feat(cache): read cache TTLs from storage settings and renew tracked TMDB details (CACHE-2)

Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
rootandClaude Opus 5.5 committed 2026-09-29 17:16:50 +08:00
1 parent 58e358e4a9
commit 85b352b8eb
10 files changed
+140 -35

No files matched your search

@@ -15,7 +15,14 @@ import {
} from "@umt/adapters";
import { DEFAULT_SETTINGS, type SitesSettings } from "@umt/contracts";
import { pgTokenBucketFor, type SiteRow, sql } from "@umt/db";
import { CACHE_REGIONS, sharedCache, type TmdbCachePolicy, tmdbCachePolicy } from "../../shared-cache.ts";
import {
CACHE_REGIONS,
renewTrackedTmdbDetail,
sharedCache,
type TmdbCachePolicy,
tmdbCachePolicy,
tmdbDetailRef,
} from "../../shared-cache.ts";
import type { ServerContext } from "../types.ts";
// 模块运行时依赖:站点适配器与限流器(每站一个实例,令牌在 PG 跨进程共享)、RSS 抓取器、TMDB 客户端、时钟。
@@ -108,7 +115,11 @@ export class SourcesRuntime {
...(baseUrl ? { baseUrl } : {}),
...(this.overrides.fetch ? { fetch: this.overrides.fetch } : {}),
// 请求响应进 PG 缓存(region tmdb.request),三进程共享;TTL / 条数上限 / 空结果短 TTL 取设置(对照 MP)
cache: { ttlMs: policy.ttl * 1000, emptyTtlMs: policy.emptyTtl * 1000 },
cache: {
ttlMs: policy.ttl * 1000,
searchTtlMs: policy.searchTtl * 1000,
emptyTtlMs: policy.emptyTtl * 1000,
},
sharedCache: tmdbSharedCache(this.ctx, policy),
})
: null;
@@ -276,17 +287,27 @@ export class SourcesRuntime {
}
}
/** TMDB 请求响应缓存:PG kv_cache region=tmdb.request;PG 出错按未命中 / 不写处理,不让 TMDB 请求失败 */
/**
* TMDB 请求响应缓存:PG kv_cache region=tmdb.request;PG 出错按未命中 / 不写处理,不让 TMDB 请求失败。
* 详情命中且属于追踪中的作品时续期(CACHE-2),续期失败只记 debug。
*/
function tmdbSharedCache(ctx: ServerContext, policy: TmdbCachePolicy): TmdbSharedCache {
const cache = sharedCache(ctx);
const warn = (err: unknown) =>
ctx.logger.warn({ err: { message: (err as Error).message } }, "TMDB 缓存读写失败,按未命中处理");
return {
get: (key) =>
cache.get(CACHE_REGIONS.tmdbRequest, key).catch((e) => {
get: async (key) => {
const v = await cache.get(CACHE_REGIONS.tmdbRequest, key).catch((e) => {
warn(e);
return null;
}),
});
if (v !== null && tmdbDetailRef(key)) {
await renewTrackedTmdbDetail(ctx, key, policy.ttl).catch((e) =>
ctx.logger.debug({ err: { message: (e as Error).message } }, "TMDB 详情缓存续期失败"),
);
}
return v;
},
set: (key, value, ttlMs) =>
cache.set(CACHE_REGIONS.tmdbRequest, key, value, { ttl: ttlMs / 1000, maxsize: policy.maxsize }).catch(warn),
};
@@ -1,18 +1,18 @@
import type { SiteDetailPage } from "@umt/adapters";
import type { SiteReleasePage } from "@umt/contracts";
import { sql } from "@umt/db";
import { sharedCache } from "../../../shared-cache.ts";
import { sharedCache, storageSettings } from "../../../shared-cache.ts";
import type { SourcesRuntime } from "../runtime.ts";
import type { Exec } from "./common.ts";
// 站点详情页扩展内容(简介 / MediaInfo / 截图 / 外部 ID / 文件 / 标签 / 上传者):
// - 只在按需抓详情时得到(工作台打开详情、release.inspect 复核),放 PG kv_cache region site.detail,
// key = source_release.id,TTL = 站点详情缓存期(至少 24 小时);不进 source_release 列,过期即丢。
// key = source_release.id,TTL = storage.cacheTtl.siteDetailDays(CACHE-2,缺省 7 天;不短于站点详情缓存期);
// 不进 source_release 列,过期即丢。促销 / H&R 不靠它:下载前复核走 refreshReleaseDetail(按站点详情缓存期判断是否重抓)。
// - 外部 ID(IMDb / 豆瓣 / TMDB)另写 source_release.raw._detail(列表快照 upsert 不覆盖它),供身份解析用。
// - 截图对外只给代理路径 /api/v1/img/site-detail/{releaseId}/{index},原始地址留在服务端。
export const DETAIL_PAGE_REGION = "site.detail";
const MIN_TTL_SEC = 86_400;
const MAX_ENTRIES = 5_000;
interface StoredPage extends SiteDetailPage {
@@ -33,8 +33,9 @@ export async function saveDetailPage(
cacheSec: number,
): Promise<void> {
const stored: StoredPage = { ...page, fetchedAt: rt.now().toISOString() };
const { siteDetailDays } = (await storageSettings(rt.ctx)).cacheTtl;
await sharedCache(rt.ctx)
.set(DETAIL_PAGE_REGION, srId, stored, { ttl: Math.max(cacheSec, MIN_TTL_SEC), maxsize: MAX_ENTRIES })
.set(DETAIL_PAGE_REGION, srId, stored, { ttl: Math.max(cacheSec, siteDetailDays * 86_400), maxsize: MAX_ENTRIES })
.catch((e) => rt.ctx.logger.warn({ err: { message: (e as Error).message } }, "详情页缓存写入失败"));
const ids: DetailIds = {
...(page.imdbId ? { imdbId: page.imdbId } : {}),
@@ -1,5 +1,6 @@
import type { DecisionOption, PoolCandidate } from "@umt/contracts";
import { sql } from "@umt/db";
import { storageSettings } from "../../../shared-cache.ts";
import type { ServerContext } from "../../types.ts";
import { jsonb } from "../rows.ts";
import type { SourcesRuntime } from "../runtime.ts";
@@ -10,7 +11,7 @@ import { closeDecisions, type Exec, openDecision } from "./common.ts";
// - 同一来源 + 同一归一化标题合并为一条事项(dedupeKey identity-group:<siteId>:<normalized>),成员 = 池里
// decision_id 指向它且仍 pending 的条目;事项的 subject 与候选选项取自代表资源(最近一次进来的那条);
// - 成员被识别 / 指定 / 忽略时从组里摘除,组空了才关闭事项;
// - unmatched 池条目超过 identity.poolArchiveDays 没有新证据就归档。
// - unmatched 池条目超过 storage.retentionDays.identityPoolArchive 天没有新证据就归档。
const OPEN = ["open", "investigating", "awaiting_user"] as const;
/** 事项证据里最多列多少条成员资源(完整列表在待识别池按标题检索) */
@@ -208,14 +209,15 @@ const ARCHIVE_EVERY_MS = 3600_000;
let lastArchiveAt = 0;
/**
* 过期归档:unmatched 池条目超过 identity.poolArchiveDays 没有新证据 → archived(仍可按状态筛出、可手动指定;
* 过期归档:unmatched 池条目超过 storage.retentionDays.identityPoolArchive 天没有新证据 → archived(仍可按状态筛出、可手动指定;
* 同一资源再被观察到或新增相近别名时回到 pending)。每进程每小时最多执行一次。
*/
export async function archiveStalePool(rt: SourcesRuntime, opts: { force?: boolean } = {}): Promise<number> {
const now = rt.now();
if (!opts.force && now.getTime() - lastArchiveAt < ARCHIVE_EVERY_MS) return 0;
lastArchiveAt = now.getTime();
const { poolArchiveDays } = await rt.ctx.settings.get("identity");
// 与存储保留任务 storage.retention.identity_pool 同一条件(那边在 system 模块里按表直接执行)
const poolArchiveDays = (await storageSettings(rt.ctx)).retentionDays.identityPoolArchive;
const cutoff = new Date(now.getTime() - poolArchiveDays * 86_400_000);
const r = await rt.ctx.db
.updateTable("identity_candidate")
@@ -110,7 +110,8 @@ export async function saveRecognizeCache(
poster_path: info.posterPath ?? null,
backdrop_path: info.backdropPath ?? null,
};
await cache.set(CACHE_REGIONS.tmdbRecognize, key, entry, { ttl: policy.ttl, maxsize: policy.maxsize });
// 识别结果 = 名称 → TMDB 条目,按搜索类 TTL(storage.cacheTtl.tmdbSearchDays)
await cache.set(CACHE_REGIONS.tmdbRecognize, key, entry, { ttl: policy.searchTtl, maxsize: policy.maxsize });
return;
}
// 负识别缓存用独立短 TTL(EMPTY_RESULT_CACHE_TTL):故障期间的空结果不被固化整个有效期
+3 -4
View File
@@ -17,10 +17,11 @@ import {
type WebSearchTestResult,
} from "@umt/contracts";
import { cachedCall, pgCacheFor, pgTokenBucketFor } from "@umt/db";
import { storageSettings } from "../../shared-cache.ts";
import type { ServerContext } from "../types.ts";
/**
* WEB-1 网页搜索服务:Brave 客户端之上的 Key 读取、每分钟上限(跨进程令牌桶)、10 分钟 PG 缓存(kv_cache region websearch)、
* WEB-1 网页搜索服务:Brave 客户端之上的 Key 读取、每分钟上限(跨进程令牌桶)、PG 缓存(kv_cache region websearch,TTL = storage.cacheTtl.webSearchDays)、
* 以及 web.fetch 的白名单——按调用方(Agent 会话 / 其它调用者)记录最近一次 web.search 返回的 URL,只允许读取这些 URL。
*/
@@ -36,8 +37,6 @@ export interface WebSearchModuleOptions {
export const WEBSEARCH_CACHE_REGION = "websearch";
/** 白名单:key = 调用方,value = 最近一次搜索的 URL */
export const WEBSEARCH_ALLOW_REGION = "websearch.allow";
/** 搜索结果缓存 10 分钟 */
export const SEARCH_CACHE_TTL_SEC = 600;
const SEARCH_CACHE_MAX = 1000;
/** 白名单保留 6 小时(会话时长上限远小于此) */
const ALLOW_TTL_SEC = 6 * 3600;
@@ -142,7 +141,7 @@ export class WebSearchService {
key,
{
region: WEBSEARCH_CACHE_REGION,
ttl: SEARCH_CACHE_TTL_SEC,
ttl: (await storageSettings(this.ctx)).cacheTtl.webSearchDays * 86_400,
maxsize: SEARCH_CACHE_MAX,
...(o.fresh ? { fresh: true } : {}),
},
+74 -9
View File
@@ -1,9 +1,10 @@
import { DEFAULT_SETTINGS } from "@umt/contracts";
import { type PgCache, pgCacheFor } from "@umt/db";
import { DEFAULT_SETTINGS, type StorageSettings, withStorageDefaults } from "@umt/contracts";
import { type PgCache, pgCacheFor, sql } from "@umt/db";
import type { ServerContext } from "./modules/types.ts";
// 进程级共享缓存入口:PG kv_cache(移植 MP app/runtime/cache.py 的 TTLCache,存储换成 PG,三进程共享)。
// region 与 TTL 规则见 docs/runbook/cache.md 与 docs/porting/moviepilot-tmdb-cache.md。
// region 与 TTL 规则见 docs/runbook/cache.md 与 docs/porting/moviepilot-tmdb-cache.md;
// CACHE-2 起各 region 的 TTL 取值来源是 settings.storage.cacheTtl(docs/runbook/storage.md),语义不变。
export const CACHE_REGIONS = {
/** TMDB 请求响应(MP TMDb.request 的 @cached):key = path?query */
@@ -12,14 +13,27 @@ export const CACHE_REGIONS = {
tmdbRecognize: "tmdb.recognize",
} as const;
const DAY = 86_400;
export function sharedCache(ctx: Pick<ServerContext, "db">): PgCache {
return pgCacheFor(ctx.db);
}
/** TMDB 缓存策略(MP:CONF.meta = META_CACHE_EXPIRE 小时、CONF.tmdb 条数、EMPTY_RESULT_CACHE_TTL 秒) */
/** settings.storage(读失败用缺省) */
export async function storageSettings(ctx: Pick<ServerContext, "settings">): Promise<StorageSettings> {
return withStorageDefaults(await ctx.settings.get("storage").catch(() => undefined));
}
/**
* TMDB 缓存策略(MP:CONF.meta = META_CACHE_EXPIRE、CONF.tmdb 条数、EMPTY_RESULT_CACHE_TTL)。
* CACHE-2:MP 只有一个 CONF.meta,这里按请求类别取 storage.cacheTtl:搜索 / find 与识别结果用 searchTtl,
* 详情(/movie/{id}、/tv/{id}、季)用 ttl(= tmdbDetailDays),空结果与未识别负缓存用 emptyTtl。
*/
export interface TmdbCachePolicy {
/** 秒 */
/** 详情 TTL(秒) */
ttl: number;
/** 搜索 / find / 识别结果 TTL(秒) */
searchTtl: number;
maxsize: number;
/** 秒 */
emptyTtl: number;
@@ -27,10 +41,61 @@ export interface TmdbCachePolicy {
export async function tmdbCachePolicy(ctx: Pick<ServerContext, "settings">): Promise<TmdbCachePolicy> {
const c = await ctx.settings.get("cache").catch(() => DEFAULT_SETTINGS.cache);
const d = DEFAULT_SETTINGS.cache;
const t = (await storageSettings(ctx)).cacheTtl;
return {
ttl: (c.tmdbTtlHours ?? d.tmdbTtlHours) * 3600,
maxsize: c.tmdbMaxSize ?? d.tmdbMaxSize,
emptyTtl: c.emptyResultTtlSec ?? d.emptyResultTtlSec,
ttl: t.tmdbDetailDays * DAY,
searchTtl: t.tmdbSearchDays * DAY,
maxsize: c.tmdbMaxSize ?? DEFAULT_SETTINGS.cache.tmdbMaxSize,
emptyTtl: t.tmdbMissDays * DAY,
};
}
/** TMDB 请求键是否是搜索类(按 searchTtl 缓存) */
export function isTmdbSearchPath(key: string): boolean {
return /^\/(search|find)\//.test(key);
}
/** 详情键 → `movie:<id>` / `tv:<id>`;不是详情返回 null */
export function tmdbDetailRef(key: string): string | null {
const m = /^\/(movie|tv)\/(\d+)(?:[/?]|$)/.exec(key);
return m ? `${m[1]}:${m[2]}` : null;
}
// ── 追踪中的作品:详情缓存命中时续期 ──
// 追踪 = 作品有未结束的订阅(active / paused / pending)且有已接受的 TMDB 身份。集合按进程缓存 5 分钟。
const TRACKED_TTL_MS = 5 * 60_000;
const trackedCache = new WeakMap<object, { at: number; set: Promise<Set<string>> }>();
export async function trackedTmdbRefs(ctx: Pick<ServerContext, "db">, now = Date.now()): Promise<Set<string>> {
const hit = trackedCache.get(ctx.db);
if (hit && now - hit.at < TRACKED_TTL_MS) return hit.set;
const set = sql<{ ref: string }>`select distinct x.entity_type || ':' || x.external_id as ref
from subscription s join external_identity x on x.media_entity_id = s.media_entity_id
where s.status in ('active', 'paused', 'pending') and x.provider = 'tmdb' and x.status = 'accepted'
and x.entity_type in ('movie', 'tv')`
.execute(ctx.db)
.then((r) => new Set(r.rows.map((x) => x.ref)));
trackedCache.set(ctx.db, { at: now, set });
set.catch(() => trackedCache.delete(ctx.db));
return set;
}
/** 测试用:清掉追踪集合缓存 */
export function resetTrackedTmdbRefs(ctx: Pick<ServerContext, "db">): void {
trackedCache.delete(ctx.db);
}
/**
* TMDB 详情缓存命中后的续期:键属于追踪中的作品时把过期时间推到 now + 详情 TTL(只延长不缩短)。
* 返回是否续期。失败不影响命中结果。
*/
export async function renewTrackedTmdbDetail(
ctx: Pick<ServerContext, "db">,
key: string,
detailTtlSec: number,
): Promise<boolean> {
const ref = tmdbDetailRef(key);
if (!ref || !(await trackedTmdbRefs(ctx)).has(ref)) return false;
return await sharedCache(ctx).renew(CACHE_REGIONS.tmdbRequest, key, detailTtlSec);
}
@@ -468,18 +468,21 @@ describe.skipIf(!up)("resolveObservation / 身份与元数据", () => {
).rows;
const opp = (await rows()).find((r) => r.key === "[未知][zh-CN]Oppenheimer-2023-None");
expect(opp?.value).toMatchObject({ id: 872585, type: "movie", title: "奥本海默" });
expect(Number(opp!.minutes)).toBeGreaterThan(23 * 60);
// 正常条目按搜索类 TTL(storage.cacheTtl.tmdbSearchDays,缺省 7 天)
expect(Number(opp!.minutes)).toBeGreaterThan(7 * 24 * 60 - 5);
expect(Number(opp!.minutes)).toBeLessThanOrEqual(7 * 24 * 60);
// 同名同年的另一条资源:命中识别缓存,不再调 TMDB 搜索
const searches = tmdb.callsTo(/\/3\/search\//).length;
const again = await release("Oppenheimer.2023.1080p.BluRay.x264-OTHER");
await resolveRelease(h.rt, { sourceReleaseId: again.id });
expect(tmdb.callsTo(/\/3\/search\//).length).toBe(searches);
// 查无此项 → 负缓存 {id: 0},TTL = 空结果短 TTL(30 分钟)
// 查无此项 → 负缓存 {id: 0},TTL = 未命中 TTL(storage.cacheTtl.tmdbMissDays,缺省 1 天)
const none = await release("Zzqxv.Nonexistent.Title.2019.1080p.WEB-DL-NOGRP");
await resolveRelease(h.rt, { sourceReleaseId: none.id });
const neg = (await rows()).find((r) => r.key === "[未知][zh-CN]Zzqxv Nonexistent Title-2019-None");
expect(neg?.value).toEqual({ id: 0 });
expect(Number(neg!.minutes)).toBeLessThanOrEqual(30);
expect(Number(neg!.minutes)).toBeLessThanOrEqual(24 * 60);
expect(Number(neg!.minutes)).toBeGreaterThan(24 * 60 - 5);
const s2 = tmdb.callsTo(/\/3\/search\//).length;
await resolveRelease(h.rt, { sourceReleaseId: none.id });
expect(tmdb.callsTo(/\/3\/search\//).length).toBe(s2);
+9 -4
View File
@@ -98,8 +98,11 @@ export interface TmdbConfig {
maxRetryAfterMs?: number;
maxConcurrent?: number;
minIntervalMs?: number;
/** 默认 max 1024、ttl 24h;空结果(搜索 0 条)用 emptyTtlMs,默认 30min(对照 MP EMPTY_RESULT_CACHE_TTL) */
cache?: { max?: number; ttlMs?: number; emptyTtlMs?: number };
/**
* 默认 max 1024、ttl 24h;空结果(搜索 0 条)用 emptyTtlMs,默认 30min(对照 MP EMPTY_RESULT_CACHE_TTL);
* searchTtlMs:/search/、/find/ 请求的 TTL(缺省同 ttlMs),其余(详情 / 季)用 ttlMs。
*/
cache?: { max?: number; ttlMs?: number; searchTtlMs?: number; emptyTtlMs?: number };
/**
* 跨进程共享的请求缓存(服务端传 PG 缓存);给了就不再用进程内 LRU。
* 语义对照 MP @cached(skip_none、业务失败不缓存、results 为空按 emptyTtlMs)。
@@ -164,6 +167,8 @@ export class TmdbClient {
for (const k of Object.keys(params).sort()) if (params[k] !== undefined) qs.set(k, params[k]!);
const key = `${path}?${qs}`;
const emptyTtl = this.cfg.cache?.emptyTtlMs ?? 30 * 60_000;
const baseTtl = this.cfg.cache?.ttlMs ?? 24 * 3600_000;
const ttl = /^\/(search|find)\//.test(path) ? (this.cfg.cache?.searchTtlMs ?? baseTtl) : baseTtl;
if (this.cfg.sharedCache) {
const shared = this.cfg.sharedCache;
const hit = await shared.get(key);
@@ -171,14 +176,14 @@ export class TmdbClient {
// 网络错误 / 业务失败在 fetchWithRetry 抛出,不会走到写缓存(MP skip_none / skip_if)
const v = await this.limiter.run(() => this.fetchWithRetry(path, qs));
if (v !== null && v !== undefined) {
await shared.set(key, v, isEmptyResult(v) ? emptyTtl : (this.cfg.cache?.ttlMs ?? 24 * 3600_000));
await shared.set(key, v, isEmptyResult(v) ? emptyTtl : ttl);
}
return v as T;
}
return this.cache.getOrLoad(
key,
() => this.limiter.run(() => this.fetchWithRetry(path, qs)),
(v) => (isEmptyResult(v) ? emptyTtl : undefined),
(v) => (isEmptyResult(v) ? emptyTtl : ttl === baseTtl ? undefined : ttl),
) as Promise<T>;
}
+8
View File
@@ -58,6 +58,14 @@ export class PgCache {
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> {
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);
return Number(r.numAffectedRows ?? 0) > 0;
}
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(
+1 -1
View File
@@ -1,5 +1,5 @@
// 统一从本包取 Kysely 与 sql,避免各包各装一份 kysely
export { type Kysely, type Selectable, sql, type Transaction } from "kysely";
export { type Kysely, type RawBuilder, type Selectable, sql, type Transaction } from "kysely";
export {
type CachedOptions,
cachedCall,