Search: find addresses by local part, domain or name on PostgreSQL and MySQL

A 3-node PostgreSQL rehearsal found IMAP SEARCH FROM "noreply" matched
0-2 messages where RocksDB matched 23 of 930. The message indexer hands
each address and display name of From/To/Cc/Bcc to the search store as
keyword text (Language::None). The built-in index splits keyword text
into lowercase runs of alphanumerics, so an address is found by its full
form, its local part, its domain or a display-name word. The SQL
backends didn't:

- PostgreSQL's text parser keeps "[email protected]" as one email
  token (host names and URLs likewise), so neither "noreply" nor
  "amazon.com" ever matched it. Keyword text is now split the same way
  as the built-in index (SpaceTokenizer) before to_tsvector on insert
  and before plainto_tsquery/phraseto_tsquery on search, still under
  the 'simple' configuration, so the GIN index keeps serving the query.
  The sort columns keep the raw text.
- MySQL's FULLTEXT parser already splits on punctuation, but InnoDB
  never indexes its stopwords ("com", "de", "www", ...) or words under
  innodb_ft_min_token_size (3), and a required +word it hasn't indexed
  matches no row. So "amazon.com", "[email protected]" or "jane doe" found
  nothing. Those words are now matched with a word-boundary REGEXP on
  the rows the indexed words select. In language text (bodies,
  subjects) they are dropped when other words remain, and only checked
  when nothing else is left, so "the invoice" no longer finds nothing
  either.

Existing PostgreSQL search indexes hold the old single-token vectors and
need a reindex (the reindexAccounts task) before address searches find
old messages. MySQL needs none: only the query changed.

store::search_tests gains test_address_search: five messages, 28
FROM/TO/CC/BCC searches by full address, local part, domain, domain
labels, display name and hyphenated local part, plus a TEXT-style OR,
with the same expected ids on every backend. It passes on RocksDB,
SQLite, PostgreSQL and MySQL; on main it fails on PostgreSQL (From
"noreply") and MySQL (From "[email protected]").

.gitea/workflows/publish.yml

@@ -3,11 +3,28 @@
 # whether a person pushed it or weekly-release.yml created it through the
 # releases API.
 #
-# The image is multi-arch (linux/amd64, linux/arm64) as before, but built in
-# one buildx run on host1 instead of one native runner per architecture: the
-# Dockerfile's builder stage runs on the build platform and cross-compiles
-# with an aarch64 linker, so only the small final stage (apt, setcap) goes
-# through QEMU for arm64. No digest-joining job is needed.
+# The image is multi-arch (linux/amd64, linux/arm64), built by two jobs on
+# the image-build runner rather than one buildx run for both. The Dockerfile's
+# builder stage runs on the build platform and cross-compiles with an aarch64
+# linker, so only the small final stage (apt, setcap) goes through QEMU for
+# arm64 -- but two release builds (LTO, one codegen unit) side by side on one
+# machine each take twice as long. Production runs amd64, so amd64 goes first
+# and on its own:
+#   * publish-amd64 pushes :<version>-amd64 and :<version>, a plain amd64
+#     image, as soon as its build is done. A deploy can start from it.
+#   * publish-arm64 then builds arm64, pushes :<version>-arm64, and replaces
+#     :<version> with the two-platform index. :latest moves only here, so it
+#     never names an image without arm64.
+#
+# Both jobs use one BuildKit builder, `gitea-builder`, whose container
+# (buildx_buildkit_gitea-builder0) and state volume stay on the runner's host
+# between jobs: a job container's `buildx create` finds the existing container
+# and reuses it and its cache. The dependency build (`cargo chef cook`) is
+# keyed on the recipe, which only a dependency change alters, so a release
+# normally compiles just the workspace. Removing that container or its volume
+# costs the next release a cold build, nothing more. The planner and dependency
+# layers for the build platform are shared, so arm64 also reuses what amd64
+# just did where it can.
 #
 # Two guards before anything is pushed:
 #   * the tag must be v<brand_version!>. The version is a string in
@@ -62,7 +79,7 @@ jobs:
           echo "version=$V" >> "$GITHUB_OUTPUT"
           echo "version $V"
 
-  publish:
+  publish-amd64:
     needs: [version]
     runs-on: docker
     container:
@@ -81,16 +98,15 @@ jobs:
           test -n "$REGISTRY" && test -n "$VERSION"
           test -n "$PACKAGE_TOKEN" || { echo "PACKAGE_TOKEN secret is not set on this repository" >&2; exit 1; }
           echo "$PACKAGE_TOKEN" | docker login -u jcoffey-dev --password-stdin "$REGISTRY"
-          docker run --privileged --rm tonistiigi/binfmt --install arm64
           docker buildx create --use --name gitea-builder --driver docker-container || docker buildx use gitea-builder
-      # Attestations off, as before: they add manifests of their own to the
-      # index, and the index should hold the two images and nothing else.
+      # Attestations off, as before: they add manifests of their own, and the
+      # index should hold the two images and nothing else.
       - run: |
           docker buildx build \
-            --platform linux/amd64,linux/arm64 \
+            --platform linux/amd64 \
             --provenance=false --sbom=false \
+            --tag "$IMAGE:$VERSION-amd64" \
             --tag "$IMAGE:$VERSION" \
-            --tag "$IMAGE:latest" \
             --push .
           docker buildx imagetools inspect "$IMAGE:$VERSION"
       # Gitea keeps a container package on its owner; linking it shows it on
@@ -103,11 +119,47 @@ jobs:
       - if: always()
         run: docker logout "$REGISTRY" || true
 
+  publish-arm64:
+    needs: [version, publish-amd64]
+    runs-on: docker
+    container:
+      image: docker:28-cli@sha256:625d9431a9f54c5a2bc90f24f0e1c3d55b1349fd857dd85035f98c2c9acbdd4d # 28-cli
+      volumes:
+        - /var/run/docker.sock:/var/run/docker.sock
+    env:
+      DOCKER_BUILDKIT: "1"
+      REGISTRY: ${{ vars.REGISTRY }}
+      IMAGE: ${{ vars.REGISTRY }}/${{ github.repository }}
+      VERSION: ${{ needs.version.outputs.version }}
+      PACKAGE_TOKEN: ${{ secrets.PACKAGE_TOKEN }}
+    steps:
+      - uses: coffey-labs/actions/checkout@fab0c4d45e0162963965f1555df27b7bed5e20ec
+      - run: |
+          echo "$PACKAGE_TOKEN" | docker login -u jcoffey-dev --password-stdin "$REGISTRY"
+          docker run --privileged --rm tonistiigi/binfmt --install arm64
+          docker buildx create --use --name gitea-builder --driver docker-container || docker buildx use gitea-builder
+      # The index is built from the two per-architecture tags rather than from
+      # :<version>, which by now is the amd64 image and would be read as such.
+      - run: |
+          docker buildx build \
+            --platform linux/arm64 \
+            --provenance=false --sbom=false \
+            --tag "$IMAGE:$VERSION-arm64" \
+            --push .
+          docker buildx imagetools create \
+            --tag "$IMAGE:$VERSION" \
+            --tag "$IMAGE:latest" \
+            "$IMAGE:$VERSION-amd64" "$IMAGE:$VERSION-arm64"
+          docker buildx imagetools inspect "$IMAGE:$VERSION"
+      - if: always()
+        run: docker logout "$REGISTRY" || true
+
   # The weekly release creates its Release (and so the tag) first; a tag
   # pushed by hand has none. Either way the tag ends up with exactly one
