Merge pull request #136 from autonomys/byte-range-downloads

feat: implement byte range downloads

Author: jfrank-summit

services/file-retriever/package.json

@@ -4,11 +4,11 @@
   "type": "module",
   "dependencies": {
     "@auto-files/rpc-apis": "workspace:*",
-    "@autonomys/asynchronous": "^1.4.36",
-    "@autonomys/auto-dag-data": "^1.4.31",
-    "@autonomys/auto-utils": "^1.4.31",
-    "@autonomys/file-caching": "^1.4.36",
-    "@autonomys/rpc": "^1.4.31",
+    "@autonomys/asynchronous": "^1.5.12",
+    "@autonomys/auto-dag-data": "^1.5.12",
+    "@autonomys/auto-utils": "^1.5.12",
+    "@autonomys/file-caching": "^1.5.12",
+    "@autonomys/rpc": "^1.5.12",
     "@ipld/dag-pb": "^4.1.3",
     "@keyvhq/core": "^2.1.1",
     "@keyvhq/sqlite": "^2.1.6",

services/file-retriever/src/http/controllers/file.ts

@@ -4,7 +4,7 @@ import { fileComposer } from '../../services/fileComposer.js'
 import { pipeline } from 'stream'
 import { logger } from '../../drivers/logger.js'
 import { asyncSafeHandler } from '../../utils/express.js'
-import { uniqueHeaderValue } from '../../utils/http.js'
+import { getByteRange, uniqueHeaderValue } from '../../utils/http.js'
 import { HttpError } from '../middlewares/error.js'
 import { dsnFetcher } from '../../services/dsnFetcher.js'
 import { isValidCID } from '../../utils/dagData.js'
@@ -56,12 +56,25 @@ fileRouter.get(
     }
 
     const rawMode = req.query.raw === 'true'
+    const byteRange = getByteRange(req)
     const ignoreCache =
       req.query.originControl === 'no-cache' ||
       uniqueHeaderValue(req.headers['x-origin-control'])?.toLowerCase() ===
         'no-cache'
 
-    const [fromCache, file] = await fileComposer.get(cid, ignoreCache)
+    const metadata = await dsnFetcher.fetchNodeMetadata(cid)
+    if (byteRange) {
+      if (byteRange[0] > Number(metadata.size)) {
+        res.set('Content-Range', `bytes */${metadata.size}`)
+        res.sendStatus(416)
+        return
+      }
+    }
+
+    const [fromCache, file] = await fileComposer.get(cid, {
+      ignoreCache,
+      byteRange,
+    })
     if (fromCache) {
       res.setHeader('x-file-origin', 'cache')
     } else {
@@ -77,10 +90,30 @@ fileRouter.get(
         `filename="${encodeURIComponent(file.filename)}"`,
       )
     }
-    if (file.size) {
+
+    // Advertise range support
+    res.set('Accept-Ranges', 'bytes')
+
+    if (byteRange && file.size != null) {
+      const fileSizeNumber = Number(file.size)
+      const startIndex = byteRange[0]
+
+      const effectiveEnd = Math.min(
+        byteRange[1] != null ? byteRange[1] : fileSizeNumber - 1,
+        fileSizeNumber - 1,
+      )
+      const contentLength = effectiveEnd - startIndex + 1
+
+      res.status(206)
+      res.set(
+        'Content-Range',
+        `bytes ${startIndex}-${effectiveEnd}/${fileSizeNumber}`,
+      )
+      res.set('Content-Length', contentLength.toString())
+    } else if (file.size != null) {
       res.set('Content-Length', file.size.toString())
     }
-    if (file.encoding && !rawMode) {
+    if (file.encoding && !rawMode && !byteRange) {
       res.set('Content-Encoding', file.encoding)
     }
 

services/file-retriever/src/repositories/dag-indexer.ts

@@ -27,6 +27,7 @@ const mapToDomain = (db: ExtendedIPLDMetadataDB): ExtendedIPLDMetadata => {
     type: db.type as MetadataType,
     linkDepth: db.link_depth,
     name: db.name,
+    size: BigInt(db.size),
     blockHeight: db.block_height,
     blockHash: db.block_hash,
     extrinsicId: db.extrinsic_id,

services/file-retriever/src/services/cache.ts

@@ -2,6 +2,7 @@ import {
   createFileCache,
   defaultMemoryAndSqliteConfig,
   ensureDirectoryExists,
+  FileCacheOptions,
   FileResponse,
 } from '@autonomys/file-caching'
 import path from 'path'
@@ -17,7 +18,8 @@ export const cache = createFileCache(
 
 const cacheWithNamespace = (namespace: string) => {
   return {
-    get: (cid: string) => cache.get(`${namespace}:${cid}`),
+    get: (cid: string, options?: FileCacheOptions) =>
+      cache.get(`${namespace}:${cid}`, options),
     set: (cid: string, fileResponse: FileResponse) =>
       cache.set(`${namespace}:${cid}`, fileResponse),
     has: (cid: string) => cache.has(`${namespace}:${cid}`),

services/file-retriever/src/services/dsnFetcher.ts

@@ -9,7 +9,11 @@ import {
   IPLDNodeData,
   MetadataType,
 } from '@autonomys/auto-dag-data'
-import { FileResponse } from '@autonomys/file-caching'
+import {
+  ByteRange,
+  FileCacheOptions,
+  FileResponse,
+} from '@autonomys/file-caching'
 import { z } from 'zod'
 import { PBNode } from '@ipld/dag-pb'
 import { HttpError } from '../http/middlewares/error.js'
@@ -29,6 +33,7 @@ import { Readable } from 'stream'
 import { ReadableStream } from 'stream/web'
 import { fileCache, nodeCache } from './cache.js'
 import { dagIndexerRepository } from '../repositories/dag-indexer.js'
+import { sliceReadable } from '../utils/readable.js'
 
 const fetchNodesSchema = z.object({
   jsonrpc: z.string(),
@@ -141,6 +146,123 @@ const fetchObjects = async (objects: ObjectMapping[]) => {
   )
 }
 
+const getNodesForPartialRetrieval = async (
+  chunks: ExtendedIPLDMetadata[],
+  byteRange: ByteRange,
+): Promise<{
+  nodes: string[]
+  firstNodeFileOffset: number
+}> => {
+  let accumulatedLength = 0
+  const nodeRange: [number | null, number | null] = [null, null]
+  let firstNodeFileOffset: number | undefined
+  let i = 0
+
+  logger.debug(
+    `getNodesForPartialRetrieval called (byteRange=[${byteRange[0]}, ${byteRange[1] ?? 'EOF'}])`,
+  )
+
+  // Searches for the first node that contains the byte range
+  while (nodeRange[0] === null && i < chunks.length) {
+    const chunk = chunks[i]
+    const chunkSize = Number((chunk.size ?? 0).valueOf())
+    // [accumulatedLength, accumulatedLength + chunkSize) // is the range of the chunk
+    if (
+      byteRange[0] >= accumulatedLength &&
+      byteRange[0] < accumulatedLength + chunkSize
+    ) {
+      nodeRange[0] = i
+      firstNodeFileOffset = accumulatedLength
+    } else {
+      accumulatedLength += chunkSize
+      i++
+    }
+  }
+
+  // Searchs for the last node that contains the byte range
+  // unless the byte range is the last byte of the file
+  if (byteRange[1]) {
+    while (nodeRange[1] === null && i < chunks.length) {
+      const chunk = chunks[i]
+      const chunkSize = Number((chunk.size ?? 0).valueOf())
+      if (
+        byteRange[1] >= accumulatedLength &&
+        byteRange[1] < accumulatedLength + chunkSize
+      ) {
+        nodeRange[1] = i
+      }
+      accumulatedLength += chunkSize
+      i++
+    }
+  }
+
+  if (nodeRange[0] == null) {
+    throw new Error('Byte range not found')
+  }
+
+  const nodes = chunks
+    .slice(nodeRange[0], nodeRange[1] === null ? undefined : nodeRange[1] + 1)
+    .map((e) => e.cid)
+
+  return {
+    nodes,
+    firstNodeFileOffset: firstNodeFileOffset ?? 0,
+  }
+}
+
+const fetchFileAsStreamWithByteRange = async (
+  cid: string,
+  byteRange: ByteRange,
+): Promise<Readable> => {
+  const chunks = await dsnFetcher.getFileChunks(cid)
+  const { nodes, firstNodeFileOffset } = await getNodesForPartialRetrieval(
+    chunks,
+    byteRange,
+  )
+
+  logger.debug(
+    `getNodesForPartialRetrieval called (byteRange=[${byteRange[0]}, ${byteRange[1] ?? 'EOF'}]) nodes=${JSON.stringify(nodes)} firstNodeFileOffset=${firstNodeFileOffset}`,
+  )
+
+  // We pass all the chunks to the fetchNode function
+  // So that we can fetch all the nodes within the same piece
+  // in one go
+  const siblings = chunks.map((e) => e.cid)
+  const stream = new ReadableStream({
+    start: async (controller) => {
+      try {
+        for (const chunk of nodes) {
+          const node = await dsnFetcher.fetchNode(chunk, siblings)
+          const data = safeIPLDDecode(node)
+          if (!data) {
+            throw new HttpError(
+              400,
+              'Bad request: Not a valid auto-dag-data IPLD node',
+            )
+          }
+
+          controller.enqueue(Buffer.from(data.data ?? []))
+        }
+
+        controller.close()
+      } catch (error) {
+        controller.error(error)
+      }
+    },
+  })
+
+  const metadata = await dsnFetcher.fetchNodeMetadata(cid)
+  const fileSize = Number(metadata.size)
+  const endIndex = byteRange[1] ?? fileSize - 1
+  const length = endIndex - byteRange[0] + 1
+
+  return sliceReadable(
+    Readable.fromWeb(stream),
+    byteRange[0] - firstNodeFileOffset,
+    length,
+  )
+}
+
 /**
  * Fetches a file as a stream
  *
@@ -150,13 +272,13 @@ const fetchObjects = async (objects: ObjectMapping[]) => {
  * @param node - The root node of the file
  * @returns A readable stream of the file
  */
-const fetchFileAsStream = async (cid: string): Promise<ReadableStream> => {
+const fetchFileAsStream = async (cid: string): Promise<Readable> => {
   const chunks = await dsnFetcher.getFileChunks(cid)
 
   // if a file is a multi-node file, we need to fetch the nodes in the correct order
   // bearing in mind there might be multiple levels of links, we need to fetch
   // all the links from the root node first and then continue with the next level
-  return new ReadableStream({
+  const stream = new ReadableStream({
     start: async (controller) => {
       try {
         for (const chunk of chunks) {
@@ -181,6 +303,8 @@ const fetchFileAsStream = async (cid: string): Promise<ReadableStream> => {
       }
     },
   })
+
+  return Readable.fromWeb(stream)
 }
 
 const getFileMetadata = (
@@ -202,7 +326,10 @@ const getFileMetadata = (
   }
 }
 
-const fetchFile = async (cid: string): Promise<FileResponse> => {
+const fetchFile = async (
+  cid: string,
+  options?: FileCacheOptions,
+): Promise<FileResponse> => {
   try {
     const metadata = await dsnFetcher.fetchNodeMetadata(cid)
     if (metadata.type !== MetadataType.File) {
@@ -215,8 +342,12 @@ const fetchFile = async (cid: string): Promise<FileResponse> => {
       `Fetching file (cid=${cid}, size=${traits.size}, mimeType=${traits.mimeType}, filename=${traits.filename}, encoding=${traits.encoding})`,
     )
 
+    const readable = options?.byteRange
+      ? await fetchFileAsStreamWithByteRange(cid, options.byteRange)
+      : await fetchFileAsStream(cid)
+
     return {
-      data: Readable.fromWeb(await fetchFileAsStream(cid)),
+      data: readable,
       ...traits,
     }
   } catch (error) {

services/file-retriever/src/services/fileComposer.ts

@@ -1,23 +1,28 @@
 import { dsnFetcher } from './dsnFetcher.js'
 import { forkStream } from '@autonomys/asynchronous'
 import { logger } from '../drivers/logger.js'
-import { FileResponse } from '@autonomys/file-caching'
+import { FileCacheOptions, FileResponse } from '@autonomys/file-caching'
 import { fileCache } from './cache.js'
 import { moderationService } from './moderation.js'
 import { HttpError } from '../http/middlewares/error.js'
 
+interface DownloadOptions extends FileCacheOptions {
+  ignoreCache?: boolean
+}
+
 const get = async (
   cid: string,
-  ignoreCache = false,
+  options: DownloadOptions = {},
 ): Promise<[fromCache: boolean, FileResponse]> => {
+  const isPartialRequest = options.byteRange != null
   const isBanned = await moderationService.isFileBanned(cid)
   if (isBanned) {
     logger.warn(`File download blocked: ${cid}`)
     throw new HttpError(451, 'Unavailable for legal reasons')
   }
 
-  if (!ignoreCache) {
-    const cachedFile = await fileCache.get(cid)
+  if (!options.ignoreCache) {
+    const cachedFile = await fileCache.get(cid, options)
     if (cachedFile) {
       logger.debug(`Cache hit for file ${cid}`)
       return [true, cachedFile]
@@ -26,10 +31,15 @@ const get = async (
 
   let start = performance.now()
   logger.debug(`Fetching file from DSN ${cid}`)
-  const file = await dsnFetcher.fetchFile(cid)
+  const file = await dsnFetcher.fetchFile(cid, options)
   let end = performance.now()
   logger.debug(`Fetching file from DSN ${cid} took ${end - start}ms`)
 
+  // Do not cache partial responses
+  if (isPartialRequest) {
+    return [false, file]
+  }
+
   start = performance.now()
   const [data, cachingStream] = await forkStream(file.data)
   end = performance.now()
@@ -44,13 +54,7 @@ const get = async (
       console.error('Error caching file', e)
     })
 
-  return [
-    false,
-    {
-      ...file,
-      data,
-    },
-  ]
+  return [false, { ...file, data }]
 }
 
 export const fileComposer = {