3.4 Structured Zmq IpcMessage Transport

Structured ZeroMQ IpcMessage Transport

Feature-gated (zmq) PUB/SUB transport that carries one compact canonical JSON WireEnvelope of IpcMessage per ZeroMQ message. Added on main @ 08d2f28 (#55; CHANGELOG Unreleased / LIM-1278).

This is a separate protocol surface from binary readout ZmqIpcBackend and from the REST server. Sources: src/zmq_message/mod.rs, src/zmq_message/transport.rs, src/zmq_message/hybrid.rs.

Three surfaces (do not mix framings)

Surface	Feature	Framing	Types
Structured ZeroMQ (this page)	zmq	one canonical JSON WireEnvelope of IpcMessage	ZmqMessagePublisher, ZmqMessageSubscriber, ZmqHybridFlowBackend
Binary readout	zmq	[i64 tick][N × f32 LE]	ZmqIpcBackend only
REST	server	HTTP JSON around IpcBackend	corpus_ipc_server

ZmqIpcBackend does not implement HybridFlowBackend. ZmqHybridFlowBackend does not implement IpcBackend.

Types

Type	Role
ZmqTransportConfig	endpoint, send/receive HWM, linger, reconnect interval/max — applied before bind/connect
ZmqMessagePublisher	binds PUB; validates with Validate, then encode_canonical_ipc_message, then send
ZmqMessageSubscriber	connects SUB (empty subscription prefix); try_recv → decode_ipc_message_json + Validate
ZmqHybridFlowBackend	HybridFlowBackend adapter over that pair
ZmqMessageError	structured-transport errors (InvalidConfig, WouldBlock, HighWaterMark, Closed, framing, envelope, …)
FramingFault	Empty or Multipart
STRUCTURED_MAX_FRAME_BYTES	512 MiB (ZMQ_MAXMSGSIZE); publisher rejects oversized encoded frames before send

Root re-exports (feature zmq): see src/lib.rs.

ZmqTransportConfig defaults (new)

Field	Default
send_hwm / receive_hwm	1000 (0 = unlimited, matching libzmq)
linger_ms	0 (discard unsent on close)
reconnect_ivl_ms	100
reconnect_ivl_max_ms	5000 (must be ≥ interval; libzmq ignores a smaller cap)

Topology and delivery

Publisher binds; subscriber connects. PUB/SUB is best-effort and at-most-once (no acknowledgements): slow-joiner loss, HWM drops, and linger 0 teardown all discard frames. A successful send means the local socket accepted the frame, not that a peer observed it. Control/config variants use the same contract.

Inbound frames above STRUCTURED_MAX_FRAME_BYTES are rejected by libzmq before allocate. Multipart input is drained up to eight extra frames: a message that fits in that drain returns a framing error and leaves the subscriber open; frames still remaining after the drain close the subscriber.

Sockets stay mutex-serialized Send + Sync with no unsafe impl Sync. ZmqMessagePublisher::bind and ZmqMessageSubscriber::connect use the crate-wide shared context from src/zmq_context.rs (#62), which stays alive until process exit, so closing one socket never tears the context down under another thread.

ZmqHybridFlowBackend

open(outbound, inbound) binds PUB + connects SUB. Identical outbound and inbound endpoints are rejected before either socket opens.

Trait method	IpcMessage variant	Direction
send_spikes	Spikes	out
send_embeddings	Embeddings	out
try_receive_gradients	GradientUpdate	in
try_receive_traces	EligibilityTraces	in

Sibling hold and protocol rejection

• The other supported inbound variant (gradient while receiving traces, or vice versa) is held one message per variant for its own method.
• A second held sibling returns BackendError::CommunicationError and consumes the new frame (held frame stays).
• Any unrelated inbound variant returns BackendError::Protocol (ProtocolError::UnexpectedVariant); the frame is consumed, not dropped quietly.
• Payload validation failures inside an envelope surface as BackendError::InvalidInput.
• A closed publisher reports Closed / mapped communication error before payload validation on send.
• Non-blocking empty receive → Ok(None) on the hybrid methods (WouldBlock on the raw subscriber).

Flow

flowchart LR
  App["Caller"] -->|"send_spikes / send_embeddings"| Hybrid["ZmqHybridFlowBackend"]
  Hybrid -->|"IpcMessage::Spikes / Embeddings"| Pub["ZmqMessagePublisher.bind"]
  Pub -->|"Validate + encode_canonical_ipc_message"| WireOut["WireEnvelope JSON frame"]
  WireOut -->|"PUB"| PeerSub["Peer SUB"]
  PeerPub["Peer PUB"] -->|"SUB"| Sub["ZmqMessageSubscriber.connect"]
  Sub -->|"decode_ipc_message_json + Validate"| Hybrid
  Hybrid -->|"try_receive_gradients / traces"| App
  Hold["held_gradient / held_trace slots"] -.-> Hybrid

Loading

Tests in-tree

Unit/live coverage under src/zmq_message/ (tests.rs, transport_tests.rs, framing_tests.rs, hybrid_live.rs) — feature zmq.