Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 3 additions & 1 deletion core/src/control.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
44 changes: 43 additions & 1 deletion core/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<AtomicBool>);
Expand Down Expand Up @@ -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()));
}

{
Expand Down Expand Up @@ -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));
Expand Down
8 changes: 7 additions & 1 deletion e2e/http-in-bound.sh
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Expand All @@ -97,13 +98,18 @@ 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
wait_ingest $IN2 && ok "http-in survives 300 connections on a pids limit of 64" \
|| 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
Expand Down
Loading