Branch data Line data Source code
1 : : // Copyright 2026 HPActor Contributors
2 : : //
3 : : // Licensed under the Apache License, Version 2.0 (the "License");
4 : : // you may not use this file except in compliance with the License.
5 : : // You may obtain a copy of the License at
6 : : //
7 : : // http://www.apache.org/licenses/LICENSE-2.0
8 : : //
9 : : // Unless required by applicable law or agreed to in writing, software
10 : : // distributed under the License is distributed on an "AS IS" BASIS,
11 : : // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 : : // See the License for the specific language governing permissions and
13 : : // limitations under the License.
14 : :
15 : : #include <hpactor/net/registrar.hpp>
16 : : #include <hpactor/net/registrar_serialization.hpp>
17 : :
18 : : #include <algorithm>
19 : : #include <arpa/inet.h>
20 : : #include <netdb.h>
21 : : #include <netinet/in.h>
22 : : #include <netinet/tcp.h>
23 : : #include <sys/socket.h>
24 : : #include <unistd.h>
25 : :
26 : : #include <cstring>
27 : :
28 : : #include <hpactor/log/logger.hpp>
29 : :
30 : : namespace hpactor {
31 : :
32 : : namespace net {
33 : :
34 : : namespace {
35 : :
36 : : constexpr size_t kUdpMagicOffset = 0;
37 : : constexpr size_t kUdpVersionOffset = 4;
38 : : constexpr size_t kUdpTypeOffset = 5;
39 : : constexpr size_t kUdpPayloadLengthOffset = 6;
40 : : constexpr size_t kUdpPayloadOffset = RegistrarHeaderSize;
41 : :
42 : : struct UdpPacketView {
43 : : RegistrarMessageType type = RegistrarMessageType::ResolveQuery;
44 : : StreamBuffer payload;
45 : : };
46 : :
47 : 0 : bool parse_udp_packet(const StreamBuffer& data, UdpPacketView& packet) {
48 : 0 : if (data.size() < RegistrarHeaderSize) {
49 : 0 : return false;
50 : : }
51 : :
52 : : uint32_t magic;
53 : 0 : memcpy(&magic, data.data() + kUdpMagicOffset, 4);
54 : 0 : magic = ntohl(magic);
55 : 0 : if (magic != RegistrarMagic) {
56 : 0 : return false;
57 : : }
58 : :
59 : 0 : if (data[kUdpVersionOffset] != RegistrarVersion) {
60 : 0 : return false;
61 : : }
62 : :
63 : : uint32_t payload_len;
64 : 0 : memcpy(&payload_len, data.data() + kUdpPayloadLengthOffset, 4);
65 : 0 : payload_len = ntohl(payload_len);
66 : 0 : if (payload_len > data.size() - kUdpPayloadOffset) {
67 : 0 : return false;
68 : : }
69 : :
70 : 0 : packet.type = static_cast<RegistrarMessageType>(data[kUdpTypeOffset]);
71 : 0 : packet.payload.assign(data.begin() + kUdpPayloadOffset,
72 : 0 : data.begin() + kUdpPayloadOffset + payload_len);
73 : 0 : return true;
74 : : }
75 : :
76 : : StreamBuffer
77 : 0 : make_udp_packet(RegistrarMessageType type, const StreamBuffer& payload) {
78 : 0 : StreamBuffer packet(RegistrarHeaderSize + payload.size());
79 : :
80 : 0 : uint32_t magic_be = htonl(RegistrarMagic);
81 : 0 : memcpy(packet.data() + kUdpMagicOffset, &magic_be, 4);
82 : 0 : packet[kUdpVersionOffset] = RegistrarVersion;
83 : 0 : packet[kUdpTypeOffset] = static_cast<uint8_t>(type);
84 : 0 : uint32_t len_be = htonl(static_cast<uint32_t>(payload.size()));
85 : 0 : memcpy(packet.data() + kUdpPayloadLengthOffset, &len_be, 4);
86 : :
87 : 0 : if (!payload.empty()) {
88 : 0 : memcpy(packet.data() + kUdpPayloadOffset, payload.data(), payload.size());
89 : : }
90 : 0 : return packet;
91 : : }
92 : :
93 : 0 : void make_udp_destination(const std::string& host, uint16_t port, sockaddr_in& dest) {
94 : 0 : memset(&dest, 0, sizeof(dest));
95 : 0 : dest.sin_family = AF_INET;
96 : 0 : dest.sin_port = htons(port);
97 : 0 : inet_pton(AF_INET, host.c_str(), &dest.sin_addr);
98 : 0 : }
99 : :
100 : : } // namespace
101 : :
102 : : // -----------------------------------------------------------------------------
103 : : // RegistrarConnection Implementation
104 : : // -----------------------------------------------------------------------------
105 : :
106 : 0 : RegistrarConnection::RegistrarConnection(EndPoint remote_endpoint,
107 : 0 : EventLoop* loop, int fd)
108 : 0 : : remote_endpoint_(remote_endpoint), loop_(loop), fd_(fd),
109 : 0 : header_buffer_(TcpHeaderSize) {}
110 : :
111 : 0 : RegistrarConnection::~RegistrarConnection() {
112 : 0 : close();
113 : 0 : }
114 : :
115 : : RegistrarConnectionPtr
116 : 0 : RegistrarConnection::accepted(int fd, EndPoint remote_endpoint, EventLoop* loop) {
117 : : auto conn = std::shared_ptr<RegistrarConnection>(
118 : 0 : new RegistrarConnection(remote_endpoint, loop, fd));
119 : 0 : conn->register_with_loop();
120 : 0 : return conn;
121 : : }
122 : :
123 : : RegistrarConnectionPtr
124 : 0 : RegistrarConnection::connecting(int fd, EndPoint remote_endpoint, EventLoop* loop) {
125 : : return std::shared_ptr<RegistrarConnection>(
126 : 0 : new RegistrarConnection(remote_endpoint, loop, fd));
127 : : }
128 : :
129 : 0 : void RegistrarConnection::register_with_loop() {
130 : 0 : if (loop_ && fd_ >= 0) {
131 : 0 : loop_->add_fd(fd_, EventLoop::Event::Read);
132 : 0 : loop_->set_read_handler(fd_, [this](int /*fd*/) { handle_read_event(); });
133 : : }
134 : 0 : }
135 : :
136 : 0 : void RegistrarConnection::set_message_handler(message_handler h) {
137 : 0 : message_handler_ = std::move(h);
138 : 0 : }
139 : :
140 : 0 : void RegistrarConnection::set_disconnect_handler(disconnect_handler h) {
141 : 0 : disconnect_handler_ = std::move(h);
142 : 0 : }
143 : :
144 : 0 : void RegistrarConnection::set_send_complete_handler(send_complete_handler h) {
145 : 0 : send_complete_handler_ = std::move(h);
146 : 0 : }
147 : :
148 : 0 : void RegistrarConnection::send_message(TcpMessageType type,
149 : : const StreamBuffer& payload) {
150 : 0 : if (fd_ < 0 || !loop_)
151 : 0 : return;
152 : :
153 : : // Build message: [Magic: 4][Version: 1][Type: 1][Length: 4][Payload: N]
154 : 0 : StreamBuffer message;
155 : 0 : message.resize(TcpHeaderSize + payload.size());
156 : :
157 : 0 : uint32_t magic_be = htonl(TcpRegistrarMagic);
158 : 0 : memcpy(message.data(), &magic_be, 4);
159 : 0 : message[4] = TcpRegistrarVersion;
160 : 0 : message[5] = static_cast<uint8_t>(type);
161 : 0 : uint32_t len_be = htonl(static_cast<uint32_t>(payload.size()));
162 : 0 : memcpy(message.data() + 6, &len_be, 4);
163 : 0 : if (!payload.empty()) {
164 : 0 : memcpy(message.data() + TcpHeaderSize, payload.data(), payload.size());
165 : : }
166 : :
167 : : // Append to write buffer and flush
168 : 0 : write_buffer_.insert(write_buffer_.end(), message.begin(), message.end());
169 : 0 : flush_write_buffer();
170 : 0 : }
171 : :
172 : 0 : void RegistrarConnection::close() {
173 : 0 : if (fd_ >= 0) {
174 : 0 : if (loop_) {
175 : 0 : loop_->remove_fd(fd_);
176 : : }
177 : 0 : ::close(fd_);
178 : 0 : fd_ = -1;
179 : : }
180 : 0 : }
181 : :
182 : 0 : void RegistrarConnection::handle_read_event() {
183 : 0 : if (fd_ < 0 || !loop_)
184 : 0 : return;
185 : :
186 : : // Non-blocking read loop (fd is known to be readable via EventLoop
187 : : // notification) This pattern follows PlainConnection: poll has_event() to
188 : : // know when to read
189 : :
190 : : // Read into header buffer first
191 : 0 : while (header_bytes_read_ < TcpHeaderSize) {
192 : : // Check if still readable (edge-triggered)
193 : 0 : if (!loop_->has_event(fd_, EventLoop::Event::Read)) {
194 : 0 : return; // Would block, wait for next notification
195 : : }
196 : :
197 : 0 : ssize_t bytes_read = recv(fd_, header_buffer_.data() + header_bytes_read_,
198 : 0 : TcpHeaderSize - header_bytes_read_, 0);
199 : 0 : if (bytes_read <= 0) {
200 : : // Connection closed or error
201 : 0 : if (disconnect_handler_) {
202 : 0 : disconnect_handler_();
203 : : }
204 : 0 : close();
205 : 0 : return;
206 : : }
207 : 0 : header_bytes_read_ += static_cast<size_t>(bytes_read);
208 : : }
209 : :
210 : : // Parse header
211 : : uint32_t magic;
212 : 0 : memcpy(&magic, header_buffer_.data(), 4);
213 : 0 : magic = ntohl(magic);
214 : :
215 : 0 : if (magic != TcpRegistrarMagic) {
216 : : // Invalid magic - consume byte and try again
217 : 0 : memmove(header_buffer_.data(), header_buffer_.data() + 1, TcpHeaderSize - 1);
218 : 0 : header_bytes_read_--;
219 : : // Loop back to try reading more header bytes
220 : 0 : return;
221 : : }
222 : :
223 : 0 : uint8_t version = header_buffer_[4];
224 : 0 : if (version != TcpRegistrarVersion) {
225 : : // Unsupported version - consume byte and try again
226 : 0 : memmove(header_buffer_.data(), header_buffer_.data() + 1, TcpHeaderSize - 1);
227 : 0 : header_bytes_read_--;
228 : 0 : return;
229 : : }
230 : :
231 : 0 : current_type_ = static_cast<TcpMessageType>(header_buffer_[5]);
232 : :
233 : : uint32_t payload_len;
234 : 0 : memcpy(&payload_len, header_buffer_.data() + 6, 4);
235 : 0 : payload_len = ntohl(payload_len);
236 : :
237 : : // Allocate payload buffer
238 : 0 : payload_buffer_.resize(payload_len);
239 : 0 : payload_bytes_read_ = 0;
240 : 0 : read_state_ = ReadState::ReadingPayload;
241 : :
242 : : // Continue to read payload
243 : 0 : handle_payload_read();
244 : : }
245 : :
246 : 0 : void RegistrarConnection::handle_payload_read() {
247 : 0 : if (fd_ < 0)
248 : 0 : return;
249 : :
250 : : // Continue reading payload
251 : 0 : while (payload_bytes_read_ < payload_buffer_.size()) {
252 : : // Check if still readable
253 : 0 : if (!loop_->has_event(fd_, EventLoop::Event::Read)) {
254 : 0 : return; // Would block, wait for next notification
255 : : }
256 : :
257 : : ssize_t bytes_read =
258 : 0 : recv(fd_, payload_buffer_.data() + payload_bytes_read_,
259 : 0 : payload_buffer_.size() - payload_bytes_read_, 0);
260 : 0 : if (bytes_read <= 0) {
261 : : // Connection closed or error
262 : 0 : if (disconnect_handler_) {
263 : 0 : disconnect_handler_();
264 : : }
265 : 0 : close();
266 : 0 : return;
267 : : }
268 : 0 : payload_bytes_read_ += static_cast<size_t>(bytes_read);
269 : : }
270 : :
271 : : // Payload complete - deliver to handler
272 : 0 : if (message_handler_) {
273 : 0 : message_handler_(current_type_, payload_buffer_);
274 : : }
275 : :
276 : : // Reset to reading next header
277 : 0 : header_bytes_read_ = 0;
278 : 0 : std::fill(header_buffer_.begin(), header_buffer_.end(), 0);
279 : 0 : payload_buffer_.clear();
280 : 0 : read_state_ = ReadState::ReadingHeader;
281 : : }
282 : :
283 : 0 : void RegistrarConnection::flush_write_buffer() {
284 : 0 : if (fd_ < 0 || loop_ == nullptr || write_buffer_.empty() || is_sending_) {
285 : 0 : return;
286 : : }
287 : :
288 : 0 : is_sending_ = true;
289 : :
290 : : struct iovec iov;
291 : 0 : iov.iov_base = write_buffer_.data();
292 : 0 : iov.iov_len = write_buffer_.size();
293 : :
294 : : // Use async_send - completion routed via EventLoop completion_callback_
295 : 0 : loop_->backend()->async_send(fd_, &iov, 1, ActorId(0),
296 : : static_cast<uint32_t>(OpType::Send));
297 : : }
298 : :
299 : 0 : void RegistrarConnection::handle_send_completion(int result) {
300 : 0 : if (send_complete_handler_) {
301 : 0 : send_complete_handler_(result);
302 : : }
303 : 0 : is_sending_ = false;
304 : :
305 : 0 : if (result < 0) {
306 : : // Send error
307 : 0 : if (disconnect_handler_) {
308 : 0 : disconnect_handler_();
309 : : }
310 : 0 : close();
311 : 0 : return;
312 : : }
313 : :
314 : : // Remove sent StreamBuffer from write buffer
315 : 0 : if (static_cast<size_t>(result) >= write_buffer_.size()) {
316 : 0 : write_buffer_.clear();
317 : : } else {
318 : 0 : write_buffer_.erase(write_buffer_.begin(), write_buffer_.begin() + result);
319 : : }
320 : :
321 : : // Continue flushing if more data
322 : 0 : if (!write_buffer_.empty()) {
323 : 0 : flush_write_buffer();
324 : : }
325 : : }
326 : :
327 : : // -----------------------------------------------------------------------------
328 : : // HostResolver Implementation
329 : : // -----------------------------------------------------------------------------
330 : :
331 : 0 : std::string HostResolver::resolve(const std::string& hostname) {
332 : 0 : std::lock_guard<std::mutex> lock(mutex_);
333 : :
334 : : // Check cache first
335 : 0 : auto it = cache_.find(hostname);
336 : 0 : if (it != cache_.end()) {
337 : 0 : if (it->second.expires_at > std::chrono::steady_clock::now()) {
338 : 0 : return it->second.ip;
339 : : }
340 : : // Expired - remove it
341 : 0 : cache_.erase(it);
342 : : }
343 : :
344 : : // Check if hostname is already an IP address
345 : : struct in_addr addr;
346 : 0 : if (inet_pton(AF_INET, hostname.c_str(), &addr) == 1) {
347 : : // It's a valid IP address, cache it
348 : 0 : cache(hostname, hostname, std::chrono::seconds(300));
349 : 0 : return hostname;
350 : : }
351 : :
352 : : // Try DNS resolution
353 : : struct addrinfo hints;
354 : 0 : std::memset(&hints, 0, sizeof(hints));
355 : 0 : hints.ai_family = AF_INET;
356 : 0 : hints.ai_socktype = SOCK_STREAM;
357 : :
358 : 0 : struct addrinfo* result = nullptr;
359 : 0 : int ret = getaddrinfo(hostname.c_str(), nullptr, &hints, &result);
360 : 0 : if (ret != 0) {
361 : 0 : return "";
362 : : }
363 : :
364 : 0 : std::string ip;
365 : 0 : if (result != nullptr) {
366 : : char ipstr[INET_ADDRSTRLEN];
367 : 0 : struct sockaddr_in* addr_in =
368 : 0 : reinterpret_cast<struct sockaddr_in*>(result->ai_addr);
369 : 0 : if (inet_ntop(AF_INET, &addr_in->sin_addr, ipstr, sizeof(ipstr)) != nullptr) {
370 : 0 : ip = ipstr;
371 : : }
372 : 0 : freeaddrinfo(result);
373 : : }
374 : :
375 : 0 : if (!ip.empty()) {
376 : 0 : cache(hostname, ip, std::chrono::seconds(300));
377 : : }
378 : :
379 : 0 : return ip;
380 : 0 : }
381 : :
382 : 0 : void HostResolver::resolve_async(const std::string& hostname,
383 : : std::function<void(std::string ip)> callback) {
384 : : // For now, do blocking resolution in a background context
385 : : // In production, this would use a thread pool
386 : 0 : std::string result = resolve(hostname);
387 : 0 : callback(result);
388 : 0 : }
389 : :
390 : 2 : std::string HostResolver::get_cached(const std::string& hostname) const {
391 : 2 : std::lock_guard<std::mutex> lock(mutex_);
392 : 2 : auto it = cache_.find(hostname);
393 : 2 : if (it != cache_.end()) {
394 : 1 : if (it->second.expires_at > std::chrono::steady_clock::now()) {
395 : 1 : return it->second.ip;
396 : : }
397 : : }
398 : 2 : return "";
399 : 2 : }
400 : :
401 : 1 : void HostResolver::cache(const std::string& hostname, const std::string& ip,
402 : : std::chrono::seconds ttl) {
403 : 1 : std::lock_guard<std::mutex> lock(mutex_);
404 : 1 : CacheEntry entry;
405 : 1 : entry.ip = ip;
406 : 1 : entry.expires_at = std::chrono::steady_clock::now() + ttl;
407 : 1 : cache_[hostname] = entry;
408 : 1 : }
409 : :
410 : 0 : void HostResolver::clear_expired() {
411 : 0 : std::lock_guard<std::mutex> lock(mutex_);
412 : 0 : auto now = std::chrono::steady_clock::now();
413 : 0 : for (auto it = cache_.begin(); it != cache_.end();) {
414 : 0 : if (it->second.expires_at <= now) {
415 : 0 : it = cache_.erase(it);
416 : : } else {
417 : 0 : ++it;
418 : : }
419 : : }
420 : 0 : }
421 : :
422 : : // -----------------------------------------------------------------------------
423 : : // NodeRegistry Implementation
424 : : // -----------------------------------------------------------------------------
425 : :
426 : 3 : NodeRegistry::NodeRegistry(const RegistrarConfig& config) : config_(config) {}
427 : :
428 : 2 : void NodeRegistry::upsert_endpoint(NodeEndpoint endpoint) {
429 : 2 : std::lock_guard<std::mutex> lock(mutex_);
430 : 2 : endpoint.last_seen = std::chrono::steady_clock::now();
431 : 2 : endpoints_[endpoint.identity.endpoint] = endpoint;
432 : 2 : }
433 : :
434 : 0 : bool NodeRegistry::remove_endpoint(EndPoint endpoint) {
435 : 0 : std::lock_guard<std::mutex> lock(mutex_);
436 : 0 : return endpoints_.erase(endpoint) > 0;
437 : 0 : }
438 : :
439 : 3 : NodeEndpoint* NodeRegistry::get(EndPoint endpoint) {
440 : 3 : std::lock_guard<std::mutex> lock(mutex_);
441 : 3 : auto it = endpoints_.find(endpoint);
442 : 3 : if (it != endpoints_.end()) {
443 : 2 : return &it->second;
444 : : }
445 : 1 : return nullptr;
446 : 3 : }
447 : :
448 : 1 : bool NodeRegistry::has(EndPoint endpoint) const {
449 : 1 : std::lock_guard<std::mutex> lock(mutex_);
450 : 1 : return endpoints_.find(endpoint) != endpoints_.end();
451 : 1 : }
452 : :
453 : 2 : std::vector<NodeEndpoint> NodeRegistry::all() const {
454 : 2 : std::lock_guard<std::mutex> lock(mutex_);
455 : 2 : std::vector<NodeEndpoint> result;
456 : 2 : result.reserve(endpoints_.size());
457 : 3 : for (const auto& [id, ep] : endpoints_) {
458 : 1 : result.push_back(ep);
459 : : }
460 : 2 : return result;
461 : 2 : }
462 : :
463 : 0 : size_t NodeRegistry::remove_expired() {
464 : 0 : std::lock_guard<std::mutex> lock(mutex_);
465 : 0 : auto now = std::chrono::steady_clock::now();
466 : 0 : size_t removed = 0;
467 : 0 : for (auto it = endpoints_.begin(); it != endpoints_.end();) {
468 : : // Static routes don't expire
469 : 0 : if (!it->second.is_static_route) {
470 : 0 : auto age = now - it->second.last_seen;
471 : 0 : if (age > config_.expiration_timeout) {
472 : 0 : it = endpoints_.erase(it);
473 : 0 : ++removed;
474 : 0 : continue;
475 : : }
476 : : }
477 : 0 : ++it;
478 : : }
479 : 0 : return removed;
480 : 0 : }
481 : :
482 : : // -----------------------------------------------------------------------------
483 : : // UdpRegistrar Implementation
484 : : // -----------------------------------------------------------------------------
485 : :
486 : 6 : UdpRegistrar::UdpRegistrar(const RegistrarConfig& config,
487 : 6 : EndPoint local_endpoint, EventLoop* loop)
488 : 6 : : config_(config), local_endpoint_(local_endpoint), loop_(loop) {}
489 : :
490 : 6 : UdpRegistrar::~UdpRegistrar() {
491 : 6 : stop();
492 : 6 : }
493 : :
494 : 0 : void UdpRegistrar::start() {
495 : 0 : if (loop_) {
496 : : // Event-driven path — use the EventLoop-integrated async methods.
497 : : // Probe server vs client mode by attempting to bind the TCP port.
498 : 0 : int test_sock = socket(AF_INET, SOCK_STREAM, 0);
499 : 0 : if (test_sock < 0) {
500 : 0 : start_client_mode_async();
501 : 0 : return;
502 : : }
503 : :
504 : 0 : int reuse = 1;
505 : 0 : setsockopt(test_sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
506 : :
507 : : struct sockaddr_in addr;
508 : 0 : memset(&addr, 0, sizeof(addr));
509 : 0 : addr.sin_family = AF_INET;
510 : 0 : addr.sin_addr.s_addr = INADDR_ANY;
511 : 0 : addr.sin_port = htons(config_.tcp_port);
512 : :
513 : 0 : if (bind(test_sock, reinterpret_cast<struct sockaddr*>(&addr),
514 : 0 : sizeof(addr)) == 0) {
515 : 0 : close(test_sock);
516 : 0 : start_server_mode_async();
517 : : } else {
518 : 0 : close(test_sock);
519 : 0 : start_client_mode_async();
520 : : }
521 : 0 : return;
522 : : }
523 : :
524 : : // Legacy path — no EventLoop available, use blocking socket I/O.
525 : 0 : int test_sock = socket(AF_INET, SOCK_STREAM, 0);
526 : 0 : if (test_sock < 0) {
527 : 0 : start_client_mode();
528 : 0 : return;
529 : : }
530 : :
531 : 0 : int reuse = 1;
532 : 0 : setsockopt(test_sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
533 : :
534 : : struct sockaddr_in addr;
535 : 0 : memset(&addr, 0, sizeof(addr));
536 : 0 : addr.sin_family = AF_INET;
537 : 0 : addr.sin_addr.s_addr = INADDR_ANY;
538 : 0 : addr.sin_port = htons(config_.tcp_port);
539 : :
540 : 0 : if (bind(test_sock, reinterpret_cast<struct sockaddr*>(&addr), sizeof(addr)) ==
541 : : 0) {
542 : 0 : close(test_sock);
543 : 0 : start_server_mode();
544 : : } else {
545 : 0 : close(test_sock);
546 : 0 : start_client_mode();
547 : : }
548 : : }
549 : :
550 : 11 : void UdpRegistrar::stop() {
551 : 11 : server_.reset();
552 : 11 : client_.reset();
553 : 11 : client_registry_.reset();
554 : :
555 : 11 : if (udp_socket_ >= 0) {
556 : 0 : if (loop_) {
557 : 0 : loop_->clear_read_handler(udp_socket_);
558 : 0 : loop_->remove_fd(udp_socket_);
559 : : }
560 : 0 : close(udp_socket_);
561 : 0 : udp_socket_ = -1;
562 : : }
563 : 11 : }
564 : :
565 : 0 : void UdpRegistrar::start_server_mode() {
566 : 0 : server_ = std::make_unique<RegistrarServer>(config_, local_endpoint_, loop_);
567 : 0 : server_->start();
568 : :
569 : 0 : setup_udp_socket();
570 : 0 : }
571 : :
572 : 0 : void UdpRegistrar::start_client_mode() {
573 : : // Create registry populated with static routes
574 : 0 : client_registry_ = std::make_unique<NodeRegistry>(config_);
575 : :
576 : : // Populate with static routes
577 : 0 : for (const auto& route : config_.static_routes) {
578 : 0 : NodeEndpoint ep;
579 : 0 : ep.identity.endpoint = route.endpoint;
580 : 0 : ep.identity.host = route.address;
581 : 0 : ep.tcp_port = route.port;
582 : 0 : ep.is_static_route = true;
583 : 0 : client_registry_->upsert_endpoint(ep);
584 : 0 : }
585 : :
586 : : // Use first static route as server if available
587 : 0 : EndPoint server_endpoint;
588 : 0 : if (!config_.static_routes.empty()) {
589 : 0 : server_endpoint = config_.static_routes[0].endpoint;
590 : : }
591 : :
592 : 0 : client_ = std::make_unique<RegistrarClient>(
593 : 0 : config_, local_endpoint_, server_endpoint, client_registry_.get(), loop_);
594 : 0 : client_->set_failover_callback([this]() { failover(); });
595 : 0 : client_->start();
596 : 0 : }
597 : :
598 : 0 : void UdpRegistrar::setup_udp_socket() {
599 : 0 : udp_socket_ = socket(AF_INET, SOCK_DGRAM, 0);
600 : 0 : if (udp_socket_ < 0)
601 : 0 : return;
602 : :
603 : 0 : int reuse = 1;
604 : 0 : setsockopt(udp_socket_, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
605 : :
606 : : struct sockaddr_in udp_addr;
607 : 0 : memset(&udp_addr, 0, sizeof(udp_addr));
608 : 0 : udp_addr.sin_family = AF_INET;
609 : 0 : udp_addr.sin_addr.s_addr = INADDR_ANY;
610 : 0 : udp_addr.sin_port = htons(config_.udp_port);
611 : 0 : if (bind(udp_socket_, reinterpret_cast<struct sockaddr*>(&udp_addr),
612 : 0 : sizeof(udp_addr)) < 0) {
613 : 0 : close(udp_socket_);
614 : 0 : udp_socket_ = -1;
615 : 0 : return;
616 : : }
617 : :
618 : 0 : if (loop_) {
619 : 0 : loop_->add_fd(udp_socket_, EventLoop::Event::Read);
620 : 0 : loop_->set_read_handler(udp_socket_,
621 : 0 : [this](int /*fd*/) { handle_udp_read_ready(); });
622 : : }
623 : : }
624 : :
625 : 0 : void UdpRegistrar::start_server_mode_async() {
626 : 0 : server_ = std::make_unique<RegistrarServer>(config_, local_endpoint_, loop_);
627 : 0 : server_->start();
628 : :
629 : 0 : setup_udp_socket();
630 : 0 : if (udp_socket_ < 0)
631 : 0 : return;
632 : :
633 : 0 : udp_recv_buffer_.resize(kUdpRecvBufferSize);
634 : 0 : issue_async_recvfrom();
635 : : }
636 : :
637 : 0 : void UdpRegistrar::start_client_mode_async() {
638 : : // Create registry populated with static routes
639 : 0 : client_registry_ = std::make_unique<NodeRegistry>(config_);
640 : :
641 : : // Populate with static routes
642 : 0 : for (const auto& route : config_.static_routes) {
643 : 0 : NodeEndpoint ep;
644 : 0 : ep.identity.endpoint = route.endpoint;
645 : 0 : ep.identity.host = route.address;
646 : 0 : ep.tcp_port = route.port;
647 : 0 : ep.is_static_route = true;
648 : 0 : client_registry_->upsert_endpoint(ep);
649 : 0 : }
650 : :
651 : : // Use first static route as server if available
652 : 0 : EndPoint server_endpoint;
653 : 0 : if (!config_.static_routes.empty()) {
654 : 0 : server_endpoint = config_.static_routes[0].endpoint;
655 : : }
656 : :
657 : 0 : client_ = std::make_unique<RegistrarClient>(
658 : 0 : config_, local_endpoint_, server_endpoint, client_registry_.get(), loop_);
659 : 0 : client_->set_failover_callback([this]() { failover(); });
660 : 0 : client_->start();
661 : 0 : }
662 : :
663 : 0 : void UdpRegistrar::issue_async_recvfrom() {
664 : 0 : if (udp_socket_ < 0 || !loop_)
665 : 0 : return;
666 : :
667 : : // Clear address storage for next recvfrom
668 : 0 : memset(&udp_src_addr_, 0, sizeof(udp_src_addr_));
669 : 0 : udp_src_addr_len_ = sizeof(udp_src_addr_);
670 : : }
671 : :
672 : 0 : void UdpRegistrar::handle_udp_read_ready() {
673 : 0 : if (udp_socket_ < 0)
674 : 0 : return;
675 : :
676 : : // Non-blocking recvfrom — the EventLoop only calls this callback
677 : : // when the fd is readable (edge-triggered epoll/kqueue).
678 : : char buffer[kUdpRecvBufferSize];
679 : : struct sockaddr_in src_addr;
680 : 0 : socklen_t src_addr_len = sizeof(src_addr);
681 : :
682 : : ssize_t bytes_read =
683 : 0 : recvfrom(udp_socket_, buffer, sizeof(buffer), 0,
684 : : reinterpret_cast<struct sockaddr*>(&src_addr), &src_addr_len);
685 : :
686 : 0 : if (bytes_read > 0) {
687 : 0 : StreamBuffer data(buffer, buffer + bytes_read);
688 : : char ip_str[INET_ADDRSTRLEN];
689 : 0 : std::string from_host;
690 : 0 : uint16_t from_port = 0;
691 : :
692 : 0 : if (inet_ntop(AF_INET, &src_addr.sin_addr, ip_str, sizeof(ip_str))) {
693 : 0 : from_host = ip_str;
694 : : }
695 : 0 : from_port = ntohs(src_addr.sin_port);
696 : :
697 : 0 : handle_udp_recv_completion(data, from_host, from_port);
698 : 0 : }
699 : : }
700 : :
701 : 0 : void UdpRegistrar::handle_udp_recv_completion(const StreamBuffer& data,
702 : : const std::string& from_host,
703 : : uint16_t from_port) {
704 : : // Call the existing handler
705 : 0 : handle_udp_packet(data, from_host, from_port);
706 : 0 : }
707 : :
708 : 0 : void UdpRegistrar::send_udp_response(const StreamBuffer& data,
709 : : const struct sockaddr_in& dest) {
710 : 0 : if (udp_socket_ < 0)
711 : 0 : return;
712 : :
713 : 0 : if (loop_) {
714 : : // Use async_sendto for async UDP send
715 : : struct iovec iov;
716 : 0 : iov.iov_base = const_cast<uint8_t*>(data.data());
717 : 0 : iov.iov_len = data.size();
718 : :
719 : 0 : loop_->backend()->async_sendto(
720 : : udp_socket_, &iov, 1, reinterpret_cast<const sockaddr*>(&dest),
721 : : sizeof(dest), ActorId(0), static_cast<uint32_t>(OpType::SendTo));
722 : : } else {
723 : : // Fallback to blocking sendto
724 : 0 : sendto(udp_socket_, data.data(), data.size(), 0,
725 : : reinterpret_cast<const struct sockaddr*>(&dest), sizeof(dest));
726 : : }
727 : : }
728 : :
729 : 0 : void UdpRegistrar::failover() {
730 : : // Stop current mode
731 : 0 : server_.reset();
732 : 0 : client_.reset();
733 : :
734 : : // Try to become server
735 : 0 : int test_sock = socket(AF_INET, SOCK_STREAM, 0);
736 : 0 : if (test_sock < 0) {
737 : 0 : start_client_mode();
738 : 0 : return;
739 : : }
740 : :
741 : 0 : int reuse = 1;
742 : 0 : setsockopt(test_sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
743 : :
744 : : struct sockaddr_in addr;
745 : 0 : memset(&addr, 0, sizeof(addr));
746 : 0 : addr.sin_family = AF_INET;
747 : 0 : addr.sin_addr.s_addr = INADDR_ANY;
748 : 0 : addr.sin_port = htons(config_.tcp_port);
749 : :
750 : 0 : if (bind(test_sock, reinterpret_cast<struct sockaddr*>(&addr), sizeof(addr)) ==
751 : : 0) {
752 : 0 : close(test_sock);
753 : 0 : start_server_mode();
754 : : } else {
755 : 0 : close(test_sock);
756 : 0 : start_client_mode();
757 : : }
758 : : }
759 : :
760 : 0 : NodeEndpoint* UdpRegistrar::get_endpoint(EndPoint endpoint) {
761 : 0 : if (server_) {
762 : 0 : return server_->registry()->get(endpoint);
763 : : }
764 : : // In client mode, could query via client
765 : 0 : return nullptr;
766 : : }
767 : :
768 : 1 : std::vector<NodeEndpoint> UdpRegistrar::get_all_endpoints() const {
769 : 1 : if (server_) {
770 : 0 : return server_->registry()->all();
771 : : }
772 : 1 : return {};
773 : : }
774 : :
775 : 0 : void UdpRegistrar::set_node_callback(node_callback cb) {
776 : 0 : node_callback_ = std::move(cb);
777 : 0 : }
778 : :
779 : 0 : void UdpRegistrar::handle_udp_packet(const StreamBuffer& data,
780 : : const std::string& from_host,
781 : : uint16_t from_port) {
782 : : // Handle incoming UDP packet for resolution
783 : : // Packet format: [Magic: 4][Version: 1][Type: 1][Length: 4]
784 : : // [Reserved: 2][Payload...]
785 : 0 : UdpPacketView packet;
786 : 0 : if (!parse_udp_packet(data, packet)) {
787 : 0 : HPACTOR_LOG_ERROR(log::LogCategory::kRegistrar, ActorId{0}, 0,
788 : : "malformed registrar packet");
789 : 0 : return;
790 : : }
791 : :
792 : 0 : switch (packet.type) {
793 : 0 : case RegistrarMessageType::ResolveQuery:
794 : 0 : handle_resolve_query(packet.payload, from_host, from_port);
795 : 0 : break;
796 : :
797 : 0 : case RegistrarMessageType::ResolveResponse:
798 : 0 : handle_resolve_response(packet.payload);
799 : 0 : break;
800 : :
801 : 0 : default:
802 : : // Unknown message type
803 : 0 : break;
804 : : }
805 : 0 : }
806 : :
807 : 0 : void UdpRegistrar::handle_resolve_query(const StreamBuffer& payload,
808 : : const std::string& from_host,
809 : : uint16_t from_port) {
810 : 0 : if (!server_) {
811 : 0 : return;
812 : : }
813 : :
814 : 0 : PbResolveQueryPayload msg;
815 : 0 : if (!parse_resolve_query_payload(payload, msg)) {
816 : 0 : return;
817 : : }
818 : :
819 : 0 : EndPoint target_endpoint = endpoint_ops::parse_endpoint(msg.target_endpoint());
820 : 0 : NodeEndpoint* ep = server_->registry()->get(target_endpoint);
821 : 0 : if (ep == nullptr) {
822 : 0 : HPACTOR_LOG_WARNING(
823 : : log::LogCategory::kRegistrar, ActorId{0},
824 : : static_cast<uint32_t>(log::LogEventId::kRegistrarResolveMiss),
825 : : "registrar resolve miss");
826 : 0 : return;
827 : : }
828 : :
829 : 0 : send_resolve_response(*ep, from_host, from_port);
830 : 0 : }
831 : :
832 : 0 : void UdpRegistrar::handle_resolve_response(const StreamBuffer& payload) {
833 : 0 : PbResolveResponsePayload msg;
834 : 0 : if (!parse_resolve_response_payload(payload, msg)) {
835 : 0 : return;
836 : : }
837 : :
838 : 0 : if (!server_) {
839 : 0 : return;
840 : : }
841 : :
842 : 0 : auto& info = msg.endpoint_info();
843 : 0 : NodeEndpoint ep;
844 : 0 : ep.identity.endpoint = endpoint_ops::parse_endpoint(info.endpoint());
845 : 0 : ep.identity.host = info.host();
846 : 0 : ep.tcp_port = static_cast<uint16_t>(info.tcp_port());
847 : 0 : ep.last_seen = std::chrono::steady_clock::now();
848 : 0 : server_->registry()->upsert_endpoint(ep);
849 : :
850 : 0 : if (node_callback_) {
851 : 0 : node_callback_(ep.identity.endpoint, true);
852 : : }
853 : 0 : }
854 : :
855 : 0 : void UdpRegistrar::send_resolve_response(const NodeEndpoint& endpoint,
856 : : const std::string& from_host,
857 : : uint16_t from_port) const {
858 : 0 : if (udp_socket_ < 0) {
859 : 0 : return;
860 : : }
861 : :
862 : : sockaddr_in dest_addr;
863 : 0 : make_udp_destination(from_host, from_port, dest_addr);
864 : :
865 : 0 : StreamBuffer response_payload = serialize_resolve_response_payload(endpoint);
866 : : StreamBuffer response =
867 : 0 : make_udp_packet(RegistrarMessageType::ResolveResponse, response_payload);
868 : 0 : sendto(udp_socket_, response.data(), response.size(), 0,
869 : : reinterpret_cast<struct sockaddr*>(&dest_addr), sizeof(dest_addr));
870 : 0 : }
871 : :
872 : : // ── IServiceDiscovery overrides ────────────────────────────────────────────
873 : :
874 : 0 : Member UdpRegistrar::to_member(const NodeEndpoint& ep) {
875 : 0 : Member m;
876 : 0 : m.identity.endpoint = ep.identity.endpoint;
877 : 0 : m.identity.host = ep.identity.host;
878 : 0 : m.identity.uds_path = ep.identity.uds_path;
879 : 0 : m.identity.acceptors = ep.identity.acceptors;
880 : 0 : m.last_seen = ep.last_seen;
881 : 0 : return m;
882 : : }
883 : :
884 : 1 : std::vector<Member> UdpRegistrar::discover_all() const {
885 : 1 : std::vector<Member> result;
886 : :
887 : : // Collect from server registry
888 : 1 : auto eps = get_all_endpoints();
889 : 1 : result.reserve(eps.size());
890 : 1 : for (const auto& ep : eps)
891 : 0 : result.push_back(to_member(ep));
892 : :
893 : : // Also collect from client registry (static routes)
894 : 1 : if (client_registry_) {
895 : 0 : auto client_eps = client_registry_->all();
896 : 0 : result.reserve(result.size() + client_eps.size());
897 : 0 : for (const auto& ep : client_eps)
898 : 0 : result.push_back(to_member(ep));
899 : 0 : }
900 : :
901 : 1 : return result;
902 : 1 : }
903 : :
904 : 1 : const Member* UdpRegistrar::discover(EndPoint ep) const {
905 : : // Check server registry first
906 : 1 : if (server_) {
907 : 0 : auto* node_ep = server_->registry()->get(ep);
908 : 0 : if (node_ep) {
909 : 0 : endpoint_to_member_[ep] = to_member(*node_ep);
910 : 0 : return &endpoint_to_member_[ep];
911 : : }
912 : : }
913 : :
914 : : // Check client registry
915 : 1 : if (client_registry_) {
916 : 0 : auto* node_ep = client_registry_->get(ep);
917 : 0 : if (node_ep) {
918 : 0 : endpoint_to_member_[ep] = to_member(*node_ep);
919 : 0 : return &endpoint_to_member_[ep];
920 : : }
921 : : }
922 : :
923 : 1 : return nullptr;
924 : : }
925 : :
926 : 0 : void UdpRegistrar::announce(Member) {
927 : : // No-op: registrar server handles membership via Register/Heartbeat.
928 : 0 : }
929 : :
930 : 0 : void UdpRegistrar::on_member_change(MemberChangeCallback cb) {
931 : 0 : member_change_cb_ = std::move(cb);
932 : 0 : }
933 : :
934 : : } // namespace net
935 : : } // namespace hpactor
|
