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

Centralize replaceable event persistence (#6660)

Browse Source 
## Why
Command persistence duplicated NIP-33 replacement SQL in the relay and
obscured the boundary between database runtime concerns and domain-store
behavior. This proving slice establishes that boundary inside the
existing `buzz-db` crate.

## What
- Centralize parameterized-replaceable coordinate locking, ordering,
replacement, watermark, and mention-indexing behavior in `buzz-db`
- Expose transaction-required replacement and ordinary-insertion seams
while keeping transaction ownership and workflow conflict messages in
the relay
- Cover concurrency, same-second ties, stale writes, replay, caller
rollback, nested rollback recovery, and mention-index atomicity with
focused PostgreSQL tests

## Risk Assessment
Medium — this changes core event persistence and command idempotency
paths, while intentionally preserving NIP-33, NIP-RS, mesh, and workflow
conflict semantics.

## References
- Architecture guardrail:
https://github.com/TheSentinel454/buzz/issues/34
- Proving slice: https://github.com/TheSentinel454/buzz/issues/3,
https://github.com/TheSentinel454/buzz/issues/4,
https://github.com/TheSentinel454/buzz/issues/6

Generated with Codex

---------

Signed-off-by: tornquist <tornquist@squareup.com>
This commit is contained in:
Luke Tornquist
2026-08-24 20:42:15 -04:00
committed by GitHub
parent e760c51820
commit 8d2d0ff5ad
6 changed files with 1140 additions and 462 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
.github/workflows/ci.yml vendored
View File

@@ -679,6 +679,19 @@ jobs:
VALUES ('00000000-0000-4000-8000-00000000c0de', 'localhost:3000')
ON CONFLICT (lower(host)) DO NOTHING
;"
- name: Replaceable persistence PostgreSQL tests
# Transaction, concurrency, and mention-index coverage for the
# replaceable-event store seam. These tests require real Postgres and
# are ignored by the infrastructure-free unit-test job.
run: |
filter='package(buzz-db) and test(/tests::(parameterized_|concurrent_parameterized_)/)'
cargo nextest run \
--archive-file target/ci/backend-integration-tests.tar.zst \
-E "${filter}" \
--run-ignored ignored-only
env:
DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz
TEST_DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz
- name: Start relay
run: |
chmod +x ./target/ci/buzz-relay

2
crates/buzz-db/src/channel.rs
View File

@@ -625,7 +625,7 @@ pub async fn lock_member_snapshot(
// this key before INSERT; migration 0032 then takes the membership key in
// the INSERT trigger. Taking both in that order avoids mixed-version
// duplicate heads without introducing a lock-order inversion.
let replacement_lock = crate::event_replacement_lock_key(
let replacement_lock = crate::replaceable::event_replacement_lock_key(
community_id,
39002,
relay_pubkey,

53
crates/buzz-db/src/event.rs
View File

@@ -6,7 +6,7 @@
use chrono::{DateTime, Utc};
use nostr::Event;
use sqlx::{PgPool, Postgres, QueryBuilder, Row, Transaction};
use sqlx::{PgConnection, PgPool, Postgres, QueryBuilder, Row, Transaction};
use uuid::Uuid;
use buzz_core::kind::{
@@ -275,6 +275,30 @@ pub async fn insert_event(
community_id: CommunityId,
event: &Event,
channel_id: Option<Uuid>,
) -> Result<(StoredEvent, bool)> {
let mut connection = pool.acquire().await?;
insert_event_on(&mut connection, community_id, event, channel_id).await
}
/// Insert a Nostr event in a caller-owned PostgreSQL transaction.
///
/// This is the transaction-composition seam for callers that must keep the
/// event insert open while performing related work. The caller owns commit or
/// rollback.
pub async fn insert_event_in_transaction(
tx: &mut Transaction<'_, Postgres>,
community_id: CommunityId,
event: &Event,
channel_id: Option<Uuid>,
) -> Result<(StoredEvent, bool)> {
insert_event_on(tx.as_mut(), community_id, event, channel_id).await
}
async fn insert_event_on(
connection: &mut PgConnection,
community_id: CommunityId,
event: &Event,
channel_id: Option<Uuid>,
) -> Result<(StoredEvent, bool)> {
let kind_u16 = event.kind.as_u16();
let kind_u32 = u32::from(kind_u16);
@@ -317,7 +341,7 @@ pub async fn insert_event(
.bind(channel_id)
.bind(d_tag.as_deref())
.bind(not_before)
.execute(pool)
.execute(connection)
.await?;
let was_inserted = result.rows_affected() > 0;
@@ -1607,6 +1631,31 @@ mod tests {
id
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn event_insert_in_existing_transaction_rolls_back_with_caller() {
let pool = setup_pool().await;
let community_uuid = make_test_community(&pool).await;
let community = CommunityId::from_uuid(community_uuid);
let event = make_text_event("caller-owned transaction");
let mut tx = pool.begin().await.expect("begin event insert transaction");
let (_, was_inserted) = insert_event_in_transaction(&mut tx, community, &event, None)
.await
.expect("insert event in caller transaction");
assert!(was_inserted);
tx.rollback().await.expect("roll back event insert");
let persisted: i64 =
sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id = $1 AND id = $2")
.bind(community_uuid)
.bind(event.id.as_bytes().as_slice())
.fetch_one(&pool)
.await
.expect("count rolled-back event");
assert_eq!(persisted, 0);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn event_insert_ttl_trigger_handles_permanent_ephemeral_duplicate_and_activation_race() {

747
crates/buzz-db/src/lib.rs
View File

@@ -8,6 +8,21 @@
//! - Events table is partitioned by month on `created_at`.
//! - No FK references to partitioned tables.
//! - Uses `sqlx::query()` (runtime) not `sqlx::query!()` (compile-time).
//!
//! ## Runtime and store ownership
//! This crate intentionally keeps database runtime and Buzz domain persistence
//! together while maintaining an internal boundary between them:
//!
//! - Runtime concerns own pool construction, writer/replica routing, transaction
//! creation, session invariants, metrics, and health support.
//! - Store concerns own domain-specific SQL, row mapping, locking, mutation
//! rules, indexes, and focused persistence tests.
//!
//! Transaction-required store operations accept [`sqlx::Transaction`] so their
//! composition requirement is visible in the type. Private connection helpers
//! are reserved for SQL primitives that are valid on any same-session
//! connection. New domains should prove this boundary incrementally instead of
//! exposing raw pools or introducing broad store traits.
/// Explicit deployment-global admin report reads.
pub mod admin_moderation;
@@ -45,6 +60,8 @@ pub mod reaction;
pub mod relay_invite;
/// Relay-level membership persistence (NIP-43).
pub mod relay_members;
/// Replaceable-event persistence and coordinate locking.
pub mod replaceable;
/// Replica freshness fence for keyset-cursor read routing.
pub mod replica_fence;
/// Thread metadata persistence.
@@ -68,37 +85,12 @@ use uuid::Uuid;
use buzz_core::{CommunityId, StoredEvent};
pub(crate) fn event_replacement_lock_key(
community_id: CommunityId,
kind: i32,
pubkey: &[u8],
coordinate: Option<&[u8]>,
) -> i64 {
let mut hash: u64 = 0xcbf29ce484222325;
let kind_bytes = kind.to_le_bytes();
for bytes in [
community_id.as_uuid().as_bytes().as_slice(),
kind_bytes.as_slice(),
pubkey,
] {
for byte in bytes {
hash ^= *byte as u64;
hash = hash.wrapping_mul(0x100000001b3);
}
}
if let Some(coordinate) = coordinate {
for byte in coordinate {
hash ^= *byte as u64;
hash = hash.wrapping_mul(0x100000001b3);
}
}
hash as i64
}
/// Extract p-tag mentions from an event and insert into the `event_mentions` table.
///
/// Called after event insertion. Failures are logged but do not block event storage.
/// Uses `INSERT ... ON CONFLICT DO NOTHING` so duplicate inserts are silently skipped.
/// This pool-owning wrapper propagates failures to its caller. Replacement writes
/// use the transaction-bound helper below so event storage and mention indexing
/// commit or roll back together. Duplicate inserts are silently skipped with
/// `INSERT ... ON CONFLICT DO NOTHING`.
pub async fn insert_mentions(
pool: &PgPool,
community_id: CommunityId,
@@ -4875,7 +4867,7 @@ impl Db {
.ok_or(DbError::InvalidTimestamp(created_at_secs))?;
// Collisions only cause extra serialization; they cannot change behavior.
let lock_key = event_replacement_lock_key(
let lock_key = replaceable::event_replacement_lock_key(
community_id,
kind_i32,
pubkey_bytes.as_slice(),
@@ -5058,8 +5050,12 @@ impl Db {
let kind_i32 = buzz_core::kind::KIND_NIP43_MEMBERSHIP_LIST as i32;
let pubkey_bytes = relay_keypair.public_key().to_bytes();
let lock_key =
event_replacement_lock_key(community_id, kind_i32, pubkey_bytes.as_slice(), None);
let lock_key = replaceable::event_replacement_lock_key(
community_id,
kind_i32,
pubkey_bytes.as_slice(),
None,
);
let mut tx = self.pool.begin().await?;
@@ -5166,243 +5162,6 @@ impl Db {
member_count,
))
}
/// Atomically replace a NIP-33 parameterized replaceable event (kind 30000–39999).
///
/// Keeps only the event with the highest `created_at` per `(kind, pubkey, d_tag)`.
/// Same-second ties are broken by lowest event `id` (deterministic ordering).
/// The entire check → retire old payload → insert runs in a single transaction
/// with an advisory lock to prevent concurrent-insert races. NIP-RS read-state
/// coordinates hard-delete the superseded payload and preserve a compact
/// ordering watermark. Buzz mesh status coordinates also hard-delete their
/// superseded heartbeat payload because only the live head has product
/// value; other NIP-33 kinds retain soft-deleted history.
///
/// **Channel policy:** NIP-33 replacement keys on `(kind, pubkey, d_tag)` globally —
/// `channel_id` is NOT part of the replacement key. This matches the Nostr spec:
/// an author's parameterized replaceable event is a single global resource identified
/// by its d-tag, regardless of which channel it was submitted to. The `channel_id`
/// parameter is stored on the new row for query scoping but does not affect replacement.
///
/// Note: `replace_addressable_event()` keys on `channel_id` because it serves
/// relay-signed NIP-29 group metadata (kind 39000–39002) where the relay is the
/// author and channel_id distinguishes groups. User-submitted NIP-33 events use
/// this function instead, where the author's pubkey + d-tag is the natural key.
#[datastore_span(name = "replace_parameterized_event", system = "postgresql")]
pub async fn replace_parameterized_event(
&self,
community_id: CommunityId,
event: &nostr::Event,
d_tag: &str,
channel_id: Option<Uuid>,
) -> Result<(StoredEvent, bool)> {
let kind_i32 = buzz_core::kind::event_kind_i32(event);
let pubkey_bytes = event.pubkey.to_bytes();
let created_at_secs = event.created_at.as_secs() as i64;
let created_at = chrono::DateTime::from_timestamp(created_at_secs, 0)
.ok_or(DbError::InvalidTimestamp(created_at_secs))?;
let lock_key = event_replacement_lock_key(
community_id,
kind_i32,
pubkey_bytes.as_slice(),
Some(d_tag.as_bytes()),
);
let mut tx = self.pool.begin().await?;
sqlx::query("SELECT pg_advisory_xact_lock($1)")
.bind(lock_key)
.execute(&mut *tx)
.await?;
let d_tag_count = event
.tags
.iter()
.filter(|tag| tag.as_slice().first().is_some_and(|part| part == "d"))
.count();
let has_exact_d_tag = event.tags.iter().any(|tag| {
let parts = tag.as_slice();
parts.len() >= 2 && parts[0] == "d" && parts[1] == d_tag
});
let read_state_t_tag_count = event
.tags
.iter()
.filter(|tag| {
let parts = tag.as_slice();
parts.len() == 2 && parts[0] == "t" && parts[1] == "read-state"
})
.count();
let is_nip_rs = kind_i32 == buzz_core::kind::KIND_READ_STATE as i32
&& d_tag_count == 1
&& has_exact_d_tag
&& d_tag.strip_prefix("read-state:").is_some_and(|slot| {
slot.len() == 32
&& slot
.bytes()
.all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
})
&& read_state_t_tag_count == 1;
let is_buzz_mesh_status = kind_i32 == buzz_core::kind::KIND_BOOKMARK_SET as i32
&& d_tag.starts_with("buzz-mesh-member-status:")
&& event.tags.iter().any(|tag| {
let parts = tag.as_slice();
parts.len() == 2 && parts[0] == "k" && parts[1] == "buzz-mesh-status"
});
let hard_delete_superseded = is_nip_rs || is_buzz_mesh_status;
// Check the live head and, for NIP-RS, the compact historical ordering
// watermark. The watermark remains after a NIP-09 coordinate deletion,
// preventing a previously accepted signed blob from being resurrected.
let existing: Option<(chrono::DateTime<chrono::Utc>, Vec<u8>)> = sqlx::query_as(
"SELECT created_at, id FROM events \
WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL \
ORDER BY created_at DESC, id ASC LIMIT 1",
)
.bind(community_id.as_uuid())
.bind(kind_i32)
.bind(pubkey_bytes.as_slice())
.bind(d_tag)
.fetch_optional(&mut *tx)
.await?;
let watermark: Option<(chrono::DateTime<chrono::Utc>, Vec<u8>)> = if is_nip_rs {
sqlx::query_as(
"SELECT created_at, event_id FROM parameterized_event_watermarks \
WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4",
)
.bind(community_id.as_uuid())
.bind(kind_i32)
.bind(pubkey_bytes.as_slice())
.bind(d_tag)
.fetch_optional(&mut *tx)
.await?
} else {
None
};
// Stale-write protection: reject if either durable ordering source
// dominates the incoming tuple. Equal timestamps use lowest event id.
let incoming_id = event.id.as_bytes().as_slice();
let dominated =
existing
.iter()
.chain(watermark.iter())
.any(|(accepted_ts, accepted_id)| {
created_at < *accepted_ts
|| (created_at == *accepted_ts && incoming_id >= accepted_id.as_slice())
});
if dominated {
tx.rollback().await?;
let received_at = chrono::Utc::now();
return Ok((
StoredEvent::with_received_at(event.clone(), received_at, channel_id, false),
false,
));
}
if existing.is_some() {
if is_nip_rs {
// Migration 0011 rejects regex-coordinate hard deletes from
// pre-fix writers. Authorize only this corrected NIP-RS delete,
// transaction-locally so pooled connections cannot leak it.
sqlx::query("SELECT set_config('buzz.nip_rs_hard_delete', 'on', true)")
.execute(&mut *tx)
.await?;
}
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(&mut *tx)
.await?;
if hard_delete_superseded {
if let Some((_, existing_id)) = &existing {
// Event first, mentions second: migration 0009's live-event
// fence uses this global lock order to avoid deadlocks.
sqlx::query(
"DELETE FROM event_mentions WHERE community_id = $1 AND event_id = $2",
)
.bind(community_id.as_uuid())
.bind(existing_id)
.execute(&mut *tx)
.await?;
}
}
}
// Insert the new event inside the transaction.
let sig_bytes = event.sig.serialize();
let tags_json = serde_json::to_value(&event.tags)?;
let received_at = chrono::Utc::now();
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) \
ON CONFLICT DO NOTHING",
)
.bind(community_id.as_uuid())
.bind(event.id.as_bytes().as_slice())
.bind(pubkey_bytes.as_slice())
.bind(created_at)
.bind(kind_i32)
.bind(&tags_json)
.bind(&event.content)
.bind(sig_bytes.as_slice())
.bind(received_at)
.bind(channel_id)
.bind(d_tag)
.bind(event::extract_not_before(event))
.execute(&mut *tx)
.await?;
let was_inserted = insert_result.rows_affected() > 0;
if !was_inserted {
tx.rollback().await?;
return Ok((
StoredEvent::with_received_at(event.clone(), received_at, channel_id, false),
false,
));
}
if is_nip_rs {
sqlx::query(
"INSERT INTO parameterized_event_watermarks \
(community_id, kind, pubkey, d_tag, created_at, event_id) \
VALUES ($1, $2, $3, $4, $5, $6) \
ON CONFLICT (community_id, kind, pubkey, d_tag) DO UPDATE SET \
created_at = EXCLUDED.created_at, event_id = EXCLUDED.event_id",
)
.bind(community_id.as_uuid())
.bind(kind_i32)
.bind(pubkey_bytes.as_slice())
.bind(d_tag)
.bind(created_at)
.bind(incoming_id)
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
// Mentions are a denormalized index — safe outside the transaction.
if let Err(e) = crate::insert_mentions(&self.pool, community_id, event, channel_id).await {
tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}");
}
Ok((
StoredEvent::with_received_at(event.clone(), received_at, channel_id, true),
true,
))
}
}
/// A full API token record.
@@ -6014,6 +5773,456 @@ mod tests {
assert_eq!(live, 0, "watermark must block stale resurrection");
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn nip_rs_transaction_operation_restores_hard_delete_opt_in() {
use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
let db = setup_db().await;
let community = CommunityId::from_uuid(make_community(&db.pool).await);
let keys = Keys::generate();
let base = Timestamp::now().as_secs();
let replace_d_tag = format!("read-state:{}", "b".repeat(32));
let victim_d_tag = format!("read-state:{}", "c".repeat(32));
let event = |d_tag: &str, content: &str, timestamp: u64| {
EventBuilder::new(
Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16),
content,
)
.tags(vec![
Tag::parse(["d", d_tag]).expect("d tag"),
Tag::parse(["t", "read-state"]).expect("t tag"),
])
.custom_created_at(Timestamp::from(timestamp))
.sign_with_keys(&keys)
.expect("sign read state")
};
let old = event(&replace_d_tag, "old", base);
let new = event(&replace_d_tag, "new", base + 1);
let victim = event(&victim_d_tag, "victim", base);
assert!(
db.replace_parameterized_event(community, &old, &replace_d_tag, None)
.await
.expect("insert old head")
.1
);
assert!(
db.replace_parameterized_event(community, &victim, &victim_d_tag, None)
.await
.expect("insert victim head")
.1
);
let mut tx = db
.begin_transaction()
.await
.expect("begin caller transaction");
let result = db
.replace_parameterized_event_in_transaction(
&mut tx,
community,
&new,
&replace_d_tag,
None,
replaceable::ParameterizedReplacePrecondition::Unconditional,
)
.await
.expect("replace inside caller transaction");
assert_eq!(
result.status,
replaceable::ParameterizedReplaceStatus::Inserted
);
let leaked: Option<String> = sqlx::query_scalar(
"SELECT NULLIF(current_setting('buzz.nip_rs_hard_delete', true), '')",
)
.fetch_one(&mut *tx)
.await
.expect("read hard-delete opt-in after replacement");
assert_ne!(leaked.as_deref(), Some("on"));
let unauthorized = sqlx::query(
"DELETE 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(&victim_d_tag)
.execute(&mut *tx)
.await;
assert!(
unauthorized.is_err(),
"replacement opt-in must not authorize later caller SQL"
);
tx.rollback().await.expect("roll back caller transaction");
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn parameterized_replacement_in_existing_transaction_honors_revision_and_rollback() {
use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
let db = setup_db().await;
let community = CommunityId::from_uuid(make_community(&db.pool).await);
let keys = Keys::generate();
let d_tag = format!("transactional-project-{}", Uuid::new_v4().simple());
let base = Timestamp::now().as_secs();
let event = |content: &str, timestamp: u64| {
EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), content)
.tags(vec![Tag::parse(["d", d_tag.as_str()]).expect("d tag")])
.custom_created_at(Timestamp::from(timestamp))
.sign_with_keys(&keys)
.expect("sign project")
};
let old = event("old", base);
let new = event("new", base + 1);
assert!(
db.replace_parameterized_event(community, &old, &d_tag, None)
.await
.expect("insert old head")
.1
);
let mut tx = db.begin_transaction().await.expect("begin replacement tx");
let outcome = db
.replace_parameterized_event_in_transaction(
&mut tx,
community,
&new,
&d_tag,
None,
replaceable::ParameterizedReplacePrecondition::ExpectedRevision(
old.id.as_bytes().as_slice(),
),
)
.await
.expect("replace inside caller transaction");
assert_eq!(
outcome.status,
replaceable::ParameterizedReplaceStatus::Inserted
);
tx.rollback().await.expect("roll back replacement tx");
let live_id: Vec<u8> = sqlx::query_scalar(
"SELECT id FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \
AND d_tag=$4 AND deleted_at IS NULL",
)
.bind(community.as_uuid())
.bind(buzz_core::kind::KIND_PROJECT as i32)
.bind(keys.public_key().to_bytes())
.bind(&d_tag)
.fetch_one(&db.pool)
.await
.expect("load live head after rollback");
assert_eq!(live_id, old.id.as_bytes().to_vec());
let mut tx = db
.begin_transaction()
.await
.expect("begin stale revision tx");
let mismatch = db
.replace_parameterized_event_in_transaction(
&mut tx,
community,
&new,
&d_tag,
None,
replaceable::ParameterizedReplacePrecondition::ExpectedRevision(
[0x42; 32].as_slice(),
),
)
.await
.expect("evaluate stale revision");
assert_eq!(
mismatch.status,
replaceable::ParameterizedReplaceStatus::RevisionMismatch
);
tx.rollback().await.expect("roll back stale revision tx");
let missing_d_tag = format!("missing-project-{}", Uuid::new_v4().simple());
let missing = EventBuilder::new(
Kind::Custom(buzz_core::kind::KIND_PROJECT as u16),
"missing",
)
.tags(vec![
Tag::parse(["d", missing_d_tag.as_str()]).expect("missing d tag")
])
.custom_created_at(Timestamp::from(base + 2))
.sign_with_keys(&keys)
.expect("sign missing project");
let mut tx = db
.begin_transaction()
.await
.expect("begin missing revision tx");
let missing_result = db
.replace_parameterized_event_in_transaction(
&mut tx,
community,
&missing,
&missing_d_tag,
None,
replaceable::ParameterizedReplacePrecondition::ExpectedRevision(
[0x24; 32].as_slice(),
),
)
.await
.expect("evaluate missing revision");
assert_eq!(
missing_result.status,
replaceable::ParameterizedReplaceStatus::RevisionMissing
);
tx.rollback().await.expect("roll back missing revision tx");
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn parameterized_replacement_rolls_back_when_mention_indexing_fails() {
use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
let admin = PgPool::connect(&admin_url().await)
.await
.expect("connect admin");
let (pool, scratch_name) = create_scratch_db(&admin, "atomic_parameterized").await;
let db = Db::from_pool(pool.clone());
let community = CommunityId::from_uuid(make_community(&pool).await);
let keys = Keys::generate();
let mentioned = Keys::generate().public_key().to_hex();
let d_tag = format!("mention-project-{}", Uuid::new_v4().simple());
let tags = || {
vec![
Tag::parse(["d", d_tag.as_str()]).expect("d tag"),
Tag::parse(["p", mentioned.as_str()]).expect("p tag"),
]
};
let base = Timestamp::now().as_secs();
let old = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), "old")
.tags(tags())
.custom_created_at(Timestamp::from(base))
.sign_with_keys(&keys)
.expect("sign old project");
let new = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), "new")
.tags(tags())
.custom_created_at(Timestamp::from(base + 1))
.sign_with_keys(&keys)
.expect("sign new project");
assert!(
db.replace_parameterized_event(community, &old, &d_tag, None)
.await
.expect("insert old project")
.1
);
sqlx::query(
"CREATE FUNCTION reject_test_mention() RETURNS trigger AS $$ \
BEGIN RAISE EXCEPTION 'injected mention failure'; END; \
$$ LANGUAGE plpgsql",
)
.execute(&pool)
.await
.expect("create failure function");
sqlx::query(
"CREATE TRIGGER reject_test_mention BEFORE INSERT ON event_mentions \
FOR EACH ROW EXECUTE FUNCTION reject_test_mention()",
)
.execute(&pool)
.await
.expect("install failure injection");
let mut tx = db
.begin_transaction()
.await
.expect("begin caller transaction");
let error = db
.replace_parameterized_event_in_transaction(
&mut tx,
community,
&new,
&d_tag,
None,
replaceable::ParameterizedReplacePrecondition::Unconditional,
)
.await
.expect_err("mention failure must fail replacement");
assert!(error.to_string().contains("injected mention failure"));
let probe: i32 = sqlx::query_scalar("SELECT 1")
.fetch_one(&mut *tx)
.await
.expect("inner failure must leave caller transaction usable");
assert_eq!(probe, 1);
let live_id: Vec<u8> = sqlx::query_scalar(
"SELECT id FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \
AND d_tag=$4 AND deleted_at IS NULL",
)
.bind(community.as_uuid())
.bind(buzz_core::kind::KIND_PROJECT as i32)
.bind(keys.public_key().to_bytes())
.bind(&d_tag)
.fetch_one(&mut *tx)
.await
.expect("load live project after failed indexing");
assert_eq!(live_id, old.id.as_bytes().to_vec());
let new_rows: i64 =
sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id=$1 AND id=$2")
.bind(community.as_uuid())
.bind(new.id.as_bytes().as_slice())
.fetch_one(&mut *tx)
.await
.expect("count rolled-back project");
assert_eq!(new_rows, 0);
tx.commit().await.expect("commit usable caller transaction");
drop_scratch_db(&admin, pool, &scratch_name).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn parameterized_duplicate_restores_live_head_inside_caller_transaction() {
use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
let db = setup_db().await;
let community = CommunityId::from_uuid(make_community(&db.pool).await);
let keys = Keys::generate();
let d_tag = format!("duplicate-project-{}", Uuid::new_v4().simple());
let base = Timestamp::now().as_secs();
let event = |content: &str, timestamp: u64| {
EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), content)
.tags(vec![Tag::parse(["d", d_tag.as_str()]).expect("d tag")])
.custom_created_at(Timestamp::from(timestamp))
.sign_with_keys(&keys)
.expect("sign project")
};
let old = event("old-live-head", base);
let duplicate = event("soft-deleted-duplicate", base + 1);
assert!(
db.replace_parameterized_event(community, &duplicate, &d_tag, None)
.await
.expect("insert future duplicate")
.1
);
sqlx::query("UPDATE events SET deleted_at=NOW() WHERE community_id=$1 AND id=$2")
.bind(community.as_uuid())
.bind(duplicate.id.as_bytes().as_slice())
.execute(&db.pool)
.await
.expect("soft-delete duplicate row");
let mut seed_tx = db
.begin_transaction()
.await
.expect("begin seed transaction");
let (_, was_inserted) =
event::insert_event_in_transaction(&mut seed_tx, community, &old, None)
.await
.expect("insert older live head");
assert!(was_inserted);
seed_tx.commit().await.expect("commit older live head");
let mut tx = db
.begin_transaction()
.await
.expect("begin caller transaction");
let result = db
.replace_parameterized_event_in_transaction(
&mut tx,
community,
&duplicate,
&d_tag,
None,
replaceable::ParameterizedReplacePrecondition::Unconditional,
)
.await
.expect("evaluate soft-deleted duplicate");
assert_eq!(
result.status,
replaceable::ParameterizedReplaceStatus::Duplicate
);
let live_id: Vec<u8> = sqlx::query_scalar(
"SELECT id FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \
AND d_tag=$4 AND deleted_at IS NULL",
)
.bind(community.as_uuid())
.bind(buzz_core::kind::KIND_PROJECT as i32)
.bind(keys.public_key().to_bytes())
.bind(&d_tag)
.fetch_one(&mut *tx)
.await
.expect("caller transaction remains usable after duplicate");
assert_eq!(live_id, old.id.as_bytes().to_vec());
tx.rollback().await.expect("roll back caller transaction");
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn concurrent_parameterized_replacement_keeps_deterministic_head() {
use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
let db = setup_db().await;
let community = CommunityId::from_uuid(make_community(&db.pool).await);
let keys = Keys::generate();
let d_tag = format!("concurrent-project-{}", Uuid::new_v4().simple());
let created_at = Timestamp::now().as_secs();
let event = |content: &str, timestamp: u64| {
EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), content)
.tags(vec![Tag::parse(["d", d_tag.as_str()]).expect("d tag")])
.custom_created_at(Timestamp::from(timestamp))
.sign_with_keys(&keys)
.expect("sign project")
};
let first = event("first", created_at);
let second = event("second", created_at);
let expected = if first.id.as_bytes() < second.id.as_bytes() {
&first
} else {
&second
};
let (first_result, second_result) = tokio::join!(
db.replace_parameterized_event(community, &first, &d_tag, None),
db.replace_parameterized_event(community, &second, &d_tag, None),
);
let first_inserted = first_result.expect("first concurrent writer").1;
let second_inserted = second_result.expect("second concurrent writer").1;
assert!(
first_inserted || second_inserted,
"at least one concurrent writer must insert",
);
let live_ids: Vec<Vec<u8>> = sqlx::query_scalar(
"SELECT id FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \
AND d_tag=$4 AND deleted_at IS NULL",
)
.bind(community.as_uuid())
.bind(buzz_core::kind::KIND_PROJECT as i32)
.bind(keys.public_key().to_bytes())
.bind(&d_tag)
.fetch_all(&db.pool)
.await
.expect("load concurrent live head");
assert_eq!(live_ids, vec![expected.id.as_bytes().to_vec()]);
assert!(
!db.replace_parameterized_event(community, expected, &d_tag, None)
.await
.expect("replay winning event")
.1,
"replaying the live event must be idempotent",
);
let stale = event("stale", created_at.saturating_sub(1));
assert!(
!db.replace_parameterized_event(community, &stale, &d_tag, None)
.await
.expect("submit stale event")
.1,
"an older event must not replace the live head",
);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn mesh_status_replacement_keeps_one_physical_row() {

417
crates/buzz-db/src/replaceable.rs Normal file
View File

@@ -0,0 +1,417 @@
//! Replaceable-event persistence and coordinate locking.
use buzz_core::{CommunityId, StoredEvent};
use buzz_datastore_tracing::datastore_span;
use chrono::{DateTime, Utc};
use sqlx::{Acquire, Postgres, Transaction};
use uuid::Uuid;
use crate::{Db, DbError, Result};
/// Result category for a parameterized-replaceable event write.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ParameterizedReplaceStatus {
/// The incoming event was inserted as the coordinate's live head.
Inserted,
/// The exact event was already accepted.
Duplicate,
/// A newer event, or lower-ID same-second event, already dominates it.
Superseded,
/// A requested current revision has no live coordinate head.
RevisionMissing,
/// The live coordinate head differs from the requested revision.
RevisionMismatch,
/// An exact replay was required, but the event is not the live head.
ReplayOnlyMiss,
}
/// Structural precondition for a parameterized-replaceable write.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ParameterizedReplacePrecondition<'a> {
/// Apply normal NIP-33 ordering without a revision precondition.
Unconditional,
/// Require the live head to match this validated event ID.
ExpectedRevision(&'a [u8]),
/// Accept only an exact live-head replay and perform no mutation otherwise.
ExactReplayOnly,
}
/// Result of a transaction-bound parameterized-replaceable event write.
#[derive(Clone, Debug)]
pub struct ParameterizedReplaceResult {
/// Stored representation of the submitted event.
pub event: StoredEvent,
/// Whether and why the coordinate accepted the event.
pub status: ParameterizedReplaceStatus,
}
impl ParameterizedReplaceResult {
fn new(
event: &nostr::Event,
received_at: DateTime<Utc>,
channel_id: Option<Uuid>,
status: ParameterizedReplaceStatus,
) -> Self {
Self {
event: StoredEvent::with_received_at(
event.clone(),
received_at,
channel_id,
status == ParameterizedReplaceStatus::Inserted,
),
status,
}
}
}
/// Derive the transaction-scoped advisory-lock key for an event coordinate.
///
/// Hash collisions only add serialization; the SQL predicates still determine
/// which rows are read or changed.
pub(crate) fn event_replacement_lock_key(
community_id: CommunityId,
kind: i32,
pubkey: &[u8],
coordinate: Option<&[u8]>,
) -> i64 {
let mut hash: u64 = 0xcbf29ce484222325;
let kind_bytes = kind.to_le_bytes();
for bytes in [
community_id.as_uuid().as_bytes().as_slice(),
kind_bytes.as_slice(),
pubkey,
] {
for byte in bytes {
hash ^= *byte as u64;
hash = hash.wrapping_mul(0x100000001b3);
}
}
if let Some(coordinate) = coordinate {
for byte in coordinate {
hash ^= *byte as u64;
hash = hash.wrapping_mul(0x100000001b3);
}
}
hash as i64
}
/// Replace a parameterized event in a caller-owned transaction.
///
/// This function acquires a transaction-scoped advisory lock but never commits
/// or rolls back the outer transaction. The typed precondition can require the
/// current live head to have an exact event ID or restrict the operation to an
/// idempotent replay.
async fn replace_parameterized_event_in_transaction_impl(
tx: &mut Transaction<'_, Postgres>,
community_id: CommunityId,
event: &nostr::Event,
d_tag: &str,
channel_id: Option<Uuid>,
precondition: ParameterizedReplacePrecondition<'_>,
) -> Result<ParameterizedReplaceResult> {
let kind_i32 = buzz_core::kind::event_kind_i32(event);
let pubkey_bytes = event.pubkey.to_bytes();
let created_at_secs = event.created_at.as_secs() as i64;
let created_at = DateTime::from_timestamp(created_at_secs, 0)
.ok_or(DbError::InvalidTimestamp(created_at_secs))?;
let received_at = Utc::now();
let lock_key = event_replacement_lock_key(
community_id,
kind_i32,
pubkey_bytes.as_slice(),
Some(d_tag.as_bytes()),
);
sqlx::query("SELECT pg_advisory_xact_lock($1)")
.bind(lock_key)
.execute(&mut **tx)
.await?;
let d_tag_count = event
.tags
.iter()
.filter(|tag| tag.as_slice().first().is_some_and(|part| part == "d"))
.count();
let has_exact_d_tag = event.tags.iter().any(|tag| {
let parts = tag.as_slice();
parts.len() >= 2 && parts[0] == "d" && parts[1] == d_tag
});
let read_state_t_tag_count = event
.tags
.iter()
.filter(|tag| {
let parts = tag.as_slice();
parts.len() == 2 && parts[0] == "t" && parts[1] == "read-state"
})
.count();
let is_nip_rs = kind_i32 == buzz_core::kind::KIND_READ_STATE as i32
&& d_tag_count == 1
&& has_exact_d_tag
&& d_tag.strip_prefix("read-state:").is_some_and(|slot| {
slot.len() == 32
&& slot
.bytes()
.all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
})
&& read_state_t_tag_count == 1;
let is_buzz_mesh_status = kind_i32 == buzz_core::kind::KIND_BOOKMARK_SET as i32
&& d_tag.starts_with("buzz-mesh-member-status:")
&& event.tags.iter().any(|tag| {
let parts = tag.as_slice();
parts.len() == 2 && parts[0] == "k" && parts[1] == "buzz-mesh-status"
});
let hard_delete_superseded = is_nip_rs || is_buzz_mesh_status;
let existing: Option<(DateTime<Utc>, Vec<u8>)> = sqlx::query_as(
"SELECT created_at, id FROM events \
WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL \
ORDER BY created_at DESC, id ASC LIMIT 1",
)
.bind(community_id.as_uuid())
.bind(kind_i32)
.bind(pubkey_bytes.as_slice())
.bind(d_tag)
.fetch_optional(&mut **tx)
.await?;
let watermark: Option<(DateTime<Utc>, Vec<u8>)> = if is_nip_rs {
sqlx::query_as(
"SELECT created_at, event_id FROM parameterized_event_watermarks \
WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4",
)
.bind(community_id.as_uuid())
.bind(kind_i32)
.bind(pubkey_bytes.as_slice())
.bind(d_tag)
.fetch_optional(&mut **tx)
.await?
} else {
None
};
let incoming_id = event.id.as_bytes().as_slice();
if existing
.as_ref()
.is_some_and(|(_, existing_id)| existing_id.as_slice() == incoming_id)
|| watermark
.as_ref()
.is_some_and(|(_, event_id)| event_id.as_slice() == incoming_id)
{
return Ok(ParameterizedReplaceResult::new(
event,
received_at,
channel_id,
ParameterizedReplaceStatus::Duplicate,
));
}
if precondition == ParameterizedReplacePrecondition::ExactReplayOnly {
return Ok(ParameterizedReplaceResult::new(
event,
received_at,
channel_id,
ParameterizedReplaceStatus::ReplayOnlyMiss,
));
}
if let ParameterizedReplacePrecondition::ExpectedRevision(expected_revision) = precondition {
let status = match existing.as_ref() {
None => Some(ParameterizedReplaceStatus::RevisionMissing),
Some((_, existing_id)) if existing_id.as_slice() != expected_revision => {
Some(ParameterizedReplaceStatus::RevisionMismatch)
}
Some(_) => None,
};
if let Some(status) = status {
return Ok(ParameterizedReplaceResult::new(
event,
received_at,
channel_id,
status,
));
}
}
let dominated = existing
.iter()
.chain(watermark.iter())
.any(|(accepted_ts, accepted_id)| {
created_at < *accepted_ts
|| (created_at == *accepted_ts && incoming_id >= accepted_id.as_slice())
});
if dominated {
return Ok(ParameterizedReplaceResult::new(
event,
received_at,
channel_id,
ParameterizedReplaceStatus::Superseded,
));
}
let mut savepoint = tx.begin().await?;
if existing.is_some() {
let previous_nip_rs_hard_delete: Option<String> = if is_nip_rs {
sqlx::query_scalar(
"SELECT NULLIF(current_setting('buzz.nip_rs_hard_delete', true), '')",
)
.fetch_one(&mut *savepoint)
.await?
} else {
None
};
if is_nip_rs {
sqlx::query("SELECT set_config('buzz.nip_rs_hard_delete', 'on', true)")
.execute(&mut *savepoint)
.await?;
}
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(&mut *savepoint)
.await?;
if is_nip_rs {
let previous_value = previous_nip_rs_hard_delete.as_deref().unwrap_or_default();
sqlx::query("SELECT set_config('buzz.nip_rs_hard_delete', $1, true)")
.bind(previous_value)
.execute(&mut *savepoint)
.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(&mut *savepoint)
.await?;
}
}
}
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) \
ON CONFLICT DO NOTHING",
)
.bind(community_id.as_uuid())
.bind(incoming_id)
.bind(pubkey_bytes.as_slice())
.bind(created_at)
.bind(kind_i32)
.bind(&tags_json)
.bind(&event.content)
.bind(sig_bytes.as_slice())
.bind(received_at)
.bind(channel_id)
.bind(d_tag)
.bind(crate::event::extract_not_before(event))
.execute(&mut *savepoint)
.await?;
if insert_result.rows_affected() == 0 {
savepoint.rollback().await?;
return Ok(ParameterizedReplaceResult::new(
event,
received_at,
channel_id,
ParameterizedReplaceStatus::Duplicate,
));
}
if is_nip_rs {
sqlx::query(
"INSERT INTO parameterized_event_watermarks \
(community_id, kind, pubkey, d_tag, created_at, event_id) \
VALUES ($1, $2, $3, $4, $5, $6) \
ON CONFLICT (community_id, kind, pubkey, d_tag) DO UPDATE SET \
created_at = EXCLUDED.created_at, event_id = EXCLUDED.event_id",
)
.bind(community_id.as_uuid())
.bind(kind_i32)
.bind(pubkey_bytes.as_slice())
.bind(d_tag)
.bind(created_at)
.bind(incoming_id)
.execute(&mut *savepoint)
.await?;
}
crate::insert_mentions_in_transaction(&mut savepoint, community_id, event, channel_id).await?;
savepoint.commit().await?;
Ok(ParameterizedReplaceResult::new(
event,
received_at,
channel_id,
ParameterizedReplaceStatus::Inserted,
))
}
impl Db {
/// Replace a NIP-33 event inside a caller-owned transaction.
///
/// The caller owns commit or rollback. Requiring [`Transaction`] here and
/// in the internal state machine makes the advisory-lock contract explicit.
pub async fn replace_parameterized_event_in_transaction(
&self,
tx: &mut Transaction<'_, Postgres>,
community_id: CommunityId,
event: &nostr::Event,
d_tag: &str,
channel_id: Option<Uuid>,
precondition: ParameterizedReplacePrecondition<'_>,
) -> Result<ParameterizedReplaceResult> {
replace_parameterized_event_in_transaction_impl(
tx,
community_id,
event,
d_tag,
channel_id,
precondition,
)
.await
}
/// Atomically replace a NIP-33 parameterized replaceable event.
///
/// Replacement keys on `(kind, pubkey, d_tag)` across channels. The
/// highest timestamp wins; same-second ties use the lowest event ID.
#[datastore_span(name = "replace_parameterized_event", system = "postgresql")]
pub async fn replace_parameterized_event(
&self,
community_id: CommunityId,
event: &nostr::Event,
d_tag: &str,
channel_id: Option<Uuid>,
) -> Result<(StoredEvent, bool)> {
let mut tx = self.pool.begin().await?;
let result = self
.replace_parameterized_event_in_transaction(
&mut tx,
community_id,
event,
d_tag,
channel_id,
ParameterizedReplacePrecondition::Unconditional,
)
.await?;
let was_inserted = result.status == ParameterizedReplaceStatus::Inserted;
if was_inserted {
tx.commit().await?;
} else {
tx.rollback().await?;
}
Ok((result.event, was_inserted))
}
}

370
crates/buzz-relay/src/handlers/command_executor.rs
View File

@@ -105,8 +105,9 @@ async fn persist_command_event(
event: &Event,
channel_id_override: Option<Uuid>,
) -> Result<PersistResult, IngestError> {
let channel_id = channel_id_override.or_else(|| extract_channel_id(event));
use buzz_db::replaceable::{ParameterizedReplacePrecondition, ParameterizedReplaceStatus};
let channel_id = channel_id_override.or_else(|| extract_channel_id(event));
let mut tx = db
.begin_transaction()
.await
@@ -118,22 +119,8 @@ async fn persist_command_event(
IngestError::Rejected(format!("restricted: community writes are fenced: {error}"))
})?;
// INSERT with ON CONFLICT DO NOTHING — idempotency guard.
let id_bytes = event.id.as_bytes();
let pubkey_bytes = event.pubkey.to_bytes();
let sig_bytes = event.sig.serialize();
let tags_json = serde_json::to_value(&event.tags)
.map_err(|e| IngestError::Internal(format!("error: serialize tags: {e}")))?;
let kind_i32 = event.kind.as_u16() as i32;
let created_at_secs = event.created_at.as_secs() as i64;
let created_at = chrono::DateTime::from_timestamp(created_at_secs, 0).ok_or_else(|| {
IngestError::Rejected(format!("invalid: bad timestamp {created_at_secs}"))
})?;
let received_at = chrono::Utc::now();
// Extract d_tag for parameterized replaceable kinds (NIP-33).
let d_tag = buzz_db::event::extract_d_tag(event);
if let Some(ref d_tag) = d_tag {
if let Some(d_tag) = d_tag.as_deref() {
if d_tag.len() > buzz_db::event::D_TAG_MAX_LEN {
return Err(IngestError::Rejected(format!(
"invalid: d tag too long ({} bytes, max {})",
@@ -142,130 +129,81 @@ async fn persist_command_event(
)));
}
// Command kinds normally use plain insert semantics, but workflow
// definitions are NIP-33 events. Serialize writers for the same
// coordinate and reject stale writes before executing the domain
// mutation, otherwise old updates can overwrite newer workflow state.
let lock_key = {
let mut h: u64 = 0xcbf29ce484222325;
for b in tenant.community().as_uuid().as_bytes() {
h ^= *b as u64;
h = h.wrapping_mul(0x100000001b3);
}
for b in kind_i32.to_le_bytes() {
h ^= b as u64;
h = h.wrapping_mul(0x100000001b3);
}
for b in pubkey_bytes.as_slice() {
h ^= *b as u64;
h = h.wrapping_mul(0x100000001b3);
}
for b in d_tag.as_bytes() {
h ^= *b as u64;
h = h.wrapping_mul(0x100000001b3);
}
h as i64
let kind = event.kind.as_u16() as i32;
let (expected_revision, revision_error) = match parse_expected_workflow_revision(
kind,
extract_tag(event, "expected-revision").as_deref(),
) {
Ok(expected_revision) => (expected_revision, None),
Err(error) => (None, Some(error)),
};
sqlx::query("SELECT pg_advisory_xact_lock($1)")
.bind(lock_key)
.execute(tx.as_mut())
.await
.map_err(|e| IngestError::Internal(format!("error: lock event coordinate: {e}")))?;
let existing: Option<(chrono::DateTime<chrono::Utc>, Vec<u8>)> = sqlx::query_as(
"SELECT created_at, id FROM events \
WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL \
ORDER BY created_at DESC, id ASC LIMIT 1",
)
.bind(tenant.community().as_uuid())
.bind(kind_i32)
.bind(pubkey_bytes.as_slice())
.bind(d_tag)
.fetch_optional(tx.as_mut())
.await
.map_err(|e| IngestError::Internal(format!("error: query event coordinate: {e}")))?;
let incoming_id = event.id.as_bytes().as_slice();
if existing
.as_ref()
.is_some_and(|(_, existing_id)| existing_id.as_slice() == incoming_id)
{
return Ok(PersistResult::Duplicate);
}
let expected_revision = extract_tag(event, "expected-revision");
validate_workflow_revision(
kind_i32,
expected_revision.as_deref(),
existing.as_ref().map(|(_, id)| id.as_slice()),
)?;
if let Some((existing_ts, existing_id)) = existing {
let dominated = created_at < existing_ts
|| (created_at == existing_ts && incoming_id >= existing_id.as_slice());
if dominated {
if kind_i32 == KIND_WORKFLOW_DEF as i32 && expected_revision.is_some() {
return Err(IngestError::Rejected(
"conflict: workflow update was superseded; refresh and try again".into(),
));
}
return Ok(PersistResult::Duplicate);
}
sqlx::query(
"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",
let precondition = if revision_error.is_some() {
ParameterizedReplacePrecondition::ExactReplayOnly
} else if let Some(expected_revision) = expected_revision.as_deref() {
ParameterizedReplacePrecondition::ExpectedRevision(expected_revision)
} else {
ParameterizedReplacePrecondition::Unconditional
};
let result = db
.replace_parameterized_event_in_transaction(
&mut tx,
tenant.community(),
event,
d_tag,
channel_id,
precondition,
)
.bind(tenant.community().as_uuid())
.bind(kind_i32)
.bind(pubkey_bytes.as_slice())
.bind(d_tag)
.execute(tx.as_mut())
.await
.map_err(|e| IngestError::Internal(format!("error: replace old event: {e}")))?;
}
.map_err(|e| {
IngestError::Internal(format!("error: replace parameterized event: {e}"))
})?;
return match result.status {
ParameterizedReplaceStatus::Inserted => Ok(PersistResult::Inserted(tx)),
ParameterizedReplaceStatus::Duplicate => Ok(PersistResult::Duplicate),
ParameterizedReplaceStatus::Superseded
if kind == KIND_WORKFLOW_DEF as i32 && expected_revision.is_some() =>
{
Err(IngestError::Rejected(
"conflict: workflow update was superseded; refresh and try again".into(),
))
}
ParameterizedReplaceStatus::Superseded => Ok(PersistResult::Duplicate),
ParameterizedReplaceStatus::RevisionMissing => Err(IngestError::Rejected(
"conflict: workflow revision does not exist".into(),
)),
ParameterizedReplaceStatus::RevisionMismatch => Err(IngestError::Rejected(
"conflict: workflow changed since it was loaded".into(),
)),
ParameterizedReplaceStatus::ReplayOnlyMiss => match revision_error {
Some(error) => Err(error),
None => Err(IngestError::Internal(
"error: replay-only replacement lacked a revision error".into(),
)),
},
};
}
let result = sqlx::query(
r#"
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(tenant.community().as_uuid())
.bind(id_bytes.as_slice())
.bind(pubkey_bytes.as_slice())
.bind(created_at)
.bind(kind_i32)
.bind(&tags_json)
.bind(&event.content)
.bind(sig_bytes.as_slice())
.bind(received_at)
.bind(channel_id)
.bind(d_tag.as_deref())
.execute(tx.as_mut())
.await
.map_err(|e| IngestError::Internal(format!("error: insert event: {e}")))?;
if result.rows_affected() == 0 {
// Duplicate — rollback (implicit on drop) and signal idempotent success.
Ok(PersistResult::Duplicate)
} else {
let (_, was_inserted) =
buzz_db::event::insert_event_in_transaction(&mut tx, tenant.community(), event, channel_id)
.await
.map_err(|e| IngestError::Internal(format!("error: insert event: {e}")))?;
if was_inserted {
Ok(PersistResult::Inserted(tx))
} else {
Ok(PersistResult::Duplicate)
}
}
fn validate_workflow_revision(
fn parse_expected_workflow_revision(
kind: i32,
expected_revision: Option<&str>,
existing_id: Option<&[u8]>,
) -> Result<(), IngestError> {
) -> Result<Option<Vec<u8>>, IngestError> {
if kind != KIND_WORKFLOW_DEF as i32 {
return Ok(());
return Ok(None);
}
let expected_id = expected_revision
expected_revision
.map(|expected| {
let id = hex::decode(expected).map_err(|_| {
IngestError::Rejected("invalid: bad expected workflow revision".into())
@@ -277,18 +215,7 @@ fn validate_workflow_revision(
}
Ok(id)
})
.transpose()?;
match (expected_id.as_deref(), existing_id) {
(None, _) => Ok(()),
(Some(_), None) => Err(IngestError::Rejected(
"conflict: workflow revision does not exist".into(),
)),
(Some(expected), Some(existing)) if expected != existing => Err(IngestError::Rejected(
"conflict: workflow changed since it was loaded".into(),
)),
(Some(_), Some(_)) => Ok(()),
}
.transpose()
}
/// Extract all `p` tag values (hex pubkeys) from an event.
@@ -425,7 +352,7 @@ async fn handle_dm_open(
.await
.map_err(|e| IngestError::Internal(format!("error: db open_dm: {e}")))?;
// Commit: event + mutation succeeded atomically.
// Finalize the idempotency record after the separate mutation succeeds.
tx.commit()
.await
.map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?;
@@ -586,7 +513,7 @@ async fn handle_dm_add_member(
.await
.map_err(|e| IngestError::Internal(format!("error: db open_dm: {e}")))?;
// Commit: event + mutation succeeded atomically.
// Finalize the idempotency record after the separate mutation succeeds.
tx.commit()
.await
.map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?;
@@ -691,7 +618,7 @@ async fn handle_dm_hide(
.await
.map_err(|e| IngestError::Internal(format!("error: db hide_dm: {e}")))?;
// Commit: event + mutation succeeded atomically.
// Finalize the idempotency record after the separate mutation succeeds.
tx.commit()
.await
.map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?;
@@ -994,7 +921,7 @@ async fn handle_workflow_trigger(
.await
.map_err(|e| IngestError::Internal(format!("error: db create_workflow_run: {e}")))?;
// Commit: event + run creation succeeded atomically.
// Finalize the idempotency record after the separate run creation succeeds.
tx.commit()
.await
.map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?;
@@ -1169,7 +1096,7 @@ async fn handle_approval_grant(
));
}
// Commit: event + approval update succeeded atomically.
// Finalize the idempotency record after the separate approval update succeeds.
tx.commit()
.await
.map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?;
@@ -1280,7 +1207,7 @@ async fn handle_approval_deny(
));
}
// Commit: event + approval denial succeeded atomically.
// Finalize the idempotency record after the separate approval denial succeeds.
tx.commit()
.await
.map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?;
@@ -1489,57 +1416,40 @@ mod tests {
.expect("workflow event")
}
fn rejection_message(result: Result<(), IngestError>) -> String {
fn rejection_message(result: Result<Option<Vec<u8>>, IngestError>) -> String {
match result {
Err(IngestError::Rejected(message)) => message,
Err(IngestError::AuthFailed(message)) => panic!("unexpected auth failure: {message}"),
Err(IngestError::Internal(message)) => panic!("unexpected internal failure: {message}"),
Ok(()) => panic!("expected revision validation to fail"),
Ok(_) => panic!("expected revision parsing to fail"),
}
}
#[test]
fn workflow_revision_accepts_create_and_matching_update() {
let existing = [0x42; 32];
assert!(validate_workflow_revision(KIND_WORKFLOW_DEF as i32, None, None).is_ok());
assert!(validate_workflow_revision(
KIND_WORKFLOW_DEF as i32,
Some(&hex::encode(existing)),
Some(&existing),
)
.is_ok());
fn workflow_revision_parser_accepts_create_and_valid_update() {
let revision = [0x42; 32];
assert_eq!(
parse_expected_workflow_revision(KIND_WORKFLOW_DEF as i32, None)
.expect("tagless workflow"),
None
);
assert_eq!(
parse_expected_workflow_revision(
KIND_WORKFLOW_DEF as i32,
Some(&hex::encode(revision)),
)
.expect("valid revision"),
Some(revision.to_vec())
);
}
#[test]
fn workflow_revision_rejects_stale_and_malformed_updates() {
let existing = [0x42; 32];
let stale = [0x24; 32];
assert_eq!(
rejection_message(validate_workflow_revision(
KIND_WORKFLOW_DEF as i32,
Some(&hex::encode(stale)),
Some(&existing),
)),
"conflict: workflow changed since it was loaded",
);
assert!(
validate_workflow_revision(KIND_WORKFLOW_DEF as i32, None, Some(&existing)).is_ok(),
"tagless legacy workflow updates remain compatible during rollout",
);
fn workflow_revision_parser_rejects_malformed_values() {
for malformed in ["not-hex", "42"] {
assert_eq!(
rejection_message(validate_workflow_revision(
rejection_message(parse_expected_workflow_revision(
KIND_WORKFLOW_DEF as i32,
Some(malformed),
Some(&existing),
)),
"invalid: bad expected workflow revision",
);
assert_eq!(
rejection_message(validate_workflow_revision(
KIND_WORKFLOW_DEF as i32,
Some(malformed),
None,
)),
"invalid: bad expected workflow revision",
);
@@ -1547,14 +1457,11 @@ mod tests {
}
#[test]
fn workflow_revision_rejects_update_for_missing_coordinate() {
fn revision_tag_does_not_change_other_command_kinds() {
assert_eq!(
rejection_message(validate_workflow_revision(
KIND_WORKFLOW_DEF as i32,
Some(&hex::encode([0x42; 32])),
None,
)),
"conflict: workflow revision does not exist",
parse_expected_workflow_revision(KIND_DM_OPEN as i32, Some("not-hex"))
.expect("non-workflow revision tag"),
None
);
}
@@ -1567,6 +1474,25 @@ mod tests {
let created_at = Timestamp::now().as_secs();
let create = workflow_event(&keys, workflow_id, created_at, None, "create");
let missing_revision = hex::encode([0x24; 32]);
let missing_revision_update = workflow_event(
&keys,
Uuid::new_v4(),
created_at,
Some(&missing_revision),
"missing-revision",
);
let error = match persist_command_event(&db, &tenant, &missing_revision_update, None).await
{
Err(error) => error,
Ok(_) => panic!("missing revision must not create a workflow"),
};
assert!(matches!(
error,
IngestError::Rejected(ref message)
if message == "conflict: workflow revision does not exist"
));
let PersistResult::Inserted(tx) = persist_command_event(&db, &tenant, &create, None)
.await
.expect("persist create")
@@ -1621,6 +1547,23 @@ mod tests {
PersistResult::Duplicate
));
let stale_revision_update = workflow_event(
&keys,
workflow_id,
created_at + 1,
Some(&create_revision),
"stale-revision",
);
let error = match persist_command_event(&db, &tenant, &stale_revision_update, None).await {
Err(error) => error,
Ok(_) => panic!("stale revision must not replace the current workflow"),
};
assert!(matches!(
error,
IngestError::Rejected(ref message)
if message == "conflict: workflow changed since it was loaded"
));
let error = match persist_command_event(&db, &tenant, &dominated_update, None).await {
Err(error) => error,
Ok(_) => panic!("distinct dominated CAS update must not report duplicate success"),
@@ -1632,8 +1575,55 @@ mod tests {
));
}
#[test]
fn revision_tag_does_not_change_other_command_kinds() {
assert!(validate_workflow_revision(KIND_DM_OPEN as i32, Some("not-hex"), None).is_ok());
#[tokio::test]
#[ignore = "requires Postgres"]
async fn workflow_persistence_replays_legacy_malformed_revision_before_validation() {
let (db, tenant) = persistence_test_context().await;
let keys = Keys::generate();
let workflow_id = Uuid::new_v4();
let created_at = Timestamp::now().as_secs();
let legacy = workflow_event(
&keys,
workflow_id,
created_at,
Some("not-hex"),
"legacy-malformed",
);
let mut tx = db.begin_transaction().await.expect("begin legacy seed");
let (_, was_inserted) = buzz_db::event::insert_event_in_transaction(
&mut tx,
tenant.community(),
&legacy,
extract_channel_id(&legacy),
)
.await
.expect("seed legacy workflow event");
assert!(was_inserted);
tx.commit().await.expect("commit legacy seed");
assert!(matches!(
persist_command_event(&db, &tenant, &legacy, None)
.await
.expect("exact legacy replay must remain idempotent"),
PersistResult::Duplicate
));
let distinct = workflow_event(
&keys,
workflow_id,
created_at + 1,
Some("not-hex"),
"distinct-malformed",
);
let error = match persist_command_event(&db, &tenant, &distinct, None).await {
Err(error) => error,
Ok(_) => panic!("distinct malformed revision must remain rejected"),
};
assert!(matches!(
error,
IngestError::Rejected(ref message)
if message == "invalid: bad expected workflow revision"
));
}
}