feat: use rollup indexes

package.json

@@ -6,7 +6,7 @@
   "type": "module",
   "scripts": {
     "start": "npm run dev",
-    "dev": "npm run build:debug && miniflare dist/worker.mjs --watch --debug -m --r2-persist",
+    "dev": "npm run build:debug && miniflare dist/worker.mjs --watch --debug -m --r2-persist --global-async-io --global-timers",
     "build": "esbuild --bundle src/index.js --format=esm --sourcemap --minify --outfile=dist/worker.mjs",
     "build:debug": "esbuild --bundle src/index.js --format=esm --outfile=dist/worker.mjs",
     "test": "npm run build:debug && node --test --experimental-vm-modules",
@@ -27,8 +27,8 @@
     "@ipld/dag-cbor": "^9.0.0",
     "@ipld/dag-json": "^10.0.1",
     "@ipld/dag-pb": "^4.0.2",
-    "@web3-storage/gateway-lib": "^3.2.0",
-    "cardex": "^2.1.0",
+    "@web3-storage/gateway-lib": "^3.2.2",
+    "cardex": "^2.2.0",
     "chardet": "^1.5.0",
     "dagula": "^7.0.0",
     "magic-bytes.js": "^1.0.12",
@@ -46,5 +46,10 @@
     "miniflare": "^2.9.0",
     "standard": "^17.0.0",
     "uint8arrays": "^4.0.3"
+  },
+  "standard": {
+    "ignore": [
+      "*.ts"
+    ]
   }
 }

scripts/r2-put.js

@@ -2,7 +2,7 @@
  * Puts data to the local persisted Miniflare R2 buckets.
  *
  * Usage:
- *   node scripts/r2-put.js test.jpg --no-wrap
+ *   node scripts/r2-put.js test.jpg --no-wrap --no-rollup
  */
 import { R2Bucket } from '@miniflare/r2'
 import { FileStorage } from '@miniflare/storage-file'
@@ -20,5 +20,8 @@ const wrapWithDirectory = process.argv.every(p => p !== '--no-wrap')
 
 const { dataCid, carCids } = await builder.add(input, { wrapWithDirectory })
 
+const rollup = process.argv.every(p => p !== '--no-rollup')
+if (rollup) await builder.rollup(dataCid, carCids)
+
 console.log(`Data CID:\n${dataCid}`)
 console.log(`CAR CIDs:\n${carCids.join('\n')}`)

src/bindings.d.ts

@@ -1,5 +1,6 @@
-import type { CID } from 'multiformats/cid'
+import type { Link } from 'multiformats/link'
 import type { Context } from '@web3-storage/gateway-lib'
+import type { CARLink } from 'cardex/api'
 import type { MemoryBudget } from './lib/mem-budget'
 
 export {}
@@ -12,8 +13,21 @@ export interface Environment {
   MAX_SHARDS: string
 }
 
-export interface CarCidsContext extends Context {
-  carCids: CID[]
+export interface IndexSource {
+  /** Bucket this index can be found in */
+  bucket: R2Bucket
+  /** Bucket key for the source */
+  key: string
+  /**
+   * Origin CAR CID the index source applies to. Will be undefined if the index
+   * source is a multi-index index, which specifies origin CAR CIDs within the
+   * index.
+   */
+  origin?: CARLink
+}
+
+export interface IndexSourcesContext extends Context {
+  indexSources: IndexSource[]
 }
 
 export interface R2GetOptions {

src/index.js

@@ -16,15 +16,14 @@ import {
 } from '@web3-storage/gateway-lib/handlers'
 import {
   withDagula,
-  withCarCids,
+  withIndexSources,
   withUnsupportedFeaturesHandler,
   withVersionHeader
 } from './middleware.js'
 
 /**
  * @typedef {import('./bindings').Environment} Environment
  * @typedef {import('@web3-storage/gateway-lib').IpfsUrlContext} IpfsUrlContext
- * @typedef {import('./bindings').CarCidsContext} CarCidsContext
  * @typedef {import('@web3-storage/gateway-lib').DagulaContext} DagulaContext
  */
 
@@ -41,15 +40,15 @@ export default {
       withUnsupportedFeaturesHandler,
       withHttpGet,
       withParsedIpfsUrl,
-      withCarCids,
+      withIndexSources,
       withDagula,
       withFixedLengthStream
     )
     return middleware(handler)(request, env, ctx)
   }
 }
 
