backup-helper: parallelise prepare hashing + curio import-pieces --result compatibility

scripts/backup-helper/commands/commit.mjs

@@ -5,6 +5,9 @@
  * provider machine and returns ready-to-commit `pieceCid`s.
  */
 
+import fs from 'node:fs/promises'
+import path from 'node:path'
+
 import { parse as parsePieceCid } from '@filoz/synapse-core/piece'
 import { fromSecp256k1 } from '@filoz/synapse-core/session-key'
 import { addPieces, createDataSet, waitForAddPieces, waitForCreateDataSet } from '@filoz/synapse-core/sp'
@@ -195,9 +198,12 @@ function extractParkingJson(stdout) {
  * @returns {Promise<ParkingResult>}
  */
 async function runParkingBinary(dir, target) {
-  let result
+  // curio toolbox import-pieces now writes its JSON result to a --result file
+  // (commit 7c8297ca "always write in json file"), not stdout. Local patch
+  // pending Natalie shipping the same fix upstream to migration-support-scripts.
+  const resultPath = path.join(dir, '.parking-result.json')
   try {
-    result = await execa({
+    await execa({
       env: {
         LANG: 'en_US.UTF-8',
         GOLOG_LOG_LEVEL: 'error',
@@ -211,17 +217,27 @@ async function runParkingBinary(dir, target) {
       target,
       '--batch-size',
       String(PARKING_BATCH_SIZE),
+      '--result',
+      resultPath,
     ])
   } catch (err) {
-    const message = err?.stderr || err?.stdout || err?.message || String(err)
+    // curio writes the result file even on failure, with an `error` field set
+    let structured
+    try {
+      structured = JSON.parse(await fs.readFile(resultPath, 'utf8'))
+    } catch {}
+    await fs.rm(resultPath, { force: true })
+    const message = structured?.error || err?.stderr || err?.stdout || err?.message || String(err)
     throw new Error(`commit: parking command failed: ${message}`)
   }
 
   let parsed
   try {
-    parsed = JSON.parse(extractParkingJson(result.stdout || ''))
+    parsed = JSON.parse(await fs.readFile(resultPath, 'utf8'))
   } catch (err) {
     throw new Error(`commit: parking command returned invalid JSON: ${err?.message || err}`)
+  } finally {
+    await fs.rm(resultPath, { force: true })
   }
 
   return /** @type {ParkingResult} */ (parkingResultSchema.parse(parsed))

scripts/backup-helper/commands/prepare.mjs

@@ -7,10 +7,10 @@
  * CAR bytes and persisted back into tracking.db.
  */
 
-import { createReadStream } from 'node:fs'
 import fs from 'node:fs/promises'
+import { fileURLToPath } from 'node:url'
+import { Worker } from 'node:worker_threads'
 
-import { calculateFromIterable } from '@filoz/synapse-core/piece'
 import pMap from 'p-map'
 
 import { pathExists, renderProgressLine } from '../../utils.js'
@@ -28,15 +28,83 @@ const PREPARE_BATCH_SIZE = 1_000
  */
 
 /**
- * @param {string} carPath
+ * Pool of worker threads that each compute a piece CID from a CAR file, so the
+ * CPU-bound (pure-JS, single-threaded) CommP hashing runs in parallel across
+ * cores. Each worker calls the same `@filoz/synapse-core/piece` hash as before,
+ * so output piece CIDs are identical — only throughput changes. DB writes and
+ * file renames stay on the main thread (sqlite is not shared with workers).
  */
-async function calculateLocalPieceCid(carPath) {
-  const stream = createReadStream(carPath)
-  try {
-    const pieceCid = await calculateFromIterable(stream)
-    return pieceCid.toString()
-  } finally {
-    stream.destroy()
+class PieceCidWorkerPool {
+  /** @param {number} size */
+  constructor(size) {
+    this.workerPath = fileURLToPath(new URL('../lib/piece-cid-worker.mjs', import.meta.url))
+    /** @type {import('node:worker_threads').Worker[]} */
+    this.workers = []
+    /** @type {import('node:worker_threads').Worker[]} */
+    this.idle = []
+    /** @type {Array<{carPath: string, resolve: (v: string) => void, reject: (e: Error) => void}>} */
+    this.queue = []
+    /** @type {Map<import('node:worker_threads').Worker, {resolve: (v: string) => void, reject: (e: Error) => void}>} */
+    this.busy = new Map()
+    for (let i = 0; i < Math.max(1, size); i++) {
+      const worker = new Worker(this.workerPath)
+      worker.on('message', (msg) => this._onMessage(worker, msg))
+      worker.on('error', (err) => this._onError(worker, err))
+      this.workers.push(worker)
+      this.idle.push(worker)
+    }
+  }
+
+  /**
+   * @param {string} carPath
+   * @returns {Promise<string>}
+   */
+  compute(carPath) {
+    return new Promise((resolve, reject) => {
+      const worker = this.idle.pop()
+      if (worker) this._assign(worker, { carPath, resolve, reject })
+      else this.queue.push({ carPath, resolve, reject })
+    })
+  }
+
+  /**
+   * @param {import('node:worker_threads').Worker} worker
+   * @param {{carPath: string, resolve: (v: string) => void, reject: (e: Error) => void}} job
+   */
+  _assign(worker, job) {
+    this.busy.set(worker, { resolve: job.resolve, reject: job.reject })
+    worker.postMessage({ carPath: job.carPath })
+  }
+
+  /**
+   * @param {import('node:worker_threads').Worker} worker
+   * @param {{pieceCid?: string, error?: string}} msg
+   */
+  _onMessage(worker, msg) {
+    const job = this.busy.get(worker)
+    this.busy.delete(worker)
+    if (job) {
+      if (msg.error) job.reject(new Error(msg.error))
+      else job.resolve(/** @type {string} */ (msg.pieceCid))
+    }
+    const next = this.queue.shift()
+    if (next) this._assign(worker, next)
+    else this.idle.push(worker)
+  }
+
+  /**
+   * @param {import('node:worker_threads').Worker} worker
+   * @param {Error} err
+   */
+  _onError(worker, err) {
+    const job = this.busy.get(worker)
+    this.busy.delete(worker)
+    if (job) job.reject(err)
+    // a crashed worker leaves the pool; remaining workers continue
+  }
+
+  async close() {
+    await Promise.all(this.workers.map((worker) => worker.terminate()))
   }
 }
 
@@ -127,6 +195,7 @@ function renderPrepareProgress(summary) {
 export async function runPrepare({ dir, concurrency }) {
   const tracking = openTrackingDb(dir)
   const workerConcurrency = concurrency ?? DEFAULT_PREPARE_CONCURRENCY
+  const pool = new PieceCidWorkerPool(workerConcurrency)
 
   try {
     const total = tracking.getDownloadStats().complete
@@ -171,7 +240,7 @@ export async function runPrepare({ dir, concurrency }) {
 
             let pieceCid = candidate.pieceCid
             if (!pieceCid) {
-              pieceCid = await calculateLocalPieceCid(workItem.carPath)
+              pieceCid = await pool.compute(workItem.carPath)
               tracking.setPieceCid(workItem.shardCid, pieceCid)
               computedPieceCid = true
             } else {
@@ -209,6 +278,7 @@ export async function runPrepare({ dir, concurrency }) {
       `prepare: done. total=${summary.total} done=${summary.done}/${summary.total} computed=${summary.computed} failed=${summary.failed}`,
     )
   } finally {
+    await pool.close()
     tracking.close()
   }
 }

scripts/backup-helper/lib/piece-cid-worker.mjs

@@ -0,0 +1,29 @@
+/**
+ * Worker thread for `backup-helper prepare`: computes a piece CID from a local
+ * CAR file using the same `@filoz/synapse-core/piece` hash as the main thread,
+ * so the CPU-bound (pure-JS) CommP hashing can run in parallel across cores.
+ *
+ * Local addition (not upstream) — parallelises prepare's hashing. The hash
+ * itself is unchanged, so output piece CIDs are identical to the single-thread
+ * path; only the throughput differs.
+ */
+import { createReadStream } from 'node:fs'
+import { parentPort } from 'node:worker_threads'
+
+import { calculateFromIterable } from '@filoz/synapse-core/piece'
+
+if (!parentPort) {
+  throw new Error('piece-cid-worker.mjs must be run as a worker thread')
+}
+
+parentPort.on('message', async ({ carPath }) => {
+  const stream = createReadStream(carPath)
+  try {
+    const pieceCid = await calculateFromIterable(stream)
+    parentPort.postMessage({ pieceCid: pieceCid.toString() })
+  } catch (err) {
+    parentPort.postMessage({ error: String(err?.message || err) })
+  } finally {
+    stream.destroy()
+  }
+})