-  # Release, created after the image exists so its pull instructions work.
+  # Release, created once the amd64 image exists so its pull instructions
+  # work; arm64 and the binaries follow.
   release:
-    needs: [version, publish]
+    needs: [version, publish-amd64]
     runs-on: light
     container:
       image: python:3.13-slim@sha256:8d9d0b8bcf6506481eae4907c18f5e3e7902e629f5f6d684f9e7c32e85e3ddf0 # 3.13-slim
@@ -131,7 +183,9 @@ jobs:
           except urllib.error.HTTPError as e:
               if e.code != 404: raise
           image = f"{os.environ['REGISTRY']}/{os.environ['REPO']}:{version}"
-          body = (f"Container image: `{image}` (linux/amd64, linux/arm64); also `:latest`.\n\n"
+          body = (f"Container image: `{image}` (linux/amd64, linux/arm64); also `:latest`. "
+                  "amd64 is published first; arm64 is added to the same tag when its build "
+                  "finishes, and `:latest` moves then.\n\n"
                   "Binaries for a host install are attached: `inbuxa-linux-amd64.tar.gz` and "
                   "`inbuxa-linux-arm64.tar.gz`, with `SHA256SUMS`. Each is the binary out of this "
                   "release's image for that architecture, so it is the same build. The image "
@@ -154,7 +208,7 @@ jobs:
   # `docker create` does not start anything, so pulling an arm64 image on an
   # amd64 runner and copying a file out of it needs no emulation.
   binaries:
-    needs: [version, publish, release]
+    needs: [version, publish-arm64, release]
     runs-on: docker
     container:
       image: docker:28-cli@sha256:625d9431a9f54c5a2bc90f24f0e1c3d55b1349fd857dd85035f98c2c9acbdd4d # 28-cli

crates/common/src/ipc.rs

@@ -335,3 +335,57 @@ impl EmailPush {
         }
     }
 }
+
+/// inbuxa: the task locks this node holds, so a graceful stop can hand them
+/// back instead of leaving the tasks blocked until the locks expire.
+pub struct TaskLocks {
+    held: parking_lot::Mutex<ahash::AHashSet<u64>>,
+    stopping: AtomicBool,
+    expiry: std::sync::atomic::AtomicU64,
+}
+
+impl TaskLocks {
+    /// How long a task lock lasts, in seconds, unless it is released first.
+    pub const DEFAULT_EXPIRY: u64 = 60 * 60;
+
+    pub fn is_stopping(&self) -> bool {
+        self.stopping.load(Ordering::Acquire)
+    }
+
+    /// Stops new claims and returns the ids of every lock still held.
+    pub fn stop(&self) -> Vec<u64> {
+        self.stopping.store(true, Ordering::Release);
+        self.held.lock().drain().collect()
+    }
+
+    pub fn insert(&self, id: u64) {
+        self.held.lock().insert(id);
+    }
+
+    pub fn remove(&self, id: u64) {
+        self.held.lock().remove(&id);
+    }
+
+    pub fn held(&self) -> usize {
+        self.held.lock().len()
+    }
+
+    pub fn expiry(&self) -> u64 {
+        self.expiry.load(Ordering::Relaxed)
+    }
+
+    /// Changes the lock lifetime; the tests shorten it.
+    pub fn set_expiry(&self, seconds: u64) {
+        self.expiry.store(seconds.max(1), Ordering::Relaxed);
+    }
+}
+
+impl Default for TaskLocks {
+    fn default() -> Self {
+        Self {
+            held: Default::default(),
+            stopping: AtomicBool::new(false),
+            expiry: std::sync::atomic::AtomicU64::new(Self::DEFAULT_EXPIRY),
+        }
+    }
+}

crates/common/src/lib.rs

