fix: make online updates recoverable

server/update.ts

@@ -59,8 +59,22 @@ export type UrlPolicy = {
   /** When allowedHosts is omitted, requests are constrained to this URL's host. */
   baseUrl?: string | URL | undefined;
   maxRedirects?: number | undefined;
+  /** Maximum time allowed for one metadata/sidecar/archive request. */
+  timeoutMs?: number | undefined;
 };
 
+/** Release an unread response body before following a redirect or returning
+ * an error. Undici keeps the underlying connection associated with a body
+ * until it is consumed or cancelled; leaving it open can exhaust sockets when
+ * an update feed repeatedly returns errors or oversized responses. */
+async function cancelResponseBody(response: Response): Promise<void> {
+  try {
+    await response.body?.cancel();
+  } catch {
+    // The body may already be consumed/closed. Cancellation is best effort.
+  }
+}
+
 function invalidVersion(): never {
   throw new Error("更新版本号无效");
 }
@@ -157,8 +171,37 @@ function metadataError(): Error {
 }
 
 const DEFAULT_METADATA_MAX_BYTES = 2 * 1024 * 1024;
+/** Maximum time allowed for one update HTTP request, including its body. */
+export const DEFAULT_UPDATE_TIMEOUT_MS = 30_000;
 export const RELEASE_NOTES_MAX_BYTES = 64 * 1024;
 
+type UpdateFetchOptions = {
+  fetchImpl?: typeof fetch | undefined;
+  maxBytes?: number | undefined;
+  timeoutMs?: number | undefined;
+};
+
+function updateTimeoutMs(options: UpdateFetchOptions): number {
+  if (options.timeoutMs !== undefined) {
+    if (!Number.isSafeInteger(options.timeoutMs) || options.timeoutMs <= 0) throw new Error("更新请求超时配置无效");
+    return options.timeoutMs;
+  }
+  const configuredSeconds = process.env.TALLYNOTE_UPDATE_TIMEOUT_SECONDS;
+  if (configuredSeconds !== undefined && configuredSeconds.trim() !== "") {
+    const seconds = Number(configuredSeconds);
+    if (!Number.isSafeInteger(seconds) || seconds <= 0) throw new Error("TALLYNOTE_UPDATE_TIMEOUT_SECONDS 必须是大于 0 的整数");
+    return seconds * 1000;
+  }
+  return DEFAULT_UPDATE_TIMEOUT_MS;
+}
+
+function beginUpdateRequest(options: UpdateFetchOptions): { signal: AbortSignal; clear: () => void } {
+  const controller = new AbortController();
+  const timer = setTimeout(() => controller.abort(), updateTimeoutMs(options));
+  timer.unref?.();
+  return { signal: controller.signal, clear: () => clearTimeout(timer) };
+}
+
 function releaseNotesText(value: unknown): string | undefined {
   if (typeof value !== "string" || value.length === 0) return undefined;
   // Gitea exposes both Markdown (body/body_html) and releaseNotes depending on
@@ -210,20 +253,29 @@ function releaseResourceUrl(value: string, current: URL, options: UrlPolicy): st
 }
 
 /** Read a fetch body without ever buffering more than the caller's bound. */
