relay: fold a resubmitted row once by its nonce

docs/changes/unreleased/1128-the-relay-folds-a-retried-batch-once.md

@@ -0,0 +1,20 @@
+---
+kind: fixed
+title: The pool relay folds a retried batch once
+pr: 1128
+surface: [engine]
+invalidates:
+  - "A resubmitted batch was folded again. The relay keyed nothing by a row's identity, so a row whose 202 the client never saw — the client holds every row that did not get one and re-sends its outbox on the next judged run and at start-up, and a timeout after the relay had already stored the batch is the ordinary case — was summed into the install's running total a second time and charged against its daily quota a second time. A row's own nonce now names it under `seen/<install>/<nonce>`, and a nonce the relay has already stored is neither folded nor charged again; a batch that was entirely already stored still answers 202 with `accepted: 0`."
+---
+
+The submit path in `relay/src/worker.js` now drops the rows whose nonce it has
+already folded before the quota check and the fold, and records the nonce of
+every row it folds under a per-row `seen/<install>/<nonce>` key with a one-week
+TTL. The key is per row and not per batch because a retry carries the rows that
+did not get a 202, which may be fewer than the batch that first sent them, and
+only the nonce each row carries survives that re-grouping. Idempotence is only
+as strong as KV, which is eventually consistent — two concurrent copies of a
+batch can still both fold a row; within what KV answers, a stored nonce is not
+folded or charged again. A nonce repeated inside one batch folds once too.
+`accepted` counts the rows folded now, so a wholly remembered batch answers 202
+with `accepted: 0` — the client needs only the 202 to mark its rows sent.

relay/src/worker.js

@@ -141,11 +141,29 @@ async function handleRows(request, env) {
     rows.push(row);
   }
 
-  // The per-install per-day quota is checked for the whole batch before
-  // anything is stored: a batch that would exceed it stores nothing.
+  // Keep only the rows the relay has not stored: a row's nonce is its
+  // 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();
+  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) {
+      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 rows) {
+  for (const row of fresh) {
     const list = byDay.get(row.day) || [];
     list.push(row);
     byDay.set(row.day, list);
@@ -159,8 +177,9 @@ async function handleRows(request, env) {
     used.set(day, current);
   }
 
-  // Fold each score into its install's running total for the cell and day.
-  for (const row of rows) {
+  // 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)}`;
     const previous = await env.POOL.get(key);
     let triple = null;
@@ -172,18 +191,37 @@ async function handleRows(request, env) {
       }
     }
     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 [day, list] of byDay) {
     await env.POOL.put(quotaKey(install, day), String(used.get(day) + list.length));
   }
-  return json({ accepted: rows.length }, 202);
+  // 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);
 }
 
 // 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.
+const SEEN_TTL = 7 * 24 * 60 * 60;
+function seenKey(install, nonce) {
+  return `seen/${install}/${nonce}`;
+}
+
 // refreshInBackground republishes the document behind a read when the
 // stored version is missing or older than PUBLISH_EVERY seconds. The read
 // itself still answers what is stored, so callers never wait on a publish.

relay/test/worker.test.js

