From 78b5dae1f608a33850137b5be5a1909390035cce Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Wed, 28 Oct 2026 11:38:30 -0400 Subject: [PATCH] fix(relay): store a kind:9000 and its membership writes in one transaction (#8606) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to #8575. ## Problem Today a kind:9000 is stored first and the membership write runs afterwards, on its own. If that write fails, the relay logs it and still returns an accepted OK. A resend of the same event then returns `duplicate:`, so the person is never added. This is the catch-log-and-accept failure that Review-Proven Rules #1 and #5 describe. ## Change kind:9000 now uses the same atomic path as kind:9002 (`persisted_atomically` in ingest). - **`buzz-db`:** new `put_members_with_event`, which runs one transaction: 1. Take the channel's membership lock, before the event insert, the TTL lock and the channel row. This matches the order the other membership writers use, so it adds no new lock cycle. 2. Store the event. If it's already stored, roll back and return `Duplicate`. 3. Run the existing `put_members` decisions and upsert. These now live in a shared `put_members_on` step on the transaction. 4. Commit. - **Failure:** nothing is stored and the client gets an `error:` rejection, so it can resend the same event and have it applied. - **Everyone refused:** if the decisions under the lock refuse every person, nothing is stored and the event is rejected with the same text validation gives. Before, a single-person add refused at that point was accepted with an empty message. - **After commit:** the live effects stay as they were. They're best-effort and read the committed state: cache invalidation, `member_joined` notices, member-added notifications and 39002 discovery. - **`NOSTR.md`:** now says a failed kind:9000 can be resent as is. ## What got simpler - kind:9000 no longer has its own store-then-apply path in ingest. It shares the atomic branch with 9002 and workflow deletion. - `apply_put_user`'s "write failed, report it as a per-person rejection inside an accepted OK" branch is gone. - The `Db::put_members` wrapper is deleted. Its only caller was the kind:9000 path. ## Identified debt (not in this PR) kind:9001 (remove), 9021 (join) and 9022 (leave) have the same gap. Each stores the event, then runs the membership write in `handle_side_effects`, and a failure is only logged. kind:9008 and 9005 follow the same store-then-apply pattern. These can move onto the same atomic shape in a follow-up. ## Tests - `failed_put_user_stores_nothing` (relay, Postgres): a test trigger fails the second person's row in a two-person add, after the event row and the first person were already written in the same transaction. It checks: - Both the send and a resend are rejected with `error:`. - No kind:9000 is stored, nobody is added, and no `member_joined` notice is posted. - With the trigger dropped, the same signed event adds both people. - One more resend returns `duplicate:` and posts no second notice. - `put_members_with_event_stores_the_event_with_its_writes` (buzz-db): checks the three outcomes: - Refused: nothing is stored. - Applied: the event is stored and the person is written. - Duplicate: the roster is left alone. - `put_user_refused_under_the_lock_stores_nothing` (relay, Postgres): a test trigger removes the sender from a private channel inside the write transaction, so validation passes and the decision under the lock refuses. For a single add and a batch, it checks the `invalid:` rejection, that nothing is stored and no notice is posted, and that the same event applies once the trigger is gone. - Mutation checks, run in Blox gates: - At `d18e88fd`: committing the refused event fails the buzz-db test; applying a duplicate fails the first two tests; storing the event in its own transaction fails the first two tests. - At `3b106623`: turning the refusal under the lock into `error:` fails the new refusal test, and so does committing the refused event. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: Luke Tornquist Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-authored-by: Claude Opus 5.5 --- NOSTR.md | 2 +- crates/buzz-db/src/store/channel_members.rs | 234 ++++++++++++++++-- crates/buzz-relay/src/handlers/ingest.rs | 39 ++- .../src/handlers/put_user_postgres_tests.rs | 201 +++++++++++++++ .../buzz-relay/src/handlers/side_effects.rs | 81 +++--- 5 files changed, 502 insertions(+), 55 deletions(-) diff --git a/NOSTR.md b/NOSTR.md index 507923d70c..61f3e49d3b 100644 --- a/NOSTR.md +++ b/NOSTR.md @@ -51,7 +51,7 @@ PGPASSWORD=buzz_dev psql -h localhost -U buzz -d buzz -c \ | **Deletions (kind:5)** | ✅ | Standard NIP-09; self-authored only. `#h` optional, `#e` required | | **User profiles (kind:0)** | ✅ | NIP-01 metadata; synced to users table (display_name, avatar, about, NIP-05). NIP-05 handles must canonicalize to this relay's domain — off-domain or invalid handles are silently cleared. If a NIP-05 handle collides with another user's (UNIQUE constraint), the handle is skipped but other profile fields (display_name, avatar, about) are still synced. | | **Group creation (kind:9007)** | ✅ | NIP-29; include `name` tag, optional `visibility`, `channel_type`, label `t` and `unlisted` tags | -| **Add user (kind:9000)** | ✅ | Open: any user, subject to target's `channel_add_policy` (`owner_only`/`nobody` can block). Private: owner/admin only. Self-add bypasses agent policy but not private-channel auth. A guest cannot add anyone. A new member with no role tag gets the channel's `join_role`; a non-member adding themselves may ask for `join_role` or a lower role. One event may name up to 100 people: one `p` tag each, with an optional role as the `p` tag's third element (NIP-29), falling back to the event's `role` tag. Each person is checked and added on their own. When the event names several people, the OK message is `members: ` followed by JSON listing the non-empty `added`, `updated` (role changed), `already_member` and `rejected` (`[{"reason", "pubkeys"}]`) results. The event is refused only when it is malformed or nobody it names may be added; then one person keeps the single-add error, and several get `invalid: members: {…}`. A resent event returns `duplicate:` without results; to retry an `error:` rejection, send a new event. Re-adding someone at their current role posts no second `member_joined` notice. | +| **Add user (kind:9000)** | ✅ | Open: any user, subject to target's `channel_add_policy` (`owner_only`/`nobody` can block). Private: owner/admin only. Self-add bypasses agent policy but not private-channel auth. A guest cannot add anyone. A new member with no role tag gets the channel's `join_role`; a non-member adding themselves may ask for `join_role` or a lower role. One event may name up to 100 people: one `p` tag each, with an optional role as the `p` tag's third element (NIP-29), falling back to the event's `role` tag. Each person is checked and added on their own. When the event names several people, the OK message is `members: ` followed by JSON listing the non-empty `added`, `updated` (role changed), `already_member` and `rejected` (`[{"reason", "pubkeys"}]`) results. The event is refused only when it is malformed or nobody it names may be added; then one person keeps the single-add error, and several get `invalid: members: {…}`. The event and its membership changes are stored together, so an `error:` rejection stores nothing and the same event can be resent; a resent stored event returns `duplicate:` without results. Re-adding someone at their current role posts no second `member_joined` notice. | | **Remove user (kind:9001)** | ✅ | Self-remove allowed (with last-owner guard). Removing others: owner/admin only. | | **Edit group metadata (kind:9002)** | ✅ | `name`/`about`/label `t`/`visibility`/`ttl`/`archived`/`join_role`/`unlisted` tags: owner/admin. `topic`/`purpose` tags: any member except a guest. Buzz applies 9002 as a partial update: an absent tag leaves that setting unchanged. | | **Admin delete event (kind:9005)** | ✅ | Event author can always delete own. Otherwise owner/admin required. Target must be in same channel. | diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 0922be5948..14f865c998 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -791,14 +791,7 @@ pub async fn put_members( targets: &[MemberPut<'_>], invited_by: Option<&[u8]>, ) -> Result>> { - let mut seen = std::collections::HashSet::with_capacity(targets.len()); - if let Some(dup) = targets.iter().find(|t| !seen.insert(t.pubkey)) { - return Err(DbError::InvalidData(format!( - "pubkey {} appears more than once", - hex::encode(dup.pubkey) - ))); - } - + check_distinct_targets(targets)?; let mut tx = crate::begin_community_write_transaction( pool, community_id, @@ -809,7 +802,100 @@ pub async fn put_members( // First statement: serialize the whole decision / owner-count / upsert // sequence against concurrent membership writes on this channel. acquire_channel_membership_lock(tx.conn(), community_id, channel_id).await?; + let results = put_members_on(&mut tx, channel_id, targets, invited_by).await?; + tx.commit().await?; + Ok(results) +} +/// The result of [`Db::put_members_with_event`]. +pub enum MemberPutEventWrite { + /// The event was stored and at least one person was written. One result + /// per target, in order, as from [`put_members`]. + Applied { + /// The stored kind:9000 event. + stored: buzz_core::StoredEvent, + /// One result per target. + results: Vec>, + }, + /// The event was already stored. Nothing changed: its membership changes + /// committed with it the first time. + Duplicate(buzz_core::StoredEvent), + /// Every person was refused, so nothing was stored. One error per target. + Refused(Vec), +} + +/// Store a kind:9000 `event` and apply its [`put_members`] writes in one +/// transaction. If any write fails, nothing is stored, so the client can +/// resend the same event and have it applied. +/// +/// The membership lock comes first, as in [`put_members`] and the channel +/// settings writers, so a second copy of the event waits for the first and +/// then finds it stored. +pub(crate) async fn put_members_with_event( + pool: &PgPool, + push_enqueue: Option, + community_id: CommunityId, + channel_id: Uuid, + event: &nostr::Event, + targets: &[MemberPut<'_>], + thread_meta: Option>, +) -> Result { + check_distinct_targets(targets)?; + let mut tx = crate::begin_community_event_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; + tx.set_push_enqueue(push_enqueue); + acquire_channel_membership_lock(tx.conn(), community_id, channel_id).await?; + + // The same writes as `Db::insert_event_with_thread_metadata`. + let (stored, inserted) = crate::event::insert_event_with_thread_metadata_tx( + &mut tx, + event, + Some(channel_id), + thread_meta, + ) + .await?; + if !inserted { + tx.rollback().await?; + return Ok(MemberPutEventWrite::Duplicate(stored)); + } + crate::insert_mentions_in_transaction(&mut tx, event, Some(channel_id)).await?; + + let actor = event.pubkey.to_bytes(); + let results = put_members_on(&mut tx, channel_id, targets, Some(&actor)).await?; + if results.iter().all(Result::is_err) { + tx.rollback().await?; + return Ok(MemberPutEventWrite::Refused( + results.into_iter().filter_map(Result::err).collect(), + )); + } + tx.commit().await?; + Ok(MemberPutEventWrite::Applied { stored, results }) +} + +fn check_distinct_targets(targets: &[MemberPut<'_>]) -> Result<()> { + let mut seen = std::collections::HashSet::with_capacity(targets.len()); + match targets.iter().find(|t| !seen.insert(t.pubkey)) { + Some(dup) => Err(DbError::InvalidData(format!( + "pubkey {} appears more than once", + hex::encode(dup.pubkey) + ))), + None => Ok(()), + } +} + +/// The decisions and upsert of [`put_members`], on a transaction that holds +/// `channel_id`'s membership lock. The caller owns the commit. +async fn put_members_on( + tx: &mut AdmittedTx, + channel_id: Uuid, + targets: &[MemberPut<'_>], + invited_by: Option<&[u8]>, +) -> Result>> { + let community_id = tx.community(); let channel = get_channel_tx(tx.conn(), community_id, channel_id).await?; let private = channel.visibility == "private"; let join_role = buzz_core::channel::ChannelJoinRole::parse_fail_closed(&channel.join_role); @@ -945,7 +1031,6 @@ pub async fn put_members( written.insert(member.pubkey.clone(), member); } } - tx.commit().await?; Ok(targets .iter() @@ -2115,17 +2200,28 @@ impl Db { .await } - /// [`Db::put_member`] for several people in one transaction (see - /// [`put_members`]). - #[datastore_span(name = "put_members", system = "postgresql")] - pub async fn put_members( + /// Stores a kind:9000 `event` and applies its [`put_members`] writes, + /// with the event's author as the actor, in one transaction (see + /// [`put_members_with_event`]). + #[datastore_span(name = "put_members_with_event", system = "postgresql")] + pub async fn put_members_with_event( &self, community_id: CommunityId, channel_id: Uuid, + event: &nostr::Event, targets: &[MemberPut<'_>], - invited_by: Option<&[u8]>, - ) -> Result>> { - put_members(&self.pool, community_id, channel_id, targets, invited_by).await + thread_meta: Option>, + ) -> Result { + put_members_with_event( + &self.pool, + self.push_enqueue.clone(), + community_id, + channel_id, + event, + targets, + thread_meta, + ) + .await } /// Removes a member from a channel. @@ -3476,6 +3572,112 @@ mod postgres_tests { assert!(matches!(err, DbError::InvalidData(_)), "{err:?}"); } + /// `put_members_with_event` stores the event only with the writes it + /// carries: a stored event applies its people, a resend changes nothing, + /// and an event whose every person is refused under the lock is not + /// stored. Mutations: commit when everyone is refused → the refused event + /// is stored → RED; write people on a duplicate → the removed person is + /// back → RED. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn put_members_with_event_stores_the_event_with_its_writes() { + let pool = setup_pool().await; + let community_id = make_test_community(&pool).await; + let community = CommunityId::from_uuid(community_id); + let (owner, outsider) = (Keys::generate(), Keys::generate()); + let (owner_pk, outsider_pk) = ( + owner.public_key().to_bytes().to_vec(), + outsider.public_key().to_bytes().to_vec(), + ); + let person = random_pubkey(); + for pk in [&owner_pk, &outsider_pk, &person] { + ensure_user(&pool, community, pk) + .await + .expect("ensure user"); + } + let channel = create_test_channel( + &pool, + community_id, + "put-members-event", + ChannelType::Stream, + ChannelVisibility::Private, + None, + &owner_pk, + None, + ) + .await + .expect("create channel"); + let event = |actor: &Keys| { + nostr::EventBuilder::new(nostr::Kind::Custom(9000), "") + .tags([ + nostr::Tag::parse(["h", &channel.id.to_string()]).unwrap(), + nostr::Tag::parse(["p", &hex::encode(&person)]).unwrap(), + ]) + .sign_with_keys(actor) + .expect("sign 9000") + }; + let targets = [MemberPut { + pubkey: &person, + requested: None, + }]; + let stored = |id: nostr::EventId| { + let pool = pool.clone(); + async move { + crate::event::get_event_by_id(&pool, community, id.as_bytes()) + .await + .expect("read event") + .is_some() + } + }; + + // A non-member may not add anyone to a private channel. + let refused = event(&outsider); + let write = + put_members_with_event(&pool, None, community, channel.id, &refused, &targets, None) + .await + .expect("refused write"); + assert!( + matches!(&write, MemberPutEventWrite::Refused(errors) + if matches!(errors.as_slice(), [DbError::AccessDenied(_)])), + "expected a refusal" + ); + assert!(!stored(refused.id).await, "a refused event is not stored"); + assert_eq!( + get_member_role(&pool, community, channel.id, &person) + .await + .expect("read role"), + None + ); + + let added = event(&owner); + let write = + put_members_with_event(&pool, None, community, channel.id, &added, &targets, None) + .await + .expect("applied write"); + let MemberPutEventWrite::Applied { results, .. } = write else { + panic!("expected the event to apply"); + }; + let put = results[0].as_ref().expect("person added"); + assert_eq!(put.prior_role, None); + assert!(stored(added.id).await); + + // A resend finds the event stored and leaves the roster alone. + remove_member(&pool, community, channel.id, &person, &owner_pk) + .await + .expect("remove person"); + let write = + put_members_with_event(&pool, None, community, channel.id, &added, &targets, None) + .await + .expect("duplicate write"); + assert!(matches!(write, MemberPutEventWrite::Duplicate(_))); + assert_eq!( + get_member_role(&pool, community, channel.id, &person) + .await + .expect("read role"), + None + ); + } + /// `put_member` enforces admission itself, so a caller that skips the /// relay validator still cannot join an announce channel above /// `join_role`, and a guest still cannot add anyone. No role tag keeps an diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index f886b900b0..c2ab9c3ba5 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -3053,7 +3053,7 @@ async fn ingest_event_inner( validate_huddle_lifecycle_event(tenant, state, &event, kind_u32).await?; // kind:9000 is checked person by person; the plan carries each result - // through storage to `apply_put_user`. + // to `persist_put_user`, which stores the event and applies it. let put_user = if kind_u32 == KIND_NIP29_PUT_USER { Some( crate::handlers::side_effects::validate_put_user(tenant, &event, state) @@ -3540,6 +3540,7 @@ async fn ingest_event_inner( }; let workflow_deletion = crate::handlers::side_effects::is_workflow_deletion(&event); + let mut message = String::new(); let (stored_event, was_inserted) = if let Some(create) = channel_create.as_ref() { let creator = event.pubkey.to_bytes(); let channel = buzz_db::channel::NewChannel { @@ -3590,6 +3591,28 @@ async fn ingest_event_inner( crate::handlers::side_effects::persist_workflow_deletion(tenant, &event, state) .await .map_err(|e| IngestError::Internal(format!("error: workflow deletion failed: {e}")))? + } else if let Some(plan) = put_user { + // One commit stores the event and applies its membership changes. A + // failure stores nothing, so the client is told and can resend it. + // Refusing everyone under the membership lock is the client's error; + // anything else is ours. + let thread_params = thread_meta.as_ref().map(|m| m.as_params()); + let (stored, inserted, ok_message) = crate::handlers::side_effects::persist_put_user( + tenant, + &event, + state, + plan, + thread_params, + ) + .await + .map_err(|e| match e.downcast_ref::() { + Some(buzz_db::DbError::AccessDenied(reason)) => { + IngestError::Rejected(format!("invalid: {reason}")) + } + _ => IngestError::Internal(format!("error: membership write failed: {e}")), + })?; + message = ok_message; + (stored, inserted) } else if kind_u32 == KIND_NIP29_EDIT_METADATA { // One commit stores the event and applies its channel changes. A // failure stores nothing, so the client is told and can resend it. @@ -3694,14 +3717,12 @@ async fn ingest_event_inner( }); } - // Workflow deletion and kind:9002 applied their changes in the insert's - // own transaction above. - let persisted_atomically = workflow_deletion || kind_u32 == KIND_NIP29_EDIT_METADATA; - let mut message = String::new(); - if let Some(plan) = put_user { - message = crate::handlers::side_effects::apply_put_user(tenant, &event, state, plan).await; - } else if !persisted_atomically && crate::handlers::side_effects::is_side_effect_kind(kind_u32) - { + // Workflow deletion, kind:9000 and kind:9002 applied their changes in the + // insert's own transaction above. + let persisted_atomically = workflow_deletion + || kind_u32 == KIND_NIP29_PUT_USER + || kind_u32 == KIND_NIP29_EDIT_METADATA; + if !persisted_atomically && crate::handlers::side_effects::is_side_effect_kind(kind_u32) { if let Err(e) = crate::handlers::side_effects::handle_side_effects(tenant, kind_u32, &event, state) .await diff --git a/crates/buzz-relay/src/handlers/put_user_postgres_tests.rs b/crates/buzz-relay/src/handlers/put_user_postgres_tests.rs index 07d4f29828..637f7b42e6 100644 --- a/crates/buzz-relay/src/handlers/put_user_postgres_tests.rs +++ b/crates/buzz-relay/src/handlers/put_user_postgres_tests.rs @@ -546,3 +546,204 @@ async fn put_user_decides_each_person_after_the_ones_before() { Some("owner") ); } + +async fn put_user_events(state: &AppState, tenant: &TenantContext, channel_id: Uuid) -> usize { + state + .db + .query_events(&EventQuery { + kinds: Some(vec![buzz_core::kind::KIND_NIP29_PUT_USER as i32]), + channel_id: Some(channel_id), + ..EventQuery::for_community(tenant.community()) + }) + .await + .expect("put-user events") + .len() +} + +/// A membership write that fails stores neither the event nor anyone it +/// names, and the same event applies in full once the failure is gone. A test +/// trigger fails the second person's row after the event row and the first +/// person were written in the same transaction. Before, the relay stored the +/// event, logged the failure and accepted it; a resend then returned +/// `duplicate:` and nobody was ever added. Mutations: store the event in its +/// own transaction → RED (event stored after the failure); accept the event +/// when the write fails → RED; apply a duplicate → RED (second notice). +#[tokio::test] +#[ignore = "requires Postgres"] +async fn failed_put_user_stores_nothing() { + let state = state().await; + let tenant = community(&state, "put-user-atomic-fail").await; + let (owner, person, failing) = (Keys::generate(), Keys::generate(), Keys::generate()); + let channel_id = create_channel(&state, &tenant, &owner, &[]).await; + let (person_hex, failing_hex) = (person.public_key().to_hex(), failing.public_key().to_hex()); + + // Each test runs in its own database, so this trigger affects only it. It + // fails the admin row, which only `failing` asks for. + for statement in [ + "CREATE FUNCTION put_user_atomic_fail() RETURNS trigger LANGUAGE plpgsql AS \ + $$ BEGIN RAISE EXCEPTION 'injected membership failure'; END $$", + "CREATE TRIGGER put_user_atomic_fail BEFORE INSERT ON channel_members \ + FOR EACH ROW WHEN (NEW.role = 'admin') EXECUTE FUNCTION put_user_atomic_fail()", + ] { + sqlx::query(statement) + .execute(state.db.pool()) + .await + .expect("install failure trigger"); + } + + let event = put_user( + &owner, + channel_id, + &[&[&person_hex], &[&failing_hex, "admin"]], + &[], + ); + for attempt in ["first send", "resend"] { + match ingest(&state, &tenant, event.clone(), &owner).await { + Ok(result) => panic!( + "{attempt}: a failed put-user was reported as {}: {:?}", + if result.accepted { + "accepted" + } else { + "refused" + }, + result.message + ), + Err(error) => assert!( + format!("{error:?}").contains("membership write failed"), + "{attempt}: {error:?}" + ), + } + assert_eq!( + put_user_events(&state, &tenant, channel_id).await, + 0, + "{attempt}" + ); + for keys in [&person, &failing] { + assert_eq!( + role_of(&state, &tenant, channel_id, keys).await, + None, + "{attempt}: nobody is added" + ); + } + assert_eq!( + joined_notices(&state, &tenant, channel_id, &person).await, + 0, + "{attempt}: no notice announces an add that did not happen" + ); + } + + // With the failure gone, the same signed event applies in full. + sqlx::query("DROP TRIGGER put_user_atomic_fail ON channel_members") + .execute(state.db.pool()) + .await + .expect("drop failure trigger"); + let retry = ingest(&state, &tenant, event.clone(), &owner) + .await + .expect("retry"); + assert!(retry.accepted, "{}", retry.message); + assert_eq!( + report(&retry.message), + serde_json::json!({ "added": [person_hex, failing_hex] }) + ); + assert_eq!( + role_of(&state, &tenant, channel_id, &person) + .await + .as_deref(), + Some("member") + ); + assert_eq!( + role_of(&state, &tenant, channel_id, &failing) + .await + .as_deref(), + Some("admin") + ); + assert_eq!(put_user_events(&state, &tenant, channel_id).await, 1); + + // A resend of the stored event is a duplicate and changes nothing. + let resend = ingest(&state, &tenant, event, &owner) + .await + .expect("resend"); + assert!(resend.accepted); + assert_eq!(resend.message, "duplicate:"); + assert_eq!( + joined_notices(&state, &tenant, channel_id, &person).await, + 1 + ); +} + +/// A refusal decided under the membership lock, after validation passed, +/// rejects the event with `invalid:` and stores nothing, for a single add and +/// for a batch. A trigger removes the actor inside the write transaction, so +/// validation sees an owner and the locked decision sees a non-member of a +/// private channel. +/// Mutations: map the under-lock `AccessDenied` to `error:` → RED; commit the +/// refused event → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn put_user_refused_under_the_lock_stores_nothing() { + let state = state().await; + let tenant = community(&state, "put-user-locked-refusal").await; + let (owner, first, second) = (Keys::generate(), Keys::generate(), Keys::generate()); + let channel_id = create_channel(&state, &tenant, &owner, &[&["visibility", "private"]]).await; + let (first_hex, second_hex) = (first.public_key().to_hex(), second.public_key().to_hex()); + + // Each test runs in its own database, so this trigger affects only it. The + // removal rolls back with the refused transaction. + for statement in [ + "CREATE FUNCTION put_user_revoke_actor() RETURNS trigger LANGUAGE plpgsql AS \ + $$ BEGIN UPDATE channel_members SET removed_at = NOW() \ + WHERE community_id = NEW.community_id AND channel_id = NEW.channel_id \ + AND pubkey = NEW.pubkey AND removed_at IS NULL; RETURN NULL; END $$", + "CREATE TRIGGER put_user_revoke_actor AFTER INSERT ON events \ + FOR EACH ROW WHEN (NEW.kind = 9000) EXECUTE FUNCTION put_user_revoke_actor()", + ] { + sqlx::query(statement) + .execute(state.db.pool()) + .await + .expect("install revoke trigger"); + } + + let reason = "private channel requires an invite from an active member"; + let single = put_user(&owner, channel_id, &[&[&first_hex]], &[]); + let batch = put_user(&owner, channel_id, &[&[&first_hex], &[&second_hex]], &[]); + for event in [single, batch.clone()] { + let message = match ingest(&state, &tenant, event, &owner).await { + Err(IngestError::Rejected(message)) => message, + Err(error) => panic!("expected an invalid: rejection, got {error:?}"), + Ok(result) => panic!("expected a rejection, got OK {}", result.message), + }; + let body = message.strip_prefix("invalid: ").expect("invalid prefix"); + if body.starts_with("members: ") { + assert_eq!( + report(body), + serde_json::json!({ + "rejected": [{ "reason": reason, "pubkeys": [first_hex, second_hex] }], + }) + ); + } else { + assert_eq!(body, reason); + } + assert_eq!(put_user_events(&state, &tenant, channel_id).await, 0); + for keys in [&first, &second] { + assert_eq!(role_of(&state, &tenant, channel_id, keys).await, None); + assert_eq!(joined_notices(&state, &tenant, channel_id, keys).await, 0); + } + assert_eq!( + role_of(&state, &tenant, channel_id, &owner) + .await + .as_deref(), + Some("owner"), + "the actor's removal rolls back with the refusal" + ); + } + + // Nothing was stored, so the same signed event applies once the actor + // keeps their role. + sqlx::query("DROP TRIGGER put_user_revoke_actor ON events") + .execute(state.db.pool()) + .await + .expect("drop revoke trigger"); + let retry = ingest(&state, &tenant, batch, &owner).await.expect("retry"); + assert!(retry.accepted, "{}", retry.message); + assert_eq!(put_user_events(&state, &tenant, channel_id).await, 1); +} diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 3fa7036b40..101ac4bc24 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -335,9 +335,9 @@ pub async fn handle_side_effects( match kind { 0 => handle_kind0_profile(tenant, event, state).await, 5 => handle_standard_deletion_event(tenant, event, state).await, - // kind:9000 belongs to the put-user path in ingest, which reports - // each person's result. - 9000 => Err(anyhow::anyhow!("kind:9000 requires apply_put_user")), + // kind:9000 belongs to the atomic put-user path in ingest, which + // reports each person's result. + 9000 => Err(anyhow::anyhow!("kind:9000 requires persist_put_user")), 9001 => handle_remove_user(tenant, event, state).await, // kind:9002 belongs to the atomic persistence path in ingest. 9002 => Err(anyhow::anyhow!("kind:9002 requires atomic persistence")), @@ -597,7 +597,7 @@ pub async fn validate_admin_event( match kind { // Ingest validates kind:9000 with `validate_put_user`, which keeps - // each person's result for `apply_put_user`. + // each person's result for `persist_put_user`. 9000 => Err(anyhow::anyhow!("kind:9000 requires validate_put_user")), 9001 => { // REMOVE_USER: self-remove allowed unless actor is the last owner; removing others requires owner/admin @@ -1760,17 +1760,27 @@ impl PutUserReport { } } -/// Apply a stored kind:9000 and return its OK message. An event naming -/// several people reports each person's result; one naming a single person -/// keeps the empty message a single add has always had. `put_member` decides -/// every admission rule again under the channel's membership lock, so a person -/// validation allowed can still be refused here. -pub(crate) async fn apply_put_user( +/// Store a validated kind:9000 and apply it, returning the stored event, +/// whether it was new, and its OK message. +/// +/// One transaction stores the event and writes every allowed person. If it +/// fails, nothing is stored and the client is told, so it can resend the same +/// event. `put_members_with_event` decides every admission rule again under +/// the channel's membership lock, so a person validation allowed can still be +/// refused there; when everyone is refused, nothing is stored and the error is +/// a [`buzz_db::DbError::AccessDenied`] with the same text validation gives. +/// +/// An event naming several people reports each person's result; one naming a +/// single person keeps the empty message a single add has always had. Live +/// effects (cache invalidation, notices, discovery events) run only after the +/// commit. +pub(crate) async fn persist_put_user( tenant: &TenantContext, event: &Event, state: &Arc, plan: PutUserPlan, -) -> String { + thread_meta: Option>, +) -> anyhow::Result<(buzz_core::StoredEvent, bool, String)> { let channel_id = plan.channel_id; let batch = plan.targets.len() > 1; let actor_bytes = event.pubkey.to_bytes().to_vec(); @@ -1790,21 +1800,30 @@ pub(crate) async fn apply_put_user( requested: target.role, }) .collect(); - let results = match state + let (stored, results) = match state .db - .put_members(tenant.community(), channel_id, &puts, Some(&actor_bytes)) - .await + .put_members_with_event(tenant.community(), channel_id, event, &puts, thread_meta) + .await? { - Ok(results) => results, - // Nothing was written. - Err(e) => { - error!(channel = %channel_id, error = %e, "NIP-29 PUT_USER write failed"); - for target in &allowed { - report.reject(&target.pubkey, "error: membership write failed".into()); + buzz_db::channel_members::MemberPutEventWrite::Applied { stored, results } => { + (stored, results) + } + buzz_db::channel_members::MemberPutEventWrite::Duplicate(stored) => { + return Ok((stored, false, String::new())); + } + buzz_db::channel_members::MemberPutEventWrite::Refused(errors) => { + for (target, error) in allowed.iter().zip(errors) { + report.reject(&target.pubkey, put_refusal_reason(error)); } - Vec::new() + let message = match report.rejected.as_slice() { + // One person keeps the single-add error text. + [(reason, _)] if !batch => reason.clone(), + _ => report.message(), + }; + return Err(buzz_db::DbError::AccessDenied(message).into()); } }; + // Every person's notification is tagged with the channel's type: read it // once. No channel row means no type to tag, so no notifications. let channel_type = if results.iter().any(Result::is_ok) { @@ -1830,13 +1849,8 @@ pub(crate) async fn apply_put_user( for (target, result) in allowed.iter().zip(results) { let put = match result { Ok(put) => put, - Err(buzz_db::DbError::AccessDenied(reason)) => { - report.reject(&target.pubkey, reason); - continue; - } - Err(e) => { - error!(channel = %channel_id, error = %e, "NIP-29 PUT_USER write failed"); - report.reject(&target.pubkey, "error: membership write failed".into()); + Err(error) => { + report.reject(&target.pubkey, put_refusal_reason(error)); continue; } }; @@ -1901,10 +1915,19 @@ pub(crate) async fn apply_put_user( rejected = report.rejected.iter().map(|(_, pubkeys)| pubkeys.len()).sum::(), "NIP-29 PUT_USER processed" ); - if batch { + let message = if batch { report.message() } else { String::new() + }; + Ok((stored, true, message)) +} + +/// The reported reason a person was refused under the membership lock. +fn put_refusal_reason(error: buzz_db::DbError) -> String { + match error { + buzz_db::DbError::AccessDenied(reason) => reason, + other => other.to_string(), } }