[inureyes]

Summary

Implements the six transport and remote pipeline residuals of #2712 in one PR, each wired into the live path (stage service, coordinator exchange, connection pool, TCP transport, router startup) and each backed by a test that fails on 752db4f.

1. Stage errors reach the coordinator. RemoteStageService::handle_activation reports an execution failure, or a middle stage's failed downstream forward, as RemoteStageResponse::Error { nonce, request_id: Some(..), state: Some(..) } to the coordinator (via send_run_error); without a reply nonce it logs one warn and sends nothing. The upstream nonce: 0 send is gone. ExchangeMatch gains error_peers (&self.stage_peers[..] for RunEntry, from_ref(peer) otherwise) and requires the Error to name the run's request_id; forward_sequence now fails in well under 1 s instead of after stage_timeout.
2. ServerAtCapacity backoff. connect_with_retry retries the at-capacity verdict with the refused-connect backoff and returns it, type preserved in the chain, only when the budget is spent; other handshake failures stay terminal. The warn goes through handshake_warn_allowed with the new AtCapacity class.
3. Warning budget. HandshakeWarnClass::RouterDrop (router-drop) and AtCapacity (at-capacity), one window slot per class, router_front.rs uses RouterDrop; WarnDecision::previous_window_ended_ago feeds the summary N cluster warnings ({class}) were logged at debug during the 60s window that ended Ms ago.
4. Unknown authenticated kind is logged. tcp_frame::UnknownAuthenticatedKind { kind } (same Display text), returned by read_frame, classified as diagnostic by handshake_warn_class, and handle_connection logs Closing connection from {peer}: {err} (version skew?) through the budget.
5. Pool reuse and accounting. ProtectedStream::probe_idle() -> IdleProbe { Idle, PendingData, Closed } (peer_closed() kept); acquire reuses only Idle and returns a PooledStream guard (Deref/DerefMut, Arc<AtomicUsize> per peer, decrement on drop) so failed and cancelled sends release their slot; PoolConfig::router() (2 retries from 50 ms, 2 s connect) is used by the router transport in startup.rs.
6. TCP rpc_call round-trips. The response is read on the pooled connection the request went out on; the connection is released only after the matching response and dropped on timeout or error. rpc_pending and the RpcResponse arm of the accept loop are removed (an RpcResponse on an accepted connection is logged at debug).

Docs: docs/distributed.md (capacity backoff, four classes, summary wording), docs/CONTINUOUS_BATCHING.md (router pool profile), and the module docs of remote_service, connection_pool, handshake_warn, tcp_frame, tcp_transport, cluster_handshake. No env vars changed.

For #2714

Item 1's addressing goes through two helpers on RemoteStageService in src/distributed/pipeline/remote_service.rs: fn reply_target(&self, _request_key: &str) -> String (today coordinator_identity with the coordinator.configured() fallback; the Logits RunResult uses it too) and async fn send_run_error(&self, nonce: u64, request_key: String, message: String) -> Result<()>, called from report_activation_failure. Switching reply_target to the per-sequence owner needs no call-site changes.

Failing-first evidence

The behavioral tests were committed first (d7fd980) and run against the unchanged code (752db4f plus only the test files): 11 of 13 failed, each with the expected failure mode.

Test	Failure on 752db4f
remote_service::tests::final_stage_execution_error_goes_to_coordinator	stage must answer within 5s: the coordinator receives nothing (the Error went upstream with nonce 0)
remote_service::tests::final_stage_drops_execution_error_without_reply_nonce	expect_silence(upstream) fails: the upstream received the nonce 0 Error
runtime::correlation_tests::error_from_last_stage_fails_forward_promptly	elapsed assertion: the exchange waited out the 5 s timeout
runtime::correlation_tests::error_from_last_stage_with_wrong_{nonce,request_id}_is_dropped	pass on both sides (main drops every non-addressed Error); they guard the new acceptance rule
tcp_transport::tests::robustness::server_at_capacity_is_retried_with_backoff	cluster handshake with 127.0.0.1:N failed: returned at once, no retry
tcp_transport::tests::robustness::server_at_capacity_exhausts_retry_budget	elapsed under 60 ms: no backoff spent
tcp_transport::tests::robustness::failed_send_releases_active_slot	total_active stayed 1
tcp_transport::tests::robustness::cancelled_send_releases_active_slot	total_active stayed 1
tcp_transport::tests::robustness::rpc_call_round_trips_over_authenticated_tcp	RPC call timed out
tcp_transport::tests::robustness::timed_out_rpc_call_drops_its_stream	total_idle was 1: the stream was pooled before the wait
tcp_transport::tests::frame_protection::authenticated_unknown_kind_is_logged_before_close	zero close lines captured
connection_pool::tests::acquire_drops_idle_connection_with_pending_data	second accept timed out: the idle connection was reused

