From f9b1811d835b7f96189b80d60992bc772fd8b32a Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 16 Oct 2026 03:10:47 +0000 Subject: [PATCH 1/7] build(asupersync): give blocking-pool threads the runtime stack size The native blocking pool built its threads with std::thread::Builder defaults, 2 MiB, while native workers run on the CLI's 16 MiB ASYNC_STACK_BYTES. Handler futures moved off a worker through off_native_thread ran on that smaller stack. A second asupersync patch passes the runtime's thread_stack_size to BlockingPoolOptions and applies it to every pool thread. --- MODULE.bazel | 8 ++- .../asupersync-blocking-pool-stack.patch | 66 +++++++++++++++++++ 2 files changed, 73 insertions(+), 1 deletion(-) create mode 100644 scripts/bazel/patches/asupersync-blocking-pool-stack.patch diff --git a/MODULE.bazel b/MODULE.bazel index ddd4068e74..b45e6c09e4 100644 --- a/MODULE.bazel +++ b/MODULE.bazel @@ -62,10 +62,16 @@ crate.annotation( # task promotes no parked follower, so one long poll delays every socket wake # and shared timer on the runtime. Delete this annotation and the patch once # the pinned or published asupersync contains the fix. +# The blocking-pool stack patch gives pool threads the runtime's +# thread_stack_size (16 MiB in the CLI); unpatched they get the standard +# library's 2 MiB while running handler futures moved off a worker. crate.annotation( crate = "asupersync", patch_args = ["-p1"], - patches = ["//scripts:bazel/patches/asupersync-reactor-handoff.patch"], + patches = [ + "//scripts:bazel/patches/asupersync-reactor-handoff.patch", + "//scripts:bazel/patches/asupersync-blocking-pool-stack.patch", + ], version = "0.6.0", ) diff --git a/scripts/bazel/patches/asupersync-blocking-pool-stack.patch b/scripts/bazel/patches/asupersync-blocking-pool-stack.patch new file mode 100644 index 0000000000..c601b520c2 --- /dev/null +++ b/scripts/bazel/patches/asupersync-blocking-pool-stack.patch @@ -0,0 +1,66 @@ +Blocking-pool threads take the runtime's thread_stack_size instead of the +standard library default (2 MiB), so work moved off a worker keeps the stack +the worker had. + +diff --git a/src/runtime/blocking_pool.rs b/src/runtime/blocking_pool.rs +index b6bac66..a13ad99 100644 +--- a/src/runtime/blocking_pool.rs ++++ b/src/runtime/blocking_pool.rs +@@ -346,6 +346,8 @@ struct BlockingPoolInner { + sleep_fn: SleepFn, + /// Thread name prefix. + thread_name_prefix: String, ++ /// Stack size for each pool thread; zero keeps the standard library's. ++ thread_stack_size: usize, + /// Callback when a thread starts. + on_thread_start: Option>, + /// Callback when a thread stops. +@@ -567,6 +569,7 @@ impl BlockingPool { + time_getter: options.time_getter, + sleep_fn: options.sleep_fn, + thread_name_prefix: options.thread_name_prefix, ++ thread_stack_size: options.thread_stack_size, + on_thread_start: options.on_thread_start, + on_thread_stop: options.on_thread_stop, + thread_handles: Mutex::new(Vec::with_capacity(max_threads)), +@@ -1189,6 +1192,8 @@ pub struct BlockingPoolOptions { + pub sleep_fn: SleepFn, + /// Thread name prefix. + pub thread_name_prefix: String, ++ /// Stack size for each pool thread; zero keeps the standard library's. ++ pub thread_stack_size: usize, + /// Callback when a thread starts. + pub on_thread_start: Option>, + /// Callback when a thread stops. +@@ -1206,6 +1211,7 @@ impl Default for BlockingPoolOptions { + time_getter: wall_clock_now, + sleep_fn: blocking_thread_sleep, + thread_name_prefix: "asupersync".to_string(), ++ thread_stack_size: 0, + on_thread_start: None, + on_thread_stop: None, + affinity_profile: BlockingPoolAffinityProfile::Disabled, +@@ -1326,7 +1332,10 @@ fn spawn_thread_on_inner(inner: &Arc) { + // panic and strand the pool in an inconsistent state. + let thread_id = inner.next_thread_id.fetch_add(1, Ordering::Relaxed); + let name = format!("{}-blocking-{}", inner.thread_name_prefix, thread_id); +- let builder = thread::Builder::new().name(name); ++ let mut builder = thread::Builder::new().name(name); ++ if inner.thread_stack_size > 0 { ++ builder = builder.stack_size(inner.thread_stack_size); ++ } + let assigned_cohort = inner + .affinity + .as_ref() +diff --git a/src/runtime/builder.rs b/src/runtime/builder.rs +index 6bddecd..bba4600 100644 +--- a/src/runtime/builder.rs ++++ b/src/runtime/builder.rs +@@ -6348,6 +6348,7 @@ impl RuntimeInner { + let options = crate::runtime::blocking_pool::BlockingPoolOptions { + idle_timeout: Duration::from_secs(10), + thread_name_prefix: format!("{}-blocking", config.thread_name_prefix), ++ thread_stack_size: config.thread_stack_size, + on_thread_start: config.on_thread_start.clone(), + on_thread_stop: config.on_thread_stop.clone(), + affinity_profile: config.blocking.affinity_profile, From b8bce81923e4102050442aedf8881b01b45eb572 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 16 Oct 2026 03:10:48 +0000 Subject: [PATCH 2/7] fix(runtime): count long native polls instead of failing the task The debug-build long-poll guard asserted after a poll had already finished, so it discarded the poll's result and failed tasks whose side effects had happened: a broker connection closed after its request was sent, a code-index worker parked failed, a schema convergence reported failure. Its 250 ms thread-CPU bound also moved 70x with host load and opt level. LongPollGuard now runs in every build, measures wall time and (on unix) thread CPU, and on a poll past LONG_POLL (100 ms wall) logs a warning and counts it against the spawn site in long_native_polls(). It never changes the poll's result. The daemon's runtime hazards line lists those sites. Beside it, so long polls can be judged by their effect: - spawn_lateness_canary() measures how late native timer wakes run; the daemon runs it for its lifetime and reports late wakes at exit. - spawn_blocking runs work inline when called on a blocking-pool thread that already runs queued work, so a full pool cannot deadlock on nested blocking work. - MAX_BLOCKING_THREADS describes the native pool it bounds. The long-dispatch test #4065 added only exercised the guard and goes with it. --- .../code_index_scheduler/tests/residency.rs | 2 +- crates/tracedecay-rusqlite-runtime/Cargo.toml | 2 +- .../src/native_runtime.rs | 280 +++++++++++++++--- crates/tracedecay/src/daemon/bootstrap.rs | 35 ++- .../src/daemon/connection_serving.rs | 119 -------- .../src/daemon/store_runtime_tests.rs | 2 +- 6 files changed, 275 insertions(+), 165 deletions(-) diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/residency.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/residency.rs index 2e5e7e6b00..88961437f9 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/residency.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/residency.rs @@ -2122,7 +2122,7 @@ async fn a_memory_activation_longer_than_one_native_poll_still_seats() { .expect("mounted scope"); super::super::graph_activation::set_injected_activation_prewarm_cpu( &scope.worktree_id, - tracedecay_runtime_core::native_runtime::LONG_POLL_CPU * 4, + tracedecay_runtime_core::native_runtime::LONG_POLL * 4, ); drop(first_pass); wait_for_live_complete_generation(®istry, fixture.path()).await; diff --git a/crates/tracedecay-rusqlite-runtime/Cargo.toml b/crates/tracedecay-rusqlite-runtime/Cargo.toml index e674286a74..248e15b6ee 100644 --- a/crates/tracedecay-rusqlite-runtime/Cargo.toml +++ b/crates/tracedecay-rusqlite-runtime/Cargo.toml @@ -32,7 +32,7 @@ tracedecay-contracts = { path = "../tracedecay-contracts" } tracedecay-store = { path = "../tracedecay-store" } tracing = "0.1" -# Thread CPU time for the debug-build long-poll guard on native tasks. +# Thread CPU time for the long-poll count on native tasks. [target.'cfg(unix)'.dependencies] libc = "0.2" diff --git a/crates/tracedecay-rusqlite-runtime/src/native_runtime.rs b/crates/tracedecay-rusqlite-runtime/src/native_runtime.rs index 357fdcb6e3..36d78b2ac0 100644 --- a/crates/tracedecay-rusqlite-runtime/src/native_runtime.rs +++ b/crates/tracedecay-rusqlite-runtime/src/native_runtime.rs @@ -24,8 +24,10 @@ use asupersync::types::PanicPayload; use asupersync::{Budget, Cx}; use futures_util::future::{AbortHandle, Aborted, BoxFuture, FutureExt, Shared, abortable}; -/// Tokio's default blocking-pool ceiling. Shard closes and SQL reads park a -/// blocking thread for their whole duration. +/// The native blocking pool's thread ceiling. Shard closes, SQL reads, and +/// [`off_native_thread`] handlers each hold a pool thread for their whole +/// duration, mostly parked, so this bounds concurrent waits rather than CPU +/// use. The value matches Tokio's `max_blocking_threads` default. pub const MAX_BLOCKING_THREADS: usize = 512; static PROCESS_RUNTIME: OnceLock> = OnceLock::new(); @@ -73,6 +75,11 @@ pub fn block_on(future: F) -> Result { /// be cancelled with that region before it starts, which would strand anything /// the caller had already queued for it. /// +/// Called on a blocking-pool thread that already runs work queued here (an +/// [`off_native_thread`] handler, say), it runs `work` inline at the call: +/// queuing would hold this pool thread while waiting for another, and a full +/// pool would deadlock on that wait. +/// /// Boxed so callers prove the context and channel wait `Send` once, here, /// instead of inside every future that awaits blocking work /// (rust-lang/rust#159228). @@ -81,6 +88,11 @@ where T: Send + 'static, F: FnOnce() -> T + Send + 'static, { + if ON_BLOCKING_POOL.get() { + let ran = std::panic::catch_unwind(std::panic::AssertUnwindSafe(work)) + .map_err(|_| "native blocking work panicked".to_owned()); + return Box::pin(std::future::ready(ran)); + } let queued = queue_blocking(work); Box::pin(async move { let (cx, mut receiver) = queued?; @@ -139,12 +151,19 @@ where let (sender, receiver) = asupersync::channel::oneshot::channel(); runtime .spawn_blocking(move || { + ON_BLOCKING_POOL.set(true); let _ = sender.send_blocking(work()); }) .ok_or("native blocking pool unavailable")?; Ok((cx, receiver)) } +thread_local! { + /// Set for good once a blocking-pool thread runs work [`queue_blocking`] + /// queued; such a thread serves the pool for its whole life. + static ON_BLOCKING_POOL: std::cell::Cell = const { std::cell::Cell::new(false) }; +} + /// Drives `settlement` on the caller's executor. A caller that stops waiting /// hands the unfinished settlement to the runtime, so accepted work still /// outlives a cancelled request. Spawning a native task for every call instead @@ -281,46 +300,77 @@ impl Future for Timeout { } } -/// The most CPU time one poll of a [`NativeTask`] may take in a debug build. +/// The wall time past which one poll of a [`NativeTask`] counts as long. /// -/// A one-worker runtime has no follower to take the reactor while its worker -/// is inside a poll, and a `block_on` root never turns the reactor. Every -/// socket wake and timer on that runtime waits for the poll to return (#3994, -/// upstream Dicklesworthstone/asupersync#83, whose multi-worker half Bazel -/// patches in). -pub const LONG_POLL_CPU: Duration = Duration::from_millis(250); - -/// Fails a debug build's native task when one poll exceeds [`LONG_POLL_CPU`] -/// of thread CPU time. CPU time, not wall time, so a worker descheduled on a -/// loaded host does not trip it. -#[cfg(all(debug_assertions, unix))] +/// While a worker is inside a poll, no other task queued on that worker runs. +/// A one-worker runtime also has no follower to take the reactor, and a +/// `block_on` root never turns it, so there every socket wake and timer waits +/// for the poll to return (#3994, upstream Dicklesworthstone/asupersync#83, +/// whose multi-worker half Bazel patches in). The waiters see wall time, +/// whether the poll computed, parked on a lock or a writer reply, or was +/// descheduled, so wall time decides what counts; thread CPU time is logged +/// beside it on unix to tell computing from parking. +/// +/// 100 ms is where a reply delayed behind the poll becomes noticeable to the +/// person waiting on a hook or MCP call. A poll's cost moves with host load +/// and opt level (a 7 ms fastbuild dispatch measured 490-590 ms of CPU at load +/// average ~1000), so this bound sets the floor of a latency report, not a +/// correctness line: a long poll is logged and counted, never failed. +pub const LONG_POLL: Duration = Duration::from_millis(100); + +static LONG_POLLS: Mutex, u64>> = + Mutex::new(BTreeMap::new()); + +/// Long native-task polls this process ran, by the site that spawned the +/// task. Each entry names work that held a native worker past [`LONG_POLL`] +/// and is a candidate for [`spawn_blocking`] when its latency matters. +#[must_use] +pub fn long_native_polls() -> BTreeMap<&'static std::panic::Location<'static>, u64> { + LONG_POLLS + .lock() + .unwrap_or_else(PoisonError::into_inner) + .clone() +} + +/// Times every poll of a [`NativeTask`] and counts the ones past +/// [`LONG_POLL`] against the task's spawn site. It never alters the poll's +/// result. Per poll it reads the monotonic clock twice and the thread CPU +/// clock once, plus once more after a long poll. struct LongPollGuard { work: F, site: &'static std::panic::Location<'static>, } -#[cfg(all(debug_assertions, unix))] impl Future for LongPollGuard { type Output = F::Output; fn poll(mut self: Pin<&mut Self>, task: &mut Context<'_>) -> Poll { - let started = thread_cpu_time(); + let started = Instant::now(); + let cpu_started = thread_cpu_time(); let poll = Pin::new(&mut self.work).poll(task); - if let (Some(started), Some(ended)) = (started, thread_cpu_time()) { - let spent = ended.saturating_sub(started); - assert!( - spent <= LONG_POLL_CPU, - "native task spawned at {} spent {spent:?} of CPU in one poll, past \ - {LONG_POLL_CPU:?}; move its synchronous work to native_runtime::spawn_blocking \ - (#3994)", - self.site, + let wall = started.elapsed(); + if wall >= LONG_POLL { + let cpu = cpu_started + .zip(thread_cpu_time()) + .map(|(started, ended)| ended.saturating_sub(started)); + *LONG_POLLS + .lock() + .unwrap_or_else(PoisonError::into_inner) + .entry(self.site) + .or_default() += 1; + tracing::warn!( + spawn_site = %self.site, + ?wall, + ?cpu, + "native task held its worker for one poll past {LONG_POLL:?}; \ + native_runtime::spawn_blocking keeps that work off the worker" ); } poll } } -#[cfg(all(debug_assertions, unix))] +#[cfg(unix)] fn thread_cpu_time() -> Option { let mut now = libc::timespec { tv_sec: 0, @@ -336,6 +386,68 @@ fn thread_cpu_time() -> Option { }) } +#[cfg(not(unix))] +fn thread_cpu_time() -> Option { + None +} + +/// How long the scheduling-lateness canary sleeps between wakes. +const CANARY_TICK: Duration = Duration::from_millis(100); + +/// What the scheduling-lateness canary measured in this process. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct SchedulingLateness { + pub ticks: u64, + /// Wakes that ran [`LONG_POLL`] or more past their deadline. + pub late_ticks: u64, + pub worst: Duration, +} + +static SCHEDULING_LATENESS: Mutex = Mutex::new(SchedulingLateness { + ticks: 0, + late_ticks: 0, + worst: Duration::ZERO, +}); + +#[must_use] +pub fn scheduling_lateness() -> SchedulingLateness { + *SCHEDULING_LATENESS + .lock() + .unwrap_or_else(PoisonError::into_inner) +} + +/// Starts a native task that sleeps on a native timer again and again and +/// records how late each wake runs. A wake runs late when no worker turns the +/// timer or picks up the woken task: a long poll on a one-worker runtime, or +/// every worker busy. That is the delay a socket reply waits behind, measured +/// directly rather than inferred from [`long_native_polls`]. A wake +/// [`LONG_POLL`] late or worse is logged. Abort the task to stop it. +#[track_caller] +pub fn spawn_lateness_canary() -> Result, String> { + NativeTask::spawn(async { + loop { + let deadline = Instant::now() + CANARY_TICK; + if let Err(error) = sleep_until(deadline).await { + tracing::warn!(%error, "scheduling-lateness canary stopped"); + return; + } + let late = deadline.elapsed(); + let mut lateness = SCHEDULING_LATENESS + .lock() + .unwrap_or_else(PoisonError::into_inner); + lateness.ticks += 1; + lateness.worst = lateness.worst.max(late); + if late >= LONG_POLL { + lateness.late_ticks += 1; + tracing::warn!( + ?late, + "native runtime woke a timer late; a long poll or busy workers held it" + ); + } + } + }) +} + /// How a [`NativeTask`] ended. #[derive(Clone, Debug, PartialEq, Eq)] pub enum NativeTaskOutcome { @@ -362,8 +474,9 @@ impl NativeTask { /// runtime. The runtime owns it, not the caller's region, so it outlives /// the request that started it. /// - /// `work` must not run CPU-heavy synchronous code inside one poll; see - /// [`LONG_POLL_CPU`]. Queue that part with [`spawn_blocking`]. + /// A poll of `work` past [`LONG_POLL`] is logged and counted against the + /// caller's location in [`long_native_polls`]. Queue blocking or + /// CPU-heavy parts with [`spawn_blocking`] when their latency matters. #[track_caller] pub fn spawn(work: F) -> Result where @@ -373,7 +486,6 @@ impl NativeTask { // Boxed so the runtime's own spawn wrappers do not nest `work` in // their `Send` proof (rust-lang/rust#159228). let work: BoxFuture<'static, T> = Box::pin(work); - #[cfg(all(debug_assertions, unix))] let work = LongPollGuard { work, site: std::panic::Location::caller(), @@ -648,10 +760,6 @@ fn start(builder: RuntimeBuilder) -> Result { } #[cfg(test)] -#[allow( - clippy::disallowed_methods, - reason = "tests run on an ambient Tokio runtime" -)] mod tests { use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; @@ -1225,25 +1333,113 @@ mod tests { ); } - #[cfg(all(debug_assertions, unix))] + #[track_caller] + fn spawned_here( + work: impl Future + Send + 'static, + ) -> (NativeTask, &'static std::panic::Location<'static>) { + ( + NativeTask::spawn(work).expect("native task"), + std::panic::Location::caller(), + ) + } + + /// The guard's window encloses the spin, so a descheduled worker only + /// lengthens the measured poll. + #[test] + fn a_long_native_poll_completes_and_is_counted_at_its_spawn_site() { + let runtime = RuntimeBuilder::multi_thread() + .worker_threads(1) + .build() + .expect("test runtime"); + let (long, short) = runtime.block_on(async { + let (task, site) = spawned_here(async { + spin(LONG_POLL + Duration::from_millis(50)); + 7 + }); + let long = (task.wait().await, site); + let (task, site) = spawned_here(async { 8 }); + (long, (task.wait().await, site)) + }); + assert_eq!(long.0, NativeTaskOutcome::Completed(7)); + assert_eq!(long_native_polls().get(long.1), Some(&1), "{}", long.1); + assert_eq!(short.0, NativeTaskOutcome::Completed(8)); + assert_eq!(long_native_polls().get(short.1), None, "{}", short.1); + } + + /// Queued instead, the inner work would wait for a second pool thread + /// while the outer one holds the first; a full pool never frees one. + #[test] + fn blocking_work_spawned_from_the_blocking_pool_runs_on_the_same_thread() { + let (outer, inner) = block_on(spawn_blocking(|| { + let inner = spawn_blocking(|| std::thread::current().id()).now_or_never(); + (std::thread::current().id(), inner) + })) + .expect("process runtime") + .expect("outer blocking work"); + assert_eq!(inner, Some(Ok(outer))); + } + + #[cfg(target_os = "linux")] + fn current_thread_stack_size() -> usize { + let mut attributes = std::mem::MaybeUninit::::uninit(); + let mut size = 0; + // SAFETY: `pthread_getattr_np` initializes `attributes` for the calling + // thread before `pthread_attr_getstacksize` reads it, and they are + // destroyed once, after that read. + unsafe { + assert_eq!( + libc::pthread_getattr_np(libc::pthread_self(), attributes.as_mut_ptr()), + 0 + ); + assert_eq!( + libc::pthread_attr_getstacksize(attributes.as_ptr(), &raw mut size), + 0 + ); + libc::pthread_attr_destroy(attributes.as_mut_ptr()); + } + size + } + + #[cfg(target_os = "linux")] #[test] - fn a_native_task_poll_past_the_cpu_bound_fails_in_debug_builds() { + fn blocking_pool_threads_get_the_runtime_thread_stack_size() { + const STACK: usize = 16 * 1024 * 1024; let runtime = RuntimeBuilder::multi_thread() .worker_threads(1) + .thread_stack_size(STACK) + .blocking_threads(1, 2) .build() .expect("test runtime"); - let outcome = runtime.block_on(async { - NativeTask::spawn(async { spin(LONG_POLL_CPU + Duration::from_millis(50)) }) - .expect("native task") + let stack = runtime + .block_on(async { spawn_blocking(current_thread_stack_size).await }) + .expect("blocking work"); + assert!(stack >= STACK, "pool thread stack is {stack} bytes"); + } + + #[test] + fn the_lateness_canary_records_a_wake_held_behind_a_long_poll() { + let runtime = RuntimeBuilder::multi_thread() + .worker_threads(1) + .build() + .expect("test runtime"); + runtime.block_on(async { + let canary = spawn_lateness_canary().expect("canary"); + sleep_until(Instant::now() + Duration::from_millis(20)) + .await + .expect("let the canary arm its first timer"); + NativeTask::spawn(async { spin(CANARY_TICK + LONG_POLL * 3) }) + .expect("spinning task") .wait() + .await; + sleep_until(Instant::now() + CANARY_TICK) .await + .expect("let the canary record its overdue wake"); + canary.abort(); }); - let NativeTaskOutcome::Panicked(payload) = outcome else { - panic!("a {LONG_POLL_CPU:?} poll must fail the task, got {outcome:?}"); - }; + let lateness = scheduling_lateness(); assert!( - payload.to_string().contains("spawn_blocking"), - "the failure names the remedy: {payload}" + lateness.late_ticks >= 1 && lateness.worst >= LONG_POLL, + "{lateness:?}" ); } diff --git a/crates/tracedecay/src/daemon/bootstrap.rs b/crates/tracedecay/src/daemon/bootstrap.rs index ad0dfe5c97..d15b69ae81 100644 --- a/crates/tracedecay/src/daemon/bootstrap.rs +++ b/crates/tracedecay/src/daemon/bootstrap.rs @@ -193,6 +193,7 @@ async fn run_foreground_loopback( }, ) .await; + let lateness_canary = spawn_lateness_canary(); let spooled_hook_opener = hook_v2_replay_consumer::spawn_spooled_hook_opener( hook_v2_replay_consumer::PortableSpoolOpenerOwners { lifecycle: lifecycle.clone(), @@ -279,6 +280,9 @@ async fn run_foreground_loopback( if let Some(opener) = spooled_hook_opener { opener.abort(); } + if let Some(canary) = lateness_canary { + canary.abort(); + } cancel_retained_session_history(&store_administration).await; let shutdown_deadline = Instant::now() + DAEMON_SHUTDOWN_COORDINATION_DEADLINE; let endpoint_cleanup = authority.cleanup_owned_endpoint(); @@ -493,8 +497,15 @@ fn log_background_shutdown_receipt(receipt: &tracedecay_daemon_service::shutdown /// The daemon's lifetime count of Tokio-only paths skipped for want of a Tokio /// runtime, and of writer waits that parked a native thread. Zero is the only -/// healthy value for either. +/// healthy value for either. Long native polls are listed by spawn site, and +/// the lateness canary reports timer wakes that ran late. fn log_runtime_hazards() { + let long_native_polls = native_runtime::long_native_polls() + .into_iter() + .map(|(site, polls)| format!("{site}={polls}")) + .collect::>() + .join(","); + let lateness = native_runtime::scheduling_lateness(); log_daemon_event( "daemon_runtime_hazards", &[ @@ -506,10 +517,28 @@ fn log_runtime_hazards() { "native_thread_writer_waits", native_runtime::native_thread_writer_waits().to_string(), ), + ("long_native_polls", long_native_polls), + ("late_timer_wakes", lateness.late_ticks.to_string()), + ( + "worst_timer_lateness_ms", + lateness.worst.as_millis().to_string(), + ), ], ); } +/// Measures, for the daemon's lifetime, how late native timer wakes run. +fn spawn_lateness_canary() -> Option> { + native_runtime::spawn_lateness_canary() + .inspect_err(|error| { + log_daemon_event( + "daemon_lateness_canary_unavailable", + &[("error", error.clone())], + ); + }) + .ok() +} + fn log_project_server_shutdown_receipt(receipt: &tracedecay_store_runtime::ShutdownTaskReceipt) { if receipt.is_clean() { return; @@ -770,6 +799,7 @@ async fn run_foreground_unix( .with_pr_autotrack_task(pr_autotrack_task) .await; let mut shutdown_signals = ShutdownSignals::listen()?; + let lateness_canary = spawn_lateness_canary(); let spooled_hook_opener = hook_v2_replay_consumer::spawn_spooled_hook_opener(engine.clone(), profile.clone()); let admission = DaemonClientAdmission::new(MAX_CONCURRENT_DAEMON_CLIENTS); @@ -850,6 +880,9 @@ async fn run_foreground_unix( if let Some(rotation) = stderr_log_rotation { rotation.abort(); } + if let Some(canary) = lateness_canary { + canary.abort(); + } // Stop accepting and unlink the socket before draining so clients that // connect during shutdown get NotFound/ConnectionRefused (which they retry // via `connect_with_restart_grace`) instead of a queued connection that diff --git a/crates/tracedecay/src/daemon/connection_serving.rs b/crates/tracedecay/src/daemon/connection_serving.rs index cff8d7bf5e..97acc7d516 100644 --- a/crates/tracedecay/src/daemon/connection_serving.rs +++ b/crates/tracedecay/src/daemon/connection_serving.rs @@ -2181,122 +2181,3 @@ mod delivery_ack_tests { ); } } - -/// Debug builds fail a native task whose poll runs longer than -/// `LONG_POLL_CPU`; a broker connection's request dispatch must not be that -/// poll, or the daemon closes the socket after the request was sent (#4056). -#[cfg(all(test, unix, debug_assertions))] -mod long_dispatch_tests { - use std::collections::HashMap; - use std::io::{BufRead, BufReader}; - use std::path::PathBuf; - use std::time::{Duration, Instant}; - - use tracedecay_daemon_protocol::{ - BrokerStream, DaemonClientIdentity, DaemonHandshake, DaemonInvocationProblem, - DaemonInvocationRequest, DaemonInvocationResponse, MovedStoreAdoption, - }; - use tracedecay_daemon_service::shutdown::DaemonLifecycle; - use tracedecay_mcp::BrokerStreamTransport; - use tracedecay_runtime_core::native_runtime::{self, NativeTask, NativeTaskOutcome}; - - use super::{DaemonInvocationState, drive_retained_invocation_responses}; - - const REQUEST_ID: &str = "request.long-dispatch"; - - fn handshake() -> DaemonHandshake { - let profile_root = PathBuf::from("/profiles/long-dispatch"); - DaemonHandshake { - project_path: None, - scope_prefix: None, - timings: false, - allow_init: false, - allow_initialize_root_routing: false, - client_identity: DaemonClientIdentity { - global_db_path: profile_root.join("global.db"), - profile_root, - }, - client_version: String::new(), - client_instance_id: "client.long-dispatch".to_owned(), - tool_list_changed_capable: false, - catalog_version: String::new(), - moved_store_adoption: MovedStoreAdoption::Never, - } - } - - fn request() -> DaemonInvocationRequest { - let now = tracedecay_contracts::now_micros(); - DaemonInvocationRequest::profile_graph_tool( - REQUEST_ID, - tracedecay_tool_catalog::ApplicationSurfaceOperation::AffectedTests, - serde_json::Map::new(), - now, - tracedecay_contracts::Deadline::new(tracedecay_domain::UtcMicros(now.0 + 30_000_000)) - .expect("deadline"), - tracedecay_contracts::CancellationContext::active("cancel.long-dispatch") - .expect("cancellation"), - ) - } - - #[test] - fn a_dispatch_longer_than_one_native_poll_still_answers_its_request() { - let (daemon_end, client) = std::os::unix::net::UnixStream::pair().expect("socket pair"); - daemon_end - .set_nonblocking(true) - .expect("nonblocking daemon end"); - let daemon_end = BrokerStream::Unix( - asupersync::net::unix::UnixStream::from_std(daemon_end).expect("daemon end"), - ); - let invocation = DaemonInvocationState::default(); - let connection = NativeTask::spawn(async move { - let mut transport = BrokerStreamTransport::new(daemon_end); - drive_retained_invocation_responses( - Ok(request()), - &mut transport, - &invocation.service, - &handshake(), - &DaemonLifecycle::default(), - &mut HashMap::new(), - |request| { - Box::pin(async move { - let spun = Instant::now(); - while spun.elapsed() < native_runtime::LONG_POLL_CPU * 2 { - std::hint::spin_loop(); - } - // Await again after the synchronous work, so the - // response is not written inside the long poll. - let _ = - native_runtime::sleep_until(Instant::now() + Duration::from_millis(1)) - .await; - DaemonInvocationResponse::problem( - request.request_id, - DaemonInvocationProblem::Unavailable, - ) - }) - }, - ) - .await - .map_err(|error| error.to_string()) - }) - .expect("native connection task"); - - client - .set_read_timeout(Some(Duration::from_secs(30))) - .expect("client read timeout"); - let mut line = String::new(); - BufReader::new(&client) - .read_line(&mut line) - .expect("read the response"); - let response: DaemonInvocationResponse = - serde_json::from_str(&line).expect("the connection answers the sent request"); - assert_eq!(response.request_id, REQUEST_ID); - client - .shutdown(std::net::Shutdown::Both) - .expect("close the client"); - assert_eq!( - native_runtime::block_on(connection.wait()).expect("process runtime"), - NativeTaskOutcome::Completed(Ok(())), - "the connection ends cleanly when its client closes" - ); - } -} diff --git a/crates/tracedecay/src/daemon/store_runtime_tests.rs b/crates/tracedecay/src/daemon/store_runtime_tests.rs index bc63b09e25..e6828265a4 100644 --- a/crates/tracedecay/src/daemon/store_runtime_tests.rs +++ b/crates/tracedecay/src/daemon/store_runtime_tests.rs @@ -1366,7 +1366,7 @@ async fn schema_convergence_completes_when_its_allocator_trim_outlasts_a_native_ .await .expect("session runtime registry"); registry.inject_registered_schema_convergence_trim_cpu_for_test( - tracedecay_runtime_core::native_runtime::LONG_POLL_CPU * 4, + tracedecay_runtime_core::native_runtime::LONG_POLL * 4, ); registry From 992b2fd262233b79c04f22e156f89dfdb660dd28 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 16 Oct 2026 03:10:49 +0000 Subject: [PATCH 3/7] fix(cli): run long-lived commands on at least two native workers The reactor handoff patch promotes a parked follower only when a runtime has two or more parkers, so on a one-CPU host daemon run and serve held every socket wake and timer behind any long poll. Commands other than the one-shot tool and status clients now start at least two native workers. The native blocking pool takes MAX_BLOCKING_THREADS directly instead of max(tokio limit, 512), which always resolved to 512. --- crates/tracedecay-cli/src/main.rs | 13 ++++++----- crates/tracedecay-cli/src/startup_tests.rs | 26 ++++++++++++++++++---- 2 files changed, 30 insertions(+), 9 deletions(-) diff --git a/crates/tracedecay-cli/src/main.rs b/crates/tracedecay-cli/src/main.rs index 25e09fc13d..e71558d09c 100644 --- a/crates/tracedecay-cli/src/main.rs +++ b/crates/tracedecay-cli/src/main.rs @@ -233,6 +233,7 @@ impl Drop for Spinner { /// an explicit stack size gives every platform the same headroom. const ASYNC_STACK_BYTES: usize = 16 * 1024 * 1024; const MAX_ASYNC_WORKER_THREADS: usize = 16; +const MIN_LONG_LIVED_WORKER_THREADS: usize = 2; /// The runtimes a command's process owns. The native runtime drives every /// command's future. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -256,9 +257,14 @@ fn async_worker_threads(command: Option<&Commands>) -> usize { if is_one_shot_daemon_client(command) { return 1; } + // At least two, even on a one-CPU host: the reactor handoff patch + // (asupersync-reactor-handoff.patch) promotes a parked follower only when + // the runtime has two or more parkers (`count < 2` returns early), so on + // one worker a long poll in `daemon run` or `serve` holds every socket + // wake and timer until it returns. std::thread::available_parallelism() .map_or(1, usize::from) - .clamp(1, MAX_ASYNC_WORKER_THREADS) + .clamp(MIN_LONG_LIVED_WORKER_THREADS, MAX_ASYNC_WORKER_THREADS) } /// Every command but `daemon run` is a daemon client or a local store tool @@ -572,10 +578,7 @@ fn async_main() -> tracedecay_domain::errors::Result { .thread_name_prefix("tracedecay-native") .worker_threads(worker_threads) .thread_stack_size(ASYNC_STACK_BYTES) - .blocking_threads( - 1, - blocking_threads.max(native_runtime::MAX_BLOCKING_THREADS), - ), + .blocking_threads(1, native_runtime::MAX_BLOCKING_THREADS), ) .map_err(|error| tracedecay_domain::errors::TraceDecayError::Config { message: error.to_string(), diff --git a/crates/tracedecay-cli/src/startup_tests.rs b/crates/tracedecay-cli/src/startup_tests.rs index 4f49b78a07..d0c6a53bb0 100644 --- a/crates/tracedecay-cli/src/startup_tests.rs +++ b/crates/tracedecay-cli/src/startup_tests.rs @@ -1,10 +1,11 @@ use super::{ AnalyticsAction, Cli, CommandFamily, CommandRuntime, Commands, DAEMON_CPU_THREADS_ENV, DEFAULT_MAX_DAEMON_CPU_THREADS, HostBundleCliOptions, PackageHookAction, RAYON_NUM_THREADS_ENV, - ScoopPackageHookAction, StderrTracingDefault, command_profile_label, command_runtime, - daemon_cpu_threads_from, is_one_shot_daemon_client, normalize_tool_reserved_global_flags, - runs_worldwide_counter_flush, should_skip_agent_install_check, should_skip_startup_maintenance, - stderr_tracing_default, try_print_cli_version, validate_host_bundle_options, + ScoopPackageHookAction, StderrTracingDefault, async_worker_threads, command_profile_label, + command_runtime, daemon_cpu_threads_from, is_one_shot_daemon_client, + normalize_tool_reserved_global_flags, runs_worldwide_counter_flush, + should_skip_agent_install_check, should_skip_startup_maintenance, stderr_tracing_default, + try_print_cli_version, validate_host_bundle_options, }; use clap::{CommandFactory, Parser}; use std::iter; @@ -390,6 +391,23 @@ fn status_is_a_one_shot_daemon_client() { ])))); } +#[test] +fn long_lived_commands_run_at_least_two_native_workers() { + for args in [&["daemon", "run"][..], &["serve"]] { + assert!( + async_worker_threads(Some(&parse_command(args))) >= 2, + "{args:?}" + ); + } + for args in [&["status"][..], &["tool", "tracedecay_status"]] { + assert_eq!( + async_worker_threads(Some(&parse_command(args))), + 1, + "{args:?}" + ); + } +} + #[test] fn only_the_foreground_daemon_owns_a_tokio_runtime() { for args in [ From 5ec9585a479f4ddeff257a0238025d72341c51b2 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 16 Oct 2026 03:10:52 +0000 Subject: [PATCH 4/7] fix(runtime): name the real spawn site in long-poll warnings spawn_detached, spawn_maintenance_task, spawn_doctor_task, and LspRuntimeSpawner::spawn wrap NativeTask::spawn without #[track_caller], so every task they start counted against the wrapper's line. They now pass their caller's location through. --- .../tracedecay-code-index-runtime/src/code_index_task_support.rs | 1 + crates/tracedecay-lsp/src/gateway/runtime.rs | 1 + crates/tracedecay/src/daemon/maintenance.rs | 1 + crates/tracedecay/src/doctor.rs | 1 + 4 files changed, 4 insertions(+) diff --git a/crates/tracedecay-code-index-runtime/src/code_index_task_support.rs b/crates/tracedecay-code-index-runtime/src/code_index_task_support.rs index 3e2b508d8a..836e9dd92a 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_task_support.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_task_support.rs @@ -102,6 +102,7 @@ pub async fn settle_owned_blocking_task( /// Starts `work` on the native runtime and detaches it. For work that never /// reaches Tokio. A refused start is logged, and `work` is dropped. +#[track_caller] pub(crate) fn spawn_detached(event: &'static str, work: impl Future + Send + 'static) { if let Err(error) = native_runtime::NativeTask::spawn(work) { tracing::warn!(event, %error, "native task did not start"); diff --git a/crates/tracedecay-lsp/src/gateway/runtime.rs b/crates/tracedecay-lsp/src/gateway/runtime.rs index 201a8a62ff..b421075861 100644 --- a/crates/tracedecay-lsp/src/gateway/runtime.rs +++ b/crates/tracedecay-lsp/src/gateway/runtime.rs @@ -42,6 +42,7 @@ pub trait LspRuntimeTask: Send + Sync { /// Runtime injection used by broker policy without taking a Tokio dependency. pub trait LspRuntimeSpawner: Send + Sync { + #[track_caller] fn spawn(&self, future: LspRuntimeFuture<()>) -> Box; } diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index db2c58b059..300da67891 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -70,6 +70,7 @@ async fn registered_project_stores( /// Starts one maintenance owner on the native runtime. A refusal is logged /// with the owner it leaves unstarted. +#[track_caller] fn spawn_maintenance_task( owner: &'static str, work: impl std::future::Future + Send + 'static, diff --git a/crates/tracedecay/src/doctor.rs b/crates/tracedecay/src/doctor.rs index 052ec06b91..6b8fb5e20a 100644 --- a/crates/tracedecay/src/doctor.rs +++ b/crates/tracedecay/src/doctor.rs @@ -286,6 +286,7 @@ where /// A daemon round trip running on the native runtime beside the blocking /// checks. The result travels over a channel because a `NativeTask` output /// must be `Clone`, and the doctor's error type is not. +#[track_caller] fn spawn_doctor_task( work: impl Future + Send + 'static, ) -> DoctorCheck { From b70a9fb51b6458ef27ec15e408df38fa01b7623f Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 16 Oct 2026 03:17:32 +0000 Subject: [PATCH 5/7] test: drop two tests that only exercised the long-poll guard a_memory_activation_longer_than_one_native_poll_still_seats (#4022) and schema_convergence_completes_when_its_allocator_trim_outlasts_a_native_poll (#4027) inject CPU so a poll trips the guard. With long polls counted instead of failed, both pass whether or not the prewarm and the trim stay on a worker. --- .../code_index_scheduler/tests/residency.rs | 35 ----------------- .../src/daemon/store_runtime_tests.rs | 38 ------------------- 2 files changed, 73 deletions(-) diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/residency.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/residency.rs index 88961437f9..114b5106f1 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/residency.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/residency.rs @@ -2093,38 +2093,3 @@ async fn readers_of_a_build_waiting_for_memory_do_not_spin_the_worker() { registry.shutdown().await; } - -/// A memory activation whose prewarm outlasts the CPU one native poll may -/// spend still seats the generation a reader demanded. The prewarm is -/// synchronous: run inside the worker's poll, the debug long-poll guard fails -/// the worker and the seat never arrives (#4016). -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn a_memory_activation_longer_than_one_native_poll_still_seats() { - let fixture = GitFixture::new(ALPHA_LIB_V1); - let store = TempDir::new().expect("store root"); - let registry = CodeIndexSchedulerRegistryV1::with_background_reconcile_permits(1, 1); - let first_pass = registry - .background_reconcile_admission() - .acquire_owned() - .await - .expect("hold the worker before its first activation"); - registry - .mount_worktree( - test_project_id(), - fixture.path(), - store.path().to_path_buf(), - ) - .await - .expect("mount"); - let scope = registry - .serving_code_scope(fixture.path()) - .await - .expect("mounted scope"); - super::super::graph_activation::set_injected_activation_prewarm_cpu( - &scope.worktree_id, - tracedecay_runtime_core::native_runtime::LONG_POLL * 4, - ); - drop(first_pass); - wait_for_live_complete_generation(®istry, fixture.path()).await; - registry.shutdown().await; -} diff --git a/crates/tracedecay/src/daemon/store_runtime_tests.rs b/crates/tracedecay/src/daemon/store_runtime_tests.rs index e6828265a4..ae6975a853 100644 --- a/crates/tracedecay/src/daemon/store_runtime_tests.rs +++ b/crates/tracedecay/src/daemon/store_runtime_tests.rs @@ -1349,44 +1349,6 @@ async fn schema_convergence_is_reported_and_completes_after_admission() { ); } -/// Convergence ends with a glibc `malloc_trim`, which walks every arena and -/// measured up to 0.8 s of CPU in a loaded test process. Run inside the -/// convergence task's poll, it stalls every socket wake on that native worker, -/// and the debug long-poll guard fails the task as it completes (#4023). -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn schema_convergence_completes_when_its_allocator_trim_outlasts_a_native_poll() { - let (_temporary, identity, project_id, project_root, _sessions_path, _database_scope) = - project_sessions_pending_convergence("project.schema-long-trim").await; - let shard_id = StoreShardIdV1::project_sessions( - identity.brain_id().clone(), - identity.profile_id().clone(), - project_id.clone(), - ); - let registry = DaemonSessionRuntimeRegistryV1::open_with_session_maintenance(identity, true) - .await - .expect("session runtime registry"); - registry.inject_registered_schema_convergence_trim_cpu_for_test( - tracedecay_runtime_core::native_runtime::LONG_POLL * 4, - ); - - registry - .project_sessions(project_id, [project_root]) - .await - .expect("project sessions admission"); - - assert_eq!( - wait_for_schema_convergence(®istry, &shard_id).await, - RegisteredSchemaConvergenceStatus::Complete - ); - tokio::time::timeout( - MUST_NOT_BLOCK, - registry.shutdown_terminal_tasks(&tracedecay_runtime_core::native_runtime::cx().unwrap()), - ) - .await - .expect("terminal shutdown joins the finished convergence task") - .expect("the convergence task must finish without failing"); -} - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn foreground_project_open_defers_historical_convergence_until_full_publication() { let (_temporary, identity, project_id, project_root, _sessions_path, _database_scope) = From 4603e1976d704e202e5c1e0bb106473014898691 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 16 Oct 2026 04:12:15 +0000 Subject: [PATCH 6/7] docs(cli): name the scheduler patch the worker floor depends on --- crates/tracedecay-cli/src/main.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/tracedecay-cli/src/main.rs b/crates/tracedecay-cli/src/main.rs index e71558d09c..a23092ea61 100644 --- a/crates/tracedecay-cli/src/main.rs +++ b/crates/tracedecay-cli/src/main.rs @@ -258,7 +258,7 @@ fn async_worker_threads(command: Option<&Commands>) -> usize { return 1; } // At least two, even on a one-CPU host: the reactor handoff patch - // (asupersync-reactor-handoff.patch) promotes a parked follower only when + // (asupersync-scheduler.patch) promotes a parked follower only when // the runtime has two or more parkers (`count < 2` returns early), so on // one worker a long poll in `daemon run` or `serve` holds every socket // wake and timer until it returns. From 783ddaf54e5b33068c9186dda797ca3e4dcb29b6 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 16 Oct 2026 04:24:03 +0000 Subject: [PATCH 7/7] fix(runtime): run blocking work inline under spawn_off_native_thread NativeTask::spawn_off_native_thread polls its work on a blocking-pool thread without marking it, so spawn_blocking called from that work queued behind the thread it held. It now marks the thread like queued blocking work does. --- .../src/native_runtime.rs | 24 ++++++++++++++++++- 1 file changed, 23 insertions(+), 1 deletion(-) diff --git a/crates/tracedecay-rusqlite-runtime/src/native_runtime.rs b/crates/tracedecay-rusqlite-runtime/src/native_runtime.rs index df1ed32984..6062c3b582 100644 --- a/crates/tracedecay-rusqlite-runtime/src/native_runtime.rs +++ b/crates/tracedecay-rusqlite-runtime/src/native_runtime.rs @@ -160,7 +160,8 @@ where thread_local! { /// Set for good once a blocking-pool thread runs work [`queue_blocking`] - /// queued; such a thread serves the pool for its whole life. + /// or [`NativeTask::spawn_off_native_thread`] queued; such a thread serves + /// the pool for its whole life. static ON_BLOCKING_POOL: std::cell::Cell = const { std::cell::Cell::new(false) }; } @@ -524,6 +525,7 @@ impl NativeTask { handle() .ok_or("native execution runtime unavailable")? .spawn_blocking(move || { + ON_BLOCKING_POOL.set(true); let polled = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { poll_on_current_thread(work) })); @@ -1523,6 +1525,26 @@ mod tests { assert_eq!(inner, Some(Ok(outer))); } + /// A fresh one-thread pool, so the thread has run no other queued work. + #[test] + fn blocking_work_from_a_task_spawned_off_native_threads_runs_on_its_thread() { + let runtime = RuntimeBuilder::multi_thread() + .worker_threads(1) + .blocking_threads(1, 1) + .build() + .expect("test runtime"); + let outcome = runtime.block_on(async { + NativeTask::spawn_off_native_thread(async { + spawn_blocking(|| std::thread::current().id()).now_or_never() + == Some(Ok(std::thread::current().id())) + }) + .expect("blocking-pool task") + .wait() + .await + }); + assert_eq!(outcome, NativeTaskOutcome::Completed(true)); + } + #[cfg(target_os = "linux")] fn current_thread_stack_size() -> usize { let mut attributes = std::mem::MaybeUninit::::uninit();