Merge pull request #5797 from onflow/petera/access-remove-dht-v0.33

[Access] Update ipfs libraries and disable DHT

cmd/access/node_builder/access_node_builder.go

@@ -593,6 +593,10 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 				),
 			}
 
+			if !builder.BitswapReprovideEnabled {
+				opts = append(opts, blob.WithReprovideInterval(-1))
+			}
+
 			var err error
 			bs, err = node.EngineRegistry.RegisterBlobService(channels.ExecutionDataService, ds, opts...)
 			if err != nil {
@@ -693,6 +697,7 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 						blob.NewTracer(node.Logger.With().Str("public_blob_service", channels.PublicExecutionDataService.String()).Logger()),
 					),
 				),
+				blob.WithParentBlobService(bs),
 			}
 
 			net := builder.AccessNodeConfig.PublicNetworkConfig.Network

cmd/execution_builder.go

@@ -362,6 +362,10 @@ func (exeNode *ExecutionNode) LoadBlobService(
 		),
 	}
 
+	if !node.BitswapReprovideEnabled {
+		opts = append(opts, blob.WithReprovideInterval(-1))
+	}
+
 	if exeNode.exeConf.blobstoreRateLimit > 0 && exeNode.exeConf.blobstoreBurstLimit > 0 {
 		opts = append(opts, blob.WithRateLimit(float64(exeNode.exeConf.blobstoreRateLimit), exeNode.exeConf.blobstoreBurstLimit))
 	}
