diff --git a/crates/spt-daemon/src/registryhost.rs b/crates/spt-daemon/src/registryhost.rs index 6422f324..3d79cbbe 100644 --- a/crates/spt-daemon/src/registryhost.rs +++ b/crates/spt-daemon/src/registryhost.rs @@ -1366,6 +1366,19 @@ mod tests { for p in ["alpha", "beta", "gamma"] { std::fs::create_dir_all(work.join(p)).unwrap(); } + // PREMISE, asserted rather than assumed: this fixture tree is OUTSIDE any git + // toplevel. project_id_for_dir tries `git remote get-url origin` FIRST, so a + // tempdir that lands inside a checkout derives THAT repo's remote for every + // fixture dir and collapses alpha/beta/gamma to one id — the assertions below + // then red as a product defect when the real fault is where TMP points + // (releases#304 / register IR-107). A failure HERE names the environment. + assert_eq!( + spt_store::project::project_id_for_dir(&work.join("alpha")), + "alpha", + "fixture dir is inside a git repo — TMP resolves into a checkout, so the id \ + came from that repo's remote instead of the folder name; this is an \ + environment fault, not a recent_projects_for defect" + ); // A SECOND spelling of project "beta" (distinct cwd, same derived id). let work2 = root.join("work2"); std::fs::create_dir_all(work2.join("beta")).unwrap(); diff --git a/crates/spt-daemon/src/wan.rs b/crates/spt-daemon/src/wan.rs index 785679c8..48b3cae3 100644 --- a/crates/spt-daemon/src/wan.rs +++ b/crates/spt-daemon/src/wan.rs @@ -878,32 +878,56 @@ pub enum PresenceRequestOutcome { /// its own access seam, and a peer that died mid-dial. Reporting any of them as /// "offline" would sell one guess as a fact about somebody's agent. // [impl->REQ-UNLISTED-PRESENCE-PROBE] +// [impl->REQ-ROSTER-WAIT-ATTRIBUTION] pub fn request_presence( brain: &mut Brain, conn_id: u64, rec: &spt_net::net::presencemsg::PresenceRecord, ) -> std::io::Result { use spt_net::net::presencemsg::{Presence, PresenceReplyDecoder}; + use spt_store::registry::RosterWait; // [impl->REQ-WAN-REPLY-BOUND] - crate::brain::refuse_unbounded_carrier(brain, PRESENCE_VERB)?; - let opened = brain.net_open_stream(conn_id, None)?; - brain.net_stream_send(opened.stream_id, &ndjson::encode_line(rec), None, true)?; - brain.net_stream_subscribe(opened.stream_id, 0)?; + let admission = RosterWait::start("presence.carrier_admission", "none"); + let bounded = crate::brain::refuse_unbounded_carrier(brain, PRESENCE_VERB); + admission.finish(format_args!("outcome={bounded:?}")); + bounded?; + let open = RosterWait::start("presence.open_stream", "carrier_per_operation_wall_reply"); + let opened = brain.net_open_stream(conn_id, None); + open.finish(format_args!("outcome={:?}", opened.as_ref().map(|_| ()))); + let opened = opened?; + let send = RosterWait::start("presence.send_finish", "carrier_per_operation_wall_reply"); + let sent = brain.net_stream_send(opened.stream_id, &ndjson::encode_line(rec), None, true); + send.finish(format_args!("outcome={sent:?}")); + sent?; + let subscribe = RosterWait::start("presence.subscribe", "carrier_per_operation_wall_reply"); + let subscribed = brain.net_stream_subscribe(opened.stream_id, 0); + subscribe.finish(format_args!("outcome={subscribed:?}")); + subscribed?; let mut decoder = PresenceReplyDecoder::new(); // BOUNDED reply read, the `request_wan` discipline (releases#289): see the // sibling verbs above for why the budget re-arms per frame and why this // reads through `read_event_until` rather than `read_peer_reply_until`. // [impl->REQ-WAN-REPLY-BOUND] let mut deadline = brain.reply_read_deadline(); + let reply_wait = RosterWait::start("presence.reply", "wall_rearmed_on_matching_data"); + let mut events = 0usize; + let mut data_frames = 0usize; + let mut reply_bytes = 0usize; + reply_wait.progress(format_args!("budget_remaining_ms={:?} producer=subscribed stream={}", deadline.map(|d| d.saturating_duration_since(std::time::Instant::now()).as_millis()), opened.stream_id)); loop { let ev = match brain.read_event_until(deadline) { Ok(ev) => ev, Err(e) if e.kind() == std::io::ErrorKind::TimedOut => { - return Ok(PresenceRequestOutcome::PeerSilent) + reply_wait.finish(format_args!("outcome=peer_silent events={events} data_frames={data_frames} reply_bytes={reply_bytes} producer=no_decoded_reply")); + return Ok(PresenceRequestOutcome::PeerSilent); + } + Err(e) => { + reply_wait.finish(format_args!("outcome=read_error events={events} data_frames={data_frames} reply_bytes={reply_bytes} error={e:?}")); + return Err(e); } - Err(e) => return Err(e), }; + events += 1; if matches!(&ev, BrokerEvent::NetStreamData { stream_id: sid, .. } if *sid == opened.stream_id) { deadline = brain.reply_read_deadline(); @@ -914,16 +938,24 @@ pub fn request_presence( bytes, .. } if sid == opened.stream_id => { + data_frames += 1; + reply_bytes += bytes.len(); + reply_wait.progress(format_args!("events={events} data_frames={data_frames} reply_bytes={reply_bytes} producer=matching_data budget_rearmed=true")); if let Some(reply) = decoder.push(&bytes).into_iter().next() { + reply_wait.finish(format_args!("outcome=answered events={events} data_frames={data_frames} reply_bytes={reply_bytes} producer=decoded_reply")); return Ok(PresenceRequestOutcome::Answered(Presence::from_token( &reply.presence, ))); } } BrokerEvent::NetStreamEof { stream_id: sid, .. } if sid == opened.stream_id => { + reply_wait.finish(format_args!("outcome=eof_without_reply events={events} data_frames={data_frames} reply_bytes={reply_bytes} producer=stream_eof")); return Ok(PresenceRequestOutcome::Answered(Presence::Unknown)); } - BrokerEvent::Error { message } => return Err(std::io::Error::other(message)), + BrokerEvent::Error { message } => { + reply_wait.finish(format_args!("outcome=broker_error events={events} data_frames={data_frames} reply_bytes={reply_bytes} error={message:?}")); + return Err(std::io::Error::other(message)); + } _ => {} } } diff --git a/crates/spt-store/src/registry.rs b/crates/spt-store/src/registry.rs index a83d06e7..80c31d63 100644 --- a/crates/spt-store/src/registry.rs +++ b/crates/spt-store/src/registry.rs @@ -17,6 +17,90 @@ use std::path::Path; use crate::liveness; +thread_local! { + static ROSTER_TRACE: std::cell::Cell = const { std::cell::Cell::new(false) }; +} + +/// Endpoint-list's opt-in scope, explicitly propagated to its probe workers. +/// Other callers of the shared SQLite routines remain silent, even when the +/// process has SPT_PUMP_TRACE set. This is not a process-global DB trace switch. +// [impl->REQ-ROSTER-WAIT-ATTRIBUTION] +pub struct RosterTraceScope { + previous: bool, + _thread: std::marker::PhantomData<*mut ()>, +} + +impl RosterTraceScope { + pub fn enter(enabled: bool) -> Self { + Self { + previous: ROSTER_TRACE.with(|v| v.replace(enabled)), + _thread: std::marker::PhantomData, + } + } + + pub fn enabled() -> bool { + ROSTER_TRACE.with(std::cell::Cell::get) + } +} + +impl Drop for RosterTraceScope { + fn drop(&mut self) { + ROSTER_TRACE.with(|v| v.set(self.previous)); + } +} + +/// One investigated roster wait. Start records survive a still-blocked producer; +/// finish records retain successful and failed elapsed time. Thread + pid join +/// nested stages to the worker's endpoint/node record, without changing the wire. +/// Budgets describe existing waits, never impose a new deadline. +// [impl->REQ-ROSTER-WAIT-ATTRIBUTION] +pub struct RosterWait { + started: Option, + stage: &'static str, + budget: &'static str, + finished: std::cell::Cell, +} + +impl RosterWait { + pub fn start(stage: &'static str, budget: &'static str) -> Self { + let wait = Self { + started: RosterTraceScope::enabled().then(std::time::Instant::now), + stage, + budget, + finished: std::cell::Cell::new(false), + }; + wait.progress(format_args!("outcome=started producer=entered")); + wait + } + + pub fn progress(&self, detail: std::fmt::Arguments<'_>) { + if let Some(started) = self.started { + spt_proto::emit_line_err!( + "ROSTER_WAIT pid={} thread={:?} stage={} elapsed_ms={} budget={} {}", + std::process::id(), + std::thread::current().id(), + self.stage, + started.elapsed().as_millis(), + self.budget, + detail + ); + } + } + + pub fn finish(&self, detail: std::fmt::Arguments<'_>) { + self.finished.set(true); + self.progress(detail); + } +} + +impl Drop for RosterWait { + fn drop(&mut self) { + if !self.finished.get() { + self.progress(format_args!("outcome=unwound producer=completion_not_observed")); + } + } +} + /// Open (creating if needed) the registry DB at `/.registry`. fn open_registry(owlery: &Path) -> rusqlite::Result { // [impl->REQ-HAZARD-REGISTRY-DIR-CREATE] SQLite creates the FILE but never @@ -26,19 +110,25 @@ fn open_registry(owlery: &Path) -> rusqlite::Result { // if creation truly fails, Connection::open surfaces the real error (4.9). let _ = std::fs::create_dir_all(owlery); let db_path = owlery.join(".registry"); - let conn = Connection::open(&db_path)?; + let wait = RosterWait::start("sqlite.open", "none"); + let opened = Connection::open(&db_path); + wait.finish(format_args!("outcome={:?}", opened.as_ref().map(|_| ()))); + let conn = opened?; // [impl->REQ-HAZARD-REGISTRY-CONCURRENT] concurrency-safe pragmas (busy_timeout // first, then WAL-with-retry) so concurrent ReadyAgent registrations don't // fail with "database is locked". crate::db::tune_connection(&conn)?; - conn.execute_batch( + let wait = RosterWait::start("sqlite.schema", "sqlite_busy_handler_per_statement"); + let schema = conn.execute_batch( "CREATE TABLE IF NOT EXISTS agents ( id TEXT PRIMARY KEY, address TEXT NOT NULL, pid INTEGER NOT NULL, started TEXT NOT NULL )", - )?; + ); + wait.finish(format_args!("outcome={schema:?}")); + schema?; Ok(conn) } @@ -81,13 +171,24 @@ pub fn register_address_with_pid( /// Look up an agent's TCP address by id (no staleness check). `None` on miss or /// any DB error — never hard-fails. pub fn lookup_address(id: &str, owlery: &Path) -> Option { - let conn = open_registry(owlery).ok()?; - conn.query_row( + let wait = RosterWait::start("sqlite.lookup_address", "sqlite_busy_handler_per_statement"); + wait.progress(format_args!("endpoint={id:?} producer=opening_registry")); + let conn = match open_registry(owlery) { + Ok(conn) => conn, + Err(error) => { + wait.finish(format_args!("outcome=open_error error={error:?}")); + return None; + } + }; + let query = RosterWait::start("sqlite.select_address", "sqlite_busy_handler_per_statement"); + let result = conn.query_row( "SELECT address FROM agents WHERE id = ?1", params![id], |row| row.get(0), - ) - .ok() + ); + query.finish(format_args!("outcome={:?}", result.as_ref().map(|_: &String| ()))); + wait.finish(format_args!("outcome={:?} producer=query_returned", result.as_ref().map(|_| ()))); + result.ok() } /// Look up an agent's owning pid by id. `None` on miss / DB error.