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 warning: in the working copy of 'crates/spt-daemon/tests/pump.rs', LF will be replaced by CRLF the next time Git touches it 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 warning: in the working copy of 'crates/spt-daemon/tests/pump.rs', LF will be replaced by CRLF the next time Git touches it crates/spt-daemon/tests/attach.rs | 6 +- crates/spt-daemon/tests/brain_read_deadline.rs | 8 +- crates/spt-daemon/tests/digest_cross_node.rs | 6 +- crates/spt-daemon/tests/inject_control_wedge.rs | 33 +++-- crates/spt-daemon/tests/input_ack_deadlock.rs | 185 ++++++++++++++---------- crates/spt-daemon/tests/pump.rs | 37 +++-- 6 files changed, 163 insertions(+), 112 deletions(-) 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 warning: in the working copy of 'crates/spt-daemon/tests/pump.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 @@ -66,8 +66,8 @@ use std::sync::Arc; use std::thread; 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}; use spt_daemon::codec::{read_frame, write_frame}; use spt_daemon::effect::{MintedOp, Minter}; use spt_daemon::frame::{Envelope, Role}; @@ -98,6 +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"; fn unique_name() -> String { let n = SEQ.fetch_add(1, Ordering::Relaxed); format!("spt-daemon-ackdl-{}-{}.sock", std::process::id(), n) @@ -114,24 +115,27 @@ fn kill_pid(pid: u32) { .output(); } -/// 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 { #[cfg(unix)] - 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(), + ], + ); #[cfg(windows)] - // `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. let (program, args) = ( - "waitfor".to_string(), + "cmd".to_string(), vec![ - "/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(), ], ); SpawnReq { @@ -148,22 +152,6 @@ 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 -} - // One scoped temp SPT_HOME (serve_attach best-effort stamps the endpoint perch's // driven_by marker — that resolution must never touch a real home). fn init_home() -> TestHome { @@ -214,6 +202,25 @@ fn wait_for_stream(brain: &mut Brain) -> Option<(u64, String)> { None } +/// 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 +} + /// Open a FRESH brain connection, send one command, and return whether the /// expected reply arrives within `deadline`. Runs the whole exchange on a spawned /// thread so a BLOCKED `read_frame` (the wedge) surfaces as a timeout — a clean @@ -275,20 +282,29 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { let broker = net_broker(&name, &dir.path().join("ackdl")); - // ── 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. let mut spawner = connect_retry(&name); let sid = spawner - .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" + ); // ── (A) The FLOOD controller: dial loopback, request a CONTROLLER attach, then // fire FLOOD_N input records back-to-back on the ONE stream WITHOUT reading. // The target's serve_attach reads them in one decoder.push() batch and calls - // 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. // // This whole leg runs on its own thread behind a result channel so a wedge // (the pre-fix deadlock) surfaces as a watchdog timeout, never a hang. ── @@ -396,9 +412,7 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { // return direction and wedge the broker's per-conn handler. for op in 1..=FLOOD_N { let line = format!("FLOODINPUT-{op}\r"); - 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) { let _ = flood_tx.send(FloodVerdict::SendFailed { op, error: format!("{error:#}"), @@ -410,10 +424,11 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { // Detach: close the send side (best-effort — the broker's draining handler // processes it and releases the controller seat). Then ABANDON the conn + - // 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). let _ = operator.net_stream_send(stream, &[], None, true); drop(operator); drop(server); @@ -425,9 +440,7 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { let flood_verdict = match flood_rx.recv_timeout(Duration::from_secs(20)) { Ok(verdict) => verdict, Err(std::sync::mpsc::RecvTimeoutError::Timeout) => FloodVerdict::WatchdogTimeout, - Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { - FloodVerdict::HelperDisconnected - } + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => FloodVerdict::HelperDisconnected, }; let (flood_sent, flood_detail) = match &flood_verdict { FloodVerdict::Sent => (true, "Sent".to_string()), @@ -464,14 +477,22 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { // ── (B) The CONCURRENT viewer attach — the non-vacuous proof the broker is NOT // wedged. A SEPARATE operator dials loopback, requests a VIEWER attach, - // 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 // 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 attach_name = name.clone(); let attach_ep = endpoint.to_string(); let attacher = thread::spawn(move || { - 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, + ) { Ok(b) => b, Err(_) => { let _ = result_tx.send((false, false)); @@ -505,7 +526,7 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { let subscribed = operator.net_stream_subscribe(stream_b, 0).is_ok(); let mut target = connect_retry(&attach_name); - let (stream_a, origin) = match wait_for_stream(&mut target) { + let (stream_a, origin) = match wait_for_latest_stream(&mut target) { Some(s) => s, None => { let _ = result_tx.send((subscribed, false)); @@ -527,31 +548,40 @@ 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. let mut got_output = false; if subscribed { let mut decoder = AttachDecoder::new(); + let mut received = Vec::new(); let deadline = Instant::now() + Duration::from_secs(8); - // 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. while Instant::now() < deadline { - 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)) { Ok(BrokerEvent::NetStreamData { stream_id, bytes, .. }) if stream_id == stream_b => { for rec in decoder.push(&bytes) { if let AttachRecord::Output { data_b64, .. } = rec { - 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); } } } @@ -560,7 +590,11 @@ fn input_flood_through_serve_attach_does_not_deadlock_broker() { } } Ok(_) => continue, - Err(_) => break, + Err(e) if e.kind() == std::io::ErrorKind::TimedOut => continue, + Err(e) => { + eprintln!("CONCURRENT_VIEWER_READ_FAILED: {e}"); + break; + } } } } @@ -641,9 +675,12 @@ 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) 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}" + ); } diff --git a/crates/spt-daemon/tests/pump.rs b/crates/spt-daemon/tests/pump.rs index b352b401..b6a7a374 100644 --- a/crates/spt-daemon/tests/pump.rs +++ b/crates/spt-daemon/tests/pump.rs @@ -169,7 +169,9 @@ fn pump_and_dispatch_self_drive_the_subnet() { // ── B (canonical home): gates + the mind A will bootstrap-pull. ────── let mut subnets = SubnetStore::load(); - subnets.create_subnet(subnet, spt_store::access::Mode::Open).expect("create subnet"); + subnets + .create_subnet(subnet, spt_store::access::Mode::Open) + .expect("create subnet"); subnets.save().expect("save subnets"); // Mesh-D6: membership is the roster — B rosters A (both the inbound gate and // the pump's push target read it). Seed the canonical roster. @@ -275,7 +277,9 @@ fn pump_and_dispatch_self_drive_the_subnet() { info::write_info(&perch_path, &rec).unwrap(); let mut a_subnets = SubnetStore::default(); - a_subnets.create_subnet(subnet, spt_store::access::Mode::Open).expect("A subnet"); + a_subnets + .create_subnet(subnet, spt_store::access::Mode::Open) + .expect("A subnet"); a_subnets.save_to(&a_paths.subnets).expect("save A subnets"); // Mesh-D6: A's membership IS the roster — both its inbound gates and its // pump push target read it. The pump derives the roster path as a sibling of @@ -527,7 +531,9 @@ fn pump_survives_a_black_holing_peer_heartbeat_advances_no_restart() { let a_root = dir.path().join("a"); std::fs::create_dir_all(&a_root).unwrap(); let mut a_subnets = SubnetStore::default(); - a_subnets.create_subnet(subnet, spt_store::access::Mode::Open).expect("A subnet"); + a_subnets + .create_subnet(subnet, spt_store::access::Mode::Open) + .expect("A subnet"); a_subnets .save_to(&a_root.join("subnet.json")) .expect("save subnets"); @@ -719,7 +725,9 @@ fn pump_w2_live_peer_advertised_amid_dead_peers_no_restart() { let a_root = dir.path().join("a"); std::fs::create_dir_all(&a_root).unwrap(); let mut a_subnets = SubnetStore::default(); - a_subnets.create_subnet(subnet, spt_store::access::Mode::Open).expect("subnet"); + a_subnets + .create_subnet(subnet, spt_store::access::Mode::Open) + .expect("subnet"); a_subnets .save_to(&a_root.join("subnet.json")) .expect("save subnets"); @@ -809,10 +817,10 @@ fn pump_w2_live_peer_advertised_amid_dead_peers_no_restart() { converge( "this node advertised to the live peer in the same round", || { - live_probe - .net_streams() - .map(|r| r.streams.iter().any(|s| s.remote_id_hex == a_hex)) - .unwrap_or(false) + live_probe + .net_streams() + .map(|r| r.streams.iter().any(|s| s.remote_id_hex == a_hex)) + .unwrap_or(false) }, ); @@ -904,17 +912,22 @@ fn pump_w2_presence_reaches_only_the_subscribing_carrier() { serde_json::from_value(a_addr).expect("addr decodes"); peer.dial(a_ep).expect("peer dials A"); - // The SUBSCRIBER receives the CONNECTED event. + // The SUBSCRIBER receives the CONNECTED event. This is a real 10s allowance: + // 200ms read slices keep the carrier responsive, and a quiet slice continues + // until the absolute deadline rather than collapsing the probe to its first tick. let mut got_connected = false; - for _ in 0..50 { - match sub.read_event_until(Some(Instant::now() + Duration::from_millis(200))) { + let connected_deadline = Instant::now() + Duration::from_secs(10); + while Instant::now() < connected_deadline { + let slice_deadline = (Instant::now() + Duration::from_millis(200)).min(connected_deadline); + match sub.read_event_until(Some(slice_deadline)) { Ok(BrokerEvent::NetPresence(ev)) => { got_connected = true; assert_eq!(ev.kind, spt_daemon::msg::PRESENCE_CONNECTED); break; } Ok(_) => continue, - Err(_) => break, + Err(e) if e.kind() == std::io::ErrorKind::TimedOut => continue, + Err(e) => panic!("subscriber presence read failed before its deadline: {e}"), } } assert!(