Merge pull request 'Release 2026.9.25.1' (#55) from bump/2026.9.25.1 into main

Every node renews its lease once a minute instead of every 30 minutes,
so the lease works as a heartbeat. x:ClusterNode reports a node Stale
once it has gone three minutes without renewing (it used to take an
hour), and Inactive after a day, as before.

Taking over a lease still needs a full hour of silence. A node that is
slow rather than gone never loses its id to another host, so snowflake
ids stay unique.

The admin dashboard's Cluster Health card counts these statuses.

crates/common/src/auth/rate_limit.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::auth::AccessToken;
@@ -18,6 +20,16 @@ impl Server {
         access_token: &AccessToken,
         addr: IpAddr,
     ) -> trc::Result<Option<InFlight>> {
+        // inbuxa: an account with unlimited requests passes both limits
+        // below anyway, so don't count its requests. The count is a write to
+        // one counter per account in the in-memory store, and concurrent
+        // requests from one account queue on that key (a row lock on SQL,
+        // conflict retries on RocksDB): in a cluster rehearsal ten parallel
+        // admin writes were accepted one after another, about 33 ms apart.
+        if access_token.has_permission(Permission::UnlimitedRequests) {
+            return Ok(None);
+        }
+
         let rate_reset = if let Some(rate) = &self.core.network.http.rate_authenticated {
             if self.is_ip_allowed(addr) {
                 None

crates/common/src/cache/reload.rs

@@ -7,7 +7,7 @@
  */
 
 use crate::{
-    Core, Server,
+    BuildServer, Core, Server,
     config::{
         server::{Listeners, tls::parse_certificates},
         storage::Storage,
@@ -245,21 +245,73 @@ fn error_object(error: &Error) -> Option<ObjectId> {
 // until someone reloaded. Writes to objects the settings are built from now
 // reload them, here and across the cluster, as ReloadSettings does.
 
-/// Coalesces the full reloads that registry writes trigger: a write waits for
-/// a reload that started after it was stored, and joins one if it can, so a
-/// burst of writes costs a reload or two rather than one each.
-#[derive(Default)]
+/// Coalesces the full reloads that registry writes trigger. A write waits
+/// for more writes before a reload starts (see [`WRITE_QUIET`]), then
+/// takes the result of the first reload that started after it was stored,
+/// so a burst of writes, or a request with many objects, costs one reload
+/// or two rather than one each.
 pub struct SettingsReloadGate {
     requested: std::sync::atomic::AtomicU64,
-    state: tokio::sync::Mutex<SettingsReloadState>,
+    reloads: std::sync::atomic::AtomicU64,
+    state: parking_lot::Mutex<SettingsReloadState>,
+    completed: tokio::sync::watch::Sender<u64>,
 }
 
 #[derive(Default)]
 struct SettingsReloadState {
-    completed: u64,
-    refused: Option<String>,
+    /// A reload is waiting for writes to settle, or running.
+    scheduled: bool,
+    /// When the oldest write not yet covered by a reload was stored, and
+    /// the newest.
+    first_write: Option<std::time::Instant>,
+    last_write: Option<std::time::Instant>,
+    /// Recent reloads, oldest first: the last write each covered, and why
+    /// it was refused, if it was.
+    results: std::collections::VecDeque<(u64, Option<String>)>,
 }
 
+impl Default for SettingsReloadGate {
+    fn default() -> Self {
+        Self {
+            requested: Default::default(),
+            reloads: Default::default(),
+            state: Default::default(),
+            completed: tokio::sync::watch::Sender::new(0),
+        }
+    }
+}
+
+impl SettingsReloadGate {
+    /// How many full reloads registry writes have run.
+    pub fn reloads(&self) -> u64 {
+        self.reloads.load(std::sync::atomic::Ordering::Relaxed)
+    }
+}
+
+impl SettingsReloadState {
+    /// The result of the reload that covered write `ticket`, once it ran.
+    fn result_for(&self, ticket: u64) -> Option<Result<(), String>> {
+        self.results
+            .iter()
+            .find(|(covers, _)| *covers >= ticket)
+            .map(|(_, refused)| refused.clone().map_or(Ok(()), Err))
+    }
+}
+
+/// How long a full reload waits after the last registry write for another.
+/// Parallel requests reach the server tens of milliseconds apart (in a
+/// cluster rehearsal, ten x:<Object>/set requests sent at once arrived about
+/// 33 ms apart and each got a reload of its own), so the window is a little
+/// over twice that. A single write pays it once, on top of the reload.
+pub const WRITE_QUIET: std::time::Duration = std::time::Duration::from_millis(75);
+
+/// The longest a full reload waits after the first write it covers, so a
+/// steady stream of writes still reloads at least this often.
+pub const WRITE_MAX_WAIT: std::time::Duration = std::time::Duration::from_millis(250);
+
+/// How many past reload results a waiting write can look up.
+const RELOAD_RESULTS: usize = 64;
+
 /// The reload a write to `object` calls for: the object to reload, or None
 /// when the running settings don't hold that object (accounts, domains and
 /// other data read as needed, stores, which take a restart, and objects with
@@ -370,21 +422,85 @@ impl Server {
             return Some(result);
         }
 
+        // inbuxa: #39 joined only writes that queued behind a running
+        // reload; requests that arrive tens of milliseconds apart never
+        // overlapped one, so each got a reload of its own. The reload now
+        // waits until writes settle (WRITE_QUIET after the last one, at
+        // most WRITE_MAX_WAIT after the first) and covers them all. It runs
+        // in a task of its own, so a request that goes away doesn't take
+        // it with it; each write then takes the result of the reload that
+        // started after it was stored.
         let gate = &self.inner.data.settings_reload;
         let ticket = gate
             .requested
             .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
             + 1;
-        let mut state = gate.state.lock().await;
-        if state.completed >= ticket {
-            // A reload that started after this write was stored has run
-            return Some(state.refused.clone().map_or(Ok(()), Err));
+        let now = std::time::Instant::now();
+        {
+            let mut state = gate.state.lock();
+            state.first_write.get_or_insert(now);
+            state.last_write = Some(now);
         }
-        let covers = gate.requested.load(std::sync::atomic::Ordering::SeqCst);
-        let result = self.reload_and_broadcast(change).await;
-        state.completed = covers;
-        state.refused = result.clone().err();
-        Some(result)
+
+        loop {
+            let mut completed = {
+                let mut state = gate.state.lock();
+                if let Some(result) = state.result_for(ticket) {
+                    return Some(result);
+                }
+                if !state.scheduled {
+                    state.scheduled = true;
+                    let server = self.clone();
+                    tokio::spawn(async move {
+                        server.run_write_reload(change).await;
+                    });
+                }
+                gate.completed.subscribe()
+            };
+            if completed.changed().await.is_err() {
+                return Some(Err("The settings reload was interrupted".to_string()));
+            }
+        }
+    }
+
+    /// Waits for registry writes to settle, then reloads the settings once
+    /// for all the writes stored so far.
+    async fn run_write_reload(&self, change: RegistryChange) {
+        let gate = &self.inner.data.settings_reload;
+        loop {
+            let deadline = {
+                let state = gate.state.lock();
+                let now = std::time::Instant::now();
+                let first = state.first_write.unwrap_or(now);
+                let last = state.last_write.unwrap_or(now);
+                (last + WRITE_QUIET).min(first + WRITE_MAX_WAIT)
+            };
+            if deadline <= std::time::Instant::now() {
+                break;
+            }
+            tokio::time::sleep_until(deadline.into()).await;
+        }
+
+        // Writes stored from here on wait for the next reload
+        let covers = {
+            let mut state = gate.state.lock();
+            state.first_write = None;
+            state.last_write = None;
+            gate.requested.load(std::sync::atomic::Ordering::SeqCst)
+        };
+        gate.reloads
+            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+        let result = self.inner.build_server().reload_and_broadcast(change).await;
+
+        {
+            let mut state = gate.state.lock();
+            if state.results.len() == RELOAD_RESULTS {
+                state.results.pop_front();
+            }
+            state.results.push_back((covers, result.err()));
+            state.scheduled = false;
+        }
+        gate.completed.send_replace(covers);
     }
 
     async fn reload_and_broadcast(&self, change: RegistryChange) -> Result<(), String> {

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

@@ -51,7 +51,9 @@ impl PostgresStore {
                 let all_fields = index.all_fields();
                 let fields = document.fields;
                 // inbuxa: keyword text (addresses, contact fields, ...) is split into
-                // words before it reaches the text parser, see keyword_terms().
+                // words before it reaches the text parser, see keyword_terms();
+                // language text gets the words inside its URLs, host names and
+                // file names added, see url_terms().
                 let keywords = primary_keys
                     .iter()
                     .chain(all_fields)
@@ -60,6 +62,9 @@ impl PostgresStore {
                             value,
                             language: Language::None,
                         }) if field.is_text() => Some(keyword_terms(value)),
+                        Some(SearchValue::Text { value, .. }) if field.is_text() => {
+                            url_terms(value)
+                        }
                         _ => None,
                     })
                     .collect::<Vec<_>>();
@@ -290,14 +295,36 @@ impl PostgresStore {
                                 continue;
                             }
                         } else {
+                            // inbuxa: a query word written as a URL, host,
+                            // file or hyphenated word also matches as its word
+                            // parts, which url_terms() indexes
+                            let parts = match value {
+                                SearchValue::Text { value, .. } => query_url_terms(value),
+                                _ => None,
+                            };
+                            let parts_pos = value_pos + 1;
                             let _ = write!(query, "@@ ({method}('{config}', ${value_pos})");
+                            if parts.is_some() {
+                                let _ = write!(query, " || {method}('{config}', ${parts_pos})");
+                            }
                             for fallback in [PG_FALLBACK_LANG, PG_UNSTEMMED_LANG] {
                                 if fallback != config && self.ts_configs.contains(fallback) {
                                     let _ =
                                         write!(query, " || {method}('{fallback}', ${value_pos})");
+                                    if parts.is_some() {
+                                        let _ = write!(
+                                            query,
+                                            " || {method}('{fallback}', ${parts_pos})"
+                                        );
+                                    }
                                 }
                             }
                             query.push(')');
+                            values.push(SqlParam::Ref(value));
+                            if let Some(parts) = parts {
+                                values.push(SqlParam::Owned(parts));
+                            }
+                            continue;
                         }
                         values.push(SqlParam::Ref(value));
                     } else if let SearchValue::KeyValues(kv) = value {
@@ -391,6 +418,75 @@ pub(crate) fn keyword_terms(value: &str) -> String {
     terms
 }
 
+// inbuxa: in language text (subject, body, attachments) PostgreSQL's parser
+// keeps a URL, a host name, a path or a file name as tokens of its own:
+// "https://x.example/shipping-support/" gives a url, a host and a url_path,
+// "invoice-2024.pdf" a file, so a body search for "shipping" or "invoice"
+// missed messages where the word appears only there, while the built-in index
+// splits them into words. The text is indexed as it was, followed by the word
+// parts of each such token (SpaceTokenizer, as keyword_terms() splits), so
+// they go through the same configuration and stemming as the words around
+// them. On sample mail the text vector grows by about 15% for a newsletter
+// full of tracking links and 30% for a short order notice with three links.
+// Plain words, and words that only carry punctuation ("end.", "(see"),
+// add nothing; hyphenated words are already split by the parser. Returns None
+// when there is nothing to add, so most text is indexed exactly as before.
+/// Characters that join the parts of a URL, host, path, address or file name.
+const URL_SEPARATORS: [char; 13] = [
+    '/', '.', '@', ':', '?', '=', '&', '#', '_', '%', '+', '~', '\\',
+];
+
+pub(crate) fn url_terms(value: &str) -> Option<String> {
+    let mut terms = String::new();
+    // Each word is added once: a phrase search still finds the first URL it
+    // is in, and a newsletter's hundred tracking links don't add a hundred
+    // positions for "utm" and "campaign"
+    let mut seen = std::collections::HashSet::new();
+    for token in value.split(|c: char| {
+        c.is_whitespace() || matches!(c, '<' | '>' | '"' | '(' | ')' | '[' | ']' | '{' | '}')
+    }) {
+        let token = token.trim_matches(|c: char| !c.is_alphanumeric());
+        if token.contains(URL_SEPARATORS) {
+            for word in SpaceTokenizer::new(token, MAX_TOKEN_LENGTH) {
+                if !seen.insert(word.clone()) {
+                    continue;
+                }
+                if terms.is_empty() {
+                    terms.reserve(value.len() + 64);
+                    terms.push_str(value);
+                    terms.push('\n');
+                } else {
+                    terms.push(' ');
+                }
+                terms.push_str(&word);
+            }
+        }
+    }
+    (!terms.is_empty()).then_some(terms)
+}
+
+/// The query side of url_terms(): each query word that is a URL, host, file
+/// name or hyphenated word replaced by its word parts, or None when there is
+/// none. It is searched in addition to the query as written, so documents
+/// indexed before url_terms() still match as they did.
+pub(crate) fn query_url_terms(value: &str) -> Option<String> {
+    let mut terms = String::with_capacity(value.len());
+    let mut changed = false;
+    for token in value.split_whitespace() {
+        let word = token.trim_matches(|c: char| !c.is_alphanumeric());
+        if !terms.is_empty() {
+            terms.push(' ');
+        }
+        if word.contains(URL_SEPARATORS) || word.contains('-') {
+            changed = true;
+            terms.push_str(&keyword_terms(word));
+        } else {
+            terms.push_str(token);
+        }
+    }
+    changed.then_some(terms)
+}
+
 pub(super) enum SqlParam<'x> {
     Ref(&'x (dyn ToSql + Sync)),
     Owned(String),

crates/store/src/build/cluster.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::{
@@ -23,6 +25,14 @@ use utils::snowflake::MAX_NODE_ID;
 
 const STALE_NODE_TIMEOUT: u64 = 60 * 60; // 1 hour
 const DEAD_NODE_TIMEOUT: u64 = 60 * 60 * 24; // 24 hours
+
+// INBUXA: every node renews its lease once a minute, so the lease doubles as
+// a heartbeat. A node not heard from in three minutes is reported Stale, which
+// is what Cluster Health on the dashboard counts. Taking over a lease still
+// needs the full hour of silence, so a node that is slow rather than gone
+// never loses its id to another host.
+const HEARTBEAT_INTERVAL: u64 = 60; // 1 minute
+const UNRESPONSIVE_NODE_TIMEOUT: u64 = 3 * HEARTBEAT_INTERVAL;
 const MAX_LEASE_RETRIES: u32 = 5;
 
 struct NodeSlot {
@@ -96,7 +106,7 @@ impl RegistryStore {
     }
 
     pub fn refresh_node_id_interval(&self) -> Duration {
-        Duration::from_secs(STALE_NODE_TIMEOUT / 2)
+        Duration::from_secs(HEARTBEAT_INTERVAL)
     }
 
     pub async fn cluster_node_list(&self) -> trc::Result<Vec<ClusterNode>> {
@@ -289,6 +299,10 @@ impl NodeSlot {
         self.elapsed > DEAD_NODE_TIMEOUT
     }
 
+    fn is_responsive(&self) -> bool {
+        self.elapsed <= UNRESPONSIVE_NODE_TIMEOUT
+    }
+
     fn is_assignable(&self) -> bool {
         self.node_id <= MAX_NODE_ID
     }
@@ -296,10 +310,10 @@ impl NodeSlot {
     fn status(&self) -> ClusterNodeStatus {
         if self.is_dead() {
             ClusterNodeStatus::Inactive
-        } else if self.is_stale() {
-            ClusterNodeStatus::Stale
-        } else {
+        } else if self.is_responsive() {
             ClusterNodeStatus::Active
+        } else {
+            ClusterNodeStatus::Stale
         }
     }
 }
@@ -314,3 +328,42 @@ impl From<NodeSlot> for ClusterNode {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn slot(elapsed: u64) -> NodeSlot {
+        NodeSlot {
+            node_id: 1,
+            hostname: "mx2.example.org".into(),
+            last_renewal: 0,
+            elapsed,
+            hash: 0,
+        }
+    }
+
+    #[test]
+    fn status_follows_the_heartbeat() {
+        assert_eq!(slot(0).status(), ClusterNodeStatus::Active);
+        assert_eq!(slot(UNRESPONSIVE_NODE_TIMEOUT).status(), ClusterNodeStatus::Active);
+        assert_eq!(slot(UNRESPONSIVE_NODE_TIMEOUT + 1).status(), ClusterNodeStatus::Stale);
+        assert_eq!(slot(DEAD_NODE_TIMEOUT).status(), ClusterNodeStatus::Stale);
+        assert_eq!(slot(DEAD_NODE_TIMEOUT + 1).status(), ClusterNodeStatus::Inactive);
+    }
+
+    #[test]
+    fn a_silent_node_keeps_its_id_for_an_hour() {
+        // Reported Stale after three minutes, but not free to take over.
+        let quiet = slot(UNRESPONSIVE_NODE_TIMEOUT + 1);
+        assert_eq!(quiet.status(), ClusterNodeStatus::Stale);
+        assert!(!quiet.is_stale());
+        assert!(slot(STALE_NODE_TIMEOUT + 1).is_stale());
+    }
+
+    #[test]
+    fn several_renewals_fit_before_a_node_looks_unresponsive() {
+        assert!(UNRESPONSIVE_NODE_TIMEOUT >= 3 * HEARTBEAT_INTERVAL);
+        assert!(HEARTBEAT_INTERVAL * 2 < STALE_NODE_TIMEOUT);
+    }
+}

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.3"
+        "2026.9.25.1"
     };
 }
 

tests/src/store/query.rs

@@ -133,6 +133,11 @@ pub async fn test(test: &TestServer) {
     println!("Running address search tests...");
     test_address_search(store.clone()).await;
 
+    // inbuxa: words inside URLs, host names and file names in body text
+    // are found on every backend
+    println!("Running URL word search tests...");
+    test_url_word_search(store.clone()).await;
+
     // Large document insert test
     println!("Running large document insert tests...");
     let mut large_text = String::with_capacity(20 * 1024 * 1024);
@@ -1129,3 +1134,79 @@ async fn test_address_search(store: SearchStore) {
         .await
         .unwrap();
 }
+
+async fn test_url_word_search(store: SearchStore) {
+    const ACCOUNT_ID: u32 = 8;
+    let bodies = [
+        "Track your parcel here: https://x.example/shipping-support/ and reply.",
+        "Reset it at https://mail.example.com/login/?password=reset&user=jane now.",
+        "Attached is invoice-2024.pdf for your records.",
+        "Shipping was fast, thanks again.",
+        "Nothing to see at www.example.org/about-us, really.",
+    ];
+
+    let mut documents = Vec::new();
+    let mut mask = RoaringBitmap::new();
+    for (document_id, body) in bodies.iter().enumerate() {
+        let mut document = IndexDocument::new(SearchIndex::Email)
+            .with_account_id(ACCOUNT_ID)
+            .with_document_id(document_id as u32);
+        document.index_text(EmailSearchField::Body, body, Language::English);
+        document.index_unsigned(EmailSearchField::ReceivedAt, document_id as u64);
+        documents.push(document);
+        mask.insert(document_id as u32);
+    }
+    store.index(documents).await.unwrap();
+    if let SearchStore::ElasticSearch(store) = &store {
+        store.refresh_index(SearchIndex::Email).await.unwrap();
+    }
+
+    for (text, expected) in [
+        // only inside a URL path, a query string or a file name
+        ("shipping", vec![0u32, 3]),
+        ("support", vec![0]),
+        ("password", vec![1]),
+        ("login", vec![1]),
+        ("jane", vec![1]),
+        ("invoice", vec![2]),
+        ("pdf", vec![2]),
+        ("2024", vec![2]),
+        // host names
+        ("example", vec![0, 1, 4]),
+        ("mail", vec![1]),
+        // written as they appear
+        ("https://x.example/shipping-support/", vec![0]),
+        ("shipping-support", vec![0]),
+        ("invoice-2024.pdf", vec![2]),
+        ("mail.example.com", vec![1]),
+        // plain words are unaffected
+        ("parcel", vec![0]),
+        ("records", vec![2]),
+        ("thanks", vec![3]),
+        // no match
+        ("billing", vec![]),
+        ("example.net", vec![]),
+    ] {
+        let ids = store
+            .query_account(
+                SearchQuery::new(SearchIndex::Email)
+                    .with_filters(vec![
+                        SearchFilter::eq(SearchField::AccountId, ACCOUNT_ID),
+                        SearchFilter::has_english_text(EmailSearchField::Body, text),
+                    ])
+                    .with_comparator(SearchComparator::ascending(EmailSearchField::ReceivedAt))
+                    .with_mask(mask.clone()),
+            )
+            .await
+            .unwrap();
+        assert_eq!(ids, expected, "Body {text:?}");
+    }
+
+    store
+        .unindex(
+            SearchQuery::new(SearchIndex::Email)
+                .with_filter(SearchFilter::eq(SearchField::AccountId, ACCOUNT_ID)),
+        )
+        .await
+        .unwrap();
+}

tests/src/system/auto_reload.rs

@@ -116,12 +116,64 @@ async fn test_write_applies(test: &TestServer) {
     for name in &names {
         assert!(has_schedule(test, name), "{name} missing");
     }
+    // A burst of separate requests shares a reload or two: each arrives
+    // tens of milliseconds after the last, so none overlaps a running
+    // reload, and the reload waits for writes to settle instead
+    let reloads = test.server.inner.data.settings_reload.reloads();
+    let started = std::time::Instant::now();
+    let burst = (0..10)
+        .map(|i| format!("autoreload-burst-{i}"))
+        .collect::<Vec<_>>();
+    let mut writes = Vec::new();
+    for name in &burst {
+        writes.push(admin.registry_create([MtaDeliverySchedule {
+            name: name.clone(),
+            queue_id,
+            ..Default::default()
+        }]));
+    }
+    for response in futures::future::join_all(writes).await {
+        assert_applied(&response);
+        schedule_ids.push(response.created_id(0));
+    }
+    let burst_reloads = test.server.inner.data.settings_reload.reloads() - reloads;
+    println!(
+        "10 concurrent writes: {burst_reloads} reload(s), {} ms",
+        started.elapsed().as_millis()
+    );
+    assert!(
+        (1..=2).contains(&burst_reloads),
+        "{burst_reloads} reloads for 10 concurrent writes"
+    );
+    for name in &burst {
+        assert!(has_schedule(test, name), "{name} missing");
+    }
+
+    // A single write still reloads promptly
+    let reloads = test.server.inner.data.settings_reload.reloads();
+    let started = std::time::Instant::now();
+    let response = admin
+        .registry_create([MtaDeliverySchedule {
+            name: "autoreload-single".into(),
+            queue_id,
+            ..Default::default()
+        }])
+        .await;
+    assert_applied(&response);
+    schedule_ids.push(response.created_id(0));
+    println!("1 write: {} ms", started.elapsed().as_millis());
+    assert_eq!(
+        test.server.inner.data.settings_reload.reloads() - reloads,
+        1
+    );
+    assert!(has_schedule(test, "autoreload-single"));
+
     // Several objects in one request: one reload
     let response = admin
         .registry_destroy(ObjectType::MtaDeliverySchedule, schedule_ids.iter())
         .await;
     assert_applied(&response);
-    for name in &names {
+    for name in names.iter().chain(&burst) {
         assert!(!has_schedule(test, name), "{name} still present");
     }