From 8b68f8af24f01ca7dbf8ad5c1d2206adef17e362 Mon Sep 17 00:00:00 2001 From: coder 0 Date: Mon, 24 Aug 2026 18:58:11 -0400 Subject: [PATCH] Support community deletion in versioned media buckets Co-authored-by: Brad Seiler Signed-off-by: Brad Seiler --- Cargo.lock | 1 + crates/buzz-db/src/deletion.rs | 193 +++++- crates/buzz-deletion/src/lib.rs | 366 +++++++++-- crates/buzz-media/Cargo.toml | 1 + crates/buzz-media/src/lib.rs | 5 +- crates/buzz-media/src/storage.rs | 573 +++++++++++++++++- crates/buzz-media/tests/versioned_minio.rs | 363 +++++++++++ crates/buzz-relay/src/api/git/transport.rs | 98 +++ .../buzz-relay/src/handlers/side_effects.rs | 75 ++- deploy/charts/buzz/values.yaml | 9 + desktop/src-tauri/Cargo.lock | 1 + 11 files changed, 1584 insertions(+), 101 deletions(-) create mode 100644 crates/buzz-media/tests/versioned_minio.rs diff --git a/Cargo.lock b/Cargo.lock index 18c53c18ca0..9ff8ccf6185 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1136,6 +1136,7 @@ dependencies = [ "infer", "mp4", "nostr 0.44.7", + "quick-xml 0.38.4", "rust-s3", "serde", "serde_json", diff --git a/crates/buzz-db/src/deletion.rs b/crates/buzz-db/src/deletion.rs index fbe69f22a68..7db0df8ee37 100644 --- a/crates/buzz-db/src/deletion.rs +++ b/crates/buzz-db/src/deletion.rs @@ -310,11 +310,11 @@ pub struct StorageManifest { pub struct PrefixManifest { /// Exact community-scoped listing prefix. pub prefix: String, - /// Objects under the prefix at enumeration time. + /// Object versions and delete markers under the prefix at enumeration time. pub object_count: u64, - /// Total object bytes under the prefix at enumeration time. + /// Total object-version bytes under the prefix at enumeration time. pub total_bytes: u64, - /// Hex SHA-256 of the newline-terminated ascending key stream. + /// Hex SHA-256 of the newline-terminated ascending version-entry stream. pub keys_digest: String, } @@ -325,10 +325,88 @@ pub struct ManifestKeyChunk { pub chunk_no: i64, /// The tenant prefix every key in this chunk lives under. pub prefix: String, - /// Strictly ascending keys. + /// Strictly ascending serialized manifest entries. pub keys: Vec, } +/// One immutable object-store manifest entry. +/// +/// Version 5 storage manifests serialize entries as +/// `key\u{1f}version_id\u{1f}kind`, where kind is `object` or +/// `delete_marker`. Version 4 manifests used bare keys. Keeping the side-table +/// column name unchanged avoids a database migration while making the stream +/// explicitly version-aware. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct StorageManifestEntry { + /// Object key. + pub key: String, + /// S3 version id. + pub version_id: String, + /// Either `object` or `delete_marker`. + pub kind: String, +} + +impl StorageManifestEntry { + /// Create a manifest entry. + pub fn new( + key: impl Into, + version_id: impl Into, + kind: impl Into, + ) -> Self { + Self { + key: key.into(), + version_id: version_id.into(), + kind: kind.into(), + } + } + + /// Serialize this entry into the chunk stream. + pub fn encode(&self) -> Result { + validate_manifest_component("key", &self.key)?; + validate_manifest_component("version id", &self.version_id)?; + validate_manifest_component("kind", &self.kind)?; + if self.kind != "object" && self.kind != "delete_marker" { + return Err(DbError::DeletionSafety(format!( + "unsupported storage manifest entry kind {}", + self.kind + ))); + } + Ok(format!( + "{}\u{1f}{}\u{1f}{}", + self.key, self.version_id, self.kind + )) + } + + /// Decode a manifest stream entry. + pub fn decode(value: &str) -> Result { + let mut parts = value.split('\u{1f}'); + let key = parts.next().unwrap_or_default(); + let version_id = parts.next().ok_or_else(|| { + DbError::DeletionSafety("storage manifest entry is missing version id".to_string()) + })?; + let kind = parts.next().ok_or_else(|| { + DbError::DeletionSafety("storage manifest entry is missing kind".to_string()) + })?; + if parts.next().is_some() { + return Err(DbError::DeletionSafety( + "storage manifest entry has too many fields".to_string(), + )); + } + let entry = Self::new(key, version_id, kind); + entry.encode()?; + Ok(entry) + } +} + +fn validate_manifest_component(name: &str, value: &str) -> Result<()> { + if value.is_empty() || value.contains(['\n', '\u{1f}']) { + return Err(DbError::DeletionSafety(format!( + "storage manifest {name} is empty or contains a reserved delimiter" + ))); + } + Ok(()) +} + /// One durable fleet-wide object-store taxonomy sweep record. #[derive(Debug, Clone, Serialize)] pub struct TaxonomySweep { @@ -358,11 +436,11 @@ type TaxonomySweepRow = ( i64, ); -/// Streaming SHA-256 over a strictly ascending key stream. +/// Streaming SHA-256 over a strictly ascending storage manifest stream. /// /// The executor's prefix enumeration and the destructive freeze's chunk -/// validation both fold keys through this, so "the chunk rows are exactly -/// the frozen enumeration" reduces to digest equality. Each key is hashed +/// validation both fold entries through this, so "the chunk rows are exactly +/// the frozen enumeration" reduces to digest equality. Each entry is hashed /// with a trailing newline so concatenation cannot alias two streams. pub struct KeyStreamDigest { hasher: Sha256, @@ -395,6 +473,17 @@ impl KeyStreamDigest { "storage key stream is not strictly ascending at {key}" ))); } + self.fold_unordered(key) + } + + /// Fold an already-canonical manifest entry whose source ordering is owned + /// by the object store, not by key lexicographic order. + /// + /// S3 `ListObjectVersions` sorts by key but orders multiple versions of one + /// key by recency with opaque version ids, so version-aware manifests cannot + /// require strictly ascending serialized entries. Digest equality still + /// binds the exact stream that was listed and chunked. + pub fn fold_unordered(&mut self, key: &str) -> Result<()> { self.hasher.update(key.as_bytes()); self.hasher.update(b"\n"); self.last = Some(key.to_owned()); @@ -2594,7 +2683,7 @@ async fn live_fenced_tables_on(conn: &mut PgConnection) -> Result Result<()> { - if manifest.version != 4 { + if !matches!(manifest.version, 4 | 5) { return Err(DbError::DeletionSafety(format!( "unsupported storage manifest version {}", manifest.version @@ -2682,12 +2771,21 @@ fn validate_manifest_key_chunks( )); } for key in &keys.0 { - if !key.starts_with(chunk_prefix.as_str()) { + let prefix_key = if manifest.version >= 5 { + StorageManifestEntry::decode(key)?.key + } else { + key.clone() + }; + if !prefix_key.starts_with(chunk_prefix.as_str()) { return Err(DbError::DeletionSafety(format!( - "frozen key {key} is outside its chunk prefix {chunk_prefix}" + "frozen key {prefix_key} is outside its chunk prefix {chunk_prefix}" ))); } - digest.fold(key)?; + if manifest.version >= 5 { + digest.fold_unordered(key)?; + } else { + digest.fold(key)?; + } } } if let Some(summary) = current { @@ -3039,6 +3137,36 @@ mod tests { assert!(validate_storage_manifest(&malformed_digest).is_err()); } + #[test] + fn frozen_inventory_digest_is_canonical_for_v5_manifest_entries() { + let entry = StorageManifestEntry::new("_meta/c/a.json", "null", "object") + .encode() + .expect("entry"); + let mut digest = KeyStreamDigest::new(); + digest.fold_unordered(&entry).expect("fold entry"); + let (keys_digest, object_count) = digest.finish(); + let inventory = FrozenInventory { + schema: SchemaManifest { + scoped_tables: vec!["events".to_string()], + row_counts: BTreeMap::from([("events".to_string(), 1)]), + fenced_tables: vec!["events".to_string()], + }, + storage: StorageManifest { + version: 5, + prefixes: vec![PrefixManifest { + prefix: "_meta/c/".to_string(), + object_count, + total_bytes: 4, + keys_digest, + }], + }, + }; + let digest = inventory.digest().unwrap(); + let round_tripped: FrozenInventory = + serde_json::from_slice(&serde_json::to_vec(&inventory).unwrap()).unwrap(); + assert_eq!(digest, round_tripped.digest().unwrap()); + } + #[test] fn key_stream_digest_requires_strict_order_and_is_chunking_invariant() { let keys = ["a/1", "a/2", "a/3"]; @@ -3100,6 +3228,49 @@ mod tests { assert!(validate_manifest_key_chunks(&storage_manifest(), &[]).is_ok()); } + #[test] + fn versioned_manifest_entries_decode_and_validate_chunks() { + let entries = vec![ + StorageManifestEntry::new("_meta/c/1", "v2", "object") + .encode() + .expect("entry 1"), + StorageManifestEntry::new("_meta/c/1", "v1", "delete_marker") + .encode() + .expect("entry 2"), + ]; + let mut digest = KeyStreamDigest::new(); + for entry in &entries { + digest.fold_unordered(entry).expect("fold version entry"); + } + let (hex_digest, count) = digest.finish(); + let mut manifest = storage_manifest(); + manifest.version = 5; + manifest.prefixes[0].object_count = count; + manifest.prefixes[0].keys_digest = hex_digest; + + let chunk = |entries: &[String]| { + vec![( + 0, + "_meta/c/".to_string(), + sqlx::types::Json(entries.to_vec()), + )] + }; + // v5 freeze validation is retry-stable: a retried freeze with the + // same canonical version-entry stream is accepted, while a drifted + // stream is rejected. + assert!(validate_manifest_key_chunks(&manifest, &chunk(&entries)).is_ok()); + assert!(validate_manifest_key_chunks(&manifest, &chunk(&entries)).is_ok()); + + let foreign = vec![StorageManifestEntry::new("_uploads/c/1", "v1", "object") + .encode() + .expect("foreign entry")]; + assert!(validate_manifest_key_chunks(&manifest, &chunk(&foreign)).is_err()); + assert!(StorageManifestEntry::decode("_meta/c/1").is_err()); + assert!(StorageManifestEntry::new("_meta/c/1", "v1", "unknown") + .encode() + .is_err()); + } + #[test] fn frozen_inventory_digest_is_stable() { let inventory = FrozenInventory { diff --git a/crates/buzz-deletion/src/lib.rs b/crates/buzz-deletion/src/lib.rs index ae3dbe4f396..f13b7d507ac 100644 --- a/crates/buzz-deletion/src/lib.rs +++ b/crates/buzz-deletion/src/lib.rs @@ -2,6 +2,7 @@ #![warn(missing_docs)] //! Shared durable whole-community deletion engine and store adapters. +use std::future::Future; #[cfg(test)] use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; @@ -10,10 +11,14 @@ use std::time::Duration; use anyhow::{Context, Result}; use buzz_db::deletion::{ ClaimedDeletion, DeletionRequest, DeletionStage, DeletionStore, FrozenInventory, - KeyStreamDigest, LeaseToken, PrefixManifest, StorageManifest, DEFAULT_LEASE_DURATION, + KeyStreamDigest, LeaseToken, PrefixManifest, StorageManifest, StorageManifestEntry, + DEFAULT_LEASE_DURATION, }; use buzz_db::{Db, DbConfig}; -use buzz_media::{is_tenant_owned_key, tenant_prefixes, MediaStorage}; +use buzz_media::{ + is_tenant_owned_key, tenant_prefixes, BulkDeleteOutcome, MediaStorage, ObjectVersionKind, + ObjectVersionRef, +}; use clap::Subcommand; use serde::Serialize; use tokio_util::sync::CancellationToken; @@ -700,6 +705,26 @@ async fn flush_chunk(services: &Services, sink: &mut ChunkSink<'_>, prefix: &str Ok(()) } +fn manifest_kind_name(kind: ObjectVersionKind) -> &'static str { + match kind { + ObjectVersionKind::Object => "object", + ObjectVersionKind::DeleteMarker => "delete_marker", + } +} + +fn manifest_chunk_deleted_detail( + prefix: &str, + key_count: usize, + outcome: &buzz_media::BulkDeleteOutcome, +) -> serde_json::Value { + serde_json::json!({ + "prefix": prefix, + "keys": key_count, + "deleted": outcome.deleted, + "already_missing": outcome.already_missing, + }) +} + /// Enumerate the target's three tenant prefixes into per-prefix summaries. /// /// Cost is O(tenant objects) regardless of fleet size. Unknown shapes inside @@ -715,36 +740,44 @@ async fn enumerate_tenant_prefixes( heartbeat_lost: Option<&CancellationToken>, mut sink: Option<&mut ChunkSink<'_>>, ) -> Result { - if services.media.bucket_versioning_detected().await? { - return Err(permanent( - "bucket versioning detected; deletion cannot prove logical absence with delete markers", - )); - } let community = *request.community_id.as_uuid(); let chunk_keys = manifest_chunk_keys(); let mut prefixes = Vec::new(); for prefix in tenant_prefixes(community) { let mut digest = KeyStreamDigest::new(); let mut total_bytes: u64 = 0; - let mut continuation = None; + let mut key_marker = None; + let mut version_id_marker = None; loop { if heartbeat_lost.is_some_and(CancellationToken::is_cancelled) { return Err(DeletionLeaseLost.into()); } let page = services .media - .list_prefix_page(&prefix, continuation.take(), LIST_PAGE_SIZE) + .list_prefix_versions_page( + &prefix, + key_marker.take(), + version_id_marker.take(), + LIST_PAGE_SIZE, + ) .await?; - for (key, size) in page.objects { - if !is_tenant_owned_key(community, &key) { + for entry in page.entries { + if !is_tenant_owned_key(community, &entry.key) { return Err(permanent(format!( - "key under a tenant prefix is outside the exact writer taxonomy: {key}" + "key under a tenant prefix is outside the exact writer taxonomy: {}", + entry.key ))); } - digest.fold(&key)?; - total_bytes = total_bytes.saturating_add(size); + let encoded = StorageManifestEntry::new( + entry.key, + entry.version_id, + manifest_kind_name(entry.kind), + ) + .encode()?; + digest.fold_unordered(&encoded)?; + total_bytes = total_bytes.saturating_add(entry.size); if let Some(sink) = sink.as_deref_mut() { - sink.buffered.push(key); + sink.buffered.push(encoded); if sink.buffered.len() >= chunk_keys { flush_chunk(services, sink, &prefix).await?; } @@ -753,12 +786,12 @@ async fn enumerate_tenant_prefixes( if !page.is_truncated { break; } - continuation = page.next_continuation_token; - if continuation.is_none() { - return Err(transient( - "truncated tenant listing page has no continuation token", - )); - } + let (next_key_marker, next_version_id_marker) = require_truncated_version_markers( + page.next_key_marker, + page.next_version_id_marker, + )?; + key_marker = Some(next_key_marker); + version_id_marker = Some(next_version_id_marker); } if let Some(sink) = sink.as_deref_mut() { flush_chunk(services, sink, &prefix).await?; @@ -772,13 +805,113 @@ async fn enumerate_tenant_prefixes( }); } let manifest = StorageManifest { - version: 4, + version: 5, prefixes, }; buzz_db::deletion::validate_storage_manifest(&manifest)?; Ok(manifest) } +fn require_truncated_version_markers( + next_key_marker: Option, + next_version_id_marker: Option, +) -> Result<(String, String)> { + match (next_key_marker, next_version_id_marker) { + (Some(key_marker), Some(version_id_marker)) => Ok((key_marker, version_id_marker)), + (None, Some(_)) => Err(transient( + "truncated tenant version listing page has no key marker", + )), + (Some(_), None) => Err(transient( + "truncated tenant version listing page has no version id marker", + )), + (None, None) => Err(transient( + "truncated tenant version listing page has no key marker or version id marker", + )), + } +} + +async fn delete_manifest_chunk_with( + chunk: &buzz_db::deletion::ManifestKeyChunk, + storage_version: i32, + delete: F, +) -> Result +where + F: FnOnce(Vec) -> Fut, + Fut: Future>, +{ + let versions = object_versions_from_manifest_chunk(chunk, storage_version)?; + delete(versions).await +} + +fn object_versions_from_manifest_chunk( + chunk: &buzz_db::deletion::ManifestKeyChunk, + storage_version: i32, +) -> Result> { + if storage_version >= 5 { + chunk + .keys + .iter() + .map(|entry| { + let entry = StorageManifestEntry::decode(entry)?; + Ok(ObjectVersionRef { + key: entry.key, + version_id: entry.version_id, + }) + }) + .collect() + } else { + Ok(chunk + .keys + .iter() + .map(|key| ObjectVersionRef { + key: key.clone(), + version_id: String::new(), + }) + .collect()) + } +} + +fn manifest_chunk_deleted_checkpoint_detail( + chunk: &buzz_db::deletion::ManifestKeyChunk, + outcome: &BulkDeleteOutcome, +) -> Result { + validate_manifest_chunk_delete_outcome(chunk, outcome)?; + Ok(manifest_chunk_deleted_detail( + &chunk.prefix, + chunk.keys.len(), + outcome, + )) +} + +fn validate_manifest_chunk_delete_outcome( + chunk: &buzz_db::deletion::ManifestKeyChunk, + outcome: &BulkDeleteOutcome, +) -> Result<()> { + if !outcome.versioned_keys.is_empty() { + return Err(transient(format!( + "bulk delete returned version metadata for {} explicit versions: {}", + outcome.versioned_keys.len(), + outcome.versioned_keys.join(",") + ))); + } + if !outcome.failed.is_empty() { + let (key, code, message) = &outcome.failed[0]; + return Err(transient(format!( + "bulk delete failed for {} key(s); first: {key}: {code}: {message}", + outcome.failed.len() + ))); + } + let acknowledged = outcome.deleted.saturating_add(outcome.already_missing); + if acknowledged != chunk.keys.len() as u64 { + return Err(transient(format!( + "bulk delete acknowledged {acknowledged} of {} keys in chunk {}", + chunk.keys.len(), + chunk.chunk_no + ))); + } + Ok(()) +} + /// Freeze the post-fence, post-drain destructive enumeration: stream the /// tenant prefixes into side-table chunks, then bind the chunk stream to the /// request row's digests atomically. @@ -1071,6 +1204,35 @@ async fn execute_stage( } match request.stage { DeletionStage::Approved => { + // Fail closed on missing version-list permission before we take the + // durable write fence. Exact-version delete permission cannot be + // proven safely here: S3 has no dry-run DeleteObjectVersion, and a + // fabricated-version delete would still be a destructive API call + // while proving less than the real tenant-prefix operation. + run_guarded_external_step( + services, + &token, + DeletionStage::Approved, + heartbeat_lost, + || async { + for prefix in tenant_prefixes(*request.community_id.as_uuid()) { + services + .media + .preflight_version_listing(&prefix) + .await + .with_context(|| { + format!( + "S3 version-list preflight failed for prefix {prefix}; \ + verify s3:ListBucketVersions and s3:DeleteObjectVersion \ + on the relay bucket before fencing" + ) + })?; + } + Ok(()) + }, + ) + .await?; + // Approval binds immutable catalog + community-prefix ownership. // Live row counts and tenant binding keys are deliberately not // equality-bound until the durable fence closes all writers. @@ -1150,51 +1312,38 @@ async fn execute_stage( let mut removed: u64 = 0; let mut already_missing: u64 = 0; while let Some(chunk) = services.store.next_pending_manifest_chunk(&token).await? { + let chunk_no = chunk.chunk_no; let outcome = run_guarded_external_step( services, &token, DeletionStage::Drained, heartbeat_lost, - || async { Ok(services.media.delete_objects(&chunk.keys).await?) }, + || async { + delete_manifest_chunk_with(&chunk, storage.version, |versions| async { + if storage.version >= 5 { + Ok(services.media.delete_object_versions(&versions).await?) + } else { + let keys = versions + .into_iter() + .map(|version| version.key) + .collect::>(); + Ok(services.media.delete_objects(&keys).await?) + } + }) + .await + }, ) .await?; - if !outcome.versioned_keys.is_empty() { - return Err(permanent(format!( - "bulk delete produced version artifacts; bucket versioning blocks \ - deletion: {}", - outcome.versioned_keys.join(",") - ))); - } - if !outcome.failed.is_empty() { - let (key, code, message) = &outcome.failed[0]; - return Err(transient(format!( - "bulk delete failed for {} key(s); first: {key}: {code}: {message}", - outcome.failed.len() - ))); + if heartbeat_lost.is_cancelled() { + return Err(DeletionLeaseLost.into()); } - let acknowledged = outcome.deleted.saturating_add(outcome.already_missing); - if acknowledged != chunk.keys.len() as u64 { - return Err(transient(format!( - "bulk delete acknowledged {acknowledged} of {} keys in chunk {}", - chunk.keys.len(), - chunk.chunk_no - ))); - } - removed += outcome.deleted; - already_missing += outcome.already_missing; + let detail = manifest_chunk_deleted_checkpoint_detail(&chunk, &outcome)?; services .store - .mark_manifest_chunk_deleted( - &token, - chunk.chunk_no, - serde_json::json!({ - "prefix": chunk.prefix, - "keys": chunk.keys.len(), - "deleted": outcome.deleted, - "already_missing": outcome.already_missing, - }), - ) + .mark_manifest_chunk_deleted(&token, chunk_no, detail) .await?; + removed += outcome.deleted; + already_missing += outcome.already_missing; } let frozen_keys: u64 = storage .prefixes @@ -1277,10 +1426,16 @@ fn token_with_current_fence(token: &LeaseToken, request: &DeletionRequest) -> Le /// empty — O(1) requests per prefix, independent of fleet size. async fn verify_storage_absence(services: &Services, request: &DeletionRequest) -> Result<()> { for prefix in tenant_prefixes(*request.community_id.as_uuid()) { - let page = services.media.list_prefix_page(&prefix, None, 1).await?; - if let Some((key, _)) = page.objects.first() { + let page = services + .media + .list_prefix_versions_page(&prefix, None, None, 1) + .await?; + if let Some(entry) = page.entries.first() { return Err(transient(format!( - "logical verification found a live target object binding: {key}" + "logical verification found a retained target object version: {}@{} ({})", + entry.key, + entry.version_id, + manifest_kind_name(entry.kind) ))); } } @@ -1821,6 +1976,99 @@ mod tests { } } + #[test] + fn truncated_version_listing_requires_key_marker() { + let error = require_truncated_version_markers(None, Some("v1".to_string())) + .expect_err("missing key marker must fail closed"); + + assert!(format!("{error:#}").contains("no key marker")); + } + + #[test] + fn truncated_version_listing_requires_version_id_marker() { + let error = require_truncated_version_markers(Some("key".to_string()), None) + .expect_err("missing version id marker must fail closed"); + + assert!(format!("{error:#}").contains("no version id marker")); + } + + #[test] + fn legacy_v4_manifest_chunk_decodes_bare_keys_for_resume_delete() { + let chunk = buzz_db::deletion::ManifestKeyChunk { + chunk_no: 3, + prefix: "_meta/community/".to_string(), + keys: vec![ + "_meta/community/a.json".to_string(), + "_meta/community/b.json".to_string(), + ], + }; + + let versions = object_versions_from_manifest_chunk(&chunk, 4).expect("decode v4 chunk"); + assert_eq!( + versions, + vec![ + ObjectVersionRef { + key: "_meta/community/a.json".to_string(), + version_id: String::new(), + }, + ObjectVersionRef { + key: "_meta/community/b.json".to_string(), + version_id: String::new(), + }, + ] + ); + } + + #[tokio::test] + async fn partial_delete_ack_fails_before_checkpoint_detail() { + let chunk = buzz_db::deletion::ManifestKeyChunk { + chunk_no: 7, + prefix: "_meta/community/".to_string(), + keys: vec![ + StorageManifestEntry::new("_meta/community/a.json", "v1", "object") + .encode() + .expect("encode manifest entry"), + StorageManifestEntry::new("_meta/community/b.json", "v2", "object") + .encode() + .expect("encode manifest entry"), + ], + }; + let delete = delete_manifest_chunk_with(&chunk, 5, |versions| async move { + assert_eq!(versions.len(), 2); + Ok(BulkDeleteOutcome { + deleted: 1, + already_missing: 0, + versioned_keys: Vec::new(), + failed: Vec::new(), + }) + }) + .await + .expect("delete call returns partial acknowledgement"); + let checkpoint = manifest_chunk_deleted_checkpoint_detail(&chunk, &delete); + + let error = checkpoint.expect_err("partial acknowledgement must be transient"); + assert!(format!("{error:#}").contains("bulk delete acknowledged 1 of 2 keys in chunk 7")); + } + + #[test] + fn manifest_chunk_checkpoint_detail_records_partial_delete_response_counts() { + let detail = manifest_chunk_deleted_detail( + "_meta/community/", + 3, + &buzz_media::BulkDeleteOutcome { + deleted: 2, + already_missing: 1, + versioned_keys: Vec::new(), + failed: Vec::new(), + }, + ); + + assert_eq!(detail["prefix"], "_meta/community/"); + assert_eq!(detail["keys"], 3); + assert_eq!(detail["deleted"], 2); + assert_eq!(detail["already_missing"], 1); + } + #[test] fn permanent_failures_are_typed_not_string_classified() { let permanent_error = permanent("catalog drift"); diff --git a/crates/buzz-media/Cargo.toml b/crates/buzz-media/Cargo.toml index 530ce69c90a..7808ecaff43 100644 --- a/crates/buzz-media/Cargo.toml +++ b/crates/buzz-media/Cargo.toml @@ -32,6 +32,7 @@ tempfile = "3" tokio-util = { version = "0.7", features = ["io"] } futures-util = "0.3" futures-core = "0.3" +quick-xml = { version = "0.38", features = ["serialize"] } [dev-dependencies] tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/buzz-media/src/lib.rs b/crates/buzz-media/src/lib.rs index b2ff12c16e9..3198e1f8301 100644 --- a/crates/buzz-media/src/lib.rs +++ b/crates/buzz-media/src/lib.rs @@ -20,7 +20,10 @@ pub use bucket_index::{ }; pub use config::{MediaConfig, S3AddressingStyle}; pub use error::MediaError; -pub use storage::{BlobHeadMeta, BlobMeta, BulkDeleteOutcome, ByteStream, MediaStorage}; +pub use storage::{ + BlobHeadMeta, BlobMeta, BulkDeleteOutcome, ByteStream, MediaStorage, ObjectVersionEntry, + ObjectVersionKind, ObjectVersionRef, ObjectVersionsPage, +}; pub use types::BlobDescriptor; pub use upload::{process_file_upload, process_upload, process_video_upload}; pub use upload_record::{ diff --git a/crates/buzz-media/src/storage.rs b/crates/buzz-media/src/storage.rs index 0f0aa7af623..abe6bdd40ea 100644 --- a/crates/buzz-media/src/storage.rs +++ b/crates/buzz-media/src/storage.rs @@ -1,5 +1,6 @@ //! S3/MinIO storage client. +use std::collections::HashMap; use std::path::Path; use std::pin::Pin; @@ -8,13 +9,198 @@ use buzz_core::tenant::{CommunityId, TenantContext}; use crate::config::{MediaConfig, S3AddressingStyle}; use crate::error::MediaError; use bytes::Bytes; +use quick_xml::events::{BytesStart, Event}; +use quick_xml::Reader; use s3::creds::Credentials; +use s3::request::Request as _; use s3::{Bucket, Region}; use serde::{Deserialize, Serialize}; /// A stream of byte chunks from S3, usable with `axum::body::Body::from_stream()`. pub type ByteStream = Pin> + Send>>; +/// The kind of versioned S3 object-store entry. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ObjectVersionKind { + /// A concrete object version with bytes. + Object, + /// A delete-marker version hiding older bytes from live-object listing. + DeleteMarker, +} + +/// One S3 object version or delete marker under a tenant prefix. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ObjectVersionEntry { + /// Object key. + pub key: String, + /// Concrete S3 version id. + pub version_id: String, + /// Whether this entry is a byte-bearing object or delete marker. + pub kind: ObjectVersionKind, + /// Byte size for object versions; zero for delete markers. + pub size: u64, +} + +/// Exact version identifier used for permanent deletion. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ObjectVersionRef { + /// Object key. + pub key: String, + /// Concrete S3 version id. + pub version_id: String, +} + +/// One `ListObjectVersions` page. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ObjectVersionsPage { + /// Object versions and delete markers returned by this page. + pub entries: Vec, + /// Next key marker for truncated listings. + pub next_key_marker: Option, + /// Next version-id marker for truncated listings. + pub next_version_id_marker: Option, + /// Whether more pages remain. + pub is_truncated: bool, +} + +#[derive(Debug, Default)] +struct ListVersionFields { + key: Option, + version_id: Option, + size: Option, +} + +fn local_name(name: &[u8]) -> &[u8] { + name.rsplit(|byte| *byte == b':').next().unwrap_or(name) +} + +fn xml_error(error: impl std::fmt::Display) -> MediaError { + MediaError::StorageError(error.to_string()) +} + +fn read_element_text( + reader: &mut Reader<&[u8]>, + start: &BytesStart<'_>, +) -> Result { + reader + .read_text(start.to_end().name()) + .map(|text| text.into_owned()) + .map_err(xml_error) +} + +fn skip_element(reader: &mut Reader<&[u8]>, start: &BytesStart<'_>) -> Result<(), MediaError> { + reader + .read_to_end(start.to_end().name()) + .map_err(xml_error)?; + Ok(()) +} + +fn parse_list_version_entry( + reader: &mut Reader<&[u8]>, + start: &BytesStart<'_>, + kind: ObjectVersionKind, +) -> Result { + let mut fields = ListVersionFields::default(); + loop { + match reader.read_event().map_err(xml_error)? { + Event::Start(child) => match local_name(child.local_name().as_ref()) { + b"Key" => fields.key = Some(read_element_text(reader, &child)?), + b"VersionId" => fields.version_id = Some(read_element_text(reader, &child)?), + b"Size" => { + let size = read_element_text(reader, &child)?; + fields.size = Some(size.parse::().map_err(xml_error)?); + } + _ => skip_element(reader, &child)?, + }, + Event::Empty(child) => match local_name(child.local_name().as_ref()) { + b"Key" => fields.key = Some(String::new()), + b"VersionId" => fields.version_id = Some(String::new()), + b"Size" => fields.size = Some(0), + _ => {} + }, + Event::End(end) if end.name().as_ref() == start.to_end().name().as_ref() => { + let key = fields.key.ok_or_else(|| { + MediaError::StorageError("ListObjectVersions entry missing Key".to_string()) + })?; + let version_id = fields.version_id.ok_or_else(|| { + MediaError::StorageError( + "ListObjectVersions entry missing VersionId".to_string(), + ) + })?; + return Ok(ObjectVersionEntry { + key, + version_id, + kind, + size: if kind == ObjectVersionKind::Object { + fields.size.unwrap_or(0) + } else { + 0 + }, + }); + } + Event::Eof => { + return Err(MediaError::StorageError( + "unexpected EOF inside ListObjectVersions entry".to_string(), + )); + } + _ => {} + } + } +} + +fn parse_object_versions_page(xml: &[u8]) -> Result { + let mut reader = Reader::from_reader(xml); + reader.config_mut().trim_text(true); + let mut entries = Vec::new(); + let mut next_key_marker = None; + let mut next_version_id_marker = None; + let mut is_truncated = false; + + loop { + match reader.read_event().map_err(xml_error)? { + Event::Start(start) => match local_name(start.local_name().as_ref()) { + b"Version" => entries.push(parse_list_version_entry( + &mut reader, + &start, + ObjectVersionKind::Object, + )?), + b"DeleteMarker" => entries.push(parse_list_version_entry( + &mut reader, + &start, + ObjectVersionKind::DeleteMarker, + )?), + b"IsTruncated" => { + let value = read_element_text(&mut reader, &start)?; + is_truncated = value.eq_ignore_ascii_case("true"); + } + b"NextKeyMarker" => { + next_key_marker = Some(read_element_text(&mut reader, &start)?); + } + b"NextVersionIdMarker" => { + next_version_id_marker = Some(read_element_text(&mut reader, &start)?); + } + b"ListVersionsResult" => {} + _ => skip_element(&mut reader, &start)?, + }, + Event::Empty(start) => match local_name(start.local_name().as_ref()) { + b"NextKeyMarker" => next_key_marker = Some(String::new()), + b"NextVersionIdMarker" => next_version_id_marker = Some(String::new()), + _ => {} + }, + Event::Eof => break, + _ => {} + } + } + + Ok(ObjectVersionsPage { + entries, + next_key_marker, + next_version_id_marker, + is_truncated, + }) +} + /// S3-compatible object storage client. pub struct MediaStorage { bucket: Box, @@ -177,24 +363,6 @@ impl MediaStorage { } } - /// Detect whether the bucket has ever had versioning enabled. - /// - /// rust-s3 exposes no GetBucketVersioning, so this writes and inspects a - /// short-lived fleet probe object instead: versioning-enabled (and - /// versioning-suspended) buckets stamp new writes with a version id. - /// Deletion refuses versioned buckets because bulk deletes without a - /// VersionId would only insert delete markers, not prove logical absence. - pub async fn bucket_versioning_detected(&self) -> Result { - let key = format!("probe/deletion-versioning-{}", uuid::Uuid::new_v4()); - self.put(&key, b"buzz deletion versioning probe", "text/plain") - .await?; - let inspected = self.bucket.head_object(&key).await; - let removed = self.bucket.delete_object(&key).await; - let (head, _) = inspected.map_err(|e| MediaError::StorageError(e.to_string()))?; - removed.map_err(|e| MediaError::StorageError(e.to_string()))?; - Ok(head.version_id.is_some()) - } - /// Bulk-delete up to one manifest chunk of keys via S3 `DeleteObjects`. /// /// Never fails on per-key outcomes: they are folded into @@ -210,6 +378,57 @@ impl MediaStorage { .iter() .map(|key| s3::serde_types::ObjectIdentifier::new(key.clone())) .collect::>(); + self.delete_object_identifiers(identifiers).await + } + + /// Non-destructively verify that versioned bucket APIs are reachable. + /// + /// `ListObjectVersions` can be proven without mutation. S3 has no equivalent + /// dry-run for `DeleteObjectVersion`: `DeleteObjects` is always destructive, + /// even for exact versions, and deleting a fabricated version id does not + /// prove permission when policies can be prefix- or tag-constrained. + /// Operators must still provision `s3:DeleteObjectVersion`; the first exact + /// version deletion remains the destructive proof. + pub async fn preflight_version_listing(&self, prefix: &str) -> Result<(), MediaError> { + self.list_prefix_versions_page(prefix, None, None, 1) + .await + .map(|_| ()) + } + + /// Bulk-delete exact object versions via S3 `DeleteObjects`. + /// + /// Every identifier includes a version id, so this removes historical + /// versions and delete markers permanently instead of adding another + /// delete marker to a versioned bucket. + pub async fn delete_object_versions( + &self, + versions: &[ObjectVersionRef], + ) -> Result { + self.delete_object_versions_with_folding(versions, fold_version_delete_result) + .await + } + + async fn delete_object_versions_with_folding( + &self, + versions: &[ObjectVersionRef], + fold: fn(s3::serde_types::DeleteObjectsResult) -> BulkDeleteOutcome, + ) -> Result { + if versions.is_empty() { + return Ok(BulkDeleteOutcome::default()); + } + let identifiers = object_version_identifiers(versions); + let result = self + .bucket + .delete_objects(identifiers) + .await + .map_err(|e| MediaError::StorageError(e.to_string()))?; + Ok(fold(result)) + } + + async fn delete_object_identifiers( + &self, + identifiers: Vec, + ) -> Result { let result = self .bucket .delete_objects(identifiers) @@ -330,6 +549,57 @@ impl MediaStorage { is_truncated: result.is_truncated, }) } + + /// One page of object versions and delete markers under a prefix. + /// + /// This uses S3 `ListObjectVersions` (`?versions`) instead of + /// `ListObjectsV2`: versioned buckets can be logically empty while still + /// retaining historical versions or delete markers, and permanent deletion + /// must enumerate both. Pagination must carry both `KeyMarker` and + /// `VersionIdMarker`; carrying only the key marker can skip siblings when a + /// key has multiple versions on a page boundary. + pub async fn list_prefix_versions_page( + &self, + prefix: &str, + key_marker: Option, + version_id_marker: Option, + max_keys: usize, + ) -> Result { + let mut query = HashMap::from([ + ("versions".to_string(), String::new()), + ("prefix".to_string(), prefix.to_string()), + ("max-keys".to_string(), max_keys.to_string()), + ]); + if let Some(marker) = key_marker { + query.insert("key-marker".to_string(), marker); + } + if let Some(marker) = version_id_marker { + query.insert("version-id-marker".to_string(), marker); + } + let bucket = self + .bucket + .with_extra_query(query) + .map_err(|e| MediaError::StorageError(e.to_string()))?; + let request = s3::request::tokio_backend::ReqwestRequest::new( + &bucket, + "/", + s3::command::Command::GetObject, + ) + .await + .map_err(|e| MediaError::StorageError(e.to_string()))?; + let response = request + .response_data(false) + .await + .map_err(|e| MediaError::StorageError(e.to_string()))?; + if response.status_code() >= 300 { + return Err(MediaError::StorageError(format!( + "list object versions failed with status {}: {}", + response.status_code(), + response.as_str().unwrap_or("") + ))); + } + parse_object_versions_page(response.as_slice()) + } } /// Per-key outcomes of one bulk `DeleteObjects` call. @@ -349,16 +619,49 @@ pub struct BulkDeleteOutcome { pub failed: Vec<(String, String, String)>, } +fn object_version_identifiers( + versions: &[ObjectVersionRef], +) -> Vec { + versions + .iter() + .map(|version| { + s3::serde_types::ObjectIdentifier::with_version( + version.key.clone(), + version.version_id.clone(), + ) + }) + .collect() +} + fn fold_bulk_delete_result(result: s3::serde_types::DeleteObjectsResult) -> BulkDeleteOutcome { + fold_delete_result(result, DeleteMode::Unversioned) +} + +fn fold_version_delete_result(result: s3::serde_types::DeleteObjectsResult) -> BulkDeleteOutcome { + fold_delete_result(result, DeleteMode::ExplicitVersion) +} + +enum DeleteMode { + Unversioned, + ExplicitVersion, +} + +fn fold_delete_result( + result: s3::serde_types::DeleteObjectsResult, + mode: DeleteMode, +) -> BulkDeleteOutcome { let mut outcome = BulkDeleteOutcome::default(); for deleted in result.deleted { - if deleted.delete_marker == Some(true) + let has_version_artifact = deleted.delete_marker == Some(true) || deleted.delete_marker_version_id.is_some() - || deleted.version_id.is_some() - { - outcome.versioned_keys.push(deleted.key); - } else { - outcome.deleted += 1; + || deleted.version_id.is_some(); + match mode { + DeleteMode::Unversioned if has_version_artifact => { + outcome.versioned_keys.push(deleted.key); + } + DeleteMode::Unversioned | DeleteMode::ExplicitVersion => { + outcome.deleted += 1; + } } } for error in result.errors { @@ -419,6 +722,228 @@ mod tests { ); } + #[test] + fn version_delete_fold_counts_explicit_version_artifacts_as_deleted() { + use s3::serde_types::{DeleteError, DeleteObjectsResult, DeletedObject}; + let result = DeleteObjectsResult { + deleted: vec![DeletedObject { + key: "versioned".to_string(), + version_id: Some("v1".to_string()), + delete_marker: Some(true), + delete_marker_version_id: Some("v1".to_string()), + }], + errors: vec![ + DeleteError { + key: "retried-version".to_string(), + code: "NoSuchVersion".to_string(), + message: "already absent".to_string(), + version_id: Some("v-gone".to_string()), + }, + DeleteError { + key: "denied-version".to_string(), + code: "AccessDenied".to_string(), + message: "denied".to_string(), + version_id: Some("v-denied".to_string()), + }, + ], + }; + + let outcome = fold_version_delete_result(result); + assert_eq!(outcome.deleted, 1); + assert_eq!(outcome.already_missing, 1); + assert!(outcome.versioned_keys.is_empty()); + assert_eq!( + outcome.failed, + vec![( + "denied-version".to_string(), + "AccessDenied".to_string(), + "denied".to_string() + )] + ); + } + + #[test] + fn object_version_identifiers_include_explicit_version_ids() { + let identifiers = object_version_identifiers(&[ + ObjectVersionRef { + key: "_meta/tenant/a.json".to_string(), + version_id: "v-object".to_string(), + }, + ObjectVersionRef { + key: "uploads/tenant/event/blob".to_string(), + version_id: "v-delete-marker".to_string(), + }, + ]); + + assert_eq!(identifiers.len(), 2); + assert_eq!(identifiers[0].key, "_meta/tenant/a.json"); + assert_eq!(identifiers[0].version_id.as_deref(), Some("v-object")); + assert_eq!(identifiers[1].key, "uploads/tenant/event/blob"); + assert_eq!( + identifiers[1].version_id.as_deref(), + Some("v-delete-marker") + ); + } + + #[tokio::test] + async fn delete_object_versions_empty_input_short_circuits_before_folding() { + let storage = MediaStorage::new(&storage_config("buzz_dev", "buzz_dev_secret")) + .expect("static client"); + let outcome = storage + .delete_object_versions_with_folding(&[], |_| BulkDeleteOutcome { + deleted: 0, + already_missing: 0, + versioned_keys: vec!["wrong-fold".to_string()], + failed: Vec::new(), + }) + .await + .expect("empty delete short-circuits before fold"); + assert_eq!(outcome, BulkDeleteOutcome::default()); + } + + #[test] + fn parse_object_versions_page_includes_objects_delete_markers_and_dual_markers() { + let page = parse_object_versions_page( + br#" + + buzz-media + _meta/tenant/ + _meta/tenant/a.json + v-old + 2 + true + _meta/tenant/a.json + v-new + + _meta/tenant/a.json + v-delete + true + + + _meta/tenant/a.json + v-new + false + 42 + +"#, + ) + .expect("parse versions page"); + + assert!(page.is_truncated); + assert_eq!(page.next_key_marker.as_deref(), Some("_meta/tenant/a.json")); + assert_eq!(page.next_version_id_marker.as_deref(), Some("v-new")); + assert_eq!( + page.entries, + vec![ + ObjectVersionEntry { + key: "_meta/tenant/a.json".to_string(), + version_id: "v-delete".to_string(), + kind: ObjectVersionKind::DeleteMarker, + size: 0, + }, + ObjectVersionEntry { + key: "_meta/tenant/a.json".to_string(), + version_id: "v-new".to_string(), + kind: ObjectVersionKind::Object, + size: 42, + }, + ] + ); + } + + #[test] + fn parse_object_versions_page_preserves_repeated_interleaved_aws_ordering() { + let page = parse_object_versions_page( + br#" + k-av33 + k-av2 + k-av11 + k-bm2 + k-bm110 +"#, + ) + .expect("parse interleaved versions page"); + + assert_eq!( + page.entries, + vec![ + ObjectVersionEntry { + key: "k-a".to_string(), + version_id: "v3".to_string(), + kind: ObjectVersionKind::Object, + size: 3, + }, + ObjectVersionEntry { + key: "k-a".to_string(), + version_id: "v2".to_string(), + kind: ObjectVersionKind::DeleteMarker, + size: 0, + }, + ObjectVersionEntry { + key: "k-a".to_string(), + version_id: "v1".to_string(), + kind: ObjectVersionKind::Object, + size: 1, + }, + ObjectVersionEntry { + key: "k-b".to_string(), + version_id: "m2".to_string(), + kind: ObjectVersionKind::DeleteMarker, + size: 0, + }, + ObjectVersionEntry { + key: "k-b".to_string(), + version_id: "m1".to_string(), + kind: ObjectVersionKind::Object, + size: 10, + }, + ] + ); + } + + #[test] + fn parse_object_versions_page_handles_marker_only_key_before_versioned_key() { + let page = parse_object_versions_page( + br#" + k-marker-onlyd-only + k-versionedv220 + k-versionedd1 + k-versionedv110 +"#, + ) + .expect("parse marker-only and versioned keys"); + + assert_eq!( + page.entries, + vec![ + ObjectVersionEntry { + key: "k-marker-only".to_string(), + version_id: "d-only".to_string(), + kind: ObjectVersionKind::DeleteMarker, + size: 0, + }, + ObjectVersionEntry { + key: "k-versioned".to_string(), + version_id: "v2".to_string(), + kind: ObjectVersionKind::Object, + size: 20, + }, + ObjectVersionEntry { + key: "k-versioned".to_string(), + version_id: "d1".to_string(), + kind: ObjectVersionKind::DeleteMarker, + size: 0, + }, + ObjectVersionEntry { + key: "k-versioned".to_string(), + version_id: "v1".to_string(), + kind: ObjectVersionKind::Object, + size: 10, + }, + ] + ); + } + fn tenant(n: u128) -> TenantContext { TenantContext::resolved( CommunityId::from_uuid(uuid::Uuid::from_u128(n)), diff --git a/crates/buzz-media/tests/versioned_minio.rs b/crates/buzz-media/tests/versioned_minio.rs new file mode 100644 index 00000000000..1e0db4b481f --- /dev/null +++ b/crates/buzz-media/tests/versioned_minio.rs @@ -0,0 +1,363 @@ +//! Live destructive versioned-bucket deletion coverage against docker-compose MinIO. +//! +//! This exercises the S3-compatible path that community deletion relies on when +//! a bucket has versioning enabled: list object versions/delete markers with +//! dual markers, delete exact `(Key, VersionId)` identifiers, retry an already +//! deleted version, and prove final `ListObjectVersions` emptiness. +//! +//! Run it against the docker-compose MinIO (creds `buzz_dev`/`buzz_dev_secret`): +//! +//! ```bash +//! docker compose up -d minio minio-init +//! cargo test -p buzz-media --test versioned_minio -- --ignored --nocapture +//! ``` +//! +//! The test creates and removes its own bucket. The MinIO container name is +//! overridable with `BUZZ_MINIO_CONTAINER`; credentials/endpoint/region/addressing +//! use the same `BUZZ_S3_*` env vars as `static_creds_minio`. + +use std::process::Command; + +use buzz_media::config::MediaConfig; +use buzz_media::storage::{MediaStorage, ObjectVersionKind, ObjectVersionRef}; + +fn env_or(name: &str, default: &str) -> String { + std::env::var(name).unwrap_or_else(|_| default.to_string()) +} + +fn minio_config(bucket: String) -> MediaConfig { + MediaConfig { + s3_endpoint: env_or("BUZZ_S3_ENDPOINT", "http://localhost:9000"), + s3_access_key: env_or("BUZZ_S3_ACCESS_KEY", "buzz_dev"), + s3_secret_key: env_or("BUZZ_S3_SECRET_KEY", "buzz_dev_secret"), + s3_bucket: bucket, + s3_region: env_or("BUZZ_S3_REGION", "us-east-1"), + s3_addressing_style: env_or("BUZZ_S3_ADDRESSING_STYLE", "path") + .parse() + .expect("BUZZ_S3_ADDRESSING_STYLE must be path or virtual"), + max_image_bytes: 50 * 1024 * 1024, + max_gif_bytes: 10 * 1024 * 1024, + max_video_bytes: 524_288_000, + max_file_bytes: 104_857_600, + public_base_url: "http://localhost:3000/media".to_string(), + upload_records_enabled: false, + upload_ip_header: None, + upload_port_header: None, + } +} + +fn run_mc(args: &[String]) -> Result<(), String> { + let container = env_or("BUZZ_MINIO_CONTAINER", "buzz-minio"); + let output = Command::new("docker") + .arg("exec") + .arg(container) + .arg("mc") + .args(args) + .output() + .map_err(|err| format!("failed to execute docker/mc: {err}"))?; + if output.status.success() { + Ok(()) + } else { + Err(format!( + "mc {:?} failed with status {}\nstdout:\n{}\nstderr:\n{}", + args, + output.status, + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + )) + } +} + +fn mc_alias(access_key: &str, secret_key: &str) -> Result<(), String> { + run_mc(&[ + "alias".to_string(), + "set".to_string(), + "local".to_string(), + "http://localhost:9000".to_string(), + access_key.to_string(), + secret_key.to_string(), + ]) +} + +async fn list_all_versions( + storage: &MediaStorage, + prefix: &str, + max_keys: usize, +) -> Vec { + let mut entries = Vec::new(); + let mut key_marker = None; + let mut version_id_marker = None; + loop { + let page = storage + .list_prefix_versions_page( + prefix, + key_marker.take(), + version_id_marker.take(), + max_keys, + ) + .await + .expect("list object versions page"); + if page.is_truncated { + assert!( + page.next_key_marker.is_some(), + "truncated ListObjectVersions page must include NextKeyMarker" + ); + assert!( + page.next_version_id_marker.is_some(), + "truncated ListObjectVersions page must include NextVersionIdMarker" + ); + } + entries.extend(page.entries); + if !page.is_truncated { + break; + } + key_marker = page.next_key_marker; + version_id_marker = page.next_version_id_marker; + } + entries +} + +fn refs_from(entries: &[buzz_media::storage::ObjectVersionEntry]) -> Vec { + entries + .iter() + .map(|entry| ObjectVersionRef { + key: entry.key.clone(), + version_id: entry.version_id.clone(), + }) + .collect() +} + +#[tokio::test] +#[ignore = "requires live docker-compose MinIO; permanently deletes exact test object versions"] +async fn never_versioned_bucket_lists_null_versions_and_exact_delete_empties_listing() { + let bucket = format!("buzz-media-never-versioned-{}", std::process::id()); + let bucket_path = format!("local/{bucket}"); + let config = minio_config(bucket.clone()); + mc_alias(&config.s3_access_key, &config.s3_secret_key).expect("configure mc alias"); + run_mc(&[ + "mb".to_string(), + "--ignore-existing".to_string(), + bucket_path.clone(), + ]) + .expect("create isolated never-versioned test bucket"); + + let storage = MediaStorage::new(&config).expect("static MinIO storage client"); + let prefix = format!("_test/never-versioned-{}/", uuid::Uuid::new_v4()); + let key = format!("{prefix}plain.bin"); + storage + .put(&key, b"plain", "application/octet-stream") + .await + .expect("put never-versioned object"); + + let listed = list_all_versions(&storage, &prefix, 2).await; + assert_eq!( + listed, + vec![buzz_media::storage::ObjectVersionEntry { + key: key.clone(), + version_id: "null".to_string(), + kind: ObjectVersionKind::Object, + size: 5, + }], + "never-versioned buckets must still enumerate exact null-version objects" + ); + + let delete = storage + .delete_object_versions(&refs_from(&listed)) + .await + .expect("delete exact null-version object"); + assert!(delete.failed.is_empty(), "{delete:?}"); + assert!(delete.versioned_keys.is_empty(), "{delete:?}"); + assert_eq!( + delete.deleted + delete.already_missing, + 1, + "exact null-version delete must account for the listed object" + ); + assert!( + list_all_versions(&storage, &prefix, 2).await.is_empty(), + "final ListObjectVersions must be empty after deleting the exact null version" + ); + + run_mc(&["rb".to_string(), "--force".to_string(), bucket_path.clone()]) + .expect("remove isolated never-versioned test bucket"); +} + +#[tokio::test] +#[ignore = "requires live docker-compose MinIO; permanently deletes exact test object versions"] +async fn versioned_bucket_exact_version_delete_reaches_final_list_versions_emptiness() { + let bucket = format!("buzz-media-versioned-{}", std::process::id()); + let bucket_path = format!("local/{bucket}"); + let config = minio_config(bucket.clone()); + mc_alias(&config.s3_access_key, &config.s3_secret_key).expect("configure mc alias"); + run_mc(&[ + "mb".to_string(), + "--ignore-existing".to_string(), + bucket_path.clone(), + ]) + .expect("create isolated versioned test bucket"); + run_mc(&[ + "version".to_string(), + "enable".to_string(), + bucket_path.clone(), + ]) + .expect("enable bucket versioning"); + + let storage = MediaStorage::new(&config).expect("static MinIO storage client"); + let prefix = format!("_test/versioned-{}/", uuid::Uuid::new_v4()); + let historical_key = format!("{prefix}historical.bin"); + let marker_only_key = format!("{prefix}marker-only.bin"); + let paginated_key = format!("{prefix}paginated.bin"); + + storage + .put(&historical_key, b"v1", "application/octet-stream") + .await + .expect("put historical v1"); + storage + .put(&historical_key, b"v2", "application/octet-stream") + .await + .expect("put historical v2"); + storage + .delete(&historical_key) + .await + .expect("delete historical current version creates delete marker"); + storage + .put(&marker_only_key, b"marker-base", "application/octet-stream") + .await + .expect("put marker-only base version"); + storage + .delete(&marker_only_key) + .await + .expect("delete current version creates marker-only delete marker"); + let marker_versions = list_all_versions(&storage, &marker_only_key, 2).await; + let marker_objects: Vec = marker_versions + .iter() + .filter(|entry| entry.kind == ObjectVersionKind::Object) + .map(|entry| ObjectVersionRef { + key: entry.key.clone(), + version_id: entry.version_id.clone(), + }) + .collect(); + assert_eq!( + marker_objects.len(), + 1, + "marker-only setup should have one object version: {marker_versions:?}" + ); + let marker_object_delete = storage + .delete_object_versions(&marker_objects) + .await + .expect("delete marker-only base object version"); + assert!( + marker_object_delete.failed.is_empty(), + "{marker_object_delete:?}" + ); + assert!( + marker_object_delete.versioned_keys.is_empty(), + "{marker_object_delete:?}" + ); + storage + .put(&paginated_key, b"page-a", "application/octet-stream") + .await + .expect("put paginated v1"); + storage + .put(&paginated_key, b"page-b", "application/octet-stream") + .await + .expect("put paginated v2"); + + let listed = list_all_versions(&storage, &prefix, 2).await; + assert!( + listed.len() >= 6, + "expected multiple versions/delete markers across small pages, got {listed:?}" + ); + assert!(listed.iter().any(|entry| { + entry.key == historical_key && entry.kind == ObjectVersionKind::Object && entry.size == 2 + })); + assert!(listed.iter().any(|entry| { + entry.key == historical_key && entry.kind == ObjectVersionKind::DeleteMarker + })); + assert!(listed.iter().any(|entry| { + entry.key == marker_only_key && entry.kind == ObjectVersionKind::DeleteMarker + })); + + let refs = refs_from(&listed); + let first_chunk = &refs[..2.min(refs.len())]; + let first_delete = storage + .delete_object_versions(first_chunk) + .await + .expect("delete first explicit version chunk"); + assert!(first_delete.failed.is_empty(), "{first_delete:?}"); + assert!(first_delete.versioned_keys.is_empty(), "{first_delete:?}"); + assert_eq!( + first_delete.deleted + first_delete.already_missing, + first_chunk.len() as u64, + "explicit version delete should account for every requested identifier" + ); + + let retry = storage + .delete_object_versions(&first_chunk[..1]) + .await + .expect("retry already-deleted explicit version"); + assert!(retry.failed.is_empty(), "{retry:?}"); + assert!(retry.versioned_keys.is_empty(), "{retry:?}"); + assert_eq!( + retry.deleted + retry.already_missing, + 1, + "retry should be idempotently accounted as deleted/already missing" + ); + + let rest_delete = storage + .delete_object_versions(&refs[first_chunk.len()..]) + .await + .expect("delete remaining explicit versions"); + assert!(rest_delete.failed.is_empty(), "{rest_delete:?}"); + assert!(rest_delete.versioned_keys.is_empty(), "{rest_delete:?}"); + assert_eq!( + rest_delete.deleted + rest_delete.already_missing, + (refs.len() - first_chunk.len()) as u64, + "remaining explicit version delete should account for every requested identifier" + ); + + let remaining = list_all_versions(&storage, &prefix, 2).await; + assert!( + remaining.is_empty(), + "final ListObjectVersions must be empty after exact-version deletion: {remaining:?}" + ); + + if run_mc(&[ + "version".to_string(), + "suspend".to_string(), + bucket_path.clone(), + ]) + .is_ok() + { + let suspended_key = format!("{prefix}suspended.bin"); + storage + .put(&suspended_key, b"suspended", "application/octet-stream") + .await + .expect("put suspended-versioning object"); + storage + .delete(&suspended_key) + .await + .expect("delete suspended-versioning object"); + let suspended_entries = list_all_versions(&storage, &prefix, 2).await; + assert!( + suspended_entries + .iter() + .any(|entry| entry.key == suspended_key), + "suspended-versioning write/delete should be visible to ListObjectVersions" + ); + let suspended_delete = storage + .delete_object_versions(&refs_from(&suspended_entries)) + .await + .expect("delete suspended-versioning entries by explicit version id"); + assert!(suspended_delete.failed.is_empty(), "{suspended_delete:?}"); + assert!( + suspended_delete.versioned_keys.is_empty(), + "{suspended_delete:?}" + ); + assert!(list_all_versions(&storage, &prefix, 2).await.is_empty()); + } else { + eprintln!("MinIO mc did not support version suspend; enabled-versioning coverage passed"); + } + + run_mc(&["rb".to_string(), "--force".to_string(), bucket_path.clone()]) + .expect("remove isolated versioned test bucket"); +} diff --git a/crates/buzz-relay/src/api/git/transport.rs b/crates/buzz-relay/src/api/git/transport.rs index 3b2241046a3..ec7af3aac65 100644 --- a/crates/buzz-relay/src/api/git/transport.rs +++ b/crates/buzz-relay/src/api/git/transport.rs @@ -2424,6 +2424,104 @@ mod track_c_tests { } } + #[tokio::test] + #[ignore = "requires Postgres and MinIO"] + async fn repo_announcement_holds_serving_lease_until_pointer_is_seeded() { + let (state, pool) = finalize_test_state().await; + let host = format!( + "git-announce-lease-{}.example", + uuid::Uuid::new_v4().simple() + ); + let community = state + .db + .ensure_configured_community(&host) + .await + .expect("create test community") + .id; + let (request, claim) = approved_deletion(&state, &host).await; + let tenant = TenantContext::resolved(community, host.clone()); + let owner_keys = Keys::generate(); + let repo = format!("repo-{}", uuid::Uuid::new_v4().simple()); + let event = EventBuilder::new(Kind::Custom(30_617), "") + .tags([Tag::parse(["d", &repo]).expect("d tag")]) + .sign_with_keys(&owner_keys) + .expect("sign announcement"); + let gate = Arc::new(crate::handlers::side_effects::GitRepoAnnouncementGate::default()); + let hooks = crate::handlers::side_effects::GitRepoAnnouncementHooks { + post_lease_gate: Some(Arc::clone(&gate)), + }; + let announce_state = Arc::clone(&state); + let announce_tenant = tenant.clone(); + let announce = tokio::spawn(async move { + let result = crate::handlers::side_effects::handle_git_repo_announcement_inner( + &announce_tenant, + &event, + &announce_state, + &hooks, + ) + .await; + let owner_hex = hex::encode(owner_keys.public_key().to_bytes()); + (result, owner_hex) + }); + + gate.reached.notified().await; + state + .db + .deletion_store() + .begin_quiescing(&claim.lease) + .await + .expect("quiesce after announcement lease"); + let error = state + .db + .deletion_store() + .fence(&claim.lease) + .await + .expect_err("announcement serving lease must block fence"); + assert!(matches!( + error, + buzz_db::DbError::ServingWritesNotDrained { .. } + )); + + gate.resume.notify_one(); + let (announce_result, owner_hex) = announce.await.expect("announcement task"); + announce_result.expect("announcement completes"); + let pointer_key = crate::api::git::manifest::pointer_key(community, &owner_hex, &repo); + assert!( + state + .git_store + .get_pointer(&pointer_key) + .await + .expect("read pointer") + .is_some(), + "announcement pointer must be durable before lease release" + ); + assert!(state + .db + .deletion_store() + .serving_writes_drained(community) + .await + .expect("serving lease released")); + let generation = state + .db + .deletion_store() + .fence(&claim.lease) + .await + .expect("fence after pointer seed"); + assert_eq!(generation, 1); + assert_eq!( + state + .db + .deletion_store() + .get(request.id) + .await + .expect("fenced request") + .stage, + buzz_db::deletion::DeletionStage::Fenced + ); + drop(state); + pool.close().await; + } + #[tokio::test] #[ignore = "requires Postgres and MinIO"] async fn finalize_push_holds_serving_lease_through_post_cas_publication() { diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 89595fbee17..d37e7b375ee 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -2578,6 +2578,31 @@ async fn handle_git_repo_announcement( event: &Event, state: &Arc, ) -> anyhow::Result<()> { + handle_git_repo_announcement_inner(tenant, event, state, &GitRepoAnnouncementHooks::default()) + .await +} + +#[derive(Default)] +pub(crate) struct GitRepoAnnouncementHooks { + #[cfg(test)] + pub(crate) post_lease_gate: Option>, +} + +#[cfg(test)] +#[derive(Default)] +pub(crate) struct GitRepoAnnouncementGate { + pub(crate) reached: tokio::sync::Notify, + pub(crate) resume: tokio::sync::Notify, +} + +pub(crate) async fn handle_git_repo_announcement_inner( + tenant: &TenantContext, + event: &Event, + state: &Arc, + hooks: &GitRepoAnnouncementHooks, +) -> anyhow::Result<()> { + #[cfg(not(test))] + let _ = hooks; // Extract repo identifier from d tag (required for NIP-33 parameterized replaceable events). let repo_id = extract_tag_value(event, "d").ok_or_else(|| anyhow::anyhow!("kind:30617 missing d tag"))?; @@ -2677,6 +2702,32 @@ async fn handle_git_repo_announcement( // other attempt already established. let reserved_by_this_attempt = matches!(outcome, ReserveOutcome::Reserved); + // The event row and name registry are ordinary database state: if deletion + // quiescing wins before they commit, the DB write fence rejects them; if + // they committed first, the destructive DB stage purges them. The manifest + // and pointer below are external S3 effects, so acquire the durable + // serving-write lease immediately before that sequence. Once acquired, + // deletion must drain this lease before it can freeze the final object list. + let serving_write = buzz_deletion::acquire_serving_write( + &state.db, + tenant.community(), + "git_repo_announcement", + ) + .await + .map_err(|e| anyhow::anyhow!("repo announcement rejected by community deletion fence: {e}"))?; + + #[cfg(test)] + if let Some(gate) = &hooks.post_lease_gate { + gate.reached.notify_one(); + gate.resume.notified().await; + } + + if let Err(error) = serving_write.verify().await { + return Err(anyhow::anyhow!( + "repo announcement lost community serving lease: {error}" + )); + } + // Establish/confirm the manifest pointer, keeping the invariant // "repo announced ⟺ pointer exists" so the read path can rely on // pointer-absent meaning never-announced (keeping `info_refs`'s fail-closed @@ -2691,10 +2742,16 @@ async fn handle_git_repo_announcement( // re-announce must accept it untouched; only an absent pointer is // repaired by seeding. Using the strict seed here would wrongly reject // every re-announce after the first push. - let pointer_result = if reserved_by_this_attempt { - seed_manifest_pointer(state, tenant, &owner_hex, &repo_id).await - } else { - ensure_manifest_pointer(state, tenant, &owner_hex, &repo_id).await + let pointer_operation = async { + if reserved_by_this_attempt { + seed_manifest_pointer(state, tenant, &owner_hex, &repo_id).await + } else { + ensure_manifest_pointer(state, tenant, &owner_hex, &repo_id).await + } + }; + let pointer_result = match serving_write.protect(pointer_operation).await { + Ok(result) => result, + Err(error) => Err(error), }; if let Err(pointer_err) = pointer_result { // A reserved name without a clone-able pointer is exactly the broken @@ -2743,7 +2800,9 @@ async fn handle_git_repo_announcement( // initial empty signal is a one-time seeding notification, not something a // re-announce should replay. if reserved_by_this_attempt { - if let Err(e) = emit_initial_ref_state(tenant, state, &owner_hex, &repo_id).await { + if let Err(e) = + emit_initial_ref_state(tenant, state, serving_write.lease(), &owner_hex, &repo_id).await + { // Non-fatal: the manifest is the source of truth; this is just the // derived notification. A failure here means subscribers miss the // "repo now exists" event, but clone/push still works. @@ -2756,6 +2815,9 @@ async fn handle_git_repo_announcement( } } + serving_write.finish().await.map_err(|e| { + anyhow::anyhow!("repo announcement lost community serving lease on release: {e}") + })?; Ok(()) } @@ -2897,6 +2959,7 @@ async fn ensure_manifest_pointer( async fn emit_initial_ref_state( tenant: &TenantContext, state: &Arc, + lease: &buzz_db::deletion::ServingWriteLease, owner_hex: &str, repo_id: &str, ) -> anyhow::Result<()> { @@ -2914,7 +2977,7 @@ async fn emit_initial_ref_state( .map_err(|e| anyhow::anyhow!("build_ref_state_event: {e}"))?; let (stored, was_inserted) = state .db - .insert_event(tenant.community(), &event, None) + .insert_event_with_serving_write_guard(lease, &event, None) .await .map_err(|e| anyhow::anyhow!("insert kind:30618: {e}"))?; if was_inserted { diff --git a/deploy/charts/buzz/values.yaml b/deploy/charts/buzz/values.yaml index ca3403a633f..448e15ca976 100644 --- a/deploy/charts/buzz/values.yaml +++ b/deploy/charts/buzz/values.yaml @@ -333,6 +333,15 @@ externalRedis: # first sweep fails AccessDenied and buzz_storage_sweep_ok stays 0 — no other # media functionality is affected. Set BUZZ_STORAGE_METRICS=off to disable # the sweep entirely on a deployment that can't grant it. +# +# Whole-community deletion (`buzz-admin deletions ...`) permanently removes +# tenant-owned object versions through the v5 deletion path. In addition to the +# ordinary media permissions, every deployment that enables community deletion +# needs bucket-level `s3:ListBucketVersions` and object-level +# `s3:DeleteObjectVersion` on this bucket before exercising deletion — including +# never-versioned buckets, which S3 exposes through `ListObjectVersions` with the +# `null` version id. +# # Note: buzz_storage_sweep_failures is a process-local gauge — on leader # failover it resets to the new leader's local count, not a global total. # Note: on a failed sweep attempt, the next retry fires on the next usage tick diff --git a/desktop/src-tauri/Cargo.lock b/desktop/src-tauri/Cargo.lock index fb60a351895..0933149f5b5 100644 --- a/desktop/src-tauri/Cargo.lock +++ b/desktop/src-tauri/Cargo.lock @@ -1191,6 +1191,7 @@ dependencies = [ "infer", "mp4", "nostr 0.44.7", + "quick-xml 0.38.4", "rust-s3", "serde", "serde_json",