From 78cfe0c0f69e1ea935910d35d860dd4c73f9ff3b Mon Sep 17 00:00:00 2001 From: Daniel Salazar Date: Wed, 12 Aug 2026 15:55:03 -0700 Subject: [PATCH] metrics: add metering buffer store coutns (#3554) --- .../metering/MeteringBufferStore.test.ts | 230 +++++++++++- .../stores/metering/MeteringBufferStore.ts | 330 ++++++++++++++++-- 2 files changed, 532 insertions(+), 28 deletions(-) diff --git a/src/backend/stores/metering/MeteringBufferStore.test.ts b/src/backend/stores/metering/MeteringBufferStore.test.ts index 4cfc36a41..dfc42f78d 100644 --- a/src/backend/stores/metering/MeteringBufferStore.test.ts +++ b/src/backend/stores/metering/MeteringBufferStore.test.ts @@ -663,19 +663,142 @@ describe('MeteringBufferStore', () => { expect(await storedTotal(key)).toBe(0); }); - it('drops an index entry whose data is gone', async () => { + it('raises the alarm for an index entry whose data is gone', async () => { + // The claim's amounts left the cache and only the entry pointing at + // them survived. Nothing downstream will ever notice the gap, so + // dropping the entry quietly is the one thing this must not do. const tag = bucketTag(key); await server.clients.redis.hset( `meter:pending:{${tag}}`, 'lostnonce', `${Date.now() - 60_000}:${key}`, ); + const logged = vi + .spyOn(console, 'error') + .mockImplementation(() => {}); + const alarmSpy = vi.spyOn(server.clients.alarm, 'create'); await target.flushCycle(); expect( await server.clients.redis.hgetall(`meter:pending:{${tag}}`), ).toEqual({}); + expect(logged).toHaveBeenCalledWith( + expect.stringContaining('lost its amounts'), + ); + expect(alarmSpy).toHaveBeenCalledWith( + `metering_claim_lost:${key}`, + expect.stringContaining('under-billed'), + { key }, + 'critical', + expect.objectContaining({ dedup: true }), + ); + logged.mockRestore(); + alarmSpy.mockRestore(); + }); + + it('stays quiet when another flush re-keyed the claim first', async () => { + // Both cycles see the same abandoned claim; the one that loses the + // rename finds nothing where the claim used to be. That is the + // ordinary outcome of the race, not usage lost, and calling it loss + // would make the alarm fire on every deploy. + const tag = bucketTag(key); + await server.clients.redis.hset( + `meter:p:{${tag}}:racednonce`, + 'total', + '25', + ); + await server.clients.redis.hset( + `meter:pending:{${tag}}`, + 'racednonce', + `${Date.now() - 60_000}:${key}`, + ); + const alarmSpy = vi.spyOn(server.clients.alarm, 'create'); + + await Promise.all([target.flushCycle(), target.flushCycle()]); + + expect(await storedTotal(key)).toBe(25); + expect(alarmSpy).not.toHaveBeenCalledWith( + `metering_claim_lost:${key}`, + expect.anything(), + expect.anything(), + expect.anything(), + expect.anything(), + ); + alarmSpy.mockRestore(); + }); + + it('retires an abandoned claim whose amounts all landed, quietly', async () => { + // Every path was written onward and taken off the claim, and then + // the deployment went away before it could retire it. Nothing was + // lost here, so the sweep has to finish the bookkeeping without + // reporting loss — and without settling, which would replace a good + // base with this claim's stale view. + const tag = bucketTag(key); + await target.incr({ key, pathAndAmountMap: { total: 40 } }); + await target.flushCycle(); + + await server.clients.redis.hset( + `meter:p:{${tag}}:appliednonce`, + '__claim', + String(Date.now() - 60_000), + ); + await server.clients.redis.hset( + `meter:pending:{${tag}}`, + 'appliednonce', + `${Date.now() - 60_000}:${key}`, + ); + const alarmSpy = vi.spyOn(server.clients.alarm, 'create'); + + await target.flushCycle(); + + expect(alarmSpy).not.toHaveBeenCalled(); + expect(await storedTotal(key)).toBe(40); + expect( + await server.clients.redis.hgetall(`meter:pending:{${tag}}`), + ).toEqual({}); + expect( + await server.clients.redis.hget( + `meter:b:{${tag}}:${key}`, + 'total', + ), + ).toBe('40'); + alarmSpy.mockRestore(); + }); + + it('keeps a claim addressable once its last amount is applied', async () => { + // Amounts come off a claim as they land, and a hash with nothing + // left in it stops existing — which would make a finished claim + // indistinguishable from one whose amounts were lost. + await target.incr({ key, pathAndAmountMap: { total: 5 } }); + + const tag = bucketTag(key); + const settle = vi + .spyOn( + server.clients.redis as unknown as { + meterSettle: () => Promise; + }, + 'meterSettle', + ) + .mockImplementation(async () => { + // Mid-settle: every amount has been applied and taken off + // the claim, but the claim itself has not been retired. + const pending = await server.clients.redis.hgetall( + `meter:pending:{${tag}}`, + ); + const nonce = Object.keys(pending)[0]!; + expect( + await server.clients.redis.exists( + `meter:p:{${tag}}:${nonce}`, + ), + ).toBe(1); + return 1; + }); + + await target.flushCycle(); + settle.mockRestore(); + + expect(await storedTotal(key)).toBe(5); }); it('gives up on a path the store will never accept', async () => { @@ -999,6 +1122,111 @@ describe('MeteringBufferStore', () => { }); }); + describe('reconciliation', () => { + it('puts back a delta that fell off its dirty set', async () => { + await target.incr({ key, pathAndAmountMap: { total: 17 } }); + + // The delta survives but the working list forgot it — a failover, + // or a set that went while the counter it pointed at stayed. From + // here nothing would ever flush this: reads keep answering from it + // right up until the TTL takes the amounts with it. + const tag = bucketTag(key); + await server.clients.redis.del(`meter:dirty:{${tag}}`); + expect(await target.flushCycle()).toBe(0); + expect(await storedTotal(key)).toBe(0); + + const logged = vi + .spyOn(console, 'error') + .mockImplementation(() => {}); + await target.reconcile(); + expect(logged).toHaveBeenCalledWith( + expect.stringContaining('recovered 1 buffered counter'), + ); + logged.mockRestore(); + + await target.flushCycle(); + expect(await storedTotal(key)).toBe(17); + }); + + it('leaves a counter that is only between the drain and its claim', async () => { + await target.incr({ key, pathAndAmountMap: { total: 8 } }); + const tag = bucketTag(key); + + // Taken off the dirty set and not yet claimed. It is not lost — the + // in-flight record is holding it — and putting it back here would + // hand the same delta to two flushes at once. + await server.clients.redis.del(`meter:dirty:{${tag}}`); + await server.clients.redis.zadd( + `meter:inflight:{${tag}}`, + String(Date.now()), + key, + ); + + await target.reconcile(); + + expect( + await server.clients.redis.scard(`meter:dirty:{${tag}}`), + ).toBe(0); + }); + + it('retires what it tracks once a counter has settled', async () => { + await target.incr({ key, pathAndAmountMap: { total: 3 } }); + const tag = bucketTag(key); + expect( + await server.clients.redis.smembers(`meter:tracked:{${tag}}`), + ).toEqual([key]); + + await target.flushCycle(); + + // Otherwise every counter the bucket ever saw stays on the list and + // each sweep costs more than the last. + expect( + await server.clients.redis.smembers(`meter:tracked:{${tag}}`), + ).toEqual([]); + }); + + it('keeps tracking a counter that took on more mid-flush', async () => { + await target.incr({ key, pathAndAmountMap: { total: 3 } }); + const tag = bucketTag(key); + + // An increment landing between the claim and the settle starts a + // fresh delta. Retiring the entry on the settle that is finishing + // would leave that delta with nothing pointing at it. + const settle = server.clients.redis as unknown as { + meterSettle: (...args: string[]) => Promise; + }; + const real = settle.meterSettle.bind(settle); + const racing = vi + .spyOn(settle, 'meterSettle') + .mockImplementation(async (...args: string[]) => { + await target.incr({ key, pathAndAmountMap: { total: 4 } }); + return real(...args); + }); + + await target.flushCycle(); + racing.mockRestore(); + + expect( + await server.clients.redis.smembers(`meter:tracked:{${tag}}`), + ).toEqual([key]); + + await target.flushCycle(); + expect(await storedTotal(key)).toBe(7); + }); + + it('reports nothing when every bucket is in order', async () => { + await target.incr({ key, pathAndAmountMap: { total: 1 } }); + const logged = vi + .spyOn(console, 'error') + .mockImplementation(() => {}); + + await target.reconcile(); + + expect(logged).not.toHaveBeenCalled(); + logged.mockRestore(); + }); + }); + describe('month boundaries', () => { it('keeps the counter for each month independent', async () => { const august = `${key}-2026-08`; diff --git a/src/backend/stores/metering/MeteringBufferStore.ts b/src/backend/stores/metering/MeteringBufferStore.ts index d3d73196e..72e60728e 100644 --- a/src/backend/stores/metering/MeteringBufferStore.ts +++ b/src/backend/stores/metering/MeteringBufferStore.ts @@ -18,6 +18,7 @@ */ import { randomUUID } from 'node:crypto'; +import { metrics } from '@opentelemetry/api'; import murmurhash from 'murmurhash'; import { chunkPathsForIncr, @@ -25,6 +26,42 @@ import { } from '../systemKv/SystemKVStore'; import { PuterStore } from '../types'; +// -- Metrics ---------------------------------------------------------- + +/** + * Counters rather than log lines, because the interesting numbers here are ones + * nothing downstream notices: amounts given up on, and claims whose amounts + * left the cache without being written. A log line can be read after the fact; + * only a metric can be alarmed on. + * + * Deliberately undimensioned — the natural dimensions (counter key, actor) are + * unbounded, and each distinct combination is its own CloudWatch series. + */ +const meter = metrics.getMeter('puter-backend'); + +const absorbedCounter = meter.createCounter('metering.buffer.absorbed', { + description: 'Metering increments taken into the cache buffer', +}); + +const settledCounter = meter.createCounter('metering.buffer.settled', { + description: 'Buffered counters written onward to the KV store', +}); + +const droppedPathsCounter = meter.createCounter('metering.buffer.dropped', { + description: + 'Counter paths the KV store would not accept, given up on and lost', +}); + +const lostClaimsCounter = meter.createCounter('metering.buffer.lost_claims', { + description: + 'Claims whose amounts left the cache before they were written onward', +}); + +const recoveredCounter = meter.createCounter('metering.buffer.recovered', { + description: + 'Buffered counters the reconciliation sweep put back on a dirty set', +}); + // -- Types ------------------------------------------------------------ /** Matches the operation shape of `SystemKVStore.incr`. */ @@ -77,6 +114,13 @@ const SETTLE_CONCURRENCY = 20; /** Roughly a minute between compression reports, so they don't flood the log. */ const CYCLES_PER_COMPRESSION_REPORT = 12; +/** + * Roughly a minute between reconciliation sweeps. Cheap enough to run more + * often, but what it recovers has already waited on whatever lost it, so a + * minute of extra delay changes nothing. + */ +const CYCLES_PER_RECONCILE = 12; + /** * How many path names this deployment remembers as unwritable. Bounded because * the names come from usage types, and a caller can invent those; forgetting @@ -98,6 +142,29 @@ const pendingKey = (tag: string, nonce: string): string => const pendingIndexKey = (tag: string): string => `meter:pending:{${tag}}`; /** Counters taken off the dirty set but not yet claimed, scored by when. */ const inflightKey = (tag: string): string => `meter:inflight:{${tag}}`; +/** + * Every counter this bucket holds amounts for, whether or not anything is + * currently pointing at it. + * + * The dirty set is what a flush works from, so a counter missing from it is a + * counter nothing will ever flush: its delta sits there until the TTL takes it, + * while reads keep answering from it as though it were fine. That is the one + * way amounts leave without anything saying so. This set is the second record + * the sweep compares against, so a delta can be found again by what it is + * rather than by a pointer that may not have survived. + */ +const trackedKey = (tag: string): string => `meter:tracked:{${tag}}`; + +/** + * Bookkeeping field kept on a claim for as long as the claim exists. + * + * Amounts come off a claim as they are applied, and a hash with no fields left + * is a hash that does not exist — so without this, a claim that finished and a + * claim whose amounts were lost look identical from the outside, and the sweep + * cannot tell which of them it is looking at. With it, the claim's key is + * present exactly while the claim is, and a missing one means real loss. + */ +const CLAIM_FIELD = '__claim'; // -- Shape helpers ---------------------------------------------------- @@ -110,6 +177,13 @@ export const pairsToAmounts = (pairs: string[]): FlatAmounts => { return out; }; +/** A claim's amounts, without the bookkeeping field every claim carries. */ +export const claimAmounts = (pairs: string[]): FlatAmounts => { + const amounts = pairsToAmounts(pairs); + delete amounts[CLAIM_FIELD]; + return amounts; +}; + /** * Cache hashes are flat, but stored counters nest on `.` the same way `incr` * paths do — so `{'a.units': 1}` becomes `{a: {units: 1}}`. @@ -205,10 +279,16 @@ export const isBilledCounter = (key: string): boolean => // -- Scripts ---------------------------------------------------------- /** - * KEYS: delta, base, dirty set. ARGV: ttl, member, then path/amount pairs. + * KEYS: delta, base, dirty set, tracked set. ARGV: ttl, member, then + * path/amount pairs. * * Returns the delta and the base together so a running total can be served * without a second round trip. Amounts are floats — costs are fractional. + * + * The counter joins both sets here. The dirty set is the working list a flush + * takes from and hands back; the tracked set is only ever added to here and + * removed from once the amounts are gone, so it stays true even when the + * working list does not. */ const INCR_SCRIPT = ` for i = 3, #ARGV, 2 do @@ -217,6 +297,8 @@ end redis.call('PEXPIRE', KEYS[1], ARGV[1]) redis.call('SADD', KEYS[3], ARGV[2]) redis.call('PEXPIRE', KEYS[3], ARGV[1]) +redis.call('SADD', KEYS[4], ARGV[2]) +redis.call('PEXPIRE', KEYS[4], ARGV[1]) return { redis.call('HGETALL', KEYS[1]), redis.call('HGETALL', KEYS[2]) } `; @@ -255,7 +337,7 @@ return { taken, redis.call('SCARD', KEYS[1]) } /** * KEYS: delta, pending, pending index, in-flight set. ARGV: nonce, index value, - * ttl, in-flight member. + * ttl, in-flight member, claim time. * * The rename is the claim, and it is atomic: if two flushes race for one delta, * one takes all of it and the other sees nothing. Increments arriving mid-flush @@ -271,6 +353,7 @@ const CLAIM_SCRIPT = ` redis.call('ZREM', KEYS[4], ARGV[4]) if redis.call('EXISTS', KEYS[1]) == 0 then return nil end redis.call('RENAME', KEYS[1], KEYS[2]) +redis.call('HSET', KEYS[2], '${CLAIM_FIELD}', ARGV[5]) redis.call('PEXPIRE', KEYS[2], ARGV[3]) redis.call('HSET', KEYS[3], ARGV[1], ARGV[2]) redis.call('PEXPIRE', KEYS[3], ARGV[3]) @@ -279,32 +362,47 @@ return redis.call('HGETALL', KEYS[2]) /** * KEYS: pending (abandoned), pending (fresh), pending index. ARGV: abandoned - * nonce, fresh nonce, fresh index value, ttl. + * nonce, fresh nonce, fresh index value, ttl, claim time. * * 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. + * + * Returns `{1, amounts}` when the re-claim was taken. When it wasn't, returns + * `{0, had}` where `had` says whether this call is the one that removed the + * index entry — which is the difference between another deployment having + * re-keyed the claim first, and the claim's amounts having left the cache while + * the entry pointing at them stayed. The first is routine; the second is usage + * lost. Only doing the removal here can tell them apart, because between a + * separate read and delete the other deployment could land either way. */ const RECLAIM_SCRIPT = ` -if redis.call('EXISTS', KEYS[1]) == 0 then return nil end +if redis.call('EXISTS', KEYS[1]) == 0 then + return { 0, redis.call('HDEL', KEYS[3], ARGV[1]) } +end redis.call('RENAME', KEYS[1], KEYS[2]) +redis.call('HSET', KEYS[2], '${CLAIM_FIELD}', ARGV[5]) 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]) +return { 1, redis.call('HGETALL', KEYS[2]) } `; /** - * KEYS: base, pending, pending index. ARGV: nonce, ttl, new total (or ''), then - * the authoritative path/value pairs. + * KEYS: base, pending, pending index, delta, tracked set. ARGV: nonce, ttl, new + * total (or ''), member, then the authoritative path/value pairs. * * Replaces the base with what the KV store now holds, which is how the base * picks up other deployments' contributions without a separate read. The total * may not move backwards: two flushes settling out of order would otherwise * briefly under-report, and this total decides whether someone may spend. + * + * The counter leaves the tracked set here, but only if nothing has started a + * fresh delta for it in the meantime — checking and removing in the same step + * is what stops an increment that landed mid-settle from being forgotten. */ const SETTLE_SCRIPT = ` local current = redis.call('HGET', KEYS[1], 'total') @@ -314,16 +412,71 @@ if ARGV[3] ~= '' and current then end if replace then redis.call('DEL', KEYS[1]) - for i = 4, #ARGV, 2 do + for i = 5, #ARGV, 2 do redis.call('HSET', KEYS[1], ARGV[i], ARGV[i + 1]) end redis.call('PEXPIRE', KEYS[1], ARGV[2]) end redis.call('DEL', KEYS[2]) redis.call('HDEL', KEYS[3], ARGV[1]) +if redis.call('EXISTS', KEYS[4]) == 0 then + redis.call('SREM', KEYS[5], ARGV[4]) +end return 1 `; +/** + * KEYS: pending, pending index, delta, tracked set. ARGV: nonce, member. + * + * Forgets a claim without touching the base, for the cases where nothing was + * written onward and replacing the base would throw away a good one. Leaves the + * tracked set on the same terms as a settle. + */ +const RETIRE_SCRIPT = ` +redis.call('DEL', KEYS[1]) +redis.call('HDEL', KEYS[2], ARGV[1]) +if redis.call('EXISTS', KEYS[3]) == 0 then + redis.call('SREM', KEYS[4], ARGV[2]) +end +return 1 +`; + +/** + * KEYS: tracked set, dirty set, in-flight set. ARGV: ttl, delta key prefix. + * + * Puts back anything the working list has lost. A counter still holding a + * delta, but on neither the dirty set nor in flight, is one no flush will ever + * reach: nothing points at it any more, and it will sit there being read from + * until its TTL takes it and the amounts go with it. + * + * A tracked counter with no delta left is the ordinary case — it settled, or it + * is between the claim and the write — and its entry is simply retired, so this + * sweep costs the same next time regardless of how much has passed through the + * bucket. + * + * The delta key is built here rather than declared, which is safe only because + * every key in this scheme carries the bucket as its hash tag and so lives in + * the one slot this script is already running against. + */ +const RECONCILE_SCRIPT = ` +local tracked = redis.call('SMEMBERS', KEYS[1]) +local recovered = 0 +for i = 1, #tracked do + local member = tracked[i] + if redis.call('SISMEMBER', KEYS[2], member) == 0 + and not redis.call('ZSCORE', KEYS[3], member) then + if redis.call('EXISTS', ARGV[2] .. member) == 1 then + redis.call('SADD', KEYS[2], member) + recovered = recovered + 1 + else + redis.call('SREM', KEYS[1], member) + end + end +end +if recovered > 0 then redis.call('PEXPIRE', KEYS[2], ARGV[1]) end +return recovered +`; + /** KEYS: base. ARGV: ttl, then path/value pairs. Returns the base. */ const SEED_SCRIPT = ` if redis.call('EXISTS', KEYS[1]) == 0 then @@ -340,8 +493,11 @@ type ScriptRunner = { meterRead(...args: string[]): Promise<[string[], string[]]>; meterDrain(...args: string[]): Promise<[string[], number]>; meterClaim(...args: string[]): Promise; - meterReclaim(...args: string[]): Promise; + /** `[1, amounts]` when re-claimed, `[0, hadIndexEntry]` when not. */ + meterReclaim(...args: string[]): Promise<[number, string[] | number]>; meterSettle(...args: string[]): Promise; + meterRetire(...args: string[]): Promise; + meterReconcile(...args: string[]): Promise; meterSeed(...args: string[]): Promise; }; @@ -368,6 +524,7 @@ export class MeteringBufferStore extends PuterStore { #flushedCount = 0; #droppedPathCount = 0; #cyclesSinceReport = 0; + #cyclesSinceReconcile = 0; /** Path names the KV store has refused on their own; see `#settle`. */ #unwritablePaths = new Set(); @@ -499,7 +656,7 @@ export class MeteringBufferStore extends PuterStore { this.#definedScripts = true; const client = this.clients.redis; client.defineCommand('meterIncr', { - numberOfKeys: 3, + numberOfKeys: 4, lua: INCR_SCRIPT, }); client.defineCommand('meterRead', { @@ -519,9 +676,17 @@ export class MeteringBufferStore extends PuterStore { lua: RECLAIM_SCRIPT, }); client.defineCommand('meterSettle', { - numberOfKeys: 3, + numberOfKeys: 5, lua: SETTLE_SCRIPT, }); + client.defineCommand('meterRetire', { + numberOfKeys: 4, + lua: RETIRE_SCRIPT, + }); + client.defineCommand('meterReconcile', { + numberOfKeys: 3, + lua: RECONCILE_SCRIPT, + }); client.defineCommand('meterSeed', { numberOfKeys: 1, lua: SEED_SCRIPT, @@ -545,11 +710,13 @@ export class MeteringBufferStore extends PuterStore { deltaKey(tag, key), baseKey(tag, key), dirtyKey(tag), + trackedKey(tag), String(BUFFER_TTL_MS), key, ...toScriptArgs(pathAndAmountMap), ); this.#absorbedCount++; + absorbedCounter.add(1); return { delta: pairsToAmounts(delta), base: pairsToAmounts(base) }; } @@ -618,6 +785,13 @@ export class MeteringBufferStore extends PuterStore { const work: Array<() => Promise> = []; let truncated = 0; + // Before the drain, so anything put back joins this cycle rather than + // waiting for the next one. + if (++this.#cyclesSinceReconcile >= CYCLES_PER_RECONCILE) { + this.#cyclesSinceReconcile = 0; + await this.reconcile(); + } + // Settled, not all-or-nothing: one bucket the cache could not answer // for must not discard the other 63 buckets' work for this cycle. const drained = await Promise.allSettled( @@ -665,10 +839,54 @@ export class MeteringBufferStore extends PuterStore { ); } + settledCounter.add(work.length); this.#reportCompression(work.length); return work.length; } + /** + * Put back anything the dirty sets have lost, across every bucket. + * + * Settled per bucket for the same reason the drain is: a bucket the cache + * could not answer for is one bucket's worth of counters left for the next + * sweep, not a reason to skip the other 63. + */ + async reconcile(): Promise { + const swept = await Promise.allSettled( + Array.from({ length: BUCKET_COUNT }, (_, bucket) => { + const tag = `m${bucket}`; + return this.#redis.meterReconcile( + trackedKey(tag), + dirtyKey(tag), + inflightKey(tag), + String(BUFFER_TTL_MS), + `meter:d:{${tag}}:`, + ); + }), + ); + + let recovered = 0; + for (const outcome of swept) { + if (outcome.status === 'fulfilled') { + recovered += Number(outcome.value); + continue; + } + console.warn( + `[metering] could not reconcile a bucket: ${(outcome.reason as Error)?.message}`, + ); + } + + if (recovered === 0) return; + + recoveredCounter.add(recovered); + // Loud because the amounts were on their way out of the cache with + // nothing left pointing at them: this is the sweep earning its keep, + // and a steady rate of it means something upstream keeps losing them. + console.error( + `[metering] recovered ${recovered} buffered counter(s) that had fallen off their dirty set — usage that would otherwise have expired unwritten`, + ); + } + /** * How many increments each write onward stood in for. This is the number * that says whether buffering is earning anything: a ratio near 1 means @@ -739,20 +957,22 @@ export class MeteringBufferStore extends PuterStore { async #flushOne(tag: string, key: string): Promise { const nonce = randomUUID().replace(/-/g, ''); + const now = Date.now(); const claimed = await this.#redis.meterClaim( deltaKey(tag, key), pendingKey(tag, nonce), pendingIndexKey(tag), inflightKey(tag), nonce, - encodePendingEntry(Date.now(), key), + encodePendingEntry(now, key), String(BUFFER_TTL_MS), key, + String(now), ); // Nothing buffered for this counter — another flush already took it. if (!claimed) return; - await this.#settle(tag, key, nonce, pairsToAmounts(claimed)); + await this.#settle(tag, key, nonce, claimAmounts(claimed)); } async #settleOrphan( @@ -760,24 +980,35 @@ export class MeteringBufferStore extends PuterStore { orphan: { nonce: string; key: string }, ): Promise { const nonce = randomUUID().replace(/-/g, ''); - const reclaimed = await this.#redis.meterReclaim( + const now = Date.now(); + const [taken, payload] = await this.#redis.meterReclaim( pendingKey(tag, orphan.nonce), pendingKey(tag, nonce), pendingIndexKey(tag), orphan.nonce, nonce, - encodePendingEntry(Date.now(), orphan.key), + encodePendingEntry(now, orphan.key), String(BUFFER_TTL_MS), + String(now), ); - 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); + + if (!taken) { + // The claim's amounts are not there. Either another flush re-keyed + // it first — in which case that flush also took the index entry, + // and this removed nothing — or the amounts left the cache while + // the entry pointing at them stayed, which is usage lost. Removing + // the entry is how the two are told apart, so the answer only + // exists here. + if (Number(payload) > 0) this.#recordLostClaim(orphan.key); return; } - await this.#settle(tag, orphan.key, nonce, pairsToAmounts(reclaimed)); + + await this.#settle( + tag, + orphan.key, + nonce, + claimAmounts(payload as string[]), + ); } /** @@ -800,7 +1031,7 @@ export class MeteringBufferStore extends PuterStore { if (Object.keys(amounts).length === 0) { // Nothing to write onward. Retire the claim rather than settling // it, which would clear a base that is still good. - await this.#retireClaim(tag, nonce); + await this.#retireClaim(tag, key, nonce); return; } @@ -872,7 +1103,7 @@ export class MeteringBufferStore extends PuterStore { if (written === 0) { // Nothing reached the store, so there is no newer base to adopt — // settling would replace a good one with this counter's old value. - await this.#retireClaim(tag, nonce); + await this.#retireClaim(tag, key, nonce); return; } @@ -881,9 +1112,12 @@ export class MeteringBufferStore extends PuterStore { baseKey(tag, key), pendingKey(tag, nonce), pendingIndexKey(tag), + deltaKey(tag, key), + trackedKey(tag), nonce, String(BUFFER_TTL_MS), flat['total'] === undefined ? '' : String(flat['total']), + key, ...toScriptArgs(flat), ); } @@ -914,6 +1148,7 @@ export class MeteringBufferStore extends PuterStore { #recordDrop(key: string, dropped: FlatAmounts, reason: string): void { const paths = Object.keys(dropped); this.#droppedPathCount += paths.length; + droppedPathsCounter.add(paths.length); console.error( `[metering] dropping ${paths.length} unwritable path(s) of ${key}: ${reason}`, ); @@ -943,9 +1178,50 @@ export class MeteringBufferStore extends PuterStore { * would also replace the base, which is only correct when something was * actually written onward. */ - async #retireClaim(tag: string, nonce: string): Promise { - await this.clients.redis.del(pendingKey(tag, nonce)); - await this.clients.redis.hdel(pendingIndexKey(tag), nonce); + async #retireClaim(tag: string, key: string, nonce: string): Promise { + await this.#redis.meterRetire( + pendingKey(tag, nonce), + pendingIndexKey(tag), + deltaKey(tag, key), + trackedKey(tag), + nonce, + key, + ); + } + + /** + * Say loudly that a claim's amounts left the cache before anything wrote + * them onward. + * + * Unlike a dropped path there is no list of what was lost — the amounts + * went with the claim, and all that survived was the entry that pointed at + * it. That is the whole reason this has to be raised: nothing else in the + * system will ever notice the gap. + */ + #recordLostClaim(key: string): void { + lostClaimsCounter.add(1); + console.error( + `[metering] a claim on ${key} lost its amounts before they were written onward`, + ); + + if (isBilledCounter(key)) { + this.clients.alarm.create( + `metering_claim_lost:${key}`, + `A claim on ${key} left the cache before it was persisted — the amounts are lost and the account is under-billed`, + { key }, + 'critical', + { dedup: true }, + ); + return; + } + + this.clients.alarm.create( + 'metering_aggregate_claim_lost', + `An aggregate claim (${key}) left the cache before it was persisted — reporting totals will under-count`, + { key }, + 'warning', + { dedup: true }, + ); } }