diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index f7f0888e96f..1ce22b8a3d1 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,28 @@ 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..b45075e4be5 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -2178,6 +2178,106 @@ 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]); +} + +/// 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) diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index b868ed8b612..2cee788086e 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,53 @@ 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 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", + 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-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] 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;