PostgreSQL search: find words inside URLs and file names in body text

After #37, address fields on PostgreSQL are split into words as the
built-in index splits them, but language text (subject, body,
attachments) still goes straight to PostgreSQL's parser, which keeps a
URL, host, path or file name as tokens of its own:
"https://x.example/shipping-support/" becomes a url, a host and a
url_path, "invoice-2024.pdf" a file. So TEXT/BODY "shipping" missed
messages where the word appears only inside a link, while RocksDB and
the other built-in backends found them: 8 messages across a handful
of searches in the rehearsal.

On insert, language text is now indexed as it was, followed by the
word parts of each token that holds a URL separator (/ . @ : ? = & # _
% + ~ \), split with SpaceTokenizer as keyword_terms() splits addresses.
The parts go through the same text search configuration as the rest of
the text, so they are stemmed like the words around them. Plain words,
words that only carry punctuation ("end.", "(see") and hyphenated words
(the parser already splits those) add nothing, so text without links
is indexed exactly as before. Each part is added once per document.
On sample mail, the text vector of a short order notice with three
links grows from 546 to 716 bytes, a newsletter with 25 tracking links
from 5586 to 6430, and a plain letter not at all.

On search, a query word written as a URL, host, file or hyphenated word
also matches as its word parts, ORed with the query as written, so
"shipping-support" or "invoice-2024.pdf" match the new parts and
documents indexed before this change still match as they did.

Existing messages keep their old vectors until they are reindexed (the
reindexAccounts task); new and reindexed messages match at once.

store::search_tests gains test_url_word_search: five bodies, 19 body
searches for words found only in a URL path, query string, host or
file name, the tokens as written, plain words and non-matches, with the
same expected ids on every backend. It passes on RocksDB, SQLite,
MySQL and PostgreSQL; on main PostgreSQL fails at the first ("shipping"
finds [3], not [0, 3]). On PostgreSQL the suite then stops at the
account sort assertion (query.rs:689) exactly as it does on main.

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),

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");
     }