Cooperate message transfer at frame boundaries

[RyanKung]

Summary

• Add a local outbound transfer scheduler that admits one frame per actor step, prioritizes DHT/control work, and cancels cleanly when routes or connection generations retire.
• Route inbound reassembly through a bounded mailbox, make storage sync batches yield per entry, and add admission/lifecycle regressions.
• Redact oversized payload diagnostics so logs keep metadata without formatting message bodies.

Validation

• cargo +nightly fmt --all -- --check
• git diff --check
• cargo test -p rings-core --features dummy swarm::transport -- --nocapture
• cargo test -p rings-core --features dummy oversized_payload_log_omits_message_body -- --nocapture
• cargo test -p rings-core --features dummy storage_sync_batch_persists_one_entry_per_step_after_validation -- --nocapture

Closes #669
Closes #672

[RyanKung]

Review

Read both commits in full. The direction is right — collapsing outbound sends into a per-peer actor that cooperates at frame boundaries and lets control-plane work preempt bulk is the correct shape. But there are three blocking problems in the landing, and the abstraction/documentation completeness is well below the bar for a module of this weight.

---

1. Blocking

1.1 yield_core_actor_step is not a yield — it is a round trip through a global timer thread

crates/core/src/message/effects.rs:30-35:

const CORE_ACTOR_STEP_YIELD: Duration = Duration::from_millis(0);
pub(crate) async fn yield_core_actor_step() { sleep(CORE_ACTOR_STEP_YIELD).await; }

crate::utils::sleep is futures_timer::Delay on native. Checking futures-timer 3.0.3: Delay::new allocates an Arc<Node<ScheduledTimer>>, pushes it onto a process-global intrusive list, and calls inner.waker.wake() to wake the global timer helper thread. On the first poll the fired bit is not yet set, so it always returns Pending and must wait for a cross-thread wakeup from that helper thread. Drop pushes onto the list and wakes the thread a second time.

So every "yield" costs: 1 Arc allocation + 2 global-list pushes + 2 cross-thread wakeups of a shared helper thread + 1 task reschedule. And it is now on: every inbound message (swarm/callback.rs:424), every effect (effects.rs:288), every storage-sync entry in every phase, every DHT candidate connect. Every node's entire message hot path is now funneled through one process-wide futures-timer thread.

Delay::poll also contains panic!("timer has gone away") — this introduces a panic site into the message hot path.

wasm is worse: window_sleep(0) lowers to setTimeout(_, 0), and the HTML spec clamps timeouts nested 5+ deep to >= 4ms. A mailbox drain / effect loop / storage batch is exactly a nested-timeout chain. A 100-entry sync batch is ~300 steps ~= 1.2 seconds.

The primitive you want is tokio::task::yield_now() (plus a microtask yield on wasm — Poll::Pending + wake_by_ref — not setTimeout). Not a sleep.

Separately, CORE_ACTOR_STEP_YIELD: Duration = from_millis(0) is a misleading knob in itself: the name says yield, the type says sleep, the value is zero — the three contradict each other.

1.2 The scheduler shreds concurrent chunked transfers within a class against each other

swarm/transport/outbound.rs:590 — after a delivery completes the transfer goes back via self.ready.push(class, ...), i.e. push_back. So a chunked message is re-queued at the tail of its class after every single frame. That is not "cross-class preemption", it is N-way round-robin within a class.

Two consequences:

• Receiver memory. The number of simultaneously in-flight partial messages goes from "1 per peer" to "queue depth" (up to 256). That directly stresses ReassemblyLimits::max_pending_messages: 512 / max_total_buffered_cost, and the constrained() profile (64 ids / 8 MiB) will essentially always hit its ceiling.
• Completion time. N concurrent bulk transfers go from FIFO completion at T, 2T, ..., NT to all of them completing at ~NT. Not one of them finishes early.

The intended semantics are reachable with push_front (or by keeping one active transfer per class until its frames are exhausted): DHT control still preempts, and bulk transfers recover FIFO ordering relative to each other.

This also makes crates/core/src/chunk.rs:7 stale — "There is no mid-message interruption, interleaving, or incremental delivery" no longer holds on the wire.

1.3 retire_active_connection_with now calls tokio::spawn while holding four std Mutex guards

swarm/transport/pending.rs:610 — self.outbound_schedulers.shutdown(attempt.peer) reaches OutboundPeerHandle::shutdown, which calls spawn_outbound_task (tokio::spawn).

That function is synchronous, and the comment directly above it states "These mutations are infallible after the DHT action commits." The last step is now a spawn that requires an active tokio runtime context — calling it off a runtime panics. The "no failure points after commit" property of that block is broken.

