Intercept fetches to R2 and use direct CARPARK pull (#141)

# Goals

When we fetch blocks from an R2 CARPark URL, we appear to get TTFB
around 400ms. Interestingly, if we instead use Cloudflare Worker's
direct R2 access, our TTFB drops to around 200ms. Bandwidth doesn't seem
to change much, but this is still an important win.

# Implementation

- Using a seperate change in blob-fetcher to allow passing a custom
fetch implementation, we now write a version of fetch that checks URLs
to see if they are CAR Park URLs. if they are, instead of using the
fetch API, we use our direct connection to CARPark in the worker to
perform the fetch request.
- Pass this custom fetch to the batching fetch used by content claims
dagula
- Also, to assemble a browser like Response object use the existing code
in withCarBlockHandler, factored out to utility file
- In the process of doing this, I discovered a bug in that code where it
was treating the last byte in the range as exclusive to the returned
content rather inclusive, and this was undetected as the test was
expecting the wrong result.

# For Discussion

Comparison traces, note the average TTFB of 200 vs 400 on requests to R2

Regular fetch:
![Screenshot 2025-01-03 at 8 42
11 PM](https://github.com/user-attachments/assets/78981630-8614-4060-8a3e-20e6b2868924)

With intercepted direct gets to R2:
![Screenshot 2025-01-04 at 3 48
55 PM](https://github.com/user-attachments/assets/52a3857a-f3eb-46e0-85c9-5337ec0491dc)

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]}"`)
   })