fix: add correct path to runParkingBinary and use the public gateway to getDataSet

Author: BravoNatalie

scripts/backup-helper/README.md

@@ -8,67 +8,73 @@ This tool turns the inventory into:
 
 The input DB is treated as strictly read-only, and all derived state lives in `tracking.db` under the output directory, so `rm -rf <dir>` is always a safe reset.
 
-## Subcommands
+## Commands
 
-```sh
-node scripts/backup-helper/index.mjs create   --db <space-inventory.db> --dir <output-dir>
-node scripts/backup-helper/index.mjs download --dir <output-dir> [--port N] [--concurrency N]
-node scripts/backup-helper/index.mjs prepare  --dir <output-dir> [--concurrency N]
-node scripts/backup-helper/index.mjs commit   --dir <output-dir> --target <curio-piece-dir> --service-url https://... --provider-address 0x... --session-key 0x... --customer-wallet 0x... [--network mainnet|calibration] [--concurrency N] [--retry]
-```
+* **`create`** — Reads the input DB, deduplicates shards by `shard_cid` across all spaces, and generates:
+
+  * `<dir>/tracking.db`
+  * `<dir>/manifest.aria2`
+
+  Safe to re-run. Produces identical output for unchanged input and never overwrites an existing `piece_cid`.
+
+* **`download`** — Starts a local aria2 RPC worker and downloads shards from `tracking.db` into:
+
+  ```
+  <dir>/shards/<shardCID>.car
+  ```
+
+  Resumable via `tracking.db`, `aria2.session`, and `.aria2` control files. If `--port` is not provided, a free localhost port is selected automatically.
 
-- `create` — read the input DB, deduplicate by `shard_cid` across every space,
-  populate `<dir>/tracking.db`, and stream `<dir>/manifest.aria2`. Idempotent
-  on re-run: produces byte-identical output for unchanged input and never
-  clobbers a previously-computed `piece_cid` in tracking.db.
-- `download` — starts a local aria2 RPC worker directly from the command,
-  then schedules shards from `tracking.db`.
-  Writes each CAR to `<dir>/shards/<shardCID>.car`. Resumable via shard status
-  in `tracking.db`, plus aria2's session and per-file `.aria2` control files.
-  `--port N` is optional; if omitted, `download` picks a free localhost port
-  for that run automatically. The `manifest.aria2` and `aria2.session` files are
-  just generated artifacts that can be useful for debugging if needed.
-- `prepare` — process completed local shard CARs from `tracking.db`, compute
-  pieceCID v2 when `piece_cid IS NULL`, and rename each prepared CAR from
-  `<shardCID>.car` to `<pieceCID>.car`. Failures land in `tracking.db`'s
-  `failures` table under `stage='prepare'`; rows are deleted on a later
-  successful attempt.
-- `commit` — uses `tracking.db` plus Synapse to park prepared pieces and commit
-  them on-chain. The parking flow shells out to
-  `curio toolbox import-pieces --source <dir> --target <target> --batch-size N`
-  and consumes its JSON `{ count, pieces }` output until `count = 0`. Commit
-  state is persisted on `root_shards` plus a single `migration_metadata` row in
-  `tracking.db`.
+* **`prepare`** — Processes downloaded CARs, computes missing Piece CID v2 values, and renames files from:
+
+  ```
+  <shardCID>.car → <pieceCID>.car
+  ```
+
+  Failures are recorded in `tracking.db` (`stage='prepare'`) and cleared automatically after a successful retry.
+
+* **`commit`** — Parks prepared pieces in Curio and performs on-chain commits concurrently using synapse-core. Piece parking is done through:
+
+  ```bash
+  curio toolbox import-pieces \
+    --source <dir> \
+    --target <target> \
+    --batch-size N
+  ```
+
+  The command continuously imports pieces into Curio while concurrently submitting eligible commits on-chain. Progress is persisted in `tracking.db` (`root_shards` and `migration_metadata`), allowing the operation to be safely resumed after interruptions.
 
 ## End-to-end workflow
 
 ```sh
 # 1. Inventory + manifest (fast: minutes for 2M-row input DBs).
 node scripts/backup-helper/index.mjs create \
-  --db /path/to/space-inventory.db \
-  --dir /path/to/backup-dir
+  --db <space-inventory.db> \
+  --dir <output-dir>
 
 # 2. Download CARs (long: hours for terabyte-scale inventories; resumable).
+# Optional: pin aria2 RPC to a specific localhost port
 node scripts/backup-helper/index.mjs download \
-  --dir /path/to/backup-dir
-
-# Optional: pin aria2 RPC to a specific localhost port for debugging.
-node scripts/backup-helper/index.mjs download \
-  --dir /path/to/backup-dir \
-  --port 6801
+  --dir <output-dir> \
+  [--port N] \
+  [--concurrency N]
 
 # 3. Compute pieceCIDs + rename CARs to pieceCID filenames.
 node scripts/backup-helper/index.mjs prepare \
-  --dir /path/to/backup-dir
+  --dir <output-dir> \
+  [--concurrency N]
 
 # 4. Park prepared pieces and commit them on-chain.
 node scripts/backup-helper/index.mjs commit \
-  --dir /path/to/backup-dir \
-  --target /path/to/curio-storage/piece \
-  --service-url https://pdp.example.com/ \
+  --dir <output-dir> \
+  --target <curio-piece-dir> \
+  --service-url https://... \
   --provider-address 0x... \
   --session-key 0x... \
-  --customer-wallet 0x...
+  --customer-wallet 0x... \
+  [--network mainnet|calibration] \
+  [--concurrency N] \
+  [--retry]
 ```
 
 ## Final Output layout
