mirror of
https://github.com/4gray/iptvnator.git
synced 2026-10-13 04:26:16 -08:00
fix(downloads): persist archive ownership throughout its lifecycle
This commit is contained in:
1 parent
7c186dcc8e
commit
edbaa2de25
18 files changed
+416
-61
No files matched your search
@@ -1039,6 +1039,10 @@ expiry. `download-schema.ts` owns the transactional CHECK/index migration;
|
||||
`download_archive_finalizations` table records write-ahead file identity/size
|
||||
proof before promotion (before writing a fallback copy), allowing startup to
|
||||
recover completed unknown-length archives and clean only their owned partials.
|
||||
The same journal stores transfer-phase descriptor identity before truncation;
|
||||
Resume checks it at open, and rejected replacements are preserved and detached
|
||||
so Retry can reserve a fresh path. A synchronous completion-commit boundary
|
||||
rejects late pause/cancel commands before awaited cleanup and persistence.
|
||||
Archive transfers validate TS framing, restart from byte zero after interruption
|
||||
and check expiry again at transfer start. Completed cards play locally and never
|
||||
route to VOD details. Contract and EOF/duration limits:
|
||||
|
||||
@@ -1850,6 +1850,10 @@ expiry. `download-schema.ts` owns the transactional CHECK/index migration;
|
||||
`download_archive_finalizations` table records write-ahead file identity/size
|
||||
proof before promotion (before writing a fallback copy), allowing startup to
|
||||
recover completed unknown-length archives and clean only their owned partials.
|
||||
The same journal stores transfer-phase descriptor identity before truncation;
|
||||
Resume checks it at open, and rejected replacements are preserved and detached
|
||||
so Retry can reserve a fresh path. A synchronous completion-commit boundary
|
||||
rejects late pause/cancel commands before awaited cleanup and persistence.
|
||||
Archive transfers validate TS framing, restart from byte zero after interruption
|
||||
and check expiry again at transfer start. Completed cards play locally and never
|
||||
route to VOD details. Contract and EOF/duration limits:
|
||||
|
||||
@@ -7,16 +7,48 @@ import type { DownloadsDatabase } from './download-task';
|
||||
|
||||
export interface ArchiveFinalizationProof {
|
||||
version: 1;
|
||||
phase?: 'finalization';
|
||||
filePath: string;
|
||||
size: number;
|
||||
partialIdentity: ArchiveFileIdentity;
|
||||
finalIdentity: ArchiveFileIdentity;
|
||||
}
|
||||
|
||||
export interface ArchivePartialProof {
|
||||
version: 1;
|
||||
phase: 'transfer';
|
||||
filePath: string;
|
||||
partialIdentity: ArchiveFileIdentity;
|
||||
}
|
||||
export type ArchiveDownloadProof =
|
||||
ArchiveFinalizationProof | ArchivePartialProof;
|
||||
|
||||
export async function recordArchivePartial(
|
||||
db: DownloadsDatabase,
|
||||
downloadId: number,
|
||||
filePath: string,
|
||||
partialIdentity: ArchiveFileIdentity
|
||||
): Promise<void> {
|
||||
await writeArchiveProof(db, downloadId, {
|
||||
version: 1,
|
||||
phase: 'transfer',
|
||||
filePath,
|
||||
partialIdentity,
|
||||
});
|
||||
}
|
||||
|
||||
export async function recordArchiveFinalization(
|
||||
db: DownloadsDatabase,
|
||||
downloadId: number,
|
||||
proof: ArchiveFinalizationProof
|
||||
): Promise<void> {
|
||||
await writeArchiveProof(db, downloadId, proof);
|
||||
}
|
||||
|
||||
async function writeArchiveProof(
|
||||
db: DownloadsDatabase,
|
||||
downloadId: number,
|
||||
proof: ArchiveDownloadProof
|
||||
): Promise<void> {
|
||||
const serialized = JSON.stringify({
|
||||
...proof,
|
||||
@@ -24,10 +56,14 @@ export async function recordArchiveFinalization(
|
||||
dev: proof.partialIdentity.dev,
|
||||
ino: proof.partialIdentity.ino,
|
||||
},
|
||||
finalIdentity: {
|
||||
dev: proof.finalIdentity.dev,
|
||||
ino: proof.finalIdentity.ino,
|
||||
},
|
||||
...(proof.phase !== 'transfer'
|
||||
? {
|
||||
finalIdentity: {
|
||||
dev: proof.finalIdentity.dev,
|
||||
ino: proof.finalIdentity.ino,
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
});
|
||||
await db
|
||||
.insert(schema.downloadArchiveFinalizations)
|
||||
@@ -38,7 +74,7 @@ export async function recordArchiveFinalization(
|
||||
});
|
||||
}
|
||||
|
||||
/** An explicitly restarted transfer must never inherit an earlier attempt's proof. */
|
||||
/** Fresh reservations must never inherit an earlier attempt's proof. */
|
||||
export async function clearArchiveFinalization(
|
||||
db: DownloadsDatabase,
|
||||
downloadId: number
|
||||
@@ -59,15 +95,21 @@ function identity(value: unknown): value is ArchiveFileIdentity {
|
||||
|
||||
export function parseArchiveFinalization(
|
||||
value: string
|
||||
): ArchiveFinalizationProof | undefined {
|
||||
): ArchiveDownloadProof | undefined {
|
||||
try {
|
||||
const proof = JSON.parse(value) as ArchiveFinalizationProof;
|
||||
return proof?.version === 1 &&
|
||||
typeof proof.filePath === 'string' &&
|
||||
isAbsolute(proof.filePath) &&
|
||||
const proof = JSON.parse(value) as ArchiveDownloadProof;
|
||||
if (
|
||||
!proof ||
|
||||
proof.version !== 1 ||
|
||||
typeof proof.filePath !== 'string' ||
|
||||
!isAbsolute(proof.filePath) ||
|
||||
!identity(proof.partialIdentity)
|
||||
)
|
||||
return undefined;
|
||||
if (proof.phase === 'transfer') return proof;
|
||||
return (proof.phase === undefined || proof.phase === 'finalization') &&
|
||||
Number.isSafeInteger(proof.size) &&
|
||||
proof.size > 0 &&
|
||||
identity(proof.partialIdentity) &&
|
||||
identity(proof.finalIdentity)
|
||||
? proof
|
||||
: undefined;
|
||||
@@ -79,9 +121,9 @@ export function parseArchiveFinalization(
|
||||
export async function readArchiveFinalizations(
|
||||
db: DownloadsDatabase,
|
||||
ids: number[]
|
||||
): Promise<Map<number, ArchiveFinalizationProof>> {
|
||||
): Promise<Map<number, ArchiveDownloadProof>> {
|
||||
if (ids.length === 0) return new Map();
|
||||
const result = new Map<number, ArchiveFinalizationProof>();
|
||||
const result = new Map<number, ArchiveDownloadProof>();
|
||||
for (let offset = 0; offset < ids.length; offset += 500) {
|
||||
const rows = await db
|
||||
.select()
|
||||
@@ -102,9 +144,10 @@ export async function readArchiveFinalizations(
|
||||
|
||||
export function verifiedArchiveSize(
|
||||
filePath: string | null,
|
||||
proof: ArchiveFinalizationProof | undefined
|
||||
proof: ArchiveDownloadProof | undefined
|
||||
): number | null {
|
||||
if (!proof || proof.filePath !== filePath) return null;
|
||||
if (!proof || proof.phase === 'transfer' || proof.filePath !== filePath)
|
||||
return null;
|
||||
try {
|
||||
const file = lstatSync(proof.filePath);
|
||||
return file.isFile() &&
|
||||
|
||||
@@ -6,11 +6,17 @@ export interface ArchiveFileIdentity {
|
||||
readonly ino: number;
|
||||
}
|
||||
|
||||
export class ArchivePartialReplacedError extends Error {
|
||||
constructor() {
|
||||
super('Archive partial changed before opening');
|
||||
}
|
||||
}
|
||||
|
||||
/** Open without truncation first: a replaced partial must not damage its target. */
|
||||
export async function openCatchupOutput(
|
||||
partialPath: string,
|
||||
expectedIdentity?: ArchiveFileIdentity,
|
||||
beforeTruncate?: () => Promise<void>
|
||||
beforeTruncate?: (identity: ArchiveFileIdentity) => Promise<void>
|
||||
): Promise<{ stream: WriteStream; identity: ArchiveFileIdentity }> {
|
||||
const before = await lstat(partialPath).catch(
|
||||
(error: NodeJS.ErrnoException) => {
|
||||
@@ -19,7 +25,7 @@ export async function openCatchupOutput(
|
||||
}
|
||||
);
|
||||
if (before && (!before.isFile() || before.nlink !== 1)) {
|
||||
throw new Error('Archive partial is not an exclusive regular file');
|
||||
throw new ArchivePartialReplacedError();
|
||||
}
|
||||
const handle = await open(
|
||||
partialPath,
|
||||
@@ -42,10 +48,10 @@ export async function openCatchupOutput(
|
||||
(current.dev !== expectedIdentity.dev ||
|
||||
current.ino !== expectedIdentity.ino))
|
||||
) {
|
||||
throw new Error('Archive partial changed before opening');
|
||||
throw new ArchivePartialReplacedError();
|
||||
}
|
||||
// Keep durable ownership evidence until the actual descriptor passes.
|
||||
await beforeTruncate?.();
|
||||
await beforeTruncate?.({ dev: current.dev, ino: current.ino });
|
||||
await handle.truncate(0);
|
||||
// All writes use this verified descriptor; never reopen by pathname.
|
||||
return {
|
||||
|
||||
@@ -14,6 +14,7 @@ import * as schema from '../../database/schema';
|
||||
import { finalizeCatchupPartial } from './download-catchup-finalize';
|
||||
import {
|
||||
recordArchiveFinalization,
|
||||
recordArchivePartial,
|
||||
parseArchiveFinalization,
|
||||
} from './download-catchup-journal';
|
||||
import { resetStaleDownloads } from './download-recovery';
|
||||
@@ -259,3 +260,32 @@ it.each(['downloading', 'queued'])(
|
||||
);
|
||||
}
|
||||
);
|
||||
|
||||
it.each([false, true])(
|
||||
'recovers transfer-phase identity without treating it as completion (replaced=%s)',
|
||||
async (replaced) => {
|
||||
const { partial } = await prepare();
|
||||
await recordArchivePartial(db, 1, filePath, partial);
|
||||
// A same-sized final is insufficient without a finalization-phase journal.
|
||||
await writeFile(filePath, 'verified archive');
|
||||
if (replaced) {
|
||||
await rename(filePath + '.part', join(directory, 'original'));
|
||||
await writeFile(filePath + '.part', 'unrelated bytes!');
|
||||
}
|
||||
await resetStaleDownloads();
|
||||
expect(updates).toContainEqual(
|
||||
expect.objectContaining(
|
||||
replaced
|
||||
? { status: 'failed', filePath: null }
|
||||
: { status: 'paused' }
|
||||
)
|
||||
);
|
||||
expect(updates.some((value) => value.status === 'completed')).toBe(
|
||||
false
|
||||
);
|
||||
expect(await readFile(filePath, 'utf8')).toBe('verified archive');
|
||||
expect(await readFile(filePath + '.part', 'utf8')).toBe(
|
||||
replaced ? 'unrelated bytes!' : 'verified archive'
|
||||
);
|
||||
}
|
||||
);
|
||||
@@ -0,0 +1,72 @@
|
||||
import {
|
||||
mkdtemp,
|
||||
lstat,
|
||||
readFile,
|
||||
rename,
|
||||
rm,
|
||||
writeFile,
|
||||
} from 'node:fs/promises';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
import { reserveFreshCatchupTarget } from './download-catchup-reservation';
|
||||
import { clearArchiveFinalization } from './download-catchup-journal';
|
||||
import type { DownloadsDatabase, DownloadTask } from './download-task';
|
||||
|
||||
jest.mock('./download-catchup-journal', () => ({
|
||||
clearArchiveFinalization: jest.fn().mockResolvedValue(undefined),
|
||||
}));
|
||||
|
||||
it.each([false, true])(
|
||||
'handles an occupied final without relocating a partial (replaced=%s)',
|
||||
async (replaced) => {
|
||||
const directory = await mkdtemp(join(tmpdir(), 'archive-collision-'));
|
||||
const filePath = join(directory, 'show.ts');
|
||||
try {
|
||||
await writeFile(filePath, 'unrelated final');
|
||||
await writeFile(filePath + '.part', 'owned bytes');
|
||||
const task: DownloadTask = {
|
||||
id: 1,
|
||||
filePath,
|
||||
directory,
|
||||
fileName: 'show.ts',
|
||||
url: 'https://provider.test/archive.ts',
|
||||
catchupExpectedPartialIdentity: await lstat(filePath + '.part'),
|
||||
};
|
||||
if (replaced) {
|
||||
await rename(
|
||||
filePath + '.part',
|
||||
join(directory, 'owned-original')
|
||||
);
|
||||
await writeFile(filePath + '.part', 'unrelated partial');
|
||||
}
|
||||
jest.mocked(clearArchiveFinalization).mockClear();
|
||||
const reserve = reserveFreshCatchupTarget(
|
||||
{} as DownloadsDatabase,
|
||||
task
|
||||
);
|
||||
if (replaced) {
|
||||
await expect(reserve).rejects.toThrow('changed');
|
||||
expect(await readFile(filePath + '.part', 'utf8')).toBe(
|
||||
'unrelated partial'
|
||||
);
|
||||
await expect(
|
||||
lstat(join(directory, 'show (1).ts.part'))
|
||||
).rejects.toMatchObject({ code: 'ENOENT' });
|
||||
expect(clearArchiveFinalization).not.toHaveBeenCalled();
|
||||
} else {
|
||||
await expect(reserve).resolves.toEqual({
|
||||
path: join(directory, 'show (1).ts'),
|
||||
partialPath: join(directory, 'show (1).ts.part'),
|
||||
filename: 'show (1).ts',
|
||||
});
|
||||
await expect(lstat(filePath + '.part')).rejects.toMatchObject({
|
||||
code: 'ENOENT',
|
||||
});
|
||||
expect(task.catchupExpectedPartialIdentity).toBeUndefined();
|
||||
}
|
||||
expect(await readFile(filePath, 'utf8')).toBe('unrelated final');
|
||||
} finally {
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
);
|
||||
@@ -0,0 +1,37 @@
|
||||
import { lstat } from 'node:fs/promises';
|
||||
import { cleanupCatchupPartial } from './download-catchup-cleanup';
|
||||
import { clearArchiveFinalization } from './download-catchup-journal';
|
||||
import { ArchivePartialReplacedError } from './download-catchup-output';
|
||||
import { reserveAvailablePartialDownloadFile } from './download-file-path';
|
||||
import type { DownloadsDatabase, DownloadTask } from './download-task';
|
||||
|
||||
/** Archives restart at zero, so never relocate a retained entry on collision. */
|
||||
export async function reserveFreshCatchupTarget(
|
||||
db: DownloadsDatabase,
|
||||
task: DownloadTask
|
||||
) {
|
||||
if (task.filePath) {
|
||||
const partial = await lstat(`${task.filePath}.part`).catch(
|
||||
(error: NodeJS.ErrnoException) => {
|
||||
if (error.code === 'ENOENT') return undefined;
|
||||
throw error;
|
||||
}
|
||||
);
|
||||
const expected = task.catchupExpectedPartialIdentity;
|
||||
if (
|
||||
partial &&
|
||||
(!expected ||
|
||||
!partial.isFile() ||
|
||||
partial.dev !== expected.dev ||
|
||||
partial.ino !== expected.ino)
|
||||
) {
|
||||
throw new ArchivePartialReplacedError();
|
||||
}
|
||||
if (!(await cleanupCatchupPartial(task.filePath, expected))) {
|
||||
throw new Error('Could not remove the owned archive partial');
|
||||
}
|
||||
}
|
||||
await clearArchiveFinalization(db, task.id);
|
||||
task.catchupExpectedPartialIdentity = undefined;
|
||||
return reserveAvailablePartialDownloadFile(task.directory, task.fileName);
|
||||
}
|
||||
@@ -1,3 +1,5 @@
|
||||
import { ArchivePartialReplacedError } from './download-catchup-output';
|
||||
import { handleDownloadFailure } from './download-finalize';
|
||||
import {
|
||||
lstat,
|
||||
mkdtemp,
|
||||
@@ -14,6 +16,8 @@ import {
|
||||
enqueueDownload,
|
||||
pauseDownload,
|
||||
cancelDownload,
|
||||
isDownloadCommitting,
|
||||
removeDownloadFromRuntime,
|
||||
} from './download-runtime';
|
||||
import { cleanupCatchupFile } from './download-catchup-cleanup';
|
||||
import { readArchiveFinalizations } from './download-catchup-journal';
|
||||
@@ -195,9 +199,11 @@ it.each([1880, null])(
|
||||
jest.mocked(cleanupCatchupFile).mockImplementationOnce(
|
||||
async (path, identity) => {
|
||||
// The file is verified, but cleanup is still awaiting filesystem I/O.
|
||||
expect(path).toBe(join(directory, 'show.ts.part'));
|
||||
expect(path).toBe(task.filePath + '.part');
|
||||
lateCommands.push(await pauseDownload(task.id));
|
||||
lateCommands.push(await cancelDownload(task.id));
|
||||
expect(isDownloadCommitting(task.id)).toBe(true);
|
||||
lateCommands.push(removeDownloadFromRuntime(task.id));
|
||||
expect(task.pauseRequested).not.toBe(true);
|
||||
expect(task.cancelRequested).not.toBe(true);
|
||||
return jest
|
||||
@@ -208,19 +214,76 @@ it.each([1880, null])(
|
||||
}
|
||||
);
|
||||
try {
|
||||
if (totalBytes === null) {
|
||||
task.filePath = join(directory, 'show.ts');
|
||||
await writeFile(
|
||||
task.filePath + '.part',
|
||||
'unproven retained file'
|
||||
);
|
||||
}
|
||||
enqueueDownload(task);
|
||||
await settled;
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
expect(lateCommands).toEqual([false, false]);
|
||||
expect(lateCommands).toEqual([false, false, false]);
|
||||
expect(terminalStatus).toBe('completed');
|
||||
expect(completionAttempts).toBe(2);
|
||||
expect(await readFile(join(directory, 'show.ts'))).toEqual(body);
|
||||
await expect(
|
||||
lstat(join(directory, 'show.ts.part'))
|
||||
).rejects.toMatchObject({ code: 'ENOENT' });
|
||||
expect(await readFile(task.filePath!)).toEqual(body);
|
||||
if (totalBytes === null) {
|
||||
expect(task.filePath).toBe(join(directory, 'show (1).ts'));
|
||||
expect(
|
||||
await readFile(join(directory, 'show.ts.part'), 'utf8')
|
||||
).toBe('unproven retained file');
|
||||
}
|
||||
await expect(lstat(task.filePath + '.part')).rejects.toMatchObject({
|
||||
code: 'ENOENT',
|
||||
});
|
||||
expect(transferCatchupToPartialFile).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
it('detaches a rejected replacement so Retry can reserve a fresh path', async () => {
|
||||
const directory = await mkdtemp(join(tmpdir(), 'archive-detached-'));
|
||||
const filePath = join(directory, 'show.ts');
|
||||
const task: DownloadTask = {
|
||||
id: 994,
|
||||
directory,
|
||||
filePath,
|
||||
fileName: 'show.ts',
|
||||
url: 'https://provider.test/show.ts',
|
||||
catchup: {
|
||||
channelName: 'News',
|
||||
startTimestamp: 100,
|
||||
stopTimestamp: 200,
|
||||
},
|
||||
};
|
||||
const updates: Record<string, unknown>[] = [];
|
||||
const db = {
|
||||
update: () => ({
|
||||
set: (value: Record<string, unknown>) => ({
|
||||
where: async () => {
|
||||
updates.push(value);
|
||||
},
|
||||
}),
|
||||
}),
|
||||
};
|
||||
try {
|
||||
await writeFile(filePath + '.part', 'unrelated user file');
|
||||
await handleDownloadFailure(
|
||||
db as never,
|
||||
task,
|
||||
undefined,
|
||||
new ArchivePartialReplacedError()
|
||||
);
|
||||
expect(updates.at(-1)).toEqual(
|
||||
expect.objectContaining({ status: 'failed', filePath: null })
|
||||
);
|
||||
expect(await readFile(filePath + '.part', 'utf8')).toBe(
|
||||
'unrelated user file'
|
||||
);
|
||||
} finally {
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
@@ -1,4 +1,4 @@
|
||||
import { clearArchiveFinalization } from './download-catchup-journal';
|
||||
import { recordArchivePartial } from './download-catchup-journal';
|
||||
import {
|
||||
mkdtemp,
|
||||
readFile,
|
||||
@@ -29,7 +29,7 @@ import { stat } from 'node:fs/promises';
|
||||
|
||||
jest.mock('./download-catchup-journal', () => ({
|
||||
...jest.requireActual('./download-catchup-journal'),
|
||||
clearArchiveFinalization: jest.fn().mockResolvedValue(undefined),
|
||||
recordArchivePartial: jest.fn().mockResolvedValue(undefined),
|
||||
}));
|
||||
jest.mock('./download-catchup-limits', () => {
|
||||
const actual = jest.requireActual('./download-catchup-limits');
|
||||
@@ -123,8 +123,14 @@ describe('TS archive transfer', () => {
|
||||
};
|
||||
try {
|
||||
await writeFile(path + '.part', 'old data');
|
||||
jest.mocked(clearArchiveFinalization).mockImplementationOnce(
|
||||
async (_db, id) => {
|
||||
jest.mocked(recordArchivePartial).mockImplementationOnce(
|
||||
async (_db, id, filePath, identity) => {
|
||||
expect(filePath).toBe(path);
|
||||
expect(identity).toEqual(
|
||||
expect.objectContaining({
|
||||
ino: (await stat(path + '.part')).ino,
|
||||
})
|
||||
);
|
||||
expect(id).toBe(task.id);
|
||||
expect(await readFile(path + '.part', 'utf8')).toBe(
|
||||
'old data'
|
||||
@@ -180,9 +186,14 @@ describe('TS archive transfer', () => {
|
||||
expect.anything()
|
||||
);
|
||||
expect(task.resumeValidator).toBeNull();
|
||||
expect(clearArchiveFinalization).toHaveBeenCalledWith(
|
||||
expect(recordArchivePartial).toHaveBeenCalledWith(
|
||||
expect.anything(),
|
||||
task.id
|
||||
task.id,
|
||||
path,
|
||||
expect.objectContaining({
|
||||
dev: expect.any(Number),
|
||||
ino: expect.any(Number),
|
||||
})
|
||||
);
|
||||
} finally {
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
@@ -250,8 +261,8 @@ it.each(['symlink', 'hardlink', 'regular file'])(
|
||||
directory,
|
||||
catchup: metadata,
|
||||
};
|
||||
const journalClears = jest.mocked(clearArchiveFinalization).mock
|
||||
.calls.length;
|
||||
const journalClears =
|
||||
jest.mocked(recordArchivePartial).mock.calls.length;
|
||||
await expect(
|
||||
transferCatchupToPartialFile({} as DownloadsDatabase, task, {
|
||||
path,
|
||||
@@ -259,9 +270,9 @@ it.each(['symlink', 'hardlink', 'regular file'])(
|
||||
filename: 'archive.ts',
|
||||
})
|
||||
).rejects.toThrow();
|
||||
expect(
|
||||
jest.mocked(clearArchiveFinalization).mock.calls.length
|
||||
).toBe(journalClears);
|
||||
expect(jest.mocked(recordArchivePartial).mock.calls.length).toBe(
|
||||
journalClears
|
||||
);
|
||||
expect(await readFile(path + '.part', 'utf8')).toBe(
|
||||
'keep this file intact'
|
||||
);
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { clearArchiveFinalization } from './download-catchup-journal';
|
||||
import { recordArchivePartial } from './download-catchup-journal';
|
||||
import {
|
||||
assertArchiveCopyHeadroom,
|
||||
createArchiveByteGuard,
|
||||
@@ -102,7 +102,8 @@ export async function transferCatchupToPartialFile(
|
||||
output = await openCatchupOutput(
|
||||
reservation.partialPath,
|
||||
task.catchupExpectedPartialIdentity,
|
||||
() => clearArchiveFinalization(db, task.id)
|
||||
(identity) =>
|
||||
recordArchivePartial(db, task.id, reservation.path, identity)
|
||||
);
|
||||
output.stream.on('error', () => undefined);
|
||||
task.catchupPartialIdentity = output.identity;
|
||||
|
||||
@@ -1,4 +1,8 @@
|
||||
import { removePartialDownloadFile } from './download-file-path';
|
||||
import type { DownloadTask } from './download-task';
|
||||
import {
|
||||
getPartialDownloadSize,
|
||||
removePartialDownloadFile,
|
||||
} from './download-file-path';
|
||||
import { constants } from 'node:fs';
|
||||
import { copyFile, link, stat, unlink } from 'node:fs/promises';
|
||||
import type { ReservedPartialDownloadFile } from './download-file-path';
|
||||
@@ -75,3 +79,12 @@ export function removePartialFile(
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
export function getPausedByteCount(task: DownloadTask): number {
|
||||
try {
|
||||
return getPartialDownloadSize(task.filePath);
|
||||
} catch (error) {
|
||||
console.error('[Downloads] Failed to inspect partial file:', error);
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import { ArchivePartialReplacedError } from './download-catchup-output';
|
||||
import { recordArchiveFinalization } from './download-catchup-journal';
|
||||
import {
|
||||
finalizePartialDownload,
|
||||
@@ -106,7 +107,12 @@ export async function handleDownloadFailure(
|
||||
.update(schema.downloads)
|
||||
.set({
|
||||
errorMessage: describeError(error),
|
||||
filePath: task.catchup && !removed ? (task.filePath ?? null) : null,
|
||||
filePath:
|
||||
task.catchup &&
|
||||
!removed &&
|
||||
!(error instanceof ArchivePartialReplacedError)
|
||||
? (task.filePath ?? null)
|
||||
: null,
|
||||
resumeValidator: null,
|
||||
status: 'failed',
|
||||
updatedAt: sql`CURRENT_TIMESTAMP`,
|
||||
@@ -338,13 +344,7 @@ export function getCompletedPartialProgress(
|
||||
}
|
||||
}
|
||||
|
||||
export function getPausedByteCount(task: DownloadTask): number {
|
||||
try {
|
||||
return getPartialDownloadSize(task.filePath);
|
||||
} catch (error) {
|
||||
console.error('[Downloads] Failed to inspect partial file:', error);
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
export { removePartialFile } from './download-file-finalize';
|
||||
export {
|
||||
getPausedByteCount,
|
||||
removePartialFile,
|
||||
} from './download-file-finalize';
|
||||
@@ -5,7 +5,7 @@ import {
|
||||
import {
|
||||
readArchiveFinalizations,
|
||||
verifiedArchiveSize,
|
||||
type ArchiveFinalizationProof,
|
||||
type ArchiveDownloadProof,
|
||||
} from './download-catchup-journal';
|
||||
import { inArray, sql } from 'drizzle-orm';
|
||||
import { lstatSync, statSync } from 'node:fs';
|
||||
@@ -18,7 +18,7 @@ import {
|
||||
|
||||
interface StaleDownload {
|
||||
contentType?: string;
|
||||
proof?: ArchiveFinalizationProof;
|
||||
proof?: ArchiveDownloadProof;
|
||||
filePath: string | null;
|
||||
id: number;
|
||||
status: string;
|
||||
@@ -193,6 +193,7 @@ export async function resetStaleDownloads(): Promise<void> {
|
||||
if (
|
||||
download.contentType === 'catchup' &&
|
||||
download.proof &&
|
||||
download.proof.phase !== 'transfer' &&
|
||||
!finalizedIds.has(download.id) &&
|
||||
verifiedArchiveSize(download.filePath, download.proof) === null
|
||||
) {
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { reserveFreshCatchupTarget } from './download-catchup-reservation';
|
||||
import {
|
||||
clearArchiveFinalization,
|
||||
readArchiveFinalizations,
|
||||
@@ -32,6 +33,7 @@ import {
|
||||
requestDownloadCancellation,
|
||||
requestDownloadPause,
|
||||
type DownloadTask,
|
||||
type DownloadsDatabase,
|
||||
} from './download-task';
|
||||
import { describeError } from './download-transfer';
|
||||
|
||||
@@ -132,9 +134,16 @@ export async function cancelDownload(downloadId: number): Promise<boolean> {
|
||||
return true;
|
||||
}
|
||||
|
||||
export function removeDownloadFromRuntime(downloadId: number): void {
|
||||
export function isDownloadCommitting(downloadId: number): boolean {
|
||||
return (
|
||||
activeDownload?.id === downloadId &&
|
||||
!!activeDownload.catchupCommitStarted
|
||||
);
|
||||
}
|
||||
|
||||
export function removeDownloadFromRuntime(downloadId: number): boolean {
|
||||
if (activeDownload?.id === downloadId) {
|
||||
requestDownloadCancellation(activeDownload);
|
||||
if (!requestDownloadCancellation(activeDownload)) return false;
|
||||
}
|
||||
|
||||
const queueIndex = downloadQueue.findIndex(
|
||||
@@ -143,6 +152,7 @@ export function removeDownloadFromRuntime(downloadId: number): void {
|
||||
if (queueIndex !== -1) {
|
||||
downloadQueue.splice(queueIndex, 1);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
async function processQueue(): Promise<void> {
|
||||
@@ -204,12 +214,16 @@ async function startDownload(task: DownloadTask): Promise<void> {
|
||||
await readArchiveFinalizations(db, [task.id])
|
||||
).get(task.id);
|
||||
task.catchupExpectedPartialIdentity = proof?.partialIdentity;
|
||||
} else {
|
||||
// No durable ownership evidence: preserve the old entry and
|
||||
// reserve a fresh destination instead of adopting it.
|
||||
if (!proof) task.filePath = null;
|
||||
}
|
||||
if (!task.filePath) {
|
||||
// A fresh reservation must not inherit a previous attempt's proof.
|
||||
await clearArchiveFinalization(db, task.id);
|
||||
}
|
||||
}
|
||||
reservation = await reserveTarget(task);
|
||||
reservation = await reserveTarget(db, task);
|
||||
task.fileName = reservation.filename;
|
||||
task.filePath = reservation.path;
|
||||
await db
|
||||
@@ -273,6 +287,7 @@ async function startDownload(task: DownloadTask): Promise<void> {
|
||||
}
|
||||
|
||||
async function reserveTarget(
|
||||
db: DownloadsDatabase,
|
||||
task: DownloadTask
|
||||
): Promise<ReservedPartialDownloadFile> {
|
||||
if (task.filePath) {
|
||||
@@ -284,6 +299,8 @@ async function reserveTarget(
|
||||
};
|
||||
}
|
||||
|
||||
if (task.catchup) return reserveFreshCatchupTarget(db, task);
|
||||
|
||||
// Something now occupies the recorded destination — possibly a file
|
||||
// the user created while this download was paused or failed. Never
|
||||
// inspect or delete it: move the retained .part to the next free
|
||||
|
||||
@@ -4,6 +4,7 @@ import {
|
||||
mockBroadcastDownloadUpdate,
|
||||
mockDownloadRow,
|
||||
mockRemoveDownloadFromRuntime,
|
||||
mockIsDownloadCommitting,
|
||||
mockRemovePartialDownloadFile,
|
||||
mockTerminalRows,
|
||||
setupDownloadsEventsHarness,
|
||||
@@ -14,6 +15,34 @@ describe('downloads events: partial-file cleanup', () => {
|
||||
await setupDownloadsEventsHarness();
|
||||
});
|
||||
|
||||
it('keeps a committing archive row and journal intact on Remove', async () => {
|
||||
const { deleteWhere } = mockDownloadRow(
|
||||
createDownloadRow('downloading')
|
||||
);
|
||||
mockIsDownloadCommitting.mockReturnValue(true);
|
||||
await expect(getHandler('DOWNLOADS_REMOVE')(null, 42)).resolves.toEqual(
|
||||
{
|
||||
success: false,
|
||||
error: 'Download is completing; try again shortly',
|
||||
}
|
||||
);
|
||||
expect(mockRemovePartialDownloadFile).not.toHaveBeenCalled();
|
||||
expect(mockRemoveDownloadFromRuntime).not.toHaveBeenCalled();
|
||||
expect(deleteWhere).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('skips completion commits when clearing terminal rows', async () => {
|
||||
const { deleteWhere } = mockTerminalRows([
|
||||
createDownloadRow('completed'),
|
||||
]);
|
||||
mockIsDownloadCommitting.mockReturnValue(true);
|
||||
await expect(
|
||||
getHandler('DOWNLOADS_CLEAR_COMPLETED')(null)
|
||||
).resolves.toEqual({ success: true });
|
||||
expect(mockRemovePartialDownloadFile).not.toHaveBeenCalled();
|
||||
expect(deleteWhere).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('removes queued resumed partial files before deleting the row', async () => {
|
||||
const { deleteWhere } = mockDownloadRow(createDownloadRow('queued'));
|
||||
|
||||
|
||||
@@ -23,6 +23,7 @@ import { resetStaleDownloads } from './download-recovery';
|
||||
import {
|
||||
broadcastDownloadUpdate,
|
||||
cancelDownload,
|
||||
isDownloadCommitting,
|
||||
pauseDownload,
|
||||
removeDownloadFromRuntime,
|
||||
setMainWindow,
|
||||
@@ -189,6 +190,11 @@ ipcMain.handle('DOWNLOADS_REMOVE', async (_event, downloadId: number) => {
|
||||
.where(eq(schema.downloads.id, downloadId))
|
||||
.limit(1);
|
||||
const row = rows[0];
|
||||
if (isDownloadCommitting(downloadId))
|
||||
return {
|
||||
success: false,
|
||||
error: 'Download is completing; try again shortly',
|
||||
};
|
||||
if (row?.filePath && removablePartialStatuses.has(row.status)) {
|
||||
try {
|
||||
removePartialDownloadFile(row.filePath);
|
||||
@@ -208,7 +214,11 @@ ipcMain.handle('DOWNLOADS_REMOVE', async (_event, downloadId: number) => {
|
||||
};
|
||||
}
|
||||
}
|
||||
removeDownloadFromRuntime(downloadId);
|
||||
if (removeDownloadFromRuntime(downloadId) === false)
|
||||
return {
|
||||
success: false,
|
||||
error: 'Download is completing; try again shortly',
|
||||
};
|
||||
await db
|
||||
.delete(schema.downloads)
|
||||
.where(eq(schema.downloads.id, downloadId));
|
||||
@@ -331,6 +341,7 @@ ipcMain.handle(
|
||||
.where(terminalFilter);
|
||||
const downloadIdsToDelete: number[] = [];
|
||||
for (const row of rows) {
|
||||
if (isDownloadCommitting(row.id)) continue;
|
||||
if (row.filePath && removablePartialStatuses.has(row.status)) {
|
||||
try {
|
||||
removePartialDownloadFile(row.filePath);
|
||||
|
||||
@@ -11,6 +11,7 @@ type IpcHandler = (_event: unknown, ...args: unknown[]) => Promise<unknown>;
|
||||
export const mockRegisteredHandlers = new Map<string, IpcHandler>();
|
||||
export const mockGetDatabase = jest.fn();
|
||||
export const mockRemoveDownloadFromRuntime = jest.fn();
|
||||
export const mockIsDownloadCommitting = jest.fn();
|
||||
export const mockBroadcastDownloadUpdate = jest.fn();
|
||||
export const mockRemovePartialDownloadFile = jest.fn();
|
||||
export const mockPauseDownload = jest.fn();
|
||||
@@ -53,6 +54,7 @@ export async function setupDownloadsEventsHarness(): Promise<void> {
|
||||
mockRegisteredHandlers.clear();
|
||||
mockGetDatabase.mockReset();
|
||||
mockRemoveDownloadFromRuntime.mockReset();
|
||||
mockIsDownloadCommitting.mockReset().mockReturnValue(false);
|
||||
mockBroadcastDownloadUpdate.mockReset();
|
||||
mockRemovePartialDownloadFile.mockReset();
|
||||
mockPauseDownload.mockReset();
|
||||
@@ -114,6 +116,7 @@ export async function setupDownloadsEventsHarness(): Promise<void> {
|
||||
jest.doMock('./download-runtime', () => ({
|
||||
broadcastDownloadUpdate: mockBroadcastDownloadUpdate,
|
||||
cancelDownload: jest.fn(),
|
||||
isDownloadCommitting: mockIsDownloadCommitting,
|
||||
pauseDownload: mockPauseDownload,
|
||||
removeDownloadFromRuntime: mockRemoveDownloadFromRuntime,
|
||||
setMainWindow: jest.fn(),
|
||||
|
||||
@@ -67,9 +67,17 @@ journal also makes startup partial cleanup identity-aware. A journaled partial
|
||||
whose identity changed is preserved and detached from the failed row; Retry
|
||||
reserves a fresh path instead of adopting or truncating it. The journal remains with
|
||||
completed archives until they are removed. An explicitly restarted transfer
|
||||
carries the journaled source identity through any destination renaming and checks
|
||||
it against the opened descriptor. Only then does it clear the previous proof,
|
||||
before truncation or writes; a rejected replacement keeps its journal for retry. Process-local proof allows immediate
|
||||
carries the journaled source identity to output opening and checks
|
||||
it against the opened descriptor. Only then does it replace the previous proof,
|
||||
before truncation or writes, replacing the old finalization proof with a
|
||||
transfer-phase journal containing the newly opened identity. That identity
|
||||
survives pause, failure and restart without claiming the file is complete. A
|
||||
rejected replacement is detached from the failed row without deletion, so Retry
|
||||
reserves a new path. Retained catch-up paths without any durable proof are also
|
||||
preserved and redirected to a fresh reservation. On a final-path collision,
|
||||
archives never relocate a retained partial: only a verified owned partial is
|
||||
discarded before reserving a new filename, because archive retries restart at zero. The transfer proof is upgraded to finalization proof only
|
||||
after validated EOF. Process-local proof allows immediate
|
||||
recovery after a transient completion DB error without waiting for a restart.
|
||||
Fallback copying observes pause/cancel between bounded 64 KiB reads/writes and
|
||||
before publication completes. Interruption removes only the owned copy and
|
||||
@@ -77,7 +85,9 @@ leaves the source for the runtime pause/cancel handler. Once publication identit
|
||||
and size pass the final check, the task synchronously enters completion commit
|
||||
before awaited partial cleanup and the SQLite completion write. Pause/cancel then
|
||||
return false without setting flags; a command is never accepted and subsequently
|
||||
overwritten by completion.
|
||||
overwritten by completion. Remove rejects this committing row before partial
|
||||
cleanup or deletion, and Clear completed skips it, preserving the cascading
|
||||
journal until completion finishes.
|
||||
A kill between exclusive copy-file creation and its identity journal commit can
|
||||
leave an unowned **empty** destination: no bytes are written before the commit.
|
||||
Recovery preserves that file rather than guessing ownership; Retry uses a
|
||||
|
||||
Reference in new issue
Block a user