Skip to content

Give the Redis write queue the folding it never had #140

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ralyodio merged 1 commit into from
Aug 19, 2026
Merged

Give the Redis write queue the folding it never had #140

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
118 changes: 9 additions & 109 deletions packages/db/src/client.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ import { randomUUID } from 'node:crypto';

import { createClient } from '@libsql/client';

import { createWriteFolder } from './writeFolder.js';
import { queueWrites } from './writeQueue.js';

/**
Expand Down Expand Up @@ -164,127 +165,26 @@ export function serializeWrites(client, opts = {}) {
const maxStatements = Number.isFinite(configured) && configured > 0 ? Math.floor(configured) : 1;
const original = client.batch.bind(client);

/** @type {Array<{ statements: unknown[], resolve: (value: unknown) => void, reject: (reason: unknown) => void }>} */
const waiting = [];
let draining = false;
// The folding itself lives in writeFolder.js, because the Redis queue needs
// exactly the same thing and having two copies of it is how they drift.
const enqueue = createWriteFolder({
run: (statements) => original(statements, 'write'),
maxStatements,
});

/**
* Put one caller behind the transaction currently in flight.
*
* The first caller starts immediately. Everyone arriving while it awaits the
* remote database accumulates in `waiting` and is folded together on the next
* turn, up to a statement ceiling that keeps one transaction bounded.
*
* @param {unknown[]} statements
* @returns {Promise<unknown>}
*/
const enqueue = (statements) =>
new Promise((resolve, reject) => {
waiting.push({ statements: Array.from(statements ?? []), resolve, reject });
void drain();
});

async function drain() {
if (draining) return;
draining = true;

try {
while (waiting.length > 0) {
const group = [];
let statementCount = 0;

while (waiting.length > 0) {
const next = waiting[0];
const size = next.statements.length;
if (group.length > 0 && statementCount + size > maxStatements) break;
group.push(waiting.shift());
statementCount += size;
}

const statements = group.flatMap((entry) => entry.statements);

try {
const results = await original(statements, 'write');
settleGroup(group, Array.from(results ?? []));
} catch (err) {
if (group.length > 1 && isStatementError(err)) {
// The transaction was rejected because one statement is bad. Run
// each caller on its own to preserve failure isolation and ordering.
for (const entry of group) {
try {
entry.resolve(await original(entry.statements, 'write'));
} catch (singleErr) {
entry.reject(singleErr);
}
}
} else {
for (const entry of group) entry.reject(err);
}
}
}
} finally {
draining = false;
// A caller can arrive after the while condition and before the flag is
// cleared. Do not leave it asleep until an unrelated later write arrives.
if (waiting.length > 0) void drain();
}
}

// The one method replaced, on the instance, rather than the whole client
// wrapped in a Proxy. A Proxy was the first attempt and it broke 154 tests:
// libSQL's client keeps private class fields, and reading one through a
// Proxy receiver throws, so every call that was merely passing through died.
// Overriding the single method that opens a transaction touches nothing else.
client.batch = (statements, mode) =>
// Only transactions that can take the write lock. A read batch is several
// selects and has no business waiting behind a crawl.
// selects and holds nothing.
mode === 'read' ? original(statements, mode) : enqueue(statements);

// `transaction()` is deliberately left alone. It hands the caller an open
// transaction to hold across awaits, which this queue cannot bound -- a lock
// acquired here and released who-knows-when is worse than no lock. Nothing in
// this codebase calls it; if something starts to, it should be a deliberate
// decision rather than a silent hole in the serialisation.
// held while the caller does anything else is the problem, not the solution.

return client;
}

/**
* Hand each caller the result rows belonging to its own statements.
*
* libSQL returns one ResultSet per statement, in order. A caller above this
* layer must never learn that its transaction shared a round trip.
*
* @param {Array<{ statements: unknown[], resolve: (value: unknown) => void }>} group
* @param {unknown[]} results
*/
function settleGroup(group, results) {
let offset = 0;
for (const entry of group) {
const end = offset + entry.statements.length;
entry.resolve(results.slice(offset, end));
offset = end;
}
}

