Intercept fetches to R2 and use direct CARPARK pull

package.json

@@ -42,7 +42,7 @@
     "@ucanto/client": "^9.0.1",
     "@ucanto/principal": "^9.0.1",
     "@ucanto/transport": "^9.1.1",
-    "@web3-storage/blob-fetcher": "^2.4.3",
+    "@web3-storage/blob-fetcher": "^2.5.0",
     "@web3-storage/capabilities": "^17.4.1",
     "@web3-storage/gateway-lib": "^5.1.2",
     "dagula": "^8.0.0",
@@ -59,7 +59,7 @@
     "@types/node-fetch": "^2.6.11",
     "@types/sinon": "^17.0.3",
     "@web3-storage/content-claims": "^5.0.0",
-    "@web3-storage/public-bucket": "^1.1.0",
+    "@web3-storage/public-bucket": "^1.4.0",
     "@web3-storage/upload-client": "^16.1.1",
     "carstream": "^2.1.0",
     "chai": "^5.1.1",

src/index.js

@@ -29,7 +29,8 @@ import {
   withUcanInvocationHandler,
   withDelegationsStorage,
   withDelegationStubs,
-  withOptionsRequest
+  withOptionsRequest,
+  withCarParkFetch
 } from './middleware/index.js'
 import { instrument } from '@microlabs/otel-cf-workers'
 import { NoopSpanProcessor, TraceIdRatioBasedSampler } from '@opentelemetry/sdk-trace-base'
@@ -71,6 +72,7 @@ const middleware = composeMiddleware(
   withParsedIpfsUrl,
   withAuthToken,
   withLocator,
+  withCarParkFetch,
 
   // TODO: replace this with a handler to fetch the real delegations
   withDelegationStubs,

src/middleware/index.js

@@ -5,6 +5,7 @@ export { withRateLimit } from './withRateLimit.js'
 export { withVersionHeader } from './withVersionHeader.js'
 export { withAuthorizedSpace } from './withAuthorizedSpace.js'
 export { withLocator } from './withLocator.js'
+export { withCarParkFetch } from './withCarParkFetch.js'
 export { withEgressTracker } from './withEgressTracker.js'
 export { withEgressClient } from './withEgressClient.js'
 export { withDelegationStubs } from './withDelegationStubs.js'

src/middleware/withCarBlockHandler.js

@@ -128,7 +128,7 @@ export async function handleCarBlock (request, env, ctx) {
       last = range.length != null ? first + range.length - 1 : obj.size - 1
     }
     headers.set('Content-Range', `bytes ${first}-${last}/${obj.size}`)
-    contentLength = last - first
+    contentLength = last - first + 1
   }
   headers.set('Content-Length', contentLength.toString())
 

src/middleware/withCarParkFetch.js

