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");