Rust PLAN_REDESIGNS §14.3.12 (D370), stage 9 of §14.3. A definition's published version may carry `starts`: a trigger, a condition and a cooldown. A matching firing creates a run of it, now. - spec: `starts` validated by the phase gate's own function (same grammar, same dormant rule; a dormant start refuses publish), a cooldown from 0m to 7d (30m default), at most ten, event.* refused. - events/starts.js: the emit path's third subscriber, beside dispatch and the gate observer. An unverified version starts nothing; one run at a time per definition and the cooldown are one conditional UPDATE on the new event_trigger_starts table; the occurrence key makes one firing reported twice one run. Ignored firings are counted by why. - event_runs.origin: the firing (trigger, label, time, subject, matched condition, declared variables), also on the run.created audit line. - Editor: a "Started by the game" panel using the gate's condition builder, with the ignored-firings line. Run console: "Started because ... fired". - No module contract change and no MODULE_API bump. Tests: eventStarts (server 21, client 8); eventRunnerSql +7 against a real MariaDB (75/75); Module-uo proof on uo.champ.boss_up against the dev database. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01E14m6SuuY6i1vASFeGDBeY
1894 lines
88 KiB
JavaScript
1894 lines
88 KiB
JavaScript
// ── The runner's raw SQL, against a real MariaDB ───────────────────────────
|
||
//
|
||
// EVENTS_PLAN.md Phase 2. `eventRunner.test.js` stubs the three tables and
|
||
// exercises everything the runner DECIDES. It cannot prove the statements whose
|
||
// whole correctness is a server contract, and on this codebase that gap has
|
||
// already cost something once: engagement's cooldown claim was green against its
|
||
// stub and always allowed the send against a real server, because the connector
|
||
// defaults `foundRows: true` and a no-op UPDATE reports 1 rather than 0.
|
||
//
|
||
// So the five statements that decide who owns what run here, for real:
|
||
//
|
||
// • **`claimStart`** — the CAS `scheduled -> starting`. "Exactly one winner" is
|
||
// `affectedRows = 1` for one caller and 0 for every other, and that is a
|
||
// property of the SERVER's answer, not of the SQL's shape.
|
||
// • **`claimTick`** — the same, for a run already in flight, and with no
|
||
// owner-matches escape clause. A live lease must refuse its own holder, or
|
||
// one `setInterval` that overran advances one run twice.
|
||
// • **`transition`** — a guarded status move. The guard is the whole thing: a
|
||
// run cancelled between the read and the write must not be transitioned.
|
||
// • **`reclaimStale`** on steps — two statements in a fixed ORDER, give-up
|
||
// before hand-back. Reversing them makes `MAX_ATTEMPTS` unreachable and the
|
||
// row cycles forever (Engagement Phase 14's defect), and **neither statement
|
||
// may touch `attempts`**.
|
||
// • **`holdNext`** — `UPDATE ... ORDER BY seq LIMIT 1` with a guard, which is
|
||
// both a MariaDB-specific syntax and a correctness claim: it must move the
|
||
// next PENDING step and only ever push a due date later.
|
||
//
|
||
// **Phase 3 added four more**, and each of them is a control a staff member
|
||
// presses against a live game world:
|
||
//
|
||
// * **`confirmParked` / `skipByHuman`** - both keyed on `status = 'running'
|
||
// AND claim_expires_at IS NULL`. That pair, and only that pair, means "a cue
|
||
// waiting on a human". If the clause let a LEASED step through, confirm would
|
||
// race the process mid-dispatch on that row.
|
||
// * **`cancelOpen`** - pending steps and parked cues, never a leased one.
|
||
// * **`lastStartedSeq`** - a `MAX(seq) ... WHERE status <> 'pending'`, which is
|
||
// what decides whether retry is offered. The first draft asked for the LOWEST
|
||
// unsettled seq instead, which is a different step whenever a phase carried
|
||
// on past an `on_failure: skip` failure.
|
||
//
|
||
// **Phase 4 added two reads**, and a read earns a place here when a stub cannot
|
||
// tell it is wrong:
|
||
//
|
||
// * **`findSchedulable`** - the query the runner runs on EVERY tick to decide
|
||
// what has a recurrence to expand. It joins a definition to its published
|
||
// version and left-joins the series, and every stub of it in
|
||
// `eventSchedule.test.js` is a hand-written object rather than that join. A
|
||
// syntax error or a wrong join direction here is a runner that materialises
|
||
// nothing, silently, for ever.
|
||
// * **`repinScheduled`** - the UPDATE publish runs over already-materialised
|
||
// occurrences. Its guard is the whole statement, and the two rows it must
|
||
// NOT touch are a run that has started and a run that is already terminal.
|
||
// * **`listInWindow`** - the calendar's real half, with a correlated subquery
|
||
// for `waiting_steps` and a LEFT JOIN that must not drop a definition with no
|
||
// series.
|
||
//
|
||
// **Phase 5 added the gate statements**, and the increment is the one on this
|
||
// list with the closest precedent: engagement's cooldown claim was green against
|
||
// its stub and always allowed the send against a real server, because the
|
||
// connector defaults `foundRows: true`. A gate's tally is the same shape of
|
||
// claim — a conditional UPDATE whose answer decides something — so it is proved
|
||
// here rather than believed.
|
||
//
|
||
// * **`GATE_COUNT`** - `tally = tally + 1` with `CASE WHEN tally + 1 >=
|
||
// needed` inside the same statement, guarded on `satisfied_at IS NULL`. Two
|
||
// emits arriving together must each add one and exactly ONE of them must
|
||
// cross the threshold; a read-then-write would let both see 2 of 3. **This
|
||
// is the statement that failed here first**: MariaDB evaluates SET
|
||
// assignments left to right with the values already assigned, so an
|
||
// increment written before the CASE made the CASE read the new tally and
|
||
// close a two-firing gate on the first firing. The ORDER of the SET list is
|
||
// load-bearing, and only a real server says so.
|
||
// * **`GATE_SATISFY`** - the same guard for the two reasons that are not a
|
||
// firing: an `after` deadline passing, and a human forcing it. A force that
|
||
// races the tick must lose harmlessly.
|
||
// * **`GATE_OPEN_FOR_TRIGGER`** - the emit path's only query, and the one
|
||
// index in this feature on a hot path. Its JOIN is what stops a gate
|
||
// belonging to a cancelled run counting for ever.
|
||
// * **`uq_evgate_phase`** - what makes `open` an INSERT IGNORE, so a process
|
||
// that died between entering a phase and opening its gate opens no second
|
||
// one on the next tick.
|
||
//
|
||
// **Phase 8 added the resource ledger**, and its unique key is the single most
|
||
// server-dependent thing in this feature. "Two events cannot hold a lease on one
|
||
// target" has to hold among LIVE rows only — last week's finished event must not
|
||
// keep this week's from leasing the same rate — and MariaDB has no partial index,
|
||
// so the encoding is a STORED generated column that goes NULL once the row is no
|
||
// longer ours. Whether multiple NULLs collide in a unique index is a property of
|
||
// the server and of nothing else, and TEAMS.md §2.5 already had to be corrected
|
||
// once on this exact shape: MariaDB refuses ON DELETE SET NULL on a foreign key
|
||
// whose column is a base column of a stored generated column (error 1901), which
|
||
// is why the expression reads `status` alone and `step_id` stays a SET NULL FK.
|
||
// Both halves of that are proved below rather than believed.
|
||
//
|
||
// Plus the two unique indexes that are load-bearing rather than tidy:
|
||
// `uq_evrun_occurrence` (which, not the claim, is what stops two runs of one
|
||
// occurrence existing) and `uq_evstep_slot` (which is what makes re-materialising
|
||
// a phase a no-op).
|
||
//
|
||
// **It SKIPS when there is no database**, deliberately: CI runs the suite with
|
||
// the pool pointed at a dead port, and a file that failed there would make every
|
||
// PR red for a reason unrelated to itself. Run it against this machine's
|
||
// container with:
|
||
//
|
||
// DB_HOST=127.0.0.1 DB_PORT=3306 DB_USER=... DB_PASSWORD=... \
|
||
// node --test test/eventRunnerSql.test.js
|
||
//
|
||
// It creates its tables in a throwaway database named after the process and
|
||
// drops it again, so it can never touch a real schema.
|
||
|
||
const { test, before, after, beforeEach } = require('node:test')
|
||
const assert = require('node:assert/strict')
|
||
const mariadb = require('mariadb')
|
||
|
||
// Trimmed to the columns these statements read or write. The ENUMs are verbatim,
|
||
// because "is `missed` a legal value" is one of the things being proved.
|
||
const SCHEMA = `
|
||
CREATE TABLE event_series (
|
||
id INT AUTO_INCREMENT PRIMARY KEY,
|
||
name VARCHAR(160) NOT NULL,
|
||
slug VARCHAR(160) NOT NULL,
|
||
ordering INT NOT NULL DEFAULT 0
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||
CREATE TABLE event_definitions (
|
||
id INT AUTO_INCREMENT PRIMARY KEY,
|
||
title VARCHAR(200) NOT NULL DEFAULT 'x',
|
||
slug VARCHAR(200) NOT NULL DEFAULT 'x',
|
||
state ENUM('draft','ready','archived') NOT NULL DEFAULT 'draft',
|
||
current_version_id INT NULL,
|
||
series_id INT NULL,
|
||
concurrency_key VARCHAR(190) NULL,
|
||
timezone VARCHAR(64) NOT NULL DEFAULT 'UTC',
|
||
grace_seconds INT NOT NULL DEFAULT 900
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||
CREATE TABLE event_versions (
|
||
id INT AUTO_INCREMENT PRIMARY KEY,
|
||
definition_id INT NOT NULL,
|
||
version INT NOT NULL DEFAULT 1,
|
||
spec JSON NOT NULL,
|
||
verified_at DATETIME NULL
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||
CREATE TABLE event_runs (
|
||
id BIGINT AUTO_INCREMENT PRIMARY KEY,
|
||
definition_id INT NOT NULL,
|
||
version_id INT NOT NULL,
|
||
scope VARCHAR(190) NOT NULL DEFAULT '',
|
||
status ENUM('scheduled','starting','running','paused','ending',
|
||
'completed','cancelled','failed','missed')
|
||
NOT NULL DEFAULT 'scheduled',
|
||
health ENUM('ok','degraded','stalled') NOT NULL DEFAULT 'ok',
|
||
current_phase VARCHAR(64) NULL,
|
||
scheduled_for DATETIME NOT NULL,
|
||
timezone VARCHAR(64) NOT NULL DEFAULT 'UTC',
|
||
concurrency_key VARCHAR(190) NULL,
|
||
params JSON NULL,
|
||
rehearsal TINYINT(1) NOT NULL DEFAULT 0,
|
||
started_at DATETIME NULL,
|
||
ended_at DATETIME NULL,
|
||
claimed_by VARCHAR(64) NULL,
|
||
claim_expires_at DATETIME NULL,
|
||
started_by INT NULL,
|
||
origin JSON NULL,
|
||
last_error VARCHAR(500) NULL,
|
||
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
||
UNIQUE KEY uq_evrun_occurrence (definition_id, scope, scheduled_for),
|
||
INDEX idx_evrun_due (status, scheduled_for),
|
||
INDEX idx_evrun_concurrency (concurrency_key, status)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||
CREATE TABLE event_run_steps (
|
||
id BIGINT AUTO_INCREMENT PRIMARY KEY,
|
||
run_id BIGINT NOT NULL,
|
||
phase VARCHAR(64) NOT NULL,
|
||
seq INT NOT NULL,
|
||
action_id VARCHAR(96) NOT NULL,
|
||
params JSON NULL,
|
||
action_version INT NOT NULL DEFAULT 1,
|
||
status ENUM('pending','running','done','failed','skipped','refused','cancelled')
|
||
NOT NULL DEFAULT 'pending',
|
||
due_at DATETIME NULL,
|
||
attempts INT NOT NULL DEFAULT 0,
|
||
on_failure VARCHAR(32) NOT NULL DEFAULT 'skip',
|
||
idempotency_key CHAR(40) NOT NULL,
|
||
claimed_by VARCHAR(64) NULL,
|
||
claim_expires_at DATETIME NULL,
|
||
last_error VARCHAR(500) NULL,
|
||
started_at DATETIME NULL,
|
||
finished_at DATETIME NULL,
|
||
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
||
UNIQUE KEY uq_evstep_slot (run_id, phase, seq),
|
||
INDEX idx_evstep_due (status, due_at)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||
CREATE TABLE event_run_phase_gates (
|
||
id BIGINT AUTO_INCREMENT PRIMARY KEY,
|
||
run_id BIGINT NOT NULL,
|
||
phase VARCHAR(64) NOT NULL,
|
||
kind ENUM('after','on') NOT NULL,
|
||
after_seconds INT NULL,
|
||
trigger_id VARCHAR(96) NULL,
|
||
conditions JSON NULL,
|
||
needed INT NOT NULL DEFAULT 1,
|
||
tally INT NOT NULL DEFAULT 0,
|
||
entered_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
due_at DATETIME NULL,
|
||
last_event JSON NULL,
|
||
last_event_at DATETIME NULL,
|
||
satisfied_at DATETIME NULL,
|
||
satisfied_by VARCHAR(16) NULL,
|
||
forced_by INT NULL,
|
||
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
||
CONSTRAINT fk_evgate_run FOREIGN KEY (run_id) REFERENCES event_runs(id) ON DELETE CASCADE,
|
||
UNIQUE KEY uq_evgate_phase (run_id, phase),
|
||
INDEX idx_evgate_open (trigger_id, satisfied_at)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||
CREATE TABLE event_run_budget (
|
||
id BIGINT AUTO_INCREMENT PRIMARY KEY,
|
||
run_id BIGINT NOT NULL,
|
||
dimension VARCHAR(96) NOT NULL,
|
||
consumed INT NOT NULL DEFAULT 0,
|
||
cap INT NULL,
|
||
effective_from VARCHAR(96) NULL,
|
||
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
||
CONSTRAINT fk_evbud_run FOREIGN KEY (run_id) REFERENCES event_runs(id) ON DELETE CASCADE,
|
||
UNIQUE KEY uq_evbud_dim (run_id, dimension)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||
CREATE TABLE event_run_resources (
|
||
id BIGINT AUTO_INCREMENT PRIMARY KEY,
|
||
run_id BIGINT NOT NULL,
|
||
step_id BIGINT NULL,
|
||
owner_module VARCHAR(64) NOT NULL,
|
||
kind VARCHAR(64) NOT NULL,
|
||
ref VARCHAR(190) NOT NULL,
|
||
payload JSON NULL,
|
||
lease_until DATETIME NULL,
|
||
status ENUM('pending','confirmed','reverting','reverted','orphaned','drifted')
|
||
NOT NULL DEFAULT 'pending',
|
||
revert_attempts INT NOT NULL DEFAULT 0,
|
||
last_error VARCHAR(500) NULL,
|
||
member_key VARCHAR(190) NULL,
|
||
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
||
live_marker TINYINT AS (IF(status IN ('pending','confirmed','reverting'), 1, NULL)) STORED,
|
||
CONSTRAINT fk_evres_run FOREIGN KEY (run_id) REFERENCES event_runs(id) ON DELETE CASCADE,
|
||
CONSTRAINT fk_evres_step FOREIGN KEY (step_id) REFERENCES event_run_steps(id) ON DELETE SET NULL,
|
||
UNIQUE KEY uq_evres_target (owner_module, kind, ref, live_marker),
|
||
INDEX idx_evres_run (run_id, status),
|
||
INDEX idx_evres_live (status, lease_until)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||
CREATE TABLE event_action_settings (
|
||
action_id VARCHAR(96) NOT NULL PRIMARY KEY,
|
||
enabled TINYINT(1) NOT NULL DEFAULT 0,
|
||
caps JSON NULL,
|
||
updated_by INT NULL,
|
||
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||
CREATE TABLE event_trigger_starts (
|
||
definition_id INT NOT NULL PRIMARY KEY,
|
||
claimed_at DATETIME NULL,
|
||
last_started_at DATETIME NULL,
|
||
last_run_id BIGINT NULL,
|
||
ignored_running INT NOT NULL DEFAULT 0,
|
||
ignored_cooldown INT NOT NULL DEFAULT 0,
|
||
ignored_unverified INT NOT NULL DEFAULT 0,
|
||
last_ignored_at DATETIME NULL,
|
||
last_ignored_reason VARCHAR(32) NULL,
|
||
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||
`
|
||
|
||
// The statements under test, verbatim from `eventRuns.db.js` and
|
||
// `eventRunSteps.db.js`. Duplicated rather than required, because requiring the
|
||
// modules would drag in `utils/db`'s pool, which the harness has already pointed
|
||
// at a dead port. The pool below leaves `foundRows` at the connector's default,
|
||
// exactly as `utils/db.js` does — pinning it here would make this file agree with
|
||
// the code by construction and prove nothing about the pool the server runs.
|
||
const CLAIM_START = `
|
||
UPDATE event_runs
|
||
SET status = 'starting', claimed_by = ?, claim_expires_at = ?,
|
||
started_at = COALESCE(started_at, NOW())
|
||
WHERE id = ? AND status = 'scheduled'`
|
||
|
||
const CLAIM_TICK = `
|
||
UPDATE event_runs
|
||
SET claimed_by = ?, claim_expires_at = ?
|
||
WHERE id = ?
|
||
AND status IN ('starting','running','ending')
|
||
AND (claim_expires_at IS NULL OR claim_expires_at < ?)`
|
||
|
||
const TRANSITION = `
|
||
UPDATE event_runs SET status = ?, current_phase = ?
|
||
WHERE id = ? AND status IN (?)`
|
||
|
||
const CLAIM_STEP = `
|
||
UPDATE event_run_steps
|
||
SET status = 'running', attempts = attempts + 1, claimed_by = ?, claim_expires_at = ?,
|
||
started_at = COALESCE(started_at, NOW())
|
||
WHERE id = ? AND status = 'pending' AND (due_at IS NULL OR due_at <= ?)`
|
||
|
||
const STEP_GIVE_UP = `
|
||
UPDATE event_run_steps
|
||
SET status = 'failed', last_error = 'gave up after repeated interruptions',
|
||
finished_at = NOW(), claimed_by = NULL, claim_expires_at = NULL
|
||
WHERE status = 'running'
|
||
AND claim_expires_at IS NOT NULL AND claim_expires_at < ?
|
||
AND attempts >= ?`
|
||
|
||
const STEP_RECLAIM = `
|
||
UPDATE event_run_steps
|
||
SET status = 'pending', claimed_by = NULL, claim_expires_at = NULL
|
||
WHERE status = 'running' AND claim_expires_at IS NOT NULL AND claim_expires_at < ?`
|
||
|
||
const HOLD_NEXT = `
|
||
UPDATE event_run_steps
|
||
SET due_at = ?
|
||
WHERE run_id = ? AND phase = ? AND seq > ? AND status = 'pending'
|
||
AND (due_at IS NULL OR due_at < ?)
|
||
ORDER BY seq LIMIT 1`
|
||
|
||
// Phase 3's four, verbatim from `eventRunSteps.db.js`.
|
||
const CONFIRM_PARKED = `
|
||
UPDATE event_run_steps
|
||
SET status = 'done', finished_at = NOW(), claimed_by = NULL,
|
||
last_error = ?
|
||
WHERE id = ? AND status = 'running' AND claim_expires_at IS NULL`
|
||
|
||
const SKIP_BY_HUMAN = `
|
||
UPDATE event_run_steps
|
||
SET status = 'skipped', finished_at = NOW(), claimed_by = NULL,
|
||
last_error = ?
|
||
WHERE id = ?
|
||
AND (status = 'pending' OR (status = 'running' AND claim_expires_at IS NULL))`
|
||
|
||
const REQUEUE = `
|
||
UPDATE event_run_steps
|
||
SET status = 'pending', attempts = 0, due_at = NULL, last_error = NULL,
|
||
claimed_by = NULL, claim_expires_at = NULL, finished_at = NULL
|
||
WHERE id = ? AND status = 'failed'`
|
||
|
||
const CANCEL_OPEN = `
|
||
UPDATE event_run_steps
|
||
SET status = 'cancelled', finished_at = NOW(), claimed_by = NULL
|
||
WHERE run_id = ?
|
||
AND (status = 'pending' OR (status = 'running' AND claim_expires_at IS NULL))`
|
||
|
||
const LAST_STARTED_SEQ = `
|
||
SELECT MAX(seq) AS seq FROM event_run_steps
|
||
WHERE run_id = ? AND phase = ? AND status <> 'pending'`
|
||
|
||
// Phase 5's four, verbatim from `eventPhaseGates.db.js`.
|
||
const GATE_OPEN = `
|
||
INSERT IGNORE INTO event_run_phase_gates
|
||
(run_id, phase, kind, after_seconds, trigger_id, conditions, needed, entered_at, due_at)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`
|
||
|
||
const GATE_COUNT = `
|
||
UPDATE event_run_phase_gates
|
||
SET satisfied_at = CASE WHEN tally + 1 >= needed THEN ? ELSE NULL END,
|
||
satisfied_by = CASE WHEN tally + 1 >= needed THEN 'condition' ELSE NULL END,
|
||
last_event = ?,
|
||
last_event_at = ?,
|
||
tally = tally + 1
|
||
WHERE id = ? AND satisfied_at IS NULL`
|
||
|
||
const GATE_SATISFY = `
|
||
UPDATE event_run_phase_gates
|
||
SET satisfied_at = ?, satisfied_by = ?, forced_by = ?
|
||
WHERE id = ? AND satisfied_at IS NULL`
|
||
|
||
const GATE_OPEN_FOR_TRIGGER = `
|
||
SELECT g.* FROM event_run_phase_gates g
|
||
JOIN event_runs r ON r.id = g.run_id
|
||
WHERE g.trigger_id = ? AND g.satisfied_at IS NULL
|
||
AND r.status IN ('running','paused')
|
||
AND r.current_phase = g.phase
|
||
LIMIT 200`
|
||
|
||
const MATERIALISE_RUN = `
|
||
INSERT IGNORE INTO event_runs (definition_id, version_id, scope, scheduled_for, concurrency_key)
|
||
VALUES (?, ?, ?, ?, ?)`
|
||
|
||
const MATERIALISE_STEP = `
|
||
INSERT IGNORE INTO event_run_steps (run_id, phase, seq, action_id, idempotency_key)
|
||
VALUES (?, ?, ?, ?, ?)`
|
||
|
||
const FIND_MISSED = `
|
||
SELECT r.id FROM event_runs r
|
||
JOIN event_definitions d ON d.id = r.definition_id
|
||
WHERE r.status = 'scheduled'
|
||
AND r.scheduled_for + INTERVAL d.grace_seconds SECOND < ?`
|
||
|
||
// Verbatim `eventDefinitions.db#findSchedulable`.
|
||
const FIND_SCHEDULABLE = `
|
||
SELECT d.id, d.title, d.slug, d.timezone, d.grace_seconds, d.concurrency_key,
|
||
d.current_version_id, d.series_id, v.spec AS version_spec,
|
||
s.name AS series_name, s.slug AS series_slug
|
||
FROM event_definitions d
|
||
JOIN event_versions v ON v.id = d.current_version_id
|
||
LEFT JOIN event_series s ON s.id = d.series_id
|
||
WHERE d.state = 'ready'
|
||
ORDER BY d.id`
|
||
|
||
// Verbatim `eventRuns.db#listInWindow`, with no optional filter applied.
|
||
const LIST_IN_WINDOW = `
|
||
SELECT r.*, d.title AS definition_title, d.slug AS definition_slug,
|
||
d.series_id AS series_id, se.name AS series_name, se.slug AS series_slug,
|
||
v.version AS version_number,
|
||
(SELECT COUNT(*) FROM event_run_steps s
|
||
WHERE s.run_id = r.id AND s.status = 'running' AND s.claim_expires_at IS NULL) AS waiting_steps
|
||
FROM event_runs r
|
||
JOIN event_definitions d ON d.id = r.definition_id
|
||
JOIN event_versions v ON v.id = r.version_id
|
||
LEFT JOIN event_series se ON se.id = d.series_id
|
||
WHERE r.scheduled_for >= ? AND r.scheduled_for < ?
|
||
ORDER BY r.scheduled_for, r.id
|
||
LIMIT 500`
|
||
|
||
// Verbatim `eventRuns.db#repinScheduled`.
|
||
const REPIN_SCHEDULED = `
|
||
UPDATE event_runs
|
||
SET version_id = ?
|
||
WHERE definition_id = ?
|
||
AND status = 'scheduled'
|
||
AND started_at IS NULL
|
||
AND version_id <> ?`
|
||
|
||
const DB = `rg_events_test_${process.pid}`
|
||
let pool = null
|
||
let available = false
|
||
|
||
const poolOpts = () => ({
|
||
host: process.env.DB_HOST || '127.0.0.1',
|
||
port: Number(process.env.DB_PORT) || 3306,
|
||
user: process.env.DB_USER || 'root',
|
||
password: process.env.DB_PASSWORD || '',
|
||
})
|
||
|
||
before(async () => {
|
||
const admin = mariadb.createPool({
|
||
...poolOpts(),
|
||
connectionLimit: 1,
|
||
connectTimeout: 2000,
|
||
initializationTimeout: 2000,
|
||
multipleStatements: true,
|
||
})
|
||
try {
|
||
await admin.query(`CREATE DATABASE ${DB}`)
|
||
available = true
|
||
} catch {
|
||
available = false
|
||
} finally {
|
||
await admin.end().catch(() => {})
|
||
}
|
||
if (!available) return
|
||
|
||
pool = mariadb.createPool({
|
||
...poolOpts(),
|
||
database: DB,
|
||
connectionLimit: 3,
|
||
multipleStatements: true,
|
||
bigIntAsNumber: true,
|
||
insertIdAsNumber: true,
|
||
})
|
||
await pool.query(SCHEMA)
|
||
})
|
||
|
||
after(async () => {
|
||
if (pool) {
|
||
await pool.query(`DROP DATABASE IF EXISTS ${DB}`).catch(() => {})
|
||
await pool.end().catch(() => {})
|
||
}
|
||
})
|
||
|
||
// Checked INSIDE each test, never as a `{ skip }` option: the option is evaluated
|
||
// when the file is read, which is before `before()` has had a chance to find out
|
||
// whether there is a database. Every test skipped unconditionally is what that
|
||
// mistake looks like, and it looks exactly like a passing suite.
|
||
const SKIP = 'no database reachable - set DB_HOST/DB_PORT/DB_USER/DB_PASSWORD to run'
|
||
const needDb = (t) => {
|
||
if (available) return false
|
||
t.skip(SKIP)
|
||
return true
|
||
}
|
||
|
||
const T0 = new Date('2026-09-02T12:00:00Z')
|
||
const later = (ms) => new Date(T0.getTime() + ms)
|
||
const rows = (r) => Number(r.affectedRows)
|
||
|
||
beforeEach(async () => {
|
||
if (!available) return
|
||
await pool.query('DELETE FROM event_run_phase_gates')
|
||
await pool.query('DELETE FROM event_run_resources')
|
||
await pool.query('DELETE FROM event_run_steps')
|
||
await pool.query('DELETE FROM event_runs')
|
||
await pool.query('DELETE FROM event_definitions')
|
||
await pool.query('DELETE FROM event_versions')
|
||
await pool.query('DELETE FROM event_series')
|
||
})
|
||
|
||
async function seedRun(over = {}) {
|
||
const def = await pool.query('INSERT INTO event_definitions (grace_seconds) VALUES (?)', [
|
||
over.graceSeconds ?? 900,
|
||
])
|
||
const r = await pool.query(
|
||
`INSERT INTO event_runs (definition_id, version_id, scope, status, scheduled_for, concurrency_key,
|
||
claimed_by, claim_expires_at)
|
||
VALUES (?, 1, ?, ?, ?, ?, ?, ?)`,
|
||
[
|
||
def.insertId,
|
||
over.scope ?? '',
|
||
over.status ?? 'scheduled',
|
||
over.scheduledFor ?? T0,
|
||
over.concurrencyKey ?? null,
|
||
over.claimedBy ?? null,
|
||
over.claimExpiresAt ?? null,
|
||
],
|
||
)
|
||
return { runId: r.insertId, definitionId: def.insertId }
|
||
}
|
||
|
||
const seedStep = async (runId, over = {}) =>
|
||
(
|
||
await pool.query(
|
||
`INSERT INTO event_run_steps (run_id, phase, seq, action_id, status, due_at, attempts,
|
||
claimed_by, claim_expires_at, idempotency_key)
|
||
VALUES (?, ?, ?, 'test.noop', ?, ?, ?, ?, ?, ?)`,
|
||
[
|
||
runId,
|
||
over.phase ?? 'main',
|
||
over.seq ?? 0,
|
||
over.status ?? 'pending',
|
||
over.dueAt ?? null,
|
||
over.attempts ?? 0,
|
||
over.claimedBy ?? null,
|
||
over.claimExpiresAt ?? null,
|
||
over.key ?? 'k'.repeat(40),
|
||
],
|
||
)
|
||
).insertId
|
||
|
||
const stepById = async (id) => (await pool.query('SELECT * FROM event_run_steps WHERE id = ?', [id]))[0]
|
||
const runById = async (id) => (await pool.query('SELECT * FROM event_runs WHERE id = ?', [id]))[0]
|
||
|
||
// ── claimStart: exactly one winner ─────────────────────────────────────────
|
||
|
||
test('claimStart: the first caller wins and every other gets zero', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedRun()
|
||
|
||
const first = await pool.query(CLAIM_START, ['host:1', later(60_000), runId])
|
||
const second = await pool.query(CLAIM_START, ['host:2', later(60_000), runId])
|
||
|
||
assert.equal(rows(first), 1, 'the winner is told 1')
|
||
assert.equal(rows(second), 0, 'the loser is told 0, not 1 with foundRows')
|
||
assert.equal((await runById(runId)).claimed_by, 'host:1')
|
||
assert.equal((await runById(runId)).status, 'starting')
|
||
})
|
||
|
||
test('claimStart: started_at is stamped once and never moved', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedRun()
|
||
await pool.query(CLAIM_START, ['host:1', later(60_000), runId])
|
||
const first = (await runById(runId)).started_at
|
||
|
||
await pool.query("UPDATE event_runs SET status = 'scheduled' WHERE id = ?", [runId])
|
||
await pool.query(CLAIM_START, ['host:2', later(60_000), runId])
|
||
|
||
assert.deepEqual((await runById(runId)).started_at, first, 'COALESCE keeps the original instant')
|
||
})
|
||
|
||
// ── claimTick: a live lease refuses even its own holder ────────────────────
|
||
|
||
test('claimTick: a live lease is not re-enterable by the process that took it', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedRun({ status: 'running', claimedBy: 'host:1', claimExpiresAt: later(60_000) })
|
||
|
||
const again = await pool.query(CLAIM_TICK, ['host:1', later(120_000), runId, T0])
|
||
assert.equal(rows(again), 0, 'a tick that overran must not advance its own run twice')
|
||
})
|
||
|
||
test('claimTick: an expired lease is takeable, by anyone', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedRun({ status: 'running', claimedBy: 'host:1', claimExpiresAt: later(-60_000) })
|
||
|
||
const taken = await pool.query(CLAIM_TICK, ['host:2', later(60_000), runId, T0])
|
||
assert.equal(rows(taken), 1)
|
||
assert.equal((await runById(runId)).claimed_by, 'host:2')
|
||
})
|
||
|
||
test('claimTick: a paused run is never claimable', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedRun({ status: 'paused' })
|
||
assert.equal(rows(await pool.query(CLAIM_TICK, ['host:1', later(60_000), runId, T0])), 0)
|
||
})
|
||
|
||
// ── transition: the guard is the whole point ───────────────────────────────
|
||
|
||
test('transition: a run cancelled underneath the tick is not transitioned', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedRun({ status: 'running' })
|
||
await pool.query("UPDATE event_runs SET status = 'cancelled' WHERE id = ?", [runId])
|
||
|
||
const moved = await pool.query(TRANSITION, ['completed', 'main', runId, 'running'])
|
||
assert.equal(rows(moved), 0)
|
||
assert.equal((await runById(runId)).status, 'cancelled')
|
||
})
|
||
|
||
test('transition: running -> running is a guarded write, not a no-op', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedRun({ status: 'running' })
|
||
|
||
// This is how the runner advances `current_phase`, and `foundRows` is exactly
|
||
// what makes it report 1 despite `status` not changing — which is the answer
|
||
// the caller needs, because what it is checking is that the run is STILL
|
||
// running, not that the status moved.
|
||
const moved = await pool.query(TRANSITION, ['running', 'closing', runId, 'running'])
|
||
assert.equal(rows(moved), 1)
|
||
assert.equal((await runById(runId)).current_phase, 'closing')
|
||
})
|
||
|
||
// ── The step claim ─────────────────────────────────────────────────────────
|
||
|
||
test('the step claim: one winner, and attempts is incremented by the claim alone', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedRun({ status: 'running' })
|
||
const stepId = await seedStep(runId)
|
||
|
||
assert.equal(rows(await pool.query(CLAIM_STEP, ['host:1', later(60_000), stepId, T0])), 1)
|
||
assert.equal(rows(await pool.query(CLAIM_STEP, ['host:2', later(60_000), stepId, T0])), 0)
|
||
assert.equal((await stepById(stepId)).attempts, 1, 'one claim, one attempt')
|
||
})
|
||
|
||
test('the step claim: a step held behind a core.wait is not due', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedRun({ status: 'running' })
|
||
const stepId = await seedStep(runId, { dueAt: later(300_000) })
|
||
|
||
assert.equal(rows(await pool.query(CLAIM_STEP, ['host:1', later(60_000), stepId, T0])), 0)
|
||
assert.equal(rows(await pool.query(CLAIM_STEP, ['host:1', later(360_000), stepId, later(301_000)])), 1)
|
||
})
|
||
|
||
// ── The reclaim: order, and what it must not touch ─────────────────────────
|
||
|
||
test('the reclaim hands a stale step back WITHOUT resetting attempts', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedRun({ status: 'running' })
|
||
const stepId = await seedStep(runId, { status: 'running', attempts: 2, claimExpiresAt: later(-1000) })
|
||
|
||
await pool.query(STEP_GIVE_UP, [T0, 3])
|
||
await pool.query(STEP_RECLAIM, [T0])
|
||
|
||
const step = await stepById(stepId)
|
||
assert.equal(step.status, 'pending')
|
||
assert.equal(step.attempts, 2, 'Engagement Phase 14: a reclaim that reset this made MAX_ATTEMPTS unreachable')
|
||
})
|
||
|
||
test('the reclaim gives up FIRST, so a spent step leaves running as failed', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedRun({ status: 'running' })
|
||
const stepId = await seedStep(runId, { status: 'running', attempts: 3, claimExpiresAt: later(-1000) })
|
||
|
||
const gaveUp = await pool.query(STEP_GIVE_UP, [T0, 3])
|
||
const reclaimed = await pool.query(STEP_RECLAIM, [T0])
|
||
|
||
assert.equal(rows(gaveUp), 1)
|
||
assert.equal(rows(reclaimed), 0, 'reversing these two makes the row retry forever')
|
||
assert.equal((await stepById(stepId)).status, 'failed')
|
||
assert.equal((await stepById(stepId)).attempts, 3)
|
||
})
|
||
|
||
test('the reclaim leaves a PARKED step alone, however long it waits', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedRun({ status: 'running' })
|
||
const stepId = await seedStep(runId, { status: 'running', attempts: 1, claimExpiresAt: null })
|
||
|
||
await pool.query(STEP_GIVE_UP, [later(365 * 24 * 3600 * 1000), 3])
|
||
await pool.query(STEP_RECLAIM, [later(365 * 24 * 3600 * 1000)])
|
||
|
||
const step = await stepById(stepId)
|
||
assert.equal(step.status, 'running', 'a NULL lease is a parked cue, not staleness')
|
||
assert.equal(step.attempts, 1)
|
||
})
|
||
|
||
// ── holdNext ───────────────────────────────────────────────────────────────
|
||
|
||
test('holdNext moves the next PENDING step of the phase, and only one', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedRun({ status: 'running' })
|
||
await seedStep(runId, { seq: 0, status: 'done' })
|
||
const second = await seedStep(runId, { seq: 1 })
|
||
const third = await seedStep(runId, { seq: 2 })
|
||
|
||
const moved = await pool.query(HOLD_NEXT, [later(300_000), runId, 'main', 0, later(300_000)])
|
||
assert.equal(rows(moved), 1)
|
||
assert.deepEqual((await stepById(second)).due_at, later(300_000))
|
||
assert.equal((await stepById(third)).due_at, null, 'a wait holds the next step, not the rest of the phase')
|
||
})
|
||
|
||
test('holdNext never pulls a due date earlier, so a re-dispatch cannot double the wait', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedRun({ status: 'running' })
|
||
await seedStep(runId, { seq: 0, status: 'done' })
|
||
const second = await seedStep(runId, { seq: 1, dueAt: later(600_000) })
|
||
|
||
const moved = await pool.query(HOLD_NEXT, [later(300_000), runId, 'main', 0, later(300_000)])
|
||
assert.equal(rows(moved), 0)
|
||
assert.deepEqual((await stepById(second)).due_at, later(600_000))
|
||
})
|
||
|
||
test('holdNext skips a step that is already running', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedRun({ status: 'running' })
|
||
await seedStep(runId, { seq: 0, status: 'done' })
|
||
await seedStep(runId, { seq: 1, status: 'running' })
|
||
const third = await seedStep(runId, { seq: 2 })
|
||
|
||
await pool.query(HOLD_NEXT, [later(300_000), runId, 'main', 0, later(300_000)])
|
||
assert.deepEqual((await stepById(third)).due_at, later(300_000))
|
||
})
|
||
|
||
// ── The two unique indexes that carry the weight ───────────────────────────
|
||
|
||
test('uq_evrun_occurrence, not the claim, is what stops two runs of one occurrence', async (t) => {
|
||
if (needDb(t)) return
|
||
const def = await pool.query('INSERT INTO event_definitions (grace_seconds) VALUES (900)')
|
||
|
||
const a = await pool.query(MATERIALISE_RUN, [def.insertId, 1, '', T0, null])
|
||
const b = await pool.query(MATERIALISE_RUN, [def.insertId, 1, '', T0, null])
|
||
|
||
assert.equal(rows(a), 1)
|
||
assert.equal(rows(b), 0, 'INSERT IGNORE answers honestly rather than raising a 1062')
|
||
const all = await pool.query('SELECT COUNT(*) AS n FROM event_runs')
|
||
assert.equal(Number(all[0].n), 1)
|
||
})
|
||
|
||
test("scope '' rather than NULL is what makes that index work at all", async (t) => {
|
||
if (needDb(t)) return
|
||
const def = await pool.query('INSERT INTO event_definitions (grace_seconds) VALUES (900)')
|
||
|
||
// The empty-string case collides, as it must. A NULL scope would NOT: multiple
|
||
// NULLs do not collide in MariaDB, which would silently permit two runs of one
|
||
// occurrence — the reason the column is NOT NULL DEFAULT ''.
|
||
await pool.query(MATERIALISE_RUN, [def.insertId, 1, '', T0, null])
|
||
assert.equal(rows(await pool.query(MATERIALISE_RUN, [def.insertId, 1, '', T0, null])), 0)
|
||
|
||
// Two different scopes are two different occurrences, which is what lets a
|
||
// worldwide event fan out across servers without colliding with itself.
|
||
assert.equal(rows(await pool.query(MATERIALISE_RUN, [def.insertId, 1, 'atlantic', T0, null])), 1)
|
||
})
|
||
|
||
test('uq_evstep_slot makes re-materialising a phase a no-op', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedRun({ status: 'running' })
|
||
|
||
assert.equal(rows(await pool.query(MATERIALISE_STEP, [runId, 'main', 0, 'test.noop', 'a'.repeat(40)])), 1)
|
||
assert.equal(rows(await pool.query(MATERIALISE_STEP, [runId, 'main', 0, 'test.noop', 'b'.repeat(40)])), 0)
|
||
|
||
const [step] = await pool.query('SELECT idempotency_key FROM event_run_steps WHERE run_id = ?', [runId])
|
||
assert.equal(step.idempotency_key, 'a'.repeat(40), 'a re-materialise cannot overwrite a key a dispatch already sent')
|
||
})
|
||
|
||
// ── The grace window is per definition ─────────────────────────────────────
|
||
|
||
test('findMissed compares against each definition’s own grace window', async (t) => {
|
||
if (needDb(t)) return
|
||
const tight = await seedRun({ graceSeconds: 600 })
|
||
const generous = await seedRun({ graceSeconds: 3600, scope: 'b' })
|
||
|
||
const missed = (await pool.query(FIND_MISSED, [later(30 * 60 * 1000)])).map((r) => Number(r.id))
|
||
assert.deepEqual(missed, [tight.runId])
|
||
assert.ok(!missed.includes(generous.runId), 'inside its own window a run starts late rather than being missed')
|
||
})
|
||
|
||
// -- Phase 3: the controls a human presses ----------------------------------
|
||
|
||
test('confirm resolves a parked cue and cannot touch a step being dispatched', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedRun({ status: 'running' })
|
||
const parked = await seedStep(runId, { seq: 0, status: 'running', claimedBy: 'host:1', claimExpiresAt: null })
|
||
const busy = await seedStep(runId, { seq: 1, status: 'running', claimedBy: 'host:1', claimExpiresAt: later(60_000), key: 'x'.repeat(40) })
|
||
|
||
assert.equal(rows(await pool.query(CONFIRM_PARKED, ['gate opened', parked])), 1)
|
||
assert.equal(rows(await pool.query(CONFIRM_PARKED, ['nope', busy])), 0, 'a live lease is a step somebody owns')
|
||
|
||
assert.equal((await stepById(parked)).status, 'done')
|
||
assert.equal((await stepById(parked)).last_error, 'gate opened')
|
||
assert.equal((await stepById(busy)).status, 'running')
|
||
})
|
||
|
||
test('a confirm of an already-confirmed cue reports 0, not 1', async (t) => {
|
||
if (needDb(t)) return
|
||
// The engagement Phase 4a shape: a connector that defaults `foundRows: true`
|
||
// reports 1 for an UPDATE that matched and changed nothing, and a control that
|
||
// read that as success would tell a second staff member their press worked.
|
||
const { runId } = await seedRun({ status: 'running' })
|
||
const parked = await seedStep(runId, { status: 'running', claimExpiresAt: null })
|
||
|
||
assert.equal(rows(await pool.query(CONFIRM_PARKED, [null, parked])), 1)
|
||
assert.equal(rows(await pool.query(CONFIRM_PARKED, [null, parked])), 0)
|
||
})
|
||
|
||
test('skip takes a pending step and a parked cue, and refuses a leased one', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedRun({ status: 'running' })
|
||
const pending = await seedStep(runId, { seq: 0, status: 'pending' })
|
||
const parked = await seedStep(runId, { seq: 1, status: 'running', claimExpiresAt: null, key: 'y'.repeat(40) })
|
||
const busy = await seedStep(runId, { seq: 2, status: 'running', claimExpiresAt: later(60_000), key: 'z'.repeat(40) })
|
||
const failed = await seedStep(runId, { seq: 3, status: 'failed', key: 'w'.repeat(40) })
|
||
|
||
assert.equal(rows(await pool.query(SKIP_BY_HUMAN, [null, pending])), 1)
|
||
assert.equal(rows(await pool.query(SKIP_BY_HUMAN, [null, parked])), 1)
|
||
assert.equal(rows(await pool.query(SKIP_BY_HUMAN, [null, busy])), 0)
|
||
assert.equal(rows(await pool.query(SKIP_BY_HUMAN, [null, failed])), 0, 'a failed step is terminal; resume carries the phase past it')
|
||
})
|
||
|
||
test('cancelOpen closes pending steps and parked cues, and leaves a leased one alone', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedRun({ status: 'running' })
|
||
const done = await seedStep(runId, { seq: 0, status: 'done' })
|
||
const busy = await seedStep(runId, { seq: 1, status: 'running', claimExpiresAt: later(60_000), key: 'p'.repeat(40) })
|
||
const parked = await seedStep(runId, { seq: 2, status: 'running', claimExpiresAt: null, key: 'q'.repeat(40) })
|
||
const pending = await seedStep(runId, { seq: 3, status: 'pending', key: 'r'.repeat(40) })
|
||
|
||
assert.equal(rows(await pool.query(CANCEL_OPEN, [runId])), 2)
|
||
|
||
assert.equal((await stepById(done)).status, 'done')
|
||
assert.equal((await stepById(busy)).status, 'running', 'nothing can recall a command already sent')
|
||
assert.equal((await stepById(parked)).status, 'cancelled', 'a cancelled run must stop claiming to wait on somebody')
|
||
assert.equal((await stepById(pending)).status, 'cancelled')
|
||
})
|
||
|
||
test('requeue only takes a failed step, and puts attempts back to zero', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedRun({ status: 'paused' })
|
||
const failed = await seedStep(runId, { seq: 0, status: 'failed', attempts: 3 })
|
||
const pending = await seedStep(runId, { seq: 1, status: 'pending', key: 's'.repeat(40) })
|
||
|
||
assert.equal(rows(await pool.query(REQUEUE, [failed])), 1)
|
||
assert.equal(rows(await pool.query(REQUEUE, [pending])), 0)
|
||
|
||
const row = await stepById(failed)
|
||
assert.equal(row.status, 'pending')
|
||
assert.equal(Number(row.attempts), 0)
|
||
assert.equal(row.due_at, null, 'a re-queued step is due now, not at the retry backoff it was left on')
|
||
})
|
||
|
||
test('lastStartedSeq names the furthest step of the phase, not the earliest unsettled one', async (t) => {
|
||
if (needDb(t)) return
|
||
// The defect this replaced: a phase that carried on past a failed step (an
|
||
// `on_failure` of `skip`) and then paused at a later one. "The lowest seq that
|
||
// is not settled" answers with the FIRST failure - a step the runner has long
|
||
// since stepped over - and retry would re-queue a row behind its own cursor.
|
||
const { runId } = await seedRun({ status: 'paused' })
|
||
await seedStep(runId, { seq: 0, status: 'failed', key: 'a'.repeat(40) })
|
||
await seedStep(runId, { seq: 1, status: 'done', key: 'b'.repeat(40) })
|
||
await seedStep(runId, { seq: 2, status: 'failed', key: 'c'.repeat(40) })
|
||
await seedStep(runId, { seq: 3, status: 'pending', key: 'd'.repeat(40) })
|
||
|
||
const [row] = await pool.query(LAST_STARTED_SEQ, [runId, 'main'])
|
||
assert.equal(Number(row.seq), 2)
|
||
})
|
||
|
||
test('lastStartedSeq is NULL for a phase nothing has touched', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedRun({ status: 'running' })
|
||
await seedStep(runId, { seq: 0, status: 'pending' })
|
||
|
||
const [row] = await pool.query(LAST_STARTED_SEQ, [runId, 'main'])
|
||
assert.equal(row.seq, null, 'a null must read as "nothing to retry", not as seq 0')
|
||
})
|
||
|
||
test('a guarded transition refuses a run that was cancelled underneath it', async (t) => {
|
||
if (needDb(t)) return
|
||
// What the admin cancel looks like from the runner's side, mid-tick: the
|
||
// guarded write returns 0 and the tick treats the run as taken rather than
|
||
// advancing a run somebody has just stopped.
|
||
const { runId } = await seedRun({ status: 'running' })
|
||
await pool.query("UPDATE event_runs SET status = 'cancelled' WHERE id = ?", [runId])
|
||
|
||
assert.equal(rows(await pool.query(TRANSITION, ['running', 'two', runId, 'running'])), 0)
|
||
assert.equal((await runById(runId)).status, 'cancelled')
|
||
})
|
||
|
||
|
||
// ── Phase 4: the two reads ────────────────────────────────────────
|
||
|
||
/** A definition with a published version, and optionally a series. */
|
||
async function seedDefinition({ state = 'ready', spec = { schedule: { kind: 'manual' } }, series = null } = {}) {
|
||
let seriesId = null
|
||
if (series) {
|
||
const s = await pool.query('INSERT INTO event_series (name, slug) VALUES (?, ?)', [series, series])
|
||
seriesId = s.insertId
|
||
}
|
||
const d = await pool.query(
|
||
'INSERT INTO event_definitions (state, series_id, timezone) VALUES (?, ?, ?)',
|
||
[state, seriesId, 'Europe/Berlin'],
|
||
)
|
||
const v = await pool.query(
|
||
'INSERT INTO event_versions (definition_id, version, spec) VALUES (?, 1, ?)',
|
||
[d.insertId, JSON.stringify(spec)],
|
||
)
|
||
await pool.query('UPDATE event_definitions SET current_version_id = ? WHERE id = ?', [
|
||
v.insertId,
|
||
d.insertId,
|
||
])
|
||
return { definitionId: d.insertId, versionId: v.insertId, seriesId }
|
||
}
|
||
|
||
test('findSchedulable returns ready definitions with their PUBLISHED spec, series or not', async (t) => {
|
||
if (needDb(t)) return
|
||
const withSeries = await seedDefinition({
|
||
spec: { schedule: { kind: 'weekly', days: ['friday'], time: '20:00' } },
|
||
series: 'royal-spy',
|
||
})
|
||
const withoutSeries = await seedDefinition({ spec: { schedule: { kind: 'manual' } } })
|
||
|
||
const found = await pool.query(FIND_SCHEDULABLE)
|
||
const ids = found.map((r) => r.id).sort((a, b) => a - b)
|
||
assert.deepEqual(ids, [withSeries.definitionId, withoutSeries.definitionId].sort((a, b) => a - b))
|
||
|
||
// The LEFT JOIN must not drop the definition that belongs to no arc — an
|
||
// inner join here would make every event outside a series unschedulable, and
|
||
// most events are outside one.
|
||
const plain = found.find((r) => r.id === withoutSeries.definitionId)
|
||
assert.equal(plain.series_name, null)
|
||
|
||
const arced = found.find((r) => r.id === withSeries.definitionId)
|
||
assert.equal(arced.series_name, 'royal-spy')
|
||
assert.equal(arced.timezone, 'Europe/Berlin')
|
||
|
||
// The spec really came back, and really came back parseable.
|
||
const spec = typeof arced.version_spec === 'string' ? JSON.parse(arced.version_spec) : arced.version_spec
|
||
assert.equal(spec.schedule.kind, 'weekly')
|
||
})
|
||
|
||
test('findSchedulable skips a draft, an archived one, and one with no published version', async (t) => {
|
||
if (needDb(t)) return
|
||
await seedDefinition({ state: 'draft' })
|
||
await seedDefinition({ state: 'archived' })
|
||
// `ready` with a dangling version pointer: the JOIN is what must drop it, and
|
||
// a definition whose version row went missing must not become a runner crash.
|
||
const orphan = await seedDefinition({ state: 'ready' })
|
||
await pool.query('DELETE FROM event_versions WHERE id = ?', [orphan.versionId])
|
||
|
||
assert.equal((await pool.query(FIND_SCHEDULABLE)).length, 0)
|
||
})
|
||
|
||
test('listInWindow is half-open on the window, and counts only PARKED steps as waiting', async (t) => {
|
||
if (needDb(t)) return
|
||
const def = await seedDefinition()
|
||
const at = async (when) => {
|
||
const r = await pool.query(
|
||
'INSERT INTO event_runs (definition_id, version_id, scope, scheduled_for) VALUES (?, ?, ?, ?)',
|
||
[def.definitionId, def.versionId, '', when],
|
||
)
|
||
return r.insertId
|
||
}
|
||
const before = await at(new Date('2026-09-01T00:00:00Z'))
|
||
const onFrom = await at(new Date('2026-09-02T00:00:00Z'))
|
||
const inside = await at(new Date('2026-09-05T00:00:00Z'))
|
||
const onTo = await at(new Date('2026-09-09T00:00:00Z'))
|
||
|
||
// `>= from AND < to` — the instant ON the upper bound belongs to the NEXT
|
||
// window. A closed interval would draw the last day of one month and the first
|
||
// of the next as the same occurrence twice.
|
||
const found = await pool.query(LIST_IN_WINDOW, [
|
||
new Date('2026-09-02T00:00:00Z'),
|
||
new Date('2026-09-09T00:00:00Z'),
|
||
])
|
||
assert.deepEqual(found.map((r) => r.id), [onFrom, inside])
|
||
assert.ok(!found.some((r) => r.id === before || r.id === onTo))
|
||
|
||
// A parked step is `running` with a NULL lease; a leased one is the runner
|
||
// mid-dispatch and is not waiting on anybody.
|
||
await seedStep(inside, { status: 'running', claimExpiresAt: null, key: 'a'.repeat(40) })
|
||
await seedStep(inside, {
|
||
seq: 1,
|
||
status: 'running',
|
||
claimedBy: 'host',
|
||
claimExpiresAt: later(60_000),
|
||
key: 'b'.repeat(40),
|
||
})
|
||
const again = await pool.query(LIST_IN_WINDOW, [
|
||
new Date('2026-09-02T00:00:00Z'),
|
||
new Date('2026-09-09T00:00:00Z'),
|
||
])
|
||
assert.equal(Number(again.find((r) => r.id === inside).waiting_steps), 1)
|
||
})
|
||
|
||
|
||
test('repinScheduled moves the occurrences that have not begun, and only those', async (t) => {
|
||
if (needDb(t)) return
|
||
// The case: an editor fixes a typo on a weekly event on Wednesday. Two Fridays
|
||
// are already materialised on v3, last Friday's run is finished, and one is in
|
||
// flight right now.
|
||
const def = await pool.query('INSERT INTO event_definitions (grace_seconds) VALUES (900)')
|
||
const mk = async (status, versionId, startedAt, when) =>
|
||
(
|
||
await pool.query(
|
||
`INSERT INTO event_runs (definition_id, version_id, scope, status, scheduled_for, started_at)
|
||
VALUES (?, ?, ?, ?, ?, ?)`,
|
||
[def.insertId, versionId, `s${when}`, status, T0, startedAt],
|
||
)
|
||
).insertId
|
||
|
||
const ahead1 = await mk('scheduled', 3, null, 1)
|
||
const ahead2 = await mk('scheduled', 3, null, 2)
|
||
const running = await mk('running', 3, T0, 3)
|
||
const done = await mk('completed', 3, T0, 4)
|
||
// Already on the new version: excluded by `version_id <> ?`, so a second
|
||
// publish of an unchanged definition is not a fleet of pointless writes.
|
||
const already = await mk('scheduled', 4, null, 5)
|
||
|
||
const moved = rows(await pool.query(REPIN_SCHEDULED, [4, def.insertId, 4]))
|
||
assert.equal(moved, 2)
|
||
|
||
const versionOf = async (id) =>
|
||
Number((await pool.query('SELECT version_id FROM event_runs WHERE id = ?', [id]))[0].version_id)
|
||
assert.equal(await versionOf(ahead1), 4)
|
||
assert.equal(await versionOf(ahead2), 4)
|
||
// A run that has begun keeps the version it pinned, for ever: that pin is what
|
||
// makes it explicable afterwards.
|
||
assert.equal(await versionOf(running), 3)
|
||
assert.equal(await versionOf(done), 3)
|
||
assert.equal(await versionOf(already), 4)
|
||
})
|
||
|
||
test('a scheduled run whose started_at is somehow set is left alone', async (t) => {
|
||
if (needDb(t)) return
|
||
// Belt and braces on the guard: `status = 'scheduled'` and `started_at IS NULL`
|
||
// are two conditions rather than one because a row that has both is the only
|
||
// row that is provably untouched.
|
||
const def = await pool.query('INSERT INTO event_definitions (grace_seconds) VALUES (900)')
|
||
const r = await pool.query(
|
||
`INSERT INTO event_runs (definition_id, version_id, scope, status, scheduled_for, started_at)
|
||
VALUES (?, 3, '', 'scheduled', ?, ?)`,
|
||
[def.insertId, T0, T0],
|
||
)
|
||
assert.equal(rows(await pool.query(REPIN_SCHEDULED, [4, def.insertId, 4])), 0)
|
||
const after = (await pool.query('SELECT version_id FROM event_runs WHERE id = ?', [r.insertId]))[0]
|
||
assert.equal(Number(after.version_id), 3)
|
||
})
|
||
|
||
|
||
// ── Phase 5: the gate statements ───────────────────────────────────────────
|
||
|
||
/** A run in `running` on one phase, plus an open gate for it. */
|
||
async function seedGate({ kind = 'on', needed = 1, trigger = 'uo.champ.boss_up', status = 'running', phase = 'boss' } = {}) {
|
||
const { runId } = await seedRun({ status })
|
||
await pool.query('UPDATE event_runs SET current_phase = ? WHERE id = ?', [phase, runId])
|
||
const g = await pool.query(GATE_OPEN, [
|
||
runId,
|
||
phase,
|
||
kind,
|
||
kind === 'after' ? 1800 : null,
|
||
kind === 'on' ? trigger : null,
|
||
kind === 'on' ? JSON.stringify({ variable: 'region', cmp: 'eq', value: 'Yew' }) : null,
|
||
needed,
|
||
T0,
|
||
kind === 'after' ? later(1800_000) : null,
|
||
])
|
||
return { runId, gateId: Number(g.insertId) }
|
||
}
|
||
|
||
const gateById = async (id) => (await pool.query('SELECT * FROM event_run_phase_gates WHERE id = ?', [id]))[0]
|
||
|
||
test('two firings arriving together each count, and exactly one crosses the threshold', async (t) => {
|
||
if (needDb(t)) return
|
||
const { gateId } = await seedGate({ needed: 2 })
|
||
|
||
// The whole reason the CASE is inside the UPDATE: a read-then-write would let
|
||
// both of these see "1 of 2" and neither satisfy, or both satisfy and advance
|
||
// one phase twice.
|
||
const [a, b] = await Promise.all([
|
||
pool.query(GATE_COUNT, [T0, null, T0, gateId]),
|
||
pool.query(GATE_COUNT, [T0, null, T0, gateId]),
|
||
])
|
||
assert.equal(rows(a) + rows(b), 2, 'both increments land')
|
||
|
||
const gate = await gateById(gateId)
|
||
assert.equal(Number(gate.tally), 2)
|
||
assert.ok(gate.satisfied_at, 'and the second one closed it')
|
||
assert.equal(gate.satisfied_by, 'condition')
|
||
})
|
||
|
||
test('the threshold is read BEFORE the increment, not after it', async (t) => {
|
||
if (needDb(t)) return
|
||
// The defect this file was written to catch, and did: MariaDB evaluates SET
|
||
// assignments left to right with the values already assigned, so an increment
|
||
// placed FIRST makes the CASE after it read `new + 1 >= needed` and close a
|
||
// two-firing gate on the first firing. Every stub of this agrees with the
|
||
// intent rather than with the server, which is exactly why it survived one.
|
||
const { gateId } = await seedGate({ needed: 2 })
|
||
await pool.query(GATE_COUNT, [T0, null, T0, gateId])
|
||
const half = await gateById(gateId)
|
||
assert.equal(Number(half.tally), 1)
|
||
assert.equal(half.satisfied_at, null, 'one of two is not two')
|
||
|
||
await pool.query(GATE_COUNT, [T0, null, T0, gateId])
|
||
assert.ok((await gateById(gateId)).satisfied_at, 'and the second one is')
|
||
})
|
||
|
||
test('an increment against a closed gate is refused, not applied', async (t) => {
|
||
if (needDb(t)) return
|
||
const { gateId } = await seedGate({ needed: 1 })
|
||
assert.equal(rows(await pool.query(GATE_COUNT, [T0, null, T0, gateId])), 1)
|
||
|
||
// `WHERE satisfied_at IS NULL` — and 0 rather than 1, which is the whole
|
||
// reason this file exists: `foundRows: true` would answer 1 for a no-op.
|
||
assert.equal(rows(await pool.query(GATE_COUNT, [T0, null, T0, gateId])), 0)
|
||
assert.equal(Number((await gateById(gateId)).tally), 1, 'the tally does not keep climbing after the phase moved on')
|
||
})
|
||
|
||
test('a force and a deadline race the same guard, and only one of them wins', async (t) => {
|
||
if (needDb(t)) return
|
||
const { gateId } = await seedGate({ kind: 'after' })
|
||
const [forced, elapsed] = await Promise.all([
|
||
pool.query(GATE_SATISFY, [T0, 'forced', 7, gateId]),
|
||
pool.query(GATE_SATISFY, [T0, 'elapsed', null, gateId]),
|
||
])
|
||
assert.equal(rows(forced) + rows(elapsed), 1, 'exactly one of the two writes')
|
||
|
||
const gate = await gateById(gateId)
|
||
assert.ok(['forced', 'elapsed'].includes(gate.satisfied_by))
|
||
// Whichever won is the one in the row, and the log records that one — never
|
||
// both.
|
||
assert.equal(gate.satisfied_by === 'forced' ? Number(gate.forced_by) : gate.forced_by, gate.satisfied_by === 'forced' ? 7 : null)
|
||
})
|
||
|
||
test('opening a gate twice writes one row', async (t) => {
|
||
if (needDb(t)) return
|
||
const { runId } = await seedGate()
|
||
// The INSERT IGNORE that makes recovery from a process which died between
|
||
// entering a phase and opening its gate uneventful.
|
||
const again = await pool.query(GATE_OPEN, [runId, 'boss', 'on', null, 'uo.champ.boss_up', null, 1, later(60_000), null])
|
||
assert.equal(rows(again), 0)
|
||
const all = await pool.query('SELECT * FROM event_run_phase_gates WHERE run_id = ?', [runId])
|
||
assert.equal(all.length, 1)
|
||
assert.equal(new Date(all[0].entered_at).getTime(), T0.getTime(), 'and the first entry time is the one that stands')
|
||
})
|
||
|
||
test('the emit path sees only gates whose run is live and on that phase', async (t) => {
|
||
if (needDb(t)) return
|
||
const live = await seedGate()
|
||
const paused = await seedGate({ status: 'paused' })
|
||
const cancelled = await seedGate({ status: 'scheduled' })
|
||
await pool.query('UPDATE event_runs SET status = ? WHERE id = ?', ['cancelled', cancelled.runId])
|
||
|
||
const moved = await seedGate()
|
||
await pool.query('UPDATE event_runs SET current_phase = ? WHERE id = ?', ['cleanup', moved.runId])
|
||
|
||
const closed = await seedGate()
|
||
await pool.query(GATE_SATISFY, [T0, 'forced', null, closed.gateId])
|
||
|
||
const found = (await pool.query(GATE_OPEN_FOR_TRIGGER, ['uo.champ.boss_up'])).map((g) => Number(g.id))
|
||
assert.deepEqual(found.sort((a, b) => a - b), [live.gateId, paused.gateId].sort((a, b) => a - b))
|
||
// `paused` counts: the world does not stop because an operator paused the
|
||
// console, and discarding firings during a pause would make pause destructive.
|
||
assert.ok(found.includes(paused.gateId))
|
||
})
|
||
|
||
test('a gate goes with its run', async (t) => {
|
||
if (needDb(t)) return
|
||
// ON DELETE CASCADE, which is what keeps an orphaned gate from outliving the
|
||
// run whose phase it was about.
|
||
const { runId, gateId } = await seedGate()
|
||
await pool.query('DELETE FROM event_run_steps WHERE run_id = ?', [runId])
|
||
await pool.query('DELETE FROM event_runs WHERE id = ?', [runId])
|
||
assert.equal((await pool.query('SELECT * FROM event_run_phase_gates WHERE id = ?', [gateId])).length, 0)
|
||
})
|
||
|
||
// ── Phase 6: the cap's conditional increment ───────────────────────────────
|
||
//
|
||
// **The plan's own acceptance criterion**: *"two concurrent steps against one cap
|
||
// proving neither over-spends"*. It belongs here rather than beside the stub for
|
||
// the reason the gate's increment does — the guard lives in a WHERE that the
|
||
// server evaluates against the pre-update row, and a stub reproduces the reading
|
||
// rather than the server. Engagement's cooldown claim was green against its stub
|
||
// and always allowed the send, because the connector defaults `foundRows: true`
|
||
// and a no-op UPDATE reports 1: exactly the mistake this shape of statement
|
||
// invites, and the reason `spend()` reads `affectedRows` at all.
|
||
//
|
||
// The SET list has ONE assignment, and that is Phase 5's lesson applied rather
|
||
// than rediscovered: MariaDB evaluates SET assignments left to right with the
|
||
// values already assigned, so nothing in this statement may read `consumed`
|
||
// after writing it. The guard stays in the WHERE.
|
||
|
||
// **Requiring a shipping model into THIS file starts a second pool**, and it is
|
||
// the only file in the suite where that matters. Everywhere else the environment
|
||
// points at a dead port and `utils/db`'s pool never connects; here it points at a
|
||
// live server, so the pool holds open connections and the process never exits —
|
||
// 49 green tests and a file that hangs until the harness kills it. Every other
|
||
// event test file already closes it in an `after`; this one now has a reason to.
|
||
//
|
||
// **And that second pool has to be aimed at the throwaway database**, which it
|
||
// was not before Phase 8 noticed. `utils/db` builds its pool at REQUIRE time from
|
||
// `DB_NAME`, and its own `dotenv.config()` reads `server/.env` — so a model test
|
||
// run on a developer's machine was reaching that developer's real schema while
|
||
// the fixtures it was asserting against were being written next door. It passed
|
||
// only because the two tables happened to exist in both. `dotenv` does not
|
||
// overwrite a variable that already exists, so setting it here, before the
|
||
// require below, is what beats the file. This file drops its database in `after`,
|
||
// so aiming the model pool at it is also what keeps the whole run disposable.
|
||
process.env.DB_NAME = DB
|
||
|
||
const budgetDb = require('../src/model/events/eventRunBudget.db')
|
||
const resourcesDb = require('../src/model/events/eventRunResources.db')
|
||
const runsDb = require('../src/model/events/eventRuns.db')
|
||
const appDb = require('../src/utils/db')
|
||
after(() => appDb.close())
|
||
|
||
const seedBudget = async (over = {}) => {
|
||
const { runId } = await seedRun({ status: 'running' })
|
||
await pool.query(
|
||
'INSERT INTO event_run_budget (run_id, dimension, consumed, cap, effective_from) VALUES (?, ?, ?, ?, ?)',
|
||
// `'cap' in over`, not `over.cap ?? 30` -- `null` is a MEANINGFUL cap here
|
||
// (uncapped) and `??` coalesces it straight back to 30, which made the
|
||
// uncapped test seed a capped row and fail against the correct statement.
|
||
[
|
||
runId,
|
||
over.dimension ?? 'uo.creatures',
|
||
over.consumed ?? 0,
|
||
'cap' in over ? over.cap : 30,
|
||
over.from ?? 'uo.creature.spawn',
|
||
],
|
||
)
|
||
return runId
|
||
}
|
||
|
||
const consumedOf = async (runId, dimension = 'uo.creatures') =>
|
||
Number(
|
||
(
|
||
await pool.query('SELECT consumed FROM event_run_budget WHERE run_id = ? AND dimension = ?', [
|
||
runId,
|
||
dimension,
|
||
])
|
||
)[0].consumed,
|
||
)
|
||
|
||
// The statement, lifted verbatim from `eventRunBudget.db.js`. Run through this
|
||
// file's pool rather than through the module, because the module builds its own
|
||
// pool from the environment while this pool points at the throwaway database.
|
||
const SPEND = `
|
||
UPDATE event_run_budget
|
||
SET consumed = consumed + ?
|
||
WHERE run_id = ? AND dimension = ? AND (cap IS NULL OR consumed + ? <= cap)`
|
||
|
||
const spend = async (runId, amount, dimension = 'uo.creatures') =>
|
||
Number((await pool.query(SPEND, [amount, runId, dimension, amount])).affectedRows) > 0
|
||
|
||
test('two steps spending one cap at once: the second is refused, not queued behind the first', async (t) => {
|
||
if (needDb(t)) return
|
||
// 28 of 30 spent, and two steps each wanting 5 arrive together. A
|
||
// read-then-write would let both see 28 and both spend, ending at 38 of 30 —
|
||
// the exact over-spend the conditional increment exists to make impossible.
|
||
const runId = await seedBudget({ consumed: 28, cap: 30 })
|
||
const [a, b] = await Promise.all([spend(runId, 5), spend(runId, 5)])
|
||
assert.equal(a, false)
|
||
assert.equal(b, false)
|
||
assert.equal(await consumedOf(runId), 28)
|
||
})
|
||
|
||
test('two steps that BOTH fit both spend, and the total is exact', async (t) => {
|
||
if (needDb(t)) return
|
||
// The other half, and the one a too-strict guard would break: a cap is not a
|
||
// lock. Three steps of 5 against 30 must all succeed and land on 15, or the
|
||
// statement is refusing legal work.
|
||
const runId = await seedBudget({ consumed: 0, cap: 30 })
|
||
const results = await Promise.all([spend(runId, 5), spend(runId, 5), spend(runId, 5)])
|
||
assert.deepEqual(results, [true, true, true])
|
||
assert.equal(await consumedOf(runId), 15)
|
||
})
|
||
|
||
test('a spend that exactly reaches the cap is allowed; one over it is not', async (t) => {
|
||
if (needDb(t)) return
|
||
// `<=`, not `<`. A cap of 30 means thirty creatures are permitted, and an
|
||
// off-by-one here is a deployment that can never use the last unit of anything
|
||
// it configured.
|
||
const runId = await seedBudget({ consumed: 25, cap: 30 })
|
||
assert.equal(await spend(runId, 5), true)
|
||
assert.equal(await consumedOf(runId), 30)
|
||
assert.equal(await spend(runId, 1), false)
|
||
assert.equal(await consumedOf(runId), 30)
|
||
})
|
||
|
||
test('a NULL cap is uncapped, and still counts', async (t) => {
|
||
if (needDb(t)) return
|
||
// The row exists so the meter has something to show; nothing bounds it. The
|
||
// distinction matters because a MISSING row is a refusal — a step spending a
|
||
// dimension its own run's version never priced — and the two must not collapse
|
||
// into one behaviour.
|
||
const runId = await seedBudget({ consumed: 0, cap: null })
|
||
assert.equal(await spend(runId, 1000000), true)
|
||
assert.equal(await consumedOf(runId), 1000000)
|
||
})
|
||
|
||
test('spending a dimension with no row is refused', async (t) => {
|
||
if (needDb(t)) return
|
||
const runId = await seedBudget()
|
||
assert.equal(await spend(runId, 1, 'uo.bosses'), false)
|
||
})
|
||
|
||
test('seeding is INSERT IGNORE: a tick that overruns cannot reset a spent cap', async (t) => {
|
||
if (needDb(t)) return
|
||
// The idempotence `materialisePhase` and `gates.open` both have. Without it a
|
||
// second seed would either error on the unique key or — worse, written as an
|
||
// upsert — hand a run that has spent 28 of 30 a fresh 0.
|
||
const { runId } = await seedRun({ status: 'running' })
|
||
const SEED =
|
||
'INSERT IGNORE INTO event_run_budget (run_id, dimension, consumed, cap, effective_from) VALUES (?, ?, 0, ?, ?)'
|
||
await pool.query(SEED, [runId, 'uo.creatures', 30, 'uo.creature.spawn'])
|
||
await spend(runId, 28)
|
||
await pool.query(SEED, [runId, 'uo.creatures', 30, 'uo.creature.spawn'])
|
||
assert.equal(await consumedOf(runId), 28)
|
||
assert.equal((await pool.query('SELECT * FROM event_run_budget WHERE run_id = ?', [runId])).length, 1)
|
||
})
|
||
|
||
test('a refund floors at zero rather than going negative', async (t) => {
|
||
if (needDb(t)) return
|
||
// `GREATEST(consumed - ?, 0)`. A negative `consumed` would make every later cap
|
||
// check lie in the permissive direction, permanently — a worse outcome than a
|
||
// refund that is slightly too small.
|
||
const runId = await seedBudget({ consumed: 3, cap: 30 })
|
||
await pool.query(
|
||
'UPDATE event_run_budget SET consumed = GREATEST(consumed - ?, 0) WHERE run_id = ? AND dimension = ?',
|
||
[10, runId, 'uo.creatures'],
|
||
)
|
||
assert.equal(await consumedOf(runId), 0)
|
||
})
|
||
|
||
test('a budget goes with its run', async (t) => {
|
||
if (needDb(t)) return
|
||
const runId = await seedBudget()
|
||
await pool.query('DELETE FROM event_runs WHERE id = ?', [runId])
|
||
assert.equal((await pool.query('SELECT * FROM event_run_budget WHERE run_id = ?', [runId])).length, 0)
|
||
})
|
||
|
||
test('a version records that a dry run passed against it', async (t) => {
|
||
if (needDb(t)) return
|
||
// §K's gate is two columns on an immutable table, and the ALTER that adds them
|
||
// has to actually apply — `verified_at` is what `runsModel.create` reads to
|
||
// decide whether a scheduled occurrence may start at all, so a column that
|
||
// silently did not exist would hold every schedule on the deployment.
|
||
await pool.query('ALTER TABLE event_versions ADD COLUMN IF NOT EXISTS verified_at DATETIME NULL')
|
||
await pool.query('ALTER TABLE event_versions ADD COLUMN IF NOT EXISTS verified_by INT NULL')
|
||
const v = await pool.query('INSERT INTO event_versions (definition_id, spec) VALUES (1, ?)', ['{}'])
|
||
const before = (await pool.query('SELECT verified_at FROM event_versions WHERE id = ?', [v.insertId]))[0]
|
||
assert.equal(before.verified_at, null)
|
||
const moved = await pool.query('UPDATE event_versions SET verified_at = ?, verified_by = ? WHERE id = ?', [
|
||
T0,
|
||
1,
|
||
v.insertId,
|
||
])
|
||
assert.equal(Number(moved.affectedRows), 1)
|
||
const after = (
|
||
await pool.query('SELECT verified_at, verified_by FROM event_versions WHERE id = ?', [v.insertId])
|
||
)[0]
|
||
assert.ok(after.verified_at instanceof Date)
|
||
assert.equal(Number(after.verified_by), 1)
|
||
})
|
||
|
||
test('a non-positive spend never reaches the database', async (t) => {
|
||
if (needDb(t)) return
|
||
// Through the shipping module rather than a copy of its statement. An action
|
||
// whose `cost()` answers 0 for its params is telling core it consumes nothing,
|
||
// and pricing that as a query would be one round trip per step behind a fact
|
||
// the caller already has.
|
||
const runId = await seedBudget({ consumed: 0, cap: 10 })
|
||
assert.equal(await budgetDb.spend(runId, 'uo.creatures', 0), true)
|
||
assert.equal(await consumedOf(runId), 0)
|
||
})
|
||
|
||
// ── The resource ledger's unique key (Phase 8) ─────────────────────────────
|
||
//
|
||
// Every test here is about a property of the SERVER. A stub can enforce whatever
|
||
// rule its author had in mind; only MariaDB can say whether this encoding of
|
||
// "unique among live rows" actually is one.
|
||
|
||
const insertResource = async (runId, over = {}) =>
|
||
(
|
||
await pool.query(
|
||
`INSERT INTO event_run_resources (run_id, step_id, owner_module, kind, ref, payload, lease_until, member_key, status)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||
[
|
||
runId,
|
||
over.stepId ?? null,
|
||
over.owner ?? 'demo',
|
||
over.kind ?? 'creature',
|
||
over.ref ?? '0xA',
|
||
over.payload ?? null,
|
||
over.leaseUntil ?? null,
|
||
over.memberKey ?? null,
|
||
over.status ?? 'pending',
|
||
],
|
||
)
|
||
).insertId
|
||
|
||
const dup = async (fn) => {
|
||
try {
|
||
await fn()
|
||
return null
|
||
} catch (err) {
|
||
return err.code || String(err.errno)
|
||
}
|
||
}
|
||
|
||
test('two runs cannot hold one target, and the refusal comes from the server', async (t) => {
|
||
if (needDb(t)) return
|
||
// §D: "the unique index is what stops two events leasing one target." Not a
|
||
// read-then-insert — two runs entering the same tick would both pass the check
|
||
// — so the whole conflict story is this one constraint answering.
|
||
const a = await seedRun()
|
||
const b = await seedRun()
|
||
await insertResource(a.runId, { kind: 'override', ref: 'demo.rate', status: 'confirmed' })
|
||
const code = await dup(() => insertResource(b.runId, { kind: 'override', ref: 'demo.rate' }))
|
||
assert.equal(code, 'ER_DUP_ENTRY')
|
||
})
|
||
|
||
test('the key is released by the three statuses that mean it is no longer ours', async (t) => {
|
||
if (needDb(t)) return
|
||
// Amended 2026-09-03. §D says "among non-reverted rows", which was written
|
||
// before the six statuses had their meanings; taken literally it makes
|
||
// `drifted` and `orphaned` hold a target for ever, so one bad night would
|
||
// disable a lease permanently with no control able to clear it. `drifted` means
|
||
// somebody else has hold of the value and this run has let go; `orphaned` means
|
||
// it vanished. Neither is a claim on the target.
|
||
for (const status of ['reverted', 'drifted', 'orphaned']) {
|
||
await pool.query('DELETE FROM event_run_resources')
|
||
const a = await seedRun()
|
||
const b = await seedRun()
|
||
await insertResource(a.runId, { kind: 'override', ref: 'demo.rate', status })
|
||
const code = await dup(() => insertResource(b.runId, { kind: 'override', ref: 'demo.rate' }))
|
||
assert.equal(code, null, `a ${status} row must not hold the target`)
|
||
}
|
||
})
|
||
|
||
test('the key is HELD by the three that mean core still believes it is ours', async (t) => {
|
||
if (needDb(t)) return
|
||
for (const status of ['pending', 'confirmed', 'reverting']) {
|
||
await pool.query('DELETE FROM event_run_resources')
|
||
const a = await seedRun()
|
||
const b = await seedRun()
|
||
await insertResource(a.runId, { kind: 'override', ref: 'demo.rate', status })
|
||
const code = await dup(() => insertResource(b.runId, { kind: 'override', ref: 'demo.rate' }))
|
||
assert.equal(code, 'ER_DUP_ENTRY', `a ${status} row must hold the target`)
|
||
}
|
||
})
|
||
|
||
test('many released rows on one target coexist, which is the whole encoding', async (t) => {
|
||
if (needDb(t)) return
|
||
// The property the NULL depends on: multiple NULLs do not collide in a unique
|
||
// index. A weekly event that leases the same rate every Saturday accumulates one
|
||
// released row per week, and the fiftieth must not fail to insert.
|
||
const a = await seedRun()
|
||
for (let i = 0; i < 5; i++) {
|
||
await insertResource(a.runId, { kind: 'override', ref: 'demo.rate', status: 'reverted' })
|
||
}
|
||
const held = await pool.query(
|
||
'SELECT COUNT(*) AS n FROM event_run_resources WHERE ref = ? AND live_marker IS NULL',
|
||
['demo.rate'],
|
||
)
|
||
assert.equal(Number(held[0].n), 5)
|
||
// And a live one still goes on top of them.
|
||
assert.equal(await dup(() => insertResource(a.runId, { kind: 'override', ref: 'demo.rate' })), null)
|
||
})
|
||
|
||
// ── claimRevert's stale-claim reclaim (Phase 16) ───────────────────────────
|
||
//
|
||
// The statement's own comment explains WHY a `reverting` row must be reclaimable;
|
||
// this proves it against a real server, because every part of the answer is the
|
||
// server's: `NOW() - INTERVAL … MICROSECOND`, whether `ON UPDATE` re-stamps, and
|
||
// above all what `affectedRows` counts. This connector sends `CLIENT_FOUND_ROWS`,
|
||
// so it counts rows MATCHED — a stub counting CHANGED rows would call the reclaim
|
||
// a failure, and one counting matched rows would miss that the second claimer
|
||
// needs the re-stamp in order to lose. Only MariaDB settles it.
|
||
|
||
const CLAIM_REVERT = `
|
||
UPDATE event_run_resources
|
||
SET status = 'reverting', updated_at = NOW()
|
||
WHERE id = ?
|
||
AND (status IN ('pending', 'confirmed', 'orphaned', 'drifted')
|
||
OR (status = 'reverting'
|
||
AND updated_at < (NOW() - INTERVAL ? MICROSECOND)))`
|
||
|
||
const claimRevert = async (id, ttlMs) =>
|
||
Number((await pool.query(CLAIM_REVERT, [id, ttlMs * 1000]))?.affectedRows || 0) > 0
|
||
|
||
const ageResource = (id, seconds) =>
|
||
pool.query('UPDATE event_run_resources SET updated_at = NOW() - INTERVAL ? SECOND WHERE id = ?', [
|
||
seconds,
|
||
id,
|
||
])
|
||
|
||
test('a reverting row whose claim has gone stale is claimable again', async (t) => {
|
||
if (needDb(t)) return
|
||
// The Phase 16 walk's finding: a process killed mid-teardown leaves the row in
|
||
// `reverting` and nothing releases it. The sweep ran every 15s for ever finding
|
||
// nothing it could claim, `cleanup_status` never left `pending`, and the manual
|
||
// retry answered 200 while doing nothing — it claims through this statement too.
|
||
const run = await seedRun()
|
||
const id = await insertResource(run.runId, { status: 'reverting' })
|
||
|
||
// Fresh: somebody else really is mid-revert on it. Left alone.
|
||
assert.equal(await claimRevert(id, 900_000), false)
|
||
|
||
// Stale: the holder is not coming back.
|
||
await ageResource(id, 1800)
|
||
assert.equal(await claimRevert(id, 900_000), true)
|
||
})
|
||
|
||
test('reclaiming re-stamps, so the second claimer of one stale row loses', async (t) => {
|
||
if (needDb(t)) return
|
||
// Under CLIENT_FOUND_ROWS the four fresh statuses need no re-stamp — the winner
|
||
// moves the row out of `status IN (…)` and the loser stops matching. A stale
|
||
// `reverting` row has no such natural change, so without writing `updated_at`
|
||
// BOTH claimers would match it and two passes would revert the same resource.
|
||
const run = await seedRun()
|
||
const id = await insertResource(run.runId, { status: 'reverting' })
|
||
await ageResource(id, 1800)
|
||
|
||
assert.equal(await claimRevert(id, 900_000), true)
|
||
assert.equal(await claimRevert(id, 900_000), false, 'the re-stamp must make the second miss')
|
||
})
|
||
|
||
test('reclaiming a stale revert does not spend a retry attempt', async (t) => {
|
||
if (needDb(t)) return
|
||
// A stale claim is a process that died, not an attempt that failed. Counting it
|
||
// would burn MAX_REVERT_ATTEMPTS on crashes — Engagement Phase 14's rule, one
|
||
// table over.
|
||
const run = await seedRun()
|
||
const id = await insertResource(run.runId, { status: 'reverting' })
|
||
await ageResource(id, 1800)
|
||
await claimRevert(id, 900_000)
|
||
const [row] = await pool.query('SELECT revert_attempts FROM event_run_resources WHERE id = ?', [id])
|
||
assert.equal(Number(row.revert_attempts), 0)
|
||
})
|
||
|
||
test('an UPDATE that releases a row frees the target at once', async (t) => {
|
||
if (needDb(t)) return
|
||
// The generated column is STORED, so this is really asking whether MariaDB
|
||
// recomputes it on UPDATE and re-indexes. Cleanup depends on it entirely: the
|
||
// moment a lease is restored, the next event may take it.
|
||
const a = await seedRun()
|
||
const b = await seedRun()
|
||
const id = await insertResource(a.runId, { kind: 'override', ref: 'demo.rate', status: 'confirmed' })
|
||
assert.equal(await dup(() => insertResource(b.runId, { kind: 'override', ref: 'demo.rate' })), 'ER_DUP_ENTRY')
|
||
await pool.query("UPDATE event_run_resources SET status = 'reverted' WHERE id = ?", [id])
|
||
assert.equal(await dup(() => insertResource(b.runId, { kind: 'override', ref: 'demo.rate' })), null)
|
||
})
|
||
|
||
test('the key is per owner and per kind, not per ref', async (t) => {
|
||
if (needDb(t)) return
|
||
// `kind` and `ref` are module-opaque strings core stores verbatim, so two
|
||
// modules using the same word for different things must not collide — and one
|
||
// module's `creature:0xA` and `item:0xA` are two objects.
|
||
const a = await seedRun()
|
||
await insertResource(a.runId, { owner: 'demo', kind: 'creature', ref: '0xA', status: 'confirmed' })
|
||
assert.equal(await dup(() => insertResource(a.runId, { owner: 'other', kind: 'creature', ref: '0xA' })), null)
|
||
assert.equal(await dup(() => insertResource(a.runId, { owner: 'demo', kind: 'item', ref: '0xA' })), null)
|
||
assert.equal(await dup(() => insertResource(a.runId, { owner: 'demo', kind: 'creature', ref: '0xB' })), null)
|
||
})
|
||
|
||
test('a deleted step leaves its resources behind, and a deleted run does not', async (t) => {
|
||
if (needDb(t)) return
|
||
// TEAMS.md §2.5's correction, held as a test: `step_id` is SET NULL and it only
|
||
// works because the generated column reads `status` alone. If `step_id` were in
|
||
// that expression MariaDB would refuse the constraint outright (error 1901), and
|
||
// the migration would fail on a fresh install rather than here.
|
||
//
|
||
// The two directions are different on purpose. A record of what was changed in
|
||
// the WORLD must outlive the row that scheduled it — `engagement_sends`'
|
||
// argument — while a deleted RUN takes its ledger with it, because the ledger
|
||
// exists to answer questions about a run.
|
||
const a = await seedRun()
|
||
const stepId = await seedStep(a.runId)
|
||
const id = await insertResource(a.runId, { stepId, status: 'confirmed' })
|
||
|
||
await pool.query('DELETE FROM event_run_steps WHERE id = ?', [stepId])
|
||
const orphaned = (await pool.query('SELECT step_id, status FROM event_run_resources WHERE id = ?', [id]))[0]
|
||
assert.equal(orphaned.step_id, null)
|
||
assert.equal(orphaned.status, 'confirmed')
|
||
|
||
await pool.query('DELETE FROM event_runs WHERE id = ?', [a.runId])
|
||
const gone = await pool.query('SELECT id FROM event_run_resources WHERE id = ?', [id])
|
||
assert.equal(gone.length, 0)
|
||
})
|
||
|
||
test('the ledger model reserves, confirms, reverts and refuses for real', async (t) => {
|
||
if (needDb(t)) return
|
||
// Through the shipping module rather than a copy of its statements, like the
|
||
// budget tests above: `reserve` reads ER_DUP_ENTRY as a refusal and looks the
|
||
// holder up to name it, and both halves of that are the connector's behaviour
|
||
// rather than this file's.
|
||
const a = await seedRun()
|
||
const b = await seedRun()
|
||
|
||
const first = await resourcesDb.reserve({ runId: a.runId, owner: 'demo', kind: 'override', ref: 'demo.rate' })
|
||
assert.equal(first.ok, true)
|
||
await resourcesDb.confirm(first.id)
|
||
|
||
const second = await resourcesDb.reserve({ runId: b.runId, owner: 'demo', kind: 'override', ref: 'demo.rate' })
|
||
assert.equal(second.ok, false)
|
||
assert.equal(second.code, 'held')
|
||
assert.equal(Number(second.holder.run_id), Number(a.runId))
|
||
assert.equal(second.holder.status, 'confirmed')
|
||
|
||
// And once it is given back the second run gets it.
|
||
await resourcesDb.markReverted(first.id)
|
||
const third = await resourcesDb.reserve({ runId: b.runId, owner: 'demo', kind: 'override', ref: 'demo.rate' })
|
||
assert.equal(third.ok, true)
|
||
})
|
||
|
||
test('failRevert increments and NEVER resets, and only a human clears it', async (t) => {
|
||
if (needDb(t)) return
|
||
// Engagement Phase 14's rule at the statement level: `revert_attempts =
|
||
// revert_attempts + 1` is written in one place, and the reset is a separate
|
||
// statement with an actor behind it. A `SET revert_attempts = ?` anywhere in
|
||
// the sweep would make the ceiling unreachable and the row cycle for ever.
|
||
const a = await seedRun()
|
||
const id = await resourcesDb
|
||
.reserve({ runId: a.runId, owner: 'demo', kind: 'creature', ref: '0xA' })
|
||
.then((r) => r.id)
|
||
await resourcesDb.confirm(id)
|
||
|
||
await resourcesDb.failRevert(id, 'the shard did not answer')
|
||
await resourcesDb.failRevert(id, 'still nothing')
|
||
let row = (await pool.query('SELECT * FROM event_run_resources WHERE id = ?', [id]))[0]
|
||
assert.equal(Number(row.revert_attempts), 2)
|
||
assert.equal(row.status, 'confirmed')
|
||
assert.equal(row.last_error, 'still nothing')
|
||
|
||
assert.equal(await resourcesDb.resetAttempts(a.runId), 1)
|
||
row = (await pool.query('SELECT * FROM event_run_resources WHERE id = ?', [id]))[0]
|
||
assert.equal(Number(row.revert_attempts), 0)
|
||
})
|
||
|
||
test('claimRevert is a compare-and-set, and reverting is not re-claimable', async (t) => {
|
||
if (needDb(t)) return
|
||
// The cleanup leg and the manual cleanup route can both be working one run at
|
||
// once. `reverting` is deliberately not claimable — a row another pass is
|
||
// mid-revert on is left alone, exactly as a step with a live claim is.
|
||
const a = await seedRun()
|
||
const id = await resourcesDb
|
||
.reserve({ runId: a.runId, owner: 'demo', kind: 'creature', ref: '0xA' })
|
||
.then((r) => r.id)
|
||
await resourcesDb.confirm(id)
|
||
|
||
assert.equal(await resourcesDb.claimRevert(id), true)
|
||
assert.equal(await resourcesDb.claimRevert(id), false, 'a second pass must not take a row mid-revert')
|
||
|
||
await resourcesDb.markReverted(id)
|
||
assert.equal(await resourcesDb.claimRevert(id), false, 'and a reverted row is finished')
|
||
})
|
||
|
||
test('the placeholder is resolvable exactly once, and only if it is a placeholder', async (t) => {
|
||
if (needDb(t)) return
|
||
// The `kind = '@step'` guard in the statement, not in JavaScript. It is what
|
||
// stops a bug elsewhere resolving a real resource — which would be core marking
|
||
// a live creature as given back without asking anyone.
|
||
const a = await seedRun()
|
||
const placeholder = await resourcesDb
|
||
.reserve({ runId: a.runId, owner: 'demo', kind: resourcesDb.STEP_KIND, ref: 'k'.repeat(40) })
|
||
.then((r) => r.id)
|
||
const real = await resourcesDb
|
||
.reserve({ runId: a.runId, owner: 'demo', kind: 'creature', ref: '0xA' })
|
||
.then((r) => r.id)
|
||
|
||
assert.equal(await resourcesDb.resolvePlaceholder(placeholder), true)
|
||
assert.equal(await resourcesDb.resolvePlaceholder(placeholder), false)
|
||
assert.equal(await resourcesDb.resolvePlaceholder(real), false)
|
||
const stillThere = (await pool.query('SELECT status FROM event_run_resources WHERE id = ?', [real]))[0]
|
||
assert.equal(stillThere.status, 'pending')
|
||
})
|
||
|
||
test('the unresolved reads see the five statuses that still want something', async (t) => {
|
||
if (needDb(t)) return
|
||
// `cleanup_status` is derived from this count, so what it includes IS the
|
||
// definition of "clean". A `drifted` row left out of it would let a run whose
|
||
// lease somebody else took call itself complete.
|
||
const a = await seedRun()
|
||
for (const status of ['pending', 'confirmed', 'reverting', 'reverted', 'orphaned', 'drifted']) {
|
||
await insertResource(a.runId, { ref: `ref-${status}`, status })
|
||
}
|
||
assert.equal(await resourcesDb.unresolvedCount(a.runId), 5)
|
||
const rows = await resourcesDb.unresolvedForRun(a.runId)
|
||
assert.deepEqual(
|
||
rows.map((r) => r.status).sort(),
|
||
['confirmed', 'drifted', 'orphaned', 'pending', 'reverting'],
|
||
)
|
||
const counts = await resourcesDb.unresolvedCounts([a.runId])
|
||
assert.equal(counts.get(a.runId) ?? counts.get(String(a.runId)), 5)
|
||
})
|
||
|
||
test('payload comes back as an object rather than as a string', async (t) => {
|
||
if (needDb(t)) return
|
||
// A lease's baseline lives in here and the cleanup sweep reads it out to pass
|
||
// to `restore`. The connector hands JSON back as text, so a missing hydration
|
||
// is a `restore(undefined)` — a lease put back to nothing, silently.
|
||
const a = await seedRun()
|
||
const id = await resourcesDb
|
||
.reserve({
|
||
runId: a.runId,
|
||
owner: 'demo',
|
||
kind: 'override',
|
||
ref: 'demo.rate',
|
||
payload: { baseline: 1, applied: 3 },
|
||
})
|
||
.then((r) => r.id)
|
||
void id
|
||
const [row] = await resourcesDb.forRun(a.runId)
|
||
assert.deepEqual(row.payload, { baseline: 1, applied: 3 })
|
||
})
|
||
|
||
test('the cleanup scan finds a run that owes something, whatever its status column says', async (t) => {
|
||
if (needDb(t)) return
|
||
await pool.query('ALTER TABLE event_runs ADD COLUMN IF NOT EXISTS cleanup_status ' +
|
||
"ENUM('not_required','pending','complete','incomplete') NOT NULL DEFAULT 'not_required'")
|
||
|
||
// **`not_required` is in the scan because of a live-walk defect**, not for
|
||
// tidiness. A run whose only resource was a LEASE never went through the
|
||
// ledger's `markRunDirty` — `core.lease` reserves its own row — so its column
|
||
// stayed `not_required` and the lease was never given back at all. A terminal
|
||
// run with an unresolved row has work to do whatever any summary column says.
|
||
const lease = await seedRun({ status: 'completed' })
|
||
await insertResource(lease.runId, { kind: 'override', ref: 'demo.rate', status: 'confirmed' })
|
||
const marked = await seedRun({ status: 'cancelled' })
|
||
await insertResource(marked.runId, { ref: '0xB', status: 'pending' })
|
||
await pool.query("UPDATE event_runs SET cleanup_status = 'pending' WHERE id = ?", [marked.runId])
|
||
|
||
// Three that must NOT be selected, one per reason.
|
||
const running = await seedRun({ status: 'running' })
|
||
await insertResource(running.runId, { ref: '0xC', status: 'confirmed' })
|
||
const done = await seedRun({ status: 'completed' })
|
||
await insertResource(done.runId, { ref: '0xD', status: 'reverted' })
|
||
const givenUp = await seedRun({ status: 'completed' })
|
||
await insertResource(givenUp.runId, { ref: '0xE', status: 'confirmed' })
|
||
await pool.query("UPDATE event_runs SET cleanup_status = 'incomplete' WHERE id = ?", [givenUp.runId])
|
||
|
||
const found = (await resourcesDb.runsNeedingCleanup(10, 3)).map((r) => Number(r.id))
|
||
assert.deepEqual(found.sort(), [Number(lease.runId), Number(marked.runId)].sort())
|
||
})
|
||
|
||
test('the scan stops selecting a run whose attempts are spent', async (t) => {
|
||
if (needDb(t)) return
|
||
// Without the bound in the join, a run whose rows are all spent would be
|
||
// selected, worked over and found to have nothing to do on every tick for the
|
||
// rest of its life.
|
||
const a = await seedRun({ status: 'completed' })
|
||
const id = await insertResource(a.runId, { status: 'confirmed' })
|
||
await pool.query("UPDATE event_runs SET cleanup_status = 'pending' WHERE id = ?", [a.runId])
|
||
assert.equal((await resourcesDb.runsNeedingCleanup(10, 3)).length, 1)
|
||
|
||
await pool.query('UPDATE event_run_resources SET revert_attempts = 3 WHERE id = ?', [id])
|
||
assert.equal((await resourcesDb.runsNeedingCleanup(10, 3)).length, 0)
|
||
})
|
||
|
||
test('setCleanupStatus is guarded, which is what stops a late row re-opening a swept run', async (t) => {
|
||
if (needDb(t)) return
|
||
const a = await seedRun({ status: 'completed' })
|
||
assert.equal(await runsDb.setCleanupStatus(a.runId, 'complete'), true)
|
||
assert.equal(await runsDb.setCleanupStatus(a.runId, 'pending', ['not_required']), false)
|
||
assert.equal(await runsDb.setCleanupStatus(a.runId, 'pending'), true)
|
||
})
|
||
|
||
// ── A trigger can start a run (Rust PLAN_REDESIGNS §14.3.12, D370) ─────────
|
||
//
|
||
// `eventStarts.test.js` proves what `starts.observe()` DECIDES against a stand-in
|
||
// whose claim keeps the rules. These prove the rules are what the server does:
|
||
// the JSON lookup over the published version, and the claim — one winner, the
|
||
// cooldown and the one-at-a-time check all inside one WHERE.
|
||
|
||
const definitionsDb = require('../src/model/events/eventDefinitions.db')
|
||
const triggerStartsDb = require('../src/model/events/eventTriggerStarts.db')
|
||
|
||
const startsOn = (on, extra = {}) => ({
|
||
schedule: { kind: 'manual' },
|
||
starts: [{ on, where: null, cooldown: '30m', dormant: false, ...extra }],
|
||
phases: [{ key: 'main', label: 'Main', steps: [] }],
|
||
})
|
||
|
||
test('findStartingOn reads the PUBLISHED version of ready definitions, and nothing else', async (t) => {
|
||
if (needDb(t)) return
|
||
const hit = await seedDefinition({ spec: startsOn('demo.bradley.destroyed') })
|
||
await seedDefinition({ state: 'draft', spec: startsOn('demo.bradley.destroyed') })
|
||
await seedDefinition({ state: 'archived', spec: startsOn('demo.bradley.destroyed') })
|
||
await seedDefinition({ spec: startsOn('demo.heli.downed') })
|
||
await seedDefinition({ spec: { schedule: { kind: 'manual' } } })
|
||
// Two starts, the trigger second: `$.starts[*].on` is every entry, not the first.
|
||
const second = await seedDefinition({
|
||
spec: { ...startsOn('demo.heli.downed'), starts: [{ on: 'demo.heli.downed' }, { on: 'demo.bradley.destroyed' }] },
|
||
})
|
||
await pool.query('UPDATE event_versions SET verified_at = ? WHERE id = ?', [T0, hit.versionId])
|
||
|
||
const found = await definitionsDb.findStartingOn('demo.bradley.destroyed')
|
||
assert.deepEqual(found.map((d) => Number(d.id)), [Number(hit.definitionId), Number(second.definitionId)])
|
||
assert.equal(found[0].version_spec.starts[0].on, 'demo.bradley.destroyed', 'the spec comes back parsed')
|
||
assert.ok(found[0].version_verified_at, 'and whether it may start unattended')
|
||
assert.equal(found[1].version_verified_at, null)
|
||
// A substring is not a trigger.
|
||
assert.equal((await definitionsDb.findStartingOn('demo.bradley')).length, 0)
|
||
})
|
||
|
||
test('two firings claiming at once: exactly one owns the start', async (t) => {
|
||
if (needDb(t)) return
|
||
const def = await seedDefinition({ spec: startsOn('demo.x', { cooldown: '0m' }) })
|
||
const results = await Promise.all(
|
||
Array.from({ length: 5 }, (_, i) =>
|
||
triggerStartsDb.claim(def.definitionId, { at: later(i * 1000), cooldownSeconds: 0, now: later(10_000) }),
|
||
),
|
||
)
|
||
assert.equal(results.filter(Boolean).length, 1)
|
||
})
|
||
|
||
test('the cooldown is measured from the last trigger-started run', async (t) => {
|
||
if (needDb(t)) return
|
||
const def = await seedDefinition({ spec: startsOn('demo.x') })
|
||
const id = def.definitionId
|
||
assert.equal(await triggerStartsDb.claim(id, { at: T0, cooldownSeconds: 1800, now: T0 }), true)
|
||
await triggerStartsDb.started(id, { at: T0, runId: 99 })
|
||
|
||
assert.equal(await triggerStartsDb.claim(id, { at: later(10 * 60_000), cooldownSeconds: 1800, now: later(10 * 60_000) }), false)
|
||
// Exactly at the boundary the cooldown has passed.
|
||
assert.equal(await triggerStartsDb.claim(id, { at: later(30 * 60_000), cooldownSeconds: 1800, now: later(30 * 60_000) }), true)
|
||
})
|
||
|
||
test('a going run blocks the claim; a future occurrence does not', async (t) => {
|
||
if (needDb(t)) return
|
||
const def = await seedDefinition({ spec: startsOn('demo.x', { cooldown: '0m' }) })
|
||
const id = def.definitionId
|
||
const run = await pool.query(
|
||
'INSERT INTO event_runs (definition_id, version_id, scope, status, scheduled_for) VALUES (?, ?, ?, ?, ?)',
|
||
[id, def.versionId, '', 'scheduled', later(24 * 60 * 60_000)],
|
||
)
|
||
assert.equal(await triggerStartsDb.isGoing(id, T0), false, 'tomorrow\'s occurrence is not going')
|
||
assert.equal(await triggerStartsDb.claim(id, { at: T0, cooldownSeconds: 0, now: T0 }), true)
|
||
await triggerStartsDb.release(id)
|
||
|
||
for (const [status, when] of [['running', T0], ['paused', T0], ['scheduled', later(-1000)]]) {
|
||
await pool.query('UPDATE event_runs SET status = ?, scheduled_for = ? WHERE id = ?', [status, when, run.insertId])
|
||
assert.equal(await triggerStartsDb.isGoing(id, T0), true, status)
|
||
assert.equal(await triggerStartsDb.claim(id, { at: T0, cooldownSeconds: 0, now: T0 }), false, status)
|
||
}
|
||
await pool.query("UPDATE event_runs SET status = 'completed' WHERE id = ?", [run.insertId])
|
||
assert.equal(await triggerStartsDb.claim(id, { at: T0, cooldownSeconds: 0, now: T0 }), true)
|
||
})
|
||
|
||
test('a claim held past a minute belongs to a dead process and can be taken', async (t) => {
|
||
if (needDb(t)) return
|
||
const def = await seedDefinition({ spec: startsOn('demo.x', { cooldown: '0m' }) })
|
||
const id = def.definitionId
|
||
assert.equal(await triggerStartsDb.claim(id, { at: T0, cooldownSeconds: 0, now: T0 }), true)
|
||
assert.equal(await triggerStartsDb.claim(id, { at: later(30_000), cooldownSeconds: 0, now: later(30_000) }), false)
|
||
assert.equal(await triggerStartsDb.claim(id, { at: later(61_000), cooldownSeconds: 0, now: later(61_000) }), true)
|
||
})
|
||
|
||
test('ignore counts by reason, and a start zeroes the counts', async (t) => {
|
||
if (needDb(t)) return
|
||
const def = await seedDefinition({ spec: startsOn('demo.x') })
|
||
const id = def.definitionId
|
||
await triggerStartsDb.ignore(id, 'running', T0)
|
||
await triggerStartsDb.ignore(id, 'running', T0)
|
||
await triggerStartsDb.ignore(id, 'cooldown', later(1000))
|
||
let row = await triggerStartsDb.forDefinition(id)
|
||
assert.equal(row.ignored_running, 2)
|
||
assert.equal(row.ignored_cooldown, 1)
|
||
assert.equal(row.last_ignored_reason, 'cooldown')
|
||
await assert.rejects(() => triggerStartsDb.ignore(id, 'bogus', T0), /unknown ignore reason/)
|
||
|
||
await triggerStartsDb.started(id, { at: later(2000), runId: 5 })
|
||
row = await triggerStartsDb.forDefinition(id)
|
||
assert.equal(row.ignored_running, 0)
|
||
assert.equal(Number(row.last_run_id), 5)
|
||
assert.equal(row.claimed_at, null)
|
||
})
|
||
|
||
test('a run carries its origin through the insert and back', async (t) => {
|
||
if (needDb(t)) return
|
||
const def = await seedDefinition()
|
||
const origin = { kind: 'trigger', triggerId: 'demo.x', variables: { grid: 'G8' } }
|
||
const runId = await runsDb.materialise({
|
||
definition_id: def.definitionId,
|
||
version_id: def.versionId,
|
||
scope: '',
|
||
scheduled_for: T0,
|
||
concurrency_key: null,
|
||
params: null,
|
||
started_by: null,
|
||
origin,
|
||
})
|
||
assert.deepEqual((await runsDb.getById(runId)).origin, origin)
|
||
})
|