diff --git a/app/data/ArchiveCompressor.mjs b/app/data/ArchiveCompressor.mjs index 22cac15..1aea019 100644 --- a/app/data/ArchiveCompressor.mjs +++ b/app/data/ArchiveCompressor.mjs @@ -18,10 +18,11 @@ import TarZstdWriter from './TarZstdWriter.mjs'; const downloadLimit = 5; const quietPeriod = 30 * 60 * 1000; -export function compressArchives(maxDays = Infinity, dryRun = false) { +export function compressArchives(maxDays = Infinity, dryRun = false, rebuildBefore = null) { let compressor = new ArchiveCompressor; compressor.maxDays = maxDays; compressor.dryRun = dryRun; + compressor.rebuildBefore = rebuildBefore; return compressor.process(); } @@ -29,15 +30,19 @@ export function compressArchives(maxDays = Infinity, dryRun = false) { export function compressArchivesFromCli(args) { let dryRun = false; let maxDays = Infinity; + let rebuildBefore = null; for (let i = 0; i < args.length; i++) { if (args[i] === '--dry-run') { dryRun = true; + } else if (args[i] === '--rebuild-before' && Number.isFinite(Date.parse(args[i + 1]))) { + rebuildBefore = Date.parse(args[++i]); } else if (args[i] === '--max-days' && /^\d+$/.test(args[i + 1])) { maxDays = Number(args[++i]); } else { throw new Error( - 'Usage: npm run data:archive:compress -- [--dry-run] [--max-days DAYS]', + 'Usage: npm run data:archive:compress -- ' + + '[--dry-run] [--max-days DAYS] [--rebuild-before TIMESTAMP]', ); } } @@ -46,13 +51,14 @@ export function compressArchivesFromCli(args) { throw new Error('--max-days must be at least 1'); } - return compressArchives(maxDays, dryRun); + return compressArchives(maxDays, dryRun, rebuildBefore); } export default class ArchiveCompressor { dryRun = false; maxDays = Infinity; + rebuildBefore = null; constructor(s3Client, archiveWriter = new TarZstdWriter) { this._client = s3Client; @@ -143,7 +149,8 @@ export default class ArchiveCompressor } this.console.log( - `Would create ${candidate.archiveKey} and ${candidate.manifestKey} ` + `Would ${candidate.exists ? 'replace' : 'create'} ` + + `${candidate.archiveKey} and ${candidate.manifestKey} ` + `from ${objects.length} files`, ); @@ -178,7 +185,7 @@ export default class ArchiveCompressor }; let manifest = createArchiveManifest(candidate.date, archive, files); - this.console.log(`Uploading ${candidate.archiveKey}`); + this.console.log(`${candidate.exists ? 'Replacing' : 'Uploading'} ${candidate.archiveKey}`); await this.upload(candidate.archiveKey, await fs.readFile(archivePath), 'application/zstd'); await this.upload( candidate.manifestKey, @@ -290,7 +297,7 @@ export default class ArchiveCompressor for (let month of months.sort()) { let listing = await this.list(month, '/'); - let existing = new Set(listing.objects.map(object => object.Key)); + let existing = new Map(listing.objects.map(object => [object.Key, object])); for (let prefix of listing.prefixes.sort()) { let date = this.dateFromPrefix(prefix); @@ -300,11 +307,22 @@ export default class ArchiveCompressor let archiveKey = `${month}${date}.tar.zst`; let manifestKey = `${archiveKey}.manifest.json`; - if (existing.has(archiveKey) && existing.has(manifestKey)) { - continue; + let exists = existing.has(archiveKey) && existing.has(manifestKey); + if (exists) { + if (this.rebuildBefore === null) { + continue; + } + + let modified = existing.get(manifestKey).LastModified?.getTime(); + if (!Number.isFinite(modified)) { + throw new Error(`S3 returned incomplete metadata for ${manifestKey}`); + } + if (modified >= this.rebuildBefore) { + continue; + } } - candidates.push({ archiveKey, date, manifestKey, prefix }); + candidates.push({ archiveKey, date, exists, manifestKey, prefix }); } } } diff --git a/app/data/ArchiveCompressor.test.mjs b/app/data/ArchiveCompressor.test.mjs index 597ca97..105ee51 100644 --- a/app/data/ArchiveCompressor.test.mjs +++ b/app/data/ArchiveCompressor.test.mjs @@ -16,6 +16,7 @@ class FakeS3Client ['2025/03/05/alpha.json', Buffer.from('alpha\n')], ['2025/03/05/nested/beta.json', Buffer.from('beta\n')], ]); + manifestLastModified = new Date('2026-08-20T00:00:00.000Z'); secondDay = false; uploads = []; @@ -54,7 +55,10 @@ class FakeS3Client ], Contents: this.completed ? [ { Key: '2025/03/2025-03-05.tar.zst' }, - { Key: '2025/03/2025-03-05.tar.zst.manifest.json' }, + { + Key: '2025/03/2025-03-05.tar.zst.manifest.json', + LastModified: this.manifestLastModified, + }, ] : [], }; } @@ -183,6 +187,41 @@ describe('ArchiveCompressor', () => { expect(archiveWriter.write).not.toHaveBeenCalled(); }); + it('rebuilds a completed archive older than the repair cutoff', async () => { + let s3Client = new FakeS3Client; + s3Client.completed = true; + let archiveWriter = { + async write(sourceDirectory, archivePath) { + await fs.writeFile(archivePath, 'replacement archive'); + }, + }; + let compressor = new ArchiveCompressor(s3Client, archiveWriter); + compressor.rebuildBefore = Date.parse('2026-08-22T16:49:12.000Z'); + compressor._console = { error: vi.fn(), log: vi.fn() }; + + await compressor.process(); + + expect(s3Client.uploads.map(upload => upload.Key)).toEqual([ + '2025/03/2025-03-05.tar.zst', + '2025/03/2025-03-05.tar.zst.manifest.json', + ]); + }); + + it('leaves a rebuilt archive newer than the repair cutoff alone', async () => { + let s3Client = new FakeS3Client; + s3Client.completed = true; + s3Client.manifestLastModified = new Date('2026-08-23T00:00:00.000Z'); + let archiveWriter = { write: vi.fn() }; + let compressor = new ArchiveCompressor(s3Client, archiveWriter); + compressor.rebuildBefore = Date.parse('2026-08-22T16:49:12.000Z'); + compressor._console = { error: vi.fn(), log: vi.fn() }; + + await compressor.process(); + + expect(s3Client.uploads).toEqual([]); + expect(archiveWriter.write).not.toHaveBeenCalled(); + }); + it('stops after a failed day and removes its temporary files', async () => { vi.useRealTimers(); let s3Client = new FakeS3Client; diff --git a/app/data/ArchiveVerifier.mjs b/app/data/ArchiveVerifier.mjs index 6040c91..cfb4b22 100644 --- a/app/data/ArchiveVerifier.mjs +++ b/app/data/ArchiveVerifier.mjs @@ -4,6 +4,8 @@ import { Transform } from 'node:stream'; import { pipeline } from 'node:stream/promises'; import * as tar from 'tar-stream'; +class ArchiveContentError extends Error {} + function processCompletion(child) { let stderr = ''; child.stderr.setEncoding('utf8'); @@ -69,6 +71,11 @@ export default class ArchiveVerifier decompressor.completed, this.verifyTar(decompressor.stream, inventory), ]); + let verificationResult = results[2]; + if (verificationResult.status === 'rejected' + && verificationResult.reason instanceof ArchiveContentError) { + throw verificationResult.reason; + } let errors = results .filter(result => result.status === 'rejected') .map(result => result.reason); @@ -115,22 +122,22 @@ export default class ArchiveVerifier await pipeline(stream, extract); if (verified.size !== manifestByPath.size) { - throw new Error('Tar archive is missing files from its manifest'); + throw new ArchiveContentError('Tar archive is missing files from its manifest'); } } async verifyEntry(header, entry, manifestByPath, sourceByPath, verified) { if (header.type !== 'file') { - throw new Error(`Tar archive contains a non-file entry: ${header.name}`); + throw new ArchiveContentError(`Tar archive contains a non-file entry: ${header.name}`); } if (verified.has(header.name)) { - throw new Error(`Tar archive contains a duplicate file: ${header.name}`); + throw new ArchiveContentError(`Tar archive contains a duplicate file: ${header.name}`); } let manifestFile = manifestByPath.get(header.name); let sourceObject = sourceByPath.get(header.name); if (!manifestFile || !sourceObject) { - throw new Error(`Tar archive contains an unexpected file: ${header.name}`); + throw new ArchiveContentError(`Tar archive contains an unexpected file: ${header.name}`); } verified.add(header.name); @@ -144,10 +151,10 @@ export default class ArchiveVerifier } if (bytes !== manifestFile.bytes) { - throw new Error(`Size does not match the manifest for ${header.name}`); + throw new ArchiveContentError(`Size does not match the manifest for ${header.name}`); } if (`sha256:${sha256.digest('hex')}` !== manifestFile.hash) { - throw new Error(`SHA-256 does not match the manifest for ${header.name}`); + throw new ArchiveContentError(`SHA-256 does not match the manifest for ${header.name}`); } let etag = sourceObject.etag; @@ -155,10 +162,10 @@ export default class ArchiveVerifier etag = etag.slice(1, -1); } if (!/^[a-fA-F0-9]{32}$/.test(etag)) { - throw new Error(`S3 ETag is not an MD5 hash for ${header.name}`); + throw new ArchiveContentError(`S3 ETag is not an MD5 hash for ${header.name}`); } if (md5.digest('hex') !== etag.toLowerCase()) { - throw new Error(`MD5 does not match the S3 ETag for ${header.name}`); + throw new ArchiveContentError(`MD5 does not match the S3 ETag for ${header.name}`); } } } diff --git a/app/data/ArchiveVerifier.test.mjs b/app/data/ArchiveVerifier.test.mjs index be02155..490de7b 100644 --- a/app/data/ArchiveVerifier.test.mjs +++ b/app/data/ArchiveVerifier.test.mjs @@ -97,12 +97,15 @@ describe('ArchiveVerifier', () => { )).rejects.toThrow('S3 ETag is not an MD5 hash for alpha.json'); }); - it('rejects a file that exists only in the tar archive', async () => { + it('reports an unexpected tar file instead of the resulting broken pipe', async () => { let archive = await createTar([ ['alpha.json', 'alpha\n'], ['extra.json', 'extra\n'], ]); - let verifier = new ArchiveVerifier(identityDecompressor); + let verifier = new ArchiveVerifier(stream => ({ + completed: Promise.reject(new Error('zstd failed (exit 70): Broken pipe')), + stream, + })); await expect(verifier.verify( Readable.from(archive), diff --git a/app/data/TarZstdWriter.mjs b/app/data/TarZstdWriter.mjs index dd266e6..b4e0811 100644 --- a/app/data/TarZstdWriter.mjs +++ b/app/data/TarZstdWriter.mjs @@ -27,6 +27,7 @@ export default class TarZstdWriter { async write(sourceDirectory, archivePath, files) { let tar = spawn('tar', ['-cf', '-', '-C', sourceDirectory, '--', ...files], { + env: { ...process.env, COPYFILE_DISABLE: '1' }, stdio: ['ignore', 'pipe', 'pipe'], }); let zstd = spawn('zstd', [ diff --git a/app/data/TarZstdWriter.test.mjs b/app/data/TarZstdWriter.test.mjs new file mode 100644 index 0000000..10d46f9 --- /dev/null +++ b/app/data/TarZstdWriter.test.mjs @@ -0,0 +1,57 @@ +import { execFile, spawn } from 'node:child_process'; +import { createReadStream } from 'node:fs'; +import fs from 'node:fs/promises'; +import os from 'node:os'; +import path from 'node:path'; +import { pipeline } from 'node:stream/promises'; +import { promisify } from 'node:util'; +import { expect, it } from 'vitest'; +import * as tar from 'tar-stream'; +import TarZstdWriter from './TarZstdWriter.mjs'; + +const execFileAsync = promisify(execFile); + +async function listFiles(stream) { + let files = []; + let extract = tar.extract(); + extract.on('entry', (header, entry, next) => { + files.push(header.name); + entry.on('end', next); + entry.resume(); + }); + await pipeline(stream, extract); + + return files; +} + +it.runIf(process.platform === 'darwin')('does not add macOS metadata files', async () => { + let temporaryDirectory = await fs.mkdtemp(path.join(os.tmpdir(), 'splatoon3ink-tar-test-')); + let sourceDirectory = path.join(temporaryDirectory, 'source'); + let archivePath = path.join(temporaryDirectory, 'archive.tar.zst'); + let rawArchivePath = path.join(temporaryDirectory, 'archive.tar'); + + try { + await fs.mkdir(sourceDirectory); + let sourcePath = path.join(sourceDirectory, 'sample.json'); + await fs.writeFile(sourcePath, '{}\n'); + await execFileAsync('xattr', ['-w', 'com.splatoon3ink.test', 'metadata', sourcePath]); + let rawTarEnvironment = { ...process.env }; + delete rawTarEnvironment.COPYFILE_DISABLE; + await execFileAsync('tar', [ + '-cf', rawArchivePath, + '-C', sourceDirectory, + '--', 'sample.json', + ], { env: rawTarEnvironment }); + expect(await listFiles(createReadStream(rawArchivePath))).toEqual([ + '._sample.json', + 'sample.json', + ]); + + await new TarZstdWriter().write(sourceDirectory, archivePath, ['sample.json']); + + let zstd = spawn('zstd', ['-dc', archivePath]); + expect(await listFiles(zstd.stdout)).toEqual(['sample.json']); + } finally { + await fs.rm(temporaryDirectory, { force: true, recursive: true }); + } +});