relay: fold a submit batch with one read and one write per distinct cell and day

docs/changes/unreleased/1136-the-relay-writes-once-per-cell-per-batch.md

@@ -0,0 +1,29 @@
+---
+kind: changed
+title: the relay writes a submit batch once per distinct cell and day
+pr: 1136
+surface: [engine]
+invalidates:
+  - "A folded row's identity lived under its own `seen/<install>/<nonce>` key, so a submit batch spent one KV `get` and one KV `put` per row: an install re-sending a 200-row outbox spent 400 writes and reads to fold a handful of cells. The nonces of a day now live together under one `seen/<install>/<day>` key, and a batch costs one read and one write per distinct (day, cell) it folds into plus one read and one write per day for that day's seen set and its quota counter — 12 writes for 200 rows over ten cells on one day."
+---
+
+The submit path in `relay/src/worker.js` groups a batch's rows by day and by
+their sheet key before touching the store, and reads a day's seen set once
+instead of once per row. `fold` now runs in memory over the rows that share a
+sheet key before the key is written, so two rows of one batch in the same
+(day, cell) fold against each other rather than against whatever KV last
+answered. The counts per batch are writes = distinct (day, cell) keys +
+distinct fresh days (the seen set) + distinct fresh days (the quota counter),
+and the same order of reads; neither scales with the rows.
+
+The idempotence contract is unchanged: a nonce already in its day's set is
+neither folded nor charged again, and a wholly remembered batch still answers
+202 with `accepted: 0`. One consequence of the new key shape: a nonce the old
+per-row key still holds is not consulted, so a retry that straddles the deploy
+may fold once more — the same eventual-consistency window the per-row key had,
+now stated on one key, where two batches from one install on one day race and
+the last put wins. The day's set holds at most `ROWS_PER_INSTALL_PER_DAY`
+nonces (16,500 bytes at the 500 default) against KV's 25 MiB value limit, and a
+quota raised past what one value holds is refused rather than allowed to drop
+nonces. `docs/design/model-pool/RUNBOOK.md` now states the per-batch cost and
+the plan's daily KV write limit.

docs/design/model-pool/RUNBOOK.md

@@ -62,9 +62,24 @@ key.
    (expect 200 and an ETag), then `codeaf pool verify`, then `codeaf pool
    status`.
 9. Know what the store holds — one running total per install, cell and day; a
