feature/pos-networking-and-syncing

Dockerfile

@@ -24,10 +24,8 @@ COPY scripts     scripts
 COPY test_data   test_data
 COPY main.go     .
 
-RUN ./scripts/install-relic.sh
-
 # build backend
-RUN GOOS=linux go build -mod=mod -a -installsuffix cgo -o bin/core -tags=relic main.go
+RUN GOOS=linux go build -mod=mod -a -installsuffix cgo -o bin/core main.go
 
 # create tiny image
 FROM alpine:edge

bls/signature.go

@@ -1,5 +1,3 @@
-//go:build relic
-
 package bls
 
 import (
@@ -11,8 +9,8 @@ import (
 	"fmt"
 	"strings"
 
-	flowCrypto "github.com/onflow/flow-go/crypto"
-	"github.com/onflow/flow-go/crypto/hash"
+	flowCrypto "github.com/onflow/crypto"
+	"github.com/onflow/crypto/hash"
 )
 
 // The signingAlgorithm for BLS keys is BLSBLS12381 which is BLS on the BLS 12-381 curve.
@@ -227,6 +225,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 +330,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_test.go

@@ -1,5 +1,3 @@
-//go:build relic
-
 package bls
 
 import (

cmd/config.go

@@ -80,7 +80,8 @@ type Config struct {
 	TimeEvents            bool
 
 	// State Syncer
-	StateChangeDir string
+	StateChangeDir                 string
+	StateSyncerMempoolTxnSyncLimit uint64
 }
 
 func LoadConfig() *Config {
@@ -176,6 +177,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,
@@ -282,6 +282,7 @@ func (node *Node) Start(exitChannels ...*chan struct{}) {
 		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

@@ -222,6 +222,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)
+}