fix(files): S-6 identify goes through the shared TMDB limiter and caps remote queries
- invoke-limits: sharedInvokeLimiter() (one semaphore set per process); agent capability routes default to it - /files/identify: catalog.lookup with includeRemote runs under the shared TMDB semaphore + 15s timeout; at most 20 distinct remote queries per request, further items fall back to local recall with a warning Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
1 parent
ba1361a9b6
commit
b4560be48b
4 files changed
+49
-8
No files matched your search
@@ -137,3 +137,13 @@ export function createInvokeLimiter(
|
|||||||
},
|
},
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
let shared: InvokeLimiter | null = null;
|
||||||
|
/**
|
||||||
|
* 进程内共享的限流器:同一 API 进程里所有 HTTP 路径上的外部 IO 调用(能力调用端点、/files/identify 等)
|
||||||
|
* 共用同一组信号量,TMDB/站点并发上限才是真正的进程级上限(验收 07 S-6)。
|
||||||
|
*/
|
||||||
|
export function sharedInvokeLimiter(): InvokeLimiter {
|
||||||
|
shared ??= createInvokeLimiter();
|
||||||
|
return shared;
|
||||||
|
}
|
||||||
@@ -14,7 +14,7 @@ import type { ModuleApp, ServerContext } from "../types.ts";
|
|||||||
import { planBinding } from "./authorize.ts";
|
import { planBinding } from "./authorize.ts";
|
||||||
import { cancelTask, retryTask, taskToApi } from "./capabilities.ts";
|
import { cancelTask, retryTask, taskToApi } from "./capabilities.ts";
|
||||||
import { route } from "./http.ts";
|
import { route } from "./http.ts";
|
||||||
import { createInvokeLimiter, type InvokeLimiter, UpstreamBusyError, UpstreamTimeoutError } from "./invoke-limits.ts";
|
import { type InvokeLimiter, sharedInvokeLimiter, UpstreamBusyError, UpstreamTimeoutError } from "./invoke-limits.ts";
|
||||||
import { AgentRunError, abortRun, appendMessage, startRun } from "./service.ts";
|
import { AgentRunError, abortRun, appendMessage, startRun } from "./service.ts";
|
||||||
|
|
||||||
// Agent 会话 API、持续授权 / 单次审批、能力目录与调用、业务任务。
|
// Agent 会话 API、持续授权 / 单次审批、能力目录与调用、业务任务。
|
||||||
@@ -89,7 +89,7 @@ const decode = (c?: string) => {
|
|||||||
export function registerAgentRoutes(
|
export function registerAgentRoutes(
|
||||||
app: ModuleApp,
|
app: ModuleApp,
|
||||||
ctx: ServerContext,
|
ctx: ServerContext,
|
||||||
limiter: InvokeLimiter = createInvokeLimiter(),
|
limiter: InvokeLimiter = sharedInvokeLimiter(),
|
||||||
) {
|
) {
|
||||||
const db = () => ctx.db;
|
const db = () => ctx.db;
|
||||||
|
|
||||||
|
|||||||
@@ -11,12 +11,16 @@ import { sql } from "@umt/db";
|
|||||||
import { type IdentityCandidate, type ParsedRelease, parseReleasePath, queryTitles, rankCandidates } from "@umt/domain";
|
import { type IdentityCandidate, type ParsedRelease, parseReleasePath, queryTitles, rankCandidates } from "@umt/domain";
|
||||||
import { localIdentityCandidate } from "@umt/mapping";
|
import { localIdentityCandidate } from "@umt/mapping";
|
||||||
import type { Principal } from "../../http/auth.ts";
|
import type { Principal } from "../../http/auth.ts";
|
||||||
|
import { EXTERNAL_INVOKE_TIMEOUT_MS, sharedInvokeLimiter } from "../agent/invoke-limits.ts";
|
||||||
import { actorOfP, errMessage, FilesError, type FmDeps, resolveIn } from "./common.ts";
|
import { actorOfP, errMessage, FilesError, type FmDeps, resolveIn } from "./common.ts";
|
||||||
import { kindOfName } from "./list.ts";
|
import { kindOfName } from "./list.ts";
|
||||||
|
|
||||||
// 识别(只读,不落库):parseReleasePath(文件 + 父/祖父目录上下文)→ 本地召回(别名 pg_trgm + 外部身份)
|
// 识别(只读,不落库):parseReleasePath(文件 + 父/祖父目录上下文)→ 本地召回(别名 pg_trgm + 外部身份)
|
||||||
// → 远端候选(catalog.lookup 能力,走 TMDB 缓存;失败只降级为提示)→ rankCandidates,每项前 5。
|
// → 远端候选(catalog.lookup 能力,走 TMDB 缓存;失败只降级为提示)→ rankCandidates,每项前 5。
|
||||||
|
|
||||||
|
/** 单次 /files/identify 请求内去重后的远端(TMDB)查询上限 */
|
||||||
|
export const REMOTE_QUERY_MAX = 20;
|
||||||
|
|
||||||
const TRGM = 0.45;
|
const TRGM = 0.45;
|
||||||
const LEVEL: Record<string, FileIdentifyCandidate["level"]> = {
|
const LEVEL: Record<string, FileIdentifyCandidate["level"]> = {
|
||||||
rule_confirmed: "high",
|
rule_confirmed: "high",
|
||||||
@@ -140,6 +144,7 @@ export async function identify(
|
|||||||
return out;
|
return out;
|
||||||
};
|
};
|
||||||
|
|
||||||
|
let remoteQueries = 0;
|
||||||
const remoteFor = async (p: ParsedRelease) => {
|
const remoteFor = async (p: ParsedRelease) => {
|
||||||
const q = p.titleCn ?? p.title ?? p.titleEn;
|
const q = p.titleCn ?? p.title ?? p.titleEn;
|
||||||
if (!q) return { list: [], warning: null };
|
if (!q) return { list: [], warning: null };
|
||||||
@@ -147,17 +152,32 @@ export async function identify(
|
|||||||
const hit = remoteCache.get(key);
|
const hit = remoteCache.get(key);
|
||||||
if (hit) return hit;
|
if (hit) return hit;
|
||||||
let res: { list: IdentityCandidate[]; warning: string | null };
|
let res: { list: IdentityCandidate[]; warning: string | null };
|
||||||
|
// 单次请求内去重后的远端查询设上限,超出的项只做本地召回(验收 07 S-6)
|
||||||
|
if (remoteQueries >= REMOTE_QUERY_MAX) {
|
||||||
|
res = {
|
||||||
|
list: [],
|
||||||
|
warning: `本次识别的远端查询已达上限 ${REMOTE_QUERY_MAX} 个,该项仅本地召回;可缩小范围后再识别`,
|
||||||
|
};
|
||||||
|
remoteCache.set(key, res);
|
||||||
|
return res;
|
||||||
|
}
|
||||||
|
remoteQueries += 1;
|
||||||
try {
|
try {
|
||||||
const out = (await d.ctx.capabilities.invoke(
|
const input = {
|
||||||
"catalog.lookup",
|
query: q,
|
||||||
{ query: q, includeRemote: true, limit: 10, ...(body.hints?.year ? { year: body.hints.year } : {}) },
|
includeRemote: true,
|
||||||
{
|
limit: 10,
|
||||||
|
...(body.hints?.year ? { year: body.hints.year } : {}),
|
||||||
|
};
|
||||||
|
// 经进程共享的 TMDB 信号量(并发 4)+ 15s 超时,与能力调用端点同一个限流器
|
||||||
|
const out = (await sharedInvokeLimiter().run("catalog.lookup", input, undefined, (signal) =>
|
||||||
|
d.ctx.capabilities.invoke("catalog.lookup", input, {
|
||||||
actorId: actorOfP(principal),
|
actorId: actorOfP(principal),
|
||||||
taskId: `files.identify:${randomUUID()}`,
|
taskId: `files.identify:${randomUUID()}`,
|
||||||
scopeIds: [],
|
scopeIds: [],
|
||||||
signal: AbortSignal.timeout(15_000),
|
signal: signal ?? AbortSignal.timeout(EXTERNAL_INVOKE_TIMEOUT_MS),
|
||||||
source: "web",
|
source: "web",
|
||||||
},
|
}),
|
||||||
)) as {
|
)) as {
|
||||||
remote?: { provider: string; entityType: string; externalId: string; title: string; year: number | null }[];
|
remote?: { provider: string; entityType: string; externalId: string; title: string; year: number | null }[];
|
||||||
};
|
};
|
||||||
|
|||||||
@@ -78,6 +78,17 @@ describe.skipIf(skip)("files 识别 / 整理 / 探测 / 刮削", () => {
|
|||||||
expect(c[0]).toMatchObject({ provider: "tmdb", externalId: "999", mediaEntityId: null, kind: "movie" });
|
expect(c[0]).toMatchObject({ provider: "tmdb", externalId: "999", mediaEntityId: null, kind: "movie" });
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("identify:去重后的远端查询单次上限 20,超出的项只本地召回并给 warning(验收 07 S-6)", async () => {
|
||||||
|
await mkdir(join(h.dl, "many"), { recursive: true });
|
||||||
|
const names = Array.from({ length: 23 }, (_, i) => `Distinct Title ${String.fromCharCode(65 + i)}.2019.1080p.mkv`);
|
||||||
|
for (const n of names) await writeFile(join(h.dl, "many", n), "x");
|
||||||
|
const r = await h.api("POST", "/files/identify", { items: [{ root: h.dlRoot, path: "many" }] }, READ_KEY);
|
||||||
|
expect(r.status).toBe(200);
|
||||||
|
expect(r.body.data.items).toHaveLength(23);
|
||||||
|
const capped = r.body.data.items.filter((i: { warnings: string[] }) => i.warnings.some((w) => w.includes("上限")));
|
||||||
|
expect(capped).toHaveLength(3);
|
||||||
|
});
|
||||||
|
|
||||||
it("organize:dryRun 返回 planId 与前后路径;源变更 → 该组 plan_stale;重新预览后执行入队 verifyAndImport", async () => {
|
it("organize:dryRun 返回 planId 与前后路径;源变更 → 该组 plan_stale;重新预览后执行入队 verifyAndImport", async () => {
|
||||||
const items = [{ root: h.dlRoot, path: "Blue.Film.2021.2160p.BluRay.x265-GRP.mkv", mediaEntityId: film }];
|
const items = [{ root: h.dlRoot, path: "Blue.Film.2021.2160p.BluRay.x265-GRP.mkv", mediaEntityId: film }];
|
||||||
const dry = await h.api("POST", "/files/organize", { idempotencyKey: key(), items, action: "copy", dryRun: true });
|
const dry = await h.api("POST", "/files/organize", { idempotencyKey: key(), items, action: "copy", dryRun: true });
|
||||||
|
|||||||
Reference in new issue
Block a user