fix(downloads): journal archive promotion before publishing files

This commit is contained in:
4gray committed 2026-09-08 01:18:24 +02:00
1 parent 9013437aff
commit 8dfd06c05f
16 files changed
+665 -88

No files matched your search

@@ -2,7 +2,7 @@ import {
getDownloadPlayPaths,
installDownloadPlayCapture,
} from './downloads.e2e-support';
import { mkdirSync, readFileSync } from 'node:fs';
import { mkdirSync, readFileSync, readdirSync } from 'node:fs';
import { join } from 'node:path';
import type { Page } from '@playwright/test';
import {
@@ -21,6 +21,7 @@ import {
switchUnifiedCollectionScope,
test,
waitForXtreamWorkspaceReady,
workspaceRoot,
} from './electron-test-fixtures';
import { fetchXtreamEpgFixture } from './portal-mock-fixtures';
@@ -429,7 +430,51 @@ test('@downloads @epg @xtream @electron downloads a completed archive into the l
await expect(app.mainWindow).toHaveURL(
/\/workspace\/downloads(?:\?.*)?$/
);
// Library remains usable after a restart, including archive metadata.
// Model termination after verified promotion but before the completion
// DB write, including a response without Content-Length. Keep only the
// real SQLite journal written by the transfer; the next process has no task.
const durableProof = await app.electronApp.evaluate(
(_electron, { dependency, file, id }) => {
const Database = process
.getBuiltinModule('module')
.createRequire(dependency)(dependency);
const db = new Database(file);
try {
const journal = db
.prepare(
'SELECT proof FROM download_archive_finalizations WHERE download_id=?'
)
.get(id) as { proof: string } | undefined;
if (!journal)
throw new Error(
'Archive promotion journal was not persisted'
);
db.prepare(
"UPDATE downloads SET status='downloading', total_bytes=NULL WHERE id=?"
).run(id);
return JSON.parse(journal.proof) as {
version: number;
filePath: string;
size: number;
};
} finally {
db.close();
}
},
{
dependency: join(workspaceRoot, 'node_modules/better-sqlite3'),
file: join(dataDir, 'databases/iptvnator.db'),
id: row.id,
}
);
expect(durableProof).toEqual(
expect.objectContaining({
version: 1,
filePath: row.filePath,
size: readFileSync(row.filePath).length,
})
);
// Library remains usable after recovery, including archive metadata.
app = await restartElectronApp(app, dataDir, {
env: { TZ: VIEWER_TIMEZONE },
});
@@ -439,6 +484,22 @@ test('@downloads @epg @xtream @electron downloads a completed archive into the l
await expect(
app.mainWindow.getByTestId(`download-library-catchup-${row.id}`)
).toBeVisible();
const recovered = await app.mainWindow.evaluate(async () =>
(await window.electron.downloadsGetList()).find(
(entry) => entry.contentType === 'catchup'
)
);
expect(recovered).toEqual(
expect.objectContaining({
id: row.id,
status: 'completed',
filePath: row.filePath,
totalBytes: durableProof.size,
})
);
expect(
readdirSync(folder).filter((name) => name.endsWith('.ts'))
).toHaveLength(1);
} finally {
await closeElectronApp(app);
}
@@ -16,6 +16,7 @@ export {
epgPrograms,
playbackPositions,
downloads,
downloadArchiveFinalizations,
recordings,
appState,
// Types
@@ -132,6 +132,46 @@ describe('archive file promotion', () => {
code: 'ENOENT',
});
});
it.each(['hardlink', 'copy'])(
'records write-ahead identity before %s can publish complete bytes',
async (mode) => {
const { reservation, identity } = await prepare();
if (mode === 'copy')
jest.mocked(link).mockRejectedValueOnce(
Object.assign(new Error('unsupported'), { code: 'ENOTSUP' })
);
let checkpoints = 0;
await finalizeCatchupPartial(
reservation,
identity,
identity.size,
async (expected) => {
checkpoints++;
if (checkpoints === 1) {
expect(expected).toEqual(
expect.objectContaining({
dev: identity.dev,
ino: identity.ino,
})
);
await expect(
lstat(reservation.path)
).rejects.toMatchObject({ code: 'ENOENT' });
} else {
const created = await lstat(reservation.path);
expect(created).toEqual(
expect.objectContaining({
dev: expected.dev,
ino: expected.ino,
size: 0,
})
);
}
}
);
expect(checkpoints).toBe(mode === 'copy' ? 2 : 1);
}
);
it('never overwrites an occupied final destination', async () => {
const { reservation, identity } = await prepare();
await writeFile(reservation.path, 'keep me');
@@ -26,7 +26,8 @@ function verify(
export async function finalizeCatchupPartial(
reservation: ReservedPartialDownloadFile,
identity: ArchiveFileIdentity | undefined,
size: number
size: number,
recordProof?: (identity: ArchiveFileIdentity) => Promise<void>
): Promise<{ size: number; identity: ArchiveFileIdentity }> {
if (!identity) throw new Error('Archive transfer identity is unavailable');
verify(await lstat(reservation.partialPath), identity, size);
@@ -37,6 +38,9 @@ export async function finalizeCatchupPartial(
let created: ArchiveFileIdentity | undefined;
try {
verify(await source.stat(), identity, size);
// A hardlink can become complete immediately: persist its expected
// identity before publishing it, while the verified partial still exists.
await recordProof?.(identity);
try {
await link(reservation.partialPath, reservation.path);
// The link we created belongs to the verified source, even if
@@ -63,6 +67,9 @@ export async function finalizeCatchupPartial(
const target = await open(reservation.path, 'wx', 0o600);
try {
created = await target.stat();
// For a copy, record the exclusively created target identity
// before the first byte, so a complete file never lacks proof.
await recordProof?.(created);
const buffer = Buffer.alloc(64 * 1024);
let position = 0;
while (position < size) {
@@ -0,0 +1,109 @@
import { inArray } from 'drizzle-orm';
import { lstatSync } from 'node:fs';
import { isAbsolute } from 'node:path';
import * as schema from '../../database/schema';
import type { ArchiveFileIdentity } from './download-catchup-output';
import type { DownloadsDatabase } from './download-task';
export interface ArchiveFinalizationProof {
version: 1;
filePath: string;
size: number;
partialIdentity: ArchiveFileIdentity;
finalIdentity: ArchiveFileIdentity;
}
export async function recordArchiveFinalization(
db: DownloadsDatabase,
downloadId: number,
proof: ArchiveFinalizationProof
): Promise<void> {
const serialized = JSON.stringify({
...proof,
partialIdentity: {
dev: proof.partialIdentity.dev,
ino: proof.partialIdentity.ino,
},
finalIdentity: {
dev: proof.finalIdentity.dev,
ino: proof.finalIdentity.ino,
},
});
await db
.insert(schema.downloadArchiveFinalizations)
.values({ downloadId, proof: serialized })
.onConflictDoUpdate({
target: schema.downloadArchiveFinalizations.downloadId,
set: { proof: serialized },
});
}
function identity(value: unknown): value is ArchiveFileIdentity {
if (!value || typeof value !== 'object') return false;
const candidate = value as ArchiveFileIdentity;
return (
Number.isSafeInteger(candidate.dev) &&
Number.isSafeInteger(candidate.ino)
);
}
export function parseArchiveFinalization(
value: string
): ArchiveFinalizationProof | undefined {
try {
const proof = JSON.parse(value) as ArchiveFinalizationProof;
return proof?.version === 1 &&
typeof proof.filePath === 'string' &&
isAbsolute(proof.filePath) &&
Number.isSafeInteger(proof.size) &&
proof.size > 0 &&
identity(proof.partialIdentity) &&
identity(proof.finalIdentity)
? proof
: undefined;
} catch {
return undefined;
}
}
export async function readArchiveFinalizations(
db: DownloadsDatabase,
ids: number[]
): Promise<Map<number, ArchiveFinalizationProof>> {
if (ids.length === 0) return new Map();
const result = new Map<number, ArchiveFinalizationProof>();
for (let offset = 0; offset < ids.length; offset += 500) {
const rows = await db
.select()
.from(schema.downloadArchiveFinalizations)
.where(
inArray(
schema.downloadArchiveFinalizations.downloadId,
ids.slice(offset, offset + 500)
)
);
for (const row of rows) {
const proof = parseArchiveFinalization(row.proof);
if (proof) result.set(row.downloadId, proof);
}
}
return result;
}
export function verifiedArchiveSize(
filePath: string | null,
proof: ArchiveFinalizationProof | undefined
): number | null {
if (!proof || proof.filePath !== filePath) return null;
try {
const file = lstatSync(proof.filePath);
return file.isFile() &&
file.dev === proof.finalIdentity.dev &&
file.ino === proof.finalIdentity.ino &&
file.size === proof.size
? proof.size
: null;
} catch {
return null;
}
}
@@ -0,0 +1,233 @@
import {
lstat,
mkdtemp,
readFile,
rename,
rm,
writeFile,
link,
} from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { getDatabase } from '../../database/connection';
import * as schema from '../../database/schema';
import { finalizeCatchupPartial } from './download-catchup-finalize';
import {
recordArchiveFinalization,
parseArchiveFinalization,
} from './download-catchup-journal';
import { resetStaleDownloads } from './download-recovery';
import type { DownloadsDatabase } from './download-task';
jest.mock('../../database/connection', () => ({ getDatabase: jest.fn() }));
jest.mock('node:fs/promises', () => {
const actual = jest.requireActual('node:fs/promises');
return { ...actual, link: jest.fn(actual.link) };
});
let directory: string;
let filePath: string;
let rows: Record<string, unknown>[];
let journals: { downloadId: number; proof: string }[];
let updates: Record<string, unknown>[];
let db: DownloadsDatabase;
beforeEach(async () => {
directory = await mkdtemp(join(tmpdir(), 'archive-recovery-'));
filePath = join(directory, 'show.ts');
rows = [
{
id: 1,
filePath,
contentType: 'catchup',
status: 'downloading',
totalBytes: null,
},
];
journals = [];
updates = [];
db = {
select: () => ({
from: (table: unknown) => ({
where: async () =>
table === schema.downloads ? rows : journals,
}),
}),
insert: () => ({
values: (row: (typeof journals)[number]) => ({
onConflictDoUpdate: async () => {
journals = [JSON.parse(JSON.stringify(row))];
},
}),
}),
update: () => ({
set: (value: Record<string, unknown>) => ({
where: async () => {
updates.push(value);
},
}),
}),
} as unknown as DownloadsDatabase;
jest.mocked(getDatabase).mockResolvedValue(db);
jest.mocked(link)
.mockReset()
.mockImplementation(jest.requireActual('node:fs/promises').link);
});
afterEach(async () => {
await rm(directory, { recursive: true, force: true });
});
async function prepare() {
await writeFile(filePath + '.part', 'verified archive');
const partial = await lstat(filePath + '.part');
return {
partial,
reservation: {
path: filePath,
partialPath: filePath + '.part',
filename: 'show.ts',
},
};
}
it.each(['hardlink', 'copy'])(
'recovers an unknown-length archive from the durable %s journal after restart',
async (mode) => {
const { partial, reservation } = await prepare();
if (mode === 'copy')
jest.mocked(link).mockRejectedValueOnce(
Object.assign(new Error('unsupported'), { code: 'ENOTSUP' })
);
await finalizeCatchupPartial(
reservation,
partial,
partial.size,
(finalIdentity) =>
recordArchiveFinalization(db, 1, {
version: 1,
filePath,
size: partial.size,
partialIdentity: partial,
finalIdentity,
})
);
// No in-memory DownloadTask or completion DB update survives this restart.
await resetStaleDownloads();
expect(updates).toContainEqual(
expect.objectContaining({
status: 'completed',
bytesDownloaded: partial.size,
totalBytes: partial.size,
})
);
expect(await readFile(filePath, 'utf8')).toBe('verified archive');
}
);
it('pauses verified bytes when termination precedes promotion', async () => {
const { partial, reservation } = await prepare();
await expect(
finalizeCatchupPartial(
reservation,
partial,
partial.size,
async (finalIdentity) => {
await recordArchiveFinalization(db, 1, {
version: 1,
filePath,
size: partial.size,
partialIdentity: partial,
finalIdentity,
});
throw new Error('simulated termination');
}
)
).rejects.toThrow('termination');
await resetStaleDownloads();
expect(updates).toContainEqual(
expect.objectContaining({
status: 'paused',
bytesDownloaded: partial.size,
})
);
expect(await readFile(reservation.partialPath, 'utf8')).toBe(
'verified archive'
);
});
it('refuses same-size replacements and preserves unrelated partials during startup cleanup', async () => {
const { partial, reservation } = await prepare();
await finalizeCatchupPartial(
reservation,
partial,
partial.size,
(finalIdentity) =>
recordArchiveFinalization(db, 1, {
version: 1,
filePath,
size: partial.size,
partialIdentity: partial,
finalIdentity,
})
);
await rename(filePath, join(directory, 'original'));
await writeFile(filePath, 'untrusted bytes!');
await writeFile(filePath + '.part', 'leave this alone');
await resetStaleDownloads();
expect(updates.some((value) => value.status === 'completed')).toBe(false);
expect(await readFile(filePath + '.part', 'utf8')).toBe('leave this alone');
expect(await readFile(filePath, 'utf8')).toBe('untrusted bytes!');
});
it('removes an owned incomplete copy before pausing its retained source', async () => {
const { partial, reservation } = await prepare();
await writeFile(filePath, 'half');
await recordArchiveFinalization(db, 1, {
version: 1,
filePath,
size: partial.size,
partialIdentity: partial,
finalIdentity: await lstat(filePath),
});
await resetStaleDownloads();
expect(updates).toContainEqual(
expect.objectContaining({
status: 'paused',
bytesDownloaded: partial.size,
})
);
await expect(lstat(filePath)).rejects.toMatchObject({ code: 'ENOENT' });
expect(await readFile(reservation.partialPath, 'utf8')).toBe(
'verified archive'
);
});
it.each([
'{}',
'null',
'bad json',
JSON.stringify({
version: 1,
filePath: '/tmp/a',
size: -1,
partialIdentity: { dev: 1, ino: 1 },
finalIdentity: { dev: 1, ino: 1 },
}),
])('ignores malformed journal %s', (value) => {
expect(parseArchiveFinalization(value)).toBeUndefined();
});
it('does not remove a valid journaled final when a retry was queued at shutdown', async () => {
const { partial, reservation } = await prepare();
await finalizeCatchupPartial(
reservation,
partial,
partial.size,
(finalIdentity) =>
recordArchiveFinalization(db, 1, {
version: 1,
filePath,
size: partial.size,
partialIdentity: partial,
finalIdentity,
})
);
rows[0].status = 'queued';
await resetStaleDownloads();
expect(await readFile(filePath, 'utf8')).toBe('verified archive');
expect(updates).toContainEqual(
expect.objectContaining({ status: 'paused' })
);
});
@@ -13,6 +13,9 @@ import { transferCatchupToPartialFile } from './download-catchup-transfer';
import { enqueueDownload } from './download-runtime';
import type { DownloadTask } from './download-task';
jest.mock('./download-catchup-journal', () => ({
recordArchiveFinalization: jest.fn().mockResolvedValue(undefined),
}));
jest.mock('../../database/connection', () => ({ getDatabase: jest.fn() }));
jest.mock('./download-catchup-transfer', () => ({
transferCatchupToPartialFile: jest.fn(),
@@ -1,3 +1,4 @@
import { recordArchiveFinalization } from './download-catchup-journal';
import { finalizePartialDownload } from './download-file-finalize';
import { cleanupCatchupPartial } from './download-catchup-cleanup';
import {
@@ -120,10 +121,21 @@ export async function completeDownloadFromPartial(
let fileSize: number;
try {
if (task.catchup) {
const partialIdentity = task.catchupPartialIdentity;
if (!partialIdentity)
throw new Error('Archive transfer identity is unavailable');
const finalized = await finalizeCatchupPartial(
reservation,
task.catchupPartialIdentity,
progress.bytesDownloaded
progress.bytesDownloaded,
(finalIdentity) =>
recordArchiveFinalization(db, task.id, {
version: 1,
filePath: reservation.path,
size: progress.bytesDownloaded,
partialIdentity,
finalIdentity,
})
);
task.catchupFinalized = {
...finalized,
@@ -1,3 +1,12 @@
import {
cleanupCatchupFile,
cleanupCatchupPartial,
} from './download-catchup-cleanup';
import {
readArchiveFinalizations,
verifiedArchiveSize,
type ArchiveFinalizationProof,
} from './download-catchup-journal';
import { inArray, sql } from 'drizzle-orm';
import { statSync } from 'node:fs';
import { getDatabase } from '../../database/connection';
@@ -8,6 +17,8 @@ import {
} from './download-file-path';
interface StaleDownload {
contentType?: string;
proof?: ArchiveFinalizationProof;
filePath: string | null;
id: number;
status: string;
@@ -40,6 +51,10 @@ function getRecoverablePartialSize(download: StaleDownload): number {
* commits the completion instead of orphaning the file and re-downloading.
*/
function getFinalizedFileSize(download: StaleDownload): number | null {
if (download.contentType === 'catchup')
return download.status === 'downloading'
? verifiedArchiveSize(download.filePath, download.proof)
: null;
if (
download.status !== 'downloading' ||
!download.filePath ||
@@ -58,7 +73,12 @@ function getFinalizedFileSize(download: StaleDownload): number | null {
}
}
function removeFailedPartial(download: StaleDownload): boolean {
async function removeFailedPartial(download: StaleDownload): Promise<boolean> {
if (download.contentType === 'catchup')
return cleanupCatchupPartial(
download.filePath,
download.proof?.partialIdentity
);
if (!download.filePath) {
return true;
}
@@ -76,7 +96,14 @@ function removeFailedPartial(download: StaleDownload): boolean {
}
}
function removeCompletedPartial(download: StaleDownload): void {
async function removeCompletedPartial(download: StaleDownload): Promise<void> {
if (download.contentType === 'catchup') {
await cleanupCatchupPartial(
download.filePath,
download.proof?.partialIdentity
);
return;
}
if (!download.filePath) {
return;
}
@@ -95,9 +122,10 @@ function removeCompletedPartial(download: StaleDownload): void {
export async function resetStaleDownloads(): Promise<void> {
try {
const db = await getDatabase();
const downloads = await db
const rows = await db
.select({
filePath: schema.downloads.filePath,
contentType: schema.downloads.contentType,
id: schema.downloads.id,
status: schema.downloads.status,
totalBytes: schema.downloads.totalBytes,
@@ -110,6 +138,19 @@ export async function resetStaleDownloads(): Promise<void> {
'completed',
])
);
const proofs = await readArchiveFinalizations(
db,
rows
.filter((row) => row.contentType === 'catchup')
.map((row) => row.id)
);
const downloads = rows.map((row) => {
const proof = proofs.get(row.id);
return {
...row,
proof: proof?.filePath === row.filePath ? proof : undefined,
};
});
const completedDownloads = downloads.filter(
(download) => download.status === 'completed'
);
@@ -125,6 +166,21 @@ export async function resetStaleDownloads(): Promise<void> {
const finalizedIds = new Set(
finalizedDownloads.map((download) => download.id)
);
// A killed copy may have left an incomplete owned destination. Remove
// only that journal-bound entry before resuming the retained source.
for (const download of staleDownloads) {
if (
download.contentType === 'catchup' &&
download.proof &&
!finalizedIds.has(download.id) &&
verifiedArchiveSize(download.filePath, download.proof) === null
) {
await cleanupCatchupFile(
download.proof.filePath,
download.proof.finalIdentity
).catch(() => undefined);
}
}
// Queued rows are recoverable even without partial bytes: a resumed
// download waiting behind an active one is persisted as 'queued' with
// its retained .part, and a never-started queued row loses nothing by
@@ -147,10 +203,12 @@ export async function resetStaleDownloads(): Promise<void> {
!recoverableIds.has(download.id) &&
!finalizedIds.has(download.id)
);
const cleanupResult = failedDownloads.map((download) => ({
...download,
partialRemoved: removeFailedPartial(download),
}));
const cleanupResult = await Promise.all(
failedDownloads.map(async (download) => ({
...download,
partialRemoved: await removeFailedPartial(download),
}))
);
const failedIdsWithRemovedPartials = cleanupResult
.filter((download) => download.partialRemoved)
.map((download) => download.id);
@@ -158,15 +216,16 @@ export async function resetStaleDownloads(): Promise<void> {
.filter((download) => !download.partialRemoved)
.map((download) => download.id);
completedDownloads.forEach(removeCompletedPartial);
await Promise.all(completedDownloads.map(removeCompletedPartial));
for (const download of finalizedDownloads) {
// The interrupted commit may also have left the .part behind.
removeCompletedPartial(download);
await removeCompletedPartial(download);
await db
.update(schema.downloads)
.set({
bytesDownloaded: download.finalizedSize,
totalBytes: download.finalizedSize,
errorMessage: null,
status: 'completed',
updatedAt: sql`CURRENT_TIMESTAMP`,
@@ -196,7 +255,9 @@ export async function resetStaleDownloads(): Promise<void> {
status: 'failed',
updatedAt: sql`CURRENT_TIMESTAMP`,
})
.where(inArray(schema.downloads.id, failedIdsWithRemovedPartials));
.where(
inArray(schema.downloads.id, failedIdsWithRemovedPartials)
);
}
if (failedIdsWithRetainedPartials.length > 0) {
@@ -207,7 +268,9 @@ export async function resetStaleDownloads(): Promise<void> {
status: 'failed',
updatedAt: sql`CURRENT_TIMESTAMP`,
})
.where(inArray(schema.downloads.id, failedIdsWithRetainedPartials));
.where(
inArray(schema.downloads.id, failedIdsWithRetainedPartials)
);
}
console.log('[Downloads] Reset stale downloads');