feat(events): a trigger can start a run
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
This commit is contained in:
@@ -12,13 +12,14 @@
|
||||
// one that decides. A spec arriving by any other route (a restore, a fixture, a
|
||||
// module shipping a definition as content) gets the same answer.
|
||||
//
|
||||
// **What this file knows, and what it deliberately refuses.** Two top-level keys
|
||||
// exist today: `schedule` and `phases`. Unknown top-level keys are REFUSED rather
|
||||
// **What this file knows, and what it deliberately refuses.** Three top-level
|
||||
// keys exist today: `schedule`, `starts` and `phases`. Unknown top-level keys are REFUSED rather
|
||||
// than preserved: a spec that silently carries `announcements` today is a spec
|
||||
// whose author believes announcements work, and the later phase that gives the
|
||||
// key meaning would inherit a corpus of unvalidated ones. The refusal list is the
|
||||
// changelog — Phase 4 added the recurrence shapes, Phase 5 added a phase's
|
||||
// `advance`, Phase 10 adds `announcements`.
|
||||
// `advance`, Phase 10 adds `announcements`, and Rust PLAN_REDESIGNS §14.3.12
|
||||
// added `starts`, the triggers that begin a run (D370).
|
||||
//
|
||||
// **Phase 4 widened `schedule` from one shape to four** — `manual`, `once`,
|
||||
// `weekly`, `monthly` — and every check on them is a check on SHAPE. The
|
||||
@@ -76,6 +77,24 @@ const MAX_AFTER_SECONDS = 30 * 86_400
|
||||
// never advances written as one that eventually does.
|
||||
const MAX_ADVANCE_COUNT = 1000
|
||||
|
||||
// A definition's `starts` (Rust PLAN_REDESIGNS §14.3.12, D370): the triggers that
|
||||
// begin a run of it. Bounded for the reason MAX_PHASES is — a paste guard, not a
|
||||
// policy, and ten is more than any event here has wanted.
|
||||
const MAX_STARTS = 10
|
||||
|
||||
// How long after one trigger-started run another may start. Thirty minutes by
|
||||
// default, zero (one run at a time is then the only bound) up to a week. It is
|
||||
// the same one-integer-one-unit grammar as `after`, for the same proofreading
|
||||
// reason, and normalised the same way.
|
||||
const DEFAULT_COOLDOWN = '30m'
|
||||
const MAX_COOLDOWN_SECONDS = 7 * 86_400
|
||||
|
||||
// Core's own lifecycle triggers cannot start a run (D370). `event.run.started`
|
||||
// starting a run would be a run starting another, and the one-at-a-time rule is
|
||||
// per definition, so it would not hold a chain of two definitions starting each
|
||||
// other. Refused by prefix because every one of them is core's and named so.
|
||||
const CORE_EVENT_PREFIX = 'event.'
|
||||
|
||||
// The four closed shapes of §E. `manual` is first because it is the default and
|
||||
// what an unscheduled draft carries; the other three are recurrences the runner
|
||||
// expands into occurrences ahead of time.
|
||||
@@ -323,6 +342,74 @@ function validateAdvance(raw, path, errors) {
|
||||
return { on: triggerId, where: checked.conditions, count, dormant: false }
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse a cooldown into seconds, or answer null. `'0m'` (or a bare `0`) is
|
||||
* allowed here and nowhere else in the duration grammar: no cooldown at all is a
|
||||
* real choice, and then one run at a time is the only bound.
|
||||
*/
|
||||
function parseCooldown(raw) {
|
||||
if (raw === 0 || raw === '0') return 0
|
||||
const m = AFTER_RE.exec(String(raw ?? ''))
|
||||
if (!m) return null
|
||||
const seconds = Number(m[1]) * AFTER_UNIT_SECONDS[m[2]]
|
||||
return seconds <= MAX_COOLDOWN_SECONDS ? seconds : null
|
||||
}
|
||||
|
||||
const formatCooldown = (seconds) => (seconds === 0 ? '0m' : formatAfter(seconds))
|
||||
|
||||
/**
|
||||
* Check a definition's `starts` and answer the normalised list (D370).
|
||||
*
|
||||
* Each entry is a phase gate's `on` and `where` — the same catalogue, the same
|
||||
* grammar, the same dormant rule — plus a `cooldown`. A trigger nobody registers
|
||||
* makes the entry dormant: it saves, and it refuses to publish.
|
||||
*/
|
||||
function validateStarts(raw, errors) {
|
||||
if (raw === undefined || raw === null) return []
|
||||
if (!Array.isArray(raw)) {
|
||||
errors.push('spec.starts: expected an array')
|
||||
return []
|
||||
}
|
||||
if (raw.length > MAX_STARTS) {
|
||||
errors.push(`spec.starts: at most ${MAX_STARTS} triggers`)
|
||||
return []
|
||||
}
|
||||
|
||||
const out = []
|
||||
raw.forEach((entry, i) => {
|
||||
const path = `spec.starts[${i}]`
|
||||
if (!isPlainObject(entry)) {
|
||||
errors.push(`${path}: expected an object`)
|
||||
return
|
||||
}
|
||||
const extra = Object.keys(entry).filter((k) => !['on', 'where', 'cooldown', 'dormant'].includes(k))
|
||||
if (extra.length) {
|
||||
errors.push(`${path}: unknown key(s) ${extra.join(', ')}`)
|
||||
return
|
||||
}
|
||||
if (typeof entry.on === 'string' && entry.on.startsWith(CORE_EVENT_PREFIX)) {
|
||||
errors.push(`${path}.on: "${entry.on}" is one of the event system's own triggers, and a run cannot start another`)
|
||||
return
|
||||
}
|
||||
|
||||
const cooldownSeconds = parseCooldown(entry.cooldown ?? DEFAULT_COOLDOWN)
|
||||
if (cooldownSeconds === null) {
|
||||
errors.push(
|
||||
`${path}.cooldown: expected a duration like "30m" — a whole number of s, m, h or d, ` +
|
||||
`from 0 to ${formatAfter(MAX_COOLDOWN_SECONDS)}`,
|
||||
)
|
||||
return
|
||||
}
|
||||
|
||||
// The trigger and its predicate are a gate's, so a gate's validator answers
|
||||
// for them: one grammar, one dormant rule, one set of messages.
|
||||
const gate = validateAdvance({ on: entry.on, where: entry.where ?? null }, path, errors)
|
||||
if (!gate) return
|
||||
out.push({ on: gate.on, where: gate.where, cooldown: formatCooldown(cooldownSeconds), dormant: gate.dormant })
|
||||
})
|
||||
return out
|
||||
}
|
||||
|
||||
/**
|
||||
* A `multiple` param's value as the list it is (MODULE_API 1.14.0, Rust
|
||||
* PLAN_REDESIGNS §12.3).
|
||||
@@ -427,7 +514,7 @@ function validate(raw, { knownActionIds = [] } = {}) {
|
||||
|
||||
if (!isPlainObject(raw)) return { ok: false, errors: ['spec: expected an object'] }
|
||||
|
||||
const allowed = new Set(['schedule', 'phases'])
|
||||
const allowed = new Set(['schedule', 'phases', 'starts'])
|
||||
for (const key of Object.keys(raw)) {
|
||||
if (!allowed.has(key)) {
|
||||
errors.push(`spec: unknown key "${key}" (Phase 1 understands ${[...allowed].join(', ')})`)
|
||||
@@ -452,6 +539,9 @@ function validate(raw, { knownActionIds = [] } = {}) {
|
||||
schedule = validateSchedule(kind, rawSchedule, errors)
|
||||
}
|
||||
|
||||
// ── starts (D370) ──
|
||||
const starts = validateStarts(raw.starts, errors)
|
||||
|
||||
// ── phases ──
|
||||
const rawPhases = raw.phases
|
||||
const phases = []
|
||||
@@ -605,7 +695,10 @@ function validate(raw, { knownActionIds = [] } = {}) {
|
||||
if (totalSteps > MAX_STEPS) errors.push(`spec: at most ${MAX_STEPS} steps in one definition`)
|
||||
|
||||
if (errors.length) return { ok: false, errors }
|
||||
return { ok: true, spec: { schedule, phases } }
|
||||
// `starts` is omitted when empty, for the reason a phase's `advance` is: a spec
|
||||
// carrying `"starts": []` everywhere is a version diff that says every
|
||||
// definition changed the first time one of them gained a trigger.
|
||||
return { ok: true, spec: starts.length ? { schedule, starts, phases } : { schedule, phases } }
|
||||
}
|
||||
|
||||
/** Every action id a spec names, dormant ones included. */
|
||||
@@ -630,6 +723,9 @@ function publishable(spec) {
|
||||
const dormant = [
|
||||
...phases.flatMap((p) => (p.steps || []).filter((s) => s.dormant).map((s) => s.actionId)),
|
||||
...phases.filter((p) => p.advance?.dormant).map((p) => p.advance.on),
|
||||
// A start on a trigger nothing registers, on the same rule (D370): a version
|
||||
// that starts on something nobody can emit is one that never starts.
|
||||
...(spec?.starts || []).filter((s) => s.dormant).map((s) => s.on),
|
||||
]
|
||||
return dormant.length ? { ok: false, dormant: [...new Set(dormant)] } : { ok: true, dormant: [] }
|
||||
}
|
||||
@@ -651,7 +747,12 @@ module.exports = {
|
||||
defaultOnFailure,
|
||||
parseAfter,
|
||||
formatAfter,
|
||||
parseCooldown,
|
||||
PHASE_KEY,
|
||||
MAX_STARTS,
|
||||
DEFAULT_COOLDOWN,
|
||||
MAX_COOLDOWN_SECONDS,
|
||||
CORE_EVENT_PREFIX,
|
||||
SCHEDULE_KINDS,
|
||||
ADVANCE_KINDS,
|
||||
MAX_ADVANCE_COUNT,
|
||||
|
||||
210
server/src/events/starts.js
Normal file
210
server/src/events/starts.js
Normal file
@@ -0,0 +1,210 @@
|
||||
// ── Starting a run from a trigger: the emit path's third subscriber ────────
|
||||
//
|
||||
// Rust PLAN_REDESIGNS §14.3.12 (D370), and EVENTS.md §E "Starting a run from a
|
||||
// trigger". *"When anyone destroys Bradley on Main, start Launch Site
|
||||
// Lockdown"*; *"when a champion's boss appears, start Champion's Call"*. Before
|
||||
// this a trigger could only advance a phase of a run already going (`gates.js`)
|
||||
// or send a notice (`engine.dispatch`); now a definition's published version may
|
||||
// carry `starts`, and a firing that matches one creates a run of it, now.
|
||||
//
|
||||
// **Same seam and posture as the other two subscribers**: called from
|
||||
// `engagementEmit.emit`, never awaited, never allowed to reject. Whether some
|
||||
// event wanted to start is core's business, and a failure of core's must not
|
||||
// become the module's control flow.
|
||||
//
|
||||
// **Four rules decide whether a matching firing starts anything:**
|
||||
//
|
||||
// 1. The version must have passed its dry run. A trigger start is unattended,
|
||||
// as a scheduled one is, and §K's gate falls on the path where nobody is.
|
||||
// 2. One run at a time per definition. A firing while a run of it is going
|
||||
// starts nothing and is counted.
|
||||
// 3. A cooldown after the last trigger-started run (default 30 minutes, 0 to
|
||||
// 7 days). A firing inside it starts nothing and is counted. Rules 2 and 3
|
||||
// are what stop twelve road Bradleys dying in a minute starting twelve runs.
|
||||
// 4. Core's own `event.*` triggers never start a run (refused at save, and
|
||||
// skipped here as well), so a run starting can never start another.
|
||||
//
|
||||
// Rules 2 and 3 are ONE conditional UPDATE (`eventTriggerStarts.db.claim`); the
|
||||
// occurrence key `UNIQUE (definition_id, scope, scheduled_for)` with
|
||||
// `scheduled_for` = the firing's time is what makes one firing reported twice
|
||||
// (two instances ingesting one feed) create one run.
|
||||
|
||||
const definitionsDb = require('../model/events/eventDefinitions.db')
|
||||
const startsDb = require('../model/events/eventTriggerStarts.db')
|
||||
const runsModel = require('../model/events/eventRuns.model')
|
||||
const conditions = require('../engagement/conditions')
|
||||
const registries = require('../modules/registries')
|
||||
const spec = require('./spec')
|
||||
const gates = require('./gates')
|
||||
const log = require('../utils/logger')('event-starts')
|
||||
|
||||
// What of a firing's data is kept on the run. The run row is admin-only, and it
|
||||
// is the answer to "why did this run happen", so the trigger's DECLARED
|
||||
// variables are kept and nothing else a module happened to send. Bounded, so
|
||||
// one oversized payload cannot make the insert the thing that fails.
|
||||
const MAX_ORIGIN_BYTES = 4096
|
||||
|
||||
/** The firing's time, in whole seconds (the column's precision), never in the future. */
|
||||
function firingTime(occurredAt, now) {
|
||||
const parsed = occurredAt ? new Date(occurredAt) : now
|
||||
const at = Number.isNaN(parsed.getTime()) || parsed > now ? now : parsed
|
||||
return new Date(Math.floor(at.getTime() / 1000) * 1000)
|
||||
}
|
||||
|
||||
const sameSecond = (a, b) =>
|
||||
a && b && Math.floor(new Date(a).getTime() / 1000) === Math.floor(new Date(b).getTime() / 1000)
|
||||
|
||||
/** The firing as the run will remember it. */
|
||||
function originOf(event, declaration, start, at) {
|
||||
const declared = (declaration?.variables || []).map((v) => v.name)
|
||||
const data = event.data || {}
|
||||
let variables = Object.fromEntries(declared.filter((n) => n in data).map((n) => [n, data[n]]))
|
||||
if (Buffer.byteLength(JSON.stringify(variables), 'utf8') > MAX_ORIGIN_BYTES) {
|
||||
// Too big to keep whole: keep the ones the start's condition named, which
|
||||
// are the reason it matched, and say the rest were dropped.
|
||||
const named = gates.variablesIn(start.where)
|
||||
variables = Object.fromEntries(named.filter((n) => n in data).map((n) => [n, data[n]]))
|
||||
variables._truncated = true
|
||||
}
|
||||
return {
|
||||
kind: 'trigger',
|
||||
triggerId: event.triggerId,
|
||||
label: declaration?.label || event.triggerId,
|
||||
occurredAt: at.toISOString(),
|
||||
subject: event.subject ?? null,
|
||||
// The start's condition in the builder's own words, so the console can say
|
||||
// what it matched on without re-reading the version.
|
||||
matched: gates.phrase(start.where),
|
||||
variables,
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Why a firing that matched did not get the claim. A duplicate of the firing
|
||||
* that started the last run is not counted: it is the same thing happening once,
|
||||
* reported twice.
|
||||
*/
|
||||
async function whyNot(definitionId, at, now) {
|
||||
const state = await startsDb.forDefinition(definitionId)
|
||||
if (state && sameSecond(state.last_started_at, at)) return 'duplicate'
|
||||
if (await startsDb.isGoing(definitionId, now)) return 'running'
|
||||
// A live claim is another firing starting it this instant: the run is going in
|
||||
// every sense but the row.
|
||||
if (state?.claimed_at && now - new Date(state.claimed_at) < startsDb.CLAIM_STALE_MS) return 'running'
|
||||
return 'cooldown'
|
||||
}
|
||||
|
||||
/** One definition, one firing. Answers what happened, for the summary and the tests. */
|
||||
async function consider(definition, event, at, now) {
|
||||
const starts = (definition.version_spec?.starts || []).filter((s) => s.on === event.triggerId && !s.dormant)
|
||||
const start = starts.find((s) => conditions.evaluate(s.where, event.data || {}))
|
||||
if (!start) return 'unmatched'
|
||||
|
||||
if (now - at > Number(definition.grace_seconds || 0) * 1000) {
|
||||
// A firing older than the grace window would make a run the missed sweep
|
||||
// takes on its next tick. Said, and not started.
|
||||
log.warn('a firing too old to start a run', { definition: definition.id, trigger: event.triggerId, at })
|
||||
return 'stale'
|
||||
}
|
||||
|
||||
if (!definition.version_verified_at) {
|
||||
await startsDb.ignore(definition.id, 'unverified', now)
|
||||
log.warn('trigger start held: the published version has never been verified', {
|
||||
definition: definition.id,
|
||||
title: definition.title,
|
||||
trigger: event.triggerId,
|
||||
})
|
||||
return 'unverified'
|
||||
}
|
||||
|
||||
const cooldownSeconds = spec.parseCooldown(start.cooldown ?? spec.DEFAULT_COOLDOWN) ?? 0
|
||||
if (!(await startsDb.claim(definition.id, { at, cooldownSeconds, now }))) {
|
||||
const reason = await whyNot(definition.id, at, now)
|
||||
if (reason !== 'duplicate') await startsDb.ignore(definition.id, reason, now)
|
||||
return reason
|
||||
}
|
||||
|
||||
const declaration = registries.eventTrigger(event.triggerId)
|
||||
let result
|
||||
try {
|
||||
result = await runsModel.create(
|
||||
definition.id,
|
||||
// Scope is empty, as a scheduled occurrence's is (org lead, 2026-09-02): a
|
||||
// scope registry does not exist, and the condition is what narrows a start
|
||||
// to one server.
|
||||
{ scope: '', scheduledFor: at, source: 'trigger', origin: originOf(event, declaration, start, at) },
|
||||
null,
|
||||
)
|
||||
} catch (err) {
|
||||
await startsDb.release(definition.id)
|
||||
throw err
|
||||
}
|
||||
|
||||
if (!result.ok) {
|
||||
await startsDb.release(definition.id)
|
||||
if (result.code === 'unverified') await startsDb.ignore(definition.id, 'unverified', now)
|
||||
log.warn('trigger start refused', { definition: definition.id, errors: result.errors })
|
||||
return 'refused'
|
||||
}
|
||||
|
||||
if (!result.created) {
|
||||
// The occurrence already existed: this firing was reported twice, and the
|
||||
// claimant that made the run has recorded it. Nothing to count.
|
||||
await startsDb.release(definition.id)
|
||||
return 'duplicate'
|
||||
}
|
||||
await startsDb.started(definition.id, { at, runId: result.run.id })
|
||||
log.info('run started by a trigger', { definition: definition.id, run: result.run.id, trigger: event.triggerId })
|
||||
return 'started'
|
||||
}
|
||||
|
||||
/**
|
||||
* The trigger stream's third subscriber.
|
||||
*
|
||||
* The common answer is one indexed-by-nothing query over the ready definitions
|
||||
* returning no rows: a deployment has tens of definitions, not thousands, and a
|
||||
* JSON column cannot carry an index. If that ever changes, a side table of
|
||||
* (trigger, definition) written at publish is the fix.
|
||||
*/
|
||||
async function observe(event, { now = new Date() } = {}) {
|
||||
const summary = { definitions: 0, started: 0, ignored: 0 }
|
||||
try {
|
||||
if (!event?.triggerId || event.triggerId.startsWith(spec.CORE_EVENT_PREFIX)) return summary
|
||||
const candidates = await definitionsDb.findStartingOn(event.triggerId)
|
||||
summary.definitions = candidates.length
|
||||
if (!candidates.length) return summary
|
||||
|
||||
const at = firingTime(event.occurredAt, now)
|
||||
for (const definition of candidates) {
|
||||
try {
|
||||
const outcome = await consider(definition, event, at, now)
|
||||
if (outcome === 'started') summary.started += 1
|
||||
if (['running', 'cooldown', 'unverified'].includes(outcome)) summary.ignored += 1
|
||||
} catch (err) {
|
||||
log.error('trigger start failed', { definition: definition.id, trigger: event.triggerId, message: err.message })
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
log.error('trigger start observation failed', { trigger: event?.triggerId, message: err.message })
|
||||
}
|
||||
return summary
|
||||
}
|
||||
|
||||
/**
|
||||
* The editor's line about a definition's trigger starts, from the state row.
|
||||
* Null when no trigger has ever tried to start it.
|
||||
*/
|
||||
function describeState(row) {
|
||||
if (!row) return null
|
||||
return {
|
||||
lastStartedAt: row.last_started_at || null,
|
||||
lastRunId: row.last_run_id || null,
|
||||
ignoredRunning: Number(row.ignored_running || 0),
|
||||
ignoredCooldown: Number(row.ignored_cooldown || 0),
|
||||
ignoredUnverified: Number(row.ignored_unverified || 0),
|
||||
lastIgnoredAt: row.last_ignored_at || null,
|
||||
lastIgnoredReason: row.last_ignored_reason || null,
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = { observe, consider, describeState, firingTime, originOf, MAX_ORIGIN_BYTES }
|
||||
Reference in New Issue
Block a user