3.2 ZmqIpcBackend

ZmqIpcBackend

Feature-gated (zmq) IpcBackend that subscribes to a remote compute PUB socket and returns the latest valid float readout on each process_batch. Implementation: src/zmq_backend.rs. Shared parser: src/zmq_readout.rs. Deprecated alias: ZmqRuntimeBackend.

Does not implement HybridFlowBackend. Structured IpcMessage PUB/SUB lives in zmq_message (ZmqHybridFlowBackend) as of 08d2f28 / #55.

Concurrency

libzmq sockets are Send + !Sync. ZmqIpcBackend stores the socket in Mutex<Option<ExclusiveSocket>> so the public type is Send + Sync without unsafe impl Sync on zmq::Socket. initialize / process_batch / reset take &mut self for the readout cache; save_state / get_spike_states take &self and do not touch the socket.

Since #62, ZmqIpcBackend::new() takes its context from the crate-private shared_context() instead of zmq::Context::new(). Every socket in the crate shares one process-wide libzmq context, held in a OnceLock until process exit, because libzmq 4.3.4 segfaults in its command mailbox when several contexts call zmq_connect concurrently. Sockets still have a single owner. The Linux build.rs API lock that #62 added was removed in #65 because it never reached downstream binaries (7.1); the crate ships no libzmq patch.

Endpoint and socket setup

On initialize:

1. Endpoint from CORPUS_IPC_ZMQ_READOUT_IPC, else ipc:///tmp/corpus_ipc_readout.ipc.
2. Create SUB, set_subscribe(b""), set_rcvhwm(16).
3. Set ZMQ_RECONNECT_IVL / ZMQ_RECONNECT_IVL_MAX before connect (defaults 100 ms / 5000 ms; override with CORPUS_IPC_ZMQ_RECONNECT_IVL_MS / CORPUS_IPC_ZMQ_RECONNECT_IVL_MAX_MS, positive ms only).
4. Store under the mutex. Idempotent after success until reset.

Publisher late-start / stop / restart at the same endpoint is handled by libzmq's reconnect state machine (subscribe filter is re-sent). There is no crate-owned retry thread. reset + initialize remains the path for switching endpoints.

Latest-value drain (process_batch)

Constants: RECEIVE_DRAIN_MAX_MESSAGES = 64, RECEIVE_DRAIN_MAX_DURATION = 5 ms, READOUT_RCVHWM = 16.

sequenceDiagram
  participant Caller
  participant ZB as "ZmqIpcBackend"
  participant Sock as "SUB socket"
  Caller->>ZB: process_batch
  loop until EAGAIN or 64 msgs or 5 ms
    ZB->>Sock: recv_bytes DONTWAIT
    alt Ok bytes
      Sock-->>ZB: frame
      ZB->>ZB: parse_readout_packet
      Note over ZB: prior valid frame increments skipped_readouts
    else EAGAIN
      Sock-->>ZB: EAGAIN
      Note over ZB: stop drain, return newest or cache
    else malformed or over cap
      Sock-->>ZB: bad frame
      ZB-->>Caller: BackendError fail-closed
    end
  end
  ZB-->>Caller: Ok Vec of f32

Loading

• A newer valid frame supersedes the previous; skipped_readouts() counts valid frames dropped by this policy (not silent libzmq HWM drops).
• EAGAIN with an empty drain returns the cached last_readout without error.
• Malformed or over-limit frames return BackendError and do not update the cache (malformed_readouts increments on parse failure).

Wire format

Bytes	Field
0..8	tick i64 LE
8..	N × f32 LE

An 88-byte frame is tick + 20 floats. Length alone cannot distinguish that from a historical 16-readout + 4-modulator layout; this backend does not infer or strip neuromodulator scores. Cap: default 1024 floats (CORPUS_IPC_ZMQ_MAX_READOUT_FLOATS). See 3.3.

Reset / Drop

reset clears cache/tick/counters and drops the socket under the mutex (zmq_close, default linger). Drop closes through exclusive ownership. After reset, callers must initialize again before process_batch.