perf(relay): index default channels and read them from the replica at AUTH

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::<Vec<_>>()
         };
-        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,

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<i64>>(
+            "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.
 ///

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)

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<Vec<Uuid>> {
     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<Vec<Uuid>> {
-        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.

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<bool> =
+        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<DateTime<Utc>>,)> = 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]

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,

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 $$;