ceremonyclient/node/consensus/time/data_time_reel.go

856 lines
20 KiB
Go
Raw Normal View History

2024-02-13 07:04:56 +00:00
package time
import (
"bytes"
2024-02-18 04:52:19 +00:00
"encoding/hex"
2024-02-13 07:04:56 +00:00
"math/big"
"sort"
2024-02-13 07:04:56 +00:00
"sync"
2024-02-16 21:46:54 +00:00
lru "github.com/hashicorp/golang-lru/v2"
2024-02-13 07:04:56 +00:00
"github.com/pkg/errors"
"go.uber.org/zap"
"source.quilibrium.com/quilibrium/monorepo/node/config"
"source.quilibrium.com/quilibrium/monorepo/node/crypto"
"source.quilibrium.com/quilibrium/monorepo/node/protobufs"
"source.quilibrium.com/quilibrium/monorepo/node/store"
"source.quilibrium.com/quilibrium/monorepo/node/tries"
)
var allBitmaskFilter = []byte{
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
}
var unknownDistance = new(big.Int).SetBytes([]byte{
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
})
type pendingFrame struct {
2024-02-18 04:52:19 +00:00
selector *big.Int
2024-02-13 07:04:56 +00:00
parentSelector *big.Int
}
type DataTimeReel struct {
rwMutex sync.RWMutex
2024-02-16 21:46:54 +00:00
running bool
2024-02-13 07:04:56 +00:00
filter []byte
engineConfig *config.EngineConfig
logger *zap.Logger
clockStore store.ClockStore
frameProver crypto.FrameProver
parentTimeReel TimeReel
origin []byte
initialInclusionProof *crypto.InclusionAggregateProof
initialProverKeys [][]byte
head *protobufs.ClockFrame
totalDistance *big.Int
headDistance *big.Int
2024-02-16 21:46:54 +00:00
lruFrames *lru.Cache[string, string]
2024-02-13 07:04:56 +00:00
proverTrie *tries.RollingFrecencyCritbitTrie
pending map[uint64][]*pendingFrame
incompleteForks map[uint64][]*pendingFrame
frames chan *protobufs.ClockFrame
newFrameCh chan *protobufs.ClockFrame
badFrameCh chan *protobufs.ClockFrame
done chan bool
}
func NewDataTimeReel(
filter []byte,
logger *zap.Logger,
clockStore store.ClockStore,
engineConfig *config.EngineConfig,
frameProver crypto.FrameProver,
origin []byte,
initialInclusionProof *crypto.InclusionAggregateProof,
initialProverKeys [][]byte,
) *DataTimeReel {
if filter == nil {
panic("filter is nil")
}
if logger == nil {
panic("logger is nil")
}
if clockStore == nil {
panic("clock store is nil")
}
if engineConfig == nil {
panic("engine config is nil")
}
if frameProver == nil {
panic("frame prover is nil")
}
2024-02-16 21:46:54 +00:00
cache, err := lru.New[string, string](10000)
if err != nil {
panic(err)
}
2024-02-13 07:04:56 +00:00
return &DataTimeReel{
2024-02-16 21:46:54 +00:00
running: false,
2024-02-13 07:04:56 +00:00
logger: logger,
filter: filter,
engineConfig: engineConfig,
clockStore: clockStore,
frameProver: frameProver,
origin: origin,
initialInclusionProof: initialInclusionProof,
initialProverKeys: initialProverKeys,
2024-02-16 21:46:54 +00:00
lruFrames: cache,
2024-02-13 07:04:56 +00:00
pending: make(map[uint64][]*pendingFrame),
incompleteForks: make(map[uint64][]*pendingFrame),
frames: make(chan *protobufs.ClockFrame),
newFrameCh: make(chan *protobufs.ClockFrame),
badFrameCh: make(chan *protobufs.ClockFrame),
done: make(chan bool),
}
}
func (d *DataTimeReel) Start() error {
trie := &tries.RollingFrecencyCritbitTrie{}
frame, err := d.clockStore.GetLatestDataClockFrame(d.filter, trie)
if err != nil && !errors.Is(err, store.ErrNotFound) {
panic(err)
}
if frame == nil {
d.head, d.proverTrie = d.createGenesisFrame()
d.totalDistance = big.NewInt(0)
2024-02-16 21:46:54 +00:00
d.headDistance = big.NewInt(0)
2024-02-13 07:04:56 +00:00
} else {
d.head = frame
2024-02-16 21:46:54 +00:00
if err != nil {
panic(err)
}
2024-02-13 07:04:56 +00:00
d.proverTrie = trie
2024-02-16 21:46:54 +00:00
d.headDistance, err = d.GetDistance(frame)
2024-02-13 07:04:56 +00:00
d.totalDistance = d.getTotalDistance(frame)
}
go d.runLoop()
return nil
}
func (d *DataTimeReel) Head() (*protobufs.ClockFrame, error) {
return d.head, nil
}
// Insert enqueues a structurally valid frame into the time reel. If the frame
// 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 {
2024-02-16 21:46:54 +00:00
if !d.running {
return nil
}
2024-02-18 04:52:19 +00:00
d.logger.Debug(
"insert frame",
zap.Uint64("frame_number", frame.FrameNumber),
zap.String("output_tag", hex.EncodeToString(frame.Output[:64])),
)
2024-02-16 21:46:54 +00:00
if d.lruFrames.Contains(string(frame.Output[:64])) {
return nil
}
d.lruFrames.Add(string(frame.Output[:64]), string(frame.ParentSelector))
2024-02-13 07:04:56 +00:00
go func() {
d.frames <- frame
}()
return nil
}
func (d *DataTimeReel) GetFrameProverTrie() *tries.RollingFrecencyCritbitTrie {
return d.proverTrie
}
func (d *DataTimeReel) NewFrameCh() <-chan *protobufs.ClockFrame {
return d.newFrameCh
}
func (d *DataTimeReel) BadFrameCh() <-chan *protobufs.ClockFrame {
return d.badFrameCh
}
func (d *DataTimeReel) Stop() {
d.done <- true
}
func (d *DataTimeReel) createGenesisFrame() (
*protobufs.ClockFrame,
*tries.RollingFrecencyCritbitTrie,
) {
if d.origin == nil {
panic("origin is nil")
}
if d.initialInclusionProof == nil {
panic("initial inclusion proof is nil")
}
if d.initialProverKeys == nil {
panic("initial prover keys is nil")
}
2024-02-14 07:11:12 +00:00
difficulty := d.engineConfig.Difficulty
if difficulty == 0 {
difficulty = 10000
}
2024-02-13 07:04:56 +00:00
frame, trie, err := d.frameProver.CreateDataGenesisFrame(
d.filter,
d.origin,
2024-02-14 07:11:12 +00:00
difficulty,
2024-02-13 07:04:56 +00:00
d.initialInclusionProof,
d.initialProverKeys,
true,
)
if err != nil {
panic(err)
}
txn, err := d.clockStore.NewTransaction()
if err != nil {
panic(err)
}
if err := d.clockStore.PutDataClockFrame(
frame,
trie,
txn,
false,
); err != nil {
panic(err)
}
if err := txn.Commit(); err != nil {
panic(err)
}
return frame, trie
}
// Main data consensus loop
func (d *DataTimeReel) runLoop() {
2024-02-16 21:46:54 +00:00
d.running = true
2024-02-13 07:04:56 +00:00
for {
select {
case frame := <-d.frames:
2024-02-18 04:52:19 +00:00
d.logger.Debug(
"processing frame",
zap.Uint64("frame_number", frame.FrameNumber),
zap.String("output_tag", hex.EncodeToString(frame.Output[:64])),
zap.Uint64("head_number", d.head.FrameNumber),
zap.String("head_output_tag", hex.EncodeToString(d.head.Output[:64])),
)
2024-02-13 07:04:56 +00:00
// Most common scenario: in order new frame is higher number
if d.head.FrameNumber < frame.FrameNumber {
2024-02-18 04:52:19 +00:00
d.logger.Debug("frame is higher")
2024-02-13 07:04:56 +00:00
parent := new(big.Int).SetBytes(frame.ParentSelector)
selector, err := frame.GetSelector()
if err != nil {
panic(err)
}
distance, err := d.GetDistance(frame)
if err != nil {
2024-02-18 04:52:19 +00:00
d.logger.Debug("no master, add pending")
2024-02-13 07:04:56 +00:00
// If the frame arrived ahead of a master, e.g. the master data is not
// synced, we'll go ahead and mark it as pending and process it when
// we can, but if we had a general fault, panic:
if !errors.Is(err, store.ErrNotFound) {
panic(err)
}
2024-02-18 04:52:19 +00:00
d.addPending(selector, parent, distance, frame)
2024-02-21 08:10:23 +00:00
d.processPending(d.head, frame)
2024-02-13 07:04:56 +00:00
continue
}
2024-02-16 09:42:37 +00:00
headSelector, err := d.head.GetSelector()
if err != nil {
panic(err)
}
// If the frame has a gap from the head or is not descendent, mark it as
// pending:
if frame.FrameNumber-d.head.FrameNumber != 1 ||
parent.Cmp(headSelector) != 0 {
2024-02-18 04:52:19 +00:00
d.logger.Debug(
2024-02-21 08:10:23 +00:00
"frame has has gap or is non-descendent, fork choice",
2024-02-18 04:52:19 +00:00
zap.Bool("has_gap", frame.FrameNumber-d.head.FrameNumber != 1),
zap.String("parent_selector", parent.Text(16)),
zap.String("head_selector", headSelector.Text(16)),
)
2024-02-21 08:10:23 +00:00
d.forkChoice(frame, distance)
d.processPending(d.head, frame)
2024-02-13 07:04:56 +00:00
continue
}
// Otherwise set it as the next and process all pending
d.setHead(frame, distance)
2024-02-21 08:10:23 +00:00
d.processPending(d.head, frame)
2024-02-13 07:04:56 +00:00
} else if d.head.FrameNumber == frame.FrameNumber {
// frames are equivalent, no need to act
if bytes.Equal(d.head.Output, frame.Output) {
2024-02-18 04:52:19 +00:00
d.logger.Debug("equivalent frame")
2024-02-21 08:10:23 +00:00
d.processPending(d.head, frame)
2024-02-13 07:04:56 +00:00
continue
}
distance, err := d.GetDistance(frame)
if err != nil {
panic(err)
}
2024-02-18 04:52:19 +00:00
d.logger.Debug(
"frame is same height",
zap.String("head_distance", d.headDistance.Text(16)),
zap.String("distance", distance.Text(16)),
)
2024-02-13 07:04:56 +00:00
// Optimization: if competing frames share a parent we can short-circuit
// fork choice
if bytes.Equal(d.head.ParentSelector, frame.ParentSelector) &&
distance.Cmp(d.headDistance) < 0 {
2024-02-18 04:52:19 +00:00
d.logger.Debug(
"frame shares parent, has shorter distance, short circuit",
)
2024-02-13 07:04:56 +00:00
d.totalDistance.Sub(d.totalDistance, d.headDistance)
d.setHead(frame, distance)
2024-02-21 08:10:23 +00:00
d.processPending(d.head, frame)
2024-02-16 09:42:37 +00:00
continue
2024-02-13 07:04:56 +00:00
}
// Choose fork
d.forkChoice(frame, distance)
2024-02-21 08:10:23 +00:00
d.processPending(d.head, frame)
2024-02-13 07:04:56 +00:00
} else {
2024-02-18 04:52:19 +00:00
d.logger.Debug("frame is lower height")
2024-02-13 07:04:56 +00:00
// tag: dusk we should have some kind of check here to avoid brutal
// thrashing
existing, _, err := d.clockStore.GetDataClockFrame(
d.filter,
frame.FrameNumber,
)
if err != nil {
// if this returns an error it's either not found (which shouldn't
// happen without corruption) or pebble is borked, either way, panic
panic(err)
}
// It's a fork, but it's behind. We need to stash it until it catches
// up (or dies off)
if !bytes.Equal(existing.Output, frame.Output) {
2024-02-18 04:52:19 +00:00
d.logger.Debug("is fork, add pending")
2024-02-13 07:04:56 +00:00
distance, err := d.GetDistance(frame)
if err != nil {
panic(err)
}
parent, selector, err := frame.GetParentAndSelector()
if err != nil {
panic(err)
}
d.addPending(selector, parent, distance, frame)
2024-02-21 08:10:23 +00:00
d.processPending(d.head, frame)
2024-02-13 07:04:56 +00:00
}
}
case <-d.done:
return
}
}
}
func (d *DataTimeReel) addPending(
selector *big.Int,
parent *big.Int,
distance *big.Int,
frame *protobufs.ClockFrame,
) {
2024-02-18 04:52:19 +00:00
d.logger.Debug(
"add pending",
zap.Uint64("head_frame_number", d.head.FrameNumber),
zap.Uint64("add_frame_number", frame.FrameNumber),
zap.String("selector", selector.Text(16)),
zap.String("parent", parent.Text(16)),
zap.String("distance", distance.Text(16)),
)
2024-02-13 07:04:56 +00:00
2024-02-18 04:52:19 +00:00
if d.head.FrameNumber <= frame.FrameNumber {
if _, ok := d.pending[frame.FrameNumber]; !ok {
d.pending[frame.FrameNumber] = []*pendingFrame{}
2024-02-16 21:46:54 +00:00
}
2024-02-13 07:04:56 +00:00
2024-02-18 04:52:19 +00:00
// avoid heavy thrashing
for _, frame := range d.pending[frame.FrameNumber] {
if frame.selector.Cmp(selector) == 0 {
d.logger.Debug("exists in pending already")
return
}
}
2024-02-13 07:04:56 +00:00
}
2024-02-16 21:46:54 +00:00
// avoid db thrashing
if existing, err := d.clockStore.GetParentDataClockFrame(
frame.Filter,
frame.FrameNumber,
2024-02-13 07:04:56 +00:00
selector.FillBytes(make([]byte, 32)),
2024-02-22 06:07:17 +00:00
true,
2024-02-16 21:46:54 +00:00
); err != nil && existing == nil {
2024-02-18 04:52:19 +00:00
d.logger.Debug(
"not stored yet, save data candidate",
zap.Uint64("frame_number", frame.FrameNumber),
zap.String("selector", selector.Text(16)),
zap.String("parent", parent.Text(16)),
zap.String("distance", distance.Text(16)),
)
2024-02-16 21:46:54 +00:00
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)
}
2024-02-13 07:04:56 +00:00
}
2024-02-18 04:52:19 +00:00
if d.head.FrameNumber <= frame.FrameNumber {
d.logger.Debug(
"accumulate in pending",
zap.Int("pending_neighbors", len(d.pending[frame.FrameNumber])),
)
2024-02-13 07:04:56 +00:00
2024-02-18 04:52:19 +00:00
d.pending[frame.FrameNumber] = append(
d.pending[frame.FrameNumber],
&pendingFrame{
selector: selector,
parentSelector: parent,
},
2024-02-13 07:04:56 +00:00
)
}
2024-02-18 04:52:19 +00:00
}
2024-02-13 07:04:56 +00:00
2024-02-18 04:52:19 +00:00
func (d *DataTimeReel) processPending(
frame *protobufs.ClockFrame,
2024-02-21 08:10:23 +00:00
lastReceived *protobufs.ClockFrame,
2024-02-18 04:52:19 +00:00
) {
2024-02-21 08:10:23 +00:00
d.logger.Debug(
"process pending",
zap.Int("pending_frame_numbers", len(d.pending)),
)
frameNumbers := []uint64{}
for f := range d.pending {
frameNumbers = append(frameNumbers, f)
2024-02-21 08:10:23 +00:00
d.logger.Debug(
"pending per frame number",
zap.Uint64("pending_frame_number", f),
zap.Int("pending_frames", len(d.pending[f])),
)
}
sort.Slice(frameNumbers, func(i, j int) bool {
2024-02-21 08:10:23 +00:00
return frameNumbers[i] > frameNumbers[j]
})
2024-02-21 08:10:23 +00:00
lastSelector, err := lastReceived.GetSelector()
if err != nil {
panic(err)
}
for _, f := range frameNumbers {
if f < d.head.FrameNumber {
delete(d.pending, f)
}
nextPending := d.pending[f]
2024-02-18 04:52:19 +00:00
d.logger.Debug(
"checking frame set",
zap.Uint64("pending_frame_number", f),
zap.Uint64("frame_number", frame.FrameNumber),
)
if f < frame.FrameNumber {
d.logger.Debug(
"purging frame set",
zap.Uint64("pending_frame_number", f),
zap.Uint64("frame_number", frame.FrameNumber),
2024-02-13 07:04:56 +00:00
)
2024-02-18 04:52:19 +00:00
delete(d.pending, f)
continue
2024-02-16 09:42:37 +00:00
}
2024-02-18 04:52:19 +00:00
// Pull the next
2024-02-21 08:10:23 +00:00
for len(nextPending) != 0 {
2024-02-18 04:52:19 +00:00
d.logger.Debug("try process next")
2024-02-16 09:42:37 +00:00
next := nextPending[0]
2024-02-21 08:10:23 +00:00
d.pending[f] = d.pending[f][1:]
if f == lastReceived.FrameNumber && next.selector.Cmp(lastSelector) == 0 {
d.pending[f] = append(d.pending[f], next)
if len(d.pending[f]) == 1 {
nextPending = nil
}
continue
}
2024-02-16 09:42:37 +00:00
2024-02-18 04:52:19 +00:00
nextFrame, err := d.clockStore.GetParentDataClockFrame(
2024-02-16 09:42:37 +00:00
d.filter,
2024-02-18 04:52:19 +00:00
f,
next.selector.FillBytes(make([]byte, 32)),
2024-02-22 06:07:17 +00:00
false,
2024-02-16 09:42:37 +00:00
)
if err != nil && !errors.Is(err, store.ErrNotFound) {
panic(err)
}
2024-02-13 07:04:56 +00:00
if nextFrame != nil {
2024-02-18 04:52:19 +00:00
d.logger.Debug("next found, send frame back in")
2024-02-13 07:04:56 +00:00
go func() {
d.frames <- nextFrame
}()
2024-02-21 08:10:23 +00:00
return
2024-02-13 07:04:56 +00:00
}
2024-02-18 04:52:19 +00:00
if len(d.pending[f]) == 0 {
d.logger.Debug("last next processing, clear list")
delete(d.pending, f)
2024-02-21 08:10:23 +00:00
nextPending = nil
2024-02-18 04:52:19 +00:00
}
2024-02-13 07:04:56 +00:00
}
}
}
func (d *DataTimeReel) setHead(frame *protobufs.ClockFrame, distance *big.Int) {
2024-02-18 04:52:19 +00:00
d.logger.Debug(
"set frame to head",
zap.Uint64("frame_number", frame.FrameNumber),
zap.String("output_tag", hex.EncodeToString(frame.Output[:64])),
zap.Uint64("head_number", d.head.FrameNumber),
zap.String("head_output_tag", hex.EncodeToString(d.head.Output[:64])),
)
2024-02-13 07:04:56 +00:00
txn, err := d.clockStore.NewTransaction()
if err != nil {
panic(err)
}
2024-02-18 04:52:19 +00:00
d.logger.Debug(
"save data",
zap.Uint64("frame_number", frame.FrameNumber),
zap.String("distance", distance.Text(16)),
)
2024-02-13 07:04:56 +00:00
if err := d.clockStore.PutDataClockFrame(
frame,
d.proverTrie,
txn,
false,
); err != nil {
panic(err)
}
if err = txn.Commit(); err != nil {
panic(err)
}
d.head = frame
d.totalDistance.Add(d.totalDistance, distance)
d.headDistance = distance
go func() {
d.newFrameCh <- frame
}()
}
// tag: dusk store the distance with the frame
func (d *DataTimeReel) getTotalDistance(frame *protobufs.ClockFrame) *big.Int {
total, err := d.GetDistance(frame)
if err != nil {
panic(err)
}
for index := frame; err == nil &&
index.FrameNumber > 0; index, err = d.clockStore.GetParentDataClockFrame(
d.filter,
index.FrameNumber-1,
index.ParentSelector,
2024-02-22 06:07:17 +00:00
true,
2024-02-13 07:04:56 +00:00
) {
distance, err := d.GetDistance(index)
if err != nil {
panic(err)
}
total.Add(total, distance)
}
return total
}
func (d *DataTimeReel) GetDistance(frame *protobufs.ClockFrame) (
*big.Int,
error,
) {
// tag: equinox master filter changes
master, err := d.clockStore.GetMasterClockFrame(
allBitmaskFilter,
frame.FrameNumber)
if err != nil {
2024-02-18 04:52:19 +00:00
return unknownDistance, errors.Wrap(err, "get distance")
2024-02-13 07:04:56 +00:00
}
masterSelector, err := master.GetSelector()
if err != nil {
2024-02-18 04:52:19 +00:00
return unknownDistance, errors.Wrap(err, "get distance")
2024-02-13 07:04:56 +00:00
}
discriminatorNode :=
d.proverTrie.FindNearest(masterSelector.FillBytes(make([]byte, 32)))
discriminator := discriminatorNode.External.Key
addr, err := frame.GetAddress()
if err != nil {
2024-02-18 04:52:19 +00:00
return unknownDistance, errors.Wrap(err, "get distance")
2024-02-13 07:04:56 +00:00
}
distance := new(big.Int).Sub(
new(big.Int).SetBytes(discriminator),
new(big.Int).SetBytes(addr),
)
distance.Abs(distance)
return distance, nil
}
func (d *DataTimeReel) forkChoice(
frame *protobufs.ClockFrame,
distance *big.Int,
) {
2024-02-18 04:52:19 +00:00
d.logger.Debug(
"fork choice",
zap.Uint64("frame_number", frame.FrameNumber),
zap.String("output_tag", hex.EncodeToString(frame.Output[:64])),
zap.Uint64("head_number", d.head.FrameNumber),
zap.String("head_output_tag", hex.EncodeToString(d.head.Output[:64])),
)
2024-02-13 07:04:56 +00:00
parentSelector, selector, err := frame.GetParentAndSelector()
if err != nil {
panic(err)
}
leftIndex := d.head
rightIndex := frame
leftTotal := new(big.Int).Set(d.headDistance)
2024-02-21 08:10:23 +00:00
overweight := big.NewInt(0)
2024-02-13 07:04:56 +00:00
rightTotal := new(big.Int).Set(distance)
left := d.head.ParentSelector
right := frame.ParentSelector
2024-02-16 09:42:37 +00:00
rightReplaySelectors := [][]byte{}
2024-02-13 07:04:56 +00:00
2024-02-21 08:10:23 +00:00
for rightIndex.FrameNumber > leftIndex.FrameNumber {
rightReplaySelectors = append(
append(
[][]byte{},
right,
),
rightReplaySelectors...,
)
rightIndex, err = d.clockStore.GetParentDataClockFrame(
d.filter,
rightIndex.FrameNumber-1,
rightIndex.ParentSelector,
2024-02-22 06:07:17 +00:00
true,
2024-02-21 08:10:23 +00:00
)
if err != nil {
// If lineage cannot be verified, set it for later
if errors.Is(err, store.ErrNotFound) {
d.addPending(selector, parentSelector, distance, frame)
return
} else {
panic(err)
}
}
right = rightIndex.ParentSelector
rightIndexDistance, err := d.GetDistance(rightIndex)
if err != nil {
panic(err)
}
// We accumulate right on left when right is longer because we cannot know
// where the left will lead and don't want it to disadvantage our comparison
overweight.Add(overweight, rightIndexDistance)
rightTotal.Add(rightTotal, rightIndexDistance)
}
2024-02-13 07:04:56 +00:00
// Walk backwards through the parents, until we find a matching parent
// selector:
for !bytes.Equal(left, right) {
2024-02-21 08:10:23 +00:00
d.logger.Debug(
"scan backwards",
zap.String("left_parent", hex.EncodeToString(leftIndex.ParentSelector)),
zap.String("right_parent", hex.EncodeToString(rightIndex.ParentSelector)),
)
2024-02-18 04:52:19 +00:00
2024-02-13 07:04:56 +00:00
rightReplaySelectors = append(
append(
[][]byte{},
right,
),
rightReplaySelectors...,
)
leftIndex, err = d.clockStore.GetParentDataClockFrame(
d.filter,
leftIndex.FrameNumber-1,
leftIndex.ParentSelector,
2024-02-22 06:07:17 +00:00
true,
2024-02-13 07:04:56 +00:00
)
if err != nil {
panic(err)
}
rightIndex, err = d.clockStore.GetParentDataClockFrame(
d.filter,
rightIndex.FrameNumber-1,
rightIndex.ParentSelector,
2024-02-22 06:07:17 +00:00
true,
2024-02-13 07:04:56 +00:00
)
if err != nil {
// If lineage cannot be verified, set it for later
if errors.Is(err, store.ErrNotFound) {
d.addPending(selector, parentSelector, distance, frame)
return
} else {
panic(err)
}
}
left = leftIndex.ParentSelector
right = rightIndex.ParentSelector
leftIndexDistance, err := d.GetDistance(leftIndex)
if err != nil {
panic(err)
}
rightIndexDistance, err := d.GetDistance(rightIndex)
if err != nil {
panic(err)
}
leftTotal.Add(leftTotal, leftIndexDistance)
rightTotal.Add(rightTotal, rightIndexDistance)
}
2024-02-18 04:52:19 +00:00
d.logger.Debug("found mutual root")
2024-02-13 07:04:56 +00:00
2024-02-16 09:42:37 +00:00
frameNumber := rightIndex.FrameNumber
2024-02-21 08:10:23 +00:00
overweight.Add(overweight, leftTotal)
2024-02-13 07:04:56 +00:00
// Choose new fork based on lightest distance sub-tree
2024-02-21 08:10:23 +00:00
if rightTotal.Cmp(overweight) > 0 {
2024-02-18 04:52:19 +00:00
d.logger.Debug("proposed fork has greater distance",
zap.String("right_total", rightTotal.Text(16)),
2024-02-21 08:10:23 +00:00
zap.String("left_total", overweight.Text(16)),
2024-02-18 04:52:19 +00:00
)
d.addPending(selector, parentSelector, distance, frame)
return
}
2024-02-13 07:04:56 +00:00
2024-02-18 04:52:19 +00:00
for {
if len(rightReplaySelectors) == 0 {
break
}
next := rightReplaySelectors[0]
rightReplaySelectors =
rightReplaySelectors[1:]
2024-02-16 09:42:37 +00:00
2024-02-18 04:52:19 +00:00
rightIndex, err = d.clockStore.GetParentDataClockFrame(
d.filter,
frameNumber,
next,
2024-02-22 06:07:17 +00:00
false,
2024-02-18 04:52:19 +00:00
)
if err != nil {
panic(err)
2024-02-16 09:42:37 +00:00
}
txn, err := d.clockStore.NewTransaction()
if err != nil {
panic(err)
}
if err := d.clockStore.PutDataClockFrame(
2024-02-18 04:52:19 +00:00
rightIndex,
2024-02-16 09:42:37 +00:00
d.proverTrie,
txn,
false,
); err != nil {
panic(err)
}
if err = txn.Commit(); err != nil {
panic(err)
2024-02-13 07:04:56 +00:00
}
2024-02-18 04:52:19 +00:00
frameNumber++
}
txn, err := d.clockStore.NewTransaction()
if err != nil {
panic(err)
2024-02-13 07:04:56 +00:00
}
2024-02-18 04:52:19 +00:00
if err := d.clockStore.PutDataClockFrame(
frame,
d.proverTrie,
txn,
false,
); err != nil {
panic(err)
}
if err = txn.Commit(); err != nil {
panic(err)
}
d.head = frame
d.totalDistance.Sub(d.totalDistance, leftTotal)
d.totalDistance.Add(d.totalDistance, rightTotal)
d.headDistance = distance
d.logger.Debug(
"set total distance",
zap.String("total_distance", d.totalDistance.Text(16)),
)
go func() {
d.newFrameCh <- frame
}()
2024-02-13 07:04:56 +00:00
}
var _ TimeReel = (*DataTimeReel)(nil)