@@ -1325,7 +1329,6 @@ func (exeNode *ExecutionNode) LoadSynchronizationEngine(
 	error,
 ) {
 	// initialize the synchronization engine
-	//var err error
 	spamConfig, err := synchronization.NewSpamDetectionConfig()
 	if err != nil {
 		return nil, fmt.Errorf("could not initialize spam detection config: %w", err)

cmd/node_builder.go

@@ -10,9 +10,10 @@ import (
 	"github.com/rs/zerolog"
 	"github.com/spf13/pflag"
 
+	"github.com/onflow/flow-go/crypto"
+
 	"github.com/onflow/flow-go/admin/commands"
 	"github.com/onflow/flow-go/config"
-	"github.com/onflow/flow-go/crypto"
 	"github.com/onflow/flow-go/fvm"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module"
@@ -172,6 +173,13 @@ type BaseConfig struct {
 
 	// FlowConfig Flow configuration.
 	FlowConfig config.FlowConfig
+
+	// DhtSystemEnabled configures whether the DHT system is enabled on Access and Execution nodes.
+	DhtSystemEnabled bool
+
+	// BitswapReprovideEnabled configures whether the Bitswap reprovide mechanism is enabled.
+	// This is only meaningful to Access and Execution nodes.
+	BitswapReprovideEnabled bool
 }
 
 // NodeConfig contains all the derived parameters such the NodeID, private keys etc. and initialized instances of
@@ -235,7 +243,7 @@ type StateExcerptAtBoot struct {
 	SealedRootBlock     *flow.Block             // The last sealed block when bootstrapped.
 	RootQC              *flow.QuorumCertificate // QC for Finalized Root Block
 	RootResult          *flow.ExecutionResult   // Result for SealedRootBlock
-	RootSeal            *flow.Seal              //Seal for RootResult
+	RootSeal            *flow.Seal              // Seal for RootResult
 	RootChainID         flow.ChainID
 	SporkID             flow.Identifier
 	LastFinalizedHeader *flow.Header // last finalized header when the node boots up
@@ -280,10 +288,12 @@ func DefaultBaseConfig() *BaseConfig {
 			Duration: 10 * time.Second,
 		},
 
-		HeroCacheMetricsEnable: false,
-		SyncCoreConfig:         chainsync.DefaultConfig(),
-		CodecFactory:           codecFactory,
-		ComplianceConfig:       compliance.DefaultConfig(),
+		HeroCacheMetricsEnable:  false,
+		SyncCoreConfig:          chainsync.DefaultConfig(),
+		CodecFactory:            codecFactory,
+		ComplianceConfig:        compliance.DefaultConfig(),
+		DhtSystemEnabled:        true,
+		BitswapReprovideEnabled: true,
 	}
 }
 

cmd/scaffold.go

@@ -24,14 +24,15 @@ import (
 	"golang.org/x/time/rate"
 	"google.golang.org/api/option"
 
+	"github.com/onflow/flow-go/crypto"
+
 	"github.com/onflow/flow-go/admin"
 	"github.com/onflow/flow-go/admin/commands"
 	"github.com/onflow/flow-go/admin/commands/common"
 	storageCommands "github.com/onflow/flow-go/admin/commands/storage"
 	"github.com/onflow/flow-go/cmd/build"
 	"github.com/onflow/flow-go/config"
 	"github.com/onflow/flow-go/consensus/hotstuff/persister"
-	"github.com/onflow/flow-go/crypto"
 	"github.com/onflow/flow-go/fvm"
 	"github.com/onflow/flow-go/fvm/environment"
 	"github.com/onflow/flow-go/model/flow"
@@ -190,6 +191,15 @@ func (fnb *FlowNodeBuilder) BaseFlags() {
 	fnb.flags.UintVar(&fnb.BaseConfig.guaranteesCacheSize, "guarantees-cache-size", bstorage.DefaultCacheSize, "collection guarantees cache size")
 	fnb.flags.UintVar(&fnb.BaseConfig.receiptsCacheSize, "receipts-cache-size", bstorage.DefaultCacheSize, "receipts cache size")
 
+	fnb.flags.BoolVar(&fnb.BaseConfig.DhtSystemEnabled,
+		"dht-enabled",
+		defaultConfig.DhtSystemEnabled,
+		"[experimental] whether to enable dht system. This is an experimental feature. Use with caution.")
+	fnb.flags.BoolVar(&fnb.BaseConfig.BitswapReprovideEnabled,
+		"bitswap-reprovide-enabled",
+		defaultConfig.BitswapReprovideEnabled,
+		"[experimental] whether to enable bitswap reproviding. This is an experimental feature. Use with caution.")
+
 	// dynamic node startup flags
 	fnb.flags.StringVar(&fnb.BaseConfig.DynamicStartupANPubkey,
 		"dynamic-startup-access-publickey",
@@ -413,10 +423,11 @@ func (fnb *FlowNodeBuilder) EnqueueNetworkInit() {
 			return publicLibp2pNode, nil
 		}
 
-		dhtActivationStatus, err := DhtSystemActivationStatus(fnb.NodeRole)
+		dhtActivationStatus, err := DhtSystemActivationStatus(fnb.NodeRole, fnb.DhtSystemEnabled)
 		if err != nil {
 			return nil, fmt.Errorf("could not determine dht activation status: %w", err)
 		}
+
 		builder, err := p2pbuilder.DefaultNodeBuilder(fnb.Logger,
 			myAddr,
 			network.PrivateNetwork,
@@ -2136,10 +2147,11 @@ func (fnb *FlowNodeBuilder) extraFlagsValidation() error {
 // DhtSystemActivationStatus parses the given role string and returns the corresponding DHT system activation status.
 // Args:
 // - roleStr: the role string to parse.
+// - enabled: whether the DHT system is configured to be enabled. Only meaningful for access and execution nodes.
 // Returns:
 // - DhtSystemActivation: the corresponding DHT system activation status.
 // - error: if the role string is invalid, returns an error.
-func DhtSystemActivationStatus(roleStr string) (p2pbuilder.DhtSystemActivation, error) {
+func DhtSystemActivationStatus(roleStr string, enabled bool) (p2pbuilder.DhtSystemActivation, error) {
 	if roleStr == "ghost" {
 		// ghost node is not a valid role, so we don't need to parse it
 		return p2pbuilder.DhtSystemDisabled, nil
@@ -2150,12 +2162,12 @@ func DhtSystemActivationStatus(roleStr string) (p2pbuilder.DhtSystemActivation,
 		// ghost role is not a valid role, so we don't need to parse it
 		return p2pbuilder.DhtSystemDisabled, fmt.Errorf("could not parse node role: %w", err)
 	}
-	if role == flow.RoleAccess || role == flow.RoleExecution {
-		// Only access and execution nodes need to run DHT;
-		// Access nodes and execution nodes need DHT to run a blob service.
-		// Moreover, access nodes run a DHT to let un-staked (public) access nodes find each other on the public network.
-		return p2pbuilder.DhtSystemEnabled, nil
+
+	// Only access and execution nodes need to run DHT; which is used by bitswap.
+	// Access nodes also run a DHT on the public network for peer discovery of un-staked nodes.
+	if role != flow.RoleAccess && role != flow.RoleExecution {
+		return p2pbuilder.DhtSystemDisabled, nil
 	}
 
-	return p2pbuilder.DhtSystemDisabled, nil
+	return p2pbuilder.DhtSystemActivation(enabled), nil
 }

cmd/scaffold_test.go

@@ -738,56 +738,78 @@ func TestDhtSystemActivationStatus(t *testing.T) {
 	tests := []struct {
 		name      string
 		roleStr   string
+		enabled   bool
 		expected  p2pbuilder.DhtSystemActivation
 		expectErr bool
 	}{
 		{
 			name:      "ghost role returns disabled",
 			roleStr:   "ghost",
+			enabled:   true,
 			expected:  p2pbuilder.DhtSystemDisabled,
 			expectErr: false,
 		},
 		{
 			name:      "access role returns enabled",
 			roleStr:   "access",
+			enabled:   true,
 			expected:  p2pbuilder.DhtSystemEnabled,
 			expectErr: false,
 		},
 		{
 			name:      "execution role returns enabled",
 			roleStr:   "execution",
+			enabled:   true,
 			expected:  p2pbuilder.DhtSystemEnabled,
 			expectErr: false,
 		},
+		{
+			name:      "access role with disabled returns disabled",
+			roleStr:   "access",
+			enabled:   false,
+			expected:  p2pbuilder.DhtSystemDisabled,
+			expectErr: false,
+		},
+		{
+			name:      "execution role with disabled returns disabled",
+			roleStr:   "execution",
+			enabled:   false,
+			expected:  p2pbuilder.DhtSystemDisabled,
+			expectErr: false,
+		},
 		{
 			name:      "collection role returns disabled",
 			roleStr:   "collection",
+			enabled:   true,
 			expected:  p2pbuilder.DhtSystemDisabled,
 			expectErr: false,
 		},
 		{
 			name:      "consensus role returns disabled",
 			roleStr:   "consensus",
+			enabled:   true,
 			expected:  p2pbuilder.DhtSystemDisabled,
 			expectErr: false,
 		},
 		{
 			name:      "verification nodes return disabled",
 			roleStr:   "verification",
+			enabled:   true,
 			expected:  p2pbuilder.DhtSystemDisabled,
 			expectErr: false,
 		},
 		{
 			name:      "invalid role returns error",
 			roleStr:   "invalidRole",
+			enabled:   true,
 			expected:  p2pbuilder.DhtSystemDisabled,
 			expectErr: true,
 		}, // Add more test cases for other roles, if needed.
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			result, err := DhtSystemActivationStatus(tt.roleStr)
+			result, err := DhtSystemActivationStatus(tt.roleStr, tt.enabled)
 			require.Equal(t, tt.expectErr, err != nil, "unexpected error status")
 			require.Equal(t, tt.expected, result, "unexpected activation status")
 		})

engine/execution/computation/computer/computer_test.go

@@ -15,9 +15,9 @@ import (
 	"github.com/onflow/cadence/runtime/sema"
 	"github.com/onflow/cadence/runtime/stdlib"
 
+	"github.com/ipfs/boxo/blockstore"
 	"github.com/ipfs/go-datastore"
 	dssync "github.com/ipfs/go-datastore/sync"
-	blockstore "github.com/ipfs/go-ipfs-blockstore"
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"

engine/execution/computation/execution_verification_test.go

@@ -6,9 +6,9 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/ipfs/boxo/blockstore"
 	"github.com/ipfs/go-datastore"
 	dssync "github.com/ipfs/go-datastore/sync"
-	blockstore "github.com/ipfs/go-ipfs-blockstore"
 	"github.com/onflow/cadence"
 	jsoncdc "github.com/onflow/cadence/encoding/json"
 	"github.com/rs/zerolog"

engine/execution/computation/manager_benchmark_test.go

@@ -7,9 +7,9 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ipfs/boxo/blockstore"
 	"github.com/ipfs/go-datastore"
 	dssync "github.com/ipfs/go-datastore/sync"
-	blockstore "github.com/ipfs/go-ipfs-blockstore"
 	"github.com/onflow/cadence/runtime"
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/mock"

engine/execution/computation/manager_test.go

@@ -9,9 +9,9 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ipfs/boxo/blockstore"
 	"github.com/ipfs/go-datastore"
 	dssync "github.com/ipfs/go-datastore/sync"
-	blockstore "github.com/ipfs/go-ipfs-blockstore"
 	"github.com/onflow/cadence"
 	jsoncdc "github.com/onflow/cadence/encoding/json"
 	"github.com/onflow/cadence/runtime/common"

engine/execution/computation/programs_test.go

@@ -6,9 +6,9 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ipfs/boxo/blockstore"
 	"github.com/ipfs/go-datastore"
 	dssync "github.com/ipfs/go-datastore/sync"
-	blockstore "github.com/ipfs/go-ipfs-blockstore"
 	"github.com/onflow/cadence"
 	"github.com/onflow/cadence/encoding/ccf"
 	"github.com/rs/zerolog"

engine/testutil/nodes.go

@@ -10,14 +10,16 @@ import (
 	"time"
 
 	"github.com/coreos/go-semver/semver"
+	"github.com/ipfs/boxo/blockstore"
 	"github.com/ipfs/go-datastore"
 	dssync "github.com/ipfs/go-datastore/sync"
-	blockstore "github.com/ipfs/go-ipfs-blockstore"
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/atomic"
 
+	"github.com/onflow/flow-go/crypto"
+
 	"github.com/onflow/flow-go/cmd/build"
 	"github.com/onflow/flow-go/consensus"
 	"github.com/onflow/flow-go/consensus/hotstuff"
@@ -26,7 +28,6 @@ import (
 	"github.com/onflow/flow-go/consensus/hotstuff/model"
 	"github.com/onflow/flow-go/consensus/hotstuff/notifications"
 	"github.com/onflow/flow-go/consensus/hotstuff/notifications/pubsub"
-	"github.com/onflow/flow-go/crypto"
 	"github.com/onflow/flow-go/engine"
 	"github.com/onflow/flow-go/engine/collection/epochmgr"
 	"github.com/onflow/flow-go/engine/collection/epochmgr/factories"

engine/verification/utils/unittest/fixture.go

@@ -5,9 +5,9 @@ import (
 	"math/rand"
 	"testing"
 
+	"github.com/ipfs/boxo/blockstore"
 	"github.com/ipfs/go-datastore"
 	dssync "github.com/ipfs/go-datastore/sync"
-	blockstore "github.com/ipfs/go-ipfs-blockstore"
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"

fvm/fvm_bench_test.go

@@ -10,9 +10,9 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/ipfs/boxo/blockstore"
 	"github.com/ipfs/go-datastore"
 	dssync "github.com/ipfs/go-datastore/sync"
-	blockstore "github.com/ipfs/go-ipfs-blockstore"
 	"github.com/onflow/atree"
 	"github.com/onflow/cadence"
 	"github.com/onflow/cadence/encoding/ccf"
@@ -24,6 +24,7 @@ import (
 
 	flow2 "github.com/onflow/flow-go-sdk"
 	"github.com/onflow/flow-go-sdk/templates"
+
 	"github.com/onflow/flow-go/engine/execution"
 	"github.com/onflow/flow-go/engine/execution/computation"
 	"github.com/onflow/flow-go/engine/execution/computation/committer"