diff --git a/src/main/app-controller.ts b/src/main/app-controller.ts index 342fc95..5f3ed1e 100644 --- a/src/main/app-controller.ts +++ b/src/main/app-controller.ts @@ -1360,6 +1360,8 @@ export class AppController { abortActiveUpdateDownload(); cancelPendingAsyncSaves(); this.manager.prepareForShutdown(); + await this.manager.shutdownAndDrain?.(deadlineAt); + this.manager.persistForShutdown?.(); if (this.downloadHealthEvaluation) { await this.waitForShutdownTask(this.downloadHealthEvaluation, deadlineAt); } diff --git a/src/main/download-manager.ts b/src/main/download-manager.ts index aed2b92..786d41f 100644 --- a/src/main/download-manager.ts +++ b/src/main/download-manager.ts @@ -83,6 +83,7 @@ import { mergeKnownTotalBytes } from "./download-size"; import { DiskCapacityError, DiskReservationCoordinator, type DiskReservationLease } from "./disk-space"; import { createRendererState } from "./renderer-state"; import { PackageOutputScope } from "./package-output-scope"; +import { ExtractionCoordinator, type ExtractionArchiveMember } from "./extraction-coordinator"; import { RollingAccountStatisticsAccumulator, addStatisticsActiveIntervalInPlace, @@ -1940,14 +1941,10 @@ export class DownloadManager extends EventEmitter { private cleanupQueue: Promise = Promise.resolve(); private packageOutputScopes = new Map(); + + private extractionCoordinator: ExtractionCoordinator; - private packagePostProcessQueue: Promise = Promise.resolve(); - - private packagePostProcessActive = 0; - - private packagePostProcessWaiters: Array<{ packageId: string; runOwnerId: string | null; resolve: (acquired: boolean) => void }> = []; - - private packagePostProcessTasks = new Map>(); + private packagePostProcessTasks = new Map>(); private packagePostProcessAbortControllers = new Map(); @@ -2088,6 +2085,7 @@ export class DownloadManager extends EventEmitter { this.session = session; this.itemCount = Object.keys(this.session.items).length; this.storagePaths = storagePaths; + this.extractionCoordinator = new ExtractionCoordinator(settings.maxParallelExtract || 1); this.statisticsLedger = seedStatisticsDayProviderBytes( loadStatisticsLedger(storagePaths.statisticsFile, startedAt), settings.providerDailyUsageBytes, @@ -2424,7 +2422,8 @@ export class DownloadManager extends EventEmitter { next.totalCompletedFilesAllTime = Math.max(next.totalCompletedFilesAllTime || 0, this.settings.totalCompletedFilesAllTime || 0); const now = nowMs(); next.totalRuntimeAllTimeMs = Math.max(next.totalRuntimeAllTimeMs || 0, this.getLiveTotalRuntimeMs(now)); - this.settings = next; + this.settings = next; + this.extractionCoordinator.resize(next.maxParallelExtract || 1); this.invalidateSettingsSnapshotCache(); this.runtimePersistedTotalMs = this.settings.totalRuntimeAllTimeMs || 0; this.runtimePersistedAt = now; @@ -2670,8 +2669,7 @@ export class DownloadManager extends EventEmitter { public abortAllPostProcessing(): void { this.abortPostProcessing("external"); - this.cancelPostProcessWaiters(); - } + } public triggerIdleExtractions(): void { if (this.session.running || !this.settings.autoExtract || !this.settings.autoExtractWhenStopped) { @@ -2987,6 +2985,7 @@ export class DownloadManager extends EventEmitter { private abortPackagePostProcessing(packageId: string, reason: string, invalidateDeferred = true): Promise[] { const tasks: Promise[] = []; + void this.extractionCoordinator.cancelPackage(packageId, reason); if (invalidateDeferred) { this.bumpPackagePostProcessVersion(packageId); } @@ -3355,8 +3354,6 @@ export class DownloadManager extends EventEmitter { this.hybridExtractedPaths.clear(); this.hybridFailedArchives.clear(); this.providerFailures.clear(); - this.packagePostProcessQueue = Promise.resolve(); - this.cancelPostProcessWaiters(); this.summary = null; this.nonResumableActive = 0; this.resetSessionTotalsIfQueueEmpty(true); @@ -4664,6 +4661,86 @@ export class DownloadManager extends EventEmitter { } } + private async extractionArchiveMembers(archivePaths: readonly string[]): Promise { + const archiveRoots = new Map(); + for (const archivePath of archivePaths) { + const resolved = path.resolve(archivePath); + const key = process.platform === "win32" ? resolved.toLocaleLowerCase("en-US") : resolved; + if (!archiveRoots.has(key)) { + archiveRoots.set(key, resolved); + } + } + const directoryFiles = new Map(); + const memberPaths = new Map(); + for (const archivePath of archiveRoots.values()) { + const directory = path.dirname(archivePath); + const directoryKey = process.platform === "win32" ? directory.toLocaleLowerCase("en-US") : directory; + let files = directoryFiles.get(directoryKey); + if (!files) { + try { + files = (await fs.promises.readdir(directory, { withFileTypes: true })) + .filter((entry) => entry.isFile()) + .map((entry) => entry.name); + } catch { + files = []; + } + directoryFiles.set(directoryKey, files); + } + for (const memberPath of collectArchiveCleanupTargets(archivePath, files)) { + const resolved = path.resolve(memberPath); + const key = process.platform === "win32" ? resolved.toLocaleLowerCase("en-US") : resolved; + if (!memberPaths.has(key)) { + memberPaths.set(key, resolved); + } + } + } + return Promise.all([...memberPaths.values()].map(async (memberPath) => { + try { + return { path: memberPath, size: (await fs.promises.stat(memberPath)).size }; + } catch { + return { path: memberPath, size: null }; + } + })); + } + + private async runCoordinatedExtraction( + pkg: PackageEntry, + archivePaths: readonly string[], + signal: AbortSignal | undefined, + operation: ( + targetDir: string, + scope: PackageOutputScope, + scheduleArchive: (archivePath: string, execute: (signal: AbortSignal) => Promise) => Promise + ) => Promise + ): Promise { + const members = await this.extractionArchiveMembers(archivePaths); + const extraction = await this.extractionCoordinator.beginOperation({ + context: { + operationId: uuidv4(), + packageId: pkg.id, + generation: this.getPackageResultGeneration(pkg.id), + runOwnerId: this.getPackageResultRunOwner(pkg.id) || "" + }, + targetPath: pkg.extractDir, + members, + acquireLease: (request) => this.diskReservations.reserve({ + phase: request.phase, + ownerId: request.ownerId, + targetPath: request.targetPath, + requiredBytes: request.requiredBytes + }) + }); + const scheduleArchive = (archivePath: string, execute: (jobSignal: AbortSignal) => Promise): Promise => + this.extractionCoordinator.scheduleArchive(extraction, archivePath, (jobSignal) => execute( + signal ? AbortSignal.any([signal, jobSignal]) : jobSignal + )); + try { + return await this.runWithPackageOutputProvenance(pkg, (targetDir, scope) => operation(targetDir, scope, scheduleArchive)); + } finally { + await extraction.finalize(); + } + } + private async removeEmptyDirectoryTree(rootDir: string): Promise { if (!rootDir) { return 0; @@ -7093,7 +7170,9 @@ export class DownloadManager extends EventEmitter { this.speedBytesPerPackage.clear(); this.speedEventsHead = 0; this.abortPostProcessing("stop", stoppedRunContext?.id); - this.cancelPostProcessWaiters(stoppedRunContext?.id); + if (stoppedRunContext) { + void this.extractionCoordinator.cancelRun(stoppedRunContext.id, "stop"); + } for (const active of this.activeTasks.values()) { active.abortReason = abortReason; active.abortController.abort(abortReason); @@ -7217,9 +7296,18 @@ export class DownloadManager extends EventEmitter { this.pacedStartReservationByItem.clear(); this.nonResumableActive = 0; this.session.summaryText = ""; - if (!this.skipShutdownPersist) { - const pkgCount = Object.keys(this.session.packages).length; - const itemCount = Object.keys(this.session.items).length; + this.emitState(true); + logger.info(`Shutdown-Vorbereitung beendet: requeued=${requeuedItems}`); + } + + public async shutdownAndDrain(deadlineAt: number): Promise { + await this.extractionCoordinator.shutdownAndDrain(deadlineAt); + } + + public persistForShutdown(): void { + if (!this.skipShutdownPersist) { + const pkgCount = Object.keys(this.session.packages).length; + const itemCount = Object.keys(this.session.items).length; logger.info(`Shutdown-Save: ${pkgCount} Pakete, ${itemCount} Items`); this.foldRuntimeIntoSettings(nowMs()); if (!this.guardBlocksSessionSave()) { @@ -7227,12 +7315,10 @@ export class DownloadManager extends EventEmitter { } saveSettings(this.storagePaths, this.settings); saveStatisticsLedger(this.storagePaths.statisticsFile, this.statisticsLedger); - } else { - logger.info(`Shutdown-Save übersprungen: skipShutdownPersist=${this.skipShutdownPersist}, blockAllPersistence=${this.blockAllPersistence}`); - } - this.emitState(true); - logger.info(`Shutdown-Vorbereitung beendet: requeued=${requeuedItems}`); - } + } else { + logger.info(`Shutdown-Save übersprungen: skipShutdownPersist=${this.skipShutdownPersist}, blockAllPersistence=${this.blockAllPersistence}`); + } + } public togglePause(): boolean { if (!this.session.running) { @@ -8672,6 +8758,7 @@ export class DownloadManager extends EventEmitter { if (runContextId !== undefined && owner !== runContextId) { continue; } + void this.extractionCoordinator.cancelPackage(packageId, reason); if (!controller.signal.aborted) { controller.abort(reason); } @@ -8701,21 +8788,23 @@ export class DownloadManager extends EventEmitter { } } - for (const controller of this.packageDeferredPostProcessAbortControllers.values()) { + for (const [packageId, controller] of this.packageDeferredPostProcessAbortControllers.entries()) { const owner = this.packageDeferredRunOwnerByController.get(controller); if (runContextId !== undefined && owner !== runContextId) { continue; } + void this.extractionCoordinator.cancelPackage(packageId, reason); if (!controller.signal.aborted) { controller.abort(reason); } } - for (const hybridSet of this.packageHybridPostProcessControllers.values()) { + for (const [packageId, hybridSet] of this.packageHybridPostProcessControllers.entries()) { for (const controller of hybridSet) { const owner = this.packageHybridRunOwnerByController.get(controller); if (runContextId !== undefined && owner !== runContextId) { continue; } + void this.extractionCoordinator.cancelPackage(packageId, reason); if (!controller.signal.aborted) { controller.abort(reason); } @@ -8723,55 +8812,6 @@ export class DownloadManager extends EventEmitter { } } - private cancelPostProcessWaiters(runOwnerId?: string): void { - const retained: typeof this.packagePostProcessWaiters = []; - for (const waiter of this.packagePostProcessWaiters) { - if (runOwnerId !== undefined && waiter.runOwnerId !== runOwnerId) { - retained.push(waiter); - } else { - waiter.resolve(false); - } - } - this.packagePostProcessWaiters = retained; - } - - private async acquirePostProcessSlot(packageId: string, runOwnerId: string | null = this.getPackageResultRunOwner(packageId)): Promise { - const maxConcurrent = Math.max(1, Math.min(8, this.settings.maxParallelExtract || 1)); - if (this.packagePostProcessActive < maxConcurrent) { - this.packagePostProcessActive += 1; - return true; - } - return new Promise((resolve) => { - this.packagePostProcessWaiters.push({ packageId, runOwnerId, resolve }); - }); - } - - private releasePostProcessSlot(): void { - if (this.packagePostProcessActive <= 0) { - this.packagePostProcessActive = 0; - return; - } - const maxConcurrent = Math.max(1, Math.min(8, this.settings.maxParallelExtract || 1)); - if (this.packagePostProcessWaiters.length === 0 || this.packagePostProcessActive > maxConcurrent) { - this.packagePostProcessActive -= 1; - return; - } - const order = this.session.packageOrder; - let bestIdx = 0; - let bestOrder = order.indexOf(this.packagePostProcessWaiters[0].packageId); - if (bestOrder === -1) bestOrder = Infinity; - for (let i = 1; i < this.packagePostProcessWaiters.length; i++) { - let pos = order.indexOf(this.packagePostProcessWaiters[i].packageId); - if (pos === -1) pos = Infinity; - if (pos < bestOrder) { - bestOrder = pos; - bestIdx = i; - } - } - const [next] = this.packagePostProcessWaiters.splice(bestIdx, 1); - next.resolve(true); - } - private runPackagePostProcessing(packageId: string): Promise { this.trackPackagePostProcessResult(packageId); const existing = this.packagePostProcessTasks.get(packageId); @@ -8793,31 +8833,12 @@ export class DownloadManager extends EventEmitter { // cannot reference its own const inside its initializer). Assigned right after. const handle: { task?: Promise } = {}; const task = (async () => { - const slotWaitStart = nowMs(); - let slotAcquired = false; try { - slotAcquired = await this.acquirePostProcessSlot( - packageId, - this.packagePostProcessRunOwnerByController.get(abortController) ?? null - ); - if (!slotAcquired) { - return; - } const startedPackage = this.session.packages[packageId]; if (startedPackage) { startedPackage.postProcessStartedAt = startedPackage.postProcessStartedAt || nowMs(); startedPackage.updatedAt = nowMs(); } - const slotWaitMs = nowMs() - slotWaitStart; - if (slotWaitMs > 100) { - logger.info(`Post-Process Slot erhalten nach ${(slotWaitMs / 1000).toFixed(1)}s Wartezeit: pkg=${packageId.slice(0, 8)}`); - const pkg = this.session.packages[packageId]; - if (pkg) { - this.logPackageForPackage(pkg, "INFO", "Post-Process-Slot erhalten", { - slotWaitMs - }); - } - } let round = 0; do { round += 1; @@ -8847,12 +8868,9 @@ export class DownloadManager extends EventEmitter { this.hybridExtractRequeue.delete(packageId); } } - } while (this.hybridExtractRequeue.has(packageId)); + } while (this.hybridExtractRequeue.has(packageId)); } finally { - if (slotAcquired) { - this.releasePostProcessSlot(); - } - // Identity guard: only clear the map entries if they still point to THIS + // Identity guard: only clear the map entries if they still point to THIS // task/controller. After an abort deletes our handle a new run can install // a fresh task+controller for the same packageId; a blind delete here would // orphan that newer task (uncancellable) and allow a duplicate concurrent run. @@ -13598,7 +13616,7 @@ export class DownloadManager extends EventEmitter { return 0; } - const result = await this.runWithPackageOutputProvenance(pkg, (targetDir, scope) => extractPackageArchives({ + const result = await this.runCoordinatedExtraction(pkg, [...readyArchives], signal, (targetDir, scope, scheduleArchive) => extractPackageArchives({ packageDir: pkg.outputDir, targetDir, cleanupMode: this.settings.cleanupMode, @@ -13611,7 +13629,7 @@ export class DownloadManager extends EventEmitter { skipPostCleanup: true, packageId, hybridMode: true, - maxParallel: this.settings.maxParallelExtract || 2, + scheduleArchive, extractCpuPriority: "high", onLog: (level, message) => this.logExtractionForItems(pkg, items, "Hybrid-Extractor", level, message), onOutput: (event) => scope.add(event), @@ -14136,40 +14154,6 @@ export class DownloadManager extends EventEmitter { fullExtractItemIds.add(entry.id); } } - const archiveSizes = await Promise.all([...fullArchiveSet].map(async (archivePath) => { - try { - return (await fs.promises.stat(archivePath)).size; - } catch { - return null; - } - })); - try { - const diskLease = await this.diskReservations.reserve({ - phase: "extract", - ownerId: packageId, - targetPath: pkg.extractDir, - requiredBytes: archiveSizes.every((size) => size === null) ? null : archiveSizes.reduce((total, size) => total + Math.max(0, size || 0), 0) - }); - diskLease.release(); - } catch (error) { - if (error instanceof DiskCapacityError) { - this.diskWaitEvents = [{ ...error.event, packageId }]; - const retryAt = error.event.retryAt; - this.packageDiskRetryAfterByPackage.set(packageId, retryAt); - for (const entry of completedItems) { - entry.fullStatus = "Warte auf Festplatte"; - entry.lastError = "Zu wenig Speicherplatz"; - entry.updatedAt = nowMs(); - } - pkg.postProcessLabel = undefined; - pkg.status = "queued"; - pkg.updatedAt = nowMs(); - this.persistSoon(); - this.emitState(); - return; - } - throw error; - } const pendingAt = nowMs(); for (const entry of completedItems) { if (!fullExtractItemIds.has(entry.id) || isExtractedLabel(entry.fullStatus)) { @@ -14180,7 +14164,9 @@ export class DownloadManager extends EventEmitter { entry.updatedAt = pendingAt; } this.emitState(); - const result = await this.runWithPackageOutputProvenance(pkg, (targetDir, scope) => extractPackageArchives({ + let result; + try { + result = await this.runCoordinatedExtraction(pkg, [...fullArchiveSet], extractAbortController.signal, (targetDir, scope, scheduleArchive) => extractPackageArchives({ packageDir: pkg.outputDir, targetDir, cleanupMode: this.settings.cleanupMode, @@ -14192,7 +14178,7 @@ export class DownloadManager extends EventEmitter { packageId, onlyArchives: fullArchiveSet, skipPostCleanup: true, - maxParallel: this.settings.maxParallelExtract || 2, + scheduleArchive, extractCpuPriority: "high", onLog: (level, message) => this.logExtractionForItems(pkg, completedItems, "Extractor", level, message), onOutput: (event) => scope.add(event), @@ -14322,7 +14308,26 @@ export class DownloadManager extends EventEmitter { } emitExtractStatus(overallLabel); } - })); + })); + } catch (error) { + if (error instanceof DiskCapacityError) { + this.diskWaitEvents = [{ ...error.event, packageId }]; + const retryAt = error.event.retryAt; + this.packageDiskRetryAfterByPackage.set(packageId, retryAt); + for (const entry of completedItems) { + entry.fullStatus = "Warte auf Festplatte"; + entry.lastError = "Zu wenig Speicherplatz"; + entry.updatedAt = nowMs(); + } + pkg.postProcessLabel = undefined; + pkg.status = "queued"; + pkg.updatedAt = nowMs(); + this.persistSoon(); + this.emitState(); + return; + } + throw error; + } logger.info(`Post-Processing Entpacken Ende: pkg=${pkg.name}, extracted=${result.extracted}, failed=${result.failed}, lastError=${result.lastError || ""}`); this.logPackageForPackage(pkg, "INFO", "Post-Processing Entpacken Ende", { extracted: result.extracted, @@ -14532,7 +14537,7 @@ export class DownloadManager extends EventEmitter { }); const nestedFailureCategories = new Map(); const nestedItems = pkg.itemIds.map((itemId) => this.session.items[itemId]).filter(Boolean) as DownloadItem[]; - const nestedResult = await this.runWithPackageOutputProvenance(pkg, (targetDir, scope) => extractPackageArchives({ + const nestedResult = await this.runCoordinatedExtraction(pkg, nestedCandidates, deferredController.signal, (targetDir, scope, scheduleArchive) => extractPackageArchives({ packageDir: pkg.extractDir, targetDir, cleanupMode: this.settings.cleanupMode, @@ -14543,7 +14548,7 @@ export class DownloadManager extends EventEmitter { signal: deferredController.signal, packageId, onlyArchives: new Set(nestedCandidates.map((p) => process.platform === "win32" ? path.resolve(p).toLowerCase() : path.resolve(p))), - maxParallel: this.settings.maxParallelExtract || 2, + scheduleArchive, extractCpuPriority: this.settings.extractCpuPriority, onLog: (level, message) => this.logPackageForPackage(pkg, level, `Nested-Extractor: ${message}`), onOutput: (event) => scope.add(event), diff --git a/src/main/extraction-coordinator.ts b/src/main/extraction-coordinator.ts index 588212f..aafee2f 100644 --- a/src/main/extraction-coordinator.ts +++ b/src/main/extraction-coordinator.ts @@ -162,7 +162,7 @@ export class ExtractionCoordinator { const members = deduplicateMembers(options.members || []); state.lease = await options.acquireLease({ phase: "extract", - ownerId: context.operationId, + ownerId: context.packageId, targetPath: String(options.targetPath || ""), requiredBytes: reservationBytes(members), memberPaths: Object.freeze(members.map((member) => member.path)) @@ -233,10 +233,7 @@ export class ExtractionCoordinator { this.resolveDrainIfIdle(state); } await this.waitUntilDeadline(Promise.all(states.map((state) => state.drain.promise)), deadlineAt); - const finalizers = states.map((state) => state.finalizePromise).filter((value): value is Promise => Boolean(value)); - if (finalizers.length > 0) { - await this.waitUntilDeadline(Promise.allSettled(finalizers), deadlineAt); - } + await this.waitForFinalization(states, deadlineAt); for (const state of states) { this.releaseLease(state); } @@ -402,4 +399,21 @@ export class ExtractionCoordinator { clearTimeout(timeout); } } + + private async waitForFinalization(states: readonly OperationState[], deadlineAt: number): Promise { + while (Date.now() < deadlineAt) { + const pending = states.filter((state) => this.operations.get(state.context.operationId) === state); + if (pending.length === 0) { + return; + } + const finalizers = pending + .map((state) => state.finalizePromise) + .filter((value): value is Promise => Boolean(value)); + const waitMs = Math.max(1, Math.min(10, deadlineAt - Date.now())); + await Promise.race([ + ...finalizers.map((finalizer) => finalizer.then(() => undefined, () => undefined)), + new Promise((resolve) => setTimeout(resolve, waitMs)) + ]); + } + } } diff --git a/src/main/extractor.ts b/src/main/extractor.ts index 3b601ea..09c49c2 100644 --- a/src/main/extractor.ts +++ b/src/main/extractor.ts @@ -60,11 +60,11 @@ export interface ExtractOptions { skipPostCleanup?: boolean; packageId?: string; hybridMode?: boolean; - maxParallel?: number; extractCpuPriority?: string; onArchiveFailure?: (failure: ExtractArchiveFailureInfo) => void; onLog?: (level: "INFO" | "WARN" | "ERROR", message: string) => void; onOutput?: (event: ExtractOutputEvent) => void; + scheduleArchive?: (archivePath: string, execute: (signal: AbortSignal) => Promise) => Promise; } export interface ExtractProgressUpdate { @@ -215,6 +215,9 @@ interface DaemonRequest { passwordCount: number; onOutput?: (event: ExtractOutputEvent) => void; targetDir: string; + aborted: boolean; + timedOut: boolean; + terminationStarted: boolean; } const activeSubstDrives = new Set(); @@ -1489,26 +1492,18 @@ function runExtractCommand( resolve(result); }; - if (timeoutMs && timeoutMs > 0) { - timeoutId = setTimeout(() => { - timedOutByWatchdog = true; - killProcessTree(child); - finish({ - ok: false, - missingCommand: false, - aborted: false, - timedOut: true, - errorText: `Entpacken Timeout nach ${Math.ceil(timeoutMs / 1000)}s` - }); - }, timeoutMs); - } + if (timeoutMs && timeoutMs > 0) { + timeoutId = setTimeout(() => { + timedOutByWatchdog = true; + killProcessTree(child); + }, timeoutMs); + } const onAbort = signal - ? (): void => { - abortedBySignal = true; - killProcessTree(child); - finish({ ok: false, missingCommand: false, aborted: true, timedOut: false, errorText: "aborted:extract" }); - } + ? (): void => { + abortedBySignal = true; + killProcessTree(child); + } : null; if (signal && onAbort) { signal.addEventListener("abort", onAbort, { once: true }); @@ -1525,8 +1520,11 @@ function runExtractCommand( onChunk?.(text); }); - child.on("error", (error) => { - const text = cleanErrorText(String(error)); + child.on("error", (error) => { + if (abortedBySignal || timedOutByWatchdog) { + return; + } + const text = cleanErrorText(String(error)); finish({ ok: false, missingCommand: text.toLowerCase().includes("enoent"), @@ -1926,8 +1924,11 @@ function startDaemon(layout: JvmExtractorLayout): boolean { } }); - child.on("error", () => { - if (daemonCurrentRequest) { + child.on("error", () => { + if (daemonCurrentRequest?.terminationStarted) { + return; + } + if (daemonCurrentRequest) { finishDaemonRequest({ ok: false, missingCommand: true, missingRuntime: true, aborted: false, timedOut: false, errorText: "Daemon process error", @@ -1937,16 +1938,31 @@ function startDaemon(layout: JvmExtractorLayout): boolean { shutdownDaemon(); }); - child.on("close", () => { - if (daemonCurrentRequest) { - const req = daemonCurrentRequest; - finishDaemonRequest({ - ok: false, missingCommand: false, missingRuntime: false, - aborted: false, timedOut: false, - errorText: cleanErrorText(req.parseState.reportedError || daemonOutput) || "Daemon process exited unexpectedly", - usedPassword: req.parseState.usedPassword, backend: req.parseState.backend - }); - } + child.on("close", () => { + if (daemonCurrentRequest) { + const req = daemonCurrentRequest; + if (req.aborted) { + finishDaemonRequest({ + ok: false, missingCommand: false, missingRuntime: false, + aborted: true, timedOut: false, errorText: "aborted:extract", + usedPassword: req.parseState.usedPassword, backend: req.parseState.backend + }); + } else if (req.timedOut) { + finishDaemonRequest({ + ok: false, missingCommand: false, missingRuntime: false, + aborted: false, timedOut: true, + errorText: `Entpacken Timeout nach ${Math.ceil((req.timeoutMs || 0) / 1000)}s`, + usedPassword: req.parseState.usedPassword, backend: req.parseState.backend + }); + } else { + finishDaemonRequest({ + ok: false, missingCommand: false, missingRuntime: false, + aborted: false, timedOut: false, + errorText: cleanErrorText(req.parseState.reportedError || daemonOutput) || "Daemon process exited unexpectedly", + usedPassword: req.parseState.usedPassword, backend: req.parseState.backend + }); + } + } fs.rm(jvmTmpDir, { recursive: true, force: true }, () => {}); daemonProcess = null; daemonReady = false; @@ -2010,37 +2026,39 @@ function sendDaemonRequest( startedAt: Date.now(), passwordCount: passwordCandidates.length, onOutput, - targetDir + targetDir, + aborted: false, + timedOut: false, + terminationStarted: false }; logger.info(`JVM Daemon Request Start: archive=${archiveName}, pwCandidates=${passwordCandidates.length}, timeoutMs=${timeoutMs || 0}, conflict=${mode}`); if (timeoutMs && timeoutMs > 0) { - daemonTimeoutId = setTimeout(() => { - const req = daemonCurrentRequest; - if (req) { - finishDaemonRequest({ - ok: false, missingCommand: false, missingRuntime: false, - aborted: false, timedOut: true, - errorText: `Entpacken Timeout nach ${Math.ceil(timeoutMs / 1000)}s`, - usedPassword: parseState.usedPassword, backend: parseState.backend - }); - } - shutdownDaemon(); - }, timeoutMs); + daemonTimeoutId = setTimeout(() => { + const req = daemonCurrentRequest; + if (req && !req.terminationStarted) { + req.timedOut = true; + req.terminationStarted = true; + try { daemonProcess?.stdin?.end(); } catch { } + if (daemonProcess) { + killProcessTree(daemonProcess); + } + } + }, timeoutMs); } if (signal) { - daemonAbortHandler = () => { - const req = daemonCurrentRequest; - if (req) { - finishDaemonRequest({ - ok: false, missingCommand: false, missingRuntime: false, - aborted: true, timedOut: false, errorText: "aborted:extract", - usedPassword: parseState.usedPassword, backend: parseState.backend - }); - } - shutdownDaemon(); - }; + daemonAbortHandler = () => { + const req = daemonCurrentRequest; + if (req && !req.terminationStarted) { + req.aborted = true; + req.terminationStarted = true; + try { daemonProcess?.stdin?.end(); } catch { } + if (daemonProcess) { + killProcessTree(daemonProcess); + } + } + }; signal.addEventListener("abort", daemonAbortHandler, { once: true }); } @@ -2207,29 +2225,18 @@ async function runJvmExtractCommand( resolve(finalResult); }; - if (timeoutMs && timeoutMs > 0) { - timeoutId = setTimeout(() => { - timedOutByWatchdog = true; - killProcessTree(child); - finish({ - ok: false, missingCommand: false, missingRuntime: false, - aborted: false, timedOut: true, - errorText: `Entpacken Timeout nach ${Math.ceil(timeoutMs / 1000)}s`, - usedPassword: parseState.usedPassword, backend: parseState.backend - }); - }, timeoutMs); - } + if (timeoutMs && timeoutMs > 0) { + timeoutId = setTimeout(() => { + timedOutByWatchdog = true; + killProcessTree(child); + }, timeoutMs); + } onAbort = signal - ? (): void => { - abortedBySignal = true; - killProcessTree(child); - finish({ - ok: false, missingCommand: false, missingRuntime: false, - aborted: true, timedOut: false, errorText: "aborted:extract", - usedPassword: parseState.usedPassword, backend: parseState.backend - }); - } + ? (): void => { + abortedBySignal = true; + killProcessTree(child); + } : null; if (signal && onAbort) { @@ -2243,8 +2250,11 @@ async function runJvmExtractCommand( flushLines(String(chunk || ""), true); }); - child.on("error", (error) => { - const text = cleanErrorText(String(error)); + child.on("error", (error) => { + if (abortedBySignal || timedOutByWatchdog) { + return; + } + const text = cleanErrorText(String(error)); finish({ ok: false, missingCommand: text.toLowerCase().includes("enoent"), missingRuntime: true, aborted: false, timedOut: false, @@ -2555,21 +2565,14 @@ export function parseNativeExtractOutput( } function failDaemonOutputCallback(req: DaemonRequest): void { - if (daemonCurrentRequest !== req || !req.parseState.outputError) { + if (daemonCurrentRequest !== req || !req.parseState.outputError || req.terminationStarted) { return; } - const message = cleanErrorText(req.parseState.outputError.message || String(req.parseState.outputError)); - finishDaemonRequest({ - ok: false, - missingCommand: false, - missingRuntime: false, - aborted: false, - timedOut: false, - errorText: message, - usedPassword: req.parseState.usedPassword, - backend: req.parseState.backend - }); - shutdownDaemon(); + req.terminationStarted = true; + try { daemonProcess?.stdin?.end(); } catch { } + if (daemonProcess) { + killProcessTree(daemonProcess); + } } function createNativeOutputCollector( @@ -3793,12 +3796,24 @@ export async function extractPackageArchives(options: ExtractOptions): Promise => { - if (options.signal?.aborted) { + const fallbackSignal = options.signal || new AbortController().signal; + let localScheduleQueue = Promise.resolve(); + const scheduleArchive = options.scheduleArchive + ? (archivePath: string, execute: (signal: AbortSignal) => Promise): Promise => options.scheduleArchive!( + archivePath, + (jobSignal) => execute(options.signal ? AbortSignal.any([options.signal, jobSignal]) : jobSignal) + ) + : (_archivePath: string, execute: (signal: AbortSignal) => Promise): Promise => { + const scheduled = localScheduleQueue.then(() => execute(fallbackSignal)); + localScheduleQueue = scheduled.then(() => undefined, () => undefined); + return scheduled; + }; + const archiveWorkerCount = Math.max(1, pendingCandidates.length); + let noExtractorEncountered = false; + let lastArchiveFinishedAt: number | null = null; + + const extractSingleArchive = async (archivePath: string, signal: AbortSignal): Promise => { + if (signal.aborted) { throw new Error("aborted:extract"); } if (noExtractorEncountered) { @@ -3876,18 +3891,18 @@ export async function extractPackageArchives(options: ExtractOptions): Promise { reportArchiveProgress(value); - }, options.signal, hybrid, onPwAttempt, false, undefined, options.onLog, emitOutput); + }, signal, hybrid, onPwAttempt, false, undefined, options.onLog, emitOutput); rememberLearnedPassword(usedPassword); } catch (error) { if (isNoExtractorError(String(error))) { - await extractZipArchive(archivePath, options.targetDir, options.conflictMode, options.signal, emitOutput, validateOutputTarget); + await extractZipArchive(archivePath, options.targetDir, options.conflictMode, signal, emitOutput, validateOutputTarget); } else { throw error; } } } else { try { - await extractZipArchive(archivePath, options.targetDir, options.conflictMode, options.signal, emitOutput, validateOutputTarget); + await extractZipArchive(archivePath, options.targetDir, options.conflictMode, signal, emitOutput, validateOutputTarget); archivePercent = 100; } catch (error) { if (!shouldFallbackToExternalZip(error)) { @@ -3896,7 +3911,7 @@ export async function extractPackageArchives(options: ExtractOptions): Promise { reportArchiveProgress(value); - }, options.signal, hybrid, onPwAttempt, false, undefined, options.onLog, emitOutput); + }, signal, hybrid, onPwAttempt, false, undefined, options.onLog, emitOutput); rememberLearnedPassword(usedPassword); } catch (externalError) { throw selectZipFallbackError(error, externalError); @@ -3907,7 +3922,7 @@ export async function extractPackageArchives(options: ExtractOptions): Promise { reportArchiveProgress(value); - }, options.signal, hybrid, onPwAttempt, packageNeedsFlatMode, flatResult, options.onLog, emitOutput); + }, signal, hybrid, onPwAttempt, packageNeedsFlatMode, flatResult, options.onLog, emitOutput); rememberLearnedPassword(usedPassword); if (flatResult.needed) packageNeedsFlatMode = true; } @@ -3974,10 +3989,10 @@ export async function extractPackageArchives(options: ExtractOptions): Promise extractSingleArchive(archivePath, signal)); } if (noExtractorEncountered) { const remaining = candidates.length - (extracted + failed); @@ -3993,7 +4008,7 @@ export async function extractPackageArchives(options: ExtractOptions): Promise extractSingleArchive(first, signal)); } catch (err) { const errText = String(err); if (/aborted:extract/i.test(errText)) throw err; @@ -4015,7 +4030,7 @@ export async function extractPackageArchives(options: ExtractOptions): Promise extractSingleArchive(queue[idx], signal)); } catch (error) { const errText = String(error); if (errText.includes("noextractor:skipped")) { @@ -4029,7 +4044,7 @@ export async function extractPackageArchives(options: ExtractOptions): Promise worker())); @@ -4051,7 +4066,7 @@ export async function extractPackageArchives(options: ExtractOptions): Promise extractSingleArchive(archivePath, signal)); retryRecovered += 1; } catch (retryError) { const errText = String(retryError); @@ -4073,7 +4088,7 @@ export async function extractPackageArchives(options: ExtractOptions): Promise extractSingleArchive(archivePath, signal)); retryRecovered += 1; } catch (retryError) { const errText = String(retryError); @@ -4142,24 +4157,26 @@ export async function extractPackageArchives(options: ExtractOptions): Promise { emitProgress(extracted + failed, `nested: ${nestedName}`, "extracting", nestedPercent, Date.now() - nestedStartedAt, undefined, undefined, nestedArchive); }, 1100); - const hybrid = Boolean(options.hybridMode); - logger.info(`Nested-Entpacke: ${nestedName} -> ${options.targetDir}${hybrid ? " (hybrid)" : ""}`); - try { - const ext = path.extname(nestedArchive).toLowerCase(); - if (ext === ".zip" && !(await shouldPreferExternalZip(nestedArchive))) { - try { - await extractZipArchive(nestedArchive, options.targetDir, options.conflictMode, options.signal, emitOutput, validateOutputTarget); - nestedPercent = 100; - } catch (zipErr) { - if (!shouldFallbackToExternalZip(zipErr)) throw zipErr; - const usedPw = await runExternalExtract(nestedArchive, options.targetDir, options.conflictMode, passwordCandidates, (v) => { nestedPercent = Math.max(nestedPercent, v); }, options.signal, hybrid, undefined, false, undefined, options.onLog, emitOutput); - rememberLearnedPassword(usedPw); - } - } else { - const usedPw = await runExternalExtract(nestedArchive, options.targetDir, options.conflictMode, passwordCandidates, (v) => { nestedPercent = Math.max(nestedPercent, v); }, options.signal, hybrid, undefined, false, undefined, options.onLog, emitOutput); - rememberLearnedPassword(usedPw); - } - extracted += 1; + const hybrid = Boolean(options.hybridMode); + logger.info(`Nested-Entpacke: ${nestedName} -> ${options.targetDir}${hybrid ? " (hybrid)" : ""}`); + try { + await scheduleArchive(nestedArchive, async (signal) => { + const ext = path.extname(nestedArchive).toLowerCase(); + if (ext === ".zip" && !(await shouldPreferExternalZip(nestedArchive))) { + try { + await extractZipArchive(nestedArchive, options.targetDir, options.conflictMode, signal, emitOutput, validateOutputTarget); + nestedPercent = 100; + } catch (zipErr) { + if (!shouldFallbackToExternalZip(zipErr)) throw zipErr; + const usedPw = await runExternalExtract(nestedArchive, options.targetDir, options.conflictMode, passwordCandidates, (v) => { nestedPercent = Math.max(nestedPercent, v); }, signal, hybrid, undefined, false, undefined, options.onLog, emitOutput); + rememberLearnedPassword(usedPw); + } + } else { + const usedPw = await runExternalExtract(nestedArchive, options.targetDir, options.conflictMode, passwordCandidates, (v) => { nestedPercent = Math.max(nestedPercent, v); }, signal, hybrid, undefined, false, undefined, options.onLog, emitOutput); + rememberLearnedPassword(usedPw); + } + }); + extracted += 1; nestedExtracted += 1; extractedArchives.add(nestedArchive); const nestedResume = await buildResumeArchive(nestedArchive, options.packageDir, options.targetDir, outputScope.records()); diff --git a/src/main/main.ts b/src/main/main.ts index bbb16f0..93df68f 100644 --- a/src/main/main.ts +++ b/src/main/main.ts @@ -1117,11 +1117,14 @@ app.on("before-quit", createBeforeQuitHandler({ powerMonitor.removeListener("resume", handlePowerResume); stopClipboardWatcher(); destroyTray(); - shutdownDaemon(); }, shutdown: async () => { - if (controller) { - await controller.shutdown(); + try { + if (controller) { + await controller.shutdown(); + } + } finally { + shutdownDaemon(); } }, continueQuit: () => app.quit(), diff --git a/tests/download-manager.test.ts b/tests/download-manager.test.ts index 306aa92..acc0f19 100644 --- a/tests/download-manager.test.ts +++ b/tests/download-manager.test.ts @@ -9,6 +9,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { DownloadManager, buildAutoRenameBaseNameFromFoldersWithOptions, extractArchiveNameFromExtractorLogMessage, getAuthoritativeRealDebridTotal, getDiskWriteWaitReason, resolveArchiveItemsFromList, resolveUnrestrictTimeoutBudgetMs, runWithLimitedConcurrency } from "../src/main/download-manager"; import { planDownloadCompletion, validateDownloadedFileCompletion } from "../src/main/download-completion"; import { DiskReservationCoordinator } from "../src/main/disk-space"; +import { ExtractionCoordinator } from "../src/main/extraction-coordinator"; import { defaultSettings } from "../src/main/constants"; import { parseDebridLinkApiKeys } from "../src/shared/debrid-link-keys"; import { getProviderUsageDayKey } from "../src/shared/provider-daily-limits"; @@ -2310,42 +2311,6 @@ describe("download manager", () => { expect((manager as any).shouldCollapseQuickPostProcessRequeue(packageId)).toBe(false); }); - it("honors maxParallelExtract for concurrent post-process slots", async () => { - const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-postprocess-slots-")); - tempDirs.push(root); - - const manager = new DownloadManager( - { - ...defaultSettings(), - token: "rd-token", - maxParallelExtract: 4 - }, - emptySession(), - createStoragePaths(path.join(root, "state")) - ); - - await (manager as any).acquirePostProcessSlot("pkg-1"); - await (manager as any).acquirePostProcessSlot("pkg-2"); - await (manager as any).acquirePostProcessSlot("pkg-3"); - await (manager as any).acquirePostProcessSlot("pkg-4"); - - expect((manager as any).packagePostProcessActive).toBe(4); - - let fifthResolved = false; - const fifth = (manager as any).acquirePostProcessSlot("pkg-5").then(() => { - fifthResolved = true; - }); - - await new Promise((resolve) => setTimeout(resolve, 30)); - expect(fifthResolved).toBe(false); - - (manager as any).releasePostProcessSlot(); - await fifth; - - expect(fifthResolved).toBe(true); - expect((manager as any).packagePostProcessActive).toBe(4); - }); - it("extractNow only re-arms completed items that are not already extracted", () => { const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-extract-now-")); tempDirs.push(root); @@ -15543,6 +15508,71 @@ describe("package lifecycle telemetry boundaries", () => { expect(pkg.outputRecords).toEqual([expect.objectContaining({ state: "partial", outputPath: path.join(extractDir, "partial.mkv") })]); }); + it("holds one deduplicated multipart lease through child close and output-scope finalization", async () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-coordinated-lease-")); + tempDirs.push(root); + const outputDir = path.join(root, "downloads"); + const extractDir = path.join(root, "extract"); + fs.mkdirSync(outputDir, { recursive: true }); + const firstPart = path.join(outputDir, "release.part1.rar"); + const secondPart = path.join(outputDir, "release.part2.rar"); + fs.writeFileSync(firstPart, Buffer.alloc(100)); + fs.writeFileSync(secondPart, Buffer.alloc(200)); + const session = emptySession(); + const pkg: PackageEntry = { + id: "coordinated-lease", + name: "coordinated-lease", + outputDir, + extractDir, + status: "completed", + itemIds: [], + cancelled: false, + enabled: true, + resultGeneration: 3, + createdAt: 1_000, + updatedAt: 1_000 + }; + session.packages[pkg.id] = pkg; + session.packageOrder = [pkg.id]; + const manager = new DownloadManager(defaultSettings(), session, createStoragePaths(path.join(root, "state"))); + const state = manager as any; + state.extractionCoordinator = new ExtractionCoordinator(1); + state.diskReservations = new DiskReservationCoordinator({ + safetyBytes: 0, + statVolume: async () => ({ path: extractDir, volumeKey: "extract-volume", freeBytes: 10_000, totalBytes: 20_000 }) + }); + const originalSync = state.syncPackageOutputScope.bind(state); + const reservationsDuringFinalization: number[] = []; + state.syncPackageOutputScope = (entry: PackageEntry, scope: unknown) => { + reservationsDuringFinalization.push(state.diskReservations.getReservedBytesByVolume().get("extract-volume") || 0); + return originalSync(entry, scope); + }; + let closeChild = () => {}; + const childClosed = new Promise((resolve) => { + closeChild = resolve; + }); + let completed = false; + + const extraction = state.runCoordinatedExtraction( + pkg, + [firstPart, firstPart.toUpperCase(), secondPart], + undefined, + async (_targetDir: string, _scope: unknown, scheduleArchive: (archivePath: string, execute: (signal: AbortSignal) => Promise) => Promise) => { + await scheduleArchive(firstPart, async () => childClosed); + } + ).then(() => { + completed = true; + }); + + await vi.waitFor(() => expect(state.diskReservations.getReservedBytesByVolume().get("extract-volume")).toBe(300)); + expect(completed).toBe(false); + closeChild(); + await extraction; + + expect(reservationsDuringFinalization).toContain(300); + expect(state.diskReservations.getReservedBytesByVolume().get("extract-volume")).toBe(0); + }); + it("uses normalized nested item paths for archive identity and leaves empty item provenance at zero", () => { const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-archive-identity-")); tempDirs.push(root); @@ -15611,73 +15641,6 @@ describe("package lifecycle telemetry boundaries", () => { expect(operations.map((operation) => operation.partCount)).toEqual([1, 1, 0]); }); - it("keeps foreign post-process waiters reserved when stopping another run", async () => { - const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-run-owned-slots-")); - tempDirs.push(root); - const session = emptySession(); - const createPackage = (id: string): PackageEntry => ({ - id, - name: id, - outputDir: path.join(root, "downloads", id), - extractDir: path.join(root, "extract", id), - status: "completed", - itemIds: [], - cancelled: false, - enabled: true, - createdAt: 1_000, - updatedAt: 1_000 - }); - const packageA = createPackage("run-a-package"); - const packageB = createPackage("run-b-package"); - session.packages[packageA.id] = packageA; - session.packages[packageB.id] = packageB; - session.packageOrder = [packageA.id, packageB.id]; - const manager = new DownloadManager( - { ...defaultSettings(), maxParallelExtract: 1 }, - session, - createStoragePaths(path.join(root, "state")) - ); - const state = manager as any; - const runA = state.createRunContext([packageA.id], 1_000, false); - const runB = state.beginActiveRunContext([packageB.id], 2_000); - session.running = true; - session.runStartedAt = 2_000; - state.runPackageIds = new Set([packageB.id]); - state.runItemIds = new Set(["run-b-item"]); - let concurrent = 1; - let peak = concurrent; - let foreignResolved = false; - - await state.acquirePostProcessSlot("active-a", runA.id); - const foreignWaiter = state.acquirePostProcessSlot("waiting-a", runA.id).then((acquired: boolean | undefined) => { - foreignResolved = true; - if (acquired !== false) { - concurrent += 1; - peak = Math.max(peak, concurrent); - } - return acquired; - }); - const stoppedWaiter = state.acquirePostProcessSlot("waiting-b", runB.id); - manager.stop(); - const stoppedResult = await stoppedWaiter; - await Promise.resolve(); - - expect(stoppedResult).toBe(false); - expect(foreignResolved).toBe(false); - expect(state.packagePostProcessActive).toBe(1); - - concurrent -= 1; - state.releasePostProcessSlot(); - const foreignResult = await foreignWaiter; - expect(foreignResult).toBe(true); - expect(state.packagePostProcessActive).toBe(1); - expect(peak).toBe(1); - - concurrent -= 1; - state.releasePostProcessSlot(); - expect(state.packagePostProcessActive).toBe(0); - }); - it("records queued, slot start and terminal timestamps around real post-processing", async () => { const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-lifecycle-boundaries-")); tempDirs.push(root); diff --git a/tests/extraction-coordinator.test.ts b/tests/extraction-coordinator.test.ts index d135acb..f57dc51 100644 --- a/tests/extraction-coordinator.test.ts +++ b/tests/extraction-coordinator.test.ts @@ -331,4 +331,35 @@ describe("ExtractionCoordinator", () => { await Promise.all([active, finalized, shutdown]); expect(events).toEqual(["waiter-cancel", "active-abort", "child-close", "scope-finalize", "lease-release"]); }); + + it("waits for scope finalization registered immediately after child drain", async () => { + const coordinator = new ExtractionCoordinator(1); + const heldLease = lease(); + const operation = await coordinator.beginOperation({ + context: context("late-finalize", "package-a", "run"), + targetPath: "C:\\target", + members: [{ path: "C:\\archives\\one.rar", size: 100 }], + acquireLease: async () => heldLease + }); + const childClose = deferred(); + const scopeClose = deferred(); + const active = coordinator.scheduleArchive(operation, "active", async () => childClose.promise); + let shutdownSettled = false; + const shutdown = coordinator.shutdownAndDrain(Date.now() + 1000).then(() => { + shutdownSettled = true; + }); + + childClose.resolve(); + await active; + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(shutdownSettled).toBe(false); + const finalization = operation.finalize(async () => scopeClose.promise); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(shutdownSettled).toBe(false); + expect(heldLease.release).not.toHaveBeenCalled(); + + scopeClose.resolve(); + await Promise.all([finalization, shutdown]); + expect(heldLease.release).toHaveBeenCalledTimes(1); + }); }); diff --git a/tests/extractor-child-close.test.ts b/tests/extractor-child-close.test.ts new file mode 100644 index 0000000..f5a3834 --- /dev/null +++ b/tests/extractor-child-close.test.ts @@ -0,0 +1,170 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +const childProcesses = vi.hoisted(() => { + class FakeEmitter { + private readonly listeners = new Map void>>(); + + public on(name: string, listener: (...args: unknown[]) => void): this { + const listeners = this.listeners.get(name) || []; + listeners.push(listener); + this.listeners.set(name, listeners); + return this; + } + + public emit(name: string, ...args: unknown[]): boolean { + for (const listener of this.listeners.get(name) || []) { + listener(...args); + } + return true; + } + } + + class FakeChild extends FakeEmitter { + public readonly stdout = new FakeEmitter(); + public readonly stderr = new FakeEmitter(); + public readonly stdin = { end: vi.fn(), write: vi.fn() }; + public readonly pid: number; + public readonly kill = vi.fn(); + + public constructor(pid: number) { + super(); + this.pid = pid; + } + } + + return { + nextPid: 10_000, + activeExtraction: null as FakeChild | null, + spawn: vi.fn((_command: string, args: string[]) => { + const child = new FakeChild(childProcesses.nextPid++); + if (args[0] === "?") { + queueMicrotask(() => child.emit("close", 0)); + } else if (args[0] === "l") { + queueMicrotask(() => { + child.stdout.emit("data", "----------\nPath = episode.mkv\nFolder = -\n"); + child.emit("close", 0); + }); + } else if (args[0] === "/PID") { + queueMicrotask(() => child.emit("close", 0)); + } else { + childProcesses.activeExtraction = child; + } + return child; + }), + spawnSync: vi.fn(() => ({ status: 1, stdout: Buffer.alloc(0), stderr: Buffer.alloc(0) })) + }; +}); + +vi.mock("node:child_process", () => ({ + spawn: childProcesses.spawn, + spawnSync: childProcesses.spawnSync +})); + +import { extractPackageArchives } from "../src/main/extractor"; + +const tempDirs: string[] = []; +const originalBackend = process.env.RD_EXTRACT_BACKEND; +const originalSevenZip = process.env.RD_7Z_BIN; + +afterEach(() => { + vi.useRealTimers(); + for (const directory of tempDirs.splice(0)) { + fs.rmSync(directory, { recursive: true, force: true }); + } + childProcesses.activeExtraction = null; + childProcesses.spawn.mockClear(); + if (originalBackend === undefined) { + delete process.env.RD_EXTRACT_BACKEND; + } else { + process.env.RD_EXTRACT_BACKEND = originalBackend; + } + if (originalSevenZip === undefined) { + delete process.env.RD_7Z_BIN; + } else { + process.env.RD_7Z_BIN = originalSevenZip; + } +}); + +describe("extractor child close lifecycle", () => { + it("keeps an aborted native archive job active until the original child closes", async () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-native-child-close-")); + tempDirs.push(root); + const packageDir = path.join(root, "package"); + const targetDir = path.join(root, "target"); + const sevenZipPath = path.join(root, "7z.exe"); + fs.mkdirSync(packageDir, { recursive: true }); + fs.writeFileSync(sevenZipPath, "fake"); + fs.writeFileSync(path.join(packageDir, "release.7z"), Buffer.from("377abcaf271c", "hex")); + process.env.RD_EXTRACT_BACKEND = "legacy"; + process.env.RD_7Z_BIN = sevenZipPath; + const controller = new AbortController(); + let settled = false; + let failure: unknown; + + const extraction = extractPackageArchives({ + packageDir, + targetDir, + cleanupMode: "none", + conflictMode: "overwrite", + removeLinks: false, + removeSamples: false, + passwordList: "", + signal: controller.signal + }).catch((error) => { + failure = error; + }).finally(() => { + settled = true; + }); + + await vi.waitFor(() => expect(childProcesses.activeExtraction).not.toBeNull()); + controller.abort("abort-test"); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(settled).toBe(false); + + childProcesses.activeExtraction?.emit("close", 1); + await extraction; + expect(String(failure)).toContain("aborted:extract"); + }); + + it("keeps a timed-out native archive job active until the original child closes", async () => { + vi.useFakeTimers(); + const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-native-timeout-close-")); + tempDirs.push(root); + const packageDir = path.join(root, "package"); + const targetDir = path.join(root, "target"); + const sevenZipPath = path.join(root, "7z.exe"); + fs.mkdirSync(packageDir, { recursive: true }); + fs.writeFileSync(sevenZipPath, "fake"); + fs.writeFileSync(path.join(packageDir, "release.7z"), Buffer.from("377abcaf271c", "hex")); + process.env.RD_EXTRACT_BACKEND = "legacy"; + process.env.RD_7Z_BIN = sevenZipPath; + let settled = false; + const results: Awaited>[] = []; + + const extraction = extractPackageArchives({ + packageDir, + targetDir, + cleanupMode: "none", + conflictMode: "overwrite", + removeLinks: false, + removeSamples: false, + passwordList: "" + }).then((value) => { + results.push(value); + }).finally(() => { + settled = true; + }); + + await vi.waitFor(() => expect(childProcesses.activeExtraction).not.toBeNull()); + await vi.advanceTimersByTimeAsync(6 * 60 * 1000); + expect(settled).toBe(false); + + childProcesses.activeExtraction?.emit("close", 1); + await extraction; + expect(results[0]).toEqual(expect.objectContaining({ failed: 1 })); + expect(results[0]?.lastError).toContain("Timeout"); + }); +}); diff --git a/tests/extractor.test.ts b/tests/extractor.test.ts index e34b3ce..1460760 100644 --- a/tests/extractor.test.ts +++ b/tests/extractor.test.ts @@ -1214,7 +1214,6 @@ describe("extractor", () => { conflictMode: "overwrite", removeLinks: false, removeSamples: false, - maxParallel: 2, passwordList: "pw1|pw2|pw3", onProgress: (update) => { if (update.phase !== "extracting" || !update.archiveName) return; @@ -1249,7 +1248,6 @@ describe("extractor", () => { conflictMode: "overwrite", removeLinks: false, removeSamples: false, - maxParallel: 4 }); expect(result.extracted).toBe(2); @@ -1274,7 +1272,6 @@ describe("extractor", () => { conflictMode: "overwrite", removeLinks: false, removeSamples: false, - maxParallel: 4, passwordList: "pw1|pw2|pw3" }); @@ -1430,6 +1427,40 @@ describe("extractor", () => { expect(fs.existsSync(path.join(targetDir, "foreign.txt"))).toBe(false); }); + it("delegates top-level and nested archive jobs through one scheduler", async () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-global-extract-scheduler-")); + tempDirs.push(root); + const packageDir = path.join(root, "pkg"); + const targetDir = path.join(root, "out"); + fs.mkdirSync(packageDir, { recursive: true }); + const first = new AdmZip(); + first.addFile("first.txt", Buffer.from("first")); + first.writeZip(path.join(packageDir, "first.zip")); + const nested = new AdmZip(); + nested.addFile("nested.txt", Buffer.from("nested")); + const second = new AdmZip(); + second.addFile("owned.zip", nested.toBuffer()); + second.writeZip(path.join(packageDir, "second.zip")); + const scheduled: string[] = []; + + const result = await extractPackageArchives({ + packageDir, + targetDir, + cleanupMode: "none", + conflictMode: "overwrite", + removeLinks: false, + removeSamples: false, + scheduleArchive: async (archivePath, execute) => { + scheduled.push(path.basename(archivePath)); + return execute(new AbortController().signal); + } + }); + + expect(result.failed).toBe(0); + expect(scheduled).toEqual(["first.zip", "second.zip", "owned.zip"]); + expect(fs.readFileSync(path.join(targetDir, "nested.txt"), "utf8")).toBe("nested"); + }); + it("resumes same-basename archives by relative path and invalidates changed multipart fingerprints", async () => { const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-resume-v2-")); tempDirs.push(root); diff --git a/tests/main-shutdown-lifecycle.test.ts b/tests/main-shutdown-lifecycle.test.ts index c7d9abb..e029138 100644 --- a/tests/main-shutdown-lifecycle.test.ts +++ b/tests/main-shutdown-lifecycle.test.ts @@ -69,9 +69,57 @@ afterEach(() => { }); describe("main shutdown lifecycle", () => { + it("drains extraction before session persistence and runtime disposal", async () => { + const drain = deferred(); + const events: string[] = []; + const controller = Object.create(AppController.prototype) as any; + controller.downloadHealthTimer = null; + controller.downloadHealthEvaluation = null; + controller.downloadHealthMonitor = null; + controller.runtimeStatsTimer = null; + controller.notificationOutbox = { drainForShutdown: vi.fn(async () => undefined) }; + controller.manager = { + suspendDownloadHealthMonitoring: vi.fn(), + prepareForShutdown: vi.fn(() => events.push("queue-close")), + shutdownAndDrain: vi.fn(async () => { + events.push("child-drain-start"); + await drain.promise; + events.push("child-drain-end"); + }), + persistForShutdown: vi.fn(() => events.push("session-persist")), + flushNotificationsForShutdown: vi.fn(async () => undefined) + }; + controller.megaWebFallback = { dispose: vi.fn(() => events.push("runtime-dispose")) }; + controller.realDebridWebFallbacks = new Map(); + controller.pendingRealDebridWebAccountIds = new Map(); + controller.allDebridWebFallback = { dispose: vi.fn() }; + controller.bestDebridWebFallback = { dispose: vi.fn() }; + controller.shutdownLogStorage = vi.fn(); + controller.audit = vi.fn(); + controller.settings = { historyRetentionMode: "never" }; + + const shutdown = controller.shutdown(); + await Promise.resolve(); + expect(events).toEqual(["queue-close", "child-drain-start"]); + drain.resolve(); + await shutdown; + + expect(events).toEqual([ + "queue-close", + "child-drain-start", + "child-drain-end", + "session-persist", + "runtime-dispose" + ]); + }); + it("AppController waits for the bounded outbox drain before disposing runtime owners", async () => { const drain = deferred(); - const manager = { prepareForShutdown: vi.fn() }; + const manager = { + prepareForShutdown: vi.fn(), + shutdownAndDrain: vi.fn(async () => undefined), + persistForShutdown: vi.fn() + }; const controller = Object.create(AppController.prototype) as any; controller.runtimeStatsTimer = null; controller.notificationOutbox = { drainForShutdown: vi.fn(() => drain.promise) }; @@ -88,6 +136,7 @@ describe("main shutdown lifecycle", () => { const shutdown = controller.shutdown(); expect(shutdown).toBeInstanceOf(Promise); + await vi.waitFor(() => expect(controller.notificationOutbox.drainForShutdown).toHaveBeenCalledTimes(1)); const drainBudget = controller.notificationOutbox.drainForShutdown.mock.calls[0][0]; expect(drainBudget).toBeGreaterThan(0); expect(drainBudget).toBeLessThanOrEqual(3000);