chore: add abort support to the reader phase

.nx/version-plans/version-plan-1778524089700.md

@@ -0,0 +1,6 @@
+---
+'@storacha/filecoin-pin-migration': patch
+'@storacha/cli': patch
+---
+
+add abort support to the reader phase

packages/cli/migrate.js

@@ -497,6 +497,7 @@ async function readInventories({
     resolver,
     state,
     spaceDIDs: /** @type {`did:key:${string}`[]} */ (spaceDIDs),
+    signal,
   })) {
     switch (event.type) {
       case 'reader:space:start':
@@ -536,6 +537,11 @@ async function readInventories({
     }
   }
 
+  if (signal.aborted) {
+    spinner.stop()
+    return { interrupted: true }
+  }
+
   spinner.succeed('Inventories ready')
   return { interrupted: signal.aborted }
 }

packages/filecoin-pin-migration/src/api.ts

@@ -325,6 +325,8 @@ interface BuildInventoriesBaseInput {
   resolver: SourceURLResolver
   /** Mutated in place; used for resume and checkpointing */
   state: MigrationState
+  /** AbortSignal for cooperative cancellation during reader I/O. */
+  signal?: AbortSignal
   options?: {
     /** Override the default indexing service URL */
     serviceURL?: URL

packages/filecoin-pin-migration/src/errors.js

@@ -114,3 +114,26 @@ export class ResumeBindingDriftError extends Error {
     return ResumeBindingDriftErrorName
   }
 }
+
+/**
+ * Abort is cooperative control flow, not a migration failure.
+ *
+ * @param {unknown} error
+ * @param {AbortSignal | undefined} [signal]
+ * @returns {boolean}
+ */
+export function isAbortError(error, signal) {
+  return (
+    signal?.aborted === true ||
+    (error instanceof DOMException && error.name === 'AbortError') ||
+    (error instanceof Error && error.name === 'AbortError')
+  )
+}
+
+/**
+ * @param {AbortSignal | undefined} signal
+ */
+export function throwIfAborted(signal) {
+  if (!signal?.aborted) return
+  throw new DOMException('The operation was aborted.', 'AbortError')
+}

packages/filecoin-pin-migration/src/reader/carpark.js

@@ -1,3 +1,5 @@
+import { isAbortError } from '../errors.js'
+
 const CARPARK_HOSTS = [
   'carpark-prod-0.r2.w3s.link',
   'carpark-prod-1.r2.w3s.link',
@@ -56,7 +58,8 @@ function createCarparkCandidateURLs(shard) {
 async function headObject(fetcher, url) {
   try {
     return await fetcher(url, { method: 'HEAD' })
-  } catch {
+  } catch (error) {
+    if (isAbortError(error)) throw error
     return null
   }
 }

packages/filecoin-pin-migration/src/reader/indexer.js

@@ -6,6 +6,7 @@ import { base32upper } from 'multiformats/bases/base32'
 import { CID } from 'multiformats/cid'
 import * as RAW from 'multiformats/codecs/raw'
 import { DEFAULT_CARPARK_CONCURRENCY } from '../constants.js'
+import { isAbortError, throwIfAborted } from '../errors.js'
 import { findCarparkLocation } from './carpark.js'
 
 /**
@@ -40,12 +41,14 @@ const EQUALS_CLAIM_PROTOCOL = 0x3e0001
  * @param {IndexingServiceReader} args.indexer
  * @param {ShardEntry[]} args.shards
  * @param {typeof fetch | undefined} args.fetcher
+ * @param {AbortSignal | undefined} [args.signal]
  * @returns {Promise<Map<string, ClaimsEntry>>}
  */
-export async function resolveClaimsIndex({ indexer, shards, fetcher }) {
+export async function resolveClaimsIndex({ indexer, shards, fetcher, signal }) {
   /** @type {Map<string, ClaimsEntry>} */
   const index = new Map()
   if (shards.length === 0) return index
+  throwIfAborted(signal)
 
   const requestedShardB58s = new Set()
   /** @type {string[]} */
@@ -70,6 +73,7 @@ export async function resolveClaimsIndex({ indexer, shards, fetcher }) {
       hashes,
       kind: 'standard',
     })
+    throwIfAborted(signal)
 
     if (claimsResult.ok) {
       primarySucceeded = true
@@ -79,7 +83,8 @@ export async function resolveClaimsIndex({ indexer, shards, fetcher }) {
         claimsResult.ok.claims.values()
       )
     }
-  } catch {
+  } catch (error) {
+    if (isAbortError(error, signal)) throw error
     // Best-effort fallback below.
   }
 
@@ -98,6 +103,7 @@ export async function resolveClaimsIndex({ indexer, shards, fetcher }) {
     index,
     shardsByB58,
     fetcher,
+    signal,
   })
 
   const missingLocationB58s = missingB58s.filter((b58) =>
@@ -112,6 +118,7 @@ export async function resolveClaimsIndex({ indexer, shards, fetcher }) {
     index,
     shardsByB58,
     fetcher,
+    signal,
   })
 
   return index
@@ -145,15 +152,26 @@ function applyPrimaryClaims(index, requestedShardB58s, claims) {
  * @param {Map<string, ClaimsEntry>} args.index
  * @param {Map<string, ShardEntry>} args.shardsByB58
  * @param {typeof fetch} args.fetcher
+ * @param {AbortSignal | undefined} [args.signal]
  */
-async function applyIPNIFallback({ b58s, index, shardsByB58, fetcher }) {
+async function applyIPNIFallback({
+  b58s,
+  index,
+  shardsByB58,
+  fetcher,
+  signal,
+}) {
   await pMap(
     b58s,
     async (b58) => {
       const shard = shardsByB58.get(b58)
       if (!shard) return
 
-      const providerResults = await fetchIPNIProviderResults(fetcher, b58)
+      const providerResults = await fetchIPNIProviderResults(
+        fetcher,
+        b58,
+        signal
+      )
       if (providerResults.length === 0) return
 
       const entry = getOrCreateClaimsEntry(index, b58)
@@ -164,7 +182,7 @@ async function applyIPNIFallback({ b58s, index, shardsByB58, fetcher }) {
         }
       }
     },
-    { concurrency: DEFAULT_IPNI_CONCURRENCY }
+    { concurrency: DEFAULT_IPNI_CONCURRENCY, signal }
   )
 }
 
@@ -176,8 +194,15 @@ async function applyIPNIFallback({ b58s, index, shardsByB58, fetcher }) {
  * @param {Map<string, ClaimsEntry>} args.index
  * @param {Map<string, ShardEntry>} args.shardsByB58
  * @param {typeof fetch} args.fetcher
+ * @param {AbortSignal | undefined} [args.signal]
  */
-async function applyCarparkFallback({ b58s, index, shardsByB58, fetcher }) {
+async function applyCarparkFallback({
+  b58s,
+  index,
+  shardsByB58,
+  fetcher,
+  signal,
+}) {
   await pMap(
     b58s,
     async (b58) => {
@@ -195,16 +220,17 @@ async function applyCarparkFallback({ b58s, index, shardsByB58, fetcher }) {
         entry.size = match.size
       }
     },
-    { concurrency: DEFAULT_CARPARK_CONCURRENCY }
+    { concurrency: DEFAULT_CARPARK_CONCURRENCY, signal }
   )
 }
 
 /**
  * @param {typeof fetch} fetcher
  * @param {string} b58
+ * @param {AbortSignal | undefined} [signal]
  * @returns {Promise<IPNIProviderResult[]>}
  */
-async function fetchIPNIProviderResults(fetcher, b58) {
+async function fetchIPNIProviderResults(fetcher, b58, signal) {
   try {
     const value = b58.charAt(0) === 'z' ? b58.substring(1) : b58
     const response = await fetcher(`${CID_CONTACT_URL}/multihash/${value}`, {
@@ -214,7 +240,8 @@ async function fetchIPNIProviderResults(fetcher, b58) {
 
     const body = /** @type {IPNIFindResponse} */ (await response.json())
     return body.MultihashResults?.[0]?.ProviderResults ?? []
-  } catch {
+  } catch (error) {
+    if (isAbortError(error, signal)) throw error
     return []
   }
 }