Add optional abort signal to fetch json and merge with timeout

packages/cli/src/__tests__/ceramic-daemon.test.ts

@@ -3,7 +3,14 @@ import { Ceramic } from '@ceramicnetwork/core'
 import { CeramicClient } from '@ceramicnetwork/http-client'
 import tmp from 'tmp-promise'
 import { CeramicDaemon } from '../ceramic-daemon.js'
-import { AnchorStatus, fetchJson, Stream, StreamUtils, IpfsApi } from '@ceramicnetwork/common'
+import {
+  AnchorStatus,
+  fetchJson,
+  Stream,
+  StreamUtils,
+  IpfsApi,
+  TimedAbortSignal,
+} from '@ceramicnetwork/common'
 import { TileDocumentHandler } from '@ceramicnetwork/stream-tile-handler'
 import { TileDocument } from '@ceramicnetwork/stream-tile'
 import { firstValueFrom } from 'rxjs'
@@ -366,7 +373,7 @@ describe('Ceramic interop: core <> http-client', () => {
     expect(json).toEqual(content3)
   })
 
-  it('times out if fetch is taking too long', async () => {
+  it('Aborts fetch if it is taking too long', async () => {
     const content1 = { test: 123 }
     const doc = await TileDocument.create(core, content1, null, { anchor: false })
 
@@ -384,11 +391,77 @@ describe('Ceramic interop: core <> http-client', () => {
       fetchJson(`http://localhost:${daemon.port}/api/v0/streams/${doc.id}/content`, {
         timeout: 1000,
       })
-    ).rejects.toThrow(`Http request timed out after 1000 ms`)
+    ).rejects.toThrow(/aborted/)
 
     clearTimeout(id)
   })
 
+  it('Aborts fetch through passed in AbortSignal', async () => {
+    const content1 = { test: 123 }
+    const doc = await TileDocument.create(core, content1, null, { anchor: false })
+
+    const loadStreamMock = jest.spyOn(core, 'loadStream')
+    let id = null
+    loadStreamMock.mockImplementation(() => {
+      return new Promise((resolve) => {
+        id = setTimeout(() => {
+          resolve(doc)
+        }, 5000)
+      })
+    })
+
+    const timedAbortSignal = new TimedAbortSignal(1000)
+
+    await expect(
+      fetchJson(`http://localhost:${daemon.port}/api/v0/streams/${doc.id}/content`, {
+        signal: timedAbortSignal.signal,
+      })
+    ).rejects.toThrow(/aborted/)
+
+    timedAbortSignal.clear()
+    clearTimeout(id)
+  })
+
+  it('Aborts fetch if taking too long even if given an AbortSignal that did not get aborted', async () => {
+    const content1 = { test: 123 }
+    const doc = await TileDocument.create(core, content1, null, { anchor: false })
+
+    const loadStreamMock = jest.spyOn(core, 'loadStream')
+    let id = null
+    loadStreamMock.mockImplementation(() => {
+      return new Promise((resolve) => {
+        id = setTimeout(() => {
+          resolve(doc)
+        }, 4000)
+      })
+    })
+
+    const controller = new AbortController()
+
+    await expect(
+      fetchJson(`http://localhost:${daemon.port}/api/v0/streams/${doc.id}/content`, {
+        signal: controller.signal,
+        timeout: 1000,
+      })
+    ).rejects.toThrow(/aborted/)
+
+    clearTimeout(id)
+  })
+
+  it('Aborts fetch if the AbortSignal given has already been aborted', async () => {
+    const content1 = { test: 123 }
+    const doc = await TileDocument.create(core, content1, null, { anchor: false })
+
+    const controller = new AbortController()
+    controller.abort()
+
+    await expect(
+      fetchJson(`http://localhost:${daemon.port}/api/v0/streams/${doc.id}/content`, {
+        signal: controller.signal,
+      })
+    ).rejects.toThrow(/aborted/)
+  })
+
   it('requestAnchor works via http api', async () => {
     const content1 = { test: 123 }
 

packages/common/src/index.ts

@@ -8,6 +8,7 @@ export * from './utils/stream-utils.js'
 export * from './utils/test-utils.js'
 export * from './utils/accountid-utils.js'
 export * from './utils/cid-utils.js'
+export * from './utils/abort-signal-utils.js'
 export * from './logger-provider.js'
 export * from './loggers.js'
 export * from './networks.js'

packages/common/src/utils/abort-signal-utils.ts

@@ -0,0 +1,45 @@
+import { timer, fromEvent, merge, Subscription } from 'rxjs'
+import { first } from 'rxjs/operators'
+
+export function mergeAbortSignals(signals: AbortSignal[]): AbortSignal {
+  const controller = new AbortController()
+
+  if (signals.length === 0) {
+    throw Error('Need abort signals to create a merged abort signal')
+  }
+
+  if (signals.some((signal) => signal.aborted)) {
+    controller.abort()
+    return controller.signal
+  }
+
+  merge(...signals.map((signal) => fromEvent(signal, 'abort')))
+    .pipe(first())
+    .subscribe(() => {
+      controller.abort()
+    })
+
+  return controller.signal
+}
+export class TimedAbortSignal {
+  private readonly _subscription: Subscription
+  readonly signal: AbortSignal
+
+  constructor(timeout: number) {
+    const controller = new AbortController()
+    this.signal = controller.signal
+
+    if (timeout <= 0) {
+      controller.abort()
+      return
+    }
+
+    this._subscription = timer(timeout).subscribe(() => {
+      controller.abort()
+    })
+  }
+
+  clear() {
+    this._subscription?.unsubscribe()
+  }
+}

packages/common/src/utils/http-utils.ts

@@ -1,11 +1,13 @@
 import fetch from 'cross-fetch'
+import { mergeAbortSignals, TimedAbortSignal } from './abort-signal-utils'
 
 const DEFAULT_FETCH_TIMEOUT = 60 * 1000 * 3 // 3 minutes
 interface FetchOpts {
   body?: any
   method?: string
   headers?: any
   timeout?: number
+  signal?: AbortSignal
 }
 
 export async function fetchJson(url: string, opts: FetchOpts = {}): Promise<any> {
@@ -17,22 +19,13 @@ export async function fetchJson(url: string, opts: FetchOpts = {}): Promise<any>
   }
 
   const timeoutLength = opts.timeout || DEFAULT_FETCH_TIMEOUT
-  const controller = new AbortController()
-  const timeout = setTimeout(() => {
-    controller.abort()
-  }, timeoutLength)
-  Object.assign(opts, {
-    signal: controller.signal,
-  })
+  const timedAbortSignal = new TimedAbortSignal(timeoutLength)
 
-  const res = await fetch(url, opts)
-    .catch((err) => {
-      if (err.name == 'AbortError') {
-        throw new Error(`Http request timed out after ${timeoutLength} ms`)
-      }
-      throw err
-    })
-    .finally(() => timeout && clearTimeout(timeout))
+  opts.signal = opts.signal
+    ? mergeAbortSignals([opts.signal, timedAbortSignal.signal])
+    : timedAbortSignal.signal
+
+  const res = await fetch(url, opts).finally(() => timedAbortSignal.clear())
 
   if (!res.ok) {
     const text = await res.text()