feat: 调整限流器和并发

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 }, "播放失败");

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<string, TorboxLimitRule>;
@@ -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;
 

src/hooks/use-limit-retry.ts

@@ -25,7 +25,7 @@ const fetchLimits = async () => {
 };
 
 export type LimitWait = {
-  /** 受限的操作名，如「获取播放地址」 */
+  /** 受限的操作名，如「获取文件地址」 */
   label: string;
   /** 恢复时间，本机毫秒时间戳 */
   until: number;

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<string, HostState> };
+const hosts = (globals[HOSTS_KEY] ??= new Map<string, HostState>());
+
+/**
+ * 占用该节点的一个并发名额，返回释放函数（可重复调用）。
+ * 排队期间客户端离开（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<void>((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);
+    }
+  });
+};

src/server/torbox-limiter.ts

@@ -53,6 +53,9 @@ const describeRule = (id: TorboxLimitId) => ({
   ruleName: ruleName(id)
 });
 
+/** 请求的来源信息（如资源类型、资源 id），仅写入拒绝时的日志和事件，用来看是谁在耗额度 */
+export type TorboxRequestContext = Record<string, unknown>;
+
 /** 额度暂时用尽：路由据此返回 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 <T>(
   endpoint: string,
-  task: (queuedMs: number) => Promise<T>
+  task: (queuedMs: number) => Promise<T>,
+  context?: TorboxRequestContext
 ) => {
   const startedAt = Date.now();
-  await reserve(endpoint);
+  await reserve(endpoint, context);
   await acquireSlot();
   try {
     return await task(Date.now() - startedAt);

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<SignedUrl & { from: DownloadUrlFrom }> => {
   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);

src/types/torbox.ts

@@ -58,7 +58,7 @@ export type LibraryResponse = {
 /** 当前受限的 TorBox 额度 */
 export type TorboxLimitState = {
   id: TorboxLimitId;
-  /** 操作名，如「获取播放地址」 */
+  /** 操作名，如「获取文件地址」 */
   label: string;
   /** 多久（毫秒）后恢复 */
   retryAfterMs: number;