Skip to content

Commit 59da830

Browse files
authored
Merge pull request #150 from remcovanmook/perf/poll-rides-request-batch
perf(api): the sync freshness poll rides along the request's D1 batch
2 parents 5c59171 + 49b2a7e commit 59da830

7 files changed

Lines changed: 258 additions & 22 deletions

File tree

‎workers/api/index.js‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ import { validateRequest, routeAdminPath, wrapHandler } from '../core/admin.js';
1010
import { handlePreflight, jsonError, H_API_AUTH, H_API_ANON, H_API_SHARED, H_NOCACHE_AUTH, H_NOCACHE_ANON, isNotModifiedSince, lastModifiedHeader, syncAlignedCacheControl } from './http.js';
1111
import { isAuthSensitive } from './auth_scope.js';
1212
import { handleList, handleDetail, handleAsSet, handleCompare, handleNotImplemented } from './handlers/index.js';
13-
import { ensureSyncFreshness, getEntityVersion, handleStatus } from './sync_state.js';
13+
import { claimSyncPoll, runSyncPoll, getEntityVersion, handleStatus } from './sync_state.js';
1414
import { withD1Stats } from '../core/d1stats.js';
1515
import { ENTITY_TAGS, ENTITIES, validateFields, validateQuery, resolveImplicitFilters, dropUnknownFilters } from './entities.js';
1616
import { getCacheStats, purgeAllCaches } from './cache.js';
@@ -144,10 +144,7 @@ async function handleRequest(request, env, ctx) {
144144
return jsonError(404, "Not found");
145145
}
146146

147-
// O(1) hot-path hook: trigger background D1 poll if 15s have passed.
148-
// Scoped to entity routes only — admin/health/status don't need it.
149147
const now = Date.now();
150-
ensureSyncFreshness(session, ctx, now);
151148

