Merge pull request 'Allowed IPs take the full settings reload after a write' (#42) from fix/allowed-ip-reload into main

write_reload_target sent AllowedIp writes to the blocked-IP reload, but
that reload rebuilds only BlockedIps. Allowed IPs are parsed into the
core's security settings (Security::parse), which only a full reload
rebuilds, so an AllowedIp write reported x:settingsReload applied: true
while the change wasn't live until the next full reload.

AllowedIp now maps to the full reload, like the other settings objects;
BlockedIp keeps its targeted reload.

system::auto_reload::settings_reload_tests now creates an allowed IP
over JMAP and checks that is_ip_allowed sees it with no ReloadSettings,
and that destroying it takes it out again. On main it fails ("allowed
IP not in the running settings").

.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
+# The image is multi-arch (linux/amd64, linux/arm64), built by two jobs on
-# one buildx run on host1 instead of one native runner per architecture: the
+# the image-build runner rather than one buildx run for both. The Dockerfile's
-# Dockerfile's builder stage runs on the build platform and cross-compiles
+# builder stage runs on the build platform and cross-compiles with an aarch64
-# with an aarch64 linker, so only the small final stage (apt, setcap) goes
+# linker, so only the small final stage (apt, setcap) goes through QEMU for
-# through QEMU for arm64. No digest-joining job is needed.
+# 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
+      # Attestations off, as before: they add manifests of their own, and the
-      # index, and the index should hold the two images and nothing else.
+      # 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/cache/reload.rs

@@ -13,20 +13,27 @@ use crate::{
         storage::Storage,
         telemetry::Telemetry,
     },
-    ipc::{QueueEvent, RegistryChange},
+    ipc::{BroadcastEvent, QueueEvent, RegistryChange},
     network::security::{BlockedIps, IpWithTtl},
 };
 use ahash::AHashMap;
 use directory::Directories;
 use registry::{
     schema::{prelude::ObjectType, structs::BlockedIp},
-    types::error::{Error, Warning},
+    types::{
+        error::{Error, Warning},
+        id::ObjectId,
+    },
 };
 use std::sync::Arc;
 use store::{LookupStores, registry::bootstrap::Bootstrap, write::now};
 
 pub struct ReloadResult {
+    /// Errors that kept the reload from being applied.
     pub errors: Vec<Error>,
+    /// inbuxa: errors in objects that already failed when the running
+    /// settings were built; logged, but they don't refuse a reload.
+    pub known_errors: Vec<Error>,
     pub warnings: Vec<Warning>,
     pub replaced_core: bool,
 }
@@ -114,42 +121,60 @@ impl Server {
                     directories: directory.directories,
                 };
 
-                // Parse tracers
+                // inbuxa: upstream swapped the core only when the whole build
+                // was free of errors, while boot runs with whatever built. So one
+                // object that failed (a DNS lookup that timed out, say) refused
+                // every later reload, cluster-wide when the reload came from
+                // ReloadSettings, and the running settings went stale. Now a
+                // reload is refused only for errors in objects that built when
+                // the running settings were built: those would be lost by
+                // applying it. Objects that already failed then are missing
+                // from the running settings anyway, as at boot, so their
+                // errors are reported but don't hold the reload back.
                 let tracers = Telemetry::parse(&mut bootstrap, &storage).await;
+                let core = Box::pin(Core::parse(&mut bootstrap, storage)).await;
+                let mut servers = Listeners::parse(&mut bootstrap).await;
 
-                if bootstrap.errors.is_empty() {
+                if !self.has_new_build_errors(&bootstrap.errors) {
-                    let core = Box::pin(Core::parse(&mut bootstrap, storage)).await;
+                    servers
+                        .parse_tcp_acceptors(&mut bootstrap, self.inner.clone())
+                        .await;
 
-                    if bootstrap.errors.is_empty() {
+                    if !self.has_new_build_errors(&bootstrap.errors) {
-                        let mut servers = Listeners::parse(&mut bootstrap).await;
+                        // Update core
-                        servers
+                        self.inner.shared_core.store(core.into());
-                            .parse_tcp_acceptors(&mut bootstrap, self.inner.clone())
-                            .await;
 
-                        if bootstrap.errors.is_empty() {
+                        // Update tracers
-                            // Update core
+                        tracers.update();
-                            self.inner.shared_core.store(core.into());
 
-                            // Update tracers
+                        // Reload queue settings
+                        self.inner
+                            .ipc
+                            .queue_tx
+                            .send(QueueEvent::ReloadSettings)
+                            .await
+                            .ok();
 
-                            tracers.update();
+                        self.record_build_errors(&bootstrap.errors);
 
-                            // Reload queue settings
+                        return Ok(ReloadResult {
-                            self.inner
+                            errors: Vec::new(),
-                                .ipc
+                            known_errors: bootstrap.errors,
-                                .queue_tx
+                            warnings: bootstrap.warnings,
-                                .send(QueueEvent::ReloadSettings)
+                            replaced_core: true,
-                                .await
+                        });
-                                .ok();
-
-                            return Ok(ReloadResult {
-                                errors: bootstrap.errors,
-                                warnings: bootstrap.warnings,
-                                replaced_core: true,
-                            });
-                        }
                     }
                 }
+
+                let (known_errors, errors) = std::mem::take(&mut bootstrap.errors)
+                    .into_iter()
+                    .partition(|error| self.is_known_build_error(error));
+                return Ok(ReloadResult {
+                    errors,
+                    known_errors,
+                    warnings: bootstrap.warnings,
+                    replaced_core: false,
+                });
             }
         }
 
@@ -163,7 +188,7 @@ impl ReloadResult {
     }
 
     pub fn log(&self) {
-        for error in &self.errors {
+        for error in self.errors.iter().chain(&self.known_errors) {
             error.log();
         }
         for warning in &self.warnings {
@@ -176,8 +201,237 @@ impl From<Bootstrap> for ReloadResult {
     fn from(bootstrap: Bootstrap) -> Self {
         Self {
             errors: bootstrap.errors,
+            known_errors: Vec::new(),
             warnings: bootstrap.warnings,
             replaced_core: false,
         }
     }
 }
+
+// inbuxa: which objects failed to build for the running settings
+impl Server {
+    /// Records the objects that failed to build for the settings now running.
+    pub fn record_build_errors(&self, errors: &[Error]) {
+        *self.inner.data.build_errors.lock() = errors.iter().filter_map(error_object).collect();
+    }
+
+    fn is_known_build_error(&self, error: &Error) -> bool {
+        error_object(error).is_some_and(|id| self.inner.data.build_errors.lock().contains(&id))
+    }
+
+    fn has_new_build_errors(&self, errors: &[Error]) -> bool {
+        errors.iter().any(|error| !self.is_known_build_error(error))
+    }
+}
+
+fn error_object(error: &Error) -> Option<ObjectId> {
+    match error {
+        Error::Validation { object_id, .. }
+        | Error::Build { object_id, .. }
+        | Error::NotFound { object_id } => Some(*object_id),
+        Error::Internal { object_id, .. } => *object_id,
+    }
+}
+
+// inbuxa: upstream applied a registry write to the running settings only on
+// an explicit x:Action ReloadSettings (Directory and Authentication aside), so
+// a new MtaDeliverySchedule, say, stayed unknown ("Queue strategy not found")
+// until someone reloaded. Writes to objects the settings are built from now
+// reload them, here and across the cluster, as ReloadSettings does.
+
+/// Coalesces the full reloads that registry writes trigger: a write waits for
+/// a reload that started after it was stored, and joins one if it can, so a
+/// burst of writes costs a reload or two rather than one each.
+#[derive(Default)]
+pub struct SettingsReloadGate {
+    requested: std::sync::atomic::AtomicU64,
+    state: tokio::sync::Mutex<SettingsReloadState>,
+}
+
+#[derive(Default)]
+struct SettingsReloadState {
+    completed: u64,
+    refused: Option<String>,
+}
+
+/// The reload a write to `object` calls for: the object to reload, or None
+/// when the running settings don't hold that object (accounts, domains and
+/// other data read as needed, stores, which take a restart, and objects with
+/// reload actions of their own, such as applications). Blocked IPs have a
+/// reload of their own; allowed IPs take the full one.
+pub fn write_reload_target(object: ObjectType) -> Option<ObjectType> {
+    match object {
+        ObjectType::Certificate => Some(ObjectType::Certificate),
+        ObjectType::MemoryLookupKey
+        | ObjectType::MemoryLookupKeyValue
+        | ObjectType::HttpLookup
+        | ObjectType::StoreLookup => Some(ObjectType::StoreLookup),
+        ObjectType::BlockedIp => Some(ObjectType::BlockedIp),
+        // Allowed IPs are part of the core's security settings
+        // (Security::parse), which only a full reload rebuilds; the blocked-IP
+        // reload doesn't touch them
+        ObjectType::AllowedIp
+        | ObjectType::AcmeProvider
+        | ObjectType::AddressBook
+        | ObjectType::AiModel
+        | ObjectType::Asn
+        | ObjectType::Authentication
+        | ObjectType::Cache
+        | ObjectType::Calendar
+        | ObjectType::CalendarAlarm
+        | ObjectType::CalendarScheduling
+        | ObjectType::ClusterRole
+        | ObjectType::DataRetention
+        | ObjectType::Directory
+        | ObjectType::DkimReportSettings
+        | ObjectType::DmarcReportSettings
+        | ObjectType::DnsResolver
+        | ObjectType::DsnReportSettings
+        | ObjectType::Email
+        | ObjectType::EventTracingLevel
+        | ObjectType::FileStorage
+        | ObjectType::Http
+        | ObjectType::HttpForm
+        | ObjectType::Imap
+        | ObjectType::Jmap
+        | ObjectType::Metrics
+        | ObjectType::MtaConnectionStrategy
+        | ObjectType::MtaDeliverySchedule
+        | ObjectType::MtaExtensions
+        | ObjectType::MtaHook
+        | ObjectType::MtaInboundSession
+        | ObjectType::MtaInboundThrottle
+        | ObjectType::MtaMilter
+        | ObjectType::MtaOutboundStrategy
+        | ObjectType::MtaOutboundThrottle
+        | ObjectType::MtaQueueQuota
+        | ObjectType::MtaRoute
+        | ObjectType::MtaStageAuth
+        | ObjectType::MtaStageConnect
+        | ObjectType::MtaStageData
+        | ObjectType::MtaStageEhlo
+        | ObjectType::MtaStageMail
+        | ObjectType::MtaStageRcpt
+        | ObjectType::MtaSts
+        | ObjectType::MtaTlsStrategy
+        | ObjectType::MtaVirtualQueue
+        | ObjectType::NetworkListener
+        | ObjectType::OidcProvider
+        | ObjectType::ReportSettings
+        | ObjectType::Search
+        | ObjectType::Security
+        | ObjectType::SenderAuth
+        | ObjectType::Sharing
+        | ObjectType::SieveSystemInterpreter
+        | ObjectType::SieveSystemScript
+        | ObjectType::SieveUserInterpreter
+        | ObjectType::SieveUserScript
+        | ObjectType::SpamClassifier
+        | ObjectType::SpamDnsblServer
+        | ObjectType::SpamDnsblSettings
+        | ObjectType::SpamFileExtension
+        | ObjectType::SpamPyzor
+        | ObjectType::SpamRule
+        | ObjectType::SpamSettings
+        | ObjectType::SpamTag
+        | ObjectType::SpfReportSettings
+        | ObjectType::SystemSettings
+        | ObjectType::TaskManager
+        | ObjectType::TlsReportSettings
+        | ObjectType::Tracer
+        | ObjectType::WebDav
+        | ObjectType::WebHook => Some(object),
+        _ => None,
+    }
+}
+
+impl Server {
+    /// Applies a stored registry write to `object` to the running settings,
+    /// and on success tells the other nodes to do the same. Returns None when
+    /// the write needs no reload, Some(Ok(())) when it was applied, and
+    /// Some(Err(reason)) when the reload was refused (the write stays stored;
+    /// ReloadSettings reports the same errors).
+    pub async fn reload_after_write(&self, object: ObjectType) -> Option<Result<(), String>> {
+        let target = write_reload_target(object)?;
+        let change = RegistryChange::Reload(target);
+
+        if matches!(
+            target,
+            ObjectType::Certificate | ObjectType::StoreLookup | ObjectType::BlockedIp
+        ) {
+            // Cheap, and limited to their own objects
+            let result = self.reload_and_broadcast(change).await;
+            return Some(result);
+        }
+
+        let gate = &self.inner.data.settings_reload;
+        let ticket = gate
+            .requested
+            .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
+            + 1;
+        let mut state = gate.state.lock().await;
+        if state.completed >= ticket {
+            // A reload that started after this write was stored has run
+            return Some(state.refused.clone().map_or(Ok(()), Err));
+        }
+        let covers = gate.requested.load(std::sync::atomic::Ordering::SeqCst);
+        let result = self.reload_and_broadcast(change).await;
+        state.completed = covers;
+        state.refused = result.clone().err();
+        Some(result)
+    }
+
+    async fn reload_and_broadcast(&self, change: RegistryChange) -> Result<(), String> {
+        match Box::pin(self.reload_registry(change)).await {
+            Ok(reload) if !reload.has_errors() => {
+                reload.log();
+                self.cluster_broadcast(BroadcastEvent::RegistryChange(change))
+                    .await;
+                Ok(())
+            }
+            Ok(reload) => {
+                reload.log();
+                let reason = describe_reload_errors(&reload.errors);
+                trc::event!(
+                    Registry(trc::RegistryEvent::BuildWarning),
+                    Details = "Settings didn't reload after a registry write",
+                    Reason = reason.clone(),
+                );
+                Err(reason)
+            }
+            Err(err) => {
+                let reason = err.to_string();
+                trc::error!(err.details("Failed to reload settings after a registry write"));
+                Err(reason)
+            }
+        }
+    }
+}
+
+/// inbuxa: a refused reload's errors in a sentence: the first one, naming its
+/// object, and how many more there are.
+pub fn describe_reload_errors(errors: &[Error]) -> String {
+    let mut description = match errors.first() {
+        Some(Error::Build { object_id, message }) => format!("{object_id}: {message}"),
+        Some(Error::Validation { object_id, errors }) => format!(
+            "{object_id}: {}",
+            errors
+                .iter()
+                .map(|err| err.to_string())
+                .collect::<Vec<_>>()
+                .join("; ")
+        ),
+        Some(Error::Internal {
+            object_id: Some(object_id),
+            error,
+        }) => format!("{object_id}: {error}"),
+        Some(Error::Internal { error, .. }) => error.to_string(),
+        Some(Error::NotFound { object_id }) => format!("{object_id} was not found"),
+        None => String::new(),
+    };
+    let more = errors.len().saturating_sub(1);
+    if more > 0 {
+        description.push_str(&format!(" ({more} more in the server log.)"));
+    }
+    description
+}

crates/common/src/config/inner.rs

@@ -93,9 +93,11 @@ impl Data {
             registry_id_gen: id_generator.clone(),
             span_id_gen: id_generator,
             queue_status: true.into(),
+            settings_reload: Default::default(),
             applications,
             logos: Default::default(),
             smtp_connectors: TlsConnectors::try_new().failed("Failed to build TLS connectors"),
+            build_errors: Default::default(),
             asn_geo_data: Default::default(),
         }
     }
@@ -234,9 +236,11 @@ impl Default for Data {
             span_id_gen: Default::default(),
             registry_id_gen: Default::default(),
             queue_status: true.into(),
+            settings_reload: Default::default(),
             applications: WebApplications::new(),
             logos: Default::default(),
             smtp_connectors: TlsConnectors::try_new().unwrap(),
+            build_errors: Default::default(),
             asn_geo_data: Default::default(),
             lookup_stores: Default::default(),
         }

crates/common/src/config/mailstore/spamfilter.rs

@@ -16,7 +16,6 @@ use mail_auth::common::resolver::ToReverseName;
 use nlp::classifier::model::{CcfhClassifier, FhClassifier};
 use registry::schema::{
     enums::{ExpressionVariable, ModelSize},
-    prelude::ObjectType,
     structs::{
         self, SpamDnsblServer, SpamDnsblSettings, SpamFileExtension, SpamPyzor, SpamRule,
         SpamSettings, SpamTag,
@@ -25,10 +24,10 @@ use registry::schema::{
 use sieve::SpamStatus;
 use std::{
     net::{IpAddr, SocketAddr},
-    time::Duration,
+    sync::Arc,
+    time::{Duration, Instant},
 };
 use store::registry::{RegistryObject, bootstrap::Bootstrap};
-use tokio::net::lookup_host;
 use utils::{cache::CacheItemWeight, glob::GlobMap};
 
 #[derive(rkyv::Archive, rkyv::Deserialize, rkyv::Serialize, Debug, Default)]
@@ -157,7 +156,11 @@ pub struct FtrlParameters {
 
 #[derive(Debug, Clone)]
 pub struct PyzorConfig {
-    pub address: SocketAddr,
+    // inbuxa: the server is resolved when a message is checked, not while the
+    // settings are built (see PyzorConfig::address)
+    pub host: String,
+    pub port: u16,
+    pub resolved: Arc<parking_lot::Mutex<Option<(SocketAddr, Instant)>>>,
     pub timeout: Duration,
     pub min_count: u64,
     pub min_wl_count: u64,
@@ -474,31 +477,15 @@ impl PyzorConfig {
             return None;
         }
 
-        let port = pyzor.port;
+        // inbuxa: upstream resolved the host here and reported a failed lookup
-        let host = pyzor.host;
+        // as a build error, so a DNS hiccup on one node refused every settings
-        let address = match lookup_host(format!("{host}:{port}"))
+        // reload on it (and, from the node that ran ReloadSettings, across the
-            .await
+        // cluster). The lookup now happens when a message is checked; a
-            .map(|mut a| a.next())
+        // failure there is logged as a Pyzor error for that message.
-        {
-            Ok(Some(address)) => address,
-            Ok(None) => {
-                bp.build_error(
-                    ObjectType::SpamPyzor.singleton(),
-                    "Invalid address: No addresses found.",
-                );
-                return None;
-            }
-            Err(err) => {
-                bp.build_error(
-                    ObjectType::SpamPyzor.singleton(),
-                    format!("Invalid address: {}", err),
-                );
-                return None;
-            }
-        };
-
         PyzorConfig {
-            address,
+            host: pyzor.host,
+            port: pyzor.port as u16,
+            resolved: Default::default(),
             timeout: pyzor.timeout.into_inner(),
             min_count: pyzor.block_count,
             min_wl_count: pyzor.allow_count,
@@ -508,6 +495,35 @@ impl PyzorConfig {
     }
 }
 
+// inbuxa: how long a resolved Pyzor address is reused
+const PYZOR_RESOLVE_TTL: Duration = Duration::from_secs(300);
+
+impl PyzorConfig {
+    /// The server's address: the host itself when it is an IP address,
+    /// otherwise the first address it resolves to, reused for five minutes.
+    pub async fn address(&self) -> std::io::Result<SocketAddr> {
+        if let Ok(ip) = self.host.parse::<IpAddr>() {
+            return Ok(SocketAddr::new(ip, self.port));
+        }
+        if let Some((address, resolved_at)) = *self.resolved.lock()
+            && resolved_at.elapsed() < PYZOR_RESOLVE_TTL
+        {
+            return Ok(address);
+        }
+        let address = tokio::net::lookup_host((self.host.as_str(), self.port))
+            .await?
+            .next()
+            .ok_or_else(|| {
+                std::io::Error::new(
+                    std::io::ErrorKind::NotFound,
+                    format!("{} has no addresses", self.host),
+                )
+            })?;
+        *self.resolved.lock() = Some((address, Instant::now()));
+        Ok(address)
+    }
+}
+
 impl ClassifierConfig {
     pub async fn parse(bp: &mut Bootstrap) -> Option<Self> {
         let classifier = bp.setting_infallible::<structs::SpamClassifier>().await;

crates/common/src/config/smtp/session.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 self::resolver::Policy;
@@ -22,7 +24,7 @@ use registry::schema::{
 };
 use smtp_proto::*;
 use std::{
-    net::{SocketAddr, ToSocketAddrs},
+    net::{IpAddr, SocketAddr},
     str::FromStr,
     time::Duration,
 };
@@ -384,19 +386,16 @@ impl SessionConfig {
                     Some(Milter {
                         enable: bp.compile_expr(id, &milter.ctx_enable()),
                         id,
-                        addrs: format!("{}:{}", milter.hostname, milter.port)
+                        // inbuxa: upstream resolved the hostname here (a
-                            .to_socket_addrs()
+                        // blocking lookup) and made a failure a build error,
-                            .map_err(|err| {
+                        // which refused the whole settings reload. An IP
-                                bp.build_error(
+                        // address is kept as is; a name is resolved on each
-                                    id,
+                        // connection (MilterClient::connect).
-                                    format!(
+                        addrs: milter
-                                        "Unable to resolve milter hostname {}: {}",
+                            .hostname
-                                        milter.hostname, err
+                            .parse::<IpAddr>()
-                                    ),
+                            .map(|ip| vec![SocketAddr::new(ip, milter.port as u16)])
-                                )
+                            .unwrap_or_default(),
-                            })
-                            .ok()?
-                            .collect(),
                         hostname: milter.hostname,
                         port: milter.port as u16,
                         timeout_connect: milter.timeout_connect.into_inner(),

crates/common/src/ipc.rs

@@ -345,8 +345,13 @@ pub struct TaskLocks {
 }
 
 impl TaskLocks {
-    /// How long a task lock lasts, in seconds, unless it is released first.
+    /// How long a task lock lasts, in seconds, unless it is released first
-    pub const DEFAULT_EXPIRY: u64 = 60 * 60;
+    /// or renewed. inbuxa: upstream held a lock for an hour, so a killed
+    /// node's tasks waited that long; the lock is now a five-minute lease
+    /// that the task manager renews every third of it while the task runs
+    /// (renew_task_locks), so a dead node's tasks run elsewhere within
+    /// minutes.
+    pub const DEFAULT_EXPIRY: u64 = 5 * 60;
 
     pub fn is_stopping(&self) -> bool {
         self.stopping.load(Ordering::Acquire)
@@ -370,6 +375,16 @@ impl TaskLocks {
         self.held.lock().len()
     }
 
+    /// inbuxa: the tasks this node holds, to renew their locks.
+    pub fn held_ids(&self) -> Vec<u64> {
+        self.held.lock().iter().copied().collect()
+    }
+
+    /// inbuxa: whether this node holds (and is running) the task.
+    pub fn is_held(&self, id: u64) -> bool {
+        self.held.lock().contains(&id)
+    }
+
     pub fn expiry(&self) -> u64 {
         self.expiry.load(Ordering::Relaxed)
     }

crates/common/src/lib.rs

@@ -161,11 +161,17 @@ pub struct Data {
     pub span_id_gen: SnowflakeIdGenerator,
     pub registry_id_gen: SnowflakeIdGenerator,
     pub queue_status: AtomicBool,
+    // inbuxa: coalesces the settings reloads registry writes trigger
+    pub settings_reload: cache::reload::SettingsReloadGate,
 
     pub applications: WebApplications,
     pub logos: Mutex<AHashMap<Box<str>, LogoCache>>,
 
     pub smtp_connectors: TlsConnectors,
+
+    // inbuxa: the objects that failed to build when the running settings
+    // were built, at boot or by the last applied reload (see reload_registry)
+    pub build_errors: Mutex<AHashSet<registry::types::id::ObjectId>>,
 }
 
 #[derive(Clone)]

crates/common/src/manager/boot.rs

@@ -240,6 +240,9 @@ impl BootManager {
                     .parse_tcp_acceptors(&mut bootstrap, inner.clone())
                     .await;
 
+                // inbuxa: a reload isn't refused over objects that failed here
+                inner.build_server().record_build_errors(&bootstrap.errors);
+
                 BootManager {
                     inner,
                     bootstrap,

crates/jmap-proto/src/method/set.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 super::ahash_is_empty;
@@ -71,6 +73,23 @@ pub struct SetResponse<T: JmapObject> {
     #[serde(rename = "notDestroyed")]
     #[serde(skip_serializing_if = "VecMap::is_empty")]
     pub not_destroyed: VecMap<MaybeInvalid<Id>, SetError<T::Property>>,
+
+    // inbuxa: on a registry write that changes the running settings, whether
+    // the server applied it
+    #[serde(rename = "x:settingsReload")]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub settings_reload: Option<SettingsReload>,
+}
+
+/// inbuxa: the settings reload that followed a registry write.
+#[derive(Debug, Clone, serde::Serialize)]
+pub struct SettingsReload {
+    /// The running settings (here and, through the cluster, on every node)
+    /// include the write.
+    pub applied: bool,
+    /// Why they don't, when they don't.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub description: Option<String>,
 }
 
 impl<'de, T: JmapObject> DeserializeArguments<'de> for SetRequest<'de, T> {
@@ -199,6 +218,7 @@ impl<T: JmapObject> SetResponse<T> {
                 not_created: VecMap::new(),
                 not_updated: VecMap::new(),
                 not_destroyed: VecMap::new(),
+                settings_reload: None,
             })
         } else {
             Err(trc::JmapEvent::RequestTooLarge.into_err())

crates/jmap/src/registry/mapping/action.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::registry::mapping::{RegistrySetResponse, map_bootstrap_error};
@@ -99,7 +101,7 @@ pub(crate) async fn action_set(
                 } else {
                     set.response
                         .not_created
-                        .append(id, map_bootstrap_error(result.errors));
+                        .append(id, reload_refused(result.errors));
                 }
             }
             Action::InvalidateCaches => {
@@ -573,3 +575,14 @@ async fn dmarc_troubleshoot(
 
     Some(request)
 }
+
+/// inbuxa: a refused reload names the object that stopped it and says the
+/// settings weren't applied; upstream passed on the first error's bare message
+/// ("Invalid address: ..."), which read like a problem with the request.
+fn reload_refused(errors: Vec<registry::types::error::Error>) -> SetError<Property> {
+    let description = format!(
+        "Settings were not reloaded. {}",
+        common::cache::reload::describe_reload_errors(&errors)
+    );
+    map_bootstrap_error(errors).with_description(description)
+}

crates/jmap/src/registry/set.rs

@@ -38,7 +38,7 @@ use directory::core::secret::{hash_secret, is_password_hash};
 use http_proto::HttpSessionData;
 use jmap_proto::{
     error::set::{SetError, SetErrorType},
-    method::set::{SetRequest, SetResponse},
+    method::set::{SetRequest, SetResponse, SettingsReload},
     object::registry::Registry,
     references::resolve::ResolveCreatedReference,
     request::{IntoValid, MaybeInvalid},
@@ -931,30 +931,28 @@ impl RegistrySet for Server {
             }
         };
 
-        // inbuxa: DIR-17: a directory or the server default applies on the
+        // inbuxa: a write to an object the running settings are built from
-        // next request, here and on every node
+        // applies at once, here and on every node (DIR-17 did this for
-        if matches!(
+        // directories and the server default; now it covers every such object)
-            object_type,
+        let mut result = result;
-            ObjectType::Directory | ObjectType::Authentication
+        if let Ok(response) = &mut result
-        ) && let Ok(response) = &result
             && (!response.created.is_empty()
                 || !response.updated.is_empty()
                 || !response.destroyed.is_empty())
+            && let Some(reload) = self.reload_after_write(object_type).await
         {
-            let change = common::ipc::RegistryChange::Reload(ObjectType::Directory);
+            response.settings_reload = Some(match reload {
-            match Box::pin(self.reload_registry(change)).await {
+                Ok(()) => SettingsReload {
-                Ok(reload) if !reload.has_errors() => {
+                    applied: true,
-                    self.cluster_broadcast(common::ipc::BroadcastEvent::RegistryChange(change))
+                    description: None,
-                        .await;
+                },
-                }
+                Err(reason) => SettingsReload {
-                Ok(_) => trc::event!(
+                    applied: false,
-                    Registry(trc::RegistryEvent::BuildWarning),
+                    description: Some(format!(
-                    Details = "Settings didn't reload after a directory change",
+                        "Saved, but the running settings were not reloaded. {reason}"
-                ),
+                    )),
-                Err(err) => {
+                },
-                    trc::error!(err.details("Failed to reload directories"));
+            });
-                }
-            }
         }
         result
     }

crates/services/src/task_manager/lock.rs

@@ -82,3 +82,42 @@ pub async fn release_task_locks(server: &Server) -> usize {
     }
     ids.len()
 }
+
+/// inbuxa: renews the lease on every task this node is running, so it stays
+/// claimed for as long as it runs while a node that dies loses its claims
+/// within one lock lifetime. Returns how many leases were renewed and how
+/// many were found lost (expired, perhaps taken by another node).
+pub async fn renew_task_locks(server: &Server) -> (usize, usize) {
+    let locks = &server.inner.ipc.task_locks;
+    let expiry = locks.expiry();
+    let (mut renewed, mut lost) = (0, 0);
+    for id in locks.held_ids() {
+        match server
+            .in_memory_store()
+            .renew_lock(KV_LOCK_TASK, &id.to_be_bytes(), expiry)
+            .await
+        {
+            Ok(true) => renewed += 1,
+            Ok(false) => {
+                // Still held here as far as this node knows; the task
+                // finishes and its lock is removed as usual
+                if locks.is_held(id) {
+                    lost += 1;
+                    trc::event!(
+                        TaskManager(TaskManagerEvent::TaskLocked),
+                        Id = id,
+                        Details = "Task lock expired while the task was running",
+                    );
+                }
+            }
+            Err(err) => {
+                trc::error!(
+                    err.details("Failed to renew task lock")
+                        .ctx(trc::Key::Id, id)
+                        .caused_by(trc::location!())
+                );
+            }
+        }
+    }
+    (renewed, lost)
+}

crates/services/src/task_manager/manager.rs

@@ -13,7 +13,7 @@ use crate::task_manager::dkim::DkimManagementTask;
 use crate::task_manager::dns::DnsManagementTask;
 use crate::task_manager::imip::SendImipTask;
 use crate::task_manager::index::SearchIndexTask;
-use crate::task_manager::lock::TaskLockManager;
+use crate::task_manager::lock::{TaskLockManager, renew_task_locks};
 use crate::task_manager::maintenance::MaintenanceTask;
 use crate::task_manager::merge_threads::MergeThreadsTask;
 use crate::task_manager::report::{self, SubmitReportTask};
@@ -24,6 +24,7 @@ use crate::task_manager::{
     TaskJob, TaskManagerIpc, TaskResult,
 };
 use common::BuildServer;
+use common::config::network::ClusterRoles;
 use common::config::server::{DEFAULT_TLS_TIMEOUT, ServerProtocol};
 use common::network::limiter::ConcurrencyLimiter;
 use common::network::{ServerInstance, TcpAcceptor};
@@ -58,10 +59,12 @@ pub fn spawn_task_manager(inner: Arc<Inner>) {
         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;
@@ -72,6 +75,28 @@ pub fn spawn_task_manager(inner: Arc<Inner>) {
 
     trc::event!(TaskManager(TaskManagerEvent::ManagerStarted));
 
+    // inbuxa: keep the leases of running tasks alive, every third of a lock
+    // lifetime, until the node stops
+    {
+        let inner = inner.clone();
+        tokio::spawn(async move {
+            let mut renewed_at = Instant::now();
+            loop {
+                tokio::time::sleep(Duration::from_secs(1)).await;
+                let locks = &inner.ipc.task_locks;
+                if locks.is_stopping() {
+                    break;
+                }
+                if renewed_at.elapsed() >= Duration::from_secs((locks.expiry() / 3).max(1)) {
+                    renewed_at = Instant::now();
+                    if locks.held() > 0 {
+                        renew_task_locks(&inner.build_server()).await;
+                    }
+                }
+            }
+        });
+    }
+
     // Create dummy server instance for alarms
     let server_instance = Arc::new(ServerInstance {
         id: "_local".to_string(),
@@ -289,26 +314,13 @@ impl TaskQueueManager for Server {
                                     .caused_by(trc::location!())
                                     .ctx(trc::Key::Value, value)
                             })?;
-                            let enabled = match task_type {
+                            // inbuxa: running here under a lease this node
-                                TaskType::IndexDocument
+                            // renews; don't hand it to a worker again
-                                | TaskType::UnindexDocument
+                            if task_locks.is_held(task_id) {
-                                | TaskType::IndexTrace => roles.search_indexing,
+                                return Ok(true);
-                                TaskType::AccountMaintenance
+                            }
-                                | TaskType::TenantMaintenance
+
-                                | TaskType::DestroyAccount => roles.account_maintenance,
+                            let enabled = task_enabled(roles, task_type);
-                                TaskType::StoreMaintenance => roles.store_maintenance,
-                                TaskType::SpamFilterMaintenance => roles.spam_training,
-                                TaskType::CalendarAlarmEmail
-                                | TaskType::CalendarAlarmNotification
-                                | TaskType::CalendarItipMessage
-                                | TaskType::MergeThreads
-                                | TaskType::DmarcReport
-                                | TaskType::TlsReport
-                                | TaskType::RestoreArchivedItem
-                                | TaskType::AcmeRenewal
-                                | TaskType::DkimManagement
-                                | TaskType::DnsManagement => true,
-                            };
 
                             if !enabled {
                                 trc::event!(
@@ -437,6 +449,48 @@ impl TaskQueueManager for Server {
     }
 }
 
+/// inbuxa: whether this node's cluster role lets it run a task type. Upstream
+/// checked the dedicated roles (search indexing, account and store
+/// maintenance, spam training) and let every node with a task manager run
+/// the rest, whatever its taskQueueProcessing setting. Every task type now
+/// answers to one ClusterTaskType:
+///
+/// - IndexDocument, UnindexDocument, IndexTrace: searchIndexing
+/// - AccountMaintenance, TenantMaintenance, DestroyAccount: accountMaintenance
+/// - StoreMaintenance: storeMaintenance
+/// - SpamFilterMaintenance: spamClassifierTraining
+/// - DmarcReport, TlsReport: outboundMta. They build and send reports to
+///   other domains (TLS reports can go straight to an HTTPS endpoint), which
+///   is the outbound MTA's business.
+/// - CalendarAlarmEmail, CalendarAlarmNotification, CalendarItipMessage,
+///   MergeThreads, RestoreArchivedItem, AcmeRenewal, DkimManagement,
+///   DnsManagement: taskQueueProcessing, the role for queue tasks with no
+///   role of their own.
+///
+/// A node that may not run a task leaves it unclaimed, so a node that may
+/// picks it up.
+pub fn task_enabled(roles: &ClusterRoles, task_type: TaskType) -> bool {
+    match task_type {
+        TaskType::IndexDocument | TaskType::UnindexDocument | TaskType::IndexTrace => {
+            roles.search_indexing
+        }
+        TaskType::AccountMaintenance | TaskType::TenantMaintenance | TaskType::DestroyAccount => {
+            roles.account_maintenance
+        }
+        TaskType::StoreMaintenance => roles.store_maintenance,
+        TaskType::SpamFilterMaintenance => roles.spam_training,
+        TaskType::DmarcReport | TaskType::TlsReport => roles.outbound_mta,
+        TaskType::CalendarAlarmEmail
+        | TaskType::CalendarAlarmNotification
+        | TaskType::CalendarItipMessage
+        | TaskType::MergeThreads
+        | TaskType::RestoreArchivedItem
+        | TaskType::AcmeRenewal
+        | TaskType::DkimManagement
+        | TaskType::DnsManagement => roles.task_manager,
+    }
+}
+
 async fn run_task(
     server: &Server,
     task: &Task,

crates/smtp/src/inbound/milter/client.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 common::config::smtp::session::Milter;
@@ -25,7 +27,19 @@ impl MilterClient<TcpStream> {
     pub async fn connect(config: &Milter, session_id: u64) -> Result<Self> {
         tokio::time::timeout(config.timeout_command, async {
             let mut last_err = Error::Disconnected;
-            for addr in &config.addrs {
+            // inbuxa: a hostname is resolved here, per connection, rather
+            // than while the settings are built
+            let resolved;
+            let addrs = if config.addrs.is_empty() {
+                resolved = tokio::net::lookup_host((config.hostname.as_str(), config.port))
+                    .await
+                    .map_err(Error::Io)?
+                    .collect::<Vec<_>>();
+                &resolved
+            } else {
+                &config.addrs
+            };
+            for addr in addrs {
                 match TcpStream::connect(addr).await {
                     Ok(stream) => {
                         return Ok(MilterClient {

crates/spam-filter/src/modules/pyzor.rs

@@ -48,19 +48,24 @@ pub(crate) async fn pyzor_check(
     // Send message to address. inbuxa: in tests, a fixed table answers
     // instead of a public server (test_response).
     #[cfg(not(feature = "test_mode"))]
-    let response = pyzor_send_message(config.address, config.timeout, &request).await;
+    let response = match tokio::time::timeout(config.timeout, config.address()).await {
+        Ok(Ok(address)) => pyzor_send_message(address, config.timeout, &request).await,
+        Ok(Err(err)) => Err(err),
+        Err(_) => Err(std::io::Error::new(
+            std::io::ErrorKind::TimedOut,
+            "Timed out resolving the Pyzor server",
+        )),
+    };
     #[cfg(feature = "test_mode")]
     let response = std::io::Result::Ok(test_response(&request));
 
-    response
+    response.map(Into::into).map_err(|err| {
-        .map(Into::into)
+        trc::SpamEvent::PyzorError
-        .map_err(|err| {
+            .into_err()
-            trc::SpamEvent::PyzorError
+            .ctx(trc::Key::Url, format!("{}:{}", config.host, config.port))
-                .into_err()
+            .reason(err)
-                .ctx(trc::Key::Url, config.address.to_string())
+            .details("Pyzor failed")
-                .reason(err)
+    })
-                .details("Pyzor failed")
-        })
 }
 
 /// inbuxa: the answers tests get, by digest, instead of a public server's,

crates/store/src/backend/mysql/blob.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 std::ops::Range;
@@ -16,7 +18,7 @@ impl MysqlStore {
         key: &[u8],
         range: Range<usize>,
     ) -> trc::Result<Option<Vec<u8>>> {
-        let mut conn = self.conn_pool.get_conn().await.map_err(into_error)?;
+        let mut conn = self.conn().await?;
         let s = conn
             .prep("SELECT v FROM t WHERE k = ?")
             .await
@@ -39,7 +41,7 @@ impl MysqlStore {
     }
 
     pub(crate) async fn put_blob(&self, key: &[u8], data: &[u8]) -> trc::Result<()> {
-        let mut conn = self.conn_pool.get_conn().await.map_err(into_error)?;
+        let mut conn = self.conn().await?;
         let s = conn
             .prep("INSERT INTO t (k, v) VALUES (?, ?) ON DUPLICATE KEY UPDATE v = VALUES(v)")
             .await
@@ -51,7 +53,7 @@ impl MysqlStore {
     }
 
     pub(crate) async fn delete_blob(&self, key: &[u8]) -> trc::Result<bool> {
-        let mut conn = self.conn_pool.get_conn().await.map_err(into_error)?;
+        let mut conn = self.conn().await?;
         let s = conn
             .prep("DELETE FROM t WHERE k = ?")
             .await

crates/store/src/backend/mysql/lookup.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 mysql_async::{Params, Row, prelude::Queryable};
@@ -16,7 +18,7 @@ impl MysqlStore {
         query: &str,
         params: &[Value<'_>],
     ) -> trc::Result<T> {
-        let mut conn = self.conn_pool.get_conn().await.map_err(into_error)?;
+        let mut conn = self.conn().await?;
         let s = conn.prep(query).await.map_err(into_error)?;
         let params = Params::Positional(params.iter().map(Into::into).collect());
 

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

@@ -32,6 +32,9 @@ impl MysqlStore {
             .max_allowed_packet(config.max_allowed_packet.map(|v| v as usize))
             .wait_timeout(config.timeout.map(|t| t.as_secs() as usize))
             .client_found_rows(true)
+            // inbuxa: notice a server that went away without closing the
+            // connection in minutes, not the system default of two hours
+            .tcp_keepalive(Some(super::POOL_KEEPALIVE_IDLE))
             .tcp_port(config.port as u16);
 
         if config.use_tls {
@@ -95,7 +98,7 @@ impl MysqlStore {
     }
 
     pub(crate) async fn create_storage_tables(&self) -> trc::Result<()> {
-        let mut conn = self.conn_pool.get_conn().await.map_err(into_error)?;
+        let mut conn = self.conn().await?;
 
         for table in [
             SUBSPACE_ACL,
@@ -169,7 +172,7 @@ impl MysqlStore {
     }
 
     pub(crate) async fn create_search_tables(&self) -> trc::Result<()> {
-        let mut conn = self.conn_pool.get_conn().await.map_err(into_error)?;
+        let mut conn = self.conn().await?;
 
         create_search_tables::<EmailSearchField>(&mut conn).await?;
         create_search_tables::<CalendarSearchField>(&mut conn).await?;

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

@@ -27,6 +27,33 @@ pub struct MysqlStore {
     pub(crate) conn_pool: Pool,
 }
 
+/// inbuxa: how long a request waits for a pooled connection (including
+/// opening one). mysql_async's pool has no wait timeout, so upstream waited
+/// forever when the server stopped answering.
+pub(crate) const POOL_WAIT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
+/// inbuxa: idle time before TCP keepalive probes start.
+pub(crate) const POOL_KEEPALIVE_IDLE: std::time::Duration = std::time::Duration::from_secs(60);
+
+impl MysqlStore {
+    /// inbuxa: a pooled connection, or an error once POOL_WAIT_TIMEOUT has
+    /// passed without one.
+    pub(crate) async fn conn(&self) -> trc::Result<mysql_async::Conn> {
+        pool_conn(&self.conn_pool, POOL_WAIT_TIMEOUT).await
+    }
+}
+
+pub(crate) async fn pool_conn(
+    pool: &Pool,
+    wait: std::time::Duration,
+) -> trc::Result<mysql_async::Conn> {
+    match tokio::time::timeout(wait, pool.get_conn()).await {
+        Ok(result) => result.map_err(into_error),
+        Err(_) => Err(trc::StoreEvent::MysqlError
+            .reason("Timed out waiting for a database connection")
+            .details(format!("No connection within {} s", wait.as_secs()))),
+    }
+}
+
 #[inline(always)]
 pub(crate) fn into_error(err: impl Display) -> trc::Error {
     trc::StoreEvent::MysqlError.reason(err)

crates/store/src/backend/mysql/read.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 super::{MysqlStore, into_error, is_timeout_error};
@@ -14,7 +16,7 @@ impl MysqlStore {
     where
         U: Deserialize + 'static,
     {
-        let mut conn = self.conn_pool.get_conn().await.map_err(into_error)?;
+        let mut conn = self.conn().await?;
         let s = conn
             .prep(format!(
                 "SELECT v FROM {} WHERE k = ?",
@@ -36,7 +38,7 @@ impl MysqlStore {
     }
 
     pub(crate) async fn key_exists(&self, key: impl Key) -> trc::Result<bool> {
-        let mut conn = self.conn_pool.get_conn().await.map_err(into_error)?;
+        let mut conn = self.conn().await?;
         let s = conn
             .prep(format!(
                 "SELECT 1 FROM {} WHERE k = ?",
@@ -56,7 +58,7 @@ impl MysqlStore {
         params: IterateParams<T>,
         mut cb: impl for<'x> FnMut(&'x [u8], &'x [u8]) -> trc::Result<bool> + Sync + Send,
     ) -> trc::Result<()> {
-        let mut conn = self.conn_pool.get_conn().await.map_err(into_error)?;
+        let mut conn = self.conn().await?;
         let table = char::from(params.begin.subspace());
         let begin = params.begin.serialize(0);
         let end = params.end.serialize(0);
@@ -155,7 +157,7 @@ impl MysqlStore {
         let key = key.into();
         let table = char::from(key.subspace());
         let key = key.serialize(0);
-        let mut conn = self.conn_pool.get_conn().await.map_err(into_error)?;
+        let mut conn = self.conn().await?;
         let s = conn
             .prep(format!("SELECT v FROM {table} WHERE k = ?"))
             .await

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,12 +21,12 @@ 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 {
     pub async fn index(&self, documents: Vec<IndexDocument>) -> trc::Result<()> {
-        let mut conn = self.conn_pool.get_conn().await.map_err(into_error)?;
+        let mut conn = self.conn().await?;
         let mut tx_opts = TxOpts::default();
         tx_opts
             .with_consistent_snapshot(false)
@@ -94,7 +96,7 @@ impl MysqlStore {
             build_sort(&mut query, sort);
         }
 
-        let mut conn = self.conn_pool.get_conn().await.map_err(into_error)?;
+        let mut conn = self.conn().await?;
         let s = conn.prep(query).await.map_err(into_error)?;
 
         conn.exec::<i64, _, _>(s, params)
@@ -108,7 +110,7 @@ impl MysqlStore {
         let mut query = format!("DELETE FROM {table} ");
         let params = build_filter(&mut query, &filter.filters);
 
-        let mut conn = self.conn_pool.get_conn().await.map_err(into_error)?;
+        let mut conn = self.conn().await?;
         let s = conn.prep(&query).await.map_err(into_error)?;
 
         match conn.exec_drop(s, params.clone()).await {
@@ -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) {
+                    let (value, mode, unindexed) = match (value, op) {
-                        (SearchValue::Text { value, .. }, SearchOperator::Equal) => {
+                        (SearchValue::Text { value, .. }, SearchOperator::Equal) => (
-                            (Value::Bytes(format!("{value:?}").into_bytes()), "BOOLEAN")
+                            Value::Bytes(format!("{value:?}").into_bytes()),
-                        }
+                            "BOOLEAN",
-                        (SearchValue::Text { value, .. }, ..) => {
+                            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() {
+                                // inbuxa: InnoDB never indexes stopwords ("com",
-                                    text_query.push(' ');
+                                // "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());
+                    if unindexed.is_empty() {
-                    values.push(value);
+                        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/mysql/write.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 super::{DELETE_CHUNK_SIZE, MIN_DELETE_CHUNK_SIZE, MysqlStore, into_error, is_timeout_error};
@@ -29,7 +31,7 @@ impl MysqlStore {
     pub(crate) async fn write(&self, mut batch: Batch<'_>) -> trc::Result<AssignedIds> {
         let start = Instant::now();
         let mut retry_count = 0;
-        let mut conn = self.conn_pool.get_conn().await.map_err(into_error)?;
+        let mut conn = self.conn().await?;
 
         loop {
             let err = match self.write_trx(&mut conn, &mut batch).await {
@@ -382,7 +384,7 @@ impl MysqlStore {
     }
 
     pub(crate) async fn purge_store(&self) -> trc::Result<()> {
-        let mut conn = self.conn_pool.get_conn().await.map_err(into_error)?;
+        let mut conn = self.conn().await?;
         for subspace in [SUBSPACE_QUOTA, SUBSPACE_COUNTER, SUBSPACE_IN_MEMORY_COUNTER] {
             purge_table(&mut conn, char::from(subspace)).await?;
         }
@@ -391,7 +393,7 @@ impl MysqlStore {
     }
 
     pub(crate) async fn delete_range(&self, from: impl Key, to: impl Key) -> trc::Result<()> {
-        let mut conn = self.conn_pool.get_conn().await.map_err(into_error)?;
+        let mut conn = self.conn().await?;
         let table = char::from(from.subspace());
         let mut from = from.serialize(0);
         let to = to.serialize(0);

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

@@ -22,11 +22,34 @@ use crate::{
 use ::registry::schema::{enums::PostgreSqlRecyclingMethod, structs};
 use ahash::AHashSet;
 use deadpool_postgres::{
-    Config, ManagerConfig, Object, Pool, PoolConfig, RecyclingMethod, Runtime,
+    Config, ManagerConfig, Object, Pool, PoolConfig, RecyclingMethod, Runtime, Timeouts,
 };
+use std::time::Duration;
 use tokio_postgres::NoTls;
 use utils::tls::rustls_client_config;
 
+/// inbuxa: how long a request waits for a pooled connection.
+pub(crate) const POOL_WAIT_TIMEOUT: Duration = Duration::from_secs(30);
+/// inbuxa: how long opening a connection may take when the store sets no
+/// timeout of its own.
+pub(crate) const POOL_CREATE_TIMEOUT: Duration = Duration::from_secs(15);
+/// inbuxa: how long checking a pooled connection before reuse may take.
+pub(crate) const POOL_RECYCLE_TIMEOUT: Duration = Duration::from_secs(10);
+/// inbuxa: idle time before TCP keepalive probes start.
+pub(crate) const POOL_KEEPALIVE_IDLE: Duration = Duration::from_secs(60);
+
+/// inbuxa: the pool's timeouts. Opening a connection is bounded by the
+/// store's own timeout when it has one; waiting for one covers at least that
+/// long, so a slow connect isn't cut short by the wait.
+pub(crate) fn pool_timeouts(connect_timeout: Option<Duration>) -> Timeouts {
+    let create = connect_timeout.unwrap_or(POOL_CREATE_TIMEOUT);
+    Timeouts {
+        wait: POOL_WAIT_TIMEOUT.max(create).into(),
+        create: create.into(),
+        recycle: POOL_RECYCLE_TIMEOUT.into(),
+    }
+}
+
 impl PostgresStore {
     pub async fn open(config: structs::PostgreSqlStore) -> Result<Store, String> {
         // inbuxa: ST-15: where the primary is, to tell a replica from it
@@ -46,9 +69,20 @@ impl PostgresStore {
                 PostgreSqlRecyclingMethod::Clean => RecyclingMethod::Clean,
             },
         });
-        if let Some(max_conn) = config.pool_max_connections {
+        // inbuxa: upstream set no pool timeouts, so a request waited for a
-            cfg.pool = PoolConfig::new(max_conn as usize).into();
+        // free connection, or for one to be made or recycled, for as long as
-        }
+        // it took: forever when the server stopped answering. A worker now
+        // gets an error instead and the task or request is retried.
+        let mut pool = config
+            .pool_max_connections
+            .map(|max_conn| PoolConfig::new(max_conn as usize))
+            .unwrap_or_default();
+        pool.timeouts = pool_timeouts(cfg.connect_timeout);
+        cfg.pool = pool.into();
+        // Notice a server that went away without closing the connection in
+        // minutes rather than the system default of two hours
+        cfg.keepalives = true.into();
+        cfg.keepalives_idle = POOL_KEEPALIVE_IDLE.into();
 
         let primary_pool = if config.use_tls {
             cfg.create_pool(

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::{
+    backend::{
-        DELETE_CHUNK_SIZE, MIN_DELETE_CHUNK_SIZE, PostgresStore, PsqlSearchField, into_error,
+        MAX_TOKEN_LENGTH,
-        into_pool_error, is_timeout_error,
+        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() {

crates/store/src/backend/redis/lookup.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 super::{RedisPool, RedisStore, into_error};
@@ -79,6 +81,30 @@ impl RedisStore {
         }
     }
 
+    // inbuxa: see InMemoryStore::renew_lock
+    pub async fn renew_lock(&self, key: &[u8], expires: u64) -> trc::Result<bool> {
+        match &self.pool {
+            RedisPool::Single(pool) => {
+                with_conn(pool, async |conn| {
+                    Self::renew_lock_(conn, key, expires).await
+                })
+                .await
+            }
+            RedisPool::Cluster(pool) => {
+                with_conn(pool, async |conn| {
+                    Self::renew_lock_(conn, key, expires).await
+                })
+                .await
+            }
+            RedisPool::Sentinel(pool) => {
+                with_conn(pool, async |conn| {
+                    Self::renew_lock_(conn, key, expires).await
+                })
+                .await
+            }
+        }
+    }
+
     pub async fn key_delete(&self, key: &[u8]) -> trc::Result<()> {
         match &self.pool {
             RedisPool::Single(pool) => {
@@ -226,6 +252,22 @@ impl RedisStore {
             .map(|reply| reply.is_some())
     }
 
+    async fn renew_lock_(
+        conn: &mut impl AsyncCommands,
+        key: &[u8],
+        expires: u64,
+    ) -> RedisResult<bool> {
+        redis::cmd("SET")
+            .arg(key)
+            .arg(now() + expires)
+            .arg("XX")
+            .arg("EX")
+            .arg(expires as i64)
+            .query_async::<Option<String>>(conn)
+            .await
+            .map(|reply| reply.is_some())
+    }
+
     async fn key_delete_(conn: &mut impl AsyncCommands, key: &[u8]) -> RedisResult<()> {
         conn.del(key).await
     }

crates/store/src/dispatch/lookup.rs

@@ -401,6 +401,57 @@ impl InMemoryStore {
         }
     }
 
+    /// inbuxa: extends a lock this node holds to `duration` seconds from now.
+    /// Returns false when the lock is gone or has expired: it may have been
+    /// taken by someone else since, so it is left alone.
+    pub async fn renew_lock(&self, prefix: u8, key: &[u8], duration: u64) -> trc::Result<bool> {
+        match self {
+            InMemoryStore::Store(store) => {
+                let key = KeyValue::<()>::build_key(prefix, key);
+                let key = ValueClass::InMemory(InMemoryClass::Key(key));
+                let Some(lock_expiry) = store
+                    .get_value::<u64>(ValueKey::from(key.clone()))
+                    .await
+                    .caused_by(trc::location!())?
+                else {
+                    return Ok(false);
+                };
+                let now = now();
+                if lock_expiry <= now {
+                    return Ok(false);
+                }
+
+                let mut batch = BatchBuilder::new();
+                batch.assert_value(key.clone(), AssertValue::U64(lock_expiry));
+                batch.set(key, (now + duration).serialize());
+                match store.write(batch.build_all()).await {
+                    Ok(_) => Ok(true),
+                    Err(err) if err.is_assertion_failure() => Ok(false),
+                    Err(err) => Err(err
+                        .details("Failed to renew lock.")
+                        .caused_by(trc::location!())),
+                }
+            }
+            InMemoryStore::Sharded(store) => {
+                Box::pin(
+                    store
+                        .member(&KeyValue::<()>::build_key(prefix, key))
+                        .renew_lock(prefix, key, duration),
+                )
+                .await
+            }
+            #[cfg(feature = "redis")]
+            InMemoryStore::Redis(store) => {
+                store
+                    .renew_lock(&KeyValue::<()>::build_key(prefix, key), duration)
+                    .await
+            }
+            InMemoryStore::Static(_) | InMemoryStore::Http(_) => {
+                Err(trc::StoreEvent::NotSupported.into_err())
+            }
+        }
+    }
+
     pub async fn remove_lock(&self, prefix: u8, key: &[u8]) -> trc::Result<()> {
         self.key_delete(KeyValue::<()>::build_key(prefix, key))
             .await

tests/src/cluster/broadcast.rs

@@ -2,6 +2,8 @@
  * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
  *
  * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
+ *
+ * Modified by Coffey Labs in 2026 for INBUXA.
  */
 
 use crate::{
@@ -11,6 +13,7 @@ use crate::{
         server::TestServerBuilder,
     },
 };
+use common::BuildServer;
 use imap_proto::ResponseType;
 use registry::{
     schema::{
@@ -18,7 +21,8 @@ use registry::{
         prelude::{ObjectType, Property, SocketAddr},
         structs::{
             ClusterListenerGroup, ClusterListenerGroupProperties, ClusterRole, ClusterTaskGroup,
-            Coordinator, Imap, NatsCoordinator, NetworkListener, RedisStore,
+            Coordinator, Imap, MtaDeliverySchedule, MtaVirtualQueue, NatsCoordinator,
+            NetworkListener, RedisStore,
         },
     },
     types::map::Map,
@@ -209,6 +213,45 @@ pub async fn cluster_tests() {
         Some("John Doe")
     );
 
+    // inbuxa: a settings write applies on every node, no ReloadSettings
+    let queue_id = admin
+        .registry_create_object(MtaVirtualQueue {
+            name: "clusterq".into(),
+            threads_per_node: 1,
+            description: None,
+        })
+        .await;
+    admin
+        .registry_create_object(MtaDeliverySchedule {
+            name: "cluster-autoreload".into(),
+            queue_id,
+            ..Default::default()
+        })
+        .await;
+    for (node_id, test) in servers.iter().enumerate() {
+        let started = std::time::Instant::now();
+        while !test
+            .server
+            .inner
+            .build_server()
+            .core
+            .smtp
+            .queue
+            .queue_strategy
+            .contains_key("cluster-autoreload")
+        {
+            assert!(
+                started.elapsed() < std::time::Duration::from_secs(5),
+                "node {node_id} didn't pick up the new delivery schedule"
+            );
+            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        }
+        println!(
+            "Node {node_id} has the new delivery schedule after {} ms",
+            started.elapsed().as_millis()
+        );
+    }
+
     // Run IMAP idle tests across nodes
     let mut node1_client = imap_client("[email protected]", "this is john's secret", 1).await;
     let mut node2_client = imap_client("[email protected]", "this is john's secret", 2).await;

tests/src/cluster/mod.rs

@@ -10,3 +10,4 @@ pub mod broadcast;
 #[cfg(feature = "nats")]
 pub mod coordinator; // inbuxa: coordinator reconnects
 pub mod stress;
+pub mod task_roles; // inbuxa: task types follow cluster roles

tests/src/cluster/task_roles.rs

@@ -0,0 +1,218 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! Two task managers with different cluster roles over one shared store:
+//! each runs only the task types its role allows, and a task one node may
+//! not run is left for the node that may. Needs a store both nodes can open
+//! (STORE=PostgreSql or MySql).
+
+use crate::utils::server::TestServerBuilder;
+use common::Server;
+use registry::{
+    schema::{
+        enums::{ClusterTaskType, IndexDocumentType},
+        structs::{
+            ClusterListenerGroup, ClusterRole, ClusterTaskGroup, ClusterTaskGroupProperties, Task,
+            TaskDnsManagement, TaskIndexDocument, TaskStatus, TaskTlsReport,
+        },
+    },
+    types::map::Map,
+};
+use std::time::{Duration, Instant};
+use store::{
+    ValueKey,
+    write::{BatchBuilder, TaskQueueClass, ValueClass},
+};
+use utils::snowflake::SnowflakeIdGenerator;
+
+const QUEUE_ROLE: &str = "tasks_queue";
+const INDEX_MTA_ROLE: &str = "tasks_index_mta";
+
+#[tokio::test(flavor = "multi_thread")]
+pub async fn task_role_tests() {
+    if matches!(
+        std::env::var("STORE").as_deref(),
+        Ok("RocksDb" | "Sqlite") | Err(_)
+    ) {
+        println!("Skipping task role tests: they need a store both nodes can open.");
+        return;
+    }
+    println!(
+        "Running task role tests on {}...",
+        std::env::var("STORE").unwrap_or_default()
+    );
+
+    // The roles, stored by a node that runs no services of its own (a node
+    // looks its role up when it starts)
+    let seed = TestServerBuilder::new("task_roles_seed")
+        .await
+        .with_object(role(QUEUE_ROLE, &[ClusterTaskType::TaskQueueProcessing]))
+        .await
+        .with_object(role(
+            INDEX_MTA_ROLE,
+            &[
+                ClusterTaskType::SearchIndexing,
+                ClusterTaskType::OutboundMta,
+            ],
+        ))
+        .await
+        .disable_services()
+        .build()
+        .await;
+
+    // Node A runs queue tasks (taskQueueProcessing) only
+    let node_a = TestServerBuilder::new_with_role(
+        "task_roles_a",
+        "node-a.example.com".into(),
+        Some(QUEUE_ROLE.into()),
+        false,
+    )
+    .await
+    .build_with_opts(false)
+    .await;
+    let server_a = node_a.server.clone();
+    let roles = &server_a.core.network.roles;
+    assert!(roles.task_manager && !roles.search_indexing && !roles.outbound_mta);
+
+    // A DNS task (taskQueueProcessing), an unindex task (searchIndexing) and
+    // a TLS report (outboundMta), all due now
+    let [dns, unindex, report] = new_task_ids();
+    let mut batch = BatchBuilder::new();
+    batch
+        .schedule_task_with_id(
+            dns,
+            Task::DnsManagement(TaskDnsManagement {
+                status: TaskStatus::now(),
+                ..Default::default()
+            }),
+        )
+        .schedule_task_with_id(
+            unindex,
+            Task::UnindexDocument(TaskIndexDocument {
+                account_id: 0u32.into(),
+                document_id: u32::MAX.into(),
+                document_type: IndexDocumentType::File,
+                status: TaskStatus::now(),
+            }),
+        )
+        .schedule_task_with_id(
+            report,
+            Task::TlsReport(TaskTlsReport {
+                report_id: u64::MAX.into(),
+                status: TaskStatus::now(),
+            }),
+        );
+    server_a.store().write(batch.build_all()).await.unwrap();
+    server_a.notify_task_queue();
+
+    // Node A runs the DNS task and leaves the other two alone. Upstream ran
+    // the TLS report here too: report tasks ran on any node with a task
+    // manager.
+    wait_until_run(&server_a, &[dns], Duration::from_secs(20)).await;
+    tokio::time::sleep(Duration::from_secs(3)).await;
+    server_a.notify_task_queue();
+    tokio::time::sleep(Duration::from_secs(2)).await;
+    assert!(
+        is_pending(&server_a, unindex).await,
+        "unindex ran on node A"
+    );
+    assert!(
+        is_pending(&server_a, report).await,
+        "TLS report ran on node A"
+    );
+
+    // Node B (search indexing and outbound MTA) comes up and picks up what
+    // node A left
+    let node_b = TestServerBuilder::new_with_role(
+        "task_roles_b",
+        "node-b.example.com".into(),
+        Some(INDEX_MTA_ROLE.into()),
+        false,
+    )
+    .await
+    .build_with_opts(false)
+    .await;
+    let server_b = node_b.server.clone();
+    let roles = &server_b.core.network.roles;
+    assert!(!roles.task_manager && roles.search_indexing && roles.outbound_mta);
+    server_b.notify_task_queue();
+    wait_until_run(&server_b, &[unindex, report], Duration::from_secs(20)).await;
+
+    // A queue task scheduled now still runs, on node A: node B may not
+    // claim it
+    let [dns] = new_task_ids();
+    let mut batch = BatchBuilder::new();
+    batch.schedule_task_with_id(
+        dns,
+        Task::DnsManagement(TaskDnsManagement {
+            status: TaskStatus::now(),
+            ..Default::default()
+        }),
+    );
+    server_b.store().write(batch.build_all()).await.unwrap();
+    server_b.notify_task_queue();
+    tokio::time::sleep(Duration::from_secs(3)).await;
+    assert!(is_pending(&server_b, dns).await, "DNS task ran on node B");
+    server_a.notify_task_queue();
+    wait_until_run(&server_a, &[dns], Duration::from_secs(20)).await;
+
+    if seed.is_reset() {
+        seed.temp_dir.delete();
+        node_a.temp_dir.delete();
+        node_b.temp_dir.delete();
+    }
+}
+
+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()),
+        }),
+    }
+}
+
+fn new_task_ids<const N: usize>() -> [u64; N] {
+    std::array::from_fn(|_| SnowflakeIdGenerator::global_id().unwrap())
+}
+
+/// Still due and never run: present, and pending.
+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_until_run(server: &Server, ids: &[u64], within: Duration) {
+    let started = Instant::now();
+    loop {
+        let mut left = 0;
+        for id in ids {
+            if is_pending(server, *id).await {
+                left += 1;
+            }
+        }
+        if left == 0 {
+            return;
+        }
+        assert!(
+            started.elapsed() < within,
+            "{left} task(s) still pending after {:?}",
+            started.elapsed()
+        );
+        tokio::time::sleep(Duration::from_millis(250)).await;
+    }
+}

tests/src/smtp/inbound/milter.rs

@@ -2,6 +2,8 @@
  * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
  *
  * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
+ *
+ * Modified by Coffey Labs in 2026 for INBUXA.
  */
 
 use crate::{
@@ -75,7 +77,7 @@ async fn milter_session() {
                 else_: "true".into(),
                 ..Default::default()
             },
-            hostname: "127.0.0.1".into(),
+            hostname: "localhost".into(), // inbuxa: resolved when the session connects
             port: 9332,
             use_tls: false,
             stages: Map::new(vec![MtaStage::Data]),

tests/src/smtp/inbound/mod.rs

@@ -2,6 +2,8 @@
  * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
  *
  * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
+ *
+ * Modified by Coffey Labs in 2026 for INBUXA.
  */
 
 use crate::utils::server::TestServer;
@@ -40,15 +42,23 @@ pub mod vrfy;
 const EVENT_TIMEOUT: Duration = Duration::from_secs(5);
 
 impl TestServer {
+    // inbuxa: registry writes reload the settings, and each reload sends the
+    // queue a ReloadSettings; read_event, try_read_event and assert_no_events
+    // pass over those (expect_reload_settings still waits for one)
     pub async fn read_event(&mut self) -> QueueEvent {
-        if let Some(event) = self.queue_events.pop_front() {
+        while let Some(event) = self.queue_events.pop_front() {
-            return event;
+            if !event.is_reload_settings() {
+                return event;
+            }
         }
 
-        match tokio::time::timeout(EVENT_TIMEOUT, self.queue_rx.recv()).await {
+        loop {
-            Ok(Some(event)) => event,
+            match tokio::time::timeout(EVENT_TIMEOUT, self.queue_rx.recv()).await {
-            Ok(None) => panic!("Channel closed."),
+                Ok(Some(event)) if event.is_reload_settings() => (),
-            Err(_) => panic!("No queue event received."),
+                Ok(Some(event)) => return event,
+                Ok(None) => panic!("Channel closed."),
+                Err(_) => panic!("No queue event received."),
+            }
         }
     }
 
@@ -78,26 +88,39 @@ impl TestServer {
     }
 
     pub async fn try_read_event(&mut self) -> Option<QueueEvent> {
-        if let Some(event) = self.queue_events.pop_front() {
+        while let Some(event) = self.queue_events.pop_front() {
-            return Some(event);
+            if !event.is_reload_settings() {
+                return Some(event);
+            }
         }
 
-        match tokio::time::timeout(EVENT_TIMEOUT, self.queue_rx.recv()).await {
+        loop {
-            Ok(Some(event)) => Some(event),
+            match tokio::time::timeout(EVENT_TIMEOUT, self.queue_rx.recv()).await {
-            Ok(None) => panic!("Channel closed."),
+                Ok(Some(event)) if event.is_reload_settings() => (),
-            Err(_) => None,
+                Ok(Some(event)) => return Some(event),
+                Ok(None) => panic!("Channel closed."),
+                Err(_) => return None,
+            }
         }
     }
 
     pub fn assert_no_events(&mut self) {
-        if let Some(event) = self.queue_events.pop_front() {
+        if let Some(event) = self
+            .queue_events
+            .iter()
+            .find(|event| !event.is_reload_settings())
+        {
             panic!("Expected empty queue but got {event:?}");
         }
+        self.queue_events.clear();
 
-        match self.queue_rx.try_recv() {
+        loop {
-            Err(TryRecvError::Empty) => (),
+            match self.queue_rx.try_recv() {
-            Ok(event) => panic!("Expected empty queue but got {event:?}"),
+                Ok(event) if event.is_reload_settings() => (),
-            Err(err) => panic!("Queue error: {err:?}"),
+                Err(TryRecvError::Empty) => break,
+                Ok(event) => panic!("Expected empty queue but got {event:?}"),
+                Err(err) => panic!("Queue error: {err:?}"),
+            }
         }
     }
 

tests/src/store/mod.rs

@@ -10,6 +10,8 @@ pub mod blob;
 pub mod import_export;
 pub mod lookup;
 pub mod ops;
+#[cfg(any(feature = "postgres", feature = "mysql"))]
+pub mod pool_timeout; // inbuxa: SQL pools give up instead of hanging
 pub mod query;
 pub mod registry;
 #[cfg(feature = "postgres")]

tests/src/store/pool_timeout.rs

@@ -0,0 +1,96 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! A database that accepts connections and then says nothing (a hung or
+//! half-dead server, a black-holed failover) gives a worker an error within
+//! the pool's timeouts. Upstream's pools had none, so the worker waited for
+//! good. No database is needed: a local listener that never answers plays
+//! the server.
+
+use registry::schema::structs::DataStore;
+use std::time::{Duration, Instant};
+use store::{Store, ValueKey, write::ValueClass};
+use tokio::net::TcpListener;
+
+/// Accepts connections on a local port and never sends a byte.
+async fn silent_server() -> u16 {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let port = listener.local_addr().unwrap().port();
+    tokio::spawn(async move {
+        let mut held = Vec::new();
+        while let Ok((socket, _)) = listener.accept().await {
+            held.push(socket);
+        }
+    });
+    port
+}
+
+/// Builds the store and reads a key; both must end, with an error for the
+/// read, well within `limit`.
+async fn assert_times_out(data_store: DataStore, limit: Duration) {
+    let started = Instant::now();
+    let result = tokio::time::timeout(limit, async {
+        match Store::build(data_store).await {
+            Ok(store) => store
+                .get_value::<u64>(ValueKey::from(ValueClass::Property(0)))
+                .await
+                .map(|_| ())
+                .map_err(|err| err.to_string()),
+            Err(err) => Err(err.to_string()),
+        }
+    })
+    .await;
+    let elapsed = started.elapsed();
+    match result {
+        Ok(Err(err)) => println!("Got {err} after {elapsed:?}"),
+        Ok(Ok(())) => panic!("a silent server answered?"),
+        Err(_) => panic!("still waiting for a connection after {elapsed:?}"),
+    }
+}
+
+#[cfg(feature = "postgres")]
+#[tokio::test(flavor = "multi_thread")]
+pub async fn postgres_pool_timeout() {
+    use registry::schema::structs::PostgreSqlStore;
+
+    let port = silent_server().await;
+    println!("Running PostgreSQL pool timeout test...");
+    // The store's own timeout bounds opening a connection, handshake
+    // included (tokio-postgres's connect_timeout covers only the TCP connect)
+    assert_times_out(
+        DataStore::PostgreSql(PostgreSqlStore {
+            host: "127.0.0.1".into(),
+            port: port as u64,
+            database: "none".into(),
+            timeout: Some(Duration::from_secs(2).into()),
+            use_tls: false,
+            ..Default::default()
+        }),
+        Duration::from_secs(20),
+    )
+    .await;
+}
+
+#[cfg(feature = "mysql")]
+#[tokio::test(flavor = "multi_thread")]
+pub async fn mysql_pool_timeout() {
+    use registry::schema::structs::MySqlStore;
+
+    let port = silent_server().await;
+    println!("Running MySQL pool timeout test...");
+    // mysql_async has no pool timeout; the store waits 30 s for a connection
+    assert_times_out(
+        DataStore::MySql(MySqlStore {
+            host: "127.0.0.1".into(),
+            port: port as u64,
+            database: "none".into(),
+            use_tls: false,
+            ..Default::default()
+        }),
+        Duration::from_secs(60),
+    )
+    .await;
+}

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

@@ -79,7 +79,33 @@ pub async fn task_lock_tests() {
         "ran before the other node's locks expired: {elapsed:?}"
     );
 
-    // 3. A graceful stop releases the locks this node holds: another node
+    // 3. A task that runs longer than a lock lifetime keeps its claim: the
+    // task manager renews the lease while this node holds it, and the claim
+    // ends when the task does. (Before, a lock simply lasted an hour.)
+    let [id] = new_task_ids(1)[..] else {
+        unreachable!()
+    };
+    assert!(server.try_lock_task(id).await, "claim {id}");
+    tokio::time::sleep(Duration::from_secs(LOCK_EXPIRY + LOCK_EXPIRY / 2)).await;
+    assert!(
+        !foreign_lock(&server, id, LOCK_EXPIRY).await,
+        "lease lapsed while the task ran"
+    );
+    server.remove_index_lock(id).await;
+    assert!(
+        foreign_lock(&server, id, LOCK_EXPIRY).await,
+        "released when the task ended"
+    );
+    let _ = server
+        .in_memory_store()
+        .remove_lock(KV_LOCK_TASK, &id.to_be_bytes())
+        .await;
+    assert!(
+        common::ipc::TaskLocks::DEFAULT_EXPIRY <= 5 * 60,
+        "a dead node's tasks wait no more than a few minutes"
+    );
+
+    // 4. 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 {

tests/src/system/auto_reload.rs

@@ -0,0 +1,220 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+// inbuxa: a registry write to an object the running settings are built from
+// applies without an x:Action ReloadSettings, and the set response says so.
+
+use crate::utils::{
+    jmap::JmapResponse,
+    server::{TestServer, TestServerBuilder},
+};
+use common::BuildServer;
+use registry::{
+    schema::{
+        enums::TracingLevel,
+        prelude::ObjectType,
+        structs::{
+            AllowedIp, CertificateManagement, DkimManagement, DnsManagement, Domain, Expression,
+            MtaDeliverySchedule, MtaStageAuth, MtaVirtualQueue, Tracer, TracerStdout,
+        },
+    },
+    types::ipmask::IpAddrOrMask,
+};
+use serde_json::Value;
+
+#[tokio::test(flavor = "multi_thread")]
+pub async fn settings_reload_tests() {
+    let mut test = TestServerBuilder::new("settings_reload_tests")
+        .await
+        .with_default_listeners()
+        .await
+        .with_object(MtaStageAuth {
+            require: Expression {
+                else_: "false".to_string(),
+                ..Default::default()
+            },
+            ..Default::default()
+        })
+        .await
+        .build()
+        .await;
+
+    let admin = test
+        .create_user_account(
+            "admin",
+            "[email protected]",
+            "these_pretzels_are_making_me_thirsty",
+            &[],
+            "Admin",
+        )
+        .await;
+    test.account("admin")
+        .assign_roles_to_account(admin.id(), &["user", "system"])
+        .await;
+    test.insert_account(admin);
+
+    test_write_applies(&test).await;
+
+    if test.is_reset() {
+        test.temp_dir.delete();
+    }
+}
+
+async fn test_write_applies(test: &TestServer) {
+    println!("Running settings reload after registry writes...");
+    let admin = test.account("[email protected]");
+
+    // A delivery schedule is in use as soon as it is saved
+    let response = admin
+        .registry_create([MtaVirtualQueue {
+            name: "autorld".into(),
+            threads_per_node: 2,
+            description: None,
+        }])
+        .await;
+    assert_applied(&response);
+    let queue_id = response.created_id(0);
+    assert!(!has_schedule(test, "autoreload-schedule"));
+    let response = admin
+        .registry_create([MtaDeliverySchedule {
+            name: "autoreload-schedule".into(),
+            queue_id,
+            ..Default::default()
+        }])
+        .await;
+    assert_applied(&response);
+    assert!(has_schedule(test, "autoreload-schedule"));
+
+    // Destroyed, it's gone at once too
+    let schedule_id = response.created_id(0);
+    let response = admin
+        .registry_destroy(ObjectType::MtaDeliverySchedule, [schedule_id])
+        .await;
+    assert_applied(&response);
+    assert!(!has_schedule(test, "autoreload-schedule"));
+
+    // Concurrent writes all end up in the running settings
+    let names = (0..8)
+        .map(|i| format!("autoreload-{i}"))
+        .collect::<Vec<_>>();
+    let mut writes = Vec::new();
+    for name in &names {
+        writes.push(admin.registry_create([MtaDeliverySchedule {
+            name: name.clone(),
+            queue_id,
+            ..Default::default()
+        }]));
+    }
+    let mut schedule_ids = Vec::new();
+    for response in futures::future::join_all(writes).await {
+        assert_applied(&response);
+        schedule_ids.push(response.created_id(0));
+    }
+    for name in &names {
+        assert!(has_schedule(test, name), "{name} missing");
+    }
+    // Several objects in one request: one reload
+    let response = admin
+        .registry_destroy(ObjectType::MtaDeliverySchedule, schedule_ids.iter())
+        .await;
+    assert_applied(&response);
+    for name in &names {
+        assert!(!has_schedule(test, name), "{name} still present");
+    }
+
+    // A write whose reload fails is stored, and the response says the
+    // settings weren't reloaded: only one console tracer is allowed.
+    let response = admin
+        .registry_create([
+            Tracer::Stdout(TracerStdout {
+                enable: true,
+                level: TracingLevel::Error,
+                ..Default::default()
+            }),
+            Tracer::Stdout(TracerStdout {
+                enable: true,
+                level: TracingLevel::Error,
+                ..Default::default()
+            }),
+        ])
+        .await;
+    let reload = settings_reload(&response).expect("x:settingsReload missing");
+    assert_eq!(reload["applied"], Value::Bool(false), "{response:?}");
+    let description = reload["description"].as_str().unwrap_or_default();
+    assert!(
+        description.starts_with("Saved, but the running settings were not reloaded. ")
+            && description.contains("Only one console tracer is allowed"),
+        "{description}"
+    );
+    let tracer_ids = [response.created_id(0), response.created_id(1)];
+    let response = admin
+        .registry_destroy(ObjectType::Tracer, tracer_ids.iter())
+        .await;
+    assert_applied(&response);
+
+    // An allowed IP is live as soon as it is saved, and gone once
+    // destroyed. It lives in the core's security settings, which the
+    // blocked-IP reload it used to get doesn't rebuild.
+    let ip: std::net::IpAddr = "198.51.100.7".parse().unwrap();
+    assert!(!is_allowed(test, ip));
+    let response = admin
+        .registry_create([AllowedIp {
+            address: IpAddrOrMask::from_ip(ip),
+            reason: Some("autoreload".into()),
+            ..Default::default()
+        }])
+        .await;
+    assert_applied(&response);
+    assert!(
+        is_allowed(test, ip),
+        "allowed IP not in the running settings"
+    );
+    let allowed_id = response.created_id(0);
+    let response = admin
+        .registry_destroy(ObjectType::AllowedIp, [allowed_id])
+        .await;
+    assert_applied(&response);
+    assert!(!is_allowed(test, ip), "destroyed allowed IP still live");
+
+    // Data that isn't part of the running settings doesn't reload them
+    let response = admin
+        .registry_create([Domain {
+            name: "autoreload.example.org".into(),
+            certificate_management: CertificateManagement::Manual,
+            dns_management: DnsManagement::Manual,
+            dkim_management: DkimManagement::Manual,
+            ..Default::default()
+        }])
+        .await;
+    assert!(settings_reload(&response).is_none(), "{response:?}");
+}
+
+fn settings_reload(response: &JmapResponse) -> Option<&Value> {
+    response.pointer("/methodResponses/0/1/x:settingsReload")
+}
+
+fn assert_applied(response: &JmapResponse) {
+    assert_eq!(
+        settings_reload(response),
+        Some(&serde_json::json!({"applied": true})),
+        "{response:?}"
+    );
+}
+
+fn has_schedule(test: &TestServer, name: &str) -> bool {
+    test.server
+        .inner
+        .build_server()
+        .core
+        .smtp
+        .queue
+        .queue_strategy
+        .contains_key(name)
+}
+
+fn is_allowed(test: &TestServer, ip: std::net::IpAddr) -> bool {
+    test.server.inner.build_server().is_ip_allowed(ip)
+}

tests/src/system/mod.rs

@@ -11,6 +11,7 @@ pub mod authentication;
 pub mod ai;
 pub mod ai_calibration;
 pub mod authorization;
+pub mod auto_reload; // inbuxa: registry writes apply at once
 pub mod branding;
 pub mod crypto;
 pub mod delivery;
@@ -20,6 +21,7 @@ pub mod monitoring;
 pub mod oidc;
 pub mod purge;
 pub mod quota;
+pub mod reload; // inbuxa: reloads and build errors
 pub mod security;
 pub mod task;
 pub mod tenant;

tests/src/system/reload.rs

@@ -0,0 +1,213 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+// inbuxa: a settings reload isn't held back by a DNS lookup, or by objects
+// that already failed when the running settings were built; an error in an
+// object that built then still refuses it, and says which object.
+
+use crate::utils::server::{TestServer, TestServerBuilder};
+use common::{BuildServer, config::mailstore::spamfilter::PyzorConfig, ipc::RegistryChange};
+use registry::schema::{
+    enums::TracingLevel,
+    prelude::{ObjectType, Property},
+    structs::{Action, Expression, MtaStageAuth, SpamPyzor, Tracer, TracerStdout},
+};
+
+#[tokio::test(flavor = "multi_thread")]
+pub async fn reload_tests() {
+    let mut test = TestServerBuilder::new("reload_tests")
+        .await
+        .with_default_listeners()
+        .await
+        .with_object(MtaStageAuth {
+            require: Expression {
+                else_: "false".to_string(),
+                ..Default::default()
+            },
+            ..Default::default()
+        })
+        .await
+        .build()
+        .await;
+
+    let admin = test
+        .create_user_account(
+            "admin",
+            "[email protected]",
+            "these_pretzels_are_making_me_thirsty",
+            &[],
+            "Admin",
+        )
+        .await;
+    test.account("admin")
+        .assign_roles_to_account(admin.id(), &["user", "system"])
+        .await;
+    test.insert_account(admin);
+
+    test_unresolvable_pyzor(&test).await;
+    test_build_errors(&test).await;
+
+    if test.is_reset() {
+        test.temp_dir.delete();
+    }
+}
+
+async fn test_unresolvable_pyzor(test: &TestServer) {
+    println!("Running reload with an unresolvable Pyzor host...");
+    let admin = test.account("[email protected]");
+
+    // Upstream resolved the host while building the settings and refused the
+    // reload when that failed.
+    admin
+        .registry_update_setting(
+            SpamPyzor {
+                enable: true,
+                host: "pyzor.invalid".into(),
+                port: 24441,
+                ..Default::default()
+            },
+            &[Property::Enable, Property::Host, Property::Port],
+        )
+        .await;
+    admin.reload_settings().await;
+
+    let pyzor = running_pyzor(test);
+    assert_eq!(pyzor.host, "pyzor.invalid");
+    assert_eq!(pyzor.port, 24441);
+    assert!(pyzor.address().await.is_err());
+
+    // An IP address needs no lookup
+    admin
+        .registry_update_setting(
+            SpamPyzor {
+                host: "192.0.2.1".into(),
+                ..Default::default()
+            },
+            &[Property::Host],
+        )
+        .await;
+    admin.reload_settings().await;
+    assert_eq!(
+        running_pyzor(test).address().await.unwrap().to_string(),
+        "192.0.2.1:24441"
+    );
+}
+
+async fn test_build_errors(test: &TestServer) {
+    println!("Running reload with build errors...");
+    let admin = test.account("[email protected]");
+    let pyzor_ratio = running_pyzor(test).ratio;
+    assert_ne!(pyzor_ratio, 0.25);
+
+    // Two console tracers: only one is allowed, so the build of one of them
+    // fails. Neither existed when the running settings were built.
+    let mut tracer_ids = Vec::new();
+    for _ in 0..2 {
+        tracer_ids.push(
+            admin
+                .registry_create_object(Tracer::Stdout(TracerStdout {
+                    enable: true,
+                    level: TracingLevel::Error,
+                    ..Default::default()
+                }))
+                .await,
+        );
+    }
+    admin
+        .registry_update_setting(
+            SpamPyzor {
+                ratio: 0.25.into(),
+                ..Default::default()
+            },
+            &[Property::Ratio],
+        )
+        .await;
+
+    // A new error refuses the reload and names the object
+    let err = admin
+        .registry_create_object_expect_err(Action::ReloadSettings)
+        .await;
+    let description = err.description.clone().unwrap_or_default();
+    assert!(
+        description.starts_with("Settings were not reloaded. ")
+            && description.contains("Tracer")
+            && description.contains("Only one console tracer is allowed"),
+        "{err:?}"
+    );
+    assert_eq!(running_pyzor(test).ratio, pyzor_ratio);
+
+    // Had the running settings been built with that tracer failing, as a
+    // restart now would, the same error doesn't hold the reload back.
+    let result = Box::pin(
+        test.server
+            .reload_registry(RegistryChange::Reload(ObjectType::DataStore)),
+    )
+    .await
+    .unwrap();
+    assert!(!result.replaced_core);
+    assert_eq!(result.errors.len(), 1, "{:?}", result.errors);
+    test.server.record_build_errors(&result.errors);
+
+    admin.reload_settings().await;
+    assert_eq!(running_pyzor(test).ratio, 0.25);
+    let result = Box::pin(
+        test.server
+            .reload_registry(RegistryChange::Reload(ObjectType::DataStore)),
+    )
+    .await
+    .unwrap();
+    assert!(result.replaced_core);
+    assert!(result.errors.is_empty());
+    assert_eq!(result.known_errors.len(), 1);
+
+    // Once fixed, the object is no longer known to fail, so a new error
+    // there refuses the reload again.
+    admin
+        .registry_destroy(ObjectType::Tracer, tracer_ids.iter())
+        .await
+        .assert_destroyed(&tracer_ids);
+    admin.reload_settings().await;
+    let result = Box::pin(
+        test.server
+            .reload_registry(RegistryChange::Reload(ObjectType::DataStore)),
+    )
+    .await
+    .unwrap();
+    assert!(result.replaced_core);
+    assert!(result.errors.is_empty() && result.known_errors.is_empty());
+
+    for _ in 0..2 {
+        tracer_ids.push(
+            admin
+                .registry_create_object(Tracer::Stdout(TracerStdout {
+                    enable: true,
+                    level: TracingLevel::Error,
+                    ..Default::default()
+                }))
+                .await,
+        );
+    }
+    admin
+        .registry_create_object_expect_err(Action::ReloadSettings)
+        .await;
+    let tracer_ids = tracer_ids.split_off(2);
+    admin
+        .registry_destroy(ObjectType::Tracer, tracer_ids.iter())
+        .await
+        .assert_destroyed(&tracer_ids);
+    admin.reload_settings().await;
+}
+
+fn running_pyzor(test: &TestServer) -> PyzorConfig {
+    test.server
+        .inner
+        .build_server()
+        .core
+        .spam
+        .pyzor
+        .clone()
+        .expect("Pyzor enabled")
+}