Support community deletion in versioned media buckets (#6738)

This commit is contained in:
Brad Seiler
2026-08-25 12:02:18 -04:00
committed by GitHub
parent bb5b9357a7
commit d12dea4e67
11 changed files with 1584 additions and 101 deletions

1
Cargo.lock generated
View File

@@ -1136,6 +1136,7 @@ dependencies = [
"infer",
"mp4",
"nostr 0.44.7",
"quick-xml 0.38.4",
"rust-s3",
"serde",
"serde_json",

View File

@@ -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<String>,
}
/// 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<String>,
version_id: impl Into<String>,
kind: impl Into<String>,
) -> 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<String> {
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<Self> {
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<BTreeSet<Strin
/// Fail closed when a community-prefix inventory has an unsafe shape.
pub fn validate_storage_manifest(manifest: &StorageManifest) -> 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 {

View File

@@ -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<StorageManifest> {
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<String>,
next_version_id_marker: Option<String>,
) -> 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<F, Fut>(
chunk: &buzz_db::deletion::ManifestKeyChunk,
storage_version: i32,
delete: F,
) -> Result<BulkDeleteOutcome>
where
F: FnOnce(Vec<ObjectVersionRef>) -> Fut,
Fut: Future<Output = Result<BulkDeleteOutcome>>,
{
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<Vec<ObjectVersionRef>> {
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<serde_json::Value> {
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::<Vec<_>>();
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 heartbeat_lost.is_cancelled() {
return Err(DeletionLeaseLost.into());
}
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
)));
}
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");

View File

@@ -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"] }

View File

@@ -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::{

View File

@@ -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<Box<dyn futures_core::Stream<Item = Result<Bytes, MediaError>> + 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<ObjectVersionEntry>,
/// Next key marker for truncated listings.
pub next_key_marker: Option<String>,
/// Next version-id marker for truncated listings.
pub next_version_id_marker: Option<String>,
/// Whether more pages remain.
pub is_truncated: bool,
}
#[derive(Debug, Default)]
struct ListVersionFields {
key: Option<String>,
version_id: Option<String>,
size: Option<u64>,
}
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<String, MediaError> {
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<ObjectVersionEntry, MediaError> {
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::<u64>().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<ObjectVersionsPage, MediaError> {
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<Bucket>,
@@ -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<bool, MediaError> {
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::<Vec<_>>();
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<BulkDeleteOutcome, MediaError> {
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<BulkDeleteOutcome, MediaError> {
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<s3::serde_types::ObjectIdentifier>,
) -> Result<BulkDeleteOutcome, MediaError> {
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<String>,
version_id_marker: Option<String>,
max_keys: usize,
) -> Result<ObjectVersionsPage, MediaError> {
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<s3::serde_types::ObjectIdentifier> {
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#"<?xml version="1.0" encoding="UTF-8"?>
<ListVersionsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
<Name>buzz-media</Name>
<Prefix>_meta/tenant/</Prefix>
<KeyMarker>_meta/tenant/a.json</KeyMarker>
<VersionIdMarker>v-old</VersionIdMarker>
<MaxKeys>2</MaxKeys>
<IsTruncated>true</IsTruncated>
<NextKeyMarker>_meta/tenant/a.json</NextKeyMarker>
<NextVersionIdMarker>v-new</NextVersionIdMarker>
<DeleteMarker>
<Key>_meta/tenant/a.json</Key>
<VersionId>v-delete</VersionId>
<IsLatest>true</IsLatest>
</DeleteMarker>
<Version>
<Key>_meta/tenant/a.json</Key>
<VersionId>v-new</VersionId>
<IsLatest>false</IsLatest>
<Size>42</Size>
</Version>
</ListVersionsResult>"#,
)
.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#"<ListVersionsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
<Version><Key>k-a</Key><VersionId>v3</VersionId><Size>3</Size></Version>
<DeleteMarker><Key>k-a</Key><VersionId>v2</VersionId></DeleteMarker>
<Version><Key>k-a</Key><VersionId>v1</VersionId><Size>1</Size></Version>
<DeleteMarker><Key>k-b</Key><VersionId>m2</VersionId></DeleteMarker>
<Version><Key>k-b</Key><VersionId>m1</VersionId><Size>10</Size></Version>
</ListVersionsResult>"#,
)
.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#"<ListVersionsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
<DeleteMarker><Key>k-marker-only</Key><VersionId>d-only</VersionId></DeleteMarker>
<Version><Key>k-versioned</Key><VersionId>v2</VersionId><Size>20</Size></Version>
<DeleteMarker><Key>k-versioned</Key><VersionId>d1</VersionId></DeleteMarker>
<Version><Key>k-versioned</Key><VersionId>v1</VersionId><Size>10</Size></Version>
</ListVersionsResult>"#,
)
.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)),

View File

@@ -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<buzz_media::storage::ObjectVersionEntry> {
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<ObjectVersionRef> {
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<ObjectVersionRef> = 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");
}

View File

@@ -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() {

View File

@@ -2578,6 +2578,31 @@ async fn handle_git_repo_announcement(
event: &Event,
state: &Arc<AppState>,
) -> 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<Arc<GitRepoAnnouncementGate>>,
}
#[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<AppState>,
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<AppState>,
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 {

View File

@@ -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

View File

@@ -1191,6 +1191,7 @@ dependencies = [
"infer",
"mp4",
"nostr 0.44.7",
"quick-xml 0.38.4",
"rust-s3",
"serde",
"serde_json",