metrics: add metering buffer store coutns (#3554)

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<unknown>;
+                    },
+                    '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<number>;
+            };
+            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`;

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<string[] | null>;
-    meterReclaim(...args: string[]): Promise<string[] | null>;
+    /** `[1, amounts]` when re-claimed, `[0, hadIndexEntry]` when not. */
+    meterReclaim(...args: string[]): Promise<[number, string[] | number]>;
     meterSettle(...args: string[]): Promise<number>;
+    meterRetire(...args: string[]): Promise<number>;
+    meterReconcile(...args: string[]): Promise<number>;
     meterSeed(...args: string[]): Promise<string[]>;
 };
 
@@ -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<string>();
@@ -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<void>> = [];
         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<void> {
+        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<void> {
         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<void> {
         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<void> {
-        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<void> {
+        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 },
+        );
     }
 }