diff --git a/src/backend/clients/dynamodb/DDBClient.test.ts b/src/backend/clients/dynamodb/DDBClient.test.ts index b8faf0470..864862ee0 100644 --- a/src/backend/clients/dynamodb/DDBClient.test.ts +++ b/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 . + * 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', () => { diff --git a/src/backend/clients/dynamodb/DDBClient.ts b/src/backend/clients/dynamodb/DDBClient.ts index f5734d164..1b8e4d2aa 100644 --- a/src/backend/clients/dynamodb/DDBClient.ts +++ b/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 }[]) { + 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 }[]) { + 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(); 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; }, diff --git a/src/backend/stores/systemKv/SystemKVStore.test.ts b/src/backend/stores/systemKv/SystemKVStore.test.ts index 758d5d62b..3719c7f79 100644 --- a/src/backend/stores/systemKv/SystemKVStore.test.ts +++ b/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( diff --git a/src/backend/stores/systemKv/SystemKVStore.ts b/src/backend/stores/systemKv/SystemKVStore.ts index a3cca8e37..360fcaa1e 100644 --- a/src/backend/stores/systemKv/SystemKVStore.ts +++ b/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> { + if (!Array.isArray(keys) || keys.length === 0) { + return { res: true, usage: emptyUsage() }; + } + + const unique = new Set(); + 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 | undefined; + | Record + | undefined; if (!startKey) { exhausted = remaining > 0; break; @@ -1076,7 +1129,8 @@ export class SystemKVStore extends PuterStore { >), ); nextKey = response.LastEvaluatedKey as - Record | undefined; + | Record + | 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 | undefined; + | Record + | 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, ), ), }; diff --git a/src/gui/src/UI/Dashboard/usageBudget.js b/src/gui/src/UI/Dashboard/usageBudget.js index d9cb26eea..f858f579e 100644 --- a/src/gui/src/UI/Dashboard/usageBudget.js +++ b/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 diff --git a/src/gui/src/UI/Dashboard/usageBudget.test.js b/src/gui/src/UI/Dashboard/usageBudget.test.js index ec77c3bee..68de6e40e 100644 --- a/src/gui/src/UI/Dashboard/usageBudget.test.js +++ b/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.