From 3592c39850a8d6fe6606c01a689d74060e4564c6 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 18 Aug 2026 16:41:55 -0700 Subject: [PATCH 1/3] feat: add internal-message enqueue and snapshot incarnation APIs Give a host package (such as a future commercial scaling layer) two things it cannot build from public API alone: a way to enqueue an internal-delivery-mode actor message that skips user authorization, with a transaction-scoped variant for atomic multi-write commits, and a way to read instance identity/recency alongside a snapshot so a derived write can fence against a stale or superseded actor incarnation. instanceId is a random UUID, not monotonic, so fencing uses createdAtMs instead. --- CHANGELOG.md | 19 +++++ docs/api.md | 7 ++ docs/parity.md | 11 ++- package.json | 2 +- src/index.ts | 7 +- src/runtime.ts | 96 ++++++++++++++++++++- src/version.ts | 2 +- test/internal-messages.test.ts | 114 +++++++++++++++++++++++++ test/process-administration.test.ts | 3 +- test/snapshot-with-incarnation.test.ts | 87 +++++++++++++++++++ 10 files changed, 342 insertions(+), 6 deletions(-) create mode 100644 test/internal-messages.test.ts create mode 100644 test/snapshot-with-incarnation.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 4818f14..8007aff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,24 @@ # Changelog +## 0.14.0 - 2026-08-18 + +- Add `runtime.enqueueInternalMessage()` and + `runtime.enqueueInternalMessageInTransaction(connection, options)`, public + entry points for a host package to enqueue an `internal`-delivery-mode + actor message without going through user authorization. The + transaction-scoped variant accepts a caller-supplied `DatabaseConnection` + so the enqueue can commit atomically alongside other writes in the same + transaction; call the new `runtime.announceInternalMessage(message)` after + that transaction commits to wake worker roles the same way a normal + enqueue does. +- Add `runtime.snapshotWithIncarnation(reference)`, returning the same + authorized fields as `snapshot()` alongside the read instance's + `instanceId`, `revision`, and `createdAtMs`, computed from one shared read. + `instanceId` is a random UUID, not a monotonically increasing value, so a + caller that needs to detect actor recreation (a new incarnation + superseding an old one, regardless of revision) should fence on + `createdAtMs` rather than comparing `instanceId` values directly. + ## 0.13.3 - 2026-08-18 - Lower the supported Node.js floor from 24.15.0 to 24.4.0. Node.js 24.4.0 is diff --git a/docs/api.md b/docs/api.md index 7ac9eaa..ae982e4 100644 --- a/docs/api.md +++ b/docs/api.md @@ -13,6 +13,11 @@ generic signatures; this index explains the supported role of every export. default. - `SolidObjectsRuntime`: installation, registration, supervision, and manager owner. The normal lifecycle is `install()`, `run(signal)`, then `close()`. + `snapshotWithIncarnation(reference)` returns the same authorized fields as + `snapshot()` alongside the read instance's `instanceId`, `revision`, and + `createdAtMs`, computed from the identical read so a caller can fence a + derived write (for example a downstream projection) against a stale or + superseded actor incarnation. - `Actor`: base class providing `ref()`, `actorId`, `currentMessage`, `observables()`, `reject()`, `emit()`, `commitAction()`, `schedule()`, `sendTo()`, and protected lifecycle hooks. @@ -25,6 +30,8 @@ generic signatures; this index explains the supported role of every export. - `ActorClass`, `ActorReference`, `ActorMessageSender`, `ActorSnapshot`, `ActorOperationNames`, `ActorQueryNames`, `StagedOperations`, and `ScheduledOperations`: inferred actor-class and fluent-dispatch types. +- `SnapshotWithIncarnation`: the `{ snapshot, instanceId, revision, +createdAtMs }` shape returned by `SolidObjectsRuntime.snapshotWithIncarnation`. - `MessageReference`: immutable durable message identity with `id`, `requestId`, actor identity, `sequence`, `status()`, `result()`, and `wait()`. - `InvocationOptions`, `AsyncInvocationOptions`, `SnapshotOptions`, and diff --git a/docs/parity.md b/docs/parity.md index 245bcb9..337a841 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -8,7 +8,7 @@ Reference: Ruby `solid_objects` 0.13.3. The JavaScript package began at the Ruby design's `0.12` capability generation; that version number did not imply earlier JavaScript releases. -The Node `0.13.3` implementation has capability parity with that reference. Its +The Node `0.14.0` implementation has capability parity with that reference. Its relational runtime, correctness boundaries, administration, diagnostics, operator dashboard, realtime projections, browser behavior, and supported adapters have native equivalents. Rails-specific rendering surfaces are @@ -16,6 +16,15 @@ replaced by transport- and framework-neutral JavaScript APIs. The partial guard row and the shared planned result-lookup row below are explicit scope boundaries, not missing Ruby capabilities. +`0.14.0` also adds `runtime.enqueueInternalMessage()`, +`runtime.enqueueInternalMessageInTransaction()`, and +`runtime.snapshotWithIncarnation()`. These are Node-only integration points +for a host package (such as a future commercial scaling layer), not ported +Ruby capabilities: Ruby's equivalent primitives (`SolidObjects::Mailbox#enqueue`, +`ActorSnapshot`) are already reachable in-process without a dedicated public +API, since Ruby has no package-privacy boundary between a gem and its own +dependents the way Node's `exports` map enforces one. + ## Status vocabulary - **Native**: the TypeScript runtime provides the capability in a Node-native diff --git a/package.json b/package.json index 30eea2d..0a82256 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "solid-objects", - "version": "0.13.3", + "version": "0.14.0", "description": "Race-free realtime state per application identity, backed by your SQL database", "type": "module", "license": "MIT", diff --git a/src/index.ts b/src/index.ts index 3e97269..34ca19c 100644 --- a/src/index.ts +++ b/src/index.ts @@ -14,7 +14,12 @@ export { type ReminderIntent, type ReminderOptions, } from "./actor.js" -export { configure, createRuntime, SolidObjectsRuntime } from "./runtime.js" +export { + configure, + createRuntime, + SolidObjectsRuntime, + type SnapshotWithIncarnation, +} from "./runtime.js" export { VERSION } from "./version.js" export { guardApplicationDatabase } from "./application-database.js" export { runCli, type CliRunOptions } from "./cli.js" diff --git a/src/runtime.ts b/src/runtime.ts index 41a92c8..fc9c3bc 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -116,6 +116,7 @@ import { waitFor, Worker } from "./worker.js" import { EffectWorker } from "./effect-worker.js" import type { WakeUpRole } from "./wake-up.js" import { withDatabaseDeadline } from "./database/deadline.js" +import type { DatabaseConnection } from "./database/types.js" interface RegisteredActor { actorClass: ActorClass @@ -124,6 +125,13 @@ interface RegisteredActor { queries: ReadonlySet } +export interface SnapshotWithIncarnation { + snapshot: ActorSnapshot + instanceId: string + revision: string + createdAtMs: number +} + interface PayloadProjectionSnapshot { state: JsonObject instanceId: string @@ -683,6 +691,27 @@ export class SolidObjectsRuntime { argumentsValue: {}, authorizationContext: options.authorizationContext, }) + const built = await this.buildSnapshotWithIncarnation(reference) + return built.snapshot + } + + async snapshotWithIncarnation( + reference: ActorReferenceCore, + options: SnapshotOptions = {}, + ): Promise> { + await this.authorize({ + kind: "query", + reference, + operation: "__snapshot__", + argumentsValue: {}, + authorizationContext: options.authorizationContext, + }) + return this.buildSnapshotWithIncarnation(reference) + } + + private async buildSnapshotWithIncarnation( + reference: ActorReferenceCore, + ): Promise> { const registered = this.fetchActor(reference.actorType) const instance = await this.repository.findInstanceByIdentity( reference.actorType, @@ -718,7 +747,12 @@ export class SolidObjectsRuntime { ) { throw new QueryMutatedState("snapshot getters must not mutate actor state or stage work") } - return readonlyCopy(snapshot) as ActorSnapshot + return { + snapshot: readonlyCopy(snapshot) as ActorSnapshot, + instanceId: instance?.id ?? "0", + revision: String(instance?.state_revision ?? 0), + createdAtMs: Number(instance?.created_at_ms ?? 0), + } } async subscriptionSnapshot(options: { @@ -1443,6 +1477,66 @@ export class SolidObjectsRuntime { } } + async enqueueInternalMessage(options: { + actorType: string + actorId: string + operation: string + argumentsValue?: JsonObject + idempotencyKey?: string + }): Promise> { + const { actorType, actorId, operation, argumentsValue = {}, idempotencyKey } = options + const actor = this.fetchActor(actorType) + if (!actor.operations.has(operation)) { + throw new UnknownOperation(`unknown operation ${JSON.stringify(operation)}`) + } + const argumentsObject = jsonObject(argumentsValue, { maxBytes: this.settings.maxPayloadBytes }) + const message = await this.repository.enqueue({ + actorType, + actorId, + operation, + deliveryMode: "internal", + arguments: argumentsObject, + initialState: initialStateFor(actor.definition), + stateVersion: actor.definition.stateVersion, + ...(idempotencyKey === undefined ? {} : { idempotencyKey }), + }) + this.announceInternalMessage(message) + return this.messageReferenceFromRow(message) + } + + async enqueueInternalMessageInTransaction( + connection: DatabaseConnection, + options: { + actorType: string + actorId: string + operation: string + argumentsValue?: JsonObject + idempotencyKey?: string + }, + ): Promise { + const { actorType, actorId, operation, argumentsValue = {}, idempotencyKey } = options + const actor = this.fetchActor(actorType) + if (!actor.operations.has(operation)) { + throw new UnknownOperation(`unknown operation ${JSON.stringify(operation)}`) + } + const argumentsObject = jsonObject(argumentsValue, { maxBytes: this.settings.maxPayloadBytes }) + return this.repository.enqueueInTransaction(connection, { + actorType, + actorId, + operation, + deliveryMode: "internal", + arguments: argumentsObject, + initialState: initialStateFor(actor.definition), + stateVersion: actor.definition.stateVersion, + ...(idempotencyKey === undefined ? {} : { idempotencyKey }), + }) + } + + announceInternalMessage(message: MessageRow): void { + this.emitInstrumentation("message.enqueued", messageInstrumentation(message)) + this.wakeUp("actors") + } + private async enqueue(options: { reference: ActorReferenceCore operation: string diff --git a/src/version.ts b/src/version.ts index 729c18a..d2a1bcf 100644 --- a/src/version.ts +++ b/src/version.ts @@ -1 +1 @@ -export const VERSION = "0.13.3" +export const VERSION = "0.14.0" diff --git a/test/internal-messages.test.ts b/test/internal-messages.test.ts new file mode 100644 index 0000000..497d311 --- /dev/null +++ b/test/internal-messages.test.ts @@ -0,0 +1,114 @@ +import { afterEach, describe, expect, it } from "vitest" +import { Actor } from "../src/actor.js" +import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" +import type { SolidObjectsConfiguration } from "../src/configuration.js" +import { sqlite } from "../src/database/sqlite.js" +import { UnknownOperation } from "../src/errors.js" + +class Counter extends Actor { + static override readonly actorType = "Counter" + + count = 0 + + increment({ amount = 1 }: { amount?: number } = {}): number { + this.count += amount + return this.count + } +} + +let runtime: SolidObjectsRuntime | undefined + +afterEach(async () => { + await runtime?.close() + runtime = undefined +}) + +describe("enqueueInternalMessage", () => { + it("processes an internal message even when authorizeMessage rejects every request", async () => { + runtime = createRuntime(configuredSettings({ authorizeMessage: () => false })) + runtime.register(Counter) + await runtime.install() + + await runtime.enqueueInternalMessage({ + actorType: Counter.actorType, + actorId: "c1", + operation: "increment", + argumentsValue: { amount: 5 }, + }) + + expect(await runtime.worker().runUntilIdle()).toBe(1) + await expect(runtime.ref(Counter, "c1").snapshot()).resolves.toMatchObject({ count: 5 }) + }) + + it("still rejects an operation the actor does not define", async () => { + runtime = createRuntime(configuredSettings()) + runtime.register(Counter) + await runtime.install() + + await expect( + runtime.enqueueInternalMessage({ + actorType: Counter.actorType, + actorId: "c1", + operation: "missing", + }), + ).rejects.toBeInstanceOf(UnknownOperation) + }) +}) + +describe("enqueueInternalMessageInTransaction", () => { + it("enqueues inside a caller-supplied transaction and runs after announce", async () => { + runtime = createRuntime(configuredSettings()) + runtime.register(Counter) + await runtime.install() + const activeRuntime = runtime + + const message = await activeRuntime.settings.database.transaction((connection) => + activeRuntime.enqueueInternalMessageInTransaction(connection, { + actorType: Counter.actorType, + actorId: "c1", + operation: "increment", + argumentsValue: { amount: 3 }, + }), + ) + activeRuntime.announceInternalMessage(message) + + expect(await activeRuntime.worker().runUntilIdle()).toBe(1) + await expect(activeRuntime.ref(Counter, "c1").snapshot()).resolves.toMatchObject({ count: 3 }) + }) + + it("rolls back with the caller's transaction if the surrounding work fails", async () => { + runtime = createRuntime(configuredSettings()) + runtime.register(Counter) + await runtime.install() + const activeRuntime = runtime + + await expect( + activeRuntime.settings.database.transaction(async (connection) => { + await activeRuntime.enqueueInternalMessageInTransaction(connection, { + actorType: Counter.actorType, + actorId: "c1", + operation: "increment", + argumentsValue: { amount: 3 }, + }) + throw new Error("caller failed after enqueueing") + }), + ).rejects.toThrow("caller failed after enqueueing") + + expect(await activeRuntime.worker().runUntilIdle()).toBe(0) + }) +}) + +function configuredSettings( + overrides: Partial = {}, +): SolidObjectsConfiguration { + return { + database: sqlite({ path: ":memory:" }), + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + pollingIntervalMilliseconds: 1, + syncPollingIntervalMilliseconds: 1, + maxAttempts: 2, + ...overrides, + } +} diff --git a/test/process-administration.test.ts b/test/process-administration.test.ts index 4d36871..a00b805 100644 --- a/test/process-administration.test.ts +++ b/test/process-administration.test.ts @@ -3,6 +3,7 @@ import { Actor } from "../src/actor.js" import { sqlite } from "../src/database/sqlite.js" import { Unauthorized } from "../src/errors.js" import { configure, type SolidObjectsRuntime } from "../src/runtime.js" +import { VERSION } from "../src/version.js" class ProcessActor extends Actor { static override readonly actorType = "ProcessActor" @@ -49,7 +50,7 @@ describe("process administration", () => { hostProcessId: process.pid, metadata: { nodeVersion: process.version, - solidObjectsVersion: "0.13.3", + solidObjectsVersion: VERSION, }, shutdownState: "running", shutdownRequestedAt: null, diff --git a/test/snapshot-with-incarnation.test.ts b/test/snapshot-with-incarnation.test.ts new file mode 100644 index 0000000..0660d7f --- /dev/null +++ b/test/snapshot-with-incarnation.test.ts @@ -0,0 +1,87 @@ +import { afterEach, describe, expect, it } from "vitest" +import { Actor } from "../src/actor.js" +import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" +import type { SolidObjectsConfiguration } from "../src/configuration.js" +import { sqlite } from "../src/database/sqlite.js" +import { Unauthorized } from "../src/errors.js" + +class Counter extends Actor { + static override readonly actorType = "Counter" + + count = 0 + + increment({ amount = 1 }: { amount?: number } = {}): number { + this.count += amount + return this.count + } +} + +let runtime: SolidObjectsRuntime | undefined + +afterEach(async () => { + await runtime?.close() + runtime = undefined +}) + +describe("snapshotWithIncarnation", () => { + it("returns the same fields as snapshot() plus instanceId, revision, and createdAtMs", async () => { + runtime = createRuntime(configuredSettings()) + runtime.register(Counter) + await runtime.install() + + const reference = runtime.ref(Counter, "c1") + await reference.increment({ amount: 4 }) + + const plain = await runtime.snapshot(reference) + const withIncarnation = await runtime.snapshotWithIncarnation(reference) + + expect(withIncarnation.snapshot).toEqual(plain) + expect(withIncarnation.instanceId).toEqual(expect.any(String)) + expect(withIncarnation.instanceId.length).toBeGreaterThan(0) + expect(withIncarnation.revision).toEqual(expect.any(String)) + expect(withIncarnation.createdAtMs).toEqual(expect.any(Number)) + expect(withIncarnation.createdAtMs).toBeGreaterThan(0) + }) + + it("gives a recreated actor a fresh instanceId and a createdAtMs no earlier than the original", async () => { + runtime = createRuntime(configuredSettings()) + runtime.register(Counter) + await runtime.install() + + const reference = runtime.ref(Counter, "c1") + await reference.increment({ amount: 1 }) + const first = await runtime.snapshotWithIncarnation(reference) + + await reference.destroy() + await reference.increment({ amount: 1 }) + const second = await runtime.snapshotWithIncarnation(reference) + + expect(second.instanceId).not.toEqual(first.instanceId) + expect(second.createdAtMs).toBeGreaterThanOrEqual(first.createdAtMs) + }) + + it("honors authorizeQuery the same way snapshot() does", async () => { + runtime = createRuntime(configuredSettings({ authorizeQuery: () => false })) + runtime.register(Counter) + await runtime.install() + + const reference = runtime.ref(Counter, "c1") + + await expect(runtime.snapshotWithIncarnation(reference)).rejects.toBeInstanceOf(Unauthorized) + }) +}) + +function configuredSettings( + overrides: Partial = {}, +): SolidObjectsConfiguration { + return { + database: sqlite({ path: ":memory:" }), + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + pollingIntervalMilliseconds: 1, + syncPollingIntervalMilliseconds: 1, + maxAttempts: 2, + ...overrides, + } +} From 8fc551e9a6a8bae69bacf2dfb1294ad118a1515d Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 18 Aug 2026 16:48:02 -0700 Subject: [PATCH 2/3] ci: retrigger a fresh run to rule out a rerun artifact on the floor job From 7dcc8fd3d3c90989ac9ed481d105a8d504b3f4bc Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Wed, 19 Aug 2026 06:50:08 -0700 Subject: [PATCH 3/3] docs: document snapshotWithIncarnation's millisecond boundary createdAtMs orders actor incarnations at the same millisecond granularity every adapter stores created_at_ms at. Destroying and recreating the same actor identity within one database-clock millisecond produces two incarnations a caller cannot order by createdAtMs alone, since neither timestamp precision nor instanceId (a random UUID) can break the tie without a schema-level monotonic sequence, which is out of scope here. Record the boundary in docs/correctness.md and docs/api.md, and make the existing recreation test's title state why it asserts >= rather than >. --- CHANGELOG.md | 4 ++++ docs/api.md | 5 ++++- docs/correctness.md | 8 ++++++++ test/snapshot-with-incarnation.test.ts | 2 +- 4 files changed, 17 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8007aff..db21465 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,10 @@ caller that needs to detect actor recreation (a new incarnation superseding an old one, regardless of revision) should fence on `createdAtMs` rather than comparing `instanceId` values directly. + `createdAtMs` orders incarnations at millisecond granularity; destroying + and recreating the same actor identity within the same millisecond + produces two incarnations a caller cannot order by `createdAtMs` alone — + see `docs/correctness.md`. ## 0.13.3 - 2026-08-18 diff --git a/docs/api.md b/docs/api.md index ae982e4..63d899a 100644 --- a/docs/api.md +++ b/docs/api.md @@ -17,7 +17,10 @@ generic signatures; this index explains the supported role of every export. `snapshot()` alongside the read instance's `instanceId`, `revision`, and `createdAtMs`, computed from the identical read so a caller can fence a derived write (for example a downstream projection) against a stale or - superseded actor incarnation. + superseded actor incarnation. `createdAtMs` orders incarnations at + millisecond granularity; see + [Limitations and non-goals](correctness.md#limitations-and-non-goals) for + the same-millisecond boundary. - `Actor`: base class providing `ref()`, `actorId`, `currentMessage`, `observables()`, `reject()`, `emit()`, `commitAction()`, `schedule()`, `sendTo()`, and protected lifecycle hooks. diff --git a/docs/correctness.md b/docs/correctness.md index 9f32c27..1a67d1b 100644 --- a/docs/correctness.md +++ b/docs/correctness.md @@ -71,6 +71,14 @@ placement, capacity, database backups, and database failover. - Redis and PostgreSQL notifications reduce wake-up latency but do not replace durable polling or become a source of truth. +- `snapshotWithIncarnation`'s `createdAtMs` orders actor incarnations at + millisecond granularity, the same precision every adapter stores + `created_at_ms` at. Destroying and recreating the same actor identity + within the same database-clock millisecond produces two incarnations with + an equal `createdAtMs`; a caller fencing a derived write on it cannot + distinguish which of the two is current in that narrow case. `instanceId` + still changes and detects that a recreation happened; it is a random UUID + and carries no order of its own. - Large documents, bulk pipelines, globally placed edge state, and global counters are outside the intended workload. Prefer an ordinary row transaction when it completely enforces the invariant. diff --git a/test/snapshot-with-incarnation.test.ts b/test/snapshot-with-incarnation.test.ts index 0660d7f..9d3481b 100644 --- a/test/snapshot-with-incarnation.test.ts +++ b/test/snapshot-with-incarnation.test.ts @@ -43,7 +43,7 @@ describe("snapshotWithIncarnation", () => { expect(withIncarnation.createdAtMs).toBeGreaterThan(0) }) - it("gives a recreated actor a fresh instanceId and a createdAtMs no earlier than the original", async () => { + it("gives a recreated actor a fresh instanceId; createdAtMs only orders it at millisecond granularity", async () => { runtime = createRuntime(configuredSettings()) runtime.register(Counter) await runtime.install()