From a50e0e2a5a6d6333f0ff4edee143e2496a7fe94e Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 13:18:27 -0700 Subject: [PATCH 01/11] feat: retry and redrive dead effects and broadcasts `DeadLetterManager` covered message dead letters only. A dead effect or broadcast reached `status = 'dead'` and nothing brought it back, so the recovery path was an operator writing an UPDATE against a runtime table. Retry was also one row at a time, and an incident produces dead rows in the hundreds. `runtime.deadLetters` keeps its message meaning and answers `effects` and `broadcasts`, so the kind rides on the receiver. `retry` returns a dead row to pending, keeps its id, and acts only on a dead row, so a second press cannot double-enqueue and cannot take a row from a worker. `redrive` opens a durable task and returns at once. A unique index on the active scope makes it idempotent in the database rather than in a read followed by a write. `runtime.run()` advances one bounded batch per pass, so a redrive never holds a transaction longer than one batch. A redrive moves what was already dead when it started. Without that bound, a row that fails again lands back in the same scope and a task whose handler is still broken would move it forever. The bound needs a failure stamp, so schema version 11 adds `failed_at_ms` to effects and broadcasts and backfills it from the availability stamp. Every retry and every task transition writes one row to `solid_objects_administration_events` under the identity that asked for it, through a new `administrationIdentity` option. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 23 ++ docs/api.md | 12 + docs/configuration.md | 21 +- docs/operations.md | 80 +++++++ docs/parity.md | 13 ++ src/configuration.ts | 23 ++ src/dead-letter-scopes.ts | 227 +++++++++++++++++++ src/dead-letters.ts | 20 ++ src/doctor.ts | 16 +- src/index.ts | 16 ++ src/platform/uuid.ts | 8 + src/redrive.ts | 325 ++++++++++++++++++++++++++ src/repository.ts | 23 +- src/runtime.ts | 74 +++++- src/schema.ts | 85 ++++++- test/dead-letter-scopes.test.ts | 198 ++++++++++++++++ test/dead-letters.test.ts | 8 +- test/doctor.test.ts | 2 +- test/instance-retention.test.ts | 4 +- test/redrive.test.ts | 388 ++++++++++++++++++++++++++++++++ 20 files changed, 1545 insertions(+), 21 deletions(-) create mode 100644 src/dead-letter-scopes.ts create mode 100644 src/redrive.ts create mode 100644 test/dead-letter-scopes.test.ts create mode 100644 test/redrive.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 8feec93..0618aae 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,29 @@ ## Unreleased +- Retry a dead effect or broadcast. `runtime.deadLetters` keeps its message + meaning and answers `effects` and `broadcasts`, so the kind rides on the + receiver. `retry` returns a dead row to pending with a zero attempt count and + no claim, keeps its id so a deduplicating handler sees the same key, and acts + only on a dead row, so a second press cannot double-enqueue. +- Redrive a whole scope. `redrive` opens a durable task, returns at once, and is + idempotent over its scope and filters, which a dashboard button needs. A + unique index on the active scope enforces that in the database. + `runtime.run()` advances one bounded batch per pass, so a redrive never holds + a transaction longer than one batch. `runtime.redrives` reads tasks back, and + `task.cancel()` stops one and leaves the rows it already moved. A redrive + moves what was dead when it started, so a still-broken handler cannot make it + run forever. +- Record who pressed what. Every retry and every redrive transition writes one + row to `solid_objects_administration_events`. The identity comes from the + authorization context through a new `administrationIdentity` option. +- Add `redriveBatchSize`, which defaults to 100, and + `redriveBatchPauseMilliseconds`, which defaults to 50. +- Add schema version 11: two tables, `solid_objects_administration_events` and + `solid_objects_redrives`, and a `failed_at_ms` stamp on effects and + broadcasts, which a redrive filters on. Existing dead rows take their + availability stamp. + - Select a wake-up adapter automatically. `wakeUp` now takes a name or an adapter, as `config.cache_store` does in Rails, and defaults to `"automatic"`. Selection prefers `SOLID_OBJECTS_REDIS_URL`, then PostgreSQL diff --git a/docs/api.md b/docs/api.md index 582e7a6..c47a312 100644 --- a/docs/api.md +++ b/docs/api.md @@ -551,6 +551,18 @@ Every manager below is available as a property on `SolidObjectsRuntime`; the class and result types are also exported for integration typing. - `runtime.deadLetters` / `DeadLetterManager`: `all()` and idempotent `retry()`. +- `runtime.deadLetters.effects` and `runtime.deadLetters.broadcasts`: a + `DeadLetterScope` for one `DeadLetterKind`. `all()` lists its dead rows as + `DeadRow` values, `retry(id)` returns one to pending, and `redrive(options)` + moves a whole scope. `RedriveOptions` names `actorType`, `failedAfter`, and + `limit`, which become the `RedriveFilters` the task records. `UnknownDeadRow` + reports an id that does not exist. +- `runtime.redrives` / `RedriveManager`: `find(id)`, `all({ status })`, + `cancel(id)`, and `advance()`, which moves one bounded batch. A `RedriveTask` + carries its id, kind, filters, `RedriveStatus`, `moved`, `remaining`, + `startedAt`, `finishedAt`, and its own `cancel()`. `RedriveScheduler` is the + component that advances tasks inside `runtime.run()`. `UnknownRedrive` and + `RedriveNotStarted` report a missing task and a task that could not open. - `runtime.reminders` / `ReminderManager`: cursor-paginated `all()` and idempotent paused-alarm `resume()`. - `runtime.processes` / `ProcessManager`: immutable role `all()` and stale-owner diff --git a/docs/configuration.md b/docs/configuration.md index f36a519..78ec0c6 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -78,15 +78,18 @@ polling remains the correctness path. ## Retention and cleanup -| Option | Default | Contract | -| ---------------------------------------- | ----------: | ---------------------------------------------------------------- | -| `retentionIntervalMilliseconds` | `3_600_000` | Non-negative automatic history-pruning cadence; `0` disables it. | -| `deadProcessCleanupIntervalMilliseconds` | `60_000` | Non-negative stale-owner recovery cadence; `0` disables it. | -| `messageRetentionMilliseconds` | 30 days | Positive default completed-message retention. | -| `messageRetentionByActorType` | `{}` | Positive per-type message-retention overrides. | -| `instanceRetentionByActorType` | `{}` | Positive per-type instance-expiration opt-ins. | -| `processRetentionMilliseconds` | 7 days | Positive stopped-process retention. | -| `pruneBatchSize` | `1_000` | Positive integer maximum rows rechecked per pruning transaction. | +| Option | Default | Contract | +| ---------------------------------------- | ----------------: | ---------------------------------------------------------------- | +| `retentionIntervalMilliseconds` | `3_600_000` | Non-negative automatic history-pruning cadence; `0` disables it. | +| `deadProcessCleanupIntervalMilliseconds` | `60_000` | Non-negative stale-owner recovery cadence; `0` disables it. | +| `messageRetentionMilliseconds` | 30 days | Positive default completed-message retention. | +| `messageRetentionByActorType` | `{}` | Positive per-type message-retention overrides. | +| `instanceRetentionByActorType` | `{}` | Positive per-type instance-expiration opt-ins. | +| `processRetentionMilliseconds` | 7 days | Positive stopped-process retention. | +| `pruneBatchSize` | `1_000` | Positive integer maximum rows rechecked per pruning transaction. | +| `redriveBatchSize` | `100` | Positive integer rows a redrive moves per transaction. | +| `redriveBatchPauseMilliseconds` | `50` | Non-negative pause between redrive batches. | +| `administrationIdentity` | `String(context)` | Names the operator recorded in an administration event. | Automatic retention prunes messages and stopped process records. Actor instance expiration remains an explicit administration action even when a type diff --git a/docs/operations.md b/docs/operations.md index bfb00d1..0fb7bf0 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -227,3 +227,83 @@ be resumed. Scheduling an existing actor operation for a different time emits `solid_objects.reminder.replaced` after the fenced actor commit. Scheduling it for the same time emits nothing. + +## Dead letters, retry, and redrive + +A message that exhausts its attempts becomes a dead letter. An effect or a +broadcast that exhausts its attempts stays in its own table with +`status = 'dead'`. All three are read and retried through one receiver, which +carries the kind: + +```ts +await runtime.deadLetters.all({ authorizationContext }) +await runtime.deadLetters.retry(deadLetterId, { authorizationContext }) + +await runtime.deadLetters.effects.all({ authorizationContext }) +await runtime.deadLetters.effects.retry(effectId, { authorizationContext }) +await runtime.deadLetters.broadcasts.retry(broadcastId, { authorizationContext }) +``` + +An effect or broadcast retry returns the row to pending with a zero attempt +count, no claim, and immediate availability, and keeps its id, so a handler that +deduplicates on the effect id still sees the same key. An effect is +at-least-once by contract, so a retried effect can run twice. + +Retry acts only on a dead row. A row that is pending, processing, or completed +comes back unchanged, so pressing a button twice cannot double-enqueue and +cannot take a row away from a worker that holds it. + +An incident produces dead rows in the hundreds, so a scope also answers +`redrive`: + +```ts +const task = await runtime.deadLetters.effects.redrive({ + actorType: "payments", + failedAfter: new Date(Date.now() - 6 * 60 * 60 * 1000), + limit: 5_000, + authorizationContext, +}) + +await task.cancel({ authorizationContext }) +``` + +`redrive` returns at once. The task is durable, and `runtime.run()` advances one +bounded batch per pass, so a redrive of thousands of rows never holds a +transaction longer than one batch. `redriveBatchSize` defaults to 100 and +`redriveBatchPauseMilliseconds` to 50. + +A redrive moves the rows that were already dead when it started. A row that +fails again lands back in the same scope, and without that bound a task whose +handler is still broken would move it forever. + +A redrive is idempotent over its scope and its filters. Starting the same one +while it runs returns the running task rather than a second one, which a +dashboard button an operator can press twice needs. A different scope or a +different filter starts its own task, and the same scope can be redriven again +once the first task finishes. + +Read tasks back with `runtime.redrives`: + +```ts +await runtime.redrives.find(task.id, { authorizationContext }) +await runtime.redrives.all({ status: "running", authorizationContext }) +``` + +A running task reports what is left to move rather than a stored estimate, +because rows die and are retried while it runs. + +Retry, redrive, and cancel each go through `authorizeAdministration` under their +own resource name: `dead_letters`, `effect_dead_letters`, +`broadcast_dead_letters`, and `redrives`. Every retry and every task transition +writes one row to `solid_objects_administration_events`, holding the action, the +kind, the subject, the identity, and when it happened. The identity comes from +`administrationIdentity`, which receives the authorization context the caller +passed and defaults to its `String` form. A refused caller writes nothing. + +The Durable Objects engine keeps its own message and outbox tables inside each +object, so these scopes and redrive cover the SQL backends. `deadLetters` there +reports its own dead rows as it did before. + +Automatic redrive on a schedule is deliberately absent. A dead row means a +person decided something, and these APIs give that person an alternative to an +`UPDATE` against a runtime table. diff --git a/docs/parity.md b/docs/parity.md index ea4d443..4136c80 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -129,6 +129,19 @@ edit of the settings makes impossible is reported once here and raised in Ruby; both runtimes refuse an unknown name when the configuration is built, so this only reaches code that changes the setting afterwards. +## Dead letters, retry, and redrive + +Both runtimes scope dead letters by kind through one receiver, retry a dead +effect or broadcast, redrive a whole scope as a durable and idempotent task, +advance it in bounded batches, and write one administration event per retry and +per task transition. The filters, the resource names, and the audit shape match. + +Two details differ. This runtime filters on a `failed_at_ms` stamp that schema +version 11 adds, while Ruby filters on the `updated_at` column Active Record +already maintains. The Durable Objects engine keeps its own message and outbox +tables inside each object, so these scopes cover the SQL backends here; its +`deadLetters` call is unchanged. + ## Realtime and browser behavior | Capability | Status | TypeScript shape or remaining work | diff --git a/src/configuration.ts b/src/configuration.ts index b55ab38..4e36d1e 100644 --- a/src/configuration.ts +++ b/src/configuration.ts @@ -82,6 +82,9 @@ export interface SolidObjectsConfiguration { instrumentation?: (event: InstrumentationEvent) => void broadcast?: (event: BroadcastEvent) => Promise wakeUp?: WakeUpSetting + redriveBatchSize?: number + redriveBatchPauseMilliseconds?: number + administrationIdentity?: (authorizationContext: unknown) => string | null | Promise } export interface BroadcastEvent { @@ -156,6 +159,14 @@ export function buildSettings(configuration: SolidObjectsConfiguration): Runtime }), processRetentionMilliseconds: configuration.processRetentionMilliseconds ?? 7 * 86_400_000, pruneBatchSize: configuration.pruneBatchSize ?? 1_000, + redriveBatchSize: configuration.redriveBatchSize ?? 100, + redriveBatchPauseMilliseconds: configuration.redriveBatchPauseMilliseconds ?? 50, + administrationIdentity: + configuration.administrationIdentity ?? + ((authorizationContext) => + authorizationContext === undefined || authorizationContext === null + ? null + : String(authorizationContext)), logger: configuration.logger ?? consoleLogger, wakeUp: configuration.wakeUp ?? "automatic", authorizeMessage: configuration.authorizeMessage ?? (() => false), @@ -280,6 +291,18 @@ function validateSettings(settings: RuntimeSettings): void { throw new TypeError(`${name} must be a non-negative integer`) } + if (!Number.isSafeInteger(settings.redriveBatchSize) || settings.redriveBatchSize < 1) { + throw new TypeError("redriveBatchSize must be a positive safe integer") + } + if ( + !Number.isFinite(settings.redriveBatchPauseMilliseconds) || + settings.redriveBatchPauseMilliseconds < 0 + ) { + throw new TypeError("redriveBatchPauseMilliseconds must not be negative") + } + if (typeof settings.administrationIdentity !== "function") { + throw new TypeError("administrationIdentity must be a function") + } if (!Number.isSafeInteger(settings.pruneBatchSize) || settings.pruneBatchSize < 1) { throw new TypeError("pruneBatchSize must be a positive safe integer") } diff --git a/src/dead-letter-scopes.ts b/src/dead-letter-scopes.ts new file mode 100644 index 0000000..503aa8d --- /dev/null +++ b/src/dead-letter-scopes.ts @@ -0,0 +1,227 @@ +import type { DatabaseConnection } from "./database/types.js" +import type { RedriveTask } from "./redrive.js" +import type { SolidObjectsRuntime } from "./runtime.js" +import { normalizeJson } from "./serialization.js" +import type { AdministrationOptions, DeepReadonly, JsonObject } from "./types.js" + +export type DeadLetterKind = "effect" | "broadcast" + +export interface DeadRow { + readonly id: string + readonly kind: DeadLetterKind + readonly actorType: string + readonly actorId: string + readonly status: string + readonly attemptCount: number + readonly availableAt: Date + readonly failedAt: Date | null + readonly error: DeepReadonly | null +} + +export interface RedriveFilters { + readonly actorType: string | null + readonly failedAfter: number | null + readonly limit: number | null +} + +export interface RedriveOptions extends AdministrationOptions { + actorType?: string + failedAfter?: Date + limit?: number +} + +interface DeadRowShape { + id: string + status: string + attempt_count: number | bigint + available_at_ms: number | bigint + failed_at_ms: number | bigint | null + error: string | null + actor_type: string + actor_id: string +} + +const TABLES: Readonly> = Object.freeze({ + effect: "effects", + broadcast: "broadcasts", +}) + +const RESOURCES: Readonly> = Object.freeze({ + effect: "effect_dead_letters", + broadcast: "broadcast_dead_letters", +}) + +export class DeadLetterScope { + constructor( + private readonly runtime: SolidObjectsRuntime, + readonly kind: DeadLetterKind, + ) {} + + get resource(): string { + return RESOURCES[this.kind] + } + + async all(options: AdministrationOptions = {}): Promise { + await this.authorize({ action: "inspect", options }) + const rows = await this.runtime.settings.database.connection((connection) => + this.matching({ connection, filters: emptyFilters() }), + ) + return Object.freeze(rows.map((row) => this.deadRow(row))) + } + + async retry(id: string, options: AdministrationOptions = {}): Promise { + await this.authorize({ action: "retry", options, resourceId: id }) + await this.runtime.recordAdministrationEvent({ + action: "dead_letter.retry", + kind: this.kind, + subjectId: id, + authorizationContext: options.authorizationContext, + }) + const row = await this.runtime.settings.database.transaction(async (connection) => { + const found = await this.find({ connection, id }) + if (found.status === "dead") await this.revive({ connection, identifiers: [id] }) + return await this.find({ connection, id }) + }) + this.runtime.wakeUpAfterRevival(this.kind) + return this.deadRow(row) + } + + async redrive(options: RedriveOptions = {}): Promise { + await this.authorize({ action: "redrive", options }) + return await this.runtime.redrives.start({ + kind: this.kind, + filters: { + actorType: options.actorType ?? null, + failedAfter: options.failedAfter ? options.failedAfter.getTime() : null, + limit: options.limit ?? null, + }, + authorizationContext: options.authorizationContext, + }) + } + + async count(input: { + connection: DatabaseConnection + filters: RedriveFilters + deadBefore?: number + }): Promise { + const { where, parameters } = this.conditions(input.filters, input.deadBefore) + const row = await input.connection.get<{ total: number | bigint }>( + `SELECT COUNT(*) AS total FROM ${this.table()} AS dead + JOIN ${this.runtime.repository.table("instances")} AS owner ON owner.id = dead.instance_id + ${where}`, + parameters, + ) + return Number(row?.total ?? 0) + } + + async matching(input: { + connection: DatabaseConnection + filters: RedriveFilters + limit?: number + deadBefore?: number + }): Promise { + const { where, parameters } = this.conditions(input.filters, input.deadBefore) + const limit = input.limit === undefined ? "" : ` LIMIT ${Number(input.limit)}` + return await input.connection.all( + `SELECT dead.id, dead.status, dead.attempt_count, dead.available_at_ms, + dead.failed_at_ms, dead.error, owner.actor_type, owner.actor_id + FROM ${this.table()} AS dead + JOIN ${this.runtime.repository.table("instances")} AS owner ON owner.id = dead.instance_id + ${where} + ORDER BY dead.failed_at_ms DESC, dead.id DESC${limit}`, + parameters, + ) + } + + async revive(input: { + connection: DatabaseConnection + identifiers: readonly string[] + }): Promise { + if (input.identifiers.length === 0) return 0 + const placeholders = input.identifiers.map(() => "?").join(", ") + const now = await input.connection.nowMilliseconds() + const result = await input.connection.run( + `UPDATE ${this.table()} + SET status = 'pending', attempt_count = 0, available_at_ms = ?, claimed_by = NULL + WHERE status = 'dead' AND id IN (${placeholders})`, + [now, ...input.identifiers], + ) + return result.changes + } + + private table(): string { + return this.runtime.repository.table(TABLES[this.kind]) + } + + // A redrive moves what was already dead when it started. Without that bound a + // row that fails again lands back in the same scope, and a task whose handler + // is still broken would move it forever. + private conditions( + filters: RedriveFilters, + deadBefore?: number, + ): { where: string; parameters: unknown[] } { + const clauses = ["dead.status = 'dead'"] + const parameters: unknown[] = [] + if (deadBefore !== undefined) { + clauses.push("dead.failed_at_ms <= ?") + parameters.push(deadBefore) + } + if (filters.actorType !== null) { + clauses.push("owner.actor_type = ?") + parameters.push(filters.actorType) + } + if (filters.failedAfter !== null) { + clauses.push("dead.failed_at_ms >= ?") + parameters.push(filters.failedAfter) + } + return { where: `WHERE ${clauses.join(" AND ")}`, parameters } + } + + private async find(input: { connection: DatabaseConnection; id: string }): Promise { + const row = await input.connection.get( + `SELECT dead.id, dead.status, dead.attempt_count, dead.available_at_ms, + dead.failed_at_ms, dead.error, owner.actor_type, owner.actor_id + FROM ${this.table()} AS dead + JOIN ${this.runtime.repository.table("instances")} AS owner ON owner.id = dead.instance_id + WHERE dead.id = ?`, + [input.id], + ) + if (!row) throw new UnknownDeadRow(`unknown ${this.kind} ${input.id}`) + return row + } + + private deadRow(row: DeadRowShape): DeadRow { + return Object.freeze({ + id: row.id, + kind: this.kind, + actorType: row.actor_type, + actorId: row.actor_id, + status: row.status, + attemptCount: Number(row.attempt_count), + availableAt: new Date(Number(row.available_at_ms)), + failedAt: row.failed_at_ms === null ? null : new Date(Number(row.failed_at_ms)), + error: row.error === null ? null : (normalizeJson(JSON.parse(row.error)) as JsonObject), + }) + } + + private async authorize(input: { + action: string + options: AdministrationOptions + resourceId?: string + }): Promise { + await this.runtime.authorizeAdministration({ + action: input.action, + resource: this.resource, + ...(input.resourceId === undefined ? {} : { resourceId: input.resourceId }), + authorizationContext: input.options.authorizationContext, + }) + } +} + +export function emptyFilters(): RedriveFilters { + return Object.freeze({ actorType: null, failedAfter: null, limit: null }) +} + +export class UnknownDeadRow extends Error { + override readonly name = "UnknownDeadRow" +} diff --git a/src/dead-letters.ts b/src/dead-letters.ts index e58a468..bfc5240 100644 --- a/src/dead-letters.ts +++ b/src/dead-letters.ts @@ -1,3 +1,4 @@ +import { DeadLetterScope, type DeadLetterKind } from "./dead-letter-scopes.js" import type { MessageReference } from "./reference.js" import type { SolidObjectsRuntime } from "./runtime.js" import type { AdministrationOptions, DeepReadonly, JsonObject, JsonValue } from "./types.js" @@ -17,6 +18,8 @@ export interface DeadLetter { } export class DeadLetterManager { + private readonly scopes = new Map() + constructor(private readonly runtime: SolidObjectsRuntime) {} all(options: AdministrationOptions = {}): Promise { @@ -26,4 +29,21 @@ export class DeadLetterManager { retry(id: string, options: AdministrationOptions = {}): Promise { return this.runtime.retryDeadLetter(id, options) } + + get effects(): DeadLetterScope { + return this.scope("effect") + } + + get broadcasts(): DeadLetterScope { + return this.scope("broadcast") + } + + scope(kind: DeadLetterKind): DeadLetterScope { + const existing = this.scopes.get(kind) + if (existing) return existing + + const created = new DeadLetterScope(this.runtime, kind) + this.scopes.set(kind, created) + return created + } } diff --git a/src/doctor.ts b/src/doctor.ts index 19cfc9c..9558b47 100644 --- a/src/doctor.ts +++ b/src/doctor.ts @@ -96,6 +96,18 @@ const EXPECTED_COLUMNS: Readonly> = { "available_at_ms", ], dead_letters: ["id", "message_id", "instance_id", "attempts", "error", "retried_message_id"], + administration_events: ["id", "action", "kind", "subject_id", "actor", "occurred_at_ms"], + redrives: [ + "id", + "kind", + "filters", + "status", + "active_scope", + "moved", + "move_limit", + "started_at_ms", + "finished_at_ms", + ], } class DoctorProbe extends Actor { @@ -231,11 +243,11 @@ export class Doctor { message: `incompatible schema identity ${wrongIdentity.schema_identity}`, }) } - if (versions.join(",") !== "1,2,3,4,5,6,7,8,9,10") { + if (versions.join(",") !== "1,2,3,4,5,6,7,8,9,10,11") { return check({ name: "schema", status: "fail", - message: `expected schema migrations 1, 2, 3, 4, 5, 6, 7, 8, 9, 10; found ${versions.join(", ")}`, + message: `expected schema migrations 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11; found ${versions.join(", ")}`, }) } return check({ diff --git a/src/index.ts b/src/index.ts index 2e0ec5a..9aee02e 100644 --- a/src/index.ts +++ b/src/index.ts @@ -67,6 +67,22 @@ export { type SelectedWakeUp, type WakeUpSelectionOptions, } from "./wake-up-selection.js" +export { + DeadLetterScope, + UnknownDeadRow, + type DeadLetterKind, + type DeadRow, + type RedriveFilters, + type RedriveOptions, +} from "./dead-letter-scopes.js" +export { + RedriveManager, + RedriveScheduler, + RedriveNotStarted, + UnknownRedrive, + type RedriveStatus, + type RedriveTask, +} from "./redrive.js" export { parseSubscriptionRequest, RealtimeManager, diff --git a/src/platform/uuid.ts b/src/platform/uuid.ts index 9c13cbd..fbc462e 100644 --- a/src/platform/uuid.ts +++ b/src/platform/uuid.ts @@ -2,3 +2,11 @@ export function randomUUID(): string { const platform = globalThis as typeof globalThis & { crypto: { randomUUID(): string } } return platform.crypto.randomUUID() } + +export async function sha256Hex(value: string): Promise { + const platform = globalThis as typeof globalThis & { + crypto: { subtle: { digest(algorithm: string, data: BufferSource): Promise } } + } + const digest = await platform.crypto.subtle.digest("SHA-256", new TextEncoder().encode(value)) + return Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join("") +} diff --git a/src/redrive.ts b/src/redrive.ts new file mode 100644 index 0000000..290613b --- /dev/null +++ b/src/redrive.ts @@ -0,0 +1,325 @@ +import type { DatabaseConnection } from "./database/types.js" +import type { DeadLetterKind, RedriveFilters } from "./dead-letter-scopes.js" +import { randomUUID, sha256Hex } from "./platform/uuid.js" +import type { SolidObjectsRuntime } from "./runtime.js" +import type { AdministrationOptions } from "./types.js" +import { waitFor } from "./worker.js" + +export type RedriveStatus = "running" | "completed" | "cancelled" + +export interface RedriveTask { + readonly id: string + readonly kind: DeadLetterKind + readonly filters: RedriveFilters + readonly status: RedriveStatus + readonly moved: number + readonly remaining: number + readonly startedAt: Date + readonly finishedAt: Date | null + cancel(options?: AdministrationOptions): Promise +} + +interface RedriveShape { + id: string + kind: DeadLetterKind + filters: string + status: RedriveStatus + active_scope: string | null + moved: number | bigint + move_limit: number | bigint | null + actor: string | null + started_at_ms: number | bigint + finished_at_ms: number | bigint | null +} + +export class RedriveManager { + constructor(private readonly runtime: SolidObjectsRuntime) {} + + async start(input: { + kind: DeadLetterKind + filters: RedriveFilters + authorizationContext?: unknown + }): Promise { + const activeScope = await activeScopeFor(input) + const existing = await this.findRow(activeScope, "active_scope") + if (existing) return await this.task(existing) + + const actor = await this.runtime.administrationIdentity(input.authorizationContext) + const opened = await this.open({ ...input, activeScope, actor }) + return await this.task(opened) + } + + async find(id: string, options: AdministrationOptions = {}): Promise { + await this.authorize({ action: "inspect", options, resourceId: id }) + return await this.task(await this.require(id)) + } + + async all( + options: AdministrationOptions & { status?: RedriveStatus } = {}, + ): Promise { + await this.authorize({ action: "inspect", options }) + const status = options.status + const rows = await this.runtime.settings.database.connection((connection) => + connection.all( + `SELECT * FROM ${this.table()} + ${status === undefined ? "" : "WHERE status = ?"} + ORDER BY started_at_ms DESC, id DESC`, + status === undefined ? [] : [status], + ), + ) + return Object.freeze(await Promise.all(rows.map((row) => this.task(row)))) + } + + async cancel(id: string, options: AdministrationOptions = {}): Promise { + await this.authorize({ action: "cancel", options, resourceId: id }) + const row = await this.require(id) + if (row.status !== "running") return await this.task(row) + + await this.close({ id, status: "cancelled" }) + await this.audit({ row, action: "redrive.cancel" }) + return await this.task(await this.require(id)) + } + + // A redrive moves rows in bounded batches, each its own short transaction, so + // one redrive cannot hold a lock long enough to starve delivery. + async advance(): Promise { + const row = await this.claim() + if (!row) return false + + const moved = await this.moveBatch(row) + if (moved > 0) return true + + await this.close({ id: row.id, status: "completed" }) + await this.audit({ row, action: "redrive.finish" }) + return false + } + + private async claim(): Promise { + return await this.runtime.settings.database.connection((connection) => + connection.get( + `SELECT * FROM ${this.table()} WHERE status = 'running' + ORDER BY started_at_ms, id LIMIT 1`, + ), + ) + } + + private async moveBatch(row: RedriveShape): Promise { + const size = this.batchSize(row) + if (size <= 0) return 0 + + const scope = this.runtime.deadLetters.scope(row.kind) + const filters = parseFilters(row.filters) + const moved = await this.runtime.settings.database.transaction(async (connection) => { + const candidates = await scope.matching({ + connection, + filters, + limit: size, + deadBefore: Number(row.started_at_ms), + }) + if (candidates.length === 0) return 0 + const revived = await scope.revive({ + connection, + identifiers: candidates.map(({ id }) => id), + }) + await connection.run(`UPDATE ${this.table()} SET moved = moved + ? WHERE id = ?`, [ + revived, + row.id, + ]) + return revived + }) + if (moved > 0) this.runtime.wakeUpAfterRevival(row.kind) + return moved + } + + private batchSize(row: RedriveShape): number { + const configured = this.runtime.settings.redriveBatchSize + if (row.move_limit === null) return configured + return Math.min(configured, Number(row.move_limit) - Number(row.moved)) + } + + private async open(input: { + kind: DeadLetterKind + filters: RedriveFilters + activeScope: string + actor: string | null + }): Promise { + const id = `redrive_${randomUUID()}` + try { + await this.runtime.settings.database.transaction(async (connection) => { + const now = await connection.nowMilliseconds() + await connection.run( + `INSERT INTO ${this.table()} + (id, kind, filters, status, active_scope, moved, move_limit, actor, started_at_ms) + VALUES (?, ?, ?, 'running', ?, 0, ?, ?, ?)`, + [ + id, + input.kind, + JSON.stringify(input.filters), + input.activeScope, + input.filters.limit, + input.actor, + now, + ], + ) + }) + } catch { + const running = await this.findRow(input.activeScope, "active_scope") + if (running) return running + throw new RedriveNotStarted(`could not start a ${input.kind} redrive`) + } + const row = await this.require(id) + await this.audit({ row, action: "redrive.start" }) + return row + } + + private async close(input: { id: string; status: RedriveStatus }): Promise { + await this.runtime.settings.database.transaction(async (connection) => { + const now = await connection.nowMilliseconds() + await connection.run( + `UPDATE ${this.table()} SET status = ?, active_scope = NULL, finished_at_ms = ? + WHERE id = ? AND status = 'running'`, + [input.status, now, input.id], + ) + }) + } + + private async audit(input: { row: RedriveShape; action: string }): Promise { + await this.runtime.writeAdministrationEvent({ + action: input.action, + kind: input.row.kind, + subjectId: input.row.id, + filters: parseFilters(input.row.filters), + actor: input.row.actor, + }) + } + + private async task(row: RedriveShape): Promise { + const manager = this + return Object.freeze({ + id: row.id, + kind: row.kind, + filters: parseFilters(row.filters), + status: row.status, + moved: Number(row.moved), + remaining: await this.remaining(row), + startedAt: new Date(Number(row.started_at_ms)), + finishedAt: row.finished_at_ms === null ? null : new Date(Number(row.finished_at_ms)), + cancel(options: AdministrationOptions = {}): Promise { + return manager.cancel(row.id, options) + }, + }) + } + + // A running task reports what is left to move rather than a stored estimate, + // because rows die and are retried while it runs. + private async remaining(row: RedriveShape): Promise { + if (row.status !== "running") return 0 + + const scope = this.runtime.deadLetters.scope(row.kind) + const matching = await this.runtime.settings.database.connection((connection) => + scope.count({ + connection, + filters: parseFilters(row.filters), + deadBefore: Number(row.started_at_ms), + }), + ) + if (row.move_limit === null) return matching + return Math.min(matching, Number(row.move_limit) - Number(row.moved)) + } + + private async require(id: string): Promise { + const row = await this.findRow(id, "id") + if (!row) throw new UnknownRedrive(`unknown redrive ${id}`) + return row + } + + private async findRow( + value: string, + column: "id" | "active_scope", + ): Promise { + return await this.runtime.settings.database.connection((connection: DatabaseConnection) => + connection.get(`SELECT * FROM ${this.table()} WHERE ${column} = ?`, [value]), + ) + } + + private table(): string { + return this.runtime.repository.table("redrives") + } + + private async authorize(input: { + action: string + options: AdministrationOptions + resourceId?: string + }): Promise { + await this.runtime.authorizeAdministration({ + action: input.action, + resource: "redrives", + ...(input.resourceId === undefined ? {} : { resourceId: input.resourceId }), + authorizationContext: input.options.authorizationContext, + }) + } +} + +// An operator starts a redrive and expects it to move, so the runtime advances +// it rather than ask the application to schedule a job. A pass that moved rows +// pauses briefly and takes the next batch; an idle pass waits longer, so an +// empty table is not polled every few milliseconds. +export class RedriveScheduler { + private stopping = false + + constructor(private readonly runtime: SolidObjectsRuntime) {} + + async run(signal: AbortSignal): Promise { + while (!signal.aborted && !this.stopping) { + const moved = await this.advance() + await waitFor(this.pauseMilliseconds(moved), signal) + } + } + + requestShutdown(): void { + this.stopping = true + } + + stopped(): boolean { + return this.stopping + } + + stop(): void { + this.stopping = true + } + + private async advance(): Promise { + try { + return await this.runtime.redrives.advance() + } catch (error) { + this.runtime.emitInstrumentation("supervisor.redrive_failed", { + errorName: error instanceof Error ? error.name : "Error", + }) + return false + } + } + + private pauseMilliseconds(moved: boolean): number { + if (moved) return this.runtime.settings.redriveBatchPauseMilliseconds + return this.runtime.settings.idlePollingIntervalMilliseconds + } +} + +async function activeScopeFor(input: { + kind: DeadLetterKind + filters: RedriveFilters +}): Promise { + return `${input.kind}:${await sha256Hex(JSON.stringify(input.filters))}` +} + +function parseFilters(filters: string): RedriveFilters { + return Object.freeze(JSON.parse(filters) as RedriveFilters) +} + +export class UnknownRedrive extends Error { + override readonly name = "UnknownRedrive" +} + +export class RedriveNotStarted extends Error { + override readonly name = "RedriveNotStarted" +} diff --git a/src/repository.ts b/src/repository.ts index 4840d28..642a005 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -1472,6 +1472,8 @@ export class Repository { async resetForTesting(): Promise { const tables = [ + "administration_events", + "redrives", "effect_recoveries", "dead_letters", "claimed_messages", @@ -1592,9 +1594,14 @@ export class Repository { const updated = await connection.run( `UPDATE ${this.table("effects")} - SET status = 'dead', error = ?, claimed_by = NULL + SET status = 'dead', error = ?, claimed_by = NULL, failed_at_ms = ? WHERE id = ? AND status = 'processing' AND claimed_by = ?`, - [JSON.stringify(errorRecord), effect.id, effect.claimed_by], + [ + JSON.stringify(errorRecord), + await connection.nowMilliseconds(), + effect.id, + effect.claimed_by, + ], ) if (updated.changes !== 1) throw new LostActivation("effect claim no longer matches") if (!effect.failure_operation) return @@ -2044,9 +2051,17 @@ export class Repository { : now + this.settings.retryDelayMilliseconds(Number(broadcast.attempt_count)) const updated = await connection.run( `UPDATE ${this.table("broadcasts")} - SET status = ?, error = ?, available_at_ms = ?, claimed_by = NULL + SET status = ?, error = ?, available_at_ms = ?, claimed_by = NULL, + failed_at_ms = ? WHERE id = ? AND status = 'processing' AND claimed_by = ?`, - [status, JSON.stringify(safeError(error)), availableAt, broadcast.id, broadcast.claimed_by], + [ + status, + JSON.stringify(safeError(error)), + availableAt, + now, + broadcast.id, + broadcast.claimed_by, + ], ) if (updated.changes !== 1) throw new LostActivation("broadcast claim no longer matches") }) diff --git a/src/runtime.ts b/src/runtime.ts index 259869d..9bfe1c7 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -14,7 +14,10 @@ import { withActorProjection, withApplicationWritesForbidden, } from "./context.js" +import { randomUUID } from "./platform/uuid.js" import { DeadLetterManager, type DeadLetter } from "./dead-letters.js" +import type { DeadLetterKind, RedriveFilters } from "./dead-letter-scopes.js" +import { RedriveManager, RedriveScheduler } from "./redrive.js" import { Doctor } from "./doctor.js" import { clearDefaultRuntime, setDefaultRuntime } from "./default-runtime.js" import { @@ -182,10 +185,24 @@ function scheduledReminderOf(row: ReminderRow): ScheduledReminder { } } +// An audit log is read in the order things happened, and two events can share +// a millisecond. The id carries the stamp and a counter, so a reader orders by +// it and sees insertion order within a process. Across processes the stamp is +// all any log can offer. +let administrationEventCounter = 0 + +function administrationEventId(occurredAtMilliseconds: number): string { + administrationEventCounter = (administrationEventCounter + 1) % 1_000_000 + const stamp = String(occurredAtMilliseconds).padStart(15, "0") + const counter = String(administrationEventCounter).padStart(6, "0") + return `${stamp}-${counter}-${randomUUID()}` +} + export class SolidObjectsRuntime { readonly settings readonly repository readonly deadLetters + readonly redrives readonly reconciliation readonly retention readonly doctor @@ -211,6 +228,7 @@ export class SolidObjectsRuntime { wakeUpAdapter: () => this.wakeUpAdapter(), }) this.deadLetters = new DeadLetterManager(this) + this.redrives = new RedriveManager(this) this.reconciliation = new ReconciliationManager(this) this.retention = new RetentionManager(this) this.doctor = new Doctor(this) @@ -871,6 +889,59 @@ export class SolidObjectsRuntime { return destroyed } + // A revived row is pending again, so the role that drains it has to be told + // rather than wait out its polling interval. + wakeUpAfterRevival(kind: DeadLetterKind): void { + this.wakeUp(kind === "effect" ? "effects" : "broadcasts") + } + + async administrationIdentity(authorizationContext: unknown): Promise { + const identity = await this.settings.administrationIdentity(authorizationContext) + return identity === null || identity === undefined ? null : String(identity).slice(0, 255) + } + + async recordAdministrationEvent(input: { + action: string + kind: string + subjectId?: string + filters?: RedriveFilters + authorizationContext?: unknown + }): Promise { + await this.writeAdministrationEvent({ + action: input.action, + kind: input.kind, + ...(input.subjectId === undefined ? {} : { subjectId: input.subjectId }), + ...(input.filters === undefined ? {} : { filters: input.filters }), + actor: await this.administrationIdentity(input.authorizationContext), + }) + } + + async writeAdministrationEvent(input: { + action: string + kind: string + subjectId?: string + filters?: RedriveFilters + actor: string | null + }): Promise { + await this.settings.database.transaction(async (connection) => { + const occurredAt = await connection.nowMilliseconds() + await connection.run( + `INSERT INTO ${this.repository.table("administration_events")} + (id, action, kind, subject_id, filters, actor, occurred_at_ms) + VALUES (?, ?, ?, ?, ?, ?, ?)`, + [ + administrationEventId(occurredAt), + input.action, + input.kind, + input.subjectId ?? null, + input.filters === undefined ? null : JSON.stringify(input.filters), + input.actor, + occurredAt, + ], + ) + }) + } + async inspectDeadLetters(options: AdministrationOptions = {}): Promise { await this.authorizeAdministration({ action: "inspect", @@ -1634,6 +1705,7 @@ export class SolidObjectsRuntime { () => () => this.broadcastWorker(), ) : []), + () => new RedriveScheduler(this), ...(this.settings.retentionIntervalMilliseconds > 0 ? [ () => @@ -2043,7 +2115,7 @@ export class SolidObjectsRuntime { if (!authorized) throw new Unauthorized(`actor ${kind} is not authorized`) } - private async authorizeAdministration(options: { + async authorizeAdministration(options: { action: string resource: string resourceId?: string diff --git a/src/schema.ts b/src/schema.ts index 9bc48d3..5c82a5e 100644 --- a/src/schema.ts +++ b/src/schema.ts @@ -11,7 +11,8 @@ const KEYED_REMINDERS_VERSION = 7 const POLLING_INDEXES_VERSION = 8 const EFFECT_RECOVERY_VERSION = 9 const INSTANCE_RETENTION_INDEX_VERSION = 10 -const LATEST_VERSION = INSTANCE_RETENTION_INDEX_VERSION +const DEAD_LETTER_REDRIVE_VERSION = 11 +const LATEST_VERSION = DEAD_LETTER_REDRIVE_VERSION export async function installSchema(options: { connection: DatabaseConnection @@ -350,6 +351,10 @@ export async function installSchema(options: { }) } + if (!installedVersions.has(DEAD_LETTER_REDRIVE_VERSION)) { + await installRedrive({ connection, family, table, prefix, schemaIdentity, createTable }) + } + if (installedVersions.has(POLLING_INDEXES_VERSION)) return const pollingIndexes = [ ["effects", `${prefix}effects_poll`, "status, available_at_ms, id"], @@ -374,6 +379,84 @@ export async function installSchema(options: { }) } +// A dead row carries no timestamp of its own, so a redrive that filters on +// when something failed needs one. Existing dead rows take their availability +// stamp, which is the closest record of their last attempt. +async function installRedrive(options: { + connection: DatabaseConnection + family: DatabaseFamily + table: (name: string) => string + prefix: string + schemaIdentity: string + createTable: (sql: string) => Promise +}): Promise { + const { connection, family, table, prefix, schemaIdentity, createTable } = options + await createTable(`CREATE TABLE IF NOT EXISTS ${table("administration_events")} ( + id TEXT PRIMARY KEY, + action TEXT NOT NULL, + kind TEXT NOT NULL, + subject_id TEXT, + filters TEXT, + actor TEXT, + occurred_at_ms INTEGER NOT NULL + ) STRICT`) + + await createTable(`CREATE TABLE IF NOT EXISTS ${table("redrives")} ( + id TEXT PRIMARY KEY, + kind TEXT NOT NULL, + filters TEXT NOT NULL, + status TEXT NOT NULL CHECK (status IN ('running', 'completed', 'cancelled')), + active_scope TEXT UNIQUE, + moved INTEGER NOT NULL DEFAULT 0 CHECK (moved >= 0), + move_limit INTEGER CHECK (move_limit IS NULL OR move_limit > 0), + actor TEXT, + started_at_ms INTEGER NOT NULL, + finished_at_ms INTEGER + ) STRICT`) + + for (const name of ["effects", "broadcasts"] as const) { + await addFailedAt({ connection, family, table: table(name) }) + } + await createIndex({ + connection, + family, + table: table("administration_events"), + name: `${prefix}admin_events_occurred`, + columns: "occurred_at_ms, id", + }) + await createIndex({ + connection, + family, + table: table("redrives"), + name: `${prefix}redrives_poll`, + columns: "status, started_at_ms, id", + }) + await recordMigration({ + connection, + table: table("schema_migrations"), + version: DEAD_LETTER_REDRIVE_VERSION, + schemaIdentity, + }) +} + +async function addFailedAt(options: { + connection: DatabaseConnection + family: DatabaseFamily + table: string +}): Promise { + // A millisecond stamp does not fit a 32-bit INTEGER, which is what every + // family but SQLite means by that word. + const type = options.family === "sqlite" ? "INTEGER" : "BIGINT" + try { + await options.connection.run(`ALTER TABLE ${options.table} ADD COLUMN failed_at_ms ${type}`) + } catch { + return + } + await options.connection.run( + `UPDATE ${options.table} SET failed_at_ms = available_at_ms WHERE status = 'dead'`, + ) +} + async function recordMigration(options: { connection: DatabaseConnection table: string diff --git a/test/dead-letter-scopes.test.ts b/test/dead-letter-scopes.test.ts new file mode 100644 index 0000000..0a1c989 --- /dev/null +++ b/test/dead-letter-scopes.test.ts @@ -0,0 +1,198 @@ +import { afterEach, describe, expect, it } from "vitest" +import { Actor } from "../src/actor.js" +import { sqlite } from "../src/database/sqlite.js" +import { Unauthorized } from "../src/errors.js" +import { configure, type SolidObjectsRuntime } from "../src/runtime.js" + +class OrderActor extends Actor { + static override readonly actorType = "RedriveOrderActor" + + count = 0 + orders = 0 + + place(): void { + this.orders += 1 + this.emit("settle", { arguments: { order: "one" } }) + } + + touch(): void { + this.count += 1 + } + + override observables(): Record { + return { count: this.count } + } +} + +class PoisonActor extends Actor { + static override readonly actorType = "RedrivePoisonActor" + static fail = true + + run(): void { + if (PoisonActor.fail) throw new Error("poison message") + } +} + +let runtime: SolidObjectsRuntime | undefined +let settle: (argumentsValue: Record) => void = () => {} +let deliver: () => void = () => {} + +afterEach(async () => { + await runtime?.close() + runtime = undefined + PoisonActor.fail = true + settle = () => {} + deliver = () => {} +}) + +async function start(): Promise { + const created = configure({ + database: sqlite({ path: ":memory:" }), + maxAttempts: 1, + retryDelayMilliseconds: () => 0, + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeAdministration: () => true, + broadcast: async () => deliver(), + logger: { debug: () => {}, info: () => {}, warn: () => {}, error: () => {} }, + }) + created.registerEffect("settle", (argumentsValue) => settle(argumentsValue)) + created.register(OrderActor) + created.register(PoisonActor) + await created.install() + runtime = created + return created +} + +async function deadEffect(active: SolidObjectsRuntime): Promise { + settle = () => { + throw new Error("settlement declined") + } + await active.ref(OrderActor, "one").send.place() + await active.worker().runUntilIdle() + await active.effectWorker().runUntilIdle() + const dead = await active.deadLetters.effects.all() + expect(dead).toHaveLength(1) + return dead[0]!.id +} + +async function deadBroadcast(active: SolidObjectsRuntime): Promise { + deliver = () => { + throw new Error("transport down") + } + await active.ref(OrderActor, "broadcast").send.touch() + await active.worker().runUntilIdle() + await active.broadcastWorker().runUntilIdle() + const dead = await active.deadLetters.broadcasts.all() + expect(dead).toHaveLength(1) + return dead[0]!.id +} + +describe("dead-letter scopes", () => { + it("returns a dead effect to pending and runs it again", async () => { + const active = await start() + const id = await deadEffect(active) + const settled: unknown[] = [] + + await active.deadLetters.effects.retry(id) + + expect(await active.deadLetters.effects.all()).toHaveLength(0) + settle = (argumentsValue) => { + settled.push(argumentsValue) + } + await active.effectWorker().runUntilIdle() + + expect(settled).toEqual([{ order: "one" }]) + }) + + it("reuses the stable effect id when it retries", async () => { + const active = await start() + const id = await deadEffect(active) + + const retried = await active.deadLetters.effects.retry(id) + + expect(retried.id).toBe(id) + }) + + it("leaves an effect that is already pending alone", async () => { + const active = await start() + const id = await deadEffect(active) + const first = await active.deadLetters.effects.retry(id) + + const second = await active.deadLetters.effects.retry(id) + + expect(second.status).toBe("pending") + expect(second.availableAt).toEqual(first.availableAt) + }) + + it("returns a dead broadcast to pending and delivers it", async () => { + const active = await start() + const id = await deadBroadcast(active) + + await active.deadLetters.broadcasts.retry(id) + + expect(await active.deadLetters.broadcasts.all()).toHaveLength(0) + deliver = () => {} + await active.broadcastWorker().runUntilIdle() + expect(await active.deadLetters.broadcasts.all()).toHaveLength(0) + }) + + it("reads and retries message dead letters as it always has", async () => { + const active = await start() + await active.ref(PoisonActor, "one").send.run() + await active.worker().runUntilIdle() + const letters = await active.deadLetters.all() + PoisonActor.fail = false + + const reference = await active.deadLetters.retry(letters[0]!.id) + await active.worker().runUntilIdle() + + expect(letters).toHaveLength(1) + expect(await reference.status()).toBe("completed") + }) + + it("reads only its own kind", async () => { + const active = await start() + const effectId = await deadEffect(active) + const broadcastId = await deadBroadcast(active) + await active.ref(PoisonActor, "one").send.run() + await active.worker().runUntilIdle() + + expect((await active.deadLetters.effects.all()).map(({ id }) => id)).toEqual([effectId]) + expect((await active.deadLetters.broadcasts.all()).map(({ id }) => id)).toEqual([broadcastId]) + expect(await active.deadLetters.all()).toHaveLength(1) + }) + + it("refuses an unauthorized caller", async () => { + const active = configure({ + database: sqlite({ path: ":memory:" }), + authorizeMessage: () => true, + }) + runtime = active + await active.install() + + await expect(active.deadLetters.effects.all()).rejects.toBeInstanceOf(Unauthorized) + await expect(active.deadLetters.effects.retry("missing")).rejects.toBeInstanceOf(Unauthorized) + await expect(active.deadLetters.broadcasts.retry("missing")).rejects.toBeInstanceOf( + Unauthorized, + ) + }) + + it("names the scope it authorizes", async () => { + const seen: { action: string; resource: string }[] = [] + const active = configure({ + database: sqlite({ path: ":memory:" }), + authorizeMessage: () => true, + authorizeAdministration: ({ action, resource }) => { + seen.push({ action, resource }) + return true + }, + }) + runtime = active + await active.install() + + await active.deadLetters.effects.all() + + expect(seen).toEqual([{ action: "inspect", resource: "effect_dead_letters" }]) + }) +}) diff --git a/test/dead-letters.test.ts b/test/dead-letters.test.ts index 9715529..f3e1be2 100644 --- a/test/dead-letters.test.ts +++ b/test/dead-letters.test.ts @@ -151,7 +151,9 @@ describe("schema migrations", () => { const broadcastColumns = await runtime.settings.database.connection((connection) => connection.all<{ name: string }>("PRAGMA table_info(solid_objects_broadcasts)"), ) - expect(versions.map(({ version }) => Number(version))).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]) + expect(versions.map(({ version }) => Number(version))).toEqual([ + 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, + ]) expect(deadLetterColumns.map(({ name }) => name)).toContain("retried_message_id") expect(broadcastColumns.map(({ name }) => name)).toContain("invalidations") expect(await installedPollingIndexes(runtime)).toEqual(POLLING_INDEX_COLUMNS) @@ -184,7 +186,9 @@ describe("schema migrations", () => { "SELECT version FROM solid_objects_schema_migrations ORDER BY version", ), ) - expect(versions.map(({ version }) => Number(version))).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]) + expect(versions.map(({ version }) => Number(version))).toEqual([ + 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, + ]) expect(await installedPollingIndexes(runtime)).toEqual(POLLING_INDEX_COLUMNS) }) diff --git a/test/doctor.test.ts b/test/doctor.test.ts index d1af544..cf5fde9 100644 --- a/test/doctor.test.ts +++ b/test/doctor.test.ts @@ -27,7 +27,7 @@ describe("runtime doctor", () => { expect(check(report, "configuration").status).toBe("pass") expect(check(report, "schema")).toMatchObject({ status: "pass", - details: { versions: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] }, + details: { versions: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11] }, }) expect(check(report, "authorization").status).toBe("pass") expect(check(report, "database").status).toBe("pass") diff --git a/test/instance-retention.test.ts b/test/instance-retention.test.ts index 046693d..96195ee 100644 --- a/test/instance-retention.test.ts +++ b/test/instance-retention.test.ts @@ -100,7 +100,9 @@ it.each(["fresh installation", "version-nine upgrade", "interrupted upgrade"])( `SELECT version FROM ${PREFIX}schema_migrations ORDER BY version`, ), ) - expect(versions.map(({ version }) => Number(version))).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]) + expect(versions.map(({ version }) => Number(version))).toEqual([ + 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, + ]) }, 30_000, ) diff --git a/test/redrive.test.ts b/test/redrive.test.ts new file mode 100644 index 0000000..988388e --- /dev/null +++ b/test/redrive.test.ts @@ -0,0 +1,388 @@ +import { afterEach, describe, expect, it } from "vitest" +import { Actor } from "../src/actor.js" +import { sqlite } from "../src/database/sqlite.js" +import { Unauthorized } from "../src/errors.js" +import { configure, type SolidObjectsRuntime } from "../src/runtime.js" + +class PaymentActor extends Actor { + static override readonly actorType = "RedrivePaymentActor" + + placed = 0 + + place(): void { + this.placed += 1 + this.emit("settle", { arguments: { order: "one" } }) + } +} + +class ShipmentActor extends Actor { + static override readonly actorType = "RedriveShipmentActor" + + count = 0 + + touch(): void { + this.count += 1 + } + + override observables(): Record { + return { count: this.count } + } +} + +let runtime: SolidObjectsRuntime | undefined +let deliver: () => void = () => {} + +afterEach(async () => { + await runtime?.close() + runtime = undefined + deliver = () => {} +}) + +async function start(): Promise { + const created = configure({ + database: sqlite({ path: ":memory:" }), + maxAttempts: 1, + retryDelayMilliseconds: () => 0, + redriveBatchSize: 10, + redriveBatchPauseMilliseconds: 0, + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeAdministration: () => true, + broadcast: async () => deliver(), + logger: { debug: () => {}, info: () => {}, warn: () => {}, error: () => {} }, + }) + created.register(PaymentActor) + created.register(ShipmentActor) + created.registerEffect("settle", () => { + throw new Error("declined") + }) + await created.install() + runtime = created + return created +} + +async function deadEffects(active: SolidObjectsRuntime, count: number): Promise { + for (let index = 0; index < count; index += 1) { + await active.ref(PaymentActor, `order-${index}`).send.place() + } + await active.worker().runUntilIdle() + await markDead(active, "effects") + expect(await active.deadLetters.effects.all()).toHaveLength(count) +} + +async function deadBroadcast(active: SolidObjectsRuntime): Promise { + await active.ref(ShipmentActor, "one").send.touch() + await active.worker().runUntilIdle() + await markDead(active, "broadcasts") +} + +async function markDead(active: SolidObjectsRuntime, table: string): Promise { + await active.settings.database.transaction(async (connection) => { + const now = await connection.nowMilliseconds() + await connection.run( + `UPDATE ${active.repository.table(table)} SET status = 'dead', failed_at_ms = ? + WHERE status <> 'dead'`, + [now], + ) + }) +} + +async function drain(active: SolidObjectsRuntime): Promise { + while (await active.redrives.advance()) continue +} + +async function pendingEffects(active: SolidObjectsRuntime): Promise { + const row = await active.settings.database.connection((connection) => + connection.get<{ total: number | bigint }>( + `SELECT COUNT(*) AS total FROM ${active.repository.table("effects")} WHERE status = 'pending'`, + ), + ) + return Number(row?.total ?? 0) +} + +describe("redrive", () => { + it("moves every matching row in bounded batches", async () => { + const active = await start() + await deadEffects(active, 25) + + const task = await active.deadLetters.effects.redrive() + expect(task.status).toBe("running") + + expect(await active.redrives.advance()).toBe(true) + expect(await pendingEffects(active)).toBe(10) + expect(await active.redrives.advance()).toBe(true) + expect(await pendingEffects(active)).toBe(20) + expect(await active.redrives.advance()).toBe(true) + expect(await pendingEffects(active)).toBe(25) + expect(await active.redrives.advance()).toBe(false) + + const finished = await active.redrives.find(task.id) + expect(finished.status).toBe("completed") + expect(finished.moved).toBe(25) + expect(finished.remaining).toBe(0) + }) + + it("stops at the limit and leaves the rest dead", async () => { + const active = await start() + await deadEffects(active, 25) + + const task = await active.deadLetters.effects.redrive({ limit: 15 }) + await drain(active) + + const finished = await active.redrives.find(task.id) + expect(finished.moved).toBe(15) + expect(finished.status).toBe("completed") + expect(await active.deadLetters.effects.all()).toHaveLength(10) + }) + + it("returns the running task when the same scope is redriven again", async () => { + const active = await start() + await deadEffects(active, 25) + + const first = await active.deadLetters.effects.redrive() + const second = await active.deadLetters.effects.redrive() + + expect(second.id).toBe(first.id) + expect(await active.redrives.all()).toHaveLength(1) + }) + + it("starts a separate task for another scope while one runs", async () => { + const active = await start() + await deadEffects(active, 5) + await deadBroadcast(active) + + const effects = await active.deadLetters.effects.redrive() + const broadcasts = await active.deadLetters.broadcasts.redrive() + + expect(broadcasts.id).not.toBe(effects.id) + expect((await active.redrives.all()).map(({ kind }) => kind).sort()).toEqual([ + "broadcast", + "effect", + ]) + }) + + it("starts a separate task for different filters", async () => { + const active = await start() + await deadEffects(active, 5) + + const first = await active.deadLetters.effects.redrive() + const second = await active.deadLetters.effects.redrive({ actorType: "RedrivePaymentActor" }) + + expect(second.id).not.toBe(first.id) + }) + + it("starts a new task once the first finishes", async () => { + const active = await start() + await deadEffects(active, 5) + const first = await active.deadLetters.effects.redrive() + await drain(active) + await markDead(active, "effects") + + const second = await active.deadLetters.effects.redrive() + + expect(second.id).not.toBe(first.id) + expect(second.status).toBe("running") + }) + + it("cancels a running task and keeps the rows it already moved", async () => { + const active = await start() + await deadEffects(active, 25) + const task = await active.deadLetters.effects.redrive() + await active.redrives.advance() + + await task.cancel() + + expect(await active.redrives.advance()).toBe(false) + const cancelled = await active.redrives.find(task.id) + expect(cancelled.status).toBe("cancelled") + expect(cancelled.moved).toBe(10) + expect(await pendingEffects(active)).toBe(10) + expect(await active.deadLetters.effects.all()).toHaveLength(15) + }) + + it("filters by actor type", async () => { + const active = await start() + await deadEffects(active, 3) + + const task = await active.deadLetters.effects.redrive({ actorType: "RedriveShipmentActor" }) + await drain(active) + + expect((await active.redrives.find(task.id)).moved).toBe(0) + expect(await active.deadLetters.effects.all()).toHaveLength(3) + }) + + it("filters by failure time", async () => { + const active = await start() + await deadEffects(active, 3) + + const task = await active.deadLetters.effects.redrive({ + failedAfter: new Date(Date.now() + 60_000), + }) + await drain(active) + + expect((await active.redrives.find(task.id)).moved).toBe(0) + }) + + it("reads tasks back by id and by status", async () => { + const active = await start() + await deadEffects(active, 5) + const task = await active.deadLetters.effects.redrive() + + expect((await active.redrives.all({ status: "running" })).map(({ id }) => id)).toEqual([ + task.id, + ]) + + await drain(active) + + expect(await active.redrives.all({ status: "running" })).toHaveLength(0) + expect((await active.redrives.all({ status: "completed" })).map(({ id }) => id)).toEqual([ + task.id, + ]) + }) + + it("reports what a running task has left to move", async () => { + const active = await start() + await deadEffects(active, 25) + + const task = await active.deadLetters.effects.redrive({ limit: 15 }) + expect(task.remaining).toBe(15) + + await active.redrives.advance() + + expect((await active.redrives.find(task.id)).remaining).toBe(5) + }) + + it( + "a running runtime advances a redrive without a caller driving it", + { timeout: 20_000 }, + async () => { + const active = await start() + await deadEffects(active, 12) + const task = await active.deadLetters.effects.redrive() + const controller = new AbortController() + const running = active.run(controller.signal) + + const deadline = Date.now() + 10_000 + while ((await active.redrives.find(task.id)).status !== "completed") { + if (Date.now() > deadline) throw new Error("the runtime did not finish the redrive") + await new Promise((resolve) => setTimeout(resolve, 20)) + } + controller.abort() + await running + + expect((await active.redrives.find(task.id)).moved).toBe(12) + }, + ) + + it("does not move a row that died after the task started", async () => { + const active = await start() + await deadEffects(active, 2) + const task = await active.deadLetters.effects.redrive() + await new Promise((resolve) => setTimeout(resolve, 5)) + await active.ref(PaymentActor, "late").send.place() + await active.worker().runUntilIdle() + await markDead(active, "effects") + + await drain(active) + + expect((await active.redrives.find(task.id)).moved).toBe(2) + expect(await active.deadLetters.effects.all()).toHaveLength(1) + }) + + it("refuses an unauthorized caller", async () => { + const active = configure({ + database: sqlite({ path: ":memory:" }), + authorizeMessage: () => true, + }) + runtime = active + await active.install() + + await expect(active.deadLetters.effects.redrive()).rejects.toBeInstanceOf(Unauthorized) + await expect(active.redrives.all()).rejects.toBeInstanceOf(Unauthorized) + await expect(active.redrives.find("missing")).rejects.toBeInstanceOf(Unauthorized) + await expect(active.redrives.cancel("missing")).rejects.toBeInstanceOf(Unauthorized) + }) +}) + +describe("administration audit", () => { + it("writes one row for each retry, including a repeat", async () => { + const active = await start() + await deadEffects(active, 1) + const id = (await active.deadLetters.effects.all())[0]!.id + + await active.deadLetters.effects.retry(id) + await active.deadLetters.effects.retry(id) + + const events = await auditRows(active) + expect(events.map(({ action }) => action)).toEqual(["dead_letter.retry", "dead_letter.retry"]) + expect(events[0]).toMatchObject({ kind: "effect", subject_id: id }) + }) + + it("writes one row for each redrive transition", async () => { + const active = await start() + await deadEffects(active, 5) + + const task = await active.deadLetters.effects.redrive() + await drain(active) + + const events = await auditRows(active) + expect(events.map(({ action }) => action)).toEqual(["redrive.start", "redrive.finish"]) + expect(events.map(({ subject_id }) => subject_id)).toEqual([task.id, task.id]) + }) + + it("writes one row when a task is cancelled", async () => { + const active = await start() + await deadEffects(active, 5) + const task = await active.deadLetters.effects.redrive() + + await task.cancel() + + expect((await auditRows(active)).map(({ action }) => action)).toEqual([ + "redrive.start", + "redrive.cancel", + ]) + }) + + it("records the identity the application names", async () => { + const active = configure({ + database: sqlite({ path: ":memory:" }), + authorizeMessage: () => true, + authorizeAdministration: () => true, + administrationIdentity: (context) => `user:${(context as { id: number }).id}`, + }) + runtime = active + await active.install() + + await active.deadLetters.effects.redrive({ authorizationContext: { id: 42 } }) + + expect((await auditRows(active))[0]?.actor).toBe("user:42") + }) + + it("writes no row when the caller is refused, and none for a read", async () => { + const active = await start() + await deadEffects(active, 1) + + await active.deadLetters.effects.all() + expect(await auditRows(active)).toHaveLength(0) + + const refusing = configure({ + database: sqlite({ path: ":memory:" }), + authorizeMessage: () => true, + }) + await refusing.install() + await expect(refusing.deadLetters.effects.retry("missing")).rejects.toBeInstanceOf(Unauthorized) + expect(await auditRows(refusing)).toHaveLength(0) + await refusing.close() + }) +}) + +async function auditRows( + active: SolidObjectsRuntime, +): Promise<{ action: string; kind: string; subject_id: string | null; actor: string | null }[]> { + return await active.settings.database.connection((connection) => + connection.all( + `SELECT action, kind, subject_id, actor FROM + ${active.repository.table("administration_events")} ORDER BY id`, + ), + ) +} From bf903ef6cca2642b477d22e9651ea3ee1e57423f Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 13:35:23 -0700 Subject: [PATCH 02/11] fix: authorize a redrive where it cannot be bypassed `RedriveManager.start` is public and ran no check of its own, so the only guard was the scope that normally calls it. A caller that reached the manager directly started a task unauthorized. The check moves into `start`, under the scope's own resource name, so there is one check and no way around it. Co-Authored-By: Claude Opus 5 (1M context) --- src/dead-letter-scopes.ts | 3 +-- src/redrive.ts | 4 ++++ test/redrive.test.ts | 16 ++++++++++++++++ 3 files changed, 21 insertions(+), 2 deletions(-) diff --git a/src/dead-letter-scopes.ts b/src/dead-letter-scopes.ts index 503aa8d..c2966e8 100644 --- a/src/dead-letter-scopes.ts +++ b/src/dead-letter-scopes.ts @@ -87,7 +87,6 @@ export class DeadLetterScope { } async redrive(options: RedriveOptions = {}): Promise { - await this.authorize({ action: "redrive", options }) return await this.runtime.redrives.start({ kind: this.kind, filters: { @@ -204,7 +203,7 @@ export class DeadLetterScope { }) } - private async authorize(input: { + async authorize(input: { action: string options: AdministrationOptions resourceId?: string diff --git a/src/redrive.ts b/src/redrive.ts index 290613b..64312f3 100644 --- a/src/redrive.ts +++ b/src/redrive.ts @@ -40,6 +40,10 @@ export class RedriveManager { filters: RedriveFilters authorizationContext?: unknown }): Promise { + await this.runtime.deadLetters.scope(input.kind).authorize({ + action: "redrive", + options: { authorizationContext: input.authorizationContext }, + }) const activeScope = await activeScopeFor(input) const existing = await this.findRow(activeScope, "active_scope") if (existing) return await this.task(existing) diff --git a/test/redrive.test.ts b/test/redrive.test.ts index 988388e..08c6710 100644 --- a/test/redrive.test.ts +++ b/test/redrive.test.ts @@ -289,6 +289,22 @@ describe("redrive", () => { expect(await active.deadLetters.effects.all()).toHaveLength(1) }) + it("refuses an unauthorized caller that reaches the manager directly", async () => { + const active = configure({ + database: sqlite({ path: ":memory:" }), + authorizeMessage: () => true, + }) + runtime = active + await active.install() + + await expect( + active.redrives.start({ + kind: "effect", + filters: { actorType: null, failedAfter: null, limit: null }, + }), + ).rejects.toBeInstanceOf(Unauthorized) + }) + it("refuses an unauthorized caller", async () => { const active = configure({ database: sqlite({ path: ":memory:" }), From 101fe1ec1d6807b8233d239d851e965fcecfe931 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 13:49:00 -0700 Subject: [PATCH 03/11] refactor: move the reasoning out of the source Eight comments explained decisions the code below them already shows, or carried a why that belongs in a commit message. They are gone, and the reasoning each one held is in the commit that introduced it and in the pull request body. Co-Authored-By: Claude Opus 5 (1M context) --- src/dead-letter-scopes.ts | 3 --- src/redrive.ts | 8 -------- src/runtime.ts | 6 ------ src/schema.ts | 5 ----- 4 files changed, 22 deletions(-) diff --git a/src/dead-letter-scopes.ts b/src/dead-letter-scopes.ts index c2966e8..a037baa 100644 --- a/src/dead-letter-scopes.ts +++ b/src/dead-letter-scopes.ts @@ -152,9 +152,6 @@ export class DeadLetterScope { return this.runtime.repository.table(TABLES[this.kind]) } - // A redrive moves what was already dead when it started. Without that bound a - // row that fails again lands back in the same scope, and a task whose handler - // is still broken would move it forever. private conditions( filters: RedriveFilters, deadBefore?: number, diff --git a/src/redrive.ts b/src/redrive.ts index 64312f3..6d036bb 100644 --- a/src/redrive.ts +++ b/src/redrive.ts @@ -84,8 +84,6 @@ export class RedriveManager { return await this.task(await this.require(id)) } - // A redrive moves rows in bounded batches, each its own short transaction, so - // one redrive cannot hold a lock long enough to starve delivery. async advance(): Promise { const row = await this.claim() if (!row) return false @@ -214,8 +212,6 @@ export class RedriveManager { }) } - // A running task reports what is left to move rather than a stored estimate, - // because rows die and are retried while it runs. private async remaining(row: RedriveShape): Promise { if (row.status !== "running") return 0 @@ -264,10 +260,6 @@ export class RedriveManager { } } -// An operator starts a redrive and expects it to move, so the runtime advances -// it rather than ask the application to schedule a job. A pass that moved rows -// pauses briefly and takes the next batch; an idle pass waits longer, so an -// empty table is not polled every few milliseconds. export class RedriveScheduler { private stopping = false diff --git a/src/runtime.ts b/src/runtime.ts index 9bfe1c7..a75e77f 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -185,10 +185,6 @@ function scheduledReminderOf(row: ReminderRow): ScheduledReminder { } } -// An audit log is read in the order things happened, and two events can share -// a millisecond. The id carries the stamp and a counter, so a reader orders by -// it and sees insertion order within a process. Across processes the stamp is -// all any log can offer. let administrationEventCounter = 0 function administrationEventId(occurredAtMilliseconds: number): string { @@ -889,8 +885,6 @@ export class SolidObjectsRuntime { return destroyed } - // A revived row is pending again, so the role that drains it has to be told - // rather than wait out its polling interval. wakeUpAfterRevival(kind: DeadLetterKind): void { this.wakeUp(kind === "effect" ? "effects" : "broadcasts") } diff --git a/src/schema.ts b/src/schema.ts index 5c82a5e..6050620 100644 --- a/src/schema.ts +++ b/src/schema.ts @@ -379,9 +379,6 @@ export async function installSchema(options: { }) } -// A dead row carries no timestamp of its own, so a redrive that filters on -// when something failed needs one. Existing dead rows take their availability -// stamp, which is the closest record of their last attempt. async function installRedrive(options: { connection: DatabaseConnection family: DatabaseFamily @@ -444,8 +441,6 @@ async function addFailedAt(options: { family: DatabaseFamily table: string }): Promise { - // A millisecond stamp does not fit a 32-bit INTEGER, which is what every - // family but SQLite means by that word. const type = options.family === "sqlite" ? "INTEGER" : "BIGINT" try { await options.connection.run(`ALTER TABLE ${options.table} ADD COLUMN failed_at_ms ${type}`) From a51fc19d3ef58317da4dc540869882e60684e6a9 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 13:53:06 -0700 Subject: [PATCH 04/11] fix: claim a redrive and write its audit atomically `claim()` was an unlocked SELECT outside the batch transaction, so two processes could take the same task. One revived the candidates, the other saw zero rows changed and marked the whole task completed while matching rows were still dead. The same gap let a batch revive rows after a cancel had committed. A pass now claims, moves, and closes inside one transaction, with `FOR UPDATE SKIP LOCKED` where the database has it, and every write guards on `status = 'running'`. Each administration event moves into the transaction that causes it, so a retry that names a row which does not exist writes nothing, and a transition that commits cannot lose its event. `failedAfter` and `limit` are validated. `new Date("nonsense").getTime()` is NaN, which JSON turns into null, so an unusable date silently became an unfiltered redrive over every dead row. The migration no longer swallows every DDL error. It asks the schema whether `failed_at_ms` exists and adds it only when missing, so a real failure surfaces instead of being recorded as a completed migration. Co-Authored-By: Claude Opus 5 (1M context) --- src/configuration.ts | 13 ++++- src/dead-letter-scopes.ts | 36 +++++++++--- src/redrive.ts | 115 +++++++++++++++++++++----------------- src/runtime.ts | 53 +++++++----------- src/schema.ts | 31 +++++++--- test/redrive.test.ts | 24 ++++++++ 6 files changed, 171 insertions(+), 101 deletions(-) diff --git a/src/configuration.ts b/src/configuration.ts index 4e36d1e..d850ee5 100644 --- a/src/configuration.ts +++ b/src/configuration.ts @@ -1,6 +1,13 @@ import { InvalidActor } from "./errors.js" import type { Database } from "./database/types.js" -import type { DeepReadonly, JsonObject, JsonValue, Logger, LongRunningComponent } from "./types.js" +import type { + AdministrationOptions, + DeepReadonly, + JsonObject, + JsonValue, + Logger, + LongRunningComponent, +} from "./types.js" import { WAKE_UP_NAMES, type WakeUpAdapter, type WakeUpSetting } from "./wake-up.js" const wakeUpNames: readonly string[] = WAKE_UP_NAMES @@ -84,7 +91,9 @@ export interface SolidObjectsConfiguration { wakeUp?: WakeUpSetting redriveBatchSize?: number redriveBatchPauseMilliseconds?: number - administrationIdentity?: (authorizationContext: unknown) => string | null | Promise + administrationIdentity?: ( + authorizationContext: AdministrationOptions["authorizationContext"], + ) => string | null | Promise } export interface BroadcastEvent { diff --git a/src/dead-letter-scopes.ts b/src/dead-letter-scopes.ts index a037baa..f2a7028 100644 --- a/src/dead-letter-scopes.ts +++ b/src/dead-letter-scopes.ts @@ -71,15 +71,17 @@ export class DeadLetterScope { async retry(id: string, options: AdministrationOptions = {}): Promise { await this.authorize({ action: "retry", options, resourceId: id }) - await this.runtime.recordAdministrationEvent({ - action: "dead_letter.retry", - kind: this.kind, - subjectId: id, - authorizationContext: options.authorizationContext, - }) + const actor = await this.runtime.administrationIdentity(options.authorizationContext) const row = await this.runtime.settings.database.transaction(async (connection) => { const found = await this.find({ connection, id }) if (found.status === "dead") await this.revive({ connection, identifiers: [id] }) + await this.runtime.writeAdministrationEvent({ + connection, + action: "dead_letter.retry", + kind: this.kind, + subjectId: id, + actor, + }) return await this.find({ connection, id }) }) this.runtime.wakeUpAfterRevival(this.kind) @@ -91,8 +93,8 @@ export class DeadLetterScope { kind: this.kind, filters: { actorType: options.actorType ?? null, - failedAfter: options.failedAfter ? options.failedAfter.getTime() : null, - limit: options.limit ?? null, + failedAfter: failedAfterFilter(options.failedAfter), + limit: limitFilter(options.limit), }, authorizationContext: options.authorizationContext, }) @@ -214,6 +216,24 @@ export class DeadLetterScope { } } +function failedAfterFilter(failedAfter: Date | undefined): number | null { + if (failedAfter === undefined) return null + + const milliseconds = failedAfter.getTime() + if (!Number.isFinite(milliseconds)) { + throw new TypeError("failedAfter must be a valid Date") + } + return milliseconds +} + +function limitFilter(limit: number | undefined): number | null { + if (limit === undefined) return null + if (!Number.isSafeInteger(limit) || limit < 1) { + throw new TypeError("limit must be a positive safe integer") + } + return limit +} + export function emptyFilters(): RedriveFilters { return Object.freeze({ actorType: null, failedAfter: null, limit: null }) } diff --git a/src/redrive.ts b/src/redrive.ts index 6d036bb..530044a 100644 --- a/src/redrive.ts +++ b/src/redrive.ts @@ -79,58 +79,64 @@ export class RedriveManager { const row = await this.require(id) if (row.status !== "running") return await this.task(row) - await this.close({ id, status: "cancelled" }) - await this.audit({ row, action: "redrive.cancel" }) + await this.runtime.settings.database.transaction((connection) => + this.close({ connection, row, status: "cancelled", action: "redrive.cancel" }), + ) return await this.task(await this.require(id)) } async advance(): Promise { - const row = await this.claim() - if (!row) return false + const outcome = await this.runtime.settings.database.transaction(async (connection) => { + const row = await this.claim(connection) + if (!row) return { moved: 0, kind: undefined } - const moved = await this.moveBatch(row) - if (moved > 0) return true + const moved = await this.moveBatch({ connection, row }) + if (moved > 0) return { moved, kind: row.kind } - await this.close({ id: row.id, status: "completed" }) - await this.audit({ row, action: "redrive.finish" }) - return false + await this.close({ connection, row, status: "completed", action: "redrive.finish" }) + return { moved: 0, kind: undefined } + }) + if (outcome.kind) this.runtime.wakeUpAfterRevival(outcome.kind) + return outcome.moved > 0 } - private async claim(): Promise { - return await this.runtime.settings.database.connection((connection) => - connection.get( - `SELECT * FROM ${this.table()} WHERE status = 'running' - ORDER BY started_at_ms, id LIMIT 1`, - ), + private async claim(connection: DatabaseConnection): Promise { + return await connection.get( + `SELECT * FROM ${this.table()} WHERE status = 'running' + ORDER BY started_at_ms, id LIMIT 1${this.lockClause()}`, ) } - private async moveBatch(row: RedriveShape): Promise { + private lockClause(): string { + return this.runtime.settings.database.family === "sqlite" ? "" : " FOR UPDATE SKIP LOCKED" + } + + private async moveBatch(input: { + connection: DatabaseConnection + row: RedriveShape + }): Promise { + const { connection, row } = input const size = this.batchSize(row) if (size <= 0) return 0 const scope = this.runtime.deadLetters.scope(row.kind) - const filters = parseFilters(row.filters) - const moved = await this.runtime.settings.database.transaction(async (connection) => { - const candidates = await scope.matching({ - connection, - filters, - limit: size, - deadBefore: Number(row.started_at_ms), - }) - if (candidates.length === 0) return 0 - const revived = await scope.revive({ - connection, - identifiers: candidates.map(({ id }) => id), - }) - await connection.run(`UPDATE ${this.table()} SET moved = moved + ? WHERE id = ?`, [ - revived, - row.id, - ]) - return revived + const candidates = await scope.matching({ + connection, + filters: parseFilters(row.filters), + limit: size, + deadBefore: Number(row.started_at_ms), }) - if (moved > 0) this.runtime.wakeUpAfterRevival(row.kind) - return moved + if (candidates.length === 0) return 0 + + const revived = await scope.revive({ + connection, + identifiers: candidates.map(({ id }) => id), + }) + await connection.run( + `UPDATE ${this.table()} SET moved = moved + ? WHERE id = ? AND status = 'running'`, + [revived, row.id], + ) + return revived } private batchSize(row: RedriveShape): number { @@ -163,30 +169,39 @@ export class RedriveManager { now, ], ) + await this.runtime.writeAdministrationEvent({ + connection, + action: "redrive.start", + kind: input.kind, + subjectId: id, + filters: input.filters, + actor: input.actor, + }) }) } catch { const running = await this.findRow(input.activeScope, "active_scope") if (running) return running throw new RedriveNotStarted(`could not start a ${input.kind} redrive`) } - const row = await this.require(id) - await this.audit({ row, action: "redrive.start" }) - return row + return await this.require(id) } - private async close(input: { id: string; status: RedriveStatus }): Promise { - await this.runtime.settings.database.transaction(async (connection) => { - const now = await connection.nowMilliseconds() - await connection.run( - `UPDATE ${this.table()} SET status = ?, active_scope = NULL, finished_at_ms = ? - WHERE id = ? AND status = 'running'`, - [input.status, now, input.id], - ) - }) - } + private async close(input: { + connection: DatabaseConnection + row: RedriveShape + status: RedriveStatus + action: string + }): Promise { + const now = await input.connection.nowMilliseconds() + const result = await input.connection.run( + `UPDATE ${this.table()} SET status = ?, active_scope = NULL, finished_at_ms = ? + WHERE id = ? AND status = 'running'`, + [input.status, now, input.row.id], + ) + if (result.changes === 0) return - private async audit(input: { row: RedriveShape; action: string }): Promise { await this.runtime.writeAdministrationEvent({ + connection: input.connection, action: input.action, kind: input.row.kind, subjectId: input.row.id, diff --git a/src/runtime.ts b/src/runtime.ts index a75e77f..2724b8c 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -889,51 +889,36 @@ export class SolidObjectsRuntime { this.wakeUp(kind === "effect" ? "effects" : "broadcasts") } - async administrationIdentity(authorizationContext: unknown): Promise { + async administrationIdentity( + authorizationContext: AdministrationOptions["authorizationContext"], + ): Promise { const identity = await this.settings.administrationIdentity(authorizationContext) return identity === null || identity === undefined ? null : String(identity).slice(0, 255) } - async recordAdministrationEvent(input: { - action: string - kind: string - subjectId?: string - filters?: RedriveFilters - authorizationContext?: unknown - }): Promise { - await this.writeAdministrationEvent({ - action: input.action, - kind: input.kind, - ...(input.subjectId === undefined ? {} : { subjectId: input.subjectId }), - ...(input.filters === undefined ? {} : { filters: input.filters }), - actor: await this.administrationIdentity(input.authorizationContext), - }) - } - async writeAdministrationEvent(input: { + connection: DatabaseConnection action: string kind: string subjectId?: string filters?: RedriveFilters actor: string | null }): Promise { - await this.settings.database.transaction(async (connection) => { - const occurredAt = await connection.nowMilliseconds() - await connection.run( - `INSERT INTO ${this.repository.table("administration_events")} - (id, action, kind, subject_id, filters, actor, occurred_at_ms) - VALUES (?, ?, ?, ?, ?, ?, ?)`, - [ - administrationEventId(occurredAt), - input.action, - input.kind, - input.subjectId ?? null, - input.filters === undefined ? null : JSON.stringify(input.filters), - input.actor, - occurredAt, - ], - ) - }) + const occurredAt = await input.connection.nowMilliseconds() + await input.connection.run( + `INSERT INTO ${this.repository.table("administration_events")} + (id, action, kind, subject_id, filters, actor, occurred_at_ms) + VALUES (?, ?, ?, ?, ?, ?, ?)`, + [ + administrationEventId(occurredAt), + input.action, + input.kind, + input.subjectId ?? null, + input.filters === undefined ? null : JSON.stringify(input.filters), + input.actor, + occurredAt, + ], + ) } async inspectDeadLetters(options: AdministrationOptions = {}): Promise { diff --git a/src/schema.ts b/src/schema.ts index 6050620..e968c33 100644 --- a/src/schema.ts +++ b/src/schema.ts @@ -1,5 +1,5 @@ import { UnsupportedDatabase } from "./errors.js" -import type { DatabaseConnection, DatabaseFamily } from "./database/types.js" +import type { DatabaseConnection, DatabaseFamily, RunResult } from "./database/types.js" const BASE_VERSION = 1 const RETRY_LINK_VERSION = 2 @@ -385,7 +385,7 @@ async function installRedrive(options: { table: (name: string) => string prefix: string schemaIdentity: string - createTable: (sql: string) => Promise + createTable: (sql: string) => Promise }): Promise { const { connection, family, table, prefix, schemaIdentity, createTable } = options await createTable(`CREATE TABLE IF NOT EXISTS ${table("administration_events")} ( @@ -441,17 +441,34 @@ async function addFailedAt(options: { family: DatabaseFamily table: string }): Promise { + if (await hasFailedAt(options)) return + const type = options.family === "sqlite" ? "INTEGER" : "BIGINT" - try { - await options.connection.run(`ALTER TABLE ${options.table} ADD COLUMN failed_at_ms ${type}`) - } catch { - return - } + await options.connection.run(`ALTER TABLE ${options.table} ADD COLUMN failed_at_ms ${type}`) await options.connection.run( `UPDATE ${options.table} SET failed_at_ms = available_at_ms WHERE status = 'dead'`, ) } +async function hasFailedAt(options: { + connection: DatabaseConnection + family: DatabaseFamily + table: string +}): Promise { + if (options.family === "sqlite") { + const columns = await options.connection.all<{ name: string }>( + `PRAGMA table_info(${options.table})`, + ) + return columns.some(({ name }) => name === "failed_at_ms") + } + const found = await options.connection.get<{ found: number | bigint }>( + `SELECT COUNT(*) AS found FROM information_schema.columns + WHERE table_name = ? AND column_name = 'failed_at_ms'`, + [options.table], + ) + return Number(found?.found ?? 0) > 0 +} + async function recordMigration(options: { connection: DatabaseConnection table: string diff --git a/test/redrive.test.ts b/test/redrive.test.ts index 08c6710..467216a 100644 --- a/test/redrive.test.ts +++ b/test/redrive.test.ts @@ -289,6 +289,30 @@ describe("redrive", () => { expect(await active.deadLetters.effects.all()).toHaveLength(1) }) + it("refuses an invalid filter rather than redrive everything", async () => { + const active = await start() + await deadEffects(active, 2) + + await expect( + active.deadLetters.effects.redrive({ failedAfter: new Date("nonsense") }), + ).rejects.toBeInstanceOf(TypeError) + await expect(active.deadLetters.effects.redrive({ limit: 0 })).rejects.toBeInstanceOf(TypeError) + await expect(active.deadLetters.effects.redrive({ limit: 1.5 })).rejects.toBeInstanceOf( + TypeError, + ) + + expect(await active.redrives.all()).toHaveLength(0) + expect(await auditRows(active)).toHaveLength(0) + }) + + it("writes no audit row when a retry names a row that does not exist", async () => { + const active = await start() + + await expect(active.deadLetters.effects.retry("missing")).rejects.toThrow() + + expect(await auditRows(active)).toHaveLength(0) + }) + it("refuses an unauthorized caller that reaches the manager directly", async () => { const active = configure({ database: sqlite({ path: ":memory:" }), From 492cd14d2a246cb008fe7d0b7fbca2bb52507a87 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 14:13:41 -0700 Subject: [PATCH 05/11] fix: look for the column in this database only `information_schema.columns` spans every schema on PostgreSQL and every database on MySQL, so a same-named table elsewhere could answer for this one. The column check would then skip the `ALTER`, record migration 11, and leave every later query referencing a column that is not there. The lookup narrows to `current_schema()` or `DATABASE()`, as the other metadata queries in this repository already do. `RedriveManager.start` and the filter parameters take the types the rest of the code uses, rather than restating `unknown`. Co-Authored-By: Claude Opus 5 (1M context) --- src/dead-letter-scopes.ts | 6 ++++-- src/redrive.ts | 2 +- src/schema.ts | 3 ++- 3 files changed, 7 insertions(+), 4 deletions(-) diff --git a/src/dead-letter-scopes.ts b/src/dead-letter-scopes.ts index f2a7028..319d744 100644 --- a/src/dead-letter-scopes.ts +++ b/src/dead-letter-scopes.ts @@ -6,6 +6,8 @@ import type { AdministrationOptions, DeepReadonly, JsonObject } from "./types.js export type DeadLetterKind = "effect" | "broadcast" +type DeadRowFilterValue = string | number + export interface DeadRow { readonly id: string readonly kind: DeadLetterKind @@ -157,9 +159,9 @@ export class DeadLetterScope { private conditions( filters: RedriveFilters, deadBefore?: number, - ): { where: string; parameters: unknown[] } { + ): { where: string; parameters: DeadRowFilterValue[] } { const clauses = ["dead.status = 'dead'"] - const parameters: unknown[] = [] + const parameters: DeadRowFilterValue[] = [] if (deadBefore !== undefined) { clauses.push("dead.failed_at_ms <= ?") parameters.push(deadBefore) diff --git a/src/redrive.ts b/src/redrive.ts index 530044a..8ed1f80 100644 --- a/src/redrive.ts +++ b/src/redrive.ts @@ -38,7 +38,7 @@ export class RedriveManager { async start(input: { kind: DeadLetterKind filters: RedriveFilters - authorizationContext?: unknown + authorizationContext?: AdministrationOptions["authorizationContext"] }): Promise { await this.runtime.deadLetters.scope(input.kind).authorize({ action: "redrive", diff --git a/src/schema.ts b/src/schema.ts index e968c33..992af65 100644 --- a/src/schema.ts +++ b/src/schema.ts @@ -461,9 +461,10 @@ async function hasFailedAt(options: { ) return columns.some(({ name }) => name === "failed_at_ms") } + const schema = options.family === "postgresql" ? "current_schema()" : "DATABASE()" const found = await options.connection.get<{ found: number | bigint }>( `SELECT COUNT(*) AS found FROM information_schema.columns - WHERE table_name = ? AND column_name = 'failed_at_ms'`, + WHERE table_schema = ${schema} AND table_name = ? AND column_name = 'failed_at_ms'`, [options.table], ) return Number(found?.found ?? 0) > 0 From 9b4f511048fe72e659ff486e0e8593f45c4ce899 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 14:25:24 -0700 Subject: [PATCH 06/11] fix: audit a message dead-letter retry A retry of an effect or a broadcast wrote an administration event and a retry of a message did not, so the one path that existed before this branch was the one the audit missed. The Ruby port records all three, and comparing the two suites is what surfaced it. The event goes in the same transaction as the retry, so a retry that commits cannot lose its event and a retry that raises cannot leave one. The suite gains the six cases the Ruby side already covered: a dead transmit effect replaying, a scope reading only dead rows, a cancel that cannot overwrite a finished task, a task reported as a frozen value, and audit rows for a broadcast and a message retry. Co-Authored-By: Claude Opus 5 (1M context) --- src/repository.ts | 2 + src/runtime.ts | 9 +++++ test/dead-letter-scopes.test.ts | 38 +++++++++++++++++++ test/redrive.test.ts | 65 +++++++++++++++++++++++++++++++++ 4 files changed, 114 insertions(+) diff --git a/src/repository.ts b/src/repository.ts index 642a005..3f06eb5 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -1140,8 +1140,10 @@ export class Repository { id: string initialState: JsonObject stateVersion: number + audit?: (connection: DatabaseConnection) => Promise }): Promise { return this.settings.database.transaction(async (connection) => { + await options.audit?.(connection) const deadLetter = await this.findDeadLetterInConnection(connection, options.id) if (!deadLetter) throw new UnknownDeadLetter(`unknown dead letter ${options.id}`) if (deadLetter.retried_message_id) { diff --git a/src/runtime.ts b/src/runtime.ts index 2724b8c..8aa3865 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -949,10 +949,19 @@ export class SolidObjectsRuntime { `unknown dead-letter operation ${JSON.stringify(deadLetter.operation)}`, ) } + const actorIdentity = await this.administrationIdentity(options.authorizationContext) const message = await this.repository.retryDeadLetter({ id, initialState: initialStateFor(actor.definition), stateVersion: actor.definition.stateVersion, + audit: (connection) => + this.writeAdministrationEvent({ + connection, + action: "dead_letter.retry", + kind: "message", + subjectId: id, + actor: actorIdentity, + }), }) this.emitInstrumentation("dead_letter.retried", { deadLetterId: id, diff --git a/test/dead-letter-scopes.test.ts b/test/dead-letter-scopes.test.ts index 0a1c989..9b044d4 100644 --- a/test/dead-letter-scopes.test.ts +++ b/test/dead-letter-scopes.test.ts @@ -3,6 +3,7 @@ import { Actor } from "../src/actor.js" import { sqlite } from "../src/database/sqlite.js" import { Unauthorized } from "../src/errors.js" import { configure, type SolidObjectsRuntime } from "../src/runtime.js" +import { TRANSMIT_EFFECT } from "../src/transmit-effect.js" class OrderActor extends Actor { static override readonly actorType = "RedriveOrderActor" @@ -19,6 +20,10 @@ class OrderActor extends Actor { this.count += 1 } + sendElsewhere(): void { + this.transmit().touch() + } + override observables(): Record { return { count: this.count } } @@ -57,6 +62,7 @@ async function start(): Promise { logger: { debug: () => {}, info: () => {}, warn: () => {}, error: () => {} }, }) created.registerEffect("settle", (argumentsValue) => settle(argumentsValue)) + created.registerEffect(TRANSMIT_EFFECT, (argumentsValue) => settle(argumentsValue)) created.register(OrderActor) created.register(PoisonActor) await created.install() @@ -137,6 +143,38 @@ describe("dead-letter scopes", () => { expect(await active.deadLetters.broadcasts.all()).toHaveLength(0) }) + it("replays a dead transmit effect", async () => { + const active = await start() + const transmitted: Record[] = [] + settle = () => { + throw new Error("carrier down") + } + await active.ref(OrderActor, "one").send.sendElsewhere() + await active.worker().runUntilIdle() + await active.effectWorker().runUntilIdle() + const dead = await active.deadLetters.effects.all() + expect(dead).toHaveLength(1) + + settle = (argumentsValue) => { + transmitted.push(argumentsValue) + } + await active.deadLetters.effects.retry(dead[0]!.id) + await active.effectWorker().runUntilIdle() + + expect(await active.deadLetters.effects.all()).toHaveLength(0) + expect(transmitted.map((argumentsValue) => argumentsValue["operation"])).toEqual(["touch"]) + }) + + it("reads only dead rows, not pending ones", async () => { + const active = await start() + const id = await deadEffect(active) + settle = () => {} + await active.ref(OrderActor, "two").send.place() + await active.worker().runUntilIdle() + + expect((await active.deadLetters.effects.all()).map((row) => row.id)).toEqual([id]) + }) + it("reads and retries message dead letters as it always has", async () => { const active = await start() await active.ref(PoisonActor, "one").send.run() diff --git a/test/redrive.test.ts b/test/redrive.test.ts index 467216a..b43511b 100644 --- a/test/redrive.test.ts +++ b/test/redrive.test.ts @@ -13,6 +13,10 @@ class PaymentActor extends Actor { this.placed += 1 this.emit("settle", { arguments: { order: "one" } }) } + + explode(): void { + throw new Error("poison message") + } } class ShipmentActor extends Actor { @@ -289,6 +293,33 @@ describe("redrive", () => { expect(await active.deadLetters.effects.all()).toHaveLength(1) }) + it("does not let a cancel overwrite a task the runner already finished", async () => { + const active = await start() + await deadEffects(active, 1) + const task = await active.deadLetters.effects.redrive() + await drain(active) + + await task.cancel() + + expect((await active.redrives.find(task.id)).status).toBe("completed") + expect((await auditRows(active)).map(({ action }) => action)).toEqual([ + "redrive.start", + "redrive.finish", + ]) + }) + + it("reports a task as a frozen value", async () => { + const active = await start() + await deadEffects(active, 1) + + const task = await active.deadLetters.effects.redrive() + + expect(Object.isFrozen(task)).toBe(true) + expect(task.kind).toBe("effect") + expect(task.startedAt).toBeInstanceOf(Date) + expect(task.finishedAt).toBeNull() + }) + it("refuses an invalid filter rather than redrive everything", async () => { const active = await start() await deadEffects(active, 2) @@ -358,6 +389,40 @@ describe("administration audit", () => { expect(events[0]).toMatchObject({ kind: "effect", subject_id: id }) }) + it("writes one row for a broadcast retry", async () => { + const active = await start() + await deadBroadcast(active) + const dead = await active.deadLetters.broadcasts.all() + + await active.deadLetters.broadcasts.retry(dead[0]!.id) + + const events = await auditRows(active) + expect(events).toHaveLength(1) + expect(events[0]).toMatchObject({ + action: "dead_letter.retry", + kind: "broadcast", + subject_id: dead[0]!.id, + }) + }) + + it("writes one row for a message retry", async () => { + const active = await start() + await active.ref(PaymentActor, "poison").send.explode() + await active.worker().runUntilIdle() + const letters = await active.deadLetters.all() + expect(letters).toHaveLength(1) + + await active.deadLetters.retry(letters[0]!.id) + + const events = await auditRows(active) + expect(events).toHaveLength(1) + expect(events[0]).toMatchObject({ + action: "dead_letter.retry", + kind: "message", + subject_id: letters[0]!.id, + }) + }) + it("writes one row for each redrive transition", async () => { const active = await start() await deadEffects(active, 5) From 34389ae56ab0a539a22d0053e1601e40cf775e33 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 14:36:33 -0700 Subject: [PATCH 07/11] test: prove an audit row follows its retry The Ruby suite covers a retry that raises after the lookup succeeds, in both the scope and the message path. The TypeScript suite covered only the lookup failure. Co-Authored-By: Claude Opus 5 (1M context) --- test/redrive.test.ts | 30 +++++++++++++++++++++++++++++- 1 file changed, 29 insertions(+), 1 deletion(-) diff --git a/test/redrive.test.ts b/test/redrive.test.ts index b43511b..7393d1e 100644 --- a/test/redrive.test.ts +++ b/test/redrive.test.ts @@ -1,4 +1,4 @@ -import { afterEach, describe, expect, it } from "vitest" +import { afterEach, describe, expect, it, vi } from "vitest" import { Actor } from "../src/actor.js" import { sqlite } from "../src/database/sqlite.js" import { Unauthorized } from "../src/errors.js" @@ -336,6 +336,34 @@ describe("redrive", () => { expect(await auditRows(active)).toHaveLength(0) }) + it("writes no audit row when the retry itself fails", async () => { + const active = await start() + await deadEffects(active, 1) + const id = (await active.deadLetters.effects.all())[0]!.id + const scope = active.deadLetters.effects + const revive = vi.spyOn(scope, "revive").mockRejectedValue(new Error("injected failure")) + + await expect(scope.retry(id)).rejects.toThrow("injected failure") + + expect(await auditRows(active)).toHaveLength(0) + revive.mockRestore() + }) + + it("writes no audit row when a message retry fails to enqueue", async () => { + const active = await start() + await active.ref(PaymentActor, "poison").send.explode() + await active.worker().runUntilIdle() + const letters = await active.deadLetters.all() + const enqueue = vi + .spyOn(active.repository, "enqueueInTransaction") + .mockRejectedValue(new Error("injected failure")) + + await expect(active.deadLetters.retry(letters[0]!.id)).rejects.toThrow("injected failure") + + expect(await auditRows(active)).toHaveLength(0) + enqueue.mockRestore() + }) + it("writes no audit row when a retry names a row that does not exist", async () => { const active = await start() From 9d7e0f9cb63ef3de34b3e8b4b61a199cb314f43d Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 14:37:07 -0700 Subject: [PATCH 08/11] docs: record dead-letter redrive parity Co-Authored-By: Claude Opus 5 (1M context) --- docs/parity.md | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/docs/parity.md b/docs/parity.md index 4136c80..d6361c3 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -142,6 +142,21 @@ already maintains. The Durable Objects engine keeps its own message and outbox tables inside each object, so these scopes cover the SQL backends here; its `deadLetters` call is unchanged. +## Dead letters, retry, and redrive + +Both runtimes scope dead letters by kind through one receiver, retry a dead +message, effect, or broadcast, redrive a whole scope as a durable and idempotent +task, bound that task to the rows that were dead when it started, advance it in +bounded batches, and write one administration event per retry and per task +transition, in the transaction that causes it. The filters, the resource names, +the audit actions, and the configuration defaults match. + +Two details differ. This runtime filters on a `failed_at_ms` stamp that schema +version 11 adds, while Ruby filters on the `updated_at` column Active Record +already maintains. The Durable Objects engine keeps its own message and outbox +tables inside each object, so these scopes cover the SQL backends here; its +`deadLetters` call is unchanged. + ## Realtime and browser behavior | Capability | Status | TypeScript shape or remaining work | From 72aa49429baad74e6a2c0fb7325d6dbb7c59d46b Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 14:45:11 -0700 Subject: [PATCH 09/11] docs: keep one parity section, and type the tests The parity round added a second "Dead letters, retry, and redrive" section beside the one already there. Only the newer, complete one remains. The new tests annotated observables, effect arguments, and result arrays as `unknown`. They name the shapes they actually carry. Co-Authored-By: Claude Opus 5 (1M context) --- docs/parity.md | 13 ------------- test/dead-letter-scopes.test.ts | 12 +++++++----- test/redrive.test.ts | 2 +- 3 files changed, 8 insertions(+), 19 deletions(-) diff --git a/docs/parity.md b/docs/parity.md index d6361c3..bbe13cc 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -131,19 +131,6 @@ only reaches code that changes the setting afterwards. ## Dead letters, retry, and redrive -Both runtimes scope dead letters by kind through one receiver, retry a dead -effect or broadcast, redrive a whole scope as a durable and idempotent task, -advance it in bounded batches, and write one administration event per retry and -per task transition. The filters, the resource names, and the audit shape match. - -Two details differ. This runtime filters on a `failed_at_ms` stamp that schema -version 11 adds, while Ruby filters on the `updated_at` column Active Record -already maintains. The Durable Objects engine keeps its own message and outbox -tables inside each object, so these scopes cover the SQL backends here; its -`deadLetters` call is unchanged. - -## Dead letters, retry, and redrive - Both runtimes scope dead letters by kind through one receiver, retry a dead message, effect, or broadcast, redrive a whole scope as a durable and idempotent task, bound that task to the rows that were dead when it started, advance it in diff --git a/test/dead-letter-scopes.test.ts b/test/dead-letter-scopes.test.ts index 9b044d4..7dbb463 100644 --- a/test/dead-letter-scopes.test.ts +++ b/test/dead-letter-scopes.test.ts @@ -24,7 +24,7 @@ class OrderActor extends Actor { this.transmit().touch() } - override observables(): Record { + override observables(): { count: number } { return { count: this.count } } } @@ -39,7 +39,9 @@ class PoisonActor extends Actor { } let runtime: SolidObjectsRuntime | undefined -let settle: (argumentsValue: Record) => void = () => {} +type SettleArguments = { order?: string; operation?: string } + +let settle: (argumentsValue: SettleArguments) => void = () => {} let deliver: () => void = () => {} afterEach(async () => { @@ -98,7 +100,7 @@ describe("dead-letter scopes", () => { it("returns a dead effect to pending and runs it again", async () => { const active = await start() const id = await deadEffect(active) - const settled: unknown[] = [] + const settled: SettleArguments[] = [] await active.deadLetters.effects.retry(id) @@ -145,7 +147,7 @@ describe("dead-letter scopes", () => { it("replays a dead transmit effect", async () => { const active = await start() - const transmitted: Record[] = [] + const transmitted: SettleArguments[] = [] settle = () => { throw new Error("carrier down") } @@ -162,7 +164,7 @@ describe("dead-letter scopes", () => { await active.effectWorker().runUntilIdle() expect(await active.deadLetters.effects.all()).toHaveLength(0) - expect(transmitted.map((argumentsValue) => argumentsValue["operation"])).toEqual(["touch"]) + expect(transmitted.map(({ operation }) => operation)).toEqual(["touch"]) }) it("reads only dead rows, not pending ones", async () => { diff --git a/test/redrive.test.ts b/test/redrive.test.ts index 7393d1e..1fa5c56 100644 --- a/test/redrive.test.ts +++ b/test/redrive.test.ts @@ -28,7 +28,7 @@ class ShipmentActor extends Actor { this.count += 1 } - override observables(): Record { + override observables(): { count: number } { return { count: this.count } } } From db23982202cd2d51ea155ffcd5e9c78b70d2ff40 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 14:48:15 -0700 Subject: [PATCH 10/11] docs: state what an administration event records An event records an authorized press, not a state transition, and the redrive transitions are the opposite. The rule was implicit in the tests. Co-Authored-By: Claude Opus 5 (1M context) --- docs/operations.md | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/docs/operations.md b/docs/operations.md index 0fb7bf0..375d061 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -298,7 +298,15 @@ own resource name: `dead_letters`, `effect_dead_letters`, writes one row to `solid_objects_administration_events`, holding the action, the kind, the subject, the identity, and when it happened. The identity comes from `administrationIdentity`, which receives the authorization context the caller -passed and defaults to its `String` form. A refused caller writes nothing. +passed and defaults to its `String` form. + +An event records an authorized press, not a state transition. Pressing retry +twice writes two rows, because an operator did two things and a log that shows +one cannot answer who pressed what. The row the event names carries the outcome. +A refused caller writes nothing, and a retry that raises after the lookup writes +nothing, because the event shares the transaction with the work. The redrive +transitions are different: `redrive.start`, `redrive.finish`, and +`redrive.cancel` are written only when the task actually changes. The Durable Objects engine keeps its own message and outbox tables inside each object, so these scopes and redrive cover the SQL backends. `deadLetters` there From 1aad83304c1d01cededb3fa69f5aadcef11d7a8f Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 14:58:38 -0700 Subject: [PATCH 11/11] docs: point the dashboard at the new scopes The Ruby dashboard guide says the dashboard does not yet surface dead effects and broadcasts, and where the API for them lives. This one did not. Co-Authored-By: Claude Opus 5 (1M context) --- docs/dashboard.md | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/docs/dashboard.md b/docs/dashboard.md index 547a525..1313c9a 100644 --- a/docs/dashboard.md +++ b/docs/dashboard.md @@ -141,6 +141,15 @@ Dead-letter retry calls `runtime.deadLetters.retry()`. It keeps the durable idempotency and the actor-operation validation of that method. If the runtime refuses a retry, the detail page shows it with status 422. +Dead effects and broadcasts have the same API, which the dashboard does not yet +surface. `runtime.deadLetters.effects` and `runtime.deadLetters.broadcasts` read +and retry their own kind, and `redrive` moves a whole scope as a durable task. +See [Operations](operations.md) for both. + +Every retry and every redrive transition writes one row to +`solid_objects_administration_events`, holding the action, the kind, the +subject, and the identity that asked for it. + `HEAD /` performs only a schema reachability query and creates no CSRF session state. Use it for liveness checks instead of polling the full dashboard.