fix(backup-helper): make commit parking retriable

Author: BravoNatalie

scripts/backup-helper/commands/commit.mjs

@@ -10,6 +10,7 @@ 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 pMap from 'p-map'
 import { createPublicClient, http } from 'viem'
 import { getAddress } from 'viem/utils'
 import { z } from 'zod'
@@ -20,6 +21,7 @@ import { openTrackingDb } from '../lib/tracking-db.mjs'
 
 const PARKING_BATCH_SIZE = 50
 const COMMIT_BATCH_SIZE = 20
+const PARKING_RECOVERY_BATCH_SIZE = 1_000
 const DEFAULT_COMMIT_CONCURRENCY = 4
 const EMPTY_POLL_INTERVAL_MS = 1_000
 
@@ -167,31 +169,57 @@ const parkingResultSchema = z
     message: 'count must match pieces length',
   })
 
+/**
+ * Curio may emit log lines before the final JSON result.
+ *
+ * @param {string} stdout
+ */
+function extractParkingJson(stdout) {
+  const trimmed = stdout.trim()
+  if (!trimmed) {
+    throw new Error('commit: parking command returned empty stdout')
+  }
+
+  const objectStart = trimmed.lastIndexOf('\n{')
+  const start = objectStart >= 0 ? objectStart + 1 : trimmed.indexOf('{')
+  if (start < 0) {
+    throw new Error('commit: parking command returned no JSON result in stdout')
+  }
+
+  return trimmed.slice(start)
+}
+
 /**
  * @param {string} dir
  * @param {string} target
  * @returns {Promise<ParkingResult>}
  */
 async function runParkingBinary(dir, target) {
-  const result = await execa({
-    env: {
-      LANG: 'en_US.UTF-8',
-    },
-  })('curio', [
-    'toolbox',
-    'import-pieces',
-    '--source',
-    shardsDir(dir),
-    '--target',
-    target,
-    '--batch-size',
-    String(PARKING_BATCH_SIZE),
-  ])
+  let result
+  try {
+    result = await execa({
+      env: {
+        LANG: 'en_US.UTF-8',
+        GOLOG_LOG_LEVEL: 'error',
+      },
+    })('curio', [
+      'toolbox',
+      'import-pieces',
+      '--source',
+      shardsDir(dir),
+      '--target',
+      target,
+      '--batch-size',
+      String(PARKING_BATCH_SIZE),
+    ])
+  } catch (err) {
+    const message = err?.stderr || err?.stdout || err?.message || String(err)
+    throw new Error(`commit: parking command failed: ${message}`)
+  }
 
   let parsed
   try {
-    console.log(`Parking binary output: ${result.stdout}`)
-    parsed = JSON.parse(result.stdout || '{}')
+    parsed = JSON.parse(extractParkingJson(result.stdout || ''))
   } catch (err) {
     throw new Error(`commit: parking command returned invalid JSON: ${err?.message || err}`)
   }
@@ -209,6 +237,63 @@ function buildCommitPieces(rows) {
   }))
 }
 
