PostgreSQL search GIN indexes without a pending list

A three-node rehearsal on PostgreSQL saw searches take about 185 ms
with 80 to 260 pages in the full-text indexes' pending lists, 2 to 6 ms
right after gin_clean_pending_list() or VACUUM, then creep back up as
mail came in. The search tables' GIN indexes were created with the
default fastupdate=on: new entries wait in an unindexed pending list
that every search scans in full until VACUUM (or 4 MB of backlog)
merges it, and autovacuum only visits an insert-only table after
thousands of inserts.

The search GIN indexes are now created WITH (fastupdate = off), so an
insert pays its index update at once. The schema step runs at every
startup (create_search_tables, via SearchStore::create_indexes), so
indexes made before this change are switched there: when an index's
reloptions don't already turn fastupdate off, ALTER INDEX ... SET
(fastupdate = off) and one gin_clean_pending_list() merge its backlog.
The ALTER takes a SHARE UPDATE EXCLUSIVE lock, which blocks neither
reads nor writes; after the first startup the step is one catalog read
per index. A failure is logged and startup goes on (search still
works, only slower).

Per-table autovacuum settings for the search tables are left alone.
The pending list was the only reason the insert threshold mattered for
search; dead tuples and freezing are served by the defaults, and table
settings would override whatever tuning the DBA has done.

MySQL is unaffected: InnoDB FULLTEXT keeps new entries in an in-memory
cache that queries read directly, with no setting like fastupdate.

store::search_gin::postgres_gin_fastupdate (new, PostgreSQL) builds
the search schema in a schema of its own and checks pg_class.reloptions:
fastupdate=off on every GIN index of a fresh schema; then, with the
option reset to the default and 500 rows pending, one startup turns it
off everywhere and leaves no pending tuples (pgstatginindex); a second
startup changes nothing. On main it fails at the first check.

crates/common/src/cache/reload.rs