-/** @type {import('@web3-storage/gateway-lib').Handler<DagulaContext & CarCidsContext & IpfsUrlContext, Environment>} */
+/** @type {import('@web3-storage/gateway-lib').Handler<DagulaContext & IpfsUrlContext, Environment>} */
 async function handler (request, env, ctx) {
   const { headers } = request
   const { searchParams } = ctx

src/lib/block-batch.js

@@ -2,9 +2,9 @@ const MAX_BYTES_BETWEEN = 1024 * 1024 * 2
 const MAX_BATCH_SIZE = 10
 
 /**
- * @typedef {import('multiformats').CID} CID
- * @typedef {{ carCid: CID, blockCid: CID, offset: number }} BatchItem
- * @typedef {{ add: (i: BatchItem) => void, next: () => BatchItem[] }} BlockBatcher
+ * @typedef {import('multiformats').UnknownLink} UnknownLink
+ * @typedef {{ carCid: import('cardex/api').CARLink, blockCid: UnknownLink, offset: number }} BatchItem
+ * @typedef {{ add: (i: BatchItem) => void, remove: (cid: UnknownLink) => void, next: () => BatchItem[] }} BlockBatcher
  */
 
 /**
@@ -21,6 +21,11 @@ export class OrderedCarBlockBatcher {
     this.#queue.push(item)
   }
 
+  /** @param {UnknownLink} cid */
+  remove (cid) {
+    this.#queue = this.#queue.filter(item => item.blockCid.toString() !== cid.toString())
+  }
+
   next () {
     const queue = this.#queue
     let prevItem = queue.shift()

src/lib/blockstore.js

@@ -5,7 +5,7 @@ import { MultiCarIndex, StreamingCarIndex } from './car-index.js'
 import { OrderedCarBlockBatcher } from './block-batch.js'
 
 /**
- * @typedef {import('multiformats').CID} CID
+ * @typedef {import('multiformats').UnknownLink} UnknownLink
  * @typedef {import('cardex/api').IndexItem} IndexEntry
  * @typedef {string} MultihashString
  * @typedef {import('dagula').Block} Block
@@ -23,31 +23,22 @@ const MAX_ENCODED_BLOCK_LENGTH = (1024 * 1024 * 2) + 39 + 61
 export class R2Blockstore {
   /**
    * @param {R2Bucket} dataBucket
-   * @param {R2Bucket} indexBucket
-   * @param {CID[]} carCids
+   * @param {import('../bindings').IndexSource[]} indexSources
    */
-  constructor (dataBucket, indexBucket, carCids) {
+  constructor (dataBucket, indexSources) {
     this._dataBucket = dataBucket
     this._idx = new MultiCarIndex()
-    for (const carCid of carCids) {
-      this._idx.addIndex(carCid, new StreamingCarIndex(async () => {
-        const idxPath = `${carCid}/${carCid}.car.idx`
-        const idxObj = await indexBucket.get(idxPath)
-        if (!idxObj) {
-          throw Object.assign(new Error(`index not found: ${carCid}`), { code: 'ERR_MISSING_INDEX' })
-        }
-        return idxObj.body
-      }))
+    for (const src of indexSources) {
+      this._idx.addIndex(new StreamingCarIndex(src))
     }
   }
 
-  /** @param {CID} cid */
+  /** @param {UnknownLink} cid */
   async get (cid) {
     // console.log(`get ${cid}`)
-    const multiIdxEntry = await this._idx.get(cid)
-    if (!multiIdxEntry) return
-    const [carCid, entry] = multiIdxEntry
-    const carPath = `${carCid}/${carCid}.car`
+    const entry = await this._idx.get(cid)
+    if (!entry) return
+    const carPath = `${entry.origin}/${entry.origin}.car`
     const range = { offset: entry.offset }
     const res = await this._dataBucket.get(carPath, { range })
     if (!res) return
@@ -188,6 +179,8 @@ export class BatchingR2Blockstore extends R2Blockstore {
             }
             blocks.forEach(b => b.resolve(block))
             pendingBlocks.delete(key)
+            // remove from batcher if queued to be read
+            batcher.remove(blockHeader.cid)
           }
         } catch {
           break
@@ -205,14 +198,13 @@ export class BatchingR2Blockstore extends R2Blockstore {
     }
   }
 
-  /** @param {CID} cid */
+  /** @param {UnknownLink} cid */
   async get (cid) {
     // console.log(`get ${cid}`)
-    const multiIdxEntry = await this._idx.get(cid)
-    if (!multiIdxEntry) return
+    const entry = await this._idx.get(cid)
+    if (!entry) return
 
-    const [carCid, entry] = multiIdxEntry
-    this.#batcher.add({ carCid, blockCid: cid, offset: entry.offset })
+    this.#batcher.add({ carCid: entry.origin, blockCid: cid, offset: entry.offset })
 
     if (!entry.multihash) throw new Error('missing entry multihash')
     const key = mhToKey(entry.multihash.bytes)

src/lib/car-index.js

@@ -3,40 +3,38 @@ import { UniversalReader } from 'cardex/universal'
 import defer from 'p-defer'
 
 /**
- * @typedef {import('multiformats').CID} CID
- * @typedef {import('cardex/multihash-index-sorted/api').MultihashIndexItem} IndexEntry
+ * @typedef {import('multiformats').UnknownLink} UnknownLink
+ * @typedef {import('cardex/multi-index/api').MultiIndexItem & import('cardex/multihash-index-sorted/api').MultihashIndexItem} IndexEntry
  * @typedef {string} MultihashString
- * @typedef {{ get: (c: CID) => Promise<IndexEntry|undefined> }} CarIndex
+ * @typedef {{ get: (c: UnknownLink) => Promise<IndexEntry|undefined> }} CarIndex
  */
 
 export class MultiCarIndex {
-  /** @type {Map<CID, CarIndex>} */
+  /** @type {CarIndex[]} */
   #idxs
 
   constructor () {
-    this.#idxs = new Map()
+    this.#idxs = []
   }
 
   /**
-   * @param {CID} carCid
    * @param {CarIndex} index
    */
-  addIndex (carCid, index) {
-    this.#idxs.set(carCid, index)
+  addIndex (index) {
+    this.#idxs.push(index)
   }
 
   /**
-   * @param {CID} cid
-   * @returns {Promise<[CID, IndexEntry] | undefined>}
+   * @param {UnknownLink} cid
+   * @returns {Promise<IndexEntry | undefined>}
    */
   async get (cid) {
     const deferred = defer()
-    const idxEntries = Array.from(this.#idxs.entries())
 
     Promise
-      .allSettled(idxEntries.map(async ([carCid, idx]) => {
+      .allSettled(this.#idxs.map(async idx => {
         const entry = await idx.get(cid)
-        if (entry) deferred.resolve([carCid, entry])
+        if (entry) deferred.resolve(entry)
       }))
       .then(results => {
         // if not already resolved, check for rejections and reject
@@ -55,6 +53,9 @@ export class MultiCarIndex {
  * @implements {CarIndex}
  */
 export class StreamingCarIndex {
+  /** @type {import('../bindings').IndexSource} */
+  #source
+
   /** @type {Map<MultihashString, IndexEntry>} */
   #idx = new Map()
 
@@ -67,22 +68,27 @@ export class StreamingCarIndex {
   /** @type {Error?} */
   #buildError = null
 
-  /** @param {() => Promise<ReadableStream<Uint8Array>>} fetchIndex */
-  constructor (fetchIndex) {
-    this.#buildIndex(fetchIndex)
+  /** @param {import('../bindings').IndexSource} source */
+  constructor (source) {
+    this.#source = source
+    this.#buildIndex()
   }
 
-  /** @param {() => Promise<ReadableStream<Uint8Array>>} fetchIndex */
-  async #buildIndex (fetchIndex) {
+  async #buildIndex () {
     this.#building = true
     try {
-      const stream = await fetchIndex()
-      const idxReader = UniversalReader.createReader({ reader: stream.getReader() })
+      const idxObj = await this.#source.bucket.get(this.#source.key)
+      if (!idxObj) {
+        throw Object.assign(new Error(`index not found: ${this.#source.key}`), { code: 'ERR_MISSING_INDEX' })
+      }
+      const idxReader = UniversalReader.createReader({ reader: idxObj.body.getReader() })
       while (true) {
         const { done, value } = await idxReader.read()
         if (done) break
 
         const entry = /** @type {IndexEntry} */(value)
+        entry.origin = entry.origin ?? this.#source.origin
+
         const key = mhToKey(entry.multihash.bytes)
 
         // set this value in the index so any future requests for this key get
@@ -122,7 +128,7 @@ export class StreamingCarIndex {
     }
   }
 
-  /** @param {CID} cid */
+  /** @param {UnknownLink} cid */
   async get (cid) {
     if (this.#buildError) {
       throw new Error('failed to build index', { cause: this.#buildError })