feat: add header for skipping the blockcache

CHANGELOG.md

@@ -16,6 +16,7 @@ The following emojis are used to highlight certain changes:
 ### Added
 
 - Tracing per request with auth header (see `RAINBOW_TRACING_AUTH`) or a fraction of requests (see `RAINBOW_SAMPLING_FRACTION`)
+- Debugging with `Rainbow-No-Blockcache` that is gated by the `Authorization` header and does not use the local block cache for the request
 
 ### Changed
 

go.mod

@@ -15,6 +15,7 @@ require (
 	github.com/ipfs/go-datastore v0.6.0
 	github.com/ipfs/go-ds-badger4 v0.1.5
 	github.com/ipfs/go-ds-flatfs v0.5.1
+	github.com/ipfs/go-ds-leveldb v0.5.0
 	github.com/ipfs/go-ipfs-delay v0.0.1
 	github.com/ipfs/go-log/v2 v2.5.1
 	github.com/ipfs/go-metrics-interface v0.0.1
@@ -176,6 +177,7 @@ require (
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/samber/lo v1.39.0 // indirect
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
+	github.com/syndtr/goleveldb v1.0.0 // indirect
 	github.com/ucarion/urlpath v0.0.0-20200424170820-7ccc79b76bbb // indirect
 	github.com/whyrusleeping/base32 v0.0.0-20170828182744-c30ac30633cc // indirect
 	github.com/whyrusleeping/cbor v0.0.0-20171005072247-63513f603b11 // indirect

go.sum

@@ -263,6 +263,8 @@ github.com/ipfs/go-ds-badger4 v0.1.5/go.mod h1:LUU2FbhNdmhAbJmMeoahVRbe4GsduAODS
 github.com/ipfs/go-ds-flatfs v0.5.1 h1:ZCIO/kQOS/PSh3vcF1H6a8fkRGS7pOfwfPdx4n/KJH4=
 github.com/ipfs/go-ds-flatfs v0.5.1/go.mod h1:RWTV7oZD/yZYBKdbVIFXTX2fdY2Tbvl94NsWqmoyAX4=
 github.com/ipfs/go-ds-leveldb v0.1.0/go.mod h1:hqAW8y4bwX5LWcCtku2rFNX3vjDZCy5LZCg+cSZvYb8=
+github.com/ipfs/go-ds-leveldb v0.5.0 h1:s++MEBbD3ZKc9/8/njrn4flZLnCuY9I79v94gBUNumo=
+github.com/ipfs/go-ds-leveldb v0.5.0/go.mod h1:d3XG9RUDzQ6V4SHi8+Xgj9j1XuEk1z82lquxrVbml/Q=
 github.com/ipfs/go-ipfs-blockstore v1.3.1 h1:cEI9ci7V0sRNivqaOr0elDsamxXFxJMMMy7PTTDQNsQ=
 github.com/ipfs/go-ipfs-blockstore v1.3.1/go.mod h1:KgtZyc9fq+P2xJUiCAzbRdhhqJHvsw8u2Dlqy2MyRTE=
 github.com/ipfs/go-ipfs-blocksutil v0.0.1 h1:Eh/H4pc1hsvhzsQoMEP3Bke/aW5P5rVM1IWFJMcGIPQ=
@@ -491,8 +493,12 @@ github.com/multiformats/go-varint v0.0.7 h1:sWSGR+f/eu5ABZA2ZpYKBILXTTs9JWpdEM/n
 github.com/multiformats/go-varint v0.0.7/go.mod h1:r8PUYw/fD/SjBCiKOoDlGF6QawOELpZAu9eioSos/OU=
 github.com/neelance/astrewrite v0.0.0-20160511093645-99348263ae86/go.mod h1:kHJEU3ofeGjhHklVoIGuVj85JJwZ6kWPaJwCIxgnFmo=
 github.com/neelance/sourcemap v0.0.0-20151028013722-8c68805598ab/go.mod h1:Qr6/a/Q4r9LP1IltGz7tA7iOK1WonHEYhu1HRBA7ZiM=
+github.com/nxadm/tail v1.4.11 h1:8feyoE3OzPrcshW5/MJ4sGESc5cqmGkGCWlco4l0bqY=
+github.com/nxadm/tail v1.4.11/go.mod h1:OTaG3NK980DZzxbRq6lEuzgU+mug70nY11sMd4JXXHc=
 github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
+github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
+github.com/onsi/ginkgo v1.16.5/go.mod h1:+E8gABHa3K6zRBolWtd+ROzc/U5bkGt0FwiG042wbpU=
 github.com/onsi/ginkgo/v2 v2.19.0 h1:9Cnnf7UHo57Hy3k6/m5k3dRfGTMXGvxhHFvkDTCTpvA=
 github.com/onsi/ginkgo/v2 v2.19.0/go.mod h1:rlwLi9PilAFJ8jCg9UE1QP6VBpd6/xj3SRC0d6TU0To=
 github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
@@ -669,6 +675,7 @@ github.com/stretchr/testify v1.8.3/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXl
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
 github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/syndtr/goleveldb v1.0.0 h1:fBdIW9lB4Iz0n9khmH8w27SJ3QEJ7+IgjPEwGSZiFdE=
 github.com/syndtr/goleveldb v1.0.0/go.mod h1:ZVVdQEZoIme9iO1Ch2Jdy24qqXrMMOU6lpPAyBWyWuQ=
 github.com/tarm/serial v0.0.0-20180830185346-98f6abe2eb07/go.mod h1:kDXzergiv9cbyO7IOYJZWg1U88JhDg3PB6klq9Hg2pA=
 github.com/tj/assert v0.0.3 h1:Df/BlaZ20mq6kuai7f5z2TvPFiwC3xaWJSDQNiIS3Rk=
@@ -1016,6 +1023,7 @@ gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMy
 gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
 gopkg.in/src-d/go-cli.v0 v0.0.0-20181105080154-d492247bbc0d/go.mod h1:z+K8VcOYVYcSwSjGebuDL6176A1XskgbtNl64NSg+n8=
 gopkg.in/src-d/go-log.v1 v1.0.1/go.mod h1:GN34hKP0g305ysm2/hctJ0Y8nWP3zxXXJ8GFabTyABE=
+gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=

handler_test.go

@@ -1,8 +1,17 @@
 package main
 
 import (
+	"context"
+	"crypto/rand"
+	"io"
 	"net/http"
 	"testing"
+	"time"
+
+	bsnet "github.com/ipfs/boxo/bitswap/network"
+	bsserver "github.com/ipfs/boxo/bitswap/server"
+	"github.com/ipfs/go-metrics-interface"
+	"github.com/libp2p/go-libp2p"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -51,3 +60,59 @@ func TestTrustless(t *testing.T) {
 		assert.Equal(t, http.StatusOK, res.StatusCode)
 	})
 }
+
+func TestNoBlockcacheHeader(t *testing.T) {
+	ts, gnd := mustTestServer(t, Config{
+		Bitswap: true,
+	})
+
+	content := make([]byte, 1024)
+	_, err := rand.Read(content)
+	require.NoError(t, err)
+	cid := mustAddFile(t, gnd, content)
+	url := ts.URL + "/ipfs/" + cid.String()
+
+	t.Run("Successful download of data with standard already cached in the node", func(t *testing.T) {
+		req, err := http.NewRequest(http.MethodGet, url, nil)
+		require.NoError(t, err)
+
+		res, err := http.DefaultClient.Do(req)
+		assert.NoError(t, err)
+		assert.Equal(t, http.StatusOK, res.StatusCode)
+		responseBody, err := io.ReadAll(res.Body)
+		assert.NoError(t, err)
+		assert.Equal(t, content, responseBody)
+	})
+
+	t.Run("When caching is explicitly skipped the data is not found", func(t *testing.T) {
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+		req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+		require.NoError(t, err)
+
+		req.Header.Set(NoBlockcacheHeader, "true")
+		_, err = http.DefaultClient.Do(req)
+		assert.ErrorIs(t, err, context.DeadlineExceeded)
+	})
+
+	t.Run("When caching is explicitly skipped the data is found if a peer has it", func(t *testing.T) {
+		newHost, err := libp2p.New()
+		require.NoError(t, err)
+
+		ctx := context.Background()
+		// pacify metrics reporting code
+		ctx = metrics.CtxScope(ctx, "test.bsserver.host")
+		bs := bsserver.New(ctx, bsnet.NewFromIpfsHost(newHost, nil), gnd.blockstore)
+		defer bs.Close()
+
+		ctx, cancel := context.WithTimeout(ctx, time.Second)
+		defer cancel()
+		req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+		require.NoError(t, err)
+
+		req.Header.Set(NoBlockcacheHeader, "true")
+		_, err = http.DefaultClient.Do(req)
+		assert.ErrorIs(t, err, context.DeadlineExceeded)
+	})
+
+}

handlers.go

@@ -1,8 +1,11 @@
 package main
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/ipfs/boxo/blockstore"
+	leveldb "github.com/ipfs/go-ds-leveldb"
 	"net/http"
 	"os"
 	"runtime"
@@ -208,7 +211,7 @@ func setupGatewayHandler(cfg Config, nd *Node, tracingAuth string) (http.Handler
 	// Add tracing.
 	handler = otelhttp.NewHandler(handler, "Gateway")
 
-	// Remove tracing headers if not authorized
+	// Remove tracing and cache skipping headers if not authorized
 	prevHandler := handler
 	handler = http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
 		if request.Header.Get("Authorization") != tracingAuth {
@@ -218,13 +221,33 @@ func setupGatewayHandler(cfg Config, nd *Node, tracingAuth string) (http.Handler
 			if request.Header.Get("Tracestate") != "" {
 				request.Header.Del("Tracestate")
 			}
+			if request.Header.Get(NoBlockcacheHeader) != "" {
+				request.Header.Del(NoBlockcacheHeader)
+			}
+		}
+
+		// Process cache skipping header
+		if noBlockCache := request.Header.Get(NoBlockcacheHeader); noBlockCache != "" {
+			ds, err := leveldb.NewDatastore("", nil)
+			if err != nil {
+				writer.WriteHeader(http.StatusInternalServerError)
+				_, _ = writer.Write([]byte(err.Error()))
+				return
+			}
+			newCtx := context.WithValue(request.Context(), NoBlockcache{}, blockstore.NewBlockstore(ds))
+			request = request.WithContext(newCtx)
 		}
+
 		prevHandler.ServeHTTP(writer, request)
 	})
 
 	return handler, nil
 }
 
