Merge pull request 'Hotfix 2026.9.24.4: report reschedules keep the task queue readable' (#49) from hotfix/2026.9.24.4 into release/2026.9.24.4

Setting deliverAt on an internal DMARC or TLS report wrote the new task
queue row with the report's object type (0x21, 0x6e) instead of the task
type (7, 8), and left the task row at its old due. The task manager's scan
failed on that row with store.data-corruption ("Failed to iterate over task
queue"), and because the error ended the whole scan, every task due after
the row stopped running on every node.

- reschedule_ops writes the new queue row through schedule_task_with_id, so
  it carries the task type and the task row gets the new due. It removes
  the row the task is actually queued under (the task's due, which differs
  from deliverAt once the task has been retried) and any row an earlier
  reschedule left at deliverAt.
- x:DmarcInternalReport/set and x:TlsInternalReport/set lock the report's
  task while they move it, as x:Task/set does, refuse while the report is
  being sent, release the locks however the request ends, and wake the task
  manager.
- The task manager logs a queue row it can't read (id, due, key, value) and
  skips it instead of ending the scan. It then repairs the row from its task:
  the row is rewritten with the task's type, and a row with no task behind
  it is removed. A row holding a report's object type for a report task is
  what the old reschedule wrote: the task is moved to that row's time, as
  the reschedule intended, and its old queue row is removed. Stores that
  already hold such a row recover on their own once it comes due.
- x:Task/query with a type filter skips an unreadable row instead of
  failing.

Test: smtp::reporting::reschedule (RocksDB and PostgreSQL). It fails on
main: x:Task/get shows the old due, and with that check removed, neither
report nor a later task ever runs.

(cherry picked from commit 1a7859a8cc)

.dockerignore

@@ -1,10 +1,16 @@
-// Ignore everything
+# Ignore everything
 *
 
-// Allow what is needed
+# Allow what is needed
 !crates
 !tests
 !resources
 
+# The patched dependency Cargo.toml's [patch.crates-io] points at. Without
+# it the build context has no vendor/, and `cargo chef cook` fails on
+# "failed to load source for dependency sieve-rs" -- which CI cannot see,
+# because CI builds from a checkout and only the image build has a context.
+!vendor
+
 !Cargo.lock
 !Cargo.toml

.gitea/workflows/ci.yml

@@ -35,6 +35,11 @@ jobs:
       - run: python3 tools/fork/name-check.py
       - if: always()
         run: python3 tools/fork/notice-check.py
+      # Cargo can patch a dependency to a directory in this repository, and
+      # the image builds from a context .dockerignore prunes to almost
+      # nothing. CI never sees the difference; a release does.
+      - if: always()
+        run: python3 tools/fork/context-check.py
 
   build:
     # Either runner (host1 or host2): the build needs no docker socket.

.gitea/workflows/publish.yml

@@ -14,8 +14,11 @@
 #     crates/types/src/branding.rs, not Cargo.toml, and the image is tagged
 #     with it, so a tag beside an unbumped macro would publish an image that
 #     reports a different version from its tag.
-#   * the tag must be on main, so an image never describes code that was never
+#   * the tag must be on main or on a release/* branch, so an image never
-#     reviewed onto the default branch.
+#     describes code that was never reviewed onto one of them. A release/*
+#     branch carries a hotfix: it starts at an earlier release tag, takes
+#     fixes through pull requests into it, and is tagged there, so production
+#     can get a fix without everything that has landed on main since.
 #
 # :latest moves with every published tag: tags are cut by the weekly release
 # (or by hand for a real release); there are no prerelease tags here.
@@ -57,8 +60,13 @@ jobs:
             echo "Refusing to publish an image that would report the wrong version." >&2
             exit 1
           fi
-          git merge-base --is-ancestor "$(git rev-parse "${TAG}^{commit}")" origin/main \
+          commit="$(git rev-parse "${TAG}^{commit}")"
-            || { echo "$TAG is not on main" >&2; exit 1; }
+          on=""
+          for ref in origin/main $(git for-each-ref --format='%(refname:short)' 'refs/remotes/origin/release/*'); do
+            if git merge-base --is-ancestor "$commit" "$ref"; then on="$ref"; break; fi
+          done
+          [ -n "$on" ] || { echo "$TAG is not on main or a release/* branch" >&2; exit 1; }
+          echo "$TAG is on $on"
           echo "version=$V" >> "$GITHUB_OUTPUT"
           echo "version $V"
 

.gitea/workflows/upstream-watch.yml

@@ -12,6 +12,9 @@
 # An issue is opened once per release: an existing one with the same title,
 # open or closed, stops a second.
 #
+# It also watches spam-filter, whose rules the server bundles
+# (resources/spam-filter/), and opens an issue for a newer release.
+#
 # Daily 06:17 UTC; run it by hand with workflow_dispatch.
 name: upstream-watch
 
@@ -64,7 +67,7 @@ jobs:
                           and key(r["tag_name"]) > key(base)),
                          key=lambda r: key(r["tag_name"]))
           if not newer:
-              print(f"Up to date: {base} is the newest upstream release."); sys.exit(0)
+              print(f"Up to date: {base} is the newest upstream release.")
 
           # Titles and bodies stay free of the upstream project's name, as the
           # rest of the fork's user-visible text does.
@@ -85,4 +88,35 @@ jobs:
                       "add any new third-party notices to `THIRD-PARTY.md`, then merge `upstream` into `main`.")
               issue = call("POST", f"{api}/issues", {"title": title, "body": body})
               print(f"{tag}: opened #{issue['number']}.")
+
+          # The spam filter rules bundled with the server (resources/spam-filter/):
+          # an issue when spam-filter publishes a newer release than the one
+          # BUNDLED_SPAM_RULES_VERSION names on main.
+          src = call("GET", f"{api}/contents/crates/common/src/manager/spam_rules.rs?ref=main")
+          import base64
+          text = base64.b64decode(src["content"]).decode()
+          m = re.search(r'BUNDLED_SPAM_RULES_VERSION: &str = "(\d+\.\d+\.\d+)"', text)
+          if not m:
+              print("Can't read BUNDLED_SPAM_RULES_VERSION from spam_rules.rs", file=sys.stderr); sys.exit(1)
+          bundled = "v" + m.group(1)
+          rels = call("GET", "https://api.github.com/repos/stalwartlabs/spam-filter/releases?per_page=30", token=None)
+          newer = sorted((r for r in rels
+                          if not r["draft"] and not r["prerelease"] and SEMVER.match(r["tag_name"])
+                          and key(r["tag_name"]) > key(bundled)),
+                         key=lambda r: key(r["tag_name"]))
+          if not newer:
+              print(f"Up to date: the bundled spam rules are {bundled}, the newest release."); sys.exit(0)
+          latest = newer[-1]
+          tag = latest["tag_name"]
+          title = f"Update the bundled spam rules to {tag}"
+          existing = {i["title"] for i in call("GET", f"{api}/issues?state=all&type=issues&q=bundled+spam+rules&limit=50")}
+          if title in existing:
+              print(f"spam rules {tag}: issue already exists."); sys.exit(0)
+          body = (f"spam-filter published {tag} on {latest['published_at'][:10]}. "
+                  f"The server bundles {bundled}.\n\n"
+                  "Update it as resources/spam-filter/README.md describes: take the rules file "
+                  f"from the {tag} release (by tag, not `latest`), set BUNDLED_SPAM_RULES_VERSION, "
+                  "and run the antispam test.")
+          issue = call("POST", f"{api}/issues", {"title": title, "body": body})
+          print(f"spam rules {tag}: opened #{issue['number']}.")
           PY

