diff --git a/CHANGELOG.md b/CHANGELOG.md index 0618aae..01dc270 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,58 @@ # 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 + `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. 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. +- 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 + 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 + 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 + `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. 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. - 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/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/api.md b/docs/api.md index c47a312..3b64938 100644 --- a/docs/api.md +++ b/docs/api.md @@ -550,6 +550,36 @@ 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`. +- 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. 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 + 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. +- 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. - `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 @@ -700,8 +730,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..d74b0af 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -87,6 +87,8 @@ 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. | +| `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/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 bbe13cc..176c1fa 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 | Planned | This is also an open Ruby roadmap item and will be implemented in both runtimes when its authorization shape is settled. | +| 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. @@ -144,6 +144,49 @@ 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. + +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. 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 +`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 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 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, +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/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/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/cloudflare/configuration.ts b/src/cloudflare/configuration.ts index 6349a0a..65268aa 100644 --- a/src/cloudflare/configuration.ts +++ b/src/cloudflare/configuration.ts @@ -26,6 +26,8 @@ export type CloudflareConfiguration = Pick< | "retryDelayMilliseconds" | "messageRetentionMilliseconds" | "pruneBatchSize" + | "retainedIdempotencyKeys" + | "retainedIdempotencyKeysBytes" > & { backend: DurableObjectsBackend effects?: Readonly> @@ -53,6 +55,8 @@ 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, + retainedIdempotencyKeysBytes: configuration.retainedIdempotencyKeysBytes ?? 16_384, } for (const name of [ "maxAttempts", @@ -64,6 +68,8 @@ export function buildCloudflareSettings(configuration: CloudflareConfiguration) "maxActivationDurationMilliseconds", "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 0fbd723..d582c1a 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" @@ -311,33 +312,16 @@ 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") - 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 - } 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) @@ -345,6 +329,58 @@ 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 async prunedReply(input: HostRequest): Promise { + const key = input.payload.idempotencyKey + if (key === undefined) return null + const remembered = (this.store.instance()?.completedIdempotencyKeys ?? []).find( + (entry) => entry.key === String(key), + ) + if (!remembered || !remembered.arguments) return null + const definition = this.definition(input.actorType) + 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 + if ( + !(await authorize({ + ...input, + operation: remembered.operation, + arguments: remembered.arguments, + })) + ) + return null + + return { pruned: true } + } + + private rememberKey(instance: Instance, message: Message): void { + const key = message.idempotencyKey + if (key === null) return + + const remembered = instance.completedIdempotencyKeys ?? [] + const entry = { key, operation: message.operation, arguments: message.arguments } + + instance.completedIdempotencyKeys = boundedKeys({ + keys: [...remembered.filter((value) => value.key !== key), entry], + count: this.settings.retainedIdempotencyKeys, + bytes: this.settings.retainedIdempotencyKeysBytes, + }) + } + private committed(identity: ActorIdentity) { const definition = this.definition(identity.actorType) const instance = this.store.instance() @@ -589,6 +625,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 +659,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 +673,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) } @@ -647,6 +687,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 8c5c944..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,6 +12,7 @@ export interface Instance extends ActorIdentity { stateVersion: number createdAt: number paused: boolean + completedIdempotencyKeys?: RememberedKey[] } export interface Message { diff --git a/src/cloudflare/runtime.ts b/src/cloudflare/runtime.ts index c046a03..bebabd3 100644 --- a/src/cloudflare/runtime.ts +++ b/src/cloudflare/runtime.ts @@ -8,8 +8,10 @@ import { ActorSetupFailed, EnqueueOutcomeUnknown, MessageFailed, + MessagePruned, Rejected, SyncTimeout, + Unauthorized, UnsupportedCapability, } from "../errors.js" import { @@ -19,6 +21,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 +130,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 }, + }).catch((error) => { + if (error instanceof Unauthorized) return null + throw error + }) + 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 : errorRecord(jsonObject(record.error)), + rejection: + record.rejection === null ? undefined : rejectionRecord(jsonObject(record.rejection)), + attempts: Number(record.attempt), + }) + } + async messageStatus( message: MessageReference, options: SnapshotOptions = {}, @@ -432,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/configuration.ts b/src/configuration.ts index d850ee5..70b860b 100644 --- a/src/configuration.ts +++ b/src/configuration.ts @@ -80,6 +80,8 @@ export interface SolidObjectsConfiguration { instanceRetentionByActorType?: Readonly> processRetentionMilliseconds?: number pruneBatchSize?: number + retainedIdempotencyKeys?: number + retainedIdempotencyKeysBytes?: number logger?: Logger authorizeMessage?: (input: AuthorizationInput) => boolean | Promise authorizeQuery?: (input: AuthorizationInput) => boolean | Promise @@ -168,6 +170,8 @@ 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: @@ -315,6 +319,18 @@ 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") + } + 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/doctor.ts b/src/doctor.ts index 9558b47..1ed7927 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" @@ -50,6 +51,7 @@ const EXPECTED_COLUMNS: Readonly> = { "activation_expires_at_ms", "activation_generation", "paused", + "completed_idempotency_keys", ], messages: [ "id", @@ -243,11 +245,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(",") !== SCHEMA_VERSIONS.join(",")) { 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 ${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 9aee02e..c5029d2 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, @@ -193,6 +194,7 @@ export { LostActivation, MailboxFull, MessageFailed, + MessagePruned, NonRetryableError, PayloadTooLarge, QueryMutatedState, diff --git a/src/outcome.ts b/src/outcome.ts new file mode 100644 index 0000000..9541464 --- /dev/null +++ b/src/outcome.ts @@ -0,0 +1,20 @@ +import type { DeepReadonly, JsonObject, JsonValue, 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/records.ts b/src/records.ts index 2ae81d7..132f4c3 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 { @@ -146,3 +147,9 @@ export interface BroadcastRow { claimed_by: string | null error: string | null } + +export interface RememberedKey { + key: string + operation: string + arguments: JsonObject +} diff --git a/src/reference.ts b/src/reference.ts index bd2de94..03ae4de 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, + 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..08cf972 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -20,11 +20,12 @@ import type { EffectRow, EnqueueInput, InstanceRow, + RememberedKey, MessageRow, 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, @@ -712,11 +713,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 +942,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 +974,7 @@ export class Repository { now, ], ) + await this.rememberKeys(connection, turn) await this.releaseClaim({ connection, turn }) return "dead" as const } @@ -999,6 +1003,45 @@ 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 rememberedIdempotencyKey(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).find( + (entry) => entry.key === input.idempotencyKey, + ) + } + + 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 }> { @@ -1020,36 +1063,48 @@ export class Repository { }) } - async messageStatus( + async messageWithStatus( id: string, - ): Promise<"ready" | "claimed" | "completed" | "rejected" | "dead" | "unknown"> { + ): 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 "unknown" - 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], - ) - return ready ? "ready" : "unknown" + if (!message) return undefined + return { message, status: await this.statusOf({ connection, message }) } }) } + async messageStatus(id: string): Promise { + return (await this.messageWithStatus(id))?.status ?? "unknown" + } + + private async statusOf(options: { + connection: DatabaseConnection + message: MessageRow + }): Promise { + const { connection, message } = options + 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 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 { return this.settings.database.connection(async (connection) => { const message = await connection.get( @@ -2218,6 +2273,43 @@ 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) + const entry = { + key, + operation: turn.message.operation, + arguments: jsonObject(JSON.parse(turn.message.arguments)), + } + + return JSON.stringify( + boundedKeys({ + keys: [...remembered.filter((value) => value.key !== key), entry], + count: this.settings.retainedIdempotencyKeys, + bytes: this.settings.retainedIdempotencyKeysBytes, + }), + ) + } + private async assertFence(connection: DatabaseConnection, turn: ClaimedTurn): Promise { const instance = await this.lockActivationFence(connection, turn.instance.id) const now = await connection.nowMilliseconds() @@ -2351,6 +2443,32 @@ function retentionPolicy(options: { } } +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 +} + +export function rememberedList(stored: string | null): RememberedKey[] { + if (stored === null) return [] + const parsed = JSON.parse(stored) as RememberedKey[] + if (!Array.isArray(parsed)) return [] + return parsed.filter( + (value) => + value !== null && + typeof value === "object" && + typeof value.key === "string" && + typeof value.operation === "string" && + value.arguments !== null && + typeof value.arguments === "object" && + !Array.isArray(value.arguments), + ) +} + function parameterList(length: number): string { return Array.from({ length }, () => "?").join(", ") } diff --git a/src/runtime.ts b/src/runtime.ts index 8aa3865..935f340 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" @@ -37,6 +38,7 @@ import { InvalidActor, LostActivation, MessageFailed, + MessagePruned, NonRetryableError, QueryMutatedState, Rejected, @@ -687,6 +689,69 @@ 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 await this.absentMessage(input) + 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, + ) + 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: found.status, + result: + snapshot.result === null + ? undefined + : (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 + : readonlyCopy(JSON.parse(snapshot.rejection) as RejectionRecord), + attempts: Number(snapshot.attempt_count), + }) + } + async messageStatus( messageReference: MessageReference, options: SnapshotOptions = {}, @@ -1658,6 +1723,121 @@ 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!, + }) + } + + // 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.rememberedIdempotencyKey({ + actorType: reference.actorType, + actorId: reference.actorId, + idempotencyKey: input.idempotencyKey, + }) + if (!remembered) return undefined + const readable = await this.readableOperation({ + actorType: reference.actorType, + actorId: reference.actorId, + operation: remembered.operation, + argumentsValue: remembered.arguments, + authorizationContext: input.authorizationContext, + }) + if (!readable) return undefined + + throw new MessagePruned(input.idempotencyKey) + } + + 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: 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 || error instanceof UnknownActorType) return false + throw error + } + } + + // 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 (error) { + if (error instanceof Unauthorized || error instanceof UnknownActorType) return false + throw error + } + } + private messageReferenceFromRow(message: MessageRow): MessageReference { return new MessageReference({ runtime: this, diff --git a/src/schema.ts b/src/schema.ts index 992af65..1ccafae 100644 --- a/src/schema.ts +++ b/src/schema.ts @@ -12,7 +12,13 @@ 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 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 @@ -355,6 +361,35 @@ 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(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/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" diff --git a/test/cloudflare/recovery.test.ts b/test/cloudflare/recovery.test.ts index e1009e6..970ee62 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) }) @@ -326,6 +331,91 @@ 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("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) { + 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?.map((entry) => entry.key) + }) + + 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(400) + 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 7afff63..4d4c29f 100644 --- a/test/cloudflare/worker.ts +++ b/test/cloudflare/worker.ts @@ -235,13 +235,17 @@ 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", authorizeSubscription: (input) => input.authorizationContext === "allowed", authorizeAdministration: (input) => input.authorizationContext === "allowed", retryDelayMilliseconds: () => 10, + retainedIdempotencyKeys: 3, + retainedIdempotencyKeysBytes: 256, 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 f3e1be2..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, - ]) + 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, - ]) + 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 cf5fde9..2cfa9aa 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] }, + details: { versions: [...SCHEMA_VERSIONS] }, }) expect(check(report, "authorization").status).toBe("pass") expect(check(report, "database").status).toBe("pass") @@ -145,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/instance-retention.test.ts b/test/instance-retention.test.ts index 96195ee..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, - ]) + 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 new file mode 100644 index 0000000..e412ec0 --- /dev/null +++ b/test/result-lookup.test.ts @@ -0,0 +1,566 @@ +import { afterEach, describe, expect, it } from "vitest" +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" + 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: JsonObject }[] = [] +let hooks: string[] = [] + +afterEach(async () => { + await runtime?.close() + runtime = undefined + CartActor.fail = false + seenOperations = [] + hooks = [] +}) + +async function start( + overrides: { retainedIdempotencyKeys?: number; retainedIdempotencyKeysBytes?: number } = {}, +): Promise { + const created = configure({ + database: sqlite({ path: ":memory:" }), + maxAttempts: 1, + ...overrides, + 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("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("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("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") + 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 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("bounds what an instance remembers by size", async () => { + 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()) { + await reference.send.with({ idempotencyKey: key }).checkout({ orderId: index }) + } + await active.worker().runUntilIdle() + + const remembered = await rememberedKeys(active) + + expect(remembered).toEqual(keys.slice(-1)) + }) + + 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") + 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 }) + await active.worker().runUntilIdle() + + expect(await rememberedKeys(active)).toEqual([]) + }) + + it("keeps request ids unique across the table", async () => { + const active = await start() + 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 = ?`, + [first.requestId, second.id], + ), + ), + ).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("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 }) + 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/, + ) + }) +}) + +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 +} + +async function deleteMessages(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")}`, + ), + ) + const remembered = JSON.parse(row?.completed_idempotency_keys ?? "[]") as { + key: string + operation: string + }[] + return remembered.map((entry) => entry.key) +} diff --git a/test/support/portable-runtime-contract.ts b/test/support/portable-runtime-contract.ts index 7763b4c..01da20b 100644 --- a/test/support/portable-runtime-contract.ts +++ b/test/support/portable-runtime-contract.ts @@ -49,6 +49,56 @@ 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("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" })