fix: handle aborted requests properly (#241)

* fix: abort signal handling when walking the path

* fix: remove all throwIfAborted calls

* chore: normalize check for aborted signal

Author: SgtPooki

packages/interop/src/abort-handling.spec.ts

@@ -0,0 +1,38 @@
+/* eslint-env mocha */
+import { createVerifiedFetch } from '@helia/verified-fetch'
+import { expect } from 'aegir/chai'
+import type { VerifiedFetch } from '@helia/verified-fetch'
+
+describe('verified-fetch abort handling', () => {
+  let verifiedFetch: VerifiedFetch
+  before(async () => {
+    if (process.env.KUBO_DIRECT_RETRIEVAL_ROUTER == null || process.env.KUBO_DIRECT_RETRIEVAL_ROUTER === '') {
+      throw new Error('KUBO_DIRECT_RETRIEVAL_ROUTER environment variable is required')
+    }
+
+    verifiedFetch = await createVerifiedFetch({
+      gateways: [process.env.KUBO_DIRECT_RETRIEVAL_ROUTER],
+      routers: [process.env.KUBO_DIRECT_RETRIEVAL_ROUTER],
+      allowInsecure: true,
+      allowLocal: true
+    })
+  })
+
+  after(async () => {
+    await verifiedFetch.stop()
+  })
+
+  it('should handle aborts properly', async function () {
+    this.timeout(2000)
+    const controller = new AbortController()
+    const timeout = setTimeout(() => {
+      controller.abort()
+    }, 70)
+
+    const fetchPromise = verifiedFetch('ipfs://QmdmQXB2mzChmMeKY47C43LxUdg1NDJ5MWcKMKxDu7RgQm/1 - Barrel - Part 1/1 - Barrel - Part 1 - alt.txt', {
+      signal: controller.signal
+    })
+    await expect(fetchPromise).to.eventually.be.rejected.with.property('name', 'AbortError')
+    clearTimeout(timeout)
+  })
+})

packages/verified-fetch/src/plugins/plugin-handle-dag-pb.ts

@@ -1,5 +1,6 @@
 import { unixfs } from '@helia/unixfs'
 import { code as dagPbCode } from '@ipld/dag-pb'
+import { AbortError } from '@libp2p/interface'
 import { exporter } from 'ipfs-unixfs-exporter'
 import { CustomProgressEvent } from 'progress-events'
 import { getStreamFromAsyncIterable } from '../utils/get-stream-from-async-iterable.js'
@@ -98,8 +99,10 @@ export class DagPbPlugin extends BasePlugin {
         path = rootFilePath
         resolvedCID = entry.cid
       } catch (err: any) {
+        if (options?.signal?.aborted) {
+          throw new AbortError(options?.signal?.reason)
+        }
         this.log.error('error loading path %c/%s', dirCid, rootFilePath, err)
-        options?.signal?.throwIfAborted()
         context.isDirectory = true
         context.directoryEntries = []
         context.modified++
@@ -161,7 +164,9 @@ export class DagPbPlugin extends BasePlugin {
 
       return response
     } catch (err: any) {
-      options?.signal?.throwIfAborted()
+      if (options?.signal?.aborted) {
+        throw new AbortError(options?.signal?.reason)
+      }
       log.error('error streaming %c/%s', cid, path, err)
       if (byteRangeContext.isRangeRequest && err.code === 'ERR_INVALID_PARAMS') {
         return badRangeResponse(resource)

packages/verified-fetch/src/plugins/plugin-handle-dag-walk.ts

@@ -29,10 +29,10 @@ export class DagWalkPlugin extends BasePlugin {
     const { cid, resource, options, withServerTiming = false } = context
     const { getBlockstore, handleServerTiming } = this.pluginOptions
     const blockstore = getBlockstore(cid, resource, options?.session ?? true, options)
+
     // TODO: migrate handlePathWalking into this plugin
     const pathDetails = await handleServerTiming('path-walking', '', async () => handlePathWalking({ ...context, blockstore, log: this.log }), withServerTiming)
 
-    context.modified++
     if (pathDetails instanceof Response) {
       this.log.trace('path walking failed')
 
@@ -42,10 +42,11 @@ export class DagWalkPlugin extends BasePlugin {
         return pathDetails
       }
 
-      // some error walking the path
+      // some other error walking the path (codec doesn't support pathing, etc..), let the next plugin try to handle it
       return null
     }
 
+    context.modified++
     context.pathDetails = pathDetails
 
     return null

packages/verified-fetch/src/utils/get-stream-from-async-iterable.ts

@@ -25,7 +25,7 @@ export async function getStreamFromAsyncIterable (iterator: AsyncIterable<Uint8A
     },
     async pull (controller) {
       const { value, done } = await reader.next()
-      if (options?.signal?.aborted === true) {
+      if (options?.signal?.aborted) {
         controller.error(new AbortError(options.signal.reason ?? 'signal aborted by user'))
         controller.close()
         return

packages/verified-fetch/src/utils/parse-url-string.ts

@@ -1,3 +1,4 @@
+import { AbortError } from '@libp2p/interface'
 import { CID } from 'multiformats/cid'
 import { getPeerIdFromString } from './get-peer-id-from-string.js'
 import { serverTiming } from './server-timing.js'
@@ -212,7 +213,9 @@ export async function parseUrlString ({ urlString, ipns, logger, withServerTimin
         resolvedPath = resolveResult?.path
         log.trace('resolved %s to %c', cidOrPeerIdOrDnsLink, cid)
       } catch (err) {
-        options?.signal?.throwIfAborted()
+        if (options?.signal?.aborted) {
+          throw new AbortError(options?.signal?.reason)
+        }
         if (peerId == null) {
           log.error('could not parse PeerId string "%s"', cidOrPeerIdOrDnsLink, err)
           errors.push(new TypeError(`Could not parse PeerId in ipns url "${cidOrPeerIdOrDnsLink}", ${(err as Error).message}`))
@@ -249,7 +252,10 @@ export async function parseUrlString ({ urlString, ipns, logger, withServerTimin
           resolvedPath = resolveResult?.path
           log.trace('resolved %s to %c', decodedDnsLinkLabel, cid)
         } catch (err: any) {
-          options?.signal?.throwIfAborted()
+          // eslint-disable-next-line max-depth
+          if (options?.signal?.aborted) {
+            throw new AbortError(options?.signal?.reason)
+          }
           log.error('could not resolve DnsLink for "%s"', cidOrPeerIdOrDnsLink, err)
           errors.push(err)
         }

packages/verified-fetch/src/utils/walk-path.ts

@@ -1,4 +1,5 @@
 import { DoesNotExistError } from '@helia/unixfs/errors'
+import { AbortError } from '@libp2p/interface'
 import { walkPath as exporterWalk } from 'ipfs-unixfs-exporter'
 import { badGatewayResponse, notFoundResponse } from './responses.js'
 import type { PluginContext } from '../plugins/types.js'
@@ -53,7 +54,10 @@ export async function handlePathWalking ({ cid, path, resource, options, blockst
   try {
     return await walkPath(blockstore, `${cid.toString()}/${path}`, options)
   } catch (err: any) {
-    options?.signal?.throwIfAborted()
+    if (options?.signal?.aborted) {
+      throw new AbortError(options?.signal?.reason)
+    }
+
     if (['ERR_NO_PROP', 'ERR_NO_TERMINAL_ELEMENT', 'ERR_NOT_FOUND'].includes(err.code)) {
       return notFoundResponse(resource)
     }

packages/verified-fetch/src/verified-fetch.ts

@@ -1,4 +1,5 @@
 import { ipns as heliaIpns } from '@helia/ipns'
+import { AbortError } from '@libp2p/interface'
 import { prefixLogger } from '@libp2p/logger'
 import { CustomProgressEvent } from 'progress-events'
 import QuickLRU from 'quick-lru'
@@ -282,7 +283,9 @@ export class VerifiedFetch {
             break
           }
         } catch (err: any) {
-          context.options?.signal?.throwIfAborted()
+          if (context.options?.signal?.aborted) {
+            throw new AbortError(context.options?.signal?.reason)
+          }
           this.log.error('Error in plugin:', plugin.constructor.name, err)
           // if fatal, short-circuit the pipeline
           if (err.name === 'PluginFatalError') {
@@ -341,7 +344,9 @@ export class VerifiedFetch {
       parsedResult = await this.handleServerTiming('parse-resource', '', async () => parseResource(resource, { ipns: this.ipns, logger: this.helia.logger }, { withServerTiming, ...options }), withServerTiming)
       this.serverTimingHeaders.push(...parsedResult.serverTimings.map(({ header }) => header))
     } catch (err: any) {
-      options?.signal?.throwIfAborted()
+      if (options?.signal?.aborted) {
+        throw new AbortError(options?.signal?.reason)
+      }
       this.log.error('error parsing resource %s', resource, err)
 
       return this.handleFinalResponse(badRequestResponse(resource.toString(), err))