/**
* Whether retrying a failed combined transaction can isolate one bad caller.
*
* Network failures and deadlines affect the database as a whole and must not be
* multiplied into N retries. SQLite statement/constraint errors are local to
* the SQL or data and are worth splitting once so neighbouring feeds survive.
*
* @param {unknown} err
* @returns {boolean}
*/
function isStatementError(err) {
const text = String(err?.message ?? err);
return /SQLITE_(?:CONSTRAINT|ERROR|MISMATCH|RANGE)|constraint failed|syntax error|no such (?:table|column)/i.test(
text,
);
}


/**
* Application-generated primary key.
*
Expand Down
124 changes: 124 additions & 0 deletions packages/db/src/writeFolder.js
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,124 @@
/**
* One write transaction at a time, with waiting callers folded into the next.
*
* Extracted from `serializeWrites` so the Redis queue can use the same thing.
* It was the in-process queue's whole advantage and the Redis path did not have
* it: `queueWrites` posted one job per caller, each job became one remote
* transaction, and with a worker at concurrency 1 that put a hard ceiling on
* cluster write throughput of one transaction's latency — about 370ms, so
* roughly 2.7 writes a second for every process combined. Measured in
* production on 2026-08-19: the import drain stopped entirely for over an hour
* behind a queue of small crawl batches, and item ingestion halved.
*
* The mechanism is deliberately dumb. The first caller runs immediately.
* Everyone who arrives while it is awaiting the database accumulates, and the
* next turn sends them as one transaction, up to a statement ceiling. SQLite
* was going to serialise them anyway; this pays the round trip once instead of
* once per caller.
*
* Nothing here knows whether `run` writes to a database directly or posts a job
* to a queue. That is the point — the folding is worth having on both paths.
*/

/**
* Hand each caller the result rows belonging to its own statements.
*
* libSQL returns one ResultSet per statement, in order. A caller above this
* layer must never learn that its transaction shared a round trip.
*
* @param {Array<{ statements: unknown[], resolve: (value: unknown) => void }>} group
* @param {unknown[]} results
*/
export function settleGroup(group, results) {
let offset = 0;
for (const entry of group) {
const end = offset + entry.statements.length;
entry.resolve(results.slice(offset, end));
offset = end;
}
}

/**
* Whether retrying a failed combined transaction can isolate one bad caller.
*
* Network failures and deadlines affect the database as a whole and must not be
* multiplied into N retries. SQLite statement/constraint errors are local to
* the SQL or data and are worth splitting once so neighbouring feeds survive.
*
* @param {unknown} err
* @returns {boolean}
*/
export function isStatementError(err) {
const text = String(err?.message ?? err);
return /SQLITE_(?:CONSTRAINT|ERROR|MISMATCH|RANGE)|constraint failed|syntax error|no such (?:table|column)/i.test(
text,
);
}

/**
* Build the folding enqueue function.
*
* @param {{
* run: (statements: unknown[]) => Promise<unknown[]>,
* maxStatements?: number,
* }} opts `run` performs one transaction's worth of statements
* @returns {(statements: unknown[]) => Promise<unknown>}
*/
export function createWriteFolder({ run, maxStatements = 1 }) {
const ceiling = Number.isFinite(maxStatements) && maxStatements > 0 ? Math.floor(maxStatements) : 1;

/** @type {Array<{ statements: unknown[], resolve: (value: unknown) => void, reject: (reason: unknown) => void }>} */
const waiting = [];
let draining = false;

async function drain() {
if (draining) return;
draining = true;

try {
while (waiting.length > 0) {
const group = [];
let statementCount = 0;

while (waiting.length > 0) {
const next = waiting[0];
const size = next.statements.length;
// The first caller always goes in, however large it is: a ceiling
// that could refuse the only waiting caller would deadlock.
if (group.length > 0 && statementCount + size > ceiling) break;
group.push(waiting.shift());
statementCount += size;
}

const statements = group.flatMap((entry) => entry.statements);

try {
const results = await run(statements);
settleGroup(group, Array.from(results ?? []));
} catch (err) {
if (group.length > 1 && isStatementError(err)) {
// The transaction was rejected because one statement is bad. Run
// each caller on its own to preserve failure isolation and ordering.
for (const entry of group) {
try {
entry.resolve(await run(entry.statements));
} catch (singleErr) {
entry.reject(singleErr);
}
}
} else {
for (const entry of group) entry.reject(err);
}
}
}
} finally {
draining = false;
}
}

return (statements) =>
new Promise((resolve, reject) => {
waiting.push({ statements: Array.from(statements ?? []), resolve, reject });
void drain();
});
}
96 changes: 88 additions & 8 deletions packages/db/src/writeQueue.js
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
import { Queue, QueueEvents, Worker } from 'bullmq';

