chore: update boxo and otlp for bitswap providing refactor

Keeps in sync with ipfs/boxo#528

Author: Jorropo

core/commands/bitswap.go

@@ -134,7 +134,6 @@ var bitswapStatCmd = &cmds.Command{
 			human, _ := req.Options[bitswapHumanOptionName].(bool)
 
 			fmt.Fprintln(w, "bitswap status")
-			fmt.Fprintf(w, "\tprovides buffer: %d / %d\n", s.ProvideBufLen, bitswap.HasBlockBufferSize)
 			fmt.Fprintf(w, "\tblocks received: %d\n", s.BlocksReceived)
 			fmt.Fprintf(w, "\tblocks sent: %d\n", s.BlocksSent)
 			if human {

core/commands/files.go

@@ -14,8 +14,6 @@ import (
 	"github.com/ipfs/kubo/core"
 	"github.com/ipfs/kubo/core/commands/cmdenv"
 
-	bservice "github.com/ipfs/boxo/blockservice"
-	offline "github.com/ipfs/boxo/exchange/offline"
 	dag "github.com/ipfs/boxo/ipld/merkledag"
 	ft "github.com/ipfs/boxo/ipld/unixfs"
 	mfs "github.com/ipfs/boxo/mfs"
@@ -162,11 +160,7 @@ var filesStatCmd = &cmds.Command{
 
 		var dagserv ipld.DAGService
 		if withLocal {
-			// an offline DAGService will not fetch from the network
-			dagserv = dag.NewDAGService(bservice.New(
-				node.Blockstore,
-				offline.Exchange(node.Blockstore),
-			))
+			dagserv = node.OfflineDAG
 		} else {
 			dagserv = node.DAG
 		}

core/commands/pin/pin.go

@@ -8,8 +8,6 @@ import (
 	"os"
 	"time"
 
-	bserv "github.com/ipfs/boxo/blockservice"
-	offline "github.com/ipfs/boxo/exchange/offline"
 	dag "github.com/ipfs/boxo/ipld/merkledag"
 	verifcid "github.com/ipfs/boxo/verifcid"
 	cid "github.com/ipfs/go-cid"
@@ -702,8 +700,7 @@ type pinVerifyOpts struct {
 func pinVerify(ctx context.Context, n *core.IpfsNode, opts pinVerifyOpts, enc cidenc.Encoder) (<-chan any, error) {
 	visited := make(map[cid.Cid]PinStatus)
 
-	bs := n.Blocks.Blockstore()
-	DAG := dag.NewDAGService(bserv.New(bs, offline.Exchange(bs)))
+	DAG := n.OfflineDAG
 	getLinks := dag.GetLinksWithDAG(DAG)
 
 	var checkPin func(root cid.Cid) PinStatus

core/core.go

@@ -81,7 +81,9 @@ type IpfsNode struct {
 	BaseBlocks                  node.BaseBlocks           // the raw blockstore, no filestore wrapping
 	GCLocker                    bstore.GCLocker           // the locker used to protect the blockstore during gc
 	Blocks                      bserv.BlockService        // the block service, get/add blocks.
+	OfflineBlocks               bserv.BlockService        `name:"offlineBlockService"` // blockservice which doesn't try to fetch from the network
 	DAG                         ipld.DAGService           // the merkle dag service, get/add objects.
+	OfflineDAG                  ipld.DAGService           `name:"offlineDagService"`    // merkle dag service which doesn't try to fetch from the network
 	IPLDFetcherFactory          fetcher.Factory           `name:"ipldFetcher"`          // fetcher that paths over the IPLD data model
 	UnixFSFetcherFactory        fetcher.Factory           `name:"unixfsFetcher"`        // fetcher that interprets UnixFS data
 	OfflineIPLDFetcherFactory   fetcher.Factory           `name:"offlineIpldFetcher"`   // fetcher that paths over the IPLD data model without fetching new blocks

core/coreapi/coreapi.go

@@ -244,7 +244,7 @@ func (api *CoreAPI) WithOptions(opts ...options.ApiOption) (coreiface.CoreAPI, e
 
 	if settings.Offline || !settings.FetchBlocks {
 		subAPI.exchange = offlinexch.Exchange(subAPI.blockstore)
-		subAPI.blocks = bserv.New(subAPI.blockstore, subAPI.exchange)
+		subAPI.blocks = bserv.New(subAPI.blockstore, nil, bserv.WithProvider(subAPI.provider))
 		subAPI.dag = dag.NewDAGService(subAPI.blocks)
 	}
 

core/coreapi/dht.go

@@ -6,9 +6,9 @@ import (
 
 	blockservice "github.com/ipfs/boxo/blockservice"
 	blockstore "github.com/ipfs/boxo/blockstore"
-	offline "github.com/ipfs/boxo/exchange/offline"
 	dag "github.com/ipfs/boxo/ipld/merkledag"
 	"github.com/ipfs/boxo/path"
+	"github.com/ipfs/boxo/provider"
 	cid "github.com/ipfs/go-cid"
 	cidutil "github.com/ipfs/go-cidutil"
 	coreiface "github.com/ipfs/kubo/core/coreiface"
@@ -99,7 +99,7 @@ func (api *DhtAPI) Provide(ctx context.Context, path path.Path, opts ...caopts.D
 	}
 
 	if settings.Recursive {
-		err = provideKeysRec(ctx, api.routing, api.blockstore, []cid.Cid{c})
+		err = provideKeysRec(ctx, api.routing, api.blockstore, api.provider, []cid.Cid{c})
 	} else {
 		err = provideKeys(ctx, api.routing, []cid.Cid{c})
 	}
@@ -120,12 +120,13 @@ func provideKeys(ctx context.Context, r routing.Routing, cids []cid.Cid) error {
 	return nil
 }
 
-func provideKeysRec(ctx context.Context, r routing.Routing, bs blockstore.Blockstore, cids []cid.Cid) error {
+func provideKeysRec(ctx context.Context, r routing.Routing, bs blockstore.Blockstore, prov provider.Provider, cids []cid.Cid) error {
 	provided := cidutil.NewStreamingSet()
 
 	errCh := make(chan error)
 	go func() {
-		dserv := dag.NewDAGService(blockservice.New(bs, offline.Exchange(bs)))
+		// FIXME: we are recreating a dag and blockservice, maybe offline varients should be shared ?
+		dserv := dag.NewDAGService(blockservice.New(bs, nil, blockservice.WithProvider(prov)))
 		for _, c := range cids {
 			err := dag.Walk(ctx, dag.GetLinksDirect(dserv), c, provided.Visitor(ctx))
 			if err != nil {

core/coreapi/pin.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 
 	bserv "github.com/ipfs/boxo/blockservice"
-	offline "github.com/ipfs/boxo/exchange/offline"
 	"github.com/ipfs/boxo/ipld/merkledag"
 	"github.com/ipfs/boxo/path"
 	pin "github.com/ipfs/boxo/pinning/pinner"
@@ -195,7 +194,8 @@ func (api *PinAPI) Verify(ctx context.Context) (<-chan coreiface.PinStatus, erro
 
 	visited := make(map[cid.Cid]*pinStatus)
 	bs := api.blockstore
-	DAG := merkledag.NewDAGService(bserv.New(bs, offline.Exchange(bs)))
+	// FIXME: we are recreating a dag and blockservice, maybe offline varients should be shared ?
+	DAG := merkledag.NewDAGService(bserv.New(bs, nil, bserv.WithProvider(api.provider)))
 	getLinks := merkledag.GetLinksWithDAG(DAG)
 
 	var checkPin func(root cid.Cid) *pinStatus

core/coreapi/unixfs.go

@@ -106,6 +106,7 @@ func (api *UnixfsAPI) Add(ctx context.Context, files files.Node, opts ...options
 	}
 	exch := api.exchange
 	pinning := api.pinning
+	prov := api.provider
 
 	if settings.OnlyHash {
 		node, err := getOrCreateNilNode()
@@ -115,9 +116,10 @@ func (api *UnixfsAPI) Add(ctx context.Context, files files.Node, opts ...options
 		addblockstore = node.Blockstore
 		exch = node.Exchange
 		pinning = node.Pinning
+		prov = nil
 	}
 
-	bserv := blockservice.New(addblockstore, exch) // hash security 001
+	bserv := blockservice.New(addblockstore, exch, blockservice.WithProvider(prov)) // hash security 001
 	dserv := merkledag.NewDAGService(bserv)
 
 	// add a sync call to the DagService

core/corehttp/gateway.go

@@ -9,8 +9,6 @@ import (
 	"net/http"
 	"time"
 
-	"github.com/ipfs/boxo/blockservice"
-	"github.com/ipfs/boxo/exchange/offline"
 	"github.com/ipfs/boxo/files"
 	"github.com/ipfs/boxo/gateway"
 	"github.com/ipfs/boxo/namesys"
@@ -79,7 +77,7 @@ func VersionOption() ServeOption {
 
 func Libp2pGatewayOption() ServeOption {
 	return func(n *core.IpfsNode, _ net.Listener, mux *http.ServeMux) (*http.ServeMux, error) {
-		bserv := blockservice.New(n.Blocks.Blockstore(), offline.Exchange(n.Blocks.Blockstore()))
+		bserv := n.OfflineBlocks
 
 		backend, err := gateway.NewBlocksBackend(bserv,
 			// GatewayOverLibp2p only returns things that are in local blockstore
@@ -118,7 +116,7 @@ func newGatewayBackend(n *core.IpfsNode) (gateway.IPFSBackend, error) {
 	pathResolver := n.UnixFSPathResolver
 
 	if cfg.Gateway.NoFetch {
-		bserv = blockservice.New(bserv.Blockstore(), offline.Exchange(bserv.Blockstore()))
+		bserv = n.OfflineBlocks
 
 		cs := cfg.Ipns.ResolveCacheSize
 		if cs == 0 {

core/node/bitswap.go

@@ -31,17 +31,16 @@ type bitswapOptionsOut struct {
 	BitswapOpts []bitswap.Option `group:"bitswap-options,flatten"`
 }
 
-// BitswapOptions creates configuration options for Bitswap from the config file
-// and whether to provide data.
-func BitswapOptions(cfg *config.Config, provide bool) interface{} {
-	return func() bitswapOptionsOut {
+// BitswapOptions creates configuration options for Bitswap from the config file.
+func BitswapOptions(cfg *config.Config) fx.Option {
+	return fx.Provide(func(routing irouting.ProvideManyRouter) bitswapOptionsOut {
 		var internalBsCfg config.InternalBitswap
 		if cfg.Internal.Bitswap != nil {
 			internalBsCfg = *cfg.Internal.Bitswap
 		}
 
 		opts := []bitswap.Option{
-			bitswap.ProvideEnabled(provide),
+			bitswap.WithContentSearch(routing),
 			bitswap.ProviderSearchDelay(internalBsCfg.ProviderSearchDelay.WithDefault(DefaultProviderSearchDelay)), // See https://github.com/ipfs/go-ipfs/issues/8807 for rationale
 			bitswap.EngineBlockstoreWorkerCount(int(internalBsCfg.EngineBlockstoreWorkerCount.WithDefault(DefaultEngineBlockstoreWorkerCount))),
 			bitswap.TaskWorkerCount(int(internalBsCfg.TaskWorkerCount.WithDefault(DefaultTaskWorkerCount))),
@@ -50,25 +49,24 @@ func BitswapOptions(cfg *config.Config, provide bool) interface{} {
 		}
 
 		return bitswapOptionsOut{BitswapOpts: opts}
-	}
+	})
 }
 
 type onlineExchangeIn struct {
 	fx.In
 
 	Mctx        helpers.MetricsCtx
 	Host        host.Host
-	Rt          irouting.ProvideManyRouter
 	Bs          blockstore.GCBlockstore
 	BitswapOpts []bitswap.Option `group:"bitswap-options"`
 }
 
 // OnlineExchange creates new LibP2P backed block exchange (BitSwap).
 // Additional options to bitswap.New can be provided via the "bitswap-options"
 // group.
-func OnlineExchange() interface{} {
-	return func(in onlineExchangeIn, lc fx.Lifecycle) exchange.Interface {
-		bitswapNetwork := network.NewFromIpfsHost(in.Host, in.Rt)
+func OnlineExchange() fx.Option {
+	return fx.Provide(func(in onlineExchangeIn, lc fx.Lifecycle) exchange.Interface {
+		bitswapNetwork := network.NewFromIpfsHost(in.Host)
 
 		exch := bitswap.New(helpers.LifecycleCtx(in.Mctx, lc), bitswapNetwork, in.Bs, in.BitswapOpts...)
 		lc.Append(fx.Hook{
@@ -77,5 +75,5 @@ func OnlineExchange() interface{} {
 			},
 		})
 		return exch
-	}
+	})
 }