-async function readBoundedResponse(response: Response, maxBytes: number, tooLargeMessage: string): Promise<Buffer> {
+async function readBoundedResponse(response: Response, maxBytes: number, tooLargeMessage: string, signal?: AbortSignal): Promise<Buffer> {
   if (!Number.isSafeInteger(maxBytes) || maxBytes <= 0) throw new Error("响应大小限制无效");
   const contentLength = response.headers.get("content-length");
   if (contentLength !== null) {
     const declared = Number(contentLength);
-    if (Number.isFinite(declared) && declared > maxBytes) throw new Error(tooLargeMessage);
+    if (Number.isFinite(declared) && declared > maxBytes) {
+      await cancelResponseBody(response);
+      throw new Error(tooLargeMessage);
+    }
   }
   if (!response.body) return Buffer.alloc(0);
   const reader = response.body.getReader();
   const chunks: Buffer[] = [];
   let total = 0;
+  let onAbort: (() => void) | undefined;
+  const abort = signal ? new Promise<never>((_, reject) => {
+    onAbort = () => reject(new Error("更新请求超时"));
+    if (signal.aborted) onAbort();
+    else signal.addEventListener("abort", onAbort, { once: true });
+  }) : undefined;
   try {
     for (;;) {
-      const result = await reader.read();
+      const result = await (abort ? Promise.race([reader.read(), abort]) : reader.read());
       if (result.done) break;
       const chunk = Buffer.from(result.value);
       if (chunk.length > maxBytes - total) {
@@ -233,7 +285,11 @@ async function readBoundedResponse(response: Response, maxBytes: number, tooLarg
       total += chunk.length;
       chunks.push(chunk);
     }
+  } catch (error) {
+    await reader.cancel().catch(() => undefined);
+    throw error;
   } finally {
+    if (signal && onAbort) signal.removeEventListener("abort", onAbort);
     reader.releaseLock();
   }
   return Buffer.concat(chunks, total);
@@ -241,84 +297,78 @@ async function readBoundedResponse(response: Response, maxBytes: number, tooLarg
 
 export async function fetchReleaseMetadata(
   metadataUrl: string | URL,
-  options: UrlPolicy & { fetchImpl?: typeof fetch | undefined; maxBytes?: number | undefined } = {},
+  options: UrlPolicy & UpdateFetchOptions = {},
 ): Promise<ReleaseMetadata> {
   const fetchImpl = options.fetchImpl ?? fetch;
   let current = validateHttpsUrl(metadataUrl, options);
   const maxRedirects = options.maxRedirects ?? 3;
   let response: Response;
   for (let redirects = 0; ; redirects += 1) {
+    const request = beginUpdateRequest(options);
     try {
-      response = await fetchImpl(current, { method: "GET", redirect: "manual", headers: { accept: "application/json" } });
+      response = await fetchImpl(current, { method: "GET", redirect: "manual", headers: { accept: "application/json" }, signal: request.signal });
     } catch {
+      request.clear();
       throw metadataError();
     }
-    if (response.status < 300 || response.status >= 400) break;
+    if (response.status < 300 || response.status >= 400) {
+      try {
+        if (response.status < 200 || response.status >= 300) {
+          await cancelResponseBody(response);
+          throw metadataError();
+        }
+        const maxBytes = Math.min(options.maxBytes ?? DEFAULT_METADATA_MAX_BYTES, DEFAULT_METADATA_MAX_BYTES);
+        const body = await readBoundedResponse(response, maxBytes, "更新发布信息过大", request.signal);
+        const payload: unknown = JSON.parse(body.toString("utf8"));
+        if (!payload || typeof payload !== "object") throw metadataError();
+        const item = payload as Record<string, unknown>;
+        const rawVersion = typeof item.version === "string" ? item.version : typeof item.tag_name === "string" ? item.tag_name : typeof item.tagName === "string" ? item.tagName : undefined;
+        if (!rawVersion) throw metadataError();
+        const version = parseSemver(rawVersion);
+        if (typeof item.tag_name === "string" && compareSemver(version, item.tag_name) !== 0) throw metadataError();
+        const assetsRaw = Array.isArray(item.assets) ? item.assets : [];
+        const assets: ReleaseAsset[] = [];
+        for (const raw of assetsRaw) {
+          if (!raw || typeof raw !== "object") continue;
+          const asset = raw as Record<string, unknown>;
+          const name = typeof asset.name === "string" ? asset.name : undefined;
+          const url = typeof asset.url === "string" ? asset.url : typeof asset.browser_download_url === "string" ? asset.browser_download_url : undefined;
+          if (!name || !url) continue;
+          let sha256: string | undefined;
+          const digest = typeof asset.sha256 === "string" ? asset.sha256 : typeof asset.digest === "string" ? asset.digest : undefined;
+          if (digest) {
+            const candidate = digest.replace(/^sha256:/i, "").toLowerCase();
+            if (/^[a-f0-9]{64}$/.test(candidate)) sha256 = candidate;
+          }
+          assets.push({ name, url: releaseResourceUrl(url, current, options), ...(sha256 ? { sha256 } : {}), ...(typeof asset.size === "number" && Number.isSafeInteger(asset.size) && asset.size >= 0 ? { size: asset.size } : {}) });
+        }
+        const notes = releaseNotesText(item.body ?? item.releaseNotes ?? item.release_notes ?? item.body_html);
+        const releaseName = releaseNameText(item.name ?? item.releaseName);
+        let releaseUrl: string | undefined;
+        if (typeof item.html_url === "string" || typeof item.url === "string") {
+          try { releaseUrl = releaseResourceUrl(typeof item.html_url === "string" ? item.html_url : item.url as string, current, options); } catch { /* optional */ }
+        }
+        return {
+          version: `${version.major}.${version.minor}.${version.patch}${version.prerelease.length ? `-${version.prerelease.join(".")}` : ""}${version.build.length ? `+${version.build.join(".")}` : ""}`,
+          ...(typeof item.tag_name === "string" ? { tagName: item.tag_name } : {}), ...(releaseName ? { releaseName } : {}), ...(typeof item.published_at === "string" ? { publishedAt: item.published_at } : {}), ...(notes ? { notes } : {}), ...(releaseUrl ? { releaseUrl } : {}), assets,
+        };
+      } catch { throw metadataError(); }
+      finally { request.clear(); }
+    }
+    await cancelResponseBody(response);
+    request.clear();
     if (redirects >= maxRedirects) throw metadataError();
     const location = response.headers.get("location");
     if (!location) throw metadataError();
     current = validateHttpsUrl(new URL(location, current), options.baseUrl ? options : { ...options, baseUrl: current });
   }
-  if (response.status < 200 || response.status >= 300) throw metadataError();
-  let payload: unknown;
-  try {
-    const maxBytes = Math.min(options.maxBytes ?? DEFAULT_METADATA_MAX_BYTES, DEFAULT_METADATA_MAX_BYTES);
-    const body = await readBoundedResponse(response, maxBytes, "更新发布信息过大");
-    payload = JSON.parse(body.toString("utf8"));
-  } catch { throw metadataError(); }
-  if (!payload || typeof payload !== "object") throw metadataError();
-  const item = payload as Record<string, unknown>;
-  const rawVersion = typeof item.version === "string" ? item.version : typeof item.tag_name === "string" ? item.tag_name : typeof item.tagName === "string" ? item.tagName : undefined;
-  if (!rawVersion) throw metadataError();
-  const version = parseSemver(rawVersion);
-  if (typeof item.tag_name === "string") {
-    try {
-      if (compareSemver(version, item.tag_name) !== 0) throw metadataError();
-    } catch {
-      throw metadataError();
-    }
-  }
-  const assetsRaw = Array.isArray(item.assets) ? item.assets : [];
-  const assets: ReleaseAsset[] = [];
-  for (const raw of assetsRaw) {
-    if (!raw || typeof raw !== "object") continue;
-    const asset = raw as Record<string, unknown>;
-    const name = typeof asset.name === "string" ? asset.name : undefined;
-    const url = typeof asset.url === "string" ? asset.url : typeof asset.browser_download_url === "string" ? asset.browser_download_url : undefined;
-    if (!name || !url) continue;
-    let sha256: string | undefined;
-    const digest = typeof asset.sha256 === "string" ? asset.sha256 : typeof asset.digest === "string" ? asset.digest : undefined;
-    if (digest) {
-      const candidate = digest.replace(/^sha256:/i, "").toLowerCase();
-      if (/^[a-f0-9]{64}$/.test(candidate)) sha256 = candidate;
-    }
-    assets.push({ name, url: releaseResourceUrl(url, current, options), ...(sha256 ? { sha256 } : {}), ...(typeof asset.size === "number" && Number.isSafeInteger(asset.size) && asset.size >= 0 ? { size: asset.size } : {}) });
-  }
-  const notes = releaseNotesText(item.body ?? item.releaseNotes ?? item.release_notes ?? item.body_html);
-  const releaseName = releaseNameText(item.name ?? item.releaseName);
-  let releaseUrl: string | undefined;
-  if (typeof item.html_url === "string" || typeof item.url === "string") {
-    try {
-      const candidate = typeof item.html_url === "string" ? item.html_url : item.url as string;
-      releaseUrl = releaseResourceUrl(candidate, current, options);
-    } catch { /* omit invalid optional release page URL */ }
-  }
-  return {
-    version: `${version.major}.${version.minor}.${version.patch}${version.prerelease.length ? `-${version.prerelease.join(".")}` : ""}${version.build.length ? `+${version.build.join(".")}` : ""}`,
-    ...(typeof item.tag_name === "string" ? { tagName: item.tag_name } : {}),
-    ...(releaseName ? { releaseName } : {}),
-    ...(typeof item.published_at === "string" ? { publishedAt: item.published_at } : {}),
-    ...(notes ? { notes } : {}),
-    ...(releaseUrl ? { releaseUrl } : {}),
-    assets,
-  };
 }
 
 /** Fetch a small text sidecar (for example SHA256SUMS) with the same
  * redirect, HTTPS and host policy used for release metadata. */
 export async function fetchReleaseText(
   textUrl: string | URL,
-  options: UrlPolicy & { fetchImpl?: typeof fetch | undefined; maxBytes?: number | undefined } = {},
+  options: UrlPolicy & UpdateFetchOptions = {},
 ): Promise<string> {
   const fetchImpl = options.fetchImpl ?? fetch;
   let current = validateHttpsUrl(textUrl, options);
@@ -328,27 +378,32 @@ export async function fetchReleaseText(
   const maxRedirects = options.maxRedirects ?? 3;
   let response: Response;
   for (let redirects = 0; ; redirects += 1) {
+    const request = beginUpdateRequest(options);
     try {
-      response = await fetchImpl(current, { method: "GET", redirect: "manual" });
+      response = await fetchImpl(current, { method: "GET", redirect: "manual", signal: request.signal });
     } catch {
+      request.clear();
       throw new Error("更新校验文件下载失败");
     }
-    if (response.status < 300 || response.status >= 400) break;
+    if (response.status < 300 || response.status >= 400) {
+      if (response.status < 200 || response.status >= 300) { await cancelResponseBody(response); request.clear(); throw new Error("更新校验文件下载失败"); }
+      const declared = Number(response.headers.get("content-length") ?? 0);
+      const maxBytes = options.maxBytes ?? 1024 * 1024;
+      if (declared > maxBytes) { await cancelResponseBody(response); request.clear(); throw new Error("更新校验文件过大"); }
+      try {
+        return (await readBoundedResponse(response, maxBytes, "更新校验文件过大", request.signal)).toString("utf8");
+      } catch (error) {
+        if (error instanceof Error && error.message === "更新校验文件过大") throw error;
+        throw new Error("更新校验文件下载失败");
+      } finally { request.clear(); }
+    }
+    await cancelResponseBody(response);
+    request.clear();
     if (redirects >= maxRedirects) throw new Error("更新校验文件下载失败");
     const location = response.headers.get("location");
     if (!location) throw new Error("更新校验文件下载失败");
     current = validateHttpsUrl(new URL(location, current), redirectPolicy);
   }
-  if (response.status < 200 || response.status >= 300) throw new Error("更新校验文件下载失败");
-  const declared = Number(response.headers.get("content-length") ?? 0);
-  const maxBytes = options.maxBytes ?? 1024 * 1024;
-  if (declared > maxBytes) throw new Error("更新校验文件过大");
-  try {
-    return (await readBoundedResponse(response, maxBytes, "更新校验文件过大")).toString("utf8");
-  } catch (error) {
-    if (error instanceof Error && error.message === "更新校验文件过大") throw error;
-    throw new Error("更新校验文件下载失败");
-  }
 }
 
 /** Fetch a bounded binary sidecar (for example an Ed25519 detached
@@ -356,7 +411,7 @@ export async function fetchReleaseText(
  * this separate from fetchReleaseText. */
 export async function fetchReleaseBytes(
   bytesUrl: string | URL,
-  options: UrlPolicy & { fetchImpl?: typeof fetch | undefined; maxBytes?: number | undefined } = {},
+  options: UrlPolicy & UpdateFetchOptions = {},
 ): Promise<Buffer> {
   const fetchImpl = options.fetchImpl ?? fetch;
   let current = validateHttpsUrl(bytesUrl, options);
@@ -366,27 +421,32 @@ export async function fetchReleaseBytes(
   const maxRedirects = options.maxRedirects ?? 3;
   let response: Response;
   for (let redirects = 0; ; redirects += 1) {
+    const request = beginUpdateRequest(options);
     try {
-      response = await fetchImpl(current, { method: "GET", redirect: "manual" });
+      response = await fetchImpl(current, { method: "GET", redirect: "manual", signal: request.signal });
     } catch {
+      request.clear();
       throw new Error("更新签名下载失败");
     }
-    if (response.status < 300 || response.status >= 400) break;
+    if (response.status < 300 || response.status >= 400) {
+      if (response.status < 200 || response.status >= 300) { await cancelResponseBody(response); request.clear(); throw new Error("更新签名下载失败"); }
+      const declared = Number(response.headers.get("content-length") ?? 0);
+      const maxBytes = options.maxBytes ?? 64 * 1024;
+      if (declared > maxBytes) { await cancelResponseBody(response); request.clear(); throw new Error("更新签名文件过大"); }
+      try {
+        return await readBoundedResponse(response, maxBytes, "更新签名文件过大", request.signal);
+      } catch (error) {
+        if (error instanceof Error && error.message === "更新签名文件过大") throw error;
+        throw new Error("更新签名下载失败");
+      } finally { request.clear(); }
+    }
+    await cancelResponseBody(response);
+    request.clear();
     if (redirects >= maxRedirects) throw new Error("更新签名下载失败");
     const location = response.headers.get("location");
     if (!location) throw new Error("更新签名下载失败");
     current = validateHttpsUrl(new URL(location, current), redirectPolicy);
   }
-  if (response.status < 200 || response.status >= 300) throw new Error("更新签名下载失败");
-  const declared = Number(response.headers.get("content-length") ?? 0);
-  const maxBytes = options.maxBytes ?? 64 * 1024;
-  if (declared > maxBytes) throw new Error("更新签名文件过大");
-  try {
-    return await readBoundedResponse(response, maxBytes, "更新签名文件过大");
-  } catch (error) {
-    if (error instanceof Error && error.message === "更新签名文件过大") throw error;
-    throw new Error("更新签名下载失败");
-  }
 }
 
 export function selectReleaseAsset(release: ReleaseMetadata, platform = detectPlatform(), runtimeHash?: string): ReleaseAsset | undefined {
@@ -438,7 +498,7 @@ export async function verifySha256(filePath: string, expected: string): Promise<
 export async function downloadReleaseAsset(
   url: string | URL,
   destination: string,
-  options: UrlPolicy & { fetchImpl?: typeof fetch | undefined; maxBytes?: number | undefined; onProgress?: ((downloadedBytes: number, totalBytes: number | null) => void) | undefined } = {},
+  options: UrlPolicy & UpdateFetchOptions & { onProgress?: ((downloadedBytes: number, totalBytes: number | null) => void) | undefined } = {},
 ): Promise<{ size: number; sha256: string }> {
   const fetchImpl = options.fetchImpl ?? fetch;
   let current = validateHttpsUrl(url, options);
@@ -448,22 +508,32 @@ export async function downloadReleaseAsset(
   const maxRedirects = options.maxRedirects ?? 3;
   let response: Response;
   for (let redirects = 0; ; redirects += 1) {
+    const request = beginUpdateRequest(options);
     try {
-      response = await fetchImpl(current, { method: "GET", redirect: "manual" });
+      response = await fetchImpl(current, { method: "GET", redirect: "manual", signal: request.signal });
     } catch {
+      request.clear();
       throw new Error("更新文件下载失败");
     }
-    if (response.status < 300 || response.status >= 400) break;
+    if (response.status < 300 || response.status >= 400) { request.clear(); break; }
+    await cancelResponseBody(response);
+    request.clear();
     if (redirects >= maxRedirects) throw new Error("更新文件下载失败");
     const location = response.headers.get("location");
     if (!location) throw new Error("更新文件下载失败");
     current = validateHttpsUrl(new URL(location, current), redirectPolicy);
   }
-  if (response.status < 200 || response.status >= 300 || !response.body) throw new Error("更新文件下载失败");
+  if (response.status < 200 || response.status >= 300 || !response.body) {
+    await cancelResponseBody(response);
+    throw new Error("更新文件下载失败");
+  }
   const declared = Number(response.headers.get("content-length") ?? 0);
   const totalBytes = Number.isSafeInteger(declared) && declared > 0 ? declared : null;
   const maxBytes = options.maxBytes ?? 512 * 1024 * 1024;
-  if (declared > maxBytes) throw new Error("更新文件超过大小限制");
+  if (declared > maxBytes) {
+    await cancelResponseBody(response);
+    throw new Error("更新文件超过大小限制");
+  }
   await mkdir(path.dirname(destination), { recursive: true, mode: 0o700 });
   const temporary = `${destination}.part-${randomUUID()}`;
   let size = 0;
@@ -475,8 +545,10 @@ export async function downloadReleaseAsset(
     hash.update(chunk);
     callback(null, chunk);
   } });
+  const request = beginUpdateRequest(options);
   try {
-    await pipeline(Readable.fromWeb(response.body as import("node:stream/web").ReadableStream), meter, createWriteStream(temporary, { flags: "wx", mode: 0o600 }));
+    const source = Readable.fromWeb(response.body as import("node:stream/web").ReadableStream, { signal: request.signal });
+    await pipeline(source, meter, createWriteStream(temporary, { flags: "wx", mode: 0o600 }));
     const fd = await open(temporary, "r");
     await fd.sync();
     await fd.close();
@@ -484,6 +556,8 @@ export async function downloadReleaseAsset(
   } catch (error) {
     await import("node:fs/promises").then(({ rm }) => rm(temporary, { force: true })).catch(() => undefined);
     throw error instanceof Error && error.message.startsWith("更新文件") ? error : new Error("更新文件下载失败");
+  } finally {
+    request.clear();
   }
   return { size, sha256: hash.digest("hex") };
 }