russell@unturf.com
3e47814cf3
shopt -s nullglob (set so outer for-loop tolerates empty *.bin) made an unmatched .inflight-* glob expand to nothing. Bare "ls >/dev/null" then succeeded by listing CWD, and our if-branch incorrectly skipped every orphan whose .inflight-* did not match. Net effect: medium-sized orphan bins (100 MiB+, valid QECCOPS1 magic, no markers) accumulated in our queue indefinitely across pool restarts. Replace with compgen -G which returns success only when our pattern matches, immune to nullglob. Adds tests/integration/test-heal-orphan-bins.sh as TCRAUDT reducer covering our contract: medium-size orphan promotes to .ready, sub-threshold truncated to .done, bad-magic DLQs to dlq/. Incident 2026-06-14: 8 vecC-tri-*-w8c.bin orphans (493 MiB each, Jun 12 emit) survived multiple foxhop pool restarts. Live factory queue visible to operator as a persistent ~8-bin floor that never drained.
507 lines
23 KiB
Bash
Executable file
507 lines
23 KiB
Bash
Executable file
#!/usr/bin/env bash
|
|
# bend-emit-pool.sh — bounded parallel emit pool for circuit-emission cells.
|
|
#
|
|
# Reads work items from $LUMBDA_QUEUE_DIR/*.lsp. For each item <tag>:
|
|
# - skips if <tag>.ready or <tag>.done exists
|
|
# - skips if <tag>.emitting marker exists (in-flight)
|
|
# - throttles to MAX concurrent emits
|
|
# - launches $LUMBDA_EMITTER_CMD <tag>.lsp, expects it to write <tag>.bin
|
|
# - touches <tag>.ready when bin size > MIN_BIN_SIZE
|
|
# - touches <tag>.done if emit fails or bin stays too small
|
|
#
|
|
# Exits when no .lsp without a marker remains AND no .emitting markers active.
|
|
# Touch $LUMBDA_QUEUE_DIR/STOP to halt new launches gracefully.
|
|
#
|
|
# Liveness:
|
|
# - Writes $LUMBDA_QUEUE_DIR/pool.heartbeat (pid + timestamp) at top of every
|
|
# loop iter via lib-heartbeat.sh::heartbeat_touch.
|
|
# - On every exit path (STOP, drain timeout, signal trap, EXIT trap)
|
|
# writes $LUMBDA_QUEUE_DIR/pool.exit-cause with reason + removes heartbeat.
|
|
# - Supervisor reads pool.heartbeat to detect silent death.
|
|
#
|
|
# Usage: bend-emit-pool.sh [MAX [QUEUE_DIR [MIN_BIN_SIZE_BYTES]]]
|
|
# MAX default 6
|
|
# QUEUE_DIR default $LUMBDA_QUEUE_DIR
|
|
# MIN_BIN_SIZE_BYTES default 268435456 (256 MB — sanity check only).
|
|
#
|
|
# Self-identifies as "bend-emit-pool" in cmdline for precise pkill targeting.
|
|
#
|
|
# Env vars consumed: see $LUMBDA_FACTORY_DIR/CONTRACT.md (LUMBDA_QUEUE_DIR,
|
|
# LUMBDA_DOMAIN_DIR, LUMBDA_REPO_DIR, LUMBDA_EMITTER_CMD, AUTO_PULL_MASTER,
|
|
# AUTO_PULL_TIMEOUT, DISK_MIN_FREE_GB, MEM_MIN_FREE_GB, POOL_MAX_IDLE_LOOPS,
|
|
# SLOW_MAX).
|
|
set -u
|
|
|
|
MAX="${1:-6}"
|
|
QUEUE_DIR="${2:-${LUMBDA_QUEUE_DIR:-/tmp/lumbda-queue}}"
|
|
MIN_BIN_SIZE="${3:-268435456}"
|
|
|
|
LUMBDA_DOMAIN_DIR="${LUMBDA_DOMAIN_DIR:-}"
|
|
LUMBDA_REPO_DIR="${LUMBDA_REPO_DIR:-$HOME/git/lumbda}"
|
|
LUMBDA_FACTORY_DIR="${LUMBDA_FACTORY_DIR:-$(dirname "$(readlink -f "$0")")}"
|
|
LUMBDA_EMITTER_CMD="${LUMBDA_EMITTER_CMD:?LUMBDA_EMITTER_CMD must be set by consumer (e.g. \"lumbda --fast \$LUMBDA_DOMAIN_DIR/lumbda/emit-stream.lsp\")}"
|
|
LIB_HEARTBEAT="${LIB_HEARTBEAT:-$LUMBDA_FACTORY_DIR/lib-heartbeat.sh}"
|
|
LIB_TIER="${LIB_TIER:-$LUMBDA_FACTORY_DIR/lib-tier.sh}"
|
|
|
|
mkdir -p "$QUEUE_DIR"
|
|
|
|
echo $$ > "$QUEUE_DIR/pool.pid"
|
|
trap 'rm -f "$QUEUE_DIR/pool.pid"' EXIT
|
|
|
|
# Self-tag in cmdline so pkill -f bend-emit-pool only matches THIS script,
|
|
# not other bash sessions that happen to mention factory paths.
|
|
exec -a bend-emit-pool bash -c '
|
|
QUEUE_DIR="'"$QUEUE_DIR"'"
|
|
MAX="'"$MAX"'"
|
|
MIN_BIN_SIZE="'"$MIN_BIN_SIZE"'"
|
|
LUMBDA_DOMAIN_DIR="'"$LUMBDA_DOMAIN_DIR"'"
|
|
LUMBDA_REPO_DIR="'"$LUMBDA_REPO_DIR"'"
|
|
LUMBDA_EMITTER_CMD="'"$LUMBDA_EMITTER_CMD"'"
|
|
LIB_HEARTBEAT="'"$LIB_HEARTBEAT"'"
|
|
LIB_TIER="'"$LIB_TIER"'"
|
|
|
|
# Source liveness library — provides heartbeat_touch / heartbeat_is_alive /
|
|
# heartbeat_exit_cause. Path resolved by parent shell before exec so inner
|
|
# shell does not need to derive its own dirname.
|
|
# shellcheck source=/dev/null
|
|
. "$LIB_HEARTBEAT"
|
|
# Source bin-size tier classifier — provides tier_of_size / tier_label_dlq.
|
|
# Pool uses these to route oversize bins (above TIER_LARGE_MAX) to DLQ at
|
|
# finalize time instead of letting them flow downstream to crush the
|
|
# dispatcher fleet.
|
|
# shellcheck source=/dev/null
|
|
. "$LIB_TIER"
|
|
|
|
echo "[$(date +%H:%M:%S)] bend-emit-pool start MAX=$MAX queue=$QUEUE_DIR"
|
|
|
|
# Exit-cause writers. Every exit path (drain, STOP, signal) records its
|
|
# reason via the library before falling off the loop. EXIT trap acts as
|
|
# safety net for unforeseen crashes; the loop body sets a more specific
|
|
# reason just before break + EXIT trap then forwards it.
|
|
POOL_EXIT_REASON=""
|
|
on_exit() {
|
|
local r="${POOL_EXIT_REASON:-unknown-exit}"
|
|
heartbeat_exit_cause "$QUEUE_DIR" "pool" "$r"
|
|
echo "[$(date +%H:%M:%S)] bend-emit-pool exit reason=$r"
|
|
}
|
|
trap on_exit EXIT
|
|
trap "POOL_EXIT_REASON=signal-int; exit 130" INT
|
|
trap "POOL_EXIT_REASON=signal-term; exit 143" TERM
|
|
|
|
# Pull origin/master before consuming a cell. Fail-soft: log a warning
|
|
# and return non-zero on failure (caller decides whether to keep emitting
|
|
# against checkout we have). Cheap (~1s fetch over LAN). Locked by
|
|
# .git/index.lock so concurrent pool restarts serialize cleanly.
|
|
pull_master_or_log() {
|
|
[ "${AUTO_PULL_MASTER:-1}" = "1" ] || return 0
|
|
local out rc
|
|
out=$(timeout "${AUTO_PULL_TIMEOUT:-30}" git -C "$LUMBDA_REPO_DIR" pull --ff-only origin master 2>&1)
|
|
rc=$?
|
|
if [ "$rc" -ne 0 ]; then
|
|
echo "[$(date +%H:%M:%S)] PULL-FAIL rc=$rc (substrate may be stale): $(echo "$out" | tr "\n" "|" | cut -c-200)"
|
|
return "$rc"
|
|
fi
|
|
if ! echo "$out" | grep -q "Already up to date"; then
|
|
echo "[$(date +%H:%M:%S)] PULL ok: $(echo "$out" | tr "\n" "|" | cut -c-200)"
|
|
fi
|
|
return 0
|
|
}
|
|
|
|
# Self-healing janitor: clear .emitting markers whose emitter PID is dead.
|
|
# Called before every slot-count check. Stale markers can accrue when emits
|
|
# get SIGKILLed (OOM, manual cleanup) before they can clean up their own
|
|
# marker.
|
|
#
|
|
# Subshell wrapper around emitter touches .ready (or .done) BEFORE removing
|
|
# its own .emitting marker. A brief window exists where emitter has exited
|
|
# (PID dead) but parent subshell has not finished tail cleanup. If this
|
|
# janitor fires in that window, it must NOT treat emit as failed —
|
|
# .ready/.done file proves completion. Likewise if a .bin already exists
|
|
# at full size, emit was real and .ready marker may already have been
|
|
# consumed by dispatcher.
|
|
#
|
|
# NOTE: comments inside bash -c body must NOT contain apostrophes — each
|
|
# closes outer single-quoted argument early + corrupts parse.
|
|
heal_stale_markers() {
|
|
for em in "$QUEUE_DIR"/*.emitting; do
|
|
[ ! -f "$em" ] && continue
|
|
pid=$(cat "$em" 2>/dev/null)
|
|
if [ -z "$pid" ] || ! kill -0 "$pid" 2>/dev/null; then
|
|
tag=$(basename "$em" .emitting)
|
|
# If ANY completion artifact already exists, the emit completed
|
|
# before this janitor fired. Silently clear stale .emitting marker
|
|
# — no HEAL log, no .done touch, no relaunch.
|
|
if [ -f "$QUEUE_DIR/${tag}.ready" ] \
|
|
|| [ -f "$QUEUE_DIR/${tag}.done" ]; then
|
|
rm -f "$em"
|
|
continue
|
|
fi
|
|
# Orphan-bin recovery: if .bin exists at full size but no .ready/.done,
|
|
# subshell wrote .bin then died (SIGTERM mid-cleanup) before touching
|
|
# .ready. Promote: bin is real, dispatcher should consume it. Truncated
|
|
# bins (< MIN_BIN_SIZE) treated as failure — rm + .done so cell does
|
|
# not re-emit endlessly.
|
|
if [ -f "$QUEUE_DIR/${tag}.bin" ]; then
|
|
sz=$(stat -c %s "$QUEUE_DIR/${tag}.bin" 2>/dev/null || echo 0)
|
|
magic=$(head -c 8 "$QUEUE_DIR/${tag}.bin" 2>/dev/null)
|
|
if [ "$magic" != "QECCOPS1" ]; then
|
|
# Header-rewrite never fired. See finalize() comment for the
|
|
# SIGKILL/OOM mechanism.
|
|
mkdir -p "$QUEUE_DIR/dlq"
|
|
echo 1 > "$QUEUE_DIR/dlq/${tag}.retries"
|
|
{
|
|
echo "tier=zero-magic bin_bytes=$sz"
|
|
echo "reason: emit-interrupted (heal-stale path) — bin header magic missing"
|
|
echo "first-8-bytes-hex: $(head -c 8 $QUEUE_DIR/${tag}.bin 2>/dev/null | xxd -p -c 8)"
|
|
} > "$QUEUE_DIR/dlq/${tag}.reason"
|
|
echo "[$(date +%H:%M:%S)] DLQ orphan-bin-zero-magic $tag sz=$sz"
|
|
rm -f "$QUEUE_DIR/${tag}.bin"
|
|
touch "$QUEUE_DIR/${tag}.done.fail"
|
|
touch "$QUEUE_DIR/${tag}.done"
|
|
elif [ "$sz" -ge "$MIN_BIN_SIZE" ]; then
|
|
touch "$QUEUE_DIR/${tag}.ready"
|
|
echo "[$(date +%H:%M:%S)] RECOVER orphan-bin $tag ($(du -h $QUEUE_DIR/${tag}.bin | cut -f1))"
|
|
else
|
|
rm -f "$QUEUE_DIR/${tag}.bin"
|
|
touch "$QUEUE_DIR/${tag}.done"
|
|
echo "[$(date +%H:%M:%S)] HEAL orphan-bin-truncated $tag sz=$sz"
|
|
fi
|
|
rm -f "$em"
|
|
continue
|
|
fi
|
|
echo "[$(date +%H:%M:%S)] HEAL stale $tag (pid=$pid dead)"
|
|
rm -f "$em"
|
|
touch "$QUEUE_DIR/${tag}.done"
|
|
fi
|
|
done
|
|
}
|
|
|
|
# Belt-and-suspenders: standalone scan for orphan .bin files whose .emitting
|
|
# marker is already gone (different death path than heal_stale_markers).
|
|
# Catches:
|
|
# - pool subshell died AFTER .emitting was racy-cleaned but
|
|
# BEFORE it touched .ready
|
|
# - pool process murdered between unrelated janitor pass + .ready touch
|
|
# - any .bin left behind by a manual operator intervention
|
|
#
|
|
# Same recovery rule as .emitting branch: full-size → .ready, truncated →
|
|
# rm + .done so cell does not re-emit forever. Skips bins already in flight
|
|
# (.inflight-N) or already complete (.ready / .done / .dispatch.log).
|
|
# Idempotent — safe to run every loop iter alongside heal_stale_markers.
|
|
heal_orphan_bins() {
|
|
shopt -s nullglob
|
|
for bin in "$QUEUE_DIR"/*.bin; do
|
|
tag=$(basename "$bin" .bin)
|
|
# Any in-progress or completion artifact = NOT orphan
|
|
[ -f "$QUEUE_DIR/${tag}.ready" ] && continue
|
|
[ -f "$QUEUE_DIR/${tag}.done" ] && continue
|
|
[ -f "$QUEUE_DIR/${tag}.emitting" ] && continue
|
|
[ -f "$QUEUE_DIR/${tag}.dispatch.log" ] && continue
|
|
# 2026-06-14 defect fix: with shopt -s nullglob (set at function top
|
|
# so the outer for loop tolerates empty *.bin), an unmatched
|
|
# .inflight-* glob expands to NOTHING. Bare "ls >/dev/null" then
|
|
# succeeds (lists CWD) + the if-branch incorrectly skips the orphan.
|
|
# Compgen returns success only when pattern matches, immune to nullglob.
|
|
if compgen -G "$QUEUE_DIR/${tag}.inflight-*" >/dev/null; then
|
|
continue
|
|
fi
|
|
sz=$(stat -c %s "$bin" 2>/dev/null || echo 0)
|
|
magic=$(head -c 8 "$bin" 2>/dev/null)
|
|
if [ "$magic" != "QECCOPS1" ]; then
|
|
# Same emit-interrupted class as finalize() + heal_stale_markers.
|
|
mkdir -p "$QUEUE_DIR/dlq"
|
|
echo 1 > "$QUEUE_DIR/dlq/${tag}.retries"
|
|
{
|
|
echo "tier=zero-magic bin_bytes=$sz"
|
|
echo "reason: emit-interrupted (orphan-bin-scan path) — bin header magic missing"
|
|
echo "first-8-bytes-hex: $(head -c 8 $bin 2>/dev/null | xxd -p -c 8)"
|
|
} > "$QUEUE_DIR/dlq/${tag}.reason"
|
|
echo "[$(date +%H:%M:%S)] DLQ orphan-bin-scan-zero-magic $tag sz=$sz"
|
|
rm -f "$bin"
|
|
touch "$QUEUE_DIR/${tag}.done.fail"
|
|
touch "$QUEUE_DIR/${tag}.done"
|
|
elif [ "$sz" -ge "$MIN_BIN_SIZE" ]; then
|
|
touch "$QUEUE_DIR/${tag}.ready"
|
|
echo "[$(date +%H:%M:%S)] RECOVER orphan-bin-scan $tag ($(du -h $bin | cut -f1))"
|
|
else
|
|
rm -f "$bin"
|
|
touch "$QUEUE_DIR/${tag}.done"
|
|
echo "[$(date +%H:%M:%S)] HEAL orphan-bin-scan-truncated $tag sz=$sz"
|
|
fi
|
|
done
|
|
shopt -u nullglob
|
|
}
|
|
|
|
DISK_MIN_FREE_GB="${DISK_MIN_FREE_GB:-30}"
|
|
# Memory-free guard. Each emit RSS sits ~3-4 GiB at peak. With MAX=6 emits
|
|
# in flight + backend + dispatchers, total memory pressure can climb fast.
|
|
# Hold launches when free RAM falls below MEM_MIN_FREE_GB (default 8 GiB).
|
|
# Mirrors disk-free pattern: pool HOLDs rather than spawning an
|
|
# overcommit-OOM-kill cascade.
|
|
MEM_MIN_FREE_GB="${MEM_MIN_FREE_GB:-8}"
|
|
# Idle tolerance — pool stays alive while feeder may push more .lsp cells.
|
|
# Default 600 (x 5s = 50 min) — long enough to outlast feeder push cadence
|
|
# in normal operation. Exit only on STOP marker OR sustained idle past this
|
|
# bound.
|
|
POOL_MAX_IDLE_LOOPS="${POOL_MAX_IDLE_LOOPS:-600}"
|
|
# Substrate freshness — pull origin/master before emitting each cell so a
|
|
# backend never consumes a .lsp against stale substrate. Fail-soft: if pull
|
|
# fails (local mods, network), log + continue with checkout we have. Cell
|
|
# still emits — log entry tells operator "this host ran against potentially
|
|
# stale substrate".
|
|
AUTO_PULL_MASTER="${AUTO_PULL_MASTER:-1}"
|
|
AUTO_PULL_TIMEOUT="${AUTO_PULL_TIMEOUT:-30}"
|
|
|
|
# Startup orphan sweep — runs BEFORE main loop so any leftover bins from a
|
|
# SIGKILL-d previous pool (whose subshell wrappers died before touching
|
|
# .ready) get promoted immediately. In-loop heal also runs, but only after
|
|
# a full poll cycle (~5s + admission checks), during which the new pool
|
|
# could start spawning fresh emits + pushing past orphans. Catching them
|
|
# at startup closes that window.
|
|
echo "[$(date +%H:%M:%S)] startup orphan scan"
|
|
heal_stale_markers
|
|
heal_orphan_bins
|
|
|
|
idle_loops=0
|
|
while true; do
|
|
# Heartbeat UNCONDITIONALLY at top of loop. Records both "I am alive"
|
|
# mtime + "my pid is X" content. Supervisor on this host reads it every
|
|
# poll cycle to decide whether to restart us.
|
|
heartbeat_touch "$QUEUE_DIR" "pool"
|
|
|
|
if [ -f "$QUEUE_DIR/STOP" ]; then
|
|
echo "[$(date +%H:%M:%S)] STOP seen"
|
|
POOL_EXIT_REASON="stop-marker"
|
|
break
|
|
fi
|
|
pull_master_or_log || true
|
|
heal_stale_markers
|
|
heal_orphan_bins
|
|
# Disk-free guard. Emits land 4-6 GiB bins; running near zero free disk
|
|
# makes emitter spin on write retries instead of erroring out. Hold
|
|
# launches until at least DISK_MIN_FREE_GB available.
|
|
free_gb=$(df -BG --output=avail / 2>/dev/null | tail -1 | tr -dc 0-9)
|
|
if [ "${free_gb:-0}" -lt "$DISK_MIN_FREE_GB" ]; then
|
|
echo "[$(date +%H:%M:%S)] HOLD disk-free=${free_gb}G < ${DISK_MIN_FREE_GB}G, sleeping"
|
|
sleep 30
|
|
continue
|
|
fi
|
|
# Memory-free guard. MAX=6 concurrent emits can each climb to ~3-4 GiB
|
|
# RSS; with backend + dispatchers also resident, sustained pressure
|
|
# approaches host limits. Hold new launches until MemAvailable >=
|
|
# MEM_MIN_FREE_GB so an OOM killer never picks a long-running emit
|
|
# at the worst moment. Same backoff sleep as disk guard.
|
|
free_mem_kb=$(grep ^MemAvailable: /proc/meminfo 2>/dev/null | tr -dc 0-9)
|
|
free_mem_gb=$(( ${free_mem_kb:-0} / 1048576 ))
|
|
if [ "$free_mem_gb" -lt "$MEM_MIN_FREE_GB" ]; then
|
|
echo "[$(date +%H:%M:%S)] HOLD mem-free=${free_mem_gb}G < ${MEM_MIN_FREE_GB}G, sleeping"
|
|
sleep 30
|
|
continue
|
|
fi
|
|
# Read autoscaler advisory + apply dynamic MAX cap when RAM/swap pressure
|
|
# detected. If advisory says slow_emit=1, cap concurrency at SLOW_MAX
|
|
# (default = MAX/2). When pressure clears the advisory flips back to 0
|
|
# and the cap restores. MEM_MIN_FREE_GB hold (above) is still the hard
|
|
# floor — advisory acts as a graceful intermediate signal.
|
|
effective_max="$MAX"
|
|
if [ -f "$QUEUE_DIR/emit.advisory" ]; then
|
|
slow=$(grep -E '^slow_emit=' "$QUEUE_DIR/emit.advisory" 2>/dev/null | tail -1 | cut -d= -f2)
|
|
if [ "${slow:-0}" = "1" ]; then
|
|
effective_max="${SLOW_MAX:-$(( MAX / 2 ))}"
|
|
[ "$effective_max" -lt 1 ] && effective_max=1
|
|
fi
|
|
fi
|
|
# Find next pending work item
|
|
next=""
|
|
for f in "$QUEUE_DIR"/*.lsp; do
|
|
[ ! -f "$f" ] && continue
|
|
tag=$(basename "$f" .lsp)
|
|
[ -f "$QUEUE_DIR/${tag}.ready" ] && continue
|
|
[ -f "$QUEUE_DIR/${tag}.done" ] && continue
|
|
# .done.fail = operator/caller marked cell as skip (incompatible config,
|
|
# poison cell, retracted batch). Without this check pool re-emits the
|
|
# cell every iter even after the operator explicitly excluded it.
|
|
[ -f "$QUEUE_DIR/${tag}.done.fail" ] && continue
|
|
[ -f "$QUEUE_DIR/${tag}.emitting" ] && continue
|
|
# Dispatcher worker claims .ready by renaming to .inflight-N (atomic).
|
|
# If worker dies mid-dispatch (backend crash, kill, OOM) .inflight-N
|
|
# stays + no .done written. Treat .inflight-* as downstream-owned and
|
|
# do not re-emit. Dispatcher reclaim path runs at bend-dispatcher
|
|
# startup + heals stale .inflight-N back to .ready or .done.
|
|
# NOTE: comments here must NOT contain apostrophes — each closes outer
|
|
# single-quoted bash -c argument early + corrupts parse.
|
|
if ls "$QUEUE_DIR/${tag}.inflight-"* >/dev/null 2>&1; then
|
|
continue
|
|
fi
|
|
# Belt-and-suspenders against HEAL race: if a .bin exists at any size,
|
|
# an emit has run. Dispatcher consumes .bin + may delete .ready when
|
|
# done, leaving no other markers — without this check main loop would
|
|
# relaunch the cell after dispatcher consumed it.
|
|
[ -f "$QUEUE_DIR/${tag}.bin" ] && continue
|
|
next="$f"
|
|
break
|
|
done
|
|
if [ -z "$next" ]; then
|
|
# No new items — wait for in-flight or for feeder to push more.
|
|
running=$(ls "$QUEUE_DIR"/*.emitting 2>/dev/null | wc -l)
|
|
if [ "$running" = "0" ]; then
|
|
idle_loops=$((idle_loops + 1))
|
|
if [ "$idle_loops" -ge "$POOL_MAX_IDLE_LOOPS" ]; then
|
|
echo "[$(date +%H:%M:%S)] pool drained + idle past $POOL_MAX_IDLE_LOOPS loops"
|
|
POOL_EXIT_REASON="drain-timeout"
|
|
break
|
|
fi
|
|
fi
|
|
sleep 5
|
|
continue
|
|
fi
|
|
idle_loops=0
|
|
# Throttle to MAX concurrent (self-healing each check). Heartbeat again
|
|
# inside throttle wait — long emits can hold us here for minutes; without
|
|
# re-touching heartbeat supervisor would see us as dead.
|
|
while true; do
|
|
heartbeat_touch "$QUEUE_DIR" "pool"
|
|
heal_stale_markers
|
|
heal_orphan_bins
|
|
running=$(ls "$QUEUE_DIR"/*.emitting 2>/dev/null | wc -l)
|
|
# Compare against effective_max (advisory-aware) rather than raw MAX
|
|
# so RAM-pressure slow-down actually reduces concurrency. effective_max
|
|
# computed once per main-loop iter against latest advisory.
|
|
[ "$running" -lt "$effective_max" ] && break
|
|
sleep 3
|
|
done
|
|
tag=$(basename "$next" .lsp)
|
|
echo "[$(date +%H:%M:%S)] launch $tag"
|
|
# Subshell: traps EXIT/HUP/TERM so partial emit still finalizes (.ready
|
|
# or .done) when parent pool dies during hot-reload. SIGHUP explicitly
|
|
# ignored — propagated from a dying parent must NOT kill the wrapper
|
|
# mid-cleanup. SIGKILL on us still leaks (no trap fires), but startup
|
|
# heal_orphan_bins promotes those bins on next pool start.
|
|
(
|
|
trap "" HUP
|
|
bin="$QUEUE_DIR/${tag}.bin"
|
|
finalize() {
|
|
if [ -f "$bin" ]; then
|
|
sz=$(stat -c %s "$bin" 2>/dev/null || echo 0)
|
|
# Bin-magic check FIRST. Interrupted emits (OOM-killed mid-write,
|
|
# SIGKILL on emitter hang) leave the 16-byte header reserve as
|
|
# zero bytes — emit-stream writes 16 zeros, streams ops, then
|
|
# seeks back to byte 0 to rewrite QECCOPS1 + n_ops at the end.
|
|
# SIGKILL between those steps strands a file with zero magic.
|
|
# Downstream backend chokes with opaque "bad magic" → bin-load-fail
|
|
# → DLQ with no useful forensic info. Route here with actual cause
|
|
# named so operators can trace back to upstream cell hang instead
|
|
# of guessing at load_ops_bin internals.
|
|
magic=$(head -c 8 "$bin" 2>/dev/null)
|
|
if [ "$magic" != "QECCOPS1" ]; then
|
|
mkdir -p "$QUEUE_DIR/dlq"
|
|
echo 1 > "$QUEUE_DIR/dlq/${tag}.retries"
|
|
{
|
|
echo "tier=zero-magic bin_bytes=$sz"
|
|
echo "reason: emit-interrupted — bin header magic missing or non-QECCOPS1 (likely SIGKILL or OOM mid-emit; emit-stream header rewrite never fired)"
|
|
echo "first-8-bytes-hex: $(head -c 8 "$bin" 2>/dev/null | xxd -p -c 8)"
|
|
echo "--- last 10 lines of emit.log ---"
|
|
tail -10 "$QUEUE_DIR/${tag}.emit.log" 2>/dev/null
|
|
} > "$QUEUE_DIR/dlq/${tag}.reason"
|
|
echo "[$(date +%H:%M:%S)] DLQ $tag zero-magic sz=$sz ($(printf %.0f $((sz/1024/1024)))MiB) — emit-interrupted"
|
|
rm -f "$bin"
|
|
touch "$QUEUE_DIR/${tag}.done.fail"
|
|
touch "$QUEUE_DIR/${tag}.done"
|
|
rm -f "$QUEUE_DIR/${tag}.emitting"
|
|
return 0
|
|
fi
|
|
# 2nd sub-class: GOOD magic + n_ops=0 — emit wrote QECCOPS1 magic +
|
|
# streamed ops but SIGKILL hit between stream-end + finalize seek-
|
|
# back-to-byte-8 that rewrites n_ops. Detect via n_ops=0
|
|
# implausibility check: any in-tier bin should have n_ops > 0;
|
|
# if not, finalize never fired.
|
|
n_ops_le=$(dd if="$bin" bs=1 count=8 skip=8 2>/dev/null | xxd -p -c 8)
|
|
if [ "$n_ops_le" = "0000000000000000" ]; then
|
|
mkdir -p "$QUEUE_DIR/dlq"
|
|
echo 1 > "$QUEUE_DIR/dlq/${tag}.retries"
|
|
{
|
|
echo "tier=zero-n_ops bin_bytes=$sz"
|
|
echo "reason: emit-finalize-never-fired — QECCOPS1 magic present but n_ops field at byte 8 is zero (SIGKILL between stream-end + emit-stream seek-back; downstream load_ops_bin would fail length-mismatch)"
|
|
echo "n_ops_le_hex: $n_ops_le"
|
|
echo "--- last 10 lines of emit.log ---"
|
|
tail -10 "$QUEUE_DIR/${tag}.emit.log" 2>/dev/null
|
|
} > "$QUEUE_DIR/dlq/${tag}.reason"
|
|
echo "[$(date +%H:%M:%S)] DLQ $tag zero-n_ops sz=$sz ($(printf %.0f $((sz/1024/1024)))MiB) — emit-finalize-never-fired"
|
|
rm -f "$bin"
|
|
touch "$QUEUE_DIR/${tag}.done.fail"
|
|
touch "$QUEUE_DIR/${tag}.done"
|
|
rm -f "$QUEUE_DIR/${tag}.emitting"
|
|
return 0
|
|
fi
|
|
# Tier-classify the bin. Tier "below-min" / "above-max" → DLQ with
|
|
# a typed reason. In-band tiers (micro/small/medium/large) get
|
|
# promoted to .ready + dispatcher routes through tier-specific
|
|
# admission. lib-tier.sh owns boundary policy.
|
|
tier=$(tier_of_size "$sz")
|
|
dlq_reason=$(tier_label_dlq "$tier")
|
|
if [ -n "$dlq_reason" ]; then
|
|
mkdir -p "$QUEUE_DIR/dlq"
|
|
echo 1 > "$QUEUE_DIR/dlq/${tag}.retries"
|
|
{
|
|
echo "tier=$tier bin_bytes=$sz"
|
|
echo "reason: $dlq_reason"
|
|
echo "--- last 10 lines of emit.log ---"
|
|
tail -10 "$QUEUE_DIR/${tag}.emit.log" 2>/dev/null
|
|
} > "$QUEUE_DIR/dlq/${tag}.reason"
|
|
echo "[$(date +%H:%M:%S)] DLQ $tag tier=$tier sz=$sz ($(printf %.0f $((sz/1024/1024)))MiB) — $dlq_reason"
|
|
rm -f "$bin"
|
|
touch "$QUEUE_DIR/${tag}.done.fail"
|
|
touch "$QUEUE_DIR/${tag}.done"
|
|
else
|
|
# In-tier bin → optional domain-specific sidecar (vorticity, etc),
|
|
# then mark .ready. Sidecar lands next to .bin. Read-only
|
|
# post-process; emit substrate untouched. Fail-soft: sidecar
|
|
# script errors log + continue — bin still promotes to .ready.
|
|
# Dispatcher reads .bin, ignores sidecar; operators + analysis
|
|
# tooling consume sidecar opportunistically.
|
|
vort="$QUEUE_DIR/${tag}.vorticity.tsv"
|
|
if [ -n "$LUMBDA_DOMAIN_DIR" ] && [ -x "$LUMBDA_DOMAIN_DIR/scripts/qubit-vorticity.py" ]; then
|
|
if "$LUMBDA_DOMAIN_DIR/scripts/qubit-vorticity.py" "$bin" --out "$vort" \
|
|
> "$QUEUE_DIR/${tag}.vorticity.log" 2>&1; then
|
|
:
|
|
else
|
|
echo "[$(date +%H:%M:%S)] vorticity $tag FAILED (see ${tag}.vorticity.log) — promoting bin anyway"
|
|
fi
|
|
fi
|
|
touch "$QUEUE_DIR/${tag}.ready"
|
|
echo "[$(date +%H:%M:%S)] ready $tag tier=$tier ($(du -h $bin | cut -f1))"
|
|
fi
|
|
else
|
|
# No bin — emitter errored before writing (undefined flag, syntax
|
|
# error, load failure, etc). Emit failures stay DETERMINISTIC —
|
|
# retry will fail the same way. Route to DLQ for forensics, mark
|
|
# .done.fail + .done so pool skips on next iter (no infinite
|
|
# relaunch loop).
|
|
mkdir -p "$QUEUE_DIR/dlq"
|
|
echo 1 > "$QUEUE_DIR/dlq/${tag}.retries"
|
|
tail -10 "$QUEUE_DIR/${tag}.emit.log" 2>/dev/null > "$QUEUE_DIR/dlq/${tag}.reason"
|
|
echo "[$(date +%H:%M:%S)] DLQ $tag (emit error — see dlq/${tag}.reason)"
|
|
touch "$QUEUE_DIR/${tag}.done.fail"
|
|
touch "$QUEUE_DIR/${tag}.done"
|
|
fi
|
|
rm -f "$QUEUE_DIR/${tag}.emitting"
|
|
}
|
|
trap finalize EXIT
|
|
trap "kill \$lpid 2>/dev/null; finalize; exit 130" INT
|
|
trap "kill \$lpid 2>/dev/null; finalize; exit 143" TERM
|
|
if [ -n "$LUMBDA_DOMAIN_DIR" ]; then
|
|
cd "$LUMBDA_DOMAIN_DIR"
|
|
fi
|
|
PYTHONUNBUFFERED=1 ECDSA_OUT_BIN="$QUEUE_DIR/${tag}.bin" \
|
|
$LUMBDA_EMITTER_CMD "$next" \
|
|
> "$QUEUE_DIR/${tag}.emit.log" 2>&1 &
|
|
lpid=$!
|
|
echo "$lpid" > "$QUEUE_DIR/${tag}.emitting"
|
|
wait "$lpid"
|
|
) &
|
|
done
|
|
wait
|
|
echo "[$(date +%H:%M:%S)] all emits done"
|
|
# Loop exited cleanly via STOP / drain; POOL_EXIT_REASON already set.
|
|
# EXIT trap calls heartbeat_exit_cause.
|
|
'
|