From 7c809f86ff65671dc9e2764ce8894e73bf10dec7 Mon Sep 17 00:00:00 2001 From: Matt Isenhower Date: Tue, 18 Aug 2026 23:12:11 -0700 Subject: [PATCH] Test archive compression workflow --- app/data/ArchiveCompressor.mjs | 66 +------- app/data/ArchiveCompressor.test.mjs | 249 ++++++++++++++++++++++++++++ app/data/ArchiveManifest.mjs | 71 ++++++++ app/data/ArchiveManifest.test.mjs | 95 +++++++++++ app/data/ArchiveStats.mjs | 17 +- app/data/TarZstdWriter.mjs | 54 ++++++ 6 files changed, 481 insertions(+), 71 deletions(-) create mode 100644 app/data/ArchiveCompressor.test.mjs create mode 100644 app/data/ArchiveManifest.mjs create mode 100644 app/data/ArchiveManifest.test.mjs create mode 100644 app/data/TarZstdWriter.mjs diff --git a/app/data/ArchiveCompressor.mjs b/app/data/ArchiveCompressor.mjs index 1b0e667..22cac15 100644 --- a/app/data/ArchiveCompressor.mjs +++ b/app/data/ArchiveCompressor.mjs @@ -3,7 +3,6 @@ import { createReadStream, createWriteStream } from 'node:fs'; import fs from 'node:fs/promises'; import os from 'node:os'; import path from 'node:path'; -import { spawn } from 'node:child_process'; import { pipeline } from 'node:stream/promises'; import * as Sentry from '@sentry/node'; import { @@ -13,31 +12,12 @@ import { S3Client, } from '@aws-sdk/client-s3'; import prefixedConsole from '../common/prefixedConsole.mjs'; +import { createArchiveManifest } from './ArchiveManifest.mjs'; +import TarZstdWriter from './TarZstdWriter.mjs'; const downloadLimit = 5; const quietPeriod = 30 * 60 * 1000; -function processCompletion(child, name) { - let stderr = ''; - child.stderr.setEncoding('utf8'); - child.stderr.on('data', chunk => { - stderr += chunk; - }); - - return new Promise((resolve, reject) => { - child.once('error', reject); - child.once('close', (code, signal) => { - if (code === 0) { - resolve(); - } else { - reject(new Error( - `${name} failed (${signal ? `signal ${signal}` : `exit ${code}`}): ${stderr.trim()}`, - )); - } - }); - }); -} - export function compressArchives(maxDays = Infinity, dryRun = false) { let compressor = new ArchiveCompressor; compressor.maxDays = maxDays; @@ -74,6 +54,11 @@ export default class ArchiveCompressor dryRun = false; maxDays = Infinity; + constructor(s3Client, archiveWriter = new TarZstdWriter) { + this._client = s3Client; + this.archiveWriter = archiveWriter; + } + async process() { if (!this.canRun) { this.console.log('Skipping archive compressor'); @@ -184,20 +169,14 @@ export default class ArchiveCompressor files.sort((a, b) => a.path.localeCompare(b.path)); this.console.log(`Compressing ${candidate.date}`); - await this.compress(sourceDirectory, archivePath, files.map(file => file.path)); + await this.archiveWriter.write(sourceDirectory, archivePath, files.map(file => file.path)); let archive = { path: candidate.archiveKey, bytes: (await fs.stat(archivePath)).size, hash: `sha256:${await this.hashFile(archivePath)}`, }; - let manifest = { - version: 1, - date: candidate.date, - createdAt: new Date().toISOString(), - archive, - files, - }; + let manifest = createArchiveManifest(candidate.date, archive, files); this.console.log(`Uploading ${candidate.archiveKey}`); await this.upload(candidate.archiveKey, await fs.readFile(archivePath), 'application/zstd'); @@ -279,33 +258,6 @@ export default class ArchiveCompressor }; } - async compress(sourceDirectory, archivePath, files) { - let tar = spawn('tar', ['-cf', '-', '-C', sourceDirectory, '--', ...files], { - stdio: ['ignore', 'pipe', 'pipe'], - }); - let zstd = spawn('zstd', [ - '-19', - '--long=27', - '--single-thread', - '-f', - '-o', archivePath, - ], { - stdio: ['pipe', 'ignore', 'pipe'], - }); - - let results = await Promise.allSettled([ - processCompletion(tar, 'tar'), - pipeline(tar.stdout, zstd.stdin), - processCompletion(zstd, 'zstd'), - ]); - let errors = results - .filter(result => result.status === 'rejected') - .map(result => result.reason); - if (errors.length > 0) { - throw new AggregateError(errors, `Could not create ${path.basename(archivePath)}`); - } - } - async hashFile(file) { let hash = crypto.createHash('sha256'); for await (let chunk of createReadStream(file)) { diff --git a/app/data/ArchiveCompressor.test.mjs b/app/data/ArchiveCompressor.test.mjs new file mode 100644 index 0000000..a93ee11 --- /dev/null +++ b/app/data/ArchiveCompressor.test.mjs @@ -0,0 +1,249 @@ +import { Readable } from 'node:stream'; +import fs from 'node:fs/promises'; +import { + GetObjectCommand, + ListObjectsV2Command, + PutObjectCommand, +} from '@aws-sdk/client-s3'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import ArchiveCompressor from './ArchiveCompressor.mjs'; + +class FakeS3Client +{ + completed = false; + downloads = []; + files = new Map([ + ['2025/03/05/alpha.json', Buffer.from('alpha\n')], + ['2025/03/05/nested/beta.json', Buffer.from('beta\n')], + ]); + secondDay = false; + uploads = []; + + async send(command) { + if (command instanceof ListObjectsV2Command) { + return this.list(command.input.Prefix); + } + + if (command instanceof GetObjectCommand) { + this.downloads.push(command.input.Key); + + return { Body: Readable.from(this.files.get(command.input.Key)) }; + } + + if (command instanceof PutObjectCommand) { + this.uploads.push(command.input); + + return {}; + } + + throw new Error(`Unexpected S3 command: ${command.constructor.name}`); + } + + list(prefix) { + if (prefix === '') { + return { CommonPrefixes: [{ Prefix: '2025/' }] }; + } + if (prefix === '2025/') { + return { CommonPrefixes: [{ Prefix: '2025/03/' }] }; + } + if (prefix === '2025/03/') { + return { + CommonPrefixes: [ + { Prefix: '2025/03/05/' }, + ...(this.secondDay ? [{ Prefix: '2025/03/06/' }] : []), + ], + Contents: this.completed ? [ + { Key: '2025/03/2025-03-05.tar.zst' }, + { Key: '2025/03/2025-03-05.tar.zst.manifest.json' }, + ] : [], + }; + } + if (/^2025\/03\/\d{2}\/$/.test(prefix)) { + return { Contents: [...this.files] + .filter(([Key]) => Key.startsWith(prefix)) + .map(([Key, body]) => ({ + Key, + LastModified: new Date('2025-03-05T23:45:00.000Z'), + Size: body.length, + })) }; + } + + throw new Error(`Unexpected S3 prefix: ${prefix}`); + } +} + +class PaginatedS3Client extends FakeS3Client +{ + async send(command) { + if (command instanceof ListObjectsV2Command && command.input.Prefix === '') { + if (!command.input.ContinuationToken) { + return { + CommonPrefixes: [], + IsTruncated: true, + NextContinuationToken: 'next-page', + }; + } + + expect(command.input.ContinuationToken).toBe('next-page'); + + return { CommonPrefixes: [{ Prefix: '2025/' }] }; + } + + return super.send(command); + } +} + +describe('ArchiveCompressor', () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(new Date('2026-08-18T12:00:00.000Z')); + process.env.AWS_S3_ENDPOINT = 'https://example.invalid'; + process.env.AWS_REGION = 'test'; + process.env.AWS_S3_ARCHIVE_BUCKET = 'archive'; + process.env.AWS_ACCESS_KEY_ID = 'key'; + process.env.AWS_SECRET_ACCESS_KEY = 'secret'; + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('downloads a day and publishes its archive and manifest', async () => { + let s3Client = new FakeS3Client; + let temporaryDirectory; + let archiveWriter = { + async write(sourceDirectory, archivePath, files) { + temporaryDirectory = sourceDirectory.slice(0, -'/source'.length); + expect(files).toEqual(['alpha.json', 'nested/beta.json']); + expect(await fs.readFile(`${sourceDirectory}/alpha.json`, 'utf8')).toBe('alpha\n'); + expect(await fs.readFile(`${sourceDirectory}/nested/beta.json`, 'utf8')).toBe('beta\n'); + await fs.writeFile(archivePath, 'fake compressed archive'); + }, + }; + let compressor = new ArchiveCompressor(s3Client, archiveWriter); + 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', + ]); + expect(s3Client.uploads[0].Body.toString()).toBe('fake compressed archive'); + expect(JSON.parse(s3Client.uploads[1].Body)).toEqual({ + version: 1, + date: '2025-03-05', + createdAt: '2026-08-18T12:00:00.000Z', + archive: { + path: '2025/03/2025-03-05.tar.zst', + bytes: 23, + hash: 'sha256:01bba9b6e21a70b88e2fb19741b8aa5e24f7d0ee368224528e7a451fcce5cbc6', + }, + files: [ + { + path: 'alpha.json', + bytes: 6, + hash: 'sha256:b6a98d9ce9a2d9149288fa3df42d377c3e42737afdcdaf714e33c0a100b51060', + }, + { + path: 'nested/beta.json', + bytes: 5, + hash: 'sha256:f2c82decdd7181cf98945929a62598db7e6b477e11f6e0eb0ae97020eff151ad', + }, + ], + }); + await expect(fs.stat(temporaryDirectory)).rejects.toMatchObject({ code: 'ENOENT' }); + }); + + it('does not download or upload files during a dry run', async () => { + let s3Client = new FakeS3Client; + let archiveWriter = { write: vi.fn() }; + let compressor = new ArchiveCompressor(s3Client, archiveWriter); + compressor.dryRun = true; + compressor._console = { error: vi.fn(), log: vi.fn() }; + + await compressor.process(); + + expect(s3Client.downloads).toEqual([]); + expect(s3Client.uploads).toEqual([]); + expect(archiveWriter.write).not.toHaveBeenCalled(); + }); + + it('leaves a completed archive alone', async () => { + let s3Client = new FakeS3Client; + s3Client.completed = true; + let archiveWriter = { write: vi.fn() }; + let compressor = new ArchiveCompressor(s3Client, archiveWriter); + compressor._console = { error: vi.fn(), log: vi.fn() }; + + await compressor.process(); + + expect(s3Client.downloads).toEqual([]); + 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; + s3Client.secondDay = true; + s3Client.files.set('2025/03/06/gamma.json', Buffer.from('gamma\n')); + let temporaryDirectory; + let archiveWriter = { + async write(sourceDirectory) { + temporaryDirectory = sourceDirectory.slice(0, -'/source'.length); + throw new Error('compression failed'); + }, + }; + let compressor = new ArchiveCompressor(s3Client, archiveWriter); + compressor._console = { error: vi.fn(), log: vi.fn() }; + + await expect(compressor.process()).rejects.toThrow('compression failed'); + + expect(s3Client.downloads).toEqual([ + '2025/03/05/alpha.json', + '2025/03/05/nested/beta.json', + ]); + expect(s3Client.uploads).toEqual([]); + await expect(fs.stat(temporaryDirectory)).rejects.toMatchObject({ code: 'ENOENT' }); + }); + + it('continues S3 discovery across paginated listings', async () => { + let s3Client = new PaginatedS3Client; + let archiveWriter = { + async write(sourceDirectory, archivePath) { + await fs.writeFile(archivePath, 'archive'); + }, + }; + let compressor = new ArchiveCompressor(s3Client, archiveWriter); + 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('does not process more than maxDays', async () => { + let s3Client = new FakeS3Client; + s3Client.secondDay = true; + s3Client.files.set('2025/03/06/gamma.json', Buffer.from('gamma\n')); + let archiveWriter = { + async write(sourceDirectory, archivePath) { + await fs.writeFile(archivePath, 'archive'); + }, + }; + let compressor = new ArchiveCompressor(s3Client, archiveWriter); + compressor.maxDays = 1; + 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', + ]); + }); +}); diff --git a/app/data/ArchiveManifest.mjs b/app/data/ArchiveManifest.mjs new file mode 100644 index 0000000..eca6bb8 --- /dev/null +++ b/app/data/ArchiveManifest.mjs @@ -0,0 +1,71 @@ +import path from 'node:path'; + +const sha256Pattern = /^sha256:[a-f0-9]{64}$/; + +export function createArchiveManifest(date, archive, files) { + return { + version: 1, + date, + createdAt: new Date().toISOString(), + archive, + files, + }; +} + +export function getArchiveManifestStats(manifest) { + return { + path: manifest.archive.path, + compressedBytes: manifest.archive.bytes, + fileCount: manifest.files.length, + originalBytes: manifest.files.reduce((total, file) => total + file.bytes, 0), + }; +} + +export function parseArchiveManifest(contents, archivePath) { + let manifest = JSON.parse(contents); + let filePaths = new Set; + if (manifest.version !== 1 + || !isDate(manifest.date) + || typeof manifest.createdAt !== 'string' + || !Number.isFinite(Date.parse(manifest.createdAt)) + || manifest.archive?.path !== archivePath + || !Number.isSafeInteger(manifest.archive?.bytes) + || manifest.archive.bytes < 1 + || !sha256Pattern.test(manifest.archive?.hash) + || !Array.isArray(manifest.files) + || manifest.files.some(file => { + if (!isRelativePath(file.path) + || filePaths.has(file.path) + || !Number.isSafeInteger(file.bytes) + || file.bytes < 0 + || !sha256Pattern.test(file.hash)) { + return true; + } + + filePaths.add(file.path); + + return false; + })) { + throw new Error(`Invalid archive manifest: ${archivePath}`); + } + + return manifest; +} + +function isDate(date) { + if (typeof date !== 'string' || !/^\d{4}-\d{2}-\d{2}$/.test(date)) { + return false; + } + + let parsedDate = new Date(`${date}T00:00:00.000Z`); + + return !Number.isNaN(parsedDate.getTime()) && parsedDate.toISOString().slice(0, 10) === date; +} + +function isRelativePath(filePath) { + return typeof filePath === 'string' + && filePath.length > 0 + && !path.posix.isAbsolute(filePath) + && path.posix.normalize(filePath) === filePath + && !filePath.split('/').includes('..'); +} diff --git a/app/data/ArchiveManifest.test.mjs b/app/data/ArchiveManifest.test.mjs new file mode 100644 index 0000000..d505c90 --- /dev/null +++ b/app/data/ArchiveManifest.test.mjs @@ -0,0 +1,95 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { + createArchiveManifest, + getArchiveManifestStats, + parseArchiveManifest, +} from './ArchiveManifest.mjs'; + +describe('ArchiveManifest', () => { + afterEach(() => { + vi.useRealTimers(); + }); + + it('creates the public manifest format', () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date('2026-08-18T12:00:00.000Z')); + let archive = { + path: '2025/03/2025-03-05.tar.zst', + bytes: 123, + hash: `sha256:${'a'.repeat(64)}`, + }; + let files = [{ + path: 'sample.json', + bytes: 456, + hash: `sha256:${'b'.repeat(64)}`, + }]; + + expect(createArchiveManifest('2025-03-05', archive, files)).toEqual({ + version: 1, + date: '2025-03-05', + createdAt: '2026-08-18T12:00:00.000Z', + archive, + files, + }); + }); + + it('parses a valid manifest', () => { + let manifest = { + version: 1, + date: '2025-03-05', + createdAt: '2026-08-18T12:00:00.000Z', + archive: { + path: '2025/03/2025-03-05.tar.zst', + bytes: 123, + hash: `sha256:${'a'.repeat(64)}`, + }, + files: [{ + path: 'sample.json', + bytes: 456, + hash: `sha256:${'b'.repeat(64)}`, + }], + }; + + expect(parseArchiveManifest( + JSON.stringify(manifest), + '2025/03/2025-03-05.tar.zst', + )).toEqual(manifest); + }); + + it('rejects a manifest without valid SHA-256 hashes', () => { + let manifest = { + version: 1, + date: '2025-03-05', + createdAt: '2026-08-18T12:00:00.000Z', + archive: { + path: '2025/03/2025-03-05.tar.zst', + bytes: 123, + hash: 'not-a-hash', + }, + files: [{ + path: 'sample.json', + bytes: 456, + hash: `sha256:${'b'.repeat(64)}`, + }], + }; + + expect(() => parseArchiveManifest( + JSON.stringify(manifest), + '2025/03/2025-03-05.tar.zst', + )).toThrow('Invalid archive manifest'); + }); + + it('summarizes the sizes recorded in a manifest', () => { + let manifest = { + archive: { path: '2025/03/2025-03-05.tar.zst', bytes: 123 }, + files: [{ bytes: 400 }, { bytes: 600 }], + }; + + expect(getArchiveManifestStats(manifest)).toEqual({ + path: '2025/03/2025-03-05.tar.zst', + compressedBytes: 123, + fileCount: 2, + originalBytes: 1000, + }); + }); +}); diff --git a/app/data/ArchiveStats.mjs b/app/data/ArchiveStats.mjs index 192cc7c..7fbda56 100644 --- a/app/data/ArchiveStats.mjs +++ b/app/data/ArchiveStats.mjs @@ -5,6 +5,7 @@ import { } from '@aws-sdk/client-s3'; import pLimit from 'p-limit'; import prefixedConsole from '../common/prefixedConsole.mjs'; +import { getArchiveManifestStats, parseArchiveManifest } from './ArchiveManifest.mjs'; const manifestSuffix = '.tar.zst.manifest.json'; const requestLimit = pLimit(10); @@ -115,22 +116,10 @@ export default class ArchiveStats throw new Error(`S3 returned no body for ${key}`); } - let manifest = JSON.parse(await response.Body.transformToString()); let archivePath = key.slice(0, -'.manifest.json'.length); - if (manifest.archive?.path !== archivePath - || !Number.isSafeInteger(manifest.archive?.bytes) - || manifest.archive.bytes < 1 - || !Array.isArray(manifest.files) - || manifest.files.some(file => !Number.isSafeInteger(file.bytes) || file.bytes < 0)) { - throw new Error(`Invalid archive manifest: ${key}`); - } + let manifest = parseArchiveManifest(await response.Body.transformToString(), archivePath); - return { - path: manifest.archive.path, - compressedBytes: manifest.archive.bytes, - fileCount: manifest.files.length, - originalBytes: manifest.files.reduce((total, file) => total + file.bytes, 0), - }; + return getArchiveManifestStats(manifest); } async list(prefix, delimiter) { diff --git a/app/data/TarZstdWriter.mjs b/app/data/TarZstdWriter.mjs new file mode 100644 index 0000000..dd266e6 --- /dev/null +++ b/app/data/TarZstdWriter.mjs @@ -0,0 +1,54 @@ +import path from 'node:path'; +import { spawn } from 'node:child_process'; +import { pipeline } from 'node:stream/promises'; + +function processCompletion(child, name) { + let stderr = ''; + child.stderr.setEncoding('utf8'); + child.stderr.on('data', chunk => { + stderr += chunk; + }); + + return new Promise((resolve, reject) => { + child.once('error', reject); + child.once('close', (code, signal) => { + if (code === 0) { + resolve(); + } else { + reject(new Error( + `${name} failed (${signal ? `signal ${signal}` : `exit ${code}`}): ${stderr.trim()}`, + )); + } + }); + }); +} + +export default class TarZstdWriter +{ + async write(sourceDirectory, archivePath, files) { + let tar = spawn('tar', ['-cf', '-', '-C', sourceDirectory, '--', ...files], { + stdio: ['ignore', 'pipe', 'pipe'], + }); + let zstd = spawn('zstd', [ + '-19', + '--long=27', + '--single-thread', + '-f', + '-o', archivePath, + ], { + stdio: ['pipe', 'ignore', 'pipe'], + }); + + let results = await Promise.allSettled([ + processCompletion(tar, 'tar'), + pipeline(tar.stdout, zstd.stdin), + processCompletion(zstd, 'zstd'), + ]); + let errors = results + .filter(result => result.status === 'rejected') + .map(result => result.reason); + if (errors.length > 0) { + throw new AggregateError(errors, `Could not create ${path.basename(archivePath)}`); + } + } +}