warning: in the working copy of 'crates/spt-daemon/tests/input_ack_deadlock.rs', LF will be replaced by CRLF the next time Git touches it diff --git a/crates/spt-daemon/tests/input_ack_deadlock.rs b/crates/spt-daemon/tests/input_ack_deadlock.rs index 43aeaf88..952122f3 100644 --- a/crates/spt-daemon/tests/input_ack_deadlock.rs +++ b/crates/spt-daemon/tests/input_ack_deadlock.rs @@ -69,2 +69,2 @@ use std::time::{Duration, Instant}; -use spt_daemon::attach::{request_attach, send_attach_input, serve_attach}; -use spt_daemon::brain::{Brain, BrokerEvent}; +use spt_daemon::attach::{request_attach, send_attach_input, send_attach_resize, serve_attach}; +use spt_daemon::brain::{Brain, BrokerEvent, PumpTrace}; @@ -100,0 +101 @@ static SEQ: AtomicU32 = AtomicU32::new(0); +const RETAINED_OUTPUT: &[u8] = b"ACKDL_OUTPUT"; @@ -117,7 +118,7 @@ fn kill_pid(pid: u32) { -/// A QUIET child: it neither reads stdin nor writes stdout (a long sleep). This is -/// the clean ack-deadlock substrate — the flooded input is consumed by the PTY -/// writer with NO echo, so the ONLY thing that can back up the brain↔broker conn is -/// the pre-fix APPLIED-ACK stream (not echoed output). An echo/flood child would -/// confound this gate with the W1 output-drain hazard (output backing up the -/// non-draining controller conn), so we deliberately avoid any child output here. -fn quiet_spawn_req(endpoint: &str) -> SpawnReq { +/// A SEEDED-THEN-QUIET child: it writes exactly one retained-output marker before +/// the flood, then neither reads stdin nor writes again. The seed makes the later +/// bounded replay diagnostic deterministic; waiting for the broker's output seq +/// before starting the flood keeps ALL child output outside the deadlock substrate. +/// During the flood, input is consumed by the PTY writer with NO echo, so the only +/// thing that can back up the brain↔broker conn is the pre-fix APPLIED-ACK stream. +fn seeded_quiet_spawn_req(endpoint: &str) -> SpawnReq { @@ -125 +126,7 @@ fn quiet_spawn_req(endpoint: &str) -> SpawnReq { - let (program, args) = ("sleep".to_string(), vec!["600".to_string()]); + let (program, args) = ( + "sh".to_string(), + vec![ + "-c".to_string(), + "printf 'ACKDL_OUTPUT\\n'; exec sleep 600".to_string(), + ], + ); @@ -127,2 +133,0 @@ fn quiet_spawn_req(endpoint: &str) -> SpawnReq { - // `waitfor` blocks up to /t seconds for a signal that never comes: it produces - // NO stdout and does NOT read stdin — the Windows "silent sleep" we need. @@ -130 +135 @@ fn quiet_spawn_req(endpoint: &str) -> SpawnReq { - "waitfor".to_string(), + "cmd".to_string(), @@ -132,3 +137,2 @@ fn quiet_spawn_req(endpoint: &str) -> SpawnReq { - "/t".to_string(), - "600".to_string(), - "AckDlNoSignal".to_string(), + "/C".to_string(), + "echo ACKDL_OUTPUT & ping -n 600 127.0.0.1 >nul".to_string(), @@ -151,16 +154,0 @@ fn quiet_spawn_req(endpoint: &str) -> SpawnReq { -fn count(hay: &[u8], needle: &[u8]) -> usize { - if needle.is_empty() || hay.len() < needle.len() { - return 0; - } - let (mut n, mut i) = (0usize, 0usize); - while i + needle.len() <= hay.len() { - if &hay[i..i + needle.len()] == needle { - n += 1; - i += needle.len(); - } else { - i += 1; - } - } - n -} - @@ -216,0 +205,19 @@ fn wait_for_stream(brain: &mut Brain) -> Option<(u64, String)> { +/// Like [`wait_for_stream`], but selects the newest peer stream. A fresh Brain +/// sees broker-global historical stream rows too; `.find()` would repeatedly +/// return the flood controller's older row instead of the later VIEWER stream. +fn wait_for_latest_stream(brain: &mut Brain) -> Option<(u64, String)> { + for _ in 0..400 { + let reply = brain.net_streams().expect("net-streams"); + if let Some(s) = reply + .streams + .iter() + .filter(|s| !s.initiated_locally) + .max_by_key(|s| s.stream_id) + { + return Some((s.stream_id, s.remote_id_hex.clone())); + } + thread::sleep(Duration::from_millis(5)); + } + None +} + @@ -278,4 +285,4 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { - // ── The quiet child: ignores stdin, produces NO output. The flooded input is - // consumed by the PTY writer without echo, so the only thing that can back up - // the brain↔broker conn is the pre-fix applied-ack stream — a clean substrate - // for the ack-deadlock gate (no W1 output-drain confound). ── + // ── The seeded-then-quiet child: emit ONE retained marker, then go silent. + // The asserted seq wait — never a blind sleep — proves the marker reached + // the broker before the flood starts. During the flood the child neither + // reads nor writes, preserving the clean applied-ack deadlock substrate. @@ -284,2 +291,10 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { - .spawn_session(quiet_spawn_req(endpoint)) - .expect("spawn quiet child"); + .spawn_session(seeded_quiet_spawn_req(endpoint)) + .expect("spawn seeded quiet child"); + let seed_deadline = Instant::now() + Duration::from_secs(10); + while broker.session_output_seq(sid).unwrap_or(0) == 0 && Instant::now() < seed_deadline { + thread::sleep(Duration::from_millis(10)); + } + assert!( + broker.session_output_seq(sid).unwrap_or(0) > 0, + "pre-flood ACKDL_OUTPUT seed must be retained before the deadlock substrate starts" + ); @@ -290,2 +305,3 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { - // send_effect_no_ack per record (post-fix). The flood child also floods - // stdout so the concurrent viewer below has output to actually receive. + // send_effect_no_ack per record (post-fix). The child is now silent: no + // concurrent output can confound the applied-ack deadlock substrate. The + // bounded receive diagnostic replays its retained pre-flood seed below. @@ -399,3 +415 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { - if let Err(error) = - send_attach_input(&mut operator, stream, line.as_bytes(), op) - { + if let Err(error) = send_attach_input(&mut operator, stream, line.as_bytes(), op) { @@ -413,4 +427,5 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { - // serve thread WITHOUT a blocking drain loop: a Whole conn's read ignores - // any deadline, so a drain loop here would hang the helper. The main thread's - // child-kill + broker drop unwinds the abandoned serve thread (serve threads - // are abandoned by design here — never block the gate on one). + // serve thread WITHOUT a drain loop: this Whole carrier supports only an + // unbounded read; asking it for a deadline is refused by name. An unbounded + // drain could hang the helper. The main thread's child-kill + broker drop + // unwinds the abandoned serve thread (serve threads are abandoned by design + // here — never block the gate on one). @@ -428,3 +443 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { - Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { - FloodVerdict::HelperDisconnected - } + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => FloodVerdict::HelperDisconnected, @@ -467 +480 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { - // subscribes, and RECEIVES the flood child's output. Pre-fix the per-conn + // subscribes, and replays the retained pre-flood output. Pre-fix the per-conn @@ -474 +487,9 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { - let mut operator = match Brain::cold_start(&attach_name, 1) { + // The result-gating viewer needs bounded reads below, so it must ride the + // Split carrier. Its 10s per-call ceiling sits above the preserved 8s outer + // observation budget; the outer loop remains the diagnostic's owner. + let mut operator = match Brain::cold_start_pump( + &attach_name, + 1, + Duration::from_secs(10), + PumpTrace::Stderr, + ) { @@ -508 +529 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { - let (stream_a, origin) = match wait_for_stream(&mut target) { + let (stream_a, origin) = match wait_for_latest_stream(&mut target) { @@ -530,8 +551,16 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { - // Real byte receipt: the viewer must actually RECEIVE PTY output — here the - // ECHOED flood input (`FLOODINPUT-…`) round-tripping back through the PTY. - // Output delivery does not go through the input path, so receiving these - // bytes proves the dispatch serviced this attach while the flood was driven. - // - // Only enter the (Whole-conn, blocking) read loop if the subscribe was - // SERVICED — if it was not (the deadlock face), there is no output coming and - // a blocking read would hang the helper; we report (false,false) and bail. + // Retained-replay contract: the pre-flood ACKDL_OUTPUT record is already in + // the broker ring, and this VIEWER asks from_seq=0. It MUST remain a viewer: + // a same-origin equal-lease CONTROL retake deliberately does not replay + // history or self-displace, making from_seq=0 inert. Resize is only an + // explicit post-subscribe stream wake. + if send_attach_resize(&mut operator, stream_b, 25, 80).is_err() { + let _ = result_tx.send((subscribed, false)); + return; + } + + // The VIEWER must receive and decode the retained ACKDL_OUTPUT seeded before + // the flood. The child remains silent throughout the flood, so output cannot + // confound the deadlock substrate. Only enter the bounded Split-carrier read + // loop if the subscribe was SERVICED — if it was not (the deadlock face), + // report (false,false) immediately. A Whole carrier would refuse this deadline + // by name rather than silently converting it to an unbounded read. @@ -540,0 +570 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { + let mut received = Vec::new(); @@ -542,3 +571,0 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { - // A Whole conn ignores the per-read deadline, but the echo child keeps - // re-emitting the flooded input, so frames keep arriving and the OUTER - // deadline check fires between them; `got_output` then breaks promptly. @@ -546 +573,2 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { - match operator.read_event_until(Some(Instant::now() + Duration::from_millis(250))) { + let slice_deadline = (Instant::now() + Duration::from_millis(250)).min(deadline); + match operator.read_event_until(Some(slice_deadline)) { @@ -552,3 +580,5 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { - let chunk = decode_bytes(&data_b64).unwrap_or_default(); - if count(&chunk, b"FLOODINPUT") >= 1 { - got_output = true; + if let Ok(chunk) = decode_bytes(&data_b64) { + received.extend_from_slice(&chunk); + got_output = received + .windows(RETAINED_OUTPUT.len()) + .any(|window| window == RETAINED_OUTPUT); @@ -563 +593,5 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { - Err(_) => break, + Err(e) if e.kind() == std::io::ErrorKind::TimedOut => continue, + Err(e) => { + eprintln!("CONCURRENT_VIEWER_READ_FAILED: {e}"); + break; + } @@ -644,5 +678,8 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { - // (4) DIAGNOSTIC (not asserted): a concurrent real loopback rc attach + its byte - // receipt. Two simultaneous loopback dials on one NetHost can race the inbound - // stream demux in this in-process rig, so the loopback-attach leg is CAPTURED - // here (never a false-red); the flood-drain + liveness probes above are the gate. - let _ = (subscribed, got_output); + // (4) The post-flood VIEWER is serviced and its bounded receive path decodes + // ACKDL_OUTPUT from the retained pre-flood ring record. The Resize above is + // only a wake; the asserted seed sequence is the stimulus contract. + assert!( + subscribed && got_output, + "the concurrent VIEWER must subscribe and replay retained ACKDL_OUTPUT; \ + subscribed={subscribed} got_output={got_output}" + );