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

feat(events): the resource ledger, leases and cleanup (Phase 8)
Some checks failed
PR Checks / client-build (pull_request) Successful in 3m15s
Details
PR Checks / server-tests (pull_request) Failing after 8m21s
Details
PR Checks / bot-tests (pull_request) Successful in 11m12s
Details

Browse Source 
Event System Phase 8 (EVENTS_PLAN.md). Docs half: RunicGateway/docs#NNN.

One table, one core action, one route, one body field, and two members added to
MODULE_API 1.10.0 in place. The safety property the whole world-write half
depends on: core now remembers what a run changed in the world, and gives it
back on every terminal path.

Four decisions settled by the org lead on 2026-09-03, all as recommended:

- A lease is acquired by a new CORE action, `core.lease`. Section F puts the
  duration bound and the two-events-one-target conflict check on core's side of
  the seam, and a lease verb per module would be both re-implemented once per
  module, advisory everywhere.
- Record-before-confirm is a PLACEHOLDER keyed by the step's idempotency key. A
  spawn's ref does not exist until the module answers, so what core writes
  before the dispatch is `kind: '@step'`, `ref` = that key. If the answer never
  comes it stands, and cleanup calls revert() with the key and no resources --
  which is why section F's revert takes the key at all.
- Cleanup is one sweep over the ledger, not synthetic step rows. The
  step-shaped version costs a second retry counter beside `revert_attempts`.
- `reconcile` is declared here and TRIGGERED BY THE MODULE, through
  `ctx.events.reconcile()`. Core has no concept of the game being up, so it
  cannot decide when to ask; it asks once at its own boot.

MODULE_API stays 1.10.0. A protocol owes a bump once it has landed on `main`;
while it is on `edge` it is amended in place, so the whole module contract
reaches an author as one version they read once.

Verify

- `npm test` -- 2025 tests, 1935 pass, 89 skipped, 1 fail. That one is the
  pre-existing engagementManifest CRLF failure, in a file this branch does not
  touch (`edge` before: 1950/1876/73/1). +75 tests.
- The unique key was proved against a REAL MariaDB, because nothing else can
  prove it: whether multiple NULLs collide in a unique index, whether a STORED
  generated column is recomputed on UPDATE, and whether the SET NULL foreign key
  survives beside it are properties of the server. eventRunnerSql.test.js gained
  16 tests; 65 pass against the container. The real schema.sql was applied to a
  fresh database and to an existing one.
- Client: 362 pass, and it builds. routes:manifest and swagger -- one route
  added, none moved.

The live walk found three defects, and two of them are the phase's real finding

Driven by a throwaway `rig` module in website/modules/, deleted before commit.

1. A lease was never given back at all. `core.lease` reserves its own ledger
   row, so it never went through the ledger's dirty-marking, so a run holding
   only a lease kept `cleanup_status = 'not_required'` and the cleanup leg --
   which selected on `pending` -- never looked at it.
2. EVENT_REVERT_MAX_ATTEMPTS meant one attempt, not three. The first failing
   sweep moved the run to `incomplete`, which took it out of the leg's own scan
   for ever. The test covering the bound asserted `<= 3` and was satisfied by 1:
   a bound has two halves, and a test that only asserts the ceiling passes
   against a floor.
3. The first fix for (2) made the console lie. Spending every row's
   `revert_attempts` was a tidy way to take a `cleanup: false` run out of a
   counter-bounded scan, and the run page then rendered "3 attempts" beside
   resources nothing had ever tried. Found by opening the page.

Both (1) and (2) are the same mistake: deriving "is there anything to do" from a
summary column instead of from the rows. Neither was visible to a unit test,
because a test that calls the sweep directly never asks what would have selected
the run.

The two properties that need the process to die were walked as the plan asks.
With the module's perform() hanging, the placeholder existed while the dispatch
was in flight and nothing was named; after taskkill and a restart the reclaim
re-dispatched the same idempotency key, the retry re-used its own placeholder,
and everything was given back. Then, with the module reporting one of two
resources as no longer in force, the boot-time reconcile marked the other
`orphaned` -- never `reverted`.

This branch does NOT bump MODULE_API_VERSION, so the integration kit stays as
Phase 7 left it: red until the Phase 16 cutover re-pins ci/core-ref.json.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
claude
2026-09-03 21:19:27 -05:00
parent 57d183e921
commit fdc118166c
29 changed files with 3928 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

