From 95da802b13793e69f41124ced5f7c905f4655a45 Mon Sep 17 00:00:00 2001 From: Daniel Salazar Date: Sat, 8 Aug 2026 06:10:01 -0700 Subject: [PATCH] fix: metering hardening (#3527) --- .../metering/MeteringBufferStore.test.ts | 26 ++++++++++ .../stores/metering/MeteringBufferStore.ts | 47 +++++++++++++++---- 2 files changed, 65 insertions(+), 8 deletions(-) diff --git a/src/backend/stores/metering/MeteringBufferStore.test.ts b/src/backend/stores/metering/MeteringBufferStore.test.ts index 9f63aa8ad..18369dbbd 100644 --- a/src/backend/stores/metering/MeteringBufferStore.test.ts +++ b/src/backend/stores/metering/MeteringBufferStore.test.ts @@ -492,6 +492,32 @@ describe('MeteringBufferStore', () => { ).toEqual({}); }); + it('re-drives an abandoned claim through exactly one of two concurrent flushes', async () => { + const tag = bucketTag(key); + const nonce = 'abandonednonce'; + + await server.clients.redis.hset( + `meter:p:{${tag}}:${nonce}`, + 'total', + '25', + ); + await server.clients.redis.hset( + `meter:pending:{${tag}}`, + nonce, + `${Date.now() - 60_000}:${key}`, + ); + + // The pending index is read without removing anything, so both + // cycles see this claim. Only one of them may write it onward — + // usage counted twice here would over-bill. + await Promise.all([target.flushCycle(), target.flushCycle()]); + + expect(await storedTotal(key)).toBe(25); + expect( + await server.clients.redis.hgetall(`meter:pending:{${tag}}`), + ).toEqual({}); + }); + it('leaves a claim that is still fresh alone', async () => { const tag = bucketTag(key); await server.clients.redis.hset( diff --git a/src/backend/stores/metering/MeteringBufferStore.ts b/src/backend/stores/metering/MeteringBufferStore.ts index b87896e72..32fdfcf63 100644 --- a/src/backend/stores/metering/MeteringBufferStore.ts +++ b/src/backend/stores/metering/MeteringBufferStore.ts @@ -191,6 +191,26 @@ redis.call('PEXPIRE', KEYS[3], ARGV[3]) return redis.call('HGETALL', KEYS[2]) `; +/** + * KEYS: pending (abandoned), pending (fresh), pending index. ARGV: abandoned + * nonce, fresh nonce, fresh index value, ttl. + * + * Re-claims an abandoned claim under a new nonce. The pending index is read + * without removing anything, so every deployment sees every abandoned claim at + * once; this rename is what stops more than one of them from writing the same + * amounts onward. Re-stamping the claim time also restarts the clock, so a + * deployment that dies holding the re-claim doesn't have it swept instantly. + */ +const RECLAIM_SCRIPT = ` +if redis.call('EXISTS', KEYS[1]) == 0 then return nil end +redis.call('RENAME', KEYS[1], KEYS[2]) +redis.call('PEXPIRE', KEYS[2], ARGV[4]) +redis.call('HDEL', KEYS[3], ARGV[1]) +redis.call('HSET', KEYS[3], ARGV[2], ARGV[3]) +redis.call('PEXPIRE', KEYS[3], ARGV[4]) +return redis.call('HGETALL', KEYS[2]) +`; + /** * KEYS: base, pending, pending index. ARGV: nonce, ttl, new total (or ''), then * the authoritative path/value pairs. @@ -233,6 +253,7 @@ type ScriptRunner = { meterIncr(...args: string[]): Promise<[string[], string[]]>; meterRead(...args: string[]): Promise<[string[], string[]]>; meterClaim(...args: string[]): Promise; + meterReclaim(...args: string[]): Promise; meterSettle(...args: string[]): Promise; meterSeed(...args: string[]): Promise; }; @@ -398,6 +419,10 @@ export class MeteringBufferStore extends PuterStore { numberOfKeys: 3, lua: CLAIM_SCRIPT, }); + client.defineCommand('meterReclaim', { + numberOfKeys: 3, + lua: RECLAIM_SCRIPT, + }); client.defineCommand('meterSettle', { numberOfKeys: 3, lua: SETTLE_SCRIPT, @@ -609,19 +634,25 @@ export class MeteringBufferStore extends PuterStore { tag: string, orphan: { nonce: string; key: string }, ): Promise { - const pairs = await this.clients.redis.hgetall( + const nonce = randomUUID().replace(/-/g, ''); + const reclaimed = await this.#redis.meterReclaim( pendingKey(tag, orphan.nonce), + pendingKey(tag, nonce), + pendingIndexKey(tag), + orphan.nonce, + nonce, + encodePendingEntry(Date.now(), orphan.key), + String(BUFFER_TTL_MS), ); - const amounts: FlatAmounts = {}; - for (const [path, amount] of Object.entries(pairs ?? {})) { - amounts[path] = Number(amount); - } - if (Object.keys(amounts).length === 0) { - // The claimed data is gone but its index entry outlived it. + if (!reclaimed) { + // Either another flush re-claimed this first, or the claimed data + // is gone and only its index entry outlived it. Dropping the entry + // covers the second case and is a no-op for the first, which has + // already re-keyed it. await this.clients.redis.hdel(pendingIndexKey(tag), orphan.nonce); return; } - await this.#settle(tag, orphan.key, orphan.nonce, amounts); + await this.#settle(tag, orphan.key, nonce, pairsToAmounts(reclaimed)); } async #settle(