From a552ce0b87ff366f0fc278ee2da9ca43e28006a1 Mon Sep 17 00:00:00 2001 From: Daniel Salazar Date: Wed, 19 Aug 2026 11:38:14 -0700 Subject: [PATCH] fix: better metrics for cache hit rates (#3609) * fix: better metrics for cache hit rates * chore: type + docs --- src/backend/clients/event/types.ts | 20 +++- .../share/ShareNotificationService.ts | 6 +- .../stores/metering/MeteringBufferStore.ts | 8 +- src/backend/stores/systemKv/SystemKVStore.ts | 92 ++++++++++++++++--- src/puter-js/src/modules/Email.js | 10 +- 5 files changed, 113 insertions(+), 23 deletions(-) diff --git a/src/backend/clients/event/types.ts b/src/backend/clients/event/types.ts index 175324dad..fa3caca1b 100644 --- a/src/backend/clients/event/types.ts +++ b/src/backend/clients/event/types.ts @@ -32,6 +32,24 @@ type GuiEvent> = { 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; @@ -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` diff --git a/src/backend/services/share/ShareNotificationService.ts b/src/backend/services/share/ShareNotificationService.ts index b80368d5c..23f216b64 100644 --- a/src/backend/services/share/ShareNotificationService.ts +++ b/src/backend/services/share/ShareNotificationService.ts @@ -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; - /** * 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. diff --git a/src/backend/stores/metering/MeteringBufferStore.ts b/src/backend/stores/metering/MeteringBufferStore.ts index 72e60728e..9e6c94a90 100644 --- a/src/backend/stores/metering/MeteringBufferStore.ts +++ b/src/backend/stores/metering/MeteringBufferStore.ts @@ -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]) } `; diff --git a/src/backend/stores/systemKv/SystemKVStore.ts b/src/backend/stores/systemKv/SystemKVStore.ts index 5326f0f50..f4806b717 100644 --- a/src/backend/stores/systemKv/SystemKVStore.ts +++ b/src/backend/stores/systemKv/SystemKVStore.ts @@ -17,6 +17,7 @@ * along with this program. If not, see . */ +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(); 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 | undefined; + | Record + | undefined; if (!startKey) { exhausted = remaining > 0; break; @@ -1163,7 +1228,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 @@ -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 | undefined; + | Record + | undefined; } while (countKey); } @@ -1554,7 +1621,8 @@ export class SystemKVStore extends PuterStore { probeUsage, writeUsage( (response.ConsumedCapacity?.CapacityUnits as - number | undefined) ?? 1, + | number + | undefined) ?? 1, ), ), }; diff --git a/src/puter-js/src/modules/Email.js b/src/puter-js/src/modules/Email.js index d88ae1f94..9b288ca02 100644 --- a/src/puter-js/src/modules/Email.js +++ b/src/puter-js/src/modules/Email.js @@ -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