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

fix(relay): create a channel and store its create in one transaction

Browse Source 
A kind:9007 used to commit its channel and members, then store the event
in a second write. A resend landing between the two found the channel but
not the event, so it was refused, and a client that withdraws refused
creates dropped a chat that existed.

The channel, its owner, its members and the event now commit together.
A resend beside an unfinished create waits on its channel insert, then is
confirmed once the create commits, or creates the channel itself if the
first rolled back. The soft-delete compensation for a failed event insert
is gone: the rollback covers it.

A failed join notice for one initial member no longer skips the notices
and notifications of the members after it.

Co-authored-by: ART <ea39c1c8acf57180161ef4ba4ed68abf8589f5291956b909a5aa991fc66664ca@buzz.block.builderlab.xyz>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Bradley Axen <baxen@squareup.com>
This commit is contained in:
Bradley Axen
2026-10-18 10:37:08 -07:00
parent c2b22afc88
commit 10adcf6dd5
3 changed files with 433 additions and 155 deletions
Download Patch File Download Diff File Expand all files Collapse all files

427
crates/buzz-db/src/store/channel.rs
View File

@@ -10,7 +10,7 @@ use uuid::Uuid;
use crate::error::{DbError, Result};
use crate::Db;
use buzz_core::CommunityId;
use buzz_core::{CommunityId, StoredEvent};
use buzz_datastore_tracing::datastore_span;
// Re-export the canonical enum definitions from buzz-core.
@@ -200,10 +200,133 @@ pub async fn create_channel_with_id(
ttl_seconds: Option<i32>,
labels: &[String],
) -> Result<(ChannelRecord, bool)> {
create_channel_with_members(
let mut tx = crate::begin_community_write_transaction(
pool,
community_id,
crate::observability::WriterOperation::EventWrite,
)
.await?;
let was_created = insert_channel_rows(
&mut tx,
channel_id,
NewChannel {
name,
channel_type,
visibility,
description,
created_by,
ttl_seconds,
labels,
members: &[],
},
)
.await?;
let row = sqlx::query(
r#"
SELECT id, name, channel_type::text AS channel_type, visibility::text AS visibility,
description, canvas,
created_by, created_at, updated_at, archived_at, deleted_at,
nip29_group_id, topic_required, max_members,
topic, topic_set_by, topic_set_at,
purpose, purpose_set_by, purpose_set_at,
ttl_seconds, ttl_deadline, labels
FROM channels WHERE community_id = $1 AND id = $2
"#,
)
.bind(community_id.as_uuid())
.bind(channel_id)
.fetch_one(tx.conn())
.await?;
let record = row_to_channel_record(row)?;
tx.commit().await?;
Ok((record, was_created))
}
/// A channel to create: the fields a kind:9007 sets.
pub struct NewChannel<'a> {
/// Display name, canonicalized before storage.
pub name: &'a str,
/// Stream, forum, DM, …
pub channel_type: ChannelType,
/// Open or private.
pub visibility: ChannelVisibility,
/// The `about` text.
pub description: Option<&'a str>,
/// The creator's 32-byte pubkey; becomes the owner.
pub created_by: &'a [u8],
/// Ephemeral channel lifetime.
pub ttl_seconds: Option<i32>,
/// Labels validated by `buzz_core::channel::parse_channel_labels`.
pub labels: &'a [String],
/// 32-byte pubkeys added as ordinary members invited by the creator. The
/// caller has authorized each addition.
pub members: &'a [Vec<u8>],
}
/// What [`Db::create_channel_with_event`] did.
#[derive(Debug)]
pub enum ChannelCreate {
/// The channel, its members and the create event committed together.
Created(Box<StoredEvent>),
/// The UUID was already taken and nothing was written. `resent` is true
/// when the stored create that took it is this same event.
Exists {
/// This exact event is the channel's stored create.
resent: bool,
},
}
/// See [`Db::create_channel_with_event`].
pub(crate) async fn create_channel_with_event(
pool: &PgPool,
push_enqueue: Option<crate::runtime::push_enqueue::PushEnqueue>,
community_id: CommunityId,
channel_id: Uuid,
channel: NewChannel<'_>,
event: &nostr::Event,
) -> Result<ChannelCreate> {
let mut tx = crate::begin_community_event_write_transaction(
pool,
community_id,
crate::observability::WriterOperation::EventWrite,
)
.await?;
tx.set_push_enqueue(push_enqueue);
if !insert_channel_rows(&mut tx, channel_id, channel).await? {
let resent = sqlx::query(
"SELECT 1 FROM events WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL LIMIT 1",
)
.bind(community_id.as_uuid())
.bind(event.id.as_bytes().as_slice())
.fetch_optional(tx.conn())
.await?
.is_some();
return Ok(ChannelCreate::Exists { resent });
}
// 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), None)
.await?;
if inserted {
crate::insert_mentions_in_transaction(&mut tx, event, Some(channel_id)).await?;
}
tx.commit().await?;
Ok(ChannelCreate::Created(Box::new(stored)))
}
/// Inserts the channel row, its owner and its initial members on `tx`.
/// Returns `false`, having written nothing, when `channel_id` already exists.
async fn insert_channel_rows(
tx: &mut crate::AdmittedTx,
channel_id: Uuid,
channel: NewChannel<'_>,
) -> Result<bool> {
let NewChannel {
name,
channel_type,
visibility,
@@ -211,29 +334,9 @@ pub async fn create_channel_with_id(
created_by,
ttl_seconds,
labels,
&[],
)
.await
}
/// [`create_channel_with_id`], also adding `members` (32-byte pubkeys) as
/// ordinary members invited by the creator, in the transaction that creates the
/// channel. The caller has authorized each addition. Nothing is added when the
/// channel already exists.
#[allow(clippy::too_many_arguments)]
pub async fn create_channel_with_members(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
name: &str,
channel_type: ChannelType,
visibility: ChannelVisibility,
description: Option<&str>,
created_by: &[u8],
ttl_seconds: Option<i32>,
labels: &[String],
members: &[Vec<u8>],
) -> Result<(ChannelRecord, bool)> {
members,
} = channel;
let community_id = tx.community();
if created_by.len() != 32 {
return Err(DbError::InvalidData(format!(
"pubkey must be 32 bytes, got {}",
@@ -252,13 +355,6 @@ pub async fn create_channel_with_members(
return Err(DbError::InvalidData("channel name is required".into()));
}
let mut tx = crate::begin_community_write_transaction(
pool,
community_id,
crate::observability::WriterOperation::EventWrite,
)
.await?;
let rows_affected = sqlx::query(
r#"
INSERT INTO channels (id, community_id, name, channel_type, visibility, description, created_by, ttl_seconds, ttl_deadline, labels)
@@ -327,26 +423,7 @@ pub async fn create_channel_with_members(
}
}
let row = sqlx::query(
r#"
SELECT id, name, channel_type::text AS channel_type, visibility::text AS visibility,
description, canvas,
created_by, created_at, updated_at, archived_at, deleted_at,
nip29_group_id, topic_required, max_members,
topic, topic_set_by, topic_set_at,
purpose, purpose_set_by, purpose_set_at,
ttl_seconds, ttl_deadline, labels
FROM channels WHERE community_id = $1 AND id = $2
"#,
)
.bind(community_id.as_uuid())
.bind(channel_id)
.fetch_one(tx.conn())
.await?;
let record = row_to_channel_record(row)?;
tx.commit().await?;
Ok((record, was_created))
Ok(was_created)
}
/// Fetches a channel record by `(community_id, id)`. Returns `ChannelNotFound` if missing or deleted.
@@ -1014,38 +1091,28 @@ impl Db {
.await
}
/// Creates a channel with a client-supplied UUID and an initial label set.
/// Creates `channel_id` from its kind:9007 `event`: the channel, its
/// owner, its initial members and the event itself commit in one
/// transaction, so a channel row always has its create event beside it.
///
/// The same as [`Db::create_channel_with_id`], plus `labels`, which the
/// caller has validated with `buzz_core::channel::parse_channel_labels`,
/// and initial `members` (see [`create_channel_with_members`]).
#[allow(clippy::too_many_arguments)]
#[datastore_span(name = "create_labeled_channel_with_id", system = "postgresql")]
pub async fn create_labeled_channel_with_id(
/// A concurrent copy of the same create waits on the first one's channel
/// insert, then sees either the committed event (`Exists { resent: true }`)
/// or, if the first rolled back, an untaken UUID.
#[datastore_span(name = "create_channel_with_event", system = "postgresql")]
pub async fn create_channel_with_event(
&self,
community_id: CommunityId,
channel_id: Uuid,
name: &str,
channel_type: ChannelType,
visibility: ChannelVisibility,
description: Option<&str>,
created_by: &[u8],
ttl_seconds: Option<i32>,
labels: &[String],
members: &[Vec<u8>],
) -> Result<(ChannelRecord, bool)> {
create_channel_with_members(
channel: NewChannel<'_>,
event: &nostr::Event,
) -> Result<ChannelCreate> {
create_channel_with_event(
&self.pool,
self.push_enqueue.clone(),
community_id,
channel_id,
name,
channel_type,
visibility,
description,
created_by,
ttl_seconds,
labels,
members,
channel,
event,
)
.await
}
@@ -1488,4 +1555,210 @@ mod postgres_tests {
"reaper should carry the archived row's community id and host"
);
}
fn create_event(creator: &Keys, channel_id: Uuid) -> nostr::Event {
nostr::EventBuilder::new(nostr::Kind::Custom(9007), "")
.tags([
nostr::Tag::parse(["h", &channel_id.to_string()]).expect("h tag"),
nostr::Tag::parse(["name", "start"]).expect("name tag"),
])
.sign_with_keys(creator)
.expect("sign 9007")
}
fn new_channel<'a>(created_by: &'a [u8], members: &'a [Vec<u8>]) -> NewChannel<'a> {
NewChannel {
name: "start",
channel_type: ChannelType::Stream,
visibility: ChannelVisibility::Private,
description: None,
created_by,
ttl_seconds: None,
labels: &[],
members,
}
}
/// Holds a create open on its own transaction (channel row and event both
/// written, not committed) while the same create runs again beside it.
async fn resend_beside_open_create(commit_first: bool) -> ChannelCreate {
let pool = setup_pool().await;
let community = CommunityId::from_uuid(make_test_community(&pool).await);
let creator = Keys::generate();
let creator_bytes = creator.public_key().to_bytes().to_vec();
let members = vec![random_pubkey()];
let channel_id = Uuid::new_v4();
let event = create_event(&creator, channel_id);
let mut first = crate::begin_community_write_transaction(
&pool,
community,
crate::observability::WriterOperation::EventWrite,
)
.await
.expect("begin first create");
assert!(insert_channel_rows(
&mut first,
channel_id,
new_channel(&creator_bytes, &members)
)
.await
.expect("first create's rows"));
crate::event::insert_event_with_thread_metadata_tx(
&mut first,
&event,
Some(channel_id),
None,
)
.await
.expect("first create's event");
let resend = tokio::spawn({
let (pool, event, creator_bytes, members) = (
pool.clone(),
event.clone(),
creator_bytes.clone(),
members.clone(),
);
async move {
create_channel_with_event(
&pool,
None,
community,
channel_id,
new_channel(&creator_bytes, &members),
&event,
)
.await
}
});
tokio::time::sleep(std::time::Duration::from_millis(300)).await;
assert!(
!resend.is_finished(),
"the resend waits for the open create instead of answering early"
);
if commit_first {
first.commit().await.expect("commit first create");
} else {
drop(first);
}
let outcome = resend.await.expect("resend task").expect("resend outcome");
let roster = sqlx::query_scalar::<_, Vec<u8>>(
"SELECT pubkey FROM channel_members WHERE community_id = $1 AND channel_id = $2 \
AND removed_at IS NULL ORDER BY pubkey",
)
.bind(community.as_uuid())
.bind(channel_id)
.fetch_all(&pool)
.await
.expect("roster");
let mut expected = vec![creator_bytes, members[0].clone()];
expected.sort();
assert_eq!(
roster, expected,
"the channel has its owner and member once"
);
outcome
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn a_resend_beside_an_open_create_confirms_it_once_committed() {
let outcome = resend_beside_open_create(true).await;
assert!(
matches!(outcome, ChannelCreate::Exists { resent: true }),
"{outcome:?}"
);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn a_resend_beside_a_failed_create_creates_the_channel() {
let outcome = resend_beside_open_create(false).await;
assert!(matches!(outcome, ChannelCreate::Created(_)), "{outcome:?}");
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn a_create_whose_event_fails_leaves_no_channel() {
let pool = setup_pool().await;
let community = CommunityId::from_uuid(make_test_community(&pool).await);
let creator = Keys::generate();
let creator_bytes = creator.public_key().to_bytes().to_vec();
let channel_id = Uuid::new_v4();
// An ephemeral kind is refused by the event insert, after the rows.
let unstorable = nostr::EventBuilder::new(nostr::Kind::Custom(20_000), "")
.sign_with_keys(&creator)
.expect("sign");
let members = [random_pubkey()];
create_channel_with_event(
&pool,
None,
community,
channel_id,
new_channel(&creator_bytes, &members),
&unstorable,
)
.await
.expect_err("the event insert fails");
assert!(
get_channel(&pool, community, channel_id).await.is_err(),
"the channel rolls back with its event"
);
let retry = create_channel_with_event(
&pool,
None,
community,
channel_id,
new_channel(&creator_bytes, &members),
&create_event(&creator, channel_id),
)
.await
.expect("retry");
assert!(matches!(retry, ChannelCreate::Created(_)), "{retry:?}");
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn another_create_for_a_taken_channel_writes_nothing() {
let pool = setup_pool().await;
let community = CommunityId::from_uuid(make_test_community(&pool).await);
let creator = Keys::generate();
let creator_bytes = creator.public_key().to_bytes().to_vec();
let channel_id = Uuid::new_v4();
let first = create_channel_with_event(
&pool,
None,
community,
channel_id,
new_channel(&creator_bytes, &[]),
&create_event(&creator, channel_id),
)
.await
.expect("first create");
assert!(matches!(first, ChannelCreate::Created(_)), "{first:?}");
let other = Keys::generate();
let other_bytes = other.public_key().to_bytes().to_vec();
let intruder = random_pubkey();
let second = create_channel_with_event(
&pool,
None,
community,
channel_id,
new_channel(&other_bytes, std::slice::from_ref(&intruder)),
&create_event(&other, channel_id),
)
.await
.expect("second create");
assert!(
matches!(second, ChannelCreate::Exists { resent: false }),
"{second:?}"
);
let members = crate::channel_members::get_members(&pool, community, channel_id)
.await
.expect("members");
assert_eq!(members.len(), 1, "only the first creator is a member");
}
}

145
crates/buzz-relay/src/handlers/ingest.rs
View File

@@ -2381,6 +2381,18 @@ fn validate_canvas_future_timestamp(event_ts: i64, now: i64) -> Result<(), &'sta
Ok(())
}
/// A validated kind:9007 whose channel is created when the event is stored.
struct ChannelCreateRequest {
channel_id: Uuid,
name: String,
channel_type: buzz_db::channel::ChannelType,
visibility: buzz_db::channel::ChannelVisibility,
description: Option<String>,
ttl_seconds: Option<i32>,
labels: Vec<String>,
members: Vec<Vec<u8>>,
}
async fn ingest_event_inner(
state: &Arc<AppState>,
tracer: &Arc<dyn buzz_conformance::Tracer>,
@@ -3050,8 +3062,9 @@ async fn ingest_event_inner(
.map_err(|e| IngestError::Rejected(format!("invalid: {e}")))?;
}
// Track pre-created channel UUID for compensation on insert failure.
let mut pre_created_channel: Option<Uuid> = None;
// A kind:9007 with an `h` tag creates its channel when it is stored, in the
// same transaction (see the storage dispatch below).
let mut channel_create: Option<ChannelCreateRequest> = None;
if kind_u32 == KIND_NIP29_CREATE_GROUP {
// Validate name tag is present and non-empty before any DB work.
@@ -3112,66 +3125,24 @@ async fn ingest_event_inner(
.unwrap_or_default();
if let Some(client_uuid) = channel_id {
let name = create_name.unwrap_or_default();
let name = buzz_core::channel::canonical_channel_name(&name);
let description = event.tags.iter().find_map(|t| {
if t.kind().to_string() == "about" {
t.content().map(|s| s.to_string())
} else {
None
}
});
let ttl_seconds = super::resolve_ttl(&event, state.config.ephemeral_ttl_override);
let actor_bytes = event.pubkey.to_bytes().to_vec();
let members = super::side_effects::initial_members(&event)
.map_err(|e| IngestError::Rejected(format!("invalid: {e}")))?;
let (_, was_created) = state
.db
.create_labeled_channel_with_id(
tenant.community(),
client_uuid,
name,
channel_type,
visibility,
description.as_deref(),
&actor_bytes,
ttl_seconds,
&labels,
&members,
)
.await
.map_err(|e| IngestError::Internal(format!("error: {e}")))?;
if !was_created {
// Resending a create whose OK was lost confirms it. Anything
// else for an existing UUID is refused before storage, so it
// never runs the create's side effects again.
let resent = state
.db
.get_event_by_id_for_event_write(tenant.community(), event.id.as_bytes())
.await
.map_err(|e| IngestError::Internal(format!("error: {e}")))?
.is_some();
return Ok(IngestResult {
event_id: event_id_hex,
accepted: resent,
message: if resent {
"duplicate:".into()
channel_create = Some(ChannelCreateRequest {
channel_id: client_uuid,
name: create_name.unwrap_or_default(),
channel_type,
visibility,
description: event.tags.iter().find_map(|t| {
if t.kind().to_string() == "about" {
t.content().map(|s| s.to_string())
} else {
"duplicate: channel already exists".into()
},
});
}
pre_created_channel = Some(client_uuid);
metrics::counter!(
"buzz_channels_created_total",
"community" => tenant.host().to_owned(),
"type" => channel_type.to_string()
)
.increment(1);
None
}
}),
ttl_seconds: super::resolve_ttl(&event, state.config.ephemeral_ttl_override),
labels,
members,
});
}
}
@@ -3422,7 +3393,49 @@ async fn ingest_event_inner(
};
let workflow_deletion = crate::handlers::side_effects::is_workflow_deletion(&event);
let (stored_event, was_inserted) = if workflow_deletion {
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 {
name: &create.name,
channel_type: create.channel_type,
visibility: create.visibility,
description: create.description.as_deref(),
created_by: &creator,
ttl_seconds: create.ttl_seconds,
labels: &create.labels,
members: &create.members,
};
match state
.db
.create_channel_with_event(tenant.community(), create.channel_id, channel, &event)
.await
.map_err(|e| IngestError::Internal(format!("error: {e}")))?
{
buzz_db::channel::ChannelCreate::Created(stored) => {
metrics::counter!(
"buzz_channels_created_total",
"community" => tenant.host().to_owned(),
"type" => create.channel_type.to_string()
)
.increment(1);
(*stored, true)
}
// Resending a create whose OK was lost confirms it. Anything else
// for a taken UUID is refused. Neither writes anything, so the
// create's side effects never run again.
buzz_db::channel::ChannelCreate::Exists { resent } => {
return Ok(IngestResult {
event_id: event_id_hex,
accepted: resent,
message: if resent {
"duplicate:".into()
} else {
"duplicate: channel already exists".into()
},
});
}
}
} else if workflow_deletion {
// A single commit owns public acceptance, domain mutation, and dispatch.
// Failure rolls everything back; identical concurrent requests cannot
// divide insertion and repair ownership between two relay workers.
@@ -3502,18 +3515,6 @@ async fn ingest_event_inner(
{
Ok(result) => result,
Err(e) => {
// Compensate: if we pre-created a channel for kind:9007,
// soft-delete it so no orphaned channel row remains.
if let Some(ch_id) = pre_created_channel {
if let Err(re) = state
.db
.soft_delete_channel(tenant.community(), ch_id)
.await
{
warn!(event_id = %event_id_hex, "channel compensation failed: {re}");
}
state.invalidate_channel_deleted(tenant);
}
return Err(match e {
buzz_db::DbError::AuthEventRejected => {
IngestError::Rejected("invalid: AUTH events cannot be stored".into())

16
crates/buzz-relay/src/handlers/side_effects.rs
View File

@@ -3,7 +3,7 @@
use std::sync::Arc;
use nostr::{Event, EventBuilder, Kind, Tag};
use tracing::{info, warn};
use tracing::{error, info, warn};
use uuid::Uuid;
use buzz_core::kind::{
@@ -2174,8 +2174,8 @@ async fn handle_create_group(
// Creator becomes owner — evict any stale negative membership lookup.
state.invalidate_membership(tenant, channel.id, &actor_bytes);
// Open channels appear in everyone's accessible set; private channels only
// affect the creator (the sole initial member).
// Open channels appear in everyone's accessible set; a private channel only
// affects its initial members, whose entries the invalidations here drop.
if visibility == buzz_db::channel::ChannelVisibility::Open {
state.invalidate_all_accessible_channels(tenant);
}
@@ -2210,7 +2210,8 @@ async fn handle_create_group(
}
// Ingest added these in the create's transaction; announce them as a
// kind:9000 would.
// kind:9000 would. A failed announcement for one member must not cost the
// rest theirs: a resend writes nothing, so nothing would retry them.
let members = if members_added {
initial_members(event)?
} else {
@@ -2218,7 +2219,7 @@ async fn handle_create_group(
};
for member in members {
state.invalidate_membership(tenant, channel.id, &member);
emit_system_message(
if let Err(e) = emit_system_message(
tenant,
state,
channel.id,
@@ -2229,7 +2230,10 @@ async fn handle_create_group(
}),
chrono::Utc::now(),
)
.await?;
.await
{
error!(channel = %channel.id, error = %e, "member_joined system message failed");
}
if let Err(e) = emit_membership_notification(
tenant,
state,