ceremonyclient/node/consensus/ceremony/consensus_frames.go

1193 lines
32 KiB
Go

package ceremony
import (
"bytes"
"context"
"encoding/binary"
"encoding/hex"
"fmt"
"io"
"math/big"
"strings"
"github.com/iden3/go-iden3-crypto/ff"
"github.com/iden3/go-iden3-crypto/poseidon"
"github.com/libp2p/go-libp2p/core/peer"
"github.com/pkg/errors"
"go.uber.org/zap"
"golang.org/x/crypto/sha3"
"google.golang.org/grpc"
"google.golang.org/protobuf/proto"
"source.quilibrium.com/quilibrium/monorepo/nekryptology/pkg/core/curves"
"source.quilibrium.com/quilibrium/monorepo/nekryptology/pkg/vdf"
"source.quilibrium.com/quilibrium/monorepo/node/consensus"
qcrypto "source.quilibrium.com/quilibrium/monorepo/node/crypto"
"source.quilibrium.com/quilibrium/monorepo/node/execution/ceremony/application"
"source.quilibrium.com/quilibrium/monorepo/node/p2p"
"source.quilibrium.com/quilibrium/monorepo/node/protobufs"
"source.quilibrium.com/quilibrium/monorepo/node/store"
"source.quilibrium.com/quilibrium/monorepo/node/tries"
)
func (e *CeremonyDataClockConsensusEngine) prove(
previousFrame *protobufs.ClockFrame,
) (*protobufs.ClockFrame, error) {
if e.state == consensus.EngineStateProving {
if !e.frameProverTrie.Contains(e.provingKeyAddress) {
e.stagedKeyCommitsMx.Lock()
e.stagedKeyCommits = make(
map[curves.PairingPoint]*protobufs.InclusionCommitment,
)
e.stagedKeyPolynomials = make(
map[curves.PairingPoint][]curves.PairingScalar,
)
e.stagedKeyCommitsMx.Unlock()
e.stagedLobbyStateTransitionsMx.Lock()
e.stagedLobbyStateTransitions = &protobufs.CeremonyLobbyStateTransition{}
e.stagedLobbyStateTransitionsMx.Unlock()
e.state = consensus.EngineStateCollecting
return previousFrame, nil
}
e.logger.Info("proving new frame")
commitments := [][]byte{}
aggregations := []*protobufs.InclusionAggregateProof{}
e.stagedKeyCommitsMx.Lock()
if len(e.stagedKeyCommits) > 0 && len(e.stagedKeyPolynomials) > 0 {
e.logger.Debug(
"adding staged key commits to frame",
zap.Uint64("frame_number", previousFrame.FrameNumber+1),
)
keyCommitments := []curves.PairingPoint{}
keyInclusions := []*protobufs.InclusionCommitment{}
keyPolynomials := [][]curves.PairingScalar{}
i := uint32(0)
for commit, inclusion := range e.stagedKeyCommits {
e.logger.Debug(
"adding staged key commit to aggregate proof",
zap.Uint64("frame_number", previousFrame.FrameNumber+1),
zap.Uint32("position", i),
)
keyCommitments = append(keyCommitments, commit)
inclusion.FrameNumber = previousFrame.FrameNumber + 1
inclusion.Position = i
keyInclusions = append(keyInclusions, inclusion)
keyPolynomials = append(keyPolynomials, e.stagedKeyPolynomials[commit])
}
proof, commitment, err := e.prover.ProveAggregate(
keyPolynomials,
keyCommitments,
)
if err != nil {
e.logger.Error("could not produce proof", zap.Error(err))
return nil, errors.Wrap(err, "prove")
}
if proof.IsIdentity() {
return nil, errors.Wrap(errors.New("invalid proof"), "prove")
}
commitments = append(commitments, commitment.ToAffineCompressed())
keyAggregation := &protobufs.InclusionAggregateProof{
Filter: e.filter,
FrameNumber: previousFrame.FrameNumber + 1,
InclusionCommitments: keyInclusions,
Proof: proof.ToAffineCompressed(),
}
aggregations = append(aggregations, keyAggregation)
e.stagedKeyCommits = make(
map[curves.PairingPoint]*protobufs.InclusionCommitment,
)
e.stagedKeyPolynomials = make(
map[curves.PairingPoint][]curves.PairingScalar,
)
}
e.stagedKeyCommitsMx.Unlock()
e.stagedLobbyStateTransitionsMx.Lock()
executionOutput := &protobufs.IntrinsicExecutionOutput{}
app, err := application.MaterializeApplicationFromFrame(previousFrame)
if err != nil {
e.stagedLobbyStateTransitions = &protobufs.CeremonyLobbyStateTransition{}
e.stagedLobbyStateTransitionsMx.Unlock()
return nil, errors.Wrap(err, "prove")
}
if e.stagedLobbyStateTransitions == nil {
e.stagedLobbyStateTransitions = &protobufs.CeremonyLobbyStateTransition{}
}
app, err = app.ApplyTransition(
previousFrame.FrameNumber,
e.stagedLobbyStateTransitions,
)
if err != nil {
e.stagedLobbyStateTransitions = &protobufs.CeremonyLobbyStateTransition{}
e.stagedLobbyStateTransitionsMx.Unlock()
return nil, errors.Wrap(err, "prove")
}
lobbyState, err := app.MaterializeLobbyStateFromApplication()
if err != nil {
e.stagedLobbyStateTransitions = &protobufs.CeremonyLobbyStateTransition{}
e.stagedLobbyStateTransitionsMx.Unlock()
return nil, errors.Wrap(err, "prove")
}
executionOutput.Address = application.CEREMONY_ADDRESS
executionOutput.Output, err = proto.Marshal(lobbyState)
if err != nil {
e.stagedLobbyStateTransitions = &protobufs.CeremonyLobbyStateTransition{}
e.stagedLobbyStateTransitionsMx.Unlock()
return nil, errors.Wrap(err, "prove")
}
executionOutput.Proof, err = proto.Marshal(e.stagedLobbyStateTransitions)
if err != nil {
e.stagedLobbyStateTransitions = &protobufs.CeremonyLobbyStateTransition{}
e.stagedLobbyStateTransitionsMx.Unlock()
return nil, errors.Wrap(err, "prove")
}
data, err := proto.Marshal(executionOutput)
if err != nil {
e.stagedLobbyStateTransitions = &protobufs.CeremonyLobbyStateTransition{}
e.stagedLobbyStateTransitionsMx.Unlock()
return nil, errors.Wrap(err, "prove")
}
e.logger.Debug("encoded execution output")
// Execution data in the ceremony is plaintext, we do not need to leverage
// full encoding for commit/proof reference.
digest := sha3.NewShake256()
_, err = digest.Write(data)
if err != nil {
e.logger.Error(
"error converting key bundle to polynomial",
zap.Error(err),
)
return nil, errors.Wrap(err, "prove")
}
expand := make([]byte, 1024)
_, err = digest.Read(expand)
if err != nil {
e.logger.Error(
"error converting key bundle to polynomial",
zap.Error(err),
)
return nil, errors.Wrap(err, "prove")
}
poly, err := e.prover.BytesToPolynomial(expand)
if err != nil {
e.stagedLobbyStateTransitions = &protobufs.CeremonyLobbyStateTransition{}
e.stagedLobbyStateTransitionsMx.Unlock()
return nil, errors.Wrap(err, "prove")
}
e.logger.Debug("proving execution output for inclusion")
polys, err := qcrypto.FFT(
poly,
*curves.BLS48581(
curves.BLS48581G1().NewGeneratorPoint(),
),
16,
false,
)
if err != nil {
e.stagedLobbyStateTransitions = &protobufs.CeremonyLobbyStateTransition{}
e.stagedLobbyStateTransitionsMx.Unlock()
return nil, errors.Wrap(err, "prove")
}
e.logger.Debug("converted execution output chunk to evaluation form")
e.logger.Debug("creating kzg commitment")
commitment, err := e.prover.Commit(polys)
if err != nil {
e.stagedLobbyStateTransitions = &protobufs.CeremonyLobbyStateTransition{}
e.stagedLobbyStateTransitionsMx.Unlock()
return nil, errors.Wrap(err, "prove")
}
e.logger.Debug("creating kzg proof")
proof, aggregate, err := e.prover.ProveAggregate(
[][]curves.PairingScalar{polys},
[]curves.PairingPoint{commitment},
)
if err != nil {
e.stagedLobbyStateTransitions = &protobufs.CeremonyLobbyStateTransition{}
e.stagedLobbyStateTransitionsMx.Unlock()
return nil, errors.Wrap(err, "prove")
}
if proof.IsIdentity() {
return nil, errors.Wrap(errors.New("invalid proof"), "prove")
}
commitments = append(commitments, aggregate.ToAffineCompressed())
e.logger.Debug("finalizing execution proof")
e.stagedLobbyStateTransitions = &protobufs.CeremonyLobbyStateTransition{}
e.stagedLobbyStateTransitionsMx.Unlock()
execInclusion := &protobufs.InclusionCommitment{
Filter: e.filter,
FrameNumber: previousFrame.FrameNumber + 1,
TypeUrl: protobufs.IntrinsicExecutionOutputType,
Data: data,
Commitment: commitment.ToAffineCompressed(),
}
execAggregation := &protobufs.InclusionAggregateProof{
Filter: e.filter,
FrameNumber: previousFrame.FrameNumber + 1,
InclusionCommitments: []*protobufs.InclusionCommitment{
execInclusion,
},
Proof: proof.ToAffineCompressed(),
}
aggregations = append(aggregations, execAggregation)
frame, err := protobufs.ProveDataClockFrame(
previousFrame,
commitments,
aggregations,
e.provingKey,
e.difficulty,
)
if err != nil {
return nil, errors.Wrap(err, "prove")
}
e.state = consensus.EngineStatePublishing
e.logger.Debug(
"returning new proven frame",
zap.Int("proof_count", len(aggregations)),
zap.Int("commitment_count", len(commitments)),
)
return frame, nil
}
return nil, nil
}
func (e *CeremonyDataClockConsensusEngine) setFrame(
frame *protobufs.ClockFrame,
) {
pubkey := []byte{}
discriminator := big.NewInt(0)
ed448PublicKey := frame.GetPublicKeySignatureEd448()
if frame.PublicKeySignature == nil && frame.FrameNumber != 0 {
e.logger.Error("could not set frame, signature invalid for non-zero frame")
return
} else if ed448PublicKey != nil {
pubkey = ed448PublicKey.PublicKey.KeyValue
}
if len(pubkey) != 0 {
var err error
discriminator, err = poseidon.HashBytes(pubkey)
if err != nil {
e.logger.Error(
"could not set frame",
zap.Error(err),
)
return
}
}
selector := new(big.Int).SetBytes(frame.ParentSelector)
l := new(big.Int).Mod(new(big.Int).Sub(selector, discriminator), ff.Modulus())
r := new(big.Int).Mod(new(big.Int).Sub(discriminator, selector), ff.Modulus())
distance := r
if l.Cmp(r) == -1 {
distance = l
}
previousSelectorBytes := [516]byte{}
copy(previousSelectorBytes[:], frame.Output[:516])
parent, err := poseidon.HashBytes(previousSelectorBytes[:])
if err != nil {
panic(errors.Wrap(err, "set frame"))
}
e.logger.Debug("set frame", zap.Uint64("frame_number", frame.FrameNumber))
e.currentDistance = distance
e.frame = frame.FrameNumber
e.parentSelector = parent.Bytes()
e.activeFrame = frame
go func() {
e.frameChan <- frame
}()
}
func (
e *CeremonyDataClockConsensusEngine,
) createGenesisFrame() *protobufs.ClockFrame {
e.logger.Info("creating genesis frame")
for _, l := range strings.Split(string(e.input), "\n") {
e.logger.Info(l)
}
b := sha3.Sum256(e.input)
v := vdf.New(e.difficulty, b)
v.Execute()
o := v.GetOutput()
inputMessage := o[:]
e.logger.Info("encoding ceremony and phase one signatories")
transcript := &protobufs.CeremonyTranscript{}
for p, s := range qcrypto.CeremonyBLS48581G1 {
transcript.G1Powers = append(
transcript.G1Powers,
&protobufs.BLS48581G1PublicKey{
KeyValue: s.ToAffineCompressed(),
},
)
e.logger.Info(fmt.Sprintf("encoded G1 power %d", p))
}
for p, s := range qcrypto.CeremonyBLS48581G2 {
transcript.G2Powers = append(
transcript.G2Powers,
&protobufs.BLS48581G2PublicKey{
KeyValue: s.ToAffineCompressed(),
},
)
e.logger.Info(fmt.Sprintf("encoded G2 power %d", p))
}
transcript.RunningG1_256Witnesses = append(
transcript.RunningG1_256Witnesses,
&protobufs.BLS48581G1PublicKey{
KeyValue: qcrypto.CeremonyRunningProducts[0].ToAffineCompressed(),
},
)
transcript.RunningG2_256Powers = append(
transcript.RunningG2_256Powers,
&protobufs.BLS48581G2PublicKey{
KeyValue: qcrypto.CeremonyPotPubKeys[len(qcrypto.CeremonyPotPubKeys)-1].
ToAffineCompressed(),
},
)
outputProof := &protobufs.CeremonyLobbyStateTransition{
TypeUrls: []string{},
TransitionInputs: [][]byte{},
}
proofBytes, err := proto.Marshal(outputProof)
if err != nil {
panic(err)
}
e.logger.Info("encoded transcript")
e.logger.Info("encoding ceremony signatories into application state")
rewardTrie := &tries.RewardCritbitTrie{}
for _, s := range qcrypto.CeremonySignatories {
pubkey := s.ToAffineCompressed()
addr, err := poseidon.HashBytes(pubkey)
if err != nil {
panic(err)
}
addrBytes := addr.Bytes()
addrBytes = append(make([]byte, 32-len(addrBytes)), addrBytes...)
rewardTrie.Add(addrBytes, 0, 50)
}
trieBytes, err := rewardTrie.Serialize()
if err != nil {
panic(err)
}
ceremonyLobbyState := &protobufs.CeremonyLobbyState{
LobbyState: 0,
CeremonyState: &protobufs.CeremonyLobbyState_CeremonyOpenState{
CeremonyOpenState: &protobufs.CeremonyOpenState{
JoinedParticipants: []*protobufs.CeremonyLobbyJoin{},
PreferredParticipants: []*protobufs.Ed448PublicKey{},
},
},
LatestTranscript: transcript,
RewardTrie: trieBytes,
}
outputBytes, err := proto.Marshal(ceremonyLobbyState)
if err != nil {
panic(err)
}
executionOutput := &protobufs.IntrinsicExecutionOutput{
Address: []byte(e.filter),
Output: outputBytes,
Proof: proofBytes,
}
data, err := proto.Marshal(executionOutput)
if err != nil {
panic(err)
}
e.logger.Info("encoded execution output")
digest := sha3.NewShake256()
_, err = digest.Write(data)
if err != nil {
panic(err)
}
expand := make([]byte, 1024)
_, err = digest.Read(expand)
if err != nil {
panic(err)
}
poly, err := e.prover.BytesToPolynomial(expand)
if err != nil {
panic(err)
}
e.logger.Info("proving execution output for inclusion")
evalPoly, err := qcrypto.FFT(
poly,
*curves.BLS48581(
curves.BLS48581G1().NewGeneratorPoint(),
),
16,
false,
)
if err != nil {
panic(err)
}
e.logger.Info(
"converted execution output chunk to evaluation form",
zap.Int("poly_size", len(evalPoly)),
)
e.logger.Info("creating kzg commitment")
commitment, err := e.prover.Commit(evalPoly)
if err != nil {
panic(err)
}
e.logger.Info("creating kzg proof")
proof, aggregate, err := e.prover.ProveAggregate(
[][]curves.PairingScalar{evalPoly},
[]curves.PairingPoint{commitment},
)
if err != nil {
panic(err)
}
e.logger.Info("finalizing execution proof")
inputMessage = append(
append([]byte{}, inputMessage...),
aggregate.ToAffineCompressed()...,
)
ceremonyExecutiveProof := &protobufs.InclusionAggregateProof{
Filter: e.filter,
FrameNumber: 0,
InclusionCommitments: []*protobufs.InclusionCommitment{
{
Filter: e.filter,
FrameNumber: 0,
Position: 0,
TypeUrl: protobufs.IntrinsicExecutionOutputType,
Data: data,
Commitment: commitment.ToAffineCompressed(),
},
},
Proof: proof.ToAffineCompressed(),
}
// Signatories are special, they don't have an inclusion proof because they
// have not broadcasted communication keys, but they still get contribution
// rights prior to PoMW, because they did produce meaningful work in the
// first phase:
e.logger.Info("encoding signatories to prover trie")
for _, s := range qcrypto.CeremonySignatories {
pubkey := s.ToAffineCompressed()
e.logger.Info("0x" + hex.EncodeToString(pubkey))
addr, err := poseidon.HashBytes(pubkey)
if err != nil {
panic(err)
}
addrBytes := addr.Bytes()
addrBytes = append(make([]byte, 32-len(addrBytes)), addrBytes...)
e.frameProverTrie.Add(addrBytes, 0)
}
e.logger.Info("proving genesis frame")
input := []byte{}
input = append(input, e.filter...)
input = binary.BigEndian.AppendUint64(input, e.frame)
input = binary.BigEndian.AppendUint64(input, uint64(0))
input = binary.BigEndian.AppendUint32(input, e.difficulty)
input = append(input, e.input...)
b = sha3.Sum256(input)
v = vdf.New(e.difficulty, b)
v.Execute()
o = v.GetOutput()
frame := &protobufs.ClockFrame{
Filter: e.filter,
FrameNumber: e.frame,
Timestamp: 0,
Difficulty: e.difficulty,
Input: inputMessage,
Output: o[:],
ParentSelector: e.parentSelector,
AggregateProofs: []*protobufs.InclusionAggregateProof{
ceremonyExecutiveProof,
},
PublicKeySignature: nil,
}
parent, distance, selector, err := frame.GetParentSelectorAndDistance()
if err != nil {
panic(err)
}
txn, err := e.clockStore.NewTransaction()
if err != nil {
panic(err)
}
if err := e.clockStore.PutCandidateDataClockFrame(
parent.Bytes(),
distance.Bytes(),
selector.Bytes(),
frame,
txn,
); err != nil {
panic(err)
}
if err := e.clockStore.PutDataClockFrame(
frame,
e.frameProverTrie,
txn,
); err != nil {
panic(err)
}
if err := txn.Commit(); err != nil {
panic(err)
}
e.setFrame(frame)
return frame
}
func (e *CeremonyDataClockConsensusEngine) commitLongestPath(
latest *protobufs.ClockFrame,
) (
*protobufs.ClockFrame,
error,
) {
current := latest
e.logger.Info(
"searching from committed frame",
zap.Uint64("frame_number", current.FrameNumber),
)
runningFrames := [][]*protobufs.ClockFrame{{current}}
commitReady := false
currentDepth := 0
for {
nextRunningFrames := [][]*protobufs.ClockFrame{}
for _, s := range runningFrames {
e.logger.Info(
"ranging over candidates for frame",
zap.Uint64("frame_number", s[currentDepth].FrameNumber),
)
selector, err := s[currentDepth].GetSelector()
if err != nil {
return nil, errors.Wrap(err, "commit longest path")
}
iter, err := e.clockStore.RangeCandidateDataClockFrames(
e.filter,
selector.Bytes(),
s[currentDepth].FrameNumber+1,
)
if err != nil {
return nil, errors.Wrap(err, "commit longest path")
}
for iter.First(); iter.Valid(); iter.Next() {
value, err := iter.Value()
if err != nil {
return nil, errors.Wrap(err, "commit longest path")
}
selectorBytes := selector.Bytes()
selectorBytes = append(
make([]byte, 32-len(selectorBytes)),
selectorBytes...,
)
nearest := e.frameProverTrie.FindNearest(
selectorBytes,
)
addr, err := value.GetAddress()
// If we got the outright nearest, then skip all this, we know this is
// the right frame for the selector.
if err != nil && bytes.Equal(nearest.Bits(), addr) {
nextRunningFrames = append(
nextRunningFrames,
append(
append([]*protobufs.ClockFrame{}, s...),
value,
),
)
break
}
// Iterated values will always be in order of shortest distance, this
// will always keep closest selected, longest path
if current.FrameNumber < value.FrameNumber {
e.logger.Info(
"setting longest path cursor to frame",
zap.Uint64("frame_number", value.FrameNumber),
)
current = value
}
e.logger.Info(
"adding candidate",
zap.Uint64("frame_number", value.FrameNumber),
)
nextRunningFrames = append(
nextRunningFrames,
append(
append([]*protobufs.ClockFrame{}, s...),
value,
),
)
}
iter.Close()
if len(nextRunningFrames) == 1 && len(nextRunningFrames[0]) > 32 {
break
}
}
if commitReady && len(nextRunningFrames) == 1 {
commitReady = false
e.logger.Info(
"consensus found, committing frames",
zap.Int("commit_depth", len(runningFrames[0])),
)
for _, s := range runningFrames[0][1:] {
s := s
txn, err := e.clockStore.NewTransaction()
if err != nil {
return nil, errors.Wrap(err, "commit longest path")
}
e.logger.Info(
"committing candidate",
zap.Uint64("frame_number", s.FrameNumber),
zap.Binary(
"prover",
s.GetPublicKeySignatureEd448().PublicKey.KeyValue,
),
)
addr, err := s.GetAddress()
if err != nil {
return nil, errors.Wrap(err, "commit longest path")
}
e.frameProverTrie.Add(addr, s.FrameNumber)
if err := e.clockStore.PutDataClockFrame(
s,
e.frameProverTrie,
txn,
); err != nil {
e.logger.Error(
"could not commit candidate",
zap.Error(err),
zap.Uint64("frame_number", s.FrameNumber),
zap.Binary("output", s.Output),
)
return nil, errors.Wrap(err, "commit longest path")
}
e.logger.Debug(
"committing aggregate proofs",
zap.Int("proof_count", len(s.AggregateProofs)),
)
for _, p := range s.AggregateProofs {
p := p
e.logger.Debug(
"committing inclusions",
zap.Int("inclusions_count", len(p.InclusionCommitments)),
)
for _, c := range p.InclusionCommitments {
c := c
switch c.TypeUrl {
case protobufs.ProvingKeyAnnouncementType:
provingKey := &protobufs.ProvingKeyAnnouncement{}
if err := proto.Unmarshal(c.Data, provingKey); err != nil {
e.logger.Error(
"could not commit candidate",
zap.Error(err),
zap.Uint64("frame_number", s.FrameNumber),
zap.Binary("commitment", c.Commitment),
)
return nil, errors.Wrap(err, "commit longest path")
}
e.logger.Debug(
"committing proving key",
zap.Uint64("frame_number", s.FrameNumber),
zap.Binary("commitment", c.Commitment),
)
if err := e.keyStore.IncludeProvingKey(c, txn); err != nil {
e.logger.Error(
"could not commit candidate",
zap.Error(err),
zap.Uint64("frame_number", s.FrameNumber),
zap.Binary("output", s.Output),
)
return nil, errors.Wrap(err, "commit longest path")
}
case protobufs.KeyBundleAnnouncementType:
bundle := &protobufs.KeyBundleAnnouncement{}
if err := proto.Unmarshal(c.Data, bundle); err != nil {
e.logger.Error(
"could not commit candidate",
zap.Error(err),
zap.Uint64("frame_number", s.FrameNumber),
zap.Binary("commitment", c.Commitment),
)
return nil, errors.Wrap(err, "commit longest path")
}
e.logger.Debug(
"committing key bundle",
zap.Uint64("frame_number", s.FrameNumber),
zap.Binary("commitment", c.Commitment),
)
if err := e.keyStore.PutKeyBundle(
bundle.ProvingKeyBytes,
c,
txn,
); err != nil {
e.logger.Error(
"could not commit candidate",
zap.Error(err),
zap.Uint64("frame_number", s.FrameNumber),
zap.Binary("output", s.Output),
)
return nil, errors.Wrap(err, "commit longest path")
}
}
}
}
if err := txn.Commit(); err != nil {
e.logger.Error(
"could not commit candidates",
zap.Error(err),
)
return nil, errors.Wrap(err, "commit longest path")
}
}
runningFrames = [][]*protobufs.ClockFrame{
{nextRunningFrames[0][currentDepth]},
}
currentDepth = 0
} else {
e.logger.Info(
"not ready to commit",
zap.Int("forks", len(nextRunningFrames)),
zap.Int("current_depth", currentDepth),
)
commitReady = len(nextRunningFrames) == 1
runningFrames = nextRunningFrames
currentDepth++
}
if len(nextRunningFrames) == 0 {
e.logger.Info("deepest consensus reached")
break
}
}
return current, nil
}
func (e *CeremonyDataClockConsensusEngine) GetMostAheadPeer() (
[]byte,
uint64,
error,
) {
e.peerMapMx.Lock()
max := e.frame
var peer []byte = nil
for _, v := range e.peerMap {
if v.maxFrame > max {
peer = v.peerId
max = v.maxFrame
}
}
size := len(e.peerMap)
e.peerMapMx.Unlock()
if peer == nil {
if size > 1 {
return nil, 0, nil
} else {
return nil, 0, p2p.ErrNoPeersAvailable
}
}
return peer, max, nil
}
func (e *CeremonyDataClockConsensusEngine) reverseOptimisticSync(
currentLatest *protobufs.ClockFrame,
maxFrame uint64,
peerId []byte,
) (*protobufs.ClockFrame, error) {
latest := currentLatest
cc, err := e.pubSub.GetDirectChannel(peerId)
if err != nil {
e.logger.Error(
"could not establish direct channel",
zap.Error(err),
)
e.peerMapMx.Lock()
e.uncooperativePeersMap[string(peerId)] = e.peerMap[string(peerId)]
delete(e.peerMap, string(peerId))
e.peerMapMx.Unlock()
e.syncingTarget = nil
return latest, errors.Wrap(err, "reverse optimistic sync")
}
client := protobufs.NewCeremonyServiceClient(cc)
from := latest.FrameNumber
if from == 0 {
from = 1
}
if maxFrame-from > 32 {
// divergence is high, ask them for the latest frame and if they
// respond with a valid answer, optimistically continue from this
// frame, if we hit a fault we'll mark them as uncooperative and move
// on
from = 1
s, err := client.GetCompressedSyncFrames(
context.Background(),
&protobufs.ClockFramesRequest{
Filter: e.filter,
FromFrameNumber: maxFrame - 32,
},
grpc.MaxCallRecvMsgSize(600*1024*1024),
)
if err != nil {
e.logger.Error(
"received error from peer",
zap.Error(err),
)
e.peerMapMx.Lock()
e.uncooperativePeersMap[string(peerId)] = e.peerMap[string(peerId)]
delete(e.peerMap, string(peerId))
e.peerMapMx.Unlock()
e.syncingTarget = nil
return latest, errors.Wrap(err, "reverse optimistic sync")
}
var syncMsg *protobufs.CeremonyCompressedSync
for syncMsg, err = s.Recv(); err == nil; syncMsg, err = s.Recv() {
e.logger.Info(
"received compressed sync frame",
zap.Uint64("from", syncMsg.FromFrameNumber),
zap.Uint64("to", syncMsg.ToFrameNumber),
zap.Int("frames", len(syncMsg.TruncatedClockFrames)),
zap.Int("proofs", len(syncMsg.Proofs)),
)
var next *protobufs.ClockFrame
if next, err = e.decompressAndStoreCandidates(
syncMsg,
e.logger.Info,
); err != nil && !errors.Is(err, ErrNoNewFrames) {
e.logger.Error(
"could not decompress and store candidate",
zap.Error(err),
)
e.peerMapMx.Lock()
e.uncooperativePeersMap[string(peerId)] = e.peerMap[string(peerId)]
delete(e.peerMap, string(peerId))
e.peerMapMx.Unlock()
if err := cc.Close(); err != nil {
e.logger.Error("error while closing connection", zap.Error(err))
}
e.syncingTarget = nil
return currentLatest, errors.Wrap(err, "reverse optimistic sync")
}
if next != nil {
latest = next
}
}
if err != nil && err != io.EOF && !errors.Is(err, ErrNoNewFrames) {
if err := cc.Close(); err != nil {
e.logger.Error("error while closing connection", zap.Error(err))
}
e.logger.Error("error while receiving sync", zap.Error(err))
e.syncingTarget = nil
return latest, errors.Wrap(err, "reverse optimistic sync")
}
}
go func() {
defer func() { e.syncingTarget = nil }()
e.logger.Info("continuing sync in background")
s, err := client.GetCompressedSyncFrames(
context.Background(),
&protobufs.ClockFramesRequest{
Filter: e.filter,
FromFrameNumber: from,
ToFrameNumber: maxFrame,
},
grpc.MaxCallRecvMsgSize(600*1024*1024),
)
if err != nil {
e.logger.Error(
"error while retrieving sync",
zap.Error(err),
)
e.peerMapMx.Lock()
e.uncooperativePeersMap[string(peerId)] = e.peerMap[string(peerId)]
delete(e.peerMap, string(peerId))
e.peerMapMx.Unlock()
if err := cc.Close(); err != nil {
e.logger.Error("error while closing connection", zap.Error(err))
}
return
} else {
var syncMsg *protobufs.CeremonyCompressedSync
for syncMsg, err = s.Recv(); err == nil; syncMsg, err = s.Recv() {
e.logger.Debug(
"received compressed sync frame",
zap.Uint64("from", syncMsg.FromFrameNumber),
zap.Uint64("to", syncMsg.ToFrameNumber),
zap.Int("frames", len(syncMsg.TruncatedClockFrames)),
zap.Int("proofs", len(syncMsg.Proofs)),
)
if _, err = e.decompressAndStoreCandidates(
syncMsg,
e.logger.Debug,
); err != nil && !errors.Is(err, ErrNoNewFrames) {
e.logger.Error(
"could not decompress and store candidate",
zap.Error(err),
)
if err := cc.Close(); err != nil {
e.logger.Error("error while closing connection", zap.Error(err))
}
return
}
}
if err != nil && err != io.EOF && !errors.Is(err, ErrNoNewFrames) {
e.logger.Error("error while receiving sync", zap.Error(err))
if err := cc.Close(); err != nil {
e.logger.Error("error while closing connection", zap.Error(err))
}
return
}
}
if err := cc.Close(); err != nil {
e.logger.Error("error while closing connection", zap.Error(err))
}
}()
return latest, nil
}
func (e *CeremonyDataClockConsensusEngine) sync(
currentLatest *protobufs.ClockFrame,
maxFrame uint64,
peerId []byte,
) (*protobufs.ClockFrame, error) {
latest := currentLatest
e.logger.Info("polling peer for new frames", zap.Binary("peer_id", peerId))
cc, err := e.pubSub.GetDirectChannel(peerId)
if err != nil {
e.logger.Error(
"could not establish direct channel",
zap.Error(err),
)
e.peerMapMx.Lock()
e.uncooperativePeersMap[string(peerId)] = e.peerMap[string(peerId)]
delete(e.peerMap, string(peerId))
e.peerMapMx.Unlock()
return latest, errors.Wrap(err, "reverse optimistic sync")
}
client := protobufs.NewCeremonyServiceClient(cc)
from := latest.FrameNumber
if from == 0 {
from = 1
}
if maxFrame > from {
from = 1
s, err := client.GetCompressedSyncFrames(
context.Background(),
&protobufs.ClockFramesRequest{
Filter: e.filter,
FromFrameNumber: maxFrame - 16,
},
grpc.MaxCallRecvMsgSize(600*1024*1024),
)
if err != nil {
e.logger.Error(
"received error from peer",
zap.Error(err),
)
e.peerMapMx.Lock()
e.uncooperativePeersMap[string(peerId)] = e.peerMap[string(peerId)]
delete(e.peerMap, string(peerId))
e.peerMapMx.Unlock()
return latest, errors.Wrap(err, "reverse optimistic sync")
}
var syncMsg *protobufs.CeremonyCompressedSync
for syncMsg, err = s.Recv(); err == nil; syncMsg, err = s.Recv() {
e.logger.Info(
"received compressed sync frame",
zap.Uint64("from", syncMsg.FromFrameNumber),
zap.Uint64("to", syncMsg.ToFrameNumber),
zap.Int("frames", len(syncMsg.TruncatedClockFrames)),
zap.Int("proofs", len(syncMsg.Proofs)),
)
var next *protobufs.ClockFrame
if next, err = e.decompressAndStoreCandidates(
syncMsg,
e.logger.Info,
); err != nil && !errors.Is(err, ErrNoNewFrames) {
e.logger.Error(
"could not decompress and store candidate",
zap.Error(err),
)
e.peerMapMx.Lock()
e.uncooperativePeersMap[string(peerId)] = e.peerMap[string(peerId)]
delete(e.peerMap, string(peerId))
e.peerMapMx.Unlock()
if err := cc.Close(); err != nil {
e.logger.Error("error while closing connection", zap.Error(err))
}
return currentLatest, errors.Wrap(err, "reverse optimistic sync")
}
if next != nil {
latest = next
}
}
if err != nil && err != io.EOF && !errors.Is(err, ErrNoNewFrames) {
if err := cc.Close(); err != nil {
e.logger.Error("error while closing connection", zap.Error(err))
}
e.logger.Error("error while receiving sync", zap.Error(err))
return latest, errors.Wrap(err, "reverse optimistic sync")
}
e.logger.Info("received new leading frame", zap.Uint64("frame_number", latest.FrameNumber))
if err := cc.Close(); err != nil {
e.logger.Error("error while closing connection", zap.Error(err))
}
}
return latest, nil
}
func (e *CeremonyDataClockConsensusEngine) collect(
currentFramePublished *protobufs.ClockFrame,
) (*protobufs.ClockFrame, error) {
if e.state == consensus.EngineStateCollecting {
e.logger.Info("collecting vdf proofs")
latest := currentFramePublished
maxFrame := uint64(0)
var peerId []byte
peerId, maxFrame, err := e.GetMostAheadPeer()
if err != nil {
e.logger.Warn("no peers available, skipping sync")
} else if peerId == nil {
e.logger.Info("currently up to date, skipping sync")
} else if e.syncingTarget == nil {
e.syncingStatus = SyncStatusAwaitingResponse
e.logger.Info(
"setting syncing target",
zap.String("peer_id", peer.ID(peerId).String()),
)
e.syncingTarget = peerId
latest, err = e.reverseOptimisticSync(latest, maxFrame, peerId)
} else if maxFrame > latest.FrameNumber {
latest, err = e.sync(latest, maxFrame, peerId)
}
go func() {
_, err = e.keyStore.GetProvingKey(e.provingKeyBytes)
if errors.Is(err, store.ErrNotFound) &&
latest.FrameNumber-e.lastKeyBundleAnnouncementFrame > 6 {
if err = e.announceKeyBundle(); err != nil {
panic(err)
}
e.lastKeyBundleAnnouncementFrame = latest.FrameNumber
}
}()
e.logger.Info(
"returning leader frame",
zap.Uint64("frame_number", latest.FrameNumber),
)
e.setFrame(latest)
e.state = consensus.EngineStateProving
return latest, nil
}
return nil, nil
}