diff --git a/apps/admin-api/Dockerfile b/apps/admin-api/Dockerfile index 8c5e6f4..4f921c2 100644 --- a/apps/admin-api/Dockerfile +++ b/apps/admin-api/Dockerfile @@ -24,9 +24,14 @@ WORKDIR /app COPY --from=builder /tmp/admin-api /usr/local/bin/admin-api COPY --from=builder /tmp/crank-migrate /usr/local/bin/crank-migrate +COPY apps/admin-api/docker-entrypoint.sh /usr/local/bin/crank-admin-entrypoint + +RUN chmod 0755 /usr/local/bin/crank-admin-entrypoint ENV CRANK_ADMIN_BIND=0.0.0.0:3001 +ENV CRANK_STORAGE_ROOT=/var/lib/crank/storage EXPOSE 3001 +ENTRYPOINT ["/usr/local/bin/crank-admin-entrypoint"] CMD ["admin-api"] diff --git a/apps/admin-api/docker-entrypoint.sh b/apps/admin-api/docker-entrypoint.sh new file mode 100644 index 0000000..075c645 --- /dev/null +++ b/apps/admin-api/docker-entrypoint.sh @@ -0,0 +1,25 @@ +#!/bin/sh +set -eu + +# Docker creates a fresh named volume as root:root 0755. The artifact store +# deliberately rejects that mode: before starting the API, provision exactly +# the private root its pinned-directory checks require. +if [ "$#" -gt 0 ] && [ "$1" = "admin-api" ]; then + storage_root="${CRANK_STORAGE_ROOT:-/var/lib/crank/storage}" + case "$storage_root" in + /*) ;; + *) + echo "CRANK_STORAGE_ROOT must be an absolute path" >&2 + exit 64 + ;; + esac + if [ -L "$storage_root" ] || { [ -e "$storage_root" ] && [ ! -d "$storage_root" ]; }; then + echo "CRANK_STORAGE_ROOT must be a directory, not a symlink or file" >&2 + exit 64 + fi + umask 077 + mkdir -p -- "$storage_root" + chmod 0700 -- "$storage_root" +fi + +exec "$@" diff --git a/apps/admin-api/src/dto.rs b/apps/admin-api/src/dto.rs index 9295358..4f22489 100644 --- a/apps/admin-api/src/dto.rs +++ b/apps/admin-api/src/dto.rs @@ -560,14 +560,15 @@ pub struct OpenApiImportCreateResponse { pub findings: Vec, } -#[derive(Clone, Debug, Serialize)] +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] pub struct OpenApiImportCreatedOperation { + pub operation_key: String, pub operation_id: String, pub name: String, pub version: u32, } -#[derive(Clone, Debug, Serialize)] +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] pub struct OpenApiImportSkippedOperation { pub operation_key: String, pub name: String, diff --git a/apps/admin-api/src/error.rs b/apps/admin-api/src/error.rs index 8e8b1fb..cbc4bc1 100644 --- a/apps/admin-api/src/error.rs +++ b/apps/admin-api/src/error.rs @@ -195,14 +195,6 @@ impl ApiError { } } - pub(crate) fn source_unavailable() -> Self { - Self::openapi_upload(OpenApiUploadLocale::En, "source_unavailable") - } - - pub(crate) fn source_integrity() -> Self { - Self::openapi_upload(OpenApiUploadLocale::En, "source_integrity") - } - pub(crate) fn rate_limited_with_context(message: impl Into, context: Value) -> Self { Self::RateLimited { message: message.into(), diff --git a/apps/admin-api/src/main.rs b/apps/admin-api/src/main.rs index 30a91c0..5270131 100644 --- a/apps/admin-api/src/main.rs +++ b/apps/admin-api/src/main.rs @@ -204,7 +204,6 @@ async fn run( verified_startup_secret_crypto(®istry, config.runtime.master_key.expose_secret()) .await?; let artifact_store = open_reconciliation_store(config.storage_root.clone()).await?; - registry.delete_expired_import_jobs().await?; let outbound_http_policy = crank_runtime::OutboundHttpPolicy::try_new_with_limits( config.runtime.outbound.allowed_hosts.clone(), config.runtime.outbound.denied_hosts.clone(), diff --git a/apps/admin-api/src/reconciliation.rs b/apps/admin-api/src/reconciliation.rs index 6dc5c91..8488742 100644 --- a/apps/admin-api/src/reconciliation.rs +++ b/apps/admin-api/src/reconciliation.rs @@ -1,5 +1,6 @@ use std::{ path::PathBuf, + sync::Arc, time::{Duration, SystemTime}, }; @@ -7,7 +8,7 @@ use async_trait::async_trait; use crank_artifacts::{ ArtifactError, ArtifactStore, ReconciliationCursor, ReconciliationMutation, ReconciliationNamespace, ReconciliationPresence, ReconciliationRegistration, - ReconciliationScanStop, + ReconciliationScanStop, TempScanCursor, }; use crank_registry::{ ArtifactClaimFinalization, ArtifactClaimFinalizeOutcome, ArtifactClaimOutcome, @@ -25,6 +26,11 @@ 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; +pub const RECONCILIATION_TICK_DEADLINE: Duration = Duration::from_secs(30); +pub const TEMP_CLEANUP_GRACE: Duration = Duration::from_secs(60 * 60); +pub const TEMP_CLEANUP_SCAN_LIMIT: usize = 512; +pub const TEMP_CLEANUP_DELETE_LIMIT: usize = 16; +pub const IMPORT_JOB_CLEANUP_LIMIT: u32 = 128; #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum Phase { @@ -191,6 +197,8 @@ pub enum ReconciliationCoordinatorError { Backend, #[error("artifact reconciliation backend exceeded its bounded contract")] Bounds, + #[error("artifact reconciliation backend exceeded its deadline")] + Deadline, } #[async_trait] @@ -215,7 +223,7 @@ struct ReconciliationCoordinator { #[derive(Clone)] struct PostgresReconciliationBackend { registry: PostgresRegistry, - store: ArtifactStore, + store: Arc, } enum PostgresReconciliationCursor { @@ -230,7 +238,7 @@ impl std::fmt::Debug for PostgresReconciliationCursor { } impl PostgresReconciliationBackend { - fn new(registry: PostgresRegistry, store: ArtifactStore) -> Self { + fn new(registry: PostgresRegistry, store: Arc) -> Self { Self { registry, store } } @@ -245,7 +253,7 @@ impl PostgresReconciliationBackend { &self, artifact_ref: crank_artifacts::ArtifactRef, ) -> Result { - let store = self.store.clone(); + let store = Arc::clone(&self.store); let result = tokio::task::spawn_blocking(move || store.reconciliation_presence(&artifact_ref)) .await @@ -518,7 +526,7 @@ impl ReconciliationBackend for PostgresReconciliationBackend { Phase::Recovery => ReconciliationNamespace::Quarantine, Phase::Sweep => ReconciliationNamespace::Final, }; - let store = self.store.clone(); + let store = Arc::clone(&self.store); let (mut scan, candidates) = tokio::task::spawn_blocking(move || { store.scan_reconciliation_namespace( namespace, @@ -560,7 +568,7 @@ impl ReconciliationBackend for PostgresReconciliationBackend { Phase::Recovery => Duration::ZERO, Phase::Sweep => RECONCILIATION_GRACE, }; - let store = self.store.clone(); + let store = Arc::clone(&self.store); let registration = tokio::task::spawn_blocking(move || { let registration = store.register_reconciliation( &candidate, @@ -621,7 +629,7 @@ impl ReconciliationBackend for PostgresReconciliationBackend { step.mutation_attempts += 1; step.physical_mutation_possible = true; - let store = self.store.clone(); + let store = Arc::clone(&self.store); let mutation = match tokio::task::spawn_blocking(move || match phase { Phase::Recovery => store.delete_quarantined_reconciliation(candidate), Phase::Sweep => store.quarantine_reconciliation(candidate), @@ -717,8 +725,8 @@ impl ReconciliationBackend for PostgresReconciliationBackend { /// async executor. pub async fn open_reconciliation_store( root: PathBuf, -) -> Result { - tokio::task::spawn_blocking(move || ArtifactStore::open(root)) +) -> Result, ReconciliationCoordinatorError> { + tokio::task::spawn_blocking(move || ArtifactStore::open(root).map(Arc::new)) .await .map_err(|_| ReconciliationCoordinatorError::Backend)? .map_err(|_| ReconciliationCoordinatorError::Backend) @@ -726,10 +734,25 @@ pub async fn open_reconciliation_store( /// 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); +pub async fn spawn_artifact_reconciliation(registry: PostgresRegistry, store: Arc) { + let cleanup_registry = registry.clone(); + let backend = PostgresReconciliationBackend::new(registry, Arc::clone(&store)); let mut coordinator = ReconciliationCoordinator::new(backend); - observe_tick(coordinator.tick().await); + let mut temp_cursor = None; + let startup_maintenance = async { + observe_import_job_cleanup(&cleanup_registry).await; + observe_tick(run_bounded_reconciliation_tick(&mut coordinator).await); + temp_cursor = observe_stale_temp_cleanup(Arc::clone(&store), None).await; + }; + if tokio::time::timeout(RECONCILIATION_TICK_DEADLINE, startup_maintenance) + .await + .is_err() + { + warn!( + name: "admin.artifact_maintenance.startup_deadline", + "startup maintenance exceeded its shared deadline; remaining work will resume in background" + ); + } tokio::spawn(async move { let mut interval = tokio::time::interval(RECONCILIATION_INTERVAL); @@ -738,11 +761,148 @@ pub async fn spawn_artifact_reconciliation(registry: PostgresRegistry, store: Ar interval.tick().await; loop { interval.tick().await; - observe_tick(coordinator.tick().await); + observe_import_job_cleanup(&cleanup_registry).await; + observe_tick(run_bounded_reconciliation_tick(&mut coordinator).await); + temp_cursor = observe_stale_temp_cleanup(Arc::clone(&store), temp_cursor).await; } }); } +async fn observe_import_job_cleanup(registry: &PostgresRegistry) { + let cleanup = async { + let mut deleted_jobs = 0_u64; + let mut detached_sources = 0_u64; + loop { + let report = registry + .cleanup_expired_import_jobs(IMPORT_JOB_CLEANUP_LIMIT) + .await?; + deleted_jobs = deleted_jobs.saturating_add(report.deleted_jobs); + detached_sources = detached_sources.saturating_add(report.detached_sources); + if !report.more_work { + return Ok::<_, crank_registry::RegistryError>((deleted_jobs, detached_sources)); + } + // Each pass commits independently. Yield before requesting the + // next lock batch so normal import traffic can make progress. + tokio::task::yield_now().await; + } + }; + match tokio::time::timeout(RECONCILIATION_TICK_DEADLINE, cleanup).await { + Ok(Ok((deleted_jobs, detached_sources))) => info!( + name: "admin.openapi_import_cleanup.tick", + deleted_jobs, + detached_sources, + more_work = false, + "OpenAPI import cleanup tick completed" + ), + Ok(Err(_)) => warn!( + name: "admin.openapi_import_cleanup.failed", + error_category = "registry", + "OpenAPI import cleanup tick will be retried" + ), + Err(_) => warn!( + name: "admin.openapi_import_cleanup.failed", + error_category = "deadline", + "OpenAPI import cleanup tick exceeded its deadline" + ), + } +} + +async fn run_bounded_reconciliation_tick( + coordinator: &mut ReconciliationCoordinator, +) -> Result { + tokio::time::timeout(RECONCILIATION_TICK_DEADLINE, coordinator.tick()) + .await + .map_err(|_| ReconciliationCoordinatorError::Deadline)? +} + +async fn observe_stale_temp_cleanup( + store: Arc, + cursor: Option, +) -> Option { + let progress = Arc::new(std::sync::Mutex::new(cursor)); + let cleanup_progress = Arc::clone(&progress); + let cleanup = async move { + let mut scanned = 0_usize; + let mut omitted = 0_usize; + let mut deleted = 0_usize; + let mut retryable = 0_usize; + loop { + let scanner = Arc::clone(&store); + let cursor = cleanup_progress + .lock() + .map_err(|_| ArtifactError::Storage)? + .clone(); + let (scan, candidates, next_cursor) = tokio::task::spawn_blocking(move || { + scanner.scan_stale_temps_after( + TEMP_CLEANUP_GRACE, + cursor, + TEMP_CLEANUP_SCAN_LIMIT, + TEMP_CLEANUP_DELETE_LIMIT, + ) + }) + .await + .map_err(|_| ArtifactError::Storage)??; + scanned = scanned.saturating_add(scan.scanned); + omitted = omitted.saturating_add(scan.omitted); + *cleanup_progress + .lock() + .map_err(|_| ArtifactError::Storage)? = Some(next_cursor); + + for candidate in candidates { + let store = Arc::clone(&store); + match tokio::task::spawn_blocking(move || store.delete_stale_temp(candidate)).await + { + Ok(Ok(())) => deleted += 1, + Ok(Err(_)) | Err(_) => retryable += 1, + } + } + if scan.complete { + return Ok::<_, ArtifactError>(( + scanned, + omitted, + deleted, + retryable, + cleanup_progress + .lock() + .map_err(|_| ArtifactError::Storage)? + .clone(), + )); + } + tokio::task::yield_now().await; + } + }; + + match tokio::time::timeout(RECONCILIATION_TICK_DEADLINE, cleanup).await { + Ok(Ok((scanned, omitted, deleted, retryable, next_cursor))) => { + info!( + name: "admin.artifact_temp_cleanup.tick", + scanned, + omitted, + deleted, + retryable, + "artifact temporary-file cleanup tick completed" + ); + next_cursor + } + Ok(Err(_)) => { + warn!( + name: "admin.artifact_temp_cleanup.failed", + error_category = "backend", + "artifact temporary-file cleanup tick will be retried" + ); + progress.lock().ok().and_then(|cursor| cursor.clone()) + } + Err(_) => { + warn!( + name: "admin.artifact_temp_cleanup.failed", + error_category = "deadline", + "artifact temporary-file cleanup tick exceeded its deadline" + ); + progress.lock().ok().and_then(|cursor| cursor.clone()) + } + } +} + fn observe_tick(result: Result) { match result { Ok(report) => info!( @@ -772,6 +932,7 @@ fn observe_tick(result: Result "backend", ReconciliationCoordinatorError::Bounds => "bounds", + ReconciliationCoordinatorError::Deadline => "deadline", }, "artifact reconciliation tick will be retried" ), diff --git a/apps/admin-api/src/reconciliation/tests.rs b/apps/admin-api/src/reconciliation/tests.rs index b567c6f..75dc5cb 100644 --- a/apps/admin-api/src/reconciliation/tests.rs +++ b/apps/admin-api/src/reconciliation/tests.rs @@ -518,7 +518,7 @@ async fn production_backend_recovers_absence_before_sweep_and_restarts_after_mut 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 backend = PostgresReconciliationBackend::new(registry.clone(), Arc::new(store.clone())); let mut coordinator = ReconciliationCoordinator::with_tick_limits( backend, StepLimits { @@ -660,7 +660,7 @@ async fn expired_claim_cursor_prevents_unsafe_prefix_starvation_across_ticks() { .unwrap(); remove_registered_artifact(&store, &absent); - let backend = PostgresReconciliationBackend::new(registry.clone(), store); + let backend = PostgresReconciliationBackend::new(registry.clone(), Arc::new(store)); let mut coordinator = ReconciliationCoordinator::new(backend); let first = coordinator.tick().await.unwrap(); assert_eq!(first.candidates, RECONCILIATION_CANDIDATE_LIMIT); diff --git a/apps/admin-api/src/routes.rs b/apps/admin-api/src/routes.rs index 47255b2..87f4ae1 100644 --- a/apps/admin-api/src/routes.rs +++ b/apps/admin-api/src/routes.rs @@ -24,23 +24,89 @@ pub async fn health() -> Json { } pub async fn readiness(State(state): State) -> impl IntoResponse { - match state.service.readiness().await { - Ok(()) => ( + let checks = state.service.readiness().await; + let postgres = if checks.postgres { + "ready" + } else { + "not_ready" + }; + let artifact_storage = if checks.artifact_storage { + "ready" + } else { + "not_ready" + }; + if checks.is_ready() { + ( StatusCode::OK, Json(json!({ "service": "admin-api", "status": "ready", - "checks": { "postgres": "ready" } + "checks": { "postgres": postgres, "artifact_storage": artifact_storage } })), - ), - Err(error) => ( + ) + } else { + ( StatusCode::SERVICE_UNAVAILABLE, Json(json!({ "service": "admin-api", "status": "not_ready", - "checks": { "postgres": "not_ready" }, - "error": error.to_string() + "checks": { "postgres": postgres, "artifact_storage": artifact_storage } })), - ), + ) + } +} + +#[cfg(test)] +mod tests { + use crate::service::ReadinessChecks; + + #[test] + fn readiness_checks_identify_a_postgres_failure() { + let checks = ReadinessChecks { + postgres: false, + artifact_storage: true, + }; + assert!(!checks.is_ready()); + assert_eq!( + if checks.postgres { + "ready" + } else { + "not_ready" + }, + "not_ready" + ); + assert_eq!( + if checks.artifact_storage { + "ready" + } else { + "not_ready" + }, + "ready" + ); + } + + #[test] + fn readiness_checks_identify_an_artifact_storage_failure() { + let checks = ReadinessChecks { + postgres: true, + artifact_storage: false, + }; + assert!(!checks.is_ready()); + assert_eq!( + if checks.postgres { + "ready" + } else { + "not_ready" + }, + "ready" + ); + assert_eq!( + if checks.artifact_storage { + "ready" + } else { + "not_ready" + }, + "not_ready" + ); } } diff --git a/apps/admin-api/src/routes/imports.rs b/apps/admin-api/src/routes/imports.rs index de24ec6..ac1f50d 100644 --- a/apps/admin-api/src/routes/imports.rs +++ b/apps/admin-api/src/routes/imports.rs @@ -70,7 +70,16 @@ async fn parse_openapi_upload( .ok_or_else(|| ApiError::openapi_upload(locale, "invalid_filename"))?; let mime_type = field .content_type() - .map(ToString::to_string) + // Axum exposes the raw header value here. Persist only the MIME + // essence so parameters and case cannot make route and service + // validation disagree. + .map(|mime| { + mime.split(';') + .next() + .unwrap_or_default() + .trim() + .to_ascii_lowercase() + }) .ok_or_else(|| ApiError::openapi_upload(locale, "invalid_media_type"))?; if !valid_upload_type(filename, &mime_type) { return Err(ApiError::openapi_upload(locale, "invalid_media_type")); @@ -123,35 +132,90 @@ fn valid_upload_type(filename: &str, mime_type: &str) -> bool { } fn openapi_upload_locale(headers: &HeaderMap) -> OpenApiUploadLocale { - let prefers_russian = headers + let Some(value) = headers .get("accept-language") .and_then(|value| value.to_str().ok()) - .is_some_and(|value| { - value.split(',').any(|range| { - let language = range.split(';').next().unwrap_or_default().trim(); - language.eq_ignore_ascii_case("ru") - || language.to_ascii_lowercase().starts_with("ru-") + else { + return OpenApiUploadLocale::En; + }; + + // RFC 9110: highest q wins; ties preserve the header's order. Only the + // locales served by this endpoint participate in negotiation. + let mut preferred = (0_u16, usize::MAX, OpenApiUploadLocale::En); + for (index, range) in value.split(',').enumerate() { + let mut parts = range.split(';'); + let language = parts.next().unwrap_or_default().trim(); + let locale = if language.eq_ignore_ascii_case("ru") + || language.to_ascii_lowercase().starts_with("ru-") + { + Some(OpenApiUploadLocale::Ru) + } else if language.eq_ignore_ascii_case("en") + || language.to_ascii_lowercase().starts_with("en-") + || language == "*" + { + Some(OpenApiUploadLocale::En) + } else { + None + }; + let Some(locale) = locale else { continue }; + let quality = match parts + .filter_map(|parameter| { + let (name, value) = parameter.trim().split_once('=')?; + name.eq_ignore_ascii_case("q").then_some(value.trim()) }) - }); - if prefers_russian { - OpenApiUploadLocale::Ru - } else { - OpenApiUploadLocale::En + .next() + { + None => Some(1_000_u16), + Some(value) => value + .parse::() + .ok() + .filter(|quality| (0.0..=1.0).contains(quality) && *quality > 0.0) + .map(|quality| (quality * 1_000.0).round() as u16), + }; + let Some(quality) = quality else { continue }; + if quality > preferred.0 || (quality == preferred.0 && index < preferred.1) { + preferred = (quality, index, locale); + } } + preferred.2 } pub async fn create_openapi_import( Path(path): Path, State(state): State, + headers: HeaderMap, Json(payload): Json, ) -> Result, ApiError> { let imported = state .service - .create_openapi_import( + .create_openapi_import_with_locale( &path.workspace_id.as_str().into(), &path.job_id.as_str().into(), payload, + openapi_upload_locale(&headers), ) .await?; Ok(Json(json!(imported))) } + +#[cfg(test)] +mod tests { + use axum::http::{HeaderMap, HeaderValue}; + + use super::*; + + fn locale(value: &str) -> OpenApiUploadLocale { + let mut headers = HeaderMap::new(); + headers.insert("accept-language", HeaderValue::from_str(value).unwrap()); + openapi_upload_locale(&headers) + } + + #[test] + fn accept_language_honours_quality_zero_and_header_order() { + assert_eq!(locale("ru;q=0, en;q=0.5"), OpenApiUploadLocale::En); + assert_eq!(locale("en;q=0.5, ru;q=0.5"), OpenApiUploadLocale::En); + assert_eq!(locale("ru-RU;q=0.9, en;q=1"), OpenApiUploadLocale::En); + assert_eq!(locale("ru, en;q=0.5"), OpenApiUploadLocale::Ru); + assert_eq!(locale("ru;q=bad, en;q=0.5"), OpenApiUploadLocale::En); + } +} diff --git a/apps/admin-api/src/service.rs b/apps/admin-api/src/service.rs index e906410..9ef758c 100644 --- a/apps/admin-api/src/service.rs +++ b/apps/admin-api/src/service.rs @@ -74,7 +74,7 @@ pub struct AdminService { pub struct AdminServiceBuilder { registry: PostgresRegistry, storage_root: PathBuf, - artifact_store: Option, + artifact_store: Option>, auth_settings: AuthSettings, secret_crypto: SecretCrypto, runtime: RuntimeExecutor, @@ -88,6 +88,18 @@ pub struct AdminServiceBuilder { pub use crate::dto::*; +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct ReadinessChecks { + pub postgres: bool, + pub artifact_storage: bool, +} + +impl ReadinessChecks { + pub const fn is_ready(self) -> bool { + self.postgres && self.artifact_storage + } +} + #[derive(Clone, Debug)] pub struct AdminAuditContext { actor: AuditActor, @@ -124,9 +136,16 @@ pub(super) struct CredentialAuditRecord<'a> { } impl AdminService { - pub async fn readiness(&self) -> Result<(), ApiError> { - self.registry.ping().await?; - Ok(()) + pub async fn readiness(&self) -> ReadinessChecks { + let postgres = self.registry.ping().await.is_ok(); + let store = Arc::clone(&self.artifact_store); + let artifact_storage = tokio::task::spawn_blocking(move || store.check_health()) + .await + .is_ok_and(|result| result.is_ok()); + ReadinessChecks { + postgres, + artifact_storage, + } } #[cfg(test)] @@ -230,7 +249,7 @@ impl AdminServiceBuilder { /// Reuses the process-wide immutable artifact authority for OpenAPI /// ingress and reconciliation. Tests may omit this and use their private /// storage root instead. - pub fn with_artifact_store(mut self, artifact_store: ArtifactStore) -> Self { + pub fn with_artifact_store(mut self, artifact_store: Arc) -> Self { self.artifact_store = Some(artifact_store); self } @@ -264,10 +283,9 @@ impl AdminServiceBuilder { pub fn build(self) -> AdminService { AdminService { registry: self.registry, - artifact_store: Arc::new( - self.artifact_store - .unwrap_or_else(|| ArtifactStore::new(&self.storage_root)), - ), + artifact_store: self + .artifact_store + .unwrap_or_else(|| Arc::new(ArtifactStore::new(&self.storage_root))), runtime: self.runtime, storage: LocalArtifactStorage::new(self.storage_root), auth_settings: self.auth_settings, diff --git a/apps/admin-api/src/service/imports.rs b/apps/admin-api/src/service/imports.rs index 3a4026e..fd0f033 100644 --- a/apps/admin-api/src/service/imports.rs +++ b/apps/admin-api/src/service/imports.rs @@ -29,6 +29,10 @@ use crate::{ }; const IMPORT_JOB_TTL_HOURS: i64 = 24; +const OPENAPI_PARSE_DEADLINE: std::time::Duration = std::time::Duration::from_secs(30); +const OPENAPI_PARSE_CONCURRENCY: usize = 4; +static OPENAPI_PARSE_SLOTS: tokio::sync::Semaphore = + tokio::sync::Semaphore::const_new(OPENAPI_PARSE_CONCURRENCY); impl AdminService { #[instrument(skip(self, upload), fields(workspace_id = %workspace_id.as_str()))] @@ -38,7 +42,6 @@ impl AdminService { upload: OpenApiUpload, ) -> Result { self.ensure_workspace_exists(workspace_id).await?; - self.registry.delete_expired_import_jobs().await?; validate_openapi_upload(&upload)?; let OpenApiUpload { @@ -72,19 +75,38 @@ impl AdminService { source_id.clone(), source.updated_at, ); - let verified = self + let verified = match self .registry - .read_artifact_source(&self.artifact_store, workspace_id, &source_id) - .await?; + .read_artifact_source( + std::sync::Arc::clone(&self.artifact_store), + workspace_id, + &source_id, + ) + .await + { + Ok(verified) => verified, + Err(error) => { + detach_guard.detach_now().await; + return Err(ApiError::from(error)); + } + }; if verified.source.blob.artifact_ref != *artifact.artifact_ref() { + detach_guard.detach_now().await; return Err(ApiError::openapi_upload(locale, "source_integrity")); } - let preview = parse_verified_preview(verified.bytes, locale).await?; + let preview = match parse_verified_preview(verified.bytes, locale).await { + Ok(preview) => preview, + Err(error) => { + detach_guard.detach_now().await; + return Err(error); + } + }; if preview .groups .iter() .all(|group| group.operations.is_empty()) { + detach_guard.detach_now().await; return Err(ApiError::openapi_upload(locale, "no_methods")); } let source_envelope = ImportJobSourceEnvelope { @@ -93,15 +115,18 @@ impl AdminService { }; let preview_value = serde_json::to_value(&preview) .map_err(|error| ApiError::internal(error.to_string()))?; + let preview_digest = preview_digest(&preview_value)?; let preview_payload = json!({ "source": { "source_id": source_envelope.source_id.as_str(), "digest": source_envelope.digest.as_str(), }, "preview": preview_value, + "preview_digest": preview_digest, }); - self.registry + if let Err(error) = self + .registry .create_import_job(CreateImportJobRequest { id: &job_id, workspace_id, @@ -114,7 +139,11 @@ impl AdminService { created_at: &now, expires_at: &expires_at, }) - .await?; + .await + { + detach_guard.detach_now().await; + return Err(ApiError::from(error)); + } detach_guard.disarm(); Ok(OpenApiImportPreviewResponse { @@ -132,9 +161,24 @@ impl AdminService { workspace_id: &WorkspaceId, job_id: &ImportJobId, payload: OpenApiImportCreatePayload, + ) -> Result { + self.create_openapi_import_with_locale( + workspace_id, + job_id, + payload, + OpenApiUploadLocale::En, + ) + .await + } + + pub async fn create_openapi_import_with_locale( + &self, + workspace_id: &WorkspaceId, + job_id: &ImportJobId, + payload: OpenApiImportCreatePayload, + locale: OpenApiUploadLocale, ) -> Result { self.ensure_workspace_exists(workspace_id).await?; - self.registry.delete_expired_import_jobs().await?; if !matches!(payload.conflict_mode.as_str(), "skip" | "rename") { return Err(ApiError::validation( @@ -185,22 +229,56 @@ impl AdminService { application_key: &application_key, conflict_mode, operations: &[], + pre_skipped: &[], finished_at: &finished_at, }) .await?; - return Ok(openapi_import_response(applied, Vec::new())); + return Ok(openapi_import_response(applied)); } - let source = import_job_source(&job.preview_payload)?; - let verified = self + let source = import_job_source(&job.preview_payload, locale)?; + let verified = match self .registry - .read_artifact_source(&self.artifact_store, workspace_id, &source.source_id) + .read_artifact_source( + std::sync::Arc::clone(&self.artifact_store), + workspace_id, + &source.source_id, + ) .await - .map_err(openapi_source_error)?; + { + Ok(verified) => verified, + Err( + error @ (RegistryError::SourceNotFound { .. } | RegistryError::SourceUnavailable), + ) => { + // Another request may have finished and detached the source + // after our unlocked Pending read. Re-read the job and ask + // the registry for its canonical, locked replay instead of + // exposing a false `source_unavailable` outcome. + let latest = self.registry.get_import_job(workspace_id, job_id).await?; + if latest.is_some_and(|latest| latest.status == ImportJobStatus::Completed) { + let applied = self + .registry + .apply_import_job(ApplyImportJobRequest { + id: job_id, + workspace_id, + application_key: &application_key, + conflict_mode, + operations: &[], + pre_skipped: &[], + finished_at: &finished_at, + }) + .await?; + return Ok(openapi_import_response(applied)); + } + return Err(openapi_source_error(locale, error)); + } + Err(error) => return Err(openapi_source_error(locale, error)), + }; if verified.source.blob.artifact_ref != source.digest { - return Err(ApiError::source_integrity()); + return Err(ApiError::openapi_upload(locale, "source_integrity")); } - let preview = parse_verified_preview(verified.bytes, OpenApiUploadLocale::En).await?; + let preview = parse_verified_preview(verified.bytes, locale).await?; + verify_preview_contract(&job.preview_payload, &preview, locale)?; let mut candidates = BTreeMap::new(); for group in &preview.groups { for operation in &group.operations { @@ -213,7 +291,7 @@ impl AdminService { for operation_key in selected { let Some(candidate) = candidates.get(&operation_key) else { - skipped.push(OpenApiImportSkippedOperation { + skipped.push(crank_registry::SkippedImportOperation { operation_key, name: String::new(), reason: "operation was not found in import preview".to_owned(), @@ -261,22 +339,21 @@ impl AdminService { application_key: &application_key, conflict_mode, operations: &operations, + pre_skipped: &skipped, finished_at: &finished_at, }) .await?; - Ok(openapi_import_response(applied, skipped)) + Ok(openapi_import_response(applied)) } } -fn openapi_import_response( - applied: ImportJobApplyResult, - mut skipped: Vec, -) -> OpenApiImportCreateResponse { +fn openapi_import_response(applied: ImportJobApplyResult) -> OpenApiImportCreateResponse { let created = applied .created .iter() .map(|operation| OpenApiImportCreatedOperation { + operation_key: operation.operation_key.clone(), operation_id: operation.operation_id.as_str().to_owned(), name: operation.name.clone(), version: operation.version, @@ -300,22 +377,29 @@ fn openapi_import_response( }) }) .collect::>(); - for operation in applied.skipped { - skipped.push(OpenApiImportSkippedOperation { - operation_key: operation.operation_key.clone(), - name: operation.name.clone(), - reason: "operation with this name already exists".to_owned(), - }); - findings.push(ImportFinding { - code: operation.reason, - severity: ImportFindingSeverity::Warning, - message: format!( - "Операция {} уже существует и была пропущена.", - operation.name - ), - operation_key: Some(operation.operation_key), - }); - } + let skipped = applied + .skipped + .into_iter() + .map(|operation| { + let reason = operation.reason.clone(); + let name = operation.name.clone(); + findings.push(ImportFinding { + code: reason.clone(), + severity: ImportFindingSeverity::Warning, + message: if name.is_empty() { + "Выбранная операция отсутствует в исходном preview и была пропущена.".to_owned() + } else { + format!("Операция {name} уже существует и была пропущена.") + }, + operation_key: Some(operation.operation_key.clone()), + }); + OpenApiImportSkippedOperation { + operation_key: operation.operation_key.clone(), + name: operation.name, + reason, + } + }) + .collect::>(); info!( name: "admin.openapi_import.completed", created = created.len(), @@ -361,14 +445,28 @@ async fn parse_verified_preview( bytes: Vec, locale: OpenApiUploadLocale, ) -> Result { - tokio::task::spawn_blocking(move || { + let started = tokio::time::Instant::now(); + let permit = tokio::time::timeout(OPENAPI_PARSE_DEADLINE, OPENAPI_PARSE_SLOTS.acquire()) + .await + .map_err(|_| ApiError::openapi_upload(locale, "parser_unavailable"))? + .map_err(|_| ApiError::openapi_upload(locale, "parser_unavailable"))?; + let parsing = tokio::task::spawn_blocking(move || { + // Keep the permit inside the blocking task. Timing out the caller + // cannot cancel CPU work already running, but abandoned parsers remain + // globally bounded and release capacity when they actually finish. + let _permit = permit; let document = std::str::from_utf8(&bytes) .map_err(|_| ApiError::openapi_upload(locale, "invalid_utf8"))?; crank_import::rest::preview_document(document) .map_err(|_| ApiError::openapi_upload(locale, "invalid_document")) - }) - .await - .map_err(|_| ApiError::openapi_upload(locale, "parser_unavailable"))? + }); + let remaining = OPENAPI_PARSE_DEADLINE + .checked_sub(started.elapsed()) + .unwrap_or(std::time::Duration::ZERO); + tokio::time::timeout(remaining, parsing) + .await + .map_err(|_| ApiError::openapi_upload(locale, "parser_unavailable"))? + .map_err(|_| ApiError::openapi_upload(locale, "parser_unavailable"))? } fn artifact_error(locale: OpenApiUploadLocale, error: ArtifactError) -> ApiError { @@ -383,30 +481,63 @@ fn artifact_error(locale: OpenApiUploadLocale, error: ArtifactError) -> ApiError } } -fn openapi_source_error(error: RegistryError) -> ApiError { +fn openapi_source_error(locale: OpenApiUploadLocale, error: RegistryError) -> ApiError { match error { RegistryError::SourceNotFound { .. } | RegistryError::SourceUnavailable => { - ApiError::source_unavailable() + ApiError::openapi_upload(locale, "source_unavailable") } - RegistryError::SourceIntegrity => ApiError::source_integrity(), + RegistryError::SourceIntegrity => ApiError::openapi_upload(locale, "source_integrity"), other => ApiError::from(other), } } -fn import_job_source(payload: &serde_json::Value) -> Result { +fn preview_digest(preview: &serde_json::Value) -> Result { + let canonical = + serde_json::to_vec(preview).map_err(|error| ApiError::internal(error.to_string()))?; + Ok(format!("{:x}", Sha256::digest(canonical))) +} + +fn verify_preview_contract( + payload: &serde_json::Value, + preview: &crank_import::rest::ImportPreview, + locale: OpenApiUploadLocale, +) -> Result<(), ApiError> { + // Pre-fingerprint jobs are legacy rolling-upgrade records. They retain the + // old reparse behavior; new jobs fail closed if parser output drifts or a + // persisted preview has been changed. + let Some(expected) = payload + .get("preview_digest") + .and_then(serde_json::Value::as_str) + else { + return Ok(()); + }; + let actual = preview_digest( + &serde_json::to_value(preview).map_err(|error| ApiError::internal(error.to_string()))?, + )?; + if actual == expected { + Ok(()) + } else { + Err(ApiError::openapi_upload(locale, "source_integrity")) + } +} + +fn import_job_source( + payload: &serde_json::Value, + locale: OpenApiUploadLocale, +) -> Result { let source = payload .get("source") - .ok_or_else(ApiError::source_unavailable)?; + .ok_or_else(|| ApiError::openapi_upload(locale, "source_unavailable"))?; let source_id = source .get("source_id") .and_then(serde_json::Value::as_str) .filter(|value| value.len() <= 132) - .ok_or_else(ApiError::source_unavailable)?; + .ok_or_else(|| ApiError::openapi_upload(locale, "source_unavailable"))?; let digest = source .get("digest") .and_then(serde_json::Value::as_str) .and_then(|value| value.parse().ok()) - .ok_or_else(ApiError::source_integrity)?; + .ok_or_else(|| ApiError::openapi_upload(locale, "source_integrity"))?; Ok(ImportJobSourceEnvelope { source_id: ArtifactSourceId::new(source_id), digest, @@ -440,6 +571,25 @@ impl SourceDetachGuard { fn disarm(&mut self) { self.armed = false; } + + async fn detach_now(&mut self) { + if !self.armed { + return; + } + self.armed = false; + // This is the normal error path: await the database transition so a + // caller receives a completed cleanup before it can retry. `Drop` + // remains only the cancellation/shutdown safety net. + let _ = self + .registry + .detach_artifact_source(DetachArtifactSourceRequest { + workspace_id: &self.workspace_id, + source_id: &self.source_id, + expected_updated_at: Some(self.expected_updated_at), + detached_at: OffsetDateTime::now_utc(), + }) + .await; + } } impl Drop for SourceDetachGuard { diff --git a/apps/admin-api/tests/config_startup.rs b/apps/admin-api/tests/config_startup.rs index 91262cb..bb3f6cc 100644 --- a/apps/admin-api/tests/config_startup.rs +++ b/apps/admin-api/tests/config_startup.rs @@ -1,4 +1,11 @@ -use std::process::Command; +use std::{ + fs, + io::Read, + os::unix::fs::PermissionsExt, + process::{Command, Stdio}, + sync::atomic::{AtomicU64, Ordering}, + time::{SystemTime, UNIX_EPOCH}, +}; use crank_registry::{MASTER_KEY_CIPHER_CONTRACT, MasterKeyIdentityCandidate, PostgresRegistry}; use crank_runtime::SecretCrypto; @@ -7,6 +14,7 @@ use time::{OffsetDateTime, format_description::well_known::Rfc3339}; const TEST_MASTER_KEY: &str = "test-master-key-00000000000000000000000000000000"; const REGISTERED_MASTER_KEY: &str = "registered-master-key-CANARY_SECRET_VALUE-00000000"; const WRONG_MASTER_KEY: &str = "wrong-master-key-CANARY_SECRET_VALUE-0000000000000"; +static NEXT_STORAGE_ROOT: AtomicU64 = AtomicU64::new(0); fn run_with(entries: &[(&str, &str)]) -> String { let mut command = Command::new(env!("CARGO_BIN_EXE_admin-api")); @@ -78,6 +86,105 @@ async fn fresh_database_startup_is_read_only() { assert!(!present); } +#[tokio::test] +async fn real_admin_startup_runs_immediate_import_maintenance() { + let unique = NEXT_STORAGE_ROOT.fetch_add(1, Ordering::Relaxed); + let epoch = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let database_url = + crank_test_support::postgres_schema_url(&format!("startup_maint_{epoch}_{unique}")).await; + let applied = Command::new(env!("CARGO_BIN_EXE_crank-migrate")) + .arg("apply") + .env("CRANK_DATABASE_URL", &database_url) + .output() + .expect("migration command executes"); + assert!( + applied.status.success(), + "{}", + String::from_utf8_lossy(&applied.stderr) + ); + let pool = sqlx::PgPool::connect(&database_url).await.unwrap(); + assert_eq!( + sqlx::query_scalar::<_, Option>("select max(version) from __crank_migrations") + .fetch_one(&pool) + .await + .unwrap(), + Some(13), + "migration binary must install the current canonical ledger", + ); + sqlx::query( + "insert into import_jobs ( + id, workspace_id, kind, source_format, source_version, status, + preview_payload, created_operation_ids, error_text, created_at, expires_at, finished_at + ) values ( + 'imp_startup_expired', 'ws_default', 'openapi', 'openapi', null, 'pending', + '{\"source\": {\"source_id\": 1}}'::jsonb, '[]'::jsonb, null, + now() - interval '2 hours', now() - interval '1 hour', null + )", + ) + .execute(&pool) + .await + .unwrap(); + + let storage_root = std::env::temp_dir().join(format!( + "crank-admin-startup-maintenance-{}-{epoch}-{unique}", + std::process::id() + )); + fs::create_dir(&storage_root).unwrap(); + fs::set_permissions(&storage_root, fs::Permissions::from_mode(0o700)).unwrap(); + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let bind_addr = listener.local_addr().unwrap().to_string(); + drop(listener); + + let mut command = Command::new(env!("CARGO_BIN_EXE_admin-api")); + for field in crank_config::field_registry() { + command.env_remove(field.env_name); + } + let mut child = command + .envs([ + ("CRANK_DATABASE_URL", database_url.as_str()), + ("CRANK_MASTER_KEY", TEST_MASTER_KEY), + ("CRANK_SESSION_SECRET", "session"), + ("CRANK_PASSWORD_PEPPER", "pepper"), + ("CRANK_BOOTSTRAP_ADMIN_EMAIL", "owner@example.test"), + ("CRANK_BOOTSTRAP_ADMIN_PASSWORD", "password"), + ("CRANK_ADMIN_BIND", bind_addr.as_str()), + ("CRANK_STORAGE_ROOT", storage_root.to_str().unwrap()), + ]) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .expect("admin binary starts"); + + let mut cleaned = false; + for _ in 0..200 { + if let Some(status) = child.try_wait().unwrap() { + let mut stderr = String::new(); + if let Some(mut stream) = child.stderr.take() { + stream.read_to_string(&mut stderr).unwrap(); + } + panic!("admin binary exited before startup maintenance: {status}: {stderr}"); + } + let remaining: i64 = + sqlx::query_scalar("select count(*) from import_jobs where id = 'imp_startup_expired'") + .fetch_one(&pool) + .await + .unwrap(); + if remaining == 0 { + cleaned = true; + break; + } + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + } + + child.kill().expect("admin binary can be stopped"); + child.wait().expect("admin binary can be reaped"); + fs::remove_dir_all(&storage_root).unwrap(); + assert!(cleaned, "real startup did not clean the expired import job"); +} + #[tokio::test] async fn master_key_mismatch_blocks_startup_with_safe_diagnostic() { let database_url = crank_test_support::postgres_schema_url("admin_master_key_mismatch").await; diff --git a/apps/admin-api/tests/integration/openapi_import.rs b/apps/admin-api/tests/integration/openapi_import.rs index bfe2687..9eda604 100644 --- a/apps/admin-api/tests/integration/openapi_import.rs +++ b/apps/admin-api/tests/integration/openapi_import.rs @@ -81,6 +81,7 @@ async fn previews_openapi_and_creates_draft_operations() { .unwrap(); assert_eq!(created.created.len(), 1); + assert_eq!(created.created[0].operation_key, "GET /v2/latest"); assert_eq!(created.created[0].name, "latest_rates"); assert!(created.skipped.is_empty()); @@ -176,6 +177,46 @@ async fn previews_openapi_and_creates_draft_operations() { assert_eq!(renamed.findings[0].code, "operation_name_renamed"); } +#[tokio::test] +#[serial] +async fn unknown_selected_operations_are_persisted_in_the_canonical_replay() { + let registry = test_registry().await; + let service = test_service( + registry, + test_storage_root("openapi_import_unknown_replay"), + test_auth_settings(), + test_secret_crypto(), + ); + let workspace_id = WorkspaceId::new("ws_default"); + let preview = service + .preview_openapi_import(&workspace_id, openapi_upload()) + .await + .unwrap(); + let payload = OpenApiImportCreatePayload { + selected_operation_keys: vec!["GET /v2/latest".to_owned(), "GET /missing".to_owned()], + server_url: Some("https://api.frankfurter.dev".to_owned()), + conflict_mode: "skip".to_owned(), + }; + + let first = service + .create_openapi_import( + &workspace_id, + &preview.job_id.as_str().into(), + payload.clone(), + ) + .await + .unwrap(); + let replayed = service + .create_openapi_import(&workspace_id, &preview.job_id.as_str().into(), payload) + .await + .unwrap(); + + assert_eq!(first.created, replayed.created); + assert_eq!(first.skipped, replayed.skipped); + assert_eq!(first.skipped.len(), 1); + assert_eq!(first.skipped[0].operation_key, "GET /missing"); +} + #[tokio::test] #[serial] async fn concurrent_openapi_import_replays_the_same_atomic_result() { @@ -231,6 +272,55 @@ async fn concurrent_openapi_import_replays_the_same_atomic_result() { assert!(conflicting_replay.is_err()); } +#[tokio::test] +#[serial] +async fn apply_fails_closed_when_the_preview_parser_contract_drifts() { + let registry = test_registry().await; + let service = test_service( + registry.clone(), + test_storage_root("openapi_import_parser_drift"), + test_auth_settings(), + test_secret_crypto(), + ); + let workspace_id = WorkspaceId::new("ws_default"); + let preview = service + .preview_openapi_import(&workspace_id, openapi_upload()) + .await + .unwrap(); + let job_id: crank_registry::ImportJobId = preview.job_id.as_str().into(); + sqlx::query( + "update import_jobs + set preview_payload = jsonb_set(preview_payload, '{preview_digest}', to_jsonb($1::text)) + where id = $2", + ) + .bind("0".repeat(64)) + .bind(job_id.as_str()) + .execute(registry.pool()) + .await + .unwrap(); + + let result = service + .create_openapi_import( + &workspace_id, + &job_id, + OpenApiImportCreatePayload { + selected_operation_keys: vec!["GET /v2/latest".to_owned()], + server_url: Some("https://api.frankfurter.dev".to_owned()), + conflict_mode: "rename".to_owned(), + }, + ) + .await; + + assert!(result.is_err()); + assert!( + service + .list_operations(&workspace_id) + .await + .unwrap() + .is_empty() + ); +} + fn openapi_upload() -> OpenApiUpload { OpenApiUpload { bytes: OPENAPI3.as_bytes().to_vec(), diff --git a/apps/admin-api/tests/integration/openapi_source.rs b/apps/admin-api/tests/integration/openapi_source.rs index dd95f2f..08fd211 100644 --- a/apps/admin-api/tests/integration/openapi_source.rs +++ b/apps/admin-api/tests/integration/openapi_source.rs @@ -385,6 +385,16 @@ async fn multipart_boundary_rejections_are_localized_and_leave_no_entities() { Some("no supported methods"), ) .await; + assert_eq!( + sqlx::query_scalar::<_, i64>( + "select count(*) from artifact_sources where lifecycle = 'active'", + ) + .fetch_one(registry.pool()) + .await + .unwrap(), + 0, + "no_methods must not leave an active source behind", + ); assert_rejected( &client, &server, @@ -432,6 +442,42 @@ async fn multipart_boundary_rejections_are_localized_and_leave_no_entities() { assert!(!rendered.contains("digest")); } +#[tokio::test(flavor = "multi_thread")] +#[serial] +async fn cleanup_is_bounded_and_malformed_expired_jobs_do_not_block_later_rows() { + let registry = test_registry().await; + for index in 0..129 { + sqlx::query( + "insert into import_jobs ( + id, workspace_id, kind, source_format, source_version, status, + preview_payload, created_operation_ids, error_text, created_at, expires_at, finished_at + ) values ( + $1, 'ws_default', 'openapi', 'openapi', null, 'pending', + '{\"source\": {\"source_id\": 1}}'::jsonb, '[]'::jsonb, null, + now() - interval '2 hours', now() - interval '1 hour', null + )", + ) + .bind(format!("imp_cleanup_{index:03}")) + .execute(registry.pool()) + .await + .unwrap(); + } + + let first = registry.cleanup_expired_import_jobs(128).await.unwrap(); + assert_eq!(first.deleted_jobs, 128); + assert!(first.more_work); + let second = registry.cleanup_expired_import_jobs(128).await.unwrap(); + assert_eq!(second.deleted_jobs, 1); + assert!(!second.more_work); + assert_eq!( + sqlx::query_scalar::<_, i64>("select count(*) from import_jobs") + .fetch_one(registry.pool()) + .await + .unwrap(), + 0, + ); +} + async fn wait_for_source_lifecycle( registry: &crank_registry::PostgresRegistry, lifecycle: ArtifactSourceLifecycle, @@ -746,11 +792,15 @@ async fn cancellation_detaches_and_restart_cleanup_recovers_a_dangling_source() preview_task.abort(); let cancellation = preview_task.await.unwrap_err(); assert!(cancellation.is_cancelled()); - wait_for_blocked_import_insert_to_stop(&observer).await; + // Dropping a sqlx query future does not synchronously cancel the backend + // statement. Release the artificial blocker first; the abandoned + // transaction can then finish and roll back before the detach fallback + // obtains another pooled connection. sqlx::query("select pg_advisory_unlock(2147483001)") .execute(&mut *lock_connection) .await .unwrap(); + wait_for_blocked_import_insert_to_stop(&observer).await; wait_for_specific_source_lifecycle( ®istry, &cancelled_source_id, diff --git a/apps/ui/js/api.js b/apps/ui/js/api.js index 5e1003b..e93bb28 100644 --- a/apps/ui/js/api.js +++ b/apps/ui/js/api.js @@ -428,7 +428,10 @@ createOpenApiImport: function(workspaceId, jobId, payload, options) { return request(API_BASE + '/workspaces/' + encodeURIComponent(workspaceId) + '/imports/openapi/' + encodeURIComponent(jobId) + '/create', { method: 'POST', - headers: headers({ 'Content-Type': 'application/json' }), + headers: headers({ + 'Content-Type': 'application/json', + 'Accept-Language': localStorage.getItem('crank_lang') === 'ru' ? 'ru' : 'en', + }), body: JSON.stringify(payload), signal: options && options.signal, }); diff --git a/apps/ui/js/i18n.js b/apps/ui/js/i18n.js index 40813b6..0aef40d 100644 --- a/apps/ui/js/i18n.js +++ b/apps/ui/js/i18n.js @@ -2082,7 +2082,7 @@ Object.assign(TRANSLATIONS.en, { 'openapi.server.label': 'Base URL', 'openapi.server.custom_aria': 'Custom Base URL', 'openapi.server.placeholder': 'Or provide a base URL, for example https://api.example.com', - 'openapi.server.later': 'Specify later', + 'openapi.server.required': 'Enter a Base URL', 'openapi.conflict.label': 'If an operation already exists', 'openapi.conflict.rename': 'Create a copy with a new name', 'openapi.conflict.skip': 'Skip it', @@ -2110,6 +2110,8 @@ Object.assign(TRANSLATIONS.en, { 'openapi.status.creating': 'Creating drafts…', 'openapi.status.created': 'Created: {created}; skipped: {skipped}.', 'openapi.status.cancelled': 'Request cancelled.', + 'openapi.status.apply_outcome_unknown': 'The create request may have completed. Retry uses the same import job and is safe.', + 'openapi.status.refresh_failed': 'Drafts were created, but the operation list could not be refreshed.', 'openapi.status.context_changed': 'The request was cancelled because the context changed.', 'openapi.error.file_required': 'Choose an OpenAPI/Swagger file first.', 'openapi.error.file_type': 'Choose one .yaml, .yml, or .json file with a matching type.', @@ -2196,7 +2198,7 @@ Object.assign(TRANSLATIONS.ru, { 'openapi.server.label': 'Base URL', 'openapi.server.custom_aria': 'Свой Base URL', 'openapi.server.placeholder': 'Или укажите свой base URL, например https://api.example.com', - 'openapi.server.later': 'Указать позже', + 'openapi.server.required': 'Укажите Base URL', 'openapi.conflict.label': 'Если операция уже существует', 'openapi.conflict.rename': 'Создать копию с новым именем', 'openapi.conflict.skip': 'Пропустить', @@ -2224,6 +2226,8 @@ Object.assign(TRANSLATIONS.ru, { 'openapi.status.creating': 'Создаю черновики…', 'openapi.status.created': 'Создано: {created}; пропущено: {skipped}.', 'openapi.status.cancelled': 'Запрос отменён.', + 'openapi.status.apply_outcome_unknown': 'Запрос на создание мог завершиться. Повтор использует ту же задачу импорта и безопасен.', + 'openapi.status.refresh_failed': 'Черновики созданы, но список операций не удалось обновить.', 'openapi.status.context_changed': 'Запрос отменён из-за изменения контекста.', 'openapi.error.file_required': 'Сначала выберите файл OpenAPI/Swagger.', 'openapi.error.file_type': 'Выберите один файл .yaml, .yml или .json с подходящим типом.', diff --git a/apps/ui/js/openapi-import.js b/apps/ui/js/openapi-import.js index 38b6748..a120d01 100644 --- a/apps/ui/js/openapi-import.js +++ b/apps/ui/js/openapi-import.js @@ -1,5 +1,7 @@ (function() { var MAX_FILE_BYTES = 256 * 1024; + var APPLY_REPLAY_STORAGE_KEY = 'crank_openapi_apply_replay_v1'; + var restoredApplyRequest = loadApplyReplay(); var state = { workspaceId: null, onImported: null, @@ -12,9 +14,58 @@ previewController: null, applyController: null, applyInFlight: false, + applyRequest: restoredApplyRequest, + applyMetadata: restoredApplyRequest && restoredApplyRequest.operationMetadata || [], + applyResult: null, + applyOutcomeUnknown: !!restoredApplyRequest, + applyRefreshFailed: false, restoreFocus: null, }; + function validApplyReplay(request) { + return request + && typeof request.workspaceId === 'string' && request.workspaceId.length > 0 && request.workspaceId.length <= 128 + && typeof request.jobId === 'string' && request.jobId.length > 0 && request.jobId.length <= 128 + && request.payload && Array.isArray(request.payload.selected_operation_keys) + && request.payload.selected_operation_keys.length <= 1024 + && request.payload.selected_operation_keys.every(function(key) { + return typeof key === 'string' && key.length > 0 && key.length <= 512; + }) + && (request.payload.server_url === null + || (typeof request.payload.server_url === 'string' && request.payload.server_url.length <= 2048)) + && (request.payload.conflict_mode === 'skip' || request.payload.conflict_mode === 'rename') + && Array.isArray(request.operationMetadata) + && request.operationMetadata.length <= 1024 + && request.operationMetadata.every(function(operation) { + return operation + && typeof operation.key === 'string' && operation.key.length > 0 && operation.key.length <= 512 + && typeof operation.method === 'string' && operation.method.length > 0 && operation.method.length <= 16 + && typeof operation.path === 'string' && operation.path.length > 0 && operation.path.length <= 2048; + }); + } + + function loadApplyReplay() { + try { + var request = JSON.parse(sessionStorage.getItem(APPLY_REPLAY_STORAGE_KEY) || 'null'); + return validApplyReplay(request) ? request : null; + } catch (_error) { + return null; + } + } + + function persistApplyReplay(request) { + if (!validApplyReplay(request)) return; + try { + sessionStorage.setItem(APPLY_REPLAY_STORAGE_KEY, JSON.stringify(request)); + } catch (_error) {} + } + + function clearApplyReplay() { + try { + sessionStorage.removeItem(APPLY_REPLAY_STORAGE_KEY); + } catch (_error) {} + } + function qs(id) { return document.getElementById(id); } @@ -31,10 +82,14 @@ return localStorage.getItem('crank_lang') === 'ru' ? 'ru' : 'en'; } - function isCurrent(revision, workspaceId, language) { + function isSameContext(revision, workspaceId, language) { return revision === state.revision && workspaceId === state.workspaceId - && language === locale() + && language === locale(); + } + + function isCurrent(revision, workspaceId, language) { + return isSameContext(revision, workspaceId, language) && !qs('openapi-import-modal').hidden; } @@ -279,7 +334,7 @@ servers.forEach(function(server) { var option = document.createElement('option'); option.value = server; - option.textContent = server || tKey('openapi.server.later'); + option.textContent = server || tKey('openapi.server.required'); serverSelect.appendChild(option); }); } @@ -425,6 +480,11 @@ function clearPreview() { state.jobId = null; state.preview = null; + state.applyRequest = null; + state.applyMetadata = []; + state.applyResult = null; + state.applyOutcomeUnknown = false; + state.applyRefreshFailed = false; state.filterQuery = ''; state.filterMethod = ''; qs('openapi-import-preview-panel').hidden = true; @@ -448,6 +508,8 @@ state.applyInFlight = false; qs('openapi-import-preview').disabled = false; qs('openapi-import-create').disabled = false; + qs('openapi-import-file-select').disabled = false; + qs('openapi-import-reset').disabled = false; qs('openapi-import-cancel').hidden = true; clearPreview(); @@ -462,8 +524,40 @@ qs('openapi-import-retry').hidden = !show; } + function lockApplyInputs(locked) { + qs('openapi-import-create').disabled = locked; + qs('openapi-import-preview').disabled = locked; + qs('openapi-import-file-select').disabled = locked; + qs('openapi-import-reset').disabled = locked; + } + + function markApplyOutcomeUnknown(options) { + if (!state.applyRequest) return false; + + abort(state.applyController); + state.applyController = null; + state.applyInFlight = false; + state.applyOutcomeUnknown = true; + persistApplyReplay(state.applyRequest); + // Until the same job is replayed, starting another import could hide an + // already committed server outcome and lead to duplicate drafts. + lockApplyInputs(true); + qs('openapi-import-cancel').hidden = true; + + if (!options || options.render !== false) { + showRetry(true); + setStatus(tKey('openapi.status.apply_outcome_unknown'), true, true); + } + + return true; + } + function cancelCurrent() { - if (!state.previewController && !state.applyController) return; + if (state.applyInFlight || state.applyController) { + markApplyOutcomeUnknown(); + return; + } + if (!state.previewController) return; invalidate(); showRetry(!!state.file); @@ -547,13 +641,32 @@ } } - function previewOperationByName(name) { + function previewOperationForCreated(created) { var groups = state.preview && state.preview.groups || []; for (var groupIndex = 0; groupIndex < groups.length; groupIndex += 1) { var operations = groups[groupIndex].operations || []; for (var operationIndex = 0; operationIndex < operations.length; operationIndex += 1) { - if (operations[operationIndex].suggested_name === name) return operations[operationIndex]; + if (created.operation_key && operations[operationIndex].key === created.operation_key) { + return operations[operationIndex]; + } + } + } + + // Older servers did not return operation_key. Keep result rendering useful + // during a rolling upgrade, but use the unambiguous key whenever it exists. + for (var fallbackGroupIndex = 0; fallbackGroupIndex < groups.length; fallbackGroupIndex += 1) { + var fallbackOperations = groups[fallbackGroupIndex].operations || []; + for (var fallbackOperationIndex = 0; fallbackOperationIndex < fallbackOperations.length; fallbackOperationIndex += 1) { + if (fallbackOperations[fallbackOperationIndex].suggested_name === created.name) { + return fallbackOperations[fallbackOperationIndex]; + } + } + } + + for (var metadataIndex = 0; metadataIndex < state.applyMetadata.length; metadataIndex += 1) { + if (created.operation_key && state.applyMetadata[metadataIndex].key === created.operation_key) { + return state.applyMetadata[metadataIndex]; } } @@ -615,7 +728,7 @@ + ''; response.created.forEach(function(operation) { - var previewOperation = previewOperationByName(operation.name); + var previewOperation = previewOperationForCreated(operation); var row = document.createElement('div'); var nameCell = document.createElement('div'); var name = document.createElement('div'); @@ -671,6 +784,118 @@ node.hidden = false; } + function createdStatus(response) { + return tfKey('openapi.status.created', { + created: (response.created || []).length, + skipped: (response.skipped || []).length, + }); + } + + async function notifyImported(revision, workspaceId, language) { + if (typeof state.onImported !== 'function') return; + + try { + await state.onImported(); + } catch (_error) { + state.applyRefreshFailed = true; + if (isCurrent(revision, workspaceId, language) && state.applyResult) { + setStatus(createdStatus(state.applyResult) + ' ' + tKey('openapi.status.refresh_failed'), false, true); + } + } + } + + async function submitApply(request) { + if (state.applyInFlight) return; + + var revision = state.revision; + var workspaceId = request.workspaceId; + var language = locale(); + var controller = new AbortController(); + + state.applyController = controller; + state.applyInFlight = true; + state.applyOutcomeUnknown = false; + state.applyRefreshFailed = false; + lockApplyInputs(true); + qs('openapi-import-cancel').hidden = false; + showRetry(false); + setStatus(tKey('openapi.status.creating')); + + try { + var response = await window.CrankApi.createOpenApiImport( + workspaceId, + request.jobId, + request.payload, + { signal: controller.signal } + ); + // Abort is advisory: a superseded request can still deliver a response. + // Only the controller installed by this attempt may mutate UI authority. + if (state.applyController !== controller) return; + state.applyOutcomeUnknown = false; + clearApplyReplay(); + state.applyController = null; + state.applyInFlight = false; + lockApplyInputs(false); + qs('openapi-import-cancel').hidden = true; + showRetry(false); + if (!isSameContext(revision, workspaceId, language)) { + // The original workspace job is now resolved. Do not render its result + // into another workspace/language context. + state.applyResult = null; + state.applyRequest = null; + return; + } + state.applyResult = response; + state.applyRequest = null; + if (isCurrent(revision, workspaceId, language)) { + setStatus(createdStatus(response)); + renderResult(response); + } + await notifyImported(revision, workspaceId, language); + } catch (error) { + // A cancelled attempt may settle after an idempotent retry has already + // installed a new controller. It must not overwrite the newer state. + if (state.applyController !== controller) return; + if (error && error.name === 'AbortError') { + state.applyOutcomeUnknown = true; + persistApplyReplay(state.applyRequest); + showRetry(true); + if (!qs('openapi-import-modal').hidden) { + setStatus(tKey('openapi.status.apply_outcome_unknown'), true, true); + } + return; + } + + if (Number.isInteger(error && error.status) + && error.status >= 400 && error.status < 500 && error.status !== 408) { + state.applyOutcomeUnknown = false; + state.applyRequest = null; + clearApplyReplay(); + showRetry(true); + if (!qs('openapi-import-modal').hidden) { + setStatus(errorStatus(error, 'openapi.error.create'), true, true); + } + return; + } + + // A browser/network failure is not proof that the server rolled the + // transaction back. Replaying this exact job is idempotent server-side. + state.applyOutcomeUnknown = true; + persistApplyReplay(state.applyRequest); + showRetry(true); + if (!qs('openapi-import-modal').hidden) { + setStatus(errorStatus(error, 'openapi.error.create') + ' ' + tKey('openapi.status.apply_outcome_unknown'), true, true); + } + } finally { + if (state.applyController === controller) { + state.applyController = null; + state.applyInFlight = false; + lockApplyInputs(state.applyOutcomeUnknown); + qs('openapi-import-cancel').hidden = true; + } + } + } + async function createDrafts() { if (state.applyInFlight) return; @@ -690,50 +915,32 @@ return; } - var revision = state.revision; - var workspaceId = state.workspaceId; - var language = locale(); - var jobId = state.jobId; - var controller = new AbortController(); - - state.applyController = controller; - state.applyInFlight = true; - qs('openapi-import-create').disabled = true; - qs('openapi-import-cancel').hidden = false; - showRetry(false); - setStatus(tKey('openapi.status.creating')); - - try { - var response = await window.CrankApi.createOpenApiImport(workspaceId, jobId, { + state.applyRequest = { + workspaceId: state.workspaceId, + jobId: state.jobId, + payload: { selected_operation_keys: keys, server_url: serverUrl, conflict_mode: qs('openapi-import-conflict-mode').value || 'rename', - }, { signal: controller.signal }); - if (!isCurrent(revision, workspaceId, language)) return; + }, + operationMetadata: (state.preview.groups || []).flatMap(function(group) { + return (group.operations || []).filter(function(operation) { + return keys.indexOf(operation.key) >= 0; + }).map(function(operation) { + return { key: operation.key, method: operation.method, path: operation.path }; + }); + }), + }; + state.applyMetadata = state.applyRequest.operationMetadata; + persistApplyReplay(state.applyRequest); + await submitApply(state.applyRequest); + } - setStatus(tfKey('openapi.status.created', { - created: (response.created || []).length, - skipped: (response.skipped || []).length, - })); - renderResult(response); - if (typeof state.onImported === 'function') await state.onImported(); - } catch (error) { - if (!isCurrent(revision, workspaceId, language)) return; - if (error && error.name === 'AbortError') { - setStatus(tKey('openapi.status.cancelled')); - return; - } - - showRetry(false); - setStatus(errorStatus(error, 'openapi.error.create'), true, true); - } finally { - if (isCurrent(revision, workspaceId, language)) { - state.applyController = null; - state.applyInFlight = false; - qs('openapi-import-create').disabled = false; - qs('openapi-import-cancel').hidden = true; - } + function retryCurrent() { + if (state.applyOutcomeUnknown && state.applyRequest) { + return submitApply(state.applyRequest); } + return preview(); } function reset() { @@ -745,7 +952,8 @@ } function open(options) { - if (state.workspaceId && state.workspaceId !== options.workspaceId) { + if (state.workspaceId && state.workspaceId !== options.workspaceId + && !(state.applyOutcomeUnknown && state.applyRequest)) { invalidate({ clearFile: true }); } @@ -754,10 +962,36 @@ state.restoreFocus = document.activeElement; qs('openapi-import-modal').hidden = false; renderFileName(); + if (state.applyResult) { + renderResult(state.applyResult); + setStatus(createdStatus(state.applyResult) + (state.applyRefreshFailed ? ' ' + tKey('openapi.status.refresh_failed') : '')); + } else if (state.applyOutcomeUnknown && state.applyRequest) { + lockApplyInputs(true); + showRetry(true); + setStatus(tKey('openapi.status.apply_outcome_unknown'), true, true); + qs('openapi-import-retry').focus(); + return; + } qs('openapi-import-file-select').focus(); } function close() { + if (state.applyInFlight || state.applyController) { + markApplyOutcomeUnknown({ render: false }); + qs('openapi-import-modal').hidden = true; + if (state.restoreFocus && typeof state.restoreFocus.focus === 'function') { + state.restoreFocus.focus(); + } + return; + } + if (state.applyOutcomeUnknown && state.applyRequest) { + persistApplyReplay(state.applyRequest); + qs('openapi-import-modal').hidden = true; + if (state.restoreFocus && typeof state.restoreFocus.focus === 'function') { + state.restoreFocus.focus(); + } + return; + } invalidate({ clearFile: true }); showRetry(false); qs('openapi-import-modal').hidden = true; @@ -796,7 +1030,7 @@ selectFile(event.target.files && event.target.files[0]); }); qs('openapi-import-preview').addEventListener('click', preview); - qs('openapi-import-retry').addEventListener('click', preview); + qs('openapi-import-retry').addEventListener('click', retryCurrent); qs('openapi-import-create').addEventListener('click', createDrafts); qs('openapi-import-cancel').addEventListener('click', cancelCurrent); qs('openapi-import-reset').addEventListener('click', reset); @@ -842,6 +1076,15 @@ }); function invalidateForContext(options) { if (!qs('openapi-import-modal').hidden) { + if (state.applyInFlight || state.applyController) { + markApplyOutcomeUnknown(); + return; + } + if (state.applyOutcomeUnknown && state.applyRequest) { + showRetry(true); + setStatus(tKey('openapi.status.apply_outcome_unknown'), true, true); + return; + } invalidate(options); renderFileName(); showRetry(!options || !options.clearFile ? !!state.file : false); @@ -861,14 +1104,30 @@ if (event.key === 'crank_lang') invalidateForContext(); }); window.addEventListener('pagehide', function() { - invalidate({ clearFile: true }); + if (state.applyInFlight || state.applyController) { + markApplyOutcomeUnknown({ render: false }); + state.file = null; + qs('openapi-import-file').value = ''; + renderFileName(); + } else if (state.applyOutcomeUnknown && state.applyRequest) { + persistApplyReplay(state.applyRequest); + } else { + invalidate({ clearFile: true }); + } + clearServerSelection(); showRetry(false); setStatus(''); }); window.addEventListener('pageshow', function(event) { if (!event.persisted || qs('openapi-import-modal').hidden) return; - invalidateForContext({ clearFile: true }); + if (state.applyOutcomeUnknown && state.applyRequest) { + lockApplyInputs(true); + showRetry(true); + setStatus(tKey('openapi.status.apply_outcome_unknown'), true, true); + } else { + invalidateForContext({ clearFile: true }); + } }); }); diff --git a/apps/ui/tests/e2e/operations.spec.js b/apps/ui/tests/e2e/operations.spec.js index ee5ef85..22c27e2 100644 --- a/apps/ui/tests/e2e/operations.spec.js +++ b/apps/ui/tests/e2e/operations.spec.js @@ -242,7 +242,106 @@ test('OpenAPI upload recovers from pagehide and a preview server error', async ( releasePreview(); }); -test('OpenAPI upload invalidates active draft creation after language or workspace changes', async ({ page }) => { +test('OpenAPI apply cancellation preserves the import job for an idempotent retry', async ({ page }) => { + await login(page); + await dismissOnboardingIfOpen(page); + await page.getByRole('button', { name: localized('Import OpenAPI', 'Импорт OpenAPI') }).click(); + + await page.route('**/imports/openapi/preview', async (route) => { + await route.fulfill({ + contentType: 'application/json', + body: JSON.stringify({ + job_id: 'job_retry_same_authority', + preview: { + source: { servers: ['https://api.example.test'] }, + findings: [], + groups: [{ + title: 'retry', + operations: [{ + key: 'get:/retry', method: 'GET', path: '/retry', suggested_name: 'preview_name', + suggested_display_name: 'Retry', input_fields: 0, output_fields: 0, + draft: { input_mapping: { rules: [] }, output_mapping: { rules: [] } }, findings: [], + }], + }], + }, + }), + }); + }); + + const requests = []; + let releaseFirst; + const firstRequest = new Promise((resolve) => { + releaseFirst = resolve; + }); + await page.route('**/imports/openapi/*/create', async (route) => { + requests.push({ url: route.request().url(), body: route.request().postData() }); + if (requests.length === 1) await firstRequest; + if (requests.length === 3) { + await route.fulfill({ + status: 409, + contentType: 'application/json', + body: JSON.stringify({ error: { code: 'import_job.application_mismatch', message: 'conflict' } }), + }); + return; + } + try { + await route.fulfill({ + contentType: 'application/json', + body: JSON.stringify({ + created: [{ + operation_key: 'get:/retry', name: 'renamed_after_conflict', operation_id: 'op_retry', version: 1, + }], + skipped: [], findings: [], + }), + }); + } catch (_error) { + // The first request is deliberately aborted by the browser. + } + }); + + await page.locator('#openapi-import-file').setInputFiles({ + name: 'retry.yaml', mimeType: 'application/yaml', buffer: Buffer.from('openapi: 3.0.3'), + }); + await page.locator('#openapi-import-preview').click(); + await expect(page.locator('#openapi-import-preview-panel')).toBeVisible(); + await page.locator('#openapi-import-create').click(); + await expect.poll(() => requests.length).toBe(1); + await page.locator('#openapi-import-cancel').click(); + await expect(page.locator('#openapi-import-status')).toContainText(localized('may have completed', 'мог завершиться')); + await expect(page.locator('#openapi-import-retry')).toBeVisible(); + await page.locator('[data-openapi-close]').last().click(); + await page.reload(); + await page.getByRole('button', { name: localized('Import OpenAPI', 'Импорт OpenAPI') }).click(); + await expect(page.locator('#openapi-import-retry')).toBeVisible(); + await page.locator('#openapi-import-retry').click(); + await expect(page.locator('#openapi-import-result')).toContainText('GET /retry'); + await expect.poll(() => requests.length).toBe(2); + // Let the cancelled first attempt answer after the retry. Its stale response + // must not overwrite or unlock the newer attempt's state. + releaseFirst(); + await page.waitForTimeout(50); + await expect(page.locator('#openapi-import-result')).toContainText('GET /retry'); + expect(requests[0].url).toContain('/imports/openapi/job_retry_same_authority/create'); + expect(requests[1].url).toBe(requests[0].url); + expect(requests[1].body).toBe(requests[0].body); + + // A received 4xx is a definitive non-commit outcome, unlike a transport + // failure. It must release the wizard instead of trapping it in replay mode. + await page.locator('#openapi-import-reset').click(); + await page.locator('#openapi-import-file').setInputFiles({ + name: 'definitive-4xx.yaml', mimeType: 'application/yaml', buffer: Buffer.from('openapi: 3.0.3'), + }); + await page.locator('#openapi-import-preview').click(); + await expect(page.locator('#openapi-import-preview-panel')).toBeVisible(); + await page.locator('#openapi-import-create').click(); + await expect.poll(() => requests.length).toBe(3); + await expect(page.locator('#openapi-import-file-select')).toBeEnabled(); + await expect(page.locator('#openapi-import-reset')).toBeEnabled(); + await expect(page.locator('#openapi-import-retry')).toBeVisible(); + expect(await page.evaluate(() => sessionStorage.getItem('crank_openapi_apply_replay_v1'))).toBeNull(); +}); + +test('OpenAPI apply preserves job authority after language or workspace changes', async ({ page }) => { await login(page); await dismissOnboardingIfOpen(page); await page.getByRole('button', { name: localized('Import OpenAPI', 'Импорт OpenAPI') }).click(); @@ -268,12 +367,17 @@ test('OpenAPI upload invalidates active draft creation after language or workspa }); }); const releases = []; + const applyRequests = []; await page.route('**/imports/openapi/*/create', async (route) => { + applyRequests.push({ url: route.request().url(), body: route.request().postData() }); await new Promise((resolve) => releases.push(resolve)); try { await route.fulfill({ contentType: 'application/json', - body: JSON.stringify({ created: [{ name: 'stale draft', operation_id: 'op_stale', version: 1 }], skipped: [], findings: [] }), + body: JSON.stringify({ + created: [{ operation_key: 'get:/active', name: 'active', operation_id: 'op_active', version: 1 }], + skipped: [], findings: [], + }), }); } catch (_error) { // The invalidation aborts a request that must not update the modal afterwards. @@ -295,23 +399,37 @@ test('OpenAPI upload invalidates active draft creation after language or workspa } await beginApply('language-change.yaml'); - await page.evaluate(() => { - window.dispatchEvent(new StorageEvent('storage', { key: 'crank_lang', newValue: 'en' })); - }); + // Exercise the actual UI language switcher path: it updates storage, + // re-renders translations and dispatches crank:langchange in one action. + await page.evaluate(() => window.setLang('en')); + await expect(page.locator('#openapi-import-status')).toContainText('may have completed'); + await expect(page.locator('#openapi-import-retry')).toBeVisible(); + await expect(page.locator('#openapi-import-file-select')).toBeDisabled(); releases.shift()(); - await expect(page.locator('#openapi-import-preview-panel')).toBeHidden(); - await expect(page.locator('#openapi-import-result')).toBeHidden(); - await expect(page.locator('#openapi-import-create')).toBeEnabled(); + await page.locator('#openapi-import-retry').click(); + await expect.poll(() => releases.length).toBeGreaterThan(0); + releases.shift()(); + await expect(page.locator('#openapi-import-result')).toContainText('GET /active'); + expect(applyRequests[1].url).toBe(applyRequests[0].url); + expect(applyRequests[1].body).toBe(applyRequests[0].body); + await page.locator('#openapi-import-reset').click(); await beginApply('workspace-change.yaml'); + const workspaceRequest = applyRequests.at(-1); await page.evaluate(() => { window.dispatchEvent(new CustomEvent('crank:workspacechange', { detail: { id: 'workspace-after-apply' } })); }); + await expect(page.locator('#openapi-import-status')).toContainText('may have completed'); + await expect(page.locator('#openapi-import-retry')).toBeVisible(); + await expect(page.locator('#openapi-import-file-select')).toBeDisabled(); releases.shift()(); - await expect(page.locator('#openapi-import-preview-panel')).toBeHidden(); - await expect(page.locator('#openapi-import-result')).toBeHidden(); - await expect(page.locator('#openapi-import-file-name')).toContainText('No file selected'); - await expect(page.locator('#openapi-import-create')).toBeEnabled(); + await page.locator('#openapi-import-retry').click(); + await expect.poll(() => releases.length).toBeGreaterThan(0); + releases.shift()(); + await expect(page.locator('#openapi-import-file-select')).toBeEnabled(); + const workspaceReplay = applyRequests.at(-1); + expect(workspaceReplay.url).toBe(workspaceRequest.url); + expect(workspaceReplay.body).toBe(workspaceRequest.body); }); test('OpenAPI upload only renders the latest selected file and clears reset or close races', async ({ page }) => { diff --git a/crates/crank-artifacts/src/housekeeping.rs b/crates/crank-artifacts/src/housekeeping.rs index 3bdbdbb..c4707b0 100644 --- a/crates/crank-artifacts/src/housekeeping.rs +++ b/crates/crank-artifacts/src/housekeeping.rs @@ -4,7 +4,7 @@ use std::{ time::{Duration, SystemTime, UNIX_EPOCH}, }; -use crate::temp_scan::{list_names, valid_temp_name}; +use crate::temp_scan::{list_names_after, valid_temp_name}; use crate::{ ArtifactError, ArtifactStore, ReconciliationMutation, ReconciliationNamespace, ReconciliationScan, ReconciliationScanStop, @@ -94,10 +94,33 @@ pub struct StaleTemp { grace: Duration, } +/// Opaque, inode-bound progress marker for the bounded stale-temp sweeper. +/// +/// It deliberately advances by shard instead of retaining a directory offset: +/// directory offsets are invalidated by a concurrent writer, while round-robin +/// shard progress prevents a busy low-numbered shard from starving all others. +#[derive(Clone, Debug)] +pub struct TempScanCursor { + root_dev: u64, + root_ino: u64, + next_shard: u8, + shard_continuation: Option, +} + +#[derive(Clone, Debug)] +struct TempShardContinuation { + shard: u8, + dev: u64, + ino: u64, + cookie: i64, +} + #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] pub struct TempScan { pub scanned: usize, pub omitted: usize, + /// True when this page completed a full round-robin traversal. + pub complete: bool, } impl ArtifactStore { @@ -324,27 +347,93 @@ impl ArtifactStore { scan_budget: usize, result_limit: usize, ) -> Result<(TempScan, Vec), ArtifactError> { + self.scan_stale_temps_after(grace, None, scan_budget, result_limit) + .map(|(report, candidates, _)| (report, candidates)) + } + + /// Resumes stale-temp cleanup from an opaque round-robin marker. A + /// continuation is valid only for this exact pinned root. + pub fn scan_stale_temps_after( + &self, + grace: Duration, + continuation: Option, + scan_budget: usize, + result_limit: usize, + ) -> Result<(TempScan, Vec, TempScanCursor), ArtifactError> { let root = self.root()?; let _lock = RootLock::shared(root)?; ensure_root_unchanged(root)?; let sha = match open_existing_dir(root.fd.as_raw_fd(), b"sha256") { Ok(fd) => fd, - Err(ArtifactError::NotFound) => return Ok((TempScan::default(), Vec::new())), + Err(ArtifactError::NotFound) => { + return Ok(( + TempScan { + complete: true, + ..TempScan::default() + }, + Vec::new(), + TempScanCursor { + root_dev: root.dev, + root_ino: root.ino, + next_shard: 0, + shard_continuation: None, + }, + )); + } Err(error) => return Err(error), }; + let (start_shard, mut shard_continuation) = match continuation { + Some(cursor) if cursor.root_dev == root.dev && cursor.root_ino == root.ino => { + (cursor.next_shard, cursor.shard_continuation) + } + Some(_) => return Err(ArtifactError::UnsafeRoot), + None => (0, None), + }; let mut report = TempScan::default(); let mut remaining = scan_budget; let mut result = Vec::new(); - for shard in (0_u8..=255).map(|value| format!("{value:02x}")) { + if remaining == 0 { + return Ok(( + report, + result, + TempScanCursor { + root_dev: root.dev, + root_ino: root.ino, + next_shard: start_shard, + shard_continuation, + }, + )); + } + let mut next_shard = start_shard; + for offset in 0_u16..=255 { if remaining == 0 { break; } + let shard_number = start_shard.wrapping_add(offset as u8); + next_shard = shard_number.wrapping_add(1); + let shard = format!("{shard_number:02x}"); let shard_fd = match open_existing_dir(sha.as_raw_fd(), shard.as_bytes()) { Ok(fd) => fd, - Err(ArtifactError::NotFound) => continue, + Err(ArtifactError::NotFound) => { + shard_continuation = None; + continue; + } Err(error) => return Err(error), }; - for name in list_names(shard_fd.as_raw_fd(), &mut remaining, &mut report)? { + let stat = stat_fd(shard_fd.as_raw_fd())?; + let resume = shard_continuation.take().and_then(|continuation| { + if continuation.shard == shard_number + && continuation.dev == stat.st_dev + && continuation.ino == stat.st_ino + { + Some(continuation.cookie) + } else { + None + } + }); + let page = list_names_after(shard_fd.as_raw_fd(), &mut remaining, &mut report, resume)?; + for entry in page.names { + let name = entry.name; if !valid_temp_name(&name) { continue; } @@ -370,6 +459,23 @@ impl ArtifactStore { { if result.len() == result_limit { report.omitted += 1; + if result_limit > 0 { + return Ok(( + report, + result, + TempScanCursor { + root_dev: root.dev, + root_ino: root.ino, + next_shard: shard_number, + shard_continuation: Some(TempShardContinuation { + shard: shard_number, + dev: stat.st_dev, + ino: stat.st_ino, + cookie: entry.cookie_before, + }), + }, + )); + } continue; } result.push(StaleTemp { @@ -385,8 +491,35 @@ impl ArtifactStore { }); } } + if let Some(cookie) = page.continuation { + return Ok(( + report, + result, + TempScanCursor { + root_dev: root.dev, + root_ino: root.ino, + next_shard: shard_number, + shard_continuation: Some(TempShardContinuation { + shard: shard_number, + dev: stat.st_dev, + ino: stat.st_ino, + cookie, + }), + }, + )); + } } - Ok((report, result)) + report.complete = true; + Ok(( + report, + result, + TempScanCursor { + root_dev: root.dev, + root_ino: root.ino, + next_shard, + shard_continuation: None, + }, + )) } /// Deletes a revalidated stale inode under an exclusive lock, then fsyncs. diff --git a/crates/crank-artifacts/src/lib.rs b/crates/crank-artifacts/src/lib.rs index 33a6d2f..62acd41 100644 --- a/crates/crank-artifacts/src/lib.rs +++ b/crates/crank-artifacts/src/lib.rs @@ -20,7 +20,9 @@ mod temp_scan; pub mod test_support; pub use error::ArtifactError; -pub use housekeeping::{ReconciliationCandidate, ReconciliationCursor, StaleTemp, TempScan}; +pub use housekeeping::{ + ReconciliationCandidate, ReconciliationCursor, StaleTemp, TempScan, TempScanCursor, +}; pub use model::{ ArtifactRef, MAX_ARTIFACT_BYTES, MAX_SOURCE_BYTES, ReconciliationMutation, ReconciliationNamespace, ReconciliationPresence, ReconciliationRegistration, diff --git a/crates/crank-artifacts/src/store.rs b/crates/crank-artifacts/src/store.rs index 45bfd57..03efe41 100644 --- a/crates/crank-artifacts/src/store.rs +++ b/crates/crank-artifacts/src/store.rs @@ -80,6 +80,15 @@ impl ArtifactStore { }) } + /// Verifies that the pinned root is still the same private directory that + /// was accepted at startup. This does not open data by pathname and is + /// suitable for a process readiness probe. + pub fn check_health(&self) -> Result<(), ArtifactError> { + let root = self.root()?; + let _lock = RootLock::shared(root)?; + ensure_root_unchanged(root) + } + /// Compatibility constructor. I/O returns the opening error fail-closed. pub fn new(root: impl AsRef) -> Self { Self { diff --git a/crates/crank-artifacts/src/temp_scan.rs b/crates/crank-artifacts/src/temp_scan.rs index 21700dd..2e101d0 100644 --- a/crates/crank-artifacts/src/temp_scan.rs +++ b/crates/crank-artifacts/src/temp_scan.rs @@ -1,5 +1,19 @@ use crate::{ArtifactError, ArtifactStore, TempScan}; +pub(crate) struct NamePage { + pub(crate) names: Vec, + /// Opaque `telldir` position immediately after the last consumed entry. + /// It is meaningful only for the same opened directory inode. + pub(crate) continuation: Option, +} + +pub(crate) struct NameEntry { + pub(crate) name: String, + /// Directory position immediately before this entry. Resuming here + /// guarantees a candidate omitted by a result limit is reconsidered. + pub(crate) cookie_before: i64, +} + pub(crate) fn valid_temp_name(name: &str) -> bool { let Some(rest) = name.strip_prefix(ArtifactStore::temp_prefix()) else { return false; @@ -20,11 +34,12 @@ pub(crate) fn valid_temp_name(name: &str) -> bool { && sequence.bytes().all(|byte| byte.is_ascii_digit()) } -pub(crate) fn list_names( +pub(crate) fn list_names_after( parent: i32, remaining: &mut usize, report: &mut TempScan, -) -> Result, ArtifactError> { + continuation: Option, +) -> Result { let dot = std::ffi::CString::new(".").map_err(|_| ArtifactError::Storage)?; let raw = unsafe { libc::openat( @@ -44,13 +59,32 @@ pub(crate) fn list_names( return Err(ArtifactError::Storage); } let mut names = Vec::new(); + if let Some(cookie) = continuation { + unsafe { + libc::seekdir(directory, cookie as libc::c_long); + } + } loop { if *remaining == 0 { - break; + let continuation = unsafe { libc::telldir(directory) }; + unsafe { + libc::closedir(directory); + } + return Ok(NamePage { + names, + continuation: (continuation >= 0).then_some(continuation as i64), + }); } unsafe { *libc::__errno_location() = 0; } + let cookie_before = unsafe { libc::telldir(directory) }; + if cookie_before < 0 { + unsafe { + libc::closedir(directory); + } + return Err(ArtifactError::Storage); + } let entry = unsafe { libc::readdir(directory) }; if entry.is_null() { if unsafe { *libc::__errno_location() } != 0 { @@ -59,7 +93,13 @@ pub(crate) fn list_names( } return Err(ArtifactError::Storage); } - break; + unsafe { + libc::closedir(directory); + } + return Ok(NamePage { + names, + continuation: None, + }); } let name = unsafe { std::ffi::CStr::from_ptr((*entry).d_name.as_ptr()) }.to_bytes(); if name == b"." || name == b".." { @@ -68,11 +108,10 @@ pub(crate) fn list_names( report.scanned += 1; *remaining -= 1; if let Ok(name) = std::str::from_utf8(name) { - names.push(name.to_owned()); + names.push(NameEntry { + name: name.to_owned(), + cookie_before: cookie_before as i64, + }); } } - unsafe { - libc::closedir(directory); - } - Ok(names) } diff --git a/crates/crank-artifacts/tests/artifact_store.rs b/crates/crank-artifacts/tests/artifact_store.rs index eac15e4..450ca42 100644 --- a/crates/crank-artifacts/tests/artifact_store.rs +++ b/crates/crank-artifacts/tests/artifact_store.rs @@ -360,6 +360,119 @@ fn housekeeping_enforces_result_limit_after_stale_validation() { assert!(temp_paths.iter().all(|path| path.exists())); } +#[test] +fn stale_temp_cleanup_round_robins_past_a_busy_early_shard() { + let root = TestRoot::new("cleanup-round-robin"); + let store = ArtifactStore::open(&root.0).unwrap(); + let sha = root.0.join("sha256"); + let early = sha.join("00"); + let late = sha.join("ff"); + fs::create_dir(&sha).unwrap(); + fs::create_dir(&early).unwrap(); + fs::create_dir(&late).unwrap(); + for directory in [&sha, &early, &late] { + fs::set_permissions(directory, fs::Permissions::from_mode(0o700)).unwrap(); + } + for index in 0..5 { + let entry = early.join(format!("unrelated-{index}")); + fs::write(&entry, b"not a temp").unwrap(); + fs::set_permissions(entry, fs::Permissions::from_mode(0o400)).unwrap(); + } + let stale = late.join(".crank-artifact-tmp-v1-00000000000000000000000000000000-1-1"); + fs::write(&stale, b"partial").unwrap(); + fs::set_permissions(&stale, fs::Permissions::from_mode(0o400)).unwrap(); + + let (_, first, cursor) = store + .scan_stale_temps_after(Duration::ZERO, None, 5, 1) + .unwrap(); + assert!(first.is_empty()); + + let (_, second, _) = store + .scan_stale_temps_after(Duration::ZERO, Some(cursor), 5, 1) + .unwrap(); + assert_eq!(second.len(), 1); + store + .delete_stale_temp(second.into_iter().next().unwrap()) + .unwrap(); + assert!(!stale.exists()); +} + +#[test] +fn stale_temp_cleanup_resumes_inside_a_busy_shard() { + let root = TestRoot::new("cleanup-intra-shard"); + let store = ArtifactStore::open(&root.0).unwrap(); + let sha = root.0.join("sha256"); + let shard = sha.join("00"); + fs::create_dir(&sha).unwrap(); + fs::create_dir(&shard).unwrap(); + for directory in [&sha, &shard] { + fs::set_permissions(directory, fs::Permissions::from_mode(0o700)).unwrap(); + } + for index in 0..12 { + let entry = shard.join(format!("unrelated-{index}")); + fs::write(&entry, b"not a temp").unwrap(); + fs::set_permissions(entry, fs::Permissions::from_mode(0o400)).unwrap(); + } + let stale = shard.join(".crank-artifact-tmp-v1-00000000000000000000000000000000-1-2"); + fs::write(&stale, b"partial").unwrap(); + fs::set_permissions(&stale, fs::Permissions::from_mode(0o400)).unwrap(); + + let mut cursor = None; + let mut discovered = None; + for _ in 0..4 { + let (_, candidates, next_cursor) = store + .scan_stale_temps_after(Duration::ZERO, cursor, 5, 1) + .unwrap(); + if let Some(candidate) = candidates.into_iter().next() { + discovered = Some(candidate); + break; + } + cursor = Some(next_cursor); + } + let candidate = discovered.expect("the temp behind one scan budget is eventually discovered"); + store.delete_stale_temp(candidate).unwrap(); + assert!(!stale.exists()); +} + +#[test] +fn stale_temp_cleanup_reconsiders_candidates_beyond_each_result_page() { + let root = TestRoot::new("cleanup-result-continuation"); + let store = ArtifactStore::open(&root.0).unwrap(); + let sha = root.0.join("sha256"); + let shard = sha.join("00"); + fs::create_dir(&sha).unwrap(); + fs::create_dir(&shard).unwrap(); + for directory in [&sha, &shard] { + fs::set_permissions(directory, fs::Permissions::from_mode(0o700)).unwrap(); + } + for index in 0..20 { + let temp = shard.join(format!( + ".crank-artifact-tmp-v1-00000000000000000000000000000000-1-{index}" + )); + fs::write(&temp, b"partial").unwrap(); + fs::set_permissions(temp, fs::Permissions::from_mode(0o400)).unwrap(); + } + + let mut cursor = None; + let mut deleted = 0; + for _ in 0..16 { + let (scan, candidates, next_cursor) = store + .scan_stale_temps_after(Duration::ZERO, cursor, 512, 3) + .unwrap(); + for candidate in candidates { + store.delete_stale_temp(candidate).unwrap(); + deleted += 1; + } + cursor = Some(next_cursor); + if scan.complete { + break; + } + } + + assert_eq!(deleted, 20, "every stale candidate must remain reachable"); + assert!(fs::read_dir(shard).unwrap().next().is_none()); +} + #[test] fn housekeeping_revalidates_subsecond_mtime() { let root = TestRoot::new("cleanup-subsecond"); @@ -479,6 +592,16 @@ fn open_rejects_non_private_root_and_existing_finals_must_be_immutable() { ); } +#[test] +fn health_check_revalidates_the_pinned_root_permissions() { + let root = TestRoot::new("health-check"); + let store = ArtifactStore::open(&root.0).unwrap(); + assert_eq!(store.check_health(), Ok(())); + + fs::set_permissions(&root.0, fs::Permissions::from_mode(0o755)).unwrap(); + assert_eq!(store.check_health(), Err(ArtifactError::UnsafeRoot)); +} + #[test] fn open_rejects_symlinked_root() { let root = TestRoot::new("root-symlink-target"); diff --git a/crates/crank-artifacts/tests/reconciliation.rs b/crates/crank-artifacts/tests/reconciliation.rs index 329a821..7e15988 100644 --- a/crates/crank-artifacts/tests/reconciliation.rs +++ b/crates/crank-artifacts/tests/reconciliation.rs @@ -3,6 +3,7 @@ use std::{ os::unix::fs::{MetadataExt, PermissionsExt}, path::PathBuf, sync::atomic::{AtomicU64, Ordering}, + sync::{Mutex, OnceLock}, time::{Duration, SystemTime}, }; @@ -19,8 +20,6 @@ use crank_artifacts::test_support::{ FaultAction, checkpoint_hits, clear_checkpoint, reset_traversal_calls, set_checkpoint, set_checkpoint_on_hit, traversal_calls, wait_until_held, }; -#[cfg(debug_assertions)] -use std::sync::{Mutex, OnceLock}; static NEXT_ROOT: AtomicU64 = AtomicU64::new(0); const FULL_SCAN_BUDGET: usize = 2048; @@ -47,7 +46,6 @@ impl Drop for TestRoot { } } -#[cfg(debug_assertions)] fn fault_guard() -> std::sync::MutexGuard<'static, ()> { static GUARD: OnceLock> = OnceLock::new(); GUARD.get_or_init(|| Mutex::new(())).lock().unwrap() @@ -55,6 +53,7 @@ fn fault_guard() -> std::sync::MutexGuard<'static, ()> { #[test] fn paginates_final_entries_without_disclosing_locations() { + let _guard = fault_guard(); let root = TestRoot::new("pages"); let store = ArtifactStore::open(&root.0).unwrap(); store.put(b"reconciliation first").unwrap(); @@ -116,6 +115,7 @@ fn paginates_final_entries_without_disclosing_locations() { #[test] fn classifies_malformed_and_unsafe_entries_with_a_usable_page() { + let _guard = fault_guard(); let root = TestRoot::new("malformed"); let store = ArtifactStore::open(&root.0).unwrap(); let stored = store.put(b"reconciliation valid").unwrap(); @@ -157,6 +157,7 @@ fn classifies_malformed_and_unsafe_entries_with_a_usable_page() { #[test] fn skips_a_valid_publish_temp_without_classifying_it_as_malformed() { + let _guard = fault_guard(); let root = TestRoot::new("valid-temp"); let store = ArtifactStore::open(&root.0).unwrap(); let stored = store.put(b"reconciliation with concurrent temp").unwrap(); @@ -181,6 +182,7 @@ fn skips_a_valid_publish_temp_without_classifying_it_as_malformed() { #[test] fn unsafe_canonical_shard_does_not_starve_later_shards() { + let _guard = fault_guard(); let root = TestRoot::new("unsafe-shard"); let store = ArtifactStore::open(&root.0).unwrap(); let sha = root.0.join("sha256"); @@ -212,6 +214,7 @@ fn unsafe_canonical_shard_does_not_starve_later_shards() { #[test] #[cfg(debug_assertions)] fn traversal_never_exceeds_the_exact_syscall_budget() { + let _guard = fault_guard(); let root = TestRoot::new("syscall-budget"); let store = ArtifactStore::open(&root.0).unwrap(); store.put(b"bounded traversal syscall accounting").unwrap(); @@ -237,6 +240,7 @@ fn traversal_never_exceeds_the_exact_syscall_budget() { #[test] fn rejects_a_cursor_from_another_store() { + let _guard = fault_guard(); let first_root = TestRoot::new("foreign-cursor-first"); let second_root = TestRoot::new("foreign-cursor-second"); let first = ArtifactStore::open(&first_root.0).unwrap(); @@ -276,6 +280,7 @@ fn disappearing_entry_after_readdir_is_skipped() { #[test] fn quarantine_and_delete_are_idempotent() { + let _guard = fault_guard(); let root = TestRoot::new("mutation"); let store = ArtifactStore::open(&root.0).unwrap(); let stored = store.put(b"reconciliation mutation").unwrap(); @@ -320,6 +325,7 @@ fn quarantine_and_delete_are_idempotent() { #[test] fn quarantine_preserves_old_inode_when_the_digest_is_republished() { + let _guard = fault_guard(); let root = TestRoot::new("no-clobber"); let store = ArtifactStore::open(&root.0).unwrap(); let stored = store.put(b"reconciliation no-clobber").unwrap(); @@ -358,6 +364,7 @@ fn quarantine_preserves_old_inode_when_the_digest_is_republished() { #[test] fn quarantine_refuses_to_replace_an_unrelated_inode() { + let _guard = fault_guard(); let root = TestRoot::new("unrelated-collision"); let store = ArtifactStore::open(&root.0).unwrap(); let stored = store.put(b"canonical final survives collision").unwrap(); @@ -394,6 +401,7 @@ fn quarantine_refuses_to_replace_an_unrelated_inode() { #[test] fn delete_fails_closed_for_a_replaced_or_hardlinked_quarantined_inode() { + let _guard = fault_guard(); let root = TestRoot::new("quarantine-replaced"); let store = ArtifactStore::open(&root.0).unwrap(); let stored = store @@ -431,6 +439,7 @@ fn delete_fails_closed_for_a_replaced_or_hardlinked_quarantined_inode() { #[test] fn delete_revalidates_a_post_scan_hardlink_without_inode_replacement() { + let _guard = fault_guard(); let root = TestRoot::new("quarantine-hardlink"); let store = ArtifactStore::open(&root.0).unwrap(); let stored = store.put(b"same quarantined inode gains a link").unwrap(); @@ -496,6 +505,7 @@ fn delete_fsync_ambiguity_is_retryable_and_recovers_as_absent() { #[test] fn rejects_a_replaced_final_inode() { + let _guard = fault_guard(); let root = TestRoot::new("replaced"); let store = ArtifactStore::open(&root.0).unwrap(); let stored = store.put(b"reconciliation original").unwrap(); @@ -773,6 +783,7 @@ fn crash_windows_are_recoverable() { #[test] fn registration_revalidates_content_and_enforces_grace_without_candidate_identity_access() { + let _guard = fault_guard(); let root = TestRoot::new("registration"); let store = ArtifactStore::open(&root.0).unwrap(); let stored = store.put(b"reconciliation registration").unwrap(); @@ -820,6 +831,7 @@ fn registration_revalidates_content_and_enforces_grace_without_candidate_identit #[test] fn namespace_scan_can_finish_quarantine_recovery_before_final_sweep() { + let _guard = fault_guard(); let root = TestRoot::new("namespace-order"); let store = ArtifactStore::open(&root.0).unwrap(); store.put(b"quarantine-first recovery").unwrap(); @@ -856,6 +868,7 @@ fn namespace_scan_can_finish_quarantine_recovery_before_final_sweep() { #[test] fn namespace_scan_rejects_a_cursor_from_a_broader_scan() { + let _guard = fault_guard(); let root = TestRoot::new("namespace-cursor-scope"); let store = ArtifactStore::open(&root.0).unwrap(); let (report, _) = store.scan_reconciliation(None, 0, 1).unwrap(); @@ -876,6 +889,7 @@ fn namespace_scan_rejects_a_cursor_from_a_broader_scan() { #[test] fn full_scan_rejects_a_quarantine_only_cursor() { + let _guard = fault_guard(); let root = TestRoot::new("namespace-cursor-narrow"); let store = ArtifactStore::open(&root.0).unwrap(); let (report, _) = store @@ -893,6 +907,7 @@ fn full_scan_rejects_a_quarantine_only_cursor() { #[test] fn presence_probe_distinguishes_final_quarantine_and_absent_without_disclosure() { + let _guard = fault_guard(); let root = TestRoot::new("presence"); let store = ArtifactStore::open(&root.0).unwrap(); let stored = store.put(b"presence probe").unwrap(); diff --git a/crates/crank-registry/src/lib.rs b/crates/crank-registry/src/lib.rs index 3b3c016..2803855 100644 --- a/crates/crank-registry/src/lib.rs +++ b/crates/crank-registry/src/lib.rs @@ -20,20 +20,20 @@ pub mod records { ArtifactReconciliationClaim, ArtifactSourceCursor, ArtifactSourceId, ArtifactSourceLifecycle, ArtifactSourcePage, ArtifactSourceRecord, ArtifactSourceSensitivity, AuthUserRecord, DescriptorKind, DescriptorMetadata, ImportJob, - ImportJobApplyResult, ImportJobId, ImportJobKind, ImportJobSourceEnvelope, 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, + ImportJobApplyResult, ImportJobCleanupReport, ImportJobId, ImportJobKind, + ImportJobSourceEnvelope, 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, }; } @@ -83,18 +83,18 @@ pub use model::{ CreateVersionRequest, CreateWorkspaceRequest, CreateYamlImportJobRequest, DecideApprovalRequest, DescriptorKind, DescriptorMetadata, DetachArtifactSourceRequest, ExpireApprovalRequest, FinishApprovalRequest, FinishImportJobRequest, ImportConflictMode, - ImportJob, ImportJobApplyResult, ImportJobId, ImportJobKind, ImportJobSourceEnvelope, - ImportJobStatus, ImportOperationDraft, InvitationRecord, InvocationHistoryLoss, - InvocationHistoryLossCategory, InvocationHistoryWriteOutcome, InvocationLogRecord, - InvocationRetentionOutcome, InvocationRetentionPolicy, InvocationRetentionStatus, - ListApprovalRequestsQuery, ListArtifactSourcesQuery, ListInvocationLogsQuery, - ListProductEventsQuery, 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, + ImportJob, ImportJobApplyResult, ImportJobCleanupReport, ImportJobId, ImportJobKind, + ImportJobSourceEnvelope, ImportJobStatus, ImportOperationDraft, InvitationRecord, + InvocationHistoryLoss, InvocationHistoryLossCategory, InvocationHistoryWriteOutcome, + InvocationLogRecord, InvocationRetentionOutcome, InvocationRetentionPolicy, + InvocationRetentionStatus, ListApprovalRequestsQuery, ListArtifactSourcesQuery, + ListInvocationLogsQuery, ListProductEventsQuery, 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/migrations.rs b/crates/crank-registry/src/migrations.rs index 92619cc..b622829 100644 --- a/crates/crank-registry/src/migrations.rs +++ b/crates/crank-registry/src/migrations.rs @@ -1,6 +1,7 @@ mod admin_auth_lifecycle_v8; mod agent_catalog_lifecycle_v9; mod approval_side_effects_v10; +mod artifact_cleanup_indexes_v13; mod artifact_metadata_v12; mod authority; mod baseline_v1; @@ -13,6 +14,7 @@ mod schema_guard; mod schema_guard_v10; mod schema_guard_v11; mod schema_guard_v12; +mod schema_guard_v13; mod schema_guard_v7; mod schema_guard_v8; mod schema_guard_v9; diff --git a/crates/crank-registry/src/migrations/artifact_cleanup_indexes_v13.rs b/crates/crank-registry/src/migrations/artifact_cleanup_indexes_v13.rs new file mode 100644 index 0000000..a4ef6e7 --- /dev/null +++ b/crates/crank-registry/src/migrations/artifact_cleanup_indexes_v13.rs @@ -0,0 +1,44 @@ +use sqlx::{Postgres, Transaction, query}; + +use super::authority::{MigrationDescriptor, MigrationError}; + +pub(super) const SOURCE: &str = include_str!("artifact_cleanup_indexes_v13.sql"); +pub(super) const SOURCE_SHA256: &str = + "abe822c967e1b2cc3966ede988e05eb2d9b2a062b539c45aaa8cad677395ab05"; + +pub(super) async fn apply( + transaction: &mut Transaction<'_, Postgres>, + descriptor: &MigrationDescriptor, +) -> Result<(), MigrationError> { + sqlx::raw_sql(SOURCE) + .execute(&mut **transaction) + .await + .map_err(|_| { + MigrationError::new( + "apply_failed", + "apply.artifact_cleanup_indexes", + Some(13), + "restore_known_good_backup", + ) + })?; + query( + "insert into __crank_migrations (version, name, checksum, phase, compatibility) + values ($1, $2, $3, $4, $5)", + ) + .bind(descriptor.version) + .bind(descriptor.name) + .bind(&descriptor.checksum) + .bind(descriptor.phase) + .bind(descriptor.compatibility) + .execute(&mut **transaction) + .await + .map_err(|_| { + MigrationError::new( + "apply_failed", + "apply.canonical_ledger", + Some(13), + "restore_known_good_backup", + ) + })?; + Ok(()) +} diff --git a/crates/crank-registry/src/migrations/artifact_cleanup_indexes_v13.sql b/crates/crank-registry/src/migrations/artifact_cleanup_indexes_v13.sql new file mode 100644 index 0000000..3afb879 --- /dev/null +++ b/crates/crank-registry/src/migrations/artifact_cleanup_indexes_v13.sql @@ -0,0 +1,19 @@ +-- Expand-only performance indexes for bounded artifact and import cleanup. +-- No data rewrite or destructive schema operation is required. +create index artifact_blobs_expired_claim_idx + on artifact_blobs(claim_expires_at, digest) + where claim_token is not null; + +create index artifact_sources_blob_lifecycle_detached_idx + on artifact_sources(blob_digest, lifecycle, detached_at); + +create index artifact_sources_openapi_dangling_idx + on artifact_sources(created_at, workspace_id, source_id) + where lifecycle = 'active' and left(source_id, 12) = 'src_openapi_'; + +create index import_jobs_expires_at_idx + on import_jobs(expires_at, id); + +create index import_jobs_openapi_source_idx + on import_jobs(workspace_id, ((preview_payload -> 'source' ->> 'source_id'))) + where preview_payload ? 'source'; diff --git a/crates/crank-registry/src/migrations/authority.rs b/crates/crank-registry/src/migrations/authority.rs index 85fe7c2..3d629e1 100644 --- a/crates/crank-registry/src/migrations/authority.rs +++ b/crates/crank-registry/src/migrations/authority.rs @@ -1,6 +1,7 @@ use super::admin_auth_lifecycle_v8; use super::agent_catalog_lifecycle_v9; use super::approval_side_effects_v10; +use super::artifact_cleanup_indexes_v13; use super::artifact_metadata_v12; use super::execution_outcome_v5; use super::master_key_identity_v7; @@ -21,8 +22,8 @@ mod contract; use contract::baseline_source_digest; use contract::validate_descriptors; const MIGRATION_LOCK_ID: i64 = 0x4352_414E_4B4D_4947; -const CURRENT_VERSION: i64 = 12; -const IMPLEMENTED_VERSIONS: &[i64] = &[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]; +const CURRENT_VERSION: i64 = 13; +const IMPLEMENTED_VERSIONS: &[i64] = &[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]; const BASELINE_SOURCE_SHA256: &str = "eb1656fc5b4b5be9ee390d237d1d58e4b2274ae5ba9b7ba06a2f3f860dfda675"; const CONSOLIDATION_SOURCE: &str = include_str!("consolidation_v2.sql"); @@ -254,6 +255,12 @@ impl MigrationAuthority { artifact_metadata_v12::SOURCE_SHA256, 11, ), + expand_descriptor( + 13, + "artifact-cleanup-indexes-v13", + artifact_cleanup_indexes_v13::SOURCE_SHA256, + 12, + ), ] } pub fn validate_sequence() -> Result<(), MigrationError> { @@ -379,6 +386,9 @@ impl MigrationAuthority { if from < 12 { artifact_metadata_v12::apply(&mut transaction, &Self::sequence()[11]).await?; } + if from < 13 { + artifact_cleanup_indexes_v13::apply(&mut transaction, &Self::sequence()[12]).await?; + } transaction .commit() .await diff --git a/crates/crank-registry/src/migrations/authority/contract.rs b/crates/crank-registry/src/migrations/authority/contract.rs index a9c8023..216ec7a 100644 --- a/crates/crank-registry/src/migrations/authority/contract.rs +++ b/crates/crank-registry/src/migrations/authority/contract.rs @@ -128,6 +128,8 @@ pub(super) fn validate_descriptors( != onboarding_product_events_v11::SOURCE_SHA256 || sha256_hex(artifact_metadata_v12::SOURCE.as_bytes()) != artifact_metadata_v12::SOURCE_SHA256 + || sha256_hex(artifact_cleanup_indexes_v13::SOURCE.as_bytes()) + != artifact_cleanup_indexes_v13::SOURCE_SHA256 { return Err(MigrationError::new( "invalid_contract", diff --git a/crates/crank-registry/src/migrations/schema_guard.rs b/crates/crank-registry/src/migrations/schema_guard.rs index 06d6a0c..5d7b69e 100644 --- a/crates/crank-registry/src/migrations/schema_guard.rs +++ b/crates/crank-registry/src/migrations/schema_guard.rs @@ -624,6 +624,13 @@ pub(super) async fn validate_schema_fingerprint( } else { super::schema_guard_v12::validate_v12_artifact_metadata(connection).await?; } + if current_version < 13 { + if !super::schema_guard_v13::validate_v13_absent(connection).await? { + return Err(schema_error(current_version)); + } + } else { + super::schema_guard_v13::validate_v13_artifact_cleanup_indexes(connection).await?; + } Ok(()) } diff --git a/crates/crank-registry/src/migrations/schema_guard_v12.rs b/crates/crank-registry/src/migrations/schema_guard_v12.rs index 1f951af..6484a5d 100644 --- a/crates/crank-registry/src/migrations/schema_guard_v12.rs +++ b/crates/crank-registry/src/migrations/schema_guard_v12.rs @@ -12,6 +12,18 @@ const OWNED_RELATIONS: &[(&str, &str)] = &[ ("artifact_sources_workspace_created_idx", "i"), ]; +const OWNED_RELATIONS_WITH_V13_INDEXES: &[(&str, &str)] = &[ + ("artifact_blobs", "r"), + ("artifact_blobs_artifact_ref_key", "i"), + ("artifact_blobs_expired_claim_idx", "i"), + ("artifact_blobs_pkey", "i"), + ("artifact_sources", "r"), + ("artifact_sources_blob_lifecycle_detached_idx", "i"), + ("artifact_sources_openapi_dangling_idx", "i"), + ("artifact_sources_pkey", "i"), + ("artifact_sources_workspace_created_idx", "i"), +]; + const BLOB_COLUMNS: &[(&str, &str, bool, Option<&str>)] = &[ ("digest", "text", false, None), ("artifact_ref", "text", false, None), @@ -141,7 +153,8 @@ pub(super) async fn validate_v12_absent( pub(super) async fn validate_v12_artifact_metadata( connection: &mut PgConnection, ) -> Result<(), MigrationError> { - if owned_relations(connection).await? != OWNED_RELATIONS { + let relations = owned_relations(connection).await?; + if relations != OWNED_RELATIONS && relations != OWNED_RELATIONS_WITH_V13_INDEXES { return Err(schema_error(12)); } validate_table_properties(connection).await?; @@ -202,7 +215,7 @@ async fn owned_relations( let kind = row .try_get::("relkind") .map_err(|_| MigrationError::storage("preflight.schema"))?; - let expected = OWNED_RELATIONS + let expected = OWNED_RELATIONS_WITH_V13_INDEXES .iter() .copied() .find(|(expected_name, expected_kind)| { @@ -346,14 +359,22 @@ async fn validate_indexes(connection: &mut PgConnection) -> Result<(), Migration .iter() .filter_map(|row| row.try_get::("index_name").ok()) .collect::>(); - if names - != [ - "artifact_blobs_artifact_ref_key", - "artifact_blobs_pkey", - "artifact_sources_pkey", - "artifact_sources_workspace_created_idx", - ] - { + let base_indexes = [ + "artifact_blobs_artifact_ref_key", + "artifact_blobs_pkey", + "artifact_sources_pkey", + "artifact_sources_workspace_created_idx", + ]; + let expanded_indexes = [ + "artifact_blobs_artifact_ref_key", + "artifact_blobs_expired_claim_idx", + "artifact_blobs_pkey", + "artifact_sources_blob_lifecycle_detached_idx", + "artifact_sources_openapi_dangling_idx", + "artifact_sources_pkey", + "artifact_sources_workspace_created_idx", + ]; + if names != base_indexes && names != expanded_indexes { return Err(schema_error(12)); } let row = rows.iter().find(|row| { diff --git a/crates/crank-registry/src/migrations/schema_guard_v13.rs b/crates/crank-registry/src/migrations/schema_guard_v13.rs new file mode 100644 index 0000000..bee1571 --- /dev/null +++ b/crates/crank-registry/src/migrations/schema_guard_v13.rs @@ -0,0 +1,178 @@ +use sqlx::{PgConnection, Row, query}; + +use super::{ + authority::MigrationError, + schema_guard::{normalize_definition, schema_error}, +}; + +struct IndexContract { + name: &'static str, + table: &'static str, + columns: &'static [&'static str], + predicate: Option<&'static str>, +} + +const CLEANUP_INDEXES: &[IndexContract] = &[ + IndexContract { + name: "artifact_blobs_expired_claim_idx", + table: "artifact_blobs", + columns: &["claim_expires_at", "digest"], + predicate: Some("claim_tokenisnotnull"), + }, + IndexContract { + name: "artifact_sources_blob_lifecycle_detached_idx", + table: "artifact_sources", + columns: &["blob_digest", "lifecycle", "detached_at"], + predicate: None, + }, + IndexContract { + name: "artifact_sources_openapi_dangling_idx", + table: "artifact_sources", + columns: &["created_at", "workspace_id", "source_id"], + predicate: Some("lifecycle='active'::textand\"left\"source_id,12='src_openapi_'::text"), + }, + IndexContract { + name: "import_jobs_expires_at_idx", + table: "import_jobs", + columns: &["expires_at", "id"], + predicate: None, + }, + IndexContract { + name: "import_jobs_openapi_source_idx", + table: "import_jobs", + columns: &[ + "workspace_id", + "preview_payload->'source'::text->>'source_id'::text", + ], + predicate: Some("preview_payload?'source'::text"), + }, +]; + +pub(super) async fn validate_v13_absent( + connection: &mut PgConnection, +) -> Result { + Ok(cleanup_indexes(connection).await?.is_empty()) +} + +pub(super) async fn validate_v13_artifact_cleanup_indexes( + connection: &mut PgConnection, +) -> Result<(), MigrationError> { + let rows = query( + "select idx.relname as index_name, t.relname as table_name, am.amname as access_method, + i.indisvalid, i.indisready, i.indisunique, i.indnkeyatts, i.indnatts, + pg_get_indexdef(i.indexrelid, 1, true) as first_column, + pg_get_indexdef(i.indexrelid, 2, true) as second_column, + pg_get_indexdef(i.indexrelid, 3, true) as third_column, + pg_get_expr(i.indpred, i.indrelid) as predicate + from pg_catalog.pg_index i + join pg_catalog.pg_class idx on idx.oid = i.indexrelid + join pg_catalog.pg_class t on t.oid = i.indrelid + join pg_catalog.pg_namespace n on n.oid = t.relnamespace + join pg_catalog.pg_am am on am.oid = idx.relam + where n.nspname = current_schema() + and idx.relname = any($1) + order by idx.relname", + ) + .bind(index_names()) + .fetch_all(connection) + .await + .map_err(|_| MigrationError::storage("preflight.schema"))?; + if rows.len() != CLEANUP_INDEXES.len() { + return Err(schema_error(13)); + } + + for row in rows { + let name = row + .try_get::("index_name") + .map_err(|_| MigrationError::storage("preflight.schema"))?; + let Some(contract) = CLEANUP_INDEXES.iter().find(|index| index.name == name) else { + return Err(schema_error(13)); + }; + if !index_matches_contract(&row, contract) { + return Err(schema_error(13)); + } + } + Ok(()) +} + +fn index_matches_contract(row: &sqlx::postgres::PgRow, contract: &IndexContract) -> bool { + let Ok(table_name) = row.try_get::("table_name") else { + return false; + }; + let Ok(access_method) = row.try_get::("access_method") else { + return false; + }; + let Ok(indnkeyatts) = row.try_get::("indnkeyatts") else { + return false; + }; + let Ok(indnatts) = row.try_get::("indnatts") else { + return false; + }; + let columns = [ + row.try_get::, _>("first_column"), + row.try_get::, _>("second_column"), + row.try_get::, _>("third_column"), + ]; + let expected_columns = contract + .columns + .iter() + .map(|column| normalize_definition(column)) + .collect::>(); + let actual_columns = columns + .iter() + .take(contract.columns.len()) + .map(|column| { + column + .as_ref() + .ok() + .and_then(|column| column.as_deref()) + .map(normalize_definition) + }) + .collect::>>(); + let trailing_columns_are_absent = columns.iter().skip(contract.columns.len()).all(|column| { + column + .as_ref() + .is_ok_and(|column| column.as_deref().is_none_or(str::is_empty)) + }); + let predicate = row + .try_get::, _>("predicate") + .ok() + .flatten() + .map(|predicate| normalize_definition(&predicate)); + + table_name == contract.table + && access_method == "btree" + && row.try_get::("indisvalid").ok() == Some(true) + && row.try_get::("indisready").ok() == Some(true) + && row.try_get::("indisunique").ok() == Some(false) + && indnkeyatts == contract.columns.len() as i16 + && indnatts == contract.columns.len() as i16 + && actual_columns.as_ref() == Some(&expected_columns) + && trailing_columns_are_absent + && predicate.as_deref() == contract.predicate +} + +fn index_names() -> Vec<&'static str> { + CLEANUP_INDEXES.iter().map(|index| index.name).collect() +} + +async fn cleanup_indexes(connection: &mut PgConnection) -> Result, MigrationError> { + query( + "select idx.relname + from pg_catalog.pg_class idx + join pg_catalog.pg_namespace n on n.oid = idx.relnamespace + where n.nspname = current_schema() + and idx.relname = any($1) + order by idx.relname", + ) + .bind(index_names()) + .fetch_all(connection) + .await + .map_err(|_| MigrationError::storage("preflight.schema"))? + .into_iter() + .map(|row| { + row.try_get("relname") + .map_err(|_| MigrationError::storage("preflight.schema")) + }) + .collect() +} diff --git a/crates/crank-registry/src/model.rs b/crates/crank-registry/src/model.rs index 96e1c07..01f4a89 100644 --- a/crates/crank-registry/src/model.rs +++ b/crates/crank-registry/src/model.rs @@ -673,6 +673,17 @@ pub struct ImportJobApplyResult { pub skipped: Vec, } +/// Result of one bounded import-source maintenance pass. +/// +/// The caller may schedule another pass when `more_work` is true. The report +/// intentionally contains only counts so it is safe to use in telemetry. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ImportJobCleanupReport { + pub deleted_jobs: u64, + pub detached_sources: u64, + pub more_work: bool, +} + #[derive(Clone, Debug, PartialEq)] pub struct ApplyImportJobRequest<'a> { pub id: &'a ImportJobId, @@ -680,6 +691,10 @@ pub struct ApplyImportJobRequest<'a> { pub application_key: &'a str, pub conflict_mode: ImportConflictMode, pub operations: &'a [ImportOperationDraft], + /// Outcomes determined before the registry transaction (for example, + /// selected OpenAPI keys absent from the immutable preview). They are + /// persisted with the canonical result so retries remain equivalent. + pub pre_skipped: &'a [SkippedImportOperation], pub finished_at: &'a OffsetDateTime, } diff --git a/crates/crank-registry/src/postgres/artifact_source.rs b/crates/crank-registry/src/postgres/artifact_source.rs index 22b4032..d868f28 100644 --- a/crates/crank-registry/src/postgres/artifact_source.rs +++ b/crates/crank-registry/src/postgres/artifact_source.rs @@ -502,8 +502,12 @@ impl PostgresRegistry { .ok_or_else(|| source_not_found(request.source_id))?; if recorded.lifecycle == ArtifactSourceLifecycle::Detached { - if recorded.detached_at == Some(request.detached_at) - && request.expected_updated_at == Some(recorded.created_at) + // `None` is an unconditional detach authority. Retrying it after + // the state transition must remain idempotent; callers that need + // optimistic fencing still supply an exact timestamp below. + if request.expected_updated_at.is_none() + || (recorded.detached_at == Some(request.detached_at) + && request.expected_updated_at == Some(recorded.created_at)) { transaction.commit().await?; return Ok(recorded); @@ -544,7 +548,7 @@ impl PostgresRegistry { pub async fn read_artifact_source( &self, - store: &ArtifactStore, + store: std::sync::Arc, workspace_id: &WorkspaceId, source_id: &ArtifactSourceId, ) -> Result { @@ -555,7 +559,6 @@ impl PostgresRegistry { return Err(RegistryError::SourceUnavailable); } - let store = store.clone(); let artifact_ref = source.blob.artifact_ref.clone(); let bytes = tokio::task::spawn_blocking(move || store.read(&artifact_ref)) .await diff --git a/crates/crank-registry/src/postgres/import_job.rs b/crates/crank-registry/src/postgres/import_job.rs index 58e8c65..e64a6a9 100644 --- a/crates/crank-registry/src/postgres/import_job.rs +++ b/crates/crank-registry/src/postgres/import_job.rs @@ -3,7 +3,7 @@ use crate::{ArtifactSourceId, ImportJobSourceEnvelope}; use crank_artifacts::ArtifactRef; const APPLICATION_RESULT_KEY: &str = "_crank_application_result"; -const IMPORT_JOB_CLEANUP_BATCH: i64 = 128; +const IMPORT_JOB_CLEANUP_BATCH: u32 = 128; const DANGLING_OPENAPI_SOURCE_GRACE: time::Duration = time::Duration::minutes(5); impl PostgresRegistry { @@ -143,7 +143,16 @@ impl PostgresRegistry { } } - pub async fn delete_expired_import_jobs(&self) -> Result { + /// Runs a single bounded cleanup pass suitable for a periodic worker. + /// + /// A corrupt historic payload is not trusted for source detachment, but it + /// must not pin every later cleanup pass: the expired job is deleted and + /// its source is subsequently eligible for the bounded dangling sweep. + pub async fn cleanup_expired_import_jobs( + &self, + limit: u32, + ) -> Result { + let limit = i64::from(limit.clamp(1, IMPORT_JOB_CLEANUP_BATCH)); let mut transaction = self.pool.begin().await?; let now = sqlx::query_scalar::<_, OffsetDateTime>("select now()") .fetch_one(&mut *transaction) @@ -156,22 +165,28 @@ impl PostgresRegistry { limit $1 for update skip locked", ) - .bind(IMPORT_JOB_CLEANUP_BATCH) + .bind(limit) .fetch_all(&mut *transaction) .await?; let mut expired_ids = Vec::with_capacity(expired.len()); + let mut detached_sources = 0; for row in &expired { expired_ids.push(row.try_get::("id")?); let workspace_id = WorkspaceId::new(row.try_get::("workspace_id")?); let payload = row.try_get::("preview_payload")?; - if let Some(source) = source_from_payload(&payload)? { - detach_source_in_transaction( + // The job itself is expired regardless of whether a legacy or + // corrupt payload can be decoded. Do not let one bad row roll + // back cleanup for every tenant. + if let Ok(Some(source)) = source_from_payload(&payload) + && detach_source_in_transaction( &mut transaction, &workspace_id, &source.source_id, now, ) - .await?; + .await? + { + detached_sources += 1; } } let deleted = if expired_ids.is_empty() { @@ -206,17 +221,42 @@ impl PostgresRegistry { for update of s skip locked", ) .bind(dangling_cutoff) - .bind(IMPORT_JOB_CLEANUP_BATCH) + .bind(limit) .fetch_all(&mut *transaction) .await?; + let has_more_dangling = dangling.len() == limit as usize; for row in dangling { let workspace_id = WorkspaceId::new(row.try_get::("workspace_id")?); let source_id = ArtifactSourceId::new(row.try_get::("source_id")?); - detach_source_in_transaction(&mut transaction, &workspace_id, &source_id, now).await?; + if detach_source_in_transaction(&mut transaction, &workspace_id, &source_id, now) + .await? + { + detached_sources += 1; + } } transaction.commit().await?; - Ok(deleted) + Ok(ImportJobCleanupReport { + deleted_jobs: deleted, + detached_sources, + more_work: expired.len() == limit as usize || has_more_dangling, + }) + } + + /// Exhausts currently visible cleanup work for startup and request paths. + /// Periodic workers should use [`Self::cleanup_expired_import_jobs`] to + /// keep each tick bounded. + pub async fn delete_expired_import_jobs(&self) -> Result { + let mut deleted_jobs = 0; + loop { + let report = self + .cleanup_expired_import_jobs(IMPORT_JOB_CLEANUP_BATCH) + .await?; + deleted_jobs += report.deleted_jobs; + if !report.more_work { + return Ok(deleted_jobs); + } + } } } @@ -286,6 +326,7 @@ async fn apply_import_job_transaction( let mut result = ImportJobApplyResult { application_key: request.application_key.to_owned(), + skipped: request.pre_skipped.to_vec(), ..ImportJobApplyResult::default() }; for draft in request.operations { @@ -360,7 +401,7 @@ async fn apply_import_job_transaction( .execute(&mut **tx) .await?; - detach_source_in_transaction( + let _ = detach_source_in_transaction( tx, request.workspace_id, &source.source_id, @@ -430,8 +471,8 @@ async fn detach_source_in_transaction( workspace_id: &WorkspaceId, source_id: &ArtifactSourceId, detached_at: OffsetDateTime, -) -> Result<(), RegistryError> { - sqlx::query( +) -> Result { + let detached = sqlx::query( "update artifact_sources set lifecycle = 'detached', updated_at = $1, detached_at = $1 where workspace_id = $2 and source_id = $3 @@ -441,8 +482,9 @@ async fn detach_source_in_transaction( .bind(workspace_id.as_str()) .bind(source_id.as_str()) .execute(&mut **transaction) - .await?; - Ok(()) + .await? + .rows_affected(); + Ok(detached == 1) } fn stored_application_result( diff --git a/crates/crank-registry/src/postgres/mod.rs b/crates/crank-registry/src/postgres/mod.rs index 9806a8c..e516250 100644 --- a/crates/crank-registry/src/postgres/mod.rs +++ b/crates/crank-registry/src/postgres/mod.rs @@ -45,8 +45,8 @@ use crate::{ CreatePlatformApiKeyRequest, CreateSecretRequest, CreateVersionRequest, CreateWorkspaceRequest, CreateYamlImportJobRequest, DecideApprovalRequest, DescriptorMetadata, ExpireApprovalRequest, FinishApprovalRequest, FinishImportJobRequest, - ImportConflictMode, ImportJob, ImportJobApplyResult, ImportJobId, ImportJobStatus, - InvitationRecord, InvocationHistoryLoss, InvocationHistoryLossCategory, + ImportConflictMode, ImportJob, ImportJobApplyResult, ImportJobCleanupReport, ImportJobId, + ImportJobStatus, InvitationRecord, InvocationHistoryLoss, InvocationHistoryLossCategory, InvocationHistoryWriteOutcome, InvocationLogRecord, InvocationRetentionOutcome, InvocationRetentionPolicy, InvocationRetentionStatus, ListApprovalRequestsQuery, ListInvocationLogsQuery, ListProductEventsQuery, MasterKeyIdentityCandidate, diff --git a/crates/crank-registry/tests/integration/artifact_sources.rs b/crates/crank-registry/tests/integration/artifact_sources.rs index 11f7d9a..70392a5 100644 --- a/crates/crank-registry/tests/integration/artifact_sources.rs +++ b/crates/crank-registry/tests/integration/artifact_sources.rs @@ -329,6 +329,16 @@ async fn source_relations_are_scoped_replayable_pageable_and_detachable() { .await .unwrap(); assert_eq!(retry, detached); + let unconditional_retry = registry + .detach_artifact_source(DetachArtifactSourceRequest { + workspace_id: &workspace_a, + source_id: &source_id, + expected_updated_at: None, + detached_at: timestamp("2026-08-26T10:05:00Z"), + }) + .await + .unwrap(); + assert_eq!(unconditional_retry, detached); assert!(matches!( registry .detach_artifact_source(DetachArtifactSourceRequest { @@ -350,7 +360,7 @@ async fn source_relations_are_scoped_replayable_pageable_and_detachable() { ); assert!(matches!( registry - .read_artifact_source(&store, &workspace_a, &source_id) + .read_artifact_source(std::sync::Arc::new(store.clone()), &workspace_a, &source_id) .await, Err(RegistryError::SourceUnavailable) )); @@ -439,7 +449,7 @@ async fn verified_read_returns_only_digest_and_size_verified_bytes() { .await .unwrap(); let verified = registry - .read_artifact_source(&store, &workspace_id, &valid_id) + .read_artifact_source(std::sync::Arc::new(store.clone()), &workspace_id, &valid_id) .await .unwrap(); assert_eq!(verified.bytes, bytes); @@ -473,7 +483,11 @@ async fn verified_read_returns_only_digest_and_size_verified_bytes() { .unwrap(); assert!(matches!( registry - .read_artifact_source(&store, &workspace_id, &missing_id) + .read_artifact_source( + std::sync::Arc::new(store.clone()), + &workspace_id, + &missing_id + ) .await, Err(RegistryError::SourceUnavailable) )); @@ -504,7 +518,11 @@ async fn verified_read_returns_only_digest_and_size_verified_bytes() { .unwrap(); assert!(matches!( registry - .read_artifact_source(&store, &workspace_id, &wrong_size_id) + .read_artifact_source( + std::sync::Arc::new(store.clone()), + &workspace_id, + &wrong_size_id + ) .await, Err(RegistryError::SourceIntegrity) )); @@ -529,7 +547,11 @@ async fn verified_read_returns_only_digest_and_size_verified_bytes() { .unwrap(); assert!(matches!( registry - .read_artifact_source(&store, &workspace_id, &unavailable_id) + .read_artifact_source( + std::sync::Arc::new(store.clone()), + &workspace_id, + &unavailable_id + ) .await, Err(RegistryError::SourceUnavailable) )); @@ -543,7 +565,7 @@ async fn verified_read_returns_only_digest_and_size_verified_bytes() { fs::write(&tampered, vec![b'x'; bytes.len()]).unwrap(); assert!(matches!( registry - .read_artifact_source(&store, &workspace_id, &valid_id) + .read_artifact_source(std::sync::Arc::new(store.clone()), &workspace_id, &valid_id) .await, Err(RegistryError::SourceIntegrity) )); @@ -749,12 +771,16 @@ async fn reconciliation_claims_are_fenced_global_and_allow_verified_revival() { .unwrap(), ArtifactClaimOutcome::ActiveReference )); + let detached_at: OffsetDateTime = sqlx::query_scalar("select clock_timestamp()") + .fetch_one(registry.pool()) + .await + .unwrap(); 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"), + detached_at, }) .await .unwrap(); diff --git a/crates/crank-registry/tests/integration/migrations.rs b/crates/crank-registry/tests/integration/migrations.rs index 7db908d..cf0cc3f 100644 --- a/crates/crank-registry/tests/integration/migrations.rs +++ b/crates/crank-registry/tests/integration/migrations.rs @@ -51,6 +51,7 @@ async fn controlled_authority_is_versioned_and_safe_under_concurrent_apply() { (10, "approval-side-effects-v10"), (11, "onboarding-product-events-v11"), (12, "artifact-metadata-v12"), + (13, "artifact-cleanup-indexes-v13"), ]; assert_eq!(rows.len(), expected.len()); for (row, (version, name)) in rows.iter().zip(expected) { @@ -106,7 +107,7 @@ async fn controlled_authority_is_versioned_and_safe_under_concurrent_apply() { assert_eq!(artifact_relations.len(), 2); assert_eq!( MigrationAuthority::preflight(first.pool()).await.unwrap(), - MigrationPreflight::Current { version: 12 } + MigrationPreflight::Current { version: 13 } ); } #[tokio::test] @@ -229,7 +230,7 @@ async fn legacy_core_baseline_is_consolidated_without_data_loss() { MigrationAuthority::preflight(&pool).await.unwrap(), MigrationPreflight::MigrationRequired { current: 1, - target: 12, + target: 13, } ); MigrationAuthority::apply(&pool).await.unwrap(); @@ -308,7 +309,13 @@ async fn future_sequence_fails_closed() { let database_url = crank_test_support::postgres_schema_url("test_future_sequence").await; let pool = sqlx::PgPool::connect(&database_url).await.unwrap(); MigrationAuthority::apply(&pool).await.unwrap(); - sqlx::query("update __crank_migrations set version = 13 where version = 12") + let current: i64 = sqlx::query_scalar("select max(version) from __crank_migrations") + .fetch_one(&pool) + .await + .unwrap(); + sqlx::query("update __crank_migrations set version = $1 where version = $2") + .bind(current + 1) + .bind(current) .execute(&pool) .await .unwrap(); @@ -336,13 +343,13 @@ async fn healthy_v2_is_reported_as_migration_required_and_upgrades_to_v3() { MigrationAuthority::preflight(&pool).await.unwrap(), MigrationPreflight::MigrationRequired { current: 2, - target: 12, + target: 13, } ); MigrationAuthority::apply(&pool).await.unwrap(); assert_eq!( MigrationAuthority::preflight(&pool).await.unwrap(), - MigrationPreflight::Current { version: 12 } + MigrationPreflight::Current { version: 13 } ); let trace_column: bool = sqlx::query_scalar( "select exists ( @@ -385,7 +392,7 @@ async fn healthy_v3_upgrades_to_v4_with_honest_legacy_snapshot_provenance() { MigrationAuthority::preflight(&pool).await.unwrap(), MigrationPreflight::MigrationRequired { current: 3, - target: 12, + target: 13, } ); MigrationAuthority::apply(&pool).await.unwrap(); @@ -445,7 +452,7 @@ async fn healthy_v4_upgrades_to_v5_without_fabricating_legacy_outcomes() { MigrationAuthority::preflight(&pool).await.unwrap(), MigrationPreflight::MigrationRequired { current: 4, - target: 12, + target: 13, } ); MigrationAuthority::apply(&pool).await.unwrap(); @@ -699,7 +706,7 @@ async fn v3_upgrade_ignores_oversized_legacy_request_ids_in_partial_index() { MigrationAuthority::apply(&pool).await.unwrap(); assert_eq!( MigrationAuthority::preflight(&pool).await.unwrap(), - MigrationPreflight::Current { version: 12 } + MigrationPreflight::Current { version: 13 } ); } async fn remove_v3_schema(pool: &sqlx::PgPool) { @@ -855,7 +862,13 @@ async fn remove_v11_schema(pool: &sqlx::PgPool) { } async fn remove_v12_schema(pool: &sqlx::PgPool) { sqlx::raw_sql( - "drop table if exists artifact_sources; + "drop index if exists import_jobs_openapi_source_idx; + drop index if exists import_jobs_expires_at_idx; + drop index if exists artifact_sources_openapi_dangling_idx; + drop index if exists artifact_sources_blob_lifecycle_detached_idx; + drop index if exists artifact_blobs_expired_claim_idx; + delete from __crank_migrations where version = 13; + drop table if exists artifact_sources; drop table if exists artifact_blobs; delete from __crank_migrations where version = 12;", ) diff --git a/crates/crank-registry/tests/integration/migrations/artifact_metadata.rs b/crates/crank-registry/tests/integration/migrations/artifact_metadata.rs index f3de502..6fd67b8 100644 --- a/crates/crank-registry/tests/integration/migrations/artifact_metadata.rs +++ b/crates/crank-registry/tests/integration/migrations/artifact_metadata.rs @@ -1,7 +1,7 @@ use super::*; #[tokio::test] -async fn healthy_v11_upgrades_to_v12_without_rewriting_prior_ledger() { +async fn healthy_v11_upgrades_to_v13_without_rewriting_prior_ledger() { let database_url = crank_test_support::postgres_schema_url("test_v11_to_v12_artifacts").await; let pool = sqlx::PgPool::connect(&database_url).await.unwrap(); MigrationAuthority::apply(&pool).await.unwrap(); @@ -27,7 +27,7 @@ async fn healthy_v11_upgrades_to_v12_without_rewriting_prior_ledger() { MigrationAuthority::preflight(&pool).await.unwrap(), MigrationPreflight::MigrationRequired { current: 11, - target: 12, + target: 13, } ); @@ -51,22 +51,22 @@ async fn healthy_v11_upgrades_to_v12_without_rewriting_prior_ledger() { .collect::>(); assert_eq!(after, prior); - let v12_applied_at: time::OffsetDateTime = - sqlx::query_scalar("select applied_at from __crank_migrations where version = 12") + let v13_applied_at: time::OffsetDateTime = + sqlx::query_scalar("select applied_at from __crank_migrations where version = 13") .fetch_one(&pool) .await .unwrap(); MigrationAuthority::apply(&pool).await.unwrap(); let replayed_at: time::OffsetDateTime = - sqlx::query_scalar("select applied_at from __crank_migrations where version = 12") + sqlx::query_scalar("select applied_at from __crank_migrations where version = 13") .fetch_one(&pool) .await .unwrap(); - assert_eq!(replayed_at, v12_applied_at); + assert_eq!(replayed_at, v13_applied_at); } #[tokio::test] -async fn v12_exact_guard_rejects_column_constraint_index_and_relation_drift() { +async fn v12_guard_still_rejects_column_constraint_index_and_relation_drift() { let database_url = crank_test_support::postgres_schema_url("test_v12_artifact_drift").await; let pool = sqlx::PgPool::connect(&database_url).await.unwrap(); MigrationAuthority::apply(&pool).await.unwrap(); @@ -158,7 +158,7 @@ async fn v12_exact_guard_rejects_column_constraint_index_and_relation_drift() { sqlx::raw_sql(restore).execute(&pool).await.unwrap(); assert_eq!( MigrationAuthority::preflight(&pool).await.unwrap(), - MigrationPreflight::Current { version: 12 }, + MigrationPreflight::Current { version: 13 }, "restore: {restore}" ); } @@ -175,3 +175,44 @@ async fn v12_exact_guard_rejects_column_constraint_index_and_relation_drift() { assert_eq!(error.code(), "partial_sequence"); assert_eq!(error.version(), Some(12)); } + +#[tokio::test] +async fn v13_guard_rejects_wrong_cleanup_index_order_and_predicate() { + let database_url = + crank_test_support::postgres_schema_url("test_v13_cleanup_index_drift").await; + let pool = sqlx::PgPool::connect(&database_url).await.unwrap(); + MigrationAuthority::apply(&pool).await.unwrap(); + + for (drift, restore) in [ + ( + "drop index artifact_sources_blob_lifecycle_detached_idx; + create index artifact_sources_blob_lifecycle_detached_idx + on artifact_sources(lifecycle, blob_digest, detached_at);", + "drop index artifact_sources_blob_lifecycle_detached_idx; + create index artifact_sources_blob_lifecycle_detached_idx + on artifact_sources(blob_digest, lifecycle, detached_at);", + ), + ( + "drop index artifact_sources_openapi_dangling_idx; + create index artifact_sources_openapi_dangling_idx + on artifact_sources(created_at, workspace_id, source_id) + where lifecycle = 'active';", + "drop index artifact_sources_openapi_dangling_idx; + create index artifact_sources_openapi_dangling_idx + on artifact_sources(created_at, workspace_id, source_id) + where lifecycle = 'active' and left(source_id, 12) = 'src_openapi_';", + ), + ] { + sqlx::raw_sql(drift).execute(&pool).await.unwrap(); + let error = MigrationAuthority::preflight(&pool).await.unwrap_err(); + assert_eq!(error.code(), "partial_sequence", "drift: {drift}"); + assert_eq!(error.version(), Some(13), "drift: {drift}"); + + sqlx::raw_sql(restore).execute(&pool).await.unwrap(); + assert_eq!( + MigrationAuthority::preflight(&pool).await.unwrap(), + MigrationPreflight::Current { version: 13 }, + "restore: {restore}" + ); + } +} diff --git a/crates/crank-registry/tests/integration/migrations/rollback.rs b/crates/crank-registry/tests/integration/migrations/rollback.rs index c9364a2..c0454c5 100644 --- a/crates/crank-registry/tests/integration/migrations/rollback.rs +++ b/crates/crank-registry/tests/integration/migrations/rollback.rs @@ -63,7 +63,7 @@ async fn failed_artifact_metadata_migration_rolls_back_schema_and_ledger() { MigrationAuthority::preflight(&pool).await.unwrap(), MigrationPreflight::MigrationRequired { current: 11, - target: 12, + target: 13, } ); } diff --git a/deploy/community/docker-compose.images.yml b/deploy/community/docker-compose.images.yml index 36059d1..30a4690 100644 --- a/deploy/community/docker-compose.images.yml +++ b/deploy/community/docker-compose.images.yml @@ -55,12 +55,24 @@ services: POSTGRES_IDLE_TIMEOUT_MS: ${POSTGRES_IDLE_TIMEOUT_MS:-600000} POSTGRES_MAX_LIFETIME_MS: ${POSTGRES_MAX_LIFETIME_MS:-1800000} + artifact-storage-init: + image: ${CRANK_ADMIN_API_IMAGE:-git.itexp.me/bsodfather/crank-community-admin-api:main} + entrypoint: ["/bin/sh", "-ec"] + command: ["install -d -m 700 -- \"$${CRANK_STORAGE_ROOT}\""] + restart: "no" + environment: + CRANK_STORAGE_ROOT: ${CRANK_STORAGE_ROOT:-/var/lib/crank/storage} + volumes: + - artifact_storage:${CRANK_STORAGE_ROOT:-/var/lib/crank/storage} + admin-api: image: ${CRANK_ADMIN_API_IMAGE:-git.itexp.me/bsodfather/crank-community-admin-api:main} restart: unless-stopped depends_on: migrate: condition: service_completed_successfully + artifact-storage-init: + condition: service_completed_successfully postgres: condition: service_healthy required: false diff --git a/deploy/community/docker-compose.yml b/deploy/community/docker-compose.yml index 1635188..484c70b 100644 --- a/deploy/community/docker-compose.yml +++ b/deploy/community/docker-compose.yml @@ -55,6 +55,19 @@ services: POSTGRES_IDLE_TIMEOUT_MS: ${POSTGRES_IDLE_TIMEOUT_MS:-600000} POSTGRES_MAX_LIFETIME_MS: ${POSTGRES_MAX_LIFETIME_MS:-1800000} + artifact-storage-init: + image: ${CRANK_ADMIN_API_IMAGE:-crank/admin-api:dev} + build: + context: ../.. + dockerfile: apps/admin-api/Dockerfile + entrypoint: ["/bin/sh", "-ec"] + command: ["install -d -m 700 -- \"$${CRANK_STORAGE_ROOT}\""] + restart: "no" + environment: + CRANK_STORAGE_ROOT: ${CRANK_STORAGE_ROOT:-/var/lib/crank/storage} + volumes: + - artifact_storage:${CRANK_STORAGE_ROOT:-/var/lib/crank/storage} + admin-api: image: ${CRANK_ADMIN_API_IMAGE:-crank/admin-api:dev} build: @@ -64,6 +77,8 @@ services: depends_on: migrate: condition: service_completed_successfully + artifact-storage-init: + condition: service_completed_successfully environment: POSTGRES_HOST: ${POSTGRES_HOST:-postgres} POSTGRES_PORT: ${POSTGRES_PORT:-5432} diff --git a/docker-compose.yml b/docker-compose.yml index 16cb902..c7d3e25 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -39,6 +39,19 @@ services: postgres: condition: service_healthy + artifact-storage-init: + image: ${CRANK_ADMIN_API_IMAGE:-crank/admin-api:dev} + build: + context: . + dockerfile: apps/admin-api/Dockerfile + entrypoint: ["/bin/sh", "-ec"] + command: ["install -d -m 700 -- \"$${CRANK_STORAGE_ROOT}\""] + restart: "no" + environment: + CRANK_STORAGE_ROOT: ${CRANK_STORAGE_ROOT:-/var/lib/crank/storage} + volumes: + - artifact_storage:${CRANK_STORAGE_ROOT:-/var/lib/crank/storage} + admin-api: image: ${CRANK_ADMIN_API_IMAGE:-crank/admin-api:dev} build: @@ -99,6 +112,8 @@ services: depends_on: migrate: condition: service_completed_successfully + artifact-storage-init: + condition: service_completed_successfully postgres: condition: service_healthy volumes: @@ -106,7 +121,7 @@ services: ports: - "${CRANK_PUBLISH_BIND:-127.0.0.1}:3001:3001" healthcheck: - test: ["CMD", "curl", "--fail", "http://127.0.0.1:3001/health"] + test: ["CMD", "curl", "--fail", "http://127.0.0.1:3001/ready"] interval: 15s timeout: 5s retries: 5 diff --git a/scripts/validate-capability-run.py b/scripts/validate-capability-run.py index 6659251..f3e0e76 100644 --- a/scripts/validate-capability-run.py +++ b/scripts/validate-capability-run.py @@ -122,6 +122,18 @@ def run_schema(schema: Any) -> dict[str, Any]: return definition +def validate_accepted_semantics(candidate: Any) -> None: + """Fail closed: accepted evidence must be an automated passing run.""" + if not isinstance(candidate, dict): + raise ValidationError('/', 'accepted evidence candidate must be an object') + if candidate.get('accepted') is not True: + raise ValidationError('/accepted', 'accepted evidence is required') + if candidate.get('execution_verdict') != 'pass': + raise ValidationError('/execution_verdict', 'accepted evidence requires execution_verdict=pass') + if candidate.get('evidence_mode') != 'automated': + raise ValidationError('/evidence_mode', 'accepted evidence requires evidence_mode=automated') + + def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--schema', required=True) @@ -133,8 +145,8 @@ def main() -> int: schema = load_json(Path(args.schema)) candidate = load_json(Path(args.candidate)) validate(candidate, run_schema(schema)) - if args.require_accepted and candidate.get('accepted') is not True: - raise ValidationError('/accepted', 'accepted evidence is required') + if args.require_accepted: + validate_accepted_semantics(candidate) except ValidationError as error: print(f'INVALID_CAPABILITY_RUN pointer={error.pointer} reason={error.message}', file=sys.stderr) return 1 diff --git a/tests/unit/test_validate_capability_run.py b/tests/unit/test_validate_capability_run.py index 9aec00e..c593117 100644 --- a/tests/unit/test_validate_capability_run.py +++ b/tests/unit/test_validate_capability_run.py @@ -94,6 +94,18 @@ class CapabilityRunValidatorTests(unittest.TestCase): self.assertNotEqual(result.returncode, 0) self.assertIn("accepted evidence is required", result.stderr) + def test_require_accepted_rejects_true_flag_without_passing_automated_evidence(self) -> None: + for field, value, expected in ( + ("execution_verdict", "fail", "accepted evidence requires execution_verdict=pass"), + ("evidence_mode", "manual_only", "accepted evidence requires evidence_mode=automated"), + ): + with self.subTest(field=field): + candidate_value = candidate() + candidate_value[field] = value + result = self.validate(json.dumps(candidate_value), require_accepted=True) + self.assertNotEqual(result.returncode, 0) + self.assertIn(expected, result.stderr) + if __name__ == "__main__": unittest.main()