From 657f89c2e89d38840f937aacdbd392f8da19cc75 Mon Sep 17 00:00:00 2001 From: Zibi Braniecki Date: Thu, 22 Oct 2026 08:59:05 -0700 Subject: [PATCH] Carry verified upload evidence through backup finalization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Successful transfers now publish logical digests from the exact transform reads, after EOF and durable wire-size verification. Hash owned immutable raw staging in place and copy only legacy unowned input. Persist a versioned completion result atomically with attempt/lease-fenced job completion by merging into the current intent, preserving concurrent updates. Sandbox-manager forwards the digests for existing restore verification and the catalog stores optional metadata/object bindings with immutable replay checks. Additive wire fields tolerate older workers and unknown optional evidence. Missing optional proof does not turn a successful ordinary upload into failure. Ambiguous roles remain uncertified; generated btrfs-send sources remain unsupported for proof-required recovery. ETag is an object identity token, not a digest; current object-store metadata provides no version ID. These proofs bind an honest authenticated uploader, not independently verified plaintext against a compromised worker. Schema deploys before catalog writers; ordinary completion stays backward compatible. Validation: payload overwrite regression failed with the old write and passes with the merge. All 62 common-jobs and 219 volume-manager tests passed; all 30 sandbox upload tests and the PostgreSQL evidence/finalization/restore test passed. All-target/all-feature Clippy for affected jobs, transfer, sandbox, catalog and internal API packages passed. Public diff review and Gitleaks passed. An opt-in 16 MiB debug-build benchmark measured approximately 56–84 ms added hashing CPU, zero extra owned-raw staging disk, and one logical-size copy for legacy raw. Workspace all-target/all-feature Clippy and affected aarch64 Linux musl compilation/lint passed; Linux/KVM live recovery remains unvalidated. AI assistance: Codex contributed implementation, tests, and this description. --- Cargo.lock | 1 + .../common/core/src/types/object_store.rs | 55 ++++ crates/libs/common/jobs/src/db.rs | 37 ++- crates/libs/common/jobs/src/db/tests.rs | 70 +++++ crates/libs/common/jobs/src/lib.rs | 40 +++ crates/libs/common/jobs/src/worker.rs | 41 ++- crates/libs/common/jobs/src/worker/tests.rs | 69 +++++ crates/libs/internal/api/server/src/lib.rs | 83 ++++-- crates/libs/internal/api/server/src/tests.rs | 30 +- .../mgr/server/src/snapshot_upload_jobs.rs | 47 +++- .../server/src/snapshot_upload_jobs/tests.rs | 38 +++ crates/libs/snapshot/store/Cargo.toml | 1 + crates/libs/snapshot/store/src/lib.rs | 1 + .../snapshot/store/src/upload_evidence.rs | 40 +++ crates/libs/vol/mgr/server/src/executor.rs | 169 +++++++++-- .../mgr/server/src/executor/logical_digest.rs | 140 ++++++++++ .../src/executor/logical_digest_tests.rs | 46 +++ .../mgr/server/src/executor/stream_push.rs | 15 +- .../server/src/executor/stream_push/tests.rs | 52 +++- .../libs/vol/mgr/server/src/executor/tests.rs | 263 +++++++++++++++++- crates/libs/vol/mgr/server/src/service.rs | 30 ++ .../libs/vol/mgr/server/src/service/tests.rs | 32 +++ docs/snapshot-upload-streaming.md | 33 +++ proto/node/api/v1/cluster.proto | 10 + proto/volume/manager/v1/volume_manager.proto | 20 +- 25 files changed, 1288 insertions(+), 75 deletions(-) create mode 100644 crates/libs/snapshot/store/src/upload_evidence.rs create mode 100644 crates/libs/vol/mgr/server/src/executor/logical_digest.rs create mode 100644 crates/libs/vol/mgr/server/src/executor/logical_digest_tests.rs diff --git a/Cargo.lock b/Cargo.lock index 80210ab877..78c6f5d255 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7083,6 +7083,7 @@ dependencies = [ "pplx-space-common-core", "pplx-space-db-test-fixtures", "pplx-space-test-env", + "serde", "serde_json", "sqlx", "thiserror 2.0.20", diff --git a/crates/libs/common/core/src/types/object_store.rs b/crates/libs/common/core/src/types/object_store.rs index a025e2da3e..4c4e2a2ce7 100644 --- a/crates/libs/common/core/src/types/object_store.rs +++ b/crates/libs/common/core/src/types/object_store.rs @@ -605,5 +605,60 @@ impl FromStr for ExactStorageUri { } } +/// Canonical stable snapshot metadata for an operator recovery approval. +/// Generated transport timestamps/sizes and node-local send-source handles are +/// deliberately excluded. Every other field, including future fields, is bound. +#[cfg(feature = "serde")] +pub fn snapshot_metadata_binding(bytes: &[u8]) -> Result, SnapshotMetadataBindingError> { + let mut value: serde_json::Value = serde_json::from_slice(bytes)?; + let map = value + .as_object_mut() + .ok_or(SnapshotMetadataBindingError::Shape)?; + if map.get("version").and_then(serde_json::Value::as_u64) != Some(2) + || [ + "sandbox_id", + "local_artifact_id", + "artifact_kind", + "artifact_origin", + "execution", + "source", + ] + .iter() + .any(|key| map.get(*key).is_none_or(serde_json::Value::is_null)) + { + return Err(SnapshotMetadataBindingError::Shape); + } + for key in ["created_at", "size_bytes", "rootfs_send_source_id"] { + map.remove(key); + } + fn sort(value: &mut serde_json::Value) { + match value { + serde_json::Value::Object(map) => { + for child in map.values_mut() { + sort(child); + } + map.sort_keys(); + } + serde_json::Value::Array(values) => { + for child in values { + sort(child); + } + } + _ => {} + } + } + sort(&mut value); + Ok(serde_json::to_vec(&value)?) +} + +#[cfg(feature = "serde")] +#[derive(Debug, thiserror::Error)] +pub enum SnapshotMetadataBindingError { + #[error("recovery requires version 2 artifact identity and execution metadata")] + Shape, + #[error("invalid snapshot metadata: {0}")] + Json(#[from] serde_json::Error), +} + #[cfg(test)] mod tests; diff --git a/crates/libs/common/jobs/src/db.rs b/crates/libs/common/jobs/src/db.rs index 5455f8d9be..cf3e1da7e5 100644 --- a/crates/libs/common/jobs/src/db.rs +++ b/crates/libs/common/jobs/src/db.rs @@ -503,11 +503,26 @@ pub fn complete_leased_job( job_id: &JobId, worker_id: &str, bytes: i64, +) -> rusqlite::Result { + complete_leased_job_with_result(conn, job_id, worker_id, None, bytes, None) +} + +pub fn complete_leased_job_with_result( + conn: &Connection, + job_id: &JobId, + worker_id: &str, + expected_attempt: Option, + bytes: i64, + payload: Option<&str>, ) -> rusqlite::Result { let rows = conn.execute( "UPDATE jobs SET status = 'completed', bytes_processed = ?3, + payload = CASE WHEN json_valid(payload) AND json_type(payload) = 'object' THEN + CASE WHEN ?4 IS NULL THEN json_remove(payload, '$.completion_result') + ELSE json_set(payload, '$.completion_result', json_object('version', 1, 'attempt', attempts, 'value', json(?4))) END + ELSE payload END, error = NULL, error_kind = NULL, failure_reason = NULL, @@ -517,8 +532,9 @@ pub fn complete_leased_job( heartbeat_at = NULL WHERE job_id = ?1 AND lease_owner = ?2 - AND status = 'running'", - params![job_id, worker_id, bytes], + AND status = 'running' + AND (?5 IS NULL OR attempts = ?5)", + params![job_id, worker_id, bytes, payload, expected_attempt], )?; Ok(rows > 0) } @@ -625,14 +641,27 @@ pub fn complete_job_attempt( job_id: &JobId, expected_attempts: i32, bytes: i64, +) -> rusqlite::Result { + complete_job_attempt_with_result(conn, job_id, expected_attempts, bytes, None) +} + +pub fn complete_job_attempt_with_result( + conn: &Connection, + job_id: &JobId, + expected_attempts: i32, + bytes: i64, + payload: Option<&str>, ) -> rusqlite::Result { let changed = conn.execute( "UPDATE jobs - SET status = 'completed', bytes_processed = ?2, + SET status = 'completed', bytes_processed = ?2, payload = CASE WHEN json_valid(payload) AND json_type(payload) = 'object' THEN + CASE WHEN ?4 IS NULL THEN json_remove(payload, '$.completion_result') + ELSE json_set(payload, '$.completion_result', json_object('version', 1, 'attempt', attempts, 'value', json(?4))) END + ELSE payload END, error = NULL, error_kind = NULL, failure_reason = NULL, completed_at = datetime('now') WHERE job_id = ?1 AND status = 'running' AND attempts = ?3", - params![job_id, bytes, expected_attempts], + params![job_id, bytes, expected_attempts, payload], )?; Ok(changed == 1) } diff --git a/crates/libs/common/jobs/src/db/tests.rs b/crates/libs/common/jobs/src/db/tests.rs index 14d1f86fb6..5e2606db5b 100644 --- a/crates/libs/common/jobs/src/db/tests.rs +++ b/crates/libs/common/jobs/src/db/tests.rs @@ -534,3 +534,73 @@ fn count_failed_jobs_by_dedup_key_counts_one_failure_kind_of_one_key() { 0 ); } +#[test] +fn completion_evidence_is_atomic_and_fenced() { + for leased in [false, true] { + let conn = mem_conn(); + let id = fixture_job_id("content-proof"); + insert_job( + &conn, + InsertJob { + job_id: &id, + kind: "push", + payload: "{\"source\":\"owned\"}", + subject_id: None, + dedup_key: None, + max_attempts: 3, + traceparent: None, + priority: JobPriority::default(), + }, + ) + .unwrap(); + conn.execute( + "UPDATE jobs SET status='running',attempts=2,lease_owner='current' WHERE job_id=?1", + [id], + ) + .unwrap(); + let complete = |owner, attempt, payload| { + if leased { + complete_leased_job_with_result(&conn, &id, owner, Some(attempt), 14, Some(payload)) + } else { + complete_job_attempt_with_result(&conn, &id, attempt, 14, Some(payload)) + } + }; + assert!(!complete("previous", 1, "wrong-proof").unwrap()); + assert!(!complete("current", 1, "stale-attempt-proof").unwrap()); + let unchanged = get_job(&conn, &id).unwrap().unwrap(); + assert_eq!(unchanged.status, "running"); + assert_eq!(unchanged.payload, "{\"source\":\"owned\"}"); + conn.execute( + "UPDATE jobs SET payload=json_set(payload,'$.updated_during_upload',1) WHERE job_id=?1", + [id], + ) + .unwrap(); + assert!( + complete( + "current", + 2, + "{\"source\":\"owned\",\"proof\":\"verified\"}" + ) + .unwrap() + ); + let completed = get_job(&conn, &id).unwrap().unwrap(); + assert_eq!(completed.status, "completed"); + assert_eq!(completed.bytes_processed, 14); + assert!(completed.payload.contains("verified")); + let payload: serde_json::Value = serde_json::from_str(&completed.payload).unwrap(); + assert_eq!(payload["updated_during_upload"], 1); + assert_eq!(payload["source"], "owned"); + assert_eq!( + crate::completion_result::(&completed).unwrap()["proof"], + "verified" + ); + let mut old_retry = completed.clone(); + old_retry.attempts += 1; + assert!(crate::completion_result::(&old_retry).is_none()); + assert!(!complete("current", 2, "replacement-proof").unwrap()); + assert_eq!( + get_job(&conn, &id).unwrap().unwrap().payload, + completed.payload + ); + } +} diff --git a/crates/libs/common/jobs/src/lib.rs b/crates/libs/common/jobs/src/lib.rs index 6b71f34b9c..8d283cfb2f 100644 --- a/crates/libs/common/jobs/src/lib.rs +++ b/crates/libs/common/jobs/src/lib.rs @@ -368,6 +368,31 @@ impl JobError { } } +/// A successful execution and an optional service-owned result containing its +/// completion evidence. The worker publishes both in the same fenced SQLite +/// transition; partial or superseded executions never publish their evidence. +#[derive(Debug)] +pub struct JobCompletion { + pub bytes_processed: i64, + pub result: Option, +} + +/// Read optional completion evidence only when its schema and producing attempt +/// match. Old workers, malformed results and rollback schemas remain readable. +pub fn completion_result(job: &Job) -> Option { + if job.status != "completed" { + return None; + } + let payload: serde_json::Value = serde_json::from_str(&job.payload).ok()?; + let envelope = payload.get("completion_result")?; + if envelope.get("version")?.as_u64()? != 1 + || envelope.get("attempt")?.as_i64()? != i64::from(job.attempts) + { + return None; + } + serde_json::from_value(envelope.get("value")?.clone()).ok() +} + /// Defines the execution logic for a job kind. /// /// Implementations must be `Clone` + `Send` + `Sync` + `'static` so they @@ -381,6 +406,21 @@ pub trait JobExecutor: Clone + Send + Sync + 'static { cancellation: tokio_util::sync::CancellationToken, ) -> Result; + /// Executors may return evidence merged into the current intent at completion. + /// Existing executors preserve their intent payload unchanged. + async fn execute_with_completion( + &self, + job: &Job, + cancellation: tokio_util::sync::CancellationToken, + ) -> Result { + self.execute(job, cancellation) + .await + .map(|bytes_processed| JobCompletion { + bytes_processed, + result: None, + }) + } + /// Remove executor-owned partial state after cancellation. This runs only /// after the execute future has stopped, so cleanup cannot race a writer. async fn cleanup_cancelled(&self, _job: &Job) -> Result<(), JobError> { diff --git a/crates/libs/common/jobs/src/worker.rs b/crates/libs/common/jobs/src/worker.rs index 69da24dcf2..d76f78e7f9 100644 --- a/crates/libs/common/jobs/src/worker.rs +++ b/crates/libs/common/jobs/src/worker.rs @@ -1583,9 +1583,11 @@ impl BackgroundWorker { self.finish_cancelled_job(job).await; return; } - let result = - tokio::time::timeout(self.job_timeout, self.executor.execute(&job, cancellation)) - .await; + let result = tokio::time::timeout( + self.job_timeout, + self.executor.execute_with_completion(&job, cancellation), + ) + .await; self.cancellation_tokens .lock() .expect("common jobs cancellation mutex poisoned") @@ -1605,8 +1607,12 @@ impl BackgroundWorker { } match result { - Ok(Ok(bytes)) => { - if self.complete_job(&job, lease_owner.as_deref(), bytes).await { + Ok(Ok(completion)) => { + let bytes = completion.bytes_processed; + if self + .complete_job(&job, lease_owner.as_deref(), &completion) + .await + { span.record("otel.status_code", "OK"); tracing::info!( job_id = %job.job_id, @@ -1724,14 +1730,33 @@ impl BackgroundWorker { /// remote work; abandoning this write strands nonleased running rows. /// A persistent database outage has no completion deadline. Bounded drain /// must report non-quiescence rather than dropping this ownership. - async fn complete_job(&self, job: &Job, lease_owner: Option<&str>, bytes: i64) -> bool { + async fn complete_job( + &self, + job: &Job, + lease_owner: Option<&str>, + completion: &crate::JobCompletion, + ) -> bool { + let result_json = completion.result.as_ref().map(serde_json::Value::to_string); let mut retry_delay = Duration::from_secs(1); loop { let result = { let conn = self.lock_db_for_background(); match lease_owner { - Some(owner) => db::complete_leased_job(&conn, &job.job_id, owner, bytes), - None => db::complete_job_attempt(&conn, &job.job_id, job.attempts, bytes), + Some(owner) => db::complete_leased_job_with_result( + &conn, + &job.job_id, + owner, + Some(job.attempts), + completion.bytes_processed, + result_json.as_deref(), + ), + None => db::complete_job_attempt_with_result( + &conn, + &job.job_id, + job.attempts, + completion.bytes_processed, + result_json.as_deref(), + ), } }; match result { diff --git a/crates/libs/common/jobs/src/worker/tests.rs b/crates/libs/common/jobs/src/worker/tests.rs index 8767361ddb..c16257a1e1 100644 --- a/crates/libs/common/jobs/src/worker/tests.rs +++ b/crates/libs/common/jobs/src/worker/tests.rs @@ -1282,3 +1282,72 @@ async fn serialized_subjects_do_not_starve_other_subjects() { driver.await.expect("queue driver task should not panic"); remove_test_db(path); } + +#[derive(Clone)] +struct EvidenceExecutor { + inner: TestExecutor, +} +#[async_trait] +impl JobExecutor for EvidenceExecutor { + async fn execute(&self, job: &Job, cancellation: CancellationToken) -> Result { + self.inner.execute(job, cancellation).await + } + async fn execute_with_completion( + &self, + job: &Job, + cancellation: CancellationToken, + ) -> Result { + let bytes_processed = self.execute(job, cancellation).await?; + Ok(crate::JobCompletion { + bytes_processed, + result: Some(serde_json::json!({"digest":"synthetic-content-proof"})), + }) + } +} + +#[tokio::test] +async fn worker_merges_completion_with_concurrent_intent_and_refuses_stale_execution() { + for superseded in [false, true] { + let path = test_db_path("completion-evidence"); + let db = Arc::new(Mutex::new(open_test_db(&path))); + let executor = EvidenceExecutor { + inner: TestExecutor::blocking(), + }; + let worker = BackgroundWorker::new( + db.clone(), + executor.clone(), + WorkerConfig::new(1, Duration::from_secs(5), 3), + ); + let id = worker.submit("test", "{\"intent\":1}", None, None).unwrap(); + executor.inner.wait_started(1).await; + { + let conn = db.lock().unwrap(); + conn.execute("UPDATE jobs SET payload=json_set(payload,'$.late_intent',2),attempts=attempts+?2 WHERE job_id=?1",rusqlite::params![id,i32::from(superseded)]).unwrap(); + } + executor.inner.release(); + tokio::time::timeout(Duration::from_secs(2), async { + while worker.inflight() != 0 { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert_eq!(worker.inflight(), 0); + let job = worker.get_job(&id).unwrap().unwrap(); + let payload: serde_json::Value = serde_json::from_str(&job.payload).unwrap(); + assert_eq!(payload["late_intent"], 2); + if superseded { + assert_eq!(job.status, "running"); + assert!(crate::completion_result::(&job).is_none()); + } else { + assert_eq!(job.status, "completed"); + assert_eq!( + crate::completion_result::(&job).unwrap()["digest"], + "synthetic-content-proof" + ); + } + drop(worker); + drop(db); + remove_test_db(path); + } +} diff --git a/crates/libs/internal/api/server/src/lib.rs b/crates/libs/internal/api/server/src/lib.rs index e0f6daaa99..561a114f84 100644 --- a/crates/libs/internal/api/server/src/lib.rs +++ b/crates/libs/internal/api/server/src/lib.rs @@ -517,6 +517,43 @@ fn parse_artifact_digests( .collect() } +struct BackupFinalization<'a> { + snapshot_id: SnapshotId, + target_state: SnapshotState, + bytes_uploaded: i64, + failure_reason: Option<&'a str>, + artifact_digests: &'a [snapshot_store::SnapshotArtifactDigest], + upload_evidence: &'a [snapshot_store::upload_evidence::ArtifactUploadEvidence], + caller: &'a backup_ownership::Caller, +} + +fn parse_upload_evidence( + digests: &[crate::proto::SnapshotArtifactDigest], +) -> Vec { + digests + .iter() + .filter_map(|entry| { + let evidence = entry.upload_evidence.as_ref()?; + if evidence.object_key.is_empty() { + return None; + } + Some(snapshot_store::upload_evidence::ArtifactUploadEvidence { + artifact: entry.artifact.parse().ok()?, + object_key: evidence.object_key.clone(), + wire_size_bytes: evidence.wire_size_bytes, + etag: evidence.etag.clone(), + version_id: evidence.version_id.clone(), + metadata_binding_digest: evidence + .metadata_binding_digest + .as_deref() + .map(str::parse) + .transpose() + .ok()?, + }) + }) + .collect() +} + fn snapshot_insert_status(error: snapshot_store::InsertSnapshotError) -> Status { match error { snapshot_store::InsertSnapshotError::IdentityConflict { .. } @@ -678,15 +715,16 @@ impl ClusterApiService { /// /// Only the attempt's source node may close it; the check precedes the /// transaction so a refused caller writes nothing. - async fn finalize_backup_attempt( - &self, - snapshot_id: SnapshotId, - target_state: SnapshotState, - bytes_uploaded: i64, - failure_reason: Option<&str>, - artifact_digests: &[snapshot_store::SnapshotArtifactDigest], - caller: &backup_ownership::Caller, - ) -> Result<(), Status> { + async fn finalize_backup_attempt(&self, input: BackupFinalization<'_>) -> Result<(), Status> { + let BackupFinalization { + snapshot_id, + target_state, + bytes_uploaded, + failure_reason, + artifact_digests, + upload_evidence, + caller, + } = input; let attempt = self .registered_backup_attempt(snapshot_id) .await? @@ -799,6 +837,17 @@ impl ClusterApiService { })?; } + if target_state.is_ready() { + snapshot_store::upload_evidence::record(&mut tx, snapshot_id, upload_evidence) + .await + .map_err(|error| match error { + snapshot_store::RecordArtifactDigestsError::Database(_) => { + Status::internal(format!("upload evidence: {error}")) + } + _ => Status::failed_precondition(format!("upload evidence: {error}")), + })?; + } + let suspend_projection = if target_state.is_ready() { project_ready_suspend_snapshot(&mut tx, snapshot_id).await? } else { @@ -1116,14 +1165,15 @@ impl ClusterApiService { } }; let artifact_digests = parse_artifact_digests(&req.artifact_digests)?; - self.finalize_backup_attempt( - SnapshotId::from_uuid(backup_attempt_id.as_uuid()), + self.finalize_backup_attempt(BackupFinalization { + snapshot_id: SnapshotId::from_uuid(backup_attempt_id.as_uuid()), target_state, - req.bytes_uploaded, - req.failure_reason.as_deref(), - &artifact_digests, - &caller, - ) + bytes_uploaded: req.bytes_uploaded, + failure_reason: req.failure_reason.as_deref(), + artifact_digests: &artifact_digests, + upload_evidence: &parse_upload_evidence(&req.artifact_digests), + caller: &caller, + }) .await?; Ok(Response::new(FinalizeArtifactBackupResponse {})) } @@ -1425,6 +1475,7 @@ impl ClusterApiService { artifact: digest.artifact.file_name().to_string(), digest: digest.digest.to_string(), size_bytes: digest.size_bytes, + upload_evidence: None, }) .collect()) } diff --git a/crates/libs/internal/api/server/src/tests.rs b/crates/libs/internal/api/server/src/tests.rs index 8bba3a08b0..3f945178df 100644 --- a/crates/libs/internal/api/server/src/tests.rs +++ b/crates/libs/internal/api/server/src/tests.rs @@ -7306,11 +7306,13 @@ async fn ecw_route_resolve_enforces_key_surface_and_organization() -> TestResult fn test_artifact_digests() -> Vec { vec![ crate::proto::SnapshotArtifactDigest { + upload_evidence: None, artifact: "memfile".to_string(), digest: format!("sha256:{}", "22".repeat(32)), size_bytes: 1 << 30, }, crate::proto::SnapshotArtifactDigest { + upload_evidence: None, artifact: "binfile".to_string(), digest: format!("sha256:{}", "11".repeat(32)), size_bytes: 4096, @@ -7357,8 +7359,32 @@ async fn finalize_to_ready_publishes_artifact_digests_to_restore_lookups() -> Te pplx_space_sbx_mgr_client::proto::ArtifactOrigin::Suspend, ) .await?; - finalize_ready_with_digests(&fixture.service, backup_attempt_id, test_artifact_digests()) - .await?; + let mut uploaded = test_artifact_digests(); + uploaded[0].upload_evidence = Some(crate::proto::SnapshotUploadEvidence { + object_key: format!("synthetic/{sandbox_id}/{backup_attempt_id}/memfile"), + wire_size_bytes: 123, + etag: Some("synthetic-etag".into()), + version_id: None, + metadata_binding_digest: None, + }); + finalize_ready_with_digests(&fixture.service, backup_attempt_id, uploaded.clone()).await?; + let evidence = + snapshot_store::upload_evidence::get(&fixture.service.pool, backup_attempt_id).await?; + assert_eq!(evidence.len(), 1); + assert_eq!(evidence[0].etag.as_deref(), Some("synthetic-etag")); + finalize_ready_with_digests(&fixture.service, backup_attempt_id, uploaded.clone()).await?; + uploaded[0].upload_evidence.as_mut().unwrap().etag = Some("changed-object".into()); + assert_eq!( + finalize_ready_with_digests(&fixture.service, backup_attempt_id, uploaded) + .await + .unwrap_err() + .code(), + Code::FailedPrecondition + ); + assert_eq!( + snapshot_store::upload_evidence::get(&fixture.service.pool, backup_attempt_id).await?, + evidence + ); let mut expected = test_artifact_digests(); expected.sort_by(|a, b| a.artifact.cmp(&b.artifact)); diff --git a/crates/libs/sbx/mgr/server/src/snapshot_upload_jobs.rs b/crates/libs/sbx/mgr/server/src/snapshot_upload_jobs.rs index c26a1cee0a..350295cb20 100644 --- a/crates/libs/sbx/mgr/server/src/snapshot_upload_jobs.rs +++ b/crates/libs/sbx/mgr/server/src/snapshot_upload_jobs.rs @@ -316,7 +316,8 @@ impl JobExecutor for SnapshotUploadExecutor { .collect::>(); let vm_total_bytes = vm_artifact_total_bytes(&artifact_bytes, vm_artifact_count); let finalize_started_at = Instant::now(); - let finalize_result = finalize_with_cluster_api_ready(&snapshot_id, vm_total_bytes).await; + let finalize_result = + finalize_with_cluster_api_ready(&snapshot_id, vm_total_bytes, &artifact_results).await; pplx_space_metrics::snapshot::delivery_duration_ms( "cluster_finalize", origin, @@ -1041,8 +1042,9 @@ fn vm_artifact_total_bytes(polled_bytes: &[u64], vm_artifact_count: usize) -> u6 } /// Completed transfer bytes and persisted queue/transfer timing. -#[derive(Debug, Clone, Copy, PartialEq)] +#[derive(Debug, Clone, PartialEq)] struct ArtifactTransferObservation { + proofs: Vec, bytes: u64, queue_duration_ms: Option, transfer_duration_ms: Option, @@ -1140,6 +1142,7 @@ async fn poll_artifact_job( sqlite_duration_ms(started_at, completed_at) }); return Ok(ArtifactTransferObservation { + proofs: job.logical_artifact_digests, bytes: job.bytes_transferred, queue_duration_ms, transfer_duration_ms, @@ -1354,11 +1357,49 @@ fn begin_artifact_backup_request( } } +fn transfer_digests( + transfers: &[ArtifactTransferObservation], +) -> Vec { + let mut seen = std::collections::BTreeSet::new(); + let mut digests = Vec::new(); + for proof in transfers.iter().flat_map(|transfer| &transfer.proofs) { + let Ok(role) = pplx_space_volmgr_client::proto::SnapshotArtifact::try_from(proof.artifact) + else { + return Vec::new(); + }; + let Some(artifact) = pplx_space_volmgr_client::snapshot_artifact_from_proto(role) else { + return Vec::new(); + }; + if !seen.insert(artifact) + || proof + .digest + .parse::() + .is_err() + { + return Vec::new(); + } + digests.push(pplx_space_node_api_client::proto::SnapshotArtifactDigest { + artifact: artifact.file_name().into(), + digest: proof.digest.clone(), + size_bytes: proof.size_bytes, + upload_evidence: Some(pplx_space_node_api_client::proto::SnapshotUploadEvidence { + object_key: proof.object_key.clone(), + wire_size_bytes: proof.wire_size_bytes, + etag: proof.etag.clone(), + version_id: proof.version_id.clone(), + metadata_binding_digest: proof.metadata_binding_digest.clone(), + }), + }); + } + digests +} + /// Flip the backup row to `state='ready'`. This is the terminal step of the /// durable worker's `execute` and makes the row visible to restore. async fn finalize_with_cluster_api_ready( snapshot_id: &str, bytes_uploaded: u64, + transfers: &[ArtifactTransferObservation], ) -> Result<(), JobError> { let bytes = bytes_uploaded_i64(snapshot_id, bytes_uploaded); let mut client = connect_cluster_api().await?; @@ -1367,7 +1408,7 @@ async fn finalize_with_cluster_api_ready( target_state: pplx_space_node_api_client::proto::SuspendSnapshotTerminalState::Ready as i32, bytes_uploaded: bytes, failure_reason: None, - artifact_digests: Vec::new(), + artifact_digests: transfer_digests(transfers), }; match client.finalize_artifact_backup(req).await { diff --git a/crates/libs/sbx/mgr/server/src/snapshot_upload_jobs/tests.rs b/crates/libs/sbx/mgr/server/src/snapshot_upload_jobs/tests.rs index 8e48ec213f..6401b1c087 100644 --- a/crates/libs/sbx/mgr/server/src/snapshot_upload_jobs/tests.rs +++ b/crates/libs/sbx/mgr/server/src/snapshot_upload_jobs/tests.rs @@ -1259,3 +1259,41 @@ fn absent_volume_manager_socket() -> std::path::PathBuf { .join(format!("space-sbx-mgr-test-{}", uuid::Uuid::now_v7())) .join("absent-volume-manager.sock") } + +#[test] +fn transfer_evidence_reaches_finalization_without_inventing_legacy_proofs() { + let proof = pplx_space_volmgr_client::proto::LogicalArtifactDigest { + artifact: pplx_space_volmgr_client::proto::SnapshotArtifact::Memory as i32, + digest: format!("sha256:{}", "11".repeat(32)), + size_bytes: 100, + object_key: "synthetic/attempt/memfile".into(), + wire_size_bytes: 40, + etag: Some("synthetic-etag".into()), + version_id: None, + metadata_binding_digest: None, + }; + let mut observation = ArtifactTransferObservation { + proofs: vec![proof], + bytes: 100, + queue_duration_ms: None, + transfer_duration_ms: None, + }; + let forwarded = transfer_digests(&[observation.clone()]); + assert_eq!(forwarded[0].artifact, "memfile"); + assert_eq!(forwarded[0].size_bytes, 100); + assert_eq!( + forwarded[0] + .upload_evidence + .as_ref() + .unwrap() + .wire_size_bytes, + 40 + ); + observation.proofs.push(observation.proofs[0].clone()); + assert!( + transfer_digests(&[observation.clone()]).is_empty(), + "ambiguous roles are not proof" + ); + observation.proofs.clear(); + assert!(transfer_digests(&[observation]).is_empty()); +} diff --git a/crates/libs/snapshot/store/Cargo.toml b/crates/libs/snapshot/store/Cargo.toml index ecfbcecce1..9573c5ce70 100644 --- a/crates/libs/snapshot/store/Cargo.toml +++ b/crates/libs/snapshot/store/Cargo.toml @@ -12,6 +12,7 @@ workspace = true [dependencies] jiff = { workspace = true } +serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } sqlx = { workspace = true, features = ["derive", "json", "postgres", "runtime-tokio", "time", "tls-rustls-aws-lc-rs", "uuid"] } thiserror = { workspace = true } diff --git a/crates/libs/snapshot/store/src/lib.rs b/crates/libs/snapshot/store/src/lib.rs index b7d1ebf002..c392d3ea18 100644 --- a/crates/libs/snapshot/store/src/lib.rs +++ b/crates/libs/snapshot/store/src/lib.rs @@ -20,6 +20,7 @@ use time::OffsetDateTime; use uuid::Uuid; pub mod backup_authority; +pub mod upload_evidence; /// Reading a snapshot catalog row failed. #[derive(Debug, thiserror::Error)] diff --git a/crates/libs/snapshot/store/src/upload_evidence.rs b/crates/libs/snapshot/store/src/upload_evidence.rs new file mode 100644 index 0000000000..9aa85b6f0c --- /dev/null +++ b/crates/libs/snapshot/store/src/upload_evidence.rs @@ -0,0 +1,40 @@ +//! Optional upload evidence, committed alongside the write-once logical digests. +use super::*; + +#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize)] +pub struct ArtifactUploadEvidence { + pub artifact: SnapshotArtifactFile, + pub object_key: String, + pub wire_size_bytes: u64, + pub etag: Option, + pub version_id: Option, + pub metadata_binding_digest: Option, +} + +pub async fn record( + conn: &mut sqlx::PgConnection, + snapshot: SnapshotId, + evidence: &[ArtifactUploadEvidence], +) -> Result<(), RecordArtifactDigestsError> { + for entry in evidence { + let changed = sqlx::query(sql!("UPDATE snapshot_artifact_digests SET upload_evidence=$3 + WHERE snapshot_id=$1 AND artifact=$2 AND (upload_evidence IS NULL OR upload_evidence=$3)")) + .bind(snapshot).bind(entry.artifact.file_name()).bind(sqlx::types::Json(entry)).execute(&mut *conn).await?; + if changed.rows_affected() != 1 { + return Err(RecordArtifactDigestsError::Conflict { + snapshot_id: snapshot, + artifact: entry.artifact, + }); + } + } + Ok(()) +} + +pub async fn get( + executor: impl sqlx::PgExecutor<'_>, + snapshot: SnapshotId, +) -> Result, sqlx::Error> { + let rows: Vec> = sqlx::query_scalar(sql!("SELECT upload_evidence FROM snapshot_artifact_digests WHERE snapshot_id=$1 AND upload_evidence IS NOT NULL ORDER BY artifact")) + .bind(snapshot).fetch_all(executor).await?; + Ok(rows.into_iter().map(|entry| entry.0).collect()) +} diff --git a/crates/libs/vol/mgr/server/src/executor.rs b/crates/libs/vol/mgr/server/src/executor.rs index c80f74e835..77b62b30f7 100644 --- a/crates/libs/vol/mgr/server/src/executor.rs +++ b/crates/libs/vol/mgr/server/src/executor.rs @@ -1,5 +1,7 @@ pub(crate) mod copy; mod decode_admission; +pub(crate) mod logical_digest; +use logical_digest::{DigestReader, DigestSink, LogicalArtifactDigest}; mod stream_push; use decode_admission::{ CappedDecode, DecodeGate, DecodedOutputCap, copy_decode_capped, is_output_cap_exceeded, @@ -538,6 +540,11 @@ fn ensure_pull_failure_reason(error: JobError) -> JobError { error.with_default_failure_reason(JobFailureReason::Internal) } +struct TransformInput { + codec_permit: Option, + digest: Option, +} + /// Compress `src` to a sibling `.zst.tmp` file using zstd. /// Returns the temp path and the original (uncompressed) size. /// The caller is responsible for deleting the temp file. @@ -546,10 +553,7 @@ fn ensure_pull_failure_reason(error: JobError) -> JobError { /// task and released only when the CPU pass exits, so a cancelled caller /// cannot free the permit while the detached `spawn_blocking` keeps /// burning CPU. -async fn compress_to_temp( - src: &Path, - codec_permit: Option, -) -> Result<(PathBuf, u64), JobError> { +async fn compress_to_temp(src: &Path, input: TransformInput) -> Result<(PathBuf, u64), JobError> { let original_size = tokio::fs::metadata(src) .await .map(|m| m.len()) @@ -561,14 +565,15 @@ async fn compress_to_temp( let tmp = tmp_path.clone(); tokio::task::spawn_blocking(move || -> Result<(), JobError> { - let _codec_permit = codec_permit; - let mut reader = std::fs::File::open(&src) + let _codec_permit = input.codec_permit; + let reader = std::fs::File::open(&src) .map_err(|e| JobError::Permanent(format!("open source for compression: {e}")))?; + let mut reader = DigestReader::new(reader, input.digest); let mut writer = std::fs::File::create(&tmp) .map_err(|e| JobError::Permanent(format!("create compression temp file: {e}")))?; zstd::stream::copy_encode(&mut reader, &mut writer, ZSTD_LEVEL) .map_err(|e| JobError::Permanent(format!("zstd compress: {e}")))?; - page_cache::advise_dontneed(&reader, &src, "compress volume upload input"); + page_cache::advise_dontneed(reader.inner(), &src, "compress volume upload input"); page_cache::advise_dontneed(&writer, &tmp, "write compressed volume upload temp"); Ok(()) }) @@ -1024,6 +1029,7 @@ async fn apply_file_mode( /// `Clone` is cheap — `S3ObjectStore` is internally `Arc`-wrapped. #[derive(Clone)] pub struct TransferExecutor { + logical_digests: Option>>>, node_manager: Option, store: Arc, config: TransferExecutorConfig, @@ -1109,6 +1115,7 @@ impl TransferExecutor { }; let shadow_comparisons = ShadowComparisonController::new(); Self { + logical_digests: None, node_manager: None, stream_push: StreamPushConfig::new(&config.snapshot_upload_mode, total), store, @@ -1265,6 +1272,36 @@ impl JobExecutor for TransferExecutor { } } + async fn execute_with_completion( + &self, + job: &Job, + cancellation: CancellationToken, + ) -> Result { + let mut executor = self.clone(); + let digests = Arc::new(std::sync::Mutex::new(Vec::new())); + executor.logical_digests = Some(digests.clone()); + let bytes_processed = executor.execute(job, cancellation).await?; + let mut proofs = digests + .lock() + .map(|proofs| proofs.clone()) + .unwrap_or_default(); + proofs.sort_by_key(|entry| entry.artifact); + let ambiguous = proofs + .windows(2) + .any(|pair| matches!(pair, [a,b] if a.artifact == b.artifact)); + let result = if proofs.is_empty() || ambiguous { + None + } else { + serde_json::to_value(proofs) + .ok() + .map(|proofs| serde_json::json!({"logical_artifact_digests":proofs})) + }; + Ok(pplx_space_common_jobs::JobCompletion { + bytes_processed, + result, + }) + } + async fn cleanup_cancelled(&self, job: &Job) -> Result<(), JobError> { match job.kind.as_str() { // The object store owns multipart abort; copies create no local files. @@ -1854,7 +1891,7 @@ async fn stage_encrypted_temp( src_path: &Path, compress: bool, resolved_dek: Option<&[u8]>, - codec_permit: Option, + input: TransformInput, ) -> Result { let enc_path = encrypted_sibling(src_path); let src_path_buf = src_path.to_path_buf(); @@ -1891,11 +1928,13 @@ async fn stage_encrypted_temp( ); let job_id = job.job_id; tokio::task::spawn_blocking(move || -> Result<(), JobError> { - let _codec_permit = codec_permit; + let _codec_permit = input.codec_permit; let src_file = std::fs::File::open(&src_path_buf) .map_err(|e| JobError::Permanent(format!("open source for encrypt: {e}")))?; - let buf_reader = - std::io::BufReader::with_capacity(ENCRYPT_PUSH_READ_BUF_BYTES, src_file); + let buf_reader = std::io::BufReader::with_capacity( + ENCRYPT_PUSH_READ_BUF_BYTES, + DigestReader::new(src_file, input.digest), + ); let mut dst_file = std::fs::File::create(&enc_path_buf) .map_err(|e| JobError::Permanent(format!("create enc temp: {e}")))?; if compress { @@ -1929,6 +1968,12 @@ async fn stage_encrypted_temp( } } +struct PushReadContext<'a> { + resolved_dek: Option<&'a [u8]>, + cancellation: CancellationToken, + digest: Option, +} + struct PushExecutionOptions<'a> { cleanup_staged: bool, resolved_dek: Option<&'a [u8]>, @@ -1944,16 +1989,47 @@ impl TransferExecutor { target: ObjectTarget<'_>, options: PushExecutionOptions<'_>, ) -> Result { + let artifact = p.effective_snapshot_artifact(); + let digest = self + .logical_digests + .as_ref() + .and_then(|_| artifact.map(DigestSink::for_artifact)); let result = self .execute_push_inner( job, p, creds, target, - options.resolved_dek, - options.cancellation, + PushReadContext { + resolved_dek: options.resolved_dek, + cancellation: options.cancellation, + digest: digest.clone(), + }, ) .await; + if let (Ok(bytes), Some(sink), Some(artifact), Some(collection)) = + (&result, digest, artifact, &self.logical_digests) + { + // Optional evidence never changes a successful transfer into failure. + // Incomplete reads and size disagreements leave the transfer uncertified. + if let Ok((digest, size_bytes, metadata_binding_digest)) = sink.completed() + && i64::try_from(size_bytes).ok() == Some(*bytes) + && let Some(object) = sink.object() + && let Ok(mut proofs) = collection.lock() + { + proofs.push(LogicalArtifactDigest { + artifact, + digest, + size_bytes, + metadata_binding_digest, + object_key: object.key, + wire_size_bytes: object.size, + etag: object.etag, + version_id: None, + }); + } + } + // Single-file pushes clean their staged source here. Artifact entries // pass `cleanup_staged = false`: a per-entry success is NOT terminal // for the artifact (a sibling may force a retry that must re-read this @@ -1971,9 +2047,13 @@ impl TransferExecutor { p: &TransferPayload, creds: &pplx_space_object_store::ObjectCredentials, target: ObjectTarget<'_>, - resolved_dek: Option<&[u8]>, - cancellation: CancellationToken, + read: PushReadContext<'_>, ) -> Result { + let PushReadContext { + resolved_dek, + cancellation, + digest, + } = read; // Prefer the staged copy when set: bytes there are immune to a later // unlink of the caller's original path between submit and execute. // Legacy jobs queued before the staging field existed fall back to @@ -2034,6 +2114,7 @@ impl TransferExecutor { options, original_size, cancellation, + digest, }, ) .await; @@ -2052,6 +2133,7 @@ impl TransferExecutor { // Ineligible sources and stores without streaming support retain the // file-backed transform and upload path. let mut transient_temps: Vec = Vec::new(); + let mut raw_digest_temp = None; let put_source: PathBuf = match &p.encryption { None => { @@ -2070,7 +2152,14 @@ impl TransferExecutor { codec_wait_started_at, ); } - let (zst_path, _) = compress_to_temp(upload_path, codec_permit).await?; + let (zst_path, _) = compress_to_temp( + upload_path, + TransformInput { + codec_permit, + digest: digest.clone(), + }, + ) + .await?; tracing::debug!( job_id = %job.job_id, subject_id = %p.subject_id, @@ -2082,7 +2171,44 @@ impl TransferExecutor { transient_temps.push(zst_path.clone()); zst_path } else { - upload_path.to_path_buf() + if let Some(sink) = digest.clone() { + let source = upload_path.to_owned(); + let owned = p.staged_path.is_some(); + let staged = tokio::task::spawn_blocking(move || { + let file = std::fs::File::open(&source)?; + let mut reader = DigestReader::new(file, Some(sink)); + if owned { + // Staging owns an immutable inode: no second full copy. + std::io::copy(&mut reader, &mut std::io::sink())?; + Ok::<_, std::io::Error>(None) + } else { + let mut temp = + tempfile::NamedTempFile::new_in(source.parent().ok_or_else( + || std::io::Error::other("upload source has no parent"), + )?)?; + std::io::copy(&mut reader, &mut temp)?; + temp.as_file_mut().flush()?; + Ok(Some(temp)) + } + }) + .await; + match staged { + Ok(Ok(Some(temp))) => { + let path = temp.path().to_owned(); + raw_digest_temp = Some(temp); + path + } + Ok(Ok(None)) => upload_path.to_path_buf(), + _ => { + if let Some(sink) = &digest { + sink.invalidate(); + } + upload_path.to_path_buf() + } + } + } else { + upload_path.to_path_buf() + } } } Some(enc) => { @@ -2121,7 +2247,10 @@ impl TransferExecutor { upload_path, self.config.compress, resolved_dek, - codec_permit, + TransformInput { + codec_permit, + digest: digest.clone(), + }, ) .await { @@ -2187,6 +2316,9 @@ impl TransferExecutor { // to a corrupt artifact. Ok(_) => match self.store.head(creds, target).await { Ok(Some(durable)) if durable.size == wire_size => { + if let Some(sink) = &digest { + sink.record_object(target.key, &durable); + } push_payload( artifact, encoding, @@ -2225,6 +2357,7 @@ impl TransferExecutor { unlink_crypto_tmp(tmp, &job.job_id, &p.subject_id, "post_upload_cleanup").await; } + drop(raw_digest_temp); upload_result } diff --git a/crates/libs/vol/mgr/server/src/executor/logical_digest.rs b/crates/libs/vol/mgr/server/src/executor/logical_digest.rs new file mode 100644 index 0000000000..e76b37bff5 --- /dev/null +++ b/crates/libs/vol/mgr/server/src/executor/logical_digest.rs @@ -0,0 +1,140 @@ +//! Logical content proof from the same reads that feed compression/encryption. +//! A partial reader never produces a proof. Publication belongs to the job's +//! fenced completion transition, not to an individual producer finishing. + +use pplx_space_common_core::types::{Digest, SnapshotArtifact}; +use pplx_space_common_jobs::JobError; +use sha2::{Digest as _, Sha256}; +use std::io::{self, Read}; +use std::sync::{Arc, Mutex}; + +#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] +pub(crate) struct LogicalArtifactDigest { + pub artifact: SnapshotArtifact, + pub digest: Digest, + pub size_bytes: u64, + #[serde(default)] + pub metadata_binding_digest: Option, + pub object_key: String, + pub wire_size_bytes: u64, + pub etag: Option, + pub version_id: Option, +} + +#[derive(Clone)] +pub(super) struct VerifiedObject { + pub key: String, + pub size: u64, + pub etag: Option, +} + +type CompletedDigest = (Digest, u64, Option); + +#[derive(Clone, Default)] +pub(super) struct DigestSink { + result: Arc>>, + metadata: bool, + object: Arc>>, +} + +impl DigestSink { + pub fn record_object(&self, key: &str, meta: &pplx_space_object_store::ObjectMeta) { + if let Ok(mut object) = self.object.lock() { + *object = Some(VerifiedObject { + key: key.into(), + size: meta.size, + etag: meta.etag.clone(), + }); + } + } + pub fn object(&self) -> Option { + self.object.lock().ok()?.clone() + } + pub fn invalidate(&self) { + if let Ok(mut result) = self.result.lock() { + *result = None; + } + } + + pub fn for_artifact(artifact: SnapshotArtifact) -> Self { + Self { + metadata: artifact == SnapshotArtifact::Metadata, + ..Self::default() + } + } + pub fn completed(&self) -> Result<(Digest, u64, Option), JobError> { + self.result + .lock() + .map_err(|_| JobError::permanent("logical digest lock poisoned"))? + .as_ref() + .copied() + .ok_or_else(|| JobError::retriable("logical upload source was not read to completion")) + } +} + +pub(super) struct DigestReader { + inner: R, + sink: Option, + hash: Sha256, + bytes: u64, + metadata_bytes: Vec, +} + +impl DigestReader { + pub fn new(inner: R, sink: Option) -> Self { + Self { + inner, + sink, + hash: Sha256::new(), + bytes: 0, + metadata_bytes: Vec::new(), + } + } + pub fn inner(&self) -> &R { + &self.inner + } +} + +impl Read for DigestReader { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + let n = self.inner.read(buf)?; + if let Some(sink) = &self.sink { + if n != 0 { + let bytes = buf + .get(..n) + .ok_or_else(|| io::Error::other("reader returned an invalid byte count"))?; + self.hash.update(bytes); + // Bounded optional metadata proof: malformed/legacy/oversized + // metadata does not fail ordinary uploads, but grants require it. + if sink.metadata && self.metadata_bytes.len() <= 1024 * 1024 { + let remaining = (1024 * 1024 + 1) - self.metadata_bytes.len(); + self.metadata_bytes + .extend(bytes.iter().take(remaining).copied()); + } + self.bytes = self + .bytes + .checked_add(n as u64) + .ok_or_else(|| io::Error::other("logical upload size overflow"))?; + } else if !buf.is_empty() { + let digest = Digest::sha256(self.hash.clone().finalize().into()); + let binding = (sink.metadata && self.metadata_bytes.len() <= 1024 * 1024) + .then(|| { + pplx_space_common_core::types::snapshot_metadata_binding( + &self.metadata_bytes, + ) + .ok() + }) + .flatten() + .map(|bytes| Digest::sha256(Sha256::digest(bytes).into())); + if let Ok(mut result) = sink.result.lock() { + *result = Some((digest, self.bytes, binding)); + } + } + } + Ok(n) + } +} + +#[cfg(test)] +#[path = "logical_digest_tests.rs"] +mod tests; diff --git a/crates/libs/vol/mgr/server/src/executor/logical_digest_tests.rs b/crates/libs/vol/mgr/server/src/executor/logical_digest_tests.rs new file mode 100644 index 0000000000..759facc035 --- /dev/null +++ b/crates/libs/vol/mgr/server/src/executor/logical_digest_tests.rs @@ -0,0 +1,46 @@ +use super::*; +#[test] +fn proof_covers_consumed_bytes_and_requires_eof() { + let sink = DigestSink::default(); + let mut reader = DigestReader::new(io::Cursor::new(b"captured-state"), Some(sink.clone())); + assert_eq!(reader.read(&mut []).unwrap(), 0); + sink.completed().expect_err("incomplete input has no proof"); + let mut first = [0; 3]; + reader.read_exact(&mut first).unwrap(); + sink.completed().expect_err("incomplete input has no proof"); + let mut rest = Vec::new(); + reader.read_to_end(&mut rest).unwrap(); + let (digest, bytes, _) = sink.completed().unwrap(); + assert_eq!(bytes, 14); + assert_eq!( + digest, + Digest::sha256(Sha256::digest(b"captured-state").into()) + ); +} + +#[test] +fn metadata_proof_binds_stable_fields_but_ignores_regenerated_values() { + let base = serde_json::json!({"version":2,"sandbox_id":"sandbox","local_artifact_id":"artifact","artifact_kind":"full","artifact_origin":"suspend","execution":{"runtime":"fc"},"source":{"template_id":"seed","content_hash":"original"},"created_at":"old","size_bytes":1}); + let read = |value: &serde_json::Value| { + let sink = DigestSink::for_artifact(SnapshotArtifact::Metadata); + let mut reader = DigestReader::new( + io::Cursor::new(serde_json::to_vec(value).unwrap()), + Some(sink.clone()), + ); + io::copy(&mut reader, &mut io::sink()).unwrap(); + sink.completed().unwrap().2 + }; + let original = read(&base).expect("valid version 2 metadata"); + let mut regenerated = base.clone(); + regenerated["created_at"] = "later".into(); + regenerated["size_bytes"] = 123.into(); + regenerated["rootfs_send_source_id"] = "node-local-handle".into(); + assert_eq!(read(®enerated), Some(original)); + regenerated["source"]["content_hash"] = "different".into(); + assert_ne!(read(®enerated), Some(original)); + assert_eq!( + read(&serde_json::json!({"version":1})), + None, + "legacy metadata remains unknown" + ); +} diff --git a/crates/libs/vol/mgr/server/src/executor/stream_push.rs b/crates/libs/vol/mgr/server/src/executor/stream_push.rs index 1ff78aa773..d7622ddd33 100644 --- a/crates/libs/vol/mgr/server/src/executor/stream_push.rs +++ b/crates/libs/vol/mgr/server/src/executor/stream_push.rs @@ -128,6 +128,7 @@ pub(super) struct StreamPushAttempt<'a> { pub options: StreamingUploadOptions, pub original_size: u64, pub cancellation: CancellationToken, + pub digest: Option, } impl TransferExecutor { @@ -145,6 +146,7 @@ impl TransferExecutor { options, original_size, cancellation, + digest, } = attempt; let cancellation = cancellation.child_token(); let _cancel_on_drop = cancellation.clone().drop_guard(); @@ -176,7 +178,7 @@ impl TransferExecutor { .into_std() .await; let source = compression_source( - file, + DigestReader::new(file, digest.clone()), original_size, ProducerAdmission { codec, @@ -213,6 +215,9 @@ impl TransferExecutor { ); match self.store.head(creds, target).await { Ok(Some(meta)) if meta.size == wire_bytes => { + if let Some(sink) = &digest { + sink.record_object(target.key, &meta); + } stream_push_payload( artifact, StreamPushPhase::Verified, @@ -240,7 +245,7 @@ struct ProducerAdmission { } fn compression_source( - file: std::fs::File, + file: DigestReader, size: u64, admission: ProducerAdmission, cancellation: CancellationToken, @@ -286,7 +291,7 @@ fn compression_source( if result.is_ok() { // Preserve the file-backed compressor's final source-cache hint. page_cache::advise_dontneed( - &input.file, + input.file.inner(), Path::new("staged snapshot"), "stream upload input", ); @@ -310,7 +315,7 @@ fn compression_source( } struct SourceReader { - file: std::fs::File, + file: DigestReader, cancellation: CancellationToken, consumed: u64, advised_through: u64, @@ -330,7 +335,7 @@ impl Read for SourceReader { let through = self.consumed / ADVICE_WINDOW * ADVICE_WINDOW; if through > self.advised_through { if let Err(error) = page_cache::advise_range_dontneed( - &self.file, + self.file.inner(), self.advised_through, through - self.advised_through, ) { diff --git a/crates/libs/vol/mgr/server/src/executor/stream_push/tests.rs b/crates/libs/vol/mgr/server/src/executor/stream_push/tests.rs index 818c28b571..78f2529e80 100644 --- a/crates/libs/vol/mgr/server/src/executor/stream_push/tests.rs +++ b/crates/libs/vol/mgr/server/src/executor/stream_push/tests.rs @@ -294,9 +294,15 @@ async fn ordinary_and_luks_retry_from_staging_and_verify_actual_wire_size() { .unwrap(); source.write_all(&bytes[60 * MIB..68 * MIB]).unwrap(); drop(source); - let (temp, logical_size) = compress_to_temp(Path::new(payload.upload_source()), None) - .await - .unwrap(); + let (temp, logical_size) = compress_to_temp( + Path::new(payload.upload_source()), + TransformInput { + codec_permit: None, + digest: None, + }, + ) + .await + .unwrap(); assert_eq!(logical_size, bytes.len() as u64); let file_encoded = tokio::fs::read(&temp).await.unwrap(); tokio::fs::remove_file(temp).await.unwrap(); @@ -304,7 +310,14 @@ async fn ordinary_and_luks_retry_from_staging_and_verify_actual_wire_size() { let creates_before = server.recorded.lock().unwrap().creates; server.short_head.store(true, Ordering::SeqCst); server.retry_part.store(true, Ordering::SeqCst); - let first = push(&executor(), &server, &payload, CancellationToken::new()).await; + let mut proof_executor = executor(); + let collection = Arc::new(std::sync::Mutex::new(Vec::new())); + proof_executor.logical_digests = Some(collection.clone()); + let first = push(&proof_executor, &server, &payload, CancellationToken::new()).await; + assert!( + collection.lock().unwrap().is_empty(), + "failed object verification cannot publish a content proof" + ); assert!(matches!(first, Err(JobError::Retriable(_)))); assert!( Path::new(payload.upload_source()).exists(), @@ -320,15 +333,30 @@ async fn ordinary_and_luks_retry_from_staging_and_verify_actual_wire_size() { server.short_head.store(false, Ordering::SeqCst); let restored = TransferPayload::from_json(&persisted).unwrap(); assert_eq!( - push(&executor(), &server, &restored, CancellationToken::new()) - .await - .unwrap(), + push( + &proof_executor, + &server, + &restored, + CancellationToken::new() + ) + .await + .unwrap(), bytes.len() as i64 ); assert!( !Path::new(payload.upload_source()).exists(), "only verified terminal success cleans staging" ); + let proofs = collection.lock().unwrap(); + assert_eq!(proofs.len(), 1); + assert_eq!(proofs[0].artifact, artifact); + assert_eq!(proofs[0].size_bytes, bytes.len() as u64); + use sha2::Digest as _; + assert_eq!( + proofs[0].digest, + pplx_space_common_core::types::Digest::sha256(sha2::Sha256::digest(&bytes).into()) + ); + drop(proofs); let state = server.recorded.lock().unwrap(); assert_eq!( state.creates, @@ -370,6 +398,7 @@ async fn deterministic_stream_failures_are_permanent_and_join_the_producer() { &test_creds(), server.target(), StreamPushAttempt { + digest: None, uploader: executor.store.streaming_upload().unwrap(), config: executor.stream_push.as_ref().unwrap(), options: StreamingUploadOptions::new(5 * MIB, 2, max_bytes as u64).unwrap(), @@ -538,7 +567,7 @@ async fn producer_error_after_eof_never_completes_an_object() { std::fs::write(&path, synthetic_bytes(6 * MIB)).unwrap(); let executor = executor(); let source = compression_source( - std::fs::File::open(path).unwrap(), + DigestReader::new(std::fs::File::open(path).unwrap(), None), 7 * MIB as u64, ProducerAdmission { codec: executor @@ -620,8 +649,11 @@ async fn compatibility_paths_keep_file_uploads_and_small_sources_bypass_codec_qu &payload, &test_creds(), super::super::tests::test_target(), - None, - CancellationToken::new() + PushReadContext { + resolved_dek: None, + cancellation: CancellationToken::new(), + digest: None + }, ) .await .unwrap(), diff --git a/crates/libs/vol/mgr/server/src/executor/tests.rs b/crates/libs/vol/mgr/server/src/executor/tests.rs index 3ba58f7c2d..56f86651a4 100644 --- a/crates/libs/vol/mgr/server/src/executor/tests.rs +++ b/crates/libs/vol/mgr/server/src/executor/tests.rs @@ -88,6 +88,8 @@ fn common_object_parent_stays_within_one_artifact_directory() { assert_eq!(common_object_parent(divergent.into_iter()), None); } +type CapturedUploads = Arc)>>>; + #[derive(Default)] pub(super) struct MockObjectStore { heads: Mutex>>, @@ -95,6 +97,7 @@ pub(super) struct MockObjectStore { head_calls: AtomicUsize, get_calls: AtomicUsize, put_calls: AtomicUsize, + captured: Option, } impl MockObjectStore { @@ -108,6 +111,7 @@ impl MockObjectStore { head_calls: AtomicUsize::new(0), get_calls: AtomicUsize::new(0), put_calls: AtomicUsize::new(0), + captured: None, } } @@ -168,6 +172,10 @@ impl ObjectStore for MockObjectStore { // so a test can simulate a put that "succeeds" yet leaves a // truncated object behind. self.put_calls.fetch_add(1, Ordering::SeqCst); + if let Some(captured) = &self.captured { + let bytes = tokio::fs::read(src).await?; + captured.lock().unwrap().push((src.to_owned(), bytes)); + } let len = tokio::fs::metadata(src).await?.len(); Ok(len) } @@ -1123,8 +1131,11 @@ async fn push_rejects_truncated_durable_object() { &payload, &test_creds(), test_target(), - None, - CancellationToken::new(), + PushReadContext { + resolved_dek: None, + cancellation: CancellationToken::new(), + digest: None, + }, ) .await; @@ -1161,8 +1172,11 @@ async fn push_accepts_fully_durable_object() { &payload, &test_creds(), test_target(), - None, - CancellationToken::new(), + PushReadContext { + resolved_dek: None, + cancellation: CancellationToken::new(), + digest: None, + }, ) .await .expect("a fully durable object must finalize successfully"); @@ -4500,7 +4514,19 @@ async fn stage_encrypted_temp_returns_partial_enc_path_on_aad_derivation_error() // `compress = false` so the test exercises the AAD-derivation // failure before any zstd init; the early-Err path is the same // regardless of the compress branch. - let result = stage_encrypted_temp(&job, &p, &enc, &post_compress, false, None, None).await; + let result = stage_encrypted_temp( + &job, + &p, + &enc, + &post_compress, + false, + None, + TransformInput { + codec_permit: None, + digest: None, + }, + ) + .await; let (partial_enc_path, err) = result.expect_err("template subject must fail derive_artifact_aads"); assert_eq!( @@ -4916,3 +4942,230 @@ fn benchmark_production_shaped_sparse_accounting_with_metrics() { outstanding_metric_calls.load(Ordering::Relaxed), ); } + +#[tokio::test] +async fn logical_upload_proof_survives_raw_compressed_and_encrypted_transforms() { + use pplx_space_common_core::types::Digest as ContentDigest; + use sha2::{Digest as _, Sha256}; + let dir = tempfile::tempdir().unwrap(); + let source = dir.path().join("memory"); + let bytes = b"the exact retained guest memory"; + tokio::fs::write(&source, bytes).await.unwrap(); + let expected = ContentDigest::sha256(Sha256::digest(bytes).into()); + let payload = push_payload(&source); + let captured = Arc::new(Mutex::new(Vec::new())); + let mut mock = MockObjectStore::new( + [Some(ObjectMeta { + size: bytes.len() as u64, + etag: None, + })], + Vec::new(), + ); + mock.captured = Some(captured.clone()); + let store = Arc::new(mock); + let mut job = test_job("push"); + job.payload = payload.to_json(); + let completion = test_executor(store) + .execute_with_completion(&job, CancellationToken::new()) + .await + .unwrap(); + let body = completion.result.unwrap(); + let proofs: Vec = + serde_json::from_value(body["logical_artifact_digests"].clone()).unwrap(); + assert_eq!(captured.lock().unwrap()[0].1, bytes); + assert_ne!( + captured.lock().unwrap()[0].0, + source, + "legacy raw source must use an owned copy" + ); + assert_eq!(proofs.len(), 1); + assert_eq!(proofs[0].wire_size_bytes, bytes.len() as u64); + assert!(!proofs[0].object_key.is_empty()); + assert_eq!(proofs[0].digest, expected); + assert_eq!(proofs[0].size_bytes, bytes.len() as u64); + let sink = DigestSink::default(); + let (compressed, _) = compress_to_temp( + &source, + TransformInput { + codec_permit: None, + digest: Some(sink.clone()), + }, + ) + .await + .unwrap(); + assert_eq!( + zstd::decode_all(std::fs::File::open(compressed).unwrap()).unwrap(), + bytes + ); + assert_eq!(sink.completed().unwrap().0, expected); + let mut payload = payload; + payload.artifact_id = Some("019e9000-0000-7000-8000-000000000001".into()); + payload.entry_role = Some("memfile".into()); + let encryption = TransferEncryption { + organization_id: "019e31d9-79a6-7132-b90f-007f1c62ddea".parse().unwrap(), + provider: KekProvider::InMemory, + master_key_identifier: "test/key-1".parse().unwrap(), + wrapped_dek: vec![0; 16], + }; + for compress in [false, true] { + let sink = DigestSink::default(); + let encrypted = stage_encrypted_temp( + &job, + &payload, + &encryption, + &source, + compress, + Some(&[7; 32]), + TransformInput { + codec_permit: None, + digest: Some(sink.clone()), + }, + ) + .await + .unwrap(); + assert_ne!(std::fs::read(encrypted).unwrap(), bytes); + assert_eq!( + sink.completed().unwrap().0, + expected, + "logical hash precedes both transforms" + ); + assert_eq!(sink.completed().unwrap().1, bytes.len() as u64); + } +} + +#[tokio::test] +async fn owned_raw_upload_hashes_in_place_including_empty_files() { + for bytes in [b"".as_slice(), b"retained immutable bytes".as_slice()] { + let dir = tempfile::tempdir().unwrap(); + let staged = dir.path().join("staged"); + std::fs::write(&staged, bytes).unwrap(); + let captured = Arc::new(Mutex::new(Vec::new())); + let mut store = MockObjectStore::new( + [Some(ObjectMeta { + size: bytes.len() as u64, + etag: Some("synthetic-etag".into()), + })], + Vec::new(), + ); + store.captured = Some(captured.clone()); + let mut payload = push_payload(&dir.path().join("original-already-unlinked")); + payload.staged_path = Some(staged.to_string_lossy().into()); + let mut job = test_job("push"); + job.payload = payload.to_json(); + let completion = test_executor(Arc::new(store)) + .execute_with_completion(&job, CancellationToken::new()) + .await + .unwrap(); + let proofs: Vec = + serde_json::from_value(completion.result.unwrap()["logical_artifact_digests"].clone()) + .unwrap(); + assert_eq!(captured.lock().unwrap()[0], (staged, bytes.to_vec())); + assert_eq!( + proofs[0].digest, + pplx_space_common_core::types::Digest::sha256(sha2::Sha256::digest(bytes).into()) + ); + assert_eq!(proofs[0].size_bytes, bytes.len() as u64); + assert_eq!(proofs[0].etag.as_deref(), Some("synthetic-etag")); + } +} + +/// Opt-in local cost measurement; independent of CI's ignored-test selection. +#[tokio::test] +async fn measure_upload_evidence_cost() { + if std::env::var_os("SPACE_MEASURE_UPLOAD_EVIDENCE").is_none() { + return; + } + let dir = tempfile::tempdir().unwrap(); + let source = dir.path().join("logical-input"); + let mut bytes = vec![0u8; 16 * 1024 * 1024]; + let mut state = 17_u64; + for byte in &mut bytes { + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + *byte = state as u8; + } + std::fs::write(&source, &bytes).unwrap(); + let cpu = || { + let value = nix::time::clock_gettime(nix::time::ClockId::CLOCK_PROCESS_CPUTIME_ID).unwrap(); + u128::try_from(value.tv_sec()).unwrap() * 1_000_000_000 + + u128::try_from(value.tv_nsec()).unwrap() + }; + for mode in [ + "owned_raw", + "legacy_raw", + "compressed", + "encrypted_compressed", + ] { + for hashing in [false, true] { + let sink = hashing.then(DigestSink::default); + let started = Instant::now(); + let cpu_started = cpu(); + let staged_bytes = match mode { + "owned_raw" => { + let mut reader = + DigestReader::new(std::fs::File::open(&source).unwrap(), sink.clone()); + std::io::copy(&mut reader, &mut std::io::sink()).unwrap(); + 0 + } + "legacy_raw" => { + let mut reader = + DigestReader::new(std::fs::File::open(&source).unwrap(), sink.clone()); + let mut temp = tempfile::NamedTempFile::new_in(dir.path()).unwrap(); + std::io::copy(&mut reader, &mut temp).unwrap() + } + "compressed" => { + let (path, _) = compress_to_temp( + &source, + TransformInput { + codec_permit: None, + digest: sink.clone(), + }, + ) + .await + .unwrap(); + let size = std::fs::metadata(&path).unwrap().len(); + std::fs::remove_file(path).unwrap(); + size + } + _ => { + let mut payload = push_payload(&source); + payload.artifact_id = Some("019e9000-0000-7000-8000-000000000001".into()); + payload.entry_role = Some("memfile".into()); + let encryption = TransferEncryption { + organization_id: "019e9000-0000-7000-8000-000000000002".parse().unwrap(), + provider: KekProvider::InMemory, + master_key_identifier: "test/key-1".parse().unwrap(), + wrapped_dek: vec![0; 16], + }; + let path = stage_encrypted_temp( + &test_job("push"), + &payload, + &encryption, + &source, + true, + Some(&[7; 32]), + TransformInput { + codec_permit: None, + digest: sink.clone(), + }, + ) + .await + .unwrap(); + let size = std::fs::metadata(&path).unwrap().len(); + std::fs::remove_file(path).unwrap(); + size + } + }; + let cpu_ms = (cpu() - cpu_started) / 1_000_000; + if let Some(sink) = sink { + assert_eq!(sink.completed().unwrap().1, bytes.len() as u64); + } + println!( + "UPLOAD_EVIDENCE_COST mode={mode} hashing={hashing} logical_bytes={} transient_bytes={staged_bytes} wall_ms={} cpu_ms={cpu_ms}", + bytes.len(), + started.elapsed().as_millis() + ); + } + } +} diff --git a/crates/libs/vol/mgr/server/src/service.rs b/crates/libs/vol/mgr/server/src/service.rs index ca6a1f012c..6da4ad6aee 100644 --- a/crates/libs/vol/mgr/server/src/service.rs +++ b/crates/libs/vol/mgr/server/src/service.rs @@ -643,6 +643,35 @@ fn job_to_proto(job: pplx_space_common_jobs::Job) -> Result "cancelled" => TransferJobStatus::Cancelled, _ => TransferJobStatus::Unspecified, }; + let logical_artifact_digests = if status == TransferJobStatus::Completed { + #[derive(serde::Deserialize)] + struct Proofs { + logical_artifact_digests: Vec, + } + let proofs = pplx_space_common_jobs::completion_result::(&job) + .map(|result| result.logical_artifact_digests) + .unwrap_or_default(); + proofs + .into_iter() + .map( + |proof| pplx_space_volmgr_client::proto::LogicalArtifactDigest { + artifact: pplx_space_volmgr_client::snapshot_artifact_to_proto(proof.artifact) + as i32, + digest: proof.digest.to_string(), + size_bytes: proof.size_bytes, + object_key: proof.object_key, + wire_size_bytes: proof.wire_size_bytes, + etag: proof.etag, + version_id: proof.version_id, + metadata_binding_digest: proof + .metadata_binding_digest + .map(|digest| digest.to_string()), + }, + ) + .collect() + } else { + Vec::new() + }; Ok(TransferJob { job_id: job.job_id.to_string(), status: transfer_job_status_code(status), @@ -657,6 +686,7 @@ fn job_to_proto(job: pplx_space_common_jobs::Job) -> Result completed_at: job.completed_at, transfer_scope_id, attempts: u32::try_from(job.attempts).unwrap_or_default(), + logical_artifact_digests, }) } diff --git a/crates/libs/vol/mgr/server/src/service/tests.rs b/crates/libs/vol/mgr/server/src/service/tests.rs index 42d61ee64a..7a82c707cd 100644 --- a/crates/libs/vol/mgr/server/src/service/tests.rs +++ b/crates/libs/vol/mgr/server/src/service/tests.rs @@ -1720,3 +1720,35 @@ fn classify_artifact_retry_rejects_divergent_wrap_on_pull() { .is_some() ); } + +#[test] +fn completed_transfer_status_survives_unreadable_optional_proofs() { + let attempt = persisted_job("completed", &sample_artifact()).attempts; + let known = serde_json::json!({"completion_result":{"version":1,"attempt":attempt,"value":{"logical_artifact_digests":[{ + "artifact":"memory", "digest":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "size_bytes":7, + "object_key":"synthetic/attempt/memfile","wire_size_bytes":7,"etag":"synthetic-etag","version_id":null + }]}}}); + let mut stale = known.clone(); + stale["completion_result"]["attempt"] = (attempt - 1).into(); + let mut future = known.clone(); + future["completion_result"]["version"] = 2.into(); + for (payload, proofs) in [ + (known.to_string(), 1), + ("{}".into(), 0), + (known.to_string().replace("memory", "future_artifact"), 0), + (stale.to_string(), 0), + (future.to_string(), 0), + ("{malformed".into(), 0), + ] { + let mut job = persisted_job("completed", &sample_artifact()); + job.payload = payload; + job.bytes_processed = 7; + let response = job_to_proto(job).expect("optional proofs cannot hide durable completion"); + assert_eq!( + response.status, + transfer_job_status_code(TransferJobStatus::Completed) + ); + assert_eq!(response.bytes_transferred, 7); + assert_eq!(response.logical_artifact_digests.len(), proofs); + } +} diff --git a/docs/snapshot-upload-streaming.md b/docs/snapshot-upload-streaming.md index cf88bff3cf..3d973170c7 100644 --- a/docs/snapshot-upload-streaming.md +++ b/docs/snapshot-upload-streaming.md @@ -536,3 +536,36 @@ Before widening enablement, use a small canary and matched controls to verify: Broad enablement remains pending these measurements. Disabling new captures stops further accumulation; existing bundles and leases still follow their normal ownership and retirement rules. + +## Completion evidence + +File-backed snapshot uploads hash the logical reads that feed compression and +optional envelope encryption. A digest is published only after EOF, logical size +agreement, and successful verification of the uploaded object's wire size. +Already-owned immutable raw staging is hashed in place; legacy unowned raw +sources need a private copy to bind the bytes uploaded. Generated btrfs-send +sources do not acquire proof from an unrelated local file. + +The common job worker merges a versioned completion result into the latest +intent payload in the same attempt/lease-fenced SQLite completion write. +Consumers require its producing attempt to match the completed job. Missing, +unknown, or malformed optional evidence leaves ordinary completion readable. +Ambiguous artifact roles produce no certified evidence. Sandbox-manager forwards +logical digests to the catalog for restore verification; optional metadata and +object identity are stored alongside them. ETags are provider identity tokens, +not cryptographic hashes. The current object-store adapter exposes no version +ID, so that optional field remains absent. + +This establishes evidence on the authenticated, honest-worker path. It does not +independently attest object plaintext against a malicious uploader. Recovery +requiring exact evidence must reject absent or mismatching proofs and retain +its original source. + +An opt-in measurement is available with +`SPACE_MEASURE_UPLOAD_EVIDENCE=1 cargo test -p pplx-space-volmgr --lib measure_upload_evidence_cost -- --nocapture`. +On a local debug build using 16 MiB of synthetic incompressible input, SHA-256 +added roughly 56–84 ms of process CPU across raw, compressed, and encrypted +compressed paths. Owned raw input used no additional temporary disk; legacy raw +used one logical-size copy. Compression and encryption used their existing wire +staging, with no extra proof copy. These are local transform measurements, not +production throughput or network latency estimates. diff --git a/proto/node/api/v1/cluster.proto b/proto/node/api/v1/cluster.proto index d5b35f1d27..d172452fef 100644 --- a/proto/node/api/v1/cluster.proto +++ b/proto/node/api/v1/cluster.proto @@ -1121,6 +1121,16 @@ message SnapshotArtifactDigest { string digest = 2; // Size in bytes of the same bytes the digest covers. uint64 size_bytes = 3; + // Optional identity of the verified uploaded object; absent on older workers. + optional SnapshotUploadEvidence upload_evidence = 4; +} + +message SnapshotUploadEvidence { + string object_key = 1; + uint64 wire_size_bytes = 2; + optional string etag = 3; + optional string version_id = 4; + optional string metadata_binding_digest = 5; } message FinalizeArtifactBackupResponse {} diff --git a/proto/volume/manager/v1/volume_manager.proto b/proto/volume/manager/v1/volume_manager.proto index 5fb717c0bd..1ff24630ef 100644 --- a/proto/volume/manager/v1/volume_manager.proto +++ b/proto/volume/manager/v1/volume_manager.proto @@ -421,6 +421,19 @@ enum TransferJobStatus { TRANSFER_JOB_STATUS_CANCELLED = 6; } +// Proof over the exact source reads before transport transforms. Published +// atomically with COMPLETED; absent for old jobs and non-snapshot transfers. +message LogicalArtifactDigest { + SnapshotArtifact artifact = 1; + string digest = 2; // sha256:<64 lowercase hex> + uint64 size_bytes = 3; + optional string metadata_binding_digest = 4; + string object_key = 5; + uint64 wire_size_bytes = 6; + optional string etag = 7; + optional string version_id = 8; +} + message TransferJob { string job_id = 1; TransferJobStatus status = 2; @@ -435,10 +448,9 @@ message TransferJob { optional string started_at = 10; optional string completed_at = 11; optional string transfer_scope_id = 12; - // Executions this job has started, retries included. A completed job - // with more than one had a failed attempt; `error` is cleared on - // completion, `attempts` never is. - uint32 attempts = 13; + // Executions this job has started, including retries. + uint32 attempts = 13; + repeated LogicalArtifactDigest logical_artifact_digests = 14; } message GetTransferJobRequest {