External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat: add read-only community deletion preview

Browse Source 
Co-authored-by: am <6e30cd56c30e030cd31bb0939b94a7c257c9a09d5ba2d92cf2735da45629f248@buzz.block.builderlab.xyz>
Signed-off-by: am <6e30cd56c30e030cd31bb0939b94a7c257c9a09d5ba2d92cf2735da45629f248@buzz.block.builderlab.xyz>
This commit is contained in:
am
2026-08-12 13:21:45 -07:00
parent c6c6e7eca7
commit 30d9609691
3 changed files with 234 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/buzz-admin/src/deletions.rs
+13
View File
@@ -11,6 +11,19 @@ pub async fn run(command: DeletionsCommand) -> anyhow::Result<i32> {
mod tests {
use clap::Parser;
#[test]
fn preview_command_parses_as_a_distinct_read_only_operation() {
let command = crate::Cli::try_parse_from([
"buzz-admin",
"deletions",
"preview",
"--host",
"community.example",
"--include-keys",
]);
assert!(command.is_ok());
}
#[test]
fn continuous_worker_command_is_not_exposed() {
let command = crate::Cli::try_parse_from(["buzz-admin", "deletions", "worker"]);
crates/buzz-db/src/deletion.rs
+48
View File
@@ -558,6 +558,24 @@ impl DeletionStore {
.unwrap_or(false)
}
/// Resolve an active deletion-preview target without creating a request.
pub async fn preview_target(&self, community_host: &str) -> Result<(CommunityId, String)> {
let row: Option<(Uuid, String)> = sqlx::query_as(
"SELECT id, host FROM communities \
WHERE lower(host) = lower($1) AND deletion_state = 'active' \
AND deleted_at IS NULL",
)
.bind(community_host)
.fetch_optional(&self.pool)
.await?;
row.map(|(id, host)| (CommunityId::from_uuid(id), host))
.ok_or_else(|| {
DbError::DeletionSafety(format!(
"community {community_host:?} is missing, fenced, or tombstoned"
))
})
}
/// Persist a request. Only active non-tombstone communities may be submitted.
pub async fn submit(
&self,
@@ -3193,6 +3211,36 @@ mod postgres_tests {
(request, inventory)
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn preview_target_reads_active_community_without_creating_request() {
let (db, store) = store().await;
let host = format!("preview-{}.example", Uuid::new_v4().simple());
let community = db
.ensure_configured_community(&host)
.await
.expect("create preview community");
let (preview_id, canonical_host) = store
.preview_target(&host.to_uppercase())
.await
.expect("resolve preview target");
assert_eq!(preview_id, community.id);
assert_eq!(canonical_host, host);
let requests: i64 = sqlx::query_scalar(
"SELECT count(*) FROM community_deletion_requests WHERE community_id = $1",
)
.bind(community.id.as_uuid())
.fetch_one(&db.pool)
.await
.expect("count preview requests");
assert_eq!(
requests, 0,
"preview must not create durable deletion state"
);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn approval_boundary_blocks_claim_until_exact_inventory_is_approved() {
crates/buzz-deletion/src/lib.rs
+173 -8
View File
@@ -10,7 +10,8 @@ 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, SchemaManifest, StorageManifest,
DEFAULT_LEASE_DURATION,
};
use buzz_db::{Db, DbConfig};
use buzz_media::{is_tenant_owned_key, tenant_prefixes, MediaStorage};
@@ -217,6 +218,15 @@ async fn acquire_serving_write_with_heartbeat(
/// CLI-only whole-community deletion commands.
#[derive(Subcommand)]
pub enum Command {
/// Read live deletion targets without creating or changing durable state.
Preview {
/// Canonical community host. Defaults to RELAY_URL's authority.
#[arg(long)]
host: Option<String>,
/// Stream concrete object-store and Redis keys as NDJSON before the summary.
#[arg(long)]
include_keys: bool,
},
/// Persist a deletion request and freeze its initial cross-store inventory.
Submit {
/// Canonical community host. Defaults to RELAY_URL's authority.
@@ -353,6 +363,39 @@ fn is_permanent_error(error: &anyhow::Error) -> bool {
})
}
#[derive(Debug, Serialize)]
struct PreviewOutput {
record: &'static str,
advisory: bool,
observation_started_at: chrono::DateTime<chrono::Utc>,
observation_completed_at: chrono::DateTime<chrono::Utc>,
warning: &'static str,
community_id: String,
community_host: String,
postgres: SchemaManifest,
object_store: PreviewStorage,
redis: PreviewRedis,
}
#[derive(Debug, Serialize)]
struct PreviewStorage {
manifest: StorageManifest,
keys_included: bool,
}
#[derive(Debug, Serialize)]
struct PreviewRedis {
observed_key_count: u64,
keys_included: bool,
}
#[derive(Debug, Serialize)]
struct PreviewKey<'a> {
record: &'static str,
store: &'static str,
key: &'a str,
}
#[derive(Debug, Serialize)]
struct RunOutput {
request_id: Uuid,
@@ -430,6 +473,55 @@ pub async fn run(command: Command) -> Result<i32> {
async fn run_with_services(command: Command, services: Services) -> Result<i32> {
match command {
Command::Preview { host, include_keys } => {
let observation_started_at = chrono::Utc::now();
let relay_url = std::env::var("RELAY_URL").ok();
let host = resolve_submit_host(host.as_deref(), relay_url.as_deref())?;
let (community_id, community_host) = services.store.preview_target(&host).await?;
let postgres = services.store.inventory_schema(community_id).await?;
let mut object_key_count = 0u64;
let object_store = enumerate_tenant_prefixes(
&services,
community_id,
None,
None,
include_keys.then_some(&mut object_key_count),
)
.await?;
if include_keys {
debug_assert_eq!(
object_key_count,
object_store
.prefixes
.iter()
.map(|prefix| prefix.object_count)
.sum::<u64>()
);
}
let redis =
preview_redis_namespace(&services.redis, community_id, include_keys).await?;
let output = PreviewOutput {
record: "deletion_preview_summary",
advisory: true,
observation_started_at,
observation_completed_at: chrono::Utc::now(),
warning: "live state may change after this read-only observation; execution freezes its authoritative manifest only after fence and drain",
community_id: community_id.to_string(),
community_host,
postgres,
object_store: PreviewStorage {
manifest: object_store,
keys_included: include_keys,
},
redis,
};
if include_keys {
print_ndjson(&output)?;
} else {
print_json(&output)?;
}
Ok(0)
}
Command::Submit {
host,
requested_by,
@@ -658,7 +750,8 @@ async fn build_inventory(
.store
.inventory_schema(request.community_id)
.await?;
let storage = enumerate_tenant_prefixes(services, request, None, None).await?;
let storage =
enumerate_tenant_prefixes(services, request.community_id, None, None, None).await?;
Ok(FrozenInventory { schema, storage })
}
@@ -693,19 +786,20 @@ async fn flush_chunk(services: &Services, sink: &mut ChunkSink<'_>, prefix: &str
/// digests.
async fn enumerate_tenant_prefixes(
services: &Services,
request: &DeletionRequest,
community: buzz_core::CommunityId,
heartbeat_lost: Option<&CancellationToken>,
mut sink: Option<&mut ChunkSink<'_>>,
mut preview_key_count: Option<&mut u64>,
) -> 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 community_uuid = *community.as_uuid();
let chunk_keys = manifest_chunk_keys();
let mut prefixes = Vec::new();
for prefix in tenant_prefixes(community) {
for prefix in tenant_prefixes(community_uuid) {
let mut digest = KeyStreamDigest::new();
let mut total_bytes: u64 = 0;
let mut continuation = None;
@@ -718,13 +812,21 @@ async fn enumerate_tenant_prefixes(
.list_prefix_page(&prefix, continuation.take(), LIST_PAGE_SIZE)
.await?;
for (key, size) in page.objects {
if !is_tenant_owned_key(community, &key) {
if !is_tenant_owned_key(community_uuid, &key) {
return Err(permanent(format!(
"key under a tenant prefix is outside the exact writer taxonomy: {key}"
)));
}
digest.fold(&key)?;
total_bytes = total_bytes.saturating_add(size);
if let Some(key_count) = preview_key_count.as_deref_mut() {
print_ndjson(&PreviewKey {
record: "deletion_preview_key",
store: "object_store",
key: &key,
})?;
*key_count = key_count.saturating_add(1);
}
if let Some(sink) = sink.as_deref_mut() {
sink.buffered.push(key);
if sink.buffered.len() >= chunk_keys {
@@ -779,8 +881,14 @@ async fn freeze_destructive_manifest(
next_chunk_no: 0,
buffered: Vec::new(),
};
let manifest =
enumerate_tenant_prefixes(services, request, Some(heartbeat_lost), Some(&mut sink)).await?;
let manifest = enumerate_tenant_prefixes(
services,
request.community_id,
Some(heartbeat_lost),
Some(&mut sink),
None,
)
.await?;
services
.store
.freeze_destructive_storage_manifest(token, &manifest)
@@ -1283,6 +1391,50 @@ async fn publish_disconnect_community(
Ok(())
}
fn preview_redis_page(include_keys: bool, keys: &[String], key_count: &mut u64) -> Result<()> {
*key_count = key_count.saturating_add(keys.len() as u64);
if include_keys {
for key in keys {
print_ndjson(&PreviewKey {
record: "deletion_preview_key",
store: "redis",
key,
})?;
}
}
Ok(())
}
async fn preview_redis_namespace(
pool: &deadpool_redis::Pool,
community: buzz_core::CommunityId,
include_keys: bool,
) -> Result<PreviewRedis> {
let mut connection = pool.get().await?;
let pattern = format!("buzz:{community}:*");
let mut cursor = 0u64;
let mut key_count = 0u64;
loop {
let (next, keys): (u64, Vec<String>) = redis::cmd("SCAN")
.arg(cursor)
.arg("MATCH")
.arg(&pattern)
.arg("COUNT")
.arg(1000)
.query_async(&mut *connection)
.await?;
preview_redis_page(include_keys, &keys, &mut key_count)?;
if next == 0 {
break;
}
cursor = next;
}
Ok(PreviewRedis {
observed_key_count: key_count,
keys_included: include_keys,
})
}
async fn purge_redis_namespace(
pool: &deadpool_redis::Pool,
community: buzz_core::CommunityId,
@@ -1420,6 +1572,11 @@ fn run_output(request: DeletionRequest) -> RunOutput {
}
}
fn print_ndjson(value: &impl Serialize) -> Result<()> {
println!("{}", serde_json::to_string(value)?);
Ok(())
}
fn print_json(value: &impl Serialize) -> Result<()> {
println!("{}", serde_json::to_string_pretty(value)?);
Ok(())
@@ -1429,6 +1586,14 @@ fn print_json(value: &impl Serialize) -> Result<()> {
mod tests {
use super::*;
#[test]
fn preview_redis_count_saturates_without_retaining_keys() {
let keys = vec!["buzz:a:one".to_string(), "buzz:a:two".to_string()];
let mut count = u64::MAX - 1;
preview_redis_page(false, &keys, &mut count).expect("count preview page");
assert_eq!(count, u64::MAX);
}
#[test]
fn submit_host_prefers_explicit_host() {
assert_eq!(