fix: better metrics for cache hit rates (#3609)
Maintain Release Merge PR / update-release-pr (push) Canceled after 0s
Notify HeyPuter / notify (push) Canceled after 0s
release-please / release-please (push) Canceled after 0s

* fix: better metrics for cache hit rates

* chore: type + docs
This commit is contained in:
Daniel Salazar
2026-08-19 11:38:14 -07:00
committed by GitHub
parent c04827fd3f
commit a552ce0b87
5 changed files with 113 additions and 23 deletions
+19 -1
View File
@@ -32,6 +32,24 @@ type GuiEvent<R = Record<string, unknown>> = {
response: R;
};
/**
* Extension-augmentable half of {@link EventMap}. Extensions that emit their own
* events declare the payload here by declaration merging, so both the emitter
* and every listener are typed against the same shape:
*
* declare module '@heyputer/backend/clients/event/types' {
* interface IExtensionEventMap {
* 'my.thing.happened': { thingId: string };
* }
* }
*
* Deliberately member-less and index-signature-free: an index signature here
* would widen `keyof EventMap` to `string` and silently disable key checking on
* every `emit` in the tree.
*/
// eslint-disable-next-line @typescript-eslint/no-empty-object-type
export interface IExtensionEventMap {}
export type EventMap = {
// ---- Server lifecycle ----
serverStart: Record<string, never>;
@@ -499,7 +517,7 @@ export type EventMap = {
* have all dropped that answer.
*/
'outer.pubsub.metering.credits-changed': { userUuid: string };
};
} & IExtensionEventMap;
/**
* Phase of a request/method lifecycle. `reject` is emitted when a `before`
@@ -18,10 +18,8 @@
*/
import { createHash, randomUUID } from 'node:crypto';
import { checkRateLimit } from '../../core/http/middleware/rateLimit.js';
import type { Actor } from '../../core/actor';
import type { LayerInstances } from '../../types';
import type { puterServices } from '../index';
import { checkRateLimit } from '../../core/http/middleware/rateLimit.js';
import { PuterService } from '../types';
import {
digestLines,
@@ -141,8 +139,6 @@ interface DigestEntryRecord {
}
export class ShareNotificationService extends PuterService {
declare protected services: LayerInstances<typeof puterServices>;
/**
* The flush timers this node owns, keyed per recipient. Timers only — the
* queued sends live in KV, where any node's flush can pick them up.
@@ -328,10 +328,16 @@ for i = 1, #abandoned do
redis.call('ZREM', KEYS[2], abandoned[i])
end
local taken = redis.call('SPOP', KEYS[1], ARGV[1]) or {}
-- Nothing taken means the dirty set is empty, including anything just put back
-- above, so the SCARD below could only be 0. Returning it directly matters
-- because most buckets are idle on most cycles: SPOP is a write and doesn't
-- count as a keyspace read, so skipping the SCARD is what keeps an idle
-- bucket's drain from registering as a cache miss.
if #taken == 0 then return { taken, 0 } end
for i = 1, #taken do
redis.call('ZADD', KEYS[2], ARGV[2], taken[i])
end
if #taken > 0 then redis.call('PEXPIRE', KEYS[2], ARGV[3]) end
redis.call('PEXPIRE', KEYS[2], ARGV[3])
return { taken, redis.call('SCARD', KEYS[1]) }
`;
+80 -12
View File
@@ -17,6 +17,7 @@
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
import { metrics } from '@opentelemetry/api';
import { PuterStore } from '../types';
import type { Actor } from '../../core/actor';
import {
@@ -47,6 +48,48 @@ import {
type KvCacheSettings,
} from './readCache';
const meter = metrics.getMeter('puter-backend');
/**
* What the read cache did with each key it was asked about, by `result`:
*
* - `hit` — answered from a cached value
* - `miss` — answered from a cached absence, which saves the same read a `hit`
* does and belongs on the same side of the ratio
* - `expired` — a cached value whose own deadline had passed, so it answered
* nothing and the key was read through
* - `blocked` — a recent write left a marker, so the read deliberately went
* through and did not populate
* - `absent` — nothing was cached; the read went through and populated
* - `error` — the cache could not be reached and the read degraded to uncached
*
* The rate worth watching is `(hit + miss) / total`. Deliberately not split by
* namespace: namespaces are per-app, so that would be unbounded cardinality.
*/
const cacheLookupCounter = meter.createCounter('kv.cache.lookup', {
description: 'KV read-cache lookups by outcome',
});
type CacheOutcomes = Record<
'hit' | 'miss' | 'expired' | 'blocked' | 'absent',
number
>;
const countedOutcomes = (): CacheOutcomes => ({
hit: 0,
miss: 0,
expired: 0,
blocked: 0,
absent: 0,
});
/** One `add` per outcome that actually occurred, rather than one per key. */
const recordCacheOutcomes = (outcomes: CacheOutcomes): void => {
for (const [result, count] of Object.entries(outcomes)) {
if (count > 0) cacheLookupCounter.add(count, { result });
}
};
// -- Types ------------------------------------------------------------
/** DynamoDB consumed-capacity units split by operation kind. */
@@ -454,6 +497,7 @@ export class SystemKVStore extends PuterStore {
const resolved = new Set<string>();
let readUnits = 0;
const now = Date.now() / 1000;
const outcomes = countedOutcomes();
keys.forEach((key, index) => {
const cached = decodeCachedRead(raw[index], key);
@@ -461,21 +505,33 @@ export class SystemKVStore extends PuterStore {
// The entry carries its own deadline and the cache TTL is
// only an upper bound on it, so an entry that lapsed since
// it was written counts as nothing cached at all.
if (cached.item.ttl && cached.item.ttl <= now) return;
if (cached.item.ttl && cached.item.ttl <= now) {
outcomes.expired++;
return;
}
outcomes.hit++;
items.push(cached.item);
resolved.add(key);
readUnits += cached.readUnits;
return;
}
if (cached.state === 'miss') {
outcomes.miss++;
resolved.add(key);
readUnits += cached.readUnits;
return;
}
if (cached.state === 'blocked') outcomes.blocked++;
else outcomes.absent++;
});
recordCacheOutcomes(outcomes);
return { items, resolved, readUnits };
} catch (e) {
// A cache that is down degrades to no cache, never to an error.
// Counted so that a cache which has stopped answering reads as
// exactly that, rather than as a cache nobody is asking.
cacheLookupCounter.add(keys.length, { result: 'error' });
console.warn(
'[kv] read cache lookup failed:',
(e as Error).message,
@@ -759,7 +815,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 +888,8 @@ export class SystemKVStore extends PuterStore {
probeUsage,
writeUsage(
response.ConsumedCapacity?.CapacityUnits as
number | undefined,
| number
| undefined,
),
),
};
@@ -925,7 +983,8 @@ export class SystemKVStore extends PuterStore {
probeUsage,
writeUsage(
(response.ConsumedCapacity?.CapacityUnits as
number | undefined) ?? 1,
| number
| undefined) ?? 1,
),
),
};
@@ -953,7 +1012,8 @@ export class SystemKVStore extends PuterStore {
await this.#invalidate(namespace, [key]);
const old = response.Attributes as
{ value?: unknown; ttl?: number } | undefined;
| { value?: unknown; ttl?: number }
| undefined;
const now = Date.now() / 1000;
const res =
old === undefined || (old.ttl && old.ttl <= now)
@@ -966,7 +1026,8 @@ export class SystemKVStore extends PuterStore {
probeUsage,
writeUsage(
(response.ConsumedCapacity?.CapacityUnits as
number | undefined) ?? 1,
| number
| undefined) ?? 1,
),
),
};
@@ -1044,7 +1105,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;
}
@@ -1123,7 +1186,8 @@ export class SystemKVStore extends PuterStore {
usage,
readUsage(
(response.ConsumedCapacity?.CapacityUnits as
number | undefined) ?? 1,
| number
| undefined) ?? 1,
),
);
return response;
@@ -1140,7 +1204,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;
@@ -1163,7 +1228,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
@@ -1199,7 +1265,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);
}
@@ -1554,7 +1621,8 @@ export class SystemKVStore extends PuterStore {
probeUsage,
writeUsage(
(response.ConsumedCapacity?.CapacityUnits as
number | undefined) ?? 1,
| number
| undefined) ?? 1,
),
),
};
+6 -4
View File
@@ -37,7 +37,7 @@ import * as utils from '../lib/utils.js';
* @typedef {Object} EmailSendResult
* @property {string | null} messageId First transport message id reported for this send, when available.
* @property {number} cost Total charge for this send, in microcents.
* @property {string[]} suppressed Recipients omitted because they opted out of this sender's mail.
* @property {string[]} suppressed Recipients omitted because they opted out of this app's mail.
* @property {string[]} failed Recipients whose delivery attempt failed. Everyone else got their copy
* retry with just these addresses. A send where every delivery fails rejects instead.
*/
@@ -66,9 +66,11 @@ import * as utils from '../lib/utils.js';
* Positional form: `await puter.email.send(to, subject, body)`.
*
* Every mail automatically gets an unsubscribe / report-abuse footer.
* Recipients who unsubscribe are dropped from future sends they come
* back in the result's `suppressed` array and a send whose `to` list
* is entirely unsubscribed is rejected.
* Unsubscribing is per app: a recipient who opts out stops hearing from
* the app they opted out of, and still hears from the other apps the same
* account runs. Opted-out recipients are dropped from that app's future
* sends they come back in the result's `suppressed` array and a send
* whose `to` list is entirely opted out is rejected.
*
* Each recipient gets a private delivery. A recipient whose delivery
* fails comes back in the result's `failed` array (everyone else got