fix(downloads): preserve archive recovery ownership and interruption

This commit is contained in:
4gray committed 2026-09-08 01:39:01 +02:00
1 parent 77a70af8e9
commit 414b15d5c8
6 files changed
+118 -6

No files matched your search

@@ -8,6 +8,7 @@ import {
writeFile,
} from 'node:fs/promises';
import { join } from 'node:path';
import { existsSync, statSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { finalizeCatchupPartial } from './download-catchup-finalize';
@@ -180,4 +181,37 @@ describe('archive file promotion', () => {
).rejects.toMatchObject({ code: 'EEXIST' });
expect(await readFile(reservation.path, 'utf8')).toBe('keep me');
});
it.each(['pause', 'cancel'])(
'stops an in-progress fallback copy on %s',
async () => {
const { reservation } = await prepare();
const bytes = Buffer.alloc(192 * 1024, 0x47);
await writeFile(reservation.partialPath, bytes);
const identity = await lstat(reservation.partialPath);
jest.mocked(link).mockRejectedValueOnce(
Object.assign(new Error('unsupported'), { code: 'ENOTSUP' })
);
let observedBytes = 0;
await expect(
finalizeCatchupPartial(
reservation,
identity,
bytes.length,
undefined,
() => {
observedBytes = existsSync(reservation.path)
? statSync(reservation.path).size
: 0;
return observedBytes > 0;
}
)
).rejects.toThrow('interrupted');
expect(observedBytes).toBeGreaterThan(0);
expect(observedBytes).toBeLessThan(bytes.length);
expect(await readFile(reservation.partialPath)).toEqual(bytes);
await expect(lstat(reservation.path)).rejects.toMatchObject({
code: 'ENOENT',
});
}
);
});
@@ -27,8 +27,13 @@ export async function finalizeCatchupPartial(
reservation: ReservedPartialDownloadFile,
identity: ArchiveFileIdentity | undefined,
size: number,
recordProof?: (identity: ArchiveFileIdentity) => Promise<void>
recordProof?: (identity: ArchiveFileIdentity) => Promise<void>,
shouldInterrupt: () => boolean = () => false
): Promise<{ size: number; identity: ArchiveFileIdentity }> {
const checkInterruption = () => {
if (shouldInterrupt()) throw new Error('Archive promotion interrupted');
};
checkInterruption();
if (!identity) throw new Error('Archive transfer identity is unavailable');
verify(await lstat(reservation.partialPath), identity, size);
const source = await open(
@@ -41,6 +46,7 @@ export async function finalizeCatchupPartial(
// A hardlink can become complete immediately: persist its expected
// identity before publishing it, while the verified partial still exists.
await recordProof?.(identity);
checkInterruption();
try {
await link(reservation.partialPath, reservation.path);
// The link we created belongs to the verified source, even if
@@ -73,6 +79,7 @@ export async function finalizeCatchupPartial(
const buffer = Buffer.alloc(64 * 1024);
let position = 0;
while (position < size) {
checkInterruption();
const { bytesRead } = await source.read(
buffer,
0,
@@ -85,6 +92,7 @@ export async function finalizeCatchupPartial(
);
let written = 0;
while (written < bytesRead) {
checkInterruption();
const { bytesWritten } = await target.write(
buffer,
written,
@@ -103,7 +111,9 @@ export async function finalizeCatchupPartial(
await target.close();
}
}
checkInterruption();
verify(await lstat(reservation.path), created, size);
checkInterruption();
await cleanupCatchupFile(reservation.partialPath, identity).catch(
() => undefined
);
@@ -231,3 +231,31 @@ it('does not remove a valid journaled final when a retry was queued at shutdown'
expect.objectContaining({ status: 'paused' })
);
});
it.each(['downloading', 'queued'])(
'detaches a replaced journaled partial instead of making %s recoverable',
async (status) => {
const { partial } = await prepare();
await recordArchiveFinalization(db, 1, {
version: 1,
filePath,
size: partial.size,
partialIdentity: partial,
finalIdentity: partial,
});
await rename(filePath + '.part', join(directory, 'original'));
await writeFile(filePath + '.part', 'unrelated user file');
rows[0].status = status;
await resetStaleDownloads();
expect(updates).toContainEqual(
expect.objectContaining({
status: 'failed',
filePath: null,
})
);
expect(updates.some((value) => value.status === 'paused')).toBe(false);
expect(await readFile(filePath + '.part', 'utf8')).toBe(
'unrelated user file'
);
}
);
@@ -137,7 +137,8 @@ export async function completeDownloadFromPartial(
size: progress.bytesDownloaded,
partialIdentity,
finalIdentity,
})
}),
() => !!(task.cancelRequested || task.pauseRequested)
);
task.catchupFinalized = {
...finalized,
@@ -8,7 +8,7 @@ import {
type ArchiveFinalizationProof,
} from './download-catchup-journal';
import { inArray, sql } from 'drizzle-orm';
import { statSync } from 'node:fs';
import { lstatSync, statSync } from 'node:fs';
import { getDatabase } from '../../database/connection';
import * as schema from '../../database/schema';
import {
@@ -25,6 +25,27 @@ interface StaleDownload {
totalBytes: number | null;
}
/** A journaled source cannot be adopted again after an unrelated replacement. */
function hasReplacedArchivePartial(download: StaleDownload): boolean {
if (
download.contentType !== 'catchup' ||
!download.proof ||
!download.filePath
)
return false;
try {
const file = lstatSync(`${download.filePath}.part`);
const expected = download.proof.partialIdentity;
return (
!file.isFile() ||
file.dev !== expected.dev ||
file.ino !== expected.ino
);
} catch (error) {
return (error as NodeJS.ErrnoException).code !== 'ENOENT';
}
}
function getRecoverablePartialSize(download: StaleDownload): number {
if (
(download.status !== 'downloading' && download.status !== 'queued') ||
@@ -193,7 +214,9 @@ export async function resetStaleDownloads(): Promise<void> {
}))
.filter(
(download) =>
download.status === 'queued' || download.bytesDownloaded > 0
!hasReplacedArchivePartial(download) &&
(download.status === 'queued' ||
download.bytesDownloaded > 0)
);
const recoverableIds = new Set(
recoverableDownloads.map((download) => download.id)
@@ -206,7 +229,11 @@ export async function resetStaleDownloads(): Promise<void> {
const cleanupResult = await Promise.all(
failedDownloads.map(async (download) => ({
...download,
partialRemoved: await removeFailedPartial(download),
// Detach a known replacement without touching it. A later retry
// reserves another path instead of truncating the unrelated file.
partialRemoved:
hasReplacedArchivePartial(download) ||
(await removeFailedPartial(download)),
}))
);
const failedIdsWithRemovedPartials = cleanupResult
+13 -1
View File
@@ -63,10 +63,22 @@ unknown-length file has durable proof before the completion-status write. Startu
requires that proof and a matching regular file, identity and size to recover an
archive; termination before promotion leaves the verified partial paused. An
owned incomplete copy is removed by journal identity before the source resumes. The
journal also makes startup partial cleanup identity-aware and remains with
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
clears the previous attempt's proof before network requests or partial-file writes. 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
leaves the source for the runtime pause/cancel handler.
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
numbered free destination. SQLite and filesystem creation cannot commit
atomically, and portable rename cannot guarantee no-clobber publication on the
filesystems that need this fallback. This bounded orphan is preferred to deleting
or overwriting an unrelated file.
An explicit cancellation of a queued/paused archive captures the selected regular
partial using the same cleanup helper; symlink entries are preserved.
A retained archive cannot use the VOD byte-count completion shortcut. Transfers