feat: Download System v2 — complete rewrite of download pipeline
Replace monolithic download-manager.ts (9500 lines) with 7 focused modules: - error-classifier.ts: 25+ typed DownloadErrorKind enum, classifier functions for network/HTTP/debrid/extraction errors — no more string matching - retry-manager.ts: Declarative per-error-kind retry policies, exponential backoff, shelving after 15 failures, state export/import - stream-writer.ts: HTTP stream → file with pre-resume validation, stall detection, NTFS-aligned buffered writing, Range-ignored detection - pipeline.ts: Single download lifecycle (unrestrict → stream → verify), throws typed errors, caller decides retry strategy - post-processor.ts: Extraction state machine with hard caps (3 attempts per archive, 5 rounds per package), no infinite loops - scheduler.ts: Queue management with priority-based slot allocation, heartbeat stall detection, global watchdog, provider cooldowns - download-manager.ts: Drop-in orchestrator (~1500 lines), same public API Fixes: 1. Hanging downloads: heartbeat-based stall detection + global watchdog 2. Wrong error classification: typed enum at point of origin 3. Unreliable resume: file size vs tracker validation, Range-ignored detection 4. Extraction loops: bounded retries with state machine 215 new unit tests for error-classifier and retry-manager (all passing). Build compiles cleanly. Same IPC interface — UI unchanged. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Sucukdeluxe
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
63b412a43f
commit
efa0909e11
@@ -21,7 +21,7 @@ import {
|
||||
import { resetDebridLinkApiKeyDailyUsage, resetProviderDailyUsage } from "../shared/provider-daily-limits";
|
||||
import { importDlcContainers } from "./container";
|
||||
import { APP_VERSION } from "./constants";
|
||||
import { DownloadManager } from "./download-manager";
|
||||
import { DownloadManager } from "./download/download-manager";
|
||||
import { fetchAllDebridHostInfo, fetchDebridLinkHostLimits } from "./debrid";
|
||||
import { parseCollectorInput } from "./link-parser";
|
||||
import { configureLogger, getLogFilePath, logger } from "./logger";
|
||||
|
||||
@@ -2,7 +2,7 @@ import http from "node:http";
|
||||
import fs from "node:fs";
|
||||
import path from "node:path";
|
||||
import { logger, getLogFilePath } from "./logger";
|
||||
import type { DownloadManager } from "./download-manager";
|
||||
import type { DownloadManager } from "./download/download-manager";
|
||||
|
||||
const DEFAULT_PORT = 9868;
|
||||
const MAX_LOG_LINES = 10000;
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,508 @@
|
||||
/**
|
||||
* error-classifier.ts — Typed error system for download pipeline.
|
||||
*
|
||||
* Every error gets classified ONCE at the point of origin into a
|
||||
* DownloadErrorKind. No post-hoc string matching needed downstream.
|
||||
*/
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Error Kinds
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export enum DownloadErrorKind {
|
||||
// Network
|
||||
NetworkReset = "network_reset",
|
||||
Timeout = "timeout",
|
||||
DnsFailure = "dns_failure",
|
||||
ConnectTimeout = "connect_timeout",
|
||||
|
||||
// HTTP
|
||||
RangeNotSatisfied = "range_not_satisfied",
|
||||
RangeIgnored = "range_ignored",
|
||||
ServerError = "server_error",
|
||||
RateLimited = "rate_limited",
|
||||
Forbidden = "forbidden",
|
||||
NotFound = "not_found",
|
||||
|
||||
// Provider / Debrid
|
||||
UnrestrictFailed = "unrestrict_failed",
|
||||
ProviderBusy = "provider_busy",
|
||||
ProviderDown = "provider_down",
|
||||
HosterUnavailable = "hoster_unavailable",
|
||||
LinkDead = "link_dead",
|
||||
QuotaExceeded = "quota_exceeded",
|
||||
|
||||
// Filesystem
|
||||
DiskFull = "disk_full",
|
||||
PermissionDenied = "permission_denied",
|
||||
FileLocked = "file_locked",
|
||||
|
||||
// Integrity / Resume
|
||||
FileCorrupt = "file_corrupt",
|
||||
FileTruncated = "file_truncated",
|
||||
ResumeUnderflow = "resume_underflow",
|
||||
|
||||
// Extraction
|
||||
WrongPassword = "wrong_password",
|
||||
ArchiveCorrupt = "archive_corrupt",
|
||||
ExtractorCrash = "extractor_crash",
|
||||
|
||||
// Write / Drain
|
||||
WriteDrainTimeout = "write_drain_timeout",
|
||||
|
||||
// Catchall
|
||||
Unknown = "unknown",
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Permanent kinds — retrying is pointless
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const PERMANENT_KINDS = new Set<DownloadErrorKind>([
|
||||
DownloadErrorKind.LinkDead,
|
||||
DownloadErrorKind.DiskFull,
|
||||
DownloadErrorKind.PermissionDenied,
|
||||
DownloadErrorKind.WrongPassword,
|
||||
]);
|
||||
|
||||
export function isPermanentKind(kind: DownloadErrorKind): boolean {
|
||||
return PERMANENT_KINDS.has(kind);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// DownloadError class
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export class DownloadError extends Error {
|
||||
readonly kind: DownloadErrorKind;
|
||||
readonly retryable: boolean;
|
||||
readonly permanent: boolean;
|
||||
readonly httpStatus?: number;
|
||||
readonly originalError?: Error;
|
||||
/** Extra context (e.g. existing bytes, expected total). */
|
||||
readonly context?: Record<string, unknown>;
|
||||
|
||||
constructor(
|
||||
kind: DownloadErrorKind,
|
||||
message: string,
|
||||
opts?: {
|
||||
httpStatus?: number;
|
||||
originalError?: Error;
|
||||
retryable?: boolean;
|
||||
permanent?: boolean;
|
||||
context?: Record<string, unknown>;
|
||||
},
|
||||
) {
|
||||
super(message);
|
||||
this.name = "DownloadError";
|
||||
this.kind = kind;
|
||||
this.retryable = opts?.retryable ?? !isPermanentKind(kind);
|
||||
this.permanent = opts?.permanent ?? isPermanentKind(kind);
|
||||
this.httpStatus = opts?.httpStatus;
|
||||
this.originalError = opts?.originalError;
|
||||
this.context = opts?.context;
|
||||
}
|
||||
|
||||
/** Compact single-line representation for logging. */
|
||||
toLogString(): string {
|
||||
const parts = [`[${this.kind}]`, this.message];
|
||||
if (this.httpStatus) parts.push(`(HTTP ${this.httpStatus})`);
|
||||
return parts.join(" ");
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Classifier: raw fetch / network errors
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export function classifyFetchError(error: unknown): DownloadError {
|
||||
const text = errorText(error);
|
||||
const lc = text.toLowerCase();
|
||||
|
||||
// Abort is not an error to classify — re-throw as-is
|
||||
if (lc.includes("aborted:") || lc.includes("abort")) {
|
||||
// Preserve abort errors unchanged so callers can check abortReason
|
||||
throw error instanceof Error ? error : new Error(text);
|
||||
}
|
||||
|
||||
// Connection timeout
|
||||
if (lc.includes("connect_timeout") || lc.includes("etimedout") || lc.includes("connection timed out")) {
|
||||
return new DownloadError(DownloadErrorKind.ConnectTimeout, text, {
|
||||
originalError: toError(error),
|
||||
});
|
||||
}
|
||||
|
||||
// DNS
|
||||
if (lc.includes("enotfound") || lc.includes("getaddrinfo") || lc.includes("dns")) {
|
||||
return new DownloadError(DownloadErrorKind.DnsFailure, text, {
|
||||
originalError: toError(error),
|
||||
});
|
||||
}
|
||||
|
||||
// Network reset
|
||||
if (
|
||||
lc.includes("fetch failed") ||
|
||||
lc.includes("socket hang up") ||
|
||||
lc.includes("econnreset") ||
|
||||
lc.includes("econnrefused") ||
|
||||
lc.includes("epipe") ||
|
||||
lc.includes("network error") ||
|
||||
lc.includes("econnaborted") ||
|
||||
lc.includes("socket closed") ||
|
||||
lc.includes("connection reset")
|
||||
) {
|
||||
return new DownloadError(DownloadErrorKind.NetworkReset, text, {
|
||||
originalError: toError(error),
|
||||
});
|
||||
}
|
||||
|
||||
// Stall / read timeout
|
||||
if (lc.includes("stall_timeout") || lc.includes("read timeout")) {
|
||||
return new DownloadError(DownloadErrorKind.Timeout, text, {
|
||||
originalError: toError(error),
|
||||
});
|
||||
}
|
||||
|
||||
// Write drain timeout
|
||||
if (lc.includes("write_drain_timeout")) {
|
||||
return new DownloadError(DownloadErrorKind.WriteDrainTimeout, text, {
|
||||
originalError: toError(error),
|
||||
});
|
||||
}
|
||||
|
||||
// Disk full
|
||||
if (lc.includes("enospc") || lc.includes("no space left")) {
|
||||
return new DownloadError(DownloadErrorKind.DiskFull, text, {
|
||||
originalError: toError(error),
|
||||
permanent: true,
|
||||
});
|
||||
}
|
||||
|
||||
// Permission denied
|
||||
if (lc.includes("eacces") || lc.includes("eperm") || lc.includes("permission denied")) {
|
||||
return new DownloadError(DownloadErrorKind.PermissionDenied, text, {
|
||||
originalError: toError(error),
|
||||
permanent: true,
|
||||
});
|
||||
}
|
||||
|
||||
// File locked (Windows)
|
||||
if (lc.includes("ebusy") || lc.includes("file is locked") || lc.includes("being used by another process")) {
|
||||
return new DownloadError(DownloadErrorKind.FileLocked, text, {
|
||||
originalError: toError(error),
|
||||
});
|
||||
}
|
||||
|
||||
// Resume underflow
|
||||
if (lc.startsWith("resume_download_underflow:")) {
|
||||
return new DownloadError(DownloadErrorKind.ResumeUnderflow, text, {
|
||||
originalError: toError(error),
|
||||
});
|
||||
}
|
||||
|
||||
// Range ignored on resume
|
||||
if (lc.startsWith("range_ignored_on_resume:")) {
|
||||
return new DownloadError(DownloadErrorKind.RangeIgnored, text, {
|
||||
originalError: toError(error),
|
||||
});
|
||||
}
|
||||
|
||||
return new DownloadError(DownloadErrorKind.Unknown, text, {
|
||||
originalError: toError(error),
|
||||
});
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Classifier: HTTP response status codes
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export interface HttpClassifyContext {
|
||||
status: number;
|
||||
statusText?: string;
|
||||
responseText?: string;
|
||||
existingBytes?: number;
|
||||
rangeHeaderSent?: boolean;
|
||||
}
|
||||
|
||||
export function classifyHttpStatus(ctx: HttpClassifyContext): DownloadError {
|
||||
const { status, statusText, responseText } = ctx;
|
||||
const body = responseText || statusText || "";
|
||||
const msg = `HTTP ${status}${body ? ": " + compactText(body) : ""}`;
|
||||
|
||||
switch (true) {
|
||||
case status === 416:
|
||||
return new DownloadError(DownloadErrorKind.RangeNotSatisfied, msg, {
|
||||
httpStatus: status,
|
||||
context: { existingBytes: ctx.existingBytes },
|
||||
});
|
||||
|
||||
case status === 429:
|
||||
return new DownloadError(DownloadErrorKind.RateLimited, msg, {
|
||||
httpStatus: status,
|
||||
});
|
||||
|
||||
case status === 403:
|
||||
return new DownloadError(DownloadErrorKind.Forbidden, msg, {
|
||||
httpStatus: status,
|
||||
});
|
||||
|
||||
case status === 404:
|
||||
return new DownloadError(DownloadErrorKind.NotFound, msg, {
|
||||
httpStatus: status,
|
||||
});
|
||||
|
||||
case status >= 500:
|
||||
return new DownloadError(DownloadErrorKind.ServerError, msg, {
|
||||
httpStatus: status,
|
||||
});
|
||||
|
||||
default:
|
||||
return new DownloadError(DownloadErrorKind.Unknown, msg, {
|
||||
httpStatus: status,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Detect when the server ignored a Range header (sent 200 instead of 206).
|
||||
* Call this AFTER receiving a 200 response when a Range header was sent.
|
||||
*/
|
||||
export function classifyRangeIgnored(
|
||||
existingBytes: number,
|
||||
contentLength: number,
|
||||
): DownloadError {
|
||||
return new DownloadError(
|
||||
DownloadErrorKind.RangeIgnored,
|
||||
`range_ignored_on_resume:${existingBytes}/${contentLength}`,
|
||||
{ context: { existingBytes, contentLength } },
|
||||
);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Classifier: unrestrict / debrid API errors
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export function classifyUnrestrictError(error: unknown): DownloadError {
|
||||
const text = errorText(error);
|
||||
const lc = text.toLowerCase();
|
||||
|
||||
// Permanent: file is dead
|
||||
if (
|
||||
lc.includes("permanent ungültig") ||
|
||||
/file.?not.?found/.test(lc) ||
|
||||
/file.?unavailable/.test(lc) ||
|
||||
/link.?is.?dead/.test(lc) ||
|
||||
lc.includes("file has been removed") ||
|
||||
lc.includes("file has been deleted") ||
|
||||
lc.includes("file is no longer available") ||
|
||||
lc.includes("file was removed") ||
|
||||
lc.includes("file was deleted")
|
||||
) {
|
||||
return new DownloadError(DownloadErrorKind.LinkDead, text, {
|
||||
originalError: toError(error),
|
||||
permanent: true,
|
||||
});
|
||||
}
|
||||
|
||||
// Provider busy / concurrent limit
|
||||
if (
|
||||
lc.includes("too many active") ||
|
||||
lc.includes("too many concurrent") ||
|
||||
lc.includes("too many downloads") ||
|
||||
lc.includes("active download") ||
|
||||
lc.includes("concurrent limit") ||
|
||||
lc.includes("slot limit") ||
|
||||
lc.includes("limit reached") ||
|
||||
lc.includes("zu viele aktive") ||
|
||||
lc.includes("zu viele gleichzeitige") ||
|
||||
lc.includes("zu viele downloads")
|
||||
) {
|
||||
return new DownloadError(DownloadErrorKind.ProviderBusy, text, {
|
||||
originalError: toError(error),
|
||||
});
|
||||
}
|
||||
|
||||
// Hoster unavailable
|
||||
if (lc.includes("hosternotavailable")) {
|
||||
return new DownloadError(DownloadErrorKind.HosterUnavailable, text, {
|
||||
originalError: toError(error),
|
||||
});
|
||||
}
|
||||
|
||||
// Quota / traffic exceeded
|
||||
if (
|
||||
lc.includes("quota") ||
|
||||
lc.includes("traffic") ||
|
||||
lc.includes("bandwidth limit") ||
|
||||
lc.includes("daily limit")
|
||||
) {
|
||||
return new DownloadError(DownloadErrorKind.QuotaExceeded, text, {
|
||||
originalError: toError(error),
|
||||
});
|
||||
}
|
||||
|
||||
// Provider temporarily down
|
||||
if (
|
||||
lc.includes("server error") ||
|
||||
lc.includes("internal server error") ||
|
||||
lc.includes("temporarily unavailable") ||
|
||||
lc.includes("temporary unavailable") ||
|
||||
lc.includes("temporarily disabled") ||
|
||||
lc.includes("try again later") ||
|
||||
lc.includes("service unavailable") ||
|
||||
lc.includes("host is down") ||
|
||||
lc.includes("maintenance") ||
|
||||
lc.includes("bad gateway") ||
|
||||
lc.includes("gateway timeout") ||
|
||||
lc.includes("cloudflare") ||
|
||||
lc.includes("worker error")
|
||||
) {
|
||||
return new DownloadError(DownloadErrorKind.ProviderDown, text, {
|
||||
originalError: toError(error),
|
||||
});
|
||||
}
|
||||
|
||||
// Generic unrestrict failure (session, login, etc.)
|
||||
if (
|
||||
lc.includes("unrestrict") ||
|
||||
lc.includes("mega-web") ||
|
||||
lc.includes("mega-debrid") ||
|
||||
lc.includes("bestdebrid") ||
|
||||
lc.includes("alldebrid") ||
|
||||
lc.includes("kein debrid") ||
|
||||
lc.includes("session-cookie") ||
|
||||
lc.includes("session cookie") ||
|
||||
lc.includes("session blockiert") ||
|
||||
lc.includes("session expired") ||
|
||||
lc.includes("invalid session") ||
|
||||
lc.includes("login ungültig") ||
|
||||
lc.includes("login liefert") ||
|
||||
lc.includes("login required") ||
|
||||
lc.includes("login failed")
|
||||
) {
|
||||
return new DownloadError(DownloadErrorKind.UnrestrictFailed, text, {
|
||||
originalError: toError(error),
|
||||
});
|
||||
}
|
||||
|
||||
return new DownloadError(DownloadErrorKind.Unknown, text, {
|
||||
originalError: toError(error),
|
||||
});
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Classifier: extraction errors
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export function classifyExtractionError(
|
||||
errorText_: string,
|
||||
category?: string,
|
||||
): DownloadError {
|
||||
const lc = (errorText_ || "").toLowerCase();
|
||||
|
||||
if (lc.includes("wrong password") || lc.includes("falsches passwort") || category === "wrong_password") {
|
||||
return new DownloadError(DownloadErrorKind.WrongPassword, errorText_, {
|
||||
permanent: true,
|
||||
});
|
||||
}
|
||||
|
||||
if (
|
||||
lc.includes("corrupt") ||
|
||||
lc.includes("unexpected end") ||
|
||||
lc.includes("broken header") ||
|
||||
lc.includes("invalid archive") ||
|
||||
lc.includes("bad signature") ||
|
||||
lc.includes("beschädigt") ||
|
||||
category === "archive_corrupt"
|
||||
) {
|
||||
return new DownloadError(DownloadErrorKind.ArchiveCorrupt, errorText_);
|
||||
}
|
||||
|
||||
if (
|
||||
lc.includes("process exited") ||
|
||||
lc.includes("process crashed") ||
|
||||
lc.includes("extractor failed") ||
|
||||
lc.includes("segmentation fault") ||
|
||||
category === "extractor_crash"
|
||||
) {
|
||||
return new DownloadError(DownloadErrorKind.ExtractorCrash, errorText_);
|
||||
}
|
||||
|
||||
if (lc.includes("enospc") || lc.includes("no space left")) {
|
||||
return new DownloadError(DownloadErrorKind.DiskFull, errorText_, {
|
||||
permanent: true,
|
||||
});
|
||||
}
|
||||
|
||||
return new DownloadError(DownloadErrorKind.Unknown, errorText_);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Convenience: wrap any unknown error into a DownloadError
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Ensure any thrown value becomes a DownloadError.
|
||||
* If already a DownloadError, return as-is.
|
||||
*/
|
||||
export function ensureDownloadError(error: unknown): DownloadError {
|
||||
if (error instanceof DownloadError) return error;
|
||||
return classifyFetchError(error);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Human-readable error messages for UI
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const KIND_LABELS: Record<DownloadErrorKind, string> = {
|
||||
[DownloadErrorKind.NetworkReset]: "Netzwerkfehler",
|
||||
[DownloadErrorKind.Timeout]: "Zeitüberschreitung",
|
||||
[DownloadErrorKind.DnsFailure]: "DNS-Fehler",
|
||||
[DownloadErrorKind.ConnectTimeout]: "Verbindungs-Timeout",
|
||||
[DownloadErrorKind.RangeNotSatisfied]: "Range-Konflikt (HTTP 416)",
|
||||
[DownloadErrorKind.RangeIgnored]: "Server ignorierte Resume",
|
||||
[DownloadErrorKind.ServerError]: "Serverfehler",
|
||||
[DownloadErrorKind.RateLimited]: "Rate-Limit erreicht",
|
||||
[DownloadErrorKind.Forbidden]: "Zugriff verweigert",
|
||||
[DownloadErrorKind.NotFound]: "Nicht gefunden",
|
||||
[DownloadErrorKind.UnrestrictFailed]: "Unrestrict fehlgeschlagen",
|
||||
[DownloadErrorKind.ProviderBusy]: "Provider ausgelastet",
|
||||
[DownloadErrorKind.ProviderDown]: "Provider nicht erreichbar",
|
||||
[DownloadErrorKind.HosterUnavailable]: "Hoster nicht verfügbar",
|
||||
[DownloadErrorKind.LinkDead]: "Link ungültig / gelöscht",
|
||||
[DownloadErrorKind.QuotaExceeded]: "Tages-Limit erreicht",
|
||||
[DownloadErrorKind.DiskFull]: "Festplatte voll",
|
||||
[DownloadErrorKind.PermissionDenied]: "Zugriff verweigert (Dateisystem)",
|
||||
[DownloadErrorKind.FileLocked]: "Datei gesperrt",
|
||||
[DownloadErrorKind.FileCorrupt]: "Datei beschädigt (CRC-Fehler)",
|
||||
[DownloadErrorKind.FileTruncated]: "Download unvollständig",
|
||||
[DownloadErrorKind.ResumeUnderflow]: "Resume-Fehler",
|
||||
[DownloadErrorKind.WrongPassword]: "Falsches Archiv-Passwort",
|
||||
[DownloadErrorKind.ArchiveCorrupt]: "Archiv beschädigt",
|
||||
[DownloadErrorKind.ExtractorCrash]: "Entpacker abgestürzt",
|
||||
[DownloadErrorKind.WriteDrainTimeout]: "Schreibvorgang blockiert",
|
||||
[DownloadErrorKind.Unknown]: "Unbekannter Fehler",
|
||||
};
|
||||
|
||||
export function errorKindLabel(kind: DownloadErrorKind): string {
|
||||
return KIND_LABELS[kind] || KIND_LABELS[DownloadErrorKind.Unknown];
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Internal helpers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
function errorText(e: unknown): string {
|
||||
if (typeof e === "string") return e;
|
||||
if (e instanceof Error) return e.message || String(e);
|
||||
return String(e ?? "");
|
||||
}
|
||||
|
||||
function toError(e: unknown): Error {
|
||||
if (e instanceof Error) return e;
|
||||
return new Error(String(e ?? ""));
|
||||
}
|
||||
|
||||
function compactText(s: string): string {
|
||||
return s.replace(/\s+/g, " ").trim().slice(0, 200);
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
/**
|
||||
* Download system v2 — public re-exports.
|
||||
*/
|
||||
|
||||
export { DownloadManager } from "./download-manager";
|
||||
export type { DownloadManagerOptions } from "./download-manager";
|
||||
export { DownloadError, DownloadErrorKind, errorKindLabel } from "./error-classifier";
|
||||
@@ -0,0 +1,314 @@
|
||||
/**
|
||||
* pipeline.ts — Single download lifecycle: unrestrict → stream → verify.
|
||||
*
|
||||
* The pipeline runs ONE download attempt. It does NOT handle retries —
|
||||
* the caller (download-manager + retry-manager) decides what to do with errors.
|
||||
* All errors thrown are typed DownloadErrors.
|
||||
*/
|
||||
|
||||
import fs from "node:fs";
|
||||
import path from "node:path";
|
||||
import { DownloadError, DownloadErrorKind, classifyUnrestrictError, classifyFetchError, ensureDownloadError } from "./error-classifier";
|
||||
import { streamToFile, type StreamResult } from "./stream-writer";
|
||||
import type { DownloadItem, PackageEntry, AppSettings, DebridProvider } from "../../shared/types";
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Types
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/** Unrestricted link result from debrid service. */
|
||||
export interface UnrestrictedLink {
|
||||
fileName: string;
|
||||
directUrl: string;
|
||||
fileSize: number | null;
|
||||
retriesUsed: number;
|
||||
skipTlsVerify?: boolean;
|
||||
provider: DebridProvider;
|
||||
providerLabel?: string;
|
||||
sourceAccountId?: string;
|
||||
sourceAccountLabel?: string;
|
||||
}
|
||||
|
||||
/** Debrid service interface — the pipeline only needs unrestrict. */
|
||||
export interface DebridUnrestrictor {
|
||||
unrestrictLink(url: string, signal: AbortSignal): Promise<UnrestrictedLink>;
|
||||
}
|
||||
|
||||
/** Integrity checker interface. */
|
||||
export interface IntegrityChecker {
|
||||
validateFile(filePath: string, packageDir: string): Promise<{ ok: boolean; message: string }>;
|
||||
}
|
||||
|
||||
export interface PipelineContext {
|
||||
item: DownloadItem;
|
||||
pkg: PackageEntry;
|
||||
settings: AppSettings;
|
||||
debridService: DebridUnrestrictor;
|
||||
integrityChecker?: IntegrityChecker;
|
||||
signal: AbortSignal;
|
||||
/** Reuse direct URL from previous attempt (skip unrestrict). */
|
||||
cachedDirectUrl?: string;
|
||||
cachedProvider?: DebridProvider;
|
||||
cachedProviderLabel?: string;
|
||||
cachedSkipTls?: boolean;
|
||||
|
||||
// Callbacks
|
||||
onStatus: (status: string, fullStatus: string) => void;
|
||||
onProgress: (downloadedBytes: number, totalBytes: number | null, speedBps: number) => void;
|
||||
onResumable: (resumable: boolean) => void;
|
||||
onFileNameOverride: (newName: string, newTargetPath: string) => void;
|
||||
onProviderInfo: (provider: DebridProvider, label?: string, accountId?: string, accountLabel?: string) => void;
|
||||
onHeartbeat: () => void;
|
||||
onDiskBusy?: (busy: boolean) => void;
|
||||
onLog: (level: "INFO" | "WARN" | "ERROR", message: string, fields?: Record<string, unknown>) => void;
|
||||
|
||||
// Path management
|
||||
claimTargetPath: (itemId: string, preferredPath: string, keepExisting?: boolean) => string;
|
||||
releaseTargetPath: (itemId: string) => void;
|
||||
}
|
||||
|
||||
export interface PipelineResult {
|
||||
success: boolean;
|
||||
downloadedBytes: number;
|
||||
totalBytes: number | null;
|
||||
directUrl: string;
|
||||
provider: DebridProvider;
|
||||
providerLabel?: string;
|
||||
resumable: boolean;
|
||||
skipTlsVerify?: boolean;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Configuration
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const DEFAULT_STALL_TIMEOUT_MS = 10_000;
|
||||
const DEFAULT_CONNECT_TIMEOUT_MS = 25_000;
|
||||
const DEFAULT_UNRESTRICT_TIMEOUT_MS = 60_000;
|
||||
const DEFAULT_LOW_THROUGHPUT_TIMEOUT_MS = 120_000;
|
||||
const DEFAULT_LOW_THROUGHPUT_MIN_BYTES = 64 * 1024;
|
||||
|
||||
function getEnvMs(name: string, defaultMs: number): number {
|
||||
const val = process.env[name];
|
||||
if (!val) return defaultMs;
|
||||
const n = Number(val);
|
||||
return Number.isFinite(n) && n >= 0 ? n : defaultMs;
|
||||
}
|
||||
|
||||
const LARGE_BINARY_RE = /\.(?:part\d+\.rar|rar|r\d{2,3}|zip(?:\.\d+)?|7z(?:\.\d+)?|tar|gz|bz2|xz|iso|mkv|mp4|avi|mov|wmv|m4v|ts|m2ts|webm|mp3|flac|aac|wav)$/i;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Main pipeline function
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export async function runPipeline(ctx: PipelineContext): Promise<PipelineResult> {
|
||||
const { item, pkg, settings, debridService, integrityChecker, signal } = ctx;
|
||||
|
||||
// Abort guard
|
||||
if (signal.aborted) throw new Error("aborted");
|
||||
|
||||
// ----- Step 1: Unrestrict -----
|
||||
let directUrl = ctx.cachedDirectUrl || "";
|
||||
let provider = ctx.cachedProvider || item.provider;
|
||||
let providerLabel = ctx.cachedProviderLabel || "";
|
||||
let skipTlsVerify = ctx.cachedSkipTls || false;
|
||||
|
||||
if (!directUrl) {
|
||||
ctx.onStatus("validating", "Link wird umgewandelt...");
|
||||
ctx.onLog("INFO", "Unrestrict started", { url: item.url });
|
||||
|
||||
const unrestrictTimeoutMs = getEnvMs("RD_UNRESTRICT_TIMEOUT_MS", DEFAULT_UNRESTRICT_TIMEOUT_MS);
|
||||
const timeoutSignal = AbortSignal.timeout(unrestrictTimeoutMs);
|
||||
const combinedSignal = AbortSignal.any([signal, timeoutSignal]);
|
||||
|
||||
let unrestricted: UnrestrictedLink;
|
||||
try {
|
||||
unrestricted = await debridService.unrestrictLink(item.url, combinedSignal);
|
||||
} catch (error) {
|
||||
if (signal.aborted) throw error;
|
||||
if (timeoutSignal.aborted) {
|
||||
throw new DownloadError(
|
||||
DownloadErrorKind.ConnectTimeout,
|
||||
`Unrestrict timeout after ${Math.ceil(unrestrictTimeoutMs / 1000)}s`,
|
||||
);
|
||||
}
|
||||
throw classifyUnrestrictError(error);
|
||||
}
|
||||
|
||||
if (signal.aborted) throw new Error("aborted");
|
||||
|
||||
directUrl = unrestricted.directUrl;
|
||||
provider = unrestricted.provider;
|
||||
providerLabel = unrestricted.providerLabel || "";
|
||||
skipTlsVerify = unrestricted.skipTlsVerify || false;
|
||||
|
||||
// Update item metadata
|
||||
ctx.onProviderInfo(
|
||||
unrestricted.provider,
|
||||
unrestricted.providerLabel,
|
||||
unrestricted.sourceAccountId,
|
||||
unrestricted.sourceAccountLabel,
|
||||
);
|
||||
|
||||
// Resolve target path
|
||||
const fileName = sanitizeFilename(unrestricted.fileName || filenameFromUrl(item.url));
|
||||
try { fs.mkdirSync(pkg.outputDir, { recursive: true }); } catch {}
|
||||
|
||||
const existingPath = (item.targetPath || "").trim();
|
||||
const canReuse = existingPath
|
||||
&& isPathInsideDir(existingPath, pkg.outputDir)
|
||||
&& (item.downloadedBytes > 0 || fs.existsSync(existingPath));
|
||||
const preferred = canReuse ? existingPath : path.join(pkg.outputDir, fileName);
|
||||
const targetPath = ctx.claimTargetPath(item.id, preferred, Boolean(canReuse));
|
||||
|
||||
// Update item fields
|
||||
item.fileName = fileName;
|
||||
item.targetPath = targetPath;
|
||||
item.totalBytes = unrestricted.fileSize;
|
||||
item.provider = unrestricted.provider;
|
||||
item.providerLabel = unrestricted.providerLabel;
|
||||
item.providerAccountId = unrestricted.sourceAccountId;
|
||||
item.providerAccountLabel = unrestricted.sourceAccountLabel;
|
||||
item.retries += unrestricted.retriesUsed;
|
||||
|
||||
ctx.onLog("INFO", "Link unrestricted", {
|
||||
provider: unrestricted.provider,
|
||||
providerLabel: unrestricted.providerLabel || "",
|
||||
fileName,
|
||||
targetPath,
|
||||
fileSize: unrestricted.fileSize,
|
||||
directUrl,
|
||||
});
|
||||
}
|
||||
|
||||
// ----- Step 2: Stream download -----
|
||||
ctx.onStatus("downloading", `Download läuft (${providerLabel || providerDisplayName(provider)})`);
|
||||
|
||||
const stallTimeoutMs = getEnvMs("RD_STALL_TIMEOUT_MS", DEFAULT_STALL_TIMEOUT_MS);
|
||||
const connectTimeoutMs = getEnvMs("RD_CONNECT_TIMEOUT_MS", DEFAULT_CONNECT_TIMEOUT_MS);
|
||||
const lowThroughputTimeoutMs = getEnvMs("RD_LOW_THROUGHPUT_TIMEOUT_MS", DEFAULT_LOW_THROUGHPUT_TIMEOUT_MS);
|
||||
const lowThroughputMinBytes = getEnvMs("RD_LOW_THROUGHPUT_MIN_BYTES", DEFAULT_LOW_THROUGHPUT_MIN_BYTES);
|
||||
|
||||
// Speed limit
|
||||
let effectiveSpeedLimit = 0;
|
||||
if (settings.speedLimitEnabled && settings.speedLimitKbps > 0) {
|
||||
effectiveSpeedLimit = settings.speedLimitKbps * 1024;
|
||||
if (settings.speedLimitMode === "global") {
|
||||
// For global mode, caller divides by active download count
|
||||
// Here we just pass the per-download share
|
||||
}
|
||||
}
|
||||
|
||||
let streamResult: StreamResult;
|
||||
try {
|
||||
streamResult = await streamToFile({
|
||||
url: directUrl,
|
||||
targetPath: item.targetPath,
|
||||
expectedBytes: item.totalBytes,
|
||||
trackedDownloadedBytes: item.downloadedBytes,
|
||||
stallTimeoutMs,
|
||||
connectTimeoutMs,
|
||||
skipTlsVerify,
|
||||
speedLimitBps: effectiveSpeedLimit,
|
||||
signal,
|
||||
onProgress: ctx.onProgress,
|
||||
onHeartbeat: ctx.onHeartbeat,
|
||||
onResumable: ctx.onResumable,
|
||||
onFileNameOverride: (newName) => {
|
||||
const newPath = path.join(pkg.outputDir, newName);
|
||||
ctx.releaseTargetPath(item.id);
|
||||
const claimedPath = ctx.claimTargetPath(item.id, newPath);
|
||||
item.fileName = newName;
|
||||
item.targetPath = claimedPath;
|
||||
ctx.onFileNameOverride(newName, claimedPath);
|
||||
},
|
||||
onLog: ctx.onLog,
|
||||
onDiskBusy: ctx.onDiskBusy,
|
||||
lowThroughputTimeoutMs,
|
||||
lowThroughputMinBytes,
|
||||
isLargeBinary: LARGE_BINARY_RE.test(item.fileName || ""),
|
||||
});
|
||||
} catch (error) {
|
||||
if (signal.aborted) throw error;
|
||||
throw ensureDownloadError(error);
|
||||
}
|
||||
|
||||
// Update item after successful download
|
||||
item.downloadedBytes = streamResult.downloadedBytes;
|
||||
item.totalBytes = streamResult.totalBytes;
|
||||
|
||||
if (signal.aborted) throw new Error("aborted");
|
||||
|
||||
// ----- Step 3: Integrity check -----
|
||||
if (integrityChecker && settings.enableIntegrityCheck) {
|
||||
ctx.onStatus("integrity_check", "Integritätsprüfung...");
|
||||
ctx.onLog("INFO", "Integrity check started", { targetPath: item.targetPath });
|
||||
|
||||
try {
|
||||
const result = await integrityChecker.validateFile(item.targetPath, pkg.outputDir);
|
||||
if (!result.ok) {
|
||||
ctx.onLog("ERROR", "Integrity check failed", { message: result.message });
|
||||
throw new DownloadError(DownloadErrorKind.FileCorrupt, result.message);
|
||||
}
|
||||
ctx.onLog("INFO", "Integrity check passed", { message: result.message });
|
||||
} catch (error) {
|
||||
if (error instanceof DownloadError) throw error;
|
||||
// Non-DownloadError from integrity check — classify
|
||||
throw new DownloadError(DownloadErrorKind.FileCorrupt, String(error));
|
||||
}
|
||||
}
|
||||
|
||||
// ----- Done -----
|
||||
return {
|
||||
success: true,
|
||||
downloadedBytes: streamResult.downloadedBytes,
|
||||
totalBytes: streamResult.totalBytes,
|
||||
directUrl,
|
||||
provider: provider!,
|
||||
providerLabel,
|
||||
resumable: streamResult.resumable,
|
||||
skipTlsVerify,
|
||||
};
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Helpers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
function sanitizeFilename(name: string): string {
|
||||
return name.replace(/[<>:"/\\|?*\x00-\x1F]/g, "_").replace(/\s+/g, " ").trim() || "download";
|
||||
}
|
||||
|
||||
function filenameFromUrl(url: string): string {
|
||||
try {
|
||||
const u = new URL(url);
|
||||
const pathParts = u.pathname.split("/").filter(Boolean);
|
||||
const last = pathParts[pathParts.length - 1] || "download";
|
||||
return decodeURIComponent(last);
|
||||
} catch {
|
||||
return "download";
|
||||
}
|
||||
}
|
||||
|
||||
function isPathInsideDir(filePath: string, dirPath: string): boolean {
|
||||
const normalizedFile = path.resolve(filePath).toLowerCase();
|
||||
const normalizedDir = path.resolve(dirPath).toLowerCase();
|
||||
return normalizedFile.startsWith(normalizedDir);
|
||||
}
|
||||
|
||||
function providerDisplayName(provider: DebridProvider | null): string {
|
||||
if (!provider) return "Debrid";
|
||||
const names: Record<string, string> = {
|
||||
realdebrid: "Real-Debrid",
|
||||
"megadebrid-api": "Mega-Debrid API",
|
||||
"megadebrid-web": "Mega-Debrid Web",
|
||||
megadebrid: "Mega-Debrid",
|
||||
bestdebrid: "BestDebrid",
|
||||
alldebrid: "AllDebrid",
|
||||
ddownload: "DDownload",
|
||||
onefichier: "1Fichier",
|
||||
debridlink: "DebridLink",
|
||||
linksnappy: "LinkSnappy",
|
||||
};
|
||||
return names[provider] || provider;
|
||||
}
|
||||
@@ -0,0 +1,409 @@
|
||||
/**
|
||||
* post-processor.ts — Extraction state machine with bounded retries.
|
||||
*
|
||||
* Each archive has a clear state (pending → extracting → done/failed).
|
||||
* No infinite loops: hard cap on retry count per archive.
|
||||
* Redownload requests are emitted as events, not handled internally.
|
||||
*/
|
||||
|
||||
import { EventEmitter } from "node:events";
|
||||
import { DownloadError, DownloadErrorKind, classifyExtractionError } from "./error-classifier";
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Types
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export interface ArchiveExtractionState {
|
||||
archiveName: string;
|
||||
status: "pending" | "extracting" | "done" | "failed";
|
||||
attempts: number;
|
||||
maxAttempts: number;
|
||||
redownloaded: boolean;
|
||||
lastError?: string;
|
||||
lastErrorKind?: DownloadErrorKind;
|
||||
}
|
||||
|
||||
export interface PackagePostProcessState {
|
||||
packageId: string;
|
||||
status: "idle" | "waiting" | "extracting" | "done" | "failed" | "aborted";
|
||||
archives: Map<string, ArchiveExtractionState>;
|
||||
startedAt: number;
|
||||
completedAt?: number;
|
||||
label?: string;
|
||||
}
|
||||
|
||||
export interface PostProcessOptions {
|
||||
packageDir: string;
|
||||
extractDir: string;
|
||||
cleanupMode: "none" | "trash" | "delete";
|
||||
conflictMode: "overwrite" | "skip" | "rename" | "ask";
|
||||
removeLinks: boolean;
|
||||
removeSamples: boolean;
|
||||
passwordList: string;
|
||||
hybridMode: boolean;
|
||||
maxParallelExtract: number;
|
||||
extractCpuPriority: string;
|
||||
signal: AbortSignal;
|
||||
}
|
||||
|
||||
export interface ExtractProgressUpdate {
|
||||
current: number;
|
||||
total: number;
|
||||
percent: number;
|
||||
archiveName: string;
|
||||
archivePercent?: number;
|
||||
phase: "extracting" | "done" | "preparing";
|
||||
archiveDone?: boolean;
|
||||
archiveSuccess?: boolean;
|
||||
}
|
||||
|
||||
export interface ExtractArchiveFailure {
|
||||
archiveName: string;
|
||||
errorText: string;
|
||||
category: string;
|
||||
suggestRedownload: boolean;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Constants
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const DEFAULT_MAX_EXTRACT_ATTEMPTS = 3;
|
||||
const SLOT_POLL_INTERVAL_MS = 500;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// PostProcessor
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export interface PostProcessorEvents {
|
||||
progress: [{ packageId: string; update: ExtractProgressUpdate }];
|
||||
"package-done": [{ packageId: string; success: boolean; errors: string[] }];
|
||||
"archive-redownload": [{ packageId: string; archiveName: string; error: string }];
|
||||
status: [{ packageId: string; label: string }];
|
||||
}
|
||||
|
||||
export class PostProcessor extends EventEmitter {
|
||||
private states = new Map<string, PackagePostProcessState>();
|
||||
private abortControllers = new Map<string, AbortController>();
|
||||
private activeTasks = new Map<string, Promise<void>>();
|
||||
private activeSlots = 0;
|
||||
private maxSlots: number;
|
||||
private slotWaiters: Array<() => void> = [];
|
||||
|
||||
/** Extraction function — injected to avoid circular dependency. */
|
||||
private extractFn: ((opts: any) => Promise<any>) | null = null;
|
||||
/** Archive candidate finder. */
|
||||
private findArchivesFn: ((dir: string) => string[] | Promise<string[]>) | null = null;
|
||||
|
||||
constructor(maxParallel: number = 2) {
|
||||
super();
|
||||
this.maxSlots = maxParallel;
|
||||
}
|
||||
|
||||
/** Inject the extraction function (from extractor.ts). */
|
||||
setExtractor(
|
||||
extractFn: (opts: any) => Promise<any>,
|
||||
findArchivesFn: (dir: string) => string[] | Promise<string[]>,
|
||||
): void {
|
||||
this.extractFn = extractFn;
|
||||
this.findArchivesFn = findArchivesFn;
|
||||
}
|
||||
|
||||
setMaxParallel(n: number): void {
|
||||
this.maxSlots = Math.max(1, n);
|
||||
}
|
||||
|
||||
/**
|
||||
* Queue a package for post-processing.
|
||||
* If already processing, mark for re-run (hybrid requeue).
|
||||
*/
|
||||
queuePackage(packageId: string, options: PostProcessOptions): void {
|
||||
const existing = this.activeTasks.get(packageId);
|
||||
if (existing) {
|
||||
// Mark for requeue — current run will check after finishing
|
||||
const state = this.states.get(packageId);
|
||||
if (state) state.status = "waiting";
|
||||
return;
|
||||
}
|
||||
|
||||
const ac = new AbortController();
|
||||
this.abortControllers.set(packageId, ac);
|
||||
|
||||
const combinedSignal = AbortSignal.any([options.signal, ac.signal]);
|
||||
|
||||
const task = this.runPostProcessing(packageId, { ...options, signal: combinedSignal });
|
||||
this.activeTasks.set(packageId, task);
|
||||
|
||||
task.finally(() => {
|
||||
this.activeTasks.delete(packageId);
|
||||
this.abortControllers.delete(packageId);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Abort processing for a specific package.
|
||||
*/
|
||||
abortPackage(packageId: string): void {
|
||||
const ac = this.abortControllers.get(packageId);
|
||||
if (ac) ac.abort();
|
||||
const state = this.states.get(packageId);
|
||||
if (state) state.status = "aborted";
|
||||
}
|
||||
|
||||
/**
|
||||
* Abort all active post-processing.
|
||||
*/
|
||||
abortAll(): void {
|
||||
for (const [id, ac] of this.abortControllers) {
|
||||
ac.abort();
|
||||
const state = this.states.get(id);
|
||||
if (state) state.status = "aborted";
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Retry extraction for a package (user-initiated).
|
||||
*/
|
||||
retryPackage(packageId: string, options: PostProcessOptions): void {
|
||||
// Reset archive states
|
||||
const state = this.states.get(packageId);
|
||||
if (state) {
|
||||
for (const archive of state.archives.values()) {
|
||||
if (archive.status === "failed") {
|
||||
archive.status = "pending";
|
||||
archive.attempts = 0;
|
||||
}
|
||||
}
|
||||
state.status = "idle";
|
||||
}
|
||||
this.queuePackage(packageId, options);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get state for a package.
|
||||
*/
|
||||
getState(packageId: string): PackagePostProcessState | undefined {
|
||||
return this.states.get(packageId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if any processing is active.
|
||||
*/
|
||||
isActive(): boolean {
|
||||
return this.activeTasks.size > 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Wait for all active tasks to complete.
|
||||
*/
|
||||
async waitAll(): Promise<void> {
|
||||
await Promise.allSettled([...this.activeTasks.values()]);
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Private
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
private async runPostProcessing(packageId: string, options: PostProcessOptions): Promise<void> {
|
||||
// Acquire slot
|
||||
await this.acquireSlot(options.signal);
|
||||
if (options.signal.aborted) return;
|
||||
|
||||
const state: PackagePostProcessState = this.states.get(packageId) || {
|
||||
packageId,
|
||||
status: "extracting",
|
||||
archives: new Map(),
|
||||
startedAt: Date.now(),
|
||||
};
|
||||
state.status = "extracting";
|
||||
state.startedAt = Date.now();
|
||||
this.states.set(packageId, state);
|
||||
|
||||
let round = 0;
|
||||
const MAX_ROUNDS = 5; // Hard cap on requeue rounds
|
||||
|
||||
try {
|
||||
do {
|
||||
round++;
|
||||
if (round > MAX_ROUNDS) {
|
||||
state.label = `Max. Runden erreicht (${MAX_ROUNDS})`;
|
||||
break;
|
||||
}
|
||||
|
||||
this.emit("status", { packageId, label: `Entpacken Runde ${round}...` });
|
||||
|
||||
try {
|
||||
await this.runExtractionRound(packageId, options, state);
|
||||
} catch (error) {
|
||||
if (options.signal.aborted) break;
|
||||
const msg = error instanceof Error ? error.message : String(error);
|
||||
state.label = `Fehler: ${msg}`;
|
||||
this.emit("status", { packageId, label: state.label });
|
||||
}
|
||||
|
||||
// Check if there are pending archives for another round
|
||||
const hasPending = [...state.archives.values()].some(a => a.status === "pending");
|
||||
if (!hasPending) break;
|
||||
|
||||
} while (!options.signal.aborted);
|
||||
|
||||
// Determine final status
|
||||
const archives = [...state.archives.values()];
|
||||
const allDone = archives.every(a => a.status === "done");
|
||||
const anyFailed = archives.some(a => a.status === "failed");
|
||||
const errors = archives
|
||||
.filter(a => a.status === "failed")
|
||||
.map(a => `${a.archiveName}: ${a.lastError || "Unbekannt"}`);
|
||||
|
||||
if (options.signal.aborted) {
|
||||
state.status = "aborted";
|
||||
} else if (allDone || archives.length === 0) {
|
||||
state.status = "done";
|
||||
} else {
|
||||
state.status = "failed";
|
||||
}
|
||||
state.completedAt = Date.now();
|
||||
|
||||
this.emit("package-done", {
|
||||
packageId,
|
||||
success: state.status === "done",
|
||||
errors,
|
||||
});
|
||||
|
||||
} finally {
|
||||
this.releaseSlot();
|
||||
}
|
||||
}
|
||||
|
||||
private async runExtractionRound(
|
||||
packageId: string,
|
||||
options: PostProcessOptions,
|
||||
state: PackagePostProcessState,
|
||||
): Promise<void> {
|
||||
if (!this.extractFn || !this.findArchivesFn) {
|
||||
throw new Error("Extractor not configured — call setExtractor()");
|
||||
}
|
||||
|
||||
// Find archives
|
||||
const archivePaths = await this.findArchivesFn(options.packageDir);
|
||||
if (archivePaths.length === 0) {
|
||||
state.label = "Keine Archive gefunden";
|
||||
return;
|
||||
}
|
||||
|
||||
// Initialize archive states for new archives
|
||||
for (const archivePath of archivePaths) {
|
||||
const name = archivePath;
|
||||
if (!state.archives.has(name)) {
|
||||
state.archives.set(name, {
|
||||
archiveName: name,
|
||||
status: "pending",
|
||||
attempts: 0,
|
||||
maxAttempts: DEFAULT_MAX_EXTRACT_ATTEMPTS,
|
||||
redownloaded: false,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// Only extract pending archives
|
||||
const pendingArchives = [...state.archives.values()]
|
||||
.filter(a => a.status === "pending")
|
||||
.map(a => a.archiveName);
|
||||
|
||||
if (pendingArchives.length === 0) return;
|
||||
|
||||
// Run extraction
|
||||
const failures: ExtractArchiveFailure[] = [];
|
||||
|
||||
await this.extractFn({
|
||||
packageDir: options.packageDir,
|
||||
targetDir: options.extractDir,
|
||||
cleanupMode: options.cleanupMode,
|
||||
conflictMode: options.conflictMode,
|
||||
removeLinks: options.removeLinks,
|
||||
removeSamples: options.removeSamples,
|
||||
passwordList: options.passwordList,
|
||||
signal: options.signal,
|
||||
hybridMode: options.hybridMode,
|
||||
maxParallel: options.maxParallelExtract,
|
||||
extractCpuPriority: options.extractCpuPriority,
|
||||
packageId,
|
||||
onlyArchives: new Set(pendingArchives),
|
||||
onProgress: (update: ExtractProgressUpdate) => {
|
||||
this.emit("progress", { packageId, update });
|
||||
|
||||
// Track individual archive completion
|
||||
if (update.archiveDone) {
|
||||
const archiveState = state.archives.get(update.archiveName);
|
||||
if (archiveState) {
|
||||
archiveState.attempts++;
|
||||
if (update.archiveSuccess) {
|
||||
archiveState.status = "done";
|
||||
}
|
||||
// If not success, onArchiveFailure will handle it
|
||||
}
|
||||
}
|
||||
},
|
||||
onArchiveFailure: (failure: ExtractArchiveFailure) => {
|
||||
failures.push(failure);
|
||||
const archiveState = state.archives.get(failure.archiveName);
|
||||
if (!archiveState) return;
|
||||
|
||||
const error = classifyExtractionError(failure.errorText, failure.category);
|
||||
archiveState.lastError = failure.errorText;
|
||||
archiveState.lastErrorKind = error.kind;
|
||||
archiveState.attempts++;
|
||||
|
||||
// Decide: retry, redownload, or fail permanently
|
||||
if (archiveState.attempts >= archiveState.maxAttempts) {
|
||||
// Max attempts reached
|
||||
if (error.kind === DownloadErrorKind.ArchiveCorrupt && !archiveState.redownloaded && failure.suggestRedownload) {
|
||||
// Request redownload (max once per archive)
|
||||
archiveState.redownloaded = true;
|
||||
archiveState.attempts = 0; // Reset for redownloaded archive
|
||||
archiveState.status = "pending";
|
||||
this.emit("archive-redownload", {
|
||||
packageId,
|
||||
archiveName: failure.archiveName,
|
||||
error: failure.errorText,
|
||||
});
|
||||
} else {
|
||||
archiveState.status = "failed";
|
||||
}
|
||||
} else {
|
||||
// Still have attempts left — mark as pending for next round
|
||||
archiveState.status = "pending";
|
||||
}
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Slot management
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
private async acquireSlot(signal: AbortSignal): Promise<void> {
|
||||
while (this.activeSlots >= this.maxSlots) {
|
||||
if (signal.aborted) return;
|
||||
await new Promise<void>(resolve => {
|
||||
this.slotWaiters.push(resolve);
|
||||
// Also poll in case signal gets aborted
|
||||
const timer = setTimeout(() => {
|
||||
const idx = this.slotWaiters.indexOf(resolve);
|
||||
if (idx >= 0) this.slotWaiters.splice(idx, 1);
|
||||
resolve();
|
||||
}, SLOT_POLL_INTERVAL_MS);
|
||||
// Clean up timer if resolved normally
|
||||
const originalResolve = resolve;
|
||||
// Just let the poll handle it
|
||||
});
|
||||
}
|
||||
this.activeSlots++;
|
||||
}
|
||||
|
||||
private releaseSlot(): void {
|
||||
this.activeSlots = Math.max(0, this.activeSlots - 1);
|
||||
const waiter = this.slotWaiters.shift();
|
||||
if (waiter) waiter();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,390 @@
|
||||
/**
|
||||
* retry-manager.ts — Declarative retry logic with per-error-kind policies.
|
||||
*
|
||||
* Each DownloadErrorKind has a RetryPolicy that determines max retries,
|
||||
* backoff strategy, and actions (reset file, switch provider, etc.).
|
||||
* The RetryManager tracks failure counts per item and decides whether
|
||||
* to retry or fail permanently.
|
||||
*/
|
||||
|
||||
import { DownloadError, DownloadErrorKind, errorKindLabel, isPermanentKind } from "./error-classifier";
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Retry Policy
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export interface RetryPolicy {
|
||||
/** Maximum retries for this error kind. 0 = fail immediately. */
|
||||
maxRetries: number;
|
||||
/** Backoff strategy. */
|
||||
backoff: "fixed" | "exponential";
|
||||
/** Base delay in milliseconds. */
|
||||
baseDelayMs: number;
|
||||
/** Maximum delay in milliseconds (cap for exponential). */
|
||||
maxDelayMs: number;
|
||||
/** Delete partial file before retry. */
|
||||
resetFile: boolean;
|
||||
/** Try a different debrid provider on retry. */
|
||||
switchProvider: boolean;
|
||||
/** Request a fresh direct link from debrid service. */
|
||||
refreshLink: boolean;
|
||||
/** Apply cooldown to current provider (ms). 0 = no cooldown. */
|
||||
providerCooldownMs: number;
|
||||
}
|
||||
|
||||
export const RETRY_POLICIES: Record<DownloadErrorKind, RetryPolicy> = {
|
||||
// -- Network --
|
||||
[DownloadErrorKind.NetworkReset]: {
|
||||
maxRetries: 3, backoff: "fixed", baseDelayMs: 300, maxDelayMs: 300,
|
||||
resetFile: true, switchProvider: false, refreshLink: false, providerCooldownMs: 0,
|
||||
},
|
||||
[DownloadErrorKind.Timeout]: {
|
||||
maxRetries: 10, backoff: "exponential", baseDelayMs: 200, maxDelayMs: 30_000,
|
||||
resetFile: false, switchProvider: false, refreshLink: false, providerCooldownMs: 0,
|
||||
},
|
||||
[DownloadErrorKind.DnsFailure]: {
|
||||
maxRetries: 2, backoff: "fixed", baseDelayMs: 5000, maxDelayMs: 5000,
|
||||
resetFile: false, switchProvider: false, refreshLink: false, providerCooldownMs: 0,
|
||||
},
|
||||
[DownloadErrorKind.ConnectTimeout]: {
|
||||
maxRetries: 4, backoff: "exponential", baseDelayMs: 2000, maxDelayMs: 30_000,
|
||||
resetFile: false, switchProvider: false, refreshLink: true, providerCooldownMs: 0,
|
||||
},
|
||||
|
||||
// -- HTTP --
|
||||
[DownloadErrorKind.RangeNotSatisfied]: {
|
||||
maxRetries: 2, backoff: "fixed", baseDelayMs: 200, maxDelayMs: 200,
|
||||
resetFile: true, switchProvider: false, refreshLink: true, providerCooldownMs: 0,
|
||||
},
|
||||
[DownloadErrorKind.RangeIgnored]: {
|
||||
maxRetries: 3, backoff: "fixed", baseDelayMs: 300, maxDelayMs: 300,
|
||||
resetFile: false, switchProvider: false, refreshLink: true, providerCooldownMs: 0,
|
||||
},
|
||||
[DownloadErrorKind.ServerError]: {
|
||||
maxRetries: 5, backoff: "exponential", baseDelayMs: 2000, maxDelayMs: 60_000,
|
||||
resetFile: false, switchProvider: false, refreshLink: true, providerCooldownMs: 0,
|
||||
},
|
||||
[DownloadErrorKind.RateLimited]: {
|
||||
maxRetries: 8, backoff: "exponential", baseDelayMs: 5000, maxDelayMs: 120_000,
|
||||
resetFile: false, switchProvider: false, refreshLink: false, providerCooldownMs: 0,
|
||||
},
|
||||
[DownloadErrorKind.Forbidden]: {
|
||||
maxRetries: 2, backoff: "fixed", baseDelayMs: 1000, maxDelayMs: 1000,
|
||||
resetFile: false, switchProvider: false, refreshLink: true, providerCooldownMs: 0,
|
||||
},
|
||||
[DownloadErrorKind.NotFound]: {
|
||||
maxRetries: 1, backoff: "fixed", baseDelayMs: 2000, maxDelayMs: 2000,
|
||||
resetFile: false, switchProvider: false, refreshLink: true, providerCooldownMs: 0,
|
||||
},
|
||||
|
||||
// -- Provider / Debrid --
|
||||
[DownloadErrorKind.UnrestrictFailed]: {
|
||||
maxRetries: 5, backoff: "exponential", baseDelayMs: 5000, maxDelayMs: 120_000,
|
||||
resetFile: false, switchProvider: true, refreshLink: false, providerCooldownMs: 20_000,
|
||||
},
|
||||
[DownloadErrorKind.ProviderBusy]: {
|
||||
maxRetries: 8, backoff: "exponential", baseDelayMs: 5000, maxDelayMs: 60_000,
|
||||
resetFile: false, switchProvider: true, refreshLink: false, providerCooldownMs: 12_000,
|
||||
},
|
||||
[DownloadErrorKind.ProviderDown]: {
|
||||
maxRetries: 5, backoff: "exponential", baseDelayMs: 10_000, maxDelayMs: 180_000,
|
||||
resetFile: false, switchProvider: true, refreshLink: false, providerCooldownMs: 30_000,
|
||||
},
|
||||
[DownloadErrorKind.HosterUnavailable]: {
|
||||
maxRetries: 5, backoff: "exponential", baseDelayMs: 5000, maxDelayMs: 30_000,
|
||||
resetFile: false, switchProvider: true, refreshLink: false, providerCooldownMs: 15_000,
|
||||
},
|
||||
[DownloadErrorKind.LinkDead]: {
|
||||
maxRetries: 0, backoff: "fixed", baseDelayMs: 0, maxDelayMs: 0,
|
||||
resetFile: false, switchProvider: false, refreshLink: false, providerCooldownMs: 0,
|
||||
},
|
||||
[DownloadErrorKind.QuotaExceeded]: {
|
||||
maxRetries: 3, backoff: "exponential", baseDelayMs: 30_000, maxDelayMs: 300_000,
|
||||
resetFile: false, switchProvider: true, refreshLink: false, providerCooldownMs: 60_000,
|
||||
},
|
||||
|
||||
// -- Filesystem --
|
||||
[DownloadErrorKind.DiskFull]: {
|
||||
maxRetries: 0, backoff: "fixed", baseDelayMs: 0, maxDelayMs: 0,
|
||||
resetFile: false, switchProvider: false, refreshLink: false, providerCooldownMs: 0,
|
||||
},
|
||||
[DownloadErrorKind.PermissionDenied]: {
|
||||
maxRetries: 0, backoff: "fixed", baseDelayMs: 0, maxDelayMs: 0,
|
||||
resetFile: false, switchProvider: false, refreshLink: false, providerCooldownMs: 0,
|
||||
},
|
||||
[DownloadErrorKind.FileLocked]: {
|
||||
maxRetries: 3, backoff: "exponential", baseDelayMs: 1000, maxDelayMs: 10_000,
|
||||
resetFile: false, switchProvider: false, refreshLink: false, providerCooldownMs: 0,
|
||||
},
|
||||
|
||||
// -- Integrity / Resume --
|
||||
[DownloadErrorKind.FileCorrupt]: {
|
||||
maxRetries: 2, backoff: "fixed", baseDelayMs: 500, maxDelayMs: 500,
|
||||
resetFile: true, switchProvider: false, refreshLink: true, providerCooldownMs: 0,
|
||||
},
|
||||
[DownloadErrorKind.FileTruncated]: {
|
||||
maxRetries: 3, backoff: "fixed", baseDelayMs: 300, maxDelayMs: 300,
|
||||
resetFile: true, switchProvider: false, refreshLink: true, providerCooldownMs: 0,
|
||||
},
|
||||
[DownloadErrorKind.ResumeUnderflow]: {
|
||||
maxRetries: 2, backoff: "fixed", baseDelayMs: 300, maxDelayMs: 300,
|
||||
resetFile: true, switchProvider: false, refreshLink: true, providerCooldownMs: 0,
|
||||
},
|
||||
|
||||
// -- Extraction --
|
||||
[DownloadErrorKind.WrongPassword]: {
|
||||
maxRetries: 0, backoff: "fixed", baseDelayMs: 0, maxDelayMs: 0,
|
||||
resetFile: false, switchProvider: false, refreshLink: false, providerCooldownMs: 0,
|
||||
},
|
||||
[DownloadErrorKind.ArchiveCorrupt]: {
|
||||
maxRetries: 1, backoff: "fixed", baseDelayMs: 1000, maxDelayMs: 1000,
|
||||
resetFile: true, switchProvider: false, refreshLink: true, providerCooldownMs: 0,
|
||||
},
|
||||
[DownloadErrorKind.ExtractorCrash]: {
|
||||
maxRetries: 1, backoff: "fixed", baseDelayMs: 2000, maxDelayMs: 2000,
|
||||
resetFile: false, switchProvider: false, refreshLink: false, providerCooldownMs: 0,
|
||||
},
|
||||
|
||||
// -- Write / Drain --
|
||||
[DownloadErrorKind.WriteDrainTimeout]: {
|
||||
maxRetries: 3, backoff: "exponential", baseDelayMs: 2000, maxDelayMs: 30_000,
|
||||
resetFile: false, switchProvider: false, refreshLink: false, providerCooldownMs: 0,
|
||||
},
|
||||
|
||||
// -- Catchall --
|
||||
[DownloadErrorKind.Unknown]: {
|
||||
maxRetries: 5, backoff: "exponential", baseDelayMs: 1000, maxDelayMs: 60_000,
|
||||
resetFile: false, switchProvider: false, refreshLink: false, providerCooldownMs: 0,
|
||||
},
|
||||
};
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Retry Actions
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export type RetryAction =
|
||||
| "reset_file"
|
||||
| "switch_provider"
|
||||
| "refresh_link"
|
||||
| "cooldown_provider"
|
||||
| "shelve";
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Retry State (per item)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export interface RetryState {
|
||||
failuresByKind: Partial<Record<DownloadErrorKind, number>>;
|
||||
totalFailures: number;
|
||||
shelveCount: number;
|
||||
lastErrorKind?: DownloadErrorKind;
|
||||
lastErrorMessage?: string;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Retry Decision
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export interface RetryDecision {
|
||||
shouldRetry: boolean;
|
||||
delayMs: number;
|
||||
actions: RetryAction[];
|
||||
/** Human-readable status message for UI (German). */
|
||||
reason: string;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Constants
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const SHELVE_THRESHOLD = 15;
|
||||
const SHELVE_DELAY_MS = 90_000;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// RetryManager
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export class RetryManager {
|
||||
private states = new Map<string, RetryState>();
|
||||
private userRetryLimit: number;
|
||||
|
||||
constructor(retryLimit: number = 0) {
|
||||
this.userRetryLimit = retryLimit;
|
||||
}
|
||||
|
||||
/** Update the user-configured retry limit. 0 = unlimited. */
|
||||
setRetryLimit(limit: number): void {
|
||||
this.userRetryLimit = Math.max(0, limit);
|
||||
}
|
||||
|
||||
/**
|
||||
* Record a failure and decide whether to retry.
|
||||
*/
|
||||
evaluate(itemId: string, error: DownloadError): RetryDecision {
|
||||
const state = this.getOrCreateState(itemId);
|
||||
const kind = error.kind;
|
||||
|
||||
// Update state
|
||||
state.failuresByKind[kind] = (state.failuresByKind[kind] || 0) + 1;
|
||||
state.totalFailures += 1;
|
||||
state.lastErrorKind = kind;
|
||||
state.lastErrorMessage = error.message;
|
||||
|
||||
const kindCount = state.failuresByKind[kind]!;
|
||||
const policy = RETRY_POLICIES[kind];
|
||||
|
||||
// Permanent errors — never retry
|
||||
if (isPermanentKind(kind) || policy.maxRetries === 0) {
|
||||
return {
|
||||
shouldRetry: false,
|
||||
delayMs: 0,
|
||||
actions: [],
|
||||
reason: errorKindLabel(kind),
|
||||
};
|
||||
}
|
||||
|
||||
// Determine effective max retries (user limit overrides if set)
|
||||
const effectiveMax = this.userRetryLimit > 0
|
||||
? Math.min(policy.maxRetries, this.userRetryLimit)
|
||||
: policy.maxRetries;
|
||||
|
||||
// Check shelving threshold BEFORE individual kind limits
|
||||
if (state.totalFailures >= SHELVE_THRESHOLD) {
|
||||
return this.shelve(state, kind);
|
||||
}
|
||||
|
||||
// Check if this specific kind exhausted its retries
|
||||
if (kindCount > effectiveMax) {
|
||||
return {
|
||||
shouldRetry: false,
|
||||
delayMs: 0,
|
||||
actions: [],
|
||||
reason: `${errorKindLabel(kind)} — Versuche erschöpft (${kindCount}/${effectiveMax})`,
|
||||
};
|
||||
}
|
||||
|
||||
// Retry — compute delay and actions
|
||||
const delayMs = this.computeDelay(policy, kindCount);
|
||||
const actions = this.computeActions(policy);
|
||||
const reason = `${errorKindLabel(kind)}, Retry ${kindCount}/${effectiveMax}`;
|
||||
|
||||
return { shouldRetry: true, delayMs, actions, reason };
|
||||
}
|
||||
|
||||
/**
|
||||
* Reset retry state for an item (manual reset by user).
|
||||
*/
|
||||
resetItem(itemId: string): void {
|
||||
this.states.delete(itemId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get current retry state for persistence.
|
||||
*/
|
||||
getState(itemId: string): RetryState | undefined {
|
||||
return this.states.get(itemId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Restore retry state from persisted session.
|
||||
*/
|
||||
restoreState(itemId: string, state: RetryState): void {
|
||||
this.states.set(itemId, { ...state });
|
||||
}
|
||||
|
||||
/**
|
||||
* Export all retry states for persistence.
|
||||
*/
|
||||
exportStates(): Record<string, RetryState> {
|
||||
const out: Record<string, RetryState> = {};
|
||||
for (const [id, state] of this.states) {
|
||||
out[id] = { ...state, failuresByKind: { ...state.failuresByKind } };
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/**
|
||||
* Import retry states from persistence.
|
||||
*/
|
||||
importStates(states: Record<string, RetryState>): void {
|
||||
this.states.clear();
|
||||
for (const [id, state] of Object.entries(states)) {
|
||||
this.states.set(id, { ...state, failuresByKind: { ...state.failuresByKind } });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove state for deleted/cancelled items.
|
||||
*/
|
||||
removeItem(itemId: string): void {
|
||||
this.states.delete(itemId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Soft-reset stale retry state. Halves counters for items that haven't
|
||||
* failed recently. Called periodically (e.g. every 10 minutes).
|
||||
*/
|
||||
softReset(): void {
|
||||
for (const state of this.states.values()) {
|
||||
if (state.totalFailures > 0) {
|
||||
for (const kind of Object.keys(state.failuresByKind) as DownloadErrorKind[]) {
|
||||
state.failuresByKind[kind] = Math.floor((state.failuresByKind[kind] || 0) / 2);
|
||||
}
|
||||
state.totalFailures = Object.values(state.failuresByKind).reduce(
|
||||
(sum, v) => sum + (v || 0), 0,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Private
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
private getOrCreateState(itemId: string): RetryState {
|
||||
let state = this.states.get(itemId);
|
||||
if (!state) {
|
||||
state = { failuresByKind: {}, totalFailures: 0, shelveCount: 0 };
|
||||
this.states.set(itemId, state);
|
||||
}
|
||||
return state;
|
||||
}
|
||||
|
||||
private shelve(state: RetryState, lastKind: DownloadErrorKind): RetryDecision {
|
||||
// Halve all counters to allow recovery
|
||||
for (const kind of Object.keys(state.failuresByKind) as DownloadErrorKind[]) {
|
||||
state.failuresByKind[kind] = Math.floor((state.failuresByKind[kind] || 0) / 2);
|
||||
}
|
||||
state.totalFailures = Object.values(state.failuresByKind).reduce(
|
||||
(sum, v) => sum + (v || 0), 0,
|
||||
);
|
||||
state.shelveCount += 1;
|
||||
|
||||
return {
|
||||
shouldRetry: true,
|
||||
delayMs: SHELVE_DELAY_MS,
|
||||
actions: ["shelve", "switch_provider", "refresh_link"],
|
||||
reason: `Viele Fehler (${SHELVE_THRESHOLD}+), pausiert für ${SHELVE_DELAY_MS / 1000}s`,
|
||||
};
|
||||
}
|
||||
|
||||
private computeDelay(policy: RetryPolicy, attempt: number): number {
|
||||
if (policy.backoff === "fixed") {
|
||||
return policy.baseDelayMs;
|
||||
}
|
||||
// Exponential: base * 1.5^(attempt-1) with jitter, capped at max
|
||||
const base = policy.baseDelayMs * Math.pow(1.5, attempt - 1);
|
||||
const capped = Math.min(base, policy.maxDelayMs);
|
||||
const jitter = capped * Math.random() * 0.5;
|
||||
return Math.floor(Math.max(capped * 0.5, capped - jitter));
|
||||
}
|
||||
|
||||
private computeActions(policy: RetryPolicy): RetryAction[] {
|
||||
const actions: RetryAction[] = [];
|
||||
if (policy.resetFile) actions.push("reset_file");
|
||||
if (policy.switchProvider) actions.push("switch_provider");
|
||||
if (policy.refreshLink) actions.push("refresh_link");
|
||||
if (policy.providerCooldownMs > 0) actions.push("cooldown_provider");
|
||||
return actions;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,492 @@
|
||||
/**
|
||||
* scheduler.ts — Queue management, slot allocation, and stall detection.
|
||||
*
|
||||
* The scheduler runs a loop that fills download slots up to maxParallel,
|
||||
* monitors heartbeats for stall detection, and provides a global watchdog.
|
||||
*/
|
||||
|
||||
import { EventEmitter } from "node:events";
|
||||
import type { DownloadItem, PackageEntry, PackagePriority, SessionState } from "../../shared/types";
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Types
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export interface SchedulerConfig {
|
||||
maxParallel: number;
|
||||
stallTimeoutMs: number;
|
||||
globalStallWatchdogMs: number;
|
||||
}
|
||||
|
||||
export interface ActiveSlot {
|
||||
itemId: string;
|
||||
packageId: string;
|
||||
abortController: AbortController;
|
||||
abortReason: "stop" | "cancel" | "reconnect" | "package_toggle" | "stall" | "shutdown" | "reset" | "none";
|
||||
resumable: boolean;
|
||||
lastHeartbeatAt: number;
|
||||
bytesAtHeartbeat: number;
|
||||
blockedOnDiskWrite: boolean;
|
||||
blockedOnDiskSince: number;
|
||||
}
|
||||
|
||||
export interface SlotRequest {
|
||||
itemId: string;
|
||||
packageId: string;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Scheduler
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export class Scheduler extends EventEmitter {
|
||||
private generation = 0;
|
||||
private running = false;
|
||||
private paused = false;
|
||||
private config: SchedulerConfig;
|
||||
|
||||
// Active downloads
|
||||
private slots = new Map<string, ActiveSlot>();
|
||||
|
||||
// Retry delays
|
||||
private retryDelays = new Map<string, number>(); // itemId → readyAtEpochMs
|
||||
|
||||
// Provider cooldowns
|
||||
private providerCooldowns = new Map<string, { cooldownUntil: number; failureCount: number }>();
|
||||
|
||||
// Reconnect state
|
||||
private reconnectUntil = 0;
|
||||
|
||||
// Global watchdog state
|
||||
private lastGlobalProgressBytes = 0;
|
||||
private lastGlobalProgressAt = 0;
|
||||
|
||||
// Scoped run (only these packages)
|
||||
private scopedPackageIds = new Set<string>();
|
||||
|
||||
constructor(config: SchedulerConfig) {
|
||||
super();
|
||||
this.config = { ...config };
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Public API
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
/** Update config at runtime (e.g. when user changes maxParallel). */
|
||||
updateConfig(partial: Partial<SchedulerConfig>): void {
|
||||
Object.assign(this.config, partial);
|
||||
}
|
||||
|
||||
/**
|
||||
* Start the scheduler loop.
|
||||
*
|
||||
* @param session Live session state
|
||||
* @param startItem Callback to start a download for a slot request
|
||||
* @param scopedIds Optional: only run these package IDs
|
||||
*/
|
||||
async start(
|
||||
session: SessionState,
|
||||
startItem: (slot: SlotRequest) => void,
|
||||
scopedIds?: string[],
|
||||
): Promise<void> {
|
||||
this.generation++;
|
||||
this.running = true;
|
||||
this.paused = false;
|
||||
this.scopedPackageIds = new Set(scopedIds || []);
|
||||
this.lastGlobalProgressBytes = 0;
|
||||
this.lastGlobalProgressAt = Date.now();
|
||||
|
||||
const myGeneration = this.generation;
|
||||
const loopIntervalMs = 120;
|
||||
let lastHeartbeatCheckAt = Date.now();
|
||||
let lastSoftResetAt = Date.now();
|
||||
|
||||
while (this.running && this.generation === myGeneration) {
|
||||
const now = Date.now();
|
||||
|
||||
// Paused — just idle
|
||||
if (this.paused) {
|
||||
await sleep(loopIntervalMs);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Reconnect wait
|
||||
if (this.reconnectUntil > now) {
|
||||
await sleep(220);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Fill slots
|
||||
const maxParallel = Math.max(1, this.config.maxParallel);
|
||||
while (this.slots.size < maxParallel) {
|
||||
const next = this.findNextItem(session, now);
|
||||
if (!next) break;
|
||||
startItem(next);
|
||||
}
|
||||
|
||||
// Heartbeat / stall check (every 2s)
|
||||
if (now - lastHeartbeatCheckAt >= 2000) {
|
||||
this.checkStalls(now);
|
||||
lastHeartbeatCheckAt = now;
|
||||
}
|
||||
|
||||
// Global stall watchdog
|
||||
this.runGlobalWatchdog(now);
|
||||
|
||||
// Soft-reset stale retry delays (every 10 min)
|
||||
if (now - lastSoftResetAt >= 600_000) {
|
||||
this.cleanupStaleRetryDelays(now);
|
||||
lastSoftResetAt = now;
|
||||
}
|
||||
|
||||
// Check if run is complete
|
||||
if (this.slots.size === 0) {
|
||||
const hasQueued = this.hasQueuedItems(session, now);
|
||||
const hasDelayed = this.hasDelayedItems(session, now);
|
||||
if (!hasQueued && !hasDelayed) {
|
||||
this.emit("run-complete");
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
await sleep(this.slots.size >= maxParallel ? 170 : loopIntervalMs);
|
||||
}
|
||||
|
||||
this.running = false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop the scheduler loop (bumps generation to exit).
|
||||
*/
|
||||
stop(): void {
|
||||
this.generation++;
|
||||
this.running = false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Pause/unpause slot allocation.
|
||||
*/
|
||||
setPaused(paused: boolean): void {
|
||||
this.paused = paused;
|
||||
}
|
||||
|
||||
get isPaused(): boolean {
|
||||
return this.paused;
|
||||
}
|
||||
|
||||
get isRunning(): boolean {
|
||||
return this.running;
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Slot management
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Register an item as actively downloading.
|
||||
*/
|
||||
claimSlot(itemId: string, packageId: string, abortController: AbortController): ActiveSlot {
|
||||
const slot: ActiveSlot = {
|
||||
itemId,
|
||||
packageId,
|
||||
abortController,
|
||||
abortReason: "none",
|
||||
resumable: true,
|
||||
lastHeartbeatAt: Date.now(),
|
||||
bytesAtHeartbeat: 0,
|
||||
blockedOnDiskWrite: false,
|
||||
blockedOnDiskSince: 0,
|
||||
};
|
||||
this.slots.set(itemId, slot);
|
||||
return slot;
|
||||
}
|
||||
|
||||
/**
|
||||
* Release a slot (download finished/failed/cancelled).
|
||||
*/
|
||||
releaseSlot(itemId: string): void {
|
||||
this.slots.delete(itemId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get active slot for an item.
|
||||
*/
|
||||
getSlot(itemId: string): ActiveSlot | undefined {
|
||||
return this.slots.get(itemId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all active slots.
|
||||
*/
|
||||
getActiveSlots(): Map<string, ActiveSlot> {
|
||||
return this.slots;
|
||||
}
|
||||
|
||||
get activeCount(): number {
|
||||
return this.slots.size;
|
||||
}
|
||||
|
||||
hasCapacity(): boolean {
|
||||
return this.slots.size < Math.max(1, this.config.maxParallel);
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Heartbeat
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Record a heartbeat from an active download.
|
||||
*/
|
||||
heartbeat(itemId: string, downloadedBytes: number): void {
|
||||
const slot = this.slots.get(itemId);
|
||||
if (slot) {
|
||||
slot.lastHeartbeatAt = Date.now();
|
||||
slot.bytesAtHeartbeat = downloadedBytes;
|
||||
}
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Retry scheduling
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Schedule a retry delay for an item.
|
||||
*/
|
||||
scheduleRetry(itemId: string, delayMs: number): void {
|
||||
this.retryDelays.set(itemId, Date.now() + Math.max(0, delayMs));
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if an item is still delayed.
|
||||
*/
|
||||
isDelayed(itemId: string, now?: number): boolean {
|
||||
const readyAt = this.retryDelays.get(itemId);
|
||||
if (!readyAt) return false;
|
||||
return readyAt > (now ?? Date.now());
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear retry delay for an item.
|
||||
*/
|
||||
clearRetryDelay(itemId: string): void {
|
||||
this.retryDelays.delete(itemId);
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Provider cooldowns
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Apply a cooldown to a provider.
|
||||
*/
|
||||
applyProviderCooldown(provider: string, cooldownMs: number): void {
|
||||
const existing = this.providerCooldowns.get(provider) || { cooldownUntil: 0, failureCount: 0 };
|
||||
existing.cooldownUntil = Date.now() + cooldownMs;
|
||||
existing.failureCount++;
|
||||
this.providerCooldowns.set(provider, existing);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get remaining cooldown for a provider (ms). 0 = not in cooldown.
|
||||
*/
|
||||
getProviderCooldownRemaining(provider: string): number {
|
||||
const entry = this.providerCooldowns.get(provider);
|
||||
if (!entry) return 0;
|
||||
const remaining = entry.cooldownUntil - Date.now();
|
||||
if (remaining <= 0) {
|
||||
entry.failureCount = 0;
|
||||
return 0;
|
||||
}
|
||||
return remaining;
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear cooldown for a provider (after success).
|
||||
*/
|
||||
clearProviderCooldown(provider: string): void {
|
||||
this.providerCooldowns.delete(provider);
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Reconnect
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Enter reconnect wait mode (429/503 backoff).
|
||||
*/
|
||||
setReconnectWait(durationMs: number): void {
|
||||
this.reconnectUntil = Date.now() + durationMs;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if currently in reconnect wait.
|
||||
*/
|
||||
isReconnecting(): boolean {
|
||||
return this.reconnectUntil > Date.now();
|
||||
}
|
||||
|
||||
/**
|
||||
* Get remaining reconnect wait time (ms).
|
||||
*/
|
||||
getReconnectRemaining(): number {
|
||||
return Math.max(0, this.reconnectUntil - Date.now());
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Abort helpers
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Abort a specific item's download.
|
||||
*/
|
||||
abortItem(itemId: string, reason: ActiveSlot["abortReason"]): void {
|
||||
const slot = this.slots.get(itemId);
|
||||
if (slot) {
|
||||
slot.abortReason = reason;
|
||||
slot.abortController.abort(reason);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Abort all active downloads.
|
||||
*/
|
||||
abortAll(reason: ActiveSlot["abortReason"]): void {
|
||||
for (const slot of this.slots.values()) {
|
||||
slot.abortReason = reason;
|
||||
slot.abortController.abort(reason);
|
||||
}
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Private: item selection
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
private findNextItem(session: SessionState, now: number): SlotRequest | null {
|
||||
const priorities: PackagePriority[] = ["high", "normal", "low"];
|
||||
|
||||
for (const prio of priorities) {
|
||||
for (const packageId of session.packageOrder) {
|
||||
const pkg = session.packages[packageId];
|
||||
if (!pkg || pkg.cancelled || !pkg.enabled) continue;
|
||||
if ((pkg.priority || "normal") !== prio) continue;
|
||||
if (this.scopedPackageIds.size > 0 && !this.scopedPackageIds.has(packageId)) continue;
|
||||
|
||||
for (const itemId of pkg.itemIds) {
|
||||
const item = session.items[itemId];
|
||||
if (!item) continue;
|
||||
if (item.status !== "queued" && item.status !== "reconnect_wait") continue;
|
||||
if (this.slots.has(itemId)) continue;
|
||||
|
||||
// Check retry delay
|
||||
const retryAt = this.retryDelays.get(itemId);
|
||||
if (retryAt && retryAt > now) continue;
|
||||
if (retryAt && retryAt <= now) this.retryDelays.delete(itemId);
|
||||
|
||||
return { itemId, packageId };
|
||||
}
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private hasQueuedItems(session: SessionState, now: number): boolean {
|
||||
for (const packageId of session.packageOrder) {
|
||||
const pkg = session.packages[packageId];
|
||||
if (!pkg || pkg.cancelled || !pkg.enabled) continue;
|
||||
if (this.scopedPackageIds.size > 0 && !this.scopedPackageIds.has(packageId)) continue;
|
||||
|
||||
for (const itemId of pkg.itemIds) {
|
||||
const item = session.items[itemId];
|
||||
if (!item) continue;
|
||||
const retryAt = this.retryDelays.get(itemId);
|
||||
if (retryAt && retryAt > now) continue;
|
||||
if (item.status === "queued" || item.status === "reconnect_wait") return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
private hasDelayedItems(session: SessionState, now: number): boolean {
|
||||
for (const [itemId, readyAt] of this.retryDelays) {
|
||||
if (readyAt <= now) continue;
|
||||
const item = session.items[itemId];
|
||||
if (!item) continue;
|
||||
if (item.status !== "queued" && item.status !== "reconnect_wait") continue;
|
||||
const pkg = session.packages[item.packageId];
|
||||
if (!pkg || pkg.cancelled || !pkg.enabled) continue;
|
||||
if (this.scopedPackageIds.size > 0 && !this.scopedPackageIds.has(item.packageId)) continue;
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Private: stall detection
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
private checkStalls(now: number): void {
|
||||
if (this.config.stallTimeoutMs <= 0) return;
|
||||
|
||||
for (const slot of this.slots.values()) {
|
||||
if (slot.blockedOnDiskWrite) continue; // Don't count disk waits
|
||||
const idleMs = now - slot.lastHeartbeatAt;
|
||||
if (idleMs > this.config.stallTimeoutMs) {
|
||||
this.emit("stall-detected", { itemId: slot.itemId, idleMs });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private runGlobalWatchdog(now: number): void {
|
||||
if (this.config.globalStallWatchdogMs <= 0) return;
|
||||
if (this.slots.size === 0) return;
|
||||
|
||||
// Sum total bytes across all active downloads
|
||||
let totalBytes = 0;
|
||||
let allDiskBlocked = true;
|
||||
for (const slot of this.slots.values()) {
|
||||
totalBytes += slot.bytesAtHeartbeat;
|
||||
if (!slot.blockedOnDiskWrite) allDiskBlocked = false;
|
||||
}
|
||||
|
||||
// If all downloads are disk-blocked, don't trigger watchdog
|
||||
if (allDiskBlocked) return;
|
||||
|
||||
if (totalBytes > this.lastGlobalProgressBytes) {
|
||||
this.lastGlobalProgressBytes = totalBytes;
|
||||
this.lastGlobalProgressAt = now;
|
||||
} else if (now - this.lastGlobalProgressAt > this.config.globalStallWatchdogMs) {
|
||||
const stalledIds = [...this.slots.values()]
|
||||
.filter(s => !s.blockedOnDiskWrite)
|
||||
.map(s => s.itemId);
|
||||
this.emit("global-stall", { itemIds: stalledIds });
|
||||
this.lastGlobalProgressAt = now; // Reset to avoid rapid-fire events
|
||||
}
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Private: cleanup
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
private cleanupStaleRetryDelays(now: number): void {
|
||||
for (const [itemId, readyAt] of this.retryDelays) {
|
||||
if (readyAt <= now) {
|
||||
this.retryDelays.delete(itemId);
|
||||
}
|
||||
}
|
||||
// Cleanup stale provider cooldowns
|
||||
for (const [provider, entry] of this.providerCooldowns) {
|
||||
if (entry.cooldownUntil <= now) {
|
||||
this.providerCooldowns.delete(provider);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Helper
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
function sleep(ms: number): Promise<void> {
|
||||
return new Promise(resolve => setTimeout(resolve, ms));
|
||||
}
|
||||
@@ -0,0 +1,732 @@
|
||||
/**
|
||||
* stream-writer.ts — HTTP streaming with validated resume, NTFS-aligned
|
||||
* buffered writing, stall detection, and speed limiting.
|
||||
*
|
||||
* This module is a pure function with no dependency on DownloadManager state.
|
||||
* All side effects happen through callbacks.
|
||||
*/
|
||||
|
||||
import fs from "node:fs";
|
||||
import path from "node:path";
|
||||
import { DownloadError, DownloadErrorKind, classifyFetchError, classifyHttpStatus, classifyRangeIgnored } from "./error-classifier";
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Constants
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const WRITE_BUFFER_SIZE = 512 * 1024;
|
||||
const ALLOCATION_UNIT_SIZE = 4096;
|
||||
const STREAM_HIGH_WATER_MARK = 512 * 1024;
|
||||
const WRITE_FLUSH_TIMEOUT_MS = 2000;
|
||||
const DISK_BUSY_THRESHOLD_MS = 300;
|
||||
const DEFAULT_DRAIN_TIMEOUT_MS = 300_000; // 5 min
|
||||
const MIN_LEGITIMATE_FILE_BYTES = 512;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Interfaces
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export interface StreamOptions {
|
||||
/** Direct download URL. */
|
||||
url: string;
|
||||
/** Target file path on disk. */
|
||||
targetPath: string;
|
||||
/** Expected total file size (from unrestrict or previous response). null = unknown. */
|
||||
expectedBytes: number | null;
|
||||
/** Previously downloaded bytes (tracked by caller for resume validation). */
|
||||
trackedDownloadedBytes: number;
|
||||
/** Stall timeout: abort if no data received for this long (ms). 0 = disabled. */
|
||||
stallTimeoutMs: number;
|
||||
/** Connection timeout (ms). 0 = disabled. */
|
||||
connectTimeoutMs: number;
|
||||
/** Skip TLS verification for this request. */
|
||||
skipTlsVerify: boolean;
|
||||
/** Speed limit in bytes/sec. 0 = no limit. */
|
||||
speedLimitBps: number;
|
||||
/** Abort signal from caller. */
|
||||
signal: AbortSignal;
|
||||
/** Called periodically with download progress. */
|
||||
onProgress: (downloadedBytes: number, totalBytes: number | null, speedBps: number) => void;
|
||||
/** Called every ~1-3s even during slow transfer, for watchdog purposes. */
|
||||
onHeartbeat: () => void;
|
||||
/** Called once after HTTP response to report resumability. */
|
||||
onResumable: (resumable: boolean) => void;
|
||||
/** Called if Content-Disposition provides a different filename. */
|
||||
onFileNameOverride?: (newName: string) => void;
|
||||
/** Called to log events. */
|
||||
onLog?: (level: "INFO" | "WARN" | "ERROR", message: string, fields?: Record<string, unknown>) => void;
|
||||
/** Called when disk is busy (backpressure). */
|
||||
onDiskBusy?: (busy: boolean) => void;
|
||||
/** Maximum inner retries on same direct URL before escalating. Default: 3. */
|
||||
maxDirectUrlRetries?: number;
|
||||
/** Low throughput timeout: abort if < minBytes in this window (ms). 0 = disabled. */
|
||||
lowThroughputTimeoutMs?: number;
|
||||
/** Minimum bytes required in lowThroughput window. */
|
||||
lowThroughputMinBytes?: number;
|
||||
/** Whether the target filename looks like a large binary (archive, video, etc.). */
|
||||
isLargeBinary?: boolean;
|
||||
}
|
||||
|
||||
export interface StreamResult {
|
||||
/** Total bytes of the complete file. */
|
||||
totalBytes: number | null;
|
||||
/** Bytes written in this session (not counting resume). */
|
||||
downloadedBytes: number;
|
||||
/** Whether the server supports Range/resume. */
|
||||
resumable: boolean;
|
||||
/** If Content-Disposition provided a new filename. */
|
||||
fileName?: string;
|
||||
/** True if the download completed (all bytes received). */
|
||||
completed: boolean;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Main function
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export async function streamToFile(opts: StreamOptions): Promise<StreamResult> {
|
||||
const {
|
||||
url,
|
||||
targetPath,
|
||||
expectedBytes,
|
||||
trackedDownloadedBytes,
|
||||
stallTimeoutMs,
|
||||
connectTimeoutMs,
|
||||
skipTlsVerify,
|
||||
speedLimitBps,
|
||||
signal,
|
||||
onProgress,
|
||||
onHeartbeat,
|
||||
onResumable,
|
||||
onFileNameOverride,
|
||||
onLog,
|
||||
onDiskBusy,
|
||||
lowThroughputTimeoutMs = 0,
|
||||
lowThroughputMinBytes = 64 * 1024,
|
||||
isLargeBinary = false,
|
||||
} = opts;
|
||||
|
||||
const maxAttempts = opts.maxDirectUrlRetries ?? 3;
|
||||
const log = onLog ?? (() => {});
|
||||
let lastError: DownloadError | null = null;
|
||||
let overriddenFileName: string | undefined;
|
||||
let effectiveTargetPath = targetPath;
|
||||
|
||||
for (let attempt = 1; attempt <= maxAttempts; attempt++) {
|
||||
// ----- Pre-resume validation -----
|
||||
let existingBytes = 0;
|
||||
try {
|
||||
const stat = await fs.promises.stat(effectiveTargetPath);
|
||||
existingBytes = stat.size;
|
||||
} catch {
|
||||
// file does not exist
|
||||
}
|
||||
|
||||
// Guard against pre-allocated sparse files: if file is much larger than
|
||||
// what we actually wrote, truncate to tracked bytes.
|
||||
if (existingBytes > 0 && trackedDownloadedBytes > 0 && existingBytes > trackedDownloadedBytes + 1_048_576) {
|
||||
try {
|
||||
await fs.promises.truncate(effectiveTargetPath, trackedDownloadedBytes);
|
||||
existingBytes = trackedDownloadedBytes;
|
||||
log("WARN", "Sparse file truncated to tracked bytes", {
|
||||
existingBytes: existingBytes,
|
||||
trackedDownloadedBytes,
|
||||
});
|
||||
} catch { /* best-effort */ }
|
||||
}
|
||||
|
||||
// If file is smaller than tracked bytes but nonzero — mismatch, could be
|
||||
// corruption from a crash. For small mismatches (<1MB), restart fresh.
|
||||
if (existingBytes > 0 && trackedDownloadedBytes > 0 && existingBytes < trackedDownloadedBytes - 1_048_576) {
|
||||
try {
|
||||
await fs.promises.rm(effectiveTargetPath, { force: true });
|
||||
existingBytes = 0;
|
||||
log("WARN", "File smaller than tracked bytes — deleted for fresh start", {
|
||||
existingBytes,
|
||||
trackedDownloadedBytes,
|
||||
});
|
||||
} catch { /* best-effort */ }
|
||||
}
|
||||
|
||||
// ----- HTTP request -----
|
||||
const headers: Record<string, string> = {};
|
||||
if (existingBytes > 0) {
|
||||
headers.Range = `bytes=${existingBytes}-`;
|
||||
}
|
||||
|
||||
log("INFO", "HTTP download attempt", {
|
||||
attempt,
|
||||
maxAttempts,
|
||||
url,
|
||||
targetPath: effectiveTargetPath,
|
||||
existingBytes,
|
||||
rangeHeader: headers.Range || "",
|
||||
});
|
||||
|
||||
// Check abort before connecting
|
||||
checkAborted(signal);
|
||||
|
||||
let response: Response;
|
||||
let connectTimer: NodeJS.Timeout | null = null;
|
||||
const connectAbortController = new AbortController();
|
||||
|
||||
// TLS skip management
|
||||
if (skipTlsVerify) acquireTlsSkip();
|
||||
try {
|
||||
if (connectTimeoutMs > 0) {
|
||||
connectTimer = setTimeout(() => connectAbortController.abort("connect_timeout"), connectTimeoutMs);
|
||||
}
|
||||
response = await fetch(url, {
|
||||
method: "GET",
|
||||
headers,
|
||||
signal: AbortSignal.any([signal, connectAbortController.signal]),
|
||||
});
|
||||
} catch (error) {
|
||||
// Rethrow abort errors
|
||||
if (signal.aborted) throw error;
|
||||
if (String(error).includes("connect_timeout")) {
|
||||
throw new DownloadError(DownloadErrorKind.ConnectTimeout, "Connection timeout", { originalError: error instanceof Error ? error : undefined });
|
||||
}
|
||||
lastError = classifyFetchError(error);
|
||||
log("WARN", "HTTP connection failed", { attempt, error: lastError.message });
|
||||
if (attempt < maxAttempts) {
|
||||
await sleep(retryDelayWithJitter(attempt, 200));
|
||||
continue;
|
||||
}
|
||||
throw lastError;
|
||||
} finally {
|
||||
if (skipTlsVerify) releaseTlsSkip();
|
||||
if (connectTimer) clearTimeout(connectTimer);
|
||||
}
|
||||
|
||||
// ----- HTTP status handling -----
|
||||
if (!response.ok && response.status !== 206) {
|
||||
if (response.status === 416 && existingBytes > 0) {
|
||||
const result = await handle416(response, existingBytes, expectedBytes, log);
|
||||
if (result) {
|
||||
onResumable(true);
|
||||
return result;
|
||||
}
|
||||
// Not complete — delete and retry
|
||||
try { await fs.promises.rm(effectiveTargetPath, { force: true }); } catch {}
|
||||
if (attempt < maxAttempts) {
|
||||
await sleep(retryDelayWithJitter(attempt, 200));
|
||||
continue;
|
||||
}
|
||||
throw classifyHttpStatus({ status: 416, existingBytes });
|
||||
}
|
||||
|
||||
const responseText = await response.text().catch(() => "");
|
||||
lastError = classifyHttpStatus({
|
||||
status: response.status,
|
||||
statusText: response.statusText,
|
||||
responseText,
|
||||
existingBytes,
|
||||
});
|
||||
log("WARN", "HTTP response not OK", { attempt, status: response.status, error: lastError.message });
|
||||
if (attempt < maxAttempts) {
|
||||
await sleep(retryDelayWithJitter(attempt, 250));
|
||||
continue;
|
||||
}
|
||||
throw lastError;
|
||||
}
|
||||
|
||||
// ----- Response analysis -----
|
||||
const acceptRanges = (response.headers.get("accept-ranges") || "").toLowerCase().includes("bytes");
|
||||
const resumable = response.status === 206 || acceptRanges;
|
||||
onResumable(resumable);
|
||||
|
||||
// Detect server ignoring Range header (200 instead of 206)
|
||||
if (existingBytes > 0 && response.status === 200) {
|
||||
const contentLength = Number(response.headers.get("content-length") || 0);
|
||||
try { await response.body?.cancel(); } catch {}
|
||||
log("WARN", "Server ignored Range header", { existingBytes, contentLength });
|
||||
throw classifyRangeIgnored(existingBytes, contentLength);
|
||||
}
|
||||
|
||||
// Parse total size
|
||||
const rawContentLength = Number(response.headers.get("content-length") || 0);
|
||||
const contentLength = Number.isFinite(rawContentLength) && rawContentLength > 0 ? rawContentLength : 0;
|
||||
const totalFromRange = parseContentRangeTotal(response.headers.get("content-range"));
|
||||
|
||||
let totalBytes = expectedBytes;
|
||||
if (!totalBytes || totalBytes <= 0) {
|
||||
if (totalFromRange) totalBytes = totalFromRange;
|
||||
else if (contentLength > 0) {
|
||||
totalBytes = response.status === 206 ? existingBytes + contentLength : contentLength;
|
||||
}
|
||||
}
|
||||
|
||||
// Content-Disposition filename (only on fresh downloads)
|
||||
if (existingBytes === 0 && onFileNameOverride) {
|
||||
const rawName = parseContentDispositionFilename(response.headers.get("content-disposition")).trim();
|
||||
const fromHeader = rawName ? sanitizeFilename(rawName) : "";
|
||||
if (fromHeader && !looksLikeOpaqueFilename(fromHeader) && fromHeader !== path.basename(targetPath)) {
|
||||
overriddenFileName = fromHeader;
|
||||
const newPath = path.join(path.dirname(targetPath), fromHeader);
|
||||
effectiveTargetPath = newPath;
|
||||
onFileNameOverride(fromHeader);
|
||||
log("INFO", "Filename from Content-Disposition", { fromHeader, newPath });
|
||||
}
|
||||
}
|
||||
|
||||
const writeMode = existingBytes > 0 && response.status === 206 ? "a" : "w";
|
||||
|
||||
log("INFO", "HTTP response accepted", {
|
||||
attempt, status: response.status, resumable, contentLength,
|
||||
totalFromRange, totalBytes, writeMode,
|
||||
});
|
||||
|
||||
// If starting fresh, delete existing file
|
||||
if (writeMode === "w" && existingBytes > 0) {
|
||||
try { await fs.promises.rm(effectiveTargetPath, { force: true }); } catch {}
|
||||
}
|
||||
|
||||
await fs.promises.mkdir(path.dirname(effectiveTargetPath), { recursive: true });
|
||||
|
||||
// ----- Sparse pre-allocation (Windows) -----
|
||||
let preAllocated = false;
|
||||
if (writeMode === "w" && totalBytes && totalBytes > 0 && process.platform === "win32") {
|
||||
try {
|
||||
const fd = await fs.promises.open(effectiveTargetPath, "w");
|
||||
try { await fd.truncate(totalBytes); preAllocated = true; } finally { await fd.close(); }
|
||||
} catch { /* best-effort */ }
|
||||
}
|
||||
|
||||
// ----- Streaming write -----
|
||||
const stream = fs.createWriteStream(effectiveTargetPath, {
|
||||
flags: preAllocated ? "r+" : writeMode === "a" ? "a" : "w",
|
||||
start: preAllocated ? 0 : undefined,
|
||||
highWaterMark: STREAM_HIGH_WATER_MARK,
|
||||
});
|
||||
let written = writeMode === "a" ? existingBytes : 0;
|
||||
let windowBytes = 0;
|
||||
let windowStarted = nowMs();
|
||||
let bodyError: unknown = null;
|
||||
|
||||
// Write buffer with 4KB NTFS alignment
|
||||
const writeBuf = Buffer.allocUnsafe(WRITE_BUFFER_SIZE);
|
||||
let writeBufPos = 0;
|
||||
let lastFlushAt = nowMs();
|
||||
|
||||
let diskBusySince = 0;
|
||||
let diskBusyNotified = false;
|
||||
const drainTimeoutMs = Math.max(30_000, Math.min(DEFAULT_DRAIN_TIMEOUT_MS, stallTimeoutMs > 0 ? stallTimeoutMs * 12 : 120_000));
|
||||
|
||||
// --- waitDrain ---
|
||||
const waitDrain = (): Promise<void> => new Promise((resolve, reject) => {
|
||||
if (signal.aborted) { reject(new Error("aborted")); return; }
|
||||
|
||||
if (onDiskBusy && !diskBusyNotified) {
|
||||
onDiskBusy(true);
|
||||
diskBusyNotified = true;
|
||||
}
|
||||
|
||||
let settled = false;
|
||||
const timeoutId = setTimeout(() => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
cleanup();
|
||||
reject(new DownloadError(DownloadErrorKind.WriteDrainTimeout, "write_drain_timeout"));
|
||||
}, drainTimeoutMs);
|
||||
|
||||
const cleanup = (): void => {
|
||||
clearTimeout(timeoutId);
|
||||
if (onDiskBusy && diskBusyNotified) {
|
||||
onDiskBusy(false);
|
||||
diskBusyNotified = false;
|
||||
}
|
||||
stream.off("drain", onDrain);
|
||||
stream.off("error", onErr);
|
||||
signal.removeEventListener("abort", onAbort);
|
||||
};
|
||||
const onDrain = (): void => { if (!settled) { settled = true; cleanup(); resolve(); } };
|
||||
const onErr = (e: Error): void => { if (!settled) { settled = true; cleanup(); reject(e); } };
|
||||
const onAbort = (): void => { if (!settled) { settled = true; cleanup(); reject(new Error("aborted")); } };
|
||||
|
||||
stream.once("drain", onDrain);
|
||||
stream.once("error", onErr);
|
||||
signal.addEventListener("abort", onAbort, { once: true });
|
||||
});
|
||||
|
||||
// --- aligned flush ---
|
||||
const alignedFlush = async (final = false): Promise<void> => {
|
||||
if (writeBufPos === 0) return;
|
||||
let toWrite = writeBufPos;
|
||||
if (!final && toWrite > ALLOCATION_UNIT_SIZE) {
|
||||
toWrite = toWrite - (toWrite % ALLOCATION_UNIT_SIZE);
|
||||
}
|
||||
const slice = Buffer.from(writeBuf.subarray(0, toWrite));
|
||||
if (!stream.write(slice)) {
|
||||
await waitDrain();
|
||||
}
|
||||
if (toWrite < writeBufPos) {
|
||||
writeBuf.copy(writeBuf, 0, toWrite, writeBufPos);
|
||||
}
|
||||
writeBufPos -= toWrite;
|
||||
lastFlushAt = nowMs();
|
||||
};
|
||||
|
||||
try {
|
||||
const body = response.body;
|
||||
if (!body) throw new DownloadError(DownloadErrorKind.Unknown, "Empty response body");
|
||||
|
||||
const reader = body.getReader();
|
||||
let lastDataAt = nowMs();
|
||||
|
||||
// Throughput window for low-throughput detection
|
||||
let throughputWindowStart = nowMs();
|
||||
let throughputWindowBytes = 0;
|
||||
|
||||
// Speed limiter state
|
||||
let speedLimitWindowStart = nowMs();
|
||||
let speedLimitWindowBytes = 0;
|
||||
|
||||
// Heartbeat timer
|
||||
const heartbeatInterval = setInterval(() => {
|
||||
if (!signal.aborted) onHeartbeat();
|
||||
}, 2000);
|
||||
|
||||
// readWithTimeout
|
||||
const readWithTimeout = async (): Promise<ReadableStreamReadResult<Uint8Array>> => {
|
||||
if (stallTimeoutMs <= 0) return reader.read();
|
||||
return new Promise<ReadableStreamReadResult<Uint8Array>>((resolve, reject) => {
|
||||
let settled = false;
|
||||
const timer = setTimeout(() => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
reject(new DownloadError(DownloadErrorKind.Timeout, "stall_timeout"));
|
||||
}, stallTimeoutMs);
|
||||
reader.read().then(result => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
clearTimeout(timer);
|
||||
resolve(result);
|
||||
}).catch(err => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
clearTimeout(timer);
|
||||
reject(err);
|
||||
});
|
||||
});
|
||||
};
|
||||
|
||||
try {
|
||||
while (true) {
|
||||
const { done, value } = await readWithTimeout();
|
||||
if (done) break;
|
||||
|
||||
lastDataAt = nowMs();
|
||||
checkAborted(signal);
|
||||
|
||||
const buffer = Buffer.isBuffer(value) ? value : Buffer.from(value.buffer, value.byteOffset, value.byteLength);
|
||||
|
||||
// Speed limiting
|
||||
if (speedLimitBps > 0) {
|
||||
speedLimitWindowBytes += buffer.length;
|
||||
const elapsed = (nowMs() - speedLimitWindowStart) / 1000;
|
||||
if (elapsed > 0.1) {
|
||||
const currentRate = speedLimitWindowBytes / elapsed;
|
||||
if (currentRate > speedLimitBps) {
|
||||
const sleepMs = Math.floor(((speedLimitWindowBytes / speedLimitBps) - elapsed) * 1000);
|
||||
if (sleepMs > 10) await sleep(Math.min(sleepMs, 1000));
|
||||
}
|
||||
if (elapsed >= 1) {
|
||||
speedLimitWindowStart = nowMs();
|
||||
speedLimitWindowBytes = 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
checkAborted(signal);
|
||||
|
||||
// Buffer incoming data for aligned writes
|
||||
let srcOffset = 0;
|
||||
while (srcOffset < buffer.length) {
|
||||
const space = WRITE_BUFFER_SIZE - writeBufPos;
|
||||
const toCopy = Math.min(space, buffer.length - srcOffset);
|
||||
buffer.copy(writeBuf, writeBufPos, srcOffset, srcOffset + toCopy);
|
||||
writeBufPos += toCopy;
|
||||
srcOffset += toCopy;
|
||||
if (writeBufPos >= Math.floor(WRITE_BUFFER_SIZE * 0.80)) {
|
||||
await alignedFlush(false);
|
||||
}
|
||||
}
|
||||
|
||||
// Time-based flush
|
||||
if (writeBufPos > 0 && nowMs() - lastFlushAt >= WRITE_FLUSH_TIMEOUT_MS) {
|
||||
await alignedFlush(false);
|
||||
}
|
||||
|
||||
// Proactive disk-busy detection
|
||||
if (stream.writableLength > 0) {
|
||||
if (diskBusySince === 0) diskBusySince = nowMs();
|
||||
} else {
|
||||
diskBusySince = 0;
|
||||
if (diskBusyNotified && onDiskBusy) {
|
||||
onDiskBusy(false);
|
||||
diskBusyNotified = false;
|
||||
}
|
||||
}
|
||||
|
||||
written += buffer.length;
|
||||
windowBytes += buffer.length;
|
||||
throughputWindowBytes += buffer.length;
|
||||
|
||||
// Early completion: all expected bytes received
|
||||
const expectedTotal = totalBytes && totalBytes > 0 ? totalBytes : 0;
|
||||
const expectedFromResponse = contentLength > 0 ? contentLength : 0;
|
||||
if (expectedTotal > 0 && written >= expectedTotal) break;
|
||||
if (expectedTotal === 0 && expectedFromResponse > 0 && (written - (writeMode === "a" ? existingBytes : 0)) >= expectedFromResponse) break;
|
||||
|
||||
// Low throughput check
|
||||
const now = nowMs();
|
||||
if (lowThroughputTimeoutMs > 0 && now - throughputWindowStart >= lowThroughputTimeoutMs) {
|
||||
if (throughputWindowBytes < lowThroughputMinBytes) {
|
||||
throw new DownloadError(DownloadErrorKind.Timeout, `slow_throughput:${throughputWindowBytes}/${lowThroughputMinBytes}`);
|
||||
}
|
||||
throughputWindowStart = now;
|
||||
throughputWindowBytes = 0;
|
||||
}
|
||||
|
||||
// Speed calculation and progress reporting
|
||||
const elapsed = Math.max((nowMs() - windowStarted) / 1000, 0.2);
|
||||
const speed = windowBytes / elapsed;
|
||||
if (elapsed >= 0.5) {
|
||||
windowStarted = nowMs();
|
||||
windowBytes = 0;
|
||||
}
|
||||
|
||||
const diskBusy = diskBusySince > 0 && nowMs() - diskBusySince >= DISK_BUSY_THRESHOLD_MS;
|
||||
onProgress(written, totalBytes, diskBusy ? 0 : Math.floor(speed));
|
||||
}
|
||||
} finally {
|
||||
clearInterval(heartbeatInterval);
|
||||
try { await reader.cancel().catch(() => {}); reader.releaseLock(); } catch {}
|
||||
}
|
||||
} catch (error) {
|
||||
bodyError = error;
|
||||
log("WARN", "Download body error", { attempt, error: errorMessage(error) });
|
||||
} finally {
|
||||
// Flush remaining buffered data
|
||||
try { await alignedFlush(true); } catch (e) { if (!bodyError) bodyError = e; }
|
||||
|
||||
// Close stream
|
||||
try {
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
if (stream.closed || stream.destroyed) { resolve(); return; }
|
||||
const onDone = (): void => { stream.off("error", onErr); resolve(); };
|
||||
const onErr = (e: Error): void => { stream.off("finish", onDone); stream.off("close", onDone); reject(e); };
|
||||
stream.once("finish", onDone);
|
||||
stream.once("close", onDone);
|
||||
stream.once("error", onErr);
|
||||
stream.end();
|
||||
});
|
||||
} catch (closeErr) {
|
||||
if (!stream.destroyed) stream.destroy();
|
||||
if (!bodyError) throw closeErr;
|
||||
log("WARN", "Stream close error suppressed", { error: errorMessage(closeErr) });
|
||||
}
|
||||
|
||||
if (!stream.destroyed) stream.destroy();
|
||||
|
||||
// fsync for pre-allocated files
|
||||
if (!bodyError && preAllocated) {
|
||||
try {
|
||||
const fd = await fs.promises.open(effectiveTargetPath, "r");
|
||||
try { await fd.datasync(); } finally { await fd.close(); }
|
||||
} catch { /* best-effort */ }
|
||||
}
|
||||
|
||||
// Truncate pre-allocated file to actual written bytes on error
|
||||
if (bodyError && preAllocated && totalBytes && written < totalBytes) {
|
||||
try { await fs.promises.truncate(effectiveTargetPath, written); } catch {}
|
||||
}
|
||||
|
||||
if (bodyError) {
|
||||
// On error: truncate pre-allocated sparse file
|
||||
if (preAllocated && totalBytes && written < totalBytes) {
|
||||
try { await fs.promises.truncate(effectiveTargetPath, written); } catch {}
|
||||
}
|
||||
if (signal.aborted) throw bodyError;
|
||||
lastError = bodyError instanceof DownloadError ? bodyError : classifyFetchError(bodyError);
|
||||
if (attempt < maxAttempts) {
|
||||
log("WARN", "Retrying after body error", { attempt, error: lastError.message });
|
||||
await sleep(retryDelayWithJitter(attempt, 250));
|
||||
continue;
|
||||
}
|
||||
throw lastError;
|
||||
}
|
||||
}
|
||||
|
||||
// ----- Post-download validation -----
|
||||
|
||||
// Tiny file detection (hoster error pages disguised as downloads)
|
||||
if (written > 0 && written < MIN_LEGITIMATE_FILE_BYTES) {
|
||||
let snippet = "";
|
||||
try { snippet = (await fs.promises.readFile(effectiveTargetPath, "utf8")).slice(0, 200).replace(/[\r\n]+/g, " ").trim(); } catch {}
|
||||
try { await fs.promises.rm(effectiveTargetPath, { force: true }); } catch {}
|
||||
log("WARN", `Tiny download detected (${written} bytes)`, { snippet });
|
||||
throw new DownloadError(DownloadErrorKind.ServerError,
|
||||
`Download too small (${written} B) — hoster error page?${snippet ? ` Content: "${snippet}"` : ""}`,
|
||||
{ httpStatus: 200 });
|
||||
}
|
||||
|
||||
// Underflow detection
|
||||
if (totalBytes && totalBytes > 0 && written < totalBytes) {
|
||||
const shortfall = totalBytes - written;
|
||||
if (preAllocated) {
|
||||
try { await fs.promises.truncate(effectiveTargetPath, written); } catch {}
|
||||
}
|
||||
if (isLargeBinary || shortfall > ALLOCATION_UNIT_SIZE) {
|
||||
log("WARN", "Download underflow", { expected: totalBytes, received: written, shortfall });
|
||||
throw new DownloadError(DownloadErrorKind.FileTruncated,
|
||||
`download_underflow:${written}/${totalBytes}`,
|
||||
{ context: { written, totalBytes, shortfall } });
|
||||
}
|
||||
}
|
||||
|
||||
// Truncate pre-allocated file to actual size
|
||||
if (preAllocated && totalBytes && written < totalBytes) {
|
||||
try { await fs.promises.truncate(effectiveTargetPath, written); } catch {}
|
||||
}
|
||||
|
||||
log("INFO", "Download complete", { attempt, resumable, written, totalBytes, targetPath: effectiveTargetPath });
|
||||
|
||||
return {
|
||||
totalBytes,
|
||||
downloadedBytes: written,
|
||||
resumable,
|
||||
fileName: overriddenFileName,
|
||||
completed: true,
|
||||
};
|
||||
}
|
||||
|
||||
// All attempts exhausted
|
||||
throw lastError ?? new DownloadError(DownloadErrorKind.Unknown, "Download failed — all attempts exhausted");
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// HTTP 416 handler
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
async function handle416(
|
||||
response: Response,
|
||||
existingBytes: number,
|
||||
expectedBytes: number | null,
|
||||
log: (level: "INFO" | "WARN" | "ERROR", msg: string, fields?: Record<string, unknown>) => void,
|
||||
): Promise<StreamResult | null> {
|
||||
await response.arrayBuffer().catch(() => undefined);
|
||||
const rangeTotal = parseContentRangeTotal(response.headers.get("content-range"));
|
||||
const resolvedTotal = (expectedBytes && expectedBytes > 0) ? expectedBytes : rangeTotal;
|
||||
|
||||
// File is already complete
|
||||
if (resolvedTotal && existingBytes === resolvedTotal) {
|
||||
log("INFO", "HTTP 416 treated as complete", { existingBytes, resolvedTotal });
|
||||
return {
|
||||
totalBytes: resolvedTotal,
|
||||
downloadedBytes: existingBytes,
|
||||
resumable: true,
|
||||
completed: true,
|
||||
};
|
||||
}
|
||||
|
||||
// No size info but substantial data — assume complete to avoid deleting multi-GB files
|
||||
if (!resolvedTotal && existingBytes > 1_048_576) {
|
||||
log("WARN", "HTTP 416 without size info — assuming complete", { existingBytes });
|
||||
return {
|
||||
totalBytes: existingBytes,
|
||||
downloadedBytes: existingBytes,
|
||||
resumable: true,
|
||||
completed: true,
|
||||
};
|
||||
}
|
||||
|
||||
// Not complete — caller should delete and retry
|
||||
return null;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Content-Disposition parser (RFC 2231 support)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
function parseContentDispositionFilename(header: string | null): string {
|
||||
if (!header) return "";
|
||||
// filename*= (RFC 2231 extended notation)
|
||||
const extMatch = /filename\*\s*=\s*(?:UTF-8|utf-8)?''(.+?)(?:;|$)/i.exec(header);
|
||||
if (extMatch) {
|
||||
try { return decodeURIComponent(extMatch[1]); } catch {}
|
||||
}
|
||||
// filename= (standard, possibly quoted)
|
||||
const stdMatch = /filename\s*=\s*"?([^";]+)"?/i.exec(header);
|
||||
if (stdMatch) return stdMatch[1].trim();
|
||||
return "";
|
||||
}
|
||||
|
||||
function parseContentRangeTotal(header: string | null): number | null {
|
||||
if (!header) return null;
|
||||
const match = /\/\s*(\d+)/.exec(header);
|
||||
if (match) {
|
||||
const total = Number(match[1]);
|
||||
return total > 0 ? total : null;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Filename utilities
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
function sanitizeFilename(name: string): string {
|
||||
return name.replace(/[<>:"/\\|?*\x00-\x1F]/g, "_").replace(/\s+/g, " ").trim();
|
||||
}
|
||||
|
||||
function looksLikeOpaqueFilename(name: string): boolean {
|
||||
return /^[a-f0-9]{20,}(\.\w+)?$/i.test(name);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// TLS skip reference counter
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
let tlsSkipRefCount = 0;
|
||||
|
||||
function acquireTlsSkip(): void {
|
||||
tlsSkipRefCount++;
|
||||
if (tlsSkipRefCount === 1) process.env.NODE_TLS_REJECT_UNAUTHORIZED = "0";
|
||||
}
|
||||
|
||||
function releaseTlsSkip(): void {
|
||||
tlsSkipRefCount--;
|
||||
if (tlsSkipRefCount <= 0) {
|
||||
tlsSkipRefCount = 0;
|
||||
delete process.env.NODE_TLS_REJECT_UNAUTHORIZED;
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Helpers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
function checkAborted(signal: AbortSignal): void {
|
||||
if (signal.aborted) throw new Error("aborted");
|
||||
}
|
||||
|
||||
function nowMs(): number {
|
||||
return Date.now();
|
||||
}
|
||||
|
||||
function sleep(ms: number): Promise<void> {
|
||||
return new Promise(resolve => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
function retryDelayWithJitter(attempt: number, baseMs: number): number {
|
||||
const base = baseMs * Math.pow(1.5, attempt - 1);
|
||||
const jitter = base * Math.random();
|
||||
return Math.floor(Math.max(base * 0.5, base - jitter));
|
||||
}
|
||||
|
||||
function errorMessage(e: unknown): string {
|
||||
if (e instanceof Error) return e.message;
|
||||
return String(e ?? "");
|
||||
}
|
||||
Reference in New Issue
Block a user