From 3473fc61d448742f66cf0beaacc61e1e85b7cf44 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 21 Sep 2026 18:08:49 -0700 Subject: [PATCH 01/10] feat: cancel actor reminders A recurring reminder could be started and never stopped. schedule() now returns a ReminderHandle, and unschedule() removes one alarm by operation, by operation and key, or by that handle. unscheduleAll() removes every key of one operation. Both cancellations are staged in the same list as the schedules, so they apply in the order the turn called them, commit with the state change that decided them, and cancel nothing when a turn throws. Cancellation works on both engines. The SQL repository deletes by the composed name, or by message_operation when every key goes. The Durable Objects store deletes by reminder name, and reads its rows to find the keys of one operation, which is cheap because one object holds one actor. The handle is a plain object, as emit already returns, so it serialises into actor state and still cancels after a deactivation. schedule() returned void before. transmit() keeps that contract through its own type, and the packaged consumer fixture now exercises the handle and both cancellations through the published surface. --- CHANGELOG.md | 13 ++ docs/api.md | 48 ++++- docs/correctness.md | 3 + src/actor.ts | 43 ++++- src/cloudflare/engine.ts | 5 + src/cloudflare/storage.ts | 11 ++ src/index.ts | 4 + src/reference.ts | 23 ++- src/repository.ts | 9 + src/types.ts | 2 + test/actor-operations.types.ts | 3 +- test/cloudflare/runtime.test.ts | 34 ++++ test/cloudflare/worker.ts | 18 ++ test/fixtures/actor-operations-consumer.mts | 7 +- test/reminder-cancellation.test.ts | 204 ++++++++++++++++++++ 15 files changed, 414 insertions(+), 13 deletions(-) create mode 100644 test/reminder-cancellation.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 237f7d5..b15dbab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,18 @@ # Changelog +## Unreleased + +- Add reminder cancellation. `unschedule()` removes one alarm by operation, by + operation and key, or by the handle `schedule()` now returns. + `unscheduleAll()` removes every key of one operation. Both stage an intent + beside the schedules, so they apply in the order the turn called them, commit + with the state change that decided them, and cancel nothing when a turn + throws. Cancellation works on the SQL backends and on Durable Objects. +- `schedule()` now returns a `ReminderHandle` (`{ name: string }`) instead of + `void`. A handle is a plain object, so it survives in actor state and still + cancels after a deactivation. Code that assigned the result to `void` needs + updating, as `emit` required in 0.15.0. + ## 0.15.2 - 2026-09-21 - Stop the deadlock between concurrent callers that create the same actor from diff --git a/docs/api.md b/docs/api.md index 7de0fe7..7e83405 100644 --- a/docs/api.md +++ b/docs/api.md @@ -68,9 +68,12 @@ createdAtMs }` shape returned by `SolidObjectsRuntime.snapshotWithIncarnation`. schedule that the reference methods use. `ActorIntents`, `EffectIntent`, `CommitActionIntent`, `ReminderIntent`, +`UnscheduleIntent`, `UnscheduleAllIntent`, `ReminderMutation`, `OutboundMessageIntent`, `ReminderOptions`, `OutboundMessageOptions`, `PayloadBroadcasts`, and `PayloadBroadcastValue` describe actor-declared -transactional work and typed personalized projections. +transactional work and typed personalized projections. `ReminderMutation` is the +union of one scheduled reminder and the two cancellations, held in one list so +they apply in the order the turn called them. `EffectFailurePayload`, `EffectSuccessPayload`, and `SerializedError` describe effect callback messages. They are also exported @@ -162,6 +165,49 @@ row per item. It also cannot strand an entry when the runtime coalesces an occurrence. Prefer it for a large queue of interchangeable items. Prefer `key` when one item needs an alarm that you can move on its own. +#### Cancelling a reminder + +`schedule()` returns a `ReminderHandle` (`{ name: string }`) naming the alarm it +armed. `unschedule()` cancels one alarm, by operation, by operation and key, or +by that handle. `unscheduleAll()` cancels every key of one operation. + +```typescript +class Subscription extends Actor { + chase: ReminderHandle | null = null + + convertToPaid(): void { + this.status = "active" + this.unschedule("trialExpired") + this.chase = this.schedule({ at: renewal, everyMilliseconds: MONTH }).chargeRenewal() + } + + cancelled(): void { + if (this.chase) this.unschedule(this.chase) + } + + shipped({ carrierId }: { carrierId: string }): void { + this.unschedule("chaseCarrier", { key: carrierId }) + } + + stopChasing(): void { + this.unscheduleAll("chaseCarrier") + } +} +``` + +A cancellation is staged like a schedule, so it commits with the state change +that decided it and a turn that throws cancels nothing. Both apply in the order +the turn called them, so cancelling and then scheduling the same name leaves it +armed at the new time. + +Cancelling an alarm that does not exist is not an error. `unschedule()` returns +nothing, because it stages an intent rather than applying one, and an answer +given at call time could be stale by the time the turn commits. + +A handle is a plain object, so it survives in actor state and still cancels +after a deactivation. Passing a handle together with `key` is a `TypeError`, +because the handle already names the key. + ### Recovering abandoned effects `emit` returns an `EffectHandle` (`{ id: string }`) on every successful call. Save diff --git a/docs/correctness.md b/docs/correctness.md index 17858a3..699fdeb 100644 --- a/docs/correctness.md +++ b/docs/correctness.md @@ -5,6 +5,9 @@ - Delivery is ordered per actor identity and at least once. - Different identities may execute concurrently. - Sequence allocation and durable enqueue are one transaction. +- A reminder can be cancelled. A cancellation commits with the state change that + decided it, and applies in the order the turn called it. An occurrence the + scheduler already claimed still runs; the cancellation removes later ones. - Concurrent callers that create the same actor produce one instance row and distinct sequences. The mailbox locks that row by its primary key, so MySQL does not upgrade a shared lock and the enqueue does not deadlock. diff --git a/src/actor.ts b/src/actor.ts index 336c66d..308e65f 100644 --- a/src/actor.ts +++ b/src/actor.ts @@ -9,6 +9,7 @@ import { createStagedOperations, type ActorReference, type ScheduledOperations, + type StagedOperationMap, type ScheduledOperationsFor, type StagedOperations, } from "./reference.js" @@ -17,6 +18,7 @@ import type { ActorIdentifier, EffectHandle, JsonObject, + ReminderHandle, JsonValue, MessageContext, } from "./types.js" @@ -108,6 +110,18 @@ export interface ReminderIntent { missedPolicy: "all" | "latest" } +export interface UnscheduleIntent { + cancel: "one" + name: string +} + +export interface UnscheduleAllIntent { + cancel: "all" + operation: string +} + +export type ReminderMutation = ReminderIntent | UnscheduleIntent | UnscheduleAllIntent + export interface OutboundMessageIntent { actorType: string actorId: string @@ -121,7 +135,7 @@ export interface ActorIntents { effects: EffectIntent[] effectRecoveries?: EffectRecoveryIntent[] commitActions: CommitActionIntent[] - reminders: ReminderIntent[] + reminders: ReminderMutation[] outboundMessages: OutboundMessageIntent[] } @@ -304,7 +318,7 @@ export abstract class Actor { transmit( this: Actor & Pick & (Partial | NoInfer), ): ScheduledOperationsFor> - transmit(): ScheduledOperations { + transmit(): StagedOperationMap { return createStagedOperationMap(this.#operations, (operation, argumentsValue) => { this.#intents.effects.push({ name: TRANSMIT_EFFECT, @@ -331,8 +345,9 @@ export abstract class Actor { const key = validatedReminderKey(options.key) return createStagedOperationMap(this.#operations, (operation, argumentsValue) => { + const name = reminderName(operation, key) this.#intents.reminders.push({ - name: reminderName(operation, key), + name, operation, atMilliseconds, arguments: jsonObject(argumentsValue), @@ -341,9 +356,31 @@ export abstract class Actor { ? {} : { intervalMilliseconds: options.everyMilliseconds }), }) + return { name } }) } + unschedule(operationOrHandle: string | ReminderHandle, options: { key?: string | number } = {}) { + if (typeof operationOrHandle === "object" && operationOrHandle !== null) { + if (options.key !== undefined) { + throw new TypeError("a reminder handle already names its key") + } + const name = (operationOrHandle as ReminderHandle).name + if (typeof name !== "string" || name.length === 0) { + throw new TypeError("unschedule requires a reminder handle returned by schedule") + } + this.#intents.reminders.push({ cancel: "one", name }) + return + } + + const key = validatedReminderKey(options.key) + this.#intents.reminders.push({ cancel: "one", name: reminderName(operationOrHandle, key) }) + } + + unscheduleAll(operation: string) { + this.#intents.reminders.push({ cancel: "all", operation }) + } + sendTo( reference: ActorReference, options: OutboundMessageOptions = {}, diff --git a/src/cloudflare/engine.ts b/src/cloudflare/engine.ts index 3cc562a..f166560 100644 --- a/src/cloudflare/engine.ts +++ b/src/cloudflare/engine.ts @@ -673,6 +673,11 @@ export class ActorEngine { }) } for (const intent of intents.reminders) { + if ("cancel" in intent) { + if (intent.cancel === "all") this.store.deleteRemindersFor(intent.operation) + else this.store.deleteReminder(intent.name) + continue + } this.store.saveReminder({ name: intent.name, generation: crypto.randomUUID(), diff --git a/src/cloudflare/storage.ts b/src/cloudflare/storage.ts index 31f23d7..6885d59 100644 --- a/src/cloudflare/storage.ts +++ b/src/cloudflare/storage.ts @@ -176,6 +176,17 @@ export class ActorStorage { ) } + deleteReminder(name: string): void { + this.storage.sql.exec("DELETE FROM reminders WHERE name = ?", name) + } + + deleteRemindersFor(operation: string): void { + const names = this.rows("SELECT record FROM reminders") + .filter((reminder) => reminder.operation === operation) + .map((reminder) => reminder.name) + for (const name of names) this.deleteReminder(name) + } + saveSubscription(subscription: Subscription): void { this.storage.sql.exec( "INSERT INTO subscriptions(id, expires_at, record) VALUES (?, ?, ?) ON CONFLICT(id) DO UPDATE SET expires_at = excluded.expires_at, record = excluded.record", diff --git a/src/index.ts b/src/index.ts index 792dc8b..077a789 100644 --- a/src/index.ts +++ b/src/index.ts @@ -15,6 +15,9 @@ export { type PayloadBroadcasts, type PayloadBroadcastValue, type ReminderIntent, + type ReminderMutation, + type UnscheduleAllIntent, + type UnscheduleIntent, type ReminderOptions, } from "./actor.js" export { @@ -127,6 +130,7 @@ export type { EffectContext, EffectFailurePayload, EffectHandle, + ReminderHandle, EffectSuccessPayload, InvocationOptions, JsonObject, diff --git a/src/reference.ts b/src/reference.ts index 6dced92..59e2015 100644 --- a/src/reference.ts +++ b/src/reference.ts @@ -10,6 +10,7 @@ import type { JsonValue, MessageStatus, SnapshotOptions, + ReminderHandle, } from "./types.js" type FunctionKeys = { @@ -70,6 +71,12 @@ type StagedMethod = Method extends (...argumentsValue: any[]) => any : (...argumentsValue: OperationArguments) => void : never +type ScheduledMethod = Method extends (...argumentsValue: any[]) => any + ? [OperationArguments] extends [never] + ? never + : (...argumentsValue: OperationArguments) => ReminderHandle + : never + type DirectMessages = { [Key in ActorOperationNames]: InvokedMethod } @@ -93,11 +100,15 @@ export type StagedOperations = { } export interface ScheduledOperations { - [operation: string]: (argumentsValue?: Record) => void + [operation: string]: (argumentsValue?: Record) => ReminderHandle } export type ScheduledOperationsFor = { - [Key in ActorOperationNames]: StagedMethod + [Key in ActorOperationNames]: ScheduledMethod +} + +export interface StagedOperationMap { + [operation: string]: (argumentsValue?: Record) => void } export type ActorReference = ActorReferenceCore & @@ -260,10 +271,10 @@ export function createStagedOperations( return createStagedOperationMap(operations, dispatch) as unknown as StagedOperations } -export function createStagedOperationMap( +export function createStagedOperationMap( operations: ReadonlySet, - dispatch: (operation: string, argumentsValue: JsonObject) => void, -): ScheduledOperations { + dispatch: (operation: string, argumentsValue: JsonObject) => Result, +): { [operation: string]: (argumentsValue?: Record) => Result } { return new Proxy( {}, { @@ -274,7 +285,7 @@ export function createStagedOperationMap( dispatch(property, operationArguments(argumentsValue)) }, }, - ) as ScheduledOperations + ) as { [operation: string]: (argumentsValue?: Record) => Result } } function createReferenceProxy( diff --git a/src/repository.ts b/src/repository.ts index 9d0611b..c4122c5 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -767,6 +767,15 @@ export class Repository { }) for (const reminder of input.intents.reminders) { + if ("cancel" in reminder) { + const column = reminder.cancel === "all" ? "message_operation" : "operation" + const value = reminder.cancel === "all" ? reminder.operation : reminder.name + await connection.run( + `DELETE FROM ${this.table("reminders")} WHERE instance_id = ? AND ${column} = ?`, + [turn.instance.id, value], + ) + continue + } const existing = await connection.get<{ id: string; run_at_ms: number | bigint }>( `SELECT id, run_at_ms FROM ${this.table("reminders")} WHERE instance_id = ? AND operation = ?`, diff --git a/src/types.ts b/src/types.ts index b74d217..1afa2c0 100644 --- a/src/types.ts +++ b/src/types.ts @@ -4,6 +4,8 @@ export type JsonObject = { [key: string]: JsonValue } export type EffectHandle = { readonly id: string } +export type ReminderHandle = { readonly name: string } + export type SerializedError = { name: string message: string diff --git a/test/actor-operations.types.ts b/test/actor-operations.types.ts index dbfe073..d8b3606 100644 --- a/test/actor-operations.types.ts +++ b/test/actor-operations.types.ts @@ -1,3 +1,4 @@ +import type { ReminderHandle } from "../src/types.js" import { expectTypeOf } from "vitest" import { Actor, @@ -41,7 +42,7 @@ export class ChatRun extends ParentRun { start() { const operations = this.schedule({ at: new Date(0), key: "watchdog" }) - expectTypeOf(operations.recoverIfStuck({ generation: 1 })).toEqualTypeOf() + expectTypeOf(operations.recoverIfStuck({ generation: 1 })).toEqualTypeOf() operations.finish() operations.optional() operations.optional({ generation: 1 }) diff --git a/test/cloudflare/runtime.test.ts b/test/cloudflare/runtime.test.ts index 8a798d2..de8d59b 100644 --- a/test/cloudflare/runtime.test.ts +++ b/test/cloudflare/runtime.test.ts @@ -58,6 +58,29 @@ describe("Durable Objects runtime", () => { expect(await runtime().ref(Counter, "destination").with({ authorizationContext }).count).toBe(1) }) + it("cancels reminders before the alarm fires", async () => { + const reference = runtime().ref(Counter, "cancelled").with({ authorizationContext }) + await reference.arm({ at: Date.now() + 60_000 }) + expect(await remainingReminders("cancelled")).toEqual(["increment"]) + + await reference.disarm() + expect(await remainingReminders("cancelled")).toEqual([]) + + const stub = env.ACTORS.getByName(JSON.stringify(["Counter", "cancelled"])) + for (let attempt = 0; attempt < 5; attempt += 1) await runDurableObjectAlarm(stub) + expect(await runtime().ref(Counter, "cancelled").with({ authorizationContext }).count).toBe(0) + }) + + it("cancels one keyed reminder and every key of an operation", async () => { + const reference = runtime().ref(Counter, "keyed-cancel").with({ authorizationContext }) + await reference.armKeyed({ at: Date.now() + 60_000, keys: ["a", "b", "c"] }) + await reference.disarmKey({ key: "b" }) + expect(await remainingReminders("keyed-cancel")).toEqual(["increment:a", "increment:c"]) + + await reference.disarmAll() + expect(await remainingReminders("keyed-cancel")).toEqual([]) + }) + it("retains a recoverable accepted message after caller timeout", async () => { const reference = runtime().ref(Counter, "delayed") const message = await reference.send @@ -84,3 +107,14 @@ describe("Durable Objects runtime", () => { ) }) }) + +async function remainingReminders(actorId: string): Promise { + const stub = env.ACTORS.getByName(JSON.stringify(["Counter", actorId])) + return runInDurableObject(stub, (instance: unknown) => { + const storage = (instance as { ctx: { storage: { sql: { exec: Function } } } }).ctx.storage + return storage.sql + .exec("SELECT name FROM reminders ORDER BY name") + .toArray() + .map((row: { name: string }) => row.name) + }) +} diff --git a/test/cloudflare/worker.ts b/test/cloudflare/worker.ts index c6c82f9..c92b0b1 100644 --- a/test/cloudflare/worker.ts +++ b/test/cloudflare/worker.ts @@ -67,6 +67,24 @@ export class Counter extends Actor { this.schedule({ at: new Date(options.at) }).increment() } + armKeyed(options: { at: number; keys: string[] }): void { + for (const key of options.keys) { + this.schedule({ at: new Date(options.at), key }).increment() + } + } + + disarm(): void { + this.unschedule("increment") + } + + disarmKey(options: { key: string }): void { + this.unschedule("increment", { key: options.key }) + } + + disarmAll(): void { + this.unscheduleAll("increment") + } + forward(options: { target: string }): void { this.sendTo(Counter.ref(options.target)).increment() } diff --git a/test/fixtures/actor-operations-consumer.mts b/test/fixtures/actor-operations-consumer.mts index c3cba74..d3256e2 100644 --- a/test/fixtures/actor-operations-consumer.mts +++ b/test/fixtures/actor-operations-consumer.mts @@ -3,6 +3,7 @@ import { type ActorReference, type ScheduledOperationsFor, type EffectOptions, + type ReminderHandle, } from "solid-objects" import type { ScheduledOperationsFor as CoreScheduledOperationsFor } from "solid-objects/core" @@ -26,8 +27,10 @@ export class ChatRun extends ParentRun { } start(): void { const operations = this.schedule({ at: new Date(0), key: "watchdog" }) - const result: void = operations.recoverIfStuck({ generation: 1 }) - void result + const handle: ReminderHandle = operations.recoverIfStuck({ generation: 1 }) + this.unschedule(handle) + this.unschedule("finish", { key: "watchdog" }) + this.unscheduleAll("finish") operations.finish() operations.optional() this.transmit().recoverIfStuck({ generation: 1 }) diff --git a/test/reminder-cancellation.test.ts b/test/reminder-cancellation.test.ts new file mode 100644 index 0000000..06b8497 --- /dev/null +++ b/test/reminder-cancellation.test.ts @@ -0,0 +1,204 @@ +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 type { ReminderHandle } from "../src/types.js" + +class Subscription extends Actor { + static override readonly actorType = "cancel-subscriptions" + + status = "trialing" + expirations = 0 + handle: ReminderHandle | null = null + + startTrial(): void { + this.handle = this.schedule({ at: new Date(Date.now() + 3_600_000) }).trialExpired() + } + + startRecurring(): void { + this.handle = this.schedule({ + at: new Date(Date.now() - 1_000), + everyMilliseconds: 60_000, + }).trialExpired() + } + + convertByName(): void { + this.status = "active" + this.unschedule("trialExpired") + } + + convertByHandle(): void { + this.status = "active" + if (this.handle) this.unschedule(this.handle) + } + + cancelThenReschedule(): void { + this.unschedule("trialExpired") + this.schedule({ at: new Date(Date.UTC(2031, 0, 1)) }).trialExpired() + } + + cancelThenFail(): void { + this.unschedule("trialExpired") + throw new Error("turn failed") + } + + cancelBadHandle(): void { + this.unschedule({ nope: "x" } as unknown as ReminderHandle) + } + + trialExpired(): void { + this.expirations += 1 + this.status = "expired" + } +} + +class Shipment extends Actor { + static override readonly actorType = "cancel-shipments" + + dispatch({ carrierIds }: { carrierIds: string[] }): void { + for (const id of carrierIds) { + this.schedule({ at: new Date(Date.now() + 3_600_000), key: id }).chaseCarrier({ + carrierId: id, + }) + } + this.schedule({ at: new Date(Date.now() + 3_600_000) }).audit() + } + + shipped({ carrierId }: { carrierId: string }): void { + this.unschedule("chaseCarrier", { key: carrierId }) + } + + stopChasing(): void { + this.unscheduleAll("chaseCarrier") + } + + chaseCarrier(_options: { carrierId: string }): void {} + audit(): void {} +} + +let runtime: SolidObjectsRuntime | undefined + +afterEach(async () => { + await runtime?.close() + runtime = undefined +}) + +async function start(): Promise { + runtime = configure({ + database: sqlite({ path: ":memory:" }), + authorizeMessage: () => true, + authorizeQuery: () => true, + pollingIntervalMilliseconds: 1, + syncPollingIntervalMilliseconds: 1, + maxAttempts: 1, + }) + runtime.register(Subscription) + runtime.register(Shipment) + await runtime.install() + return runtime +} + +async function reminderNames(started: SolidObjectsRuntime): Promise { + const rows = await started.settings.database.connection((connection) => + connection.all<{ operation: string }>( + `SELECT operation FROM solid_objects_reminders ORDER BY operation`, + ), + ) + return rows.map((row) => row.operation) +} + +describe("reminder cancellation", () => { + it("returns a handle naming the reminder", async () => { + const started = await start() + const reference = Subscription.ref("alice") + await reference.startTrial() + + expect(await reference.handle).toEqual({ name: "trialExpired" }) + expect(await reminderNames(started)).toEqual(["trialExpired"]) + }) + + it("cancels by name", async () => { + const started = await start() + const reference = Subscription.ref("alice") + await reference.startTrial() + await reference.convertByName() + + expect(await reminderNames(started)).toEqual([]) + expect(await reference.status).toBe("active") + }) + + it("cancels by handle", async () => { + const started = await start() + const reference = Subscription.ref("alice") + await reference.startTrial() + await reference.convertByHandle() + + expect(await reminderNames(started)).toEqual([]) + }) + + it("stops a recurring reminder", async () => { + const started = await start() + const reference = Subscription.ref("alice") + await reference.startRecurring() + expect(await started.reminderScheduler().runOnce()).toBe(1) + await reference.convertByName() + + expect(await reminderNames(started)).toEqual([]) + expect(await started.reminderScheduler().runOnce()).toBe(0) + }) + + it("cancelling an absent reminder is not an error", async () => { + const started = await start() + await Subscription.ref("alice").convertByName() + + expect(await reminderNames(started)).toEqual([]) + }) + + it("a failed turn cancels nothing", async () => { + const started = await start() + const reference = Subscription.ref("alice") + await reference.startTrial() + await expect(reference.cancelThenFail()).rejects.toThrow() + + expect(await reminderNames(started)).toEqual(["trialExpired"]) + }) + + it("cancel then schedule in one turn leaves the new time", async () => { + const started = await start() + const reference = Subscription.ref("alice") + await reference.startTrial() + await reference.cancelThenReschedule() + + const rows = await started.settings.database.connection((connection) => + connection.all<{ run_at_ms: number | bigint }>( + `SELECT run_at_ms FROM solid_objects_reminders`, + ), + ) + expect(rows).toHaveLength(1) + expect(Number(rows[0]!.run_at_ms)).toBe(Date.UTC(2031, 0, 1)) + }) + + it("rejects a malformed handle", async () => { + const started = await start() + await expect(Subscription.ref("alice").cancelBadHandle()).rejects.toThrow() + expect(await reminderNames(started)).toEqual([]) + }) + + it("cancels one key and leaves its siblings", async () => { + const started = await start() + const reference = Shipment.ref("truck") + await reference.dispatch({ carrierIds: ["a", "b", "c"] }) + await reference.shipped({ carrierId: "b" }) + + expect(await reminderNames(started)).toEqual(["audit", "chaseCarrier:a", "chaseCarrier:c"]) + }) + + it("cancels every key of one operation", async () => { + const started = await start() + const reference = Shipment.ref("truck") + await reference.dispatch({ carrierIds: ["a", "b", "c"] }) + await reference.stopChasing() + + expect(await reminderNames(started)).toEqual(["audit"]) + }) +}) From 3d933015cf272ed937ae15e237c384fed7d75cd8 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 21 Sep 2026 21:10:35 -0700 Subject: [PATCH 02/10] fix: correct the reminder cancellation contract Four defects from review. transmit() kept a public overload returning ScheduledOperationsFor, so callers saw the scheduling map after its calls started returning a handle. Only the implementation signature had changed, which callers never see. transmit() now has its own TransmittedOperationsFor, and both the type test and the packaged consumer fixture assert that its calls return nothing. unscheduleAll compared message_operation alone. That column is null on every row written before the keyed-reminder migration, where the name is still the operation, so an old recurring reminder survived the cancel and kept firing. It now compares COALESCE(message_operation, operation), which is the fallback the five dispatch sites already use. A cancellation could land between claimReminder and enqueueReminder. The reload then found no row and raised LostActivation, which is not UnknownOperation, so the scheduler rethrew and a supervised scheduler exited. enqueueReminder now separates a cancelled reminder from a lost claim: a row that is gone returns false and the occurrence is dropped quietly, while a row whose claim changed still raises. The two dispatches nested their cancellation branches inside an `in` check. ReminderIntent now carries an absent `cancel`, so the union discriminates on one property and both dispatches read as guard clauses. The contract said an occurrence already claimed still runs. That was wrong in the window this change defines: a cancellation cannot recall a message the scheduler already wrote, but it does pre-empt a claimed occurrence that has not been enqueued. --- CHANGELOG.md | 4 ++- docs/api.md | 3 ++ docs/correctness.md | 6 ++-- src/actor.ts | 25 ++++++++-------- src/cloudflare/engine.ts | 9 ++++-- src/index.ts | 1 + src/reference.ts | 4 +++ src/repository.ts | 30 +++++++++++++------ src/runtime.ts | 3 +- test/actor-operations.types.ts | 2 +- test/fixtures/actor-operations-consumer.mts | 3 +- test/reminder-cancellation.test.ts | 32 +++++++++++++++++++++ 12 files changed, 93 insertions(+), 29 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b15dbab..8d028d6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,7 +7,9 @@ `unscheduleAll()` removes every key of one operation. Both stage an intent beside the schedules, so they apply in the order the turn called them, commit with the state change that decided them, and cancel nothing when a turn - throws. Cancellation works on the SQL backends and on Durable Objects. + throws. Cancellation works on the SQL backends and on Durable Objects. A + cancellation that lands on an occurrence the scheduler claimed but has not yet + enqueued pre-empts it, and the scheduler continues rather than failing. - `schedule()` now returns a `ReminderHandle` (`{ name: string }`) instead of `void`. A handle is a plain object, so it survives in actor state and still cancels after a deactivation. Code that assigned the result to `void` needs diff --git a/docs/api.md b/docs/api.md index 7e83405..cd54606 100644 --- a/docs/api.md +++ b/docs/api.md @@ -45,6 +45,9 @@ authorization, capability boundaries, and release validation. - `Actor`: base class providing `ref()`, `actorId`, `currentMessage`, `observables()`, `reject()`, `emit()`, `transmit()`, `commitAction()`, `schedule()`, `sendTo()`, and protected lifecycle hooks. + `ScheduledOperationsFor` types the map `schedule()` returns, whose calls return a + `ReminderHandle`. `TransmittedOperationsFor` types the map `transmit()` + returns, whose calls return nothing. - `broadcastValue(value)`: mark an observable so its changed value enters the durable invalidation envelope. - `broadcastInvalidation(value)`: compare the real observable value but put diff --git a/docs/correctness.md b/docs/correctness.md index 699fdeb..f94cf3b 100644 --- a/docs/correctness.md +++ b/docs/correctness.md @@ -6,8 +6,10 @@ - Different identities may execute concurrently. - Sequence allocation and durable enqueue are one transaction. - A reminder can be cancelled. A cancellation commits with the state change that - decided it, and applies in the order the turn called it. An occurrence the - scheduler already claimed still runs; the cancellation removes later ones. + decided it, and applies in the order the turn called it. A cancellation cannot + recall an occurrence the scheduler already turned into a message. It does + pre-empt one the scheduler claimed but has not yet enqueued, and the scheduler + treats that as ordinary work rather than a failure. - Concurrent callers that create the same actor produce one instance row and distinct sequences. The mailbox locks that row by its primary key, so MySQL does not upgrade a shared lock and the enqueue does not deadlock. diff --git a/src/actor.ts b/src/actor.ts index 308e65f..c7576e5 100644 --- a/src/actor.ts +++ b/src/actor.ts @@ -11,6 +11,7 @@ import { type ScheduledOperations, type StagedOperationMap, type ScheduledOperationsFor, + type TransmittedOperationsFor, type StagedOperations, } from "./reference.js" import { jsonObject, normalizeJson } from "./serialization.js" @@ -101,6 +102,7 @@ export interface CommitActionIntent { } export interface ReminderIntent { + cancel?: undefined /** Without a key this is the operation. */ name: string operation: string @@ -317,7 +319,7 @@ export abstract class Actor { transmit( this: Actor & Pick & (Partial | NoInfer), - ): ScheduledOperationsFor> + ): TransmittedOperationsFor> transmit(): StagedOperationMap { return createStagedOperationMap(this.#operations, (operation, argumentsValue) => { this.#intents.effects.push({ @@ -361,20 +363,19 @@ export abstract class Actor { } unschedule(operationOrHandle: string | ReminderHandle, options: { key?: string | number } = {}) { - if (typeof operationOrHandle === "object" && operationOrHandle !== null) { - if (options.key !== undefined) { - throw new TypeError("a reminder handle already names its key") - } - const name = (operationOrHandle as ReminderHandle).name - if (typeof name !== "string" || name.length === 0) { - throw new TypeError("unschedule requires a reminder handle returned by schedule") - } - this.#intents.reminders.push({ cancel: "one", name }) + if (typeof operationOrHandle === "string") { + const key = validatedReminderKey(options.key) + this.#intents.reminders.push({ cancel: "one", name: reminderName(operationOrHandle, key) }) return } + if (options.key !== undefined) throw new TypeError("a reminder handle already names its key") - const key = validatedReminderKey(options.key) - this.#intents.reminders.push({ cancel: "one", name: reminderName(operationOrHandle, key) }) + const name = operationOrHandle?.name + if (typeof name !== "string" || name.length === 0) { + throw new TypeError("unschedule requires a reminder handle returned by schedule") + } + + this.#intents.reminders.push({ cancel: "one", name }) } unscheduleAll(operation: string) { diff --git a/src/cloudflare/engine.ts b/src/cloudflare/engine.ts index f166560..95d0a3b 100644 --- a/src/cloudflare/engine.ts +++ b/src/cloudflare/engine.ts @@ -673,9 +673,12 @@ export class ActorEngine { }) } for (const intent of intents.reminders) { - if ("cancel" in intent) { - if (intent.cancel === "all") this.store.deleteRemindersFor(intent.operation) - else this.store.deleteReminder(intent.name) + if (intent.cancel === "all") { + this.store.deleteRemindersFor(intent.operation) + continue + } + if (intent.cancel === "one") { + this.store.deleteReminder(intent.name) continue } this.store.saveReminder({ diff --git a/src/index.ts b/src/index.ts index 077a789..b86be1b 100644 --- a/src/index.ts +++ b/src/index.ts @@ -109,6 +109,7 @@ export { type ActorSnapshot, type ScheduledOperations, type ScheduledOperationsFor, + type TransmittedOperationsFor, type StagedOperations, } from "./reference.js" export type { diff --git a/src/reference.ts b/src/reference.ts index 59e2015..bd2de94 100644 --- a/src/reference.ts +++ b/src/reference.ts @@ -107,6 +107,10 @@ export type ScheduledOperationsFor = { [Key in ActorOperationNames]: ScheduledMethod } +export type TransmittedOperationsFor = { + [Key in ActorOperationNames]: StagedMethod +} + export interface StagedOperationMap { [operation: string]: (argumentsValue?: Record) => void } diff --git a/src/repository.ts b/src/repository.ts index c4122c5..28bc301 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -767,12 +767,18 @@ export class Repository { }) for (const reminder of input.intents.reminders) { - if ("cancel" in reminder) { - const column = reminder.cancel === "all" ? "message_operation" : "operation" - const value = reminder.cancel === "all" ? reminder.operation : reminder.name + if (reminder.cancel === "all") { await connection.run( - `DELETE FROM ${this.table("reminders")} WHERE instance_id = ? AND ${column} = ?`, - [turn.instance.id, value], + `DELETE FROM ${this.table("reminders")} + WHERE instance_id = ? AND COALESCE(message_operation, operation) = ?`, + [turn.instance.id, reminder.operation], + ) + continue + } + if (reminder.cancel === "one") { + await connection.run( + `DELETE FROM ${this.table("reminders")} WHERE instance_id = ? AND operation = ?`, + [turn.instance.id, reminder.name], ) continue } @@ -1759,8 +1765,8 @@ export class Repository { async enqueueReminder( reminder: ReminderRow, options: { nowMilliseconds?: number } = {}, - ): Promise { - await this.settings.database.transaction(async (connection) => { + ): Promise { + return this.settings.database.transaction(async (connection) => { const now = options.nowMilliseconds ?? (await connection.nowMilliseconds()) const claimed = await connection.get( `SELECT reminders.*, instances.actor_type, instances.actor_id @@ -1769,7 +1775,14 @@ export class Repository { WHERE reminders.id = ? AND reminders.status = 'scheduled' AND reminders.claimed_by = ?`, [reminder.id, reminder.claimed_by], ) - if (!claimed) throw new LostActivation("reminder claim no longer matches") + if (!claimed) { + const surviving = await connection.get<{ id: string }>( + `SELECT id FROM ${this.table("reminders")} WHERE id = ?`, + [reminder.id], + ) + if (!surviving) return false + throw new LostActivation("reminder claim no longer matches") + } await this.enqueueInTransaction(connection, { actorType: claimed.actor_type, actorId: claimed.actor_id, @@ -1800,6 +1813,7 @@ export class Repository { claimed.claimed_by, ], ) + return true }) } diff --git a/src/runtime.ts b/src/runtime.ts index 26ed144..5a7b99b 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -1421,7 +1421,8 @@ export class SolidObjectsRuntime { if (!actor.operations.has(dispatchOperation)) { throw new UnknownOperation(`unknown reminder operation ${JSON.stringify(dispatchOperation)}`) } - await this.repository.enqueueReminder(reminder, options) + if (!(await this.repository.enqueueReminder(reminder, options))) return + this.wakeUp("actors") this.emitInstrumentation("reminder.enqueued", { reminderId: reminder.id, diff --git a/test/actor-operations.types.ts b/test/actor-operations.types.ts index d8b3606..fae5562 100644 --- a/test/actor-operations.types.ts +++ b/test/actor-operations.types.ts @@ -46,7 +46,7 @@ export class ChatRun extends ParentRun { operations.finish() operations.optional() operations.optional({ generation: 1 }) - this.transmit().recoverIfStuck({ generation: 1 }) + expectTypeOf(this.transmit().recoverIfStuck({ generation: 1 })).toEqualTypeOf() this.transmit().finish() this.transmit().optional() this.emit("run_model", { onSuccess: "finish", onFailure: "failTurn" }) diff --git a/test/fixtures/actor-operations-consumer.mts b/test/fixtures/actor-operations-consumer.mts index d3256e2..497e449 100644 --- a/test/fixtures/actor-operations-consumer.mts +++ b/test/fixtures/actor-operations-consumer.mts @@ -33,7 +33,8 @@ export class ChatRun extends ParentRun { this.unscheduleAll("finish") operations.finish() operations.optional() - this.transmit().recoverIfStuck({ generation: 1 }) + const transmitted: void = this.transmit().recoverIfStuck({ generation: 1 }) + void transmitted this.emit("run_model", { onFailure: "finish" }) // @ts-expect-error operation typo operations.recoverIfStcuk({ generation: 1 }) diff --git a/test/reminder-cancellation.test.ts b/test/reminder-cancellation.test.ts index 06b8497..7b5a2c3 100644 --- a/test/reminder-cancellation.test.ts +++ b/test/reminder-cancellation.test.ts @@ -42,6 +42,10 @@ class Subscription extends Actor { throw new Error("turn failed") } + stopAllTrials(): void { + this.unscheduleAll("trialExpired") + } + cancelBadHandle(): void { this.unschedule({ nope: "x" } as unknown as ReminderHandle) } @@ -193,6 +197,34 @@ describe("reminder cancellation", () => { expect(await reminderNames(started)).toEqual(["audit", "chaseCarrier:a", "chaseCarrier:c"]) }) + it("cancels a reminder migrated before message_operation existed", async () => { + const started = await start() + const reference = Subscription.ref("alice") + await reference.startTrial() + // A row written before the keyed-reminder migration carries no + // message_operation, and its name is still the operation. + await started.settings.database.connection((connection) => + connection.run(`UPDATE solid_objects_reminders SET message_operation = NULL`), + ) + + await reference.stopAllTrials() + + expect(await reminderNames(started)).toEqual([]) + }) + + it("a cancel that lands on a claimed occurrence does not fail the scheduler", async () => { + const started = await start() + const reference = Subscription.ref("alice") + await reference.startRecurring() + + const claimed = await started.repository.claimReminder("test-scheduler") + expect(claimed).toBeDefined() + await reference.convertByName() + + await expect(started.repository.enqueueReminder(claimed!)).resolves.toBe(false) + expect(await started.reminderScheduler().runOnce()).toBe(0) + }) + it("cancels every key of one operation", async () => { const started = await start() const reference = Shipment.ref("truck") From e504508a024c0874d2751fce67a0cf236dbad236 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 21 Sep 2026 21:37:30 -0700 Subject: [PATCH 03/10] refactor: flatten the missing claim branch The two failure outcomes nested inside one check. They read as guard clauses now, and the existence query still runs only when the reload found nothing, so a successful enqueue makes no extra read. --- src/repository.ts | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/src/repository.ts b/src/repository.ts index 28bc301..bcf0ad2 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -1775,14 +1775,14 @@ export class Repository { WHERE reminders.id = ? AND reminders.status = 'scheduled' AND reminders.claimed_by = ?`, [reminder.id, reminder.claimed_by], ) - if (!claimed) { - const surviving = await connection.get<{ id: string }>( + const surviving = + !claimed && + (await connection.get<{ id: string }>( `SELECT id FROM ${this.table("reminders")} WHERE id = ?`, [reminder.id], - ) - if (!surviving) return false - throw new LostActivation("reminder claim no longer matches") - } + )) + if (!claimed && !surviving) return false + if (!claimed) throw new LostActivation("reminder claim no longer matches") await this.enqueueInTransaction(connection, { actorType: claimed.actor_type, actorId: claimed.actor_id, From 3e805a292a6b354f115a85e59ddee33731b2a2b4 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 21 Sep 2026 21:43:51 -0700 Subject: [PATCH 04/10] fix: narrow durable object state and correct the return docs The reminder helper asserted an unknown instance into a fabricated shape and typed exec as Function, which threw away the row type. It takes the state the callback already supplies, as the neighbouring test does, and names the row type on exec. The typed operation reference still said both maps return void. A scheduled call returns a handle now and a transmitted one returns void, so it says that. --- docs/api.md | 8 +++++--- test/cloudflare/runtime.test.ts | 11 +++++------ 2 files changed, 10 insertions(+), 9 deletions(-) diff --git a/docs/api.md b/docs/api.md index cd54606..c99895b 100644 --- a/docs/api.md +++ b/docs/api.md @@ -407,9 +407,11 @@ effect results and the handler's declared argument/result types in agreement. ### Typed operation references `schedule` and `transmit` infer this actor's operation names and arguments, including -inside actor methods and for inherited application operations. The returned -`ScheduledOperationsFor` values return `void` and preserve required, -optional, and zero-argument operation signatures. No non-null assertion is needed: +inside actor methods and for inherited application operations. A scheduled operation +call returns a `ReminderHandle`, through `ScheduledOperationsFor`, and a +transmitted one returns `void`, through `TransmittedOperationsFor`. Both +preserve required, optional, and zero-argument operation signatures. No non-null +assertion is needed: ```typescript class ChatRun extends Actor { diff --git a/test/cloudflare/runtime.test.ts b/test/cloudflare/runtime.test.ts index de8d59b..f12d83d 100644 --- a/test/cloudflare/runtime.test.ts +++ b/test/cloudflare/runtime.test.ts @@ -110,11 +110,10 @@ describe("Durable Objects runtime", () => { async function remainingReminders(actorId: string): Promise { const stub = env.ACTORS.getByName(JSON.stringify(["Counter", actorId])) - return runInDurableObject(stub, (instance: unknown) => { - const storage = (instance as { ctx: { storage: { sql: { exec: Function } } } }).ctx.storage - return storage.sql - .exec("SELECT name FROM reminders ORDER BY name") + return runInDurableObject(stub, (_object, state) => + state.storage.sql + .exec<{ name: string }>("SELECT name FROM reminders ORDER BY name") .toArray() - .map((row: { name: string }) => row.name) - }) + .map((row) => row.name), + ) } From 00b2fb9b40565fc7dfc568cdc584f71645f2de64 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 21 Sep 2026 22:00:04 -0700 Subject: [PATCH 05/10] docs: make the cancellation example compile The example cancelled chaseCarrier from a Subscription that never declared or scheduled it, and used status, renewal, and MONTH without declaring them. It is two classes now, one per case, each declaring the operations it cancels. --- docs/api.md | 48 +++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 45 insertions(+), 3 deletions(-) diff --git a/docs/api.md b/docs/api.md index c99895b..3f61c3d 100644 --- a/docs/api.md +++ b/docs/api.md @@ -175,17 +175,57 @@ armed. `unschedule()` cancels one alarm, by operation, by operation and key, or by that handle. `unscheduleAll()` cancels every key of one operation. ```typescript +const MONTH = 30 * 24 * 60 * 60 * 1000 + class Subscription extends Actor { - chase: ReminderHandle | null = null + static override readonly actorType = "subscriptions" + + status = "trialing" + renewal: ReminderHandle | null = null + + startTrial(): void { + this.schedule({ at: new Date(Date.now() + (14 * MONTH) / 30) }).trialExpired() + } convertToPaid(): void { this.status = "active" this.unschedule("trialExpired") - this.chase = this.schedule({ at: renewal, everyMilliseconds: MONTH }).chargeRenewal() + this.renewal = this.schedule({ + at: new Date(Date.now() + MONTH), + everyMilliseconds: MONTH, + }).chargeRenewal() } cancelled(): void { - if (this.chase) this.unschedule(this.chase) + this.status = "cancelled" + if (this.renewal) this.unschedule(this.renewal) + } + + trialExpired(): void { + this.status = "expired" + } + + chargeRenewal(): void {} +} +``` + +`this.unschedule(this.renewal)` and `this.unschedule("chargeRenewal")` cancel the +same alarm. Prefer the handle when the actor already stored one, because it +cannot drift from the name that armed the reminder. + +A keyed alarm cancels by the key that armed it, and `unscheduleAll()` cancels +every key of one operation: + +```typescript +class Shipment extends Actor { + static override readonly actorType = "shipments" + + dispatch({ carrierIds }: { carrierIds: string[] }): void { + for (const carrierId of carrierIds) { + this.schedule({ at: new Date(Date.now() + MONTH / 30), key: carrierId }).chaseCarrier({ + carrierId, + }) + } } shipped({ carrierId }: { carrierId: string }): void { @@ -195,6 +235,8 @@ class Subscription extends Actor { stopChasing(): void { this.unscheduleAll("chaseCarrier") } + + chaseCarrier(_options: { carrierId: string }): void {} } ``` From ab5fd6aa7aa0fc717762fb8373782625938204df Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 21 Sep 2026 22:12:15 -0700 Subject: [PATCH 06/10] fix: refuse an unknown operation in unschedule schedule() already threw UnknownOperation for an operation the actor does not declare, because the staged operation map asserts it. unschedule() and unscheduleAll() took any string, composed a name from it, and deleted nothing. A typo cancelled quietly and left a recurring reminder running, which is the failure this feature exists to prevent. Both now check the operation against the actor's declared operations and throw the same error. A handle skips the check, because the schedule() call that produced it was already checked. --- CHANGELOG.md | 3 +++ docs/api.md | 5 +++++ src/actor.ts | 8 ++++++++ test/reminder-cancellation.test.ts | 19 +++++++++++++++++++ 4 files changed, 35 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8d028d6..152d0a9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,9 @@ ## Unreleased +- Refuse an unknown operation in `unschedule()` and `unscheduleAll()`. + `schedule()` already threw `UnknownOperation` for one, so a typo cancelled + nothing quietly and left a recurring reminder running. - Add reminder cancellation. `unschedule()` removes one alarm by operation, by operation and key, or by the handle `schedule()` now returns. `unscheduleAll()` removes every key of one operation. Both stage an intent diff --git a/docs/api.md b/docs/api.md index 3f61c3d..c27af38 100644 --- a/docs/api.md +++ b/docs/api.md @@ -240,6 +240,11 @@ class Shipment extends Actor { } ``` +`unschedule()` and `unscheduleAll()` refuse an operation the actor does not +declare, with the `UnknownOperation` that `schedule()` already throws, so a typo +fails the turn rather than cancelling nothing. A handle skips that check, because +the `schedule()` call that produced it was already checked. + A cancellation is staged like a schedule, so it commits with the state change that decided it and a turn that throws cancels nothing. Both apply in the order the turn called them, so cancelling and then scheduling the same name leaves it diff --git a/src/actor.ts b/src/actor.ts index c7576e5..5ba01d4 100644 --- a/src/actor.ts +++ b/src/actor.ts @@ -364,6 +364,7 @@ export abstract class Actor { unschedule(operationOrHandle: string | ReminderHandle, options: { key?: string | number } = {}) { if (typeof operationOrHandle === "string") { + this.#assertOperation(operationOrHandle) const key = validatedReminderKey(options.key) this.#intents.reminders.push({ cancel: "one", name: reminderName(operationOrHandle, key) }) return @@ -379,9 +380,16 @@ export abstract class Actor { } unscheduleAll(operation: string) { + this.#assertOperation(operation) this.#intents.reminders.push({ cancel: "all", operation }) } + #assertOperation(operation: string): void { + if (this.#operations.has(operation)) return + + throw new UnknownOperation(`unknown operation ${JSON.stringify(operation)}`) + } + sendTo( reference: ActorReference, options: OutboundMessageOptions = {}, diff --git a/test/reminder-cancellation.test.ts b/test/reminder-cancellation.test.ts index 7b5a2c3..9fc2e2b 100644 --- a/test/reminder-cancellation.test.ts +++ b/test/reminder-cancellation.test.ts @@ -46,6 +46,14 @@ class Subscription extends Actor { this.unscheduleAll("trialExpired") } + cancelUnknown(): void { + this.unschedule("noSuchOperation") + } + + cancelAllUnknown(): void { + this.unscheduleAll("noSuchOperation") + } + cancelBadHandle(): void { this.unschedule({ nope: "x" } as unknown as ReminderHandle) } @@ -182,6 +190,17 @@ describe("reminder cancellation", () => { expect(Number(rows[0]!.run_at_ms)).toBe(Date.UTC(2031, 0, 1)) }) + it("refuses an unknown operation instead of cancelling nothing", async () => { + const started = await start() + const reference = Subscription.ref("alice") + await reference.startTrial() + + await expect(reference.cancelUnknown()).rejects.toThrow() + await expect(reference.cancelAllUnknown()).rejects.toThrow() + + expect(await reminderNames(started)).toEqual(["trialExpired"]) + }) + it("rejects a malformed handle", async () => { const started = await start() await expect(Subscription.ref("alice").cancelBadHandle()).rejects.toThrow() From 6fcd528a5ee0e6b23507e41fda5aaa050e2ed615 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 21 Sep 2026 22:43:03 -0700 Subject: [PATCH 07/10] feat: read the schedule from an actor reminder returns one armed alarm as a ScheduledReminder and reminders lists every key of one operation, matching what the Ruby port already offers. Both are async, because a TypeScript actor holds no rows and reads its own through a reader the runtime supplies at hydration. That is one injection point, so both engines get it: the SQL runtime reads the reminders table for the instance, and the Durable Objects engine reads the object's own store. A read starts from the committed rows and applies the intents staged so far, so an actor that schedules and then reads sees what the commit will write, and one that cancels and then reads sees the alarm gone. key and intervalMilliseconds are null rather than undefined. The first draft used undefined and a test caught it: returning a ScheduledReminder straight from an operation failed with InvalidPayload, because undefined does not serialise. Returning one is the obvious thing to do, so the type makes it work. A projection has no reader and throws rather than reporting an armed alarm as absent, which is the failure this whole feature exists to prevent. ScheduledReminder rather than ReminderStatus, because the administration API already exports that name for the status string. --- CHANGELOG.md | 4 + docs/api.md | 29 +++++++ docs/correctness.md | 3 + src/actor.ts | 119 +++++++++++++++++++++++++---- src/cloudflare/engine.ts | 23 +++++- src/definition.ts | 4 +- src/index.ts | 2 + src/repository.ts | 9 +++ src/runtime.ts | 18 +++++ src/types.ts | 15 ++++ test/cloudflare/runtime.test.ts | 9 +++ test/cloudflare/worker.ts | 5 ++ test/reminder-cancellation.test.ts | 63 ++++++++++++++- 13 files changed, 286 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 152d0a9..d1503a1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## Unreleased +- Add reminder reading. `reminder()` returns one armed alarm as a + `ScheduledReminder`, and `reminders()` lists every key of one operation. Both + apply the intents staged so far in the turn, so a read agrees with what the + commit will write. Reading works on the SQL backends and on Durable Objects. - Refuse an unknown operation in `unschedule()` and `unscheduleAll()`. `schedule()` already threw `UnknownOperation` for one, so a typo cancelled nothing quietly and left a recurring reminder running. diff --git a/docs/api.md b/docs/api.md index c27af38..29b5000 100644 --- a/docs/api.md +++ b/docs/api.md @@ -70,6 +70,9 @@ createdAtMs }` shape returned by `SolidObjectsRuntime.snapshotWithIncarnation`. `DestroyOptions`: the options for authorization, idempotency, time, and schedule that the reference methods use. +`ScheduledReminder` is one armed reminder as an actor reads it, and +`ReminderReader` is how a runtime supplies them. + `ActorIntents`, `EffectIntent`, `CommitActionIntent`, `ReminderIntent`, `UnscheduleIntent`, `UnscheduleAllIntent`, `ReminderMutation`, `OutboundMessageIntent`, `ReminderOptions`, `OutboundMessageOptions`, @@ -245,6 +248,32 @@ declare, with the `UnknownOperation` that `schedule()` already throws, so a typo fails the turn rather than cancelling nothing. A handle skips that check, because the `schedule()` call that produced it was already checked. +#### Reading the schedule + +`reminder()` returns the armed alarm as a `ScheduledReminder`, or `undefined`. +`reminders()` returns every key of one operation. Both are async, because an +actor reads its own rows rather than holding them in memory: + +```typescript +async nextChargeAt(): Promise { + return (await this.reminder("chargeRenewal"))?.runAtMilliseconds ?? null +} + +async pendingCarriers(): Promise<(string | null)[]> { + return (await this.reminders("chaseCarrier")).map((reminder) => reminder.key) +} +``` + +A read applies the intents staged so far in the turn, so an actor that schedules +and then reads sees what the commit will write, and one that cancels and then +reads sees the alarm gone. + +`key` and `intervalMilliseconds` are `null` rather than `undefined` when absent, +so a `ScheduledReminder` returns from an operation without a serialization error. + +Reading is available during a turn. A projection has no reader and throws, rather +than reporting an armed alarm as absent. + A cancellation is staged like a schedule, so it commits with the state change that decided it and a turn that throws cancels nothing. Both apply in the order the turn called them, so cancelling and then scheduling the same name leaves it diff --git a/docs/correctness.md b/docs/correctness.md index f94cf3b..e34af95 100644 --- a/docs/correctness.md +++ b/docs/correctness.md @@ -5,6 +5,9 @@ - Delivery is ordered per actor identity and at least once. - Different identities may execute concurrently. - Sequence allocation and durable enqueue are one transaction. +- An actor reads its own schedule. A read applies the intents staged so far in + the turn, so it agrees with what the commit will write rather than with what + the turn began with. - A reminder can be cancelled. A cancellation commits with the state change that decided it, and applies in the order the turn called it. A cancellation cannot recall an occurrence the scheduler already turned into a message. It does diff --git a/src/actor.ts b/src/actor.ts index 5ba01d4..c2746a2 100644 --- a/src/actor.ts +++ b/src/actor.ts @@ -2,6 +2,7 @@ import { currentMessage, currentRuntime } from "./context.js" import { getDefaultRuntime } from "./default-runtime.js" import type { StateMigration } from "./definition.js" import { InvalidRejectionCode, Rejected, UnknownOperation } from "./errors.js" +import type { ReminderStatus } from "./reminder-administration.js" import { TRANSMIT_EFFECT } from "./transmit-effect.js" import { randomUUID } from "./platform/uuid.js" import { @@ -20,6 +21,8 @@ import type { EffectHandle, JsonObject, ReminderHandle, + ReminderReader, + ScheduledReminder, JsonValue, MessageContext, } from "./types.js" @@ -179,6 +182,66 @@ function validatedReminderKey(key: string | number | undefined): string | undefi * reverse, and it is refused here rather than at the insert, once the turn is * already doing work. */ +function handleName(handle: ReminderHandle, key: string | number | undefined): string { + if (key !== undefined) throw new TypeError("a reminder handle already names its key") + + const name = handle?.name + if (typeof name !== "string" || name.length === 0) { + throw new TypeError("a reminder handle returned by schedule is required") + } + + return name +} + +function reminderKeyOf(name: string, operation: string): string | null { + return name === operation ? null : name.slice(operation.length + 1) +} + +function reminderStatusOf(options: { + name: string + operation: string + runAtMilliseconds: number + intervalMilliseconds: number | null + missedPolicy: "all" | "latest" + status: ReminderStatus +}): ScheduledReminder { + return { + name: options.name, + operation: options.operation, + key: reminderKeyOf(options.name, options.operation), + runAtMilliseconds: options.runAtMilliseconds, + intervalMilliseconds: options.intervalMilliseconds, + missedPolicy: options.missedPolicy, + status: options.status, + handle: { name: options.name }, + } +} + +function applyReminderIntent(view: Map, intent: ReminderMutation): void { + if (intent.cancel === "all") { + for (const [name, status] of view) { + if (status.operation === intent.operation) view.delete(name) + } + return + } + if (intent.cancel === "one") { + view.delete(intent.name) + return + } + + view.set( + intent.name, + reminderStatusOf({ + name: intent.name, + operation: intent.operation, + runAtMilliseconds: intent.atMilliseconds, + intervalMilliseconds: intent.intervalMilliseconds ?? null, + missedPolicy: intent.missedPolicy, + status: "scheduled", + }), + ) +} + function reminderName(operation: string, key: string | undefined): string { if (key === undefined) return operation @@ -211,6 +274,8 @@ export abstract class Actor { } readonly #actorId: string + #readReminders: ReminderReader | undefined + readonly #intents: ActorIntents = { effects: [], commitActions: [], @@ -363,25 +428,50 @@ export abstract class Actor { } unschedule(operationOrHandle: string | ReminderHandle, options: { key?: string | number } = {}) { + this.#intents.reminders.push({ + cancel: "one", + name: this.#reminderNameOf(operationOrHandle, options), + }) + } + + unscheduleAll(operation: string) { + this.#assertOperation(operation) + this.#intents.reminders.push({ cancel: "all", operation }) + } + + async reminder( + operationOrHandle: string | ReminderHandle, + options: { key?: string | number } = {}, + ): Promise { + return (await this.#reminderView()).get(this.#reminderNameOf(operationOrHandle, options)) + } + + async reminders(operation: string): Promise { + this.#assertOperation(operation) + const view = await this.#reminderView() + return [...view.values()].filter((status) => status.operation === operation) + } + + #reminderNameOf( + operationOrHandle: string | ReminderHandle, + options: { key?: string | number }, + ): string { if (typeof operationOrHandle === "string") { this.#assertOperation(operationOrHandle) - const key = validatedReminderKey(options.key) - this.#intents.reminders.push({ cancel: "one", name: reminderName(operationOrHandle, key) }) - return - } - if (options.key !== undefined) throw new TypeError("a reminder handle already names its key") - - const name = operationOrHandle?.name - if (typeof name !== "string" || name.length === 0) { - throw new TypeError("unschedule requires a reminder handle returned by schedule") + return reminderName(operationOrHandle, validatedReminderKey(options.key)) } - this.#intents.reminders.push({ cancel: "one", name }) + return handleName(operationOrHandle, options.key) } - unscheduleAll(operation: string) { - this.#assertOperation(operation) - this.#intents.reminders.push({ cancel: "all", operation }) + async #reminderView(): Promise> { + if (!this.#readReminders) { + throw new TypeError("reading reminders is not available outside an actor turn") + } + const view = new Map() + for (const status of await this.#readReminders()) view.set(status.name, status) + for (const intent of this.#intents.reminders) applyReminderIntent(view, intent) + return view } #assertOperation(operation: string): void { @@ -413,8 +503,9 @@ export abstract class Actor { } /** @internal */ - prepare(operations: ReadonlySet): void { + prepare(operations: ReadonlySet, readReminders?: ReminderReader): void { this.#operations = operations + this.#readReminders = readReminders } /** @internal */ diff --git a/src/cloudflare/engine.ts b/src/cloudflare/engine.ts index 95d0a3b..7269248 100644 --- a/src/cloudflare/engine.ts +++ b/src/cloudflare/engine.ts @@ -34,6 +34,7 @@ import type { JsonObject, JsonValue, SerializedError, + ScheduledReminder, } from "../types.js" import type { CloudflareSettings } from "./configuration.js" import { actorName, callHost, type ActorIdentity, type HostRequest } from "./protocol.js" @@ -357,6 +358,21 @@ export class ActorEngine { return { definition, instance, state } } + private readReminders = async (): Promise => + this.store.rows("SELECT record FROM reminders ORDER BY name").map((reminder) => ({ + name: reminder.name, + operation: reminder.operation, + key: + reminder.name === reminder.operation + ? null + : reminder.name.slice(reminder.operation.length + 1), + runAtMilliseconds: reminder.at, + intervalMilliseconds: reminder.interval, + missedPolicy: reminder.missed, + status: reminder.status, + handle: { name: reminder.name }, + })) + private async snapshot(identity: ActorIdentity): Promise { const { definition, instance, state } = this.committed(identity) const actor = hydrateActor({ definition, actorId: identity.actorId, state }) @@ -387,7 +403,11 @@ export class ActorEngine { }): Promise { const { input, payloadNames } = options const { definition, instance, state } = this.committed(input) - const actor = hydrateActor({ definition, actorId: input.actorId, state }) + const actor = hydrateActor({ + definition, + actorId: input.actorId, + state, + }) const identity = { actorType: input.actorType, actorId: input.actorId, @@ -486,6 +506,7 @@ export class ActorEngine { storedVersion: instance.stateVersion, storedState: instance.state, }), + readReminders: this.readReminders, }) if (this.cached?.actor !== actor) { await withActorContext({ actor, runtime: this.runtime }, () => actor.activate()) diff --git a/src/definition.ts b/src/definition.ts index 7936282..37ffb25 100644 --- a/src/definition.ts +++ b/src/definition.ts @@ -2,6 +2,7 @@ import { Actor, type ActorClass } from "./actor.js" import { withApplicationWritesForbidden } from "./context.js" import { ApplicationWriteForbidden, InvalidActor, StateMigrationError } from "./errors.js" import { deepCopy, jsonObject, normalizeJson } from "./serialization.js" +import type { ReminderReader } from "./types.js" import type { JsonObject } from "./types.js" export type PayloadBroadcastHandler = ( @@ -146,10 +147,11 @@ export function hydrateActor(options: { definition: ValidatedActorDefinition actorId: string state: JsonObject + readReminders?: ReminderReader }): ActorType { const { definition, actorId, state } = options const actor = new definition.actorClass(actorId) - actor.prepare(new Set(definition.operations)) + actor.prepare(new Set(definition.operations), options.readReminders) const target = actor as unknown as Record for (const key of definition.stateKeys) target[key] = deepCopy(state[key]) return actor diff --git a/src/index.ts b/src/index.ts index b86be1b..769b8b6 100644 --- a/src/index.ts +++ b/src/index.ts @@ -132,6 +132,8 @@ export type { EffectFailurePayload, EffectHandle, ReminderHandle, + ReminderReader, + ScheduledReminder, EffectSuccessPayload, InvocationOptions, JsonObject, diff --git a/src/repository.ts b/src/repository.ts index bcf0ad2..0e7296d 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -1169,6 +1169,15 @@ export class Repository { }) } + async remindersForInstance(instanceId: string): Promise { + return this.settings.database.connection((connection) => + connection.all( + `SELECT * FROM ${this.table("reminders")} WHERE instance_id = ? ORDER BY operation`, + [instanceId], + ), + ) + } + async findInstanceByIdentity( actorType: string, actorId: string, diff --git a/src/runtime.ts b/src/runtime.ts index 5a7b99b..03a058c 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -117,6 +117,7 @@ import type { MessageContext, MessageStatus, SnapshotOptions, + ScheduledReminder, } from "./types.js" import { SolidObjectsTestHelper } from "./test-helper.js" import { waitFor, Worker } from "./worker.js" @@ -165,6 +166,21 @@ type CommitActionHandler = ( context: CommitActionContext, ) => unknown | Promise +function scheduledReminderOf(row: ReminderRow): ScheduledReminder { + const operation = row.message_operation ?? row.operation + const name = row.operation + return { + name, + operation, + key: name === operation ? null : name.slice(operation.length + 1), + runAtMilliseconds: Number(row.run_at_ms), + intervalMilliseconds: row.interval_ms === null ? null : Number(row.interval_ms), + missedPolicy: row.missed_policy, + status: row.status, + handle: { name }, + } +} + export class SolidObjectsRuntime { readonly settings readonly repository @@ -1117,6 +1133,8 @@ export class SolidObjectsRuntime { definition, actorId: turn.message.actor_id, state: deepCopy(state), + readReminders: async () => + (await this.repository.remindersForInstance(turn.instance.id)).map(scheduledReminderOf), }) } catch (error) { throw new ActorSetupFailed(error) diff --git a/src/types.ts b/src/types.ts index 1afa2c0..8414e59 100644 --- a/src/types.ts +++ b/src/types.ts @@ -1,3 +1,5 @@ +import type { ReminderStatus } from "./reminder-administration.js" + export type JsonPrimitive = null | boolean | number | string export type JsonValue = JsonPrimitive | JsonValue[] | { [key: string]: JsonValue } export type JsonObject = { [key: string]: JsonValue } @@ -6,6 +8,19 @@ export type EffectHandle = { readonly id: string } export type ReminderHandle = { readonly name: string } +export interface ScheduledReminder { + readonly name: string + readonly operation: string + readonly key: string | null + readonly runAtMilliseconds: number + readonly intervalMilliseconds: number | null + readonly missedPolicy: "all" | "latest" + readonly status: ReminderStatus + readonly handle: ReminderHandle +} + +export type ReminderReader = () => Promise + export type SerializedError = { name: string message: string diff --git a/test/cloudflare/runtime.test.ts b/test/cloudflare/runtime.test.ts index f12d83d..ea2c516 100644 --- a/test/cloudflare/runtime.test.ts +++ b/test/cloudflare/runtime.test.ts @@ -71,6 +71,15 @@ describe("Durable Objects runtime", () => { expect(await runtime().ref(Counter, "cancelled").with({ authorizationContext }).count).toBe(0) }) + it("reads its own schedule inside a durable object", async () => { + const reference = runtime().ref(Counter, "reads-schedule").with({ authorizationContext }) + expect(await reference.readArmed()).toBeNull() + + await reference.arm({ at: Date.now() + 60_000 }) + + expect(await reference.readArmed()).toEqual({ name: "increment", interval: null }) + }) + it("cancels one keyed reminder and every key of an operation", async () => { const reference = runtime().ref(Counter, "keyed-cancel").with({ authorizationContext }) await reference.armKeyed({ at: Date.now() + 60_000, keys: ["a", "b", "c"] }) diff --git a/test/cloudflare/worker.ts b/test/cloudflare/worker.ts index c92b0b1..7afff63 100644 --- a/test/cloudflare/worker.ts +++ b/test/cloudflare/worker.ts @@ -73,6 +73,11 @@ export class Counter extends Actor { } } + async readArmed(): Promise<{ name: string; interval: number | null } | null> { + const found = await this.reminder("increment") + return found ? { name: found.name, interval: found.intervalMilliseconds } : null + } + disarm(): void { this.unschedule("increment") } diff --git a/test/reminder-cancellation.test.ts b/test/reminder-cancellation.test.ts index 9fc2e2b..9f5d253 100644 --- a/test/reminder-cancellation.test.ts +++ b/test/reminder-cancellation.test.ts @@ -2,7 +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 type { ReminderHandle } from "../src/types.js" +import type { ReminderHandle, ScheduledReminder } from "../src/types.js" class Subscription extends Actor { static override readonly actorType = "cancel-subscriptions" @@ -46,6 +46,20 @@ class Subscription extends Actor { this.unscheduleAll("trialExpired") } + async readTrial(): Promise { + return (await this.reminder("trialExpired")) ?? null + } + + async readAfterStagedSchedule(): Promise { + this.schedule({ at: new Date(Date.UTC(2031, 0, 1)) }).trialExpired() + return (await this.reminder("trialExpired"))?.runAtMilliseconds ?? null + } + + async readAfterStagedCancel(): Promise { + this.unschedule("trialExpired") + return (await this.reminder("trialExpired")) !== undefined + } + cancelUnknown(): void { this.unschedule("noSuchOperation") } @@ -84,6 +98,10 @@ class Shipment extends Actor { this.unscheduleAll("chaseCarrier") } + async pendingKeys(): Promise<(string | null)[]> { + return (await this.reminders("chaseCarrier")).map((status) => status.key).sort() + } + chaseCarrier(_options: { carrierId: string }): void {} audit(): void {} } @@ -190,6 +208,49 @@ describe("reminder cancellation", () => { expect(Number(rows[0]!.run_at_ms)).toBe(Date.UTC(2031, 0, 1)) }) + it("reads an armed reminder", async () => { + await start() + const reference = Subscription.ref("alice") + await reference.startRecurring() + + expect(await reference.readTrial()).toEqual({ + name: "trialExpired", + operation: "trialExpired", + key: null, + runAtMilliseconds: expect.any(Number), + intervalMilliseconds: 60_000, + missedPolicy: "latest", + status: "scheduled", + handle: { name: "trialExpired" }, + }) + }) + + it("reads nothing when no reminder is armed", async () => { + await start() + expect(await Subscription.ref("alice").readTrial()).toBeNull() + }) + + it("sees a schedule staged earlier in the same turn", async () => { + await start() + expect(await Subscription.ref("alice").readAfterStagedSchedule()).toBe(Date.UTC(2031, 0, 1)) + }) + + it("sees a cancel staged earlier in the same turn", async () => { + await start() + const reference = Subscription.ref("alice") + await reference.startTrial() + + expect(await reference.readAfterStagedCancel()).toBe(false) + }) + + it("lists every key of one operation", async () => { + await start() + const reference = Shipment.ref("truck") + await reference.dispatch({ carrierIds: ["a", "b", "c"] }) + + expect(await reference.pendingKeys()).toEqual(["a", "b", "c"]) + }) + it("refuses an unknown operation instead of cancelling nothing", async () => { const started = await start() const reference = Subscription.ref("alice") From 93db131a0a786704501c5a7f50654df82ff7f2d7 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 21 Sep 2026 23:02:22 -0700 Subject: [PATCH 08/10] fix: read reminders from a projection too A projection hydrated its actor without a reader, so an observable that asked what was armed threw instead of answering. The Ruby port reads from an observable, and a test there covers it, so this side did the same. Both Durable Objects projections and the SQL snapshot projection now pass the reader they already have the instance for. Record the one difference that stays. ScheduledReminder carries no occurrence count, because the SQL backends track one and Durable Objects does not, and reporting it for one backend only would be worse than leaving it out. --- docs/api.md | 9 +++++++-- src/cloudflare/engine.ts | 8 +++++++- src/runtime.ts | 6 ++++++ test/reminder-cancellation.test.ts | 28 ++++++++++++++++++++++++++++ 4 files changed, 48 insertions(+), 3 deletions(-) diff --git a/docs/api.md b/docs/api.md index 29b5000..cfb6b8c 100644 --- a/docs/api.md +++ b/docs/api.md @@ -271,8 +271,13 @@ reads sees the alarm gone. `key` and `intervalMilliseconds` are `null` rather than `undefined` when absent, so a `ScheduledReminder` returns from an operation without a serialization error. -Reading is available during a turn. A projection has no reader and throws, rather -than reporting an armed alarm as absent. +Reading is available during a turn and from a snapshot projection, so an +observable can report what is armed. `reminder()` and `reminders()` refuse an +operation the actor does not declare, as `schedule()` and `unschedule()` do. + +`ScheduledReminder` carries no occurrence count. The SQL backends track one and +Durable Objects does not, so it is left out rather than reported for one backend +only. A cancellation is staged like a schedule, so it commits with the state change that decided it and a turn that throws cancels nothing. Both apply in the order diff --git a/src/cloudflare/engine.ts b/src/cloudflare/engine.ts index 7269248..0fbd723 100644 --- a/src/cloudflare/engine.ts +++ b/src/cloudflare/engine.ts @@ -375,7 +375,12 @@ export class ActorEngine { private async snapshot(identity: ActorIdentity): Promise { const { definition, instance, state } = this.committed(identity) - const actor = hydrateActor({ definition, actorId: identity.actorId, state }) + const actor = hydrateActor({ + definition, + actorId: identity.actorId, + state, + readReminders: this.readReminders, + }) const before = stableJson(actorState(actor, definition.stateKeys)) const snapshot: JsonObject = { ...state } await withActorProjection({ actor, runtime: this.runtime }, async () => { @@ -407,6 +412,7 @@ export class ActorEngine { definition, actorId: input.actorId, state, + readReminders: this.readReminders, }) const identity = { actorType: input.actorType, diff --git a/src/runtime.ts b/src/runtime.ts index 03a058c..88c13a0 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -795,6 +795,12 @@ export class SolidObjectsRuntime { definition: registered.definition, actorId: options.actorId, state, + ...(instance === undefined + ? {} + : { + readReminders: async () => + (await this.repository.remindersForInstance(instance.id)).map(scheduledReminderOf), + }), }) return readonlyCopy({ actorType: options.actorType, diff --git a/test/reminder-cancellation.test.ts b/test/reminder-cancellation.test.ts index 9f5d253..38e3ac5 100644 --- a/test/reminder-cancellation.test.ts +++ b/test/reminder-cancellation.test.ts @@ -78,6 +78,22 @@ class Subscription extends Actor { } } +class Observed extends Actor { + static override readonly actorType = "cancel-observed" + + armed = false + + arm(): void { + this.schedule({ at: new Date(Date.UTC(2030, 0, 1)) }).ping() + } + + async armedName(): Promise { + return (await this.reminder("ping"))?.name ?? null + } + + ping(): void {} +} + class Shipment extends Actor { static override readonly actorType = "cancel-shipments" @@ -124,6 +140,7 @@ async function start(): Promise { }) runtime.register(Subscription) runtime.register(Shipment) + runtime.register(Observed) await runtime.install() return runtime } @@ -251,6 +268,17 @@ describe("reminder cancellation", () => { expect(await reference.pendingKeys()).toEqual(["a", "b", "c"]) }) + it("reads the schedule from a snapshot projection", async () => { + const started = await start() + const reference = Observed.ref("one") + await reference.arm() + + const snapshot = await started.snapshot(Observed.ref("one")) + + expect(snapshot).toBeDefined() + expect(await reference.armedName()).toBe("ping") + }) + it("refuses an unknown operation instead of cancelling nothing", async () => { const started = await start() const reference = Subscription.ref("alice") From 4c6a1a5df1a134d8928f9081248383e1ff623fc3 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 21 Sep 2026 23:17:16 -0700 Subject: [PATCH 09/10] fix: do not report a spent reminder as armed A one-shot keeps its row after it fires, as completed, and the view returned it. A next-run lookup reported an old time rather than nothing, and an existence check refused to re-arm an alarm that could never fire again. The shared view now drops completed rows, so both engines agree without either having to remember. buildSnapshotWithIncarnation hydrated without a reader, so a snapshot getter that read its schedule threw. The earlier test missed it because it used an operation rather than a getter, so the read ran in an ordinary turn. It is a getter now and asserts through the snapshot. Inline the two single-use helpers that built a ScheduledReminder, and replace the double assertion in the malformed-handle test with an expected type error. --- CHANGELOG.md | 4 +++ docs/api.md | 3 ++ src/actor.ts | 50 ++++++++---------------------- src/runtime.ts | 4 +++ test/reminder-cancellation.test.ts | 26 +++++++++++++--- 5 files changed, 45 insertions(+), 42 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d1503a1..15d6144 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ `ScheduledReminder`, and `reminders()` lists every key of one operation. Both apply the intents staged so far in the turn, so a read agrees with what the commit will write. Reading works on the SQL backends and on Durable Objects. +- Leave a one-shot reminder that already fired out of `reminder()` and + `reminders()`. Its row stays as `completed`, so a next-run lookup reported an + old time rather than nothing, and an existence check refused to re-arm an + alarm that could never fire again. - Refuse an unknown operation in `unschedule()` and `unscheduleAll()`. `schedule()` already threw `UnknownOperation` for one, so a typo cancelled nothing quietly and left a recurring reminder running. diff --git a/docs/api.md b/docs/api.md index cfb6b8c..204aefe 100644 --- a/docs/api.md +++ b/docs/api.md @@ -275,6 +275,9 @@ Reading is available during a turn and from a snapshot projection, so an observable can report what is armed. `reminder()` and `reminders()` refuse an operation the actor does not declare, as `schedule()` and `unschedule()` do. +A one-shot that already fired is not reported. Its row stays as `completed`, and +an alarm that cannot fire again is not armed. + `ScheduledReminder` carries no occurrence count. The SQL backends track one and Durable Objects does not, so it is left out rather than reported for one backend only. diff --git a/src/actor.ts b/src/actor.ts index c2746a2..7a6e718 100644 --- a/src/actor.ts +++ b/src/actor.ts @@ -2,7 +2,6 @@ import { currentMessage, currentRuntime } from "./context.js" import { getDefaultRuntime } from "./default-runtime.js" import type { StateMigration } from "./definition.js" import { InvalidRejectionCode, Rejected, UnknownOperation } from "./errors.js" -import type { ReminderStatus } from "./reminder-administration.js" import { TRANSMIT_EFFECT } from "./transmit-effect.js" import { randomUUID } from "./platform/uuid.js" import { @@ -193,30 +192,6 @@ function handleName(handle: ReminderHandle, key: string | number | undefined): s return name } -function reminderKeyOf(name: string, operation: string): string | null { - return name === operation ? null : name.slice(operation.length + 1) -} - -function reminderStatusOf(options: { - name: string - operation: string - runAtMilliseconds: number - intervalMilliseconds: number | null - missedPolicy: "all" | "latest" - status: ReminderStatus -}): ScheduledReminder { - return { - name: options.name, - operation: options.operation, - key: reminderKeyOf(options.name, options.operation), - runAtMilliseconds: options.runAtMilliseconds, - intervalMilliseconds: options.intervalMilliseconds, - missedPolicy: options.missedPolicy, - status: options.status, - handle: { name: options.name }, - } -} - function applyReminderIntent(view: Map, intent: ReminderMutation): void { if (intent.cancel === "all") { for (const [name, status] of view) { @@ -229,17 +204,16 @@ function applyReminderIntent(view: Map, intent: Remin return } - view.set( - intent.name, - reminderStatusOf({ - name: intent.name, - operation: intent.operation, - runAtMilliseconds: intent.atMilliseconds, - intervalMilliseconds: intent.intervalMilliseconds ?? null, - missedPolicy: intent.missedPolicy, - status: "scheduled", - }), - ) + view.set(intent.name, { + name: intent.name, + operation: intent.operation, + key: intent.name === intent.operation ? null : intent.name.slice(intent.operation.length + 1), + runAtMilliseconds: intent.atMilliseconds, + intervalMilliseconds: intent.intervalMilliseconds ?? null, + missedPolicy: intent.missedPolicy, + status: "scheduled", + handle: { name: intent.name }, + }) } function reminderName(operation: string, key: string | undefined): string { @@ -469,7 +443,9 @@ export abstract class Actor { throw new TypeError("reading reminders is not available outside an actor turn") } const view = new Map() - for (const status of await this.#readReminders()) view.set(status.name, status) + for (const reminder of await this.#readReminders()) { + if (reminder.status !== "completed") view.set(reminder.name, reminder) + } for (const intent of this.#intents.reminders) applyReminderIntent(view, intent) return view } diff --git a/src/runtime.ts b/src/runtime.ts index 88c13a0..8af5fee 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -743,6 +743,10 @@ export class SolidObjectsRuntime { definition: registered.definition, actorId: reference.actorId, state, + readReminders: async () => + instance === undefined + ? [] + : (await this.repository.remindersForInstance(instance.id)).map(scheduledReminderOf), }) const stateBefore = stableJson(actorState(actor, registered.definition.stateKeys)) const intentCount = actor.intentCount() diff --git a/test/reminder-cancellation.test.ts b/test/reminder-cancellation.test.ts index 38e3ac5..0599814 100644 --- a/test/reminder-cancellation.test.ts +++ b/test/reminder-cancellation.test.ts @@ -69,7 +69,8 @@ class Subscription extends Actor { } cancelBadHandle(): void { - this.unschedule({ nope: "x" } as unknown as ReminderHandle) + // @ts-expect-error a malformed handle exercises the runtime check + this.unschedule({ nope: "x" }) } trialExpired(): void { @@ -87,11 +88,15 @@ class Observed extends Actor { this.schedule({ at: new Date(Date.UTC(2030, 0, 1)) }).ping() } - async armedName(): Promise { - return (await this.reminder("ping"))?.name ?? null + get armedName(): Promise { + return this.reminder("ping").then((found) => found?.name ?? null) } ping(): void {} + + armDue(): void { + this.schedule({ at: new Date(Date.now() - 1_000) }).ping() + } } class Shipment extends Actor { @@ -268,6 +273,18 @@ describe("reminder cancellation", () => { expect(await reference.pendingKeys()).toEqual(["a", "b", "c"]) }) + it("does not report a one-shot that already fired", async () => { + const started = await start() + const reference = Observed.ref("one") + await reference.armDue() + expect(await reference.armedName).toBe("ping") + + expect(await started.reminderScheduler().runOnce()).toBe(1) + await started.worker().runUntilIdle() + + expect(await reference.armedName).toBeNull() + }) + it("reads the schedule from a snapshot projection", async () => { const started = await start() const reference = Observed.ref("one") @@ -275,8 +292,7 @@ describe("reminder cancellation", () => { const snapshot = await started.snapshot(Observed.ref("one")) - expect(snapshot).toBeDefined() - expect(await reference.armedName()).toBe("ping") + expect(snapshot.armedName).toBe("ping") }) it("refuses an unknown operation instead of cancelling nothing", async () => { From 66b11ccc1e3a774fdc17564ccabe895342b0e1b1 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 21 Sep 2026 23:28:15 -0700 Subject: [PATCH 10/10] test: read the schedule from an activation hook The Ruby port covers a read from on_activate and this side did not. The test starts a runtime, arms an alarm, closes it, and starts a second one against the same file, so onActivate runs again with the reminder already armed. It fails when the turn reader is removed. --- test/reminder-cancellation.test.ts | 44 ++++++++++++++++++++++++++++++ 1 file changed, 44 insertions(+) diff --git a/test/reminder-cancellation.test.ts b/test/reminder-cancellation.test.ts index 0599814..c579481 100644 --- a/test/reminder-cancellation.test.ts +++ b/test/reminder-cancellation.test.ts @@ -1,3 +1,6 @@ +import { mkdtemp, rm } from "node:fs/promises" +import { tmpdir } from "node:os" +import { join } from "node:path" import { afterEach, describe, expect, it } from "vitest" import { Actor } from "../src/actor.js" import { sqlite } from "../src/database/sqlite.js" @@ -83,6 +86,11 @@ class Observed extends Actor { static override readonly actorType = "cancel-observed" armed = false + seenOnActivate: string | null = null + + protected override async onActivate(): Promise { + this.seenOnActivate = (await this.reminder("ping"))?.name ?? null + } arm(): void { this.schedule({ at: new Date(Date.UTC(2030, 0, 1)) }).ping() @@ -134,6 +142,21 @@ afterEach(async () => { runtime = undefined }) +function configuredRuntime(path: string): SolidObjectsRuntime { + runtime = configure({ + database: sqlite({ path }), + authorizeMessage: () => true, + authorizeQuery: () => true, + pollingIntervalMilliseconds: 1, + syncPollingIntervalMilliseconds: 1, + maxAttempts: 1, + }) + runtime.register(Subscription) + runtime.register(Shipment) + runtime.register(Observed) + return runtime +} + async function start(): Promise { runtime = configure({ database: sqlite({ path: ":memory:" }), @@ -285,6 +308,27 @@ describe("reminder cancellation", () => { expect(await reference.armedName).toBeNull() }) + it("reads the schedule from an activation hook", async () => { + const directory = await mkdtemp(join(tmpdir(), "reminder-hook-")) + const path = join(directory, "hook.sqlite3") + try { + const first = configuredRuntime(path) + await first.install() + await Observed.ref("one").arm() + await first.close() + + // A second runtime activates the actor fresh, so onActivate runs again + // with the reminder already armed. + const second = configuredRuntime(path) + await second.install() + expect(await Observed.ref("one").seenOnActivate).toBe("ping") + await second.close() + runtime = undefined + } finally { + await rm(directory, { recursive: true, force: true }) + } + }) + it("reads the schedule from a snapshot projection", async () => { const started = await start() const reference = Observed.ref("one")