diff --git a/go-libp2p-blossomsub/blossomsub.go b/go-libp2p-blossomsub/blossomsub.go index 2f3a521..9005a1e 100644 --- a/go-libp2p-blossomsub/blossomsub.go +++ b/go-libp2p-blossomsub/blossomsub.go @@ -33,7 +33,7 @@ var ( BlossomSubHistoryGossip = 3 BlossomSubDlazy = 6 BlossomSubGossipFactor = 0.25 - BlossomSubGossipRetransmission = 3 + BlossomSubGossipRetransmission = 1 BlossomSubHeartbeatInitialDelay = 100 * time.Millisecond BlossomSubHeartbeatInterval = 1 * time.Second BlossomSubFanoutTTL = 60 * time.Second @@ -703,9 +703,6 @@ func (bs *BlossomSubRouter) handleIHave(p peer.ID, ctl *pb.ControlMessage) []*pb iwantlst = append(iwantlst, mid) } - // ask in random order - shuffleStrings(iwantlst) - // truncate to the messages we are actually asking for and update the iasked counter iwantlst = iwantlst[:iask] bs.iasked[p] += iask diff --git a/node/config/version.go b/node/config/version.go index 1db14af..b98b164 100644 --- a/node/config/version.go +++ b/node/config/version.go @@ -10,11 +10,11 @@ func GetMinimumVersionCutoff() time.Time { } func GetMinimumVersion() []byte { - return []byte{0x01, 0x04, 0x0F} + return []byte{0x01, 0x04, 0x10} } func GetVersion() []byte { - return []byte{0x01, 0x04, 0x0F} + return []byte{0x01, 0x04, 0x10} } func GetVersionString() string { diff --git a/node/consensus/ceremony/broadcast_messaging.go b/node/consensus/ceremony/broadcast_messaging.go index 4350540..faf2984 100644 --- a/node/consensus/ceremony/broadcast_messaging.go +++ b/node/consensus/ceremony/broadcast_messaging.go @@ -1,7 +1,9 @@ package ceremony import ( + "encoding/binary" "strings" + "time" "github.com/iden3/go-iden3-crypto/poseidon" "github.com/pkg/errors" @@ -9,6 +11,7 @@ import ( "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/anypb" "source.quilibrium.com/quilibrium/monorepo/go-libp2p-blossomsub/pb" + "source.quilibrium.com/quilibrium/monorepo/node/config" "source.quilibrium.com/quilibrium/monorepo/node/protobufs" ) @@ -36,11 +39,46 @@ func (e *CeremonyDataClockConsensusEngine) publishProof( peers, max, err := e.GetMostAheadPeer(head.FrameNumber) if err != nil || len(peers) == 0 || head.FrameNumber > max { - if err := e.publishMessage(e.filter, frame); err != nil { - return errors.Wrap( - err, - "publish proof", - ) + timestamp := time.Now().UnixMilli() + msg := binary.BigEndian.AppendUint64([]byte{}, frame.FrameNumber) + msg = append(msg, config.GetVersion()...) + msg = binary.BigEndian.AppendUint64(msg, uint64(timestamp)) + sig, err := e.pubSub.SignMessage(msg) + if err != nil { + panic(err) + } + + e.peerMapMx.Lock() + e.peerMap[string(e.pubSub.GetPeerID())] = &peerInfo{ + peerId: e.pubSub.GetPeerID(), + multiaddr: "", + maxFrame: frame.FrameNumber, + version: config.GetVersion(), + signature: sig, + publicKey: e.pubSub.GetPublicKey(), + timestamp: timestamp, + totalDistance: e.dataTimeReel.GetTotalDistance().FillBytes( + make([]byte, 256), + ), + } + list := &protobufs.CeremonyPeerListAnnounce{ + PeerList: []*protobufs.CeremonyPeer{}, + } + list.PeerList = append(list.PeerList, &protobufs.CeremonyPeer{ + PeerId: e.pubSub.GetPeerID(), + Multiaddr: "", + MaxFrame: frame.FrameNumber, + Version: config.GetVersion(), + Signature: sig, + PublicKey: e.pubSub.GetPublicKey(), + Timestamp: timestamp, + TotalDistance: e.dataTimeReel.GetTotalDistance().FillBytes( + make([]byte, 256), + ), + }) + e.peerMapMx.Unlock() + if err := e.publishMessage(e.filter, list); err != nil { + e.logger.Debug("error publishing message", zap.Error(err)) } } diff --git a/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go b/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go index 0894162..d40b440 100644 --- a/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go +++ b/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go @@ -489,7 +489,6 @@ func (e *CeremonyDataClockConsensusEngine) runLoop() { } } case <-time.After(20 * time.Second): - e.logger.Info("no frames received, kicking off") dataFrame, err := e.dataTimeReel.Head() if err != nil { panic(err) diff --git a/node/consensus/master/broadcast_messaging.go b/node/consensus/master/broadcast_messaging.go index 53bde8d..ea5c998 100644 --- a/node/consensus/master/broadcast_messaging.go +++ b/node/consensus/master/broadcast_messaging.go @@ -219,11 +219,10 @@ func (e *MasterClockConsensusEngine) publishProof( if err != nil || len(peers) == 0 { // publish if we don't see anyone (empty peer list) or if we're the most // ahead: - if err := e.publishMessage(e.filter, frame); err != nil { - return errors.Wrap( - err, - "publish proof", - ) + e.report.MasterHeadFrame = frame.FrameNumber + + if err := e.publishMessage(e.filter, e.report); err != nil { + e.logger.Debug("error publishing message", zap.Error(err)) } } diff --git a/node/p2p/blossomsub.go b/node/p2p/blossomsub.go index 6cf05ca..fb59efa 100644 --- a/node/p2p/blossomsub.go +++ b/node/p2p/blossomsub.go @@ -60,7 +60,7 @@ var BITMASK_ALL = []byte{ // While we iterate through these next phases, we're going to aggressively // enforce keeping updated. This will be achieved through announce strings // that will vary with each update -var ANNOUNCE = "quilibrium-1.4.8-sunset-ceasefire" +var ANNOUNCE = "quilibrium-1.4.16-sunset-noctilucent" func getPeerID(p2pConfig *config.P2PConfig) peer.ID { peerPrivKey, err := hex.DecodeString(p2pConfig.PeerPrivKey) @@ -191,7 +191,7 @@ func NewBlossomSub( GossipThreshold: -2000, PublishThreshold: -5000, GraylistThreshold: -10000, - AcceptPXThreshold: 100, + AcceptPXThreshold: 1, OpportunisticGraftThreshold: 2, })) @@ -351,18 +351,8 @@ func initDHT( ) defaultBootstrapPeers := p2pConfig.BootstrapPeers - connected := 0 - for connected < 2 { - i, err := rand.Int( - rand.Reader, - big.NewInt(int64(len(defaultBootstrapPeers))), - ) - if err != nil { - panic(err) - } - - peerAddr := defaultBootstrapPeers[i.Int64()] + for _, peerAddr := range defaultBootstrapPeers { peerinfo, err := peer.AddrInfoFromString(peerAddr) if err != nil { panic(err) @@ -370,7 +360,7 @@ func initDHT( if peerinfo.ID == h.ID() || h.Network().Connectedness(peerinfo.ID) == network.Connected || - connected >= 2 { + h.Network().Connectedness(peerinfo.ID) == network.CannotConnect { continue } @@ -381,7 +371,6 @@ func initDHT( "connected to peer", zap.String("peer_id", peerinfo.ID.String()), ) - connected++ } } } @@ -391,12 +380,8 @@ func initDHT( go func() { for { time.Sleep(30 * time.Second) - // try to assert some stability, never go below min peers for data - // consensus: - if len(h.Network().Peers()) < 2 { - logger.Info("reconnecting to peers") - reconnect() - } + logger.Debug("reconnecting to peers") + reconnect() } }() @@ -567,7 +552,8 @@ func discoverPeers( for peer := range peerChan { peer := peer if peer.ID == h.ID() || - h.Network().Connectedness(peer.ID) == network.Connected { + h.Network().Connectedness(peer.ID) == network.Connected || + h.Network().Connectedness(peer.ID) == network.CannotConnect { continue } @@ -593,7 +579,7 @@ func discoverPeers( go func() { for { time.Sleep(5 * time.Minute) - if len(h.Network().Peers()) < 4 { + if len(h.Network().Peers()) < 16 { discover() } }