ceremonyclient/node/app/node.go

package app

import (
	"bytes"
	"errors"

	"go.uber.org/zap"
	"source.quilibrium.com/quilibrium/monorepo/node/consensus"
	"source.quilibrium.com/quilibrium/monorepo/node/consensus/master"
	"source.quilibrium.com/quilibrium/monorepo/node/execution"
	"source.quilibrium.com/quilibrium/monorepo/node/execution/intrinsics/ceremony"
	"source.quilibrium.com/quilibrium/monorepo/node/execution/intrinsics/ceremony/application"
	"source.quilibrium.com/quilibrium/monorepo/node/keys"
	"source.quilibrium.com/quilibrium/monorepo/node/p2p"
	"source.quilibrium.com/quilibrium/monorepo/node/store"
	"source.quilibrium.com/quilibrium/monorepo/node/tries"
)

type Node struct {
	logger      *zap.Logger
	clockStore  store.ClockStore
	keyManager  keys.KeyManager
	pubSub      p2p.PubSub
	execEngines map[string]execution.ExecutionEngine
	engine      consensus.ConsensusEngine
}

type DHTNode struct {
	pubSub p2p.PubSub
	quit   chan struct{}
}

func newDHTNode(
	pubSub p2p.PubSub,
) (*DHTNode, error) {
	return &DHTNode{
		pubSub: pubSub,
		quit:   make(chan struct{}),
	}, nil
}

func newNode(
	logger *zap.Logger,
	clockStore store.ClockStore,
	keyManager keys.KeyManager,
	pubSub p2p.PubSub,
	ceremonyExecutionEngine *ceremony.CeremonyExecutionEngine,
	engine consensus.ConsensusEngine,
) (*Node, error) {
	if engine == nil {
		return nil, errors.New("engine must not be nil")
	}

	execEngines := make(map[string]execution.ExecutionEngine)
	if ceremonyExecutionEngine != nil {
		execEngines[ceremonyExecutionEngine.GetName()] = ceremonyExecutionEngine
	}

	return &Node{
		logger,
		clockStore,
		keyManager,
		pubSub,
		execEngines,
		engine,
	}, nil
}

func (n *Node) RunRepair() {
	intrinsicFilter := append(
		p2p.GetBloomFilter(application.CEREMONY_ADDRESS, 256, 3),
		p2p.GetBloomFilterIndices(application.CEREMONY_ADDRESS, 65536, 24)...,
	)
	n.logger.Info("check store and repair if needed, this may take a few minutes")
	proverTrie := &tries.RollingFrecencyCritbitTrie{}
	head, err := n.clockStore.GetLatestDataClockFrame(intrinsicFilter, proverTrie)
	if err == nil && head != nil {
		for head != nil && head.FrameNumber != 0 {
			prev := head
			head, err = n.clockStore.GetStagedDataClockFrame(
				intrinsicFilter,
				head.FrameNumber-1,
				head.ParentSelector,
				true,
			)
			if err != nil {
				panic(err)
			}
			compare, _, err := n.clockStore.GetDataClockFrame(
				intrinsicFilter,
				prev.FrameNumber-1,
				true,
			)
			if err != nil {
				panic(err)
			}
			if !bytes.Equal(head.Output, compare.Output) {
				n.logger.Warn(
					"repairing frame",
					zap.Uint64("frame_number", head.FrameNumber),
				)
				head, err = n.clockStore.GetStagedDataClockFrame(
					intrinsicFilter,
					prev.FrameNumber-1,
					prev.ParentSelector,
					true,
				)
				if err != nil {
					panic(err)
				}

				txn, err := n.clockStore.NewTransaction()
				if err != nil {
					panic(err)
				}

				selector, err := head.GetSelector()
				if err != nil {
					panic(err)
				}

				err = n.clockStore.CommitDataClockFrame(
					intrinsicFilter,
					head.FrameNumber,
					selector.FillBytes(make([]byte, 32)),
					proverTrie,
					txn,
					true,
				)
				if err != nil {
					panic(err)
				}

				if err = txn.Commit(); err != nil {
					panic(err)
				}
			}
		}
	}
	n.logger.Info("check complete")
}

func (d *DHTNode) Start() {
	<-d.quit
}

func (d *DHTNode) Stop() {
	go func() {
		d.quit <- struct{}{}
	}()
}

func (n *Node) Start() {
	err := <-n.engine.Start()
	if err != nil {
		panic(err)
	}

	// TODO: add config mapping to engine name/frame registration
	for _, e := range n.execEngines {
		n.engine.RegisterExecutor(e, 0)
	}
}

func (n *Node) Stop() {
	err := <-n.engine.Stop(false)
	if err != nil {
		panic(err)
	}
}

func (n *Node) GetLogger() *zap.Logger {
	return n.logger
}

func (n *Node) GetClockStore() store.ClockStore {
	return n.clockStore
}

func (n *Node) GetKeyManager() keys.KeyManager {
	return n.keyManager
}

func (n *Node) GetPubSub() p2p.PubSub {
	return n.pubSub
}

func (n *Node) GetMasterClock() *master.MasterClockConsensusEngine {
	return n.engine.(*master.MasterClockConsensusEngine)
}

func (n *Node) GetExecutionEngines() []execution.ExecutionEngine {
	list := []execution.ExecutionEngine{}
	for _, e := range n.execEngines {
		list = append(list, e)
	}
	return list
}