TEST: enable provide sweep by default

.github/workflows/sharness.yml

@@ -4,10 +4,10 @@ on:
   workflow_dispatch:
   pull_request:
     paths-ignore:
-      - '**/*.md'
+      - "**/*.md"
   push:
     branches:
-      - 'master'
+      - "master"
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.event_name }}-${{ github.event_name == 'push' && github.sha || github.ref }}

config/reprovider.go

@@ -8,6 +8,14 @@ import (
 const (
 	DefaultReproviderInterval = time.Hour * 22 // https://github.com/ipfs/kubo/pull/9326
 	DefaultReproviderStrategy = "all"
+
+	DefaultReproviderSweepEnabled                  = true
+	DefaultReproviderSweepMaxWorkers               = 4
+	DefaultReproviderSweepDedicatedPeriodicWorkers = 2
+	DefaultReproviderSweepDedicatedBurstWorkers    = 1
+	DefaultReproviderSweepMaxProvideConnsPerWorker = 16
+	DefaultReproviderSweepKeyStoreBatchSize        = 1 << 14 // ~544 KiB per batch (1 multihash = 34 bytes)
+	DefaultReproviderSweepOfflineDelay             = 2 * time.Hour
 )
 
 type ReproviderStrategy int
@@ -24,6 +32,23 @@ const (
 type Reprovider struct {
 	Interval *OptionalDuration `json:",omitempty"` // Time period to reprovide locally stored objects to the network
 	Strategy *OptionalString   `json:",omitempty"` // Which keys to announce
+
+	Sweep Sweep
+}
+
+// Sweep configuration describes how the Sweeping Reprovider is configured if enabled.
+type Sweep struct {
+	Enabled Flag `json:",omitempty"`
+
+	MaxWorkers               *OptionalInteger // Max number of concurrent workers performing a provide operation.
+	DedicatedPeriodicWorkers *OptionalInteger // Number of workers dedicated to periodic reprovides.
+	DedicatedBurstWorkers    *OptionalInteger // Number of workers dedicated to initial provides or burst reproviding keyspace regions after a period of inactivity.
+	MaxProvideConnsPerWorker *OptionalInteger // Number of connections that a worker is able to open to send provider records during a (re)provide operation.
+
+	KeyStoreGCInterval *OptionalDuration // Interval for garbage collection in KeyStore.
+	KeyStoreBatchSize  *OptionalInteger  // Number of multihashes to keep in memory when gc'ing the KeyStore.
+
+	OfflineDelay *OptionalDuration // Delay after which the provides changes state from Disconnected to Offline.
 }
 
 func ParseReproviderStrategy(s string) ReproviderStrategy {

core/commands/provide.go

@@ -1,6 +1,7 @@
 package commands
 
 import (
+	"errors"
 	"fmt"
 	"io"
 	"text/tabwriter"
@@ -118,7 +119,12 @@ This interface is not stable and may change from release to release.
 			return ErrNotOnline
 		}
 
-		stats, err := nd.Provider.Stat()
+		provideSys, ok := nd.Provider.(provider.System)
+		if !ok {
+			return errors.New("stats not available with experimental sweeping provider (Reprovider.Sweep.Enabled=true)")
+		}
+
+		stats, err := provideSys.Stat()
 		if err != nil {
 			return err
 		}

core/commands/routing.go

@@ -11,6 +11,8 @@ import (
 
 	"github.com/ipfs/kubo/config"
 	cmdenv "github.com/ipfs/kubo/core/commands/cmdenv"
+	"github.com/ipfs/kubo/core/node"
+	mh "github.com/multiformats/go-multihash"
 
 	dag "github.com/ipfs/boxo/ipld/merkledag"
 	"github.com/ipfs/boxo/ipns"
@@ -207,9 +209,9 @@ var provideRefRoutingCmd = &cmds.Command{
 		go func() {
 			defer cancel()
 			if rec {
-				provideErr = provideKeysRec(ctx, nd.Routing, nd.DAG, cids)
+				provideErr = provideCidsRec(ctx, nd.Provider, nd.DAG, cids)
 			} else {
-				provideErr = provideKeys(ctx, nd.Routing, cids)
+				provideErr = provideCids(nd.Provider, cids)
 			}
 			if provideErr != nil {
 				routing.PublishQueryEvent(ctx, &routing.QueryEvent{
@@ -274,8 +276,12 @@ Trigger reprovider to announce our data to network.
 		if cfg.Reprovider.Interval.WithDefault(config.DefaultReproviderInterval) == 0 {
 			return errors.New("invalid configuration: Reprovider.Interval is set to '0'")
 		}
+		provideSys, ok := nd.Provider.(*node.LegacyProvider)
+		if !ok {
+			return errors.New("manual reprovide not available with experimental sweeping provider (Reprovider.Sweep.Enabled=true)")
+		}
 
-		err = nd.Provider.Reprovide(req.Context)
+		err = provideSys.Reprovide(req.Context)
 		if err != nil {
 			return err
 		}
@@ -284,39 +290,25 @@ Trigger reprovider to announce our data to network.
 	},
 }
 
-func provideKeys(ctx context.Context, r routing.Routing, cids []cid.Cid) error {
-	for _, c := range cids {
-		err := r.Provide(ctx, c, true)
-		if err != nil {
-			return err
-		}
+func provideCids(prov node.DHTProvider, cids []cid.Cid) error {
+	mhs := make([]mh.Multihash, len(cids))
+	for i, c := range cids {
+		mhs[i] = c.Hash()
 	}
-	return nil
+	return prov.StartProviding(true, mhs...)
 }
 
-func provideKeysRec(ctx context.Context, r routing.Routing, dserv ipld.DAGService, cids []cid.Cid) error {
-	provided := cid.NewSet()
+func provideCidsRec(ctx context.Context, prov node.DHTProvider, dserv ipld.DAGService, cids []cid.Cid) error {
 	for _, c := range cids {
 		kset := cid.NewSet()
-
 		err := dag.Walk(ctx, dag.GetLinksDirect(dserv), c, kset.Visit)
 		if err != nil {
 			return err
 		}
-
-		for _, k := range kset.Keys() {
-			if provided.Has(k) {
-				continue
-			}
-
-			err = r.Provide(ctx, k, true)
-			if err != nil {
-				return err
-			}
-			provided.Add(k)
+		if err = provideCids(prov, kset.Keys()); err != nil {
+			return err
 		}
 	}
-
 	return nil
 }
 

core/core.go

@@ -92,31 +92,31 @@ type IpfsNode struct {
 	RecordValidator             record.Validator
 
 	// Online
-	PeerHost                  p2phost.Host               `optional:"true"` // the network host (server+client)
-	Peering                   *peering.PeeringService    `optional:"true"`
-	Filters                   *ma.Filters                `optional:"true"`
-	Bootstrapper              io.Closer                  `optional:"true"` // the periodic bootstrapper
-	Routing                   irouting.ProvideManyRouter `optional:"true"` // the routing system. recommend ipfs-dht
-	ContentDiscovery          routing.ContentDiscovery   `optional:"true"` // the discovery part of the routing system
-	DNSResolver               *madns.Resolver            // the DNS resolver
-	IPLDPathResolver          pathresolver.Resolver      `name:"ipldPathResolver"`          // The IPLD path resolver
-	UnixFSPathResolver        pathresolver.Resolver      `name:"unixFSPathResolver"`        // The UnixFS path resolver
-	OfflineIPLDPathResolver   pathresolver.Resolver      `name:"offlineIpldPathResolver"`   // The IPLD path resolver that uses only locally available blocks
-	OfflineUnixFSPathResolver pathresolver.Resolver      `name:"offlineUnixFSPathResolver"` // The UnixFS path resolver that uses only locally available blocks
-	Exchange                  exchange.Interface         // the block exchange + strategy
-	Bitswap                   *bitswap.Bitswap           `optional:"true"` // The Bitswap instance
-	Namesys                   namesys.NameSystem         // the name system, resolves paths to hashes
-	Provider                  provider.System            // the value provider system
-	ProvidingStrategy         config.ReproviderStrategy  `optional:"true"`
-	ProvidingKeyChanFunc      provider.KeyChanFunc       `optional:"true"`
-	IpnsRepub                 *ipnsrp.Republisher        `optional:"true"`
-	ResourceManager           network.ResourceManager    `optional:"true"`
+	PeerHost                  p2phost.Host              `optional:"true"` // the network host (server+client)
+	Peering                   *peering.PeeringService   `optional:"true"`
+	Filters                   *ma.Filters               `optional:"true"`
+	Bootstrapper              io.Closer                 `optional:"true"` // the periodic bootstrapper
+	ContentDiscovery          routing.ContentDiscovery  `optional:"true"` // the discovery part of the routing system
+	DNSResolver               *madns.Resolver           // the DNS resolver
+	IPLDPathResolver          pathresolver.Resolver     `name:"ipldPathResolver"`          // The IPLD path resolver
+	UnixFSPathResolver        pathresolver.Resolver     `name:"unixFSPathResolver"`        // The UnixFS path resolver
+	OfflineIPLDPathResolver   pathresolver.Resolver     `name:"offlineIpldPathResolver"`   // The IPLD path resolver that uses only locally available blocks
+	OfflineUnixFSPathResolver pathresolver.Resolver     `name:"offlineUnixFSPathResolver"` // The UnixFS path resolver that uses only locally available blocks
+	Exchange                  exchange.Interface        // the block exchange + strategy
+	Bitswap                   *bitswap.Bitswap          `optional:"true"` // The Bitswap instance
+	Namesys                   namesys.NameSystem        // the name system, resolves paths to hashes
+	ProvidingStrategy         config.ReproviderStrategy `optional:"true"`
+	ProvidingKeyChanFunc      provider.KeyChanFunc      `optional:"true"`
+	IpnsRepub                 *ipnsrp.Republisher       `optional:"true"`
+	ResourceManager           network.ResourceManager   `optional:"true"`
 
 	PubSub   *pubsub.PubSub             `optional:"true"`
 	PSRouter *psrouter.PubsubValueStore `optional:"true"`
 
-	DHT       *ddht.DHT       `optional:"true"`
-	DHTClient routing.Routing `name:"dhtc" optional:"true"`
+	Routing   irouting.ProvideManyRouter `optional:"true"` // the routing system. recommend ipfs-dht
+	Provider  node.DHTProvider           // the value provider system
+	DHT       *ddht.DHT                  `optional:"true"`
+	DHTClient routing.Routing            `name:"dhtc" optional:"true"`
 
 	P2P *p2p.P2P `optional:"true"`
 

core/coreapi/coreapi.go

@@ -23,10 +23,8 @@ import (
 	dag "github.com/ipfs/boxo/ipld/merkledag"
 	pathresolver "github.com/ipfs/boxo/path/resolver"
 	pin "github.com/ipfs/boxo/pinning/pinner"
-	provider "github.com/ipfs/boxo/provider"
 	offlineroute "github.com/ipfs/boxo/routing/offline"
 	ipld "github.com/ipfs/go-ipld-format"
-	logging "github.com/ipfs/go-log/v2"
 	"github.com/ipfs/kubo/config"
 	coreiface "github.com/ipfs/kubo/core/coreiface"
 	"github.com/ipfs/kubo/core/coreiface/options"
@@ -45,8 +43,6 @@ import (
 	"github.com/ipfs/kubo/repo"
 )
 
-var log = logging.Logger("coreapi")
-
 type CoreAPI struct {
 	nctx context.Context
 
@@ -73,7 +69,7 @@ type CoreAPI struct {
 	ipldPathResolver   pathresolver.Resolver
 	unixFSPathResolver pathresolver.Resolver
 
-	provider          provider.System
+	provider          node.DHTProvider
 	providingStrategy config.ReproviderStrategy
 
 	pubSub *pubsub.PubSub

core/coreapi/routing.go

@@ -15,9 +15,10 @@ import (
 	cidutil "github.com/ipfs/go-cidutil"
 	coreiface "github.com/ipfs/kubo/core/coreiface"
 	caopts "github.com/ipfs/kubo/core/coreiface/options"
+	"github.com/ipfs/kubo/core/node"
 	"github.com/ipfs/kubo/tracing"
 	peer "github.com/libp2p/go-libp2p/core/peer"
-	routing "github.com/libp2p/go-libp2p/core/routing"
+	mh "github.com/multiformats/go-multihash"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
 )
@@ -148,9 +149,9 @@ func (api *RoutingAPI) Provide(ctx context.Context, path path.Path, opts ...caop
 	}
 
 	if settings.Recursive {
-		err = provideKeysRec(ctx, api.routing, api.blockstore, []cid.Cid{c})
+		err = provideKeysRec(ctx, api.provider, api.blockstore, []cid.Cid{c})
 	} else {
-		err = provideKeys(ctx, api.routing, []cid.Cid{c})
+		err = api.provider.StartProviding(false, c.Hash())
 	}
 	if err != nil {
 		return err
@@ -159,41 +160,64 @@ func (api *RoutingAPI) Provide(ctx context.Context, path path.Path, opts ...caop
 	return nil
 }
 
-func provideKeys(ctx context.Context, r routing.Routing, cids []cid.Cid) error {
-	for _, c := range cids {
-		err := r.Provide(ctx, c, true)
-		if err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func provideKeysRec(ctx context.Context, r routing.Routing, bs blockstore.Blockstore, cids []cid.Cid) error {
+func provideKeysRec(ctx context.Context, prov node.DHTProvider, bs blockstore.Blockstore, cids []cid.Cid) error {
 	provided := cidutil.NewStreamingSet()
 
-	errCh := make(chan error)
+	// Error channel with buffer size 1 to avoid blocking the goroutine
+	errCh := make(chan error, 1)
 	go func() {
+		// Always close provided.New to signal completion
+		defer close(provided.New)
+		// Also close error channel to distinguish between "no error" and "pending error"
+		defer close(errCh)
+
 		dserv := dag.NewDAGService(blockservice.New(bs, offline.Exchange(bs)))
 		for _, c := range cids {
-			err := dag.Walk(ctx, dag.GetLinksDirect(dserv), c, provided.Visitor(ctx))
-			if err != nil {
-				errCh <- err
+			if err := dag.Walk(ctx, dag.GetLinksDirect(dserv), c, provided.Visitor(ctx)); err != nil {
+				// Send error to channel. If context is cancelled while trying to send,
+				// exit immediately as the main loop will return ctx.Err()
+				select {
+				case errCh <- err:
+					// Error sent successfully, exit goroutine
+				case <-ctx.Done():
+					// Context cancelled, exit without sending error
+					return
+				}
+				return
 			}
 		}
+		// All CIDs walked successfully, goroutine will exit and channels will close
 	}()
 
+	keys := make([]mh.Multihash, 0)
 	for {
 		select {
-		case k := <-provided.New:
-			err := r.Provide(ctx, k, true)
-			if err != nil {
-				return err
-			}
-		case err := <-errCh:
-			return err
 		case <-ctx.Done():
+			// Context cancelled, return immediately
 			return ctx.Err()
+		case err := <-errCh:
+			// Received error from DAG walk, return it
+			return err
+		case c, ok := <-provided.New:
+			if !ok {
+				// Channel closed means goroutine finished.
+				// CRITICAL: Check for any error that was sent just before channel closure.
+				// This handles the race where error is sent to errCh but main loop
+				// sees provided.New close first.
+				select {
+				case err := <-errCh:
+					if err != nil {
+						return err
+					}
+					// errCh closed with nil, meaning success
+				default:
+					// No pending error in errCh
+				}
+				// All CIDs successfully processed, start providing
+				return prov.StartProviding(true, keys...)
+			}
+			// Accumulate the CID for providing
+			keys = append(keys, c.Hash())
 		}
 	}
 }