diff --git a/CHANGELOG.md b/CHANGELOG.md index 6c34ef4..e05a297 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,21 @@ All notable changes to EigenScript are documented here. ## [Unreleased] ### Added +- **`lib/supervise` — observer-native supervision (#409).** A supervisor over + the #408 cooperative task layer that, beyond BEAM-style crash-restart, also + catches the silently **wedged** worker — alive, not crashed, never timing + out, but no longer making progress — a signal BEAM structurally cannot see. + Workers publish raw progress with `heartbeat of [slot, value]`; the + supervisor samples each alive worker every tick, feeds a **bounded** signature + (`1.5 + v % 8`, dodging the #294 flat-entropy blind spot) into that slot's + observer trajectory, and restarts a worker whose signature has gone `stable` + after a warm-up window (the EigenOS red-title pattern, generalized) or that + died with an uncaught raise — each up to a per-worker restart-intensity cap. + `supervisor_new`, `supervise of [sup, fn, slot]`, `supervise_step`, and + `supervise_run of [sup, max_ticks]`. Deterministic by construction; flagship + `examples/supervisor_tree.eigs`. Surfaced a gap: a task cannot learn its own + id (no `task_self`), so the message-link supervision variant is not yet + expressible — filed separately. - **Lint `W018` — `e.kind` compared against an out-of-set error kind (#469).** A `catch` handler comparing a caught error's `.kind` against a string that is a near-miss of a real kind — a case variant (`"IO"`), a single-character typo diff --git a/docs/STDLIB.md b/docs/STDLIB.md index 4d1425a..9ab99b3 100644 --- a/docs/STDLIB.md +++ b/docs/STDLIB.md @@ -752,6 +752,19 @@ and the claim). `lock_new`, `lock_acquire`, `lock_release`, and if the body raises). Acquire only from within tasks — it spins with `task_yield`, so it needs the scheduler active. +### lib/supervise.eigs +Observer-native supervision over the #408 task layer. Beyond BEAM-style +crash-restart, it also detects the silently **wedged** worker — alive, not +crashed, never timing out, but no longer making progress — by sampling each +worker's progress trajectory and seeing it go `stable` (the EigenOS +red-title pattern generalized). Workers publish raw progress with +`heartbeat of [slot, value]`; the supervisor turns it into a bounded +signature (dodging the #294 flat-entropy blind spot). `supervisor_new`, +`supervise of [sup, fn, slot]`, `supervise_step`, and `supervise_run of +[sup, max_ticks]` (drive to settled); wedged and crashed workers are +restarted from their spec under a per-worker restart-intensity cap. Eight +slots. See `examples/supervisor_tree.eigs`. + ### lib/store.eigs EigenStore high-level layer: `find`, `find_one`, `upsert`, `bulk_put`, `to_dataframe`. diff --git a/examples/supervisor_tree.eigs b/examples/supervisor_tree.eigs new file mode 100644 index 0000000..79c5b0d --- /dev/null +++ b/examples/supervisor_tree.eigs @@ -0,0 +1,46 @@ +# ============================================================ +# supervisor_tree.eigs — observer-native supervision (#409) +# ============================================================ +# +# A worker pool of three tasks. Two are healthy; one silently WEDGES partway +# through — it never crashes and never times out, so a crash-reactive +# supervisor (BEAM-style) would watch it hang forever. Here the supervisor +# samples each worker's progress trajectory and sees the frozen one go +# `stable`, then restarts it from its spec. The restarted worker runs to +# completion. Deterministic: this prints identically on every run. + +load_file of "lib/supervise.eigs" + +# A worker advances a counter, publishing progress each step. The one in the +# "wedge slot" freezes on its FIRST life (infinite yield, progress stuck), and +# on its restarted life runs cleanly to the end. +wedge_slot is 1 +lives is [0, 0, 0] + +define worker(slot) as: + lives is set_at of [lives, slot, lives[slot] + 1] + life is lives[slot] + n is 0 + loop while n < 12: + heartbeat of [slot, n] + # First life of the wedge-slot worker gets stuck at step 4. + if slot == wedge_slot and life == 1 and n == 4: + loop while 1 == 1: + heartbeat of [slot, 4] # frozen — alive but no progress + task_yield of null + task_yield of null + n is n + 1 + return f"worker {slot} finished" + +sup is supervisor_new of 3 +supervise of [sup, worker, 0] +supervise of [sup, worker, 1] +supervise of [sup, worker, 2] + +restarts is supervise_run of [sup, 1000] + +print of f"restarts performed: {restarts}" +print of f"worker 0 completed: {_sup_done[0]}" +print of f"worker 1 completed: {_sup_done[1]}" +print of f"worker 2 completed: {_sup_done[2]}" +print of "supervision tree settled" diff --git a/lib/supervise.eigs b/lib/supervise.eigs new file mode 100644 index 0000000..2856bc7 --- /dev/null +++ b/lib/supervise.eigs @@ -0,0 +1,227 @@ +# ============================================================ +# Supervise Library — observer-native supervision (#408 task layer) +# ============================================================ +# +# BEAM-style supervision restarts a worker when it CRASHES. An observer +# supervisor also catches the worker that is silently WEDGED — alive, not +# crashed, never timing out, but no longer making progress — a signal BEAM +# structurally cannot see, because its processes carry no free per-value +# trajectory measurement. This is the EigenOS window manager's red-title +# pattern (a hung window detected via `stable of`), generalized over the #408 +# cooperative task layer. +# +# The model (share-nothing messages are unnecessary here — cooperative tasks +# share the module-global heap on one thread): +# * A worker publishes RAW progress each step with `heartbeat of [slot, v]`. +# * The SUPERVISOR samples each alive worker's progress every tick, feeds a +# BOUNDED signature (1.5 + v % 8 — never a raw counter, which hits the +# #294 flat-entropy blind spot) into that slot's observer trajectory, and +# flags a worker whose signature has gone `stable` (frozen) after a +# warm-up window. Sampling on the SUPERVISOR side is what makes a wedge +# visible: a frozen worker stops calling heartbeat, so only the poller +# keeps adding the (now constant) samples that read `stable`. +# * A worker that returns normally is `done`; one that raises is `crashed` +# (caught by the wrapper, so a supervised crash never fails the process). +# * Wedged and crashed workers are restarted from their spec, up to +# `max_restarts` per worker (restart-intensity cap). +# +# How to use: +# load_file of "lib/supervise.eigs" +# +# define worker(slot) as: +# n is 0 +# loop while n < 100: +# heartbeat of [slot, n] # publish progress +# task_yield of null +# n is n + 1 +# return "ok" +# +# sup is supervisor_new of 3 # up to 3 restarts per worker +# supervise of [sup, worker, 0] # worker in slot 0 +# supervise of [sup, worker, 1] +# restarts is supervise_run of [sup, 500] # drive until done / 500 ticks +# +# Eight slots (the WM/task-table precedent). Slots are the caller's index +# space; one supervised worker per slot. + +# --- Shared per-slot state (indexed by slot 0..7) ------------------------ +_sup_progress is [0, 0, 0, 0, 0, 0, 0, 0] # each worker's latest raw progress +_sup_done is [0, 0, 0, 0, 0, 0, 0, 0] # 1 when the worker returned normally +_sup_crashed is [0, 0, 0, 0, 0, 0, 0, 0] # 1 when the worker raised +_sup_warm is [0, 0, 0, 0, 0, 0, 0, 0] # observer samples since (re)start + +# --- Observer trajectory scalars (one NAMED binding per slot; a list element +# has no trajectory, #262, so the eight signals need eight named scalars) --- +_sup_o0 is 0.0 +_sup_o1 is 0.0 +_sup_o2 is 0.0 +_sup_o3 is 0.0 +_sup_o4 is 0.0 +_sup_o5 is 0.0 +_sup_o6 is 0.0 +_sup_o7 is 0.0 + +# _sup_feed(slot, sig) — write the bounded signature into slot's trajectory. +# `+ 0.0` forces a fresh value so an aliased iterate can't smuggle in an old +# trajectory (#262). +define _sup_feed(pair) as: + slot is pair[0] + sig is pair[1] + if slot == 0: + _sup_o0 is sig + 0.0 + elif slot == 1: + _sup_o1 is sig + 0.0 + elif slot == 2: + _sup_o2 is sig + 0.0 + elif slot == 3: + _sup_o3 is sig + 0.0 + elif slot == 4: + _sup_o4 is sig + 0.0 + elif slot == 5: + _sup_o5 is sig + 0.0 + elif slot == 6: + _sup_o6 is sig + 0.0 + elif slot == 7: + _sup_o7 is sig + 0.0 + else: + throw of "supervise: slot out of range (8 slots)" + return 1 + +# _sup_stable(slot) — 1 if slot's signature trajectory has frozen. +define _sup_stable(slot) as: + if slot == 0: + return stable of _sup_o0 + elif slot == 1: + return stable of _sup_o1 + elif slot == 2: + return stable of _sup_o2 + elif slot == 3: + return stable of _sup_o3 + elif slot == 4: + return stable of _sup_o4 + elif slot == 5: + return stable of _sup_o5 + elif slot == 6: + return stable of _sup_o6 + elif slot == 7: + return stable of _sup_o7 + throw of "supervise: slot out of range (8 slots)" + +# --- Public API ---------------------------------------------------------- + +# supervisor_new(max_restarts) — a fresh supervisor with a per-worker restart +# cap. +define supervisor_new(max_restarts) as: + return {"children": [], "max_restarts": max_restarts} + +# heartbeat([slot, value]) — a worker calls this each step to publish its raw +# progress. Any value that ADVANCES while the worker is healthy works; the +# supervisor turns it into a bounded observer signature. +define heartbeat(pair) as: + slot is pair[0] + value is pair[1] + _sup_progress is set_at of [_sup_progress, slot, value] + return 1 + +# _run_child([fn, slot]) — the wrapper task. Runs the worker, records normal +# completion; catches a raise as `crashed` so a supervised crash never turns +# into an uncaught task death (which would fail the process, #493). +define _run_child(pair) as: + fn is pair[0] + slot is pair[1] + try: + r is fn of slot + _sup_done is set_at of [_sup_done, slot, 1] + return r + catch e: + _sup_crashed is set_at of [_sup_crashed, slot, 1] + return null + +# _reset_slot(slot) — clear a slot's bookkeeping before a (re)start. +define _reset_slot(slot) as: + _sup_progress is set_at of [_sup_progress, slot, 0] + _sup_done is set_at of [_sup_done, slot, 0] + _sup_crashed is set_at of [_sup_crashed, slot, 0] + _sup_warm is set_at of [_sup_warm, slot, 0] + return 1 + +# supervise([sup, fn, slot]) — spawn `fn` as a supervised worker in `slot`. +# `fn` must accept its slot index and publish progress via `heartbeat`. +define supervise(triple) as: + sup is triple[0] + fn is triple[1] + slot is triple[2] + _reset_slot of slot + id is task_spawn of [_run_child, [fn, slot]] + child is {"fn": fn, "slot": slot, "id": id, "restarts": 0} + sup.children is append of [sup.children, child] + return child + +# _restart([sup, child]) — restart a worker if under its cap. Returns 1 if it +# restarted, else 0. +define _restart(pair) as: + sup is pair[0] + child is pair[1] + if child.restarts >= sup.max_restarts: + return 0 + slot is child.slot + if (task_alive of child.id) == 1: + task_kill of child.id + _reset_slot of slot + child.id is task_spawn of [_run_child, [child.fn, slot]] + child.restarts is child.restarts + 1 + return 1 + +# supervise_step(sup) — one supervision tick. Feeds each alive worker's +# progress into its trajectory and restarts wedged (alive + frozen after the +# warm-up window) or crashed (dead, raised) workers under the cap. Returns the +# number of restarts performed this tick. +define supervise_step(sup) as: + restarted is 0 + for child in sup.children: + slot is child.slot + if (task_alive of child.id) == 1: + sig is 1.5 + (_sup_progress[slot] % 8) + _sup_feed of [slot, sig] + _sup_warm is set_at of [_sup_warm, slot, _sup_warm[slot] + 1] + warmed is _sup_warm[slot] >= 10 + if warmed == 1 and (_sup_stable of slot) == 1: + restarted is restarted + (_restart of [sup, child]) + else: + if _sup_crashed[slot] == 1: + restarted is restarted + (_restart of [sup, child]) + return restarted + +# _all_settled(sup) — every worker has finished normally, or exhausted its +# restart cap while still failing (so no further supervision can help). +define _all_settled(sup) as: + for child in sup.children: + slot is child.slot + finished is (task_alive of child.id) == 0 + done is _sup_done[slot] == 1 + capped is child.restarts >= sup.max_restarts + if done == 1: + settled is 1 + elif finished == 1 and capped == 1: + settled is 1 + else: + settled is 0 + if settled == 0: + return 0 + return 1 + +# supervise_run([sup, max_ticks]) — drive supervision from the main task until +# every worker is settled or `max_ticks` elapse. Deterministic: the same +# program replays identically. Returns the total number of restarts performed. +define supervise_run(pair) as: + sup is pair[0] + max_ticks is pair[1] + t is 0 + total is 0 + loop while t < max_ticks: + if (_all_settled of sup) == 1: + return total + total is total + (supervise_step of sup) + task_yield of null + t is t + 1 + return total diff --git a/tests/run_all_tests.sh b/tests/run_all_tests.sh index 3e10361..0ad551d 100755 --- a/tests/run_all_tests.sh +++ b/tests/run_all_tests.sh @@ -2179,6 +2179,12 @@ check_eigs_suite "cooperative tasks: yield/join/deadlock/teardown (#408)" test_t # with_lock releases + re-raises on abort. check_eigs_suite "lib/sync cooperative locks (#488)" test_sync.eigs "All tests passed" 1 +# lib/supervise — observer-native supervision over the #408 task layer (#409): +# a wedged worker (alive, progress frozen) is detected via its observer +# trajectory and restarted; a crashed worker is restarted; a healthy worker is +# left alone; restart-intensity is capped. +check_eigs_suite "lib/supervise observer supervision (#409)" test_supervise.eigs "All tests passed" 1 + # #408 determinism-by-construction: a task program with cooperative yields must # print byte-identically on two fresh processes (the signature property — the # interleaving is a pure function of program order, no tape). diff --git a/tests/test_supervise.eigs b/tests/test_supervise.eigs new file mode 100644 index 0000000..e83ed24 --- /dev/null +++ b/tests/test_supervise.eigs @@ -0,0 +1,73 @@ +# lib/supervise.eigs — observer-native supervision over the #408 task layer: +# a wedged worker (alive, progress frozen, no crash, no timeout) is detected +# via its observer trajectory and restarted; a crashed worker is restarted; +# a healthy worker is left alone; restart-intensity is capped. Deterministic +# by construction (the cooperative task layer). +load_file of "lib/test.eigs" +load_file of "lib/supervise.eigs" + +# --- 1. Healthy worker is never restarted; wedged worker (frozen progress) is +# detected and restarted once, then recovers ------------------------ +define healthy(slot) as: + n is 0 + loop while n < 20: + heartbeat of [slot, n] + task_yield of null + n is n + 1 + return "ok" + +wedge_lives is 0 +define flaky(slot) as: + wedge_lives is wedge_lives + 1 + life is wedge_lives + if life == 1: + # WEDGE: alive forever, progress frozen — no crash, no timeout. + loop while 1 == 1: + heartbeat of [slot, 5] + task_yield of null + n is 0 + loop while n < 20: + heartbeat of [slot, n] + task_yield of null + n is n + 1 + return "recovered" + +sup is supervisor_new of 3 +supervise of [sup, healthy, 0] +supervise of [sup, flaky, 1] +r1 is supervise_run of [sup, 500] +assert_eq of [r1, 1, "wedged worker detected + restarted exactly once"] +assert_eq of [_sup_done[0], 1, "healthy worker completed (never restarted)"] +assert_eq of [_sup_done[1], 1, "flaky worker recovered after its restart"] + +# --- 2. Crashed worker is restarted, and the caught crash does NOT fail the +# run (supervised crash is recovered, not fatal) -------------------- +crash_lives is 0 +define crasher(slot) as: + crash_lives is crash_lives + 1 + life is crash_lives + heartbeat of [slot, 1] + task_yield of null + if life == 1: + throw of "boom" + return "recovered" + +sup2 is supervisor_new of 3 +supervise of [sup2, crasher, 0] +r2 is supervise_run of [sup2, 200] +assert_eq of [r2, 1, "crashed worker restarted exactly once"] +assert_eq of [_sup_done[0], 1, "crashed worker recovered on restart"] + +# --- 3. Restart-intensity cap: a worker that always wedges is restarted at +# most max_restarts times, then supervision gives up ---------------- +define always_wedge(slot) as: + loop while 1 == 1: + heartbeat of [slot, 7] + task_yield of null + +sup3 is supervisor_new of 2 +supervise of [sup3, always_wedge, 0] +r3 is supervise_run of [sup3, 400] +assert_eq of [r3, 2, "restart-intensity cap honored (max_restarts=2)"] + +test_summary of null