diff --git a/node/consensus/ceremony/broadcast_messaging.go b/node/consensus/ceremony/broadcast_messaging.go index 31aff40..8331909 100644 --- a/node/consensus/ceremony/broadcast_messaging.go +++ b/node/consensus/ceremony/broadcast_messaging.go @@ -41,7 +41,7 @@ func (e *CeremonyDataClockConsensusEngine) publishProof( panic(err) } - peers, max, err := e.GetMostAheadPeer() + 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( diff --git a/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go b/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go index 2e332b0..e97f6f4 100644 --- a/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go +++ b/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go @@ -566,6 +566,7 @@ func (e *CeremonyDataClockConsensusEngine) Stop(force bool) <-chan error { wg.Wait() e.logger.Info("execution engines stopped") + e.dataTimeReel.Stop() e.state = consensus.EngineStateStopped e.engineMx.Lock() diff --git a/node/consensus/ceremony/consensus_frames.go b/node/consensus/ceremony/consensus_frames.go index 2b279d6..ae33ec9 100644 --- a/node/consensus/ceremony/consensus_frames.go +++ b/node/consensus/ceremony/consensus_frames.go @@ -143,24 +143,21 @@ func (e *CeremonyDataClockConsensusEngine) prove( return frame, nil } -func (e *CeremonyDataClockConsensusEngine) GetMostAheadPeer() ( +func (e *CeremonyDataClockConsensusEngine) GetMostAheadPeer( + frameNumber uint64, +) ( []byte, uint64, error, ) { - frame, err := e.dataTimeReel.Head() - if err != nil { - panic(err) - } - e.logger.Info( "checking peer list", zap.Int("peers", len(e.peerMap)), zap.Int("uncooperative_peers", len(e.uncooperativePeersMap)), - zap.Uint64("current_head_frame", frame.FrameNumber), + zap.Uint64("current_head_frame", frameNumber), ) - max := frame.FrameNumber + max := frameNumber var peer []byte = nil e.peerMapMx.RLock() for _, v := range e.peerMap { @@ -514,7 +511,7 @@ func (e *CeremonyDataClockConsensusEngine) collect( latest := currentFramePublished for { - peerId, maxFrame, err := e.GetMostAheadPeer() + peerId, maxFrame, err := e.GetMostAheadPeer(latest.FrameNumber) if maxFrame > latest.FrameNumber { e.syncingStatus = SyncStatusSynchronizing if err != nil { diff --git a/node/consensus/master/master_clock_consensus_engine.go b/node/consensus/master/master_clock_consensus_engine.go index 7450c55..b89ffb9 100644 --- a/node/consensus/master/master_clock_consensus_engine.go +++ b/node/consensus/master/master_clock_consensus_engine.go @@ -325,7 +325,7 @@ func (e *MasterClockConsensusEngine) performBandwidthTest(peerID []byte) { cc, err := e.pubSub.GetDirectChannel(peerID, "validation") if err != nil { - e.logger.Info( + e.logger.Debug( "could not connect to peer for validation", zap.String("peer_id", base58.Encode(peerID)), ) @@ -348,7 +348,7 @@ func (e *MasterClockConsensusEngine) performBandwidthTest(peerID []byte) { end := time.Now().UnixMilli() if err != nil && err != io.EOF { cc.Close() - e.logger.Info( + e.logger.Debug( "peer returned error", zap.String("peer_id", base58.Encode(peerID)), zap.Error(err), @@ -360,7 +360,7 @@ func (e *MasterClockConsensusEngine) performBandwidthTest(peerID []byte) { cc.Close() if !bytes.Equal(verification, validation.Validation) { - e.logger.Info( + e.logger.Debug( "peer provided invalid verification", zap.String("peer_id", base58.Encode(peerID)), ) @@ -370,7 +370,7 @@ func (e *MasterClockConsensusEngine) performBandwidthTest(peerID []byte) { } if end-start > 2000 { - e.logger.Info( + e.logger.Debug( "peer has slow bandwidth, scoring out", zap.String("peer_id", base58.Encode(peerID)), ) diff --git a/node/execution/intrinsics/ceremony/ceremony_execution_engine.go b/node/execution/intrinsics/ceremony/ceremony_execution_engine.go index 1fe04b1..502d297 100644 --- a/node/execution/intrinsics/ceremony/ceremony_execution_engine.go +++ b/node/execution/intrinsics/ceremony/ceremony_execution_engine.go @@ -458,11 +458,11 @@ func (e *CeremonyExecutionEngine) Start() <-chan error { } // Stop implements ExecutionEngine -func (*CeremonyExecutionEngine) Stop(force bool) <-chan error { +func (e *CeremonyExecutionEngine) Stop(force bool) <-chan error { errChan := make(chan error) go func() { - errChan <- nil + errChan <- <-e.clock.Stop(force) }() return errChan diff --git a/node/store/clock.go b/node/store/clock.go index 0fc6804..bedf592 100644 --- a/node/store/clock.go +++ b/node/store/clock.go @@ -1251,6 +1251,21 @@ func (p *PebbleClockStore) Compact( ); err != nil { return errors.Wrap(err, "compact") } + + if err := p.db.DeleteRange( + clockDataParentIndexKey( + make([]byte, 32), + 0, + make([]byte, 32), + ), + clockDataParentIndexKey( + bytes.Repeat([]byte{0xff}, 32), + 0, + bytes.Repeat([]byte{0xff}, 32), + ), + ); err != nil { + return errors.Wrap(err, "compact") + } } if dataFilter != nil && !cleared { diff --git a/node/store/pebble.go b/node/store/pebble.go index a9072bd..07fbaea 100644 --- a/node/store/pebble.go +++ b/node/store/pebble.go @@ -78,7 +78,7 @@ func (p *PebbleDB) CompactAll() error { return errors.Wrap(err, "compact all") } - if err := p.Compact(first, last, true); err != nil { + if err := p.Compact(first, last, false); err != nil { return errors.Wrap(err, "compact all") }