jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-02 15:31:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(db): classify superseded rows by stored shape; bound 0058 lock

Browse Source 
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>
This commit is contained in:
Elrond
2026-10-09 09:20:01 -04:00
parent 9861fbd5af
commit 0585608f82
4 changed files with 237 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
crates/buzz-db/src/runtime/migration.rs
View File

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

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

@@ -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<RetentionFreeClass> {
if !may_be_retention_free(kind) {
return Ok(RetentionFreeClass::default());
) -> Result<bool> {
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<Utc>,
) -> Result<bool> {
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)
}

239
crates/buzz-db/src/store/replaceable.rs
View File

@@ -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<Utc>, Vec<u8>)> = 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<bool> = 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<Tag>,
duplicate_t as fn(&str) -> Vec<Tag>,
(0, 0, 0),
),
(
"nonconforming then conforming",
duplicate_t as fn(&str) -> Vec<Tag>,
conforming as fn(&str) -> Vec<Tag>,
(1, 0, 1),
),
] {
let keys = Keys::generate();
let d_tag = format!("read-state:{}", "f".repeat(32));
let sign = |content: &str, tags: Vec<Tag>, 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::<Vec<_>>(),
)
.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,

4
migrations/0058_drop_nip_rs_mesh_migration_triggers.sql
View File

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