# flybus A small local RPC and pub/sub bus with immutable file-backed artifacts, for Tokio. One library, one router, one wire protocol. Messages are small JSON envelopes carrying metadata and artifact references. Large immutable bytes live in a file store the router manages. Ownership follows deliveries and explicit holds. The router moves messages and tracks ownership; it does not know what the messages mean. This crate implements the Flybus v1 draft (session-framework design, `bus-v1`, draft 1 of 2026-09-18), using the scalar encodings of its companion `ipc-v1` (`Id`, `U64`, `Digest`). Where this crate narrows or extends the draft, the difference is listed under [Differences from the draft](#differences-from-the-draft). Every sentence of the draft's sections 2 to 11 is audited against this code, with the test that proves it, in `docs/design/session-framework/bus-conformance.md`. Nothing in the crate is specific to a game, a brain or a stream. It is a workspace member; `flysim` embeds a router for the feed (`FLY_FEED_VIA=bus`, `flysim::feedbus`) and `fly-edge` subscribes to it (`docs/design/flybus.md`, "Feed over the bus"). ## Layout | Module | Contents | | --- | --- | | `wire` (public) | Scalars, strict JSON, `Envelope`, `ArtifactRef`, `Attachment`, `Location`, framing, `CONTRACT` and `contract_digest()` | | `error` (public) | `ErrorCode`, `Dispatch`, `BusError` | | `limits` (public) | `Limits` and its hello encoding | | `policy` (public) | `Policy`, `Grants`, `Pattern` | | `router` | `Router`, `RouterConfig`, `RouterStats`, `UnixListenerHandle`; the state machine in `router/state.rs` | | `client` | `Client` and every handle type | | `store` | The file store (router side) and location resolution (client side) | | `transport` | `Transport`, the `Stream` trait | Dependencies: `tokio`, `serde`, `serde_json`, `sha2` and `libc` (for `flock`, `posix_fallocate` and `O_NOFOLLOW`), all already in the workspace lockfile. ## API ### Router ```rust let mut config = RouterConfig::new("/run/example/bus-store"); // default Limits, closed Policy config.policy = Policy::closed() .client("coordinator", Grants { call: vec![Pattern::prefix("agent.")], ..Grants::default() }) .client("agent-a", Grants { register: vec![Pattern::exact("agent.fly-a")], ..Grants::default() }); let router = Router::new(config)?; // creates /, cleans orphans let listener = router.listen_unix_as("/run/example/coordinator.sock", "coordinator").await?; let transport = router.connect_in_memory_as("coordinator"); router.serve_as(any_async_read_write_stream, "coordinator"); router.stats(); // RouterStats (logical registry counters) router.shutdown(); // notices, then closes every connection ``` - `Router` is `Clone`; all clones are one router. `serve_as`, `connect_in_memory_as` and `listen_unix_as` bind the launcher's expected participant id to the transport before Hello. A mismatching Hello is refused before registration or routing. - The unbound `serve`, `connect_in_memory` and `listen_unix` entry points are only for explicit `Policy::open()` trusted/test deployments. Restricted policies refuse unbound transports. Identity in open/unbound mode is self-asserted and is not authentication. - `RouterConfig::hello_timeout` defaults to 5 seconds. `max_clients` counts all accepted connections, pending and active. - `router_id()`, `store_id()`, `store_root()` and `store_dir()` report the incarnation and paths. - `RouterStats` has `connections`, `services`, `topics`, `subscriptions`, `calls` (correlation records, detached included), `active_calls`, `artifacts`, `sealed_artifacts`, `artifact_roots`, `owners`, `reply_capabilities`, `queued`, `store_bytes` and `retained_bytes`. - `Limits` (defaults in brackets): `max_clients` [64], `max_services` [256], `max_topics` [512], `max_subscriptions_per_client` [128], `max_subscriptions` [1024], `max_active_calls_per_client` [64], `max_service_queued` / `max_service_in_flight` [16 / 16], `max_latest_in_flight` [2], `max_bounded_queued` / `max_bounded_in_flight` [64 / 16], `max_owners_per_client` [256], `reserved_owners_per_client` [64], `max_store_bytes` [512 MiB], `max_artifact_bytes` [128 MiB], `max_retained_bytes` [128 MiB], `max_queued_bytes_per_client` [1 MiB], `max_control_frames` / `max_control_bytes` [128 / 1 MiB]. `Router::new` refuses a configuration `Limits::validate` rejects. - `Policy::open()` explicitly enables trusted unbound transports and admits any client id with every grant. `Policy::closed()` admits only the ids added with `.client(id, grants)`. `.with_default(Some(grants))` admits unlisted ids too. `Grants` lists `register`, `call`, `publish`, `subscribe` and `manage_topics` (declare, clear, delete) as `Pattern::Any`, `Pattern::exact(name)` or `Pattern::prefix(prefix)`. An empty list grants nothing. ### Client ```rust let bus = Client::connect(transport, ClientConfig::new("coordinator", store_root)).await?; let bus = Client::connect_unix(socket_path, ClientConfig::new("coordinator", store_root)).await?; bus.info(); // SessionInfo: router_id, connection_id, identity, selected_major, // selected_minor, contract_digest, limits bus.closed(); // Some(reason) once the connection has closed bus.control_errors();// fire-and-forget releases/consumes/unregisters the router refused bus.close().await; // flush queued releases, close, wait for the reader to stop ``` `ClientConfig` has `client_id`, `client_incarnation` (generated when `None`; a reconnect needs a new one), `store_root` (the router's) and `control_lane_capacity` [4096]. `Client` is `Clone`. The connection closes when the client and every handle made from it are dropped, or on `close()`. After that, handles are inert and the router has released what they owned. #### RPC ```rust let mut svc = bus.register("agent.fly-a", ServiceConfig { max_queued: 16, max_in_flight: 16 }).await?; while let Some(req) = svc.next().await { // Request req.method(); req.payload(); req.caller(); // authenticated only on a launcher-bound transport let frame = req.artifact("frame")?; // shares the request's delivery guard req.reply(outcome_map, &[("result", &artifact)]).await?; // Ok(true) routed, Ok(false) detached } // dropping a Request consumes its delivery let mut pending = bus.call("agent.fly-a", Some(&incarnation), "Agent.Prepare", payload, &[("frame", &frame)]).await?; let result = pending.result().await?; // RpcResult: outcome(), responder(), artifact() let state = pending.cancel().await?; // CancelState let result = bus.call_and_wait(service, pin, method, payload, &attachments).await?; ``` - `register` is exclusive; `Service::incarnation()` is the id callers pin. Dropping the `Service` unregisters it. - `call` returns once the router has admitted the call. The `PendingCall` then yields exactly one terminal outcome. `result()` is cancel-safe: wrap it in `tokio::time::timeout`, then call it again or `cancel()`. There is no deadline parameter; the deadline belongs to the caller. - `cancel()` returns `CancelledBeforeDispatch`, `ExecutionUnknown`, `Completed` or `CallGone`. After any state except `Completed`, `result()` fails with `CALL_GONE`. The error's dispatch is `not-dispatched` only for `CancelledBeforeDispatch`. - Dropping an unfinished `PendingCall` sends a best-effort `rpc.cancel`. The reactor consumes any result that still arrives. - `Request::responder()` returns a `Responder` that can reply after the request itself has been dropped. Request delivery credit returns normally; a separate router-visible reply capability preserves correlation until the last `Request`/`Responder` is dropped or a reply is admitted. If the final capability is dropped while the caller is still attached, the call terminates with `CALL_GONE` and dispatch `dispatched` while the route remains live, or the existing `NO_SERVICE`/`dispatched` terminal if the route has ended. Capabilities share the service connection's owner bound. - A call fails with `call.failed` data in its `BusError`: - Queued calls whose service unregisters or disconnects fail with `NO_SERVICE` and dispatch `not-dispatched`. - Dispatched calls whose service disconnects without replying fail with `NO_SERVICE` and dispatch `dispatched`. - Dispatched calls whose last responder is released without replying fail with `CALL_GONE` and dispatch `dispatched`. - A lost connection fails pending calls with `ROUTER_LOST` and dispatch `unknown`. #### Pub/sub ```rust bus.declare_topic("session.demo.snapshots", Retained::Latest).await?; // TopicInfo bus.clear_topic(name).await?; // bool: a retained value was released bus.delete_topic(name).await?; // bool; CONFLICT while it has subscribers let mut sub = bus.subscribe(name, SubscriptionConfig::latest().in_flight(1).replay(true)).await?; let receipt = bus.publish(name, payload, &[("frame", &frame)]).await?; // topic_sequence, subscribers, replaced let msg = sub.next().await; // Option; try_next() does not wait msg.topic_sequence(); msg.replaced(); msg.topic_incarnation(); msg.payload(); msg.artifact("frame")?; ``` - `SubscriptionConfig::latest()` is 1 queued and 2 in flight. `SubscriptionConfig::bounded()` is 64 queued and 16 in flight. `.queued(n)`, `.in_flight(n)` and `.replay(bool)` adjust them. - A credit returns only when the delivery is consumed: when the `Message` and every artifact or `ArtifactFile` taken from it have been dropped. - A retained replay into a bounded subscription is admitted atomically under the same queued envelope-byte quota as an ordinary bounded publication. Latest replay remains outside that byte pool and is bounded by one slot per subscription. - Dropping a `Subscription` unsubscribes and discards its queue. Messages already handed out stay valid. #### Artifacts ```rust let mut writer = bus.artifacts().allocate(len, "image/x-rgba").await?; // ArtifactWriter: io::Write writer.write_all(&pixels)?; let frame = writer.seal().await?; // or seal_with_digest(Some(sha256_hex)) frame.reference(); // ArtifactRef let bytes = frame.read_all().await?; // Vec let file = frame.open().await?; // ArtifactFile: io::Read + io::Seek, keeps the handle let kept = frame.retain().await?; // an independent explicit hold ``` - `Artifact` is a read-only, cloneable handle on one owner: a delivery or an explicit hold. Its last clone (including any `ArtifactFile`) releases that owner. - `ArtifactWriter` is unique. Dropping it unsealed releases the staging storage. Writes past the allocated length fail. Unwritten bytes read as zeros. - Sealing copies staging into a fresh read-only (0444) file and unlinks staging. A descriptor the producer kept or duplicated afterwards writes only to the unlinked staging inode. Sealing therefore needs quota for both copies while it runs. ### Errors `BusError { code: ErrorCode, message, dispatch: Dispatch }`. The codes are the draft's `INVALID_ENVELOPE`, `VERSION_MISMATCH`, `NOT_AUTHORIZED`, `NO_SERVICE`, `TARGET_CHANGED`, `BACKPRESSURE`, `CALL_GONE`, `ARTIFACT_UNSEALED`, `ARTIFACT_GONE`, `OWNER_INVALID`, `QUOTA_EXCEEDED`, `STORE_FAILURE` and `ROUTER_LOST`, plus `CONFLICT`, `NO_TOPIC` and `ARTIFACT_MISMATCH`. `Dispatch` is `not-dispatched`, `dispatched` or `unknown`. Refusals before admission are `not-dispatched`. A command in flight when the connection is lost reports `unknown`. ## Wire summary - Frames are a `u32` little-endian length followed by 1..=65,536 bytes of UTF-8 JSON. The length is checked before any allocation. - Every frame is parsed strictly: - Duplicate keys at any depth, invalid UTF-8, non-finite numbers and trailing bytes are refused. Nesting is limited to 128 levels. - Unknown fields are refused in the envelope, attachments, references and every management body. `payload` and `outcome` are opaque objects. - Frame or envelope errors close the connection after a `connection.closing` notice. They include a bad length, bad JSON, an unknown envelope field, a wrong kind, a non-null `replyTo`, a command id that is not canonical or does not increase, a command before `bus.hello`, a second hello, and a major/minor other than 1.0 after hello. A body-level error (unknown op, bad field, out-of-range value, attachments on a management op) gets an `INVALID_ENVELOPE` reply and the connection stays up. - The SDK applies symmetric checks to router frames: exact 1.0 version, strictly increasing canonical `bus-` ids, kind/replyTo/attachment direction rules, expected reply operation, delivery correlations and complete strict reply, delivery and notice body parsing. - Ids: - Commands: `msg-`, strictly increasing per connection. - Router envelopes: `bus-`. Connections: `conn-`. Services: `svc-`. Topics: `top-`. Subscriptions: `sub-`. Calls: `call-`, increasing per client. - Artifacts: `a-`. Deliveries: `dlv-`. Holds and writers: `own-`. - `routerId` is `router-<16 hex>` and `storeId` is `store-<16 hex>` (same tag). - Delivery and hold serials have separate per-connection watermarks. - Notices: - `call.failed {callId, code, message, dispatch}` - `route.removed {name, serviceIncarnation, reason}` - `subscription.closed {subscriptionId, topic, topicIncarnation, reason}` - `connection.closing {code, message}` - The hello reply's `limits` object gives counts as JSON integers and byte sizes as U64 strings. Its fields: `maxEnvelopeBytes`, `maxAttachments`, `maxBatch`, `maxClients`, `maxServices`, `maxTopics`, `maxSubscriptionsPerClient`, `maxSubscriptions`, `maxActiveCallsPerClient`, `maxServiceQueued`, `maxServiceInFlight`, `maxLatestInFlight`, `maxBoundedQueued`, `maxBoundedInFlight`, `maxOwnersPerClient`, `reservedOwnersPerClient`, `maxControlFrames`, `maxStoreBytes`, `maxArtifactBytes`, `maxRetainedBytes`, `maxQueuedBytesPerClient` and `maxControlBytes`. - `contractDigest` is the SHA-256 of `wire::CONTRACT`, a text listing of every operation, delivery and notice shape. The client refuses a router whose digest differs from its own. ## Semantics worth knowing - **Admission is all or nothing.** `rpc.call`, `rpc.reply` and `publish` validate every attachment and every bound before changing any state. A refused publication creates no delivery, does not move the retained value and spends no topic sequence number. A refused syntactically valid call id advances the issued-id watermark even when semantic admission is refused, while creating no call, delivery or root. Reuse or decrease is rejected. - **Ownership.** Each sealed artifact has a root count. Roots are held by queued deliveries, delivered-but-unconsumed deliveries, explicit holds and a retained topic value. A delivery that names one artifact twice holds one root. Fan-out adds roots and never copies bytes. When the last root goes, the artifact leaves the quota and its file is unlinked, outside the router lock. A process that still has the file open keeps its pages until it closes it. - **Owner budget.** `max_owners_per_client` counts deliveries handed to the client and not consumed, explicit holds and writers. Topic deliveries, holds and writers may use all but `reserved_owners_per_client`; the reserve is for RPC requests and results. A topic delivery over budget waits in its queue. A hold or allocation over budget fails with `QUOTA_EXCEEDED`. Request reply capabilities consume the same finite bound independently of request delivery credit, preventing retained responders from growing detached correlation state without limit. - **Lanes.** The router sends each client its replies and notices first, then RPC requests and results, then topic messages. Topic messages still get one frame after 16 consecutive higher-priority frames. Router envelope ids are assigned only after this scheduler selects the next item, so they increase in actual write order. Replies and notices waiting for one client are bounded by `max_control_frames` / `max_control_bytes`; exceeding either closes that client with a `connection.closing` notice. The client sends consumes, releases and cancels ahead of ordinary commands and batches up to 64 ids per command. A full client control lane closes the connection rather than drop a release. - **No waiting on readers.** A client that stops reading stalls only its own writer task. The router keeps admitting until that client's queues refuse, then returns `BACKPRESSURE` to publishers of bounded topics. Connection teardown is ordered against each synchronous `poll_write`/`poll_flush` call without holding a mutex across an await: it marks the stream closing once, without waiting, and only then waits for a poll already in progress, so no later poll reaches the transport however long the writer had been holding it. Teardown either observes a complete frame before reclaiming its delivery owner, or marks a partial frame canceled before cleanup and appends no final notice to the truncated stream. At a frame boundary, normal final notices are still attempted. - **RPC.** First dispatch is FIFO per service, which includes per caller. A call is marked dispatched when its bytes are about to be written. Replies correlate by call id and may arrive in any order. A second reply to one call fails with `CALL_GONE`. A reply to a detached call returns `routed:false` and creates no roots. Unregistering a service fails its queued calls; dispatched calls can still be answered while a `Request` or `Responder` retains reply capability. Both cancel/consume orderings retire cleanly. - **Seals** copy (and hash, when a digest was given) on Tokio's blocking pool, alongside the connection's reader, so a large copy does not hold up that client's releases. If the writer is released or disconnects mid-seal, the copy is discarded and the artifact is never published. ## Differences from the draft 1. **Extra error codes.** `CONFLICT` covers a duplicate registration, a conflicting topic redeclaration and deleting a topic that has subscribers. `NO_TOPIC` covers publishing or subscribing to an undeclared topic. `ARTIFACT_MISMATCH` covers a seal whose length or digest is wrong, and a reference that disagrees with the artifact it names. All three are now amendments to the draft (bus-v1 section 12), which its inclusive error list allows. 2. **Topics must be declared.** Publish and subscribe fail with `NO_TOPIC` otherwise. `topic.delete` of an unknown topic returns `deleted:false`. `topic.clear` of an unknown topic returns `NO_TOPIC`. 3. **When notices are sent.** - `subscription.closed` is sent only when the router shuts down, because `topic.delete` requires no subscribers and nothing else closes a subscription. - `route.removed` goes to callers with queued or dispatched calls on the removed registration, not to every client. - `connection.closing` is an extra notice that precedes every router-initiated close. 4. **Byte budgets.** No longer a difference: the draft's section 9 table was amended on 2026-09-22 to name the bounded pool and to bound latest slots separately. - `max_queued_bytes_per_client` counts only `bounded` subscriptions. A `latest` slot is bounded by subscription count times envelope size, because a latest subscriber may never be the reason a publication is refused. - `max_retained_bytes` (not in the draft's table) counts the artifact bytes pinned by retained values, once per topic. 5. **Delivery size.** Admission computes the delivery's size with the router-added ids at their longest. If that would exceed 65,536 bytes it refuses with `INVALID_ENVELOPE`, so an inbound envelope near the limit can be refused even though it fits. 6. **Identity.** One live connection per client id. The router remembers each client id's last incarnation and refuses its reuse. Launcher-bound `*_as` endpoints authenticate the Hello id; trusted open/unbound endpoints do not. With an open policy this is one small record per distinct client id ever seen. 7. **Seal reply.** The reply's `ownerId` is the writer's own id, now an explicit hold. 8. **No `budget` argument on calls, and no router executable.** Timeouts are the caller's (`tokio::time::timeout` plus `cancel`); the draft's section 2 sketch was amended on 2026-09-22 to show the deadline there too. The draft's executable is optional; embed `Router`. 9. **Wire strictness.** Management bodies reject unknown fields. After hello, envelopes must carry `minor: 0`. 10. **Reply capability release.** The SDK sends `rpc.responder.release {callId, requestDeliveryId} -> {released}` when the last local reply capability is dropped, an operation the draft does not list and now carries as an amendment (bus-v1 section 12). This keeps request consumption independent from late-reply correlation while bounding that correlation under the service connection's owner limit. For an attached dispatched call, final release atomically retires the correlation and caller slot and emits `call.failed` with dispatch `dispatched`: `CALL_GONE` while the route remains live, or `NO_SERVICE` after route loss. ## Limitations - **One host, one user.** Clients must run as the router's user and see the same store root. The store directory is 0700, staging files 0600, sealed files 0444 and the socket 0600. Bus authority (grants, connection-scoped owner ids, location grants) is enforced on bus operations, not on the filesystem: a process running as the same user can read, mutate, replace, chmod or unlink store files directly and can deny service. Mode 0600 authenticates only the shared OS user, not a Flybus participant. - **No memory maps.** Artifacts are read through `std::fs::File` (`ArtifactFile`) or `read_all()`; there is no mmap API. `ArtifactFile` reads are blocking I/O. - **Stats are logical.** `RouterStats` counts registry entries. Unlinked files that are still open, client memory and OS pages are not in it. - **Fairness is modest.** Fairness between clients is Tokio's scheduling plus a yield after every 32 commands a connection sends. Within a connection, subscriptions and services are served round robin. There is no weighting. - **Orphan cleanup needs a restart.** Store directories left by a stopped router are removed only when a new router starts on the same root. A directory counts as orphaned when it carries the store marker and its `flock` is free. - **Rust only.** There are no other language bindings. - **Two perf runs, not capacity data.** `tests/perf.rs` (ignored by default) measures 640x480 RGBA frames at 60 Hz over a Unix socket to three latest-mode consumers, one delayed 40 ms per frame, with 1, 2 and 4 agent services called every frame. The router gets its own two-thread runtime with a distinct thread name, so its CPU is separable from the clients' in the same process. Two release runs on a shared 4-CPU development VM: producer copy into staging p50 0.4 ms, seal copy p50 1.1 to 1.4 ms, consumer readback p50 0.7 to 1.0 ms, publish admission p50 0.4 ms, RPC round trip p50 0.5 to 1.1 ms; router 0.18 to 0.22 cores, whole process 0.33 to 0.46; 11 to 17 MB RSS; store peak 3.7 MB and zero live after drain. The second run's p99s were three to five times the first's because other work shared the host. The full table, and what it does not claim, are in `docs/design/session-framework/bus-conformance.md`. ## Tests ```text cargo test -p flybus # unit + integration cargo test --release -p flybus --test perf -- --ignored --nocapture # the measurement above cargo run -p flybus --example demo # counter RPC, observer, held frame ``` Every integration test runs twice, once over the in-memory transport and once over a Unix socket, through the same router code: - `tests/wire.rs`: hello negotiation and refusals, malformed frames of every kind, body errors that keep the connection, the exact 65,536-byte limit and control-lane exhaustion. - `tests/rpc.rs`: exclusive and pinned registration, authority, FIFO dispatch with out-of-order completion, backpressure, all four cancel states, single replies, service disconnect and unregister, forged replies and call ids, an endpoint cache replaying an artifact result, and abandoned calls. - `tests/sol_review_races.rs`: launcher-bound identity and Hello resource bounds, detached RPC orderings, responder/service teardown, reply/cancel races, replay admission, call-id watermarking, strict client router validation, unsent rollback and shared task shutdown. - `tests/pubsub.rs`: bounded FIFO with atomic backpressure, latest coalescing, credits, retention and incarnations, zero-subscriber publication, unsubscribe, validation, quotas, a saturated subscriber that does not block RPC, and shutdown notices. - `tests/artifacts.rs`: allocate, seal and read; unsealed use; immutability against live writable descriptors; length and digest checks; quotas; one physical object shared by fan-out and collected after the last consumer; extracted artifacts and holds; atomic admission; release watermarks; connection-scoped owners; abandoned futures; disconnects; and router restarts. - `tests/integration.rs`: two agents called in parallel with a forwarded frame, an environment service, committed snapshot publication, a slow latest consumer and a bounded recorder. - `tests/bus_acceptance.rs`: the implementation guide's BUS-01/02/03 acceptance bullets that the suites above do not already prove, one test per bullet - a lost result, a retransmission fixture, no failover onto a replacement registration, a status RPC answering while another handler is delayed, and a disconnect that reclaims ownership without touching an open file - plus `both_transports_produce_equivalent_behaviour_traces`, which replays one RPC scenario and one pub/sub-and-artifact scenario through the `Trace` recorder in `tests/common/mod.rs` and requires the two transports to record the same 29 behaviour events. `Trace::record` panics on a router-issued id, so a trace cannot drift into operational detail. `FLYBUS_TRACE=1` prints it. It also holds the two rules an audit reviewer found cited but unproven: a latest subscriber flooded with 100 publications of 60 KB never refuses one, and a topic named exactly like a router notice is still delivered as topic data. - `tests/example_demo.rs`: runs `examples/demo.rs` and asserts every line it prints.