feat(download): serve HTTP Range on the proxy route

lib/storage.ts

@@ -65,6 +65,35 @@ export class ObjectNotFoundError extends Error {
   }
 }
 
+/** `size` is omitted when the backend didn't report one, so the route can skip `content-range`. */
+export class RangeNotSatisfiableError extends Error {
+  constructor(
+    objectName: string,
+    public readonly size?: number,
+  ) {
+    super(`Range not satisfiable for ${objectName}${size === undefined ? '' : ` (size ${size})`}`)
+    this.name = 'RangeNotSatisfiableError'
+  }
+}
+
+/** An omitted `end` means open-ended; the backend resolves it against the object length. */
+export interface RangeRequest {
+  start: number
+  end?: number
+}
+
+export interface ByteRange {
+  start: number
+  end: number
+}
+
+/** `range` is set only when a range was requested and the adapter honoured it. */
+export interface DownloadStream {
+  stream: Readable
+  size?: number
+  range?: ByteRange
+}
+
 export class Storage {
   static async fromEnv() {
     const storage = new Storage({
@@ -133,11 +162,17 @@ export class Storage {
     if (actualPartCount < location.partCount) throw new ObjectNotFoundError(partsFolder)
   }
 
-  private async downloadFromCacheEntryLocation(location: StorageLocation) {
-    if (location.mergedAt) return this.adapter.createDownloadStream(`${location.folderName}/merged`)
+  private async downloadFromCacheEntryLocation(
+    location: StorageLocation,
+    range?: RangeRequest,
+  ): Promise<DownloadStream> {
+    if (location.mergedAt)
+      return this.adapter.createDownloadStream(`${location.folderName}/merged`, range)
 
     await this.ensurePartsExist(location)
-    return Readable.from(this.streamParts(location))
+    // No `range`: an unmerged entry is concatenated from its parts as it is read, so
+    // there is nothing to seek into. Served whole under a 200, which Range allows.
+    return { stream: Readable.from(this.streamParts(location)) }
   }
 
   private async pumpPartsToStreams(
@@ -165,7 +200,7 @@ export class Storage {
     if (location.partsDeletedAt) throw new Error('No parts to feed for location with deleted parts')
 
     for (let i = 0; i < location.partCount; i++) {
-      const partStream = await this.adapter.createDownloadStream(
+      const { stream: partStream } = await this.adapter.createDownloadStream(
         `${location.folderName}/parts/${i}`,
       )
 
@@ -524,7 +559,7 @@ export class Storage {
     }
   }
 
-  async download(cacheEntryId: string): Promise<Readable | undefined> {
+  async download(cacheEntryId: string, range?: RangeRequest): Promise<DownloadStream | undefined> {
     const protectedLocation = await this.db.transaction().execute(async (tx) => {
       let query = tx
         .selectFrom('storage_locations')
@@ -551,8 +586,8 @@ export class Storage {
 
     try {
       if (storageLocation.mergedAt) {
-        const stream = await this.downloadFromCacheEntryLocation(storageLocation)
-        return this.protectDownloadStream(stream, readerLeaseId)
+        const download = await this.downloadFromCacheEntryLocation(storageLocation, range)
+        return { ...download, stream: this.protectDownloadStream(download.stream, readerLeaseId) }
       }
 
       await this.ensurePartsExist(storageLocation)
@@ -568,8 +603,8 @@ export class Storage {
           }),
       )
       if (!merge) {
-        const stream = await this.downloadFromCacheEntryLocation(storageLocation)
-        return this.protectDownloadStream(stream, readerLeaseId)
+        const download = await this.downloadFromCacheEntryLocation(storageLocation, range)
+        return { ...download, stream: this.protectDownloadStream(download.stream, readerLeaseId) }
       }
 
       this.pumpPartsToStreams(storageLocation, responseStream, mergerStream).catch((err) => {
@@ -579,7 +614,7 @@ export class Storage {
           logger.warn(`Stale cache entry ${cacheEntryId}: ${err.message}`)
       })
 
-      return this.protectDownloadStream(responseStream, readerLeaseId)
+      return { stream: this.protectDownloadStream(responseStream, readerLeaseId) }
     } catch (err) {
       await releaseReaderLease(this.db, readerLeaseId)
       if (err instanceof ObjectNotFoundError) {
@@ -789,7 +824,7 @@ export class Storage {
 export const getStorage = createSingletonPromise(async () => Storage.fromEnv())
 
 export interface StorageAdapter {
-  createDownloadStream(objectName: string): Promise<Readable>
+  createDownloadStream(objectName: string, range?: RangeRequest): Promise<DownloadStream>
   /**
    * Uploads must be atomically visible: an object never exists partially, and
    * overwriting an object never disturbs active readers of the previous
@@ -995,19 +1030,30 @@ class S3Adapter implements StorageAdapter {
     return deleted
   }
 
-  async createDownloadStream(objectName: string) {
+  async createDownloadStream(objectName: string, range?: RangeRequest): Promise<DownloadStream> {
     try {
       const response = await this.s3.send(
         new GetObjectCommand({
           Bucket: this.bucket,
           Key: `${this.keyPrefix}/${objectName}`,
+          Range: range ? `bytes=${range.start}-${range.end ?? ''}` : undefined,
         }),
       )
       if (!response.Body) throw new Error('No body in S3 get object response')
 
-      return response.Body as Readable
+      const stream = response.Body as Readable
+      if (response.$metadata.httpStatusCode !== 206) return { stream, size: response.ContentLength }
+
+      const served = parseContentRange(response.ContentRange)
+      if (!served) {
+        stream.destroy()
+        throw new Error(`S3 answered 206 with unparseable Content-Range: ${response.ContentRange}`)
+      }
+      return { stream, size: served.size, range: { start: served.start, end: served.end } }
     } catch (err: any) {
       if (err.name === 'NoSuchKey') throw new ObjectNotFoundError(objectName)
+      if (err.name === 'InvalidRange')
+        throw new RangeNotSatisfiableError(objectName, parseActualObjectSize(err))
       throw err
     }
   }
@@ -1160,14 +1206,20 @@ class FileSystemAdapter implements StorageAdapter {
     return folder
   }
 
-  async createDownloadStream(objectName: string) {
+  async createDownloadStream(objectName: string, range?: RangeRequest): Promise<DownloadStream> {
     const filePath = this.safePath(objectName)
+    let size: number
     try {
-      await fs.access(filePath)
+      const stat = await fs.stat(filePath)
+      size = stat.size
     } catch {
       throw new ObjectNotFoundError(objectName)
     }
-    return createReadStream(filePath)
+    if (!range) return { stream: createReadStream(filePath), size }
+
+    const served = clampRange(range, size)
+    if (!served) throw new RangeNotSatisfiableError(objectName, size)
+    return { stream: createReadStream(filePath, served), size, range: served }
   }
 
   async objectExists(objectName: string) {
@@ -1317,11 +1369,22 @@ class GcsAdapter implements StorageAdapter {
     this.bucket = gcs.bucket(bucket)
   }
 
-  async createDownloadStream(objectName: string) {
+  async createDownloadStream(objectName: string, range?: RangeRequest): Promise<DownloadStream> {
     const file = this.bucket.file(`${this.keyPrefix}/${objectName}`)
-    const [exists] = await file.exists()
-    if (!exists) throw new ObjectNotFoundError(objectName)
-    return file.createReadStream()
+    // `getMetadata` proves existence and carries the size, so no second round-trip.
+    let size: number
+    try {
+      const [metadata] = await file.getMetadata()
+      size = Number(metadata.size)
+    } catch (err: any) {
+      if (err.code === 404) throw new ObjectNotFoundError(objectName)
+      throw err
+    }
+    if (!range) return { stream: file.createReadStream(), size }
+
+    const served = clampRange(range, size)
+    if (!served) throw new RangeNotSatisfiableError(objectName, size)
+    return { stream: file.createReadStream(served), size, range: served }
   }
 
   async objectExists(objectName: string) {
@@ -1405,3 +1468,24 @@ class GcsAdapter implements StorageAdapter {
       .then((res) => res[0])
   }
 }
+
+function clampRange(range: RangeRequest, size: number): ByteRange | undefined {
+  if (range.start >= size) return
+  return { start: range.start, end: Math.min(range.end ?? size - 1, size - 1) }
+}
+
+const CONTENT_RANGE_RE = /^bytes (\d+)-(\d+)\/(\d+)$/
+
+function parseContentRange(header: string | undefined) {
+  if (!header) return
+  const m = CONTENT_RANGE_RE.exec(header)
+  if (!m) return
+  return { start: Number(m[1]), end: Number(m[2]), size: Number(m[3]) }
+}
+
+// AWS and MinIO put the object size here; other S3-compatible servers may not.
+function parseActualObjectSize(err: unknown): number | undefined {
+  const raw = (err as { ActualObjectSize?: unknown }).ActualObjectSize
+  const size = Number(raw)
+  return Number.isSafeInteger(size) && size >= 0 ? size : undefined
+}

routes/download/[cacheEntryId].ts

@@ -1,12 +1,28 @@
-import { Readable } from 'node:stream'
+import type { RangeRequest } from '~/lib/storage'
+import { pipeline } from 'node:stream/promises'
 import { z } from 'zod'
 import { logger } from '~/lib/logger'
-import { getStorage } from '~/lib/storage'
+import { getStorage, RangeNotSatisfiableError } from '~/lib/storage'
 
 const pathParamsSchema = z.object({
   cacheEntryId: z.string(),
 })
 
+// Single ranges only; multi-range and suffix (`bytes=-500`) fall through to a 200.
+const RANGE_RE = /^bytes=(\d+)-(\d*)$/i
+
+function parseRange(header: string | undefined): RangeRequest | undefined {
+  if (!header) return
+  const m = RANGE_RE.exec(header.trim())
+  if (!m) return
+  const start = Number(m[1])
+  if (!Number.isSafeInteger(start)) return
+  if (m[2] === '') return { start }
+  const end = Number(m[2])
+  if (!Number.isSafeInteger(end) || end < start) return
+  return { start, end }
+}
+
 export default defineEventHandler(async (event) => {
   const parsedPathParams = pathParamsSchema.safeParse(event.context.params)
   if (!parsedPathParams.success)
@@ -16,28 +32,64 @@ export default defineEventHandler(async (event) => {
     })
 
   const { cacheEntryId } = parsedPathParams.data
-
   const storage = await getStorage()
-  const stream = await storage.download(cacheEntryId)
-  if (!stream)
+
+  const range = parseRange(getHeader(event, 'range'))
+
+  // Unmerged entries ignore Range, so clients must key off the status, not this header.
+  setHeader(event, 'accept-ranges', 'bytes')
+
+  let download
+  try {
+    download = await storage.download(cacheEntryId, range)
+  } catch (err) {
+    if (err instanceof RangeNotSatisfiableError) {
+      // An empty object has no satisfiable range, but S3 answers `bytes=0-` with an
+      // empty 200 while the other adapters raise. Normalise to the 200 so the
+      // response does not depend on which backend is configured.
+      if (err.size === 0) {
+        setHeader(event, 'content-length', 0)
+        return send(event)
+      }
+      setResponseStatus(event, 416, 'Range Not Satisfiable')
+      if (err.size !== undefined) setHeader(event, 'content-range', `bytes */${err.size}`)
+      return send(event)
+    }
+    throw err
+  }
+  if (!download)
     throw createError({
       statusCode: 404,
       message: 'Cache file not found',
     })
 
+  if (range && download.range && download.size !== undefined) {
+    setResponseStatus(event, 206)
+    setHeader(
+      event,
+      'content-range',
+      `bytes ${download.range.start}-${download.range.end}/${download.size}`,
+    )
+    setHeader(event, 'content-length', download.range.end - download.range.start + 1)
+  } else if (download.size !== undefined) {
+    setHeader(event, 'content-length', download.size)
+  }
+
+  // Take over the response from h3: `sendStream` applies no backpressure and does not
+  // notice client aborts, whereas `pipeline` destroys the source and releases its lease.
+  // `_handled` is an h3 v1 internal and will need replacing when h3 v2 lands.
+  event._handled = true
   try {
-    await sendStream(event, Readable.toWeb(stream) as ReadableStream)
+    await pipeline(download.stream, event.node.res)
   } catch (err) {
-    // Once the response has started flushing, we can't surface stream errors
-    // as an HTTP error — Nitro's default error handler would call
-    // `setResponseHeaders` after headers were already sent and crash with
-    // ERR_HTTP_HEADERS_SENT (logged as an unhandled error). Client aborts on
-    // long downloads are expected (cancelled jobs, parallel runners), so we
-    // log and swallow once headers are out.
+    // Client went away mid-body. Expected on cancelled jobs and parallel runners.
+    if ((err as NodeJS.ErrnoException).code === 'ERR_STREAM_PREMATURE_CLOSE') {
+      logger.debug(`Client aborted /download/${cacheEntryId}: ${(err as Error).message}`)
+      return
+    }
+    // Headers are out, so Nitro's error handler would crash with ERR_HTTP_HEADERS_SENT.
     if (event.node.res.headersSent) {
-      if (event.node.req.destroyed)
-        logger.debug(`Client aborted /download/${cacheEntryId}: ${(err as Error).message}`)
-      else logger.error(`Download stream failed for ${cacheEntryId}`, { error: err })
+      logger.error(`Download stream failed for ${cacheEntryId}`, { error: err })
       return
     }
     throw err

tests/cleanup-lifecycle.test.ts

@@ -249,15 +249,15 @@ describe('cleanup lifecycle', () => {
       const activePartsDownload = await storage.download(entryId)
       expect(mergingDownload).toBeDefined()
       expect(activePartsDownload).toBeDefined()
-      for await (const _chunk of mergingDownload!) void _chunk
+      for await (const _chunk of mergingDownload!.stream) void _chunk
       await storage.waitForOngoingMerges()
 
       const taskModule = await import('~/tasks/cleanup/parts')
       const task = taskModule.default
       await task.run({} as never)
       expect(await storage.adapter.countFilesInFolder(`${folderName}/parts`)).toBe(1)
 
-      for await (const _chunk of activePartsDownload!) void _chunk
+      for await (const _chunk of activePartsDownload!.stream) void _chunk
 
       await vi.waitFor(
         async () => {
@@ -270,7 +270,7 @@ describe('cleanup lifecycle', () => {
       const mergedDownload = await storage.download(entryId)
       expect(mergedDownload).toBeDefined()
       let restored = ''
-      for await (const chunk of mergedDownload!) restored += chunk.toString()
+      for await (const chunk of mergedDownload!.stream) restored += chunk.toString()
       expect(restored).toBe('cache-data')
     } finally {
       await db.deleteFrom('storage_locations').where('id', '=', locationId).execute()
@@ -320,7 +320,7 @@ describe('cleanup lifecycle', () => {
     await task.run({} as never)
     expect(await storage.adapter.countFilesInFolder(folderName)).toBe(1)
 
-    for await (const _chunk of download!) void _chunk
+    for await (const _chunk of download!.stream) void _chunk
 
     await vi.waitFor(
       async () => {
@@ -446,7 +446,7 @@ describe('cleanup lifecycle', () => {
     try {
       const download = await storage.download(entryId)
       expect(download).toBeDefined()
-      download!.on('error', () => undefined)
+      download!.stream.on('error', () => undefined)
       await db
         .deleteFrom('storage_reader_leases')
         .where('storageLocationId', '=', locationId)
@@ -456,9 +456,10 @@ describe('cleanup lifecycle', () => {
       // The renewal fires on the fake timer, but the lease-lost DB query resolves on a
       // real round-trip — wait for the resulting destroy instead of asserting synchronously.
       // Plain 'close' wait (not events.once, which rejects on the error-destroy).
-      if (!download!.destroyed) await new Promise((resolve) => download!.once('close', resolve))
+      if (!download!.stream.destroyed)
+        await new Promise((resolve) => download!.stream.once('close', resolve))
 
-      expect(download!.destroyed).toBe(true)
+      expect(download!.stream.destroyed).toBe(true)
     } finally {
       vi.useRealTimers()
       await db.deleteFrom('storage_locations').where('id', '=', locationId).execute()