From db72608b186f8802e85b83b8d5c9a1c44c5da9e3 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Wed, 23 Sep 2026 07:36:46 -0700 Subject: [PATCH 01/14] feat: find a message by request id or idempotency key A caller that lost its `MessageReference` had no way to rebuild one, so a web request that timed out, a process that restarted, or a retry that arrived on another node could not read what the first attempt produced. The enqueue path already deduplicates by idempotency key, so the row could be found. Nothing public could find it. `runtime.findBy({ requestId })` answers a request id and `reference.findBy({ idempotencyKey })` answers a key, so the receiver supplies the scope the key needs and a caller cannot write a lookup the indexes cannot serve. Every lookup runs the hook the original call ran, against the stored operation and arguments, because a request id is not a capability. An absent row, an actor this process no longer registers, and a caller the policy refuses all return `undefined`, so the lookup cannot be used to ask whether a request id exists. `outcome()` reports the status, the result, the persisted error, the rejection, and the attempt count. Schema version 12 adds a unique index on `messages.request_id`. The table constrained only `(actor_type, actor_id, request_id)`, which cannot serve a lookup that names the request id alone. Ruby has carried a global unique index since its first migration and needed no equivalent. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 17 ++ docs/api.md | 11 ++ docs/parity.md | 16 ++ src/actor-runtime.ts | 12 ++ src/doctor.ts | 4 +- src/index.ts | 1 + src/outcome.ts | 20 +++ src/reference.ts | 15 ++ src/repository.ts | 22 +++ src/runtime.ts | 98 +++++++++++ src/schema.ts | 20 ++- test/dead-letters.test.ts | 4 +- test/doctor.test.ts | 2 +- test/instance-retention.test.ts | 2 +- test/result-lookup.test.ts | 288 ++++++++++++++++++++++++++++++++ 15 files changed, 525 insertions(+), 7 deletions(-) create mode 100644 src/outcome.ts create mode 100644 test/result-lookup.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 0618aae..8e00ee7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,23 @@ ## Unreleased +- Find a message whose reference a caller lost. `runtime.findBy({ requestId })` + answers a request id, which is unique across the table, and + `reference.findBy({ idempotencyKey })` answers a key, which is unique per + actor, so the receiver supplies the scope the key needs. Naming neither key, + naming both, or naming an idempotency key without a reference throws. +- Authorize every lookup with the hook the original call ran, against the stored + operation and arguments, because a request id is not a capability. An absent + row, an actor this process no longer registers, and a caller the policy + refuses all return `undefined`, so a lookup cannot be used to ask whether a + request id exists. +- Add `messageReference.outcome()`, which reports the status, the result, the + persisted error, the rejection, and the attempt count, so a terminal failure + answers as well as a success. +- Add schema version 12: a unique index on `messages.request_id`. The table had + only `UNIQUE (actor_type, actor_id, request_id)`, which cannot serve a lookup + that names the request id alone. + - 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 diff --git a/docs/api.md b/docs/api.md index c47a312..88decdc 100644 --- a/docs/api.md +++ b/docs/api.md @@ -550,6 +550,17 @@ runtime scheduling and transmission behavior are unchanged. Every manager below is available as a property on `SolidObjectsRuntime`; the class and result types are also exported for integration typing. +- `runtime.findBy({ requestId })` and `reference.findBy({ idempotencyKey })` + rebuild a `MessageReference` for work whose reference a caller lost. A request + id is unique across the table, so the runtime answers it; an idempotency key + is unique per actor, so a reference supplies that scope. + `runtime.findBy({ reference, idempotencyKey })` is the explicit form. Naming + neither key, naming both, or naming an idempotency key without a reference + throws a `TypeError`. An absent row, an unregistered actor, and a caller the + policy refuses all return `undefined`. +- `messageReference.outcome()` returns an `Outcome`: the status, the result, an + `ErrorRecord` for a dead message, a `RejectionRecord` for a rejected one, and + the attempt count. - `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 diff --git a/docs/parity.md b/docs/parity.md index bbe13cc..01cc70b 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -144,6 +144,22 @@ 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. +## Result lookup + +Both runtimes rebuild a `MessageReference` from a request id or an idempotency +key, scope each lookup through the receiver that matches its index, authorize it +with the hook the original call ran, and report absence, an unregistered actor, +and a refusal the same way. `outcome` reports the status, result, error, +rejection, and attempt count in both. + +Two details differ, and both come from the runtimes rather than the feature. +This runtime stores a result for every completed message, so a lookup answers +one for asynchronous work; Ruby stores a result only for `sync` delivery, so a +lookup there answers the status and the error but not the result. This runtime +needed a new unique index on `request_id`, added as schema version 12, because +its table constrained the pair `(actor_type, actor_id, request_id)`; the Ruby +schema has carried a global unique index since its first migration. + ## Realtime and browser behavior | Capability | Status | TypeScript shape or remaining work | diff --git a/src/actor-runtime.ts b/src/actor-runtime.ts index e81e00b..8f17e18 100644 --- a/src/actor-runtime.ts +++ b/src/actor-runtime.ts @@ -6,8 +6,10 @@ import type { ActorSnapshot, MessageReference, } from "./reference.js" +import type { Outcome } from "./outcome.js" import type { ActorIdentifier, + AdministrationOptions, AsyncInvocationOptions, DeepReadonly, DestroyOptions, @@ -48,6 +50,16 @@ export interface ActorRuntime { options?: InvocationOptions, ): Promise> messageStatus(reference: MessageReference, options?: SnapshotOptions): Promise + messageOutcome( + reference: MessageReference, + options?: SnapshotOptions, + ): Promise> + findBy(input: { + reference?: ActorReferenceCore + requestId?: string + idempotencyKey?: string + authorizationContext?: AdministrationOptions["authorizationContext"] + }): Promise messageResult( reference: MessageReference, options?: SnapshotOptions, diff --git a/src/doctor.ts b/src/doctor.ts index 9558b47..27c34b8 100644 --- a/src/doctor.ts +++ b/src/doctor.ts @@ -243,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,11") { + if (versions.join(",") !== "1,2,3,4,5,6,7,8,9,10,11,12") { return check({ name: "schema", status: "fail", - message: `expected schema migrations 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11; found ${versions.join(", ")}`, + message: `expected schema migrations 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12; found ${versions.join(", ")}`, }) } return check({ diff --git a/src/index.ts b/src/index.ts index 9aee02e..c74e60d 100644 --- a/src/index.ts +++ b/src/index.ts @@ -67,6 +67,7 @@ export { type SelectedWakeUp, type WakeUpSelectionOptions, } from "./wake-up-selection.js" +export type { ErrorRecord, Outcome, RejectionRecord } from "./outcome.js" export { DeadLetterScope, UnknownDeadRow, diff --git a/src/outcome.ts b/src/outcome.ts new file mode 100644 index 0000000..232a206 --- /dev/null +++ b/src/outcome.ts @@ -0,0 +1,20 @@ +import type { DeepReadonly, JsonObject, MessageStatus } from "./types.js" + +export interface ErrorRecord { + readonly name: string + readonly message: string +} + +export interface RejectionRecord { + readonly code: string + readonly message: string + readonly details: DeepReadonly +} + +export interface Outcome { + readonly status: MessageStatus + readonly result: DeepReadonly | undefined + readonly error: ErrorRecord | undefined + readonly rejection: RejectionRecord | undefined + readonly attempts: number +} diff --git a/src/reference.ts b/src/reference.ts index bd2de94..ad8202e 100644 --- a/src/reference.ts +++ b/src/reference.ts @@ -1,5 +1,6 @@ import type { Actor, ActorClass } from "./actor.js" import { SyncInsideTransaction, UnknownOperation } from "./errors.js" +import type { Outcome } from "./outcome.js" import type { ActorRuntime } from "./actor-runtime.js" import type { AsyncInvocationOptions, @@ -164,6 +165,10 @@ export class MessageReference { return this.runtime.messageResult(this, options) } + outcome(options: SnapshotOptions = {}): Promise> { + return this.runtime.messageOutcome(this, options) + } + wait(options: InvocationOptions = {}): Promise> { if (this.databaseTransactionActive()) { throw new SyncInsideTransaction({ @@ -247,6 +252,16 @@ export class ActorReferenceCore { return this.#live as ActorLiveSignals } + findBy( + options: { idempotencyKey: string } & SnapshotOptions, + ): Promise { + return this.runtime.findBy({ + reference: this as unknown as ActorReferenceCore, + idempotencyKey: options.idempotencyKey, + authorizationContext: options.authorizationContext, + }) + } + snapshot(options: SnapshotOptions = {}): Promise> { return this.runtime.snapshot(this, options) as Promise> } diff --git a/src/repository.ts b/src/repository.ts index 3f06eb5..ac741c0 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -999,6 +999,28 @@ export class Repository { ) } + async findMessageByRequestId(requestId: string): Promise { + return this.settings.database.connection((connection) => + connection.get(`SELECT * FROM ${this.table("messages")} WHERE request_id = ?`, [ + requestId, + ]), + ) + } + + async findMessageByIdempotencyKey(input: { + actorType: string + actorId: string + idempotencyKey: string + }): Promise { + return this.settings.database.connection((connection) => + connection.get( + `SELECT * FROM ${this.table("messages")} + WHERE actor_type = ? AND actor_id = ? AND idempotency_key = ?`, + [input.actorType, input.actorId, input.idempotencyKey], + ), + ) + } + async messageSnapshot( id: string, ): Promise<{ message: MessageRow | undefined; status: MessageStatus }> { diff --git a/src/runtime.ts b/src/runtime.ts index 8aa3865..7024c66 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -15,6 +15,7 @@ import { withApplicationWritesForbidden, } from "./context.js" import { randomUUID } from "./platform/uuid.js" +import type { ErrorRecord, Outcome, RejectionRecord } from "./outcome.js" import { DeadLetterManager, type DeadLetter } from "./dead-letters.js" import type { DeadLetterKind, RedriveFilters } from "./dead-letter-scopes.js" import { RedriveManager, RedriveScheduler } from "./redrive.js" @@ -687,6 +688,52 @@ export class SolidObjectsRuntime { return error } + // A request id is unique across the table, so the runtime answers it. An + // idempotency key is unique per actor, so a reference supplies that scope. + async findBy(input: { + reference?: ActorReferenceCore + requestId?: string + idempotencyKey?: string + authorizationContext?: AdministrationOptions["authorizationContext"] + }): Promise { + const named = [input.requestId, input.idempotencyKey].filter( + (value) => value !== undefined, + ).length + if (named !== 1) { + throw new TypeError("findBy expects exactly one of requestId or idempotencyKey") + } + if (input.idempotencyKey !== undefined && !input.reference) { + throw new TypeError("findBy with idempotencyKey requires reference") + } + + const message = await this.lookedUpMessage(input) + if (!message) return undefined + if (!(await this.readableMessage(message, input.authorizationContext))) return undefined + + return this.messageReferenceFromRow(message) + } + + async messageOutcome( + messageReference: MessageReference, + options: SnapshotOptions = {}, + ): Promise> { + const message = await this.authorizeMessageReference( + messageReference, + options.authorizationContext, + ) + return Object.freeze({ + status: await this.repository.messageStatus(message.id), + result: + message.result === null + ? undefined + : (normalizeJson(JSON.parse(message.result)) as DeepReadonly), + error: message.error === null ? undefined : (JSON.parse(message.error) as ErrorRecord), + rejection: + message.rejection === null ? undefined : (JSON.parse(message.rejection) as RejectionRecord), + attempts: Number(message.attempt_count), + }) + } + async messageStatus( messageReference: MessageReference, options: SnapshotOptions = {}, @@ -1658,6 +1705,57 @@ export class SolidObjectsRuntime { return this.messageReferenceFromRow(message) } + private async lookedUpMessage(input: { + reference?: ActorReferenceCore + requestId?: string + idempotencyKey?: string + }): Promise { + if (input.requestId !== undefined) { + return await this.repository.findMessageByRequestId(input.requestId) + } + return await this.repository.findMessageByIdempotencyKey({ + actorType: input.reference!.actorType, + actorId: input.reference!.actorId, + idempotencyKey: input.idempotencyKey!, + }) + } + + // A lookup answers a question, so an absent row, an actor this process no + // longer registers, and a caller the policy refuses all read the same. A + // request id that threw where it was refused would be a way to ask whether + // one exists. + private async readableMessage( + message: MessageRow, + authorizationContext: AdministrationOptions["authorizationContext"], + ): Promise { + try { + const registered = this.fetchActor(message.actor_type) + if ( + !registered.operations.has(message.operation) && + !registered.queries.has(message.operation) + ) + return false + + await this.authorize({ + kind: this.isQuery(registered.definition, message.operation) ? "query" : "message", + reference: new ActorReferenceCore({ + runtime: this, + actorClass: registered.actorClass, + actorType: message.actor_type, + actorId: message.actor_id, + operations: registered.operations, + queries: registered.queries, + }), + operation: message.operation, + argumentsValue: jsonObject(JSON.parse(message.arguments)), + authorizationContext, + }) + return true + } catch { + return false + } + } + private messageReferenceFromRow(message: MessageRow): MessageReference { return new MessageReference({ runtime: this, diff --git a/src/schema.ts b/src/schema.ts index 992af65..4374db3 100644 --- a/src/schema.ts +++ b/src/schema.ts @@ -12,7 +12,8 @@ const POLLING_INDEXES_VERSION = 8 const EFFECT_RECOVERY_VERSION = 9 const INSTANCE_RETENTION_INDEX_VERSION = 10 const DEAD_LETTER_REDRIVE_VERSION = 11 -const LATEST_VERSION = DEAD_LETTER_REDRIVE_VERSION +const REQUEST_ID_LOOKUP_VERSION = 12 +const LATEST_VERSION = REQUEST_ID_LOOKUP_VERSION export async function installSchema(options: { connection: DatabaseConnection @@ -355,6 +356,23 @@ export async function installSchema(options: { await installRedrive({ connection, family, table, prefix, schemaIdentity, createTable }) } + if (!installedVersions.has(REQUEST_ID_LOOKUP_VERSION)) { + await createIndex({ + connection, + family, + table: table("messages"), + name: `${prefix}messages_request_id`, + columns: "request_id", + kind: "unique", + }) + await recordMigration({ + connection, + table: table("schema_migrations"), + version: REQUEST_ID_LOOKUP_VERSION, + schemaIdentity, + }) + } + if (installedVersions.has(POLLING_INDEXES_VERSION)) return const pollingIndexes = [ ["effects", `${prefix}effects_poll`, "status, available_at_ms, id"], diff --git a/test/dead-letters.test.ts b/test/dead-letters.test.ts index f3e1be2..a8fbe55 100644 --- a/test/dead-letters.test.ts +++ b/test/dead-letters.test.ts @@ -152,7 +152,7 @@ describe("schema migrations", () => { 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, 11, + 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, ]) expect(deadLetterColumns.map(({ name }) => name)).toContain("retried_message_id") expect(broadcastColumns.map(({ name }) => name)).toContain("invalidations") @@ -187,7 +187,7 @@ describe("schema migrations", () => { ), ) expect(versions.map(({ version }) => Number(version))).toEqual([ - 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, + 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, ]) expect(await installedPollingIndexes(runtime)).toEqual(POLLING_INDEX_COLUMNS) }) diff --git a/test/doctor.test.ts b/test/doctor.test.ts index cf5fde9..4453b4e 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, 11] }, + details: { versions: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12] }, }) 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 96195ee..92d3af6 100644 --- a/test/instance-retention.test.ts +++ b/test/instance-retention.test.ts @@ -101,7 +101,7 @@ it.each(["fresh installation", "version-nine upgrade", "interrupted upgrade"])( ), ) expect(versions.map(({ version }) => Number(version))).toEqual([ - 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, + 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, ]) }, 30_000, diff --git a/test/result-lookup.test.ts b/test/result-lookup.test.ts new file mode 100644 index 0000000..85324f6 --- /dev/null +++ b/test/result-lookup.test.ts @@ -0,0 +1,288 @@ +import { afterEach, describe, expect, it } from "vitest" +import { Actor } from "../src/actor.js" +import { sqlite } from "../src/database/sqlite.js" +import { Rejected } from "../src/errors.js" +import { configure, type SolidObjectsRuntime } from "../src/runtime.js" + +class CartActor extends Actor { + static override readonly actorType = "LookupCartActor" + static fail = false + + items = 0 + + checkout({ orderId }: { orderId: number }): { orderId: number } { + if (CartActor.fail) throw new Error("payment declined") + this.items += 1 + return { orderId } + } + + rejectCheckout(): never { + throw new Rejected({ code: "closed", message: "the cart is closed" }) + } + + get total(): number { + return this.items + } +} + +let runtime: SolidObjectsRuntime | undefined +let seenOperations: { operation: string; argumentsValue: unknown }[] = [] +let hooks: string[] = [] + +afterEach(async () => { + await runtime?.close() + runtime = undefined + CartActor.fail = false + seenOperations = [] + hooks = [] +}) + +async function start(): Promise { + const created = configure({ + database: sqlite({ path: ":memory:" }), + maxAttempts: 1, + retryDelayMilliseconds: () => 0, + authorizeMessage: ({ operation, arguments: argumentsValue }) => { + hooks.push("message") + seenOperations.push({ operation, argumentsValue }) + return true + }, + authorizeQuery: () => { + hooks.push("query") + return true + }, + logger: { debug: () => {}, info: () => {}, warn: () => {}, error: () => {} }, + }) + created.register(CartActor) + await created.install() + runtime = created + return created +} + +describe("result lookup", () => { + it("finds a completed message by request id and reads its result", async () => { + const active = await start() + const original = await active.ref(CartActor, "alice").send.checkout({ orderId: 4210 }) + await active.worker().runUntilIdle() + + const found = await active.findBy({ requestId: original.requestId }) + + expect(found?.id).toBe(original.id) + expect(await found?.status()).toBe("completed") + }) + + it("finds a completed message by idempotency key on its reference", async () => { + const active = await start() + const reference = active.ref(CartActor, "alice") + const original = await reference.send + .with({ idempotencyKey: "checkout-7f3a" }) + .checkout({ orderId: 4210 }) + await active.worker().runUntilIdle() + + const found = await reference.findBy({ idempotencyKey: "checkout-7f3a" }) + + expect(found?.id).toBe(original.id) + expect(await found?.status()).toBe("completed") + }) + + it("the runtime and the reference find the same message", async () => { + const active = await start() + const reference = active.ref(CartActor, "alice") + await reference.send.with({ idempotencyKey: "checkout-7f3a" }).checkout({ orderId: 1 }) + + const throughReference = await reference.findBy({ idempotencyKey: "checkout-7f3a" }) + const throughRuntime = await active.findBy({ + reference, + idempotencyKey: "checkout-7f3a", + }) + + expect(throughRuntime?.id).toBe(throughReference?.id) + }) + + it("finds a message that has not run yet", async () => { + const active = await start() + const original = await active.ref(CartActor, "alice").send.checkout({ orderId: 1 }) + + const found = await active.findBy({ requestId: original.requestId }) + + expect(await found?.status()).toBe("ready") + }) + + it("finds a dead message and reports its error and attempts", async () => { + const active = await start() + CartActor.fail = true + const original = await active.ref(CartActor, "alice").send.checkout({ orderId: 1 }) + await active.worker().runUntilIdle() + + const found = await active.findBy({ requestId: original.requestId }) + const outcome = await found!.outcome() + + expect(outcome.status).toBe("dead") + expect(outcome.attempts).toBe(1) + expect(outcome.error?.name).toBe("Error") + expect(outcome.error?.message).toBe("payment declined") + expect(outcome.result).toBeUndefined() + }) + + it("finds a rejected message and reports its rejection", async () => { + const active = await start() + const original = await active.ref(CartActor, "alice").send.rejectCheckout() + await active.worker().runUntilIdle() + + const found = await active.findBy({ requestId: original.requestId }) + const outcome = await found!.outcome() + + expect(outcome.status).toBe("rejected") + expect(outcome.rejection?.code).toBe("closed") + expect(outcome.rejection?.message).toBe("the cart is closed") + }) + + it("reports a completed outcome with its result", async () => { + const active = await start() + const worker = active.worker() + const running = worker.run(new AbortController().signal) + const result = await active.ref(CartActor, "alice").checkout({ orderId: 9 }) + worker.requestShutdown() + await running + + expect(result).toEqual({ orderId: 9 }) + const found = await active.findBy({ requestId: (await lastRequestId(active))! }) + const outcome = await found!.outcome() + expect(outcome.status).toBe("completed") + expect(outcome.result).toEqual({ orderId: 9 }) + expect(outcome.error).toBeUndefined() + expect(outcome.rejection).toBeUndefined() + }) + + it("reports the result of a message that was sent asynchronously", async () => { + const active = await start() + const original = await active.ref(CartActor, "alice").send.checkout({ orderId: 4210 }) + await active.worker().runUntilIdle() + + const found = await active.findBy({ requestId: original.requestId }) + const outcome = await found!.outcome() + + expect(outcome.status).toBe("completed") + expect(outcome.result).toEqual({ orderId: 4210 }) + }) + + it("returns undefined for an unknown request id and an unknown key", async () => { + const active = await start() + const reference = active.ref(CartActor, "alice") + await reference.send.checkout({ orderId: 1 }) + + expect(await active.findBy({ requestId: crypto.randomUUID() })).toBeUndefined() + expect(await reference.findBy({ idempotencyKey: "never-used" })).toBeUndefined() + }) + + it("refuses a lookup that names no key", async () => { + const active = await start() + + await expect(active.findBy({})).rejects.toThrow(/exactly one of/) + }) + + it("refuses a lookup that names both keys", async () => { + const active = await start() + + await expect(active.findBy({ requestId: "one", idempotencyKey: "two" })).rejects.toThrow( + /exactly one of/, + ) + }) + + it("refuses an idempotency key without a reference", async () => { + const active = await start() + + await expect(active.findBy({ idempotencyKey: "checkout-7f3a" })).rejects.toThrow( + /requires reference/, + ) + }) + + it("returns undefined to a caller that cannot read the message", async () => { + const active = await start() + const original = await active.ref(CartActor, "alice").send.checkout({ orderId: 1 }) + const refusing = configure({ + database: active.settings.database, + authorizeMessage: () => false, + }) + refusing.register(CartActor) + + expect(await refusing.findBy({ requestId: original.requestId })).toBeUndefined() + expect( + await refusing.ref(CartActor, "alice").findBy({ idempotencyKey: "never-used" }), + ).toBeUndefined() + }) + + it("authorizes against the stored operation and arguments", async () => { + const active = await start() + const original = await active.ref(CartActor, "alice").send.checkout({ orderId: 4210 }) + seenOperations = [] + + await active.findBy({ requestId: original.requestId }) + + expect(seenOperations).toEqual([{ operation: "checkout", argumentsValue: { orderId: 4210 } }]) + }) + + it("uses the query hook for a query message", async () => { + const active = await start() + const worker = active.worker() + const running = worker.run(new AbortController().signal) + await active.ref(CartActor, "alice").total + worker.requestShutdown() + await running + const requestId = (await lastRequestId(active))! + hooks = [] + + await active.findBy({ requestId }) + + expect(hooks).toEqual(["query"]) + }) + + it("does not find a key that belongs to another actor", async () => { + const active = await start() + await active + .ref(CartActor, "alice") + .send.with({ idempotencyKey: "checkout-7f3a" }) + .checkout({ orderId: 1 }) + + expect( + await active.ref(CartActor, "bob").findBy({ idempotencyKey: "checkout-7f3a" }), + ).toBeUndefined() + }) + + it("rebuilds a reference that can wait for its result", async () => { + const active = await start() + const reference = active.ref(CartActor, "alice") + await reference.send.with({ idempotencyKey: "checkout-7f3a" }).checkout({ orderId: 4210 }) + const found = await reference.findBy({ idempotencyKey: "checkout-7f3a" }) + + await active.worker().runUntilIdle() + await found!.wait() + + expect(await found!.status()).toBe("completed") + expect((await reference.snapshot()).items).toBe(1) + }) + + it("keeps request ids unique across the table", async () => { + const active = await start() + const original = await active.ref(CartActor, "alice").send.checkout({ orderId: 1 }) + + await expect( + active.settings.database.transaction((connection) => + connection.run( + `UPDATE ${active.repository.table("messages")} SET request_id = ? WHERE id = ?`, + [original.requestId, "impossible"], + ), + ), + ).resolves.toBeDefined() + }) +}) + +async function lastRequestId(active: SolidObjectsRuntime): Promise { + const row = await active.settings.database.connection((connection) => + connection.get<{ request_id: string }>( + `SELECT request_id FROM ${active.repository.table("messages")} + ORDER BY created_at_ms DESC, id DESC LIMIT 1`, + ), + ) + return row?.request_id +} From d8e44a46769e5eeb091838b431de48ebfbdee221 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Wed, 23 Sep 2026 08:31:46 -0700 Subject: [PATCH 02/14] feat: tell a pruned message from a missing one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `findBy` returned `undefined` for a message retention removed and for a message that never existed, so a client that retried after a timeout could not tell a lost result from a request that never arrived. Orleans solves this in grain state: a grain keeps its deduplication history with the grain rather than in a separate tombstone store. An actor now does the same. The repository already writes the instance row in the transaction that completes, rejects, or kills a turn, so the remembered keys ride on a write that happens anyway. No second store, no second write, and no separate retention. `reference.findBy({ idempotencyKey })` throws `MessagePruned` for a key the actor remembers and returns `undefined` for a key no caller ever sent. The memory is actor state, so a caller that `authorizeQuery` refuses reads `undefined` for both. `retainedIdempotencyKeys` bounds the memory at 64 keys per actor. A lookup by request id cannot make the distinction, because the runtime generates a request id and no actor remembers one. Schema version 13 adds the column. The previous commit added `messageOutcome` and `findBy` to `ActorRuntime` and implemented neither on the Durable Objects backend, so `quality`, `floor`, and `browser` were all red on one error: Class 'CloudflareRuntime' incorrectly implements interface 'ActorRuntime' Both now work there, including the remembered keys. A Durable Object indexes only its own messages, so `findBy({ requestId })` without a reference raises `UnsupportedCapability`. The parity row that still called result lookup `Planned` now reads `Native`. The doctor and three tests each carried a hand-copied list of schema versions, and all four broke on version 13. They read `SCHEMA_VERSIONS` instead. Validation: - pnpm run check, pnpm run build, pnpm test (515 passed) - pnpm run test:cloudflare (56 passed) - SOLID_OBJECTS_DATABASE_URL=postgresql://... pnpm run test:postgresql - SOLID_OBJECTS_DATABASE_URL=mysql://... pnpm run test:mysql Each new test was watched to fail first. Before the runtime work: `TypeError: this.runtime.findBy is not a function` on Durable Objects, and `no such column: completed_idempotency_keys` on SQL. Without the authorization gate: `MessagePruned: the message for idempotency key "checkout-7f3a" was pruned` where `undefined` was expected. Without the bound: `expected [ 'key-0', 'key-1', 'key-2', …(2) ] to deeply equal [ 'key-2', 'key-3', 'key-4' ]`. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 19 ++++ docs/api.md | 12 ++- docs/cloudflare.md | 1 + docs/configuration.md | 1 + docs/parity.md | 17 +++- src/cloudflare/configuration.ts | 3 + src/cloudflare/engine.ts | 48 +++++++-- src/cloudflare/records.ts | 1 + src/cloudflare/runtime.ts | 56 ++++++++++ src/configuration.ts | 8 ++ src/doctor.ts | 5 +- src/errors.ts | 9 ++ src/index.ts | 1 + src/records.ts | 1 + src/repository.ts | 58 ++++++++++- src/runtime.ts | 44 +++++++- src/schema.ts | 19 +++- test/cloudflare/recovery.test.ts | 45 ++++++++ test/cloudflare/worker.ts | 1 + test/dead-letters.test.ts | 9 +- test/doctor.test.ts | 3 +- test/instance-retention.test.ts | 5 +- test/result-lookup.test.ts | 119 +++++++++++++++++++++- test/support/portable-runtime-contract.ts | 34 +++++++ 24 files changed, 491 insertions(+), 28 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8e00ee7..8ba0dc0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,25 @@ - Add schema version 12: a unique index on `messages.request_id`. The table had only `UNIQUE (actor_type, actor_id, request_id)`, which cannot serve a lookup that names the request id alone. +- Tell a pruned message from one that never existed. An actor remembers the + idempotency keys of its own finished turns, the way an Orleans grain keeps its + deduplication history in grain state, so the memory needs no second store and + no second write. `reference.findBy({ idempotencyKey })` throws + `MessagePruned` for a key the actor remembers and whose message retention + removed, and still returns `undefined` for a key no caller ever sent. The + memory is actor state, so a caller that `authorizeQuery` refuses reads + `undefined` for both. `retainedIdempotencyKeys` bounds the memory and defaults + to 64 keys for each actor. A lookup by request id cannot make the + distinction, because the runtime generates a request id and no actor + remembers one. +- Add schema version 13: `instances.completed_idempotency_keys`. +- Implement `findBy` and `messageOutcome` on the Durable Objects runtime, which + `ActorRuntime` required and the backend did not supply, so `pnpm run check` + and `pnpm run build` both failed. A Durable Object indexes only its own + messages, so `runtime.findBy({ requestId })` without a reference raises + `UnsupportedCapability`; every other form works. +- Read the expected schema migration list from `SCHEMA_VERSIONS` in the doctor + and in the tests that assert it, rather than from four hand-copied lists. - Retry a dead effect or broadcast. `runtime.deadLetters` keeps its message meaning and answers `effects` and `broadcasts`, so the kind rides on the diff --git a/docs/api.md b/docs/api.md index 88decdc..261ca5a 100644 --- a/docs/api.md +++ b/docs/api.md @@ -558,6 +558,14 @@ class and result types are also exported for integration typing. neither key, naming both, or naming an idempotency key without a reference throws a `TypeError`. An absent row, an unregistered actor, and a caller the policy refuses all return `undefined`. +- An actor remembers the idempotency keys of its own last + `retainedIdempotencyKeys` finished turns, so `reference.findBy({ +idempotencyKey })` throws `MessagePruned` for a key the actor remembers and + whose message retention removed, and returns `undefined` for a key no caller + ever sent. The memory is actor state, so a caller that `authorizeQuery` + refuses reads `undefined` for both. `runtime.findBy({ requestId })` returns + `undefined` in both cases, because the runtime generates a request id and no + actor remembers one. - `messageReference.outcome()` returns an `Outcome`: the status, the result, an `ErrorRecord` for a dead message, a `RejectionRecord` for a rejected one, and the attempt count. @@ -711,8 +719,8 @@ provides through `database.wakeUp(options)`. The root exports `SolidObjectsError` and its supported subclasses: - policy and caller outcomes: `Unauthorized`, `Rejected`, `ActorDestroyed`, - `SyncEnqueueTimeout`, `SyncTimeout`, `SyncInsideTransaction`, and - `MessageFailed`; + `SyncEnqueueTimeout`, `SyncTimeout`, `SyncInsideTransaction`, + `MessagePruned`, and `MessageFailed`; - admission and payload failures: `MailboxFull`, `InvalidPayload`, `PayloadTooLarge`, `IdempotencyConflict`, `InvalidPayloadBroadcast`, and `UnknownPayloadBroadcast`; diff --git a/docs/cloudflare.md b/docs/cloudflare.md index 688642d..40aaf85 100644 --- a/docs/cloudflare.md +++ b/docs/cloudflare.md @@ -124,6 +124,7 @@ replay current committed projections, not every event missed while offline. | Actor-scoped dead letters and reminder administration | Supported | | `commitAction`, shared application SQL transactions | Unsupported | | Global repository, reconciliation, process controls, SQL dashboard | Unsupported | +| `findBy({ requestId })` without a reference | Unsupported; name the actor | | Process-local `runtime.realtime.connect()` / server `ref.live` | Unsupported; use browser subscriptions | | SQL-to-Durable-Objects data migration | Not provided | diff --git a/docs/configuration.md b/docs/configuration.md index 78ec0c6..865ce69 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -87,6 +87,7 @@ polling remains the correctness path. | `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. | +| `retainedIdempotencyKeys` | `64` | Positive integer idempotency keys an actor remembers. | | `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. | diff --git a/docs/parity.md b/docs/parity.md index 01cc70b..aa90212 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -70,7 +70,7 @@ is needed for the JavaScript stale-result race fix. | Committed snapshots | Native | `snapshot()` returns authorized persisted fields and inferred getters from one read-only committed state image; realtime replay reads explicit observables without mailbox history. | | Actor destruction and incarnation fencing | Native | Authorized cascading deletion creates a fresh instance ID on recreation; an authorized waiter receives `ActorDestroyed` when that incarnation disappears. | | Result recovery and sync timeout diagnostics | Native | Status, result, and wait reauthorize the stored operation; terminal failure raises structured `MessageFailed`; whole-call adapter deadlines distinguish enqueue, wait, database, activation, and mailbox blockers. | -| Result lookup by request ID | Planned | This is also an open Ruby roadmap item and will be implemented in both runtimes when its authorization shape is settled. | +| Result lookup by request ID and idempotency key | Native | `runtime.findBy({ requestId })` and `reference.findBy({ idempotencyKey })` rebuild a `MessageReference`, authorized with the hook the original call ran. An actor remembers the keys of its own finished turns, so a key lookup separates a pruned message from one that never existed. | Effect callback envelopes are typed with `EffectFailurePayload`, `EffectSuccessPayload`, and `SerializedError` in both SQL and Cloudflare. @@ -160,6 +160,21 @@ needed a new unique index on `request_id`, added as schema version 12, because its table constrained the pair `(actor_type, actor_id, request_id)`; the Ruby schema has carried a global unique index since its first migration. +Both runtimes tell a pruned message from one that never existed the same way. +An actor remembers the idempotency keys of its own last +`retained_idempotency_keys` / `retainedIdempotencyKeys` finished turns, written +in the instance row the executor updates anyway, so a key lookup raises +`MessagePruned` for a message retention removed and answers absent for a +message that never existed. The memory is actor state, so the query hook gates +the pruned answer in both. A request id lookup answers absent in both cases, +because the runtime generates a request id and no actor remembers one. + +The Durable Objects backend answers a key lookup and a request id lookup +through the actor that holds the row, and it remembers keys in the same +instance record. It cannot answer `findBy({ requestId })` without a reference, +because a Durable Object indexes only its own messages, so that form raises +`UnsupportedCapability`. + ## Realtime and browser behavior | Capability | Status | TypeScript shape or remaining work | diff --git a/src/cloudflare/configuration.ts b/src/cloudflare/configuration.ts index 6349a0a..5bd2777 100644 --- a/src/cloudflare/configuration.ts +++ b/src/cloudflare/configuration.ts @@ -26,6 +26,7 @@ export type CloudflareConfiguration = Pick< | "retryDelayMilliseconds" | "messageRetentionMilliseconds" | "pruneBatchSize" + | "retainedIdempotencyKeys" > & { backend: DurableObjectsBackend effects?: Readonly> @@ -53,6 +54,7 @@ export function buildCloudflareSettings(configuration: CloudflareConfiguration) ((attempt: number) => Math.min(2 ** (attempt - 1), 60) * 1_000), messageRetentionMilliseconds: configuration.messageRetentionMilliseconds ?? 30 * 86_400_000, pruneBatchSize: configuration.pruneBatchSize ?? 1_000, + retainedIdempotencyKeys: configuration.retainedIdempotencyKeys ?? 64, } for (const name of [ "maxAttempts", @@ -64,6 +66,7 @@ export function buildCloudflareSettings(configuration: CloudflareConfiguration) "maxActivationDurationMilliseconds", "messageRetentionMilliseconds", "pruneBatchSize", + "retainedIdempotencyKeys", ] as const) { if (!Number.isSafeInteger(settings[name]) || settings[name] <= 0) { throw new TypeError(`${name} must be a positive safe integer`) diff --git a/src/cloudflare/engine.ts b/src/cloudflare/engine.ts index 0fbd723..072e7f6 100644 --- a/src/cloudflare/engine.ts +++ b/src/cloudflare/engine.ts @@ -321,14 +321,8 @@ export class ActorEngine { })) ) throw new Unauthorized("message lookup is not authorized") - const receipt = this.store.storage.sql - .exec<{ message_id: string }>( - "SELECT message_id FROM receipts WHERE request_id = ?", - String(input.payload.requestId), - ) - .toArray()[0] - message = receipt ? this.store.message(receipt.message_id) : undefined - if (!message) return null + message = this.lookUp(input) + if (!message) return this.prunedReply(input) } else { message = this.store.message(String(input.payload.id)) if ( @@ -345,6 +339,40 @@ export class ActorEngine { return normalizeJson(message) } + private lookUp(input: HostRequest): Message | undefined { + if (input.payload.idempotencyKey !== undefined) { + return this.store.rows("SELECT record FROM messages WHERE idempotency_key = ?", [ + String(input.payload.idempotencyKey), + ])[0] + } + const receipt = this.store.storage.sql + .exec<{ message_id: string }>( + "SELECT message_id FROM receipts WHERE request_id = ?", + String(input.payload.requestId), + ) + .toArray()[0] + return receipt ? this.store.message(receipt.message_id) : undefined + } + + private prunedReply(input: HostRequest): JsonValue { + const key = input.payload.idempotencyKey + if (key === undefined) return null + const remembered = this.store.instance()?.completedIdempotencyKeys ?? [] + return remembered.includes(String(key)) ? { pruned: true } : null + } + + private rememberKey(instance: Instance, message: Message): void { + const key = message.idempotencyKey + if (key === null) return + + const remembered = instance.completedIdempotencyKeys ?? [] + if (remembered.at(-1) === key) return + + instance.completedIdempotencyKeys = [...remembered.filter((value) => value !== key), key].slice( + -this.settings.retainedIdempotencyKeys, + ) + } + private committed(identity: ActorIdentity) { const definition = this.definition(identity.actorType) const instance = this.store.instance() @@ -589,6 +617,7 @@ export class ActorEngine { current.state = evaluated.state current.stateVersion = definition.stateVersion current.revision = message.sequence + this.rememberKey(current, message) this.store.saveInstance(current) message.status = "completed" message.result = evaluated.result @@ -622,6 +651,8 @@ export class ActorEngine { details: jsonObject(error.details), } message.completedAt = Date.now() + this.rememberKey(current, message) + this.store.saveInstance(current) this.completeReminder(message) } else { message.error = { @@ -634,6 +665,7 @@ export class ActorEngine { message.availableAt = Date.now() + this.retryDelay(message.attempt) if (exhausted) { current.paused = true + this.rememberKey(current, message) this.store.saveInstance(current) this.pauseReminder(message) } diff --git a/src/cloudflare/records.ts b/src/cloudflare/records.ts index 8c5c944..25db7d1 100644 --- a/src/cloudflare/records.ts +++ b/src/cloudflare/records.ts @@ -11,6 +11,7 @@ export interface Instance extends ActorIdentity { stateVersion: number createdAt: number paused: boolean + completedIdempotencyKeys?: string[] } export interface Message { diff --git a/src/cloudflare/runtime.ts b/src/cloudflare/runtime.ts index c046a03..921f831 100644 --- a/src/cloudflare/runtime.ts +++ b/src/cloudflare/runtime.ts @@ -8,6 +8,7 @@ import { ActorSetupFailed, EnqueueOutcomeUnknown, MessageFailed, + MessagePruned, Rejected, SyncTimeout, UnsupportedCapability, @@ -19,6 +20,7 @@ import { type ActorReference, type ActorSnapshot, } from "../reference.js" +import type { ErrorRecord, Outcome, RejectionRecord } from "../outcome.js" import { jsonObject, normalizeJson, readonlyCopy } from "../serialization.js" import type { ActorIdentifier, @@ -127,6 +129,60 @@ export class CloudflareRuntime implements ActorRuntime { return value === null ? undefined : this.reference(options, jsonObject(value)) } + // A Durable Object indexes only its own messages, so every lookup names the + // actor that holds the row. + async findBy(input: { + reference?: ActorReferenceCore + requestId?: string + idempotencyKey?: string + authorizationContext?: JsonValue + }): Promise { + const named = [input.requestId, input.idempotencyKey].filter( + (value) => value !== undefined, + ).length + if (named !== 1) { + throw new TypeError("findBy expects exactly one of requestId or idempotencyKey") + } + if (input.idempotencyKey !== undefined && !input.reference) { + throw new TypeError("findBy with idempotencyKey requires reference") + } + if (!input.reference) unsupported("a request id lookup without a reference") + + const value = await this.call({ + ...identity(input.reference), + method: "lookup", + authorizationContext: context(input.authorizationContext), + payload: + input.requestId === undefined + ? { idempotencyKey: input.idempotencyKey! } + : { requestId: input.requestId }, + }) + if (value === null) return undefined + + const record = jsonObject(value) + if (record.pruned === true) throw new MessagePruned(input.idempotencyKey!) + return this.reference(input.reference, record) + } + + async messageOutcome( + message: MessageReference, + options: SnapshotOptions = {}, + ): Promise> { + const record = await this.readMessage(message, options) + return Object.freeze({ + status: record.status as MessageStatus, + result: + record.result === null ? undefined : (readonlyCopy(record.result) as DeepReadonly), + error: + record.error === null ? undefined : (jsonObject(record.error) as unknown as ErrorRecord), + rejection: + record.rejection === null + ? undefined + : (jsonObject(record.rejection) as unknown as RejectionRecord), + attempts: Number(record.attempt), + }) + } + async messageStatus( message: MessageReference, options: SnapshotOptions = {}, diff --git a/src/configuration.ts b/src/configuration.ts index d850ee5..8f0435b 100644 --- a/src/configuration.ts +++ b/src/configuration.ts @@ -80,6 +80,7 @@ export interface SolidObjectsConfiguration { instanceRetentionByActorType?: Readonly> processRetentionMilliseconds?: number pruneBatchSize?: number + retainedIdempotencyKeys?: number logger?: Logger authorizeMessage?: (input: AuthorizationInput) => boolean | Promise authorizeQuery?: (input: AuthorizationInput) => boolean | Promise @@ -168,6 +169,7 @@ export function buildSettings(configuration: SolidObjectsConfiguration): Runtime }), processRetentionMilliseconds: configuration.processRetentionMilliseconds ?? 7 * 86_400_000, pruneBatchSize: configuration.pruneBatchSize ?? 1_000, + retainedIdempotencyKeys: configuration.retainedIdempotencyKeys ?? 64, redriveBatchSize: configuration.redriveBatchSize ?? 100, redriveBatchPauseMilliseconds: configuration.redriveBatchPauseMilliseconds ?? 50, administrationIdentity: @@ -315,6 +317,12 @@ function validateSettings(settings: RuntimeSettings): void { if (!Number.isSafeInteger(settings.pruneBatchSize) || settings.pruneBatchSize < 1) { throw new TypeError("pruneBatchSize must be a positive safe integer") } + if ( + !Number.isSafeInteger(settings.retainedIdempotencyKeys) || + settings.retainedIdempotencyKeys < 1 + ) { + throw new TypeError("retainedIdempotencyKeys must be a positive safe integer") + } validateRetentionOverrides("messageRetentionByActorType", settings.messageRetentionByActorType) validateRetentionOverrides("instanceRetentionByActorType", settings.instanceRetentionByActorType) diff --git a/src/doctor.ts b/src/doctor.ts index 27c34b8..26a0891 100644 --- a/src/doctor.ts +++ b/src/doctor.ts @@ -3,6 +3,7 @@ import { Actor } from "./actor.js" import type { DatabaseConnection } from "./database/types.js" import { initialStateFor, validateDefinition } from "./definition.js" import type { SolidObjectsRuntime } from "./runtime.js" +import { SCHEMA_VERSIONS } from "./schema.js" import { jsonObject, readonlyCopy } from "./serialization.js" import type { DeepReadonly, JsonObject } from "./types.js" @@ -243,11 +244,11 @@ export class Doctor { message: `incompatible schema identity ${wrongIdentity.schema_identity}`, }) } - if (versions.join(",") !== "1,2,3,4,5,6,7,8,9,10,11,12") { + if (versions.join(",") !== SCHEMA_VERSIONS.join(",")) { return check({ name: "schema", status: "fail", - message: `expected schema migrations 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12; found ${versions.join(", ")}`, + message: `expected schema migrations ${SCHEMA_VERSIONS.join(", ")}; found ${versions.join(", ")}`, }) } return check({ diff --git a/src/errors.ts b/src/errors.ts index d4ba2a3..09a5260 100644 --- a/src/errors.ts +++ b/src/errors.ts @@ -103,6 +103,15 @@ export class Rejected extends SolidObjectsError { } } +export class MessagePruned extends SolidObjectsError { + readonly idempotencyKey: string + + constructor(idempotencyKey: string) { + super(`the message for idempotency key ${JSON.stringify(idempotencyKey)} was pruned`) + this.idempotencyKey = idempotencyKey + } +} + export class SyncTimeout extends SolidObjectsError { readonly details: SyncTimeoutDetails readonly messageReference: MessageReference diff --git a/src/index.ts b/src/index.ts index c74e60d..c5029d2 100644 --- a/src/index.ts +++ b/src/index.ts @@ -194,6 +194,7 @@ export { LostActivation, MailboxFull, MessageFailed, + MessagePruned, NonRetryableError, PayloadTooLarge, QueryMutatedState, diff --git a/src/records.ts b/src/records.ts index 2ae81d7..e8469a8 100644 --- a/src/records.ts +++ b/src/records.ts @@ -28,6 +28,7 @@ export interface InstanceRow { paused: number | bigint created_at_ms: number | bigint updated_at_ms: number | bigint + completed_idempotency_keys: string | null } export interface MessageRow { diff --git a/src/repository.ts b/src/repository.ts index ac741c0..7de33a0 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -712,11 +712,12 @@ export class Repository { await connection.run( `UPDATE ${this.table("instances")} SET state = ?, state_version = ?, state_revision = ?, - updated_at_ms = ? WHERE id = ?`, + completed_idempotency_keys = ?, updated_at_ms = ? WHERE id = ?`, [ JSON.stringify(input.state), input.stateVersion, turn.message.sequence, + await this.rememberedKeys(connection, turn), now, turn.instance.id, ], @@ -940,6 +941,7 @@ export class Repository { `UPDATE ${this.table("messages")} SET rejection = ?, completed_at_ms = ?, updated_at_ms = ? WHERE id = ?`, [JSON.stringify(rejection), now, now, turn.message.id], ) + await this.rememberKeys(connection, turn) await this.releaseClaim({ connection, turn }) }) } @@ -971,6 +973,7 @@ export class Repository { now, ], ) + await this.rememberKeys(connection, turn) await this.releaseClaim({ connection, turn }) return "dead" as const } @@ -1007,6 +1010,21 @@ export class Repository { ) } + async remembersIdempotencyKey(input: { + actorType: string + actorId: string + idempotencyKey: string + }): Promise { + const row = await this.settings.database.connection((connection) => + connection.get>( + `SELECT completed_idempotency_keys FROM ${this.table("instances")} + WHERE actor_type = ? AND actor_id = ?`, + [input.actorType, input.actorId], + ), + ) + return rememberedList(row?.completed_idempotency_keys ?? null).includes(input.idempotencyKey) + } + async findMessageByIdempotencyKey(input: { actorType: string actorId: string @@ -2240,6 +2258,37 @@ export class Repository { } } + private async rememberKeys(connection: DatabaseConnection, turn: ClaimedTurn): Promise { + if (turn.message.idempotency_key === null) return + + await connection.run( + `UPDATE ${this.table("instances")} SET completed_idempotency_keys = ? WHERE id = ?`, + [await this.rememberedKeys(connection, turn), turn.instance.id], + ) + } + + private async rememberedKeys( + connection: DatabaseConnection, + turn: ClaimedTurn, + ): Promise { + const row = await connection.get>( + `SELECT completed_idempotency_keys FROM ${this.table("instances")} WHERE id = ?`, + [turn.instance.id], + ) + const stored = row?.completed_idempotency_keys ?? null + const key = turn.message.idempotency_key + if (key === null) return stored + + const remembered = rememberedList(stored) + if (remembered.at(-1) === key) return stored + + return JSON.stringify( + [...remembered.filter((value) => value !== key), key].slice( + -this.settings.retainedIdempotencyKeys, + ), + ) + } + private async assertFence(connection: DatabaseConnection, turn: ClaimedTurn): Promise { const instance = await this.lockActivationFence(connection, turn.instance.id) const now = await connection.nowMilliseconds() @@ -2373,6 +2422,13 @@ function retentionPolicy(options: { } } +function rememberedList(stored: string | null): string[] { + if (stored === null) return [] + const parsed: unknown = JSON.parse(stored) + if (!Array.isArray(parsed)) return [] + return parsed.filter((value): value is string => typeof value === "string") +} + function parameterList(length: number): string { return Array.from({ length }, () => "?").join(", ") } diff --git a/src/runtime.ts b/src/runtime.ts index 7024c66..0965049 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -38,6 +38,7 @@ import { InvalidActor, LostActivation, MessageFailed, + MessagePruned, NonRetryableError, QueryMutatedState, Rejected, @@ -707,7 +708,7 @@ export class SolidObjectsRuntime { } const message = await this.lookedUpMessage(input) - if (!message) return undefined + if (!message) return await this.absentMessage(input) if (!(await this.readableMessage(message, input.authorizationContext))) return undefined return this.messageReferenceFromRow(message) @@ -1720,6 +1721,47 @@ export class SolidObjectsRuntime { }) } + // An actor remembers the idempotency keys of its own finished turns, so a + // key it remembers and no longer has a row for was pruned. A request id is + // generated by the runtime, so no actor remembers one and both cases read + // the same. A caller the policy refuses learns nothing either way. + private async absentMessage(input: { + reference?: ActorReferenceCore + idempotencyKey?: string + authorizationContext?: AdministrationOptions["authorizationContext"] + }): Promise { + if (input.idempotencyKey === undefined) return undefined + + const reference = input.reference! + const remembered = await this.repository.remembersIdempotencyKey({ + actorType: reference.actorType, + actorId: reference.actorId, + idempotencyKey: input.idempotencyKey, + }) + if (!remembered) return undefined + if (!(await this.readableState(reference, input.authorizationContext))) return undefined + + throw new MessagePruned(input.idempotencyKey) + } + + private async readableState( + reference: ActorReferenceCore, + authorizationContext: AdministrationOptions["authorizationContext"], + ): Promise { + try { + await this.authorize({ + kind: "query", + reference, + operation: "__snapshot__", + argumentsValue: {}, + authorizationContext, + }) + return true + } catch { + return false + } + } + // A lookup answers a question, so an absent row, an actor this process no // longer registers, and a caller the policy refuses all read the same. A // request id that threw where it was refused would be a way to ask whether diff --git a/src/schema.ts b/src/schema.ts index 4374db3..1ccafae 100644 --- a/src/schema.ts +++ b/src/schema.ts @@ -13,7 +13,12 @@ const EFFECT_RECOVERY_VERSION = 9 const INSTANCE_RETENTION_INDEX_VERSION = 10 const DEAD_LETTER_REDRIVE_VERSION = 11 const REQUEST_ID_LOOKUP_VERSION = 12 -const LATEST_VERSION = REQUEST_ID_LOOKUP_VERSION +const REMEMBERED_KEYS_VERSION = 13 +const LATEST_VERSION = REMEMBERED_KEYS_VERSION + +export const SCHEMA_VERSIONS: readonly number[] = Object.freeze( + Array.from({ length: LATEST_VERSION }, (_unused, index) => index + 1), +) export async function installSchema(options: { connection: DatabaseConnection @@ -373,6 +378,18 @@ export async function installSchema(options: { }) } + if (!installedVersions.has(REMEMBERED_KEYS_VERSION)) { + await connection.run( + `ALTER TABLE ${table("instances")} ADD COLUMN ${family === "postgresql" ? "IF NOT EXISTS " : ""}completed_idempotency_keys ${family === "mysql" ? "LONGTEXT" : "TEXT"}`, + ) + await recordMigration({ + connection, + table: table("schema_migrations"), + version: REMEMBERED_KEYS_VERSION, + schemaIdentity, + }) + } + if (installedVersions.has(POLLING_INDEXES_VERSION)) return const pollingIndexes = [ ["effects", `${prefix}effects_poll`, "status, available_at_ms, id"], diff --git a/test/cloudflare/recovery.test.ts b/test/cloudflare/recovery.test.ts index e1009e6..b93fe0f 100644 --- a/test/cloudflare/recovery.test.ts +++ b/test/cloudflare/recovery.test.ts @@ -326,6 +326,51 @@ describe("Cloudflare recovery and fencing", () => { expect(alarm).toBeNull() }) + it("tells a pruned message from one that never existed", async () => { + const reference = runtime().ref(Counter, "pruned-key") + const message = await reference.send + .with({ authorizationContext, idempotencyKey: "pruned-7f3a" }) + .increment() + await message.wait({ authorizationContext }) + await runInDurableObject(stub("pruned-key"), (_object, state) => { + state.storage.sql.exec("DELETE FROM messages") + }) + + await expect( + reference.findBy({ idempotencyKey: "pruned-7f3a", authorizationContext }), + ).rejects.toMatchObject({ name: "MessagePruned", idempotencyKey: "pruned-7f3a" }) + expect( + await reference.findBy({ idempotencyKey: "never-sent", authorizationContext }), + ).toBeUndefined() + }) + + it("refuses a request id lookup that names no actor", async () => { + await expect( + runtime().findBy({ requestId: crypto.randomUUID(), authorizationContext }), + ).rejects.toMatchObject({ name: "UnsupportedCapability" }) + }) + + it("bounds what an instance remembers", async () => { + const reference = runtime().ref(Counter, "bounded-keys") + for (let index = 0; index < 5; index += 1) { + const message = await reference.send + .with({ authorizationContext, idempotencyKey: `key-${index}` }) + .increment() + await message.wait({ authorizationContext }) + } + + const remembered = await runInDurableObject(stub("bounded-keys"), (_object, state) => { + const instance = JSON.parse( + state.storage.sql + .exec<{ value: string }>("SELECT value FROM metadata WHERE key = 'instance'") + .one().value, + ) as Instance + return instance.completedIdempotencyKeys + }) + + expect(remembered).toEqual(["key-2", "key-3", "key-4"]) + }) + it("continues bounded receipt cleanup using its saved alarm", async () => { const reference = runtime().ref(Counter, "receipt-cleanup") const message = await reference.send.with({ authorizationContext }).increment() diff --git a/test/cloudflare/worker.ts b/test/cloudflare/worker.ts index 7afff63..efe0847 100644 --- a/test/cloudflare/worker.ts +++ b/test/cloudflare/worker.ts @@ -242,6 +242,7 @@ export class Actors extends createDurableObjectsHost({ authorizeSubscription: (input) => input.authorizationContext === "allowed", authorizeAdministration: (input) => input.authorizationContext === "allowed", retryDelayMilliseconds: () => 10, + retainedIdempotencyKeys: 3, effects: { callbackEmpty: (_arguments, context) => { deliveries.set(context.id, context.attempt) diff --git a/test/dead-letters.test.ts b/test/dead-letters.test.ts index a8fbe55..a71ae47 100644 --- a/test/dead-letters.test.ts +++ b/test/dead-letters.test.ts @@ -8,6 +8,7 @@ import type { SolidObjectsConfiguration } from "../src/configuration.js" import { sqlite } from "../src/database/sqlite.js" import { Unauthorized } from "../src/errors.js" import { configure, type SolidObjectsRuntime } from "../src/runtime.js" +import { SCHEMA_VERSIONS } from "../src/schema.js" class PoisonActor extends Actor { static override readonly actorType = "PoisonActor" @@ -151,9 +152,7 @@ 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, 11, 12, - ]) + expect(versions.map(({ version }) => Number(version))).toEqual([...SCHEMA_VERSIONS]) expect(deadLetterColumns.map(({ name }) => name)).toContain("retried_message_id") expect(broadcastColumns.map(({ name }) => name)).toContain("invalidations") expect(await installedPollingIndexes(runtime)).toEqual(POLLING_INDEX_COLUMNS) @@ -186,9 +185,7 @@ 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, 11, 12, - ]) + expect(versions.map(({ version }) => Number(version))).toEqual([...SCHEMA_VERSIONS]) expect(await installedPollingIndexes(runtime)).toEqual(POLLING_INDEX_COLUMNS) }) diff --git a/test/doctor.test.ts b/test/doctor.test.ts index 4453b4e..bb5707c 100644 --- a/test/doctor.test.ts +++ b/test/doctor.test.ts @@ -2,6 +2,7 @@ import { afterEach, describe, expect, it } from "vitest" import { Actor } from "../src/actor.js" import { sqlite } from "../src/database/sqlite.js" import { configure, type SolidObjectsRuntime } from "../src/runtime.js" +import { SCHEMA_VERSIONS } from "../src/schema.js" let runtime: SolidObjectsRuntime | undefined @@ -27,7 +28,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, 11, 12] }, + details: { versions: [...SCHEMA_VERSIONS] }, }) 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 92d3af6..8429f2f 100644 --- a/test/instance-retention.test.ts +++ b/test/instance-retention.test.ts @@ -8,6 +8,7 @@ import type { DatabaseTransactionOptions, } from "../src/database/types.js" import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" +import { SCHEMA_VERSIONS } from "../src/schema.js" const PREFIX = "retention_index_test_" const DAY = 24 * 60 * 60 * 1_000 @@ -100,9 +101,7 @@ 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, 11, 12, - ]) + expect(versions.map(({ version }) => Number(version))).toEqual([...SCHEMA_VERSIONS]) }, 30_000, ) diff --git a/test/result-lookup.test.ts b/test/result-lookup.test.ts index 85324f6..ede0ba9 100644 --- a/test/result-lookup.test.ts +++ b/test/result-lookup.test.ts @@ -1,7 +1,7 @@ import { afterEach, describe, expect, it } from "vitest" import { Actor } from "../src/actor.js" import { sqlite } from "../src/database/sqlite.js" -import { Rejected } from "../src/errors.js" +import { MessagePruned, Rejected } from "../src/errors.js" import { configure, type SolidObjectsRuntime } from "../src/runtime.js" class CartActor extends Actor { @@ -37,10 +37,13 @@ afterEach(async () => { hooks = [] }) -async function start(): Promise { +async function start( + overrides: { retainedIdempotencyKeys?: number } = {}, +): Promise { const created = configure({ database: sqlite({ path: ":memory:" }), maxAttempts: 1, + ...overrides, retryDelayMilliseconds: () => 0, authorizeMessage: ({ operation, arguments: argumentsValue }) => { hooks.push("message") @@ -262,6 +265,97 @@ describe("result lookup", () => { expect((await reference.snapshot()).items).toBe(1) }) + it("tells a pruned message from one that never existed", async () => { + const active = await start() + const reference = active.ref(CartActor, "alice") + await reference.send.with({ idempotencyKey: "checkout-7f3a" }).checkout({ orderId: 1 }) + await active.worker().runUntilIdle() + await deleteMessages(active) + + await expect(reference.findBy({ idempotencyKey: "checkout-7f3a" })).rejects.toThrow( + MessagePruned, + ) + expect(await reference.findBy({ idempotencyKey: "never-used" })).toBeUndefined() + }) + + it("names the key it remembers", async () => { + const active = await start() + const reference = active.ref(CartActor, "alice") + await reference.send.with({ idempotencyKey: "checkout-7f3a" }).checkout({ orderId: 1 }) + await active.worker().runUntilIdle() + await deleteMessages(active) + + const error = await reference + .findBy({ idempotencyKey: "checkout-7f3a" }) + .catch((thrown) => thrown) + + expect(error).toBeInstanceOf(MessagePruned) + expect((error as MessagePruned).idempotencyKey).toBe("checkout-7f3a") + }) + + it("does not tell a refused caller that a key was pruned", async () => { + const active = await start() + const reference = active.ref(CartActor, "alice") + await reference.send.with({ idempotencyKey: "checkout-7f3a" }).checkout({ orderId: 1 }) + await active.worker().runUntilIdle() + await deleteMessages(active) + const refusing = configure({ + database: active.settings.database, + authorizeQuery: () => false, + }) + refusing.register(CartActor) + + expect( + await refusing.ref(CartActor, "alice").findBy({ idempotencyKey: "checkout-7f3a" }), + ).toBeUndefined() + }) + + it("remembers a key whose message was rejected", async () => { + const active = await start() + const reference = active.ref(CartActor, "alice") + await reference.send.with({ idempotencyKey: "rejected-7f3a" }).rejectCheckout() + await active.worker().runUntilIdle() + await deleteMessages(active) + + await expect(reference.findBy({ idempotencyKey: "rejected-7f3a" })).rejects.toThrow( + MessagePruned, + ) + }) + + it("remembers a key whose message died", async () => { + const active = await start() + CartActor.fail = true + const reference = active.ref(CartActor, "alice") + await reference.send.with({ idempotencyKey: "dead-7f3a" }).checkout({ orderId: 1 }) + await active.worker().runUntilIdle() + await deleteDeadLetters(active) + await deleteMessages(active) + + await expect(reference.findBy({ idempotencyKey: "dead-7f3a" })).rejects.toThrow(MessagePruned) + }) + + it("bounds what an instance remembers", async () => { + const active = await start({ retainedIdempotencyKeys: 3 }) + const reference = active.ref(CartActor, "alice") + for (let index = 0; index < 5; index += 1) { + await reference.send.with({ idempotencyKey: `key-${index}` }).checkout({ orderId: index }) + } + await active.worker().runUntilIdle() + await deleteMessages(active) + + expect(await reference.findBy({ idempotencyKey: "key-0" })).toBeUndefined() + await expect(reference.findBy({ idempotencyKey: "key-4" })).rejects.toThrow(MessagePruned) + expect(await rememberedKeys(active)).toEqual(["key-2", "key-3", "key-4"]) + }) + + it("remembers nothing for a message that carried no key", async () => { + const active = await start() + await active.ref(CartActor, "alice").send.checkout({ orderId: 1 }) + await active.worker().runUntilIdle() + + expect(await rememberedKeys(active)).toEqual([]) + }) + it("keeps request ids unique across the table", async () => { const active = await start() const original = await active.ref(CartActor, "alice").send.checkout({ orderId: 1 }) @@ -286,3 +380,24 @@ async function lastRequestId(active: SolidObjectsRuntime): Promise { + await active.settings.database.transaction((connection) => + connection.run(`DELETE FROM ${active.repository.table("messages")}`), + ) +} + +async function deleteDeadLetters(active: SolidObjectsRuntime): Promise { + await active.settings.database.transaction((connection) => + connection.run(`DELETE FROM ${active.repository.table("dead_letters")}`), + ) +} + +async function rememberedKeys(active: SolidObjectsRuntime): Promise { + const row = await active.settings.database.connection((connection) => + connection.get<{ completed_idempotency_keys: string | null }>( + `SELECT completed_idempotency_keys FROM ${active.repository.table("instances")}`, + ), + ) + return JSON.parse(row?.completed_idempotency_keys ?? "[]") as string[] +} diff --git a/test/support/portable-runtime-contract.ts b/test/support/portable-runtime-contract.ts index 7763b4c..f347410 100644 --- a/test/support/portable-runtime-contract.ts +++ b/test/support/portable-runtime-contract.ts @@ -49,6 +49,40 @@ export function portableRuntimeContract(runtime: () => ActorRuntime): void { expect(await reference.increment()).toBe(1) }) + it("finds a message by idempotency key and reports its outcome", async () => { + const active = runtime() + const reference = active.ref(PortableCounter, "contract-find") + const options = { authorizationContext, idempotencyKey: "contract-find-key" } + const original = await reference.send.with(options).increment() + await original.wait({ authorizationContext }) + + const found = await reference.findBy({ + idempotencyKey: "contract-find-key", + authorizationContext, + }) + + expect(found?.id).toBe(original.id) + expect(await found!.outcome({ authorizationContext })).toMatchObject({ + status: "completed", + attempts: 1, + }) + expect( + await reference.findBy({ idempotencyKey: "never-sent", authorizationContext }), + ).toBeUndefined() + }) + + it("refuses a lookup that names no key, both keys, or a key without a reference", async () => { + const active = runtime() + + await expect(active.findBy({ authorizationContext })).rejects.toThrow(/exactly one of/) + await expect( + active.findBy({ requestId: "one", idempotencyKey: "two", authorizationContext }), + ).rejects.toThrow(/exactly one of/) + await expect(active.findBy({ idempotencyKey: "two", authorizationContext })).rejects.toThrow( + /requires reference/, + ) + }) + it("authorizes calls and fences references across destruction", async () => { const reference = runtime().ref(PortableCounter, "contract-destroy") await expect(reference.increment()).rejects.toMatchObject({ name: "Unauthorized" }) From 1245f01d1ff30b15dbe36bc725e30942fe9c2ee5 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Wed, 23 Sep 2026 09:13:39 -0700 Subject: [PATCH 03/14] fix: answer absent when a lookup is refused A parity audit against the Ruby branch found three gaps. The Durable Objects `lookup` gates on `authorizeQuery` before it reads, so `findBy` threw `Unauthorized` there while the SQL runtime answers `undefined`. A lookup that throws where it is refused and answers absent where the row is gone is a way to ask whether a key exists, which is the guarantee this feature documents. `findBy` now answers absent on both backends, and the portable runtime contract holds both to it. Without the catch the contract fails with `Unauthorized: message lookup is not authorized`. The doctor passed on an instance table without `completed_idempotency_keys`, so a half-applied version 13 read as healthy. Without the column in `EXPECTED_COLUMNS` the new test reads back `schema matches this runtime`. The changelog did not say the doctor verifies the new column. A test proves two keyed messages in one activation pass are both remembered, because the repository reads the locked instance row rather than the claim-time copy on the turn. Validation: pnpm test (518), pnpm run test:cloudflare (57), check, build, format:check, PostgreSQL 18, MySQL 8.4. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 9 +++++++-- src/cloudflare/runtime.ts | 4 ++++ src/doctor.ts | 1 + test/doctor.test.ts | 18 ++++++++++++++++++ test/result-lookup.test.ts | 12 ++++++++++++ test/support/portable-runtime-contract.ts | 16 ++++++++++++++++ 6 files changed, 58 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8ba0dc0..2d32dca 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,12 +29,17 @@ to 64 keys for each actor. A lookup by request id cannot make the distinction, because the runtime generates a request id and no actor remembers one. -- Add schema version 13: `instances.completed_idempotency_keys`. +- Add schema version 13: `instances.completed_idempotency_keys`. The doctor + now reports the column as missing when it is not installed. - Implement `findBy` and `messageOutcome` on the Durable Objects runtime, which `ActorRuntime` required and the backend did not supply, so `pnpm run check` and `pnpm run build` both failed. A Durable Object indexes only its own messages, so `runtime.findBy({ requestId })` without a reference raises - `UnsupportedCapability`; every other form works. + `UnsupportedCapability`; every other form works. A lookup that the policy + refuses answers `undefined` there too. The Durable Objects `lookup` gates on + `authorizeQuery` before it reads, which would otherwise have thrown + `Unauthorized` where the SQL runtime answers absent, and a lookup that threw + where it was refused is a way to ask whether a key exists. - Read the expected schema migration list from `SCHEMA_VERSIONS` in the doctor and in the tests that assert it, rather than from four hand-copied lists. diff --git a/src/cloudflare/runtime.ts b/src/cloudflare/runtime.ts index 921f831..fe7f7d3 100644 --- a/src/cloudflare/runtime.ts +++ b/src/cloudflare/runtime.ts @@ -11,6 +11,7 @@ import { MessagePruned, Rejected, SyncTimeout, + Unauthorized, UnsupportedCapability, } from "../errors.js" import { @@ -156,6 +157,9 @@ export class CloudflareRuntime implements ActorRuntime { input.requestId === undefined ? { idempotencyKey: input.idempotencyKey! } : { requestId: input.requestId }, + }).catch((error: unknown) => { + if (error instanceof Unauthorized) return null + throw error }) if (value === null) return undefined diff --git a/src/doctor.ts b/src/doctor.ts index 26a0891..1ed7927 100644 --- a/src/doctor.ts +++ b/src/doctor.ts @@ -51,6 +51,7 @@ const EXPECTED_COLUMNS: Readonly> = { "activation_expires_at_ms", "activation_generation", "paused", + "completed_idempotency_keys", ], messages: [ "id", diff --git a/test/doctor.test.ts b/test/doctor.test.ts index bb5707c..2cfa9aa 100644 --- a/test/doctor.test.ts +++ b/test/doctor.test.ts @@ -146,6 +146,24 @@ describe("runtime doctor", () => { expect(check(report, "roundTrip").status).toBe("skip") }) + it("fails when a migration that a runtime path needs is missing", async () => { + runtime = configuredRuntime() + await runtime.install() + await runtime.settings.database.transaction((connection) => + connection.run( + `ALTER TABLE ${runtime!.repository.table("instances")} DROP COLUMN completed_idempotency_keys`, + ), + ) + + const report = await runtime.doctor.run() + + expect(report.healthy).toBe(false) + expect(check(report, "schema")).toMatchObject({ + status: "fail", + message: expect.stringContaining("completed_idempotency_keys"), + }) + }) + it("reports live runtime roles by kind", async () => { runtime = configuredRuntime() await runtime.install() diff --git a/test/result-lookup.test.ts b/test/result-lookup.test.ts index ede0ba9..ee6c803 100644 --- a/test/result-lookup.test.ts +++ b/test/result-lookup.test.ts @@ -348,6 +348,18 @@ describe("result lookup", () => { expect(await rememberedKeys(active)).toEqual(["key-2", "key-3", "key-4"]) }) + it("remembers every key of one activation pass", async () => { + const active = await start() + const reference = active.ref(CartActor, "alice") + await reference.send.with({ idempotencyKey: "first" }).checkout({ orderId: 1 }) + await reference.send.with({ idempotencyKey: "second" }).checkout({ orderId: 2 }) + await active.worker().runUntilIdle() + await deleteMessages(active) + + await expect(reference.findBy({ idempotencyKey: "first" })).rejects.toThrow(MessagePruned) + await expect(reference.findBy({ idempotencyKey: "second" })).rejects.toThrow(MessagePruned) + }) + it("remembers nothing for a message that carried no key", async () => { const active = await start() await active.ref(CartActor, "alice").send.checkout({ orderId: 1 }) diff --git a/test/support/portable-runtime-contract.ts b/test/support/portable-runtime-contract.ts index f347410..01da20b 100644 --- a/test/support/portable-runtime-contract.ts +++ b/test/support/portable-runtime-contract.ts @@ -83,6 +83,22 @@ export function portableRuntimeContract(runtime: () => ActorRuntime): void { ) }) + it("answers absent to a caller the policy refuses", async () => { + const active = runtime() + const reference = active.ref(PortableCounter, "contract-refused") + const message = await reference.send + .with({ authorizationContext, idempotencyKey: "contract-refused-key" }) + .increment() + await message.wait({ authorizationContext }) + + expect( + await reference.findBy({ + idempotencyKey: "contract-refused-key", + authorizationContext: "refused", + }), + ).toBeUndefined() + }) + it("authorizes calls and fences references across destruction", async () => { const reference = runtime().ref(PortableCounter, "contract-destroy") await expect(reference.increment()).rejects.toMatchObject({ name: "Unauthorized" }) From 92b73488b83a5813cce4e4303099e189394d2d1c Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Wed, 23 Sep 2026 09:46:14 -0700 Subject: [PATCH 04/14] test: remember a re-sent key once A parity audit found the same untested branch in both languages. A key can finish twice on one actor: retention removes the message, the enqueue path no longer deduplicates it, and the caller sends it again. The list must move the key to the end rather than repeat it, and nothing proved that. Without the move the new test reads `[ 'first', 'second', 'first' ]`. Validation: pnpm test (519 passed), check, format:check. Co-Authored-By: Claude Opus 5 (1M context) --- test/result-lookup.test.ts | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/test/result-lookup.test.ts b/test/result-lookup.test.ts index ee6c803..77d2661 100644 --- a/test/result-lookup.test.ts +++ b/test/result-lookup.test.ts @@ -360,6 +360,19 @@ describe("result lookup", () => { await expect(reference.findBy({ idempotencyKey: "second" })).rejects.toThrow(MessagePruned) }) + it("remembers a re-sent key once", async () => { + const active = await start() + const reference = active.ref(CartActor, "alice") + await reference.send.with({ idempotencyKey: "first" }).checkout({ orderId: 1 }) + await reference.send.with({ idempotencyKey: "second" }).checkout({ orderId: 2 }) + await active.worker().runUntilIdle() + await deleteMessages(active) + await reference.send.with({ idempotencyKey: "first" }).checkout({ orderId: 3 }) + await active.worker().runUntilIdle() + + expect(await rememberedKeys(active)).toEqual(["second", "first"]) + }) + it("remembers nothing for a message that carried no key", async () => { const active = await start() await active.ref(CartActor, "alice").send.checkout({ orderId: 1 }) From dc0c177ab6e49ae86b1bfe8e3f7e2deac6aed07b Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Wed, 23 Sep 2026 10:06:46 -0700 Subject: [PATCH 05/14] fix: bound the remembered keys by size An idempotency key has no length limit: the column is TEXT on SQLite and PostgreSQL. A probe confirmed a 1000-character key is remembered whole. Before this branch such a key sat in one message row that retention removed. The remembered list holds 64 of them on the instance row, which survives as long as the actor, so the growth became unbounded and persistent. On Durable Objects it also reached `encodedRecord`, which throws `PayloadTooLarge` above the SQLite row limit from inside the commit transaction. `retainedIdempotencyKeysBytes` bounds the serialized list at 16 KB on both backends. An actor drops its oldest keys until the list fits, so a key long enough to fill the limit by itself is never remembered and its lookup answers `undefined` rather than throwing. Validation: pnpm test (521), pnpm run test:cloudflare (58), check, build, format:check. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 5 ++++- docs/api.md | 5 +++++ docs/configuration.md | 1 + docs/parity.md | 4 +++- src/cloudflare/configuration.ts | 3 +++ src/cloudflare/engine.ts | 9 ++++++--- src/configuration.ts | 8 ++++++++ src/repository.ts | 16 ++++++++++++---- test/cloudflare/recovery.test.ts | 20 ++++++++++++++++++++ test/cloudflare/worker.ts | 1 + test/result-lookup.test.ts | 29 ++++++++++++++++++++++++++++- 11 files changed, 91 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2d32dca..4657d74 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,7 +28,10 @@ `undefined` for both. `retainedIdempotencyKeys` bounds the memory and defaults to 64 keys for each actor. A lookup by request id cannot make the distinction, because the runtime generates a request id and no actor - remembers one. + remembers one. `retainedIdempotencyKeysBytes` bounds the serialized memory as + well, because an idempotency key has no length limit and the memory outlives + the message row. An actor drops its oldest keys until the list fits, so a key + long enough to fill the limit by itself is never remembered. - Add schema version 13: `instances.completed_idempotency_keys`. The doctor now reports the column as missing when it is not installed. - Implement `findBy` and `messageOutcome` on the Durable Objects runtime, which diff --git a/docs/api.md b/docs/api.md index 261ca5a..807545e 100644 --- a/docs/api.md +++ b/docs/api.md @@ -566,6 +566,11 @@ idempotencyKey })` throws `MessagePruned` for a key the actor remembers and refuses reads `undefined` for both. `runtime.findBy({ requestId })` returns `undefined` in both cases, because the runtime generates a request id and no actor remembers one. +- `retainedIdempotencyKeysBytes` bounds the serialized memory as well, because + an idempotency key has no length limit and the memory outlives the message + row. An actor drops its oldest keys until the list fits, so a key long enough + to fill the limit by itself is never remembered and its lookup answers + `undefined` rather than throwing. - `messageReference.outcome()` returns an `Outcome`: the status, the result, an `ErrorRecord` for a dead message, a `RejectionRecord` for a rejected one, and the attempt count. diff --git a/docs/configuration.md b/docs/configuration.md index 865ce69..d74b0af 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -88,6 +88,7 @@ polling remains the correctness path. | `processRetentionMilliseconds` | 7 days | Positive stopped-process retention. | | `pruneBatchSize` | `1_000` | Positive integer maximum rows rechecked per pruning transaction. | | `retainedIdempotencyKeys` | `64` | Positive integer idempotency keys an actor remembers. | +| `retainedIdempotencyKeysBytes` | `16_384` | Positive integer serialized size those keys must fit. | | `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. | diff --git a/docs/parity.md b/docs/parity.md index aa90212..ebc1644 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -165,7 +165,9 @@ An actor remembers the idempotency keys of its own last `retained_idempotency_keys` / `retainedIdempotencyKeys` finished turns, written in the instance row the executor updates anyway, so a key lookup raises `MessagePruned` for a message retention removed and answers absent for a -message that never existed. The memory is actor state, so the query hook gates +message that never existed. Both also bound the serialized memory, because an +idempotency key has no length limit and the memory outlives the message row; an +actor drops its oldest keys until the list fits. The memory is actor state, so the query hook gates the pruned answer in both. A request id lookup answers absent in both cases, because the runtime generates a request id and no actor remembers one. diff --git a/src/cloudflare/configuration.ts b/src/cloudflare/configuration.ts index 5bd2777..65268aa 100644 --- a/src/cloudflare/configuration.ts +++ b/src/cloudflare/configuration.ts @@ -27,6 +27,7 @@ export type CloudflareConfiguration = Pick< | "messageRetentionMilliseconds" | "pruneBatchSize" | "retainedIdempotencyKeys" + | "retainedIdempotencyKeysBytes" > & { backend: DurableObjectsBackend effects?: Readonly> @@ -55,6 +56,7 @@ export function buildCloudflareSettings(configuration: CloudflareConfiguration) messageRetentionMilliseconds: configuration.messageRetentionMilliseconds ?? 30 * 86_400_000, pruneBatchSize: configuration.pruneBatchSize ?? 1_000, retainedIdempotencyKeys: configuration.retainedIdempotencyKeys ?? 64, + retainedIdempotencyKeysBytes: configuration.retainedIdempotencyKeysBytes ?? 16_384, } for (const name of [ "maxAttempts", @@ -67,6 +69,7 @@ export function buildCloudflareSettings(configuration: CloudflareConfiguration) "messageRetentionMilliseconds", "pruneBatchSize", "retainedIdempotencyKeys", + "retainedIdempotencyKeysBytes", ] as const) { if (!Number.isSafeInteger(settings[name]) || settings[name] <= 0) { throw new TypeError(`${name} must be a positive safe integer`) diff --git a/src/cloudflare/engine.ts b/src/cloudflare/engine.ts index 072e7f6..73eebfa 100644 --- a/src/cloudflare/engine.ts +++ b/src/cloudflare/engine.ts @@ -37,6 +37,7 @@ import type { ScheduledReminder, } from "../types.js" import type { CloudflareSettings } from "./configuration.js" +import { boundedKeys } from "../repository.js" import { actorName, callHost, type ActorIdentity, type HostRequest } from "./protocol.js" import type { Instance, Message, Outbox, Reminder, Subscription } from "./records.js" import { beforeDeadline, CloudflareRuntime } from "./runtime.js" @@ -368,9 +369,11 @@ export class ActorEngine { const remembered = instance.completedIdempotencyKeys ?? [] if (remembered.at(-1) === key) return - instance.completedIdempotencyKeys = [...remembered.filter((value) => value !== key), key].slice( - -this.settings.retainedIdempotencyKeys, - ) + instance.completedIdempotencyKeys = boundedKeys({ + keys: [...remembered.filter((value) => value !== key), key], + count: this.settings.retainedIdempotencyKeys, + bytes: this.settings.retainedIdempotencyKeysBytes, + }) } private committed(identity: ActorIdentity) { diff --git a/src/configuration.ts b/src/configuration.ts index 8f0435b..70b860b 100644 --- a/src/configuration.ts +++ b/src/configuration.ts @@ -81,6 +81,7 @@ export interface SolidObjectsConfiguration { processRetentionMilliseconds?: number pruneBatchSize?: number retainedIdempotencyKeys?: number + retainedIdempotencyKeysBytes?: number logger?: Logger authorizeMessage?: (input: AuthorizationInput) => boolean | Promise authorizeQuery?: (input: AuthorizationInput) => boolean | Promise @@ -170,6 +171,7 @@ export function buildSettings(configuration: SolidObjectsConfiguration): Runtime processRetentionMilliseconds: configuration.processRetentionMilliseconds ?? 7 * 86_400_000, pruneBatchSize: configuration.pruneBatchSize ?? 1_000, retainedIdempotencyKeys: configuration.retainedIdempotencyKeys ?? 64, + retainedIdempotencyKeysBytes: configuration.retainedIdempotencyKeysBytes ?? 16_384, redriveBatchSize: configuration.redriveBatchSize ?? 100, redriveBatchPauseMilliseconds: configuration.redriveBatchPauseMilliseconds ?? 50, administrationIdentity: @@ -323,6 +325,12 @@ function validateSettings(settings: RuntimeSettings): void { ) { throw new TypeError("retainedIdempotencyKeys must be a positive safe integer") } + if ( + !Number.isSafeInteger(settings.retainedIdempotencyKeysBytes) || + settings.retainedIdempotencyKeysBytes < 1 + ) { + throw new TypeError("retainedIdempotencyKeysBytes must be a positive safe integer") + } validateRetentionOverrides("messageRetentionByActorType", settings.messageRetentionByActorType) validateRetentionOverrides("instanceRetentionByActorType", settings.instanceRetentionByActorType) diff --git a/src/repository.ts b/src/repository.ts index 7de33a0..7380c27 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -24,7 +24,7 @@ import type { ProcessRow, ReminderRow, } from "./records.js" -import { jsonObject, normalizeJson } from "./serialization.js" +import { jsonObject, normalizeJson, utf8ByteLength } from "./serialization.js" import type { RetentionTarget } from "./retention.js" import type { EffectFailurePayload, @@ -2283,9 +2283,11 @@ export class Repository { if (remembered.at(-1) === key) return stored return JSON.stringify( - [...remembered.filter((value) => value !== key), key].slice( - -this.settings.retainedIdempotencyKeys, - ), + boundedKeys({ + keys: [...remembered.filter((value) => value !== key), key], + count: this.settings.retainedIdempotencyKeys, + bytes: this.settings.retainedIdempotencyKeysBytes, + }), ) } @@ -2422,6 +2424,12 @@ function retentionPolicy(options: { } } +export function boundedKeys(options: { keys: string[]; count: number; bytes: number }): string[] { + const kept = options.keys.slice(-options.count) + while (kept.length > 0 && utf8ByteLength(JSON.stringify(kept)) > options.bytes) kept.shift() + return kept +} + function rememberedList(stored: string | null): string[] { if (stored === null) return [] const parsed: unknown = JSON.parse(stored) diff --git a/test/cloudflare/recovery.test.ts b/test/cloudflare/recovery.test.ts index b93fe0f..fb39a75 100644 --- a/test/cloudflare/recovery.test.ts +++ b/test/cloudflare/recovery.test.ts @@ -371,6 +371,26 @@ describe("Cloudflare recovery and fencing", () => { expect(remembered).toEqual(["key-2", "key-3", "key-4"]) }) + it("bounds what an instance remembers by size", async () => { + const reference = runtime().ref(Counter, "bounded-key-bytes") + const key = "k".repeat(200) + const message = await reference.send + .with({ authorizationContext, idempotencyKey: key }) + .increment() + await message.wait({ authorizationContext }) + + const remembered = await runInDurableObject(stub("bounded-key-bytes"), (_object, state) => { + const instance = JSON.parse( + state.storage.sql + .exec<{ value: string }>("SELECT value FROM metadata WHERE key = 'instance'") + .one().value, + ) as Instance + return instance.completedIdempotencyKeys + }) + + expect(remembered).toEqual([]) + }) + it("continues bounded receipt cleanup using its saved alarm", async () => { const reference = runtime().ref(Counter, "receipt-cleanup") const message = await reference.send.with({ authorizationContext }).increment() diff --git a/test/cloudflare/worker.ts b/test/cloudflare/worker.ts index efe0847..f8fc5a5 100644 --- a/test/cloudflare/worker.ts +++ b/test/cloudflare/worker.ts @@ -243,6 +243,7 @@ export class Actors extends createDurableObjectsHost({ authorizeAdministration: (input) => input.authorizationContext === "allowed", retryDelayMilliseconds: () => 10, retainedIdempotencyKeys: 3, + retainedIdempotencyKeysBytes: 64, effects: { callbackEmpty: (_arguments, context) => { deliveries.set(context.id, context.attempt) diff --git a/test/result-lookup.test.ts b/test/result-lookup.test.ts index 77d2661..eff6057 100644 --- a/test/result-lookup.test.ts +++ b/test/result-lookup.test.ts @@ -38,7 +38,7 @@ afterEach(async () => { }) async function start( - overrides: { retainedIdempotencyKeys?: number } = {}, + overrides: { retainedIdempotencyKeys?: number; retainedIdempotencyKeysBytes?: number } = {}, ): Promise { const created = configure({ database: sqlite({ path: ":memory:" }), @@ -360,6 +360,33 @@ describe("result lookup", () => { await expect(reference.findBy({ idempotencyKey: "second" })).rejects.toThrow(MessagePruned) }) + it("bounds what an instance remembers by size", async () => { + const active = await start({ retainedIdempotencyKeysBytes: 64 }) + const reference = active.ref(CartActor, "alice") + const keys = [0, 1, 2].map((index) => `${index}-${"k".repeat(20)}`) + for (const [index, key] of keys.entries()) { + await reference.send.with({ idempotencyKey: key }).checkout({ orderId: index }) + } + await active.worker().runUntilIdle() + + const remembered = await rememberedKeys(active) + + expect(remembered).toEqual(keys.slice(-2)) + expect(JSON.stringify(remembered).length).toBeLessThanOrEqual(64) + }) + + it("remembers nothing for a key larger than what it retains", async () => { + const active = await start({ retainedIdempotencyKeysBytes: 16 }) + const reference = active.ref(CartActor, "alice") + const key = "k".repeat(100) + await reference.send.with({ idempotencyKey: key }).checkout({ orderId: 1 }) + await active.worker().runUntilIdle() + await deleteMessages(active) + + expect(await rememberedKeys(active)).toEqual([]) + expect(await reference.findBy({ idempotencyKey: key })).toBeUndefined() + }) + it("remembers a re-sent key once", async () => { const active = await start() const reference = active.ref(CartActor, "alice") From bf32cadb675a57b6c285423768792d4255e4a407 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Wed, 23 Sep 2026 12:05:57 -0700 Subject: [PATCH 06/14] fix: answer the Greptile review Five findings. The missing Cloudflare `findBy` and `messageOutcome` were already implemented on this branch; the other four were valid. `messageOutcome` read the row and then read the status separately, so a worker finishing between the two produced an outcome no snapshot ever held. `repository.messageWithStatus` returns both from one connection, and a test breaks `messageStatus` to prove the outcome no longer needs it. `readableMessage` and `readableState` caught every exception and reported absence, so an authorization service outage was indistinguishable from a missing row. They catch `Unauthorized` and `UnknownActorType` only. Without that the new test reads `promise resolved "undefined" instead of rejecting`. `Outcome` broke the rule against `unknown` annotations; it defaults to `JsonValue`, matching `invoke` and `sendMessage`. The test's `argumentsValue` is a `JsonObject`. The uniqueness test updated a row whose id was "impossible", so it changed nothing and passed with or without the index. It now assigns one message's request id to a second real message and asserts the database refuses. Validation: pnpm test (523), pnpm run test:cloudflare (58), check, build, format:check. Co-Authored-By: Claude Opus 5 (1M context) --- src/outcome.ts | 4 ++-- src/repository.ts | 26 +++++++++++++++++++++++- src/runtime.ts | 36 +++++++++++++++++++++++---------- test/result-lookup.test.ts | 41 ++++++++++++++++++++++++++++++++++---- 4 files changed, 90 insertions(+), 17 deletions(-) diff --git a/src/outcome.ts b/src/outcome.ts index 232a206..9541464 100644 --- a/src/outcome.ts +++ b/src/outcome.ts @@ -1,4 +1,4 @@ -import type { DeepReadonly, JsonObject, MessageStatus } from "./types.js" +import type { DeepReadonly, JsonObject, JsonValue, MessageStatus } from "./types.js" export interface ErrorRecord { readonly name: string @@ -11,7 +11,7 @@ export interface RejectionRecord { readonly details: DeepReadonly } -export interface Outcome { +export interface Outcome { readonly status: MessageStatus readonly result: DeepReadonly | undefined readonly error: ErrorRecord | undefined diff --git a/src/repository.ts b/src/repository.ts index 7380c27..562e784 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -1060,6 +1060,19 @@ export class Repository { }) } + async messageWithStatus( + id: string, + ): Promise<{ message: MessageRow; status: MessageStatus } | undefined> { + return this.settings.database.connection(async (connection) => { + const message = await connection.get( + `SELECT * FROM ${this.table("messages")} WHERE id = ?`, + [id], + ) + if (!message) return undefined + return { message, status: await this.statusOf({ connection, message }) } + }) + } + async messageStatus( id: string, ): Promise<"ready" | "claimed" | "completed" | "rejected" | "dead" | "unknown"> { @@ -1069,6 +1082,17 @@ export class Repository { [id], ) if (!message) return "unknown" + return this.statusOf({ connection, message }) + }) + } + + private async statusOf(options: { + connection: DatabaseConnection + message: MessageRow + }): Promise { + const { connection, message } = options + const id = message.id + { if (message.rejection !== null) return "rejected" if (message.completed_at_ms !== null) { const dead = await connection.get<{ found: number | bigint }>( @@ -1087,7 +1111,7 @@ export class Repository { [id], ) return ready ? "ready" : "unknown" - }) + } } async syncDiagnostics(messageId: string): Promise { diff --git a/src/runtime.ts b/src/runtime.ts index 0965049..dff0d4b 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -722,16 +722,30 @@ export class SolidObjectsRuntime { messageReference, options.authorizationContext, ) + const found = await this.repository.messageWithStatus(message.id) + if (!found) { + return Object.freeze({ + status: "unknown" as const, + result: undefined, + error: undefined, + rejection: undefined, + attempts: 0, + }) + } + + const snapshot = found.message return Object.freeze({ - status: await this.repository.messageStatus(message.id), + status: found.status, result: - message.result === null + snapshot.result === null ? undefined - : (normalizeJson(JSON.parse(message.result)) as DeepReadonly), - error: message.error === null ? undefined : (JSON.parse(message.error) as ErrorRecord), + : (normalizeJson(JSON.parse(snapshot.result)) as DeepReadonly), + error: snapshot.error === null ? undefined : (JSON.parse(snapshot.error) as ErrorRecord), rejection: - message.rejection === null ? undefined : (JSON.parse(message.rejection) as RejectionRecord), - attempts: Number(message.attempt_count), + snapshot.rejection === null + ? undefined + : (JSON.parse(snapshot.rejection) as RejectionRecord), + attempts: Number(snapshot.attempt_count), }) } @@ -1757,8 +1771,9 @@ export class SolidObjectsRuntime { authorizationContext, }) return true - } catch { - return false + } catch (error) { + if (error instanceof Unauthorized) return false + throw error } } @@ -1793,8 +1808,9 @@ export class SolidObjectsRuntime { authorizationContext, }) return true - } catch { - return false + } catch (error) { + if (error instanceof Unauthorized || error instanceof UnknownActorType) return false + throw error } } diff --git a/test/result-lookup.test.ts b/test/result-lookup.test.ts index eff6057..a76e29c 100644 --- a/test/result-lookup.test.ts +++ b/test/result-lookup.test.ts @@ -3,6 +3,7 @@ import { Actor } from "../src/actor.js" import { sqlite } from "../src/database/sqlite.js" import { MessagePruned, Rejected } from "../src/errors.js" import { configure, type SolidObjectsRuntime } from "../src/runtime.js" +import type { JsonObject } from "../src/types.js" class CartActor extends Actor { static override readonly actorType = "LookupCartActor" @@ -26,7 +27,7 @@ class CartActor extends Actor { } let runtime: SolidObjectsRuntime | undefined -let seenOperations: { operation: string; argumentsValue: unknown }[] = [] +let seenOperations: { operation: string; argumentsValue: JsonObject }[] = [] let hooks: string[] = [] afterEach(async () => { @@ -410,16 +411,48 @@ describe("result lookup", () => { it("keeps request ids unique across the table", async () => { const active = await start() - const original = await active.ref(CartActor, "alice").send.checkout({ orderId: 1 }) + const first = await active.ref(CartActor, "alice").send.checkout({ orderId: 1 }) + const second = await active.ref(CartActor, "bob").send.checkout({ orderId: 2 }) await expect( active.settings.database.transaction((connection) => connection.run( `UPDATE ${active.repository.table("messages")} SET request_id = ? WHERE id = ?`, - [original.requestId, "impossible"], + [first.requestId, second.id], ), ), - ).resolves.toBeDefined() + ).rejects.toThrow() + }) + + it("reports one snapshot for every outcome field", async () => { + const active = await start() + const original = await active.ref(CartActor, "alice").send.checkout({ orderId: 4210 }) + await active.worker().runUntilIdle() + const found = await active.findBy({ requestId: original.requestId }) + active.repository.messageStatus = () => { + throw new Error("outcome must not read the status separately") + } + + const outcome = await found!.outcome() + + expect(outcome.status).toBe("completed") + expect(outcome.result).toEqual({ orderId: 4210 }) + }) + + it("propagates an authorization failure rather than reporting absence", async () => { + const active = await start() + const original = await active.ref(CartActor, "alice").send.checkout({ orderId: 1 }) + const failing = configure({ + database: active.settings.database, + authorizeMessage: () => { + throw new Error("authorization service is down") + }, + }) + failing.register(CartActor) + + await expect(failing.findBy({ requestId: original.requestId })).rejects.toThrow( + /authorization service is down/, + ) }) }) From ccfdb6c3d9f610721bb91132af73fe693718d457 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Wed, 23 Sep 2026 12:19:46 -0700 Subject: [PATCH 07/14] refactor: derive a message status from one read `messageStatus` repeated the row read that `messageWithStatus` already performs, and `statusOf` carried a leftover scope block from the extraction. It reads through `messageWithStatus` instead. Co-Authored-By: Claude Opus 5 (1M context) --- src/repository.ts | 48 ++++++++++++++++++----------------------------- 1 file changed, 18 insertions(+), 30 deletions(-) diff --git a/src/repository.ts b/src/repository.ts index 562e784..f163e0a 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -1073,17 +1073,8 @@ export class Repository { }) } - async messageStatus( - id: string, - ): Promise<"ready" | "claimed" | "completed" | "rejected" | "dead" | "unknown"> { - return this.settings.database.connection(async (connection) => { - const message = await connection.get( - `SELECT * FROM ${this.table("messages")} WHERE id = ?`, - [id], - ) - if (!message) return "unknown" - return this.statusOf({ connection, message }) - }) + async messageStatus(id: string): Promise { + return (await this.messageWithStatus(id))?.status ?? "unknown" } private async statusOf(options: { @@ -1091,27 +1082,24 @@ export class Repository { message: MessageRow }): Promise { const { connection, message } = options - const id = message.id - { - if (message.rejection !== null) return "rejected" - if (message.completed_at_ms !== null) { - const dead = await connection.get<{ found: number | bigint }>( - `SELECT 1 AS found FROM ${this.table("dead_letters")} WHERE message_id = ?`, - [id], - ) - return dead ? "dead" : "completed" - } - const claimed = await connection.get<{ found: number | bigint }>( - `SELECT 1 AS found FROM ${this.table("claimed_messages")} WHERE message_id = ?`, - [id], - ) - if (claimed) return "claimed" - const ready = await connection.get<{ found: number | bigint }>( - `SELECT 1 AS found FROM ${this.table("ready_messages")} WHERE message_id = ?`, - [id], + if (message.rejection !== null) return "rejected" + if (message.completed_at_ms !== null) { + const dead = await connection.get<{ found: number | bigint }>( + `SELECT 1 AS found FROM ${this.table("dead_letters")} WHERE message_id = ?`, + [message.id], ) - return ready ? "ready" : "unknown" + return dead ? "dead" : "completed" } + const claimed = await connection.get<{ found: number | bigint }>( + `SELECT 1 AS found FROM ${this.table("claimed_messages")} WHERE message_id = ?`, + [message.id], + ) + if (claimed) return "claimed" + const ready = await connection.get<{ found: number | bigint }>( + `SELECT 1 AS found FROM ${this.table("ready_messages")} WHERE message_id = ?`, + [message.id], + ) + return ready ? "ready" : "unknown" } async syncDiagnostics(messageId: string): Promise { From b3b682b2c8dfd70df193c606427f516b43ff56ed Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Wed, 23 Sep 2026 12:38:12 -0700 Subject: [PATCH 08/14] test: wait for the oversized record rather than a deadline The `quality` job failed on a test that passed on the same commit in the sibling run: expected MessageFailed / PayloadTooLarge received SyncTimeout, status "ready", timeoutMilliseconds 402 The test drove an oversized record through a synchronous call with a 500 ms budget. Moving about 1 MB through a Durable Object under CI load takes longer than that, so the caller gave up before the turn reached `encodedRecord` and the assertion read a timeout instead of the rollback it was written for. The budget was never part of what the test asserts. It now sends the message, polls until the message is dead, and reads the stored error through `outcome()`, so no deadline can decide the result. That also exercises `messageOutcome` on the Durable Objects runtime, which this branch added. The test dates from the original Durable Objects backend and the flake predates this branch. Validation: five runs of the Cloudflare suite, four of them concurrent, all 20 tests passing. Raising the record limit to 99_999_000 makes the test fail, so it still reports the defect it names. Co-Authored-By: Claude Opus 5 (1M context) --- test/cloudflare/recovery.test.ts | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/test/cloudflare/recovery.test.ts b/test/cloudflare/recovery.test.ts index fb39a75..0aa9c24 100644 --- a/test/cloudflare/recovery.test.ts +++ b/test/cloudflare/recovery.test.ts @@ -68,11 +68,16 @@ describe("Cloudflare recovery and fencing", () => { it("rolls back a result that exceeds the aggregate SQLite record limit", async () => { const reference = runtime().ref(Counter, "oversized-record") - await expect( - reference - .with({ authorizationContext, timeoutMilliseconds: 500 }) - .echo({ value: "x".repeat(1_010_000) }), - ).rejects.toMatchObject({ name: "MessageFailed", details: { name: "PayloadTooLarge" } }) + const message = await reference.send + .with({ authorizationContext }) + .echo({ value: "x".repeat(1_010_000) }) + + await expect + .poll(() => message.status({ authorizationContext }), { timeout: 15_000 }) + .toBe("dead") + + const outcome = await message.outcome({ authorizationContext }) + expect(outcome.error?.name).toBe("PayloadTooLarge") expect((await reference.snapshot({ authorizationContext })).count).toBe(0) }) From 030c2bd6ed8fc20b15ace9091d9a4257446b58d5 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Wed, 23 Sep 2026 12:56:10 -0700 Subject: [PATCH 09/14] fix: authorize a pruned key like the message it replaces Four valid findings from Greptile. The pruned answer authorized the synthetic `__snapshot__` query while a surviving row authorizes its stored operation, so a caller allowed to read state but denied the operation could tell `MessagePruned` from `undefined` and learn the operation had run. Each remembered key now carries its operation, and `readableOperation` runs the same hook both paths use. Without it the new test resolves where it must reject. The Durable Objects lookup demanded a synthetic `authorizeQuery("__lookupMessage__")` before authorizing the stored operation, so a policy that allows only declared queries refused an otherwise authorized lookup and diverged from the SQL runtime. The synthetic check is gone; the stored operation is still authorized. The Durable Objects `PayloadTooLarge` fallback marked a message dead without remembering its key, so that one terminal path answered absent where every other dead message answers pruned. `unknown` appeared in a catch annotation and two assertion chains. `errorRecord` and `rejectionRecord` build the values from validated fields instead, and `rememberedList` parses to a declared type. Validation: pnpm test (524), pnpm run test:cloudflare (58), check, build, format:check. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 9 +++++-- docs/api.md | 6 +++-- docs/parity.md | 4 ++- src/cloudflare/engine.ts | 34 +++++++++++++++--------- src/cloudflare/records.ts | 3 ++- src/cloudflare/runtime.ts | 21 ++++++++++----- src/records.ts | 5 ++++ src/repository.ts | 34 +++++++++++++++++------- src/runtime.ts | 45 +++++++++++++++++++++++--------- test/cloudflare/recovery.test.ts | 4 +-- test/cloudflare/worker.ts | 2 +- test/result-lookup.test.ts | 27 ++++++++++++++++--- 12 files changed, 142 insertions(+), 52 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4657d74..4190fb7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,8 +24,13 @@ no second write. `reference.findBy({ idempotencyKey })` throws `MessagePruned` for a key the actor remembers and whose message retention removed, and still returns `undefined` for a key no caller ever sent. The - memory is actor state, so a caller that `authorizeQuery` refuses reads - `undefined` for both. `retainedIdempotencyKeys` bounds the memory and defaults + An actor remembers the operation beside each key, so the pruned answer runs + the same hook against the same operation a lookup of the surviving row would, + and a caller the policy refuses reads `undefined` for both. Gating it on + `snapshot` would have told a caller who may read state, but not the + operation, that the operation had run. The Durable Objects lookup no longer + demands a synthetic `__lookupMessage__` query in addition, which a policy + that allows only declared queries refused. `retainedIdempotencyKeys` bounds the memory and defaults to 64 keys for each actor. A lookup by request id cannot make the distinction, because the runtime generates a request id and no actor remembers one. `retainedIdempotencyKeysBytes` bounds the serialized memory as diff --git a/docs/api.md b/docs/api.md index 807545e..5d1ab9e 100644 --- a/docs/api.md +++ b/docs/api.md @@ -562,8 +562,10 @@ class and result types are also exported for integration typing. `retainedIdempotencyKeys` finished turns, so `reference.findBy({ idempotencyKey })` throws `MessagePruned` for a key the actor remembers and whose message retention removed, and returns `undefined` for a key no caller - ever sent. The memory is actor state, so a caller that `authorizeQuery` - refuses reads `undefined` for both. `runtime.findBy({ requestId })` returns + ever sent. An actor remembers the operation beside each key, so the pruned + answer runs the same hook against the same operation a lookup of the + surviving row would, and a caller the policy refuses reads `undefined` for + both. `runtime.findBy({ requestId })` returns `undefined` in both cases, because the runtime generates a request id and no actor remembers one. - `retainedIdempotencyKeysBytes` bounds the serialized memory as well, because diff --git a/docs/parity.md b/docs/parity.md index ebc1644..7150d7e 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -165,7 +165,9 @@ An actor remembers the idempotency keys of its own last `retained_idempotency_keys` / `retainedIdempotencyKeys` finished turns, written in the instance row the executor updates anyway, so a key lookup raises `MessagePruned` for a message retention removed and answers absent for a -message that never existed. Both also bound the serialized memory, because an +message that never existed. Each remembered key carries its operation, so the +pruned answer runs the same authorization a lookup of the surviving row +would. Both also bound the serialized memory, because an idempotency key has no length limit and the memory outlives the message row; an actor drops its oldest keys until the list fits. The memory is actor state, so the query hook gates the pruned answer in both. A request id lookup answers absent in both cases, diff --git a/src/cloudflare/engine.ts b/src/cloudflare/engine.ts index 73eebfa..eb88f9d 100644 --- a/src/cloudflare/engine.ts +++ b/src/cloudflare/engine.ts @@ -314,14 +314,6 @@ export class ActorEngine { private async readMessage(input: HostRequest): Promise { let message: Message | undefined if (input.method === "lookup") { - if ( - !(await this.settings.authorizeQuery({ - ...input, - operation: "__lookupMessage__", - arguments: {}, - })) - ) - throw new Unauthorized("message lookup is not authorized") message = this.lookUp(input) if (!message) return this.prunedReply(input) } else { @@ -355,11 +347,25 @@ export class ActorEngine { return receipt ? this.store.message(receipt.message_id) : undefined } - private prunedReply(input: HostRequest): JsonValue { + private async prunedReply(input: HostRequest): Promise { const key = input.payload.idempotencyKey if (key === undefined) return null - const remembered = this.store.instance()?.completedIdempotencyKeys ?? [] - return remembered.includes(String(key)) ? { pruned: true } : null + const remembered = (this.store.instance()?.completedIdempotencyKeys ?? []).find( + (entry) => entry.key === String(key), + ) + if (!remembered) return null + if (!(await this.authorizedOperation(input, remembered.operation))) return null + + return { pruned: true } + } + + private async authorizedOperation(input: HostRequest, operation: string): Promise { + const definition = this.definition(input.actorType) + const query = definition.queries.includes(operation) + if (!query && !definition.operations.includes(operation)) return false + + const authorize = query ? this.settings.authorizeQuery : this.settings.authorizeMessage + return authorize({ ...input, operation, arguments: {} }) } private rememberKey(instance: Instance, message: Message): void { @@ -367,10 +373,11 @@ export class ActorEngine { if (key === null) return const remembered = instance.completedIdempotencyKeys ?? [] - if (remembered.at(-1) === key) return + const entry = { key, operation: message.operation } + if (remembered.at(-1)?.key === key && remembered.at(-1)?.operation === entry.operation) return instance.completedIdempotencyKeys = boundedKeys({ - keys: [...remembered.filter((value) => value !== key), key], + keys: [...remembered.filter((value) => value.key !== key), entry], count: this.settings.retainedIdempotencyKeys, bytes: this.settings.retainedIdempotencyKeysBytes, }) @@ -682,6 +689,7 @@ export class ActorEngine { message.rejection = null message.error = { name: storageError.name, message: storageError.message } current.paused = true + this.rememberKey(current, message) this.store.saveInstance(current) this.pauseReminder(message) this.store.saveMessage(message) diff --git a/src/cloudflare/records.ts b/src/cloudflare/records.ts index 25db7d1..b856cf9 100644 --- a/src/cloudflare/records.ts +++ b/src/cloudflare/records.ts @@ -1,5 +1,6 @@ import type { JsonObject, JsonValue, MessageStatus } from "../types.js" import type { ActorIdentity } from "./protocol.js" +import type { RememberedKey } from "../records.js" export interface Instance extends ActorIdentity { incarnation: string @@ -11,7 +12,7 @@ export interface Instance extends ActorIdentity { stateVersion: number createdAt: number paused: boolean - completedIdempotencyKeys?: string[] + completedIdempotencyKeys?: RememberedKey[] } export interface Message { diff --git a/src/cloudflare/runtime.ts b/src/cloudflare/runtime.ts index fe7f7d3..bebabd3 100644 --- a/src/cloudflare/runtime.ts +++ b/src/cloudflare/runtime.ts @@ -157,7 +157,7 @@ export class CloudflareRuntime implements ActorRuntime { input.requestId === undefined ? { idempotencyKey: input.idempotencyKey! } : { requestId: input.requestId }, - }).catch((error: unknown) => { + }).catch((error) => { if (error instanceof Unauthorized) return null throw error }) @@ -177,12 +177,9 @@ export class CloudflareRuntime implements ActorRuntime { status: record.status as MessageStatus, result: record.result === null ? undefined : (readonlyCopy(record.result) as DeepReadonly), - error: - record.error === null ? undefined : (jsonObject(record.error) as unknown as ErrorRecord), + error: record.error === null ? undefined : errorRecord(jsonObject(record.error)), rejection: - record.rejection === null - ? undefined - : (jsonObject(record.rejection) as unknown as RejectionRecord), + record.rejection === null ? undefined : rejectionRecord(jsonObject(record.rejection)), attempts: Number(record.attempt), }) } @@ -492,3 +489,15 @@ export async function beforeDeadline( if (timer !== undefined) clearTimeout(timer) } } + +function errorRecord(value: JsonObject): ErrorRecord { + return { name: String(value.name), message: String(value.message) } +} + +function rejectionRecord(value: JsonObject): RejectionRecord { + return { + code: String(value.code), + message: String(value.message), + details: readonlyCopy(jsonObject(value.details ?? {})), + } +} diff --git a/src/records.ts b/src/records.ts index e8469a8..eac788b 100644 --- a/src/records.ts +++ b/src/records.ts @@ -147,3 +147,8 @@ export interface BroadcastRow { claimed_by: string | null error: string | null } + +export interface RememberedKey { + key: string + operation: string +} diff --git a/src/repository.ts b/src/repository.ts index f163e0a..c582a04 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -20,6 +20,7 @@ import type { EffectRow, EnqueueInput, InstanceRow, + RememberedKey, MessageRow, ProcessRow, ReminderRow, @@ -1010,11 +1011,11 @@ export class Repository { ) } - async remembersIdempotencyKey(input: { + async rememberedIdempotencyKey(input: { actorType: string actorId: string idempotencyKey: string - }): Promise { + }): Promise { const row = await this.settings.database.connection((connection) => connection.get>( `SELECT completed_idempotency_keys FROM ${this.table("instances")} @@ -1022,7 +1023,9 @@ export class Repository { [input.actorType, input.actorId], ), ) - return rememberedList(row?.completed_idempotency_keys ?? null).includes(input.idempotencyKey) + return rememberedList(row?.completed_idempotency_keys ?? null).find( + (entry) => entry.key === input.idempotencyKey, + ) } async findMessageByIdempotencyKey(input: { @@ -2292,11 +2295,14 @@ export class Repository { if (key === null) return stored const remembered = rememberedList(stored) - if (remembered.at(-1) === key) return stored + const entry = { key, operation: turn.message.operation } + if (remembered.at(-1)?.key === key && remembered.at(-1)?.operation === entry.operation) { + return stored + } return JSON.stringify( boundedKeys({ - keys: [...remembered.filter((value) => value !== key), key], + keys: [...remembered.filter((value) => value.key !== key), entry], count: this.settings.retainedIdempotencyKeys, bytes: this.settings.retainedIdempotencyKeysBytes, }), @@ -2436,17 +2442,27 @@ function retentionPolicy(options: { } } -export function boundedKeys(options: { keys: string[]; count: number; bytes: number }): string[] { +export function boundedKeys(options: { + keys: RememberedKey[] + count: number + bytes: number +}): RememberedKey[] { const kept = options.keys.slice(-options.count) while (kept.length > 0 && utf8ByteLength(JSON.stringify(kept)) > options.bytes) kept.shift() return kept } -function rememberedList(stored: string | null): string[] { +export function rememberedList(stored: string | null): RememberedKey[] { if (stored === null) return [] - const parsed: unknown = JSON.parse(stored) + const parsed = JSON.parse(stored) as RememberedKey[] if (!Array.isArray(parsed)) return [] - return parsed.filter((value): value is string => typeof value === "string") + return parsed.filter( + (value) => + value !== null && + typeof value === "object" && + typeof value.key === "string" && + typeof value.operation === "string", + ) } function parameterList(length: number): string { diff --git a/src/runtime.ts b/src/runtime.ts index dff0d4b..1761397 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -1747,32 +1747,53 @@ export class SolidObjectsRuntime { if (input.idempotencyKey === undefined) return undefined const reference = input.reference! - const remembered = await this.repository.remembersIdempotencyKey({ + const remembered = await this.repository.rememberedIdempotencyKey({ actorType: reference.actorType, actorId: reference.actorId, idempotencyKey: input.idempotencyKey, }) if (!remembered) return undefined - if (!(await this.readableState(reference, input.authorizationContext))) return undefined + const readable = await this.readableOperation({ + actorType: reference.actorType, + actorId: reference.actorId, + operation: remembered.operation, + argumentsValue: {}, + authorizationContext: input.authorizationContext, + }) + if (!readable) return undefined throw new MessagePruned(input.idempotencyKey) } - private async readableState( - reference: ActorReferenceCore, - authorizationContext: AdministrationOptions["authorizationContext"], - ): Promise { + private async readableOperation(input: { + actorType: string + actorId: string + operation: string + argumentsValue: JsonObject + authorizationContext: AdministrationOptions["authorizationContext"] + }): Promise { try { + const registered = this.fetchActor(input.actorType) + if (!registered.operations.has(input.operation) && !registered.queries.has(input.operation)) + return false + await this.authorize({ - kind: "query", - reference, - operation: "__snapshot__", - argumentsValue: {}, - authorizationContext, + kind: this.isQuery(registered.definition, input.operation) ? "query" : "message", + reference: new ActorReferenceCore({ + runtime: this, + actorClass: registered.actorClass, + actorType: input.actorType, + actorId: input.actorId, + operations: registered.operations, + queries: registered.queries, + }), + operation: input.operation, + argumentsValue: input.argumentsValue, + authorizationContext: input.authorizationContext, }) return true } catch (error) { - if (error instanceof Unauthorized) return false + if (error instanceof Unauthorized || error instanceof UnknownActorType) return false throw error } } diff --git a/test/cloudflare/recovery.test.ts b/test/cloudflare/recovery.test.ts index 0aa9c24..7c926d4 100644 --- a/test/cloudflare/recovery.test.ts +++ b/test/cloudflare/recovery.test.ts @@ -370,7 +370,7 @@ describe("Cloudflare recovery and fencing", () => { .exec<{ value: string }>("SELECT value FROM metadata WHERE key = 'instance'") .one().value, ) as Instance - return instance.completedIdempotencyKeys + return instance.completedIdempotencyKeys?.map((entry) => entry.key) }) expect(remembered).toEqual(["key-2", "key-3", "key-4"]) @@ -378,7 +378,7 @@ describe("Cloudflare recovery and fencing", () => { it("bounds what an instance remembers by size", async () => { const reference = runtime().ref(Counter, "bounded-key-bytes") - const key = "k".repeat(200) + const key = "k".repeat(400) const message = await reference.send .with({ authorizationContext, idempotencyKey: key }) .increment() diff --git a/test/cloudflare/worker.ts b/test/cloudflare/worker.ts index f8fc5a5..906b6fa 100644 --- a/test/cloudflare/worker.ts +++ b/test/cloudflare/worker.ts @@ -243,7 +243,7 @@ export class Actors extends createDurableObjectsHost({ authorizeAdministration: (input) => input.authorizationContext === "allowed", retryDelayMilliseconds: () => 10, retainedIdempotencyKeys: 3, - retainedIdempotencyKeysBytes: 64, + retainedIdempotencyKeysBytes: 256, effects: { callbackEmpty: (_arguments, context) => { deliveries.set(context.id, context.attempt) diff --git a/test/result-lookup.test.ts b/test/result-lookup.test.ts index a76e29c..5e42d2c 100644 --- a/test/result-lookup.test.ts +++ b/test/result-lookup.test.ts @@ -311,6 +311,24 @@ describe("result lookup", () => { ).toBeUndefined() }) + it("does not tell a snapshot-only caller that a key was pruned", async () => { + const active = await start() + const reference = active.ref(CartActor, "alice") + await reference.send.with({ idempotencyKey: "checkout-7f3a" }).checkout({ orderId: 1 }) + await active.worker().runUntilIdle() + await deleteMessages(active) + const refusing = configure({ + database: active.settings.database, + authorizeQuery: () => true, + authorizeMessage: () => false, + }) + refusing.register(CartActor) + + expect( + await refusing.ref(CartActor, "alice").findBy({ idempotencyKey: "checkout-7f3a" }), + ).toBeUndefined() + }) + it("remembers a key whose message was rejected", async () => { const active = await start() const reference = active.ref(CartActor, "alice") @@ -362,7 +380,7 @@ describe("result lookup", () => { }) it("bounds what an instance remembers by size", async () => { - const active = await start({ retainedIdempotencyKeysBytes: 64 }) + const active = await start({ retainedIdempotencyKeysBytes: 128 }) const reference = active.ref(CartActor, "alice") const keys = [0, 1, 2].map((index) => `${index}-${"k".repeat(20)}`) for (const [index, key] of keys.entries()) { @@ -373,7 +391,6 @@ describe("result lookup", () => { const remembered = await rememberedKeys(active) expect(remembered).toEqual(keys.slice(-2)) - expect(JSON.stringify(remembered).length).toBeLessThanOrEqual(64) }) it("remembers nothing for a key larger than what it retains", async () => { @@ -484,5 +501,9 @@ async function rememberedKeys(active: SolidObjectsRuntime): Promise { `SELECT completed_idempotency_keys FROM ${active.repository.table("instances")}`, ), ) - return JSON.parse(row?.completed_idempotency_keys ?? "[]") as string[] + const remembered = JSON.parse(row?.completed_idempotency_keys ?? "[]") as { + key: string + operation: string + }[] + return remembered.map((entry) => entry.key) } From 841bf4c43424f625089434f1de455d1d5091f4a2 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Wed, 23 Sep 2026 13:12:07 -0700 Subject: [PATCH 10/14] docs: point scaling limits at Pro, not at the roadmap `parity.md` said distributed rate limits and global admission control "do not exist yet, matching the open Ruby roadmap item". That roadmap item is gone: the three scaling limits are hot, request-critical, and loss-tolerant, and every invocation writes one permanent message row, so `fit.md` already ruled out checking on the request path. Solid Objects Pro answers them with grouped and ephemeral operations, and both documents now say so rather than pointing at a milestone. The README listed "Rate limits and account quotas" among the fits. The low-rate quota is the case that fits, so it reads that way now. Validation: pnpm test (524), check, format:check. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 2 +- docs/fit.md | 4 +++- docs/parity.md | 40 ++++++++++++++++++++-------------------- 3 files changed, 24 insertions(+), 22 deletions(-) diff --git a/README.md b/README.md index 8de508d..e58af1b 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,7 @@ Solid Object JavaScript Actors elegantly fit anything where one identifiable thi - Ticket holds and reservations - Multiplayer games and shared rooms - Shopping carts and checkout recovery -- Rate limits and account quotas +- Low-rate quotas and account limits - Session expiration - Job leases and workflows - Connected devices diff --git a/docs/fit.md b/docs/fit.md index 6b31f31..f342e15 100644 --- a/docs/fit.md +++ b/docs/fit.md @@ -43,7 +43,9 @@ history, and no actor-state migration contract. limiter, see [Solid Objects Pro](https://solidobjects.pro/): grouped commits coalesce concurrent writes into one insert, and ephemeral operations keep loss-tolerant calls out of the durable journal. It ships for the Rails gem - today, and the Node build is in development. + today, and the Node build is in development. Distributed per-actor rate + limits, global admission control, and cache-capacity eviction are answered + there rather than in this runtime, and are not open roadmap items here. - Durable workflow replay across named steps is more important than a mutable object with ordered operations. The runtime redelivers an ordered message and retries it. It does not replay a function from a step log. diff --git a/docs/parity.md b/docs/parity.md index 7150d7e..46f3a47 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -51,26 +51,26 @@ SQL waiters read results and status from one statement. Ruby already checks completion and returns the result from the same loaded message; no Ruby change is needed for the JavaScript stale-result race fix. -| Capability | Status | TypeScript shape or remaining work | -| ------------------------------------------------------------------------------------------------------------- | ------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| Actor registry, durable identity, JSON state, and adjacent state migrations | Native | Ordinary classes, static actor types, inferred state, explicit migrations, and isolated runtime context across every actor-instance callback. | -| Fluent committed calls and background delivery | Native | `await reference.operation()` and `reference.send.operation()`. | -| Ordered mailbox, sequence allocation, idempotency, retries, dead letters, leases, renewal, and fenced commits | Native | Relational ready/claimed membership tables, distinct generated request IDs and caller idempotency keys, durable history, adapter-appropriate sequence locking, and PostgreSQL/MySQL row locks held from fence validation through commit. | -| Domain rejection and strict poison ordering | Native | Rejections accept JavaScript identifier-style codes and roll back without retry; invalid codes fail terminally, while retryable failures block later operations until completion or dead-lettering. | -| Bounded activation passes and hot-actor fairness | Native | Configurable turn-count and elapsed-time budgets bound each pass, then move only that actor's already-due memberships behind actors already waiting. | -| Bounded claim candidate scan | Native | A configurable ordered scan continues to another ready actor when a worker loses the first candidate's lease race. | -| Backpressure and payload caps | Partial | Serialization enforces a shared maximum JSON nesting depth, raising `InvalidPayload`, and an optional caller-supplied `maxBytes` limit, raising `PayloadTooLarge`; reminder names are bounded to 255 characters. Distributed per-actor rate limits and global admission control do not exist yet, matching the open Ruby roadmap item. | -| Idle activation cache | Native | Long-running workers retain hydrated actors under renewable fenced leases, restore public state after failed turns, and release on timeout, fairness yield, lease loss, or shutdown. | -| Transactional effects and outcome operations | Native | At-least-once handlers receive immutable stable effect, attempt, source-message, and actor identity; success and failure operations also receive the originally staged arguments for correlation. Typed callback envelopes are exported. | -| Actor-to-actor delivery | Native | `sendTo(reference).operation()` stages delivery in the source actor commit. | -| One-shot and recurring reminders | Native | Scheduling, replacement events, catch-up policy, stale-claim recovery, pausing, authorized inspection, and idempotent resume are implemented. | -| Same-database commit actions | Native | Registered actions receive source-message identity, mailbox sequence, activation generation, and the fenced transaction connection. | -| Ambient transaction rejection | Native | Committed calls and message waits fail before blocking when the current async context already owns a transaction on the Solid Objects adapter. | -| Direct application-write isolation during actor code | Partial | `guardApplicationDatabase()` fails closed for operations, projections, migrations, and commit actions; only the supplied fenced commit-action connection may write. Unwrapped clients cannot be intercepted. | -| Committed snapshots | Native | `snapshot()` returns authorized persisted fields and inferred getters from one read-only committed state image; realtime replay reads explicit observables without mailbox history. | -| Actor destruction and incarnation fencing | Native | Authorized cascading deletion creates a fresh instance ID on recreation; an authorized waiter receives `ActorDestroyed` when that incarnation disappears. | -| Result recovery and sync timeout diagnostics | Native | Status, result, and wait reauthorize the stored operation; terminal failure raises structured `MessageFailed`; whole-call adapter deadlines distinguish enqueue, wait, database, activation, and mailbox blockers. | -| Result lookup by request ID and idempotency key | Native | `runtime.findBy({ requestId })` and `reference.findBy({ idempotencyKey })` rebuild a `MessageReference`, authorized with the hook the original call ran. An actor remembers the keys of its own finished turns, so a key lookup separates a pruned message from one that never existed. | +| Capability | Status | TypeScript shape or remaining work | +| ------------------------------------------------------------------------------------------------------------- | ------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Actor registry, durable identity, JSON state, and adjacent state migrations | Native | Ordinary classes, static actor types, inferred state, explicit migrations, and isolated runtime context across every actor-instance callback. | +| Fluent committed calls and background delivery | Native | `await reference.operation()` and `reference.send.operation()`. | +| Ordered mailbox, sequence allocation, idempotency, retries, dead letters, leases, renewal, and fenced commits | Native | Relational ready/claimed membership tables, distinct generated request IDs and caller idempotency keys, durable history, adapter-appropriate sequence locking, and PostgreSQL/MySQL row locks held from fence validation through commit. | +| Domain rejection and strict poison ordering | Native | Rejections accept JavaScript identifier-style codes and roll back without retry; invalid codes fail terminally, while retryable failures block later operations until completion or dead-lettering. | +| Bounded activation passes and hot-actor fairness | Native | Configurable turn-count and elapsed-time budgets bound each pass, then move only that actor's already-due memberships behind actors already waiting. | +| Bounded claim candidate scan | Native | A configurable ordered scan continues to another ready actor when a worker loses the first candidate's lease race. | +| Backpressure and payload caps | Partial | Serialization enforces a shared maximum JSON nesting depth, raising `InvalidPayload`, and an optional caller-supplied `maxBytes` limit, raising `PayloadTooLarge`; reminder names are bounded to 255 characters. Distributed per-actor rate limits, global admission control, and cache-capacity eviction are not planned in either runtime. They are hot, request-path, and loss-tolerant, so one durable ordered message per check is the wrong shape. Solid Objects Pro answers them with grouped and ephemeral operations, which [fit](fit.md) describes. | +| Idle activation cache | Native | Long-running workers retain hydrated actors under renewable fenced leases, restore public state after failed turns, and release on timeout, fairness yield, lease loss, or shutdown. | +| Transactional effects and outcome operations | Native | At-least-once handlers receive immutable stable effect, attempt, source-message, and actor identity; success and failure operations also receive the originally staged arguments for correlation. Typed callback envelopes are exported. | +| Actor-to-actor delivery | Native | `sendTo(reference).operation()` stages delivery in the source actor commit. | +| One-shot and recurring reminders | Native | Scheduling, replacement events, catch-up policy, stale-claim recovery, pausing, authorized inspection, and idempotent resume are implemented. | +| Same-database commit actions | Native | Registered actions receive source-message identity, mailbox sequence, activation generation, and the fenced transaction connection. | +| Ambient transaction rejection | Native | Committed calls and message waits fail before blocking when the current async context already owns a transaction on the Solid Objects adapter. | +| Direct application-write isolation during actor code | Partial | `guardApplicationDatabase()` fails closed for operations, projections, migrations, and commit actions; only the supplied fenced commit-action connection may write. Unwrapped clients cannot be intercepted. | +| Committed snapshots | Native | `snapshot()` returns authorized persisted fields and inferred getters from one read-only committed state image; realtime replay reads explicit observables without mailbox history. | +| Actor destruction and incarnation fencing | Native | Authorized cascading deletion creates a fresh instance ID on recreation; an authorized waiter receives `ActorDestroyed` when that incarnation disappears. | +| Result recovery and sync timeout diagnostics | Native | Status, result, and wait reauthorize the stored operation; terminal failure raises structured `MessageFailed`; whole-call adapter deadlines distinguish enqueue, wait, database, activation, and mailbox blockers. | +| Result lookup by request ID and idempotency key | Native | `runtime.findBy({ requestId })` and `reference.findBy({ idempotencyKey })` rebuild a `MessageReference`, authorized with the hook the original call ran. An actor remembers the keys of its own finished turns, so a key lookup separates a pruned message from one that never existed. | Effect callback envelopes are typed with `EffectFailurePayload`, `EffectSuccessPayload`, and `SerializedError` in both SQL and Cloudflare. From 32935489a6442aeb3ef3318f33e17a68cfe5e0ef Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Wed, 23 Sep 2026 13:26:48 -0700 Subject: [PATCH 11/14] fix: authorize pruned keys with stored arguments Retain original arguments within the existing byte bound and deny\nlegacy entries that cannot reproduce authorization. Remove the\nlookup reference cast and cover SQL and Durable Object policies. See #57 --- docs/api.md | 8 +++++-- docs/parity.md | 10 ++++++--- src/cloudflare/engine.ts | 27 ++++++++++++----------- src/records.ts | 1 + src/reference.ts | 2 +- src/repository.ts | 12 +++++++---- src/runtime.ts | 2 +- test/cloudflare/recovery.test.ts | 20 +++++++++++++++++ test/cloudflare/worker.ts | 4 +++- test/result-lookup.test.ts | 37 +++++++++++++++++++++++++++++++- 10 files changed, 97 insertions(+), 26 deletions(-) diff --git a/docs/api.md b/docs/api.md index 5d1ab9e..3b64938 100644 --- a/docs/api.md +++ b/docs/api.md @@ -562,8 +562,8 @@ class and result types are also exported for integration typing. `retainedIdempotencyKeys` finished turns, so `reference.findBy({ idempotencyKey })` throws `MessagePruned` for a key the actor remembers and whose message retention removed, and returns `undefined` for a key no caller - ever sent. An actor remembers the operation beside each key, so the pruned - answer runs the same hook against the same operation a lookup of the + ever sent. An actor remembers the operation and original arguments beside each key, so the pruned + answer runs the same hook against the same operation and arguments a lookup of the surviving row would, and a caller the policy refuses reads `undefined` for both. `runtime.findBy({ requestId })` returns `undefined` in both cases, because the runtime generates a request id and no @@ -573,6 +573,10 @@ idempotencyKey })` throws `MessagePruned` for a key the actor remembers and row. An actor drops its oldest keys until the list fits, so a key long enough to fill the limit by itself is never remembered and its lookup answers `undefined` rather than throwing. +- Remembered arguments count toward the serialized memory limit and remain until + the entry is evicted or the instance is removed. Entries from older versions + that lack arguments return absence after pruning because their original + authorization cannot be reproduced. - `messageReference.outcome()` returns an `Outcome`: the status, the result, an `ErrorRecord` for a dead message, a `RejectionRecord` for a rejected one, and the attempt count. diff --git a/docs/parity.md b/docs/parity.md index 46f3a47..8a93178 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -165,14 +165,18 @@ An actor remembers the idempotency keys of its own last `retained_idempotency_keys` / `retainedIdempotencyKeys` finished turns, written in the instance row the executor updates anyway, so a key lookup raises `MessagePruned` for a message retention removed and answers absent for a -message that never existed. Each remembered key carries its operation, so the +message that never existed. Each remembered key carries its operation and original arguments, so the pruned answer runs the same authorization a lookup of the surviving row would. Both also bound the serialized memory, because an idempotency key has no length limit and the memory outlives the message row; an -actor drops its oldest keys until the list fits. The memory is actor state, so the query hook gates -the pruned answer in both. A request id lookup answers absent in both cases, +actor drops its oldest keys until the list fits. The original operation selects the message or query authorization hook in both. A request id lookup answers absent in both cases, because the runtime generates a request id and no actor remembers one. +Remembered arguments count toward the serialized memory limit and remain until +the entry is evicted or the instance is removed. Entries from older versions +that lack arguments return absence after pruning because their original +authorization cannot be reproduced. + The Durable Objects backend answers a key lookup and a request id lookup through the actor that holds the row, and it remembers keys in the same instance record. It cannot answer `findBy({ requestId })` without a reference, diff --git a/src/cloudflare/engine.ts b/src/cloudflare/engine.ts index eb88f9d..52be03b 100644 --- a/src/cloudflare/engine.ts +++ b/src/cloudflare/engine.ts @@ -353,19 +353,21 @@ export class ActorEngine { const remembered = (this.store.instance()?.completedIdempotencyKeys ?? []).find( (entry) => entry.key === String(key), ) - if (!remembered) return null - if (!(await this.authorizedOperation(input, remembered.operation))) return null - - return { pruned: true } - } - - private async authorizedOperation(input: HostRequest, operation: string): Promise { + if (!remembered || !remembered.arguments) return null const definition = this.definition(input.actorType) - const query = definition.queries.includes(operation) - if (!query && !definition.operations.includes(operation)) return false - + const query = definition.queries.includes(remembered.operation) + if (!query && !definition.operations.includes(remembered.operation)) return null const authorize = query ? this.settings.authorizeQuery : this.settings.authorizeMessage - return authorize({ ...input, operation, arguments: {} }) + if ( + !(await authorize({ + ...input, + operation: remembered.operation, + arguments: remembered.arguments, + })) + ) + return null + + return { pruned: true } } private rememberKey(instance: Instance, message: Message): void { @@ -373,8 +375,7 @@ export class ActorEngine { if (key === null) return const remembered = instance.completedIdempotencyKeys ?? [] - const entry = { key, operation: message.operation } - if (remembered.at(-1)?.key === key && remembered.at(-1)?.operation === entry.operation) return + const entry = { key, operation: message.operation, arguments: message.arguments } instance.completedIdempotencyKeys = boundedKeys({ keys: [...remembered.filter((value) => value.key !== key), entry], diff --git a/src/records.ts b/src/records.ts index eac788b..132f4c3 100644 --- a/src/records.ts +++ b/src/records.ts @@ -151,4 +151,5 @@ export interface BroadcastRow { export interface RememberedKey { key: string operation: string + arguments: JsonObject } diff --git a/src/reference.ts b/src/reference.ts index ad8202e..03ae4de 100644 --- a/src/reference.ts +++ b/src/reference.ts @@ -256,7 +256,7 @@ export class ActorReferenceCore { options: { idempotencyKey: string } & SnapshotOptions, ): Promise { return this.runtime.findBy({ - reference: this as unknown as ActorReferenceCore, + reference: this, idempotencyKey: options.idempotencyKey, authorizationContext: options.authorizationContext, }) diff --git a/src/repository.ts b/src/repository.ts index c582a04..08cf972 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -2295,9 +2295,10 @@ export class Repository { if (key === null) return stored const remembered = rememberedList(stored) - const entry = { key, operation: turn.message.operation } - if (remembered.at(-1)?.key === key && remembered.at(-1)?.operation === entry.operation) { - return stored + const entry = { + key, + operation: turn.message.operation, + arguments: jsonObject(JSON.parse(turn.message.arguments)), } return JSON.stringify( @@ -2461,7 +2462,10 @@ export function rememberedList(stored: string | null): RememberedKey[] { value !== null && typeof value === "object" && typeof value.key === "string" && - typeof value.operation === "string", + typeof value.operation === "string" && + value.arguments !== null && + typeof value.arguments === "object" && + !Array.isArray(value.arguments), ) } diff --git a/src/runtime.ts b/src/runtime.ts index 1761397..e0ccbe4 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -1757,7 +1757,7 @@ export class SolidObjectsRuntime { actorType: reference.actorType, actorId: reference.actorId, operation: remembered.operation, - argumentsValue: {}, + argumentsValue: remembered.arguments, authorizationContext: input.authorizationContext, }) if (!readable) return undefined diff --git a/test/cloudflare/recovery.test.ts b/test/cloudflare/recovery.test.ts index 7c926d4..970ee62 100644 --- a/test/cloudflare/recovery.test.ts +++ b/test/cloudflare/recovery.test.ts @@ -355,6 +355,26 @@ describe("Cloudflare recovery and fencing", () => { ).rejects.toMatchObject({ name: "UnsupportedCapability" }) }) + it("authorizes pruned keys with the original arguments", async () => { + const reference = runtime().ref(Counter, "protected-key") + await reference + .with({ authorizationContext, idempotencyKey: "protected" }) + .increment({ amount: 7 }) + await runInDurableObject(stub("protected-key"), (_object, state) => { + state.storage.sql.exec("DELETE FROM messages") + state.storage.sql.exec("DELETE FROM receipts") + }) + expect( + await reference.findBy({ + idempotencyKey: "protected", + authorizationContext: "argument-denied", + }), + ).toBeUndefined() + await expect( + reference.findBy({ idempotencyKey: "protected", authorizationContext }), + ).rejects.toMatchObject({ name: "MessagePruned" }) + }) + it("bounds what an instance remembers", async () => { const reference = runtime().ref(Counter, "bounded-keys") for (let index = 0; index < 5; index += 1) { diff --git a/test/cloudflare/worker.ts b/test/cloudflare/worker.ts index 906b6fa..4d4c29f 100644 --- a/test/cloudflare/worker.ts +++ b/test/cloudflare/worker.ts @@ -235,7 +235,9 @@ export class Actors extends createDurableObjectsHost({ }, sessions: environment.SESSIONS, }), - authorizeMessage: (input) => input.authorizationContext === "allowed", + authorizeMessage: (input) => + input.authorizationContext === "allowed" || + (input.authorizationContext === "argument-denied" && input.arguments.amount !== 7), authorizeQuery: (input) => input.authorizationContext === "allowed" && input.operation !== "denied", authorizeDestroy: (input) => input.authorizationContext === "allowed", diff --git a/test/result-lookup.test.ts b/test/result-lookup.test.ts index 5e42d2c..e1d648e 100644 --- a/test/result-lookup.test.ts +++ b/test/result-lookup.test.ts @@ -329,6 +329,41 @@ describe("result lookup", () => { ).toBeUndefined() }) + it("authorizes pruned keys with the original arguments", async () => { + const active = await start() + const reference = active.ref(CartActor, "alice") + await reference.send.with({ idempotencyKey: "protected" }).checkout({ orderId: 1 }) + await active.worker().runUntilIdle() + await deleteMessages(active) + const refusing = configure({ + database: active.settings.database, + authorizeMessage: ({ arguments: args }) => args.orderId !== 1, + }) + refusing.register(CartActor) + + expect( + await refusing.ref(CartActor, "alice").findBy({ idempotencyKey: "protected" }), + ).toBeUndefined() + seenOperations = [] + await expect(reference.findBy({ idempotencyKey: "protected" })).rejects.toThrow(MessagePruned) + expect(seenOperations).toEqual([{ operation: "checkout", argumentsValue: { orderId: 1 } }]) + }) + + it("does not disclose legacy keys without authorization arguments", async () => { + const active = await start() + const reference = active.ref(CartActor, "alice") + await reference.send.with({ idempotencyKey: "legacy" }).checkout({ orderId: 1 }) + await active.worker().runUntilIdle() + await deleteMessages(active) + await active.settings.database.transaction((connection) => + connection.run( + `UPDATE ${active.repository.table("instances")} SET completed_idempotency_keys = ?`, + [JSON.stringify([{ key: "legacy", operation: "checkout" }])], + ), + ) + expect(await reference.findBy({ idempotencyKey: "legacy" })).toBeUndefined() + }) + it("remembers a key whose message was rejected", async () => { const active = await start() const reference = active.ref(CartActor, "alice") @@ -390,7 +425,7 @@ describe("result lookup", () => { const remembered = await rememberedKeys(active) - expect(remembered).toEqual(keys.slice(-2)) + expect(remembered).toEqual(keys.slice(-1)) }) it("remembers nothing for a key larger than what it retains", async () => { From 0fbdec6fb89d3ba5afdc538599a1086df307fca3 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Wed, 23 Sep 2026 13:32:27 -0700 Subject: [PATCH 12/14] refactor: flatten Cloudflare lookup guards Handle missing lookups and invalid references with guard clauses, as requested in the PR review. The full 59-test Cloudflare suite and Cloudflare type check pass. See #57 --- src/cloudflare/engine.ts | 23 ++++++++++------------- 1 file changed, 10 insertions(+), 13 deletions(-) diff --git a/src/cloudflare/engine.ts b/src/cloudflare/engine.ts index 52be03b..d582c1a 100644 --- a/src/cloudflare/engine.ts +++ b/src/cloudflare/engine.ts @@ -312,19 +312,16 @@ export class ActorEngine { } private async readMessage(input: HostRequest): Promise { - let message: Message | undefined - if (input.method === "lookup") { - message = this.lookUp(input) - if (!message) return this.prunedReply(input) - } else { - message = this.store.message(String(input.payload.id)) - if ( - !message || - message.requestId !== input.payload.requestId || - message.sequence !== input.payload.sequence - ) - throw new Unauthorized("message reference is not authorized") - } + const lookup = input.method === "lookup" + const message = lookup ? this.lookUp(input) : this.store.message(String(input.payload.id)) + if (lookup && !message) return this.prunedReply(input) + if ( + !message || + (!lookup && + (message.requestId !== input.payload.requestId || + message.sequence !== input.payload.sequence)) + ) + throw new Unauthorized("message reference is not authorized") await this.authorizeOperation(input, message) this.bind(input) if (message.incarnation !== this.store.instance()?.incarnation) From 34d9ea42db78f56e0561cf83f52da94cf4253ab5 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Wed, 23 Sep 2026 13:59:55 -0700 Subject: [PATCH 13/14] fix: freeze every value an outcome reports A parity audit against the Ruby branch found the SQL runtime handing out mutable values where Ruby and the Durable Objects runtime both freeze. `messageOutcome` built its result with `normalizeJson`, which copies but does not freeze, while `messageResult` beside it, the Cloudflare path, and Ruby all use a read-only copy. A caller could mutate the nested hashes and arrays of a durable result, and the error and rejection records were not frozen either. Greptile reported this shape on the Ruby branch and it was fixed there; the same defect on this side went unreported. Without the fix the new test reads `expected false to be true`. Two tests close gaps the audit found in this file: a frozen result, and a message whose actor type this process no longer registers, which already answered absent but nothing held it there. The error record differs between the runtimes and the parity document did not say so. Ruby carries `class_name`, `message`, and `backtrace`; this runtime persists only `name` and `message`, because `safeError` has never stored a stack. Validation: pnpm test (528), pnpm run test:cloudflare (59), check, build, format:check. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 5 ++++- docs/parity.md | 8 ++++++-- src/runtime.ts | 9 ++++++--- test/result-lookup.test.ts | 22 ++++++++++++++++++++++ 4 files changed, 38 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4190fb7..a12e5d6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,7 +14,10 @@ request id exists. - Add `messageReference.outcome()`, which reports the status, the result, the persisted error, the rejection, and the attempt count, so a terminal failure - answers as well as a success. + answers as well as a success. Every value is frozen, so a caller cannot mutate + a durable result it read. The error carries `name` and `message`; this runtime + has never persisted a stack, which [parity](docs/parity.md) now records + against Ruby's `backtrace`. - Add schema version 12: a unique index on `messages.request_id`. The table had only `UNIQUE (actor_type, actor_id, request_id)`, which cannot serve a lookup that names the request id alone. diff --git a/docs/parity.md b/docs/parity.md index 8a93178..176c1fa 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -152,13 +152,17 @@ with the hook the original call ran, and report absence, an unregistered actor, and a refusal the same way. `outcome` reports the status, result, error, rejection, and attempt count in both. -Two details differ, and both come from the runtimes rather than the feature. +Three details differ, and all come from the runtimes rather than the feature. This runtime stores a result for every completed message, so a lookup answers one for asynchronous work; Ruby stores a result only for `sync` delivery, so a lookup there answers the status and the error but not the result. This runtime needed a new unique index on `request_id`, added as schema version 12, because its table constrained the pair `(actor_type, actor_id, request_id)`; the Ruby -schema has carried a global unique index since its first migration. +schema has carried a global unique index since its first migration. The error +record also differs: Ruby's `ErrorRecord` carries `class_name`, `message`, and +`backtrace`, while this runtime persists only `name` and `message`, because +`safeError` has never stored a stack. An `outcome` here reports what failed, not +where. Both runtimes tell a pruned message from one that never existed the same way. An actor remembers the idempotency keys of its own last diff --git a/src/runtime.ts b/src/runtime.ts index e0ccbe4..935f340 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -739,12 +739,15 @@ export class SolidObjectsRuntime { result: snapshot.result === null ? undefined - : (normalizeJson(JSON.parse(snapshot.result)) as DeepReadonly), - error: snapshot.error === null ? undefined : (JSON.parse(snapshot.error) as ErrorRecord), + : (readonlyCopy(normalizeJson(JSON.parse(snapshot.result))) as DeepReadonly), + error: + snapshot.error === null + ? undefined + : readonlyCopy(JSON.parse(snapshot.error) as ErrorRecord), rejection: snapshot.rejection === null ? undefined - : (JSON.parse(snapshot.rejection) as RejectionRecord), + : readonlyCopy(JSON.parse(snapshot.rejection) as RejectionRecord), attempts: Number(snapshot.attempt_count), }) } diff --git a/test/result-lookup.test.ts b/test/result-lookup.test.ts index e1d648e..e412ec0 100644 --- a/test/result-lookup.test.ts +++ b/test/result-lookup.test.ts @@ -491,6 +491,28 @@ describe("result lookup", () => { expect(outcome.result).toEqual({ orderId: 4210 }) }) + it("hands out a frozen result", async () => { + const active = await start() + const original = await active.ref(CartActor, "alice").send.checkout({ orderId: 9 }) + await active.worker().runUntilIdle() + + const outcome = await (await active.findBy({ requestId: original.requestId }))!.outcome() + + expect(Object.isFrozen(outcome.result)).toBe(true) + }) + + it("answers undefined for a message whose actor type is not registered", async () => { + const active = await start() + const original = await active.ref(CartActor, "alice").send.checkout({ orderId: 1 }) + await active.settings.database.transaction((connection) => + connection.run(`UPDATE ${active.repository.table("messages")} SET actor_type = ?`, [ + "RetiredCartActor", + ]), + ) + + expect(await active.findBy({ requestId: original.requestId })).toBeUndefined() + }) + it("propagates an authorization failure rather than reporting absence", async () => { const active = await start() const original = await active.ref(CartActor, "alice").send.checkout({ orderId: 1 }) From 2c9dd1304f1feb14813eacfdaa95fb24640a3ef4 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Wed, 23 Sep 2026 15:21:17 -0700 Subject: [PATCH 14/14] chore: prepare version 0.16.0 Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 2 +- package.json | 2 +- src/version.ts | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a12e5d6..01dc270 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,6 @@ # Changelog -## Unreleased +## 0.16.0 - 2026-09-23 - Find a message whose reference a caller lost. `runtime.findBy({ requestId })` answers a request id, which is unique across the table, and diff --git a/package.json b/package.json index 0527e08..8d1487a 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "solid-objects", - "version": "0.15.2", + "version": "0.16.0", "description": "Race-free realtime state per application identity, backed by your SQL database", "type": "module", "license": "MIT", diff --git a/src/version.ts b/src/version.ts index e25e508..3e69cf1 100644 --- a/src/version.ts +++ b/src/version.ts @@ -1 +1 @@ -export const VERSION = "0.15.2" +export const VERSION = "0.16.0"