feat: add client-side max request duration timeout to Spark (#99) (#124)

* feat: add client-side max request duration timeout to Spark (#99)

* refactor(fetchCAR): cleanup based on review feedback

- Moved MAX_REQUEST_DURATION_MS to constants
- Fixed indentation
- Removed maxTimeout field in favor of unified timeout

* Refactor fetchCAR timeouts for clarity and testability

* feat: add jitter to delay between tasks (#125)

feat: add jitter to delay between tasks

Introduce random jitter in the delay between subsequent tasks (retrieval
checks).

The goal is to better spread out the requests from the network in the
time, to avoid too many nodes hitting services like cid.contact at the
same time.

Signed-off-by: Miroslav Bajtoš <oss@bajtos.net>
Co-authored-by: Julian Gruber <julian@juliangruber.com>

* v1.18.2

* chore: rename maxFetchDurationMs to maxRequestDurationMs for consistency

* feat: add client-side max request duration timeout to Spark (#99)

* clean up

* fix indentation

* clean up

---------

Signed-off-by: Miroslav Bajtoš <oss@bajtos.net>
Co-authored-by: Miroslav Bajtoš <oss@bajtos.net>
Co-authored-by: Julian Gruber <julian@juliangruber.com>

Author: 3 people

lib/constants.js

@@ -5,3 +5,4 @@ export const MAX_JITTER_BETWEEN_TASKS_IN_MS = 10_000 // 10 seconds
 export const RPC_URL = 'https://api.node.glif.io/'
 export const RPC_AUTH = 'KZLIUb9ejreYOm-mZFM3UNADE0ux6CrHjxnS2D2Qgb8='
 export const MINER_TO_PEERID_CONTRACT_ADDRESS = '0x14183aD016Ddc83D638425D6328009aa390339Ce' // Contract address on the Filecoin EVM
+export const MAX_REQUEST_DURATION_MS = 90_000

lib/spark.js

@@ -1,7 +1,7 @@
 /* global Zinnia */
 
 import { ActivityState } from './activity-state.js'
-import { SPARK_VERSION, MAX_CAR_SIZE, APPROX_ROUND_LENGTH_IN_MS, MAX_JITTER_BETWEEN_TASKS_IN_MS } from './constants.js'
+import { SPARK_VERSION, MAX_CAR_SIZE, APPROX_ROUND_LENGTH_IN_MS, MAX_JITTER_BETWEEN_TASKS_IN_MS, MAX_REQUEST_DURATION_MS } from './constants.js'
 import { queryTheIndex } from './ipni-client.js'
 import { assertOkResponse } from './http-assertions.js'
 import { getIndexProviderPeerId as defaultGetIndexProvider } from './miner-info.js'
@@ -83,20 +83,27 @@ export default class Spark {
     }
   }
 
-  async fetchCAR (protocol, address, cid, stats) {
+  async fetchCAR (protocol, address, cid, stats, { maxRequestDurationMs = MAX_REQUEST_DURATION_MS } = {}) {
     // Abort if no progress was made for 60 seconds
     const controller = new AbortController()
     const { signal } = controller
-    let timeout
+
+    let requestIdleTimeout
+    let maxDurationTimeout
+
     const resetTimeout = () => {
-      if (timeout) {
-        clearTimeout(timeout)
+      if (requestIdleTimeout) {
+        clearTimeout(requestIdleTimeout)
       }
-      timeout = setTimeout(() => {
+      requestIdleTimeout = setTimeout(() => {
         stats.timeout = true
         controller.abort()
       }, 60_000)
     }
+    maxDurationTimeout = setTimeout(() => {
+      stats.timeout = true
+      controller.abort()
+    }, maxRequestDurationMs)
 
     // WebCrypto API does not support streams yet, the hashing function requires entire data
     // to be provided at once. See https://github.com/w3c/webcrypto/issues/73
@@ -157,7 +164,8 @@ export default class Spark {
         stats.statusCode = mapErrorToStatusCode(err)
       }
     } finally {
-      clearTimeout(timeout)
+      clearTimeout(requestIdleTimeout)
+      clearTimeout(maxDurationTimeout)
     }
 
     stats.endAt = new Date()

test/spark.js

@@ -420,3 +420,31 @@ test('calculateDelayBeforeNextTask() introduces random jitter for zero tasks in
     `expected delay values to be within 1 second of each other. Actual values: ${delay1} <> ${delay2}`
   )
 })
+
+test('fetchCAR triggers timeout after long retrieval', async () => {
+  const sleep = (ms) => new Promise(resolve => setTimeout(resolve, ms))
+  const fetch = async (_url, { signal }) => {
+    return {
+      status: 200,
+      ok: true,
+      body: (async function * () {
+        while (true) {
+          if (signal.aborted) {
+            throw new DOMException('Aborted', 'AbortError')
+          }
+          yield new Uint8Array([0])
+          await sleep(500)
+        }
+      })()
+    }
+  }
+
+  const spark = new Spark({ fetch })
+  const stats = newStats()
+
+  await spark.fetchCAR('http', '/dns/example.com/tcp/80/http', KNOWN_CID, stats, {
+    maxRequestDurationMs: 0
+  })
+
+  assertEquals(stats.timeout, true)
+})