diff --git a/ipc-codegen/SCHEMA_SPEC.md b/ipc-codegen/SCHEMA_SPEC.md index 79f75956e935..d4f1ddabe05e 100644 --- a/ipc-codegen/SCHEMA_SPEC.md +++ b/ipc-codegen/SCHEMA_SPEC.md @@ -156,12 +156,18 @@ is serialized. The golden corpus pins these encodings across all languages. ### Framing -All messages use length-prefix framing: +Framing is owned by ipc-runtime, below this spec: every message travels as ``` -[4 bytes: payload length, little-endian uint32][payload: msgpack bytes] +[4 bytes: length, LE uint32][8 bytes: request id, LE uint64][payload: msgpack bytes] ``` +where the length counts the id plus the payload. The request id is assigned +by the client and echoed on the response; responses arrive in completion +order and are correlated by id, entirely inside the transports — the msgpack +payloads this spec describes never contain the id, and generated code never +sees it. + ### Request wire format A request is a 1-element msgpack array wrapping a `[name, payload]` pair: diff --git a/ipc-runtime/README.md b/ipc-runtime/README.md index 15118999154d..c1088fe9cf33 100644 --- a/ipc-runtime/README.md +++ b/ipc-runtime/README.md @@ -53,7 +53,7 @@ ipc-runtime/ shm_server.hpp # single-client (SPSC) SHM server mpsc_shm_client.hpp # multi-client SHM client (one slot per client) mpsc_shm_server.hpp # multi-client SHM server - shm_common.hpp # length-prefix framing over the rings + shm_common.hpp # length + request-id framing over the rings shm/ # lock-free SPSC/MPSC ring buffer primitives serve_helper.{hpp,cpp} # ipc::make_server / make_client (path-suffix dispatch) signal_handlers.{hpp,cpp} # ipc::install_default_signal_handlers @@ -122,13 +122,18 @@ public: static std::unique_ptr create_socket(const std::string& socket_path); static std::unique_ptr create_shm(const std::string& base_name); static std::unique_ptr create_mpsc_shm(const std::string& base_name, - std::size_t client_id); - - virtual bool connect() = 0; - virtual bool send(const void* data, size_t len, uint64_t timeout_ns) = 0; - virtual std::span receive(uint64_t timeout_ns) = 0; - virtual void release(size_t message_size) = 0; - virtual void close() = 0; + std::size_t client_id = kAutoClientId); + + virtual bool connect() = 0; + // Explicit-id primitives — for pipelining callers that own their own pairing. + virtual bool send(uint64_t request_id, const void* data, size_t len, + uint64_t timeout_ns) = 0; + virtual std::span receive(uint64_t timeout_ns, uint64_t& request_id) = 0; + // Serial convenience — auto-assigns ids and verifies the echo (one in flight). + bool send(const void* data, size_t len, uint64_t timeout_ns); + std::span receive(uint64_t timeout_ns); + virtual void release(size_t message_size) = 0; + virtual void close() = 0; }; class IpcServer { @@ -144,19 +149,22 @@ public: std::size_t request_ring_size = DEFAULT_RING_SIZE, std::size_t response_ring_size = DEFAULT_RING_SIZE); - virtual bool listen() = 0; - virtual int wait_for_data(uint64_t timeout_ns) = 0; - virtual std::span receive(int client_id) = 0; - virtual void release(int client_id, size_t message_size) = 0; - virtual bool send(int client_id, const void* data, size_t len) = 0; - virtual void close() = 0; + virtual bool listen() = 0; + virtual int wait_for_data(uint64_t timeout_ns) = 0; + virtual std::span receive(int client_id, uint64_t& request_id) = 0; + virtual void release(int client_id, size_t message_size) = 0; + virtual bool send(int client_id, uint64_t request_id, const void* data, size_t len) = 0; + virtual void close() = 0; virtual void request_shutdown(); // NOT signal-safe (wakes waiters) void request_shutdown_from_signal() noexcept; // signal-safe variant - virtual void run(const Handler& handler); // event loop + virtual void run(const Handler& handler); // serial event loop (echoes ids) + void run_reactor(const AsyncHandler& handler); // async loop: handlers respond from any + // thread; responses sent in completion order }; std::unique_ptr make_server(const std::string& path, const ServerOptions& = {}); -std::unique_ptr make_client(const std::string& path, std::size_t shm_client_id = 0); +std::unique_ptr make_client(const std::string& path, + std::size_t shm_client_id = kAutoClientId); void install_default_signal_handlers(IpcServer& server); @@ -200,9 +208,9 @@ Two transport-specific clients: | Class | Transport | Sync / Async | |----------------------|----------------------------|--------------------------------------------------------------| -| `UdsIpcClient` | Node `net.Socket` | async only | -| `NapiShmSyncClient` | MPSC-SHM via NAPI bridge | sync | -| `NapiShmAsyncClient` | MPSC-SHM via NAPI bridge | async (C++ poll thread + ThreadSafeFunction bridge) | +| `UdsIpcClient` | Node `net.Socket` | async; pipelines, pairs responses to callers by request id | +| `NapiShmSyncClient` | MPSC-SHM via NAPI bridge | sync (one in flight) | +| `NapiShmAsyncClient` | MPSC-SHM via NAPI bridge | async (C++ poll thread + ThreadSafeFunction bridge); pipelines, pairs by request id | `UdsIpcServer` is provided for in-process tests; production servers are in C++. @@ -228,18 +236,28 @@ definition of the transport limits and defaults: ## Wire framing -Both transports use a 4-byte little-endian length prefix in front of every -message: +Both transports frame every message as a length prefix, a request id, and +the payload; the length counts the id plus the payload: ``` -┌───────────────────────┬────────────────────────┐ -│ Length (uint32 le) │ Payload (Length bytes) │ -└───────────────────────┴────────────────────────┘ +┌────────────────────┬─────────────────────────┬──────────────────────────────┐ +│ Length (uint32 le) │ Request id (uint64 le) │ Payload (Length − 8 bytes) │ +└────────────────────┴─────────────────────────┴──────────────────────────────┘ ``` -Framing is handled inside `IpcServer::receive` / `IpcClient::recv`; callers -deal in whole messages. The codegen's `Command` / `Response` NamedUnion -sits inside that payload — see `ipc-codegen/SCHEMA_SPEC.md`. +The id is client-assigned (per-connection random-start counter; 0 is +reserved for server-initiated frames) and echoed verbatim on the response. +Clients correlate responses by id, so the server sends responses in +**completion order** — there is no FIFO contract on the wire, and a slow +request never delays a fast one's response. A frame whose id matches +nothing outstanding is a stale leftover on SHM (rings persist across slot +occupants — it is released and skipped) and a fatal desync on UDS (the +kernel guarantees a fresh stream, so the connection is failed loudly). + +Framing is handled inside `IpcServer::receive` / `IpcClient::receive`; +callers deal in whole payloads and never see ids unless they use the +explicit-id primitives. The codegen's `Command` / `Response` NamedUnion +sits inside the payload — see `ipc-codegen/SCHEMA_SPEC.md`. ## Performance characteristics diff --git a/ipc-runtime/cpp/ipc_runtime/c_abi.cpp b/ipc-runtime/cpp/ipc_runtime/c_abi.cpp index d02d8db8c1f7..0e5d3d3d85eb 100644 --- a/ipc-runtime/cpp/ipc_runtime/c_abi.cpp +++ b/ipc-runtime/cpp/ipc_runtime/c_abi.cpp @@ -129,12 +129,14 @@ int ipc_server_wait_for_data(ipc_server_t* server, uint64_t timeout_ns) return server && server->impl ? server->impl->wait_for_data(timeout_ns) : -1; } -ipc_status_t ipc_server_receive(ipc_server_t* server, int client_id, const uint8_t** out, size_t* out_len) +ipc_status_t ipc_server_receive( + ipc_server_t* server, int client_id, uint64_t* request_id_out, const uint8_t** out, size_t* out_len) { - if (!server || !server->impl || !out || !out_len) { + if (!server || !server->impl || !request_id_out || !out || !out_len) { return IPC_ERR_RECV; } - auto view = server->impl->receive(client_id); + uint64_t request_id = 0; + auto view = server->impl->receive(client_id, request_id); // data() == nullptr is error/timeout; a non-null empty view is a valid // zero-length message. if (view.data() == nullptr) { @@ -142,6 +144,7 @@ ipc_status_t ipc_server_receive(ipc_server_t* server, int client_id, const uint8 *out_len = 0; return IPC_ERR_RECV; } + *request_id_out = request_id; *out = view.data(); *out_len = view.size(); return IPC_OK; @@ -154,9 +157,9 @@ void ipc_server_release(ipc_server_t* server, int client_id, size_t msg_size) } } -bool ipc_server_send(ipc_server_t* server, int client_id, const uint8_t* data, size_t len) +bool ipc_server_send(ipc_server_t* server, int client_id, uint64_t request_id, const uint8_t* data, size_t len) { - return server && server->impl ? server->impl->send(client_id, data, len) : false; + return server && server->impl ? server->impl->send(client_id, request_id, data, len) : false; } void ipc_server_run(ipc_server_t* server, ipc_server_handler_fn handler, void* ctx) diff --git a/ipc-runtime/cpp/ipc_runtime/c_abi.h b/ipc-runtime/cpp/ipc_runtime/c_abi.h index ef8641497b85..127ca6f45ced 100644 --- a/ipc-runtime/cpp/ipc_runtime/c_abi.h +++ b/ipc-runtime/cpp/ipc_runtime/c_abi.h @@ -45,47 +45,46 @@ typedef enum { IPC_OK = 0, IPC_ERR_RECV = -5 } ipc_status_t; /* --- Options ----------------------------------------------------------- */ typedef struct { - size_t max_shm_clients; /* default: 2 */ - size_t shm_request_ring_size; /* default: 4 MiB */ - size_t shm_response_ring_size; /* default: 4 MiB */ - int socket_backlog; /* default: 1 */ + size_t max_shm_clients; /* default: 2 */ + size_t shm_request_ring_size; /* default: 4 MiB */ + size_t shm_response_ring_size; /* default: 4 MiB */ + int socket_backlog; /* default: 1 */ } ipc_server_options_t; /* Populate `opts` with the same defaults ipc::ServerOptions{} provides. */ -void ipc_server_options_default(ipc_server_options_t *opts); +void ipc_server_options_default(ipc_server_options_t* opts); /* --- Server ------------------------------------------------------------ */ typedef struct ipc_server ipc_server_t; /* Pick UDS vs MPSC-SHM by suffix. Returns NULL if suffix unrecognised. */ -ipc_server_t *ipc_make_server(const char *path, - const ipc_server_options_t *opts); +ipc_server_t* ipc_make_server(const char* path, const ipc_server_options_t* opts); -ipc_server_t *ipc_server_create_socket(const char *path, int max_clients); -ipc_server_t *ipc_server_create_mpsc_shm(const char *base_name, +ipc_server_t* ipc_server_create_socket(const char* path, int max_clients); +ipc_server_t* ipc_server_create_mpsc_shm(const char* base_name, size_t max_clients, size_t request_ring_size, size_t response_ring_size); -void ipc_server_destroy(ipc_server_t *server); +void ipc_server_destroy(ipc_server_t* server); -bool ipc_server_listen(ipc_server_t *server); -void ipc_server_close(ipc_server_t *server); -void ipc_server_request_shutdown(ipc_server_t *server); +bool ipc_server_listen(ipc_server_t* server); +void ipc_server_close(ipc_server_t* server); +void ipc_server_request_shutdown(ipc_server_t* server); /* Returns client_id ≥ 0, or -1 on timeout/error. */ -int ipc_server_wait_for_data(ipc_server_t *server, uint64_t timeout_ns); +int ipc_server_wait_for_data(ipc_server_t* server, uint64_t timeout_ns); /* On success: *out / *out_len reference an internal buffer valid until - * ipc_server_release(). Returns IPC_OK or a negative status. */ -ipc_status_t ipc_server_receive(ipc_server_t *server, int client_id, - const uint8_t **out, size_t *out_len); + * ipc_server_release(), and *request_id_out holds the client-assigned id to + * echo via ipc_server_send(). Returns IPC_OK or a negative status. */ +ipc_status_t ipc_server_receive( + ipc_server_t* server, int client_id, uint64_t* request_id_out, const uint8_t** out, size_t* out_len); -void ipc_server_release(ipc_server_t *server, int client_id, size_t msg_size); +void ipc_server_release(ipc_server_t* server, int client_id, size_t msg_size); -bool ipc_server_send(ipc_server_t *server, int client_id, const uint8_t *data, - size_t len); +bool ipc_server_send(ipc_server_t* server, int client_id, uint64_t request_id, const uint8_t* data, size_t len); /* Convenience event loop. The handler is called for each incoming message; * it writes the response into a buffer it owns and stores the pointer + @@ -97,44 +96,39 @@ bool ipc_server_send(ipc_server_t *server, int client_id, const uint8_t *data, * setting *resp_len_out = 0) sends a zero-length response frame. To exit * the loop, call ipc_server_request_shutdown() from inside the handler. */ -typedef void (*ipc_server_handler_fn)(int client_id, const uint8_t *req, - size_t req_len, uint8_t **resp_out, - size_t *resp_len_out, void *ctx); +typedef void (*ipc_server_handler_fn)( + int client_id, const uint8_t* req, size_t req_len, uint8_t** resp_out, size_t* resp_len_out, void* ctx); -void ipc_server_run(ipc_server_t *server, ipc_server_handler_fn handler, - void *ctx); +void ipc_server_run(ipc_server_t* server, ipc_server_handler_fn handler, void* ctx); /* Install SIGTERM/SIGINT graceful-shutdown + SIGBUS/SIGSEGV close + * parent-death monitoring (prctl on linux, kqueue NOTE_EXIT on macOS) wired to * `server`. */ -void ipc_install_default_signal_handlers(ipc_server_t *server); +void ipc_install_default_signal_handlers(ipc_server_t* server); /* --- Client ------------------------------------------------------------ */ typedef struct ipc_client ipc_client_t; -ipc_client_t *ipc_make_client(const char *path, size_t shm_client_id); +ipc_client_t* ipc_make_client(const char* path, size_t shm_client_id); -ipc_client_t *ipc_client_create_socket(const char *socket_path); -ipc_client_t *ipc_client_create_mpsc_shm(const char *base_name, - size_t client_id); +ipc_client_t* ipc_client_create_socket(const char* socket_path); +ipc_client_t* ipc_client_create_mpsc_shm(const char* base_name, size_t client_id); -void ipc_client_destroy(ipc_client_t *client); +void ipc_client_destroy(ipc_client_t* client); -bool ipc_client_connect(ipc_client_t *client); -void ipc_client_close(ipc_client_t *client); +bool ipc_client_connect(ipc_client_t* client); +void ipc_client_close(ipc_client_t* client); -bool ipc_client_send(ipc_client_t *client, const uint8_t *data, size_t len, - uint64_t timeout_ns); +bool ipc_client_send(ipc_client_t* client, const uint8_t* data, size_t len, uint64_t timeout_ns); /* On success: IPC_OK with *out / *out_len referencing an internal buffer * valid until ipc_client_release(). A zero-length response is IPC_OK with * *out_len == 0 (still followed by ipc_client_release(0)). IPC_ERR_RECV * means timeout or disconnect. timeout_ns == 0 means infinite. */ -ipc_status_t ipc_client_receive(ipc_client_t *client, uint64_t timeout_ns, - const uint8_t **out, size_t *out_len); +ipc_status_t ipc_client_receive(ipc_client_t* client, uint64_t timeout_ns, const uint8_t** out, size_t* out_len); -void ipc_client_release(ipc_client_t *client, size_t msg_size); +void ipc_client_release(ipc_client_t* client, size_t msg_size); #ifdef __cplusplus } /* extern "C" */ diff --git a/ipc-runtime/cpp/ipc_runtime/constants.hpp b/ipc-runtime/cpp/ipc_runtime/constants.hpp index ddc054a04f16..77256e5ecfb7 100644 --- a/ipc-runtime/cpp/ipc_runtime/constants.hpp +++ b/ipc-runtime/cpp/ipc_runtime/constants.hpp @@ -21,6 +21,16 @@ namespace ipc { */ inline constexpr uint32_t MAX_FRAME_SIZE = 256U * 1024 * 1024; // 256 MiB +/** + * Every frame carries a client-assigned request id (little-endian u64) between + * the length prefix and the payload; the server echoes it on the response. + * Clients correlate responses by id, so the server may complete requests in + * any order — there is no FIFO contract on the wire. Ids are per-connection + * (random-start counter); 0 is reserved for server-initiated frames such as + * protocol errors. + */ +inline constexpr size_t FRAME_ID_SIZE = 8; + /** * Total budget for connect() retry loops, covering the window where the * server process is still starting up. Shared by UDS and SHM clients. diff --git a/ipc-runtime/cpp/ipc_runtime/ipc_client.hpp b/ipc-runtime/cpp/ipc_runtime/ipc_client.hpp index 55b67441ded3..b2d6692c6aaa 100644 --- a/ipc-runtime/cpp/ipc_runtime/ipc_client.hpp +++ b/ipc-runtime/cpp/ipc_runtime/ipc_client.hpp @@ -4,6 +4,7 @@ #include #include #include +#include #include #include #include @@ -39,18 +40,20 @@ class IpcClient { virtual bool connect() = 0; /** - * @brief Send a message to the server - * @param data Pointer to message data - * @param len Length of message in bytes + * @brief Send a request frame carrying an explicit request id + * @param request_id Caller-chosen id; the server echoes it on the response + * @param data Pointer to message payload + * @param len Length of payload in bytes * @param timeout_ns Timeout in nanoseconds (0 = infinite) * @return true if sent successfully, false on error or timeout */ - virtual bool send(const void* data, size_t len, uint64_t timeout_ns) = 0; + virtual bool send(uint64_t request_id, const void* data, size_t len, uint64_t timeout_ns) = 0; /** - * @brief Receive a message from the server (zero-copy for shared memory) + * @brief Receive a response frame (zero-copy for shared memory) * @param timeout_ns Timeout in nanoseconds (0 = infinite) - * @return Span of message data. data() == nullptr means error/timeout; + * @param request_id Out: the echoed request id of the received frame + * @return Span of message payload. data() == nullptr means error/timeout; * a non-null span of size 0 is a valid zero-length message. * * The span remains valid until release() is called or the next recv(). @@ -59,7 +62,61 @@ class IpcClient { * * Must be followed by release() to consume the message. */ - virtual std::span receive(uint64_t timeout_ns) = 0; + virtual std::span receive(uint64_t timeout_ns, uint64_t& request_id) = 0; + + /** + * @brief Send with an auto-assigned request id (serial call pattern) + * + * Convenience for one-request-in-flight clients (the generated C++ IPC + * clients): assigns the next id internally; the matching receive() overload + * verifies the echo. + */ + bool send(const void* data, size_t len, uint64_t timeout_ns) + { + return send(++last_request_id_, data, len, timeout_ns); + } + + /** + * @brief Whether frames from a previous connection can legitimately appear. + * + * SHM rings persist across occupants: a reclaimed MPSC slot (or a restarted + * client reattaching to SPSC rings) can hold leftover responses addressed to + * the previous occupant. Random-start request ids make those recognisable; + * transports where this is expected return true so the serial receive() + * discards them instead of treating them as a fatal desync. Sockets return + * false — the kernel guarantees a fresh stream, so a foreign frame there + * means the correlation is genuinely broken. + */ + virtual bool may_have_stale_frames() const { return false; } + + /** + * @brief Receive the response to the last auto-id send() + * + * Serial-contract counterpart of send(data, len, timeout). A frame whose + * echoed id does not match the last sent id is either an anticipated + * leftover from a ring's previous occupant (may_have_stale_frames() — + * released and skipped, keeping the wait for the real response) or a + * genuine desync (the connection is closed and the call fails rather than + * delivering another request's payload). + */ + std::span receive(uint64_t timeout_ns) + { + while (true) { + uint64_t echoed = 0; + auto payload = receive(timeout_ns, echoed); + if (payload.data() == nullptr || echoed == last_request_id_) { + return payload; + } + if (!may_have_stale_frames()) { + close(); + return {}; + } + // Stale leftover already sitting in the ring — consume and retry; + // draining it is immediate, so the awaited response keeps + // effectively the full timeout. + release(payload.size()); + } + } /** * @brief Wake any thread blocked in receive()/send() (for shutdown). @@ -83,6 +140,19 @@ class IpcClient { */ virtual void close() = 0; + protected: + // Auto-assigned request ids start at a random point per client instance so + // a stale frame left in a recycled SHM ring slot by a previous occupant + // cannot collide with the new occupant's ids. + uint64_t last_request_id_ = random_request_id_start(); + + static uint64_t random_request_id_start() + { + std::random_device rd; + return (static_cast(rd()) << 16) + 1; + } + + public: // Factory methods. static std::unique_ptr create_socket(const std::string& socket_path); // Single-client SHM: one request ring and one response ring. Use this diff --git a/ipc-runtime/cpp/ipc_runtime/ipc_server.hpp b/ipc-runtime/cpp/ipc_runtime/ipc_server.hpp index e05863b17681..70981fdb4e93 100644 --- a/ipc-runtime/cpp/ipc_runtime/ipc_server.hpp +++ b/ipc-runtime/cpp/ipc_runtime/ipc_server.hpp @@ -6,15 +6,14 @@ #include #include #include +#include #include #include -#include #include #include #include #include #include -#include #include #include @@ -106,15 +105,17 @@ class IpcServer { * @brief Receive next message from a specific client * * Blocks until a complete message is available. Returns a span pointing to - * the message data. For shared memory, this is a zero-copy view directly into + * the message payload (the frame's request id is stripped into + * `request_id`). For shared memory, this is a zero-copy view directly into * the ring buffer. For sockets, this is a view into an internal buffer. * * The message remains valid until release() is called with the message size. * * @param client_id Client to receive from - * @return Span of message data (empty only on error/disconnect) + * @param request_id Out: the client-assigned id to echo on the response + * @return Span of message payload (empty only on error/disconnect) */ - virtual std::span receive(int client_id) = 0; + virtual std::span receive(int client_id, uint64_t& request_id) = 0; /** * @brief Release/consume the previously received message @@ -129,13 +130,14 @@ class IpcServer { virtual void release(int client_id, size_t message_size) = 0; /** - * @brief Send a message to a specific client + * @brief Send a response frame to a specific client * @param client_id Client to send to - * @param data Pointer to message data - * @param len Length of message in bytes + * @param request_id Echoed request id (0 for server-initiated frames) + * @param data Pointer to message payload + * @param len Length of payload in bytes * @return true if sent successfully, false on error */ - virtual bool send(int client_id, const void* data, size_t len) = 0; + virtual bool send(int client_id, uint64_t request_id, const void* data, size_t len) = 0; /** * @brief Close the server and all client connections @@ -245,7 +247,8 @@ class IpcServer { // Receive message (blocks until complete message available, zero-copy for // SHM). A null data() means error/timeout; a non-null empty span is a // valid zero-length request. - auto request = receive(client_id); + uint64_t request_id = 0; + auto request = receive(client_id, request_id); if (request.data() == nullptr) { continue; } @@ -254,7 +257,7 @@ class IpcServer { // response, and skipping it would deadlock the waiting client. try { auto response = handler(client_id, request); - send(client_id, response.data(), response.size()); + send(client_id, request_id, response.data(), response.size()); } catch (const std::exception& e) { // A handler or send failure here is unrecoverable for this // request — e.g. a response larger than the ring can never be @@ -282,16 +285,16 @@ class IpcServer { * The reactor thread owns ALL ring/socket I/O: it reads each request and is * the sole caller of send(). It never blocks on the handler. For each request * it copies the bytes into a runtime-owned buffer, release()s the ring slot - * immediately, assigns a per-connection sequence number, and invokes the - * handler with a `respond` callback. The handler does decode/dispatch, any - * classification/ordering, and chooses its own thread; it calls `respond` - * exactly once when the result is ready (inline or later, from any thread). + * immediately, and invokes the handler with a `respond` callback. The handler + * does decode/dispatch, any classification/ordering, and chooses its own + * thread; it calls `respond` exactly once when the result is ready (inline or + * later, from any thread). * - * Responses are sent in per-connection request order (a small reorder stash - * keyed by the sequence number), so the wire stays FIFO with no request-id - * envelope. Because only the reactor calls send(), each response ring keeps + * Responses are sent in COMPLETION order, each carrying its request's echoed + * id — clients correlate by id, so a slow request never delays a fast one's + * response. Because only the reactor calls send(), each response ring keeps * its single-producer/lock-free property and writes are serial for free; the - * only lock is over the in-process stash, never over a ring. + * only lock is over the in-process completion queue, never over a ring. * * All scheduling policy — concurrency, ordering, and any inline-vs-deferred * fast path — lives in the handler, not here. A handler that responds inline @@ -299,48 +302,35 @@ class IpcServer { */ void run_reactor(const AsyncHandler& handler) { - struct Conn { - uint64_t next_send_seq = 0; // next sequence to release to the wire - std::map> stash; // completed-but-not-yet-in-order responses + struct Completion { + int client_id; + uint64_t request_id; + std::vector response; }; - std::mutex mtx; // guards `conns` only (never a ring) - std::unordered_map conns; // per-connection reorder state - std::unordered_map next_seq; // reactor-only: next sequence to assign - std::atomic inflight{ 0 }; // requests whose respond() has not fired + std::mutex mtx; // guards `completed` only (never a ring) + std::deque completed; + std::atomic inflight{ 0 }; // requests whose respond() has not fired - // True iff some connection has its next-expected response ready. Called - // (under mtx) inside the SHM wait's seq-latched window, so a response - // posted just before the futex_wait is not slept through. + // Called (under mtx) inside the SHM wait's seq-latched window, so a + // completion posted just before the futex_wait is not slept through. auto have_ready = [&]() -> bool { std::lock_guard lock(mtx); - for (auto& [client, conn] : conns) { - if (conn.stash.count(conn.next_send_seq) != 0) { - return true; - } - } - return false; + return !completed.empty(); }; - // Reactor-only: emit every response that is now next-in-sequence. Collect - // under the lock, then send() outside it — send() can block on ring - // backpressure and must never hold the stash mutex. + // Reactor-only: send every completed response. Collect under the lock, + // then send() outside it — send() can block on ring backpressure and + // must never hold the completion mutex. A send to a since-disconnected + // client fails harmlessly (client ids are never reused). auto drain_and_send = [&]() { - std::vector>> ready; + std::deque ready; { std::lock_guard lock(mtx); - for (auto& [client, conn] : conns) { - auto it = conn.stash.find(conn.next_send_seq); - while (it != conn.stash.end()) { - ready.emplace_back(client, std::move(it->second)); - conn.stash.erase(it); - conn.next_send_seq++; - it = conn.stash.find(conn.next_send_seq); - } - } + ready.swap(completed); } - for (auto& [client, bytes] : ready) { - send(client, bytes.data(), bytes.size()); + for (auto& c : ready) { + send(c.client_id, c.request_id, c.response.data(), c.response.size()); } }; @@ -353,7 +343,8 @@ class IpcServer { continue; } - auto request = receive(client_id); + uint64_t request_id = 0; + auto request = receive(client_id, request_id); if (request.data() == nullptr) { continue; } @@ -364,31 +355,32 @@ class IpcServer { auto buf = std::make_shared>(request.begin(), request.end()); release(client_id, request.size()); - uint64_t seq = next_seq[client_id]++; inflight.fetch_add(1, std::memory_order_relaxed); - // respond(): invoked exactly once, possibly on another thread. Stash - // the response (push) then notify the reactor to send it — push before - // notify so the reactor's have_ready predicate observes it and a SHM - // wake is never lost. Holds `buf` alive until invoked. Captures reactor - // locals by reference, valid because run_reactor does not return until - // inflight hits 0 (quiesce) and the final respond drives it there. - Respond respond = [this, client_id, seq, buf, &mtx, &conns, &inflight](std::vector response) { - { - std::lock_guard lock(mtx); - conns[client_id].stash.emplace(seq, std::move(response)); - } - inflight.fetch_sub(1, std::memory_order_release); - notify(); - }; + // respond(): invoked exactly once, possibly on another thread. Queue + // the completion (push) then notify the reactor to send it — push + // before notify so the reactor's have_ready predicate observes it and + // a SHM wake is never lost. Holds `buf` alive until invoked. Captures + // reactor locals by reference, valid because run_reactor does not + // return until inflight hits 0 (quiesce) and the final respond drives + // it there. + Respond respond = + [this, client_id, request_id, buf, &mtx, &completed, &inflight](std::vector response) { + { + std::lock_guard lock(mtx); + completed.push_back({ client_id, request_id, std::move(response) }); + } + inflight.fetch_sub(1, std::memory_order_release); + notify(); + }; handler(client_id, std::span(*buf), std::move(respond)); drain_and_send(); // emit immediately if the handler responded inline } // Quiesce before returning: in-flight handlers capture this frame's state - // (mtx, conns, inflight), so we must not unwind until every respond() has - // fired. + // (mtx, completed, inflight), so we must not unwind until every respond() + // has fired. while (inflight.load(std::memory_order_acquire) > 0) { drain_and_send(); wait_for_data_or_ready(10000000, have_ready); diff --git a/ipc-runtime/cpp/ipc_runtime/mpsc_shm_client.hpp b/ipc-runtime/cpp/ipc_runtime/mpsc_shm_client.hpp index 318c6bd53761..c718c3574874 100644 --- a/ipc-runtime/cpp/ipc_runtime/mpsc_shm_client.hpp +++ b/ipc-runtime/cpp/ipc_runtime/mpsc_shm_client.hpp @@ -83,43 +83,48 @@ class MpscShmClient : public IpcClient { return false; } - bool send(const void* data, size_t len, uint64_t timeout_ns) override + bool send(uint64_t request_id, const void* data, size_t len, uint64_t timeout_ns) override { if (!producer_.has_value()) { return false; } - // Claim space for length prefix + data - size_t total_size = sizeof(uint32_t) + len; + // Claim space for length prefix + request id + data + size_t total_size = sizeof(uint32_t) + FRAME_ID_SIZE + len; void* buf = producer_->claim(total_size, normalize_call_timeout(timeout_ns)); if (buf == nullptr) { return false; } - // Write length prefix + data - auto len_u32 = static_cast(len); + // Write length prefix + request id + data + auto len_u32 = static_cast(FRAME_ID_SIZE + len); std::memcpy(buf, &len_u32, sizeof(uint32_t)); - std::memcpy(static_cast(buf) + sizeof(uint32_t), data, len); + std::memcpy(static_cast(buf) + sizeof(uint32_t), &request_id, FRAME_ID_SIZE); + std::memcpy(static_cast(buf) + sizeof(uint32_t) + FRAME_ID_SIZE, data, len); // Publish (rings doorbell to wake server) producer_->publish(total_size); return true; } - std::span receive(uint64_t timeout_ns) override + std::span receive(uint64_t timeout_ns, uint64_t& request_id) override { if (!response_ring_.has_value()) { return {}; } - return ring_receive_msg(response_ring_.value(), normalize_call_timeout(timeout_ns)); + return ring_receive_msg(response_ring_.value(), normalize_call_timeout(timeout_ns), request_id); } + // Rings persist across occupants (slot reclaim / reattach), so leftover + // frames from a previous connection are expected and skipped by id. + bool may_have_stale_frames() const override { return true; } + void release(size_t message_size) override { if (!response_ring_.has_value()) { return; } - response_ring_->release(sizeof(uint32_t) + message_size); + response_ring_->release(sizeof(uint32_t) + FRAME_ID_SIZE + message_size); } void close() override diff --git a/ipc-runtime/cpp/ipc_runtime/mpsc_shm_server.hpp b/ipc-runtime/cpp/ipc_runtime/mpsc_shm_server.hpp index ee7af0edd465..e35444299206 100644 --- a/ipc-runtime/cpp/ipc_runtime/mpsc_shm_server.hpp +++ b/ipc-runtime/cpp/ipc_runtime/mpsc_shm_server.hpp @@ -107,7 +107,7 @@ class MpscShmServer : public IpcServer { return request_consumer_.has_value() && request_consumer_->has_data(); } - std::span receive(int client_id) override + std::span receive(int client_id, uint64_t& request_id) override { if (!request_consumer_.has_value() || client_id < 0 || static_cast(client_id) >= max_clients_) { return {}; @@ -121,17 +121,22 @@ class MpscShmServer : public IpcServer { std::memcpy(&msg_len, len_ptr, sizeof(uint32_t)); // A prefix larger than the send side could legally publish means the - // ring is corrupt — error out instead of waiting for it. - if (msg_len > MAX_FRAME_SIZE || msg_len > request_ring_size_ / 2 - sizeof(uint32_t)) { - throw std::runtime_error("MpscShmServer::receive: corrupt length prefix (" + std::to_string(msg_len) + - " bytes exceeds ring/frame limits)"); + // ring is corrupt — error out instead of waiting for it. A frame + // shorter than the request-id field means the peer speaks the id-less + // protocol. + if (msg_len > MAX_FRAME_SIZE || msg_len > request_ring_size_ / 2 - sizeof(uint32_t) || + msg_len < FRAME_ID_SIZE) { + throw std::runtime_error("MpscShmServer::receive: invalid length prefix (" + std::to_string(msg_len) + + " bytes) — corrupt ring or protocol mismatch"); } void* msg_ptr = request_consumer_->peek(static_cast(client_id), sizeof(uint32_t) + msg_len, 100000000); if (msg_ptr == nullptr) { return {}; } - return std::span(static_cast(msg_ptr) + sizeof(uint32_t), msg_len); + std::memcpy(&request_id, static_cast(msg_ptr) + sizeof(uint32_t), FRAME_ID_SIZE); + return std::span(static_cast(msg_ptr) + sizeof(uint32_t) + FRAME_ID_SIZE, + msg_len - FRAME_ID_SIZE); } void release(int client_id, size_t message_size) override @@ -139,15 +144,15 @@ class MpscShmServer : public IpcServer { if (!request_consumer_.has_value() || client_id < 0 || static_cast(client_id) >= max_clients_) { return; } - request_consumer_->release(static_cast(client_id), sizeof(uint32_t) + message_size); + request_consumer_->release(static_cast(client_id), sizeof(uint32_t) + FRAME_ID_SIZE + message_size); } - bool send(int client_id, const void* data, size_t len) override + bool send(int client_id, uint64_t request_id, const void* data, size_t len) override { if (client_id < 0 || static_cast(client_id) >= response_rings_.size()) { return false; } - return ring_send_msg(response_rings_[static_cast(client_id)], data, len, 100000000); + return ring_send_msg(response_rings_[static_cast(client_id)], request_id, data, len, 100000000); } void close() override diff --git a/ipc-runtime/cpp/ipc_runtime/shm.test.cpp b/ipc-runtime/cpp/ipc_runtime/shm.test.cpp index 3f599cf09d41..701696e77b47 100644 --- a/ipc-runtime/cpp/ipc_runtime/shm.test.cpp +++ b/ipc-runtime/cpp/ipc_runtime/shm.test.cpp @@ -35,7 +35,7 @@ TEST(ShmTest, SingleClientSmallRingHighVolume) constexpr size_t NUM_ITERATIONS = 10000000; // Sizing ensures that no matter that state of the internal ring buffer, we // can't deadlock. - constexpr size_t MAX_MSG_SIZE = (RING_SIZE / 2) - 4; + constexpr size_t MAX_MSG_SIZE = (RING_SIZE / 2) - 4 - FRAME_ID_SIZE; // Use short name for macOS compatibility (31-char limit) std::string wrap_test_shm = "shm_wrap_" + std::to_string(getpid()); @@ -56,7 +56,8 @@ TEST(ShmTest, SingleClientSmallRingHighVolume) continue; } - auto request_buf = server->receive(client_id); + uint64_t request_id = 0; + auto request_buf = server->receive(client_id, request_id); // std::cerr << "Server received " << request.size() << " bytes" << '\n'; if (request_buf.empty()) { @@ -83,7 +84,7 @@ TEST(ShmTest, SingleClientSmallRingHighVolume) } // Retry send until success. - while (!server->send(client_id, request.data(), request.size())) { + while (!server->send(client_id, request_id, request.data(), request.size())) { // Timeout - retry (response ring might be full) std::cerr << iter << " Server send size " << request.size() << " timeout, retrying..." << '\n'; dynamic_cast(server.get())->debug_dump(); @@ -104,63 +105,68 @@ TEST(ShmTest, SingleClientSmallRingHighVolume) std::mt19937 gen(rd()); std::uniform_int_distribution size_dist(1, MAX_MSG_SIZE); - // Store sizes for each iteration so receiver knows what to expect - std::vector iteration_sizes(NUM_ITERATIONS); - for (size_t i = 0; i < NUM_ITERATIONS; i++) { - iteration_sizes[i] = size_dist(gen); - // iteration_sizes[i] = MAX_MSG_SIZE - 1; + // Size of each request, indexed by request id. Ids are 1-based (id 0 is + // reserved for server-initiated frames — see constants.hpp), so the array + // has NUM_ITERATIONS + 1 slots with slot 0 unused. + std::vector request_sizes(NUM_ITERATIONS + 1); + for (size_t id = 1; id <= NUM_ITERATIONS; id++) { + request_sizes[id] = size_dist(gen); + // request_sizes[id] = MAX_MSG_SIZE - 1; } // Sender thread: continuously send requests std::thread sender_thread([&]() { std::vector send_buffer(MAX_MSG_SIZE); - for (size_t iter = 0; iter < NUM_ITERATIONS; iter++) { - size_t size = iteration_sizes[iter]; - // std::cerr << "Client: Iteration " << iter << ": sending " << size << " - // bytes" << '\n'; + for (size_t id = 1; id <= NUM_ITERATIONS; id++) { + size_t size = request_sizes[id]; + // std::cerr << "Client: id " << id << ": sending " << size << " bytes" << '\n'; - // Fill buffer with iteration-specific pattern - // First byte is iteration number (mod 256), rest is XOR pattern with - // offset - uint8_t iter_byte = static_cast(iter & 0xFF); + // Fill buffer with an id-specific pattern: first byte is the id + // (mod 256), rest is an XOR pattern with offset. + uint8_t id_byte = static_cast(id & 0xFF); for (size_t i = 0; i < size; i++) { - send_buffer[i] = static_cast((iter_byte ^ i) & 0xFF); + send_buffer[i] = static_cast((id_byte ^ i) & 0xFF); } - // Retry send until success - timeouts are expected under extreme load - while (!client->send(send_buffer.data(), size, 100000000)) { + // Retry send until success - timeouts are expected under extreme load. + // Explicit request id: the receiver thread runs concurrently, so the + // serial auto-id convenience API cannot be used. + while (!client->send(id, send_buffer.data(), size, 100000000)) { // Timeout - retry (ring might be full, server might be slow) - std::cerr << iter << " Client send size " << size << " timeout, retrying..." << '\n'; + std::cerr << id << " Client send size " << size << " timeout, retrying..." << '\n'; dynamic_cast(client.get())->debug_dump(); } } }); - // Receiver thread: continuously receive and validate responses + // Receiver thread: continuously receive and validate responses. The echoed + // request id selects the expected size and pattern, so validation is + // independent of response order. std::thread receiver_thread([&]() { - for (size_t iter = 0; iter < NUM_ITERATIONS; iter++) { - size_t expected_size = iteration_sizes[iter]; - + for (size_t n = 0; n < NUM_ITERATIONS; n++) { // Retry recv until success - timeouts are expected under extreme load std::span response; - while ((response = client->receive(100000000)).empty()) { - std::cerr << iter << " Client receive timeout, retrying..." << '\n'; + uint64_t rid = 0; + while ((response = client->receive(100000000, rid)).empty()) { + std::cerr << n << " Client receive timeout, retrying..." << '\n'; // Timeout - retry } - // std::cerr << "Client received response of " << response.size() << " - // bytes" << '\n'; + ASSERT_GE(rid, 1U); + ASSERT_LE(rid, NUM_ITERATIONS); + size_t expected_size = request_sizes[rid]; + // std::cerr << "Client received response of " << response.size() << " bytes" << '\n'; - ASSERT_EQ(response.size(), expected_size) << "Size mismatch at iteration " << iter; + ASSERT_EQ(response.size(), expected_size) << "Size mismatch for request id " << rid; - // Validate entire response - check iteration byte and pattern - uint8_t iter_byte = static_cast(iter & 0xFF); + // Validate entire response - check id byte and pattern + uint8_t id_byte = static_cast(rid & 0xFF); if (response.size() > 0) { - ASSERT_EQ(response[0], iter_byte) << "Iteration byte mismatch at iteration " << iter; + ASSERT_EQ(response[0], id_byte) << "Id byte mismatch for request id " << rid; for (size_t i = 0; i < response.size(); i++) { - uint8_t expected = static_cast((iter_byte ^ i) & 0xFF); + uint8_t expected = static_cast((id_byte ^ i) & 0xFF); if (response[i] != expected) { - FAIL() << "Data corruption at iteration " << iter << " offset " << i + FAIL() << "Data corruption for request id " << rid << " offset " << i << ": expected=" << (int)expected << " actual=" << (int)response[i]; } } @@ -250,7 +256,8 @@ TEST(ShmTest, TimeoutDoesNotWrapAbove4Seconds) EXPECT_LT(elapsed.count(), 4400); sender.join(); - auto request = server->receive(0); + uint64_t drain_id = 0; + auto request = server->receive(0, drain_id); if (!request.empty()) { server->release(0, request.size()); } @@ -278,7 +285,8 @@ TEST(ShmTest, RingRejectsCorruptLengthPrefix) std::memcpy(buf, &bogus_len, sizeof(bogus_len)); producer.publish(8); - EXPECT_THROW((void)ring_receive_msg(consumer, 10'000'000ULL), std::runtime_error); + uint64_t corrupt_id = 0; + EXPECT_THROW((void)ring_receive_msg(consumer, 10'000'000ULL, corrupt_id), std::runtime_error); SpscShm::unlink(ring_name); } @@ -310,13 +318,14 @@ TEST(ShmTest, MpscEchoTwoClients) if (client_id < 0) { continue; } - auto request_buf = server->receive(client_id); + uint64_t request_id = 0; + auto request_buf = server->receive(client_id, request_id); if (request_buf.empty()) { continue; } std::vector request(request_buf.begin(), request_buf.end()); server->release(client_id, request.size()); - while (!server->send(client_id, request.data(), request.size())) { + while (!server->send(client_id, request_id, request.data(), request.size())) { // Retry if the client's response ring is full. } } @@ -439,7 +448,7 @@ class ReactorTestPool { // the reactor must observe inside its futex-arm window (see MpscConsumer::notify // / wait_for_data). Pipeline N requests on one connection whose handlers sleep // LONGER for earlier indices, so completions arrive reversed: a lost wake would -// stall, and a missing reorder buffer would deliver out of order. +// stall. Responses arrive in completion order, correlated by echoed id. TEST(ShmTest, MpscReactorPipelinedConcurrencyAndOrder) { constexpr uint32_t N = 16; @@ -468,16 +477,18 @@ TEST(ShmTest, MpscReactorPipelinedConcurrencyAndOrder) ASSERT_TRUE(client->connect()) << "MPSC reactor client failed to connect"; auto t0 = std::chrono::steady_clock::now(); - for (uint32_t i = 0; i < N; i++) { - while (!client->send(&i, sizeof(i), 100'000'000ULL)) { + for (uint32_t id = 1; id <= N; id++) { + while (!client->send(id, &id, sizeof(id), 100'000'000ULL)) { // Retry on a transient full request ring. } } bool stalled = false; - for (uint32_t i = 0; i < N; i++) { + std::vector seen(N + 1, false); + for (uint32_t n = 0; n < N; n++) { std::span resp; + uint64_t rid = 0; size_t empties = 0; - while ((resp = client->receive(100'000'000ULL)).empty()) { + while ((resp = client->receive(100'000'000ULL, rid)).empty()) { if (++empties > 50) { // 5s grace — a lost wake shows up as a stall stalled = true; break; @@ -489,13 +500,22 @@ TEST(ShmTest, MpscReactorPipelinedConcurrencyAndOrder) ASSERT_EQ(resp.size(), sizeof(uint32_t)); uint32_t got = 0; std::memcpy(&got, resp.data(), sizeof(got)); - EXPECT_EQ(got, i) << "responses must arrive in per-connection request order"; + ASSERT_GE(rid, 1U); + ASSERT_LE(rid, N); + EXPECT_EQ(got, static_cast(rid)) << "response payload does not match its echoed request id"; + EXPECT_FALSE(seen[rid]) << "duplicate response for request id " << rid; + seen[rid] = true; client->release(resp.size()); } + if (!stalled) { + for (uint32_t id = 1; id <= N; id++) { + EXPECT_TRUE(seen[id]) << "request id " << id << " was never answered"; + } + } auto ms = std::chrono::duration_cast(std::chrono::steady_clock::now() - t0).count(); EXPECT_FALSE(stalled) << "receiver stalled — a completion wake was lost over MPSC"; - // Serial would be the sum of sleeps (~456ms); 8 workers should be far less. + // Serial would be the sum of sleeps (~440ms); 8 workers should be far less. EXPECT_LT(ms, 250) << "pipelined requests did not execute concurrently (took " << ms << "ms)"; client->close(); @@ -518,7 +538,7 @@ TEST(ShmTest, MpscSingleClientPipelinedFlood) { constexpr size_t RING_SIZE = 2UL * 1024; constexpr size_t NUM_ITERATIONS = 200000; - constexpr size_t MAX_MSG_SIZE = (RING_SIZE / 2) - 4; + constexpr size_t MAX_MSG_SIZE = (RING_SIZE / 2) - 4 - FRAME_ID_SIZE; constexpr size_t NUM_CLIENTS = 1; std::string base_name = "shm_mpscflood_" + std::to_string(getpid()); @@ -535,7 +555,8 @@ TEST(ShmTest, MpscSingleClientPipelinedFlood) if (client_id < 0) { continue; } - auto request_buf = server->receive(client_id); + uint64_t request_id = 0; + auto request_buf = server->receive(client_id, request_id); if (request_buf.empty()) { continue; } @@ -548,7 +569,7 @@ TEST(ShmTest, MpscSingleClientPipelinedFlood) break; } } - while (!server->send(client_id, request.data(), request.size())) { + while (!server->send(client_id, request_id, request.data(), request.size())) { // Response ring full - retry. } } @@ -562,9 +583,9 @@ TEST(ShmTest, MpscSingleClientPipelinedFlood) std::random_device rd; std::mt19937 gen(rd()); std::uniform_int_distribution size_dist(1, MAX_MSG_SIZE); - std::vector iteration_sizes(NUM_ITERATIONS); - for (size_t i = 0; i < NUM_ITERATIONS; i++) { - iteration_sizes[i] = size_dist(gen); + std::vector request_sizes(NUM_ITERATIONS + 1); + for (size_t id = 1; id <= NUM_ITERATIONS; id++) { + request_sizes[id] = size_dist(gen); } // Set by the receiver if it gives up; lets the sender abandon its send-retry @@ -573,13 +594,13 @@ TEST(ShmTest, MpscSingleClientPipelinedFlood) std::thread sender_thread([&]() { std::vector send_buffer(MAX_MSG_SIZE); - for (size_t iter = 0; iter < NUM_ITERATIONS && !abort_flood.load(std::memory_order_acquire); iter++) { - size_t size = iteration_sizes[iter]; - uint8_t iter_byte = static_cast(iter & 0xFF); + for (size_t id = 1; id <= NUM_ITERATIONS && !abort_flood.load(std::memory_order_acquire); id++) { + size_t size = request_sizes[id]; + uint8_t id_byte = static_cast(id & 0xFF); for (size_t i = 0; i < size; i++) { - send_buffer[i] = static_cast((iter_byte ^ i) & 0xFF); + send_buffer[i] = static_cast((id_byte ^ i) & 0xFF); } - while (!client->send(send_buffer.data(), size, 100000000)) { + while (!client->send(id, send_buffer.data(), size, 100000000)) { if (abort_flood.load(std::memory_order_acquire)) { return; } @@ -590,22 +611,26 @@ TEST(ShmTest, MpscSingleClientPipelinedFlood) std::atomic received{ 0 }; std::atomic stalled{ false }; std::thread receiver_thread([&]() { - for (size_t iter = 0; iter < NUM_ITERATIONS; iter++) { - size_t expected_size = iteration_sizes[iter]; + for (size_t n = 0; n < NUM_ITERATIONS; n++) { std::span response; + uint64_t rid = 0; size_t empties = 0; // 50 * 100ms = 5s grace; a lost message shows up as a stall here. - while ((response = client->receive(100000000)).empty()) { + while ((response = client->receive(100000000, rid)).empty()) { if (++empties > 50) { stalled.store(true); abort_flood.store(true, std::memory_order_release); return; } } - bool ok = response.size() == expected_size; - uint8_t iter_byte = static_cast(iter & 0xFF); + // The echoed id identifies the request, so validation is + // independent of response order. + bool ok = rid >= 1 && rid <= NUM_ITERATIONS; + size_t expected_size = ok ? request_sizes[rid] : 0; + ok = ok && response.size() == expected_size; + uint8_t id_byte = static_cast(rid & 0xFF); for (size_t i = 0; ok && i < response.size(); i++) { - ok = response[i] == static_cast((iter_byte ^ i) & 0xFF); + ok = response[i] == static_cast((id_byte ^ i) & 0xFF); } client->release(response.size()); if (!ok) { @@ -661,13 +686,14 @@ TEST(ShmTest, MpscSingleClientBurst) if (client_id < 0) { continue; } - auto request_buf = server->receive(client_id); + uint64_t request_id = 0; + auto request_buf = server->receive(client_id, request_id); if (request_buf.empty()) { continue; } std::vector request(request_buf.begin(), request_buf.end()); server->release(client_id, request.size()); - while (!server->send(client_id, request.data(), request.size())) { + while (!server->send(client_id, request_id, request.data(), request.size())) { // Response ring full - retry. } } @@ -680,22 +706,27 @@ TEST(ShmTest, MpscSingleClientBurst) size_t total_received = 0; bool stalled = false; + // The payload tag doubles as the request id (1-based, unique across rounds). + uint32_t next_tag = 1; for (size_t round = 0; round < NUM_ROUNDS && !stalled; round++) { // Fire the whole burst with no interleaved receive: BURST requests and // their responses are all in flight before we drain any. for (size_t i = 0; i < BURST; i++) { std::vector payload(MSG_SIZE, 0); - uint32_t tag = static_cast(round * BURST + i); + uint32_t tag = next_tag++; std::memcpy(payload.data(), &tag, sizeof(tag)); - while (!client->send(payload.data(), payload.size(), 1'000'000'000ULL)) { + while (!client->send(tag, payload.data(), payload.size(), 1'000'000'000ULL)) { // Ring full - retry. } } - // Drain the burst; responses come back in FIFO (send) order. + // Drain the burst; each response is paired to its request by the + // echoed id (the serve loop is serial here, but the contract is + // correlation, not order). for (size_t i = 0; i < BURST; i++) { std::span resp; + uint64_t rid = 0; size_t empties = 0; - while ((resp = client->receive(100'000'000ULL)).empty()) { + while ((resp = client->receive(100'000'000ULL, rid)).empty()) { if (++empties > 50) { // 5s grace stalled = true; break; @@ -706,7 +737,7 @@ TEST(ShmTest, MpscSingleClientBurst) } uint32_t tag = 0; std::memcpy(&tag, resp.data(), sizeof(tag)); - EXPECT_EQ(tag, static_cast(round * BURST + i)) << "lost/out-of-order at round " << round; + EXPECT_EQ(static_cast(tag), rid) << "payload does not match its echoed id at round " << round; EXPECT_EQ(resp.size(), MSG_SIZE); client->release(resp.size()); total_received++; @@ -724,6 +755,60 @@ TEST(ShmTest, MpscSingleClientBurst) server->close(); } +// A reclaimed slot's response ring can hold a leftover frame addressed to the +// previous occupant. The serial convenience receive() must recognise it by id, +// release it, and keep waiting for the real response — closing here would fail +// the new occupant's first call through no fault of its own. +TEST(ShmTest, SerialClientSkipsStaleFrameFromPreviousOccupant) +{ + constexpr size_t RING_SIZE = 4UL * 1024; + std::string base_name = "shm_stale_" + std::to_string(getpid()); + auto server = IpcServer::create_mpsc_shm(base_name, 1, RING_SIZE, RING_SIZE); + ASSERT_TRUE(server->listen()); + + std::atomic server_running{ true }; + std::thread server_thread([&]() { + while (server_running.load(std::memory_order_acquire)) { + server->accept(); + int client_id = server->wait_for_data(10000000); // 10ms + if (client_id < 0) { + continue; + } + uint64_t request_id = 0; + auto request = server->receive(client_id, request_id); + if (request.data() == nullptr) { + continue; + } + std::vector payload(request.begin(), request.end()); + server->release(client_id, payload.size()); + // Inject a leftover addressed to a previous occupant's id, THEN the + // real echo. The serial client must skip the first frame. + uint8_t junk[3] = { 0xBA, 0xAD, 0x01 }; + while (!server->send(client_id, 0xDEADBEEFULL, junk, sizeof(junk))) { + } + while (!server->send(client_id, request_id, payload.data(), payload.size())) { + } + } + }); + + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + auto client = IpcClient::create_mpsc_shm(base_name, 0); + ASSERT_TRUE(client->connect()); + + uint8_t msg[4] = { 1, 2, 3, 4 }; + ASSERT_TRUE(client->send(msg, sizeof(msg), 1'000'000'000ULL)); + auto resp = client->receive(2'000'000'000ULL); + ASSERT_EQ(resp.size(), sizeof(msg)) << "stale frame was not skipped"; + EXPECT_EQ(0, std::memcmp(resp.data(), msg, sizeof(msg))); + client->release(resp.size()); + + client->close(); + server_running.store(false); + server->request_shutdown(); + server_thread.join(); + server->close(); +} + TEST(ShmTest, MpscReactorMultiClientResponseRouting) { constexpr uint32_t NUM_CLIENTS = 4; @@ -739,9 +824,9 @@ TEST(ShmTest, MpscReactorMultiClientResponseRouting) server->run_reactor([&pool](int, std::span req, IpcServer::Respond respond) { std::vector r(req.begin(), req.end()); pool.enqueue([r = std::move(r), respond = std::move(respond)]() mutable { - uint32_t seq = 0; - std::memcpy(&seq, r.data() + sizeof(uint32_t), sizeof(uint32_t)); - std::this_thread::sleep_for(std::chrono::microseconds(20 * (seq % 8))); + uint32_t id = 0; + std::memcpy(&id, r.data() + sizeof(uint32_t), sizeof(uint32_t)); + std::this_thread::sleep_for(std::chrono::microseconds(20 * (id % 8))); respond(std::move(r)); }); }); @@ -749,7 +834,7 @@ TEST(ShmTest, MpscReactorMultiClientResponseRouting) std::this_thread::sleep_for(std::chrono::milliseconds(100)); std::atomic wrong_client{ 0 }; - std::atomic out_of_order{ 0 }; + std::atomic mispaired{ 0 }; std::atomic stalls{ 0 }; auto run_client = [&](uint32_t c) { @@ -758,15 +843,19 @@ TEST(ShmTest, MpscReactorMultiClientResponseRouting) stalls++; return; } - for (uint32_t s = 0; s < K; s++) { - uint32_t msg[2] = { c, s }; - while (!client->send(msg, sizeof(msg), 100'000'000ULL)) { + for (uint32_t id = 1; id <= K; id++) { + uint32_t msg[2] = { c, id }; + while (!client->send(id, msg, sizeof(msg), 100'000'000ULL)) { } } - for (uint32_t s = 0; s < K; s++) { + // Responses arrive in completion order; pair each to its request via + // the echoed id and require every request answered exactly once. + std::vector seen(K + 1, false); + for (uint32_t n = 0; n < K; n++) { std::span resp; + uint64_t rid = 0; size_t empties = 0; - while ((resp = client->receive(100'000'000ULL)).empty()) { + while ((resp = client->receive(100'000'000ULL, rid)).empty()) { if (++empties > 50) { stalls++; break; @@ -780,8 +869,10 @@ TEST(ShmTest, MpscReactorMultiClientResponseRouting) if (got[0] != c) { wrong_client++; } - if (got[1] != s) { - out_of_order++; + if (rid < 1 || rid > K || got[1] != static_cast(rid) || seen[rid]) { + mispaired++; + } else { + seen[rid] = true; } client->release(resp.size()); } @@ -801,7 +892,7 @@ TEST(ShmTest, MpscReactorMultiClientResponseRouting) server->close(); EXPECT_EQ(wrong_client.load(), 0) << "responses were routed to the wrong client"; - EXPECT_EQ(out_of_order.load(), 0) << "responses arrived out of order within a client"; + EXPECT_EQ(mispaired.load(), 0) << "a response did not pair with its request id exactly once"; EXPECT_EQ(stalls.load(), 0) << "a client stalled waiting for its responses"; } @@ -820,9 +911,9 @@ TEST(ShmTest, MpscReactorAutoClaimMultiClient) server->run_reactor([&pool](int, std::span req, IpcServer::Respond respond) { std::vector r(req.begin(), req.end()); pool.enqueue([r = std::move(r), respond = std::move(respond)]() mutable { - uint32_t seq = 0; - std::memcpy(&seq, r.data() + sizeof(uint32_t), sizeof(uint32_t)); - std::this_thread::sleep_for(std::chrono::microseconds(20 * (seq % 8))); + uint32_t id = 0; + std::memcpy(&id, r.data() + sizeof(uint32_t), sizeof(uint32_t)); + std::this_thread::sleep_for(std::chrono::microseconds(20 * (id % 8))); respond(std::move(r)); }); }); @@ -830,7 +921,7 @@ TEST(ShmTest, MpscReactorAutoClaimMultiClient) std::this_thread::sleep_for(std::chrono::milliseconds(100)); std::atomic wrong_client{ 0 }; - std::atomic out_of_order{ 0 }; + std::atomic mispaired{ 0 }; std::atomic stalls{ 0 }; auto run_client = [&](uint32_t c) { @@ -839,15 +930,19 @@ TEST(ShmTest, MpscReactorAutoClaimMultiClient) stalls++; return; } - for (uint32_t s = 0; s < K; s++) { - uint32_t msg[2] = { c, s }; - while (!client->send(msg, sizeof(msg), 100'000'000ULL)) { + for (uint32_t id = 1; id <= K; id++) { + uint32_t msg[2] = { c, id }; + while (!client->send(id, msg, sizeof(msg), 100'000'000ULL)) { } } - for (uint32_t s = 0; s < K; s++) { + // Responses arrive in completion order; pair each to its request via + // the echoed id and require every request answered exactly once. + std::vector seen(K + 1, false); + for (uint32_t n = 0; n < K; n++) { std::span resp; + uint64_t rid = 0; size_t empties = 0; - while ((resp = client->receive(100'000'000ULL)).empty()) { + while ((resp = client->receive(100'000'000ULL, rid)).empty()) { if (++empties > 50) { stalls++; break; @@ -861,8 +956,10 @@ TEST(ShmTest, MpscReactorAutoClaimMultiClient) if (got[0] != c) { wrong_client++; } - if (got[1] != s) { - out_of_order++; + if (rid < 1 || rid > K || got[1] != static_cast(rid) || seen[rid]) { + mispaired++; + } else { + seen[rid] = true; } client->release(resp.size()); } @@ -882,7 +979,7 @@ TEST(ShmTest, MpscReactorAutoClaimMultiClient) server->close(); EXPECT_EQ(wrong_client.load(), 0) << "self-allocated clients aliased onto a shared slot"; - EXPECT_EQ(out_of_order.load(), 0) << "responses arrived out of order within a client"; + EXPECT_EQ(mispaired.load(), 0) << "a response did not pair with its request id exactly once"; EXPECT_EQ(stalls.load(), 0) << "a client stalled"; } diff --git a/ipc-runtime/cpp/ipc_runtime/shm/README.md b/ipc-runtime/cpp/ipc_runtime/shm/README.md index 45a4bba41231..ac15655f0514 100644 --- a/ipc-runtime/cpp/ipc_runtime/shm/README.md +++ b/ipc-runtime/cpp/ipc_runtime/shm/README.md @@ -185,8 +185,8 @@ public: ## Usage Examples These use the message framing helpers from `../shm_common.hpp` -(`ring_send_msg` / `ring_receive_msg`), which add a 4-byte length prefix and -take care of the matched claim/peek sizing. +(`ring_send_msg` / `ring_receive_msg`), which add the 4-byte length prefix +plus the 8-byte request id and take care of the matched claim/peek sizing. **Producer process:** ```cpp @@ -200,7 +200,7 @@ int main() { std::string msg = "hello from producer"; while (true) { // Blocks up to 1s for ring space; false on timeout. - ipc::ring_send_msg(tx, msg.data(), msg.size(), 1'000'000'000); + ipc::ring_send_msg(tx, /*request_id=*/1, msg.data(), msg.size(), 1'000'000'000); } } ``` @@ -215,12 +215,13 @@ int main() { while (true) { // Blocks up to 1s for a whole message; empty data() on timeout. - auto msg = ipc::ring_receive_msg(rx, 1'000'000'000); + uint64_t request_id = 0; + auto msg = ipc::ring_receive_msg(rx, 1'000'000'000, request_id); if (msg.data() == nullptr) { continue; // timeout } std::cout << "Received: " << std::string(msg.begin(), msg.end()) << "\n"; - rx.release(4 + msg.size()); // length prefix + payload + rx.release(4 + 8 + msg.size()); // length prefix + request id + payload } } ``` diff --git a/ipc-runtime/cpp/ipc_runtime/shm_client.hpp b/ipc-runtime/cpp/ipc_runtime/shm_client.hpp index 79722cfe1b49..66a650135c33 100644 --- a/ipc-runtime/cpp/ipc_runtime/shm_client.hpp +++ b/ipc-runtime/cpp/ipc_runtime/shm_client.hpp @@ -74,28 +74,32 @@ class ShmClient : public IpcClient { return false; } - bool send(const void* data, size_t len, uint64_t timeout_ns) override + bool send(uint64_t request_id, const void* data, size_t len, uint64_t timeout_ns) override { if (!request_ring_.has_value()) { return false; } - return ring_send_msg(request_ring_.value(), data, len, normalize_call_timeout(timeout_ns)); + return ring_send_msg(request_ring_.value(), request_id, data, len, normalize_call_timeout(timeout_ns)); } - std::span receive(uint64_t timeout_ns) override + std::span receive(uint64_t timeout_ns, uint64_t& request_id) override { if (!response_ring_.has_value()) { return {}; } - return ring_receive_msg(response_ring_.value(), normalize_call_timeout(timeout_ns)); + return ring_receive_msg(response_ring_.value(), normalize_call_timeout(timeout_ns), request_id); } + // Rings persist across occupants (slot reclaim / reattach), so leftover + // frames from a previous connection are expected and skipped by id. + bool may_have_stale_frames() const override { return true; } + void release(size_t message_size) override { if (!response_ring_.has_value()) { return; } - response_ring_->release(sizeof(uint32_t) + message_size); + response_ring_->release(sizeof(uint32_t) + FRAME_ID_SIZE + message_size); } void close() override diff --git a/ipc-runtime/cpp/ipc_runtime/shm_common.hpp b/ipc-runtime/cpp/ipc_runtime/shm_common.hpp index ccc273210bd9..4dc12cc885a7 100644 --- a/ipc-runtime/cpp/ipc_runtime/shm_common.hpp +++ b/ipc-runtime/cpp/ipc_runtime/shm_common.hpp @@ -10,27 +10,34 @@ namespace ipc { -inline bool ring_send_msg(SpscShm& ring, const void* data, size_t len, uint64_t timeout_ns) +/** + * Ring messages carry the ipc-runtime frame: [4B length][8B request id][payload], + * where the length counts the id plus the payload. The id is client-assigned on + * requests and echoed by the server on responses; receivers use it to correlate, + * so there is no ordering contract between frames. + */ +inline bool ring_send_msg(SpscShm& ring, uint64_t request_id, const void* data, size_t len, uint64_t timeout_ns) { // Prevent sending messages larger than half the ring buffer capacity. // This simplifies wrap-around logic. - if (len > ring.capacity() / 2 - 4) { + if (FRAME_ID_SIZE + len > ring.capacity() / 2 - 4) { throw std::runtime_error("ring_send_msg: message too large for ring " "buffer, must be <= half capacity minus 4 bytes"); } - // Atomic send: claim space for entire message (length + data) - size_t total_size = 4 + len; + // Atomic send: claim space for entire message (length + id + data) + size_t total_size = 4 + FRAME_ID_SIZE + len; void* buf = ring.claim(total_size, timeout_ns); if (buf == nullptr) { return false; // Timeout or no space - nothing published yet (atomic // failure) } - // Write length prefix and message data together - auto len_u32 = static_cast(len); + // Write length prefix, request id, and message data together + auto len_u32 = static_cast(FRAME_ID_SIZE + len); std::memcpy(buf, &len_u32, 4); - std::memcpy(static_cast(buf) + 4, data, len); + std::memcpy(static_cast(buf) + 4, &request_id, FRAME_ID_SIZE); + std::memcpy(static_cast(buf) + 4 + FRAME_ID_SIZE, data, len); // Publish entire message atomically ring.publish(total_size); @@ -38,7 +45,7 @@ inline bool ring_send_msg(SpscShm& ring, const void* data, size_t len, uint64_t return true; } -inline std::span ring_receive_msg(SpscShm& ring, uint64_t timeout_ns) +inline std::span ring_receive_msg(SpscShm& ring, uint64_t timeout_ns, uint64_t& request_id) { // Peek the length prefix (4 bytes) void* len_ptr = ring.peek(4, timeout_ns); @@ -52,10 +59,11 @@ inline std::span ring_receive_msg(SpscShm& ring, uint64_t timeout // Validate before waiting on the claimed size: the send side can never // legally publish more than capacity/2 - 4 bytes, so a larger prefix - // means the ring is corrupt. Waiting would only ever time out. - if (msg_len > MAX_FRAME_SIZE || msg_len > ring.capacity() / 2 - 4) { - throw std::runtime_error("ring_receive_msg: corrupt length prefix (" + std::to_string(msg_len) + - " bytes exceeds ring/frame limits)"); + // means the ring is corrupt — and a frame shorter than the request-id + // field means the peer speaks the id-less protocol. + if (msg_len > MAX_FRAME_SIZE || msg_len > ring.capacity() / 2 - 4 || msg_len < FRAME_ID_SIZE) { + throw std::runtime_error("ring_receive_msg: invalid length prefix (" + std::to_string(msg_len) + + " bytes) — corrupt ring or protocol mismatch"); } // Now peek the message data @@ -64,8 +72,10 @@ inline std::span ring_receive_msg(SpscShm& ring, uint64_t timeout return {}; // Timeout } - // Return span directly into ring buffer (zero-copy!) - return std::span(static_cast(msg_ptr) + 4, msg_len); + std::memcpy(&request_id, static_cast(msg_ptr) + 4, FRAME_ID_SIZE); + + // Return payload span directly into ring buffer (zero-copy!) + return std::span(static_cast(msg_ptr) + 4 + FRAME_ID_SIZE, msg_len - FRAME_ID_SIZE); } } // namespace ipc diff --git a/ipc-runtime/cpp/ipc_runtime/shm_server.hpp b/ipc-runtime/cpp/ipc_runtime/shm_server.hpp index 3df6851e3bef..f614db58742b 100644 --- a/ipc-runtime/cpp/ipc_runtime/shm_server.hpp +++ b/ipc-runtime/cpp/ipc_runtime/shm_server.hpp @@ -80,13 +80,13 @@ class ShmServer : public IpcServer { return -1; // Timeout } - std::span receive([[maybe_unused]] int client_id) override + std::span receive([[maybe_unused]] int client_id, uint64_t& request_id) override { if (!request_ring_.has_value()) { return {}; } // TODO: Plumb timeout. - return ring_receive_msg(request_ring_.value(), 100000000); // 100ms timeout + return ring_receive_msg(request_ring_.value(), 100000000, request_id); // 100ms timeout } void release([[maybe_unused]] int client_id, size_t message_size) override @@ -94,15 +94,15 @@ class ShmServer : public IpcServer { if (!request_ring_.has_value()) { return; } - request_ring_->release(sizeof(uint32_t) + message_size); + request_ring_->release(sizeof(uint32_t) + FRAME_ID_SIZE + message_size); } - bool send([[maybe_unused]] int client_id, const void* data, size_t len) override + bool send([[maybe_unused]] int client_id, uint64_t request_id, const void* data, size_t len) override { if (!response_ring_.has_value()) { return false; } - return ring_send_msg(response_ring_.value(), data, len, 100000000); + return ring_send_msg(response_ring_.value(), request_id, data, len, 100000000); } void close() override diff --git a/ipc-runtime/cpp/ipc_runtime/signal_handlers.cpp b/ipc-runtime/cpp/ipc_runtime/signal_handlers.cpp index 52cf6aecd650..41ec1f9db100 100644 --- a/ipc-runtime/cpp/ipc_runtime/signal_handlers.cpp +++ b/ipc-runtime/cpp/ipc_runtime/signal_handlers.cpp @@ -137,6 +137,11 @@ void install_default_signal_handlers(IpcServer& server) (void)std::signal(SIGINT, graceful_shutdown_handler); (void)std::signal(SIGBUS, fatal_error_handler); (void)std::signal(SIGSEGV, fatal_error_handler); + // A client that disconnects with responses still in flight must produce + // EPIPE on the server's send(), never a process-killing SIGPIPE. send() + // already passes MSG_NOSIGNAL where available; this covers every other + // write to a peer-closed fd. + (void)std::signal(SIGPIPE, SIG_IGN); setup_parent_death_monitoring(); } diff --git a/ipc-runtime/cpp/ipc_runtime/signal_handlers.hpp b/ipc-runtime/cpp/ipc_runtime/signal_handlers.hpp index 83f0442d78a3..abd935e02c8c 100644 --- a/ipc-runtime/cpp/ipc_runtime/signal_handlers.hpp +++ b/ipc-runtime/cpp/ipc_runtime/signal_handlers.hpp @@ -8,6 +8,8 @@ * (graceful drain; the run() loop exits on its next poll iteration) * - SIGBUS / SIGSEGV → best-effort unlink of the server's socket/SHM * files (cached at install time) + _Exit(128 + sig) + * - SIGPIPE → SIG_IGN (a peer-closed fd yields EPIPE from write/send + * instead of killing the process) * - Parent-process death watch via prctl(PR_SET_PDEATHSIG) on Linux * and a kqueue NOTE_EXIT watcher on macOS — so spawn-and-forget * services die with their parent rather than turning into orphans. diff --git a/ipc-runtime/cpp/ipc_runtime/socket.test.cpp b/ipc-runtime/cpp/ipc_runtime/socket.test.cpp index c9e1cd48cc25..4c4b4eed7820 100644 --- a/ipc-runtime/cpp/ipc_runtime/socket.test.cpp +++ b/ipc-runtime/cpp/ipc_runtime/socket.test.cpp @@ -6,6 +6,7 @@ #include #include #include +#include #include #include #include @@ -84,10 +85,10 @@ class TestPool { }; // run_reactor() must (a) execute pipelined requests on one connection -// concurrently across the pool, and (b) still deliver responses in -// per-connection request order even when handlers complete out of order. The -// handler sleeps LONGER for earlier indices, so completions arrive roughly -// reversed — the reorder buffer has to hold them until each is next-in-sequence. +// concurrently across the pool, and (b) send each response as its handler +// completes, correlated by echoed request id — there is no FIFO contract. The +// handler sleeps LONGER for earlier ids, so completions (and responses) arrive +// roughly reversed. TEST(SocketTest, ReactorPipelinedConcurrencyAndOrder) { std::string path = test_socket_path("reactor"); @@ -113,20 +114,46 @@ TEST(SocketTest, ReactorPipelinedConcurrencyAndOrder) ASSERT_TRUE(client->connect()); auto t0 = std::chrono::steady_clock::now(); - for (uint32_t i = 0; i < N; i++) { - ASSERT_TRUE(client->send(&i, sizeof(i), 1'000'000'000ULL)); + // Request ids are 1-based (id 0 is reserved for server-initiated frames — + // see constants.hpp); the id doubles as the payload. + for (uint32_t id = 1; id <= N; id++) { + ASSERT_TRUE(client->send(id, &id, sizeof(id), 1'000'000'000ULL)); } - for (uint32_t i = 0; i < N; i++) { - auto resp = client->receive(5'000'000'000ULL); + // Responses arrive in completion order (earlier indices sleep longer, so + // they arrive out of send order); the echoed request id pairs each frame + // with its request. + std::vector seen(N + 1, false); + bool in_send_order = true; + uint64_t prev_rid = 0; + for (uint32_t n = 0; n < N; n++) { + uint64_t rid = 0; + auto resp = client->receive(5'000'000'000ULL, rid); ASSERT_EQ(resp.size(), sizeof(uint32_t)); uint32_t got = 0; std::memcpy(&got, resp.data(), sizeof(got)); - EXPECT_EQ(got, i) << "responses must arrive in per-connection request order"; + ASSERT_GE(rid, 1U); + ASSERT_LE(rid, N); + EXPECT_EQ(got, static_cast(rid)) << "response payload does not match its echoed request id"; + EXPECT_FALSE(seen[rid]) << "duplicate response for request id " << rid; + seen[rid] = true; + // Ids are sent in increasing order and each arrives exactly once, so a + // monotonically increasing drain means FIFO arrival. + if (rid < prev_rid) { + in_send_order = false; + } + prev_rid = rid; client->release(resp.size()); } auto ms = std::chrono::duration_cast(std::chrono::steady_clock::now() - t0).count(); - // Serial execution would be the sum of all sleeps (~456ms). With 8 workers + for (uint32_t id = 1; id <= N; id++) { + EXPECT_TRUE(seen[id]) << "request id " << id << " was never answered"; + } + // Reversed sleeps guarantee out-of-order completions; a fully in-order + // arrival would mean responses are being re-serialized somewhere. + EXPECT_FALSE(in_send_order) << "responses arrived strictly in send order — head-of-line blocking is back?"; + + // Serial execution would be the sum of all sleeps (~440ms). With 8 workers // it should be a small multiple of the longest single sleep; allow headroom. EXPECT_LT(ms, 250) << "pipelined requests did not execute concurrently (took " << ms << "ms)"; @@ -268,4 +295,243 @@ TEST(SocketTest, ClientRejectsOversizedLengthPrefix) ::unlink(path.c_str()); } +// A connection that dies with a request still in flight leaves a late respond(). Client ids are +// never reused, so that response has nowhere valid to go — the reactor must drop it, and the next +// connection (a fresh id) must see only its own frames. Positional clients (the TS AsyncApi) +// depend on this: a single leaked frame shifts every subsequent response onto the wrong caller. +// +// Protocol: request = [tag]; response = [tag, client_id]. Tag 'A' defers its response behind a +// test-controlled gate (scripted completion order, no sleeps-as-sync); any other tag responds +// inline. The client_id echo pins the never-reused-id invariant directly. +// Sockets have no ring reuse: a frame whose id matches nothing the client sent +// means the correlation is genuinely broken, and the serial convenience +// receive() must close rather than skip (contrast the SHM stale-frame test). +TEST(SocketTest, SerialClientClosesOnForeignFrame) +{ + std::string path = test_socket_path("foreign"); + auto server = IpcServer::create_socket(path, 2); + ASSERT_TRUE(server->listen()); + + std::atomic server_running{ true }; + std::thread server_thread([&]() { + while (server_running.load(std::memory_order_acquire)) { + server->accept(); + int client_id = server->wait_for_data(10000000); // 10ms + if (client_id < 0) { + continue; + } + uint64_t request_id = 0; + auto request = server->receive(client_id, request_id); + if (request.data() == nullptr) { + continue; + } + std::vector payload(request.begin(), request.end()); + server->release(client_id, payload.size()); + // Respond with a wrong id first: over UDS this is a protocol error, + // not a leftover, and the client must refuse the stream. + uint8_t junk[3] = { 0xBA, 0xAD, 0x02 }; + server->send(client_id, 0xDEADBEEFULL, junk, sizeof(junk)); + server->send(client_id, request_id, payload.data(), payload.size()); + } + }); + + auto client = IpcClient::create_socket(path); + ASSERT_TRUE(client->connect()); + + uint8_t msg[4] = { 5, 6, 7, 8 }; + ASSERT_TRUE(client->send(msg, sizeof(msg), 1'000'000'000ULL)); + auto resp = client->receive(2'000'000'000ULL); + EXPECT_EQ(resp.data(), nullptr) << "foreign frame over UDS must fail the call, not be skipped"; + + client->close(); + server_running.store(false); + server->request_shutdown(); + server_thread.join(); + server->close(); +} + +TEST(SocketTest, ReactorDropsStaleResponsesAndNeverReusesIds) +{ + std::string path = test_socket_path("staleresp"); + auto server = IpcServer::create_socket(path, 4); + ASSERT_TRUE(server->listen()); + + std::mutex gate_m; + std::condition_variable gate_cv; + bool gate_open = false; + auto release_gate = [&] { + { + std::lock_guard lock(gate_m); + gate_open = true; + } + gate_cv.notify_all(); + }; + + std::promise a_request_seen; // fulfilled with the client_id that sent 'A' + TestPool pool(2); + + std::thread server_thread([&] { + server->run_reactor([&](int client_id, std::span req, IpcServer::Respond respond) { + uint8_t tag = req[0]; + if (tag == 'A') { + a_request_seen.set_value(client_id); + pool.enqueue([&gate_m, &gate_cv, &gate_open, client_id, respond = std::move(respond)]() mutable { + std::unique_lock lock(gate_m); + gate_cv.wait(lock, [&] { return gate_open; }); + respond({ 'A', static_cast(client_id) }); + }); + } else { + respond({ tag, static_cast(client_id) }); + } + }); + }); + + // Connection A: send 'A' (its response is now in flight behind the gate), then vanish. + auto client_a = IpcClient::create_socket(path); + ASSERT_TRUE(client_a->connect()); + uint8_t tag_a = 'A'; + ASSERT_TRUE(client_a->send(&tag_a, 1, 1'000'000'000ULL)); + auto a_seen = a_request_seen.get_future(); + ASSERT_EQ(a_seen.wait_for(std::chrono::seconds(2)), std::future_status::ready) << "server never saw A's request"; + int a_id = a_seen.get(); + client_a->close(); + + // Let the reactor observe A's EOF before B connects — the window in which A's id would be + // freed for reuse if ids were recycled. + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + + // Connection B: sends 'B' (inline response), THEN A's zombie completes. + auto client_b = IpcClient::create_socket(path); + ASSERT_TRUE(client_b->connect()); + uint8_t tag_b = 'B'; + ASSERT_TRUE(client_b->send(&tag_b, 1, 1'000'000'000ULL)); + std::this_thread::sleep_for(std::chrono::milliseconds(50)); // let B's response reach the stash + release_gate(); + + auto first = client_b->receive(2'000'000'000ULL); + ASSERT_EQ(first.size(), 2U) << "no response frame reached connection B"; + uint8_t first_tag = first[0]; + uint8_t first_id = first[1]; + client_b->release(first.size()); + + EXPECT_EQ(first_tag, 'B') << "connection B's first response frame carries the dead connection A's payload " + "(tag '" + << static_cast(first_tag) << "', id " << int(first_id) + << ") — a stale response was delivered across connections"; + EXPECT_NE(int(first_id), a_id) << "client id was reused across connections"; + + // And there must be exactly one frame: a leaked zombie shifts B's real response into a + // second frame (which a positional client would hand to the NEXT caller). + auto extra = client_b->receive(300'000'000ULL); + EXPECT_TRUE(extra.empty()) << "extra frame leaked to connection B (tag '" + << static_cast(extra.empty() ? '?' : extra[0]) << "')"; + if (!extra.empty()) { + client_b->release(extra.size()); + } + + client_b->close(); + server->request_shutdown(); + server_thread.join(); + server->close(); +} + +// Control for the scenario above: when the first connection's response completes and is read +// BEFORE it disconnects, the second connection (fresh id) sees exactly its own response. Pins the +// invariant and validates the harness. +TEST(SocketTest, ReactorSequentialConnectionsAreIndependent) +{ + std::string path = test_socket_path("seq_conns"); + auto server = IpcServer::create_socket(path, 4); + ASSERT_TRUE(server->listen()); + + std::thread server_thread([&] { + server->run_reactor([&](int client_id, std::span req, IpcServer::Respond respond) { + respond({ req[0], static_cast(client_id) }); + }); + }); + + auto client_a = IpcClient::create_socket(path); + ASSERT_TRUE(client_a->connect()); + uint8_t tag_a = 'A'; + ASSERT_TRUE(client_a->send(&tag_a, 1, 1'000'000'000ULL)); + auto resp_a = client_a->receive(2'000'000'000ULL); + ASSERT_EQ(resp_a.size(), 2U); + EXPECT_EQ(resp_a[0], 'A'); + uint8_t a_id = resp_a[1]; + client_a->release(resp_a.size()); + client_a->close(); + + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + + auto client_b = IpcClient::create_socket(path); + ASSERT_TRUE(client_b->connect()); + uint8_t tag_b = 'B'; + ASSERT_TRUE(client_b->send(&tag_b, 1, 1'000'000'000ULL)); + auto resp_b = client_b->receive(2'000'000'000ULL); + ASSERT_EQ(resp_b.size(), 2U); + EXPECT_EQ(resp_b[0], 'B'); + EXPECT_NE(resp_b[1], a_id) << "client id was reused across connections"; + client_b->release(resp_b.size()); + auto extra = client_b->receive(300'000'000ULL); + EXPECT_TRUE(extra.empty()); + + client_b->close(); + server->request_shutdown(); + server_thread.join(); + server->close(); +} + +// The reactor must survive a client that disconnects with responses still in flight: the +// reactor drops the late responses (their connection's state is gone) instead of writing them +// to the dead fd, and any +// write that does hit a peer-closed fd yields EPIPE (MSG_NOSIGNAL / SO_NOSIGPIPE), never a +// process-killing SIGPIPE. NOTE: an in-process peer-closed write can be absorbed by kernel +// buffering, so this test alone cannot prove SIGPIPE immunity — the cross-process guard is +// yarn-project/world-state's wsdb churn test, where the server lives in its own process. +TEST(SocketTest, ReactorSurvivesResponseToDeadClient) +{ + std::string path = test_socket_path("sigpipe"); + auto server = IpcServer::create_socket(path, 4); + ASSERT_TRUE(server->listen()); + + std::mutex gate_m; + std::condition_variable gate_cv; + bool gate_open = false; + TestPool pool(2); + + std::thread server_thread([&] { + server->run_reactor([&](int, std::span req, IpcServer::Respond respond) { + std::vector big(64 * 1024, req[0]); // big frames: force multiple send() calls + pool.enqueue([&gate_m, &gate_cv, &gate_open, big = std::move(big), respond = std::move(respond)]() mutable { + std::unique_lock lock(gate_m); + gate_cv.wait(lock, [&] { return gate_open; }); + respond(std::move(big)); + }); + }); + }); + + // Pipeline several requests, then vanish without reading anything. + auto client = IpcClient::create_socket(path); + ASSERT_TRUE(client->connect()); + for (uint8_t t = 0; t < 4; t++) { + ASSERT_TRUE(client->send(&t, 1, 1'000'000'000ULL)); + } + std::this_thread::sleep_for(std::chrono::milliseconds(100)); // let the reactor ingest all four + client->close(); + + // Release all four responses; the reactor must drop or fail them without dying. + { + std::lock_guard lock(gate_m); + gate_open = true; + } + gate_cv.notify_all(); + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + + // If we are still alive, the server survived its client's mid-flight death. + server->request_shutdown(); + server_thread.join(); + server->close(); + SUCCEED(); +} + } // namespace diff --git a/ipc-runtime/cpp/ipc_runtime/socket_client.cpp b/ipc-runtime/cpp/ipc_runtime/socket_client.cpp index c99cd8c8e9e6..3adde5e9f049 100644 --- a/ipc-runtime/cpp/ipc_runtime/socket_client.cpp +++ b/ipc-runtime/cpp/ipc_runtime/socket_client.cpp @@ -119,7 +119,7 @@ int SocketClient::recv_exact(void* buf, size_t len, bool& partial) return 1; } -bool SocketClient::send(const void* data, size_t len, uint64_t timeout_ns) +bool SocketClient::send(uint64_t request_id, const void* data, size_t len, uint64_t timeout_ns) { if (fd_ < 0) { errno = EINVAL; @@ -132,11 +132,12 @@ bool SocketClient::send(const void* data, size_t len, uint64_t timeout_ns) apply_timeout(SO_SNDTIMEO, applied_send_timeout_ns_, timeout_ns); - // Send length prefix (4 bytes, little-endian), then message data, - // looping on partial writes. - auto msg_len = static_cast(len); + // Send length prefix (4 bytes, little-endian), request id (8 bytes, + // little-endian), then message data, looping on partial writes. + auto msg_len = static_cast(FRAME_ID_SIZE + len); bool partial = false; - if (send_exact(&msg_len, sizeof(msg_len), partial) != 1 || send_exact(data, len, partial) != 1) { + if (send_exact(&msg_len, sizeof(msg_len), partial) != 1 || send_exact(&request_id, FRAME_ID_SIZE, partial) != 1 || + send_exact(data, len, partial) != 1) { if (partial) { // Part of the frame is on the wire — the stream is desynced and // unusable. Close rather than silently corrupting later frames. @@ -147,7 +148,7 @@ bool SocketClient::send(const void* data, size_t len, uint64_t timeout_ns) return true; } -std::span SocketClient::receive(uint64_t timeout_ns) +std::span SocketClient::receive(uint64_t timeout_ns, uint64_t& request_id) { if (fd_ < 0) { return {}; @@ -166,12 +167,22 @@ std::span SocketClient::receive(uint64_t timeout_ns) return {}; } - // A corrupt/malicious prefix must not drive the allocation below. - if (msg_len > MAX_FRAME_SIZE) { + // A corrupt/malicious prefix must not drive the allocation below. A frame + // shorter than the request-id field means the peer speaks the id-less + // protocol — close rather than misparse. + if (msg_len > MAX_FRAME_SIZE || msg_len < FRAME_ID_SIZE) { close_internal(); return {}; } + // Read the echoed request id (8 bytes, little-endian). + request_id = 0; + if (recv_exact(&request_id, FRAME_ID_SIZE, partial) != 1) { + close_internal(); + return {}; + } + msg_len -= FRAME_ID_SIZE; + // Ensure buffer is large enough. Keep at least one byte so data() is // non-null for zero-length messages (null data() signals failure). if (recv_buffer_.size() < msg_len || recv_buffer_.empty()) { diff --git a/ipc-runtime/cpp/ipc_runtime/socket_client.hpp b/ipc-runtime/cpp/ipc_runtime/socket_client.hpp index 8d495adf3092..0844f72230a6 100644 --- a/ipc-runtime/cpp/ipc_runtime/socket_client.hpp +++ b/ipc-runtime/cpp/ipc_runtime/socket_client.hpp @@ -29,8 +29,8 @@ class SocketClient : public IpcClient { SocketClient& operator=(SocketClient&&) = delete; bool connect() override; - bool send(const void* data, size_t len, uint64_t timeout_ns) override; - std::span receive(uint64_t timeout_ns) override; + bool send(uint64_t request_id, const void* data, size_t len, uint64_t timeout_ns) override; + std::span receive(uint64_t timeout_ns, uint64_t& request_id) override; void release(size_t message_size) override; void close() override; diff --git a/ipc-runtime/cpp/ipc_runtime/socket_server.cpp b/ipc-runtime/cpp/ipc_runtime/socket_server.cpp index 2d5995fa606d..e284425551a1 100644 --- a/ipc-runtime/cpp/ipc_runtime/socket_server.cpp +++ b/ipc-runtime/cpp/ipc_runtime/socket_server.cpp @@ -4,6 +4,7 @@ #include #include #include +#include #include #include #include @@ -48,13 +49,14 @@ void SocketServer::close() void SocketServer::close_internal() { // Close all client connections - for (int fd : client_fds_) { + for (const auto& [client_id, fd] : client_fds_) { if (fd >= 0) { ::close(fd); } } client_fds_.clear(); fd_to_client_id_.clear(); + recv_buffers_.clear(); num_clients_ = 0; if (wake_read_fd_ >= 0) { @@ -80,19 +82,6 @@ void SocketServer::close_internal() ::unlink(socket_path_.c_str()); } -int SocketServer::find_free_slot() -{ - // Look for existing free slot - for (size_t i = 0; i < client_fds_.size(); i++) { - if (client_fds_[i] == -1) { - return static_cast(i); - } - } - - // No free slot found, allocate new one at end - return static_cast(client_fds_.size()); -} - bool SocketServer::setup_wake_pipe() { int fds[2]; @@ -151,10 +140,10 @@ void SocketServer::notify() [[maybe_unused]] ssize_t n = ::write(wake_write_fd_, &one, 1); } -bool SocketServer::send(int client_id, const void* data, size_t len) +bool SocketServer::send(int client_id, uint64_t request_id, const void* data, size_t len) { - if (client_id < 0 || static_cast(client_id) >= client_fds_.size() || - client_fds_[static_cast(client_id)] < 0) { + auto fd_it = client_fds_.find(client_id); + if (fd_it == client_fds_.end()) { errno = EINVAL; return false; } @@ -164,18 +153,29 @@ bool SocketServer::send(int client_id, const void* data, size_t len) return false; } - int fd = client_fds_[static_cast(client_id)]; + int fd = fd_it->second; - // Send length prefix (4 bytes) then message data, looping on partial - // writes — a short write after the prefix would permanently desync the - // stream for this connection. - auto msg_len = static_cast(len); - const uint8_t* parts[2] = { reinterpret_cast(&msg_len), static_cast(data) }; - size_t part_lens[2] = { sizeof(msg_len), len }; - for (int part = 0; part < 2; part++) { + // Send length prefix (4 bytes), echoed request id (8 bytes), then message + // data, looping on partial writes — a short write after the prefix would + // permanently desync the stream for this connection. + // + // MSG_NOSIGNAL: a peer that closed with responses still in flight must + // yield EPIPE here, not a process-killing SIGPIPE. macOS has no + // MSG_NOSIGNAL; accept() sets SO_NOSIGPIPE on the fd instead. +#ifdef MSG_NOSIGNAL + constexpr int send_flags = MSG_NOSIGNAL; +#else + constexpr int send_flags = 0; +#endif + auto msg_len = static_cast(FRAME_ID_SIZE + len); + const uint8_t* parts[3] = { reinterpret_cast(&msg_len), + reinterpret_cast(&request_id), + static_cast(data) }; + size_t part_lens[3] = { sizeof(msg_len), FRAME_ID_SIZE, len }; + for (int part = 0; part < 3; part++) { size_t total_sent = 0; while (total_sent < part_lens[part]) { - ssize_t n = ::send(fd, parts[part] + total_sent, part_lens[part] - total_sent, 0); + ssize_t n = ::send(fd, parts[part] + total_sent, part_lens[part] - total_sent, send_flags); if (n < 0) { if (errno == EINTR) { continue; // Interrupted, retry @@ -199,20 +199,18 @@ void SocketServer::release(int client_id, size_t message_size) (void)message_size; } -std::span SocketServer::receive(int client_id) +std::span SocketServer::receive(int client_id, uint64_t& request_id) { - if (client_id < 0 || static_cast(client_id) >= client_fds_.size() || - client_fds_[static_cast(client_id)] < 0) { + auto fd_it = client_fds_.find(client_id); + if (fd_it == client_fds_.end()) { return {}; } - int fd = client_fds_[static_cast(client_id)]; - const auto client_idx = static_cast(client_id); - - // Ensure buffers are sized for this client - if (client_idx >= recv_buffers_.size()) { - recv_buffers_.resize(client_idx + 1); - } + int fd = fd_it->second; + // Default-constructs this client's buffer on first use. disconnect_client() erases the + // map entry and invalidates this reference — every disconnect path below returns + // immediately without touching the buffer again. + std::vector& recv_buffer = recv_buffers_[client_id]; // Read length prefix (4 bytes) - must loop until all bytes received (MSG_WAITALL unreliable on macOS) uint32_t msg_len = 0; @@ -233,26 +231,44 @@ std::span SocketServer::receive(int client_id) total_read += static_cast(n); } - // A corrupt/malicious prefix must not drive the allocation below. - if (msg_len > MAX_FRAME_SIZE) { + // A corrupt/malicious prefix must not drive the allocation below. A frame + // shorter than the request-id field means the peer speaks the id-less + // protocol — disconnect rather than misparse. + if (msg_len > MAX_FRAME_SIZE || msg_len < FRAME_ID_SIZE) { + fprintf(stderr, "ipc: client %d sent an invalid frame (len=%u) — protocol mismatch?\n", client_id, msg_len); disconnect_client(client_id); return {}; } + // Read the request id (8 bytes, little-endian). + request_id = 0; + total_read = 0; + while (total_read < FRAME_ID_SIZE) { + ssize_t n = ::recv(fd, reinterpret_cast(&request_id) + total_read, FRAME_ID_SIZE - total_read, 0); + if (n <= 0) { + if (n < 0 && errno == EINTR) { + continue; // Interrupted, retry + } + disconnect_client(client_id); + return {}; + } + total_read += static_cast(n); + } + msg_len -= FRAME_ID_SIZE; + // Resize buffer if needed to fit length prefix + message size_t total_size = sizeof(uint32_t) + msg_len; - if (recv_buffers_[client_idx].size() < total_size) { - recv_buffers_[client_idx].resize(total_size); + if (recv_buffer.size() < total_size) { + recv_buffer.resize(total_size); } // Store length prefix in buffer - std::memcpy(recv_buffers_[client_idx].data(), &msg_len, sizeof(uint32_t)); + std::memcpy(recv_buffer.data(), &msg_len, sizeof(uint32_t)); // Read message data - must loop until all bytes received (MSG_WAITALL unreliable on macOS) total_read = 0; while (total_read < msg_len) { - ssize_t n = - ::recv(fd, recv_buffers_[client_idx].data() + sizeof(uint32_t) + total_read, msg_len - total_read, 0); + ssize_t n = ::recv(fd, recv_buffer.data() + sizeof(uint32_t) + total_read, msg_len - total_read, 0); if (n < 0) { if (errno == EINTR) { continue; // Interrupted, retry @@ -268,7 +284,7 @@ std::span SocketServer::receive(int client_id) total_read += static_cast(n); } - return std::span(recv_buffers_[client_idx].data() + sizeof(uint32_t), msg_len); + return std::span(recv_buffer.data() + sizeof(uint32_t), msg_len); } #ifdef __APPLE__ @@ -388,15 +404,16 @@ int SocketServer::accept() fcntl(client_fd, F_SETFL, flags & ~O_NONBLOCK); } - // Find free slot (or allocate new one) - int client_id = find_free_slot(); +#ifdef SO_NOSIGPIPE + // No MSG_NOSIGNAL on macOS: suppress SIGPIPE per-fd so a send() to a + // disconnected peer yields EPIPE instead of killing the process. + int nosigpipe = 1; + setsockopt(client_fd, SOL_SOCKET, SO_NOSIGPIPE, &nosigpipe, sizeof(nosigpipe)); +#endif - // Store client fd - const auto client_id_unsigned = static_cast(client_id); - if (client_id_unsigned >= client_fds_.size()) { - client_fds_.resize(client_id_unsigned + 1, -1); - } - client_fds_[static_cast(client_id)] = client_fd; + // Fresh, never-reused connection id. + int client_id = next_client_id_++; + client_fds_[client_id] = client_fd; fd_to_client_id_[client_fd] = client_id; num_clients_++; @@ -468,23 +485,23 @@ int SocketServer::wait_for_data(uint64_t timeout_ns) void SocketServer::disconnect_client(int client_id) { - if (client_id < 0 || static_cast(client_id) >= client_fds_.size()) { + auto fd_it = client_fds_.find(client_id); + if (fd_it == client_fds_.end()) { return; } - int fd = client_fds_[static_cast(client_id)]; - if (fd >= 0) { - // For kqueue, we don't need explicit deletion - closing the fd removes it automatically - // But we can explicitly remove it for clarity - struct kevent ev; - EV_SET(&ev, fd, EVFILT_READ, EV_DELETE, 0, 0, nullptr); - kevent(fd_, &ev, 1, nullptr, 0, nullptr); - - ::close(fd); - fd_to_client_id_.erase(fd); - client_fds_[static_cast(client_id)] = -1; - num_clients_--; - } + int fd = fd_it->second; + // For kqueue, we don't need explicit deletion - closing the fd removes it automatically + // But we can explicitly remove it for clarity + struct kevent ev; + EV_SET(&ev, fd, EVFILT_READ, EV_DELETE, 0, 0, nullptr); + kevent(fd_, &ev, 1, nullptr, 0, nullptr); + + ::close(fd); + fd_to_client_id_.erase(fd); + client_fds_.erase(fd_it); + recv_buffers_.erase(client_id); + num_clients_--; } #else @@ -606,15 +623,9 @@ int SocketServer::accept() fcntl(client_fd, F_SETFL, flags & ~O_NONBLOCK); } - // Find free slot (or allocate new one) - int client_id = find_free_slot(); - - // Store client fd - const auto client_id_unsigned = static_cast(client_id); - if (client_id_unsigned >= client_fds_.size()) { - client_fds_.resize(client_id_unsigned + 1, -1); - } - client_fds_[static_cast(client_id)] = client_fd; + // Fresh, never-reused connection id. + int client_id = next_client_id_++; + client_fds_[client_id] = client_fd; fd_to_client_id_[client_fd] = client_id; num_clients_++; @@ -680,18 +691,18 @@ int SocketServer::wait_for_data(uint64_t timeout_ns) void SocketServer::disconnect_client(int client_id) { - if (client_id < 0 || static_cast(client_id) >= client_fds_.size()) { + auto fd_it = client_fds_.find(client_id); + if (fd_it == client_fds_.end()) { return; } - int fd = client_fds_[static_cast(client_id)]; - if (fd >= 0) { - epoll_ctl(fd_, EPOLL_CTL_DEL, fd, nullptr); - ::close(fd); - fd_to_client_id_.erase(fd); - client_fds_[static_cast(client_id)] = -1; - num_clients_--; - } + int fd = fd_it->second; + epoll_ctl(fd_, EPOLL_CTL_DEL, fd, nullptr); + ::close(fd); + fd_to_client_id_.erase(fd); + client_fds_.erase(fd_it); + recv_buffers_.erase(client_id); + num_clients_--; } #endif diff --git a/ipc-runtime/cpp/ipc_runtime/socket_server.hpp b/ipc-runtime/cpp/ipc_runtime/socket_server.hpp index b1e1bb1effbe..61313d5dd8a7 100644 --- a/ipc-runtime/cpp/ipc_runtime/socket_server.hpp +++ b/ipc-runtime/cpp/ipc_runtime/socket_server.hpp @@ -6,6 +6,7 @@ #include #include #include +#include #include namespace ipc { @@ -32,9 +33,9 @@ class SocketServer : public IpcServer { bool listen() override; int accept() override; int wait_for_data(uint64_t timeout_ns) override; - std::span receive(int client_id) override; + std::span receive(int client_id, uint64_t& request_id) override; void release(int client_id, size_t message_size) override; - bool send(int client_id, const void* data, size_t len) override; + bool send(int client_id, uint64_t request_id, const void* data, size_t len) override; void close() override; // Wake a thread blocked in wait_for_data() by writing the self-pipe whose @@ -50,7 +51,6 @@ class SocketServer : public IpcServer { private: void close_internal(); void disconnect_client(int client_id); - int find_free_slot(); // Create the self-pipe and register its read end with the epoll/kqueue // instance. Returns false on failure. Called from listen(). bool setup_wake_pipe(); @@ -60,12 +60,17 @@ class SocketServer : public IpcServer { std::string socket_path_; int initial_max_clients_; int listen_fd_ = -1; - int fd_ = -1; // kqueue or epoll fd - int wake_read_fd_ = -1; // self-pipe read end (in the event set) - int wake_write_fd_ = -1; // self-pipe write end (poked by notify()) - std::vector client_fds_; // client_id -> fd - std::unordered_map fd_to_client_id_; // fd -> client_id (for fast lookup) - std::vector> recv_buffers_; // client_id -> recv buffer + int fd_ = -1; // kqueue or epoll fd + int wake_read_fd_ = -1; // self-pipe read end (in the event set) + int wake_write_fd_ = -1; // self-pipe write end (poked by notify()) + // Client ids are monotonic and never reused: a connection's identity must not be + // inheritable by a later connection, or state addressed to a dead client (a late + // handler response, reorder bookkeeping) could reach its successor. Contrast with + // the SHM transport, whose ids are physical ring indices and must recycle. + int next_client_id_ = 0; + std::unordered_map client_fds_; // client_id -> fd + std::unordered_map fd_to_client_id_; // fd -> client_id (for fast lookup) + std::unordered_map> recv_buffers_; // client_id -> recv buffer int num_clients_ = 0; }; diff --git a/ipc-runtime/cpp/napi/msgpack_client_async.cpp b/ipc-runtime/cpp/napi/msgpack_client_async.cpp index 64f08978b31e..db0b1d27077f 100644 --- a/ipc-runtime/cpp/napi/msgpack_client_async.cpp +++ b/ipc-runtime/cpp/napi/msgpack_client_async.cpp @@ -62,15 +62,17 @@ void MsgpackClientAsync::poll_responses() constexpr uint64_t TIMEOUT_NS = 1'000'000'000; // 1s while (!shutdown_.load(std::memory_order_acquire)) { - std::span response = client_->receive(TIMEOUT_NS); + uint64_t request_id = 0; + std::span response = client_->receive(TIMEOUT_NS, request_id); // data() == nullptr means timeout; a non-null empty span is a valid // zero-length response and must be delivered. if (response.data() == nullptr) { continue; // timeout — keep polling (and re-check shutdown) } - // Copy out — span is invalidated by release(). - auto* response_data = new std::vector(response.begin(), response.end()); + // Copy out — span is invalidated by release(). The echoed request id + // travels with the payload so the TS side can pair it to its promise. + auto* response_data = new Completion{ request_id, { response.begin(), response.end() } }; client_->release(response.size()); std::lock_guard lock(tsfn_mutex_); @@ -82,11 +84,12 @@ void MsgpackClientAsync::poll_responses() delete response_data; continue; } - auto status = tsfn_.NonBlockingCall( - response_data, [](Napi::Env env, Napi::Function js_callback, std::vector* data) { - auto js_buffer = Napi::Buffer::Copy(env, data->data(), data->size()); - js_callback.Call({ js_buffer }); - delete data; + auto status = + tsfn_.NonBlockingCall(response_data, [](Napi::Env env, Napi::Function js_callback, Completion* completion) { + auto js_buffer = + Napi::Buffer::Copy(env, completion->payload.data(), completion->payload.size()); + js_callback.Call({ Napi::BigInt::New(env, completion->request_id), js_buffer }); + delete completion; }); if (status != napi_ok) { // Failed to queue — likely process exiting. Drop the response. @@ -99,20 +102,25 @@ Napi::Value MsgpackClientAsync::call(const Napi::CallbackInfo& info) { Napi::Env env = info.Env(); - if (info.Length() < 1 || !info[0].IsBuffer()) { - throw Napi::TypeError::New(env, "First argument must be a Buffer"); + if (info.Length() < 2 || !info[0].IsBigInt() || !info[1].IsBuffer()) { + throw Napi::TypeError::New(env, "Expected (request_id: bigint, payload: Buffer)"); } if (shutdown_.load(std::memory_order_acquire)) { throw Napi::Error::New(env, "Client is closed"); } - auto input_buffer = info[0].As>(); + bool lossless = false; + uint64_t request_id = info[0].As().Uint64Value(&lossless); + if (!lossless) { + throw Napi::TypeError::New(env, "request_id must fit in an unsigned 64-bit integer"); + } + auto input_buffer = info[1].As>(); const uint8_t* input_data = input_buffer.Data(); size_t input_len = input_buffer.Length(); // Single non-blocking attempt: claim() treats timeout 1 as an immediate - // check (0 would be normalized to infinite). TS owns the promise queue. - if (!client_->send(input_data, input_len, 1)) { + // check (0 would be normalized to infinite). TS owns the promise map. + if (!client_->send(request_id, input_data, input_len, 1)) { throw Napi::Error::New(env, "Failed to send request, ring buffer full. Make it bigger?"); } diff --git a/ipc-runtime/cpp/napi/msgpack_client_async.hpp b/ipc-runtime/cpp/napi/msgpack_client_async.hpp index efd11ef43a0f..33f80189202f 100644 --- a/ipc-runtime/cpp/napi/msgpack_client_async.hpp +++ b/ipc-runtime/cpp/napi/msgpack_client_async.hpp @@ -6,6 +6,7 @@ #include #include #include +#include namespace ipc::napi { @@ -25,6 +26,11 @@ namespace ipc::napi { * TS owns the queue (single-threaded JS makes that natural), so we don't need * a C++-side mutex/queue. */ +struct Completion { + uint64_t request_id; + std::vector payload; +}; + class MsgpackClientAsync : public Napi::ObjectWrap { public: MsgpackClientAsync(const Napi::CallbackInfo& info); diff --git a/ipc-runtime/ts/src/shm_client.test.ts b/ipc-runtime/ts/src/shm_client.test.ts new file mode 100644 index 000000000000..02d8b999394a --- /dev/null +++ b/ipc-runtime/ts/src/shm_client.test.ts @@ -0,0 +1,72 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { NapiShmAsyncClient, NapiMsgpackClientAsync } from "./shm_client.js"; + +/** + * Drives NapiShmAsyncClient through a mock addon, so the id-pairing logic is + * testable without the native module or a live server. + */ +class MockAddon implements NapiMsgpackClientAsync { + public deliver!: (requestId: bigint, response: Buffer) => void; + public sent: Array<{ requestId: bigint; input: Buffer }> = []; + public acquires = 0; + public releases = 0; + public closed = false; + + setResponseCallback(cb: (requestId: bigint, response: Buffer) => void): void { + this.deliver = cb; + } + call(requestId: bigint, input: Buffer): void { + this.sent.push({ requestId, input }); + } + acquire(): void { + this.acquires++; + } + release(): void { + this.releases++; + } + close(): void { + this.closed = true; + } +} + +test("shm async client discards a stale frame and still resolves the live call", async () => { + const addon = new MockAddon(); + const client = new NapiShmAsyncClient(addon); + + const pending = client.call(new Uint8Array([1, 2, 3])); + assert.equal(addon.sent.length, 1); + const liveId = addon.sent[0].requestId; + + // A leftover frame from a ring's previous occupant: unknown id. Must be + // discarded — not resolve the live call, not reject anything. + addon.deliver(liveId ^ 0xdeadbeefn, Buffer.from([0xba, 0xad])); + + // The real response still pairs and resolves. + addon.deliver(liveId, Buffer.from([9, 9])); + assert.deepEqual(await pending, new Uint8Array([9, 9])); + + // Refcount stayed balanced: one acquire for the call, one release when the + // live response drained the map (the stale frame must not release). + assert.equal(addon.acquires, 1); + assert.equal(addon.releases, 1); + + await client.destroy(); +}); + +test("shm async client pairs out-of-order responses to the right callers", async () => { + const addon = new MockAddon(); + const client = new NapiShmAsyncClient(addon); + + const a = client.call(new Uint8Array([0xaa])); + const b = client.call(new Uint8Array([0xbb])); + const [idA, idB] = addon.sent.map((s) => s.requestId); + + // Complete in reverse order; each caller must get its own payload. + addon.deliver(idB, Buffer.from([2])); + addon.deliver(idA, Buffer.from([1])); + assert.deepEqual(await b, new Uint8Array([2])); + assert.deepEqual(await a, new Uint8Array([1])); + + await client.destroy(); +}); diff --git a/ipc-runtime/ts/src/shm_client.ts b/ipc-runtime/ts/src/shm_client.ts index 19bcd51c5f76..53b43456afc5 100644 --- a/ipc-runtime/ts/src/shm_client.ts +++ b/ipc-runtime/ts/src/shm_client.ts @@ -13,9 +13,10 @@ import { IpcClientAsync, IpcClientSync } from "./types.js"; * prebuilt addon shipped in this package's `build/-/` directory. * * Note on the async contract: `MsgpackClientAsync.call` is *fire and - * forget*. Responses arrive via `setResponseCallback` in FIFO order on a - * background-thread → main-thread bridge (Napi::ThreadSafeFunction). - * The TS wrapper below owns the request queue and matches responses. + * forget*. Responses arrive via `setResponseCallback` in COMPLETION order on + * a background-thread → main-thread bridge (Napi::ThreadSafeFunction), each + * carrying its echoed request id. The TS wrapper below owns the pending map + * and pairs responses to callers by id. */ export interface NapiMsgpackClientSync { call(input: Buffer): Buffer; @@ -23,8 +24,8 @@ export interface NapiMsgpackClientSync { } export interface NapiMsgpackClientAsync { - setResponseCallback(cb: (response: Buffer) => void): void; - call(input: Buffer): void; + setResponseCallback(cb: (requestId: bigint, response: Buffer) => void): void; + call(requestId: bigint, input: Buffer): void; acquire(): void; release(): void; /** Stop the native poll thread, release any held TSFN ref, close the client. */ @@ -55,36 +56,47 @@ interface PendingCallback { /** * Wraps the fire-and-forget async NAPI msgpack client behind the - * `IpcClientAsync` interface. Owns a FIFO queue of pending calls; the C++ - * background polling thread invokes `setResponseCallback` once per - * response, and this wrapper matches it to the next queued caller. + * `IpcClientAsync` interface. Owns a map of pending calls keyed by request + * id; the C++ background polling thread invokes `setResponseCallback` once + * per response (in completion order), and this wrapper pairs it to its + * caller by the echoed id. Ids start at a random point per client so a + * stale frame left in a recycled SHM ring slot by a previous occupant + * cannot pair with a live call. * * `acquire` / `release` are reference-count hooks the NAPI exposes so the * libuv loop is kept alive only while requests are outstanding — without * them a `node script.js` would never exit naturally. */ export class NapiShmAsyncClient implements IpcClientAsync { - private readonly pending: PendingCallback[] = []; + private readonly pending = new Map(); + private nextRequestId = + (BigInt(Math.floor(Math.random() * 0xffffffff)) << 16n) + 1n; private destroyed = false; constructor(private inner: NapiMsgpackClientAsync) { - this.inner.setResponseCallback((response: Buffer) => { + this.inner.setResponseCallback((requestId: bigint, response: Buffer) => { if (this.destroyed) { // Late response delivered after destroy(); the native close already // balanced the TSFN reference. return; } - const cb = this.pending.shift(); + const cb = this.pending.get(requestId); if (cb) { + this.pending.delete(requestId); cb.resolve(new Uint8Array(response)); - if (this.pending.length === 0) { + if (this.pending.size === 0) { this.inner.release(); } } else { - // Protocol desync — every response should match a pending call. - // Don't release: no acquire was taken for an orphan response. + // SHM rings persist across occupants (slot reclaim / reattach), so a + // frame addressed to a previous occupant's id is an anticipated + // leftover — discard it and keep serving live calls. Log it so that if + // a genuinely lost pairing ever hangs a caller, the evidence is in the + // log rather than silently dropped. Don't release: no acquire was + // taken for an orphan response. console.warn( - "NapiShmAsyncClient: dropping response with no pending caller", + `NapiShmAsyncClient: discarding response for unknown request id ${requestId} ` + + "(stale frame from a previous ring occupant?)", ); } }); @@ -100,16 +112,17 @@ export class NapiShmAsyncClient implements IpcClientAsync { ? input : Buffer.from(input.buffer, input.byteOffset, input.byteLength); return new Promise((resolve, reject) => { - if (this.pending.length === 0) { + const requestId = this.nextRequestId++; + if (this.pending.size === 0) { this.inner.acquire(); } - this.pending.push({ resolve, reject }); + this.pending.set(requestId, { resolve, reject }); try { - this.inner.call(buf); + this.inner.call(requestId, buf); } catch (err: any) { - // Send failed — unwind the queue entry we just added. - this.pending.pop(); - if (this.pending.length === 0) { + // Send failed — unwind the map entry we just added. + this.pending.delete(requestId); + if (this.pending.size === 0) { this.inner.release(); } reject( @@ -127,12 +140,13 @@ export class NapiShmAsyncClient implements IpcClientAsync { } this.destroyed = true; // Reject anything still in flight. - while (this.pending.length > 0) { - const cb = this.pending.shift(); - cb?.reject(new Error("ipc-runtime SHM client destroyed before response")); + const err = new Error("ipc-runtime SHM client destroyed before response"); + for (const cb of this.pending.values()) { + cb.reject(err); } + this.pending.clear(); // Stops the native poll thread and releases the TSFN reference taken - // when the queue went 0 → 1 — without this, Node never exits when + // when the map went 0 → 1 — without this, Node never exits when // destroyed with calls in flight. this.inner.close(); } diff --git a/ipc-runtime/ts/src/spawned_backend.test.ts b/ipc-runtime/ts/src/spawned_backend.test.ts index a630b52f33ac..472492d9a6c1 100644 --- a/ipc-runtime/ts/src/spawned_backend.test.ts +++ b/ipc-runtime/ts/src/spawned_backend.test.ts @@ -24,11 +24,13 @@ const server = net.createServer(conn => { while (buf.length >= 4) { const len = buf.readUInt32LE(0); if (buf.length < 4 + len) return; + const requestId = buf.readBigUInt64LE(4); // frame = [len][8B id][payload] buf = buf.subarray(4 + len); const payload = Buffer.from(String(process.pid)); - const out = Buffer.alloc(4 + payload.length); - out.writeUInt32LE(payload.length, 0); - payload.copy(out, 4); + const out = Buffer.alloc(12 + payload.length); + out.writeUInt32LE(payload.length + 8, 0); + out.writeBigUInt64LE(requestId, 4); + payload.copy(out, 12); conn.write(out); } }); diff --git a/ipc-runtime/ts/src/uds.test.ts b/ipc-runtime/ts/src/uds.test.ts index 07d36a1e1036..eed1a6fd2c52 100644 --- a/ipc-runtime/ts/src/uds.test.ts +++ b/ipc-runtime/ts/src/uds.test.ts @@ -110,6 +110,62 @@ test("client rejects oversized frame from server", async () => { } }); +test("client fails all pending calls on a response with an unknown request id", async () => { + const socketPath = tmpSocketPath("unknown_id"); + // Raw server that answers with a well-formed frame whose request id matches + // nothing the client sent — the correlation-desync case. + const rawServer = net.createServer((conn) => { + conn.once("data", () => { + const frame = Buffer.allocUnsafe(12 + 1); + frame.writeUInt32LE(1 + 8, 0); + frame.writeBigUInt64LE(0xdeadbeefn, 4); + frame.writeUInt8(42, 12); + conn.write(frame); + }); + }); + await new Promise((resolve) => + rawServer.listen(socketPath, () => resolve()), + ); + + const client = await UdsIpcClient.connect(socketPath); + try { + await assert.rejects( + client.call(new Uint8Array([1])), + /unknown request id/, + ); + } finally { + await client.destroy(); + rawServer.close(); + fs.rmSync(socketPath, { force: true }); + } +}); + +test("client fails loudly on an id-less (old-protocol) frame", async () => { + const socketPath = tmpSocketPath("idless"); + // Raw server speaking the pre-envelope-id protocol: [4B len][payload] with + // len < 8. + const rawServer = net.createServer((conn) => { + conn.once("data", () => { + const frame = Buffer.allocUnsafe(4 + 1); + frame.writeUInt32LE(1, 0); + frame.writeUInt8(42, 4); + conn.write(frame); + }); + }); + await new Promise((resolve) => + rawServer.listen(socketPath, () => resolve()), + ); + + const client = await UdsIpcClient.connect(socketPath); + try { + await assert.rejects(client.call(new Uint8Array([1])), /protocol mismatch/); + } finally { + await client.destroy(); + rawServer.close(); + fs.rmSync(socketPath, { force: true }); + } +}); + test("server drops connection on oversized frame", async () => { const socketPath = tmpSocketPath("oversize_srv"); const server = await UdsIpcServer.listen(socketPath, (_id, req) => req); diff --git a/ipc-runtime/ts/src/uds_client.ts b/ipc-runtime/ts/src/uds_client.ts index 9127704f5d0a..4c6173572223 100644 --- a/ipc-runtime/ts/src/uds_client.ts +++ b/ipc-runtime/ts/src/uds_client.ts @@ -25,15 +25,19 @@ export interface UdsIpcClientConnectOptions { /** * Async IPC client over a Unix Domain Socket. Wire format matches the C++ * ipc::IpcServer/IpcClient socket transport: 4-byte little-endian length - * prefix followed by `length` bytes of msgpack payload, per direction. + * prefix, 8-byte little-endian request id, then the msgpack payload (the + * length counts the id plus the payload), per direction. * - * Supports pipelining: multiple concurrent `call()` invocations are queued - * FIFO and matched with responses in order. Pipelining keeps the server-side - * socket window full and matches the native client behaviour. + * Supports pipelining: each call carries a unique request id which the + * server echoes on the response, so responses are paired to callers by id + * and the server may complete requests in any order. Ids start at a random + * point per connection. */ export class UdsIpcClient implements IpcClientAsync { private buffer: Buffer = Buffer.alloc(0); - private pending: PendingCall[] = []; + private pending = new Map(); + private nextRequestId = + (BigInt(Math.floor(Math.random() * 0xffffffff)) << 16n) + 1n; private destroyed = false; /** Set once the socket has errored/closed; new calls fail fast. */ private closed = false; @@ -67,7 +71,7 @@ export class UdsIpcClient implements IpcClientAsync { /** Number of in-flight calls awaiting a response. */ get inflight(): number { - return this.pending.length; + return this.pending.size; } /** Underlying socket — exposed for ref/unref control (event-loop tuning). */ @@ -85,10 +89,12 @@ export class UdsIpcClient implements IpcClientAsync { ); } return new Promise((resolve, reject) => { - this.pending.push({ resolve, reject }); - const lenBuf = Buffer.allocUnsafe(4); - lenBuf.writeUInt32LE(input.length, 0); - this.conn.write(lenBuf); + const requestId = this.nextRequestId++; + this.pending.set(requestId, { resolve, reject }); + const header = Buffer.allocUnsafe(12); + header.writeUInt32LE(input.length + 8, 0); // length counts id + payload + header.writeBigUInt64LE(requestId, 4); + this.conn.write(header); this.conn.write(input); }); } @@ -118,25 +124,45 @@ export class UdsIpcClient implements IpcClientAsync { ); return; } + if (len < 8) { + // Shorter than the request-id field: the server speaks the id-less + // protocol. Fail loudly instead of misparsing. + this.conn.destroy(); + this.failAll( + new IpcTransportError( + `UdsIpcClient: ${len}-byte frame is shorter than the request-id field — ` + + "IPC protocol mismatch (envelope ids); update the peer binary/package", + ), + ); + return; + } if (this.buffer.length < 4 + len) return; - const payload = this.buffer.subarray(4, 4 + len); + const requestId = this.buffer.readBigUInt64LE(4); + const payload = this.buffer.subarray(12, 4 + len); this.buffer = this.buffer.subarray(4 + len); - const next = this.pending.shift(); + const next = this.pending.get(requestId); if (next) { + this.pending.delete(requestId); next.resolve(new Uint8Array(payload)); } else { - // Protocol desync — every response should match a pending call. - console.warn( - `UdsIpcClient: dropping ${len}-byte response with no pending caller`, + // A response that pairs with no pending call means the stream's + // correlation is broken — fail everything loudly rather than + // continuing on a connection we can no longer trust. + this.conn.destroy(); + this.failAll( + new IpcTransportError( + `UdsIpcClient: response for unknown request id ${requestId} — protocol desync`, + ), ); + return; } } } private failAll(err: Error): void { this.closed = true; - const pending = this.pending; - this.pending = []; + const pending = [...this.pending.values()]; + this.pending.clear(); for (const p of pending) p.reject(err); } } @@ -144,11 +170,12 @@ export class UdsIpcClient implements IpcClientAsync { /** * Connect to `socketPath`, retrying "server not ready" errors until * `timeoutMs` elapses: ENOENT (socket file not created yet), ECONNREFUSED - * (the window between the server's bind() and listen()), and EAGAIN (Linux + * (the window between the server's bind() and listen()), EAGAIN (Linux * reports this for a UDS connect when the accept backlog is momentarily - * full). Other errors fail immediately. Each attempt is also capped at the - * remaining budget, so a bound-but-never-accepting server cannot hang the - * connect past the deadline. + * full), and ECONNRESET (a connect racing the server's accept loop under + * connection churn). Other errors fail immediately. Each attempt is also + * capped at the remaining budget, so a bound-but-never-accepting server + * cannot hang the connect past the deadline. */ async function connectWithRetry( socketPath: string, @@ -166,6 +193,7 @@ async function connectWithRetry( const code = (err as NodeJS.ErrnoException).code; if ( code !== "ECONNREFUSED" && + code !== "ECONNRESET" && code !== "ENOENT" && code !== "ETIMEDOUT" && code !== "EAGAIN" diff --git a/ipc-runtime/ts/src/uds_server.ts b/ipc-runtime/ts/src/uds_server.ts index 309cee40a7d0..6d07a874643c 100644 --- a/ipc-runtime/ts/src/uds_server.ts +++ b/ipc-runtime/ts/src/uds_server.ts @@ -14,9 +14,11 @@ export type IpcServerHandler = ( ) => Promise | Uint8Array; /** - * UDS server with the same 4-byte-LE-length-prefix wire as UdsIpcClient and - * the C++ ipc::IpcServer socket transport. Accepts multiple concurrent - * connections; handler invocations are serialised per-connection. + * UDS server with the same wire format as UdsIpcClient and the C++ + * ipc::IpcServer socket transport: 4-byte LE length prefix, 8-byte LE request + * id (echoed on the response), then the payload; the length counts the id + * plus the payload. Accepts multiple concurrent connections; handler + * invocations are serialised per-connection. * * Signal handling is the caller's responsibility (unlike the C++ server's * install_default_signal_handlers); the socket file is unlinked on close() @@ -119,10 +121,22 @@ export class UdsIpcServer { ); return; } + if (len < 8) { + // Shorter than the request-id field: the peer speaks the id-less + // protocol. Drop the connection with a clear reason. + conn.destroy( + new Error( + `UdsIpcServer: ${len}-byte frame is shorter than the request-id field — ` + + "IPC protocol mismatch (envelope ids); update the peer binary/package", + ), + ); + return; + } if (buffer.length < 4 + len) break; + const requestId = buffer.readBigUInt64LE(4); // Copy into a standalone Buffer (not a subarray view, and not a plain Uint8Array): handlers // decode with msgpackr, which relies on Buffer semantics for correct string/binary decoding. - const payload = Buffer.from(buffer.subarray(4, 4 + len)); + const payload = Buffer.from(buffer.subarray(12, 4 + len)); buffer = buffer.subarray(4 + len); const prev = chain; @@ -130,9 +144,10 @@ export class UdsIpcServer { await prev; try { const resp = await handler(clientId, payload); - const lenBuf = Buffer.allocUnsafe(4); - lenBuf.writeUInt32LE(resp.length, 0); - conn.write(lenBuf); + const header = Buffer.allocUnsafe(12); + header.writeUInt32LE(resp.length + 8, 0); // length counts id + payload + header.writeBigUInt64LE(requestId, 4); + conn.write(header); conn.write(resp); } catch (err) { conn.destroy(err as Error); diff --git a/yarn-project/world-state/src/native/ipc_churn_correlation.test.ts b/yarn-project/world-state/src/native/ipc_churn_correlation.test.ts new file mode 100644 index 000000000000..c3ef34b8aa93 --- /dev/null +++ b/yarn-project/world-state/src/native/ipc_churn_correlation.test.ts @@ -0,0 +1,402 @@ +import { Fr } from '@aztec/foundation/curves/bn254'; +import { type IpcClientAsync, UdsIpcClient, createNapiShmAsyncClient } from '@aztec/ipc-runtime'; +import { MerkleTreeId } from '@aztec/stdlib/trees'; +import { AsyncApi } from '@aztec/wsdb'; + +import { jest } from '@jest/globals'; + +import { NativeWorldStateService } from './native_world_state.js'; + +// Load tests for the wsdb IPC response-correlation invariant: every response a client resolves must +// be the response to ITS OWN request. The TS client correlates positionally (no request IDs), so any +// server-side misordering — reactor reorder stash, per-fork scheduler, or slot reuse across +// disconnects — surfaces here as a caller receiving another caller's payload. +// +// Detection design: connections must never share observable state, or a cross-connection swap of +// same-type responses is invisible (identical requests have identical answers). Each connection +// plants a private fork with leaves derived from its own seed, so ANY frame delivered to the wrong +// connection produces a visibly wrong value — wrong index, wrong root, wrong size — not just a +// wrong-type decode. +// +// Legs: +// C: single connection, sequential (sanity — must always pass). +// A: single connection, heavy pipelined contention (per-fork write serialisation → out-of-order +// completions feed the reorder stash) — models the long-lived world-state client. +// D: multi-connection read/write soak, NO connection churn. Readers verify recorded roots and +// exact leaf indices on their private forks; writers pipeline append→read-after-write with +// exact-index asserts. Independent of the disconnect-cleanup fixes — any failure here is a +// distinct IPC/reorder/scheduler bug. +// B: leg D's workload plus mid-flight churn — clients destroyed with calls in flight and +// replaced, recycling server slots. Guards the reactor's disconnect cleanup and the server's +// survival of mid-flight death. +// +// WSDB_SOAK_MS extends the D/B legs for grinding sessions (default a few seconds for CI). + +const SOAK_MS = Number(process.env.WSDB_SOAK_MS ?? 4_000); +jest.setTimeout(Math.max(300_000, SOAK_MS * 4)); + +const TRANSPORT = (process.env.WSDB_TRANSPORT === 'shm' ? 'shm' : 'uds') as 'shm' | 'uds'; +// Typed as plain number: the generated AsyncApi carries tree ids as numbers, so enum-typed +// comparisons against response fields trip no-unsafe-enum-comparison. +const TREE: number = MerkleTreeId.NOTE_HASH_TREE; +const N = 64; +/** Private leaves planted per connection identity. */ +const M = 32; + +/** Per-connection private state: a fork only this logical client touches. */ +interface ConnIdentity { + forkId: number; + rev: { forkId: number; blockNumber: number; includeUncommitted: boolean }; + /** Leaves unique to this identity (and disjoint from every other identity and the seed fork). */ + leaves: Uint8Array[]; + /** Tree size before this identity's leaves were appended — leaf i lives at baseSize + i. */ + baseSize: number; + /** Tree size including all writes so far (advanced by writer rounds). */ + size: number; + /** Root recorded after planting; stable until the identity writes again. */ + root: Uint8Array; +} + +/** Unique leaf value: disjoint across identities, write rounds, and the shared seed fork. */ +const identityLeaf = (seed: number, i: number): Uint8Array => + new Fr((BigInt(0xc0ffee) << 64n) + (BigInt(seed) << 32n) + BigInt(i) + 1n).toBuffer(); + +/** Progress line every ~15s so long soaks show liveness and rate instead of going dark. */ +const makeHeartbeat = (leg: string) => { + const start = Date.now(); + let lastLog = start; + return (rounds: number) => { + const now = Date.now(); + if (now - lastLog >= 15_000) { + lastLog = now; + const elapsedS = (now - start) / 1000; + // eslint-disable-next-line no-console + console.log( + `[soak ${leg}] ${new Date(now).toISOString()} rounds=${rounds} ` + + `elapsed=${elapsedS.toFixed(0)}s rate=${(rounds / elapsedS).toFixed(1)}/s`, + ); + } + }; +}; + +describe(`wsdb IPC churn/contention correlation (transport=${TRANSPORT})`, () => { + let ws: NativeWorldStateService; + let ipcPath: string | undefined; + let leaves: Fr[]; + // The generated AsyncApi is bytes-in/bytes-out (wire `Fr = Uint8Array`); the facade normally + // converts, but these tests drive AsyncApi directly, so pass raw 32-byte buffers. + let leafBytes: Uint8Array[]; + /** A shared fork seeded with `leaves`; read with includeUncommitted. */ + let seedForkId: number; + let seedRevision: { forkId: number; blockNumber: number; includeUncommitted: boolean }; + + const makeClient = async (clientId: number): Promise<{ api: AsyncApi; backend: IpcClientAsync }> => { + const backend: IpcClientAsync = + TRANSPORT === 'shm' + ? createNapiShmAsyncClient(ipcPath!.replace(/\.shm$/, ''), { clientId }) + : await UdsIpcClient.connect(ipcPath!, { connectTimeoutMs: 5_000 }); + return { api: new AsyncApi(backend), backend }; + }; + + /** Create a private fork and plant this identity's unique leaves, recording size and root. */ + const plantIdentity = async (api: AsyncApi, seed: number): Promise => { + const forkId = (await api.createFork({ latest: true, blockNumber: 0 })).forkId; + const rev = { forkId, blockNumber: 0xffffffff, includeUncommitted: true }; + const before = await api.getTreeInfo({ treeId: TREE, revision: rev }); + const baseSize = Number(before.size); + const idLeaves = Array.from({ length: M }, (_, i) => identityLeaf(seed, i)); + await api.appendLeaves({ treeId: TREE, leaves: idLeaves, forkId }); + const after = await api.getTreeInfo({ treeId: TREE, revision: rev }); + if (Number(after.size) !== baseSize + M) { + throw new Error(`plant(seed=${seed}): size ${after.size} != ${baseSize + M}`); + } + return { forkId, rev, leaves: idLeaves, baseSize, size: baseSize + M, root: after.root }; + }; + + /** + * Pipelined reads whose responses each prove they belong to this identity's own requests: exact + * leaf indices on the private fork, and (when the identity has not written since planting) the + * recorded root. A frame swapped from any other connection fails these by value. + */ + const identityChecks = (api: AsyncApi, id: ConnIdentity, tag: string, opts?: { checkRoot?: boolean }) => { + const checks: Promise[] = []; + for (let i = 0; i < M; i++) { + const expectIdx = id.baseSize + i; + checks.push( + api.findLeafIndices({ treeId: TREE, revision: id.rev, leaves: [id.leaves[i]], startIndex: 0 }).then(r => { + const got = r.indices?.[0]; + if (got === null || got === undefined || Number(got) !== expectIdx) { + throw new Error(`${tag}: findLeafIndices(own leaf ${i}) returned ${String(got)}, expected ${expectIdx}`); + } + }), + ); + checks.push( + api.getSiblingPath({ treeId: TREE, revision: id.rev, leafIndex: expectIdx }).then(r => { + if (!Array.isArray(r.path) || r.path.length === 0) { + throw new Error(`${tag}: getSiblingPath(${expectIdx}) returned empty/invalid path`); + } + }), + ); + checks.push( + api.getTreeInfo({ treeId: TREE, revision: id.rev }).then(r => { + if (r.treeId !== TREE || Number(r.size) < id.baseSize + M) { + throw new Error(`${tag}: getTreeInfo returned treeId=${r.treeId} size=${r.size} (min ${id.baseSize + M})`); + } + if (opts?.checkRoot && Buffer.compare(r.root, id.root) !== 0) { + throw new Error(`${tag}: getTreeInfo root mismatch — response belongs to another fork/connection`); + } + }), + ); + checks.push( + api.getStateReference({ revision: id.rev }).then(r => { + if (!Array.isArray(r.state) || r.state.length === 0 || r.state.some(t => t.size === undefined)) { + throw new Error(`${tag}: getStateReference malformed: ${JSON.stringify(r).slice(0, 160)}`); + } + }), + ); + } + return checks; + }; + + /** + * One writer round: pipeline an append of a fresh unique leaf and, behind it on the same + * connection, a read of that leaf — the per-fork scheduler must order the read after the write, + * and the leaf must land at exactly the pre-append size. Also refreshes the identity's root. + */ + const writerRound = async (api: AsyncApi, id: ConnIdentity, seed: number, round: number, tag: string) => { + const leaf = identityLeaf(seed, M + round); + const expectIdx = id.size; + await Promise.all([ + api.appendLeaves({ treeId: TREE, leaves: [leaf], forkId: id.forkId }), + api.findLeafIndices({ treeId: TREE, revision: id.rev, leaves: [leaf], startIndex: 0 }).then(r => { + const got = r.indices?.[0]; + if (got === null || got === undefined || Number(got) !== expectIdx) { + throw new Error( + `${tag}: read-after-write of round-${round} leaf returned ${String(got)}, expected ${expectIdx}`, + ); + } + }), + ...identityChecks(api, id, tag), + ]); + id.size++; + id.root = (await api.getTreeInfo({ treeId: TREE, revision: id.rev })).root; + }; + + beforeAll(async () => { + ws = await NativeWorldStateService.tmp(); + try { + ipcPath = ws.getIpcPath(); + } catch { + ipcPath = undefined; // in-process build — no IPC to exercise + return; + } + leaves = Array.from({ length: N }, (_, i) => new Fr(BigInt(i) * 0x1_0000_0001n + 1n)); + leafBytes = leaves.map(l => l.toBuffer()); + + // Seed a shared fork (forks are shared world-state objects, not per-connection). Legs A/C use + // it; the multi-connection legs deliberately do NOT — see the detection-design note above. + const seeder = await makeClient(1); + try { + seedForkId = (await seeder.api.createFork({ latest: true, blockNumber: 0 })).forkId; + seedRevision = { forkId: seedForkId, blockNumber: 0xffffffff, includeUncommitted: true }; + await seeder.api.appendLeaves({ treeId: TREE, leaves: leafBytes, forkId: seedForkId }); + // Sanity: the seeded state must be readable exactly as planted, or every later check is void. + const probe = await seeder.api.findLeafIndices({ + treeId: TREE, + revision: seedRevision, + leaves: [leafBytes[5]], + startIndex: 0, + }); + if (Number(probe.indices?.[0]) !== 5) { + throw new Error(`seed verification failed: ${JSON.stringify(probe)}`); + } + } finally { + await seeder.backend.destroy(); + } + }); + + afterAll(async () => { + await ws?.close(); + }); + + // Shared-fork variant used by the single-connection legs (no cross-connection ambiguity there). + const selfCheckingReads = (api: AsyncApi, tag: string): Promise[] => { + const checks: Promise[] = []; + for (let i = 0; i < N; i++) { + checks.push( + api.getTreeInfo({ treeId: TREE, revision: seedRevision }).then(r => { + if (r.treeId !== TREE) { + throw new Error(`${tag}: getTreeInfo returned treeId=${r.treeId}, expected ${TREE}`); + } + }), + ); + checks.push( + api.getStateReference({ revision: seedRevision }).then(r => { + if (!Array.isArray(r.state) || r.state.length === 0 || r.state.some(t => t.size === undefined)) { + throw new Error(`${tag}: getStateReference malformed: ${JSON.stringify(r).slice(0, 160)}`); + } + }), + ); + const idx = i; + checks.push( + api.getSiblingPath({ treeId: TREE, revision: seedRevision, leafIndex: idx }).then(r => { + if (!Array.isArray(r.path) || r.path.length === 0) { + throw new Error(`${tag}: getSiblingPath(index=${idx}) returned empty/invalid path`); + } + }), + ); + checks.push( + api.findLeafIndices({ treeId: TREE, revision: seedRevision, leaves: [leafBytes[i]], startIndex: 0 }).then(r => { + const got = r.indices?.[0]; + if (got === null || got === undefined || Number(got) !== i) { + throw new Error(`${tag}: findLeafIndices(leaf ${i}) returned index ${String(got)}, expected ${i}`); + } + }), + ); + } + return checks; + }; + + it('Leg C — single connection, sequential (sanity)', async () => { + if (!ipcPath) { + return; + } + const { api, backend } = await makeClient(1); + try { + for (let iter = 0; iter < 40; iter++) { + const idx = iter % N; + const r = await api.findLeafIndices({ + treeId: TREE, + revision: seedRevision, + leaves: [leafBytes[idx]], + startIndex: 0, + }); + const got = r.indices?.[0]; + if (got === null || got === undefined || Number(got) !== idx) { + throw new Error(`Leg C: findLeafIndices(${idx}) returned ${String(got)}`); + } + } + } finally { + await backend.destroy(); + } + }); + + it('Leg A — single connection, contended pipeline (writes + reads interleaved)', async () => { + if (!ipcPath) { + return; + } + const { api, backend } = await makeClient(1); + try { + // Private scratch fork for writes: uncommitted reads on it queue behind each write, driving + // the per-fork scheduler out of arrival order on this single connection. Reads of the seed + // fork run concurrently (different fork), stacking more reorder pressure. + const scratchForkId = (await api.createFork({ latest: true, blockNumber: 0 })).forkId; + const scratchRev = { forkId: scratchForkId, blockNumber: 0xffffffff, includeUncommitted: true }; + for (let iter = 0; iter < 25; iter++) { + const inflight: Promise[] = [...selfCheckingReads(api, `A/iter${iter}`)]; + for (let w = 0; w < 8; w++) { + const extra = [new Fr(BigInt(0x5eed_0000) + BigInt(iter * 100 + w)).toBuffer()]; + inflight.push(api.appendLeaves({ treeId: TREE, leaves: extra, forkId: scratchForkId })); + inflight.push( + api.getTreeInfo({ treeId: TREE, revision: scratchRev }).then(r => { + if (r.treeId !== TREE) { + throw new Error(`A/iter${iter}: scratch getTreeInfo returned treeId=${r.treeId}`); + } + }), + ); + } + await Promise.all(inflight); + } + } finally { + await backend.destroy(); + } + }); + + it('Leg D — multi-connection read/write soak, no churn', async () => { + if (!ipcPath) { + return; + } + const READERS = 3; + const WRITERS = 3; + const conns = await Promise.all(Array.from({ length: READERS + WRITERS }, (_, i) => makeClient(i + 1))); + try { + // Distinct identity per connection; readers keep theirs immutable so roots stay checkable. + const ids = []; + for (let c = 0; c < conns.length; c++) { + ids.push(await plantIdentity(conns[c].api, 100 + c)); + } + + const deadline = Date.now() + SOAK_MS; + const heartbeat = makeHeartbeat('D'); + let round = 0; + do { + const inflight: Promise[] = []; + for (let c = 0; c < READERS; c++) { + inflight.push( + Promise.all(identityChecks(conns[c].api, ids[c], `D/round${round}/reader${c}`, { checkRoot: true })), + ); + } + for (let c = READERS; c < conns.length; c++) { + inflight.push(writerRound(conns[c].api, ids[c], 100 + c, round, `D/round${round}/writer${c}`)); + } + // Fail fast: the first mispaired/malformed response aborts the soak with its tag. + await Promise.all(inflight); + round++; + heartbeat(round); + } while (Date.now() < deadline); + expect(round).toBeGreaterThan(0); + } finally { + await Promise.all(conns.map(({ backend }) => backend.destroy().catch(() => {}))); + } + }); + + it('Leg B — churned multi-connection workload (slot recycling)', async () => { + if (!ipcPath) { + return; + } + const CONNS = 6; + const clients = await Promise.all(Array.from({ length: CONNS }, (_, i) => makeClient(i + 1))); + const ids: ConnIdentity[] = []; + for (let c = 0; c < CONNS; c++) { + ids.push(await plantIdentity(clients[c].api, 200 + c)); + } + try { + const deadline = Date.now() + SOAK_MS; + const heartbeat = makeHeartbeat('B'); + let iter = 0; + do { + // Every connection pipelines identity-verified reads (one writer mixes in writes). Only + // the round's victim may reject, and only with its own destroy error — anything else + // (mispaired value, decode failure, another connection's socket dying) aborts the soak + // immediately with its tag. + const victim = iter % (CONNS - 1); // never the writer, so its size/root model stays valid + const perConn = clients.map(({ api }, c) => { + const tag = `B/iter${iter}/conn${c}`; + const work = + c === CONNS - 1 + ? writerRound(api, ids[c], 200 + c, iter, tag) + : Promise.all(identityChecks(api, ids[c], tag)).then(() => undefined); + return work.catch((e: Error) => { + if (c === victim && /destroyed/.test(e.message)) { + return; // the victim's own in-flight calls reject when we destroy it below + } + throw new Error(`${tag}: ${e.message}`); + }); + }); + + // Churn: while those are in flight, destroy the victim (leaving its requests in flight + // server-side) and immediately replace it (reusing the freed server slot). The + // replacement gets a fresh identity — its old fork stays behind, like a killed AVM sim's. + await clients[victim].backend.destroy().catch(() => {}); + clients[victim] = await makeClient(victim + 1); + + await Promise.all(perConn); + ids[victim] = await plantIdentity(clients[victim].api, 200 + CONNS + iter); + iter++; + heartbeat(iter); + } while (Date.now() < deadline); + } finally { + await Promise.all(clients.map(({ backend }) => backend.destroy().catch(() => {}))); + } + }); +}); diff --git a/yarn-project/world-state/src/native/wsdb_sigpipe_death.test.ts b/yarn-project/world-state/src/native/wsdb_sigpipe_death.test.ts new file mode 100644 index 000000000000..7badf2fcd980 --- /dev/null +++ b/yarn-project/world-state/src/native/wsdb_sigpipe_death.test.ts @@ -0,0 +1,86 @@ +import { Fr } from '@aztec/foundation/curves/bn254'; +import { type IpcClientAsync, UdsIpcClient } from '@aztec/ipc-runtime'; +import { MerkleTreeId } from '@aztec/stdlib/trees'; +import { AsyncApi } from '@aztec/wsdb'; + +import { jest } from '@jest/globals'; + +import { NativeWorldStateService } from './native_world_state.js'; + +// The wsdb server must survive a client that disconnects with requests still in flight: the +// reactor drops the connection's late responses (their connection's reorder state is gone, and +// client ids are never reused) and any write that races the disconnect yields EPIPE, not a +// process-killing SIGPIPE. This is the cross-process guard for +// that invariant — an in-process C++ test cannot reliably raise SIGPIPE because the kernel may +// buffer writes to a just-closed local peer. A long-lived "monitor" connection (models the TS +// world-state client, which never disconnects) proves the blast radius: it must keep reading +// correct answers while an unrelated peer churns. Mid-flight client death is routine in +// production: AVM simulator processes are killed on cancellation and teardown while their wsdb +// requests are outstanding. + +jest.setTimeout(120_000); +const TREE = MerkleTreeId.NOTE_HASH_TREE; + +describe('wsdb server survives a mid-flight client disconnect', () => { + let ws: NativeWorldStateService; + let ipcPath: string | undefined; + let seedRev: { forkId: number; blockNumber: number; includeUncommitted: boolean }; + let leafBytes: Uint8Array[]; + + const connect = async (): Promise<{ api: AsyncApi; backend: IpcClientAsync }> => { + const backend = await UdsIpcClient.connect(ipcPath!, { connectTimeoutMs: 5_000 }); + return { api: new AsyncApi(backend), backend }; + }; + + beforeAll(async () => { + ws = await NativeWorldStateService.tmp(); + try { + ipcPath = ws.getIpcPath(); + } catch { + ipcPath = undefined; // in-process build — no IPC to exercise + return; + } + // The generated AsyncApi is bytes-in/bytes-out (wire `Fr = Uint8Array`), so pass raw buffers. + leafBytes = Array.from({ length: 64 }, (_, i) => new Fr(BigInt(i) * 0x1_0000_0001n + 1n).toBuffer()); + const seeder = await connect(); + const forkId = (await seeder.api.createFork({ latest: true, blockNumber: 0 })).forkId; + seedRev = { forkId, blockNumber: 0xffffffff, includeUncommitted: true }; + await seeder.api.appendLeaves({ treeId: TREE, leaves: leafBytes, forkId }); + await seeder.backend.destroy(); + }); + + afterAll(async () => { + await ws?.close(); + }); + + it('a long-lived monitor keeps working after an unrelated client disconnects mid-flight', async () => { + if (!ipcPath) { + return; + } + const monitor = await connect(); + try { + const before = await monitor.api.getStateReference({ revision: seedRev }); + expect(Array.isArray(before.state) && before.state.length > 0).toBe(true); + + // Victim pipelines many reads, then destroys its connection WITHOUT draining them. The + // server completes them after the disconnect and must drop them cleanly. + for (let round = 0; round < 20; round++) { + const victim = await connect(); + for (let i = 0; i < 200; i++) { + void victim.api.getStateReference({ revision: seedRev }).catch(() => {}); + void victim.api.getSiblingPath({ treeId: TREE, revision: seedRev, leafIndex: i % 64 }).catch(() => {}); + } + await victim.backend.destroy(); // close mid-flight + + // The monitor must still get a correct answer; if the server died this read never + // resolves (or the connection resets), and if a stale frame leaked the decode fails. + const r = await monitor.api.getStateReference({ revision: seedRev }); + if (!Array.isArray(r.state) || r.state.length === 0) { + throw new Error(`round ${round}: monitor read malformed after churn: ${JSON.stringify(r).slice(0, 160)}`); + } + } + } finally { + await monitor.backend.destroy().catch(() => {}); + } + }); +});