@@ -0,0 +1,80 @@
+import { withSimpleSpan } from '@web3-storage/blob-fetcher/tracing/tracing'
+import { createHandler } from '@web3-storage/public-bucket/server'
+// eslint-disable-next-line
+import * as BucketAPI from '@web3-storage/public-bucket'
+
+/** @implements {BucketAPI.Bucket} */
+export class TraceBucket {
+  #bucket
+
+  /**
+   *
+   * @param {BucketAPI.Bucket} bucket
+   */
+  constructor (bucket) {
+    this.#bucket = bucket
+  }
+
+  /** @param {string} key */
+  head (key) {
+    return withSimpleSpan('bucket.head', this.#bucket.head, this.#bucket)(key)
+  }
+
+  /**
+   * @param {string} key
+   * @param {BucketAPI.GetOptions} [options]
+   */
+  get (key, options) {
+    return withSimpleSpan('bucket.get', this.#bucket.get, this.#bucket)(key, options)
+  }
+}
+
+/**
+ * @import {
+ *   Middleware,
+ *   Context as MiddlewareContext
+ * } from '@web3-storage/gateway-lib'
+ * @import {
+ *   CarParkFetchContext,
+ *   CarParkFetchEnvironment
+ * } from './withCarParkFetch.types.js'
+ */
+
+/**
+ * 20MiB should allow the worker to process ~4-5 concurrent requests that
+ * require a batch at the maximum size.
+ */
+const MAX_BATCH_SIZE = 20 * 1024 * 1024
+
+/**
+ * Adds {@link CarParkFetchContext.fetch} to the context. This version of fetch
+ * will pull directly from R2 CARPARK when present
+ *
+ * @type {Middleware<CarParkFetchContext, MiddlewareContext, CarParkFetchEnvironment>}
+ */
+export function withCarParkFetch (handler) {
+  return async (request, env, ctx) => {
+    // if carpark public bucket is not set, just use default
+    if (!env.CARPARK_PUBLIC_BUCKET_URL) {
+      return handler(request, env, { ...ctx, fetch: globalThis.fetch })
+    }
+    const bucket = new TraceBucket(/** @type {import('@web3-storage/public-bucket').Bucket} */ (env.CARPARK))
+    const bucketHandler = createHandler({ bucket, maxBatchSize: MAX_BATCH_SIZE })
+
+    /**
+     *
+     * @param {globalThis.RequestInfo | URL} input
+     * @param {globalThis.RequestInit} [init]
+     * @returns {Promise<globalThis.Response>}
+     */
+    const fetch = async (input, init) => {
+      const request = input instanceof Request ? input : new Request(input, init)
+      // check whether request is going to CARPARK
+      if (env.CARPARK_PUBLIC_BUCKET_URL && request.url.startsWith(env.CARPARK_PUBLIC_BUCKET_URL)) {
+        return bucketHandler(request)
+      }
+      return globalThis.fetch(input, init)
+    }
+    return handler(request, env, { ...ctx, fetch })
+  }
+}

src/middleware/withCarParkFetch.types.ts

@@ -0,0 +1,14 @@
+import {
+  Environment as MiddlewareEnvironment,
+  Context as MiddlewareContext,
+} from '@web3-storage/gateway-lib'
+import { R2Bucket } from '@cloudflare/workers-types'
+
+export interface CarParkFetchEnvironment extends MiddlewareEnvironment {
+  CARPARK: R2Bucket
+  CARPARK_PUBLIC_BUCKET_URL?: string
+}
+
+export interface CarParkFetchContext extends MiddlewareContext {
+  fetch: typeof globalThis.fetch
+}

src/middleware/withContentClaimsDagula.js

@@ -10,6 +10,7 @@ import * as BatchingFetcher from '@web3-storage/blob-fetcher/fetcher/batching'
  *   Middleware,
  * } from '@web3-storage/gateway-lib'
  * @import { LocatorContext } from './withLocator.types.js'
+ * @import { CarParkFetchContext } from './withCarParkFetch.types.js'
  * @import { Environment } from './withContentClaimsDagula.types.js'
  */
 
@@ -18,16 +19,16 @@ import * as BatchingFetcher from '@web3-storage/blob-fetcher/fetcher/batching'
  *
  * @type {(
  *   Middleware<
- *     BlockContext & DagContext & UnixfsContext & IpfsUrlContext & LocatorContext,
- *     IpfsUrlContext & LocatorContext,
+ *     BlockContext & DagContext & UnixfsContext & IpfsUrlContext & LocatorContext & CarParkFetchContext,
+ *     IpfsUrlContext & LocatorContext & CarParkFetchContext,
  *     Environment
  *   >
  * )}
  */
 export function withContentClaimsDagula (handler) {
   return async (request, env, ctx) => {
     const { locator } = ctx
-    const fetcher = BatchingFetcher.create(locator)
+    const fetcher = BatchingFetcher.create(locator, ctx.fetch)
     const dagula = new Dagula({
       async get (cid) {
         const res = await fetcher.fetch(cid.multihash)

test/miniflare/freeway.spec.js

@@ -338,7 +338,7 @@ describe('freeway', () => {
 
     const res = await miniflare.dispatchFetch(`http://localhost:8787/ipfs/${shards[0]}`, {
       headers: {
-        Range: `bytes=${rootBlock.blockOffset}-${rootBlock.blockOffset + rootBlock.blockLength}`
+        Range: `bytes=${rootBlock.blockOffset}-${rootBlock.blockOffset + rootBlock.blockLength - 1}`
       }
     })
     assert(res.ok)
@@ -349,7 +349,7 @@ describe('freeway', () => {
     const contentLength = parseInt(res.headers.get('Content-Length') ?? '0')
     assert(contentLength)
     assert.equal(contentLength, rootBlock.bytes.length)
-    assert.equal(res.headers.get('Content-Range'), `bytes ${rootBlock.blockOffset}-${rootBlock.blockOffset + rootBlock.blockLength}/${obj.size}`)
+    assert.equal(res.headers.get('Content-Range'), `bytes ${rootBlock.blockOffset}-${rootBlock.blockOffset + rootBlock.blockLength - 1}/${obj.size}`)
     assert.equal(res.headers.get('Content-Type'), 'application/vnd.ipld.car; version=1;')
     assert.equal(res.headers.get('Etag'), `"${shards[0]}"`)
   })