digest request

Author: samliok

epoch.go

@@ -167,6 +167,8 @@ func (e *Epoch) HandleMessage(msg *Message, from NodeID) error {
 		return e.handleReplicationResponse(msg.ReplicationResponse, from)
 	case msg.ReplicationRequest != nil && e.ReplicationEnabled:
 		return e.handleReplicationRequest(msg.ReplicationRequest, from)
+	case msg.BlockDigestRequest != nil: 
+		return e.handleBlockDigestRequest(msg.BlockDigestRequest, from)
 	default:
 		e.Logger.Debug("Invalid message type", zap.Stringer("from", from))
 		return nil
@@ -623,7 +625,7 @@ func (e *Epoch) handleFinalizeVoteMessage(message *FinalizeVote, from NodeID) er
 	vote := message.Finalization
 
 	e.Logger.Verbo("Received finalize vote",
-		zap.Stringer("from", from), zap.Uint64("round", vote.Round))
+		zap.Stringer("from", from), zap.Uint64("round", vote.Round), zap.Uint64("nextSeqToCommit", e.nextSeqToCommit()))
 
 	// Only process a point to point finalizations.
 	// This is needed to prevent a malicious node from sending us a finalization of a different node for a future round.
@@ -1113,7 +1115,7 @@ func (e *Epoch) rebroadcastPastFinalizeVotes() error {
 			}
 			finalizeVoteMessage = msg
 		}
-		e.Logger.Debug("Rebroadcasting finalization", zap.Uint64("round", r), zap.Uint64("seq", finalizeVoteMessage.FinalizeVote.Finalization.Seq))
+		e.Logger.Debug("Rebroadcasting finalize vote", zap.Uint64("round", r), zap.Uint64("seq", finalizeVoteMessage.FinalizeVote.Finalization.Seq))
 		e.Comm.Broadcast(finalizeVoteMessage)
 	}
 
@@ -2886,6 +2888,35 @@ func (e *Epoch) haveNotFinalizedNotarizedRound() (uint64, bool) {
 	return minRoundNum, found
 }
 
