scripts/backfill_manifest_file_sizes.mjs
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env bun
/**
* Backfill manifest.file_size from trusted object storage metadata.
*
* Dry-run by default:
* bun scripts/backfill_manifest_file_sizes.mjs --app-version-id=180988804
*
* Apply updates:
* bun scripts/backfill_manifest_file_sizes.mjs --app-version-id=180988804 --apply
* bun scripts/backfill_manifest_file_sizes.mjs --all --apply --workers=16
*/
import { appendFileSync, existsSync, mkdirSync, readFileSync, writeFileSync } from 'node:fs'
import { dirname, resolve } from 'node:path'
import process from 'node:process'
import { fileURLToPath } from 'node:url'
import { S3Client } from '@bradenmacdonald/s3-lite-client'
import { parse } from 'dotenv'
import pg from 'pg'
const __dirname = dirname(fileURLToPath(import.meta.url))
function hasFlag(name) {
return process.argv.includes(name)
}
function getArgValue(name) {
const prefix = `${name}=`
const match = process.argv.find(arg => arg.startsWith(prefix))
if (match) {
const value = match.slice(prefix.length)
if (!value)
throw new Error(`${name} requires a value`)
return value
}
const index = process.argv.indexOf(name)
if (index !== -1) {
const value = process.argv[index + 1]
if (!value || value.startsWith('--'))
throw new Error(`${name} requires a value`)
return value
}
return undefined
}
function getTarget() {
if (hasFlag('--help') || hasFlag('-h'))
return 'prod'
const target = getArgValue('--target') ?? (hasFlag('--local') ? 'local' : 'prod')
if (target !== 'prod' && target !== 'local')
throw new Error('--target must be "prod" or "local"')
return target
}
const target = getTarget()
const sharedEnvPaths = [
'../.env',
]
const targetEnvPaths = target === 'prod'
? [
'../internal/cloudflare/.env.prod',
]
: [
'../.env.local',
'../internal/cloudflare/.env.local',
]
function loadEnvFiles(envPaths) {
const loaded = {}
for (const envPath of envPaths) {
const resolvedPath = resolve(__dirname, envPath)
if (!existsSync(resolvedPath))
continue
Object.assign(loaded, parse(readFileSync(resolvedPath)))
}
return loaded
}
const targetEnv = loadEnvFiles(targetEnvPaths)
const runtimeEnv = loadEnvFiles([...sharedEnvPaths, ...targetEnvPaths])
for (const [key, value] of Object.entries(runtimeEnv)) {
process.env[key] = value
}
const DB_URL_ENV_KEYS = [
'MAIN_SUPABASE_DB_URL',
'DATABASE_URL',
'POSTGRES_URL',
'SUPABASE_DB_URL',
'SUPABASE_DB_DIRECT_URL',
'DIRECT_URL',
]
const MAX_CANDIDATE_BATCH_SIZE = 1000
const PERCENT_ENCODED_OCTET_RE = /%[0-9a-f]{2}/i
const FAILED_CSV_HEADERS = [
'id',
'app_id',
'app_version_id',
'version_name',
'file_name',
's3_path',
'attempted_s3_path',
'attempted_s3_paths',
'status',
'method',
'reason',
'attempts',
'error_name',
'error_status',
'error_message',
]
function sleep(ms) {
return new Promise(resolve => setTimeout(resolve, ms))
}
function getNumberArg(name, fallback) {
const value = getArgValue(name)
if (value === undefined)
return fallback
const parsed = Number.parseInt(value, 10)
if (!Number.isFinite(parsed) || parsed <= 0)
throw new Error(`${name} must be a positive integer`)
return parsed
}
function getOptionalNumberArg(name) {
const value = getArgValue(name)
if (value === undefined)
return null
const parsed = Number.parseInt(value, 10)
if (!Number.isFinite(parsed) || parsed < 0)
throw new Error(`${name} must be a positive integer`)
return parsed
}
function getFixedBatchSizeArg() {
const value = getArgValue('--batch-size')
if (value === undefined)
return MAX_CANDIDATE_BATCH_SIZE
const parsed = Number.parseInt(value, 10)
if (!Number.isFinite(parsed) || parsed <= 0)
throw new Error('--batch-size must be a positive integer')
if (parsed !== MAX_CANDIDATE_BATCH_SIZE)
console.warn(`--batch-size=${parsed} is ignored; candidate reads are fixed to ${MAX_CANDIDATE_BATCH_SIZE}`)
return MAX_CANDIDATE_BATCH_SIZE
}
function getDatabaseUrl(databaseEnv) {
for (const key of DB_URL_ENV_KEYS) {
const value = databaseEnv[key]
if (value)
return value
}
throw new Error(`Missing Postgres URL in ${targetEnvPaths.join(', ')}. Set one of: ${DB_URL_ENV_KEYS.join(', ')}`)
}
function isLocalDatabaseUrl(databaseUrl) {
try {
let { hostname } = new URL(databaseUrl)
if (hostname.startsWith('[') && hostname.endsWith(']'))
hostname = hostname.slice(1, -1)
return ['127.0.0.1', 'localhost', '::1'].includes(hostname)
}
catch {
return databaseUrl.includes('127.0.0.1')
|| databaseUrl.includes('localhost')
|| databaseUrl.includes('::1')
|| databaseUrl.includes('[::1]')
}
}
function getSafeDatabaseUrl() {
const databaseUrl = getDatabaseUrl(targetEnv)
if (target === 'prod' && isLocalDatabaseUrl(databaseUrl)) {
throw new Error('Refusing to use a local Postgres URL for the default prod target. Pass --target=local only when you intentionally want local.')
}
return databaseUrl
}
function describeDatabaseUrl(databaseUrl) {
try {
const { host } = new URL(databaseUrl)
return host
}
catch {
return 'unknown host'
}
}
function getRequiredEnv(name) {
const value = process.env[name]
if (!value)
throw new Error(`Missing env var: ${name}`)
return value
}
function getStorageEndpoint() {
const endpoint = getRequiredEnv('S3_ENDPOINT')
return endpoint.includes('://') ? endpoint : `https://${endpoint}`
}
function parseObjectSizeFromHeaders(contentRange, contentLength) {
if (contentRange && contentRange.includes('/')) {
const total = Number.parseInt(contentRange.split('/').at(1) ?? '0', 10)
if (Number.isFinite(total) && total > 0)
return total
}
if (contentLength) {
const len = Number.parseInt(contentLength, 10)
if (Number.isFinite(len) && len > 0)
return len
}
return 0
}
function serializeError(error) {
if (error instanceof Error) {
return {
message: error.message,
name: error.name,
status: error.status ?? error.statusCode ?? error.code,
}
}
return { message: String(error) }
}
function decodeStoragePathSegments(s3Path) {
try {
return s3Path.split('/').map(segment => decodeURIComponent(segment)).join('/')
}
catch {
return null
}
}
function encodeStoragePathSegments(s3Path) {
return s3Path.split('/').map(segment => encodeURIComponent(segment)).join('/')
}
function getStorageCandidatePaths(s3Path) {
const candidates = [s3Path]
if (PERCENT_ENCODED_OCTET_RE.test(s3Path)) {
const decoded = decodeStoragePathSegments(s3Path)
if (decoded && decoded !== s3Path)
candidates.push(decoded)
const encoded = encodeStoragePathSegments(s3Path)
if (encoded !== s3Path)
candidates.push(encoded)
}
return [...new Set(candidates)]
}
function markStorageCandidateResult(result, attemptedS3Path, attemptedS3Paths) {
const suffix = attemptedS3Path === attemptedS3Paths[0] ? '' : '_candidate'
return {
...result,
attempted_s3_path: attemptedS3Path,
attempted_s3_paths: attemptedS3Paths,
method: `${result.method ?? 'unknown'}${suffix}`,
reason: suffix && result.reason ? `${result.reason}${suffix}` : result.reason,
}
}
async function getObjectSizeWithLegacyFallback(s3, s3Path) {
const candidatePaths = getStorageCandidatePaths(s3Path)
let lastResult = null
for (const candidatePath of candidatePaths) {
const result = markStorageCandidateResult(await getObjectSize(s3, candidatePath), candidatePath, candidatePaths)
lastResult = result
if (result.size > 0 || shouldRetryStorageResult(result))
return result
}
return lastResult
}
async function getObjectSize(s3, s3Path) {
try {
const stat = await s3.statObject(s3Path, {
headers: { 'Accept-Encoding': 'identity' },
})
const size = Number.isFinite(stat.size) ? stat.size : 0
if (size > 0)
return { method: 'head', size }
}
catch (error) {
const rangeResult = await getObjectSizeWithRange(s3, s3Path, 'head_error')
if (rangeResult.size > 0)
return rangeResult
return { ...rangeResult, error: serializeError(error) }
}
return await getObjectSizeWithRange(s3, s3Path, 'missing_head_size')
}
function shouldRetryStorageResult(result) {
if (result.size > 0)
return false
if (result.status === 404)
return false
return true
}
async function getObjectSizeWithRetry(s3, s3Path, attempts) {
let lastResult = null
for (let attempt = 1; attempt <= attempts; attempt++) {
const result = await getObjectSizeWithLegacyFallback(s3, s3Path)
lastResult = { ...result, attempts: attempt }
if (!shouldRetryStorageResult(lastResult))
return lastResult
if (attempt < attempts)
await sleep(250 * attempt)
}
return lastResult ?? { attempts: 0, method: 'unknown', size: 0 }
}
async function getObjectSizeWithRange(s3, s3Path, reason) {
try {
const url = await s3.getPresignedUrl('GET', s3Path, {
parameters: { 'x-id': 'GetObject' },
})
const response = await fetch(url, {
method: 'GET',
headers: {
'Accept-Encoding': 'identity',
'Range': 'bytes=0-0',
},
})
const contentRange = response.headers.get('content-range') || response.headers.get('Content-Range')
const contentLength = response.headers.get('content-length') || response.headers.get('Content-Length')
const size = response.status === 206 && contentRange
? parseObjectSizeFromHeaders(contentRange, null)
: response.status === 200
? parseObjectSizeFromHeaders(null, contentLength)
: 0
await response.body?.cancel()
return {
contentLength,
contentRange,
method: 'range',
reason,
size,
status: response.status,
statusText: response.statusText,
}
}
catch (error) {
return {
error: serializeError(error),
method: 'range',
reason,
size: 0,
}
}
}
function isRetryableDatabaseError(error) {
const code = error?.code ?? error?.errno
const message = String(error?.message ?? '')
const retryableCodes = [
'40001',
'40P01',
'53300',
'57P01',
'57P02',
'57P03',
'ECONNRESET',
'ECONNREFUSED',
'ETIMEDOUT',
'EPIPE',
]
return retryableCodes.includes(code) || /Connection terminated|timeout|ECONNRESET|ECONNREFUSED|ETIMEDOUT|terminating connection/i.test(message)
}
async function queryWithRetry(pool, sql, params, label, attempts) {
let lastError = null
for (let attempt = 1; attempt <= attempts; attempt++) {
try {
return await pool.query(sql, params)
}
catch (error) {
lastError = error
if (!isRetryableDatabaseError(error) || attempt >= attempts)
break
console.warn(`${label} failed, retrying ${attempt}/${attempts}: ${error.message}`)
await sleep(500 * attempt)
}
}
throw lastError
}
function appendCommonFilters(where, params, { appId, appVersionId, endId, includeDeleted, startId }) {
if (startId !== undefined && startId !== null) {
params.push(startId)
where.push(`m.id > $${params.length}`)
}
if (endId !== undefined && endId !== null) {
params.push(endId)
where.push(`m.id <= $${params.length}`)
}
where.push('(m.file_size IS NULL OR m.file_size <= 0)')
where.push('m.s3_path IS NOT NULL')
if (!includeDeleted)
where.push('av.deleted = false')
if (appVersionId) {
params.push(appVersionId)
where.push(`m.app_version_id = $${params.length}`)
}
if (appId) {
params.push(appId)
where.push(`av.app_id = $${params.length}`)
}
}
function buildCandidateQuery({ afterId, appId, appVersionId, endId, includeDeleted, limit }) {
const params = [afterId]
const where = [
'm.id > $1',
]
appendCommonFilters(where, params, { appId, appVersionId, endId, includeDeleted, startId: null })
params.push(limit)
return {
params,
// Keep every page as a bounded manifest.id range scan so the primary-key index carries the backfill.
sql: `
SELECT
m.id,
m.app_version_id,
m.file_name,
m.s3_path,
m.file_size,
av.app_id,
av.name AS version_name,
av.deleted
FROM public.manifest m
INNER JOIN public.app_versions av ON av.id = m.app_version_id
WHERE ${where.join(' AND ')}
ORDER BY m.id
LIMIT $${params.length}
`,
}
}
function buildBulkUpdateQuery(rows) {
const params = []
const values = rows.map((row) => {
params.push(row.id, row.size)
const idIndex = params.length - 1
const sizeIndex = params.length
return `($${idIndex}::bigint, $${sizeIndex}::bigint)`
}).join(', ')
return {
params,
sql: `
UPDATE public.manifest AS m
SET file_size = v.file_size
FROM (VALUES ${values}) AS v(id, file_size)
WHERE m.id = v.id
AND (m.file_size IS NULL OR m.file_size <= 0)
`,
}
}
function createWorkerReport() {
return {
batches: 0,
checked: 0,
currentBatch: null,
currentFirstId: null,
currentLastId: null,
done: false,
fixed: 0,
lastId: null,
missingSize: 0,
pages: 0,
unchanged: 0,
}
}
function writeReport(outputPath, report) {
writeFileSync(outputPath, `${JSON.stringify(report, null, 2)}\n`)
}
function csvValue(value) {
if (value === undefined || value === null)
return ''
const text = typeof value === 'object' ? JSON.stringify(value) : String(value)
return /[",\n\r]/.test(text) ? `"${text.replaceAll('"', '""')}"` : text
}
function failedCsvRow(error) {
return FAILED_CSV_HEADERS.map((header) => {
if (header === 'error_message')
return csvValue(error.error?.message)
if (header === 'error_name')
return csvValue(error.error?.name)
if (header === 'error_status')
return csvValue(error.error?.status)
return csvValue(error[header])
}).join(',')
}
function createFailedCsv(outputPath) {
writeFileSync(outputPath, `${FAILED_CSV_HEADERS.join(',')}\n`)
}
function appendFailedCsvRows(outputPath, errors) {
if (errors.length === 0)
return
appendFileSync(outputPath, `${errors.map(failedCsvRow).join('\n')}\n`)
}
function defaultFailedCsvPath(outputPath) {
return outputPath.toLowerCase().endsWith('.json')
? outputPath.replace(/\.json$/i, '-failed.csv')
: `${outputPath}-failed.csv`
}
function createProgressLogger(report, apply) {
const startedAt = Date.now()
let lastLogAt = 0
return (force = false) => {
const now = Date.now()
if (!force && now - lastLogAt < 1000)
return
lastLogAt = now
const elapsedSeconds = Math.max(1, Math.round((now - startedAt) / 1000))
const rate = Math.round(report.checked / elapsedSeconds)
const activeWorkers = Object.values(report.workers).filter(worker => !worker.done).length
console.log(`Checked ${report.checked}, ${apply ? 'fixed' : 'fixable'} ${apply ? report.fixed : report.unchanged}, missing ${report.missingSize}, rate ${rate}/s, active workers ${activeWorkers}, batches ${report.claimedBatches}, cursor id ${report.lastClaimedId}`)
}
}
async function processCandidates({ apply, dbAttempts, pool, s3, storageAttempts, verbose }, candidates) {
const results = await Promise.all(candidates.map(async (row) => {
const storage = await getObjectSizeWithRetry(s3, row.s3_path, storageAttempts)
if (verbose) {
console.log(`${row.id} ${row.file_name} size=${storage.size} method=${storage.method}${storage.status ? ` status=${storage.status}` : ''} attempts=${storage.attempts}`)
}
return { row, storage }
}))
const rowsWithSize = results
.filter(result => result.storage.size > 0)
.map(result => ({
id: result.row.id,
size: result.storage.size,
}))
let fixed = 0
if (apply && rowsWithSize.length > 0) {
const update = buildBulkUpdateQuery(rowsWithSize)
const updateResult = await queryWithRetry(pool, update.sql, update.params, 'bulk manifest file_size update', dbAttempts)
fixed = updateResult.rowCount ?? 0
}
const missingErrors = results
.filter(result => result.storage.size <= 0)
.map(result => ({
app_id: result.row.app_id,
app_version_id: result.row.app_version_id,
attempted_s3_path: result.storage.attempted_s3_path,
attempted_s3_paths: result.storage.attempted_s3_paths,
attempts: result.storage.attempts,
error: result.storage.error,
file_name: result.row.file_name,
id: result.row.id,
method: result.storage.method,
reason: result.storage.reason,
s3_path: result.row.s3_path,
status: result.storage.status,
version_name: result.row.version_name,
}))
return {
checked: results.length,
fixed,
missingErrors,
missingSize: missingErrors.length,
unchanged: rowsWithSize.length - fixed,
}
}
function mergePageReport(report, workerReport, pageReport, failedCsvPath) {
report.checked += pageReport.checked
report.fixed += pageReport.fixed
report.missingSize += pageReport.missingSize
report.unchanged += pageReport.unchanged
report.errors.push(...pageReport.missingErrors)
appendFailedCsvRows(failedCsvPath, pageReport.missingErrors)
workerReport.checked += pageReport.checked
workerReport.fixed += pageReport.fixed
workerReport.missingSize += pageReport.missingSize
workerReport.unchanged += pageReport.unchanged
workerReport.pages += 1
}
function createBatchClaimer({ appId, appVersionId, batchSize, dbAttempts, endId, includeDeleted, limit, pool, report, startId, writeProgress }) {
let afterId = startId
let claimChain = Promise.resolve()
let done = false
let remaining = limit
return async function claimBatch(workerIndex) {
const claim = claimChain.then(async () => {
if (done || remaining <= 0)
return null
const query = buildCandidateQuery({
afterId,
appId,
appVersionId,
endId,
includeDeleted,
limit: Math.min(batchSize, remaining),
})
const candidates = (await queryWithRetry(pool, query.sql, query.params, `worker ${workerIndex} candidate read`, dbAttempts)).rows
if (candidates.length === 0) {
done = true
return null
}
const firstId = candidates[0].id
const lastId = candidates[candidates.length - 1].id
const batch = {
candidates,
firstId,
index: report.claimedBatches,
lastId,
}
afterId = lastId
remaining -= candidates.length
report.claimedBatches += 1
report.lastClaimedId = lastId
writeProgress()
return batch
})
claimChain = claim.catch(() => {})
return claim
}
}
async function runBatchWorker({ claimBatch, failedCsvPath, options, report, workerIndex, writeProgress }) {
const workerReport = report.workers[workerIndex]
try {
while (true) {
const batch = await claimBatch(workerIndex)
if (!batch)
break
workerReport.batches += 1
workerReport.currentBatch = batch.index
workerReport.currentFirstId = batch.firstId
workerReport.currentLastId = batch.lastId
workerReport.lastId = batch.lastId
writeProgress()
const pageReport = await processCandidates(options, batch.candidates)
mergePageReport(report, workerReport, pageReport, failedCsvPath)
writeProgress()
}
}
catch (error) {
workerReport.error = serializeError(error)
throw error
}
finally {
workerReport.done = true
writeProgress(true)
}
}
async function main() {
if (hasFlag('--help') || hasFlag('-h')) {
console.log(`Usage:
bun scripts/backfill_manifest_file_sizes.mjs --app-version-id=<id> [--apply]
bun scripts/backfill_manifest_file_sizes.mjs --app-id=<app_id> [--limit=1000] [--apply]
bun scripts/backfill_manifest_file_sizes.mjs --all --apply
Options:
--apply Update manifest.file_size. Without this, dry-run only.
--all Scan all manifest rows with missing size.
--app-version-id Restrict to one bundle id.
--app-id Restrict to one app id.
--limit Max rows to scan without --all. Default: 1000.
--batch-size Backward-compatible alias; candidate reads are always 1000.
--workers Parallel shared-cursor workers. Default: 8 for --all, 1 otherwise.
--concurrency Backward-compatible no-op; storage uses full-batch parallelism for maximum throughput.
--storage-attempts Storage metadata attempts per file. Default: 3.
--db-attempts DB read/update attempts. Default: 5.
--start-id Exclusive lower manifest.id bound for resume.
--end-id Inclusive upper manifest.id bound.
--report Report JSON output path.
--failed-csv Failed metadata CSV output path.
--include-deleted Include deleted bundles.
--target prod|local Env target. Default: prod.
--local Alias for --target=local.
--verbose Print every checked row.
`)
return
}
const apply = hasFlag('--apply')
const all = hasFlag('--all')
const includeDeleted = hasFlag('--include-deleted')
const verbose = hasFlag('--verbose')
const appVersionIdRaw = getArgValue('--app-version-id')
const appVersionId = appVersionIdRaw ? Number.parseInt(appVersionIdRaw, 10) : null
const appId = getArgValue('--app-id') ?? null
const limit = all ? Number.POSITIVE_INFINITY : getNumberArg('--limit', 1000)
const workers = getNumberArg('--workers', all ? 8 : 1)
const batchSize = getFixedBatchSizeArg()
const storageAttempts = getNumberArg('--storage-attempts', 3)
const dbAttempts = getNumberArg('--db-attempts', 5)
const startId = getOptionalNumberArg('--start-id') ?? 0
const endId = getOptionalNumberArg('--end-id')
const reportPathArg = getArgValue('--report')
const failedCsvPathArg = getArgValue('--failed-csv')
if (!all && !appVersionId && !appId)
throw new Error('Pass --app-version-id, --app-id, or --all')
if (appVersionIdRaw && (!Number.isFinite(appVersionId) || appVersionId <= 0))
throw new Error('--app-version-id must be a positive integer')
if (!all && workers !== 1)
throw new Error('--workers is only supported with --all')
if (endId !== null && endId <= startId)
throw new Error('--end-id must be greater than --start-id')
const databaseUrl = getSafeDatabaseUrl()
console.log(`Using ${target} database target: ${describeDatabaseUrl(databaseUrl)}`)
const outputDir = resolve(__dirname, '../tmp/manifest_file_size_backfill')
const outputPath = reportPathArg
? resolve(process.cwd(), reportPathArg)
: resolve(outputDir, `manifest-file-size-backfill-${Date.now()}.json`)
const failedCsvPath = failedCsvPathArg
? resolve(process.cwd(), failedCsvPathArg)
: defaultFailedCsvPath(outputPath)
mkdirSync(dirname(outputPath), { recursive: true })
mkdirSync(dirname(failedCsvPath), { recursive: true })
createFailedCsv(failedCsvPath)
const workerCount = all ? workers : 1
const storageFanoutPerWorker = batchSize
const effectiveStorageFanout = storageFanoutPerWorker * workerCount
const poolMax = Math.min(Math.max(workerCount * 2 + 4, 4), 40)
const pool = new pg.Pool({
connectionString: databaseUrl,
max: poolMax,
ssl: target === 'prod' ? { rejectUnauthorized: false } : false,
})
const s3 = new S3Client({
accessKey: getRequiredEnv('S3_ACCESS_KEY_ID'),
bucket: getRequiredEnv('S3_BUCKET'),
endPoint: getStorageEndpoint(),
pathStyle: true,
region: process.env.S3_REGION || 'auto',
secretKey: getRequiredEnv('S3_SECRET_ACCESS_KEY'),
})
const report = {
apply,
appId,
appVersionId,
batchSize,
checked: 0,
claimedBatches: 0,
dbAttempts,
endedAt: null,
endId,
errors: [],
failedCsvPath,
fixed: 0,
includeDeleted,
lastClaimedId: startId,
maxBatchSize: MAX_CANDIDATE_BATCH_SIZE,
missingSize: 0,
poolMax,
scannedAt: new Date().toISOString(),
startId,
storageAttempts,
storageFanoutPerWorker,
effectiveStorageFanout,
target,
unchanged: 0,
workerCount,
workers: {},
}
const logProgress = createProgressLogger(report, apply)
const writeProgress = (force = false) => {
writeReport(outputPath, report)
logProgress(force)
}
try {
console.log(`Scanning manifest.id > ${startId}${endId ? ` and <= ${endId}` : ''} with ${workerCount} workers, page size ${batchSize}, storage fan-out ${storageFanoutPerWorker} per worker (${effectiveStorageFanout} effective), DB pool max ${poolMax}`)
const claimBatch = createBatchClaimer({
appId,
appVersionId,
batchSize,
dbAttempts,
endId,
includeDeleted,
limit,
pool,
report,
startId,
writeProgress,
})
for (let workerIndex = 0; workerIndex < workerCount; workerIndex++) {
report.workers[workerIndex] = createWorkerReport()
}
writeProgress(true)
const workerResults = await Promise.allSettled(Array.from({ length: workerCount }, (_, workerIndex) => runBatchWorker({
claimBatch,
failedCsvPath,
options: {
apply,
dbAttempts,
pool,
s3,
storageAttempts,
verbose,
},
report,
workerIndex,
writeProgress,
})))
const failedWorkers = workerResults
.map((result, index) => ({ index, result }))
.filter(({ result }) => result.status === 'rejected')
if (failedWorkers.length > 0) {
throw new Error(`${failedWorkers.length} backfill workers failed: ${failedWorkers.map(({ index, result }) => `worker ${index}: ${result.reason?.message ?? result.reason}`).join('; ')}`)
}
}
finally {
report.endedAt = new Date().toISOString()
writeProgress(true)
await pool.end()
}
console.log('\nSummary')
console.log(` Mode: ${apply ? 'apply' : 'dry-run'}`)
console.log(` Checked: ${report.checked}`)
console.log(` ${apply ? 'Fixed' : 'Fixable'}: ${apply ? report.fixed : report.unchanged}`)
console.log(` Missing size: ${report.missingSize}`)
console.log(` Report: ${outputPath}`)
console.log(` Failed CSV: ${failedCsvPath}`)
if (report.missingSize > 0)
process.exitCode = 1
}
main().catch((error) => {
console.error(error)
process.exit(1)
})