diff --git a/crates/common/src/auth/rate_limit.rs b/crates/common/src/auth/rate_limit.rs index 9202044..755748c 100644 --- a/crates/common/src/auth/rate_limit.rs +++ b/crates/common/src/auth/rate_limit.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * 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> { + // 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 diff --git a/crates/common/src/cache/reload.rs b/crates/common/src/cache/reload.rs index e8fd7b7..6e20d49 100644 --- a/crates/common/src/cache/reload.rs +++ b/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 { // 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, + reloads: std::sync::atomic::AtomicU64, + state: parking_lot::Mutex, + completed: tokio::sync::watch::Sender, } #[derive(Default)] struct SettingsReloadState { - completed: u64, - refused: Option, + /// 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, + last_write: Option, + /// Recent reloads, oldest first: the last write each covered, and why + /// it was refused, if it was. + results: std::collections::VecDeque<(u64, Option)>, } +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> { + 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:/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> { diff --git a/tests/src/system/auto_reload.rs b/tests/src/system/auto_reload.rs index d7a89c0..95dfa88 100644 --- a/tests/src/system/auto_reload.rs +++ b/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::>(); + 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"); }