Settings writes: wait for a burst to settle before reloading

A cluster rehearsal sent ten x:<Object>/set requests at once and got
ten full reloads on every node. #39's coalescing only joined writes
that queued behind a running reload, but the requests reached the
server about 33 ms apart and a reload takes tens of milliseconds, so
none overlapped one.

A full reload after a registry write now waits for writes to settle:
75 ms after the last one, and at most 250 ms after the first it
covers, so a steady stream still reloads at least four times a
second. 75 ms is a little over twice the gap the rehearsal saw between
requests. A single write pays it once: in the tests a settings write
takes about 140 ms instead of 60. The reload runs in a task of its
own, so a request that goes away doesn't cancel it for the others.
Each write takes the result of the first reload that started after it
was stored (the gate keeps the last 64 results), so applied true or
false still describes the reload that covered that write.

The 33 ms gap was a queue on the server, not password hashing: Basic
credentials are cached per Authorization header, so they are checked
once. Every authenticated HTTP request counted itself against the
account's rate limit by incrementing one counter per account in the
in-memory store, so parallel requests from one account queued on that
key: a row lock on PostgreSQL (a few round trips to the database
each) and conflict retries with a 50-300 ms backoff on RocksDB. An
account with the unlimitedRequests permission (administrators, by
default) passes the rate and concurrency limits anyway, so its
requests are no longer counted. Ten parallel Core/echo calls as the
admin now finish in 1-4 ms; before, they finished one after another
over 20 ms on a local PostgreSQL and 300-450 ms on RocksDB. Other
accounts still count every request.

system::auto_reload::settings_reload_tests: ten concurrent writes now
take one reload (the gate counts them; at most two allowed), all are
applied: true and in the running settings, and a single write takes
exactly one reload. RocksDB and PostgreSQL, 1 reload in 141-196 ms.
With the old behavior (no wait, requests counted) the same writes
took 5 reloads; without the wait but with the rate fix, 2.
cluster::broadcast (3 nodes, PostgreSQL + NATS) and system::reload
still pass.

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> {

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