427
server/src/events/cleanup.js Normal file
View File

@@ -0,0 +1,427 @@
// ── Giving back what a run took ────────────────────────────────────────────
//
// EVENTS.md §C ("Cleanup is generated, never authored"), §L and its two ledger
// rules, and Phase 8 of EVENTS_PLAN.md. `events/ledger.js` is the write half;
// this is the undo half, plus the reconcile that answers "is any of it still
// there?" after something outside core restarted.
//
// **Cleanup is derived from the ledger, never authored.** An operator cannot be
// relied on to write the undo, and an aborted run never reaches the phase they
// wrote it in — so there is no cleanup phase in a spec and no `on_teardown` on an
// action. There is one function, it reads rows, and it runs on EVERY terminal
// path: completion, cancellation and abort alike.
//
// **It is not built out of `event_run_steps` rows** (org lead, 2026-09-03). The
// plan's phrase is "cleanup steps are generated from the ledger", and the
// tempting reading is a synthetic phase of real step rows so the console's
// per-step retry comes free. It is the wrong shape here for one concrete reason:
// `event_run_resources` already carries `revert_attempts` and `last_error`, so
// synthetic steps would put a second retry counter beside the first and the two
// would disagree the first time a step reverted three of its four resources.
// The manual retry the API surface promises is a route over the ledger —
// `POST /admin/events/runs/:runId/cleanup` — rather than a step control.
//
// **Where it runs from.** One place: the runner's cleanup leg, which finds
// terminal runs that still owe the world something and works their rows. Hooking
// each terminal path instead would be four call sites, three of which are inside
// a request, and none of which would survive the process dying mid-cleanup. The
// leg is ordered AFTER advance in the tick, so a run that completes in one tick
// is cleaned in the same one.
//
// **The scan's WHERE clause cost two live-walk findings, in opposite
// directions.** A run whose only resource was a LEASE never went through
// `ledger.markRunDirty` — `core.lease` reserves its own row — so its
// `cleanup_status` stayed `not_required` and the lease was never given back at
// all. And a run whose first sweep failed was moved to `incomplete` by that very
// sweep, so it was never picked up again: `MAX_REVERT_ATTEMPTS` meant ONE attempt
// rather than three. The first is why `not_required` is in the scan; the second
// is why `incomplete` is written HERE only once nothing retryable is left.
//
// **Rule 2 is what the bounds are for.** A revert that never succeeds must stay
// visible rather than cycle: `MAX_REVERT_ATTEMPTS` stops the automatic retry, the
// run reaches `completed` with `cleanup_status = 'incomplete'`, and the rows stay
// on the console with their last error. Only a human's cleanup clears the
// counter — Engagement Phase 14's rule, whose defect was a sweep that reset every
// stale row and made the ceiling unreachable for ever.
const resourcesDb = require('../model/events/eventRunResources.db')
const runsDb = require('../model/events/eventRuns.db')
const logDb = require('../model/events/eventRunLog.db')
const stepsDb = require('../model/events/eventRunSteps.db')
const registries = require('../modules/registries')
const { withDeadline } = require('./dispatch')
const log = require('../utils/logger')('events')
// How many times the automatic sweep will ask before leaving a resource for a
// human. Three, like a step's, and for the same reason: a fourth attempt against
// a shard that has answered the same way three times is not new information.
const MAX_REVERT_ATTEMPTS = Number(process.env.EVENT_REVERT_MAX_ATTEMPTS) || 3
// The bound on one revert call, when the action that made the resource is gone
// and there is no `budgetMs` to read. A restore is a round trip like any other.
const DEFAULT_REVERT_BUDGET_MS = 10_000
// How many runs one cleanup leg looks at, and how many resource groups it works
// per run. Bounds rather than targets, exactly like `RUN_BATCH`: the tick runs
// again, and an unbounded teardown is how one run's bad night stalls every other.
const CLEANUP_RUN_BATCH = Number(process.env.EVENT_CLEANUP_RUN_BATCH) || 10
const CLEANUP_GROUPS_PER_RUN = Number(process.env.EVENT_CLEANUP_GROUPS_PER_RUN) || 25
/**
* Classify one revert answer, with `dispatch.classify`'s posture: no shape a
* failure can take may read as success.
*
* The extra value here is `drifted`. It is NOT an error — the module did exactly
* what it was asked and found somebody else's value in place — so it is a third
* outcome rather than a failure with a flag, and the row it produces is the one
* §L wants surfaced beside the unreverted ones.
*/
function classifyRevert(raw, what) {
if (raw && raw.__timedOut) return { outcome: 'retry', error: raw.error }
if (raw === null || typeof raw !== 'object' || Array.isArray(raw)) {
return { outcome: 'retry', error: `${what} answered with no envelope` }
}
if (raw.ok === true) {
// §L, and the Rust wipe: "gone, and that is fine" is a successful revert. A
// module never has to distinguish "I deleted it" from "it was not there".
return { outcome: 'done', failed: Array.isArray(raw.failed) ? raw.failed.map(String) : [] }
}
if (raw.drifted === true) {
return {
outcome: 'drifted',
error: `the value is now ${JSON.stringify(raw.current)} rather than what this run applied, so it was left alone`,
}
}
return {
outcome: raw.retry === false ? 'terminal' : 'retry',
error: raw.error ? String(raw.error) : `${what} refused`,
}
}
/** Call a lease's `restore`, under a deadline, never throwing. */
async function restoreLease(row) {
const lease = registries.eventLease(row.ref)
if (!lease) {
// The module that owned it is uninstalled or failed to boot. Not a failure to
// retry away — nothing will change until an operator reinstalls it — and not
// an orphan either, because core has no idea whether the value is still
// applied. It stays unresolved with the reason on it, which is exactly what
// `cleanup_status = 'incomplete'` is for.
return { outcome: 'terminal', error: `no module registers the lease "${row.ref}"` }
}
const payload = row.payload || {}
let raw
try {
raw = await withDeadline(
() => lease.restore(payload.baseline, { expected: payload.applied, runId: row.run_id }),
DEFAULT_REVERT_BUDGET_MS,
row.ref,
)
} catch (err) {
return { outcome: 'retry', error: err.message }
}
return classifyRevert(raw, row.ref)
}
/** Call an action's `revert` over a group of its rows, under a deadline, never throwing. */
async function revertGroup(actionId, rows, idempotencyKey) {
const action = registries.eventAction(actionId)
if (!action || typeof action.revert !== 'function') {
return {
outcome: 'terminal',
error: action
? `${actionId} declares no revert()`
: `no module registers "${actionId}", so its resources cannot be given back`,
}
}
const payload = rows
.filter((r) => r.kind !== resourcesDb.STEP_KIND)
.map((r) => ({ kind: r.kind, ref: r.ref, payload: r.payload || null, memberKey: r.member_key || null }))
let raw
try {
raw = await withDeadline(
() => action.revert({ runId: rows[0].run_id, resources: payload, idempotencyKey }),
action.budgetMs || DEFAULT_REVERT_BUDGET_MS,
actionId,
)
} catch (err) {
// A module should not throw from `revert` any more than from `perform`, and
// one that does has produced a transient failure rather than a crashed sweep.
log.warn('event revert threw', { action: actionId, message: err.message })
return { outcome: 'retry', error: err.message }
}
return classifyRevert(raw, actionId)
}
/**
* Work one run's ledger once.
*
* Answers what it found and what it managed, and sets `cleanup_status` from the
* rows that are left rather than from what it did — the two differ whenever
* another writer touched the run, and the rows are the truth.
*
* `resetAttempts` is the human's flag. It is never set by the automatic leg.
*/
async function cleanupRun(run, { resetAttempts = false, actor = null } = {}) {
const summary = { attempted: 0, reverted: 0, drifted: 0, failed: 0, remaining: 0 }
if (resetAttempts) {
const cleared = await resourcesDb.resetAttempts(run.id)
await runsDb.setCleanupStatus(run.id, 'pending')
if (cleared) {
await logDb.write({
runId: run.id,
kind: 'cleanup.retry',
detail: { resources: cleared, by: actor },
})
}
}
const rows = await resourcesDb.unresolvedForRun(run.id, {
maxAttempts: resetAttempts ? null : MAX_REVERT_ATTEMPTS,
})
// **Grouped by the step that made them**, because that is what names the verb:
// the resource row records the module and the opaque names, the step records
// the action, and `revert()` takes a LIST so one round trip can give back
// twelve creatures. A lease is its own group of one — core restores it through
// the lease registry rather than through any action, which is the split §F
// draws and the reason `core.lease` needs no `revert()` of its own.
const leases = rows.filter((r) => r.kind === 'override')
const byStep = new Map()
for (const row of rows) {
if (row.kind === 'override') continue
const key = row.step_id === null ? `orphan:${row.id}` : `step:${row.step_id}`
if (!byStep.has(key)) byStep.set(key, [])
byStep.get(key).push(row)
}
const groups = [...leases.map((r) => ({ lease: r })), ...[...byStep.values()].map((rs) => ({ rows: rs }))]
for (const group of groups.slice(0, CLEANUP_GROUPS_PER_RUN)) {
if (group.lease) {
const row = group.lease
if (!(await resourcesDb.claimRevert(row.id))) continue
summary.attempted += 1
const verdict = await restoreLease(row)
await applyVerdict(run, [row], verdict, summary, row.ref)
continue
}
const rs = group.rows
// The step is what names the action and carries the idempotency key. A row
// whose step was deleted keeps the action id in its own payload, which is why
// the placeholder writes one.
const step = rs[0].step_id === null ? null : await stepsDb.getById(rs[0].step_id)
const actionId = step?.action_id || rs[0].payload?.action || null
if (!actionId) {
await noteUnrevertable(run, rs, 'nothing records which action created this', summary)
continue
}
const claimed = []
for (const row of rs) if (await resourcesDb.claimRevert(row.id)) claimed.push(row)
if (!claimed.length) continue
summary.attempted += claimed.length
const verdict = await revertGroup(actionId, claimed, step?.idempotency_key || rs[0].ref)
await applyVerdict(run, claimed, verdict, summary, actionId)
}
summary.remaining = await resourcesDb.unresolvedCount(run.id)
// **`incomplete` means "finished with, and not finished"**, so it is written
// only once there is nothing left this sweep will try. Writing it after the
// FIRST failure — which is what the first draft did — took the run straight out
// of the leg's own scan, and `MAX_REVERT_ATTEMPTS` quietly meant one attempt
// rather than three. Found by the live walk, watching `revert_attempts` sit at
// 1 through half a minute of ticks.
const retryable = await resourcesDb.unresolvedForRun(run.id, { maxAttempts: MAX_REVERT_ATTEMPTS })
const status = summary.remaining === 0 ? 'complete' : retryable.length ? 'pending' : 'incomplete'
await runsDb.setCleanupStatus(run.id, status)
if (summary.attempted > 0) {
await logDb.write({
runId: run.id,
kind: 'cleanup.swept',
detail: { ...summary, by: actor },
})
}
return summary
}
/** Write one verdict across the rows it covers, and count it. */
async function applyVerdict(run, rows, verdict, summary, what) {
for (const row of rows) {
if (verdict.outcome === 'done' && !(verdict.failed || []).includes(row.ref)) {
await resourcesDb.markReverted(row.id)
summary.reverted += 1
continue
}
if (verdict.outcome === 'drifted') {
await resourcesDb.failRevert(row.id, verdict.error, 'drifted')
summary.drifted += 1
continue
}
const error =
verdict.outcome === 'done'
? `${what} could not give "${row.ref}" back`
: verdict.error
await resourcesDb.failRevert(row.id, error, 'confirmed')
summary.failed += 1
}
await logDb.write({
runId: run.id,
kind: verdict.outcome === 'done' ? 'cleanup.reverted' : 'cleanup.failed',
detail: {
what,
outcome: verdict.outcome,
resources: rows.map((r) => `${r.kind}:${r.ref}`),
...(verdict.error ? { error: verdict.error } : {}),
},
})
}
/** A group core cannot even name a verb for. Counted as failed, and said once. */
async function noteUnrevertable(run, rows, reason, summary) {
for (const row of rows) {
if (!(await resourcesDb.claimRevert(row.id))) continue
await resourcesDb.failRevert(row.id, reason, 'confirmed')
summary.attempted += 1
summary.failed += 1
}
await logDb.write({
runId: run.id,
kind: 'cleanup.failed',
detail: { what: null, outcome: 'terminal', resources: rows.map((r) => `${r.kind}:${r.ref}`), error: reason },
})
}
/**
* The cleanup leg of the tick: every TERMINAL run with something left to give
* back.
*
* Terminal only. A run still in flight has a ledger that is still growing, and
* reverting a resource the next step is about to use would be core undoing an
* event while it is happening.
*/
async function sweep() {
// The ceiling goes INTO the query, so a run whose rows are all spent is not
// selected, worked over and found to have nothing to do on every tick for the
// rest of its life. It is also what excludes a run an admin cancelled without
// cleanup, whose counters were spent deliberately.
const candidates = await resourcesDb.runsNeedingCleanup(CLEANUP_RUN_BATCH, MAX_REVERT_ATTEMPTS)
let swept = 0
for (const candidate of candidates) {
if (!runsDb.TERMINAL.includes(candidate.status)) continue
try {
await cleanupRun(candidate)
swept += 1
} catch (err) {
log.error('event cleanup failed', { run: candidate.id, message: err.message })
}
}
return swept
}
/**
* Ask one module which of its ledgered resources the game still has (§L, and
* §N7's "the shard stays stateless about events").
*
* **Core cannot know when to ask**, and that is not an omission: §F says core has
* no concept of the game being up, because a module with six sidecars cannot
* answer that question in the singular. So the module triggers this, through
* `ctx.events.reconcile()`, when it sees its own reconnect — module-uo already
* watches `bootId` for exactly that. Core also asks once at boot, for its own
* restart.
*
* **A resource the module no longer has becomes `orphaned`, never `reverted`.**
* Reverting it would be core recording that it put something back when what
* actually happened is that the thing vanished while nobody was looking, and the
* two are different sentences to the operator reading the console afterwards.
*
* A module with no `reconcile()` on the action is not broken: core keeps
* believing its own ledger, which is precisely the behaviour before this phase.
*/
async function reconcileModule(owner) {
const rows = await resourcesDb.liveForModule(owner)
const summary = { asked: 0, inForce: 0, orphaned: 0, unanswered: 0 }
if (!rows.length) return summary
const byStep = new Map()
for (const row of rows) {
if (row.kind === resourcesDb.STEP_KIND) continue // nothing to ask about yet
const key = row.step_id === null ? `orphan:${row.id}` : `step:${row.step_id}`
if (!byStep.has(key)) byStep.set(key, [])
byStep.get(key).push(row)
}
for (const group of byStep.values()) {
const step = group[0].step_id === null ? null : await stepsDb.getById(group[0].step_id)
const actionId = step?.action_id || group[0].payload?.action || null
const action = actionId ? registries.eventAction(actionId) : null
if (!action || typeof action.reconcile !== 'function') {
summary.unanswered += group.length
continue
}
summary.asked += group.length
let raw
try {
raw = await withDeadline(
() =>
action.reconcile({
runId: group[0].run_id,
resources: group.map((r) => ({ kind: r.kind, ref: r.ref, payload: r.payload || null })),
}),
action.budgetMs || DEFAULT_REVERT_BUDGET_MS,
actionId,
)
} catch (err) {
log.warn('event reconcile threw', { action: actionId, message: err.message })
raw = null
}
// Same posture as everywhere else: nothing that is not an explicit answer
// counts as one. A module that could not answer leaves the ledger alone,
// because "I do not know" must never be read as "it is gone".
if (!raw || raw.__timedOut || raw.ok !== true || !Array.isArray(raw.inForce)) {
summary.unanswered += group.length
continue
}
const held = new Set(raw.inForce.map(String))
for (const row of group) {
if (held.has(row.ref)) {
summary.inForce += 1
continue
}
await resourcesDb.markOrphaned(row.id, 'the module reports this is no longer in force')
summary.orphaned += 1
await logDb.write({
runId: row.run_id,
kind: 'resource.orphaned',
detail: { module: owner, resource: `${row.kind}:${row.ref}`, action: actionId },
})
}
}
return summary
}
/** Ask every module that owns a live row. Core's own boot-time sweep. */
async function reconcileAll() {
const owners = await resourcesDb.modulesWithLiveRows()
const out = {}
for (const owner of owners) {
try {
out[owner] = await reconcileModule(owner)
} catch (err) {
log.error('event reconcile failed', { module: owner, message: err.message })
}
}
return out
}
module.exports = {
MAX_REVERT_ATTEMPTS,
classifyRevert,
cleanupRun,
sweep,
reconcileModule,
reconcileAll,
}