+/**
+ * @param {string} serviceUrl
+ * @param {string} pieceCid
+ */
+async function isPieceAvailable(serviceUrl, pieceCid) {
+  const baseUrl = serviceUrl.endsWith('/') ? serviceUrl : `${serviceUrl}/`
+  const url = new URL(`piece/${pieceCid}`, baseUrl)
+  const response = await fetch(url, { method: 'HEAD' })
+
+  if (response.status === 200) return true
+  if (response.status === 404) return false
+  if (!response.ok) {
+    throw new Error(`commit: parking recovery HEAD ${url.toString()} returned ${response.status}`)
+  }
+
+  return true
+}
+
+/**
+ * Last-resort recovery for the crash window where Curio already parked files
+ * but the DB was not updated before the process failed.
+ *
+ * @param {object} args
+ * @param {TrackingDb} args.tracking
+ * @param {string} args.serviceUrl
+ * @param {number} args.concurrency
+ */
+async function recoverParkedPieces({ tracking, serviceUrl, concurrency }) {
+  let afterPieceCid = ''
+  let recoveredCount = 0
+
+  while (true) {
+    const pendingPieceCids = tracking.listPendingCommitPieceCids(PARKING_RECOVERY_BATCH_SIZE, afterPieceCid)
+    if (pendingPieceCids.length === 0) return recoveredCount
+
+    const recovered = await pMap(
+      pendingPieceCids,
+      async (pieceCid) => {
+        try {
+          return (await isPieceAvailable(serviceUrl, pieceCid)) ? pieceCid : null
+        } catch (err) {
+          const message = `parking recovery probe failed: ${err?.message || err}`
+          console.error(`commit: ${message} for piece ${pieceCid}`)
+          tracking.markPendingCommitPieceError(pieceCid, message)
+          return null
+        }
+      },
+      { concurrency },
+    )
+
+    const recoveredPieceCids = recovered.filter((pieceCid) => pieceCid != null)
+    tracking.markParkedByPieceCids(recoveredPieceCids)
+    recoveredCount += recoveredPieceCids.length
+    afterPieceCid = pendingPieceCids[pendingPieceCids.length - 1]
+  }
+}
+
 /**
  * @param {object} args
  * @param {ReturnType<typeof fromSecp256k1>} args.sessionKey
@@ -291,7 +376,18 @@ export async function runCommit({
       try {
         while (true) {
           const parkingResult = await runParkingBinary(dir, target)
-          if (parkingResult.count === 0) return
+          if (parkingResult.count === 0) {
+            const recovered = await recoverParkedPieces({
+              tracking,
+              serviceUrl,
+              concurrency: commitConcurrency,
+            })
+            if (recovered === 0) return
+
+            renderCommitProgress(tracking.getCommitStats())
+            continue
+          }
+
           tracking.markParkedByPieceCids(parkingResult.pieces)
           renderCommitProgress(tracking.getCommitStats())
         }

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

@@ -354,11 +354,30 @@ export function openTrackingDb(dir) {
   const markParkedByPieceCidStmt = db.prepare(`
     UPDATE root_shards
     SET commit_status = '${COMMIT_STATUS.parked}',
+        last_commit_error = NULL,
+        updated_at = ?
+    WHERE piece_cid = ?
+      AND commit_status = '${COMMIT_STATUS.pending}'
+  `)
+
+  const markPendingCommitPieceErrorStmt = db.prepare(`
+    UPDATE root_shards
+    SET last_commit_error = ?,
         updated_at = ?
     WHERE piece_cid = ?
       AND commit_status = '${COMMIT_STATUS.pending}'
   `)
 
+  const pendingCommitPieceCidsStmt = db.prepare(`
+    SELECT DISTINCT piece_cid
+    FROM root_shards
+    WHERE commit_status = '${COMMIT_STATUS.pending}'
+      AND piece_cid IS NOT NULL
+      AND piece_cid > ?
+    ORDER BY piece_cid
+    LIMIT ?
+  `)
+
   const claimCommitCandidatesStmt = db.prepare(`
     SELECT root_cid, shard_cid, piece_cid
     FROM root_shards
@@ -777,6 +796,23 @@ export function openTrackingDb(dir) {
       return changed
     },
 
+    /**
+     * @param {string} pieceCid
+     * @param {string} error
+     */
+    markPendingCommitPieceError(pieceCid, error) {
+      return Number(markPendingCommitPieceErrorStmt.run(error, now(), pieceCid).changes || 0)
+    },
+
+    /**
+     * @param {number} limit
+     * @param {string} afterPieceCid
+     * @returns {string[]}
+     */
+    listPendingCommitPieceCids(limit, afterPieceCid = '') {
+      return pendingCommitPieceCidsStmt.all(afterPieceCid, limit).map((row) => row.piece_cid?.toString())
+    },
+
     /**
      * @param {number} limit
      * @returns {CommitCandidate[]}