From f48f177a10a376375b7adafbc6e33582b215061e Mon Sep 17 00:00:00 2001 From: Cassandra Heart <7929478+CassOnMars@users.noreply.github.com> Date: Wed, 10 Jan 2024 00:58:38 -0600 Subject: [PATCH] v1.2.2 - sneaky sync troubles (#35) --- node/app/db_console.go | 4 +- .../ceremony_data_clock_consensus_engine.go | 11 ++- node/consensus/ceremony/consensus_frames.go | 86 ++++++++++++++----- node/consensus/ceremony/peer_messaging.go | 7 +- node/consensus/consensus_engine.go | 2 +- node/main.go | 2 +- 6 files changed, 84 insertions(+), 28 deletions(-) diff --git a/node/app/db_console.go b/node/app/db_console.go index 7c13e31..b04b409 100644 --- a/node/app/db_console.go +++ b/node/app/db_console.go @@ -818,11 +818,11 @@ func logoVersion(width int) string { out += " ####################################### ########\n" out += " ############################# ##\n" out += " \n" - out += " Quilibrium Node - v1.2.1 – Dawn\n" + out += " Quilibrium Node - v1.2.2 – Dawn\n" out += " \n" out += " DB Console\n" } else { - out = "Quilibrium Node - v1.2.1 – Dawn - DB Console\n" + out = "Quilibrium Node - v1.2.2 – Dawn - DB Console\n" } return out } diff --git a/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go b/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go index fa40990..90a1f94 100644 --- a/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go +++ b/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go @@ -249,6 +249,8 @@ func (e *CeremonyDataClockConsensusEngine) Start() <-chan error { } go func() { + thresholdBeforeConfirming := 4 + for { time.Sleep(30 * time.Second) @@ -291,7 +293,8 @@ func (e *CeremonyDataClockConsensusEngine) Start() <-chan error { if v == nil { continue } - if v.timestamp <= time.Now().UnixMilli()-UNCOOPERATIVE_PEER_INFO_TTL { + if v.timestamp <= time.Now().UnixMilli()-UNCOOPERATIVE_PEER_INFO_TTL || + thresholdBeforeConfirming > 0 { deletes = append(deletes, v) } } @@ -303,11 +306,17 @@ func (e *CeremonyDataClockConsensusEngine) Start() <-chan error { if err := e.publishMessage(e.filter, list); err != nil { e.logger.Debug("error publishing message", zap.Error(err)) } + + if thresholdBeforeConfirming > 0 { + thresholdBeforeConfirming-- + } } }() go func() { e.logger.Info("waiting for peer list mappings") + // We need to re-tune this so that libp2p's peerstore activation threshold + // considers DHT peers to be correct: time.Sleep(30 * time.Second) for e.state < consensus.EngineStateStopping { peerCount := e.pubSub.GetNetworkPeersCount() diff --git a/node/consensus/ceremony/consensus_frames.go b/node/consensus/ceremony/consensus_frames.go index a1f29b3..b3e15fc 100644 --- a/node/consensus/ceremony/consensus_frames.go +++ b/node/consensus/ceremony/consensus_frames.go @@ -12,6 +12,7 @@ import ( "math/big" "os" "strings" + "time" "github.com/iden3/go-iden3-crypto/ff" "github.com/iden3/go-iden3-crypto/poseidon" @@ -913,20 +914,16 @@ func (e *CeremonyDataClockConsensusEngine) GetMostAheadPeer() ( max := e.frame.FrameNumber var peer []byte = nil for _, v := range e.peerMap { - if v.maxFrame > max { + _, ok := e.uncooperativePeersMap[string(v.peerId)] + if v.maxFrame > max && !ok { peer = v.peerId max = v.maxFrame } } - size := len(e.peerMap) e.peerMapMx.Unlock() if peer == nil { - if size > 1 { - return nil, 0, nil - } else { - return nil, 0, p2p.ErrNoPeersAvailable - } + return nil, 0, p2p.ErrNoPeersAvailable } return peer, max, nil @@ -948,10 +945,11 @@ func (e *CeremonyDataClockConsensusEngine) sync( e.peerMapMx.Lock() if _, ok := e.peerMap[string(peerId)]; ok { e.uncooperativePeersMap[string(peerId)] = e.peerMap[string(peerId)] + e.uncooperativePeersMap[string(peerId)].timestamp = time.Now().UnixMilli() delete(e.peerMap, string(peerId)) } e.peerMapMx.Unlock() - return latest, errors.Wrap(err, "reverse optimistic sync") + return latest, errors.Wrap(err, "sync") } client := protobufs.NewCeremonyServiceClient(cc) @@ -981,11 +979,15 @@ func (e *CeremonyDataClockConsensusEngine) sync( e.peerMapMx.Lock() if _, ok := e.peerMap[string(peerId)]; ok { e.uncooperativePeersMap[string(peerId)] = e.peerMap[string(peerId)] + e.uncooperativePeersMap[string(peerId)].timestamp = time.Now().UnixMilli() delete(e.peerMap, string(peerId)) } e.peerMapMx.Unlock() - return latest, errors.Wrap(err, "reverse optimistic sync") + return latest, errors.Wrap(err, "sync") } + + firstPass := true + var syncMsg *protobufs.CeremonyCompressedSync for syncMsg, err = s.Recv(); err == nil; syncMsg, err = s.Recv() { e.logger.Info( @@ -995,6 +997,31 @@ func (e *CeremonyDataClockConsensusEngine) sync( zap.Int("frames", len(syncMsg.TruncatedClockFrames)), zap.Int("proofs", len(syncMsg.Proofs)), ) + + // This can only happen if we get a peer with state that was initially + // farther ahead, but something happened. However, this has a sticking + // effect that doesn't go away for them until they're caught up again, + // so let's not penalize their score and make everyone else suffer, + // let's just move on: + if syncMsg.FromFrameNumber == 0 && + syncMsg.ToFrameNumber == 0 && + firstPass { + if err := cc.Close(); err != nil { + e.logger.Error("error while closing connection", zap.Error(err)) + } + + e.peerMapMx.Lock() + if _, ok := e.peerMap[string(peerId)]; ok { + e.uncooperativePeersMap[string(peerId)] = e.peerMap[string(peerId)] + e.uncooperativePeersMap[string(peerId)].timestamp = time.Now(). + UnixMilli() + delete(e.peerMap, string(peerId)) + } + e.peerMapMx.Unlock() + + return currentLatest, errors.Wrap(ErrNoNewFrames, "sync") + } + var next *protobufs.ClockFrame if next, err = e.decompressAndStoreCandidates( peerId, @@ -1008,6 +1035,8 @@ func (e *CeremonyDataClockConsensusEngine) sync( e.peerMapMx.Lock() if _, ok := e.peerMap[string(peerId)]; ok { e.uncooperativePeersMap[string(peerId)] = e.peerMap[string(peerId)] + e.uncooperativePeersMap[string(peerId)].timestamp = time.Now(). + UnixMilli() delete(e.peerMap, string(peerId)) } e.peerMapMx.Unlock() @@ -1016,18 +1045,28 @@ func (e *CeremonyDataClockConsensusEngine) sync( e.logger.Error("error while closing connection", zap.Error(err)) } - return currentLatest, errors.Wrap(err, "reverse optimistic sync") + return currentLatest, errors.Wrap(err, "sync") } if next != nil { latest = next } } if err != nil && err != io.EOF && !errors.Is(err, ErrNoNewFrames) { + e.logger.Error("error while receiving sync", zap.Error(err)) + if err := cc.Close(); err != nil { e.logger.Error("error while closing connection", zap.Error(err)) } - e.logger.Error("error while receiving sync", zap.Error(err)) - return latest, errors.Wrap(err, "reverse optimistic sync") + + e.peerMapMx.Lock() + if _, ok := e.peerMap[string(peerId)]; ok { + e.uncooperativePeersMap[string(peerId)] = e.peerMap[string(peerId)] + e.uncooperativePeersMap[string(peerId)].timestamp = time.Now().UnixMilli() + delete(e.peerMap, string(peerId)) + } + e.peerMapMx.Unlock() + + return latest, errors.Wrap(err, "sync") } e.logger.Info( @@ -1052,13 +1091,20 @@ func (e *CeremonyDataClockConsensusEngine) collect( latest = e.previousHead e.syncingStatus = SyncStatusNotSyncing } - peerId, maxFrame, err := e.GetMostAheadPeer() - if err != nil { - e.logger.Warn("no peers available, skipping sync") - } else if peerId == nil { - e.logger.Info("currently up to date, skipping sync") - } else if maxFrame-2 > latest.FrameNumber { - latest, err = e.sync(latest, maxFrame, peerId) + for { + peerId, maxFrame, err := e.GetMostAheadPeer() + if err != nil { + e.logger.Warn("no peers available, skipping sync") + break + } else if peerId == nil { + e.logger.Info("currently up to date, skipping sync") + break + } else if maxFrame-2 > latest.FrameNumber { + latest, err = e.sync(latest, maxFrame, peerId) + if err == nil { + break + } + } } e.logger.Info( @@ -1068,7 +1114,7 @@ func (e *CeremonyDataClockConsensusEngine) collect( e.logger.Info("selecting leader") - latest, err = e.commitLongestPath(latest) + latest, err := e.commitLongestPath(latest) if err != nil { e.logger.Error("could not collect longest path", zap.Error(err)) latest, _, err = e.clockStore.GetDataClockFrame(e.filter, 0) diff --git a/node/consensus/ceremony/peer_messaging.go b/node/consensus/ceremony/peer_messaging.go index 48de54b..df30bf0 100644 --- a/node/consensus/ceremony/peer_messaging.go +++ b/node/consensus/ceremony/peer_messaging.go @@ -85,7 +85,7 @@ func (e *CeremonyDataClockConsensusEngine) GetCompressedSyncFrames( ) if err != nil { from = 1 - e.logger.Info("peer fully out of sync, rewinding sync head to start") + e.logger.Debug("peer fully out of sync, rewinding sync head to start") break } @@ -95,8 +95,8 @@ func (e *CeremonyDataClockConsensusEngine) GetCompressedSyncFrames( parent, ) if err != nil { - from = 1 - e.logger.Info("peer fully out of sync, rewinding sync head to start") + from = frame.FrameNumber - 16 + e.logger.Debug("peer fully out of sync, rewinding sync head to min") break } @@ -167,6 +167,7 @@ func (e *CeremonyDataClockConsensusEngine) decompressAndStoreCandidates( e.peerMapMx.Lock() if _, ok := e.peerMap[string(peerId)]; ok { e.uncooperativePeersMap[string(peerId)] = e.peerMap[string(peerId)] + e.uncooperativePeersMap[string(peerId)].timestamp = time.Now().UnixMilli() delete(e.peerMap, string(peerId)) } e.peerMapMx.Unlock() diff --git a/node/consensus/consensus_engine.go b/node/consensus/consensus_engine.go index 02649b5..ae32cea 100644 --- a/node/consensus/consensus_engine.go +++ b/node/consensus/consensus_engine.go @@ -59,5 +59,5 @@ func GetMinimumVersion() []byte { } func GetVersion() []byte { - return []byte{0x01, 0x02, 0x01} + return []byte{0x01, 0x02, 0x02} } diff --git a/node/main.go b/node/main.go index c2bd6b9..3f6fb07 100644 --- a/node/main.go +++ b/node/main.go @@ -249,5 +249,5 @@ func printLogo() { func printVersion() { fmt.Println(" ") - fmt.Println(" Quilibrium Node - v1.2.1 – Dawn") + fmt.Println(" Quilibrium Node - v1.2.2 – Dawn") }