merge

block/manager_test.go

@@ -55,7 +55,7 @@ func TestInitialState(t *testing.T) {
 	privKey, _, _ := crypto.GenerateEd25519Key(rand.Reader)
 	p2pClient, err := p2p.NewClient(config.P2PConfig{
 		ListenAddress:                config.DefaultListenAddress,
-		GossipedBlocksCacheSize:      50,
+		GossipSubCacheSize:           50,
 		BootstrapRetryTime:           30 * time.Second,
 		BlockSyncRequestIntervalTime: 30 * time.Second,
 	}, privKey, "TestChain", emptyStore, pubsubServer, datastore.NewMapDatastore(), logger)

config/config_test.go

@@ -220,7 +220,7 @@ func fullNodeConfig() config.NodeConfig {
 			Port: 9090,
 		},
 		P2PConfig: config.P2PConfig{
-			GossipedBlocksCacheSize:      50,
+			GossipSubCacheSize:           50,
 			BootstrapRetryTime:           30 * time.Second,
 			BlockSyncRequestIntervalTime: 30 * time.Second,
 			ListenAddress:                config.DefaultListenAddress,

config/defaults.go

@@ -37,7 +37,7 @@ func DefaultConfig(home, chainId string) *NodeConfig {
 			PrometheusListenAddr: ":2112",
 		},
 		P2PConfig: P2PConfig{
-			GossipedBlocksCacheSize:      50,
+			GossipSubCacheSize:           50,
 			BootstrapRetryTime:           30 * time.Second,
 			BlockSyncRequestIntervalTime: 30 * time.Second,
 			ListenAddress:                DefaultListenAddress,

config/flags.go

@@ -62,7 +62,7 @@ func AddNodeFlags(cmd *cobra.Command) {
 	cmd.Flags().String(FlagP2PListenAddress, def.P2PConfig.ListenAddress, "P2P listen address")
 	cmd.Flags().String(FlagP2PBootstrapNodes, def.P2PConfig.BootstrapNodes, "P2P bootstrap nodes")
 	cmd.Flags().Duration(FlagP2PBootstrapRetryTime, def.P2PConfig.BootstrapRetryTime, "P2P bootstrap time")
-	cmd.Flags().Uint64(FlagP2PGossipCacheSize, uint64(def.P2PConfig.GossipedBlocksCacheSize), "P2P Gossiped blocks cache size")
+	cmd.Flags().Uint64(FlagP2PGossipCacheSize, uint64(def.P2PConfig.GossipSubCacheSize), "P2P Gossiped blocks cache size")
 }
 
 func BindDymintFlags(cmd *cobra.Command, v *viper.Viper) error {

config/p2p.go

@@ -14,7 +14,7 @@ type P2PConfig struct {
 	// List of nodes persistent P2P nodes
 	PersistentNodes string `mapstructure:"p2p_persistent_nodes"`
 	// Size of the Gossipsub router cache
-	GossipedBlocksCacheSize int `mapstructure:"p2p_gossiped_blocks_cache_size"`
+	GossipSubCacheSize int `mapstructure:"p2p_gossip_cache_size"`
 	// Time interval a node tries to bootstrap again, in case no nodes connected
 	BootstrapRetryTime time.Duration `mapstructure:"p2p_bootstrap_retry_time"`
 	// Param used to enable block sync from p2p
@@ -27,8 +27,8 @@ type P2PConfig struct {
 
 // Validate P2PConfig
 func (c P2PConfig) Validate() error {
-	if c.GossipedBlocksCacheSize <= 0 {
-		return fmt.Errorf("gossipsub cache size must be positive")
+	if c.GossipSubCacheSize < 0 {
+		return fmt.Errorf("gossipsub cache size cannot be negative")
 	}
 	if c.BootstrapRetryTime <= 0 {
 		return fmt.Errorf("bootstrap time must be positive")

config/toml.go

@@ -102,7 +102,7 @@ p2p_bootstrap_nodes = "{{ .P2PConfig.BootstrapNodes }}"
 p2p_persistent_nodes = "{{ .P2PConfig.PersistentNodes }}"
 
 # max number of cached messages by gossipsub protocol
-p2p_gossiped_blocks_cache_size = {{ .P2PConfig.GossipedBlocksCacheSize }}
+p2p_gossip_cache_size = {{ .P2PConfig.GossipSubCacheSize }}
 
 # time interval to check if no p2p nodes are connected to bootstrap again
 p2p_bootstrap_retry_time = "{{ .P2PConfig.BootstrapRetryTime }}"

node/node_test.go

@@ -66,7 +66,7 @@ func TestMempoolDirectly(t *testing.T) {
 		DBPath:  "",
 		P2PConfig: config.P2PConfig{
 			ListenAddress:                config.DefaultListenAddress,
-			GossipedBlocksCacheSize:      50,
+			GossipSubCacheSize:           50,
 			BootstrapRetryTime:           30 * time.Second,
 			BootstrapNodes:               "",
 			BlockSyncRequestIntervalTime: 30 * time.Second,

p2p/client.go

@@ -432,10 +432,9 @@ func (c *Client) tryConnect(ctx context.Context, peer peer.AddrInfo) {
 }
 
 func (c *Client) setupGossiping(ctx context.Context) error {
-	pubsub.GossipSubHistoryGossip = c.conf.GossipedBlocksCacheSize
-	pubsub.GossipSubHistoryLength = c.conf.GossipedBlocksCacheSize
+	pubsub.GossipSubHistoryGossip = c.conf.GossipSubCacheSize
+	pubsub.GossipSubHistoryLength = c.conf.GossipSubCacheSize
 
-	// We add WithSeenMessagesTTL (with 1 year time) option to avoid ever requesting already seen blocks
 	ps, err := pubsub.NewGossipSub(ctx, c.Host)
 	if err != nil {
 		return err

p2p/client_test.go

@@ -34,7 +34,7 @@ func TestClientStartup(t *testing.T) {
 	store := store.New(store.NewDefaultInMemoryKVStore())
 	client, err := p2p.NewClient(config.P2PConfig{
 		ListenAddress:                config.DefaultListenAddress,
-		GossipedBlocksCacheSize:      50,
+		GossipSubCacheSize:           50,
 		BootstrapRetryTime:           30 * time.Second,
 		BlockSyncRequestIntervalTime: 30 * time.Second,
 	}, privKey, "TestChain", store, pubsubServer, datastore.NewMapDatastore(), log.TestingLogger())
@@ -272,8 +272,8 @@ func TestSeedStringParsing(t *testing.T) {
 			logger := &testutil.MockLogger{}
 			store := store.New(store.NewDefaultInMemoryKVStore())
 			client, err := p2p.NewClient(config.P2PConfig{
-				GossipedBlocksCacheSize: 50,
-				BootstrapRetryTime:      30 * time.Second,
+				GossipSubCacheSize: 50,
+				BootstrapRetryTime: 30 * time.Second,
 			}, privKey, "TestNetwork", store, pubsubServer, datastore.NewMapDatastore(), logger)
 			require.NoError(err)
 			require.NotNil(client)

p2p/gossip.go

@@ -13,6 +13,9 @@ import (
 	"github.com/dymensionxyz/dymint/types"
 )
 
+// buffer size used by gossipSub router to consume received packets (blocks or txs). packets are dropped in case buffer overflows. in case of blocks, it can buffer up to 5 minutes (assuming 200ms block rate)
+const pubsubBufferSize = 3000
+
 // GossipMessage represents message gossiped via P2P network (e.g. transaction, Block etc).
 type GossipMessage struct {
 	Data []byte
@@ -50,8 +53,7 @@ func NewGossiper(host host.Host, ps *pubsub.PubSub, topicStr string, msgHandler
 	if err != nil {
 		return nil, err
 	}
-
-	subscription, err := topic.Subscribe()
+	subscription, err := topic.Subscribe(pubsub.WithBufferSize(max(pubsub.GossipSubHistoryGossip, pubsubBufferSize)))
 	if err != nil {
 		return nil, err
 	}

rpc/client/client.go

@@ -718,7 +718,6 @@ func (c *Client) Status(ctx context.Context) (*ctypes.ResultStatus, error) {
 	if validator == nil {
 		return nil, fmt.Errorf("find proposer %s in the valSet", string(latest.Header.ProposerAddress))
 	}
-
 	state, err := c.node.Store.LoadState()
 	if err != nil {
 		return nil, fmt.Errorf("load the last saved state: %w", err)

rpc/client/client_test.go

@@ -105,7 +105,7 @@ func TestGenesisChunked(t *testing.T) {
 		P2PConfig: config.P2PConfig{
 			ListenAddress:                config.DefaultListenAddress,
 			BootstrapNodes:               "",
-			GossipedBlocksCacheSize:      50,
+			GossipSubCacheSize:           50,
 			BootstrapRetryTime:           30 * time.Second,
 			BlockSyncRequestIntervalTime: 30 * time.Second,
 		},
@@ -707,7 +707,7 @@ func TestValidatorSetHandling(t *testing.T) {
 		P2PConfig: config.P2PConfig{
 			ListenAddress:                config.DefaultListenAddress,
 			BootstrapNodes:               "",
-			GossipedBlocksCacheSize:      50,
+			GossipSubCacheSize:           50,
 			BootstrapRetryTime:           30 * time.Second,
 			BlockSyncRequestIntervalTime: 30 * time.Second,
 		},
@@ -865,7 +865,7 @@ func getRPCInternal(t *testing.T, sequencer bool) (*tmmocks.MockApplication, *cl
 		P2PConfig: config.P2PConfig{
 			ListenAddress:                config.DefaultListenAddress,
 			BootstrapNodes:               "",
-			GossipedBlocksCacheSize:      50,
+			GossipSubCacheSize:           50,
 			BootstrapRetryTime:           30 * time.Second,
 			BlockSyncRequestIntervalTime: 30 * time.Second,
 		},
@@ -977,7 +977,7 @@ func TestMempool2Nodes(t *testing.T) {
 		P2PConfig: config.P2PConfig{
 			ListenAddress:                "/ip4/127.0.0.1/tcp/9001",
 			BootstrapNodes:               "",
-			GossipedBlocksCacheSize:      50,
+			GossipSubCacheSize:           50,
 			BootstrapRetryTime:           30 * time.Second,
 			BlockSyncRequestIntervalTime: 30 * time.Second,
 		},
@@ -1009,7 +1009,7 @@ func TestMempool2Nodes(t *testing.T) {
 			ListenAddress:                "/ip4/127.0.0.1/tcp/9002",
 			BootstrapNodes:               "/ip4/127.0.0.1/tcp/9001/p2p/" + id1.String(),
 			BootstrapRetryTime:           30 * time.Second,
-			GossipedBlocksCacheSize:      50,
+			GossipSubCacheSize:           50,
 			BlockSyncRequestIntervalTime: 30 * time.Second,
 		},
 		MempoolConfig: *tmcfg.DefaultMempoolConfig(),

rpc/json/service_test.go

@@ -314,7 +314,7 @@ func getRPC(t *testing.T) (*tmmocks.MockApplication, *client.Client) {
 		},
 		P2PConfig: config.P2PConfig{
 			ListenAddress:                config.DefaultListenAddress,
-			GossipedBlocksCacheSize:      50,
+			GossipSubCacheSize:           50,
 			BootstrapRetryTime:           30 * time.Second,
 			BlockSyncRequestIntervalTime: 30 * time.Second,
 		},

testutil/block.go

@@ -92,7 +92,7 @@ func GetManagerWithProposerKey(conf config.BlockManagerConfig, proposerKey crypt
 	// Init p2p client and validator
 	p2pKey, _, _ := crypto.GenerateEd25519Key(rand.Reader)
 	p2pClient, err := p2p.NewClient(config.P2PConfig{
-		GossipedBlocksCacheSize:      50,
+		GossipSubCacheSize:           50,
 		BootstrapRetryTime:           30 * time.Second,
 		BlockSyncRequestIntervalTime: 30 * time.Second,
 	}, p2pKey, "TestChain", managerStore, pubsubServer, datastore.NewMapDatastore(), logger)

testutil/p2p.go

@@ -112,7 +112,7 @@ func StartTestNetwork(ctx context.Context, t *testing.T, n int, conf map[int]Hos
 	for i := 0; i < n; i++ {
 		client, err := p2p.NewClient(config.P2PConfig{
 			BootstrapNodes:               seeds[i],
-			GossipedBlocksCacheSize:      50,
+			GossipSubCacheSize:           50,
 			BootstrapRetryTime:           30 * time.Second,
 			BlockSyncRequestIntervalTime: 30 * time.Second,
 			ListenAddress:                config.DefaultListenAddress,