diff --git a/.alfonso/release-notes/next/log-session-prefix.md b/.alfonso/release-notes/next/log-session-prefix.md new file mode 100644 index 000000000..b090c2d7f --- /dev/null +++ b/.alfonso/release-notes/next/log-session-prefix.md @@ -0,0 +1 @@ +Log lines no longer carry the wrong session id: slow-bind warnings name the bind's own session, and lines not tied to a session carry no session tag (#1008, thanks @iceteaSA). diff --git a/benchmarks/aft-search/slice-descriptors/land-log-session-prefix.json b/benchmarks/aft-search/slice-descriptors/land-log-session-prefix.json new file mode 100644 index 000000000..ebd574711 --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/land-log-session-prefix.json @@ -0,0 +1,8 @@ +{ + "targeted_mechanism": "none", + "kind": "harness", + "slice_class": "non_ranking", + "fixtures": [ + "harness-goldens" + ] +} diff --git a/crates/aft/src/log_ctx.rs b/crates/aft/src/log_ctx.rs index fdc33ea8c..5fa68d81f 100644 --- a/crates/aft/src/log_ctx.rs +++ b/crates/aft/src/log_ctx.rs @@ -9,23 +9,20 @@ //! Background threads spawned during request handling (search-index pre-warm, //! semantic-index build) **must** capture the session id before spawning and //! re-install it on the new thread via [`set_session`] or [`with_session`]. +//! +//! A line logged outside any session scope carries no session tag. Threads +//! such as the request loop, the subc frame loop and executor workers serve +//! many sessions in turn, so "the last session this thread served" says +//! nothing about who owns a later between-request line; tagging with it sent +//! operators to the wrong session. Code that logs about a specific session +//! from such a thread (a pending bind, a background task) must enter that +//! session's scope with [`with_session`] first. use std::cell::RefCell; thread_local! { /// Current session id for log tagging. `None` means "no session context". static CURRENT_SESSION: RefCell> = const { RefCell::new(None) }; - - /// Most-recent non-None session id observed on this thread. Used as a - /// fallback for between-request log sites that have no inherent session - /// (filesystem watcher invalidation, gitignore rebuilds triggered by - /// the OS, etc.) so their lines still carry a session tag. - /// - /// This is best-effort: in a multi-session setup the tag may attribute - /// project-wide events to whichever session was active most recently. - /// That's acceptable because the vast majority of AFT users have one - /// active session per project and the alternative is an untagged line. - static LAST_SESSION: RefCell> = const { RefCell::new(None) }; } /// Set the current thread-local session id. @@ -33,11 +30,6 @@ thread_local! { /// Call this at the start of a background thread that captured the session id /// from the parent request loop. pub fn set_session(session: Option) { - if let Some(sid) = session.as_deref() { - LAST_SESSION.with(|s| { - *s.borrow_mut() = Some(sid.to_string()); - }); - } CURRENT_SESSION.with(|s| { *s.borrow_mut() = session; }); @@ -47,10 +39,6 @@ struct SessionGuard(Option); impl Drop for SessionGuard { fn drop(&mut self) { - // Restore CURRENT_SESSION directly (don't go through set_session, - // which would also overwrite LAST_SESSION — the guard's job is to - // restore the previous CURRENT_SESSION value, not to alter the - // last-session fallback). CURRENT_SESSION.with(|s| { *s.borrow_mut() = self.0.take(); }); @@ -75,21 +63,12 @@ pub fn current_session() -> Option { } /// Return the current session id prefix string, e.g. `"[ses_abcd1234] "`, -/// or an empty string if no session id has ever been observed on this thread. -/// -/// Resolution order: -/// 1. Current request's session (set via `with_session`). -/// 2. Most-recent session observed on this thread (fallback for -/// between-request log sites like watcher invalidation that have no -/// inherent session context). +/// or an empty string when no session scope is active on this thread. /// /// The stored session id may already carry the `ses_` prefix (OpenCode's /// real session IDs do); detect that and avoid double-prefixing. pub fn session_prefix() -> String { - let sid_opt = CURRENT_SESSION - .with(|s| s.borrow().clone()) - .or_else(|| LAST_SESSION.with(|s| s.borrow().clone())); - match sid_opt.as_deref() { + match current_session().as_deref() { Some(sid) if sid.starts_with("ses_") => format!("[{}] ", sid), Some(sid) => format!("[ses_{}] ", sid), None => String::new(), @@ -108,22 +87,18 @@ fn write_session_message( message: std::fmt::Arguments<'_>, ) -> std::fmt::Result { CURRENT_SESSION.with(|current| { - LAST_SESSION.with(|last| { - let current = current.borrow(); - let last = last.borrow(); - if let Some(sid) = current.as_deref().or(last.as_deref()) { - writer.write_str(if sid.starts_with("ses_") { - "[" - } else { - "[ses_" - })?; - writer.write_str(sid)?; - writer.write_str("] ")?; - } - Ok(()) - }) + if let Some(sid) = current.borrow().as_deref() { + writer.write_str(if sid.starts_with("ses_") { + "[" + } else { + "[ses_" + })?; + writer.write_str(sid)?; + writer.write_str("] ")?; + } + Ok(()) })?; - // Release the session borrows before user-provided Display implementations + // Release the session borrow before user-provided Display implementations // run; formatting a message may itself change the logging context. writer.write_fmt(message) } @@ -277,17 +252,14 @@ mod tests { } } - /// Reset both thread-locals so each test starts from a clean slate. - /// Rust runs tests on a thread pool — without this helper a previous - /// test in the same thread could leak its LAST_SESSION value into - /// the next one. + /// Reset the thread-local so each test starts from a clean slate. + /// Rust runs tests on a thread pool, and a test that panics inside + /// `set_session` (rather than `with_session`) could otherwise leak its + /// session into the next test on the same thread. fn reset_session_state() { CURRENT_SESSION.with(|s| { *s.borrow_mut() = None; }); - LAST_SESSION.with(|s| { - *s.borrow_mut() = None; - }); } #[test] @@ -328,12 +300,7 @@ mod tests { assert_eq!(session_prefix(), "[ses_abcd1234] "); }); - // After exiting with_session, CURRENT is None but LAST_SESSION - // still carries "abcd1234" — the prefix falls back to it. - assert_eq!(session_prefix(), "[ses_abcd1234] "); - - // Truly empty starting state: no prefix. - reset_session_state(); + // Once the scope has ended there is no session context, so no prefix. assert_eq!(session_prefix(), ""); } @@ -354,46 +321,48 @@ mod tests { CURRENT_SESSION.with(|s| { assert_eq!(s.borrow().as_deref(), Some("direct")); }); + assert_eq!(session_prefix(), "[ses_direct] "); set_session(None); CURRENT_SESSION.with(|s| { assert!(s.borrow().is_none()); }); - // LAST_SESSION still remembers it for the fallback. - LAST_SESSION.with(|s| { - assert_eq!(s.borrow().as_deref(), Some("direct")); - }); + assert_eq!(session_prefix(), ""); } - /// Regression: between-request log sites (filesystem watcher - /// invalidation, gitignore rebuilds) run outside a `with_session` - /// scope but should still carry a session tag derived from the most - /// recent active session on this thread. Without the LAST_SESSION - /// fallback these lines appeared untagged in the plugin log. + /// A line logged after a session scope ends, on a thread that served that + /// session, carries no session tag. Threads like the request loop and the + /// subc frame loop serve many sessions in turn, so tagging an unscoped + /// line with the previous session (as an earlier fallback did) attributed + /// project-wide or other-session events to whichever session ran last. #[test] - fn session_prefix_falls_back_to_last_session_after_with_session_exits() { + fn unscoped_line_after_scope_ends_carries_no_session() { reset_session_state(); - // Simulate a request handled on this thread. with_session(Some("ses_first".to_string()), || { - assert_eq!(session_prefix(), "[ses_first] "); + assert_eq!( + format_session_message(format_args!("inside")), + "[ses_first] inside" + ); }); - // CURRENT_SESSION is now None, but a between-request log site - // should still get a tag from LAST_SESSION. assert!(current_session().is_none()); - assert_eq!(session_prefix(), "[ses_first] "); + assert_eq!( + format_session_message(format_args!("invalidated 1 files")), + "invalidated 1 files" + ); } + /// A thread that served session A and then logs about session B inside + /// B's scope tags the line with B only. #[test] - fn last_session_updates_with_each_with_session_call() { + fn scoped_line_carries_its_own_session_not_the_previous_one() { reset_session_state(); - with_session(Some("ses_first".to_string()), || {}); - assert_eq!(session_prefix(), "[ses_first] "); - - with_session(Some("ses_second".to_string()), || {}); - // Newer session wins as the fallback. - assert_eq!(session_prefix(), "[ses_second] "); + with_session(Some("ses_A".to_string()), || {}); + let line = with_session(Some("ses_B".to_string()), || { + format_session_message(format_args!("pending bind")) + }); + assert_eq!(line, "[ses_B] pending bind"); } #[test] @@ -410,17 +379,19 @@ mod tests { ); } + /// `with_session(None, ...)` (synthetic or system requests with no + /// session id) runs its closure with no session tag, even nested inside + /// a real session's scope, and restores that scope afterwards. #[test] - fn with_session_none_does_not_clear_last_session() { + fn with_session_none_runs_untagged_and_restores_outer_scope() { reset_session_state(); - with_session(Some("ses_real".to_string()), || {}); - assert_eq!(session_prefix(), "[ses_real] "); - - // A subsequent `with_session(None, ...)` call (which can happen - // for synthetic / system requests with no session_id) must not - // erase the fallback tag from earlier real traffic. - with_session(None, || {}); - assert_eq!(session_prefix(), "[ses_real] "); + with_session(Some("ses_real".to_string()), || { + with_session(None, || { + assert_eq!(session_prefix(), ""); + }); + assert_eq!(session_prefix(), "[ses_real] "); + }); + assert_eq!(session_prefix(), ""); } } diff --git a/crates/aft/src/subc/bash.rs b/crates/aft/src/subc/bash.rs index cc905c606..67e3bdebf 100644 --- a/crates/aft/src/subc/bash.rs +++ b/crates/aft/src/subc/bash.rs @@ -785,12 +785,15 @@ fn detach_held_bash_sync(target: drain::BashDetachTarget, cancelled: bool) { // process alive until the daemon's scope.describe response confirms the run ended; the // command's normal hard timeout still applies while confirmation waits. } else if let Err(error) = target.registry.promote(&target.task_id, &target.session_id) { - crate::slog_warn!( - "subc attach: bash background handoff failed task_id={:?} session={:?} error={:?}", - target.task_id, - target.session_id, - error - ); + // Runs on a shared blocking-pool thread; tag with the task's session. + crate::log_ctx::with_session(Some(target.session_id.clone()), || { + crate::slog_warn!( + "subc attach: bash background handoff failed task_id={:?} session={:?} error={:?}", + target.task_id, + target.session_id, + error + ); + }); } else if cancelled { // The caller saw only the cancellation; name the still-running task // on the session's next bash call. @@ -2375,7 +2378,10 @@ pub(super) fn submit_deferred_bash( let reply = await_runon_launch_reply(spawn_ctx.bash_background().clone(), &session_id, &response, capture_deadline, &cancel, &claim, server_completion, Some(capture_reply_deadline), background_runon).await.unwrap_or(response); foreground_runon_reply(reply, foreground_runon_deadline, background_runon, &cancel, foreground_runon_timeout, startup_window, worker_session, format_context.bash_watch_available.unwrap_or(worker_session)) }; - crate::slog_warn!("bash startup reply deadline channel={} corr={corr} elapsed_ms={} admitted={} code={} steps={}", route.channel, received_at.elapsed().as_millis(), admitted.load(Ordering::Relaxed), response.data.get("code").and_then(Value::as_str).unwrap_or("promoted"), startup_steps.format_steps()); + // This task runs on the shared subc runtime outside any session scope. + crate::log_ctx::with_session(Some(session_id.to_string()), || { + crate::slog_warn!("bash startup reply deadline channel={} corr={corr} elapsed_ms={} admitted={} code={} steps={}", route.channel, received_at.elapsed().as_millis(), admitted.load(Ordering::Relaxed), response.data.get("code").and_then(Value::as_str).unwrap_or("promoted"), startup_steps.format_steps()); + }); let result = (!cancel.connection.is_cancelled() && !cancel.route.is_cancelled() && claim.try_claim_for_wait_task()).then(|| bash_result_from_response(response, &format_context)); send_bash_deferred_completion(&completion_tx, &task_metrics, route, corr, flags, ver, root_for_task.clone(), request_id.clone(), result, false).await; answered_startup = true; @@ -3584,9 +3590,12 @@ async fn send_ready_bash_result( .get("message") .and_then(Value::as_str) .unwrap_or(&result.text); - crate::slog_warn!( - "subc attach: late bash result discarded after handoff task_id={task_id:?} session={session_id:?} error={error:?}" - ); + // Runs on the shared subc runtime; tag with the task's session. + crate::log_ctx::with_session(Some(session_id.to_string()), || { + crate::slog_warn!( + "subc attach: late bash result discarded after handoff task_id={task_id:?} session={session_id:?} error={error:?}" + ); + }); } else if claim.claimed_by_module_loop() { send_bash_deferred_completion( completion_tx, diff --git a/crates/aft/src/subc/health.rs b/crates/aft/src/subc/health.rs index 3f49ba4b6..952392193 100644 --- a/crates/aft/src/subc/health.rs +++ b/crates/aft/src/subc/health.rs @@ -1173,14 +1173,18 @@ impl DispatchPathMetrics { continue; } logged.insert(key, now); - crate::slog_warn!( - "subc bg wake: pending watch remains unacked root={} session={} task={} watch={} age_ms={}", - root.as_path().display(), - session, - task_id, - watch_id, - age_ms - ); + // One scan covers every session's watches, so log each inside + // its own session's scope to tag it with that session. + crate::log_ctx::with_session(Some(session.clone()), || { + crate::slog_warn!( + "subc bg wake: pending watch remains unacked root={} session={} task={} watch={} age_ms={}", + root.as_path().display(), + session, + task_id, + watch_id, + age_ms + ); + }); } } logged.retain(|_, last| { @@ -1641,16 +1645,19 @@ pub(super) fn warn_slow_pending_binds( in_flight_writer: None, reader_admissions_while_promoted_writer_waited: 0, }); - crate::slog_warn!( - "{}", - pending_bind_breadcrumb( - *route, - &pending.bind_root_id, - age, - &pending.configure_request_id, - &snapshot, - ) - ); + let breadcrumb = pending_bind_breadcrumb( + *route, + &pending.bind_root_id, + age, + &pending.configure_request_id, + &snapshot, + ); + // The frame loop serves every session, so tag the line with the + // session that requested this bind rather than whatever scope (if + // any) the thread happens to be in. + crate::log_ctx::with_session(pending.session.clone(), || { + crate::slog_warn!("{breadcrumb}"); + }); } } diff --git a/crates/aft/src/subc/mod.rs b/crates/aft/src/subc/mod.rs index c3a33c651..0a70760cd 100644 --- a/crates/aft/src/subc/mod.rs +++ b/crates/aft/src/subc/mod.rs @@ -1448,6 +1448,10 @@ struct PendingBind { /// removed, running configures return at their next checkpoint) instead of /// only marking bookkeeping. cancellation: crate::executor::JobCancellation, + /// Session that requested this bind. Periodic warnings about the bind run + /// on the frame loop, which serves every session, so they log inside this + /// session's scope to carry the right session tag. + session: Option, } struct RouteBindCompletion { @@ -7966,6 +7970,7 @@ async fn handle_control_request( scopes.bind(&route_identity); } let configure_session = route_identity.session.clone(); + let bind_session = configure_session.clone(); let root_was_live = live_roots.contains_key(&bind_root_id); metrics.mark_frame_loop_wait( "executor.scheduler/register_actor", @@ -8036,6 +8041,7 @@ async fn handle_control_request( ver: frame.header.ver, flags: frame.header.flags, cancellation: configure_cancellation, + session: Some(bind_session), }, ); @@ -19107,6 +19113,7 @@ mod tests { ver: PROTOCOL_VERSION, flags: control_flags(), cancellation: crate::executor::JobCancellation::new(), + session: None, }, )]); let mut installed_route_epochs = HashMap::from([(route.channel, route.epoch)]); @@ -19195,6 +19202,7 @@ mod tests { ver: PROTOCOL_VERSION, flags: control_flags(), cancellation: crate::executor::JobCancellation::new(), + session: None, }, )]); let mut installed_route_epochs = HashMap::from([(route.channel, route.epoch)]); @@ -19247,6 +19255,7 @@ mod tests { ver: PROTOCOL_VERSION, flags: control_flags(), cancellation: crate::executor::JobCancellation::new(), + session: None, }; let young = route_key(9, 1); let held = route_key(10, 1); @@ -19288,6 +19297,70 @@ mod tests { assert!(!pending_binds[&young].deadline_reported); } + /// The frame loop serves every session. After it has served session A, + /// the slow-bind warning for a bind that session B requested must carry + /// B's tag, and one for a bind with no session must carry no tag; neither + /// may carry A's. + #[test] + fn slow_pending_bind_warning_carries_the_bind_session_not_the_last_served() { + let (_dir, root) = test_root("route-bind-slow-warn-session"); + let pending_bind = |corr: u64, session: Option<&str>| PendingBind { + bind_root_id: root.clone(), + inserted_new_actor: false, + cancelled: false, + configure_request_id: format!("subc-bind-{corr}"), + started_at: Instant::now() + .checked_sub(DISPATCH_PATH_BIND_WARN_AFTER + Duration::from_secs(1)) + .expect("monotonic clock is older than the bind age"), + warned_half_deadline: false, + deadline_reported: false, + corr, + ver: PROTOCOL_VERSION, + flags: control_flags(), + cancellation: crate::executor::JobCancellation::new(), + session: session.map(str::to_string), + }; + let owned = route_key(352, 1); + let unowned = route_key(353, 1); + let mut pending_binds = HashMap::from([ + (owned, pending_bind(352, Some("ses_B"))), + (unowned, pending_bind(353, None)), + ]); + let executor = Executor::new(); + + let ((), lines) = crate::logging::capture_log_lines(|| { + log_ctx::with_session(Some("ses_A".to_string()), || { + crate::slog_info!("serving session A"); + }); + warn_slow_pending_binds(&mut pending_binds, &executor); + }); + + assert_eq!( + lines.first().map(String::as_str), + Some("INFO [ses_A] serving session A") + ); + let owned_line = lines + .iter() + .find(|line| line.contains("route 352@1")) + .unwrap_or_else(|| panic!("no warning for route 352: {lines:?}")); + let unowned_line = lines + .iter() + .find(|line| line.contains("route 353@1")) + .unwrap_or_else(|| panic!("no warning for route 353: {lines:?}")); + assert!( + owned_line.starts_with("WARN [ses_B] subc attach: "), + "{owned_line}" + ); + assert!( + unowned_line.starts_with("WARN subc attach: "), + "{unowned_line}" + ); + assert!( + !lines[1..].iter().any(|line| line.contains("ses_A")), + "{lines:?}" + ); + } + /// A bind whose deadline passes while it waits behind a reader on its root /// is refused with its own retryable code, and the message names the /// reader's job, its tool and how long it has run, so whoever retries the @@ -19348,6 +19421,7 @@ mod tests { ver: PROTOCOL_VERSION, flags: control_flags(), cancellation: bind_token, + session: None, }, )]); let mut installed_route_epochs = HashMap::from([(route.channel, route.epoch)]); @@ -19540,6 +19614,7 @@ mod tests { ver: PROTOCOL_VERSION, flags: control_flags(), cancellation, + session: None, }, )]); let mut installed = HashMap::from([(route.channel, route.epoch)]); @@ -19782,6 +19857,7 @@ mod tests { ver: PROTOCOL_VERSION, flags: control_flags(), cancellation, + session: None, }, )]); let mut installed = HashMap::from([(route.channel, route.epoch)]); diff --git a/crates/aft/tests/integration/log_ctx_test.rs b/crates/aft/tests/integration/log_ctx_test.rs index aec5c0aa1..b2a5ffc2e 100644 --- a/crates/aft/tests/integration/log_ctx_test.rs +++ b/crates/aft/tests/integration/log_ctx_test.rs @@ -87,13 +87,12 @@ fn slog_macros_prepend_session_tag() { ); }); - // Case 2: Without explicit session id — verify LAST_SESSION fallback. - // session_prefix() falls back to the most recently set session id so that - // between-request log lines (filesystem watcher invalidation, gitignore - // matcher rebuilds) still carry a session tag. This thread had Case 1 - // set "abcd1234", so the fallback should return that. + // Case 2: Without a session scope there is no tag, even though this + // thread served "abcd1234" in Case 1: an unscoped line is not + // attributed to whichever session the thread served last. aft::log_ctx::with_session(None, || { let prefix = aft::log_ctx::session_prefix(); - assert_eq!(prefix, "[ses_abcd1234] "); + assert_eq!(prefix, ""); }); + assert_eq!(aft::log_ctx::session_prefix(), ""); }