import { createWriteFolder } from './writeFolder.js';

/**
* The write path, moved out of the process and into Redis.
*
Expand Down Expand Up @@ -203,13 +205,11 @@ export function connectionFor(url) {
* @returns {import('@libsql/client').Client & { closeWriteQueue: () => Promise<void> }}
*/
export function queueWrites(client, opts) {
const connection = connectionFor(opts.url);
// The Redis instance is shared with every other app in the Railway project,
// so the keyspace is claimed explicitly rather than left on BullMQ's default.
const prefix = opts.prefix ?? '{rssamplifier}';

const queue = new Queue(WRITE_QUEUE, { connection, prefix });
const events = new QueueEvents(WRITE_QUEUE, { connection, prefix });
const { queue, events } = sharedQueue(opts.url, prefix);
const original = client.batch.bind(client);

/**
Expand All @@ -236,20 +236,100 @@ export function queueWrites(client, opts) {
return /** @type {object[]} */ (encoded).map(decodeResult);
};

// Callers waiting at the same moment are folded into one job, which is one
// transaction. Without this every caller was its own job and the worker --
// correctly at concurrency 1, because SQLite has one writer -- could only
// retire them at one transaction's latency each. See writeFolder.js.
const fold = createWriteFolder({ run: enqueue, maxStatements: groupStatements(opts) });

client.batch = (statements, mode) =>
mode === 'read' ? original(statements, mode) : enqueue(statements);
mode === 'read' ? original(statements, mode) : fold(statements);

// `transaction()` is left alone for the reason serializeWrites leaves it
// alone: it hands out a lock held across awaits, which no queue can bound.

return Object.assign(client, {
closeWriteQueue: async () => {
await queue.close();
await events.close();
},
closeWriteQueue: () => releaseQueue(opts.url, prefix),
});
}

/**
* How many statements one queued transaction may carry.
*
* Its own knob rather than `TURSO_WRITE_GROUP_STATEMENTS`, because that one is
* set to 1 in production -- folding off -- for the in-process path, where a
* five-crawl group was once measured to exceed the 30-second request deadline.
* On the queued path folding is not an optimisation, it is the difference
* between a working queue and a 2.7-writes-a-second ceiling, so it defaults on.
*
* Fifty is deliberately modest: one crawl's worth of statements, so a group is
* a handful of callers rather than a transaction big enough to hit the deadline
* the in-process canary hit.
*
* @param {{ maxStatements?: number }} opts
* @returns {number}
*/
function groupStatements(opts) {
const configured = Number(opts.maxStatements ?? process.env['TURSO_QUEUE_GROUP_STATEMENTS']);
return Number.isFinite(configured) && configured > 0 ? Math.floor(configured) : 50;
}

/**
* One Queue and one QueueEvents per (url, prefix), shared by every caller.
*
* `connect()` is called freely -- once per request handler in the web app, more
* in the poller -- and each call used to build its own pair. Each pair opens
* Redis connections and registers listeners, so the poller logged
* `MaxListenersExceededWarning: 11 closing listeners added to [Queue]` within
* minutes and kept climbing. Refcounted rather than cached outright so that
* `closeWriteQueue` still means something to a caller that owns the last one.
*
* @type {Map<string, { queue: import('bullmq').Queue, events: import('bullmq').QueueEvents, refs: number }>}
*/
const shared = new Map();

/**
* @param {string} url
* @param {string} prefix
*/
function sharedQueue(url, prefix) {
const key = `${prefix}\u0000${url}`;
const found = shared.get(key);

if (found) {
found.refs += 1;
return found;
}

const connection = connectionFor(url);
const entry = {
queue: new Queue(WRITE_QUEUE, { connection, prefix }),
events: new QueueEvents(WRITE_QUEUE, { connection, prefix }),
refs: 1,
};

shared.set(key, entry);
return entry;
}

/**
* @param {string} url
* @param {string} prefix
* @returns {Promise<void>}
*/
async function releaseQueue(url, prefix) {
const key = `${prefix}\u0000${url}`;
const found = shared.get(key);
if (!found) return;

found.refs -= 1;
if (found.refs > 0) return;

shared.delete(key);
await found.queue.close();
await found.events.close();
}

/**
* The single consumer that actually writes.
*
Expand Down
Loading
Loading