+const NoBlockcacheHeader = "Rainbow-No-Blockcache"
+
+type NoBlockcache struct{}
+
 // MutexFractionOption allows to set runtime.SetMutexProfileFraction via HTTP
 // using POST request with parameter 'fraction'.
 func MutexFractionOption(path string, mux *http.ServeMux) *http.ServeMux {

setup.go

@@ -22,6 +22,8 @@ import (
 	"github.com/ipfs/boxo/namesys"
 	"github.com/ipfs/boxo/path/resolver"
 	"github.com/ipfs/boxo/peering"
+	blocks "github.com/ipfs/go-block-format"
+	"github.com/ipfs/go-cid"
 	"github.com/ipfs/go-datastore"
 	badger4 "github.com/ipfs/go-ds-badger4"
 	flatfs "github.com/ipfs/go-ds-flatfs"
@@ -277,6 +279,10 @@ func SetupWithLibp2p(ctx context.Context, cfg Config, key crypto.PrivKey, dnsCac
 			// See also comment in blockservice.
 			blockstore.WriteThrough(),
 		)
+		blkst = &switchingBlockstore{
+			baseBlockstore:      blkst,
+			contextSwitchingKey: NoBlockcache{},
+		}
 		blkst = blockstore.NewIdStore(blkst)
 		n.blockstore = blkst
 
@@ -494,3 +500,50 @@ func setupNamesys(cfg Config, vs routing.ValueStore, blocker *nopfs.Blocker) (na
 	ns = nopfsipfs.WrapNameSystem(ns, blocker)
 	return ns, nil
 }
+
+type switchingBlockstore struct {
+	baseBlockstore      blockstore.Blockstore
+	contextSwitchingKey any
+}
+
+func (s *switchingBlockstore) getBlockstore(ctx context.Context) blockstore.Blockstore {
+	alternativeBlockstore, ok := ctx.Value(s.contextSwitchingKey).(blockstore.Blockstore)
+	if ok {
+		return alternativeBlockstore
+	}
+	return s.baseBlockstore
+}
+
+func (s *switchingBlockstore) DeleteBlock(ctx context.Context, c cid.Cid) error {
+	return s.getBlockstore(ctx).DeleteBlock(ctx, c)
+}
+
+func (s *switchingBlockstore) Has(ctx context.Context, c cid.Cid) (bool, error) {
+	return s.getBlockstore(ctx).Has(ctx, c)
+}
+
+func (s *switchingBlockstore) Get(ctx context.Context, c cid.Cid) (blocks.Block, error) {
+	return s.getBlockstore(ctx).Get(ctx, c)
+}
+
+func (s *switchingBlockstore) GetSize(ctx context.Context, c cid.Cid) (int, error) {
+	return s.getBlockstore(ctx).GetSize(ctx, c)
+}
+
+func (s *switchingBlockstore) Put(ctx context.Context, block blocks.Block) error {
+	return s.getBlockstore(ctx).Put(ctx, block)
+}
+
+func (s *switchingBlockstore) PutMany(ctx context.Context, blocks []blocks.Block) error {
+	return s.getBlockstore(ctx).PutMany(ctx, blocks)
+}
+
+func (s *switchingBlockstore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
+	return s.getBlockstore(ctx).AllKeysChan(ctx)
+}
+
+func (s *switchingBlockstore) HashOnRead(enabled bool) {
+	s.baseBlockstore.HashOnRead(enabled)
+}
+
+var _ blockstore.Blockstore = (*switchingBlockstore)(nil)