From 08570dfa43ae5f82059eca4498edaecfcaedbca1 Mon Sep 17 00:00:00 2001 From: Matt Isenhower Date: Sat, 22 Aug 2026 08:54:39 -0700 Subject: [PATCH] Add verified archive pruning command --- app/data/ArchivePruner.mjs | 328 ++++++++++++++++++++++++++++++ app/data/ArchivePruner.test.mjs | 171 ++++++++++++++++ app/data/ArchiveVerifier.mjs | 164 +++++++++++++++ app/data/ArchiveVerifier.test.mjs | 124 +++++++++++ app/index.mjs | 2 + package-lock.json | 21 +- package.json | 3 + 7 files changed, 801 insertions(+), 12 deletions(-) create mode 100644 app/data/ArchivePruner.mjs create mode 100644 app/data/ArchivePruner.test.mjs create mode 100644 app/data/ArchiveVerifier.mjs create mode 100644 app/data/ArchiveVerifier.test.mjs diff --git a/app/data/ArchivePruner.mjs b/app/data/ArchivePruner.mjs new file mode 100644 index 0000000..b84ff36 --- /dev/null +++ b/app/data/ArchivePruner.mjs @@ -0,0 +1,328 @@ +import path from 'node:path'; +import * as Sentry from '@sentry/node'; +import { + DeleteObjectsCommand, + GetObjectCommand, + ListObjectsV2Command, + S3Client, +} from '@aws-sdk/client-s3'; +import prefixedConsole from '../common/prefixedConsole.mjs'; +import { parseArchiveManifest } from './ArchiveManifest.mjs'; +import ArchiveVerifier from './ArchiveVerifier.mjs'; + +const retentionPeriod = 7 * 24 * 60 * 60 * 1000; + +export function pruneArchives(maxDays = Infinity, dryRun = false) { + let pruner = new ArchivePruner; + pruner.maxDays = maxDays; + pruner.dryRun = dryRun; + + return pruner.process(); +} + +export function pruneArchivesFromCli(args) { + let dryRun = false; + let maxDays = Infinity; + + for (let i = 0; i < args.length; i++) { + if (args[i] === '--dry-run') { + dryRun = true; + } else if (args[i] === '--max-days' && /^\d+$/.test(args[i + 1])) { + maxDays = Number(args[++i]); + } else { + throw new Error( + 'Usage: npm run data:archive:prune -- [--dry-run] [--max-days DAYS]', + ); + } + } + + if (maxDays < 1) { + throw new Error('--max-days must be at least 1'); + } + + return pruneArchives(maxDays, dryRun); +} + +export default class ArchivePruner +{ + dryRun = false; + maxDays = Infinity; + + constructor(s3Client, archiveVerifier = new ArchiveVerifier) { + this._client = s3Client; + this.archiveVerifier = archiveVerifier; + } + + async process() { + if (!this.canRun) { + this.console.log('Skipping archive pruner'); + + return; + } + + let currentDate; + + try { + let candidates = await this.getCandidates(); + this.console.log(`Found ${candidates.length} dates to verify`); + let processedDays = 0; + + for (let candidate of candidates) { + if (processedDays >= this.maxDays) { + break; + } + + currentDate = candidate.date; + let manifest = await this.getManifest(candidate); + if (!this.dryRun && !this.isEligible(manifest)) { + this.console.log( + `Skipping ${candidate.date}; not eligible for deletion until ${this.eligibleDate(manifest)}`, + ); + continue; + } + + let sourceObjects = await this.getSourceObjects(candidate.prefix); + let archiveStream = await this.getArchive(candidate.archiveKey); + await this.archiveVerifier.verify(archiveStream, manifest, sourceObjects); + processedDays++; + + if (this.dryRun) { + if (this.isEligible(manifest)) { + this.console.log( + `Verified ${candidate.date}; would delete ${this.fileCount(sourceObjects)}`, + ); + } else { + this.console.log( + `Verified ${candidate.date}; not eligible for deletion until ${this.eligibleDate(manifest)}`, + ); + } + continue; + } + + let currentObjects = await this.getSourceObjects(candidate.prefix); + this.assertObjectsUnchanged(sourceObjects, currentObjects, candidate.date); + await this.deleteObjects(currentObjects); + this.console.log(`Pruned ${candidate.date}; deleted ${this.fileCount(currentObjects)}`); + } + + this.console.log( + this.dryRun + ? `Verified ${processedDays} daily archives` + : `Pruned ${processedDays} daily archives`, + ); + } catch (e) { + this.console.error(e); + Sentry.withScope(scope => { + if (currentDate) { + scope.setTag('archive.date', currentDate); + } + Sentry.captureException(e); + }); + await Sentry.flush(2000).catch(() => {}); + throw e; + } + } + + get console() { + this._console ??= prefixedConsole('Archive Pruner'); + + return this._console; + } + + get canRun() { + return process.env.AWS_S3_ENDPOINT + && process.env.AWS_REGION + && process.env.AWS_S3_ARCHIVE_BUCKET + && process.env.AWS_ACCESS_KEY_ID + && process.env.AWS_SECRET_ACCESS_KEY; + } + + get s3Client() { + return this._client ??= new S3Client({ + endpoint: process.env.AWS_S3_ENDPOINT, + region: process.env.AWS_REGION, + requestChecksumCalculation: 'WHEN_REQUIRED', + responseChecksumValidation: 'WHEN_REQUIRED', + credentials: { + accessKeyId: process.env.AWS_ACCESS_KEY_ID, + secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY, + }, + }); + } + + async getCandidates() { + let candidates = []; + let years = (await this.list('', '/')).prefixes.filter(prefix => /^\d{4}\/$/.test(prefix)); + + for (let year of years.sort()) { + let months = (await this.list(year, '/')).prefixes + .filter(prefix => /^\d{4}\/\d{2}\/$/.test(prefix)); + + for (let month of months.sort()) { + let listing = await this.list(month, '/'); + let existing = new Set(listing.objects.map(object => object.Key)); + + for (let prefix of listing.prefixes.sort()) { + let date = this.dateFromPrefix(prefix); + if (!date) { + continue; + } + + let archiveKey = `${month}${date}.tar.zst`; + let manifestKey = `${archiveKey}.manifest.json`; + let hasArchive = existing.has(archiveKey); + let hasManifest = existing.has(manifestKey); + if (hasArchive !== hasManifest) { + throw new Error(`Archive and manifest are incomplete for ${date}`); + } + if (hasArchive) { + candidates.push({ archiveKey, date, manifestKey, prefix }); + } + } + } + } + + return candidates.sort((a, b) => a.date.localeCompare(b.date)); + } + + async getManifest(candidate) { + let response = await this.s3Client.send(new GetObjectCommand({ + Bucket: process.env.AWS_S3_ARCHIVE_BUCKET, + Key: candidate.manifestKey, + })); + if (!response.Body) { + throw new Error(`S3 returned no body for ${candidate.manifestKey}`); + } + + return parseArchiveManifest( + await response.Body.transformToString(), + candidate.archiveKey, + ); + } + + async getArchive(key) { + let response = await this.s3Client.send(new GetObjectCommand({ + Bucket: process.env.AWS_S3_ARCHIVE_BUCKET, + Key: key, + })); + if (!response.Body) { + throw new Error(`S3 returned no body for ${key}`); + } + + return response.Body; + } + + async getSourceObjects(prefix) { + let listing = await this.list(prefix); + + return listing.objects + .filter(object => object.Key && object.Key !== prefix && !object.Key.endsWith('/')) + .map(object => { + if (!Number.isFinite(object.Size) || typeof object.ETag !== 'string') { + throw new Error(`S3 returned incomplete metadata for ${object.Key}`); + } + + return { + path: this.relativePath(prefix, object.Key), + bytes: object.Size, + etag: object.ETag, + key: object.Key, + }; + }) + .sort((a, b) => a.path.localeCompare(b.path)); + } + + async deleteObjects(objects) { + for (let i = 0; i < objects.length; i += 1000) { + let batch = objects.slice(i, i + 1000); + let response = await this.s3Client.send(new DeleteObjectsCommand({ + Bucket: process.env.AWS_S3_ARCHIVE_BUCKET, + Delete: { + Objects: batch.map(object => ({ Key: object.key })), + Quiet: true, + }, + })); + if (response.Errors?.length) { + let failures = response.Errors.map(error => `${error.Key}: ${error.Code}`).join(', '); + throw new Error(`Could not delete archive source files: ${failures}`); + } + } + } + + async list(prefix, delimiter) { + let prefixes = []; + let objects = []; + let continuationToken; + + do { + let response = await this.s3Client.send(new ListObjectsV2Command({ + Bucket: process.env.AWS_S3_ARCHIVE_BUCKET, + ContinuationToken: continuationToken, + Delimiter: delimiter, + Prefix: prefix, + })); + prefixes.push(...(response.CommonPrefixes ?? []).map(item => item.Prefix).filter(Boolean)); + objects.push(...(response.Contents ?? [])); + + if (response.IsTruncated && !response.NextContinuationToken) { + throw new Error(`S3 listing for ${prefix} was truncated without a continuation token`); + } + continuationToken = response.IsTruncated ? response.NextContinuationToken : undefined; + } while (continuationToken); + + return { objects, prefixes }; + } + + isEligible(manifest) { + return Date.parse(manifest.createdAt) <= Date.now() - retentionPeriod; + } + + eligibleDate(manifest) { + return new Date(Date.parse(manifest.createdAt) + retentionPeriod).toISOString().slice(0, 10); + } + + assertObjectsUnchanged(expected, actual, date) { + let unchanged = expected.length === actual.length + && expected.every((object, index) => { + let current = actual[index]; + + return object.key === current.key + && object.bytes === current.bytes + && object.etag === current.etag; + }); + if (!unchanged) { + throw new Error(`Source files changed while verifying ${date}`); + } + } + + fileCount(objects) { + return `${objects.length} ${objects.length === 1 ? 'file' : 'files'}`; + } + + dateFromPrefix(prefix) { + let match = prefix.match(/^(\d{4})\/(\d{2})\/(\d{2})\/$/); + if (!match) { + return null; + } + + let date = `${match[1]}-${match[2]}-${match[3]}`; + let parsedDate = new Date(`${date}T00:00:00.000Z`); + + return !Number.isNaN(parsedDate.getTime()) && parsedDate.toISOString().slice(0, 10) === date + ? date + : null; + } + + relativePath(prefix, key) { + let relativePath = key.slice(prefix.length); + if (!key.startsWith(prefix) + || !relativePath + || path.posix.normalize(relativePath) !== relativePath + || path.posix.isAbsolute(relativePath) + || relativePath.split('/').includes('..')) { + throw new Error(`Invalid archive object path: ${key}`); + } + + return relativePath; + } +} diff --git a/app/data/ArchivePruner.test.mjs b/app/data/ArchivePruner.test.mjs new file mode 100644 index 0000000..bbc34a8 --- /dev/null +++ b/app/data/ArchivePruner.test.mjs @@ -0,0 +1,171 @@ +import { Readable } from 'node:stream'; +import { + DeleteObjectsCommand, + GetObjectCommand, + ListObjectsV2Command, +} from '@aws-sdk/client-s3'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import ArchivePruner from './ArchivePruner.mjs'; + +const archiveKey = '2025/03/2025-03-05.tar.zst'; +const manifestKey = `${archiveKey}.manifest.json`; + +function manifest(createdAt = '2026-08-21T00:00:00.000Z') { + return { + version: 1, + date: '2025-03-05', + createdAt, + archive: { + path: archiveKey, + bytes: 7, + hash: `sha256:${'a'.repeat(64)}`, + }, + files: [{ + path: 'alpha.json', + bytes: 6, + hash: `sha256:${'b'.repeat(64)}`, + }], + }; +} + +class FakeS3Client +{ + deletes = []; + events = []; + manifest = manifest(); + manifestMissing = false; + + async send(command) { + if (command instanceof ListObjectsV2Command) { + return this.list(command.input.Prefix); + } + if (command instanceof GetObjectCommand && command.input.Key === manifestKey) { + return { + Body: { + transformToString: async () => JSON.stringify(this.manifest), + }, + }; + } + if (command instanceof GetObjectCommand && command.input.Key === archiveKey) { + return { Body: Readable.from('archive') }; + } + if (command instanceof DeleteObjectsCommand) { + this.events.push('delete'); + this.deletes.push(...command.input.Delete.Objects.map(object => object.Key)); + + 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/' }], + Contents: [ + { Key: archiveKey }, + ...(this.manifestMissing ? [] : [{ Key: manifestKey }]), + ], + }; + } + if (prefix === '2025/03/05/') { + return { + Contents: [{ + ETag: '"9f9f90dbe3e5ee1218c86b8839db1995"', + Key: '2025/03/05/alpha.json', + Size: 6, + }], + }; + } + + throw new Error(`Unexpected S3 prefix: ${prefix}`); + } +} + +describe('ArchivePruner', () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(new Date('2026-08-22T12: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('fully verifies recent archives during a dry run without deleting files', async () => { + let s3Client = new FakeS3Client; + let verifier = { verify: vi.fn() }; + let pruner = new ArchivePruner(s3Client, verifier); + pruner.dryRun = true; + pruner._console = { error: vi.fn(), log: vi.fn() }; + + await pruner.process(); + + expect(verifier.verify).toHaveBeenCalledOnce(); + expect(s3Client.deletes).toEqual([]); + expect(pruner._console.log).toHaveBeenCalledWith( + 'Verified 2025-03-05; not eligible for deletion until 2026-08-28', + ); + }); + + it('deletes an eligible day only after its archive is verified', async () => { + let s3Client = new FakeS3Client; + s3Client.manifest = manifest('2026-08-01T00:00:00.000Z'); + let verifier = { + verify: vi.fn(async () => { + s3Client.events.push('verify'); + }), + }; + let pruner = new ArchivePruner(s3Client, verifier); + pruner._console = { error: vi.fn(), log: vi.fn() }; + + await pruner.process(); + + expect(s3Client.events).toEqual(['verify', 'delete']); + expect(s3Client.deletes).toEqual(['2025/03/05/alpha.json']); + expect(pruner._console.log).toHaveBeenCalledWith('Pruned 2025-03-05; deleted 1 file'); + }); + + it('stops when an archive exists without its manifest', async () => { + let s3Client = new FakeS3Client; + s3Client.manifestMissing = true; + let verifier = { verify: vi.fn() }; + let pruner = new ArchivePruner(s3Client, verifier); + pruner._console = { error: vi.fn(), log: vi.fn() }; + + await expect(pruner.process()).rejects.toThrow( + 'Archive and manifest are incomplete for 2025-03-05', + ); + + expect(verifier.verify).not.toHaveBeenCalled(); + expect(s3Client.deletes).toEqual([]); + }); + + it('stops without deleting when archive verification fails', async () => { + let s3Client = new FakeS3Client; + s3Client.manifest = manifest('2026-08-01T00:00:00.000Z'); + let verifier = { + verify: vi.fn(async () => { + throw new Error('archive does not match'); + }), + }; + let pruner = new ArchivePruner(s3Client, verifier); + pruner._console = { error: vi.fn(), log: vi.fn() }; + + await expect(pruner.process()).rejects.toThrow('archive does not match'); + + expect(s3Client.deletes).toEqual([]); + }); +}); diff --git a/app/data/ArchiveVerifier.mjs b/app/data/ArchiveVerifier.mjs new file mode 100644 index 0000000..6040c91 --- /dev/null +++ b/app/data/ArchiveVerifier.mjs @@ -0,0 +1,164 @@ +import crypto from 'node:crypto'; +import { spawn } from 'node:child_process'; +import { Transform } from 'node:stream'; +import { pipeline } from 'node:stream/promises'; +import * as tar from 'tar-stream'; + +function processCompletion(child) { + 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( + `zstd failed (${signal ? `signal ${signal}` : `exit ${code}`}): ${stderr.trim()}`, + )); + } + }); + }); +} + +function decompressZstd(stream) { + let zstd = spawn('zstd', ['-dc'], { + stdio: ['pipe', 'pipe', 'pipe'], + }); + + return { + completed: Promise.all([ + pipeline(stream, zstd.stdin), + processCompletion(zstd), + ]), + stream: zstd.stdout, + }; +} + +export default class ArchiveVerifier +{ + constructor(decompress = decompressZstd) { + this.decompress = decompress; + } + + async verify(archiveStream, manifest, sourceObjects) { + let inventory; + try { + inventory = this.matchFiles(manifest.files, sourceObjects); + } catch (error) { + archiveStream.destroy?.(); + throw error; + } + + let archiveBytes = 0; + let archiveHash = crypto.createHash('sha256'); + let inspector = new Transform({ + transform(chunk, encoding, callback) { + archiveBytes += chunk.length; + archiveHash.update(chunk); + callback(null, chunk); + }, + }); + let decompressor = this.decompress(inspector); + let results = await Promise.allSettled([ + pipeline(archiveStream, inspector), + decompressor.completed, + this.verifyTar(decompressor.stream, inventory), + ]); + let errors = results + .filter(result => result.status === 'rejected') + .map(result => result.reason); + if (errors.length === 1) { + throw errors[0]; + } + if (errors.length > 0) { + throw new AggregateError(errors, 'Could not verify archive'); + } + + if (archiveBytes !== manifest.archive.bytes) { + throw new Error('Archive size does not match its manifest'); + } + if (`sha256:${archiveHash.digest('hex')}` !== manifest.archive.hash) { + throw new Error('Archive SHA-256 does not match its manifest'); + } + } + + matchFiles(manifestFiles, sourceObjects) { + let manifestByPath = new Map(manifestFiles.map(file => [file.path, file])); + let sourceByPath = new Map(sourceObjects.map(object => [object.path, object])); + if (manifestByPath.size !== manifestFiles.length + || sourceByPath.size !== sourceObjects.length + || manifestByPath.size !== sourceByPath.size + || [...manifestByPath].some(([filePath, file]) => { + let source = sourceByPath.get(filePath); + + return !source || source.bytes !== file.bytes; + })) { + throw new Error('Live source files do not match the archive manifest'); + } + + return { manifestByPath, sourceByPath }; + } + + async verifyTar(stream, { manifestByPath, sourceByPath }) { + let extract = tar.extract(); + let verified = new Set; + extract.on('entry', (header, entry, next) => { + this.verifyEntry(header, entry, manifestByPath, sourceByPath, verified) + .then(next) + .catch(error => extract.destroy(error)); + }); + await pipeline(stream, extract); + + if (verified.size !== manifestByPath.size) { + throw new Error('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}`); + } + if (verified.has(header.name)) { + throw new Error(`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}`); + } + verified.add(header.name); + + let bytes = 0; + let md5 = crypto.createHash('md5'); + let sha256 = crypto.createHash('sha256'); + for await (let chunk of entry) { + bytes += chunk.length; + md5.update(chunk); + sha256.update(chunk); + } + + if (bytes !== manifestFile.bytes) { + throw new Error(`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}`); + } + + let etag = sourceObject.etag; + if (etag.startsWith('"') && etag.endsWith('"')) { + 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}`); + } + if (md5.digest('hex') !== etag.toLowerCase()) { + throw new Error(`MD5 does not match the S3 ETag for ${header.name}`); + } + } +} diff --git a/app/data/ArchiveVerifier.test.mjs b/app/data/ArchiveVerifier.test.mjs new file mode 100644 index 0000000..be02155 --- /dev/null +++ b/app/data/ArchiveVerifier.test.mjs @@ -0,0 +1,124 @@ +import { Readable } from 'node:stream'; +import { describe, expect, it } from 'vitest'; +import * as tar from 'tar-stream'; +import ArchiveVerifier from './ArchiveVerifier.mjs'; + +async function createTar(files) { + let pack = tar.pack(); + for (let [name, contents] of files) { + pack.entry({ + mode: 0o644, + mtime: new Date(0), + name, + }, contents); + } + pack.finalize(); + + let chunks = []; + for await (let chunk of pack) { + chunks.push(chunk); + } + + return Buffer.concat(chunks); +} + +function manifestFor(archive, fileHash = 'b6a98d9ce9a2d9149288fa3df42d377c3e42737afdcdaf714e33c0a100b51060') { + return { + archive: { + bytes: archive.length, + hash: 'sha256:5a500b4490cc8d86f7c6970ae673b15933dc1d29f45e6098b730572ec5d2396a', + }, + files: [{ + path: 'alpha.json', + bytes: 6, + hash: `sha256:${fileHash}`, + }], + }; +} + +function sourceObject(etag = '9f9f90dbe3e5ee1218c86b8839db1995') { + return { + path: 'alpha.json', + bytes: 6, + etag: `"${etag}"`, + }; +} + +const identityDecompressor = stream => ({ completed: Promise.resolve(), stream }); + +describe('ArchiveVerifier', () => { + it('verifies a tar stream against its manifest and live S3 objects', async () => { + let archive = await createTar([['alpha.json', 'alpha\n']]); + let verifier = new ArchiveVerifier(identityDecompressor); + + await expect(verifier.verify( + Readable.from(archive), + manifestFor(archive), + [sourceObject()], + )).resolves.toBeUndefined(); + }); + + it('rejects duplicate live S3 paths', async () => { + let archive = await createTar([['alpha.json', 'alpha\n']]); + let source = sourceObject(); + let verifier = new ArchiveVerifier(identityDecompressor); + let archiveStream = new Readable({ + read() { + this.destroy(new Error('Archive should not be read')); + }, + }); + + await expect(verifier.verify( + archiveStream, + manifestFor(archive), + [source, source], + )).rejects.toThrow('Live source files do not match the archive manifest'); + }); + + it('rejects a file whose SHA-256 does not match the manifest', async () => { + let archive = await createTar([['alpha.json', 'alpha\n']]); + let verifier = new ArchiveVerifier(identityDecompressor); + + await expect(verifier.verify( + Readable.from(archive), + manifestFor(archive, '0'.repeat(64)), + [sourceObject()], + )).rejects.toThrow('SHA-256 does not match the manifest for alpha.json'); + }); + + it('rejects an S3 ETag that is not a plain MD5 hash', async () => { + let archive = await createTar([['alpha.json', 'alpha\n']]); + let verifier = new ArchiveVerifier(identityDecompressor); + + await expect(verifier.verify( + Readable.from(archive), + manifestFor(archive), + [sourceObject('multipart-etag-2')], + )).rejects.toThrow('S3 ETag is not an MD5 hash for alpha.json'); + }); + + it('rejects a file that exists only in the tar archive', async () => { + let archive = await createTar([ + ['alpha.json', 'alpha\n'], + ['extra.json', 'extra\n'], + ]); + let verifier = new ArchiveVerifier(identityDecompressor); + + await expect(verifier.verify( + Readable.from(archive), + manifestFor(archive), + [sourceObject()], + )).rejects.toThrow('Tar archive contains an unexpected file: extra.json'); + }); + + it('rejects a file whose MD5 does not match its S3 ETag', async () => { + let archive = await createTar([['alpha.json', 'alpha\n']]); + let verifier = new ArchiveVerifier(identityDecompressor); + + await expect(verifier.verify( + Readable.from(archive), + manifestFor(archive), + [sourceObject('0'.repeat(32))], + )).rejects.toThrow('MD5 does not match the S3 ETag for alpha.json'); + }); +}); diff --git a/app/index.mjs b/app/index.mjs index d9ce218..be96b89 100644 --- a/app/index.mjs +++ b/app/index.mjs @@ -10,6 +10,7 @@ import BlueskyClient from './social/clients/BlueskyClient.mjs'; import ThreadsClient from './social/clients/ThreadsClient.mjs'; import { archiveData } from './data/DataArchiver.mjs'; import { compressArchivesFromCli } from './data/ArchiveCompressor.mjs'; +import { pruneArchivesFromCli } from './data/ArchivePruner.mjs'; import { reportArchiveStatsFromCli } from './data/ArchiveStats.mjs'; import { sentryInit } from './common/sentry.mjs'; import { sync, syncUpload, syncDownload } from './sync/index.mjs'; @@ -31,6 +32,7 @@ const actions = { warmCaches, dataArchive: archiveData, archiveCompress: (...args) => compressArchivesFromCli(args), + archivePrune: (...args) => pruneArchivesFromCli(args), archiveStats: (...args) => reportArchiveStatsFromCli(args), sync, syncUpload, diff --git a/package-lock.json b/package-lock.json index ec33a84..fdea65b 100644 --- a/package-lock.json +++ b/package-lock.json @@ -30,6 +30,7 @@ "s3-sync-client": "^4.3.1", "sharp": "^0.34.5", "sirv": "^3.0.2", + "tar-stream": "^3.2.0", "twitter-api-v2": "^1.29.0", "vue": "^3.5.28", "vue-i18n": "^11.2.8", @@ -5872,11 +5873,10 @@ } }, "node_modules/bare-fs": { - "version": "4.5.4", - "resolved": "https://registry.npmjs.org/bare-fs/-/bare-fs-4.5.4.tgz", - "integrity": "sha512-POK4oplfA7P7gqvetNmCs4CNtm9fNsx+IAh7jH7GgU0OJdge2rso0R20TNWVq6VoWcCvsTdlNDaleLHGaKx8CA==", + "version": "4.8.0", + "resolved": "https://registry.npmjs.org/bare-fs/-/bare-fs-4.8.0.tgz", + "integrity": "sha512-fM+MhCvdQhZ7NV6S95a07gPSqjIYKn6mFaXfx266wN3ajZGl/+1AzH+ubkXQ0fFZvOe2nk9VHkzdYkQE5zMV3Q==", "license": "Apache-2.0", - "optional": true, "dependencies": { "bare-events": "^2.5.4", "bare-path": "^3.0.0", @@ -5885,7 +5885,7 @@ "fast-fifo": "^1.3.2" }, "engines": { - "bare": ">=1.16.0" + "bare": ">=1.28.0" }, "peerDependencies": { "bare-buffer": "*" @@ -5901,7 +5901,6 @@ "resolved": "https://registry.npmjs.org/bare-os/-/bare-os-3.6.2.tgz", "integrity": "sha512-T+V1+1srU2qYNBmJCXZkUY5vQ0B4FSlL3QDROnKQYOqeiQR8UbjNHlPa+TIbM4cuidiN9GaTaOZgSEgsvPbh5A==", "license": "Apache-2.0", - "optional": true, "engines": { "bare": ">=1.14.0" } @@ -5911,7 +5910,6 @@ "resolved": "https://registry.npmjs.org/bare-path/-/bare-path-3.0.0.tgz", "integrity": "sha512-tyfW2cQcB5NN8Saijrhqn0Zh7AnFNsnczRcuWODH0eYAXBsJ5gVxAUuNr7tsHSC6IZ77cA0SitzT+s47kot8Mw==", "license": "Apache-2.0", - "optional": true, "dependencies": { "bare-os": "^3.0.1" } @@ -5921,7 +5919,6 @@ "resolved": "https://registry.npmjs.org/bare-stream/-/bare-stream-2.7.0.tgz", "integrity": "sha512-oyXQNicV1y8nc2aKffH+BUHFRXmx6VrPzlnaEvMhram0nPBrKcEdcyBg5r08D0i8VxngHFAiVyn1QKXpSG0B8A==", "license": "Apache-2.0", - "optional": true, "dependencies": { "streamx": "^2.21.0" }, @@ -5943,7 +5940,6 @@ "resolved": "https://registry.npmjs.org/bare-url/-/bare-url-2.3.2.tgz", "integrity": "sha512-ZMq4gd9ngV5aTMa5p9+UfY0b3skwhHELaDkhEHetMdX0LRkW9kzaym4oo/Eh+Ghm0CCDuMTsRIGM/ytUc1ZYmw==", "license": "Apache-2.0", - "optional": true, "dependencies": { "bare-path": "^3.0.0" } @@ -10914,12 +10910,13 @@ } }, "node_modules/tar-stream": { - "version": "3.1.7", - "resolved": "https://registry.npmjs.org/tar-stream/-/tar-stream-3.1.7.tgz", - "integrity": "sha512-qJj60CXt7IU1Ffyc3NJMjh6EkuCFej46zUqJ4J7pqYlThyd9bO0XBTmcOIhSzZJVWfsLks0+nle/j538YAW9RQ==", + "version": "3.2.0", + "resolved": "https://registry.npmjs.org/tar-stream/-/tar-stream-3.2.0.tgz", + "integrity": "sha512-ojzvCvVaNp6aOTFmG7jaRD0meowIAuPc3cMMhSgKiVWws1GyHbGd/xvnyuRKcKlMpt3qvxx6r0hreCNITP9hIg==", "license": "MIT", "dependencies": { "b4a": "^1.6.4", + "bare-fs": "^4.5.5", "fast-fifo": "^1.2.0", "streamx": "^2.15.0" } diff --git a/package.json b/package.json index d709956..63427ed 100644 --- a/package.json +++ b/package.json @@ -25,6 +25,8 @@ "data:archive": "node app/index.mjs dataArchive", "data:archive:compress": "node app/index.mjs archiveCompress", "data:archive:compress:dry-run": "node app/index.mjs archiveCompress --dry-run", + "data:archive:prune": "node app/index.mjs archivePrune", + "data:archive:prune:dry-run": "node app/index.mjs archivePrune --dry-run", "data:archive:stats": "node app/index.mjs archiveStats", "sync": "node app/index.mjs sync", "sync:upload": "node app/index.mjs syncUpload", @@ -53,6 +55,7 @@ "s3-sync-client": "^4.3.1", "sharp": "^0.34.5", "sirv": "^3.0.2", + "tar-stream": "^3.2.0", "twitter-api-v2": "^1.29.0", "vue": "^3.5.28", "vue-i18n": "^11.2.8",