@@ -46,6 +46,10 @@ class KV {
 
 const INSTALL = '0123456789abcdef0123456789abcdef';
 
+// Two distinct 32-hex nonces, one per row when a test folds more than one.
+const NONCE_A = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa';
+const NONCE_B = 'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb';
+
 function payload(overrides = {}) {
   return {
     schema: 1, metric: 'role_quality', role: 'worker',
@@ -55,9 +59,9 @@ function payload(overrides = {}) {
   };
 }
 
-function line(overrides = {}) {
+function line({ nonce = '0123456789abcdef0123456789abcdef', ...overrides } = {}) {
   return JSON.stringify({
-    schema: 1, day: '2026-09-17', nonce: '0123456789abcdef0123456789abcdef',
+    schema: 1, day: '2026-09-17', nonce,
     payload: payload(overrides),
   });
 }
@@ -138,10 +142,64 @@ test('submit refuses an oversized body and too many lines', async () => {
 
 test('submit folds a batch into sheets and answers 202', async () => {
   const e = env();
-  const body = line({ score: 50 }) + '\n' + line({ score: 60 }) + '\n';
+  const body = line({ nonce: NONCE_A, score: 50 }) + '\n' + line({ nonce: NONCE_B, score: 60 }) + '\n';
+  const res = await worker.fetch(post(body, { 'X-Codeaf-Install': INSTALL }), e);
+  assert.equal(res.status, 202);
+  assert.deepEqual(await res.json(), { accepted: 2 });
+  const stored = JSON.parse(e.POOL.map.get(
+    `sheet/${INSTALL}/2026-09-17/worker|z-ai/glm-5.3-flash|anthropic/claude-opus-5|task|M`,
+  ));
+  assert.deepEqual(stored, { n: 2, s: 110, s2: 6100 });
+  assert.equal(e.POOL.map.get(`quota/${INSTALL}/2026-09-17`), '2');
+});
+
+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';
+  const first = await worker.fetch(post(body, { 'X-Codeaf-Install': INSTALL }), e);
+  assert.equal(first.status, 202);
+  assert.deepEqual(await first.json(), { accepted: 2 });
+
+  const second = await worker.fetch(post(body, { 'X-Codeaf-Install': INSTALL }), e);
+  assert.equal(second.status, 202);
+  assert.deepEqual(await second.json(), { accepted: 0 });
+
+  const stored = JSON.parse(e.POOL.map.get(
+    `sheet/${INSTALL}/2026-09-17/worker|z-ai/glm-5.3-flash|anthropic/claude-opus-5|task|M`,
+  ));
+  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');
+});
+
+test('a batch that repeats a nonce already stored and adds one new row folds only the new row', async () => {
+  const e = env();
+  const first = await worker.fetch(
+    post(line({ nonce: NONCE_A, score: 50 }), { 'X-Codeaf-Install': INSTALL }), e,
+  );
+  assert.equal(first.status, 202);
+
+  const body = line({ nonce: NONCE_A, score: 90 }) + '\n' + line({ nonce: NONCE_B, score: 60 }) + '\n';
+  const res = await worker.fetch(post(body, { 'X-Codeaf-Install': INSTALL }), e);
+  assert.equal(res.status, 202);
+  assert.deepEqual(await res.json(), { accepted: 1 });
+
+  const stored = JSON.parse(e.POOL.map.get(
+    `sheet/${INSTALL}/2026-09-17/worker|z-ai/glm-5.3-flash|anthropic/claude-opus-5|task|M`,
+  ));
+  assert.deepEqual(stored, { n: 2, s: 110, s2: 6100 });
+  assert.equal(e.POOL.map.get(`quota/${INSTALL}/2026-09-17`), '2');
+});
+
+test('a nonce repeated inside one batch folds once', async () => {
+  const e = env();
+  const body = line({ nonce: NONCE_A, score: 50 }) + '\n' + line({ nonce: NONCE_A, score: 90 })
+    + '\n' + line({ nonce: NONCE_B, score: 60 }) + '\n';
   const res = await worker.fetch(post(body, { 'X-Codeaf-Install': INSTALL }), e);
   assert.equal(res.status, 202);
   assert.deepEqual(await res.json(), { accepted: 2 });
+
   const stored = JSON.parse(e.POOL.map.get(
     `sheet/${INSTALL}/2026-09-17/worker|z-ai/glm-5.3-flash|anthropic/claude-opus-5|task|M`,
   ));
@@ -151,7 +209,7 @@ test('submit folds a batch into sheets and answers 202', async () => {
 
 test('a row on the exec or run door is accepted and folded, and a row on a door outside the four is refused naming them', async () => {
   const e = env();
-  const body = line({ door: 'exec' }) + '\n' + line({ door: 'run' }) + '\n';
+  const body = line({ nonce: NONCE_A, door: 'exec' }) + '\n' + line({ nonce: NONCE_B, door: 'run' }) + '\n';
   const res = await worker.fetch(post(body, { 'X-Codeaf-Install': INSTALL }), e);
   assert.equal(res.status, 202);
   assert.deepEqual(await res.json(), { accepted: 2 });
@@ -166,7 +224,7 @@ test('a row on the exec or run door is accepted and folded, and a row on a door
 
 test('submit refuses a batch past the daily quota and stores nothing', async () => {
   const e = env({ ROWS_PER_INSTALL_PER_DAY: '1' });
-  const body = line() + '\n' + line() + '\n';
+  const body = line({ nonce: NONCE_A }) + '\n' + line({ nonce: NONCE_B }) + '\n';
   const res = await worker.fetch(post(body, { 'X-Codeaf-Install': INSTALL }), e);
   assert.equal(res.status, 429);
   assert.equal(e.POOL.map.size, 0);