Skip to content

Commit c906298

Browse files
feat(router): kill workers on clean shutdown
Workers held RocksDB locks and the binary image open after the router exited, causing EPERM failures when npx tried to replace the cached binary on the next update. - Add `Registry::kill_all` to drain the map and reap every Ready worker process deterministically on Ctrl+C / SIGTERM - Return `ProxyCtx` from `build_router_app` so the shutdown handler in `main.rs` can reach the registry without extra Arc plumbing - Wire the Job Object kill-on-close as a backstop for the npx path where Node hard-terminates the router before the signal handler runs - Add tests: `kill_all` empties the registry and reaps children; no-op on an empty registry
1 parent 75e2dcb commit c906298

6 files changed

Lines changed: 267 additions & 23 deletions

File tree

‎npm/vibervn-context-engine/bin/cli.js‎

Lines changed: 92 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
#!/usr/bin/env node
22

3-
const { execFileSync } = require('child_process')
3+
const { spawn, execFileSync } = require('child_process')
44
const { existsSync } = require('fs')
55
const path = require('path')
66

@@ -48,15 +48,96 @@ if (!existsSync(binPath)) {
4848
process.exit(1)
4949
}
5050

51-
try {
52-
execFileSync(binPath, process.argv.slice(2), {
53-
stdio: 'inherit',
54-
env: process.env,
55-
})
56-
process.exit(0)
57-
} catch (e) {
58-
if (e.signal) {
59-
process.kill(process.pid, e.signal)
51+
// Run the binary as a CHILD we own, not a synchronous block, so we can forward
52+
// stop signals and — critically on Windows — tear down the WHOLE process tree.
53+
//
54+
// Why this matters: the binary runs in "router" mode and spawns one child
55+
// "worker" process per repo. Those workers hold a RocksDB lock AND keep the
56+
// `context-engine-rs.exe` image open. A worker only self-exits after an idle
57+
// window (default 300s), so if the router dies but its workers linger, the next
58+
// `npx vibervn-context-engine@latest` cannot replace the cached binary and npm
59+
// fails with `EPERM: operation not permitted, unlink ...context-engine-rs.exe`.
60+
//
61+
// On Windows there is no real SIGINT to forward: Node's `child.kill('SIGINT')`
62+
// maps to TerminateProcess, which kills ONLY the router and leaves the workers
63+
// orphaned. So on Windows we kill the whole tree with `taskkill /T /F` (the
64+
// router is still alive when we snapshot, so its workers are in the tree). On
65+
// POSIX we forward the real signal to the child, which the router handles to
66+
// kill its workers before exiting. The binary's Job Object (Windows
67+
// kill-on-close) remains the backstop for anything either path races past.
68+
const child = spawn(binPath, process.argv.slice(2), {
69+
stdio: 'inherit',
70+
env: process.env,
71+
})
72+
73+
// Stop signals we forward / translate. Listed once so we can both register and
74+
// (in the exit handler) DE-register them — see the re-raise note below.
75+
const STOP_SIGNALS = ['SIGINT', 'SIGTERM', 'SIGHUP', 'SIGBREAK']
76+
77+
let terminating = false
78+
let exited = false
79+
80+
function terminate(signal) {
81+
if (terminating) return
82+
terminating = true
83+
84+
// Guard: if the child is already gone, do NOTHING. Two reasons:
85+
// - there is nothing left to kill, and
86+
// - on Windows `taskkill /T /F <pid>` against a dead PID could match a REUSED
87+
// pid and kill an unrelated process tree. `exited` is set by the 'exit'
88+
// handler; `child.exitCode !== null` also means it has exited; `child.pid`
89+
// is undefined if spawn never succeeded (the 'error' handler ran).
90+
if (exited || child.exitCode !== null || child.pid === undefined) return
91+
92+
if (process.platform === 'win32') {
93+
// Kill the router AND its worker children as one tree. /F so a worker that
94+
// ignores the polite close still dies and releases the .exe + RocksDB lock.
95+
// The child is still alive here (guarded above), so its workers are in the
96+
// tree snapshot taskkill takes. Wrapped: a child that dies in the gap makes
97+
// taskkill exit non-zero (execFileSync throws) — caught, then a best-effort
98+
// direct kill (also caught) so the handler can never crash the wrapper.
99+
try {
100+
execFileSync('taskkill', ['/pid', String(child.pid), '/T', '/F'], {
101+
stdio: 'ignore',
102+
})
103+
} catch {
104+
try {
105+
child.kill()
106+
} catch {}
107+
}
108+
} else {
109+
// POSIX: forward the real signal; the router kills its workers, then exits.
110+
try {
111+
child.kill(signal || 'SIGTERM')
112+
} catch {}
60113
}
61-
process.exit(e.status ?? 1)
62114
}
115+
116+
// Forward the usual stop signals. SIGINT = Ctrl+C; SIGTERM = supervisor/kill;
117+
// SIGHUP = terminal closed (POSIX). Windows only delivers SIGINT/SIGBREAK here.
118+
// Each is registered exactly once; the `terminating` guard makes the body
119+
// idempotent if several arrive (or one arrives twice).
120+
for (const sig of STOP_SIGNALS) {
121+
process.on(sig, () => terminate(sig))
122+
}
123+
124+
child.on('error', (err) => {
125+
console.error(`Failed to launch context-engine binary: ${err.message}`)
126+
process.exit(1)
127+
})
128+
129+
child.on('exit', (code, signal) => {
130+
exited = true
131+
if (signal) {
132+
// Re-raise the signal on ourselves so our exit status reflects it, matching
133+
// normal child-process signal semantics. We MUST drop our own signal
134+
// listeners first: otherwise re-raising e.g. SIGINT would just re-enter
135+
// `terminate()` (a no-op now, but it would also stop the default
136+
// termination), leaving the wrapper hung instead of exiting. (Mostly a POSIX
137+
// path — on Windows child death surfaces as an exit code, not a signal.)
138+
for (const sig of STOP_SIGNALS) process.removeAllListeners(sig)
139+
process.kill(process.pid, signal)
140+
return
141+
}
142+
process.exit(code ?? 0)
143+
})

‎src/main.rs‎

Lines changed: 89 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
use std::future::IntoFuture;
12
use std::path::PathBuf;
23

34
use clap::Parser;
@@ -254,7 +255,7 @@ async fn run_router(cli: &Cli, bind: &str) {
254255
// and the embedding/data dirs, but it does NOT start an IndexEngine or open
255256
// any repo DB — that is the whole point. `boot_router` loads settings and
256257
// resolves dirs only.
257-
let app = match router::build_router_app(router::RouterBootOptions {
258+
let (app, proxy) = match router::build_router_app(router::RouterBootOptions {
258259
data_dir: cli.data_dir.clone(),
259260
embeddings_dir: cli.embeddings_dir.clone(),
260261
bind: bind.to_owned(),
@@ -263,7 +264,7 @@ async fn run_router(cli: &Cli, bind: &str) {
263264
})
264265
.await
265266
{
266-
Ok(app) => app,
267+
Ok(pair) => pair,
267268
Err(e) => {
268269
eprintln!("error: {e:#}");
269270
std::process::exit(2);
@@ -278,8 +279,90 @@ async fn run_router(cli: &Cli, bind: &str) {
278279
});
279280

280281
info!("Context Engine router listening on http://{addr}");
281-
axum::serve(listener, app).await.unwrap_or_else(|e| {
282-
eprintln!("router server error: {e}");
283-
std::process::exit(1);
284-
});
282+
283+
// Graceful stop: when a shutdown signal arrives, kill every live worker
284+
// FIRST (releasing each worker's RocksDB LOCK + its hold on the binary image)
285+
// and only then exit. Without this, the router would die but its
286+
// idle-but-alive workers would keep running for up to `worker_idle_secs`
287+
// (default 300s), pinning `context-engine-rs.exe` on Windows so a follow-up
288+
// `npx ...@latest` cannot replace the cached binary (EPERM unlink).
289+
//
290+
// We RACE the server against the signal rather than using
291+
// `with_graceful_shutdown` (which drains in-flight connections AFTER the
292+
// signal): an idle SSE/keep-alive connection — e.g. a browser holding
293+
// `/api/repos/:id/index-events` — would otherwise make Ctrl+C hang. The
294+
// worker teardown is crash-safe (RocksDB LOCK releases on process exit;
295+
// `file_meta` self-heals the next run), so a hard kill + immediate exit is
296+
// the correct, responsive stop. `process::exit(0)` then drops the Job Object
297+
// handle → kill-on-close catches any worker `kill_all` raced past (e.g. one
298+
// mid-spawn). See `router::registry::Registry::kill_all` + `router::jobobject`.
299+
//
300+
// This deterministic teardown runs only on paths where the router actually
301+
// receives the signal: `just rs-dev`, or launching the binary directly. On
302+
// the npx path the Node wrapper hard-TERMINATES the router (Windows has no
303+
// real SIGINT to forward), so this never runs there — that path is covered by
304+
// the wrapper's `taskkill /T` over the whole tree plus the Job Object.
305+
let server = axum::serve(listener, app).into_future();
306+
tokio::select! {
307+
res = server => {
308+
res.unwrap_or_else(|e| {
309+
eprintln!("router server error: {e}");
310+
std::process::exit(1);
311+
});
312+
}
313+
_ = shutdown_signal() => {
314+
info!("shutdown signal received; killing live workers");
315+
proxy.registry.kill_all().await;
316+
info!("workers killed; router exiting");
317+
std::process::exit(0);
318+
}
319+
}
320+
}
321+
322+
/// Resolve once any OS stop signal arrives. On Windows we additionally wake on
323+
/// the console-close / logoff / shutdown control events (Ctrl+C alone does not
324+
/// cover a closed console window). On Unix we also wake on SIGTERM (the signal a
325+
/// supervisor/`kill` sends), not just Ctrl+C (SIGINT). First signal wins.
326+
async fn shutdown_signal() {
327+
#[cfg(windows)]
328+
{
329+
use tokio::signal::windows;
330+
// `ctrl_c` covers Ctrl+C; the others cover the console window being
331+
// closed / the user logging off / the system shutting down. Any one of
332+
// them should trigger the same orderly worker teardown.
333+
let mut ctrl_close = match windows::ctrl_close() {
334+
Ok(s) => s,
335+
Err(_) => {
336+
let _ = tokio::signal::ctrl_c().await;
337+
return;
338+
}
339+
};
340+
let mut ctrl_shutdown = windows::ctrl_shutdown().ok();
341+
let mut ctrl_logoff = windows::ctrl_logoff().ok();
342+
tokio::select! {
343+
_ = tokio::signal::ctrl_c() => {}
344+
_ = ctrl_close.recv() => {}
345+
_ = async { match ctrl_shutdown.as_mut() { Some(s) => { s.recv().await; }, None => std::future::pending().await } } => {}
346+
_ = async { match ctrl_logoff.as_mut() { Some(s) => { s.recv().await; }, None => std::future::pending().await } } => {}
347+
}
348+
}
349+
#[cfg(unix)]
350+
{
351+
use tokio::signal::unix::{SignalKind, signal};
352+
let mut term = match signal(SignalKind::terminate()) {
353+
Ok(s) => s,
354+
Err(_) => {
355+
let _ = tokio::signal::ctrl_c().await;
356+
return;
357+
}
358+
};
359+
tokio::select! {
360+
_ = tokio::signal::ctrl_c() => {}
361+
_ = term.recv() => {}
362+
}
363+
}
364+
#[cfg(not(any(windows, unix)))]
365+
{
366+
let _ = tokio::signal::ctrl_c().await;
367+
}
285368
}

‎src/router/mod.rs‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -79,7 +79,12 @@ pub struct RouterState {
7979

8080
/// Build the router-mode axum app: load settings + resolve dirs (NO IndexEngine,
8181
/// NO repo DB opens), create the Job Object, and wire global + proxy routes.
82-
pub async fn build_router_app(opts: RouterBootOptions) -> Result<Router> {
82+
///
83+
/// Returns the wired `Router` plus a clone of the [`ProxyCtx`] so the caller's
84+
/// shutdown handler can reach the worker [`registry::Registry`] (to kill live
85+
/// workers on Ctrl+C). The router's own `RouterState` keeps the original; the
86+
/// returned clone shares the same `Arc`-backed registry + Job Object.
87+
pub async fn build_router_app(opts: RouterBootOptions) -> Result<(Router, ProxyCtx)> {
8388
// The router does not open RocksDB, but a worker it spawns will — and the
8489
// worker reads these same env-derived bounds at its own boot. Setting them
8590
// here is harmless and keeps parity if the router is ever extended.
@@ -230,6 +235,11 @@ pub async fn build_router_app(opts: RouterBootOptions) -> Result<Router> {
230235
});
231236
}
232237

238+
// Clone the proxy context BEFORE `state` is moved into `.with_state(state)`
239+
// below — the caller's shutdown handler needs it to reach the worker registry.
240+
// Shares the same Arc-backed registry + Job Object as the router's state.
241+
let proxy = state.proxy.clone();
242+
233243
let app = Router::new()
234244
// ── Global endpoints served NATIVELY (no worker, no DB) ──────────────
235245
.route("/", get(serve_index))
@@ -305,7 +315,7 @@ pub async fn build_router_app(opts: RouterBootOptions) -> Result<Router> {
305315
// Global multi-repo `/mcp`: proxying handler (repo per call).
306316
.merge(Router::new().nest_service("/mcp", mcp_service));
307317

308-
Ok(app)
318+
Ok((app, proxy))
309319
}
310320

311321
// ── Native global handlers ──────────────────────────────────────────────────

‎src/router/registry.rs‎

Lines changed: 70 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -350,6 +350,38 @@ impl Registry {
350350
}
351351
}
352352

353+
/// Kill EVERY live worker and block until each is reaped. Used by the
354+
/// router's shutdown handler (Ctrl+C / SIGTERM / console-close) so a clean
355+
/// stop releases every worker's RocksDB LOCK *and* its hold on the binary
356+
/// image before the router exits — otherwise an idle-but-alive worker keeps
357+
/// running for up to `worker_idle_secs` (default 300s) after the router is
358+
/// gone, which on Windows pins `context-engine-rs.exe` and makes a follow-up
359+
/// `npx ...@latest` fail to replace the cached binary (EPERM unlink).
360+
///
361+
/// This is the DETERMINISTIC teardown for paths where the router actually
362+
/// receives the signal (`just rs-dev`, running the binary directly). On the
363+
/// npx path the Node wrapper hard-terminates the router (Windows has no real
364+
/// SIGINT), so this handler may never run there — that path is covered by the
365+
/// wrapper's `taskkill /T` (whole tree) plus the Job Object kill-on-close
366+
/// backstop. Belt-and-suspenders: drain the map first so a concurrent
367+
/// `acquire`/`reap` can't resurrect or double-reap an entry mid-shutdown.
368+
pub async fn kill_all(&self) {
369+
let entries: Vec<WorkerState> = {
370+
let mut map = self.inner.write().await;
371+
map.drain().map(|(_repo, state)| state).collect()
372+
};
373+
for state in entries {
374+
if let WorkerState::Ready(w) = state {
375+
let mut child = w.child.lock().await;
376+
let _ = child.kill();
377+
let _ = child.wait();
378+
}
379+
// `Spawning` holds no child handle here (the driver owns it until it
380+
// publishes), so there is nothing to kill — the Job Object backstop
381+
// covers a worker that was mid-spawn when shutdown hit.
382+
}
383+
}
384+
353385
/// Snapshot of repos that currently have a Ready worker (for diagnostics /
354386
/// the router's "which repos are warm" view).
355387
pub async fn ready_repos(&self) -> Vec<String> {
@@ -666,8 +698,44 @@ mod tests {
666698
w.kill_and_wait().await; // cleanup
667699
}
668700

669-
/// (f) ABA SAFETY: the watchdog reaps a slow spawn's `Spawning(ready_A)`, a
670-
/// waiter re-elects a fresh `Spawning(ready_B)`, then the original slow driver
701+
/// (g) SHUTDOWN: `kill_all` terminates every live `Ready` worker and empties
702+
/// the registry, so a clean router stop releases each worker's RocksDB LOCK +
703+
/// its hold on the binary image (the EPERM-on-update root cause) instead of
704+
/// leaving idle-but-alive workers running until their 300s idle timer fires.
705+
#[tokio::test]
706+
async fn kill_all_kills_live_workers_and_empties_registry() {
707+
let reg = Registry::new();
708+
let w1 = live_ready_worker();
709+
let w2 = live_ready_worker();
710+
install_ready(&reg, "repoK1", w1.clone()).await;
711+
install_ready(&reg, "repoK2", w2.clone()).await;
712+
713+
reg.kill_all().await;
714+
715+
// Registry is fully drained.
716+
assert!(
717+
reg.ready_repos().await.is_empty() && reg.spawning_repos().await.is_empty(),
718+
"kill_all must drain every entry"
719+
);
720+
// Both children are actually dead: try_wait reports an exit status.
721+
for w in [&w1, &w2] {
722+
let mut child = w.child.lock().await;
723+
assert!(
724+
matches!(child.try_wait(), Ok(Some(_))),
725+
"kill_all must have killed + reaped the worker child"
726+
);
727+
}
728+
}
729+
730+
/// (g') `kill_all` on an empty registry is a no-op (idempotent / safe to call
731+
/// even when no workers ever spawned).
732+
#[tokio::test]
733+
async fn kill_all_on_empty_registry_is_noop() {
734+
let reg = Registry::new();
735+
reg.kill_all().await;
736+
assert!(reg.ready_repos().await.is_empty());
737+
}
738+
671739
/// finally completes. `publish_ready(ready_A)` must report SUPERSEDED (false)
672740
/// and must NOT clobber `ready_B` into `Ready` (that would leave two workers
673741
/// racing one LOCK). The re-elected `Spawning(ready_B)` stays intact.

‎tests/e2e_worker_lifecycle.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -74,7 +74,8 @@ async fn start_router(home: &TempDir) -> SocketAddr {
7474
worker_exe: Some(worker_exe()),
7575
})
7676
.await
77-
.expect("router app");
77+
.expect("router app")
78+
.0;
7879
tokio::spawn(async move {
7980
axum::serve(listener, app).await.expect("serve");
8081
});

‎tests/router_integration.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,8 @@ async fn start_router(home: &TempDir) -> SocketAddr {
4343
worker_exe: None,
4444
})
4545
.await
46-
.expect("router app builds");
46+
.expect("router app builds")
47+
.0;
4748
tokio::spawn(async move {
4849
axum::serve(listener, app).await.expect("router serve");
4950
});

0 commit comments

Comments
 (0)