Repository: ipfs/boxo (language: go)

Issue:
gateway:  remote backend implementations from bifrost-gateway
## About gateway backends

Right now, `boxo/gateway` comes with only one implementation of local backend (`BlocksBackend` in  [`gateway/blocks_backend.go`](https://github.com/ipfs/boxo/blob/main/gateway/blocks_backend.go)).

Remote backends that follow https://specs.ipfs.tech/http-gateways/trustless-gateway/ as data transfer protocol were created in 2023 as part of Project Rhea, but are hard to discover and use outside the no longer actively maintained [bifrost-gateway](https://github.com/ipfs/bifrost-gateway).

## Proposed improvement

We should salvage that work and make these backends useful to boxo users by moving remote backend implementations from `biforst-gateway` to `boxo/gateway`

Namely:
- `remote_blocks_backend.go` (porting https://github.com/ipfs/bifrost-gateway/blob/main/blockstore_proxy.go)  
- `remote_car_backend.go`  (porting https://github.com/ipfs/bifrost-gateway/blob/main/lib/graph_gateway.go)


This will not only benefit boxo users, but also allow us to add these backends to `rainbow` (https://github.com/ipfs/rainbow/issues/88) as alternative to libp2p one, allowing us to sunset  and archive [bifrost-gateway](https://github.com/ipfs/bifrost-gateway).


Discussion / hints:
Triage notes:
- having "Fetch" abstraction based on this would be useful for refactoring/maintaining `ipget` or sole RPC in kubo (nothing to maintain + same behavior as gateway)
- 

Fix patch:
diff --git a/.github/workflows/gateway-conformance.yml b/.github/workflows/gateway-conformance.yml
index c9c3eb072..0928d2650 100644
--- a/.github/workflows/gateway-conformance.yml
+++ b/.github/workflows/gateway-conformance.yml
@@ -1,17 +1,23 @@
 name: Gateway Conformance
+# This workflow runs https://github.com/ipfs/gateway-conformance
+# against different backend implementations of boxo/gateway
 
 on:
   push:
     branches:
       - main
   pull_request:
+  workflow_dispatch:
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.event_name }}-${{ github.event_name == 'push' && github.sha || github.ref }}
   cancel-in-progress: true
 
 jobs:
-  gateway-conformance:
+  # This test uses a static CAR file as a local blockstore,
+  # allowing us to test conformance against BlocksBackend (gateway/backend_blocks.go)
+  # which is used by implementations like Kubo
+  local-block-backend:
     runs-on: ubuntu-latest
     steps:
       # 1. Download the gateway-conformance fixtures
@@ -21,35 +27,171 @@ jobs:
           output: fixtures
           merged: true
 
-      # 2. Build the car-gateway
+      # 2. Build the gateway binary
+      - name: Checkout boxo
+        uses: actions/checkout@v4
+        with:
+          path: boxo
+      - name: Setup Go
+        uses: actions/setup-go@v5
+        with:
+          go-version-file: 'boxo/examples/go.mod'
+          cache-dependency-path: "boxo/**/*.sum"
+      - name: Build test-gateway
+        run: go build -o test-gateway
+        working-directory: boxo/examples/gateway/car-file
+
+      # 3. Start the gateway binary
+      - name: Start test-gateway
+        run: boxo/examples/gateway/car-file/test-gateway -c fixtures/fixtures.car -p 8040 &
+
+      # 4. Run the gateway-conformance tests
+      - name: Run gateway-conformance tests
+        uses: ipfs/gateway-conformance/.github/actions/test@v0.5
+        with:
+          gateway-url: http://127.0.0.1:8040
+          json: output.json
+          xml: output.xml
+          html: output.html
+          markdown: output.md
+          subdomain-url: http://example.net
+          specs: -trustless-ipns-gateway,-path-ipns-gateway,-subdomain-ipns-gateway,-dnslink-gateway
+
+      # 5. Upload the results
+      - name: Upload MD summary
+        if: failure() || success()
+        run: cat output.md >> $GITHUB_STEP_SUMMARY
+      - name: Upload HTML report
+        if: failure() || success()
+        uses: actions/upload-artifact@v4
+        with:
+          name: gateway-conformance_local-block-backend.html
+          path: output.html
+      - name: Upload JSON report
+        if: failure() || success()
+        uses: actions/upload-artifact@v4
+        with:
+          name: gateway-conformance_local-block-backend.json
+          path: output.json
+
+  # This test uses remote block gateway (?format=raw) as a remote blockstore,
+  # allowing us to test conformance against RemoteBlocksBackend
+  # (gateway/backend_blocks.go) which is used by implementations like
+  # rainbow configured to use with remote block backend
+  # Ref. https://specs.ipfs.tech/http-gateways/trustless-gateway/#block-responses-application-vnd-ipld-raw
+  remote-block-backend:
+    runs-on: ubuntu-latest
+    steps:
+      # 1. Download the gateway-conformance fixtures
+      - name: Download gateway-conformance fixtures
+        uses: ipfs/gateway-conformance/.github/actions/extract-fixtures@v0.5
+        with:
+          output: fixtures
+          merged: true
+
+      # 2. Build the gateway binaries
+      - name: Checkout boxo
+        uses: actions/checkout@v4
+        with:
+          path: boxo
       - name: Setup Go
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v5
+        with:
+          go-version-file: 'boxo/examples/go.mod'
+          cache-dependency-path: "boxo/**/*.sum"
+      - name: Build remote-block-backend # it will act as a trustless CAR gateway
+        run: go build -o remote-block-backend
+        working-directory: boxo/examples/gateway/car-file
+      - name: Build test-gateway # this one will be used for tests, it will use previous one as its remote block backend
+        run: go build -o test-gateway
+        working-directory: boxo/examples/gateway/proxy-blocks
+
+      # 3. Start the gateway binaries
+      - name: Start remote HTTP backend that serves application/vnd.ipld.raw
+        run: boxo/examples/gateway/car-file/remote-block-backend -c fixtures/fixtures.car -p 8030 & # this endpoint will respond to application/vnd.ipld.car requests
+      - name: Start gateway that uses the remote block backend
+        run: boxo/examples/gateway/proxy-blocks/test-gateway -g http://127.0.0.1:8030 -p 8040 &
+
+      # 4. Run the gateway-conformance tests
+      - name: Run gateway-conformance tests
+        uses: ipfs/gateway-conformance/.github/actions/test@v0.5
+        with:
+          gateway-url: http://127.0.0.1:8040 # we test gateway that is backed by a remote block gateway
+          json: output.json
+          xml: output.xml
+          html: output.html
+          markdown: output.md
+          subdomain-url: http://example.net
+          specs: -trustless-ipns-gateway,-path-ipns-gateway,-subdomain-ipns-gateway,-dnslink-gateway
+          args: -skip 'TestGatewayCache/.*_for_%2Fipfs%2F_with_only-if-cached_succeeds_when_in_local_datastore'
+
+      # 5. Upload the results
+      - name: Upload MD summary
+        if: failure() || success()
+        run: cat output.md >> $GITHUB_STEP_SUMMARY
+      - name: Upload HTML report
+        if: failure() || success()
+        uses: actions/upload-artifact@v4
         with:
-          go-version: 1.21.x
+          name: gateway-conformance_remote-block-backend.html
+          path: output.html
+      - name: Upload JSON report
+        if: failure() || success()
+        uses: actions/upload-artifact@v4
+        with:
+          name: gateway-conformance_remote-block-backend.json
+          path: output.json
+
+  # This test uses remote CAR gateway (?format=car, IPIP-402)
+  # allowing us to test conformance against remote CarFetcher backend.
+  # (gateway/backend_car_fetcher.go) which is used by implementations like
+  # rainbow configured to use with remote car backend
+  # Ref. https://specs.ipfs.tech/http-gateways/trustless-gateway/#car-responses-application-vnd-ipld-car
+  remote-car-backend:
+    runs-on: ubuntu-latest
+    steps:
+      # 1. Download the gateway-conformance fixtures
+      - name: Download gateway-conformance fixtures
+        uses: ipfs/gateway-conformance/.github/actions/extract-fixtures@v0.5
+        with:
+          output: fixtures
+          merged: true
+
+      # 2. Build the gateway binaries
       - name: Checkout boxo
         uses: actions/checkout@v4
         with:
           path: boxo
-      - name: Build car-gateway
-        run: go build -o car-gateway
-        working-directory: boxo/examples/gateway/car
+      - name: Setup Go
+        uses: actions/setup-go@v5
+        with:
+          go-version-file: 'boxo/examples/go.mod'
+          cache-dependency-path: "boxo/**/*.sum"
+      - name: Build remote-car-backend # it will act as a trustless CAR gateway
+        run: go build -o remote-car-backend
+        working-directory: boxo/examples/gateway/car-file
+      - name: Build test-gateway # this one will be used for tests, it will use previous one as its remote CAR backend
+        run: go build -o test-gateway
+        working-directory: boxo/examples/gateway/proxy-car
 
-      # 3. Start the car-gateway
-      - name: Start car-gateway
-        run: boxo/examples/gateway/car/car-gateway -c fixtures/fixtures.car -p 8040 &
+      # 3. Start the gateway binaries
+      - name: Start remote HTTP backend that serves application/vnd.ipld.car (IPIP-402)
+        run: boxo/examples/gateway/car-file/remote-car-backend -c fixtures/fixtures.car -p 8030 & # this endpoint will respond to application/vnd.ipld.raw requests
+      - name: Start gateway that uses the remote CAR backend
+        run: boxo/examples/gateway/proxy-car/test-gateway -g http://127.0.0.1:8030 -p 8040 &
 
       # 4. Run the gateway-conformance tests
       - name: Run gateway-conformance tests
         uses: ipfs/gateway-conformance/.github/actions/test@v0.5
         with:
-          gateway-url: http://127.0.0.1:8040
+          gateway-url: http://127.0.0.1:8040 # we test gateway that is backed by a remote car gateway
           json: output.json
           xml: output.xml
           html: output.html
           markdown: output.md
           subdomain-url: http://example.net
           specs: -trustless-ipns-gateway,-path-ipns-gateway,-subdomain-ipns-gateway,-dnslink-gateway
-          args: -skip 'TestGatewayCar/GET_response_for_application/vnd.ipld.car/Header_Content-Length'
+          args: -skip 'TestGatewayCache/.*_for_%2Fipfs%2F_with_only-if-cached_succeeds_when_in_local_datastore'
 
       # 5. Upload the results
       - name: Upload MD summary
@@ -57,13 +199,13 @@ jobs:
         run: cat output.md >> $GITHUB_STEP_SUMMARY
       - name: Upload HTML report
         if: failure() || success()
-        uses: actions/upload-artifact@v3
+        uses: actions/upload-artifact@v4
         with:
-          name: gateway-conformance.html
+          name: gateway-conformance_remote-car-backend.html
           path: output.html
       - name: Upload JSON report
         if: failure() || success()
-        uses: actions/upload-artifact@v3
+        uses: actions/upload-artifact@v4
         with:
-          name: gateway-conformance.json
+          name: gateway-conformance_remote-car-backend.json
           path: output.json
diff --git a/CHANGELOG.md b/CHANGELOG.md
index f2b810bac..ef6a25e86 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -16,7 +16,9 @@ The following emojis are used to highlight certain changes:
 
 ### Added
 
-* `gateway` now includes `NewRemoteBlocksBackend` which allows you to create a gateway backend that uses one or multiple other gateways as backend. These gateways must support RAW block requests (`application/vnd.ipld.raw`), as well as IPNS Record requests (`application/vnd.ipfs.ipns-record`). With this, we also introduced a `NewCacheBlockStore`, `NewRemoteBlockstore` and `NewRemoteValueStore`.
+* ✨ `gateway` has new backend possibilities:
+  * `NewRemoteBlocksBackend` allows you to create a gateway backend that uses one or multiple other gateways as backend. These gateways must support RAW block requests (`application/vnd.ipld.raw`), as well as IPNS Record requests (`application/vnd.ipfs.ipns-record`). With this, we also introduced `NewCacheBlockStore`, `NewRemoteBlockstore` and `NewRemoteValueStore`.
+  * `NewRemoteCarBackend` allows you to create a gateway backend that uses one or multiple Trustless Gateways as backend. These gateways must support CAR requests (`application/vnd.ipld.car`), as well as the extensions describe in [IPIP-402](https://specs.ipfs.tech/ipips/ipip-0402/). With this, we also introduced `NewCarBackend`, `NewRemoteCarFetcher` and `NewRetryCarFetcher`.
 
 ### Changed
 
diff --git a/examples/README.md b/examples/README.md
index fa5408732..d1d0021d8 100644
--- a/examples/README.md
+++ b/examples/README.md
@@ -27,6 +27,7 @@ Once you have your example finished, do not forget to run `go mod tidy` and addi
 ## Examples and Tutorials
 
 - [Fetching a UnixFS file by CID](./unixfs-file-cid)
-- [Gateway backed by a CAR file](./gateway/car)
-- [Gateway backed by a remote blockstore and IPNS resolver](./gateway/proxy)
+- [Gateway backed by a local blockstore in form of a CAR file](./gateway/car-file)
+- [Gateway backed by a remote (HTTP) blockstore and IPNS resolver](./gateway/proxy-blocks)
+- [Gateway backed by a remote (HTTP) CAR Gateway](./gateway/proxy-car)
 - [Delegated Routing V1 Command Line Client](./routing/delegated-routing-client/)
diff --git a/examples/gateway/car/README.md b/examples/gateway/car-file/README.md
similarity index 81%
rename from examples/gateway/car/README.md
rename to examples/gateway/car-file/README.md
index 2fea3fa66..2645d7b17 100644
--- a/examples/gateway/car/README.md
+++ b/examples/gateway/car-file/README.md
@@ -1,13 +1,16 @@
-# HTTP Gateway backed by a CAR File
+# HTTP Gateway backed by a CAR File as BlocksBackend
 
 This is an example that shows how to build a Gateway backed by the contents of
 a CAR file. A [CAR file](https://ipld.io/specs/transport/car/) is a Content
 Addressable aRchive that contains blocks.
 
+The `main.go` sets up a `blockService` backed by a static CAR file,
+and then uses it to initialize `gateway.NewBlocksBackend(blockService)`.
+
 ## Build
 
 ```bash
-> go build -o car-gateway
+> go build -o gateway
 ```
 
 ## Usage
@@ -23,7 +26,7 @@ Then, you can start the gateway with:
 
 
 ```
-./car-gateway -c data.car -p 8040
+./gateway -c data.car -p 8040
 ```
 
 ### Subdomain gateway
diff --git a/examples/gateway/car/main.go b/examples/gateway/car-file/main.go
similarity index 100%
rename from examples/gateway/car/main.go
rename to examples/gateway/car-file/main.go
diff --git a/examples/gateway/proxy/README.md b/examples/gateway/proxy-blocks/README.md
similarity index 97%
rename from examples/gateway/proxy/README.md
rename to examples/gateway/proxy-blocks/README.md
index 4164aad1e..505ecb131 100644
--- a/examples/gateway/proxy/README.md
+++ b/examples/gateway/proxy-blocks/README.md
@@ -18,7 +18,7 @@ gateway using `?format=ipns-record`. In addition, DNSLink lookups are done local
 ## Build
 
 ```bash
-> go build -o verifying-proxy
+> go build -o gateway
 ```
 
 ## Usage
@@ -28,7 +28,7 @@ types. Once you have it, run the proxy gateway with its address as the host para
 
 
 ```
-./verifying-proxy -g https://ipfs.io -p 8040
+./gateway -g https://trustless-gateway.link -p 8040
 ```
 
 ### Subdomain gateway
diff --git a/examples/gateway/proxy/main.go b/examples/gateway/proxy-blocks/main.go
similarity index 98%
rename from examples/gateway/proxy/main.go
rename to examples/gateway/proxy-blocks/main.go
index b1c155015..2953133c0 100644
--- a/examples/gateway/proxy/main.go
+++ b/examples/gateway/proxy-blocks/main.go
@@ -28,7 +28,7 @@ func main() {
 	defer (func() { _ = tp.Shutdown(ctx) })()
 
 	// Creates the gateway with the remote block store backend.
-	backend, err := gateway.NewRemoteBlocksBackend([]string{*gatewayUrlPtr})
+	backend, err := gateway.NewRemoteBlocksBackend([]string{*gatewayUrlPtr}, nil)
 	if err != nil {
 		log.Fatal(err)
 	}
diff --git a/examples/gateway/proxy-car/README.md b/examples/gateway/proxy-car/README.md
new file mode 100644
index 000000000..c06a1a657
--- /dev/null
+++ b/examples/gateway/proxy-car/README.md
@@ -0,0 +1,51 @@
+# Gateway as Proxy for Trustless CAR Remote Backend
+
+This is an example of building a "verifying proxy" Gateway that has no
+local on-disk blockstore, but instead, uses `application/vnd.ipld.car` and
+`application/vnd.ipfs.ipns-record` responses from a remote HTTP server that
+implements CAR support from [Trustless Gateway
+Specification](https://specs.ipfs.tech/http-gateways/trustless-gateway/).
+
+**NOTE:** the remote CAR backend MUST implement [IPIP-0402: Partial CAR Support on Trustless Gateways](https://specs.ipfs.tech/ipips/ipip-0402/)
+
+## Build
+
+```bash
+> go build -o gateway
+```
+
+## Usage
+
+First, you need a compliant gateway that supports both [CAR requests](https://www.iana.org/assignments/media-types/application/vnd.ipld.car) and IPNS Record response
+types. Once you have it, run the proxy gateway with its address as the host parameter:
+
+```
+./gateway -g https://trustless-gateway.link -p 8040
+```
+
+### Subdomain gateway
+
+Now you can access the gateway in [`localhost:8040`](http://localhost:8040/ipfs/bafybeiaysi4s6lnjev27ln5icwm6tueaw2vdykrtjkwiphwekaywqhcjze). It will
+behave like a regular [subdomain gateway](https://docs.ipfs.tech/how-to/address-ipfs-on-web/#subdomain-gateway),
+except for the fact that it runs no libp2p, and has no local blockstore.
+All data is provided by a remote trustless gateway, fetched as CAR files and IPNS Records, and verified locally.
+
+### Path gateway
+
+If you don't need Origin isolation and only care about hosting flat files,
+a plain [path gateway](https://docs.ipfs.tech/how-to/address-ipfs-on-web/#path-gateway) at
+[`127.0.0.1:8040`](http://127.0.0.1:8040/ipfs/bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi)
+may suffice.
+
+### DNSLink gateway
+
+Gateway supports hosting of [DNSLink](https://dnslink.dev/) websites. All you need is to pass `Host` header with FQDN that has DNSLink set up:
+
+```console
+$ curl -sH 'Host: en.wikipedia-on-ipfs.org' 'http://127.0.0.1:8080/wiki/' | head -3
+<!DOCTYPE html><html class="client-js"><head>
+  <meta charset="UTF-8">
+  <title>Wikipedia, the free encyclopedia</title>
+```
+
+Put it behind a reverse proxy terminating TLS (like Nginx) and voila!
diff --git a/examples/gateway/proxy-car/main.go b/examples/gateway/proxy-car/main.go
new file mode 100644
index 000000000..d03904549
--- /dev/null
+++ b/examples/gateway/proxy-car/main.go
@@ -0,0 +1,45 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"log"
+	"net/http"
+	"strconv"
+
+	"github.com/ipfs/boxo/examples/gateway/common"
+	"github.com/ipfs/boxo/gateway"
+)
+
+func main() {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	gatewayUrlPtr := flag.String("g", "", "gateway to proxy to")
+	port := flag.Int("p", 8040, "port to run this gateway from")
+	flag.Parse()
+
+	// Setups up tracing. This is optional and only required if the implementer
+	// wants to be able to enable tracing.
+	tp, err := common.SetupTracing(ctx, "CAR Gateway Example")
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer (func() { _ = tp.Shutdown(ctx) })()
+
+	// Creates the gateway with the remote car (IPIP-402) backend.
+	backend, err := gateway.NewRemoteCarBackend([]string{*gatewayUrlPtr}, nil)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	handler := common.NewHandler(backend)
+
+	log.Printf("Listening on http://localhost:%d", *port)
+	log.Printf("Try loading an image: http://localhost:%d/ipfs/bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi", *port)
+	log.Printf("Try browsing Wikipedia snapshot: http://localhost:%d/ipfs/bafybeiaysi4s6lnjev27ln5icwm6tueaw2vdykrtjkwiphwekaywqhcjze", *port)
+	log.Printf("Metrics available at http://127.0.0.1:%d/debug/metrics/prometheus", *port)
+	if err := http.ListenAndServe(":"+strconv.Itoa(*port), handler); err != nil {
+		log.Fatal(err)
+	}
+}
diff --git a/examples/go.mod b/examples/go.mod
index 9290f9158..ebae8a7e4 100644
--- a/examples/go.mod
+++ b/examples/go.mod
@@ -60,7 +60,11 @@ require (
 	github.com/huin/goupnp v1.3.0 // indirect
 	github.com/ipfs/bbloom v0.0.4 // indirect
 	github.com/ipfs/go-bitfield v1.1.0 // indirect
+	github.com/ipfs/go-blockservice v0.5.0 // indirect
+	github.com/ipfs/go-ipfs-blockstore v1.3.0 // indirect
 	github.com/ipfs/go-ipfs-delay v0.0.1 // indirect
+	github.com/ipfs/go-ipfs-ds-help v1.1.0 // indirect
+	github.com/ipfs/go-ipfs-exchange-interface v0.2.0 // indirect
 	github.com/ipfs/go-ipfs-pq v0.0.3 // indirect
 	github.com/ipfs/go-ipfs-redirects-file v0.1.1 // indirect
 	github.com/ipfs/go-ipfs-util v0.0.3 // indirect
@@ -69,9 +73,12 @@ require (
 	github.com/ipfs/go-ipld-legacy v0.2.1 // indirect
 	github.com/ipfs/go-log v1.0.5 // indirect
 	github.com/ipfs/go-log/v2 v2.5.1 // indirect
+	github.com/ipfs/go-merkledag v0.11.0 // indirect
 	github.com/ipfs/go-metrics-interface v0.0.1 // indirect
 	github.com/ipfs/go-peertaskqueue v0.8.1 // indirect
 	github.com/ipfs/go-unixfsnode v1.9.0 // indirect
+	github.com/ipfs/go-verifcid v0.0.2 // indirect
+	github.com/ipld/go-car v0.6.2 // indirect
 	github.com/ipld/go-codec-dagpb v1.6.0 // indirect
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
 	github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
diff --git a/examples/go.sum b/examples/go.sum
index 27414405f..50ed55608 100644
--- a/examples/go.sum
+++ b/examples/go.sum
@@ -136,6 +136,7 @@ github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OI
 github.com/google/pprof v0.0.0-20231229205709-960ae82b1e42 h1:dHLYa5D8/Ta0aLR2XcPsrkpAgGeFs6thhMcQK0oQ0n8=
 github.com/google/pprof v0.0.0-20231229205709-960ae82b1e42/go.mod h1:czg5+yv1E0ZGTi6S6vVK1mke0fV+FaUhNGcd6VRS9Ik=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
+github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.5.0 h1:1p67kYwdtXjb0gL0BPiP1Av9wiZPo5A8z2cWkTZ+eyU=
 github.com/google/uuid v1.5.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
@@ -167,13 +168,17 @@ github.com/ipfs/bbloom v0.0.4 h1:Gi+8EGJ2y5qiD5FbsbpX/TMNcJw8gSqr7eyjHa4Fhvs=
 github.com/ipfs/bbloom v0.0.4/go.mod h1:cS9YprKXpoZ9lT0n/Mw/a6/aFV6DTjTLYHeA+gyqMG0=
 github.com/ipfs/go-bitfield v1.1.0 h1:fh7FIo8bSwaJEh6DdTWbCeZ1eqOaOkKFI74SCnsWbGA=
 github.com/ipfs/go-bitfield v1.1.0/go.mod h1:paqf1wjq/D2BBmzfTVFlJQ9IlFOZpg422HL0HqsGWHU=
+github.com/ipfs/go-bitswap v0.11.0 h1:j1WVvhDX1yhG32NTC9xfxnqycqYIlhzEzLXG/cU1HyQ=
+github.com/ipfs/go-bitswap v0.11.0/go.mod h1:05aE8H3XOU+LXpTedeAS0OZpcO1WFsj5niYQH9a1Tmk=
 github.com/ipfs/go-block-format v0.2.0 h1:ZqrkxBA2ICbDRbK8KJs/u0O3dlp6gmAuuXUJNiW1Ycs=
 github.com/ipfs/go-block-format v0.2.0/go.mod h1:+jpL11nFx5A/SPpsoBn6Bzkra/zaArfSmsknbPMYgzM=
 github.com/ipfs/go-blockservice v0.5.0 h1:B2mwhhhVQl2ntW2EIpaWPwSCxSuqr5fFA93Ms4bYLEY=
 github.com/ipfs/go-blockservice v0.5.0/go.mod h1:W6brZ5k20AehbmERplmERn8o2Ni3ZZubvAxaIUeaT6w=
+github.com/ipfs/go-cid v0.0.5/go.mod h1:plgt+Y5MnOey4vO4UlUazGqdbEXuFYitED67FexhXog=
 github.com/ipfs/go-cid v0.0.6/go.mod h1:6Ux9z5e+HpkQdckYoX1PG/6xqKspzlEIR5SDmgqgC/I=
 github.com/ipfs/go-cid v0.4.1 h1:A/T3qGvxi4kpKWWcPC/PgbvDA2bjVLO7n4UeVwnbs/s=
 github.com/ipfs/go-cid v0.4.1/go.mod h1:uQHwDeX4c6CtyrFwdqyhpNcxVewur1M7l7fNU7LKwZk=
+github.com/ipfs/go-datastore v0.5.0/go.mod h1:9zhEApYMTl17C8YDp7JmU7sQZi2/wqiYh73hakZ90Bk=
 github.com/ipfs/go-datastore v0.6.0 h1:JKyz+Gvz1QEZw0LsX1IBn+JFCJQH4SJVFtM4uWU0Myk=
 github.com/ipfs/go-datastore v0.6.0/go.mod h1:rt5M3nNbSO/8q1t4LNkLyUwRs8HupMeN/8O4Vn9YAT8=
 github.com/ipfs/go-detect-race v0.0.1 h1:qX/xay2W3E4Q1U7d9lNs1sU9nvguX0a7319XbyQ6cOk=
@@ -184,6 +189,7 @@ github.com/ipfs/go-ipfs-blocksutil v0.0.1 h1:Eh/H4pc1hsvhzsQoMEP3Bke/aW5P5rVM1IW
 github.com/ipfs/go-ipfs-blocksutil v0.0.1/go.mod h1:Yq4M86uIOmxmGPUHv/uI7uKqZNtLb449gwKqXjIsnRk=
 github.com/ipfs/go-ipfs-chunker v0.0.5 h1:ojCf7HV/m+uS2vhUGWcogIIxiO5ubl5O57Q7NapWLY8=
 github.com/ipfs/go-ipfs-chunker v0.0.5/go.mod h1:jhgdF8vxRHycr00k13FM8Y0E+6BoalYeobXmUyTreP8=
+github.com/ipfs/go-ipfs-delay v0.0.0-20181109222059-70721b86a9a8/go.mod h1:8SP1YXK1M1kXuc4KJZINY3TQQ03J2rwBG9QfXmbRPrw=
 github.com/ipfs/go-ipfs-delay v0.0.1 h1:r/UXYyRcddO6thwOnhiznIAiSvxMECGgtv35Xs1IeRQ=
 github.com/ipfs/go-ipfs-delay v0.0.1/go.mod h1:8SP1YXK1M1kXuc4KJZINY3TQQ03J2rwBG9QfXmbRPrw=
 github.com/ipfs/go-ipfs-ds-help v1.1.0 h1:yLE2w9RAsl31LtfMt91tRZcrx+e61O5mDxFRR994w4Q=
@@ -196,6 +202,8 @@ github.com/ipfs/go-ipfs-pq v0.0.3 h1:YpoHVJB+jzK15mr/xsWC574tyDLkezVrDNeaalQBsTE
 github.com/ipfs/go-ipfs-pq v0.0.3/go.mod h1:btNw5hsHBpRcSSgZtiNm/SLj5gYIZ18AKtv3kERkRb4=
 github.com/ipfs/go-ipfs-redirects-file v0.1.1 h1:Io++k0Vf/wK+tfnhEh63Yte1oQK5VGT2hIEYpD0Rzx8=
 github.com/ipfs/go-ipfs-redirects-file v0.1.1/go.mod h1:tAwRjCV0RjLTjH8DR/AU7VYvfQECg+lpUy2Mdzv7gyk=
+github.com/ipfs/go-ipfs-routing v0.3.0 h1:9W/W3N+g+y4ZDeffSgqhgo7BsBSJwPMcyssET9OWevc=
+github.com/ipfs/go-ipfs-routing v0.3.0/go.mod h1:dKqtTFIql7e1zYsEuWLyuOU+E0WJWW8JjbTPLParDWo=
 github.com/ipfs/go-ipfs-util v0.0.3 h1:2RFdGez6bu2ZlZdI+rWfIdbQb1KudQp3VGwPtdNCmE0=
 github.com/ipfs/go-ipfs-util v0.0.3/go.mod h1:LHzG1a0Ig4G+iZ26UUOMjHd+lfM84LZCrn17xAKWBvs=
 github.com/ipfs/go-ipld-cbor v0.1.0 h1:dx0nS0kILVivGhfWuB6dUpMa/LAwElHPw1yOGYopoYs=
@@ -221,6 +229,8 @@ github.com/ipfs/go-unixfsnode v1.9.0 h1:ubEhQhr22sPAKO2DNsyVBW7YB/zA8Zkif25aBvz8
 github.com/ipfs/go-unixfsnode v1.9.0/go.mod h1:HxRu9HYHOjK6HUqFBAi++7DVoWAHn0o4v/nZ/VA+0g8=
 github.com/ipfs/go-verifcid v0.0.2 h1:XPnUv0XmdH+ZIhLGKg6U2vaPaRDXb9urMyNVCE7uvTs=
 github.com/ipfs/go-verifcid v0.0.2/go.mod h1:40cD9x1y4OWnFXbLNJYRe7MpNvWlMn3LZAG5Wb4xnPU=
+github.com/ipld/go-car v0.6.2 h1:Hlnl3Awgnq8icK+ze3iRghk805lu8YNq3wlREDTF2qc=
+github.com/ipld/go-car v0.6.2/go.mod h1:oEGXdwp6bmxJCZ+rARSkDliTeYnVzv3++eXajZ+Bmr8=
 github.com/ipld/go-car/v2 v2.13.1 h1:KnlrKvEPEzr5IZHKTXLAEub+tPrzeAFQVRlSQvuxBO4=
 github.com/ipld/go-car/v2 v2.13.1/go.mod h1:QkdjjFNGit2GIkpQ953KBwowuoukoM75nP/JI1iDJdo=
 github.com/ipld/go-codec-dagpb v1.6.0 h1:9nYazfyu9B1p3NAgfVdpRco3Fs2nFC72DqVsMj6rOcc=
@@ -251,6 +261,7 @@ github.com/klauspost/cpuid/v2 v2.2.6/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZY
 github.com/koron/go-ssdp v0.0.4 h1:1IDwrghSKYM7yLf7XCzbByg2sJ/JcNOZRXS2jczTwz0=
 github.com/koron/go-ssdp v0.0.4/go.mod h1:oDXq+E5IL5q0U8uSBcoAXzTzInwy5lEgC91HoKtbmZk=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
@@ -337,6 +348,7 @@ github.com/multiformats/go-multiaddr-dns v0.3.1 h1:QgQgR+LQVt3NPTjbrLLpsaT2ufAA2
 github.com/multiformats/go-multiaddr-dns v0.3.1/go.mod h1:G/245BRQ6FJGmryJCrOuTdB37AMA5AMOVuO6NY3JwTk=
 github.com/multiformats/go-multiaddr-fmt v0.1.0 h1:WLEFClPycPkp4fnIzoFoV9FVd49/eQsuaL3/CWe167E=
 github.com/multiformats/go-multiaddr-fmt v0.1.0/go.mod h1:hGtDIW4PU4BqJ50gW2quDuPVjyWNZxToGUh/HwTZYJo=
+github.com/multiformats/go-multibase v0.0.1/go.mod h1:bja2MqRZ3ggyXtZSEDKpl0uO/gviWFaSteVbWT51qgs=
 github.com/multiformats/go-multibase v0.0.3/go.mod h1:5+1R4eQrT3PkYZ24C3W2Ue2tPwIdYQD509ZjSb5y9Oc=
 github.com/multiformats/go-multibase v0.2.0 h1:isdYCVLvksgWlMW9OZRYJEa9pZETFivncJHmHnnd87g=
 github.com/multiformats/go-multibase v0.2.0/go.mod h1:bFBZX4lKCA/2lyOFSAoKH5SS6oPyjtnzK/XTFDPkNuk=
@@ -700,6 +712,7 @@ google.golang.org/protobuf v1.32.0 h1:pPC6BG5ex8PDFnkbrGU3EixyhKcQ2aDuBS36lqK/C7
 google.golang.org/protobuf v1.32.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
diff --git a/gateway/backend.go b/gateway/backend.go
new file mode 100644
index 000000000..ae54b14f1
--- /dev/null
+++ b/gateway/backend.go
@@ -0,0 +1,173 @@
+package gateway
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/ipfs/boxo/ipns"
+	"github.com/ipfs/boxo/namesys"
+	"github.com/ipfs/boxo/path"
+	"github.com/ipfs/boxo/path/resolver"
+	"github.com/ipfs/go-cid"
+	routinghelpers "github.com/libp2p/go-libp2p-routing-helpers"
+	"github.com/libp2p/go-libp2p/core/routing"
+	"github.com/prometheus/client_golang/prometheus"
+	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
+)
+
+type backendOptions struct {
+	ns namesys.NameSystem
+	vs routing.ValueStore
+	r  resolver.Resolver
+
+	// Only used by [CarBackend]:
+	promRegistry    prometheus.Registerer
+	getBlockTimeout time.Duration
+}
+
+// WithNameSystem sets the name system to use with the different backends. If not set
+// it will use the default DNSLink resolver generated by [NewDNSResolver] along
+// with any configured [routing.ValueStore].
+func WithNameSystem(ns namesys.NameSystem) BackendOption {
+	return func(opts *backendOptions) error {
+		opts.ns = ns
+		return nil
+	}
+}
+
+// WithValueStore sets the [routing.ValueStore] to use with the different backends.
+func WithValueStore(vs routing.ValueStore) BackendOption {
+	return func(opts *backendOptions) error {
+		opts.vs = vs
+		return nil
+	}
+}
+
+// WithResolver sets the [resolver.Resolver] to use with the different backends.
+func WithResolver(r resolver.Resolver) BackendOption {
+	return func(opts *backendOptions) error {
+		opts.r = r
+		return nil
+	}
+}
+
+// WithPrometheusRegistry sets the registry to use with [CarBackend].
+func WithPrometheusRegistry(reg prometheus.Registerer) BackendOption {
+	return func(opts *backendOptions) error {
+		opts.promRegistry = reg
+		return nil
+	}
+}
+
+const DefaultGetBlockTimeout = time.Second * 60
+
+// WithGetBlockTimeout sets a custom timeout when getting blocks from the
+// [CarFetcher] to use with [CarBackend]. By default, [DefaultGetBlockTimeout]
+// is used.
+func WithGetBlockTimeout(dur time.Duration) BackendOption {
+	return func(opts *backendOptions) error {
+		opts.getBlockTimeout = dur
+		return nil
+	}
+}
+
+type BackendOption func(options *backendOptions) error
+
+// baseBackend contains some common backend functionalities that are shared by
+// different backend implementations.
+type baseBackend struct {
+	routing routing.ValueStore
+	namesys namesys.NameSystem
+}
+
+func newBaseBackend(vs routing.ValueStore, ns namesys.NameSystem) (baseBackend, error) {
+	if vs == nil {
+		vs = routinghelpers.Null{}
+	}
+
+	if ns == nil {
+		dns, err := NewDNSResolver(nil, nil)
+		if err != nil {
+			return baseBackend{}, err
+		}
+
+		ns, err = namesys.NewNameSystem(vs, namesys.WithDNSResolver(dns))
+		if err != nil {
+			return baseBackend{}, err
+		}
+	}
+
+	return baseBackend{
+		routing: vs,
+		namesys: ns,
+	}, nil
+}
+
+func (bb *baseBackend) ResolveMutable(ctx context.Context, p path.Path) (path.ImmutablePath, time.Duration, time.Time, error) {
+	switch p.Namespace() {
+	case path.IPNSNamespace:
+		res, err := namesys.Resolve(ctx, bb.namesys, p)
+		if err != nil {
+			return path.ImmutablePath{}, 0, time.Time{}, err
+		}
+		ip, err := path.NewImmutablePath(res.Path)
+		if err != nil {
+			return path.ImmutablePath{}, 0, time.Time{}, err
+		}
+		return ip, res.TTL, res.LastMod, nil
+	case path.IPFSNamespace:
+		ip, err := path.NewImmutablePath(p)
+		return ip, 0, time.Time{}, err
+	default:
+		return path.ImmutablePath{}, 0, time.Time{}, NewErrorStatusCode(fmt.Errorf("unsupported path namespace: %s", p.Namespace()), http.StatusNotImplemented)
+	}
+}
+
+func (bb *baseBackend) GetIPNSRecord(ctx context.Context, c cid.Cid) ([]byte, error) {
+	if bb.routing == nil {
+		return nil, NewErrorStatusCode(errors.New("IPNS Record responses are not supported by this gateway"), http.StatusNotImplemented)
+	}
+
+	name, err := ipns.NameFromCid(c)
+	if err != nil {
+		return nil, NewErrorStatusCode(err, http.StatusBadRequest)
+	}
+
+	return bb.routing.GetValue(ctx, string(name.RoutingKey()))
+}
+
+func (bb *baseBackend) GetDNSLinkRecord(ctx context.Context, hostname string) (path.Path, error) {
+	if bb.namesys != nil {
+		p, err := path.NewPath("/ipns/" + hostname)
+		if err != nil {
+			return nil, err
+		}
+		res, err := bb.namesys.Resolve(ctx, p, namesys.ResolveWithDepth(1))
+		if err == namesys.ErrResolveRecursion {
+			err = nil
+		}
+		return res.Path, err
+	}
+
+	return nil, NewErrorStatusCode(errors.New("not implemented"), http.StatusNotImplemented)
+}
+
+// newRemoteHTTPClient creates a new [http.Client] that is optimized for retrieving
+// multiple blocks from a single gateway concurrently.
+func newRemoteHTTPClient() *http.Client {
+	transport := &http.Transport{
+		MaxIdleConns:        1000,
+		MaxConnsPerHost:     100,
+		MaxIdleConnsPerHost: 100,
+		IdleConnTimeout:     90 * time.Second,
+		ForceAttemptHTTP2:   true,
+	}
+
+	return &http.Client{
+		Timeout:   DefaultGetBlockTimeout,
+		Transport: otelhttp.NewTransport(transport),
+	}
+}
diff --git a/gateway/blocks_backend.go b/gateway/backend_blocks.go
similarity index 85%
rename from gateway/blocks_backend.go
rename to gateway/backend_blocks.go
index d85c2846b..42440dfcd 100644
--- a/gateway/blocks_backend.go
+++ b/gateway/backend_blocks.go
@@ -8,18 +8,16 @@ import (
 	"io"
 	"net/http"
 	"strings"
-	"time"
 
 	"github.com/ipfs/boxo/blockservice"
 	blockstore "github.com/ipfs/boxo/blockstore"
+	"github.com/ipfs/boxo/exchange/offline"
 	"github.com/ipfs/boxo/fetcher"
 	bsfetcher "github.com/ipfs/boxo/fetcher/impl/blockservice"
 	"github.com/ipfs/boxo/files"
 	"github.com/ipfs/boxo/ipld/merkledag"
 	ufile "github.com/ipfs/boxo/ipld/unixfs/file"
 	uio "github.com/ipfs/boxo/ipld/unixfs/io"
-	"github.com/ipfs/boxo/ipns"
-	"github.com/ipfs/boxo/namesys"
 	"github.com/ipfs/boxo/path"
 	"github.com/ipfs/boxo/path/resolver"
 	blocks "github.com/ipfs/go-block-format"
@@ -38,8 +36,6 @@ import (
 	"github.com/ipld/go-ipld-prime/traversal"
 	"github.com/ipld/go-ipld-prime/traversal/selector"
 	selectorparse "github.com/ipld/go-ipld-prime/traversal/selector/parse"
-	routinghelpers "github.com/libp2p/go-libp2p-routing-helpers"
-	"github.com/libp2p/go-libp2p/core/routing"
 	mc "github.com/multiformats/go-multicodec"
 
 	// Ensure basic codecs are registered.
@@ -51,54 +47,18 @@ import (
 
 // BlocksBackend is an [IPFSBackend] implementation based on a [blockservice.BlockService].
 type BlocksBackend struct {
+	baseBackend
 	blockStore   blockstore.Blockstore
 	blockService blockservice.BlockService
 	dagService   format.DAGService
 	resolver     resolver.Resolver
-
-	// Optional routing system to handle /ipns addresses.
-	namesys namesys.NameSystem
-	routing routing.ValueStore
 }
 
 var _ IPFSBackend = (*BlocksBackend)(nil)
 
-type blocksBackendOptions struct {
-	ns namesys.NameSystem
-	vs routing.ValueStore
-	r  resolver.Resolver
-}
-
-// WithNameSystem sets the name system to use with the [BlocksBackend]. If not set
-// it will use the default DNSLink resolver generated by [NewDNSResolver] along
-// with any configured [routing.ValueStore].
-func WithNameSystem(ns namesys.NameSystem) BlocksBackendOption {
-	return func(opts *blocksBackendOptions) error {
-		opts.ns = ns
-		return nil
-	}
-}
-
-// WithValueStore sets the [routing.ValueStore] to use with the [BlocksBackend].
-func WithValueStore(vs routing.ValueStore) BlocksBackendOption {
-	return func(opts *blocksBackendOptions) error {
-		opts.vs = vs
-		return nil
-	}
-}
-
-// WithResolver sets the [resolver.Resolver] to use with the [BlocksBackend].
-func WithResolver(r resolver.Resolver) BlocksBackendOption {
-	return func(opts *blocksBackendOptions) error {
-		opts.r = r
-		return nil
-	}
-}
-
-type BlocksBackendOption func(options *blocksBackendOptions) error
-
-func NewBlocksBackend(blockService blockservice.BlockService, opts ...BlocksBackendOption) (*BlocksBackend, error) {
-	var compiledOptions blocksBackendOptions
+// NewBlocksBackend creates a new [BlocksBackend] backed by a [blockservice.BlockService].
+func NewBlocksBackend(blockService blockservice.BlockService, opts ...BackendOption) (*BlocksBackend, error) {
+	var compiledOptions backendOptions
 	for _, o := range opts {
 		if err := o(&compiledOptions); err != nil {
 			return nil, err
@@ -108,50 +68,51 @@ func NewBlocksBackend(blockService blockservice.BlockService, opts ...BlocksBack
 	// Setup the DAG services, which use the CAR block store.
 	dagService := merkledag.NewDAGService(blockService)
 
-	// Setup a name system so that we are able to resolve /ipns links.
-	var (
-		ns namesys.NameSystem
-		vs routing.ValueStore
-		r  resolver.Resolver
-	)
-
-	vs = compiledOptions.vs
-	if vs == nil {
-		vs = routinghelpers.Null{}
-	}
-
-	ns = compiledOptions.ns
-	if ns == nil {
-		dns, err := NewDNSResolver(nil, nil)
-		if err != nil {
-			return nil, err
-		}
-
-		ns, err = namesys.NewNameSystem(vs, namesys.WithDNSResolver(dns))
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	r = compiledOptions.r
+	// Setup the [resolver.Resolver] if not provided.
+	r := compiledOptions.r
 	if r == nil {
-		// Setup the UnixFS resolver.
 		fetcherCfg := bsfetcher.NewFetcherConfig(blockService)
 		fetcherCfg.PrototypeChooser = dagpb.AddSupportToChooser(bsfetcher.DefaultPrototypeChooser)
 		fetcher := fetcherCfg.WithReifier(unixfsnode.Reify)
 		r = resolver.NewBasicResolver(fetcher)
 	}
 
+	// Setup the [baseBackend] which takes care of some shared functionality, such
+	// as resolving /ipns links.
+	baseBackend, err := newBaseBackend(compiledOptions.vs, compiledOptions.ns)
+	if err != nil {
+		return nil, err
+	}
+
 	return &BlocksBackend{
+		baseBackend:  baseBackend,
 		blockStore:   blockService.Blockstore(),
 		blockService: blockService,
 		dagService:   dagService,
 		resolver:     r,
-		routing:      vs,
-		namesys:      ns,
 	}, nil
 }
 
+// NewRemoteBlocksBackend creates a new [BlocksBackend] backed by one or more
+// gateways. These gateways must support RAW block requests and IPNS Record
+// requests. See [NewRemoteBlockstore] and [NewRemoteValueStore] for more details.
+//
+// To create a more custom [BlocksBackend], please use [NewBlocksBackend] directly.
+func NewRemoteBlocksBackend(gatewayURL []string, httpClient *http.Client, opts ...BackendOption) (*BlocksBackend, error) {
+	blockStore, err := NewRemoteBlockstore(gatewayURL, httpClient)
+	if err != nil {
+		return nil, err
+	}
+
+	valueStore, err := NewRemoteValueStore(gatewayURL, httpClient)
+	if err != nil {
+		return nil, err
+	}
+
+	blockService := blockservice.New(blockStore, offline.Exchange(blockStore))
+	return NewBlocksBackend(blockService, append(opts, WithValueStore(valueStore))...)
+}
+
 func (bb *BlocksBackend) Get(ctx context.Context, path path.ImmutablePath, ranges ...ByteRange) (ContentPathMetadata, *GetResponse, error) {
 	md, nd, err := bb.getNode(ctx, path)
 	if err != nil {
@@ -367,9 +328,17 @@ func (bb *BlocksBackend) GetCAR(ctx context.Context, p path.ImmutablePath, param
 		unixfsnode.AddUnixFSReificationToLinkSystem(&lsys)
 		lsys.StorageReadOpener = blockOpener(ctx, blockGetter)
 
+		// First resolve the path since we always need to.
+		lastCid, remainder, err := pathResolver.ResolveToLastNode(ctx, p)
+		if err != nil {
+			// io.PipeWriter.CloseWithError always returns nil.
+			_ = w.CloseWithError(err)
+			return
+		}
+
 		// TODO: support selectors passed as request param: https://github.com/ipfs/kubo/issues/8769
 		// TODO: this is very slow if blocks are remote due to linear traversal. Do we need deterministic traversals here?
-		carWriteErr := walkGatewaySimpleSelector(ctx, p, params, &lsys, pathResolver)
+		carWriteErr := walkGatewaySimpleSelector(ctx, lastCid, nil, remainder, params, &lsys)
 
 		// io.PipeWriter.CloseWithError always returns nil.
 		_ = w.CloseWithError(carWriteErr)
@@ -379,29 +348,49 @@ func (bb *BlocksBackend) GetCAR(ctx context.Context, p path.ImmutablePath, param
 }
 
 // walkGatewaySimpleSelector walks the subgraph described by the path and terminal element parameters
-func walkGatewaySimpleSelector(ctx context.Context, p path.ImmutablePath, params CarParams, lsys *ipld.LinkSystem, pathResolver resolver.Resolver) error {
-	// First resolve the path since we always need to.
-	lastCid, remainder, err := pathResolver.ResolveToLastNode(ctx, p)
-	if err != nil {
-		return err
-	}
-
+func walkGatewaySimpleSelector(ctx context.Context, lastCid cid.Cid, terminalBlk blocks.Block, remainder []string, params CarParams, lsys *ipld.LinkSystem) error {
 	lctx := ipld.LinkContext{Ctx: ctx}
 	pathTerminalCidLink := cidlink.Link{Cid: lastCid}
 
 	// If the scope is the block, now we only need to retrieve the root block of the last element of the path.
 	if params.Scope == DagScopeBlock {
-		_, err = lsys.LoadRaw(lctx, pathTerminalCidLink)
+		_, err := lsys.LoadRaw(lctx, pathTerminalCidLink)
 		return err
 	}
 
-	// If we're asking for everything then give it
-	if params.Scope == DagScopeAll {
-		lastCidNode, err := lsys.Load(lctx, pathTerminalCidLink, basicnode.Prototype.Any)
+	pc := dagpb.AddSupportToChooser(func(lnk ipld.Link, lnkCtx ipld.LinkContext) (ipld.NodePrototype, error) {
+		if tlnkNd, ok := lnkCtx.LinkNode.(schema.TypedLinkNode); ok {
+			return tlnkNd.LinkTargetNodePrototype(), nil
+		}
+		return basicnode.Prototype.Any, nil
+	})
+
+	np, err := pc(pathTerminalCidLink, lctx)
+	if err != nil {
+		return err
+	}
+
+	var lastCidNode datamodel.Node
+	if terminalBlk != nil {
+		decoder, err := lsys.DecoderChooser(pathTerminalCidLink)
 		if err != nil {
 			return err
 		}
+		nb := np.NewBuilder()
+		blockData := terminalBlk.RawData()
+		if err := decoder(nb, bytes.NewReader(blockData)); err != nil {
+			return err
+		}
+		lastCidNode = nb.Build()
+	} else {
+		lastCidNode, err = lsys.Load(lctx, pathTerminalCidLink, np)
+		if err != nil {
+			return err
+		}
+	}
 
+	// If we're asking for everything then give it
+	if params.Scope == DagScopeAll {
 		sel, err := selector.ParseSelector(selectorparse.CommonSelector_ExploreAllRecursively)
 		if err != nil {
 			return err
@@ -427,23 +416,6 @@ func walkGatewaySimpleSelector(ctx context.Context, p path.ImmutablePath, params
 	// From now on, dag-scope=entity!
 	// Since we need more of the graph load it to figure out what we have
 	// This includes determining if the terminal node is UnixFS or not
-	pc := dagpb.AddSupportToChooser(func(lnk ipld.Link, lnkCtx ipld.LinkContext) (ipld.NodePrototype, error) {
-		if tlnkNd, ok := lnkCtx.LinkNode.(schema.TypedLinkNode); ok {
-			return tlnkNd.LinkTargetNodePrototype(), nil
-		}
-		return basicnode.Prototype.Any, nil
-	})
-
-	np, err := pc(pathTerminalCidLink, lctx)
-	if err != nil {
-		return err
-	}
-
-	lastCidNode, err := lsys.Load(lctx, pathTerminalCidLink, np)
-	if err != nil {
-		return err
-	}
-
 	if pbn, ok := lastCidNode.(dagpb.PBNode); !ok {
 		// If it's not valid dag-pb then we're done
 		return nil
@@ -630,55 +602,6 @@ func (bb *BlocksBackend) getPathRoots(ctx context.Context, contentPath path.Immu
 	return pathRoots, lastPath, remainder, nil
 }
 
-func (bb *BlocksBackend) ResolveMutable(ctx context.Context, p path.Path) (path.ImmutablePath, time.Duration, time.Time, error) {
-	switch p.Namespace() {
-	case path.IPNSNamespace:
-		res, err := namesys.Resolve(ctx, bb.namesys, p)
-		if err != nil {
-			return path.ImmutablePath{}, 0, time.Time{}, err
-		}
-		ip, err := path.NewImmutablePath(res.Path)
-		if err != nil {
-			return path.ImmutablePath{}, 0, time.Time{}, err
-		}
-		return ip, res.TTL, res.LastMod, nil
-	case path.IPFSNamespace:
-		ip, err := path.NewImmutablePath(p)
-		return ip, 0, time.Time{}, err
-	default:
-		return path.ImmutablePath{}, 0, time.Time{}, NewErrorStatusCode(fmt.Errorf("unsupported path namespace: %s", p.Namespace()), http.StatusNotImplemented)
-	}
-}
-
-func (bb *BlocksBackend) GetIPNSRecord(ctx context.Context, c cid.Cid) ([]byte, error) {
-	if bb.routing == nil {
-		return nil, NewErrorStatusCode(errors.New("IPNS Record responses are not supported by this gateway"), http.StatusNotImplemented)
-	}
-
-	name, err := ipns.NameFromCid(c)
-	if err != nil {
-		return nil, NewErrorStatusCode(err, http.StatusBadRequest)
-	}
-
-	return bb.routing.GetValue(ctx, string(name.RoutingKey()))
-}
-
-func (bb *BlocksBackend) GetDNSLinkRecord(ctx context.Context, hostname string) (path.Path, error) {
-	if bb.namesys != nil {
-		p, err := path.NewPath("/ipns/" + hostname)
-		if err != nil {
-			return nil, err
-		}
-		res, err := bb.namesys.Resolve(ctx, p, namesys.ResolveWithDepth(1))
-		if err == namesys.ErrResolveRecursion {
-			err = nil
-		}
-		return res.Path, err
-	}
-
-	return nil, NewErrorStatusCode(errors.New("not implemented"), http.StatusNotImplemented)
-}
-
 func (bb *BlocksBackend) IsCached(ctx context.Context, p path.Path) bool {
 	rp, _, err := bb.resolvePath(ctx, p)
 	if err != nil {
@@ -711,11 +634,10 @@ func (bb *BlocksBackend) ResolvePath(ctx context.Context, path path.ImmutablePat
 func (bb *BlocksBackend) resolvePath(ctx context.Context, p path.Path) (path.ImmutablePath, []string, error) {
 	var err error
 	if p.Namespace() == path.IPNSNamespace {
-		res, err := namesys.Resolve(ctx, bb.namesys, p)
+		p, _, _, err = bb.baseBackend.ResolveMutable(ctx, p)
 		if err != nil {
 			return path.ImmutablePath{}, nil, err
 		}
-		p = res.Path
 	}
 
 	if p.Namespace() != path.IPFSNamespace {
diff --git a/gateway/backend_car.go b/gateway/backend_car.go
new file mode 100644
index 000000000..d2b33a0fc
--- /dev/null
+++ b/gateway/backend_car.go
@@ -0,0 +1,1147 @@
+package gateway
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/hashicorp/go-multierror"
+	"github.com/ipfs/boxo/files"
+	"github.com/ipfs/boxo/ipld/merkledag"
+	"github.com/ipfs/boxo/ipld/unixfs"
+	"github.com/ipfs/boxo/path"
+	"github.com/ipfs/boxo/path/resolver"
+	blocks "github.com/ipfs/go-block-format"
+	"github.com/ipfs/go-cid"
+	format "github.com/ipfs/go-ipld-format"
+	"github.com/ipfs/go-unixfsnode"
+	ufsData "github.com/ipfs/go-unixfsnode/data"
+	carv2 "github.com/ipld/go-car/v2"
+	"github.com/ipld/go-car/v2/storage"
+	dagpb "github.com/ipld/go-codec-dagpb"
+	"github.com/ipld/go-ipld-prime"
+	"github.com/ipld/go-ipld-prime/datamodel"
+	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
+	"github.com/ipld/go-ipld-prime/node/basicnode"
+	"github.com/ipld/go-ipld-prime/schema"
+	"github.com/ipld/go-ipld-prime/traversal"
+	"github.com/multiformats/go-multicodec"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+var ErrFetcherUnexpectedEOF = fmt.Errorf("failed to fetch IPLD data")
+
+type CarBackend struct {
+	baseBackend
+	fetcher         CarFetcher
+	pc              traversal.LinkTargetNodePrototypeChooser
+	metrics         *CarBackendMetrics
+	getBlockTimeout time.Duration
+}
+
+type CarBackendMetrics struct {
+	contextAlreadyCancelledMetric prometheus.Counter
+	carFetchAttemptMetric         prometheus.Counter
+	carBlocksFetchedMetric        prometheus.Counter
+	carParamsMetric               *prometheus.CounterVec
+
+	bytesRangeStartMetric prometheus.Histogram
+	bytesRangeSizeMetric  prometheus.Histogram
+}
+
+// NewCarBackend returns an [IPFSBackend] backed by a [CarFetcher].
+func NewCarBackend(f CarFetcher, opts ...BackendOption) (*CarBackend, error) {
+	compiledOptions := backendOptions{
+		getBlockTimeout: DefaultGetBlockTimeout,
+	}
+	for _, o := range opts {
+		if err := o(&compiledOptions); err != nil {
+			return nil, err
+		}
+	}
+
+	// Setup the [baseBackend] which takes care of some shared functionality, such
+	// as resolving /ipns links.
+	baseBackend, err := newBaseBackend(compiledOptions.vs, compiledOptions.ns)
+	if err != nil {
+		return nil, err
+	}
+
+	var promReg prometheus.Registerer = prometheus.NewRegistry()
+	if compiledOptions.promRegistry != nil {
+		promReg = compiledOptions.promRegistry
+	}
+
+	return &CarBackend{
+		baseBackend:     baseBackend,
+		fetcher:         f,
+		metrics:         registerCarBackendMetrics(promReg),
+		getBlockTimeout: compiledOptions.getBlockTimeout,
+		pc: dagpb.AddSupportToChooser(func(lnk ipld.Link, lnkCtx ipld.LinkContext) (ipld.NodePrototype, error) {
+			if tlnkNd, ok := lnkCtx.LinkNode.(schema.TypedLinkNode); ok {
+				return tlnkNd.LinkTargetNodePrototype(), nil
+			}
+			return basicnode.Prototype.Any, nil
+		}),
+	}, nil
+}
+
+// NewRemoteCarBackend creates a new [CarBackend] instance backed by one or more
+// gateways. These gateways must support partial CAR requests, as described in
+// [IPIP-402], as well as IPNS Record requests. See [NewRemoteCarFetcher] and
+// [NewRemoteValueStore] for more details.
+//
+// If you want to create a more custom [CarBackend] with only remote IPNS Record
+// resolution, or only remote CAR fetching, we recommend using [NewCarBackend]
+// directly.
+//
+// [IPIP-402]: https://specs.ipfs.tech/ipips/ipip-0402/
+func NewRemoteCarBackend(gatewayURL []string, httpClient *http.Client, opts ...BackendOption) (*CarBackend, error) {
+	carFetcher, err := NewRemoteCarFetcher(gatewayURL, httpClient)
+	if err != nil {
+		return nil, err
+	}
+
+	valueStore, err := NewRemoteValueStore(gatewayURL, httpClient)
+	if err != nil {
+		return nil, err
+	}
+
+	return NewCarBackend(carFetcher, append(opts, WithValueStore(valueStore))...)
+}
+
+func registerCarBackendMetrics(promReg prometheus.Registerer) *CarBackendMetrics {
+	// How many CAR Fetch attempts we had? Need this to calculate % of various car request types.
+	// We only count attempts here, because success/failure with/without retries are provided by caboose:
+	// - ipfs_caboose_fetch_duration_car_success_count
+	// - ipfs_caboose_fetch_duration_car_failure_count
+	// - ipfs_caboose_fetch_duration_car_peer_success_count
+	// - ipfs_caboose_fetch_duration_car_peer_failure_count
+	carFetchAttemptMetric := prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace: "ipfs",
+		Subsystem: "gw_car_backend",
+		Name:      "car_fetch_attempts",
+		Help:      "The number of times a CAR fetch was attempted by IPFSBackend.",
+	})
+	promReg.MustRegister(carFetchAttemptMetric)
+
+	contextAlreadyCancelledMetric := prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace: "ipfs",
+		Subsystem: "gw_car_backend",
+		Name:      "car_fetch_context_already_cancelled",
+		Help:      "The number of times context is already cancelled when a CAR fetch was attempted by IPFSBackend.",
+	})
+	promReg.MustRegister(contextAlreadyCancelledMetric)
+
+	// How many blocks were read via CARs?
+	// Need this as a baseline to reason about error ratio vs raw_block_recovery_attempts.
+	carBlocksFetchedMetric := prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace: "ipfs",
+		Subsystem: "gw_car_backend",
+		Name:      "car_blocks_fetched",
+		Help:      "The number of blocks successfully read via CAR fetch.",
+	})
+	promReg.MustRegister(carBlocksFetchedMetric)
+
+	carParamsMetric := prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "ipfs",
+		Subsystem: "gw_car_backend",
+		Name:      "car_fetch_params",
+		Help:      "How many times specific CAR parameter was used during CAR data fetch.",
+	}, []string{"dagScope", "entityRanges"}) // we use 'ranges' instead of 'bytes' here because we only count the number of ranges present
+	promReg.MustRegister(carParamsMetric)
+
+	bytesRangeStartMetric := prometheus.NewHistogram(prometheus.HistogramOpts{
+		Namespace: "ipfs",
+		Subsystem: "gw_car_backend",
+		Name:      "range_request_start",
+		Help:      "Tracks where did the range request start.",
+		Buckets:   prometheus.ExponentialBuckets(1024, 2, 24), // 1024 bytes to 8 GiB
+	})
+	promReg.MustRegister(bytesRangeStartMetric)
+
+	bytesRangeSizeMetric := prometheus.NewHistogram(prometheus.HistogramOpts{
+		Namespace: "ipfs",
+		Subsystem: "gw_car_backend",
+		Name:      "range_request_size",
+		Help:      "Tracks the size of range requests.",
+		Buckets:   prometheus.ExponentialBuckets(256*1024, 2, 10), // From 256KiB to 100MiB
+	})
+	promReg.MustRegister(bytesRangeSizeMetric)
+
+	return &CarBackendMetrics{
+		contextAlreadyCancelledMetric,
+		carFetchAttemptMetric,
+		carBlocksFetchedMetric,
+		carParamsMetric,
+		bytesRangeStartMetric,
+		bytesRangeSizeMetric,
+	}
+}
+
+func (api *CarBackend) fetchCAR(ctx context.Context, p path.ImmutablePath, params CarParams, cb DataCallback) error {
+	api.metrics.carFetchAttemptMetric.Inc()
+	var ipldError error
+	fetchErr := api.fetcher.Fetch(ctx, p, params, func(p path.ImmutablePath, reader io.Reader) error {
+		return checkRetryableError(&ipldError, func() error {
+			return cb(p, reader)
+		})
+	})
+
+	if ipldError != nil {
+		fetchErr = ipldError
+	} else if fetchErr != nil {
+		fetchErr = blockstoreErrToGatewayErr(fetchErr)
+	}
+
+	return fetchErr
+}
+
+// resolvePathWithRootsAndBlock takes a path and linksystem and returns the set of non-terminal cids, the terminal cid, the remainder, and the block corresponding to the terminal cid
+func resolvePathWithRootsAndBlock(ctx context.Context, p path.ImmutablePath, unixFSLsys *ipld.LinkSystem) (ContentPathMetadata, blocks.Block, error) {
+	md, terminalBlk, err := resolvePathToLastWithRoots(ctx, p, unixFSLsys)
+	if err != nil {
+		return ContentPathMetadata{}, nil, err
+	}
+
+	terminalCid := md.LastSegment.RootCid()
+
+	if terminalBlk == nil {
+		lctx := ipld.LinkContext{Ctx: ctx}
+		lnk := cidlink.Link{Cid: terminalCid}
+		blockData, err := unixFSLsys.LoadRaw(lctx, lnk)
+		if err != nil {
+			return ContentPathMetadata{}, nil, err
+		}
+		terminalBlk, err = blocks.NewBlockWithCid(blockData, terminalCid)
+		if err != nil {
+			return ContentPathMetadata{}, nil, err
+		}
+	}
+
+	return md, terminalBlk, err
+}
+
+// resolvePathToLastWithRoots takes a path and linksystem and returns the set of non-terminal cids, the terminal cid,
+// the remainder pathing, the last block loaded, and the last node loaded.
+//
+// Note: the block returned will be nil if the terminal element is a link or the path is just a CID
+func resolvePathToLastWithRoots(ctx context.Context, p path.ImmutablePath, unixFSLsys *ipld.LinkSystem) (ContentPathMetadata, blocks.Block, error) {
+	root, segments := p.RootCid(), p.Segments()[2:]
+	if len(segments) == 0 {
+		return ContentPathMetadata{
+			PathSegmentRoots: []cid.Cid{},
+			LastSegment:      p,
+		}, nil, nil
+	}
+
+	unixFSLsys.NodeReifier = unixfsnode.Reify
+	defer func() { unixFSLsys.NodeReifier = nil }()
+
+	var cids []cid.Cid
+	cids = append(cids, root)
+
+	pc := dagpb.AddSupportToChooser(func(lnk ipld.Link, lnkCtx ipld.LinkContext) (ipld.NodePrototype, error) {
+		if tlnkNd, ok := lnkCtx.LinkNode.(schema.TypedLinkNode); ok {
+			return tlnkNd.LinkTargetNodePrototype(), nil
+		}
+		return basicnode.Prototype.Any, nil
+	})
+
+	loadNode := func(ctx context.Context, c cid.Cid) (blocks.Block, ipld.Node, error) {
+		lctx := ipld.LinkContext{Ctx: ctx}
+		rootLnk := cidlink.Link{Cid: c}
+		np, err := pc(rootLnk, lctx)
+		if err != nil {
+			return nil, nil, err
+		}
+		nd, blockData, err := unixFSLsys.LoadPlusRaw(lctx, rootLnk, np)
+		if err != nil {
+			return nil, nil, err
+		}
+		blk, err := blocks.NewBlockWithCid(blockData, c)
+		if err != nil {
+			return nil, nil, err
+		}
+		return blk, nd, nil
+	}
+
+	nextBlk, nextNd, err := loadNode(ctx, root)
+	if err != nil {
+		return ContentPathMetadata{}, nil, err
+	}
+
+	depth := 0
+	for i, elem := range segments {
+		nextNd, err = nextNd.LookupBySegment(ipld.ParsePathSegment(elem))
+		if err != nil {
+			return ContentPathMetadata{}, nil, err
+		}
+		if nextNd.Kind() == ipld.Kind_Link {
+			depth = 0
+			lnk, err := nextNd.AsLink()
+			if err != nil {
+				return ContentPathMetadata{}, nil, err
+			}
+			cidLnk, ok := lnk.(cidlink.Link)
+			if !ok {
+				return ContentPathMetadata{}, nil, fmt.Errorf("link is not a cidlink: %v", cidLnk)
+			}
+			cids = append(cids, cidLnk.Cid)
+
+			if i < len(segments)-1 {
+				nextBlk, nextNd, err = loadNode(ctx, cidLnk.Cid)
+				if err != nil {
+					return ContentPathMetadata{}, nil, err
+				}
+			}
+		} else {
+			depth++
+		}
+	}
+
+	// if last node is not a link, just return it's cid, add path to remainder and return
+	if nextNd.Kind() != ipld.Kind_Link {
+		md, err := contentMetadataFromRootsAndRemainder(cids, segments[len(segments)-depth:])
+		if err != nil {
+			return ContentPathMetadata{}, nil, err
+		}
+
+		// return the cid and the remainder of the path
+		return md, nextBlk, nil
+	}
+
+	md, err := contentMetadataFromRootsAndRemainder(cids, nil)
+	return md, nil, err
+}
+
+func contentMetadataFromRootsAndRemainder(roots []cid.Cid, remainder []string) (ContentPathMetadata, error) {
+	if len(roots) == 0 {
+		return ContentPathMetadata{}, errors.New("invalid pathRoots given with length 0")
+	}
+
+	p, err := path.Join(path.FromCid(roots[len(roots)-1]), remainder...)
+	if err != nil {
+		return ContentPathMetadata{}, err
+	}
+
+	imPath, err := path.NewImmutablePath(p)
+	if err != nil {
+		return ContentPathMetadata{}, err
+	}
+
+	md := ContentPathMetadata{
+		PathSegmentRoots:     roots[:len(roots)-1],
+		LastSegmentRemainder: remainder,
+		LastSegment:          imPath,
+	}
+	return md, nil
+}
+
+var errNotUnixFS = fmt.Errorf("data was not unixfs")
+
+func (api *CarBackend) Get(ctx context.Context, path path.ImmutablePath, byteRanges ...ByteRange) (ContentPathMetadata, *GetResponse, error) {
+	rangeCount := len(byteRanges)
+	api.metrics.carParamsMetric.With(prometheus.Labels{"dagScope": "entity", "entityRanges": strconv.Itoa(rangeCount)}).Inc()
+
+	carParams := CarParams{Scope: DagScopeEntity}
+
+	// fetch CAR with &bytes= to get minimal set of blocks for the request
+	// Note: majority of requests have 0 or max 1 ranges. if there are more ranges than one,
+	// that is a niche edge cache we don't prefetch as CAR and use fallback blockstore instead.
+	if rangeCount > 0 {
+		r := byteRanges[0]
+		carParams.Range = &DagByteRange{
+			From: int64(r.From),
+		}
+
+		// TODO: move to boxo or to loadRequestIntoSharedBlockstoreAndBlocksGateway after we pass params in a humane way
+		api.metrics.bytesRangeStartMetric.Observe(float64(r.From))
+
+		if r.To != nil {
+			carParams.Range.To = r.To
+
+			// TODO: move to boxo or to loadRequestIntoSharedBlockstoreAndBlocksGateway after we pass params in a humane way
+			api.metrics.bytesRangeSizeMetric.Observe(float64(*r.To) - float64(r.From) + 1)
+		}
+	}
+
+	md, terminalElem, err := fetchWithPartialRetries(ctx, path, carParams, loadTerminalEntity, api.metrics, api.fetchCAR, api.getBlockTimeout)
+	if err != nil {
+		return ContentPathMetadata{}, nil, err
+	}
+
+	var resp *GetResponse
+
+	switch typedTerminalElem := terminalElem.(type) {
+	case *GetResponse:
+		resp = typedTerminalElem
+	case *backpressuredFile:
+		resp = NewGetResponseFromReader(typedTerminalElem, typedTerminalElem.size)
+	case *backpressuredHAMTDirIterNoRecursion:
+		ch := make(chan unixfs.LinkResult)
+		go func() {
+			defer close(ch)
+			for typedTerminalElem.Next() {
+				l := typedTerminalElem.Link()
+				select {
+				case ch <- l:
+				case <-ctx.Done():
+					return
+				}
+			}
+			if err := typedTerminalElem.Err(); err != nil {
+				select {
+				case ch <- unixfs.LinkResult{Err: err}:
+				case <-ctx.Done():
+					return
+				}
+			}
+		}()
+		resp = NewGetResponseFromDirectoryListing(typedTerminalElem.dagSize, ch, nil)
+	default:
+		return ContentPathMetadata{}, nil, fmt.Errorf("invalid data type")
+	}
+
+	return md, resp, nil
+}
+
+// loadTerminalEntity returns either a [*GetResponse], [*backpressuredFile], or [*backpressuredHAMTDirIterNoRecursion]
+func loadTerminalEntity(ctx context.Context, c cid.Cid, blk blocks.Block, lsys *ipld.LinkSystem, params CarParams, getLsys lsysGetter) (interface{}, error) {
+	var err error
+	if lsys == nil {
+		lsys, err = getLsys(ctx, c, params)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	lctx := ipld.LinkContext{Ctx: ctx}
+
+	if c.Type() != uint64(multicodec.DagPb) {
+		var blockData []byte
+
+		if blk != nil {
+			blockData = blk.RawData()
+		} else {
+			blockData, err = lsys.LoadRaw(lctx, cidlink.Link{Cid: c})
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		f := files.NewBytesFile(blockData)
+		if params.Range != nil && params.Range.From != 0 {
+			if _, err := f.Seek(params.Range.From, io.SeekStart); err != nil {
+				return nil, err
+			}
+		}
+
+		return NewGetResponseFromReader(f, int64(len(blockData))), nil
+	}
+
+	blockData, pbn, ufsFieldData, fieldNum, err := loadUnixFSBase(ctx, c, blk, lsys)
+	if err != nil {
+		return nil, err
+	}
+
+	switch fieldNum {
+	case ufsData.Data_Symlink:
+		if !ufsFieldData.FieldData().Exists() {
+			return nil, fmt.Errorf("invalid UnixFS symlink object")
+		}
+		lnkTarget := string(ufsFieldData.FieldData().Must().Bytes())
+		f := NewGetResponseFromSymlink(files.NewLinkFile(lnkTarget, nil).(*files.Symlink), int64(len(lnkTarget)))
+		return f, nil
+	case ufsData.Data_Metadata:
+		return nil, fmt.Errorf("UnixFS Metadata unsupported")
+	case ufsData.Data_HAMTShard, ufsData.Data_Directory:
+		blk, err := blocks.NewBlockWithCid(blockData, c)
+		if err != nil {
+			return nil, fmt.Errorf("could not create block: %w", err)
+		}
+		dirRootNd, err := merkledag.ProtoNodeConverter(blk, pbn)
+		if err != nil {
+			return nil, fmt.Errorf("could not create dag-pb universal block from UnixFS directory root: %w", err)
+		}
+		pn, ok := dirRootNd.(*merkledag.ProtoNode)
+		if !ok {
+			return nil, fmt.Errorf("could not create dag-pb node from UnixFS directory root: %w", err)
+		}
+
+		dirDagSize, err := pn.Size()
+		if err != nil {
+			return nil, fmt.Errorf("could not get cumulative size from dag-pb node: %w", err)
+		}
+
+		switch fieldNum {
+		case ufsData.Data_Directory:
+			ch := make(chan unixfs.LinkResult, pbn.Links.Length())
+			defer close(ch)
+			iter := pbn.Links.Iterator()
+			for !iter.Done() {
+				_, v := iter.Next()
+				c := v.Hash.Link().(cidlink.Link).Cid
+				var name string
+				var size int64
+				if v.Name.Exists() {
+					name = v.Name.Must().String()
+				}
+				if v.Tsize.Exists() {
+					size = v.Tsize.Must().Int()
+				}
+				lnk := unixfs.LinkResult{Link: &format.Link{
+					Name: name,
+					Size: uint64(size),
+					Cid:  c,
+				}}
+				ch <- lnk
+			}
+			return NewGetResponseFromDirectoryListing(dirDagSize, ch, nil), nil
+		case ufsData.Data_HAMTShard:
+			dirNd, err := unixfsnode.Reify(lctx, pbn, lsys)
+			if err != nil {
+				return nil, fmt.Errorf("could not reify sharded directory: %w", err)
+			}
+
+			d := &backpressuredHAMTDirIterNoRecursion{
+				dagSize:   dirDagSize,
+				linksItr:  dirNd.MapIterator(),
+				dirCid:    c,
+				lsys:      lsys,
+				getLsys:   getLsys,
+				ctx:       ctx,
+				closed:    make(chan error),
+				hasClosed: false,
+			}
+			return d, nil
+		default:
+			return nil, fmt.Errorf("not a basic or HAMT directory: should be unreachable")
+		}
+	case ufsData.Data_Raw, ufsData.Data_File:
+		nd, err := unixfsnode.Reify(lctx, pbn, lsys)
+		if err != nil {
+			return nil, err
+		}
+
+		fnd, ok := nd.(datamodel.LargeBytesNode)
+		if !ok {
+			return nil, fmt.Errorf("could not process file since it did not present as large bytes")
+		}
+		f, err := fnd.AsLargeBytes()
+		if err != nil {
+			return nil, err
+		}
+
+		fileSize, err := f.Seek(0, io.SeekEnd)
+		if err != nil {
+			return nil, fmt.Errorf("unable to get UnixFS file size: %w", err)
+		}
+
+		from := int64(0)
+		var byteRange DagByteRange
+		if params.Range != nil {
+			from = params.Range.From
+			byteRange = *params.Range
+		}
+		_, err = f.Seek(from, io.SeekStart)
+		if err != nil {
+			return nil, fmt.Errorf("unable to get reset UnixFS file reader: %w", err)
+		}
+
+		return &backpressuredFile{ctx: ctx, fileCid: c, byteRange: byteRange, size: fileSize, f: f, getLsys: getLsys, closed: make(chan error)}, nil
+	default:
+		return nil, fmt.Errorf("unknown UnixFS field type")
+	}
+}
+
+func (api *CarBackend) GetAll(ctx context.Context, path path.ImmutablePath) (ContentPathMetadata, files.Node, error) {
+	api.metrics.carParamsMetric.With(prometheus.Labels{"dagScope": "all", "entityRanges": "0"}).Inc()
+	return fetchWithPartialRetries(ctx, path, CarParams{Scope: DagScopeAll}, loadTerminalUnixFSElementWithRecursiveDirectories, api.metrics, api.fetchCAR, api.getBlockTimeout)
+}
+
+type loadTerminalElement[T any] func(ctx context.Context, c cid.Cid, blk blocks.Block, lsys *ipld.LinkSystem, params CarParams, getLsys lsysGetter) (T, error)
+type fetchCarFn = func(ctx context.Context, path path.ImmutablePath, params CarParams, cb DataCallback) error
+
+type terminalPathType[T any] struct {
+	resp T
+	err  error
+	md   ContentPathMetadata
+}
+
+type nextReq struct {
+	c      cid.Cid
+	params CarParams
+}
+
+func fetchWithPartialRetries[T any](ctx context.Context, p path.ImmutablePath, initialParams CarParams, resolveTerminalElementFn loadTerminalElement[T], metrics *CarBackendMetrics, fetchCAR fetchCarFn, timeout time.Duration) (ContentPathMetadata, T, error) {
+	var zeroReturnType T
+
+	terminalPathElementCh := make(chan terminalPathType[T], 1)
+
+	go func() {
+		cctx, cancel := context.WithCancel(ctx)
+		defer cancel()
+
+		hasSentAsyncData := false
+		var closeCh <-chan error
+
+		sendRequest := make(chan nextReq, 1)
+		sendResponse := make(chan *ipld.LinkSystem, 1)
+		getLsys := func(ctx context.Context, c cid.Cid, params CarParams) (*ipld.LinkSystem, error) {
+			select {
+			case sendRequest <- nextReq{c: c, params: params}:
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			}
+
+			select {
+			case lsys := <-sendResponse:
+				return lsys, nil
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			}
+		}
+
+		params := initialParams
+
+		err := fetchCAR(cctx, p, params, func(_ path.ImmutablePath, reader io.Reader) error {
+			gb, err := carToLinearBlockGetter(cctx, reader, timeout, metrics)
+			if err != nil {
+				return err
+			}
+
+			lsys := getCarLinksystem(gb)
+
+			if hasSentAsyncData {
+				_, _, err = resolvePathToLastWithRoots(cctx, p, lsys)
+				if err != nil {
+					return err
+				}
+
+				select {
+				case sendResponse <- lsys:
+				case <-cctx.Done():
+					return cctx.Err()
+				}
+			} else {
+				// First resolve the path since we always need to.
+				md, terminalBlk, err := resolvePathWithRootsAndBlock(cctx, p, lsys)
+				if err != nil {
+					return err
+				}
+
+				if len(md.LastSegmentRemainder) > 0 {
+					terminalPathElementCh <- terminalPathType[T]{err: errNotUnixFS}
+					return nil
+				}
+
+				if hasSentAsyncData {
+					select {
+					case sendResponse <- lsys:
+					case <-ctx.Done():
+						return ctx.Err()
+					}
+				}
+
+				terminalCid := md.LastSegment.RootCid()
+
+				nd, err := resolveTerminalElementFn(cctx, terminalCid, terminalBlk, lsys, params, getLsys)
+				if err != nil {
+					return err
+				}
+
+				ndAc, ok := any(nd).(awaitCloser)
+				if !ok {
+					terminalPathElementCh <- terminalPathType[T]{
+						resp: nd,
+						md:   md,
+					}
+					return nil
+				}
+
+				hasSentAsyncData = true
+				terminalPathElementCh <- terminalPathType[T]{
+					resp: nd,
+					md:   md,
+				}
+
+				closeCh = ndAc.AwaitClose()
+			}
+
+			select {
+			case closeErr := <-closeCh:
+				return closeErr
+			case req := <-sendRequest:
+				// set path and params for next iteration
+				p = path.FromCid(req.c)
+				if err != nil {
+					return err
+				}
+				params = req.params
+				return ErrPartialResponse{StillNeed: []CarResource{{Path: p, Params: params}}}
+			case <-cctx.Done():
+				return cctx.Err()
+			}
+		})
+
+		if !hasSentAsyncData && err != nil {
+			terminalPathElementCh <- terminalPathType[T]{err: err}
+			return
+		}
+
+		if err != nil {
+			lsys := getCarLinksystem(func(ctx context.Context, cid cid.Cid) (blocks.Block, error) {
+				return nil, multierror.Append(ErrFetcherUnexpectedEOF, format.ErrNotFound{Cid: cid})
+			})
+			for {
+				select {
+				case <-closeCh:
+					return
+				case <-sendRequest:
+				case sendResponse <- lsys:
+				case <-cctx.Done():
+					return
+				}
+			}
+		}
+	}()
+
+	select {
+	case t := <-terminalPathElementCh:
+		if t.err != nil {
+			return ContentPathMetadata{}, zeroReturnType, t.err
+		}
+		return t.md, t.resp, nil
+	case <-ctx.Done():
+		return ContentPathMetadata{}, zeroReturnType, ctx.Err()
+	}
+}
+
+func (api *CarBackend) GetBlock(ctx context.Context, p path.ImmutablePath) (ContentPathMetadata, files.File, error) {
+	api.metrics.carParamsMetric.With(prometheus.Labels{"dagScope": "block", "entityRanges": "0"}).Inc()
+
+	var md ContentPathMetadata
+	var f files.File
+	// TODO: if path is `/ipfs/cid`, we should use ?format=raw
+	err := api.fetchCAR(ctx, p, CarParams{Scope: DagScopeBlock}, func(_ path.ImmutablePath, reader io.Reader) error {
+		gb, err := carToLinearBlockGetter(ctx, reader, api.getBlockTimeout, api.metrics)
+		if err != nil {
+			return err
+		}
+		lsys := getCarLinksystem(gb)
+
+		// First resolve the path since we always need to.
+		var terminalBlk blocks.Block
+		md, terminalBlk, err = resolvePathToLastWithRoots(ctx, p, lsys)
+		if err != nil {
+			return err
+		}
+
+		var blockData []byte
+		if terminalBlk != nil {
+			blockData = terminalBlk.RawData()
+		} else {
+			lctx := ipld.LinkContext{Ctx: ctx}
+			lnk := cidlink.Link{Cid: md.LastSegment.RootCid()}
+			blockData, err = lsys.LoadRaw(lctx, lnk)
+			if err != nil {
+				return err
+			}
+		}
+
+		f = files.NewBytesFile(blockData)
+		return nil
+	})
+
+	if err != nil {
+		return ContentPathMetadata{}, nil, err
+	}
+
+	return md, f, nil
+}
+
+func (api *CarBackend) Head(ctx context.Context, p path.ImmutablePath) (ContentPathMetadata, *HeadResponse, error) {
+	api.metrics.carParamsMetric.With(prometheus.Labels{"dagScope": "entity", "entityRanges": "1"}).Inc()
+
+	// TODO:  we probably want to move this either to boxo, or at least to loadRequestIntoSharedBlockstoreAndBlocksGateway
+	api.metrics.bytesRangeStartMetric.Observe(0)
+	api.metrics.bytesRangeSizeMetric.Observe(3071)
+
+	var md ContentPathMetadata
+	var n *HeadResponse
+	// TODO: fallback to dynamic fetches in case we haven't requested enough data
+	rangeTo := int64(3071)
+	err := api.fetchCAR(ctx, p, CarParams{Scope: DagScopeEntity, Range: &DagByteRange{From: 0, To: &rangeTo}}, func(_ path.ImmutablePath, reader io.Reader) error {
+		gb, err := carToLinearBlockGetter(ctx, reader, api.getBlockTimeout, api.metrics)
+		if err != nil {
+			return err
+		}
+		lsys := getCarLinksystem(gb)
+
+		// First resolve the path since we always need to.
+		var terminalBlk blocks.Block
+		md, terminalBlk, err = resolvePathWithRootsAndBlock(ctx, p, lsys)
+		if err != nil {
+			return err
+		}
+
+		terminalCid := md.LastSegment.RootCid()
+		lctx := ipld.LinkContext{Ctx: ctx}
+		pathTerminalCidLink := cidlink.Link{Cid: terminalCid}
+
+		// Load the block at the root of the terminal path element
+		dataBytes := terminalBlk.RawData()
+
+		// It's not UnixFS if there is a remainder or it's not dag-pb
+		if len(md.LastSegmentRemainder) > 0 || terminalCid.Type() != uint64(multicodec.DagPb) {
+			n = NewHeadResponseForFile(files.NewBytesFile(dataBytes), int64(len(dataBytes)))
+			return nil
+		}
+
+		// Let's figure out if the terminal element is valid UnixFS and if so what kind
+		np, err := api.pc(pathTerminalCidLink, lctx)
+		if err != nil {
+			return err
+		}
+
+		nodeDecoder, err := lsys.DecoderChooser(pathTerminalCidLink)
+		if err != nil {
+			return err
+		}
+
+		nb := np.NewBuilder()
+		err = nodeDecoder(nb, bytes.NewReader(dataBytes))
+		if err != nil {
+			return err
+		}
+		lastCidNode := nb.Build()
+
+		if pbn, ok := lastCidNode.(dagpb.PBNode); !ok {
+			// This shouldn't be possible since we already checked for dag-pb usage
+			return fmt.Errorf("node was not go-codec-dagpb node")
+		} else if !pbn.FieldData().Exists() {
+			// If it's not valid UnixFS then just return the block bytes
+			n = NewHeadResponseForFile(files.NewBytesFile(dataBytes), int64(len(dataBytes)))
+			return nil
+		} else if unixfsFieldData, decodeErr := ufsData.DecodeUnixFSData(pbn.Data.Must().Bytes()); decodeErr != nil {
+			// If it's not valid UnixFS then just return the block bytes
+			n = NewHeadResponseForFile(files.NewBytesFile(dataBytes), int64(len(dataBytes)))
+			return nil
+		} else {
+			switch fieldNum := unixfsFieldData.FieldDataType().Int(); fieldNum {
+			case ufsData.Data_Directory, ufsData.Data_HAMTShard:
+				dirRootNd, err := merkledag.ProtoNodeConverter(terminalBlk, lastCidNode)
+				if err != nil {
+					return fmt.Errorf("could not create dag-pb universal block from UnixFS directory root: %w", err)
+				}
+				pn, ok := dirRootNd.(*merkledag.ProtoNode)
+				if !ok {
+					return fmt.Errorf("could not create dag-pb node from UnixFS directory root: %w", err)
+				}
+
+				sz, err := pn.Size()
+				if err != nil {
+					return fmt.Errorf("could not get cumulative size from dag-pb node: %w", err)
+				}
+
+				n = NewHeadResponseForDirectory(int64(sz))
+				return nil
+			case ufsData.Data_Symlink:
+				fd := unixfsFieldData.FieldData()
+				if fd.Exists() {
+					n = NewHeadResponseForSymlink(int64(len(fd.Must().Bytes())))
+					return nil
+				}
+				// If there is no target then it's invalid so just return the block
+				NewHeadResponseForFile(files.NewBytesFile(dataBytes), int64(len(dataBytes)))
+				return nil
+			case ufsData.Data_Metadata:
+				n = NewHeadResponseForFile(files.NewBytesFile(dataBytes), int64(len(dataBytes)))
+				return nil
+			case ufsData.Data_Raw, ufsData.Data_File:
+				ufsNode, err := unixfsnode.Reify(lctx, pbn, lsys)
+				if err != nil {
+					return err
+				}
+				fileNode, ok := ufsNode.(datamodel.LargeBytesNode)
+				if !ok {
+					return fmt.Errorf("data not a large bytes node despite being UnixFS bytes")
+				}
+				f, err := fileNode.AsLargeBytes()
+				if err != nil {
+					return err
+				}
+
+				fileSize, err := f.Seek(0, io.SeekEnd)
+				if err != nil {
+					return fmt.Errorf("unable to get UnixFS file size: %w", err)
+				}
+				_, err = f.Seek(0, io.SeekStart)
+				if err != nil {
+					return fmt.Errorf("unable to get reset UnixFS file reader: %w", err)
+				}
+
+				out, err := io.ReadAll(io.LimitReader(f, 3072))
+				if errors.Is(err, io.EOF) {
+					n = NewHeadResponseForFile(files.NewBytesFile(out), fileSize)
+					return nil
+				}
+				return err
+			}
+		}
+		return nil
+	})
+
+	if err != nil {
+		return ContentPathMetadata{}, nil, err
+	}
+
+	return md, n, nil
+}
+
+func (api *CarBackend) ResolvePath(ctx context.Context, p path.ImmutablePath) (ContentPathMetadata, error) {
+	api.metrics.carParamsMetric.With(prometheus.Labels{"dagScope": "block", "entityRanges": "0"}).Inc()
+
+	var md ContentPathMetadata
+	err := api.fetchCAR(ctx, p, CarParams{Scope: DagScopeBlock}, func(_ path.ImmutablePath, reader io.Reader) error {
+		gb, err := carToLinearBlockGetter(ctx, reader, api.getBlockTimeout, api.metrics)
+		if err != nil {
+			return err
+		}
+		lsys := getCarLinksystem(gb)
+
+		// First resolve the path since we always need to.
+		md, _, err = resolvePathToLastWithRoots(ctx, p, lsys)
+		if err != nil {
+			return err
+		}
+
+		return err
+	})
+
+	if err != nil {
+		return ContentPathMetadata{}, err
+	}
+
+	return md, nil
+}
+
+func (api *CarBackend) GetCAR(ctx context.Context, p path.ImmutablePath, params CarParams) (ContentPathMetadata, io.ReadCloser, error) {
+	numRanges := "0"
+	if params.Range != nil {
+		numRanges = "1"
+	}
+	api.metrics.carParamsMetric.With(prometheus.Labels{"dagScope": string(params.Scope), "entityRanges": numRanges}).Inc()
+	rootCid, err := getRootCid(p)
+	if err != nil {
+		return ContentPathMetadata{}, nil, err
+	}
+
+	switch params.Order {
+	case DagOrderUnspecified, DagOrderUnknown, DagOrderDFS:
+	default:
+		return ContentPathMetadata{}, nil, fmt.Errorf("unsupported dag order %q", params.Order)
+	}
+
+	r, w := io.Pipe()
+	go func() {
+		numBlocksSent := 0
+		var cw storage.WritableCar
+		var blockBuffer []blocks.Block
+		err = api.fetchCAR(ctx, p, params, func(_ path.ImmutablePath, reader io.Reader) error {
+			numBlocksThisCall := 0
+			gb, err := carToLinearBlockGetter(ctx, reader, api.getBlockTimeout, api.metrics)
+			if err != nil {
+				return err
+			}
+			teeBlock := func(ctx context.Context, c cid.Cid) (blocks.Block, error) {
+				blk, err := gb(ctx, c)
+				if err != nil {
+					return nil, err
+				}
+				if numBlocksThisCall >= numBlocksSent {
+					if cw == nil {
+						blockBuffer = append(blockBuffer, blk)
+					} else {
+						err = cw.Put(ctx, blk.Cid().KeyString(), blk.RawData())
+						if err != nil {
+							return nil, fmt.Errorf("error writing car block: %w", err)
+						}
+					}
+					numBlocksSent++
+				}
+				numBlocksThisCall++
+				return blk, nil
+			}
+			l := getCarLinksystem(teeBlock)
+
+			var isNotFound bool
+
+			// First resolve the path since we always need to.
+			md, terminalBlk, err := resolvePathWithRootsAndBlock(ctx, p, l)
+			if err != nil {
+				if isErrNotFound(err) {
+					isNotFound = true
+				} else {
+					return err
+				}
+			}
+
+			if len(md.LastSegmentRemainder) > 0 {
+				return nil
+			}
+
+			if cw == nil {
+				var roots []cid.Cid
+				if isNotFound {
+					roots = emptyRoot
+				} else {
+					roots = []cid.Cid{md.LastSegment.RootCid()}
+				}
+
+				cw, err = storage.NewWritable(w, roots, carv2.WriteAsCarV1(true), carv2.AllowDuplicatePuts(params.Duplicates.Bool()))
+				if err != nil {
+					// io.PipeWriter.CloseWithError always returns nil.
+					_ = w.CloseWithError(err)
+					return nil
+				}
+				for _, blk := range blockBuffer {
+					err = cw.Put(ctx, blk.Cid().KeyString(), blk.RawData())
+					if err != nil {
+						_ = w.CloseWithError(fmt.Errorf("error writing car block: %w", err))
+						return nil
+					}
+				}
+				blockBuffer = nil
+			}
+
+			if !isNotFound {
+				params.Duplicates = DuplicateBlocksIncluded
+				err = walkGatewaySimpleSelector(ctx, terminalBlk.Cid(), terminalBlk, []string{}, params, l)
+				if err != nil {
+					return err
+				}
+			}
+
+			return nil
+		})
+
+		_ = w.CloseWithError(err)
+	}()
+
+	return ContentPathMetadata{
+		PathSegmentRoots: []cid.Cid{rootCid},
+		LastSegment:      path.FromCid(rootCid),
+		ContentType:      "",
+	}, r, nil
+}
+
+func getRootCid(imPath path.ImmutablePath) (cid.Cid, error) {
+	imPathStr := imPath.String()
+	if !strings.HasPrefix(imPathStr, "/ipfs/") {
+		return cid.Undef, fmt.Errorf("path does not have /ipfs/ prefix")
+	}
+
+	firstSegment, _, _ := strings.Cut(imPathStr[6:], "/")
+	rootCid, err := cid.Decode(firstSegment)
+	if err != nil {
+		return cid.Undef, err
+	}
+
+	return rootCid, nil
+}
+
+func (api *CarBackend) IsCached(ctx context.Context, path path.Path) bool {
+	return false
+}
+
+var _ IPFSBackend = (*CarBackend)(nil)
+
+func checkRetryableError(e *error, fn func() error) error {
+	err := fn()
+	retry, processedErr := isRetryableError(err)
+	if retry {
+		return processedErr
+	}
+	*e = processedErr
+	return nil
+}
+
+func isRetryableError(err error) (bool, error) {
+	if errors.Is(err, ErrFetcherUnexpectedEOF) {
+		return false, err
+	}
+
+	if format.IsNotFound(err) {
+		return true, err
+	}
+	initialErr := err
+
+	// Checks if err is of a type that does not implement the .Is interface and
+	// cannot be directly compared to. Therefore, errors.Is cannot be used.
+	for {
+		_, ok := err.(*resolver.ErrNoLink)
+		if ok {
+			return false, err
+		}
+
+		_, ok = err.(datamodel.ErrWrongKind)
+		if ok {
+			return false, err
+		}
+
+		_, ok = err.(datamodel.ErrNotExists)
+		if ok {
+			return false, err
+		}
+
+		errNoSuchField, ok := err.(schema.ErrNoSuchField)
+		if ok {
+			// Convert into a more general error type so the gateway code can know what this means
+			// TODO: Have either a more generally usable error type system for IPLD errors (e.g. a base type indicating that data cannot exist)
+			// or at least have one that is specific to the gateway consumer and part of the Backend contract instead of this being implicit
+			err = datamodel.ErrNotExists{Segment: errNoSuchField.Field}
+			return false, err
+		}
+
+		err = errors.Unwrap(err)
+		if err == nil {
+			return true, initialErr
+		}
+	}
+}
+
+// blockstoreErrToGatewayErr translates underlying blockstore error into one that gateway code will return as HTTP 502 or 504
+// it also makes sure Retry-After hint from remote blockstore will be passed to HTTP client, if present.
+func blockstoreErrToGatewayErr(err error) error {
+	if errors.Is(err, &ErrorStatusCode{}) ||
+		errors.Is(err, &ErrorRetryAfter{}) {
+		// already correct error
+		return err
+	}
+
+	// All timeouts should produce 504 Gateway Timeout
+	if errors.Is(err, context.DeadlineExceeded) ||
+		// Unfortunately this is not an exported type so we have to check for the content.
+		strings.Contains(err.Error(), "Client.Timeout exceeded") {
+		return fmt.Errorf("%w: %s", ErrGatewayTimeout, err.Error())
+	}
+
+	// (Saturn) errors that support the RetryAfter interface need to be converted
+	// to the correct gateway error, such that the HTTP header is set.
+	for v := err; v != nil; v = errors.Unwrap(v) {
+		if r, ok := v.(interface{ RetryAfter() time.Duration }); ok {
+			return NewErrorRetryAfter(err, r.RetryAfter())
+		}
+	}
+
+	// everything else returns 502 Bad Gateway
+	return fmt.Errorf("%w: %s", ErrBadGateway, err.Error())
+}
diff --git a/gateway/backend_car_fetcher.go b/gateway/backend_car_fetcher.go
new file mode 100644
index 000000000..cf9d2ec04
--- /dev/null
+++ b/gateway/backend_car_fetcher.go
@@ -0,0 +1,169 @@
+package gateway
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"math/rand"
+	"net/http"
+	"net/url"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/ipfs/boxo/path"
+)
+
+type DataCallback func(p path.ImmutablePath, reader io.Reader) error
+
+// CarFetcher powers a [CarBackend].
+type CarFetcher interface {
+	Fetch(ctx context.Context, path path.ImmutablePath, params CarParams, cb DataCallback) error
+}
+
+type remoteCarFetcher struct {
+	httpClient *http.Client
+	gatewayURL []string
+	rand       *rand.Rand
+}
+
+// NewRemoteCarFetcher returns a [CarFetcher] that is backed by one or more gateways
+// that support partial [CAR requests], as described in [IPIP-402]. You can optionally
+// pass your own [http.Client].
+//
+// [CAR requests]: https://www.iana.org/assignments/media-types/application/vnd.ipld.car
+// [IPIP-402]: https://specs.ipfs.tech/ipips/ipip-0402
+func NewRemoteCarFetcher(gatewayURL []string, httpClient *http.Client) (CarFetcher, error) {
+	if len(gatewayURL) == 0 {
+		return nil, errors.New("missing gateway URLs to which to proxy")
+	}
+
+	if httpClient == nil {
+		httpClient = newRemoteHTTPClient()
+	}
+
+	return &remoteCarFetcher{
+		gatewayURL: gatewayURL,
+		httpClient: httpClient,
+		rand:       rand.New(rand.NewSource(time.Now().Unix())),
+	}, nil
+}
+
+func (ps *remoteCarFetcher) Fetch(ctx context.Context, path path.ImmutablePath, params CarParams, cb DataCallback) error {
+	url := contentPathToCarUrl(path, params)
+
+	urlStr := fmt.Sprintf("%s%s", ps.getRandomGatewayURL(), url.String())
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, urlStr, nil)
+	if err != nil {
+		return err
+	}
+	log.Debugw("car fetch", "url", req.URL)
+	req.Header.Set("Accept", "application/vnd.ipld.car;order=dfs;dups=y")
+	resp, err := ps.httpClient.Do(req)
+	if err != nil {
+		return err
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		errData, err := io.ReadAll(resp.Body)
+		if err != nil {
+			err = fmt.Errorf("could not read error message: %w", err)
+		} else {
+			err = fmt.Errorf("%q", string(errData))
+		}
+		return fmt.Errorf("http error from car gateway: %s: %w", resp.Status, err)
+	}
+
+	err = cb(path, resp.Body)
+	if err != nil {
+		resp.Body.Close()
+		return err
+	}
+	return resp.Body.Close()
+}
+
+func (ps *remoteCarFetcher) getRandomGatewayURL() string {
+	return ps.gatewayURL[ps.rand.Intn(len(ps.gatewayURL))]
+}
+
+// contentPathToCarUrl returns an URL that allows retrieval of specified resource
+// from a trustless gateway that implements IPIP-402
+func contentPathToCarUrl(path path.ImmutablePath, params CarParams) *url.URL {
+	return &url.URL{
+		Path:     path.String(),
+		RawQuery: carParamsToString(params),
+	}
+}
+
+// carParamsToString converts CarParams to URL parameters compatible with IPIP-402
+func carParamsToString(params CarParams) string {
+	paramsBuilder := strings.Builder{}
+	paramsBuilder.WriteString("format=car") // always send explicit format in URL, this  makes debugging easier, even when Accept header was set
+	if params.Scope != "" {
+		paramsBuilder.WriteString("&dag-scope=")
+		paramsBuilder.WriteString(string(params.Scope))
+	}
+	if params.Range != nil {
+		paramsBuilder.WriteString("&entity-bytes=")
+		paramsBuilder.WriteString(strconv.FormatInt(params.Range.From, 10))
+		paramsBuilder.WriteString(":")
+		if params.Range.To != nil {
+			paramsBuilder.WriteString(strconv.FormatInt(*params.Range.To, 10))
+		} else {
+			paramsBuilder.WriteString("*")
+		}
+	}
+	return paramsBuilder.String()
+}
+
+type retryCarFetcher struct {
+	inner   CarFetcher
+	retries int
+}
+
+// NewRetryCarFetcher returns a [CarFetcher] that retries to fetch up to the given
+// [allowedRetries] using the [inner] [CarFetcher]. If the inner fetcher returns
+// an [ErrPartialResponse] error, then the number of retries is reset to the initial
+// maximum allowed retries.
+func NewRetryCarFetcher(inner CarFetcher, allowedRetries int) (CarFetcher, error) {
+	if allowedRetries <= 0 {
+		return nil, errors.New("number of retries must be a number larger than 0")
+	}
+
+	return &retryCarFetcher{
+		inner:   inner,
+		retries: allowedRetries,
+	}, nil
+}
+
+func (r *retryCarFetcher) Fetch(ctx context.Context, path path.ImmutablePath, params CarParams, cb DataCallback) error {
+	return r.fetch(ctx, path, params, cb, r.retries)
+}
+
+func (r *retryCarFetcher) fetch(ctx context.Context, path path.ImmutablePath, params CarParams, cb DataCallback, retriesLeft int) error {
+	err := r.inner.Fetch(ctx, path, params, cb)
+	if err == nil {
+		return nil
+	}
+
+	if retriesLeft > 0 {
+		retriesLeft--
+	} else {
+		return fmt.Errorf("retry fetcher out of retries: %w", err)
+	}
+
+	switch t := err.(type) {
+	case ErrPartialResponse:
+		if len(t.StillNeed) > 1 {
+			return errors.New("only a single request at a time is supported")
+		}
+
+		// Resets the number of retries for partials, mimicking Caboose logic.
+		retriesLeft = r.retries
+
+		return r.fetch(ctx, t.StillNeed[0].Path, t.StillNeed[0].Params, cb, retriesLeft)
+	default:
+		return r.fetch(ctx, path, params, cb, retriesLeft)
+	}
+}
diff --git a/gateway/backend_car_files.go b/gateway/backend_car_files.go
new file mode 100644
index 000000000..c384bbe2c
--- /dev/null
+++ b/gateway/backend_car_files.go
@@ -0,0 +1,697 @@
+package gateway
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+
+	"github.com/ipfs/boxo/files"
+	"github.com/ipfs/boxo/ipld/unixfs"
+	blocks "github.com/ipfs/go-block-format"
+	"github.com/ipfs/go-cid"
+	format "github.com/ipfs/go-ipld-format"
+	"github.com/ipfs/go-unixfsnode"
+	ufsData "github.com/ipfs/go-unixfsnode/data"
+	"github.com/ipfs/go-unixfsnode/hamt"
+	ufsiter "github.com/ipfs/go-unixfsnode/iter"
+	dagpb "github.com/ipld/go-codec-dagpb"
+	"github.com/ipld/go-ipld-prime"
+	"github.com/ipld/go-ipld-prime/datamodel"
+	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
+	"github.com/ipld/go-ipld-prime/node/basicnode"
+	"github.com/ipld/go-ipld-prime/schema"
+	"github.com/multiformats/go-multicodec"
+)
+
+type awaitCloser interface {
+	AwaitClose() <-chan error
+}
+
+type backpressuredFile struct {
+	size    int64
+	f       io.ReadSeeker
+	getLsys lsysGetter
+
+	ctx       context.Context
+	fileCid   cid.Cid
+	byteRange DagByteRange
+	retErr    error
+
+	closed chan error
+}
+
+func (b *backpressuredFile) AwaitClose() <-chan error {
+	return b.closed
+}
+
+func (b *backpressuredFile) Close() error {
+	close(b.closed)
+	return nil
+}
+
+func (b *backpressuredFile) Size() (int64, error) {
+	return b.size, nil
+}
+
+func (b *backpressuredFile) Read(p []byte) (n int, err error) {
+	if b.retErr == nil {
+		n, err = b.f.Read(p)
+		if err == nil || err == io.EOF {
+			return n, err
+		}
+
+		if n > 0 {
+			b.retErr = err
+			return n, nil
+		}
+	} else {
+		err = b.retErr
+	}
+
+	from, seekErr := b.f.Seek(0, io.SeekCurrent)
+	if seekErr != nil {
+		// Return the seek error since by this point seeking failures like this should be impossible
+		return 0, seekErr
+	}
+
+	// we had an error while reading so attempt to reset the underlying reader
+	for {
+		if b.ctx.Err() != nil {
+			return 0, b.ctx.Err()
+		}
+
+		retry, processedErr := isRetryableError(err)
+		if !retry {
+			return 0, processedErr
+		}
+
+		var nd files.Node
+		nd, err = loadTerminalUnixFSElementWithRecursiveDirectories(b.ctx, b.fileCid, nil, nil, CarParams{Scope: DagScopeEntity, Range: &DagByteRange{From: from, To: b.byteRange.To}}, b.getLsys)
+		if err != nil {
+			continue
+		}
+
+		f, ok := nd.(files.File)
+		if !ok {
+			return 0, fmt.Errorf("not a file, should be unreachable")
+		}
+
+		b.f = f
+		break
+	}
+
+	// now that we've reset the reader try reading again
+	return b.Read(p)
+}
+
+func (b *backpressuredFile) Seek(offset int64, whence int) (int64, error) {
+	return b.f.Seek(offset, whence)
+}
+
+var _ files.File = (*backpressuredFile)(nil)
+var _ awaitCloser = (*backpressuredFile)(nil)
+
+type singleUseDirectory struct {
+	dirIter files.DirIterator
+	closed  chan error
+}
+
+func (b *singleUseDirectory) AwaitClose() <-chan error {
+	return b.closed
+}
+
+func (b *singleUseDirectory) Close() error {
+	close(b.closed)
+	return nil
+}
+
+func (b *singleUseDirectory) Size() (int64, error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (b *singleUseDirectory) Entries() files.DirIterator {
+	return b.dirIter
+}
+
+var _ files.Directory = (*singleUseDirectory)(nil)
+var _ awaitCloser = (*singleUseDirectory)(nil)
+
+type backpressuredFlatDirIter struct {
+	linksItr *dagpb.PBLinks__Itr
+	lsys     *ipld.LinkSystem
+	getLsys  lsysGetter
+	ctx      context.Context
+
+	curName string
+	curFile files.Node
+
+	err error
+}
+
+func (it *backpressuredFlatDirIter) Name() string {
+	return it.curName
+}
+
+func (it *backpressuredFlatDirIter) Node() files.Node {
+	return it.curFile
+}
+
+func (it *backpressuredFlatDirIter) Next() bool {
+	if it.err != nil {
+		return false
+	}
+
+	iter := it.linksItr
+	if iter.Done() {
+		return false
+	}
+
+	_, v := iter.Next()
+	c := v.Hash.Link().(cidlink.Link).Cid
+	var name string
+	if v.Name.Exists() {
+		name = v.Name.Must().String()
+	}
+
+	var nd files.Node
+	var err error
+	params := CarParams{Scope: DagScopeAll}
+	for {
+		if it.ctx.Err() != nil {
+			it.err = it.ctx.Err()
+			return false
+		}
+		if err != nil {
+			it.lsys, err = it.getLsys(it.ctx, c, params)
+			continue
+		}
+		nd, err = loadTerminalUnixFSElementWithRecursiveDirectories(it.ctx, c, nil, it.lsys, params, it.getLsys)
+		if err != nil {
+			if ctxErr := it.ctx.Err(); ctxErr != nil {
+				continue
+			}
+			retry, processedErr := isRetryableError(err)
+			if retry {
+				err = processedErr
+				continue
+			}
+			it.err = processedErr
+			return false
+		}
+		break
+	}
+
+	it.curName = name
+	it.curFile = nd
+	return true
+}
+
+func (it *backpressuredFlatDirIter) Err() error {
+	return it.err
+}
+
+var _ files.DirIterator = (*backpressuredFlatDirIter)(nil)
+
+type backpressuredHAMTDirIter struct {
+	linksItr ipld.MapIterator
+	dirCid   cid.Cid
+
+	lsys    *ipld.LinkSystem
+	getLsys lsysGetter
+	ctx     context.Context
+
+	curName      string
+	curFile      files.Node
+	curProcessed int
+
+	err error
+}
+
+func (it *backpressuredHAMTDirIter) Name() string {
+	return it.curName
+}
+
+func (it *backpressuredHAMTDirIter) Node() files.Node {
+	return it.curFile
+}
+
+func (it *backpressuredHAMTDirIter) Next() bool {
+	if it.err != nil {
+		return false
+	}
+
+	iter := it.linksItr
+	if iter.Done() {
+		return false
+	}
+
+	/*
+		Since there is no way to make a graph request for part of a HAMT during errors we can either fill in the HAMT with
+		block requests, or we can re-request the HAMT and skip over the parts we already have.
+
+		Here we choose the latter, however in the event of a re-request we request the entity rather than the entire DAG as
+		a compromise between more requests and over-fetching data.
+	*/
+
+	var err error
+	for {
+		if it.ctx.Err() != nil {
+			it.err = it.ctx.Err()
+			return false
+		}
+
+		retry, processedErr := isRetryableError(err)
+		if !retry {
+			it.err = processedErr
+			return false
+		}
+
+		var nd ipld.Node
+		if err != nil {
+			var lsys *ipld.LinkSystem
+			lsys, err = it.getLsys(it.ctx, it.dirCid, CarParams{Scope: DagScopeEntity})
+			if err != nil {
+				continue
+			}
+
+			_, pbn, ufsFieldData, _, ufsBaseErr := loadUnixFSBase(it.ctx, it.dirCid, nil, lsys)
+			if ufsBaseErr != nil {
+				err = ufsBaseErr
+				continue
+			}
+
+			nd, err = hamt.NewUnixFSHAMTShard(it.ctx, pbn, ufsFieldData, lsys)
+			if err != nil {
+				err = fmt.Errorf("could not reify sharded directory: %w", err)
+				continue
+			}
+
+			iter = nd.MapIterator()
+			for i := 0; i < it.curProcessed; i++ {
+				_, _, err = iter.Next()
+				if err != nil {
+					continue
+				}
+			}
+
+			it.linksItr = iter
+		}
+
+		var k, v ipld.Node
+		k, v, err = iter.Next()
+		if err != nil {
+			retry, processedErr = isRetryableError(err)
+			if retry {
+				err = processedErr
+				continue
+			}
+			it.err = processedErr
+			return false
+		}
+
+		var name string
+		name, err = k.AsString()
+		if err != nil {
+			it.err = err
+			return false
+		}
+		var lnk ipld.Link
+		lnk, err = v.AsLink()
+		if err != nil {
+			it.err = err
+			return false
+		}
+
+		cl, ok := lnk.(cidlink.Link)
+		if !ok {
+			it.err = fmt.Errorf("link not a cidlink")
+			return false
+		}
+
+		c := cl.Cid
+		params := CarParams{Scope: DagScopeAll}
+		var childNd files.Node
+		for {
+			if it.ctx.Err() != nil {
+				it.err = it.ctx.Err()
+				return false
+			}
+
+			if err != nil {
+				retry, processedErr = isRetryableError(err)
+				if !retry {
+					it.err = processedErr
+					return false
+				}
+
+				it.lsys, err = it.getLsys(it.ctx, c, params)
+				continue
+			}
+
+			childNd, err = loadTerminalUnixFSElementWithRecursiveDirectories(it.ctx, c, nil, it.lsys, params, it.getLsys)
+			if err != nil {
+				continue
+			}
+			break
+		}
+
+		it.curName = name
+		it.curFile = childNd
+		it.curProcessed++
+		break
+	}
+
+	return true
+}
+
+func (it *backpressuredHAMTDirIter) Err() error {
+	return it.err
+}
+
+var _ files.DirIterator = (*backpressuredHAMTDirIter)(nil)
+
+type backpressuredHAMTDirIterNoRecursion struct {
+	dagSize  uint64
+	linksItr ipld.MapIterator
+	dirCid   cid.Cid
+
+	lsys    *ipld.LinkSystem
+	getLsys lsysGetter
+	ctx     context.Context
+
+	curLnk       unixfs.LinkResult
+	curProcessed int
+
+	closed    chan error
+	hasClosed bool
+	err       error
+}
+
+func (it *backpressuredHAMTDirIterNoRecursion) AwaitClose() <-chan error {
+	return it.closed
+}
+
+func (it *backpressuredHAMTDirIterNoRecursion) Link() unixfs.LinkResult {
+	return it.curLnk
+}
+
+func (it *backpressuredHAMTDirIterNoRecursion) Next() bool {
+	defer func() {
+		if it.linksItr.Done() || it.err != nil {
+			if !it.hasClosed {
+				it.hasClosed = true
+				close(it.closed)
+			}
+		}
+	}()
+
+	if it.err != nil {
+		return false
+	}
+
+	iter := it.linksItr
+	if iter.Done() {
+		return false
+	}
+
+	/*
+		Since there is no way to make a graph request for part of a HAMT during errors we can either fill in the HAMT with
+		block requests, or we can re-request the HAMT and skip over the parts we already have.
+
+		Here we choose the latter, however in the event of a re-request we request the entity rather than the entire DAG as
+		a compromise between more requests and over-fetching data.
+	*/
+
+	var err error
+	for {
+		if it.ctx.Err() != nil {
+			it.err = it.ctx.Err()
+			return false
+		}
+
+		retry, processedErr := isRetryableError(err)
+		if !retry {
+			it.err = processedErr
+			return false
+		}
+
+		var nd ipld.Node
+		if err != nil {
+			var lsys *ipld.LinkSystem
+			lsys, err = it.getLsys(it.ctx, it.dirCid, CarParams{Scope: DagScopeEntity})
+			if err != nil {
+				continue
+			}
+
+			_, pbn, ufsFieldData, _, ufsBaseErr := loadUnixFSBase(it.ctx, it.dirCid, nil, lsys)
+			if ufsBaseErr != nil {
+				err = ufsBaseErr
+				continue
+			}
+
+			nd, err = hamt.NewUnixFSHAMTShard(it.ctx, pbn, ufsFieldData, lsys)
+			if err != nil {
+				err = fmt.Errorf("could not reify sharded directory: %w", err)
+				continue
+			}
+
+			iter = nd.MapIterator()
+			for i := 0; i < it.curProcessed; i++ {
+				_, _, err = iter.Next()
+				if err != nil {
+					continue
+				}
+			}
+
+			it.linksItr = iter
+		}
+
+		var k, v ipld.Node
+		k, v, err = iter.Next()
+		if err != nil {
+			retry, processedErr = isRetryableError(err)
+			if retry {
+				err = processedErr
+				continue
+			}
+			it.err = processedErr
+			return false
+		}
+
+		var name string
+		name, err = k.AsString()
+		if err != nil {
+			it.err = err
+			return false
+		}
+
+		var lnk ipld.Link
+		lnk, err = v.AsLink()
+		if err != nil {
+			it.err = err
+			return false
+		}
+
+		cl, ok := lnk.(cidlink.Link)
+		if !ok {
+			it.err = fmt.Errorf("link not a cidlink")
+			return false
+		}
+
+		c := cl.Cid
+
+		pbLnk, ok := v.(*ufsiter.IterLink)
+		if !ok {
+			it.err = fmt.Errorf("HAMT value is not a dag-pb link")
+			return false
+		}
+
+		cumulativeDagSize := uint64(0)
+		if pbLnk.Substrate.Tsize.Exists() {
+			cumulativeDagSize = uint64(pbLnk.Substrate.Tsize.Must().Int())
+		}
+
+		it.curLnk = unixfs.LinkResult{
+			Link: &format.Link{
+				Name: name,
+				Size: cumulativeDagSize,
+				Cid:  c,
+			},
+		}
+		it.curProcessed++
+		break
+	}
+
+	return true
+}
+
+func (it *backpressuredHAMTDirIterNoRecursion) Err() error {
+	return it.err
+}
+
+var _ awaitCloser = (*backpressuredHAMTDirIterNoRecursion)(nil)
+
+/*
+1. Run traversal to get the top-level response
+2. Response can do a callback for another response
+*/
+
+type lsysGetter = func(ctx context.Context, c cid.Cid, params CarParams) (*ipld.LinkSystem, error)
+
+func loadUnixFSBase(ctx context.Context, c cid.Cid, blk blocks.Block, lsys *ipld.LinkSystem) ([]byte, dagpb.PBNode, ufsData.UnixFSData, int64, error) {
+	lctx := ipld.LinkContext{Ctx: ctx}
+	pathTerminalCidLink := cidlink.Link{Cid: c}
+
+	var blockData []byte
+	var err error
+
+	if blk != nil {
+		blockData = blk.RawData()
+	} else {
+		blockData, err = lsys.LoadRaw(lctx, pathTerminalCidLink)
+		if err != nil {
+			return nil, nil, nil, 0, err
+		}
+	}
+
+	if c.Type() == uint64(multicodec.Raw) {
+		return blockData, nil, nil, 0, nil
+	}
+
+	// decode the terminal block into a node
+	pc := dagpb.AddSupportToChooser(func(lnk ipld.Link, lnkCtx ipld.LinkContext) (ipld.NodePrototype, error) {
+		if tlnkNd, ok := lnkCtx.LinkNode.(schema.TypedLinkNode); ok {
+			return tlnkNd.LinkTargetNodePrototype(), nil
+		}
+		return basicnode.Prototype.Any, nil
+	})
+
+	np, err := pc(pathTerminalCidLink, lctx)
+	if err != nil {
+		return nil, nil, nil, 0, err
+	}
+
+	decoder, err := lsys.DecoderChooser(pathTerminalCidLink)
+	if err != nil {
+		return nil, nil, nil, 0, err
+	}
+	nb := np.NewBuilder()
+	if err := decoder(nb, bytes.NewReader(blockData)); err != nil {
+		return nil, nil, nil, 0, err
+	}
+	lastCidNode := nb.Build()
+
+	if pbn, ok := lastCidNode.(dagpb.PBNode); !ok {
+		// If it's not valid dag-pb then we're done
+		return nil, nil, nil, 0, errNotUnixFS
+	} else if !pbn.FieldData().Exists() {
+		// If it's not valid UnixFS then we're done
+		return nil, nil, nil, 0, errNotUnixFS
+	} else if unixfsFieldData, decodeErr := ufsData.DecodeUnixFSData(pbn.Data.Must().Bytes()); decodeErr != nil {
+		return nil, nil, nil, 0, errNotUnixFS
+	} else {
+		switch fieldNum := unixfsFieldData.FieldDataType().Int(); fieldNum {
+		case ufsData.Data_Symlink, ufsData.Data_Metadata, ufsData.Data_Raw, ufsData.Data_File, ufsData.Data_Directory, ufsData.Data_HAMTShard:
+			return nil, pbn, unixfsFieldData, fieldNum, nil
+		default:
+			return nil, nil, nil, 0, errNotUnixFS
+		}
+	}
+}
+
+func loadTerminalUnixFSElementWithRecursiveDirectories(ctx context.Context, c cid.Cid, blk blocks.Block, lsys *ipld.LinkSystem, params CarParams, getLsys lsysGetter) (files.Node, error) {
+	var err error
+	if lsys == nil {
+		lsys, err = getLsys(ctx, c, params)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	lctx := ipld.LinkContext{Ctx: ctx}
+	blockData, pbn, ufsFieldData, fieldNum, err := loadUnixFSBase(ctx, c, blk, lsys)
+	if err != nil {
+		return nil, err
+	}
+
+	if c.Type() == uint64(multicodec.Raw) {
+		return files.NewBytesFile(blockData), nil
+	}
+
+	switch fieldNum {
+	case ufsData.Data_Symlink:
+		if !ufsFieldData.FieldData().Exists() {
+			return nil, fmt.Errorf("invalid UnixFS symlink object")
+		}
+		lnkTarget := string(ufsFieldData.FieldData().Must().Bytes())
+		f := files.NewLinkFile(lnkTarget, nil)
+		return f, nil
+	case ufsData.Data_Metadata:
+		return nil, fmt.Errorf("UnixFS Metadata unsupported")
+	case ufsData.Data_HAMTShard, ufsData.Data_Directory:
+		switch fieldNum {
+		case ufsData.Data_Directory:
+			d := &singleUseDirectory{&backpressuredFlatDirIter{
+				ctx:      ctx,
+				linksItr: pbn.Links.Iterator(),
+				lsys:     lsys,
+				getLsys:  getLsys,
+			}, make(chan error)}
+			return d, nil
+		case ufsData.Data_HAMTShard:
+			dirNd, err := unixfsnode.Reify(lctx, pbn, lsys)
+			if err != nil {
+				return nil, fmt.Errorf("could not reify sharded directory: %w", err)
+			}
+
+			d := &singleUseDirectory{
+				&backpressuredHAMTDirIter{
+					linksItr: dirNd.MapIterator(),
+					dirCid:   c,
+					lsys:     lsys,
+					getLsys:  getLsys,
+					ctx:      ctx,
+				}, make(chan error),
+			}
+			return d, nil
+		default:
+			return nil, fmt.Errorf("not a basic or HAMT directory: should be unreachable")
+		}
+	case ufsData.Data_Raw, ufsData.Data_File:
+		nd, err := unixfsnode.Reify(lctx, pbn, lsys)
+		if err != nil {
+			return nil, err
+		}
+
+		fnd, ok := nd.(datamodel.LargeBytesNode)
+		if !ok {
+			return nil, fmt.Errorf("could not process file since it did not present as large bytes")
+		}
+		f, err := fnd.AsLargeBytes()
+		if err != nil {
+			return nil, err
+		}
+
+		fileSize, err := f.Seek(0, io.SeekEnd)
+		if err != nil {
+			return nil, fmt.Errorf("unable to get UnixFS file size: %w", err)
+		}
+
+		from := int64(0)
+		var byteRange DagByteRange
+		if params.Range != nil {
+			byteRange = *params.Range
+			from = params.Range.From
+		}
+		_, err = f.Seek(from, io.SeekStart)
+		if err != nil {
+			return nil, fmt.Errorf("unable to get reset UnixFS file reader: %w", err)
+		}
+
+		return &backpressuredFile{ctx: ctx, fileCid: c, byteRange: byteRange, size: fileSize, f: f, getLsys: getLsys, closed: make(chan error)}, nil
+	default:
+		return nil, fmt.Errorf("unknown UnixFS field type")
+	}
+}
diff --git a/gateway/backend_car_traversal.go b/gateway/backend_car_traversal.go
new file mode 100644
index 000000000..544935b04
--- /dev/null
+++ b/gateway/backend_car_traversal.go
@@ -0,0 +1,137 @@
+package gateway
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"sync"
+	"time"
+
+	"github.com/ipfs/boxo/verifcid"
+	blocks "github.com/ipfs/go-block-format"
+	"github.com/ipfs/go-cid"
+	"github.com/ipfs/go-unixfsnode"
+	"github.com/ipld/go-car"
+	"github.com/ipld/go-ipld-prime"
+	"github.com/ipld/go-ipld-prime/datamodel"
+	"github.com/ipld/go-ipld-prime/linking"
+	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
+	"github.com/multiformats/go-multihash"
+)
+
+type getBlock func(ctx context.Context, cid cid.Cid) (blocks.Block, error)
+
+var errNilBlock = ErrInvalidResponse{Message: "received a nil block with no error"}
+
+func carToLinearBlockGetter(ctx context.Context, reader io.Reader, timeout time.Duration, metrics *CarBackendMetrics) (getBlock, error) {
+	cr, err := car.NewCarReaderWithOptions(reader, car.WithErrorOnEmptyRoots(false))
+	if err != nil {
+		return nil, err
+	}
+
+	cbCtx, cncl := context.WithCancel(ctx)
+
+	type blockRead struct {
+		block blocks.Block
+		err   error
+	}
+
+	blkCh := make(chan blockRead, 1)
+	go func() {
+		defer cncl()
+		defer close(blkCh)
+		for {
+			blk, rdErr := cr.Next()
+			select {
+			case blkCh <- blockRead{blk, rdErr}:
+				if rdErr != nil {
+					cncl()
+				}
+			case <-cbCtx.Done():
+				return
+			}
+		}
+	}()
+
+	isFirstBlock := true
+	mx := sync.Mutex{}
+
+	return func(ctx context.Context, c cid.Cid) (blocks.Block, error) {
+		mx.Lock()
+		defer mx.Unlock()
+		if err := verifcid.ValidateCid(verifcid.DefaultAllowlist, c); err != nil {
+			return nil, err
+		}
+
+		isId, bdata := extractIdentityMultihashCIDContents(c)
+		if isId {
+			return blocks.NewBlockWithCid(bdata, c)
+		}
+
+		// initially set a higher timeout here so that if there's an initial timeout error we get it from the car reader.
+		var t *time.Timer
+		if isFirstBlock {
+			t = time.NewTimer(timeout * 2)
+		} else {
+			t = time.NewTimer(timeout)
+		}
+		var blkRead blockRead
+		var ok bool
+		select {
+		case blkRead, ok = <-blkCh:
+			if !t.Stop() {
+				<-t.C
+			}
+			t.Reset(timeout)
+		case <-t.C:
+			return nil, ErrGatewayTimeout
+		}
+		if !ok || blkRead.err != nil {
+			if !ok || errors.Is(blkRead.err, io.EOF) {
+				return nil, io.ErrUnexpectedEOF
+			}
+			return nil, blockstoreErrToGatewayErr(blkRead.err)
+		}
+		if blkRead.block != nil {
+			metrics.carBlocksFetchedMetric.Inc()
+			if !blkRead.block.Cid().Equals(c) {
+				return nil, ErrInvalidResponse{Message: fmt.Sprintf("received block with cid %s, expected %s", blkRead.block.Cid(), c)}
+			}
+			return blkRead.block, nil
+		}
+		return nil, errNilBlock
+	}, nil
+}
+
+// extractIdentityMultihashCIDContents will check if a given CID has an identity multihash and if so return true and
+// the bytes encoded in the digest, otherwise will return false.
+// Taken from https://github.com/ipfs/boxo/blob/b96767cc0971ca279feb36e7844e527a774309ab/blockstore/idstore.go#L30
+func extractIdentityMultihashCIDContents(k cid.Cid) (bool, []byte) {
+	// Pre-check by calling Prefix(), this much faster than extracting the hash.
+	if k.Prefix().MhType != multihash.IDENTITY {
+		return false, nil
+	}
+
+	dmh, err := multihash.Decode(k.Hash())
+	if err != nil || dmh.Code != multihash.IDENTITY {
+		return false, nil
+	}
+	return true, dmh.Digest
+}
+
+func getCarLinksystem(fn getBlock) *ipld.LinkSystem {
+	lsys := cidlink.DefaultLinkSystem()
+	lsys.StorageReadOpener = func(linkContext linking.LinkContext, link datamodel.Link) (io.Reader, error) {
+		c := link.(cidlink.Link).Cid
+		blk, err := fn(linkContext.Ctx, c)
+		if err != nil {
+			return nil, err
+		}
+		return bytes.NewReader(blk.RawData()), nil
+	}
+	lsys.TrustedStorage = true
+	unixfsnode.AddUnixFSReificationToLinkSystem(&lsys)
+	return &lsys
+}
diff --git a/gateway/blockstore.go b/gateway/blockstore.go
index f5043abe0..11e51b93e 100644
--- a/gateway/blockstore.go
+++ b/gateway/blockstore.go
@@ -34,12 +34,19 @@ var _ blockstore.Blockstore = (*cacheBlockStore)(nil)
 // NewCacheBlockStore creates a new [blockstore.Blockstore] that caches blocks
 // in memory using a two queue cache. It can be useful, for example, when paired
 // with a proxy blockstore (see [NewRemoteBlockstore]).
-func NewCacheBlockStore(size int) (blockstore.Blockstore, error) {
+//
+// If the given [prometheus.Registerer] is nil, a new one will be created using
+// [prometheus.NewRegistry].
+func NewCacheBlockStore(size int, reg prometheus.Registerer) (blockstore.Blockstore, error) {
 	c, err := lru.New2Q[string, []byte](size)
 	if err != nil {
 		return nil, err
 	}
 
+	if reg == nil {
+		reg = prometheus.NewRegistry()
+	}
+
 	cacheHitsMetric := prometheus.NewCounter(prometheus.CounterOpts{
 		Namespace: "ipfs",
 		Subsystem: "http",
@@ -54,12 +61,12 @@ func NewCacheBlockStore(size int) (blockstore.Blockstore, error) {
 		Help:      "The number of global block cache requests.",
 	})
 
-	err = prometheus.Register(cacheHitsMetric)
+	err = reg.Register(cacheHitsMetric)
 	if err != nil {
 		return nil, err
 	}
 
-	err = prometheus.Register(cacheRequestsMetric)
+	err = reg.Register(cacheRequestsMetric)
 	if err != nil {
 		return nil, err
 	}
@@ -151,18 +158,23 @@ type remoteBlockstore struct {
 }
 
 // NewRemoteBlockstore creates a new [blockstore.Blockstore] that is backed by one
-// or more gateways that support RAW block requests. See the [Trustless Gateway]
-// specification for more details.
+// or more gateways that support [RAW block] requests. See the [Trustless Gateway]
+// specification for more details. You can optionally pass your own [http.Client].
 //
 // [Trustless Gateway]: https://specs.ipfs.tech/http-gateways/trustless-gateway/
-func NewRemoteBlockstore(gatewayURL []string) (blockstore.Blockstore, error) {
+// [RAW block]: https://www.iana.org/assignments/media-types/application/vnd.ipld.raw
+func NewRemoteBlockstore(gatewayURL []string, httpClient *http.Client) (blockstore.Blockstore, error) {
 	if len(gatewayURL) == 0 {
-		return nil, errors.New("missing gateway URLs to which to proxy")
+		return nil, errors.New("missing remote block backend URL")
+	}
+
+	if httpClient == nil {
+		httpClient = newRemoteHTTPClient()
 	}
 
 	return &remoteBlockstore{
 		gatewayURL: gatewayURL,
-		httpClient: newRemoteHTTPClient(),
+		httpClient: httpClient,
 		rand:       rand.New(rand.NewSource(time.Now().Unix())),
 		// Enables block validation by default. Important since we are
 		// proxying block requests to untrusted gateways.
@@ -185,7 +197,7 @@ func (ps *remoteBlockstore) fetch(ctx context.Context, c cid.Cid) (blocks.Block,
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("http error from block gateway: %s", resp.Status)
+		return nil, fmt.Errorf("http error from remote block backend: %s", resp.Status)
 	}
 
 	rb, err := io.ReadAll(resp.Body)
diff --git a/gateway/errors.go b/gateway/errors.go
index 79cedcee0..c245ae4c1 100644
--- a/gateway/errors.go
+++ b/gateway/errors.go
@@ -10,9 +10,11 @@ import (
 	"time"
 
 	"github.com/ipfs/boxo/gateway/assets"
+	"github.com/ipfs/boxo/path"
 	"github.com/ipfs/boxo/path/resolver"
 	"github.com/ipfs/go-cid"
 	"github.com/ipld/go-ipld-prime/datamodel"
+	"github.com/ipld/go-ipld-prime/schema"
 )
 
 var (
@@ -127,6 +129,42 @@ func (e *ErrorStatusCode) Unwrap() error {
 	return e.Err
 }
 
+// ErrInvalidResponse can be returned from a [DataCallback] to indicate that
+// the data provided for the requested resource was explicitly 'incorrect',
+// for example, when received blocks did not belong to the requested dag,
+// or non-car-conforming data was returned.
+type ErrInvalidResponse struct {
+	Message string
+}
+
+func (e ErrInvalidResponse) Error() string {
+	return e.Message
+}
+
+// ErrPartialResponse can be returned from a [DataCallback] to indicate that some of the requested resource
+// was successfully fetched, and that instead of retrying the full resource, that there are
+// one or more more specific resources that should be fetched (via StillNeed) to complete the request.
+//
+// This primitive allows for resume mechanism that is useful when a big CAR
+// stream gets truncated due to network error, HTTP middleware timeout, etc,
+// but some useful blocks were received and should not be fetched again.
+type ErrPartialResponse struct {
+	error
+	StillNeed []CarResource
+}
+
+type CarResource struct {
+	Path   path.ImmutablePath
+	Params CarParams
+}
+
+func (epr ErrPartialResponse) Error() string {
+	if epr.error != nil {
+		return fmt.Sprintf("partial response: %s", epr.error.Error())
+	}
+	return "received a partial CAR response from the backend"
+}
+
 func webError(w http.ResponseWriter, r *http.Request, c *Config, err error, defaultCode int) {
 	code := defaultCode
 
@@ -184,7 +222,7 @@ func webError(w http.ResponseWriter, r *http.Request, c *Config, err error, defa
 // isErrNotFound returns true for IPLD errors that should return 4xx errors (e.g. the path doesn't exist, the data is
 // the wrong type, etc.), rather than issues with just finding and retrieving the data.
 func isErrNotFound(err error) bool {
-	if errors.Is(err, &resolver.ErrNoLink{}) {
+	if errors.Is(err, &resolver.ErrNoLink{}) || errors.Is(err, schema.ErrNoSuchField{}) {
 		return true
 	}
 
diff --git a/gateway/handler_unixfs_dir.go b/gateway/handler_unixfs_dir.go
index 098a77b6a..7a49dcafc 100644
--- a/gateway/handler_unixfs_dir.go
+++ b/gateway/handler_unixfs_dir.go
@@ -121,11 +121,9 @@ func (i *handler) serveDirectory(ctx context.Context, w http.ResponseWriter, r *
 			i.unixfsDirIndexGetMetric.WithLabelValues(originalContentPath.Namespace()).Observe(time.Since(rq.begin).Seconds())
 		}
 		return success
-	}
-
-	if isErrNotFound(err) {
+	} else if isErrNotFound(err) {
 		rq.logger.Debugw("no index.html; noop", "path", idxPath)
-	} else if err != nil {
+	} else {
 		i.webError(w, r, err, http.StatusInternalServerError)
 		return false
 	}
diff --git a/gateway/remote_blocks_backend.go b/gateway/remote_blocks_backend.go
deleted file mode 100644
index 5b96385d8..000000000
--- a/gateway/remote_blocks_backend.go
+++ /dev/null
@@ -1,53 +0,0 @@
-package gateway
-
-import (
-	"net/http"
-	"time"
-
-	"github.com/ipfs/boxo/blockservice"
-	"github.com/ipfs/boxo/exchange/offline"
-	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
-)
-
-// TODO: make this configurable via BlocksBackendOption
-const getBlockTimeout = time.Second * 60
-
-// NewRemoteBlocksBackend creates a new [BlocksBackend] instance backed by one
-// or more gateways. These gateways must support RAW block requests and IPNS
-// Record requests. See [NewRemoteBlockstore] and [NewRemoteValueStore] for
-// more details.
-//
-// If you want to create a more custom [BlocksBackend] with only remote IPNS
-// Record resolution, or only remote block fetching, we recommend using
-// [NewBlocksBackend] directly.
-func NewRemoteBlocksBackend(gatewayURL []string, opts ...BlocksBackendOption) (*BlocksBackend, error) {
-	blockStore, err := NewRemoteBlockstore(gatewayURL)
-	if err != nil {
-		return nil, err
-	}
-
-	valueStore, err := NewRemoteValueStore(gatewayURL)
-	if err != nil {
-		return nil, err
-	}
-
-	blockService := blockservice.New(blockStore, offline.Exchange(blockStore))
-	return NewBlocksBackend(blockService, append(opts, WithValueStore(valueStore))...)
-}
-
-// newRemoteHTTPClient creates a new [http.Client] that is optimized for retrieving
-// multiple blocks from a single gateway concurrently.
-func newRemoteHTTPClient() *http.Client {
-	transport := &http.Transport{
-		MaxIdleConns:        1000,
-		MaxConnsPerHost:     100,
-		MaxIdleConnsPerHost: 100,
-		IdleConnTimeout:     90 * time.Second,
-		ForceAttemptHTTP2:   true,
-	}
-
-	return &http.Client{
-		Timeout:   getBlockTimeout,
-		Transport: otelhttp.NewTransport(transport),
-	}
-}
diff --git a/gateway/value_store.go b/gateway/value_store.go
index d494fc212..ead5a44e7 100644
--- a/gateway/value_store.go
+++ b/gateway/value_store.go
@@ -20,19 +20,23 @@ type remoteValueStore struct {
 	rand       *rand.Rand
 }
 
-// NewRemoteValueStore creates a new [routing.ValueStore] that is backed by one
-// or more gateways that support IPNS Record requests. See the [Trustless Gateway]
-// specification for more details.
+// NewRemoteValueStore creates a new [routing.ValueStore] backed by one or more
+// gateways that support IPNS Record requests. See the [Trustless Gateway]
+// specification for more details. You can optionally pass your own [http.Client].
 //
 // [Trustless Gateway]: https://specs.ipfs.tech/http-gateways/trustless-gateway/
-func NewRemoteValueStore(gatewayURL []string) (routing.ValueStore, error) {
+func NewRemoteValueStore(gatewayURL []string, httpClient *http.Client) (routing.ValueStore, error) {
 	if len(gatewayURL) == 0 {
 		return nil, errors.New("missing gateway URLs to which to proxy")
 	}
 
+	if httpClient == nil {
+		httpClient = newRemoteHTTPClient()
+	}
+
 	return &remoteValueStore{
 		gatewayURL: gatewayURL,
-		httpClient: newRemoteHTTPClient(),
+		httpClient: httpClient,
 		rand:       rand.New(rand.NewSource(time.Now().Unix())),
 	}, nil
 }
diff --git a/go.mod b/go.mod
index 4ef91d9f8..a6c113337 100644
--- a/go.mod
+++ b/go.mod
@@ -13,6 +13,7 @@ require (
 	github.com/gogo/protobuf v1.3.2
 	github.com/google/uuid v1.5.0
 	github.com/gorilla/mux v1.8.1
+	github.com/hashicorp/go-multierror v1.1.1
 	github.com/hashicorp/golang-lru/v2 v2.0.7
 	github.com/ipfs/bbloom v0.0.4
 	github.com/ipfs/go-bitfield v1.1.0
@@ -30,6 +31,7 @@ require (
 	github.com/ipfs/go-metrics-interface v0.0.1
 	github.com/ipfs/go-peertaskqueue v0.8.1
 	github.com/ipfs/go-unixfsnode v1.9.0
+	github.com/ipld/go-car v0.6.2
 	github.com/ipld/go-car/v2 v2.13.1
 	github.com/ipld/go-codec-dagpb v1.6.0
 	github.com/ipld/go-ipld-prime v0.21.0
@@ -103,14 +105,19 @@ require (
 	github.com/gorilla/websocket v1.5.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.19.0 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
-	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/hashicorp/golang-lru v1.0.2 // indirect
 	github.com/huin/goupnp v1.3.0 // indirect
+	github.com/ipfs/go-blockservice v0.5.0 // indirect
+	github.com/ipfs/go-ipfs-blockstore v1.3.0 // indirect
+	github.com/ipfs/go-ipfs-ds-help v1.1.0 // indirect
+	github.com/ipfs/go-ipfs-exchange-interface v0.2.0 // indirect
 	github.com/ipfs/go-ipfs-pq v0.0.3 // indirect
 	github.com/ipfs/go-ipfs-util v0.0.3 // indirect
 	github.com/ipfs/go-ipld-cbor v0.1.0 // indirect
 	github.com/ipfs/go-log v1.0.5 // indirect
+	github.com/ipfs/go-merkledag v0.11.0 // indirect
 	github.com/ipfs/go-unixfs v0.4.5 // indirect
+	github.com/ipfs/go-verifcid v0.0.2 // indirect
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
 	github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
 	github.com/klauspost/compress v1.17.4 // indirect
diff --git a/go.sum b/go.sum
index bf51a10ed..388ee6d9f 100644
--- a/go.sum
+++ b/go.sum
@@ -137,6 +137,7 @@ github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OI
 github.com/google/pprof v0.0.0-20231229205709-960ae82b1e42 h1:dHLYa5D8/Ta0aLR2XcPsrkpAgGeFs6thhMcQK0oQ0n8=
 github.com/google/pprof v0.0.0-20231229205709-960ae82b1e42/go.mod h1:czg5+yv1E0ZGTi6S6vVK1mke0fV+FaUhNGcd6VRS9Ik=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
+github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.5.0 h1:1p67kYwdtXjb0gL0BPiP1Av9wiZPo5A8z2cWkTZ+eyU=
 github.com/google/uuid v1.5.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
@@ -170,17 +171,21 @@ github.com/ipfs/bbloom v0.0.4 h1:Gi+8EGJ2y5qiD5FbsbpX/TMNcJw8gSqr7eyjHa4Fhvs=
 github.com/ipfs/bbloom v0.0.4/go.mod h1:cS9YprKXpoZ9lT0n/Mw/a6/aFV6DTjTLYHeA+gyqMG0=
 github.com/ipfs/go-bitfield v1.1.0 h1:fh7FIo8bSwaJEh6DdTWbCeZ1eqOaOkKFI74SCnsWbGA=
 github.com/ipfs/go-bitfield v1.1.0/go.mod h1:paqf1wjq/D2BBmzfTVFlJQ9IlFOZpg422HL0HqsGWHU=
+github.com/ipfs/go-bitswap v0.11.0 h1:j1WVvhDX1yhG32NTC9xfxnqycqYIlhzEzLXG/cU1HyQ=
+github.com/ipfs/go-bitswap v0.11.0/go.mod h1:05aE8H3XOU+LXpTedeAS0OZpcO1WFsj5niYQH9a1Tmk=
 github.com/ipfs/go-block-format v0.0.2/go.mod h1:AWR46JfpcObNfg3ok2JHDUfdiHRgWhJgCQF+KIgOPJY=
 github.com/ipfs/go-block-format v0.2.0 h1:ZqrkxBA2ICbDRbK8KJs/u0O3dlp6gmAuuXUJNiW1Ycs=
 github.com/ipfs/go-block-format v0.2.0/go.mod h1:+jpL11nFx5A/SPpsoBn6Bzkra/zaArfSmsknbPMYgzM=
 github.com/ipfs/go-blockservice v0.5.0 h1:B2mwhhhVQl2ntW2EIpaWPwSCxSuqr5fFA93Ms4bYLEY=
 github.com/ipfs/go-blockservice v0.5.0/go.mod h1:W6brZ5k20AehbmERplmERn8o2Ni3ZZubvAxaIUeaT6w=
 github.com/ipfs/go-cid v0.0.1/go.mod h1:GHWU/WuQdMPmIosc4Yn1bcCT7dSeX4lBafM7iqUPQvM=
+github.com/ipfs/go-cid v0.0.5/go.mod h1:plgt+Y5MnOey4vO4UlUazGqdbEXuFYitED67FexhXog=
 github.com/ipfs/go-cid v0.0.6/go.mod h1:6Ux9z5e+HpkQdckYoX1PG/6xqKspzlEIR5SDmgqgC/I=
 github.com/ipfs/go-cid v0.4.1 h1:A/T3qGvxi4kpKWWcPC/PgbvDA2bjVLO7n4UeVwnbs/s=
 github.com/ipfs/go-cid v0.4.1/go.mod h1:uQHwDeX4c6CtyrFwdqyhpNcxVewur1M7l7fNU7LKwZk=
 github.com/ipfs/go-cidutil v0.1.0 h1:RW5hO7Vcf16dplUU60Hs0AKDkQAVPVplr7lk97CFL+Q=
 github.com/ipfs/go-cidutil v0.1.0/go.mod h1:e7OEVBMIv9JaOxt9zaGEmAoSlXW9jdFZ5lP/0PwcfpA=
+github.com/ipfs/go-datastore v0.5.0/go.mod h1:9zhEApYMTl17C8YDp7JmU7sQZi2/wqiYh73hakZ90Bk=
 github.com/ipfs/go-datastore v0.6.0 h1:JKyz+Gvz1QEZw0LsX1IBn+JFCJQH4SJVFtM4uWU0Myk=
 github.com/ipfs/go-datastore v0.6.0/go.mod h1:rt5M3nNbSO/8q1t4LNkLyUwRs8HupMeN/8O4Vn9YAT8=
 github.com/ipfs/go-detect-race v0.0.1 h1:qX/xay2W3E4Q1U7d9lNs1sU9nvguX0a7319XbyQ6cOk=
@@ -191,6 +196,7 @@ github.com/ipfs/go-ipfs-blocksutil v0.0.1 h1:Eh/H4pc1hsvhzsQoMEP3Bke/aW5P5rVM1IW
 github.com/ipfs/go-ipfs-blocksutil v0.0.1/go.mod h1:Yq4M86uIOmxmGPUHv/uI7uKqZNtLb449gwKqXjIsnRk=
 github.com/ipfs/go-ipfs-chunker v0.0.5 h1:ojCf7HV/m+uS2vhUGWcogIIxiO5ubl5O57Q7NapWLY8=
 github.com/ipfs/go-ipfs-chunker v0.0.5/go.mod h1:jhgdF8vxRHycr00k13FM8Y0E+6BoalYeobXmUyTreP8=
+github.com/ipfs/go-ipfs-delay v0.0.0-20181109222059-70721b86a9a8/go.mod h1:8SP1YXK1M1kXuc4KJZINY3TQQ03J2rwBG9QfXmbRPrw=
 github.com/ipfs/go-ipfs-delay v0.0.1 h1:r/UXYyRcddO6thwOnhiznIAiSvxMECGgtv35Xs1IeRQ=
 github.com/ipfs/go-ipfs-delay v0.0.1/go.mod h1:8SP1YXK1M1kXuc4KJZINY3TQQ03J2rwBG9QfXmbRPrw=
 github.com/ipfs/go-ipfs-ds-help v1.1.0 h1:yLE2w9RAsl31LtfMt91tRZcrx+e61O5mDxFRR994w4Q=
@@ -203,6 +209,8 @@ github.com/ipfs/go-ipfs-pq v0.0.3 h1:YpoHVJB+jzK15mr/xsWC574tyDLkezVrDNeaalQBsTE
 github.com/ipfs/go-ipfs-pq v0.0.3/go.mod h1:btNw5hsHBpRcSSgZtiNm/SLj5gYIZ18AKtv3kERkRb4=
 github.com/ipfs/go-ipfs-redirects-file v0.1.1 h1:Io++k0Vf/wK+tfnhEh63Yte1oQK5VGT2hIEYpD0Rzx8=
 github.com/ipfs/go-ipfs-redirects-file v0.1.1/go.mod h1:tAwRjCV0RjLTjH8DR/AU7VYvfQECg+lpUy2Mdzv7gyk=
+github.com/ipfs/go-ipfs-routing v0.3.0 h1:9W/W3N+g+y4ZDeffSgqhgo7BsBSJwPMcyssET9OWevc=
+github.com/ipfs/go-ipfs-routing v0.3.0/go.mod h1:dKqtTFIql7e1zYsEuWLyuOU+E0WJWW8JjbTPLParDWo=
 github.com/ipfs/go-ipfs-util v0.0.1/go.mod h1:spsl5z8KUnrve+73pOhSVZND1SIxPW5RyBCNzQxlJBc=
 github.com/ipfs/go-ipfs-util v0.0.3 h1:2RFdGez6bu2ZlZdI+rWfIdbQb1KudQp3VGwPtdNCmE0=
 github.com/ipfs/go-ipfs-util v0.0.3/go.mod h1:LHzG1a0Ig4G+iZ26UUOMjHd+lfM84LZCrn17xAKWBvs=
@@ -229,6 +237,8 @@ github.com/ipfs/go-unixfsnode v1.9.0 h1:ubEhQhr22sPAKO2DNsyVBW7YB/zA8Zkif25aBvz8
 github.com/ipfs/go-unixfsnode v1.9.0/go.mod h1:HxRu9HYHOjK6HUqFBAi++7DVoWAHn0o4v/nZ/VA+0g8=
 github.com/ipfs/go-verifcid v0.0.2 h1:XPnUv0XmdH+ZIhLGKg6U2vaPaRDXb9urMyNVCE7uvTs=
 github.com/ipfs/go-verifcid v0.0.2/go.mod h1:40cD9x1y4OWnFXbLNJYRe7MpNvWlMn3LZAG5Wb4xnPU=
+github.com/ipld/go-car v0.6.2 h1:Hlnl3Awgnq8icK+ze3iRghk805lu8YNq3wlREDTF2qc=
+github.com/ipld/go-car v0.6.2/go.mod h1:oEGXdwp6bmxJCZ+rARSkDliTeYnVzv3++eXajZ+Bmr8=
 github.com/ipld/go-car/v2 v2.13.1 h1:KnlrKvEPEzr5IZHKTXLAEub+tPrzeAFQVRlSQvuxBO4=
 github.com/ipld/go-car/v2 v2.13.1/go.mod h1:QkdjjFNGit2GIkpQ953KBwowuoukoM75nP/JI1iDJdo=
 github.com/ipld/go-codec-dagpb v1.6.0 h1:9nYazfyu9B1p3NAgfVdpRco3Fs2nFC72DqVsMj6rOcc=
@@ -260,6 +270,7 @@ github.com/klauspost/cpuid/v2 v2.2.6/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZY
 github.com/koron/go-ssdp v0.0.4 h1:1IDwrghSKYM7yLf7XCzbByg2sJ/JcNOZRXS2jczTwz0=
 github.com/koron/go-ssdp v0.0.4/go.mod h1:oDXq+E5IL5q0U8uSBcoAXzTzInwy5lEgC91HoKtbmZk=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
@@ -720,6 +731,7 @@ google.golang.org/protobuf v1.32.0 h1:pPC6BG5ex8PDFnkbrGU3EixyhKcQ2aDuBS36lqK/C7
 google.golang.org/protobuf v1.32.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=


Test patch:
diff --git a/examples/gateway/car/main_test.go b/examples/gateway/car-file/main_test.go
similarity index 100%
rename from examples/gateway/car/main_test.go
rename to examples/gateway/car-file/main_test.go
diff --git a/examples/gateway/car/test.car b/examples/gateway/car-file/test.car
similarity index 100%
rename from examples/gateway/car/test.car
rename to examples/gateway/car-file/test.car
diff --git a/examples/gateway/proxy/main_test.go b/examples/gateway/proxy-blocks/main_test.go
similarity index 99%
rename from examples/gateway/proxy/main_test.go
rename to examples/gateway/proxy-blocks/main_test.go
index 309ffb59e..8cb86bbff 100644
--- a/examples/gateway/proxy/main_test.go
+++ b/examples/gateway/proxy-blocks/main_test.go
@@ -21,7 +21,7 @@ const (
 )
 
 func newProxyGateway(t *testing.T, rs *httptest.Server) *httptest.Server {
-	backend, err := gateway.NewRemoteBlocksBackend([]string{rs.URL})
+	backend, err := gateway.NewRemoteBlocksBackend([]string{rs.URL}, nil)
 	require.NoError(t, err)
 	handler := common.NewHandler(backend)
 	ts := httptest.NewServer(handler)
diff --git a/gateway/backend_car_fetcher_test.go b/gateway/backend_car_fetcher_test.go
new file mode 100644
index 000000000..383f20c2d
--- /dev/null
+++ b/gateway/backend_car_fetcher_test.go
@@ -0,0 +1,60 @@
+package gateway
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/ipfs/boxo/path"
+)
+
+func TestContentPathToCarUrl(t *testing.T) {
+	negativeOffset := int64(-42)
+	testCases := []struct {
+		contentPath string // to be turned into ImmutablePath
+		carParams   CarParams
+		expectedUrl string // url.URL.String()
+	}{
+		{
+			contentPath: "/ipfs/bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi",
+			carParams:   CarParams{},
+			expectedUrl: "/ipfs/bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi?format=car",
+		},
+		{
+			contentPath: "/ipfs/bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi",
+			carParams:   CarParams{Scope: "entity", Range: &DagByteRange{From: 0, To: nil}},
+			expectedUrl: "/ipfs/bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi?format=car&dag-scope=entity&entity-bytes=0:*",
+		},
+		{
+			contentPath: "/ipfs/bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi",
+			carParams:   CarParams{Scope: "block"},
+			expectedUrl: "/ipfs/bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi?format=car&dag-scope=block",
+		},
+		{
+			contentPath: "/ipfs/bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi",
+			carParams:   CarParams{Scope: "entity", Range: &DagByteRange{From: 4, To: &negativeOffset}},
+			expectedUrl: "/ipfs/bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi?format=car&dag-scope=entity&entity-bytes=4:-42",
+		},
+		{
+			// a regression test for case described in https://github.com/ipfs/gateway-conformance/issues/115
+			contentPath: "/ipfs/bafybeiaysi4s6lnjev27ln5icwm6tueaw2vdykrtjkwiphwekaywqhcjze/I/Auditorio_de_Tenerife%2C_Santa_Cruz_de_Tenerife%2C_España%2C_2012-12-15%2C_DD_02.jpg.webp",
+			carParams:   CarParams{Scope: "entity", Range: &DagByteRange{From: 0, To: nil}},
+			expectedUrl: "/ipfs/bafybeiaysi4s6lnjev27ln5icwm6tueaw2vdykrtjkwiphwekaywqhcjze/I/Auditorio_de_Tenerife%252C_Santa_Cruz_de_Tenerife%252C_Espa%C3%B1a%252C_2012-12-15%252C_DD_02.jpg.webp?format=car&dag-scope=entity&entity-bytes=0:*",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run("TestContentPathToCarUrl", func(t *testing.T) {
+			p, err := path.NewPath(tc.contentPath)
+			require.NoError(t, err)
+
+			contentPath, err := path.NewImmutablePath(p)
+			require.NoError(t, err)
+
+			result := contentPathToCarUrl(contentPath, tc.carParams).String()
+			if result != tc.expectedUrl {
+				t.Errorf("Expected %q, but got %q", tc.expectedUrl, result)
+			}
+		})
+	}
+}
diff --git a/gateway/backend_car_test.go b/gateway/backend_car_test.go
new file mode 100644
index 000000000..eebd8e19b
--- /dev/null
+++ b/gateway/backend_car_test.go
@@ -0,0 +1,1100 @@
+package gateway
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	_ "embed"
+
+	"github.com/ipfs/boxo/blockservice"
+	"github.com/ipfs/boxo/exchange/offline"
+	"github.com/ipfs/boxo/files"
+	"github.com/ipfs/boxo/ipld/merkledag"
+	unixfile "github.com/ipfs/boxo/ipld/unixfs/file"
+	"github.com/ipfs/boxo/path"
+	"github.com/ipfs/go-cid"
+	carv2 "github.com/ipld/go-car/v2"
+	carbs "github.com/ipld/go-car/v2/blockstore"
+	"github.com/ipld/go-car/v2/storage"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+//go:embed testdata/directory-with-multilayer-hamt-and-multiblock-files.car
+var dirWithMultiblockHAMTandFiles []byte
+
+func TestCarBackendTar(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	requestNum := 0
+	s := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		requestNum++
+		switch requestNum {
+		case 1:
+			// Expect the full request, but return one that terminates in the middle of the HAMT
+			expectedUri := "/ipfs/bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi", // root dir
+				"bafybeifdv255wmsrh75vcsrtkcwyktvewgihegeeyhhj2ju4lzt4lqfoze", // basicDir
+				"bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa", // exampleA
+				"bafkreie5noke3mb7hqxukzcy73nl23k6lxszxi5w3dtmuwz62wnvkpsscm",
+				"bafkreih4ephajybraj6wnxsbwjwa77fukurtpl7oj7t7pfq545duhot7cq",
+				"bafkreigu7buvm3cfunb35766dn7tmqyh2um62zcio63en2btvxuybgcpue",
+				"bafkreicll3huefkc3qnrzeony7zcfo7cr3nbx64hnxrqzsixpceg332fhe",
+				"bafkreifst3pqztuvj57lycamoi7z34b4emf7gawxs74nwrc2c7jncmpaqm",
+				"bafybeid3trcauvcp7fxaai23gkz3qexmlfxnnejgwm57hdvre472dafvha", // exampleB
+				"bafkreihgbi345degbcyxaf5b3boiyiaxhnuxdysvqmbdyaop2swmhh3s3m",
+				"bafkreiaugmh5gal5rgiems6gslcdt2ixfncahintrmcqvrgxqamwtlrmz4",
+				"bafkreiaxwwb7der2qvmteymgtlj7ww7w5vc44phdxfnexog3vnuwdkxuea",
+				"bafkreic5zyan5rk4ccfum4d4mu3h5eqsllbudlj4texlzj6xdgxvldzngi",
+				"bafybeignui4g7l6cvqyy4t6vnbl2fjtego4ejmpcia77jhhwnksmm4bejm", // hamtDir
+			}); err != nil {
+				panic(err)
+			}
+		case 2:
+			// Expect a request for the HAMT only and give it
+			// Note: this is an implementation detail, it could be in the future that we request less or more data
+			// (e.g. requesting the blocks to fill out the HAMT, or with spec changes asking for HAMT ranges, or asking for the HAMT and its children)
+			expectedUri := "/ipfs/bafybeignui4g7l6cvqyy4t6vnbl2fjtego4ejmpcia77jhhwnksmm4bejm"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeignui4g7l6cvqyy4t6vnbl2fjtego4ejmpcia77jhhwnksmm4bejm", // hamtDir
+				"bafybeiccgo7euew77gkqkhezn3pozfrciiibqz2u3spdqmgjvd5wqskipm",
+				"bafybeihjydob4eq5j4m43whjgf5cgftthc42kjno3g24sa3wcw7vonbmfy",
+			}); err != nil {
+				panic(err)
+			}
+		case 3:
+			// Starting here expect requests for each file in the directory
+			expectedUri := "/ipfs/bafybeid3trcauvcp7fxaai23gkz3qexmlfxnnejgwm57hdvre472dafvha"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeid3trcauvcp7fxaai23gkz3qexmlfxnnejgwm57hdvre472dafvha", // exampleB
+				"bafkreihgbi345degbcyxaf5b3boiyiaxhnuxdysvqmbdyaop2swmhh3s3m",
+				"bafkreiaugmh5gal5rgiems6gslcdt2ixfncahintrmcqvrgxqamwtlrmz4",
+				"bafkreiaxwwb7der2qvmteymgtlj7ww7w5vc44phdxfnexog3vnuwdkxuea",
+				"bafkreic5zyan5rk4ccfum4d4mu3h5eqsllbudlj4texlzj6xdgxvldzngi",
+			}); err != nil {
+				panic(err)
+			}
+		case 4:
+			// Expect a request for one of the directory items and give it
+			expectedUri := "/ipfs/bafkreih2grj7p2bo5yk2guqazxfjzapv6hpm3mwrinv6s3cyayd72ke5he"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafkreih2grj7p2bo5yk2guqazxfjzapv6hpm3mwrinv6s3cyayd72ke5he", // exampleD
+			}); err != nil {
+				panic(err)
+			}
+		case 5:
+			// Expect a request for one of the directory items and give it
+			expectedUri := "/ipfs/bafkreidqhbqn5htm5qejxpb3hps7dookudo3nncfn6al6niqibi5lq6fee"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafkreidqhbqn5htm5qejxpb3hps7dookudo3nncfn6al6niqibi5lq6fee", // exampleC
+			}); err != nil {
+				panic(err)
+			}
+		case 6:
+			// Expect a request for one of the directory items and give part of it
+			expectedUri := "/ipfs/bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa", // exampleA
+				"bafkreie5noke3mb7hqxukzcy73nl23k6lxszxi5w3dtmuwz62wnvkpsscm",
+				"bafkreih4ephajybraj6wnxsbwjwa77fukurtpl7oj7t7pfq545duhot7cq",
+			}); err != nil {
+				panic(err)
+			}
+		case 7:
+			// Expect a partial request for one of the directory items and give it
+			expectedUri := "/ipfs/bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa", // exampleA
+				"bafkreigu7buvm3cfunb35766dn7tmqyh2um62zcio63en2btvxuybgcpue",
+				"bafkreicll3huefkc3qnrzeony7zcfo7cr3nbx64hnxrqzsixpceg332fhe",
+				"bafkreifst3pqztuvj57lycamoi7z34b4emf7gawxs74nwrc2c7jncmpaqm",
+			}); err != nil {
+				panic(err)
+			}
+		default:
+			t.Fatal("unsupported request number")
+		}
+	}))
+	defer s.Close()
+
+	bs, err := NewRemoteCarFetcher([]string{s.URL}, nil)
+	require.NoError(t, err)
+
+	fetcher, err := NewRetryCarFetcher(bs, 3)
+	require.NoError(t, err)
+
+	backend, err := NewCarBackend(fetcher)
+	require.NoError(t, err)
+
+	p := path.FromCid(cid.MustParse("bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi"))
+	_, nd, err := backend.GetAll(ctx, p)
+	require.NoError(t, err)
+
+	assertNextEntryNameEquals := func(t *testing.T, dirIter files.DirIterator, expectedName string) {
+		t.Helper()
+		require.True(t, dirIter.Next(), dirIter.Err())
+		require.Equal(t, expectedName, dirIter.Name())
+	}
+
+	robs, err := carbs.NewReadOnly(bytes.NewReader(dirWithMultiblockHAMTandFiles), nil)
+	require.NoError(t, err)
+
+	dsrv := merkledag.NewDAGService(blockservice.New(robs, offline.Exchange(robs)))
+	assertFileEqual := func(t *testing.T, expectedCidString string, receivedFile files.File) {
+		t.Helper()
+
+		expected := cid.MustParse(expectedCidString)
+		receivedFileData, err := io.ReadAll(receivedFile)
+		require.NoError(t, err)
+		nd, err := dsrv.Get(ctx, expected)
+		require.NoError(t, err)
+		expectedFile, err := unixfile.NewUnixfsFile(ctx, dsrv, nd)
+		require.NoError(t, err)
+
+		expectedFileData, err := io.ReadAll(expectedFile.(files.File))
+		require.NoError(t, err)
+		require.True(t, bytes.Equal(expectedFileData, receivedFileData))
+	}
+
+	rootDirIter := nd.(files.Directory).Entries()
+	assertNextEntryNameEquals(t, rootDirIter, "basicDir")
+
+	basicDirIter := rootDirIter.Node().(files.Directory).Entries()
+	assertNextEntryNameEquals(t, basicDirIter, "exampleA")
+	assertFileEqual(t, "bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa", basicDirIter.Node().(files.File))
+
+	assertNextEntryNameEquals(t, basicDirIter, "exampleB")
+	assertFileEqual(t, "bafybeid3trcauvcp7fxaai23gkz3qexmlfxnnejgwm57hdvre472dafvha", basicDirIter.Node().(files.File))
+
+	assertNextEntryNameEquals(t, rootDirIter, "hamtDir")
+	hamtDirIter := rootDirIter.Node().(files.Directory).Entries()
+
+	assertNextEntryNameEquals(t, hamtDirIter, "exampleB")
+	assertFileEqual(t, "bafybeid3trcauvcp7fxaai23gkz3qexmlfxnnejgwm57hdvre472dafvha", hamtDirIter.Node().(files.File))
+
+	assertNextEntryNameEquals(t, hamtDirIter, "exampleD-hamt-collide-exampleB-seed-364")
+	assertFileEqual(t, "bafkreih2grj7p2bo5yk2guqazxfjzapv6hpm3mwrinv6s3cyayd72ke5he", hamtDirIter.Node().(files.File))
+
+	assertNextEntryNameEquals(t, hamtDirIter, "exampleC-hamt-collide-exampleA-seed-52")
+	assertFileEqual(t, "bafkreidqhbqn5htm5qejxpb3hps7dookudo3nncfn6al6niqibi5lq6fee", hamtDirIter.Node().(files.File))
+
+	assertNextEntryNameEquals(t, hamtDirIter, "exampleA")
+	assertFileEqual(t, "bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa", hamtDirIter.Node().(files.File))
+
+	require.False(t, rootDirIter.Next() || basicDirIter.Next() || hamtDirIter.Next())
+}
+
+func TestCarBackendTarAtEndOfPath(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	requestNum := 0
+	s := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		requestNum++
+		switch requestNum {
+		case 1:
+			// Expect the full request, but return one that terminates in the middle of the path
+			expectedUri := "/ipfs/bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi/hamtDir"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi", // root dir
+			}); err != nil {
+				panic(err)
+			}
+		case 2:
+			// Expect the full request and give the path and the children from one of the HAMT nodes but not the other
+			// Note: this is an implementation detail, it could be in the future that we request less or more data
+			expectedUri := "/ipfs/bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi/hamtDir"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi", // root dir
+				"bafybeignui4g7l6cvqyy4t6vnbl2fjtego4ejmpcia77jhhwnksmm4bejm", // hamtDir
+				"bafybeiccgo7euew77gkqkhezn3pozfrciiibqz2u3spdqmgjvd5wqskipm",
+				"bafybeid3trcauvcp7fxaai23gkz3qexmlfxnnejgwm57hdvre472dafvha", // exampleB
+				"bafkreihgbi345degbcyxaf5b3boiyiaxhnuxdysvqmbdyaop2swmhh3s3m",
+				"bafkreiaugmh5gal5rgiems6gslcdt2ixfncahintrmcqvrgxqamwtlrmz4",
+				"bafkreiaxwwb7der2qvmteymgtlj7ww7w5vc44phdxfnexog3vnuwdkxuea",
+				"bafkreic5zyan5rk4ccfum4d4mu3h5eqsllbudlj4texlzj6xdgxvldzngi",
+				"bafkreih2grj7p2bo5yk2guqazxfjzapv6hpm3mwrinv6s3cyayd72ke5he", // exampleD
+			}); err != nil {
+				panic(err)
+			}
+		case 3:
+			// Expect a request for the HAMT only and give it
+			// Note: this is an implementation detail, it could be in the future that we request less or more data
+			// (e.g. requesting the blocks to fill out the HAMT, or with spec changes asking for HAMT ranges, or asking for the HAMT and its children)
+			expectedUri := "/ipfs/bafybeignui4g7l6cvqyy4t6vnbl2fjtego4ejmpcia77jhhwnksmm4bejm"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeignui4g7l6cvqyy4t6vnbl2fjtego4ejmpcia77jhhwnksmm4bejm", // hamtDir
+				"bafybeiccgo7euew77gkqkhezn3pozfrciiibqz2u3spdqmgjvd5wqskipm",
+				"bafybeihjydob4eq5j4m43whjgf5cgftthc42kjno3g24sa3wcw7vonbmfy",
+			}); err != nil {
+				panic(err)
+			}
+		case 4:
+			// Expect a request for one of the directory items and give it
+			expectedUri := "/ipfs/bafkreidqhbqn5htm5qejxpb3hps7dookudo3nncfn6al6niqibi5lq6fee"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafkreidqhbqn5htm5qejxpb3hps7dookudo3nncfn6al6niqibi5lq6fee", // exampleC
+			}); err != nil {
+				panic(err)
+			}
+		case 5:
+			// Expect a request for the multiblock file in the directory and give some of it
+			expectedUri := "/ipfs/bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa", // exampleA
+				"bafkreie5noke3mb7hqxukzcy73nl23k6lxszxi5w3dtmuwz62wnvkpsscm",
+				"bafkreih4ephajybraj6wnxsbwjwa77fukurtpl7oj7t7pfq545duhot7cq",
+				"bafkreigu7buvm3cfunb35766dn7tmqyh2um62zcio63en2btvxuybgcpue",
+			}); err != nil {
+				panic(err)
+			}
+		case 6:
+			// Expect a request for the rest of the multiblock file in the directory and give it
+			expectedUri := "/ipfs/bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa?format=car&dag-scope=entity&entity-bytes=768:*"
+			if request.RequestURI != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa", // exampleA
+				"bafkreicll3huefkc3qnrzeony7zcfo7cr3nbx64hnxrqzsixpceg332fhe",
+				"bafkreifst3pqztuvj57lycamoi7z34b4emf7gawxs74nwrc2c7jncmpaqm",
+			}); err != nil {
+				panic(err)
+			}
+		default:
+			t.Fatal("unsupported request number")
+		}
+	}))
+	defer s.Close()
+
+	bs, err := NewRemoteCarFetcher([]string{s.URL}, nil)
+	require.NoError(t, err)
+	fetcher, err := NewRetryCarFetcher(bs, 3)
+	require.NoError(t, err)
+
+	backend, err := NewCarBackend(fetcher)
+	require.NoError(t, err)
+
+	p, err := path.Join(path.FromCid(cid.MustParse("bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi")), "hamtDir")
+	require.NoError(t, err)
+
+	imPath, err := path.NewImmutablePath(p)
+	require.NoError(t, err)
+
+	_, nd, err := backend.GetAll(ctx, imPath)
+	require.NoError(t, err)
+
+	assertNextEntryNameEquals := func(t *testing.T, dirIter files.DirIterator, expectedName string) {
+		t.Helper()
+		require.True(t, dirIter.Next())
+		require.Equal(t, expectedName, dirIter.Name())
+	}
+
+	robs, err := carbs.NewReadOnly(bytes.NewReader(dirWithMultiblockHAMTandFiles), nil)
+	require.NoError(t, err)
+
+	dsrv := merkledag.NewDAGService(blockservice.New(robs, offline.Exchange(robs)))
+	assertFileEqual := func(t *testing.T, expectedCidString string, receivedFile files.File) {
+		t.Helper()
+
+		expected := cid.MustParse(expectedCidString)
+		receivedFileData, err := io.ReadAll(receivedFile)
+		require.NoError(t, err)
+		nd, err := dsrv.Get(ctx, expected)
+		require.NoError(t, err)
+		expectedFile, err := unixfile.NewUnixfsFile(ctx, dsrv, nd)
+		require.NoError(t, err)
+
+		expectedFileData, err := io.ReadAll(expectedFile.(files.File))
+		require.NoError(t, err)
+		require.True(t, bytes.Equal(expectedFileData, receivedFileData))
+	}
+
+	hamtDirIter := nd.(files.Directory).Entries()
+
+	assertNextEntryNameEquals(t, hamtDirIter, "exampleB")
+	assertFileEqual(t, "bafybeid3trcauvcp7fxaai23gkz3qexmlfxnnejgwm57hdvre472dafvha", hamtDirIter.Node().(files.File))
+
+	assertNextEntryNameEquals(t, hamtDirIter, "exampleD-hamt-collide-exampleB-seed-364")
+	assertFileEqual(t, "bafkreih2grj7p2bo5yk2guqazxfjzapv6hpm3mwrinv6s3cyayd72ke5he", hamtDirIter.Node().(files.File))
+
+	assertNextEntryNameEquals(t, hamtDirIter, "exampleC-hamt-collide-exampleA-seed-52")
+	assertFileEqual(t, "bafkreidqhbqn5htm5qejxpb3hps7dookudo3nncfn6al6niqibi5lq6fee", hamtDirIter.Node().(files.File))
+
+	assertNextEntryNameEquals(t, hamtDirIter, "exampleA")
+	assertFileEqual(t, "bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa", hamtDirIter.Node().(files.File))
+
+	require.False(t, hamtDirIter.Next())
+}
+
+func sendBlocks(ctx context.Context, carFixture []byte, writer io.Writer, cidStrList []string) error {
+	rd, err := storage.OpenReadable(bytes.NewReader(carFixture))
+	if err != nil {
+		return err
+	}
+
+	cw, err := storage.NewWritable(writer, []cid.Cid{cid.MustParse("bafkqaaa")}, carv2.WriteAsCarV1(true), carv2.AllowDuplicatePuts(true))
+	if err != nil {
+		return err
+	}
+
+	for _, s := range cidStrList {
+		c := cid.MustParse(s)
+		blockData, err := rd.Get(ctx, c.KeyString())
+		if err != nil {
+			return err
+		}
+
+		if err := cw.Put(ctx, c.KeyString(), blockData); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func TestCarBackendGetFile(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	requestNum := 0
+	s := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		requestNum++
+		switch requestNum {
+		case 1:
+			// Expect the full request, but return one that terminates in the middle of the path
+			expectedUri := "/ipfs/bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi/hamtDir/exampleA"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi", // root dir
+			}); err != nil {
+				panic(err)
+			}
+		case 2:
+			// Expect the full request, but return one that terminates in the middle of the file
+			// Note: this is an implementation detail, it could be in the future that we request less data (e.g. partial path)
+			expectedUri := "/ipfs/bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi/hamtDir/exampleA"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi", // root dir
+				"bafybeignui4g7l6cvqyy4t6vnbl2fjtego4ejmpcia77jhhwnksmm4bejm", // hamt root
+			}); err != nil {
+				panic(err)
+			}
+
+		case 3:
+			// Expect the full request and return the path and most of the file
+			// Note: this is an implementation detail, it could be in the future that we request less data (e.g. partial path and file range)
+			expectedUri := "/ipfs/bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi/hamtDir/exampleA"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi", // root dir
+				"bafybeignui4g7l6cvqyy4t6vnbl2fjtego4ejmpcia77jhhwnksmm4bejm", // hamt root
+				"bafybeihjydob4eq5j4m43whjgf5cgftthc42kjno3g24sa3wcw7vonbmfy", // inner hamt
+				"bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa", // file root
+				"bafkreie5noke3mb7hqxukzcy73nl23k6lxszxi5w3dtmuwz62wnvkpsscm", // file chunks start here
+				"bafkreih4ephajybraj6wnxsbwjwa77fukurtpl7oj7t7pfq545duhot7cq",
+			}); err != nil {
+				panic(err)
+			}
+
+		case 4:
+			// Expect a request for the remainder of the file
+			// Note: this is an implementation detail, it could be that the requester really asks for more information
+			expectedUri := "/ipfs/bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa", // file root
+				"bafkreigu7buvm3cfunb35766dn7tmqyh2um62zcio63en2btvxuybgcpue", // middle of the file starts here
+				"bafkreicll3huefkc3qnrzeony7zcfo7cr3nbx64hnxrqzsixpceg332fhe",
+				"bafkreifst3pqztuvj57lycamoi7z34b4emf7gawxs74nwrc2c7jncmpaqm",
+			}); err != nil {
+				panic(err)
+			}
+
+		default:
+			t.Fatal("unsupported request number")
+		}
+	}))
+	defer s.Close()
+
+	bs, err := NewRemoteCarFetcher([]string{s.URL}, nil)
+	require.NoError(t, err)
+	fetcher, err := NewRetryCarFetcher(bs, 3)
+	require.NoError(t, err)
+
+	backend, err := NewCarBackend(fetcher)
+	require.NoError(t, err)
+
+	trustedGatewayServer := httptest.NewServer(NewHandler(Config{DeserializedResponses: true}, backend))
+	defer trustedGatewayServer.Close()
+
+	resp, err := http.Get(trustedGatewayServer.URL + "/ipfs/bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi/hamtDir/exampleA")
+	require.NoError(t, err)
+
+	data, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+
+	robs, err := carbs.NewReadOnly(bytes.NewReader(dirWithMultiblockHAMTandFiles), nil)
+	require.NoError(t, err)
+
+	dsrv := merkledag.NewDAGService(blockservice.New(robs, offline.Exchange(robs)))
+	fileRootNd, err := dsrv.Get(ctx, cid.MustParse("bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa"))
+	require.NoError(t, err)
+	uio, err := unixfile.NewUnixfsFile(ctx, dsrv, fileRootNd)
+	require.NoError(t, err)
+	f := uio.(files.File)
+	expectedFileData, err := io.ReadAll(f)
+	require.NoError(t, err)
+	require.True(t, bytes.Equal(data, expectedFileData))
+}
+
+func TestCarBackendGetFileRangeRequest(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	requestNum := 0
+	s := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		requestNum++
+		switch requestNum {
+		case 1:
+			// Expect the full request, but return one that terminates at the root block
+			expectedUri := "/ipfs/bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa", // file root
+			}); err != nil {
+				panic(err)
+			}
+		case 2:
+			// Expect the full request, and return the whole file which should be invalid
+			expectedUri := "/ipfs/bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa", // file root
+				"bafkreie5noke3mb7hqxukzcy73nl23k6lxszxi5w3dtmuwz62wnvkpsscm", // file chunks start here
+				"bafkreih4ephajybraj6wnxsbwjwa77fukurtpl7oj7t7pfq545duhot7cq",
+				"bafkreigu7buvm3cfunb35766dn7tmqyh2um62zcio63en2btvxuybgcpue",
+				"bafkreicll3huefkc3qnrzeony7zcfo7cr3nbx64hnxrqzsixpceg332fhe",
+				"bafkreifst3pqztuvj57lycamoi7z34b4emf7gawxs74nwrc2c7jncmpaqm",
+			}); err != nil {
+				panic(err)
+			}
+		case 3:
+			// Expect the full request and return the first block
+			expectedUri := "/ipfs/bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa", // file root
+				"bafkreih4ephajybraj6wnxsbwjwa77fukurtpl7oj7t7pfq545duhot7cq",
+			}); err != nil {
+				panic(err)
+			}
+
+		case 4:
+			// Expect a request for the remainder of the file
+			// Note: this is an implementation detail, it could be that the requester really asks for more information
+			expectedUri := "/ipfs/bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa", // file root
+				"bafkreigu7buvm3cfunb35766dn7tmqyh2um62zcio63en2btvxuybgcpue",
+				"bafkreicll3huefkc3qnrzeony7zcfo7cr3nbx64hnxrqzsixpceg332fhe",
+			}); err != nil {
+				panic(err)
+			}
+
+		default:
+			t.Fatal("unsupported request number")
+		}
+	}))
+	defer s.Close()
+
+	bs, err := NewRemoteCarFetcher([]string{s.URL}, nil)
+	require.NoError(t, err)
+	fetcher, err := NewRetryCarFetcher(bs, 3)
+	require.NoError(t, err)
+
+	backend, err := NewCarBackend(fetcher)
+	require.NoError(t, err)
+
+	trustedGatewayServer := httptest.NewServer(NewHandler(Config{DeserializedResponses: true}, backend))
+	defer trustedGatewayServer.Close()
+
+	req, err := http.NewRequestWithContext(ctx, "GET", trustedGatewayServer.URL+"/ipfs/bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa", nil)
+	require.NoError(t, err)
+	startIndex := 256
+	endIndex := 750
+	req.Header.Set("Range", fmt.Sprintf("bytes=%d-%d", startIndex, endIndex))
+	resp, err := http.DefaultClient.Do(req)
+	require.NoError(t, err)
+
+	data, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+
+	robs, err := carbs.NewReadOnly(bytes.NewReader(dirWithMultiblockHAMTandFiles), nil)
+	require.NoError(t, err)
+
+	dsrv := merkledag.NewDAGService(blockservice.New(robs, offline.Exchange(robs)))
+	fileRootNd, err := dsrv.Get(ctx, cid.MustParse("bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa"))
+	require.NoError(t, err)
+	uio, err := unixfile.NewUnixfsFile(ctx, dsrv, fileRootNd)
+	require.NoError(t, err)
+	f := uio.(files.File)
+	_, err = f.Seek(int64(startIndex), io.SeekStart)
+	require.NoError(t, err)
+	expectedFileData, err := io.ReadAll(io.LimitReader(f, int64(endIndex)-int64(startIndex)+1))
+	require.NoError(t, err)
+	require.True(t, bytes.Equal(data, expectedFileData))
+	require.Equal(t, 4, requestNum)
+}
+
+func TestCarBackendGetFileWithBadBlockReturned(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	requestNum := 0
+	s := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		requestNum++
+		switch requestNum {
+		case 1:
+			// Expect the full request, but return one that terminates at the root block
+			expectedUri := "/ipfs/bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa", // file root
+			}); err != nil {
+				panic(err)
+			}
+		case 2:
+			// Expect the full request, but return a totally unrelated block
+			expectedUri := "/ipfs/bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi", // file root
+			}); err != nil {
+				panic(err)
+			}
+		case 3:
+			// Expect the full request and return most of the file
+			// Note: this is an implementation detail, it could be in the future that we request less data (e.g. partial path and file range)
+			expectedUri := "/ipfs/bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa", // file root
+				"bafkreie5noke3mb7hqxukzcy73nl23k6lxszxi5w3dtmuwz62wnvkpsscm", // file chunks start here
+				"bafkreih4ephajybraj6wnxsbwjwa77fukurtpl7oj7t7pfq545duhot7cq",
+			}); err != nil {
+				panic(err)
+			}
+
+		case 4:
+			// Expect a request for the remainder of the file
+			// Note: this is an implementation detail, it could be that the requester really asks for more information
+			expectedUri := "/ipfs/bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa", // file root
+				"bafkreigu7buvm3cfunb35766dn7tmqyh2um62zcio63en2btvxuybgcpue", // middle of the file starts here
+				"bafkreicll3huefkc3qnrzeony7zcfo7cr3nbx64hnxrqzsixpceg332fhe",
+				"bafkreifst3pqztuvj57lycamoi7z34b4emf7gawxs74nwrc2c7jncmpaqm",
+			}); err != nil {
+				panic(err)
+			}
+
+		default:
+			t.Fatal("unsupported request number")
+		}
+	}))
+	defer s.Close()
+
+	bs, err := NewRemoteCarFetcher([]string{s.URL}, nil)
+	require.NoError(t, err)
+	fetcher, err := NewRetryCarFetcher(bs, 3)
+	require.NoError(t, err)
+
+	backend, err := NewCarBackend(fetcher)
+	require.NoError(t, err)
+
+	trustedGatewayServer := httptest.NewServer(NewHandler(Config{DeserializedResponses: true}, backend))
+	defer trustedGatewayServer.Close()
+
+	resp, err := http.Get(trustedGatewayServer.URL + "/ipfs/bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa")
+	require.NoError(t, err)
+
+	data, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+
+	robs, err := carbs.NewReadOnly(bytes.NewReader(dirWithMultiblockHAMTandFiles), nil)
+	require.NoError(t, err)
+
+	dsrv := merkledag.NewDAGService(blockservice.New(robs, offline.Exchange(robs)))
+	fileRootNd, err := dsrv.Get(ctx, cid.MustParse("bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa"))
+	require.NoError(t, err)
+	uio, err := unixfile.NewUnixfsFile(ctx, dsrv, fileRootNd)
+	require.NoError(t, err)
+	f := uio.(files.File)
+	expectedFileData, err := io.ReadAll(f)
+	require.NoError(t, err)
+	require.True(t, bytes.Equal(data, expectedFileData))
+}
+
+func TestCarBackendGetHAMTDirectory(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	requestNum := 0
+	s := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		requestNum++
+		fmt.Println(requestNum, request.URL.Path)
+		switch requestNum {
+		case 1:
+			// Expect the full request, but return one that terminates in the middle of the path
+			expectedUri := "/ipfs/bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi/hamtDir/"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi", // root dir
+			}); err != nil {
+				panic(err)
+			}
+		case 2:
+			// Expect the full request, but return one that terminates in the middle of the HAMT
+			// Note: this is an implementation detail, it could be in the future that we request less data (e.g. partial path)
+			expectedUri := "/ipfs/bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi/hamtDir/"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi", // root dir
+				"bafybeignui4g7l6cvqyy4t6vnbl2fjtego4ejmpcia77jhhwnksmm4bejm", // hamt root
+				"bafybeiccgo7euew77gkqkhezn3pozfrciiibqz2u3spdqmgjvd5wqskipm", // inner hamt nodes start here
+			}); err != nil {
+				panic(err)
+			}
+		case 3:
+			// Expect a request for a non-existent index.html file
+			// Note: this is an implementation detail related to the directory request above
+			// Note: the order of cases 3 and 4 here are implementation specific as well
+			expectedUri := "/ipfs/bafybeignui4g7l6cvqyy4t6vnbl2fjtego4ejmpcia77jhhwnksmm4bejm/index.html"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeignui4g7l6cvqyy4t6vnbl2fjtego4ejmpcia77jhhwnksmm4bejm", // hamt root
+				"bafybeiccgo7euew77gkqkhezn3pozfrciiibqz2u3spdqmgjvd5wqskipm", // inner hamt nodes start here
+			}); err != nil {
+				panic(err)
+			}
+		case 4:
+			// Expect a request for the full HAMT and return it
+			// Note: this is an implementation detail, it could be in the future that we request more or less data
+			// (e.g. ask for the full path, ask for index.html first, make a spec change to allow asking for index.html with a fallback to the directory, etc.)
+			expectedUri := "/ipfs/bafybeignui4g7l6cvqyy4t6vnbl2fjtego4ejmpcia77jhhwnksmm4bejm"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeignui4g7l6cvqyy4t6vnbl2fjtego4ejmpcia77jhhwnksmm4bejm", // hamt root
+				"bafybeiccgo7euew77gkqkhezn3pozfrciiibqz2u3spdqmgjvd5wqskipm", // inner hamt nodes start here
+				"bafybeihjydob4eq5j4m43whjgf5cgftthc42kjno3g24sa3wcw7vonbmfy",
+			}); err != nil {
+				panic(err)
+			}
+
+		default:
+			t.Fatal("unsupported request number")
+		}
+	}))
+	defer s.Close()
+
+	bs, err := NewRemoteCarFetcher([]string{s.URL}, nil)
+	require.NoError(t, err)
+	fetcher, err := NewRetryCarFetcher(bs, 3)
+	require.NoError(t, err)
+
+	backend, err := NewCarBackend(fetcher)
+	require.NoError(t, err)
+
+	trustedGatewayServer := httptest.NewServer(NewHandler(Config{DeserializedResponses: true}, backend))
+	defer trustedGatewayServer.Close()
+
+	resp, err := http.Get(trustedGatewayServer.URL + "/ipfs/bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi/hamtDir/")
+	require.NoError(t, err)
+
+	data, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+
+	if strings.Count(string(data), ">exampleD-hamt-collide-exampleB-seed-364<") == 1 &&
+		strings.Count(string(data), ">exampleC-hamt-collide-exampleA-seed-52<") == 1 &&
+		strings.Count(string(data), ">exampleA<") == 1 &&
+		strings.Count(string(data), ">exampleB<") == 1 {
+		return
+	}
+	t.Fatal("directory does not contain the expected links")
+}
+
+func TestCarBackendGetCAR(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	requestNum := 0
+	s := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		requestNum++
+		switch requestNum {
+		case 1:
+			// Expect the full request, but return one that terminates in the middle of the path
+			expectedUri := "/ipfs/bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi", // root dir
+			}); err != nil {
+				panic(err)
+			}
+		case 2:
+			// Expect the full request, but return one that terminates in the middle of the HAMT
+			// Note: this is an implementation detail, it could be in the future that we request less data (e.g. partial path)
+			expectedUri := "/ipfs/bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi", // root dir
+				"bafybeignui4g7l6cvqyy4t6vnbl2fjtego4ejmpcia77jhhwnksmm4bejm", // hamt root
+			}); err != nil {
+				panic(err)
+			}
+
+		case 3:
+			// Expect the full request and return the full HAMT
+			// Note: this is an implementation detail, it could be in the future that we request less data (e.g. requesting the blocks to fill out the HAMT, or with spec changes asking for HAMT ranges)
+			expectedUri := "/ipfs/bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi"
+			if request.URL.Path != expectedUri {
+				panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+			}
+
+			if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+				"bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi", // root dir
+				"bafybeifdv255wmsrh75vcsrtkcwyktvewgihegeeyhhj2ju4lzt4lqfoze", // basicDir
+				"bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa", // exampleA
+				"bafkreie5noke3mb7hqxukzcy73nl23k6lxszxi5w3dtmuwz62wnvkpsscm",
+				"bafkreih4ephajybraj6wnxsbwjwa77fukurtpl7oj7t7pfq545duhot7cq",
+				"bafkreigu7buvm3cfunb35766dn7tmqyh2um62zcio63en2btvxuybgcpue",
+				"bafkreicll3huefkc3qnrzeony7zcfo7cr3nbx64hnxrqzsixpceg332fhe",
+				"bafkreifst3pqztuvj57lycamoi7z34b4emf7gawxs74nwrc2c7jncmpaqm",
+				"bafybeid3trcauvcp7fxaai23gkz3qexmlfxnnejgwm57hdvre472dafvha", // exampleB
+				"bafkreihgbi345degbcyxaf5b3boiyiaxhnuxdysvqmbdyaop2swmhh3s3m",
+				"bafkreiaugmh5gal5rgiems6gslcdt2ixfncahintrmcqvrgxqamwtlrmz4",
+				"bafkreiaxwwb7der2qvmteymgtlj7ww7w5vc44phdxfnexog3vnuwdkxuea",
+				"bafkreic5zyan5rk4ccfum4d4mu3h5eqsllbudlj4texlzj6xdgxvldzngi",
+				"bafybeignui4g7l6cvqyy4t6vnbl2fjtego4ejmpcia77jhhwnksmm4bejm", // hamtDir
+				"bafybeiccgo7euew77gkqkhezn3pozfrciiibqz2u3spdqmgjvd5wqskipm",
+				"bafybeid3trcauvcp7fxaai23gkz3qexmlfxnnejgwm57hdvre472dafvha", // exampleB
+				"bafkreihgbi345degbcyxaf5b3boiyiaxhnuxdysvqmbdyaop2swmhh3s3m",
+				"bafkreiaugmh5gal5rgiems6gslcdt2ixfncahintrmcqvrgxqamwtlrmz4",
+				"bafkreiaxwwb7der2qvmteymgtlj7ww7w5vc44phdxfnexog3vnuwdkxuea",
+				"bafkreic5zyan5rk4ccfum4d4mu3h5eqsllbudlj4texlzj6xdgxvldzngi",
+				"bafkreih2grj7p2bo5yk2guqazxfjzapv6hpm3mwrinv6s3cyayd72ke5he", // exampleD
+				"bafybeihjydob4eq5j4m43whjgf5cgftthc42kjno3g24sa3wcw7vonbmfy",
+				"bafkreidqhbqn5htm5qejxpb3hps7dookudo3nncfn6al6niqibi5lq6fee", // exampleC
+				"bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa", // exampleA
+				"bafkreie5noke3mb7hqxukzcy73nl23k6lxszxi5w3dtmuwz62wnvkpsscm",
+				"bafkreih4ephajybraj6wnxsbwjwa77fukurtpl7oj7t7pfq545duhot7cq",
+				"bafkreigu7buvm3cfunb35766dn7tmqyh2um62zcio63en2btvxuybgcpue",
+				"bafkreicll3huefkc3qnrzeony7zcfo7cr3nbx64hnxrqzsixpceg332fhe",
+				"bafkreifst3pqztuvj57lycamoi7z34b4emf7gawxs74nwrc2c7jncmpaqm",
+			}); err != nil {
+				panic(err)
+			}
+
+		default:
+			t.Fatal("unsupported request number")
+		}
+	}))
+	defer s.Close()
+
+	bs, err := NewRemoteCarFetcher([]string{s.URL}, nil)
+	require.NoError(t, err)
+	fetcher, err := NewRetryCarFetcher(bs, 3)
+	require.NoError(t, err)
+
+	backend, err := NewCarBackend(fetcher)
+	require.NoError(t, err)
+
+	p := path.FromCid(cid.MustParse("bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi"))
+	var carReader io.Reader
+	_, carReader, err = backend.GetCAR(ctx, p, CarParams{Scope: DagScopeAll})
+	require.NoError(t, err)
+
+	carBytes, err := io.ReadAll(carReader)
+	require.NoError(t, err)
+	carReader = bytes.NewReader(carBytes)
+
+	blkReader, err := carv2.NewBlockReader(carReader)
+	require.NoError(t, err)
+
+	responseCarBlock := []string{
+		"bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi", // root dir
+		"bafybeifdv255wmsrh75vcsrtkcwyktvewgihegeeyhhj2ju4lzt4lqfoze", // basicDir
+		"bafybeigcisqd7m5nf3qmuvjdbakl5bdnh4ocrmacaqkpuh77qjvggmt2sa", // exampleA
+		"bafkreie5noke3mb7hqxukzcy73nl23k6lxszxi5w3dtmuwz62wnvkpsscm",
+		"bafkreih4ephajybraj6wnxsbwjwa77fukurtpl7oj7t7pfq545duhot7cq",
+		"bafkreigu7buvm3cfunb35766dn7tmqyh2um62zcio63en2btvxuybgcpue",
+		"bafkreicll3huefkc3qnrzeony7zcfo7cr3nbx64hnxrqzsixpceg332fhe",
+		"bafkreifst3pqztuvj57lycamoi7z34b4emf7gawxs74nwrc2c7jncmpaqm",
+		"bafybeid3trcauvcp7fxaai23gkz3qexmlfxnnejgwm57hdvre472dafvha", // exampleB
+		"bafkreihgbi345degbcyxaf5b3boiyiaxhnuxdysvqmbdyaop2swmhh3s3m",
+		"bafkreiaugmh5gal5rgiems6gslcdt2ixfncahintrmcqvrgxqamwtlrmz4",
+		"bafkreiaxwwb7der2qvmteymgtlj7ww7w5vc44phdxfnexog3vnuwdkxuea",
+		"bafkreic5zyan5rk4ccfum4d4mu3h5eqsllbudlj4texlzj6xdgxvldzngi",
+		"bafybeignui4g7l6cvqyy4t6vnbl2fjtego4ejmpcia77jhhwnksmm4bejm", // hamtDir
+		"bafybeiccgo7euew77gkqkhezn3pozfrciiibqz2u3spdqmgjvd5wqskipm",
+		"bafkreih2grj7p2bo5yk2guqazxfjzapv6hpm3mwrinv6s3cyayd72ke5he", // exampleD
+		"bafybeihjydob4eq5j4m43whjgf5cgftthc42kjno3g24sa3wcw7vonbmfy",
+		"bafkreidqhbqn5htm5qejxpb3hps7dookudo3nncfn6al6niqibi5lq6fee", // exampleC
+	}
+
+	for i := 0; i < len(responseCarBlock); i++ {
+		expectedCid := cid.MustParse(responseCarBlock[i])
+		blk, err := blkReader.Next()
+		require.NoError(t, err)
+		require.True(t, blk.Cid().Equals(expectedCid))
+	}
+	_, err = blkReader.Next()
+	require.ErrorIs(t, err, io.EOF)
+}
+
+func TestCarBackendPassthroughErrors(t *testing.T) {
+	t.Run("PathTraversalError", func(t *testing.T) {
+		pathTraversalTest := func(t *testing.T, traversal func(ctx context.Context, p path.ImmutablePath, backend *CarBackend) error) {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			var requestNum int
+			s := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+				requestNum++
+				switch requestNum {
+				case 1:
+					// Expect the full request, but return one that terminates in the middle of the path
+					expectedUri := "/ipfs/bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi/hamtDir/exampleA"
+					if request.URL.Path != expectedUri {
+						panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+					}
+
+					if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+						"bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi", // root dir
+					}); err != nil {
+						panic(err)
+					}
+				case 2:
+					// Expect the full request, but return one that terminates in the middle of the file
+					// Note: this is an implementation detail, it could be in the future that we request less data (e.g. partial path)
+					expectedUri := "/ipfs/bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi/hamtDir/exampleA"
+					if request.URL.Path != expectedUri {
+						panic(fmt.Errorf("expected URI %s, got %s", expectedUri, request.RequestURI))
+					}
+
+					if err := sendBlocks(ctx, dirWithMultiblockHAMTandFiles, writer, []string{
+						"bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi", // root dir
+						"bafybeignui4g7l6cvqyy4t6vnbl2fjtego4ejmpcia77jhhwnksmm4bejm", // hamt root
+					}); err != nil {
+						panic(err)
+					}
+				default:
+					t.Fatal("unsupported request number")
+				}
+			}))
+			defer s.Close()
+
+			bs, err := NewRemoteCarFetcher([]string{s.URL}, nil)
+			require.NoError(t, err)
+
+			p, err := path.NewPath("/ipfs/bafybeid3fd2xxdcd3dbj7trb433h2aqssn6xovjbwnkargjv7fuog4xjdi/hamtDir/exampleA")
+			require.NoError(t, err)
+
+			imPath, err := path.NewImmutablePath(p)
+			require.NoError(t, err)
+
+			bogusErr := NewErrorStatusCode(fmt.Errorf("this is a test error"), 418)
+
+			clientRequestNum := 0
+
+			fetcher, err := NewRetryCarFetcher(&fetcherWrapper{fn: func(ctx context.Context, path path.ImmutablePath, params CarParams, cb DataCallback) error {
+				clientRequestNum++
+				if clientRequestNum > 2 {
+					return bogusErr
+				}
+				return bs.Fetch(ctx, path, params, cb)
+			}}, 3)
+			require.NoError(t, err)
+
+			backend, err := NewCarBackend(fetcher)
+			require.NoError(t, err)
+
+			err = traversal(ctx, imPath, backend)
+			parsedErr := &ErrorStatusCode{}
+			if errors.As(err, &parsedErr) {
+				if parsedErr.StatusCode == bogusErr.StatusCode {
+					return
+				}
+			}
+			t.Fatal("error did not pass through")
+		}
+		t.Run("Block", func(t *testing.T) {
+			pathTraversalTest(t, func(ctx context.Context, p path.ImmutablePath, backend *CarBackend) error {
+				_, _, err := backend.GetBlock(ctx, p)
+				return err
+			})
+		})
+		t.Run("File", func(t *testing.T) {
+			pathTraversalTest(t, func(ctx context.Context, p path.ImmutablePath, backend *CarBackend) error {
+				_, _, err := backend.Get(ctx, p)
+				return err
+			})
+		})
+	})
+}
+
+type fetcherWrapper struct {
+	fn func(ctx context.Context, path path.ImmutablePath, params CarParams, cb DataCallback) error
+}
+
+func (w *fetcherWrapper) Fetch(ctx context.Context, path path.ImmutablePath, params CarParams, cb DataCallback) error {
+	return w.fn(ctx, path, params, cb)
+}
+
+type testErr struct {
+	message    string
+	retryAfter time.Duration
+}
+
+func (e *testErr) Error() string {
+	return e.message
+}
+
+func (e *testErr) RetryAfter() time.Duration {
+	return e.retryAfter
+}
+
+func TestGatewayErrorRetryAfter(t *testing.T) {
+	originalErr := &testErr{message: "test", retryAfter: time.Minute}
+	var (
+		convertedErr error
+		gatewayErr   *ErrorRetryAfter
+	)
+
+	// Test unwrapped
+	convertedErr = blockstoreErrToGatewayErr(originalErr)
+	ok := errors.As(convertedErr, &gatewayErr)
+	assert.True(t, ok)
+	assert.EqualValues(t, originalErr.retryAfter, gatewayErr.RetryAfter)
+
+	// Test wrapped.
+	convertedErr = blockstoreErrToGatewayErr(fmt.Errorf("wrapped error: %w", originalErr))
+	ok = errors.As(convertedErr, &gatewayErr)
+	assert.True(t, ok)
+	assert.EqualValues(t, originalErr.retryAfter, gatewayErr.RetryAfter)
+}
diff --git a/gateway/gateway_test.go b/gateway/gateway_test.go
index 031a184a5..289faad01 100644
--- a/gateway/gateway_test.go
+++ b/gateway/gateway_test.go
@@ -20,7 +20,7 @@ import (
 )
 
 func TestGatewayGet(t *testing.T) {
-	ts, backend, root := newTestServerAndNode(t, nil, "fixtures.car")
+	ts, backend, root := newTestServerAndNode(t, "fixtures.car")
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -96,7 +96,7 @@ func TestGatewayGet(t *testing.T) {
 func TestHeaders(t *testing.T) {
 	t.Parallel()
 
-	ts, backend, root := newTestServerAndNode(t, nil, "headers-test.car")
+	ts, backend, root := newTestServerAndNode(t, "headers-test.car")
 
 	var (
 		rootCID = "bafybeidbcy4u6y55gsemlubd64zk53xoxs73ifd6rieejxcr7xy46mjvky"
@@ -121,7 +121,7 @@ func TestHeaders(t *testing.T) {
 	t.Run("Cache-Control uses TTL for /ipns/ when it is known", func(t *testing.T) {
 		t.Parallel()
 
-		ts, backend, root := newTestServerAndNode(t, nil, "ipns-hostname-redirects.car")
+		ts, backend, root := newTestServerAndNode(t, "ipns-hostname-redirects.car")
 		backend.namesys["/ipns/example.net"] = newMockNamesysItem(path.FromCid(root), time.Second*30)
 		backend.namesys["/ipns/example.com"] = newMockNamesysItem(path.FromCid(root), time.Second*55)
 		backend.namesys["/ipns/unknown.com"] = newMockNamesysItem(path.FromCid(root), 0)
@@ -420,7 +420,7 @@ func TestHeaders(t *testing.T) {
 }
 
 func TestGoGetSupport(t *testing.T) {
-	ts, _, root := newTestServerAndNode(t, nil, "fixtures.car")
+	ts, _, root := newTestServerAndNode(t, "fixtures.car")
 
 	// mimic go-get
 	req := mustNewRequest(t, http.MethodGet, ts.URL+"/ipfs/"+root.String()+"?go-get=1", nil)
@@ -432,7 +432,7 @@ func TestRedirects(t *testing.T) {
 	t.Parallel()
 
 	t.Run("IPNS Base58 Multihash Redirect", func(t *testing.T) {
-		ts, _, _ := newTestServerAndNode(t, nil, "fixtures.car")
+		ts, _, _ := newTestServerAndNode(t, "fixtures.car")
 
 		t.Run("ED25519 Base58-encoded key", func(t *testing.T) {
 			t.Parallel()
@@ -453,7 +453,7 @@ func TestRedirects(t *testing.T) {
 
 	t.Run("URI Query Redirects", func(t *testing.T) {
 		t.Parallel()
-		ts, _, _ := newTestServerAndNode(t, mockNamesys{}, "fixtures.car")
+		ts, _, _ := newTestServerAndNode(t, "fixtures.car")
 
 		cid := "QmbWqxBEKC3P8tqsKc98xmWNzrzDtRLMiMPL8wBuTGsMnR"
 		for _, test := range []struct {
@@ -492,7 +492,7 @@ func TestRedirects(t *testing.T) {
 	t.Run("IPNS Hostname Redirects", func(t *testing.T) {
 		t.Parallel()
 
-		ts, backend, root := newTestServerAndNode(t, nil, "ipns-hostname-redirects.car")
+		ts, backend, root := newTestServerAndNode(t, "ipns-hostname-redirects.car")
 		backend.namesys["/ipns/example.net"] = newMockNamesysItem(path.FromCid(root), 0)
 
 		// make request to directory containing index.html
@@ -555,9 +555,11 @@ func TestRedirects(t *testing.T) {
 
 			// Check statuses and body.
 			require.Equal(t, http.StatusOK, res.StatusCode)
-			body, err := io.ReadAll(res.Body)
-			require.NoError(t, err)
-			require.Equal(t, "hello world\n", string(body))
+			if method != http.MethodHead {
+				body, err := io.ReadAll(res.Body)
+				require.NoError(t, err)
+				require.Equal(t, "hello world\n", string(body))
+			}
 
 			// Check Etag.
 			etag := res.Header.Get("Etag")
@@ -948,7 +950,7 @@ func TestPanicStatusCode(t *testing.T) {
 
 func TestBrowserErrorHTML(t *testing.T) {
 	t.Parallel()
-	ts, _, root := newTestServerAndNode(t, nil, "fixtures.car")
+	ts, _, root := newTestServerAndNode(t, "fixtures.car")
 
 	t.Run("plain error if request does not have Accept: text/html", func(t *testing.T) {
 		t.Parallel()
diff --git a/gateway/handler_unixfs_dir_test.go b/gateway/handler_unixfs_dir_test.go
index e44708687..5727d50c5 100644
--- a/gateway/handler_unixfs_dir_test.go
+++ b/gateway/handler_unixfs_dir_test.go
@@ -12,7 +12,7 @@ import (
 
 func TestIPNSHostnameBacklinks(t *testing.T) {
 	// Test if directory listing on DNSLink Websites have correct backlinks.
-	ts, backend, root := newTestServerAndNode(t, nil, "dir-special-chars.car")
+	ts, backend, root := newTestServerAndNode(t, "dir-special-chars.car")
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
diff --git a/gateway/testdata/directory-with-multilayer-hamt-and-multiblock-files.car b/gateway/testdata/directory-with-multilayer-hamt-and-multiblock-files.car
new file mode 100644
index 000000000..cb2a4875d
Binary files /dev/null and b/gateway/testdata/directory-with-multilayer-hamt-and-multiblock-files.car differ
diff --git a/gateway/utilities_test.go b/gateway/utilities_test.go
index 68db84041..22f5750fa 100644
--- a/gateway/utilities_test.go
+++ b/gateway/utilities_test.go
@@ -27,7 +27,7 @@ import (
 )
 
 func mustNewRequest(t *testing.T, method string, path string, body io.Reader) *http.Request {
-	r, err := http.NewRequest(http.MethodGet, path, body)
+	r, err := http.NewRequest(method, path, body)
 	require.NoError(t, err)
 	return r
 }
@@ -224,7 +224,7 @@ func (mb *mockBackend) resolvePathNoRootsReturned(ctx context.Context, ip path.P
 	return md.LastSegment, nil
 }
 
-func newTestServerAndNode(t *testing.T, ns mockNamesys, fixturesFile string) (*httptest.Server, *mockBackend, cid.Cid) {
+func newTestServerAndNode(t *testing.T, fixturesFile string) (*httptest.Server, *mockBackend, cid.Cid) {
 	backend, root := newMockBackend(t, fixturesFile)
 	ts := newTestServer(t, backend)
 	return ts, backend, root