Support adaptive data feed slice replay

Author: thaaddeus

ARCHITECTURE.md

@@ -13,18 +13,18 @@ Main Thread                         Worker Thread
   │                                     │
   │── Start replay ──→                  │
   │                         Fetch data slice from API
-  │                         Cache to disk (.gz file)
-  │  ←── message (sliceKey, path) ──    │
+  │                         Cache to disk (.gz/.zst file)
+  │  ←── message (sliceKey, path, size) ──
   │                         Fetch next slice...
   │                                     │
   Read cached file from disk            │
-  Decompress (gunzip)                   │
+  Decompress (gzip/zstd)                │
   Split by newlines                     │
   Parse JSON messages                   │
   Yield {localTimestamp, message}       │
 ```
 
-Worker thread pre-fetches and caches slices while the main thread processes the current one. This keeps I/O and CPU pipelined.
+Worker thread pre-fetches and caches slices while the main thread processes the current one. This keeps I/O and CPU pipelined. Normal replay fetches the first and last minute as one-minute requests, uses the returned suggested slice size for the middle of the range, and caches multi-minute responses as start-minute files with a `.size-{minutes}` suffix. One-minute cache paths keep the legacy filename.
 
 ## Real-time Streaming
 

src/handy.ts

@@ -628,6 +628,7 @@ async function _downloadFile(requestOptions: RequestOptions, url: string, downlo
 
   try {
     // based on https://github.com/nodejs/node/issues/28172 - only reliable way to consume response stream and avoiding all the 'gotchas'
+    let responseHeaders: Record<string, string> = {}
     await new Promise<void>((resolve, reject) => {
       const req = https
         .get(url, requestOptions, (res) => {
@@ -642,6 +643,7 @@ async function _downloadFile(requestOptions: RequestOptions, url: string, downlo
               reject(new HttpError(statusCode!, body, url))
             })
           } else {
+            responseHeaders = parseNodeResponseHeaders(res.headers)
             if (appendContentEncodingExtension) {
               const contentEncoding = asSingleHeaderValue(res.headers['content-encoding'])
               if (contentEncoding === 'zstd') {
@@ -682,7 +684,8 @@ async function _downloadFile(requestOptions: RequestOptions, url: string, downlo
     await rename(tmpFilePath, finalDownloadPath)
 
     return {
-      downloadPath: finalDownloadPath
+      downloadPath: finalDownloadPath,
+      headers: responseHeaders
     }
   } finally {
     tmpFileCleanups.delete(tmpFilePath)

src/replay.ts

@@ -42,7 +42,7 @@ export async function* replay<T extends Exchange, U extends boolean = false, Z e
 
   const fromDate = parseAsUTCDate(from)
   const toDate = parseAsUTCDate(to)
-  const cachedSlicePaths = new Map<string, string>()
+  const cachedSlicePaths = new Map<string, { slicePath: string; sliceSize: number }>()
   let replayError
   debug('replay for exchange: %s started - from: %s, to: %s, filters: %o', exchange, fromDate.toISOString(), toDate.toISOString(), filters)
 
@@ -65,7 +65,10 @@ export async function* replay<T extends Exchange, U extends boolean = false, Z e
   const worker = new ReliableWorker(payload)
 
   worker.on('message', (message: WorkerMessage) => {
-    cachedSlicePaths.set(message.sliceKey, message.slicePath)
+    cachedSlicePaths.set(message.sliceKey, {
+      slicePath: message.slicePath,
+      sliceSize: message.sliceSize
+    })
   })
 
   worker.on('error', (err) => {
@@ -100,23 +103,24 @@ export async function* replay<T extends Exchange, U extends boolean = false, Z e
 
       debug('getting slice: %s, exchange: %s', sliceKey, exchange)
 
-      let cachedSlicePath
-      while (cachedSlicePath === undefined) {
-        cachedSlicePath = cachedSlicePaths.get(sliceKey)
+      let cachedSlice
+      while (cachedSlice === undefined) {
+        cachedSlice = cachedSlicePaths.get(sliceKey)
 
         // if something went wrong(network issue, auth issue, gunzip issue etc)
         if (replayError !== undefined) {
           throw replayError
         }
 
-        if (cachedSlicePath === undefined) {
+        if (cachedSlice === undefined) {
           // if response for requested date is not ready yet wait 100ms and try again
           debug('waiting for slice: %s, exchange: %s', sliceKey, exchange)
           await wait(100)
         }
       }
 
       // response is a path to file on disk let' read it as stream
+      const { slicePath: cachedSlicePath, sliceSize } = cachedSlice
       const isZstdSlice = cachedSlicePath.endsWith('.zst')
       const linesStream = createReadStream(cachedSlicePath, { highWaterMark: CHUNK_SIZE })
         // decompress it while preserving the on-disk cache in the negotiated wire format
@@ -185,8 +189,8 @@ export async function* replay<T extends Exchange, U extends boolean = false, Z e
       if (autoCleanup) {
         await cleanupSlice(cachedSlicePath)
       }
-      // move one minute forward
-      currentSliceDate.setUTCMinutes(currentSliceDate.getUTCMinutes() + 1)
+      // move by the number of minutes covered by this cached response
+      currentSliceDate.setUTCMinutes(currentSliceDate.getUTCMinutes() + sliceSize)
     }
 
     debug(

src/worker.ts

@@ -7,6 +7,8 @@ import type { DataFeedCompression } from './options.ts'
 import { Exchange, Filter } from './types.ts'
 const debug = dbg('tardis-dev')
 
+const DEFAULT_DATA_FEED_SLICE_SIZE = 1
+
 if (isMainThread) {
   debug('current worker is not meant to run in main thread')
 } else {
@@ -77,15 +79,32 @@ async function getDataFeedSlices(payload: WorkerJobPayload) {
     }
   } else {
     // fetch last slice - it will tell us if user has access to the end of requested date range and data is available
-    await getDataFeedSlice(payload, minutesCountToFetch - 1, filters, cacheDir)
+    // also fetch it from API to get current suggested slice size headers
+    const lastSlice = await getDataFeedSlice(payload, minutesCountToFetch - 1, filters, cacheDir, DEFAULT_DATA_FEED_SLICE_SIZE, false)
 
     // fetch first slice - it will tell us if user has access to the beginning of requested date range
-    await getDataFeedSlice(payload, 0, filters, cacheDir)
+    const firstSlice =
+      minutesCountToFetch === 1 ? lastSlice : await getDataFeedSlice(payload, 0, filters, cacheDir, DEFAULT_DATA_FEED_SLICE_SIZE, false)
+
+    const replaySliceSize =
+      filters.length === 0 ? DEFAULT_DATA_FEED_SLICE_SIZE : Math.max(firstSlice.suggestedSliceSize, lastSlice.suggestedSliceSize)
+    const sliceOffsets: number[] = []
+    for (let offset = 1; offset < minutesCountToFetch - 1; offset += replaySliceSize) {
+      sliceOffsets.push(offset)
+    }
 
     // it both begining and end date of the range is accessible fetch all remaning slices concurently with CONCURRENCY_LIMIT
     await pMap(
-      sequence(minutesCountToFetch, 1), // this will produce Iterable sequence from 1 to minutesCountToFetch
-      (offset) => getDataFeedSlice(payload, offset, filters, cacheDir),
+      sliceOffsets,
+      async (offset) => {
+        let currentOffset = offset
+        let remainingSliceSize = Math.min(replaySliceSize, minutesCountToFetch - 1 - offset)
+        while (remainingSliceSize > 0) {
+          const result = await getDataFeedSlice(payload, currentOffset, filters, cacheDir, remainingSliceSize)
+          currentOffset += result.sliceSize
+          remainingSliceSize -= result.sliceSize
+        }
+      },
       { concurrency: CONCURRENCY_LIMIT }
     )
   }
@@ -95,51 +114,73 @@ async function getDataFeedSlice(
   { exchange, fromDate, endpoint, apiKey, dataFeedCompression, userAgent }: WorkerJobPayload,
   offset: number,
   filters: object[],
-  cacheDir: string
+  cacheDir: string,
+  requestedSliceSize = DEFAULT_DATA_FEED_SLICE_SIZE,
+  useCache = true
 ) {
   const sliceTimestamp = addMinutes(fromDate, offset)
   const sliceKey = sliceTimestamp.toISOString()
-  const sliceBasePath = `${cacheDir}/${formatDateToPath(sliceTimestamp)}.json`
+  const sliceSizeSuffix = requestedSliceSize === DEFAULT_DATA_FEED_SLICE_SIZE ? '' : `.size-${requestedSliceSize}`
+  const sliceBasePath = `${cacheDir}/${formatDateToPath(sliceTimestamp)}${sliceSizeSuffix}.json`
   const zstdSlicePath = `${sliceBasePath}.zst`
   const gzipSlicePath = `${sliceBasePath}.gz`
-  const cachedSlicePath = existsSync(zstdSlicePath) ? zstdSlicePath : existsSync(gzipSlicePath) ? gzipSlicePath : undefined
+  let cachedSlicePath
+  if (useCache) {
+    cachedSlicePath = existsSync(zstdSlicePath) ? zstdSlicePath : existsSync(gzipSlicePath) ? gzipSlicePath : undefined
+  }
+
+  if (cachedSlicePath !== undefined) {
+    debug('getDataFeedSlice already cached: %s, sliceSize: %d', sliceKey, requestedSliceSize)
+    const message: WorkerMessage = {
+      sliceKey,
+      slicePath: cachedSlicePath,
+      sliceSize: requestedSliceSize
+    }
+    parentPort!.postMessage(message)
+    return {
+      sliceSize: requestedSliceSize,
+      suggestedSliceSize: DEFAULT_DATA_FEED_SLICE_SIZE
+    }
+  }
 
   let url = `${endpoint}/data-feeds/${exchange}?from=${fromDate.toISOString()}&offset=${offset}&compression=${dataFeedCompression}`
+  if (requestedSliceSize > DEFAULT_DATA_FEED_SLICE_SIZE) {
+    url += `&sliceSize=${requestedSliceSize}`
+  }
 
   if (filters.length > 0) {
     url += `&filters=${encodeURIComponent(JSON.stringify(filters))}`
   }
 
-  const slicePath =
-    cachedSlicePath ||
-    (
-      await download({
-        apiKey,
-        downloadPath: sliceBasePath,
-        url,
-        userAgent,
-        appendContentEncodingExtension: true,
-        acceptEncoding: dataFeedCompression === 'gzip' ? 'gzip' : 'zstd, gzip'
-      })
-    ).downloadPath
-
-  if (cachedSlicePath === undefined) {
-    debug('getDataFeedSlice fetched from API and cached, %s', sliceKey)
-  } else {
-    debug('getDataFeedSlice already cached: %s', sliceKey)
-  }
+  const downloadResult = await download({
+    apiKey,
+    downloadPath: sliceBasePath,
+    url,
+    userAgent,
+    appendContentEncodingExtension: true,
+    acceptEncoding: dataFeedCompression === 'gzip' ? 'gzip' : 'zstd, gzip'
+  })
+  const responseSliceSize = Number(downloadResult.headers['x-slice-size'])
+  const suggestedSliceSize = Number(downloadResult.headers['x-suggested-slice-size'] ?? DEFAULT_DATA_FEED_SLICE_SIZE)
 
-  // everything went well (already cached or successfull cached) let's communicate it to parent thread
+  debug('getDataFeedSlice fetched from API and cached, %s, sliceSize: %d', sliceKey, responseSliceSize)
   const message: WorkerMessage = {
     sliceKey,
-    slicePath
+    slicePath: downloadResult.downloadPath,
+    sliceSize: responseSliceSize
   }
   parentPort!.postMessage(message)
+
+  return {
+    sliceSize: responseSliceSize,
+    suggestedSliceSize
+  }
 }
 
 export type WorkerMessage = {
   sliceKey: string
   slicePath: string
+  sliceSize: number
 }
 
 export type WorkerJobPayload = {

test/replay.test.ts

@@ -1,7 +1,11 @@
+import { existsSync, mkdtempSync, readdirSync, rmSync } from 'node:fs'
+import os from 'node:os'
+import path from 'node:path'
 import {
   clearCache,
   Exchange,
   EXCHANGES,
+  init,
   normalizeBookChanges,
   normalizeDerivativeTickers,
   normalizeTrades,
@@ -120,6 +124,60 @@ describe('replay', () => {
     1000 * 60 * 10
   )
 
+  test(
+    'replays real data using a multi-minute data feed slice',
+    async () => {
+      const cacheDir = mkdtempSync(path.join(os.tmpdir(), 'tardis-node-replay-slices-'))
+
+      try {
+        init({ cacheDir })
+
+        const messages = []
+        for await (const { message } of replay({
+          exchange: 'bitmex',
+          from: '2019-05-01T00:00:00.000Z',
+          to: '2019-05-01T00:12:00.000Z',
+          filters: [
+            {
+              channel: 'trade',
+              symbols: ['ETHUSD']
+            }
+          ],
+          skipDecoding: true
+        })) {
+          messages.push(message)
+        }
+
+        const cacheFiles = listFiles(cacheDir)
+        const multiMinuteSliceFiles = cacheFiles.filter((filePath) => /\.size-(?:[2-9]|10)\.json\.(?:gz|zst)$/.test(filePath))
+
+        expect(messages.length).toBeGreaterThan(0)
+        expect(multiMinuteSliceFiles.length).toBeGreaterThan(0)
+
+        const cacheFilesAfterFirstReplay = cacheFiles.sort()
+        for await (const _ of replay({
+          exchange: 'bitmex',
+          from: '2019-05-01T00:00:00.000Z',
+          to: '2019-05-01T00:12:00.000Z',
+          filters: [
+            {
+              channel: 'trade',
+              symbols: ['ETHUSD']
+            }
+          ],
+          skipDecoding: true
+        })) {
+        }
+
+        expect(listFiles(cacheDir).sort()).toEqual(cacheFilesAfterFirstReplay)
+      } finally {
+        init()
+        rmSync(cacheDir, { force: true, recursive: true })
+      }
+    },
+    1000 * 60 * 10
+  )
+
   test(
     'replays raw Coinbase data feed for 1st of Jun 2019 (ZEC-USDC trades)',
     async () => {
@@ -409,3 +467,14 @@ describe('replay', () => {
     1000 * 60 * 10
   )
 })
+
+function listFiles(directory: string): string[] {
+  if (existsSync(directory) === false) {
+    return []
+  }
+
+  return readdirSync(directory, { withFileTypes: true }).flatMap((entry) => {
+    const entryPath = path.join(directory, entry.name)
+    return entry.isDirectory() ? listFiles(entryPath) : [entryPath]
+  })
+}