@@ -84,27 +90,27 @@ node scripts/backup-helper/index.mjs commit \
 
 ## Requirements
 
-- Node.js **24** (uses the built-in `node:sqlite`; no native `better-sqlite3`
+* Node.js **24** (uses the built-in `node:sqlite`; no native `better-sqlite3`
   install required).
-- `aria2c` on PATH for the `download` subcommand.
-  - macOS: `brew install aria2`
-  - Debian/Ubuntu: `sudo apt install aria2`
-  - Windows: download from <https://aria2.github.io/> and add `aria2c.exe` to PATH.
-- `@filoz/synapse-core` for pieceCID v2 streaming hash
+* `aria2c` on PATH for the `download` subcommand.
+  * macOS: `brew install aria2`
+  * Debian/Ubuntu: `sudo apt install aria2`
+  * Windows: download from <https://aria2.github.io/> and add `aria2c.exe` to PATH.
+* `@filoz/synapse-core` for pieceCID v2 streaming hash
 
 ## Caveats
 
-- **The input `space-inventory.db` is read-only.** The tool opens it with the
+* **The input `space-inventory.db` is read-only.** The tool opens it with the
   `{ readOnly: true }` option and never writes. All derived state lives in
   `<dir>/tracking.db`, separate from the client's deliverable.
-- **One `<dir>` per client.** The output directory is the unit of "a client's
+* **One `<dir>` per client.** The output directory is the unit of "a client's
   backup". Different clients should run in different directories.
-- **Don't run two backup-helper commands against the same `<dir>` at the same
+* **Don't run two backup-helper commands against the same `<dir>` at the same
   time.** .
-- **`download` uses one local aria2 RPC daemon per run.** By default it picks a
+* **`download` uses one local aria2 RPC daemon per run.** By default it picks a
   free localhost port automatically; you can override it with `--port N` when
   you need a predictable port for debugging.
-- **`download` performance is tuned for Cloudflare R2-hosted shards.** The
+* **`download` performance is tuned for Cloudflare R2-hosted shards.** The
   aria2 worker saturates bandwidth horizontally via
   `--max-concurrent-downloads` and uses size-based per-file `split` /
   `max-connection-per-server` settings. Tune via `--concurrency N`; current

scripts/backup-helper/commands/commit.mjs

@@ -10,10 +10,12 @@ import { fromSecp256k1 } from '@filoz/synapse-core/session-key'
 import { addPieces, createDataSet, waitForAddPieces, waitForCreateDataSet } from '@filoz/synapse-core/sp'
 import { getDataSet } from '@filoz/synapse-core/warm-storage'
 import { execa } from 'execa'
+import { createPublicClient, http } from 'viem'
 import { getAddress } from 'viem/utils'
 import { z } from 'zod'
 
 import { renderProgressLine } from '../../utils.js'
+import { shardsDir } from '../lib/layout.mjs'
 import { openTrackingDb } from '../lib/tracking-db.mjs'
 
 const PARKING_BATCH_SIZE = 50
@@ -28,6 +30,12 @@ const EMPTY_POLL_INTERVAL_MS = 1_000
  * @property {string} pieceCid
  */
 
+/**
+ * @typedef {object} EnsuredDataSet
+ * @property {number} dataSetId
+ * @property {bigint} clientDataSetId
+ */
+
 /** @typedef {ReturnType<typeof openTrackingDb>} TrackingDb */
 
 function sleep(ms) {
@@ -58,25 +66,28 @@ async function createSessionKey(sessionKeyStr, customerWallet, chain) {
  * @param {ReturnType<typeof fromSecp256k1>} args.sessionKey
  * @param {string} args.serviceUrl
  * @param {`0x${string}`} args.providerAddress
+ * @param {import('viem').PublicClient} args.publicClient
+ * @returns {Promise<EnsuredDataSet>}
  */
-async function ensureDataSet({ tracking, sessionKey, serviceUrl, providerAddress }) {
+async function ensureDataSet({ tracking, sessionKey, serviceUrl, providerAddress, publicClient }) {
   const metadata = tracking.getMigrationMetadata()
   if (!metadata) {
     throw new Error('commit: migration metadata was not initialized')
   }
 
   if (metadata.dataSetId != null) {
+    console.log(`Dataset ${metadata.dataSetId} already exists, skipping creation...`)
     if (metadata.clientDataSetId == null) {
-      const dataSetInfo = await getDataSet(sessionKey.client, {
+      const dataSetInfo = await getDataSet(publicClient, {
         dataSetId: BigInt(metadata.dataSetId),
       })
       if (!dataSetInfo) {
         throw new Error(`commit: dataset ${metadata.dataSetId} was not found on-chain`)
       }
-      tracking.setMigrationClientDataSetId(Number(dataSetInfo.clientDataSetId))
+      tracking.setMigrationClientDataSetId(dataSetInfo.clientDataSetId)
       return {
         dataSetId: metadata.dataSetId,
-        clientDataSetId: Number(dataSetInfo.clientDataSetId),
+        clientDataSetId: dataSetInfo.clientDataSetId,
       }
     }
 
@@ -90,6 +101,11 @@ async function ensureDataSet({ tracking, sessionKey, serviceUrl, providerAddress
     }
   }
 
+  console.log(`Creating dataset...
+    payee=${providerAddress} 
+    payer=${metadata.clientWallet} 
+    serviceURL=${serviceUrl}
+  `)
   const createResult = await createDataSet(sessionKey.client, {
     cdn: true,
     payee: providerAddress,
@@ -100,20 +116,30 @@ async function ensureDataSet({ tracking, sessionKey, serviceUrl, providerAddress
       withIPFSIndexing: '',
     },
   })
+
+  console.log(`Waiting for dataset to be created...
+    txHash=${createResult.txHash}
+    statusUrl=${createResult.statusUrl}
+  `)
+
   const { dataSetId } = await waitForCreateDataSet(createResult)
-  const dataSetInfo = await getDataSet(sessionKey.client, { dataSetId })
+  console.log(`Dataset created: id=${dataSetId.toString()}`)
+
+  const dataSetInfo = await getDataSet(publicClient, { dataSetId })
   if (!dataSetInfo) {
     throw new Error(`commit: dataset ${dataSetId} was created but could not be fetched`)
   }
 
+  console.log(`Saving dataset info: dataSetId=${dataSetId.toString()} clientDataSetId=${dataSetInfo.clientDataSetId}`)
+
   tracking.markMigrationDataSetCreated({
     dataSetId: Number(dataSetId),
-    clientDataSetId: Number(dataSetInfo.clientDataSetId),
+    clientDataSetId: dataSetInfo.clientDataSetId,
   })
 
   return {
     dataSetId: Number(dataSetId),
-    clientDataSetId: Number(dataSetInfo.clientDataSetId),
+    clientDataSetId: dataSetInfo.clientDataSetId,
   }
 }
 
@@ -148,7 +174,7 @@ async function runParkingBinary(dir, target) {
     'toolbox',
     'import-pieces',
     '--source',
-    dir,
+    shardsDir(dir),
     '--target',
     target,
     '--batch-size',
@@ -180,7 +206,7 @@ function buildCommitPieces(rows) {
  * @param {ReturnType<typeof fromSecp256k1>} args.sessionKey
  * @param {string} args.serviceUrl
  * @param {number} args.dataSetId
- * @param {number} args.clientDataSetId
+ * @param {bigint} args.clientDataSetId
  * @param {TrackingDb} args.tracking
  * @param {CommitRow[]} args.rows
  */
@@ -189,7 +215,7 @@ async function commitBatch({ sessionKey, serviceUrl, dataSetId, clientDataSetId,
     const addResult = await addPieces(sessionKey.client, {
       serviceURL: serviceUrl,
       dataSetId: BigInt(dataSetId),
-      clientDataSetId: BigInt(clientDataSetId),
+      clientDataSetId,
       pieces: buildCommitPieces(rows),
     })
     const addStatus = await waitForAddPieces({ statusUrl: addResult.statusUrl })
@@ -241,12 +267,16 @@ export async function runCommit({
       tracking.resetCommitRowsForRetry()
     }
 
+    console.log(`crate session key...`)
     const session = await createSessionKey(/** @type {`0x${string}`} */ (sessionKey), customerWallet, chain)
+    const publicClient = createPublicClient({ chain, transport: http() })
+
     const ensuredDataSet = await ensureDataSet({
       tracking,
       sessionKey: session,
       serviceUrl,
       providerAddress,
+      publicClient,
     })
 
     const parkingLane = (async () => {

scripts/backup-helper/lib/tracking-db.mjs

@@ -86,7 +86,7 @@ const MIGRATION_STATE = {
  * @property {string} serviceUrl
  * @property {string} providerAddress
  * @property {number | null} dataSetId
- * @property {number | null} clientDataSetId
+ * @property {bigint | null} clientDataSetId
  * @property {string} state
  * @property {number} updatedAt
  */
@@ -723,7 +723,7 @@ export function openTrackingDb(dir) {
         serviceUrl: row.service_url.toString(),
         providerAddress: row.provider_address.toString(),
         dataSetId: row.data_set_id != null ? Number(row.data_set_id) : null,
-        clientDataSetId: row.client_data_set_id != null ? Number(row.client_data_set_id) : null,
+        clientDataSetId: row.client_data_set_id != null ? BigInt(row.client_data_set_id.toString()) : null,
         state: row.state.toString(),
         updatedAt: Number(row.updated_at),
       }
@@ -737,19 +737,19 @@ export function openTrackingDb(dir) {
     },
 
     /**
-     * @param {number} clientDataSetId
+     * @param {bigint} clientDataSetId
      */
     setMigrationClientDataSetId(clientDataSetId) {
-      updateMigrationClientDataSetIdStmt.run(clientDataSetId, now())
+      updateMigrationClientDataSetIdStmt.run(clientDataSetId.toString(), now())
     },
 
     /**
      * @param {object} metadata
      * @param {number} metadata.dataSetId
-     * @param {number} metadata.clientDataSetId
+     * @param {bigint} metadata.clientDataSetId
      */
     markMigrationDataSetCreated({ dataSetId, clientDataSetId }) {
-      updateMigrationMetadataStmt.run(dataSetId, clientDataSetId, MIGRATION_STATE.migrating, now())
+      updateMigrationMetadataStmt.run(dataSetId, clientDataSetId.toString(), MIGRATION_STATE.migrating, now())
     },
 
     resetCommitRowsForRetry() {