diff --git a/CHANGELOG.md b/CHANGELOG.md index db21465..d3af3ff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,82 @@ # Changelog -## 0.14.0 - 2026-08-18 +## 0.14.0 - 2026-08-22 + +- Add `solid-objects/database/shared-sqlite-wasm`, the transparent + multi-tab database. Every tab runs an ordinary + `configure -> install -> ref` flow against the same shared database; the + adapter elects one holder per origin with the Web Locks API, sends every + other tab's SQL over a `BroadcastChannel` session to the holder, and + fails over onto the same OPFS state when the holder's tab dies. The + runtime's leases and fencing arbitrate the tabs' workers exactly as they + arbitrate Node processes. A Playwright test proves plain actor references + incrementing one durable counter from two tabs with failover. +- Add `solid-objects/browser/tab-host`, the multi-tab host that completes + milestone M3 of the in-browser runtime plan + ([#17](https://github.com/cardmagic/solid-objects-js/issues/17)). Every + tab starts a candidate host; the Web Locks API elects one leader per + origin, and only the leader opens the database and runs the runtime. + Tabs invoke actors through a `BroadcastChannel` client that retries with + idempotent request ids. When the leader's tab dies, the lock releases, + the next host promotes, and the runtime continues from the same OPFS + state. A Playwright test proves shared state across two tabs and + failover after the leader closes. +- Share the transmit wire contract with the Ruby gem. The golden fixture + file `compatibility/transmit-envelopes.json` is committed to both + repositories with a consuming test on each side; `receiveTransmitEnvelope` + now defaults a missing `arguments` to an empty object, matching the Ruby + ingest and the staging side. +- Add `solid-objects/transmit`, milestone M4 of the plan. An actor + stages a transmit intent with `this.transmit().operation(arguments)` (or with + `emit(TRANSMIT_EFFECT, ...)` for a different target) in the same + transaction as its state change. `registerTransmit` drains the outbox + with at-least-once delivery and per-actor order (an ordered drain up to + the claimed effect's mailbox sequence), and `receiveTransmitEnvelope` gives + the server an idempotent ingest keyed on the effect id. Vitest covers + order under transmit failures, replay deduplication, and recovery after + an offline period; a Playwright test drains a browser outbox into the + Node server runtime. +- Change `TurnContextStore` to synchronous scoping. The store no longer + stays set across `await` boundaries, so an open transaction scope cannot + leak into interleaved tasks and trip the inside-transaction guards. In + the browser those guards are best-effort; Node keeps full + `AsyncLocalStorage` semantics. +- Retry OPFS SAH pool acquisition. The upstream module caches a failed + initialization; the adapter now passes `forceReinitIfPreviouslyFailed`, + so a new leader can claim the pool after the old tab dies. + +- Add a platform seam for async context propagation + (`src/platform/context-store.ts`). Shared modules no longer import + `node:async_hooks` directly. Node entry points register an + `AsyncLocalStorage` factory. A `TurnContextStore` gives a browser host a + turn-scoped store for serialized actor turns. This is milestone M1 of the + in-browser runtime plan + ([#17](https://github.com/cardmagic/solid-objects-js/issues/17)). +- Route UUID generation through `src/platform/uuid.ts`, which uses the + standard `crypto.randomUUID()`. Shared modules no longer import + `node:crypto`. +- Add `check:browser-imports` to `pnpm run check`. The script walks the + import graph of the browser-safe modules and fails when a `node:` module + or a server-only driver reaches that graph. +- Add `solid-objects/browser/host`, the entry point for a runtime host + inside a browser worker. An import registers the browser platform: a + turn-scoped context store and a browser host identity. The module + re-exports the core runtime API and the WASM adapter. A Playwright test + runs the full runtime in a Chromium module worker and proves durable + actor state across a page reload. This is the first stage of milestone M3 + ([#17](https://github.com/cardmagic/solid-objects-js/issues/17)). +- Replace the `Buffer.byteLength` payload size check with `TextEncoder`, so + serialization works without the Node `Buffer` global. +- Route the process identity (hostname, process id, runtime version) + through `src/platform/host-identity.ts`. The repository no longer imports + `node:os` or reads `process.pid` directly. +- Add `solid-objects/database/sqlite-wasm`, a browser-safe `Database` + adapter on `@sqlite.org/sqlite-wasm` (an optional peer dependency). The + full runtime passes its round-trip test against this adapter in Node, and + a Playwright test proves transactions, rollback, and OPFS persistence + across a page reload in Chromium. This is milestone M2 of the in-browser + runtime plan + ([#17](https://github.com/cardmagic/solid-objects-js/issues/17)). - Add `runtime.enqueueInternalMessage()` and `runtime.enqueueInternalMessageInTransaction(connection, options)`, public diff --git a/README.md b/README.md index 7b63f03..13140d1 100644 --- a/README.md +++ b/README.md @@ -17,11 +17,16 @@ Solid Objects keeps the state, the queued operations, the retries, the reminders, the effects, and the realtime invalidations in the database the application already operates. +The same runtime also runs inside a browser worker on SQLite WASM, with +durable actor state in the origin's private file system. See +[Solid Objects in the browser](#solid-objects-in-the-browser). + > **Early release:** the correctness core has automated coverage. That coverage -> includes the supported databases, the Chromium browser client, process -> recovery, and the packaged artifacts. The TypeScript implementation is still -> new. There is one deployed first-party reference application. There is no -> measured scale and no third-party production use yet. Read the +> includes the supported databases, the Chromium browser client, the browser +> runtime, process recovery, and the packaged artifacts. The TypeScript +> implementation is still new. There is one deployed first-party reference +> application. There is no measured scale and no third-party production use +> yet. Read the > [delivery boundaries](#delivery-boundaries) before you use it for important > data. @@ -284,6 +289,59 @@ personalized payloads, and framework-neutral component refresh. Applications provide authentication, WebSocket transport, and rendering. See the [browser protocol](docs/browser-protocol.md) and [authorization guide](docs/authorization.md). +## Solid Objects in the browser + +The full runtime runs inside a browser module worker. Actors look exactly +like they do in Node; the database is SQLite WASM, and persistent storage +lives in the origin's private file system (OPFS), so actor state survives +page reloads. + +```javascript +import { Actor, configure, sharedSqliteWasm } from "solid-objects/browser/host" + +class Counter extends Actor { + static actorType = "Counter" + + count = 0 + + increment({ amount = 1 } = {}) { + this.count += amount + return this.count + } +} + +const runtime = configure({ + database: sharedSqliteWasm({ path: "app.db" }), + authorizeMessage: () => true, + authorizeQuery: () => true, +}) +await runtime.install() + +await Counter.ref("page-hits").increment() +``` + +That code runs identically in every tab. `sharedSqliteWasm` elects one +database holder per origin through the Web Locks API, carries the other +tabs' SQL to it over a `BroadcastChannel`, and fails over onto the same +durable state when the holder's tab dies. Use `sqliteWasm` directly for a +single dedicated worker. + +Two companions complete the local-first story: + +- `solid-objects/browser/tab-host` runs one runtime for all tabs when the + application prefers request-level routing: the leader's worker executes + every operation, and other tabs invoke through a `BroadcastChannel` client + by name. +- `solid-objects/transmit` drains the transactional effects outbox to a + server runtime with at-least-once delivery, per-actor order, and an + idempotent server ingest, so offline writes reconcile when the network + returns. + +The wire shapes are documented in the +[browser protocol](docs/browser-protocol.md), the API in the +[public API reference](docs/api.md), and the platform boundaries in +[supported versions](docs/support.md). + ## Comparison These systems solve different coordination problems. The table describes their @@ -322,8 +380,10 @@ and operational data access. - TypeScript 5.9 or newer for TypeScript applications - SQLite through `node:sqlite`, PostgreSQL 14 or newer, or MySQL 8.0 or newer with InnoDB -- optional `pg`, `mysql2`, or `redis` peer dependency only for the selected - adapter +- optional `pg`, `mysql2`, `redis`, or `@sqlite.org/sqlite-wasm` peer + dependency only for the selected adapter +- for the browser runtime: a browser with OPFS for persistent storage and the + Web Locks API for the multi-tab host [Supported versions](docs/support.md) records the exact CI matrix and the boundaries. diff --git a/compatibility/transmit-envelopes.json b/compatibility/transmit-envelopes.json new file mode 100644 index 0000000..74aa5d1 --- /dev/null +++ b/compatibility/transmit-envelopes.json @@ -0,0 +1,89 @@ +{ + "description": "Golden transmit envelopes shared by solid_objects and solid-objects-js. The JS bridge must produce and accept these; SolidObjects::Transmission.receive must accept the valid ones, apply the duplicate pair once, and reject the malformed ones.", + "valid": [ + { + "name": "increment with arguments", + "envelope": { + "effectId": "fixture-effect-0001", + "actorType": "transmit-counters", + "actorId": "fixture-counter", + "operation": "increment", + "arguments": { "amount": 2 } + }, + "idempotencyKey": "transmit:fixture-effect-0001" + }, + { + "name": "increment without arguments", + "envelope": { + "effectId": "fixture-effect-0002", + "actorType": "transmit-counters", + "actorId": "fixture-counter", + "operation": "increment" + }, + "idempotencyKey": "transmit:fixture-effect-0002" + } + ], + "duplicatePair": [ + { + "effectId": "fixture-effect-0003", + "actorType": "transmit-counters", + "actorId": "fixture-counter", + "operation": "increment", + "arguments": { "amount": 1 } + }, + { + "effectId": "fixture-effect-0003", + "actorType": "transmit-counters", + "actorId": "fixture-counter", + "operation": "increment", + "arguments": { "amount": 1 } + } + ], + "malformed": [ + { + "name": "missing effectId", + "envelope": { + "actorType": "transmit-counters", + "actorId": "fixture-counter", + "operation": "increment" + } + }, + { + "name": "empty actorId", + "envelope": { + "effectId": "fixture-effect-0004", + "actorType": "transmit-counters", + "actorId": "", + "operation": "increment" + } + }, + { + "name": "snake_case keys", + "envelope": { + "effect_id": "fixture-effect-0005", + "actor_type": "transmit-counters", + "actor_id": "fixture-counter", + "operation": "increment" + } + }, + { + "name": "non-string operation", + "envelope": { + "effectId": "fixture-effect-0006", + "actorType": "transmit-counters", + "actorId": "fixture-counter", + "operation": 7 + } + }, + { + "name": "arguments not an object", + "envelope": { + "effectId": "fixture-effect-0007", + "actorType": "transmit-counters", + "actorId": "fixture-counter", + "operation": "increment", + "arguments": [1] + } + } + ] +} diff --git a/docs/api.md b/docs/api.md index ca32e0d..a41cc23 100644 --- a/docs/api.md +++ b/docs/api.md @@ -21,8 +21,8 @@ generic signatures; this index explains the supported role of every export. [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. + `observables()`, `reject()`, `emit()`, `transmit()`, `commitAction()`, + `schedule()`, `sendTo()`, and protected lifecycle hooks. - `broadcastValue(value)`: mark an observable so its changed value enters the durable invalidation envelope. - `broadcastInvalidation(value)`: compare the real observable value but put @@ -275,6 +275,49 @@ and `SyncTimeoutWaitingOn` type timeout diagnostics. See - `SQLiteDatabase`: `Database` implementation and `close()` owner. - `SQLiteDatabaseOptions`: path, busy timeout, and lock retry options. +## `solid-objects/database/sqlite-wasm` + +- `sqliteWasm(options)`: construct `SQLiteWasmDatabase` asynchronously. The + first call loads the `@sqlite.org/sqlite-wasm` module. +- `SQLiteWasmDatabase`: `Database` implementation on SQLite WASM and `close()` + owner. It runs in a browser and in Node. One host owns the database file; + the adapter serializes access on one connection. +- `SQLiteWasmDatabaseOptions`: `path` plus a `storage` mode. `"temporary"` + (the default) keeps data for the life of the process or page. + `"persistent"` stores data in the browser origin's OPFS through the SQLite + SAH pool VFS, and fails fast where OPFS is unavailable. + +## `solid-objects/database/shared-sqlite-wasm` + +Many browser tabs, one database, no visible infrastructure. Every tab +constructs the same shared database and runs an ordinary +`configure → install → ref` flow; the adapter hides the coordination. The +Web Locks API elects one holder per origin. The holder opens the real +SQLite WASM database; every other instance sends its SQL over a +`BroadcastChannel` session to the holder, through the same serialized +access queue. When the holder dies, the lock releases, the next instance +opens the pool, and the runtime's leases and fencing arbitrate the tabs' +workers exactly as they arbitrate Node processes. + +- `sharedSqliteWasm(options)`: construct `SharedSQLiteWasmDatabase`. +- `SharedSQLiteWasmDatabase`: `Database` implementation with a `role()` + probe (`connecting`, `holder`, or `remote`) and `close()`. +- `SharedSQLiteWasmDatabaseOptions`: `path`, an optional election `name` + (defaults to the path), the `storage` mode (persistent by default except + for `:memory:`), and the request, retry, session-idle, and open-attempt + tuning knobs. +- `SharedDatabaseFailover`: the retryable rejection an in-flight statement + receives when the holder changes mid-operation. A session that has not + executed a statement yet retries automatically; anything later surfaces, + because replaying partially executed work is not safe. +- `SharedDatabaseUnavailable`: the rejection for a closed instance, a + timed-out request, or an idle session the holder reclaimed. + +A transaction that dies with its holder rolls back with the pool, which is +the same at-least-once story as a crashed Node process. Sessions that stay +idle longer than `sessionIdleTimeoutMilliseconds` (default 10 seconds) are +reclaimed so a dead tab cannot hold the database hostage. + ## `solid-objects/database/postgresql` - `postgresql(options)`: construct `PostgreSQLDatabase`. @@ -323,6 +366,132 @@ component registry reacts to names in either location. The wire format, trust boundary, revision rules, and component semantics are in [Browser protocol](browser-protocol.md). +## `solid-objects/browser/host` + +The entry point for a runtime host inside a browser worker. An import of +this module registers the browser platform: a turn-scoped context store and +a browser host identity. Do not import it in the same process as the Node +entry points; the last registration wins. + +- Re-exports `Actor`, `broadcastInvalidation`, `broadcastValue`, `configure`, + `createRuntime`, `SolidObjectsRuntime`, and `VERSION` from the core, and + `sqliteWasm`, `SQLiteWasmDatabase`, and `SQLiteWasmDatabaseOptions` from + the WASM adapter, so a worker needs one import. +- The turn-scoped context store expects serialized actor turns. One worker + hosts one runtime. A page talks to that worker through messages, not + through direct actor references. +- The store scopes only the synchronous part of a callback and restores + the previous scope in strict stack order, so an interleaved task never + observes another turn's scope. The cost of that isolation: after the + first `await` inside an actor operation, `currentActor()`, + `applicationWritesForbidden()`, and the database deadline read as unset. + Keep guarded application-database writes in synchronous actor code or in + commit actions; Node keeps full `AsyncLocalStorage` propagation. +- Alarms and reminders fire only while the hosting worker is alive. + +## `solid-objects/browser/tab-host` + +Many tabs, one runtime. Each tab starts a candidate host; the Web Locks API +elects one leader per origin. The leader starts the runtime, runs its +workers, and serves invocations from every tab over a `BroadcastChannel`. +When the leader's tab dies, the lock releases and the next host promotes. + +- `startTabHost(options)`: join the election. `TabHostOptions` carries the + election `name` and a `startRuntime` callback; the callback runs only on + promotion, so a follower never opens the database. It returns a + `TabHostRuntimeHandle` with the runtime and an optional `close`. +- `TabHost`: `role()`, `leadership()` (a promise that resolves on + promotion), and `close()`. +- `connectTabClient(options)`: connect from any tab. `TabClientOptions` + carries the election `name` plus retry and timeout intervals. +- `TabClient.invoke(invocation)`: send a `TabInvocation` (`actorType`, + `actorId`, `operation`, `arguments`). The client retries until a leader + answers; the leader enqueues with the request id as the idempotency key, + so a resend applies once. +- `TabInvocationTimeout` and `TabInvocationFailed`: the client-side errors. + +The election needs the Web Locks API. Every current browser provides it; +Node provides `navigator.locks` from 24.5, so Node-side use of this module +needs a newer Node than the package floor. `startTabHost` fails fast with a +clear error where the API is missing. + +A tab dies without a clean shutdown, so failover speed follows the fence +settings. Give the browser runtime a short `leaseDurationMilliseconds` and +`processAliveThresholdMilliseconds` (for example 750), with a +`leaseRenewalIntervalMilliseconds` below the lease (for example 250), so a +new leader reclaims a dead tab's activations before sync invocations time +out. When `startRuntime` fails, close the database in a catch block; an +open SAH pool otherwise blocks the next candidate until the worker dies. + +## `solid-objects/transmit` + +The transactional outbox bridge between a local runtime and a server +runtime. An actor stages a transmit intent with `this.transmit()` +in the same transaction as its state change. The effect worker drains the +outbox with at-least-once delivery, per-actor order, and retry backoff. + +- `actor.transmit()`: the fluent staging surface. `this.transmit().increment( +{ amount })` stages a transmit intent that replays the operation on the + server twin of the same actor, in the same transaction as the local + state change. +- `TRANSMIT_EFFECT`: the effect name (`solid-objects.transmit`) underneath + `transmit()`. Stage it directly with `emit()` when the target differs from + the source: the staged arguments hold `operation`, `arguments`, and an + optional target `actorType` and `actorId`. +- `registerTransmit(options)`: register the drain handler on the local + runtime. `RegisterTransmitOptions` carries the runtime and a `deliver` + callback that carries a `TransmitEnvelope` to the server; throw from + `deliver` while offline and the effect retries with backoff. Give a + browser runtime a generous `maxAttempts`; an effect that exhausts its + attempts during a long offline period lands in dead letters, and + `runtime.deadLetters.retry` re-queues it. +- `receiveTransmitEnvelope(options)`: idempotent server ingest. `arguments` + is optional in the envelope and defaults to an empty object, matching the + staging side and the Ruby ingest. It enqueues an + internal message with `transmit:` as the idempotency key, so a + replayed envelope applies once. The host must authenticate the sender + before this call; internal delivery skips `authorizeMessage`. The call + belongs inside whatever route the host application gives the transmit + callback to post to: + + ```typescript + import { IdempotencyConflict, InvalidPayload, receiveTransmitEnvelope } from "solid-objects" + + async function handleSyncRoute(request: Request): Promise { + const sender = await authenticate(request) + if (!sender) return new Response("Forbidden", { status: 403 }) + try { + await receiveTransmitEnvelope({ runtime, envelope: await request.json() }) + return Response.json({}) + } catch (error) { + if (error instanceof InvalidPayload || error instanceof IdempotencyConflict) { + return new Response(null, { status: 422 }) + } + throw error + } + } + ``` + + The example uses a Fetch-style handler; any HTTP framework works. The 422 + matters: it tells the sending outbox to dead-letter the effect instead of + retrying it. `InvalidPayload` marks a malformed envelope, and + `IdempotencyConflict` marks a replay whose arguments changed; both are + permanently unappliable, and a 500 would make the outbox retry them + forever. + +- Per-actor order comes from an ordered drain: a claimed transmit effect + transmits every undelivered envelope for its actor up to its own mailbox + sequence, oldest first. A duplicate transmission is safe; the server + deduplicates by effect id. Run one effect worker per local runtime for + the order guarantee. +- `InvalidTransmitEnvelope`: the non-retryable rejection for malformed staged + arguments; the effect dead-letters instead of retrying forever. + +The tab host and transmit modules are browser-safe and also run in Node. +The transmit wire contract is shared with the Ruby gem +([solid-objects-ruby#49](https://github.com/cardmagic/solid-objects-ruby/pull/49)); +`compatibility/transmit-envelopes.json` pins it in both repositories. + ## `solid-objects/web` - `createDashboard(options)` creates an immutable `SolidObjectsDashboard` with diff --git a/docs/architecture.md b/docs/architecture.md index 24e51c6..645af2d 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -41,8 +41,10 @@ transactions immediately. PostgreSQL and MySQL use bounded pools. They keep each transaction on one checked-out client. They store timestamps and sequences as 64-bit integers. They lock an actor's instance row during mailbox sequence allocation. MySQL creates InnoDB tables and retries only the side-effect-free -enqueue transaction when InnoDB chooses it as a deadlock victim. Every adapter -uses database time and the same fencing predicates. +enqueue transaction when InnoDB chooses it as a deadlock victim. SQLite WASM +serializes one in-process connection the same way as the Node SQLite adapter. +It stores its file in the browser origin's OPFS through the SAH pool VFS. +Every adapter uses database time and the same fencing predicates. Synchronous invocation carries a monotonic deadline into adapter operations. SQLite bounds its process-local access queue and busy timeout. Outside a caller @@ -56,11 +58,14 @@ JavaScript actor code that already runs is cooperative. The runtime does not preempt it by force. If it outlives the caller's wait, leases and fenced commits stay authoritative. -Each database adapter also tracks its active transaction through Node's async -context. A committed call or message wait fails early when the same logical call -stack already owns a Solid Objects transaction. It fails before enqueue or -polling. It does not wait for a connection or a serialized SQLite slot that it -cannot release. +Each database adapter also tracks its active transaction through a platform +context store: `AsyncLocalStorage` in Node, and a turn-scoped store in the +browser that covers only synchronous code. A committed call or message wait +fails early when the same logical call stack already owns a Solid Objects +transaction. It fails before enqueue or polling. It does not wait for a +connection or a serialized SQLite slot that it cannot release. The SQLite WASM +adapter also captures each operation's absolute deadline at access entry, so +deadline enforcement does not depend on ambient context after an `await`. PostgreSQL notifications are an opt-in latency layer. One event-driven client per runtime listens on role-specific channels. It listens before the worker @@ -121,5 +126,24 @@ broadcast events through an application-owned shared transport and calls another broker optional for a single Node process while making the cross-process boundary explicit. -Solid Objects provides database-backed state coordination for Node.js. It does -not reproduce Cloudflare's placement or edge-runtime guarantees. +The browser hosts the same runtime through three layers. A platform seam +supplies what Node builtins supplied before: a context store, a host identity, +and UUID generation; `solid-objects/browser/host` registers the browser +implementations on import. A dedicated module worker owns the runtime and the +OPFS database, because OPFS sync access handles exist only in dedicated +workers. Across tabs, `solid-objects/browser/tab-host` elects one leader per +origin through the Web Locks API; only the leader opens the database and runs +workers, other tabs invoke through a versioned `BroadcastChannel` protocol +with idempotent request ids, and a dead leader's lock release promotes the +next candidate onto the same durable state. + +`solid-objects/transmit` connects a local runtime to a server runtime +through the existing effects outbox. An actor stages a transmit intent in the same +transaction as its state change; the effect worker drains the outbox with +at-least-once delivery and retry backoff. Per-actor order comes from an +ordered drain up to the claimed effect's mailbox sequence, and the server +ingest deduplicates on the effect id through the normal idempotency-key path. + +Solid Objects provides database-backed state coordination for Node.js and the +browser. It does not reproduce Cloudflare's placement or edge-runtime +guarantees. diff --git a/docs/browser-protocol.md b/docs/browser-protocol.md index 0546049..aae360b 100644 --- a/docs/browser-protocol.md +++ b/docs/browser-protocol.md @@ -105,3 +105,62 @@ several WebSocket processes, the configured `broadcast` callback publishes the committed envelope through a shared transport and each process passes received envelopes to `runtime.realtime.publish()`. The session fence safely drops the duplicate seen by a process that both claimed and received the same event. + +## Tab host channel protocol + +`solid-objects/browser/tab-host` uses a second, unrelated wire surface: a +`BroadcastChannel` between tabs of one origin. Every envelope carries +`protocol: "solid-objects-tab-host"` and `version: 1`; a listener ignores +anything else. Three kinds exist: + +- `invoke`: a client request with a `requestId` (a UUID the client generates), + the target `actorType`, `actorId`, `operation`, and a JSON `arguments` + object. +- `result`: the leader's answer for one `requestId`, with either an `ok` + value or a named error. +- `leader-online`: the announcement a new leader posts on promotion. Clients + re-post their pending requests when they see it. + +The client retries an `invoke` on an interval until a `result` arrives or its +timeout passes. The leader enqueues each request with `tab:` as the +idempotency key, so a retried or re-posted request applies once. The channel +is same-origin plumbing between the application's own tabs; it carries no +authentication, so the trust boundary is the origin. + +## Sync envelope + +`solid-objects/transmit` transmits one JSON envelope per staged transmit +effect: `effectId`, target `actorType` and `actorId`, `operation`, and an +optional `arguments` object that defaults to an empty object on both +ingests. The transport belongs to the host application; the +Playwright suite posts envelopes over `fetch`. The server calls +`receiveTransmitEnvelope`, which enqueues an internal message with +`transmit:` as the idempotency key, so a replayed envelope applies once. +Internal delivery skips `authorizeMessage`; the host must authenticate the +sender before that call. + +## Shared database channel protocol + +`solid-objects/database/shared-sqlite-wasm` uses a third wire surface: a +`BroadcastChannel` that carries SQL sessions from every tab to the current +database holder. Every envelope carries +`protocol: "solid-objects-shared-sqlite"` and `version: 1`. Seven kinds +exist: + +- `ping` and `pong`: holder discovery. A new instance pings until a holder + answers with its `epoch`. +- `holder-online`: the announcement a new holder posts on promotion, with a + fresh `epoch`. +- `open`: start a session (`connection` or `transaction`) with a client + `sessionId`. +- `statement`: one `run`, `get`, `all`, or `now` operation inside a session. +- `close`: finish a session with `commit`, `rollback`, or `end`. +- `result`: the holder's answer for one `requestId`. + +Every request carries the `epoch` it targets. A holder rejects requests from +another epoch, so a client learns about a failover from a fast rejection +rather than a timeout. A session that has not executed a statement retries +against the new holder automatically; later failures surface as +`SharedDatabaseFailover`, because a partially executed session must not +replay. The channel is same-origin plumbing with the origin as its trust +boundary, the same as the tab host protocol. diff --git a/docs/correctness.md b/docs/correctness.md index 9f09bf2..ab9a748 100644 --- a/docs/correctness.md +++ b/docs/correctness.md @@ -71,6 +71,12 @@ 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. +- The browser platform has no `AsyncLocalStorage`. Its context store covers + only the synchronous part of a callback. After the first `await` inside an + actor operation, the ambient guards (`applicationWritesForbidden()` and the + inside-transaction check) read as unset. Durable-state fencing, mailbox + ordering, and the SQLite WASM deadline enforcement do not depend on those + guards. The guards are best-effort in the browser and exact in Node. - `snapshotWithIncarnation`'s `createdAtMs` orders actor incarnations to the millisecond. Every adapter stores `created_at_ms` at that same precision. If you destroy and recreate the same actor identity inside one database-clock diff --git a/docs/fit.md b/docs/fit.md index 99b9670..aa0f41a 100644 --- a/docs/fit.md +++ b/docs/fit.md @@ -14,6 +14,9 @@ or realtime projections. invalidations atomically. - The application already operates SQLite, PostgreSQL, or MySQL and should keep durable coordination there. +- A local-first application needs the same actor model in the browser: + durable per-user state on SQLite WASM, one runtime shared across tabs, and + an outbox that syncs to a server runtime when the network allows. ## Prefer a row transaction when diff --git a/docs/parity.md b/docs/parity.md index 20fbeee..2fe9239 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -107,6 +107,67 @@ gap between them. | Personalized payload broadcasts | Native | Static typed projections run against committed state under each fresh subscriber context, reauthorize as queries, isolate failures, and carry independent revision fences. | | Real-browser compatibility suite | Native | Playwright exercises subscription replay over native WebSocket, incarnation/revision fences, payload delivery, component batching, and cancellation in Chromium. | +## JavaScript-only: the browser runtime + +`0.14.0` ships an in-browser runtime with SQLite WASM storage +([#17](https://github.com/cardmagic/solid-objects-js/issues/17)). The +runtime itself is a JavaScript-only capability: the Ruby gem has no browser +target, so no Ruby parity row exists for milestones M1 through M3. The +transmit family (milestone M4) started here but is not JavaScript-only; the +next section tracks it as a shared capability. All four milestones are +complete: + +- M1: the shared modules no longer import Node built-in modules, a + registered platform factory supplies async context propagation, and + `pnpm run check` enforces a Node-free import graph for the browser-safe + modules. +- M2: `solid-objects/database/sqlite-wasm` implements the `Database` + contract on SQLite WASM. The full runtime passes a round-trip test + against it, and Playwright proves OPFS persistence across a page reload. +- M3: `solid-objects/browser/host` hosts the full runtime in a browser + module worker on OPFS storage, with durable actor state across page + reloads proven in Chromium. `solid-objects/browser/tab-host` elects one + leader per origin with the Web Locks API and serves every tab over a + `BroadcastChannel`; Playwright proves shared state across two tabs and + failover with durable continuation after the leader tab closes. + `solid-objects/database/shared-sqlite-wasm` goes further: it moves the + election behind the `Database` seam, so every tab runs the ordinary + `configure -> install -> ref` flow and the runtime's own leases and + fencing arbitrate the tabs' workers. The plan + named a `SharedWorker` as the host; Web Locks election between dedicated + workers replaced it, because OPFS sync access handles exist only in + dedicated workers. +- M4: `solid-objects/transmit` drains the local effects outbox to a + server runtime with at-least-once delivery, per-actor order, and an + idempotent server ingest. Vitest proves order under transmit failures, + replay deduplication, and recovery after an offline period, on SQLite, + PostgreSQL, and MySQL. + +## Shared capability: the transmit family + +The transmit family is the one part of the browser work that both runtimes +share. The Ruby gem adopts it in +[solid-objects-ruby#49](https://github.com/cardmagic/solid-objects-ruby/pull/49) +(proposals [#47](https://github.com/cardmagic/solid-objects-ruby/issues/47) +and [#48](https://github.com/cardmagic/solid-objects-ruby/issues/48)): +`SolidObjects::Transmission.receive` is the ingest, and `Actor#transmit` +with `register_transmit` is the staging side. Identifiers differ by +runtime idiom; the wire contract is what both sides guarantee: + +- envelope keys are camelCase (`effectId`, `actorType`, `actorId`, + `operation`, and an optional `arguments` that defaults to an empty + object); +- the ingest idempotency key is `transmit:`, byte for byte; +- a replay with changed arguments raises the idempotency conflict on both + sides and leaves the first application intact. + +`compatibility/transmit-envelopes.json` is committed to both repositories +with a consuming test on each side, so the contract is enforced from both +sides of the repository boundary. Manual cross-runtime QA (Node to Rails +and Rails to Node) ran in solid-objects-ruby#49; the one disagreement it +found (the optional `arguments` default) is fixed and pinned by the shared +fixture. + ## Rails-specific surfaces Rails generators, Active Record models/controllers, Turbo rendering, and diff --git a/docs/support.md b/docs/support.md index c9c87e4..9562f10 100644 --- a/docs/support.md +++ b/docs/support.md @@ -2,18 +2,32 @@ ## Runtime support -| Component | Supported or tested range | -| -------------- | ----------------------------------------------------------- | -| Node.js | 24.4.0 or newer; CI runs 24.4.0 and 24.15.0 | -| TypeScript | 5.9 or newer for TypeScript applications | -| SQLite | Node's built-in `node:sqlite` on the supported Node runtime | -| PostgreSQL | 14 or newer; CI runs 14 and 18 | -| MySQL | 8.0 or newer with InnoDB; CI runs 8.0 and 8.4 | -| Redis wake-up | Optional; CI runs Redis 7 | -| Browser client | Chromium through Playwright | - -The package is ESM-only. PostgreSQL, MySQL, and Redis require their optional -peer dependency. SQLite has no driver dependency beyond Node.js. +| Component | Supported or tested range | +| --------------- | ----------------------------------------------------------- | +| Node.js | 24.4.0 or newer; CI runs 24.4.0 and 24.15.0 | +| TypeScript | 5.9 or newer for TypeScript applications | +| SQLite | Node's built-in `node:sqlite` on the supported Node runtime | +| PostgreSQL | 14 or newer; CI runs 14 and 18 | +| MySQL | 8.0 or newer with InnoDB; CI runs 8.0 and 8.4 | +| Redis wake-up | Optional; CI runs Redis 7 | +| Browser client | Chromium through Playwright | +| SQLite WASM | `@sqlite.org/sqlite-wasm` 3.50 or newer; optional | +| Browser runtime | Chromium through Playwright; OPFS for persistent storage | + +The package is ESM-only. PostgreSQL, MySQL, Redis, and SQLite WASM require +their optional peer dependency. The Node SQLite adapter has no driver +dependency beyond Node.js. + +The browser runtime needs two platform capabilities: + +- Persistent storage uses the OPFS SAH pool VFS, which needs a secure context + and a dedicated worker. `sqliteWasm({ storage: "persistent" })` fails fast + where OPFS is unavailable; temporary storage works everywhere the WASM + module loads. +- The tab host election uses the Web Locks API. Every current browser + provides it. Node.js provides `navigator.locks` from 24.5, so Node-side use + of `solid-objects/browser/tab-host` needs a newer Node than the package + floor; the tab host test suite skips on older Node. The Node.js floor is 24.4.0 because the SQLite adapter reads integer columns as `BigInt`. Node.js 24.4.0 is the first release that accepts `readBigInts` on the @@ -39,7 +53,11 @@ The default suite exercises: Database jobs run the real adapter suites against PostgreSQL and MySQL servers. The Redis job runs wake-up behavior against a real Redis server. The browser job uses native WebSocket connections and Chromium for replay, payload, -component, dashboard, and revision-fence behavior. +component, dashboard, and revision-fence behavior. It also runs the browser +runtime suites in Chromium module workers: SQLite WASM transactions and OPFS +persistence across page reloads, the full runtime with durable actor state, +deadline rollback under the turn-scoped context store, two tabs on one +runtime with leader failover, and the sync bridge drain into a Node runtime. The quality job also: diff --git a/package.json b/package.json index 0a82256..3c31408 100644 --- a/package.json +++ b/package.json @@ -46,6 +46,14 @@ "types": "./dist/database/sqlite.d.ts", "import": "./dist/database/sqlite.js" }, + "./database/sqlite-wasm": { + "types": "./dist/database/sqlite-wasm.d.ts", + "import": "./dist/database/sqlite-wasm.js" + }, + "./database/shared-sqlite-wasm": { + "types": "./dist/database/shared-sqlite-wasm.d.ts", + "import": "./dist/database/shared-sqlite-wasm.js" + }, "./database/postgresql": { "types": "./dist/database/postgresql.d.ts", "import": "./dist/database/postgresql.js" @@ -62,6 +70,18 @@ "types": "./dist/browser/index.d.ts", "import": "./dist/browser/index.js" }, + "./browser/host": { + "types": "./dist/browser/host.d.ts", + "import": "./dist/browser/host.js" + }, + "./browser/tab-host": { + "types": "./dist/browser/tab-host.d.ts", + "import": "./dist/browser/tab-host.js" + }, + "./transmit": { + "types": "./dist/transmit.d.ts", + "import": "./dist/transmit.js" + }, "./web": { "types": "./dist/web/index.d.ts", "import": "./dist/web/index.js" @@ -69,7 +89,8 @@ }, "scripts": { "build": "pnpm run clean && tsc -p tsconfig.build.json && tsc -p tsconfig.quickstart-build.json && node scripts/prepare-executable.mjs", - "check": "pnpm run check:parameters && pnpm run check:documentation && tsc -p tsconfig.json --noEmit && tsc -p tsconfig.examples.json --noEmit", + "check": "pnpm run check:parameters && pnpm run check:documentation && pnpm run check:browser-imports && tsc -p tsconfig.json --noEmit && tsc -p tsconfig.examples.json --noEmit", + "check:browser-imports": "node scripts/check-browser-imports.mjs", "check:documentation": "node scripts/check-documentation.mjs", "check:parameters": "node scripts/check-parameter-style.mjs", "clean": "node scripts/clean.mjs", @@ -91,6 +112,7 @@ }, "devDependencies": { "@playwright/test": "^1.62.1", + "@sqlite.org/sqlite-wasm": "3.53.0-build1", "@types/node": "^24.0.0", "@types/pg": "^8.21.0", "@types/ws": "^8.18.1", @@ -104,11 +126,15 @@ "ws": "^8.21.3" }, "peerDependencies": { + "@sqlite.org/sqlite-wasm": ">=3.50.0-build1", "mysql2": "^3.23.3", "pg": "^8.23.0", "redis": "^6.2.1" }, "peerDependenciesMeta": { + "@sqlite.org/sqlite-wasm": { + "optional": true + }, "mysql2": { "optional": true }, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 5a19860..86ccf64 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -11,6 +11,9 @@ importers: '@playwright/test': specifier: ^1.62.1 version: 1.62.1 + '@sqlite.org/sqlite-wasm': + specifier: 3.53.0-build1 + version: 3.53.0-build1 '@types/node': specifier: ^24.0.0 version: 24.13.3 @@ -215,6 +218,10 @@ packages: '@rolldown/pluginutils@1.0.1': resolution: {integrity: sha512-2j9bGt5Jh8hj+vPtgzPtl72j0yRxHAyumoo6TNfAjsLB04UtpSvPbPcDcBMxz7n+9CYB0c1GxQFxYRg2jimqGw==} + '@sqlite.org/sqlite-wasm@3.53.0-build1': + resolution: {integrity: sha512-PfWPWN2n+/37doa8oh2/oUXk4OOsRYZsxc1W1sDXIGb/Pu5Yrb+f2eyYpgQMGITVX7HVgxhs9P18Rc6I97ym/g==} + engines: {node: '>=22'} + '@standard-schema/spec@1.1.0': resolution: {integrity: sha512-l2aFy5jALhniG5HgqrD6jXLi/rUWrKvqN/qJx6yoJsgKhblVd+iqqU4RCXavm/jPityDo5TCvKMnpjKnOriy0w==} @@ -816,6 +823,8 @@ snapshots: '@rolldown/pluginutils@1.0.1': {} + '@sqlite.org/sqlite-wasm@3.53.0-build1': {} + '@standard-schema/spec@1.1.0': {} '@types/chai@5.2.3': diff --git a/scripts/check-browser-imports.mjs b/scripts/check-browser-imports.mjs new file mode 100644 index 0000000..b5ac6ca --- /dev/null +++ b/scripts/check-browser-imports.mjs @@ -0,0 +1,75 @@ +import fs from "node:fs" +import path from "node:path" +import process from "node:process" +import ts from "typescript" + +const repositoryRoot = path.resolve(import.meta.dirname, "..") +const browserSafeRoots = [ + "src/browser/host.ts", + "src/browser/index.ts", + "src/browser/tab-host.ts", + "src/transmit.ts", + "src/context.ts", + "src/database/deadline.ts", + "src/database/shared-sqlite-wasm.ts", + "src/database/sqlite-wasm.ts", + "src/database/transaction-context.ts", + "src/database/types.ts", + "src/platform/context-store.ts", + "src/platform/turn-context-store.ts", + "src/platform/uuid.ts", +] +const forbiddenPackages = new Set(["mysql2", "pg", "redis", "ws"]) + +const violations = [] +const visited = new Set() +const roots = process.argv.slice(2).length > 0 ? process.argv.slice(2) : browserSafeRoots +for (const root of roots) visit(path.resolve(repositoryRoot, root)) + +if (violations.length > 0) { + process.stderr.write(`${violations.join("\n")}\n`) + process.exitCode = 1 +} + +function visit(filePath) { + if (visited.has(filePath)) return + visited.add(filePath) + const source = ts.createSourceFile( + filePath, + fs.readFileSync(filePath, "utf8"), + ts.ScriptTarget.Latest, + true, + ) + for (const statement of source.statements) { + const specifier = valueImportSpecifier(statement) + if (specifier === undefined) continue + if (specifier.startsWith("node:") || forbiddenPackages.has(specifier)) { + violations.push( + `${path.relative(repositoryRoot, filePath)} imports ${specifier}, which the browser cannot load`, + ) + continue + } + if (!specifier.startsWith(".")) continue + visit(resolveRelativeImport({ from: filePath, specifier })) + } +} + +function valueImportSpecifier(statement) { + if (ts.isImportDeclaration(statement)) { + if (statement.importClause?.isTypeOnly) return undefined + if (!ts.isStringLiteral(statement.moduleSpecifier)) return undefined + return statement.moduleSpecifier.text + } + if (ts.isExportDeclaration(statement)) { + if (statement.isTypeOnly) return undefined + if (!statement.moduleSpecifier || !ts.isStringLiteral(statement.moduleSpecifier)) + return undefined + return statement.moduleSpecifier.text + } + return undefined +} + +function resolveRelativeImport(options) { + const resolved = path.resolve(path.dirname(options.from), options.specifier) + return resolved.replace(/\.js$/, ".ts") +} diff --git a/scripts/check-documentation.mjs b/scripts/check-documentation.mjs index ae8093c..eea8ff1 100644 --- a/scripts/check-documentation.mjs +++ b/scripts/check-documentation.mjs @@ -55,10 +55,15 @@ const configurationReference = await readFile( const entryPoints = [ "src/index.ts", "src/database/sqlite.ts", + "src/database/sqlite-wasm.ts", + "src/database/shared-sqlite-wasm.ts", "src/database/postgresql.ts", "src/database/mysql.ts", "src/wake-up/redis.ts", "src/browser/index.ts", + "src/browser/host.ts", + "src/browser/tab-host.ts", + "src/transmit.ts", "src/web/index.ts", ] diff --git a/src/actor.ts b/src/actor.ts index 26051c2..95748e0 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 { TRANSMIT_EFFECT } from "./transmit-effect.js" import { createStagedOperationMap, createStagedOperations, @@ -222,6 +223,15 @@ export abstract class Actor { }) } + transmit(): ScheduledOperations { + return createStagedOperationMap(this.#operations, (operation, argumentsValue) => { + this.#intents.effects.push({ + name: TRANSMIT_EFFECT, + arguments: jsonObject({ operation, arguments: argumentsValue }), + }) + }) + } + commitAction(name: string, argumentsValue: Record = {}): void { this.#intents.commitActions.push({ name, arguments: jsonObject(argumentsValue) }) } diff --git a/src/broadcast-worker.ts b/src/broadcast-worker.ts index 57c9653..670ee47 100644 --- a/src/broadcast-worker.ts +++ b/src/broadcast-worker.ts @@ -1,4 +1,4 @@ -import { randomUUID } from "node:crypto" +import { randomUUID } from "./platform/uuid.js" import type { SolidObjectsRuntime } from "./runtime.js" import { withProcessHeartbeat } from "./worker.js" import { PollingBackoff } from "./polling-backoff.js" diff --git a/src/browser/host.ts b/src/browser/host.ts new file mode 100644 index 0000000..6a6e9f7 --- /dev/null +++ b/src/browser/host.ts @@ -0,0 +1,48 @@ +import { registerContextStoreFactory } from "../platform/context-store.js" +import { registerHostIdentity } from "../platform/host-identity.js" +import { TurnContextStore } from "../platform/turn-context-store.js" + +registerContextStoreFactory(() => new TurnContextStore()) +registerHostIdentity({ + hostname: globalThis.location?.hostname ?? "browser", + hostProcessId: randomHostProcessId(), + runtimeVersion: "browser", +}) + +export { Actor, broadcastInvalidation, broadcastValue } from "../actor.js" +export { configure, createRuntime, SolidObjectsRuntime } from "../runtime.js" +export { VERSION } from "../version.js" +export { + sqliteWasm, + SQLiteWasmDatabase, + type SQLiteWasmDatabaseOptions, +} from "../database/sqlite-wasm.js" +export { + sharedSqliteWasm, + SharedSQLiteWasmDatabase, + SharedDatabaseFailover, + SharedDatabaseUnavailable, + type SharedSQLiteWasmDatabaseOptions, +} from "../database/shared-sqlite-wasm.js" +export { + connectTabClient, + startTabHost, + type TabClient, + type TabClientOptions, + type TabHost, + type TabHostOptions, + type TabHostRuntimeHandle, + type TabInvocation, +} from "./tab-host.js" +export { + registerTransmit, + TRANSMIT_EFFECT, + type RegisterTransmitOptions, + type TransmitEnvelope, +} from "../transmit.js" + +function randomHostProcessId(): number { + const values = new Uint32Array(1) + globalThis.crypto.getRandomValues(values) + return values[0] ?? 1 +} diff --git a/src/browser/tab-host.ts b/src/browser/tab-host.ts new file mode 100644 index 0000000..ae87f78 --- /dev/null +++ b/src/browser/tab-host.ts @@ -0,0 +1,280 @@ +import { randomUUID } from "../platform/uuid.js" +import { requireWebLocks } from "../platform/web-locks.js" +import type { SolidObjectsRuntime } from "../runtime.js" +import type { DeepReadonly, JsonObject, JsonValue } from "../types.js" + +const PROTOCOL = "solid-objects-tab-host" +const VERSION = 1 +const NAME_PREFIX = "solid-objects:tab-host:" + +export interface TabHostRuntimeHandle { + runtime: SolidObjectsRuntime + close?: () => Promise +} + +export interface TabHostOptions { + name: string + startRuntime: () => Promise + onError?: (error: Error) => void +} + +export interface TabHost { + role(): "follower" | "leader" + leadership(): Promise + close(): Promise +} + +export interface TabInvocation { + actorType: string + actorId: string + operation: string + arguments?: JsonObject +} + +export interface TabClientOptions { + name: string + timeoutMilliseconds?: number + retryIntervalMilliseconds?: number +} + +export interface TabClient { + invoke(invocation: TabInvocation): Promise> + close(): void +} + +export class TabInvocationTimeout extends Error { + constructor(invocation: TabInvocation) { + super( + `no tab host answered ${invocation.actorType}/${invocation.actorId} ` + + `${invocation.operation} before the timeout`, + ) + this.name = "TabInvocationTimeout" + } +} + +export class TabInvocationFailed extends Error { + constructor(details: { name: string; message: string }) { + super(details.message) + this.name = details.name + } +} + +interface InvokeRequest { + protocol: typeof PROTOCOL + version: typeof VERSION + kind: "invoke" + requestId: string + actorType: string + actorId: string + operation: string + arguments: JsonObject +} + +interface InvokeResult { + protocol: typeof PROTOCOL + version: typeof VERSION + kind: "result" + requestId: string + outcome: + | { ok: true; value: DeepReadonly } + | { ok: false; error: { name: string; message: string } } +} + +interface LeaderAnnouncement { + protocol: typeof PROTOCOL + version: typeof VERSION + kind: "leader-online" +} + +type TabMessage = InvokeRequest | InvokeResult | LeaderAnnouncement + +export function startTabHost(options: TabHostOptions): TabHost { + const channelName = `${NAME_PREFIX}${options.name}` + const queueAbort = new AbortController() + let currentRole: "follower" | "leader" = "follower" + let closed = false + let promoteToLeader = () => {} + const leadershipPromise = new Promise((resolve) => { + promoteToLeader = resolve + }) + let releaseLeadership = () => {} + const heldLeadership = new Promise((resolve) => { + releaseLeadership = resolve + }) + let leaderCleanup: (() => Promise) | undefined + + const lockRequest = requireWebLocks() + .request(channelName, { signal: queueAbort.signal }, async () => { + if (closed) return + currentRole = "leader" + promoteToLeader() + const handle = await options.startRuntime() + const runAbort = new AbortController() + const running = handle.runtime.run(runAbort.signal) + const channel = new BroadcastChannel(channelName) + channel.onmessage = (event: MessageEvent) => { + const message = parseTabMessage(event.data) + if (message) void serveRequest({ handle, channel, message }) + } + leaderCleanup = async () => { + channel.onmessage = null + channel.close() + runAbort.abort() + await running.catch(() => undefined) + await (handle.close?.() ?? handle.runtime.close()) + } + const announcement: LeaderAnnouncement = { + protocol: PROTOCOL, + version: VERSION, + kind: "leader-online", + } + channel.postMessage(announcement) + if (closed) return + await heldLeadership + }) + .catch((error: unknown) => { + if (queueAbort.signal.aborted) return + options.onError?.(error instanceof Error ? error : new Error(String(error))) + throw error + }) + + return { + role: () => currentRole, + leadership: () => leadershipPromise, + close: async () => { + if (closed) return + closed = true + queueAbort.abort() + releaseLeadership() + await lockRequest.catch(() => undefined) + await leaderCleanup?.() + }, + } +} + +export function connectTabClient(options: TabClientOptions): TabClient { + const channelName = `${NAME_PREFIX}${options.name}` + const timeoutMilliseconds = options.timeoutMilliseconds ?? 10_000 + const retryIntervalMilliseconds = options.retryIntervalMilliseconds ?? 500 + const channel = new BroadcastChannel(channelName) + interface PendingInvocation { + request: InvokeRequest + resolve: (value: DeepReadonly) => void + reject: (error: Error) => void + retryTimer: ReturnType + timeoutTimer: ReturnType + } + const pending = new Map() + + const settle = (requestId: string, finish: (entry: PendingInvocation) => void) => { + const entry = pending.get(requestId) + if (!entry) return + clearInterval(entry.retryTimer) + clearTimeout(entry.timeoutTimer) + pending.delete(requestId) + finish(entry) + } + + channel.onmessage = (event: MessageEvent) => { + const message = parseTabMessage(event.data) + if (!message) return + if (message.kind === "leader-online") { + for (const entry of pending.values()) channel.postMessage(entry.request) + return + } + if (message.kind !== "result") return + settle(message.requestId, (entry) => { + if (message.outcome.ok) entry.resolve(message.outcome.value) + else entry.reject(new TabInvocationFailed(message.outcome.error)) + }) + } + + return { + invoke: (invocation) => + new Promise>((resolve, reject) => { + const request: InvokeRequest = { + protocol: PROTOCOL, + version: VERSION, + kind: "invoke", + requestId: randomUUID(), + actorType: invocation.actorType, + actorId: invocation.actorId, + operation: invocation.operation, + arguments: invocation.arguments ?? {}, + } + const entry: PendingInvocation = { + request, + resolve, + reject, + retryTimer: setInterval(() => channel.postMessage(request), retryIntervalMilliseconds), + timeoutTimer: setTimeout( + () => + settle(request.requestId, (held) => + held.reject(new TabInvocationTimeout(invocation)), + ), + timeoutMilliseconds, + ), + } + pending.set(request.requestId, entry) + channel.postMessage(request) + }), + close: () => { + for (const requestId of [...pending.keys()]) { + settle(requestId, (entry) => entry.reject(new Error("tab client closed"))) + } + channel.onmessage = null + channel.close() + }, + } +} + +async function serveRequest(input: { + handle: TabHostRuntimeHandle + channel: BroadcastChannel + message: TabMessage +}): Promise { + const { handle, channel, message } = input + if (message.kind !== "invoke") return + const request = message + let outcome: InvokeResult["outcome"] + try { + const enqueued = await handle.runtime.enqueueInternalMessage({ + actorType: request.actorType, + actorId: request.actorId, + operation: request.operation, + argumentsValue: request.arguments, + idempotencyKey: `tab:${request.requestId}`, + }) + outcome = { ok: true, value: await enqueued.wait() } + } catch (error) { + outcome = { + ok: false, + error: { + name: error instanceof Error ? error.name : "Error", + message: error instanceof Error ? error.message : String(error), + }, + } + } + const result: InvokeResult = { + protocol: PROTOCOL, + version: VERSION, + kind: "result", + requestId: request.requestId, + outcome, + } + channel.postMessage(result) +} + +function parseTabMessage(value: unknown): TabMessage | undefined { + if (typeof value !== "object" || value === null) return undefined + const candidate = value as Partial + if (candidate.protocol !== PROTOCOL || candidate.version !== VERSION) return undefined + if ( + candidate.kind === "invoke" || + candidate.kind === "result" || + candidate.kind === "leader-online" + ) { + return candidate as TabMessage + } + return undefined +} diff --git a/src/cli.ts b/src/cli.ts index 7069aa2..02c660c 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -1,3 +1,4 @@ +import "./platform/node.js" import { resolve } from "node:path" import { pathToFileURL } from "node:url" import { SolidObjectsRuntime } from "./runtime.js" diff --git a/src/context.ts b/src/context.ts index f520e67..5e13564 100644 --- a/src/context.ts +++ b/src/context.ts @@ -1,4 +1,4 @@ -import { AsyncLocalStorage } from "node:async_hooks" +import { createContextStore } from "./platform/context-store.js" import type { Actor } from "./actor.js" import type { SolidObjectsRuntime } from "./runtime.js" import type { MessageContext } from "./types.js" @@ -16,7 +16,7 @@ interface ActorExecutionContext { message?: MessageContext } -const storage = new AsyncLocalStorage() +const storage = createContextStore() export function currentActor(): Actor | undefined { return storage.getStore()?.actor diff --git a/src/database/deadline.ts b/src/database/deadline.ts index 3340414..77fab28 100644 --- a/src/database/deadline.ts +++ b/src/database/deadline.ts @@ -1,11 +1,11 @@ -import { AsyncLocalStorage } from "node:async_hooks" +import { createContextStore } from "../platform/context-store.js" import { DatabaseDeadlineExceeded } from "../errors.js" interface DatabaseDeadline { expiresAt: number } -const deadlines = new AsyncLocalStorage() +const deadlines = createContextStore() export function withDatabaseDeadline( options: { timeoutMilliseconds: number }, diff --git a/src/database/mysql.ts b/src/database/mysql.ts index 611b3e9..6210162 100644 --- a/src/database/mysql.ts +++ b/src/database/mysql.ts @@ -1,3 +1,4 @@ +import "../platform/node.js" import mysqlDriver, { type Pool, type PoolConnection, diff --git a/src/database/postgresql.ts b/src/database/postgresql.ts index f45722d..1e57a41 100644 --- a/src/database/postgresql.ts +++ b/src/database/postgresql.ts @@ -1,3 +1,4 @@ +import "../platform/node.js" import { Pool, TypeOverrides, types, type PoolClient, type PoolConfig } from "pg" import { postgresqlSql } from "./postgresql-sql.js" import type { Database, DatabaseConnection, RunResult } from "./types.js" diff --git a/src/database/shared-sqlite-wasm.ts b/src/database/shared-sqlite-wasm.ts new file mode 100644 index 0000000..69e359f --- /dev/null +++ b/src/database/shared-sqlite-wasm.ts @@ -0,0 +1,688 @@ +import { randomUUID } from "../platform/uuid.js" +import { requireWebLocks } from "../platform/web-locks.js" +import { DatabaseDeadlineExceeded } from "../errors.js" +import { requireDatabaseDeadlineRemaining } from "./deadline.js" +import { databaseTransactionActive, withDatabaseTransaction } from "./transaction-context.js" +import { sqliteWasm, type SQLiteWasmDatabase } from "./sqlite-wasm.js" +import type { Database, DatabaseConnection, RunResult } from "./types.js" + +const PROTOCOL = "solid-objects-shared-sqlite" +const VERSION = 1 +const NAME_PREFIX = "solid-objects:shared-sqlite:" + +export interface SharedSQLiteWasmDatabaseOptions { + path: string + name?: string + storage?: "temporary" | "persistent" + requestTimeoutMilliseconds?: number + retryIntervalMilliseconds?: number + sessionIdleTimeoutMilliseconds?: number + openAttempts?: number + onError?: (error: Error) => void +} + +export class SharedDatabaseFailover extends Error { + constructor() { + super("the shared SQLite database failed over; retry the operation") + this.name = "SharedDatabaseFailover" + } +} + +export class SharedDatabaseUnavailable extends Error { + constructor(message: string) { + super(message) + this.name = "SharedDatabaseUnavailable" + } +} + +type SessionMode = "connection" | "transaction" +type StatementOperation = "run" | "get" | "all" | "now" +type CloseOutcome = "commit" | "rollback" | "end" + +interface Envelope { + protocol: typeof PROTOCOL + version: typeof VERSION +} + +interface PingMessage extends Envelope { + kind: "ping" + requestId: string +} + +interface PongMessage extends Envelope { + kind: "pong" + requestId: string + epoch: string +} + +interface HolderOnlineMessage extends Envelope { + kind: "holder-online" + epoch: string +} + +interface OpenMessage extends Envelope { + kind: "open" + requestId: string + epoch: string + sessionId: string + mode: SessionMode +} + +interface StatementMessage extends Envelope { + kind: "statement" + requestId: string + epoch: string + sessionId: string + operation: StatementOperation + sql: string + parameters: readonly unknown[] +} + +interface CloseSessionMessage extends Envelope { + kind: "close" + requestId: string + epoch: string + sessionId: string + outcome: CloseOutcome +} + +interface ResultMessage extends Envelope { + kind: "result" + requestId: string + outcome: { ok: true; value: unknown } | { ok: false; error: { name: string; message: string } } +} + +type SharedMessage = + | PingMessage + | PongMessage + | HolderOnlineMessage + | OpenMessage + | StatementMessage + | CloseSessionMessage + | ResultMessage + +class SessionOpenRejected extends Error { + constructor(readonly reason: Error) { + super(reason.message) + this.name = "SessionOpenRejected" + } +} + +class RollbackSignal extends Error { + constructor() { + super("rollback requested") + this.name = "RollbackSignal" + } +} + +interface PendingRequest { + epoch: string + resolve: (value: unknown) => void + reject: (error: Error) => void + timer: ReturnType +} + +interface HolderSession { + connection?: DatabaseConnection + finishResolve: () => void + finishReject: (error: Error) => void + settled: Promise + watchdog?: ReturnType +} + +export function sharedSqliteWasm( + options: SharedSQLiteWasmDatabaseOptions, +): SharedSQLiteWasmDatabase { + return new SharedSQLiteWasmDatabase(options) +} + +export class SharedSQLiteWasmDatabase implements Database { + readonly family = "sqlite" as const + readonly schemaIdentity = "solid-objects-wasm-v1" + private readonly options: SharedSQLiteWasmDatabaseOptions + private readonly channel: BroadcastChannel + private readonly electionAbort = new AbortController() + private readonly pending = new Map() + private readonly holderSessions = new Map() + private readonly epochWaiters: Array<(epoch: string) => void> = [] + private readonly electionLoop: Promise + private roleValue: "connecting" | "holder" | "remote" = "connecting" + private currentEpoch: string | undefined + private underlying: SQLiteWasmDatabase | undefined + private releaseHold: (() => void) | undefined + private pingTimer: ReturnType | undefined + private closed = false + + constructor(options: SharedSQLiteWasmDatabaseOptions) { + this.options = options + this.channel = new BroadcastChannel(`${NAME_PREFIX}${this.electionName()}`) + this.channel.onmessage = (event: MessageEvent) => { + const message = parseSharedMessage(event.data) + if (message) void this.receive(message) + } + this.pingTimer = setInterval(() => { + if (this.currentEpoch !== undefined || this.roleValue === "holder") { + this.stopPinging() + return + } + this.post({ ...envelope(), kind: "ping", requestId: randomUUID() }) + }, this.retryIntervalMilliseconds()) + this.electionLoop = this.runElection() + } + + role(): "connecting" | "holder" | "remote" { + return this.roleValue + } + + async connection( + callback: (connection: DatabaseConnection) => Promise, + ): Promise { + return this.dispatchSession({ mode: "connection", callback }) + } + + async transaction( + callback: (connection: DatabaseConnection) => Promise, + ): Promise { + return withDatabaseTransaction(this, () => + this.dispatchSession({ mode: "transaction", callback }), + ) + } + + private async dispatchSession(input: { + mode: SessionMode + callback: (connection: DatabaseConnection) => Promise + }): Promise { + const attempts = 3 + for (let attempt = 0; ; attempt += 1) { + if (this.roleValue === "holder" && this.underlying) { + return input.mode === "transaction" + ? this.underlying.transaction(input.callback) + : this.underlying.connection(input.callback) + } + try { + return await this.remoteSession(input) + } catch (error) { + if (!(error instanceof SessionOpenRejected)) throw error + if (attempt >= attempts - 1) throw error.reason + requireDatabaseDeadlineRemaining() + await delay(this.retryIntervalMilliseconds()) + } + } + } + + transactionActive(): boolean { + return databaseTransactionActive(this) + } + + async close(): Promise { + if (this.closed) return + this.closed = true + this.stopPinging() + this.electionAbort.abort() + this.releaseHold?.() + this.rejectInFlight(new SharedDatabaseUnavailable("the shared SQLite database is closed")) + await this.electionLoop.catch(() => undefined) + this.channel.onmessage = null + this.channel.close() + } + + private async remoteSession(input: { + mode: SessionMode + callback: (connection: DatabaseConnection) => Promise + }): Promise { + const initialRemaining = requireDatabaseDeadlineRemaining() + const deadlineExpiresAtMilliseconds = + initialRemaining === undefined ? undefined : performance.now() + initialRemaining + const epoch = await this.requireEpoch().catch((error: Error) => { + throw new SessionOpenRejected(error) + }) + const sessionId = randomUUID() + await this.request({ + ...envelope(), + kind: "open", + requestId: randomUUID(), + epoch, + sessionId, + mode: input.mode, + }).catch((error: Error) => { + throw new SessionOpenRejected(error) + }) + const connection = new SharedRemoteConnection({ + database: this, + sessionId, + epoch, + deadlineExpiresAtMilliseconds, + }) + try { + const result = await input.callback(connection) + requireDatabaseDeadlineRemaining() + requireCapturedDeadline(deadlineExpiresAtMilliseconds) + await this.request({ + ...envelope(), + kind: "close", + requestId: randomUUID(), + epoch, + sessionId, + outcome: input.mode === "transaction" ? "commit" : "end", + }) + return result + } catch (error) { + await this.request({ + ...envelope(), + kind: "close", + requestId: randomUUID(), + epoch, + sessionId, + outcome: input.mode === "transaction" ? "rollback" : "end", + }).catch(() => undefined) + throw error + } + } + + sendStatement(input: { + sessionId: string + epoch: string + operation: StatementOperation + sql: string + parameters: readonly unknown[] + }): Promise { + return this.request({ + ...envelope(), + kind: "statement", + requestId: randomUUID(), + epoch: input.epoch, + sessionId: input.sessionId, + operation: input.operation, + sql: input.sql, + parameters: input.parameters, + }) + } + + private request(message: OpenMessage | StatementMessage | CloseSessionMessage): Promise { + if (this.closed) { + return Promise.reject(new SharedDatabaseUnavailable("the shared SQLite database is closed")) + } + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + this.pending.delete(message.requestId) + reject( + new SharedDatabaseUnavailable( + `no shared SQLite database holder answered within ${this.requestTimeoutMilliseconds()}ms`, + ), + ) + }, this.requestTimeoutMilliseconds()) + this.pending.set(message.requestId, { epoch: message.epoch, resolve, reject, timer }) + this.post(message) + }) + } + + private async requireEpoch(): Promise { + if (this.currentEpoch !== undefined) return this.currentEpoch + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + const index = this.epochWaiters.indexOf(waiter) + if (index >= 0) this.epochWaiters.splice(index, 1) + reject( + new SharedDatabaseUnavailable( + `no shared SQLite database holder announced within ${this.requestTimeoutMilliseconds()}ms`, + ), + ) + }, this.requestTimeoutMilliseconds()) + const waiter = (epoch: string) => { + clearTimeout(timer) + resolve(epoch) + } + this.epochWaiters.push(waiter) + }) + } + + private adoptEpoch(epoch: string): void { + if (this.currentEpoch === epoch) return + if (this.currentEpoch !== undefined) this.rejectInFlight(new SharedDatabaseFailover()) + this.currentEpoch = epoch + if (this.roleValue !== "holder") this.roleValue = "remote" + this.stopPinging() + for (const waiter of this.epochWaiters.splice(0)) waiter(epoch) + } + + private rejectInFlight(error: Error): void { + for (const [requestId, entry] of [...this.pending]) { + this.pending.delete(requestId) + clearTimeout(entry.timer) + entry.reject(error) + } + } + + private async receive(message: SharedMessage): Promise { + if (message.kind === "result") { + const entry = this.pending.get(message.requestId) + if (!entry) return + this.pending.delete(message.requestId) + clearTimeout(entry.timer) + if (message.outcome.ok) entry.resolve(message.outcome.value) + else entry.reject(rebuildError(message.outcome.error)) + return + } + if (message.kind === "holder-online" || message.kind === "pong") { + if (this.roleValue !== "holder") this.adoptEpoch(message.epoch) + return + } + if (this.roleValue !== "holder" || this.currentEpoch === undefined) return + if (message.kind === "ping") { + this.post({ + ...envelope(), + kind: "pong", + requestId: message.requestId, + epoch: this.currentEpoch, + }) + return + } + if (message.epoch !== this.currentEpoch) { + this.respond(message.requestId, { + ok: false, + error: describeError(new SharedDatabaseFailover()), + }) + return + } + if (message.kind === "open") await this.handleOpen(message) + else if (message.kind === "statement") await this.handleStatement(message) + else await this.handleClose(message) + } + + private async handleOpen(message: OpenMessage): Promise { + const underlying = this.underlying + if (!underlying) return + let finishResolve = () => {} + let finishReject: (error: Error) => void = () => {} + const done = new Promise((resolve, reject) => { + finishResolve = resolve + finishReject = reject + }) + let ready: (session: HolderSession) => void = () => {} + const acquired = new Promise((resolve) => { + ready = resolve + }) + const runner = + message.mode === "transaction" + ? underlying.transaction.bind(underlying) + : underlying.connection.bind(underlying) + const session: HolderSession = { + finishResolve, + finishReject, + settled: runner(async (connection) => { + session.connection = connection + ready(session) + await done + return null + }), + } + session.settled.catch(() => undefined) + this.holderSessions.set(message.sessionId, session) + await acquired + this.touchSession(message.sessionId) + this.respond(message.requestId, { ok: true, value: null }) + } + + private async handleStatement(message: StatementMessage): Promise { + const session = this.holderSessions.get(message.sessionId) + if (!session?.connection) { + this.respond(message.requestId, { + ok: false, + error: describeError(new SharedDatabaseFailover()), + }) + return + } + this.touchSession(message.sessionId) + try { + const value = await executeStatement({ connection: session.connection, message }) + this.respond(message.requestId, { ok: true, value }) + } catch (error) { + this.respond(message.requestId, { ok: false, error: describeError(error) }) + } + } + + private async handleClose(message: CloseSessionMessage): Promise { + const session = this.holderSessions.get(message.sessionId) + if (!session) { + this.respond(message.requestId, { ok: true, value: null }) + return + } + this.dropSession(message.sessionId) + if (message.outcome === "rollback") { + session.finishReject(new RollbackSignal()) + await session.settled.catch(() => undefined) + this.respond(message.requestId, { ok: true, value: null }) + return + } + session.finishResolve() + try { + await session.settled + this.respond(message.requestId, { ok: true, value: null }) + } catch (error) { + this.respond(message.requestId, { ok: false, error: describeError(error) }) + } + } + + private touchSession(sessionId: string): void { + const session = this.holderSessions.get(sessionId) + if (!session) return + if (session.watchdog) clearTimeout(session.watchdog) + session.watchdog = setTimeout(() => { + this.dropSession(sessionId) + session.finishReject( + new SharedDatabaseUnavailable("the shared SQLite session timed out while idle"), + ) + }, this.sessionIdleTimeoutMilliseconds()) + } + + private dropSession(sessionId: string): void { + const session = this.holderSessions.get(sessionId) + if (!session) return + if (session.watchdog) clearTimeout(session.watchdog) + this.holderSessions.delete(sessionId) + } + + private async runElection(): Promise { + try { + requireWebLocks() + } catch (error) { + this.options.onError?.(error instanceof Error ? error : new Error(String(error))) + return + } + while (!this.closed) { + try { + await requireWebLocks().request( + `${NAME_PREFIX}${this.electionName()}`, + { signal: this.electionAbort.signal }, + () => this.serveAsHolder(), + ) + return + } catch (error) { + if (this.closed || this.electionAbort.signal.aborted) return + this.options.onError?.(error instanceof Error ? error : new Error(String(error))) + await delay(this.retryIntervalMilliseconds()) + } + } + } + + private async serveAsHolder(): Promise { + if (this.closed) return + const underlying = await this.openUnderlyingWithRetry() + if (this.closed) { + await underlying.close() + return + } + this.underlying = underlying + this.roleValue = "holder" + this.stopPinging() + const epoch = randomUUID() + this.rejectInFlight(new SharedDatabaseFailover()) + this.currentEpoch = epoch + for (const waiter of this.epochWaiters.splice(0)) waiter(epoch) + this.post({ ...envelope(), kind: "holder-online", epoch }) + await new Promise((resolve) => { + this.releaseHold = resolve + if (this.closed) resolve() + }) + for (const sessionId of [...this.holderSessions.keys()]) { + const session = this.holderSessions.get(sessionId) + this.dropSession(sessionId) + session?.finishReject( + new SharedDatabaseUnavailable("the shared SQLite database holder is closed"), + ) + await session?.settled.catch(() => undefined) + } + await underlying.close() + } + + private async openUnderlyingWithRetry(): Promise { + const attempts = this.options.openAttempts ?? 50 + let lastError: unknown + for (let attempt = 0; attempt < attempts; attempt += 1) { + if (this.closed) break + try { + return await sqliteWasm({ path: this.options.path, storage: this.storageMode() }) + } catch (error) { + lastError = error + await delay(100) + } + } + throw lastError instanceof Error + ? lastError + : new SharedDatabaseUnavailable("the shared SQLite pool could not be opened") + } + + private respond(requestId: string, outcome: ResultMessage["outcome"]): void { + this.post({ ...envelope(), kind: "result", requestId, outcome }) + } + + private post(message: SharedMessage): void { + if (this.closed) return + this.channel.postMessage(message) + queueMicrotask(() => { + void this.receive(message) + }) + } + + private stopPinging(): void { + if (this.pingTimer === undefined) return + clearInterval(this.pingTimer) + this.pingTimer = undefined + } + + private electionName(): string { + return this.options.name ?? this.options.path + } + + private storageMode(): "temporary" | "persistent" { + return this.options.storage ?? (this.options.path === ":memory:" ? "temporary" : "persistent") + } + + private requestTimeoutMilliseconds(): number { + return this.options.requestTimeoutMilliseconds ?? 15_000 + } + + private retryIntervalMilliseconds(): number { + return this.options.retryIntervalMilliseconds ?? 200 + } + + private sessionIdleTimeoutMilliseconds(): number { + return this.options.sessionIdleTimeoutMilliseconds ?? 10_000 + } +} + +class SharedRemoteConnection implements DatabaseConnection { + constructor( + private readonly context: { + database: SharedSQLiteWasmDatabase + sessionId: string + epoch: string + deadlineExpiresAtMilliseconds: number | undefined + }, + ) {} + + async run(sql: string, parameters: readonly unknown[] = []): Promise { + return (await this.statement({ operation: "run", sql, parameters })) as RunResult + } + + async get( + sql: string, + parameters: readonly unknown[] = [], + ): Promise { + return (await this.statement({ operation: "get", sql, parameters })) as Row | undefined + } + + async all(sql: string, parameters: readonly unknown[] = []): Promise { + return (await this.statement({ operation: "all", sql, parameters })) as Row[] + } + + async nowMilliseconds(): Promise { + return (await this.statement({ operation: "now", sql: "", parameters: [] })) as number + } + + private statement(input: { + operation: StatementOperation + sql: string + parameters: readonly unknown[] + }): Promise { + requireDatabaseDeadlineRemaining() + requireCapturedDeadline(this.context.deadlineExpiresAtMilliseconds) + return this.context.database.sendStatement({ + sessionId: this.context.sessionId, + epoch: this.context.epoch, + operation: input.operation, + sql: input.sql, + parameters: input.parameters, + }) + } +} + +async function executeStatement(input: { + connection: DatabaseConnection + message: StatementMessage +}): Promise { + const { connection, message } = input + if (message.operation === "run") return connection.run(message.sql, message.parameters) + if (message.operation === "get") return connection.get(message.sql, message.parameters) + if (message.operation === "all") return connection.all(message.sql, message.parameters) + return connection.nowMilliseconds() +} + +function requireCapturedDeadline(expiresAtMilliseconds: number | undefined): void { + if (expiresAtMilliseconds === undefined) return + if (performance.now() >= expiresAtMilliseconds) { + throw new DatabaseDeadlineExceeded("database deadline exceeded") + } +} + +function envelope(): Envelope { + return { protocol: PROTOCOL, version: VERSION } +} + +function parseSharedMessage(value: unknown): SharedMessage | undefined { + if (typeof value !== "object" || value === null) return undefined + const candidate = value as Partial + if (candidate.protocol !== PROTOCOL || candidate.version !== VERSION) return undefined + const kinds = ["ping", "pong", "holder-online", "open", "statement", "close", "result"] + if (typeof candidate.kind !== "string" || !kinds.includes(candidate.kind)) return undefined + return candidate as SharedMessage +} + +function describeError(error: unknown): { name: string; message: string } { + if (error instanceof Error) return { name: error.name, message: error.message } + return { name: "Error", message: String(error) } +} + +function rebuildError(details: { name: string; message: string }): Error { + if (details.name === "DatabaseDeadlineExceeded") { + return new DatabaseDeadlineExceeded(details.message) + } + if (details.name === "SharedDatabaseFailover") return new SharedDatabaseFailover() + const error = new Error(details.message) + error.name = details.name + return error +} + +function delay(milliseconds: number): Promise { + return new Promise((resolve) => setTimeout(resolve, milliseconds)) +} diff --git a/src/database/sqlite-wasm.ts b/src/database/sqlite-wasm.ts new file mode 100644 index 0000000..6949f5c --- /dev/null +++ b/src/database/sqlite-wasm.ts @@ -0,0 +1,242 @@ +import sqlite3InitModule, { + type BindableValue, + type Database as WasmDatabaseHandle, + type Sqlite3Static, + type SqlValue, +} from "@sqlite.org/sqlite-wasm" +import type { Database, DatabaseConnection, RunResult } from "./types.js" +import { requireDatabaseDeadlineRemaining } from "./deadline.js" +import { DatabaseDeadlineExceeded } from "../errors.js" +import { databaseTransactionActive, withDatabaseTransaction } from "./transaction-context.js" + +export interface SQLiteWasmDatabaseOptions { + path: string + storage?: "temporary" | "persistent" +} + +interface SQLiteWasmHandles { + handle: WasmDatabaseHandle + sqlite3: Sqlite3Static +} + +interface SQLiteWasmConnectionOptions { + handles: SQLiteWasmHandles + requireDeadlineRemaining: () => void +} + +class SQLiteWasmConnection implements DatabaseConnection { + private readonly handles: SQLiteWasmHandles + private readonly requireDeadlineRemaining: () => void + + constructor(options: SQLiteWasmConnectionOptions) { + this.handles = options.handles + this.requireDeadlineRemaining = options.requireDeadlineRemaining + } + + async run(sql: string, parameters: readonly unknown[] = []): Promise { + requireDatabaseDeadlineRemaining() + this.requireDeadlineRemaining() + this.execute({ sql, parameters }) + const output: RunResult = { changes: this.handles.handle.changes() } + const lastInsertRowid = this.handles.sqlite3.capi.sqlite3_last_insert_rowid(this.handles.handle) + if (Number(lastInsertRowid) !== 0) output.lastInsertId = String(lastInsertRowid) + return output + } + + async get( + sql: string, + parameters: readonly unknown[] = [], + ): Promise { + const rows = await this.all(sql, parameters) + return rows[0] + } + + async all(sql: string, parameters: readonly unknown[] = []): Promise { + requireDatabaseDeadlineRemaining() + this.requireDeadlineRemaining() + const resultRows: Record[] = [] + this.execute({ sql, parameters, resultRows }) + return resultRows as Row[] + } + + async nowMilliseconds(): Promise { + const row = await this.get<{ now_ms: number | bigint }>( + "SELECT CAST((julianday('now') - 2440587.5) * 86400000 AS INTEGER) AS now_ms", + ) + if (row === undefined) throw new Error("SQLite returned no clock row") + return Number(row.now_ms) + } + + private execute(options: { + sql: string + parameters: readonly unknown[] + resultRows?: Record[] + }): void { + const bind = options.parameters.map(normalizeParameter) + this.handles.handle.exec({ + sql: options.sql, + rowMode: "object", + ...(bind.length > 0 ? { bind } : {}), + ...(options.resultRows ? { resultRows: options.resultRows } : {}), + }) + } +} + +export class SQLiteWasmDatabase implements Database { + readonly family = "sqlite" as const + readonly schemaIdentity = "solid-objects-wasm-v1" + private readonly handles: SQLiteWasmHandles + private readonly databaseConnection: SQLiteWasmConnection + private accessTail: Promise = Promise.resolve() + private closed = false + private activeDeadlineExpiresAtMilliseconds: number | undefined + + constructor(handles: SQLiteWasmHandles) { + this.handles = handles + this.handles.handle.exec("PRAGMA foreign_keys = ON") + this.databaseConnection = new SQLiteWasmConnection({ + handles, + requireDeadlineRemaining: () => this.requireActiveDeadlineRemaining(), + }) + } + + async connection( + callback: (connection: DatabaseConnection) => Promise, + ): Promise { + return this.withAccess(() => callback(this.databaseConnection)) + } + + async transaction( + callback: (connection: DatabaseConnection) => Promise, + ): Promise { + return withDatabaseTransaction(this, () => this.withAccess(() => this.runTransaction(callback))) + } + + transactionActive(): boolean { + return databaseTransactionActive(this) + } + + async close(): Promise { + if (this.closed) return + await this.accessTail + if (this.closed) return + this.handles.handle.close() + this.closed = true + } + + private async withAccess(callback: () => Promise): Promise { + const initialRemaining = requireDatabaseDeadlineRemaining() + const deadlineExpiresAtMilliseconds = + initialRemaining === undefined ? undefined : performance.now() + initialRemaining + const previous = this.accessTail + let release = () => {} + this.accessTail = new Promise((resolve) => { + release = resolve + }) + if (initialRemaining !== undefined) { + try { + await waitForAccess(previous, initialRemaining) + } catch (error) { + void previous.then(release, release) + throw error + } + } else { + await previous + } + try { + if (initialRemaining !== undefined) requireDatabaseDeadlineRemaining() + if (this.closed) throw new Error("SQLite WASM database is closed") + this.activeDeadlineExpiresAtMilliseconds = deadlineExpiresAtMilliseconds + this.requireActiveDeadlineRemaining() + return await callback() + } finally { + this.activeDeadlineExpiresAtMilliseconds = undefined + release() + } + } + + private requireActiveDeadlineRemaining(): void { + const expiresAt = this.activeDeadlineExpiresAtMilliseconds + if (expiresAt === undefined) return + if (performance.now() >= expiresAt) { + throw new DatabaseDeadlineExceeded("database deadline exceeded") + } + } + + private async runTransaction( + callback: (connection: DatabaseConnection) => Promise, + ): Promise { + this.handles.handle.exec("BEGIN IMMEDIATE") + try { + const result = await callback(this.databaseConnection) + requireDatabaseDeadlineRemaining() + this.requireActiveDeadlineRemaining() + this.handles.handle.exec("COMMIT") + return result + } catch (error) { + this.handles.handle.exec("ROLLBACK") + throw error + } + } +} + +let sqlite3ModulePromise: Promise | undefined + +function loadSqlite3(): Promise { + sqlite3ModulePromise ??= sqlite3InitModule() + return sqlite3ModulePromise +} + +export async function sqliteWasm(options: SQLiteWasmDatabaseOptions): Promise { + const sqlite3 = await loadSqlite3() + const handle = await openHandle({ sqlite3, options }) + return new SQLiteWasmDatabase({ handle, sqlite3 }) +} + +async function openHandle(context: { + sqlite3: Sqlite3Static + options: SQLiteWasmDatabaseOptions +}): Promise { + if (context.options.storage !== "persistent") { + return new context.sqlite3.oo1.DB(context.options.path, "c") + } + const reinitOptions = { + forceReinitIfPreviouslyFailed: true, + } as Parameters[0] + const pool = await context.sqlite3.installOpfsSAHPoolVfs(reinitOptions) + return new pool.OpfsSAHPoolDb(context.options.path) +} + +function waitForAccess(access: Promise, timeoutMilliseconds: number): Promise { + return new Promise((resolve, reject) => { + const timeout = setTimeout( + () => reject(new DatabaseDeadlineExceeded("database deadline exceeded")), + timeoutMilliseconds, + ) + void access.then( + () => { + clearTimeout(timeout) + resolve() + }, + (error: unknown) => { + clearTimeout(timeout) + reject(error) + }, + ) + }) +} + +function normalizeParameter(value: unknown): BindableValue { + if (value === undefined) return null + if ( + value === null || + typeof value === "string" || + typeof value === "number" || + typeof value === "bigint" || + value instanceof Uint8Array + ) + return value + + if (typeof value === "boolean") return value ? 1 : 0 + throw new TypeError(`unsupported SQLite parameter ${typeof value}`) +} diff --git a/src/database/sqlite.ts b/src/database/sqlite.ts index 4ad2ce7..c2d4f9f 100644 --- a/src/database/sqlite.ts +++ b/src/database/sqlite.ts @@ -1,3 +1,4 @@ +import "../platform/node.js" import { DatabaseSync } from "node:sqlite" import type { Database, DatabaseConnection, RunResult } from "./types.js" import { diff --git a/src/database/transaction-context.ts b/src/database/transaction-context.ts index 2b51ab3..1d69dc1 100644 --- a/src/database/transaction-context.ts +++ b/src/database/transaction-context.ts @@ -1,11 +1,11 @@ -import { AsyncLocalStorage } from "node:async_hooks" +import { createContextStore } from "../platform/context-store.js" interface TransactionScope { database: object active: boolean } -const transactionScopes = new AsyncLocalStorage() +const transactionScopes = createContextStore() export function withDatabaseTransaction( database: object, diff --git a/src/doctor.ts b/src/doctor.ts index f8eee21..c000be8 100644 --- a/src/doctor.ts +++ b/src/doctor.ts @@ -1,4 +1,4 @@ -import { randomUUID } from "node:crypto" +import { randomUUID } from "./platform/uuid.js" import { Actor } from "./actor.js" import type { DatabaseConnection } from "./database/types.js" import { initialStateFor, validateDefinition } from "./definition.js" diff --git a/src/effect-worker.ts b/src/effect-worker.ts index afbd669..68f4653 100644 --- a/src/effect-worker.ts +++ b/src/effect-worker.ts @@ -1,4 +1,4 @@ -import { randomUUID } from "node:crypto" +import { randomUUID } from "./platform/uuid.js" import type { SolidObjectsRuntime } from "./runtime.js" import { withProcessHeartbeat } from "./worker.js" import { PollingBackoff } from "./polling-backoff.js" diff --git a/src/index.ts b/src/index.ts index 34ca19c..fe2b8b8 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,3 +1,5 @@ +import "./platform/node.js" + export { Actor, broadcastInvalidation, @@ -21,6 +23,14 @@ export { type SnapshotWithIncarnation, } from "./runtime.js" export { VERSION } from "./version.js" +export { + receiveTransmitEnvelope, + registerTransmit, + TRANSMIT_EFFECT, + InvalidTransmitEnvelope, + type RegisterTransmitOptions, + type TransmitEnvelope, +} from "./transmit.js" export { guardApplicationDatabase } from "./application-database.js" export { runCli, type CliRunOptions } from "./cli.js" export { Worker } from "./worker.js" diff --git a/src/platform/context-store.ts b/src/platform/context-store.ts new file mode 100644 index 0000000..3c1ba7e --- /dev/null +++ b/src/platform/context-store.ts @@ -0,0 +1,38 @@ +export interface ContextStore { + run(store: Store, callback: () => Result): Result + getStore(): Store | undefined +} + +export type ContextStoreFactory = () => ContextStore + +export class ContextStoreFactoryMissing extends Error { + constructor() { + super( + "no context store factory is registered; " + + "import a platform entry point before actor or database work starts", + ) + this.name = "ContextStoreFactoryMissing" + } +} + +let registeredFactory: ContextStoreFactory | undefined + +export function registerContextStoreFactory(factory: ContextStoreFactory): void { + registeredFactory = factory +} + +export function createContextStore(): ContextStore { + let instance: ContextStore | undefined + return { + run(store: Store, callback: () => Result): Result { + if (!instance) { + if (!registeredFactory) throw new ContextStoreFactoryMissing() + instance = registeredFactory() + } + return instance.run(store, callback) + }, + getStore(): Store | undefined { + return instance?.getStore() + }, + } +} diff --git a/src/platform/host-identity.ts b/src/platform/host-identity.ts new file mode 100644 index 0000000..a0c9c21 --- /dev/null +++ b/src/platform/host-identity.ts @@ -0,0 +1,26 @@ +export interface HostIdentity { + hostname: string + hostProcessId: number + runtimeVersion: string +} + +export class HostIdentityMissing extends Error { + constructor() { + super( + "no host identity is registered; " + + "import a platform entry point before actor or database work starts", + ) + this.name = "HostIdentityMissing" + } +} + +let registeredIdentity: HostIdentity | undefined + +export function registerHostIdentity(identity: HostIdentity): void { + registeredIdentity = identity +} + +export function hostIdentity(): HostIdentity { + if (!registeredIdentity) throw new HostIdentityMissing() + return registeredIdentity +} diff --git a/src/platform/node.ts b/src/platform/node.ts new file mode 100644 index 0000000..a880561 --- /dev/null +++ b/src/platform/node.ts @@ -0,0 +1,11 @@ +import { AsyncLocalStorage } from "node:async_hooks" +import { hostname } from "node:os" +import { registerContextStoreFactory } from "./context-store.js" +import { registerHostIdentity } from "./host-identity.js" + +registerContextStoreFactory(() => new AsyncLocalStorage()) +registerHostIdentity({ + hostname: hostname(), + hostProcessId: process.pid, + runtimeVersion: process.version, +}) diff --git a/src/platform/turn-context-store.ts b/src/platform/turn-context-store.ts new file mode 100644 index 0000000..38969e4 --- /dev/null +++ b/src/platform/turn-context-store.ts @@ -0,0 +1,19 @@ +import type { ContextStore } from "./context-store.js" + +export class TurnContextStore implements ContextStore { + #current: Store | undefined + + run(store: Store, callback: () => Result): Result { + const previous = this.#current + this.#current = store + try { + return callback() + } finally { + this.#current = previous + } + } + + getStore(): Store | undefined { + return this.#current + } +} diff --git a/src/platform/uuid.ts b/src/platform/uuid.ts new file mode 100644 index 0000000..e2051e4 --- /dev/null +++ b/src/platform/uuid.ts @@ -0,0 +1,3 @@ +export function randomUUID(): string { + return globalThis.crypto.randomUUID() +} diff --git a/src/platform/web-locks.ts b/src/platform/web-locks.ts new file mode 100644 index 0000000..ac86d93 --- /dev/null +++ b/src/platform/web-locks.ts @@ -0,0 +1,7 @@ +export function requireWebLocks(): LockManager { + const locks = globalThis.navigator?.locks + if (!locks) { + throw new Error("the Web Locks API is unavailable; leader election cannot run") + } + return locks +} diff --git a/src/reminder-scheduler.ts b/src/reminder-scheduler.ts index f6e3f6c..96b7fe8 100644 --- a/src/reminder-scheduler.ts +++ b/src/reminder-scheduler.ts @@ -1,4 +1,4 @@ -import { randomUUID } from "node:crypto" +import { randomUUID } from "./platform/uuid.js" import { UnknownOperation } from "./errors.js" import type { SolidObjectsRuntime } from "./runtime.js" import { PollingBackoff } from "./polling-backoff.js" diff --git a/src/repository.ts b/src/repository.ts index 6b4606a..33671b6 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -1,5 +1,5 @@ -import { randomUUID } from "node:crypto" -import { hostname } from "node:os" +import { randomUUID } from "./platform/uuid.js" +import { hostIdentity } from "./platform/host-identity.js" import { ActorDestroyed, IdempotencyConflict, @@ -65,9 +65,12 @@ export class Repository { [ processId, kind, - hostname(), - process.pid, - JSON.stringify({ solidObjectsVersion: VERSION, nodeVersion: process.version }), + hostIdentity().hostname, + hostIdentity().hostProcessId, + JSON.stringify({ + solidObjectsVersion: VERSION, + nodeVersion: hostIdentity().runtimeVersion, + }), now, now, ], @@ -126,7 +129,11 @@ export class Repository { WHERE shutdown_state <> 'stopped' AND heartbeat_at_ms > ? AND (hostname <> ? OR host_process_id <> ?) LIMIT 1`, - [now - this.settings.processAliveThresholdMilliseconds, hostname(), process.pid], + [ + now - this.settings.processAliveThresholdMilliseconds, + hostIdentity().hostname, + hostIdentity().hostProcessId, + ], ) return row !== undefined }) diff --git a/src/serialization.ts b/src/serialization.ts index e1e9634..23b6af1 100644 --- a/src/serialization.ts +++ b/src/serialization.ts @@ -2,12 +2,13 @@ import { InvalidPayload, PayloadTooLarge } from "./errors.js" import type { DeepReadonly, JsonValue } from "./types.js" const MAX_NESTING = 100 +const utf8Encoder = new TextEncoder() export function normalizeJson(value: unknown, options: { maxBytes?: number } = {}): JsonValue { const normalized = normalize(value, 0) const encoded = JSON.stringify(normalized) - if (options.maxBytes !== undefined && Buffer.byteLength(encoded) > options.maxBytes) { + if (options.maxBytes !== undefined && utf8Encoder.encode(encoded).length > options.maxBytes) { throw new PayloadTooLarge(`serialized value exceeds ${options.maxBytes} bytes`) } diff --git a/src/transmit-effect.ts b/src/transmit-effect.ts new file mode 100644 index 0000000..a168eae --- /dev/null +++ b/src/transmit-effect.ts @@ -0,0 +1 @@ +export const TRANSMIT_EFFECT = "solid-objects.transmit" diff --git a/src/transmit.ts b/src/transmit.ts new file mode 100644 index 0000000..981cd26 --- /dev/null +++ b/src/transmit.ts @@ -0,0 +1,142 @@ +import { InvalidPayload, NonRetryableError } from "./errors.js" +import type { SolidObjectsRuntime } from "./runtime.js" +import type { EffectContext, JsonObject, JsonValue } from "./types.js" + +export { TRANSMIT_EFFECT } from "./transmit-effect.js" +import { TRANSMIT_EFFECT } from "./transmit-effect.js" + +export interface TransmitEnvelope { + effectId: string + actorType: string + actorId: string + operation: string + arguments: JsonObject +} + +export interface RegisterTransmitOptions { + runtime: SolidObjectsRuntime + deliver: (envelope: TransmitEnvelope) => Promise + effectName?: string +} + +export class InvalidTransmitEnvelope extends NonRetryableError { + constructor(message: string) { + super(message) + this.name = "InvalidTransmitEnvelope" + } +} + +export function registerTransmit(options: RegisterTransmitOptions): void { + const effectName = options.effectName ?? TRANSMIT_EFFECT + options.runtime.registerEffect(effectName, async (argumentsValue, context) => { + parseTransmitEnvelope({ argumentsValue, context }) + const undelivered = await undeliveredEnvelopesThrough({ + runtime: options.runtime, + effectName, + context, + }) + for (const envelope of undelivered) await options.deliver(envelope) + return null + }) +} + +export async function receiveTransmitEnvelope(options: { + runtime: SolidObjectsRuntime + envelope: TransmitEnvelope +}): Promise<{ messageId: string }> { + const { envelope } = options + for (const field of ["effectId", "actorType", "actorId", "operation"] as const) { + if (typeof envelope[field] !== "string" || envelope[field].length === 0) { + throw new InvalidPayload(`transmit envelope requires a non-empty ${field}`) + } + } + const argumentsValue = envelope.arguments ?? {} + if (!isJsonObject(argumentsValue)) { + throw new InvalidPayload("transmit envelope arguments must be a JSON object") + } + const message = await options.runtime.enqueueInternalMessage({ + actorType: envelope.actorType, + actorId: envelope.actorId, + operation: envelope.operation, + argumentsValue, + idempotencyKey: `transmit:${envelope.effectId}`, + }) + return { messageId: message.id } +} + +function parseTransmitEnvelope(input: { + argumentsValue: JsonObject + context: EffectContext +}): TransmitEnvelope { + const { argumentsValue, context } = input + const operation = argumentsValue.operation + if (typeof operation !== "string" || operation.length === 0) { + throw new InvalidTransmitEnvelope("transmit effect arguments require a non-empty operation") + } + const targetArguments = argumentsValue.arguments ?? {} + if (!isJsonObject(targetArguments)) { + throw new InvalidTransmitEnvelope( + "transmit effect arguments must hold a JSON object in arguments", + ) + } + const actorType = argumentsValue.actorType ?? context.actorType + const actorId = argumentsValue.actorId ?? context.actorId + for (const [field, value] of [ + ["actorType", actorType], + ["actorId", actorId], + ] as const) { + if (typeof value !== "string" || value.length === 0) { + throw new InvalidTransmitEnvelope(`transmit effect ${field} must be a non-empty string`) + } + } + return { + effectId: context.id, + actorType: actorType as string, + actorId: actorId as string, + operation, + arguments: targetArguments, + } +} + +async function undeliveredEnvelopesThrough(input: { + runtime: SolidObjectsRuntime + effectName: string + context: EffectContext +}): Promise { + const { runtime, effectName, context } = input + const effects = runtime.repository.table("effects") + const messages = runtime.repository.table("messages") + const rows = await runtime.settings.database.connection((connection) => + connection.all<{ id: string; arguments: string }>( + `SELECT effects.id, effects.arguments FROM ${effects} effects + JOIN ${messages} messages ON messages.id = effects.message_id + WHERE effects.name = ? + AND effects.status IN ('pending', 'processing') + AND messages.actor_type = ? + AND messages.actor_id = ? + AND messages.sequence <= (SELECT sequence FROM ${messages} WHERE id = ?) + ORDER BY messages.sequence, effects.id`, + [effectName, context.actorType, context.actorId, context.sourceMessageId], + ), + ) + const envelopes: TransmitEnvelope[] = [] + for (const row of rows) { + const argumentsValue: JsonValue = JSON.parse(row.arguments) + if (!isJsonObject(argumentsValue)) continue + try { + envelopes.push( + parseTransmitEnvelope({ + argumentsValue, + context: { ...context, id: row.id }, + }), + ) + } catch (error) { + if (!(error instanceof InvalidTransmitEnvelope)) throw error + } + } + return envelopes +} + +function isJsonObject(value: unknown): value is JsonObject { + return typeof value === "object" && value !== null && !Array.isArray(value) +} diff --git a/src/web/index.ts b/src/web/index.ts index 54b37a4..38af78e 100644 --- a/src/web/index.ts +++ b/src/web/index.ts @@ -1,3 +1,4 @@ +import "../platform/node.js" import { randomBytes, timingSafeEqual } from "node:crypto" import { Unauthorized, SolidObjectsError } from "../errors.js" import { diff --git a/src/worker.ts b/src/worker.ts index 89eef67..8bb05f0 100644 --- a/src/worker.ts +++ b/src/worker.ts @@ -1,4 +1,4 @@ -import { randomUUID } from "node:crypto" +import { randomUUID } from "./platform/uuid.js" import type { Actor } from "./actor.js" import { ActorSetupFailed, diff --git a/test/browser-server.mjs b/test/browser-server.mjs index 15cf08f..fbfad9e 100644 --- a/test/browser-server.mjs +++ b/test/browser-server.mjs @@ -1,11 +1,18 @@ import { createServer } from "node:http" import { readFile } from "node:fs/promises" import { extname, resolve } from "node:path" -import { Actor, configure } from "../dist/index.js" +import { Actor, configure, receiveTransmitEnvelope } from "../dist/index.js" import { sqlite } from "../dist/database/sqlite.js" import { createDashboard, createNodeDashboardHandler } from "../dist/web/index.js" const root = resolve(import.meta.dirname, "../dist") +const sqliteWasmRoot = resolve(import.meta.dirname, "../node_modules/@sqlite.org/sqlite-wasm/dist") +const browserFixtureRoot = resolve(import.meta.dirname, "browser") +const contentTypes = { + ".js": "text/javascript; charset=utf-8", + ".mjs": "text/javascript; charset=utf-8", + ".wasm": "application/wasm", +} class DashboardBrowserActor extends Actor { static actorType = "DashboardBrowserActor" count = 0 @@ -13,6 +20,14 @@ class DashboardBrowserActor extends Actor { this.count += 1 } } +class TransmitCounter extends Actor { + static actorType = "TransmitCounter" + count = 0 + increment({ amount = 1 } = {}) { + this.count += amount + return this.count + } +} const runtime = configure({ database: sqlite({ path: ":memory:" }), authorizeMessage: () => true, @@ -20,6 +35,7 @@ const runtime = configure({ authorizeAdministration: () => true, }) runtime.register(DashboardBrowserActor) +runtime.register(TransmitCounter) await runtime.install() await DashboardBrowserActor.ref("browser-room").increment() const sessionValues = new Map() @@ -69,22 +85,82 @@ const server = createServer(async (request, response) => { response.end("") return } + if (pathname === "/sync" && request.method === "POST") { + try { + const envelope = JSON.parse(await readBody(request)) + await receiveTransmitEnvelope({ runtime, envelope }) + await runtime.testing.drain({ roles: ["actors"] }) + response.writeHead(200, { "content-type": "application/json" }) + response.end("{}") + } catch (error) { + response.writeHead(422, { "content-type": "application/json" }) + response.end(JSON.stringify({ error: String(error?.message ?? error) })) + } + return + } + if (pathname === "/sync-state") { + const actorId = new URL(request.url ?? "/", "http://127.0.0.1").searchParams.get("actorId") + const snapshot = await runtime + .ref(TransmitCounter, actorId ?? "missing") + .snapshot() + .catch(() => ({ count: 0 })) + response.writeHead(200, { "content-type": "application/json" }) + response.end(JSON.stringify(snapshot)) + return + } + if ( + pathname === "/sqlite-wasm-worker.mjs" || + pathname === "/runtime-worker.mjs" || + pathname === "/tab-host-worker.mjs" || + pathname === "/transmit-worker.mjs" || + pathname === "/shared-db-worker.mjs" + ) { + await serveFile({ response, path: resolve(browserFixtureRoot, pathname.slice(1)) }) + return + } + if (pathname.startsWith("/vendor/sqlite-wasm/")) { + const vendorPath = resolve(sqliteWasmRoot, `.${pathname.slice("/vendor/sqlite-wasm".length)}`) + if (!vendorPath.startsWith(`${sqliteWasmRoot}/`)) { + response.writeHead(404) + response.end() + return + } + await serveFile({ response, path: vendorPath }) + return + } const path = resolve(root, `.${pathname}`) if (!path.startsWith(`${root}/`)) { response.writeHead(404) response.end() return } + await serveFile({ response, path }) +}) + +function readBody(request) { + return new Promise((resolve, reject) => { + const chunks = [] + request.on("data", (chunk) => chunks.push(chunk)) + request.on("end", () => resolve(Buffer.concat(chunks).toString("utf-8"))) + request.on("error", reject) + }) +} + +async function serveFile({ response, path }) { try { - const contents = await readFile(path) - response.writeHead(200, { - "content-type": extname(path) === ".js" ? "text/javascript; charset=utf-8" : "text/plain", - }) + let contents = await readFile(path) + const contentType = contentTypes[extname(path)] ?? "text/plain" + if (extname(path) === ".js" || extname(path) === ".mjs") { + contents = contents + .toString("utf-8") + .replaceAll('"@sqlite.org/sqlite-wasm"', '"/vendor/sqlite-wasm/index.mjs"') + } + response.writeHead(200, { "content-type": contentType }) response.end(contents) } catch { response.writeHead(404) response.end() } -}) +} server.listen(4179, "127.0.0.1") diff --git a/test/browser/runtime-worker.mjs b/test/browser/runtime-worker.mjs new file mode 100644 index 0000000..a5cc365 --- /dev/null +++ b/test/browser/runtime-worker.mjs @@ -0,0 +1,60 @@ +import { Actor, configure, sqliteWasm } from "/browser/host.js" + +class BrowserCounter extends Actor { + static actorType = "BrowserCounter" + + count = 0 + + increment() { + this.count += 1 + return this.count + } +} + +self.onmessage = async (event) => { + try { + const report = await exercise(event.data) + postMessage({ ok: true, report }) + } catch (error) { + postMessage({ + ok: false, + message: String((error && error.message) || error), + stack: String((error && error.stack) || ""), + }) + } +} + +async function exercise(instructions) { + const database = await openWithRetry() + const runtime = configure({ + database, + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + pollingIntervalMilliseconds: 5, + syncPollingIntervalMilliseconds: 5, + }) + try { + await runtime.install() + const counter = BrowserCounter.ref(instructions.actorId) + const count = await counter.increment() + const snapshot = await counter.snapshot() + return { count, snapshot } + } finally { + await runtime.close() + await database.close() + } +} + +async function openWithRetry() { + let lastError + for (let attempt = 0; attempt < 20; attempt += 1) { + try { + return await sqliteWasm({ path: "solid-objects-runtime.db", storage: "persistent" }) + } catch (error) { + lastError = error + await new Promise((resolve) => setTimeout(resolve, 100)) + } + } + throw lastError +} diff --git a/test/browser/runtime.browser.ts b/test/browser/runtime.browser.ts new file mode 100644 index 0000000..2a180c4 --- /dev/null +++ b/test/browser/runtime.browser.ts @@ -0,0 +1,43 @@ +import { expect, test, type Page } from "@playwright/test" + +interface RuntimeReport { + count: number + snapshot: { count: number } +} + +function runRuntimePhase(page: Page, actorId: string): Promise { + return page.evaluate( + (actorIdValue) => + new Promise((resolve, reject) => { + const worker = new Worker("/runtime-worker.mjs", { type: "module" }) + worker.onmessage = (event) => { + worker.terminate() + if (event.data.ok) { + resolve(event.data.report as RuntimeReport) + return + } + reject(new Error(`${event.data.message}\n${event.data.stack}`)) + } + worker.onerror = (event) => { + worker.terminate() + reject(new Error(event.message)) + } + worker.postMessage({ actorId: actorIdValue }) + }), + actorId, + ) +} + +test("runs the full runtime in a browser worker with durable actor state", async ({ page }) => { + await page.goto("/") + + const first = await runRuntimePhase(page, "browser-counter") + expect(first.count).toBe(1) + expect(first.snapshot).toEqual({ count: 1 }) + + await page.reload() + + const second = await runRuntimePhase(page, "browser-counter") + expect(second.count).toBe(2) + expect(second.snapshot).toEqual({ count: 2 }) +}) diff --git a/test/browser/shared-db-worker.mjs b/test/browser/shared-db-worker.mjs new file mode 100644 index 0000000..669e5d7 --- /dev/null +++ b/test/browser/shared-db-worker.mjs @@ -0,0 +1,47 @@ +import { Actor, configure, sharedSqliteWasm } from "/browser/host.js" + +class Counter extends Actor { + static actorType = "Counter" + + count = 0 + + increment({ amount = 1 } = {}) { + this.count += amount + return this.count + } +} + +const database = sharedSqliteWasm({ path: "shared-app.db", name: "shared-e2e" }) +const runtime = configure({ + database, + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + pollingIntervalMilliseconds: 5, + syncPollingIntervalMilliseconds: 5, + idlePollingIntervalMilliseconds: 100, + processAliveThresholdMilliseconds: 750, + leaseDurationMilliseconds: 750, + leaseRenewalIntervalMilliseconds: 250, +}) +const installed = runtime.install() + +self.onmessage = async (event) => { + const { requestId, command, actorId } = event.data + try { + if (command === "role") { + postMessage({ requestId, ok: true, value: database.role() }) + return + } + await installed + const value = await Counter.ref(actorId).increment() + postMessage({ requestId, ok: true, value }) + } catch (error) { + postMessage({ + requestId, + ok: false, + message: String((error && error.message) || error), + stack: String((error && error.stack) || ""), + }) + } +} diff --git a/test/browser/shared-db.browser.ts b/test/browser/shared-db.browser.ts new file mode 100644 index 0000000..cf4a364 --- /dev/null +++ b/test/browser/shared-db.browser.ts @@ -0,0 +1,52 @@ +import { expect, test, type Page } from "@playwright/test" + +async function startWorker(page: Page): Promise { + await page.goto("/") + await page.evaluate(() => { + const worker = new Worker("/shared-db-worker.mjs", { type: "module" }) + const waiters = new Map< + string, + { resolve: (value: unknown) => void; reject: (error: Error) => void } + >() + worker.onmessage = (event) => { + const waiter = waiters.get(event.data.requestId) + if (!waiter) return + waiters.delete(event.data.requestId) + if (event.data.ok) waiter.resolve(event.data.value) + else waiter.reject(new Error(`${event.data.message}\n${event.data.stack}`)) + } + const send = (command: string, actorId?: string) => + new Promise((resolve, reject) => { + const requestId = crypto.randomUUID() + waiters.set(requestId, { resolve, reject }) + worker.postMessage({ requestId, command, actorId }) + }) + Object.assign(window, { __sharedSend: send }) + }) +} + +function send(page: Page, options: { command: string; actorId?: string }): Promise { + return page.evaluate( + ({ command, actorId }) => + ( + window as unknown as { __sharedSend(command: string, actorId?: string): Promise } + ).__sharedSend(command, actorId), + options, + ) +} + +test("plain refs share one durable database across tabs and fail over", async ({ context }) => { + const actorId = `transparent-${Date.now()}` + const pageA = await context.newPage() + const pageB = await context.newPage() + await startWorker(pageA) + await startWorker(pageB) + + expect(await send(pageA, { command: "increment", actorId })).toBe(1) + expect(await send(pageB, { command: "increment", actorId })).toBe(2) + + await pageA.close() + + expect(await send(pageB, { command: "increment", actorId })).toBe(3) + await expect.poll(() => send(pageB, { command: "role" })).toBe("holder") +}) diff --git a/test/browser/sqlite-wasm-worker.mjs b/test/browser/sqlite-wasm-worker.mjs new file mode 100644 index 0000000..d2e3a86 --- /dev/null +++ b/test/browser/sqlite-wasm-worker.mjs @@ -0,0 +1,83 @@ +import { registerContextStoreFactory } from "/platform/context-store.js" +import { TurnContextStore } from "/platform/turn-context-store.js" +import { sqliteWasm } from "/database/sqlite-wasm.js" +import { withDatabaseDeadline } from "/database/deadline.js" +import { DatabaseDeadlineExceeded } from "/errors.js" + +registerContextStoreFactory(() => new TurnContextStore()) + +self.onmessage = async (event) => { + try { + const report = await exercise(event.data) + postMessage({ ok: true, report }) + } catch (error) { + postMessage({ + ok: false, + message: String((error && error.message) || error), + stack: String((error && error.stack) || ""), + }) + } +} + +async function exercise(instructions) { + const database = await openWithRetry() + try { + await database.connection((connection) => + connection.run("CREATE TABLE IF NOT EXISTS visits(id INTEGER PRIMARY KEY, phase TEXT)"), + ) + await database.transaction((connection) => + connection.run("INSERT INTO visits(phase) VALUES (?)", [instructions.phase]), + ) + let rollbackMessage + try { + await database.transaction(async (connection) => { + await connection.run("INSERT INTO visits(phase) VALUES (?)", ["rolled-back"]) + throw new Error("abort") + }) + } catch (error) { + rollbackMessage = String(error.message) + } + let deadlineOutcome = "not-run" + try { + await withDatabaseDeadline({ timeoutMilliseconds: 25 }, () => + database.transaction(async (connection) => { + await connection.run("INSERT INTO visits(phase) VALUES (?)", ["deadline-overrun"]) + await new Promise((resolve) => setTimeout(resolve, 100)) + }), + ) + deadlineOutcome = "committed" + } catch (error) { + deadlineOutcome = + error instanceof DatabaseDeadlineExceeded ? "rolled-back" : `error:${error.message}` + } + const overrun = await database.connection((connection) => + connection.all("SELECT phase FROM visits WHERE phase = 'deadline-overrun'"), + ) + const rows = await database.connection((connection) => + connection.all("SELECT phase FROM visits WHERE phase <> 'deadline-overrun' ORDER BY id"), + ) + const now = await database.connection((connection) => connection.nowMilliseconds()) + return { + phases: rows.map((row) => row.phase), + rollbackMessage, + deadlineOutcome, + overrunRows: overrun.length, + clockSkewMilliseconds: Math.abs(now - Date.now()), + } + } finally { + await database.close() + } +} + +async function openWithRetry() { + let lastError + for (let attempt = 0; attempt < 20; attempt += 1) { + try { + return await sqliteWasm({ path: "solid-objects-browser.db", storage: "persistent" }) + } catch (error) { + lastError = error + await new Promise((resolve) => setTimeout(resolve, 100)) + } + } + throw lastError +} diff --git a/test/browser/sqlite-wasm.browser.ts b/test/browser/sqlite-wasm.browser.ts new file mode 100644 index 0000000..b2402ac --- /dev/null +++ b/test/browser/sqlite-wasm.browser.ts @@ -0,0 +1,48 @@ +import { expect, test, type Page } from "@playwright/test" + +interface WorkerReport { + phases: string[] + rollbackMessage: string + deadlineOutcome: string + overrunRows: number + clockSkewMilliseconds: number +} + +function runWorkerPhase(page: Page, phase: string): Promise { + return page.evaluate( + (phaseName) => + new Promise((resolve, reject) => { + const worker = new Worker("/sqlite-wasm-worker.mjs", { type: "module" }) + worker.onmessage = (event) => { + worker.terminate() + if (event.data.ok) { + resolve(event.data.report as WorkerReport) + return + } + reject(new Error(`${event.data.message}\n${event.data.stack}`)) + } + worker.onerror = (event) => { + worker.terminate() + reject(new Error(event.message)) + } + worker.postMessage({ phase: phaseName }) + }), + phase, + ) +} + +test("persists SQLite WASM state in OPFS across page reloads", async ({ page }) => { + await page.goto("/") + + const first = await runWorkerPhase(page, "first") + expect(first.phases).toEqual(["first"]) + expect(first.rollbackMessage).toBe("abort") + expect(first.deadlineOutcome).toBe("rolled-back") + expect(first.overrunRows).toBe(0) + expect(first.clockSkewMilliseconds).toBeLessThan(5_000) + + await page.reload() + + const second = await runWorkerPhase(page, "second") + expect(second.phases).toEqual(["first", "second"]) +}) diff --git a/test/browser/tab-host-worker.mjs b/test/browser/tab-host-worker.mjs new file mode 100644 index 0000000..ecd6db0 --- /dev/null +++ b/test/browser/tab-host-worker.mjs @@ -0,0 +1,97 @@ +import { Actor, configure, connectTabClient, sqliteWasm, startTabHost } from "/browser/host.js" + +class TabCounter extends Actor { + static actorType = "TabCounter" + + count = 0 + + increment({ amount = 1 } = {}) { + this.count += amount + return this.count + } +} + +let lastHostError = "" + +const host = startTabHost({ + name: "tab-host-e2e", + onError: (error) => { + lastHostError = `${error.name}: ${error.message}` + }, + startRuntime: async () => { + const database = await openWithRetry() + try { + const runtime = configure({ + database, + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + pollingIntervalMilliseconds: 5, + syncPollingIntervalMilliseconds: 5, + idlePollingIntervalMilliseconds: 100, + processAliveThresholdMilliseconds: 750, + leaseDurationMilliseconds: 750, + leaseRenewalIntervalMilliseconds: 250, + }) + runtime.register(TabCounter) + await runtime.install() + return { + runtime, + close: async () => { + await runtime.close() + await database.close() + }, + } + } catch (error) { + await database.close() + throw error + } + }, +}) + +const client = connectTabClient({ + name: "tab-host-e2e", + retryIntervalMilliseconds: 100, + timeoutMilliseconds: 15_000, +}) + +self.onmessage = async (event) => { + const { requestId, command, actorId } = event.data + try { + if (command === "role") { + postMessage({ requestId, ok: true, value: host.role() }) + return + } + if (command === "diagnose") { + postMessage({ requestId, ok: true, value: `${host.role()} ${lastHostError}` }) + return + } + const value = await client.invoke({ + actorType: "TabCounter", + actorId, + operation: "increment", + arguments: {}, + }) + postMessage({ requestId, ok: true, value }) + } catch (error) { + postMessage({ + requestId, + ok: false, + message: String((error && error.message) || error), + stack: String((error && error.stack) || ""), + }) + } +} + +async function openWithRetry() { + let lastError + for (let attempt = 0; attempt < 50; attempt += 1) { + try { + return await sqliteWasm({ path: "solid-objects-tab-host.db", storage: "persistent" }) + } catch (error) { + lastError = error + await new Promise((resolve) => setTimeout(resolve, 100)) + } + } + throw lastError +} diff --git a/test/browser/tab-host.browser.ts b/test/browser/tab-host.browser.ts new file mode 100644 index 0000000..fbf58d5 --- /dev/null +++ b/test/browser/tab-host.browser.ts @@ -0,0 +1,54 @@ +import { expect, test, type Page } from "@playwright/test" + +async function startHostWorker(page: Page): Promise { + await page.goto("/") + await page.evaluate(() => { + const worker = new Worker("/tab-host-worker.mjs", { type: "module" }) + const waiters = new Map< + string, + { resolve: (value: unknown) => void; reject: (error: Error) => void } + >() + worker.onmessage = (event) => { + const waiter = waiters.get(event.data.requestId) + if (!waiter) return + waiters.delete(event.data.requestId) + if (event.data.ok) waiter.resolve(event.data.value) + else waiter.reject(new Error(`${event.data.message}\n${event.data.stack}`)) + } + const send = (command: string, actorId?: string) => + new Promise((resolve, reject) => { + const requestId = crypto.randomUUID() + waiters.set(requestId, { resolve, reject }) + worker.postMessage({ requestId, command, actorId }) + }) + Object.assign(window, { __tabHostSend: send }) + }) +} + +function send(page: Page, options: { command: string; actorId?: string }): Promise { + return page.evaluate( + ({ command, actorId }) => + ( + window as unknown as { __tabHostSend(command: string, actorId?: string): Promise } + ).__tabHostSend(command, actorId), + options, + ) +} + +test("shares one durable runtime across tabs and fails over", async ({ context }) => { + const actorId = `shared-${Date.now()}` + const pageA = await context.newPage() + const pageB = await context.newPage() + await startHostWorker(pageA) + await expect.poll(() => send(pageA, { command: "role" })).toBe("leader") + await startHostWorker(pageB) + + expect(await send(pageA, { command: "increment", actorId })).toBe(1) + expect(await send(pageB, { command: "increment", actorId })).toBe(2) + expect(await send(pageB, { command: "role" })).toBe("follower") + + await pageA.close() + + expect(await send(pageB, { command: "increment", actorId })).toBe(3) + await expect.poll(() => send(pageB, { command: "role" })).toBe("leader") +}) diff --git a/test/browser/transmit-worker.mjs b/test/browser/transmit-worker.mjs new file mode 100644 index 0000000..5d80c6c --- /dev/null +++ b/test/browser/transmit-worker.mjs @@ -0,0 +1,68 @@ +import { Actor, configure, registerTransmit, sqliteWasm } from "/browser/host.js" + +class TransmitCounter extends Actor { + static actorType = "TransmitCounter" + + count = 0 + + increment({ amount = 1 } = {}) { + this.count += amount + this.transmit().increment({ amount }) + return this.count + } +} + +let stopRunning + +self.onmessage = async (event) => { + const { requestId, command, actorId, amounts } = event.data + try { + if (command === "stop") { + await stopRunning?.() + postMessage({ requestId, ok: true, value: "stopped" }) + return + } + const database = await sqliteWasm({ path: "sync-local.db" }) + const runtime = configure({ + database, + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + pollingIntervalMilliseconds: 5, + syncPollingIntervalMilliseconds: 5, + }) + registerTransmit({ + runtime, + deliver: async (envelope) => { + const response = await fetch("/sync", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(envelope), + }) + if (!response.ok) throw new Error(`sync transport failed with ${response.status}`) + }, + }) + runtime.register(TransmitCounter) + await runtime.install() + const runAbort = new AbortController() + const running = runtime.run(runAbort.signal) + stopRunning = async () => { + runAbort.abort() + await running.catch(() => undefined) + await runtime.close() + await database.close() + } + let count = 0 + for (const amount of amounts) { + count = await runtime.ref(TransmitCounter, actorId).increment({ amount }) + } + postMessage({ requestId, ok: true, value: count }) + } catch (error) { + postMessage({ + requestId, + ok: false, + message: String((error && error.message) || error), + stack: String((error && error.stack) || ""), + }) + } +} diff --git a/test/browser/transmit.browser.ts b/test/browser/transmit.browser.ts new file mode 100644 index 0000000..099f296 --- /dev/null +++ b/test/browser/transmit.browser.ts @@ -0,0 +1,46 @@ +import { expect, test, type Page } from "@playwright/test" + +function runWorkerCommand( + page: Page, + options: { command: string; actorId?: string; amounts?: number[] }, +): Promise { + return page.evaluate( + (input) => + new Promise((resolve, reject) => { + const scope = window as unknown as { __syncWorker?: Worker } + scope.__syncWorker ??= new Worker("/transmit-worker.mjs", { type: "module" }) + const worker = scope.__syncWorker + const requestId = crypto.randomUUID() + const onMessage = (event: MessageEvent) => { + if (event.data.requestId !== requestId) return + worker.removeEventListener("message", onMessage) + if (event.data.ok) resolve(event.data.value) + else reject(new Error(`${event.data.message}\n${event.data.stack}`)) + } + worker.addEventListener("message", onMessage) + worker.postMessage({ requestId, ...input }) + }), + options, + ) +} + +test("drains the browser outbox into the server runtime", async ({ page, request }) => { + const actorId = `transmit-${Date.now()}` + await page.goto("/") + + const localCount = await runWorkerCommand(page, { command: "run", actorId, amounts: [2, 3] }) + expect(localCount).toBe(5) + + await expect + .poll( + async () => { + const response = await request.get(`/sync-state?actorId=${actorId}`) + const body = (await response.json()) as { count?: number } + return body.count ?? 0 + }, + { timeout: 15_000 }, + ) + .toBe(5) + + await runWorkerCommand(page, { command: "stop" }) +}) diff --git a/test/mysql.test.ts b/test/mysql.test.ts index 080db32..7aee4fb 100644 --- a/test/mysql.test.ts +++ b/test/mysql.test.ts @@ -1,7 +1,8 @@ import { afterEach, describe, expect, it } from "vitest" import { Actor } from "../src/actor.js" import { mysql, mysqlSql, type MySQLDatabase } from "../src/database/mysql.js" -import { configure, type SolidObjectsRuntime } from "../src/runtime.js" +import { configure, createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" +import { receiveTransmitEnvelope, registerTransmit } from "../src/transmit.js" import { SyncTimeout } from "../src/errors.js" import { DatabaseDeadlineExceeded } from "../src/errors.js" import { withDatabaseDeadline } from "../src/database/deadline.js" @@ -9,6 +10,20 @@ import type { JsonObject } from "../src/types.js" import { createDashboard } from "../src/web/index.js" const connectionString = process.env.SOLID_OBJECTS_DATABASE_URL +class TransmitProofCounter extends Actor { + static override readonly actorType = "TransmitProofCounter" + + count = 0 + applied: number[] = [] + + increment({ amount = 1 }: { amount?: number } = {}): number { + this.count += amount + this.applied = [...this.applied, amount] + this.transmit().increment!({ amount }) + return this.count + } +} + const describeMySQL = connectionString?.startsWith("mysql:") ? describe : describe.skip const quietLogger = { debug: () => {}, info: () => {}, warn: () => {}, error: () => {} } @@ -117,6 +132,68 @@ describe("MySQL SQL compatibility", () => { }) describeMySQL("MySQL adapter", () => { + it("stages, drains, and ingests transmit envelopes on MySQL", async () => { + if (!connectionString) throw new Error("MySQL connection string is required") + const localDatabase = mysql({ connectionString, maximumConnections: 5 }) + const serverDatabase = mysql({ connectionString, maximumConnections: 5 }) + const settings = { + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + pollingIntervalMilliseconds: 1, + syncPollingIntervalMilliseconds: 1, + maxAttempts: 8, + retryDelayMilliseconds: () => 0, + logger: quietLogger, + } + const local = createRuntime({ + database: localDatabase, + tableNamePrefix: "transmit_local_", + ...settings, + }) + const server = createRuntime({ + database: serverDatabase, + tableNamePrefix: "transmit_server_", + ...settings, + }) + try { + let failuresRemaining = 1 + registerTransmit({ + runtime: local, + deliver: async (envelope) => { + if (failuresRemaining > 0) { + failuresRemaining -= 1 + throw new Error("network down") + } + await receiveTransmitEnvelope({ runtime: server, envelope }) + await receiveTransmitEnvelope({ runtime: server, envelope }) + }, + }) + server.register(TransmitProofCounter) + await local.install() + await server.install() + + const actorId = `proof-${crypto.randomUUID()}` + const counter = local.ref(TransmitProofCounter, actorId) + await counter.increment({ amount: 1 }) + await counter.increment({ amount: 2 }) + await local.testing.drain({ roles: ["actors", "effects"], maxPasses: 20 }) + await server.testing.drain({ roles: ["actors"] }) + + await expect(server.ref(TransmitProofCounter, actorId).snapshot()).resolves.toEqual({ + count: 3, + applied: [1, 2], + }) + } finally { + await local.testing.reset() + await server.testing.reset() + await local.close() + await server.close() + await localDatabase.close() + await serverDatabase.close() + } + }) + it("enforces deadlines without leaking session settings", async () => { if (!connectionString) throw new Error("MySQL connection string is required") database = mysql({ connectionString, maximumConnections: 1 }) diff --git a/test/platform-context-store.test.ts b/test/platform-context-store.test.ts new file mode 100644 index 0000000..e909757 --- /dev/null +++ b/test/platform-context-store.test.ts @@ -0,0 +1,104 @@ +import { describe, expect, it } from "vitest" +import { + ContextStoreFactoryMissing, + createContextStore, + registerContextStoreFactory, +} from "../src/platform/context-store.js" +import { TurnContextStore } from "../src/platform/turn-context-store.js" + +describe("context store registry", () => { + it("returns undefined from getStore before any run", () => { + const store = createContextStore<{ value: number }>() + expect(store.getStore()).toBeUndefined() + }) + + it("throws a clear error when run executes without a registered factory", () => { + const store = createContextStore<{ value: number }>() + expect(() => store.run({ value: 1 }, () => "ignored")).toThrow(ContextStoreFactoryMissing) + }) + + it("uses the AsyncLocalStorage factory after the node platform module loads", async () => { + await import("../src/platform/node.js") + const store = createContextStore<{ value: number }>() + const observed = await store.run({ value: 7 }, async () => { + await Promise.resolve() + return store.getStore()?.value + }) + expect(observed).toBe(7) + expect(store.getStore()).toBeUndefined() + }) + + it("keeps the AsyncLocalStorage store across interleaved async runs", async () => { + await import("../src/platform/node.js") + const store = createContextStore<{ value: number }>() + const results = await Promise.all( + [1, 2, 3].map((value) => + store.run({ value }, async () => { + await new Promise((resolve) => setTimeout(resolve, 5 - value)) + return store.getStore()?.value + }), + ), + ) + expect(results).toEqual([1, 2, 3]) + }) + + it("lets a later registration replace the factory for new stores", async () => { + registerContextStoreFactory(() => new TurnContextStore()) + const store = createContextStore<{ value: string }>() + const observed = store.run({ value: "turn" }, () => store.getStore()?.value) + expect(observed).toBe("turn") + }) +}) + +describe("turn context store", () => { + it("scopes a synchronous run and restores the previous store", () => { + const store = new TurnContextStore<{ value: number }>() + const observed = store.run({ value: 1 }, () => { + const outer = store.getStore()?.value + const inner = store.run({ value: 2 }, () => store.getStore()?.value) + const restored = store.getStore()?.value + return { outer, inner, restored } + }) + expect(observed).toEqual({ outer: 1, inner: 2, restored: 1 }) + expect(store.getStore()).toBeUndefined() + }) + + it("restores the previous store when the callback throws", () => { + const store = new TurnContextStore<{ value: number }>() + expect(() => + store.run({ value: 1 }, () => { + throw new Error("boom") + }), + ).toThrow("boom") + expect(store.getStore()).toBeUndefined() + }) + + it("scopes only the synchronous part of an async callback", async () => { + const store = new TurnContextStore<{ value: number }>() + const observed = await store.run({ value: 4 }, async () => { + const beforeAwait = store.getStore()?.value + await Promise.resolve() + const afterAwait = store.getStore()?.value + return { beforeAwait, afterAwait } + }) + expect(observed).toEqual({ beforeAwait: 4, afterAwait: undefined }) + expect(store.getStore()).toBeUndefined() + }) + + it("never leaks a store into tasks that interleave with an async callback", async () => { + const store = new TurnContextStore<{ value: number }>() + let releaseCallback = () => {} + const held = new Promise((resolve) => { + releaseCallback = resolve + }) + const running = store.run({ value: 9 }, async () => { + await held + }) + const interleaved = await new Promise((resolve) => + setTimeout(() => resolve(store.getStore()?.value), 1), + ) + releaseCallback() + await running + expect(interleaved).toBeUndefined() + }) +}) diff --git a/test/platform-host-identity.test.ts b/test/platform-host-identity.test.ts new file mode 100644 index 0000000..74dde60 --- /dev/null +++ b/test/platform-host-identity.test.ts @@ -0,0 +1,26 @@ +import { describe, expect, it } from "vitest" +import { + HostIdentityMissing, + hostIdentity, + registerHostIdentity, +} from "../src/platform/host-identity.js" + +describe("host identity registry", () => { + it("throws a clear error before registration", () => { + expect(() => hostIdentity()).toThrow(HostIdentityMissing) + }) + + it("returns the node identity after the node platform module loads", async () => { + await import("../src/platform/node.js") + const identity = hostIdentity() + expect(identity.hostname.length).toBeGreaterThan(0) + expect(identity.hostProcessId).toBe(process.pid) + expect(identity.runtimeVersion).toBe(process.version) + }) + + it("lets a later registration replace the identity", () => { + const replacement = { hostname: "browser-host", hostProcessId: 7, runtimeVersion: "browser" } + registerHostIdentity(replacement) + expect(hostIdentity()).toEqual(replacement) + }) +}) diff --git a/test/platform-uuid.test.ts b/test/platform-uuid.test.ts new file mode 100644 index 0000000..8cad122 --- /dev/null +++ b/test/platform-uuid.test.ts @@ -0,0 +1,15 @@ +import { describe, expect, it } from "vitest" +import { randomUUID } from "../src/platform/uuid.js" + +const UUID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/ + +describe("platform uuid", () => { + it("returns a version 4 UUID", () => { + expect(randomUUID()).toMatch(UUID_PATTERN) + }) + + it("returns a distinct value on every call", () => { + const values = new Set(Array.from({ length: 100 }, () => randomUUID())) + expect(values.size).toBe(100) + }) +}) diff --git a/test/postgresql.test.ts b/test/postgresql.test.ts index baf0f3f..a1aa8c0 100644 --- a/test/postgresql.test.ts +++ b/test/postgresql.test.ts @@ -7,7 +7,8 @@ import { postgresqlWakeUp, type PostgreSQLDatabase, } from "../src/database/postgresql.js" -import { configure, type SolidObjectsRuntime } from "../src/runtime.js" +import { configure, createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" +import { receiveTransmitEnvelope, registerTransmit } from "../src/transmit.js" import type { RealtimeEnvelope } from "../src/browser/index.js" import { SyncTimeout } from "../src/errors.js" import { DatabaseDeadlineExceeded } from "../src/errors.js" @@ -95,6 +96,20 @@ class PostgreSQLMigratingActor extends Actor { } } +class TransmitProofCounter extends Actor { + static override readonly actorType = "TransmitProofCounter" + + count = 0 + applied: number[] = [] + + increment({ amount = 1 }: { amount?: number } = {}): number { + this.count += amount + this.applied = [...this.applied, amount] + this.transmit().increment!({ amount }) + return this.count + } +} + let runtime: SolidObjectsRuntime | undefined let database: PostgreSQLDatabase | undefined let wakeUps: PostgreSQLWakeUpAdapter[] = [] @@ -151,6 +166,68 @@ describe("PostgreSQL SQL parameters", () => { }) describePostgreSQL("PostgreSQL adapter", () => { + it("stages, drains, and ingests transmit envelopes on PostgreSQL", async () => { + if (!connectionString) throw new Error("PostgreSQL connection string is required") + const localDatabase = postgresql({ connectionString, maximumConnections: 5 }) + const serverDatabase = postgresql({ connectionString, maximumConnections: 5 }) + const settings = { + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + pollingIntervalMilliseconds: 1, + syncPollingIntervalMilliseconds: 1, + maxAttempts: 8, + retryDelayMilliseconds: () => 0, + logger: quietLogger, + } + const local = createRuntime({ + database: localDatabase, + tableNamePrefix: "transmit_local_", + ...settings, + }) + const server = createRuntime({ + database: serverDatabase, + tableNamePrefix: "transmit_server_", + ...settings, + }) + try { + let failuresRemaining = 1 + registerTransmit({ + runtime: local, + deliver: async (envelope) => { + if (failuresRemaining > 0) { + failuresRemaining -= 1 + throw new Error("network down") + } + await receiveTransmitEnvelope({ runtime: server, envelope }) + await receiveTransmitEnvelope({ runtime: server, envelope }) + }, + }) + server.register(TransmitProofCounter) + await local.install() + await server.install() + + const actorId = `proof-${crypto.randomUUID()}` + const counter = local.ref(TransmitProofCounter, actorId) + await counter.increment({ amount: 1 }) + await counter.increment({ amount: 2 }) + await local.testing.drain({ roles: ["actors", "effects"], maxPasses: 20 }) + await server.testing.drain({ roles: ["actors"] }) + + await expect(server.ref(TransmitProofCounter, actorId).snapshot()).resolves.toEqual({ + count: 3, + applied: [1, 2], + }) + } finally { + await local.testing.reset() + await server.testing.reset() + await local.close() + await server.close() + await localDatabase.close() + await serverDatabase.close() + } + }) + it("enforces and clears database deadlines", async () => { if (!connectionString) throw new Error("PostgreSQL connection string is required") database = postgresql({ connectionString, maximumConnections: 1 }) diff --git a/test/shared-sqlite-wasm.test.ts b/test/shared-sqlite-wasm.test.ts new file mode 100644 index 0000000..c2c5cce --- /dev/null +++ b/test/shared-sqlite-wasm.test.ts @@ -0,0 +1,197 @@ +import { afterEach, describe, expect, it } from "vitest" +import "../src/platform/node.js" +import { Actor } from "../src/actor.js" +import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" +import { + sharedSqliteWasm, + type SharedSQLiteWasmDatabase, +} from "../src/database/shared-sqlite-wasm.js" + +class SharedCounter extends Actor { + static override readonly actorType = "SharedCounter" + + count = 0 + + increment({ amount = 1 }: { amount?: number } = {}): number { + this.count += amount + return this.count + } +} + +const databases: SharedSQLiteWasmDatabase[] = [] +const runtimes: SolidObjectsRuntime[] = [] + +afterEach(async () => { + try { + for (const runtime of [...runtimes].reverse()) await runtime.close() + for (const database of [...databases].reverse()) await database.close() + } finally { + runtimes.length = 0 + databases.length = 0 + } +}) + +function shared(name: string, overrides: { sessionIdleTimeoutMilliseconds?: number } = {}) { + const database = sharedSqliteWasm({ path: ":memory:", name, ...overrides }) + databases.push(database) + return database +} + +async function eventually(condition: () => boolean): Promise { + for (let attempt = 0; attempt < 200; attempt += 1) { + if (condition()) return + await new Promise((resolve) => setTimeout(resolve, 25)) + } + throw new Error("condition never became true") +} + +const describeShared = globalThis.navigator?.locks ? describe : describe.skip + +describeShared("shared SQLite WASM adapter", () => { + it("executes statements from the holder and a remote instance", async () => { + const name = `statements-${crypto.randomUUID()}` + const first = shared(name) + const second = shared(name) + + await first.connection(async (connection) => { + await connection.run("CREATE TABLE records(id INTEGER PRIMARY KEY, name TEXT)") + await connection.run("INSERT INTO records(name) VALUES (?)", ["from-first"]) + }) + const result = await second.connection((connection) => + connection.run("INSERT INTO records(name) VALUES (?)", ["from-second"]), + ) + expect(result.changes).toBe(1) + expect(result.lastInsertId).toBe("2") + + const rows = await second.connection((connection) => + connection.all<{ name: string }>("SELECT name FROM records ORDER BY id"), + ) + expect(rows).toEqual([{ name: "from-first" }, { name: "from-second" }]) + + const now = await second.connection((connection) => connection.nowMilliseconds()) + expect(Math.abs(now - Date.now())).toBeLessThan(5_000) + expect([first.role(), second.role()].sort()).toEqual(["holder", "remote"]) + }) + + it("commits and rolls back remote transactions", async () => { + const name = `transactions-${crypto.randomUUID()}` + const first = shared(name) + const second = shared(name) + await first.connection((connection) => + connection.run("CREATE TABLE records(id INTEGER PRIMARY KEY)"), + ) + + await second.transaction(async (connection) => { + expect(second.transactionActive()).toBe(true) + await connection.run("INSERT INTO records(id) VALUES (1)") + }) + await expect( + second.transaction(async (connection) => { + await connection.run("INSERT INTO records(id) VALUES (2)") + throw new Error("boom") + }), + ).rejects.toThrow("boom") + + const rows = await first.connection((connection) => + connection.all<{ id: number | bigint }>("SELECT id FROM records ORDER BY id"), + ) + expect(rows.map((row) => Number(row.id))).toEqual([1]) + }) + + it("serializes sessions across instances", async () => { + const name = `serialize-${crypto.randomUUID()}` + const first = shared(name) + const second = shared(name) + await first.connection((connection) => connection.run("SELECT 1")) + + const order: string[] = [] + let release: (() => void) | undefined + const blocked = second.connection(async () => { + order.push("second:start") + await new Promise((resolve) => { + release = resolve + }) + order.push("second:end") + }) + await eventually(() => release !== undefined) + const chased = first.connection(async () => { + order.push("first:start") + order.push("first:end") + }) + await new Promise((resolve) => setTimeout(resolve, 50)) + release?.() + await Promise.all([blocked, chased]) + + expect(order).toEqual(["second:start", "second:end", "first:start", "first:end"]) + }) + + it("runs two full runtimes against one shared database", async () => { + const name = `runtimes-${crypto.randomUUID()}` + const first = shared(name) + const second = shared(name) + const settings = { + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + pollingIntervalMilliseconds: 1, + syncPollingIntervalMilliseconds: 1, + } + const firstRuntime = createRuntime({ database: first, ...settings }) + const secondRuntime = createRuntime({ database: second, ...settings }) + runtimes.push(firstRuntime, secondRuntime) + + await firstRuntime.install() + await secondRuntime.install() + + expect(await firstRuntime.ref(SharedCounter, "shared").increment({ amount: 2 })).toBe(2) + expect(await secondRuntime.ref(SharedCounter, "shared").increment()).toBe(3) + expect(await secondRuntime.ref(SharedCounter, "shared").snapshot()).toEqual({ count: 3 }) + }) + + it("fails over to the next instance when the holder closes", async () => { + const name = `failover-${crypto.randomUUID()}` + const first = shared(name) + const second = shared(name) + await first.connection((connection) => connection.run("SELECT 1")) + await second.connection((connection) => connection.run("SELECT 1")) + expect(first.role()).toBe("holder") + + await first.close() + databases.splice(databases.indexOf(first), 1) + + await second.connection(async (connection) => { + await connection.run("CREATE TABLE IF NOT EXISTS records(id INTEGER PRIMARY KEY)") + await connection.run("INSERT INTO records(id) VALUES (1)") + }) + expect(second.role()).toBe("holder") + }) + + it("recovers the session slot when a remote instance dies mid-session", async () => { + const name = `watchdog-${crypto.randomUUID()}` + const first = shared(name, { sessionIdleTimeoutMilliseconds: 100 }) + const second = shared(name) + await first.connection((connection) => connection.run("SELECT 1")) + + let opened = false + let releaseAbandoned = () => {} + const held = new Promise((resolve) => { + releaseAbandoned = resolve + }) + const abandoned = second + .connection(async () => { + opened = true + await held + }) + .catch(() => undefined) + await eventually(() => opened) + await second.close() + databases.splice(databases.indexOf(second), 1) + + const value = await first.connection((connection) => + connection.get<{ answer: number | bigint }>("SELECT 42 AS answer"), + ) + expect(Number(value?.answer)).toBe(42) + releaseAbandoned() + await abandoned + }) +}) diff --git a/test/sqlite-wasm.test.ts b/test/sqlite-wasm.test.ts new file mode 100644 index 0000000..e5f84e0 --- /dev/null +++ b/test/sqlite-wasm.test.ts @@ -0,0 +1,213 @@ +import { afterEach, describe, expect, it } from "vitest" +import "../src/platform/node.js" +import { Actor } from "../src/actor.js" +import { configure, type SolidObjectsRuntime } from "../src/runtime.js" +import { sqliteWasm, type SQLiteWasmDatabase } from "../src/database/sqlite-wasm.js" +import { withDatabaseDeadline } from "../src/database/deadline.js" +import { DatabaseDeadlineExceeded } from "../src/errors.js" + +class WasmCounter extends Actor { + static override readonly actorType = "WasmCounter" + + count = 0 + + increment({ amount = 1 }: { amount?: number } = {}): number { + this.count += amount + return this.count + } +} + +let database: SQLiteWasmDatabase | undefined +let runtime: SolidObjectsRuntime | undefined + +afterEach(async () => { + await runtime?.close() + await database?.close() + database = undefined + runtime = undefined +}) + +describe("SQLite WASM adapter", () => { + it("reports the sqlite family and a wasm schema identity", async () => { + database = await sqliteWasm({ path: ":memory:" }) + expect(database.family).toBe("sqlite") + expect(database.schemaIdentity).toBe("solid-objects-wasm-v1") + }) + + it("runs statements and reports changes and the last insert id", async () => { + database = await sqliteWasm({ path: ":memory:" }) + await database.connection((connection) => + connection.run("CREATE TABLE records(id INTEGER PRIMARY KEY, name TEXT)"), + ) + const result = await database.connection((connection) => + connection.run("INSERT INTO records(name) VALUES (?)", ["first"]), + ) + expect(result.changes).toBe(1) + expect(result.lastInsertId).toBe("1") + }) + + it("reads single rows and row sets with bound parameters", async () => { + database = await sqliteWasm({ path: ":memory:" }) + await database.connection(async (connection) => { + await connection.run("CREATE TABLE records(id INTEGER PRIMARY KEY, name TEXT)") + await connection.run("INSERT INTO records(name) VALUES (?), (?)", ["first", "second"]) + }) + const row = await database.connection((connection) => + connection.get<{ name: string }>("SELECT name FROM records WHERE id = ?", [2]), + ) + expect(row).toEqual({ name: "second" }) + const missing = await database.connection((connection) => + connection.get<{ name: string }>("SELECT name FROM records WHERE id = ?", [99]), + ) + expect(missing).toBeUndefined() + const rows = await database.connection((connection) => + connection.all<{ name: string }>("SELECT name FROM records ORDER BY id"), + ) + expect(rows).toEqual([{ name: "first" }, { name: "second" }]) + }) + + it("normalizes boolean and undefined parameters", async () => { + database = await sqliteWasm({ path: ":memory:" }) + await database.connection(async (connection) => { + await connection.run("CREATE TABLE flags(id INTEGER PRIMARY KEY, active INTEGER, note TEXT)") + await connection.run("INSERT INTO flags(active, note) VALUES (?, ?)", [true, undefined]) + }) + const row = await database.connection((connection) => + connection.get<{ active: number; note: null }>("SELECT active, note FROM flags"), + ) + expect(row).toEqual({ active: 1, note: null }) + }) + + it("rejects unsupported parameters", async () => { + database = await sqliteWasm({ path: ":memory:" }) + await expect( + database.connection((connection) => connection.run("SELECT ?", [Symbol("nope") as never])), + ).rejects.toThrow("unsupported SQLite parameter") + }) + + it("reports wall-clock time from the database", async () => { + database = await sqliteWasm({ path: ":memory:" }) + const now = await database.connection((connection) => connection.nowMilliseconds()) + expect(Math.abs(now - Date.now())).toBeLessThan(5_000) + }) + + it("commits a transaction and reports it active inside the callback", async () => { + database = await sqliteWasm({ path: ":memory:" }) + await database.connection((connection) => + connection.run("CREATE TABLE records(id INTEGER PRIMARY KEY)"), + ) + expect(database.transactionActive()).toBe(false) + await database.transaction(async (connection) => { + expect(database?.transactionActive()).toBe(true) + await connection.run("INSERT INTO records(id) VALUES (1)") + }) + expect(database.transactionActive()).toBe(false) + const row = await database.connection((connection) => + connection.get<{ count: number | bigint }>("SELECT COUNT(*) AS count FROM records"), + ) + expect(Number(row?.count)).toBe(1) + }) + + it("rolls a transaction back when the callback throws", async () => { + database = await sqliteWasm({ path: ":memory:" }) + await database.connection((connection) => + connection.run("CREATE TABLE records(id INTEGER PRIMARY KEY)"), + ) + await expect( + database.transaction(async (connection) => { + await connection.run("INSERT INTO records(id) VALUES (1)") + throw new Error("boom") + }), + ).rejects.toThrow("boom") + const row = await database.connection((connection) => + connection.get<{ count: number | bigint }>("SELECT COUNT(*) AS count FROM records"), + ) + expect(Number(row?.count)).toBe(0) + }) + + it("serializes concurrent access", async () => { + database = await sqliteWasm({ path: ":memory:" }) + const order: string[] = [] + await Promise.all([ + database.connection(async () => { + order.push("first:start") + await new Promise((resolve) => setTimeout(resolve, 10)) + order.push("first:end") + }), + database.connection(async () => { + order.push("second:start") + order.push("second:end") + }), + ]) + expect(order).toEqual(["first:start", "first:end", "second:start", "second:end"]) + }) + + it("expires queued work when the database deadline passes first", async () => { + database = await sqliteWasm({ path: ":memory:" }) + let release: (() => void) | undefined + const blocked = database.connection( + () => + new Promise((resolve) => { + release = resolve + }), + ) + await new Promise((resolve) => setTimeout(resolve, 1)) + let queuedRan = false + await expect( + withDatabaseDeadline({ timeoutMilliseconds: 1 }, () => + database!.connection(async () => { + queuedRan = true + }), + ), + ).rejects.toBeInstanceOf(DatabaseDeadlineExceeded) + release?.() + await blocked + expect(queuedRan).toBe(false) + }) + + it("rolls back a transaction that finishes after its deadline", async () => { + database = await sqliteWasm({ path: ":memory:" }) + await database.connection((connection) => + connection.run("CREATE TABLE records(id INTEGER PRIMARY KEY)"), + ) + await expect( + withDatabaseDeadline({ timeoutMilliseconds: 10 }, () => + database!.transaction(async (connection) => { + await connection.run("INSERT INTO records(id) VALUES (1)") + await new Promise((resolve) => setTimeout(resolve, 30)) + }), + ), + ).rejects.toBeInstanceOf(DatabaseDeadlineExceeded) + const row = await database.connection((connection) => + connection.get<{ count: number | bigint }>("SELECT COUNT(*) AS count FROM records"), + ) + expect(Number(row?.count)).toBe(0) + }) + + it("closes idempotently and rejects work afterwards", async () => { + database = await sqliteWasm({ path: ":memory:" }) + await database.close() + await database.close() + await expect(database.connection((connection) => connection.run("SELECT 1"))).rejects.toThrow() + }) + + it("runs the full runtime against WASM storage", async () => { + database = await sqliteWasm({ path: ":memory:" }) + runtime = configure({ + database, + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + pollingIntervalMilliseconds: 1, + syncPollingIntervalMilliseconds: 1, + maxAttempts: 2, + }) + await runtime.install() + const counter = WasmCounter.ref("wasm-primary") + + expect(await counter.increment({ amount: 3 })).toBe(3) + expect(await counter.increment()).toBe(4) + expect(await counter.count).toBe(4) + expect(await counter.snapshot()).toEqual({ count: 4 }) + }) +}) diff --git a/test/tab-host.test.ts b/test/tab-host.test.ts new file mode 100644 index 0000000..310e3a4 --- /dev/null +++ b/test/tab-host.test.ts @@ -0,0 +1,141 @@ +import { afterEach, describe, expect, it } from "vitest" +import "../src/platform/node.js" +import { Actor } from "../src/actor.js" +import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" +import { sqlite } from "../src/database/sqlite.js" +import { + connectTabClient, + startTabHost, + type TabClient, + type TabHost, +} from "../src/browser/tab-host.js" + +class TabCounter extends Actor { + static override readonly actorType = "TabCounter" + + count = 0 + + increment({ amount = 1 }: { amount?: number } = {}): number { + this.count += amount + return this.count + } +} + +const hosts: TabHost[] = [] +const clients: TabClient[] = [] + +afterEach(async () => { + for (const client of clients) client.close() + await Promise.all(hosts.map((host) => host.close())) + hosts.length = 0 + clients.length = 0 +}) + +function trackedHost(name: string): TabHost { + const host = startTabHost({ + name, + startRuntime: async () => { + const runtime: SolidObjectsRuntime = createRuntime({ + database: sqlite({ path: ":memory:" }), + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + pollingIntervalMilliseconds: 1, + syncPollingIntervalMilliseconds: 1, + }) + runtime.register(TabCounter) + await runtime.install() + return { runtime } + }, + }) + hosts.push(host) + return host +} + +function trackedClient(name: string): TabClient { + const client = connectTabClient({ + name, + retryIntervalMilliseconds: 50, + timeoutMilliseconds: 10_000, + }) + clients.push(client) + return client +} + +const describeTabHost = globalThis.navigator?.locks ? describe : describe.skip + +describeTabHost("tab host", () => { + it("elects one leader and serves invocations from any participant", async () => { + const name = `election-${crypto.randomUUID()}` + const first = trackedHost(name) + await first.leadership() + const second = trackedHost(name) + const client = trackedClient(name) + + const initial = await client.invoke({ + actorType: "TabCounter", + actorId: "shared", + operation: "increment", + arguments: { amount: 2 }, + }) + const next = await client.invoke({ + actorType: "TabCounter", + actorId: "shared", + operation: "increment", + }) + + expect(initial).toBe(2) + expect(next).toBe(3) + expect(first.role()).toBe("leader") + expect(second.role()).toBe("follower") + }) + + it("fails over to the next host when the leader closes", async () => { + const name = `failover-${crypto.randomUUID()}` + const first = trackedHost(name) + await first.leadership() + const second = trackedHost(name) + const client = trackedClient(name) + + expect( + await client.invoke({ actorType: "TabCounter", actorId: "solo", operation: "increment" }), + ).toBe(1) + + const closing = first.close() + const served = client.invoke({ + actorType: "TabCounter", + actorId: "solo", + operation: "increment", + }) + await closing + await second.leadership() + + expect(await served).toBe(1) + expect(second.role()).toBe("leader") + }) + + it("reports an invocation error from the leader", async () => { + const name = `errors-${crypto.randomUUID()}` + const host = trackedHost(name) + await host.leadership() + const client = trackedClient(name) + + await expect( + client.invoke({ actorType: "TabCounter", actorId: "solo", operation: "missing" }), + ).rejects.toThrow(/missing/) + }) + + it("times out when no host exists", async () => { + const name = `nobody-${crypto.randomUUID()}` + const client = connectTabClient({ + name, + retryIntervalMilliseconds: 20, + timeoutMilliseconds: 100, + }) + clients.push(client) + + await expect( + client.invoke({ actorType: "TabCounter", actorId: "solo", operation: "increment" }), + ).rejects.toThrow(/no tab host answered/) + }) +}) diff --git a/test/transmit-fixtures.test.ts b/test/transmit-fixtures.test.ts new file mode 100644 index 0000000..6b0639f --- /dev/null +++ b/test/transmit-fixtures.test.ts @@ -0,0 +1,129 @@ +import { readFileSync } from "node:fs" +import { afterEach, describe, expect, it } from "vitest" +import "../src/platform/node.js" +import { Actor } from "../src/actor.js" +import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" +import { sqlite } from "../src/database/sqlite.js" +import { + receiveTransmitEnvelope, + registerTransmit, + type TransmitEnvelope, +} from "../src/transmit.js" +import { InvalidPayload } from "../src/errors.js" + +interface FixtureFile { + valid: Array<{ name: string; envelope: TransmitEnvelope; idempotencyKey: string }> + duplicatePair: TransmitEnvelope[] + malformed: Array<{ name: string; envelope: TransmitEnvelope }> +} + +const fixtures: FixtureFile = JSON.parse( + readFileSync(new URL("../compatibility/transmit-envelopes.json", import.meta.url), "utf8"), +) + +class TransmitCounter extends Actor { + static override readonly actorType = "transmit-counters" + + value = 0 + + increment({ amount = 1 }: { amount?: number } = {}): number { + this.value += amount + this.transmit().increment!({ amount }) + return this.value + } +} + +const runtimes: SolidObjectsRuntime[] = [] + +afterEach(async () => { + await Promise.all(runtimes.map((runtime) => runtime.close())) + runtimes.length = 0 +}) + +async function fixtureRuntime(): Promise { + const runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + pollingIntervalMilliseconds: 1, + syncPollingIntervalMilliseconds: 1, + maxAttempts: 8, + retryDelayMilliseconds: () => 0, + }) + runtimes.push(runtime) + await runtime.install() + runtime.register(TransmitCounter) + return runtime +} + +describe("shared transmit envelope fixtures", () => { + it("accepts every valid fixture envelope exactly once", async () => { + const runtime = await fixtureRuntime() + + for (const fixture of fixtures.valid) { + const first = await receiveTransmitEnvelope({ runtime, envelope: fixture.envelope }) + const replay = await receiveTransmitEnvelope({ runtime, envelope: fixture.envelope }) + expect(replay.messageId, fixture.name).toBe(first.messageId) + const stored = await runtime.settings.database.connection((connection) => + connection.get<{ idempotency_key: string }>( + `SELECT idempotency_key FROM ${runtime.repository.table("messages")} WHERE id = ?`, + [first.messageId], + ), + ) + expect(stored?.idempotency_key, fixture.name).toBe(fixture.idempotencyKey) + } + await runtime.testing.drain({ roles: ["actors"] }) + + expect(await runtime.ref(TransmitCounter, "fixture-counter").snapshot()).toEqual({ + value: 3, + }) + }) + + it("applies the duplicate fixture pair once", async () => { + const runtime = await fixtureRuntime() + + const results = [] + for (const envelope of fixtures.duplicatePair) { + results.push(await receiveTransmitEnvelope({ runtime, envelope })) + } + await runtime.testing.drain({ roles: ["actors"] }) + + expect(new Set(results.map((result) => result.messageId)).size).toBe(1) + expect(await runtime.ref(TransmitCounter, "fixture-counter").snapshot()).toEqual({ + value: 1, + }) + }) + + it("rejects every malformed fixture envelope", async () => { + const runtime = await fixtureRuntime() + + for (const fixture of fixtures.malformed) { + await expect( + receiveTransmitEnvelope({ runtime, envelope: fixture.envelope }), + fixture.name, + ).rejects.toThrow(InvalidPayload) + } + }) + + it("stages an envelope that matches the fixture byte for byte", async () => { + const runtime = await fixtureRuntime() + const delivered: TransmitEnvelope[] = [] + registerTransmit({ + runtime, + deliver: async (envelope) => { + delivered.push(envelope) + }, + }) + + await runtime.ref(TransmitCounter, "fixture-counter").increment({ amount: 2 }) + await runtime.testing.drain({ roles: ["actors", "effects"] }) + + const fixture = fixtures.valid[0]! + const { effectId, ...staged } = delivered[0]! + const { effectId: fixtureEffectId, ...expected } = fixture.envelope + expect(staged).toEqual(expected) + expect(fixture.idempotencyKey).toBe(`transmit:${fixtureEffectId}`) + expect(effectId).not.toHaveLength(0) + }) +}) diff --git a/test/transmit.test.ts b/test/transmit.test.ts new file mode 100644 index 0000000..abdd8eb --- /dev/null +++ b/test/transmit.test.ts @@ -0,0 +1,324 @@ +import { afterEach, describe, expect, it } from "vitest" +import "../src/platform/node.js" +import { Actor } from "../src/actor.js" +import { IdempotencyConflict } from "../src/errors.js" +import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" +import type { SolidObjectsConfiguration } from "../src/configuration.js" +import { sqlite } from "../src/database/sqlite.js" +import { + receiveTransmitEnvelope, + registerTransmit, + TRANSMIT_EFFECT, + type TransmitEnvelope, +} from "../src/transmit.js" + +class TransmitCounter extends Actor { + static override readonly actorType = "TransmitCounter" + + count = 0 + + increment({ amount = 1 }: { amount?: number } = {}): number { + this.count += amount + this.transmit().increment!({ amount }) + return this.count + } +} + +class EmitCounter extends Actor { + static override readonly actorType = "TransmitCounter" + + count = 0 + + increment({ amount = 1 }: { amount?: number } = {}): number { + this.count += amount + this.emit(TRANSMIT_EFFECT, { + arguments: { operation: "increment", arguments: { amount } }, + }) + return this.count + } +} + +class ServerTransmitCounter extends Actor { + static override readonly actorType = "TransmitCounter" + + count = 0 + applied: number[] = [] + + increment({ amount = 1 }: { amount?: number } = {}): number { + this.count += amount + this.applied = [...this.applied, amount] + return this.count + } +} + +class AuditLog extends Actor { + static override readonly actorType = "AuditLog" + + events: string[] = [] + + record({ eventName }: { eventName: string }): number { + this.events = [...this.events, eventName] + return this.events.length + } +} + +class Reporter extends Actor { + static override readonly actorType = "Reporter" + + report({ eventName }: { eventName: string }): void { + this.emit(TRANSMIT_EFFECT, { + arguments: { + actorType: "AuditLog", + actorId: "audit-primary", + operation: "record", + arguments: { eventName }, + }, + }) + } +} + +const runtimes: SolidObjectsRuntime[] = [] + +afterEach(async () => { + await Promise.all(runtimes.map((runtime) => runtime.close())) + runtimes.length = 0 +}) + +function testRuntime(overrides: Partial = {}): SolidObjectsRuntime { + const runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + pollingIntervalMilliseconds: 1, + syncPollingIntervalMilliseconds: 1, + maxAttempts: 8, + retryDelayMilliseconds: () => 0, + ...overrides, + }) + runtimes.push(runtime) + return runtime +} + +async function pairedRuntimes(options: { + deliver: (envelope: TransmitEnvelope) => Promise +}): Promise<{ local: SolidObjectsRuntime; server: SolidObjectsRuntime }> { + const local = testRuntime() + const server = testRuntime() + registerTransmit({ runtime: local, deliver: options.deliver }) + await local.install() + await server.install() + return { local, server } +} + +describe("sync bridge", () => { + it("delivers staged transmit effects to the server actor", async () => { + const delivered: TransmitEnvelope[] = [] + const { local, server } = await pairedRuntimes({ + deliver: async (envelope) => { + delivered.push(envelope) + await receiveTransmitEnvelope({ runtime: server, envelope }) + }, + }) + server.register(ServerTransmitCounter) + + await local.ref(TransmitCounter, "counter-1").increment({ amount: 2 }) + await local.testing.drain({ roles: ["actors", "effects"] }) + await server.testing.drain({ roles: ["actors"] }) + + expect(delivered).toHaveLength(1) + expect(delivered[0]).toMatchObject({ + actorType: "TransmitCounter", + actorId: "counter-1", + operation: "increment", + arguments: { amount: 2 }, + }) + expect(await server.ref(ServerTransmitCounter, "counter-1").snapshot()).toEqual({ + count: 2, + applied: [2], + }) + }) + + it("deduplicates a replayed envelope on the server", async () => { + let captured: TransmitEnvelope | undefined + const { local, server } = await pairedRuntimes({ + deliver: async (envelope) => { + captured = envelope + await receiveTransmitEnvelope({ runtime: server, envelope }) + }, + }) + server.register(ServerTransmitCounter) + + await local.ref(TransmitCounter, "counter-1").increment({ amount: 3 }) + await local.testing.drain({ roles: ["actors", "effects"] }) + if (!captured) throw new Error("deliver never ran") + await receiveTransmitEnvelope({ runtime: server, envelope: captured }) + await receiveTransmitEnvelope({ runtime: server, envelope: captured }) + await server.testing.drain({ roles: ["actors"] }) + + expect(await server.ref(ServerTransmitCounter, "counter-1").snapshot()).toEqual({ + count: 3, + applied: [3], + }) + }) + + it("keeps per-actor order when an early envelope fails first", async () => { + let failuresRemaining = 1 + const { local, server } = await pairedRuntimes({ + deliver: async (envelope) => { + const amount = Number((envelope.arguments as { amount?: number }).amount) + if (amount === 1 && failuresRemaining > 0) { + failuresRemaining -= 1 + throw new Error("network down") + } + await receiveTransmitEnvelope({ runtime: server, envelope }) + }, + }) + server.register(ServerTransmitCounter) + + const counter = local.ref(TransmitCounter, "ordered") + await counter.increment({ amount: 1 }) + await counter.increment({ amount: 2 }) + await local.testing.drain({ roles: ["actors", "effects"], maxPasses: 20 }) + await server.testing.drain({ roles: ["actors"] }) + + expect(await server.ref(ServerTransmitCounter, "ordered").snapshot()).toEqual({ + count: 3, + applied: [1, 2], + }) + }) + + it("recovers in order after an offline period", async () => { + let online = false + const { local, server } = await pairedRuntimes({ + deliver: async (envelope) => { + if (!online) throw new Error("offline") + await receiveTransmitEnvelope({ runtime: server, envelope }) + }, + }) + server.register(ServerTransmitCounter) + + const counter = local.ref(TransmitCounter, "offline") + await counter.increment({ amount: 1 }) + await counter.increment({ amount: 2 }) + await counter.increment({ amount: 3 }) + await local.testing.drain({ roles: ["actors", "effects"], maxPasses: 5 }) + await server.testing.drain({ roles: ["actors"] }) + expect(await server.ref(ServerTransmitCounter, "offline").snapshot()).toEqual({ + count: 0, + applied: [], + }) + + online = true + await local.testing.drain({ roles: ["actors", "effects"], maxPasses: 30 }) + await server.testing.drain({ roles: ["actors"] }) + + expect(await server.ref(ServerTransmitCounter, "offline").snapshot()).toEqual({ + count: 6, + applied: [1, 2, 3], + }) + }) + + it("delivers a raw emit the same way as transmit", async () => { + const { local, server } = await pairedRuntimes({ + deliver: async (envelope) => { + await receiveTransmitEnvelope({ runtime: server, envelope }) + }, + }) + server.register(ServerTransmitCounter) + + await local.ref(EmitCounter, "emitted").increment({ amount: 4 }) + await local.testing.drain({ roles: ["actors", "effects"] }) + await server.testing.drain({ roles: ["actors"] }) + + expect(await server.ref(ServerTransmitCounter, "emitted").snapshot()).toEqual({ + count: 4, + applied: [4], + }) + }) + + it("routes an explicit target to a different server actor", async () => { + const { local, server } = await pairedRuntimes({ + deliver: async (envelope) => { + await receiveTransmitEnvelope({ runtime: server, envelope }) + }, + }) + server.register(AuditLog) + + await local.ref(Reporter, "reporter-1").report({ eventName: "signed_in" }) + await local.testing.drain({ roles: ["actors", "effects"] }) + await server.testing.drain({ roles: ["actors"] }) + + expect(await server.ref(AuditLog, "audit-primary").snapshot()).toEqual({ + events: ["signed_in"], + }) + }) + + it("accepts an envelope without arguments, matching the Ruby ingest", async () => { + const server = testRuntime() + await server.install() + server.register(ServerTransmitCounter) + + await receiveTransmitEnvelope({ + runtime: server, + envelope: { + effectId: "effect-no-arguments", + actorType: "TransmitCounter", + actorId: "defaulted", + operation: "increment", + } as never, + }) + await server.testing.drain({ roles: ["actors"] }) + + expect(await server.ref(ServerTransmitCounter, "defaulted").snapshot()).toEqual({ + count: 1, + applied: [1], + }) + }) + + it("raises IdempotencyConflict when a replay changes the arguments", async () => { + const server = testRuntime() + await server.install() + server.register(ServerTransmitCounter) + const envelope = { + effectId: "effect-conflict", + actorType: "TransmitCounter", + actorId: "conflicted", + operation: "increment", + arguments: { amount: 1 }, + } + + await receiveTransmitEnvelope({ runtime: server, envelope }) + await expect( + receiveTransmitEnvelope({ + runtime: server, + envelope: { ...envelope, arguments: { amount: 2 } }, + }), + ).rejects.toBeInstanceOf(IdempotencyConflict) + await server.testing.drain({ roles: ["actors"] }) + + expect(await server.ref(ServerTransmitCounter, "conflicted").snapshot()).toEqual({ + count: 1, + applied: [1], + }) + }) + + it("rejects a malformed envelope on the server", async () => { + const server = testRuntime() + await server.install() + server.register(TransmitCounter) + + await expect( + receiveTransmitEnvelope({ + runtime: server, + envelope: { + effectId: "effect-1", + actorType: "TransmitCounter", + actorId: "counter-1", + operation: "", + arguments: {}, + }, + }), + ).rejects.toThrow() + }) +})