From 9245ef36dff19a8af08b10c0ff0bad22dab7f809 Mon Sep 17 00:00:00 2001 From: Spencer Morris Date: Fri, 16 Oct 2026 11:13:53 -0400 Subject: [PATCH 1/3] fix(shared): preserve SQLite reader shutdown finalizers --- packages/shared/src/nodeSqliteClient.test.ts | 35 ++++++++++++++++++++ packages/shared/src/nodeSqliteClient.ts | 28 ++++++++++------ 2 files changed, 53 insertions(+), 10 deletions(-) diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index 1817cc34c22c4..b502b8f7d70b7 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -705,6 +705,41 @@ it.effect("lets a read be cancelled while its worker is still starting", () => ), ); +it.effect.each([false, true])( + "allows a shutdown finalizer to read after interruption (warm reader: %s)", + (warm) => + 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 ('continuation intent')`; + const read = sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly); + if (warm) yield* read; + const registered = yield* Deferred.make(); + let observed: number | undefined; + const fiber = yield* Effect.scoped( + Effect.gen(function* () { + yield* Effect.addFinalizer(() => + read.pipe( + Effect.andThen((count) => + Effect.sync(() => { + observed = count; + }), + ), + Effect.orDie, + ), + ); + yield* Deferred.succeed(registered, undefined); + return yield* Effect.never; + }), + ).pipe(Effect.forkChild); + yield* Deferred.await(registered); + yield* Fiber.interrupt(fiber); + assert.equal(observed, 1); + }), + ), +); + it.effect("logs one warning when concurrent first reads find the reader unavailable", () => { const { warnings, layer: logs } = captureWarnings(); return withFileClient((sql, filename) => diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index ec60097c63e54..81c9e6d880908 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -47,9 +47,13 @@ export interface SqliteClientConfig { readonly checkpointWorker?: boolean | undefined; } -const ReadOnly = Context.Reference("@t3tools/shared/nodeSqliteClient/ReadOnly", { - defaultValue: () => false, -}); +/** + * Set by `readOnly`. Also preserves the caller's interruption policy across + * SqlClient's uninterruptible transaction acquisition. + */ +const ReadOnly = Context.Reference< + ((effect: Effect.Effect) => Effect.Effect) | undefined +>("@t3tools/shared/nodeSqliteClient/ReadOnly", { defaultValue: () => undefined }); /** * Wrap the whole transaction, as in @@ -58,7 +62,9 @@ const ReadOnly = Context.Reference("@t3tools/shared/nodeSqliteClient/Re * with "attempt to write a readonly database". */ export const readOnly = (effect: Effect.Effect): Effect.Effect => - Effect.provideService(effect, ReadOnly, true); + Effect.uninterruptibleMask((restore) => + restore(Effect.provideService(effect, ReadOnly, restore)), + ); interface Acquired { readonly connection: "writer" | "reader"; @@ -441,15 +447,17 @@ const makeReader = Effect.fnUntraced(function* (options: SqliteClientConfig) { } 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. + // Waiting happens before any lock is taken. Ordinary reads may cancel, + // while interrupted shutdown finalizers must retain their protection. return Effect.flatMap( - Effect.interruptible( - Effect.timeoutOption( + Effect.withFiber((fiber) => { + const restore = Context.get(fiber.context, ReadOnly); + const startup = Effect.timeoutOption( Effect.promise(() => reader.started), startupTimeout, - ), - ), + ); + return restore === undefined ? startup : restore(startup); + }), (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. From a64ba6b18950a1bd9dfdce0ac4073574cddbf555 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 17:20:18 +0000 Subject: [PATCH 2/3] test(shared): cover SQLite reader timeouts during shutdown --- packages/shared/src/nodeSqliteClient.test.ts | 42 ++++++++++---------- 1 file changed, 22 insertions(+), 20 deletions(-) diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index b502b8f7d70b7..2ffc45f6e3355 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -705,39 +705,41 @@ it.effect("lets a read be cancelled while its worker is still starting", () => ), ); -it.effect.each([false, true])( - "allows a shutdown finalizer to read after interruption (warm reader: %s)", - (warm) => - 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 ('continuation intent')`; +it.effect.each(["cold", "warm", "stalled"])( + "completes an interrupted shutdown finalizer's read with a %s reader", + (state) => { + const { warnings, layer: logs } = captureWarnings(); + return withReaderClient((sql) => { + const shutdown = Effect.gen(function* () { const read = sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly); - if (warm) yield* read; + if (state === "warm") yield* read; const registered = yield* Deferred.make(); + const finalizing = yield* Deferred.make(); let observed: number | undefined; const fiber = yield* Effect.scoped( Effect.gen(function* () { yield* Effect.addFinalizer(() => - read.pipe( - Effect.andThen((count) => - Effect.sync(() => { - observed = count; - }), - ), - Effect.orDie, - ), + Effect.gen(function* () { + yield* Deferred.succeed(finalizing, undefined); + observed = yield* read; + }).pipe(Effect.orDie), ); yield* Deferred.succeed(registered, undefined); return yield* Effect.never; }), ).pipe(Effect.forkChild); yield* Deferred.await(registered); - yield* Fiber.interrupt(fiber); + const interrupted = yield* Effect.forkChild(Fiber.interrupt(fiber)); + yield* Deferred.await(finalizing); + if (state === "stalled") yield* TestClock.adjust("10 seconds"); + yield* Fiber.join(interrupted); + assert.equal(observed, 1); - }), - ), + assert.lengthOf(warnings, state === "stalled" ? 1 : 0); + }); + return state === "stalled" ? silently(shutdown) : shutdown; + }).pipe(Effect.provide(logs)); + }, ); it.effect("logs one warning when concurrent first reads find the reader unavailable", () => { From c306ed44966628b6a11ffed4ca3aba860900dac9 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 17:23:48 +0000 Subject: [PATCH 3/3] fix(shared): simplify SQLite reader interruption preservation --- packages/shared/src/nodeSqliteClient.ts | 25 +++++++++++-------------- 1 file changed, 11 insertions(+), 14 deletions(-) diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index 81c9e6d880908..79eab6d84c463 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -47,13 +47,11 @@ export interface SqliteClientConfig { readonly checkpointWorker?: boolean | undefined; } -/** - * Set by `readOnly`. Also preserves the caller's interruption policy across - * SqlClient's uninterruptible transaction acquisition. - */ -const ReadOnly = Context.Reference< - ((effect: Effect.Effect) => Effect.Effect) | undefined ->("@t3tools/shared/nodeSqliteClient/ReadOnly", { defaultValue: () => undefined }); +// SqlClient masks transaction acquisition, so readOnly carries the caller's interruption policy. +const ReadOnly = Context.Reference( + "@t3tools/shared/nodeSqliteClient/ReadOnly", + { defaultValue: () => undefined }, +); /** * Wrap the whole transaction, as in @@ -447,16 +445,15 @@ const makeReader = Effect.fnUntraced(function* (options: SqliteClientConfig) { } catch (cause) { return turnOff(undefined, String(cause)); } - // Waiting happens before any lock is taken. Ordinary reads may cancel, - // while interrupted shutdown finalizers must retain their protection. return Effect.flatMap( Effect.withFiber((fiber) => { - const restore = Context.get(fiber.context, ReadOnly); - const startup = Effect.timeoutOption( - Effect.promise(() => reader.started), - startupTimeout, + const restore = Context.get(fiber.context, ReadOnly) ?? identity; + return restore( + Effect.timeoutOption( + Effect.promise(() => reader.started), + startupTimeout, + ), ); - return restore === undefined ? startup : restore(startup); }), (startup): Effect.Effect => // Another acquire may have timed out and retired this worker while this one