From df0a3206d06cf2ef6eb49bf5296f707b4e472a1a Mon Sep 17 00:00:00 2001 From: Pyseph <45090858+Pyseph@users.noreply.github.com> Date: Sat, 31 Oct 2026 04:14:07 +0100 Subject: [PATCH 1/6] perf(shared): a queued SQLite read takes the first reader that frees up A read that found every reader busy picked one and waited on that reader's own lock, even when another reader freed up first. Admission is now one semaphore with a permit per reader; a read waits for any permit and takes the lowest idle slot. --- packages/shared/src/nodeSqliteClient.test.ts | 34 ++++++++++++++ packages/shared/src/nodeSqliteClient.ts | 47 ++++++++++++-------- 2 files changed, 63 insertions(+), 18 deletions(-) diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index ac786993fa0bf..3be4def3c4331 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -259,6 +259,40 @@ it.effect("serves another read while a read-only transaction is open", () => ), ); +it.effect("gives a queued read the first reader that frees up", () => + withReaderClient((sql) => + Effect.gen(function* () { + const holders = yield* Effect.forEach([0, 1, 2, 3], () => + Effect.gen(function* () { + const held = yield* Deferred.make(); + const release = yield* Deferred.make(); + const fiber = yield* sql + .withTransaction( + countEntries(sql).pipe( + Effect.andThen(Deferred.succeed(held, undefined)), + Effect.andThen(Deferred.await(release)), + ), + ) + .pipe(SqliteClient.readOnly, Effect.forkChild); + yield* Deferred.await(held); + return { fiber, release }; + }), + ); + const queued = yield* countEntries(sql).pipe(SqliteClient.readOnly, (effect) => + Effect.forkChild(effect, { startImmediately: true }), + ); + yield* Deferred.succeed(holders[1]!.release, undefined); + yield* Fiber.join(holders[1]!.fiber); + // Pinned to the first reader it saw, this read would wait for that reader's transaction. + assert.equal(yield* Fiber.join(queued), 1); + for (const { fiber, release } of [holders[0]!, holders[2]!, holders[3]!]) { + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(fiber); + } + }), + ), +); + // regression: #19226 read-only transactions run on the separate reader connection it.effect("rejects a write in a read-only transaction", () => withReaderClient((sql) => diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index c0edfdb868d1f..749a2d38456d4 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -1088,36 +1088,47 @@ const make = Effect.fn("makeWithDatabase")(function* ( ([acquire, lock]) => ({ acquire, lock, - claims: 0, + claimed: false, }), ), ) : []; + // One permit per slot, so a queued read takes whichever reader frees first. + const readerAdmission = yield* Semaphore.make(readers.length); const writer = { name: "writer" as const, lock: semaphore, connection }; let activeWriter: ConnectionOwner | undefined; // Callers must run route uninterruptibly and pass `claimed` a finalizer that runs `release` once. const route = (claimed: (release: Effect.Effect) => Effect.Effect) => Effect.withFiber((fiber) => { - if (readers.length === 0 || readerPool.unavailable || !Context.get(fiber.context, ReadOnly)) { + const restore = Context.get(fiber.context, ReadOnly); + if (readers.length === 0 || readerPool.unavailable || restore === undefined) { return Effect.succeed(writer); } - // Least claimed wins and ties go to the lowest slot, so idle slots never start a worker. - let slot = readers[0]!; - for (const candidate of readers) if (candidate.claims < slot.claims) slot = candidate; - slot.claims++; - const reader = slot; - return Effect.andThen( - claimed( - Effect.sync(() => { - reader.claims--; - }), - ), - Effect.map(reader.acquire, (readerConnection) => - readerConnection === undefined - ? writer - : { name: "reader" as const, lock: reader.lock, connection: readerConnection }, - ), + const claim = Effect.andThen( + restore(readerAdmission.take(1)), + Effect.suspend(() => { + // Holding a permit leaves a slot idle; the lowest wins, so idle slots never start a worker. + const reader = readers.find((slot) => !slot.claimed)!; + reader.claimed = true; + const release = Effect.andThen( + Effect.sync(() => { + reader.claimed = false; + }), + readerAdmission.release(1), + ); + return Effect.as(claimed(release), reader); + }), + ); + return Effect.flatMap( + // A scheduler yield after taking the permit must not precede its release finalizer. + Effect.provideService(claim, Scheduler.PreventSchedulerYield, true), + (reader) => + Effect.map(reader.acquire, (readerConnection) => + readerConnection === undefined + ? writer + : { name: "reader" as const, lock: reader.lock, connection: readerConnection }, + ), ); }); From 4f55253489342dd16565b306ed6471060dcc57e6 Mon Sep 17 00:00:00 2001 From: Pyseph <45090858+Pyseph@users.noreply.github.com> Date: Sat, 31 Oct 2026 05:14:13 +0100 Subject: [PATCH 2/6] docs(shared): say what the reader admission comment guarantees The semaphore stops a queued read from being tied to one busy reader; it does not make waiters first in, first out. --- packages/shared/src/nodeSqliteClient.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index 749a2d38456d4..43b18cc21994b 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -1093,7 +1093,7 @@ const make = Effect.fn("makeWithDatabase")(function* ( ), ) : []; - // One permit per slot, so a queued read takes whichever reader frees first. + // One permit per slot, so a queued read waits for any reader, not one it picked while all were busy. const readerAdmission = yield* Semaphore.make(readers.length); const writer = { name: "writer" as const, lock: semaphore, connection }; From 985d25d665acdbf30a000c5a22433740ca381b4f Mon Sep 17 00:00:00 2001 From: Pyseph <45090858+Pyseph@users.noreply.github.com> Date: Sat, 31 Oct 2026 05:18:51 +0100 Subject: [PATCH 3/6] fix(shared): lock wait includes waiting for reader admission Queueing moved from each reader's lock into routing, which ran before the lock-wait timer started, so a read that waited for a busy pool reported almost no db.lock_wait_ms. Both acquisition paths now start timing before routing. --- packages/shared/src/nodeSqliteClient.test.ts | 13 +++- packages/shared/src/nodeSqliteClient.ts | 75 +++++++++++--------- 2 files changed, 52 insertions(+), 36 deletions(-) diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index 3be4def3c4331..03cb52a3a8474 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -9,6 +9,7 @@ import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; import * as SqlClient from "effect/sql/SqlClient"; import * as Statement from "effect/sql/Statement"; +import * as TestClock from "effect/testing/TestClock"; import type * as Tracer from "effect/Tracer"; import { vi } from "vite-plus/test"; @@ -278,13 +279,21 @@ it.effect("gives a queued read the first reader that frees up", () => return { fiber, release }; }), ); - const queued = yield* countEntries(sql).pipe(SqliteClient.readOnly, (effect) => - Effect.forkChild(effect, { startImmediately: true }), + const queuedSpan = yield* Deferred.make(); + const observeQueued: Statement.Transformer = (statement, _sql, _fiber, span) => + Deferred.succeed(queuedSpan, span).pipe(Effect.as(statement)); + const queued = yield* countEntries(sql).pipe( + Effect.provideService(Statement.CurrentTransformer, observeQueued), + Effect.provideService(Statement.SpanPropagationEnabled, true), + SqliteClient.readOnly, + (effect) => Effect.forkChild(effect, { startImmediately: true }), ); + yield* TestClock.adjust("1 second"); yield* Deferred.succeed(holders[1]!.release, undefined); yield* Fiber.join(holders[1]!.fiber); // Pinned to the first reader it saw, this read would wait for that reader's transaction. assert.equal(yield* Fiber.join(queued), 1); + assert.equal((yield* Deferred.await(queuedSpan)).attributes.get("db.lock_wait_ms"), 1000); for (const { fiber, release } of [holders[0]!, holders[2]!, holders[3]!]) { yield* Deferred.succeed(release, undefined); yield* Fiber.join(fiber); diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index 43b18cc21994b..d9ad406c8f08a 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -1150,9 +1150,13 @@ const make = Effect.fn("makeWithDatabase")(function* ( Effect.sync(() => { release = next; }); - return Effect.flatMap(route(claimed), (target) => restore(borrowTarget(target, use))).pipe( - Effect.ensuring(Effect.suspend(() => release)), - ); + // Started before routing, so lock wait includes waiting for reader admission. + return Effect.withFiber((fiber) => { + const start = fiber.getRef(Clock.Clock).monotonicTimeNanosUnsafe(); + return Effect.flatMap(route(claimed), (target) => + restore(borrowTarget(target, use, start)), + ); + }).pipe(Effect.ensuring(Effect.suspend(() => release))); }); const borrowTarget = ( target: { @@ -1161,10 +1165,10 @@ const make = Effect.fn("makeWithDatabase")(function* ( readonly connection: Connection; }, use: (connection: Connection) => Effect.Effect, + start: bigint, ): Effect.Effect => Effect.withFiber((fiber) => { const clock = fiber.getRef(Clock.Clock); - const start = clock.monotonicTimeNanosUnsafe(); const ownerAtWaitStart = target.name === "writer" ? activeWriter : undefined; const owner = target.name === "writer" ? connectionOwner(fiber) : undefined; const useConnection = Effect.suspend(() => { @@ -1218,37 +1222,40 @@ const make = Effect.fn("makeWithDatabase")(function* ( Effect.withFiber((fiber) => Scope.addFinalizer(Context.getUnsafe(fiber.context, Scope.Scope), release), ); - const transactionAcquirer = Effect.uninterruptible(route(claimedUntilScopeCloses)).pipe( - Effect.flatMap((target) => - Effect.uninterruptibleMask((restore) => { - const fiber = Fiber.getCurrent()!; - const scope = Context.getUnsafe(fiber.context, Scope.Scope); - const clock = fiber.getRef(Clock.Clock); - const start = clock.monotonicTimeNanosUnsafe(); - const pending = fiber.getRef(TransactionAcquire); - const ownerAtWaitStart = target.name === "writer" ? activeWriter : undefined; - const owner = target.name === "writer" ? connectionOwner(fiber) : undefined; - const acquire = Effect.tap((pending?.restore ?? restore)(target.lock.take(1)), () => { - if (target.name === "writer") activeWriter = owner; - if (pending !== undefined) { - pending.acquired = acquired(target, clock, start); - pending.ownerAtWaitStart = ownerAtWaitStart; - } - return Scope.addFinalizer( - scope, - Effect.sync(() => { - if (target.name === "writer" && activeWriter === owner) activeWriter = undefined; - }).pipe(Effect.andThen(target.lock.release(1))), + // Started before routing, so lock wait includes waiting for reader admission. + const transactionAcquirer = Effect.withFiber((routing) => { + const start = routing.getRef(Clock.Clock).monotonicTimeNanosUnsafe(); + return Effect.uninterruptible(route(claimedUntilScopeCloses)).pipe( + Effect.flatMap((target) => + Effect.uninterruptibleMask((restore) => { + const fiber = Fiber.getCurrent()!; + const scope = Context.getUnsafe(fiber.context, Scope.Scope); + const clock = fiber.getRef(Clock.Clock); + const pending = fiber.getRef(TransactionAcquire); + const ownerAtWaitStart = target.name === "writer" ? activeWriter : undefined; + const owner = target.name === "writer" ? connectionOwner(fiber) : undefined; + const acquire = Effect.tap((pending?.restore ?? restore)(target.lock.take(1)), () => { + if (target.name === "writer") activeWriter = owner; + if (pending !== undefined) { + pending.acquired = acquired(target, clock, start); + pending.ownerAtWaitStart = ownerAtWaitStart; + } + return Scope.addFinalizer( + scope, + Effect.sync(() => { + if (target.name === "writer" && activeWriter === owner) activeWriter = undefined; + }).pipe(Effect.andThen(target.lock.release(1))), + ); + }); + return Effect.as( + // A scheduler yield after taking the permit must not precede its release finalizer. + Effect.provideService(acquire, Scheduler.PreventSchedulerYield, true), + target.connection, ); - }); - return Effect.as( - // A scheduler yield after taking the permit must not precede its release finalizer. - Effect.provideService(acquire, Scheduler.PreventSchedulerYield, true), - target.connection, - ); - }), - ), - ); + }), + ), + ); + }); const client = yield* Client.make({ acquirer: transactionAcquirer, From 626491b1d9f32c769ae2fe82fb88f26122b5e21c Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 2 Nov 2026 01:48:46 +0000 Subject: [PATCH 4/6] fix(shared): preserve transaction cancellation during reader admission --- packages/shared/src/nodeSqliteClient.test.ts | 4 ++++ packages/shared/src/nodeSqliteClient.ts | 2 +- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index ee90689f009e8..3e2c95025933f 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -309,6 +309,10 @@ it.effect("gives a queued read the first reader that frees up", () => return { fiber, release }; }), ); + const cancelled = yield* Effect.uninterruptibleMask((restore) => + SqliteClient.readOnly(restore(sql.withTransaction(countEntries(sql)))), + ).pipe((effect) => Effect.forkChild(effect, { startImmediately: true })); + yield* Fiber.interrupt(cancelled); const queuedSpan = yield* Deferred.make(); const observeQueued: Statement.Transformer = (statement, _sql, _fiber, span) => Deferred.succeed(queuedSpan, span).pipe(Effect.as(statement)); diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index c2b4792767c7f..f30b8972b4aa2 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -1108,7 +1108,7 @@ const make = Effect.fn("makeWithDatabase")(function* ( return Effect.succeed(writer); } const claim = Effect.andThen( - restore(readerAdmission.take(1)), + (fiber.getRef(TransactionAcquire)?.restore ?? restore)(readerAdmission.take(1)), Effect.suspend(() => { // Holding a permit leaves a slot idle; the lowest wins, so idle slots never start a worker. const reader = readers.find((slot) => !slot.claimed)!; From 9d4286d3dc95a71e4ce9ce7641eb24c54e2e294f Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 14 Nov 2026 11:43:51 +0000 Subject: [PATCH 5/6] fix(shared): queued standalone reads stay cancellable under a mask Reader admission for a standalone read waited with the restore that readOnly captured. When readOnly was entered inside an uninterruptible mask, that restore kept the mask, so cancelling a queued read waited for a reader to free up. Standalone reads now wait with the borrower's own restore; transaction acquisition still prefers TransactionAcquire's. Co-Authored-By: Claude Opus 5.5 --- packages/shared/src/nodeSqliteClient.test.ts | 4 ++++ packages/shared/src/nodeSqliteClient.ts | 15 ++++++++++----- 2 files changed, 14 insertions(+), 5 deletions(-) diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index 32cb0dbd604df..d0cbbe19f18d2 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -450,6 +450,10 @@ it.effect("gives a queued read the first reader that frees up", () => SqliteClient.readOnly(restore(sql.withTransaction(countEntries(sql)))), ).pipe((effect) => Effect.forkChild(effect, { startImmediately: true })); yield* Fiber.interrupt(cancelled); + const cancelledRead = yield* Effect.uninterruptibleMask((restore) => + SqliteClient.readOnly(restore(countEntries(sql))), + ).pipe((effect) => Effect.forkChild(effect, { startImmediately: true })); + yield* Fiber.interrupt(cancelledRead); const queuedSpan = yield* Deferred.make(); const observeQueued: Statement.Transformer = (statement, _sql, _fiber, span) => Deferred.succeed(queuedSpan, span).pipe(Effect.as(statement)); diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index f08c0f144a172..b13aadbb8c640 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -895,14 +895,19 @@ const make = Effect.fn("makeWithDatabase")(function* ( })); let activeWriter: ConnectionOwner | undefined; // Callers must run route uninterruptibly and pass `claimed` a finalizer that runs `release` once. - const route = (claimed: (release: Effect.Effect) => Effect.Effect) => + // Standalone reads pass their own `restore`: readOnly's may have been captured inside a mask. + const route = ( + claimed: (release: Effect.Effect) => Effect.Effect, + restore?: typeof Effect.interruptible, + ) => Effect.withFiber((fiber): Effect.Effect => { - const restore = Context.get(fiber.context, ReadOnly); - if (readers.length === 0 || readerPool.unavailable || restore === undefined) { + const readOnlyRestore = Context.get(fiber.context, ReadOnly); + if (readers.length === 0 || readerPool.unavailable || readOnlyRestore === undefined) { return writer; } + const admit = restore ?? fiber.getRef(TransactionAcquire)?.restore ?? readOnlyRestore; const claim = Effect.andThen( - (fiber.getRef(TransactionAcquire)?.restore ?? restore)(readerAdmission.take(1)), + admit(readerAdmission.take(1)), Effect.suspend(() => { // Holding a permit leaves a slot idle; the lowest wins, so idle slots never start a worker. const reader = readers.find((slot) => !slot.claimed)!; @@ -951,7 +956,7 @@ const make = Effect.fn("makeWithDatabase")(function* ( // Started before routing, so lock wait includes waiting for reader admission. return Effect.withFiber((fiber) => { const start = fiber.getRef(Clock.Clock).monotonicTimeNanosUnsafe(); - return Effect.flatMap(route(claimed), (target) => + return Effect.flatMap(route(claimed, restore), (target) => restore(borrowTarget(target, use, start)), ); }).pipe(Effect.ensuring(Effect.suspend(() => release))); From d459060e881840b889323af234d6c948a761edf9 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 14 Nov 2026 12:02:28 +0000 Subject: [PATCH 6/6] fix(shared): reader admission keeps the caller's interruption policy Routing fell back to readOnly's captured restore for reservations and used it again for the reader's startup wait. Work masked inside readOnly therefore became interruptible once admitted: a cancelled sql.reserve or withTransaction skipped its body after a reader freed up, where main ran it. route now takes the caller's restore for both waits; the transaction acquirer passes the same restore it uses for the reader lock. Co-Authored-By: Claude Opus 5.5 --- packages/shared/src/nodeSqliteClient.test.ts | 24 ++++++ packages/shared/src/nodeSqliteClient.ts | 90 ++++++++++---------- 2 files changed, 68 insertions(+), 46 deletions(-) diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index d0cbbe19f18d2..aea1c956391df 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -463,6 +463,28 @@ it.effect("gives a queued read the first reader that frees up", () => SqliteClient.readOnly, (effect) => Effect.forkChild(effect, { startImmediately: true }), ); + // Masked read-only work keeps its cancellation pending and still runs once a reader frees up. + let maskedRan = 0; + const maskedRuns = yield* Effect.forEach( + [ + Effect.scoped( + Effect.flatMap(sql.reserve, (connection) => + connection.executeUnprepared("SELECT 1", [], undefined), + ), + ).pipe(Effect.asVoid), + sql.withTransaction(countEntries(sql)).pipe(Effect.asVoid), + ], + (effect) => + effect.pipe( + Effect.andThen(Effect.sync(() => maskedRan++)), + Effect.uninterruptible, + SqliteClient.readOnly, + (masked) => Effect.forkChild(masked, { startImmediately: true }), + ), + ); + const interrupting = yield* Effect.forEach(maskedRuns, (fiber) => + Effect.forkChild(Fiber.interrupt(fiber), { startImmediately: true }), + ); yield* TestClock.adjust("1 second"); yield* Deferred.succeed(holders[1]!.release, undefined); yield* Fiber.join(holders[1]!.fiber); @@ -473,6 +495,8 @@ it.effect("gives a queued read the first reader that frees up", () => yield* Deferred.succeed(release, undefined); yield* Fiber.join(fiber); } + yield* Effect.forEach(interrupting, Fiber.join); + assert.equal(maskedRan, 2); }), ), ); diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index b13aadbb8c640..5cc46c542fbf5 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -895,19 +895,17 @@ const make = Effect.fn("makeWithDatabase")(function* ( })); let activeWriter: ConnectionOwner | undefined; // Callers must run route uninterruptibly and pass `claimed` a finalizer that runs `release` once. - // Standalone reads pass their own `restore`: readOnly's may have been captured inside a mask. + // Waits use the caller's `restore`, not readOnly's, which may have been captured outside a mask. const route = ( claimed: (release: Effect.Effect) => Effect.Effect, - restore?: typeof Effect.interruptible, + restore: typeof Effect.interruptible, ) => Effect.withFiber((fiber): Effect.Effect => { - const readOnlyRestore = Context.get(fiber.context, ReadOnly); - if (readers.length === 0 || readerPool.unavailable || readOnlyRestore === undefined) { + if (readers.length === 0 || readerPool.unavailable || !Context.get(fiber.context, ReadOnly)) { return writer; } - const admit = restore ?? fiber.getRef(TransactionAcquire)?.restore ?? readOnlyRestore; const claim = Effect.andThen( - admit(readerAdmission.take(1)), + restore(readerAdmission.take(1)), Effect.suspend(() => { // Holding a permit leaves a slot idle; the lowest wins, so idle slots never start a worker. const reader = readers.find((slot) => !slot.claimed)!; @@ -925,14 +923,16 @@ const make = Effect.fn("makeWithDatabase")(function* ( // A scheduler yield after taking the permit must not precede its release finalizer. Effect.provideService(claim, Scheduler.PreventSchedulerYield, true), (reader) => - Effect.flatMap(reader.acquire, (readerConnection) => - readerConnection === undefined - ? writer - : Effect.succeed({ - name: "reader", - lock: reader.lock, - connection: readerConnection, - }), + Effect.flatMap( + Effect.provideService(reader.acquire, ReadOnly, restore), + (readerConnection) => + readerConnection === undefined + ? writer + : Effect.succeed({ + name: "reader", + lock: reader.lock, + connection: readerConnection, + }), ), ); }); @@ -995,39 +995,37 @@ const make = Effect.fn("makeWithDatabase")(function* ( Scope.addFinalizer(Context.getUnsafe(fiber.context, Scope.Scope), release), ); // Started before routing, so lock wait includes waiting for reader admission. - const transactionAcquirer = Effect.withFiber((routing) => { - const start = routing.getRef(Clock.Clock).monotonicTimeNanosUnsafe(); - return Effect.uninterruptible(route(claimedUntilScopeCloses)).pipe( - Effect.flatMap((target) => - Effect.uninterruptibleMask((restore) => { - const fiber = Fiber.getCurrent()!; - const scope = Context.getUnsafe(fiber.context, Scope.Scope); - const clock = fiber.getRef(Clock.Clock); - const pending = fiber.getRef(TransactionAcquire); - const ownerAtWaitStart = target.name === "writer" ? activeWriter : undefined; - const owner = target.name === "writer" ? connectionOwner(fiber) : undefined; - const acquire = Effect.tap((pending?.restore ?? restore)(target.lock.take(1)), () => { - if (target.name === "writer") activeWriter = owner; - if (pending !== undefined) { - pending.acquired = acquired(target, clock, start); - pending.ownerAtWaitStart = ownerAtWaitStart; - } - return Scope.addFinalizer( - scope, - Effect.sync(() => { - if (target.name === "writer" && activeWriter === owner) activeWriter = undefined; - }).pipe(Effect.andThen(target.lock.release(1))), - ); - }); - return Effect.as( - // A scheduler yield after taking the permit must not precede its release finalizer. - Effect.provideService(acquire, Scheduler.PreventSchedulerYield, true), - target.connection, + const transactionAcquirer = Effect.uninterruptibleMask((maskRestore) => + Effect.withFiber((fiber) => { + const start = fiber.getRef(Clock.Clock).monotonicTimeNanosUnsafe(); + const pending = fiber.getRef(TransactionAcquire); + const restore = pending?.restore ?? maskRestore; + return Effect.flatMap(route(claimedUntilScopeCloses, restore), (target) => { + const scope = Context.getUnsafe(fiber.context, Scope.Scope); + const clock = fiber.getRef(Clock.Clock); + const ownerAtWaitStart = target.name === "writer" ? activeWriter : undefined; + const owner = target.name === "writer" ? connectionOwner(fiber) : undefined; + const acquire = Effect.tap(restore(target.lock.take(1)), () => { + if (target.name === "writer") activeWriter = owner; + if (pending !== undefined) { + pending.acquired = acquired(target, clock, start); + pending.ownerAtWaitStart = ownerAtWaitStart; + } + return Scope.addFinalizer( + scope, + Effect.sync(() => { + if (target.name === "writer" && activeWriter === owner) activeWriter = undefined; + }).pipe(Effect.andThen(target.lock.release(1))), ); - }), - ), - ); - }); + }); + return Effect.as( + // A scheduler yield after taking the permit must not precede its release finalizer. + Effect.provideService(acquire, Scheduler.PreventSchedulerYield, true), + target.connection, + ); + }); + }), + ); const client = yield* Client.make({ driver: "node:sqlite",