Also, the shutdown command travels over the same mpsc that is capacity-256, so it can sit behind 256 already-queued commands for an unbounded time, during which the worker keeps admitting frames on a retired generation. Shutdown is an immediate-effect operation; it should be an AtomicBool flag or a dedicated oneshot, not a message queued behind business traffic and delivered by a spawned task.

---

2. Dead / unreachable paths

2.1 The third branch of TransferQueues::pop can never execute

outbound.rs:204-224. The first if covers has_control && (consec < BURST || !has_lower). What falls through is only: (a) !has_control, or (b) has_control && consec >= BURST && has_lower. In (b), pop_lower() is necessarily Some — has_lower means at least one of the three queues is non-empty, and pop_lower walks all three — so it returns there. In (a), the third if self.has_control() is constantly false. The self.control.pop_front() at the bottom is unreachable.

2.2 The worker holds its own sender, so the channel never closes

OutboundWorker keeps a clone of its input channel's sender (so spawn_delivery_wait can post back). Consequently input_closed, CommandPoll::Closed, and the if self.input_closed && self.waiting.is_empty() exit at outbound.rs:513 are all unreachable in production; the three unit tests around them only reach those states artificially.

Worse: if that path were reachable, run() would spin hot. With input_closed = true, ready empty and waiting non-empty, the loop does not return, self.receiver.next().await on a terminated stream returns None immediately, and it goes around again — pinned CPU, and the waiting transfers can never resolve because delivery events arrive on the same closed channel. Today only the fact that the channel cannot close prevents this. The worker also never self-terminates as a result.

---

3. Abstraction

3.1 The same structure is hand-traversed five times. push (4 arms), pop_lower_from (3 arms), LowerClass::next (3 arms), has_lower (3 checks), len (4 sums). A [VecDeque<T>; 3] plus a usize cursor ((c + 1) % 3) removes the entire LowerClass type and four of those matches. As written, adding a class means editing five places.

3.2 SendCompletion and OutboundCompletion are the same two-variant enum, written twice one module apart, with a From between them. One type is enough.

3.3 OutboundTransfer encodes an exclusive choice as two Option<Sender>s. Exactly one of first_result / final_result is Some (determined by completion), but nothing enforces it; the else if let Some(sender) = self.first_result.take() at outbound.rs:386 exists purely to paper over that unenforced invariant. One Sender plus an OutboundCompletion field is the exact model. Likewise admitted_frames: usize is only ever compared against 0 and 1 (is_before_first_frame / mark_frame_admitted) — the saturating_add is noise on what is really a bool/tri-state.

3.4 One observation, two recording mechanisms. admit_one_frame (async) awaits record_measurement directly; handle_delivery (sync) spawns a detached task for it (spawn_measurement / spawn_cancel_measurement). Write ordering on the same counter becomes nondeterministic. handle_command already runs in an async context, so making handle_delivery async unifies both.

3.5 enqueue_transfer returns Error::ChannelSendMessageFailed on queue overflow (outbound.rs:574). Queue overflow is not a channel send failure, and this error propagates all the way to the send_message caller. Also, capacity is enforced twice (mpsc 256 + active_transfer_count() >= 256), so the real in-flight ceiling is 512 and OUTBOUND_TRANSFER_QUEUE_CAPACITY names neither actual bound.

3.6 with_current(|_| self.outbound_schedulers.handle(did)) misuses that API. Its documented contract is "evaluate a condition while this generation cannot be retired", and the closure signature is FnOnce(&SwarmConnection) -> T. Here the closure ignores the connection and just borrows the lifecycle lock to protect a side effect (creating a worker, spawning a task). Lock ordering matches the retire path so there is no deadlock, but this is using a read-only assertion interface as a critical section. Either give it an honest name, or make scheduler registration an explicit lifecycle operation.

3.7 StorageSyncBatch copies the whole batch and stores each ack twice. validate: msg.data.iter().cloned().collect() — msg is borrowed, and the old code cloned only the accepted entry; now the entire batch is cloned up front. persist.push_back(ack.clone()) + accepted.push(ack) keeps every ack (including its storage entry payload) in memory twice. persist can be a cursor index into accepted, and validate a lifetime-bound index into &msg.data. Also StorageSyncBatchPhase::Ack is a degenerate phase: the Persist -> Ack transition returns Pending, costing one extra yield before returning a result that is already assembled (same for run_effects, which yields once more after its last effect before returning).

3.8 The Preservation comment was dropped. "batch validation is complete before the first storage effect. Invalid placement data cannot leave a partially-written batch." The invariant is still upheld (implicitly, by phase ordering), but it is the whole atomicity argument for the batch and should be restated at the state-machine / type level.

