communication: vendored Rust transport (rust/) + RustZMQCommunicator (RFC #130 Step 1)

[npuichigo]

Step 1 of the migration plan in #130, as requested — self-contained: the Rust transport is vendored into rust/ (no dependency on the mstar-rs repo), with RustZMQCommunicator landing side-by-side so it can be selected per-entity rather than replacing the pyzmq class in one cut.

rust/ — the ZMQ transport (opaque byte frames over PUSH/PULL; ipc + tcp endpoints; wakeup-fd polling via zmq_poll raw-fd items; a typed Codec layer as the pickle→msgpack migration seam) + a PyO3 module mstar_rust. Build with maturin develop (or pip install .) in rust/.

RustZMQCommunicator — drop-in for ZMQCommunicator: same constructor/methods/endpoints (ipc prefix scheme and the MSTAR_ZMQ_* TCP host/port map), wire-compatible with unwrapped pyzmq entities in both directions (pickle-default codec=(dumps, loads) over a no-framing transport — migrate one process at a time); register_event_for_poll forwards the EventWakeup fd to the Rust poller with drain semantics unchanged; and a buffering bridge for the one semantic mismatch (your poller reports readiness without consuming; the Rust receive consumes — a frame consumed during a poll is delivered by the next get_all_new_messages, FIFO intact).

Tests — test/modular/test_rust_communicator.py (pytest.importorskip("mstar_rust"), so CI skips until the extension is built): pickle interop pyzmq↔Rust both directions on one mesh; an EventWakeup fire cuts wait_for_work to ~50 ms against a 2 s timeout; lossless FIFO readiness polls. All 3 pass locally against the vendored build.

[NSagan271]

Overall looks good. I also cloned the repo and did spot checks on the Orpheus, Qwen3-Omni, and Bagel benchmarks. Only comment from testing: I got some build warnings about dead code, should maybe add #[warn(unused)] where appropriate

Also inviting @merceod to comment as well if you have time, since this is the first PR adding Rust to our system.

[npuichigo]

Thanks for the careful review — all comments addressed in b0e29f2..3fc0753:

• Dead-code warnings: the warnings came from the typed ZmqCommunicator<M, C>/Codec layer being unreachable from a cdylib-only crate with a private module. Rather than #[allow]ing it, the crate now also builds as a Rust library (crate-type = ["cdylib", "rlib"] + pub mod communicator) — that layer is the API the later migration steps (Rust conductor/API server) consume, and cargo build is warning-free.
• blocking honored in get_all_new_messages on both communicators (the pyzmq one had the same latent bug), with tests for both.
• MSTAR_RUST_ZMQ is 0/1/AUTO; default left at 0 pending your "fully tested" call — flipping to AUTO is a one-word change whenever you want it.
• Docs: docs/environment_variables.rst starts the env-var documentation precedent; Rust setup instructions are in installation.rst + README (with explicit maturin develop --release, since debug builds cost real latency on the receive path).
• Codec is a small class mirroring the Rust trait; _endpoint/_tcp_port are hoisted to BaseCommunicator.

CI is green on the branch. Happy to iterate further, and looking forward to @merceod's take on the Rust layout.

[npuichigo]

One more change worth flagging (2f1870a): swapped the typed layer's default codec from bincode to MessagePack (rmp-serde). Two reasons:

1. bincode is a governance risk — the 1.x line we pinned is frozen, and upstream has moved off GitHub with no commitment to further releases there.
2. msgpack is where the codec seam is headed anyway: it's the language-neutral wire this migration targets, and to_vec_named frames (field-name maps) are directly readable by Python peers with the msgpack package — so the same codec serves Rust-internal typed messaging and future cross-language edges.

The transport is unaffected (it moves opaque frames); only the Rust-internal typed default changes. cargo test and the interop pytest are green.

[npuichigo]

A/B of the two communicators (same pickle codec both sides, ipc endpoints, single box — isolates the transport layer):

Metric	pyzmq	Rust	Ratio
RTT p50 (2 hops)	68.4 µs	54.1 µs	1.26×
RTT p99	82.8 µs	86.9 µs	0.95× (parity)
Throughput, 1 KB msgs	75.9k msg/s	127.0k msg/s	1.67×
Throughput, 64 KB	159 MB/s	191 MB/s	1.20×
Throughput, 1 MB	193 MB/s	277 MB/s	1.43×
Drain of a 512-msg burst	3.49 ms	0.97 ms	3.59×

The deltas line up with the implementation differences: small-message throughput is the per-message FFI path, the 1 MB rate is the single-copy receive (zmq::Message end to end), and the burst drain is get_all_new_messages picking up the whole queue in one crossing instead of one per message. p99 is honest parity — at the tail, scheduler noise dominates both.

To be clear about magnitude: pyzmq is already fast, and at today's control-message rates neither transport is a bottleneck. The claim this supports is "parity or better on every metric, measured" — the migration doesn't cost anything on the way to the later steps.

[npuichigo]

Is there anything else that needs to be updated?

[NSagan271]

Is there anything else that needs to be updated?

Not that I can think of right now. I will take another look through the PR later today before approving just to be sure.

[NSagan271]

Minor comments, then it should be good to go on my end.

[NSagan271]

Looks good on my end, and I jut did one last spot check (orpheus benchmark). I'd wait for a review/go-ahead from @merceod before merging.

[merceod]

@npuichigo Thanks for the PR. Most things look great to me!

I left a few comments. Please let me know if you disagree with any of them (I wrote the comments based on my understanding of the code so please correct me if I'm am wrong somewhere). Most of the comments are minor changes and overall the code looks great.

@NSagan271 please take a look too.

One thing:

I think it might be worth adding one comment in the README about what AUTO-by-default means: on any machine where the extension is built, the whole mesh switches to Rust with no configuration change (the current "selectable per process" framing undersells this and since all entities of one mstar-serve launch inherit the same environment, the flag is effectively per-launch-tree, not per-entity; true per-process mixing only happens across separately launched processes). Fine as designed — just deserves saying.

[npuichigo]

@merceod Thanks — all five are correct, and the mutex one especially (nice probe; the two-thread measurement is exactly the failure mode the later multi-threaded consumers would have hit). Everything is addressed in 838fb63:

• Lock held across send — the peers map now stores Arc<Mutex<Socket>>: the map lock covers only lookup/insert, the per-socket lock serializes sends to one peer (per-peer FIFO preserved, since a zmq socket is single-user anyway), and an HWM-blocked send now stalls only that peer, not the process's whole control plane (or register_peer).

• poll_for_messages parity — added the pyzmq twin (the poller without consuming, wakeup drained exactly as in wait_for_work), so the factory's drop-in guarantee holds again; a parity test drives the same contract through both classes.

• Silent decode drop — took your primary suggestion: Codec::decode -> Result<M, CommError>, with try_recv/recv/recv_timeout returning Result<Option<M>> and drain Result<Vec<M>>, so codec/version skew is an error distinct from "no message" in the API Steps 4/5 inherit. CommError::Decode carries the frame length; a new test proves garbage surfaces as Err, never None.

• Transport observability — the factory logs control mesh transport: rust <version> (MSTAR_RUST_ZMQ=...) (or pyzmq) at INFO on construction; mstar_rust now exports __version__ (the crate version), and the factory warns when it differs from the tree's expected constant — the stale-wheel-under-AUTO case you described now announces itself. README got the AUTO-by-default operational note from your review body.

• POLLNVAL on wakeup fds — you're right, and it's cheap: recv_or_wake now checks the wakeup items' revents for POLLERR/POLLNVAL and returns a distinct WakeFdError, which the Python binding raises as a loud RuntimeError instead of letting every blocking wait degrade into an instant-timeout spin.

Stack rebased on the new head. Happy to adjust any of these if you'd prefer a different shape.

[merceod]

@merceod Thanks — all five are correct, and the mutex one especially (nice probe; the two-thread measurement is exactly the failure mode the later multi-threaded consumers would have hit). Everything is addressed in 838fb63:

• Lock held across send — the peers map now stores Arc<Mutex<Socket>>: the map lock covers only lookup/insert, the per-socket lock serializes sends to one peer (per-peer FIFO preserved, since a zmq socket is single-user anyway), and an HWM-blocked send now stalls only that peer, not the process's whole control plane (or register_peer).
• poll_for_messages parity — added the pyzmq twin (the poller without consuming, wakeup drained exactly as in wait_for_work), so the factory's drop-in guarantee holds again; a parity test drives the same contract through both classes.
• Silent decode drop — took your primary suggestion: Codec::decode -> Result<M, CommError>, with try_recv/recv/recv_timeout returning Result<Option<M>> and drain Result<Vec<M>>, so codec/version skew is an error distinct from "no message" in the API Steps 4/5 inherit. CommError::Decode carries the frame length; a new test proves garbage surfaces as Err, never None.
• Transport observability — the factory logs control mesh transport: rust <version> (MSTAR_RUST_ZMQ=...) (or pyzmq) at INFO on construction; mstar_rust now exports __version__ (the crate version), and the factory warns when it differs from the tree's expected constant — the stale-wheel-under-AUTO case you described now announces itself. README got the AUTO-by-default operational note from your review body.
• POLLNVAL on wakeup fds — you're right, and it's cheap: recv_or_wake now checks the wakeup items' revents for POLLERR/POLLNVAL and returns a distinct WakeFdError, which the Python binding raises as a loud RuntimeError instead of letting every blocking wait degrade into an instant-timeout spin.

Stack rebased on the new head. Happy to adjust any of these if you'd prefer a different shape.

Hey @npuichigo

Everything looks good.

Confirmed the lock fix. Re-ran the same two-thread probe on 838fb63: healthy-peer send under an HWM-blocked flood is now 0.2 ms (was ~6 s), per-peer FIFO intact.

On the SNDTIMEO half (my first comment): let's explicitly defer it out of this PR rather than leave it hanging. Two reasons: no call site today handles a send error, so a timeout would just turn a stalled send into an unhandled exception mid-loop (worse than the stall); and adding it only on the Rust side would break send-semantics parity between the two transports. The right home for it is the worker-liveness/failure-propagation work, where a timed-out send has somewhere to go (fail the affected requests, mask the peer). Can you open a small tracking issue referencing this thread so it doesn't get lost - then this is resolved from my side?

One optional nit while you're in the file: in the recv_or_wake fd guard, PollEvents::from_bits_truncate(0x20) truncates to empty (0x20 isn't a defined zmq flag) — the guard works because libzmq folds raw-fd POLLNVAL/POLLERR into ZMQ_POLLERR, which the other half of the check catches. The dead half can be dropped for clarity.

@NSagan271 thoughts?

[npuichigo]

Both done:

• Dead half of the fd guard dropped in 5b18c1d — you're right that from_bits_truncate(0x20) truncates to empty; kept only the POLLERR check with a comment noting libzmq folds a raw fd's POLLNVAL/POLLERR into ZMQ_POLLERR (verified the loud-failure behavior is unchanged, crate tests green).
• Tracking issue for the SNDTIMEO/failure-propagation deferral: Control-mesh send timeout (SNDTIMEO) + peer-failure propagation #173 — framed exactly as you laid out (no call site can handle a send error today; single-transport SNDTIMEO would break parity; belongs with worker-liveness where a timed-out send can fail requests / mask the peer).

[merceod]

Perfect!

[merceod]

LGTM!