jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 04:30:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(db): validate channel roster snapshots in the app writer (#8316)

Browse Source 
## Summary

Phase 1 of retiring the channel roster snapshot trigger
(`trg_events_guard_channel_roster_snapshot`, migration 0032).

Every admitted writer that can store a kind 39002 roster snapshot now
runs the same check the trigger runs, on the same transaction and under
the same lock:

- `LockedMemberSnapshot::replace_member_event` (relay's live roster
publisher) validates the snapshot it already holds the membership lock
for, before soft-deleting the old head.
- `insert_event_on` and `insert_event_with_thread_metadata_tx` (generic
event writes).
- `replace_parameterized_event_in_transaction_impl` (inside its existing
savepoint) and `Db::replace_addressable_event` (admin reconciliation
path), after dominance checks and before the old head is deleted.

The validator (`validate_channel_roster_snapshot`) is a direct port of
`guard_channel_roster_snapshot()`: four-field `p` tags, 64-char hex
pubkey, empty relay field, valid role, and an exact match against active
`channel_members` rows. It raises the same `23514` messages, so callers
still see `DbError::Sqlx(sqlx::Error::Database(..))` with the same code
and text.

**Lock order:** replacement key, then channel-membership key, then the
canonical read, then the old-head `UPDATE` and the write. Legacy writers
took the membership key later, at the trigger's INSERT, after the
old-head row locks. Taking it before the `UPDATE` is not the same order,
but it is strictly safer: it removes the row-lock/membership-key cycle.
The membership writers in `channel_members.rs` take the same membership
key before they change rows. Not every membership writer does yet:
`execute_kick_with_marker` (reached from report enforcement) and
`deploy_kick_member` in `relay_admin_actions.rs` set `removed_at`
without the key, so a kick can commit after a roster passes validation
and before that roster commits, leaving the removed member listed. The
retained trigger has the same gap, so this PR neither causes nor closes
it. When the trigger also runs it re-takes a key this transaction
already holds; transaction advisory locks are re-entrant, so this adds
no wait.

The validator is one verdict query on the happy path. On failure it
raises the trigger's exact error via a `DO` block, so mixed-version
behavior is identical; phase 2 can switch to a typed `DbError`.

**Unchanged on purpose:** the trigger, its function, migration 0032, and
the startup probe `verify_channel_roster_fence_behavior`. Pods still on
the old version depend on them. Phase 2 drops them once this version is
running everywhere.

## Tests

New PG tests in `channel_members.rs` `postgres_tests`, each with only
the roster trigger disabled (the helper asserts the catalog check then
fails):

- `app_roster_guard_matches_trigger_for_roster_input_table`: 19 shape
and membership cases (3/5-field tags, short and non-hex pubkeys, relay
field, unknown role, duplicate, role mismatch, missing/extra member,
uppercase hex, non-`p` tags, empty rosters, all roles). For each one,
the app outcome through `Db::replace_addressable_event` must equal the
trigger's outcome on a raw INSERT, including SQLSTATE and message.
-
`app_roster_guard_locked_writer_rejects_stale_and_accepts_current_without_trigger`
-
`app_roster_guard_covers_generic_and_parameterized_writers_without_trigger`
-
`app_roster_guard_rechecks_after_concurrent_membership_commit_without_trigger`:
the 0032 race. A newer membership commits while an older snapshot waits
on the membership key. The barrier is `pg_locks` with a bound, not
sleeps.

`stale_legacy_roster_cannot_replace_new_locked_snapshot`
(`replaceable.rs`) and `desired_schema_rejects_stale_legacy_roster_role`
(`channel_members.rs`) now simulate the old binary inline (replacement
key, old-head `UPDATE`, raw INSERT) instead of calling the app API, so
they test the retained trigger again. The replaceable.rs one keeps its
one-connection pool, so it still catches self-deadlock.

`observability_source.rs` gains
`every_production_event_insert_runs_the_roster_guard`, which requires a
real guard call in every production `INSERT INTO events` function, with
explicit exceptions (NIP-43 membership and push lease rows, both
`channel_id` NULL, and the two rolled-back startup probes), plus a
self-test.

Falsified: removing one guard call fails the source policy; dropping the
role clause or `lower()` fails the parity table; disabling the trigger
fails both legacy tests. The PG tests are picked up by the `postgres-ci`
nextest filter (`test(/postgres_tests::/)`); the source policy runs in
`just test-unit`.

### Coverage limits (review follow-up)

- `every_production_event_insert_runs_the_roster_guard` is a presence
check. It rejects missing, commented, and quoted calls, but it does not
prove a call is awaited, reachable, or passed the event's real kind and
channel. Behavior is covered by the PG tests above. The test now says
this in its doc comment.
- Exemptions match by function header. A kind 39002 insert added inside
one of the four exempt functions would pass the policy. This is
documented next to the exemption list.
- The 19-case parity table and the race test drive
`Db::replace_addressable_event` only. The generic, parameterized, and
locked writers get the mismatch case only, and the parameterized path's
membership-key-before-delete ordering is not exercised concurrently. All
five writers call the same validator, so this PR doesn't widen these
tests.
- `disable_roster_trigger` commits `DISABLE TRIGGER` on the test
database. That is safe under the per-test database harness the
`postgres-ci` profile uses, but not against a shared database. Its doc
comment now says so.

## Validation

Run on a Blox workstation (private PG 16 and Redis) against a tree
byte-identical to this commit:

- `cargo fmt --all -- --check`, `cargo clippy --workspace --all-targets
-- -D warnings`, `cargo nextest run -p buzz-db --lib`: pass.
- `cargo nextest run -p buzz-db --test observability_source`: 19 passed.
- `scripts/postgres-test-run.sh -p <pkg>` for buzz-admin, buzz-audit,
buzz-deletion, buzz-push-gateway, buzz-relay, buzz-search,
buzz-workflow: pass.
- buzz-db: the full run passed 513 and was stopped at a 240 s deadline
on `writer_pool_rejects_non_read_committed_database_default`, which
hangs the same way on the unmodified base. A second run excluding the
three host-flaky runtime tests already reproduced on the base passed all
511 others.

## Simpler?

Not yet. While the trigger is still in place, the check runs twice.
Phase 2 deletes the trigger, its function, the startup catalog and
behavior probes, the migration.rs equivalence asserts, and the pgschema
reconcile entry, which leaves one check in Rust.

## Phase 2 (separate PR, after fleet rollout)

**Prerequisite before claiming complete roster serialization (phase 2 or
anywhere else):** make both kick paths take the channel membership key
before they change rows, and add a concurrent kick-vs-roster-publication
test. The retained trigger does not prove this; it has the same gap.


A new migration drops the trigger from `events` and its partitions, then
drops `guard_channel_roster_snapshot()`. It also removes the schema
files and includes, `verify_channel_roster_fence*` and the relay startup
call, the migration.rs asserts, and the probe tests, and updates
`scripts/reconcile-schema-after-pgschema.sql`.

---------

Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
This commit is contained in:
Luke Tornquist
2026-10-16 17:07:25 -04:00
committed by GitHub
parent d79446316b
commit 97e1f6a082
4 changed files with 784 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

600
crates/buzz-db/src/store/channel_members.rs
View File

@@ -197,6 +197,94 @@ async fn acquire_channel_membership_lock(
Ok(())
}
/// Check the same p-tag shape and canonical active membership as migration 0032.
/// The caller holds this channel's membership advisory lock on the admitted
/// transaction before calling. The membership writers in this module take that
/// key before reading roles or changing rows. Not every writer does yet:
/// `execute_kick_with_marker` and `deploy_kick_member` (relay_admin_actions.rs)
/// set `removed_at` without it, so a kick can commit after this check passes
/// and before the roster commits. The retained trigger has the same gap. Both
/// kick paths must take the key, with a concurrent kick-vs-roster-publication
/// test, before anything relies on complete serialization (phase 2 included).
/// Legacy replacement takes the replacement key, locks
/// old-head rows in its UPDATE, then takes the membership key at trigger INSERT.
/// The app path takes replacement key -> membership key -> old-head row locks,
/// removing the row-lock/membership-key cycle rather than preserving it.
///
/// On failure, DO raises the trigger's exact PostgreSQL error during mixed-version
/// rollout. Phase 2 can replace this with a typed DbError once legacy pods retire.
pub(crate) async fn validate_channel_roster_snapshot(
tx: &mut AdmittedTx,
channel_id: Uuid,
tags: &serde_json::Value,
) -> Result<()> {
let verdict: i32 = sqlx::query_scalar(
r#"
SELECT CASE WHEN EXISTS (
SELECT 1 FROM jsonb_array_elements($3::jsonb) AS roster_tag(tag_json)
WHERE roster_tag.tag_json->>0 = 'p'
AND (jsonb_array_length(roster_tag.tag_json) <> 4
OR COALESCE(roster_tag.tag_json->>1, '') !~ '^[0-9a-fA-F]{64}$'
OR roster_tag.tag_json->>2 <> ''
OR COALESCE(roster_tag.tag_json->>3, '')
NOT IN ('owner', 'admin', 'bot', 'member', 'guest'))
) THEN 1
WHEN
(SELECT COALESCE(array_agg(encode(cm.pubkey, 'hex') || ':' || cm.role::text
ORDER BY cm.pubkey), ARRAY[]::text[])
FROM channel_members cm
WHERE cm.community_id = $1 AND cm.channel_id = $2
AND cm.removed_at IS NULL)
IS DISTINCT FROM
(SELECT COALESCE(array_agg(lower(roster_tag.tag_json->>1) || ':' ||
(roster_tag.tag_json->>3)
ORDER BY decode(roster_tag.tag_json->>1, 'hex')),
ARRAY[]::text[])
FROM jsonb_array_elements($3::jsonb) AS roster_tag(tag_json)
WHERE roster_tag.tag_json->>0 = 'p')
THEN 2 ELSE 0 END
"#,
)
.bind(tx.community().as_uuid())
.bind(channel_id)
.bind(tags)
.fetch_one(tx.conn())
.await?;
match verdict {
0 => return Ok(()),
1 => sqlx::query(
"DO $$ BEGIN RAISE EXCEPTION 'kind 39002 roster contains an invalid p tag' USING ERRCODE = '23514'; END $$",
),
_ => sqlx::query(
"DO $$ BEGIN RAISE EXCEPTION 'kind 39002 roster does not match canonical channel membership' USING ERRCODE = '23514'; END $$",
),
}
.execute(tx.conn())
.await?;
Err(DbError::InvalidData(
"roster check did not reject invalid snapshot".into(),
))
}
/// Apply the app-side roster fence to a general event writer. The trigger
/// reacquires this transaction's advisory key until the fleet has upgraded;
/// PostgreSQL advisory locks are reentrant for their current owner.
pub(crate) async fn guard_channel_roster_event(
tx: &mut AdmittedTx,
kind: i32,
channel_id: Option<Uuid>,
tags: &serde_json::Value,
) -> Result<()> {
if kind == 39002 {
if let Some(channel_id) = channel_id {
let community_id = tx.community();
acquire_channel_membership_lock(tx.conn(), community_id, channel_id).await?;
validate_channel_roster_snapshot(tx, channel_id, tags).await?;
}
}
Ok(())
}
// ── Transaction-level membership helpers (for commit_participant_join) ────────
/// Acquire the per-channel membership advisory lock on a caller-owned transaction.
@@ -398,11 +486,12 @@ impl LockedMemberSnapshot {
));
}
}
let tags = serde_json::to_value(&event.tags)?;
validate_channel_roster_snapshot(&mut self.tx, channel_id, &tags).await?;
sqlx::query("UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND channel_id = $4 AND deleted_at IS NULL")
.bind(community_id.as_uuid()).bind(kind).bind(pubkey.as_slice()).bind(channel_id)
.execute(self.tx.conn()).await?;
let received_at = Utc::now();
let tags = serde_json::to_value(&event.tags)?;
let sig = event.sig.serialize();
let inserted = sqlx::query("INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11) ON CONFLICT DO NOTHING")
.bind(community_id.as_uuid()).bind(event.id.as_bytes().as_slice())
@@ -3081,6 +3170,447 @@ mod postgres_tests {
(community, channel.id, owner_a, owner_b)
}
/// Commits `DISABLE TRIGGER` on the test database. Safe only under the
/// per-test database wrapper; do not run these against a shared database,
/// or later trigger-dependent tests there will see no trigger.
async fn disable_roster_trigger(pool: &PgPool) {
let partitions: Vec<String> = sqlx::query_scalar(
"SELECT quote_ident(n.nspname) || '.' || quote_ident(c.relname) \
FROM pg_inherits i JOIN pg_class c ON c.oid = i.inhrelid \
JOIN pg_namespace n ON n.oid = c.relnamespace \
WHERE i.inhparent = 'public.events'::regclass",
)
.fetch_all(pool)
.await
.expect("list event partitions");
for partition in partitions {
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {partition} DISABLE TRIGGER trg_events_guard_channel_roster_snapshot"
)))
.execute(pool)
.await
.expect("disable only the roster trigger on partition");
}
sqlx::query(
"ALTER TABLE ONLY events DISABLE TRIGGER trg_events_guard_channel_roster_snapshot",
)
.execute(pool)
.await
.expect("disable only the roster trigger on parent");
assert!(
verify_channel_roster_fence_catalog(pool).await.is_err(),
"app-only tests must run without the catalog-verified roster fence"
);
}
fn signed_roster(
keys: &Keys,
channel_id: Uuid,
members: &[(&[u8], &str, &str)],
) -> nostr::Event {
use nostr::{EventBuilder, Kind, Tag};
let mut tags = vec![Tag::parse(["d", &channel_id.to_string()]).expect("d tag")];
for (pubkey, relay, role) in members {
tags.push(Tag::parse(["p", &hex::encode(pubkey), relay, role]).expect("p tag"));
}
EventBuilder::new(Kind::Custom(39002), "")
.tags(tags)
.sign_with_keys(keys)
.expect("sign roster")
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn app_roster_guard_locked_writer_rejects_stale_and_accepts_current_without_trigger() {
let pool = setup_pool().await;
disable_roster_trigger(&pool).await;
let community_uuid = make_test_community(&pool).await;
let community = CommunityId::from_uuid(community_uuid);
let owner = random_pubkey();
let channel = create_test_channel(
&pool,
community_uuid,
"app-roster-locked",
ChannelType::Stream,
ChannelVisibility::Open,
None,
&owner,
None,
)
.await
.expect("create channel");
let keys = Keys::generate();
let stale = signed_roster(&keys, channel.id, &[]);
let mut snapshot =
lock_member_snapshot(&pool, community, channel.id, &keys.public_key().to_bytes())
.await
.expect("lock snapshot");
let error = snapshot
.replace_member_event(&stale)
.await
.expect_err("app must reject stale roster without trigger");
assert!(matches!(error, DbError::Sqlx(sqlx::Error::Database(ref db))
if db.code().as_deref() == Some("23514")
&& db.message() == "kind 39002 roster does not match canonical channel membership"));
drop(snapshot);
let current = signed_roster(&keys, channel.id, &[(&owner, "", "owner")]);
let mut snapshot =
lock_member_snapshot(&pool, community, channel.id, &keys.public_key().to_bytes())
.await
.expect("lock current snapshot");
assert!(
snapshot
.replace_member_event(&current)
.await
.expect("current roster")
.1
);
snapshot.release().await.expect("commit current roster");
let live_id: Vec<u8> = sqlx::query_scalar(
"SELECT id FROM events WHERE community_id=$1 AND channel_id=$2 \
AND kind=39002 AND deleted_at IS NULL",
)
.bind(community_uuid)
.bind(channel.id)
.fetch_one(&pool)
.await
.expect("load current roster");
assert_eq!(live_id, current.id.as_bytes().to_vec());
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn app_roster_guard_matches_trigger_for_roster_input_table() {
use nostr::{EventBuilder, Kind, Tag};
use serde_json::json;
let pool = setup_pool().await;
verify_channel_roster_fence_catalog(&pool)
.await
.expect("trigger side must start with the real fence enabled");
let owner = random_pubkey();
let owner_hex = hex::encode(&owner);
let extra_hex = hex::encode(random_pubkey());
let invalid = "kind 39002 roster contains an invalid p tag";
let mismatch = "kind 39002 roster does not match canonical channel membership";
let mut cases = vec![
(
"three-field p",
json!([["p", owner_hex, ""]]),
Some("owner"),
Some(invalid),
),
(
"five-field p",
json!([["p", owner_hex, "", "owner", "extra"]]),
Some("owner"),
Some(invalid),
),
(
"63-char pubkey",
json!([["p", &owner_hex[..63], "", "owner"]]),
Some("owner"),
Some(invalid),
),
(
"non-hex pubkey",
json!([["p", "z".repeat(64), "", "owner"]]),
Some("owner"),
Some(invalid),
),
(
"nonempty relay",
json!([["p", owner_hex, "wss://relay.invalid", "owner"]]),
Some("owner"),
Some(invalid),
),
(
"unknown role",
json!([["p", owner_hex, "", "moderator"]]),
Some("owner"),
Some(invalid),
),
(
"duplicate member",
json!([["p", owner_hex, "", "owner"], ["p", owner_hex, "", "owner"]]),
Some("owner"),
Some(mismatch),
),
(
"role mismatch",
json!([["p", owner_hex, "", "admin"]]),
Some("owner"),
Some(mismatch),
),
("missing member", json!([]), Some("owner"), Some(mismatch)),
(
"extra member",
json!([
["p", owner_hex, "", "owner"],
["p", extra_hex, "", "member"]
]),
Some("owner"),
Some(mismatch),
),
(
"uppercase real member",
json!([["p", owner_hex.to_uppercase(), "", "owner"]]),
Some("owner"),
None,
),
(
"non-p tags ignored",
json!([
["p", owner_hex, "", "owner"],
["name", "ignored"],
["x"],
["P", "not a p tag"]
]),
Some("owner"),
None,
),
("empty roster and membership", json!([]), None, None),
(
"nonempty roster and empty membership",
json!([["p", owner_hex, "", "owner"]]),
None,
Some(mismatch),
),
];
for role in ["owner", "admin", "bot", "member", "guest"] {
cases.push((role, json!([["p", owner_hex, "", role]]), Some(role), None));
}
assert_ne!(
owner_hex,
owner_hex.to_uppercase(),
"uppercase case must change the input"
);
let database_error = |error: sqlx::Error| match error {
sqlx::Error::Database(error) => (
error.code().expect("SQLSTATE").into_owned(),
error.message().to_owned(),
),
error => panic!("expected a database verdict, got {error:?}"),
};
let keys = Keys::generate();
let mut measured = Vec::new();
// Measure all trigger verdicts first with raw inserts, bypassing every
// app guard. Each savepoint and transaction is rolled back so the app
// later sees exactly the same canonical membership and no event head.
for (name, tags, role, expected_error) in cases {
let community_uuid = make_test_community(&pool).await;
let community = CommunityId::from_uuid(community_uuid);
let channel = create_test_channel(
&pool,
community_uuid,
name,
ChannelType::Stream,
ChannelVisibility::Open,
None,
&owner,
None,
)
.await
.expect("create case channel");
if let Some(role) = role {
sqlx::query("UPDATE channel_members SET role=$3::member_role WHERE community_id=$1 AND channel_id=$2")
.bind(community_uuid)
.bind(channel.id)
.bind(role)
.execute(&pool)
.await
.expect("set canonical case role");
} else {
sqlx::query("UPDATE channel_members SET removed_at=NOW() WHERE community_id=$1 AND channel_id=$2")
.bind(community_uuid)
.bind(channel.id)
.execute(&pool)
.await
.expect("empty active membership");
}
let mut tags: Vec<Vec<String>> = serde_json::from_value(tags).expect("case tag arrays");
tags.insert(0, vec!["d".into(), channel.id.to_string()]);
let event = EventBuilder::new(Kind::Custom(39002), "")
.tags(tags.iter().map(|tag| Tag::parse(tag).expect("case tag")))
.sign_with_keys(&keys)
.expect("sign case");
let mut tx = pool.begin().await.expect("trigger transaction");
sqlx::query("SAVEPOINT roster_case")
.execute(&mut *tx)
.await
.expect("case savepoint");
let trigger = sqlx::query(
"INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag) \
VALUES ($1, $2, $3, $4, 39002, $5, '', $6, NOW(), $7, $8)",
)
.bind(community_uuid)
.bind(event.id.as_bytes().as_slice())
.bind(event.pubkey.to_bytes().as_slice())
.bind(DateTime::from_timestamp(event.created_at.as_secs() as i64, 0).expect("timestamp"))
.bind(serde_json::to_value(&event.tags).expect("tags"))
.bind(event.sig.serialize().as_slice())
.bind(channel.id)
.bind(channel.id.to_string())
.execute(&mut *tx)
.await
.map(|result| result.rows_affected() == 1)
.map_err(database_error);
sqlx::query("ROLLBACK TO SAVEPOINT roster_case")
.execute(&mut *tx)
.await
.expect("undo trigger case");
tx.rollback().await.expect("undo trigger transaction");
let expected =
expected_error.map_or(Ok(true), |message| Err(("23514".into(), message.into())));
assert_eq!(trigger, expected, "trigger contract: {name}");
measured.push((name, community, channel.id, event, trigger));
}
// Only the app side runs with the trigger disabled; this helper also
// asserts that the retained catalog check now rejects the database.
disable_roster_trigger(&pool).await;
let db = Db::from_pool(pool.clone());
for (name, community, channel, event, trigger) in measured {
let app = db
.replace_addressable_event(community, &event, Some(channel))
.await
.map(|(_, inserted)| inserted)
.map_err(|error| match error {
DbError::Sqlx(error) => database_error(error),
error => panic!("expected app database verdict for {name}, got {error:?}"),
});
assert_eq!(app, trigger, "app/trigger parity: {name}");
}
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn app_roster_guard_covers_generic_and_parameterized_writers_without_trigger() {
let pool = setup_pool().await;
disable_roster_trigger(&pool).await;
let community_uuid = make_test_community(&pool).await;
let community = CommunityId::from_uuid(community_uuid);
let owner = random_pubkey();
let channel = create_test_channel(
&pool,
community_uuid,
"app-roster-generic",
ChannelType::Stream,
ChannelVisibility::Open,
None,
&owner,
None,
)
.await
.expect("create channel");
let stale = signed_roster(&Keys::generate(), channel.id, &[]);
let db = Db::from_pool(pool.clone());
for result in [
db.insert_event(community, &stale, Some(channel.id)).await,
db.insert_event_with_thread_metadata(community, &stale, Some(channel.id), None)
.await,
db.replace_parameterized_event(
community,
&stale,
&channel.id.to_string(),
Some(channel.id),
)
.await,
] {
let error = result.expect_err("every event writer must reject stale roster");
assert!(matches!(error, DbError::Sqlx(sqlx::Error::Database(ref db))
if db.code().as_deref() == Some("23514")
&& db.message() == "kind 39002 roster does not match canonical channel membership"));
}
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn app_roster_guard_rechecks_after_concurrent_membership_commit_without_trigger() {
let pool = setup_pool().await;
disable_roster_trigger(&pool).await;
let community_uuid = make_test_community(&pool).await;
let community = CommunityId::from_uuid(community_uuid);
let owner = random_pubkey();
let newcomer = random_pubkey();
let channel = create_test_channel(
&pool,
community_uuid,
"app-roster-race",
ChannelType::Stream,
ChannelVisibility::Open,
None,
&owner,
None,
)
.await
.expect("create channel");
let old = signed_roster(&Keys::generate(), channel.id, &[(&owner, "", "owner")]);
let mut holder = pool.begin().await.expect("hold membership transaction");
acquire_channel_membership_lock(&mut holder, community, channel.id)
.await
.expect("hold membership key");
sqlx::query(
"INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) \
VALUES ($1, $2, $3, 'member', $4)",
)
.bind(community_uuid)
.bind(channel.id)
.bind(&newcomer)
.bind(&owner)
.execute(&mut *holder)
.await
.expect("stage newer canonical roster");
let writer_pool = pool.clone();
let old_id = old.id.as_bytes().to_vec();
let writer = tokio::spawn(async move {
Db::from_pool(writer_pool)
.replace_addressable_event(community, &old, Some(channel.id))
.await
});
// A waiting advisory lock proves the old publication has reached the
// app fence before the newer canonical membership commits.
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 locktype='advisory' \
AND NOT granted AND database=(SELECT oid FROM pg_database WHERE datname=current_database()))",
)
.fetch_one(&pool)
.await
.expect("observe waiting roster writer");
if waiting { break; }
tokio::task::yield_now().await;
}
})
.await
.expect("old roster writer must wait for membership key");
holder
.commit()
.await
.expect("commit newer canonical roster");
let error = writer
.await
.expect("join old writer")
.expect_err("old in-flight roster must be rejected");
assert!(matches!(error, DbError::Sqlx(sqlx::Error::Database(ref db))
if db.code().as_deref() == Some("23514")
&& db.message() == "kind 39002 roster does not match canonical channel membership"));
let count: i64 = sqlx::query_scalar("SELECT count(*) FROM events WHERE id=$1")
.bind(old_id)
.fetch_one(&pool)
.await
.expect("count rejected old roster");
assert_eq!(count, 0);
}
/// A captured roster holds the same lock as membership writers until the
/// publisher explicitly releases it. This is the freshness fence used by
/// rolling-deploy reconciliation.
@@ -3804,15 +4334,61 @@ mod postgres_tests {
.1
);
let stale = roster("member", base + 1);
let error = db
.replace_addressable_event(community, &stale, Some(channel))
// Reproduce the old binary on one connection: replacement key ->
// old-head UPDATE -> raw INSERT. Do not use the app API, whose guard
// rejects before the retained trigger can exercise this legacy fence.
let error = async {
let mut tx = pool.begin().await.expect("begin legacy replacement");
let relay_pubkey = relay_keys.public_key().to_bytes();
sqlx::query("SELECT pg_advisory_xact_lock($1)")
.bind(crate::store::replaceable::event_replacement_lock_key(
community,
39002,
&relay_pubkey,
Some(channel.as_bytes().as_slice()),
))
.execute(&mut *tx)
.await
.expect("legacy replacement key");
let deleted = sqlx::query(
"UPDATE events SET deleted_at = NOW() \
WHERE community_id=$1 AND kind=39002 AND pubkey=$2 \
AND channel_id=$3 AND deleted_at IS NULL",
)
.bind(community_uuid)
.bind(relay_pubkey.as_slice())
.bind(channel)
.execute(&mut *tx)
.await
.expect_err("desired-state fence must reject stale role");
assert!(matches!(
error,
DbError::Sqlx(sqlx::Error::Database(ref db_error))
.expect("legacy soft-delete of current head");
assert_eq!(deleted.rows_affected(), 1, "legacy path must delete the live head");
let result = sqlx::query(
"INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag) \
VALUES ($1, $2, $3, $4, 39002, $5, $6, $7, NOW(), $8, $9)",
)
.bind(community_uuid)
.bind(stale.id.as_bytes().as_slice())
.bind(relay_pubkey.as_slice())
.bind(chrono::DateTime::from_timestamp(stale.created_at.as_secs() as i64, 0).expect("timestamp"))
.bind(serde_json::to_value(&stale.tags).expect("tags"))
.bind(&stale.content)
.bind(stale.sig.serialize().as_slice())
.bind(channel)
.bind(channel.to_string())
.execute(&mut *tx)
.await;
tx.rollback().await.expect("restore live head after legacy attempt");
result
}
.await
.expect_err("desired-state fence must reject stale role");
assert!(
matches!(error, sqlx::Error::Database(ref db_error)
if db_error.code().as_deref() == Some("23514")
));
&& db_error.message() == "kind 39002 roster does not match canonical channel membership"),
"expected trigger roster fence check violation, got {error:?}"
);
let live_id: Vec<u8> = sqlx::query_scalar(
"SELECT id FROM events WHERE community_id=$1 AND channel_id=$2 \
AND kind=39002 AND deleted_at IS NULL",
@@ -3823,6 +4399,14 @@ mod postgres_tests {
.await
.expect("load desired-state live roster");
assert_eq!(live_id, fresh.id.as_bytes().to_vec());
let stale_rows: i64 =
sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id=$1 AND id=$2")
.bind(community_uuid)
.bind(stale.id.as_bytes().as_slice())
.fetch_one(&pool)
.await
.expect("count rejected stale roster");
assert_eq!(stale_rows, 0, "legacy stale insert must roll back");
drop_scratch_db(&admin, pool, &scratch_name).await;
}

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

@@ -482,6 +482,8 @@ async fn insert_event_on(
let received_at = Utc::now();
let d_tag = extract_d_tag(event);
let not_before = extract_not_before(event);
crate::channel_members::guard_channel_roster_event(tx, kind_i32, channel_id, &tags_json)
.await?;
let result = sqlx::query(
r#"
INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before)
@@ -1784,6 +1786,9 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
let d_tag = extract_d_tag(event);
let not_before = extract_not_before(event);
crate::channel_members::guard_channel_roster_event(tx, kind_i32, channel_id, &tags_json)
.await?;
let result = sqlx::query(
r#"
INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before)

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

@@ -259,6 +259,9 @@ async fn replace_parameterized_event_in_transaction_impl(
.execute(tx.conn())
.await?;
let written: Result<bool> = async {
let tags_json = serde_json::to_value(&event.tags)?;
crate::channel_members::guard_channel_roster_event(tx, kind_i32, channel_id, &tags_json)
.await?;
if existing.is_some() {
let previous_nip_rs_hard_delete: Option<String> = if is_nip_rs {
sqlx::query_scalar(
@@ -309,7 +312,6 @@ async fn replace_parameterized_event_in_transaction_impl(
}
let sig_bytes = event.sig.serialize();
let tags_json = serde_json::to_value(&event.tags)?;
let insert_result = sqlx::query(
"INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) \
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) \
@@ -485,6 +487,12 @@ impl Db {
}
}
let tags_json = serde_json::to_value(&event.tags)?;
crate::channel_members::guard_channel_roster_event(
&mut tx, kind_i32, channel_id, &tags_json,
)
.await?;
// Soft-delete the old event (if any). IS NOT DISTINCT FROM for NULL safety.
sqlx::query(
"UPDATE events SET deleted_at = NOW() \
@@ -501,7 +509,6 @@ impl Db {
// Insert the new event inside the same transaction.
let sig_bytes = event.sig.serialize();
let tags_json = serde_json::to_value(&event.tags)?;
let received_at = chrono::Utc::now();
let d_tag = crate::event::extract_d_tag(event);
@@ -921,25 +928,61 @@ mod postgres_tests {
.await
.expect("commit B and release locks");
// The legacy canonical path takes the replacement key, soft-deletes B,
// then attempts its newer-timestamp stale A. Migration 0032 rejects the
// INSERT; transaction rollback must restore B. A one-connection pool
// proves the lock order does not turn this compatibility path into a
// self-deadlock.
let error = tokio::time::timeout(
Duration::from_secs(3),
db.replace_addressable_event(community, &stale_a, Some(channel)),
)
// Reproduce the old binary on one connection: replacement key ->
// old-head UPDATE -> raw INSERT. Do not use the app API, whose guard
// rejects before the retained trigger can exercise this legacy fence.
// A size-one pool and bounded attempt also catch self-deadlock.
let error = tokio::time::timeout(Duration::from_secs(3), async {
let mut tx = pool.begin().await.expect("begin legacy replacement");
let relay_pubkey = relay_keys.public_key().to_bytes();
sqlx::query("SELECT pg_advisory_xact_lock($1)")
.bind(crate::store::replaceable::event_replacement_lock_key(
community,
39002,
&relay_pubkey,
Some(channel.as_bytes().as_slice()),
))
.execute(&mut *tx)
.await
.expect("legacy replacement key");
let deleted = sqlx::query(
"UPDATE events SET deleted_at = NOW() \
WHERE community_id=$1 AND kind=39002 AND pubkey=$2 \
AND channel_id=$3 AND deleted_at IS NULL",
)
.bind(community_uuid)
.bind(relay_pubkey.as_slice())
.bind(channel)
.execute(&mut *tx)
.await
.expect("legacy soft-delete of current head");
assert_eq!(deleted.rows_affected(), 1, "legacy path must delete the live head");
let result = sqlx::query(
"INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag) \
VALUES ($1, $2, $3, $4, 39002, $5, $6, $7, NOW(), $8, $9)",
)
.bind(community_uuid)
.bind(stale_a.id.as_bytes().as_slice())
.bind(relay_pubkey.as_slice())
.bind(chrono::DateTime::from_timestamp(stale_a.created_at.as_secs() as i64, 0).expect("timestamp"))
.bind(serde_json::to_value(&stale_a.tags).expect("tags"))
.bind(&stale_a.content)
.bind(stale_a.sig.serialize().as_slice())
.bind(channel)
.bind(channel.to_string())
.execute(&mut *tx)
.await;
tx.rollback().await.expect("restore live head after legacy attempt");
result
})
.await
.expect("legacy replacement must not deadlock")
.expect_err("stale captured roster A must be rejected");
assert!(
matches!(
error,
DbError::Sqlx(sqlx::Error::Database(ref db_error))
if db_error.code().as_deref() == Some("23514")
),
"expected roster fence check violation, got {error:?}"
matches!(error, sqlx::Error::Database(ref db_error)
if db_error.code().as_deref() == Some("23514")
&& db_error.message() == "kind 39002 roster does not match canonical channel membership"),
"expected trigger roster fence check violation, got {error:?}"
);
let live_ids: Vec<Vec<u8>> = sqlx::query_scalar(

129
crates/buzz-db/tests/observability_source.rs
View File

@@ -1877,8 +1877,7 @@ mod tests {
);
}
#[test]
fn every_production_event_insert_runs_the_follow_up_hook() {
fn production_event_writer_source_files() -> Vec<std::path::PathBuf> {
use std::path::{Path, PathBuf};
fn collect_rs_files(dir: &Path, out: &mut Vec<PathBuf>) {
@@ -1903,6 +1902,12 @@ fn every_production_event_insert_runs_the_follow_up_hook() {
}
}
files
}
#[test]
fn every_production_event_insert_runs_the_follow_up_hook() {
let files = production_event_writer_source_files();
let mut hooked_writers = 0usize;
let mut violations = Vec::new();
for path in files {
@@ -1935,3 +1940,123 @@ fn every_production_event_insert_runs_the_follow_up_hook() {
"expected at least seven hooked event writers, found {hooked_writers}"
);
}
// Fixed channel-less kinds cannot be channel rosters. The startup probes must
// exercise the retained DB backstops directly and roll back all their writes.
// Exemptions match by function header, so a kind 39002 insert added inside one
// of these functions would pass the policy; keep them single-purpose.
const EVENT_INSERT_ROSTER_GUARD_EXCEPTIONS: [&str; 4] = [
// NIP-43 membership is kind 13534 with channel_id NULL.
"pub async fn publish_nip43_membership_locked(",
// Push leases are kind 30350 with channel_id NULL.
"pub async fn accept_lease_event(",
// Rolled-back replica-floor startup probe.
"pub async fn verify_floor_guard_behavior(",
// Rolled-back roster-fence startup probe.
"pub async fn verify_channel_roster_fence_behavior(",
];
fn event_insert_roster_guard_violations(production_source: &str) -> Vec<String> {
function_slices(production_source)
.into_iter()
.filter(|function_source| {
let header = function_header(function_source);
let code = mask_comments_and_literals(function_source);
inserts_events_row(function_source)
&& !EVENT_INSERT_ROSTER_GUARD_EXCEPTIONS
.iter()
.any(|exception| header.starts_with(exception))
&& !code.contains("guard_channel_roster_event(")
&& !code.contains("validate_channel_roster_snapshot(")
})
.map(|function_source| function_header(function_source).to_owned())
.collect()
}
#[test]
fn event_insert_roster_guard_policy_rejects_writers_without_the_guard() {
let source = r#"
pub(crate) async fn unguarded_writer(tx: &mut AdmittedTx) {
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)");
}
pub(crate) async fn guarded_writer(tx: &mut AdmittedTx) {
guard_channel_roster_event(tx, kind, channel, tags).await?;
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)");
}
pub(crate) async fn locked_writer(tx: &mut AdmittedTx) {
validate_channel_roster_snapshot(tx, channel, tags).await?;
sqlx::query("insert into public.events (community_id, id) VALUES ($1, $2)");
}
pub(crate) async fn mention_writer(tx: &mut AdmittedTx) {
sqlx::query("INSERT INTO event_mentions (community_id, event_id) VALUES ($1, $2)");
}
pub(crate) async fn commented_guard_writer(tx: &mut AdmittedTx) {
// guard_channel_roster_event(tx, kind, channel, tags).await?;
/* validate_channel_roster_snapshot(tx, channel, tags).await?; */
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)");
}
pub(crate) async fn quoted_guard_writer(tx: &mut AdmittedTx) {
tracing::debug!("guard_channel_roster_event(");
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)");
}
pub async fn publish_nip43_membership_locked(tx: &mut AdmittedTx) {
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)");
}
pub async fn accept_lease_event(tx: &mut AdmittedTx) {
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)");
}
pub async fn verify_floor_guard_behavior(pool: &PgPool) {
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)");
}
pub async fn verify_channel_roster_fence_behavior(pool: &PgPool) {
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)");
}
#[cfg(test)]
mod tests {
fn fixture() {
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)");
}
}
"#;
assert_eq!(
event_insert_roster_guard_violations(&strip_cfg_test_items(source)),
[
"pub(crate) async fn unguarded_writer(tx: &mut AdmittedTx) {",
"pub(crate) async fn commented_guard_writer(tx: &mut AdmittedTx) {",
"pub(crate) async fn quoted_guard_writer(tx: &mut AdmittedTx) {",
],
"only real roster-guard calls in production writers count"
);
}
/// Presence check: every production `INSERT INTO events` function must contain
/// a real (uncommented, unquoted) roster-guard call. It does not prove the call
/// is awaited, reachable, or given the event's real kind and channel; the PG
/// `app_roster_guard_*` tests cover behavior.
#[test]
fn every_production_event_insert_runs_the_roster_guard() {
let mut checked_writers = 0;
let mut violations = Vec::new();
for path in production_event_writer_source_files() {
let source = std::fs::read_to_string(&path).expect("read source file");
let production = strip_cfg_test_items(&source);
checked_writers += function_slices(&production)
.into_iter()
.filter(|function| inserts_events_row(function))
.count();
violations.extend(
event_insert_roster_guard_violations(&production)
.into_iter()
.map(|header| format!("{}: {header}", path.display())),
);
}
assert!(
violations.is_empty(),
"production `INSERT INTO events` writers must run the app-side roster guard: {violations:?}"
);
// Five guarded writers plus the four explicitly exempt writers/probes.
assert!(
checked_writers >= 9,
"expected at least nine event writers, found {checked_writers}"
);
}