diff --git a/crates/spt-daemon/tests/input_ack_deadlock.rs b/crates/spt-daemon/tests/input_ack_deadlock.rs index 92b06fbc..eb706292 100644 --- a/crates/spt-daemon/tests/input_ack_deadlock.rs +++ b/crates/spt-daemon/tests/input_ack_deadlock.rs @@ -98,7 +98,7 @@ use spt_test_support::TestHome; const FLOOD_N: u64 = 64; static SEQ: AtomicU32 = AtomicU32::new(0); -const RETAINED_OUTPUT: &[u8] = b"ACKDL_OUTPUT"; +const POST_FLOOD_OUTPUT: &[u8] = b"ACKDL_LIVE_OUTPUT"; fn unique_name() -> String { let n = SEQ.fetch_add(1, Ordering::Relaxed); format!("spt-daemon-ackdl-{}-{}.sock", std::process::id(), n) @@ -115,27 +115,31 @@ fn kill_pid(pid: u32) { .output(); } -/// 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 { +/// A SEEDED-THEN-QUIET child: it emits one pre-flood marker, ignores stdin, and +/// emits a second marker only after the test creates `live_trigger`. The file is +/// created after the viewer has subscribed, so the second marker can only arrive +/// as live fan-out, never in the cold-attach repaint. +fn seeded_quiet_spawn_req(endpoint: &str, live_trigger: &std::path::Path) -> SpawnReq { + let trigger = live_trigger.display(); #[cfg(unix)] let (program, args) = ( "sh".to_string(), vec![ "-c".to_string(), - "printf 'ACKDL_OUTPUT\\n'; exec sleep 600".to_string(), + format!( + "printf 'ACKDL_OUTPUT\\n'; while [ ! -f '{trigger}' ]; do sleep 0.02; done; printf 'ACKDL_LIVE_OUTPUT\\n'; exec sleep 600" + ), ], ); #[cfg(windows)] let (program, args) = ( - "cmd".to_string(), + "powershell".to_string(), vec![ - "/C".to_string(), - "echo ACKDL_OUTPUT & ping -n 600 127.0.0.1 >nul".to_string(), + "-NoProfile".to_string(), + "-Command".to_string(), + format!( + "$Host.UI.RawUI.WindowTitle='ackdl'; Write-Output 'ACKDL_OUTPUT'; while (-not (Test-Path -LiteralPath '{trigger}')) {{ Start-Sleep -Milliseconds 20 }}; Write-Output 'ACKDL_LIVE_OUTPUT'; Start-Sleep -Seconds 600" + ), ], ); SpawnReq { @@ -279,6 +283,7 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { let dir = tempfile::tempdir().expect("tempdir"); let name = unique_name(); let endpoint = "ack-deadlock-ep"; + let live_trigger = dir.path().join("emit-live-output"); let broker = net_broker(&name, &dir.path().join("ackdl")); @@ -288,7 +293,7 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { // reads nor writes, preserving the clean applied-ack deadlock substrate. let mut spawner = connect_retry(&name); let sid = spawner - .spawn_session(seeded_quiet_spawn_req(endpoint)) + .spawn_session(seeded_quiet_spawn_req(endpoint, &live_trigger)) .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 { @@ -484,7 +489,8 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { // subscribes, and replays the retained pre-flood output. Pre-fix the per-conn // handler is deadlocked on the flood's ack stream and this subscribe is never // serviced. Run on its own thread; the outer recv_timeout is the hard ceiling. ── - let (result_tx, result_rx) = std::sync::mpsc::channel::<(bool, bool)>(); + let (result_tx, result_rx) = std::sync::mpsc::channel::<(bool, bool, bool)>(); + let (viewer_ready_tx, viewer_ready_rx) = std::sync::mpsc::channel::(); let attach_name = name.clone(); let attach_ep = endpoint.to_string(); let attacher = thread::spawn(move || { @@ -499,14 +505,14 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { ) { Ok(b) => b, Err(_) => { - let _ = result_tx.send((false, false)); + let _ = result_tx.send((false, false, false)); return; } }; let dialed = match operator.net_dial_loopback() { Ok(d) => d, Err(_) => { - let _ = result_tx.send((false, false)); + let _ = result_tx.send((false, false, false)); return; } }; @@ -521,7 +527,7 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { ) { Ok(s) => s, Err(_) => { - let _ = result_tx.send((false, false)); + let _ = result_tx.send((false, false, false)); return; } }; @@ -533,7 +539,7 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { let (stream_a, origin) = match wait_for_latest_stream(&mut target) { Some(s) => s, None => { - let _ = result_tx.send((subscribed, false)); + let _ = result_tx.send((subscribed, false, false)); return; } }; @@ -558,17 +564,18 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { // 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)); + let _ = viewer_ready_tx.send(false); + let _ = result_tx.send((subscribed, false, 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. - let mut got_output = false; + let _ = viewer_ready_tx.send(subscribed); + + // A cold attach is entitled to a repaint, not retained ring bytes. First + // prove that repaint is non-empty and decodable; then keep reading until + // the marker emitted only after the main thread's post-subscribe trigger + // arrives as live fan-out. + let mut got_decodable_output = false; + let mut got_live_output = false; if subscribed { let mut decoder = AttachDecoder::new(); let mut received = Vec::new(); @@ -582,14 +589,15 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { for rec in decoder.push(&bytes) { if let AttachRecord::Output { data_b64, .. } = rec { if let Ok(chunk) = decode_bytes(&data_b64) { + got_decodable_output |= !chunk.is_empty(); received.extend_from_slice(&chunk); - got_output = received - .windows(RETAINED_OUTPUT.len()) - .any(|window| window == RETAINED_OUTPUT); + got_live_output = received + .windows(POST_FLOOD_OUTPUT.len()) + .any(|window| window == POST_FLOOD_OUTPUT); } } } - if got_output { + if got_live_output { break; } } @@ -601,19 +609,30 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { } } } + eprintln!( + "[DEBUG-ackdl-measure] received={:?}", + String::from_utf8_lossy(&received) + ); } // Report the verdict, then ABANDON the conn + serve thread WITHOUT a blocking // drain loop (the main thread's child-kill + broker drop unwinds them). - let _ = result_tx.send((subscribed, got_output)); + let _ = result_tx.send((subscribed, got_decodable_output, got_live_output)); let _ = operator.net_stream_send(stream_b, &[], None, true); drop(operator); drop(server); }); - let (subscribed, got_output) = result_rx + let viewer_ready = viewer_ready_rx + .recv_timeout(Duration::from_secs(20)) + .unwrap_or(false); + if viewer_ready { + std::fs::write(&live_trigger, b"go").expect("release post-flood live output"); + } + + let (subscribed, got_decodable_output, got_live_output) = result_rx .recv_timeout(Duration::from_secs(30)) - .unwrap_or((false, false)); + .unwrap_or((false, false, false)); // ── Teardown: both verdicts are already captured off the watchdog'd channels, // so the GATE never depends on a helper thread terminating. Kill the echo @@ -634,11 +653,11 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { eprintln!( "=== P1b ACK-DEADLOCK GATE: flood_n={FLOOD_N} flood_verdict={flood_detail} \ broker_alive(net_status)={broker_alive} sessions_answered={sessions_answered} \ - concurrent_attach_subscribed={subscribed} attach_received_output={got_output} \ - (fixed = flood_verdict=Sent + broker_alive + sessions_answered true; pre-fix \ - on a small-pipe/forkpty platform: the applied-ack stream fills the return \ - direction → the per-conn handler deadlocks → flood watchdog times out. The \ - loopback leg [subscribed/got_output] is a captured diagnostic, not asserted.) ===" + concurrent_attach_subscribed={subscribed} repaint_decodable={got_decodable_output} \ + post_flood_live_output={got_live_output} (fixed = flood_verdict=Sent + broker_alive + \ + sessions_answered + subscribed + repaint_decodable + post_flood_live_output true; \ + pre-fix on a small-pipe/forkpty platform: the applied-ack stream fills the return \ + direction → the per-conn handler deadlocks → flood watchdog times out.) ===" ); // (1) THE CRUX (the non-vacuous deadlock proof): all FLOOD_N records SEND through @@ -679,14 +698,15 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { the broker's session table / dispatch must not be globally frozen \ (REQ-HAZARD-INPUT-ACK-BACKPRESSURE)." ); - // (4) The post-flood VIEWER is serviced and its bounded receive path decodes - // ACKDL_OUTPUT from the retained pre-flood ring record. The old two-dial - // demux-race hedge manifested as stale first-row stream selection; - // wait_for_latest_stream selects the later VIEWER, making this assert sound. - // Resize above is only a wake; the asserted seed sequence is the stimulus. + // (4) The post-flood VIEWER is serviced, receives a non-empty decodable cold + // repaint, then receives ACKDL_LIVE_OUTPUT as live fan-out after its + // subscription. The repaint proves the served viewer path is alive without + // assuming that a cold attach replays retained ring bytes; the second marker + // proves output still flows after the flood. assert!( - subscribed && got_output, - "the concurrent VIEWER must subscribe and replay retained ACKDL_OUTPUT; \ - subscribed={subscribed} got_output={got_output}" + subscribed && got_decodable_output && got_live_output, + "the concurrent VIEWER must subscribe, decode its repaint, and receive \ + post-flood live output; subscribed={subscribed} \ + got_decodable_output={got_decodable_output} got_live_output={got_live_output}" ); }