From 7ccd9f9ab094d7a700d7c44c27a471110a76f41d Mon Sep 17 00:00:00 2001 From: Cassandra Heart Date: Sat, 16 Mar 2024 19:33:48 -0500 Subject: [PATCH] fix: nodes with long-tail syncs or not in validator trie hit OOM due to accumulation of state transition messages that do not clear --- .../ceremony_data_clock_consensus_engine.go | 31 ++++++++++--------- node/consensus/ceremony/consensus_frames.go | 6 ++-- node/consensus/ceremony/message_handler.go | 6 +++- 3 files changed, 25 insertions(+), 18 deletions(-) diff --git a/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go b/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go index 033f46a..94a3fd5 100644 --- a/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go +++ b/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go @@ -460,17 +460,17 @@ func (e *CeremonyDataClockConsensusEngine) runLoop() { e.frameChan <- latestFrame }() + var nextFrame *protobufs.ClockFrame + if nextFrame, err = e.prove(latestFrame); err != nil { + e.logger.Error("could not prove", zap.Error(err)) + e.state = consensus.EngineStateCollecting + continue + } + if bytes.Equal( e.frameProverTrie.FindNearest(e.provingKeyAddress).External.Key, e.provingKeyAddress, ) { - var nextFrame *protobufs.ClockFrame - if nextFrame, err = e.prove(latestFrame); err != nil { - e.logger.Error("could not prove", zap.Error(err)) - e.state = consensus.EngineStateCollecting - continue - } - e.dataTimeReel.Insert(nextFrame) if err = e.publishProof(nextFrame); err != nil { @@ -502,14 +502,17 @@ func (e *CeremonyDataClockConsensusEngine) runLoop() { e.frameChan <- latestFrame }() - if e.frameProverTrie.Contains(e.provingKeyAddress) { - var nextFrame *protobufs.ClockFrame - if nextFrame, err = e.prove(latestFrame); err != nil { - e.logger.Error("could not prove", zap.Error(err)) - e.state = consensus.EngineStateCollecting - continue - } + var nextFrame *protobufs.ClockFrame + if nextFrame, err = e.prove(latestFrame); err != nil { + e.logger.Error("could not prove", zap.Error(err)) + e.state = consensus.EngineStateCollecting + continue + } + if bytes.Equal( + e.frameProverTrie.FindNearest(e.provingKeyAddress).External.Key, + e.provingKeyAddress, + ) { e.dataTimeReel.Insert(nextFrame) if err = e.publishProof(nextFrame); err != nil { diff --git a/node/consensus/ceremony/consensus_frames.go b/node/consensus/ceremony/consensus_frames.go index 1245c7f..dda317d 100644 --- a/node/consensus/ceremony/consensus_frames.go +++ b/node/consensus/ceremony/consensus_frames.go @@ -454,9 +454,6 @@ func (e *CeremonyDataClockConsensusEngine) collect( e.logger.Info("collecting vdf proofs") latest := currentFramePublished - if e.syncingStatus == SyncStatusFailed { - e.syncingStatus = SyncStatusNotSyncing - } // With the increase of network size, constrain down to top thirty for i := 0; i < 30; i++ { @@ -468,6 +465,7 @@ func (e *CeremonyDataClockConsensusEngine) collect( e.logger.Info("currently up to date, skipping sync") break } else if maxFrame-2 > latest.FrameNumber { + e.syncingStatus = SyncStatusSynchronizing latest, err = e.sync(latest, maxFrame, peerId) if err == nil { break @@ -475,6 +473,8 @@ func (e *CeremonyDataClockConsensusEngine) collect( } } + e.syncingStatus = SyncStatusNotSyncing + if latest.FrameNumber < currentFramePublished.FrameNumber { latest = currentFramePublished } diff --git a/node/consensus/ceremony/message_handler.go b/node/consensus/ceremony/message_handler.go index 880148e..045a740 100644 --- a/node/consensus/ceremony/message_handler.go +++ b/node/consensus/ceremony/message_handler.go @@ -30,7 +30,11 @@ func (e *CeremonyDataClockConsensusEngine) runMessageHandler() { peer, ok := e.peerMap[string(message.From)] e.peerMapMx.RUnlock() - if ok && bytes.Compare(peer.version, config.GetMinimumVersion()) >= 0 { + if ok && bytes.Compare(peer.version, config.GetMinimumVersion()) >= 0 && + bytes.Equal( + e.frameProverTrie.FindNearest(e.provingKeyAddress).External.Key, + e.provingKeyAddress, + ) && e.syncingStatus == SyncStatusNotSyncing { for name := range e.executionEngines { name := name go func() error {