+func (e *Epoch) handleBlockDigestRequest(req *BlockDigestRequest, from NodeID) error {
+	e.Logger.Debug("Received block digest request", zap.Stringer("from", from), zap.Uint64("seq", req.Seq))
+	block, notarizationOrFinalization, ok := e.locateBlock(req.Seq, req.Digest[:])
+
+	if !ok {
+		e.Logger.Debug("Block not found for digest request", zap.Uint64("seq", req.Seq), zap.Stringer("digest", req.Digest))
+		return nil
+	}
+
+	if notarizationOrFinalization == nil {
+		e.Logger.Debug("No notarization or finalization found for block digest request", zap.Uint64("seq", req.Seq), zap.Stringer("digest", req.Digest))
+		return nil
+	}
+
+	qr := VerifiedQuorumRound{
+		VerifiedBlock: block,
+		Notarization:  notarizationOrFinalization.Notarization,
+		Finalization:  notarizationOrFinalization.Finalization,
+	}
+	
+	response := &VerifiedReplicationResponse{
+		Data: []VerifiedQuorumRound{qr},
+	}
+
+	msg := &Message{VerifiedReplicationResponse: response}
+	e.Comm.Send(msg, from)
+	return nil
+}
+
 func (e *Epoch) handleReplicationResponse(resp *ReplicationResponse, from NodeID) error {
 	if !e.ReplicationEnabled {
 		return nil

msg.go

@@ -22,6 +22,7 @@ type Message struct {
 	ReplicationResponse         *ReplicationResponse
 	VerifiedReplicationResponse *VerifiedReplicationResponse
 	ReplicationRequest          *ReplicationRequest
+	BlockDigestRequest		    *BlockDigestRequest
 }
 
 type EmptyVoteMetadata struct {
@@ -359,3 +360,8 @@ type VerifiedFinalizedBlock struct {
 type verifiableMessage interface {
 	Verify() error
 }
+
+type BlockDigestRequest struct {
+	Seq uint64
+	Digest Digest
+}

out.txt

@@ -12,6 +12,18 @@ import (
 	"go.uber.org/zap"
 )
 
+type missingBlock struct {
+	digest Digest
+	seq  uint64
+}
+
+func missingBlockFromBlock(block Block) missingBlock {
+	return missingBlock{
+		digest: block.BlockHeader().Digest,
+		seq:    block.BlockHeader().Seq,
+	}
+}
+
 type finalizedQuorumRound struct {
 	block        Block
 	finalization *Finalization
@@ -33,7 +45,7 @@ type ReplicationState struct {
 	// it means a prior notarization for that block exists but is missing.
 	// Since we may not know which round that dependency belongs to,
 	// digestTimeouts ensures we re-request the missing digest until it arrives.
-	digestTimeouts *TimeoutHandler[Digest]
+	digestTimeouts *TimeoutHandler[missingBlock]
 
 	// emptyRoundTimeouts handles timeouts for fetching missing empty round notarizations.
 	// When replication encounters a notarized block that depends on an empty round we haven't received,
@@ -42,6 +54,9 @@ type ReplicationState struct {
 
 	roundRequestor        *requestor
 	finalizationRequestor *requestor
+
+	sender sender
+	epochLock  *sync.Mutex
 }
 
 func NewReplicationState(logger Logger, comm Communication, myNodeID NodeID, maxRoundWindow uint64, enabled bool, start time.Time, lock *sync.Mutex) *ReplicationState {
@@ -64,6 +79,10 @@ func NewReplicationState(logger Logger, comm Communication, myNodeID NodeID, max
 		// round replication
 		rounds:         make(map[uint64]*QuorumRound),
 		roundRequestor: newRequestor(logger, start, lock, maxRoundWindow, comm, false),
+
+		// sender for missing dependencies
+		sender: comm,
+		epochLock:  lock,
 	}
 
 	r.digestTimeouts = NewTimeoutHandler(logger, "digest", start, DefaultReplicationRequestTimeout, r.requestDigests)
@@ -110,15 +129,15 @@ func (r *ReplicationState) storeSequence(block Block, finalization *Finalization
 	}
 
 	r.finalizationRequestor.removeTask(finalization.Finalization.Seq)
-	r.digestTimeouts.RemoveTask(block.BlockHeader().Digest)
+	r.digestTimeouts.RemoveTask(missingBlockFromBlock(block))
 }
 
 // storeRound adds or updates a quorum round in the replication state.
 // If the round already exists, it merges any missing notarizations or empty notarizations
 // from the provided quorum round. Otherwise, it stores the new round as is.
 func (r *ReplicationState) storeRound(qr *QuorumRound) {
 	if qr.Block != nil {
-		r.digestTimeouts.RemoveTask(qr.Block.BlockHeader().Digest)
+		r.digestTimeouts.RemoveTask(missingBlockFromBlock(qr.Block))
 	}
 
 	existing, exists := r.rounds[qr.GetRound()]
@@ -222,7 +241,7 @@ func (r *ReplicationState) ResendFinalizationRequest(seq uint64, signers []NodeI
 // TODO: in a future PR, these requests will be sent as specific digest requests.
 func (r *ReplicationState) CreateDependencyTasks(parent *Digest, parentSeq uint64, emptyRounds []uint64) {
 	if parent != nil {
-		r.digestTimeouts.AddTask(*parent)
+		r.digestTimeouts.AddTask(missingBlock{digest: *parent, seq: parentSeq})
 	}
 
 	if len(emptyRounds) > 0 {
@@ -296,9 +315,41 @@ func (r *ReplicationState) GetBlockWithSeq(seq uint64) Block {
 	return nil
 }
 
-func (r *ReplicationState) requestDigests(digests []Digest) {
-	// TODO: In a future PR, I will add a message that requests a specific digest.
-	r.logger.Debug("Not implemented yet", zap.Stringers("Digests", digests))
+func (r *ReplicationState) requestDigests(missingBlocks []missingBlock) {
+	// grab the lock since this is called in the timeout handler goroutine
+	r.epochLock.Lock()
+	defer r.epochLock.Unlock()
+
+	signedQuorum := r.roundRequestor.getHighestObserved()
+	if signedQuorum == nil {
+		signedQuorum = r.finalizationRequestor.getHighestObserved()
+	}
+	if signedQuorum == nil {
+		r.logger.Warn("Replication State cannot request missing block digests, no known nodes to request from")
+		return
+	}
+
+	randInt, err := rand.Int(rand.Reader, big.NewInt(int64(len(signedQuorum.signers))))
+	if err != nil {
+		r.logger.Info("Replication State failed to generate random starting index", zap.Error(err))
+		return
+	}
+	startingIndex := int(randInt.Int64())
+
+	for i, mb := range missingBlocks {
+		// grab the node to send it to
+		index := (i + startingIndex) % len(signedQuorum.signers)
+		node := signedQuorum.signers[index]
+
+		r.logger.Debug("Replication State requesting missing block digest", zap.Uint64("seq", mb.seq), zap.Stringer("digest", &mb.digest))
+		blockRequest := &BlockDigestRequest{
+			Seq:    mb.seq,
+			Digest: mb.digest,
+		}
+		r.sender.Send(&Message{
+			BlockDigestRequest: blockRequest,
+		}, node)
+	}
 }
 
 func (r *ReplicationState) requestEmptyRounds(emptyRounds []uint64) {

replication_state.go

@@ -1328,7 +1328,6 @@ func allowFinalizeVotes(msg *simplex.Message, from, to simplex.NodeID) bool {
 // TestReplicationChain tests that a node can both empty notarizations and notarizations for the same round.
 func TestReplicationChain(t *testing.T) {
 	// Digest message requests are needed for this test
-	t.Skip()
 	nodes := []simplex.NodeID{{1}, {2}, {3}, {4}}
 	net := NewInMemNetwork(t, nodes)
 
@@ -1348,11 +1347,11 @@ func TestReplicationChain(t *testing.T) {
 	fullNode2.Silence()
 	// node 3 will not receive finalize votes & finalizations
 	blockFinalize3 := NewSimplexNode(t, nodes[2], net, newNodeConfig(nodes[2]))
-	blockFinalize3.Silence()
+	// blockFinalize3.Silence()
 	// lagging node is disconnected initially. It initially receives only empty notarizations
 	// but then later receives notarizations and must send finalize votes for them
 	laggingNode := NewSimplexNode(t, nodes[3], net, newNodeConfig(nodes[3]))
-
+	laggingNode.Silence()
 	net.StartInstances()
 	net.Disconnect(laggingNode.E.ID)
 
@@ -1408,7 +1407,32 @@ func TestReplicationChain(t *testing.T) {
 	net.SetAllNodesMessageFilter(allowFinalizeVotes)
 
 	for _, n := range net.Instances {
-		n.TickUntilRoundAdvanced(numNotarizations+1, simplex.DefaultReplicationRequestTimeout)
-		require.Equal(t, numNotarizations+1-missedNotarizations, n.Storage.NumBlocks())
+		// the message filter blocks finalizations for blockFinalize3
+		if n.E.ID.Equals(blockFinalize3.E.ID) {
+			continue
+		}
+
+		for {
+			numBlocks := n.Storage.NumBlocks()
+			if numBlocks >= numNotarizations-missedNotarizations {
+				break
+			}
+
+			net.AdvanceTime(simplex.DefaultReplicationRequestTimeout)
+		}
+	}
+
+	// just for fun check the blockFinalize3 also replicated properly
+	net.SetAllNodesMessageFilter(AllowAllMessages)
+	for {
+		numBlocks := blockFinalize3.Storage.NumBlocks()
+		fmt.Println("blockFinalize3 num blocks:", numBlocks)
+		if numBlocks == numNotarizations-missedNotarizations {
+			break
+		}
+
+		net.AdvanceTime(simplex.DefaultReplicationRequestTimeout)
+		time.Sleep(100 * time.Millisecond)
+		// break
 	}
 }

replication_test.go

@@ -162,3 +162,9 @@ func (n *InMemNetwork) AdvanceWithoutLeader(round uint64, laggingNodeId simplex.
 		require.Equal(n.t, record.EmptyNotarizationRecordType, recordType)
 	}
 }
+
+func (n *InMemNetwork) AdvanceTime(duration time.Duration) {
+	for _, instance := range n.Instances {
+		instance.AdvanceTime(duration)	
+	}
+}

testutil/network.go

@@ -135,22 +135,3 @@ func (t *TestNode) TimeoutOnRound(round uint64) {
 		time.Sleep(50 * time.Millisecond)
 	}
 }
-
-func (t *TestNode) TickUntilRoundAdvanced(round uint64, tick time.Duration) {
-	timeout := time.NewTimer(time.Minute)
-	defer timeout.Stop()
-
-	for {
-		if t.E.Metadata().Round >= round {
-			return
-		}
-
-		select {
-		case <-time.After(time.Millisecond * 10):
-			t.AdvanceTime(tick)
-			continue
-		case <-timeout.C:
-			require.Fail(t.t, "timed out waiting to enter round", "current round %d, waiting for round %d", t.E.Metadata().Round, round)
-		}
-	}
-}