detect: a window on a quiet source closes (VEJAS_IDLE_CLOSE_SECS)

CHANGELOG.md

@@ -7,6 +7,18 @@ is `0`, minor versions may carry breaking changes — they are called out here.
 
 ## [Unreleased]
 
+### Fixed — a window on a quiet source closes
+- **A window fed by a source that goes quiet closes**: a detect unit sets an
+  idle grace, `VEJAS_IDLE_CLOSE_SECS` (60 seconds by default, 0 turns it
+  off). Once a subject's type has sent nothing that long, its event time
+  moves on with the wall clock, less the grace, and the unit checks while it
+  has nothing to read. A brute force on a sparse source (VPN logons, one
+  application's log) was raised only when that source spoke again; it is now
+  raised about a grace after its window ends. The types' clocks are in the
+  unit's snapshot, so a restarted unit closes the windows it restored on time
+  too (varpulis #286). `e2e/detect` D7 covers it: three VPN failures, then
+  silence; the previous runtime never raises it.
+
 ## [0.3.2] — 2026-09-23
 
 ### Fixed — detect units, from the Varpulis engine (varpulis #284, #285)

core/Cargo.toml

@@ -22,7 +22,7 @@ webpki-roots = "0.26"
 # The Varpulis CEP engine, embedded as a library (ADR-0031): compile a VPL
 # program, feed it events, publish its emits. No async runtime in its tree —
 # its own CI (scripts/check-engine-deps.py) fails if one ever appears.
-varpulis-engine = { git = "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/varpulis/varpulis", rev = "4e3dbb5eca2c985bfc5b4f568d1118a96123ef45" }
+varpulis-engine = { git = "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/varpulis/varpulis", rev = "92f07b966709bacc042ed97a71e16e8dd625c84d" }
 
 [[bin]]
 name = "vejas-runtime"

core/src/main.rs

@@ -1374,6 +1374,15 @@ fn supervise_vjs(handle: Arc<Handle>, root: PathBuf) {
     });
 }
 
+/// Where a detect unit publishes an emit: the subject its `.to()` names, else
+/// `<root>.detect.<unit>.<stream>`.
+fn emit_subject(emit: &varpulis_engine::Emit, subj_root: &str, stem: &str) -> String {
+    emit.sink
+        .as_ref()
+        .and_then(|s| s.topic.clone())
+        .unwrap_or_else(|| format!("{subj_root}.detect.{stem}.{}", emit.stream))
+}
+
 /// A detect unit (ADR-0031): a VPL program run by the embedded Varpulis engine.
 ///
 /// One durable pull consumer per `.from()` subject, each on a thread of its
@@ -1404,6 +1413,15 @@ fn supervise_detect(handle: Arc<Handle>, _root: PathBuf) {
     );
     let snapshot_every_acks: u64 =
         env::var("VEJAS_SNAPSHOT_ACKS").ok().and_then(|v| v.parse().ok()).unwrap_or(10_000).max(1);
+    // A window whose sources go quiet closes about this long after its end:
+    // once a source has sent nothing for the grace, its event time moves on
+    // with the wall clock (varpulis `Program::set_idle_grace`), at every
+    // batch and on the ticks of an idle unit. 0 turns it off: the unit is
+    // then judged in event time alone, like a replay.
+    let idle_grace = match env::var("VEJAS_IDLE_CLOSE_SECS").ok().and_then(|v| v.parse::<u64>().ok()).unwrap_or(60) {
+        0 => None,
+        secs => Some(Duration::from_secs(secs)),
+    };
     let mut delay = Duration::from_secs(1);
     'outer: loop {
         if !RUNNING.load(Ordering::SeqCst) || handle.stop.load(Ordering::SeqCst) {
@@ -1494,6 +1512,9 @@ fn supervise_detect(handle: Arc<Handle>, _root: PathBuf) {
                     None => {}
                 }
             }
