diff --git a/packages/contracts/test/db-sync.test.ts b/packages/contracts/test/db-sync.test.ts index 026c736f..28462f3c 100644 --- a/packages/contracts/test/db-sync.test.ts +++ b/packages/contracts/test/db-sync.test.ts @@ -149,6 +149,8 @@ const INTERNAL_TABLES = [ // C2-A 跨进程限流桶(0041)与 PG 缓存(0042):服务端内部,缓存经 /cache/* 的专用 DTO 暴露 "RateLimitBucket", "KvCache", + // 统一元数据缓存命中统计(0086):服务端内部,经 GET /cache/regions 的 hits 字段暴露 + "CacheStat", // 站点请求计数(0046):服务端内部,经 GET /sites/{id}/request-stats 的专用 DTO 暴露 "SiteRequestCount", // W9-I 进程心跳(0034):服务端内部,经 /system/health 的专用 DTO 暴露 diff --git a/packages/db/migrations/0086_metadata_cache.ts b/packages/db/migrations/0086_metadata_cache.ts new file mode 100644 index 00000000..3da25625 --- /dev/null +++ b/packages/db/migrations/0086_metadata_cache.ts @@ -0,0 +1,41 @@ +import type { Kysely } from "kysely"; +import { dropTables, execDdl } from "../src/migration-helpers.ts"; + +// 统一元数据缓存(docs/runbook/cache.md): +// - kv_cache.fresh_until:新鲜期截止。过了 fresh_until、还没到 expires_at 的条目是「旧值」: +// MetadataCache 先返回旧值再后台刷新(stale-while-revalidate);PgCache.get 只返回新鲜条目。null = 没有宽限期。 +// - kv_cache.pinned:关注中作品的条目,写入时按条数上限淘汰(LRU)跳过它们;到期仍会过期。 +// - cache_stat:按天 × region 汇总各进程的命中统计(进程内计数,定期把增量加进来)。 + +// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型 +export async function up(db: Kysely): Promise { + await execDdl( + db, + ` +alter table kv_cache add column fresh_until timestamptz; +alter table kv_cache add column pinned boolean not null default false; +create table cache_stat ( + day date not null, + region text not null, + l1_hits bigint not null default 0, + l2_hits bigint not null default 0, + stale_hits bigint not null default 0, + misses bigint not null default 0, + failures bigint not null default 0, + refreshes bigint not null default 0, + updated_at timestamptz not null default now(), + primary key (day, region) +); +`, + ); +} + +// biome-ignore lint/suspicious/noExplicitAny: 同上 +export async function down(db: Kysely): Promise { + await execDdl( + db, + `${dropTables(["cache_stat"])} +alter table kv_cache drop column if exists pinned; +alter table kv_cache drop column if exists fresh_until;`, + ); +} diff --git a/packages/db/src/cache.ts b/packages/db/src/cache.ts index e36d0a60..0e692c3f 100644 --- a/packages/db/src/cache.ts +++ b/packages/db/src/cache.ts @@ -1,4 +1,5 @@ import { type Kysely, sql } from "kysely"; +import pg from "pg"; import type { DB } from "./types.ts"; /** @@ -24,10 +25,37 @@ export interface SetOptions { ttl?: number; /** region 条数上限;缺省 DEFAULT_CACHE_SIZE */ maxsize?: number; + /** + * 宽限期(秒,0086):> 0 时 fresh_until = now + ttl、expires_at = now + ttl + grace, + * 宽限期内的条目由 MetadataCache 先返回旧值再后台刷新;PgCache.get 只认新鲜条目。缺省 0。 + */ + grace?: number; + /** 关注中作品的条目:按条数上限淘汰时跳过(到期照样过期)。缺省 false */ + pinned?: boolean; +} + +/** 跨进程 L1 失效通知频道(payload 见 invalidationPayload) */ +export const CACHE_INVALIDATE_CHANNEL = "umt_cache_invalidate"; + +/** 通知内容:`{"r":region,"k":key}`;k 缺省 = 整个 region;r 为 "*" = 全部。超长 key 退化为整 region */ +function invalidationPayload(region: string, key?: string): string { + const full = JSON.stringify(key === undefined ? { r: region } : { r: region, k: key }); + return full.length > 7000 ? JSON.stringify({ r: region }) : full; +} + +/** getEntry 的结果:stale = 已过新鲜期、还在宽限期内 */ +export interface CacheEntry { + value: T; + stale: boolean; + /** 新鲜期截止(毫秒);没有宽限期时同 expiresAt */ + freshUntil: number; + /** 硬过期(毫秒) */ + expiresAt: number; + pinned: boolean; } export class PgCache { - constructor(private readonly db: AnyDb) {} + constructor(readonly db: AnyDb) {} /** MemoryBackend.set */ async set(region: string, key: string, value: unknown, opts: SetOptions = {}): Promise { @@ -36,36 +64,83 @@ export class PgCache { await this.delete(region, key); return; } - await sql`insert into kv_cache (region, key, value, expires_at, updated_at, accessed_at) - values (${region}, ${key}, ${JSON.stringify(value ?? null)}::jsonb, now() + make_interval(secs => ${ttl}), now(), now()) + const grace = Math.max(0, opts.grace ?? 0); + await sql`insert into kv_cache (region, key, value, expires_at, fresh_until, pinned, updated_at, accessed_at) + values (${region}, ${key}, ${JSON.stringify(value ?? null)}::jsonb, + now() + make_interval(secs => ${ttl + grace}), + ${grace > 0 ? sql`now() + make_interval(secs => ${ttl})` : sql`null`}, + ${opts.pinned === true}, now(), now()) on conflict (region, key) do update set value = excluded.value, expires_at = excluded.expires_at, + fresh_until = excluded.fresh_until, pinned = excluded.pinned, updated_at = excluded.updated_at, accessed_at = excluded.accessed_at`.execute(this.db); await this.evict(region, opts.maxsize ?? DEFAULT_CACHE_SIZE); } - /** TLRUCache 写入时的淘汰:先删已过期,再按最近使用从旧到新删到 maxsize */ + /** + * TLRUCache 写入时的淘汰:先删已过期,再按最近使用从旧到新删到 maxsize。 + * 0086:pinned(关注中作品)的条目不参与 LRU,但计入条数;刚写入的那条总留着。 + */ private async evict(region: string, maxsize: number): Promise { await sql`delete from kv_cache where region = ${region} and expires_at <= now()`.execute(this.db); - await sql`delete from kv_cache where region = ${region} and key in ( - select key from kv_cache where region = ${region} - order by accessed_at desc, updated_at desc offset ${Math.max(1, Math.floor(maxsize))})`.execute(this.db); + await sql`delete from kv_cache where region = ${region} and not pinned and key in ( + select key from kv_cache where region = ${region} and not pinned + order by accessed_at desc, updated_at desc + offset greatest(1, ${Math.floor(maxsize)} - (select count(*) from kv_cache where region = ${region} and pinned)))`.execute( + this.db, + ); } - /** MemoryBackend.get:未命中 / 过期返回 null;命中刷新最近使用时间 */ + /** MemoryBackend.get:未命中 / 过期 / 已过新鲜期返回 null;命中刷新最近使用时间 */ async get(region: string, key: string): Promise { const r = await sql<{ value: T }>`update kv_cache set accessed_at = now() - where region = ${region} and key = ${key} and expires_at > now() returning value`.execute(this.db); + where region = ${region} and key = ${key} and expires_at > now() + and (fresh_until is null or fresh_until > now()) returning value`.execute(this.db); return r.rows.length ? (r.rows[0]!.value as T) : null; } - /** 续期:未过期的条目把 expires_at 推到 now + ttl(只延长不缩短);返回是否有行被续期 */ - async renew(region: string, key: string, ttl: number): Promise { + /** 读一条(含宽限期内的旧值),命中刷新最近使用时间;MetadataCache 的 L2 读 */ + async getEntry(region: string, key: string): Promise | null> { + const r = await sql<{ value: T; expires_at: Date; fresh_until: Date | null; pinned: boolean; now: Date }>` + update kv_cache set accessed_at = now() + where region = ${region} and key = ${key} and expires_at > now() + returning value, expires_at, fresh_until, pinned, now() as now`.execute(this.db); + const row = r.rows[0]; + if (!row) return null; + const expiresAt = new Date(row.expires_at).getTime(); + const freshUntil = row.fresh_until ? new Date(row.fresh_until).getTime() : expiresAt; + return { + value: row.value as T, + stale: freshUntil <= new Date(row.now).getTime(), + freshUntil, + expiresAt, + pinned: !!row.pinned, + }; + } + + /** + * 续期:新鲜的条目把过期时间推到 now + ttl(只延长不缩短,有宽限期的保留原宽限长度);返回是否有行被续期。 + * opts.pin = true 时顺带标为 pinned。旧值(已过新鲜期)不续期:该刷新就刷新。 + */ + async renew(region: string, key: string, ttl: number, opts: { pin?: boolean } = {}): Promise { if (ttl <= 0) return false; - const r = await sql`update kv_cache set expires_at = greatest(expires_at, now() + make_interval(secs => ${ttl})) - where region = ${region} and key = ${key} and expires_at > now()`.execute(this.db); + const r = await sql`update kv_cache set + expires_at = greatest(expires_at, + now() + make_interval(secs => ${ttl}) + coalesce(expires_at - fresh_until, interval '0')), + fresh_until = case when fresh_until is null then null + else greatest(fresh_until, now() + make_interval(secs => ${ttl})) end, + pinned = pinned or ${opts.pin === true} + where region = ${region} and key = ${key} and expires_at > now() + and (fresh_until is null or fresh_until > now())`.execute(this.db); return Number(r.numAffectedRows ?? 0) > 0; } + /** 通知各进程丢掉 L1(删除 / 清空 / 手动刷新时);失败不抛 */ + async notifyInvalidate(region: string, key?: string): Promise { + await sql`select pg_notify(${CACHE_INVALIDATE_CHANNEL}, ${invalidationPayload(region, key)})` + .execute(this.db) + .catch(() => {}); + } + async exists(region: string, key: string): Promise { const r = await sql`select 1 from kv_cache where region = ${region} and key = ${key} and expires_at > now()`.execute( @@ -74,13 +149,16 @@ export class PgCache { return r.rows.length > 0; } + /** 删除并通知各进程丢掉这条 L1 */ async delete(region: string, key: string): Promise { await sql`delete from kv_cache where region = ${region} and key = ${key}`.execute(this.db); + await this.notifyInvalidate(region, key); } - /** 清一个区,返回删除行数 */ + /** 清一个区,返回删除行数;通知各进程丢掉这个区的 L1 */ async clear(region: string): Promise { const r = await sql`delete from kv_cache where region = ${region}`.execute(this.db); + await this.notifyInvalidate(region); return Number(r.numAffectedRows ?? 0); } @@ -88,14 +166,24 @@ export class PgCache { async items( region: string, limit = 5000, - ): Promise<{ key: string; value: T; expiresAt: Date; updatedAt: Date; bytes: number }[]> { - const r = await sql<{ key: string; value: T; expires_at: Date; updated_at: Date; bytes: number }>` - select key, value, expires_at, updated_at, pg_column_size(value)::int as bytes + ): Promise<{ key: string; value: T; expiresAt: Date; freshUntil: Date; updatedAt: Date; bytes: number }[]> { + const r = await sql<{ + key: string; + value: T; + expires_at: Date; + fresh_until: Date; + updated_at: Date; + bytes: number; + }>` + select key, value, expires_at, coalesce(fresh_until, expires_at) as fresh_until, updated_at, + pg_column_size(value)::int as bytes from kv_cache where region = ${region} and expires_at > now() order by key limit ${limit}`.execute(this.db); return r.rows.map((x) => ({ key: x.key, value: x.value, expiresAt: x.expires_at, + /** 新鲜期截止(有宽限期的条目早于 expiresAt) */ + freshUntil: x.fresh_until, updatedAt: x.updated_at, bytes: Number(x.bytes), })); @@ -181,3 +269,428 @@ export function pgCacheFor(db: AnyDb): PgCache { } return c; } + +// ── 统一元数据缓存(TMDB / 豆瓣 / Bangumi 共用;docs/runbook/cache.md) ── +// 两级:L1 进程内存(按 region 限条数的 LRU,TTL 短)+ L2 PG kv_cache(三进程共享)。读 L1 → L2 → 源,写两级都写。 +// L1 只在跨进程失效通知(LISTEN umt_cache_invalidate)在线时启用:离线或断线重连时整体清空并停用, +// 避免别的进程清空 / 删除后本进程还拿着旧值。 + +/** 宽限期:TTL 的 50%,上限 7 天 */ +export const STALE_GRACE_RATIO = 0.5; +export const STALE_GRACE_MAX_SEC = 7 * 86_400; +export const defaultGrace = (ttlSec: number) => Math.min(Math.max(0, ttlSec) * STALE_GRACE_RATIO, STALE_GRACE_MAX_SEC); + +export interface FetchPolicy { + /** 秒 */ + ttl: number; + /** 空结果 TTL(秒);缺省同 ttl */ + emptyTtl?: number; + /** 空结果判定;缺省按假值(null / 空数组 / 空对象) */ + isEmpty?: (v: T) => boolean; + maxsize?: number; + /** 宽限期(秒);缺省 defaultGrace(本条 TTL);0 = 过期就回源 */ + grace?: number; + /** 缺省 true:null / undefined 不缓存 */ + skipNone?: boolean; + /** 返回真值时不缓存(业务失败的响应) */ + skipIf?: (v: T) => boolean; + /** true:跳过读(手动刷新),照常写,并通知各进程丢掉这条 L1 */ + fresh?: boolean; + /** + * 关注中作品的条目:写入时标 pinned(LRU 不淘汰),L2 命中时按 ttl 续期。 + * 只在写入 / L2 命中时调用,失败按 false。 + */ + pin?: () => Promise; +} + +export interface CacheCounters { + l1Hits: number; + l2Hits: number; + /** 先返回旧值(宽限期内)的次数 */ + staleHits: number; + misses: number; + /** 回源失败(网络、被拦、5xx 等抛错) */ + failures: number; + /** 后台刷新次数 */ + refreshes: number; +} + +const zero = (): CacheCounters => ({ l1Hits: 0, l2Hits: 0, staleHits: 0, misses: 0, failures: 0, refreshes: 0 }); + +export interface MetadataCacheOptions { + /** L1 每个 region 的条数上限,缺省 1000 */ + l1MaxPerRegion?: number; + /** L1 TTL(毫秒),缺省 5 分钟;不超过条目在 L2 的新鲜期 */ + l1TtlMs?: number; + /** 是否立即启用 L1(测试用;服务端由失效监听上线时打开) */ + l1?: boolean; + now?: () => number; + logger?: { warn(obj: object, msg: string): void }; +} + +interface L1Entry { + value: unknown; + freshUntil: number; + /** L1 内的过期(min(写入 + l1Ttl, 硬过期)) */ + expiresAt: number; +} + +const clone = (v: T): T => (v === null || typeof v !== "object" ? v : structuredClone(v)); + +export class MetadataCache { + private readonly l1 = new Map>(); + private readonly inflight = new Map>(); + private readonly counters = new Map(); + private l1On: boolean; + private readonly now: () => number; + + constructor( + readonly pg: PgCache, + private readonly opts: MetadataCacheOptions = {}, + ) { + this.l1On = opts.l1 ?? false; + this.now = opts.now ?? Date.now; + } + + // ── L1 ── + + get l1Enabled(): boolean { + return this.l1On; + } + + /** 打开 / 关闭 L1;状态变化时都整体清空 */ + setL1Enabled(on: boolean): void { + if (this.l1On !== on) this.l1.clear(); + this.l1On = on; + } + + /** 丢掉本进程 L1:给 key 丢一条,不给丢整个 region,region 为 "*" 全部丢 */ + dropLocal(region: string, key?: string): void { + if (region === "*") this.l1.clear(); + else if (key === undefined) this.l1.delete(region); + else this.l1.get(region)?.delete(key); + } + + /** L1 当前条数(测试 / 诊断) */ + l1Size(region: string): number { + return this.l1.get(region)?.size ?? 0; + } + + private l1Get(region: string, key: string): L1Entry | null { + if (!this.l1On) return null; + const m = this.l1.get(region); + const e = m?.get(key); + if (!m || !e) return null; + if (e.expiresAt <= this.now()) { + m.delete(key); + return null; + } + m.delete(key); + m.set(key, e); + return e; + } + + private l1Set(region: string, key: string, value: unknown, freshUntil: number, hardExpiresAt: number): void { + if (!this.l1On) return; + let m = this.l1.get(region); + if (!m) { + m = new Map(); + this.l1.set(region, m); + } + m.delete(key); + m.set(key, { + value: clone(value), + freshUntil, + expiresAt: Math.min(this.now() + (this.opts.l1TtlMs ?? 5 * 60_000), hardExpiresAt), + }); + const max = this.opts.l1MaxPerRegion ?? 1000; + while (m.size > max) { + const oldest = m.keys().next().value; + if (oldest === undefined) break; + m.delete(oldest); + } + } + + // ── 统计 ── + + private count(region: string, k: keyof CacheCounters, n = 1): void { + let c = this.counters.get(region); + if (!c) { + c = zero(); + this.counters.set(region, c); + } + c[k] += n; + } + + /** 本进程还没写进 cache_stat 的计数 */ + pendingStats(): Map { + return new Map([...this.counters].map(([r, c]) => [r, { ...c }])); + } + + /** + * 把本进程的计数增量加进 cache_stat(按天 × region),成功后清零;失败保留到下次。 + * 顺带删掉 30 天前的统计行。返回写入的 region 数。 + */ + async flushStats(): Promise { + const batch = [...this.counters]; + if (!batch.length) return 0; + this.counters.clear(); + let done = 0; + try { + for (const [region, c] of batch) { + await sql`insert into cache_stat (day, region, l1_hits, l2_hits, stale_hits, misses, failures, refreshes, updated_at) + values (current_date, ${region}, ${c.l1Hits}, ${c.l2Hits}, ${c.staleHits}, ${c.misses}, ${c.failures}, ${c.refreshes}, now()) + on conflict (day, region) do update set + l1_hits = cache_stat.l1_hits + excluded.l1_hits, l2_hits = cache_stat.l2_hits + excluded.l2_hits, + stale_hits = cache_stat.stale_hits + excluded.stale_hits, misses = cache_stat.misses + excluded.misses, + failures = cache_stat.failures + excluded.failures, refreshes = cache_stat.refreshes + excluded.refreshes, + updated_at = now()`.execute(this.pg.db); + done++; + } + await sql`delete from cache_stat where day < current_date - 30`.execute(this.pg.db); + return done; + } catch (err) { + // 没写进去的加回去 + for (const [region, c] of batch.slice(done)) + for (const k of Object.keys(c) as (keyof CacheCounters)[]) this.count(region, k, c[k]); + throw err; + } + } + + private warn(obj: object, msg: string) { + this.opts.logger?.warn(obj, msg); + } + + // ── 读写 ── + + /** 只读(不回源):L1 → L2 新鲜条目;记命中 / 未命中 */ + async get(region: string, key: string): Promise { + const l1 = this.l1Get(region, key); + if (l1 && l1.freshUntil > this.now()) { + this.count(region, "l1Hits"); + return clone(l1.value) as T; + } + const e = await this.pg.getEntry(region, key).catch((err) => { + this.warn({ region, err: { message: (err as Error).message } }, "缓存读取失败,按未命中处理"); + return null; + }); + if (!e || e.stale || e.value === null || e.value === undefined) { + this.count(region, "misses"); + return null; + } + this.count(region, "l2Hits"); + this.l1Set(region, key, e.value, e.freshUntil, e.expiresAt); + return e.value; + } + + /** 直接写两级(ttl ≤ 0 删除) */ + async set(region: string, key: string, value: unknown, opts: SetOptions = {}): Promise { + await this.pg.set(region, key, value, opts); + const ttl = opts.ttl ?? DEFAULT_CACHE_TTL; + if (ttl <= 0) { + this.dropLocal(region, key); + return; + } + const fresh = this.now() + ttl * 1000; + this.l1Set(region, key, value, fresh, fresh + Math.max(0, opts.grace ?? 0) * 1000); + } + + async delete(region: string, key: string): Promise { + this.dropLocal(region, key); + await this.pg.delete(region, key); + } + + async clear(region: string): Promise { + this.dropLocal(region); + return await this.pg.clear(region); + } + + /** + * 带回源的读(MP @cached 的调用体 + 两级 + 过期先用旧值 + 同请求合并 + 统计): + * - L1 / L2 新鲜命中直接返回;L2 命中且 pin() 为真时续期; + * - 宽限期内的旧值先返回,后台刷新(同 key 同时只刷一次),刷新失败继续用旧值并记一次警告; + * - 未命中:同进程同 key 只发一次 load;load 抛错不写缓存(网络、被拦、5xx);结果按 skipNone / skipIf 决定是否写。 + * PG 读写失败按未命中 / 不写处理,不让业务请求失败。 + */ + async fetch(region: string, key: string, p: FetchPolicy, load: () => Promise): Promise { + if (!p.fresh) { + const l1 = this.l1Get(region, key); + if (l1) { + if (l1.freshUntil > this.now()) { + this.count(region, "l1Hits"); + return clone(l1.value) as T; + } + this.count(region, "staleHits"); + this.revalidate(region, key, p, load); + return clone(l1.value) as T; + } + const e = await this.pg.getEntry(region, key).catch((err) => { + this.warn({ region, err: { message: (err as Error).message } }, "缓存读取失败,按未命中处理"); + return null; + }); + if (e && this.shouldCache(p, e.value)) { + this.l1Set(region, key, e.value, e.freshUntil, e.expiresAt); + if (!e.stale) { + this.count(region, "l2Hits"); + if (p.pin) await this.renewPinned(region, key, p); + return e.value; + } + this.count(region, "staleHits"); + this.revalidate(region, key, p, load); + return e.value; + } + } + this.count(region, "misses"); + return (await this.loadOnce(region, key, p, load)) as T; + } + + private shouldCache(p: FetchPolicy, v: T | null | undefined): v is T { + if ((p.skipNone ?? true) && (v === null || v === undefined)) return false; + if (p.skipIf && v !== null && v !== undefined && p.skipIf(v)) return false; + return true; + } + + private async renewPinned(region: string, key: string, p: FetchPolicy): Promise { + try { + if (await p.pin!()) await this.pg.renew(region, key, p.ttl, { pin: true }); + } catch (err) { + this.warn({ region, err: { message: (err as Error).message } }, "缓存续期失败"); + } + } + + private loadOnce(region: string, key: string, p: FetchPolicy, load: () => Promise): Promise { + const id = `${region}\u0000${key}`; + const pending = this.inflight.get(id); + if (pending) return pending as Promise; + const run = (async () => { + let v: T; + try { + v = await load(); + } catch (err) { + this.count(region, "failures"); + throw err; + } + if (this.shouldCache(p, v)) await this.write(region, key, p, v); + return v; + })().finally(() => this.inflight.delete(id)); + this.inflight.set(id, run); + return run; + } + + /** 宽限期内的后台刷新:同 key 已在刷新 / 加载就不再发 */ + private revalidate(region: string, key: string, p: FetchPolicy, load: () => Promise): void { + if (this.inflight.has(`${region}\u0000${key}`)) return; + this.count(region, "refreshes"); + this.loadOnce(region, key, { ...p, fresh: false }, load).catch((err) => + this.warn({ region, key, err: { message: (err as Error).message } }, "缓存后台刷新失败,继续用旧值"), + ); + } + + private async write(region: string, key: string, p: FetchPolicy, v: T): Promise { + const empty = p.isEmpty ? p.isEmpty(v) : isFalsy(v); + const ttl = empty && p.emptyTtl !== undefined ? p.emptyTtl : p.ttl; + const grace = p.grace ?? defaultGrace(ttl); + let pinned = false; + if (p.pin) pinned = await p.pin().catch(() => false); + try { + await this.pg.set(region, key, v, { ttl, grace, pinned, ...(p.maxsize ? { maxsize: p.maxsize } : {}) }); + } catch (err) { + this.warn({ region, err: { message: (err as Error).message } }, "缓存写入失败"); + } + const fresh = this.now() + ttl * 1000; + this.l1Set(region, key, v, fresh, fresh + grace * 1000); + if (p.fresh) await this.pg.notifyInvalidate(region, key); + } +} + +const metadataCaches = new WeakMap(); + +/** 同一个 Kysely 实例共用一个 MetadataCache(L2 用 pgCacheFor(db)) */ +export function metadataCacheFor(db: AnyDb, opts?: MetadataCacheOptions): MetadataCache { + let c = metadataCaches.get(db); + if (!c) { + c = new MetadataCache(pgCacheFor(db), opts); + metadataCaches.set(db, c); + } + return c; +} + +export interface InvalidationListenerOptions { + connectionString: string; + logger?: { warn(obj: object, msg: string): void; info(obj: object, msg: string): void }; + /** 断线重连退避上限,缺省 10s */ + maxReconnectMs?: number; +} + +/** + * 跨进程 L1 失效:一条专用连接 LISTEN umt_cache_invalidate,收到通知丢掉对应 L1。 + * 连上时清空并启用 L1;断开时停用(清空)L1 并退避重连,重连成功再整体清一次后启用。 + */ +export async function listenCacheInvalidation( + cache: MetadataCache, + opts: InvalidationListenerOptions, +): Promise<{ stop(): Promise; connected(): boolean }> { + let client: pg.Client | null = null; + let stopped = false; + let timer: ReturnType | null = null; + + const connect = async () => { + const c = new pg.Client({ connectionString: opts.connectionString }); + c.on("notification", (msg) => { + if (msg.channel !== CACHE_INVALIDATE_CHANNEL) return; + try { + const p = JSON.parse(msg.payload ?? "{}") as { r?: string; k?: string }; + if (typeof p.r === "string") cache.dropLocal(p.r, typeof p.k === "string" ? p.k : undefined); + } catch { + cache.dropLocal("*"); + } + }); + const lost = (err?: Error) => { + if (client !== c) return; + client = null; + cache.setL1Enabled(false); + c.removeAllListeners("notification"); + c.end().catch(() => {}); + if (!stopped) { + opts.logger?.warn({ err: err ? { message: err.message } : undefined }, "缓存失效监听断开,L1 停用并准备重连"); + reconnect(500); + } + }; + c.on("error", lost); + c.on("end", () => lost()); + await c.connect(); + await c.query(`listen ${CACHE_INVALIDATE_CHANNEL}`); + client = c; + cache.setL1Enabled(false); + cache.setL1Enabled(true); + }; + const reconnect = (delay: number) => { + if (stopped || timer) return; + timer = setTimeout(async () => { + timer = null; + try { + await connect(); + opts.logger?.info({}, "缓存失效监听已重连"); + } catch (err) { + opts.logger?.warn({ err: { message: (err as Error).message } }, "缓存失效监听重连失败"); + reconnect(Math.min(delay * 2, opts.maxReconnectMs ?? 10_000)); + } + }, delay); + }; + + await connect(); + return { + connected: () => client !== null, + async stop() { + stopped = true; + if (timer) clearTimeout(timer); + const c = client; + client = null; + cache.setL1Enabled(false); + if (c) { + c.removeAllListeners("notification"); + await c.end().catch(() => {}); + } + }, + }; +} diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts index 031e3928..7c7a2962 100644 --- a/packages/db/src/index.ts +++ b/packages/db/src/index.ts @@ -1,13 +1,24 @@ // 统一从本包取 Kysely 与 sql,避免各包各装一份 kysely export { type Kysely, type RawBuilder, type Selectable, sql, type Transaction } from "kysely"; export { + CACHE_INVALIDATE_CHANNEL, + type CacheCounters, type CachedOptions, + type CacheEntry, cachedCall, DEFAULT_CACHE_SIZE, DEFAULT_CACHE_TTL, + defaultGrace, + type FetchPolicy, + listenCacheInvalidation, + MetadataCache, + type MetadataCacheOptions, + metadataCacheFor, PgCache, pgCacheFor, type SetOptions as CacheSetOptions, + STALE_GRACE_MAX_SEC, + STALE_GRACE_RATIO, } from "./cache.ts"; export { type CreateDbOptions, createDb } from "./client.ts"; export { createMigrator, MIGRATIONS_DIR, type MigratorOptions, migrateToLatest, resetSchema } from "./migrator.ts"; diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index 041d46d2..3bbf065e 100644 --- a/packages/db/src/types.ts +++ b/packages/db/src/types.ts @@ -617,6 +617,18 @@ export interface BrowserTabTable { updated_at: GeneratedTimestamptz; } +export interface CacheStatTable { + day: string; + region: string; + l1_hits: Generated; + l2_hits: Generated; + stale_hits: Generated; + misses: Generated; + failures: Generated; + refreshes: Generated; + updated_at: GeneratedTimestamptz; +} + export interface CapabilityInvocationTable { id: Generated; idempotency_key: string; @@ -883,6 +895,8 @@ export interface KvCacheTable { expires_at: Timestamptz; updated_at: GeneratedTimestamptz; accessed_at: GeneratedTimestamptz; + fresh_until: Timestamptz | null; + pinned: Generated; } export interface LibraryRootTable { @@ -1798,6 +1812,7 @@ export interface DB { browser_provider: BrowserProviderTable; browser_session: BrowserSessionTable; browser_tab: BrowserTabTable; + cache_stat: CacheStatTable; capability_invocation: CapabilityInvocationTable; content_unit: ContentUnitTable; decision: DecisionTable; @@ -1929,6 +1944,9 @@ export type BrowserSessionUpdate = Updateable; export type BrowserTabRow = Selectable; export type NewBrowserTab = Insertable; export type BrowserTabUpdate = Updateable; +export type CacheStatRow = Selectable; +export type NewCacheStat = Insertable; +export type CacheStatUpdate = Updateable; export type CapabilityInvocationRow = Selectable; export type NewCapabilityInvocation = Insertable; export type CapabilityInvocationUpdate = Updateable; diff --git a/packages/db/test/metadata-cache.test.ts b/packages/db/test/metadata-cache.test.ts new file mode 100644 index 00000000..441ce55a --- /dev/null +++ b/packages/db/test/metadata-cache.test.ts @@ -0,0 +1,209 @@ +import { Kysely, PostgresDialect, sql } from "kysely"; +import pg from "pg"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { type DB, listenCacheInvalidation, MetadataCache, migrateToLatest, PgCache } from "../src/index.ts"; + +// 统一元数据缓存(0086):两级读写、跨进程失效、过期先用旧值、同请求合并、失败不缓存、pinned 不被淘汰、命中统计。 +const url = process.env.DATABASE_URL ?? "postgres://umt:umt@localhost:55433/umt"; +const skip = process.env.UMT_SKIP_DB_TESTS === "1"; +const schema = `umt_mcache_${process.pid}_${Date.now().toString(36)}`; + +const until = async (cond: () => boolean | Promise, ms = 5000) => { + const end = Date.now() + ms; + while (!(await cond())) { + if (Date.now() > end) throw new Error("等待超时"); + await new Promise((r) => setTimeout(r, 20)); + } +}; + +describe.skipIf(skip)("MetadataCache(真实 PG)", () => { + let admin: Kysely; + let db: Kysely; + /** 另一个「进程」:独立连接池 */ + let db2: Kysely; + + const pool = () => + new Kysely({ + dialect: new PostgresDialect({ + pool: new pg.Pool({ connectionString: url, max: 4, options: `-c search_path=${schema},public` }), + }), + }); + + beforeAll(async () => { + admin = new Kysely({ dialect: new PostgresDialect({ pool: new pg.Pool({ connectionString: url, max: 1 }) }) }); + await sql.raw(`create schema ${schema}`).execute(admin); + db = pool(); + db2 = pool(); + await migrateToLatest(db, { migrationTableSchema: schema }); + }); + + afterAll(async () => { + await db?.destroy(); + await db2?.destroy(); + await sql.raw(`drop schema if exists ${schema} cascade`).execute(admin); + await admin?.destroy(); + }); + + it("读 L1 → L2 → 源,写两级;另一进程从 L2 命中;统计各级命中", async () => { + const a = new MetadataCache(new PgCache(db), { l1: true }); + const b = new MetadataCache(new PgCache(db2), { l1: true }); + let n = 0; + const load = async () => ({ v: ++n }); + const p = { ttl: 600 }; + expect(await a.fetch("r1", "k", p, load)).toEqual({ v: 1 }); + expect(await a.fetch("r1", "k", p, load)).toEqual({ v: 1 }); + expect(await b.fetch("r1", "k", p, load)).toEqual({ v: 1 }); + expect(await b.fetch("r1", "k", p, load)).toEqual({ v: 1 }); + expect(n).toBe(1); + expect(a.pendingStats().get("r1")).toMatchObject({ misses: 1, l1Hits: 1, l2Hits: 0 }); + expect(b.pendingStats().get("r1")).toMatchObject({ misses: 0, l2Hits: 1, l1Hits: 1 }); + // L1 返回副本:调用方改了不污染缓存 + const got = await a.fetch<{ v: number }>("r1", "k", p, load); + got.v = 99; + expect(await a.fetch("r1", "k", p, load)).toEqual({ v: 1 }); + // 有宽限期:fresh_until = now + ttl,expires_at 再加 ttl 的 50% + const r = await sql<{ g: number }>`select extract(epoch from expires_at - fresh_until)::int as g + from kv_cache where region = 'r1' and key = 'k'`.execute(db); + expect(r.rows[0]!.g).toBe(300); + // 统计写进 cache_stat 后清零 + expect(await a.flushStats()).toBe(1); + await b.flushStats(); + expect(a.pendingStats().size).toBe(0); + const s = await sql<{ l1: number; l2: number; m: number }>`select l1_hits::int as l1, l2_hits::int as l2, + misses::int as m from cache_stat where region = 'r1'`.execute(db); + expect(s.rows[0]).toEqual({ l1: 4, l2: 1, m: 1 }); + }); + + it("跨进程失效:删除 / 清空通知其它进程丢 L1;监听断开时 L1 停用,重连后清空再启用", async () => { + const a = new MetadataCache(new PgCache(db)); + const b = new MetadataCache(new PgCache(db2)); + const la = await listenCacheInvalidation(a, { connectionString: url }); + const lb = await listenCacheInvalidation(b, { connectionString: url }); + try { + expect(a.l1Enabled && b.l1Enabled).toBe(true); + const p = { ttl: 600 }; + await a.fetch("inv", "x", p, async () => 1); + await a.fetch("inv", "y", p, async () => 2); + await b.fetch("inv", "x", p, async () => 0); + await b.fetch("inv", "y", p, async () => 0); + expect(b.l1Size("inv")).toBe(2); + // A 删 x:B 的 L1 只丢 x + await a.delete("inv", "x"); + await until(() => b.l1Size("inv") === 1); + expect(await b.fetch("inv", "x", p, async () => 10)).toBe(10); + // A 清空整个 region + await a.clear("inv"); + await until(() => b.l1Size("inv") === 0); + expect(await b.fetch("inv", "y", p, async () => 20)).toBe(20); + // 手动刷新(fresh)写入后也通知 + await a.fetch("inv", "y", { ...p, fresh: true }, async () => 21); + await until(() => b.l1Size("inv") === 0); + expect(await b.fetch("inv", "y", p, async () => 0)).toBe(21); + // 断开 B 的监听连接:L1 停用;自动重连后重新启用且是空的 + expect(b.l1Size("inv")).toBe(1); + await sql`select pg_terminate_backend(pid) from pg_stat_activity + where query ilike 'listen umt_cache_invalidate%' and pid <> pg_backend_pid()`.execute(db2); + await until(() => !b.l1Enabled || b.l1Size("inv") === 0); + expect(b.l1Size("inv")).toBe(0); + await until(() => b.l1Enabled && lb.connected(), 10_000); + expect(b.l1Size("inv")).toBe(0); + } finally { + await la.stop(); + await lb.stop(); + } + expect(a.l1Enabled).toBe(false); + }); + + it("过期在宽限期内:先返回旧值、后台只刷一次;源失败继续用旧值并记警告", async () => { + const warns: string[] = []; + const c = new MetadataCache(new PgCache(db), { l1: false, logger: { warn: (_o, m) => warns.push(m) } }); + let n = 0; + let gate: () => void = () => {}; + const slow = () => + new Promise((ok) => { + n++; + gate = () => ok(100 + n); + }); + await c.fetch("swr", "k", { ttl: 0.2, grace: 60 }, async () => 1); + await new Promise((r) => setTimeout(r, 300)); + const p = { ttl: 600 }; + const [x, y] = await Promise.all([c.fetch("swr", "k", p, slow), c.fetch("swr", "k", p, slow)]); + expect([x, y]).toEqual([1, 1]); + await until(() => n === 1); + gate(); + await until(async () => (await new PgCache(db).get("swr", "k")) === 101); + expect(n).toBe(1); + expect(c.pendingStats().get("swr")).toMatchObject({ staleHits: 2, refreshes: 1 }); + // 再次过期,源失败:返回旧值,不删旧值,记一次警告 + await c.fetch("swr", "k2", { ttl: 0.2, grace: 60 }, async () => "old"); + await new Promise((r) => setTimeout(r, 300)); + const failing = async (): Promise => { + throw new Error("HTTP 503"); + }; + expect(await c.fetch("swr", "k2", p, failing)).toBe("old"); + await until(() => warns.length === 1); + expect(warns[0]).toContain("继续用旧值"); + expect(await c.fetch("swr", "k2", p, async () => "new")).toBe("old"); + // 过了宽限期 = 硬过期,回源 + await c.fetch("swr", "k3", { ttl: 0.1, grace: 0.1 }, async () => "a"); + await new Promise((r) => setTimeout(r, 300)); + expect(await c.fetch("swr", "k3", p, async () => "b")).toBe("b"); + // PgCache.get 只认新鲜条目 + await c.fetch("swr", "k4", { ttl: 0.1, grace: 60 }, async () => "x"); + await new Promise((r) => setTimeout(r, 200)); + expect(await new PgCache(db).get("swr", "k4")).toBeNull(); + expect(await new PgCache(db).exists("swr", "k4")).toBe(true); + }); + + it("同请求合并:并发同 key 只回源一次;失败不缓存并计数", async () => { + const c = new MetadataCache(new PgCache(db), { l1: true }); + let n = 0; + const load = async () => { + n++; + await new Promise((r) => setTimeout(r, 50)); + return { ok: true }; + }; + const all = await Promise.all(Array.from({ length: 5 }, () => c.fetch("sf", "k", { ttl: 60 }, load))); + expect(n).toBe(1); + expect(all.every((v) => v.ok)).toBe(true); + let m = 0; + const fail = async () => { + m++; + throw new Error("ECONNRESET"); + }; + await expect(c.fetch("sf", "bad", { ttl: 60 }, fail)).rejects.toThrow("ECONNRESET"); + await expect(c.fetch("sf", "bad", { ttl: 60 }, fail)).rejects.toThrow("ECONNRESET"); + expect(m).toBe(2); + expect(await new PgCache(db).exists("sf", "bad")).toBe(false); + expect(c.pendingStats().get("sf")).toMatchObject({ failures: 2 }); + // skipIf / null 不缓存;空结果按 emptyTtl + await c.fetch("sf", "none", { ttl: 60 }, async () => null); + expect(await new PgCache(db).exists("sf", "none")).toBe(false); + await c.fetch("sf", "empty", { ttl: 3600, emptyTtl: 60, isEmpty: (v: number[]) => !v.length }, async () => []); + const t = await sql<{ s: number }>`select extract(epoch from fresh_until - now())::int as s + from kv_cache where region = 'sf' and key = 'empty'`.execute(db); + expect(t.rows[0]!.s).toBeLessThanOrEqual(60); + }); + + it("pinned:按条数上限淘汰时跳过;L2 命中时续期", async () => { + const pgc = new PgCache(db); + const c = new MetadataCache(pgc, { l1: false }); + const tick = () => new Promise((r) => setTimeout(r, 5)); + await c.fetch("pin", "watched", { ttl: 100, maxsize: 2, pin: async () => true }, async () => "w"); + for (const k of ["a", "b", "c", "d"]) { + await tick(); + await c.fetch("pin", k, { ttl: 100, maxsize: 2 }, async () => k); + } + expect((await pgc.items("pin")).map((i) => i.key).sort()).toEqual(["d", "watched"]); + // 续期:新鲜条目命中且 pin 为真 → 过期时间推到 now + ttl(保留宽限长度) + await sql`update kv_cache set fresh_until = now() + interval '10 seconds', expires_at = now() + interval '60 seconds' + where region = 'pin' and key = 'watched'`.execute(db); + expect(await c.fetch("pin", "watched", { ttl: 3600, pin: async () => true }, async () => "x")).toBe("w"); + const r = await sql<{ f: number; g: number }>`select extract(epoch from fresh_until - now())::int as f, + extract(epoch from expires_at - fresh_until)::int as g from kv_cache where region = 'pin' and key = 'watched'`.execute( + db, + ); + expect(r.rows[0]!.f).toBeGreaterThan(3500); + expect(r.rows[0]!.g).toBe(50); + }); +}); diff --git a/packages/db/test/schema.test.ts b/packages/db/test/schema.test.ts index 0b21163b..dae595e2 100644 --- a/packages/db/test/schema.test.ts +++ b/packages/db/test/schema.test.ts @@ -132,6 +132,8 @@ const EXPECTED_TABLES = [ "episode_mapping_rule", // 站点请求失败留样(0077) "site_failure_sample", + // 缓存命中统计(0086,按天 × region) + "cache_stat", ].sort(); /** 不走 entityTable() 的表:追加型/引擎自有表,没有 uuid id + created_at/updated_at 三件套 */ @@ -167,6 +169,8 @@ const NON_ENTITY_TABLES = [ "site_request_count", // 失败留样:只追加,只有 created_at "site_failure_sample", + // 缓存命中统计:(day, region) 主键 + "cache_stat", ]; describe.skipIf(skip)("schema migrations (real PG)", () => {