From 0cfa50ac70b26208367c56dc5ddf3fcef61808e4 Mon Sep 17 00:00:00 2001 From: Jemiah Westerman Date: Thu, 5 Nov 2026 15:42:41 -0800 Subject: [PATCH] perf(db): use exact event timestamps for transactional deletion (#9130) ## Summary Preserve the target event's already-known signed timestamp through deletion planning and use `(community_id, id, created_at)` for the canonical transactional read, soft delete, and retention-free purge. Known-target operations can prune unrelated monthly partitions instead of repeatedly probing by ID alone. Public ID-only deletion callers retain their discovery step and reuse the discovered timestamp for subsequent mutations. Operator deletion similarly reuses its existing discovery result. Canonical kind/channel classification and locking remain database-owned inside the transaction. ## Scope / compatibility - Combined-tree verification also passed: repository-wide `just ci`, 220 unit/source-policy tests, and the full 722-test buzz-db/search PostgreSQL suite. All three temporary timestamp-constraint mutations were killed by functional/plan assertions and restored. - Independent of companion search-hydration and push-enqueue PRs; based directly on `main`. - Deploy/code-only change. No schema, migrations, backfills, persistent/serialized data formats, indexes, triggers, functions, rules, procedures, or foreign keys added. - Runtime `DeletionTarget` now carries the target timestamp; no wire payload changes. - Community admission, canvas replacement locking, replay repair, atomicity, retention-free purge/mention cleanup, artifact restrictions, and once-only thread accounting remain intact. - Extract deletion helpers into a focused private module, preserving existing public reexports. - Existing/new relays coexist against the unchanged schema; no operator prebuild or rollout dependency. ## Verification - Compiled `buzz-db`, `buzz-search`, and `buzz-relay` library/tests. - 218 infrastructure-free unit/source-policy tests passed. - 23 focused PostgreSQL tests passed: cross-month/exact-key/tenant isolation, missing/wrong/deleted targets, rollback and concurrent replay/counters, retention-free cleanup in desired and migration schema modes, canvas locking, and existing deletion/admin regressions. - Actual production prepared SELECT/UPDATE/purge SQL prunes to one event month under custom and generic plans. - Targeted clippy with `-D warnings`, formatting, diff checks, PostgreSQL discovery, and differential file-size checks passed. - Live isolated relay smoke: signed message and reply ingestion, atomic deletion, deleted payload absent from queries, thread counters return to zero, and safe deletion replay. - Independent local review completed: no remaining blockers. No production latency/workload benchmark is claimed. --------- Signed-off-by: Jemiah Westerman Co-authored-by: Son of Anton <8e0ef2ab02d69a7ac089ef5eca90d1a3fbbdc546f6322dc34d0c4f3deecea7f4@buzz.block.builderlab.xyz> Co-authored-by: Claude Opus 5.5 --- crates/buzz-db/src/store/event.rs | 348 +--------- crates/buzz-db/src/store/event_deletion.rs | 406 ++++++++++++ .../store/event_deletion/postgres_tests.rs | 594 ++++++++++++++++++ crates/buzz-db/src/store/mod.rs | 2 + .../buzz-db/src/store/relay_admin_actions.rs | 23 +- .../buzz-relay/src/handlers/side_effects.rs | 17 +- 6 files changed, 1037 insertions(+), 353 deletions(-) create mode 100644 crates/buzz-db/src/store/event_deletion.rs create mode 100644 crates/buzz-db/src/store/event_deletion/postgres_tests.rs diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 2ee6b30a52..bc7a121173 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -1646,344 +1646,11 @@ fn count_events_sql(q: &EventQuery) -> QueryBuilder { qb } -/// SQL predicate over an `events` row for coordinates that carry no -/// historical value: NIP-RS read state and Buzz mesh heartbeats. Their -/// deletion is physical, including the mention index, so soft-deleted payloads -/// never accumulate. This is exactly the classification used by the migration -/// 0011 / 0019 purge triggers this path replaces. -macro_rules! retention_free_event_predicate { - () => { - "((kind = 30078 \ - AND d_tag ~ '^read-state:[0-9a-f]{32}$' \ - AND (SELECT count(*) \ - FROM jsonb_array_elements(CASE WHEN jsonb_typeof(tags) = 'array' THEN tags ELSE '[]'::jsonb END) tag \ - WHERE jsonb_typeof(tag) = 'array' AND tag->0 = '\"d\"'::jsonb) = 1 \ - AND EXISTS (SELECT 1 \ - FROM jsonb_array_elements(CASE WHEN jsonb_typeof(tags) = 'array' THEN tags ELSE '[]'::jsonb END) tag \ - WHERE jsonb_typeof(tag) = 'array' AND jsonb_array_length(tag) >= 2 \ - AND jsonb_typeof(tag->1) = 'string' AND tag->>0 = 'd' AND tag->>1 = d_tag) \ - AND (SELECT count(*) \ - FROM jsonb_array_elements(CASE WHEN jsonb_typeof(tags) = 'array' THEN tags ELSE '[]'::jsonb END) tag \ - WHERE tag = '[\"t\", \"read-state\"]'::jsonb) = 1) \ - OR (kind = 30003 \ - AND d_tag LIKE 'buzz-mesh-member-status:%' \ - AND tags @> '[[\"k\", \"buzz-mesh-status\"]]'::jsonb))" - }; -} - -/// Kinds that can hold a retention-free coordinate. Cheap Rust gate so -/// ordinary deletions skip the purge statement entirely. -fn may_be_retention_free(kind: i32) -> bool { - kind == 30078 || kind == 30003 -} - -/// Soft-delete the live row for an addressable coordinate -/// `(kind, pubkey, d_tag)` — the NIP-33 replacement key — provided it is not -/// newer than the deletion request. -/// -/// Used by `handle_a_tag_deletion` to honour NIP-09 a-tag deletions for any -/// parameterized-replaceable kind. The WHERE clause mirrors -/// `replace_parameterized_event` so the coordinate semantics stay consistent: -/// `channel_id` is intentionally NOT in the key (NIP-33 replacement is global -/// per the spec — `channel_id` is stored for query scoping, not identity). -/// -/// `deletion_created_at_secs` is the deletion event's own `created_at`. NIP-09 -/// scopes an `a`-tag deletion to versions at or before that instant, so a -/// delayed or replayed tombstone signed between two versions must not erase the -/// newer replacement. `events.created_at` is immutable per row, so the predicate -/// guarantees a tombstone can never erase a version newer than itself — the UPDATE -/// re-evaluates its WHERE clause after any lock wait, so a replacement that races -/// the deletion and lands with a later `created_at` is always spared. -/// -/// This does NOT guarantee deletion completeness when a same-coordinate -/// replacement races the deletion: the deletion may evaluate its predicate before -/// the replacement arrives, miss the incoming head, and return `Ok(false)`. That -/// outcome is state-identical to the deletion having arrived first (old head -/// gone, new head present), which is a valid Nostr ordering — Nostr never fixes -/// the order of concurrent writes from different signers, and even same-signer -/// ordering is advisory. The return value feeds only a debug log, not a -/// correctness gate. -/// -/// Returns `Ok(true)` if a row was deleted, `Ok(false)` if no live row matched -/// (already deleted, never existed, or strictly newer than the deletion). -pub async fn soft_delete_by_coordinate( - pool: &PgPool, - community_id: CommunityId, - kind: i32, - pubkey: &[u8], - d_tag: &str, - deletion_created_at_secs: i64, -) -> Result { - let deletion_created_at = DateTime::from_timestamp(deletion_created_at_secs, 0) - .ok_or(DbError::InvalidTimestamp(deletion_created_at_secs))?; - let mut tx = crate::begin_community_event_write_transaction( - pool, - community_id, - crate::observability::WriterOperation::EventWrite, - ) - .await?; - let purged = if may_be_retention_free(kind) { - purge_retention_free_events( - &mut tx, - RetentionFreeTarget::Coordinate { - kind, - pubkey, - d_tag, - created_at_or_before: deletion_created_at, - }, - ) - .await? - } else { - 0 - }; - // `AND NOT COALESCE(, false)` holds the invariant in SQL rather than through - // statement order: under READ COMMITTED this UPDATE takes a fresh snapshot, - // so a retention-free head committed by a racing replacement after the - // purge ran is spared instead of soft-deleted (the "deletion arrived - // first" outcome documented above). COALESCE keeps a NULL predicate - // (e.g. NULL `d_tag`) on the ordinary soft-delete path, as the triggers do. - let result = sqlx::query(concat!( - "UPDATE events SET deleted_at = NOW() \ - WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL \ - AND created_at <= $5 AND NOT COALESCE(", - retention_free_event_predicate!(), - ", false)" - )) - .bind(community_id.as_uuid()) - .bind(kind) - .bind(pubkey) - .bind(d_tag) - .bind(deletion_created_at) - .execute(tx.conn()) - .await?; - - tx.commit().await?; - - Ok(purged > 0 || result.rows_affected() > 0) -} - -/// Purge live retention-free rows matching `$target`, plus their mention rows, -/// and return the number of events removed. -macro_rules! purge_retention_free_sql { - ($target:literal) => { - concat!( - "WITH purged AS ( \ - DELETE FROM events \ - WHERE community_id = $1 AND ", - $target, - " AND deleted_at IS NULL AND ", - retention_free_event_predicate!(), - " RETURNING id \ - ), purged_mentions AS ( \ - DELETE FROM event_mentions mention USING purged \ - WHERE mention.community_id = $1 AND mention.event_id = purged.id \ - ) \ - SELECT count(*) FROM purged" - ) - }; -} - -/// Which live rows a deletion targets. -enum RetentionFreeTarget<'a> { - Id(&'a [u8]), - Coordinate { - kind: i32, - pubkey: &'a [u8], - d_tag: &'a str, - created_at_or_before: DateTime, - }, -} - -/// Physically delete live retention-free rows matched by `target`, plus their -/// mention rows, in the caller's transaction. The generic soft delete that -/// follows excludes the same predicate, so a retention-free row is either -/// purged here or left live, never soft-deleted. Returns the number of events -/// removed. -async fn purge_retention_free_events( - tx: &mut AdmittedTx, - target: RetentionFreeTarget<'_>, -) -> Result { - let community_id = tx.community(); - // Migration 0011 fences NIP-RS hard deletes behind a transaction-local - // opt-in. The fence and this opt-in are removed together once the - // migration-only triggers are dropped. - sqlx::query("SELECT set_config('buzz.nip_rs_hard_delete', 'on', true)") - .execute(tx.conn()) - .await?; - - let purged: i64 = match target { - RetentionFreeTarget::Id(event_id) => { - sqlx::query_scalar(purge_retention_free_sql!("id = $2")) - .bind(community_id.as_uuid()) - .bind(event_id) - .fetch_one(tx.conn()) - .await? - } - RetentionFreeTarget::Coordinate { - kind, - pubkey, - d_tag, - created_at_or_before, - } => { - sqlx::query_scalar(purge_retention_free_sql!( - "kind = $2 AND pubkey = $3 AND d_tag = $4 AND created_at <= $5" - )) - .bind(community_id.as_uuid()) - .bind(kind) - .bind(pubkey) - .bind(d_tag) - .bind(created_at_or_before) - .fetch_one(tx.conn()) - .await? - } - }; - - Ok(purged as u64) -} - -/// An event a NIP-09 (kind 5) or NIP-29 (kind 9005) deletion soft-deletes in -/// the deletion's own insert transaction, with the thread it belongs to. -#[derive(Clone, Copy, Debug)] -pub struct DeletionTarget<'a> { - /// The event to soft-delete. - pub event_id: &'a [u8], - /// Its direct parent, whose `reply_count` drops when it is deleted. - pub parent_event_id: Option<&'a [u8]>, - /// Its thread root, whose `descendant_count` drops when it is deleted. - pub root_event_id: Option<&'a [u8]>, -} - -/// Atomically soft-delete an event and decrement thread reply counters. -/// -/// Wraps the delete + counter update in a single transaction so a crash between -/// them cannot leave counters permanently inflated. Returns `Ok(true)` if the -/// event was deleted this call. -/// -/// When the target event is a kind-40100 (canvas) event, this function derives -/// the target's `kind` and `channel_id` from the database inside the same -/// transaction and acquires the same `(community, kind, channel)` advisory lock -/// used by [`insert_canvas_head_checked`] before the UPDATE. This prevents a -/// concurrent tagged write from observing a head that is simultaneously being -/// removed. The serialization invariant is owned entirely by this function; -/// callers do not classify the target kind. -pub async fn soft_delete_event_and_update_thread( - pool: &PgPool, - community_id: CommunityId, - event_id: &[u8], - parent_event_id: Option<&[u8]>, - root_event_id: Option<&[u8]>, -) -> Result { - let mut tx = crate::begin_community_event_write_transaction( - pool, - community_id, - crate::observability::WriterOperation::EventWrite, - ) - .await?; - let deleted = soft_delete_event_and_update_thread_in_tx( - &mut tx, - event_id, - parent_event_id, - root_event_id, - ) - .await?; - tx.commit().await?; - Ok(deleted) -} - -/// Transaction-scoped body of [`soft_delete_event_and_update_thread`]. -/// -/// Callers that must fence the delete with their own writes (e.g. the admin -/// action lease/marker) run this inside their transaction; the caller commits. -pub(crate) async fn soft_delete_event_and_update_thread_in_tx( - tx: &mut AdmittedTx, - event_id: &[u8], - parent_event_id: Option<&[u8]>, - root_event_id: Option<&[u8]>, -) -> Result { - let community_id = tx.community(); - use crate::store::replaceable::event_replacement_lock_key; - - // Derive the target event's kind and channel_id inside the transaction so - // that the serialization decision cannot be bypassed by any caller. - let target: Option<(i32, Option)> = sqlx::query_as( - "SELECT kind, channel_id FROM events \ - WHERE community_id = $1 AND id = $2", - ) - .bind(community_id.as_uuid()) - .bind(event_id) - .fetch_optional(tx.conn()) - .await?; - - // Relay-signed move removals are the source channel's only replay record. - if target.is_some_and(|(kind, _)| kind == 45011) { - return Err(DbError::InvalidData( - "artifact removal markers cannot be deleted".into(), - )); - } - - if let Some((kind, Some(channel_id))) = target { - if kind == KIND_CANVAS as i32 { - let lock_key = event_replacement_lock_key( - community_id, - kind, - &[], - Some(channel_id.as_bytes().as_slice()), - ); - sqlx::query("SELECT pg_advisory_xact_lock($1)") - .bind(lock_key) - .execute(tx.conn()) - .await?; - } - } - - let purged = if target.is_some_and(|(kind, _)| may_be_retention_free(kind)) { - purge_retention_free_events(tx, RetentionFreeTarget::Id(event_id)).await? - } else { - 0 - }; - let result = sqlx::query(concat!( - "UPDATE events SET deleted_at = NOW() \ - WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL AND NOT COALESCE(", - retention_free_event_predicate!(), - ", false)" - )) - .bind(community_id.as_uuid()) - .bind(event_id) - .execute(tx.conn()) - .await?; - - let deleted = purged > 0 || result.rows_affected() > 0; - - if deleted { - sqlx::query("UPDATE artifact_heads SET modified_at=clock_timestamp() WHERE community_id=$1 AND artifact_id=(SELECT artifact_id FROM artifact_revisions WHERE community_id=$1 AND event_id=$2) AND event_id=$2") - .bind(community_id.as_uuid()).bind(event_id).execute(tx.conn()).await?; - if let Some(pid) = parent_event_id { - sqlx::query( - "UPDATE thread_metadata \ - SET reply_count = GREATEST(reply_count - 1, 0) \ - WHERE community_id = $1 AND event_id = $2", - ) - .bind(community_id.as_uuid()) - .bind(pid) - .execute(tx.conn()) - .await?; - - if let Some(root_id) = root_event_id { - sqlx::query( - "UPDATE thread_metadata \ - SET descendant_count = GREATEST(descendant_count - 1, 0) \ - WHERE community_id = $1 AND event_id = $2", - ) - .bind(community_id.as_uuid()) - .bind(root_id) - .execute(tx.conn()) - .await?; - } - } - } - - Ok(deleted) -} +// Preserve the public event-store deletion API after the focused extraction. +pub(crate) use super::event_deletion::soft_delete_event_and_update_thread_in_tx; +pub use super::event_deletion::{ + soft_delete_by_coordinate, soft_delete_event_and_update_thread, DeletionTarget, +}; /// Returns the `created_at` timestamp of the most recent non-deleted event in a channel. pub async fn get_last_message_at( @@ -3289,6 +2956,7 @@ impl Db { crate::event::soft_delete_event_and_update_thread_in_tx( &mut tx, target.event_id, + Some(target.created_at), target.parent_event_id, target.root_event_id, ) @@ -6982,6 +6650,8 @@ mod postgres_tests { let deletion = deletion_of(&reaction); let targets = [DeletionTarget { event_id: reaction.id.as_bytes(), + created_at: DateTime::from_timestamp(reaction.created_at.as_secs() as i64, 0) + .expect("target timestamp"), parent_event_id: None, root_event_id: None, }]; @@ -7060,6 +6730,8 @@ mod postgres_tests { .expect("insert torn deletion"); let targets = [DeletionTarget { event_id: reaction.id.as_bytes(), + created_at: DateTime::from_timestamp(reaction.created_at.as_secs() as i64, 0) + .expect("target timestamp"), parent_event_id: None, root_event_id: None, }]; diff --git a/crates/buzz-db/src/store/event_deletion.rs b/crates/buzz-db/src/store/event_deletion.rs new file mode 100644 index 0000000000..44f0dfee15 --- /dev/null +++ b/crates/buzz-db/src/store/event_deletion.rs @@ -0,0 +1,406 @@ +//! Atomic event deletion, retention-free purging, and thread accounting. + +use buzz_core::{kind::KIND_CANVAS, CommunityId}; +use chrono::{DateTime, Utc}; +use sqlx::PgPool; +use uuid::Uuid; + +use crate::{ + error::{DbError, Result}, + AdmittedTx, +}; + +/// SQL predicate over an `events` row for coordinates that carry no +/// historical value: NIP-RS read state and Buzz mesh heartbeats. Their +/// deletion is physical, including the mention index, so soft-deleted payloads +/// never accumulate. This is exactly the classification used by the migration +/// 0011 / 0019 purge triggers this path replaces. +macro_rules! retention_free_event_predicate { + () => { + "((kind = 30078 \ + AND d_tag ~ '^read-state:[0-9a-f]{32}$' \ + AND (SELECT count(*) \ + FROM jsonb_array_elements(CASE WHEN jsonb_typeof(tags) = 'array' THEN tags ELSE '[]'::jsonb END) tag \ + WHERE jsonb_typeof(tag) = 'array' AND tag->0 = '\"d\"'::jsonb) = 1 \ + AND EXISTS (SELECT 1 \ + FROM jsonb_array_elements(CASE WHEN jsonb_typeof(tags) = 'array' THEN tags ELSE '[]'::jsonb END) tag \ + WHERE jsonb_typeof(tag) = 'array' AND jsonb_array_length(tag) >= 2 \ + AND jsonb_typeof(tag->1) = 'string' AND tag->>0 = 'd' AND tag->>1 = d_tag) \ + AND (SELECT count(*) \ + FROM jsonb_array_elements(CASE WHEN jsonb_typeof(tags) = 'array' THEN tags ELSE '[]'::jsonb END) tag \ + WHERE tag = '[\"t\", \"read-state\"]'::jsonb) = 1) \ + OR (kind = 30003 \ + AND d_tag LIKE 'buzz-mesh-member-status:%' \ + AND tags @> '[[\"k\", \"buzz-mesh-status\"]]'::jsonb))" + }; +} + +/// Kinds that can hold a retention-free coordinate. Cheap Rust gate so +/// ordinary deletions skip the purge statement entirely. +fn may_be_retention_free(kind: i32) -> bool { + kind == 30078 || kind == 30003 +} + +/// Soft-delete the live row for an addressable coordinate +/// `(kind, pubkey, d_tag)` — the NIP-33 replacement key — provided it is not +/// newer than the deletion request. +/// +/// Used by `handle_a_tag_deletion` to honour NIP-09 a-tag deletions for any +/// parameterized-replaceable kind. The WHERE clause mirrors +/// `replace_parameterized_event` so the coordinate semantics stay consistent: +/// `channel_id` is intentionally NOT in the key (NIP-33 replacement is global +/// per the spec — `channel_id` is stored for query scoping, not identity). +/// +/// `deletion_created_at_secs` is the deletion event's own `created_at`. NIP-09 +/// scopes an `a`-tag deletion to versions at or before that instant, so a +/// delayed or replayed tombstone signed between two versions must not erase the +/// newer replacement. `events.created_at` is immutable per row, so the predicate +/// guarantees a tombstone can never erase a version newer than itself — the UPDATE +/// re-evaluates its WHERE clause after any lock wait, so a replacement that races +/// the deletion and lands with a later `created_at` is always spared. +/// +/// This does NOT guarantee deletion completeness when a same-coordinate +/// replacement races the deletion: the deletion may evaluate its predicate before +/// the replacement arrives, miss the incoming head, and return `Ok(false)`. That +/// outcome is state-identical to the deletion having arrived first (old head +/// gone, new head present), which is a valid Nostr ordering — Nostr never fixes +/// the order of concurrent writes from different signers, and even same-signer +/// ordering is advisory. The return value feeds only a debug log, not a +/// correctness gate. +/// +/// Returns `Ok(true)` if a row was deleted, `Ok(false)` if no live row matched +/// (already deleted, never existed, or strictly newer than the deletion). +pub async fn soft_delete_by_coordinate( + pool: &PgPool, + community_id: CommunityId, + kind: i32, + pubkey: &[u8], + d_tag: &str, + deletion_created_at_secs: i64, +) -> Result { + let deletion_created_at = DateTime::from_timestamp(deletion_created_at_secs, 0) + .ok_or(DbError::InvalidTimestamp(deletion_created_at_secs))?; + let mut tx = crate::begin_community_event_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; + let purged = if may_be_retention_free(kind) { + purge_retention_free_events( + &mut tx, + RetentionFreeTarget::Coordinate { + kind, + pubkey, + d_tag, + created_at_or_before: deletion_created_at, + }, + ) + .await? + } else { + 0 + }; + // `AND NOT COALESCE(, false)` holds the invariant in SQL rather than through + // statement order: under READ COMMITTED this UPDATE takes a fresh snapshot, + // so a retention-free head committed by a racing replacement after the + // purge ran is spared instead of soft-deleted (the "deletion arrived + // first" outcome documented above). COALESCE keeps a NULL predicate + // (e.g. NULL `d_tag`) on the ordinary soft-delete path, as the triggers do. + let result = sqlx::query(concat!( + "UPDATE events SET deleted_at = NOW() \ + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL \ + AND created_at <= $5 AND NOT COALESCE(", + retention_free_event_predicate!(), + ", false)" + )) + .bind(community_id.as_uuid()) + .bind(kind) + .bind(pubkey) + .bind(d_tag) + .bind(deletion_created_at) + .execute(tx.conn()) + .await?; + + tx.commit().await?; + + Ok(purged > 0 || result.rows_affected() > 0) +} + +/// Purge live retention-free rows matching `$target`, plus their mention rows, +/// and return the number of events removed. +macro_rules! purge_retention_free_sql { + ($target:literal) => { + concat!( + "WITH purged AS ( \ + DELETE FROM events \ + WHERE community_id = $1 AND ", + $target, + " AND deleted_at IS NULL AND ", + retention_free_event_predicate!(), + " RETURNING id \ + ), purged_mentions AS ( \ + DELETE FROM event_mentions mention USING purged \ + WHERE mention.community_id = $1 AND mention.event_id = purged.id \ + ) \ + SELECT count(*) FROM purged" + ) + }; +} + +// The known-target statements below are executed in production and referenced +// directly by the pruning tests. They run with `.persistent(false)`: a cached +// prepared statement may switch to a generic plan after five executions, and on +// PostgreSQL 16 a generic plan takes `AccessShareLock` on every `events` +// partition (and its indexes) during plan validation, before runtime pruning. +// An unnamed statement is always planned with its bound parameters, so only the +// partition selected by `created_at` is locked. +const SELECT_KNOWN_EVENT_SQL: &str = "SELECT kind, channel_id, created_at FROM events \ + WHERE community_id = $1 AND id = $2 AND created_at = $3"; +const SOFT_DELETE_KNOWN_EVENT_SQL: &str = concat!( + "UPDATE events SET deleted_at = NOW() \ + WHERE community_id = $1 AND id = $2 AND created_at = $3 \ + AND deleted_at IS NULL AND NOT COALESCE(", + retention_free_event_predicate!(), + ", false)" +); +const PURGE_KNOWN_EVENT_SQL: &str = purge_retention_free_sql!("id = $2 AND created_at = $3"); + +/// Which live rows a deletion targets. +enum RetentionFreeTarget<'a> { + Event { + event_id: &'a [u8], + created_at: DateTime, + }, + Coordinate { + kind: i32, + pubkey: &'a [u8], + d_tag: &'a str, + created_at_or_before: DateTime, + }, +} + +/// Physically delete live retention-free rows matched by `target`, plus their +/// mention rows, in the caller's transaction. The generic soft delete that +/// follows excludes the same predicate, so a retention-free row is either +/// purged here or left live, never soft-deleted. Returns the number of events +/// removed. +async fn purge_retention_free_events( + tx: &mut AdmittedTx, + target: RetentionFreeTarget<'_>, +) -> Result { + let community_id = tx.community(); + // Migration 0011 fences NIP-RS hard deletes behind a transaction-local + // opt-in. The fence and this opt-in are removed together once the + // migration-only triggers are dropped. + sqlx::query("SELECT set_config('buzz.nip_rs_hard_delete', 'on', true)") + .execute(tx.conn()) + .await?; + + let purged: i64 = match target { + RetentionFreeTarget::Event { + event_id, + created_at, + } => { + sqlx::query_scalar(PURGE_KNOWN_EVENT_SQL) + .persistent(false) + .bind(community_id.as_uuid()) + .bind(event_id) + .bind(created_at) + .fetch_one(tx.conn()) + .await? + } + RetentionFreeTarget::Coordinate { + kind, + pubkey, + d_tag, + created_at_or_before, + } => { + sqlx::query_scalar(purge_retention_free_sql!( + "kind = $2 AND pubkey = $3 AND d_tag = $4 AND created_at <= $5" + )) + .bind(community_id.as_uuid()) + .bind(kind) + .bind(pubkey) + .bind(d_tag) + .bind(created_at_or_before) + .fetch_one(tx.conn()) + .await? + } + }; + + Ok(purged as u64) +} + +/// An event a NIP-09 (kind 5) or NIP-29 (kind 9005) deletion soft-deletes in +/// the deletion's own insert transaction, with the thread it belongs to. +#[derive(Clone, Copy, Debug)] +pub struct DeletionTarget<'a> { + /// The event to soft-delete. + pub event_id: &'a [u8], + /// The target's exact signed timestamp, not the deletion's timestamp. + pub created_at: DateTime, + /// Its direct parent, whose `reply_count` drops when it is deleted. + pub parent_event_id: Option<&'a [u8]>, + /// Its thread root, whose `descendant_count` drops when it is deleted. + pub root_event_id: Option<&'a [u8]>, +} + +/// Atomically soft-delete an event and decrement thread reply counters. +/// +/// Wraps the delete + counter update in a single transaction so a crash between +/// them cannot leave counters permanently inflated. Returns `Ok(true)` if the +/// event was deleted this call. +/// +/// When the target event is a kind-40100 (canvas) event, this function derives +/// the target's `kind` and `channel_id` from the database inside the same +/// transaction and acquires the same `(community, kind, channel)` advisory lock +/// used by [`super::event::insert_canvas_head_checked`] before the UPDATE. This prevents a +/// concurrent tagged write from observing a head that is simultaneously being +/// removed. The serialization invariant is owned entirely by this function; +/// callers do not classify the target kind. +pub async fn soft_delete_event_and_update_thread( + pool: &PgPool, + community_id: CommunityId, + event_id: &[u8], + parent_event_id: Option<&[u8]>, + root_event_id: Option<&[u8]>, +) -> Result { + let mut tx = crate::begin_community_event_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; + let deleted = soft_delete_event_and_update_thread_in_tx( + &mut tx, + event_id, + None, + parent_event_id, + root_event_id, + ) + .await?; + tx.commit().await?; + Ok(deleted) +} + +/// Transaction-scoped body of [`soft_delete_event_and_update_thread`]. +/// +/// Callers that must fence the delete with their own writes (e.g. the admin +/// action lease/marker) run this inside their transaction; the caller commits. +/// `None` discovers the target timestamp once; `Some` never falls back to ID-only. +pub(crate) async fn soft_delete_event_and_update_thread_in_tx( + tx: &mut AdmittedTx, + event_id: &[u8], + created_at: Option>, + parent_event_id: Option<&[u8]>, + root_event_id: Option<&[u8]>, +) -> Result { + let community_id = tx.community(); + use crate::store::replaceable::event_replacement_lock_key; + + // ID-only compatibility callers discover the immutable timestamp once. + // Planned targets already have it; every subsequent events-table access + // uses the exact partition key. Kind/channel remain canonical DB reads, + // never caller-supplied classification (including for deleted rows). + let target: Option<(i32, Option, DateTime)> = match created_at { + Some(created_at) => { + sqlx::query_as(SELECT_KNOWN_EVENT_SQL) + .persistent(false) + .bind(community_id.as_uuid()) + .bind(event_id) + .bind(created_at) + .fetch_optional(tx.conn()) + .await? + } + None => { + sqlx::query_as( + "SELECT kind, channel_id, created_at FROM events \ + WHERE community_id = $1 AND id = $2", + ) + .bind(community_id.as_uuid()) + .bind(event_id) + .fetch_optional(tx.conn()) + .await? + } + }; + let Some((kind, channel_id, created_at)) = target else { + return Ok(false); + }; + + // Relay-signed move removals are the source channel's only replay record. + if kind == 45011 { + return Err(DbError::InvalidData( + "artifact removal markers cannot be deleted".into(), + )); + } + + if let Some(channel_id) = channel_id { + if kind == KIND_CANVAS as i32 { + let lock_key = event_replacement_lock_key( + community_id, + kind, + &[], + Some(channel_id.as_bytes().as_slice()), + ); + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(lock_key) + .execute(tx.conn()) + .await?; + } + } + + let purged = if may_be_retention_free(kind) { + purge_retention_free_events( + tx, + RetentionFreeTarget::Event { + event_id, + created_at, + }, + ) + .await? + } else { + 0 + }; + let result = sqlx::query(SOFT_DELETE_KNOWN_EVENT_SQL) + .persistent(false) + .bind(community_id.as_uuid()) + .bind(event_id) + .bind(created_at) + .execute(tx.conn()) + .await?; + + let deleted = purged > 0 || result.rows_affected() > 0; + + if deleted { + sqlx::query("UPDATE artifact_heads SET modified_at=clock_timestamp() WHERE community_id=$1 AND artifact_id=(SELECT artifact_id FROM artifact_revisions WHERE community_id=$1 AND event_id=$2) AND event_id=$2") + .bind(community_id.as_uuid()).bind(event_id).execute(tx.conn()).await?; + if let Some(pid) = parent_event_id { + sqlx::query( + "UPDATE thread_metadata \ + SET reply_count = GREATEST(reply_count - 1, 0) \ + WHERE community_id = $1 AND event_id = $2", + ) + .bind(community_id.as_uuid()) + .bind(pid) + .execute(tx.conn()) + .await?; + + if let Some(root_id) = root_event_id { + sqlx::query( + "UPDATE thread_metadata \ + SET descendant_count = GREATEST(descendant_count - 1, 0) \ + WHERE community_id = $1 AND event_id = $2", + ) + .bind(community_id.as_uuid()) + .bind(root_id) + .execute(tx.conn()) + .await?; + } + } + } + + Ok(deleted) +} + +#[cfg(test)] +mod postgres_tests; diff --git a/crates/buzz-db/src/store/event_deletion/postgres_tests.rs b/crates/buzz-db/src/store/event_deletion/postgres_tests.rs new file mode 100644 index 0000000000..28cd85cccf --- /dev/null +++ b/crates/buzz-db/src/store/event_deletion/postgres_tests.rs @@ -0,0 +1,594 @@ +//! Functional coverage through admitted production deletion entry points. + +use super::*; +use crate::{event::ThreadMetadataParams, Db}; +use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; +use sqlx::postgres::PgPoolOptions; + +const JAN: u64 = 1_768_003_200; // 2026-01-10 UTC +const MAY: u64 = 1_778_371_200; // 2026-05-10 UTC + +async fn fixture() -> (Db, CommunityId, Uuid) { + let pool = PgPoolOptions::new() + .max_connections(1) + .connect(&crate::test_support::database_url()) + .await + .expect("test database"); + let db = Db::from_pool(pool); + let (community, channel) = tenant(&db.pool).await; + (db, community, channel) +} + +async fn tenant(pool: &PgPool) -> (CommunityId, Uuid) { + let community = Uuid::new_v4(); + let channel = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community) + .bind(format!("deletion-{}.example", community.simple())) + .execute(pool) + .await + .expect("community"); + sqlx::query( + "INSERT INTO channels (id, community_id, name, created_by) VALUES ($1,$2,'deletion',$3)", + ) + .bind(channel) + .bind(community) + .bind(vec![7_u8; 32]) + .execute(pool) + .await + .expect("channel"); + (CommunityId::from_uuid(community), channel) +} + +fn event(kind: u16, secs: u64, tags: Vec) -> nostr::Event { + EventBuilder::new(Kind::Custom(kind), "deletion test") + .tags(tags) + .custom_created_at(Timestamp::from(secs)) + .sign_with_keys(&Keys::generate()) + .expect("signed event") +} + +fn timestamp(event: &nostr::Event) -> DateTime { + DateTime::from_timestamp(event.created_at.as_secs() as i64, 0).expect("timestamp") +} + +fn target(event: &nostr::Event) -> DeletionTarget<'_> { + DeletionTarget { + event_id: event.id.as_bytes(), + created_at: timestamp(event), + parent_event_id: None, + root_event_id: None, + } +} + +async fn counts(db: &Db, community: CommunityId, id: &[u8]) -> (i64, i64, i64) { + let (rows, live): (i64, i64) = sqlx::query_as( + "SELECT count(*), count(*) FILTER (WHERE deleted_at IS NULL) FROM events WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()).bind(id).fetch_one(&db.pool).await.expect("event counts"); + let mentions = sqlx::query_scalar( + "SELECT count(*) FROM event_mentions WHERE community_id=$1 AND event_id=$2", + ) + .bind(community.as_uuid()) + .bind(id) + .fetch_one(&db.pool) + .await + .expect("mention counts"); + (rows, live, mentions) +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn known_targets_across_months_are_exact_and_tenant_scoped() { + let (db, community, channel) = fixture().await; + let (other, other_channel) = tenant(&db.pool).await; + let jan = event(7, JAN, vec![]); + let may = event(7, MAY, vec![]); + for ev in [&jan, &may] { + db.insert_event(community, ev, Some(channel)) + .await + .expect("insert"); + db.insert_event(other, ev, Some(other_channel)) + .await + .expect("other copy"); + } + // Deliberately older than both targets: e-target deletion has no inferred + // causal chronology, and the deletion's month is not a target key. + let deletion = event(5, JAN - 86_400, vec![]); + let missing = event(7, JAN + 1, vec![]); + let wrong = DeletionTarget { + created_at: timestamp(&may), + ..target(&jan) + }; + let (_, inserted, deleted) = db + .insert_deletion_event( + community, + &deletion, + Some(channel), + None, + &[wrong, target(&missing), target(&jan), target(&may)], + ) + .await + .expect("delete targets"); + assert!(inserted); + assert_eq!(deleted, [false, false, true, true]); + for ev in [&jan, &may] { + assert_eq!(counts(&db, community, ev.id.as_bytes()).await.1, 0); + assert_eq!(counts(&db, other, ev.id.as_bytes()).await.1, 1); + } + assert_eq!( + db.insert_deletion_event( + community, + &deletion, + Some(channel), + None, + &[target(&jan), target(&may)] + ) + .await + .expect("replay") + .2, + [false, false] + ); + // ID-only callers still work and discover the immutable timestamp in tx. + assert!( + soft_delete_event_and_update_thread(&db.pool, other, jan.id.as_bytes(), None, None) + .await + .expect("compatibility delete") + ); + assert!(!soft_delete_event_and_update_thread( + &db.pool, + other, + missing.id.as_bytes(), + None, + None + ) + .await + .expect("missing compatibility target")); +} + +async fn retention_free_contract() { + let (db, community, channel) = fixture().await; + let (other, other_channel) = tenant(&db.pool).await; + let slot = format!("read-state:{}", "b".repeat(32)); + let marker = event(45011, JAN, vec![]); + db.insert_event(community, &marker, Some(channel)) + .await + .expect("undeletable marker"); + for (kind, d, extra) in [ + (30078, slot.as_str(), ["t", "read-state"]), + ( + 30003, + "buzz-mesh-member-status:test", + ["k", "buzz-mesh-status"], + ), + ] { + for secs in [JAN, MAY] { + let tags = vec![ + Tag::parse(["d", d]).expect("d"), + Tag::parse(extra).expect("classification"), + Tag::parse(["p", &"c".repeat(64)]).expect("mention"), + ]; + let ev = event(kind, secs, tags); + for (tenant, ch) in [(community, channel), (other, other_channel)] { + db.replace_parameterized_event(tenant, &ev, d, Some(ch)) + .await + .expect("retention-free insert"); + assert_eq!(counts(&db, tenant, ev.id.as_bytes()).await, (1, 1, 1)); + } + let deletion = event(5, JAN - 1, vec![]); + let wrong = DeletionTarget { + created_at: timestamp(&ev) + chrono::Duration::seconds(1), + ..target(&ev) + }; + assert_eq!( + db.insert_deletion_event(community, &deletion, None, None, &[wrong]) + .await + .expect("wrong timestamp") + .2, + [false] + ); + assert_eq!(counts(&db, community, ev.id.as_bytes()).await, (1, 1, 1)); + // A later forbidden target must roll back both the physical purge + // and its mention cleanup, including the migration NIP-RS guard. + assert!(db + .insert_deletion_event( + community, + &deletion, + None, + None, + &[target(&ev), target(&marker)] + ) + .await + .is_err()); + assert_eq!(counts(&db, community, ev.id.as_bytes()).await, (1, 1, 1)); + assert_eq!( + db.insert_deletion_event(community, &deletion, None, None, &[target(&ev)]) + .await + .expect("hard purge") + .2, + [true] + ); + assert_eq!(counts(&db, community, ev.id.as_bytes()).await, (0, 0, 0)); + assert_eq!(counts(&db, other, ev.id.as_bytes()).await, (1, 1, 1)); + assert_eq!( + db.insert_deletion_event(community, &deletion, None, None, &[target(&ev)]) + .await + .expect("replay purge") + .2, + [false] + ); + assert!(soft_delete_event_and_update_thread( + &db.pool, + other, + ev.id.as_bytes(), + None, + None + ) + .await + .expect("ID-only purge")); + assert_eq!(counts(&db, other, ev.id.as_bytes()).await, (0, 0, 0)); + } + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn known_target_retention_free_purge_cleans_mentions() { + retention_free_contract().await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_known_target_retention_free_purge_cleans_mentions() { + crate::migration::run_migrations( + &PgPool::connect(&crate::test_support::database_url()) + .await + .expect("migration pool"), + ) + .await + .expect("migrate"); + retention_free_contract().await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn known_target_batch_rollback_then_replay_repairs_and_counts_once() { + let (db, community, channel) = fixture().await; + let root = event(9, JAN, vec![]); + db.insert_event_with_thread_metadata( + community, + &root, + Some(channel), + Some(ThreadMetadataParams { + event_id: root.id.as_bytes(), + event_created_at: timestamp(&root), + channel_id: channel, + parent_event_id: None, + parent_event_created_at: None, + root_event_id: None, + root_event_created_at: None, + depth: 0, + broadcast: false, + }), + ) + .await + .expect("root"); + let reply = event(9, MAY, vec![]); + db.insert_event_with_thread_metadata( + community, + &reply, + Some(channel), + Some(ThreadMetadataParams { + event_id: reply.id.as_bytes(), + event_created_at: timestamp(&reply), + channel_id: channel, + parent_event_id: Some(root.id.as_bytes()), + parent_event_created_at: Some(timestamp(&root)), + root_event_id: Some(root.id.as_bytes()), + root_event_created_at: Some(timestamp(&root)), + depth: 1, + broadcast: false, + }), + ) + .await + .expect("reply"); + let counter = || async { + sqlx::query_as::<_, (i32, i32)>("SELECT reply_count, descendant_count FROM thread_metadata WHERE community_id=$1 AND event_id=$2") + .bind(community.as_uuid()).bind(root.id.as_bytes().as_slice()) + .fetch_one(&db.pool).await.expect("counters") + }; + assert_eq!(counter().await, (1, 1)); + // A forbidden removal marker makes the second target fail after the first + // was mutated. No test-only triggers or copied deletion implementation. + let marker = event(45011, JAN, vec![]); + db.insert_event(community, &marker, Some(channel)) + .await + .expect("marker"); + let reply_target = DeletionTarget { + parent_event_id: Some(root.id.as_bytes()), + root_event_id: Some(root.id.as_bytes()), + ..target(&reply) + }; + let deletion = event(5, MAY + 1, vec![]); + assert!(db + .insert_deletion_event( + community, + &deletion, + Some(channel), + None, + &[reply_target, target(&marker)] + ) + .await + .is_err()); + assert_eq!(counts(&db, community, deletion.id.as_bytes()).await.0, 0); + assert_eq!(counts(&db, community, reply.id.as_bytes()).await.1, 1); + assert_eq!(counter().await, (1, 1)); + // Emulate a historical torn write, then concurrently replay it from two + // independent connections: exactly one transition decrements counters. + db.insert_event(community, &deletion, Some(channel)) + .await + .expect("torn deletion"); + let concurrent = Db::from_pool( + PgPool::connect(&crate::test_support::database_url()) + .await + .expect("independent pool"), + ); + let targets = [reply_target]; + let (a, b) = tokio::join!( + db.insert_deletion_event(community, &deletion, Some(channel), None, &targets), + concurrent.insert_deletion_event(community, &deletion, Some(channel), None, &targets), + ); + let (_, inserted_a, a) = a.expect("first replay"); + let (_, inserted_b, b) = b.expect("second replay"); + assert!(!inserted_a && !inserted_b); + assert_ne!(a, b); + assert_eq!(counter().await, (0, 0)); + assert_eq!( + db.insert_deletion_event(community, &deletion, Some(channel), None, &targets) + .await + .expect("repeat replay") + .2, + [false] + ); + assert_eq!(counter().await, (0, 0)); +} + +fn event_relations(plan: &serde_json::Value, relations: &mut Vec) { + if let Some(name) = plan.get("Relation Name").and_then(|name| name.as_str()) { + // Generic plans can list nonexecuted target tables / subplans. Only + // executed relation-access nodes measure runtime pruning. + if name.starts_with("events_p") + && plan + .get("Actual Loops") + .and_then(|loops| loops.as_u64()) + .unwrap_or(0) + > 0 + { + relations.push(name.to_owned()); + } + } + match plan { + serde_json::Value::Array(items) => { + for item in items { + event_relations(item, relations); + } + } + serde_json::Value::Object(items) => { + for item in items.values() { + event_relations(item, relations); + } + } + _ => {} + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn production_known_target_statements_prune_to_one_month() { + let (db, community, channel) = fixture().await; + let slot = format!("read-state:{}", "a".repeat(32)); + let ev = event( + 30078, + JAN, + vec![ + Tag::parse(["d", &slot]).expect("d"), + Tag::parse(["t", "read-state"]).expect("t"), + ], + ); + db.replace_parameterized_event(community, &ev, &slot, Some(channel)) + .await + .expect("insert"); + let deletion = event(5, MAY, vec![]); + db.insert_deletion_event(community, &deletion, None, None, &[target(&ev)]) + .await + .expect("exercise production statements"); + // Explain the exact statements production executes, with their real + // bindings. Production runs them unnamed (`.persistent(false)`), so they + // are planned with the bound parameters; the lock test below covers the + // generic-plan hazard of a cached prepared statement. ANALYZE executes the + // real statement, but every EXPLAIN is rolled back. + for statement in [ + SELECT_KNOWN_EVENT_SQL, + SOFT_DELETE_KNOWN_EVENT_SQL, + PURGE_KNOWN_EVENT_SQL, + ] { + assert!( + statement.contains("created_at = $3"), + "missing partition key: {statement}" + ); + let mut tx = crate::begin_community_event_write_transaction( + &db.pool, + community, + crate::observability::WriterOperation::EventWrite, + ) + .await + .expect("admit explain"); + sqlx::query("SET LOCAL plan_cache_mode = force_custom_plan") + .execute(tx.conn()) + .await + .expect("plan mode"); + sqlx::query(sqlx::AssertSqlSafe(format!( + "PREPARE deletion_pruning_probe (uuid, bytea, timestamptz) AS {statement}" + ))) + .execute(tx.conn()) + .await + .expect("prepare production SQL"); + let explain = format!( + "EXPLAIN (ANALYZE, FORMAT JSON) EXECUTE deletion_pruning_probe ('{}', decode('{}', 'hex'), '{}')", + community.as_uuid(), ev.id.to_hex(), timestamp(&ev).to_rfc3339() + ); + let plan: serde_json::Value = sqlx::query_scalar(sqlx::AssertSqlSafe(explain)) + .fetch_one(tx.conn()) + .await + .expect("production EXPLAIN"); + let mut relations = vec![]; + event_relations(&plan, &mut relations); + assert!(!relations.is_empty(), "no event access: {plan}"); + assert!( + relations.iter().all(|name| name == "events_p2026_01"), + "unrelated monthly partitions: {plan}" + ); + sqlx::query("DEALLOCATE deletion_pruning_probe") + .execute(tx.conn()) + .await + .expect("deallocate probe"); + tx.rollback().await.expect("rollback explain"); + } +} + +/// Monthly `events` partitions (not their indexes) this backend holds a +/// relation lock on right now. +async fn locked_event_partitions(tx: &mut AdmittedTx) -> Vec { + sqlx::query_scalar( + "SELECT DISTINCT c.relname::text FROM pg_locks l \ + JOIN pg_class c ON c.oid = l.relation \ + WHERE l.locktype = 'relation' AND l.pid = pg_backend_pid() \ + AND c.relkind = 'r' AND c.relname LIKE 'events\\_p%' ORDER BY 1", + ) + .fetch_all(tx.conn()) + .await + .expect("observe partition locks") +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn production_known_target_deletion_locks_one_partition() { + let (db, community, channel) = fixture().await; + let slot = format!("read-state:{}", "b".repeat(32)); + let ev = event( + 30078, + JAN, + vec![ + Tag::parse(["d", &slot]).expect("d"), + Tag::parse(["t", "read-state"]).expect("t"), + ], + ); + db.replace_parameterized_event(community, &ev, &slot, Some(channel)) + .await + .expect("insert"); + // The assertion is only meaningful if other partitions exist to be locked. + let partitions: i64 = + sqlx::query_scalar("SELECT count(*) FROM pg_inherits WHERE inhparent = 'events'::regclass") + .fetch_one(&db.pool) + .await + .expect("partition count"); + assert!(partitions > 1, "need several partitions, got {partitions}"); + + // The fixture pool has one connection, so every round reuses its + // statement cache. A persistent prepared statement is custom-planned for + // its first five executions and may then flip to a generic plan, which + // locks every partition during plan validation; the production statements + // are unnamed and must lock only the target month in every round. + for round in 0..10 { + let mut tx = crate::begin_community_event_write_transaction( + &db.pool, + community, + crate::observability::WriterOperation::EventWrite, + ) + .await + .expect("admit deletion"); + // Known-target select, purge and soft delete all run on the production + // path (read-state events are retention-free). + let deleted = soft_delete_event_and_update_thread_in_tx( + &mut tx, + ev.id.as_bytes(), + Some(timestamp(&ev)), + None, + None, + ) + .await + .expect("production deletion"); + assert!(deleted, "round {round}"); + let locked = locked_event_partitions(&mut tx).await; + // Roll back so the next round deletes the same row again. + tx.rollback().await.expect("rollback"); + assert_eq!( + locked, + ["events_p2026_01"], + "round {round}: known-target deletion must lock only the target month" + ); + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn known_canvas_target_uses_canonical_replacement_lock() { + let (db, community, channel) = fixture().await; + let canvas = event(KIND_CANVAS as u16, JAN, vec![]); + db.insert_event(community, &canvas, Some(channel)) + .await + .expect("canvas"); + // The fixture's single connection identifies the backend that will delete. + let deleter_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&db.pool) + .await + .expect("deleter pid"); + let observer = PgPool::connect(&crate::test_support::database_url()) + .await + .expect("lock pool"); + let mut holder = observer.begin().await.expect("holder transaction"); + let key = crate::store::replaceable::event_replacement_lock_key( + community, + KIND_CANVAS as i32, + &[], + Some(channel.as_bytes().as_slice()), + ); + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(key) + .execute(&mut *holder) + .await + .expect("hold replacement key"); + let delete_db = Db::from_pool(db.pool.clone()); + let id = canvas.id.to_bytes(); + let delete = tokio::spawn(async move { + let deletion = event(5, MAY, vec![]); + delete_db + .insert_deletion_event( + community, + &deletion, + Some(channel), + None, + &[target(&canvas)], + ) + .await + }); + // Observe an actual advisory-lock wait, not merely a task that hasn't yet + // been scheduled. Removing the canonical lock makes this test fail. + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let waiting: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM pg_locks WHERE pid=$1 AND locktype='advisory' AND NOT granted)") + .bind(deleter_pid).fetch_one(&mut *holder).await.expect("observe lock wait"); + if waiting { break; } + assert!(!delete.is_finished(), "known canvas deletion bypassed replacement lock"); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }).await.expect("deletion must wait on replacement lock"); + holder.rollback().await.expect("release replacement key"); + let result = tokio::time::timeout(std::time::Duration::from_secs(5), delete) + .await + .expect("delete completes") + .expect("join") + .expect("delete"); + assert_eq!(result.2, [true]); + assert_eq!(counts(&db, community, &id).await.1, 0); +} diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index b5578cb796..93841f7ba8 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -24,6 +24,8 @@ pub mod dm; pub mod entity_deletion; /// Event storage and retrieval. pub mod event; +/// Atomic event deletion and retention-free purging. +mod event_deletion; /// App-owned push enqueue and channel TTL refresh for inserted events. pub(crate) mod event_follow_up; /// Per-community NIP-FI issuer bindings. diff --git a/crates/buzz-db/src/store/relay_admin_actions.rs b/crates/buzz-db/src/store/relay_admin_actions.rs index f1fa4af9be..fc2fb32c98 100644 --- a/crates/buzz-db/src/store/relay_admin_actions.rs +++ b/crates/buzz-db/src/store/relay_admin_actions.rs @@ -814,13 +814,13 @@ pub async fn execute_delete_with_marker( // NIP-AR: an artifact revision leaves current state only through its own // `op=delete` or a kind-9005 redaction. Both are fanned out, so open syncs // see them; this delete would retire a head with no live signal. - let kind: Option = - sqlx::query_scalar("SELECT kind FROM events WHERE community_id = $1 AND id = $2") + let target: Option<(i32, chrono::DateTime)> = + sqlx::query_as("SELECT kind, created_at FROM events WHERE community_id = $1 AND id = $2") .bind(community_id.as_uuid()) .bind(target_event_id) .fetch_optional(tx.conn()) .await?; - if kind == Some(45010) { + if target.is_some_and(|(kind, _)| kind == 45010) { return Err(crate::error::DbError::InvalidData( "artifact revisions cannot be deleted by admin action; use op=delete or kind 9005 redaction" .into(), @@ -830,13 +830,16 @@ pub async fn execute_delete_with_marker( // Canonical delete + thread_metadata counters, fenced by this transaction. // Counters move only when the row transitions to deleted, so a second // action against an already-deleted target leaves them unchanged. - crate::event::soft_delete_event_and_update_thread_in_tx( - &mut tx, - target_event_id, - parent_event_id, - root_event_id, - ) - .await?; + if let Some((_, created_at)) = target { + crate::event::soft_delete_event_and_update_thread_in_tx( + &mut tx, + target_event_id, + Some(created_at), + parent_event_id, + root_event_id, + ) + .await?; + } let marker = sqlx::query( r#" diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 0740206c2a..23aaaf5e97 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -3487,12 +3487,19 @@ pub(crate) async fn persist_deletion( }; let targets: Vec> = planned .iter() - .map(|p| buzz_db::DeletionTarget { - event_id: &p.id, - parent_event_id: p.parent_id.as_deref(), - root_event_id: p.root_id.as_deref(), + .map(|p| { + Ok(buzz_db::DeletionTarget { + event_id: &p.id, + created_at: chrono::DateTime::from_timestamp( + p.target.event.created_at.as_secs() as i64, + 0, + ) + .ok_or_else(|| anyhow::anyhow!("invalid deletion target timestamp"))?, + parent_event_id: p.parent_id.as_deref(), + root_event_id: p.root_id.as_deref(), + }) }) - .collect(); + .collect::>()?; let (stored, inserted, deleted) = state .db .insert_deletion_event(tenant.community(), event, channel_id, thread_meta, &targets)