ceremonyclient/node/consensus/time/data_time_reel.go
Cassandra Heart 1b810d624c
v1.2.7 (#48)
2024-02-16 15:46:54 -06:00

721 lines
16 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package time
import (
"bytes"
"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"
"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 {
parentSelector *big.Int
distance *big.Int
}
type DataTimeReel struct {
rwMutex sync.RWMutex
running bool
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
lruFrames *lru.Cache[string, string]
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")
}
cache, err := lru.New[string, string](10000)
if err != nil {
panic(err)
}
return &DataTimeReel{
running: false,
logger: logger,
filter: filter,
engineConfig: engineConfig,
clockStore: clockStore,
frameProver: frameProver,
origin: origin,
initialInclusionProof: initialInclusionProof,
initialProverKeys: initialProverKeys,
lruFrames: cache,
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)
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)
}
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 {
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
}()
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")
}
difficulty := d.engineConfig.Difficulty
if difficulty == 0 {
difficulty = 10000
}
frame, trie, err := d.frameProver.CreateDataGenesisFrame(
d.filter,
d.origin,
difficulty,
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() {
d.running = true
for {
select {
case frame := <-d.frames:
// Most common scenario: in order new frame is higher number
if d.head.FrameNumber < frame.FrameNumber {
parent := new(big.Int).SetBytes(frame.ParentSelector)
selector, err := frame.GetSelector()
if err != nil {
panic(err)
}
distance, err := d.GetDistance(frame)
if err != nil {
// 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)
}
d.addPending(selector, parent, unknownDistance, frame)
continue
}
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 {
d.addPending(selector, parent, distance, frame)
continue
}
// Otherwise set it as the next and process all pending
d.setHead(frame, distance)
d.processPending(frame)
} else if d.head.FrameNumber == frame.FrameNumber {
// frames are equivalent, no need to act
if bytes.Equal(d.head.Output, frame.Output) {
continue
}
distance, err := d.GetDistance(frame)
if err != nil {
panic(err)
}
// 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 {
d.totalDistance.Sub(d.totalDistance, d.headDistance)
d.setHead(frame, distance)
d.processPending(d.head)
continue
}
// Choose fork
d.forkChoice(frame, distance)
d.processPending(d.head)
} else {
// 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) {
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)
d.processPending(d.head)
}
}
case <-d.done:
return
}
}
}
func (d *DataTimeReel) addPending(
selector *big.Int,
parent *big.Int,
distance *big.Int,
frame *protobufs.ClockFrame,
) {
if _, ok := d.pending[frame.FrameNumber]; !ok {
d.pending[frame.FrameNumber] = []*pendingFrame{}
}
// avoid heavy thrashing
for _, frame := range d.pending[frame.FrameNumber] {
if frame.parentSelector.Cmp(parent) == 0 {
return
}
}
if distance.Cmp(unknownDistance) == 0 {
distance = new(big.Int).Set(unknownDistance)
distance.Sub(distance, big.NewInt(int64(len(d.pending[frame.FrameNumber]))))
}
// avoid db thrashing
if existing, err := d.clockStore.GetParentDataClockFrame(
frame.Filter,
frame.FrameNumber,
selector.FillBytes(make([]byte, 32)),
); 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(
d.pending[frame.FrameNumber],
&pendingFrame{
parentSelector: parent,
distance: distance,
},
)
}
func (d *DataTimeReel) processPending(frame *protobufs.ClockFrame) {
neighbors := false
// Flush the current pending frames
neighborPending, ok := d.pending[frame.FrameNumber]
for ok && neighborPending != nil {
next := neighborPending[0]
d.pending[frame.FrameNumber] =
d.pending[frame.FrameNumber][1:]
if len(d.pending[frame.FrameNumber]) == 0 {
delete(d.pending, frame.FrameNumber)
}
nextFrame, err := d.clockStore.GetCandidateDataClockFrame(
d.filter,
frame.FrameNumber,
next.parentSelector.FillBytes(make([]byte, 32)),
next.distance.FillBytes(make([]byte, 32)),
)
if err != nil && !errors.Is(err, store.ErrNotFound) {
panic(err)
}
if nextFrame != nil {
neighbors = true
go func() {
d.frames <- nextFrame
}()
}
neighborPending, ok = d.pending[frame.FrameNumber]
}
above := false
if !neighbors {
// Pull the next
nextPending, ok := d.pending[frame.FrameNumber+1]
if ok {
next := nextPending[0]
d.pending[frame.FrameNumber+1] =
d.pending[frame.FrameNumber+1][1:]
if len(d.pending[frame.FrameNumber+1]) == 0 {
delete(d.pending, frame.FrameNumber+1)
}
nextFrame, err := d.clockStore.GetCandidateDataClockFrame(
d.filter,
frame.FrameNumber+1,
next.parentSelector.FillBytes(make([]byte, 32)),
next.distance.FillBytes(make([]byte, 32)),
)
if err != nil && !errors.Is(err, store.ErrNotFound) {
panic(err)
}
if nextFrame != nil {
above = true
go func() {
d.frames <- nextFrame
}()
}
}
}
if !above {
// Pull below
min := frame.FrameNumber
for k := range d.pending {
if k < min {
min = k
}
}
if min == frame.FrameNumber {
return
}
nextPending, ok := d.pending[min]
if ok {
next := nextPending[0]
d.pending[min] =
d.pending[min][1:]
if len(d.pending[min]) == 0 {
delete(d.pending, min)
}
nextFrame, err := d.clockStore.GetCandidateDataClockFrame(
d.filter,
min,
next.parentSelector.FillBytes(make([]byte, 32)),
next.distance.FillBytes(make([]byte, 32)),
)
if err != nil && !errors.Is(err, store.ErrNotFound) {
panic(err)
}
if nextFrame != nil {
go func() {
d.frames <- nextFrame
}()
}
}
}
}
func (d *DataTimeReel) setHead(frame *protobufs.ClockFrame, distance *big.Int) {
txn, err := d.clockStore.NewTransaction()
if err != nil {
panic(err)
}
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,
) {
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 {
return nil, errors.Wrap(err, "get distance")
}
masterSelector, err := master.GetSelector()
if err != nil {
return nil, errors.Wrap(err, "get distance")
}
discriminatorNode :=
d.proverTrie.FindNearest(masterSelector.FillBytes(make([]byte, 32)))
discriminator := discriminatorNode.External.Key
addr, err := frame.GetAddress()
if err != nil {
return nil, errors.Wrap(err, "get distance")
}
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,
) {
parentSelector, selector, err := frame.GetParentAndSelector()
if err != nil {
panic(err)
}
leftIndex := d.head
rightIndex := frame
leftTotal := new(big.Int).Set(d.headDistance)
rightTotal := new(big.Int).Set(distance)
left := d.head.ParentSelector
right := frame.ParentSelector
rightReplaySelectors := [][]byte{}
// Walk backwards through the parents, until we find a matching parent
// selector:
for !bytes.Equal(left, right) {
rightReplaySelectors = append(
append(
[][]byte{},
right,
),
rightReplaySelectors...,
)
leftIndex, err = d.clockStore.GetParentDataClockFrame(
d.filter,
leftIndex.FrameNumber-1,
leftIndex.ParentSelector,
)
if err != nil {
panic(err)
}
rightIndex, err = d.clockStore.GetParentDataClockFrame(
d.filter,
rightIndex.FrameNumber-1,
rightIndex.ParentSelector,
)
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)
}
frameNumber := rightIndex.FrameNumber
// Choose new fork based on lightest distance sub-tree
if rightTotal.Cmp(leftTotal) < 0 {
for {
if len(rightReplaySelectors) == 0 {
break
}
next := rightReplaySelectors[0]
rightReplaySelectors =
rightReplaySelectors[1:]
rightIndex, err = d.clockStore.GetParentDataClockFrame(
d.filter,
frameNumber,
next,
)
if err != nil {
panic(err)
}
txn, err := d.clockStore.NewTransaction()
if err != nil {
panic(err)
}
if err := d.clockStore.PutDataClockFrame(
rightIndex,
d.proverTrie,
txn,
false,
); err != nil {
panic(err)
}
if err = txn.Commit(); err != nil {
panic(err)
}
frameNumber++
}
txn, err := d.clockStore.NewTransaction()
if err != nil {
panic(err)
}
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
go func() {
d.newFrameCh <- frame
}()
}
}
var _ TimeReel = (*DataTimeReel)(nil)