feat(db): 统一元数据缓存层 MetadataCache
- 两级:L1 进程内存(按 region 限条数的 LRU、5 分钟)+ L2 PG kv_cache;读 L1 → L2 → 源,写两级 - 迁移 0086:kv_cache 加 fresh_until(过期先用旧值的宽限期)与 pinned(LRU 淘汰跳过);新表 cache_stat 按天 × region 汇总命中 - stale-while-revalidate:宽限期 = TTL 的 50%(上限 7 天),旧值先返回、同 key 后台只刷一次,刷新失败继续用旧值并记警告 - 同进程同 key 并发只回源一次;回源抛错不缓存;PG 读写失败按未命中 / 不写处理 - PgCache.delete / clear 发 pg_notify(umt_cache_invalidate),listenCacheInvalidation 收到后丢 L1;监听在线才启用 L1,断线停用、重连后整体清空 - PgCache.get 只认新鲜条目,getEntry 带回旧值;renew 只续新鲜条目并可标 pinned Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei
This commit is contained in:
rootandClaude Opus 5.5 committed
1 parent
0f656f0f58
commit
ad472a1a19
7 files changed
+815
-17
No files matched your search
+530
-17
@@ -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<T> {
|
||||
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<void> {
|
||||
@@ -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<void> {
|
||||
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<T = unknown>(region: string, key: string): Promise<T | null> {
|
||||
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<boolean> {
|
||||
/** 读一条(含宽限期内的旧值),命中刷新最近使用时间;MetadataCache 的 L2 读 */
|
||||
async getEntry<T = unknown>(region: string, key: string): Promise<CacheEntry<T> | 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<boolean> {
|
||||
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<void> {
|
||||
await sql`select pg_notify(${CACHE_INVALIDATE_CHANNEL}, ${invalidationPayload(region, key)})`
|
||||
.execute(this.db)
|
||||
.catch(() => {});
|
||||
}
|
||||
|
||||
async exists(region: string, key: string): Promise<boolean> {
|
||||
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<void> {
|
||||
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<number> {
|
||||
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<T = unknown>(
|
||||
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<T> {
|
||||
/** 秒 */
|
||||
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<boolean>;
|
||||
}
|
||||
|
||||
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 = <T>(v: T): T => (v === null || typeof v !== "object" ? v : structuredClone(v));
|
||||
|
||||
export class MetadataCache {
|
||||
private readonly l1 = new Map<string, Map<string, L1Entry>>();
|
||||
private readonly inflight = new Map<string, Promise<unknown>>();
|
||||
private readonly counters = new Map<string, CacheCounters>();
|
||||
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<string, CacheCounters> {
|
||||
return new Map([...this.counters].map(([r, c]) => [r, { ...c }]));
|
||||
}
|
||||
|
||||
/**
|
||||
* 把本进程的计数增量加进 cache_stat(按天 × region),成功后清零;失败保留到下次。
|
||||
* 顺带删掉 30 天前的统计行。返回写入的 region 数。
|
||||
*/
|
||||
async flushStats(): Promise<number> {
|
||||
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<T = unknown>(region: string, key: string): Promise<T | null> {
|
||||
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<T>(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<void> {
|
||||
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<void> {
|
||||
this.dropLocal(region, key);
|
||||
await this.pg.delete(region, key);
|
||||
}
|
||||
|
||||
async clear(region: string): Promise<number> {
|
||||
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<T>(region: string, key: string, p: FetchPolicy<T>, load: () => Promise<T>): Promise<T> {
|
||||
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<T>(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<T>(p: FetchPolicy<T>, 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<T>(region: string, key: string, p: FetchPolicy<T>): Promise<void> {
|
||||
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<T>(region: string, key: string, p: FetchPolicy<T>, load: () => Promise<T>): Promise<T> {
|
||||
const id = `${region}\u0000${key}`;
|
||||
const pending = this.inflight.get(id);
|
||||
if (pending) return pending as Promise<T>;
|
||||
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<T>(region: string, key: string, p: FetchPolicy<T>, load: () => Promise<T>): 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<T>(region: string, key: string, p: FetchPolicy<T>, v: T): Promise<void> {
|
||||
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<object, MetadataCache>();
|
||||
|
||||
/** 同一个 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<void>; connected(): boolean }> {
|
||||
let client: pg.Client | null = null;
|
||||
let stopped = false;
|
||||
let timer: ReturnType<typeof setTimeout> | 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(() => {});
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in new issue
Block a user