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.

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;

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() };
     }