Review of POS mempool (remove _connectFailingTxn)

bls/signature.go

@@ -227,6 +227,14 @@ func (publicKey *PublicKey) FromString(publicKeyString string) (*PublicKey, erro
 	return publicKey, err
 }
 
+func (publicKey *PublicKey) ToAbbreviatedString() string {
+	str := publicKey.ToString()
+	if len(str) <= 8 {
+		return str
+	}
+	return str[:8] + "..." + str[len(str)-8:]
+}
+
 func (publicKey *PublicKey) MarshalJSON() ([]byte, error) {
 	// This is called automatically by the JSON library when converting a
 	// bls.PublicKey to JSON. This is useful when passing a bls.PublicKey
@@ -324,6 +332,14 @@ func (signature *Signature) FromString(signatureString string) (*Signature, erro
 	return signature, nil
 }
 
+func (signature *Signature) ToAbbreviatedString() string {
+	str := signature.ToString()
+	if len(str) <= 8 {
+		return str
+	}
+	return str[:8] + "..." + str[len(str)-8:]
+}
+
 func (signature *Signature) MarshalJSON() ([]byte, error) {
 	// This is called automatically by the JSON library when converting a
 	// bls.Signature to JSON. This is useful when passing a bls.Signature

bls/signature_no_relic.go

@@ -84,6 +84,10 @@ func (publicKey *PublicKey) FromString(publicKeyString string) (*PublicKey, erro
 	panic(BLSNoRelicError)
 }
 
+func (publicKey *PublicKey) ToAbbreviatedString() string {
+	panic(BLSNoRelicError)
+}
+
 func (publicKey *PublicKey) MarshalJSON() ([]byte, error) {
 	panic(BLSNoRelicError)
 }
@@ -136,6 +140,10 @@ func (signature *Signature) FromString(signatureString string) (*Signature, erro
 	panic(BLSNoRelicError)
 }
 
+func (signature *Signature) ToAbbreviatedString() string {
+	panic(BLSNoRelicError)
+}
+
 func (signature *Signature) MarshalJSON() ([]byte, error) {
 	panic(BLSNoRelicError)
 }

cmd/config.go

@@ -49,11 +49,13 @@ type Config struct {
 	PosTimeoutBaseDurationMilliseconds     uint64
 
 	// Mempool
-	MempoolBackupIntervalMillis             uint64
-	MaxMempoolPosSizeBytes                  uint64
-	MempoolFeeEstimatorNumMempoolBlocks     uint64
-	MempoolFeeEstimatorNumPastBlocks        uint64
-	AugmentedBlockViewRefreshIntervalMillis uint64
+	MempoolBackupIntervalMillis                uint64
+	MaxMempoolPosSizeBytes                     uint64
+	MempoolFeeEstimatorNumMempoolBlocks        uint64
+	MempoolFeeEstimatorNumPastBlocks           uint64
+	MempoolMaxValidationViewConnects           uint64
+	TransactionValidationRefreshIntervalMillis uint64
+	AugmentedBlockViewRefreshIntervalMillis    uint64
 
 	// Mining
 	MinerPublicKeys  []string
@@ -80,7 +82,8 @@ type Config struct {
 	TimeEvents            bool
 
 	// State Syncer
-	StateChangeDir string
+	StateChangeDir                 string
+	StateSyncerMempoolTxnSyncLimit uint64
 }
 
 func LoadConfig() *Config {
@@ -130,6 +133,8 @@ func LoadConfig() *Config {
 	config.MaxMempoolPosSizeBytes = viper.GetUint64("max-mempool-pos-size-bytes")
 	config.MempoolFeeEstimatorNumMempoolBlocks = viper.GetUint64("mempool-fee-estimator-num-mempool-blocks")
 	config.MempoolFeeEstimatorNumPastBlocks = viper.GetUint64("mempool-fee-estimator-num-past-blocks")
+	config.MempoolMaxValidationViewConnects = viper.GetUint64("mempool-max-validation-view-connects")
+	config.TransactionValidationRefreshIntervalMillis = viper.GetUint64("transaction-validation-refresh-interval-millis")
 	config.AugmentedBlockViewRefreshIntervalMillis = viper.GetUint64("augmented-block-view-refresh-interval-millis")
 
 	// Peers
@@ -176,6 +181,7 @@ func LoadConfig() *Config {
 
 	// State Syncer
 	config.StateChangeDir = viper.GetString("state-change-dir")
+	config.StateSyncerMempoolTxnSyncLimit = viper.GetUint64("state-syncer-mempool-txn-sync-limit")
 
 	return &config
 }

cmd/node.go

@@ -27,12 +27,13 @@ import (
 )
 
 type Node struct {
-	Server   *lib.Server
-	ChainDB  *badger.DB
-	TXIndex  *lib.TXIndex
-	Params   *lib.DeSoParams
-	Config   *Config
-	Postgres *lib.Postgres
+	Server    *lib.Server
+	ChainDB   *badger.DB
+	TXIndex   *lib.TXIndex
+	Params    *lib.DeSoParams
+	Config    *Config
+	Postgres  *lib.Postgres
+	Listeners []net.Listener
 
 	// IsRunning is false when a NewNode is created, set to true on Start(), set to false
 	// after Stop() is called. Mainly used in testing.
@@ -117,8 +118,7 @@ func (node *Node) Start(exitChannels ...*chan struct{}) {
 
 	// This just gets localhost listening addresses on the protocol port.
 	// Such as [{127.0.0.1 18000 } {::1 18000 }], and associated listener structs.
-	listeningAddrs, listeners := GetAddrsToListenOn(node.Config.ProtocolPort)
-	_ = listeningAddrs
+	_, node.Listeners = GetAddrsToListenOn(node.Config.ProtocolPort)
 
 	// If --connect-ips is not passed, we will connect the addresses from
 	// --add-ips, DNSSeeds, and DNSSeedGenerators.
@@ -238,7 +238,7 @@ func (node *Node) Start(exitChannels ...*chan struct{}) {
 	shouldRestart := false
 	node.Server, err, shouldRestart = lib.NewServer(
 		node.Params,
-		listeners,
+		node.Listeners,
 		desoAddrMgr,
 		node.Config.ConnectIPs,
 		node.ChainDB,
@@ -279,9 +279,12 @@ func (node *Node) Start(exitChannels ...*chan struct{}) {
 		node.Config.MempoolBackupIntervalMillis,
 		node.Config.MempoolFeeEstimatorNumMempoolBlocks,
 		node.Config.MempoolFeeEstimatorNumPastBlocks,
+		node.Config.MempoolMaxValidationViewConnects,
+		node.Config.TransactionValidationRefreshIntervalMillis,
 		node.Config.AugmentedBlockViewRefreshIntervalMillis,
 		node.Config.PosBlockProductionIntervalMilliseconds,
 		node.Config.PosTimeoutBaseDurationMilliseconds,
+		node.Config.StateSyncerMempoolTxnSyncLimit,
 	)
 	if err != nil {
 		// shouldRestart can be true if, on the previous run, we did not finish flushing all ancestral

cmd/run.go

@@ -102,9 +102,14 @@ func SetupRunFlags(cmd *cobra.Command) {
 		"The number of future blocks to break the PoS mempool into when estimating txn fee for the next block.")
 	cmd.PersistentFlags().Uint64("mempool-fee-estimator-num-past-blocks", 50,
 		"The number of past blocks to use when estimating txn fee for the next block from the PoS mempool.")
+	cmd.PersistentFlags().Uint64("mempool-max-validation-view-connects", 10000,
+		"The maximum number of connects that the mempool transaction validation routine will perform.")
+	cmd.PersistentFlags().Uint64("transaction-validation-refresh-interval-millis", 10,
+		"The frequency in milliseconds with which the transaction validation routine is run in mempool. "+
+			"The default value is 10 milliseconds.")
 	cmd.PersistentFlags().Uint64("augmented-block-view-refresh-interval-millis", 10,
 		"The frequency in milliseconds with which the augmented block view will be refreshed. "+
-			"The default value is 100 milliseconds.")
+			"The default value is 10 milliseconds.")
 
 	// Peers
 	cmd.PersistentFlags().StringSlice("connect-ips", []string{},
@@ -222,6 +227,8 @@ func SetupRunFlags(cmd *cobra.Command) {
 	cmd.PersistentFlags().Bool("time-events", false, "Enable simple event timer, helpful in hands-on performance testing")
 	cmd.PersistentFlags().String("state-change-dir", "", "The directory for state change logs. WARNING: Changing this "+
 		"from an empty string to a non-empty string (or from a non-empty string to the empty string) requires a resync.")
+	cmd.PersistentFlags().Uint("state-syncer-mempool-txn-sync-limit", 10000, "The maximum number of transactions to "+
+		"process in the mempool tx state syncer at a time.")
 	cmd.PersistentFlags().VisitAll(func(flag *pflag.Flag) {
 		viper.BindPFlag(flag.Name, flag)
 	})

collections/concurrent_map.go

@@ -0,0 +1,80 @@
+package collections
+
+import "sync"
+
+type ConcurrentMap[Key comparable, Value any] struct {
+	mtx sync.RWMutex
+	m   map[Key]Value
+}
+
+func NewConcurrentMap[Key comparable, Value any]() *ConcurrentMap[Key, Value] {
+	return &ConcurrentMap[Key, Value]{
+		m: make(map[Key]Value),
+	}
+}
+
+func (cm *ConcurrentMap[Key, Value]) Set(key Key, val Value) {
+	cm.mtx.Lock()
+	defer cm.mtx.Unlock()
+
+	cm.m[key] = val
+}
+
+func (cm *ConcurrentMap[Key, Value]) Remove(key Key) {
+	cm.mtx.Lock()
+	defer cm.mtx.Unlock()
+
+	_, ok := cm.m[key]
+	if !ok {
+		return
+	}
+	delete(cm.m, key)
+}
+
+func (cm *ConcurrentMap[Key, Value]) Get(key Key) (Value, bool) {
+	cm.mtx.RLock()
+	defer cm.mtx.RUnlock()
+
+	val, ok := cm.m[key]
+	return val, ok
+}
+
+func (cm *ConcurrentMap[Key, Value]) Clone() *ConcurrentMap[Key, Value] {
+	cm.mtx.RLock()
+	defer cm.mtx.RUnlock()
+
+	clone := NewConcurrentMap[Key, Value]()
+	for key, val := range cm.m {
+		clone.Set(key, val)
+	}
+	return clone
+}
+
+func (cm *ConcurrentMap[Key, Value]) ToMap() map[Key]Value {
+	cm.mtx.RLock()
+	defer cm.mtx.RUnlock()
+
+	index := make(map[Key]Value)
+	for key, node := range cm.m {
+		index[key] = node
+	}
+	return index
+}
+
+func (cm *ConcurrentMap[Key, Value]) GetAll() []Value {
+	cm.mtx.RLock()
+	defer cm.mtx.RUnlock()
+
+	var vals []Value
+	for _, val := range cm.m {
+		vals = append(vals, val)
+	}
+	return vals
+}
+
+func (cm *ConcurrentMap[Key, Value]) Count() int {
+	cm.mtx.RLock()
+	defer cm.mtx.RUnlock()
+
+	return len(cm.m)
+}

collections/concurrent_map_test.go

@@ -0,0 +1,61 @@
+package collections
+
+import (
+	"fmt"
+	"testing"
+)
+
+func TestConcurrentMap(t *testing.T) {
+	m := NewConcurrentMap[string, int]()
+	control := make(map[string]int)
+
+	// test add
+	for ii := 0; ii < 100; ii++ {
+		key := fmt.Sprintf("%v", ii)
+		m.Set(key, ii)
+		control[key] = ii
+	}
+
+	for key, val := range control {
+		if mVal, ok := m.Get(key); !ok || mVal != val {
+			t.Errorf("Expected %d, got %d", val, m.m[key])
+		}
+	}
+
+	// test remove
+	for ii := 0; ii < 50; ii++ {
+		key := fmt.Sprintf("%v", ii)
+		m.Remove(key)
+		delete(control, key)
+	}
+
+	for key, val := range control {
+		if mVal, ok := m.Get(key); !ok || mVal != val {
+			t.Errorf("Expected %d, got %d", val, m.m[key])
+		}
+	}
+
+	// test copy
+	copy := m.ToMap()
+	for key, val := range control {
+		if mVal, ok := copy[key]; !ok || mVal != val {
+			t.Errorf("Expected %d, got %d", val, m.m[key])
+		}
+	}
+	if len(copy) != len(control) {
+		t.Errorf("Expected %d, got %d", len(control), len(copy))
+	}
+
+	// test get all
+	vals := m.GetAll()
+	for _, val := range vals {
+		if _, ok := control[fmt.Sprintf("%v", val)]; !ok {
+			t.Errorf("Expected %d, got %d", val, m.m[fmt.Sprintf("%v", val)])
+		}
+	}
+
+	// test size
+	if m.Count() != len(control) {
+		t.Errorf("Expected %d, got %d", len(control), m.Count())
+	}
+}

consensus/integration_test_types.go

@@ -104,6 +104,10 @@ func (node *validatorNode) GetStakeAmount() *uint256.Int {
 	return node.stake
 }
 
+func (node *validatorNode) GetDomains() [][]byte {
+	return [][]byte{}
+}
+
 func (node *validatorNode) ProcessBlock(incomingBlock *block) {
 	node.lock.Lock()
 	defer node.lock.Unlock()

consensus/types.go

@@ -92,6 +92,7 @@ type BlockHash interface {
 type Validator interface {
 	GetPublicKey() *bls.PublicKey
 	GetStakeAmount() *uint256.Int
+	GetDomains() [][]byte
 }
 
 type AggregateQuorumCertificate interface {

consensus/types_internal.go

@@ -35,6 +35,10 @@ func (v *validator) GetStakeAmount() *uint256.Int {
 	return v.stakeAmount
 }
 
+func (v *validator) GetDomains() [][]byte {
+	return [][]byte{}
+}
+
 ////////////////////////////////////////////////////////////////////////
 // AggregateQuorumCertificate interface implementation for internal use.
 // We use this type for unit tests, and to construct timeout QCs for