From ea787f6dbb7a2d97468905d183e7e8a810982585 Mon Sep 17 00:00:00 2001 From: Matt Isenhower Date: Tue, 18 Aug 2026 14:04:25 -0700 Subject: [PATCH 1/7] Add daily archive generator --- app/cron.mjs | 2 + app/data/ArchiveGenerator.mjs | 396 ++++++++++++++++++++++++++++++++++ app/index.mjs | 5 +- docker/app/Dockerfile | 5 + package.json | 1 + 5 files changed, 407 insertions(+), 2 deletions(-) create mode 100644 app/data/ArchiveGenerator.mjs diff --git a/app/cron.mjs b/app/cron.mjs index 2a18eb2..0581c07 100644 --- a/app/cron.mjs +++ b/app/cron.mjs @@ -4,6 +4,7 @@ import { update } from './data/index.mjs'; import { warmCaches } from './splatnet/index.mjs'; import { sendStatuses } from './social/index.mjs'; import { archiveData } from './data/DataArchiver.mjs'; +import { generateArchives } from './data/ArchiveGenerator.mjs'; import { updateAvatars } from './social/updateAvatars.mjs'; let updating = false; @@ -62,5 +63,6 @@ export default function() { }, null, true); new CronJob('30 * * * *', updateAvatars, null, true); + new CronJob('30 0 * * *', () => generateArchives(10), null, true, 'UTC'); new CronJob('0 55 4 * * *', restartToRefreshConfig, null, true); } diff --git a/app/data/ArchiveGenerator.mjs b/app/data/ArchiveGenerator.mjs new file mode 100644 index 0000000..cde67cc --- /dev/null +++ b/app/data/ArchiveGenerator.mjs @@ -0,0 +1,396 @@ +import crypto from 'node:crypto'; +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 { + GetObjectCommand, + ListObjectsV2Command, + PutObjectCommand, + S3Client, +} from '@aws-sdk/client-s3'; +import prefixedConsole from '../common/prefixedConsole.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 generateArchives(maxDays = Infinity) { + let generator = new ArchiveGenerator; + generator.maxDays = maxDays; + + return generator.process(); +} + +export function generateArchivesFromCli(args) { + if (args.length === 0) { + return generateArchives(); + } + + if (args.length !== 2 || args[0] !== '--max-days' || !/^\d+$/.test(args[1])) { + throw new Error('Usage: npm run data:archive:generate -- [--max-days DAYS]'); + } + + let maxDays = Number(args[1]); + if (maxDays < 1) { + throw new Error('--max-days must be at least 1'); + } + + return generateArchives(maxDays); +} + +export default class ArchiveGenerator +{ + maxDays = Infinity; + + async process() { + if (!this.canRun) { + this.console.log('Skipping archive generator'); + + return; + } + + let currentDate; + let generated = 0; + + try { + let candidates = await this.getCandidates(); + this.console.log(`Found ${candidates.length} dates to archive`); + + for (let candidate of candidates) { + if (generated >= this.maxDays) { + break; + } + + currentDate = candidate.date; + if (await this.archiveDate(candidate)) { + generated++; + } + } + } 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; + } + + this.console.log(`Generated ${generated} archives`); + } + + // Properties + + get console() { + this._console ??= prefixedConsole('Archive Generator'); + + 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, + }, + }); + } + + // Archive generation + + async archiveDate(candidate) { + let objects = await this.getSourceObjects(candidate.prefix); + if (objects.length === 0) { + return false; + } + + let newestObject = Math.max(...objects.map(object => object.lastModified.getTime())); + if (newestObject > Date.now() - quietPeriod) { + this.console.log(`Skipping ${candidate.date}; its source files are still changing`); + + return false; + } + + let temporaryDirectory = await fs.mkdtemp( + path.join(os.tmpdir(), `splatoon3ink-${candidate.date}-`), + ); + let sourceDirectory = path.join(temporaryDirectory, 'source'); + let archivePath = path.join(temporaryDirectory, `${candidate.date}.tar.zst`); + + try { + await fs.mkdir(sourceDirectory); + this.console.log(`Downloading ${objects.length} files for ${candidate.date}`); + let files = await this.downloadObjects(candidate.prefix, objects, sourceDirectory); + 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)); + + 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, + }; + + this.console.log(`Uploading ${candidate.archiveKey}`); + await this.upload(candidate.archiveKey, await fs.readFile(archivePath), 'application/zstd'); + await this.upload( + candidate.manifestKey, + Buffer.from(`${JSON.stringify(manifest, null, 2)}\n`), + 'application/json', + ); + + return true; + } finally { + await fs.rm(temporaryDirectory, { force: true, recursive: true }); + } + } + + async downloadObjects(prefix, objects, sourceDirectory) { + let files = []; + let nextObject = 0; + let error; + let worker = async () => { + while (!error && nextObject < objects.length) { + let index = nextObject++; + try { + files[index] = await this.downloadObject(prefix, objects[index], sourceDirectory); + } catch (e) { + error ??= e; + } + } + }; + + let workers = Array.from({ length: Math.min(downloadLimit, objects.length) }, worker); + await Promise.all(workers); + if (error) { + throw error; + } + + return files; + } + + async downloadObject(prefix, object, sourceDirectory) { + let relativePath = this.relativePath(prefix, object.key); + let destination = path.join(sourceDirectory, relativePath); + await fs.mkdir(path.dirname(destination), { recursive: true }); + + let response = await this.s3Client.send(new GetObjectCommand({ + Bucket: process.env.AWS_S3_ARCHIVE_BUCKET, + Key: object.key, + })); + if (!response.Body) { + throw new Error(`S3 returned no body for ${object.key}`); + } + + await pipeline(response.Body, createWriteStream(destination, { flags: 'wx' })); + let bytes = (await fs.stat(destination)).size; + if (bytes !== object.bytes) { + throw new Error(`Downloaded size does not match S3 listing for ${object.key}`); + } + + return { + path: relativePath, + bytes, + hash: `sha256:${await this.hashFile(destination)}`, + }; + } + + 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)) { + hash.update(chunk); + } + + return hash.digest('hex'); + } + + async upload(key, body, contentType) { + await this.s3Client.send(new PutObjectCommand({ + ACL: 'public-read', + Body: body, + Bucket: process.env.AWS_S3_ARCHIVE_BUCKET, + ContentLength: body.length, + ContentType: contentType, + Key: key, + })); + } + + // S3 discovery + + 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 || date >= new Date().toISOString().slice(0, 10)) { + continue; + } + + let archiveKey = `${month}${date}.tar.zst`; + let manifestKey = `${archiveKey}.manifest.json`; + if (existing.has(archiveKey) && existing.has(manifestKey)) { + continue; + } + + candidates.push({ archiveKey, date, manifestKey, prefix }); + } + } + } + + return candidates.sort((a, b) => a.date.localeCompare(b.date)); + } + + 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 (!object.LastModified || !Number.isFinite(object.Size)) { + throw new Error(`S3 returned incomplete metadata for ${object.Key}`); + } + + return { + key: object.Key, + bytes: object.Size, + lastModified: object.LastModified, + }; + }); + } + + 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 }; + } + + // Helpers + + 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`); + if (Number.isNaN(parsedDate.getTime()) || parsedDate.toISOString().slice(0, 10) !== date) { + return null; + } + + return date; + } + + 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/index.mjs b/app/index.mjs index 67bb15e..49dadbb 100644 --- a/app/index.mjs +++ b/app/index.mjs @@ -9,6 +9,7 @@ import ImageWriter from './social/clients/ImageWriter.mjs'; import BlueskyClient from './social/clients/BlueskyClient.mjs'; import ThreadsClient from './social/clients/ThreadsClient.mjs'; import { archiveData } from './data/DataArchiver.mjs'; +import { generateArchivesFromCli } from './data/ArchiveGenerator.mjs'; import { sentryInit } from './common/sentry.mjs'; import { sync, syncUpload, syncDownload } from './sync/index.mjs'; import { updateAvatars } from './social/updateAvatars.mjs'; @@ -28,6 +29,7 @@ const actions = { splatnet: update, warmCaches, dataArchive: archiveData, + archiveGenerate: (...args) => generateArchivesFromCli(args), sync, syncUpload, syncDownload, @@ -38,8 +40,7 @@ const command = process.argv[2]; const params = process.argv.slice(3); const action = actions[command]; if (action) { - action(...params); + await action(...params); } else { console.error(`Unrecognized command: ${command}`); } - diff --git a/docker/app/Dockerfile b/docker/app/Dockerfile index 412506b..5367c78 100644 --- a/docker/app/Dockerfile +++ b/docker/app/Dockerfile @@ -4,6 +4,11 @@ FROM node:22 WORKDIR /app ENV PUPPETEER_SKIP_DOWNLOAD=true +# Archive compression +RUN apt-get update \ + && apt-get install -y --no-install-recommends zstd \ + && rm -rf /var/lib/apt/lists/* + # Install NPM dependencies COPY package*.json ./ RUN npm ci diff --git a/package.json b/package.json index a98b70b..2b67305 100644 --- a/package.json +++ b/package.json @@ -23,6 +23,7 @@ "splatnet:all": "node app/index.mjs splatnet all", "warmCaches": "node app/index.mjs warmCaches", "data:archive": "node app/index.mjs dataArchive", + "data:archive:generate": "node app/index.mjs archiveGenerate", "sync": "node app/index.mjs sync", "sync:upload": "node app/index.mjs syncUpload", "sync:download": "node app/index.mjs syncDownload" From e690a10da8b746906d8c8c8c5a66d921c4a05aff Mon Sep 17 00:00:00 2001 From: Matt Isenhower Date: Tue, 18 Aug 2026 14:12:15 -0700 Subject: [PATCH 2/7] Add archive generator dry run --- app/data/ArchiveGenerator.mjs | 70 ++++++++++++++++++++++++++--------- package.json | 1 + 2 files changed, 54 insertions(+), 17 deletions(-) diff --git a/app/data/ArchiveGenerator.mjs b/app/data/ArchiveGenerator.mjs index cde67cc..4cfe2c3 100644 --- a/app/data/ArchiveGenerator.mjs +++ b/app/data/ArchiveGenerator.mjs @@ -38,32 +38,40 @@ function processCompletion(child, name) { }); } -export function generateArchives(maxDays = Infinity) { +export function generateArchives(maxDays = Infinity, dryRun = false) { let generator = new ArchiveGenerator; generator.maxDays = maxDays; + generator.dryRun = dryRun; return generator.process(); } export function generateArchivesFromCli(args) { - if (args.length === 0) { - return generateArchives(); + 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:generate -- [--dry-run] [--max-days DAYS]', + ); + } } - if (args.length !== 2 || args[0] !== '--max-days' || !/^\d+$/.test(args[1])) { - throw new Error('Usage: npm run data:archive:generate -- [--max-days DAYS]'); - } - - let maxDays = Number(args[1]); if (maxDays < 1) { throw new Error('--max-days must be at least 1'); } - return generateArchives(maxDays); + return generateArchives(maxDays, dryRun); } export default class ArchiveGenerator { + dryRun = false; maxDays = Infinity; async process() { @@ -86,7 +94,10 @@ export default class ArchiveGenerator } currentDate = candidate.date; - if (await this.archiveDate(candidate)) { + let completed = this.dryRun + ? await this.previewDate(candidate) + : await this.archiveDate(candidate); + if (completed) { generated++; } } @@ -102,7 +113,9 @@ export default class ArchiveGenerator throw e; } - this.console.log(`Generated ${generated} archives`); + this.console.log( + this.dryRun ? `Would generate ${generated} archives` : `Generated ${generated} archives`, + ); } // Properties @@ -136,16 +149,23 @@ export default class ArchiveGenerator // Archive generation - async archiveDate(candidate) { - let objects = await this.getSourceObjects(candidate.prefix); - if (objects.length === 0) { + async previewDate(candidate) { + let objects = await this.getReadyObjects(candidate); + if (!objects) { return false; } - let newestObject = Math.max(...objects.map(object => object.lastModified.getTime())); - if (newestObject > Date.now() - quietPeriod) { - this.console.log(`Skipping ${candidate.date}; its source files are still changing`); + this.console.log( + `Would generate ${candidate.archiveKey} and ${candidate.manifestKey} ` + + `from ${objects.length} files`, + ); + return true; + } + + async archiveDate(candidate) { + let objects = await this.getReadyObjects(candidate); + if (!objects) { return false; } @@ -191,6 +211,22 @@ export default class ArchiveGenerator } } + async getReadyObjects(candidate) { + let objects = await this.getSourceObjects(candidate.prefix); + if (objects.length === 0) { + return null; + } + + let newestObject = Math.max(...objects.map(object => object.lastModified.getTime())); + if (newestObject > Date.now() - quietPeriod) { + this.console.log(`Skipping ${candidate.date}; its source files are still changing`); + + return null; + } + + return objects; + } + async downloadObjects(prefix, objects, sourceDirectory) { let files = []; let nextObject = 0; diff --git a/package.json b/package.json index 2b67305..bde0867 100644 --- a/package.json +++ b/package.json @@ -24,6 +24,7 @@ "warmCaches": "node app/index.mjs warmCaches", "data:archive": "node app/index.mjs dataArchive", "data:archive:generate": "node app/index.mjs archiveGenerate", + "data:archive:generate:dry-run": "node app/index.mjs archiveGenerate --dry-run", "sync": "node app/index.mjs sync", "sync:upload": "node app/index.mjs syncUpload", "sync:download": "node app/index.mjs syncDownload" From aa7dfdfc4b620cc1c63677eba06ad83403d1cc75 Mon Sep 17 00:00:00 2001 From: Matt Isenhower Date: Tue, 18 Aug 2026 19:50:12 -0700 Subject: [PATCH 3/7] Rename archive generator to compressor --- app/cron.mjs | 4 +-- ...iveGenerator.mjs => ArchiveCompressor.mjs} | 36 ++++++++++--------- app/index.mjs | 4 +-- package.json | 4 +-- 4 files changed, 25 insertions(+), 23 deletions(-) rename app/data/{ArchiveGenerator.mjs => ArchiveCompressor.mjs} (93%) diff --git a/app/cron.mjs b/app/cron.mjs index 0581c07..e20ad3e 100644 --- a/app/cron.mjs +++ b/app/cron.mjs @@ -4,7 +4,7 @@ import { update } from './data/index.mjs'; import { warmCaches } from './splatnet/index.mjs'; import { sendStatuses } from './social/index.mjs'; import { archiveData } from './data/DataArchiver.mjs'; -import { generateArchives } from './data/ArchiveGenerator.mjs'; +import { compressArchives } from './data/ArchiveCompressor.mjs'; import { updateAvatars } from './social/updateAvatars.mjs'; let updating = false; @@ -63,6 +63,6 @@ export default function() { }, null, true); new CronJob('30 * * * *', updateAvatars, null, true); - new CronJob('30 0 * * *', () => generateArchives(10), null, true, 'UTC'); + new CronJob('30 0 * * *', () => compressArchives(10), null, true, 'UTC'); new CronJob('0 55 4 * * *', restartToRefreshConfig, null, true); } diff --git a/app/data/ArchiveGenerator.mjs b/app/data/ArchiveCompressor.mjs similarity index 93% rename from app/data/ArchiveGenerator.mjs rename to app/data/ArchiveCompressor.mjs index 4cfe2c3..1b0e667 100644 --- a/app/data/ArchiveGenerator.mjs +++ b/app/data/ArchiveCompressor.mjs @@ -38,15 +38,15 @@ function processCompletion(child, name) { }); } -export function generateArchives(maxDays = Infinity, dryRun = false) { - let generator = new ArchiveGenerator; - generator.maxDays = maxDays; - generator.dryRun = dryRun; +export function compressArchives(maxDays = Infinity, dryRun = false) { + let compressor = new ArchiveCompressor; + compressor.maxDays = maxDays; + compressor.dryRun = dryRun; - return generator.process(); + return compressor.process(); } -export function generateArchivesFromCli(args) { +export function compressArchivesFromCli(args) { let dryRun = false; let maxDays = Infinity; @@ -57,7 +57,7 @@ export function generateArchivesFromCli(args) { maxDays = Number(args[++i]); } else { throw new Error( - 'Usage: npm run data:archive:generate -- [--dry-run] [--max-days DAYS]', + 'Usage: npm run data:archive:compress -- [--dry-run] [--max-days DAYS]', ); } } @@ -66,30 +66,30 @@ export function generateArchivesFromCli(args) { throw new Error('--max-days must be at least 1'); } - return generateArchives(maxDays, dryRun); + return compressArchives(maxDays, dryRun); } -export default class ArchiveGenerator +export default class ArchiveCompressor { dryRun = false; maxDays = Infinity; async process() { if (!this.canRun) { - this.console.log('Skipping archive generator'); + this.console.log('Skipping archive compressor'); return; } let currentDate; - let generated = 0; + let compressed = 0; try { let candidates = await this.getCandidates(); this.console.log(`Found ${candidates.length} dates to archive`); for (let candidate of candidates) { - if (generated >= this.maxDays) { + if (compressed >= this.maxDays) { break; } @@ -98,7 +98,7 @@ export default class ArchiveGenerator ? await this.previewDate(candidate) : await this.archiveDate(candidate); if (completed) { - generated++; + compressed++; } } } catch (e) { @@ -114,14 +114,16 @@ export default class ArchiveGenerator } this.console.log( - this.dryRun ? `Would generate ${generated} archives` : `Generated ${generated} archives`, + this.dryRun + ? `Would compress ${compressed} daily archives` + : `Compressed ${compressed} daily archives`, ); } // Properties get console() { - this._console ??= prefixedConsole('Archive Generator'); + this._console ??= prefixedConsole('Archive Compressor'); return this._console; } @@ -147,7 +149,7 @@ export default class ArchiveGenerator }); } - // Archive generation + // Archive compression async previewDate(candidate) { let objects = await this.getReadyObjects(candidate); @@ -156,7 +158,7 @@ export default class ArchiveGenerator } this.console.log( - `Would generate ${candidate.archiveKey} and ${candidate.manifestKey} ` + `Would create ${candidate.archiveKey} and ${candidate.manifestKey} ` + `from ${objects.length} files`, ); diff --git a/app/index.mjs b/app/index.mjs index 49dadbb..cb1f4b7 100644 --- a/app/index.mjs +++ b/app/index.mjs @@ -9,7 +9,7 @@ import ImageWriter from './social/clients/ImageWriter.mjs'; import BlueskyClient from './social/clients/BlueskyClient.mjs'; import ThreadsClient from './social/clients/ThreadsClient.mjs'; import { archiveData } from './data/DataArchiver.mjs'; -import { generateArchivesFromCli } from './data/ArchiveGenerator.mjs'; +import { compressArchivesFromCli } from './data/ArchiveCompressor.mjs'; import { sentryInit } from './common/sentry.mjs'; import { sync, syncUpload, syncDownload } from './sync/index.mjs'; import { updateAvatars } from './social/updateAvatars.mjs'; @@ -29,7 +29,7 @@ const actions = { splatnet: update, warmCaches, dataArchive: archiveData, - archiveGenerate: (...args) => generateArchivesFromCli(args), + archiveCompress: (...args) => compressArchivesFromCli(args), sync, syncUpload, syncDownload, diff --git a/package.json b/package.json index bde0867..6297c02 100644 --- a/package.json +++ b/package.json @@ -23,8 +23,8 @@ "splatnet:all": "node app/index.mjs splatnet all", "warmCaches": "node app/index.mjs warmCaches", "data:archive": "node app/index.mjs dataArchive", - "data:archive:generate": "node app/index.mjs archiveGenerate", - "data:archive:generate:dry-run": "node app/index.mjs archiveGenerate --dry-run", + "data:archive:compress": "node app/index.mjs archiveCompress", + "data:archive:compress:dry-run": "node app/index.mjs archiveCompress --dry-run", "sync": "node app/index.mjs sync", "sync:upload": "node app/index.mjs syncUpload", "sync:download": "node app/index.mjs syncDownload" From 660eb824e9f9a6f900cbe374acc1a5483affabf4 Mon Sep 17 00:00:00 2001 From: Matt Isenhower Date: Tue, 18 Aug 2026 22:54:23 -0700 Subject: [PATCH 4/7] Add archive compression statistics --- app/data/ArchiveStats.mjs | 200 ++++++++++++++++++++++++++++++++++++++ app/index.mjs | 2 + package.json | 1 + 3 files changed, 203 insertions(+) create mode 100644 app/data/ArchiveStats.mjs diff --git a/app/data/ArchiveStats.mjs b/app/data/ArchiveStats.mjs new file mode 100644 index 0000000..192cc7c --- /dev/null +++ b/app/data/ArchiveStats.mjs @@ -0,0 +1,200 @@ +import { + GetObjectCommand, + ListObjectsV2Command, + S3Client, +} from '@aws-sdk/client-s3'; +import pLimit from 'p-limit'; +import prefixedConsole from '../common/prefixedConsole.mjs'; + +const manifestSuffix = '.tar.zst.manifest.json'; +const requestLimit = pLimit(10); + +export function reportArchiveStats(verbose = false) { + let stats = new ArchiveStats; + stats.verbose = verbose; + + return stats.process(); +} + +export function reportArchiveStatsFromCli(args) { + if (args.some(argument => argument !== '--verbose')) { + throw new Error('Usage: npm run data:archive:stats -- [--verbose]'); + } + + return reportArchiveStats(args.includes('--verbose')); +} + +export default class ArchiveStats +{ + verbose = false; + + async process() { + if (!this.canRun) { + this.console.log('Skipping archive stats'); + + return; + } + + this.console.log('Reading archive manifests...'); + let keys = await this.getManifestKeys(); + let archives = await Promise.all(keys.map(key => requestLimit(() => this.readManifest(key)))); + archives.sort((a, b) => a.path.localeCompare(b.path)); + + if (this.verbose) { + for (let archive of archives) { + this.console.log(this.describeArchive(archive)); + } + } + + let originalBytes = archives.reduce((total, archive) => total + archive.originalBytes, 0); + let compressedBytes = archives.reduce((total, archive) => total + archive.compressedBytes, 0); + let fileCount = archives.reduce((total, archive) => total + archive.fileCount, 0); + let savedBytes = originalBytes - compressedBytes; + + this.console.log( + `${archives.length} archives containing ${fileCount} files: ` + + `${this.formatBytes(originalBytes)} -> ${this.formatBytes(compressedBytes)}; ` + + `saved ${this.formatBytes(savedBytes)} (${this.formatPercent(savedBytes, originalBytes)}), ` + + `${this.formatRatio(originalBytes, compressedBytes)}:1 compression`, + ); + } + + // Properties + + get console() { + this._console ??= prefixedConsole('Archive Stats'); + + 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, + }, + }); + } + + // Manifests + + async getManifestKeys() { + let years = (await this.list('', '/')).prefixes.filter(prefix => /^\d{4}\/$/.test(prefix)); + let yearListings = await Promise.all(years.map(year => { + return requestLimit(() => this.list(year, '/')); + })); + let months = yearListings.flatMap(listing => listing.prefixes) + .filter(prefix => /^\d{4}\/\d{2}\/$/.test(prefix)); + let monthListings = await Promise.all(months.map(month => { + return requestLimit(() => this.list(month, '/')); + })); + + return monthListings.flatMap(listing => listing.objects) + .map(object => object.Key) + .filter(key => key && key.endsWith(manifestSuffix)); + } + + async readManifest(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}`); + } + + 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}`); + } + + return { + path: manifest.archive.path, + compressedBytes: manifest.archive.bytes, + fileCount: manifest.files.length, + originalBytes: manifest.files.reduce((total, file) => total + file.bytes, 0), + }; + } + + 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 }; + } + + // Formatting + + describeArchive(archive) { + let savedBytes = archive.originalBytes - archive.compressedBytes; + + return `${archive.path}: ${archive.fileCount} ${archive.fileCount === 1 ? 'file' : 'files'}, ` + + `${this.formatBytes(archive.originalBytes)} -> ${this.formatBytes(archive.compressedBytes)}, ` + + `${this.formatPercent(savedBytes, archive.originalBytes)} saved, ` + + `${this.formatRatio(archive.originalBytes, archive.compressedBytes)}:1`; + } + + formatBytes(bytes) { + let units = ['B', 'KB', 'MB', 'GB', 'TB']; + let unit = 0; + let value = bytes; + + while (Math.abs(value) >= 1000 && unit < units.length - 1) { + value /= 1000; + unit++; + } + + return `${new Intl.NumberFormat('en-US', { maximumFractionDigits: 2 }).format(value)} ${units[unit]}`; + } + + formatPercent(part, total) { + if (total === 0) { + return '0%'; + } + + return `${new Intl.NumberFormat('en-US', { maximumFractionDigits: 2 }).format(part / total * 100)}%`; + } + + formatRatio(originalBytes, compressedBytes) { + if (compressedBytes === 0) { + return '0'; + } + + return new Intl.NumberFormat('en-US', { maximumFractionDigits: 1 }) + .format(originalBytes / compressedBytes); + } +} diff --git a/app/index.mjs b/app/index.mjs index cb1f4b7..d9ce218 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 { reportArchiveStatsFromCli } from './data/ArchiveStats.mjs'; import { sentryInit } from './common/sentry.mjs'; import { sync, syncUpload, syncDownload } from './sync/index.mjs'; import { updateAvatars } from './social/updateAvatars.mjs'; @@ -30,6 +31,7 @@ const actions = { warmCaches, dataArchive: archiveData, archiveCompress: (...args) => compressArchivesFromCli(args), + archiveStats: (...args) => reportArchiveStatsFromCli(args), sync, syncUpload, syncDownload, diff --git a/package.json b/package.json index 6297c02..d709956 100644 --- a/package.json +++ b/package.json @@ -25,6 +25,7 @@ "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:stats": "node app/index.mjs archiveStats", "sync": "node app/index.mjs sync", "sync:upload": "node app/index.mjs syncUpload", "sync:download": "node app/index.mjs syncDownload" From 7c809f86ff65671dc9e2764ce8894e73bf10dec7 Mon Sep 17 00:00:00 2001 From: Matt Isenhower Date: Tue, 18 Aug 2026 23:12:11 -0700 Subject: [PATCH 5/7] 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)}`); + } + } +} From 2f6dcee0eae6afcf1bc235004d3bb7f3f5dcf5c8 Mon Sep 17 00:00:00 2001 From: Matt Isenhower Date: Thu, 20 Aug 2026 14:29:02 -0700 Subject: [PATCH 6/7] Clarify archive compression statistics --- app/data/ArchiveStats.mjs | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/app/data/ArchiveStats.mjs b/app/data/ArchiveStats.mjs index 7fbda56..762924f 100644 --- a/app/data/ArchiveStats.mjs +++ b/app/data/ArchiveStats.mjs @@ -54,8 +54,9 @@ export default class ArchiveStats this.console.log( `${archives.length} archives containing ${fileCount} files: ` - + `${this.formatBytes(originalBytes)} -> ${this.formatBytes(compressedBytes)}; ` - + `saved ${this.formatBytes(savedBytes)} (${this.formatPercent(savedBytes, originalBytes)}), ` + + `${this.formatBytes(originalBytes)} source data -> ${this.formatBytes(compressedBytes)} compressed; ` + + `projected source-file savings after pruning ${this.formatBytes(savedBytes)} ` + + `(${this.formatPercent(savedBytes, originalBytes)}; manifests excluded), ` + `${this.formatRatio(originalBytes, compressedBytes)}:1 compression`, ); } @@ -158,12 +159,12 @@ export default class ArchiveStats } formatBytes(bytes) { - let units = ['B', 'KB', 'MB', 'GB', 'TB']; + let units = ['B', 'KiB', 'MiB', 'GiB', 'TiB']; let unit = 0; let value = bytes; - while (Math.abs(value) >= 1000 && unit < units.length - 1) { - value /= 1000; + while (Math.abs(value) >= 1024 && unit < units.length - 1) { + value /= 1024; unit++; } From 57ed36eb72d1d12dc9e3be832d670da23e8d4b92 Mon Sep 17 00:00:00 2001 From: Matt Isenhower Date: Thu, 20 Aug 2026 14:35:29 -0700 Subject: [PATCH 7/7] Fix flaky archive download assertion --- app/data/ArchiveCompressor.test.mjs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/app/data/ArchiveCompressor.test.mjs b/app/data/ArchiveCompressor.test.mjs index a93ee11..597ca97 100644 --- a/app/data/ArchiveCompressor.test.mjs +++ b/app/data/ArchiveCompressor.test.mjs @@ -200,7 +200,7 @@ describe('ArchiveCompressor', () => { await expect(compressor.process()).rejects.toThrow('compression failed'); - expect(s3Client.downloads).toEqual([ + expect(s3Client.downloads.toSorted()).toEqual([ '2025/03/05/alpha.json', '2025/03/05/nested/beta.json', ]);