jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-07 09:52:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(db): use exact event timestamps for transactional deletion (#9130)

Browse Source 
## 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 <jemiah@squareup.com>
Co-authored-by: Son of Anton <8e0ef2ab02d69a7ac089ef5eca90d1a3fbbdc546f6322dc34d0c4f3deecea7f4@buzz.block.builderlab.xyz>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Jemiah Westerman
2026-11-05 15:42:41 -08:00
committed by GitHub
parent b6d4403149
commit 0cfa50ac70
6 changed files with 1037 additions and 353 deletions
Download Patch File Download Diff File Expand all files Collapse all files

348
crates/buzz-db/src/store/event.rs
View File

@@ -1646,344 +1646,11 @@ fn count_events_sql(q: &EventQuery) -> QueryBuilder<sqlx::Postgres> {
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<bool> {
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(<predicate>, 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<Utc>,
},
}
/// 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<u64> {
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<bool> {
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<bool> {
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<Uuid>)> = 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,
}];

406
crates/buzz-db/src/store/event_deletion.rs Normal file
View File

@@ -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<bool> {
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(<predicate>, 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<Utc>,
},
Coordinate {
kind: i32,
pubkey: &'a [u8],
d_tag: &'a str,
created_at_or_before: DateTime<Utc>,
},
}
/// 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<u64> {
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<Utc>,
/// 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<bool> {
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<DateTime<Utc>>,
parent_event_id: Option<&[u8]>,
root_event_id: Option<&[u8]>,
) -> Result<bool> {
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<Uuid>, DateTime<Utc>)> = 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;

594
crates/buzz-db/src/store/event_deletion/postgres_tests.rs Normal file
View File

@@ -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<Tag>) -> 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<Utc> {
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<String>) {
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<String> {
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);
}

2
crates/buzz-db/src/store/mod.rs
View File

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

23
crates/buzz-db/src/store/relay_admin_actions.rs
View File

@@ -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<i32> =
sqlx::query_scalar("SELECT kind FROM events WHERE community_id = $1 AND id = $2")
let target: Option<(i32, chrono::DateTime<chrono::Utc>)> =
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#"

17
crates/buzz-relay/src/handlers/side_effects.rs
View File

@@ -3487,12 +3487,19 @@ pub(crate) async fn persist_deletion(
};
let targets: Vec<buzz_db::DeletionTarget<'_>> = 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::<anyhow::Result<_>>()?;
let (stored, inserted, deleted) = state
.db
.insert_deletion_event(tenant.community(), event, channel_id, thread_meta, &targets)