diff --git a/node/app/db_console.go b/node/app/db_console.go index 40a4b9b..7ee5b61 100644 --- a/node/app/db_console.go +++ b/node/app/db_console.go @@ -886,11 +886,11 @@ func logoVersion(width int) string { out += " ####################################### ########\n" out += " ############################# ##\n" out += " \n" - out += " Quilibrium Node - v1.2.6 – Dawn\n" + out += " Quilibrium Node - v1.2.7 – Dawn\n" out += " \n" out += " DB Console\n" } else { - out = "Quilibrium Node - v1.2.6 – Dawn - DB Console\n" + out = "Quilibrium Node - v1.2.7 – 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 bd031fd..aefa019 100644 --- a/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go +++ b/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go @@ -425,6 +425,8 @@ func (e *CeremonyDataClockConsensusEngine) runLoop() { e.frameProverTrie.FindNearest(e.provingKeyAddress).External.Key, e.provingKeyAddress, ) { + e.dataTimeReel.Insert(nextFrame) + if err = e.publishProof(nextFrame); err != nil { e.logger.Error("could not publish", zap.Error(err)) e.state = consensus.EngineStateCollecting @@ -453,6 +455,8 @@ func (e *CeremonyDataClockConsensusEngine) runLoop() { } if e.frameProverTrie.Contains(e.provingKeyAddress) { + e.dataTimeReel.Insert(nextFrame) + if err = e.publishProof(nextFrame); err != nil { e.logger.Error("could not publish", zap.Error(err)) e.state = consensus.EngineStateCollecting diff --git a/node/consensus/ceremony/consensus_frames.go b/node/consensus/ceremony/consensus_frames.go index 88579c6..d196c18 100644 --- a/node/consensus/ceremony/consensus_frames.go +++ b/node/consensus/ceremony/consensus_frames.go @@ -346,6 +346,10 @@ func (e *CeremonyDataClockConsensusEngine) collect( } } + if latest.FrameNumber < currentFramePublished.FrameNumber { + latest = currentFramePublished + } + e.logger.Info( "returning leader frame", zap.Uint64("frame_number", latest.FrameNumber), diff --git a/node/consensus/consensus_engine.go b/node/consensus/consensus_engine.go index bc6bd61..94646b6 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, 0x06} + return []byte{0x01, 0x02, 0x07} } diff --git a/node/consensus/time/data_time_reel.go b/node/consensus/time/data_time_reel.go index ed1a535..b52c70f 100644 --- a/node/consensus/time/data_time_reel.go +++ b/node/consensus/time/data_time_reel.go @@ -5,6 +5,7 @@ import ( "math/big" "sync" + lru "github.com/hashicorp/golang-lru/v2" "github.com/pkg/errors" "go.uber.org/zap" "source.quilibrium.com/quilibrium/monorepo/node/config" @@ -35,6 +36,7 @@ type pendingFrame struct { type DataTimeReel struct { rwMutex sync.RWMutex + running bool filter []byte engineConfig *config.EngineConfig @@ -49,6 +51,7 @@ type DataTimeReel struct { head *protobufs.ClockFrame totalDistance *big.Int headDistance *big.Int + lruFrames *lru.Cache[string, string] proverTrie *tries.RollingFrecencyCritbitTrie pending map[uint64][]*pendingFrame incompleteForks map[uint64][]*pendingFrame @@ -88,7 +91,13 @@ func NewDataTimeReel( panic("frame prover is nil") } + cache, err := lru.New[string, string](10000) + if err != nil { + panic(err) + } + return &DataTimeReel{ + running: false, logger: logger, filter: filter, engineConfig: engineConfig, @@ -97,6 +106,7 @@ func NewDataTimeReel( origin: origin, initialInclusionProof: initialInclusionProof, initialProverKeys: initialProverKeys, + lruFrames: cache, pending: make(map[uint64][]*pendingFrame), incompleteForks: make(map[uint64][]*pendingFrame), frames: make(chan *protobufs.ClockFrame), @@ -116,9 +126,14 @@ func (d *DataTimeReel) Start() error { if frame == nil { d.head, d.proverTrie = d.createGenesisFrame() d.totalDistance = big.NewInt(0) + d.headDistance = big.NewInt(0) } else { d.head = frame + if err != nil { + panic(err) + } d.proverTrie = trie + d.headDistance, err = d.GetDistance(frame) d.totalDistance = d.getTotalDistance(frame) } @@ -135,6 +150,16 @@ func (d *DataTimeReel) Head() (*protobufs.ClockFrame, error) { // is the next one in sequence, it advances the reel head forward and emits a // new frame on the new frame channel. func (d *DataTimeReel) Insert(frame *protobufs.ClockFrame) error { + if !d.running { + return nil + } + + if d.lruFrames.Contains(string(frame.Output[:64])) { + return nil + } + + d.lruFrames.Add(string(frame.Output[:64]), string(frame.ParentSelector)) + go func() { d.frames <- frame }() @@ -214,6 +239,7 @@ func (d *DataTimeReel) createGenesisFrame() ( // Main data consensus loop func (d *DataTimeReel) runLoop() { + d.running = true for { select { case frame := <-d.frames: @@ -324,9 +350,11 @@ func (d *DataTimeReel) addPending( d.pending[frame.FrameNumber] = []*pendingFrame{} } - txn, err := d.clockStore.NewTransaction() - if err != nil { - panic(err) + // avoid heavy thrashing + for _, frame := range d.pending[frame.FrameNumber] { + if frame.parentSelector.Cmp(parent) == 0 { + return + } } if distance.Cmp(unknownDistance) == 0 { @@ -334,20 +362,30 @@ func (d *DataTimeReel) addPending( distance.Sub(distance, big.NewInt(int64(len(d.pending[frame.FrameNumber])))) } - err = d.clockStore.PutCandidateDataClockFrame( - parent.FillBytes(make([]byte, 32)), - distance.FillBytes(make([]byte, 32)), + // avoid db thrashing + if existing, err := d.clockStore.GetParentDataClockFrame( + frame.Filter, + frame.FrameNumber, selector.FillBytes(make([]byte, 32)), - frame, - txn, - ) - if err != nil { - txn.Abort() - panic(err) - } - - if err = txn.Commit(); err != nil { - panic(err) + ); err != nil && existing == nil { + txn, err := d.clockStore.NewTransaction() + if err != nil { + panic(err) + } + err = d.clockStore.PutCandidateDataClockFrame( + parent.FillBytes(make([]byte, 32)), + distance.FillBytes(make([]byte, 32)), + selector.FillBytes(make([]byte, 32)), + frame, + txn, + ) + if err != nil { + txn.Abort() + panic(err) + } + if err = txn.Commit(); err != nil { + panic(err) + } } d.pending[frame.FrameNumber] = append( diff --git a/node/go.mod b/node/go.mod index 6c1e63b..e5e19bd 100644 --- a/node/go.mod +++ b/node/go.mod @@ -97,7 +97,7 @@ require ( github.com/hashicorp/errwrap v1.1.0 // indirect github.com/hashicorp/go-multierror v1.1.1 // indirect github.com/hashicorp/golang-lru v0.5.4 // indirect - github.com/hashicorp/golang-lru/v2 v2.0.2 // indirect + github.com/hashicorp/golang-lru/v2 v2.0.2 github.com/huin/goupnp v1.2.0 // indirect github.com/iden3/go-iden3-crypto v0.0.15 github.com/ipfs/boxo v0.8.0 // indirect diff --git a/node/main.go b/node/main.go index e6ea937..b899ad2 100644 --- a/node/main.go +++ b/node/main.go @@ -284,5 +284,5 @@ func printLogo() { func printVersion() { fmt.Println(" ") - fmt.Println(" Quilibrium Node - v1.2.6 – Dawn") + fmt.Println(" Quilibrium Node - v1.2.7 – Dawn") }