feat: CAR-stream session for trustless-gateway block broker

.github/dictionary.txt

@@ -1,3 +1,5 @@
+backpressures
 dont
+dups
 fleek
 republisher

packages/block-brokers/.aegir.js

@@ -1,6 +1,47 @@
+import { CarWriter } from '@ipld/car'
 import cors from 'cors'
+import { CID } from 'multiformats/cid'
+import * as raw from 'multiformats/codecs/raw'
+import { sha256 } from 'multiformats/hashes/sha2'
 import polka from 'polka'
 
+/**
+ * Fixed raw blocks used by the CAR-stream session tests. The test recomputes
+ * these from the same bytes, so the CIDs match without passing them around.
+ */
+const CAR_BLOCK_BYTES = [
+  Uint8Array.from([1, 2, 3, 4]),
+  Uint8Array.from([5, 6, 7, 8]),
+  Uint8Array.from([9, 10, 11, 12])
+]
+
+async function makeRawBlock (bytes) {
+  return { cid: CID.createV1(raw.code, await sha256.digest(bytes)), bytes }
+}
+
+async function buildCar (blocks, roots) {
+  const { writer, out } = CarWriter.create(roots)
+  const chunks = []
+  const collecting = (async () => {
+    for await (const chunk of out) {
+      chunks.push(chunk)
+    }
+  })()
+  for (const block of blocks) {
+    await writer.put(block)
+  }
+  await writer.close()
+  await collecting
+  const total = chunks.reduce((n, c) => n + c.length, 0)
+  const car = new Uint8Array(total)
+  let offset = 0
+  for (const chunk of chunks) {
+    car.set(chunk, offset)
+    offset += chunk.length
+  }
+  return car
+}
+
 
 /**
  * Middleware to log requests
@@ -99,18 +140,59 @@ const options = {
       await badGateway.listen()
       const { port: badGatewayPort } = badGateway.server.address()
 
+      // Gateway that serves a CAR of a small DAG, plus the raw blocks. The CAR
+      // contains the root and the first leaf but NOT the second leaf, so the
+      // session must gap-fill that one over ?format=raw.
+      const blocks = await Promise.all(CAR_BLOCK_BYTES.map(makeRawBlock))
+      const carBytes = await buildCar([blocks[0], blocks[1]], [blocks[0].cid])
+      const blockByCid = new Map(blocks.map(b => [b.cid.toString(), b.bytes]))
+      let carRequests = 0
+
+      const carGateway = polka({ port: 0, host: '127.0.0.1' })
+      carGateway.use(cors())
+      carGateway.all('/car-requests', (req, res) => {
+        res.writeHead(200, { 'content-type': 'application/json' })
+        res.end(JSON.stringify({ carRequests }))
+      })
+      carGateway.all('/ipfs/:cid', (req, res) => {
+        // no-store so the browser HTTP cache does not satisfy getCar's
+        // `cache: 'force-cache'` across tests, which would hide CAR requests
+        // from the counter below.
+        const format = new URL(req.url, 'http://localhost').searchParams.get('format')
+        if (format === 'car') {
+          carRequests++
+          res.writeHead(200, { 'content-type': 'application/vnd.ipld.car', 'cache-control': 'no-store' })
+          res.end(carBytes)
+          return
+        }
+        const bytes = blockByCid.get(req.params.cid)
+        if (bytes == null) {
+          res.writeHead(404)
+          res.end()
+          return
+        }
+        res.writeHead(200, { 'content-type': 'application/vnd.ipld.raw', 'content-length': bytes.length, 'cache-control': 'no-store' })
+        res.end(bytes)
+      })
+
+      await carGateway.listen()
+      const { port: carGatewayPort } = carGateway.server.address()
+
       return {
         goodGateway,
         badGateway,
+        carGateway,
         env: {
           TRUSTLESS_GATEWAY: `http://127.0.0.1:${goodGatewayPort}`,
-          BAD_TRUSTLESS_GATEWAY: `http://127.0.0.1:${badGatewayPort}`
+          BAD_TRUSTLESS_GATEWAY: `http://127.0.0.1:${badGatewayPort}`,
+          CAR_GATEWAY: `http://127.0.0.1:${carGatewayPort}`
         }
       }
     },
     async after (options, before) {
       await before.goodGateway.server.close()
       await before.badGateway.server.close()
+      await before.carGateway.server.close()
     }
   }
 }

packages/block-brokers/package.json

@@ -51,6 +51,7 @@
     "@helia/bitswap": "^3.2.3",
     "@helia/interface": "^6.2.1",
     "@helia/utils": "^2.5.2",
+    "@ipld/car": "^5.4.3",
     "@libp2p/interface": "^3.2.0",
     "@libp2p/peer-id": "^6.0.6",
     "@libp2p/utils": "^7.0.15",

packages/block-brokers/src/trustless-gateway/broker.ts

@@ -1,3 +1,4 @@
+import { createTrustlessGatewayCarSession } from './car-session.ts'
 import { DEFAULT_ALLOW_INSECURE, DEFAULT_ALLOW_LOCAL } from './index.ts'
 import { createTrustlessGatewaySession } from './session.ts'
 import { findHttpGatewayProviders } from './utils.ts'
@@ -45,6 +46,7 @@ export class TrustlessGatewayBlockBroker implements BlockBroker<TrustlessGateway
   private readonly routing: Routing
   private readonly log: Logger
   private readonly logger: ComponentLogger
+  private readonly carStreamSessions: boolean
 
   constructor (components: TrustlessGatewayBlockBrokerComponents, init: TrustlessGatewayBlockBrokerInit = {}) {
     this.log = components.logger.forComponent('helia:trustless-gateway-block-broker')
@@ -53,6 +55,7 @@ export class TrustlessGatewayBlockBroker implements BlockBroker<TrustlessGateway
     this.allowInsecure = init.allowInsecure ?? DEFAULT_ALLOW_INSECURE
     this.allowLocal = init.allowLocal ?? DEFAULT_ALLOW_LOCAL
     this.transformRequestInit = init.transformRequestInit
+    this.carStreamSessions = init.carStreamSessions ?? false
   }
 
   async retrieve (cid: CID, options: BlockRetrievalOptions<TrustlessGatewayGetBlockProgressEvents> = {}): Promise<Uint8Array> {
@@ -99,7 +102,8 @@ export class TrustlessGatewayBlockBroker implements BlockBroker<TrustlessGateway
   }
 
   createSession (options: CreateTrustlessGatewaySessionOptions = {}): SessionBlockBroker<TrustlessGatewayGetBlockProgressEvents> {
-    return createTrustlessGatewaySession({
+    const create = this.carStreamSessions ? createTrustlessGatewayCarSession : createTrustlessGatewaySession
+    return create({
       logger: this.logger,
       routing: this.routing
     }, {