diff --git a/CHANGELOG.md b/CHANGELOG.md index b1d15e6..be117b3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,13 @@ is `0`, minor versions may carry breaking changes — they are called out here. ## [Unreleased] +### Fixed +- **A burst of connections during boot could end the version watch for good.** Its + first connection panicked when the OS had no thread to give: nothing showed it, + and a promoted version no longer reached that instance until a restart. It + needs a thread limit and a burst at that moment. The watch and the control loop + now carry a name and start again after a panic. + ## [0.4.0] — 2026-11-06 A security release: upgrade from any 0.3.x, and rotate any secret a shared or diff --git a/core/src/control.rs b/core/src/control.rs index 7bb4b66..65f724e 100644 --- a/core/src/control.rs +++ b/core/src/control.rs @@ -27,7 +27,9 @@ pub fn start(registry: Registry, root: PathBuf) { let Some(tenant) = env::var("VEJAS_TENANT").ok().filter(|t| !t.is_empty()) else { return; }; - thread::spawn(move || supervise(registry, root, tenant)); + crate::background("control", move || { + supervise(registry.clone(), root.clone(), tenant.clone()) + }); } fn supervise(registry: Registry, root: PathBuf, tenant: String) { diff --git a/core/src/main.rs b/core/src/main.rs index 1092efd..b056160 100644 --- a/core/src/main.rs +++ b/core/src/main.rs @@ -1929,6 +1929,32 @@ fn contained(attempt: impl FnOnce() -> Result<(), String>) -> Result<(), String> }) } +/// Start a background loop of the runtime itself, which no supervisor watches. +/// `nats::connect` spawns threads and panics when the OS has none, and that used +/// to end the loop for good with nothing to show for it: a burst of connections +/// during boot took `version_watch`, and a promoted version no longer reached this +/// instance. The thread carries a name, and a panic starts the loop again. +fn background(name: &'static str, body: impl Fn() + Send + 'static) { + let spawned = thread::Builder::new().name(name.into()).spawn(move || { + while RUNNING.load(Ordering::SeqCst) { + let run = contained(|| { + body(); + Ok(()) + }); + match run { + Ok(()) => break, + Err(e) => { + eprintln!("[vejas] {name}: {e}; starting it again"); + thread::sleep(Duration::from_secs(2)); + } + } + } + }); + if let Err(e) = spawned { + eprintln!("[vejas] {name}: cannot start its thread: {e}"); + } +} + /// Raises a flag when dropped, so that a helper thread stops on every way out of /// an attempt, a panic included. struct StopOnDrop(Arc); @@ -4709,7 +4735,7 @@ fn main() { { let registry = registry.clone(); let root = root.clone(); - thread::spawn(move || version_watch(registry, root)); + background("version-watch", move || version_watch(registry.clone(), root.clone())); } { @@ -4741,6 +4767,22 @@ mod supervision_tests { assert_eq!(err, "panic: boom 7"); } + #[test] + fn a_background_loop_starts_again_after_a_panic() { + let (tx, rx) = std::sync::mpsc::channel(); + let runs = AtomicU64::new(0); + background("test-loop", move || { + let n = runs.fetch_add(1, Ordering::SeqCst); + tx.send(n).unwrap(); + if n == 0 { + panic!("no thread to give"); + } + }); + let timeout = Duration::from_secs(10); + assert_eq!(rx.recv_timeout(timeout), Ok(0)); + assert_eq!(rx.recv_timeout(timeout), Ok(1), "the loop came back"); + } + #[test] fn a_slot_comes_back_when_its_thread_panics() { let open = Arc::new(std::sync::atomic::AtomicUsize::new(0)); diff --git a/e2e/http-in-bound.sh b/e2e/http-in-bound.sh index d7f2625..501712f 100755 --- a/e2e/http-in-bound.sh +++ b/e2e/http-in-bound.sh @@ -77,7 +77,8 @@ case "$C" in esac wait $HOLD 2>/dev/null wait_ingest $IN && ok "ingestion is back once the connections are gone" || ko "http-in did not recover" -curl -s -m 5 "$H/metrics" | grep -q 'vejas_http_rejected_total{unit="connector:http_in"}' \ +METRICS=$(curl -s -m 5 "$H/metrics") +[[ "$METRICS" == *'vejas_http_rejected_total{unit="connector:http_in"}'* ]] \ && ok "the refusals are on /metrics" || ko "no vejas_http_rejected_total on /metrics" topology "$H" | grep -q 'running' && ok "the connector still reports running" || ko "the connector left running" @@ -97,6 +98,9 @@ else debian:trixie-slim vejas-runtime > /dev/null for _ in $(seq 1 100); do curl -sf -o /dev/null "$H2/healthz" && break; sleep 0.1; done wait_ingest $IN2 || ko "http-in never came up in the container" + # Past the boot: a burst that lands on the first NATS connections is another + # case, covered by background(), and it made this check flaky. + sleep 5 hold $IN2 300 3; HOLD=$! sleep 1 wait $HOLD 2>/dev/null @@ -104,6 +108,8 @@ else || ko "http-in died on the pids limit" P=$(docker logs "$CT" 2>&1 | grep -c panicked) [ "$P" = "0" ] && ok "no panic in the log" || ko "$P panic(s) in the log" + W=$(docker exec "$CT" sh -c 'cat /proc/1/task/*/comm' | grep -c version-watch) + [ "$W" -ge 1 ] && ok "the version watch is alive" || ko "the version watch died" UNIT=$(topology "$H2") echo "$UNIT" | grep -q '"restarts": 0' && ok "no restart was needed" || ko "the connector restarted: $UNIT" fi