diff --git a/docs/changes/unreleased/1128-the-relay-folds-a-retried-batch-once.md b/docs/changes/unreleased/1128-the-relay-folds-a-retried-batch-once.md new file mode 100644 index 0000000000..4698ab0dba --- /dev/null +++ b/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//`, 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//` 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. diff --git a/relay/src/worker.js b/relay/src/worker.js index 3fa59b2386..3d60322f7f 100644 --- a/relay/src/worker.js +++ b/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,11 +191,16 @@ 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. @@ -184,6 +208,20 @@ function quotaKey(install, day) { return `quota/${install}/${day}`; } +// A folded row's identity lives under seen//. 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. diff --git a/relay/test/worker.test.js b/relay/test/worker.test.js index e8168706a7..e0a0a44520 100644 --- a/relay/test/worker.test.js +++ b/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);