From 773dd40782e95be696c5142e0ba4a7e107640fcc Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 16 Oct 2026 15:48:26 +0000 Subject: [PATCH 1/3] refactor(work): record duplicate adjudications as one writer job AdjudicateDuplicate now runs the contracts duplicate service inside one ExactSqlHandle::transact job over WorkSqliteJobStorage. The command replay check, the attempt checks, the revision compare and the append commit together, and the awaiting task parks no native thread. The append moves to a WorkDuplicateAdjudicationRecordPortV1 without a Send + Sync bound, implemented only by the job storage; the handle-based append had no other production caller and is deleted. The current Work and topology generations the command must cite are still read from the product graph before the job, on a reader, and passed in owned, so the graph read never holds the writer. --- .../src/work/registered.rs | 29 ++++ .../tests/application_suite/main.rs | 1 + .../tests/application_suite/work_leaf_jobs.rs | 151 ++++++++++++++++++ .../work_run_control_jobs.rs | 21 ++- crates/tracedecay-contracts/src/lib.rs | 10 +- .../src/work_duplicate_adjudication.rs | 71 ++++---- .../src/invocation/work/request_dispatch.rs | 101 ++++++------ .../src/work/duplicate_adjudication.rs | 47 +++--- .../src/work_run_control.rs | 2 +- .../work_duplicate_adjudication_storage.rs | 149 ++++++++--------- 10 files changed, 378 insertions(+), 204 deletions(-) create mode 100644 crates/tracedecay-application/tests/application_suite/work_leaf_jobs.rs diff --git a/crates/tracedecay-application/src/work/registered.rs b/crates/tracedecay-application/src/work/registered.rs index 70ba77dc34..a579c16f1f 100644 --- a/crates/tracedecay-application/src/work/registered.rs +++ b/crates/tracedecay-application/src/work/registered.rs @@ -283,6 +283,35 @@ impl RegisteredWorkApplicationServicesV1 { .unwrap_or_else(|_| Err(run_control_unavailable())) } + /// Records one duplicate adjudication as a single writer transaction: the + /// replay check, the revision compare and the append commit together. + /// `current_evidence` is read before the job, outside the writer. + pub async fn adjudicate_duplicate( + &self, + context: tracedecay_contracts::RequestContext, + command: tracedecay_domain::WorkDuplicateAdjudicationCommandV1, + current_evidence: tracedecay_domain::WorkDuplicateAdjudicationEvidenceV1, + ) -> Result< + tracedecay_contracts::WorkDuplicateAdjudicationAppendOutcomeV1, + tracedecay_contracts::ApplicationProblem, + > { + self.storage + .transact(move |storage| { + tracedecay_contracts::WorkDuplicateAdjudicationServiceV1::new(storage).adjudicate( + &context, + command.clone(), + current_evidence.clone(), + ) + }) + .await + .unwrap_or_else(|_| { + Err( + tracedecay_contracts::WorkDuplicateAdjudicationStorageErrorV1::Unavailable + .into(), + ) + }) + } + pub fn attempts( &self, ) -> &tracedecay_contracts::WorkAttemptService< diff --git a/crates/tracedecay-application/tests/application_suite/main.rs b/crates/tracedecay-application/tests/application_suite/main.rs index 948236d7d1..24b64fcf06 100644 --- a/crates/tracedecay-application/tests/application_suite/main.rs +++ b/crates/tracedecay-application/tests/application_suite/main.rs @@ -17,6 +17,7 @@ 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_leaf_jobs; mod work_run_control_jobs; mod work_service_composition; mod workflow_topology_contract; diff --git a/crates/tracedecay-application/tests/application_suite/work_leaf_jobs.rs b/crates/tracedecay-application/tests/application_suite/work_leaf_jobs.rs new file mode 100644 index 0000000000..3766cf0943 --- /dev/null +++ b/crates/tracedecay-application/tests/application_suite/work_leaf_jobs.rs @@ -0,0 +1,151 @@ +use tracedecay_application::work::RegisteredWorkApplicationServicesV1; +use tracedecay_contracts::{ + WorkAttemptStoragePort, WorkDuplicateAdjudicationAppendOutcomeV1, + WorkDuplicateAdjudicationPortV1, WorkDuplicateAdjudicationStorageErrorV1, +}; +use tracedecay_domain::test_fixtures::id; +use tracedecay_domain::{ + CoverageStateV1, DuplicateEffectOutcomeV1, DuplicateEffortKindV1, ProjectId, + ProjectionGenerationId, QuantityEvidenceClassV1, RunId, UtcMicros, WorkAttemptIdentityV1, + WorkCommandId, WorkDuplicateAdjudicationCommandV1, WorkDuplicateAdjudicationEvidenceV1, + WorkDuplicateAdjudicationQuantitiesV1, WorkTopologyGenerationRefV1, +}; +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 crate::work_run_control_jobs::{authority, context, leased_attempt_of}; + +const PROJECT: &str = "project.run-control-jobs"; + +async fn project_runtime( + profile: &std::path::Path, + project: &std::path::Path, +) -> RegisteredGlobalDbTestRuntime { + tracedecay_global_db::register_registered_schema_installer(); + RegisteredGlobalDbTestRuntime::project(profile, project, id::(PROJECT)) + .await + .expect("registered project store opens") +} + +/// Runs `job` on a native runtime worker and returns its output with the +/// writer and reader waits it parked that worker on. +async fn on_native_worker( + job: impl Future + Send + 'static, +) -> (T, u64, u64) { + let writer_waits = native_thread_writer_waits(); + let reader_waits = native_thread_reader_waits(); + let native = NativeTask::spawn(job).expect("native task spawns"); + let NativeTaskOutcome::Completed(output) = native.wait().await else { + panic!("the native task completes"); + }; + ( + output, + native_thread_writer_waits() - writer_waits, + native_thread_reader_waits() - reader_waits, + ) +} + +fn evidence() -> WorkDuplicateAdjudicationEvidenceV1 { + WorkDuplicateAdjudicationEvidenceV1 { + work_generation: id::("generation.work.leaf-jobs"), + topology_generation: WorkTopologyGenerationRefV1::new(format!("sha256:{}", "1".repeat(64))) + .unwrap(), + } +} + +fn duplicate_command( + command_id: &str, + pair: (&WorkAttemptIdentityV1, &WorkAttemptIdentityV1), +) -> WorkDuplicateAdjudicationCommandV1 { + WorkDuplicateAdjudicationCommandV1 { + expected_revision: None, + first_attempt: pair.0.clone(), + second_attempt: pair.1.clone(), + evidence: evidence(), + verdict: DuplicateEffortKindV1::ExactDuplicate, + quantities: WorkDuplicateAdjudicationQuantitiesV1 { + wall_micros: Some(10), + token_count: None, + cost_micros: None, + test_count: None, + effect_count: None, + evidence: QuantityEvidenceClassV1::OwnerReceipt, + effect_outcome: DuplicateEffectOutcomeV1::NotApplicable, + coverage: CoverageStateV1::Known, + }, + reason: "independent review".to_owned(), + command_id: id::(command_id), + occurred_at: UtcMicros(50_000), + } + .canonicalized() +} + +#[tokio::test] +async fn duplicate_adjudication_awaited_on_a_native_worker_parks_no_native_thread() { + let profile = tempfile::tempdir().expect("profile"); + let project = tempfile::tempdir().expect("project"); + let runtime = project_runtime(profile.path(), project.path()).await; + let database = runtime.project_database().expect("project database"); + let storage = database.work_storage().expect("Work storage"); + let first = leased_attempt_of(id::("run.leaf-jobs.1"), "leaf-jobs.1"); + let second = leased_attempt_of(id::("run.leaf-jobs.2"), "leaf-jobs.2"); + for attempt in [&first, &second] { + storage + .insert(&authority(), attempt) + .expect("the attempt is stored"); + } + let pair = (first.identity().clone(), second.identity().clone()); + let work = RegisteredWorkApplicationServicesV1::attach(database).expect("Work services"); + let (outcome, writer_waits, reader_waits) = on_native_worker(async move { + let appended = work + .adjudicate_duplicate( + context(), + duplicate_command("command.leaf-jobs.first", (&pair.0, &pair.1)), + evidence(), + ) + .await; + let conflicting = work + .adjudicate_duplicate( + context(), + duplicate_command("command.leaf-jobs.second", (&pair.0, &pair.1)), + evidence(), + ) + .await; + (appended, conflicting) + }) + .await; + assert_eq!( + writer_waits, 0, + "the adjudication parked no native worker on the writer" + ); + assert_eq!( + reader_waits, 0, + "the adjudication parked no native worker on a reader" + ); + let (appended, conflicting) = outcome; + let WorkDuplicateAdjudicationAppendOutcomeV1::Appended(receipt) = + appended.expect("the first adjudication commits") + else { + panic!("the first adjudication appends"); + }; + assert_eq!(receipt.revision().get(), 1); + assert_eq!( + conflicting, + Err(WorkDuplicateAdjudicationStorageErrorV1::RevisionConflict.into()), + "a second command for the same pair at no revision is refused by the revision compare" + ); + let (first_attempt, second_attempt) = ( + &receipt.command().first_attempt, + &receipt.command().second_attempt, + ); + assert_eq!( + storage + .latest_duplicate_adjudication_for_pair(&authority(), first_attempt, second_attempt) + .expect("the pair reads"), + Some(receipt), + "the job committed exactly the first receipt" + ); +} 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 index 7f091f7bd1..98e0e08d71 100644 --- a/crates/tracedecay-application/tests/application_suite/work_run_control_jobs.rs +++ b/crates/tracedecay-application/tests/application_suite/work_run_control_jobs.rs @@ -39,7 +39,7 @@ fn run() -> RunId { id::("run.run-control-jobs") } -fn context() -> RequestContext { +pub(crate) fn context() -> RequestContext { let scope = ResolvedScope::new( id::(PROJECT), id::(REPOSITORY), @@ -71,7 +71,7 @@ fn context() -> RequestContext { .unwrap() } -fn authority() -> WorkAuthority { +pub(crate) fn authority() -> WorkAuthority { WorkAuthority::new( id::(PROJECT), id::(REPOSITORY), @@ -91,9 +91,18 @@ fn route() -> WorkProviderRouteV1 { } fn leased_attempt() -> WorkAttemptV1 { - let identity = - WorkAttemptIdentityV1::new(task(), run(), id::("attempt.run-control-jobs")) - .unwrap(); + leased_attempt_of(run(), "run-control-jobs") +} + +/// A leased attempt of `task()` in `run_id` whose attempt and lease ids end +/// in `suffix`. +pub(crate) fn leased_attempt_of(run_id: RunId, suffix: &str) -> WorkAttemptV1 { + let identity = WorkAttemptIdentityV1::new( + task(), + run_id, + id::(&format!("attempt.{suffix}")), + ) + .unwrap(); let binding = WorkAttemptProjectionBindingV1::new( WorkGraphVersionV1::new(3).unwrap(), WorkProductEventSequenceV1::new(7).unwrap(), @@ -149,7 +158,7 @@ fn leased_attempt() -> WorkAttemptV1 { binding, envelope, WorkLeaseFenceV1::new( - id::("lease.run-control-jobs"), + id::(&format!("lease.{suffix}")), WorkFenceEpochV1::new(1).unwrap(), ) .unwrap(), diff --git a/crates/tracedecay-contracts/src/lib.rs b/crates/tracedecay-contracts/src/lib.rs index 929724cb42..b14aad7ca6 100644 --- a/crates/tracedecay-contracts/src/lib.rs +++ b/crates/tracedecay-contracts/src/lib.rs @@ -466,11 +466,11 @@ pub use work_catalog::{ pub use work_duplicate_adjudication::{ MAX_WORK_DUPLICATE_CLASSIFICATION_ATTEMPTS_V1, PrepareWorkDuplicateAdjudicationRequestV1, WorkDuplicateAdjudicationAppendOutcomeV1, WorkDuplicateAdjudicationPortV1, - WorkDuplicateAdjudicationServiceV1, WorkDuplicateAdjudicationStorageErrorV1, - WorkDuplicateAdjudicationWriteV1, WorkDuplicateAttemptClassificationReadV1, - WorkDuplicateAttemptClassificationRequestV1, WorkDuplicateAttemptClassificationV1, - WorkDuplicateClassificationUnavailableReasonV1, prepare_work_duplicate_adjudication, - work_duplicate_adjudication_input_digest, + WorkDuplicateAdjudicationRecordPortV1, WorkDuplicateAdjudicationServiceV1, + WorkDuplicateAdjudicationStorageErrorV1, WorkDuplicateAdjudicationWriteV1, + WorkDuplicateAttemptClassificationReadV1, WorkDuplicateAttemptClassificationRequestV1, + WorkDuplicateAttemptClassificationV1, WorkDuplicateClassificationUnavailableReasonV1, + prepare_work_duplicate_adjudication, work_duplicate_adjudication_input_digest, }; pub use work_evidence::{ MAX_WORK_ROOTED_EVIDENCE_SOURCES_V1, VerifiedWorkEvidenceRootV1, WorkAnchorHydrationFuture, diff --git a/crates/tracedecay-contracts/src/work_duplicate_adjudication.rs b/crates/tracedecay-contracts/src/work_duplicate_adjudication.rs index bdaa5f64cf..6db88338b6 100644 --- a/crates/tracedecay-contracts/src/work_duplicate_adjudication.rs +++ b/crates/tracedecay-contracts/src/work_duplicate_adjudication.rs @@ -158,13 +158,19 @@ impl WorkDuplicateAdjudicationAppendOutcomeV1 { } } -pub trait WorkDuplicateAdjudicationPortV1: Send + Sync { +/// The replay check, revision compare and append of one adjudication. +/// +/// It carries no `Send + Sync` bound so a storage scoped to one writer +/// transaction can implement it and run the whole record atomically. +pub trait WorkDuplicateAdjudicationRecordPortV1 { fn compare_and_record_duplicate_adjudication( &self, authority: &WorkAuthority, write: &WorkDuplicateAdjudicationWriteV1, ) -> Result; +} +pub trait WorkDuplicateAdjudicationPortV1: Send + Sync { fn latest_duplicate_adjudications_for_attempts( &self, authority: &WorkAuthority, @@ -185,14 +191,16 @@ pub struct WorkDuplicateAdjudicationServiceV1 { storage: S, } -impl WorkDuplicateAdjudicationServiceV1 -where - S: WorkDuplicateAdjudicationPortV1, -{ +impl WorkDuplicateAdjudicationServiceV1 { pub const fn new(storage: S) -> Self { Self { storage } } +} +impl WorkDuplicateAdjudicationServiceV1 +where + S: WorkDuplicateAdjudicationRecordPortV1, +{ #[tracing::instrument( name = "application.work.duplicate.adjudicate", level = "trace", @@ -221,10 +229,15 @@ where current_evidence, }, ) - .map_err(storage_problem)?; + .map_err(ApplicationProblem::from)?; Ok(outcome) } +} +impl WorkDuplicateAdjudicationServiceV1 +where + S: WorkDuplicateAdjudicationPortV1, +{ #[allow(clippy::too_many_arguments)] #[tracing::instrument(name = "application.work.duplicate.prepare", level = "trace", skip_all)] pub fn prepare_adjudication( @@ -248,7 +261,7 @@ where let latest = self .storage .latest_duplicate_adjudication_for_pair(&authority, first_attempt, second_attempt) - .map_err(storage_problem)?; + .map_err(ApplicationProblem::from)?; prepare_work_duplicate_adjudication( request, evidence, @@ -290,7 +303,7 @@ where &request.topology_generation, &attempts, ) - .map_err(storage_problem)?; + .map_err(ApplicationProblem::from)?; Ok(classify_complete_attempt_relations( &authority, request.work_generation, @@ -386,32 +399,30 @@ fn invalid_problem() -> ApplicationProblem { ) } -fn storage_problem(error: WorkDuplicateAdjudicationStorageErrorV1) -> ApplicationProblem { - match error { - WorkDuplicateAdjudicationStorageErrorV1::NotFoundOrNotAuthorized => { - ApplicationProblem::not_found_or_not_authorized(RetryDirective::Never) - } - WorkDuplicateAdjudicationStorageErrorV1::RevisionConflict => ApplicationProblem::conflict( - "application.work.duplicate-adjudication.revision-conflict", - "The duplicate Work adjudication changed after this command was prepared.", - ), - WorkDuplicateAdjudicationStorageErrorV1::IdempotencyConflict => { - ApplicationProblem::conflict( +impl From for ApplicationProblem { + fn from(error: WorkDuplicateAdjudicationStorageErrorV1) -> Self { + match error { + WorkDuplicateAdjudicationStorageErrorV1::NotFoundOrNotAuthorized => { + Self::not_found_or_not_authorized(RetryDirective::Never) + } + WorkDuplicateAdjudicationStorageErrorV1::RevisionConflict => Self::conflict( + "application.work.duplicate-adjudication.revision-conflict", + "The duplicate Work adjudication changed after this command was prepared.", + ), + WorkDuplicateAdjudicationStorageErrorV1::IdempotencyConflict => Self::conflict( "application.work.duplicate-adjudication.idempotency-conflict", "The duplicate Work adjudication command identity was already used with different input.", - ) - } - WorkDuplicateAdjudicationStorageErrorV1::EvidenceStale => { - ApplicationProblem::stale(SafeDiagnostic { + ), + WorkDuplicateAdjudicationStorageErrorV1::EvidenceStale => Self::stale(SafeDiagnostic { code: "application.work.duplicate-adjudication.evidence-stale".to_owned(), message: "The duplicate Work adjudication cites evidence that is not the current Work and topology generation; prepare it again.".to_owned(), - }) - } - WorkDuplicateAdjudicationStorageErrorV1::Unavailable => { - ApplicationProblem::unavailable(SafeDiagnostic { - code: "application.work.duplicate-adjudication.unavailable".to_owned(), - message: "The duplicate Work adjudication authority is unavailable.".to_owned(), - }) + }), + WorkDuplicateAdjudicationStorageErrorV1::Unavailable => { + Self::unavailable(SafeDiagnostic { + code: "application.work.duplicate-adjudication.unavailable".to_owned(), + message: "The duplicate Work adjudication authority is unavailable.".to_owned(), + }) + } } } } 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 087acc1450..d168845c89 100644 --- a/crates/tracedecay-daemon-service/src/invocation/work/request_dispatch.rs +++ b/crates/tracedecay-daemon-service/src/invocation/work/request_dispatch.rs @@ -619,58 +619,59 @@ pub(super) async fn dispatch_work_application( } } WorkApplicationInvocationV1::AdjudicateDuplicate(command) => { - let _span = tracing::trace_span!("daemon.service.work.adjudicate_duplicate").entered(); - { - let authority = match tracedecay_domain::WorkAuthority::new( - context.scope().project_id.clone(), - context.scope().repository_id.clone(), - context.scope().worktree_id.clone(), - context.actor().clone(), - context.grant().digest.clone(), - ) { - Ok(authority) => authority, - Err(_) => { - return DaemonInvocationResponse::problem( - request_id, - DaemonInvocationProblem::InvalidRequest, - ); - } - }; - let adjudicated = preparation::current_duplicate_adjudication_evidence( - ®istered, - &context, - capability, - &use_case, - observed_at, - ) - .and_then(|evidence| { - services - .duplicate_adjudications() - .adjudicate(&context, command, evidence) - }); - if let Ok(outcome) = &adjudicated { - let _observation = - tracedecay_application::observability::record_work_duplicate_observation( - observability_producer.as_deref(), - context.scope().project_id.as_str(), - &authority, - outcome.receipt(), - ); + let authority = match tracedecay_domain::WorkAuthority::new( + context.scope().project_id.clone(), + context.scope().repository_id.clone(), + context.scope().worktree_id.clone(), + context.actor().clone(), + context.grant().digest.clone(), + ) { + Ok(authority) => authority, + Err(_) => { + return DaemonInvocationResponse::problem( + request_id, + DaemonInvocationProblem::InvalidRequest, + ); } - complete_work_effect( - ®istered, - request_id, - &context, - canonical_request_id, - operation_key, - use_case, - input_digest, - adjudicated, - observed_at, - deadline, - WorkApplicationOutcomeV1::AdjudicateDuplicate, - ) + }; + let adjudicated = tracing::Instrument::instrument( + async { + let evidence = preparation::current_duplicate_adjudication_evidence( + ®istered, + &context, + capability, + &use_case, + observed_at, + )?; + services + .adjudicate_duplicate(context.clone(), command, evidence) + .await + }, + tracing::trace_span!("daemon.service.work.adjudicate_duplicate"), + ) + .await; + if let Ok(outcome) = &adjudicated { + let _observation = + tracedecay_application::observability::record_work_duplicate_observation( + observability_producer.as_deref(), + context.scope().project_id.as_str(), + &authority, + outcome.receipt(), + ); } + complete_work_effect( + ®istered, + request_id, + &context, + canonical_request_id, + operation_key, + use_case, + input_digest, + adjudicated, + observed_at, + deadline, + WorkApplicationOutcomeV1::AdjudicateDuplicate, + ) } WorkApplicationInvocationV1::AdjudicateLeak(command) => { let adjudicated: Pin< diff --git a/crates/tracedecay-rusqlite-runtime/src/work/duplicate_adjudication.rs b/crates/tracedecay-rusqlite-runtime/src/work/duplicate_adjudication.rs index dab9858f92..8f916a82c1 100644 --- a/crates/tracedecay-rusqlite-runtime/src/work/duplicate_adjudication.rs +++ b/crates/tracedecay-rusqlite-runtime/src/work/duplicate_adjudication.rs @@ -2,9 +2,9 @@ use tracedecay_contracts::{ MAX_WORK_DUPLICATE_CLASSIFICATION_ATTEMPTS_V1, WorkDuplicateAdjudicationAppendOutcomeV1, - WorkDuplicateAdjudicationPortV1, WorkDuplicateAdjudicationStorageErrorV1, - WorkDuplicateAdjudicationWriteV1, WorkOwnerObservationReceiptV1, - work_duplicate_adjudication_input_digest, + WorkDuplicateAdjudicationPortV1, WorkDuplicateAdjudicationRecordPortV1, + WorkDuplicateAdjudicationStorageErrorV1, WorkDuplicateAdjudicationWriteV1, + WorkOwnerObservationReceiptV1, work_duplicate_adjudication_input_digest, }; use tracedecay_domain::{ ProjectionGenerationId, WorkAttemptIdentityV1, WorkAuthority, @@ -12,15 +12,17 @@ use tracedecay_domain::{ WorkDuplicateAdjudicationRevisionV1, WorkTopologyGenerationRefV1, }; -use crate::exact_sql::{ExactSqlTransaction, ExactSqlValue}; +use crate::exact_sql::ExactSqlValue; use crate::work::{ - RegisteredWorkQuery, WorkSqliteStorage, authority_params_owned, exact_sql_statement, - exact_sql_text, registered_work_query, + RegisteredWorkQuery, RegisteredWorkWrite, WorkSqliteStorage, authority_params_owned, + exact_sql_statement, exact_sql_text, registered_work_query, }; +use crate::work_run_control::WorkSqliteJobStorage; type StorageError = WorkDuplicateAdjudicationStorageErrorV1; -impl WorkDuplicateAdjudicationPortV1 for WorkSqliteStorage { +/// The job's transaction commits the append, and rolls back every refusal. +impl WorkDuplicateAdjudicationRecordPortV1 for WorkSqliteJobStorage<'_> { fn compare_and_record_duplicate_adjudication( &self, authority: &WorkAuthority, @@ -31,15 +33,11 @@ impl WorkDuplicateAdjudicationPortV1 for WorkSqliteStorage { .command .relation_ref(authority) .map_err(|_| StorageError::Unavailable)?; - let transaction = self - .handle() - .begin_immediate() - .map_err(|_| StorageError::Unavailable)?; + let transaction = self.transaction; if let Some((stored_digest, receipt)) = - replay_by_command(&transaction, authority, write.command.command_id.as_str())? + replay_by_command(transaction, authority, write.command.command_id.as_str())? { - let _ = transaction.rollback(); return if stored_digest == write.canonical_input_digest.as_str() { Ok(WorkDuplicateAdjudicationAppendOutcomeV1::Replayed(receipt)) } else { @@ -47,18 +45,16 @@ impl WorkDuplicateAdjudicationPortV1 for WorkSqliteStorage { }; } if write.command.evidence != write.current_evidence { - let _ = transaction.rollback(); return Err(StorageError::EvidenceStale); } - require_attempt(&transaction, authority, &write.command.first_attempt)?; - require_attempt(&transaction, authority, &write.command.second_attempt)?; + require_attempt(transaction, authority, &write.command.first_attempt)?; + require_attempt(transaction, authority, &write.command.second_attempt)?; let current_receipt = - current_adjudication(&transaction, authority, relation_digest.as_str())?; + current_adjudication(transaction, authority, relation_digest.as_str())?; let current = current_receipt.as_ref().map(|receipt| receipt.revision()); if current != write.command.expected_revision { - let _ = transaction.rollback(); return Err(StorageError::RevisionConflict); } let revision = match current { @@ -72,13 +68,12 @@ impl WorkDuplicateAdjudicationPortV1 for WorkSqliteStorage { write.canonical_input_digest.clone(), ) .map_err(|_| StorageError::Unavailable)?; - insert_receipt(&transaction, authority, &receipt)?; - transaction - .commit() - .map_err(|_| StorageError::Unavailable)?; + insert_receipt(transaction, authority, &receipt)?; Ok(WorkDuplicateAdjudicationAppendOutcomeV1::Appended(receipt)) } +} +impl WorkDuplicateAdjudicationPortV1 for WorkSqliteStorage { fn latest_duplicate_adjudications_for_attempts( &self, authority: &WorkAuthority, @@ -184,7 +179,7 @@ fn admit_write( } fn replay_by_command( - transaction: &ExactSqlTransaction, + transaction: &impl RegisteredWorkQuery, authority: &WorkAuthority, command_id: &str, ) -> Result, StorageError> { @@ -242,7 +237,7 @@ fn current_adjudication( } fn require_attempt( - transaction: &ExactSqlTransaction, + transaction: &impl RegisteredWorkQuery, authority: &WorkAuthority, identity: &WorkAttemptIdentityV1, ) -> Result<(), StorageError> { @@ -270,7 +265,7 @@ fn require_attempt( } fn insert_receipt( - transaction: &ExactSqlTransaction, + transaction: &impl RegisteredWorkWrite, authority: &WorkAuthority, receipt: &WorkDuplicateAdjudicationReceiptV1, ) -> Result<(), StorageError> { @@ -280,7 +275,7 @@ fn insert_receipt( ) .map_err(|_| StorageError::Unavailable)?; transaction - .execute( + .work_execute( exact_sql_statement( "INSERT INTO work_duplicate_adjudications_v1 ( project_id, repository_id, worktree_id, actor_id, policy_digest, diff --git a/crates/tracedecay-rusqlite-runtime/src/work_run_control.rs b/crates/tracedecay-rusqlite-runtime/src/work_run_control.rs index 840771ef8c..1eba82467b 100644 --- a/crates/tracedecay-rusqlite-runtime/src/work_run_control.rs +++ b/crates/tracedecay-rusqlite-runtime/src/work_run_control.rs @@ -32,7 +32,7 @@ use crate::work::{ /// running the job. #[derive(Clone, Copy)] pub struct WorkSqliteJobStorage<'job> { - transaction: &'job ExactSqlJobTransaction<'job>, + pub(crate) transaction: &'job ExactSqlJobTransaction<'job>, } impl WorkSqliteStorage { diff --git a/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/work_duplicate_adjudication_storage.rs b/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/work_duplicate_adjudication_storage.rs index f9b2a2d03f..855ceae5f4 100644 --- a/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/work_duplicate_adjudication_storage.rs +++ b/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/work_duplicate_adjudication_storage.rs @@ -5,8 +5,9 @@ use std::num::NonZeroU16; use tracedecay_contracts::{ WorkDuplicateAdjudicationAppendOutcomeV1, WorkDuplicateAdjudicationPortV1, - WorkDuplicateAdjudicationStorageErrorV1, WorkDuplicateAdjudicationWriteV1, - WorkOwnerObservationKindV1, WorkOwnerObservationMarkOutcomeV1, WorkOwnerObservationReceiptV1, + WorkDuplicateAdjudicationRecordPortV1, WorkDuplicateAdjudicationStorageErrorV1, + WorkDuplicateAdjudicationWriteV1, WorkOwnerObservationKindV1, + WorkOwnerObservationMarkOutcomeV1, WorkOwnerObservationReceiptV1, WorkOwnerObservationStoragePortV1, work_duplicate_adjudication_input_digest, }; use tracedecay_domain::{ @@ -104,6 +105,19 @@ fn write_observing( } } +fn record( + store: &RegisteredWorkStore, + authority: &WorkAuthority, + write: &WorkDuplicateAdjudicationWriteV1, +) -> Result { + let (authority, write) = (authority.clone(), write.clone()); + tracedecay_rusqlite_runtime::native_runtime::block_on(store.storage().transact( + move |storage| storage.compare_and_record_duplicate_adjudication(&authority, &write), + )) + .expect("native runtime") + .expect("the writer runs the job") +} + fn insert_attempts(connection: &rusqlite::Connection) { insert_attempts_for(connection, &authority()); } @@ -152,14 +166,8 @@ fn pending_scan_cursor_does_not_skip_the_same_command_in_another_authority() { None, DuplicateEffortKindV1::ExactDuplicate, )); - store - .storage() - .compare_and_record_duplicate_adjudication(&authority(), &write) - .unwrap(); - store - .storage() - .compare_and_record_duplicate_adjudication(&second_authority, &write) - .unwrap(); + record(&store, &authority(), &write).unwrap(); + record(&store, &second_authority, &write).unwrap(); let first = store .storage() @@ -185,20 +193,14 @@ fn adjudication_is_revision_cas_and_exact_replay_on_the_registered_store() { None, DuplicateEffortKindV1::ExactDuplicate, )); - let appended = store - .storage() - .compare_and_record_duplicate_adjudication(&authority(), &first) - .unwrap(); + let appended = record(&store, &authority(), &first).unwrap(); let WorkDuplicateAdjudicationAppendOutcomeV1::Appended(receipt) = appended else { panic!("first adjudication must append") }; assert_eq!(receipt.revision().get(), 1); assert!(matches!( - store - .storage() - .compare_and_record_duplicate_adjudication(&authority(), &first) - .unwrap(), + record(&store, &authority(), &first).unwrap(), WorkDuplicateAdjudicationAppendOutcomeV1::Replayed(_) )); assert_eq!(store.count("work_duplicate_adjudications_v1"), 1); @@ -209,10 +211,7 @@ fn adjudication_is_revision_cas_and_exact_replay_on_the_registered_store() { DuplicateEffortKindV1::NotDuplicate, )); assert_eq!( - store - .storage() - .compare_and_record_duplicate_adjudication(&authority(), &stale) - .unwrap_err(), + record(&store, &authority(), &stale).unwrap_err(), WorkDuplicateAdjudicationStorageErrorV1::RevisionConflict ); @@ -224,10 +223,7 @@ fn adjudication_is_revision_cas_and_exact_replay_on_the_registered_store() { changed_pair.second_attempt = attempt("task.duplicate.3", "run.duplicate.3", "attempt.duplicate.3"); assert_eq!( - store - .storage() - .compare_and_record_duplicate_adjudication(&authority(), &write(changed_pair)) - .unwrap_err(), + record(&store, &authority(), &write(changed_pair)).unwrap_err(), WorkDuplicateAdjudicationStorageErrorV1::RevisionConflict, "an adjudication revision cannot change its exact attempt pair" ); @@ -237,10 +233,7 @@ fn adjudication_is_revision_cas_and_exact_replay_on_the_registered_store() { Some(WorkDuplicateAdjudicationRevisionV1::initial()), DuplicateEffortKindV1::NotDuplicate, )); - let corrected = store - .storage() - .compare_and_record_duplicate_adjudication(&authority(), &correction) - .unwrap(); + let corrected = record(&store, &authority(), &correction).unwrap(); let WorkDuplicateAdjudicationAppendOutcomeV1::Appended(receipt) = corrected else { panic!("correction must append") }; @@ -284,13 +277,12 @@ fn adjudication_refuses_evidence_the_owner_did_not_observe_but_replays_committed DuplicateEffortKindV1::ExactDuplicate, ); assert_eq!( - store - .storage() - .compare_and_record_duplicate_adjudication( - &authority(), - &write_observing(fabricated, advanced.clone()), - ) - .unwrap_err(), + record( + &store, + &authority(), + &write_observing(fabricated, advanced.clone()), + ) + .unwrap_err(), WorkDuplicateAdjudicationStorageErrorV1::EvidenceStale ); assert_eq!(store.count("work_duplicate_adjudications_v1"), 0); @@ -300,19 +292,13 @@ fn adjudication_refuses_evidence_the_owner_did_not_observe_but_replays_committed None, DuplicateEffortKindV1::ExactDuplicate, ); - let appended = store - .storage() - .compare_and_record_duplicate_adjudication(&authority(), &write(committed.clone())) - .unwrap(); + let appended = record(&store, &authority(), &write(committed.clone())).unwrap(); assert!(matches!( appended, WorkDuplicateAdjudicationAppendOutcomeV1::Appended(_) )); assert!(matches!( - store - .storage() - .compare_and_record_duplicate_adjudication( - &authority(), + record(&store, &authority(), &write_observing(committed, advanced), ) .unwrap(), @@ -331,10 +317,7 @@ fn adjudication_refuses_attempts_outside_the_exact_work_authority() { DuplicateEffortKindV1::Unknown, )); assert_eq!( - store - .storage() - .compare_and_record_duplicate_adjudication(&authority(), &write) - .unwrap_err(), + record(&store, &authority(), &write).unwrap_err(), WorkDuplicateAdjudicationStorageErrorV1::NotFoundOrNotAuthorized ); assert_eq!(store.count("work_duplicate_adjudications_v1"), 0); @@ -346,17 +329,16 @@ fn adjudication_persists_a_pending_observation_marker_with_the_receipt() { "duplicate-adjudication-observation-marker", insert_attempts, ); - store - .storage() - .compare_and_record_duplicate_adjudication( - &authority(), - &write(command( - "command.duplicate.observation-marker", - None, - DuplicateEffortKindV1::ExactDuplicate, - )), - ) - .unwrap(); + record( + &store, + &authority(), + &write(command( + "command.duplicate.observation-marker", + None, + DuplicateEffortKindV1::ExactDuplicate, + )), + ) + .unwrap(); let (state, digest): (String, String) = store.inspect(|connection| { connection @@ -378,17 +360,16 @@ fn one_exact_attempt_pair_has_one_revisioned_adjudication_identity() { "duplicate-adjudication-pair-identity", insert_attempts, ); - store - .storage() - .compare_and_record_duplicate_adjudication( - &authority(), - &write(command( - "command.duplicate.pair-identity.first", - None, - DuplicateEffortKindV1::ExactDuplicate, - )), - ) - .unwrap(); + record( + &store, + &authority(), + &write(command( + "command.duplicate.pair-identity.first", + None, + DuplicateEffortKindV1::ExactDuplicate, + )), + ) + .unwrap(); let duplicate_identity = command( "command.duplicate.pair-identity.second", @@ -396,10 +377,7 @@ fn one_exact_attempt_pair_has_one_revisioned_adjudication_identity() { DuplicateEffortKindV1::ExactDuplicate, ); assert_eq!( - store - .storage() - .compare_and_record_duplicate_adjudication(&authority(), &write(duplicate_identity),) - .unwrap_err(), + record(&store, &authority(), &write(duplicate_identity),).unwrap_err(), WorkDuplicateAdjudicationStorageErrorV1::RevisionConflict, "a second command cannot create a second relation for one exact attempt pair" ); @@ -412,17 +390,16 @@ fn pending_duplicate_receipt_is_recoverable_and_exactly_marked_durable() { "duplicate-adjudication-observation-recovery", insert_attempts, ); - let appended = store - .storage() - .compare_and_record_duplicate_adjudication( - &authority(), - &write(command( - "command.duplicate.observation-recovery", - None, - DuplicateEffortKindV1::ExactDuplicate, - )), - ) - .unwrap(); + let appended = record( + &store, + &authority(), + &write(command( + "command.duplicate.observation-recovery", + None, + DuplicateEffortKindV1::ExactDuplicate, + )), + ) + .unwrap(); let store = store.restart("duplicate-adjudication-observation-recovery-restarted"); let pending = store From 5bc4fd0a8c04a89aa616aa653dc5904f01e8e583 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 16 Oct 2026 16:15:00 +0000 Subject: [PATCH 2/3] refactor(work): record leak adjudications as one writer job AdjudicateLeak now runs the contracts leak service inside one ExactSqlHandle::transact job over WorkSqliteJobStorage. The command replay read, the verdict, the attempt check, the revision compare and the append commit together, and the awaiting task parks no native thread. The leak evidence the verdict reads (the attempt, its placement and effect holder, the global-db delivery census and the process registry's live-owner answer) is still collected before the job and handed to it owned, so no global-db read or registry lookup runs on the writer. WorkLeakAdjudicationStoragePortV1 drops its Send + Sync bound; the job storage is its only implementation and the handle-based one is deleted. --- .../src/work/registered.rs | 29 +++++ .../tests/application_suite/work_leaf_jobs.rs | 107 +++++++++++++++++- .../src/work_leak_adjudication.rs | 55 +++++---- .../src/invocation/work/leak_adjudication.rs | 13 ++- .../src/work/leak_adjudication.rs | 52 ++++----- .../work_leak_adjudication_storage.rs | 66 ++++++----- 6 files changed, 230 insertions(+), 92 deletions(-) diff --git a/crates/tracedecay-application/src/work/registered.rs b/crates/tracedecay-application/src/work/registered.rs index a579c16f1f..4fd3832b91 100644 --- a/crates/tracedecay-application/src/work/registered.rs +++ b/crates/tracedecay-application/src/work/registered.rs @@ -312,6 +312,35 @@ impl RegisteredWorkApplicationServicesV1 { }) } + /// Records one leak adjudication as a single writer transaction: the + /// command replay check, the verdict over `evidence` and the + /// revision-compared append commit together. `evidence` is collected + /// before the job, outside the writer. + pub async fn adjudicate_leak( + &self, + context: tracedecay_contracts::RequestContext, + command: tracedecay_contracts::AdjudicateWorkLeakCommandV1, + evidence: E, + scan_started_at: tracedecay_domain::UtcMicros, + scan_deadline: tracedecay_domain::UtcMicros, + ) -> Result< + tracedecay_contracts::WorkLeakAdjudicationOutcomeV1, + tracedecay_contracts::ApplicationProblem, + > + where + E: tracedecay_contracts::WorkLeakEvidencePortV1 + Clone + 'static, + { + self.storage + .transact(move |storage| { + tracedecay_contracts::WorkLeakAdjudicationServiceV1::new(storage, evidence.clone()) + .adjudicate(&context, command.clone(), scan_started_at, scan_deadline) + }) + .await + .unwrap_or_else(|_| { + Err(tracedecay_contracts::WorkLeakAdjudicationStorageErrorV1::Unavailable.into()) + }) + } + pub fn attempts( &self, ) -> &tracedecay_contracts::WorkAttemptService< diff --git a/crates/tracedecay-application/tests/application_suite/work_leaf_jobs.rs b/crates/tracedecay-application/tests/application_suite/work_leaf_jobs.rs index 3766cf0943..3219878f05 100644 --- a/crates/tracedecay-application/tests/application_suite/work_leaf_jobs.rs +++ b/crates/tracedecay-application/tests/application_suite/work_leaf_jobs.rs @@ -1,14 +1,17 @@ use tracedecay_application::work::RegisteredWorkApplicationServicesV1; use tracedecay_contracts::{ - WorkAttemptStoragePort, WorkDuplicateAdjudicationAppendOutcomeV1, - WorkDuplicateAdjudicationPortV1, WorkDuplicateAdjudicationStorageErrorV1, + AdjudicateWorkLeakCommandV1, VerifiedWorkLeakEvidenceV1, WorkAttemptStoragePort, + WorkDuplicateAdjudicationAppendOutcomeV1, WorkDuplicateAdjudicationPortV1, + WorkDuplicateAdjudicationStorageErrorV1, WorkLeakAdjudicationOutcomeV1, + WorkLeakAdjudicationStorageErrorV1, WorkLeakEvidenceErrorV1, WorkLeakEvidencePortV1, }; use tracedecay_domain::test_fixtures::id; use tracedecay_domain::{ - CoverageStateV1, DuplicateEffectOutcomeV1, DuplicateEffortKindV1, ProjectId, + CoverageStateV1, DuplicateEffectOutcomeV1, DuplicateEffortKindV1, LeakOwnerClassV1, ProjectId, ProjectionGenerationId, QuantityEvidenceClassV1, RunId, UtcMicros, WorkAttemptIdentityV1, - WorkCommandId, WorkDuplicateAdjudicationCommandV1, WorkDuplicateAdjudicationEvidenceV1, - WorkDuplicateAdjudicationQuantitiesV1, WorkTopologyGenerationRefV1, + WorkAuthority, WorkCommandId, WorkDuplicateAdjudicationCommandV1, + WorkDuplicateAdjudicationEvidenceV1, WorkDuplicateAdjudicationQuantitiesV1, + WorkExecutionLeakKindV1, WorkExecutionLeakRecoveryV1, WorkTopologyGenerationRefV1, }; use tracedecay_global_db::tests::harness::RegisteredGlobalDbTestRuntime; use tracedecay_rusqlite_runtime::exact_sql::native_thread_reader_waits; @@ -149,3 +152,97 @@ async fn duplicate_adjudication_awaited_on_a_native_worker_parks_no_native_threa "the job committed exactly the first receipt" ); } + +/// A completed scan that found the attempt running without a live owner. +#[derive(Clone)] +struct OwnerlessAttemptScan; + +impl WorkLeakEvidencePortV1 for OwnerlessAttemptScan { + fn inspect( + &self, + _authority: &WorkAuthority, + command: &AdjudicateWorkLeakCommandV1, + scan_started_at: UtcMicros, + _scan_deadline: UtcMicros, + ) -> Result { + Ok(VerifiedWorkLeakEvidenceV1 { + attempt: command.attempt.clone(), + kind: WorkExecutionLeakKindV1::AttemptWithoutLiveOwner, + recovery: WorkExecutionLeakRecoveryV1::Pending, + owner_class: LeakOwnerClassV1::Work, + coverage: CoverageStateV1::Known, + detection_horizon_micros: command.detection_horizon_micros, + scan_started_at, + scan_completed_at: UtcMicros(scan_started_at.0 + 1), + evidence_refs: vec!["work-leak:attempt-without-live-owner:fixture".to_owned()], + }) + } +} + +#[tokio::test] +async fn leak_adjudication_awaited_on_a_native_worker_parks_no_native_thread() { + let profile = tempfile::tempdir().expect("profile"); + let project = tempfile::tempdir().expect("project"); + let runtime = project_runtime(profile.path(), project.path()).await; + let database = runtime.project_database().expect("project database"); + let attempt = leased_attempt_of(id::("run.leaf-jobs.leak"), "leaf-jobs.leak"); + database + .work_storage() + .expect("Work storage") + .insert(&authority(), &attempt) + .expect("the attempt is stored"); + let command = |command_id: &str| AdjudicateWorkLeakCommandV1 { + adjudication_id: "adjudication.leaf-jobs.leak".to_owned(), + expected_revision: None, + attempt: attempt.identity().clone(), + detection_horizon_micros: 1_000, + command_id: id::(command_id), + }; + let (first, second) = ( + command("command.leaf-jobs.leak"), + command("command.leaf-jobs.leak.2"), + ); + let work = RegisteredWorkApplicationServicesV1::attach(database).expect("Work services"); + let (outcomes, writer_waits, reader_waits) = on_native_worker(async move { + let adjudicate = |command: AdjudicateWorkLeakCommandV1| { + work.adjudicate_leak( + context(), + command, + OwnerlessAttemptScan, + UtcMicros(50_000), + UtcMicros(60_000), + ) + }; + ( + adjudicate(first.clone()).await, + adjudicate(first).await, + adjudicate(second).await, + ) + }) + .await; + assert_eq!( + writer_waits, 0, + "the adjudication parked no native worker on the writer" + ); + assert_eq!( + reader_waits, 0, + "the adjudication parked no native worker on a reader" + ); + let (appended, replayed, conflicting) = outcomes; + let WorkLeakAdjudicationOutcomeV1::Appended(receipt) = + appended.expect("the first adjudication commits") + else { + panic!("the first adjudication appends"); + }; + assert_eq!(receipt.revision, 1); + assert_eq!( + replayed, + Ok(WorkLeakAdjudicationOutcomeV1::Replayed(receipt)), + "the same command replays the committed receipt" + ); + assert_eq!( + conflicting, + Err(WorkLeakAdjudicationStorageErrorV1::RevisionConflict.into()), + "a second command at no revision is refused by the revision compare" + ); +} diff --git a/crates/tracedecay-contracts/src/work_leak_adjudication.rs b/crates/tracedecay-contracts/src/work_leak_adjudication.rs index 6f0e5b45d9..defb7a5484 100644 --- a/crates/tracedecay-contracts/src/work_leak_adjudication.rs +++ b/crates/tracedecay-contracts/src/work_leak_adjudication.rs @@ -231,7 +231,12 @@ pub enum WorkLeakAdjudicationStorageErrorV1 { Unavailable, } -pub trait WorkLeakAdjudicationStoragePortV1: Send + Sync { +/// The replay read and the revision-compared append of one leak adjudication. +/// +/// It carries no `Send + Sync` bound so a storage scoped to one writer +/// transaction can implement it and run the replay check and the append +/// atomically. +pub trait WorkLeakAdjudicationStoragePortV1 { fn leak_by_command( &self, authority: &WorkAuthority, @@ -279,14 +284,12 @@ where if let Some(receipt) = self .storage .leak_by_command(&authority, &command.command_id) - .map_err(storage_problem)? + .map_err(ApplicationProblem::from)? { return if receipt.command == command { Ok(WorkLeakAdjudicationOutcomeV1::Replayed(receipt)) } else { - Err(storage_problem( - WorkLeakAdjudicationStorageErrorV1::IdempotencyConflict, - )) + Err(WorkLeakAdjudicationStorageErrorV1::IdempotencyConflict.into()) }; } let evidence = self @@ -320,7 +323,7 @@ where }, }, ) - .map_err(storage_problem) + .map_err(ApplicationProblem::from) } } @@ -360,24 +363,28 @@ fn evidence_problem(error: WorkLeakEvidenceErrorV1) -> ApplicationProblem { } } -fn storage_problem(error: WorkLeakAdjudicationStorageErrorV1) -> ApplicationProblem { - match error { - WorkLeakAdjudicationStorageErrorV1::NotFoundOrNotAuthorized => { - ApplicationProblem::not_found_or_not_authorized(RetryDirective::Never) - } - WorkLeakAdjudicationStorageErrorV1::RevisionConflict => ApplicationProblem::conflict( - "application.work-leak.revision-conflict", - "The Work leak adjudication changed before publication.", - ), - WorkLeakAdjudicationStorageErrorV1::IdempotencyConflict => ApplicationProblem::conflict( - "application.work-leak.idempotency-conflict", - "The Work leak command identity was already used with different input.", - ), - WorkLeakAdjudicationStorageErrorV1::Unavailable => { - ApplicationProblem::unavailable(SafeDiagnostic { - code: "application.work-leak.unavailable".to_owned(), - message: "The Work leak adjudication authority is unavailable.".to_owned(), - }) +impl From for ApplicationProblem { + fn from(error: WorkLeakAdjudicationStorageErrorV1) -> Self { + match error { + WorkLeakAdjudicationStorageErrorV1::NotFoundOrNotAuthorized => { + ApplicationProblem::not_found_or_not_authorized(RetryDirective::Never) + } + WorkLeakAdjudicationStorageErrorV1::RevisionConflict => ApplicationProblem::conflict( + "application.work-leak.revision-conflict", + "The Work leak adjudication changed before publication.", + ), + WorkLeakAdjudicationStorageErrorV1::IdempotencyConflict => { + ApplicationProblem::conflict( + "application.work-leak.idempotency-conflict", + "The Work leak command identity was already used with different input.", + ) + } + WorkLeakAdjudicationStorageErrorV1::Unavailable => { + ApplicationProblem::unavailable(SafeDiagnostic { + code: "application.work-leak.unavailable".to_owned(), + message: "The Work leak adjudication authority is unavailable.".to_owned(), + }) + } } } } diff --git a/crates/tracedecay-daemon-service/src/invocation/work/leak_adjudication.rs b/crates/tracedecay-daemon-service/src/invocation/work/leak_adjudication.rs index 6891ee1e32..96b6bde576 100644 --- a/crates/tracedecay-daemon-service/src/invocation/work/leak_adjudication.rs +++ b/crates/tracedecay-daemon-service/src/invocation/work/leak_adjudication.rs @@ -4,6 +4,7 @@ use tracedecay_contracts::clock::now_micros; use tracedecay_contracts::{ApplicationProblem, Deadline, RequestContext, SafeDiagnostic}; use tracedecay_domain::{ManifestDigest, UtcMicros, canonical_sha256}; +#[derive(Clone)] pub(super) struct DaemonWorkLeakEvidenceV1 { pub(super) attempt: tracedecay_domain::WorkAttemptV1, pub(super) has_live_owner: bool, @@ -222,9 +223,15 @@ pub(super) async fn adjudicate_leak( &storage, ) .await?; - let service = - tracedecay_contracts::WorkLeakAdjudicationServiceV1::new(storage.clone(), evidence); - service.adjudicate(context, command, observed_at, scan_deadline) + services + .adjudicate_leak( + context.clone(), + command, + evidence, + observed_at, + scan_deadline, + ) + .await } #[tracing::instrument( diff --git a/crates/tracedecay-rusqlite-runtime/src/work/leak_adjudication.rs b/crates/tracedecay-rusqlite-runtime/src/work/leak_adjudication.rs index 6a1ba243cb..c1b0f92c58 100644 --- a/crates/tracedecay-rusqlite-runtime/src/work/leak_adjudication.rs +++ b/crates/tracedecay-rusqlite-runtime/src/work/leak_adjudication.rs @@ -6,30 +6,26 @@ use tracedecay_contracts::{ }; use tracedecay_domain::{WorkAuthority, canonical_sha256}; -use crate::exact_sql::{ExactSqlTransaction, ExactSqlValue}; +use crate::exact_sql::ExactSqlValue; use crate::work::{ - WorkSqliteStorage, authority_params_owned, exact_sql_integer, exact_sql_statement, - exact_sql_text, registered_work_query, + RegisteredWorkQuery, RegisteredWorkWrite, authority_params_owned, exact_sql_integer, + exact_sql_statement, exact_sql_text, registered_work_query, }; +use crate::work_run_control::WorkSqliteJobStorage; type StorageError = WorkLeakAdjudicationStorageErrorV1; -impl WorkLeakAdjudicationStoragePortV1 for WorkSqliteStorage { +/// The job's transaction commits the append, and rolls back every refusal. +impl WorkLeakAdjudicationStoragePortV1 for WorkSqliteJobStorage<'_> { fn leak_by_command( &self, authority: &WorkAuthority, command_id: &tracedecay_domain::WorkCommandId, ) -> Result, StorageError> { - let transaction = self - .handle() - .begin_deferred() - .map_err(|_| StorageError::Unavailable)?; - let receipt = replay_by_command(&transaction, authority, command_id.as_str())? - .map(|(_, receipt)| receipt); - transaction - .commit() - .map_err(|_| StorageError::Unavailable)?; - Ok(receipt) + Ok( + replay_by_command(self.transaction, authority, command_id.as_str())? + .map(|(_, receipt)| receipt), + ) } fn compare_and_record_leak( @@ -38,36 +34,28 @@ impl WorkLeakAdjudicationStoragePortV1 for WorkSqliteStorage { write: &WorkLeakAdjudicationWriteV1, ) -> Result { validate_write(write)?; - let transaction = self - .handle() - .begin_immediate() - .map_err(|_| StorageError::Unavailable)?; + let transaction = self.transaction; if let Some((digest, receipt)) = replay_by_command( - &transaction, + transaction, authority, write.receipt.command.command_id.as_str(), )? { - let _ = transaction.rollback(); return if digest == write.receipt.canonical_input_digest.as_str() { Ok(WorkLeakAdjudicationOutcomeV1::Replayed(receipt)) } else { Err(StorageError::IdempotencyConflict) }; } - require_attempt(&transaction, authority, write)?; + require_attempt(transaction, authority, write)?; let current = current_revision( - &transaction, + transaction, authority, &write.receipt.command.adjudication_id, )?; if current != write.receipt.command.expected_revision { - let _ = transaction.rollback(); return Err(StorageError::RevisionConflict); } - insert_receipt(&transaction, authority, write)?; - transaction - .commit() - .map_err(|_| StorageError::Unavailable)?; + insert_receipt(transaction, authority, write)?; Ok(WorkLeakAdjudicationOutcomeV1::Appended( write.receipt.clone(), )) @@ -100,7 +88,7 @@ fn validate_write(write: &WorkLeakAdjudicationWriteV1) -> Result<(), StorageErro } fn replay_by_command( - transaction: &ExactSqlTransaction, + transaction: &impl RegisteredWorkQuery, authority: &WorkAuthority, command_id: &str, ) -> Result, StorageError> { @@ -150,7 +138,7 @@ fn replay_by_command( } fn require_attempt( - transaction: &ExactSqlTransaction, + transaction: &impl RegisteredWorkQuery, authority: &WorkAuthority, write: &WorkLeakAdjudicationWriteV1, ) -> Result<(), StorageError> { @@ -179,7 +167,7 @@ fn require_attempt( } fn current_revision( - transaction: &ExactSqlTransaction, + transaction: &impl RegisteredWorkQuery, authority: &WorkAuthority, adjudication_id: &str, ) -> Result, StorageError> { @@ -207,7 +195,7 @@ fn current_revision( } fn insert_receipt( - transaction: &ExactSqlTransaction, + transaction: &impl RegisteredWorkWrite, authority: &WorkAuthority, write: &WorkLeakAdjudicationWriteV1, ) -> Result<(), StorageError> { @@ -218,7 +206,7 @@ fn insert_receipt( ) .map_err(|_| StorageError::Unavailable)?; transaction - .execute( + .work_execute( exact_sql_statement( "INSERT INTO work_leak_adjudications_v1 ( project_id, repository_id, worktree_id, actor_id, policy_digest, diff --git a/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/work_leak_adjudication_storage.rs b/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/work_leak_adjudication_storage.rs index 3fa698b98c..bd3b344843 100644 --- a/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/work_leak_adjudication_storage.rs +++ b/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/work_leak_adjudication_storage.rs @@ -25,6 +25,7 @@ use tracedecay_domain::{ }; use common::fixture_abs_root; +use tracedecay_rusqlite_runtime::native_runtime::block_on; use work_registered_store::RegisteredWorkStore; use tracedecay_domain::test_fixtures::id; @@ -186,6 +187,39 @@ fn leak_receipt(attempt: &WorkAttemptV1) -> WorkLeakAdjudicationReceiptV1 { } } +fn record( + store: &RegisteredWorkStore, + authority: &WorkAuthority, + receipt: &WorkLeakAdjudicationReceiptV1, +) -> Result { + let (authority, receipt) = (authority.clone(), receipt.clone()); + block_on(store.storage().transact(move |storage| { + storage.compare_and_record_leak( + &authority, + &WorkLeakAdjudicationWriteV1 { + receipt: receipt.clone(), + }, + ) + })) + .expect("native runtime") + .expect("the writer runs the job") +} + +fn replay( + store: &RegisteredWorkStore, + authority: &WorkAuthority, + receipt: &WorkLeakAdjudicationReceiptV1, +) -> Result, WorkLeakAdjudicationStorageErrorV1> { + let (authority, command_id) = (authority.clone(), receipt.command.command_id.clone()); + block_on( + store + .storage() + .transact(move |storage| storage.leak_by_command(&authority, &command_id)), + ) + .expect("native runtime") + .expect("the writer runs the job") +} + #[test] fn leak_adjudication_replays_exact_receipt_across_restart() { let mut store = RegisteredWorkStore::start("leak-adjudication-replay"); @@ -193,15 +227,7 @@ fn leak_adjudication_replays_exact_receipt_across_restart() { let attempt = terminal_attempt(); store.storage().insert(&authority, &attempt).unwrap(); let receipt = leak_receipt(&attempt); - let outcome = store - .storage() - .compare_and_record_leak( - &authority, - &WorkLeakAdjudicationWriteV1 { - receipt: receipt.clone(), - }, - ) - .unwrap(); + let outcome = record(&store, &authority, &receipt).unwrap(); assert_eq!( outcome, WorkLeakAdjudicationOutcomeV1::Appended(receipt.clone()) @@ -209,13 +235,7 @@ fn leak_adjudication_replays_exact_receipt_across_restart() { assert_eq!(store.count("work_leak_adjudications_v1"), 1); store = store.restart("leak-adjudication-replay"); - assert_eq!( - store - .storage() - .leak_by_command(&authority, &receipt.command.command_id) - .unwrap(), - Some(receipt), - ); + assert_eq!(replay(&store, &authority, &receipt).unwrap(), Some(receipt),); } #[test] @@ -238,19 +258,9 @@ fn leak_adjudication_replay_rejects_corrupted_scalar_truth() { let attempt = terminal_attempt(); store.storage().insert(&authority, &attempt).unwrap(); let receipt = leak_receipt(&attempt); - store - .storage() - .compare_and_record_leak( - &authority, - &WorkLeakAdjudicationWriteV1 { - receipt: receipt.clone(), - }, - ) - .unwrap(); + record(&store, &authority, &receipt).unwrap(); assert_eq!( - store - .storage() - .leak_by_command(&authority, &receipt.command.command_id), + replay(&store, &authority, &receipt), Err(WorkLeakAdjudicationStorageErrorV1::Unavailable), ); } From 76cc22d4abcac22c61a09900f7c54f5a2ad933f2 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 16 Oct 2026 16:48:46 +0000 Subject: [PATCH 3/3] refactor(work): admit and release placements as writer jobs AdmitPlacement and ReleasePlacement now run the contracts placement service inside one ExactSqlHandle::transact job over WorkSqliteJobStorage, and the awaiting task parks no native thread. The Git observation (discovery, status and the rev-list count) no longer runs inside a storage transaction. Admit observes command.target before the job and hands the job the owned observation; the replay check, the exclusivity read and the publication commit together. Release reads the placement on a reader, refuses a stale expected authority version before touching Git, observes the target, and the job then publishes only while the placement still has that target and version, refusing with the existing authority conflict when another release committed in between. publish_placement moves to a WorkPlacementPublicationPort, and WorkPlacementStoragePort drops its Send + Sync bound so the job storage can implement both. The handle-based publication had no other production caller and is deleted; the handle keeps the reads that preflight, status and cleanup use, sharing their SQL with the job. work_authority is exported so the release pre-read uses the same authority the service derives. --- .../src/work/registered.rs | 82 +++++ .../tests/application_suite/work_leaf_jobs.rs | 211 +++++++++++- .../work_run_control_jobs.rs | 2 +- crates/tracedecay-contracts/src/lib.rs | 8 +- crates/tracedecay-contracts/src/work.rs | 4 +- .../src/work_placement.rs | 181 ++++++----- .../contracts_suite/work_placement_service.rs | 6 +- .../contracts_suite/work_topology_view.rs | 12 +- .../src/invocation/work/request_dispatch.rs | 82 ++--- .../src/work_placement.rs | 304 +++++++++++------- .../rusqlite_suite/work_placement_storage.rs | 147 ++++----- 11 files changed, 712 insertions(+), 327 deletions(-) diff --git a/crates/tracedecay-application/src/work/registered.rs b/crates/tracedecay-application/src/work/registered.rs index 4fd3832b91..b5ebb2ef3b 100644 --- a/crates/tracedecay-application/src/work/registered.rs +++ b/crates/tracedecay-application/src/work/registered.rs @@ -341,6 +341,84 @@ impl RegisteredWorkApplicationServicesV1 { }) } + /// Admits one placement as a single writer transaction: the replay check, + /// the exclusivity read and the publication commit together. `observe` + /// reads the target (Git) before the job, never on the writer. + pub async fn admit_placement( + &self, + context: tracedecay_contracts::RequestContext, + command: tracedecay_contracts::AdmitWorkPlacementCommand, + observe: impl FnOnce( + &tracedecay_domain::WorkPlacementTargetV1, + ) -> Result< + tracedecay_domain::WorkPlacementObservationV1, + tracedecay_contracts::ApplicationProblem, + >, + ) -> Result { + let observation = observe(&command.target)?; + self.storage + .transact(move |storage| { + tracedecay_contracts::WorkPlacementService::new(storage).admit_placement( + &context, + command.clone(), + |_| Ok(observation), + ) + }) + .await + .unwrap_or_else(|_| Err(placement_unavailable())) + } + + /// Releases one placement. The placement is read on a reader and its + /// target observed before the job; the job then publishes only while the + /// placement still has the observed target and the expected authority + /// version, so the Git observation never runs on the writer. + pub async fn release_placement( + &self, + context: tracedecay_contracts::RequestContext, + command: tracedecay_contracts::ReleaseWorkPlacementCommand, + observe: impl FnOnce( + &tracedecay_domain::WorkPlacementTargetV1, + ) -> Result< + tracedecay_domain::WorkPlacementObservationV1, + tracedecay_contracts::ApplicationProblem, + >, + ) -> Result { + let authority = tracedecay_contracts::work_authority(&context)?; + let identity = tracedecay_domain::WorkPlacementIdentityV1::new( + command.task_id.clone(), + command.run_id.clone(), + ); + let current = self + .storage + .load_placement_async(&authority, &identity) + .await? + .ok_or(tracedecay_contracts::WorkPlacementStorageError::NotFoundOrNotAuthorized)?; + if current.authority_version() != command.expected_authority_version { + return Err(tracedecay_contracts::WorkPlacementStorageError::AuthorityConflict.into()); + } + let observation = observe(current.target())?; + let observed_target = current.target().clone(); + self.storage + .transact(move |storage| { + tracedecay_contracts::WorkPlacementService::new(storage).release( + &context, + command.clone(), + |target| { + if target == &observed_target { + Ok(observation) + } else { + Err( + tracedecay_contracts::WorkPlacementStorageError::AuthorityConflict + .into(), + ) + } + }, + ) + }) + .await + .unwrap_or_else(|_| Err(placement_unavailable())) + } + pub fn attempts( &self, ) -> &tracedecay_contracts::WorkAttemptService< @@ -480,6 +558,10 @@ pub fn authorized_work_intelligence_service( )) } +fn placement_unavailable() -> tracedecay_contracts::ApplicationProblem { + tracedecay_contracts::WorkPlacementStorageError::Unavailable.into() +} + fn run_control_unavailable() -> tracedecay_contracts::ApplicationProblem { tracedecay_contracts::WorkRunControlStorageError::Unavailable.into() } diff --git a/crates/tracedecay-application/tests/application_suite/work_leaf_jobs.rs b/crates/tracedecay-application/tests/application_suite/work_leaf_jobs.rs index 3219878f05..4c931d8eba 100644 --- a/crates/tracedecay-application/tests/application_suite/work_leaf_jobs.rs +++ b/crates/tracedecay-application/tests/application_suite/work_leaf_jobs.rs @@ -1,17 +1,23 @@ +use std::sync::{Arc, Mutex}; use tracedecay_application::work::RegisteredWorkApplicationServicesV1; + use tracedecay_contracts::{ - AdjudicateWorkLeakCommandV1, VerifiedWorkLeakEvidenceV1, WorkAttemptStoragePort, + AdjudicateWorkLeakCommandV1, AdmitWorkPlacementCommand, ApplicationProblem, + ReleaseWorkPlacementCommand, VerifiedWorkLeakEvidenceV1, WorkAttemptStoragePort, WorkDuplicateAdjudicationAppendOutcomeV1, WorkDuplicateAdjudicationPortV1, WorkDuplicateAdjudicationStorageErrorV1, WorkLeakAdjudicationOutcomeV1, WorkLeakAdjudicationStorageErrorV1, WorkLeakEvidenceErrorV1, WorkLeakEvidencePortV1, + WorkPlacementStorageError, WorkPlacementStoragePort, }; -use tracedecay_domain::test_fixtures::id; +use tracedecay_domain::test_fixtures::{fixture_abs_root, id}; use tracedecay_domain::{ CoverageStateV1, DuplicateEffectOutcomeV1, DuplicateEffortKindV1, LeakOwnerClassV1, ProjectId, ProjectionGenerationId, QuantityEvidenceClassV1, RunId, UtcMicros, WorkAttemptIdentityV1, WorkAuthority, WorkCommandId, WorkDuplicateAdjudicationCommandV1, WorkDuplicateAdjudicationEvidenceV1, WorkDuplicateAdjudicationQuantitiesV1, - WorkExecutionLeakKindV1, WorkExecutionLeakRecoveryV1, WorkTopologyGenerationRefV1, + WorkExecutionLeakKindV1, WorkExecutionLeakRecoveryV1, WorkPlacementIdentityV1, + WorkPlacementKindV1, WorkPlacementObservationV1, WorkPlacementStateV1, WorkPlacementTargetV1, + WorkTopologyGenerationRefV1, }; use tracedecay_global_db::tests::harness::RegisteredGlobalDbTestRuntime; use tracedecay_rusqlite_runtime::exact_sql::native_thread_reader_waits; @@ -19,7 +25,7 @@ use tracedecay_rusqlite_runtime::native_runtime::{ NativeTask, NativeTaskOutcome, native_thread_writer_waits, }; -use crate::work_run_control_jobs::{authority, context, leased_attempt_of}; +use crate::work_run_control_jobs::{authority, context, leased_attempt_of, task}; const PROJECT: &str = "project.run-control-jobs"; @@ -246,3 +252,200 @@ async fn leak_adjudication_awaited_on_a_native_worker_parks_no_native_thread() { "a second command at no revision is refused by the revision compare" ); } + +fn placement_target() -> WorkPlacementTargetV1 { + WorkPlacementTargetV1::new( + WorkPlacementKindV1::LinkedWorktree, + Some(fixture_abs_root("/workspace/leaf-jobs-placement")), + false, + true, + ) + .unwrap() +} + +/// A Git observer stand-in that reports a clean target and records the name +/// of every thread it ran on. +#[derive(Clone, Default)] +struct RecordingObserver { + threads: Arc>>>, +} + +impl RecordingObserver { + fn observe( + &self, + _target: &WorkPlacementTargetV1, + ) -> Result { + self.threads + .lock() + .unwrap() + .push(std::thread::current().name().map(str::to_owned)); + Ok(WorkPlacementObservationV1 { + dirty_tracked_paths: 0, + untracked_paths: 0, + unique_commits: Some(0), + readable: true, + active_holder: false, + network_required: false, + observed_at: UtcMicros(50_000), + }) + } + + fn observer( + &self, + ) -> impl Fn(&WorkPlacementTargetV1) -> Result + + Send + + Sync + + 'static { + let recorder = self.clone(); + move |target| recorder.observe(target) + } + + fn threads(&self) -> Vec> { + self.threads.lock().unwrap().clone() + } +} + +fn admit_command(run: &str) -> AdmitWorkPlacementCommand { + AdmitWorkPlacementCommand { + task_id: task(), + run_id: id::(run), + target: placement_target(), + retention_eligible_at: None, + occurred_at: UtcMicros(50_000), + } +} + +fn release_command(run: &str, expected_authority_version: u64) -> ReleaseWorkPlacementCommand { + ReleaseWorkPlacementCommand { + task_id: task(), + run_id: id::(run), + expected_authority_version, + occurred_at: UtcMicros(60_000), + } +} + +#[tokio::test] +async fn placement_admit_and_release_awaited_on_a_native_worker_observe_off_the_writer() { + let profile = tempfile::tempdir().expect("profile"); + let project = tempfile::tempdir().expect("project"); + let runtime = project_runtime(profile.path(), project.path()).await; + let database = runtime.project_database().expect("project database"); + let work = RegisteredWorkApplicationServicesV1::attach(database).expect("Work services"); + let observer = RecordingObserver::default(); + let (admit_observer, release_observer) = (observer.observer(), observer.observer()); + let (outcome, writer_waits, reader_waits) = on_native_worker(async move { + let admitted = work + .admit_placement( + context(), + admit_command("run.leaf-jobs.placement"), + admit_observer, + ) + .await?; + let released = work + .release_placement( + context(), + release_command("run.leaf-jobs.placement", admitted.authority_version()), + release_observer, + ) + .await?; + Ok::<_, ApplicationProblem>((admitted.state(), released.state())) + }) + .await; + assert_eq!( + writer_waits, 0, + "the placement parked no native worker on the writer" + ); + assert_eq!( + reader_waits, 0, + "the placement parked no native worker on a reader" + ); + assert_eq!( + outcome, + Ok(( + WorkPlacementStateV1::Admitted, + WorkPlacementStateV1::Released + )) + ); + let threads = observer.threads(); + assert_eq!( + threads.len(), + 2, + "admit and release each observed the target once" + ); + assert!( + threads + .iter() + .all(|name| name.as_deref() != Some("tracedecay-rusqlite-writer")), + "the Git observation never ran on the writer thread: {threads:?}" + ); + assert_eq!( + database + .work_storage() + .expect("Work storage") + .load_placement( + &authority(), + &WorkPlacementIdentityV1::new(task(), id::("run.leaf-jobs.placement")), + ) + .expect("the placement reads") + .map(|placement| placement.state()), + Some(WorkPlacementStateV1::Released), + "the release committed" + ); +} + +#[tokio::test] +async fn placement_release_refuses_a_placement_that_changed_after_its_observation() { + let profile = tempfile::tempdir().expect("profile"); + let project = tempfile::tempdir().expect("project"); + let runtime = project_runtime(profile.path(), project.path()).await; + let database = runtime.project_database().expect("project database"); + let work = RegisteredWorkApplicationServicesV1::attach(database).expect("Work services"); + let interloper = RegisteredWorkApplicationServicesV1::attach(database).expect("Work services"); + let clean = RecordingObserver::default(); + let observer = clean.observer(); + let outcome = NativeTask::spawn(async move { + let admitted = work + .admit_placement(context(), admit_command("run.leaf-jobs.raced"), observer) + .await + .map_err(|problem| format!("admit: {problem:?}"))?; + let version = admitted.authority_version(); + work.release_placement( + context(), + release_command("run.leaf-jobs.raced", version), + move |target| { + // Another release of the same placement commits while this + // one is observing its target. + let target = target.clone(); + std::thread::spawn(move || { + let observer = RecordingObserver::default(); + tracedecay_rusqlite_runtime::native_runtime::block_on( + interloper.release_placement( + context(), + release_command("run.leaf-jobs.raced", version), + move |_| observer.observe(&target), + ), + ) + }) + .join() + .expect("the interloper joins") + .expect("native runtime") + .expect("the interloping release commits"); + clean.observe(&placement_target()) + }, + ) + .await + .map_err(|problem| format!("release: {problem:?}")) + }) + .expect("native task spawns"); + let NativeTaskOutcome::Completed(outcome) = outcome.wait().await else { + panic!("the native task completes"); + }; + assert_eq!( + outcome, + Err(format!( + "release: {:?}", + ApplicationProblem::from(WorkPlacementStorageError::AuthorityConflict) + )), + "the job refuses with the authority conflict instead of publishing over the newer release" + ); +} 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 index 98e0e08d71..e5e284df64 100644 --- a/crates/tracedecay-application/tests/application_suite/work_run_control_jobs.rs +++ b/crates/tracedecay-application/tests/application_suite/work_run_control_jobs.rs @@ -31,7 +31,7 @@ const REPOSITORY: &str = "repository.run-control-jobs"; const WORKTREE: &str = "worktree.run-control-jobs"; const ACTOR: &str = "actor.run-control-jobs"; -fn task() -> TaskId { +pub(crate) fn task() -> TaskId { id::("task.run-control-jobs") } diff --git a/crates/tracedecay-contracts/src/lib.rs b/crates/tracedecay-contracts/src/lib.rs index b14aad7ca6..55c7782800 100644 --- a/crates/tracedecay-contracts/src/lib.rs +++ b/crates/tracedecay-contracts/src/lib.rs @@ -434,7 +434,9 @@ pub use surface_contracts::{ CodeSymbolSearchSurfaceRequest, CodeTimelineSurfaceRequest, CodeTypeHierarchySurfaceRequest, NativeIntegrationSurfaceRequest, PrimitiveCodeSurfaceRequest, primitive_code_into_primitive, }; -pub use work::{WorkRoutingSnapshotErrorV1, WorkRoutingSnapshotPortV1, WorkRoutingSnapshotV1}; +pub use work::{ + WorkRoutingSnapshotErrorV1, WorkRoutingSnapshotPortV1, WorkRoutingSnapshotV1, work_authority, +}; pub use work_artifact_hydration::{ WorkArtifactHydrationRequestV1, WorkArtifactHydrationService, WorkArtifactHydrationV1, WorkAttemptArtifactsV1, WorkAttemptEvidencePageV1, WorkAttemptEvidenceReadPort, @@ -527,8 +529,8 @@ pub use work_owner_observation::{ }; pub use work_placement::{ AdmitWorkPlacementCommand, ReleaseWorkPlacementCommand, WorkPlacementPreflightRequestV1, - WorkPlacementReadingV1, WorkPlacementService, WorkPlacementStatusRequestV1, - WorkPlacementStorageError, WorkPlacementStoragePort, + WorkPlacementPublicationPort, WorkPlacementReadingV1, WorkPlacementService, + WorkPlacementStatusRequestV1, WorkPlacementStorageError, WorkPlacementStoragePort, }; pub use work_product::{ AcceptWorkProposalDispositionV1, AcceptWorkProposalRequestV1, AcceptWorkTaskRequestV1, diff --git a/crates/tracedecay-contracts/src/work.rs b/crates/tracedecay-contracts/src/work.rs index 5a14dd50b1..ae8dbef917 100644 --- a/crates/tracedecay-contracts/src/work.rs +++ b/crates/tracedecay-contracts/src/work.rs @@ -58,9 +58,7 @@ pub trait WorkRoutingSnapshotPortV1: Send + Sync { ) -> Result; } -pub(crate) fn work_authority( - context: &RequestContext, -) -> Result { +pub fn work_authority(context: &RequestContext) -> Result { WorkAuthority::new( context.scope().project_id.clone(), context.scope().repository_id.clone(), diff --git a/crates/tracedecay-contracts/src/work_placement.rs b/crates/tracedecay-contracts/src/work_placement.rs index 1a63ad7bf5..64c69ce2a8 100644 --- a/crates/tracedecay-contracts/src/work_placement.rs +++ b/crates/tracedecay-contracts/src/work_placement.rs @@ -47,7 +47,10 @@ pub enum WorkPlacementStorageError { } /// The durable placement relations, one per run. -pub trait WorkPlacementStoragePort: Send + Sync { +/// +/// It carries no `Send + Sync` bound so a storage scoped to one writer +/// transaction can implement it. +pub trait WorkPlacementStoragePort { fn load_placement( &self, authority: &WorkAuthority, @@ -72,7 +75,10 @@ pub trait WorkPlacementStoragePort: Send + Sync { ) -> Result { Err(WorkPlacementStorageError::Unavailable) } +} +/// The publication an admission or a release makes. +pub trait WorkPlacementPublicationPort: WorkPlacementStoragePort { /// Publishes `next` under a compare-and-swap on the authority version the /// caller read. `expected` is `None` only for the first admission. fn publish_placement( @@ -141,14 +147,16 @@ pub struct WorkPlacementService { storage: S, } -impl WorkPlacementService -where - S: WorkPlacementStoragePort, -{ +impl WorkPlacementService { pub const fn new(storage: S) -> Self { Self { storage } } +} +impl WorkPlacementService +where + S: WorkPlacementStoragePort, +{ /// Cleanup-only exact-scope census, intentionally independent of the /// current caller's actor and policy lineage. pub fn has_target_holder_in_exact_repository_root( @@ -159,7 +167,7 @@ where ) -> Result { self.storage .has_target_holder_in_exact_repository_root(project_id, repository_id, root) - .map_err(storage_problem) + .map_err(ApplicationProblem::from) } /// Evaluates a placement without changing anything. @@ -182,6 +190,75 @@ where self.evaluate(&authority, identity, request.target, observe) } + /// Reads the durable placement relation for one run. + #[tracing::instrument(name = "application.work.placement.status", level = "trace", skip_all)] + pub fn status( + &self, + context: &RequestContext, + request: &WorkPlacementStatusRequestV1, + ) -> Result { + let authority = work_authority(context)?; + let identity = + WorkPlacementIdentityV1::new(request.task_id.clone(), request.run_id.clone()); + Ok( + match self + .storage + .load_placement(&authority, &identity) + .map_err(ApplicationProblem::from)? + { + Some(placement) => WorkPlacementReadingV1::Placed { placement }, + None => WorkPlacementReadingV1::Absent, + }, + ) + } + + /// Observes the target and folds in the exclusivity fact only storage knows. + fn evaluate( + &self, + authority: &WorkAuthority, + identity: WorkPlacementIdentityV1, + target: WorkPlacementTargetV1, + observe: impl FnOnce( + &WorkPlacementTargetV1, + ) -> Result, + ) -> Result { + let mut observation = observe(&target)?; + observation.active_holder = self.has_foreign_holder(authority, &identity, &target)?; + Ok(WorkPlacementPreflightV1::evaluate( + identity, + target, + observation, + )) + } + + /// Whether a *different* run already holds this exclusive target. + /// + /// The run's own placement is not its own blocker, which is what lets a + /// repeat preflight of an already-admitted placement stay admissible. + fn has_foreign_holder( + &self, + authority: &WorkAuthority, + identity: &WorkPlacementIdentityV1, + target: &WorkPlacementTargetV1, + ) -> Result { + if !target.kind().is_exclusive() { + return Ok(false); + } + let Some(root) = target.root() else { + return Ok(false); + }; + Ok(self + .storage + .target_holder(authority, root) + .map_err(ApplicationProblem::from)? + .is_some_and(|holder| &holder != identity)) + } +} + +impl WorkPlacementService +where + S: WorkPlacementPublicationPort, +{ /// Admits a placement from a fresh, unblocked preflight. /// /// The preflight is re-run here rather than trusted from a prior call: an @@ -202,7 +279,7 @@ where let existing = self .storage .load_placement(&authority, &identity) - .map_err(storage_problem)?; + .map_err(ApplicationProblem::from)?; if let Some(existing) = existing { // Re-admitting the same target is an idempotent replay; a different // target under the same run identity is a conflict, never a move. @@ -229,32 +306,10 @@ where .map_err(contract_problem)?; self.storage .publish_placement(&authority, None, &placement) - .map_err(storage_problem)?; + .map_err(ApplicationProblem::from)?; Ok(placement) } - /// Reads the durable placement relation for one run. - #[tracing::instrument(name = "application.work.placement.status", level = "trace", skip_all)] - pub fn status( - &self, - context: &RequestContext, - request: &WorkPlacementStatusRequestV1, - ) -> Result { - let authority = work_authority(context)?; - let identity = - WorkPlacementIdentityV1::new(request.task_id.clone(), request.run_id.clone()); - Ok( - match self - .storage - .load_placement(&authority, &identity) - .map_err(storage_problem)? - { - Some(placement) => WorkPlacementReadingV1::Placed { placement }, - None => WorkPlacementReadingV1::Absent, - }, - ) - } - /// Gives the target up, or quarantines it when removal is blocked. /// /// This never deletes. It publishes what the fresh cleanup preflight found, @@ -275,7 +330,7 @@ where let current = self .storage .load_placement(&authority, &identity) - .map_err(storage_problem)? + .map_err(ApplicationProblem::from)? .ok_or_else(not_found_problem)?; if current.authority_version() != command.expected_authority_version { return Err(authority_conflict_problem()); @@ -289,61 +344,23 @@ where .map_err(contract_problem)?; self.storage .publish_placement(&authority, Some(current.authority_version()), &next) - .map_err(storage_problem)?; + .map_err(ApplicationProblem::from)?; Ok(next) } - - /// Observes the target and folds in the exclusivity fact only storage knows. - fn evaluate( - &self, - authority: &WorkAuthority, - identity: WorkPlacementIdentityV1, - target: WorkPlacementTargetV1, - observe: impl FnOnce( - &WorkPlacementTargetV1, - ) -> Result, - ) -> Result { - let mut observation = observe(&target)?; - observation.active_holder = self.has_foreign_holder(authority, &identity, &target)?; - Ok(WorkPlacementPreflightV1::evaluate( - identity, - target, - observation, - )) - } - - /// Whether a *different* run already holds this exclusive target. - /// - /// The run's own placement is not its own blocker, which is what lets a - /// repeat preflight of an already-admitted placement stay admissible. - fn has_foreign_holder( - &self, - authority: &WorkAuthority, - identity: &WorkPlacementIdentityV1, - target: &WorkPlacementTargetV1, - ) -> Result { - if !target.kind().is_exclusive() { - return Ok(false); - } - let Some(root) = target.root() else { - return Ok(false); - }; - Ok(self - .storage - .target_holder(authority, root) - .map_err(storage_problem)? - .is_some_and(|holder| &holder != identity)) - } } -fn storage_problem(error: WorkPlacementStorageError) -> ApplicationProblem { - match error { - WorkPlacementStorageError::NotFoundOrNotAuthorized => not_found_problem(), - WorkPlacementStorageError::AuthorityConflict => authority_conflict_problem(), - WorkPlacementStorageError::Unavailable => ApplicationProblem::unavailable(SafeDiagnostic { - code: "application.work-placement.storage-unavailable".to_owned(), - message: "The Work placement authority is unavailable.".to_owned(), - }), +impl From for ApplicationProblem { + fn from(error: WorkPlacementStorageError) -> Self { + match error { + WorkPlacementStorageError::NotFoundOrNotAuthorized => not_found_problem(), + WorkPlacementStorageError::AuthorityConflict => authority_conflict_problem(), + WorkPlacementStorageError::Unavailable => { + ApplicationProblem::unavailable(SafeDiagnostic { + code: "application.work-placement.storage-unavailable".to_owned(), + message: "The Work placement authority is unavailable.".to_owned(), + }) + } + } } } diff --git a/crates/tracedecay-contracts/tests/contracts_suite/work_placement_service.rs b/crates/tracedecay-contracts/tests/contracts_suite/work_placement_service.rs index d8c744b159..1ddba8cfea 100644 --- a/crates/tracedecay-contracts/tests/contracts_suite/work_placement_service.rs +++ b/crates/tracedecay-contracts/tests/contracts_suite/work_placement_service.rs @@ -20,8 +20,8 @@ use tracedecay_contracts::{ AdmitWorkPlacementCommand, ApplicationProblem, ApplicationProblemKind, CancellationContext, CapabilityGrantSnapshot, Deadline, DisclosureClass, ReleaseWorkPlacementCommand, RequestContext, RequestId, ResolvedScope, WorkPlacementPreflightRequestV1, - WorkPlacementReadingV1, WorkPlacementService, WorkPlacementStatusRequestV1, - WorkPlacementStorageError, WorkPlacementStoragePort, + WorkPlacementPublicationPort, WorkPlacementReadingV1, WorkPlacementService, + WorkPlacementStatusRequestV1, WorkPlacementStorageError, WorkPlacementStoragePort, }; use tracedecay_domain::{ ActorId, ProjectId, RepositoryId, RunId, TaskId, UtcMicros, WorkAuthority, @@ -146,7 +146,9 @@ impl WorkPlacementStoragePort for TestStore { }) .map(|((_, identity), _)| identity.clone())) } +} +impl WorkPlacementPublicationPort for TestStore { fn publish_placement( &self, authority: &WorkAuthority, diff --git a/crates/tracedecay-contracts/tests/contracts_suite/work_topology_view.rs b/crates/tracedecay-contracts/tests/contracts_suite/work_topology_view.rs index 8069b128b3..217cfe4c55 100644 --- a/crates/tracedecay-contracts/tests/contracts_suite/work_topology_view.rs +++ b/crates/tracedecay-contracts/tests/contracts_suite/work_topology_view.rs @@ -14,11 +14,11 @@ use tracedecay_contracts::{ Deadline, DisclosureClass, ExecutionTopologyViewV1, GenerateProposalRequest, RequestContext, RequestId, ResolvedScope, StartWorkAttemptCommand, WorkAttemptListCoverageV1, WorkAttemptService, WorkAttemptTopologyBindingV1, WorkAttemptTopologyStateV1, - WorkIntelligenceServiceV1, WorkPlacementReadingV1, WorkPlacementService, - WorkPlacementStorageError, WorkPlacementStoragePort, WorkProductAttemptServiceV1, - WorkProductAuthorizedRelationScopeV1, WorkProductSelectionScopeV1, WorkRoutingSnapshotErrorV1, - WorkRoutingSnapshotPortV1, WorkRoutingSnapshotV1, WorkTopologyViewRequestV1, - execution_topology_view, + WorkIntelligenceServiceV1, WorkPlacementPublicationPort, WorkPlacementReadingV1, + WorkPlacementService, WorkPlacementStorageError, WorkPlacementStoragePort, + WorkProductAttemptServiceV1, WorkProductAuthorizedRelationScopeV1, WorkProductSelectionScopeV1, + WorkRoutingSnapshotErrorV1, WorkRoutingSnapshotPortV1, WorkRoutingSnapshotV1, + WorkTopologyViewRequestV1, execution_topology_view, }; use tracedecay_domain::configuration::safe_work_topology_policy_v1; use tracedecay_domain::{ @@ -122,7 +122,9 @@ impl WorkPlacementStoragePort for PlacementStore { }) .map(|((_, identity), _)| identity.clone())) } +} +impl WorkPlacementPublicationPort for PlacementStore { fn publish_placement( &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 d168845c89..f41a54f89c 100644 --- a/crates/tracedecay-daemon-service/src/invocation/work/request_dispatch.rs +++ b/crates/tracedecay-daemon-service/src/invocation/work/request_dispatch.rs @@ -979,27 +979,27 @@ pub(super) async fn dispatch_work_application( } } WorkApplicationInvocationV1::AdmitPlacement(command) => { - let _span = tracing::trace_span!("daemon.service.work.admit_placement").entered(); - { - let placement_root = project_root.clone(); - complete_work_effect( - ®istered, - request_id, - &context, - canonical_request_id, - operation_key, - use_case, - input_digest, - services - .placement() - .admit_placement(&context, command, |target| { - observe_placement_target(placement_root.as_deref(), target, observed_at) - }), - observed_at, - deadline, - WorkApplicationOutcomeV1::AdmitPlacement, - ) - } + let placement_root = project_root.clone(); + let admitted = tracing::Instrument::instrument( + services.admit_placement(context.clone(), command, |target| { + observe_placement_target(placement_root.as_deref(), target, observed_at) + }), + tracing::trace_span!("daemon.service.work.admit_placement"), + ) + .await; + complete_work_effect( + ®istered, + request_id, + &context, + canonical_request_id, + operation_key, + use_case, + input_digest, + admitted, + observed_at, + deadline, + WorkApplicationOutcomeV1::AdmitPlacement, + ) } WorkApplicationInvocationV1::PlacementStatus(request) => { let _span = tracing::trace_span!("daemon.service.work.placement_status").entered(); @@ -1020,25 +1020,27 @@ pub(super) async fn dispatch_work_application( } } WorkApplicationInvocationV1::ReleasePlacement(command) => { - let _span = tracing::trace_span!("daemon.service.work.release_placement").entered(); - { - let placement_root = project_root.clone(); - complete_work_effect( - ®istered, - request_id, - &context, - canonical_request_id, - operation_key, - use_case, - input_digest, - services.placement().release(&context, command, |target| { - observe_placement_target(placement_root.as_deref(), target, observed_at) - }), - observed_at, - deadline, - WorkApplicationOutcomeV1::ReleasePlacement, - ) - } + let placement_root = project_root.clone(); + let released = tracing::Instrument::instrument( + services.release_placement(context.clone(), command, |target| { + observe_placement_target(placement_root.as_deref(), target, observed_at) + }), + tracing::trace_span!("daemon.service.work.release_placement"), + ) + .await; + complete_work_effect( + ®istered, + request_id, + &context, + canonical_request_id, + operation_key, + use_case, + input_digest, + released, + observed_at, + deadline, + WorkApplicationOutcomeV1::ReleasePlacement, + ) } WorkApplicationInvocationV1::AdvanceJob(request) => { let advanced = match preparation::repository_selection(&context) { diff --git a/crates/tracedecay-rusqlite-runtime/src/work_placement.rs b/crates/tracedecay-rusqlite-runtime/src/work_placement.rs index 9b1a4d96fd..5539520803 100644 --- a/crates/tracedecay-rusqlite-runtime/src/work_placement.rs +++ b/crates/tracedecay-rusqlite-runtime/src/work_placement.rs @@ -8,17 +8,73 @@ //! produces a *typed* refusal; the index is what makes the rule survive a crash //! between the read and the write. -use tracedecay_contracts::{WorkPlacementStorageError, WorkPlacementStoragePort}; +use std::time::Duration; + +use tracedecay_contracts::{ + WorkPlacementPublicationPort, WorkPlacementStorageError, WorkPlacementStoragePort, +}; use tracedecay_domain::{ ProjectId, RepositoryId, RunId, TaskId, WorkAuthority, WorkPlacementIdentityV1, WorkPlacementKindV1, WorkPlacementStateV1, WorkPlacementV1, }; -use crate::exact_sql::ExactSqlValue; +use crate::exact_sql::{ExactSqlRows, ExactSqlStatement, ExactSqlValue}; use crate::work::{ - WorkSqliteStorage, authority_params_owned, exact_sql_statement, exact_sql_text, - registered_work_query, + RegisteredWorkQuery, RegisteredWorkWrite, WorkSqliteStorage, authority_params_owned, + exact_sql_statement, exact_sql_text, registered_work_query, }; +use crate::work_run_control::WorkSqliteJobStorage; + +impl WorkSqliteStorage { + /// Reads one run's placement on a reader; the awaiting task parks no + /// thread. + pub async fn load_placement_async( + &self, + authority: &WorkAuthority, + identity: &WorkPlacementIdentityV1, + ) -> Result, WorkPlacementStorageError> { + let rows = self + .handle() + .query_async( + load_placement_statement(authority, identity)?, + Duration::from_secs(5), + ) + .await + .map_err(|_| WorkPlacementStorageError::Unavailable)?; + placement_from_rows(&rows) + } +} + +impl WorkPlacementStoragePort for WorkSqliteJobStorage<'_> { + fn load_placement( + &self, + authority: &WorkAuthority, + identity: &WorkPlacementIdentityV1, + ) -> Result, WorkPlacementStorageError> { + load_placement_from(self.transaction, authority, identity) + } + + fn target_holder( + &self, + authority: &WorkAuthority, + root: &str, + ) -> Result, WorkPlacementStorageError> { + target_holder_from(self.transaction, authority, root) + } +} + +/// The job's transaction commits the publication, and rolls back a refused +/// compare-and-swap. +impl WorkPlacementPublicationPort for WorkSqliteJobStorage<'_> { + fn publish_placement( + &self, + authority: &WorkAuthority, + expected: Option, + next: &WorkPlacementV1, + ) -> Result<(), WorkPlacementStorageError> { + publish_placement_tx(self.transaction, authority, expected, next) + } +} impl WorkPlacementStoragePort for WorkSqliteStorage { fn load_placement( @@ -26,28 +82,7 @@ impl WorkPlacementStoragePort for WorkSqliteStorage { authority: &WorkAuthority, identity: &WorkPlacementIdentityV1, ) -> Result, WorkPlacementStorageError> { - let rows = registered_work_query( - self.handle(), - "SELECT placement_payload FROM work_placements_v1 - WHERE project_id = ?1 AND repository_id = ?2 AND worktree_id = ?3 - AND actor_id = ?4 AND policy_digest = ?5 - AND task_id = ?6 AND run_id = ?7", - authority_params_owned(authority) - .into_iter() - .chain(identity_params(identity)) - .collect(), - ) - .map_err(|_| WorkPlacementStorageError::Unavailable)?; - let Some(payload) = rows - .rows - .first() - .and_then(|row| exact_sql_text(&row.values, 0)) - else { - return Ok(None); - }; - serde_json::from_str(payload) - .map(Some) - .map_err(|_| WorkPlacementStorageError::Unavailable) + load_placement_from(self.handle(), authority, identity) } fn target_holder( @@ -55,30 +90,7 @@ impl WorkPlacementStoragePort for WorkSqliteStorage { authority: &WorkAuthority, root: &str, ) -> Result, WorkPlacementStorageError> { - let rows = registered_work_query( - self.handle(), - "SELECT task_id, run_id FROM work_placements_v1 - WHERE project_id = ?1 AND repository_id = ?2 AND worktree_id = ?3 - AND actor_id = ?4 AND policy_digest = ?5 - AND target_root = ?6 AND state IN ('admitted', 'quarantined')", - authority_params_owned(authority) - .into_iter() - .chain([ExactSqlValue::Text(root.to_owned())]) - .collect(), - ) - .map_err(|_| WorkPlacementStorageError::Unavailable)?; - let Some(row) = rows.rows.first() else { - return Ok(None); - }; - let task_id = - exact_sql_text(&row.values, 0).ok_or(WorkPlacementStorageError::Unavailable)?; - let run_id = - exact_sql_text(&row.values, 1).ok_or(WorkPlacementStorageError::Unavailable)?; - let task_id = - TaskId::new(task_id.to_owned()).map_err(|_| WorkPlacementStorageError::Unavailable)?; - let run_id = - RunId::new(run_id.to_owned()).map_err(|_| WorkPlacementStorageError::Unavailable)?; - Ok(Some(WorkPlacementIdentityV1::new(task_id, run_id))) + target_holder_from(self.handle(), authority, root) } fn has_target_holder_in_exact_repository_root( @@ -102,34 +114,134 @@ impl WorkPlacementStoragePort for WorkSqliteStorage { .map_err(|_| WorkPlacementStorageError::Unavailable)?; Ok(!rows.rows.is_empty()) } +} - fn publish_placement( - &self, - authority: &WorkAuthority, - expected: Option, - next: &WorkPlacementV1, - ) -> Result<(), WorkPlacementStorageError> { - let payload = - serde_json::to_string(next).map_err(|_| WorkPlacementStorageError::Unavailable)?; - let authority_version = i64::try_from(next.authority_version()) - .map_err(|_| WorkPlacementStorageError::Unavailable)?; - let target_root = next.target().root().map_or(ExactSqlValue::Null, |root| { - ExactSqlValue::Text(root.to_owned()) - }); - let transaction = self - .handle() - .begin_immediate() - .map_err(|_| WorkPlacementStorageError::Unavailable)?; +fn load_placement_statement( + authority: &WorkAuthority, + identity: &WorkPlacementIdentityV1, +) -> Result { + exact_sql_statement( + "SELECT placement_payload FROM work_placements_v1 + WHERE project_id = ?1 AND repository_id = ?2 AND worktree_id = ?3 + AND actor_id = ?4 AND policy_digest = ?5 + AND task_id = ?6 AND run_id = ?7", + authority_params_owned(authority) + .into_iter() + .chain(identity_params(identity)) + .collect(), + ) + .map_err(|_| WorkPlacementStorageError::Unavailable) +} - let changed = match expected { - None => transaction - .execute( - exact_sql_statement( - "INSERT OR IGNORE INTO work_placements_v1 ( +fn placement_from_rows( + rows: &ExactSqlRows, +) -> Result, WorkPlacementStorageError> { + let Some(payload) = rows + .rows + .first() + .and_then(|row| exact_sql_text(&row.values, 0)) + else { + return Ok(None); + }; + serde_json::from_str(payload) + .map(Some) + .map_err(|_| WorkPlacementStorageError::Unavailable) +} + +fn load_placement_from( + source: &impl RegisteredWorkQuery, + authority: &WorkAuthority, + identity: &WorkPlacementIdentityV1, +) -> Result, WorkPlacementStorageError> { + let rows = source + .work_query(load_placement_statement(authority, identity)?) + .map_err(|_| WorkPlacementStorageError::Unavailable)?; + placement_from_rows(&rows) +} + +fn target_holder_from( + source: &impl RegisteredWorkQuery, + authority: &WorkAuthority, + root: &str, +) -> Result, WorkPlacementStorageError> { + let rows = registered_work_query( + source, + "SELECT task_id, run_id FROM work_placements_v1 + WHERE project_id = ?1 AND repository_id = ?2 AND worktree_id = ?3 + AND actor_id = ?4 AND policy_digest = ?5 + AND target_root = ?6 AND state IN ('admitted', 'quarantined')", + authority_params_owned(authority) + .into_iter() + .chain([ExactSqlValue::Text(root.to_owned())]) + .collect(), + ) + .map_err(|_| WorkPlacementStorageError::Unavailable)?; + let Some(row) = rows.rows.first() else { + return Ok(None); + }; + let task_id = exact_sql_text(&row.values, 0).ok_or(WorkPlacementStorageError::Unavailable)?; + let run_id = exact_sql_text(&row.values, 1).ok_or(WorkPlacementStorageError::Unavailable)?; + let task_id = + TaskId::new(task_id.to_owned()).map_err(|_| WorkPlacementStorageError::Unavailable)?; + let run_id = + RunId::new(run_id.to_owned()).map_err(|_| WorkPlacementStorageError::Unavailable)?; + Ok(Some(WorkPlacementIdentityV1::new(task_id, run_id))) +} + +fn publish_placement_tx( + transaction: &impl RegisteredWorkWrite, + authority: &WorkAuthority, + expected: Option, + next: &WorkPlacementV1, +) -> Result<(), WorkPlacementStorageError> { + let payload = + serde_json::to_string(next).map_err(|_| WorkPlacementStorageError::Unavailable)?; + let authority_version = i64::try_from(next.authority_version()) + .map_err(|_| WorkPlacementStorageError::Unavailable)?; + let target_root = next.target().root().map_or(ExactSqlValue::Null, |root| { + ExactSqlValue::Text(root.to_owned()) + }); + + let changed = match expected { + None => transaction + .work_execute( + exact_sql_statement( + "INSERT OR IGNORE INTO work_placements_v1 ( project_id, repository_id, worktree_id, actor_id, policy_digest, task_id, run_id, kind, target_root, state, authority_version, placement_payload ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)", + authority_params_owned(authority) + .into_iter() + .chain(identity_params(next.identity())) + .chain([ + ExactSqlValue::Text(kind_text(next.target().kind())), + target_root, + ExactSqlValue::Text(state_text(next.state())), + ExactSqlValue::Integer(authority_version), + ExactSqlValue::Text(payload), + ]) + .collect(), + ) + .map_err(|_| WorkPlacementStorageError::Unavailable)?, + ) + // `INSERT OR IGNORE` also absorbs the exclusivity index + // violation, so a second holder of the same root lands in the + // same typed conflict as a racing first admission. + .map_err(|_| WorkPlacementStorageError::Unavailable)?, + Some(expected) => { + let expected_version = + i64::try_from(expected).map_err(|_| WorkPlacementStorageError::Unavailable)?; + transaction + .work_execute( + exact_sql_statement( + "UPDATE work_placements_v1 SET + kind = ?8, target_root = ?9, state = ?10, + authority_version = ?11, placement_payload = ?12 + WHERE project_id = ?1 AND repository_id = ?2 AND worktree_id = ?3 + AND actor_id = ?4 AND policy_digest = ?5 + AND task_id = ?6 AND run_id = ?7 + AND authority_version = ?13", authority_params_owned(authority) .into_iter() .chain(identity_params(next.identity())) @@ -139,55 +251,19 @@ impl WorkPlacementStoragePort for WorkSqliteStorage { ExactSqlValue::Text(state_text(next.state())), ExactSqlValue::Integer(authority_version), ExactSqlValue::Text(payload), + ExactSqlValue::Integer(expected_version), ]) .collect(), ) .map_err(|_| WorkPlacementStorageError::Unavailable)?, ) - // `INSERT OR IGNORE` also absorbs the exclusivity index - // violation, so a second holder of the same root lands in the - // same typed conflict as a racing first admission. - .map_err(|_| WorkPlacementStorageError::Unavailable)?, - Some(expected) => { - let expected_version = - i64::try_from(expected).map_err(|_| WorkPlacementStorageError::Unavailable)?; - transaction - .execute( - exact_sql_statement( - "UPDATE work_placements_v1 SET - kind = ?8, target_root = ?9, state = ?10, - authority_version = ?11, placement_payload = ?12 - WHERE project_id = ?1 AND repository_id = ?2 AND worktree_id = ?3 - AND actor_id = ?4 AND policy_digest = ?5 - AND task_id = ?6 AND run_id = ?7 - AND authority_version = ?13", - authority_params_owned(authority) - .into_iter() - .chain(identity_params(next.identity())) - .chain([ - ExactSqlValue::Text(kind_text(next.target().kind())), - target_root, - ExactSqlValue::Text(state_text(next.state())), - ExactSqlValue::Integer(authority_version), - ExactSqlValue::Text(payload), - ExactSqlValue::Integer(expected_version), - ]) - .collect(), - ) - .map_err(|_| WorkPlacementStorageError::Unavailable)?, - ) - .map_err(|_| WorkPlacementStorageError::Unavailable)? - } - }; - if changed.changed_rows != 1 { - let _ = transaction.rollback(); - return Err(WorkPlacementStorageError::AuthorityConflict); + .map_err(|_| WorkPlacementStorageError::Unavailable)? } - transaction - .commit() - .map_err(|_| WorkPlacementStorageError::Unavailable)?; - Ok(()) + }; + if changed.changed_rows != 1 { + return Err(WorkPlacementStorageError::AuthorityConflict); } + Ok(()) } fn identity_params(identity: &WorkPlacementIdentityV1) -> [ExactSqlValue; 2] { diff --git a/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/work_placement_storage.rs b/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/work_placement_storage.rs index b8eefa8b74..0a386e98b9 100644 --- a/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/work_placement_storage.rs +++ b/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/work_placement_storage.rs @@ -12,7 +12,9 @@ use crate::work_registered_store; use std::collections::BTreeSet; -use tracedecay_contracts::{WorkPlacementStorageError, WorkPlacementStoragePort}; +use tracedecay_contracts::{ + WorkPlacementPublicationPort, WorkPlacementStorageError, WorkPlacementStoragePort, +}; use tracedecay_domain::{ ActorId, ProjectId, RepositoryId, RunId, TaskId, UtcMicros, WorkAuthority, WorkPlacementBlockerV1, WorkPlacementIdentityV1, WorkPlacementKindV1, @@ -45,6 +47,23 @@ fn authority_in_worktree_with_policy(actor: &str, worktree: &str, policy: char) .unwrap() } +/// Publishes `next` in one writer job. +fn publish( + store: &RegisteredWorkStore, + authority: &WorkAuthority, + expected: Option, + next: &WorkPlacementV1, +) -> Result<(), WorkPlacementStorageError> { + let (authority, next) = (authority.clone(), next.clone()); + tracedecay_rusqlite_runtime::native_runtime::block_on( + store + .storage() + .transact(move |storage| storage.publish_placement(&authority, expected, &next)), + ) + .expect("native runtime") + .expect("the writer runs the job") +} + #[test] fn authority_in_worktree_a_targeting_root_b_blocks_cleanup_of_b_across_lineage() { let store = RegisteredWorkStore::start("placement-cleanup-holder-scope"); @@ -60,18 +79,20 @@ fn authority_in_worktree_a_targeting_root_b_blocks_cleanup_of_b_across_lineage() ); let old_root = fixture_abs_root("/workspace/placement-target-b"); let other_root = fixture_abs_root("/workspace/placement-other-actor"); - store - .storage() - .publish_placement(&old_policy, None, &admitted("run.old-policy", &old_root)) - .unwrap(); - store - .storage() - .publish_placement( - &other_actor, - None, - &admitted("run.other-actor", &other_root), - ) - .unwrap(); + publish( + &store, + &old_policy, + None, + &admitted("run.old-policy", &old_root), + ) + .unwrap(); + publish( + &store, + &other_actor, + None, + &admitted("run.other-actor", &other_root), + ) + .unwrap(); for (authority, root) in [ (&old_policy, old_root.as_str()), @@ -154,10 +175,7 @@ fn the_first_admission_inserts_and_a_racing_first_admission_conflicts() { let store = RegisteredWorkStore::start("placement-first"); let authority = authority("actor.placement.first"); let placement = admitted("run.a", &ROOT); - store - .storage() - .publish_placement(&authority, None, &placement) - .unwrap(); + publish(&store, &authority, None, &placement).unwrap(); assert_eq!( store .storage() @@ -170,9 +188,7 @@ fn the_first_admission_inserts_and_a_racing_first_admission_conflicts() { Some(identity("run.a")) ); assert_eq!( - store - .storage() - .publish_placement(&authority, None, &placement) + publish(&store, &authority, None, &placement) .expect_err("a racing first admission conflicts"), WorkPlacementStorageError::AuthorityConflict ); @@ -182,34 +198,28 @@ fn the_first_admission_inserts_and_a_racing_first_admission_conflicts() { fn the_database_refuses_a_second_holder_of_the_same_managed_root() { let store = RegisteredWorkStore::start("placement-exclusive"); let authority = authority("actor.placement.exclusive"); - store - .storage() - .publish_placement(&authority, None, &admitted("run.a", &ROOT)) - .unwrap(); + publish(&store, &authority, None, &admitted("run.a", &ROOT)).unwrap(); // A different run naming the same root is refused by the exclusivity index // even though its own row does not exist yet. assert_eq!( - store - .storage() - .publish_placement(&authority, None, &admitted("run.b", &ROOT)) + publish(&store, &authority, None, &admitted("run.b", &ROOT)) .expect_err("a held root is exclusive"), WorkPlacementStorageError::AuthorityConflict ); assert_eq!(store.count("work_placements_v1"), 1); // A different root is unaffected. - store - .storage() - .publish_placement( - &authority, - None, - &admitted( - "run.c", - &fixture_abs_root("/workspace/placement-storage-other"), - ), - ) - .unwrap(); + publish( + &store, + &authority, + None, + &admitted( + "run.c", + &fixture_abs_root("/workspace/placement-storage-other"), + ), + ) + .unwrap(); assert_eq!(store.count("work_placements_v1"), 2); } @@ -218,10 +228,7 @@ fn a_released_placement_frees_its_root_and_a_quarantined_one_does_not() { let store = RegisteredWorkStore::start("placement-release"); let authority = authority("actor.placement.release"); let placement = admitted("run.a", &ROOT); - store - .storage() - .publish_placement(&authority, None, &placement) - .unwrap(); + publish(&store, &authority, None, &placement).unwrap(); let quarantined = placement .release( @@ -229,23 +236,20 @@ fn a_released_placement_frees_its_root_and_a_quarantined_one_does_not() { UtcMicros(400), ) .unwrap(); - store - .storage() - .publish_placement( - &authority, - Some(placement.authority_version()), - &quarantined, - ) - .unwrap(); + publish( + &store, + &authority, + Some(placement.authority_version()), + &quarantined, + ) + .unwrap(); // Quarantine retains the bytes, so the root is still held. assert_eq!( store.storage().target_holder(&authority, &ROOT).unwrap(), Some(identity("run.a")) ); assert_eq!( - store - .storage() - .publish_placement(&authority, None, &admitted("run.b", &ROOT)) + publish(&store, &authority, None, &admitted("run.b", &ROOT)) .expect_err("a quarantined root is still held"), WorkPlacementStorageError::AuthorityConflict ); @@ -253,20 +257,20 @@ fn a_released_placement_frees_its_root_and_a_quarantined_one_does_not() { let released = quarantined .release(BTreeSet::new(), UtcMicros(600)) .unwrap(); - store - .storage() - .publish_placement(&authority, Some(quarantined.authority_version()), &released) - .unwrap(); + publish( + &store, + &authority, + Some(quarantined.authority_version()), + &released, + ) + .unwrap(); assert_eq!(released.state(), WorkPlacementStateV1::Released); assert_eq!( store.storage().target_holder(&authority, &ROOT).unwrap(), None ); // Only now can another run take it. - store - .storage() - .publish_placement(&authority, None, &admitted("run.b", &ROOT)) - .unwrap(); + publish(&store, &authority, None, &admitted("run.b", &ROOT)).unwrap(); } #[test] @@ -275,25 +279,22 @@ fn a_stale_version_conflicts_and_rows_survive_a_restart_per_authority() { let mine = authority("actor.placement.mine"); let peer = authority("actor.placement.peer"); let placement = admitted("run.a", &ROOT); - store - .storage() - .publish_placement(&mine, None, &placement) - .unwrap(); + publish(&store, &mine, None, &placement).unwrap(); let released = placement.release(BTreeSet::new(), UtcMicros(400)).unwrap(); assert_eq!( - store - .storage() - .publish_placement(&mine, Some(placement.authority_version() + 9), &released) - .expect_err("stale authority version"), + publish( + &store, + &mine, + Some(placement.authority_version() + 9), + &released + ) + .expect_err("stale authority version"), WorkPlacementStorageError::AuthorityConflict ); // Another actor holds nothing here, and the same root is free for it. assert_eq!(store.storage().target_holder(&peer, &ROOT).unwrap(), None); - store - .storage() - .publish_placement(&peer, None, &admitted("run.a", &ROOT)) - .unwrap(); + publish(&store, &peer, None, &admitted("run.a", &ROOT)).unwrap(); let restarted = store.restart("placement-isolation"); assert_eq!(