feat(ipc): request-id envelopes on every frame; responses in completion order

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:

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<IpcClient> create_socket(const std::string& socket_path);
   static std::unique_ptr<IpcClient> create_shm(const std::string& base_name);
   static std::unique_ptr<IpcClient> 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<const uint8_t>   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<const uint8_t>   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<const uint8_t>           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<const uint8_t> 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<const uint8_t> 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<IpcServer> make_server(const std::string& path, const ServerOptions& = {});
-std::unique_ptr<IpcClient> make_client(const std::string& path, std::size_t shm_client_id = 0);
+std::unique_ptr<IpcClient> 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
 

ipc-runtime/cpp/ipc_runtime/c_abi.cpp

@@ -129,19 +129,22 @@ 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) {
         *out = nullptr;
         *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)

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" */

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.