Sucukdeluxe
67978cb81f
Separate provisional upload success from persistently confirmed cleanup state, invalidate prior confirmations before retries, and promote completion only inside the final queue persistence handshake. Preserve source files when history or queue persistence fails and add regression coverage for restart, retry, rollback, and real Electron finalization paths.
476 lines
17 KiB
JavaScript
476 lines
17 KiB
JavaScript
const RETRY_DELAYS = [100, 250, 500, 1000, 2000];
|
|
const TERMINAL_STATUSES = new Set(['done', 'error', 'aborted', 'skipped', 'failed', 'missing-account']);
|
|
const crypto = require('node:crypto');
|
|
|
|
function createSourceFileCleanup(options) {
|
|
if (!options || !options.fs || !options.path) {
|
|
throw new TypeError('createSourceFileCleanup requires fs and path');
|
|
}
|
|
|
|
const fs = options.fs;
|
|
const path = options.path;
|
|
const platform = options.platform || process.platform;
|
|
const isEnabled = typeof options.isEnabled === 'function' ? options.isEnabled : () => false;
|
|
const audit = typeof options.audit === 'function' ? options.audit : () => {};
|
|
const journal = options.journal || null;
|
|
const wait = typeof options.wait === 'function'
|
|
? options.wait
|
|
: (milliseconds) => new Promise((resolve) => setTimeout(resolve, milliseconds));
|
|
const groups = new Map();
|
|
const leases = new Map();
|
|
|
|
function canonicalize(file) {
|
|
const resolved = path.resolve(file);
|
|
return platform === 'win32' ? resolved.toLowerCase() : resolved;
|
|
}
|
|
|
|
function cloneFingerprint(fingerprint) {
|
|
if (!fingerprint) return null;
|
|
return {
|
|
type: 'file',
|
|
size: fingerprint.size,
|
|
mtimeMs: fingerprint.mtimeMs,
|
|
birthtimeMs: fingerprint.birthtimeMs,
|
|
dev: fingerprint.dev,
|
|
ino: fingerprint.ino
|
|
};
|
|
}
|
|
|
|
function isFingerprint(fingerprint) {
|
|
return Boolean(
|
|
fingerprint &&
|
|
fingerprint.type === 'file' &&
|
|
Number.isFinite(fingerprint.size) &&
|
|
Number.isFinite(fingerprint.mtimeMs) &&
|
|
Number.isFinite(fingerprint.birthtimeMs) &&
|
|
Number.isFinite(fingerprint.dev) &&
|
|
Number.isFinite(fingerprint.ino)
|
|
);
|
|
}
|
|
|
|
function fingerprintFromStat(stat) {
|
|
if (!stat.isFile()) return null;
|
|
return {
|
|
type: 'file',
|
|
size: stat.size,
|
|
mtimeMs: stat.mtimeMs,
|
|
birthtimeMs: stat.birthtimeMs,
|
|
dev: stat.dev,
|
|
ino: stat.ino
|
|
};
|
|
}
|
|
|
|
function fingerprintsMatch(left, right) {
|
|
return Boolean(
|
|
left &&
|
|
right &&
|
|
left.type === right.type &&
|
|
left.size === right.size &&
|
|
left.mtimeMs === right.mtimeMs &&
|
|
left.birthtimeMs === right.birthtimeMs &&
|
|
left.dev === right.dev &&
|
|
left.ino === right.ino
|
|
);
|
|
}
|
|
|
|
function uniqueStrings(values) {
|
|
return [...new Set((Array.isArray(values) ? values : []).filter((value) => typeof value === 'string' && value.length > 0))];
|
|
}
|
|
|
|
function normalizeStatus(status) {
|
|
return TERMINAL_STATUSES.has(status) ? status : 'pending';
|
|
}
|
|
|
|
function createManifest(input, canonicalFile) {
|
|
const requiredHosters = uniqueStrings(input.requiredHosters);
|
|
const confirmedHosters = new Set(
|
|
uniqueStrings(input.confirmedHosters).filter((hoster) => requiredHosters.includes(hoster))
|
|
);
|
|
const jobs = new Map();
|
|
|
|
for (const job of Array.isArray(input.jobs) ? input.jobs : []) {
|
|
if (!job || typeof job.jobId !== 'string' || typeof job.hoster !== 'string') continue;
|
|
const currentRound = job.currentRound !== false;
|
|
const status = currentRound ? 'pending' : normalizeStatus(job.status);
|
|
jobs.set(job.jobId, Object.freeze({
|
|
jobId: job.jobId,
|
|
hoster: job.hoster,
|
|
status,
|
|
currentRound
|
|
}));
|
|
}
|
|
|
|
return {
|
|
token: input.token || input.sourceCleanupToken,
|
|
file: path.resolve(input.file),
|
|
canonicalFile,
|
|
requiredHosters: Object.freeze(requiredHosters),
|
|
confirmedHosters,
|
|
jobs,
|
|
suppliedFingerprint: isFingerprint(input.fingerprint) ? cloneFingerprint(input.fingerprint) : null,
|
|
fingerprint: null,
|
|
registrationOutcome: null,
|
|
registrationError: null,
|
|
audited: false,
|
|
finalizationPromise: null
|
|
};
|
|
}
|
|
|
|
async function inspectRegistration(manifest) {
|
|
try {
|
|
const stat = await fs.promises.lstat(manifest.file);
|
|
const currentFingerprint = fingerprintFromStat(stat);
|
|
if (!currentFingerprint) {
|
|
manifest.registrationOutcome = 'unsafe-source-type';
|
|
return;
|
|
}
|
|
if (manifest.suppliedFingerprint && !fingerprintsMatch(manifest.suppliedFingerprint, currentFingerprint)) {
|
|
manifest.fingerprint = manifest.suppliedFingerprint;
|
|
manifest.registrationOutcome = 'source-changed';
|
|
return;
|
|
}
|
|
manifest.fingerprint = manifest.suppliedFingerprint || currentFingerprint;
|
|
} catch (error) {
|
|
manifest.registrationOutcome = error && error.code === 'ENOENT' ? 'source-missing' : 'failed';
|
|
manifest.registrationError = error;
|
|
}
|
|
}
|
|
|
|
async function registerGroups(inputGroups) {
|
|
const fingerprints = {};
|
|
for (const input of Array.isArray(inputGroups) ? inputGroups : []) {
|
|
const token = input && (input.token || input.sourceCleanupToken);
|
|
if (typeof token !== 'string' || token.length === 0 || typeof input.file !== 'string' || input.file.length === 0) {
|
|
throw new TypeError('source cleanup groups require token and file');
|
|
}
|
|
const existing = groups.get(token);
|
|
if (existing) {
|
|
if (existing.finalizationPromise) throw new Error('source cleanup group is already finalizing');
|
|
if (canonicalize(input.file) !== existing.canonicalFile) throw new Error('source cleanup token changed file');
|
|
existing.requiredHosters = Object.freeze(uniqueStrings([
|
|
...existing.requiredHosters,
|
|
...input.requiredHosters
|
|
]));
|
|
const incomingJobs = Array.isArray(input.jobs) ? input.jobs : [];
|
|
const currentRoundHosters = new Set(
|
|
incomingJobs
|
|
.filter((job) => job && typeof job.hoster === 'string' && job.currentRound !== false)
|
|
.map((job) => job.hoster)
|
|
);
|
|
for (const hoster of currentRoundHosters) existing.confirmedHosters.delete(hoster);
|
|
for (const hoster of uniqueStrings(input.confirmedHosters)) {
|
|
if (existing.requiredHosters.includes(hoster) && !currentRoundHosters.has(hoster)) {
|
|
existing.confirmedHosters.add(hoster);
|
|
}
|
|
}
|
|
for (const job of incomingJobs) {
|
|
if (!job || typeof job.jobId !== 'string' || typeof job.hoster !== 'string') continue;
|
|
const previous = existing.jobs.get(job.jobId);
|
|
const incomingCurrentRound = job.currentRound !== false;
|
|
const currentRound = Boolean((previous && previous.currentRound) || incomingCurrentRound);
|
|
const status = incomingCurrentRound
|
|
? 'pending'
|
|
: normalizeStatus(previous ? previous.status : job.status);
|
|
existing.jobs.set(job.jobId, Object.freeze({
|
|
jobId: job.jobId,
|
|
hoster: job.hoster,
|
|
status,
|
|
currentRound
|
|
}));
|
|
}
|
|
fingerprints[token] = cloneFingerprint(existing.fingerprint);
|
|
continue;
|
|
}
|
|
|
|
const canonicalFile = canonicalize(input.file);
|
|
const manifest = createManifest(input, canonicalFile);
|
|
const leaseToken = leases.get(canonicalFile);
|
|
if (leaseToken && leaseToken !== token) {
|
|
manifest.registrationOutcome = 'blocked';
|
|
manifest.registrationError = new Error('active-source-lease');
|
|
const leaseOwner = groups.get(leaseToken);
|
|
if (leaseOwner) {
|
|
leaseOwner.registrationOutcome = 'blocked';
|
|
leaseOwner.registrationError = new Error('active-source-lease');
|
|
}
|
|
} else {
|
|
leases.set(canonicalFile, token);
|
|
await inspectRegistration(manifest);
|
|
}
|
|
groups.set(token, manifest);
|
|
fingerprints[token] = cloneFingerprint(manifest.fingerprint);
|
|
}
|
|
return fingerprints;
|
|
}
|
|
|
|
function settle(event) {
|
|
if (!event || !TERMINAL_STATUSES.has(event.status)) return false;
|
|
const token = event.token || event.sourceCleanupToken;
|
|
const manifest = groups.get(token);
|
|
if (!manifest || manifest.finalizationPromise) return false;
|
|
const job = manifest.jobs.get(event.jobId);
|
|
if (!job || !job.currentRound || job.hoster !== event.hoster) return false;
|
|
if (typeof event.file === 'string' && canonicalize(event.file) !== manifest.canonicalFile) return false;
|
|
manifest.jobs.set(event.jobId, Object.freeze({ ...job, status: event.status }));
|
|
return true;
|
|
}
|
|
|
|
function markSkipped(jobId) {
|
|
let changed = false;
|
|
for (const manifest of groups.values()) {
|
|
if (manifest.finalizationPromise) continue;
|
|
const job = manifest.jobs.get(jobId);
|
|
if (!job || !job.currentRound) continue;
|
|
manifest.jobs.set(jobId, Object.freeze({ ...job, status: 'skipped' }));
|
|
changed = true;
|
|
}
|
|
return changed;
|
|
}
|
|
|
|
function blockingStatuses(manifest) {
|
|
const blocking = [];
|
|
for (const hoster of manifest.requiredHosters) {
|
|
const jobs = [...manifest.jobs.values()].filter((job) => job.hoster === hoster);
|
|
const currentJobs = jobs.filter((job) => job.currentRound);
|
|
if (currentJobs.length > 0) {
|
|
const currentBlocker = currentJobs.find((job) => job.status !== 'done');
|
|
if (!currentBlocker) continue;
|
|
const status = currentJobs.map((job) => job.status).find((value) => value !== 'pending') || 'pending';
|
|
blocking.push({ hoster, status });
|
|
continue;
|
|
}
|
|
if (manifest.confirmedHosters.has(hoster)) continue;
|
|
const statuses = jobs.map((job) => job.status).filter((status) => status !== 'done');
|
|
const status = statuses.find((value) => value !== 'pending') || 'pending';
|
|
blocking.push({ hoster, status });
|
|
}
|
|
return blocking;
|
|
}
|
|
|
|
async function emitAudit(manifest, outcome, details = {}) {
|
|
if (manifest.audited) return;
|
|
manifest.audited = true;
|
|
await writeAudit(manifest, outcome, details);
|
|
}
|
|
|
|
async function writeAudit(manifest, outcome, details = {}) {
|
|
const event = {
|
|
timestamp: new Date().toISOString(),
|
|
outcome,
|
|
file: manifest.file,
|
|
hosters: [...manifest.requiredHosters],
|
|
...details
|
|
};
|
|
if (event.error instanceof Error) event.error = event.error.message;
|
|
try {
|
|
return (await audit(event)) !== false;
|
|
} catch {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
async function currentSourceOutcome(manifest) {
|
|
try {
|
|
const stat = await fs.promises.lstat(manifest.file);
|
|
const currentFingerprint = fingerprintFromStat(stat);
|
|
if (!currentFingerprint) return { outcome: 'unsafe-source-type', trigger: 'unsafe-source-type' };
|
|
if (!fingerprintsMatch(manifest.fingerprint, currentFingerprint)) {
|
|
return { outcome: 'source-changed', trigger: 'source-fingerprint-mismatch' };
|
|
}
|
|
return null;
|
|
} catch (error) {
|
|
if (error && error.code === 'ENOENT') return { outcome: 'source-missing', trigger: 'source-missing' };
|
|
return { outcome: 'failed', trigger: 'source-stat-failed', error };
|
|
}
|
|
}
|
|
|
|
async function unlinkWithRetries(file) {
|
|
let attempts = 0;
|
|
while (true) {
|
|
attempts += 1;
|
|
try {
|
|
await fs.promises.unlink(file);
|
|
return { attempts };
|
|
} catch (error) {
|
|
const retryable = platform === 'win32' && error && (error.code === 'EBUSY' || error.code === 'EPERM');
|
|
if (!retryable || attempts > RETRY_DELAYS.length) throw Object.assign(error, { cleanupAttempts: attempts });
|
|
await wait(RETRY_DELAYS[attempts - 1]);
|
|
}
|
|
}
|
|
}
|
|
|
|
async function restoreStagedFile(manifest, stagedFile) {
|
|
try {
|
|
await fs.promises.rename(stagedFile, manifest.file);
|
|
return null;
|
|
} catch (error) {
|
|
return error;
|
|
}
|
|
}
|
|
|
|
async function clearJournal(manifest) {
|
|
if (!journal || typeof journal.clear !== 'function') return;
|
|
try { await journal.clear(manifest.token); } catch {}
|
|
}
|
|
|
|
async function stageSource(manifest) {
|
|
const stagedFile = path.join(
|
|
path.dirname(manifest.file),
|
|
`.${path.basename(manifest.file)}.mhu-delete-${crypto.randomUUID()}`
|
|
);
|
|
if (!journal || typeof journal.plan !== 'function') {
|
|
return { outcome: 'blocked', trigger: 'delete-journal-unavailable' };
|
|
}
|
|
try {
|
|
await journal.plan({ token: manifest.token, file: manifest.file, stagedFile });
|
|
} catch (error) {
|
|
return { outcome: 'blocked', trigger: 'delete-journal-write-failed', error };
|
|
}
|
|
try {
|
|
await fs.promises.rename(manifest.file, stagedFile);
|
|
} catch (error) {
|
|
await clearJournal(manifest);
|
|
return {
|
|
outcome: error && error.code === 'ENOENT' ? 'source-missing' : 'failed',
|
|
trigger: error && error.code === 'ENOENT' ? 'source-missing' : 'source-stage-failed',
|
|
error
|
|
};
|
|
}
|
|
try {
|
|
const stat = await fs.promises.lstat(stagedFile);
|
|
const fingerprint = fingerprintFromStat(stat);
|
|
if (fingerprintsMatch(manifest.fingerprint, fingerprint)) return { stagedFile };
|
|
const restoreError = await restoreStagedFile(manifest, stagedFile);
|
|
if (!restoreError) await clearJournal(manifest);
|
|
return {
|
|
outcome: 'source-changed',
|
|
trigger: 'source-fingerprint-mismatch-after-stage',
|
|
error: restoreError,
|
|
stagedFile: restoreError ? stagedFile : undefined
|
|
};
|
|
} catch (error) {
|
|
const restoreError = await restoreStagedFile(manifest, stagedFile);
|
|
if (!restoreError) await clearJournal(manifest);
|
|
return {
|
|
outcome: 'failed',
|
|
trigger: 'staged-source-stat-failed',
|
|
error,
|
|
restoreError,
|
|
stagedFile: restoreError ? stagedFile : undefined
|
|
};
|
|
}
|
|
}
|
|
|
|
async function finalize(manifest, barriers) {
|
|
try {
|
|
let enabled;
|
|
try {
|
|
enabled = Boolean(await isEnabled());
|
|
} catch (error) {
|
|
await emitAudit(manifest, 'failed', { trigger: 'setting-check-failed', error });
|
|
return 'failed';
|
|
}
|
|
if (!enabled) {
|
|
await emitAudit(manifest, 'setting-disabled', { trigger: 'setting-disabled' });
|
|
return 'setting-disabled';
|
|
}
|
|
if (!barriers.historyPersisted || !barriers.queuePersisted) {
|
|
await emitAudit(manifest, 'blocked', {
|
|
trigger: 'persistence-barrier-incomplete',
|
|
historyPersisted: Boolean(barriers.historyPersisted),
|
|
queuePersisted: Boolean(barriers.queuePersisted)
|
|
});
|
|
return 'blocked';
|
|
}
|
|
if (manifest.registrationOutcome) {
|
|
await emitAudit(manifest, manifest.registrationOutcome, {
|
|
trigger: manifest.registrationOutcome === 'blocked' ? 'active-source-lease' : manifest.registrationOutcome,
|
|
error: manifest.registrationError
|
|
});
|
|
return manifest.registrationOutcome;
|
|
}
|
|
const blocking = blockingStatuses(manifest);
|
|
if (blocking.length > 0) {
|
|
await emitAudit(manifest, 'blocked', {
|
|
trigger: 'required-hoster-not-done',
|
|
blockingStatuses: blocking
|
|
});
|
|
return 'blocked';
|
|
}
|
|
const sourceOutcome = await currentSourceOutcome(manifest);
|
|
if (sourceOutcome) {
|
|
await emitAudit(manifest, sourceOutcome.outcome, sourceOutcome);
|
|
return sourceOutcome.outcome;
|
|
}
|
|
const auditPersisted = await writeAudit(manifest, 'delete-approved', {
|
|
trigger: 'all-selected-hosters-succeeded'
|
|
});
|
|
if (!auditPersisted) {
|
|
await emitAudit(manifest, 'blocked', { trigger: 'audit-write-failed' });
|
|
return 'blocked';
|
|
}
|
|
const staged = await stageSource(manifest);
|
|
if (!staged.stagedFile || staged.outcome) {
|
|
await emitAudit(manifest, staged.outcome || 'failed', staged);
|
|
return staged.outcome || 'failed';
|
|
}
|
|
const commitPersisted = await writeAudit(manifest, 'source-staged', {
|
|
trigger: 'verified-source-staged'
|
|
});
|
|
if (!commitPersisted) {
|
|
const restoreError = await restoreStagedFile(manifest, staged.stagedFile);
|
|
if (!restoreError) await clearJournal(manifest);
|
|
await emitAudit(manifest, 'blocked', {
|
|
trigger: 'audit-commit-write-failed',
|
|
restoreError,
|
|
stagedFile: restoreError ? staged.stagedFile : undefined
|
|
});
|
|
return 'blocked';
|
|
}
|
|
try {
|
|
const result = await unlinkWithRetries(staged.stagedFile);
|
|
await clearJournal(manifest);
|
|
await emitAudit(manifest, 'deleted', {
|
|
trigger: 'all-selected-hosters-succeeded',
|
|
attempts: result.attempts
|
|
});
|
|
return 'deleted';
|
|
} catch (error) {
|
|
const restoreError = await restoreStagedFile(manifest, staged.stagedFile);
|
|
if (!restoreError) await clearJournal(manifest);
|
|
await emitAudit(manifest, 'failed', {
|
|
trigger: 'unlink-failed',
|
|
attempts: error.cleanupAttempts || 1,
|
|
error,
|
|
restoreError,
|
|
stagedFile: restoreError ? staged.stagedFile : undefined
|
|
});
|
|
return 'failed';
|
|
}
|
|
} catch (error) {
|
|
await emitAudit(manifest, 'failed', { trigger: 'cleanup-failed', error });
|
|
return 'failed';
|
|
} finally {
|
|
if (leases.get(manifest.canonicalFile) === manifest.token) leases.delete(manifest.canonicalFile);
|
|
}
|
|
}
|
|
|
|
async function finishBatch(barriers = {}) {
|
|
const pending = [];
|
|
for (const manifest of groups.values()) {
|
|
if (!manifest.finalizationPromise) {
|
|
manifest.finalizationPromise = finalize(manifest, {
|
|
historyPersisted: barriers.historyPersisted === true,
|
|
queuePersisted: barriers.queuePersisted === true
|
|
});
|
|
}
|
|
pending.push(manifest.finalizationPromise);
|
|
}
|
|
return Promise.all(pending);
|
|
}
|
|
|
|
return Object.freeze({ registerGroups, settle, markSkipped, finishBatch });
|
|
}
|
|
|
|
module.exports = { createSourceFileCleanup };
|