diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index b7b9b7a0866a2..aea1c956391df 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -15,6 +15,7 @@ import * as SqlError from "effect/sql/SqlError"; import * as Statement from "effect/sql/Statement"; import { ChildProcess, ChildProcessSpawner } from "effect/process"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import type * as Tracer from "effect/Tracer"; import { vi } from "vite-plus/test"; @@ -426,6 +427,80 @@ 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 cancelled = yield* Effect.uninterruptibleMask((restore) => + 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)); + const queued = yield* countEntries(sql).pipe( + Effect.provideService(Statement.CurrentTransformer, observeQueued), + Effect.provideService(Statement.SpanPropagationEnabled, true), + 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); + // 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); + } + yield* Effect.forEach(interrupting, Fiber.join); + assert.equal(maskedRan, 2); + }), + ), +); + // 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 ab14675c7bfc0..5cc46c542fbf5 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -875,11 +875,13 @@ const make = Effect.fn("makeWithDatabase")(function* ( ([acquire, lock]) => ({ acquire, lock, - claims: 0, + claimed: false, }), ), ) : []; + // 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); interface Target { readonly name: Acquired["connection"]; @@ -893,31 +895,45 @@ 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) => + // 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, + ) => Effect.withFiber((fiber): Effect.Effect => { if (readers.length === 0 || readerPool.unavailable || !Context.get(fiber.context, ReadOnly)) { return 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.flatMap(reader.acquire, (readerConnection) => - readerConnection === undefined - ? writer - : Effect.succeed({ - name: "reader", - 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.flatMap( + Effect.provideService(reader.acquire, ReadOnly, restore), + (readerConnection) => + readerConnection === undefined + ? writer + : Effect.succeed({ + name: "reader", + lock: reader.lock, + connection: readerConnection, + }), + ), ); }); @@ -937,17 +953,21 @@ 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, restore), (target) => + restore(borrowTarget(target, use, start)), + ); + }).pipe(Effect.ensuring(Effect.suspend(() => release))); }); const borrowTarget = ( target: Target, 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(() => { @@ -974,17 +994,18 @@ 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()!; + // Started before routing, so lock wait includes waiting for reader admission. + 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 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)), () => { + const acquire = Effect.tap(restore(target.lock.take(1)), () => { if (target.name === "writer") activeWriter = owner; if (pending !== undefined) { pending.acquired = acquired(target, clock, start); @@ -1002,8 +1023,8 @@ const make = Effect.fn("makeWithDatabase")(function* ( Effect.provideService(acquire, Scheduler.PreventSchedulerYield, true), target.connection, ); - }), - ), + }); + }), ); const client = yield* Client.make({