From 71f96677f80bc6ad55976b313a70968b70ba8335 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 16 Oct 2026 12:48:21 +0000 Subject: [PATCH] refactor(work): run pause and resume as one writer job PauseRun and ResumeRun now run the contracts run-control service inside one ExactSqlHandle::transact job over WorkSqliteJobStorage, a Work storage that holds the job transaction and no handle. The frontier read, the workflow journal read and the CAS publication commit together, and the awaiting task parks no native thread. The pause/resume reads and publication move to a WorkRunControlTransitionPort without a Send + Sync bound so the job-scoped storage can implement it. Their bodies, the workflow schema check and the run journal replay are shared between the handle and job paths through RegisteredWorkQuery and a new RegisteredWorkWrite trait. The handle-based transition stays for the workflow fan-out run-control sync, which is still synchronous. --- .../src/work/registered.rs | 43 +++ .../tests/application_suite/main.rs | 1 + .../work_run_control_jobs.rs | 253 +++++++++++++ crates/tracedecay-contracts/src/lib.rs | 4 +- .../src/work_run_control.rs | 113 +++--- .../work_run_control_service.rs | 104 +++--- .../src/invocation/work/request_dispatch.rs | 112 +++--- .../tracedecay-rusqlite-runtime/src/work.rs | 4 - .../src/work/sql.rs | 36 ++ .../src/work_run_control.rs | 345 +++++++++++------- .../src/workflow.rs | 24 +- .../src/workflow/run_journal.rs | 27 +- .../rusqlite_suite/work_attempt_storage.rs | 7 +- .../work_run_control_storage.rs | 69 +++- 14 files changed, 821 insertions(+), 321 deletions(-) create mode 100644 crates/tracedecay-application/tests/application_suite/work_run_control_jobs.rs diff --git a/crates/tracedecay-application/src/work/registered.rs b/crates/tracedecay-application/src/work/registered.rs index a5e9c83513..70ba77dc34 100644 --- a/crates/tracedecay-application/src/work/registered.rs +++ b/crates/tracedecay-application/src/work/registered.rs @@ -75,6 +75,7 @@ pub struct RegisteredWorkApplicationServicesV1 { duplicate_adjudications: tracedecay_contracts::WorkDuplicateAdjudicationServiceV1< tracedecay_rusqlite_runtime::work::WorkSqliteStorage, >, + storage: tracedecay_rusqlite_runtime::work::WorkSqliteStorage, } /// The Work product graph authority: its verified reads and its journaled @@ -241,9 +242,47 @@ impl RegisteredWorkApplicationServicesV1 { duplicate_adjudications: tracedecay_contracts::WorkDuplicateAdjudicationServiceV1::new( storage.clone(), ), + storage, }) } + /// Pauses one run as a single writer transaction: the frontier read, the + /// decision and the publication commit together or not at all. + pub async fn pause_run( + &self, + context: tracedecay_contracts::RequestContext, + command: tracedecay_contracts::PauseWorkRunCommand, + ) -> Result< + tracedecay_contracts::WorkRunControlTransitionReceiptV1, + tracedecay_contracts::ApplicationProblem, + > { + self.storage + .transact(move |storage| { + tracedecay_contracts::WorkRunControlService::new(storage) + .pause_with_receipt(&context, command.clone()) + }) + .await + .unwrap_or_else(|_| Err(run_control_unavailable())) + } + + /// Resumes one run as a single writer transaction. + pub async fn resume_run( + &self, + context: tracedecay_contracts::RequestContext, + command: tracedecay_contracts::ResumeWorkRunCommand, + ) -> Result< + tracedecay_contracts::WorkRunControlTransitionReceiptV1, + tracedecay_contracts::ApplicationProblem, + > { + self.storage + .transact(move |storage| { + tracedecay_contracts::WorkRunControlService::new(storage) + .resume_with_receipt(&context, command.clone()) + }) + .await + .unwrap_or_else(|_| Err(run_control_unavailable())) + } + pub fn attempts( &self, ) -> &tracedecay_contracts::WorkAttemptService< @@ -383,6 +422,10 @@ pub fn authorized_work_intelligence_service( )) } +fn run_control_unavailable() -> tracedecay_contracts::ApplicationProblem { + tracedecay_contracts::WorkRunControlStorageError::Unavailable.into() +} + fn attach_error(operation: &str, error: impl std::fmt::Display) -> TraceDecayError { TraceDecayError::Database { operation: operation.to_owned(), diff --git a/crates/tracedecay-application/tests/application_suite/main.rs b/crates/tracedecay-application/tests/application_suite/main.rs index 7d361aa71c..948236d7d1 100644 --- a/crates/tracedecay-application/tests/application_suite/main.rs +++ b/crates/tracedecay-application/tests/application_suite/main.rs @@ -17,5 +17,6 @@ mod observability_rollup_convergence; // root: the work_rollup bench mounts it, and the suite itself uses a subset. pub mod observability_runtime_contract; mod registered_scope_route; +mod work_run_control_jobs; mod work_service_composition; mod workflow_topology_contract; diff --git a/crates/tracedecay-application/tests/application_suite/work_run_control_jobs.rs b/crates/tracedecay-application/tests/application_suite/work_run_control_jobs.rs new file mode 100644 index 0000000000..7f091f7bd1 --- /dev/null +++ b/crates/tracedecay-application/tests/application_suite/work_run_control_jobs.rs @@ -0,0 +1,253 @@ +use std::collections::BTreeSet; + +use tracedecay_application::work::RegisteredWorkApplicationServicesV1; +use tracedecay_contracts::{ + CancellationContext, CapabilityGrantSnapshot, Deadline, DisclosureClass, PauseWorkRunCommand, + RequestContext, RequestId, ResolvedScope, ResumeWorkRunCommand, WorkAttemptStoragePort, + WorkRunControlReadingV1, WorkRunControlRequestV1, +}; +use tracedecay_domain::test_fixtures::{digest, fixture_abs_root, id}; +use tracedecay_domain::{ + ActorId, AttemptId, CommitId, ConfigurationRevisionId, ConfigurationSnapshotId, ProjectId, + ProposalId, ProviderId, RepositoryId, RunId, TaskId, UtcMicros, WorkApprovalPolicy, + WorkAttemptIdentityV1, WorkAttemptProjectionBindingV1, WorkAttemptStateV1, WorkAttemptV1, + WorkAuthority, WorkCancellationStateV1, WorkEffectStateV1, WorkEgressPolicy, + WorkExecutableReference, WorkExecutionEnvelopeV1, WorkExecutionLimits, WorkExecutionSnapshot, + WorkExecutionSnapshotInput, WorkFallbackTopology, WorkFenceEpochV1, WorkFilesystemPolicy, + WorkGraphVersionV1, WorkLeaseFenceV1, WorkLeaseId, WorkProductEventSequenceV1, + WorkProductSourceWatermarkV1, WorkProviderBackendV1, WorkProviderProtocol, WorkProviderRouteId, + WorkProviderRouteV1, WorkRecoveryStateV1, WorkRunControlReasonV1, WorkRunControlStateV1, + WorkSandboxPolicy, WorkflowOperationRef, WorktreeId, +}; +use tracedecay_global_db::tests::harness::RegisteredGlobalDbTestRuntime; +use tracedecay_rusqlite_runtime::exact_sql::native_thread_reader_waits; +use tracedecay_rusqlite_runtime::native_runtime::{ + NativeTask, NativeTaskOutcome, native_thread_writer_waits, +}; +use tracedecay_tool_catalog::{CapabilityId, UseCaseId}; + +const PROJECT: &str = "project.run-control-jobs"; +const REPOSITORY: &str = "repository.run-control-jobs"; +const WORKTREE: &str = "worktree.run-control-jobs"; +const ACTOR: &str = "actor.run-control-jobs"; + +fn task() -> TaskId { + id::("task.run-control-jobs") +} + +fn run() -> RunId { + id::("run.run-control-jobs") +} + +fn context() -> RequestContext { + let scope = ResolvedScope::new( + id::(PROJECT), + id::(REPOSITORY), + id::(WORKTREE), + None, + ) + .unwrap(); + let grant = CapabilityGrantSnapshot::new( + id("grant.work.run-control-jobs"), + 1, + digest('a'), + id::("actor.issuer"), + UtcMicros(1), + UtcMicros(100_000), + scope.clone(), + BTreeSet::from([CapabilityId::new("capability.work.pause_run").unwrap()]), + BTreeSet::from([UseCaseId::new("use-case.work.pause_run").unwrap()]), + DisclosureClass::Sensitive, + ) + .unwrap(); + RequestContext::new( + id::(ACTOR), + scope, + grant, + RequestId::new("request.run-control-jobs").unwrap(), + Deadline::new(UtcMicros(90_000)).unwrap(), + CancellationContext::active("cancel.run-control-jobs").unwrap(), + ) + .unwrap() +} + +fn authority() -> WorkAuthority { + WorkAuthority::new( + id::(PROJECT), + id::(REPOSITORY), + id::(WORKTREE), + id::(ACTOR), + digest('a'), + ) + .unwrap() +} + +fn route() -> WorkProviderRouteV1 { + WorkProviderRouteV1::new( + id::("provider.work.claude-code-cli"), + id::("route.run-control-jobs.claude-code.v1"), + ) + .unwrap() +} + +fn leased_attempt() -> WorkAttemptV1 { + let identity = + WorkAttemptIdentityV1::new(task(), run(), id::("attempt.run-control-jobs")) + .unwrap(); + let binding = WorkAttemptProjectionBindingV1::new( + WorkGraphVersionV1::new(3).unwrap(), + WorkProductEventSequenceV1::new(7).unwrap(), + WorkProductSourceWatermarkV1::new(Default::default()).unwrap(), + digest('f'), + id::("proposal.run-control-jobs"), + ) + .unwrap(); + let snapshot = WorkExecutionSnapshot::new(WorkExecutionSnapshotInput { + configuration_revision_id: id::("configuration-revision.rcj.1"), + configuration_snapshot_id: id::("configuration-snapshot.rcj.1"), + effective_behavior_digest: digest('c'), + resolution_provenance_digest: digest('d'), + route: route(), + backend: WorkProviderBackendV1::ClaudeCodeCli, + protocol: WorkProviderProtocol::ClaudeStreamJson, + model: "claude-test".to_owned(), + executable: WorkExecutableReference::new( + "executable.claude.code-cli".to_owned(), + digest('e'), + ) + .unwrap(), + sandbox: WorkSandboxPolicy::Required, + approval: WorkApprovalPolicy::Never, + filesystem: WorkFilesystemPolicy::WorkspaceWrite, + egress: WorkEgressPolicy::Deny, + environment_allowlist: BTreeSet::new(), + credential_references: BTreeSet::new(), + limits: WorkExecutionLimits::new(128_000, 8_192, 16_384, 16_384, 65_536, 1).unwrap(), + deadline: UtcMicros(1_000_000), + fallback: WorkFallbackTopology::Disabled, + topology: tracedecay_domain::safe_work_topology_policy_v1(), + }) + .unwrap(); + let envelope = WorkExecutionEnvelopeV1::new( + identity.clone(), + binding.clone(), + id::("operation.run-control-jobs.execute-provider"), + snapshot, + id::(PROJECT), + id::(REPOSITORY), + id::(WORKTREE), + fixture_abs_root("/tmp/run-control-jobs"), + None, + id::("0123456789abcdef0123456789abcdef01234567"), + "Execute the admitted provider step.".to_owned(), + 1, + WorkEffectStateV1::Observational, + ) + .unwrap(); + WorkAttemptV1::new( + identity, + binding, + envelope, + WorkLeaseFenceV1::new( + id::("lease.run-control-jobs"), + WorkFenceEpochV1::new(1).unwrap(), + ) + .unwrap(), + WorkAttemptStateV1::Leased, + None, + Vec::new(), + WorkCancellationStateV1::None, + WorkRecoveryStateV1::Fresh, + route(), + None, + None, + ) + .unwrap() +} + +#[tokio::test] +async fn pause_and_resume_awaited_on_a_native_worker_park_no_native_thread() { + tracedecay_global_db::register_registered_schema_installer(); + let profile = tempfile::tempdir().expect("profile"); + let project = tempfile::tempdir().expect("project"); + let runtime = RegisteredGlobalDbTestRuntime::project( + profile.path(), + project.path(), + id::(PROJECT), + ) + .await + .expect("registered project store opens"); + let database = runtime.project_database().expect("project database"); + database + .work_storage() + .expect("Work storage") + .insert(&authority(), &leased_attempt()) + .expect("the run holds one leased attempt"); + let work = RegisteredWorkApplicationServicesV1::attach(database).expect("Work services"); + let writer_waits = native_thread_writer_waits(); + let reader_waits = native_thread_reader_waits(); + let native = NativeTask::spawn(async move { + let paused = work + .pause_run( + context(), + PauseWorkRunCommand { + task_id: task(), + run_id: run(), + reason: WorkRunControlReasonV1::OperatorRequest, + expected_authority_version: None, + occurred_at: UtcMicros(50_000), + }, + ) + .await + .map_err(|problem| format!("pause: {problem:?}"))? + .control; + let resumed = work + .resume_run( + context(), + ResumeWorkRunCommand { + task_id: task(), + run_id: run(), + reason: WorkRunControlReasonV1::OperatorRequest, + expected_authority_version: paused.authority().get(), + occurred_at: UtcMicros(60_000), + }, + ) + .await + .map_err(|problem| format!("resume: {problem:?}"))? + .control; + Ok::<_, String>((paused.state(), resumed.state())) + }) + .expect("native task spawns"); + let NativeTaskOutcome::Completed(outcome) = native.wait().await else { + panic!("the pause and resume task completes"); + }; + assert_eq!( + native_thread_writer_waits() - writer_waits, + 0, + "pause and resume parked no native worker on the writer" + ); + assert_eq!( + native_thread_reader_waits() - reader_waits, + 0, + "pause and resume parked no native worker on a reader" + ); + let (paused_state, resumed_state) = outcome.expect("pause and resume commit"); + assert_eq!(paused_state, WorkRunControlStateV1::Paused); + assert_eq!(resumed_state, WorkRunControlStateV1::Running); + let final_state = RegisteredWorkApplicationServicesV1::attach(database) + .expect("Work services") + .run_control() + .read( + &context(), + &WorkRunControlRequestV1 { + task_id: task(), + run_id: run(), + }, + ) + .expect("the run control reads"); + assert!( + matches!(&final_state, WorkRunControlReadingV1::Controlled { control, .. } + if control.state() == WorkRunControlStateV1::Running), + "the resume committed: {final_state:?}" + ); +} diff --git a/crates/tracedecay-contracts/src/lib.rs b/crates/tracedecay-contracts/src/lib.rs index 4c5ecc9d97..929724cb42 100644 --- a/crates/tracedecay-contracts/src/lib.rs +++ b/crates/tracedecay-contracts/src/lib.rs @@ -566,8 +566,8 @@ pub use work_retry::{ pub use work_run_control::{ PauseWorkRunCommand, ResumeWorkRunCommand, WorkRunAdmissionV1, WorkRunControlFrontierV1, WorkRunControlReadingV1, WorkRunControlRequestV1, WorkRunControlService, - WorkRunControlStorageError, WorkRunControlStoragePort, WorkRunControlTransitionReceiptV1, - WorkRunLiveAttemptV1, + WorkRunControlStorageError, WorkRunControlStoragePort, WorkRunControlTransitionPort, + WorkRunControlTransitionReceiptV1, WorkRunLiveAttemptV1, }; pub use work_synthesis::{ AdmitWorkSynthesisCommand, WorkSynthesisAdmissionRecordV1, WorkSynthesisAdmissionV1, diff --git a/crates/tracedecay-contracts/src/work_run_control.rs b/crates/tracedecay-contracts/src/work_run_control.rs index daa7f4a193..17b357a44c 100644 --- a/crates/tracedecay-contracts/src/work_run_control.rs +++ b/crates/tracedecay-contracts/src/work_run_control.rs @@ -104,9 +104,11 @@ pub struct WorkRunControlTransitionReceiptV1 { pub blocked_intervals: Vec, } -/// The durable run-control rows and the attempt evidence they are derived -/// from. -pub trait WorkRunControlStoragePort: Send + Sync { +/// The reads and the publication one pause or resume transition makes. +/// +/// It carries no `Send + Sync` bound so a storage scoped to one writer +/// transaction can implement it and run the whole transition atomically. +pub trait WorkRunControlTransitionPort { /// Reads all mutable evidence needed by a pause or resume from one storage /// snapshot. `None` means the run has no durable attempt. fn run_control_frontier( @@ -116,16 +118,6 @@ pub trait WorkRunControlStoragePort: Send + Sync { run_id: &RunId, ) -> Result, WorkRunControlStorageError>; - /// The admitted deadline and live attempt frontier for one run, or `None` - /// when the run holds no durable attempt at all. Retry-superseded recovery - /// records still count toward history, but are absent from the frontier. - fn run_admission( - &self, - authority: &WorkAuthority, - task_id: &TaskId, - run_id: &RunId, - ) -> Result, WorkRunControlStorageError>; - /// Resolves the canonical workflow journal binding for every active /// attempt of one run. Journal replay happens only while a pause is about /// to create interval evidence, never on ordinary reads or reservations. @@ -136,6 +128,31 @@ pub trait WorkRunControlStoragePort: Send + Sync { run_id: &RunId, ) -> Result, WorkRunControlStorageError>; + /// Publishes only while the complete mutable frontier is still identical + /// to `expected`. Implementations must perform the comparison and control + /// CAS in the same write transaction. + fn publish_run_control_at_frontier( + &self, + authority: &WorkAuthority, + expected: &WorkRunControlFrontierV1, + next: &WorkRunControlV1, + blocked_intervals: &[WorkBlockedIntervalReceiptV1], + ) -> Result<(), WorkRunControlStorageError>; +} + +/// The durable run-control rows and the attempt evidence they are derived +/// from. +pub trait WorkRunControlStoragePort: Send + Sync { + /// The admitted deadline and live attempt frontier for one run, or `None` + /// when the run holds no durable attempt at all. Retry-superseded recovery + /// records still count toward history, but are absent from the frontier. + fn run_admission( + &self, + authority: &WorkAuthority, + task_id: &TaskId, + run_id: &RunId, + ) -> Result, WorkRunControlStorageError>; + /// The published control row for one run, or `None` when the run has never /// been controlled. fn load_run_control( @@ -155,17 +172,6 @@ pub trait WorkRunControlStoragePort: Send + Sync { blocked_intervals: &[WorkBlockedIntervalReceiptV1], ) -> Result<(), WorkRunControlStorageError>; - /// Publishes only while the complete mutable frontier is still identical - /// to `expected`. Implementations must perform the comparison and control - /// CAS in the same write transaction. - fn publish_run_control_at_frontier( - &self, - authority: &WorkAuthority, - expected: &WorkRunControlFrontierV1, - next: &WorkRunControlV1, - blocked_intervals: &[WorkBlockedIntervalReceiptV1], - ) -> Result<(), WorkRunControlStorageError>; - /// The still-open receipts for one exact run. Resume closes precisely these /// rows under the same control compare-and-swap; it does not reconstruct /// an interval from a current clock or current live-attempt query. @@ -274,14 +280,16 @@ pub struct WorkRunControlService { storage: S, } -impl WorkRunControlService -where - S: WorkRunControlStoragePort, -{ +impl WorkRunControlService { pub const fn new(storage: S) -> Self { Self { storage } } +} +impl WorkRunControlService +where + S: WorkRunControlTransitionPort, +{ /// Fences new reservations for one run. pub fn pause( &self, @@ -305,7 +313,7 @@ where let frontier = self .storage .run_control_frontier(&authority, &command.task_id, &command.run_id) - .map_err(storage_problem)? + .map_err(ApplicationProblem::from)? .ok_or_else(not_found_problem)?; check_expected( frontier.control.as_ref(), @@ -317,7 +325,7 @@ where let workflow_attempts = self .storage .workflow_bound_live_attempts(&authority, &command.task_id, &command.run_id) - .map_err(storage_problem)?; + .map_err(ApplicationProblem::from)?; let workflow_steps = workflow_steps_for_live_attempts(&frontier.admission.live_attempts, workflow_attempts)?; let current = match frontier.control.clone() { @@ -360,7 +368,7 @@ where .map_err(contract_problem)?; self.storage .publish_run_control_at_frontier(&authority, &frontier, &next, &blocked_intervals) - .map_err(storage_problem)?; + .map_err(ApplicationProblem::from)?; Ok(WorkRunControlTransitionReceiptV1 { control: next, blocked_intervals, @@ -394,7 +402,7 @@ where let frontier = self .storage .run_control_frontier(&authority, &command.task_id, &command.run_id) - .map_err(storage_problem)? + .map_err(ApplicationProblem::from)? .ok_or_else(not_found_problem)?; let current = frontier.control.clone().ok_or_else(|| { // A run that was never paused has nothing to resume, and @@ -428,13 +436,18 @@ where .map_err(contract_problem)?; self.storage .publish_run_control_at_frontier(&authority, &frontier, &next, &blocked_intervals) - .map_err(storage_problem)?; + .map_err(ApplicationProblem::from)?; Ok(WorkRunControlTransitionReceiptV1 { control: next, blocked_intervals, }) } +} +impl WorkRunControlService +where + S: WorkRunControlStoragePort, +{ /// Reads the published control state for one run. #[tracing::instrument(name = "application.work.run_control.read", level = "trace", skip_all)] pub fn read( @@ -447,7 +460,7 @@ where let control = self .storage .load_run_control(&authority, &request.task_id, &request.run_id) - .map_err(storage_problem)?; + .map_err(ApplicationProblem::from)?; Ok(match control { Some(control) => WorkRunControlReadingV1::Controlled { control, @@ -482,7 +495,7 @@ where let control = self .storage .load_run_control(&authority, task_id, run_id) - .map_err(storage_problem)?; + .map_err(ApplicationProblem::from)?; match control { Some(control) if !control.admits_reservation() => Err(ApplicationProblem::conflict( "application.work-run-control.paused", @@ -511,7 +524,7 @@ where let authority = work_authority(context)?; self.storage .next_settled_blocked_intervals_for_observation(&authority, limit) - .map_err(storage_problem) + .map_err(ApplicationProblem::from) } } } @@ -533,7 +546,7 @@ where let authority = work_authority(context)?; self.storage .mark_settled_blocked_interval_durable(&authority, receipt) - .map_err(storage_problem) + .map_err(ApplicationProblem::from) } } } @@ -546,7 +559,7 @@ where ) -> Result { self.storage .run_admission(authority, task_id, run_id) - .map_err(storage_problem)? + .map_err(ApplicationProblem::from)? .ok_or_else(not_found_problem) } } @@ -572,15 +585,15 @@ fn check_expected( } } -fn storage_problem(error: WorkRunControlStorageError) -> ApplicationProblem { - match error { - WorkRunControlStorageError::NotFoundOrNotAuthorized => not_found_problem(), - WorkRunControlStorageError::AuthorityConflict => authority_conflict_problem(), - WorkRunControlStorageError::Unavailable => { - ApplicationProblem::unavailable(SafeDiagnostic { +impl From for ApplicationProblem { + fn from(error: WorkRunControlStorageError) -> Self { + match error { + WorkRunControlStorageError::NotFoundOrNotAuthorized => not_found_problem(), + WorkRunControlStorageError::AuthorityConflict => authority_conflict_problem(), + WorkRunControlStorageError::Unavailable => Self::unavailable(SafeDiagnostic { code: "application.work-run-control.storage-unavailable".to_owned(), message: "The Work run control authority is unavailable.".to_owned(), - }) + }), } } } @@ -631,15 +644,17 @@ fn workflow_steps_for_live_attempts( .insert(attempt.attempt_id.clone(), attempt) .is_some() { - return Err(storage_problem(WorkRunControlStorageError::Unavailable)); + return Err(ApplicationProblem::from( + WorkRunControlStorageError::Unavailable, + )); } } live_attempts .iter() .map(|attempt_id| { - by_attempt - .remove(attempt_id) - .ok_or_else(|| storage_problem(WorkRunControlStorageError::AuthorityConflict)) + by_attempt.remove(attempt_id).ok_or_else(|| { + ApplicationProblem::from(WorkRunControlStorageError::AuthorityConflict) + }) }) .collect() } diff --git a/crates/tracedecay-contracts/tests/contracts_suite/work_run_control_service.rs b/crates/tracedecay-contracts/tests/contracts_suite/work_run_control_service.rs index f36cffa6ea..6dd7867053 100644 --- a/crates/tracedecay-contracts/tests/contracts_suite/work_run_control_service.rs +++ b/crates/tracedecay-contracts/tests/contracts_suite/work_run_control_service.rs @@ -21,7 +21,7 @@ use tracedecay_contracts::{ DisclosureClass, PauseWorkRunCommand, RequestContext, RequestId, ResolvedScope, ResumeWorkRunCommand, WorkRunAdmissionV1, WorkRunControlFrontierV1, WorkRunControlReadingV1, WorkRunControlRequestV1, WorkRunControlService, WorkRunControlStorageError, - WorkRunControlStoragePort, WorkRunLiveAttemptV1, + WorkRunControlStoragePort, WorkRunControlTransitionPort, WorkRunLiveAttemptV1, }; use tracedecay_domain::{ ActorId, AttemptId, ProjectId, RepositoryId, RunId, TaskId, UtcMicros, WorkAuthority, @@ -128,7 +128,7 @@ impl TestStore { } } -impl WorkRunControlStoragePort for TestStore { +impl WorkRunControlTransitionPort for TestStore { fn run_control_frontier( &self, authority: &WorkAuthority, @@ -155,34 +155,6 @@ impl WorkRunControlStoragePort for TestStore { })) } - fn run_admission( - &self, - authority: &WorkAuthority, - task_id: &TaskId, - run_id: &RunId, - ) -> Result, WorkRunControlStorageError> { - Ok(self - .admissions - .lock() - .unwrap() - .get(&(authority.clone(), task_id.clone(), run_id.clone())) - .cloned()) - } - - fn load_run_control( - &self, - authority: &WorkAuthority, - task_id: &TaskId, - run_id: &RunId, - ) -> Result, WorkRunControlStorageError> { - Ok(self - .controls - .lock() - .unwrap() - .get(&(authority.clone(), task_id.clone(), run_id.clone())) - .cloned()) - } - fn workflow_bound_live_attempts( &self, authority: &WorkAuthority, @@ -221,6 +193,57 @@ impl WorkRunControlStoragePort for TestStore { Ok(attempts) } + fn publish_run_control_at_frontier( + &self, + authority: &WorkAuthority, + expected: &WorkRunControlFrontierV1, + next: &WorkRunControlV1, + blocked_intervals: &[WorkBlockedIntervalReceiptV1], + ) -> Result<(), WorkRunControlStorageError> { + let current = self + .run_control_frontier(authority, next.task_id(), next.run_id())? + .ok_or(WorkRunControlStorageError::AuthorityConflict)?; + if ¤t != expected { + return Err(WorkRunControlStorageError::AuthorityConflict); + } + self.publish_run_control( + authority, + expected.control.as_ref().map(WorkRunControlV1::authority), + next, + blocked_intervals, + ) + } +} + +impl WorkRunControlStoragePort for TestStore { + fn run_admission( + &self, + authority: &WorkAuthority, + task_id: &TaskId, + run_id: &RunId, + ) -> Result, WorkRunControlStorageError> { + Ok(self + .admissions + .lock() + .unwrap() + .get(&(authority.clone(), task_id.clone(), run_id.clone())) + .cloned()) + } + + fn load_run_control( + &self, + authority: &WorkAuthority, + task_id: &TaskId, + run_id: &RunId, + ) -> Result, WorkRunControlStorageError> { + Ok(self + .controls + .lock() + .unwrap() + .get(&(authority.clone(), task_id.clone(), run_id.clone())) + .cloned()) + } + fn publish_run_control( &self, authority: &WorkAuthority, @@ -261,27 +284,6 @@ impl WorkRunControlStoragePort for TestStore { Ok(()) } - fn publish_run_control_at_frontier( - &self, - authority: &WorkAuthority, - expected: &WorkRunControlFrontierV1, - next: &WorkRunControlV1, - blocked_intervals: &[WorkBlockedIntervalReceiptV1], - ) -> Result<(), WorkRunControlStorageError> { - let current = self - .run_control_frontier(authority, next.task_id(), next.run_id())? - .ok_or(WorkRunControlStorageError::AuthorityConflict)?; - if ¤t != expected { - return Err(WorkRunControlStorageError::AuthorityConflict); - } - self.publish_run_control( - authority, - expected.control.as_ref().map(WorkRunControlV1::authority), - next, - blocked_intervals, - ) - } - fn open_blocked_intervals( &self, authority: &WorkAuthority, diff --git a/crates/tracedecay-daemon-service/src/invocation/work/request_dispatch.rs b/crates/tracedecay-daemon-service/src/invocation/work/request_dispatch.rs index 1f898690b8..087acc1450 100644 --- a/crates/tracedecay-daemon-service/src/invocation/work/request_dispatch.rs +++ b/crates/tracedecay-daemon-service/src/invocation/work/request_dispatch.rs @@ -879,64 +879,64 @@ pub(super) async fn dispatch_work_application( } } WorkApplicationInvocationV1::PauseRun(command) => { - let _span = tracing::trace_span!("daemon.service.work.pause_run").entered(); - { - let transition = services.run_control().pause_with_receipt(&context, command); - let open_receipts = transition - .as_ref() - .map_or_default(|transition| transition.blocked_intervals.clone()); - let response = complete_work_effect( - ®istered, - request_id, - &context, - canonical_request_id, - operation_key, - use_case, - input_digest, - transition.map(|transition| transition.control), - observed_at, - deadline, - WorkApplicationOutcomeV1::PauseRun, - ); - offer_work_blocked_interval_receipts( - ®istered.durable_write_signal, - observability_producer.as_deref(), - context.scope().project_id.as_str(), - &open_receipts, - ); - response - } + let transition = tracing::Instrument::instrument( + services.pause_run(context.clone(), command), + tracing::trace_span!("daemon.service.work.pause_run"), + ) + .await; + let open_receipts = transition + .as_ref() + .map_or_default(|transition| transition.blocked_intervals.clone()); + let response = complete_work_effect( + ®istered, + request_id, + &context, + canonical_request_id, + operation_key, + use_case, + input_digest, + transition.map(|transition| transition.control), + observed_at, + deadline, + WorkApplicationOutcomeV1::PauseRun, + ); + offer_work_blocked_interval_receipts( + ®istered.durable_write_signal, + observability_producer.as_deref(), + context.scope().project_id.as_str(), + &open_receipts, + ); + response } WorkApplicationInvocationV1::ResumeRun(command) => { - let _span = tracing::trace_span!("daemon.service.work.resume_run").entered(); - { - let transition = services - .run_control() - .resume_with_receipt(&context, command); - let settled_receipts = transition - .as_ref() - .map_or_default(|transition| transition.blocked_intervals.clone()); - let response = complete_work_effect( - ®istered, - request_id, - &context, - canonical_request_id, - operation_key, - use_case, - input_digest, - transition.map(|transition| transition.control), - observed_at, - deadline, - WorkApplicationOutcomeV1::ResumeRun, - ); - offer_work_blocked_interval_receipts( - ®istered.durable_write_signal, - observability_producer.as_deref(), - context.scope().project_id.as_str(), - &settled_receipts, - ); - response - } + let transition = tracing::Instrument::instrument( + services.resume_run(context.clone(), command), + tracing::trace_span!("daemon.service.work.resume_run"), + ) + .await; + let settled_receipts = transition + .as_ref() + .map_or_default(|transition| transition.blocked_intervals.clone()); + let response = complete_work_effect( + ®istered, + request_id, + &context, + canonical_request_id, + operation_key, + use_case, + input_digest, + transition.map(|transition| transition.control), + observed_at, + deadline, + WorkApplicationOutcomeV1::ResumeRun, + ); + offer_work_blocked_interval_receipts( + ®istered.durable_write_signal, + observability_producer.as_deref(), + context.scope().project_id.as_str(), + &settled_receipts, + ); + response } WorkApplicationInvocationV1::RunControl(request) => { let _span = tracing::trace_span!("daemon.service.work.run_control").entered(); diff --git a/crates/tracedecay-rusqlite-runtime/src/work.rs b/crates/tracedecay-rusqlite-runtime/src/work.rs index 6ee91ec990..ad598bee12 100644 --- a/crates/tracedecay-rusqlite-runtime/src/work.rs +++ b/crates/tracedecay-rusqlite-runtime/src/work.rs @@ -44,8 +44,4 @@ impl WorkSqliteStorage { pub(crate) fn handle(&self) -> &ExactSqlHandle { self.retained.handle() } - - pub(crate) fn retained_exact_sql(&self) -> RetainedExactSqlCapability { - self.retained.clone() - } } diff --git a/crates/tracedecay-rusqlite-runtime/src/work/sql.rs b/crates/tracedecay-rusqlite-runtime/src/work/sql.rs index b64d9c0518..81a90e3caf 100644 --- a/crates/tracedecay-rusqlite-runtime/src/work/sql.rs +++ b/crates/tracedecay-rusqlite-runtime/src/work/sql.rs @@ -72,6 +72,42 @@ impl RegisteredWorkQuery for ExactSqlTransaction { } } +impl RegisteredWorkQuery for crate::exact_sql::ExactSqlJobTransaction<'_> { + fn work_query( + &self, + statement: ExactSqlStatement, + ) -> Result { + self.query(statement) + } +} + +/// A write transaction a Work storage body runs its statements in: an +/// interactive transaction or a writer job. +pub(crate) trait RegisteredWorkWrite: RegisteredWorkQuery { + fn work_execute( + &self, + statement: ExactSqlStatement, + ) -> Result; +} + +impl RegisteredWorkWrite for ExactSqlTransaction { + fn work_execute( + &self, + statement: ExactSqlStatement, + ) -> Result { + self.execute(statement) + } +} + +impl RegisteredWorkWrite for crate::exact_sql::ExactSqlJobTransaction<'_> { + fn work_execute( + &self, + statement: ExactSqlStatement, + ) -> Result { + self.execute(statement) + } +} + pub(crate) fn registered_work_query( source: &impl RegisteredWorkQuery, sql: &str, diff --git a/crates/tracedecay-rusqlite-runtime/src/work_run_control.rs b/crates/tracedecay-rusqlite-runtime/src/work_run_control.rs index 5ac74391f4..840771ef8c 100644 --- a/crates/tracedecay-rusqlite-runtime/src/work_run_control.rs +++ b/crates/tracedecay-rusqlite-runtime/src/work_run_control.rs @@ -11,8 +11,8 @@ use tracedecay_contracts::{ WorkAttemptStorageError, WorkRunAdmissionV1, WorkRunControlFrontierV1, - WorkRunControlStorageError, WorkRunControlStoragePort, WorkRunLiveAttemptV1, - WorkflowRunStorageError, WorkflowRunStoragePort, + WorkRunControlStorageError, WorkRunControlStoragePort, WorkRunControlTransitionPort, + WorkRunLiveAttemptV1, WorkflowRunStorageError, }; use tracedecay_domain::{ RunId, TaskId, UtcMicros, WorkAttemptV1, WorkAuthority, WorkBlockedIntervalClosureV1, @@ -20,14 +20,74 @@ use tracedecay_domain::{ WorkRunControlV1, }; -use crate::exact_sql::{ExactSqlTransaction, ExactSqlValue}; +use crate::exact_sql::{ExactSqlError, ExactSqlJobTransaction, ExactSqlTransaction, ExactSqlValue}; use crate::work::{ - ACTIVE_ATTEMPT_PREDICATE, RegisteredWorkQuery, WorkSqliteStorage, authority_params_owned, - exact_sql_integer, exact_sql_statement, exact_sql_text, registered_work_query, + ACTIVE_ATTEMPT_PREDICATE, RegisteredWorkQuery, RegisteredWorkWrite, WorkSqliteStorage, + authority_params_owned, exact_sql_integer, exact_sql_statement, exact_sql_text, + registered_work_query, }; -use crate::workflow::WorkflowSqliteAuthority; -impl WorkRunControlStoragePort for WorkSqliteStorage { +/// Work storage scoped to one writer job. It holds the job's transaction and +/// no handle, so none of its statements can wait on the writer thread that is +/// running the job. +#[derive(Clone, Copy)] +pub struct WorkSqliteJobStorage<'job> { + transaction: &'job ExactSqlJobTransaction<'job>, +} + +impl WorkSqliteStorage { + /// Runs `job` over Work storage scoped to one writer transaction: it + /// commits when the job returns `Ok` and rolls back when it returns `Err`, + /// and the awaiting task parks no thread. + pub async fn transact(&self, job: F) -> Result, ExactSqlError> + where + F: Fn(WorkSqliteJobStorage<'_>) -> Result + Send + Sync + 'static, + R: Send + 'static, + E: Send + 'static, + { + self.handle() + .transact(move |transaction| job(WorkSqliteJobStorage { transaction })) + .await + } +} + +impl WorkRunControlTransitionPort for WorkSqliteJobStorage<'_> { + fn run_control_frontier( + &self, + authority: &WorkAuthority, + task_id: &TaskId, + run_id: &RunId, + ) -> Result, WorkRunControlStorageError> { + run_control_frontier_from(self.transaction, authority, task_id, run_id) + } + + fn workflow_bound_live_attempts( + &self, + authority: &WorkAuthority, + task_id: &TaskId, + run_id: &RunId, + ) -> Result, WorkRunControlStorageError> { + workflow_bound_live_attempts_from(self.transaction, authority, task_id, run_id) + } + + fn publish_run_control_at_frontier( + &self, + authority: &WorkAuthority, + expected: &WorkRunControlFrontierV1, + next: &WorkRunControlV1, + blocked_intervals: &[WorkBlockedIntervalReceiptV1], + ) -> Result<(), WorkRunControlStorageError> { + publish_run_control_at_frontier_tx( + self.transaction, + authority, + expected, + next, + blocked_intervals, + ) + } +} + +impl WorkRunControlTransitionPort for WorkSqliteStorage { fn run_control_frontier( &self, authority: &WorkAuthority, @@ -48,125 +108,31 @@ impl WorkRunControlStoragePort for WorkSqliteStorage { } } - fn run_admission( - &self, - authority: &WorkAuthority, - task_id: &TaskId, - run_id: &RunId, - ) -> Result, WorkRunControlStorageError> { - run_admission_from(self.handle(), authority, task_id, run_id) - } - fn workflow_bound_live_attempts( &self, authority: &WorkAuthority, task_id: &TaskId, run_id: &RunId, ) -> Result, WorkRunControlStorageError> { - let sql = format!( - "SELECT attempt.attempt_payload FROM work_attempts_v1 AS attempt - WHERE attempt.project_id = ?1 AND attempt.repository_id = ?2 - AND attempt.worktree_id = ?3 AND attempt.actor_id = ?4 - AND attempt.policy_digest = ?5 AND attempt.task_id = ?6 - AND attempt.run_id = ?7 AND {ACTIVE_ATTEMPT_PREDICATE} - ORDER BY attempt.rowid" - ); - let rows = registered_work_query( - self.handle(), - &sql, - authority_params_owned(authority) - .into_iter() - .chain(run_params(task_id, run_id)) - .collect(), - ) - .map_err(|_| WorkRunControlStorageError::Unavailable)?; - let attempts = rows - .rows - .iter() - .map(|row| { - let payload = exact_sql_text(&row.values, 0) - .ok_or(WorkRunControlStorageError::Unavailable)?; - attempt_from_payload(payload) - }) - .collect::, _>>()?; - let workflow = WorkflowSqliteAuthority::from_retained_exact_sql(self.retained_exact_sql()) - .map_err(|_| WorkRunControlStorageError::Unavailable)?; - let projection = match WorkflowRunStoragePort::projection(&workflow, run_id) { - Ok(projection) if projection.run_id() == run_id => Some(projection), - Ok(_) => return Err(WorkRunControlStorageError::Unavailable), - Err(WorkflowRunStorageError::NotFound) => None, - Err( - WorkflowRunStorageError::VersionConflict - | WorkflowRunStorageError::IdempotencyConflict - | WorkflowRunStorageError::InvalidHistory - | WorkflowRunStorageError::Unavailable, - ) => return Err(WorkRunControlStorageError::Unavailable), - }; - attempts - .into_iter() - .filter(|attempt| { - projection.as_ref().is_none_or(|projection| { - projection - .planned_fan_out_attempt(attempt.identity()) - .is_none_or(|planned| { - projection.active_fan_out_attempt(planned) == attempt.identity() - }) - }) - }) - .map(|attempt| { - let step_id = match projection.as_ref() { - None => None, - Some(projection) => { - let planned_attempt = - projection.planned_fan_out_attempt(attempt.identity()); - let mut matching_steps = projection - .fan_out_plans() - .values() - .filter(|plan| { - plan.children - .iter() - .any(|child| Some(&child.attempt_identity) == planned_attempt) - }) - .map(|plan| plan.step_id.clone()); - let step = matching_steps.next(); - if matching_steps.next().is_some() { - return Err(WorkRunControlStorageError::Unavailable); - } - step - } - }; - Ok(WorkRunLiveAttemptV1 { - attempt_id: attempt.identity().attempt_id().clone(), - step_id, - }) - }) - .collect() + workflow_bound_live_attempts_from(self.handle(), authority, task_id, run_id) } - fn load_run_control( - &self, - authority: &WorkAuthority, - task_id: &TaskId, - run_id: &RunId, - ) -> Result, WorkRunControlStorageError> { - load_run_control_from(self.handle(), authority, task_id, run_id) - } - - fn publish_run_control( + fn publish_run_control_at_frontier( &self, authority: &WorkAuthority, - expected: Option, + expected: &WorkRunControlFrontierV1, next: &WorkRunControlV1, blocked_intervals: &[WorkBlockedIntervalReceiptV1], ) -> Result<(), WorkRunControlStorageError> { { - let _span = tracing::trace_span!("rusqlite.work_run_control.txn.publish").entered(); + let _span = + tracing::trace_span!("rusqlite.work_run_control.txn.publish_at_frontier").entered(); { let transaction = self .handle() .begin_immediate() .map_err(|_| WorkRunControlStorageError::Unavailable)?; - if let Err(error) = publish_run_control_tx( + if let Err(error) = publish_run_control_at_frontier_tx( &transaction, authority, expected, @@ -183,42 +149,45 @@ impl WorkRunControlStoragePort for WorkSqliteStorage { } } } +} - fn publish_run_control_at_frontier( +impl WorkRunControlStoragePort for WorkSqliteStorage { + fn run_admission( &self, authority: &WorkAuthority, - expected: &WorkRunControlFrontierV1, + task_id: &TaskId, + run_id: &RunId, + ) -> Result, WorkRunControlStorageError> { + run_admission_from(self.handle(), authority, task_id, run_id) + } + + fn load_run_control( + &self, + authority: &WorkAuthority, + task_id: &TaskId, + run_id: &RunId, + ) -> Result, WorkRunControlStorageError> { + load_run_control_from(self.handle(), authority, task_id, run_id) + } + + fn publish_run_control( + &self, + authority: &WorkAuthority, + expected: Option, next: &WorkRunControlV1, blocked_intervals: &[WorkBlockedIntervalReceiptV1], ) -> Result<(), WorkRunControlStorageError> { { - let _span = - tracing::trace_span!("rusqlite.work_run_control.txn.publish_at_frontier").entered(); + let _span = tracing::trace_span!("rusqlite.work_run_control.txn.publish").entered(); { let transaction = self .handle() .begin_immediate() .map_err(|_| WorkRunControlStorageError::Unavailable)?; - let current = { - let _span = - tracing::trace_span!("rusqlite.work_run_control.cas.frontier").entered(); - { - run_control_frontier_from( - &transaction, - authority, - next.task_id(), - next.run_id(), - ) - } - }?; - if current.as_ref() != Some(expected) { - let _ = transaction.rollback(); - return Err(WorkRunControlStorageError::AuthorityConflict); - } if let Err(error) = publish_run_control_tx( &transaction, authority, - expected.control.as_ref().map(WorkRunControlV1::authority), + expected, next, blocked_intervals, ) { @@ -348,6 +317,114 @@ impl WorkRunControlStoragePort for WorkSqliteStorage { } } +fn workflow_bound_live_attempts_from( + source: &impl RegisteredWorkQuery, + authority: &WorkAuthority, + task_id: &TaskId, + run_id: &RunId, +) -> Result, WorkRunControlStorageError> { + let sql = format!( + "SELECT attempt.attempt_payload FROM work_attempts_v1 AS attempt + WHERE attempt.project_id = ?1 AND attempt.repository_id = ?2 + AND attempt.worktree_id = ?3 AND attempt.actor_id = ?4 + AND attempt.policy_digest = ?5 AND attempt.task_id = ?6 + AND attempt.run_id = ?7 AND {ACTIVE_ATTEMPT_PREDICATE} + ORDER BY attempt.rowid" + ); + let rows = registered_work_query( + source, + &sql, + authority_params_owned(authority) + .into_iter() + .chain(run_params(task_id, run_id)) + .collect(), + ) + .map_err(|_| WorkRunControlStorageError::Unavailable)?; + let attempts = rows + .rows + .iter() + .map(|row| { + let payload = + exact_sql_text(&row.values, 0).ok_or(WorkRunControlStorageError::Unavailable)?; + attempt_from_payload(payload) + }) + .collect::, _>>()?; + crate::workflow::require_workflow_schema(source) + .map_err(|_| WorkRunControlStorageError::Unavailable)?; + let projection = match crate::workflow::run_journal::projection_from(source, run_id) { + Ok(projection) if projection.run_id() == run_id => Some(projection), + Ok(_) => return Err(WorkRunControlStorageError::Unavailable), + Err(WorkflowRunStorageError::NotFound) => None, + Err( + WorkflowRunStorageError::VersionConflict + | WorkflowRunStorageError::IdempotencyConflict + | WorkflowRunStorageError::InvalidHistory + | WorkflowRunStorageError::Unavailable, + ) => return Err(WorkRunControlStorageError::Unavailable), + }; + attempts + .into_iter() + .filter(|attempt| { + projection.as_ref().is_none_or(|projection| { + projection + .planned_fan_out_attempt(attempt.identity()) + .is_none_or(|planned| { + projection.active_fan_out_attempt(planned) == attempt.identity() + }) + }) + }) + .map(|attempt| { + let step_id = match projection.as_ref() { + None => None, + Some(projection) => { + let planned_attempt = projection.planned_fan_out_attempt(attempt.identity()); + let mut matching_steps = projection + .fan_out_plans() + .values() + .filter(|plan| { + plan.children + .iter() + .any(|child| Some(&child.attempt_identity) == planned_attempt) + }) + .map(|plan| plan.step_id.clone()); + let step = matching_steps.next(); + if matching_steps.next().is_some() { + return Err(WorkRunControlStorageError::Unavailable); + } + step + } + }; + Ok(WorkRunLiveAttemptV1 { + attempt_id: attempt.identity().attempt_id().clone(), + step_id, + }) + }) + .collect() +} + +fn publish_run_control_at_frontier_tx( + transaction: &impl RegisteredWorkWrite, + authority: &WorkAuthority, + expected: &WorkRunControlFrontierV1, + next: &WorkRunControlV1, + blocked_intervals: &[WorkBlockedIntervalReceiptV1], +) -> Result<(), WorkRunControlStorageError> { + let current = { + let _span = tracing::trace_span!("rusqlite.work_run_control.cas.frontier").entered(); + run_control_frontier_from(transaction, authority, next.task_id(), next.run_id()) + }?; + if current.as_ref() != Some(expected) { + return Err(WorkRunControlStorageError::AuthorityConflict); + } + publish_run_control_tx( + transaction, + authority, + expected.control.as_ref().map(WorkRunControlV1::authority), + next, + blocked_intervals, + ) +} + fn run_control_frontier_from( source: &impl RegisteredWorkQuery, authority: &WorkAuthority, @@ -486,7 +563,7 @@ fn open_blocked_intervals_from( } fn publish_run_control_tx( - transaction: &ExactSqlTransaction, + transaction: &impl RegisteredWorkWrite, authority: &WorkAuthority, expected: Option, next: &WorkRunControlV1, @@ -501,7 +578,7 @@ fn publish_run_control_tx( { match expected { None => transaction - .execute( + .work_execute( exact_sql_statement( "INSERT OR IGNORE INTO work_run_controls_v1 ( project_id, repository_id, worktree_id, actor_id, policy_digest, @@ -524,7 +601,7 @@ fn publish_run_control_tx( let expected_version = i64::try_from(expected.get()) .map_err(|_| WorkRunControlStorageError::Unavailable)?; transaction - .execute( + .work_execute( exact_sql_statement( "UPDATE work_run_controls_v1 SET state = ?8, authority_version = ?9, control_payload = ?10 @@ -710,7 +787,7 @@ impl BlockedIntervalObservationCursor { } fn persist_blocked_intervals( - transaction: &ExactSqlTransaction, + transaction: &impl RegisteredWorkWrite, authority: &WorkAuthority, task_id: &TaskId, run_id: &RunId, @@ -736,7 +813,7 @@ fn persist_blocked_intervals( .checked_sub(1) .ok_or(WorkRunControlStorageError::AuthorityConflict)?; transaction - .execute( + .work_execute( exact_sql_statement( "UPDATE work_blocked_intervals_v1 SET interval_revision = ?12, settled = 1, observability_durable = 0, @@ -768,7 +845,7 @@ fn persist_blocked_intervals( return Err(WorkRunControlStorageError::AuthorityConflict); } transaction - .execute( + .work_execute( exact_sql_statement( "INSERT INTO work_blocked_intervals_v1 ( project_id, repository_id, worktree_id, actor_id, policy_digest, diff --git a/crates/tracedecay-rusqlite-runtime/src/workflow.rs b/crates/tracedecay-rusqlite-runtime/src/workflow.rs index b14969a431..514f79503b 100644 --- a/crates/tracedecay-rusqlite-runtime/src/workflow.rs +++ b/crates/tracedecay-rusqlite-runtime/src/workflow.rs @@ -20,6 +20,7 @@ use crate::exact_sql::{ ExactSqlValue, integer_column as sql_integer, text_column as sql_text, }; use crate::repository::RetainedExactSqlCapability; +use crate::work::RegisteredWorkQuery; mod census; mod disposition; mod effect_holder; @@ -277,8 +278,8 @@ pub enum WorkflowSqliteAuthorityBuildError { Unavailable, } -fn require_workflow_schema( - handle: &ExactSqlHandle, +pub(crate) fn require_workflow_schema( + source: &impl RegisteredWorkQuery, ) -> Result<(), WorkflowSqliteAuthorityBuildError> { let table_parameters = WORKFLOW_TABLE_CONTRACTS_V1 .iter() @@ -286,8 +287,8 @@ fn require_workflow_schema( .map(|(index, _)| format!("?{}", index + 1)) .collect::>() .join(", "); - let rows = handle - .query( + let rows = source + .work_query( ExactSqlStatement::new( format!( "SELECT name, sql FROM sqlite_master @@ -301,7 +302,6 @@ fn require_workflow_schema( .collect(), ) .map_err(|_| WorkflowSqliteAuthorityBuildError::Unavailable)?, - Duration::from_secs(5), ) .map_err(|_| WorkflowSqliteAuthorityBuildError::Unavailable)?; let actual = rows @@ -323,8 +323,8 @@ fn require_workflow_schema( { return Err(WorkflowSqliteAuthorityBuildError::ResetRequired); } - let schema = handle - .query( + let schema = source + .work_query( ExactSqlStatement::new( "SELECT singleton, schema_version, definition_digest FROM workflow_schema ORDER BY singleton" @@ -332,7 +332,6 @@ fn require_workflow_schema( Vec::new(), ) .map_err(|_| WorkflowSqliteAuthorityBuildError::Unavailable)?, - Duration::from_secs(5), ) .map_err(|_| WorkflowSqliteAuthorityBuildError::Unavailable)?; let valid_schema = schema.rows.len() == 1 @@ -352,20 +351,19 @@ fn require_workflow_schema( return Err(WorkflowSqliteAuthorityBuildError::ResetRequired); } for table in WORKFLOW_TABLE_CONTRACTS_V1 { - require_columns(handle, table)?; + require_columns(source, table)?; } Ok(()) } fn require_columns( - handle: &ExactSqlHandle, + source: &impl RegisteredWorkQuery, table: &WorkflowTableContractV1, ) -> Result<(), WorkflowSqliteAuthorityBuildError> { - let columns = handle - .query( + let columns = source + .work_query( ExactSqlStatement::new(format!("PRAGMA table_info({})", table.name), Vec::new()) .map_err(|_| WorkflowSqliteAuthorityBuildError::Unavailable)?, - Duration::from_secs(5), ) .map_err(|_| WorkflowSqliteAuthorityBuildError::Unavailable)?; let exact = columns.rows.len() == table.columns.len() diff --git a/crates/tracedecay-rusqlite-runtime/src/workflow/run_journal.rs b/crates/tracedecay-rusqlite-runtime/src/workflow/run_journal.rs index 3180e04e55..bb1ab8162d 100644 --- a/crates/tracedecay-rusqlite-runtime/src/workflow/run_journal.rs +++ b/crates/tracedecay-rusqlite-runtime/src/workflow/run_journal.rs @@ -21,6 +21,7 @@ use super::{ ExactSqlTransaction, ExactSqlValue, WorkflowSqliteAuthority, encode_json, execute_tx, query_tx, sql_text, }; +use crate::work::{RegisteredWorkQuery, registered_work_query}; fn run_journal_unavailable(_: E) -> WorkflowRunStorageError { WorkflowRunStorageError::Unavailable @@ -35,11 +36,11 @@ fn decode_event( } fn history_tx( - transaction: &ExactSqlTransaction, + source: &impl RegisteredWorkQuery, run_id: &RunId, ) -> Result, WorkflowRunStorageError> { - let rows = query_tx( - transaction, + let rows = registered_work_query( + source, "SELECT event_payload, event_digest FROM workflow_run_journal WHERE run_id = ?1 ORDER BY sequence", vec![ExactSqlValue::Text(run_id.as_str().to_owned())], @@ -60,6 +61,19 @@ fn rebuild(history: &[WorkflowRunEvent]) -> Result Result { + let history = history_tx(source, run_id)?; + if history.is_empty() { + return Err(WorkflowRunStorageError::NotFound); + } + rebuild(&history) +} + pub(crate) fn append_in_transaction( transaction: &ExactSqlTransaction, request: &WorkflowRunAppendRequest, @@ -108,12 +122,9 @@ impl WorkflowRunStoragePort for WorkflowSqliteAuthority { .handle() .begin_deferred() .map_err(run_journal_unavailable)?; - let history = history_tx(&transaction, run_id)?; + let projection = projection_from(&transaction, run_id); let _ = transaction.rollback(); - if history.is_empty() { - return Err(WorkflowRunStorageError::NotFound); - } - rebuild(&history) + projection } fn append( diff --git a/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/work_attempt_storage.rs b/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/work_attempt_storage.rs index 6131d0bbb4..b219144546 100644 --- a/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/work_attempt_storage.rs +++ b/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/work_attempt_storage.rs @@ -21,9 +21,10 @@ use tracedecay_contracts::{ WorkAttemptStoragePort, WorkOwnerObservationMarkOutcomeV1, WorkOwnerObservationStoragePortV1, WorkRetryAttemptOutcomeV1, WorkRetryCauseV1, WorkRetryFailureSelectorV1, WorkRetryReceiptV1, WorkRetrySourceV1, WorkRetryStoragePortV1, WorkRetryWriteV1, WorkRunControlStoragePort, - WorkSynthesisAdmissionRecordV1, WorkSynthesisAdmissionStoragePort, WorkSynthesisAdmissionV1, - WorkSynthesisEvidenceGroupV1, WorkSynthesisInsertOutcome, WorkSynthesisSourceEnvelopeV1, - WorkSynthesisSourceOutcomeV1, WorkSynthesisSourceSetV1, WorkflowSynthesisDraft, + WorkRunControlTransitionPort, WorkSynthesisAdmissionRecordV1, + WorkSynthesisAdmissionStoragePort, WorkSynthesisAdmissionV1, WorkSynthesisEvidenceGroupV1, + WorkSynthesisInsertOutcome, WorkSynthesisSourceEnvelopeV1, WorkSynthesisSourceOutcomeV1, + WorkSynthesisSourceSetV1, WorkflowSynthesisDraft, }; use tracedecay_domain::configuration::TopologyConcurrencyPolicyV1; use tracedecay_domain::{ diff --git a/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/work_run_control_storage.rs b/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/work_run_control_storage.rs index 6d4ba5f425..31655ae898 100644 --- a/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/work_run_control_storage.rs +++ b/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/work_run_control_storage.rs @@ -16,7 +16,7 @@ use std::collections::BTreeSet; use tracedecay_contracts::{ WorkAttemptStorageError, WorkAttemptStoragePort, WorkRunControlStorageError, - WorkRunControlStoragePort, + WorkRunControlStoragePort, WorkRunControlTransitionPort, }; use tracedecay_domain::{ ActorId, AttemptId, CommitId, ConfigurationRevisionId, ConfigurationSnapshotId, ProjectId, @@ -397,6 +397,73 @@ fn publication_refuses_an_attempt_frontier_changed_after_snapshot() { ); } +#[test] +fn a_run_control_job_publishes_with_its_frontier_read_or_not_at_all() { + use tracedecay_rusqlite_runtime::native_runtime::block_on; + + let store = RegisteredWorkStore::start_with_setup("run-control-job", |connection| { + install_workflow_schema(connection).unwrap(); + }); + let authority = authority("actor.run-control.job"); + store + .storage() + .insert(&authority, &attempt("attempt.job.1")) + .unwrap(); + let pause_in_job = |verdict: Result<(), &'static str>| { + let authority = authority.clone(); + block_on(store.storage().transact(move |storage| { + let frontier = storage + .run_control_frontier(&authority, &task(), &run()) + .map_err(|error| format!("{error:?}"))? + .ok_or("the run has an attempt")?; + let bound = storage + .workflow_bound_live_attempts(&authority, &task(), &run()) + .map_err(|error| format!("{error:?}"))?; + let control = paused( + WorkRunControlReasonV1::OperatorRequest, + UtcMicros(400), + bound + .into_iter() + .map(|attempt| attempt.attempt_id) + .collect(), + ); + storage + .publish_run_control_at_frontier(&authority, &frontier, &control, &[]) + .map_err(|error| format!("{error:?}"))?; + verdict.map(|()| control).map_err(str::to_owned) + })) + .expect("native runtime") + .expect("the writer runs the job") + }; + + assert_eq!( + pause_in_job(Err("refused after publishing")), + Err("refused after publishing".to_owned()) + ); + assert_eq!( + store + .storage() + .load_run_control(&authority, &task(), &run()) + .unwrap(), + None, + "a refused job leaves no control row" + ); + + let control = pause_in_job(Ok(())).expect("the accepted job commits"); + assert_eq!( + control.fenced_attempts(), + [id::("attempt.job.1")] + ); + assert_eq!( + store + .storage() + .load_run_control(&authority, &task(), &run()) + .unwrap(), + Some(control), + "an accepted job commits its control row" + ); +} + #[test] fn a_later_lexical_attempt_cannot_replace_the_run_admission() { let store = RegisteredWorkStore::start("run-control-first-admission");