diff --git a/CHANGELOG.md b/CHANGELOG.md index 9b52921..423c355 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,7 +1,18 @@ # Changelog -## Unreleased - +## 0.17.7 - 2026-10-12 + +- Renew retained activations between short turns and across worker passes, + even when synchronous work prevents renewal timers from running. +- Keep an activation with an outstanding message claim discoverable after + lease loss. The next worker pass recovers the message in mailbox order + without a process restart, including with `maxAttempts: 1`. Reclaiming + increments `attempt_count`; lease loss alone does not create a dead letter. +- Run activation lease regressions in PostgreSQL and MySQL CI, reusing and + resetting their test tables so repeated runs leave no messages or extra tables. + Control the server-database test clocks to avoid runner scheduling delays; + SQLite keeps real synchronous waits with a 200 ms lease. + Allow 15 seconds for the 34-turn test's database round trips in CI. - Add `scripts/check-compatibility-fixtures.mjs` and a workflow with the name `Compatibility fixtures`. The script compares the `*.json` files in two directories by name and by bytes. It fails with one line for each file that @@ -10,9 +21,8 @@ of solid-objects-ruby at `main`. It runs on each push to `main`, one time each day, and on demand. Before this change, no job compared the shared fixtures. -- `docs/parity.md` now cites Ruby `solid_objects` 0.17.3. That Ruby release - adds guides, tests, and one development dependency, and does not change the - runtime, so the ledger has no new row. +- Track Ruby `solid_objects` 0.17.4 in the parity ledger, including renewal + between short turns and preservation of activation fences for claimed messages. ## 0.17.6 - 2026-10-09 diff --git a/docs/parity.md b/docs/parity.md index 26e2080..5105402 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -4,7 +4,7 @@ This ledger tracks capability parity with the Ruby `solid_objects` gem. Parity preserves a capability and its correctness or security boundary. It does not copy a Rails API into Node. -Reference: Ruby `solid_objects` 0.17.3. The JavaScript package began at the Ruby +Reference: Ruby `solid_objects` 0.17.4. The JavaScript package began at the Ruby design's `0.12` capability generation; that version number did not imply earlier JavaScript releases. @@ -54,7 +54,7 @@ is needed for the JavaScript stale-result race fix. | Query and projection purity | Native | Both runtimes reject state mutation and staged effects, recovery checks, commit actions, reminders, or outbound messages in queries, observable projections, and personalized payloads. Projection guards compare the complete staged work, including same-count replacements. Query and observable violations fail terminally with `QueryMutatedState`; payload violations fail only that payload. Snapshot projection reads enforce the same rule. | | JSON property preservation | Native | Reserved property names such as `__proto__`, `constructor`, and `prototype` remain ordinary data, including through size checks and immutable copies. Both suites consume `compatibility/json-values.json`. | | 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. | +| Ordered mailbox, sequence allocation, idempotency, retries, dead letters, leases, renewal, and fenced commits | Native | Both runtimes renew between turns when due and retain the fence while a claim remains, allowing expired-turn recovery in mailbox order even with one allowed attempt. Reclaiming increments the attempt count; lease loss alone does not dead-letter a message. 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. | diff --git a/package.json b/package.json index ad152fb..baff34b 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "solid-objects", - "version": "0.17.6", + "version": "0.17.7", "description": "SQL-backed virtual actor library for TypeScript and Node.js, with durable state, ordered operations, and automatic activation on SQLite, PostgreSQL, or MySQL", "type": "module", "license": "MIT", @@ -118,8 +118,8 @@ "test": "vitest run", "test:browser": "pnpm run build && playwright test", "test:coverage": "vitest run --coverage", - "test:postgresql": "vitest run test/postgresql.test.ts test/effect-recovery.test.ts test/instance-retention.test.ts test/transmit-ordering.test.ts", - "test:mysql": "vitest run test/mysql.test.ts test/effect-recovery.test.ts test/instance-retention.test.ts test/transmit-ordering.test.ts", + "test:postgresql": "vitest run test/postgresql.test.ts test/activation-lease.test.ts test/effect-recovery.test.ts test/instance-retention.test.ts test/transmit-ordering.test.ts", + "test:mysql": "vitest run test/mysql.test.ts test/activation-lease.test.ts test/effect-recovery.test.ts test/instance-retention.test.ts test/transmit-ordering.test.ts", "test:package": "node scripts/release-artifact-smoke.mjs", "test:recovery": "pnpm run build && node examples/failure-recovery/demo.ts", "test:at-least-once": "pnpm run build && node examples/at-least-once/demo.ts", diff --git a/src/repository.ts b/src/repository.ts index 153d02a..f3da4cf 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -651,13 +651,18 @@ export class Repository { SET activation_owner_id = NULL, activation_token = NULL, activation_expires_at_ms = NULL, updated_at_ms = ? WHERE id = ? AND activation_owner_id = ? AND activation_token = ? - AND activation_generation = ?`, + AND activation_generation = ? + AND NOT EXISTS ( + SELECT 1 FROM ${this.table("claimed_messages")} claimed + WHERE claimed.instance_id = ? + )`, [ now, activation.instanceId, activation.processId, activation.activationToken, activation.activationGeneration, + activation.instanceId, ], ) }) diff --git a/src/version.ts b/src/version.ts index 426eb9c..7baf0ef 100644 --- a/src/version.ts +++ b/src/version.ts @@ -1 +1 @@ -export const VERSION = "0.17.6" +export const VERSION = "0.17.7" diff --git a/src/worker.ts b/src/worker.ts index fbc68e8..762ef10 100644 --- a/src/worker.ts +++ b/src/worker.ts @@ -87,7 +87,9 @@ export class Worker { actor: execution.actor, turn, lastUsedAt: now, - renewAt: now + this.runtime.settings.leaseRenewalIntervalMilliseconds, + renewAt: + cached?.renewAt ?? + passStartedAt + this.runtime.settings.leaseRenewalIntervalMilliseconds, } this.activations.set(turn.instance.id, cached) passExhausted = @@ -95,6 +97,7 @@ export class Worker { performance.now() - passStartedAt >= this.runtime.settings.maxActivationDurationMilliseconds if (passExhausted) break + await this.maintainCachedActivations() turn = await this.runtime.repository.claim(this.processId, { activation: activationLease(cached.turn), }) diff --git a/test/activation-lease.test.ts b/test/activation-lease.test.ts new file mode 100644 index 0000000..2c89993 --- /dev/null +++ b/test/activation-lease.test.ts @@ -0,0 +1,243 @@ +import { postgresql } from "../src/database/postgresql.js" +import { mysql } from "../src/database/mysql.js" +import { afterEach, describe, expect, it, vi } from "vitest" +import { Actor } from "../src/actor.js" +import type { InstrumentationEvent } from "../src/configuration.js" +import { sqlite } from "../src/database/sqlite.js" +import type { Database, DatabaseConnection } from "../src/database/types.js" +import { configure, type SolidObjectsRuntime } from "../src/runtime.js" + +class LeaseCounter extends Actor { + static override readonly actorType = "LeaseCounter" + static beforeCommit: () => void | Promise = () => {} + count = 0 + + async increment(): Promise { + this.count += 1 + await LeaseCounter.beforeCommit() + return this.count + } +} + +let runtime: SolidObjectsRuntime +let elapsedMilliseconds: number | undefined +const events: InstrumentationEvent[] = [] + +afterEach(async () => { + await runtime?.repository.resetForTesting() + await runtime?.close() + vi.useRealTimers() + vi.restoreAllMocks() + elapsedMilliseconds = undefined + events.length = 0 + LeaseCounter.beforeCommit = () => {} +}) + +async function setup(): Promise { + const database = databaseFor(process.env.SOLID_OBJECTS_DATABASE_URL) + runtime = configure({ + database: withControlledServerClock(database), + tableNamePrefix: "lease_test_", + authorizeMessage: () => true, + authorizeQuery: () => true, + leaseDurationMilliseconds: 200, + leaseRenewalIntervalMilliseconds: 50, + maxAttempts: 1, + instrumentation: (event) => events.push(event), + }) + runtime.register(LeaseCounter) + await runtime.install() + await runtime.repository.resetForTesting() + if (database.family !== "sqlite") { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }) + } +} + +function databaseFor(connectionString: string | undefined): Database { + if (connectionString?.startsWith("postgres")) return postgresql({ connectionString }) + if (connectionString?.startsWith("mysql")) return mysql({ connectionString }) + return sqlite({ path: ":memory:" }) +} + +function withControlledServerClock(database: Database): Database { + if (database.family === "sqlite") return database + elapsedMilliseconds = 0 + const startedAt = Date.now() + vi.spyOn(performance, "now").mockImplementation(() => elapsedMilliseconds!) + const clockedConnection = (connection: DatabaseConnection): DatabaseConnection => ({ + run: connection.run.bind(connection), + get: connection.get.bind(connection), + all: connection.all.bind(connection), + nowMilliseconds: async () => startedAt + elapsedMilliseconds!, + }) + return { + family: database.family, + schemaIdentity: database.schemaIdentity, + transactionActive: () => database.transactionActive?.() ?? false, + connection: (callback) => + database.connection((connection) => callback(clockedConnection(connection))), + transaction: (callback, options) => + database.transaction((connection) => callback(clockedConnection(connection)), options), + close: () => database.close(), + } +} + +function blockFor(milliseconds: number): void { + if (elapsedMilliseconds !== undefined) { + elapsedMilliseconds += milliseconds + return + } + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, milliseconds) +} + +async function storedLease(): Promise<{ + activation_expires_at_ms: number | bigint + activation_generation: number | bigint +}> { + const row = await runtime.settings.database.connection((connection) => + connection.get<{ + activation_expires_at_ms: number | bigint + activation_generation: number | bigint + }>( + `SELECT activation_expires_at_ms, activation_generation FROM ${runtime.repository.table("instances")}`, + ), + ) + if (!row) throw new Error("missing instance") + return row +} + +describe("activation lease recovery", () => { + it("renews a kept activation across continuous short passes for five lease durations", async () => { + await setup() + const worker = runtime.worker() + const reference = LeaseCounter.ref("continuous") + LeaseCounter.beforeCommit = () => blockFor(30) + const messages = [] + let firstExpiration = 0 + for (let index = 0; index < 34; index += 1) { + messages.push(await reference.send.increment()) + await worker.runOnce() + if (index === 0) firstExpiration = Number((await storedLease()).activation_expires_at_ms) + } + + expect(events.filter((event) => event.name === "solid_objects.activation.lost")).toEqual([]) + expect(Number((await storedLease()).activation_expires_at_ms)).toBeGreaterThan( + firstExpiration + 800, + ) + expect(Number((await storedLease()).activation_generation)).toBe(1) + for (const [index, message] of messages.entries()) { + expect(await message.status()).toBe("completed") + expect(await message.result()).toBe(index + 1) + expect( + Number((await runtime.repository.messageWithStatus(message.id))?.message.attempt_count), + ).toBe(1) + } + expect(events.filter((event) => event.name === "solid_objects.message.completed")).toHaveLength( + 34, + ) + await worker.stop() + }, 15_000) + + it("renews between short turns in a single pass without timer ticks", async () => { + await setup() + const worker = runtime.worker() + const messages = [] + for (let index = 0; index < 10; index += 1) { + messages.push(await LeaseCounter.ref("draining").send.increment()) + } + LeaseCounter.beforeCommit = () => blockFor(30) + + expect(await worker.runOnce()).toBe(10) + expect(events.filter((event) => event.name === "solid_objects.activation.lost")).toEqual([]) + for (const message of messages) expect(await message.status()).toBe("completed") + await worker.stop() + }) + + it("recovers an expired turn on the same worker with maxAttempts set to one", async () => { + await setup() + const worker = runtime.worker() + const reference = LeaseCounter.ref("expired") + const first = await reference.send.increment() + const second = await reference.send.increment() + LeaseCounter.beforeCommit = () => blockFor(220) + + await worker.runOnce() + expect(await first.status()).not.toBe("completed") + const instance = await runtime.settings.database.connection((connection) => + connection.get<{ state: string }>( + `SELECT state FROM ${runtime.repository.table("instances")}`, + ), + ) + expect(JSON.parse(instance!.state)).toEqual({ count: 0 }) + expect(events.filter((event) => event.name === "solid_objects.activation.lost")).toHaveLength(1) + LeaseCounter.beforeCommit = () => {} + + expect(await worker.runOnce()).toBe(2) + expect(await first.result()).toBe(1) + expect(await second.result()).toBe(2) + expect( + events.filter((event) => event.name === "solid_objects.recovery.reclaimed"), + ).toHaveLength(1) + expect(events.filter((event) => event.name === "solid_objects.message.failed")).toEqual([]) + expect( + events + .filter((event) => event.name === "solid_objects.message.started") + .map((event) => [event.messageId, event.attempt]), + ).toEqual([ + [first.id, 1], + [first.id, 2], + [second.id, 1], + ]) + expect( + Number((await runtime.repository.messageWithStatus(first.id))?.message.attempt_count), + ).toBe(2) + expect(events.filter((event) => event.name === "solid_objects.dead_letter.created")).toEqual([]) + expect(events.filter((event) => event.name === "solid_objects.message.completed")).toHaveLength( + 2, + ) + await worker.stop() + }) + + it("preserves a second worker's takeover when the expired first turn returns", async () => { + await setup() + const firstWorker = runtime.worker() + const secondWorker = runtime.worker() + const reference = LeaseCounter.ref("takeover") + const first = await reference.send.increment() + const second = await reference.send.increment() + LeaseCounter.beforeCommit = async () => { + LeaseCounter.beforeCommit = () => {} + blockFor(220) + expect(await secondWorker.runOnce()).toBe(2) + } + + await firstWorker.runOnce() + expect(await first.result()).toBe(1) + expect(await second.result()).toBe(2) + expect( + events.filter((event) => event.name === "solid_objects.recovery.reclaimed"), + ).toHaveLength(1) + expect(events.filter((event) => event.name === "solid_objects.message.failed")).toEqual([]) + expect( + events + .filter((event) => event.name === "solid_objects.message.started") + .map((event) => [event.messageId, event.attempt]), + ).toEqual([ + [first.id, 1], + [first.id, 2], + [second.id, 1], + ]) + expect( + Number((await runtime.repository.messageWithStatus(first.id))?.message.attempt_count), + ).toBe(2) + expect(events.filter((event) => event.name === "solid_objects.activation.lost")).toHaveLength(1) + expect(events.filter((event) => event.name === "solid_objects.message.completed")).toHaveLength( + 2, + ) + const third = await reference.send.increment() + expect(await secondWorker.runOnce()).toBe(1) + expect(await third.result()).toBe(3) + await firstWorker.stop() + await secondWorker.stop() + }) +})