diff --git a/apps/server/src/orchestration-v2/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration-v2/Layers/ProjectionSnapshotQuery.ts index 8772f12b0509..044f25e2fce0 100644 --- a/apps/server/src/orchestration-v2/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration-v2/Layers/ProjectionSnapshotQuery.ts @@ -12,6 +12,7 @@ import { OrchestrationV2PlanArtifact, OrchestrationV2TurnItemJson, } from "@t3tools/contracts"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; @@ -53,7 +54,12 @@ const make = Effect.gen(function* () { const shell = (location: "active" | "archive" = "active", unsettledOnly = false) => wrap( Effect.gen(function* () { - const snapshot = yield* projections.getShellSnapshot({ location, unsettledOnly }); + // The full shell read takes seconds on a large database: run it on the + // SQLite reader worker, off the event loop and the writer. Every caller + // reads committed state; inside an open transaction it stays on the writer. + const snapshot = yield* projections + .getShellSnapshot({ location, unsettledOnly }) + .pipe(NodeSqliteClient.readOnly); // The V2 store returns archived rows in `archivedThreads`, never in `threads`. const threads = location === "archive" ? snapshot.archivedThreads : snapshot.threads; const names = new Map( diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index 3fb5d2794748..1506f04a9f12 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -2308,6 +2308,55 @@ it.effect("cancels tracked setup before provider work is released", () => }), ); +it.effect("stops an in-flight branch rename when tracked setup is cancelled", () => + Effect.gen(function* () { + const setupEntered = yield* Deferred.make(); + const branchNameStarted = yield* Deferred.make(); + const allowBranchName = yield* Deferred.make(); + const branchNameInterrupted = yield* Deferred.make(); + const renameAttempted = yield* Deferred.make(); + const harness = makeHarness({ + renameBranch: (input) => + Deferred.succeed(renameAttempted, undefined).pipe(Effect.as({ branch: input.newBranch })), + runSetup: () => Deferred.succeed(setupEntered, undefined).pipe(Effect.andThen(Effect.never)), + generateBranchName: () => + Deferred.succeed(branchNameStarted, undefined).pipe( + Effect.andThen(Deferred.await(allowBranchName)), + Effect.as({ branch: "generated-branch" }), + Effect.onInterrupt(() => Deferred.succeed(branchNameInterrupted, undefined)), + ), + }); + yield* Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const tracker = yield* WorktreeSetupTracker.WorktreeSetupTracker; + const threads = yield* ThreadManagement.ThreadManagementService; + const launched = yield* launches.launch( + launchInput({ + command: "launch:cancel-renaming", + thread: "thread:cancel-renaming", + message: "Start", + workspace: { type: "worktree", baseRef: "main" }, + }), + ); + yield* Deferred.await(setupEntered); + yield* Deferred.await(branchNameStarted); + assert.isTrue(yield* tracker.cancel(launched.threadId)); + // Let a rename that survived the cancel finish, then check it never ran. + yield* Deferred.succeed(allowBranchName, undefined); + yield* Effect.raceFirst( + Deferred.await(branchNameInterrupted), + Deferred.await(renameAttempted), + ); + assert.isTrue( + yield* Deferred.isDone(branchNameInterrupted), + "the cancelled setup's branch rename kept running", + ); + assert.equal(harness.renameBranch.mock.calls.length, 0); + assert.isNull((yield* threads.getThreadProjection(launched.threadId)).thread.worktreePath); + }).pipe(Effect.provide(harness.layer)); + }), +); + it.effect.each([0, 1])("releases an async setup before its completion with exit %s", (exitCode) => Effect.gen(function* () { const completion = yield* Deferred.make<{ exitCode: number | null; durationMs: number }>(); diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index 97e7b574ad9c..9ac2d5159926 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -23,6 +23,7 @@ import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; @@ -259,6 +260,9 @@ const make = Effect.gen(function* () { let createdWorktreePath: string | null = null; let setupTerminalId: string | null = null; let workspaceRecorded = false; + // The background branch rename records the worktree again when it lands, + // so cleanup stops it before forgetting a removed worktree. + let branchRename: Fiber.Fiber | null = null; if (input.workspaceStrategy.type === "worktree") { yield* setupTracker.begin({ threadId, @@ -446,7 +450,7 @@ const make = Effect.gen(function* () { ) { const oldBranch = branch; const worktreeCwd = worktreePath; - yield* generateBranchNameFor(worktreeCwd, initialMessage).pipe( + branchRename = yield* generateBranchNameFor(worktreeCwd, initialMessage).pipe( Effect.flatMap(({ branch: newBranch, exactName }) => git.renameBranch({ cwd: worktreeCwd, @@ -586,6 +590,7 @@ const make = Effect.gen(function* () { yield* terminals .close({ threadId, terminalId: setupTerminalId, deleteHistory: true }) .pipe(Effect.ignore); + if (branchRename !== null) yield* Fiber.interrupt(branchRename); const removedPath = createdWorktreePath; // The thread forgets the worktree only once it is gone; a failed // removal leaves the directory for the user to clean up rather than diff --git a/apps/server/src/orchestration-v2/ThreadPullRequestService.ts b/apps/server/src/orchestration-v2/ThreadPullRequestService.ts index a1dabb267b2f..a7c38e244c11 100644 --- a/apps/server/src/orchestration-v2/ThreadPullRequestService.ts +++ b/apps/server/src/orchestration-v2/ThreadPullRequestService.ts @@ -11,6 +11,7 @@ import { type ThreadLinkedPullRequest, } from "@t3tools/contracts"; import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import * as Cause from "effect/Cause"; // T3-CUSTOM(expbkt3): the idle-thread rotation reads the clock. import * as Clock from "effect/Clock"; @@ -133,10 +134,12 @@ export const make = Effect.gen(function* () { */ const readThreadSnapshot = ({ threadId, backfill }: RefreshRequest) => threadId === null - ? orchestrator.getShellSnapshot({ - location: "active", - unsettledOnly: !(backfill || pendingBackfill.size > 0), - }) + ? orchestrator + .getShellSnapshot({ + location: "active", + unsettledOnly: !(backfill || pendingBackfill.size > 0), + }) + .pipe(NodeSqliteClient.readOnly) : Effect.gen(function* () { // Read the sequence first. The thread is then at least this new, so a // sync guarded by the sequence is rejected rather than missing a change. diff --git a/apps/server/src/orchestration-v2/http.ts b/apps/server/src/orchestration-v2/http.ts index 478947320fc1..b50ddb17d375 100644 --- a/apps/server/src/orchestration-v2/http.ts +++ b/apps/server/src/orchestration-v2/http.ts @@ -9,6 +9,8 @@ import { TurnItemId, type OrchestrationProjectShell, } from "@t3tools/contracts"; +// T3-CUSTOM(expbkt3): shell snapshot reads run on the SQLite reader worker. +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import * as Effect from "effect/Effect"; // T3-CUSTOM(expbkt3): resolve the bound operator. import * as Option from "effect/Option"; @@ -145,16 +147,20 @@ export const layer = HttpApiBuilder.group( ); const loadShellSnapshot = Effect.fn("http.orchestration.loadShellSnapshot")(function* () { - const base = yield* sql.withTransaction( - Effect.gen(function* () { - const threads = yield* threadManagement.getShellSnapshot({ location: "active" }); - return buildActiveShellSnapshot({ - projects: yield* projectStore.listShells(), - threads, - snapshotSequence: yield* applicationEvents.latestApplicationSequence, - }); - }), - ); + // T3-CUSTOM(expbkt3): read on the SQLite reader worker, off the event loop and the + // writer (upstream does this in loadShellSnapshotParts, which stage does not have). + const base = yield* sql + .withTransaction( + Effect.gen(function* () { + const threads = yield* threadManagement.getShellSnapshot({ location: "active" }); + return buildActiveShellSnapshot({ + projects: yield* projectStore.listShells(), + threads, + snapshotSequence: yield* applicationEvents.latestApplicationSequence, + }); + }), + ) + .pipe(NodeSqliteClient.readOnly); const projects = yield* enrichProjectShells(base.projects); // T3-CUSTOM(expbkt3): HTTP shell visibility matches the native socket stream. return filterNativeShell({ ...base, projects }, yield* currentActorUserId); diff --git a/apps/server/src/persistence/Sqlite.ts b/apps/server/src/persistence/Sqlite.ts index 0f248b5a2387..142ff045946e 100644 --- a/apps/server/src/persistence/Sqlite.ts +++ b/apps/server/src/persistence/Sqlite.ts @@ -35,6 +35,7 @@ export const layerFromPath = Effect.fn("makeSqlitePersistenceLive")(function* (d layerSetup, NodeSqliteClient.layer({ filename: dbPath, + readerWorker: true, spanAttributes: { "db.name": path.basename(dbPath), "service.name": "t3code-server", diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index decc4ed912d7..0c424bc476d1 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -143,6 +143,8 @@ import { FORK_WS_RPCS, } from "@t3tools/contracts"; import { resolveServerBackgroundActivitySettings } from "@t3tools/shared/backgroundActivitySettings"; +// T3-CUSTOM(expbkt3): shell reads run on the SQLite reader worker. +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import { HttpRouter, HttpServerRequest, @@ -1015,16 +1017,20 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc }, ); const loadSnapshot = Effect.fn("ws.orchestrationV2.loadShellSnapshot")(function* () { - const base = yield* sql.withTransaction( - Effect.gen(function* () { - const threads = yield* threadManagement.getShellSnapshot({ location: "active" }); - return buildActiveShellSnapshot({ - projects: yield* projects.listShells(), - threads, - snapshotSequence: yield* applicationEvents.latestApplicationSequence, - }); - }), - ); + // T3-CUSTOM(expbkt3): read on the SQLite reader worker, off the event loop and the + // writer (upstream does this in loadShellSnapshotParts, which stage does not have). + const base = yield* sql + .withTransaction( + Effect.gen(function* () { + const threads = yield* threadManagement.getShellSnapshot({ location: "active" }); + return buildActiveShellSnapshot({ + projects: yield* projects.listShells(), + threads, + snapshotSequence: yield* applicationEvents.latestApplicationSequence, + }); + }), + ) + .pipe(NodeSqliteClient.readOnly); const enriched = yield* enrichProjectShells(base.projects); return { snapshot: { ...base, projects: enriched.projects } as OrchestrationV2ShellSnapshot, @@ -1923,6 +1929,8 @@ const layerWsRpc = ( }), ) .pipe( + // T3-CUSTOM(expbkt3): read on the SQLite reader worker; enrichment runs after it. + NodeSqliteClient.readOnly, Effect.flatMap((snapshot) => enrichProjectShells(snapshot.projects).pipe( Effect.map(({ projects }) => ({ ...snapshot, projects })), diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index 83fe851ac8ad..040baa7ba130 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -1,14 +1,67 @@ import * as NodeSqlite from "node:sqlite"; +import type * as NodeWorkerThreads from "node:worker_threads"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; +import * as Logger from "effect/Logger"; import * as Path from "effect/Path"; import * as SqlClient from "effect/sql/SqlClient"; +import * as TestClock from "effect/testing/TestClock"; +import { vi } from "vite-plus/test"; import * as SqliteClient from "./nodeSqliteClient.ts"; +const startedWorkers = vi.hoisted((): Array => []); +// When on, workers run a script that never announces, like a worker stuck opening the file. +const silentWorkers = vi.hoisted(() => ({ on: false, readyOnTerminate: false })); + +vi.mock("node:worker_threads", async (importOriginal) => { + const actual = await importOriginal(); + class RecordedWorker extends actual.Worker { + constructor(...args: ConstructorParameters) { + super( + ...((silentWorkers.on + ? ["setInterval(() => {}, 1 << 30);", { eval: true }] + : args) as ConstructorParameters), + ); + startedWorkers.push(this); + } + override terminate() { + // A "ready" already queued when termination starts reaches the parent before "exit". + if (silentWorkers.readyOnTerminate) this.emit("message", { type: "ready" }); + return super.terminate(); + } + } + return { ...actual, Worker: RecordedWorker }; +}); + +const withReaderClient = ( + use: (sql: SqlClient.SqlClient) => Effect.Effect, +) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-sqlite-reader-" }); + const filename = path.join(directory, "state.sqlite"); + return yield* Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`PRAGMA journal_mode = WAL`; + yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`; + yield* sql`INSERT INTO entries VALUES ('first')`; + return yield* use(sql); + }).pipe(Effect.provide(SqliteClient.layer({ filename, readerWorker: true }))); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)); + +const countEntries = (sql: SqlClient.SqlClient) => + Effect.map( + sql<{ count: number }>`SELECT COUNT(*) AS count FROM entries`, + (rows) => rows[0]?.count, + ); + const layer = it.layer(SqliteClient.layer({ filename: ":memory:" })); layer("NodeSqliteClient", (it) => { @@ -182,3 +235,318 @@ it.effect( }).pipe(Effect.provide(SqliteClient.layer({ filename }))); }).pipe(Effect.provide(NodeServices.layer)), ); + +it.effect("lets the writer commit while a read-only transaction is open", () => + withReaderClient((sql) => + Effect.gen(function* () { + const snapshotTaken = yield* Deferred.make(); + const writer = yield* Effect.forkChild( + Deferred.await(snapshotTaken).pipe( + Effect.andThen(sql`INSERT INTO entries VALUES ('second')`), + ), + ); + const counts = yield* sql + .withTransaction( + Effect.gen(function* () { + const before = yield* countEntries(sql); + yield* Deferred.succeed(snapshotTaken, undefined); + // Holding the writer lock here would deadlock this join. + yield* Fiber.join(writer); + const after = yield* countEntries(sql); + return { before, after }; + }), + ) + .pipe(SqliteClient.readOnly); + + // The reader keeps the snapshot it began with, and later reads see the commit. + assert.deepEqual(counts, { before: 1, after: 1 }); + assert.equal(yield* countEntries(sql).pipe(SqliteClient.readOnly), 2); + }), + ), +); + +it.effect("rejects a write in a read-only transaction", () => + withReaderClient((sql) => + Effect.gen(function* () { + const error = yield* sql + .withTransaction(sql`INSERT INTO entries VALUES ('blocked')`) + .pipe(SqliteClient.readOnly, Effect.flip); + + assert.equal(error._tag, "SqlError"); + assert.include(String(error.reason.cause), "readonly database"); + assert.equal(yield* countEntries(sql), 1); + }), + ), +); + +it.effect("keeps the reader read-only even if a statement turns query_only off", () => + withReaderClient((sql) => + Effect.gen(function* () { + const error = yield* sql + .withTransaction( + Effect.gen(function* () { + yield* sql`PRAGMA query_only = OFF`; + yield* sql`INSERT INTO entries VALUES ('blocked')`; + }), + ) + .pipe(SqliteClient.readOnly, Effect.flip); + + assert.equal(error._tag, "SqlError"); + assert.include(String(error.reason.cause), "readonly database"); + assert.equal(yield* countEntries(sql), 1); + }), + ), +); + +it.effect("keeps read-only work inside a write transaction on the writer", () => + withReaderClient((sql) => + sql.withTransaction( + Effect.gen(function* () { + yield* sql`INSERT INTO entries VALUES ('uncommitted')`; + // Only the writer can see its own uncommitted row. + const count = yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly); + assert.equal(count, 2); + }), + ), + ), +); + +it.effect("fails a read-only transaction whose reader stops, then starts a new reader", () => + withReaderClient((sql) => + Effect.gen(function* () { + const startedBefore = startedWorkers.length; + const error = yield* sql + .withTransaction( + Effect.gen(function* () { + yield* countEntries(sql); + yield* Effect.promise(() => startedWorkers.at(-1)!.terminate()); + return yield* countEntries(sql); + }), + ) + .pipe(SqliteClient.readOnly, Effect.flip); + + assert.equal(error._tag, "SqlError"); + assert.include(String(error.reason.cause), "Database reader exited"); + assert.equal(yield* countEntries(sql).pipe(SqliteClient.readOnly), 1); + assert.equal(startedWorkers.length, startedBefore + 2); + }), + ), +); + +it.effect("uses the writer for read-only work on an in-memory database", () => + Effect.gen(function* () { + const startedBefore = startedWorkers.length; + const sql = yield* SqlClient.SqlClient; + yield* sql`CREATE TABLE memory_entries(name TEXT NOT NULL)`; + yield* sql`INSERT INTO memory_entries VALUES ('kept')`; + + const rows = yield* sql + .withTransaction(sql<{ name: string }>`SELECT name FROM memory_entries`) + .pipe(SqliteClient.readOnly); + + assert.deepEqual(rows, [{ name: "kept" }]); + assert.equal(startedWorkers.length, startedBefore); + }).pipe(Effect.provide(SqliteClient.layer({ filename: ":memory:", readerWorker: true }))), +); + +it.effect("keeps a nested transaction inside a read-only transaction on the reader", () => + withReaderClient((sql) => + Effect.gen(function* () { + const snapshotTaken = yield* Deferred.make(); + const writer = yield* Effect.forkChild( + Deferred.await(snapshotTaken).pipe( + Effect.andThen(sql`INSERT INTO entries VALUES ('second')`), + ), + ); + // Callers wrap a store read that opens its own transaction, as the + // shell snapshot routes do around getShellSnapshot. + const counts = yield* sql + .withTransaction( + Effect.gen(function* () { + const outer = yield* countEntries(sql); + yield* Deferred.succeed(snapshotTaken, undefined); + yield* Fiber.join(writer); + const nested = yield* sql.withTransaction(countEntries(sql)); + return { outer, nested }; + }), + ) + .pipe(SqliteClient.readOnly); + + assert.deepEqual(counts, { outer: 1, nested: 1 }); + }), + ), +); + +const withFileClient = ( + setup: ( + sql: SqlClient.SqlClient, + filename: string, + ) => Effect.Effect, +) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-sqlite-reader-off-" }); + const filename = path.join(directory, "state.sqlite"); + return yield* Effect.gen(function* () { + return yield* setup(yield* SqlClient.SqlClient, filename); + }).pipe(Effect.provide(SqliteClient.layer({ filename, readerWorker: true }))); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)); + +it.effect("uses the writer when the reader cannot open the database", () => + withFileClient((sql, filename) => + Effect.gen(function* () { + yield* sql`PRAGMA journal_mode = WAL`; + yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`; + yield* sql`INSERT INTO entries VALUES ('kept')`; + // The writer keeps its open file; a read-only open of the path now fails. + yield* (yield* FileSystem.FileSystem).remove(filename); + const startedBefore = startedWorkers.length; + + assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1); + assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1); + assert.equal(startedWorkers.length, startedBefore + 1); + }), + ), +); + +it.effect("uses the writer for a database that is not in WAL mode", () => + withFileClient((sql) => + Effect.gen(function* () { + yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`; + const startedBefore = startedWorkers.length; + + // A write inside readOnly only succeeds on the writer: the reader is read-only. + yield* sql + .withTransaction(sql`INSERT INTO entries VALUES ('on the writer')`) + .pipe(SqliteClient.readOnly); + assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1); + assert.equal(startedWorkers.length, startedBefore + 1); + }), + ), +); + +const captureWarnings = () => { + const warnings: Array = []; + const logger = Logger.make(({ logLevel, message }) => { + if (logLevel === "Warn") warnings.push(message); + }); + return { warnings, layer: Logger.layer([logger], { mergeWithExisting: false }) }; +}; + +const silently = (effect: Effect.Effect) => + Effect.acquireUseRelease( + Effect.sync(() => { + silentWorkers.on = true; + }), + () => effect, + () => + Effect.sync(() => { + silentWorkers.on = false; + silentWorkers.readyOnTerminate = false; + }), + ); + +it.effect("turns the reader off when its worker never opens the database", () => { + const { warnings, layer: logs } = captureWarnings(); + return withFileClient((sql) => + Effect.gen(function* () { + yield* sql`PRAGMA journal_mode = WAL`; + yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`; + yield* sql`INSERT INTO entries VALUES ('kept')`; + const read = yield* silently( + Effect.gen(function* () { + const fiber = yield* Effect.forkChild( + sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), + ); + yield* Effect.yieldNow; + yield* TestClock.adjust("10 seconds"); + return yield* Fiber.join(fiber); + }), + ); + + assert.equal(read, 1); + assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1); + assert.lengthOf(warnings, 1); + }), + ).pipe(Effect.provide(logs)); +}); + +it.effect("lets a read be cancelled while its worker is still starting", () => + withFileClient((sql) => + Effect.gen(function* () { + yield* sql`PRAGMA journal_mode = WAL`; + yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`; + yield* silently( + Effect.gen(function* () { + const fiber = yield* Effect.forkChild( + sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), + ); + yield* Effect.yieldNow; + // Never returns if the wait for the worker were uninterruptible: the + // test clock has not moved, so the startup deadline cannot fire. + yield* Fiber.interrupt(fiber); + }), + ); + + // The client stays usable: the next read waits out the deadline, then uses the writer. + const next = yield* Effect.forkChild( + sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), + ); + yield* Effect.yieldNow; + yield* TestClock.adjust("10 seconds"); + assert.equal(yield* Fiber.join(next), 0); + }), + ), +); + +it.effect("logs one warning when concurrent first reads find the reader unavailable", () => { + const { warnings, layer: logs } = captureWarnings(); + return withFileClient((sql, filename) => + Effect.gen(function* () { + yield* sql`PRAGMA journal_mode = WAL`; + yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`; + yield* (yield* FileSystem.FileSystem).remove(filename); + const startedBefore = startedWorkers.length; + + const counts = yield* Effect.all( + [ + sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), + sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), + ], + { concurrency: "unbounded" }, + ); + + assert.deepEqual(counts, [0, 0]); + assert.equal(startedWorkers.length, startedBefore + 1); + assert.lengthOf(warnings, 1); + }), + ).pipe(Effect.provide(logs)); +}); + +it.effect("does not hand out a worker that reports ready after another read timed it out", () => + withFileClient((sql) => + Effect.gen(function* () { + yield* sql`PRAGMA journal_mode = WAL`; + yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`; + yield* sql`INSERT INTO entries VALUES ('kept')`; + const read = () => sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly); + const counts = yield* silently( + Effect.gen(function* () { + const first = yield* Effect.forkChild(read()); + yield* Effect.yieldNow; + yield* TestClock.adjust("5 seconds"); + const second = yield* Effect.forkChild(read()); + yield* Effect.yieldNow; + // The first read reaches its deadline and retires the worker; the second + // is still waiting when the worker's "ready" arrives, before it exits. + silentWorkers.readyOnTerminate = true; + yield* TestClock.adjust("5 seconds"); + return [yield* Fiber.join(first), yield* Fiber.join(second)]; + }), + ); + + assert.deepEqual(counts, [1, 1]); + }), + ), +); diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index 6b05d8ac6ca1..3045cac5fd27 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -5,6 +5,7 @@ * @module SqliteClient */ import * as NodeSqlite from "node:sqlite"; +import * as NodeWorkerThreads from "node:worker_threads"; import * as Cache from "effect/Cache"; import * as Duration from "effect/Duration"; @@ -13,6 +14,7 @@ import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import { identity } from "effect/Function"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Predicate from "effect/Predicate"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; @@ -36,8 +38,38 @@ export interface SqliteClientConfig { readonly spanAttributes?: Record | undefined; readonly transformResultNames?: ((str: string) => string) | undefined; readonly transformQueryNames?: ((str: string) => string) | undefined; + // T3-CUSTOM(expbkt3): BEGIN upstream 8659fcf145 (#19226) as cherry-picked onto stage. + // Later upstream commits to this file (#19305, #19324, #19417, #19425, #19559) are + // not on stage, so these lines differ from upstream main. Upstream owns all of it: + // on the next upstream merge take upstream's side and drop the markers. + /** + * Run transactions marked with `readOnly` on a second connection in a worker + * thread, so long reads block neither the event loop nor writes. Ignored for + * in-memory databases, which a second connection cannot share. + */ + readonly readerWorker?: boolean | undefined; + /** How long a reader worker may take to open the database before reads use the writer. Default 10 seconds. */ + readonly readerStartupTimeout?: Duration.Input | undefined; } +/** + * Set by `readOnly`. A top-level transaction started while it is true runs on + * the reader worker when the client has one, and on the writer otherwise. + */ +const ReadOnly = Context.Reference("@t3tools/shared/nodeSqliteClient/ReadOnly", { + defaultValue: () => false, +}); + +/** + * Marks the SQL in `effect` as read-only. Wrap the whole transaction, as in + * `sql.withTransaction(reads).pipe(readOnly)`. Inside an open write + * transaction the reads keep using the writer, and a write attempted on the + * reader fails with "attempt to write a readonly database". + */ +export const readOnly = (effect: Effect.Effect): Effect.Effect => + Effect.provideService(effect, ReadOnly, true); +// T3-CUSTOM(expbkt3): END + export class UnsupportedNodeSqliteVersionError extends Schema.TaggedError()( "UnsupportedNodeSqliteVersionError", { @@ -99,6 +131,308 @@ const classifyError = (cause: unknown, message: string, operation: string) => { return classifySqliteError(cause, { message, operation }); }; +// Evaluated source instead of a worker file, so the reader also starts from +// bundles and single-executable builds that have no module file on disk. +const readerWorkerSource = ` +const { parentPort, workerData } = require("node:worker_threads"); +const { DatabaseSync } = require("node:sqlite"); +let db; +try { + // A read-only open cannot be undone by SQL, unlike PRAGMA query_only. + const next = new DatabaseSync(workerData.filename, { + readOnly: true, + allowExtension: workerData.allowExtension, + }); + next.exec("PRAGMA busy_timeout = 5000;"); + // Without WAL a reader's shared lock blocks the writer's commits, so refuse. + const mode = String(next.prepare("PRAGMA journal_mode").get().journal_mode).toLowerCase(); + if (mode !== "wal") throw new Error("database uses journal_mode " + mode + ", not wal"); + db = next; + parentPort.postMessage({ type: "ready" }); +} catch (error) { + parentPort.postMessage({ type: "unavailable", message: String(error && error.message ? error.message : error) }); +} +const statements = new Map(); +const prepare = (sql, cached) => { + if (!cached) return db.prepare(sql); + let statement = statements.get(sql); + if (statement === undefined) { + statement = db.prepare(sql); + if (statements.size >= workerData.prepareCacheSize) statements.delete(statements.keys().next().value); + } else { + statements.delete(sql); + } + statements.set(sql, statement); + return statement; +}; +parentPort.on("message", ({ id, sql, params, cached, values, raw, safeIntegers }) => { + let operation = "open"; + try { + if (db === undefined) throw new Error("database reader is unavailable"); + operation = "prepare"; + const statement = prepare(sql, cached); + operation = "execute"; + statement.setReadBigInts(safeIntegers); + if (statement.columns().length > 0) { + statement.setReturnArrays(values); + parentPort.postMessage({ id, rows: statement.all(...params) }); + } else { + const result = statement.run(...params); + parentPort.postMessage({ id, rows: raw ? result : [] }); + } + } catch (error) { + parentPort.postMessage({ + id, + error: { + operation, + message: String(error && error.message ? error.message : error), + code: error && error.code, + errcode: error && error.errcode, + errstr: error && error.errstr, + }, + }); + } +}); +`; + +interface ReaderRequest { + readonly sql: string; + readonly params: ReadonlyArray; + readonly cached: boolean; + readonly values: boolean; + readonly raw: boolean; +} + +type ReaderStartup = + | { readonly type: "ready" } + | { readonly type: "unavailable"; readonly message: string }; + +type ReaderReply = + | { readonly id: number; readonly rows: ReadonlyArray } + | { + readonly id: number; + readonly error: { + readonly operation: "open" | "prepare" | "execute"; + readonly message: string; + readonly code?: unknown; + readonly errcode?: unknown; + readonly errstr?: unknown; + }; + }; + +const readerFailureMessages = { + open: "Failed to open database", + prepare: "Failed to prepare statement", + execute: "Failed to execute statement", +} as const; + +const failed = (cause: unknown, message: string, operation: string) => + new SqlError({ reason: classifyError(cause, message, operation) }); + +// COMMIT, ROLLBACK and ROLLBACK TO SAVEPOINT. On a reader they change nothing, +// so they succeed once the worker is gone instead of failing the cleanup. +const endsReadTransaction = (sql: string) => /^\s*ROLLBACK\b|^\s*COMMIT\b/i.test(sql); + +// T3-CUSTOM(expbkt3): BEGIN upstream 8659fcf145 (#19226), see the note on readerWorker. +/** + * Read-only connections served by one worker thread at a time. Each worker + * gets its own `Connection`, so a transaction stays on the worker it began on: + * once that worker exits, its statements fail rather than silently continuing + * outside the transaction on a replacement. The next acquire starts a new one. + * A worker that cannot open the database read-only, or finds it outside WAL + * mode, turns the reader off for the life of the client: reads use the writer. + */ +const makeReader = Effect.fnUntraced(function* (options: SqliteClientConfig) { + let current: + | { + readonly worker: NodeWorkerThreads.Worker; + readonly connection: Connection; + readonly started: Promise; + } + | undefined; + let unavailable = false; + let nextRequestId = 0; + + const start = () => { + const worker = new NodeWorkerThreads.Worker(readerWorkerSource, { + eval: true, + workerData: { + filename: options.filename, + allowExtension: options.allowExtension ?? false, + prepareCacheSize: options.prepareCacheSize ?? 200, + }, + }); + let announce: (startup: ReaderStartup) => void = () => {}; + const started = new Promise((resolve) => { + announce = resolve; + }); + let stoppedMessage: string | undefined; + const pending = new Map< + number, + { readonly endsTransaction: boolean; readonly resume: (reply: ReaderReply) => void } + >(); + const stop = (message: string) => { + stoppedMessage = message; + announce({ type: "unavailable", message }); + if (current?.worker === worker) current = undefined; + for (const [id, request] of pending) { + request.resume( + request.endsTransaction + ? { id, rows: [] } + : { id, error: { operation: "execute", message } }, + ); + } + pending.clear(); + }; + worker.on("message", (reply: ReaderReply | ReaderStartup) => { + if ("type" in reply) { + announce(reply); + // An idle reader must not keep the process alive. + if (pending.size === 0) worker.unref(); + return; + } + const request = pending.get(reply.id); + if (request === undefined) return; + pending.delete(reply.id); + if (pending.size === 0) worker.unref(); + request.resume(reply); + }); + worker.on("error", (cause) => stop(`Database reader failed: ${cause.message}`)); + worker.on("exit", (code) => stop(`Database reader exited with code ${code}`)); + + const request = (input: ReaderRequest) => + Effect.withFiber, SqlError>((fiber) => { + const endsTransaction = endsReadTransaction(input.sql); + if (stoppedMessage !== undefined) { + return endsTransaction + ? Effect.succeed([]) + : Effect.fail( + failed(new Error(stoppedMessage), readerFailureMessages.execute, "execute"), + ); + } + const safeIntegers = Boolean(Context.get(fiber.context, Client.SafeIntegers)); + return Effect.callback, SqlError>((resume) => { + const id = nextRequestId++; + pending.set(id, { + endsTransaction, + resume: (reply) => { + if ("rows" in reply) { + resume(Effect.succeed(reply.rows)); + return; + } + const { operation, message, ...codes } = reply.error; + resume( + Effect.fail( + failed( + Object.assign(new Error(message), codes), + readerFailureMessages[operation], + operation, + ), + ), + ); + }, + }); + if (pending.size === 1) worker.ref(); + // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node workers do not accept a target origin. + worker.postMessage({ id, safeIntegers, ...input }); + }); + }); + + const connection = identity({ + execute(sql, params, rowTransform) { + const effect = request({ sql, params, cached: true, values: false, raw: false }); + return rowTransform ? Effect.map(effect, rowTransform) : effect; + }, + executeRaw(sql, params) { + return request({ sql, params, cached: true, values: false, raw: true }); + }, + executeValues(sql, params) { + return request({ sql, params, cached: true, values: true, raw: false }); + }, + executeValuesUnprepared(sql, params) { + return request({ sql, params: params ?? [], cached: false, values: true, raw: false }); + }, + executeUnprepared(sql, params, rowTransform) { + // A read-only connection rejects BEGIN IMMEDIATE and BEGIN EXCLUSIVE, + // and a deferred BEGIN is all a read snapshot needs. + const statement = /^\s*BEGIN\b/i.test(sql) ? "BEGIN" : sql; + const effect = request({ + sql: statement, + params: params ?? [], + cached: false, + values: false, + raw: false, + }); + return rowTransform ? Effect.map(effect, rowTransform) : effect; + }, + executeStream(_sql, _params) { + return Stream.die(new UnsupportedNodeSqliteOperationError()); + }, + }); + current = { worker, connection, started }; + return current; + }; + + yield* Effect.addFinalizer(() => + Effect.promise(async () => { + await current?.worker.terminate(); + }), + ); + + const startupTimeout = Duration.fromInputUnsafe( + options.readerStartupTimeout ?? Duration.seconds(10), + ); + + // Runs once: concurrent first acquires share one startup and may all see it fail. + const turnOff = (reader: typeof current, reason: string) => + Effect.suspend(() => { + if (reader !== undefined) { + if (current === reader) current = undefined; + void reader.worker.terminate(); + } + if (unavailable) return Effect.succeed(undefined); + unavailable = true; + return Effect.logWarning("SQLite reader worker unavailable; reads use the main connection", { + reason, + }).pipe(Effect.as(undefined)); + }); + + /** The live worker's connection once it is ready, or undefined to use the writer. */ + return Effect.suspend((): Effect.Effect => { + if (unavailable) return Effect.succeed(undefined); + let reader: NonNullable; + try { + reader = current ?? start(); + } catch (cause) { + return turnOff(undefined, String(cause)); + } + // SqlClient acquires transaction connections uninterruptibly. Waiting for the + // worker happens before any lock is taken, so let a cancelled request leave. + return Effect.flatMap( + Effect.interruptible( + Effect.timeoutOption( + Effect.promise(() => reader.started), + startupTimeout, + ), + ), + (startup): Effect.Effect => + // Another acquire may have timed out and retired this worker while this one + // waited; a late "ready" must not hand out a worker that is being terminated. + unavailable || current !== reader + ? Effect.succeed(undefined) + : Option.isSome(startup) && startup.value.type === "ready" + ? Effect.succeed(reader.connection) + : turnOff( + reader, + Option.isSome(startup) && startup.value.type === "unavailable" + ? startup.value.message + : `worker did not open the database within ${Duration.toMillis(startupTimeout)} ms`, + ), + ); + }); +}); +// T3-CUSTOM(expbkt3): END + const make = Effect.fn("makeWithDatabase")(function* ( options: SqliteClientConfig, ): Effect.fn.Return { @@ -261,16 +595,42 @@ const make = Effect.fn("makeWithDatabase")(function* ( const semaphore = yield* Semaphore.make(1); const connection = yield* makeConnection; + // T3-CUSTOM(expbkt3): BEGIN upstream 8659fcf145 (#19226), see the note on readerWorker. + const acquireReader = + options.readerWorker && options.filename !== ":memory:" && options.filename !== "" + ? yield* makeReader(options) + : undefined; + const readerSemaphore = yield* Semaphore.make(1); - const acquirer = semaphore.withPermits(1)(Effect.succeed(connection)); - const transactionAcquirer = Effect.uninterruptibleMask((restore) => { - const fiber = Fiber.getCurrent()!; - const scope = Context.getUnsafe(fiber.context, Scope.Scope); - return Effect.as( - Effect.tap(restore(semaphore.take(1)), () => Scope.addFinalizer(scope, semaphore.release(1))), - connection, - ); - }); + const writer = { lock: semaphore, connection }; + // Picks the reader only for `readOnly` work, and the writer whenever the + // reader is turned off. + const route = Effect.withFiber((fiber) => + acquireReader !== undefined && Context.get(fiber.context, ReadOnly) + ? Effect.map(acquireReader, (readerConnection) => + readerConnection === undefined + ? writer + : { lock: readerSemaphore, connection: readerConnection }, + ) + : Effect.succeed(writer), + ); + + const acquirer = Effect.flatMap(route, (target) => + target.lock.withPermits(1)(Effect.succeed(target.connection)), + ); + const transactionAcquirer = Effect.flatMap(route, (target) => + Effect.uninterruptibleMask((restore) => { + const fiber = Fiber.getCurrent()!; + const scope = Context.getUnsafe(fiber.context, Scope.Scope); + return Effect.as( + Effect.tap(restore(target.lock.take(1)), () => + Scope.addFinalizer(scope, target.lock.release(1)), + ), + target.connection, + ); + }), + ); + // T3-CUSTOM(expbkt3): END return yield* Client.make({ acquirer,