From 505e287c86cd16274b2db782224e0da822c0ef9b Mon Sep 17 00:00:00 2001
From: Daniel Salazar <daniel.salazar@puter.com>
Date: Sat, 10 Oct 2026 21:20:11 -0400
Subject: [PATCH] fix(ddb): settle in-flight batch chunks before rethrowing

A failed chunk stopped unstarted chunks but rejected while others were
still retrying, so writes could land after the caller had handled the
error and invalidated its cache.
---
 .../clients/dynamodb/DDBClient.test.ts        | 36 +++++++++++++++++++
 src/backend/clients/dynamodb/DDBClient.ts     | 13 +++++--
 2 files changed, 46 insertions(+), 3 deletions(-)

diff --git a/src/backend/clients/dynamodb/DDBClient.test.ts b/src/backend/clients/dynamodb/DDBClient.test.ts
index c5c560934..c1c522b59 100644
--- a/src/backend/clients/dynamodb/DDBClient.test.ts
+++ b/src/backend/clients/dynamodb/DDBClient.test.ts
@@ -871,6 +871,42 @@ describe('DDBClient — batch write chunks and TTL sweep', () => {
         expect(calls.length).toBeLessThanOrEqual(4);
     });
 
+    it('lets chunks already in flight finish before rejecting', async () => {
+        let replies = 0;
+        handlers = {
+            BatchWriteItem: (body) => {
+                replies += 1;
+                if (replies === 1) {
+                    return {
+                        status: 400,
+                        __type: 'com.amazon.coral.validate#ValidationException',
+                        message: 'bad item',
+                    };
+                }
+                // The rest need one retry, so they are still writing when the
+                // failure comes back.
+                return replies <= 4
+                    ? { UnprocessedItems: body.RequestItems }
+                    : { UnprocessedItems: {} };
+            },
+        };
+
+        await expect(
+            stubClient().batchPut(
+                Array.from({ length: 100 }, (_, i) => ({
+                    table: TABLE,
+                    item: { pk: `item-${i}` },
+                })),
+            ),
+        ).rejects.toThrow('bad item');
+        const sentBeforeReject = calls.length;
+        await new Promise((resolve) => setTimeout(resolve, 200));
+
+        // Nothing is written after the caller has seen the error.
+        expect(calls.length).toBe(sentBeforeReject);
+        expect(inFlight).toBe(0);
+    });
+
     it('retries the unprocessed deletes of the TTL sweep', async () => {
         const expired = [{ pk: { S: 'old' } }];
         let batchWrites = 0;
diff --git a/src/backend/clients/dynamodb/DDBClient.ts b/src/backend/clients/dynamodb/DDBClient.ts
index 560ad761b..50e3cad7b 100644
--- a/src/backend/clients/dynamodb/DDBClient.ts
+++ b/src/backend/clients/dynamodb/DDBClient.ts
@@ -44,7 +44,7 @@ import { Agent as httpsAgent } from 'node:https';
 import { HttpError } from '../../core/http';
 import type { IConfig, IDynamoConfig } from '../../types';
 import { BoundedTtlMap } from '../../util/boundedTtlMap.js';
-import { runWithConcurrencyLimit } from '../../util/concurrency.js';
+import { runWithConcurrencyLimitSettled } from '../../util/concurrency.js';
 import { Span } from '../../util/span.js';
 import { PuterClient } from '../types';
 import {
@@ -367,10 +367,12 @@ export class DDBClient extends PuterClient {
 
         const client = await this.#getDocumentClient();
         const capacity = new ConsumedCapacityTally();
-        // Once a chunk fails, the rest aren't started.
+        // Once a chunk fails, the rest aren't started. Those in flight settle
+        // before the error is thrown, so no write lands after the caller's
+        // cleanup.
         let failed = false;
 
-        await runWithConcurrencyLimit(
+        const outcomes = await runWithConcurrencyLimitSettled(
             chunkValues(params, MAX_BATCH_WRITE_ITEMS),
             BATCH_WRITE_CONCURRENCY,
             async (chunk) => {
@@ -414,6 +416,11 @@ export class DDBClient extends PuterClient {
                 }
             },
         );
+        const rejected = outcomes.find(
+            (outcome): outcome is PromiseRejectedResult =>
+                outcome.status === 'rejected',
+        );
+        if (rejected) throw rejected.reason;
 
         return { ConsumedCapacity: capacity.entries() };
     }