v1.4.7 (#125)

* move to a message processor channel model for ceremony

* switch to goroutine

* readjust

* cut down volume

* keep distance data asserted

* bring it back so more bootstrap nodes can exist

* bump the version, it's go time

go-libp2p-blossomsub/pubsub.go

@@ -260,18 +260,18 @@ func NewPubSub(ctx context.Context, h host.Host, rt PubSubRouter, opts ...Option
 		peerFilter:            DefaultPeerFilter,
 		disc:                  &discover{},
 		maxMessageSize:        DefaultMaxMessageSize,
-		peerOutboundQueueSize: 128,
+		peerOutboundQueueSize: 32,
 		signID:                h.ID(),
 		signKey:               nil,
 		signPolicy:            StrictSign,
-		incoming:              make(chan *RPC, 128),
+		incoming:              make(chan *RPC, 32),
 		newPeers:              make(chan struct{}, 1),
 		newPeersPend:          make(map[peer.ID]struct{}),
 		newPeerStream:         make(chan network.Stream),
 		newPeerError:          make(chan peer.ID),
 		peerDead:              make(chan struct{}, 1),
 		peerDeadPend:          make(map[peer.ID]struct{}),
-		deadPeerBackoff:       newBackoff(ctx, 128, BackoffCleanupInterval, MaxBackoffAttempts),
+		deadPeerBackoff:       newBackoff(ctx, 1000, BackoffCleanupInterval, MaxBackoffAttempts),
 		cancelCh:              make(chan *Subscription),
 		getPeers:              make(chan *listPeerReq),
 		addSub:                make(chan *addSubReq),
@@ -280,7 +280,7 @@ func NewPubSub(ctx context.Context, h host.Host, rt PubSubRouter, opts ...Option
 		addBitmask:            make(chan *addBitmaskReq),
 		rmBitmask:             make(chan *rmBitmaskReq),
 		getBitmasks:           make(chan *bitmaskReq),
-		sendMsg:               make(chan *Message, 128),
+		sendMsg:               make(chan *Message, 32),
 		addVal:                make(chan *addValReq),
 		rmVal:                 make(chan *rmValReq),
 		eval:                  make(chan func()),
@@ -979,7 +979,7 @@ func (p *PubSub) notifySubs(msg *Message) {
 	for f := range subs {
 		select {
 		case f.ch <- msg:
-		case <-time.After(15 * time.Millisecond):
+		case <-time.After(5 * time.Millisecond):
 			// it's unreasonable to immediately fall over because a subscriber didn't
 			// answer, message delivery sometimes lands next nanosecond and dropping
 			// it when there's room is absurd.

node/config/version.go

@@ -14,7 +14,7 @@ func GetMinimumVersion() []byte {
 }
 
 func GetVersion() []byte {
-	return []byte{0x01, 0x04, 0x06}
+	return []byte{0x01, 0x04, 0x07}
 }
 
 func GetVersionString() string {

node/consensus/ceremony/broadcast_messaging.go

@@ -3,16 +3,16 @@ package ceremony
 import (
 	"bytes"
 	"encoding/binary"
-	"source.quilibrium.com/quilibrium/monorepo/node/config"
 	"strings"
 	"time"
 
+	"source.quilibrium.com/quilibrium/monorepo/node/config"
+
 	"github.com/iden3/go-iden3-crypto/poseidon"
 	pcrypto "github.com/libp2p/go-libp2p/core/crypto"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/pkg/errors"
 	"go.uber.org/zap"
-	"golang.org/x/sync/errgroup"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/anypb"
 	"source.quilibrium.com/quilibrium/monorepo/go-libp2p-blossomsub/pb"
@@ -20,6 +20,98 @@ import (
 	"source.quilibrium.com/quilibrium/monorepo/node/protobufs"
 )
 
+func (e *CeremonyDataClockConsensusEngine) runMessageHandler() {
+	for {
+		select {
+		case message := <-e.messageProcessorCh:
+			msg := &protobufs.Message{}
+
+			if err := proto.Unmarshal(message.Data, msg); err != nil {
+				continue
+			}
+
+			for name := range e.executionEngines {
+				name := name
+				go func() error {
+					messages, err := e.executionEngines[name].ProcessMessage(
+						msg.Address,
+						msg,
+					)
+					if err != nil {
+						e.logger.Debug(
+							"could not process message for engine",
+							zap.Error(err),
+							zap.String("engine_name", name),
+						)
+						return nil
+					}
+
+					for _, appMessage := range messages {
+						appMsg := &anypb.Any{}
+						err := proto.Unmarshal(appMessage.Payload, appMsg)
+						if err != nil {
+							e.logger.Error(
+								"could not unmarshal app message",
+								zap.Error(err),
+								zap.String("engine_name", name),
+							)
+							continue
+						}
+
+						switch appMsg.TypeUrl {
+						case protobufs.CeremonyLobbyStateTransitionType:
+							t := &protobufs.CeremonyLobbyStateTransition{}
+							err := proto.Unmarshal(appMsg.Value, t)
+							if err != nil {
+								continue
+							}
+
+							if err := e.handleCeremonyLobbyStateTransition(t); err != nil {
+								continue
+							}
+						}
+					}
+
+					return nil
+				}()
+			}
+
+			any := &anypb.Any{}
+			if err := proto.Unmarshal(msg.Payload, any); err != nil {
+				continue
+			}
+
+			go func() {
+				switch any.TypeUrl {
+				case protobufs.ClockFrameType:
+					e.peerMapMx.RLock()
+					if peer, ok := e.peerMap[string(message.From)]; !ok ||
+						bytes.Compare(peer.version, config.GetMinimumVersion()) < 0 {
+						return
+					}
+					e.peerMapMx.RUnlock()
+					if err := e.handleClockFrameData(
+						message.From,
+						msg.Address,
+						any,
+						false,
+					); err != nil {
+						return
+					}
+				case protobufs.CeremonyPeerListAnnounceType:
+					if err := e.handleCeremonyPeerListAnnounce(
+						message.From,
+						msg.Address,
+						any,
+					); err != nil {
+						return
+					}
+				}
+			}()
+		}
+	}
+}
+
 func (e *CeremonyDataClockConsensusEngine) handleMessage(
 	message *pb.Message,
 ) error {
@@ -29,96 +121,10 @@ func (e *CeremonyDataClockConsensusEngine) handleMessage(
 		zap.Binary("from", message.From),
 		zap.Binary("signature", message.Signature),
 	)
-	msg := &protobufs.Message{}
 
-	if err := proto.Unmarshal(message.Data, msg); err != nil {
+	go func() {
-		return errors.Wrap(err, "handle message")
+		e.messageProcessorCh <- message
-	}
+	}()
-
-	eg := errgroup.Group{}
-	eg.SetLimit(len(e.executionEngines))
-
-	for name := range e.executionEngines {
-		name := name
-		eg.Go(func() error {
-			messages, err := e.executionEngines[name].ProcessMessage(
-				msg.Address,
-				msg,
-			)
-			if err != nil {
-				e.logger.Debug(
-					"could not process message for engine",
-					zap.Error(err),
-					zap.String("engine_name", name),
-				)
-				return errors.Wrap(err, "handle message")
-			}
-
-			for _, appMessage := range messages {
-				appMsg := &anypb.Any{}
-				err := proto.Unmarshal(appMessage.Payload, appMsg)
-				if err != nil {
-					e.logger.Error(
-						"could not unmarshal app message",
-						zap.Error(err),
-						zap.String("engine_name", name),
-					)
-					return errors.Wrap(err, "handle message")
-				}
-
-				switch appMsg.TypeUrl {
-				case protobufs.CeremonyLobbyStateTransitionType:
-					t := &protobufs.CeremonyLobbyStateTransition{}
-					err := proto.Unmarshal(appMsg.Value, t)
-					if err != nil {
-						return errors.Wrap(err, "handle message")
-					}
-
-					if err := e.handleCeremonyLobbyStateTransition(t); err != nil {
-						return errors.Wrap(err, "handle message")
-					}
-				}
-			}
-
-			return nil
-		})
-	}
-
-	if err := eg.Wait(); err != nil {
-		e.logger.Debug("rejecting invalid message", zap.Error(err))
-		return nil
-	}
-
-	any := &anypb.Any{}
-	if err := proto.Unmarshal(msg.Payload, any); err != nil {
-		return errors.Wrap(err, "handle message")
-	}
-
-	switch any.TypeUrl {
-	case protobufs.ClockFrameType:
-		e.peerMapMx.RLock()
-		if peer, ok := e.peerMap[string(message.From)]; !ok ||
-			bytes.Compare(peer.version, config.GetMinimumVersion()) < 0 {
-			return nil
-		}
-		e.peerMapMx.RUnlock()
-		if err := e.handleClockFrameData(
-			message.From,
-			msg.Address,
-			any,
-			false,
-		); err != nil {
-			return errors.Wrap(err, "handle message")
-		}
-	case protobufs.CeremonyPeerListAnnounceType:
-		if err := e.handleCeremonyPeerListAnnounce(
-			message.From,
-			msg.Address,
-			any,
-		); err != nil {
-			return errors.Wrap(err, "handle message")
-		}
-	}
 
 	return nil
 }

node/consensus/ceremony/ceremony_data_clock_consensus_engine.go

@@ -16,6 +16,7 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/protobuf/types/known/anypb"
+	"source.quilibrium.com/quilibrium/monorepo/go-libp2p-blossomsub/pb"
 	"source.quilibrium.com/quilibrium/monorepo/nekryptology/pkg/core/curves"
 	"source.quilibrium.com/quilibrium/monorepo/node/config"
 	"source.quilibrium.com/quilibrium/monorepo/node/consensus"
@@ -103,6 +104,7 @@ type CeremonyDataClockConsensusEngine struct {
 	lastKeyBundleAnnouncementFrame uint64
 	peerMap                        map[string]*peerInfo
 	uncooperativePeersMap          map[string]*peerInfo
+	messageProcessorCh             chan *pb.Message
 }
 
 var _ consensus.DataConsensusEngine = (*CeremonyDataClockConsensusEngine)(nil)
@@ -233,6 +235,7 @@ func NewCeremonyDataClockConsensusEngine(
 		masterTimeReel:            masterTimeReel,
 		dataTimeReel:              dataTimeReel,
 		statsClient:               statsClient,
+		messageProcessorCh:        make(chan *pb.Message, 128),
 	}
 
 	logger.Info("constructing consensus engine")
@@ -270,6 +273,10 @@ func (e *CeremonyDataClockConsensusEngine) Start() <-chan error {
 		panic(err)
 	}
 
+	go func() {
+		e.runLoop()
+	}()
+
 	e.logger.Info("subscribing to pubsub messages")
 	e.pubSub.Subscribe(e.filter, e.handleMessage, true)
 

node/consensus/ceremony/consensus_frames.go

@@ -4,9 +4,11 @@ import (
 	"bytes"
 	"context"
 	"io"
-	"source.quilibrium.com/quilibrium/monorepo/node/config"
+	"math/big"
 	"time"
 
+	"source.quilibrium.com/quilibrium/monorepo/node/config"
+
 	"github.com/pkg/errors"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
@@ -165,7 +167,10 @@ func (e *CeremonyDataClockConsensusEngine) GetMostAheadPeer() (
 		_, ok := e.uncooperativePeersMap[string(v.peerId)]
 		if v.maxFrame > max &&
 			v.timestamp > config.GetMinimumVersionCutoff().UnixMilli() &&
-			bytes.Compare(v.version, config.GetMinimumVersion()) >= 0 && !ok {
+			bytes.Compare(v.version, config.GetMinimumVersion()) >= 0 &&
+			new(big.Int).SetBytes(
+				v.totalDistance,
+			).Cmp(e.dataTimeReel.GetTotalDistance()) < 0 && !ok {
 			peer = v.peerId
 			max = v.maxFrame
 		}

node/consensus/time/data_time_reel.go

@@ -615,6 +615,14 @@ func (d *DataTimeReel) setHead(frame *protobufs.ClockFrame, distance *big.Int) {
 
 	d.head = frame
 	d.totalDistance.Add(d.totalDistance, distance)
+
+	d.clockStore.SetTotalDistance(
+		d.filter,
+		frame.FrameNumber,
+		selector.FillBytes(make([]byte, 32)),
+		d.totalDistance,
+	)
+
 	d.headDistance = distance
 	go func() {
 		d.newFrameCh <- frame
@@ -623,7 +631,21 @@ func (d *DataTimeReel) setHead(frame *protobufs.ClockFrame, distance *big.Int) {
 
 // tag: dusk – store the distance with the frame
 func (d *DataTimeReel) getTotalDistance(frame *protobufs.ClockFrame) *big.Int {
-	total, err := d.GetDistance(frame)
+	selector, err := frame.GetSelector()
+	if err != nil {
+		panic(err)
+	}
+
+	total, err := d.clockStore.GetTotalDistance(
+		d.filter,
+		frame.FrameNumber,
+		selector.FillBytes(make([]byte, 32)),
+	)
+	if err == nil && total != nil {
+		return total
+	}
+
+	total, err = d.GetDistance(frame)
 	if err != nil {
 		panic(err)
 	}
@@ -643,6 +665,13 @@ func (d *DataTimeReel) getTotalDistance(frame *protobufs.ClockFrame) *big.Int {
 		total.Add(total, distance)
 	}
 
+	d.clockStore.SetTotalDistance(
+		d.filter,
+		frame.FrameNumber,
+		selector.FillBytes(make([]byte, 32)),
+		total,
+	)
+
 	return total
 }
 
@@ -883,6 +912,14 @@ func (d *DataTimeReel) forkChoice(
 		"set total distance",
 		zap.String("total_distance", d.totalDistance.Text(16)),
 	)
+
+	d.clockStore.SetTotalDistance(
+		d.filter,
+		frame.FrameNumber,
+		selector.FillBytes(make([]byte, 32)),
+		d.totalDistance,
+	)
+
 	go func() {
 		d.newFrameCh <- frame
 	}()

node/main.go

@@ -209,6 +209,14 @@ func RunSelfTestIfNeeded(
 
 	cores := runtime.GOMAXPROCS(0)
 	memory := memory.TotalMemory()
+	d, err := os.Stat(filepath.Join(configDir, "store"))
+	if d == nil {
+		err := os.Mkdir(filepath.Join(configDir, "store"), 0755)
+		if err != nil {
+			panic(err)
+		}
+	}
+
 	f, err := os.Stat(filepath.Join(configDir, "SELF_TEST"))
 
 	if f != nil {

node/p2p/blossomsub.go

@@ -22,7 +22,6 @@ import (
 	"github.com/libp2p/go-libp2p/core/protocol"
 	"github.com/libp2p/go-libp2p/p2p/discovery/routing"
 	"github.com/libp2p/go-libp2p/p2p/discovery/util"
-	"github.com/libp2p/go-libp2p/p2p/net/connmgr"
 	"github.com/mr-tron/base58"
 	"github.com/pkg/errors"
 	"go.uber.org/zap"
@@ -107,17 +106,6 @@ func NewBlossomSub(
 		}
 	}
 
-	if isBootstrapPeer {
-		mgr, err := connmgr.NewConnManager(512, 8192)
-		if err != nil {
-			panic(err)
-		}
-
-		opts = append(opts,
-			libp2p.ConnectionManager(mgr),
-		)
-	}
-
 	var privKey crypto.PrivKey
 	if p2pConfig.PeerPrivKey != "" {
 		peerPrivKey, err := hex.DecodeString(p2pConfig.PeerPrivKey)

node/store/clock.go

@@ -3,6 +3,7 @@ package store
 import (
 	"bytes"
 	"encoding/binary"
+	"math/big"
 
 	"github.com/cockroachdb/pebble"
 	"github.com/iden3/go-iden3-crypto/poseidon"
@@ -76,6 +77,17 @@ type ClockStore interface {
 		masterFilter []byte,
 		dataFilter []byte,
 	) error
+	GetTotalDistance(
+		filter []byte,
+		frameNumber uint64,
+		selector []byte,
+	) (*big.Int, error)
+	SetTotalDistance(
+		filter []byte,
+		frameNumber uint64,
+		selector []byte,
+		totalDistance *big.Int,
+	) error
 }
 
 type PebbleClockStore struct {
@@ -268,6 +280,7 @@ const CLOCK_MASTER_FRAME_DATA = 0x00
 const CLOCK_DATA_FRAME_DATA = 0x01
 const CLOCK_DATA_FRAME_CANDIDATE_DATA = 0x02
 const CLOCK_DATA_FRAME_FRECENCY_DATA = 0x03
+const CLOCK_DATA_FRAME_DISTANCE_DATA = 0x04
 const CLOCK_MASTER_FRAME_INDEX_EARLIEST = 0x10 | CLOCK_MASTER_FRAME_DATA
 const CLOCK_MASTER_FRAME_INDEX_LATEST = 0x20 | CLOCK_MASTER_FRAME_DATA
 const CLOCK_MASTER_FRAME_INDEX_PARENT = 0x30 | CLOCK_MASTER_FRAME_DATA
@@ -402,6 +415,18 @@ func clockProverTrieKey(filter []byte, frameNumber uint64) []byte {
 	return key
 }
 
+func clockDataTotalDistanceKey(
+	filter []byte,
+	frameNumber uint64,
+	selector []byte,
+) []byte {
+	key := []byte{CLOCK_FRAME, CLOCK_DATA_FRAME_DISTANCE_DATA}
+	key = binary.BigEndian.AppendUint64(key, frameNumber)
+	key = append(key, filter...)
+	key = append(key, rightAlign(selector, 32)...)
+	return key
+}
+
 func (p *PebbleClockStore) NewTransaction() (Transaction, error) {
 	return p.db.NewBatch(), nil
 }
@@ -1287,3 +1312,39 @@ func (p *PebbleClockStore) Compact(
 
 	return nil
 }
+
+func (p *PebbleClockStore) GetTotalDistance(
+	filter []byte,
+	frameNumber uint64,
+	selector []byte,
+) (*big.Int, error) {
+	value, closer, err := p.db.Get(
+		clockDataTotalDistanceKey(filter, frameNumber, selector),
+	)
+	if err != nil {
+		if errors.Is(err, pebble.ErrNotFound) {
+			return nil, ErrNotFound
+		}
+
+		return nil, errors.Wrap(err, "get total distance")
+	}
+
+	defer closer.Close()
+	dist := new(big.Int).SetBytes(value)
+
+	return dist, nil
+}
+
+func (p *PebbleClockStore) SetTotalDistance(
+	filter []byte,
+	frameNumber uint64,
+	selector []byte,
+	totalDistance *big.Int,
+) error {
+	err := p.db.Set(
+		clockDataTotalDistanceKey(filter, frameNumber, selector),
+		totalDistance.Bytes(),
+	)
+
+	return errors.Wrap(err, "set total distance")
+}