221
server/src/events/ledger.js Normal file
View File

@@ -0,0 +1,221 @@
// ── Recording what a run changed in the world ──────────────────────────────
//
// EVENTS.md §D and §L, and Phase 8 of EVENTS_PLAN.md. The write half of the
// resource ledger; `events/cleanup.js` is the read-and-undo half.
//
// **Rule 1 is the whole reason this file is not two lines inside `drainStep`.**
// A resource is recorded BEFORE it is confirmed. The obstacle is that a spawn's
// serial does not exist until the module answers, so there is nothing to write a
// row about yet — which is why what goes in before the dispatch is a PLACEHOLDER
// keyed by the step's idempotency key rather than by the object:
//
// pre-dispatch INSERT pending { kind: '@step', ref: <idempotency key> }
// answer INSERT confirmed { kind: 'creature', ref: '0x40001234' } × n
// resolve the placeholder
// ack lost the placeholder is still `pending`
// cleanup revert({ idempotencyKey, resources: [] })
//
// That last line is why §F's `revert({ runId, resources, idempotencyKey })` takes
// the key at all. A module that half-ran and never answered is reachable by its
// key and by nothing else, and Phase 11's plugin-side key ledger is what makes
// answering it exact. Until then the contract is still honest, because §L
// requires reverting something that does not exist to be a SUCCESS.
//
// **Recording is idempotent, and the database is what makes it so.** A retry
// re-dispatches the same idempotency key, and a module that answers with the same
// resources twice must not produce two rows. `uq_evres_target` refuses the second
// insert, and this file reads that refusal as "already recorded" rather than as an
// error — the same posture `materialisePhase`'s INSERT IGNORE takes.
//
// **A lease does not use the placeholder.** Its target is knowable before the
// dispatch — it is the lease id the step names — so `core.lease` reserves the
// real row first, which is both a stronger form of rule 1 and the only place the
// two-events-one-target refusal can happen before the world has been written to.
const resourcesDb = require('../model/events/eventRunResources.db')
const runsDb = require('../model/events/eventRuns.db')
const registries = require('../modules/registries')
const log = require('../utils/logger')('events')
// Which reversible classes get a pre-dispatch placeholder. `none` is gone once
// done and `self` undoes itself, so neither has anything core could come back
// for; `override` reserves its own target instead (see the header). That leaves
// `ledger` — the class that declares `revert()`, which is exactly the class whose
// refs core cannot know until the module speaks.
const PLACEHOLDER_CLASSES = ['ledger']
// A resource `kind` may be anything a module likes except core's own reserved
// one. Bounded to the column, and refused rather than truncated: a truncated ref
// is a cleanup call naming the wrong object.
const MAX_KIND = 64
const MAX_REF = 190
const MAX_MEMBER_KEY = 190
/** Does this action produce anything core will have to come back for? */
const ledgers = (action) => action && (action.reversible === 'ledger' || action.reversible === 'override')
/**
* Turn one entry of a module's `resources` array into a row, or say why not.
*
* Every failure here is the module's mistake rather than the world's, so none of
* them is a retry: a badly shaped resource will be just as badly shaped on the
* second attempt. They are logged and dropped, and the step still counts as done
* — because it IS done; something happened in the world, and refusing to record
* it would be the one outcome worse than recording it imperfectly.
*/
function normalise(entry, actionId) {
if (!entry || typeof entry !== 'object' || Array.isArray(entry)) {
return { ok: false, reason: `${actionId} reported a resource that is not an object` }
}
const kind = String(entry.kind || '')
const ref = String(entry.ref === undefined || entry.ref === null ? '' : entry.ref)
if (!kind || kind.length > MAX_KIND) {
return { ok: false, reason: `${actionId} reported a resource with a bad kind "${entry.kind}"` }
}
if (kind === resourcesDb.STEP_KIND) {
// Core's own. A module that could write one would be a module that could make
// its own step's placeholder look resolved.
return { ok: false, reason: `${actionId} reported a resource of the reserved kind "${kind}"` }
}
if (!ref || ref.length > MAX_REF) {
return { ok: false, reason: `${actionId} reported a resource with a bad ref "${entry.ref}"` }
}
const memberKey = entry.memberKey === undefined || entry.memberKey === null ? null : String(entry.memberKey)
if (memberKey !== null && memberKey.length > MAX_MEMBER_KEY) {
return { ok: false, reason: `${actionId} reported a resource with an over-long memberKey` }
}
let leaseUntil = null
if (entry.until !== undefined && entry.until !== null) {
const at = new Date(entry.until)
if (Number.isNaN(at.getTime())) {
return { ok: false, reason: `${actionId} reported a resource with a bad until "${entry.until}"` }
}
leaseUntil = at
}
// A borrowed value must name a lease core knows how to give back. Core restores
// an `override` through the lease registry — that is the split §F draws — so a
// ref naming nothing registered is a resource core would be recording with no
// way to undo it, which is the promise rule 2 exists to stop core making.
if (kind === 'override' && !registries.eventLease(ref)) {
return { ok: false, reason: `${actionId} reported a lease "${ref}" no module registers` }
}
return {
ok: true,
row: {
kind,
ref,
payload: entry.payload === undefined ? null : entry.payload,
leaseUntil,
memberKey,
},
}
}
/**
* Write the pre-dispatch placeholder for a step that is about to change the
* world. Answers the row id, or null when this action ledgers nothing.
*
* **A duplicate is not a failure.** A retry re-uses its step's idempotency key, so
* the second attempt's placeholder collides with the first's — and finding it
* already there is the correct answer, not an error. The existing row is reused.
*/
async function reserveStep(run, step, action) {
if (!ledgers(action) || !PLACEHOLDER_CLASSES.includes(action.reversible)) return null
const owner = action.owner || 'core'
const reserved = await resourcesDb.reserve({
runId: run.id,
stepId: step.id,
owner,
kind: resourcesDb.STEP_KIND,
ref: step.idempotency_key,
payload: { action: action.id, phase: step.phase, seq: step.seq },
})
if (reserved.ok) {
await markRunDirty(run.id)
return reserved.id
}
// The only way a '@step' row collides is with this step's own earlier attempt,
// because an idempotency key is minted once per step and never varies by
// attempt (§E). Reuse it.
const existing = await resourcesDb.findByTarget(owner, resourcesDb.STEP_KIND, step.idempotency_key)
return existing ? existing.id : null
}
/**
* Record what a module said it made, and close out the placeholder.
*
* Answers `{ recorded, rejected }` — how many rows went in, and the reasons any
* entry was dropped. Never throws: a step that changed the world has changed it,
* and a ledger that threw would turn a bookkeeping problem into a failed step and
* then into a retry of a world write that already happened.
*/
async function recordAnswer({ run, step, action, placeholderId, resources }) {
const out = { recorded: 0, rejected: [] }
if (!ledgers(action)) return out
const owner = action.owner || 'core'
const list = Array.isArray(resources) ? resources : []
for (const entry of list) {
const parsed = normalise(entry, action.id)
if (!parsed.ok) {
out.rejected.push(parsed.reason)
log.warn('event resource rejected', { run: run.id, step: step.id, reason: parsed.reason })
continue
}
try {
const reserved = await resourcesDb.reserve({
runId: run.id,
stepId: step.id,
owner,
kind: parsed.row.kind,
ref: parsed.row.ref,
payload: parsed.row.payload,
leaseUntil: parsed.row.leaseUntil,
memberKey: parsed.row.memberKey,
})
if (!reserved.ok) {
// Already ledgered — by this step's own earlier attempt, or (a module bug
// rather than a race) by another run that still holds the same target.
// Either way there is a live row for it and a second would be the double
// cleanup the unique key exists to prevent.
if (reserved.holder && reserved.holder.run_id !== run.id) {
out.rejected.push(`${parsed.row.kind} "${parsed.row.ref}" is already held by run ${reserved.holder.run_id}`)
}
continue
}
await resourcesDb.confirm(reserved.id)
out.recorded += 1
} catch (err) {
// Bookkeeping must not become the step's control flow.
out.rejected.push(err.message)
log.error('event resource insert failed', { run: run.id, step: step.id, message: err.message })
}
}
if (out.recorded > 0) await markRunDirty(run.id)
// The placeholder's job is over the moment the real rows exist. It is resolved
// even when the module reported nothing at all — an action that ledgers and
// then answers `ok` with an empty list is saying "I made nothing", and holding
// its placeholder open would make cleanup call `revert()` for a step that has
// nothing to give back on every terminal path for ever.
if (placeholderId) await resourcesDb.resolvePlaceholder(placeholderId)
return out
}
/**
* There is now something to clean up. Idempotent and guarded, so it can never
* walk a run back from `complete` or `incomplete` to `pending` — only a human's
* cleanup does that, and it does it deliberately.
*/
async function markRunDirty(runId) {
await runsDb.setCleanupStatus(runId, 'pending', ['not_required'])
}
module.exports = { ledgers, normalise, reserveStep, recordAnswer, markRunDirty, PLACEHOLDER_CLASSES }