From 59c550db0f0c1be41f3952bf0b4244f3e4303374 Mon Sep 17 00:00:00 2001 From: Cassandra Heart Date: Fri, 13 Oct 2023 23:05:44 -0500 Subject: [PATCH] =?UTF-8?q?1.1.3=20=E2=80=93=20Rewinding=20heads=20and=20o?= =?UTF-8?q?bvious=20fork=20comparison,=20additional=20bugfixes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../consensus/ceremony/broadcast_messaging.go | 7 +++ .../ceremony_data_clock_consensus_engine.go | 4 +- node/consensus/ceremony/consensus_frames.go | 45 +++++++++++++--- node/consensus/ceremony/peer_messaging.go | 54 ++++++++++++++++++- node/consensus/master/broadcast_messaging.go | 1 + node/consensus/master/consensus_frames.go | 1 + node/consensus/master/peer_messaging.go | 2 + .../application/ceremony_application.go | 2 + .../ceremony_application_in_progress.go | 2 + .../application/ceremony_application_open.go | 1 + node/main.go | 4 +- node/p2p/blossomsub.go | 1 + node/protobufs/clock.pb.go | 50 ++++++++++------- node/protobufs/clock.proto | 3 ++ node/store/clock.go | 2 + 15 files changed, 150 insertions(+), 29 deletions(-) diff --git a/node/consensus/ceremony/broadcast_messaging.go b/node/consensus/ceremony/broadcast_messaging.go index 9cae3f5..6ee1350 100644 --- a/node/consensus/ceremony/broadcast_messaging.go +++ b/node/consensus/ceremony/broadcast_messaging.go @@ -212,6 +212,13 @@ func (e *CeremonyDataClockConsensusEngine) handleCeremonyPeerListAnnounce( zap.String("announced_peer", peer.ID(pr.peerId).String()), zap.Int64("frame_distance", dst), ) + e.peerMap[string(p.PeerId)] = &peerInfo{ + peerId: p.PeerId, + multiaddr: p.Multiaddr, + maxFrame: p.MaxFrame, + direct: false, + lastSeen: time.Now().Unix(), + } } else if dst < -4 { e.logger.Debug( "peer sent announcement with lower frame index for peer", diff --git a/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go b/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go index 48b20af..90f1776 100644 --- a/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go +++ b/node/consensus/ceremony/ceremony_data_clock_consensus_engine.go @@ -87,7 +87,6 @@ type CeremonyDataClockConsensusEngine struct { peerAnnounceMap map[string]*protobufs.CeremonyPeerListAnnounce peerMap map[string]*peerInfo uncooperativePeersMap map[string]*peerInfo - fullResync bool } var _ consensus.DataConsensusEngine = (*CeremonyDataClockConsensusEngine)(nil) @@ -265,16 +264,19 @@ func (e *CeremonyDataClockConsensusEngine) Start( case consensus.EngineStateCollecting: if latestFrame, err = e.collect(latestFrame); err != nil { e.logger.Error("could not collect", zap.Error(err)) + e.state = consensus.EngineStateCollecting errChan <- err } case consensus.EngineStateProving: if latestFrame, err = e.prove(latestFrame); err != nil { e.logger.Error("could not prove", zap.Error(err)) + e.state = consensus.EngineStateCollecting errChan <- err } case consensus.EngineStatePublishing: if err = e.publishProof(latestFrame); err != nil { e.logger.Error("could not publish", zap.Error(err)) + e.state = consensus.EngineStateCollecting errChan <- err } } diff --git a/node/consensus/ceremony/consensus_frames.go b/node/consensus/ceremony/consensus_frames.go index a02a53e..a87d55d 100644 --- a/node/consensus/ceremony/consensus_frames.go +++ b/node/consensus/ceremony/consensus_frames.go @@ -688,6 +688,7 @@ func (e *CeremonyDataClockConsensusEngine) commitLongestPath( ) for _, s := range runningFrames[0][1:] { + s := s txn, err := e.clockStore.NewTransaction() if err != nil { return nil, errors.Wrap(err, "commit longest path") @@ -728,12 +729,14 @@ func (e *CeremonyDataClockConsensusEngine) commitLongestPath( ) 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{} @@ -885,7 +888,6 @@ func (e *CeremonyDataClockConsensusEngine) collect( zap.String("peer_id", peer.ID(peerId).String()), ) - willPerformFullResync := false cc, err := e.pubSub.GetDirectChannel(peerId) if err != nil { e.logger.Error( @@ -898,9 +900,11 @@ func (e *CeremonyDataClockConsensusEngine) collect( e.peerMapMx.Unlock() } else { from := latest.FrameNumber + originalFrom := from + originalLatest := latest if from == 0 { from = 1 - } else if maxFrame-from > 32 && !e.fullResync { + } else if maxFrame-from > 32 { // divergence is high, we need to confirm we're not in a fork from = 1 latest, _, err = e.clockStore.GetDataClockFrame(e.filter, 0) @@ -912,7 +916,6 @@ func (e *CeremonyDataClockConsensusEngine) collect( ) panic(err) } - willPerformFullResync = true } client := protobufs.NewCeremonyServiceClient(cc) @@ -922,6 +925,7 @@ func (e *CeremonyDataClockConsensusEngine) collect( Filter: e.filter, FromFrameNumber: from, ToFrameNumber: maxFrame, + ParentSelector: latest.ParentSelector, }, grpc.MaxCallRecvMsgSize(400*1024*1024), ) @@ -930,13 +934,19 @@ func (e *CeremonyDataClockConsensusEngine) collect( "error while retrieving sync", zap.Error(err), ) + latest = originalLatest e.peerMapMx.Lock() e.uncooperativePeersMap[string(peerId)] = e.peerMap[string(peerId)] delete(e.peerMap, string(peerId)) e.peerMapMx.Unlock() } else { var syncMsg *protobufs.CeremonyCompressedSync + askedFrom := from for syncMsg, err = s.Recv(); err == nil; syncMsg, err = s.Recv() { + if syncMsg.FromFrameNumber < askedFrom { + askedFrom = syncMsg.FromFrameNumber + } + e.logger.Info( "received compressed sync frame", zap.Uint64("from", syncMsg.FromFrameNumber), @@ -945,21 +955,42 @@ func (e *CeremonyDataClockConsensusEngine) collect( zap.Int("proofs", len(syncMsg.Proofs)), ) if err = e.decompressAndStoreCandidates(syncMsg); err != nil { + e.logger.Error( + "could not decompress and store candidate", + zap.Error(err), + ) + from = originalFrom + latest = originalLatest break } } if err != nil && err != io.EOF { e.logger.Error("error while receiving sync", zap.Error(err)) + from = originalFrom + latest = originalLatest + askedFrom = from + } + + if askedFrom < from { + e.logger.Info( + "peer provided deeper history due to fork, rewinding consensus " + + "head", + ) + askedLatest, _, err := e.clockStore.GetDataClockFrame( + e.filter, + askedFrom-1, + ) + if err != nil { + e.logger.Error("error while receiving sync", zap.Error(err)) + } else { + latest = askedLatest + } } } if err := cc.Close(); err != nil { e.logger.Error("error while closing connection", zap.Error(err)) } - - if willPerformFullResync { - e.fullResync = true - } } } diff --git a/node/consensus/ceremony/peer_messaging.go b/node/consensus/ceremony/peer_messaging.go index 9ec8aee..e2ceca2 100644 --- a/node/consensus/ceremony/peer_messaging.go +++ b/node/consensus/ceremony/peer_messaging.go @@ -76,7 +76,7 @@ func (e *CeremonyDataClockConsensusEngine) GetCompressedSyncFrames( from := request.FromFrameNumber - _, _, err := e.clockStore.GetDataClockFrame( + frame, _, err := e.clockStore.GetDataClockFrame( request.Filter, from, ) @@ -108,6 +108,43 @@ func (e *CeremonyDataClockConsensusEngine) GetCompressedSyncFrames( } } + parent := request.ParentSelector + if parent != nil { + if !bytes.Equal(frame.ParentSelector, parent) { + e.logger.Info( + "peer specified out of consensus head, seeking backwards for fork", + ) + } + + for !bytes.Equal(frame.ParentSelector, parent) { + ours, err := e.clockStore.GetParentDataClockFrame( + e.filter, + frame.FrameNumber-1, + frame.ParentSelector, + ) + if err != nil { + from = 1 + e.logger.Info("peer fully out of sync, rewinding sync head to start") + break + } + + theirs, err := e.clockStore.GetParentDataClockFrame( + e.filter, + frame.FrameNumber-1, + parent, + ) + if err != nil { + from = 1 + e.logger.Info("peer fully out of sync, rewinding sync head to start") + break + } + + from-- + frame = ours + parent = theirs.ParentSelector + } + } + max := e.frame to := request.ToFrameNumber @@ -151,7 +188,14 @@ func (e *CeremonyDataClockConsensusEngine) GetCompressedSyncFrames( func (e *CeremonyDataClockConsensusEngine) decompressAndStoreCandidates( syncMsg *protobufs.CeremonyCompressedSync, ) error { + if len(syncMsg.TruncatedClockFrames) != int( + syncMsg.ToFrameNumber-syncMsg.FromFrameNumber+1, + ) { + return errors.New("invalid continuity for compressed sync response") + } + for _, frame := range syncMsg.TruncatedClockFrames { + frame := frame commits := (len(frame.Input) - 516) / 74 e.logger.Info( "processing frame", @@ -167,6 +211,7 @@ func (e *CeremonyDataClockConsensusEngine) decompressAndStoreCandidates( commit := frame.Input[516+(j*74) : 516+((j+1)*74)] var aggregateProof *protobufs.InclusionProofsMap for _, a := range syncMsg.Proofs { + a := a if bytes.Equal(a.FrameCommit, commit) { e.logger.Info( "found matching proof", @@ -197,6 +242,8 @@ func (e *CeremonyDataClockConsensusEngine) decompressAndStoreCandidates( } for k, c := range aggregateProof.Commitments { + k := k + c := c e.logger.Info( "adding inclusion commitment", zap.Uint64("frame_number", frame.FrameNumber), @@ -217,7 +264,12 @@ func (e *CeremonyDataClockConsensusEngine) decompressAndStoreCandidates( output = &protobufs.IntrinsicExecutionOutput{} } for l, h := range c.SegmentHashes { + l := l + h := h + for _, s := range syncMsg.Segments { + s := s + if bytes.Equal(s.Hash, h) { if output != nil { if l == 0 { diff --git a/node/consensus/master/broadcast_messaging.go b/node/consensus/master/broadcast_messaging.go index 62cfa91..ba59f07 100644 --- a/node/consensus/master/broadcast_messaging.go +++ b/node/consensus/master/broadcast_messaging.go @@ -54,6 +54,7 @@ func (e *MasterClockConsensusEngine) handleMessage(message *pb.Message) error { } for _, m := range messages { + m := m if err := e.publishMessage(m.Address, m); err != nil { e.logger.Error( "could not publish message for engine", diff --git a/node/consensus/master/consensus_frames.go b/node/consensus/master/consensus_frames.go index 32837dc..53693a2 100644 --- a/node/consensus/master/consensus_frames.go +++ b/node/consensus/master/consensus_frames.go @@ -240,6 +240,7 @@ func ( } for _, frame := range committedSet { + frame := frame if err = e.clockStore.PutMasterClockFrame(frame, txn); err != nil { e.logger.Error("error while committing frame", zap.Error(err)) return nil, errors.Wrap(err, "confirm latest frame") diff --git a/node/consensus/master/peer_messaging.go b/node/consensus/master/peer_messaging.go index 1c746a5..5afedbb 100644 --- a/node/consensus/master/peer_messaging.go +++ b/node/consensus/master/peer_messaging.go @@ -50,6 +50,7 @@ func (e *MasterClockConsensusEngine) handleSync(message *pb.Message) error { } for _, m := range messages { + m := m if err := e.publishMessage(e.filter, m); err != nil { e.logger.Error( "could not publish message for engine", @@ -128,6 +129,7 @@ func (e *MasterClockConsensusEngine) handleClockFramesResponse( } for _, frame := range response.ClockFrames { + frame := frame e.logger.Debug( "processing clock frame", zap.Binary("sender", peerID), diff --git a/node/execution/ceremony/application/ceremony_application.go b/node/execution/ceremony/application/ceremony_application.go index a3384e7..fd3c815 100644 --- a/node/execution/ceremony/application/ceremony_application.go +++ b/node/execution/ceremony/application/ceremony_application.go @@ -855,6 +855,7 @@ func (a *CeremonyApplication) ApplyTransition( a.StateCount = 0 a.RoundCount = 0 for _, p := range a.ActiveParticipants { + p := p if _, ok := droppedProversMap[string(p.KeyValue)]; !ok { a.NextRoundPreferredParticipants = append( append( @@ -974,6 +975,7 @@ func (a *CeremonyApplication) ApplyTransition( a.StateCount = 0 a.RoundCount = 0 for _, p := range a.ActiveParticipants { + p := p if _, ok := droppedProversMap[string(p.KeyValue)]; !ok { a.NextRoundPreferredParticipants = append( append( diff --git a/node/execution/ceremony/application/ceremony_application_in_progress.go b/node/execution/ceremony/application/ceremony_application_in_progress.go index f7f9a4e..abcc6d7 100644 --- a/node/execution/ceremony/application/ceremony_application_in_progress.go +++ b/node/execution/ceremony/application/ceremony_application_in_progress.go @@ -54,6 +54,7 @@ func (a *CeremonyApplication) applySeenProverAttestation( replaced := false for i, att := range a.LatestSeenProverAttestations { + att := att if bytes.Equal( att.SeenProverKey.KeyValue, seenProverAttestation.SeenProverKey.KeyValue, @@ -126,6 +127,7 @@ func (a *CeremonyApplication) applyDroppedProverAttestation( replaced := false for i, att := range a.DroppedParticipantAttestations { + att := att if bytes.Equal( att.DroppedProverKey.KeyValue, droppedProverAttestation.DroppedProverKey.KeyValue, diff --git a/node/execution/ceremony/application/ceremony_application_open.go b/node/execution/ceremony/application/ceremony_application_open.go index 3d06ba2..ab13d50 100644 --- a/node/execution/ceremony/application/ceremony_application_open.go +++ b/node/execution/ceremony/application/ceremony_application_open.go @@ -51,6 +51,7 @@ func (a *CeremonyApplication) applyLobbyJoin( prepend := false nextRoundPreferredParticipants := []*protobufs.Ed448PublicKey{} for _, p := range a.NextRoundPreferredParticipants { + p := p if !bytes.Equal(p.KeyValue, signature.PublicKey.KeyValue) { nextRoundPreferredParticipants = append( nextRoundPreferredParticipants, diff --git a/node/main.go b/node/main.go index 4b9e964..b0752eb 100644 --- a/node/main.go +++ b/node/main.go @@ -1,3 +1,5 @@ +//go:build !js && !wasm + package main import ( @@ -228,5 +230,5 @@ func printLogo() { func printVersion() { fmt.Println(" ") - fmt.Println(" Quilibrium Node - v1.1.2 – Dawn") + fmt.Println(" Quilibrium Node - v1.1.3 – Dawn") } diff --git a/node/p2p/blossomsub.go b/node/p2p/blossomsub.go index d9bffd0..fde7ab9 100644 --- a/node/p2p/blossomsub.go +++ b/node/p2p/blossomsub.go @@ -449,6 +449,7 @@ func discoverPeers( } for peer := range peerChan { + peer := peer if peer.ID == h.ID() { continue } diff --git a/node/protobufs/clock.pb.go b/node/protobufs/clock.pb.go index 903372b..d66fcf4 100644 --- a/node/protobufs/clock.pb.go +++ b/node/protobufs/clock.pb.go @@ -206,6 +206,9 @@ type ClockFramesRequest struct { // The latest frame in the range requested, if provided. Capped to a maximum // size of 128 frames. ToFrameNumber uint64 `protobuf:"varint,3,opt,name=to_frame_number,json=toFrameNumber,proto3" json:"to_frame_number,omitempty"` + // The optional parent selector. If provided, will perform a check to confirm + // continuity, otherwise, will rewind the sync head to the beginning. + ParentSelector []byte `protobuf:"bytes,4,opt,name=parent_selector,json=parentSelector,proto3" json:"parent_selector,omitempty"` } func (x *ClockFramesRequest) Reset() { @@ -261,6 +264,13 @@ func (x *ClockFramesRequest) GetToFrameNumber() uint64 { return 0 } +func (x *ClockFramesRequest) GetParentSelector() []byte { + if x != nil { + return x.ParentSelector + } + return nil +} + // Represents a response for a range of clock frames. Used to stay synchronized // to the latest state. type ClockFramesResponse struct { @@ -374,7 +384,7 @@ var file_clock_proto_rawDesc = []byte{ 0x52, 0x17, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x4b, 0x65, 0x79, 0x53, 0x69, 0x67, 0x6e, 0x61, 0x74, 0x75, 0x72, 0x65, 0x45, 0x64, 0x34, 0x34, 0x38, 0x42, 0x16, 0x0a, 0x14, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x5f, 0x6b, 0x65, 0x79, 0x5f, 0x73, 0x69, 0x67, 0x6e, 0x61, 0x74, 0x75, 0x72, - 0x65, 0x22, 0x80, 0x01, 0x0a, 0x12, 0x43, 0x6c, 0x6f, 0x63, 0x6b, 0x46, 0x72, 0x61, 0x6d, 0x65, + 0x65, 0x22, 0xa9, 0x01, 0x0a, 0x12, 0x43, 0x6c, 0x6f, 0x63, 0x6b, 0x46, 0x72, 0x61, 0x6d, 0x65, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x16, 0x0a, 0x06, 0x66, 0x69, 0x6c, 0x74, 0x65, 0x72, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x66, 0x69, 0x6c, 0x74, 0x65, 0x72, 0x12, 0x2a, 0x0a, 0x11, 0x66, 0x72, 0x6f, 0x6d, 0x5f, 0x66, 0x72, 0x61, 0x6d, 0x65, 0x5f, 0x6e, @@ -382,24 +392,26 @@ var file_clock_proto_rawDesc = []byte{ 0x6d, 0x46, 0x72, 0x61, 0x6d, 0x65, 0x4e, 0x75, 0x6d, 0x62, 0x65, 0x72, 0x12, 0x26, 0x0a, 0x0f, 0x74, 0x6f, 0x5f, 0x66, 0x72, 0x61, 0x6d, 0x65, 0x5f, 0x6e, 0x75, 0x6d, 0x62, 0x65, 0x72, 0x18, 0x03, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0d, 0x74, 0x6f, 0x46, 0x72, 0x61, 0x6d, 0x65, 0x4e, 0x75, - 0x6d, 0x62, 0x65, 0x72, 0x22, 0xca, 0x01, 0x0a, 0x13, 0x43, 0x6c, 0x6f, 0x63, 0x6b, 0x46, 0x72, - 0x61, 0x6d, 0x65, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x16, 0x0a, 0x06, - 0x66, 0x69, 0x6c, 0x74, 0x65, 0x72, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x66, 0x69, - 0x6c, 0x74, 0x65, 0x72, 0x12, 0x2a, 0x0a, 0x11, 0x66, 0x72, 0x6f, 0x6d, 0x5f, 0x66, 0x72, 0x61, - 0x6d, 0x65, 0x5f, 0x6e, 0x75, 0x6d, 0x62, 0x65, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x04, 0x52, - 0x0f, 0x66, 0x72, 0x6f, 0x6d, 0x46, 0x72, 0x61, 0x6d, 0x65, 0x4e, 0x75, 0x6d, 0x62, 0x65, 0x72, - 0x12, 0x26, 0x0a, 0x0f, 0x74, 0x6f, 0x5f, 0x66, 0x72, 0x61, 0x6d, 0x65, 0x5f, 0x6e, 0x75, 0x6d, - 0x62, 0x65, 0x72, 0x18, 0x03, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0d, 0x74, 0x6f, 0x46, 0x72, 0x61, - 0x6d, 0x65, 0x4e, 0x75, 0x6d, 0x62, 0x65, 0x72, 0x12, 0x47, 0x0a, 0x0c, 0x63, 0x6c, 0x6f, 0x63, - 0x6b, 0x5f, 0x66, 0x72, 0x61, 0x6d, 0x65, 0x73, 0x18, 0x04, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x24, - 0x2e, 0x71, 0x75, 0x69, 0x6c, 0x69, 0x62, 0x72, 0x69, 0x75, 0x6d, 0x2e, 0x6e, 0x6f, 0x64, 0x65, - 0x2e, 0x63, 0x6c, 0x6f, 0x63, 0x6b, 0x2e, 0x70, 0x62, 0x2e, 0x43, 0x6c, 0x6f, 0x63, 0x6b, 0x46, - 0x72, 0x61, 0x6d, 0x65, 0x52, 0x0b, 0x63, 0x6c, 0x6f, 0x63, 0x6b, 0x46, 0x72, 0x61, 0x6d, 0x65, - 0x73, 0x42, 0x3a, 0x5a, 0x38, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x71, 0x75, 0x69, 0x6c, - 0x69, 0x62, 0x72, 0x69, 0x75, 0x6d, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x71, 0x75, 0x69, 0x6c, 0x69, - 0x62, 0x72, 0x69, 0x75, 0x6d, 0x2f, 0x6d, 0x6f, 0x6e, 0x6f, 0x72, 0x65, 0x70, 0x6f, 0x2f, 0x6e, - 0x6f, 0x64, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x73, 0x62, 0x06, 0x70, - 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x6d, 0x62, 0x65, 0x72, 0x12, 0x27, 0x0a, 0x0f, 0x70, 0x61, 0x72, 0x65, 0x6e, 0x74, 0x5f, 0x73, + 0x65, 0x6c, 0x65, 0x63, 0x74, 0x6f, 0x72, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0e, 0x70, + 0x61, 0x72, 0x65, 0x6e, 0x74, 0x53, 0x65, 0x6c, 0x65, 0x63, 0x74, 0x6f, 0x72, 0x22, 0xca, 0x01, + 0x0a, 0x13, 0x43, 0x6c, 0x6f, 0x63, 0x6b, 0x46, 0x72, 0x61, 0x6d, 0x65, 0x73, 0x52, 0x65, 0x73, + 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x66, 0x69, 0x6c, 0x74, 0x65, 0x72, 0x18, + 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x66, 0x69, 0x6c, 0x74, 0x65, 0x72, 0x12, 0x2a, 0x0a, + 0x11, 0x66, 0x72, 0x6f, 0x6d, 0x5f, 0x66, 0x72, 0x61, 0x6d, 0x65, 0x5f, 0x6e, 0x75, 0x6d, 0x62, + 0x65, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0f, 0x66, 0x72, 0x6f, 0x6d, 0x46, 0x72, + 0x61, 0x6d, 0x65, 0x4e, 0x75, 0x6d, 0x62, 0x65, 0x72, 0x12, 0x26, 0x0a, 0x0f, 0x74, 0x6f, 0x5f, + 0x66, 0x72, 0x61, 0x6d, 0x65, 0x5f, 0x6e, 0x75, 0x6d, 0x62, 0x65, 0x72, 0x18, 0x03, 0x20, 0x01, + 0x28, 0x04, 0x52, 0x0d, 0x74, 0x6f, 0x46, 0x72, 0x61, 0x6d, 0x65, 0x4e, 0x75, 0x6d, 0x62, 0x65, + 0x72, 0x12, 0x47, 0x0a, 0x0c, 0x63, 0x6c, 0x6f, 0x63, 0x6b, 0x5f, 0x66, 0x72, 0x61, 0x6d, 0x65, + 0x73, 0x18, 0x04, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x24, 0x2e, 0x71, 0x75, 0x69, 0x6c, 0x69, 0x62, + 0x72, 0x69, 0x75, 0x6d, 0x2e, 0x6e, 0x6f, 0x64, 0x65, 0x2e, 0x63, 0x6c, 0x6f, 0x63, 0x6b, 0x2e, + 0x70, 0x62, 0x2e, 0x43, 0x6c, 0x6f, 0x63, 0x6b, 0x46, 0x72, 0x61, 0x6d, 0x65, 0x52, 0x0b, 0x63, + 0x6c, 0x6f, 0x63, 0x6b, 0x46, 0x72, 0x61, 0x6d, 0x65, 0x73, 0x42, 0x3a, 0x5a, 0x38, 0x73, 0x6f, + 0x75, 0x72, 0x63, 0x65, 0x2e, 0x71, 0x75, 0x69, 0x6c, 0x69, 0x62, 0x72, 0x69, 0x75, 0x6d, 0x2e, + 0x63, 0x6f, 0x6d, 0x2f, 0x71, 0x75, 0x69, 0x6c, 0x69, 0x62, 0x72, 0x69, 0x75, 0x6d, 0x2f, 0x6d, + 0x6f, 0x6e, 0x6f, 0x72, 0x65, 0x70, 0x6f, 0x2f, 0x6e, 0x6f, 0x64, 0x65, 0x2f, 0x70, 0x72, 0x6f, + 0x74, 0x6f, 0x62, 0x75, 0x66, 0x73, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( diff --git a/node/protobufs/clock.proto b/node/protobufs/clock.proto index 6062b26..81b6f59 100644 --- a/node/protobufs/clock.proto +++ b/node/protobufs/clock.proto @@ -71,6 +71,9 @@ message ClockFramesRequest { // The latest frame in the range requested, if provided. Capped to a maximum // size of 128 frames. uint64 to_frame_number = 3; + // The optional parent selector. If provided, will perform a check to confirm + // continuity, otherwise, will rewind the sync head to the beginning. + bytes parent_selector = 4; } // Represents a response for a range of clock frames. Used to stay synchronized diff --git a/node/store/clock.go b/node/store/clock.go index cea894f..6ec668f 100644 --- a/node/store/clock.go +++ b/node/store/clock.go @@ -1341,6 +1341,8 @@ func (p *PebbleClockStore) GetCompressedDataClockFrames( } for k, v := range proofs { + k := k + v := v value, closer, err := p.db.Get(dataProofMetadataKey(filter, []byte(k))) if err != nil { if errors.Is(err, pebble.ErrNotFound) {