diff --git a/src/app/api/stream/[type]/[id]/[fileId]/[[...name]]/route.ts b/src/app/api/stream/[type]/[id]/[fileId]/[[...name]]/route.ts index 7164e86..d2a39f0 100644 --- a/src/app/api/stream/[type]/[id]/[fileId]/[[...name]]/route.ts +++ b/src/app/api/stream/[type]/[id]/[fileId]/[[...name]]/route.ts @@ -2,6 +2,7 @@ import type { NextRequest } from "next/server"; import { IMAGE_BROWSER_MAX_AGE_SEC } from "@/constants/image-cache"; import { guard } from "@/server/auth"; +import { acquireCdnSlot, releaseWhenDone } from "@/server/cdn-limiter"; import { isMockMode } from "@/server/env"; import { cacheWhileStreaming, @@ -128,20 +129,24 @@ const proxy = async ( } } + /** 占用的下载节点并发名额,响应体传完(或出错、被取消)后释放 */ + let release = () => {}; try { let upstream: Response; - /** 下载节点:实测同一节点并发超过 20 会 429,出错时记录便于核对 */ + /** 下载节点:实测同一节点并发超过 20 会 429,本地按节点限制并发,出错时记录便于核对 */ let cdnHost = ""; /** 本次用的直链:来源与已用时长,出错时一并记录 */ let link: { attempt?: number; from?: DownloadUrlFrom; ageSec?: number } = {}; for (let attempt = 1; ; attempt += 1) { + // 被拒绝后重新签发的直链可能换了节点,先放掉上一次的名额 + release(); // 首次用缓存的直链,被拒绝后重新签发 const { url, signedAt, from } = await requestDownloadUrl( type, id, fileId, - { fresh: attempt > 1 } + { fresh: attempt > 1, context: media } ); cdnHost = new URL(url).host; link = { @@ -154,6 +159,7 @@ const proxy = async ( { ...media, method, cdnHost, ...link }, URL_FROM_LABEL[from] ); + release = await acquireCdnSlot(cdnHost, req.signal); upstream = await fetch(url, { method, headers: range ? { Range: range } : undefined, @@ -226,8 +232,12 @@ const proxy = async ( let body = method === "HEAD" ? null : upstream.body; if (cacheable && body) body = cacheWhileStreaming(cacheKey, upstream) ?? body; - return new Response(body, { status: upstream.status, headers }); + return new Response(releaseWhenDone(body, release), { + status: upstream.status, + headers + }); } catch (error) { + release(); // 播放器关闭或跳转时会主动断开,不算故障 if (req.signal.aborted) log.debug(media, "客户端中断请求"); else log.error({ ...media, method, err: error }, "播放失败"); diff --git a/src/constants/torbox-limits.ts b/src/constants/torbox-limits.ts index 71a9637..0820b4f 100644 --- a/src/constants/torbox-limits.ts +++ b/src/constants/torbox-limits.ts @@ -8,7 +8,7 @@ const MINUTE = 60 * 1000; const HOUR = 60 * MINUTE; export type TorboxLimitRule = { - /** 提示语里的操作名,如「获取播放地址过于频繁,请 3 分钟后再试」;以中文结尾,后面直接接「过于频繁」 */ + /** 提示语里的操作名,如「获取文件地址过于频繁,请 3 分钟后再试」;以中文结尾,后面直接接「过于频繁」 */ label: string; /** 任意 windowMs 时长内最多 limit 次(滑动窗口) */ limit: number; @@ -49,13 +49,13 @@ export const TORBOX_LIMITS = { }, /** * 文档未写。实测 1 秒 1 次、2 秒 1 次都在约第 100 次 429,Retry-After 300 秒, - * 约 20 次/分钟才不容易触发,即 5 分钟约 100 次的额度。取 80,平均 16 次/分钟, - * 同时容许图片墙一次加载几十张 + * 约 20 次/分钟才不容易触发,即 5 分钟约 100 次的额度。实际用量里 5 分钟内发出过约 91 次 + * 也没有 429,取 90,平均 18 次/分钟,同时容许图片墙一次加载几十张 */ requestDownload: { - label: "获取播放地址", + label: "获取文件地址", match: /^\/(torrents|usenet|webdl)\/requestdl$/, - limit: 80, + limit: 90, windowMs: 5 * MINUTE } } satisfies Record; @@ -76,6 +76,12 @@ export type TorboxLimitedResource = keyof typeof TORBOX_LIMITS_BY_RESOURCE; /** 同时进行中的 API 请求上限:实测约 40 个并发会把全部接口一起限住 */ export const TORBOX_MAX_CONCURRENT = 10; +/** + * 同一下载节点同时进行中的请求上限:实测并发超过 20 会 429(图片墙一次并发 17 个以上时出现), + * 留一点余量取 16,超出的排队等待 + */ +export const TORBOX_CDN_MAX_CONCURRENT = 16; + /** 本地额度用尽时最多排队等这么久,超过直接提示稍后再试,不让请求长时间挂起 */ export const TORBOX_MAX_WAIT_MS = 10 * 1000; diff --git a/src/hooks/use-limit-retry.ts b/src/hooks/use-limit-retry.ts index 4cf071d..0da0b0d 100644 --- a/src/hooks/use-limit-retry.ts +++ b/src/hooks/use-limit-retry.ts @@ -25,7 +25,7 @@ const fetchLimits = async () => { }; export type LimitWait = { - /** 受限的操作名,如「获取播放地址」 */ + /** 受限的操作名,如「获取文件地址」 */ label: string; /** 恢复时间,本机毫秒时间戳 */ until: number; diff --git a/src/server/cdn-limiter.ts b/src/server/cdn-limiter.ts new file mode 100644 index 0000000..1d1daf1 --- /dev/null +++ b/src/server/cdn-limiter.ts @@ -0,0 +1,99 @@ +// 下载节点并发限制:图片墙一次会并发拉几十张图,同一节点并发过多会返回 429。 +// 与 TorBox API 的额度无关(不扣额度、不记入 SQLite),只在本进程内按节点主机名限制同时进行的请求数 +import { TORBOX_CDN_MAX_CONCURRENT } from "@/constants/torbox-limits"; + +type HostState = { + active: number; + /** 等待名额的请求,先到先得 */ + queue: (() => void)[]; +}; + +// 挂在 globalThis 上,避免各路由打包后各持一份 +const HOSTS_KEY = Symbol.for("torbox-admin.cdn-hosts"); +const globals = globalThis as { [HOSTS_KEY]?: Map }; +const hosts = (globals[HOSTS_KEY] ??= new Map()); + +/** + * 占用该节点的一个并发名额,返回释放函数(可重复调用)。 + * 排队期间客户端离开(signal 中断)则放弃排队并抛出中断原因; + * 占用期间 signal 中断也会自动释放,调用方只需保证正常结束或出错时调用 + */ +export const acquireCdnSlot = async (host: string, signal?: AbortSignal) => { + signal?.throwIfAborted(); + const state = hosts.get(host) ?? { active: 0, queue: [] }; + hosts.set(host, state); + + /** 名额直接交给下一个等待者,active 不变 */ + const free = () => { + const next = state.queue.shift(); + if (next) next(); + else { + state.active -= 1; + if (state.active === 0) hosts.delete(host); + } + }; + + if (state.active < TORBOX_CDN_MAX_CONCURRENT) { + state.active += 1; + } else { + await new Promise((resolve, reject) => { + const entry = () => { + signal?.removeEventListener("abort", onAbort); + resolve(); + }; + const onAbort = () => { + const index = state.queue.indexOf(entry); + if (index >= 0) state.queue.splice(index, 1); + reject(signal!.reason); + }; + signal?.addEventListener("abort", onAbort, { once: true }); + state.queue.push(entry); + }); + } + + // 刚交到名额时客户端离开:监听不到已发生的中断,这里直接放弃 + if (signal?.aborted) { + free(); + throw signal.reason; + } + + let released = false; + const release = () => { + if (released) return; + released = true; + signal?.removeEventListener("abort", release); + free(); + }; + signal?.addEventListener("abort", release, { once: true }); + return release; +}; + +/** 响应体读完、出错或被取消后释放名额;没有响应体时立即释放 */ +export const releaseWhenDone = ( + body: ReadableStream | null, + release: () => void +) => { + if (!body) { + release(); + return null; + } + const reader = body.getReader(); + return new ReadableStream({ + async pull(controller) { + try { + const { done, value } = await reader.read(); + if (done) { + release(); + controller.close(); + } else controller.enqueue(value); + } catch (error) { + release(); + controller.error(error); + } + }, + cancel(reason) { + release(); + return reader.cancel(reason); + } + }); +}; diff --git a/src/server/torbox-limiter.ts b/src/server/torbox-limiter.ts index 27e74bc..091778d 100644 --- a/src/server/torbox-limiter.ts +++ b/src/server/torbox-limiter.ts @@ -53,6 +53,9 @@ const describeRule = (id: TorboxLimitId) => ({ ruleName: ruleName(id) }); +/** 请求的来源信息(如资源类型、资源 id),仅写入拒绝时的日志和事件,用来看是谁在耗额度 */ +export type TorboxRequestContext = Record; + /** 额度暂时用尽:路由据此返回 429 和 Retry-After */ export class RateLimitError extends Error { constructor( @@ -149,7 +152,7 @@ const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); * 在全部额度里各记一次;等待超过 TORBOX_MAX_WAIT_MS 的直接拒绝,不占额度。 * 驱动是同步的:检查与记录之间不会穿插其他请求 */ -const reserve = async (endpoint: string) => { +const reserve = async (endpoint: string, context?: TorboxRequestContext) => { const ids = getLimitIds(endpoint); const startedAt = Date.now(); const deadline = startedAt + TORBOX_MAX_WAIT_MS; @@ -188,6 +191,7 @@ const reserve = async (endpoint: string) => { const fields = { endpoint, + ...context, limitedBy, ...describeRule(blocking), usage: describeUsage(ids, now) @@ -205,7 +209,8 @@ const reserve = async (endpoint: string) => { limitedBy, rule: blocking, retryAfterMs: waitMs, - usage: fields.usage + usage: fields.usage, + ...(context ? { detail: context } : {}) }); throw new RateLimitError(blocking, waitMs, limitedBy); } @@ -246,10 +251,11 @@ const releaseSlot = () => { */ export const limitTorbox = async ( endpoint: string, - task: (queuedMs: number) => Promise + task: (queuedMs: number) => Promise, + context?: TorboxRequestContext ) => { const startedAt = Date.now(); - await reserve(endpoint); + await reserve(endpoint, context); await acquireSlot(); try { return await task(Date.now() - startedAt); diff --git a/src/server/torbox.ts b/src/server/torbox.ts index f2cad05..96db6a5 100644 --- a/src/server/torbox.ts +++ b/src/server/torbox.ts @@ -10,7 +10,8 @@ import { getRateLimitHeaders, limitTorbox, pauseTorbox, - RateLimitError + RateLimitError, + type TorboxRequestContext } from "@/server/torbox-limiter"; import { SOURCES, TORBOX_API } from "@/constants/torbox"; import { TORBOX_DOWNLOAD_URL_TTL_MS } from "@/constants/torbox-limits"; @@ -66,64 +67,78 @@ const apiKey = getTorboxApiKey; const log = createLogger("torbox"); /** 所有 TorBox 接口请求的唯一出口,经过统一限流 */ -const torboxFetch = (path: string, init: RequestInit = {}) => { +const torboxFetch = ( + path: string, + init: RequestInit = {}, + context?: TorboxRequestContext +) => { // 只记录接口路径:查询串里有资源 id,requestdl 的查询串还带着 API Key const endpoint = path.split("?")[0]; - return limitTorbox(endpoint, async (queuedMs) => { - const startedAt = Date.now(); - let res: Response; - try { - // body 为 FormData 时不能手动设置 Content-Type,fetch 会自动带上 boundary - res = await fetch(TORBOX_API + path, { - ...init, - headers: { - Authorization: `Bearer ${apiKey()}`, - Accept: "application/json" + return limitTorbox( + endpoint, + async (queuedMs) => { + const startedAt = Date.now(); + let res: Response; + try { + // body 为 FormData 时不能手动设置 Content-Type,fetch 会自动带上 boundary + res = await fetch(TORBOX_API + path, { + ...init, + headers: { + Authorization: `Bearer ${apiKey()}`, + Accept: "application/json" + }, + signal: AbortSignal.timeout(15000), + cache: "no-store" + }); + } catch (error) { + // 超时、网络错误:请求已发出,没有响应 + emitTorboxEvent({ + endpoint, + sent: true, + durationMs: Date.now() - startedAt, + queuedMs, + detail: { error: errorMessage(error) } + }); + throw error; + } + const durationMs = Date.now() - startedAt; + if (res.status === 429) + throw await pauseTorbox(endpoint, res, { durationMs, queuedMs }); + // 实际限制尚不明确:响应带了限流相关的头就记为 info,便于对照额度 + const rateLimitHeaders = getRateLimitHeaders(res.headers); + log[rateLimitHeaders ? "info" : "debug"]( + { + endpoint, + status: res.status, + durationMs, + queuedMs, + rateLimitHeaders }, - signal: AbortSignal.timeout(15000), - cache: "no-store" - }); - } catch (error) { - // 超时、网络错误:请求已发出,没有响应 + "TorBox 请求" + ); + const body = await res.json().catch(() => ({})); + const failed = !res.ok || body.success === false; emitTorboxEvent({ endpoint, sent: true, - durationMs: Date.now() - startedAt, + status: res.status, + durationMs, queuedMs, - detail: { error: errorMessage(error) } - }); - throw error; - } - const durationMs = Date.now() - startedAt; - if (res.status === 429) - throw await pauseTorbox(endpoint, res, { durationMs, queuedMs }); - // 实际限制尚不明确:响应带了限流相关的头就记为 info,便于对照额度 - const rateLimitHeaders = getRateLimitHeaders(res.headers); - log[rateLimitHeaders ? "info" : "debug"]( - { endpoint, status: res.status, durationMs, queuedMs, rateLimitHeaders }, - "TorBox 请求" - ); - const body = await res.json().catch(() => ({})); - const failed = !res.ok || body.success === false; - emitTorboxEvent({ - endpoint, - sent: true, - status: res.status, - durationMs, - queuedMs, - // TorBox 的业务错误码(如账号额度用尽)也可能是限流的信号 - ...(rateLimitHeaders || failed - ? { - detail: { - rateLimitHeaders, - ...(failed ? { error: body.error, message: body.detail } : {}) + // TorBox 的业务错误码(如账号额度用尽)也可能是限流的信号 + ...(rateLimitHeaders || failed + ? { + detail: { + rateLimitHeaders, + ...(failed ? { error: body.error, message: body.detail } : {}) + } } - } - : {}) - }); - if (failed) throw new Error(body.detail ?? res.statusText); - return body.data; - }); + : {}) + }); + if (failed) throw new Error(body.detail ?? res.statusText); + return body.data; + }, + context + ); }; export const isSourceType = (value: string): value is SourceType => @@ -305,7 +320,8 @@ const MOCK_PLAY_ERROR = const signDownloadUrl = async ( type: SourceType, id: string, - fileId: string + fileId: string, + context?: TorboxRequestContext ) => { const source = SOURCES.find((s) => s.type === type)!; const params = new URLSearchParams({ @@ -314,7 +330,7 @@ const signDownloadUrl = async ( file_id: fileId, redirect: "false" }); - const url = await torboxFetch(`/${type}/requestdl?${params}`); + const url = await torboxFetch(`/${type}/requestdl?${params}`, {}, context); if (typeof url !== "string" || !url) throw new Error("TorBox 未返回下载地址"); return url; }; @@ -373,6 +389,8 @@ const signing = (signingGlobals[SIGNING_KEY] ??= new Map()); type RequestDownloadUrlOptions = { /** 不用缓存、重新签发:缓存的直链被下载节点拒绝时使用 */ fresh?: boolean; + /** 本次请求的来源(如图片还是视频),额度用尽被拒绝时记入日志 */ + context?: TorboxRequestContext; }; /** @@ -391,7 +409,7 @@ export const requestDownloadUrl = async ( type: SourceType, id: string, fileId: string, - { fresh = false }: RequestDownloadUrlOptions = {} + { fresh = false, context }: RequestDownloadUrlOptions = {} ): Promise => { if (isMockMode()) throw new Error(MOCK_PLAY_ERROR); @@ -405,7 +423,7 @@ export const requestDownloadUrl = async ( const pending = signing.get(key); if (pending) return { ...(await pending), from: "shared" }; - const signed = signDownloadUrl(type, id, fileId) + const signed = signDownloadUrl(type, id, fileId, context) .then((url) => saveUrl(key, url, Date.now())) .finally(() => signing.delete(key)); signing.set(key, signed); diff --git a/src/types/torbox.ts b/src/types/torbox.ts index 8a544bf..9964e3e 100644 --- a/src/types/torbox.ts +++ b/src/types/torbox.ts @@ -58,7 +58,7 @@ export type LibraryResponse = { /** 当前受限的 TorBox 额度 */ export type TorboxLimitState = { id: TorboxLimitId; - /** 操作名,如「获取播放地址」 */ + /** 操作名,如「获取文件地址」 */ label: string; /** 多久(毫秒)后恢复 */ retryAfterMs: number;