feat: batch delete for flush + metering fixes (#3593)

src/backend/clients/dynamodb/DDBClient.test.ts

@@ -3,18 +3,19 @@
  *
  * This file is part of Puter.
  *
- * Puter is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as published
- * by the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
+ * Puter is free software: you can redistribute it and/or modify it under the
+ * terms of the GNU Affero General Public License as published by the Free
+ * Software Foundation, either version 3 of the License, or (at your option) any
+ * later version.
  *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU Affero General Public License for more details.
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+ * FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
+ * details.
  *
  * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ * along with this program. If not, see
+ * [https://www.gnu.org/licenses/](https://www.gnu.org/licenses/).
  */
 
 import { createServer, type Server } from 'node:http';
@@ -302,6 +303,35 @@ describe('DDBClient — batch writes', () => {
         const stored = await client.query(TABLE, { pk: 'bulk' });
         expect(stored.Items).toHaveLength(60);
     });
+
+    it('reports no consumed capacity for an empty delete batch', async () => {
+        await expect(client.batchDel([])).resolves.toEqual({
+            ConsumedCapacity: [],
+        });
+    });
+
+    it('batch-deletes across chunks, leaving unrelated items alone', async () => {
+        const items = Array.from({ length: 30 }, (_, index) => ({
+            table: TABLE,
+            item: { pk: 'bulk-del', sk: `item-${index}`, index },
+        }));
+        await client.batchPut(items);
+
+        // Delete all but the last item — 29 keys still exercises chunking
+        // paths shared with batchPut while proving deletion is targeted.
+        const result = await client.batchDel(
+            items.slice(0, -1).map(({ table, item }) => ({
+                table,
+                key: { pk: item.pk, sk: item.sk },
+            })),
+        );
+
+        expect(result.ConsumedCapacity).toHaveLength(1);
+        expect(result.ConsumedCapacity[0].TableName).toBe(TABLE);
+
+        const stored = await client.query(TABLE, { pk: 'bulk-del' });
+        expect(stored.Items?.map((item) => item.sk)).toEqual(['item-29']);
+    });
 });
 
 describe('DDBClient — expired item sweep', () => {

src/backend/clients/dynamodb/DDBClient.ts

@@ -222,6 +222,37 @@ export class DDBClient extends PuterClient {
         'db.batch_size': params.length,
     }))
     async batchPut(params: { table: string; item: Record<string, unknown> }[]) {
+        return this.#batchWrite(
+            params.map(({ table, item }) => ({
+                table,
+                request: { PutRequest: { Item: item } },
+            })),
+        );
+    }
+
+    @Span('ddb.batchDel', (params: unknown[]) => ({
+        'db.batch_size': params.length,
+    }))
+    async batchDel(params: { table: string; key: Record<string, unknown> }[]) {
+        return this.#batchWrite(
+            params.map(({ table, key }) => ({
+                table,
+                request: { DeleteRequest: { Key: key } },
+            })),
+        );
+    }
+
+    // Shared BatchWriteItem plumbing for batchPut/batchDel: 25-item chunks,
+    // UnprocessedItems retried with capped exponential backoff, consumed
+    // capacity accumulated per table across every request.
+    async #batchWrite(
+        params: {
+            table: string;
+            request: NonNullable<
+                BatchWriteCommandInput['RequestItems']
+            >[string][number];
+        }[],
+    ) {
         const consumedCapacityByTable = new Map<string, number>();
         if (params.length === 0) {
             return { ConsumedCapacity: [] };
@@ -258,11 +289,7 @@ export class DDBClient extends PuterClient {
             let requestItems = chunk.reduce(
                 (acc, curr) => {
                     const tableRequests = acc[curr.table] ?? [];
-                    tableRequests.push({
-                        PutRequest: {
-                            Item: curr.item,
-                        },
-                    });
+                    tableRequests.push(curr.request);
                     acc[curr.table] = tableRequests;
                     return acc;
                 },

src/backend/stores/systemKv/SystemKVStore.test.ts

@@ -232,6 +232,48 @@ describe('SystemKVStore', () => {
         });
     });
 
+    describe('batchDel', () => {
+        it('removes every key in the batch and leaves the rest', async () => {
+            await target.batchPut(
+                {
+                    items: [
+                        { key: 'bd1', value: 'v1' },
+                        { key: 'bd2', value: 'v2' },
+                        { key: 'bd3', value: 'v3' },
+                    ],
+                },
+                opts,
+            );
+            await target.batchDel({ keys: ['bd1', 'bd3'] }, opts);
+            const result = await target.get(
+                { key: ['bd1', 'bd2', 'bd3'] },
+                opts,
+            );
+            expect(result.res).toEqual([null, 'v2', null]);
+        });
+
+        it('is a no-op for an empty keys array', async () => {
+            const result = await target.batchDel({ keys: [] }, opts);
+            expect(result.res).toBe(true);
+        });
+
+        it('tolerates missing keys and duplicates in the batch', async () => {
+            await target.set({ key: 'bd-only', value: 'v' }, opts);
+            const result = await target.batchDel(
+                { keys: ['bd-only', 'bd-only', 'never-existed'] },
+                opts,
+            );
+            expect(result.res).toBe(true);
+            expect((await target.get({ key: 'bd-only' }, opts)).res).toBeNull();
+        });
+
+        it('rejects when any key is oversized', async () => {
+            await expect(
+                target.batchDel({ keys: ['ok', 'a'.repeat(1025)] }, opts),
+            ).rejects.toMatchObject({ statusCode: 400 });
+        });
+    });
+
     describe('list', () => {
         beforeEach(async () => {
             await target.batchPut(

src/backend/stores/systemKv/SystemKVStore.ts

@@ -759,7 +759,8 @@ export class SystemKVStore extends PuterStore {
                 fetched = response.Item ? [response.Item as KvCachedItem] : [];
                 fetchUnits = Number(
                     (response.ConsumedCapacity?.CapacityUnits as
-                        number | undefined) ?? 0,
+                        | number
+                        | undefined) ?? 0,
                 );
             }
 
@@ -831,7 +832,8 @@ export class SystemKVStore extends PuterStore {
                 probeUsage,
                 writeUsage(
                     response.ConsumedCapacity?.CapacityUnits as
-                        number | undefined,
+                        | number
+                        | undefined,
                 ),
             ),
         };
@@ -925,12 +927,59 @@ export class SystemKVStore extends PuterStore {
                 probeUsage,
                 writeUsage(
                     (response.ConsumedCapacity?.CapacityUnits as
-                        number | undefined) ?? 1,
+                        | number
+                        | undefined) ?? 1,
                 ),
             ),
         };
     }
 
+    async batchDel(
+        { keys }: { keys: string[] },
+        opts?: KVOpts,
+    ): Promise<KVResult<boolean>> {
+        if (!Array.isArray(keys) || keys.length === 0) {
+            return { res: true, usage: emptyUsage() };
+        }
+
+        const unique = new Set<string>();
+        for (const key of keys) {
+            const k = String(key);
+            assertKey(k);
+            unique.add(k);
+        }
+        const uniqueKeys = [...unique];
+
+        const actor = ensureActor(opts);
+        const namespace = getNamespace(actor, opts);
+
+        // One private key refuses the batch — same posture as batchPut:
+        // no partial success to probe with.
+        const probeUsage = await this.#assertNonePrivate(
+            namespace,
+            uniqueKeys,
+            opts,
+        );
+
+        const response = await this.clients.dynamo.batchDel(
+            uniqueKeys.map((key) => ({
+                table: this.tableName,
+                key: { namespace, key },
+            })),
+        );
+        await this.#invalidate(namespace, uniqueKeys);
+        const units =
+            response.ConsumedCapacity?.reduce(
+                (acc, curr) => acc + Number(curr.CapacityUnits ?? 0),
+                0,
+            ) ?? unique.size;
+
+        return {
+            res: true,
+            usage: addUsage(probeUsage, writeUsage(units || unique.size)),
+        };
+    }
+
     async list(
         {
             as,
@@ -957,7 +1006,9 @@ export class SystemKVStore extends PuterStore {
             | { key: string; value: unknown }[]
             | {
                   items:
-                      string[] | unknown[] | { key: string; value: unknown }[];
+                      | string[]
+                      | unknown[]
+                      | { key: string; value: unknown }[];
                   cursor?: string;
                   total?: number;
               }
@@ -1036,7 +1087,8 @@ export class SystemKVStore extends PuterStore {
                 usage,
                 readUsage(
                     (response.ConsumedCapacity?.CapacityUnits as
-                        number | undefined) ?? 1,
+                        | number
+                        | undefined) ?? 1,
                 ),
             );
             return response;
@@ -1053,7 +1105,8 @@ export class SystemKVStore extends PuterStore {
                 const skip = await runQuery(remaining, startKey, 'COUNT');
                 remaining -= Number(skip.Count ?? 0);
                 startKey = skip.LastEvaluatedKey as
-                    Record<string, unknown> | undefined;
+                    | Record<string, unknown>
+                    | undefined;
                 if (!startKey) {
                     exhausted = remaining > 0;
                     break;
@@ -1076,7 +1129,8 @@ export class SystemKVStore extends PuterStore {
                     >),
                 );
                 nextKey = response.LastEvaluatedKey as
-                    Record<string, unknown> | undefined;
+                    | Record<string, unknown>
+                    | undefined;
                 pages++;
                 if (normalizedLimit === undefined) {
                     // Legacy full listing: follow continuation pages so the
@@ -1112,7 +1166,8 @@ export class SystemKVStore extends PuterStore {
                 const counted = await runQuery(0, countKey, 'COUNT');
                 total += Number(counted.Count ?? 0);
                 countKey = counted.LastEvaluatedKey as
-                    Record<string, unknown> | undefined;
+                    | Record<string, unknown>
+                    | undefined;
             } while (countKey);
         }
 
@@ -1139,26 +1194,29 @@ export class SystemKVStore extends PuterStore {
         );
 
         const entries = response.Items ?? [];
-        const results = (
-            await Promise.all(
-                entries.map(async (entry) => {
-                    try {
-                        return await this.clients.dynamo.del(this.tableName, {
-                            namespace,
-                            key: entry.key,
-                        });
-                    } catch (e) {
-                        console.error('[kv] flush delete failed', entry.key, e);
-                        return null;
-                    }
-                }),
-            )
-        ).filter(Boolean);
-
-        const deleteUnits = results.reduce(
-            (acc, r) => acc + Number(r?.ConsumedCapacity?.CapacityUnits ?? 0),
-            0,
-        );
+        // One BatchWriteItem fan-out (25-item chunks with retries inside the
+        // client) instead of an unbounded Promise.all of single deletes.
+        // Failure posture matches the old per-item loop: log and fall through
+        // to invalidation — a partial flush must still drop cached reads for
+        // every key the query saw.
+        let deleteUnits = 0;
+        if (entries.length > 0) {
+            try {
+                const deleted = await this.clients.dynamo.batchDel(
+                    entries.map((entry) => ({
+                        table: this.tableName,
+                        key: { namespace, key: entry.key },
+                    })),
+                );
+                deleteUnits =
+                    deleted.ConsumedCapacity?.reduce(
+                        (acc, curr) => acc + Number(curr.CapacityUnits ?? 0),
+                        0,
+                    ) ?? 0;
+            } catch (e) {
+                console.error('[kv] flush batch delete failed', e);
+            }
+        }
         usage = addUsage(usage, writeUsage(deleteUnits));
 
         // Exactly the keys the query saw, which is also exactly what was
@@ -1464,7 +1522,8 @@ export class SystemKVStore extends PuterStore {
                     probeUsage,
                     writeUsage(
                         (response.ConsumedCapacity?.CapacityUnits as
-                            number | undefined) ?? 1,
+                            | number
+                            | undefined) ?? 1,
                     ),
                 ),
             };

src/gui/src/UI/Dashboard/usageBudget.js

@@ -48,9 +48,16 @@ export const usageBudget = (usage, allowanceInfo) => {
         ? Math.max(0, allowanceInfo.monthUsageAllowance)
         : 0;
     const total = Number.isFinite(usage?.total) ? Math.max(0, usage.total) : 0;
-    const allowanceUsed = Number.isFinite(usage?.allowanceUsed)
-        ? Math.max(0, usage.allowanceUsed)
-        : Math.min(total, capacity);
+    // Allowance-charged spend is a subset of spend, so a reported value past
+    // the total is corrupt (a raced or repeated server write) — same clamp
+    // the server applies when it computes `remaining`. Without it the two
+    // surfaces disagree: the bar overstates while remaining stays right.
+    const allowanceUsed = Math.min(
+        Number.isFinite(usage?.allowanceUsed)
+            ? Math.max(0, usage.allowanceUsed)
+            : Math.min(total, capacity),
+        total,
+    );
     const addons = allowanceInfo?.addons ?? {};
     const purchased = Number.isFinite(addons.purchasedCredits)
         ? addons.purchasedCredits

src/gui/src/UI/Dashboard/usageBudget.test.js

@@ -73,6 +73,22 @@ describe('usageBudget', () => {
         expect(budget.percent).toBe(10);
     });
 
+    it('never trusts a reported allowanceUsed past the month total', () => {
+        // A corrupt record: the split grew past the spend it splits (a raced
+        // server write). The bar reads the total — the same clamp the server
+        // applies to remaining — instead of overstating past 100%.
+        const budget = usageBudget(
+            { total: 1_840_000_000, allowanceUsed: 20_722_300_000 },
+            info(19_000_000_000, {
+                purchasedCredits: 40_000_000_000,
+                consumedPurchaseCredits: 40_000_000_000,
+            }),
+        );
+        expect(budget.used).toBe(1_840_000_000);
+        expect(budget.percent).toBe(10);
+        expect(budget.barPercent).toBeCloseTo(9.68, 1);
+    });
+
     it('falls back to the capped total for records without the split', () => {
         // Legacy record: no allowanceUsed. Everything up to the allowance
         // counts, and an overshot total still reads as a full plan.