Skip to content
Merged
143 changes: 143 additions & 0 deletions packages/shared/src/nodeSqliteClient.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -238,6 +238,149 @@ it.effect("lets the writer commit while a read-only transaction is open", () =>
),
);

// regression: #24660 one read-only transaction must not hold every other read behind it
it.effect("runs an unrelated read while a read-only transaction holds a reader", () =>
withReaderClient((sql) =>
Effect.gen(function* () {
const held = yield* Deferred.make<void>();
const release = yield* Deferred.make<void>();
const holder = 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);

// The holder ends only after this read, so a single reader permit deadlocks here.
const count = yield* countEntries(sql).pipe(SqliteClient.readOnly);
yield* Deferred.succeed(release, undefined);
yield* Fiber.join(holder);

assert.equal(count, 1);
}),
),
);

it.effect("serves a waiting read before a read requested after it", () =>
withReaderClient((sql) =>
Effect.gen(function* () {
const startedBefore = startedWorkers.length;
const finished: Array<string> = [];
const read = (name: string) =>
countEntries(sql).pipe(
SqliteClient.readOnly,
Effect.andThen(Effect.sync(() => finished.push(name))),
);
const releases = yield* Effect.forEach([0, 1, 2, 3], () => Deferred.make<void>());
const holders = yield* Effect.forEach(releases, (release, index) =>
Effect.gen(function* () {
const held = yield* Deferred.make<void>();
const holder = yield* sql
.withTransaction(
countEntries(sql).pipe(
Effect.andThen(Deferred.succeed(held, undefined)),
Effect.andThen(Deferred.await(release)),
),
)
.pipe(
SqliteClient.readOnly,
// The later read asks for a reader in the same turn this transaction returns one.
Effect.andThen(index === 0 ? read("later") : Effect.void),
Effect.forkChild,
);
yield* Deferred.await(held);
return holder;
}),
);
const earlier = yield* Effect.forkChild(read("earlier"), { startImmediately: true });

yield* Deferred.succeed(releases[0]!, undefined);
yield* Fiber.join(holders[0]!);
const order = [...finished];
yield* Effect.forEach(releases, (release) => Deferred.succeed(release, undefined));
yield* Fiber.join(earlier);
yield* Effect.forEach(holders, Fiber.join);

assert.deepEqual(order, ["earlier", "later"]);
// A fifth worker would mean the holders never filled the pool.
assert.equal(startedWorkers.length, startedBefore + 4);
}),
),
);

const holdReaders = (sql: SqlClient.SqlClient, releases: ReadonlyArray<Deferred.Deferred<void>>) =>
Effect.forEach(releases, (release) =>
Effect.gen(function* () {
const held = yield* Deferred.make<void>();
const holder = 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 holder;
}),
);

it.effect("gives the reader to the next read after a queued read is interrupted", () =>
withReaderClient((sql) =>
Effect.gen(function* () {
const releases = yield* Effect.forEach([0, 1, 2, 3], () => Deferred.make<void>());
const holders = yield* holdReaders(sql, releases);
const abandoned = yield* Effect.forkChild(countEntries(sql).pipe(SqliteClient.readOnly), {
startImmediately: true,
});
yield* Fiber.interrupt(abandoned);
const next = yield* Effect.forkChild(countEntries(sql).pipe(SqliteClient.readOnly), {
startImmediately: true,
});

yield* Deferred.succeed(releases[0]!, undefined);
assert.equal(yield* Fiber.join(next), 1);
yield* Effect.forEach(releases, (release) => Deferred.succeed(release, undefined));
yield* Effect.forEach(holders, Fiber.join);
}),
),
);

it.effect("moves queued reads to the writer in arrival order when the readers turn off", () =>
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')`;
const finished: Array<string> = [];
const read = (name: string) =>
countEntries(sql).pipe(
SqliteClient.readOnly,
Effect.andThen(Effect.sync(() => finished.push(name))),
);
const releases = yield* Effect.forEach([0, 1, 2], () => Deferred.make<void>());
const holders = yield* holdReaders(sql, releases);

// The fourth reader can't open the removed file, which turns the readers off with reads queued.
yield* (yield* FileSystem.FileSystem).remove(filename);
const reads = yield* Effect.forEach(["starting", "first", "second", "third"], (name) =>
Effect.forkChild(read(name), { startImmediately: true }),
);
yield* Effect.forEach(reads, Fiber.join);

assert.deepEqual(
finished.filter((name) => name !== "starting"),
["first", "second", "third"],
);
yield* Effect.forEach(releases, (release) => Deferred.succeed(release, undefined));
yield* Effect.forEach(holders, Fiber.join);
}),
),
);

// regression: #19226 read-only transactions run on the separate reader connection
it.effect("rejects a write in a read-only transaction", () =>
withReaderClient((sql) =>
Expand Down
Loading