fix(agent): download.recheck 复用下载同步的 qB 客户端
qB 客户端缓存改为按 ServerContext 共享(downloaderClientFor),Agent 不再自己 new 客户端, 与下载同步、面板共用同一登录会话和凭证处理。 Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei
This commit is contained in:
1 parent
74f2c7649e
commit
3d3a4cefc6
2 files changed
+35
-28
No files matched your search
@@ -117,12 +117,42 @@ export const DEFAULT_OPTIONS: AoOptions = {
|
|||||||
|
|
||||||
const str = (v: unknown) => (typeof v === "string" && v.length > 0 ? v : undefined);
|
const str = (v: unknown) => (typeof v === "string" && v.length > 0 ? v : undefined);
|
||||||
|
|
||||||
|
/** qB 客户端按 ServerContext 共享缓存(实例配置变了重建):下载同步、面板、Agent 的 download.recheck 共用同一登录会话 */
|
||||||
|
const sharedClients = new WeakMap<ServerContext, Map<string, { stamp: string; client: DownloaderClient }>>();
|
||||||
|
|
||||||
|
/** 按实例取 qB 客户端(凭证现场经 secrets 解出,解析失败按无凭证处理并记警告) */
|
||||||
|
export async function downloaderClientFor(ctx: ServerContext, inst: DownloaderInstanceRow): Promise<DownloaderClient> {
|
||||||
|
let clients = sharedClients.get(ctx);
|
||||||
|
if (!clients) {
|
||||||
|
clients = new Map();
|
||||||
|
sharedClients.set(ctx, clients);
|
||||||
|
}
|
||||||
|
const stamp = `${inst.updated_at.toISOString()}|${inst.base_url}`;
|
||||||
|
const hit = clients.get(inst.id);
|
||||||
|
if (hit && hit.stamp === stamp) return hit.client;
|
||||||
|
const settings = (inst.settings ?? {}) as Record<string, unknown>;
|
||||||
|
let secret: string | undefined;
|
||||||
|
if (inst.secret_ref) {
|
||||||
|
try {
|
||||||
|
secret = await ctx.secrets.resolve(inst.secret_ref);
|
||||||
|
} catch (err) {
|
||||||
|
ctx.logger.warn({ err: { message: (err as Error).message } }, "凭证解析失败");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
const client = createQbClient({
|
||||||
|
instanceId: inst.id,
|
||||||
|
baseUrl: inst.base_url,
|
||||||
|
...qbAuthConfig({ authKind: settings.authKind, username: inst.username, secret: secret ?? null }),
|
||||||
|
});
|
||||||
|
clients.set(inst.id, { stamp, client });
|
||||||
|
return client;
|
||||||
|
}
|
||||||
|
|
||||||
/** 用 ctx 组装默认端口;overrides 逐项覆盖(测试) */
|
/** 用 ctx 组装默认端口;overrides 逐项覆盖(测试) */
|
||||||
export function createPorts(
|
export function createPorts(
|
||||||
ctx: ServerContext,
|
ctx: ServerContext,
|
||||||
overrides: Partial<Omit<AoPorts, "options">> & { options?: Partial<AoOptions> } = {},
|
overrides: Partial<Omit<AoPorts, "options">> & { options?: Partial<AoOptions> } = {},
|
||||||
): AoPorts {
|
): AoPorts {
|
||||||
const clients = new Map<string, { stamp: string; client: DownloaderClient }>();
|
|
||||||
const resolveSecret = async (ref: string | null): Promise<string | undefined> => {
|
const resolveSecret = async (ref: string | null): Promise<string | undefined> => {
|
||||||
if (!ref) return undefined;
|
if (!ref) return undefined;
|
||||||
try {
|
try {
|
||||||
@@ -159,20 +189,7 @@ export function createPorts(
|
|||||||
if (r.created) await enqueueNewEntityRefresh(ctx, r.mediaEntityId);
|
if (r.created) await enqueueNewEntityRefresh(ctx, r.mediaEntityId);
|
||||||
return r.mediaEntityId;
|
return r.mediaEntityId;
|
||||||
},
|
},
|
||||||
async downloader(inst) {
|
downloader: (inst) => downloaderClientFor(ctx, inst),
|
||||||
const stamp = `${inst.updated_at.toISOString()}|${inst.base_url}`;
|
|
||||||
const hit = clients.get(inst.id);
|
|
||||||
if (hit && hit.stamp === stamp) return hit.client;
|
|
||||||
const settings = (inst.settings ?? {}) as Record<string, unknown>;
|
|
||||||
const secret = await resolveSecret(inst.secret_ref);
|
|
||||||
const client = createQbClient({
|
|
||||||
instanceId: inst.id,
|
|
||||||
baseUrl: inst.base_url,
|
|
||||||
...qbAuthConfig({ authKind: settings.authKind, username: inst.username, secret: secret ?? null }),
|
|
||||||
});
|
|
||||||
clients.set(inst.id, { stamp, client });
|
|
||||||
return client;
|
|
||||||
},
|
|
||||||
async torrentSource({ release, site }) {
|
async torrentSource({ release, site }) {
|
||||||
// 永久性条件用 SiteError:submitDownload 据此把计划判失败并释放 Slot,而不是反复重试、Slot 一直 reserved
|
// 永久性条件用 SiteError:submitDownload 据此把计划判失败并释放 Slot,而不是反复重试、Slot 一直 reserved
|
||||||
if (!site) throw new SiteError("", "not_found", "站点不存在,无法获取种子");
|
if (!site) throw new SiteError("", "not_found", "站点不存在,无法获取种子");
|
||||||
|
|||||||
@@ -1,7 +1,6 @@
|
|||||||
import { createQbClient, type DownloaderClient, qbAuthConfig } from "@umt/adapters";
|
|
||||||
import { CapabilityError, type CapabilityRegistry } from "@umt/agent";
|
import { CapabilityError, type CapabilityRegistry } from "@umt/agent";
|
||||||
import { capabilitySpec, type DownloadStatus } from "@umt/contracts";
|
import { capabilitySpec, type DownloadStatus } from "@umt/contracts";
|
||||||
import type { DownloaderInstanceRow } from "@umt/db";
|
import { downloaderClientFor } from "../acquisition-organize/ports.ts";
|
||||||
import type { ServerContext } from "../types.ts";
|
import type { ServerContext } from "../types.ts";
|
||||||
import { planBinding } from "./authorize.ts";
|
import { planBinding } from "./authorize.ts";
|
||||||
import type { AgentModuleOptions } from "./runner.ts";
|
import type { AgentModuleOptions } from "./runner.ts";
|
||||||
@@ -15,16 +14,6 @@ const RECHECK_FROM: DownloadStatus[] = ["missing", "failed", "stalled", "paused"
|
|||||||
/** resume 允许的起点 */
|
/** resume 允许的起点 */
|
||||||
const RESUME_FROM: DownloadStatus[] = ["paused", "stalled", "failed"];
|
const RESUME_FROM: DownloadStatus[] = ["paused", "stalled", "failed"];
|
||||||
|
|
||||||
async function defaultClient(ctx: ServerContext, inst: DownloaderInstanceRow): Promise<DownloaderClient> {
|
|
||||||
const settings = (inst.settings ?? {}) as Record<string, unknown>;
|
|
||||||
const secret = inst.secret_ref ? await ctx.secrets.resolve(inst.secret_ref) : undefined;
|
|
||||||
return createQbClient({
|
|
||||||
instanceId: inst.id,
|
|
||||||
baseUrl: inst.base_url,
|
|
||||||
...qbAuthConfig({ authKind: settings.authKind, username: inst.username, secret: secret ?? null }),
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
/** 持续授权的对象范围:按下载任务 → 计划 → Slot → 作品核对(authorize 对非计划类能力拿不到目标作品) */
|
/** 持续授权的对象范围:按下载任务 → 计划 → Slot → 作品核对(authorize 对非计划类能力拿不到目标作品) */
|
||||||
async function checkGrantScope(ctx: ServerContext, grantId: string, planId: string | null) {
|
async function checkGrantScope(ctx: ServerContext, grantId: string, planId: string | null) {
|
||||||
const g = await ctx.db.selectFrom("automation_grant").select("scope").where("id", "=", grantId).executeTakeFirst();
|
const g = await ctx.db.selectFrom("automation_grant").select("scope").where("id", "=", grantId).executeTakeFirst();
|
||||||
@@ -75,7 +64,8 @@ export function registerDownloadRecoveryCapabilities(
|
|||||||
.where("id", "=", dt.instance_id)
|
.where("id", "=", dt.instance_id)
|
||||||
.executeTakeFirst();
|
.executeTakeFirst();
|
||||||
if (!inst) throw new CapabilityError("not_found", "下载器实例不存在");
|
if (!inst) throw new CapabilityError("not_found", "下载器实例不存在");
|
||||||
const client = opts.downloader ? await opts.downloader(inst) : await defaultClient(ctx, inst);
|
// 与下载同步共用按实例缓存的客户端(同一登录会话与凭证处理),不另建
|
||||||
|
const client = opts.downloader ? await opts.downloader(inst) : await downloaderClientFor(ctx, inst);
|
||||||
try {
|
try {
|
||||||
if (input.action === "recheck") await client.recheck([dt.torrent_hash]);
|
if (input.action === "recheck") await client.recheck([dt.torrent_hash]);
|
||||||
else await client.resume([dt.torrent_hash]);
|
else await client.resume([dt.torrent_hash]);
|
||||||
|
|||||||
Reference in new issue
Block a user