Publish amd64 first, then arm64, on a builder that keeps its cache

Two release builds side by side on one machine each take twice as long,
and production only needs amd64. publish-amd64 now pushes :<version> as
soon as the amd64 build is done; publish-arm64 builds arm64 afterwards,
then replaces :<version> with the two-platform index and moves :latest.

Both jobs use one named BuildKit builder whose container outlives the
job, so the dependency layer (cargo chef cook) is reused until the
dependencies change. The release is created after amd64; the binaries
are attached once arm64 is in.

.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/manager/backup.rs

@@ -23,6 +23,13 @@ use utils::{UnwrapFailure, codec::leb128::Leb128_};
 
 pub(super) const MAGIC_MARKER: u8 = 123;
 
+// inbuxa: blobs kept under a fixed name instead of a content hash. Nothing
+// links to them, so the export names them outright.
+const NAMED_BLOBS: &[&[u8]] = &[
+    crate::manager::SPAM_CLASSIFIER_KEY,
+    crate::manager::SPAM_TRAINER_KEY,
+];
+
 #[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)]
 pub(super) enum Family {
     Data = 0,
@@ -143,15 +150,21 @@ impl Core {
                     .await
                     .failed("Failed to iterate over data store");
 
-                for hash in blobs {
+                // inbuxa: the trained spam classifier and its trainer state are
+                // blobs stored under fixed names with no blob link, so the walk
+                // over links above never reaches them.
+                let named = NAMED_BLOBS.iter().map(|key| key.to_vec());
+                for key in blobs
+                    .into_iter()
+                    .map(|hash| hash.as_slice().to_vec())
+                    .chain(named)
+                {
                     if let Some(blob) = blob_store
-                        .get_blob(hash.as_slice(), 0..usize::MAX)
+                        .get_blob(&key, 0..usize::MAX)
                         .await
                         .failed("Failed to get blob")
                     {
-                        writer
+                        writer.send((key, blob)).failed("Failed to send key");
-                            .send((hash.as_slice().to_vec(), blob))
-                            .failed("Failed to send key");
                     }
                 }
             }),
@@ -323,7 +336,13 @@ impl Family {
                 SUBSPACE_REGISTRY_IDX,
                 SUBSPACE_REGISTRY_PK,
                 SUBSPACE_DIRECTORY,
-                store::SUBSPACE_INBUXA, // inbuxa: masked email
+                // inbuxa: registry objects the upstream list left out, so an
+                // export dropped them: archived items (undelete) and spam
+                // training samples. Their indexes and id counters already
+                // travel in this family and in `data`, so they ride along.
+                SUBSPACE_DELETED_ITEMS,
+                SUBSPACE_SPAM_SAMPLES,
+                store::SUBSPACE_INBUXA, // inbuxa: the fork's own data (masked email, undelete, policies)
             ],
             Family::Changelog => &[SUBSPACE_LOGS],
             Family::Queue => &[SUBSPACE_QUEUE_MESSAGE, SUBSPACE_QUEUE_EVENT],

crates/common/src/manager/boot.rs

@@ -54,6 +54,13 @@ Options:
   -o, --console                    Open the store console
   -h, --help                       Print help
   -V, --version                    Print version
+
+An export holds everything in the data and blob stores except short-lived
+in-memory state (rate limits, locks, greylisting) and the full-text search
+index, which belongs to one search backend. An import into an empty store
+queues the index to be rebuilt when the server next starts. EXPORT_TYPES
+limits an export to some of: data, registry, blob, changelog, queue, report,
+telemetry, tasks.
 "#
 );
 
