diff --git a/crates/tracedecay-application/src/work/registered.rs b/crates/tracedecay-application/src/work/registered.rs index 70ba77dc34..b5ebb2ef3b 100644 --- a/crates/tracedecay-application/src/work/registered.rs +++ b/crates/tracedecay-application/src/work/registered.rs @@ -283,6 +283,142 @@ 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(), + ) + }) + } + + /// 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()) + }) + } + + /// 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< @@ -422,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/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..4c931d8eba --- /dev/null +++ b/crates/tracedecay-application/tests/application_suite/work_leaf_jobs.rs @@ -0,0 +1,451 @@ +use std::sync::{Arc, Mutex}; +use tracedecay_application::work::RegisteredWorkApplicationServicesV1; + +use tracedecay_contracts::{ + AdjudicateWorkLeakCommandV1, AdmitWorkPlacementCommand, ApplicationProblem, + ReleaseWorkPlacementCommand, VerifiedWorkLeakEvidenceV1, WorkAttemptStoragePort, + WorkDuplicateAdjudicationAppendOutcomeV1, WorkDuplicateAdjudicationPortV1, + WorkDuplicateAdjudicationStorageErrorV1, WorkLeakAdjudicationOutcomeV1, + WorkLeakAdjudicationStorageErrorV1, WorkLeakEvidenceErrorV1, WorkLeakEvidencePortV1, + WorkPlacementStorageError, WorkPlacementStoragePort, +}; +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, WorkPlacementIdentityV1, + WorkPlacementKindV1, WorkPlacementObservationV1, WorkPlacementStateV1, WorkPlacementTargetV1, + 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, task}; + +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" + ); +} + +/// 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" + ); +} + +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 7f091f7bd1..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") } @@ -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..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, @@ -466,11 +468,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, @@ -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_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-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-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/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-daemon-service/src/invocation/work/request_dispatch.rs b/crates/tracedecay-daemon-service/src/invocation/work/request_dispatch.rs index 087acc1450..f41a54f89c 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< @@ -978,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(); @@ -1019,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/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/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/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/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 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), ); } 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!(