diff --git a/Cargo.lock b/Cargo.lock index 34cf8bb..284d804 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -26,6 +26,7 @@ dependencies = [ "axum", "axum-extra", "base64", + "crank-artifacts", "crank-community-auth", "crank-community-mcp", "crank-config", @@ -823,6 +824,7 @@ dependencies = [ "crank-mapping", "crank-schema", "crank-test-support", + "rand 0.10.2", "serde", "serde_json", "sha2 0.10.9", diff --git a/apps/admin-api/Cargo.toml b/apps/admin-api/Cargo.toml index 13faa4f..bd33168 100644 --- a/apps/admin-api/Cargo.toml +++ b/apps/admin-api/Cargo.toml @@ -15,11 +15,13 @@ name = "crank-migrate" path = "src/bin/crank-migrate.rs" [dependencies] +async-trait = "0.1" argon2.workspace = true axum.workspace = true axum-extra.workspace = true base64.workspace = true crank-community-auth = { path = "../../crates/crank-community-auth" } +crank-artifacts = { path = "../../crates/crank-artifacts" } crank-config = { path = "../../crates/crank-config" } crank-core = { path = "../../crates/crank-core" } crank-import = { path = "../../crates/crank-import" } @@ -38,14 +40,13 @@ sha2.workspace = true sqlx.workspace = true thiserror.workspace = true time.workspace = true -tokio = { workspace = true, features = ["fs"] } +tokio = { workspace = true, features = ["fs", "time"] } tracing.workspace = true tracing-subscriber.workspace = true url.workspace = true uuid.workspace = true [dev-dependencies] -async-trait = "0.1" crank-community-mcp = { path = "../../crates/crank-community-mcp" } crank-test-support = { path = "../../crates/crank-test-support" } metrics.workspace = true diff --git a/apps/admin-api/src/error.rs b/apps/admin-api/src/error.rs index 465155d..a8904b3 100644 --- a/apps/admin-api/src/error.rs +++ b/apps/admin-api/src/error.rs @@ -611,6 +611,13 @@ impl From for ApiError { "artifact source failed integrity verification", json!({ "error_code": "artifact_source_integrity" }), ), + RegistryError::ArtifactClaimInProgress => Self::conflict_with_context( + "artifact reconciliation is in progress", + json!({ + "error_code": "artifact_claim_in_progress", + "recovery": "retry" + }), + ), RegistryError::InvalidArtifactSource { field } => Self::validation_with_context( "artifact source metadata is invalid", json!({ diff --git a/apps/admin-api/src/lib.rs b/apps/admin-api/src/lib.rs index b68dce2..21f09b4 100644 --- a/apps/admin-api/src/lib.rs +++ b/apps/admin-api/src/lib.rs @@ -5,6 +5,7 @@ pub mod error; pub mod import_guidance; pub mod pool_metrics; pub mod rate_limit; +pub mod reconciliation; pub mod request_context; pub mod routes; pub mod service; diff --git a/apps/admin-api/src/main.rs b/apps/admin-api/src/main.rs index 9d0e37c..7882089 100644 --- a/apps/admin-api/src/main.rs +++ b/apps/admin-api/src/main.rs @@ -4,6 +4,7 @@ use admin_api::{ app::build_app, auth::{AuthSettings, BootstrapAdminConfig}, pool_metrics::spawn_postgres_pool_metrics, + reconciliation::{open_reconciliation_store, spawn_artifact_reconciliation}, service::AdminServiceBuilder, state::AppState, }; @@ -202,6 +203,7 @@ async fn run( let secret_crypto = verified_startup_secret_crypto(®istry, config.runtime.master_key.expose_secret()) .await?; + let artifact_store = open_reconciliation_store(config.storage_root.clone()).await?; let outbound_http_policy = crank_runtime::OutboundHttpPolicy::try_new_with_limits( config.runtime.outbound.allowed_hosts.clone(), config.runtime.outbound.denied_hosts.clone(), @@ -216,7 +218,7 @@ async fn run( let identity_provider = PasswordIdentityProvider::new(registry.clone(), auth_settings.password_pepper.clone()); let service = AdminServiceBuilder::new( - registry, + registry.clone(), config.storage_root.clone(), auth_settings, secret_crypto, @@ -229,6 +231,7 @@ async fn run( if config.demo_seed { service.seed_demo_assets().await?; } + spawn_artifact_reconciliation(registry, artifact_store).await; spawn_invocation_log_cleanup(service.clone(), config.invocation_log_retention_days); let state = AppState { service, diff --git a/apps/admin-api/src/reconciliation.rs b/apps/admin-api/src/reconciliation.rs new file mode 100644 index 0000000..6dc5c91 --- /dev/null +++ b/apps/admin-api/src/reconciliation.rs @@ -0,0 +1,783 @@ +use std::{ + path::PathBuf, + time::{Duration, SystemTime}, +}; + +use async_trait::async_trait; +use crank_artifacts::{ + ArtifactError, ArtifactStore, ReconciliationCursor, ReconciliationMutation, + ReconciliationNamespace, ReconciliationPresence, ReconciliationRegistration, + ReconciliationScanStop, +}; +use crank_registry::{ + ArtifactClaimFinalization, ArtifactClaimFinalizeOutcome, ArtifactClaimOutcome, + ArtifactClaimRecheckOutcome, ArtifactClaimToken, ArtifactExpiredClaimProbe, + ClaimArtifactReconciliationRequest, ClaimExpiredArtifactReconciliationRequest, + PostgresRegistry, +}; +use time::OffsetDateTime; +use tokio::time::MissedTickBehavior; +use tracing::{info, warn}; + +pub const RECONCILIATION_INTERVAL: Duration = Duration::from_secs(15 * 60); +pub const RECONCILIATION_GRACE: Duration = Duration::from_secs(24 * 60 * 60); +pub const RECONCILIATION_LEASE: Duration = Duration::from_secs(5 * 60); +pub const RECONCILIATION_TRAVERSAL_LIMIT: usize = 4_096; +pub const RECONCILIATION_CANDIDATE_LIMIT: usize = 32; +pub const RECONCILIATION_MUTATION_LIMIT: usize = 32; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum Phase { + Recovery, + Sweep, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +struct StepLimits { + traversal_syscalls: usize, + candidates: usize, + mutations: usize, +} + +#[derive(Debug)] +struct Step { + cursor: Option, + scan_complete: bool, + traversal_syscalls: usize, + candidates: usize, + mutation_attempts: usize, + classifications: ReconciliationClassificationCounters, + physical_mutation_possible: bool, + retryable: bool, +} + +/// Bounded, identity-free classifications observed during one or more scans. +/// +/// The fixed fields are the complete telemetry vocabulary: no digest, path, +/// token, source or workspace value can be attached to an item classification. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct ReconciliationClassificationCounters { + pub scanner_malformed: usize, + pub scanner_unsafe: usize, + pub recovery_present: usize, + pub recovery_safety: usize, + pub recovery_retryable: usize, + pub registration_integrity: usize, + pub registration_safety: usize, + pub registration_retryable: usize, + pub claim_metadata_conflict: usize, + pub mutation_integrity: usize, + pub mutation_safety: usize, + pub mutation_retryable: usize, +} + +impl ReconciliationClassificationCounters { + fn checked_add(self, other: Self) -> Option { + Some(Self { + scanner_malformed: self + .scanner_malformed + .checked_add(other.scanner_malformed)?, + scanner_unsafe: self.scanner_unsafe.checked_add(other.scanner_unsafe)?, + recovery_present: self.recovery_present.checked_add(other.recovery_present)?, + recovery_safety: self.recovery_safety.checked_add(other.recovery_safety)?, + recovery_retryable: self + .recovery_retryable + .checked_add(other.recovery_retryable)?, + registration_integrity: self + .registration_integrity + .checked_add(other.registration_integrity)?, + registration_safety: self + .registration_safety + .checked_add(other.registration_safety)?, + registration_retryable: self + .registration_retryable + .checked_add(other.registration_retryable)?, + claim_metadata_conflict: self + .claim_metadata_conflict + .checked_add(other.claim_metadata_conflict)?, + mutation_integrity: self + .mutation_integrity + .checked_add(other.mutation_integrity)?, + mutation_safety: self.mutation_safety.checked_add(other.mutation_safety)?, + mutation_retryable: self + .mutation_retryable + .checked_add(other.mutation_retryable)?, + }) + } + + fn record_registration_error(&mut self, error: ArtifactError) -> bool { + match error { + ArtifactError::Integrity + | ArtifactError::EmptySource + | ArtifactError::SourceTooLarge => { + self.registration_integrity += 1; + false + } + ArtifactError::UnsafeRoot + | ArtifactError::InvalidReference + | ArtifactError::NotFound => { + self.registration_safety += 1; + false + } + ArtifactError::Storage => { + self.registration_retryable += 1; + true + } + } + } + + fn record_mutation_error(&mut self, error: ArtifactError) -> bool { + match error { + ArtifactError::Integrity + | ArtifactError::EmptySource + | ArtifactError::SourceTooLarge => self.mutation_integrity += 1, + ArtifactError::UnsafeRoot + | ArtifactError::InvalidReference + | ArtifactError::NotFound => self.mutation_safety += 1, + ArtifactError::Storage => self.mutation_retryable += 1, + } + true + } + + fn record_recovery_presence(&mut self, presence: ReconciliationPresence) -> bool { + match presence { + ReconciliationPresence::Final | ReconciliationPresence::Quarantine => { + self.recovery_present += 1; + false + } + ReconciliationPresence::Unsafe => { + self.recovery_safety += 1; + false + } + ReconciliationPresence::Retryable => { + self.recovery_retryable += 1; + true + } + ReconciliationPresence::Absent => false, + } + } + + fn record_mutation_presence(&mut self, presence: ReconciliationPresence) -> bool { + match presence { + ReconciliationPresence::Absent => false, + ReconciliationPresence::Unsafe => { + self.mutation_safety += 1; + true + } + ReconciliationPresence::Final + | ReconciliationPresence::Quarantine + | ReconciliationPresence::Retryable => { + self.mutation_retryable += 1; + true + } + } + } +} + +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct ReconciliationTickReport { + pub traversal_syscalls: usize, + pub candidates: usize, + pub mutation_attempts: usize, + pub classifications: ReconciliationClassificationCounters, + pub recovery_completed: bool, + pub cycle_completed: bool, + pub retryable: bool, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, thiserror::Error)] +pub enum ReconciliationCoordinatorError { + #[error("artifact reconciliation backend is unavailable")] + Backend, + #[error("artifact reconciliation backend exceeded its bounded contract")] + Bounds, +} + +#[async_trait] +trait ReconciliationBackend: Send + Sync { + type Cursor: Send; + + async fn step( + &self, + phase: Phase, + cursor: Option, + limits: StepLimits, + ) -> Result, ReconciliationCoordinatorError>; +} + +struct ReconciliationCoordinator { + backend: B, + phase: Phase, + cursor: Option, + tick_limits: StepLimits, +} + +#[derive(Clone)] +struct PostgresReconciliationBackend { + registry: PostgresRegistry, + store: ArtifactStore, +} + +enum PostgresReconciliationCursor { + ExpiredClaim(ArtifactExpiredClaimProbe), + Filesystem(ReconciliationCursor), +} + +impl std::fmt::Debug for PostgresReconciliationCursor { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("PostgresReconciliationCursor(..)") + } +} + +impl PostgresReconciliationBackend { + fn new(registry: PostgresRegistry, store: ArtifactStore) -> Self { + Self { registry, store } + } + + async fn database_now(&self) -> Result { + self.registry + .artifact_reconciliation_now() + .await + .map_err(|_| ReconciliationCoordinatorError::Backend) + } + + async fn presence( + &self, + artifact_ref: crank_artifacts::ArtifactRef, + ) -> Result { + let store = self.store.clone(); + let result = + tokio::task::spawn_blocking(move || store.reconciliation_presence(&artifact_ref)) + .await + .map_err(|_| ReconciliationCoordinatorError::Backend)?; + Ok(match result { + Ok(presence) => presence, + Err(ArtifactError::Storage) => ReconciliationPresence::Retryable, + Err(_) => ReconciliationPresence::Unsafe, + }) + } +} + +fn reconciliation_lease_expires_at(claimed_at: OffsetDateTime) -> OffsetDateTime { + claimed_at + + time::Duration::seconds( + i64::try_from(RECONCILIATION_LEASE.as_secs()) + .expect("the fixed reconciliation lease fits i64"), + ) +} + +impl ReconciliationCoordinator { + fn new(backend: B) -> Self { + Self { + backend, + phase: Phase::Recovery, + cursor: None, + tick_limits: StepLimits { + traversal_syscalls: RECONCILIATION_TRAVERSAL_LIMIT, + candidates: RECONCILIATION_CANDIDATE_LIMIT, + mutations: RECONCILIATION_MUTATION_LIMIT, + }, + } + } + + #[cfg(test)] + fn with_tick_limits(backend: B, tick_limits: StepLimits) -> Self { + Self { + backend, + phase: Phase::Recovery, + cursor: None, + tick_limits, + } + } + + async fn tick(&mut self) -> Result { + let mut report = ReconciliationTickReport::default(); + + loop { + let limits = StepLimits { + traversal_syscalls: self + .tick_limits + .traversal_syscalls + .saturating_sub(report.traversal_syscalls), + candidates: self + .tick_limits + .candidates + .saturating_sub(report.candidates), + mutations: self + .tick_limits + .mutations + .saturating_sub(report.mutation_attempts), + }; + if limits.traversal_syscalls == 0 || limits.candidates == 0 || limits.mutations == 0 { + break; + } + + let step = match self + .backend + .step(self.phase, self.cursor.take(), limits) + .await + { + Ok(step) => step, + Err(error) => { + self.cursor = None; + self.phase = Phase::Recovery; + return Err(error); + } + }; + if step.traversal_syscalls > limits.traversal_syscalls + || step.candidates > limits.candidates + || step.mutation_attempts > limits.mutations + || step.mutation_attempts > step.candidates + { + self.cursor = None; + self.phase = Phase::Recovery; + return Err(ReconciliationCoordinatorError::Bounds); + } + + report.traversal_syscalls += step.traversal_syscalls; + report.candidates += step.candidates; + report.mutation_attempts += step.mutation_attempts; + report.classifications = report + .classifications + .checked_add(step.classifications) + .ok_or_else(|| { + self.cursor = None; + self.phase = Phase::Recovery; + ReconciliationCoordinatorError::Bounds + })?; + report.retryable |= step.retryable; + + let made_progress = + step.traversal_syscalls != 0 || step.candidates != 0 || step.mutation_attempts != 0; + if step.physical_mutation_possible { + // A cursor is valid only while the namespace is unchanged. A + // possible rename/unlink includes ambiguous fsync outcomes. + drop(step.cursor); + self.cursor = None; + self.phase = Phase::Recovery; + } else { + self.cursor = step.cursor; + if step.scan_complete { + self.cursor = None; + match self.phase { + Phase::Recovery => { + report.recovery_completed = true; + self.phase = Phase::Sweep; + } + Phase::Sweep => { + report.cycle_completed = true; + self.phase = Phase::Recovery; + break; + } + } + } + } + + if step.retryable || (!made_progress && !step.scan_complete) { + break; + } + } + + Ok(report) + } +} + +#[async_trait] +impl ReconciliationBackend for PostgresReconciliationBackend { + type Cursor = PostgresReconciliationCursor; + + async fn step( + &self, + phase: Phase, + cursor: Option, + limits: StepLimits, + ) -> Result, ReconciliationCoordinatorError> { + let mut recovered_candidates = 0; + let mut classifications = ReconciliationClassificationCounters::default(); + let mut recovery_retryable = false; + let mut filesystem_cursor = None; + if phase == Phase::Recovery { + let after = match cursor { + Some(PostgresReconciliationCursor::ExpiredClaim(probe)) => Some(probe), + Some(PostgresReconciliationCursor::Filesystem(cursor)) => { + filesystem_cursor = Some(cursor); + None + } + None => None, + }; + if filesystem_cursor.is_none() { + let database_now = self.database_now().await?; + let probe_limit = u32::try_from(limits.candidates).unwrap_or(u32::MAX); + let probes = self + .registry + .list_expired_artifact_reconciliation_probes_after( + database_now, + after.as_ref(), + probe_limit, + ) + .await + .map_err(|_| ReconciliationCoordinatorError::Backend)?; + let page_full = probes.len() == probe_limit as usize; + let mut last_processed = after; + for probe in probes { + recovered_candidates += 1; + last_processed = Some(probe.clone()); + let presence = self.presence(probe.artifact_ref().clone()).await?; + if presence != ReconciliationPresence::Absent { + recovery_retryable |= classifications.record_recovery_presence(presence); + continue; + } + + let claimed_at = self.database_now().await?; + let claim = self + .registry + .claim_expired_artifact_reconciliation( + &probe, + ClaimExpiredArtifactReconciliationRequest { + token: ArtifactClaimToken::generate(), + claimed_at, + lease_expires_at: reconciliation_lease_expires_at(claimed_at), + }, + ) + .await + .map_err(|_| ReconciliationCoordinatorError::Backend)?; + let Some(claim) = claim else { + continue; + }; + + // Recheck after the DB CAS. A concurrent filesystem put + // does not consult PostgreSQL, so the first probe alone + // cannot prove absence at finalization time. + let presence = self.presence(claim.artifact_ref().clone()).await?; + if presence != ReconciliationPresence::Absent { + classifications.record_recovery_presence(presence); + recovery_retryable = true; + continue; + } + + let recovered_at = self.database_now().await?; + let result = self + .registry + .recover_artifact_reconciliation_claim( + &claim, + ArtifactClaimFinalization::AlreadyAbsent, + recovered_at, + ) + .await; + let recovered = matches!(result, Ok(ArtifactClaimFinalizeOutcome::Unavailable)); + if recovered { + let presence = self.presence(claim.artifact_ref().clone()).await?; + if presence != ReconciliationPresence::Absent { + classifications.record_recovery_presence(presence); + recovery_retryable = true; + } + } + return Ok(Step { + cursor: last_processed.map(PostgresReconciliationCursor::ExpiredClaim), + scan_complete: false, + traversal_syscalls: 0, + candidates: recovered_candidates, + mutation_attempts: 1, + classifications, + physical_mutation_possible: false, + retryable: recovery_retryable || !recovered, + }); + } + + if page_full { + return Ok(Step { + cursor: last_processed.map(PostgresReconciliationCursor::ExpiredClaim), + scan_complete: false, + traversal_syscalls: 0, + candidates: recovered_candidates, + mutation_attempts: 0, + classifications, + physical_mutation_possible: false, + retryable: recovery_retryable, + }); + } + } + } else if let Some(PostgresReconciliationCursor::Filesystem(cursor)) = cursor { + filesystem_cursor = Some(cursor); + } + + let scan_candidate_limit = limits.candidates.saturating_sub(recovered_candidates); + if scan_candidate_limit == 0 { + return Ok(Step { + cursor: filesystem_cursor.map(PostgresReconciliationCursor::Filesystem), + scan_complete: false, + traversal_syscalls: 0, + candidates: recovered_candidates, + mutation_attempts: 0, + classifications, + physical_mutation_possible: false, + retryable: recovery_retryable, + }); + } + let namespace = match phase { + Phase::Recovery => ReconciliationNamespace::Quarantine, + Phase::Sweep => ReconciliationNamespace::Final, + }; + let store = self.store.clone(); + let (mut scan, candidates) = tokio::task::spawn_blocking(move || { + store.scan_reconciliation_namespace( + namespace, + filesystem_cursor, + limits.traversal_syscalls, + scan_candidate_limit, + ) + }) + .await + .map_err(|_| ReconciliationCoordinatorError::Backend)? + .map_err(|_| ReconciliationCoordinatorError::Backend)?; + + let mut step = Step { + cursor: scan + .continuation + .take() + .map(PostgresReconciliationCursor::Filesystem), + scan_complete: scan.stop == ReconciliationScanStop::Complete, + traversal_syscalls: scan.traversal_syscalls, + candidates: recovered_candidates, + mutation_attempts: 0, + classifications: classifications + .checked_add(ReconciliationClassificationCounters { + scanner_malformed: scan.malformed, + scanner_unsafe: scan.unsafe_entries, + ..ReconciliationClassificationCounters::default() + }) + .ok_or(ReconciliationCoordinatorError::Bounds)?, + physical_mutation_possible: false, + retryable: recovery_retryable || scan.stop == ReconciliationScanStop::Retryable, + }; + + for candidate in candidates { + if step.candidates == limits.candidates || step.mutation_attempts == limits.mutations { + break; + } + step.candidates += 1; + let registration_grace = match phase { + Phase::Recovery => Duration::ZERO, + Phase::Sweep => RECONCILIATION_GRACE, + }; + let store = self.store.clone(); + let registration = tokio::task::spawn_blocking(move || { + let registration = store.register_reconciliation( + &candidate, + registration_grace, + SystemTime::now(), + ); + (candidate, registration) + }) + .await + .map_err(|_| ReconciliationCoordinatorError::Backend)?; + let (candidate, registration) = registration; + let artifact = match registration { + Ok(ReconciliationRegistration::Registered(artifact)) => artifact, + Ok(ReconciliationRegistration::NotEligible) => continue, + Err(error) => { + step.retryable |= step.classifications.record_registration_error(error); + continue; + } + }; + + let claimed_at = self.database_now().await?; + let lease_expires_at = reconciliation_lease_expires_at(claimed_at); + let detached_grace = time::Duration::seconds( + i64::try_from(RECONCILIATION_GRACE.as_secs()) + .expect("the fixed reconciliation grace fits i64"), + ); + let claim = self + .registry + .claim_artifact_reconciliation(ClaimArtifactReconciliationRequest { + artifact: &artifact, + token: ArtifactClaimToken::generate(), + claimed_at, + lease_expires_at, + detached_grace, + }) + .await + .map_err(|_| ReconciliationCoordinatorError::Backend)?; + let claim = match claim { + ArtifactClaimOutcome::Claimed(claim) => claim, + ArtifactClaimOutcome::HeldByOther + | ArtifactClaimOutcome::ActiveReference + | ArtifactClaimOutcome::DetachedReferenceInGrace => continue, + ArtifactClaimOutcome::MetadataConflict => { + step.classifications.claim_metadata_conflict += 1; + continue; + } + }; + + let rechecked_at = self.database_now().await?; + let recheck = self + .registry + .recheck_artifact_reconciliation_claim(&claim, rechecked_at, detached_grace) + .await + .map_err(|_| ReconciliationCoordinatorError::Backend)?; + if recheck != ArtifactClaimRecheckOutcome::Mutate { + continue; + } + + step.mutation_attempts += 1; + step.physical_mutation_possible = true; + let store = self.store.clone(); + let mutation = match tokio::task::spawn_blocking(move || match phase { + Phase::Recovery => store.delete_quarantined_reconciliation(candidate), + Phase::Sweep => store.quarantine_reconciliation(candidate), + }) + .await + { + Ok(mutation) => mutation, + Err(_) => Err(ArtifactError::Storage), + }; + let mut observation = match mutation { + Ok(ReconciliationMutation::Quarantined) + | Ok(ReconciliationMutation::AlreadyQuarantined) => { + ArtifactClaimFinalization::Quarantined + } + Ok(ReconciliationMutation::Deleted) => ArtifactClaimFinalization::Deleted, + Ok(ReconciliationMutation::AlreadyAbsent) => { + ArtifactClaimFinalization::AlreadyAbsent + } + Ok(ReconciliationMutation::Retryable) => { + step.classifications.mutation_retryable += 1; + step.retryable = true; + ArtifactClaimFinalization::Retryable + } + Err(error) => { + step.retryable |= step.classifications.record_mutation_error(error); + ArtifactClaimFinalization::Retryable + } + }; + if matches!( + observation, + ArtifactClaimFinalization::Deleted | ArtifactClaimFinalization::AlreadyAbsent + ) { + // `unavailable` means both canonical locations are confirmed + // absent. A concurrent republish can recreate final bytes + // without consulting PostgreSQL, so unlink alone is not proof. + let presence = self.presence(claim.artifact_ref().clone()).await?; + observation = match presence { + ReconciliationPresence::Absent => ArtifactClaimFinalization::AlreadyAbsent, + presence => { + step.retryable |= step.classifications.record_mutation_presence(presence); + ArtifactClaimFinalization::Retryable + } + }; + } + let observed_at = self.database_now().await?; + let finalization = match phase { + Phase::Recovery => { + self.registry + .recover_artifact_reconciliation_claim(&claim, observation, observed_at) + .await + } + Phase::Sweep => { + self.registry + .finalize_artifact_reconciliation_claim(&claim, observation, observed_at) + .await + } + }; + let finalized_as_expected = + match observation { + ArtifactClaimFinalization::Deleted + | ArtifactClaimFinalization::AlreadyAbsent => matches!( + finalization.as_ref(), + Ok(ArtifactClaimFinalizeOutcome::Unavailable) + ), + ArtifactClaimFinalization::Quarantined + | ArtifactClaimFinalization::Retryable => matches!( + finalization.as_ref(), + Ok(ArtifactClaimFinalizeOutcome::Retained) + ), + }; + if !finalized_as_expected { + step.retryable = true; + } + if matches!( + finalization.as_ref(), + Ok(ArtifactClaimFinalizeOutcome::Unavailable) + ) { + let presence = self.presence(claim.artifact_ref().clone()).await?; + step.retryable |= step.classifications.record_mutation_presence(presence); + } + + // The scan cursor and every remaining observation came from the + // pre-mutation namespace. Return immediately so the coordinator + // drops them and restarts a recovery cycle from `None`. + break; + } + + Ok(step) + } +} + +/// Opens the protected root without running filesystem syscalls on Tokio's +/// async executor. +pub async fn open_reconciliation_store( + root: PathBuf, +) -> Result { + tokio::task::spawn_blocking(move || ArtifactStore::open(root)) + .await + .map_err(|_| ReconciliationCoordinatorError::Backend)? + .map_err(|_| ReconciliationCoordinatorError::Backend) +} + +/// Runs the immediate bounded startup cycle, then schedules 15-minute ticks. +/// Recovery remains ahead of the final sweep even when it spans several ticks. +pub async fn spawn_artifact_reconciliation(registry: PostgresRegistry, store: ArtifactStore) { + let backend = PostgresReconciliationBackend::new(registry, store); + let mut coordinator = ReconciliationCoordinator::new(backend); + observe_tick(coordinator.tick().await); + + tokio::spawn(async move { + let mut interval = tokio::time::interval(RECONCILIATION_INTERVAL); + interval.set_missed_tick_behavior(MissedTickBehavior::Skip); + // The immediate tick was run above as part of startup composition. + interval.tick().await; + loop { + interval.tick().await; + observe_tick(coordinator.tick().await); + } + }); +} + +fn observe_tick(result: Result) { + match result { + Ok(report) => info!( + name: "admin.artifact_reconciliation.tick", + traversal_syscalls = report.traversal_syscalls, + candidates = report.candidates, + mutation_attempts = report.mutation_attempts, + scanner_malformed = report.classifications.scanner_malformed, + scanner_unsafe = report.classifications.scanner_unsafe, + recovery_present = report.classifications.recovery_present, + recovery_safety = report.classifications.recovery_safety, + recovery_retryable = report.classifications.recovery_retryable, + registration_integrity = report.classifications.registration_integrity, + registration_safety = report.classifications.registration_safety, + registration_retryable = report.classifications.registration_retryable, + claim_metadata_conflict = report.classifications.claim_metadata_conflict, + mutation_integrity = report.classifications.mutation_integrity, + mutation_safety = report.classifications.mutation_safety, + mutation_retryable = report.classifications.mutation_retryable, + recovery_completed = report.recovery_completed, + cycle_completed = report.cycle_completed, + retryable = report.retryable, + "artifact reconciliation tick completed" + ), + Err(error) => warn!( + name: "admin.artifact_reconciliation.failed", + error_category = match error { + ReconciliationCoordinatorError::Backend => "backend", + ReconciliationCoordinatorError::Bounds => "bounds", + }, + "artifact reconciliation tick will be retried" + ), + } +} + +#[cfg(test)] +#[path = "reconciliation/tests.rs"] +mod tests; diff --git a/apps/admin-api/src/reconciliation/tests.rs b/apps/admin-api/src/reconciliation/tests.rs new file mode 100644 index 0000000..b567c6f --- /dev/null +++ b/apps/admin-api/src/reconciliation/tests.rs @@ -0,0 +1,690 @@ +use std::{ + collections::VecDeque, + fs, + os::unix::fs::{PermissionsExt, symlink}, + path::{Path, PathBuf}, + sync::{ + Arc, Mutex, + atomic::{AtomicUsize, Ordering}, + }, +}; + +use async_trait::async_trait; +use crank_artifacts::{ReconciliationRegistration, RegisteredArtifact}; +use crank_registry::MigrationAuthority; + +use super::*; + +static NEXT_TEST_ROOT: AtomicUsize = AtomicUsize::new(0); + +struct TestRoot(PathBuf); + +impl TestRoot { + fn new() -> Self { + let path = std::env::temp_dir().join(format!( + "crank-admin-reconciliation-{}-{}", + std::process::id(), + NEXT_TEST_ROOT.fetch_add(1, Ordering::Relaxed) + )); + fs::create_dir(&path).unwrap(); + fs::set_permissions(&path, fs::Permissions::from_mode(0o700)).unwrap(); + Self(path) + } +} + +impl Drop for TestRoot { + fn drop(&mut self) { + let _ = fs::set_permissions(&self.0, fs::Permissions::from_mode(0o700)); + let _ = fs::remove_dir_all(&self.0); + } +} + +fn artifact_path(root: &Path, artifact: &RegisteredArtifact) -> PathBuf { + let digest = artifact.artifact_ref().digest_hex(); + root.join("sha256").join(&digest[..2]).join(digest) +} + +fn age_artifact(root: &Path, artifact: &RegisteredArtifact) { + fs::File::open(artifact_path(root, artifact)) + .unwrap() + .set_modified(SystemTime::now() - RECONCILIATION_GRACE - Duration::from_secs(60)) + .unwrap(); +} + +fn remove_registered_artifact(store: &ArtifactStore, artifact: &RegisteredArtifact) { + let (_, candidates) = store + .scan_reconciliation_namespace(ReconciliationNamespace::Final, None, 4_096, 32) + .unwrap(); + let candidate = candidates + .into_iter() + .find(|candidate| { + matches!( + store.register_reconciliation(candidate, Duration::ZERO, SystemTime::now()), + Ok(ReconciliationRegistration::Registered(found)) if found == *artifact + ) + }) + .expect("the final artifact is discoverable"); + assert!(matches!( + store.quarantine_reconciliation(candidate), + Ok(ReconciliationMutation::Quarantined | ReconciliationMutation::AlreadyQuarantined) + )); + + let (_, candidates) = store + .scan_reconciliation_namespace(ReconciliationNamespace::Quarantine, None, 4_096, 32) + .unwrap(); + let candidate = candidates + .into_iter() + .find(|candidate| { + matches!( + store.register_reconciliation(candidate, Duration::ZERO, SystemTime::now()), + Ok(ReconciliationRegistration::Registered(found)) if found == *artifact + ) + }) + .expect("the quarantined artifact is discoverable"); + assert!(matches!( + store.delete_quarantined_reconciliation(candidate), + Ok(ReconciliationMutation::Deleted | ReconciliationMutation::AlreadyAbsent) + )); +} + +#[derive(Clone, Copy)] +struct StepTemplate { + scan_complete: bool, + traversal_syscalls: usize, + candidates: usize, + mutation_attempts: usize, + classifications: ReconciliationClassificationCounters, + physical_mutation_possible: bool, + retryable: bool, + return_cursor: bool, + error: bool, +} + +impl StepTemplate { + fn complete() -> Self { + Self { + scan_complete: true, + traversal_syscalls: 1, + candidates: 0, + mutation_attempts: 0, + classifications: ReconciliationClassificationCounters::default(), + physical_mutation_possible: false, + retryable: false, + return_cursor: false, + error: false, + } + } +} + +struct CursorEvidence { + alive: Arc, +} + +impl CursorEvidence { + fn new(alive: Arc, maximum: &AtomicUsize) -> Self { + let current = alive.fetch_add(1, Ordering::SeqCst) + 1; + maximum.fetch_max(current, Ordering::SeqCst); + Self { alive } + } +} + +impl Drop for CursorEvidence { + fn drop(&mut self) { + self.alive.fetch_sub(1, Ordering::SeqCst); + } +} + +struct ScriptedBackend { + steps: Mutex>, + calls: Mutex>, + alive: Arc, + maximum: AtomicUsize, +} + +impl ScriptedBackend { + fn new(steps: impl IntoIterator) -> Self { + Self { + steps: Mutex::new(steps.into_iter().collect()), + calls: Mutex::new(Vec::new()), + alive: Arc::new(AtomicUsize::new(0)), + maximum: AtomicUsize::new(0), + } + } +} + +#[async_trait] +impl ReconciliationBackend for Arc { + type Cursor = CursorEvidence; + + async fn step( + &self, + phase: Phase, + cursor: Option, + limits: StepLimits, + ) -> Result, ReconciliationCoordinatorError> { + self.calls + .lock() + .unwrap() + .push((phase, cursor.is_some(), limits)); + drop(cursor); + let template = self.steps.lock().unwrap().pop_front().unwrap(); + if template.error { + return Err(ReconciliationCoordinatorError::Backend); + } + let cursor = template + .return_cursor + .then(|| CursorEvidence::new(self.alive.clone(), &self.maximum)); + Ok(Step { + cursor, + scan_complete: template.scan_complete, + traversal_syscalls: template.traversal_syscalls, + candidates: template.candidates, + mutation_attempts: template.mutation_attempts, + classifications: template.classifications, + physical_mutation_possible: template.physical_mutation_possible, + retryable: template.retryable, + }) + } +} + +#[tokio::test] +async fn recovery_always_completes_before_final_sweep() { + let backend = Arc::new(ScriptedBackend::new([ + StepTemplate::complete(), + StepTemplate::complete(), + ])); + let mut coordinator = ReconciliationCoordinator::new(backend.clone()); + + let report = coordinator.tick().await.unwrap(); + + assert!(report.recovery_completed); + assert!(report.cycle_completed); + let phases = backend + .calls + .lock() + .unwrap() + .iter() + .map(|call| call.0) + .collect::>(); + assert_eq!(phases, vec![Phase::Recovery, Phase::Sweep]); +} + +#[tokio::test] +async fn possible_mutation_discards_cursor_and_restarts_recovery() { + let paged = StepTemplate { + scan_complete: false, + traversal_syscalls: 2, + candidates: 1, + mutation_attempts: 0, + classifications: ReconciliationClassificationCounters::default(), + physical_mutation_possible: false, + retryable: true, + return_cursor: true, + error: false, + }; + let mutation = StepTemplate { + scan_complete: false, + traversal_syscalls: 1, + candidates: 1, + mutation_attempts: 1, + classifications: ReconciliationClassificationCounters::default(), + physical_mutation_possible: true, + retryable: false, + return_cursor: true, + error: false, + }; + let backend = Arc::new(ScriptedBackend::new([ + StepTemplate::complete(), + paged, + mutation, + StepTemplate::complete(), + StepTemplate::complete(), + ])); + let mut coordinator = ReconciliationCoordinator::new(backend.clone()); + + let first = coordinator.tick().await.unwrap(); + assert!(first.retryable); + assert_eq!(backend.alive.load(Ordering::SeqCst), 1); + + let second = coordinator.tick().await.unwrap(); + assert!(second.cycle_completed); + let calls = backend.calls.lock().unwrap(); + assert_eq!( + calls + .iter() + .map(|(phase, had_cursor, _)| (*phase, *had_cursor)) + .collect::>(), + vec![ + (Phase::Recovery, false), + (Phase::Sweep, false), + (Phase::Sweep, true), + (Phase::Recovery, false), + (Phase::Sweep, false), + ] + ); + assert_eq!(backend.alive.load(Ordering::SeqCst), 0); + assert_eq!(backend.maximum.load(Ordering::SeqCst), 1); +} + +#[tokio::test] +async fn backend_error_discards_cursor_and_restarts_recovery() { + let paged = StepTemplate { + scan_complete: false, + traversal_syscalls: 1, + candidates: 1, + mutation_attempts: 0, + classifications: ReconciliationClassificationCounters::default(), + physical_mutation_possible: false, + retryable: true, + return_cursor: true, + error: false, + }; + let failure = StepTemplate { + error: true, + ..StepTemplate::complete() + }; + let backend = Arc::new(ScriptedBackend::new([ + StepTemplate::complete(), + paged, + failure, + StepTemplate::complete(), + StepTemplate::complete(), + ])); + let mut coordinator = ReconciliationCoordinator::new(backend.clone()); + + assert!(coordinator.tick().await.unwrap().retryable); + assert_eq!(backend.alive.load(Ordering::SeqCst), 1); + assert_eq!( + coordinator.tick().await, + Err(ReconciliationCoordinatorError::Backend) + ); + assert_eq!(backend.alive.load(Ordering::SeqCst), 0); + assert!(coordinator.tick().await.unwrap().cycle_completed); + + let calls = backend.calls.lock().unwrap(); + assert_eq!( + calls + .iter() + .map(|(phase, had_cursor, _)| (*phase, *had_cursor)) + .collect::>(), + vec![ + (Phase::Recovery, false), + (Phase::Sweep, false), + (Phase::Sweep, true), + (Phase::Recovery, false), + (Phase::Sweep, false), + ] + ); +} + +#[tokio::test] +async fn backend_cannot_exceed_tick_limits() { + let backend = Arc::new(ScriptedBackend::new([StepTemplate { + scan_complete: false, + traversal_syscalls: RECONCILIATION_TRAVERSAL_LIMIT + 1, + candidates: 0, + mutation_attempts: 0, + classifications: ReconciliationClassificationCounters::default(), + physical_mutation_possible: false, + retryable: false, + return_cursor: false, + error: false, + }])); + let mut coordinator = ReconciliationCoordinator::new(backend); + + assert_eq!( + coordinator.tick().await, + Err(ReconciliationCoordinatorError::Bounds) + ); +} + +#[tokio::test] +async fn tick_stops_at_the_exact_candidate_and_mutation_limits() { + let backend = Arc::new(ScriptedBackend::new( + (0..RECONCILIATION_MUTATION_LIMIT).map(|_| StepTemplate { + scan_complete: false, + traversal_syscalls: 1, + candidates: 1, + mutation_attempts: 1, + classifications: ReconciliationClassificationCounters::default(), + physical_mutation_possible: true, + retryable: false, + return_cursor: true, + error: false, + }), + )); + let mut coordinator = ReconciliationCoordinator::new(backend.clone()); + + let report = coordinator.tick().await.unwrap(); + + assert_eq!(report.candidates, RECONCILIATION_CANDIDATE_LIMIT); + assert_eq!(report.mutation_attempts, RECONCILIATION_MUTATION_LIMIT); + assert_eq!(report.traversal_syscalls, RECONCILIATION_MUTATION_LIMIT); + assert_eq!( + backend.calls.lock().unwrap().len(), + RECONCILIATION_MUTATION_LIMIT + ); + assert_eq!(backend.alive.load(Ordering::SeqCst), 0); + assert_eq!(backend.maximum.load(Ordering::SeqCst), 1); +} + +#[tokio::test] +async fn coordinator_retains_at_most_one_cursor_between_ticks() { + let backend = Arc::new(ScriptedBackend::new((0..128).map(|_| StepTemplate { + scan_complete: false, + traversal_syscalls: 1, + candidates: 0, + mutation_attempts: 0, + classifications: ReconciliationClassificationCounters::default(), + physical_mutation_possible: false, + retryable: true, + return_cursor: true, + error: false, + }))); + let mut coordinator = ReconciliationCoordinator::new(backend.clone()); + + for _ in 0..128 { + coordinator.tick().await.unwrap(); + assert_eq!(backend.alive.load(Ordering::SeqCst), 1); + } + assert_eq!(backend.maximum.load(Ordering::SeqCst), 1); + drop(coordinator); + assert_eq!(backend.alive.load(Ordering::SeqCst), 0); +} + +#[test] +fn registration_failures_have_a_closed_redacted_classification() { + let mut counters = ReconciliationClassificationCounters::default(); + + assert!(!counters.record_registration_error(ArtifactError::Integrity)); + assert!(!counters.record_registration_error(ArtifactError::UnsafeRoot)); + assert!(counters.record_registration_error(ArtifactError::Storage)); + assert!(!counters.record_registration_error(ArtifactError::NotFound)); + assert!(!counters.record_registration_error(ArtifactError::InvalidReference)); + assert!(!counters.record_registration_error(ArtifactError::EmptySource)); + assert!(!counters.record_registration_error(ArtifactError::SourceTooLarge)); + + assert_eq!( + counters, + ReconciliationClassificationCounters { + scanner_malformed: 0, + scanner_unsafe: 0, + recovery_present: 0, + recovery_safety: 0, + recovery_retryable: 0, + registration_integrity: 3, + registration_safety: 3, + registration_retryable: 1, + claim_metadata_conflict: 0, + mutation_integrity: 0, + mutation_safety: 0, + mutation_retryable: 0, + } + ); + let debug = format!("{counters:?}"); + assert!(!debug.contains("sha256:")); + assert!(!debug.contains('/')); +} + +#[test] +fn mutation_failures_are_classified_and_always_retryable() { + let mut counters = ReconciliationClassificationCounters::default(); + + assert!(counters.record_mutation_error(ArtifactError::Integrity)); + assert!(counters.record_mutation_error(ArtifactError::UnsafeRoot)); + assert!(counters.record_mutation_error(ArtifactError::Storage)); + assert!(counters.record_mutation_error(ArtifactError::NotFound)); + assert!(counters.record_mutation_error(ArtifactError::InvalidReference)); + assert!(counters.record_mutation_error(ArtifactError::EmptySource)); + assert!(counters.record_mutation_error(ArtifactError::SourceTooLarge)); + + assert_eq!(counters.mutation_integrity, 3); + assert_eq!(counters.mutation_safety, 3); + assert_eq!(counters.mutation_retryable, 1); +} + +#[tokio::test] +async fn production_backend_recovers_absence_before_sweep_and_restarts_after_mutation() { + let database_url = + crank_test_support::postgres_schema_url("admin_reconciliation_backend").await; + let pool = sqlx::PgPool::connect(&database_url).await.unwrap(); + MigrationAuthority::apply(&pool).await.unwrap(); + let registry = PostgresRegistry::connect(&database_url).await.unwrap(); + let root = TestRoot::new(); + let store = ArtifactStore::open(&root.0).unwrap(); + let now = OffsetDateTime::now_utc(); + + let expired = store.put_registered(b"expired physical absence\n").unwrap(); + let expired_claim = registry + .claim_artifact_reconciliation(ClaimArtifactReconciliationRequest { + artifact: &expired, + token: ArtifactClaimToken::generate(), + claimed_at: now - time::Duration::minutes(10), + lease_expires_at: now - time::Duration::minutes(5), + detached_grace: time::Duration::hours(24), + }) + .await + .unwrap(); + assert!(matches!(expired_claim, ArtifactClaimOutcome::Claimed(_))); + remove_registered_artifact(&store, &expired); + sqlx::query( + "update artifact_blobs + set claim_expires_at = clock_timestamp() - interval '1 second' + where digest = $1", + ) + .bind(expired.artifact_ref().digest_hex()) + .execute(registry.pool()) + .await + .unwrap(); + + let corrupted = store + .put_registered(b"integrity-blocked-candidate\n") + .unwrap(); + let corrupted_path = artifact_path(&root.0, &corrupted); + let corrupted_size = fs::metadata(&corrupted_path).unwrap().len() as usize; + fs::set_permissions(&corrupted_path, fs::Permissions::from_mode(0o600)).unwrap(); + fs::write(&corrupted_path, vec![b'x'; corrupted_size]).unwrap(); + fs::set_permissions(&corrupted_path, fs::Permissions::from_mode(0o400)).unwrap(); + age_artifact(&root.0, &corrupted); + + let later = store.put_registered(b"later-valid-candidate\n").unwrap(); + age_artifact(&root.0, &later); + let metadata_conflict = store + .put_registered(b"metadata-conflict-candidate\n") + .unwrap(); + age_artifact(&root.0, &metadata_conflict); + sqlx::query( + "insert into artifact_blobs + (digest, artifact_ref, size_bytes, storage_lifecycle, created_at, updated_at) + values ($1, $2, $3, 'available', $4, $4)", + ) + .bind(metadata_conflict.artifact_ref().digest_hex()) + .bind(metadata_conflict.artifact_ref().as_str()) + .bind(i64::try_from(metadata_conflict.size_bytes()).unwrap() + 1) + .bind(now) + .execute(registry.pool()) + .await + .unwrap(); + let shard = artifact_path(&root.0, &later) + .parent() + .unwrap() + .to_path_buf(); + fs::write(shard.join("not-a-digest"), b"malformed").unwrap(); + let unsafe_name = format!( + "{}{}", + &later.artifact_ref().digest_hex()[..2], + "f".repeat(62) + ); + assert_ne!(unsafe_name, later.artifact_ref().digest_hex()); + symlink("not-a-digest", shard.join(unsafe_name)).unwrap(); + + let backend = PostgresReconciliationBackend::new(registry.clone(), store.clone()); + let mut coordinator = ReconciliationCoordinator::with_tick_limits( + backend, + StepLimits { + traversal_syscalls: RECONCILIATION_TRAVERSAL_LIMIT, + candidates: RECONCILIATION_CANDIDATE_LIMIT, + mutations: 1, + }, + ); + + // The single mutation slot is consumed by expired-claim recovery. The old + // final candidate proves Sweep did not run ahead of Recovery. + let first = coordinator.tick().await.unwrap(); + assert_eq!(first.mutation_attempts, 1); + assert!(!first.recovery_completed); + assert_eq!( + store.reconciliation_presence(later.artifact_ref()).unwrap(), + ReconciliationPresence::Final + ); + let lifecycle: String = + sqlx::query_scalar("select storage_lifecycle from artifact_blobs where digest = $1") + .bind(expired.artifact_ref().digest_hex()) + .fetch_one(registry.pool()) + .await + .unwrap(); + assert_eq!(lifecycle, "unavailable"); + + // The real final sweep classifies blocked entries and still reaches the + // later valid item. Its physical mutation invalidates the final cursor. + let second = coordinator.tick().await.unwrap(); + assert!(second.recovery_completed); + assert_eq!(second.mutation_attempts, 1); + assert_eq!( + store.reconciliation_presence(later.artifact_ref()).unwrap(), + ReconciliationPresence::Quarantine + ); + + // Model the next scheduled/restarted lease window. Because the coordinator + // reset to Recovery and discarded its final cursor, quarantine is handled + // before another final sweep. + sqlx::query("update artifact_blobs set claim_expires_at = $1 where digest = $2") + .bind(OffsetDateTime::now_utc() - time::Duration::seconds(1)) + .bind(later.artifact_ref().digest_hex()) + .execute(registry.pool()) + .await + .unwrap(); + let third = coordinator.tick().await.unwrap(); + assert_eq!(third.mutation_attempts, 1); + assert_eq!( + store.reconciliation_presence(later.artifact_ref()).unwrap(), + ReconciliationPresence::Absent + ); + + let fourth = coordinator.tick().await.unwrap(); + assert!(fourth.cycle_completed); + let classifications = first + .classifications + .checked_add(second.classifications) + .and_then(|value| value.checked_add(third.classifications)) + .and_then(|value| value.checked_add(fourth.classifications)) + .unwrap(); + assert!(classifications.scanner_malformed > 0); + assert!(classifications.scanner_unsafe > 0); + assert!(classifications.registration_integrity > 0); + assert!(classifications.claim_metadata_conflict > 0); + assert_eq!( + store + .reconciliation_presence(corrupted.artifact_ref()) + .unwrap(), + ReconciliationPresence::Unsafe + ); + assert_eq!(classifications.registration_safety, 0); + assert_eq!(classifications.registration_retryable, 0); +} + +#[tokio::test] +async fn expired_claim_cursor_prevents_unsafe_prefix_starvation_across_ticks() { + let database_url = + crank_test_support::postgres_schema_url("admin_reconciliation_starvation").await; + let pool = sqlx::PgPool::connect(&database_url).await.unwrap(); + MigrationAuthority::apply(&pool).await.unwrap(); + let registry = PostgresRegistry::connect(&database_url).await.unwrap(); + let root = TestRoot::new(); + let store = ArtifactStore::open(&root.0).unwrap(); + let now = OffsetDateTime::now_utc(); + + for index in 0..=RECONCILIATION_CANDIDATE_LIMIT { + let artifact = store + .put_registered(format!("unsafe expired claim {index}\n").as_bytes()) + .unwrap(); + let outcome = registry + .claim_artifact_reconciliation(ClaimArtifactReconciliationRequest { + artifact: &artifact, + token: ArtifactClaimToken::generate(), + claimed_at: now - time::Duration::minutes(20), + lease_expires_at: now - time::Duration::minutes(15), + detached_grace: time::Duration::hours(24), + }) + .await + .unwrap(); + assert!(matches!(outcome, ArtifactClaimOutcome::Claimed(_))); + sqlx::query( + "update artifact_blobs + set claim_expires_at = clock_timestamp() - interval '10 minutes' + where digest = $1", + ) + .bind(artifact.artifact_ref().digest_hex()) + .execute(registry.pool()) + .await + .unwrap(); + fs::set_permissions( + artifact_path(&root.0, &artifact), + fs::Permissions::from_mode(0o600), + ) + .unwrap(); + } + + let absent = store + .put_registered(b"absent after unsafe prefix\n") + .unwrap(); + let outcome = registry + .claim_artifact_reconciliation(ClaimArtifactReconciliationRequest { + artifact: &absent, + token: ArtifactClaimToken::generate(), + claimed_at: now - time::Duration::minutes(15), + lease_expires_at: now - time::Duration::minutes(10), + detached_grace: time::Duration::hours(24), + }) + .await + .unwrap(); + assert!(matches!(outcome, ArtifactClaimOutcome::Claimed(_))); + sqlx::query( + "update artifact_blobs + set claim_expires_at = clock_timestamp() - interval '5 minutes' + where digest = $1", + ) + .bind(absent.artifact_ref().digest_hex()) + .execute(registry.pool()) + .await + .unwrap(); + remove_registered_artifact(&store, &absent); + + let backend = PostgresReconciliationBackend::new(registry.clone(), store); + let mut coordinator = ReconciliationCoordinator::new(backend); + let first = coordinator.tick().await.unwrap(); + assert_eq!(first.candidates, RECONCILIATION_CANDIDATE_LIMIT); + assert_eq!(first.mutation_attempts, 0); + assert_eq!( + first.classifications.recovery_safety, + RECONCILIATION_CANDIDATE_LIMIT + ); + let first_lifecycle: String = + sqlx::query_scalar("select storage_lifecycle from artifact_blobs where digest = $1") + .bind(absent.artifact_ref().digest_hex()) + .fetch_one(registry.pool()) + .await + .unwrap(); + assert_eq!(first_lifecycle, "available"); + + let second = coordinator.tick().await.unwrap(); + assert!(second.classifications.recovery_safety > 0); + assert!(second.mutation_attempts > 0); + let second_lifecycle: String = + sqlx::query_scalar("select storage_lifecycle from artifact_blobs where digest = $1") + .bind(absent.artifact_ref().digest_hex()) + .fetch_one(registry.pool()) + .await + .unwrap(); + assert_eq!(second_lifecycle, "unavailable"); +} diff --git a/crates/crank-artifacts/src/housekeeping.rs b/crates/crank-artifacts/src/housekeeping.rs index 9fdef3c..3bdbdbb 100644 --- a/crates/crank-artifacts/src/housekeeping.rs +++ b/crates/crank-artifacts/src/housekeeping.rs @@ -8,21 +8,21 @@ use crate::temp_scan::{list_names, valid_temp_name}; use crate::{ ArtifactError, ArtifactStore, ReconciliationMutation, ReconciliationNamespace, ReconciliationScan, ReconciliationScanStop, + recovery::{FINAL_NAMESPACE, namespace_name, namespace_number}, store::{ QUARANTINE_DIR, RootLock, check_file, checkpointed, ensure_root_unchanged, fsync_fd, open_existing_dir, open_or_create_dir, rename_no_replace_at, stat_fd, unlinkat, }, }; -const FINAL_NAMESPACE: u8 = 0; -const QUARANTINE_NAMESPACE: u8 = 1; - /// Opaque bounded-scan continuation, valid only for an unchanged namespace. /// Discard it after any put, quarantine, delete, or external mutation. pub struct ReconciliationCursor { root_dev: u64, root_ino: u64, + namespace_start: u8, namespace: u8, + namespace_end: u8, shard: u8, state: ReconciliationCursorState, } @@ -67,29 +67,17 @@ impl fmt::Debug for ReconciliationCursor { /// Opaque inode-bound evidence that exposes neither digest nor filesystem path. #[derive(Clone)] pub struct ReconciliationCandidate { - root_dev: u64, - root_ino: u64, - namespace: ReconciliationNamespace, - shard: String, - shard_dev: u64, - shard_ino: u64, - name: String, - dev: u64, - ino: u64, - modified_seconds: i64, - modified_nanoseconds: i64, -} - -impl fmt::Debug for ReconciliationCandidate { - fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - formatter.write_str("ReconciliationCandidate(..)") - } -} - -impl ReconciliationCandidate { - pub fn namespace(&self) -> ReconciliationNamespace { - self.namespace - } + pub(crate) root_dev: u64, + pub(crate) root_ino: u64, + pub(crate) namespace: ReconciliationNamespace, + pub(crate) shard: String, + pub(crate) shard_dev: u64, + pub(crate) shard_ino: u64, + pub(crate) name: String, + pub(crate) dev: u64, + pub(crate) ino: u64, + pub(crate) modified_seconds: i64, + pub(crate) modified_nanoseconds: i64, } /// Opaque single-use evidence for a stale temporary inode under the pinned root. @@ -113,25 +101,31 @@ pub struct TempScan { } impl ArtifactStore { - /// Streams entries without disclosing paths/digests, with opaque continuation. - pub fn scan_reconciliation( + pub(crate) fn scan_reconciliation_bounded( &self, + start_namespace: ReconciliationNamespace, + end_namespace: ReconciliationNamespace, continuation: Option, scan_budget: usize, result_limit: usize, ) -> Result<(ReconciliationScan, Vec), ArtifactError> { let root = self.root()?; + let namespace = namespace_number(start_namespace); + let namespace_end = namespace_number(end_namespace); let mut cursor = match continuation { Some(cursor) => { if cursor.root_dev != root.dev || cursor.root_ino != root.ino - || cursor.namespace > QUARANTINE_NAMESPACE + || cursor.namespace_start != namespace + || cursor.namespace < namespace + || cursor.namespace > cursor.namespace_end + || cursor.namespace_end != namespace_end { return Err(ArtifactError::UnsafeRoot); } cursor } - None => reconciliation_cursor(root.dev, root.ino), + None => reconciliation_cursor(root.dev, root.ino, namespace, namespace_end), }; let report = ReconciliationScan::empty(None); let _lock = match RootLock::shared(root) { @@ -177,7 +171,7 @@ impl ArtifactStore { } let mut entries = Vec::new(); - while cursor.namespace <= QUARANTINE_NAMESPACE { + while cursor.namespace <= cursor.namespace_end { if entries.len() == result_limit { return Ok(reconciliation_page( report, @@ -438,11 +432,18 @@ impl ArtifactStore { } } -fn reconciliation_cursor(root_dev: u64, root_ino: u64) -> ReconciliationCursor { +fn reconciliation_cursor( + root_dev: u64, + root_ino: u64, + namespace: u8, + namespace_end: u8, +) -> ReconciliationCursor { ReconciliationCursor { root_dev, root_ino, - namespace: FINAL_NAMESPACE, + namespace_start: namespace, + namespace, + namespace_end, shard: 0, state: ReconciliationCursorState::OpenNamespace, } @@ -865,14 +866,6 @@ fn parse_dirent(buffer: &[u8], offset: usize) -> Result<(Vec, usize, i64), A Ok((raw_name[..end].to_vec(), after, cookie)) } -fn namespace_name(namespace: u8) -> &'static [u8] { - match namespace { - FINAL_NAMESPACE => b"sha256", - QUARANTINE_NAMESPACE => QUARANTINE_DIR, - _ => unreachable!("validated reconciliation namespace"), - } -} - fn advance_reconciliation_shard(cursor: &mut ReconciliationCursor, namespace: OwnedFd) { if cursor.shard == u8::MAX { advance_namespace(cursor); @@ -883,12 +876,16 @@ fn advance_reconciliation_shard(cursor: &mut ReconciliationCursor, namespace: Ow } fn advance_namespace(cursor: &mut ReconciliationCursor) { - cursor.namespace = cursor.namespace.saturating_add(1); + cursor.namespace = if cursor.namespace >= cursor.namespace_end { + cursor.namespace_end.saturating_add(1) + } else { + cursor.namespace + 1 + }; cursor.shard = 0; cursor.state = ReconciliationCursorState::OpenNamespace; } -fn ensure_reconciliation_root( +pub(crate) fn ensure_reconciliation_root( root: &crate::store::Root, candidate: &ReconciliationCandidate, ) -> Result<(), ArtifactError> { @@ -904,7 +901,7 @@ fn ensure_reconciliation_root( Ok(()) } -fn revalidate_candidate( +pub(crate) fn revalidate_candidate( candidate: &ReconciliationCandidate, shard: i32, ) -> Result<(), ArtifactError> { diff --git a/crates/crank-artifacts/src/lib.rs b/crates/crank-artifacts/src/lib.rs index 39d475a..33a6d2f 100644 --- a/crates/crank-artifacts/src/lib.rs +++ b/crates/crank-artifacts/src/lib.rs @@ -11,6 +11,7 @@ mod error; mod housekeeping; mod legacy; mod model; +mod recovery; mod store; mod temp_scan; @@ -22,7 +23,7 @@ pub use error::ArtifactError; pub use housekeeping::{ReconciliationCandidate, ReconciliationCursor, StaleTemp, TempScan}; pub use model::{ ArtifactRef, MAX_ARTIFACT_BYTES, MAX_SOURCE_BYTES, ReconciliationMutation, - ReconciliationNamespace, ReconciliationScan, ReconciliationScanStop, RegisteredArtifact, - StoredArtifact, + ReconciliationNamespace, ReconciliationPresence, ReconciliationRegistration, + ReconciliationScan, ReconciliationScanStop, RegisteredArtifact, StoredArtifact, }; pub use store::ArtifactStore; diff --git a/crates/crank-artifacts/src/model.rs b/crates/crank-artifacts/src/model.rs index 912fa9c..cbdf493 100644 --- a/crates/crank-artifacts/src/model.rs +++ b/crates/crank-artifacts/src/model.rs @@ -77,6 +77,66 @@ pub struct RegisteredArtifact { pub(crate) size_bytes: usize, } +/// Result of validating a scanned reconciliation candidate for registry use. +/// +/// A candidate that is younger than the configured grace period never yields +/// a registration capability. The enum is intentionally redacted by its +/// `Debug` implementation so an eligibility report cannot disclose the +/// physical artifact identity. +#[derive(Clone)] +pub enum ReconciliationRegistration { + NotEligible, + Registered(RegisteredArtifact), +} + +/// Redacted physical state observed by an expired-claim recovery probe. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ReconciliationPresence { + Final, + Quarantine, + Absent, + Unsafe, + Retryable, +} + +impl fmt::Debug for ReconciliationRegistration { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("ReconciliationRegistration(..)") + } +} + +impl PartialEq for ReconciliationRegistration { + fn eq(&self, other: &Self) -> bool { + match (self, other) { + (Self::NotEligible, Self::NotEligible) => true, + (Self::Registered(left), Self::Registered(right)) => left == right, + _ => false, + } + } +} + +impl Eq for ReconciliationRegistration {} + +impl ReconciliationRegistration { + pub fn is_eligible(&self) -> bool { + matches!(self, Self::Registered(_)) + } + + pub fn artifact(&self) -> Option<&RegisteredArtifact> { + match self { + Self::NotEligible => None, + Self::Registered(artifact) => Some(artifact), + } + } + + pub fn into_artifact(self) -> Option { + match self { + Self::NotEligible => None, + Self::Registered(artifact) => Some(artifact), + } + } +} + /// Namespace in which reconciliation observed an artifact inode. It does not /// disclose the artifact's digest or filesystem location. #[derive(Clone, Copy, Debug, Eq, PartialEq)] diff --git a/crates/crank-artifacts/src/recovery.rs b/crates/crank-artifacts/src/recovery.rs new file mode 100644 index 0000000..3c9b077 --- /dev/null +++ b/crates/crank-artifacts/src/recovery.rs @@ -0,0 +1,224 @@ +use std::{ + fmt, + os::fd::{AsRawFd, OwnedFd}, + time::{Duration, SystemTime, UNIX_EPOCH}, +}; + +use crate::{ + ArtifactError, ArtifactRef, ArtifactStore, ReconciliationCandidate, ReconciliationCursor, + ReconciliationNamespace, ReconciliationPresence, ReconciliationRegistration, + ReconciliationScan, + housekeeping::{ensure_reconciliation_root, revalidate_candidate}, + store::{ + QUARANTINE_DIR, Root, RootLock, checkpointed, ensure_root_unchanged, open_existing_dir, + open_existing_file, read_verified, stat_fd, + }, +}; + +pub(crate) const FINAL_NAMESPACE: u8 = 0; +pub(crate) const QUARANTINE_NAMESPACE: u8 = 1; + +pub(crate) fn namespace_name(namespace: u8) -> &'static [u8] { + match namespace { + FINAL_NAMESPACE => b"sha256", + QUARANTINE_NAMESPACE => QUARANTINE_DIR, + _ => unreachable!("validated reconciliation namespace"), + } +} + +pub(crate) fn namespace_number(namespace: ReconciliationNamespace) -> u8 { + match namespace { + ReconciliationNamespace::Final => FINAL_NAMESPACE, + ReconciliationNamespace::Quarantine => QUARANTINE_NAMESPACE, + } +} + +impl fmt::Debug for ReconciliationCandidate { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("ReconciliationCandidate(..)") + } +} + +impl ReconciliationCandidate { + pub fn namespace(&self) -> ReconciliationNamespace { + self.namespace + } + + /// Returns whether the candidate has aged past `grace` at `now`. + /// + /// This only reports the timestamp captured by the scanner. Callers that + /// need a registration capability must use [`ArtifactStore::register_reconciliation`], + /// which repeats inode and content validation while holding the store lock. + pub fn is_grace_eligible(&self, grace: Duration, now: SystemTime) -> bool { + let modified = UNIX_EPOCH + .checked_add(Duration::new( + self.modified_seconds.max(0) as u64, + self.modified_nanoseconds.max(0) as u32, + )) + .unwrap_or(SystemTime::UNIX_EPOCH); + now.duration_since(modified).is_ok_and(|age| age >= grace) + } +} + +impl ArtifactStore { + /// Scans only one reconciliation namespace with a bounded opaque cursor. + /// + /// A quarantine scan ends after quarantine and never falls through to + /// final entries, so recovery can finish before the normal final sweep. + pub fn scan_reconciliation_namespace( + &self, + namespace: ReconciliationNamespace, + continuation: Option, + scan_budget: usize, + result_limit: usize, + ) -> Result<(ReconciliationScan, Vec), ArtifactError> { + self.scan_reconciliation_bounded( + namespace, + namespace, + continuation, + scan_budget, + result_limit, + ) + } + + /// Streams entries without disclosing paths/digests, with opaque continuation. + pub fn scan_reconciliation( + &self, + continuation: Option, + scan_budget: usize, + result_limit: usize, + ) -> Result<(ReconciliationScan, Vec), ArtifactError> { + self.scan_reconciliation_bounded( + ReconciliationNamespace::Final, + ReconciliationNamespace::Quarantine, + continuation, + scan_budget, + result_limit, + ) + } + + /// Revalidates a scanned candidate and mints a registration capability + /// only for a file that is old enough and whose bytes still match its + /// canonical name. The candidate remains usable for a later mutation. + pub fn register_reconciliation( + &self, + candidate: &ReconciliationCandidate, + grace: Duration, + now: SystemTime, + ) -> Result { + let root = self.root()?; + let _lock = RootLock::shared(root)?; + ensure_reconciliation_root(root, candidate)?; + let namespace = namespace_number(candidate.namespace); + let namespace_root = open_existing_dir(root.fd.as_raw_fd(), namespace_name(namespace))?; + let shard = open_existing_dir(namespace_root.as_raw_fd(), candidate.shard.as_bytes())?; + revalidate_candidate(candidate, shard.as_raw_fd())?; + + let file = open_existing_file(shard.as_raw_fd(), candidate.name.as_bytes())?; + let stat = stat_fd(file.as_raw_fd())?; + if stat.st_dev != candidate.dev + || stat.st_ino != candidate.ino + || stat.st_mtime != candidate.modified_seconds + || stat.st_mtime_nsec != candidate.modified_nanoseconds + { + return Err(ArtifactError::UnsafeRoot); + } + if !candidate.is_grace_eligible(grace, now) { + return Ok(ReconciliationRegistration::NotEligible); + } + + let artifact_ref = + ArtifactRef::from_digest_hex(&candidate.name).map_err(|_| ArtifactError::UnsafeRoot)?; + let bytes = read_verified(&file, &artifact_ref)?; + Ok(ReconciliationRegistration::Registered( + crate::RegisteredArtifact { + artifact_ref, + size_bytes: bytes.len(), + }, + )) + } + + /// Convenience form that uses the current wall clock for grace checks. + pub fn register_reconciliation_candidate( + &self, + candidate: &ReconciliationCandidate, + grace: Duration, + ) -> Result { + self.register_reconciliation(candidate, grace, SystemTime::now()) + } + + /// Probes final and quarantine for an expired claim without disclosing its + /// physical identity. Quarantine takes precedence over final. + pub fn reconciliation_presence( + &self, + artifact_ref: &ArtifactRef, + ) -> Result { + let root = self.root()?; + let _lock = match RootLock::shared(root) { + Ok(lock) => lock, + Err(ArtifactError::Storage) => return Ok(ReconciliationPresence::Retryable), + Err(error) => return Err(error), + }; + match ensure_root_unchanged(root) { + Ok(()) => {} + Err(ArtifactError::Storage) => return Ok(ReconciliationPresence::Retryable), + Err(error) => return Err(error), + } + + match probe_namespace(root, QUARANTINE_NAMESPACE, artifact_ref) { + Ok(Some(presence)) => return Ok(presence), + Ok(None) => {} + Err(presence) => return Ok(presence), + } + match probe_namespace(root, FINAL_NAMESPACE, artifact_ref) { + Ok(Some(presence)) => Ok(presence), + Ok(None) => Ok(ReconciliationPresence::Absent), + Err(presence) => Ok(presence), + } + } +} + +fn probe_namespace( + root: &Root, + namespace: u8, + artifact_ref: &ArtifactRef, +) -> Result, ReconciliationPresence> { + let namespace_root = probe_dir(root.fd.as_raw_fd(), namespace_name(namespace))?; + let Some(namespace_root) = namespace_root else { + return Ok(None); + }; + let shard = probe_dir(namespace_root.as_raw_fd(), artifact_ref.shard().as_bytes())?; + let Some(shard) = shard else { + return Ok(None); + }; + let file = match checkpointed("reconciliation_presence_open", || { + open_existing_file(shard.as_raw_fd(), artifact_ref.digest_hex().as_bytes()) + }) { + Ok(fd) => fd, + Err(ArtifactError::NotFound) => return Ok(None), + Err(ArtifactError::Integrity | ArtifactError::UnsafeRoot) => { + return Err(ReconciliationPresence::Unsafe); + } + Err(_) => return Err(ReconciliationPresence::Retryable), + }; + match read_verified(&file, artifact_ref) { + Ok(_) => Ok(Some(if namespace == FINAL_NAMESPACE { + ReconciliationPresence::Final + } else { + ReconciliationPresence::Quarantine + })), + Err(ArtifactError::Integrity | ArtifactError::UnsafeRoot) => { + Err(ReconciliationPresence::Unsafe) + } + Err(_) => Err(ReconciliationPresence::Retryable), + } +} + +fn probe_dir(parent: i32, name: &[u8]) -> Result, ReconciliationPresence> { + match open_existing_dir(parent, name) { + Ok(fd) => Ok(Some(fd)), + Err(ArtifactError::NotFound) => Ok(None), + Err(ArtifactError::UnsafeRoot) => Err(ReconciliationPresence::Unsafe), + Err(_) => Err(ReconciliationPresence::Retryable), + } +} diff --git a/crates/crank-artifacts/tests/reconciliation.rs b/crates/crank-artifacts/tests/reconciliation.rs index 58baf5b..329a821 100644 --- a/crates/crank-artifacts/tests/reconciliation.rs +++ b/crates/crank-artifacts/tests/reconciliation.rs @@ -3,14 +3,15 @@ use std::{ os::unix::fs::{MetadataExt, PermissionsExt}, path::PathBuf, sync::atomic::{AtomicU64, Ordering}, + time::{Duration, SystemTime}, }; #[cfg(debug_assertions)] -use std::{process::Command, sync::Arc, thread, time::Duration}; +use std::{process::Command, sync::Arc, thread}; use crank_artifacts::{ ArtifactError, ArtifactStore, ReconciliationMutation, ReconciliationNamespace, - ReconciliationScanStop, + ReconciliationPresence, ReconciliationRegistration, ReconciliationScanStop, }; #[cfg(debug_assertions)] @@ -769,3 +770,208 @@ fn crash_windows_are_recoverable() { ); } } + +#[test] +fn registration_revalidates_content_and_enforces_grace_without_candidate_identity_access() { + let root = TestRoot::new("registration"); + let store = ArtifactStore::open(&root.0).unwrap(); + let stored = store.put(b"reconciliation registration").unwrap(); + let (_, mut candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); + let candidate = candidates.pop().unwrap(); + + assert_eq!(format!("{candidate:?}"), "ReconciliationCandidate(..)"); + assert!(!candidate.is_grace_eligible(Duration::from_secs(3600), SystemTime::now())); + assert_eq!( + format!( + "{:?}", + store + .register_reconciliation(&candidate, Duration::from_secs(3600), SystemTime::now()) + .unwrap() + ), + "ReconciliationRegistration(..)" + ); + + let registration = store + .register_reconciliation( + &candidate, + Duration::ZERO, + SystemTime::now() + Duration::from_secs(3600), + ) + .unwrap(); + let registered = match registration { + ReconciliationRegistration::Registered(registered) => registered, + ReconciliationRegistration::NotEligible => panic!("zero grace must be eligible"), + }; + assert_eq!(registered.artifact_ref(), &stored.artifact_ref); + assert_eq!( + registered.size_bytes(), + b"reconciliation registration".len() + ); + + // The candidate remains usable for the physical mutation after the + // registration bridge borrowed it. + assert_eq!( + store.quarantine_reconciliation(candidate).unwrap(), + ReconciliationMutation::Quarantined + ); +} + +#[test] +fn namespace_scan_can_finish_quarantine_recovery_before_final_sweep() { + let root = TestRoot::new("namespace-order"); + let store = ArtifactStore::open(&root.0).unwrap(); + store.put(b"quarantine-first recovery").unwrap(); + let (_, candidates) = store + .scan_reconciliation_namespace(ReconciliationNamespace::Final, None, FULL_SCAN_BUDGET, 8) + .unwrap(); + store + .quarantine_reconciliation(candidates.into_iter().next().unwrap()) + .unwrap(); + store.put(b"quarantine-first recovery").unwrap(); + + let (quarantine_report, quarantine_candidates) = store + .scan_reconciliation_namespace( + ReconciliationNamespace::Quarantine, + None, + FULL_SCAN_BUDGET, + 8, + ) + .unwrap(); + assert_eq!(quarantine_report.stop, ReconciliationScanStop::Complete); + assert_eq!(quarantine_report.final_entries, 0); + assert_eq!(quarantine_report.quarantined_entries, 1); + assert_eq!(quarantine_candidates.len(), 1); + assert!(quarantine_report.continuation.is_none()); + + let (final_report, final_candidates) = store + .scan_reconciliation_namespace(ReconciliationNamespace::Final, None, FULL_SCAN_BUDGET, 8) + .unwrap(); + assert_eq!(final_report.stop, ReconciliationScanStop::Complete); + assert_eq!(final_report.final_entries, 1); + assert_eq!(final_report.quarantined_entries, 0); + assert_eq!(final_candidates.len(), 1); +} + +#[test] +fn namespace_scan_rejects_a_cursor_from_a_broader_scan() { + let root = TestRoot::new("namespace-cursor-scope"); + let store = ArtifactStore::open(&root.0).unwrap(); + let (report, _) = store.scan_reconciliation(None, 0, 1).unwrap(); + let cursor = report + .continuation + .expect("zero-budget scan retains cursor"); + + assert!(matches!( + store.scan_reconciliation_namespace( + ReconciliationNamespace::Quarantine, + Some(cursor), + FULL_SCAN_BUDGET, + 1, + ), + Err(ArtifactError::UnsafeRoot) + )); +} + +#[test] +fn full_scan_rejects_a_quarantine_only_cursor() { + let root = TestRoot::new("namespace-cursor-narrow"); + let store = ArtifactStore::open(&root.0).unwrap(); + let (report, _) = store + .scan_reconciliation_namespace(ReconciliationNamespace::Quarantine, None, 0, 1) + .unwrap(); + let cursor = report + .continuation + .expect("zero-budget namespace scan retains cursor"); + + assert!(matches!( + store.scan_reconciliation(Some(cursor), FULL_SCAN_BUDGET, 1), + Err(ArtifactError::UnsafeRoot) + )); +} + +#[test] +fn presence_probe_distinguishes_final_quarantine_and_absent_without_disclosure() { + let root = TestRoot::new("presence"); + let store = ArtifactStore::open(&root.0).unwrap(); + let stored = store.put(b"presence probe").unwrap(); + + assert_eq!( + store.reconciliation_presence(&stored.artifact_ref).unwrap(), + ReconciliationPresence::Final + ); + let (_, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); + store + .quarantine_reconciliation(candidates.into_iter().next().unwrap()) + .unwrap(); + assert_eq!( + store.reconciliation_presence(&stored.artifact_ref).unwrap(), + ReconciliationPresence::Quarantine + ); + // A republish can leave both locations during recovery. Quarantine takes + // precedence so the stale inode is still scheduled for deletion. + store.put(b"presence probe").unwrap(); + assert_eq!( + store.reconciliation_presence(&stored.artifact_ref).unwrap(), + ReconciliationPresence::Quarantine + ); + let (_, candidates) = store + .scan_reconciliation_namespace( + ReconciliationNamespace::Quarantine, + None, + FULL_SCAN_BUDGET, + 8, + ) + .unwrap(); + store + .delete_quarantined_reconciliation(candidates.into_iter().next().unwrap()) + .unwrap(); + assert_eq!( + store.reconciliation_presence(&stored.artifact_ref).unwrap(), + ReconciliationPresence::Final + ); + let (_, candidates) = store + .scan_reconciliation_namespace(ReconciliationNamespace::Final, None, FULL_SCAN_BUDGET, 8) + .unwrap(); + store + .quarantine_reconciliation(candidates.into_iter().next().unwrap()) + .unwrap(); + let (_, candidates) = store + .scan_reconciliation_namespace( + ReconciliationNamespace::Quarantine, + None, + FULL_SCAN_BUDGET, + 8, + ) + .unwrap(); + store + .delete_quarantined_reconciliation(candidates.into_iter().next().unwrap()) + .unwrap(); + assert_eq!( + store.reconciliation_presence(&stored.artifact_ref).unwrap(), + ReconciliationPresence::Absent + ); +} + +#[cfg(debug_assertions)] +#[test] +fn presence_probe_classifies_storage_failure_as_retryable() { + let _guard = fault_guard(); + let root = TestRoot::new("presence-retryable"); + let store = ArtifactStore::open(&root.0).unwrap(); + let stored = store.put(b"presence retryable").unwrap(); + + set_checkpoint("reconciliation_presence_open", FaultAction::Fail); + assert_eq!( + store.reconciliation_presence(&stored.artifact_ref).unwrap(), + ReconciliationPresence::Retryable + ); + clear_checkpoint(); + assert_eq!( + store.reconciliation_presence(&stored.artifact_ref).unwrap(), + ReconciliationPresence::Final + ); +} diff --git a/crates/crank-registry/Cargo.toml b/crates/crank-registry/Cargo.toml index 6cb30bc..7ced393 100644 --- a/crates/crank-registry/Cargo.toml +++ b/crates/crank-registry/Cargo.toml @@ -11,6 +11,7 @@ crank-artifacts = { path = "../crank-artifacts" } crank-core = { path = "../crank-core" } crank-mapping = { path = "../crank-mapping" } crank-schema = { path = "../crank-schema" } +rand.workspace = true serde.workspace = true serde_json.workspace = true sha2.workspace = true diff --git a/crates/crank-registry/src/error.rs b/crates/crank-registry/src/error.rs index 39b6ed1..40dcdd4 100644 --- a/crates/crank-registry/src/error.rs +++ b/crates/crank-registry/src/error.rs @@ -156,6 +156,8 @@ pub enum RegistryError { SourceUnavailable, #[error("artifact source integrity verification failed")] SourceIntegrity, + #[error("artifact reconciliation is already claimed")] + ArtifactClaimInProgress, #[error("artifact source metadata is invalid for field {field}")] InvalidArtifactSource { field: &'static str }, } diff --git a/crates/crank-registry/src/lib.rs b/crates/crank-registry/src/lib.rs index 31d773e..c1e8bf9 100644 --- a/crates/crank-registry/src/lib.rs +++ b/crates/crank-registry/src/lib.rs @@ -15,28 +15,32 @@ pub mod records { pub use crate::model::{ AdminBootstrapContractRecord, AgentSummary, AgentVersionRecord, AppendProductEventOutcome, AppliedImportOperation, ApprovalRequestRecord, ArtifactBlobLifecycle, ArtifactBlobRecord, - ArtifactDigest, ArtifactSourceCursor, ArtifactSourceId, ArtifactSourceLifecycle, - ArtifactSourcePage, ArtifactSourceRecord, ArtifactSourceSensitivity, AuthUserRecord, - DescriptorKind, DescriptorMetadata, ImportJob, ImportJobApplyResult, ImportJobId, - ImportJobKind, ImportJobStatus, InvitationRecord, InvocationHistoryLoss, - InvocationHistoryLossCategory, InvocationHistoryWriteOutcome, InvocationLogRecord, - InvocationRetentionOutcome, InvocationRetentionPolicy, InvocationRetentionStatus, - MasterKeyIdentityRecord, MasterKeyRotationRecord, MasterKeyRotationStatus, - MembershipRecord, OnboardingMilestoneResult, OnboardingPresentationMilestone, - OperationAgentRef, OperationSampleMetadata, OperationSummary, OperationUsageSummary, - OperationVersionRecord, Page, PlatformApiKeyRecord, ProductEventRecord, - PublishedAgentCatalog, PublishedAgentTool, RegistryOperation, SampleKind, SecretRecord, - SecretVersionRecord, SessionRecord, SkippedImportOperation, UsageAgentBreakdown, - UsageBucket, UsageOperationBreakdown, UsageOutcomeBreakdown, UsageOutcomeGroup, - UsageRollupRecord, UsageSummary, UsageTimelinePoint, VerifiedArtifactSource, - WorkspaceMembershipRecord, WorkspaceRecord, WorkspaceUpstream, WorkspaceUpstreamId, - YamlImportJob, YamlImportJobCompletion, YamlImportJobId, YamlImportJobStatus, + ArtifactClaimFinalization, ArtifactClaimFinalizeOutcome, ArtifactClaimOutcome, + ArtifactClaimRecheckOutcome, ArtifactClaimToken, ArtifactDigest, ArtifactExpiredClaimProbe, + ArtifactReconciliationClaim, ArtifactSourceCursor, ArtifactSourceId, + ArtifactSourceLifecycle, ArtifactSourcePage, ArtifactSourceRecord, + ArtifactSourceSensitivity, AuthUserRecord, DescriptorKind, DescriptorMetadata, ImportJob, + ImportJobApplyResult, ImportJobId, ImportJobKind, ImportJobStatus, InvitationRecord, + InvocationHistoryLoss, InvocationHistoryLossCategory, InvocationHistoryWriteOutcome, + InvocationLogRecord, InvocationRetentionOutcome, InvocationRetentionPolicy, + InvocationRetentionStatus, MasterKeyIdentityRecord, MasterKeyRotationRecord, + MasterKeyRotationStatus, MembershipRecord, OnboardingMilestoneResult, + OnboardingPresentationMilestone, OperationAgentRef, OperationSampleMetadata, + OperationSummary, OperationUsageSummary, OperationVersionRecord, Page, + PlatformApiKeyRecord, ProductEventRecord, PublishedAgentCatalog, PublishedAgentTool, + RegistryOperation, SampleKind, SecretRecord, SecretVersionRecord, SessionRecord, + SkippedImportOperation, UsageAgentBreakdown, UsageBucket, UsageOperationBreakdown, + UsageOutcomeBreakdown, UsageOutcomeGroup, UsageRollupRecord, UsageSummary, + UsageTimelinePoint, VerifiedArtifactSource, WorkspaceMembershipRecord, WorkspaceRecord, + WorkspaceUpstream, WorkspaceUpstreamId, YamlImportJob, YamlImportJobCompletion, + YamlImportJobId, YamlImportJobStatus, }; } pub mod requests { pub use crate::model::{ AdminSecurityAuditRequest, AppendProductEventRequest, ApplyImportJobRequest, + ClaimArtifactReconciliationRequest, ClaimExpiredArtifactReconciliationRequest, ConsumeAdminBootstrapContractRequest, CreateAdminBootstrapContractRequest, CreateAgentDraftVersionRequest, CreateAgentRequest, CreateApprovalRequest, CreateArtifactSourceRequest, CreateImportJobRequest, CreateInvitationRequest, @@ -67,9 +71,12 @@ pub use model::{ AdminBootstrapContractRecord, AdminSecurityAuditRequest, AgentStateExpectation, AgentSummary, AgentVersionRecord, AppendProductEventOutcome, AppendProductEventRequest, AppliedImportOperation, ApplyImportJobRequest, ApprovalRequestRecord, ArtifactBlobLifecycle, - ArtifactBlobRecord, ArtifactDigest, ArtifactSourceCursor, ArtifactSourceId, + ArtifactBlobRecord, ArtifactClaimFinalization, ArtifactClaimFinalizeOutcome, + ArtifactClaimOutcome, ArtifactClaimRecheckOutcome, ArtifactClaimToken, ArtifactDigest, + ArtifactExpiredClaimProbe, ArtifactReconciliationClaim, ArtifactSourceCursor, ArtifactSourceId, ArtifactSourceLifecycle, ArtifactSourcePage, ArtifactSourceRecord, ArtifactSourceSensitivity, - AuthUserRecord, ConsumeAdminBootstrapContractRequest, CreateAdminBootstrapContractRequest, + AuthUserRecord, ClaimArtifactReconciliationRequest, ClaimExpiredArtifactReconciliationRequest, + ConsumeAdminBootstrapContractRequest, CreateAdminBootstrapContractRequest, CreateAgentDraftVersionRequest, CreateAgentRequest, CreateApprovalRequest, CreateArtifactSourceRequest, CreateImportJobRequest, CreateInvitationRequest, CreateInvocationLogRequest, CreatePlatformApiKeyRequest, CreateSecretRequest, @@ -81,12 +88,13 @@ pub use model::{ InvocationHistoryWriteOutcome, InvocationLogRecord, InvocationRetentionOutcome, InvocationRetentionPolicy, InvocationRetentionStatus, ListApprovalRequestsQuery, ListArtifactSourcesQuery, ListInvocationLogsQuery, ListProductEventsQuery, - MASTER_KEY_CIPHER_CONTRACT, MAX_ARTIFACT_SOURCE_PAGE_SIZE, MasterKeyIdentityCandidate, - MasterKeyIdentityRecord, MasterKeyRotationRecord, MasterKeyRotationStatus, MembershipRecord, - OnboardingMilestoneResult, OnboardingPresentationMilestone, OperationAgentRef, - OperationSampleMetadata, OperationStateExpectation, OperationSummary, OperationUsageSummary, - OperationVersionRecord, Page, PlatformApiKeyRecord, ProductEventRecord, PublishAgentRequest, - PublishRequest, PublishedAgentCatalog, PublishedAgentTool, RecordOnboardingCompletionRequest, + MASTER_KEY_CIPHER_CONTRACT, MAX_ARTIFACT_CLAIM_RECOVERY_BATCH, MAX_ARTIFACT_SOURCE_PAGE_SIZE, + MasterKeyIdentityCandidate, MasterKeyIdentityRecord, MasterKeyRotationRecord, + MasterKeyRotationStatus, MembershipRecord, OnboardingMilestoneResult, + OnboardingPresentationMilestone, OperationAgentRef, OperationSampleMetadata, + OperationStateExpectation, OperationSummary, OperationUsageSummary, OperationVersionRecord, + Page, PlatformApiKeyRecord, ProductEventRecord, PublishAgentRequest, PublishRequest, + PublishedAgentCatalog, PublishedAgentTool, RecordOnboardingCompletionRequest, RecordOnboardingMilestoneRequest, RecoverAdminPasswordRequest, RegistryOperation, RotateSecretRequest, SampleKind, SaveAgentBindingsRequest, SaveAgentCatalogConfigRequest, SaveAuthProfileRequest, SaveDescriptorMetadataRequest, SaveSampleMetadataRequest, diff --git a/crates/crank-registry/src/model/artifact_source.rs b/crates/crank-registry/src/model/artifact_source.rs index 2e711d0..906a9aa 100644 --- a/crates/crank-registry/src/model/artifact_source.rs +++ b/crates/crank-registry/src/model/artifact_source.rs @@ -1,5 +1,8 @@ +use std::fmt; + use crank_artifacts::{ArtifactRef, RegisteredArtifact}; use crank_core::WorkspaceId; +use rand::random; use serde::{Deserialize, Serialize}; use time::OffsetDateTime; @@ -7,6 +10,189 @@ define_registry_id!(ArtifactSourceId); define_registry_id!(ArtifactDigest); pub const MAX_ARTIFACT_SOURCE_PAGE_SIZE: u32 = 100; +pub const MAX_ARTIFACT_CLAIM_RECOVERY_BATCH: u32 = 32; +const ARTIFACT_CLAIM_TOKEN_BYTES: usize = 32; + +/// An unpredictable, bounded fencing token for one reconciliation attempt. +/// +/// It is intentionally opaque: callers can retain it in an +/// [`ArtifactReconciliationClaim`], but cannot format or extract it for logs, +/// metrics, or another process. +#[derive(Clone, PartialEq, Eq)] +pub struct ArtifactClaimToken(String); + +impl ArtifactClaimToken { + pub fn generate() -> Self { + let bytes = random::<[u8; ARTIFACT_CLAIM_TOKEN_BYTES]>(); + let mut value = String::with_capacity(ARTIFACT_CLAIM_TOKEN_BYTES * 2); + for byte in bytes { + use std::fmt::Write as _; + + let _ = write!(value, "{byte:02x}"); + } + Self(value) + } + + pub(crate) fn as_str(&self) -> &str { + &self.0 + } +} + +impl fmt::Debug for ArtifactClaimToken { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("ArtifactClaimToken(..)") + } +} + +/// Opaque transient authority to recheck and finalize a reconciliation item. +/// +/// It has no `Display` implementation and its `Debug` form is redacted so a +/// durable claim token cannot cross an observability boundary by accident. +#[derive(Clone)] +pub struct ArtifactReconciliationClaim { + artifact_ref: ArtifactRef, + token: ArtifactClaimToken, +} + +impl ArtifactReconciliationClaim { + pub(crate) fn new(artifact_ref: ArtifactRef, token: ArtifactClaimToken) -> Self { + Self { + artifact_ref, + token, + } + } + + /// Opaque content-addressed identity for a trusted artifact-store probe. + /// + /// This must not be formatted into telemetry or an external response. + pub fn artifact_ref(&self) -> &ArtifactRef { + &self.artifact_ref + } + + pub(crate) fn token(&self) -> &ArtifactClaimToken { + &self.token + } +} + +impl fmt::Debug for ArtifactReconciliationClaim { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("ArtifactReconciliationClaim(..)") + } +} + +#[derive(Clone)] +pub struct ClaimArtifactReconciliationRequest<'a> { + pub artifact: &'a RegisteredArtifact, + pub token: ArtifactClaimToken, + pub claimed_at: OffsetDateTime, + pub lease_expires_at: OffsetDateTime, + pub detached_grace: time::Duration, +} + +impl fmt::Debug for ClaimArtifactReconciliationRequest<'_> { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("ClaimArtifactReconciliationRequest(..)") + } +} + +/// Bounded recovery lease acquisition for one expired durable claim. +/// +/// Recovery needs no `RegisteredArtifact`: it recovers an identity validated +/// when the previous claim was acquired. The caller must still revalidate +/// physical presence through the artifact store before finalization. +#[derive(Clone, Debug)] +pub struct ClaimExpiredArtifactReconciliationRequest { + pub token: ArtifactClaimToken, + pub claimed_at: OffsetDateTime, + pub lease_expires_at: OffsetDateTime, +} + +/// Opaque compare-and-swap evidence for an expired claim discovered by a +/// bounded recovery read. It does not acquire a lease by itself. +#[derive(Clone)] +pub struct ArtifactExpiredClaimProbe { + artifact_ref: ArtifactRef, + prior_token: String, + claim_expires_at: OffsetDateTime, +} + +impl ArtifactExpiredClaimProbe { + pub(crate) fn new( + artifact_ref: ArtifactRef, + prior_token: String, + claim_expires_at: OffsetDateTime, + ) -> Self { + Self { + artifact_ref, + prior_token, + claim_expires_at, + } + } + + /// Opaque content-addressed identity for a trusted artifact-store probe. + /// This must not be formatted into telemetry or an external response. + pub fn artifact_ref(&self) -> &ArtifactRef { + &self.artifact_ref + } + + pub(crate) fn prior_token(&self) -> &str { + &self.prior_token + } + + pub(crate) fn claim_expires_at(&self) -> OffsetDateTime { + self.claim_expires_at + } +} + +impl fmt::Debug for ArtifactExpiredClaimProbe { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("ArtifactExpiredClaimProbe(..)") + } +} + +#[derive(Clone, Debug)] +pub enum ArtifactClaimOutcome { + /// This caller owns the fenced claim and may perform the next recheck. + Claimed(ArtifactReconciliationClaim), + /// A current token already owns the artifact; retry after its lease. + HeldByOther, + /// The artifact is still referenced by at least one active source. + ActiveReference, + /// A detached source remains within the caller's grace window. + DetachedReferenceInGrace, + /// A row with the same digest does not match store-validated metadata. + MetadataConflict, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ArtifactClaimRecheckOutcome { + /// The caller still owns the claim and no global reference blocks I/O. + Mutate, + /// An active relation now protects the artifact. + ActiveReference, + /// A detached relation is still protected by grace. + DetachedReferenceInGrace, + /// The token was finalized, replaced, or expired. + Stale, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ArtifactClaimFinalization { + Deleted, + AlreadyAbsent, + Quarantined, + Retryable, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ArtifactClaimFinalizeOutcome { + /// Confirmed absence was durably recorded as V12 `unavailable`. + Unavailable, + /// An ambiguous/mid-flight mutation remains fenced for recovery. + Retained, + /// The supplied token no longer owns the row. + Stale, +} #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] diff --git a/crates/crank-registry/src/postgres/artifact_source.rs b/crates/crank-registry/src/postgres/artifact_source.rs index fc0bc97..22b4032 100644 --- a/crates/crank-registry/src/postgres/artifact_source.rs +++ b/crates/crank-registry/src/postgres/artifact_source.rs @@ -3,19 +3,373 @@ use crank_core::WorkspaceId; use sqlx::{Postgres, Row, Transaction, postgres::PgRow}; use crate::{ - ArtifactBlobLifecycle, ArtifactBlobRecord, ArtifactDigest, ArtifactSourceCursor, + ArtifactBlobLifecycle, ArtifactBlobRecord, ArtifactClaimFinalization, + ArtifactClaimFinalizeOutcome, ArtifactClaimOutcome, ArtifactClaimRecheckOutcome, + ArtifactDigest, ArtifactExpiredClaimProbe, ArtifactReconciliationClaim, ArtifactSourceCursor, ArtifactSourceId, ArtifactSourceLifecycle, ArtifactSourcePage, ArtifactSourceRecord, - ArtifactSourceSensitivity, CreateArtifactSourceRequest, DetachArtifactSourceRequest, - ListArtifactSourcesQuery, MAX_ARTIFACT_SOURCE_PAGE_SIZE, RegistryError, VerifiedArtifactSource, + ArtifactSourceSensitivity, ClaimArtifactReconciliationRequest, + ClaimExpiredArtifactReconciliationRequest, CreateArtifactSourceRequest, + DetachArtifactSourceRequest, ListArtifactSourcesQuery, MAX_ARTIFACT_CLAIM_RECOVERY_BATCH, + MAX_ARTIFACT_SOURCE_PAGE_SIZE, RegistryError, VerifiedArtifactSource, }; use super::PostgresRegistry; +mod reconciliation; +use reconciliation::{ + checked_cutoff, database_now, insert_blob_if_missing_at_database_time, validate_claim_lease, +}; + const SOURCE_ID_PREFIX: &str = "src_"; const MAX_SOURCE_ID_SUFFIX_BYTES: usize = 128; const MAX_MIME_TYPE_BYTES: usize = 255; impl PostgresRegistry { + /// Acquires a short, fenced reconciliation lease after checking references + /// across every workspace. A lease may be replayed by its original token, + /// or stolen only after expiry. + pub async fn claim_artifact_reconciliation( + &self, + request: ClaimArtifactReconciliationRequest<'_>, + ) -> Result { + let lease_duration = validate_claim_lease(request.claimed_at, request.lease_expires_at)?; + if request.detached_grace.is_negative() { + return Err(RegistryError::InvalidArtifactSource { + field: "detached_grace", + }); + } + if !valid_artifact_size(request.artifact.size_bytes()) { + return Err(RegistryError::InvalidArtifactSource { + field: "size_bytes", + }); + } + + let mut transaction = self.pool().begin().await?; + let artifact_ref = request.artifact.artifact_ref(); + let size_bytes = i64::try_from(request.artifact.size_bytes()).map_err(|_| { + RegistryError::InvalidArtifactSource { + field: "size_bytes", + } + })?; + insert_blob_if_missing_at_database_time(&mut transaction, artifact_ref, size_bytes).await?; + let row = lock_blob(&mut transaction, artifact_ref.digest_hex()).await?; + let database_now = database_now(&mut transaction).await?; + if !blob_metadata_matches(&row, artifact_ref, size_bytes)? { + transaction.rollback().await?; + return Ok(ArtifactClaimOutcome::MetadataConflict); + } + if global_active_reference(&mut transaction, artifact_ref.digest_hex()).await? { + transaction.rollback().await?; + return Ok(ArtifactClaimOutcome::ActiveReference); + } + if global_detached_reference_in_grace( + &mut transaction, + artifact_ref.digest_hex(), + checked_cutoff(database_now, request.detached_grace)?, + ) + .await? + { + transaction.rollback().await?; + return Ok(ArtifactClaimOutcome::DetachedReferenceInGrace); + } + + let claim_token = row.try_get::, _>("claim_token")?; + let claim_expires_at = + row.try_get::, _>("claim_expires_at")?; + if let (Some(current_token), Some(expires_at)) = (claim_token, claim_expires_at) { + if current_token == request.token.as_str() && expires_at > database_now { + transaction.commit().await?; + return Ok(ArtifactClaimOutcome::Claimed( + ArtifactReconciliationClaim::new(artifact_ref.clone(), request.token), + )); + } + if expires_at > database_now || current_token == request.token.as_str() { + transaction.rollback().await?; + return Ok(ArtifactClaimOutcome::HeldByOther); + } + } + + let updated = sqlx::query( + "update artifact_blobs + set claim_token = $1, claim_expires_at = $2, updated_at = greatest(updated_at, $3) + where digest = $4", + ) + .bind(request.token.as_str()) + .bind(database_now + lease_duration) + .bind(database_now) + .bind(artifact_ref.digest_hex()) + .execute(&mut *transaction) + .await? + .rows_affected(); + debug_assert_eq!(updated, 1); + transaction.commit().await?; + Ok(ArtifactClaimOutcome::Claimed( + ArtifactReconciliationClaim::new(artifact_ref.clone(), request.token), + )) + } + + /// Reads a bounded set of expired claims without taking their leases. + /// + /// Recovery first probes physical presence through the artifact store. Only + /// a confirmed absence should call `claim_expired_artifact_reconciliation`. + /// Leaving final or quarantine entries untouched lets the ordinary scanner + /// reclaim them with a store-validated `RegisteredArtifact`. + pub async fn list_expired_artifact_reconciliation_probes( + &self, + now: time::OffsetDateTime, + limit: u32, + ) -> Result, RegistryError> { + self.list_expired_artifact_reconciliation_probes_after(now, None, limit) + .await + } + + /// Continues the bounded expired-claim read after an opaque prior probe. + /// + /// Keyset pagination prevents an unsafe or still-present first claim from + /// starving later absent claims without persisting worker state in V12. + pub async fn list_expired_artifact_reconciliation_probes_after( + &self, + now: time::OffsetDateTime, + after: Option<&ArtifactExpiredClaimProbe>, + limit: u32, + ) -> Result, RegistryError> { + let limit = limit.min(MAX_ARTIFACT_CLAIM_RECOVERY_BATCH); + if limit == 0 { + return Ok(Vec::new()); + } + let after_expires_at = after.map(ArtifactExpiredClaimProbe::claim_expires_at); + let after_digest = after.map(|probe| probe.artifact_ref().digest_hex()); + let rows = sqlx::query( + "select artifact_ref, claim_token, claim_expires_at + from artifact_blobs + where claim_token is not null + and claim_expires_at <= $1 + and ( + $2::timestamptz is null + or (claim_expires_at, digest) > ($2::timestamptz, $3::text) + ) + order by claim_expires_at asc, digest asc + limit $4", + ) + .bind(now) + .bind(after_expires_at) + .bind(after_digest) + .bind(i64::from(limit)) + .fetch_all(self.pool()) + .await?; + rows.into_iter() + .map(|row| { + let artifact_ref = ArtifactRef::parse(&row.try_get::("artifact_ref")?) + .map_err(|_| RegistryError::InvalidArtifactSource { + field: "artifact_ref", + })?; + let prior_token = row.try_get::, _>("claim_token")?.ok_or( + RegistryError::InvalidArtifactSource { + field: "claim_token", + }, + )?; + let claim_expires_at = row + .try_get::, _>("claim_expires_at")? + .ok_or(RegistryError::InvalidArtifactSource { + field: "claim_expires_at", + })?; + Ok(ArtifactExpiredClaimProbe::new( + artifact_ref, + prior_token, + claim_expires_at, + )) + }) + .collect() + } + + /// Compare-and-swap steals an expired probe after recovery observed absence. + /// + /// The caller must probe physical state again after a successful claim: a + /// concurrent scanner may have changed the inode between the read and CAS. + pub async fn claim_expired_artifact_reconciliation( + &self, + probe: &ArtifactExpiredClaimProbe, + request: ClaimExpiredArtifactReconciliationRequest, + ) -> Result, RegistryError> { + let lease_duration = validate_claim_lease(request.claimed_at, request.lease_expires_at)?; + if request.token.as_str() == probe.prior_token() { + return Err(RegistryError::InvalidArtifactSource { + field: "claim_token", + }); + } + let mut transaction = self.pool().begin().await?; + let Some(row) = + lock_blob_optional(&mut transaction, probe.artifact_ref().digest_hex()).await? + else { + transaction.rollback().await?; + return Ok(None); + }; + let database_now = database_now(&mut transaction).await?; + let current_token = row.try_get::, _>("claim_token")?; + let expires_at = row.try_get::, _>("claim_expires_at")?; + if current_token.as_deref() != Some(probe.prior_token()) + || expires_at.is_none_or(|value| value > database_now) + { + transaction.rollback().await?; + return Ok(None); + } + let claimed = sqlx::query( + "update artifact_blobs + set claim_token = $1, + claim_expires_at = $2, + updated_at = greatest(updated_at, $3) + where digest = $4 + and claim_token = $5 + and claim_expires_at <= $3", + ) + .bind(request.token.as_str()) + .bind(database_now + lease_duration) + .bind(database_now) + .bind(probe.artifact_ref().digest_hex()) + .bind(probe.prior_token()) + .execute(&mut *transaction) + .await? + .rows_affected() + == 1; + transaction.commit().await?; + Ok(claimed + .then(|| ArtifactReconciliationClaim::new(probe.artifact_ref().clone(), request.token))) + } + + /// Rechecks an owned claim immediately before blocking filesystem I/O. + /// + /// `create_artifact_source` holds the same blob lock and rejects any active + /// claim, so a `Mutate` outcome fences the recheck-to-filesystem window. + pub async fn recheck_artifact_reconciliation_claim( + &self, + claim: &ArtifactReconciliationClaim, + _now: time::OffsetDateTime, + detached_grace: time::Duration, + ) -> Result { + if detached_grace.is_negative() { + return Err(RegistryError::InvalidArtifactSource { + field: "detached_grace", + }); + } + let mut transaction = self.pool().begin().await?; + let Some(row) = + lock_blob_optional(&mut transaction, claim.artifact_ref().digest_hex()).await? + else { + transaction.rollback().await?; + return Ok(ArtifactClaimRecheckOutcome::Stale); + }; + let database_now = database_now(&mut transaction).await?; + let current_token = row.try_get::, _>("claim_token")?; + let expires_at = row.try_get::, _>("claim_expires_at")?; + if current_token.as_deref() != Some(claim.token().as_str()) + || expires_at.is_none_or(|value| value <= database_now) + { + transaction.rollback().await?; + return Ok(ArtifactClaimRecheckOutcome::Stale); + } + if global_active_reference(&mut transaction, claim.artifact_ref().digest_hex()).await? { + transaction.commit().await?; + return Ok(ArtifactClaimRecheckOutcome::ActiveReference); + } + if global_detached_reference_in_grace( + &mut transaction, + claim.artifact_ref().digest_hex(), + checked_cutoff(database_now, detached_grace)?, + ) + .await? + { + transaction.commit().await?; + return Ok(ArtifactClaimRecheckOutcome::DetachedReferenceInGrace); + } + transaction.commit().await?; + Ok(ArtifactClaimRecheckOutcome::Mutate) + } + + /// Records a filesystem observation without performing filesystem I/O. + /// + /// Only confirmed absence clears the claim and transitions V12 metadata to + /// `unavailable`; every ambiguous observation deliberately retains it for + /// a later bounded recovery attempt. + pub async fn finalize_artifact_reconciliation_claim( + &self, + claim: &ArtifactReconciliationClaim, + observation: ArtifactClaimFinalization, + _observed_at: time::OffsetDateTime, + ) -> Result { + let mut transaction = self.pool().begin().await?; + let Some(row) = + lock_blob_optional(&mut transaction, claim.artifact_ref().digest_hex()).await? + else { + transaction.rollback().await?; + return Ok(ArtifactClaimFinalizeOutcome::Stale); + }; + let database_now = database_now(&mut transaction).await?; + let current_token = row.try_get::, _>("claim_token")?; + let expires_at = row.try_get::, _>("claim_expires_at")?; + if current_token.as_deref() != Some(claim.token().as_str()) + || expires_at.is_none_or(|value| value <= database_now) + { + transaction.rollback().await?; + return Ok(ArtifactClaimFinalizeOutcome::Stale); + } + + let (lifecycle, clear_claim, outcome) = match observation { + ArtifactClaimFinalization::Deleted | ArtifactClaimFinalization::AlreadyAbsent => ( + "unavailable", + true, + ArtifactClaimFinalizeOutcome::Unavailable, + ), + ArtifactClaimFinalization::Quarantined | ArtifactClaimFinalization::Retryable => { + ("available", false, ArtifactClaimFinalizeOutcome::Retained) + } + }; + let updated = if clear_claim { + sqlx::query( + "update artifact_blobs + set storage_lifecycle = $1, claim_token = null, claim_expires_at = null, + updated_at = greatest(updated_at, $2) + where digest = $3 and claim_token = $4 and claim_expires_at > $2", + ) + .bind(lifecycle) + .bind(database_now) + .bind(claim.artifact_ref().digest_hex()) + .bind(claim.token().as_str()) + .execute(&mut *transaction) + .await? + .rows_affected() + } else { + sqlx::query( + "update artifact_blobs + set updated_at = greatest(updated_at, $1) + where digest = $2 and claim_token = $3 and claim_expires_at > $1", + ) + .bind(database_now) + .bind(claim.artifact_ref().digest_hex()) + .bind(claim.token().as_str()) + .execute(&mut *transaction) + .await? + .rows_affected() + }; + if updated != 1 { + transaction.rollback().await?; + return Ok(ArtifactClaimFinalizeOutcome::Stale); + } + transaction.commit().await?; + Ok(outcome) + } + + /// Alias used by startup recovery once its filesystem probe has produced a + /// typed observation. The registry never performs the probe itself. + pub async fn recover_artifact_reconciliation_claim( + &self, + claim: &ArtifactReconciliationClaim, + observation: ArtifactClaimFinalization, + recovered_at: time::OffsetDateTime, + ) -> Result { + self.finalize_artifact_reconciliation_claim(claim, observation, recovered_at) + .await + } + pub async fn create_artifact_source( &self, request: CreateArtifactSourceRequest<'_>, @@ -254,6 +608,41 @@ async fn ensure_blob( field: "size_bytes", } })?; + insert_blob_if_missing( + transaction, + request.artifact.artifact_ref(), + size_bytes, + request.created_at, + ) + .await?; + + let row = lock_blob(transaction, digest).await?; + if !blob_metadata_matches(&row, request.artifact.artifact_ref(), size_bytes)? { + return Err(source_conflict(request.source_id)); + } + if row.try_get::, _>("claim_token")?.is_some() { + return Err(RegistryError::ArtifactClaimInProgress); + } + if row.try_get::("storage_lifecycle")? == "unavailable" { + sqlx::query( + "update artifact_blobs + set storage_lifecycle = 'available', updated_at = greatest(updated_at, $1) + where digest = $2 and claim_token is null", + ) + .bind(request.created_at) + .bind(digest) + .execute(&mut **transaction) + .await?; + } + Ok(()) +} + +async fn insert_blob_if_missing( + transaction: &mut Transaction<'_, Postgres>, + artifact_ref: &ArtifactRef, + size_bytes: i64, + created_at: time::OffsetDateTime, +) -> Result<(), RegistryError> { sqlx::query( "insert into artifact_blobs ( digest, artifact_ref, size_bytes, storage_lifecycle, @@ -261,25 +650,87 @@ async fn ensure_blob( ) values ($1, $2, $3, 'available', null, null, $4, $4) on conflict (digest) do nothing", ) - .bind(digest) - .bind(request.artifact.artifact_ref().as_str()) + .bind(artifact_ref.digest_hex()) + .bind(artifact_ref.as_str()) .bind(size_bytes) - .bind(request.created_at) + .bind(created_at) .execute(&mut **transaction) .await?; + Ok(()) +} - let row = sqlx::query( - "select artifact_ref, size_bytes from artifact_blobs where digest = $1 for update", +async fn lock_blob( + transaction: &mut Transaction<'_, Postgres>, + digest: &str, +) -> Result { + lock_blob_optional(transaction, digest) + .await? + .ok_or(RegistryError::InvalidArtifactSource { + field: "artifact_blob", + }) +} + +async fn lock_blob_optional( + transaction: &mut Transaction<'_, Postgres>, + digest: &str, +) -> Result, RegistryError> { + sqlx::query( + "select artifact_ref, size_bytes, storage_lifecycle, claim_token, claim_expires_at + from artifact_blobs + where digest = $1 + for update", + ) + .bind(digest) + .fetch_optional(&mut **transaction) + .await + .map_err(Into::into) +} + +fn blob_metadata_matches( + row: &PgRow, + artifact_ref: &ArtifactRef, + size_bytes: i64, +) -> Result { + Ok( + row.try_get::("artifact_ref")? == artifact_ref.as_str() + && row.try_get::("size_bytes")? == size_bytes, + ) +} + +async fn global_active_reference( + transaction: &mut Transaction<'_, Postgres>, + digest: &str, +) -> Result { + sqlx::query_scalar::<_, bool>( + "select exists( + select 1 from artifact_sources + where blob_digest = $1 and lifecycle = 'active' + )", ) .bind(digest) .fetch_one(&mut **transaction) - .await?; - let recorded_ref = row.try_get::("artifact_ref")?; - let recorded_size = row.try_get::("size_bytes")?; - if recorded_ref != request.artifact.artifact_ref().as_str() || recorded_size != size_bytes { - return Err(source_conflict(request.source_id)); - } - Ok(()) + .await + .map_err(Into::into) +} + +async fn global_detached_reference_in_grace( + transaction: &mut Transaction<'_, Postgres>, + digest: &str, + detached_after: time::OffsetDateTime, +) -> Result { + sqlx::query_scalar::<_, bool>( + "select exists( + select 1 from artifact_sources + where blob_digest = $1 + and lifecycle = 'detached' + and detached_at > $2 + )", + ) + .bind(digest) + .bind(detached_after) + .fetch_one(&mut **transaction) + .await + .map_err(Into::into) } async fn get_source_in_transaction( diff --git a/crates/crank-registry/src/postgres/artifact_source/reconciliation.rs b/crates/crank-registry/src/postgres/artifact_source/reconciliation.rs new file mode 100644 index 0000000..5ceb6bf --- /dev/null +++ b/crates/crank-registry/src/postgres/artifact_source/reconciliation.rs @@ -0,0 +1,68 @@ +use crank_artifacts::ArtifactRef; +use sqlx::{Postgres, Transaction}; + +use crate::{PostgresRegistry, RegistryError}; + +const MAX_ARTIFACT_CLAIM_LEASE: time::Duration = time::Duration::minutes(5); + +impl PostgresRegistry { + /// Returns the PostgreSQL clock used by production reconciliation leases. + pub async fn artifact_reconciliation_now(&self) -> Result { + sqlx::query_scalar("select clock_timestamp()") + .fetch_one(self.pool()) + .await + .map_err(RegistryError::from) + } +} + +pub(super) async fn insert_blob_if_missing_at_database_time( + transaction: &mut Transaction<'_, Postgres>, + artifact_ref: &ArtifactRef, + size_bytes: i64, +) -> Result<(), RegistryError> { + sqlx::query( + "insert into artifact_blobs ( + digest, artifact_ref, size_bytes, storage_lifecycle, + claim_token, claim_expires_at, created_at, updated_at + ) values ($1, $2, $3, 'available', null, null, clock_timestamp(), clock_timestamp()) + on conflict (digest) do nothing", + ) + .bind(artifact_ref.digest_hex()) + .bind(artifact_ref.as_str()) + .bind(size_bytes) + .execute(&mut **transaction) + .await?; + Ok(()) +} + +pub(super) fn validate_claim_lease( + claimed_at: time::OffsetDateTime, + lease_expires_at: time::OffsetDateTime, +) -> Result { + let duration = lease_expires_at - claimed_at; + if duration <= time::Duration::ZERO || duration > MAX_ARTIFACT_CLAIM_LEASE { + return Err(RegistryError::InvalidArtifactSource { + field: "claim_lease", + }); + } + Ok(duration) +} + +pub(super) fn checked_cutoff( + now: time::OffsetDateTime, + grace: time::Duration, +) -> Result { + now.checked_sub(grace) + .ok_or(RegistryError::InvalidArtifactSource { + field: "detached_grace", + }) +} + +pub(super) async fn database_now( + transaction: &mut Transaction<'_, Postgres>, +) -> Result { + sqlx::query_scalar("select clock_timestamp()") + .fetch_one(&mut **transaction) + .await + .map_err(Into::into) +} diff --git a/crates/crank-registry/tests/integration.rs b/crates/crank-registry/tests/integration.rs index 906a9f4..29b0e0a 100644 --- a/crates/crank-registry/tests/integration.rs +++ b/crates/crank-registry/tests/integration.rs @@ -1,6 +1,7 @@ mod integration { mod agents_usage; mod approval; + mod artifact_reconciliation; mod artifact_sources; mod common; mod credential_touch; diff --git a/crates/crank-registry/tests/integration/artifact_reconciliation.rs b/crates/crank-registry/tests/integration/artifact_reconciliation.rs new file mode 100644 index 0000000..390726c --- /dev/null +++ b/crates/crank-registry/tests/integration/artifact_reconciliation.rs @@ -0,0 +1,214 @@ +use std::time::Duration as StdDuration; + +use crank_artifacts::ArtifactStore; +use crank_registry::{ + ArtifactClaimFinalization, ArtifactClaimFinalizeOutcome, ArtifactClaimOutcome, + ArtifactClaimToken, ClaimArtifactReconciliationRequest, + ClaimExpiredArtifactReconciliationRequest, RegistryError, +}; + +use super::{ + artifact_sources::{TestRoot, timestamp}, + common::TestDatabase, +}; + +#[tokio::test] +async fn locked_database_time_and_expired_recovery_are_fenced() { + let database = TestDatabase::new().await; + let registry = database.registry().await; + let raw_pool = database.raw_pool().await; + let root = TestRoot::new("database-time"); + let store = ArtifactStore::open(&root.0).unwrap(); + let caller_time = timestamp("2020-01-01T00:00:00Z"); + + let blocked = store.put_registered(b"row lock expiry\n").unwrap(); + assert!(matches!( + registry + .claim_artifact_reconciliation(ClaimArtifactReconciliationRequest { + artifact: &blocked, + token: ArtifactClaimToken::generate(), + claimed_at: caller_time, + lease_expires_at: caller_time + time::Duration::seconds(1), + detached_grace: time::Duration::hours(24), + }) + .await + .unwrap(), + ArtifactClaimOutcome::Claimed(_) + )); + let mut blocker = raw_pool.begin().await.unwrap(); + sqlx::query("select digest from artifact_blobs where digest = $1 for update") + .bind(blocked.artifact_ref().digest_hex()) + .fetch_one(&mut *blocker) + .await + .unwrap(); + let waiting_registry = registry.clone(); + let waiting_artifact = blocked.clone(); + let waiting = tokio::spawn(async move { + waiting_registry + .claim_artifact_reconciliation(ClaimArtifactReconciliationRequest { + artifact: &waiting_artifact, + token: ArtifactClaimToken::generate(), + claimed_at: caller_time, + lease_expires_at: caller_time + time::Duration::minutes(5), + detached_grace: time::Duration::hours(24), + }) + .await + }); + tokio::time::sleep(StdDuration::from_millis(1_100)).await; + blocker.commit().await.unwrap(); + assert!(matches!( + waiting.await.unwrap().unwrap(), + ArtifactClaimOutcome::Claimed(_) + )); + + let recoverable = store.put_registered(b"single winner\n").unwrap(); + let prior_token = ArtifactClaimToken::generate(); + let prior_claim = match registry + .claim_artifact_reconciliation(ClaimArtifactReconciliationRequest { + artifact: &recoverable, + token: prior_token.clone(), + claimed_at: caller_time, + lease_expires_at: caller_time + time::Duration::minutes(5), + detached_grace: time::Duration::hours(24), + }) + .await + .unwrap() + { + ArtifactClaimOutcome::Claimed(claim) => claim, + outcome => panic!("unexpected claim outcome: {outcome:?}"), + }; + sqlx::query( + "update artifact_blobs + set claim_expires_at = clock_timestamp() - interval '1 second' + where digest = $1", + ) + .bind(recoverable.artifact_ref().digest_hex()) + .execute(&raw_pool) + .await + .unwrap(); + assert!(matches!( + registry + .claim_artifact_reconciliation(ClaimArtifactReconciliationRequest { + artifact: &recoverable, + token: prior_token.clone(), + claimed_at: caller_time, + lease_expires_at: caller_time + time::Duration::minutes(5), + detached_grace: time::Duration::hours(24), + }) + .await + .unwrap(), + ArtifactClaimOutcome::HeldByOther + )); + let recovery_now = registry.artifact_reconciliation_now().await.unwrap(); + let probe = registry + .list_expired_artifact_reconciliation_probes(recovery_now, 1) + .await + .unwrap() + .pop() + .unwrap(); + assert!(matches!( + registry + .claim_expired_artifact_reconciliation( + &probe, + ClaimExpiredArtifactReconciliationRequest { + token: prior_token, + claimed_at: caller_time, + lease_expires_at: caller_time + time::Duration::minutes(5), + }, + ) + .await, + Err(RegistryError::InvalidArtifactSource { + field: "claim_token" + }) + )); + + let first_registry = registry.clone(); + let second_registry = registry.clone(); + let first_probe = probe.clone(); + let (first, second) = tokio::join!( + async move { + first_registry + .claim_expired_artifact_reconciliation( + &first_probe, + ClaimExpiredArtifactReconciliationRequest { + token: ArtifactClaimToken::generate(), + claimed_at: caller_time, + lease_expires_at: caller_time + time::Duration::minutes(5), + }, + ) + .await + }, + async move { + second_registry + .claim_expired_artifact_reconciliation( + &probe, + ClaimExpiredArtifactReconciliationRequest { + token: ArtifactClaimToken::generate(), + claimed_at: caller_time, + lease_expires_at: caller_time + time::Duration::minutes(5), + }, + ) + .await + } + ); + let first = first.unwrap(); + let second = second.unwrap(); + assert_eq!( + usize::from(first.is_some()) + usize::from(second.is_some()), + 1 + ); + let winner = first.or(second).unwrap(); + + sqlx::query("update artifact_blobs set claim_expires_at = clock_timestamp() where digest = $1") + .bind(recoverable.artifact_ref().digest_hex()) + .execute(&raw_pool) + .await + .unwrap(); + for claim in [&winner, &prior_claim] { + assert_eq!( + registry + .finalize_artifact_reconciliation_claim( + claim, + ArtifactClaimFinalization::AlreadyAbsent, + caller_time, + ) + .await + .unwrap(), + ArtifactClaimFinalizeOutcome::Stale + ); + } + + let invalid = store.put_registered(b"invalid bounds\n").unwrap(); + assert!(matches!( + registry + .claim_artifact_reconciliation(ClaimArtifactReconciliationRequest { + artifact: &invalid, + token: ArtifactClaimToken::generate(), + claimed_at: caller_time, + lease_expires_at: caller_time + time::Duration::minutes(5), + detached_grace: time::Duration::MAX, + }) + .await, + Err(RegistryError::InvalidArtifactSource { + field: "detached_grace" + }) + )); + assert!(matches!( + registry + .claim_artifact_reconciliation(ClaimArtifactReconciliationRequest { + artifact: &invalid, + token: ArtifactClaimToken::generate(), + claimed_at: caller_time, + lease_expires_at: caller_time + + time::Duration::minutes(5) + + time::Duration::seconds(1), + detached_grace: time::Duration::ZERO, + }) + .await, + Err(RegistryError::InvalidArtifactSource { + field: "claim_lease" + }) + )); + + database.cleanup().await; +} diff --git a/crates/crank-registry/tests/integration/artifact_sources.rs b/crates/crank-registry/tests/integration/artifact_sources.rs index 31a1499..11f7d9a 100644 --- a/crates/crank-registry/tests/integration/artifact_sources.rs +++ b/crates/crank-registry/tests/integration/artifact_sources.rs @@ -5,24 +5,30 @@ use std::{ os::unix::fs::PermissionsExt, path::PathBuf, sync::atomic::{AtomicU64, Ordering}, + time::{Duration as StdDuration, SystemTime}, }; -use crank_artifacts::{ArtifactRef, ArtifactStore}; +use crank_artifacts::{ + ArtifactRef, ArtifactStore, ReconciliationMutation, ReconciliationPresence, + ReconciliationRegistration, RegisteredArtifact, +}; use crank_core::{Workspace, WorkspaceId, WorkspaceStatus}; use crank_registry::{ - ArtifactSourceId, ArtifactSourceLifecycle, ArtifactSourceSensitivity, - CreateArtifactSourceRequest, CreateWorkspaceRequest, DetachArtifactSourceRequest, - ListArtifactSourcesQuery, RegistryError, + ArtifactClaimFinalization, ArtifactClaimFinalizeOutcome, ArtifactClaimOutcome, + ArtifactClaimRecheckOutcome, ArtifactClaimToken, ArtifactSourceId, ArtifactSourceLifecycle, + ArtifactSourceSensitivity, ClaimArtifactReconciliationRequest, + ClaimExpiredArtifactReconciliationRequest, CreateArtifactSourceRequest, CreateWorkspaceRequest, + DetachArtifactSourceRequest, ListArtifactSourcesQuery, RegistryError, }; use serde_json::json; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; static NEXT_ROOT: AtomicU64 = AtomicU64::new(0); -struct TestRoot(PathBuf); +pub(super) struct TestRoot(pub(super) PathBuf); impl TestRoot { - fn new(name: &str) -> Self { + pub(super) fn new(name: &str) -> Self { let path = std::env::temp_dir().join(format!( "crank-registry-artifacts-{name}-{}-{}", std::process::id(), @@ -41,10 +47,26 @@ impl Drop for TestRoot { } } -fn timestamp(value: &str) -> OffsetDateTime { +pub(super) fn timestamp(value: &str) -> OffsetDateTime { OffsetDateTime::parse(value, &Rfc3339).unwrap() } +fn reconciliation_candidate_for( + store: &ArtifactStore, + artifact: &RegisteredArtifact, +) -> crank_artifacts::ReconciliationCandidate { + let (_, candidates) = store.scan_reconciliation(None, 4096, 32).unwrap(); + candidates + .into_iter() + .find(|candidate| { + matches!( + store.register_reconciliation(candidate, StdDuration::ZERO, SystemTime::now()), + Ok(ReconciliationRegistration::Registered(found)) if found == *artifact + ) + }) + .expect("registered reconciliation candidate") +} + async fn create_workspace(registry: &crank_registry::PostgresRegistry, id: &str) -> WorkspaceId { let workspace_id = WorkspaceId::new(id); let created_at = timestamp("2026-08-26T10:00:00Z"); @@ -534,3 +556,300 @@ async fn verified_read_returns_only_digest_and_size_verified_bytes() { database.cleanup().await; } + +#[tokio::test] +async fn reconciliation_claims_are_fenced_global_and_allow_verified_revival() { + let database = TestDatabase::new().await; + let registry = database.registry().await; + let raw_pool = database.raw_pool().await; + let workspace_a = create_workspace(®istry, "ws_claim_a").await; + let workspace_b = create_workspace(®istry, "ws_claim_b").await; + let root = TestRoot::new("claims"); + let store = ArtifactStore::open(&root.0).unwrap(); + let unreferenced = store.put_registered(b"unreferenced\n").unwrap(); + let started = timestamp("2026-08-27T10:00:00Z"); + let lease = timestamp("2026-08-27T10:05:00Z"); + let token = ArtifactClaimToken::generate(); + assert_eq!(format!("{token:?}"), "ArtifactClaimToken(..)"); + + let request = ClaimArtifactReconciliationRequest { + artifact: &unreferenced, + token: token.clone(), + claimed_at: started, + lease_expires_at: lease, + detached_grace: time::Duration::hours(24), + }; + assert_eq!( + format!("{request:?}"), + "ClaimArtifactReconciliationRequest(..)" + ); + let claim = match registry + .claim_artifact_reconciliation(request) + .await + .unwrap() + { + ArtifactClaimOutcome::Claimed(claim) => claim, + outcome => panic!("unexpected claim outcome: {outcome:?}"), + }; + assert_eq!(format!("{claim:?}"), "ArtifactReconciliationClaim(..)"); + assert!(matches!( + registry + .claim_artifact_reconciliation(ClaimArtifactReconciliationRequest { + artifact: &unreferenced, + token, + claimed_at: timestamp("2026-08-27T10:01:00Z"), + lease_expires_at: lease, + detached_grace: time::Duration::hours(24), + }) + .await + .unwrap(), + ArtifactClaimOutcome::Claimed(_) + )); + assert_eq!( + registry + .recheck_artifact_reconciliation_claim(&claim, started, time::Duration::hours(24)) + .await + .unwrap(), + ArtifactClaimRecheckOutcome::Mutate + ); + assert!(matches!( + registry + .create_artifact_source(CreateArtifactSourceRequest { + workspace_id: &workspace_a, + source_id: &ArtifactSourceId::new("src_blocked"), + artifact: &unreferenced, + mime_type: "text/plain", + sensitivity: ArtifactSourceSensitivity::Internal, + created_at: started, + }) + .await, + Err(RegistryError::ArtifactClaimInProgress) + )); + assert_eq!( + registry + .finalize_artifact_reconciliation_claim( + &claim, + ArtifactClaimFinalization::Quarantined, + timestamp("2026-08-27T10:01:00Z"), + ) + .await + .unwrap(), + ArtifactClaimFinalizeOutcome::Retained + ); + + let final_candidate = reconciliation_candidate_for(&store, &unreferenced); + assert!(matches!( + store.quarantine_reconciliation(final_candidate), + Ok(ReconciliationMutation::Quarantined | ReconciliationMutation::AlreadyQuarantined) + )); + let quarantined_candidate = reconciliation_candidate_for(&store, &unreferenced); + assert!(matches!( + store.delete_quarantined_reconciliation(quarantined_candidate), + Ok(ReconciliationMutation::Deleted | ReconciliationMutation::AlreadyAbsent) + )); + assert_eq!( + store + .reconciliation_presence(unreferenced.artifact_ref()) + .unwrap(), + ReconciliationPresence::Absent + ); + + sqlx::query( + "update artifact_blobs + set claim_expires_at = clock_timestamp() - interval '1 second' + where digest = $1", + ) + .bind(unreferenced.artifact_ref().digest_hex()) + .execute(&raw_pool) + .await + .unwrap(); + let recovery_now = registry.artifact_reconciliation_now().await.unwrap(); + let probes = registry + .list_expired_artifact_reconciliation_probes(recovery_now, 1) + .await + .unwrap(); + assert_eq!( + format!("{:?}", probes.first().expect("expired claim probe")), + "ArtifactExpiredClaimProbe(..)" + ); + let winner = registry + .claim_expired_artifact_reconciliation( + probes.first().expect("expired claim probe"), + ClaimExpiredArtifactReconciliationRequest { + token: ArtifactClaimToken::generate(), + claimed_at: recovery_now, + lease_expires_at: recovery_now + time::Duration::minutes(5), + }, + ) + .await + .unwrap() + .expect("expired claim was not recovered"); + assert_eq!( + registry + .finalize_artifact_reconciliation_claim( + &claim, + ArtifactClaimFinalization::Deleted, + timestamp("2026-08-27T10:05:01Z"), + ) + .await + .unwrap(), + ArtifactClaimFinalizeOutcome::Stale + ); + assert_eq!( + registry + .recover_artifact_reconciliation_claim( + &winner, + ArtifactClaimFinalization::AlreadyAbsent, + timestamp("2026-08-27T10:05:02Z"), + ) + .await + .unwrap(), + ArtifactClaimFinalizeOutcome::Unavailable + ); + let republished = store.put_registered(b"unreferenced\n").unwrap(); + let revived = registry + .create_artifact_source(CreateArtifactSourceRequest { + workspace_id: &workspace_a, + source_id: &ArtifactSourceId::new("src_revived"), + artifact: &republished, + mime_type: "text/plain", + sensitivity: ArtifactSourceSensitivity::Internal, + created_at: timestamp("2026-08-27T10:06:00Z"), + }) + .await + .unwrap(); + assert_eq!( + revived.blob.lifecycle, + crank_registry::ArtifactBlobLifecycle::Available + ); + + let referenced = store.put_registered(b"referenced\n").unwrap(); + let active_id = ArtifactSourceId::new("src_active"); + let active = registry + .create_artifact_source(CreateArtifactSourceRequest { + workspace_id: &workspace_b, + source_id: &active_id, + artifact: &referenced, + mime_type: "text/plain", + sensitivity: ArtifactSourceSensitivity::Internal, + created_at: started, + }) + .await + .unwrap(); + assert!(matches!( + registry + .claim_artifact_reconciliation(ClaimArtifactReconciliationRequest { + artifact: &referenced, + token: ArtifactClaimToken::generate(), + claimed_at: timestamp("2026-08-27T10:01:00Z"), + lease_expires_at: lease, + detached_grace: time::Duration::hours(24), + }) + .await + .unwrap(), + ArtifactClaimOutcome::ActiveReference + )); + registry + .detach_artifact_source(DetachArtifactSourceRequest { + workspace_id: &workspace_b, + source_id: &active_id, + expected_updated_at: Some(active.updated_at), + detached_at: timestamp("2026-08-27T10:02:00Z"), + }) + .await + .unwrap(); + assert!(matches!( + registry + .claim_artifact_reconciliation(ClaimArtifactReconciliationRequest { + artifact: &referenced, + token: ArtifactClaimToken::generate(), + claimed_at: timestamp("2026-08-27T10:03:00Z"), + lease_expires_at: lease, + detached_grace: time::Duration::hours(24), + }) + .await + .unwrap(), + ArtifactClaimOutcome::DetachedReferenceInGrace + )); + + let concurrent = store.put_registered(b"concurrent claim\n").unwrap(); + let first_registry = registry.clone(); + let second_registry = registry.clone(); + let first_artifact = concurrent.clone(); + let second_artifact = concurrent.clone(); + let (first, second) = tokio::join!( + async move { + first_registry + .claim_artifact_reconciliation(ClaimArtifactReconciliationRequest { + artifact: &first_artifact, + token: ArtifactClaimToken::generate(), + claimed_at: timestamp("2026-08-27T11:00:00Z"), + lease_expires_at: timestamp("2026-08-27T11:05:00Z"), + detached_grace: time::Duration::hours(24), + }) + .await + }, + async move { + second_registry + .claim_artifact_reconciliation(ClaimArtifactReconciliationRequest { + artifact: &second_artifact, + token: ArtifactClaimToken::generate(), + claimed_at: timestamp("2026-08-27T11:00:00Z"), + lease_expires_at: timestamp("2026-08-27T11:05:00Z"), + detached_grace: time::Duration::hours(24), + }) + .await + } + ); + let first = first.unwrap(); + let second = second.unwrap(); + assert_eq!( + usize::from(matches!(first, ArtifactClaimOutcome::Claimed(_))) + + usize::from(matches!(second, ArtifactClaimOutcome::Claimed(_))), + 1 + ); + assert_eq!( + usize::from(matches!(first, ArtifactClaimOutcome::HeldByOther)) + + usize::from(matches!(second, ArtifactClaimOutcome::HeldByOther)), + 1 + ); + + let attach_race = store.put_registered(b"attach race\n").unwrap(); + let claim_registry = registry.clone(); + let attach_registry = registry.clone(); + let claim_artifact = attach_race.clone(); + let attach_artifact = attach_race.clone(); + let attach_workspace = workspace_a.clone(); + let (claim_race, attach_race) = tokio::join!( + async move { + claim_registry + .claim_artifact_reconciliation(ClaimArtifactReconciliationRequest { + artifact: &claim_artifact, + token: ArtifactClaimToken::generate(), + claimed_at: timestamp("2026-08-27T12:00:00Z"), + lease_expires_at: timestamp("2026-08-27T12:05:00Z"), + detached_grace: time::Duration::hours(24), + }) + .await + }, + async move { + attach_registry + .create_artifact_source(CreateArtifactSourceRequest { + workspace_id: &attach_workspace, + source_id: &ArtifactSourceId::new("src_attach_race"), + artifact: &attach_artifact, + mime_type: "text/plain", + sensitivity: ArtifactSourceSensitivity::Internal, + created_at: timestamp("2026-08-27T12:00:00Z"), + }) + .await + } + ); + match (claim_race.unwrap(), attach_race) { + (ArtifactClaimOutcome::ActiveReference, Ok(_)) => {} + (ArtifactClaimOutcome::Claimed(_), Err(RegistryError::ArtifactClaimInProgress)) => {} + (claim, attach) => panic!("unsafe claim/attach race outcome: {claim:?}, {attach:?}"), + } + + database.cleanup().await; +}