RunicGateway/website
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

feat(events): the runner (Phase 2)
All checks were successful
PR Checks / bot-tests (pull_request) Successful in 32s
Details
PR Checks / client-build (pull_request) Successful in 33s
Details
PR Checks / server-tests (pull_request) Successful in 5m29s
Details

Browse Source 
`utils/eventRunner.js`, the eighth poller, wired into server.js beside
engagementWorker. Its tick reclaims stale leases, sweeps occurrences past their
grace window into `missed`, advances each due run through its phases, and drains
that phase's steps in `seq` order. The three core actions from Phase 1 get real
bodies, so a published event started from the existing run route now announces,
waits and completes on its own.

No routes are added: a runner has no surface, and the live controls stay Phase
3's.

Four things the org lead settled (2026-09-02): a parked step is `running` with a
NULL lease; `await: 'human'` and `holdFor` are ordinary success-envelope members
rather than special cases keyed on an action id; a run whose concurrency key is
held stays `scheduled` and lets its grace window decide; and `n` in §L's
`retry(n)` is a runner constant.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
claude
2026-09-02 06:32:24 -05:00
parent d88906e43c
commit 2e964cfeee
10 changed files with 2527 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

510
server/src/utils/eventRunner.js Normal file
View File

@@ -0,0 +1,510 @@
// ── The event runner ───────────────────────────────────────────────────────
//
// EVENTS.md §E, and Phase 2 of EVENTS_PLAN.md. The eighth poller: same
// `setInterval` + `unref()` + `stop()` shape as `announceWorker`, the three Team
// sweepers and `engagementWorker`, wired into `server.js` beside them. In the
// **website** process rather than the bot, which cannot load module code.
//
// Its tick does four things, in this order:
//
// 1. **reclaim** — release leases whose holder died, never touching `attempts`
// 2. **materialise** — sweep occurrences past their grace window into `missed`
// 3. **advance** — claim each due run and move it through its phases
// 4. **prune** — the `event_run_log` retention sweep, on its own long clock
//
// **What "materialise" means in this phase.** §E's tick materialises due
// occurrences from a recurrence; the spec validator accepts `kind: 'manual'`
// alone until Phase 4, so there is no recurrence to expand and the only
// occurrences that exist are the ones an admin created. What that leaves for this
// leg is the half that is already real and already needed: the grace window. A
// run whose instant passed while the process was down does not start late and
// silently — it becomes `missed`, which is terminal and which a human can see
// (§L). Phase 4 adds the expansion above it.
//
// **Two properties this file must not lose**, both already paid for once on this
// codebase:
//
// - **A reclaim never resets `attempts`.** Engagement Phase 14's defect: a sweep
// that returned every stale row to its start state made the attempt ceiling
// unreachable, so the row cycled forever, never terminal, and therefore never
// eligible for any retention sweep.
// - **The unique index, not the claim, is what prevents a double run.** The claim
// decides *who* advances an occurrence; `uq_evrun_occurrence` is what stops two
// of them existing. Neither substitutes for the other.
//
// §N4 settled this deployment as single-instance, so the `--scale app=2` rig the
// engagement workstream used is deliberately not built. Every claim path is here
// exactly as §E specifies anyway, because the multi-instance case is not what
// they are for on a single container: the CAS is what protects a tick that
// overran into the next one, and the lease and its reclaim are what recover a
// step whose process died mid-dispatch. Both happen with one app.
const os = require('os')
const runsDb = require('../model/events/eventRuns.db')
const stepsDb = require('../model/events/eventRunSteps.db')
const logDb = require('../model/events/eventRunLog.db')
const versionsDb = require('../model/events/eventVersions.db')
const registries = require('../modules/registries')
const { dispatchStep } = require('../events/dispatch')
const log = require('./logger')('event-runner')
const POLL_MS = Number(process.env.EVENT_POLL_MS) || 15_000
// How many runs one sweep looks at, and how many steps it will drain from one
// run. Bounds rather than targets: the tick runs again in POLL_MS, and an
// unbounded batch is how a backlog turns one tick into a stall. The step bound
// also caps how long one run can hold the tick, which is what keeps a
// forty-step phase from starving every other run on the deployment.
const RUN_BATCH = Number(process.env.EVENT_RUN_BATCH) || 50
const STEPS_PER_TICK = Number(process.env.EVENT_STEPS_PER_TICK) || 25
// A step's retries (org lead, 2026-09-02). §L specifies `retry(n) -> skip |
// pause | abort_run` and names no `n`; it lives here, as one number an operator
// can change, rather than in a column no authoring surface would ever show.
//
// Flat backoff rather than exponential, for the reason the outbox's is flat:
// `due_at` is also the event's own clock, and a doubling backoff pushes a step
// arbitrarily far past the moment the event was about.
const MAX_ATTEMPTS = Number(process.env.EVENT_STEP_MAX_ATTEMPTS) || 3
const RETRY_MS = Number(process.env.EVENT_STEP_RETRY_MS) || 60_000
// How long a claim may look alive before the reclaim takes it back. The run
// lease has to outlast a whole tick's work on one run; a step's is computed from
// its own action's `budgetMs` (see `leaseFor`), because a registry that lets an
// action declare an hour would otherwise have its steps reclaimed and
// re-dispatched fifty-nine minutes before they answered.
const RUN_LEASE_MS = Number(process.env.EVENT_RUN_LEASE_MS) || 15 * 60 * 1000
const STEP_LEASE_MARGIN_MS = 60_000
// The log retention horizon, and how often the sweep runs. It is folded into
// this tick rather than given a ninth timer because it shares the tick's only
// real dependency — a database — and a sweep that runs four times a day does not
// need an interval of its own. Only TERMINAL runs are ever eligible, which is the
// rule Engagement Phase 14 arrived at.
const LOG_RETENTION_DAYS = Number(process.env.EVENT_LOG_RETENTION_DAYS) || 90
const PRUNE_EVERY_MS = 6 * 60 * 60 * 1000
// Who this process is, for `claimed_by`. Host and pid, so a stranded claim in the
// table names the thing that stranded it.
const OWNER = `${os.hostname()}:${process.pid}`.slice(0, 64)
const RUN_TERMINAL = runsDb.TERMINAL
/** The lease a step's dispatch gets: its action's own budget, plus a margin. */
function leaseFor(step, now) {
const action = registries.eventAction(step.action_id)
const budget = action?.budgetMs || 10_000
return new Date(now.getTime() + budget + STEP_LEASE_MARGIN_MS)
}
// ── The disposition of a step that has run out of road ─────────────────────
//
// **All three dispositions write the step `failed`.** `on_failure` says what
// happens to the RUN, not what happened to the step, and a step that was
// attempted three times and never worked is `failed` under every one of them.
// `skipped` is reserved for a step a human skipped from the run console (Phase
// 3) — a status that meant both "nobody ran this" and "this failed and we moved
// on" would make the run console's summary line unreadable.
async function applyFailure(run, step, error) {
await stepsDb.finish(step.id, 'failed', error)
await logDb.write({
runId: run.id,
stepId: step.id,
kind: 'step.status',
phase: step.phase,
detail: { to: 'failed', action: step.action_id, attempts: step.attempts + 1, onFailure: step.on_failure, error },
})
// A run that lost a step is degraded whatever happens next. Health is not
// status (§E): a run can be genuinely running and degraded at once, and the
// admin surface needs to say so without claiming the run stopped.
if (await runsDb.setHealth(run.id, 'degraded')) {
await logDb.write({ runId: run.id, kind: 'run.health', detail: { to: 'degraded', because: step.action_id } })
}
if (step.on_failure === 'abort_run') {
// §L: the disposition for `irreversible`. Nothing further is dispatched, and
// the pending steps are cancelled rather than left looking due forever.
const cancelled = await stepsDb.cancelPending(run.id)
await runsDb.transition(run.id, ['starting', 'running', 'ending'], 'failed', { error })
await logDb.write({
runId: run.id,
kind: 'run.status',
phase: step.phase,
detail: { to: 'failed', because: step.action_id, cancelledSteps: cancelled },
})
return 'stop'
}
if (step.on_failure === 'pause') {
// The default for `change`, and the right one when the world is half-altered:
// stop advancing and wait for a human. `paused` is excluded from `findDue`,
// so nothing here picks it up again — Phase 3's resume control is the only
// thing that moves it.
await runsDb.transition(run.id, ['starting', 'running'], 'paused', { error })
await logDb.write({
runId: run.id,
kind: 'run.status',
phase: step.phase,
detail: { to: 'paused', because: step.action_id },
})
return 'stop'
}
// 'skip': the run carries on, degraded, and the failure is on the record.
return 'continue'
}
/**
* Claim one step, dispatch it, and record what came back.
*
* Answers `'continue'` (the phase may proceed), `'stop'` (it may not, for now or
* ever) or `'taken'` (somebody else claimed it first).
*/
async function drainStep(run, step, now, carry = {}) {
if (!(await stepsDb.claim(step.id, OWNER, leaseFor(step, now), now))) return 'taken'
const result = await dispatchStep(step, { run })
if (result.actionVersionDrift) {
await logDb.write({
runId: run.id,
stepId: step.id,
kind: 'step.status',
phase: step.phase,
detail: { action: step.action_id, versionDrift: result.actionVersionDrift },
})
}
if (result.outcome === 'parked') {
// The GM cue. The step stays `running` with a NULL lease: genuinely in
// flight, nothing holding it, so the stale reclaim passes it by and a cue
// posted on Friday is still waiting on Monday. Phase 3's confirm control is
// what ends it.
await stepsDb.park(step.id, null)
await logDb.write({
runId: run.id,
stepId: step.id,
kind: 'step.parked',
phase: step.phase,
detail: { action: step.action_id, params: step.params },
})
return 'stop'
}
if (result.outcome === 'done') {
await stepsDb.finish(step.id, 'done', null)
if (result.holdSeconds > 0) {
// `core.wait`, and any module action that answers `holdFor`. The pause is
// the NEXT step's `due_at` and it is set here, by the runner, because a
// `perform()` that slept would hold its claim for the duration.
const until = new Date(now.getTime() + result.holdSeconds * 1000)
if (!(await stepsDb.holdNext(run.id, step.phase, step.seq, until))) {
// **Nothing after it in this phase**, which is the case a wait written as
// the last step of a phase produces. Dropping the hold here would make
// "announce, wait five minutes, then the next phase" start the next phase
// at once — a wait that silently meant nothing. The later phase's steps do
// not exist yet, so the instant is carried out to `advanceRun` and applied
// when they are materialised.
carry.holdUntil = until
}
}
await logDb.write({
runId: run.id,
stepId: step.id,
kind: 'step.status',
phase: step.phase,
detail: { to: 'done', action: step.action_id, holdSeconds: result.holdSeconds || 0 },
})
return 'continue'
}
if (result.outcome === 'retry' && step.attempts + 1 < MAX_ATTEMPTS) {
await stepsDb.reschedule(step.id, new Date(now.getTime() + RETRY_MS), result.error)
await logDb.write({
runId: run.id,
stepId: step.id,
kind: 'step.retry',
phase: step.phase,
detail: { action: step.action_id, attempt: step.attempts + 1, of: MAX_ATTEMPTS, error: result.error },
})
// Degraded from the FIRST retry, not from the eventual failure. An event
// whose announcements are landing on the second attempt is having trouble
// now, and that is when an operator wants to know.
if (await runsDb.setHealth(run.id, 'degraded')) {
await logDb.write({ runId: run.id, kind: 'run.health', detail: { to: 'degraded', because: step.action_id } })
}
return 'stop'
}
// Terminal, or transient with the attempts spent. Same disposition either way:
// §L's `retry(n) -> ...` has arrived at the arrow.
return applyFailure(run, step, result.error)
}
/**
* Advance one claimed run as far as it will go this tick.
*
* The loop is bounded by `STEPS_PER_TICK` and exits on the first thing it cannot
* get past — a parked step, a step whose `due_at` is in the future, a step
* somebody else holds, or a phase that is not finished.
*/
async function advanceRun(run, now) {
const version = await versionsDb.getById(run.version_id)
const phases = version?.spec?.phases
if (!Array.isArray(phases) || !phases.length) {
// The pinned version is unreadable. `version_id`'s foreign key RESTRICTs
// precisely so this cannot be a deleted row, so it is corruption rather than
// an ordinary race — terminal, named, and not retried.
await runsDb.transition(run.id, ['scheduled', 'starting', 'running', 'ending'], 'failed', {
error: 'the pinned version has no phases',
})
await logDb.write({ runId: run.id, kind: 'run.status', detail: { to: 'failed', because: 'pinned version has no phases' } })
return 'failed'
}
let phaseKey = run.current_phase
if (run.status === 'starting') {
// Entering the first phase. `materialisePhase` is INSERT IGNORE against
// `uq_evstep_slot`, so doing it again over the rows Phase 1's `create()`
// already wrote is a no-op — which is what makes recovery from a process that
// died between the claim and here uneventful.
const first = phases[0]
await stepsDb.materialisePhase(run.id, first.key, first.steps || [])
if (!(await runsDb.transition(run.id, 'starting', 'running', { phase: first.key }))) return 'taken'
phaseKey = first.key
await logDb.write({ runId: run.id, kind: 'run.status', phase: first.key, detail: { from: 'starting', to: 'running' } })
}
if (run.status === 'ending') {
// A run that reached the wind-down and then lost its process. Phase 8 puts
// cleanup here; until then `ending` is a state a run passes through rather
// than one it does work in, and completing it is the whole recovery.
await runsDb.transition(run.id, 'ending', 'completed')
await logDb.write({ runId: run.id, kind: 'run.status', detail: { from: 'ending', to: 'completed' } })
return 'completed'
}
// A hold a `core.wait` could not place because nothing followed it in its own
// phase. It crosses the phase boundary with the run rather than being dropped.
const carry = {}
for (let n = 0; n < STEPS_PER_TICK; n++) {
const phaseIndex = phases.findIndex((p) => p.key === phaseKey)
if (phaseIndex < 0) {
await runsDb.transition(run.id, ['running'], 'failed', { error: `phase "${phaseKey}" is not in the pinned version` })
await logDb.write({ runId: run.id, kind: 'run.status', detail: { to: 'failed', because: `unknown phase "${phaseKey}"` } })
return 'failed'
}
const step = await stepsDb.nextOpenStep(run.id, phaseKey)
if (step && step.status === 'running') return 'in-flight' // parked, or somebody's dispatch
if (step && step.due_at && new Date(step.due_at) > now) return 'waiting' // behind a core.wait
if (step) {
const outcome = await drainStep({ ...run, current_phase: phaseKey }, step, now, carry)
if (outcome === 'continue') continue
return outcome === 'taken' ? 'taken' : 'stopped'
}
// Every step of this phase is terminal.
await logDb.write({ runId: run.id, kind: 'phase.completed', phase: phaseKey, detail: { index: phaseIndex } })
const next = phases[phaseIndex + 1]
if (!next) {
// §E's `ending` exists for the reason `sending` does in the outbox — it is
// what a claim sets — so the run passes through it even though Phase 2 has
// no cleanup to do there. Phase 8 is what gives it work.
if (!(await runsDb.transition(run.id, 'running', 'ending'))) return 'taken'
await logDb.write({ runId: run.id, kind: 'run.status', phase: phaseKey, detail: { from: 'running', to: 'ending' } })
await runsDb.transition(run.id, 'ending', 'completed')
await logDb.write({ runId: run.id, kind: 'run.status', detail: { from: 'ending', to: 'completed' } })
return 'completed'
}
await stepsDb.materialisePhase(run.id, next.key, next.steps || [])
if (carry.holdUntil) {
// `seq > -1` is the first step of the phase just created. Applied after
// materialisation because that is the first moment there is a row to hold.
await stepsDb.holdNext(run.id, next.key, -1, carry.holdUntil)
carry.holdUntil = null
}
// `running -> running` is not a no-op: it is a guarded write of
// `current_phase` that fails if the run stopped being `running` underneath
// this tick, which is what a cancel from the admin surface looks like.
if (!(await runsDb.transition(run.id, 'running', 'running', { phase: next.key }))) return 'taken'
phaseKey = next.key
await logDb.write({ runId: run.id, kind: 'phase.entered', phase: next.key, detail: { steps: (next.steps || []).length } })
}
return 'bounded' // more to do; the next tick picks it up
}
/** Claim one due run, work it, and hand the lease back if it is still in flight. */
async function processRun(run, now = new Date()) {
if (run.status === 'scheduled') {
const holder = await runsDb.concurrencyHolder(run.concurrency_key, run.id)
if (holder) {
// The org lead's answer for a held key (2026-09-02): hold at `scheduled`
// and let the grace window decide. Nothing is destroyed, nothing starts
// silently late, and if the holder outlasts the window the missed sweep
// makes this run terminal and visible.
//
// Logged only when the reason CHANGES. A blocked run is re-examined every
// tick, and a line per tick for the length of a grace window would bury the
// one line that matters under a thousand identical ones.
const message = `held: run ${holder.id} has concurrency key "${run.concurrency_key}"`
if (run.last_error !== message) {
await runsDb.transition(run.id, 'scheduled', 'scheduled', { error: message })
await logDb.write({
runId: run.id,
kind: 'run.blocked',
detail: { concurrencyKey: run.concurrency_key, heldBy: holder.id, holderStatus: holder.status },
})
}
return 'blocked'
}
if (!(await runsDb.claimStart(run.id, OWNER, new Date(now.getTime() + RUN_LEASE_MS)))) return 'taken'
await logDb.write({ runId: run.id, kind: 'run.status', detail: { from: 'scheduled', to: 'starting' } })
run = { ...run, status: 'starting' }
} else if (!(await runsDb.claimTick(run.id, OWNER, new Date(now.getTime() + RUN_LEASE_MS), now))) {
return 'taken'
}
try {
const outcome = await advanceRun(run, now)
// A run that is not finished must not keep its lease: it would be
// unadvanceable until the lease expired, which would turn every `core.wait`
// into `max(wait, RUN_LEASE_MS)`. A terminal transition already cleared it.
if (!['completed', 'failed'].includes(outcome)) await runsDb.releaseClaim(run.id, OWNER)
return outcome
} catch (err) {
await runsDb.releaseClaim(run.id, OWNER)
throw err
}
}
/** Occurrences that passed their own grace window while nothing was running (§L). */
async function sweepMissed(now) {
const missed = await runsDb.findMissed(now)
let n = 0
for (const run of missed) {
if (await runsDb.transition(run.id, 'scheduled', 'missed', { error: 'the grace window passed' })) {
await stepsDb.cancelPending(run.id)
await logDb.write({
runId: run.id,
kind: 'run.status',
detail: { to: 'missed', scheduledFor: run.scheduled_for },
})
n += 1
}
}
return n
}
let lastPruneAt = 0
async function prune(now) {
if (now.getTime() - lastPruneAt < PRUNE_EVERY_MS) return 0
lastPruneAt = now.getTime()
const before = new Date(now.getTime() - LOG_RETENTION_DAYS * 24 * 60 * 60 * 1000)
const deleted = await logDb.pruneTerminal(before)
if (deleted) log.info('event run log pruned', { deleted, before, retentionDays: LOG_RETENTION_DAYS })
return deleted
}
async function tick(now = new Date()) {
try {
await runsDb.reclaimStale(now)
await stepsDb.reclaimStale(now, MAX_ATTEMPTS)
} catch (err) {
log.error('failed to reclaim stale claims', { message: err.message })
}
try {
await sweepMissed(now)
} catch (err) {
log.error('missed sweep failed', { message: err.message })
}
let due
try {
due = await runsDb.findDue(now, RUN_BATCH)
} catch (err) {
log.error('failed to load due runs', { message: err.message })
return
}
const counts = {}
for (const run of due || []) {
try {
const outcome = await processRun(run, now)
counts[outcome] = (counts[outcome] || 0) + 1
} catch (err) {
log.error('run failed', { run: run.id, message: err.message })
}
}
if (due && due.length) log.info('event runs swept', { due: due.length, ...counts })
try {
await prune(now)
} catch (err) {
log.error('log prune failed', { message: err.message })
}
}
let timer = null
// Guards against this process running two ticks over the same runs at once.
// `setInterval` fires whether or not the last callback returned, and the CAS
// alone does not cover it now that a live lease is not re-enterable by its own
// owner — a second tick would simply find every run claimed and do nothing
// useful, one query at a time, for as long as the first one ran.
let ticking = false
function start() {
if (timer) return timer
timer = setInterval(() => {
if (ticking) {
log.warn('event tick still running; skipping this interval')
return
}
ticking = true
tick()
.catch((err) => log.error('event tick failed', { message: err.message }))
.finally(() => {
ticking = false
})
}, POLL_MS)
if (timer.unref) timer.unref() // don't keep the event loop alive (tests, shutdown)
log.info('event runner started', { pollMs: POLL_MS, owner: OWNER, maxAttempts: MAX_ATTEMPTS })
return timer
}
function stop() {
if (timer) {
clearInterval(timer)
timer = null
}
}
module.exports = {
start,
stop,
tick,
processRun,
advanceRun,
drainStep,
sweepMissed,
prune,
OWNER,
POLL_MS,
MAX_ATTEMPTS,
RETRY_MS,
RUN_LEASE_MS,
LOG_RETENTION_DAYS,
RUN_TERMINAL,
}