@@ -155,12 +155,6 @@ impl Server {
                             .await
                             .ok();
 
-                        // inbuxa: the task manager reads the node's role on
-                        // every scan; scan now, so a role that gained task
-                        // types starts claiming them without waiting out the
-                        // refresh interval
-                        self.inner.ipc.task_tx.notify_one();
-
                         self.record_build_errors(&bootstrap.errors);
 
                         return Ok(ReloadResult {

crates/services/src/task_manager/manager.rs

@@ -55,12 +55,23 @@ const PERPETUAL_RETRY_MIN_DELAY: u64 = 3600;
 const PERPETUAL_RETRY_MAX_DELAY: u64 = 21600;
 
 pub fn spawn_task_manager(inner: Arc<Inner>) {
-    // inbuxa: upstream didn't start the task manager on a node whose role
-    // had no task types at boot, so adding one later did nothing until a
-    // restart. It now always runs and reads the role on every scan and
-    // before every job (task_enabled), so a role change applies at the next
-    // settings reload.
-    let is_clustered = inner.build_server().core.storage.coordinator.is_enabled();
+    let is_clustered = {
+        let server = inner.build_server();
+        let roles = &server.core.network.roles;
+
+        // inbuxa: outbound_mta too, which now governs report tasks
+        if !roles.account_maintenance
+            && !roles.store_maintenance
+            && !roles.search_indexing
+            && !roles.spam_training
+            && !roles.outbound_mta
+            && !roles.task_manager
+        {
+            return;
+        }
+
+        server.core.storage.coordinator.is_enabled()
+    };
 
     trc::event!(TaskManager(TaskManagerEvent::ManagerStarted));
 
@@ -140,23 +151,20 @@ pub fn spawn_task_manager(inner: Arc<Inner>) {
                     let server = inner.build_server();
                     let batch_size = server.core.email.index_batch_size;
                     let mut batch = Vec::with_capacity(batch_size);
-                    if let Some(task) = fetch_enabled_task(&server, job).await {
+                    if let Some(task) = fetch_task(&server, job).await {
                         batch.push(task);
                     }
 
                     while batch.len() < batch_size {
                         match rx.try_recv() {
                             Ok(job) => {
-                                if let Some(task) = fetch_enabled_task(&server, job).await {
+                                if let Some(task) = fetch_task(&server, job).await {
                                     batch.push(task);
                                 }
                             }
                             Err(_) => break,
                         }
                     }
-                    if batch.is_empty() {
-                        continue;
-                    }
 
                     // Dispatch. inbuxa: on a task of its own, so a panic
                     // releases the batch's locks and leaves this worker
@@ -197,8 +205,7 @@ pub fn spawn_task_manager(inner: Arc<Inner>) {
                     let server = inner.build_server();
                     let mut refresh_queue = false;
 
-                    if let Some(TaskDetails { task, info }) = fetch_enabled_task(&server, job).await
-                    {
+                    if let Some(TaskDetails { task, info }) = fetch_task(&server, job).await {
                         // inbuxa: on a task of its own, as above
                         let run = {
                             let server = server.clone();
@@ -267,17 +274,6 @@ impl TaskQueueManager for Server {
         if task_locks.is_stopping() {
             return Duration::from_secs(QUEUE_REFRESH_INTERVAL);
         }
-        // inbuxa: with no task type enabled by this node's role there is
-        // nothing to claim; a settings reload wakes the manager when that
-        // changes
-        let roles = &self.core.network.roles;
-        if !(0..TaskType::COUNT as u16)
-            .filter_map(TaskType::from_id)
-            .any(|task_type| task_enabled(roles, task_type))
-        {
-            ipc.locked.clear();
-            return Duration::from_secs(QUEUE_REFRESH_INTERVAL);
-        }
         let lock_expiry = task_locks.expiry();
         let now_timestamp = now();
         let from_key = ValueKey::<ValueClass> {
@@ -300,6 +296,7 @@ impl TaskQueueManager for Server {
         let mut tasks = Vec::new();
         let now = Instant::now();
         let mut next_event = None;
+        let roles = &self.core.network.roles;
         ipc.revision += 1;
         let _ = self
             .store()
@@ -547,25 +544,6 @@ async fn run_task(
     }
 }
 
-/// inbuxa: reads a claimed task when this node's role still allows its type.
-/// The role may have changed since the task was claimed (a settings reload in
-/// between); the claim is then handed back at once for a node that may run
-/// it, rather than held until the lease runs out.
-async fn fetch_enabled_task(server: &Server, job: TaskJob) -> Option<TaskDetails> {
-    if task_enabled(&server.core.network.roles, job.typ) {
-        fetch_task(server, job).await
-    } else {
-        trc::event!(
-            TaskManager(TaskManagerEvent::TaskIgnored),
-            Id = job.id,
-            Details = job.typ.as_str(),
-            Reason = "Task type was disabled by cluster roles after it was claimed.",
-        );
-        server.remove_index_lock(job.id).await;
-        None
-    }
-}
-
 /// Reads a claimed task. When it is gone or can't be read, the claim is
 /// released: inbuxa: holding it would block the task, everywhere, until
 /// the lock expired.

crates/smtp/src/lib.rs

@@ -44,14 +44,7 @@ impl StartQueueManager for BootManager {
 impl SpawnQueueManager for IpcReceivers {
     fn spawn_queue_manager(&mut self, inner: Arc<Inner>) {
         let core = inner.shared_core.load();
-        // inbuxa: upstream started these only when the node's role included
-        // outboundMta at boot, so turning the role on later did nothing and
-        // turning it off left them delivering until a restart. They now run
-        // on every node and follow the role live (see Queue::start and the
-        // report scheduler). This also drains the queue channel on nodes
-        // without the role, where every queued message's refresh used to sit
-        // in a channel nobody read until it filled and queueing blocked.
-        if !core.storage.registry.is_recovery_mode() {
+        if !core.storage.registry.is_recovery_mode() && core.network.roles.outbound_mta {
             // Spawn queue manager
             self.queue_rx.take().unwrap().spawn(inner.clone());
 

crates/smtp/src/queue/manager.rs

@@ -2,8 +2,6 @@
  * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
  *
  * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
- *
- * Modified by Coffey Labs in 2026 for INBUXA.
  */
 
 use super::{Message, QueueId, Status, spool::SmtpSpool};
@@ -41,9 +39,6 @@ pub struct Queue {
     pub urgent_refresh: bool,
     pub last_scan: Instant,
     pub last_full_scan: Instant,
-    /// inbuxa: whether this node's role included outboundMta when last
-    /// checked (None before the first check)
-    pub role_enabled: Option<bool>,
 }
 
 #[derive(Debug)]
@@ -72,9 +67,6 @@ impl SpawnQueue for mpsc::Receiver<QueueEvent> {
 const BACK_PRESSURE_WARN_INTERVAL: Duration = Duration::from_secs(60);
 const MIN_SCAN_INTERVAL: Duration = Duration::from_millis(100);
 const FULL_SCAN_INTERVAL: Duration = Duration::from_secs(QUEUE_REFRESH / 2);
-/// inbuxa: how often a node without the outbound MTA role looks at its role
-/// again when nothing else wakes it (a settings reload does)
-const ROLE_RECHECK_INTERVAL: Duration = Duration::from_secs(30);
 
 impl Queue {
     pub fn new(core: Arc<Inner>, rx: mpsc::Receiver<QueueEvent>) -> Self {
@@ -95,7 +87,6 @@ impl Queue {
             urgent_refresh: false,
             last_scan: now.checked_sub(MIN_SCAN_INTERVAL).unwrap_or(now),
             last_full_scan: now,
-            role_enabled: None,
         }
     }
 
@@ -132,27 +123,6 @@ impl Queue {
                 continue;
             }
 
-            // inbuxa: follow the node's role live. Without outboundMta the
-            // queue claims nothing new; deliveries already running finish
-            // and report back as usual, releasing their locks. When the role
-            // comes back, the whole queue is scanned at once.
-            let role_enabled = self.core.shared_core.load().network.roles.outbound_mta;
-            if self.role_enabled.replace(role_enabled) == Some(false) && role_enabled {
-                trc::event!(
-                    Queue(trc::QueueEvent::Started),
-                    Details = "This node's cluster role now includes outboundMta",
-                );
-                self.scan_from = 0;
-                self.pending_refresh = true;
-                self.urgent_refresh = true;
-            }
-            if !role_enabled {
-                self.pending_refresh = false;
-                self.urgent_refresh = false;
-                self.next_refresh = Instant::now() + ROLE_RECHECK_INTERVAL;
-                continue;
-            }
-
             self.pending_refresh |= refresh_queue;
             if !self.pending_refresh && self.next_refresh > Instant::now() {
                 continue;

crates/smtp/src/reporting/scheduler.rs

@@ -2,8 +2,6 @@
  * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
  *
  * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
- *
- * Modified by Coffey Labs in 2026 for INBUXA.
  */
 
 use super::{dmarc::DmarcReporting, tls::TlsReporting};
@@ -20,14 +18,6 @@ impl SpawnReport for mpsc::Receiver<ReportingEvent> {
         tokio::spawn(async move {
             while let Some(event) = self.recv().await {
                 let server = inner.build_server();
-                // inbuxa: reports are the outbound MTA's business, as at
-                // boot, but the role is read per event so a change applies
-                // without a restart. Events that arrive while the role is
-                // off are dropped, as they were on a node started without it
-                if !matches!(event, ReportingEvent::Stop) && !server.core.network.roles.outbound_mta
-                {
-                    continue;
-                }
                 match event {
                     ReportingEvent::Dmarc(event) => server.schedule_dmarc(event).await,
                     ReportingEvent::Tls(event) => server.schedule_tls(event).await,

crates/store/src/backend/postgres/main.rs

@@ -265,12 +265,21 @@ async fn create_search_tables<T: SearchableField + PsqlSearchField + 'static>(
     for field in T::all_fields() {
         if field.is_text() || field.is_json() {
             let column_name = field.column();
+            // inbuxa: with GIN's default fastupdate=on, new entries wait in
+            // an unindexed pending list that every search scans in full
+            // until a VACUUM (or 4 MB of backlog) merges it. On a mailbox
+            // taking steady mail that list never drains and searches slow
+            // from milliseconds to hundreds of them. Pay the index update
+            // at insert time instead.
+            let index_name = format!("gin_{table_name}_{column_name}");
             let create_index_query = format!(
-                "CREATE INDEX IF NOT EXISTS gin_{table_name}_{column_name} ON {table_name} USING GIN({column_name})",
+                "CREATE INDEX IF NOT EXISTS {index_name} ON {table_name} USING GIN({column_name}) WITH (fastupdate = off)",
             );
             conn.execute(&create_index_query, &[])
                 .await
                 .map_err(into_error)?;
+            // Indexes made before this change keep fastupdate=on
+            disable_gin_fastupdate(conn, &index_name).await;
         }
 
         if field.is_indexed() {
@@ -287,6 +296,69 @@ async fn create_search_tables<T: SearchableField + PsqlSearchField + 'static>(
     Ok(())
 }
 
+/// inbuxa: turns fastupdate off on a GIN index made with the default and
+/// merges the pending list it has built up. Idempotent: an index that already
+/// has the option is left alone, so this costs one catalog read per index at
+/// startup. A failure is logged and startup goes on, since search still works,
+/// only slower.
+async fn disable_gin_fastupdate(conn: &Object, index_name: &str) {
+    if let Err(err) = try_disable_gin_fastupdate(conn, index_name).await {
+        trc::event!(
+            Store(trc::StoreEvent::PostgresqlError),
+            Details = format!("Failed to turn off fastupdate on search index {index_name}"),
+            Reason = err.to_string(),
+        );
+    }
+}
+
+async fn try_disable_gin_fastupdate(conn: &Object, index_name: &str) -> trc::Result<()> {
+    let options = conn
+        .query_opt(
+            "SELECT COALESCE(reloptions, '{}')::text[] FROM pg_class WHERE oid = to_regclass($1)",
+            &[&index_name],
+        )
+        .await
+        .map_err(into_error)?
+        .map(|row| row.try_get::<_, Vec<String>>(0))
+        .transpose()
+        .map_err(into_error)?;
+    let Some(options) = options else {
+        return Ok(());
+    };
+    if gin_fastupdate_is_off(&options) {
+        return Ok(());
+    }
+    // SET (fastupdate) takes a SHARE UPDATE EXCLUSIVE lock, which doesn't
+    // block reads or writes. Turning it off stops new entries going to the
+    // pending list but doesn't flush the entries already there.
+    conn.execute(
+        &format!("ALTER INDEX {index_name} SET (fastupdate = off)"),
+        &[],
+    )
+    .await
+    .map_err(into_error)?;
+    conn.query_one(
+        "SELECT gin_clean_pending_list($1::text::regclass)",
+        &[&index_name],
+    )
+    .await
+    .map_err(into_error)?;
+    Ok(())
+}
+
+/// Whether a relation's reloptions turn GIN's fastupdate off.
+fn gin_fastupdate_is_off(options: &[String]) -> bool {
+    options.iter().any(|option| {
+        option.split_once('=').is_some_and(|(name, value)| {
+            name.trim().eq_ignore_ascii_case("fastupdate")
+                && matches!(
+                    value.trim().to_ascii_lowercase().as_str(),
+                    "off" | "false" | "no" | "0" | "f" | "n"
+                )
+        })
+    })
+}
+
 async fn discover_ts_configs(pool: &Pool) -> AHashSet<&'static str> {
     let mut ts_configs = AHashSet::from_iter([PG_FALLBACK_LANG, PG_UNSTEMMED_LANG]);
 

tests/src/cluster/live_roles.rs

@@ -1,297 +0,0 @@
-/*
- * SPDX-FileCopyrightText: 2026 Coffey Labs
- *
- * SPDX-License-Identifier: AGPL-3.0-only
- */
-
-//! A node follows edits to its cluster role without a restart: outbound
-//! delivery and report tasks start when the role gains outboundMta and stop
-//! when it loses it. Upstream decided at boot whether the queue, report and
-//! task managers ran at all. Needs a store the seed and the node can share
-//! (STORE=PostgreSql or MySql).
-
-use crate::utils::server::{TestServer, TestServerBuilder};
-use common::Server;
-use registry::{
-    schema::{
-        enums::ClusterTaskType,
-        prelude::{Object, ObjectType},
-        structs::{
-            ClusterListenerGroup, ClusterRole, ClusterTaskGroup, ClusterTaskGroupProperties, Task,
-            TaskStatus, TaskTlsReport,
-        },
-    },
-    types::{id::ObjectId, map::Map},
-};
-use smtp::{
-    queue::{Message, Status},
-    reporting::send::MtaReportSend,
-};
-use std::time::{Duration, Instant};
-use store::{
-    Deserialize, IterateParams, ValueKey,
-    registry::write::{RegistryWrite, RegistryWriteResult},
-    write::{AlignedBytes, Archive, BatchBuilder, QueueClass, TaskQueueClass, ValueClass},
-};
-use types::id::Id;
-use utils::snowflake::SnowflakeIdGenerator;
-
-const BUSY_ROLE: &str = "live_role_busy";
-const IDLE_ROLE: &str = "live_role_idle";
-const WITH: &[ClusterTaskType] = &[
-    ClusterTaskType::PushNotifications,
-    ClusterTaskType::OutboundMta,
-];
-const WITHOUT: &[ClusterTaskType] = &[ClusterTaskType::PushNotifications];
-const RCPT_DOMAIN: &str = "live-role.invalid";
-
-#[tokio::test(flavor = "multi_thread")]
-pub async fn live_role_tests() {
-    if matches!(
-        std::env::var("STORE").as_deref(),
-        Ok("RocksDb" | "Sqlite") | Err(_)
-    ) {
-        println!("Skipping live role tests: they need a store the nodes can share.");
-        return;
-    }
-    println!(
-        "Running live role tests on {}...",
-        std::env::var("STORE").unwrap_or_default()
-    );
-
-    // Two roles: one with outboundMta, one with no task type at all
-    let seed = TestServerBuilder::new("live_roles_seed").await;
-    let busy_id = seed.insert_object(role(BUSY_ROLE, WITH)).await;
-    let idle_id = seed.insert_object(role(IDLE_ROLE, WITHOUT)).await;
-    let seed = seed.disable_services().build().await;
-    let registry = seed.server.clone();
-
-    // 1. The rehearsal case: a node started with outboundMta has it taken
-    // away. Upstream kept delivering, report messages included, until a
-    // restart.
-    let node = start_node("live_roles_busy", BUSY_ROLE).await;
-    let server = node.server.clone();
-    assert!(server.core.network.roles.outbound_mta);
-    let (msg, task) = queue_work(&server, "busy-before").await;
-    assert_runs(&server, &msg, task).await;
-
-    set_role(&registry, &server, busy_id, role(BUSY_ROLE, WITHOUT)).await;
-    let (msg, task) = queue_work(&server, "busy-off").await;
-    assert_idle(&server, &msg, task).await;
-
-    // Given back, it takes up the work left waiting
-    set_role(&registry, &server, busy_id, role(BUSY_ROLE, WITH)).await;
-    assert_runs(&server, &msg, task).await;
-
-    // Off again, so it leaves the next node's work alone
-    set_role(&registry, &server, busy_id, role(BUSY_ROLE, WITHOUT)).await;
-
-    // 2. A node started with no task type at all gains outboundMta.
-    // Upstream never started its queue, report or task manager, so the
-    // role did nothing until a restart.
-    let node2 = start_node("live_roles_idle", IDLE_ROLE).await;
-    let server2 = node2.server.clone();
-    assert!(!server2.core.network.roles.outbound_mta);
-    let (msg, task) = queue_work(&server2, "idle-off").await;
-    assert_idle(&server2, &msg, task).await;
-    set_role(&registry, &server2, idle_id, role(IDLE_ROLE, WITH)).await;
-    assert_runs(&server2, &msg, task).await;
-    set_role(&registry, &server2, idle_id, role(IDLE_ROLE, WITHOUT)).await;
-
-    if seed.is_reset() {
-        seed.temp_dir.delete();
-        node.temp_dir.delete();
-        node2.temp_dir.delete();
-    }
-}
-
-async fn start_node(name: &str, role: &str) -> TestServer {
-    TestServerBuilder::new_with_role(
-        name,
-        format!("{name}.example.com").replace('_', "-"),
-        Some(role.into()),
-        false,
-    )
-    .await
-    .build_with_opts(false)
-    .await
-}
-
-/// Neither the message nor the report task is touched.
-async fn assert_idle(server: &Server, msg: &str, task: u64) {
-    tokio::time::sleep(Duration::from_secs(4)).await;
-    server.notify_task_queue();
-    tokio::time::sleep(Duration::from_secs(1)).await;
-    assert!(
-        !attempted(server, msg).await,
-        "delivery attempted without outboundMta"
-    );
-    assert!(
-        is_pending(server, task).await,
-        "report task claimed without outboundMta"
-    );
-}
-
-/// Delivery of the message is attempted and the report task runs.
-async fn assert_runs(server: &Server, msg: &str, task: u64) {
-    wait_for(Duration::from_secs(20), "message delivery attempt", || {
-        attempted(server, msg)
-    })
-    .await;
-    wait_for(Duration::from_secs(20), "report task to run", || async {
-        !is_pending(server, task).await
-    })
-    .await;
-}
-
-fn role(name: &str, tasks: &[ClusterTaskType]) -> ClusterRole {
-    ClusterRole {
-        name: name.into(),
-        description: None,
-        listeners: ClusterListenerGroup::EnableAll,
-        tasks: ClusterTaskGroup::EnableSome(ClusterTaskGroupProperties {
-            task_types: Map::new(tasks.to_vec()),
-        }),
-    }
-}
-
-/// Stores a new version of a role and reloads the node's settings, as a
-/// JMAP write to the role does.
-async fn set_role(registry: &Server, node: &Server, id: Id, new: ClusterRole) {
-    let enabled = matches!(&new.tasks, ClusterTaskGroup::EnableSome(group)
-        if group.task_types.iter().any(|t| *t == ClusterTaskType::OutboundMta));
-    let old = registry
-        .registry()
-        .get(ObjectId::new(ObjectType::ClusterRole, id))
-        .await
-        .unwrap()
-        .expect("role not found");
-    let new = Object::from(new);
-    let result = registry
-        .registry()
-        .write(RegistryWrite::update(id, &new, &old))
-        .await
-        .unwrap();
-    assert!(
-        matches!(result, RegistryWriteResult::Success(_)),
-        "role update refused"
-    );
-    assert_eq!(
-        node.reload_after_write(ObjectType::ClusterRole).await,
-        Some(Ok(()))
-    );
-    assert_eq!(
-        node.inner.shared_core.load().network.roles.outbound_mta,
-        enabled
-    );
-}
-
-/// Queues a message to an unreachable domain and schedules a TLS report
-/// task, both due now. Returns the recipient's local part and the task id.
-async fn queue_work(server: &Server, name: &str) -> (String, u64) {
-    let local = format!("{name}-{}", SnowflakeIdGenerator::global_id().unwrap());
-    let rcpt = format!("{local}@{RCPT_DOMAIN}");
-    server
-        .send_autogenerated(
-            "[email protected]",
-            [rcpt.as_str()].into_iter(),
-            format!(
-                "From: [email protected]\r\nTo: {rcpt}\r\n\
-                 Subject: live role test\r\n\r\nTest\r\n"
-            )
-            .into_bytes(),
-            None,
-            0,
-        )
-        .await;
-    assert!(
-        queued_recipient(server, &rcpt).await.is_some(),
-        "message to {rcpt} was not queued"
-    );
-
-    let task = SnowflakeIdGenerator::global_id().unwrap();
-    let mut batch = BatchBuilder::new();
-    batch.schedule_task_with_id(
-        task,
-        Task::TlsReport(TaskTlsReport {
-            report_id: u64::MAX.into(),
-            status: TaskStatus::now(),
-        }),
-    );
-    server.store().write(batch.build_all()).await.unwrap();
-    server.notify_task_queue();
-
-    (rcpt, task)
-}
-
-/// Whether delivery to `rcpt` was tried: the message is gone, or its
-/// recipient is no longer scheduled or has a retry count.
-async fn attempted(server: &Server, rcpt: &str) -> bool {
-    match queued_recipient(server, rcpt).await {
-        None => true,
-        Some((status_scheduled, retries)) => !status_scheduled || retries > 0,
-    }
-}
-
-/// The queued recipient `rcpt`: whether it is still scheduled, and how many
-/// times delivery was retried.
-async fn queued_recipient(server: &Server, rcpt: &str) -> Option<(bool, u32)> {
-    let mut found = None;
-    server
-        .store()
-        .iterate(
-            IterateParams::new(
-                ValueKey::from(ValueClass::Queue(QueueClass::Message(0))),
-                ValueKey::from(ValueClass::Queue(QueueClass::Message(u64::MAX))),
-            ),
-            |_, value| {
-                let message = <Archive<AlignedBytes> as Deserialize>::deserialize(value)?
-                    .deserialize::<Message>()?;
-                if let Some(recipient) = message
-                    .recipients
-                    .iter()
-                    .find(|recipient| recipient.address.as_ref() == rcpt)
-                {
-                    found = Some((
-                        matches!(recipient.status, Status::Scheduled),
-                        recipient.retry.inner,
-                    ));
-                    return Ok(false);
-                }
-                Ok(true)
-            },
-        )
-        .await
-        .unwrap();
-    found
-}
-
-async fn is_pending(server: &Server, id: u64) -> bool {
-    matches!(
-        server
-            .store()
-            .get_value::<Task>(ValueKey::from(ValueClass::TaskQueue(
-                TaskQueueClass::Task { id },
-            )))
-            .await
-            .unwrap()
-            .map(|task| task.status().clone()),
-        Some(TaskStatus::Pending(_))
-    )
-}
-
-async fn wait_for<F, Fut>(within: Duration, what: &str, mut check: F)
-where
-    F: FnMut() -> Fut,
-    Fut: Future<Output = bool>,
-{
-    let started = Instant::now();
-    while !check().await {
-        assert!(
-            started.elapsed() < within,
-            "still waiting for the {what} after {:?}",
-            started.elapsed()
-        );
-        tokio::time::sleep(Duration::from_millis(250)).await;
-    }
-}

tests/src/cluster/mod.rs

@@ -7,7 +7,6 @@
  */
 
 pub mod broadcast;
-pub mod live_roles; // inbuxa: role edits apply without a restart
 #[cfg(feature = "nats")]
 pub mod coordinator; // inbuxa: coordinator reconnects
 pub mod stress;

tests/src/store/mod.rs

@@ -21,6 +21,8 @@ pub mod replica_mysql; // inbuxa: read replicas on MySQL
 #[cfg(all(feature = "postgres", feature = "redis"))]
 pub mod replica_cluster; // inbuxa: read replicas across nodes
 pub mod scaleout; // inbuxa: scale-out storage
+#[cfg(feature = "postgres")]
+pub mod search_gin; // inbuxa: GIN indexes without a pending list
 #[cfg(any(feature = "postgres", feature = "mysql"))]
 pub mod sql_timeout;
 pub mod task_locks; // inbuxa: task locks across nodes

tests/src/store/search_gin.rs

@@ -0,0 +1,159 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! PostgreSQL full-text GIN indexes are built with fastupdate off, and an
+//! index made earlier with the default is switched over at startup. With
+//! fastupdate on, new entries wait in a pending list that every search scans
+//! in full until VACUUM merges it.
+
+use crate::utils::storage::build_data_store;
+use registry::schema::structs::DataStore;
+use store::{Rows, SearchStore, Store};
+
+const SCHEMA: &str = "gin_fastupdate_test";
+
+#[tokio::test(flavor = "multi_thread")]
+pub async fn postgres_gin_fastupdate() {
+    println!("Running PostgreSQL GIN fastupdate test...");
+
+    // Work in a schema of our own so the shared search tables are untouched
+    let admin = Store::build(build_data_store("PostgreSql", "").await)
+        .await
+        .expect("Failed to connect to PostgreSQL");
+    for query in [
+        format!("DROP SCHEMA IF EXISTS {SCHEMA} CASCADE"),
+        format!("CREATE SCHEMA {SCHEMA}"),
+    ] {
+        admin.sql_query::<usize>(&query, vec![]).await.unwrap();
+    }
+    let DataStore::PostgreSql(mut config) = build_data_store("PostgreSql", "").await else {
+        unreachable!()
+    };
+    config.options = Some(format!("-c search_path={SCHEMA}"));
+    let store = Store::build(DataStore::PostgreSql(config))
+        .await
+        .expect("Failed to connect to PostgreSQL");
+    let search = SearchStore::Store(store.clone());
+
+    // A fresh schema
+    search.create_indexes().await.unwrap();
+    let indexes = gin_indexes(&admin).await;
+    assert!(
+        indexes.len() >= 4,
+        "expected the search GIN indexes, found {indexes:?}"
+    );
+    for (name, options) in &indexes {
+        assert!(
+            options.contains("fastupdate=off"),
+            "fresh index {name} has options {options:?}"
+        );
+    }
+
+    // A schema from before the change: the same indexes, made with the
+    // default fastupdate=on, and a pending list with something in it
+    for (name, _) in &indexes {
+        admin
+            .sql_query::<usize>(
+                &format!("ALTER INDEX {SCHEMA}.{name} RESET (fastupdate)"),
+                vec![],
+            )
+            .await
+            .unwrap();
+    }
+    for (name, options) in gin_indexes(&admin).await {
+        assert!(
+            !options.contains("fastupdate"),
+            "index {name} still has options {options:?}"
+        );
+    }
+    admin
+        .sql_query::<usize>(
+            &format!(
+                "INSERT INTO {SCHEMA}.s_email (accid, docid, subj, body) \
+                 SELECT 1, n, to_tsvector('simple', 'pending subject ' || n), \
+                 to_tsvector('simple', 'pending body text ' || n) \
+                 FROM generate_series(1, 500) n"
+            ),
+            vec![],
+        )
+        .await
+        .unwrap();
+    assert!(
+        pending_tuples(&admin, "gin_s_email_body").await > 0,
+        "no pending list to merge"
+    );
+
+    // Startup on the existing schema switches every index over and merges
+    // what was pending
+    search.create_indexes().await.unwrap();
+    for (name, options) in gin_indexes(&admin).await {
+        assert!(
+            options.contains("fastupdate=off"),
+            "existing index {name} has options {options:?} after startup"
+        );
+    }
+    assert_eq!(pending_tuples(&admin, "gin_s_email_body").await, 0);
+
+    // And a second startup changes nothing
+    search.create_indexes().await.unwrap();
+    for (name, options) in gin_indexes(&admin).await {
+        assert!(options.contains("fastupdate=off"), "{name}: {options:?}");
+    }
+
+    admin
+        .sql_query::<usize>(&format!("DROP SCHEMA {SCHEMA} CASCADE"), vec![])
+        .await
+        .unwrap();
+}
+
+/// The GIN indexes in the test schema with their reloptions.
+async fn gin_indexes(admin: &Store) -> Vec<(String, String)> {
+    admin
+        .sql_query::<Rows>(
+            &format!(
+                "SELECT c.relname::text, COALESCE(array_to_string(c.reloptions, ','), '') \
+                 FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace \
+                 JOIN pg_am a ON a.oid = c.relam \
+                 WHERE n.nspname = '{SCHEMA}' AND c.relkind = 'i' AND a.amname = 'gin' \
+                 ORDER BY 1"
+            ),
+            vec![],
+        )
+        .await
+        .unwrap()
+        .rows
+        .into_iter()
+        .map(|row| {
+            let mut values = row.values.into_iter();
+            (
+                values.next().unwrap().to_str().into_owned(),
+                values.next().unwrap().to_str().into_owned(),
+            )
+        })
+        .collect()
+}
+
+/// Tuples waiting in a GIN index's pending list (pgstattuple is a contrib
+/// extension the test database has).
+async fn pending_tuples(admin: &Store, index: &str) -> i64 {
+    admin
+        .sql_query::<usize>("CREATE EXTENSION IF NOT EXISTS pgstattuple", vec![])
+        .await
+        .unwrap();
+    admin
+        .sql_query::<Rows>(
+            &format!("SELECT pending_tuples FROM pgstatginindex('{SCHEMA}.{index}'::regclass)"),
+            vec![],
+        )
+        .await
+        .unwrap()
+        .rows
+        .into_iter()
+        .next()
+        .and_then(|row| row.values.into_iter().next())
+        .map(|value| value.to_str().parse::<i64>().unwrap())
+        .unwrap()
+}