@@ -256,10 +263,10 @@ impl BootManager {
                 telemetry.enable();
 
                 // Parse settings and restore
-                Box::pin(Core::parse(&mut bootstrap, storage))
+                let core = Box::pin(Core::parse(&mut bootstrap, storage)).await;
-                    .await
+                let imported = core.restore(path).await;
-                    .restore(path)
+                // inbuxa: the search index isn't exported; rebuild it
-                    .await;
+                core.queue_reindex(&imported).await;
                 std::process::exit(0);
             }
             StoreOp::Console => {

crates/common/src/manager/restore.rs

@@ -9,15 +9,22 @@
 use super::backup::MAGIC_MARKER;
 use crate::{Core, DATABASE_SCHEMA_VERSION};
 use lz4_flex::frame::FrameDecoder;
-use registry::schema::enums::CompressionAlgo;
+use registry::{
+    schema::{
+        enums::{CompressionAlgo, TaskStoreMaintenanceType},
+        structs::{Task, TaskStatus, TaskStoreMaintenance},
+    },
+    types::EnumImpl,
+};
 use std::{
     fs::File,
     io::{BufReader, ErrorKind, Read},
     path::{Path, PathBuf},
 };
 use store::{
-    BlobStore, IterateParams, SUBSPACE_BLOBS, SUBSPACE_COUNTER, SUBSPACE_INDEXES, SUBSPACE_QUOTA,
+    BlobStore, IterateParams, SUBSPACE_BLOBS, SUBSPACE_COUNTER, SUBSPACE_INDEXES,
-    SUBSPACE_REGISTRY_PK, Store, U32_LEN,
+    SUBSPACE_PROPERTY, SUBSPACE_QUOTA, SUBSPACE_REGISTRY_PK, SUBSPACE_TELEMETRY_SPAN, Store,
+    U32_LEN,
     write::{
         AnyClass, AnyKey, BatchBuilder, ValueClass,
         key::{DeserializeBigEndian, is_node_id_key},
@@ -27,7 +34,9 @@ use types::{collection::Collection, field::Field};
 use utils::{UnwrapFailure, failed};
 
 impl Core {
-    pub async fn restore(&self, src: PathBuf) {
+    /// Imports an export into an empty store and returns the subspaces it
+    /// wrote. inbuxa: the caller hands them to [`Core::queue_reindex`].
+    pub async fn restore(&self, src: PathBuf) -> Vec<u8> {
         // Backup the core
         let paths = if src.is_dir() {
             let mut paths = Vec::new();
@@ -64,6 +73,13 @@ impl Core {
             std::process::exit(1);
         }
 
+        let mut imported = paths
+            .iter()
+            .map(|path| KeyValueReader::new(path).subspace)
+            .collect::<Vec<_>>();
+        imported.sort_unstable();
+        imported.dedup();
+
         let mut tasks = Vec::new();
         for path in paths {
             let storage = self.storage.clone();
@@ -76,6 +92,54 @@ impl Core {
         for task in tasks {
             task.await.failed("Failed to wait for task");
         }
+
+        imported
+    }
+
+    /// inbuxa: an export never carries the full-text index. It is built by
+    /// and for one search backend (the SQL stores index into their own
+    /// tables, the key-value stores into a subspace, external engines keep it
+    /// themselves), so it would be wrong or unreadable after a move to
+    /// another one. Instead, an import queues the same reindex tasks an
+    /// administrator can queue by hand (`reindexAccounts` and
+    /// `reindexTelemetry` store maintenance), and the server rebuilds the
+    /// index for whatever search store it is configured with once it starts.
+    pub async fn queue_reindex(&self, imported: &[u8]) -> Vec<TaskStoreMaintenanceType> {
+        let mut queued = Vec::new();
+        if imported.contains(&SUBSPACE_PROPERTY) {
+            queued.push(TaskStoreMaintenanceType::ReindexAccounts);
+        }
+        if imported.contains(&SUBSPACE_TELEMETRY_SPAN) {
+            queued.push(TaskStoreMaintenanceType::ReindexTelemetry);
+        }
+        if queued.is_empty() {
+            return queued;
+        }
+
+        let mut batch = BatchBuilder::new();
+        for maintenance_type in &queued {
+            batch.schedule_task(Task::StoreMaintenance(TaskStoreMaintenance {
+                maintenance_type: *maintenance_type,
+                status: TaskStatus::now(),
+                shard_index: None,
+            }));
+        }
+        self.storage
+            .data
+            .write(batch.build_all())
+            .await
+            .failed("Failed to queue the reindex tasks");
+
+        println!(
+            "Queued {} to rebuild the search index; it runs when the server starts.",
+            queued
+                .iter()
+                .map(|t| t.as_str())
+                .collect::<Vec<_>>()
+                .join(" and ")
+        );
+
+        queued
     }
 }
 
@@ -125,17 +189,22 @@ async fn restore_file(store: Store, blob_store: BlobStore, path: &Path) {
         }
         SUBSPACE_COUNTER | SUBSPACE_QUOTA => {
             while let Some((key, value)) = reader.next() {
-                batch.add(
+                let class = ValueClass::Any(AnyClass {
-                    ValueClass::Any(AnyClass {
+                    subspace: reader.subspace,
-                        subspace: reader.subspace,
+                    key,
-                        key,
+                });
-                    }),
+                let value = u64::from_le_bytes(
-                    u64::from_le_bytes(
+                    value
-                        value
+                        .try_into()
-                            .try_into()
+                        .expect("Failed to deserialize counter/quota"),
-                            .expect("Failed to deserialize counter/quota"),
+                ) as i64;
-                    ) as i64,
+                // inbuxa: the SQL stores add a negative amount with an UPDATE,
-                );
+                // which does nothing to a row that isn't there yet, so a
+                // negative counter vanished on import. Create the row first.
+                if value < 0 {
+                    batch.add(class.clone(), 0);
+                }
+                batch.add(class, value);
                 if batch.is_large_batch() {
                     store
                         .write(batch.build_all())

crates/jmap/src/inbuxa/telemetry.rs

@@ -427,9 +427,24 @@ pub(crate) async fn trace_query(
             }
             None => false,
         },
-        Property::QueueId => match value.as_str() {
+        // The queue id column is an integer on every search backend, and
+        // holds a trace's first queue id; the keywords carry all of them
+        Property::QueueId => match value
+            .as_str()
+            .and_then(|v| v.trim().parse::<u64>().ok())
+            .or_else(|| value.as_u64())
+        {
             Some(queue_id) => {
-                search.push(SearchFilter::eq(TracingSearchField::QueueId, queue_id.to_string()));
+                search.extend([
+                    SearchFilter::Or,
+                    SearchFilter::eq(TracingSearchField::QueueId, queue_id),
+                    SearchFilter::has_text(
+                        TracingSearchField::Keywords,
+                        queue_id.to_string(),
+                        nlp::language::Language::None,
+                    ),
+                    SearchFilter::End,
+                ]);
                 true
             }
             None => false,

crates/services/src/broadcast/subscriber.rs

@@ -26,7 +26,7 @@ pub fn spawn_broadcast_subscriber(inner: Arc<Inner>, mut shutdown_rx: watch::Rec
     };
 
     tokio::spawn(async move {
-        let mut retry_count = 0;
+        let mut retry_count: u32 = 0;
 
         trc::event!(Cluster(ClusterEvent::SubscriberStart));
 
@@ -53,7 +53,7 @@ pub fn spawn_broadcast_subscriber(inner: Arc<Inner>, mut shutdown_rx: watch::Rec
                     );
 
                     match tokio::time::timeout(
-                        Duration::from_secs(1 << retry_count.max(6)),
+                        subscribe_retry_delay(retry_count),
                         shutdown_rx.changed(),
                     )
                     .await
@@ -62,7 +62,7 @@ pub fn spawn_broadcast_subscriber(inner: Arc<Inner>, mut shutdown_rx: watch::Rec
                             break;
                         }
                         Err(_) => {
-                            retry_count += 1;
+                            retry_count = retry_count.saturating_add(1);
                             continue;
                         }
                     }
@@ -234,6 +234,11 @@ pub fn spawn_broadcast_subscriber(inner: Arc<Inner>, mut shutdown_rx: watch::Rec
     });
 }
 
+/// Delay before the next subscribe attempt: 1 s, 2 s, 4 s ... capped at 64 s.
+fn subscribe_retry_delay(retry_count: u32) -> Duration {
+    Duration::from_secs(1u64 << retry_count.min(6))
+}
+
 fn log_event(event: &BroadcastEvent) -> trc::Value {
     match event {
         BroadcastEvent::PushNotification(notification) => match notification {
@@ -296,3 +301,19 @@ fn log_event(event: &BroadcastEvent) -> trc::Value {
         BroadcastEvent::QueueRefresh => "QueueRefresh".into(),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::subscribe_retry_delay;
+    use std::time::Duration;
+
+    #[test]
+    fn subscribe_retry_backoff_grows_then_caps() {
+        let schedule: Vec<u64> = (0..10)
+            .map(|n| subscribe_retry_delay(n).as_secs())
+            .collect();
+        assert_eq!(schedule, vec![1, 2, 4, 8, 16, 32, 64, 64, 64, 64]);
+        // No shift overflow at the top of the range.
+        assert_eq!(subscribe_retry_delay(u32::MAX), Duration::from_secs(64));
+    }
+}

crates/services/src/task_manager/index.rs

@@ -567,20 +567,14 @@ async fn build_contact_document(
 }
 
 
-// inbuxa: MON-16: a trace's search document, when trace search is on:
+// inbuxa: MON-16: a trace's search document, when trace search is on
-// its event types, queue ids, and addresses, their domains, hosts, IPs,
-// message ids and account names as keywords
 async fn build_tracing_span_document(
     server: &Server,
     span_id: u64,
 ) -> trc::Result<Option<IndexDocument>> {
     use common::telemetry::tracers::store::MaybeTrace;
-    use registry::schema::{enums::SearchTracingField, structs::Search};
+    use registry::schema::structs::Search;
-    use store::{
+    use store::write::{TelemetryClass, ValueClass};
-        search::TracingSearchField,
-        write::{TelemetryClass, ValueClass},
-    };
-    use trc::Key;
 
     let settings = server
         .registry()
@@ -590,7 +584,6 @@ async fn build_tracing_span_document(
     if !settings.index_telemetry {
         return Ok(None);
     }
-    let wants = |field: SearchTracingField| settings.index_tracing_fields.iter().any(|f| *f == field);
     let Some(MaybeTrace(Some(trace))) = server
         .tracing_store()
         .get_value::<MaybeTrace>(ValueKey::from(ValueClass::Telemetry(TelemetryClass::Span(
@@ -601,23 +594,67 @@ async fn build_tracing_span_document(
         return Ok(None);
     };
 
+    Ok(Some(trace_search_document(
+        span_id,
+        &trace,
+        &settings
+            .index_tracing_fields
+            .iter()
+            .copied()
+            .collect::<Vec<_>>(),
+    )))
+}
+
+/// inbuxa: MON-16: the search document for a stored trace.
+///
+/// The event type and queue id columns are integers on every search backend
+/// (BIGINT on PostgreSQL and MySQL, long on Elasticsearch), and each holds a
+/// single value per trace: the event type is the trace's opening event, the
+/// one `x:Trace/query` filters on, and the queue id is the first queue id the
+/// trace mentions. Every queue id also goes into the keywords, so a session
+/// that queued several messages is found by any of them.
+pub fn trace_search_document(
+    span_id: u64,
+    trace: &registry::schema::structs::Trace,
+    fields: &[registry::schema::enums::SearchTracingField],
+) -> IndexDocument {
+    use registry::schema::{enums::SearchTracingField, structs::TraceValue};
+    use store::search::TracingSearchField;
+    use trc::Key;
+
+    let wants = |field: SearchTracingField| fields.contains(&field);
     let mut document = IndexDocument::new(SearchIndex::Tracing).with_id(span_id);
+    if wants(SearchTracingField::EventType)
+        && let Some(first) = trace.events.iter().next()
+    {
+        document.index_unsigned(TracingSearchField::EventType, first.event.to_id() as u64);
+    }
+
     let mut seen = store::ahash::AHashSet::new();
+    let mut queue_id_indexed = false;
     for event in trace.events.iter() {
-        if wants(SearchTracingField::EventType) && seen.insert(event.event.as_str().to_string()) {
-            document.index_keyword(TracingSearchField::EventType, event.event.as_str());
-        }
         for kv in event.key_values.iter() {
             let text = match &kv.value {
-                registry::schema::structs::TraceValue::String(v) => v.value.clone(),
+                TraceValue::String(v) => v.value.clone(),
-                registry::schema::structs::TraceValue::UnsignedInt(v) => v.value.to_string(),
+                TraceValue::UnsignedInt(v) => v.value.to_string(),
-                registry::schema::structs::TraceValue::IpAddr(v) => v.value.to_string(),
+                TraceValue::IpAddr(v) => v.value.to_string(),
                 _ => continue,
             };
             match kv.key {
-                Key::QueueId if wants(SearchTracingField::QueueId) => {
+                Key::QueueId => {
-                    if seen.insert(format!("q:{text}")) {
+                    let Ok(queue_id) = text.parse::<u64>() else {
-                        document.index_keyword(TracingSearchField::QueueId, &text);
+                        continue;
+                    };
+                    if wants(SearchTracingField::QueueId) && !queue_id_indexed {
+                        document.index_unsigned(TracingSearchField::QueueId, queue_id);
+                        queue_id_indexed = true;
+                    }
+                    if wants(SearchTracingField::Keywords) && seen.insert(format!("k:{text}")) {
+                        document.index_text(
+                            TracingSearchField::Keywords,
+                            &text,
+                            nlp::language::Language::None,
+                        );
                     }
                 }
                 Key::From
@@ -648,7 +685,7 @@ async fn build_tracing_span_document(
             }
         }
     }
-    Ok(Some(document))
+    document
 }
 
 // inbuxa: UD-1, UD-4: archives a deleted file, event or contact noted at

docs/spec/features/monitoring.md

@@ -212,10 +212,15 @@ unchanged.
 - **MON-16.** With `indexTelemetry` on, storing a trace schedules an
   `IndexTrace` task. The task builds one document for `SearchIndex::Tracing`
   with the fields named in `indexTracingFields`:
-  - `eventType`: every event type in the trace;
+  - `eventType`: the trace's opening event, as its numeric id;
-  - `queueId`: every `queueId` value;
+  - `queueId`: the first `queueId` value, as an integer;
   - `keywords`: every address in `from` and `to`, each address's domain, every
-    `domain`, `hostname`, `remoteIp`, `messageId` and `accountName` value.
+    `domain`, `hostname`, `remoteIp`, `messageId` and `accountName` value,
+    and every `queueId` value.
+  The event type and queue id are single integer columns on every search
+  backend (BIGINT on PostgreSQL and MySQL), so the `queueId` filter matches
+  the column or any queue id in the keywords, and a session that queued
+  several messages is found by each of them.
   So searching `example.org` finds every trace to or from that domain, as the
   upstream suite expects. With `indexTelemetry` off nothing is indexed, and
   the `text` and `queueId` filters are refused (see "Interfaces").

tests/src/store/import_export.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::utils::{
@@ -9,14 +11,22 @@ use crate::utils::{
     server::TestServer,
     temp_dir::TempDir,
 };
-use ::registry::schema::enums::CompressionAlgo;
+use ::registry::schema::{
+    enums::{CompressionAlgo, TaskStoreMaintenanceType},
+    prelude::ObjectType,
+    structs::Task,
+};
 use ahash::AHashSet;
-use common::{DATABASE_SCHEMA_VERSION, manager::backup::BackupParams};
+use common::{
+    DATABASE_SCHEMA_VERSION,
+    manager::{SPAM_CLASSIFIER_KEY, SPAM_TRAINER_KEY, backup::BackupParams},
+};
 use store::{
     rand,
     write::{
         AnyClass, AnyKey, BatchBuilder, BlobLink, BlobOp, Operation, QueueClass, QueueEvent,
-        RegistryClass, ValueClass, key::KeySerializer,
+        RegistryClass, TaskQueueClass, ValueClass,
+        key::{DeserializeBigEndian, KeySerializer},
     },
     *,
 };
@@ -167,6 +177,50 @@ pub async fn test(test: &TestServer) {
     }
     db.write(batch.build_all()).await.unwrap();
 
+    // inbuxa: registry objects kept outside the registry subspace (archived
+    // items for undelete, spam training samples, directory entries) and the
+    // fork's own subspace. Exports used to leave the first two behind.
+    println!("Creating archived items, spam samples and fork data...");
+    let mut batch = BatchBuilder::new();
+    for item_id in [1u64, 2, 3] {
+        for object in [
+            ObjectType::ArchivedItem,
+            ObjectType::SpamTrainingSample,
+            ObjectType::Account,
+        ] {
+            batch.set(
+                ValueClass::Registry(RegistryClass::Item {
+                    object_id: object as u16,
+                    item_id,
+                }),
+                random_bytes(item_id as usize * 64),
+            );
+        }
+        batch.set(
+            ValueClass::Any(AnyClass {
+                subspace: SUBSPACE_INBUXA,
+                key: [b'U', b'x']
+                    .into_iter()
+                    .chain(item_id.to_be_bytes())
+                    .collect(),
+            }),
+            random_bytes(32),
+        );
+    }
+    db.write(batch.build_all()).await.unwrap();
+
+    // inbuxa: the trained spam classifier lives in blobs with fixed names
+    let mut named_blobs = Vec::new();
+    for key in [SPAM_CLASSIFIER_KEY, SPAM_TRAINER_KEY] {
+        let data = random_bytes(4096);
+        test.server
+            .blob_store()
+            .put_blob(key, &data, CompressionAlgo::Lz4)
+            .await
+            .unwrap();
+        named_blobs.push((key, data));
+    }
+
     // Create directory data
     println!("Creating directory data...");
     let mut batch = BatchBuilder::new();
@@ -185,6 +239,17 @@ pub async fn test(test: &TestServer) {
     println!("Calculating store hash...");
     let snapshot = Snapshot::new(&db).await;
     assert!(!snapshot.keys.is_empty(), "Store hash counts are empty",);
+    for subspace in [
+        SUBSPACE_DELETED_ITEMS,
+        SUBSPACE_SPAM_SAMPLES,
+        SUBSPACE_INBUXA,
+    ] {
+        assert!(
+            snapshot.keys.iter().any(|k| k.subspace == subspace),
+            "No test data in subspace {}",
+            char::from(subspace)
+        );
+    }
 
     // Export store
     println!("Exporting store...");
@@ -210,22 +275,188 @@ pub async fn test(test: &TestServer) {
             .finalize(),
     );
     db.write(batch.build_all()).await.unwrap();
-    test.server.core.restore(temp_dir.path.clone()).await;
+    for (key, _) in &named_blobs {
+        test.server.blob_store().delete_blob(key).await.unwrap();
+    }
+    let imported = test.server.core.restore(temp_dir.path.clone()).await;
     let mut batch = BatchBuilder::new();
     batch.clear(ValueClass::NodeId(0));
     db.write(batch.build_all()).await.unwrap();
+    for subspace in [
+        SUBSPACE_DELETED_ITEMS,
+        SUBSPACE_SPAM_SAMPLES,
+        SUBSPACE_INBUXA,
+    ] {
+        assert!(
+            imported.contains(&subspace),
+            "Subspace {} was not exported",
+            char::from(subspace)
+        );
+    }
 
     // Verify hash
     print!("Verifying store hash...");
     snapshot.assert_is_eq(&Snapshot::new(&db).await);
+    assert_named_blobs(test.server.blob_store(), &named_blobs).await;
     println!(" GREAT SUCCESS!");
 
+    // inbuxa: import the same export into a fresh store of another backend,
+    // the way a move from one database to another does it
+    #[cfg(all(feature = "rocks", feature = "sqlite"))]
+    cross_backend(test, &db, &temp_dir, &named_blobs).await;
+
     // Destroy store
+    for (key, _) in &named_blobs {
+        test.server.blob_store().delete_blob(key).await.unwrap();
+    }
     store_destroy(&db).await;
     store_assert_is_empty(&db, db.clone().into(), true).await;
     temp_dir.delete();
 }
 
+#[cfg(all(feature = "rocks", feature = "sqlite"))]
+async fn cross_backend(
+    test: &TestServer,
+    source: &Store,
+    export: &TempDir,
+    named_blobs: &[(&[u8], Vec<u8>)],
+) {
+    let source_type = std::env::var("STORE").unwrap();
+    let target_type = if source_type.eq_ignore_ascii_case("sqlite") {
+        "RocksDb"
+    } else {
+        "Sqlite"
+    };
+    println!("Importing the export into a fresh {target_type} store...");
+
+    let target_dir = TempDir::new("art_vandelay_cross_backend", true);
+    let target = Store::build(
+        crate::utils::storage::build_data_store(target_type, &target_dir.path.to_string_lossy())
+            .await,
+    )
+    .await
+    .unwrap();
+    target.create_tables().await.unwrap();
+    store_destroy(&target).await;
+
+    let mut core = test.server.core.as_ref().clone();
+    core.storage.data = target.clone();
+    core.storage.blob = target.clone().into();
+    let imported = core.restore(export.path.clone()).await;
+
+    // Counters are stored differently by the SQL and key-value backends, so
+    // compare their keys here and their values through the counter API.
+    print!("Verifying {target_type} store hash...");
+    Snapshot::new_portable(source)
+        .await
+        .assert_is_eq(&Snapshot::new_portable(&target).await);
+    for subspace in [SUBSPACE_COUNTER, SUBSPACE_QUOTA] {
+        let mut keys = Vec::new();
+        source
+            .iterate(
+                IterateParams::new(
+                    AnyKey {
+                        subspace,
+                        key: vec![0u8],
+                    },
+                    AnyKey {
+                        subspace,
+                        key: vec![u8::MAX; 10],
+                    },
+                )
+                .no_values(),
+                |key, _| {
+                    keys.push(key.to_vec());
+                    Ok(true)
+                },
+            )
+            .await
+            .unwrap();
+        for key in keys {
+            let class = || {
+                ValueClass::Any(AnyClass {
+                    subspace,
+                    key: key.clone(),
+                })
+            };
+            assert_eq!(
+                source.get_counter(class()).await.unwrap(),
+                target.get_counter(class()).await.unwrap(),
+                "Counter mismatch in {} for {key:?}",
+                char::from(subspace)
+            );
+        }
+    }
+    assert_named_blobs(&core.storage.blob, named_blobs).await;
+    println!(" GREAT SUCCESS!");
+
+    // The search index isn't exported; the import queues its rebuild
+    let queued = core.queue_reindex(&imported).await;
+    let expected = [
+        TaskStoreMaintenanceType::ReindexAccounts,
+        TaskStoreMaintenanceType::ReindexTelemetry,
+    ];
+    assert_eq!(queued, expected);
+    let mut task_ids = Vec::new();
+    target
+        .iterate(
+            IterateParams::new(
+                AnyKey {
+                    subspace: SUBSPACE_TASK_QUEUE,
+                    key: vec![0u8],
+                },
+                AnyKey {
+                    subspace: SUBSPACE_TASK_QUEUE,
+                    key: vec![u8::MAX; 20],
+                },
+            )
+            .no_values(),
+            |key, _| {
+                if key.deserialize_be_u64(0)? == 0 {
+                    task_ids.push(key.deserialize_be_u64(U64_LEN)?);
+                }
+                Ok(true)
+            },
+        )
+        .await
+        .unwrap();
+    let mut found = Vec::new();
+    for id in task_ids {
+        match target
+            .get_value::<Task>(ValueKey::from(ValueClass::TaskQueue(
+                TaskQueueClass::Task { id },
+            )))
+            .await
+            .unwrap()
+        {
+            Some(Task::StoreMaintenance(task)) => found.push(task.maintenance_type),
+            other => panic!("Unexpected task {other:?}"),
+        }
+    }
+    found.sort_by_key(|t| *t as u16);
+    assert_eq!(found, expected, "Queued tasks don't match");
+
+    store_destroy(&target).await;
+    drop(core);
+    drop(target);
+    target_dir.delete();
+}
+
+async fn assert_named_blobs(blob_store: &BlobStore, named_blobs: &[(&[u8], Vec<u8>)]) {
+    for (key, data) in named_blobs {
+        assert_eq!(
+            blob_store
+                .get_blob(key, 0..usize::MAX)
+                .await
+                .unwrap()
+                .as_ref(),
+            Some(data),
+            "Blob {} was not restored",
+            String::from_utf8_lossy(key)
+        );
+    }
+}
+
 #[derive(Debug, PartialEq, Eq)]
 struct Snapshot {
     keys: AHashSet<KeyValue>,
@@ -240,7 +471,19 @@ struct KeyValue {
 
 impl Snapshot {
     async fn new(db: &Store) -> Self {
-        let is_sql = db.is_sql();
+        Self::build(db, !db.is_sql(), true).await
+    }
+
+    /// Comparable across backends: no counter values, which the SQL and
+    /// key-value stores encode differently, and no blobs, which only live in
+    /// the data store when it doubles as the blob store.
+    #[cfg(all(feature = "rocks", feature = "sqlite"))]
+    async fn new_portable(db: &Store) -> Self {
+        Self::build(db, false, false).await
+    }
+
+    async fn build(db: &Store, counter_values: bool, with_blobs: bool) -> Self {
+        let is_sql = !counter_values;
 
         let mut keys = AHashSet::new();
 
@@ -265,7 +508,12 @@ impl Snapshot {
             (SUBSPACE_QUOTA, !is_sql),
             (SUBSPACE_REPORT_OUT, true),
             (SUBSPACE_REPORT_IN, true),
+            (SUBSPACE_DIRECTORY, true),
+            (SUBSPACE_INBUXA, true),
         ] {
+            if subspace == SUBSPACE_BLOBS && !with_blobs {
+                continue;
+            }
             let from_key = AnyKey {
                 subspace,
                 key: vec![0u8],

tests/src/store/query.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::{store::deflate_test_resource, utils::server::TestServer};
@@ -122,6 +124,10 @@ pub async fn test(test: &TestServer) {
     println!("Running global id filtering tests...");
     test_global(store.clone()).await;
 
+    // inbuxa: trace documents as the index task builds them
+    println!("Running trace document tests...");
+    test_trace_documents(store.clone()).await;
+
     // Large document insert test
     println!("Running large document insert tests...");
     let mut large_text = String::with_capacity(20 * 1024 * 1024);
@@ -809,3 +815,160 @@ async fn test_global(store: SearchStore) {
         AHashSet::from_iter([3, 4, 5])
     );
 }
+
+// inbuxa: MON-16: documents built by the index task from stored traces go
+// into every search backend (the SQL backends type etyp and qid as BIGINT)
+// and are found again by queue id and keyword.
+async fn test_trace_documents(store: SearchStore) {
+    use registry::schema::{
+        enums::SearchTracingField,
+        structs::{
+            Trace, TraceEvent, TraceKeyValue, TraceValue, TraceValueString,
+            TraceValueUnsignedInt,
+        },
+    };
+    use services::task_manager::index::trace_search_document;
+    use trc::{DeliveryEvent, EventType, Key, SmtpEvent};
+
+    let kv_u = |key: Key, value: u64| TraceKeyValue {
+        key,
+        value: TraceValue::UnsignedInt(TraceValueUnsignedInt { value }),
+    };
+    let kv_s = |key: Key, value: &str| TraceKeyValue {
+        key,
+        value: TraceValue::String(TraceValueString {
+            value: value.to_string(),
+        }),
+    };
+    let event = |event: EventType, key_values: Vec<TraceKeyValue>| TraceEvent {
+        event,
+        key_values: key_values.into(),
+        ..Default::default()
+    };
+    let fields = [
+        SearchTracingField::EventType,
+        SearchTracingField::QueueId,
+        SearchTracingField::Keywords,
+    ];
+
+    // An SMTP session that queued two messages, and a delivery attempt
+    let session = Trace {
+        events: vec![
+            event(
+                EventType::Smtp(SmtpEvent::ConnectionStart),
+                vec![kv_s(Key::RemoteIp, "192.0.2.7")],
+            ),
+            event(
+                EventType::Smtp(SmtpEvent::MailFrom),
+                vec![kv_s(Key::From, "[email protected]")],
+            ),
+            event(
+                EventType::Smtp(SmtpEvent::RcptTo),
+                vec![kv_u(Key::QueueId, 9_000_000_001), kv_s(Key::To, "[email protected]")],
+            ),
+            event(
+                EventType::Smtp(SmtpEvent::RcptTo),
+                vec![kv_u(Key::QueueId, 9_000_000_002)],
+            ),
+        ]
+        .into(),
+    };
+    let delivery = Trace {
+        events: vec![event(
+            EventType::Delivery(DeliveryEvent::AttemptStart),
+            vec![kv_u(Key::QueueId, 9_000_000_003), kv_s(Key::Hostname, "relay.example.net")],
+        )]
+        .into(),
+    };
+    let documents = vec![
+        trace_search_document(100, &session, &fields),
+        trace_search_document(101, &delivery, &fields),
+    ];
+    assert!(
+        documents
+            .iter()
+            .all(|d| d.has_field(&SearchField::Tracing(TracingSearchField::QueueId))
+                && d.has_field(&SearchField::Tracing(TracingSearchField::EventType))),
+        "trace documents carry a queue id and an event type"
+    );
+    store.index(documents).await.unwrap();
+    if let SearchStore::ElasticSearch(store) = &store {
+        store.refresh_index(SearchIndex::Tracing).await.unwrap();
+    }
+
+    let query = |filters: Vec<SearchFilter>| {
+        let store = store.clone();
+        async move {
+            store
+                .query_global(
+                    SearchQuery::new(SearchIndex::Tracing)
+                        .with_filter(SearchFilter::ge(SearchField::Id, 100u64))
+                        .with_filters(filters),
+                )
+                .await
+                .unwrap()
+                .into_iter()
+                .collect::<AHashSet<_>>()
+        }
+    };
+    // By queue id, the way x:Trace/query asks: the queue id column, or any
+    // queue id in the keywords
+    let by_queue_id = |queue_id: u64| {
+        vec![
+            SearchFilter::Or,
+            SearchFilter::eq(TracingSearchField::QueueId, queue_id),
+            SearchFilter::has_text(
+                TracingSearchField::Keywords,
+                queue_id.to_string(),
+                Language::None,
+            ),
+            SearchFilter::End,
+        ]
+    };
+    assert_eq!(query(by_queue_id(9_000_000_001)).await, AHashSet::from_iter([100]));
+    assert_eq!(query(by_queue_id(9_000_000_002)).await, AHashSet::from_iter([100]));
+    assert_eq!(query(by_queue_id(9_000_000_003)).await, AHashSet::from_iter([101]));
+    assert_eq!(query(by_queue_id(9_000_000_004)).await, AHashSet::new());
+    assert_eq!(
+        query(vec![SearchFilter::eq(TracingSearchField::QueueId, 9_000_000_003u64)]).await,
+        AHashSet::from_iter([101])
+    );
+    // By opening event type
+    assert_eq!(
+        query(vec![SearchFilter::eq(
+            TracingSearchField::EventType,
+            EventType::Delivery(DeliveryEvent::AttemptStart).to_id() as u64,
+        )])
+        .await,
+        AHashSet::from_iter([101])
+    );
+    // By keyword: an address, lowercased, and its domain
+    assert_eq!(
+        query(vec![SearchFilter::has_text(
+            TracingSearchField::Keywords,
+            "example.org",
+            Language::None,
+        )])
+        .await,
+        AHashSet::from_iter([100])
+    );
+    assert_eq!(
+        query(vec![SearchFilter::has_text(
+            TracingSearchField::Keywords,
+            "relay.example.net",
+            Language::None,
+        )])
+        .await,
+        AHashSet::from_iter([101])
+    );
+
+    for id in [100u64, 101] {
+        store
+            .unindex(
+                SearchQuery::new(SearchIndex::Tracing)
+                    .with_filter(SearchFilter::eq(SearchField::Id, id)),
+            )
+            .await
+            .unwrap();
+    }
+}

tests/src/system/monitoring.rs

@@ -148,6 +148,73 @@ pub async fn test(test: &mut TestServer) {
         "test 9: to"
     );
 
+    // MON-16: the queueId filter finds the traces that name a queue id (the
+    // session that queued the message and its delivery attempt) through the
+    // search index, given as a string or a number (the index column is an
+    // integer)
+    fn queue_ids(value: &Value, out: &mut Vec<u64>) {
+        match value {
+            Value::Object(map) => {
+                if map.get("key").and_then(|k| k.as_str()) == Some("queueId")
+                    && let Some(id) = map
+                        .get("value")
+                        .and_then(|v| v.get("value").unwrap_or(v).as_u64())
+                {
+                    out.push(id);
+                }
+                map.values().for_each(|v| queue_ids(v, out));
+            }
+            Value::Array(list) => list.iter().for_each(|v| queue_ids(v, out)),
+            _ => {}
+        }
+    }
+    let with_ids = traces
+        .iter()
+        .map(|t| {
+            let mut ids = Vec::new();
+            queue_ids(t, &mut ids);
+            (t["id"].as_str().unwrap().to_string(), ids)
+        })
+        .collect::<Vec<_>>();
+    let queue_id = with_ids
+        .iter()
+        .find_map(|(_, ids)| ids.first().copied())
+        .expect("MON-16: a trace with a queue id");
+    let mut expected = with_ids
+        .iter()
+        .filter(|(_, ids)| ids.contains(&queue_id))
+        .map(|(id, _)| id.clone())
+        .collect::<Vec<_>>();
+    expected.sort();
+    for filter in [json!(queue_id.to_string()), json!(queue_id)] {
+        let response = admin
+            .jmap_method_call("x:Trace/query", json!({"filter": {"queueId": filter}}))
+            .await;
+        let mut found = response
+            .0
+            .pointer("/methodResponses/0/1/ids")
+            .and_then(|ids| ids.as_array())
+            .map(|ids| {
+                ids.iter()
+                    .filter_map(|id| id.as_str().map(str::to_string))
+                    .collect::<Vec<_>>()
+            })
+            .unwrap_or_default();
+        found.sort();
+        assert_eq!(found, expected, "MON-16: queueId {filter}: {response:?}");
+    }
+    let response = admin
+        .jmap_method_call(
+            "x:Trace/query",
+            json!({"filter": {"queueId": (queue_id ^ 0x5a5a_5a5a).to_string()}}),
+        )
+        .await;
+    assert_eq!(
+        response.0.pointer("/methodResponses/0/1/ids"),
+        Some(&json!([])),
+        "MON-16: an unknown queue id"
+    );
+
     // Acceptance test 24: destroy removes a trace; create is refused
     let trace_id = traces[0]["id"].as_str().unwrap().to_string();
     let response = admin