diff --git a/architecture/gateway.md b/architecture/gateway.md index f541397b8..f08bab7e3 100644 --- a/architecture/gateway.md +++ b/architecture/gateway.md @@ -680,7 +680,10 @@ refresh bootstrap material; sandboxes receive minted access tokens instead. Committed configuration mutations publish component and scope identifiers, never configuration payloads, to a bounded coalescing scheduler. The scheduler admits a fixed number of delivery workers and builds the latest full snapshot -for each affected active sandbox. Fleet fanout waits for worker capacity before +for each affected active sandbox. Delivery admission allows at least 64 workers, +while snapshot build concurrency remains tied to database pool capacity. This +absorbs scoped bursts without increasing concurrent database-backed builds. +Fleet fanout waits for worker capacity before admitting another recipient. An async router owns session lookup, message sizing, sequence allocation, and enqueue. Its local implementation uses the process-local supervisor registry. A future HA implementation can resolve the diff --git a/crates/openshell-server/src/config_delivery.rs b/crates/openshell-server/src/config_delivery.rs index 53ab1d4c3..6cf27cfe2 100644 --- a/crates/openshell-server/src/config_delivery.rs +++ b/crates/openshell-server/src/config_delivery.rs @@ -38,6 +38,9 @@ const MAX_ACTIVE_FANOUT_WORKERS: usize = 64; /// pool busy without stacking every waiter on the acquire timeout. const SNAPSHOT_BUILDS_PER_DB_CONNECTION: usize = 2; const MIN_CONCURRENT_SNAPSHOT_BUILDS: usize = 4; +/// Admit scoped bursts independently of the database build bound. Workers +/// waiting to build still count toward this limit. +const MIN_CONCURRENT_DELIVERY_WORKERS: usize = 64; /// One complete configuration component awaiting delivery to a supervisor. #[derive(Clone)] @@ -210,11 +213,15 @@ impl Default for ConfigDeliveryQueue { impl ConfigDeliveryQueue { #[must_use] pub fn new(max_concurrent_builds: usize) -> Self { + Self::with_limits(max_concurrent_builds, max_concurrent_builds) + } + + fn with_limits(max_concurrent_builds: usize, max_delivery_workers: usize) -> Self { let max_concurrent_builds = max_concurrent_builds.max(1); Self { pending: Mutex::default(), fanout_pending: Mutex::default(), - delivery_permits: Arc::new(Semaphore::new(max_concurrent_builds)), + delivery_permits: Arc::new(Semaphore::new(max_delivery_workers.max(1))), build_permits: Semaphore::new(max_concurrent_builds), } } @@ -223,11 +230,10 @@ impl ConfigDeliveryQueue { #[must_use] pub fn for_db_connections(max_connections: u32) -> Self { let max_connections = usize::try_from(max_connections).unwrap_or(usize::MAX); - Self::new( - max_connections - .saturating_mul(SNAPSHOT_BUILDS_PER_DB_CONNECTION) - .max(MIN_CONCURRENT_SNAPSHOT_BUILDS), - ) + let builds = max_connections + .saturating_mul(SNAPSHOT_BUILDS_PER_DB_CONNECTION) + .max(MIN_CONCURRENT_SNAPSHOT_BUILDS); + Self::with_limits(builds, builds.max(MIN_CONCURRENT_DELIVERY_WORKERS)) } #[cfg(test)] @@ -672,6 +678,9 @@ mod tests { #[test] fn build_bound_is_sized_from_the_database_pool() { + let local = ConfigDeliveryQueue::for_db_connections(5); + assert_eq!(local.max_concurrent_builds(), 10); + assert_eq!(local.delivery_permits.available_permits(), 64); assert_eq!( ConfigDeliveryQueue::for_db_connections(10).max_concurrent_builds(), 20