From 0585608f828e95bdbe1d96d2ba83f244983b4825 Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Fri, 9 Oct 2026 09:20:01 -0400 Subject: [PATCH] fix(db): classify superseded rows by stored shape; bound 0058 lock Replacement now removes superseded versions through the same remove_coordinate_versions_in_tx primitive as NIP-09 coordinate deletion, so each row is purged or soft-deleted by its own shape rather than the incoming event's. 0058 sets lock_timeout before its DROPs. Tests pin mixed-shape replacement, run the NIP-09 classification table through replacement, delete through the production path in the NIP-RS replay test, and run the trigger parity probe through run_migrations. Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/src/runtime/migration.rs | 9 +- crates/buzz-db/src/store/event.rs | 76 +++--- crates/buzz-db/src/store/replaceable.rs | 239 ++++++++++++++---- ...58_drop_nip_rs_mesh_migration_triggers.sql | 4 + 4 files changed, 237 insertions(+), 91 deletions(-) diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 237dda2bdd..38c164cf32 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -747,6 +747,7 @@ mod postgres_tests { assert_eq!(migrations[65].version, 66); let retire_nip_rs = strip_sql_comments(migrations[65].sql.as_str()); for statement in [ + "SET LOCAL lock_timeout = '5s';", "DROP TRIGGER trg_events_nip_rs_watermark ON events;", "DROP TRIGGER trg_events_guard_nip_rs_hard_delete ON events;", "DROP TRIGGER trg_events_purge_soft_deleted_nip_rs ON events;", @@ -763,6 +764,11 @@ mod postgres_tests { "0066 must run {statement}" ); } + assert!( + retire_nip_rs.find("SET LOCAL lock_timeout").unwrap() + < retire_nip_rs.find("DROP TRIGGER").unwrap(), + "0058 must bound its ACCESS EXCLUSIVE lock on events before the first DROP" + ); assert_eq!(migrations[55].version, 56); assert!(migrations[55] .sql @@ -2806,8 +2812,7 @@ mod postgres_tests { // Every migration-only trigger on the event tables is retired, so a // fully migrated database carries exactly the desired-state triggers. - MIGRATOR - .run(&migrated) + run_migrations(&migrated) .await .expect("apply remaining migrations"); for table in ["events", "event_mentions"] { diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 2bdf9dfbbd..ffcbd52366 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -1190,8 +1190,8 @@ macro_rules! mesh_status_event_predicate { /// historical value: NIP-RS read state and Buzz mesh heartbeats. Their /// deletion is physical, including the mention index, so soft-deleted payloads /// never accumulate. These predicates are the only definition of the class: -/// the replace path asks PostgreSQL through [`classify_retention_free`] rather -/// than re-implementing them in Rust. +/// every removal path decides row by row through +/// [`remove_coordinate_versions_in_tx`] rather than re-implementing them in Rust. macro_rules! retention_free_event_predicate { () => { concat!( @@ -1210,38 +1210,21 @@ fn may_be_retention_free(kind: i32) -> bool { kind == 30078 || kind == 30003 } -/// Which retention-free class an event belongs to. -#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] -pub(crate) struct RetentionFreeClass { - /// A conforming NIP-RS read-state coordinate. - pub(crate) nip_rs: bool, - /// A Buzz mesh heartbeat. - pub(crate) mesh_status: bool, -} - -impl RetentionFreeClass { - /// Superseded and deleted rows of this event are removed physically. - pub(crate) fn is_retention_free(self) -> bool { - self.nip_rs || self.mesh_status - } -} - -/// Classify an event that has not been written yet, using the same SQL -/// predicates as the purge and delete paths so the two cannot drift. -pub(crate) async fn classify_retention_free( +/// Whether an event that has not been written yet is a conforming NIP-RS +/// read-state coordinate, which advances the stale-write watermark. Uses the +/// same SQL predicate as the purge and delete paths so the two cannot drift. +pub(crate) async fn is_nip_rs_event( conn: &mut sqlx::PgConnection, kind: i32, d_tag: &str, tags: &serde_json::Value, -) -> Result { - if !may_be_retention_free(kind) { - return Ok(RetentionFreeClass::default()); +) -> Result { + if kind != 30078 { + return Ok(false); } - let (nip_rs, mesh_status): (bool, bool) = sqlx::query_as(concat!( + let nip_rs: bool = sqlx::query_scalar(concat!( "SELECT COALESCE(", nip_rs_event_predicate!(), - ", false), COALESCE(", - mesh_status_event_predicate!(), ", false) \ FROM (SELECT $1::integer AS kind, $2::text AS d_tag, $3::jsonb AS tags) event_row" )) @@ -1250,10 +1233,7 @@ pub(crate) async fn classify_retention_free( .bind(tags) .fetch_one(conn) .await?; - Ok(RetentionFreeClass { - nip_rs, - mesh_status, - }) + Ok(nip_rs) } /// Soft-delete the live row for an addressable coordinate @@ -1301,14 +1281,35 @@ pub async fn soft_delete_by_coordinate( crate::observability::WriterOperation::EventWrite, ) .await?; + let deleted = + remove_coordinate_versions_in_tx(&mut tx, kind, pubkey, d_tag, deletion_created_at).await?; + tx.commit().await?; + Ok(deleted) +} + +/// Remove every live version of `(kind, pubkey, d_tag)` created at or before +/// `created_at_or_before`, in the caller's transaction. Each row is classified +/// by its own stored shape: retention-free rows are purged physically with +/// their mentions, every other row is soft-deleted. NIP-09 coordinate deletion +/// and parameterized replacement both remove versions through this function. +/// +/// Returns `true` if any row was removed. +pub(crate) async fn remove_coordinate_versions_in_tx( + tx: &mut AdmittedTx, + kind: i32, + pubkey: &[u8], + d_tag: &str, + created_at_or_before: DateTime, +) -> Result { + let community_id = tx.community(); let purged = if may_be_retention_free(kind) { purge_retention_free_events( - &mut tx, + tx, RetentionFreeTarget::Coordinate { kind, pubkey, d_tag, - created_at_or_before: deletion_created_at, + created_at_or_before, }, ) .await? @@ -1319,8 +1320,9 @@ pub async fn soft_delete_by_coordinate( // 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. + // first" outcome documented on [`soft_delete_by_coordinate`]). COALESCE + // keeps a NULL predicate (e.g. NULL `d_tag`) on the ordinary soft-delete + // path. 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 \ @@ -1332,12 +1334,10 @@ pub async fn soft_delete_by_coordinate( .bind(kind) .bind(pubkey) .bind(d_tag) - .bind(deletion_created_at) + .bind(created_at_or_before) .execute(tx.conn()) .await?; - tx.commit().await?; - Ok(purged > 0 || result.rows_affected() > 0) } diff --git a/crates/buzz-db/src/store/replaceable.rs b/crates/buzz-db/src/store/replaceable.rs index a9306c06d2..5254c85604 100644 --- a/crates/buzz-db/src/store/replaceable.rs +++ b/crates/buzz-db/src/store/replaceable.rs @@ -132,10 +132,7 @@ async fn replace_parameterized_event_in_transaction_impl( .await?; let tags_json = serde_json::to_value(&event.tags)?; - let class = - crate::event::classify_retention_free(tx.conn(), kind_i32, d_tag, &tags_json).await?; - let is_nip_rs = class.nip_rs; - let hard_delete_superseded = class.is_retention_free(); + let is_nip_rs = crate::event::is_nip_rs_event(tx.conn(), kind_i32, d_tag, &tags_json).await?; let existing: Option<(DateTime, Vec)> = sqlx::query_as( "SELECT created_at, id FROM events \ @@ -231,30 +228,18 @@ async fn replace_parameterized_event_in_transaction_impl( .await?; let written: Result = async { if existing.is_some() { - let statement = if hard_delete_superseded { - "DELETE FROM events \ - WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL" - } else { - "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" - }; - sqlx::query(statement) - .bind(community_id.as_uuid()) - .bind(kind_i32) - .bind(pubkey_bytes.as_slice()) - .bind(d_tag) - .execute(tx.conn()) - .await?; - - if hard_delete_superseded { - if let Some((_, existing_id)) = &existing { - sqlx::query("DELETE FROM event_mentions WHERE community_id = $1 AND event_id = $2") - .bind(community_id.as_uuid()) - .bind(existing_id) - .execute(tx.conn()) - .await?; - } - } + // Every live version is at or before the incoming event (dominance + // was checked above). Each is removed by its own stored shape, so + // a retention-free head superseded by a nonconforming event is + // still purged, and a nonconforming head keeps legacy retention. + crate::event::remove_coordinate_versions_in_tx( + tx, + kind_i32, + pubkey_bytes.as_slice(), + d_tag, + created_at, + ) + .await?; } let sig_bytes = event.sig.serialize(); @@ -962,32 +947,46 @@ mod postgres_tests { .expect("count NIP-RS rows"); assert_eq!(rows, 1, "superseded payload must be physically deleted"); - sqlx::query( - "UPDATE events SET deleted_at=NOW() WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3", - ) - .bind(community.as_uuid()) - .bind(keys.public_key().to_bytes()) - .bind(&d_tag) - .execute(&db.pool) - .await - .expect("simulate NIP-09 coordinate deletion"); - assert!( - !db.replace_parameterized_event(community, &old, &d_tag, None) - .await - .expect("replay old") - .1 + event::soft_delete_by_coordinate( + &db.pool, + community, + i32::from(buzz_core::kind::KIND_READ_STATE as u16), + &keys.public_key().to_bytes(), + &d_tag, + (base + 1) as i64, + ) + .await + .expect("NIP-09 coordinate deletion"), + "coordinate deletion must remove the live head" ); - let live: i64 = sqlx::query_scalar( - "SELECT count(*) FROM events WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3 AND deleted_at IS NULL", + + // Neither the superseded head nor an exact replay of the deleted head + // may come back: the watermark holds the deleted head's (created_at, id). + for (label, replay) in [("old", &old), ("deleted head", &new)] { + assert!( + !db.replace_parameterized_event(community, replay, &d_tag, None) + .await + .expect("replay") + .1, + "replay of {label} must be rejected" + ); + } + let (rows, live): (i64, i64) = sqlx::query_as( + "SELECT count(*), count(*) FILTER (WHERE deleted_at IS NULL) FROM events \ + WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3", ) .bind(community.as_uuid()) .bind(keys.public_key().to_bytes()) .bind(&d_tag) .fetch_one(&db.pool) .await - .expect("count live NIP-RS rows"); - assert_eq!(live, 0, "watermark must block stale resurrection"); + .expect("count NIP-RS rows"); + assert_eq!( + (rows, live), + (0, 0), + "deletion is physical and the watermark blocks resurrection" + ); } #[tokio::test] @@ -1595,10 +1594,130 @@ mod postgres_tests { } } - /// NIP-09 deletion of a retention-free coordinate (NIP-RS read state, mesh - /// status) is physical in application code; no database trigger takes part. - /// Ordinary addressable events, and look-alikes that fail classification, - /// keep the soft-delete contract. + /// Replacement removes each superseded row by its own stored shape, not by + /// the incoming event's: a conforming read-state head superseded by a + /// nonconforming event is still purged with its mentions, and a + /// nonconforming head superseded by a conforming event keeps legacy + /// retention. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn mixed_shape_replacement_classifies_each_superseded_row() { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + let db = setup_db().await; + let community = CommunityId::from_uuid(make_community(&db.pool).await); + let base = Timestamp::now().as_secs(); + let conforming = |d_tag: &str| { + vec![ + Tag::parse(["d", d_tag]).expect("d tag"), + Tag::parse(["t", "read-state"]).expect("t tag"), + ] + }; + let duplicate_t = |d_tag: &str| { + vec![ + Tag::parse(["d", d_tag]).expect("d tag"), + Tag::parse(["t", "read-state"]).expect("first t tag"), + Tag::parse(["t", "read-state"]).expect("second t tag"), + ] + }; + + for (case, old_tags, new_tags, expected_old) in [ + ( + "conforming then nonconforming", + conforming as fn(&str) -> Vec, + duplicate_t as fn(&str) -> Vec, + (0, 0, 0), + ), + ( + "nonconforming then conforming", + duplicate_t as fn(&str) -> Vec, + conforming as fn(&str) -> Vec, + (1, 0, 1), + ), + ] { + let keys = Keys::generate(); + let d_tag = format!("read-state:{}", "f".repeat(32)); + let sign = |content: &str, tags: Vec, at: u64| { + EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), + content, + ) + .tags(tags) + .custom_created_at(Timestamp::from(at)) + .sign_with_keys(&keys) + .expect("sign event") + }; + let old = sign("old", old_tags(&d_tag), base); + let new = sign("new", new_tags(&d_tag), base + 1); + + assert!( + db.replace_parameterized_event(community, &old, &d_tag, None) + .await + .expect("insert old") + .1, + "{case}: old head stored" + ); + sqlx::query( + "INSERT INTO event_mentions \ + (community_id, pubkey_hex, event_id, event_created_at, channel_id, event_kind) \ + SELECT community_id, $3, id, created_at, NULL, kind \ + FROM events WHERE community_id = $1 AND id = $2", + ) + .bind(community.as_uuid()) + .bind(old.id.as_bytes().as_slice()) + .bind("c".repeat(64)) + .execute(&db.pool) + .await + .expect("index mention"); + assert!( + db.replace_parameterized_event(community, &new, &d_tag, None) + .await + .expect("replace") + .1, + "{case}: newer head replaces" + ); + + 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(old.id.as_bytes().as_slice()) + .fetch_one(&db.pool) + .await + .expect("count old rows"); + let mentions: i64 = sqlx::query_scalar( + "SELECT count(*) FROM event_mentions WHERE community_id=$1 AND event_id=$2", + ) + .bind(community.as_uuid()) + .bind(old.id.as_bytes().as_slice()) + .fetch_one(&db.pool) + .await + .expect("count old mentions"); + assert_eq!( + (rows, live, mentions), + expected_old, + "{case}: superseded row classified by its own shape" + ); + let coordinate_live: i64 = sqlx::query_scalar( + "SELECT count(*) FROM events WHERE community_id=$1 AND kind=30078 \ + AND pubkey=$2 AND d_tag=$3 AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(keys.public_key().to_bytes()) + .bind(&d_tag) + .fetch_one(&db.pool) + .await + .expect("count live heads"); + assert_eq!(coordinate_live, 1, "{case}: exactly the newer head is live"); + } + } + + /// NIP-09 deletion and replacement of a retention-free coordinate (NIP-RS + /// read state, mesh status) are physical in application code; no database + /// trigger takes part. Ordinary addressable events, and look-alikes that + /// fail classification, keep the soft-delete contract. Every removal path + /// classifies through the same SQL predicate, so one table pins all three. #[tokio::test] #[ignore = "requires Postgres"] async fn nip09_deletion_purges_retention_free_rows_and_mentions() { @@ -1620,6 +1739,9 @@ mod postgres_tests { enum Delete { Coordinate, Id, + /// Supersede through `replace_parameterized_event` with a newer + /// event of the same shape at the same coordinate. + Replace, } let db = setup_db().await; @@ -1838,10 +1960,25 @@ mod postgres_tests { }; for (label, kind, d_tag, tags, purged) in cases { - for mode in [Delete::Coordinate, Delete::Id] { + for mode in [Delete::Coordinate, Delete::Id, Delete::Replace] { let (keys, id, created_at) = store(label, kind, d_tag.clone(), tags.clone()).await; + let newer = EventBuilder::new(Kind::Custom(kind), format!("{label}-newer")) + .tags( + tags.iter() + .map(|tag| Tag::parse(tag.clone()).expect("tag")) + .collect::>(), + ) + .custom_created_at(Timestamp::from(created_at + 1)) + .sign_with_keys(&keys) + .expect("sign newer event"); let delete = || async { match mode { + Delete::Replace => { + db.replace_parameterized_event(community, &newer, &d_tag, None) + .await + .expect("replace") + .1 + } Delete::Coordinate => event::soft_delete_by_coordinate( &db.pool, community, diff --git a/migrations/0058_drop_nip_rs_mesh_migration_triggers.sql b/migrations/0058_drop_nip_rs_mesh_migration_triggers.sql index ce4afa4c43..cac8d46653 100644 --- a/migrations/0058_drop_nip_rs_mesh_migration_triggers.sql +++ b/migrations/0058_drop_nip_rs_mesh_migration_triggers.sql @@ -13,6 +13,10 @@ -- without them. Dropping a trigger on the partitioned parent also drops its -- partition clones. Published migrations 0009, 0010, 0011 and 0019 stay -- checksum-frozen. +-- +-- DROP TRIGGER takes ACCESS EXCLUSIVE on events and every partition; fail the +-- deployment rather than queue relay reads and writes behind a long holder. +SET LOCAL lock_timeout = '5s'; DROP TRIGGER trg_events_nip_rs_watermark ON events; DROP TRIGGER trg_events_guard_nip_rs_hard_delete ON events;