From f3d75d31fe85215ce5dfe54833a06bea8ddb9143 Mon Sep 17 00:00:00 2001 From: bravonatalie Date: Mon, 11 May 2026 15:30:03 -0300 Subject: [PATCH] chore: add abort support to the reader phase --- .../version-plan-1778524089700.md | 6 + packages/cli/migrate.js | 6 + packages/filecoin-pin-migration/src/api.ts | 2 + packages/filecoin-pin-migration/src/errors.js | 23 +++ .../src/reader/carpark.js | 5 +- .../src/reader/indexer.js | 45 ++++-- .../src/reader/reader.js | 84 +++++++--- .../test/reader.spec.js | 143 ++++++++++++++++++ 8 files changed, 286 insertions(+), 28 deletions(-) create mode 100644 .nx/version-plans/version-plan-1778524089700.md diff --git a/.nx/version-plans/version-plan-1778524089700.md b/.nx/version-plans/version-plan-1778524089700.md new file mode 100644 index 000000000..52d91deb6 --- /dev/null +++ b/.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 diff --git a/packages/cli/migrate.js b/packages/cli/migrate.js index 17c5c98f7..3f720cbb5 100644 --- a/packages/cli/migrate.js +++ b/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 } } diff --git a/packages/filecoin-pin-migration/src/api.ts b/packages/filecoin-pin-migration/src/api.ts index 797e5c268..cd3e399fb 100644 --- a/packages/filecoin-pin-migration/src/api.ts +++ b/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 diff --git a/packages/filecoin-pin-migration/src/errors.js b/packages/filecoin-pin-migration/src/errors.js index 56c5ecb66..19ac8f7d6 100644 --- a/packages/filecoin-pin-migration/src/errors.js +++ b/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') +} diff --git a/packages/filecoin-pin-migration/src/reader/carpark.js b/packages/filecoin-pin-migration/src/reader/carpark.js index e97154637..3c6b727bb 100644 --- a/packages/filecoin-pin-migration/src/reader/carpark.js +++ b/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 } } diff --git a/packages/filecoin-pin-migration/src/reader/indexer.js b/packages/filecoin-pin-migration/src/reader/indexer.js index c04047384..3bd7542ad 100644 --- a/packages/filecoin-pin-migration/src/reader/indexer.js +++ b/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>} */ -export async function resolveClaimsIndex({ indexer, shards, fetcher }) { +export async function resolveClaimsIndex({ indexer, shards, fetcher, signal }) { /** @type {Map} */ 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} args.index * @param {Map} 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} args.index * @param {Map} 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} */ -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 [] } } diff --git a/packages/filecoin-pin-migration/src/reader/reader.js b/packages/filecoin-pin-migration/src/reader/reader.js index a13fd63dd..1affb7ff2 100644 --- a/packages/filecoin-pin-migration/src/reader/reader.js +++ b/packages/filecoin-pin-migration/src/reader/reader.js @@ -5,6 +5,7 @@ import { DEFAULT_SHARD_LIST_CONCURRENCY, DEFAULT_STOP_ON_ERROR, } from '../constants.js' +import { isAbortError, throwIfAborted } from '../errors.js' import { checkpointInventoryPage } from '../state.js' import { resolveClaimsIndex } from './indexer.js' @@ -54,13 +55,21 @@ export async function* buildMigrationInventories({ state, spaceDIDs, uploadRootsBySpace, + signal, options, }) { - const indexer = - options?.indexer ?? new IndexingClient({ serviceURL: options?.serviceURL }) const stopOnError = options?.stopOnError ?? DEFAULT_STOP_ON_ERROR const shardListConcurrency = DEFAULT_SHARD_LIST_CONCURRENCY - const fetcher = options?.fetcher ?? globalThis.fetch + const fetcher = createSignalAwareFetch( + options?.fetcher ?? globalThis.fetch, + signal + ) + const indexer = + options?.indexer ?? + new IndexingClient({ + serviceURL: options?.serviceURL, + fetch: fetcher, + }) if (spaceDIDs && uploadRootsBySpace) { throw new TypeError( 'buildMigrationInventories: pass either "spaceDIDs" or "uploadRootsBySpace", not both' @@ -74,6 +83,7 @@ export async function* buildMigrationInventories({ ) for (const did of dids) { + if (signal?.aborted) return const spaceDID = /** @type {SpaceDID} */ (did) // Space already fully read — skip if ( @@ -83,19 +93,26 @@ export async function* buildMigrationInventories({ continue } - yield* buildSpaceInventory({ - client, - indexer, - resolver, - spaceDID, - state, - selectedUploadRoots: uploadRootsBySpace?.[spaceDID], - stopOnError, - shardListConcurrency, - fetcher, - }) + try { + yield* buildSpaceInventory({ + client, + indexer, + resolver, + spaceDID, + state, + selectedUploadRoots: uploadRootsBySpace?.[spaceDID], + stopOnError, + shardListConcurrency, + fetcher, + signal, + }) + } catch (error) { + if (isAbortError(error, signal)) return + throw error + } } + if (signal?.aborted) return state.phase = 'planning' yield { type: 'reader:complete' } yield { type: 'state:checkpoint', state } @@ -122,6 +139,7 @@ export async function* buildMigrationInventories({ * @param {boolean} args.stopOnError * @param {number} args.shardListConcurrency * @param {typeof fetch | undefined} args.fetcher + * @param {AbortSignal | undefined} args.signal * @returns {AsyncGenerator} */ async function* buildSpaceInventory({ @@ -134,10 +152,13 @@ async function* buildSpaceInventory({ stopOnError, shardListConcurrency, fetcher, + signal, }) { yield { type: 'reader:space:start', spaceDID } + throwIfAborted(signal) await client.setCurrentSpace(spaceDID) + throwIfAborted(signal) const spaceName = client.currentSpace?.()?.name || undefined let cursor = state.readerProgressCursors?.[spaceDID] @@ -145,10 +166,13 @@ async function* buildSpaceInventory({ selectedUploadRoots != null ? new Set(selectedUploadRoots) : undefined do { + throwIfAborted(signal) const page = await client.capability.upload.list({ cursor, size: 100, + signal, }) + throwIfAborted(signal) const uploadsInPage = selectedRoots ? page.results.filter((upload) => selectedRoots.has(upload.root.toString()) @@ -160,11 +184,12 @@ async function* buildSpaceInventory({ uploadsInPage, async (upload) => { const root = upload.root.toString() - const shards = await listShardsFromStore(client, upload.root) + const shards = await listShardsFromStore(client, upload.root, signal) return { root, shards } }, - { concurrency: shardListConcurrency } + { concurrency: shardListConcurrency, signal } ) + throwIfAborted(signal) // Phase 2: query the primary indexer, then repair missing claims from IPNI. const allShards = uploadsWithShards.flatMap((u) => u.shards) @@ -172,7 +197,9 @@ async function* buildSpaceInventory({ indexer, shards: allShards, fetcher, + signal, }) + throwIfAborted(signal) // Phase 3: extract per-shard results from the claims index (pure, no I/O) /** @type {ResolvedShard[]} */ @@ -262,14 +289,18 @@ async function* buildSpaceInventory({ * * @param {import('@storacha/client').Client} client * @param {UnknownLink} root + * @param {AbortSignal | undefined} signal * @returns {Promise} */ -async function listShardsFromStore(client, root) { +async function listShardsFromStore(client, root, signal) { /** @type {ShardEntry[]} */ const shards = [] let cursor do { - const page = await client.capability.upload.shard.list(root, { cursor }) + const page = await client.capability.upload.shard.list(root, { + cursor, + signal, + }) for (const link of page.results) { const b58 = base58btc.encode(link.multihash.bytes) shards.push({ cidStr: link.toString(), multihash: link.multihash, b58 }) @@ -330,3 +361,20 @@ function extractShard(claimsIndex, shard, root, resolver) { return { ok: resolved } } + +/** + * @param {typeof fetch | undefined} fetcher + * @param {AbortSignal | undefined} signal + * @returns {typeof fetch | undefined} + */ +function createSignalAwareFetch(fetcher, signal) { + if (typeof fetcher !== 'function' || !signal) return fetcher + + return /** @type {typeof fetch} */ ( + (input, init = {}) => + fetcher(input, { + ...init, + signal: init.signal ?? signal, + }) + ) +} diff --git a/packages/filecoin-pin-migration/test/reader.spec.js b/packages/filecoin-pin-migration/test/reader.spec.js index eee497dd5..23f8879d4 100644 --- a/packages/filecoin-pin-migration/test/reader.spec.js +++ b/packages/filecoin-pin-migration/test/reader.spec.js @@ -41,6 +41,10 @@ async function collectInventory(gen, state, spaceDID) { return state.spacesInventories[spaceDID] } +function createAbortError() { + return new DOMException('The operation was aborted.', 'AbortError') +} + describe('buildMigrationInventories', () => { describe('single space — basic inventory', () => { it('resolves shards and builds flat inventory with root on each shard', async () => { @@ -979,4 +983,143 @@ describe('buildMigrationInventories', () => { expect(state.readerProgressCursors).toBeUndefined() }) }) + + describe('abort support', () => { + it('stops after the last checkpointed page and leaves state resumable', async () => { + const rootA = await createTestCID('root-abort-page-a') + const rootB = await createTestCID('root-abort-page-b') + const shardA = await createTestCID('shard-abort-page-a') + const shardB = await createTestCID('shard-abort-page-b') + const pieceCid = createPieceCID() + const shardAB58 = base58btc.encode(shardA.multihash.bytes) + const shardBB58 = base58btc.encode(shardB.multihash.bytes) + + const client = createMockClient( + [ + { results: [{ root: rootA }], cursor: '1' }, + { results: [{ root: rootB }] }, + ], + new Map([ + [rootA.toString(), [shardA]], + [rootB.toString(), [shardB]], + ]) + ) + const indexer = createMockIndexer( + new Map([ + [ + shardAB58, + { + claims: buildShardClaims(shardA, { + locationURLs: ['https://r2.example/abort-a'], + pieceCid, + }), + }, + ], + [ + shardBB58, + { + claims: buildShardClaims(shardB, { + locationURLs: ['https://r2.example/abort-b'], + pieceCid, + }), + }, + ], + ]) + ) + + const ac = new AbortController() + const state = createMockInitialState() + /** @type {API.MigrationEvent[]} */ + const events = [] + + for await (const event of buildMigrationInventories({ + client, + resolver: claimsResolver, + state, + spaceDIDs: [SPACE_DID], + signal: ac.signal, + options: { indexer }, + })) { + events.push(event) + if (event.type === 'state:checkpoint') { + ac.abort() + } + } + + expect(events.map((event) => event.type)).toEqual([ + 'reader:space:start', + 'state:checkpoint', + ]) + expect(state.phase).toBe('reading') + expect(state.readerProgressCursors).toEqual({ [SPACE_DID]: '1' }) + expect(state.spacesInventories[SPACE_DID]?.uploads).toEqual([ + rootA.toString(), + ]) + expect(state.spacesInventories[SPACE_DID]?.shards).toHaveLength(1) + }) + + it('returns cleanly when a reader request is aborted in flight', async () => { + const client = /** @type {import('@storacha/client').Client} */ ( + /** @type {unknown} */ ({ + spaces() { + return [] + }, + async setCurrentSpace(/** @type {API.SpaceDID} */ _did) {}, + capability: { + upload: { + async list( + /** @type {{ signal?: AbortSignal } | undefined} */ options + ) { + return await new Promise((_resolve, reject) => { + if (options?.signal?.aborted) { + reject(createAbortError()) + return + } + options?.signal?.addEventListener( + 'abort', + () => reject(createAbortError()), + { once: true } + ) + }) + }, + shard: { + async list( + /** @type {API.UnknownLink} */ _root, + /** @type {unknown} */ _options + ) { + return { results: [] } + }, + }, + }, + }, + }) + ) + + const ac = new AbortController() + const state = createMockInitialState() + /** @type {API.MigrationEvent[]} */ + const events = [] + + const run = (async () => { + for await (const event of buildMigrationInventories({ + client, + resolver: claimsResolver, + state, + spaceDIDs: [SPACE_DID], + signal: ac.signal, + options: { indexer: createMockIndexer(new Map()) }, + })) { + events.push(event) + } + })() + + ac.abort() + await run + + expect(events.map((event) => event.type)).toEqual(['reader:space:start']) + expect(state.phase).toBe('reading') + expect(state.spacesInventories[SPACE_DID]).toBeUndefined() + expect(state.readerProgressCursors).toBeUndefined() + }) + }) })