+            // After the restore: a restored window's sources start counting
+            // their silence from now.
+            program.set_idle_grace(idle_grace);
             match resume_seq {
                 Some(seq) => {
                     // The server cannot move an existing consumer: re-create it after the snapshot.
@@ -1588,7 +1609,28 @@ fn supervise_detect(handle: Arc<Handle>, _root: PathBuf) {
                     // the rest without waiting, bounded.
                     let first = match rx.recv_timeout(Duration::from_millis(500)) {
                         Ok(m) => m,
-                        Err(std::sync::mpsc::RecvTimeoutError::Timeout) => continue,
+                        Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
+                            // Nothing to feed: close what time has closed on
+                            // quiet sources. Those emits stand for no message,
+                            // so there is nothing to ack; a crash before the
+                            // next snapshot replays and closes them again.
+                            if idle_grace.is_some() {
+                                match program.tick() {
+                                    Ok(emits) if !emits.is_empty() => {
+                                        for emit in &emits {
+                                            let subject = emit_subject(emit, &subj_root, &stem);
+                                            if let Err(e) = nc.publish(&subject, emit.to_payload()) {
+                                                eprintln!("[vejas] {}: publish {subject}: {e}", handle.spec.name);
+                                            }
+                                        }
+                                        metrics::observe_tick_emits(&handle.spec.name, emits.len() as u64);
+                                    }
+                                    Ok(_) => {}
+                                    Err(e) => eprintln!("[vejas] {}: tick: {e}", handle.spec.name),
+                                }
+                            }
+                            continue;
+                        }
                         Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
                             return Err("the consumer reader stopped".into())
                         }
@@ -1650,13 +1692,7 @@ fn supervise_detect(handle: Arc<Handle>, _root: PathBuf) {
                                 let mut ok = true;
                                 let mut subjects = Vec::with_capacity(emits.len());
                                 for emit in &emits {
-                                    let subject = emit
-                                        .sink
-                                        .as_ref()
-                                        .and_then(|s| s.topic.clone())
-                                        .unwrap_or_else(|| {
-                                            format!("{subj_root}.detect.{stem}.{}", emit.stream)
-                                        });
+                                    let subject = emit_subject(emit, &subj_root, &stem);
                                     if let Err(e) = nc.publish(&subject, emit.to_payload()) {
                                         eprintln!("[vejas] {}: publish {subject}: {e}", handle.spec.name);
                                         ok = false;

core/src/metrics.rs

@@ -80,6 +80,13 @@ pub fn observe(unit: &str, ok: bool, emits: u64, secs: f64) {
     }
 }
 
+/// Emits published by a detect unit's tick: windows that time closed while
+/// nothing arrived. They count as emits, not as processed events.
+pub fn observe_tick_emits(unit: &str, emits: u64) {
+    let mut map = units().lock().unwrap();
+    map.entry(unit.to_string()).or_default().emits += emits;
+}
+
 /// One pull round of the consumer loop and how many messages it returned.
 /// Empty rounds count too: they are the idle long-poll cadence.
 pub fn observe_fetch(unit: &str, messages: usize) {

docs/book/src/concepts/detects.md

@@ -54,9 +54,12 @@ machine it connected to: one alert, on `vx.alerts.lateral`.
   end**: on any event of those types, not only one of the window's own. A
   brute force counted per address on Security events is raised by the next
   Security event of any kind, even when the attacker got in and stopped. A
-  window fed by several types waits for the slowest; one that no event
-  follows stays open, so a source that can go quiet should send a heartbeat
-  of its own type. A stream whose events arrive late declares
+  window fed by several types waits for the slowest. A type that sends
+  nothing for `VEJAS_IDLE_CLOSE_SECS` (60 by default; 0 turns it off) has
+  its event time move on with the wall clock, less that grace, so a brute
+  force on a sparse source (VPN logons) is raised about a minute after its
+  window ends even if the source never speaks again; the unit also checks
+  while it has nothing to read. A stream whose events arrive late declares
   `.watermark(out_of_order: 30s)` to hold its windows that much longer.
 - **Types.** A string `event_type` in the payload names the event type;
   without it, the type is the one the `.from()` binding declares for that

e2e/detect/run.sh

@@ -18,6 +18,9 @@
 #   D6 count close    a count per address closes on the next event of its source
 #                     past the window, from any address: a brute force whose
 #                     attacker got in and stopped is still raised
+#   D7 quiet source   a count on a source that goes silent still closes: past
+#                     the idle grace (VEJAS_IDLE_CLOSE_SECS, 1 s here) the
+#                     source's event time moves on with the wall clock
 #   D3 vpl-check      the engine's verdict on the CLI: ok / refused, with exit codes
 #   D4 topology       /topology lists the units under "detects", lang vpl, running
 #
@@ -118,10 +121,35 @@ stream Brute = Failed
     .to(Bus, topic: "vxt.brute")
 VPL
 
+cat > "$ROOT/detects/quiet.vpl" <<'VPL'
+connector Bus = nats (
+    url: "ignored-by-vejas: the bus is NATS_URL"
+)
+
+event Vpn:
+    ip: str
+    status: str
+
+stream Logons = Vpn
+    .from(Bus, topic: "vxt.vpn")
+
+stream Failed = Vpn
+    .where(status == "failure")
+
+stream Brute = Failed
+    .partition_by(ip)
+    .window(5s)
+    .aggregate(ip: last(ip), n: count())
+    .where(n >= 3)
+    .emit(rule: "vpn_brute_force", ip: ip, n: n)
+    .to(Bus, topic: "vxt.vpnbrute")
+VPL
+
 start_nats() { "$NATSD" -js -sd "$WORK/js" -a 127.0.0.1 -p "$NATS_P" > /dev/null 2>&1 & NATS_PID=$!; sleep 0.5; }
 start_rt() {
   VEJAS_ROOT="$ROOT" NATS_URL="$URL" VEJAS_STREAM=TTEST VEJAS_SUBJECT_ROOT=vxt \
     VEJAS_HTTP_ADDR="127.0.0.1:$HTTP_P" VEJAS_ACK_WAIT_SECS=1 VEJAS_SNAPSHOT_SECS="${SNAP_SECS:-1}" \
+    VEJAS_IDLE_CLOSE_SECS=1 \
     "$BIN" >> "$WORK/rt.log" 2>&1 &
   RT_PID=$!
   until curl -sf -o /dev/null "http://127.0.0.1:$HTTP_P/healthz"; do sleep 0.05; done
@@ -157,14 +185,14 @@ echo "── D4 topology"
 deadline=$((SECONDS+10))
 while [ "$SECONDS" -lt "$deadline" ]; do
   topo=$(curl -s "http://127.0.0.1:$HTTP_P/topology")
-  [ "$(printf '%s' "$topo" | grep -o '"status":"running"' | wc -l)" -ge 3 ] && break
+  [ "$(printf '%s' "$topo" | grep -o '"status":"running"' | wc -l)" -ge 4 ] && break
   sleep 0.2
 done
-python3 - "$topo" <<'PY' && ok "three detects listed, lang vpl, running" || bad "topology: $topo"
+python3 - "$topo" <<'PY' && ok "four detects listed, lang vpl, running" || bad "topology: $topo"
 import json, sys
 t=json.loads(sys.argv[1]); d=t.get("detects", [])
 names=sorted(x["name"] for x in d)
-assert names==["detect:bruteforce","detect:lateral","detect:threshold"], names
+assert names==["detect:bruteforce","detect:lateral","detect:quiet","detect:threshold"], names
 assert all(x["lang"]=="vpl" for x in d), d
 assert all(x["status"]=="running" for x in d), [x["status"] for x in d]
 PY
@@ -243,6 +271,22 @@ assert len(alerts)==1 and alerts[0]["ip"]=="10.0.0.66" and alerts[0]["n"]==3, al
 PY
 kill "$SUB_PID" 2>/dev/null; pkill -f "[s]ub vxt.brute" 2>/dev/null
 
+echo "── D7 a count on a source that goes quiet still closes (idle grace)"
+( timeout 30 "$NATS" -s "$URL" sub vxt.vpnbrute --raw > "$WORK/vpnbrute.txt" 2>/dev/null ) & SUB_PID=$!
+sleep 0.5
+for s in 00 01 02; do
+  "$NATS" -s "$URL" pub vxt.vpn "{\"event_type\":\"Vpn\",\"@timestamp\":\"2026-09-21T19:00:${s}Z\",\"ip\":\"10.0.0.88\",\"status\":\"failure\"}" > /dev/null 2>&1
+done
+# Nothing else is ever published on the VPN subject.
+deadline=$((SECONDS+15)); while [ "$SECONDS" -lt "$deadline" ]; do grep -q '10.0.0.88' "$WORK/vpnbrute.txt" && break; sleep 0.2; done
+sleep 1
+python3 - "$WORK/vpnbrute.txt" <<'PY' && ok "the silent source's brute force is raised once the grace has passed, with its count" || bad "quiet source: $(cat "$WORK/vpnbrute.txt")"
+import json, sys
+alerts=[json.loads(l) for l in open(sys.argv[1]) if l.startswith("{")]
+assert len(alerts)==1 and alerts[0]["ip"]=="10.0.0.88" and alerts[0]["n"]==3, alerts
+PY
+kill "$SUB_PID" 2>/dev/null; pkill -f "[s]ub vxt.vpnbrute" 2>/dev/null
+
 echo "── D5 a stateful sequence survives kill -9 (snapshot, resume by sequence)"
 ( timeout 60 "$NATS" -s "$URL" sub vxt.lateral --raw > "$WORK/lateral5.txt" 2>/dev/null ) & SUB_PID=$!
 sleep 0.5