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/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. diff --git a/docs/operations.md b/docs/operations.md index bfb00d1..375d061 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -227,3 +227,91 @@ 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. + +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 +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..bbe13cc 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -129,6 +129,21 @@ 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 +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 | diff --git a/src/configuration.ts b/src/configuration.ts index b55ab38..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 @@ -82,6 +89,11 @@ export interface SolidObjectsConfiguration { instrumentation?: (event: InstrumentationEvent) => void broadcast?: (event: BroadcastEvent) => Promise wakeUp?: WakeUpSetting + redriveBatchSize?: number + redriveBatchPauseMilliseconds?: number + administrationIdentity?: ( + authorizationContext: AdministrationOptions["authorizationContext"], + ) => string | null | Promise } export interface BroadcastEvent { @@ -156,6 +168,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 +300,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..319d744 --- /dev/null +++ b/src/dead-letter-scopes.ts @@ -0,0 +1,245 @@ +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" + +type DeadRowFilterValue = string | number + +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 }) + 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) + return this.deadRow(row) + } + + async redrive(options: RedriveOptions = {}): Promise { + return await this.runtime.redrives.start({ + kind: this.kind, + filters: { + actorType: options.actorType ?? null, + failedAfter: failedAfterFilter(options.failedAfter), + limit: limitFilter(options.limit), + }, + 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]) + } + + private conditions( + filters: RedriveFilters, + deadBefore?: number, + ): { where: string; parameters: DeadRowFilterValue[] } { + const clauses = ["dead.status = 'dead'"] + const parameters: DeadRowFilterValue[] = [] + 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), + }) + } + + 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, + }) + } +} + +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 }) +} + +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..8ed1f80 --- /dev/null +++ b/src/redrive.ts @@ -0,0 +1,336 @@ +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?: AdministrationOptions["authorizationContext"] + }): 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) + + 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.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 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({ connection, row }) + if (moved > 0) return { moved, kind: row.kind } + + 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(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 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 candidates = await scope.matching({ + connection, + filters: parseFilters(row.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 = ? AND status = 'running'`, + [revived, row.id], + ) + return revived + } + + 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, + ], + ) + 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`) + } + return await this.require(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 + + await this.runtime.writeAdministrationEvent({ + connection: input.connection, + 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) + }, + }) + } + + 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, + }) + } +} + +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..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) { @@ -1472,6 +1474,8 @@ export class Repository { async resetForTesting(): Promise { const tables = [ + "administration_events", + "redrives", "effect_recoveries", "dead_letters", "claimed_messages", @@ -1592,9 +1596,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 +2053,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..8aa3865 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,20 @@ function scheduledReminderOf(row: ReminderRow): ScheduledReminder { } } +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 +224,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 +885,42 @@ export class SolidObjectsRuntime { return destroyed } + wakeUpAfterRevival(kind: DeadLetterKind): void { + this.wakeUp(kind === "effect" ? "effects" : "broadcasts") + } + + 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 writeAdministrationEvent(input: { + connection: DatabaseConnection + action: string + kind: string + subjectId?: string + filters?: RedriveFilters + actor: string | null + }): Promise { + 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 { await this.authorizeAdministration({ action: "inspect", @@ -899,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, @@ -1634,6 +1693,7 @@ export class SolidObjectsRuntime { () => () => this.broadcastWorker(), ) : []), + () => new RedriveScheduler(this), ...(this.settings.retentionIntervalMilliseconds > 0 ? [ () => @@ -2043,7 +2103,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..992af65 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 @@ -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,97 @@ export async function installSchema(options: { }) } +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 { + if (await hasFailedAt(options)) return + + const type = options.family === "sqlite" ? "INTEGER" : "BIGINT" + 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 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_schema = ${schema} AND 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/dead-letter-scopes.test.ts b/test/dead-letter-scopes.test.ts new file mode 100644 index 0000000..7dbb463 --- /dev/null +++ b/test/dead-letter-scopes.test.ts @@ -0,0 +1,238 @@ +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" +import { TRANSMIT_EFFECT } from "../src/transmit-effect.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 + } + + sendElsewhere(): void { + this.transmit().touch() + } + + override observables(): { count: number } { + 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 +type SettleArguments = { order?: string; operation?: string } + +let settle: (argumentsValue: SettleArguments) => 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.registerEffect(TRANSMIT_EFFECT, (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: SettleArguments[] = [] + + 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("replays a dead transmit effect", async () => { + const active = await start() + const transmitted: SettleArguments[] = [] + 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(({ operation }) => 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() + 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..1fa5c56 --- /dev/null +++ b/test/redrive.test.ts @@ -0,0 +1,521 @@ +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" +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" } }) + } + + explode(): void { + throw new Error("poison message") + } +} + +class ShipmentActor extends Actor { + static override readonly actorType = "RedriveShipmentActor" + + count = 0 + + touch(): void { + this.count += 1 + } + + override observables(): { count: number } { + 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("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) + + 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 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() + + 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:" }), + 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:" }), + 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 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) + + 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`, + ), + ) +}