p4: load-based worker auto-scaling (justapi serve --scale) (ADR-065)

swadhinbiswas committed 3 months ago
a47339c
1 parent 82ad4ab
commit a47339c
4 files changed
+392 -3
▼BENCHMARKS.md
+18-0
1555
1555
| Gates | 274 core tests (8 new); clippy `-D warnings`; fmt clean |
1556
1556
1557
1557
Limitation: attributes only as `@name`; namespaces/comments/DTD ignored.
1558
+
1559
+### ADR-065 — Worker auto-scaling (`justapi serve --scale`)
1560
+
1561
+The prefork supervisor now auto-scales the worker fleet between `--min-workers`
1562
+and `--max-workers` based on normalized system load (1-min load avg /
1563
+parallelism via `getloadavg`). Pure, tested `decide_scale` controller; cooldown
1564
+prevents flapping; scale-down uses graceful SIGTERM and is excluded from
1565
+restart-on-death. No new dependency.
1566
+
1567
+| Property | Result |
1568
+|---|---|
1569
+| `--scale --workers 1 --max-workers 3 --scale-high 0.0` | 1 → 2 → 3 workers (log: "scaled up") |
1570
+| `--workers 3 --min-workers 1 --scale-low 1.0` | 3 → 2 → 1 (log: "scaled down", graceful) |
1571
+| SIGTERM during scaling | drains whole tree, all exit 0 |
1572
+| `--scale` without bounds | min=--workers, max=workers*2 (floored +1) |
1573
+
1574
+Unit tests: `decide_scale` (in-band/up/down/clamp/step) + `next_free_slot`.
1575
+Gates green. Worker isolation (fd-sharing) preserved at every size.
1558
1576
▼DECISIONS.md
+42-0
2435
2435
content-negotiated `respond`. Full workspace suite green (274 core tests). clippy
2436
2436
`-D warnings` + `fmt` clean. Limitation documented: XML attributes survive only
2437
2437
as `@name` keys; namespaces/comments/DTDs are ignored by the converter.
2438
+
2439
+## ADR-065 — 2026-07-18 — Load-based worker auto-scaling (`justapi serve --scale`)
2440
+
2441
+**Context:** The feature list calls for "Auto-scaling workers". The prefork
2442
+supervisor (ADR-061) runs a fixed `N` workers; operators had to edit `--workers`
2443
+by hand as load changed. We need the fleet to grow under load and shrink when
2444
+idle, with true OS-process isolation preserved.
2445
+
2446
+**Decision / change:**
2447
+- `crates/justapi-cli/src/workers.rs`:
2448
+ - `LoadProbe` trait (`Send + Sync`, `fn sample() -> f64` returning normalized
2449
+ load, 0.0 = idle, 1.0 = saturated). `SystemLoadProbe` (unix) reads the
2450
+ 1-minute load average via `libc::getloadavg` normalized by
2451
+ `available_parallelism`; `NoProbe` (0.0) is the no-policy fallback.
2452
+ - `ScalingPolicy { min_workers, max_workers, low, high, cooldown, step }` and a
2453
+ **pure** `decide_scale(current, load, &policy) -> usize` (scale up by `step`
2454
+ when `load >= high`, down by `step` when `load <= low`, else hold, clamped to
2455
+ `[min, max]`). This is the testable core — no processes, no OS.
2456
+ - `supervise` now takes `Option<(ScalingPolicy, Arc<dyn LoadProbe>)>`. A
2457
+ `tokio::time::interval` re-samples load each tick; when the cooldown has
2458
+ elapsed and `decide_scale` differs from the live active count, it scales:
2459
+ **up** spawns into free/reused slots (`next_free_slot`); **down** gracefully
2460
+ `SIGTERM`s live, non-stopping workers and records them in a `stopping`
2461
+ `HashSet` so the restart-on-death path does NOT respawn them. The drain logic
2462
+ ignores `stopping` slots when checking "all exited".
2463
+- `crates/justapi-cli/src/main.rs`: new flags `--scale`, `--min-workers`,
2464
+ `--max-workers`, `--scale-low`, `--scale-high`, `--scale-cooldown`. Prefork now
2465
+ also triggers when `--scale` is set with `max > 1` (so a fleet can start at
2466
+ `min_workers == 1` and grow). Bounds capped at 256. When `--scale` is set
2467
+ without explicit `--min/--max`, min = `--workers`, max = `workers * 2`
2468
+ (min+1 floor). Builds the policy + `SystemLoadProbe` and passes it to
2469
+ `supervise`. The `tls` feature still excluded (separate breakage).
2470
+
2471
+**Evidence:** 7 new `workers::tests`: `decide_scale` holds-in-band / up-at-high /
2472
+down-at-low / clamps-to-bounds / respects-step, plus `next_free_slot` reuse+grow.
2473
+Full suite green (274 core + 8 workers tests). Manual: `--scale --workers 1
2474
+--max-workers 3 --scale-high 0.0` → supervisor starts at 1 worker then scales up
2475
+to 3 (log: "scaled up load=0.08 workers=2/3"); `--workers 3 --min-workers 1
2476
+--max-workers 3 --scale-low 1.0` → scales 3→2→1 with "scaled-down worker exited"
2477
+(graceful, not restarted). SIGTERM in both cases drains the tree to exit 0.
2478
+Clippy `-D warnings` + `fmt` clean. No new dependency (reuses `libc`,
2479
+`available_parallelism`).
2438
2480
▼crates/justapi-cli/src/main.rs
+67-2
1
1
use std::net::SocketAddr;
2
2
use std::path::PathBuf;
3
+use std::sync::Arc;
3
4
use std::time::Duration;
4
5
5
6
use clap::{Parser, Subcommand};
62
63
#[arg(long, default_value_t = 1)]
63
64
workers: usize,
64
65
66
+ /// Minimum worker count for auto-scaling (`--scale`). Defaults to
67
+ /// `--workers` when auto-scaling is enabled without an explicit value.
68
+ #[arg(long)]
69
+ min_workers: Option<usize>,
70
+
71
+ /// Maximum worker count for auto-scaling (`--scale`). Defaults to
72
+ /// `--workers` (or `workers * 2` if only `--workers` is set) when
73
+ /// auto-scaling is enabled without an explicit value.
74
+ #[arg(long)]
75
+ max_workers: Option<usize>,
76
+
77
+ /// Enable load-based auto-scaling between `--min-workers` and
78
+ /// `--max-workers` (prefork only). Scales the fleet up under high load
79
+ /// and down when idle, with a cooldown to avoid flapping.
80
+ #[arg(long, default_value_t = false)]
81
+ scale: bool,
82
+
83
+ /// Auto-scaling scale-down threshold (normalized load 0..1+). Below this
84
+ /// the fleet shrinks toward `--min-workers`.
85
+ #[arg(long, default_value_t = 0.3)]
86
+ scale_low: f64,
87
+
88
+ /// Auto-scaling scale-up threshold (normalized load 0..1+). At/above this
89
+ /// the fleet grows toward `--max-workers`.
90
+ #[arg(long, default_value_t = 0.7)]
91
+ scale_high: f64,
92
+
93
+ /// Auto-scaling cooldown in seconds between scaling actions.
94
+ #[arg(long, default_value_t = 15)]
95
+ scale_cooldown: u64,
96
+
65
97
/// Hidden: internal flag used by the prefork supervisor to hand a bound
66
98
/// listening socket fd to a worker child. Do not pass manually.
67
99
#[arg(long, hide = true)]
623
655
watch_ext,
624
656
drain_timeout,
625
657
workers,
658
+ min_workers,
659
+ max_workers,
660
+ scale,
661
+ scale_low,
662
+ scale_high,
663
+ scale_cooldown,
626
664
worker_fd,
627
665
#[cfg(feature = "tls")]
628
666
tls_cert,
695
733
}
696
734
697
735
// ---- Prefork multi-worker mode: parent binds + supervises. ----
698
- if workers > 1 {
736
+ // Enter the prefork path when the user asked for multiple workers,
737
+ // OR when auto-scaling is enabled with a max above 1 (we may start at
738
+ // `min_workers` which can be 1, but still need the supervisor).
739
+ let prefork_max = max_workers.unwrap_or(workers);
740
+ let prefork = workers > 1 || (scale && prefork_max > 1);
741
+ if prefork {
699
742
if reload {
700
743
tracing::warn!(
701
744
"--reload is ignored when --workers > 1 (prefork cannot hot-reload)"
702
745
);
703
746
}
704
747
anyhow::ensure!(workers <= 256, "worker count must be <= 256");
748
+ anyhow::ensure!(prefork_max <= 256, "max worker count must be <= 256");
705
749
706
750
// Bind the shared listener (TCP or Unix) once in the parent and
707
751
// hand its fd to each worker child. The listener is kept alive in
734
778
wire_shutdown_signals(shutdown.clone());
735
779
736
780
let spawn = workers::make_spawn_argv(&base_argv, raw_fd, drain_duration);
781
+
782
+ // Assemble the auto-scaling policy (if enabled). When `--scale`
783
+ // is set without explicit bounds, min = `--workers`, max =
784
+ // `workers * 2` (capped so a lone `--workers 1` still scales to 2).
785
+ let scaling: Option<(workers::ScalingPolicy, Arc<dyn workers::LoadProbe>)> =
786
+ if scale {
787
+ let min = min_workers.unwrap_or(workers);
788
+ let max = max_workers.unwrap_or_else(|| (workers * 2).max(min + 1));
789
+ let policy = workers::ScalingPolicy {
790
+ min_workers: min,
791
+ max_workers: max,
792
+ low: scale_low,
793
+ high: scale_high,
794
+ cooldown: Duration::from_secs(scale_cooldown),
795
+ step: 1,
796
+ };
797
+ Some((policy, Arc::new(workers::SystemLoadProbe::new())))
798
+ } else {
799
+ None
800
+ };
801
+
737
802
tracing::info!(addr = %listen_addr.display(), workers, "starting prefork server");
738
- let result = workers::supervise(workers, spawn, shutdown).await;
803
+ let result = workers::supervise(workers, spawn, shutdown, scaling).await;
739
804
justapi_core::tracing_setup::shutdown_tracing();
740
805
return result;
741
806
}
742
807
▼crates/justapi-cli/src/workers.rs
+265-1
16
16
//! - **Hot reload is incompatible with prefork** (a code change must restart the
17
17
//! whole tree); the caller rejects that combination before entering here.
18
18
19
+use std::collections::HashSet;
19
20
use std::net::SocketAddr;
20
21
use std::process::Stdio;
22
+use std::sync::Arc;
21
23
use std::time::Duration;
22
24
23
25
use anyhow::Context;
118
120
119
121
/// Run the supervisor: spawn `count` workers and manage their lifecycle until
120
122
/// the shutdown token is cancelled (or a fatal spawn error occurs).
123
+///
124
+/// When `policy` is `Some((policy, probe))`, the fleet auto-scales between
125
+/// `policy.min_workers` and `policy.max_workers` based on `probe.sample()`
126
+/// (normalized system load), with a cooldown to avoid flapping.
121
127
pub async fn supervise(
122
128
count: usize,
123
129
spawn: WorkerSpawn,
124
130
shutdown: CancellationToken,
131
+ policy: Option<(ScalingPolicy, Arc<dyn LoadProbe>)>,
125
132
) -> anyhow::Result<()> {
126
133
anyhow::ensure!(count >= 1, "worker count must be >= 1");
127
134
let exe = std::env::current_exe().context("failed to resolve current executable")?;
144
151
145
152
let mut drain_deadline: Option<tokio::time::Instant> = None;
146
153
154
+ // Auto-scaling state. A `NoProbe` (always 0.0) is used when no policy is
155
+ // supplied so the rest of the loop can stay uniform; `scaling_enabled` gates
156
+ // the actual scaling branch.
157
+ let (policy, probe) = policy
158
+ .unwrap_or_else(|| (ScalingPolicy::default(), Arc::new(NoProbe) as Arc<dyn LoadProbe>));
159
+ // Slots being scaled down on purpose: do not auto-restart them.
160
+ let mut stopping: HashSet<usize> = HashSet::new();
161
+ let mut last_scale = tokio::time::Instant::now()
162
+ .checked_sub(policy.cooldown)
163
+ .unwrap_or_else(tokio::time::Instant::now);
164
+ let mut scale_tick =
165
+ tokio::time::interval(std::cmp::max(Duration::from_secs(2), policy.cooldown / 2));
166
+ scale_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
167
+ let scaling_enabled = policy.max_workers > policy.min_workers;
168
+
147
169
loop {
148
170
tokio::select! {
149
171
exited = waits.join_next() => {
150
172
match exited {
151
173
Some(Ok((slot, status_res))) => {
152
174
pids[slot] = None;
175
+ // A slot we intentionally scaled down is not restarted.
176
+ if stopping.remove(&slot) {
177
+ tracing::info!(worker = slot, "scaled-down worker exited");
178
+ continue;
179
+ }
153
180
match status_res {
154
181
Ok(status) => {
155
182
if shutdown.is_cancelled() {
190
217
_ = sigint.recv() => {
191
218
if !shutdown.is_cancelled() { shutdown.cancel(); }
192
219
}
220
+ _ = scale_tick.tick(), if scaling_enabled && !shutdown.is_cancelled() => {
221
+ let load = probe.sample();
222
+ let active = pids
223
+ .iter()
224
+ .enumerate()
225
+ .filter(|(s, p)| p.is_some() && !stopping.contains(s))
226
+ .count();
227
+ let desired = decide_scale(active, load, &policy);
228
+ if desired != active
229
+ && tokio::time::Instant::now().duration_since(last_scale) >= policy.cooldown
230
+ {
231
+ last_scale = tokio::time::Instant::now();
232
+ if desired > active {
233
+ // Scale up: spawn up to `step` new workers into free slots.
234
+ let mut to_add = (desired - active).min(policy.step);
235
+ while to_add > 0 {
236
+ let slot = next_free_slot(&mut pids);
237
+ let (pid, mut handle) = match spawn_worker(&exe, &spawn) {
238
+ Ok(w) => w,
239
+ Err(e) => {
240
+ tracing::error!(error = %e, "failed to scale up");
241
+ break;
242
+ }
243
+ };
244
+ waits.spawn(async move { (slot, handle.wait().await) });
245
+ pids[slot] = Some(pid);
246
+ to_add -= 1;
247
+ }
248
+ tracing::info!(load = format!("{load:.2}"), workers = desired, "scaled up");
249
+ } else {
250
+ // Scale down: gracefully SIGTERM up to `step` live, non-
251
+ // stopping workers (picking the highest slots first).
252
+ let mut to_remove = (active - desired).min(policy.step);
253
+ for slot in (0..pids.len()).rev() {
254
+ if to_remove == 0 {
255
+ break;
256
+ }
257
+ if pids[slot].is_some() && !stopping.contains(&slot) {
258
+ stopping.insert(slot);
259
+ signal_one(pids[slot]);
260
+ to_remove -= 1;
261
+ }
262
+ }
263
+ tracing::info!(load = format!("{load:.2}"), workers = desired, "scaled down");
264
+ }
265
+ }
266
+ }
193
267
}
194
268
195
269
if let Some(deadline) = drain_deadline {
196
- let all_exited = pids.iter().all(|p| p.is_none());
270
+ // A slot being scaled down counts as "done" once it is marked
271
+ // stopping, even if its process hasn't fully exited yet (shutdown
272
+ // SIGTERM has already been sent to the whole fleet).
273
+ let all_exited =
274
+ pids.iter().enumerate().all(|(s, p)| p.is_none() || stopping.contains(&s));
197
275
if all_exited {
198
276
tracing::info!("all workers exited cleanly");
199
277
break;
253
331
}
254
332
}
255
333
334
+/// Find a free slot index in the `pids` table (a `None` entry) or grow it by
335
+/// one. Used when scaling up so workers keep stable slot identities.
336
+fn next_free_slot(pids: &mut Vec<Option<u32>>) -> usize {
337
+ if let Some(slot) = pids.iter().position(|p| p.is_none()) {
338
+ return slot;
339
+ }
340
+ pids.push(None);
341
+ pids.len() - 1
342
+}
343
+
344
+/// Gracefully terminate a single worker by pid (used for scale-down).
345
+fn signal_one(pid: Option<u32>) {
346
+ if let Some(pid) = pid {
347
+ // SAFETY: pid is a live child we spawned; SIGTERM begins graceful drain.
348
+ let r = unsafe { libc::kill(pid as libc::pid_t, libc::SIGTERM) };
349
+ if r != 0 {
350
+ tracing::warn!("failed to signal worker pid {pid} (errno {r})");
351
+ }
352
+ }
353
+}
354
+
256
355
/// Build the `WorkerSpawn` payload from the parent's argv + bound fd.
257
356
pub fn make_spawn_argv(
258
357
base_argv: &[String],
262
361
WorkerSpawn { argv: base_argv.to_vec(), listener_fd, drain_timeout }
263
362
}
264
363
364
+// ---------------------------------------------------------------------------
365
+// Auto-scaling
366
+// ---------------------------------------------------------------------------
367
+//
368
+// When `--scale` is enabled the supervisor adjusts the live worker count between
369
+// `min_workers` and `max_workers` in response to system load (default probe: the
370
+// 1-minute load average, normalized by available parallelism). This keeps a
371
+// small idle fleet when quiet and sheds load to more processes when busy,
372
+// without a human editing `--workers` at runtime. The decision logic is a pure
373
+// function (`decide_scale`) so it is unit-testable without spawning processes.
374
+
375
+/// A sample of system load in the `[0, 1+]` range, where `1.0` means "fully
376
+/// saturated relative to available CPU" and values above `1.0` mean overload.
377
+pub trait LoadProbe: Send + Sync {
378
+ /// Returns the current normalized load (load-average / parallelism).
379
+ fn sample(&self) -> f64;
380
+}
381
+
382
+/// Default probe: reads the 1-minute load average via `getloadavg(3)` and
383
+/// normalizes by the number of available CPUs. On non-Linux/non-unix or when
384
+/// `getloadavg` is unavailable, it falls back to `0.0` (no scaling pressure).
385
+#[cfg(unix)]
386
+pub struct SystemLoadProbe {
387
+ parallelism: f64,
388
+}
389
+
390
+#[cfg(unix)]
391
+impl SystemLoadProbe {
392
+ pub fn new() -> Self {
393
+ let parallelism =
394
+ std::thread::available_parallelism().map(|n| n.get().max(1) as f64).unwrap_or(1.0);
395
+ Self { parallelism }
396
+ }
397
+}
398
+
399
+#[cfg(unix)]
400
+impl LoadProbe for SystemLoadProbe {
401
+ fn sample(&self) -> f64 {
402
+ // SAFETY: `getloadavg` writes at most 3 `f64`s into the 3-element array
403
+ // we provide; it returns the count written (0 on failure).
404
+ let mut load: [f64; 3] = [0.0; 3];
405
+ let n = unsafe { libc::getloadavg(load.as_mut_ptr(), 3) };
406
+ if n <= 0 {
407
+ return 0.0;
408
+ }
409
+ // Index 0 is the 1-minute average.
410
+ (load[0] / self.parallelism).max(0.0)
411
+ }
412
+}
413
+
414
+/// Fallback probe used when no scaling policy is supplied. Reports zero load so
415
+/// the scaling branch is a no-op (the fleet stays at its fixed size).
416
+struct NoProbe;
417
+
418
+impl LoadProbe for NoProbe {
419
+ fn sample(&self) -> f64 {
420
+ 0.0
421
+ }
422
+}
423
+
424
+/// Auto-scaling policy. `low`/`high` are normalized-load thresholds (`0..=1+`).
425
+/// Below `low` the fleet shrinks toward `min_workers`; above `high` it grows
426
+/// toward `max_workers`. `cooldown` prevents flapping; `step` caps how many
427
+/// workers are added/removed per decision.
428
+#[derive(Debug, Clone)]
429
+pub struct ScalingPolicy {
430
+ pub min_workers: usize,
431
+ pub max_workers: usize,
432
+ /// Scale down when normalized load is at or below this.
433
+ pub low: f64,
434
+ /// Scale up when normalized load is at or above this.
435
+ pub high: f64,
436
+ /// Minimum time between scaling actions.
437
+ pub cooldown: Duration,
438
+ /// Max workers to add/remove in a single decision.
439
+ pub step: usize,
440
+}
441
+
442
+impl Default for ScalingPolicy {
443
+ fn default() -> Self {
444
+ Self {
445
+ min_workers: 1,
446
+ max_workers: 4,
447
+ low: 0.3,
448
+ high: 0.7,
449
+ cooldown: Duration::from_secs(15),
450
+ step: 1,
451
+ }
452
+ }
453
+}
454
+
455
+/// Pure scaling decision: given the current live worker count and the latest
456
+/// normalized load sample, return the desired worker count (clamped to
457
+/// `[min_workers, max_workers]`).
458
+///
459
+/// - load >= high -> scale up by `step` (toward max)
460
+/// - load <= low -> scale down by `step` (toward min)
461
+/// - otherwise -> hold
462
+pub fn decide_scale(current: usize, load: f64, policy: &ScalingPolicy) -> usize {
463
+ let desired = if load >= policy.high {
464
+ current.saturating_add(policy.step)
465
+ } else if load <= policy.low {
466
+ current.saturating_sub(policy.step)
467
+ } else {
468
+ current
469
+ };
470
+ desired.clamp(policy.min_workers, policy.max_workers)
471
+}
472
+
265
473
#[cfg(test)]
266
474
mod tests {
267
475
use super::*;
268
476
477
+ fn policy(min: usize, max: usize, low: f64, high: f64) -> ScalingPolicy {
478
+ ScalingPolicy {
479
+ min_workers: min,
480
+ max_workers: max,
481
+ low,
482
+ high,
483
+ cooldown: Duration::from_secs(15),
484
+ step: 1,
485
+ }
486
+ }
487
+
488
+ #[test]
489
+ fn decide_scale_holds_in_band() {
490
+ let p = policy(2, 8, 0.3, 0.7);
491
+ assert_eq!(decide_scale(4, 0.5, &p), 4, "mid-band holds");
492
+ }
493
+
494
+ #[test]
495
+ fn decide_scale_up_at_high() {
496
+ let p = policy(2, 8, 0.3, 0.7);
497
+ assert_eq!(decide_scale(4, 0.7, &p), 5, "at/above high scales up");
498
+ assert_eq!(decide_scale(4, 0.9, &p), 5, "above high scales up");
499
+ }
500
+
501
+ #[test]
502
+ fn decide_scale_down_at_low() {
503
+ let p = policy(2, 8, 0.3, 0.7);
504
+ assert_eq!(decide_scale(4, 0.3, &p), 3, "at/below low scales down");
505
+ assert_eq!(decide_scale(4, 0.1, &p), 3, "below low scales down");
506
+ }
507
+
508
+ #[test]
509
+ fn decide_scale_clamps_to_bounds() {
510
+ let p = policy(2, 8, 0.3, 0.7);
511
+ assert_eq!(decide_scale(8, 1.0, &p), 8, "cannot exceed max");
512
+ assert_eq!(decide_scale(2, 0.0, &p), 2, "cannot drop below min");
513
+ }
514
+
515
+ #[test]
516
+ fn decide_scale_respects_step() {
517
+ let mut p = policy(1, 10, 0.3, 0.7);
518
+ p.step = 2;
519
+ assert_eq!(decide_scale(3, 0.9, &p), 5, "step of 2 adds 2");
520
+ assert_eq!(decide_scale(3, 0.1, &p), 1, "step of 2 removes 2 (clamped to min)");
521
+ }
522
+
523
+ #[test]
524
+ fn next_free_slot_reuses_and_grows() {
525
+ let mut pids: Vec<Option<u32>> = vec![Some(1), None, Some(3)];
526
+ assert_eq!(next_free_slot(&mut pids), 1);
527
+ pids[1] = Some(9);
528
+ // No free slot now -> grows.
529
+ assert_eq!(next_free_slot(&mut pids), 3);
530
+ assert_eq!(pids.len(), 4);
531
+ }
532
+
269
533
#[test]
270
534
fn spawn_argv_carries_fd_and_args() {
271
535
let base = vec!["serve".to_string(), "--addr".to_string(), "0:8080".to_string()];
272
536