152149
// Write methods on API paths → 501 Not Implemented
153150
if (WRITE_METHODS.has(request.method)) {
@@ -268,9 +265,17 @@ async function handleRequest(request, env, ctx) {
268265
// header dict, avoiding a second Response + Headers allocation.
269266
/** @type {HandlerContext} */
270267
const hc = { request, db, d1, ctx, entityTag, filters, opts, rawPath: cachePath, queryString, authenticated, hApi: hEntity, entityVersionMs, userId: shared ? null : userId, paging, authMs };
268+
// Sync freshness poll (every 15 s per isolate): rides along the
269+
// request's first D1 call in the same batch — no extra round trip, and
270+
// no separate query racing the request's own. A request that never
271+
// reaches D1 (L1/L2 hit) runs it after the response instead.
272+
const poll = claimSyncPoll(now);
273+
if (poll) db.carry(poll);
271274
const response = id > 0
272275
? await handleDetail(hc, id)
273276
: await handleList(hc);
277+
const unsent = db.takeRider();
278+
if (unsent) ctx.waitUntil(runSyncPoll(session, unsent));
274279
recordCacheTier(env, hc, response.status, shared, since > 0);
275280
return response;
276281
}

‎workers/api/sync_state.js‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,5 +30,7 @@ const _sync = createSyncState({
3030
});
3131

3232
export const ensureSyncFreshness = _sync.ensureSyncFreshness;
33+
export const claimSyncPoll = _sync.claimPoll;
34+
export const runSyncPoll = _sync.runPoll;
3335
export const handleStatus = _sync.handleStatus;
3436
export const getEntityVersion = _sync.getEntityVersion;

‎workers/core/d1stats.js‎

Lines changed: 43 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -10,10 +10,18 @@
1010
* `.first()` returns no meta, so the wrapper serves it through `.all()`:
1111
* same single call, same data (D1's first() fetches the result set and
1212
* returns the first row).
13+
*
14+
* Riders: `carry(rider)` attaches one extra read statement (the sync
15+
* freshness poll) to the request's FIRST D1 call — sent in the same batch,
16+
* so it costs no round trip; its results go to `rider.apply`, not to the
17+
* caller, and its meta does not count toward the request's stats. If the
18+
* request makes no D1 call, `takeRider()` hands it back for the caller to
19+
* run separately.
1320
*/
1421

1522
/**
1623
* @typedef {{ calls: number, sqlMs: number, rowsRead: number }} D1Stats
24+
* @typedef {{ sql: string, apply: (results: Record<string, any>[]) => void }} Rider
1725
*/
1826

1927
/**
@@ -32,19 +40,41 @@ export function withD1Stats(db) {
3240
if (Number.isFinite(meta.rows_read)) stats.rowsRead += meta.rows_read;
3341
};
3442

43+
/** @type {Rider|null} */
44+
let rider = null;
45+
46+
/** @param {Rider} r @param {any} result */
47+
const deliver = (r, result) => {
48+
try { r.apply(result?.results || []); } catch (err) { console.error('[d1] rider apply failed:', err); }
49+
};
50+
51+
/**
52+
* Executes one statement's all(), sending the pending rider (if any) in
53+
* the same batch.
54+
* @param {any} st
55+
*/
56+
const allWithRider = async (st) => {
57+
const r = rider;
58+
if (!r) return st.all();
59+
rider = null;
60+
const [main, extra] = await db.batch([st, db.prepare(r.sql)]);
61+
deliver(r, extra);
62+
return main;
63+
};
64+
3565
/** @param {any} st */
3666
const wrap = (st) => ({
3767
__inner: st,
3868
bind: (/** @type {any[]} */ ...args) => wrap(st.bind(...args)),
3969
all: async () => {
4070
stats.calls++;
41-
const r = await st.all();
71+
const r = await allWithRider(st);
4272
record(r?.meta);
4373
return r;
4474
},
4575
first: async (/** @type {string} */ col) => {
4676
stats.calls++;
47-
const r = await st.all();
77+
const r = await allWithRider(st);
4878
record(r?.meta);
4979
const row = r?.results?.[0] ?? null;
5080
if (col === undefined) return row;
@@ -63,10 +93,19 @@ export function withD1Stats(db) {
6393
prepare: (/** @type {string} */ sql) => wrap(db.prepare(sql)),
6494
batch: async (/** @type {any[]} */ stmts) => {
6595
stats.calls++;
66-
const results = await db.batch(stmts.map(s => s?.__inner ?? s)); // ap-ok: cold path, one map per batch
67-
for (const r of results) record(r?.meta);
96+
const inner = stmts.map(s => s?.__inner ?? s); // ap-ok: cold path, one map per batch
97+
const r = rider;
98+
rider = null;
99+
if (r) inner.push(db.prepare(r.sql));
100+
const results = await db.batch(inner);
101+
if (r) deliver(r, results.pop());
102+
for (const res of results) record(res?.meta);
68103
return results;
69104
},
105+
/** Attaches a read statement to the next D1 call (see file overview). @param {Rider} r */
106+
carry: (r) => { rider = r; },
107+
/** The rider, if no D1 call took it (then cleared). @returns {Rider|null} */
108+
takeRider: () => { const r = rider; rider = null; return r; },
70109
withSession: () => session,
71110
getBookmark: () => db.getBookmark?.() ?? null,
72111
};

‎workers/core/sync_state.js‎

Lines changed: 62 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,9 @@
1818
*
1919
* Factory returns:
2020
* ensureSyncFreshness(db, ctx, now) — O(1) hot-path hook
21+
* claimPoll(now) → {sql, apply} | null — the due poll as a statement + result
22+
* step, to send with a query the request makes anyway
23+
* runPoll(db, poll) — run a claimed poll on its own
2124
* handleStatus(request, db, ctx) — pre-encoded /status handler (needs statusHeaders)
2225
* getEntityVersion(tag) — returns last_modified_at for L2 key versioning
2326
* refresh(db) — force a synchronous poll (cold boot / tests)
@@ -40,6 +43,8 @@ import { encoder, jsonError } from './http.js';
4043
* @returns {{
4144
* refresh: (db: D1Session) => Promise<void>,
4245
* ensureSyncFreshness: (db: D1Session, ctx: ExecutionContext, now: number) => void,
46+
* claimPoll: (now: number) => ({ sql: string, apply: (results: Record<string, any>[]) => void } | null),
47+
* runPoll: (db: D1Session, poll: { sql: string, apply: (results: Record<string, any>[]) => void }) => Promise<void>,
4348
* handleStatus: (request: Request, db: D1Session, ctx: ExecutionContext) => Promise<Response>,
4449
* getEntityVersion: (tag: string) => number,
4550
* }}
@@ -61,11 +66,12 @@ export function createSyncState({ entityTags, onEntityChange, statusHeaders, che
6166
/** Pre-encoded JSON response for /status. @type {Uint8Array|null} */
6267
let statusPayload = null;
6368

69+
/** The poll statement (tiny: one row per entity). */
70+
const POLL_SQL = 'SELECT entity, last_sync, row_count, updated_at, last_modified_at FROM "_sync_meta" ORDER BY entity';
71+
6472
/**
65-
* Background task: queries _sync_meta via D1 read-replication.
66-
* Compares each entity's last_modified_at against the in-memory snapshot
67-
* and calls onEntityChange for the ones that moved. Rebuilds the
68-
* pre-encoded /status payload when statusHeaders is configured.
73+
* Background task: queries _sync_meta via D1 read-replication and
74+
* applies the result (see applyPoll).
6975
*
7076
* Runs inside ctx.waitUntil() — allocations here do not affect the
7177
* response path.
@@ -75,17 +81,30 @@ export function createSyncState({ entityTags, onEntityChange, statusHeaders, che
7581
*/
7682
async function refresh(db) {
7783
try {
78-
const rows = await db.prepare(
79-
'SELECT entity, last_sync, row_count, updated_at, last_modified_at FROM "_sync_meta" ORDER BY entity'
80-
).all();
81-
82-
if (!rows || !rows.results) return;
84+
const rows = await db.prepare(POLL_SQL).all();
85+
if (rows && rows.results) applyPoll(rows.results);
86+
} catch (err) {
87+
// Background D1 errors must not crash the isolate.
88+
// The previous statusPayload and knownModifiedAt remain valid.
89+
console.error('[sync] background poll failed:', err);
90+
}
91+
}
8392

93+
/**
94+
* Applies a _sync_meta result: compares each entity's last_modified_at
95+
* against the in-memory snapshot and calls onEntityChange for the ones
96+
* that moved; rebuilds the pre-encoded /status payload when statusHeaders
97+
* is configured.
98+
*
99+
* @param {Record<string, any>[]} results - _sync_meta rows.
100+
*/
101+
function applyPoll(results) {
102+
try {
84103
// Build a lookup from the D1 results for fast access.
85104
// Use a Map to avoid dynamic-key objects.
86105
/** @type {Map<string, {last_sync: number, row_count: number, updated_at: string, last_modified_at: number}>} */
87106
const dbState = new Map();
88-
for (const row of rows.results) {
107+
for (const row of results) {
89108
dbState.set(
90109
/** @type {string} */ (row.entity),
91110
{
@@ -133,8 +152,38 @@ export function createSyncState({ entityTags, onEntityChange, statusHeaders, che
133152
}
134153

135154
} catch (err) {
136-
// Background D1 errors must not crash the isolate.
137-
// The previous statusPayload and knownModifiedAt remain valid.
155+
// A malformed result must not crash the isolate.
156+
console.error('[sync] applying poll result failed:', err);
157+
}
158+
}
159+
160+
/**
161+
* Claims the poll if checkIntervalMs has elapsed (and marks it taken):
162+
* the statement and its result step, for the caller to send in the same
163+
* D1 call as a query it makes anyway — no extra round trip. null when
164+
* not due. A claimed poll the caller never sends must go to runPoll.
165+
*
166+
* @param {number} now
167+
* @returns {{ sql: string, apply: (results: Record<string, any>[]) => void } | null}
168+
*/
169+
function claimPoll(now) {
170+
if (now - lastCheck <= checkIntervalMs) return null;
171+
lastCheck = now;
172+
return { sql: POLL_SQL, apply: applyPoll };
173+
}
174+
175+
/**
176+
* Runs a claimed poll on its own (when the request made no D1 call).
177+
*
178+
* @param {D1Session} db
179+
* @param {{ sql: string, apply: (results: Record<string, any>[]) => void }} poll
180+
* @returns {Promise<void>}
181+
*/
182+
async function runPoll(db, poll) {
183+
try {
184+
const rows = await db.prepare(poll.sql).all();
185+
if (rows && rows.results) poll.apply(rows.results);
186+
} catch (err) {
138187
console.error('[sync] background poll failed:', err);
139188
}
140189
}
@@ -206,5 +255,5 @@ export function createSyncState({ entityTags, onEntityChange, statusHeaders, che
206255
return knownModifiedAt.get(tag) || 0;
207256
}
208257

209-
return { refresh, ensureSyncFreshness, handleStatus, getEntityVersion };
258+
return { refresh, ensureSyncFreshness, claimPoll, runPoll, handleStatus, getEntityVersion };
210259
}

‎workers/tests/lib/sqlite_d1.js‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -106,7 +106,9 @@ export function createSqliteD1({ maxValueBytes = Infinity, oversizeError = 'toob
106106
return { success: true, meta: { changes: Number(info.changes) }, results: [] };
107107
},
108108
// D1 batch semantics: reads return rows, writes return meta.changes.
109-
exec: async () => (isRead
109+
// Batched reads count like all()/first() (requests may batch their
110+
// main query with the freshness poll, core/d1stats.js riders).
111+
exec: async () => (isRead && (stats.queries++, isPayload && stats.payloadQueries++, true)
110112
? { success: true, meta: {}, results: stmt.all(...args) }
111113
: { success: true, meta: { changes: Number(stmt.run(...args).changes) }, results: [] }),
112114
};
Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
/**
2+
* @fileoverview The sync freshness poll rides along a request's first D1
3+
* call (same batch, no extra round trip); a request that makes no D1 call
4+
* runs it separately after the response.
5+
*/
6+
import { describe, it, before, after, mock } from 'node:test';
7+
import assert from 'node:assert/strict';
8+
import { createSqliteD1, registerAssetLoader, envFor, mockCtx } from '../../lib/sqlite_d1.js';
9+
10+
registerAssetLoader();
11+
const { default: apiWorker } = await import('../../../api/index.js');
12+
13+
const TS = '2026-01-01T00:00:00Z';
14+
/** @type {any} */ let env;
15+
/** @type {string[][]} */ const batches = [];
16+
/** @type {string[]} */ const direct = [];
17+
18+
before(() => {
19+
mock.timers.enable({ apis: ['Date'], now: Date.parse('2026-10-10T12:00:00Z') });
20+
const { sqlite, db } = createSqliteD1();
21+
sqlite.exec(`INSERT INTO "peeringdb_organization" (id, name, social_media, status, created, updated) VALUES (1, 'Org', '[]', 'ok', '${TS}', '${TS}')`);
22+
const net = sqlite.prepare(`INSERT INTO "peeringdb_network" (id, org_id, name, asn, social_media, info_types, ixp_update_exclude, status, created, updated) VALUES (?, 1, ?, ?, '[]', '[]', '[]', 'ok', '${TS}', '${TS}')`);
23+
net.run(100, 'Net 100', 64100); net.run(101, 'Net 101', 64101);
24+
env = envFor({
25+
...db,
26+
withSession() { return this; },
27+
batch: async (/** @type {any[]} */ stmts) => { batches.push(stmts.map((s) => s.__sql)); return db.batch(stmts); },
28+
prepare: (/** @type {string} */ sql) => {
29+
const wrap = (/** @type {any} */ st) => ({
30+
...st, __sql: sql,
31+
bind: (/** @type {any[]} */ ...a) => wrap(st.bind(...a)),
32+
all: async () => { direct.push(sql); return st.all(); },
33+
first: async () => { direct.push(sql); return st.first(); },
34+
});
35+
return wrap(db.prepare(sql));
36+
},
37+
});
38+
});
39+
after(() => mock.timers.reset());
40+
41+
const reset = () => { batches.length = 0; direct.length = 0; };
42+
const get = async (/** @type {string} */ path) => {
43+
const res = await apiWorker.fetch(new Request(`https://api.pdbfe.dev/api/${path}`), env, mockCtx);
44+
assert.equal(res.status, 200);
45+
return res;
46+
};
47+
const isPoll = (/** @type {string} */ sql) => sql.includes('_sync_meta');
48+
49+
describe('sync freshness poll', () => {
50+
it('due + the request queries D1: one batch carries both, no separate poll query', async () => {
51+
reset();
52+
await get('net/100');
53+
assert.equal(batches.length, 1);
54+
assert.equal(batches[0].length, 2);
55+
assert.ok(batches[0].some(isPoll), 'poll in the batch');
56+
assert.equal(direct.filter(isPoll).length, 0);
57+
});
58+
59+
it('not due (within 15 s): no poll at all', async () => {
60+
reset();
61+
mock.timers.tick(5_000);
62+
await get('net/101');
63+
assert.equal([...batches.flat(), ...direct].filter(isPoll).length, 0);
64+
});
65+
66+
it('due + an L1 hit (no D1 call): the poll runs on its own after the response', async () => {
67+
reset();
68+
mock.timers.tick(16_000);
69+
await get('net/100'); // cached by the first test
70+
assert.equal(batches.length, 0, 'no batch: the request itself made no D1 call');
71+
assert.equal(direct.filter(isPoll).length, 1, 'poll sent separately');
72+
});
73+
});

0 commit comments

Comments
 (0)