Dockerfile

@@ -19,6 +19,10 @@ RUN export DEBIAN_FRONTEND=noninteractive && \
     g++-x86-64-linux-gnu binutils-x86-64-linux-gnu
 RUN rustup target add "$(cat /target.txt)"
 COPY --from=planner /recipe.json /recipe.json
+# inbuxa: [patch.crates-io] points sieve-rs at vendor/, and the recipe only
+# carries the workspace's own manifests, so cooking the dependencies needs the
+# vendored crate itself (the context allows it since #27; this puts it here).
+COPY vendor/ vendor/
 RUN RUSTFLAGS="$(cat /flags.txt)" cargo chef cook --target "$(cat /target.txt)" --release --no-default-features --features "sqlite postgres mysql rocks s3 redis azure nats" --recipe-path /recipe.json
 COPY . .
 RUN RUSTFLAGS="$(cat /flags.txt)" cargo build --target "$(cat /target.txt)" --release -p inbuxa --no-default-features --features "sqlite postgres mysql rocks s3 redis azure nats"

THIRD-PARTY.md

@@ -24,7 +24,7 @@ carry their own license files.
 | `crates/common/src/network/acme/directory.rs`, `crates/common/src/network/acme/jose.rs`, `crates/common/src/network/acme/order.rs` | [rustls-acme](https://github.com/FlorianUekermann/rustls-acme) (MIT or Apache-2.0) | Copyright (c) Florian Uekermann |
 | `crates/types/src/id.rs` | [crockford](https://github.com/archer884/crockford) (MIT or Apache-2.0) | Copyright (c) 2017 J/A <archer884@gmail.com> |
 | `crates/nlp/src/tokenizers/types.rs` | test cases from [linkify](https://github.com/robinst/linkify) (MIT or Apache-2.0) | Copyright (c) 2017 Robin Stocker |
-| `tests/resources/smtp/antispam/spam-filter-rules.json.gz` | the published rules of [spam-filter](https://github.com/stalwartlabs/spam-filter) v3.0.2, unmodified, for the spam filter's tests (MIT or Apache-2.0) | Copyright (C) 2024, Stalwart Labs LLC |
+| `resources/spam-filter/spam-filter-rules.json.gz` | the published rules of [spam-filter](https://github.com/stalwartlabs/spam-filter) v3.0.2, unmodified, built into the server as its default spam rules (MIT or Apache-2.0) | Copyright (C) 2024, Stalwart Labs LLC |
 
 Each notice above applies with this permission notice:
 

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

@@ -243,7 +243,8 @@ impl SpamFilterConfig {
                 spam_threshold: spam.score_spam.into_inner() as f32,
             },
             grey_list_expiry: spam.greylist_for.map(|d| d.into_inner().as_secs()),
-            spam_rules_url: spam.spam_filter_rules_url,
+            // inbuxa: unset, empty or upstream's old default means the bundled rules
+            spam_rules_url: crate::manager::spam_rules::rules_url(spam.spam_filter_rules_url),
             url_client: utils::http::http_client_builder(true)
                 .pool_max_idle_per_host(0)
                 .redirect(reqwest::redirect::Policy::none())

crates/common/src/manager/defaults.rs

@@ -530,13 +530,22 @@ async fn insert_safe_defaults(bp: &mut Bootstrap) -> trc::Result<()> {
         use store::write::BatchBuilder;
         use types::id::Id;
 
-        if bp.registry.count_object(ObjectType::SpamRule).await? == 0
+        // inbuxa: rules are always to hand, since a copy ships with the server
-            && bp
+        // (spam_rules). They load on first boot, and again when the bundled
-                .registry
+        // version differs from the one last loaded, which only adds what's
+        // missing: new tags and rules, never a changed score.
+        let rules_url = super::spam_rules::rules_url(
+            bp.registry
                 .object::<SpamSettings>(Id::singleton())
                 .await?
-                .is_none_or(|spam| spam.spam_filter_rules_url.is_some())
+                .and_then(|spam| spam.spam_filter_rules_url),
-        {
+        );
+        let bundled_is_new = rules_url.is_none()
+            && super::spam_rules::applied_version(&bp.data_store)
+                .await?
+                .as_deref()
+                != Some(super::spam_rules::BUNDLED_SPAM_RULES_VERSION);
+        if bp.registry.count_object(ObjectType::SpamRule).await? == 0 || bundled_is_new {
             let mut batch = BatchBuilder::new();
             batch.schedule_task(Task::SpamFilterMaintenance(TaskSpamFilterMaintenance {
                 maintenance_type: TaskSpamFilterMaintenanceType::UpdateRules,

crates/common/src/manager/mod.rs

@@ -22,6 +22,7 @@ pub mod console;
 pub mod defaults;
 pub mod first_party;
 pub mod restore;
+pub mod spam_rules; // inbuxa: rules bundled with the server
 
 pub const SPAM_TRAINER_KEY: &[u8] = "INBUXA_SPAM_TRAIN_DATA.lz4".as_bytes();
 pub const SPAM_CLASSIFIER_KEY: &[u8] = "INBUXA_SPAM_CLASSIFIER_MODEL.lz4".as_bytes();

crates/common/src/manager/spam_rules.rs

@@ -0,0 +1,103 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! inbuxa: the spam filter rules that ship with the server.
+//!
+//! Upstream fetches its latest published rules from GitHub at run time, so
+//! scoring changes with a release nobody here tested and depends on reaching
+//! it. The fork embeds a pinned copy (resources/spam-filter/, with its version
+//! and license) and uses it whenever no other source is configured. The rules
+//! URL remains an operator override (`https://` or `file://`).
+//!
+//! Loading rules only ever adds what's missing, never changes an existing rule
+//! or score. They load on first boot, and again whenever the bundled version
+//! differs from the one last applied, so an upgrade brings new tags (the AI
+//! classifier's `LLM_*` scores, say) to an install that already had rules.
+
+use std::io::Read;
+use store::{
+    SUBSPACE_INBUXA, Store, ValueKey,
+    write::{AnyClass, BatchBuilder, ValueClass},
+};
+use trc::AddContext;
+
+/// The version of spam-filter the embedded rules come from.
+pub const BUNDLED_SPAM_RULES_VERSION: &str = "3.0.2";
+
+static BUNDLED_SPAM_RULES: &[u8] =
+    include_bytes!("../../../../resources/spam-filter/spam-filter-rules.json.gz");
+
+/// Upstream's default rules source, the value every install created before
+/// the rules were bundled has saved. Read only to treat it as unset.
+const LEGACY_DEFAULT_URL: &str =
+    "https://github.com/stalwartlabs/spam-filter/releases/latest/download/spam-filter-rules.json.gz";
+
+/// The URL to fetch rules from, or `None` for the bundled rules. An empty
+/// setting and upstream's old default both mean the bundled rules.
+pub fn rules_url(configured: Option<String>) -> Option<String> {
+    configured.filter(|url| !url.trim().is_empty() && url != LEGACY_DEFAULT_URL)
+}
+
+/// The bundled rules, uncompressed: the same JSON the rules URL serves.
+pub fn bundled_rules() -> Result<Vec<u8>, String> {
+    let mut json = Vec::new();
+    mail_auth::flate2::read::GzDecoder::new(BUNDLED_SPAM_RULES)
+        .read_to_end(&mut json)
+        .map_err(|err| format!("Failed to decompress the bundled spam rules: {err}"))?;
+    Ok(json)
+}
+
+fn applied_key() -> ValueClass {
+    ValueClass::Any(AnyClass {
+        subspace: SUBSPACE_INBUXA,
+        key: b"Sr".to_vec(),
+    })
+}
+
+/// The bundled version last loaded into the registry, if any.
+pub async fn applied_version(data: &Store) -> trc::Result<Option<String>> {
+    data.get_value::<String>(ValueKey::from(applied_key()))
+        .await
+        .caused_by(trc::location!())
+}
+
+/// Records that the bundled rules of this version have been loaded.
+pub async fn set_applied_version(data: &Store, version: &str) -> trc::Result<()> {
+    let mut batch = BatchBuilder::new();
+    batch.set(applied_key(), version.as_bytes().to_vec());
+    data.write(batch.build_all())
+        .await
+        .caused_by(trc::location!())
+        .map(|_| ())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn upstream_default_and_empty_mean_bundled() {
+        assert_eq!(rules_url(None), None);
+        assert_eq!(rules_url(Some(String::new())), None);
+        assert_eq!(rules_url(Some("  ".into())), None);
+        assert_eq!(rules_url(Some(LEGACY_DEFAULT_URL.into())), None);
+        assert_eq!(
+            rules_url(Some("file:///srv/rules.json.gz".into())).as_deref(),
+            Some("file:///srv/rules.json.gz")
+        );
+    }
+
+    #[test]
+    fn bundled_rules_parse_and_score_the_ai_tags() {
+        let rules: serde_json::Value = serde_json::from_slice(&bundled_rules().unwrap()).unwrap();
+        let tags = rules["SpamTag"].as_array().unwrap();
+        for (tag, score) in [("LLM_UNSOLICITED_HIGH", 3.0), ("LLM_LEGITIMATE_HIGH", -3.0)] {
+            let found = tags.iter().find(|t| t["tag"] == tag).unwrap();
+            assert_eq!(found["score"].as_f64(), Some(score), "{tag}");
+        }
+        assert!(!rules["SpamRule"].as_array().unwrap().is_empty());
+    }
+}

crates/jmap/src/registry/mapping/report.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::{
@@ -15,22 +17,42 @@ use jmap_proto::{error::set::SetError, types::state::State};
 use jmap_tools::{Key, Value};
 use registry::{
     jmap::IntoValue,
-    schema::prelude::{Object, ObjectInner, ObjectType, Property},
+    schema::{
+        prelude::{Object, ObjectInner, ObjectType, Property},
+        structs::Task,
+    },
     types::{EnumImpl, datetime::UTCDateTime},
 };
+use services::task_manager::lock::TaskLockManager;
 use smtp::reporting::index::{ExternalReportIndex, InternalReportIndex};
 use std::str::FromStr;
 use store::{
     U64_LEN, ValueKey,
     registry::{RegistryFilter, RegistryFilterValue, RegistryQuery},
-    write::{BatchBuilder, RegistryClass, ValueClass, key::KeySerializer},
+    write::{BatchBuilder, RegistryClass, TaskQueueClass, ValueClass, key::KeySerializer},
 };
 use trc::AddContext;
 use types::id::Id;
 
 pub(crate) async fn report_set(
-    mut set: RegistrySetResponse<'_>,
+    set: RegistrySetResponse<'_>,
 ) -> trc::Result<RegistrySetResponse<'_>> {
+    // inbuxa: task locks taken to reschedule reports are released however
+    // the request ends; a held lock is renewed, so a leaked one would keep
+    // the report's task from ever running
+    let server = set.server;
+    let mut locked_tasks = Vec::new();
+    let result = report_set_locked(set, &mut locked_tasks).await;
+    for task_id in locked_tasks {
+        server.remove_index_lock(task_id).await;
+    }
+    result
+}
+
+async fn report_set_locked<'x>(
+    mut set: RegistrySetResponse<'x>,
+    locked_tasks: &mut Vec<u64>,
+) -> trc::Result<RegistrySetResponse<'x>> {
     let object_id = set.object_type.to_id();
 
     // Reports cannot be created
@@ -89,12 +111,45 @@ pub(crate) async fn report_set(
                 .get_value::<Object>(ValueKey::from(key.clone()))
                 .await?
             {
+                // inbuxa: the report's task shares its id. Hold the task
+                // while its queue rows move, as x:Task/set does, and move the
+                // row the task is actually queued under
+                if !set.server.try_lock_task(item_id).await {
+                    set.response.not_updated.append(
+                        id,
+                        SetError::forbidden().with_description(
+                            "The report is being sent and cannot be rescheduled".to_string(),
+                        ),
+                    );
+                    continue;
+                }
+                locked_tasks.push(item_id);
+                let queued = set
+                    .server
+                    .store()
+                    .get_value::<Task>(ValueKey::from(ValueClass::TaskQueue(
+                        TaskQueueClass::Task { id: item_id },
+                    )))
+                    .await?;
+
                 match &mut report_obj.inner {
                     ObjectInner::DmarcInternalReport(report) => {
-                        report.reschedule_ops(&mut batch, item_id, report_obj.revision, deliver_at);
+                        report.reschedule_ops(
+                            &mut batch,
+                            item_id,
+                            report_obj.revision,
+                            deliver_at,
+                            queued.as_ref(),
+                        );
                     }
                     ObjectInner::TlsInternalReport(report) => {
-                        report.reschedule_ops(&mut batch, item_id, report_obj.revision, deliver_at);
+                        report.reschedule_ops(
+                            &mut batch,
+                            item_id,
+                            report_obj.revision,
+                            deliver_at,
+                            queued.as_ref(),
+                        );
                     }
                     _ => {}
                 }
@@ -156,6 +211,9 @@ pub(crate) async fn report_set(
             .write(batch.build_all())
             .await
             .caused_by(trc::location!())?;
+        // inbuxa: a rescheduled report may now be due sooner than the task
+        // manager's next scan
+        set.server.notify_task_queue();
     }
 
     Ok(set)

crates/jmap/src/registry/mapping/task.rs

@@ -463,15 +463,10 @@ pub(crate) async fn task_query(
                 .set_values(typ.is_some()),
             |key, value| {
                 if let Some(typ) = typ {
-                    let task_type =
+                    // inbuxa: a row whose type can't be read matches no type
-                        TaskType::from_id(value.deserialize_be_u16(0)?).ok_or_else(|| {
+                    // filter; the task manager logs and repairs it
-                            trc::StoreEvent::DataCorruption
+                    let task_type = value.deserialize_be_u16(0).ok().and_then(TaskType::from_id);
-                                .into_err()
+                    if task_type != Some(typ) {
-                                .ctx(trc::Key::Key, key.to_vec())
-                                .ctx(trc::Key::Value, value.to_vec())
-                                .caused_by(trc::location!())
-                        })?;
-                    if task_type != typ {
                         return Ok(true);
                     }
                 }

crates/registry/src/schema/structs_impl.rs

@@ -40215,7 +40215,7 @@ impl Default for SpamSettings {
             score_reject: Float::new(0.0f64),
             score_spam: Float::new(5.0f64),
             trust_replies: true,
-            spam_filter_rules_url: Some("https://github.com/stalwartlabs/spam-filter/releases/latest/download/spam-filter-rules.json.gz".to_string()),
+            spam_filter_rules_url: None,
         }
     }
 }

crates/services/src/task_manager/manager.rs

@@ -2,6 +2,8 @@
  * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
  *
  * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
+ *
+ * Modified by Coffey Labs in 2026 for INBUXA.
  */
 
 use crate::task_manager::acme::AcmeTask;
@@ -27,6 +29,7 @@ use common::network::limiter::ConcurrencyLimiter;
 use common::network::{ServerInstance, TcpAcceptor};
 use common::{Inner, Server};
 use registry::schema::enums::TaskType;
+use registry::schema::prelude::ObjectType;
 use registry::schema::structs::{
     Task, TaskManager, TaskRetryStrategy, TaskStatus, TaskStatusFailed, TaskStatusRetry,
 };
@@ -337,6 +340,7 @@ impl TaskQueueManager for Server {
 
         // Retrieve tasks pending to be processed
         let mut tasks = Vec::new();
+        let mut unreadable = Vec::new();
         let now = Instant::now();
         let mut next_event = None;
         let roles = &self.core.network.roles;
@@ -351,12 +355,21 @@ impl TaskQueueManager for Server {
                         let task_id = key.deserialize_be_u64(U64_LEN)?;
 
                         if task_due <= now_timestamp {
-                            let task_type_idx = value.deserialize_be_u16(0)?;
+                            // inbuxa: a row whose task type can't be read is
-                            let task_type = TaskType::from_id(task_type_idx).ok_or_else(|| {
+                            // set aside, not allowed to end the scan: every
-                                trc::StoreEvent::DataCorruption
+                            // task due after it would wait behind it
-                                    .caused_by(trc::location!())
+                            let Some((task_type_idx, task_type)) = value
-                                    .ctx(trc::Key::Value, value)
+                                .deserialize_be_u16(0)
-                            })?;
+                                .ok()
+                                .and_then(|idx| TaskType::from_id(idx).map(|typ| (idx, typ)))
+                            else {
+                                unreadable.push(UnreadableDueRow {
+                                    due: task_due,
+                                    id: task_id,
+                                    value: value.to_vec(),
+                                });
+                                return Ok(true);
+                            };
                             let enabled = match task_type {
                                 TaskType::IndexDocument
                                 | TaskType::UnindexDocument
@@ -446,6 +459,11 @@ impl TaskQueueManager for Server {
                 );
             });
 
+        if !unreadable.is_empty() && repair_due_rows(self, unreadable).await {
+            // Look again at once for the rows that were rewritten
+            self.notify_task_queue();
+        }
+
         if !tasks.is_empty() {
             trc::event!(
                 TaskManager(TaskManagerEvent::TaskAcquired),
@@ -686,3 +704,114 @@ impl TaskResult {
         )
     }
 }
+
+/// inbuxa: a task queue row whose task type could not be read.
+struct UnreadableDueRow {
+    due: u64,
+    id: u64,
+    value: Vec<u8>,
+}
+
+/// inbuxa: logs each unreadable queue row and repairs it from the task it
+/// schedules. The task row says what the task is, so the queue row is
+/// rewritten with that task's type; a row with no task behind it is removed.
+///
+/// Rescheduling an internal DMARC or TLS report wrote the report's object
+/// type into the queue row instead of the task type. Such a row is the time
+/// an administrator chose, so the task is moved to it as the reschedule
+/// meant to do: the task row takes that due, and a queue row left at the
+/// task's previous due is removed. Returns whether any row was repaired.
+async fn repair_due_rows(server: &Server, rows: Vec<UnreadableDueRow>) -> bool {
+    let mut repaired = false;
+    for row in rows {
+        let UnreadableDueRow { due, id, value } = row;
+        trc::error!(
+            trc::StoreEvent::DataCorruption
+                .into_err()
+                .id(id)
+                .ctx(trc::Key::Due, trc::Value::Timestamp(due))
+                .ctx(
+                    trc::Key::Key,
+                    [due.to_be_bytes(), id.to_be_bytes()].concat()
+                )
+                .ctx(trc::Key::Value, value.clone())
+                .details("Unreadable task queue row skipped")
+                .caused_by(trc::location!())
+        );
+
+        let task_key = ValueClass::TaskQueue(TaskQueueClass::Task { id });
+        let due_key = ValueClass::TaskQueue(TaskQueueClass::Due { id, due });
+        let task = match server
+            .store()
+            .get_value::<Task>(ValueKey::from(task_key.clone()))
+            .await
+        {
+            Ok(task) => task,
+            Err(err) => {
+                trc::error!(
+                    err.id(id)
+                        .details("Failed to read the task of an unreadable queue row.")
+                        .caused_by(trc::location!())
+                );
+                continue;
+            }
+        };
+
+        let mut batch = BatchBuilder::new();
+        let action = if let Some(mut task) = task {
+            let task_type = task.object_type();
+            batch.assert_value(task_key.clone(), AssertValue::Some);
+            if rescheduled_report_type(&value) == Some(task_type) {
+                let old_due = task.due_timestamp();
+                if old_due != due {
+                    batch.clear(ValueClass::TaskQueue(TaskQueueClass::Due {
+                        id,
+                        due: old_due,
+                    }));
+                }
+                task.set_status(TaskStatus::at(due as i64));
+            }
+            batch
+                .set(due_key, task_type.to_id().serialize())
+                .set(task_key, task.to_pickled_vec());
+            "Rewrote the queue row from its task."
+        } else {
+            batch.clear(due_key);
+            "Removed a queue row with no task."
+        };
+
+        match server.store().write(batch.build_all()).await {
+            Ok(_) => {
+                repaired = true;
+                trc::event!(
+                    TaskManager(TaskManagerEvent::TaskIgnored),
+                    Id = id,
+                    Due = trc::Value::Timestamp(due),
+                    Reason = action,
+                );
+            }
+            Err(err) if err.matches(trc::EventType::Store(trc::StoreEvent::AssertValueFailed)) => {
+                // The task went away meanwhile; the next scan looks again
+            }
+            Err(err) => {
+                trc::error!(
+                    err.id(id)
+                        .details("Failed to repair an unreadable queue row.")
+                        .caused_by(trc::location!())
+                );
+            }
+        }
+    }
+    repaired
+}
+
+/// inbuxa: the task type a report reschedule meant, when a queue row holds
+/// an internal report's object type (the value that reschedule wrote).
+fn rescheduled_report_type(value: &[u8]) -> Option<TaskType> {
+    let id = u16::from_be_bytes(value.get(..2)?.try_into().ok()?);
+    match ObjectType::from_id(id)? {
+        ObjectType::DmarcInternalReport => Some(TaskType::DmarcReport),
+        ObjectType::TlsInternalReport => Some(TaskType::TlsReport),
+        _ => None,
+    }
+}

crates/services/src/task_manager/spam_classifier.rs

@@ -2,13 +2,15 @@
  * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
  *
  * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
+ *
+ * Modified by Coffey Labs in 2026 for INBUXA.
  */
 
 use crate::task_manager::{TaskFailureType, TaskResult};
 use common::{
     Server,
     ipc::{BroadcastEvent, RegistryChange},
-    manager::{SPAM_CLASSIFIER_KEY, SPAM_TRAINER_KEY, fetch_resource},
+    manager::{SPAM_CLASSIFIER_KEY, SPAM_TRAINER_KEY, fetch_resource, spam_rules},
 };
 use registry::{
     schema::{
@@ -106,6 +108,7 @@ struct RuleUpdateResult {
 
 async fn update_spam_rules(server: &Server) -> trc::Result<TaskResult> {
     let started = Instant::now();
+    let bundled = server.core.spam.spam_rules_url.is_none();
     let rules = match fetch_spam_rules(server).await {
         Ok(rules) => rules,
         Err(err) => {
@@ -289,24 +292,31 @@ async fn update_spam_rules(server: &Server) -> trc::Result<TaskResult> {
         Elapsed = started.elapsed(),
     );
 
+    // inbuxa: so the next start knows these bundled rules are in
+    if bundled {
+        spam_rules::set_applied_version(server.store(), spam_rules::BUNDLED_SPAM_RULES_VERSION)
+            .await?;
+    }
+
     Ok(TaskResult::Success(vec![]))
 }
 
 async fn fetch_spam_rules(server: &Server) -> Result<Rules, RuleUpdateError> {
-    let Some(rules_url) = server.core.spam.spam_rules_url.as_ref() else {
+    // inbuxa: no URL means the rules bundled with the server
-        return Err(RuleUpdateError {
+    let bytes = match server.core.spam.spam_rules_url.as_ref() {
-            typ: TaskFailureType::Permanent,
+        Some(rules_url) => fetch_resource(rules_url, None, Duration::from_secs(60), 1024 * 500)
-            reason: "Spam rules resource URL not configured".to_string(),
-        });
-    };
-    let rules_json: AHashMap<String, Vec<serde_json::Value>> =
-        fetch_resource(rules_url, None, Duration::from_secs(60), 1024 * 500)
             .await
             .map_err(|reason| RuleUpdateError {
                 typ: TaskFailureType::Temporary,
                 reason,
-            })
+            }),
-            .and_then(|bytes| {
+        None => spam_rules::bundled_rules().map_err(|reason| RuleUpdateError {
+            typ: TaskFailureType::Permanent,
+            reason,
+        }),
+    };
+    let rules_json: AHashMap<String, Vec<serde_json::Value>> =
+        bytes.and_then(|bytes| {
             serde_json::from_slice(&bytes).map_err(|err| RuleUpdateError {
                 typ: TaskFailureType::Permanent,
                 reason: format!("Failed to parse spam rules JSON: {err}"),

crates/smtp/src/reporting/index.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 registry::{
@@ -40,35 +42,49 @@ pub trait InternalReportIndex: ObjectImpl {
 
     fn primary_key(&self) -> ValueClass;
 
+    /// Moves the report's delivery, and its queued task, to `at`.
+    ///
+    /// inbuxa: the new queue row carries the task's type, as
+    /// `schedule_task_with_id` writes it, and the task row gets the new due
+    /// too. `queued` is the task as stored: its due, not the report's
+    /// `deliverAt`, is the queue row that exists (they differ once the task
+    /// has been retried).
     fn reschedule_ops(
         &mut self,
         batch: &mut BatchBuilder,
         item_id: u64,
         revision: u64,
         at: UTCDateTime,
+        queued: Option<&Task>,
     ) {
         let current_deliver_at = self.deliver_at();
+        let current_due = current_deliver_at.timestamp() as u64;
+        let queued_due = queued.map_or(current_due, |task| task.due_timestamp());
+        let new_due = at.timestamp() as u64;
 
-        if current_deliver_at != at {
+        if current_deliver_at != at || queued_due != new_due {
             let object = Self::OBJECT;
             let object_id = object.to_id();
             let key = ValueClass::Registry(RegistryClass::Item { object_id, item_id });
 
             self.set_deliver_at(at);
 
+            batch.assert_value(key.clone(), AssertValue::Hash(revision));
+            if queued_due != new_due {
+                batch.clear(ValueClass::TaskQueue(TaskQueueClass::Due {
+                    id: item_id,
+                    due: queued_due,
+                }));
+            }
+            // A row an earlier reschedule left at the report's deliverAt
+            if current_due != new_due && current_due != queued_due {
+                batch.clear(ValueClass::TaskQueue(TaskQueueClass::Due {
+                    id: item_id,
+                    due: current_due,
+                }));
+            }
             batch
-                .assert_value(key.clone(), AssertValue::Hash(revision))
+                .schedule_task_with_id(item_id, self.task(item_id))
-                .clear(ValueClass::TaskQueue(TaskQueueClass::Due {
-                    id: item_id,
-                    due: current_deliver_at.timestamp() as u64,
-                }))
-                .set(
-                    ValueClass::TaskQueue(TaskQueueClass::Due {
-                        id: item_id,
-                        due: at.timestamp() as u64,
-                    }),
-                    object_id.serialize(),
-                )
                 .set(key, self.to_pickled_vec());
         }
     }

crates/types/src/branding.rs

@@ -81,7 +81,7 @@ fn legacy_setting(name: &str, is_set: impl Fn(&str) -> bool) -> Option<String> {
 #[macro_export]
 macro_rules! brand_version {
     () => {
-        "2026.9.24"
+        "2026.9.24.4"
     };
 }
 

docs/spec/features/ai-spam-classification.md

@@ -102,7 +102,10 @@ Permissions: `sysSpamLlmGet`, `sysSpamLlmUpdate`.
 - **Tags and scores.** The classifier's tags are ordinary spam tags, scored
   by `x:SpamTag` entries like every other tag: `Score` (a number), `Discard`
   or `Reject`. The documented defaults are `LLM_UNSOLICITED_HIGH` 3.0 and
-  `LLM_LEGITIMATE_HIGH` −3.0. A tag with no entry scores 0.
+  `LLM_LEGITIMATE_HIGH` −3.0. A tag with no entry scores 0. The server ships
+  those entries in its bundled spam rules (`resources/spam-filter/`), loaded
+  on first boot and again when the bundled version changes, so an install
+  that predates them gains them on upgrade (added 2026-09-23).
 - **`interactAi`** permission ("Interact with AI models"): lets an account's
   own Sieve scripts call `llm_prompt`. This repository's default roles give it
   to users, tenant administrators and superusers

resources/schema/schema.json.sha256

@@ -1 +1 @@
-rWqJwNJkqgKsbC1eqcEmmIAMtJPmnlDlThR2ZtW_N1c
+VbnFuwCOTBh0s2T-NuRhb2JaJr8Jl5s3LgXv4Pv2sTg

resources/spam-filter/README.md

@@ -0,0 +1,32 @@
+# Bundled spam filter rules
+
+`spam-filter-rules.json.gz` is the published rules file of
+[spam-filter](https://github.com/stalwartlabs/spam-filter) **v3.0.2**,
+unmodified. The server embeds it (`crates/common/src/manager/spam_rules.rs`)
+and loads it whenever no other rules source is configured, so a release
+scores mail with the rules it was tested with, offline and with nothing to
+fetch. The rules URL setting stays an operator override.
+
+The rules are dual-licensed MIT or Apache-2.0, Copyright (C) 2024, Stalwart
+Labs LLC; the fork takes them under MIT, with the notice in `THIRD-PARTY.md`.
+
+They include the scores for the AI classifier's tags (`LLM_*`, 3.0 for the
+high-confidence spam categories, −3.0 for legitimate), which match
+`docs/spec/features/ai-spam-classification.md`.
+
+## Updating
+
+The `upstream-watch` workflow opens an issue when spam-filter publishes a
+newer release. To take it:
+
+1. Download `spam-filter-rules.json.gz` from that release, pinned by tag
+   (`releases/download/vX.Y.Z/…`, not `latest`), over this file.
+2. Set `BUNDLED_SPAM_RULES_VERSION` in `spam_rules.rs` and the version in
+   this README and in `THIRD-PARTY.md`.
+3. Run the antispam test (`STORE=RocksDb RUST_MIN_STACK=16777216 cargo test
+   -p tests --lib -- smtp::inbound::antispam::antispam --exact`) and fix
+   expectations the new rules change, knowingly.
+
+On the next start each server loads the new version once. Loading only adds
+rules and tags that are missing; it never changes an existing one, so an
+operator's own adjustments survive.

tests/src/smtp/inbound/antispam.rs

@@ -86,19 +86,10 @@ async fn antispam() {
         .registry_create_object(SpamSettings {
             score_spam: Float::new(5.0),
             // inbuxa: the rules carry the scores the expectations are written
-            // against, so they're pinned (spam-filter v3.0.2, beside the test
+            // against. Unset, the server uses the rules bundled with it
-            // cases) rather than read from a developer's own checkout, which
+            // (resources/spam-filter/), the path production takes;
-            // left every score at zero. SPAM_RULES_URL still overrides.
+            // SPAM_RULES_URL tests another set.
-            spam_filter_rules_url: std::env::var("SPAM_RULES_URL")
+            spam_filter_rules_url: std::env::var("SPAM_RULES_URL").ok(),
-                .unwrap_or_else(|_| {
-                    concat!(
-                        "file://",
-                        env!("CARGO_MANIFEST_DIR"),
-                        "/resources/smtp/antispam/spam-filter-rules.json.gz"
-                    )
-                    .to_string()
-                })
-                .into(),
             ..Default::default()
         })
         .await;

tests/src/smtp/reporting/mod.rs

@@ -2,9 +2,12 @@
  * 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.
  */
 
 pub mod analyze;
 pub mod dmarc;
+pub mod reschedule; // inbuxa: report reschedules and unreadable queue rows
 pub mod scheduler;
 pub mod tls;

tests/src/smtp/reporting/reschedule.rs

@@ -0,0 +1,370 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! Rescheduling an internal DMARC or TLS report over JMAP moves its task: the
+//! task runs at the new time, x:Task/get shows the new due, and tasks due
+//! after it still run. A task queue row whose type can't be read is logged
+//! and repaired rather than stopping every task due after it, including the
+//! rows an earlier reschedule wrote with the report's object type.
+
+use crate::utils::server::{TestServer, TestServerBuilder};
+use common::{
+    Server,
+    config::smtp::report::AggregateFrequency,
+    ipc::{DmarcEvent, PolicyType, TlsEvent},
+};
+use mail_auth::{
+    common::parse::TxtRecordParser,
+    dmarc::Dmarc,
+    mta_sts::TlsRpt,
+    report::{ActionDisposition, DmarcResult, Record},
+};
+use registry::{
+    schema::{
+        enums::{TaskStoreMaintenanceType, TaskType},
+        prelude::{ObjectType, Property},
+        structs::{
+            DmarcInternalReport, DmarcReportSettings, Expression, Task, TaskStatus,
+            TaskStoreMaintenance, TlsInternalReport, TlsReportSettings,
+        },
+    },
+    types::{EnumImpl, ObjectImpl, datetime::UTCDateTime},
+};
+use serde_json::json;
+use smtp::reporting::{index::InternalReportIndex, send::MtaReportSend};
+use std::{
+    sync::Arc,
+    time::{Duration, Instant},
+};
+use store::{
+    SerializeInfallible, ValueKey,
+    write::{BatchBuilder, RegistryClass, TaskQueueClass, ValueClass, now},
+};
+use types::id::Id;
+use utils::snowflake::SnowflakeIdGenerator;
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial_test::serial]
+async fn report_reschedule() {
+    let mut test = TestServerBuilder::new("smtp_report_reschedule")
+        .await
+        .with_http_listener(19057)
+        .await
+        .capture_queue()
+        .build()
+        .await;
+
+    let admin = test.account("admin");
+    admin
+        .registry_create_object(TlsReportSettings {
+            max_report_size: Expression {
+                else_: "1024".into(),
+                ..Default::default()
+            },
+            ..Default::default()
+        })
+        .await;
+    admin
+        .registry_create_object(DmarcReportSettings {
+            aggregate_max_report_size: Expression {
+                else_: "1024".into(),
+                ..Default::default()
+            },
+            ..Default::default()
+        })
+        .await;
+    admin.reload_settings().await;
+    test.reload_core();
+    test.expect_reload_settings().await;
+    let admin = test.account("admin");
+
+    // A daily DMARC and TLS report, due a day from now
+    schedule_dmarc(&test, "foobar.org").await;
+    schedule_tls(&test, "foobar.org").await;
+    let dmarc_id = wait_for_report::<DmarcInternalReport>(&test, "foobar.org").await;
+    let tls_id = wait_for_report::<TlsInternalReport>(&test, "foobar.org").await;
+
+    // Reschedule both to a few seconds from now, with a task due after them
+    let at = now() + 3;
+    let later = marker_task(&test.server, at + 3).await;
+    for (object, id, task_type) in [
+        (
+            ObjectType::DmarcInternalReport,
+            dmarc_id,
+            TaskType::DmarcReport,
+        ),
+        (ObjectType::TlsInternalReport, tls_id, TaskType::TlsReport),
+    ] {
+        admin
+            .registry_update_object(
+                object,
+                id,
+                json!({
+                    Property::DeliverAt: UTCDateTime::from_timestamp(at as i64),
+                }),
+            )
+            .await;
+
+        // x:Task/get shows the new due, and the queue row carries the task's
+        // type. Upstream wrote the report's object type there and left the
+        // task at its old due
+        let task = admin.registry_get::<Task>(id).await;
+        assert_eq!(task.object_type(), task_type);
+        assert_eq!(
+            task.due_timestamp(),
+            at,
+            "{object:?} task due not moved: {task:?}"
+        );
+        assert_eq!(
+            queue_row(&test.server, id.id(), at).await,
+            Some(task_type.to_id().serialize()),
+            "{object:?} queue row"
+        );
+    }
+
+    // Both reports go out at the new time, and the later task still runs
+    wait_until_run(&test.server, &[dmarc_id.id(), tls_id.id(), later]).await;
+    assert!(now() >= at, "the reports went out before their new time");
+    assert!(
+        admin
+            .registry_get_all::<DmarcInternalReport>()
+            .await
+            .is_empty()
+    );
+    assert!(
+        admin
+            .registry_get_all::<TlsInternalReport>()
+            .await
+            .is_empty()
+    );
+
+    // Rows an earlier reschedule may have left in a store: one with the
+    // report's object type and the task left at its old due, and one that
+    // is unreadable and has no task behind it. Neither may hold back a task
+    // due after them.
+    schedule_dmarc(&test, "foobar.net").await;
+    let dmarc_id = wait_for_report::<DmarcInternalReport>(&test, "foobar.net").await;
+    let at = now() + 2;
+    let old_due = old_style_reschedule(&test.server, dmarc_id.id(), at).await;
+    let orphan = SnowflakeIdGenerator::global_id().unwrap();
+    let mut batch = BatchBuilder::new();
+    batch.set(
+        ValueClass::TaskQueue(TaskQueueClass::Due {
+            id: orphan,
+            due: at,
+        }),
+        vec![0xff, 0xff],
+    );
+    test.server.store().write(batch.build_all()).await.unwrap();
+    let later = marker_task(&test.server, at + 2).await;
+
+    wait_until_run(&test.server, &[dmarc_id.id(), later]).await;
+    assert!(
+        admin
+            .registry_get_all::<DmarcInternalReport>()
+            .await
+            .is_empty()
+    );
+    assert_eq!(queue_row(&test.server, orphan, at).await, None);
+    assert_eq!(queue_row(&test.server, dmarc_id.id(), at).await, None);
+    assert_eq!(queue_row(&test.server, dmarc_id.id(), old_due).await, None);
+
+    // x:Task/query by type skips an unreadable row rather than failing
+    let mut batch = BatchBuilder::new();
+    let due = now() + 3600;
+    batch.set(
+        ValueClass::TaskQueue(TaskQueueClass::Due { id: orphan, due }),
+        vec![0xff, 0xff],
+    );
+    test.server.store().write(batch.build_all()).await.unwrap();
+    admin
+        .registry_query_ids(
+            ObjectType::Task,
+            vec![(Property::Type, TaskType::DmarcReport.as_str())],
+            Vec::<&str>::new(),
+        )
+        .await;
+    let mut batch = BatchBuilder::new();
+    batch.clear(ValueClass::TaskQueue(TaskQueueClass::Due {
+        id: orphan,
+        due,
+    }));
+    test.server.store().write(batch.build_all()).await.unwrap();
+
+    if test.is_reset() {
+        test.temp_dir.delete();
+    }
+}
+
+async fn schedule_dmarc(test: &TestServer, domain: &str) {
+    test.server
+        .schedule_report(DmarcEvent {
+            domain: domain.to_string(),
+            report_record: Record::new()
+                .with_source_ip("192.168.1.2".parse().unwrap())
+                .with_action_disposition(ActionDisposition::Pass)
+                .with_dmarc_dkim_result(DmarcResult::Pass)
+                .with_dmarc_spf_result(DmarcResult::Fail)
+                .with_envelope_from("[email protected]")
+                .with_envelope_to("[email protected]")
+                .with_header_from("[email protected]"),
+            dmarc_record: Arc::new(
+                Dmarc::parse(format!("v=DMARC1; p=reject; rua=mailto:reports@{domain}").as_bytes())
+                    .unwrap(),
+            ),
+            interval: AggregateFrequency::Daily,
+            span_id: 0,
+        })
+        .await;
+}
+
+async fn schedule_tls(test: &TestServer, domain: &str) {
+    test.server
+        .schedule_report(TlsEvent {
+            domain: domain.to_string(),
+            policy: PolicyType::None,
+            failure: None,
+            tls_record: Arc::new(
+                TlsRpt::parse(format!("v=TLSRPTv1;rua=mailto:reports@{domain}").as_bytes())
+                    .unwrap(),
+            ),
+            interval: AggregateFrequency::Daily,
+            span_id: 0,
+        })
+        .await;
+}
+
+trait ReportDomain: ObjectImpl {
+    fn report_domain(&self) -> &str;
+}
+
+impl ReportDomain for DmarcInternalReport {
+    fn report_domain(&self) -> &str {
+        &self.domain
+    }
+}
+
+impl ReportDomain for TlsInternalReport {
+    fn report_domain(&self) -> &str {
+        &self.domain
+    }
+}
+
+async fn wait_for_report<T: ReportDomain>(test: &TestServer, domain: &str) -> Id {
+    let admin = test.account("admin");
+    for _ in 0..100 {
+        if let Some((id, _)) = admin
+            .registry_get_all::<T>()
+            .await
+            .into_iter()
+            .find(|(_, report)| report.report_domain() == domain)
+        {
+            return id;
+        }
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+    panic!("No {} for {domain}", T::OBJECT.as_str());
+}
+
+/// A task that succeeds when it runs, due at `due`.
+async fn marker_task(server: &Server, due: u64) -> u64 {
+    let id = SnowflakeIdGenerator::global_id().unwrap();
+    let mut batch = BatchBuilder::new();
+    batch.schedule_task_with_id(
+        id,
+        Task::StoreMaintenance(TaskStoreMaintenance {
+            maintenance_type: TaskStoreMaintenanceType::RemoveLockDav,
+            shard_index: Some(0),
+            status: TaskStatus::at(due as i64),
+        }),
+    );
+    server.store().write(batch.build_all()).await.unwrap();
+    server.notify_task_queue();
+    id
+}
+
+/// What the reschedule before this fix wrote: the report's object type in
+/// the new queue row, and the task row left at its old due. Returns that
+/// old due.
+async fn old_style_reschedule(server: &Server, item_id: u64, at: u64) -> u64 {
+    let object_id = ObjectType::DmarcInternalReport.to_id();
+    let key = ValueClass::Registry(RegistryClass::Item { object_id, item_id });
+    let mut report = server
+        .store()
+        .get_value::<DmarcInternalReport>(ValueKey::from(key.clone()))
+        .await
+        .unwrap()
+        .unwrap();
+    let old_due = report.deliver_at().timestamp() as u64;
+    report.set_deliver_at(UTCDateTime::from_timestamp(at as i64));
+    let mut batch = BatchBuilder::new();
+    batch
+        .clear(ValueClass::TaskQueue(TaskQueueClass::Due {
+            id: item_id,
+            due: old_due,
+        }))
+        .set(
+            ValueClass::TaskQueue(TaskQueueClass::Due {
+                id: item_id,
+                due: at,
+            }),
+            object_id.serialize(),
+        )
+        .set(key, report.to_pickled_vec());
+    server.store().write(batch.build_all()).await.unwrap();
+    server.notify_task_queue();
+    old_due
+}
+
+struct RawValue(Vec<u8>);
+
+impl store::Deserialize for RawValue {
+    fn deserialize(bytes: &[u8]) -> trc::Result<Self> {
+        Ok(RawValue(bytes.to_vec()))
+    }
+}
+
+async fn queue_row(server: &Server, id: u64, due: u64) -> Option<Vec<u8>> {
+    server
+        .store()
+        .get_value::<RawValue>(ValueKey::from(ValueClass::TaskQueue(TaskQueueClass::Due {
+            id,
+            due,
+        })))
+        .await
+        .unwrap()
+        .map(|raw| raw.0)
+}
+
+async fn task_exists(server: &Server, id: u64) -> bool {
+    server
+        .store()
+        .get_value::<Task>(ValueKey::from(ValueClass::TaskQueue(
+            TaskQueueClass::Task { id },
+        )))
+        .await
+        .unwrap()
+        .is_some()
+}
+
+async fn wait_until_run(server: &Server, ids: &[u64]) {
+    let started = Instant::now();
+    loop {
+        let mut pending = Vec::new();
+        for id in ids {
+            if task_exists(server, *id).await {
+                pending.push(*id);
+            }
+        }
+        if pending.is_empty() {
+            return;
+        }
+        if started.elapsed() > Duration::from_secs(30) {
+            panic!("tasks {pending:?} never ran");
+        }
+        tokio::time::sleep(Duration::from_millis(200)).await;
+    }
+}

tools/fork/context-check.py

@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+# SPDX-FileCopyrightText: 2026 Coffey Labs
+# SPDX-License-Identifier: AGPL-3.0-or-later
+"""Every path Cargo patches has to be in the image's build context.
+
+Cargo.toml's [patch.crates-io] can point at a directory in this repository,
+and the Dockerfile builds from a context that .dockerignore prunes to almost
+nothing. Those two facts met on 2026-09-23: a vendored, patched sieve-rs
+landed, CI stayed green -- it builds from a checkout, where the directory is
+simply there -- and the release build failed on
+
+    failed to load source for dependency `sieve-rs`
+    failed to read /build/vendor/sieve-rs/Cargo.toml
+
+after a tag had already been pushed. This is seconds, and it runs beside the
+other fork checks rather than waiting for a release to find out.
+
+Being in the context isn't enough on its own: the Dockerfile cooks the
+dependencies (`cargo chef cook`) before it copies the tree in, from a recipe
+that carries only the workspace's manifests. So each patched path must also be
+copied into that stage before the cook step, or the same error comes back
+there -- as it did for 2026.9.24.2, the first tag after the context fix.
+"""
+
+import re
+import sys
+from pathlib import Path
+
+root = Path(__file__).resolve().parents[2]
+
+
+def patched_paths(manifest: Path) -> list[str]:
+    """Directories named by a [patch...] section's `path = "..."` entries."""
+    out, in_patch = [], False
+    for line in manifest.read_text().splitlines():
+        stripped = line.strip()
+        if stripped.startswith("["):
+            in_patch = stripped.startswith("[patch")
+            continue
+        if not in_patch:
+            continue
+        m = re.search(r'path\s*=\s*"([^"]+)"', stripped)
+        if m:
+            out.append(m.group(1))
+    return out
+
+
+def allowed(dockerignore: Path) -> set[str]:
+    """The first path segment of every re-inclusion rule."""
+    keep = set()
+    for line in dockerignore.read_text().splitlines():
+        stripped = line.strip()
+        if stripped.startswith("!"):
+            keep.add(stripped[1:].strip("/").split("/")[0])
+    return keep
+
+
+def copied_before_cook(dockerfile: Path) -> list[str] | None:
+    """Sources COPY'd into the stage that runs `cargo chef cook`, before it.
+
+    None when no stage cooks. A `COPY . .` covers everything.
+    """
+    stage: list[str] = []
+    for line in dockerfile.read_text().splitlines():
+        stripped = line.strip()
+        if re.match(r"(?i)^FROM\s", stripped):
+            stage = []
+            continue
+        if "cargo chef cook" in stripped:
+            return stage
+        m = re.match(r"(?i)^COPY\s+(?!--from)(.+)$", stripped)
+        if m:
+            parts = m.group(1).split()
+            stage.extend(p.strip("./").split("/")[0] or "." for p in parts[:-1])
+    return None
+
+
+def main() -> int:
+    paths = patched_paths(root / "Cargo.toml")
+    if not paths:
+        print("no patched paths to check")
+        return 0
+    keep = allowed(root / ".dockerignore")
+    bad = []
+    for p in paths:
+        top = p.strip("/").split("/")[0]
+        if top not in keep:
+            bad.append((p, top))
+        elif not (root / p).is_dir():
+            bad.append((p, None))
+    for path, top in bad:
+        if top is None:
+            print(f"Cargo.toml patches {path}, which does not exist", file=sys.stderr)
+        else:
+            print(
+                f"Cargo.toml patches {path}, but .dockerignore does not re-include {top!r}:\n"
+                f"  the image build would not see it, and cargo would fail on it.\n"
+                f"  Add `!{top}` to .dockerignore.",
+                file=sys.stderr,
+            )
+    copied = copied_before_cook(root / "Dockerfile")
+    if copied is not None and "." not in copied:
+        for p in paths:
+            top = p.strip("/").split("/")[0]
+            if top not in copied:
+                print(
+                    f"Cargo.toml patches {p}, but the Dockerfile doesn't copy {top!r} into the\n"
+                    f"  stage that runs `cargo chef cook` before that step, so cooking the\n"
+                    f"  dependencies fails on it. Add `COPY {top}/ {top}/` before the cook.",
+                    file=sys.stderr,
+                )
+                bad.append((p, top))
+    if bad:
+        return 1
+    print(f"build context and cook stage include every patched path: {', '.join(paths)}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tools/fork/name-allowlist.txt

@@ -23,6 +23,6 @@ crates/migration/src/lib.rs	"STALWART_SPAM_CLASSIFIER_MODEL.lz4"
 crates/migration/src/lib.rs	"STALWART_SPAM_TRAIN_DATA.lz4"
 crates/types/src/branding.rs	"STALWART"
 
-# OPEN, not yet decided (2026-09-22): upstream's published spam-filter rules,
+# Upstream's old default rules source, read only to treat it as unset: the
-# which the server downloads at runtime from this address.
+# server uses the rules bundled with it (resources/spam-filter/).
-crates/registry/src/schema/structs_impl.rs	"https://github.com/stalwartlabs/spam-filter/releases/latest/download/spam-filter-rules.json.gz"
+crates/common/src/manager/spam_rules.rs	"https://github.com/stalwartlabs/spam-filter/releases/latest/download/spam-filter-rules.json.gz"

tools/fork/renames.py

@@ -46,6 +46,10 @@ TEXT_RENAMES = [
     # that must match their containers and identity provider (database users,
     # passwords, an OIDC audience), and name their databases explicitly.
     ('"stalwart".to_string()', '"inbuxa".to_string()', ('crates',)),
+    # The spam filter rules ship with the server (common::manager::spam_rules);
+    # upstream's default of fetching its latest from GitHub becomes unset.
+    ('spam_filter_rules_url: Some("https://github.com/stalwartlabs/spam-filter/releases/latest/download/spam-filter-rules.json.gz".to_string()),',
+     'spam_filter_rules_url: None,', ('crates',)),
 ]
 ROOTS = ('crates', 'tests', 'resources')
 SKIP_SUFFIXES = {'.md', '.txt'}
@@ -58,6 +62,10 @@ SCHEMA_HASH = Path('resources/schema/schema.json.sha256')
 SCHEMA_RENAMES = [
     ('"stalwart"', '"inbuxa"'),
     ('vnd.stalwart', 'vnd.inbuxa'),
+    # The bundled spam rules: no default URL, and say what empty means.
+    ('"spamFilterRulesUrl":"https://github.com/stalwartlabs/spam-filter/releases/latest/download/spam-filter-rules.json.gz",', ''),
+    ('"URL to download spam filter rules from"',
+     '"URL to download spam filter rules from. Empty uses the rules bundled with the server."'),
 ]