Tests that use a new API cannot compile on 752db4f; git grep of the symbol on that commit returns nothing, which is the evidence: handshake_warn::tests::router_drops_do_not_suppress_diagnostic_warnings and rollover_summary_reports_the_window_age (HandshakeWarnClass::RouterDrop, WarnDecision::previous_window_ended_ago), classifies_unknown_authenticated_kind_as_diagnostic and frame_protection::read_frame_reports_unknown_authenticated_kind_typed (UnknownAuthenticatedKind), connection_pool::tests::pool_config_router_values and router_pool_config_fails_fast (PoolConfig::router), and the probe_idle assertions added to peer_closed_probe_is_non_consuming (IdleProbe). The behavioral side of each is covered by the table above (items 3 and 4) or by disaggregated_router_e2e (item 5's failover).

Verification

GB10 (spark-102), --profile test-fast --features cuda,test-utils, built outside the lock and run under gpu-lock run --tag issue-2712 with MLXCEL_REQUIRE_MODELS=1 MLXCEL_SDPA_DETERMINISTIC=1 -- --test-threads=1:

Command	Result
cargo fmt --check	clean
cargo clippy --profile test-fast --features cuda,test-utils -p mlxcel --lib --tests -- -D warnings	clean (one deprecated TcpStream::set_linger in a new test was replaced by socket2::SockRef::set_linger on the way)
cargo test ... --lib distributed::	1513 passed, 0 failed, 4 ignored (42 s)
cargo test ... --lib server::router_front	35 passed, 0 failed
cargo test ... --test zero_config_cluster_init -- --include-ignored	7 passed, 0 failed
cargo test ... --test pipeline_remote_real_models -- --include-ignored --skip glm_moe_dsa	10 passed, 0 failed, 1 filtered (82 s)
cargo test ... --test pipeline_server_remote_real_models -- --include-ignored --skip glm_moe_dsa	7 passed, 0 failed, 1 filtered (61 s)
cargo test ... --test pipeline_ci_multi_stage_real_models -- --include-ignored --skip glm_moe_dsa	4 passed, 0 failed
cargo test ... --test disaggregated_serving_e2e -- --include-ignored	2 passed, 0 failed
cargo test ... --test disaggregated_router_e2e -- --include-ignored	7 passed, 0 failed (53 s)

The new tests (25 selected by name, including the adjusted full_connection_cap_reports_capacity_to_the_client, panicking_rpc_handler_releases_its_connection_slot and peer_closed_probe_is_non_consuming) pass on the branch; greedy parity on the real-model targets is unchanged against #2708.

glm_moe_dsa is skipped as the issue allows (418.6 GB checkpoint). make verify-env-var-docs not run: no env var added or changed.

Closes #2712

[inureyes]

Implementation Review Summary

Intent

Fix the six #2712 robustness residuals on the cluster transport, remote pipeline and router, each wired into the live path with a test that fails on 752db4f.

Findings Addressed

None needed: no CRITICAL or HIGH findings at 89f5c19.

Remaining Items

• remote_service.rs:617-621: on a non-entry stage, failures before the downstream send (no downstream peer, sequence_length, RequestId parse, serialize_mlx_array) still propagate with ?, so they end in the coordinator's stage timeout and leave the request's caches in place. The issue only names the execution error and the failed send_activation, so this is outside its scope (LOW).
• connection_pool.rs:279: PooledStream does not record its peer, so release(other_peer, guard) would return the connection to the wrong idle list. The active count stays correct, and every caller passes the same key it acquired with (LOW).
• tcp_transport.rs:315: after the retry budget is spent, Transport::connect logs the "Failed to pre-connect" warning without the rate limiter. This happens once per connect call, not once per attempt (LOW).

Verification

• All stated requirements implemented (items 1 to 6, per-item tests present and asserting what the issue specifies)
• No placeholder/mock code remaining
• Integrated into project code flow (stage service, coordinator exchange, router startup PoolConfig::router(), TCP transport)
• Project conventions followed
• Existing modules reused where applicable
• No unintended structural changes (edits stay inside the item regions; the fix(distributed): tighten tensor decode limits to the model shape #2716 deserialize_* lines and the fix(server): disaggregated router parity residuals from #2700 #2715 router_front.rs regions are untouched)
• Tests pass: cargo check and clippy on --lib --tests (-D warnings), cargo fmt --check, plus lib modules distributed::connection_pool (10), distributed::tcp_transport (38), distributed::handshake_warn (6), distributed::tcp_frame (5) and distributed::pipeline:: (368), all under gpu-lock with --test-threads=1

[inureyes]

Security and Performance Review

Reviewed at 89f5c19: no CRITICAL, HIGH or MEDIUM findings, so nothing was committed.

Checked and clean

• RunEntry Error accepted from any stage peer: it still needs the random per-exchange nonce and request_id == Some(<run id>). Only nonce holders can match, and each could already stall the request into the same end (server_runtime.rs:154 panics on any forward error, timeouts included). The Error's state and stage_index are ignored, and every entry-stage RunEntry error already carries Some(request_id).
• PooledStream pairs one increment with one decrement per guard, covering failed, cancelled and released sends. probe_idle is a non-blocking MSG_PEEK.
• UnknownAuthenticatedKind logs the bound SocketAddr identity and a u8, so the line cannot be injected into. It needs the secret and is rate limited under Diagnostic.
• Removing rpc_pending also closes a hole: any authenticated peer could resolve another peer's pending RPC with a guessed sequential id.

Remaining (LOW, not fixed)

• connection_pool.rs:338, :394: at-capacity retries have no jitter, and each one is a full connect plus handshake. N concurrent acquires against a full peer make up to 6N handshakes (3N with the router profile), each writing an unrated server warn (tcp_transport.rs:614). The peer must be authenticated and the total is bounded. Suggested fix: rate limit that warn under AtCapacity and add jitter.
• tcp_transport.rs:414: rpc_call reads with the 256 MiB max_frame_size. The length header is unauthenticated until the body is opened (tcp_frame.rs:225), so an on-path party can force one large allocation per in-flight call, as on accepted connections today.
• remote_service.rs:737: the no-nonce failure warn is not rate limited, like the logits-drop warn at :667. Only the authenticated upstream can trigger it.

Verification

• cargo fmt --check and clippy --lib --tests -D warnings are clean.
• Under gpu-lock with --test-threads=1: distributed::{connection_pool,tcp_transport,tcp_frame,handshake_warn,pipeline::} ran 427 tests, all passed.