From 77473dcfb1d3850bd2a4d70fc7275e0754fb13b4 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 15 Oct 2026 19:22:05 +0000 Subject: [PATCH 1/2] test(daemon): pin zero native writer waits for Work/Workflow dispatch Drive a workflow registration, a run read, a Work attempt list and a fan-out binding read through the daemon invocation service on a native worker, and assert native_thread_writer_waits() does not move. On master it moves by 21. --- .../src/daemon/invocation_tests/work_tests.rs | 182 ++++++++++++++++-- 1 file changed, 164 insertions(+), 18 deletions(-) diff --git a/crates/tracedecay/src/daemon/invocation_tests/work_tests.rs b/crates/tracedecay/src/daemon/invocation_tests/work_tests.rs index ba5ad25e8f..149f46fad2 100644 --- a/crates/tracedecay/src/daemon/invocation_tests/work_tests.rs +++ b/crates/tracedecay/src/daemon/invocation_tests/work_tests.rs @@ -924,14 +924,18 @@ impl RegisteredWorkFixture { grant_digest.clone(), actor.clone(), UtcMicros(1), - UtcMicros(10_000), + // Workflow dispatch admits at the wall clock, not the request's + // observed time. + UtcMicros(tracedecay_contracts::now_micros().0 + 3_600_000_000), scope.clone(), tracedecay_contracts::WORK_APPLICATION_OPERATION_IDS_V1 .iter() + .chain(&tracedecay_contracts::WORKFLOW_APPLICATION_OPERATION_IDS) .map(|(_, capability, _)| CapabilityId::new(*capability).expect("capability")) .collect(), tracedecay_contracts::WORK_APPLICATION_OPERATION_IDS_V1 .iter() + .chain(&tracedecay_contracts::WORKFLOW_APPLICATION_OPERATION_IDS) .map(|(_, _, use_case)| UseCaseId::new(*use_case).expect("use case")) .collect(), DisclosureClass::Sensitive, @@ -987,23 +991,165 @@ impl RegisteredWorkFixture { request_id: &'static str, request: WorkApplicationInvocationV1, ) -> DaemonInvocationOutcome { - self.service - .invoke( - &self.registry, - Some(self.project.path()), - None, - None, - None, - DaemonInvocationRequest::work_application( - request_id, + invoke_registered( + self.service.clone(), + Arc::clone(&self.registry), + self.project.path().to_path_buf(), + DaemonInvocationRequest::work_application( + request_id, + request, + UtcMicros(100), + Deadline::new(UtcMicros(1_000)).expect("deadline"), + CancellationContext::active(format!("cancel.{request_id}")).expect("cancellation"), + ), + ) + .await + } +} + +async fn invoke_registered( + service: DaemonInvocationService, + registry: Arc>, + project_root: std::path::PathBuf, + request: DaemonInvocationRequest, +) -> DaemonInvocationOutcome { + service + .invoke(®istry, Some(&project_root), None, None, None, request) + .await + .outcome +} + +/// Broker-socket connections run on native runtime workers. Work and Workflow +/// storage ports are synchronous and wait on the `SQLite` writer, so dispatch +/// must run them off those workers: a parked worker drives no other task. +#[tokio::test] +async fn work_and_workflow_dispatch_never_park_a_native_worker_on_the_writer() { + let fixture = RegisteredWorkFixture::start("native-writer-waits").await; + let wall_clock = tracedecay_contracts::now_micros(); + let workflow = |request_id: &str, request| { + DaemonInvocationRequest::workflow_application( + request_id, + request, + wall_clock, + Deadline::new(UtcMicros(wall_clock.0 + 600_000_000)).expect("deadline"), + CancellationContext::active(format!("cancel.{request_id}")).expect("cancellation"), + ) + }; + let digest = |byte: char| { + ManifestDigest::new(format!("sha256:{}", byte.to_string().repeat(64))).expect("digest") + }; + let definition = tracedecay_domain::WorkflowDefinition::new( + tracedecay_domain::WorkflowDefinitionId::new("workflow.native-writer-waits") + .expect("definition id"), + 1, + fixture.scope.project_id.clone(), + vec![tracedecay_domain::WorkflowStep { + step_id: tracedecay_domain::WorkflowStepId::new("prepare").expect("step id"), + operation: tracedecay_domain::WorkflowOperationRef::new("operation.work.start_attempt") + .expect("operation"), + predecessors: BTreeSet::new(), + inputs: Vec::new(), + outputs: Vec::new(), + fan_out: None, + }], + digest('a'), + digest('b'), + digest('c'), + ) + .expect("workflow definition"); + let requests = vec![ + workflow( + "request.workflow.native-writer-waits.register", + tracedecay_daemon_protocol::WorkflowApplicationInvocation::RegisterDefinition( + tracedecay_contracts::WorkflowDefinitionRegisterRequest { definition }, + ), + ), + workflow( + "request.workflow.native-writer-waits.get-run", + tracedecay_daemon_protocol::WorkflowApplicationInvocation::GetRun( + tracedecay_contracts::WorkflowRunGetRequest { + run_id: tracedecay_domain::RunId::new("run.native-writer-waits") + .expect("run id"), + }, + ), + ), + DaemonInvocationRequest::work_application( + "request.work.native-writer-waits.list", + WorkApplicationInvocationV1::ListAttempts( + tracedecay_contracts::WorkAttemptListRequestV1 { + page_size: 10, + cursor: None, + }, + ), + UtcMicros(100), + Deadline::new(UtcMicros(1_000)).expect("deadline"), + CancellationContext::active("cancel.request.work.native-writer-waits.list") + .expect("cancellation"), + ), + ]; + let service = fixture.service.clone(); + let registry = Arc::clone(&fixture.registry); + let project_root = fixture.project.path().to_path_buf(); + + let writer_waits = tracedecay_runtime_core::native_runtime::native_thread_writer_waits(); + let served = tracedecay_runtime_core::native_runtime::NativeTask::spawn(async move { + let mut outcomes = Vec::new(); + for request in requests { + outcomes.push( + invoke_registered( + service.clone(), + Arc::clone(®istry), + project_root.clone(), request, - UtcMicros(100), - Deadline::new(UtcMicros(1_000)).expect("deadline"), - CancellationContext::active(format!("cancel.{request_id}")) - .expect("cancellation"), - ), + ) + .await, + ); + } + let binding = service + .work_fan_out_binding( + Some(&project_root), + &tracedecay_domain::WorkAttemptIdentityV1::new( + TaskId::new("task.native-writer-waits").expect("task id"), + tracedecay_domain::RunId::new("run.native-writer-waits").expect("run id"), + tracedecay_domain::AttemptId::new("attempt.native-writer-waits") + .expect("attempt id"), + ) + .expect("attempt identity"), ) - .await - .outcome - } + .await; + (outcomes, binding) + }) + .expect("native worker task") + .wait() + .await; + let tracedecay_runtime_core::native_runtime::NativeTaskOutcome::Completed((outcomes, binding)) = + served + else { + panic!("the native worker must serve every request: {served:?}"); + }; + assert_eq!( + tracedecay_runtime_core::native_runtime::native_thread_writer_waits(), + writer_waits, + "Work and Workflow dispatch must not park a native worker on the writer" + ); + + let [register, get_run, list] = outcomes.as_slice() else { + panic!("three outcomes: {outcomes:?}"); + }; + assert!( + matches!( + register, + DaemonInvocationOutcome::WorkflowApplication { .. } + ), + "registration must reach the workflow effect journal: {register:?}" + ); + assert!( + matches!(get_run, DaemonInvocationOutcome::ApplicationProblem { .. }), + "an unknown run must be read from storage and refused: {get_run:?}" + ); + assert!( + matches!(list, DaemonInvocationOutcome::WorkApplication { .. }), + "the attempt list must be answered: {list:?}" + ); + assert_eq!(binding, None, "no workflow owns an unknown attempt"); } From 43569e8a41298eae9104dcf7d18cc5d8af08de49 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 15 Oct 2026 19:22:07 +0000 Subject: [PATCH 2/2] perf(daemon): serve Work/Workflow requests off native workers Work and Workflow storage ports are synchronous and wait on the SQLite writer by parking their thread. Since #4003 broker-socket connections run on native workers, so every such wait parked a native worker (594 waits across daemon_suite). native_runtime::off_native_thread moves a future from a native thread to one blocking-pool thread and polls it there with a thread-park waker; other callers poll in place. execute_work_application, execute_workflow_application and work_fan_out_binding run through it, so every caller (including guided-job steps that recurse into Work dispatch) leaves the native workers in one place, one pool thread per request. Fixes #4046 --- .../src/invocation/observability_producer.rs | 20 ++++-- .../src/invocation/work.rs | 19 +++++- .../src/invocation/work/workflow_dispatch.rs | 35 +++++++++- .../src/exact_sql/mod.rs | 2 +- .../src/native_runtime.rs | 68 +++++++++++++++++++ 5 files changed, 136 insertions(+), 8 deletions(-) diff --git a/crates/tracedecay-daemon-service/src/invocation/observability_producer.rs b/crates/tracedecay-daemon-service/src/invocation/observability_producer.rs index 027fbf5e4d..b36ecb3093 100644 --- a/crates/tracedecay-daemon-service/src/invocation/observability_producer.rs +++ b/crates/tracedecay-daemon-service/src/invocation/observability_producer.rs @@ -238,13 +238,23 @@ impl DaemonInvocationService { .project_runtimes .get::(project_root?) .await?; - let workflow = - tracedecay_application::work::RegisteredWorkflowApplicationServicesV1::attach( - &runtime.database, + let identity = identity.clone(); + // The port waits on the writer; keep that wait off native workers. + tracedecay_runtime_core::native_runtime::off_native_thread(async move { + let workflow = + tracedecay_application::work::RegisteredWorkflowApplicationServicesV1::attach( + &runtime.database, + ) + .ok()?; + tracedecay_contracts::WorkflowRunStoragePort::fan_out_binding( + workflow.effects(), + &identity, ) - .ok()?; - tracedecay_contracts::WorkflowRunStoragePort::fan_out_binding(workflow.effects(), identity) .ok() .flatten() + }) + .await + .ok() + .flatten() } } diff --git a/crates/tracedecay-daemon-service/src/invocation/work.rs b/crates/tracedecay-daemon-service/src/invocation/work.rs index d3e8e0b186..23176d8317 100644 --- a/crates/tracedecay-daemon-service/src/invocation/work.rs +++ b/crates/tracedecay-daemon-service/src/invocation/work.rs @@ -106,8 +106,9 @@ pub async fn execute_work_application( deadline: Deadline, cancellation: CancellationContext, ) -> DaemonInvocationResponse { + let refused_request_id = request_id.clone(); // Keep the Work dispatch and activity-publication frame out of invocation callers. - Box::pin(async move { + off_native_worker(refused_request_id, async move { let activity_database = registered.database.clone(); let activity_root = project_root.clone(); let mutates = work_invocation_mutates(&request); @@ -146,6 +147,22 @@ pub async fn execute_work_application( .await } +/// Serves a Work or Workflow request off native runtime workers. Their storage +/// ports are synchronous and park the calling thread on every writer wait, +/// and a parked worker drives no other task. +async fn off_native_worker( + request_id: String, + request: impl Future + Send + 'static, +) -> DaemonInvocationResponse { + match tracedecay_runtime_core::native_runtime::off_native_thread(request).await { + Ok(response) => response, + Err(error) => { + tracing::error!(%error, "Work request did not run on the blocking pool"); + DaemonInvocationResponse::problem(request_id, DaemonInvocationProblem::Unavailable) + } + } +} + fn publish_committed_task_activity_in_background( database: tracedecay_global_db::RegisteredGlobalDbLeaseV1, project_root: PathBuf, diff --git a/crates/tracedecay-daemon-service/src/invocation/work/workflow_dispatch.rs b/crates/tracedecay-daemon-service/src/invocation/work/workflow_dispatch.rs index 15c264ac60..036849bcd6 100644 --- a/crates/tracedecay-daemon-service/src/invocation/work/workflow_dispatch.rs +++ b/crates/tracedecay-daemon-service/src/invocation/work/workflow_dispatch.rs @@ -30,8 +30,41 @@ use super::workflow_run_control::{ use super::{RegisteredWorkRuntime, work_request_context, workflow_census}; #[allow(clippy::too_many_arguments)] -#[tracing::instrument(name = "daemon.service.workflow.execute", level = "trace", skip_all)] pub(crate) async fn execute_workflow_application( + registered: RegisteredWorkRuntime, + attempt_processes: Arc, + observability_producer: Option< + Arc, + >, + project_root: PathBuf, + request_id: String, + request: WorkflowApplicationInvocation, + observed_at: UtcMicros, + deadline: Deadline, + cancellation: CancellationContext, + worktree_holder_admission: tracedecay_agent_hosts::native_integration::WorktreeHolderAdmissionFenceV1, +) -> DaemonInvocationResponse { + super::off_native_worker( + request_id.clone(), + workflow_application( + registered, + attempt_processes, + observability_producer, + project_root, + request_id, + request, + observed_at, + deadline, + cancellation, + worktree_holder_admission, + ), + ) + .await +} + +#[allow(clippy::too_many_arguments)] +#[tracing::instrument(name = "daemon.service.workflow.execute", level = "trace", skip_all)] +async fn workflow_application( registered: RegisteredWorkRuntime, attempt_processes: Arc, observability_producer: Option< diff --git a/crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs b/crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs index 486440e3d2..ffb49fea29 100644 --- a/crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs @@ -120,7 +120,7 @@ pub struct ExactSqlHandle { /// (a worker or a `block_on` caller; its blocking pool carries no runtime /// handle) still parks that thread. Each such wait is logged and counted in /// [`native_thread_writer_waits`] so the caller can be moved onto -/// `native_runtime::spawn_blocking`. +/// `native_runtime::off_native_thread` or `native_runtime::spawn_blocking`. #[track_caller] fn recv_writer_reply( response: async_channel::Receiver, diff --git a/crates/tracedecay-rusqlite-runtime/src/native_runtime.rs b/crates/tracedecay-rusqlite-runtime/src/native_runtime.rs index fd10182ae7..357fdcb6e3 100644 --- a/crates/tracedecay-rusqlite-runtime/src/native_runtime.rs +++ b/crates/tracedecay-rusqlite-runtime/src/native_runtime.rs @@ -91,6 +91,44 @@ where }) } +/// Runs `future` where a thread may park: from a native runtime thread it +/// moves the whole future to one blocking-pool thread and polls it there; +/// anywhere else it polls in place. +/// +/// For request handlers that call synchronous storage ports, which wait on +/// the writer by parking their thread. Native wakes reach the pool thread +/// through an ordinary `Waker`, so the handler's awaits still work. As with +/// [`spawn_blocking`], dropping the returned future abandons only the wait: +/// a moved future runs to completion. +pub fn off_native_thread(future: F) -> BoxFuture<'static, Result> +where + F: Future + Send + 'static, + F::Output: Send + 'static, +{ + if Runtime::current_handle().is_none() { + return Box::pin(future.map(Ok)); + } + spawn_blocking(move || poll_on_current_thread(future)) +} + +fn poll_on_current_thread(future: F) -> F::Output { + struct Unpark(std::thread::Thread); + impl std::task::Wake for Unpark { + fn wake(self: std::sync::Arc) { + self.0.unpark(); + } + } + let waker = std::task::Waker::from(std::sync::Arc::new(Unpark(std::thread::current()))); + let mut context = Context::from_waker(&waker); + let mut future = std::pin::pin!(future); + loop { + if let Poll::Ready(output) = future.as_mut().poll(&mut context) { + return output; + } + std::thread::park(); + } +} + fn queue_blocking(work: F) -> Result<(Cx, asupersync::channel::oneshot::Receiver), String> where T: Send + 'static, @@ -774,6 +812,36 @@ mod tests { ); } + #[test] + fn off_native_thread_moves_a_native_caller_and_keeps_its_awaits_working() { + let off_native = || async { + let parked_thread = Runtime::current_handle().is_none(); + let slept = sleep_until(Instant::now() + Duration::from_millis(5)).await; + let blocking = spawn_blocking(|| 7).await; + (parked_thread, slept, blocking) + }; + let task = NativeTask::spawn(async move { + let caller_was_native = Runtime::current_handle().is_some(); + (caller_was_native, off_native_thread(off_native()).await) + }) + .expect("native task"); + assert_eq!( + block_on(task.wait()).expect("process runtime"), + NativeTaskOutcome::Completed((true, Ok((true, Ok(()), Ok(7))))) + ); + let in_place = std::thread::spawn(move || { + let thread = std::thread::current().id(); + off_native_thread(async move { std::thread::current().id() == thread }).now_or_never() + }) + .join() + .expect("plain thread"); + assert_eq!( + in_place, + Some(Ok(true)), + "a non-native caller polls in place" + ); + } + #[tokio::test(flavor = "multi_thread")] async fn a_past_deadline_takes_a_ready_value_and_elapses_a_pending_one() { let past = Instant::now() - Duration::from_millis(1);