From 1bc773c420218ea9add3588d9d693d97f6490dae Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Fri, 6 Nov 2026 17:03:18 -0500 Subject: [PATCH 1/5] perf(relay): index default channels and read them from the replica at AUTH On a community that does not require relay membership, every admitted human AUTH lists the default channels the person has never joined. That read walked every open channel in the community on the writer pool to find the few with auto_join set. - Migration 0085 adds a partial index on channels (community_id) holding only effective default channels (flagged, open, unarchived, not deleted stream channels), built inline under 1s lock / 5s statement budgets as 0049 does, with an exact-definition check so a prebuilt CONCURRENTLY index verifies. - Db::list_never_joined_auto_join_channels_routed reads the bounded replica, falling back to the writer as pending_default_channel_join_routed does. put_first_member decides again under the membership lock, so a stale answer can only skip a channel joined since or defer one flagged since. No behavior change. Signed-off-by: Luke Tornquist Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 --- crates/buzz-db/src/runtime/migration.rs | 33 +++++++--- crates/buzz-db/src/runtime/tests.rs | 56 +++++++++++++++++ crates/buzz-db/src/store/channel.rs | 61 +++++++++++++++++-- .../buzz-relay/src/handlers/side_effects.rs | 2 +- migrations/0087_channels_auto_join_index.sql | 45 ++++++++++++++ schema/tables/public/channels.sql | 7 +++ 6 files changed, 188 insertions(+), 16 deletions(-) create mode 100644 migrations/0087_channels_auto_join_index.sql diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index f7f0888e96f..cc1c22491a9 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -902,7 +902,7 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 86); + assert_eq!(migrations.len(), 87); assert!(migrations[83] .sql .as_str() @@ -915,6 +915,15 @@ mod postgres_tests { .sql .as_str() .contains("CREATE TABLE media_uploaders")); + // Default-channel index (0087): bounded inline build, plain index in + // the desired schema. + let auto_join_index = strip_sql_comments(migrations[86].sql.as_str()); + assert!(auto_join_index.contains("SET LOCAL lock_timeout = '1s';")); + assert!(auto_join_index.contains("SET LOCAL statement_timeout = '5s';")); + assert!(auto_join_index.contains("CREATE INDEX idx_channels_auto_join")); + assert!(!auto_join_index.contains("CONCURRENTLY")); + assert!(crate::test_support::desired_state_schema_sql() + .contains("CREATE INDEX idx_channels_auto_join ON channels (community_id)")); for (index, migration) in migrations.iter().enumerate() { assert_eq!(migration.version, (index + 1) as i64); } @@ -3090,20 +3099,24 @@ mod postgres_tests { "column parity mismatch for channels: schema.sql desired state has drifted \ from the migrations; update schema/tables/public/channels.sql to match" ); - // The listed-name index (0083) is compared on its own for that reason. - let listed_name = |shapes: Vec<(String, String, String)>| { + // The listed-name (0083) and default-channel (0087) indexes are + // compared on their own for that reason. + let checked_channel_indexes = |shapes: Vec<(String, String, String)>| { shapes .into_iter() - .filter(|(name, _, _)| name == "idx_channels_listed_name") + .filter(|(name, _, _)| { + name == "idx_channels_listed_name" || name == "idx_channels_auto_join" + }) .collect::>() }; - let desired_listed_name = listed_name(index_shapes(&desired, "channels").await); - assert_eq!(desired_listed_name.len(), 1, "desired state has the index"); + let desired_channel_indexes = + checked_channel_indexes(index_shapes(&desired, "channels").await); + assert_eq!(desired_channel_indexes.len(), 2, "desired state has both indexes"); assert_eq!( - desired_listed_name, - listed_name(index_shapes(&migrated, "channels").await), - "idx_channels_listed_name parity mismatch: update \ - schema/tables/public/channels.sql to match migration 0083" + desired_channel_indexes, + checked_channel_indexes(index_shapes(&migrated, "channels").await), + "channels index parity mismatch: update schema/tables/public/channels.sql \ + to match migrations 0083 and 0087" ); assert_eq!( events_key_defs(&desired).await, diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index bab213081b2..43c9e182d94 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -2178,6 +2178,62 @@ async fn pending_default_channel_join_routed_reads_the_bounded_replica() { assert_eq!(token(pending.expect("entry too old")), writer_token); } +/// The per-AUTH never-joined default-channel read is bounded-routed like the +/// pending-join read. A different flagged channel on each pool proves which +/// pool answered. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn never_joined_auto_join_channels_routed_reads_the_bounded_replica() { + let (writer, _writer_db) = ScratchDb::migrated(&admin_url().await, "autojoin_w").await; + let (replica, _replica_db) = ScratchDb::migrated(&admin_url().await, "autojoin_r").await; + + let community = Uuid::new_v4(); + let person = [0xddu8; 32]; + let (writer_channel, replica_channel) = (Uuid::new_v4(), Uuid::new_v4()); + for (pool, channel) in [(&writer, writer_channel), (&replica, replica_channel)] { + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community) + .bind(format!("autojoin-routing-{}.example", community.simple())) + .execute(pool) + .await + .expect("insert community"); + sqlx::query( + "INSERT INTO channels \ + (id, community_id, name, channel_type, visibility, created_by, auto_join) \ + VALUES ($1, $2, $3, 'stream', 'open', $4, true)", + ) + .bind(channel) + .bind(community) + .bind(format!("autojoin-routing-{channel}")) + .bind([0xeeu8; 32].as_slice()) + .execute(pool) + .await + .expect("seed flagged channel"); + } + let cid = CommunityId::from_uuid(community); + + let mut db = Db::from_pools(writer.clone(), replica.clone()); + db.fence().force_open_for_tests(chrono::Utc::now()); + + // Budget unset ⇒ writer. + let ids = db.list_never_joined_auto_join_channels_routed(cid, &person).await; + assert_eq!(ids.expect("gate off"), vec![writer_channel]); + + // Budget set + fresh entry ⇒ replica. + db.set_replica_read_max_age_for_tests(Some(std::time::Duration::from_secs(5))); + let ids = db.list_never_joined_auto_join_channels_routed(cid, &person).await; + assert_eq!(ids.expect("gate on"), vec![replica_channel]); + + // Entry older than the budget ⇒ fail closed to the writer. + db.fence().close(); + db.fence().force_open_for_tests_at( + chrono::Utc::now(), + std::time::Instant::now() - std::time::Duration::from_secs(10), + ); + let ids = db.list_never_joined_auto_join_channels_routed(cid, &person).await; + assert_eq!(ids.expect("entry too old"), vec![writer_channel]); +} + /// Community separation across every routed seam, verified on /// REPLICA-SERVED reads. /// diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index b868ed8b612..9ea4c5c62e5 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -721,6 +721,16 @@ pub async fn list_never_joined_auto_join_channels( crate::observability::WriterOperation::Authorization, ) .await?; + list_never_joined_auto_join_channels_on(&mut connection, community_id, pubkey).await +} + +/// [`list_never_joined_auto_join_channels`] on a caller-held connection. +/// `idx_channels_auto_join` (migration 0087) serves the outer scan. +async fn list_never_joined_auto_join_channels_on( + connection: &mut sqlx::PgConnection, + community_id: CommunityId, + pubkey: &[u8], +) -> Result> { let ids = sqlx::query_scalar( r#" SELECT c.id @@ -737,7 +747,7 @@ pub async fn list_never_joined_auto_join_channels( ) .bind(community_id.as_uuid()) .bind(pubkey) - .fetch_all(&mut *connection) + .fetch_all(connection) .await?; Ok(ids) } @@ -2141,14 +2151,55 @@ impl Db { /// Lists the effective `auto_join` channels `pubkey` has never been a /// member of, active or removed (see - /// [`list_never_joined_auto_join_channels`]). - #[datastore_span(name = "list_never_joined_auto_join_channels", system = "postgresql")] - pub async fn list_never_joined_auto_join_channels( + /// [`list_never_joined_auto_join_channels`]), on the bounded replica arm. + /// + /// This runs on every admitted human AUTH to a community that does not + /// require relay membership, so it stays off the writer. A stale answer + /// is safe: [`Db::put_first_member`] decides again under the membership + /// lock, so a channel joined since is skipped, and a channel flagged since + /// is joined on the next admission. Falls back to the writer as + /// [`Db::pending_default_channel_join_routed`] does. + #[datastore_span( + name = "list_never_joined_auto_join_channels_routed", + system = "postgresql" + )] + pub async fn list_never_joined_auto_join_channels_routed( &self, community_id: CommunityId, pubkey: &[u8], ) -> Result> { - list_never_joined_auto_join_channels(&self.pool, community_id, pubkey).await + let path = "list_never_joined_auto_join_channels"; + match self + .route_read( + path, + crate::RoutePredicate::Bounded, + crate::observability::ReaderOperation::Authorization, + ) + .await? + { + crate::RouteDecision::Replica(mut tx, _entry, reason) => { + match list_never_joined_auto_join_channels_on(&mut tx, community_id, pubkey).await + { + Ok(ids) => { + Self::record_route(path, "replica", reason); + Ok(ids) + } + Err(e) if e.is_statement_cancelled() => { + Self::record_route(path, "replica", "statement_cancelled"); + Err(e) + } + Err(e) => { + tracing::warn!(path, "replica read failed; re-running on writer: {e}"); + Self::record_route(path, "writer", "replica_error"); + list_never_joined_auto_join_channels(&self.pool, community_id, pubkey) + .await + } + } + } + crate::RouteDecision::Writer => { + list_never_joined_auto_join_channels(&self.pool, community_id, pubkey).await + } + } } /// Lists channels during startup reconciliation. diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 0bcc267a3cb..d76cf8dfa8e 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -3195,7 +3195,7 @@ async fn join_never_joined_channels( let member = hex::encode(pubkey); let channels = match state .db - .list_never_joined_auto_join_channels(tenant.community(), pubkey) + .list_never_joined_auto_join_channels_routed(tenant.community(), pubkey) .await { Ok(channels) => channels, diff --git a/migrations/0087_channels_auto_join_index.sql b/migrations/0087_channels_auto_join_index.sql new file mode 100644 index 00000000000..d6dda1059ef --- /dev/null +++ b/migrations/0087_channels_auto_join_index.sql @@ -0,0 +1,45 @@ +-- Effective default channels per community: the open, unarchived stream +-- channels with `auto_join` set. On a community that does not require relay +-- membership, every human AUTH lists the ones the person has never joined +-- (`list_never_joined_auto_join_channels` in crates/buzz-db/src/store/channel.rs); +-- without this index that read walks every open channel in the community to +-- find the few that are flagged. The partial index holds only flagged rows, so +-- it stays tiny and the read becomes one index probe. Additive and expand-only: +-- the running relay ignores it. +-- +-- Built inline under the budgets below, as 0049 does (docs/database.md Rule 6). +-- A deployment that applies migrations out of band and cannot build it within +-- them prebuilds the exact index first, then applies this migration, which +-- only verifies it: +-- CREATE INDEX CONCURRENTLY idx_channels_auto_join ON public.channels (community_id) +-- WHERE auto_join AND channel_type = 'stream' AND visibility = 'open' +-- AND archived_at IS NULL AND deleted_at IS NULL; +SET LOCAL lock_timeout = '1s'; +SET LOCAL statement_timeout = '5s'; +-- pg_get_indexdef qualifies the enum casts below (`public.channel_type`) when +-- `public` is not on the search_path; pin it so the exact compare holds. +SET LOCAL search_path = public, pg_catalog; +-- IF NOT EXISTS still requests a writer-conflicting ShareLock before checking +-- whether the index exists. Bypass CREATE entirely for a prebuilt index, then +-- validate it below; a failed concurrent build must never count as success. +DO $$ +BEGIN + IF to_regclass('public.idx_channels_auto_join') IS NULL THEN + CREATE INDEX idx_channels_auto_join + ON public.channels (community_id) + WHERE auto_join AND channel_type = 'stream' AND visibility = 'open' + AND archived_at IS NULL AND deleted_at IS NULL; + END IF; + + IF NOT EXISTS ( + SELECT 1 FROM pg_index i + JOIN pg_class c ON c.oid = i.indexrelid + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = 'public' AND c.relname = 'idx_channels_auto_join' + AND i.indisvalid AND i.indisready AND i.indislive + AND pg_get_indexdef(i.indexrelid) = + 'CREATE INDEX idx_channels_auto_join ON public.channels USING btree (community_id) WHERE (auto_join AND (channel_type = ''stream''::channel_type) AND (visibility = ''open''::channel_visibility) AND (archived_at IS NULL) AND (deleted_at IS NULL))' + ) THEN + RAISE EXCEPTION 'Migration 0087 stopped: idx_channels_auto_join is invalid or has a different definition. Drop it, build it again (see the comment at the top of this migration), then migrate again.'; + END IF; +END $$; diff --git a/schema/tables/public/channels.sql b/schema/tables/public/channels.sql index 9162b502a64..68d8453c69a 100644 --- a/schema/tables/public/channels.sql +++ b/schema/tables/public/channels.sql @@ -70,3 +70,10 @@ CREATE INDEX idx_channels_id_live ON channels (id) INCLUDE (community_id) CREATE UNIQUE INDEX idx_channels_listed_name ON channels (community_id, lower(name)) WHERE visibility = 'open' AND NOT unlisted AND channel_type IN ('stream', 'forum') AND deleted_at IS NULL; +-- Effective default channels (open, unarchived, flagged streams), read on every +-- AUTH to a community that does not require relay membership. Partial, so it +-- holds only flagged rows. Created by migration 0087; must match +-- list_never_joined_auto_join_channels in crates/buzz-db/src/store/channel.rs. +CREATE INDEX idx_channels_auto_join ON channels (community_id) + WHERE auto_join AND channel_type = 'stream' AND visibility = 'open' + AND archived_at IS NULL AND deleted_at IS NULL; From 54cb6c3bd58e31d67846c0ac6647f43dbbdae275 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Fri, 6 Nov 2026 17:07:46 -0500 Subject: [PATCH 2/5] style: cargo fmt Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- crates/buzz-db/src/runtime/migration.rs | 6 +++++- crates/buzz-db/src/runtime/tests.rs | 12 +++++++++--- 2 files changed, 14 insertions(+), 4 deletions(-) diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index cc1c22491a9..1ce22b8a3d1 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -3111,7 +3111,11 @@ mod postgres_tests { }; let desired_channel_indexes = checked_channel_indexes(index_shapes(&desired, "channels").await); - assert_eq!(desired_channel_indexes.len(), 2, "desired state has both indexes"); + assert_eq!( + desired_channel_indexes.len(), + 2, + "desired state has both indexes" + ); assert_eq!( desired_channel_indexes, checked_channel_indexes(index_shapes(&migrated, "channels").await), diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index 43c9e182d94..298a80ecb22 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -2216,12 +2216,16 @@ async fn never_joined_auto_join_channels_routed_reads_the_bounded_replica() { db.fence().force_open_for_tests(chrono::Utc::now()); // Budget unset ⇒ writer. - let ids = db.list_never_joined_auto_join_channels_routed(cid, &person).await; + let ids = db + .list_never_joined_auto_join_channels_routed(cid, &person) + .await; assert_eq!(ids.expect("gate off"), vec![writer_channel]); // Budget set + fresh entry ⇒ replica. db.set_replica_read_max_age_for_tests(Some(std::time::Duration::from_secs(5))); - let ids = db.list_never_joined_auto_join_channels_routed(cid, &person).await; + let ids = db + .list_never_joined_auto_join_channels_routed(cid, &person) + .await; assert_eq!(ids.expect("gate on"), vec![replica_channel]); // Entry older than the budget ⇒ fail closed to the writer. @@ -2230,7 +2234,9 @@ async fn never_joined_auto_join_channels_routed_reads_the_bounded_replica() { chrono::Utc::now(), std::time::Instant::now() - std::time::Duration::from_secs(10), ); - let ids = db.list_never_joined_auto_join_channels_routed(cid, &person).await; + let ids = db + .list_never_joined_auto_join_channels_routed(cid, &person) + .await; assert_eq!(ids.expect("entry too old"), vec![writer_channel]); } From e4be50791574adb604750ad4a5d8f74b8e5df504 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Fri, 6 Nov 2026 17:08:27 -0500 Subject: [PATCH 3/5] style: cargo fmt channel.rs Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- crates/buzz-db/src/store/channel.rs | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index 9ea4c5c62e5..5ce644c0446 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -2178,8 +2178,7 @@ impl Db { .await? { crate::RouteDecision::Replica(mut tx, _entry, reason) => { - match list_never_joined_auto_join_channels_on(&mut tx, community_id, pubkey).await - { + match list_never_joined_auto_join_channels_on(&mut tx, community_id, pubkey).await { Ok(ids) => { Self::record_route(path, "replica", reason); Ok(ids) @@ -2191,8 +2190,7 @@ impl Db { Err(e) => { tracing::warn!(path, "replica read failed; re-running on writer: {e}"); Self::record_route(path, "writer", "replica_error"); - list_never_joined_auto_join_channels(&self.pool, community_id, pubkey) - .await + list_never_joined_auto_join_channels(&self.pool, community_id, pubkey).await } } } From 8b74366555480b3686875b268f21544dcfbbce66 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Fri, 6 Nov 2026 17:41:56 -0500 Subject: [PATCH 4/5] test(db): scope 0083's writer-lock check to 0083; pin 0087 over a prebuild The 0083 prebuild test held a channels writer lock while running every migration, so 0087's inline index build hit its lock timeout. Run it only through 0083, and add the matching check that 0087 over a concurrent prebuild neither waits for writers nor replaces the index. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- crates/buzz-db/src/runtime/tests.rs | 38 +++++++++++++++++++ .../listed_channel_name_postgres_tests.rs | 3 +- 2 files changed, 40 insertions(+), 1 deletion(-) diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index 298a80ecb22..b45075e4be5 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -2240,6 +2240,44 @@ async fn never_joined_auto_join_channels_routed_reads_the_bounded_replica() { assert_eq!(ids.expect("entry too old"), vec![writer_channel]); } +/// Migration 0087 over an operator's concurrent prebuild of +/// `idx_channels_auto_join` only checks the catalog. It neither waits for +/// channel writers nor replaces the prebuild. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn auto_join_index_migration_accepts_a_prebuild_without_waiting_for_writers() { + let (pool, _scratch) = + ScratchDb::migrated_through(&admin_url().await, "auto_join_prebuild", 86).await; + sqlx::query( + "CREATE INDEX CONCURRENTLY idx_channels_auto_join ON public.channels (community_id) \ + WHERE auto_join AND channel_type = 'stream' AND visibility = 'open' \ + AND archived_at IS NULL AND deleted_at IS NULL", + ) + .execute(&pool) + .await + .expect("prebuild idx_channels_auto_join"); + let oid = || async { + sqlx::query_scalar::<_, Option>( + "SELECT to_regclass('public.idx_channels_auto_join')::oid::bigint", + ) + .fetch_one(&pool) + .await + .expect("look up index") + }; + let prebuilt = oid().await; + + let mut writer = pool.begin().await.unwrap(); + sqlx::query("LOCK TABLE channels IN ROW EXCLUSIVE MODE") + .execute(&mut *writer) + .await + .unwrap(); + crate::migration::run_migrations_through(&pool, 87) + .await + .expect("accept the prebuild without waiting for channel writers"); + writer.rollback().await.unwrap(); + assert_eq!(oid().await, prebuilt, "the prebuild is not replaced"); +} + /// Community separation across every routed seam, verified on /// REPLICA-SERVED reads. /// diff --git a/crates/buzz-db/src/runtime/tests/listed_channel_name_postgres_tests.rs b/crates/buzz-db/src/runtime/tests/listed_channel_name_postgres_tests.rs index 096eee36f5f..aefb535a844 100644 --- a/crates/buzz-db/src/runtime/tests/listed_channel_name_postgres_tests.rs +++ b/crates/buzz-db/src/runtime/tests/listed_channel_name_postgres_tests.rs @@ -639,7 +639,8 @@ async fn migration_schema_listed_name_index_validates_a_prebuild() { let oid = index_oid(&pool).await; // A channel writer holds its lock through the migration. Over a prebuild, // 0083 must not request the ShareLock a CREATE INDEX takes, even with - // IF NOT EXISTS: that waits behind writers and queues new ones. + // IF NOT EXISTS: that waits behind writers and queues new ones. Stop at + // 0083: later migrations that build their own index inline need the lock. let mut writer = pool.begin().await.unwrap(); sqlx::query("LOCK TABLE channels IN ROW EXCLUSIVE MODE") .execute(&mut *writer) From 99c07da83aee4d72cf97199b2363923e93f29ca4 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Fri, 6 Nov 2026 19:09:28 -0500 Subject: [PATCH 5/5] fix(db): skip a default-channel join once the channel is unflagged The AUTH auto-join list can now come from the bounded replica, so a channel unflagged within the replica budget could still be listed. put_first_member already decides membership history under the channel lock; it now also skips a channel that is no longer auto_join, so a stale candidate writes nothing. Both callers are default-channel joins. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- crates/buzz-db/src/store/channel.rs | 4 +- crates/buzz-db/src/store/channel_members.rs | 46 ++++++++++++++++++++- 2 files changed, 46 insertions(+), 4 deletions(-) diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index 5ce644c0446..2cee788086e 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -2156,8 +2156,8 @@ impl Db { /// This runs on every admitted human AUTH to a community that does not /// require relay membership, so it stays off the writer. A stale answer /// is safe: [`Db::put_first_member`] decides again under the membership - /// lock, so a channel joined since is skipped, and a channel flagged since - /// is joined on the next admission. Falls back to the writer as + /// lock, so a channel joined or unflagged since is skipped, and a channel + /// flagged since is joined on the next admission. Falls back to the writer as /// [`Db::pending_default_channel_join_routed`] does. #[datastore_span( name = "list_never_joined_auto_join_channels_routed", diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 3be879aaf00..3c9a8435ed1 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -725,7 +725,9 @@ pub async fn put_members( /// Self-join `pubkey` to `channel_id` as [`put_member`] with no actor and no /// requested role (so the channel's `join_role` applies), but only if /// `pubkey` has never had a membership row in the channel, active or removed. -/// Returns `None`, writing nothing, when such a row exists. +/// Returns `None`, writing nothing, when such a row exists or the channel is +/// no longer a default (`auto_join`) channel. Callers may pass candidates +/// from a stale (replica) read: both are decided here, on the writer. /// /// The row is checked under the [`put_member`] locks, so a leave, a join or /// another first join of the same person commits wholly before or after this @@ -775,6 +777,18 @@ async fn put_first_member_with_pending( None }; lock_channel_for_member_put(&mut tx, channel_id).await?; + // Only default channels are joined here. A missing channel falls through + // to the put_members_on error. + let auto_join: Option = + sqlx::query_scalar("SELECT auto_join FROM channels WHERE community_id = $1 AND id = $2") + .bind(community_id.as_uuid()) + .bind(channel_id) + .fetch_optional(tx.conn()) + .await?; + if auto_join == Some(false) { + tx.rollback().await?; + return Ok(None); + } let history: Option<(Option>,)> = sqlx::query_as( "SELECT removed_at FROM channel_members \ WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3", @@ -2853,7 +2867,8 @@ mod postgres_tests { /// undone (RED); take the locks after the row check → racing joins both /// write (RED); drop the `NOT EXISTS` from /// `list_never_joined_auto_join_channels` → the left channel is still - /// listed (RED). + /// listed (RED); drop the `auto_join` recheck from `put_first_member` → + /// an unflagged channel is joined (RED). #[tokio::test] #[ignore = "requires Postgres"] async fn first_member_put_joins_once_and_never_undoes_a_leave() { @@ -2947,6 +2962,33 @@ mod postgres_tests { .expect("role"), None ); + + // A stale (replica) candidate for a channel unflagged since writes + // nothing. + assert!(put_first(unflagged).await.expect("put unflagged").is_none()); + update_channel( + &pool, + community, + announcements, + ChannelUpdate { + auto_join: Some(false), + ..Default::default() + }, + ) + .await + .expect("unflag channel"); + let newcomer = random_pubkey(); + assert!(put_first_member(&pool, community, announcements, &newcomer) + .await + .expect("put after unflag") + .is_none()); + assert_eq!( + get_member_role(&pool, community, announcements, &newcomer) + .await + .expect("role"), + None, + "an unflagged channel is not joined" + ); } #[tokio::test]