perf(shared): queued SQLite reads stop waiting on one reader while another is free

packages/shared/src/nodeSqliteClient.test.ts

@@ -11,6 +11,7 @@ import * as SqlClient from "effect/sql/SqlClient";
 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";
 
@@ -289,6 +290,52 @@ 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<void>();
+          const release = yield* Deferred.make<void>();
+          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 queuedSpan = yield* Deferred.make<Tracer.Span>();
+      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);
+      }
+    }),
+  ),
+);
+
 // regression: #19226 read-only transactions run on the separate reader connection
 it.effect("rejects a write in a read-only transaction", () =>
   withReaderClient((sql) =>

packages/shared/src/nodeSqliteClient.ts

@@ -1090,36 +1090,47 @@ 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);
 
   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<void>) => Effect.Effect<void>) =>
     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(
+        (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)!;
+          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 },
+          ),
       );
     });
 
@@ -1141,9 +1152,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 = <A, E, R>(
     target: {
@@ -1152,10 +1167,10 @@ const make = Effect.fn("makeWithDatabase")(function* (
       readonly connection: Connection;
     },
     use: (connection: Connection) => Effect.Effect<A, E, R>,
+    start: bigint,
   ): Effect.Effect<A, E, R> =>
     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(() => {
@@ -1209,37 +1224,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,