-   per-install per-day quota counter; and the last index document and its
-   signature — and how to wipe it: `wrangler kv key list` to see the keys and
-   `wrangler kv key delete <key>` to drop one.
+   per-install per-day quota counter; the nonces it has folded for that install
+   and day (`seen/<install>/<day>`, one week's TTL); and the last index document
+   and its signature — and how to wipe it: `wrangler kv key list` to see the
+   keys and `wrangler kv key delete <key>` to drop one.
+
+   What a batch costs the meter: one KV read and one KV write for each distinct
+   (day, cell) it folds into, one read per day the batch carries (that day's
+   seen set), and one read and one write per day it has fresh rows on (that
+   day's seen set and its quota counter) — neither scaling with the row count,
+   so a 200-row batch over ten cells on one day is 12 writes where one write per
+   row was 200. The relay is deployed on Cloudflare Workers' `plan: <operator
+   fills in>` plan; neither this runbook nor `wrangler.toml` names one, so fill
+   it in. Cloudflare's free tier allows 1,000 KV writes and 100,000 KV reads per
+   day (a paid plan is metered, not capped). The day's seen set is at most
+   `ROWS_PER_INSTALL_PER_DAY` nonces of 32 hex characters plus a separator —
+   16,500 bytes at the default 500, against KV's 25 MiB value limit — so a plan
+   that raises the quota past what one value holds is refused rather than
+   allowed to drop nonces.
 10. Rotating the signing key: generate a new seed, add the new public key to the
     binary's trusted key list, then redeploy with the new secret.
 

relay/src/worker.js

@@ -145,42 +145,72 @@ async function handleRows(request, env) {
   // identity, and the client re-sends every row that did not get a 202, so an
   // install's retry is the ordinary case rather than an error. A nonce
   // repeated inside one batch is collapsed here too, by the set, before the
-  // store is consulted.
-  const fresh = [];
-  const batchNonces = new Set();
+  // store is consulted, and the rows are grouped by day because a day's nonces
+  // live together under one key.
+  //
+  // Per batch this costs a KV read and a KV write for each distinct (day,
+  // cell) key it folds into, one read for every day the batch carries (that
+  // day's seen set) and one read and one write for every day it has fresh rows
+  // on (that day's seen set and its quota counter). A batch of 200 rows over
+  // ten cells on one day is 12 writes and 12 reads, not 200 of each.
+  const seenInBatch = new Set();
+  const byDay = new Map();
   for (const row of rows) {
-    if (batchNonces.has(row.nonce)) {
-      continue;
-    }
-    batchNonces.add(row.nonce);
-    if (await env.POOL.get(seenKey(install, row.nonce)) !== null) {
+    if (seenInBatch.has(row.nonce)) {
       continue;
     }
-    fresh.push(row);
-  }
-
-  // The per-install per-day quota is charged for the rows this batch actually
-  // folds, so a retried row is not charged a second time.
-  const limit = intVar(env, 'ROWS_PER_INSTALL_PER_DAY', 500);
-  const byDay = new Map();
-  for (const row of fresh) {
+    seenInBatch.add(row.nonce);
     const list = byDay.get(row.day) || [];
     list.push(row);
     byDay.set(row.day, list);
   }
-  const used = new Map();
+
+  // A day's seen set is one KV value, so a quota raised past what one value
+  // holds would silently drop nonces: refuse rather than lose them.
+  const limit = intVar(env, 'ROWS_PER_INSTALL_PER_DAY', 500);
+  if (limit * SEEN_BYTES_PER_NONCE > MAX_VALUE_BYTES) {
+    return json({ error: 'ROWS_PER_INSTALL_PER_DAY is larger than one seen key holds' }, 500);
+  }
+
+  // One get per day names the nonces already folded, so freshness costs one
+  // read per day rather than one per row.
+  const storedSeen = new Map();
+  const freshByDay = new Map();
   for (const [day, list] of byDay) {
+    const seen = parseSeen(await env.POOL.get(seenKey(install, day)));
+    storedSeen.set(day, seen);
+    const fresh = list.filter((row) => !seen.has(row.nonce));
+    if (fresh.length > 0) {
+      freshByDay.set(day, fresh);
+    }
+  }
+
+  // The per-install per-day quota is charged for the rows this batch actually
+  // folds, so a retried row is not charged a second time.
+  const charged = new Map();
+  for (const [day, list] of freshByDay) {
     const current = parseInt(await env.POOL.get(quotaKey(install, day)), 10) || 0;
     if (current + list.length > limit) {
       return json({ error: 'daily quota exceeded' }, 429);
     }
-    used.set(day, current);
+    charged.set(day, current + list.length);
   }
 
-  // Fold each score into its install's running total for the cell and day, and
-  // record the row's nonce so its retry is not folded again.
-  for (const row of fresh) {
-    const key = `sheet/${install}/${row.day}/${cellKey(row.payload)}`;
+  // Fold every fresh row into its install's running total for the cell and day,
+  // grouping the rows that share a sheet key so each key is read once and
+  // written once and the rows on it are folded in memory rather than by
+  // read-after-write, one row at a time.
+  let accepted = 0;
+  const byKey = new Map();
+  for (const list of freshByDay.values()) {
+    for (const row of list) {
+      const key = `sheet/${install}/${row.day}/${cellKey(row.payload)}`;
+      const group = byKey.get(key) || [];
+      group.push(row);
+      byKey.set(key, group);
+    }
+  }
+  for (const [key, group] of byKey) {
     const previous = await env.POOL.get(key);
     let triple = null;
     if (previous !== null) {
@@ -190,36 +220,68 @@ async function handleRows(request, env) {
         triple = null;
       }
     }
-    await env.POOL.put(key, JSON.stringify(fold(triple, row.payload.score)));
-    await env.POOL.put(seenKey(install, row.nonce), '1', { expirationTtl: SEEN_TTL });
+    for (const row of group) {
+      triple = fold(triple, row.payload.score);
+    }
+    await env.POOL.put(key, JSON.stringify(triple));
+    accepted += group.length;
   }
-  for (const [day, list] of byDay) {
-    await env.POOL.put(quotaKey(install, day), String(used.get(day) + list.length));
+
+  // Record the nonces just folded and charge the days they were folded on: one
+  // put of the day's seen set and one of its quota counter, per day.
+  for (const [day, list] of freshByDay) {
+    const seen = storedSeen.get(day);
+    for (const row of list) {
+      seen.add(row.nonce);
+    }
+    await env.POOL.put(seenKey(install, day), joinSeen(seen), { expirationTtl: SEEN_TTL });
+    await env.POOL.put(quotaKey(install, day), String(charged.get(day)));
   }
   // A batch that was entirely already stored still answers 202: the client
   // needs only the 202 to mark its rows sent, and giving it anything else
   // would leave the outbox retrying rows the relay already holds. `accepted`
   // counts the rows folded now, so it is 0 for such a batch.
-  return json({ accepted: fresh.length }, 202);
+  return json({ accepted }, 202);
 }
 
 // quotaKey names one install's per-day quota counter.
 function quotaKey(install, day) {
   return `quota/${install}/${day}`;
 }
 
-// A folded row's identity lives under seen/<install>/<nonce>. It is per row,
-// not per batch: the client re-sends its outbox row by row, and a retry may
-// carry fewer rows than the batch that first sent them, so only the nonce each
-// row carries survives re-grouping. Idempotence is only as strong as KV — reads
-// and writes are eventually consistent, so two concurrent copies of a batch can
-// both see a nonce unread and fold it twice; within what KV answers, a nonce
-// already stored is not folded again or charged again. The TTL is a week: the
-// outbox retries within days, so a week outlasts any retry still in flight and
-// the store does not grow without bound.
+// A day's seen set lives under seen/<install>/<day>: the nonces already folded
+// for that install and day, joined by newlines, with a one-week TTL. It is per
+// day and not per batch because the client re-groups its rows on retry — a
+// retry carries the rows that did not get a 202, which may be fewer than the
+// batch that first sent them — so only the nonce each row carries survives,
+// and the day's set of them is what a retry is checked against: one get and
+// one put per day rather than one per row. At 32 hex characters per nonce plus
+// a separator that is 33 bytes, the relay's own ROWS_PER_INSTALL_PER_DAY (500
+// by default) bounds the value at 16,500 bytes, under KV's 25 MiB value limit.
+// Idempotence is only as strong as KV — reads and writes are eventually
+// consistent, so two concurrent copies of a batch can still both see a nonce
+// absent and fold it twice — and two batches from one install on one day now
+// race on this single key, where the last put wins and a retry may re-fold;
+// within what KV answers, a nonce already stored is neither folded again nor
+// charged again. The TTL is a week: the outbox retries within days, so a week
+// outlasts any retry still in flight and the store does not grow without
+// bound.
 const SEEN_TTL = 7 * 24 * 60 * 60;
-function seenKey(install, nonce) {
-  return `seen/${install}/${nonce}`;
+const SEEN_BYTES_PER_NONCE = 33; // 32 hex characters plus a separator
+const MAX_VALUE_BYTES = 25 << 20; // KV's per-value limit
+function seenKey(install, day) {
+  return `seen/${install}/${day}`;
+}
+
+// parseSeen reads one day's seen set; a key the store has not written yet is
+// an empty set.
+function parseSeen(value) {
+  return value === null ? new Set() : new Set(value.split('\n').filter((nonce) => nonce !== ''));
+}
+
+// joinSeen writes one day's seen set back as the value its key holds.
+function joinSeen(seen) {
+  return [...seen].join('\n');
 }
 
 // refreshInBackground republishes the document behind a read when the

relay/test/worker.test.js

@@ -11,8 +11,11 @@ class KV {
   constructor() {
     this.map = new Map();
     this.expiry = new Map();
+    this.gets = 0;
+    this.puts = 0;
   }
   async get(key) {
+    this.gets++;
     if (!this.map.has(key)) {
       return null;
     }
@@ -25,6 +28,7 @@ class KV {
     return this.map.get(key);
   }
   async put(key, value, options = {}) {
+    this.puts++;
     this.map.set(key, value);
     if (Number.isFinite(options.expirationTtl)) {
       this.expiry.set(key, Date.now() + options.expirationTtl * 1000);
@@ -50,6 +54,12 @@ const INSTALL = '0123456789abcdef0123456789abcdef';
 const NONCE_A = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa';
 const NONCE_B = 'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb';
 
+// A distinct 32-hex nonce from a row's index, for a batch that needs more
+// than two rows.
+function nonce(i) {
+  return i.toString(16).padStart(32, '0');
+}
+
 function payload(overrides = {}) {
   return {
     schema: 1, metric: 'role_quality', role: 'worker',
@@ -153,6 +163,30 @@ test('submit folds a batch into sheets and answers 202', async () => {
   assert.equal(e.POOL.map.get(`quota/${INSTALL}/2026-09-17`), '2');
 });
 
+test('a batch of six rows over two cells on one day costs one read and one write per distinct cell and day, not per row', async () => {
+  const e = env();
+  let body = '';
+  for (let i = 0; i < 6; i++) {
+    // Six rows, one day, two cells: the door varies, everything else is held.
+    body += line({ nonce: nonce(i), door: i % 2 === 0 ? 'task' : 'do' }) + '\n';
+  }
+  const res = await worker.fetch(post(body, { 'X-Codeaf-Install': INSTALL }), e);
+  assert.equal(res.status, 202);
+  assert.deepEqual(await res.json(), { accepted: 6 });
+
+  // Two sheet keys, the day's seen set, the day's quota counter: 2 + 1 + 1
+  // writes, and the same order of reads, where one write per row is 13 today.
+  assert.ok(e.POOL.puts <= 4, `expected at most 4 writes, saw ${e.POOL.puts}`);
+  assert.ok(e.POOL.gets <= 4, `expected at most 4 reads, saw ${e.POOL.gets}`);
+
+  const cell = `sheet/${INSTALL}/2026-09-17/worker|z-ai/glm-5.3-flash|anthropic/claude-opus-5|`;
+  assert.deepEqual(JSON.parse(e.POOL.map.get(`${cell}task|M`)), { n: 3, s: 150, s2: 7500 });
+  assert.deepEqual(JSON.parse(e.POOL.map.get(`${cell}do|M`)), { n: 3, s: 150, s2: 7500 });
+  assert.equal(e.POOL.map.get(`quota/${INSTALL}/2026-09-17`), '6');
+  // The day's nonces live together under one key, still carrying its TTL.
+  assert.equal(typeof e.POOL.expiry.get(`seen/${INSTALL}/2026-09-17`), 'number');
+});
+
 test('a retried batch is folded once, charged once, and still answers 202', async () => {
   const e = env();
   const body = line({ nonce: NONCE_A, score: 50 }) + '\n' + line({ nonce: NONCE_B, score: 60 }) + '\n';
@@ -170,7 +204,7 @@ test('a retried batch is folded once, charged once, and still answers 202', asyn
   assert.deepEqual(stored, { n: 2, s: 110, s2: 6100 });
   assert.equal(e.POOL.map.get(`quota/${INSTALL}/2026-09-17`), '2');
   // The identity survives the retry only as long as its TTL; a week is stored.
-  assert.equal(typeof e.POOL.expiry.get(`seen/${INSTALL}/${NONCE_A}`), 'number');
+  assert.equal(typeof e.POOL.expiry.get(`seen/${INSTALL}/2026-09-17`), 'number');
 });
 
 test('a batch that repeats a nonce already stored and adds one new row folds only the new row', async () => {