3.9 The mailbox has the wrong shape. In swarm/callback.rs, mailbox is provably of length <= 1 (each iteration pops one and pushes at most one), so VecDeque behind a &mut out-parameter is over-general and works against the pure-core / imperative-shell split. handle_payload returning Option<Bytes> with let mut next = Some(first); while let Some(data) = next.take() { ...; next = self.handle_payload(...)?; } makes the <= 1 bound a type-level fact. Note also that the "bounded mailbox" in the PR description has no bound in it.

3.10 Bytes::copy_from_slice(msg) adds a full copy of every inbound message just to seed the mailbox. The overwhelming majority of messages are not chunks, so that copy is pure waste. A Cow/enum lets the first iteration borrow &[u8].

3.11 handle_payload(&self, _cid: &str, ...)'s cid is now a dead parameter. Renaming it to _cid only silences the warning; drain_message_mailbox just threads it through. The only real use of cid on the whole drain path is Did::from_str(cid) in on_message. It should be removed from both signatures.

3.12 crate::dht::entry::PlacedEntry and std::mem::take are written as inline fully-qualified paths, inconsistent with the repo's one-item-per-use import style.

---

4. Tests

4.1 oversized_payload_log_omits_message_body is vacuous. secret_body is never passed to log_oversized_payload, so assert!(!logs_contain(secret_body)) is trivially true. And the old code already logged only message_kind: &'static str — the field set is identical. "Redact oversized payload diagnostics" in the PR description redacts nothing; this is a pure struct-ification refactor (and OversizedPayloadLog + log_oversized_payload have exactly one call site). Either call it a refactor, or make the test drive the oversize branch of do_send_payload_with_completion with a real CustomMessage body.

4.2 pending_disconnected_before_data_channel_open_is_not_reported has its assertion weakened with no explanation. assert_eq!(events, vec![Connected]) became contains / !contains — from "the sequence is exactly this" to a containment check. This is unrelated to all three summary bullets. If the new yields changed event timing enough to produce extra events, that is a behavior change that needs explaining, not an assertion rewrite.

4.3 A test seam was cut into production code. do_send_payload_with_completion_observing(..., observe_before_scheduler_submit: impl FnOnce()) puts a callback hook on the main send path; the production entry point is now just ..._observing(..., || {}). A dummy controlled queue or an existing lifecycle hook is the right way to get deterministic injection here.

4.4 reassembled_chunks_are_drained_by_mailbox_without_recursive_callback_entry does not test what it is named after. depth=16 passes equally under the old recursive implementation (on_message is an async_trait boxed future; 16 levels will not overflow). The test demonstrates validates == 17 / inbounds == 1, not the non-recursive property.

4.5 dht_control_frame_runs_while_bulk_transfer_waits_for_delivery overclaims in its assertion messages. The bulk transfer is in waiting and the ready queue is empty, so the probe being admitted has nothing to do with class priority — this tests pipelining, not preemption. The message "new control work must be admitted before any new bulk frame" describes a property the test does not exercise. Actual cross-class preemption is covered only by the pure TransferQueues unit tests; there is no end-to-end evidence for it.

4.6 continuous_storage_repair_reaches_remote_owners_across_three_nodes retries with for _ in 0..12. Either 12 is a derived upper bound (in which case write the derivation), or the nondeterminism has not been root-caused.

---

5. Documentation

5.1 outbound.rs is 919 lines with zero doc comments (grep -c '^//!' = 0).

This is the sole implementation of the node's outbound scheduling policy, and none of the following is written down: class priority semantics; OUTBOUND_CONTROL_BURST = 4 (which means control takes 80% of frame slots under sustained load — a policy decision); OUTBOUND_COMMAND_DRAIN_BUDGET = 32; OUTBOUND_TRANSFER_QUEUE_CAPACITY = 256 (none of the three constants explains how its value was chosen); the one-frame-per-step actor invariant; the round-robin semantics. delivery.rs and chunk.rs in the same tree are heavily documented with categorical notation.

One invariant in particular has to be written down: send order is not preserved across classes. This is safe today only because the ordering-sensitive message pairs (E2eHandshake* / E2eStreamFrame, the entry operations) happen to land in the same class. That is a design constraint, not a coincidence — undocumented, the next Message variant assigned to the wrong class is a silent protocol bug.

5.2 crates/core/src/chunk.rs:7 module doc is now stale (see 1.2).

---

6. Minor

6.1 The outbound_schedulers.shutdown(peer) at swarm/transport/connection.rs:361 looks misplaced: that branch handles an orphaned raw connection with no lifecycle record, but creating a scheduler requires an admitted connection, so there should be nothing to shut down there. It also shuts down unconditionally by Did with no generation check. The real shutdown point is retire_active_connection_with.

6.2 Placement of TransferClass / OutboundMessageMeta: "which transfer class does this message belong to" is a property of the message, but it lives in swarm::transport::outbound. crate::message fits the repo's layering better. The exhaustive match means nothing can be missed, so this is low priority.