fix: provider merge conflicts (#10989)

Co-authored-by: Marcin Rataj <lidel@lidel.org>

Author: guillaumemichel

config/provide.go

@@ -19,7 +19,7 @@ const (
 	DefaultProvideDHTDedicatedPeriodicWorkers = 2
 	DefaultProvideDHTDedicatedBurstWorkers    = 1
 	DefaultProvideDHTMaxProvideConnsPerWorker = 16
-	DefaultProvideDHTKeyStoreBatchSize        = 1 << 14 // ~544 KiB per batch (1 multihash = 34 bytes)
+	DefaultProvideDHTKeystoreBatchSize        = 1 << 14 // ~544 KiB per batch (1 multihash = 34 bytes)
 	DefaultProvideDHTOfflineDelay             = 2 * time.Hour
 )
 
@@ -79,9 +79,9 @@ type ProvideDHT struct {
 	// Default: DefaultProvideDHTMaxProvideConnsPerWorker
 	MaxProvideConnsPerWorker *OptionalInteger `json:",omitempty"`
 
-	// KeyStoreBatchSize sets the batch size for keystore operations during reprovide refresh (sweep mode only).
-	// Default: DefaultProvideDHTKeyStoreBatchSize
-	KeyStoreBatchSize *OptionalInteger `json:",omitempty"`
+	// KeystoreBatchSize sets the batch size for keystore operations during reprovide refresh (sweep mode only).
+	// Default: DefaultProvideDHTKeystoreBatchSize
+	KeystoreBatchSize *OptionalInteger `json:",omitempty"`
 
 	// OfflineDelay sets the delay after which the provider switches from Disconnected to Offline state (sweep mode only).
 	// Default: DefaultProvideDHTOfflineDelay
@@ -150,11 +150,11 @@ func ValidateProvideConfig(cfg *Provide) error {
 		}
 	}
 
-	// Validate KeyStoreBatchSize
-	if !cfg.DHT.KeyStoreBatchSize.IsDefault() {
-		batchSize := cfg.DHT.KeyStoreBatchSize.WithDefault(DefaultProvideDHTKeyStoreBatchSize)
+	// Validate KeystoreBatchSize
+	if !cfg.DHT.KeystoreBatchSize.IsDefault() {
+		batchSize := cfg.DHT.KeystoreBatchSize.WithDefault(DefaultProvideDHTKeystoreBatchSize)
 		if batchSize <= 0 {
-			return fmt.Errorf("Provide.DHT.KeyStoreBatchSize must be positive, got %d", batchSize)
+			return fmt.Errorf("Provide.DHT.KeystoreBatchSize must be positive, got %d", batchSize)
 		}
 	}
 

core/node/provider.go

@@ -24,6 +24,7 @@ import (
 	"github.com/libp2p/go-libp2p-kad-dht/fullrt"
 	dht_pb "github.com/libp2p/go-libp2p-kad-dht/pb"
 	dhtprovider "github.com/libp2p/go-libp2p-kad-dht/provider"
+	"github.com/libp2p/go-libp2p-kad-dht/provider/buffered"
 	ddhtprovider "github.com/libp2p/go-libp2p-kad-dht/provider/dual"
 	"github.com/libp2p/go-libp2p-kad-dht/provider/keystore"
 	routinghelpers "github.com/libp2p/go-libp2p-routing-helpers"
@@ -84,7 +85,7 @@ type DHTProvider interface {
 	// The keys are not deleted from the keystore, so they will continue to be
 	// reprovided as scheduled.
 	Clear() int
-	// RefreshSchedule scans the KeyStore for any keys that are not currently
+	// RefreshSchedule scans the Keystore for any keys that are not currently
 	// scheduled for reproviding. If such keys are found, it schedules their
 	// associated keyspace region to be reprovided.
 	//
@@ -314,13 +315,40 @@ func SweepingProviderOpt(cfg *config.Config) fx.Option {
 	}
 	sweepingReprovider := fx.Provide(func(in providerInput) (DHTProvider, *keystore.ResettableKeystore, error) {
 		ds := in.Repo.Datastore()
-		keyStore, err := keystore.NewResettableKeystore(ds,
+		ks, err := keystore.NewResettableKeystore(ds,
 			keystore.WithPrefixBits(16),
 			keystore.WithDatastorePath("/provider/keystore"),
-			keystore.WithBatchSize(int(cfg.Provide.DHT.KeyStoreBatchSize.WithDefault(config.DefaultProvideDHTKeyStoreBatchSize))),
+			keystore.WithBatchSize(int(cfg.Provide.DHT.KeystoreBatchSize.WithDefault(config.DefaultProvideDHTKeystoreBatchSize))),
 		)
 		if err != nil {
-			return &NoopProvider{}, nil, err
+			return nil, nil, err
+		}
+		// Constants for buffered provider configuration
+		// These values match the upstream defaults from go-libp2p-kad-dht and have been battle-tested
+		const (
+			// bufferedDsName is the datastore namespace used by the buffered provider.
+			// The dsqueue persists operations here to handle large data additions without
+			// being memory-bound, allowing operations on hardware with limited RAM and
+			// enabling core operations to return instantly while processing happens async.
+			bufferedDsName = "bprov"
+
+			// bufferedBatchSize controls how many operations are dequeued and processed
+			// together from the datastore queue. The worker processes up to this many
+			// operations at once, grouping them by type for efficiency.
+			bufferedBatchSize = 1 << 10 // 1024 items
+
+			// bufferedIdleWriteTime is an implementation detail of go-dsqueue that controls
+			// how long the datastore buffer waits for new multihashes to arrive before
+			// flushing in-memory items to the datastore. This does NOT affect providing speed -
+			// provides happen as fast as possible via a dedicated worker that continuously
+			// processes the queue regardless of this timing.
+			bufferedIdleWriteTime = time.Minute
+		)
+
+		bufferedProviderOpts := []buffered.Option{
+			buffered.WithBatchSize(bufferedBatchSize),
+			buffered.WithDsName(bufferedDsName),
+			buffered.WithIdleWriteTime(bufferedIdleWriteTime),
 		}
 		var impl dhtImpl
 		switch inDht := in.DHT.(type) {
@@ -331,7 +359,7 @@ func SweepingProviderOpt(cfg *config.Config) fx.Option {
 		case *dual.DHT:
 			if inDht != nil {
 				prov, err := ddhtprovider.New(inDht,
-					ddhtprovider.WithKeystore(keyStore),
+					ddhtprovider.WithKeystore(ks),
 
 					ddhtprovider.WithReprovideInterval(reprovideInterval),
 					ddhtprovider.WithMaxReprovideDelay(time.Hour),
@@ -346,8 +374,7 @@ func SweepingProviderOpt(cfg *config.Config) fx.Option {
 				if err != nil {
 					return nil, nil, err
 				}
-				_ = prov
-				return prov, keyStore, nil
+				return buffered.New(prov, ds, bufferedProviderOpts...), ks, nil
 			}
 		case *fullrt.FullRT:
 			if inDht != nil {
@@ -365,7 +392,7 @@ func SweepingProviderOpt(cfg *config.Config) fx.Option {
 			selfAddrsFunc = func() []ma.Multiaddr { return impl.Host().Addrs() }
 		}
 		opts := []dhtprovider.Option{
-			dhtprovider.WithKeystore(keyStore),
+			dhtprovider.WithKeystore(ks),
 			dhtprovider.WithPeerID(impl.Host().ID()),
 			dhtprovider.WithRouter(impl),
 			dhtprovider.WithMessageSender(impl.MessageSender()),
@@ -387,16 +414,19 @@ func SweepingProviderOpt(cfg *config.Config) fx.Option {
 		}
 
 		prov, err := dhtprovider.New(opts...)
-		return prov, keyStore, err
+		if err != nil {
+			return nil, nil, err
+		}
+		return buffered.New(prov, ds, bufferedProviderOpts...), ks, nil
 	})
 
 	type keystoreInput struct {
 		fx.In
 		Provider    DHTProvider
-		KeyStore    *keystore.ResettableKeystore
+		Keystore    *keystore.ResettableKeystore
 		KeyProvider provider.KeyChanFunc
 	}
-	initKeyStore := fx.Invoke(func(lc fx.Lifecycle, in keystoreInput) {
+	initKeystore := fx.Invoke(func(lc fx.Lifecycle, in keystoreInput) {
 		// Skip keystore initialization for NoopProvider
 		if _, ok := in.Provider.(*NoopProvider); ok {
 			return
@@ -407,12 +437,12 @@ func SweepingProviderOpt(cfg *config.Config) fx.Option {
 			done   = make(chan struct{})
 		)
 
-		syncKeyStore := func(ctx context.Context) error {
+		syncKeystore := func(ctx context.Context) error {
 			kcf, err := in.KeyProvider(ctx)
 			if err != nil {
 				return err
 			}
-			if err := in.KeyStore.ResetCids(ctx, kcf); err != nil {
+			if err := in.Keystore.ResetCids(ctx, kcf); err != nil {
 				return err
 			}
 			if err := in.Provider.RefreshSchedule(); err != nil {
@@ -424,7 +454,7 @@ func SweepingProviderOpt(cfg *config.Config) fx.Option {
 		lc.Append(fx.Hook{
 			OnStart: func(ctx context.Context) error {
 				// Set the KeyProvider as a garbage collection function for the
-				// keystore. Periodically purge the KeyStore from all its keys and
+				// keystore. Periodically purge the Keystore from all its keys and
 				// replace them with the keys that needs to be reprovided, coming from
 				// the KeyChanFunc. So far, this is the less worse way to remove CIDs
 				// that shouldn't be reprovided from the provider's state.
@@ -434,7 +464,7 @@ func SweepingProviderOpt(cfg *config.Config) fx.Option {
 					// which can take a while.
 					strategy := cfg.Provide.Strategy.WithDefault(config.DefaultProvideStrategy)
 					logger.Infow("provider keystore sync started", "strategy", strategy)
-					if err := syncKeyStore(ctx); err != nil {
+					if err := syncKeystore(ctx); err != nil {
 						logger.Errorw("provider keystore sync failed", "err", err, "strategy", strategy)
 					} else {
 						logger.Infow("provider keystore sync completed", "strategy", strategy)
@@ -454,7 +484,7 @@ func SweepingProviderOpt(cfg *config.Config) fx.Option {
 						case <-gcCtx.Done():
 							return
 						case <-ticker.C:
-							if err := syncKeyStore(gcCtx); err != nil {
+							if err := syncKeystore(gcCtx); err != nil {
 								logger.Errorw("provider keystore sync", "err", err)
 							}
 						}
@@ -471,18 +501,16 @@ func SweepingProviderOpt(cfg *config.Config) fx.Option {
 				case <-ctx.Done():
 					return ctx.Err()
 				}
-
 				// Keystore data isn't purged, on close, but it will be overwritten
 				// when the node starts again.
-
-				return in.KeyStore.Close()
+				return in.Keystore.Close()
 			},
 		})
 	})
 
 	return fx.Options(
 		sweepingReprovider,
-		initKeyStore,
+		initKeystore,
 	)
 }