@@ -279,6 +279,8 @@ pub struct HttpAuthCache {
 pub struct Ipc {
     pub push_tx: mpsc::Sender<PushEvent>,
     pub task_tx: Arc<Notify>,
+    // inbuxa: task locks held by this node, released on a graceful stop
+    pub task_locks: Arc<crate::ipc::TaskLocks>,
     pub queue_tx: mpsc::Sender<QueueEvent>,
     pub report_tx: mpsc::Sender<ReportingEvent>,
     pub broadcast_tx: Option<mpsc::Sender<BroadcastEvent>>,

crates/common/src/manager/boot.rs

@@ -297,6 +297,7 @@ pub fn build_ipc(has_pubsub: bool) -> (Ipc, IpcReceivers) {
             report_tx,
             broadcast_tx: has_pubsub.then_some(broadcast_tx),
             task_tx: Arc::new(Notify::new()),
+            task_locks: Arc::new(crate::ipc::TaskLocks::default()),
             train_task_controller: Arc::new(TrainTaskController::default()),
         },
         IpcReceivers {

crates/main/src/main.rs

@@ -109,6 +109,10 @@ async fn main() -> std::io::Result<()> {
     // Wait for shutdown signal
     wait_for_shutdown().await;
 
+    // inbuxa: hand back the task locks this node holds, so other nodes can
+    // run those tasks now rather than when the locks expire
+    services::task_manager::lock::release_task_locks(&inner.build_server()).await;
+
     // Shutdown collector
     Collector::shutdown();
 

crates/services/src/task_manager/index.rs

@@ -91,7 +91,15 @@ impl SearchIndexTask for Server {
                             build_contact_document(self, account_id, document_id).await
                         }
                         IndexDocumentType::File => {
-                            // File indexing not implemented yet
+                            // File indexing not implemented yet. inbuxa: still
+                            // one result per task: update_tasks pairs them by
+                            // position, and a missing one shifts every result
+                            // after it onto the wrong task
+                            results.push(IndexTaskResult {
+                                task_type: TaskType::Insert,
+                                index: task.document_type,
+                                result: TaskResult::Ignored,
+                            });
                             continue;
                         }
                     };

crates/services/src/task_manager/lock.rs

@@ -2,6 +2,8 @@
  * 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 crate::task_manager::*;
@@ -13,13 +15,21 @@ pub trait TaskLockManager: Sync + Send {
 
 impl TaskLockManager for Server {
     async fn try_lock_task(&self, id: u64) -> bool {
+        // inbuxa: a node that is stopping claims nothing new
+        let locks = &self.inner.ipc.task_locks;
+        if locks.is_stopping() {
+            return false;
+        }
+
         match self
             .in_memory_store()
-            .try_lock(KV_LOCK_TASK, &id.to_be_bytes(), DEFAULT_LOCK_EXPIRY)
+            .try_lock(KV_LOCK_TASK, &id.to_be_bytes(), locks.expiry())
             .await
         {
             Ok(result) => {
-                if !result {
+                if result {
+                    locks.insert(id);
+                } else {
                     trc::event!(
                         TaskManager(TaskManagerEvent::TaskLocked),
                         Id = id,
@@ -48,5 +58,27 @@ impl TaskLockManager for Server {
                     .caused_by(trc::location!())
             );
         }
+        self.inner.ipc.task_locks.remove(id);
     }
 }
+
+/// inbuxa: on a graceful stop, stops claiming tasks and releases every task
+/// lock this node holds, so the rest of the cluster can pick the tasks up at
+/// once instead of after the lock expires. Returns how many were released.
+pub async fn release_task_locks(server: &Server) -> usize {
+    let ids = server.inner.ipc.task_locks.stop();
+    for id in &ids {
+        if let Err(err) = server
+            .in_memory_store()
+            .remove_lock(KV_LOCK_TASK, &id.to_be_bytes())
+            .await
+        {
+            trc::error!(
+                err.details("Failed to release task lock on shutdown")
+                    .ctx(trc::Key::Id, *id)
+                    .caused_by(trc::location!())
+            );
+        }
+    }
+    ids.len()
+}

crates/services/src/task_manager/manager.rs

@@ -2,6 +2,8 @@
  * 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 crate::task_manager::acme::AcmeTask;
@@ -18,7 +20,7 @@ use crate::task_manager::report::{self, SubmitReportTask};
 use crate::task_manager::restore_item::RestoreItemTask;
 use crate::task_manager::spam_classifier::SpamFilterMaintenanceTask;
 use crate::task_manager::{
-    DEFAULT_LOCK_EXPIRY, Locked, QUEUE_REFRESH_INTERVAL, TaskDetails, TaskFailureType, TaskInfo,
+    CLAIM_RECHECK_INTERVAL, Locked, QUEUE_REFRESH_INTERVAL, TaskDetails, TaskFailureType, TaskInfo,
     TaskJob, TaskManagerIpc, TaskResult,
 };
 use common::BuildServer;
@@ -124,72 +126,47 @@ 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);
-                    match server
-                        .store()
-                        .get_value::<Task>(ValueKey::from(ValueClass::TaskQueue(
-                            TaskQueueClass::Task { id: job.id },
-                        )))
-                        .await
-                    {
-                        Ok(Some(task)) => {
-                            batch.push(TaskDetails { task, info: job });
-                        }
-                        Ok(None) => {
-                            trc::event!(
-                                TaskManager(TaskManagerEvent::TaskIgnored),
-                                Id = job.id,
-                                Reason = "Task not found in store, likely already processed.",
-                            );
-                        }
-                        Err(err) => {
-                            trc::error!(
-                                err.id(job.id)
-                                    .details("Failed to retrieve task details.")
-                                    .caused_by(trc::location!())
-                            );
-                        }
+                    if let Some(task) = fetch_task(&server, job).await {
+                        batch.push(task);
                     }
 
                     while batch.len() < batch_size {
                         match rx.try_recv() {
                             Ok(job) => {
-                                match server
-                                    .store()
-                                    .get_value::<Task>(ValueKey::from(ValueClass::TaskQueue(
-                                        TaskQueueClass::Task { id: job.id },
-                                    )))
-                                    .await
-                                {
-                                    Ok(Some(task)) => {
-                                        batch.push(TaskDetails { task, info: job });
-                                    }
-                                    Ok(None) => {
-                                        trc::event!(
-                                            TaskManager(TaskManagerEvent::TaskIgnored),
-                                            Id = job.id,
-                                            Reason = "Task not found in store, likely already processed.",
-                                        );
-                                    }
-                                    Err(err) => {
-                                        trc::error!(
-                                            err.id(job.id)
-                                                .details("Failed to retrieve task details.")
-                                                .caused_by(trc::location!())
-                                        );
-                                    }
+                                if let Some(task) = fetch_task(&server, job).await {
+                                    batch.push(task);
                                 }
                             }
                             Err(_) => break,
                         }
                     }
 
-                    // Dispatch
+                    // Dispatch. inbuxa: on a task of its own, so a panic
+                    // releases the batch's locks and leaves this worker
+                    // running; a dead worker would keep claiming tasks it
+                    // can never run
                     let mut refresh_queue = false;
-                    let results = server.index(&batch).await.into_iter().map(|r| {
-                        refresh_queue |= r.result.is_retry();
-                        r.result
-                    });
-                    update_tasks(&server, &mut batch, results).await;
+                    let ids = batch.iter().map(|task| task.info.id).collect::<Vec<_>>();
+                    let run = {
+                        let server = server.clone();
+                        tokio::spawn(async move {
+                            let results = server.index(&batch).await;
+                            (batch, results)
+                        })
+                    };
+                    match run.await {
+                        Ok((mut batch, results)) => {
+                            let results = results.into_iter().map(|r| {
+                                refresh_queue |= r.result.is_retry();
+                                r.result
+                            });
+                            update_tasks(&server, &mut batch, results).await;
+                        }
+                        Err(err) => {
+                            worker_failed(&server, &ids, err).await;
+                            refresh_queue = true;
+                        }
+                    }
 
                     if refresh_queue || rx.is_empty() {
                         server.notify_task_queue();
@@ -203,83 +180,31 @@ pub fn spawn_task_manager(inner: Arc<Inner>) {
                     let server = inner.build_server();
                     let mut refresh_queue = false;
 
-                    match server
-                        .store()
-                        .get_value::<Task>(ValueKey::from(ValueClass::TaskQueue(
-                            TaskQueueClass::Task { id: job.id },
-                        )))
-                        .await
-                    {
-                        Ok(Some(task)) => {
-                            let result = match &task {
-                                Task::CalendarAlarmEmail(task) => {
-                                    server.send_email_alarm(task, server_instance.clone()).await
-                                }
-                                Task::CalendarAlarmNotification(task) => {
-                                    server.send_display_alarm(task).await
-                                }
-                                Task::CalendarItipMessage(task) => {
-                                    server.send_imip(task, server_instance.clone()).await
-                                }
-                                Task::MergeThreads(task) => server.merge_threads(task).await,
-                                Task::DmarcReport(task) => {
-                                    server
-                                        .submit_report(report::ReportId::Dmarc(task.report_id.id()))
-                                        .await
-                                }
-                                Task::TlsReport(task) => {
-                                    server
-                                        .submit_report(report::ReportId::Tls(task.report_id.id()))
-                                        .await
-                                }
-                                Task::RestoreArchivedItem(task) => server.restore_item(task).await,
-                                Task::DestroyAccount(task) => server.destroy_account(task).await,
-                                Task::AccountMaintenance(task) => {
-                                    server.account_maintenance(task).await
-                                }
-                                Task::TenantMaintenance(task) => {
-                                    server.tenant_maintenance(task).await
-                                }
-                                Task::StoreMaintenance(task) => {
-                                    server.store_maintenance(task).await
-                                }
-                                Task::SpamFilterMaintenance(task) => {
-                                    Box::pin(server.spam_filter_maintenance(task)).await
-                                }
-                                Task::AcmeRenewal(task) => server.acme_management(task).await,
-                                Task::DkimManagement(task_dkim_rotation) => {
-                                    server.dkim_management(task_dkim_rotation).await
-                                }
-                                Task::DnsManagement(task_dns_management) => {
-                                    server.dns_management(task_dns_management).await
-                                }
-                                Task::IndexDocument(_)
-                                | Task::UnindexDocument(_)
-                                | Task::IndexTrace(_) => unreachable!(),
-                            };
+                    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();
+                            let server_instance = server_instance.clone();
+                            tokio::spawn(async move {
+                                let result = run_task(&server, &task, server_instance).await;
+                                (task, result)
+                            })
+                        };
+                        match run.await {
+                            Ok((task, result)) => {
+                                refresh_queue = result.is_retry();
 
-                            refresh_queue = result.is_retry();
-
-                            update_tasks(
-                                &server,
-                                &mut [TaskDetails { task, info: job }],
-                                vec![result],
-                            )
-                            .await;
-                        }
-                        Ok(None) => {
-                            trc::event!(
-                                TaskManager(TaskManagerEvent::TaskIgnored),
-                                Id = job.id,
-                                Reason = "Task not found in store, likely already processed.",
-                            );
-                        }
-                        Err(err) => {
-                            trc::error!(
-                                err.id(job.id)
-                                    .details("Failed to retrieve task details.")
-                                    .caused_by(trc::location!())
-                            );
+                                update_tasks(
+                                    &server,
+                                    &mut [TaskDetails { task, info }],
+                                    vec![result],
+                                )
+                                .await;
+                            }
+                            Err(err) => {
+                                worker_failed(&server, &[info.id], err).await;
+                                refresh_queue = true;
+                            }
                         }
                     }
 
@@ -318,6 +243,13 @@ pub(crate) trait TaskQueueManager: Sync + Send {
 
 impl TaskQueueManager for Server {
     async fn process_tasks(&self, ipc: &mut TaskManagerIpc) -> Duration {
+        // inbuxa: a node that is stopping has released its locks and claims
+        // nothing new
+        let task_locks = &self.inner.ipc.task_locks;
+        if task_locks.is_stopping() {
+            return Duration::from_secs(QUEUE_REFRESH_INTERVAL);
+        }
+        let lock_expiry = task_locks.expiry();
         let now_timestamp = now();
         let from_key = ValueKey::<ValueClass> {
             account_id: 0,
@@ -393,9 +325,7 @@ impl TaskQueueManager for Server {
                                     let locked = entry.get_mut();
                                     if locked.expires <= now || locked.due < task_due {
                                         locked.expires = Instant::now()
-                                            + std::time::Duration::from_secs(
-                                                DEFAULT_LOCK_EXPIRY + 1,
-                                            );
+                                            + std::time::Duration::from_secs(lock_expiry + 1);
                                         locked.due = task_due;
                                         tasks.push((
                                             TaskJob {
@@ -411,9 +341,7 @@ impl TaskQueueManager for Server {
                                 Entry::Vacant(entry) => {
                                     entry.insert(Locked {
                                         expires: Instant::now()
-                                            + std::time::Duration::from_secs(
-                                                DEFAULT_LOCK_EXPIRY + 1,
-                                            ),
+                                            + std::time::Duration::from_secs(lock_expiry + 1),
                                         due: task_due,
                                         revision: ipc.revision,
                                     });
@@ -464,12 +392,26 @@ impl TaskQueueManager for Server {
             let tx = &ipc.txs[task_type_idx as usize];
 
             if tx.capacity() > 0 {
-                if self.try_lock_task(task_job.id).await && tx.send(task_job).await.is_err() {
+                let id = task_job.id;
+                if !self.try_lock_task(id).await {
+                    // inbuxa: another node holds the task. Look again after a
+                    // short while rather than a full lock lifetime from now:
+                    // the holder may have claimed it after this scan began,
+                    // or run on a clock ahead of this one, and waiting the
+                    // whole lifetime again would leave the task stuck for
+                    // another hour past its lock if that holder died
+                    if let Some(locked) = ipc.locked.get_mut(&id) {
+                        locked.expires =
+                            Instant::now() + Duration::from_secs(claim_recheck_interval(lock_expiry));
+                    }
+                } else if tx.send(task_job).await.is_err() {
                     trc::event!(
                         Server(trc::ServerEvent::ThreadError),
                         Details = "Error sending task.",
                         CausedBy = trc::location!()
                     );
+                    // inbuxa: nothing will run it here, so don't hold it
+                    self.remove_index_lock(id).await;
                 }
             } else {
                 // If the channel is full, release the lock so it can be picked up in the next iteration
@@ -481,9 +423,117 @@ impl TaskQueueManager for Server {
         let now = Instant::now();
         ipc.locked
             .retain(|_, locked| locked.expires > now && locked.revision == ipc.revision);
-        Duration::from_secs(next_event.map_or(QUEUE_REFRESH_INTERVAL, |timestamp| {
+        let sleep_for = Duration::from_secs(next_event.map_or(QUEUE_REFRESH_INTERVAL, |timestamp| {
             timestamp.saturating_sub(store::write::now())
-        }))
+        }));
+
+        // inbuxa: wake up when a claim held elsewhere is due to be tried
+        // again, rather than only on the next task or refresh
+        ipc.locked
+            .values()
+            .map(|locked| locked.expires.saturating_duration_since(now))
+            .min()
+            .map_or(sleep_for, |recheck| sleep_for.min(recheck.max(Duration::from_secs(1))))
+    }
+}
+
+async fn run_task(
+    server: &Server,
+    task: &Task,
+    server_instance: Arc<ServerInstance>,
+) -> TaskResult {
+    match task {
+        Task::CalendarAlarmEmail(task) => {
+            server.send_email_alarm(task, server_instance.clone()).await
+        }
+        Task::CalendarAlarmNotification(task) => {
+            server.send_display_alarm(task).await
+        }
+        Task::CalendarItipMessage(task) => {
+            server.send_imip(task, server_instance.clone()).await
+        }
+        Task::MergeThreads(task) => server.merge_threads(task).await,
+        Task::DmarcReport(task) => {
+            server
+                .submit_report(report::ReportId::Dmarc(task.report_id.id()))
+                .await
+        }
+        Task::TlsReport(task) => {
+            server
+                .submit_report(report::ReportId::Tls(task.report_id.id()))
+                .await
+        }
+        Task::RestoreArchivedItem(task) => server.restore_item(task).await,
+        Task::DestroyAccount(task) => server.destroy_account(task).await,
+        Task::AccountMaintenance(task) => {
+            server.account_maintenance(task).await
+        }
+        Task::TenantMaintenance(task) => {
+            server.tenant_maintenance(task).await
+        }
+        Task::StoreMaintenance(task) => {
+            server.store_maintenance(task).await
+        }
+        Task::SpamFilterMaintenance(task) => {
+            Box::pin(server.spam_filter_maintenance(task)).await
+        }
+        Task::AcmeRenewal(task) => server.acme_management(task).await,
+        Task::DkimManagement(task_dkim_rotation) => {
+            server.dkim_management(task_dkim_rotation).await
+        }
+        Task::DnsManagement(task_dns_management) => {
+            server.dns_management(task_dns_management).await
+        }
+        Task::IndexDocument(_)
+        | Task::UnindexDocument(_)
+        | Task::IndexTrace(_) => unreachable!(),
+    }
+}
+
+/// 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.
+async fn fetch_task(server: &Server, job: TaskJob) -> Option<TaskDetails> {
+    match server
+        .store()
+        .get_value::<Task>(ValueKey::from(ValueClass::TaskQueue(TaskQueueClass::Task {
+            id: job.id,
+        })))
+        .await
+    {
+        Ok(Some(task)) => Some(TaskDetails { task, info: job }),
+        Ok(None) => {
+            trc::event!(
+                TaskManager(TaskManagerEvent::TaskIgnored),
+                Id = job.id,
+                Reason = "Task not found in store, likely already processed.",
+            );
+            server.remove_index_lock(job.id).await;
+            None
+        }
+        Err(err) => {
+            trc::error!(
+                err.id(job.id)
+                    .details("Failed to retrieve task details.")
+                    .caused_by(trc::location!())
+            );
+            server.remove_index_lock(job.id).await;
+            None
+        }
+    }
+}
+
+/// inbuxa: a task panicked: its locks are released so it runs again, here or
+/// on another node, and the worker carries on.
+async fn worker_failed(server: &Server, ids: &[u64], err: tokio::task::JoinError) {
+    trc::event!(
+        Server(trc::ServerEvent::ThreadError),
+        Details = "Task worker failed",
+        Reason = err.to_string(),
+        CausedBy = trc::location!()
+    );
+    for id in ids {
+        server.remove_index_lock(*id).await;
     }
 }
 
@@ -614,6 +664,13 @@ async fn update_tasks(
     }
 }
 
+/// inbuxa: how long to wait before trying again to claim a task another node
+/// holds: a twelfth of the lock lifetime, so five minutes for the one-hour
+/// lock, never more than that and never under a second.
+pub(crate) fn claim_recheck_interval(lock_expiry: u64) -> u64 {
+    (lock_expiry / 12).clamp(1, CLAIM_RECHECK_INTERVAL)
+}
+
 pub fn perpetual_retry_time(typ: TaskType, attempt: u64) -> Option<u64> {
     matches!(
         typ,

crates/services/src/task_manager/mod.rs

@@ -35,7 +35,9 @@ pub mod scheduler;
 pub mod spam_classifier;
 
 const QUEUE_REFRESH_INTERVAL: u64 = 60 * 5; // 5 minutes
-const DEFAULT_LOCK_EXPIRY: u64 = 60 * 60; // 1 hour
+// inbuxa: the lock lifetime (one hour) lives in common::ipc::TaskLocks, per
+// server, so a graceful stop can release the locks and the tests can shorten it
+const CLAIM_RECHECK_INTERVAL: u64 = 60 * 5; // 5 minutes
 
 pub(crate) struct TaskManagerIpc {
     txs: [mpsc::Sender<TaskJob>; TaskType::COUNT],

crates/store/src/backend/mysql/search.rs

@@ -2,6 +2,8 @@
  * 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 crate::{
@@ -19,7 +21,7 @@ use crate::{
     write::SearchIndex,
 };
 use mysql_async::{IsolationLevel, TxOpts, Value, prelude::Queryable};
-use nlp::tokenizers::word::WordTokenizer;
+use nlp::{language::Language, tokenizers::word::WordTokenizer};
 use std::fmt::Write;
 
 impl MysqlStore {
@@ -146,6 +148,20 @@ impl MysqlStore {
     }
 }
 
+// inbuxa: InnoDB's default full-text stopword list
+// (INFORMATION_SCHEMA.INNODB_FT_DEFAULT_STOPWORD) and innodb_ft_min_token_size
+// default; words outside these are not in a FULLTEXT index.
+const FT_STOPWORDS: &[&str] = &[
+    "a", "about", "an", "are", "as", "at", "be", "by", "com", "de", "en", "for", "from", "how",
+    "i", "in", "is", "it", "la", "of", "on", "or", "that", "the", "this", "to", "was", "what",
+    "when", "where", "who", "will", "with", "und", "www",
+];
+const FT_MIN_TOKEN_SIZE: usize = 3;
+
+fn is_ft_indexed(word: &str) -> bool {
+    word.chars().count() >= FT_MIN_TOKEN_SIZE && !FT_STOPWORDS.contains(&word)
+}
+
 fn build_filter(query: &mut String, filters: &[SearchFilter]) -> Vec<Value> {
     if filters.is_empty() {
         return Vec::new();
@@ -171,30 +187,77 @@ fn build_filter(query: &mut String, filters: &[SearchFilter]) -> Vec<Value> {
 
                 if field.is_text() && matches!(op, SearchOperator::Equal | SearchOperator::Contains)
                 {
-                    let (value, mode) = match (value, op) {
-                        (SearchValue::Text { value, .. }, SearchOperator::Equal) => {
-                            (Value::Bytes(format!("{value:?}").into_bytes()), "BOOLEAN")
-                        }
-                        (SearchValue::Text { value, .. }, ..) => {
+                    let (value, mode, unindexed) = match (value, op) {
+                        (SearchValue::Text { value, .. }, SearchOperator::Equal) => (
+                            Value::Bytes(format!("{value:?}").into_bytes()),
+                            "BOOLEAN",
+                            Vec::new(),
+                        ),
+                        (SearchValue::Text { value, language }, ..) => {
                             let mut text_query = String::with_capacity(value.len() + 1);
+                            let mut unindexed = Vec::new();
 
                             for item in WordTokenizer::new(value, MAX_TOKEN_LENGTH) {
-                                if !text_query.is_empty() {
-                                    text_query.push(' ');
+                                // inbuxa: InnoDB never indexes stopwords ("com",
+                                // "de", "www", ...) or words under
+                                // innodb_ft_min_token_size, and a required
+                                // (+word) term it has not indexed matches no row,
+                                // so "example.com" or "[email protected]" found
+                                // nothing. Such words are matched with a
+                                // word-boundary REGEXP instead.
+                                if is_ft_indexed(&item.word) {
+                                    if !text_query.is_empty() {
+                                        text_query.push(' ');
+                                    }
+                                    text_query.push('+');
+                                    text_query.push_str(&item.word);
+                                } else {
+                                    unindexed.push(item.word);
                                 }
-                                text_query.push('+');
-                                text_query.push_str(&item.word);
                             }
 
-                            (Value::Bytes(text_query.into_bytes()), "BOOLEAN")
+                            // For language text (bodies, subjects) the unindexed
+                            // words are noise words and only checked when nothing
+                            // else is left to match; keyword text (addresses,
+                            // contact fields) checks every word, as the other
+                            // backends do.
+                            if !text_query.is_empty() && !matches!(language, Language::None) {
+                                unindexed.clear();
+                            }
+
+                            (Value::Bytes(text_query.into_bytes()), "BOOLEAN", unindexed)
                         }
                         _ => {
                             debug_assert!(false, "Invalid search value for text field");
                             continue;
                         }
                     };
-                    let _ = write!(query, "MATCH({}) AGAINST(? IN {mode} MODE)", field.column());
-                    values.push(value);
+                    if unindexed.is_empty() {
+                        let _ =
+                            write!(query, "MATCH({}) AGAINST(? IN {mode} MODE)", field.column());
+                        values.push(value);
+                    } else {
+                        query.push('(');
+                        let is_empty = matches!(&value, Value::Bytes(v) if v.is_empty());
+                        if !is_empty {
+                            let _ = write!(
+                                query,
+                                "MATCH({}) AGAINST(? IN {mode} MODE) AND ",
+                                field.column()
+                            );
+                            values.push(value);
+                        }
+                        for (i, word) in unindexed.iter().enumerate() {
+                            if i > 0 {
+                                query.push_str(" AND ");
+                            }
+                            let _ = write!(query, "{} REGEXP ?", field.column());
+                            values.push(Value::Bytes(
+                                format!("(^|[^[:alnum:]]){word}([^[:alnum:]]|$)").into_bytes(),
+                            ));
+                        }
+                        query.push(')');
+                    }
                 } else if let SearchValue::KeyValues(kv) = value {
                     let (key, value) = kv.iter().next().unwrap();
 

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

@@ -2,12 +2,17 @@
  * 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 crate::{
-    backend::postgres::{
-        DELETE_CHUNK_SIZE, MIN_DELETE_CHUNK_SIZE, PostgresStore, PsqlSearchField, into_error,
-        into_pool_error, is_timeout_error,
+    backend::{
+        MAX_TOKEN_LENGTH,
+        postgres::{
+            DELETE_CHUNK_SIZE, MIN_DELETE_CHUNK_SIZE, PostgresStore, PsqlSearchField, into_error,
+            into_pool_error, is_timeout_error,
+        },
     },
     search::{
         IndexDocument, SearchComparator, SearchDocumentId, SearchFilter, SearchOperator,
@@ -15,7 +20,7 @@ use crate::{
     },
     write::SearchIndex,
 };
-use nlp::language::Language;
+use nlp::{language::Language, tokenizers::space::SpaceTokenizer};
 use std::fmt::Write;
 use tokio_postgres::{
     IsolationLevel,
@@ -43,6 +48,19 @@ impl PostgresStore {
             let primary_keys = index.primary_keys();
             let all_fields = index.all_fields();
             let fields = document.fields;
+            // inbuxa: keyword text (addresses, contact fields, ...) is split into
+            // words before it reaches the text parser, see keyword_terms().
+            let keywords = primary_keys
+                .iter()
+                .chain(all_fields)
+                .map(|field| match fields.get(field) {
+                    Some(SearchValue::Text {
+                        value,
+                        language: Language::None,
+                    }) if field.is_text() => Some(keyword_terms(value)),
+                    _ => None,
+                })
+                .collect::<Vec<_>>();
             let mut values = Vec::with_capacity(fields.len() + 2);
             let mut query = format!("INSERT INTO {} (", index.psql_table());
 
@@ -74,7 +92,20 @@ impl PostgresStore {
                         (0, PG_UNSTEMMED_LANG)
                     };
 
-                    if field.is_text() {
+                    if let Some(keywords) = &keywords[i] {
+                        let _ = write!(&mut query, "to_tsvector('{language}',{value_ref})");
+                        values.push(keywords as &(dyn ToSql + Sync));
+                        if field.sort_column().is_some() {
+                            let value_ref = format!("${}", values.len() + 1);
+                            if text_len > 255 {
+                                let _ = write!(&mut query, ",left({value_ref},255)");
+                            } else {
+                                let _ = write!(&mut query, ",{value_ref}");
+                            }
+                            values.push(value as &(dyn ToSql + Sync));
+                        }
+                        continue;
+                    } else if field.is_text() {
                         let _ = write!(&mut query, "to_tsvector('{language}',{value_ref})");
                     } else if text_len > 512 {
                         query.push_str("left(");
@@ -134,6 +165,7 @@ impl PostgresStore {
     ) -> trc::Result<Vec<R>> {
         let mut query = format!("SELECT {} FROM {}", R::field().column(), index.psql_table());
         let params = self.build_filter(&mut query, filters);
+        let params = params.iter().map(SqlParam::as_sql).collect::<Vec<_>>();
         if !sort.is_empty() {
             build_sort(&mut query, sort);
         }
@@ -155,6 +187,7 @@ impl PostgresStore {
         let table = filter.index.psql_table();
         let mut where_clause = String::new();
         let params = self.build_filter(&mut where_clause, &filter.filters);
+        let params = params.iter().map(SqlParam::as_sql).collect::<Vec<_>>();
         let conn = self.conn_pool.get().await.map_err(into_pool_error)?;
         let s = conn
             .prepare_cached(&format!("DELETE FROM {table}{where_clause}"))
@@ -196,7 +229,7 @@ impl PostgresStore {
         &self,
         query: &mut String,
         filters: &'x [SearchFilter],
-    ) -> Vec<&'x (dyn ToSql + Sync)> {
+    ) -> Vec<SqlParam<'x>> {
         if filters.is_empty() {
             return Vec::new();
         }
@@ -237,6 +270,10 @@ impl PostgresStore {
 
                         if matches!(language, Language::None) {
                             let _ = write!(query, "@@ {method}('{config}', ${value_pos})");
+                            if let SearchValue::Text { value, .. } = value {
+                                values.push(SqlParam::Owned(keyword_terms(value)));
+                                continue;
+                            }
                         } else {
                             let _ = write!(query, "@@ ({method}('{config}', ${value_pos})");
                             for fallback in [PG_FALLBACK_LANG, PG_UNSTEMMED_LANG] {
@@ -247,18 +284,18 @@ impl PostgresStore {
                             }
                             query.push(')');
                         }
-                        values.push(value as &(dyn ToSql + Sync));
+                        values.push(SqlParam::Ref(value));
                     } else if let SearchValue::KeyValues(kv) = value {
                         query.push_str(field.column());
                         query.push(' ');
 
                         let (key, value) = kv.iter().next().unwrap();
-                        values.push(key as &(dyn ToSql + Sync));
+                        values.push(SqlParam::Ref(key));
 
                         if !value.is_empty() {
                             let _ = write!(query, "->> ${value_pos} ");
                             op.write_pqsql(query, values.len() + 1);
-                            values.push(value as &(dyn ToSql + Sync));
+                            values.push(SqlParam::Ref(value));
                         } else {
                             let _ = write!(query, " ? ${value_pos}");
                         }
@@ -267,7 +304,7 @@ impl PostgresStore {
                         query.push(' ');
 
                         op.write_pqsql(query, value_pos);
-                        values.push(value as &(dyn ToSql + Sync));
+                        values.push(SqlParam::Ref(value));
                     }
                 }
                 SearchFilter::And | SearchFilter::Or => {
@@ -321,6 +358,38 @@ impl PostgresStore {
     }
 }
 
+// inbuxa: PostgreSQL's text parser keeps "[email protected]" (and host names,
+// URLs, file paths, ...) as a single token, so a search for "user" or
+// "example.com" never matched an address. Keyword text is split into words the
+// same way the built-in index splits it (SpaceTokenizer: lowercase runs of
+// alphanumerics) on both the indexing and the query side, so a full address,
+// its local part, its domain and the display-name words all match, as they do
+// on the other backends.
+pub(crate) fn keyword_terms(value: &str) -> String {
+    let mut terms = String::with_capacity(value.len());
+    for token in SpaceTokenizer::new(value, MAX_TOKEN_LENGTH) {
+        if !terms.is_empty() {
+            terms.push(' ');
+        }
+        terms.push_str(&token);
+    }
+    terms
+}
+
+pub(super) enum SqlParam<'x> {
+    Ref(&'x (dyn ToSql + Sync)),
+    Owned(String),
+}
+
+impl SqlParam<'_> {
+    fn as_sql(&self) -> &(dyn ToSql + Sync) {
+        match self {
+            SqlParam::Ref(value) => *value,
+            SqlParam::Owned(value) => value,
+        }
+    }
+}
+
 fn build_sort(query: &mut String, sort: &[SearchComparator]) {
     query.push_str(" ORDER BY ");
     for (i, comparator) in sort.iter().enumerate() {

tests/src/store/mod.rs

@@ -21,6 +21,7 @@ pub mod replica_cluster; // inbuxa: read replicas across nodes
 pub mod scaleout; // inbuxa: scale-out storage
 #[cfg(any(feature = "postgres", feature = "mysql"))]
 pub mod sql_timeout;
+pub mod task_locks; // inbuxa: task locks across nodes
 
 use crate::utils::server::TestServerBuilder;
 use std::io::Read;

tests/src/store/query.rs

@@ -128,6 +128,11 @@ pub async fn test(test: &TestServer) {
     println!("Running trace document tests...");
     test_trace_documents(store.clone()).await;
 
+    // inbuxa: address fields match by full address, local part, domain and
+    // display name on every backend
+    println!("Running address search tests...");
+    test_address_search(store.clone()).await;
+
     // Large document insert test
     println!("Running large document insert tests...");
     let mut large_text = String::with_capacity(20 * 1024 * 1024);
@@ -972,3 +977,155 @@ async fn test_trace_documents(store: SearchStore) {
             .unwrap();
     }
 }
+
+// inbuxa: the message indexer passes each display name and each address of
+// From/To/Cc/Bcc as keyword text (Language::None). The built-in index splits
+// that text into words, so an address is found by its full form, its local
+// part, its domain or a display-name word; PostgreSQL kept the whole address
+// as one token and MySQL dropped stopwords such as "com" and words under three
+// characters. The expected results below are the built-in (RocksDB/SQLite)
+// results and must be the same on every backend.
+async fn test_address_search(store: SearchStore) {
+    const ACCOUNT_ID: u32 = 7;
+    let messages: [[&[(&str, &str)]; 4]; 5] = [
+        // From, To, Cc, Bcc
+        [
+            &[("Amazon.com", "[email protected]")],
+            &[("Jane Doe", "[email protected]")],
+            &[],
+            &[],
+        ],
+        [
+            &[("", "[email protected]")],
+            &[("", "[email protected]")],
+            &[("Jane Doe", "[email protected]")],
+            &[],
+        ],
+        [
+            &[("GitHub", "[email protected]")],
+            &[("Jo Li", "[email protected]")],
+            &[],
+            &[("Audit", "[email protected]")],
+        ],
+        [
+            &[("Jane Doe", "[email protected]")],
+            &[("Amazon Web Services", "[email protected]")],
+            &[("Bob", "[email protected]")],
+            &[("", "[email protected]")],
+        ],
+        [
+            &[("Newsletter", "[email protected]")],
+            &[("", "[email protected]")],
+            &[],
+            &[],
+        ],
+    ];
+    let fields = [
+        EmailSearchField::From,
+        EmailSearchField::To,
+        EmailSearchField::Cc,
+        EmailSearchField::Bcc,
+    ];
+
+    let mut documents = Vec::new();
+    let mut mask = RoaringBitmap::new();
+    for (document_id, message) in messages.iter().enumerate() {
+        let mut document = IndexDocument::new(SearchIndex::Email)
+            .with_account_id(ACCOUNT_ID)
+            .with_document_id(document_id as u32);
+        for (field, addresses) in fields.iter().zip(message.iter()) {
+            for (name, address) in addresses.iter() {
+                if !name.is_empty() {
+                    document.index_text(field.clone(), name, Language::None);
+                }
+                document.index_text(field.clone(), address, Language::None);
+            }
+        }
+        document.index_unsigned(EmailSearchField::ReceivedAt, document_id as u64);
+        documents.push(document);
+        mask.insert(document_id as u32);
+    }
+    store.index(documents).await.unwrap();
+    if let SearchStore::ElasticSearch(store) = &store {
+        store.refresh_index(SearchIndex::Email).await.unwrap();
+    }
+
+    for (field, text, expected) in [
+        // full address
+        (EmailSearchField::From, "[email protected]", vec![0u32]),
+        (EmailSearchField::To, "[email protected]", vec![0]),
+        (EmailSearchField::Cc, "[email protected]", vec![1]),
+        (EmailSearchField::Bcc, "[email protected]", vec![3]),
+        (EmailSearchField::To, "[email protected]", vec![1, 2]),
+        // local part
+        (EmailSearchField::From, "noreply", vec![0, 2]),
+        (EmailSearchField::To, "jo", vec![1, 2]),
+        (EmailSearchField::Cc, "bob", vec![3]),
+        (EmailSearchField::Bcc, "audit", vec![2]),
+        // domain
+        (EmailSearchField::From, "amazon.com", vec![0, 1]),
+        (EmailSearchField::From, "amazon", vec![0, 1]),
+        (EmailSearchField::To, "example.org", vec![0, 4]),
+        (EmailSearchField::To, "io.de", vec![1, 2]),
+        (EmailSearchField::Cc, "example.net", vec![3]),
+        (EmailSearchField::Bcc, "example.org", vec![2]),
+        (EmailSearchField::From, "www.example.com", vec![4]),
+        (EmailSearchField::From, "com", vec![0, 1, 2, 4]),
+        // display name
+        (EmailSearchField::From, "Jane", vec![3]),
+        (EmailSearchField::From, "jane doe", vec![3]),
+        (EmailSearchField::To, "Web Services", vec![3]),
+        (EmailSearchField::To, "Li", vec![2]),
+        (EmailSearchField::Cc, "Doe", vec![1]),
+        (EmailSearchField::Bcc, "Audit", vec![2]),
+        // hyphenated local part
+        (EmailSearchField::From, "shipment-tracking", vec![1]),
+        (EmailSearchField::From, "tracking", vec![1]),
+        // no match
+        (EmailSearchField::From, "amazon.org", vec![]),
+        (EmailSearchField::To, "noreply", vec![]),
+        (EmailSearchField::Bcc, "jane", vec![]),
+    ] {
+        let ids = store
+            .query_account(
+                SearchQuery::new(SearchIndex::Email)
+                    .with_filters(vec![
+                        SearchFilter::eq(SearchField::AccountId, ACCOUNT_ID),
+                        SearchFilter::has_keyword(field.clone(), text),
+                    ])
+                    .with_comparator(SearchComparator::ascending(EmailSearchField::ReceivedAt))
+                    .with_mask(mask.clone()),
+            )
+            .await
+            .unwrap();
+        assert_eq!(ids, expected, "{field:?} {text:?}");
+    }
+
+    // TEXT-style search across all address fields
+    let ids = store
+        .query_account(
+            SearchQuery::new(SearchIndex::Email)
+                .with_filters(vec![
+                    SearchFilter::eq(SearchField::AccountId, ACCOUNT_ID),
+                    SearchFilter::Or,
+                    SearchFilter::has_keyword(EmailSearchField::From, "example.org"),
+                    SearchFilter::has_keyword(EmailSearchField::To, "example.org"),
+                    SearchFilter::has_keyword(EmailSearchField::Cc, "example.org"),
+                    SearchFilter::has_keyword(EmailSearchField::Bcc, "example.org"),
+                    SearchFilter::End,
+                ])
+                .with_comparator(SearchComparator::ascending(EmailSearchField::ReceivedAt))
+                .with_mask(mask.clone()),
+        )
+        .await
+        .unwrap();
+    assert_eq!(ids, vec![0, 1, 2, 3, 4]);
+
+    store
+        .unindex(
+            SearchQuery::new(SearchIndex::Email)
+                .with_filter(SearchFilter::eq(SearchField::AccountId, ACCOUNT_ID)),
+        )
+        .await
+        .unwrap();
+}

tests/src/store/task_locks.rs

@@ -0,0 +1,184 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! Task locks across nodes: tasks claimed by a node that then disappears
+//! run elsewhere once its locks expire, and a node that stops gracefully
+//! hands its locks back at once. The other node is played by writing its
+//! locks straight into the shared in-memory store, as a node that claimed
+//! the tasks and died leaves them.
+
+use crate::utils::server::TestServerBuilder;
+use common::{KV_LOCK_TASK, Server};
+use registry::schema::{
+    enums::IndexDocumentType,
+    structs::{Task, TaskIndexDocument, TaskStatus},
+};
+use services::task_manager::lock::{TaskLockManager, release_task_locks};
+use std::time::{Duration, Instant};
+use store::{
+    ValueKey,
+    write::{BatchBuilder, TaskQueueClass, ValueClass},
+};
+use utils::snowflake::SnowflakeIdGenerator;
+
+// Short enough for a test, long enough that the recheck interval (a twelfth
+// of it) is well below it
+const LOCK_EXPIRY: u64 = 12;
+
+#[tokio::test(flavor = "multi_thread")]
+pub async fn task_lock_tests() {
+    let test = TestServerBuilder::new("task_lock_tests")
+        .await
+        .build()
+        .await;
+    let server = test.server.clone();
+    println!(
+        "Running task lock tests on {}...",
+        std::env::var("STORE").unwrap_or_default()
+    );
+    server.inner.ipc.task_locks.set_expiry(LOCK_EXPIRY);
+
+    // 1. Another node claimed the tasks and died. Its locks block them until
+    // they expire; then this node runs them, without waiting for anything
+    // else to wake it
+    let ids = new_task_ids(4);
+    for id in &ids {
+        assert!(foreign_lock(&server, *id, LOCK_EXPIRY).await);
+    }
+    schedule(&server, &ids).await;
+    let started = Instant::now();
+    server.notify_task_queue();
+    tokio::time::sleep(Duration::from_secs(3)).await;
+    assert_eq!(pending(&server, &ids).await, ids.len(), "held by the other node");
+    wait_until_done(&server, &ids, Duration::from_secs(LOCK_EXPIRY + 10)).await;
+    let elapsed = started.elapsed();
+    assert!(
+        elapsed >= Duration::from_secs(LOCK_EXPIRY - 2),
+        "ran before the other node's locks expired: {elapsed:?}"
+    );
+
+    // 2. The other node's locks outlive what this node expects: claimed just
+    // after this node looked, or by a node whose clock runs ahead. This node
+    // keeps checking at the recheck interval, so the tasks run soon after
+    // those locks expire, not a whole lock lifetime later
+    let held_for = LOCK_EXPIRY + LOCK_EXPIRY / 2;
+    let ids = new_task_ids(4);
+    for id in &ids {
+        assert!(foreign_lock(&server, *id, held_for).await);
+    }
+    schedule(&server, &ids).await;
+    let started = Instant::now();
+    server.notify_task_queue();
+    wait_until_done(&server, &ids, Duration::from_secs(held_for + 8)).await;
+    let elapsed = started.elapsed();
+    assert!(
+        elapsed >= Duration::from_secs(held_for - 2),
+        "ran before the other node's locks expired: {elapsed:?}"
+    );
+
+    // 3. A graceful stop releases the locks this node holds: another node
+    // can claim those tasks at once, and this one claims nothing more
+    let ids = new_task_ids(3);
+    for id in &ids {
+        assert!(server.try_lock_task(*id).await, "claim {id}");
+    }
+    assert_eq!(server.inner.ipc.task_locks.held(), ids.len());
+    for id in &ids {
+        assert!(
+            !foreign_lock(&server, *id, LOCK_EXPIRY).await,
+            "held while this node runs"
+        );
+    }
+    assert_eq!(release_task_locks(&server).await, ids.len());
+    assert_eq!(server.inner.ipc.task_locks.held(), 0);
+    for id in &ids {
+        assert!(
+            foreign_lock(&server, *id, LOCK_EXPIRY).await,
+            "released on stop: {id}"
+        );
+    }
+    let [id] = new_task_ids(1)[..] else {
+        unreachable!()
+    };
+    assert!(!server.try_lock_task(id).await, "a stopping node claims nothing");
+
+    for id in ids {
+        let _ = server
+            .in_memory_store()
+            .remove_lock(KV_LOCK_TASK, &id.to_be_bytes())
+            .await;
+    }
+    if test.is_reset() {
+        test.temp_dir.delete();
+    }
+}
+
+fn new_task_ids(count: usize) -> Vec<u64> {
+    (0..count)
+        .map(|_| SnowflakeIdGenerator::global_id().unwrap())
+        .collect()
+}
+
+/// The other node's claim on a task, as its task manager takes it.
+async fn foreign_lock(server: &Server, id: u64, seconds: u64) -> bool {
+    server
+        .in_memory_store()
+        .try_lock(KV_LOCK_TASK, &id.to_be_bytes(), seconds)
+        .await
+        .unwrap()
+}
+
+/// Unindex tasks for files that don't exist: files aren't search-indexed and
+/// there is no undelete note, so running one only drops it from the queue.
+async fn schedule(server: &Server, ids: &[u64]) {
+    let mut batch = BatchBuilder::new();
+    for (n, id) in ids.iter().enumerate() {
+        batch.schedule_task_with_id(
+            *id,
+            Task::UnindexDocument(TaskIndexDocument {
+                account_id: 0u32.into(),
+                document_id: (u32::MAX - n as u32).into(),
+                document_type: IndexDocumentType::File,
+                status: TaskStatus::now(),
+            }),
+        );
+    }
+    server.store().write(batch.build_all()).await.unwrap();
+}
+
+async fn pending(server: &Server, ids: &[u64]) -> usize {
+    let mut count = 0;
+    for id in ids {
+        if server
+            .store()
+            .get_value::<Task>(ValueKey::from(ValueClass::TaskQueue(
+                TaskQueueClass::Task { id: *id },
+            )))
+            .await
+            .unwrap()
+            .is_some()
+        {
+            count += 1;
+        }
+    }
+    count
+}
+
+async fn wait_until_done(server: &Server, ids: &[u64], within: Duration) {
+    let started = Instant::now();
+    loop {
+        let left = pending(server, ids).await;
+        if left == 0 {
+            return;
+        }
+        assert!(
+            started.elapsed() < within,
+            "{left} task(s) still pending after {:?}",
+            started.elapsed()
+        );
+        tokio::time::sleep(Duration::from_millis(250)).await;
+    }
+}