tests for late coming messages

agreement/actions.go

@@ -322,14 +322,14 @@ func (a rezeroAction) do(ctx context.Context, s *Service) {
 
 	removeList := make([]round, 0)
 
-	for r, _ := range s.historicalClocks {
-		if a.Round < r+credentialRoundLag {
-			removeList = append(removeList, r)
+	for round := range s.historicalClocks {
+		if a.Round < round+credentialRoundLag {
+			removeList = append(removeList, round)
 		}
 	}
 
-	for _, r := range removeList {
-		delete(s.historicalClocks, r)
+	for _, round := range removeList {
+		delete(s.historicalClocks, round)
 	}
 }
 

agreement/events.go

@@ -22,7 +22,6 @@ import (
 
 	"github.com/algorand/go-algorand/logging"
 	"github.com/algorand/go-algorand/protocol"
-	"github.com/algorand/go-algorand/util/timers"
 )
 
 // An event represents the communication of an event to a state machine.
@@ -980,7 +979,7 @@ func (e checkpointEvent) AttachConsensusVersion(v ConsensusVersionView) external
 	return e
 }
 
-func getTimestampForEvent(eventRound round, d time.Duration, currentRound round, historicalClocks map[round]timers.Clock[TimeoutType]) time.Duration {
+func getTimestampForEvent(eventRound round, d time.Duration, currentRound round, historicalClocks map[round]historicalClock) time.Duration {
 	if eventRound > currentRound {
 		return time.Duration(1)
 	} else if eventRound == currentRound {
@@ -994,7 +993,7 @@ func getTimestampForEvent(eventRound round, d time.Duration, currentRound round,
 // AttachValidatedAt looks for a validated proposal or vote inside a
 // payloadVerified or voteVerified messageEvent, and attaches the given time to
 // the proposal's validatedAt field.
-func (e messageEvent) AttachValidatedAt(d time.Duration, currentRound round, historicalClocks map[round]timers.Clock[TimeoutType]) messageEvent {
+func (e messageEvent) AttachValidatedAt(d time.Duration, currentRound round, historicalClocks map[round]historicalClock) messageEvent {
 	switch e.T {
 	case payloadVerified:
 		e.Input.Proposal.validatedAt = getTimestampForEvent(e.Input.Proposal.Round(), d, currentRound, historicalClocks)
@@ -1007,7 +1006,7 @@ func (e messageEvent) AttachValidatedAt(d time.Duration, currentRound round, his
 // AttachReceivedAt looks for an unauthenticatedProposal inside a
 // payloadPresent or votePresent messageEvent, and attaches the given
 // time to the proposal's receivedAt field.
-func (e messageEvent) AttachReceivedAt(d time.Duration, currentRound round, historicalClocks map[round]timers.Clock[TimeoutType]) messageEvent {
+func (e messageEvent) AttachReceivedAt(d time.Duration, currentRound round, historicalClocks map[round]historicalClock) messageEvent {
 	if e.T == payloadPresent {
 		e.Input.UnauthenticatedProposal.receivedAt = getTimestampForEvent(e.Input.UnauthenticatedProposal.Round(), d, currentRound, historicalClocks)
 	} else if e.T == votePresent {

agreement/player_test.go

@@ -3325,6 +3325,63 @@ func TestPlayerRetainsEarlyReceivedValidatedAtOneSample(t *testing.T) {
 	require.Equal(t, time.Duration(1), pWhite.lowestCredentialArrivals.history[0])
 }
 
+type fixedHistoricalClock struct{}
+
+const fixedClockDuration = 900 * time.Millisecond
+
+func (fixedHistoricalClock) Since() time.Duration {
+	return fixedClockDuration
+}
+
+// test that ReceivedAt and ValidateAt timing information are retained in
+// proposalStore when the payloadPresent, payloadVerified, and voteVerified
+// events are processed credentialRoundLag after the round they belong to, and
+// that all timings are available when the ensureAction is called for the block.
+func TestPlayerRetainsLateReceivedValidatedAtOneSample(t *testing.T) {
+	partitiontest.PartitionTest(t)
+	const r = round(20239)
+	const p = period(0)
+	pWhite, pM, helper := setupP(t, r-1, p, soft)
+
+	historicalClocks := map[round]historicalClock{
+		r - credentialRoundLag - 1: fixedHistoricalClock{},
+	}
+
+	// send voteVerified message
+	pP, pV := helper.MakeRandomProposalPayload(t, r-credentialRoundLag-1)
+	vVote := helper.MakeVerifiedVote(t, 0, r-credentialRoundLag-1, p, propose, *pV)
+	inMsg := messageEvent{T: voteVerified, Input: message{Vote: vVote, UnauthenticatedVote: vVote.u()}}
+	inMsg = inMsg.AttachValidatedAt(401*time.Millisecond, r-1, historicalClocks)
+	err, panicErr := pM.transition(inMsg)
+	require.NoError(t, err)
+	require.NoError(t, panicErr)
+
+	// send voteVerified message
+	pP, pV = helper.MakeRandomProposalPayload(t, r-1)
+	vVote = helper.MakeVerifiedVote(t, 0, r-1, p, propose, *pV)
+	inMsg = messageEvent{T: voteVerified, Input: message{Vote: vVote, UnauthenticatedVote: vVote.u()}}
+	inMsg = inMsg.AttachValidatedAt(501*time.Millisecond, r-1, nil)
+	err, panicErr = pM.transition(inMsg)
+	require.NoError(t, err)
+	require.NoError(t, panicErr)
+
+	// send payloadPresent message
+	m := message{UnauthenticatedProposal: pP.u()}
+	inMsg = messageEvent{T: payloadPresent, Input: m}
+	inMsg = inMsg.AttachReceivedAt(time.Second, r-1, nil)
+	err, panicErr = pM.transition(inMsg)
+	require.NoError(t, err)
+	require.NoError(t, panicErr)
+
+	assertCorrectReceivedAtSet(t, pWhite, pM, helper, r, p, pP, pV, m, protocol.ConsensusFuture, time.Second)
+
+	// assert lowest vote validateAt time was recorded into payloadArrivals
+	require.NotZero(t, dynamicFilterCredentialArrivalHistory)
+	require.Equal(t, pWhite.lowestCredentialArrivals.writePtr, 1)
+	require.False(t, pWhite.lowestCredentialArrivals.isFull())
+	require.Equal(t, fixedClockDuration, pWhite.lowestCredentialArrivals.history[0])
+}
+
 // test that ReceivedAt and ValidateAt timing information are retained in proposalStore
 // when the payloadPresent, payloadVerified, and voteVerified events are processed, and that all timings
 // are available when the ensureAction is called for the block. The history should be kept for the last
@@ -3535,6 +3592,94 @@ func TestPlayerRetainsEarlyReceivedValidatedAtPPOneSample(t *testing.T) {
 	require.Equal(t, 502*time.Millisecond, pWhite.lowestCredentialArrivals.history[0])
 }
 
+// test that ReceivedAt and ValidateAt timing information are retained in
+// proposalStore when the payloadPresent (as part of the CompoundMessage
+// encoding used by PP messages), payloadVerified, and voteVerified events are
+// processed credentialRoundLag after the round they belong to, and that all
+// timings are available when the ensureAction is called for the block.
+func TestPlayerRetainsLateReceivedValidatedAtPPOneSample(t *testing.T) {
+	partitiontest.PartitionTest(t)
+
+	const r = round(20239)
+	const p = period(0)
+	pWhite, pM, helper := setupP(t, r-1, p, soft)
+	pP, pV := helper.MakeRandomProposalPayload(t, r-1)
+
+	// Move to round r, no credentials arrived.
+	// send voteVerified message
+	vVote := helper.MakeVerifiedVote(t, 0, r-1, p, propose, *pV)
+	inMsg := messageEvent{T: voteVerified, Input: message{Vote: vVote, UnauthenticatedVote: vVote.u()}}
+	inMsg = inMsg.AttachValidatedAt(501*time.Millisecond, r-1, nil)
+	err, panicErr := pM.transition(inMsg)
+	require.NoError(t, err)
+	require.NoError(t, panicErr)
+
+	// send payloadPresent message
+	m := message{UnauthenticatedProposal: pP.u()}
+	inMsg = messageEvent{T: payloadPresent, Input: m}
+	inMsg = inMsg.AttachReceivedAt(time.Second, r-1, nil)
+	err, panicErr = pM.transition(inMsg)
+	require.NoError(t, err)
+	require.NoError(t, panicErr)
+
+	assertCorrectReceivedAtSet(t, pWhite, pM, helper, r, p, pP, pV, m, protocol.ConsensusFuture, time.Second)
+	require.False(t, pWhite.lowestCredentialArrivals.isFull())
+	require.Equal(t, pWhite.lowestCredentialArrivals.writePtr, 0)
+
+	historicalClocks := map[round]historicalClock{
+		r - credentialRoundLag: fixedHistoricalClock{},
+	}
+	// create a PP message for the round we're going to take the sample from when round r-1 ends
+	// Now we're going to pretend we got the message credentialRoundLag too late.
+	vVote = helper.MakeVerifiedVote(t, 0, r-credentialRoundLag, p, propose, *pV)
+	unverifiedVoteMsg := message{UnauthenticatedVote: vVote.u()}
+	proposalMsg := message{UnauthenticatedProposal: pP.u()}
+	compoundMsg := messageEvent{T: votePresent, Input: unverifiedVoteMsg,
+		Tail: &messageEvent{T: payloadPresent, Input: proposalMsg}}
+	inMsg = compoundMsg.AttachReceivedAt(time.Second, r, historicalClocks) // call AttachReceivedAt like demux would
+	err, panicErr = pM.transition(inMsg)
+	require.NoError(t, err)
+	require.NoError(t, panicErr)
+
+	verifiedVoteMsg := message{Vote: vVote, UnauthenticatedVote: vVote.u()}
+	inMsg = messageEvent{T: voteVerified, Input: verifiedVoteMsg, TaskIndex: 1}
+	inMsg = inMsg.AttachValidatedAt(502*time.Millisecond, r, historicalClocks)
+	err, panicErr = pM.transition(inMsg)
+	require.NoError(t, err)
+	require.NoError(t, panicErr)
+
+	// send payloadPresent message
+	m = message{UnauthenticatedProposal: pP.u()}
+	inMsg = messageEvent{T: payloadPresent, Input: m}
+	inMsg = inMsg.AttachReceivedAt(time.Second, r-credentialRoundLag, historicalClocks)
+	err, panicErr = pM.transition(inMsg)
+	require.NoError(t, err)
+	require.NoError(t, panicErr)
+
+	// move to round r+1, triggering history update
+	vVote = helper.MakeVerifiedVote(t, 0, r, p, propose, *pV)
+	inMsg = messageEvent{T: voteVerified, Input: message{Vote: vVote, UnauthenticatedVote: vVote.u()}}
+	inMsg = inMsg.AttachValidatedAt(501*time.Millisecond, r, nil)
+	err, panicErr = pM.transition(inMsg)
+	require.NoError(t, err)
+	require.NoError(t, panicErr)
+
+	// send payloadPresent message
+	m = message{UnauthenticatedProposal: pP.u()}
+	inMsg = messageEvent{T: payloadPresent, Input: m}
+	inMsg = inMsg.AttachReceivedAt(time.Second, r, nil)
+	err, panicErr = pM.transition(inMsg)
+	require.NoError(t, err)
+	require.NoError(t, panicErr)
+	moveToRound(t, pWhite, pM, helper, r+1, p, pP, pV, m, protocol.ConsensusFuture)
+
+	// assert lowest vote validateAt time was recorded into payloadArrivals
+	require.NotNil(t, dynamicFilterCredentialArrivalHistory)
+	require.Equal(t, pWhite.lowestCredentialArrivals.writePtr, 1)
+	require.False(t, pWhite.lowestCredentialArrivals.isFull())
+	require.Equal(t, fixedClockDuration, pWhite.lowestCredentialArrivals.history[0])
+}
+
 // test that ReceivedAt and ValidateAt timing information are retained in
 // proposalStore when the payloadPresent (as part of the CompoundMessage
 // encoding used by PP messages), payloadVerified, and voteVerified events are
@@ -3762,6 +3907,97 @@ func TestPlayerRetainsEarlyReceivedValidatedAtAVPPOneSample(t *testing.T) {
 	require.Equal(t, time.Duration(1), pWhite.lowestCredentialArrivals.history[0])
 }
 
+// test that ReceivedAt and ValidateAt timing information are retained in
+// proposalStore when the voteVerified event comes in first (as part of the AV
+// message before PP), then the payloadPresent (as part of the CompoundMessage
+// encoding used by PP messages) and payloadVerified events are processed
+// credentialRoundLag after the round they belong to, and that all timings are
+// available when the ensureAction is called for the block.
+func TestPlayerRetainsLateReceivedValidatedAtAVPPOneSample(t *testing.T) {
+	partitiontest.PartitionTest(t)
+
+	const r = round(20239)
+	const p = period(0)
+	pWhite, pM, helper := setupP(t, r-1, p, soft)
+	pP, pV := helper.MakeRandomProposalPayload(t, r-1)
+
+	// Move to round r, no credentials arrived.
+	// send voteVerified message
+	vVote := helper.MakeVerifiedVote(t, 0, r-1, p, propose, *pV)
+	inMsg := messageEvent{T: voteVerified, Input: message{Vote: vVote, UnauthenticatedVote: vVote.u()}}
+	inMsg = inMsg.AttachValidatedAt(501*time.Millisecond, r-1, nil)
+	err, panicErr := pM.transition(inMsg)
+	require.NoError(t, err)
+	require.NoError(t, panicErr)
+
+	// send payloadPresent message
+	m := message{UnauthenticatedProposal: pP.u()}
+	inMsg = messageEvent{T: payloadPresent, Input: m}
+	inMsg = inMsg.AttachReceivedAt(time.Second, r-1, nil)
+	err, panicErr = pM.transition(inMsg)
+	require.NoError(t, err)
+	require.NoError(t, panicErr)
+
+	assertCorrectReceivedAtSet(t, pWhite, pM, helper, r, p, pP, pV, m, protocol.ConsensusFuture, time.Second)
+	require.False(t, pWhite.lowestCredentialArrivals.isFull())
+	require.Equal(t, pWhite.lowestCredentialArrivals.writePtr, 0)
+
+	// send votePresent message (mimicking the first AV message validating)
+	vVote = helper.MakeVerifiedVote(t, 0, r-credentialRoundLag, p, propose, *pV)
+	unverifiedVoteMsg := message{UnauthenticatedVote: vVote.u()}
+	inMsg = messageEvent{T: votePresent, Input: unverifiedVoteMsg}
+	err, panicErr = pM.transition(inMsg)
+	require.NoError(t, err)
+	require.NoError(t, panicErr)
+
+	// make sure vote verify requests
+	verifyEvent := ev(cryptoAction{T: verifyVote, M: unverifiedVoteMsg, Round: r - 1, Period: p, Step: propose, TaskIndex: 1})
+	require.Truef(t, pM.getTrace().Contains(verifyEvent), "Player should verify vote")
+
+	historicalClocks := map[round]historicalClock{
+		r - credentialRoundLag: fixedHistoricalClock{},
+	}
+	// send voteVerified, pretend we're credentialRoundLag after the message was sent
+	verifiedVoteMsg := message{Vote: vVote, UnauthenticatedVote: vVote.u()}
+	inMsg = messageEvent{T: voteVerified, Input: verifiedVoteMsg, TaskIndex: 1}
+	inMsg = inMsg.AttachValidatedAt(502*time.Millisecond, r, historicalClocks)
+	err, panicErr = pM.transition(inMsg)
+	require.NoError(t, err)
+	require.NoError(t, panicErr)
+
+	// create a PP message for an arbitrary proposal/payload similar to setupCompoundMessage
+	proposalMsg := message{UnauthenticatedProposal: pP.u()}
+	compoundMsg := messageEvent{T: votePresent, Input: unverifiedVoteMsg,
+		Tail: &messageEvent{T: payloadPresent, Input: proposalMsg}}
+	inMsg = compoundMsg.AttachReceivedAt(time.Second, r-credentialRoundLag, nil) // call AttachReceivedAt like demux would
+	err, panicErr = pM.transition(inMsg)
+	require.NoError(t, err)
+	require.NoError(t, panicErr)
+
+	// move to round r+1, triggering history update
+	vVote = helper.MakeVerifiedVote(t, 0, r, p, propose, *pV)
+	inMsg = messageEvent{T: voteVerified, Input: message{Vote: vVote, UnauthenticatedVote: vVote.u()}}
+	inMsg = inMsg.AttachValidatedAt(time.Second, r, nil)
+	err, panicErr = pM.transition(inMsg)
+	require.NoError(t, err)
+	require.NoError(t, panicErr)
+
+	// send payloadPresent message
+	m = message{UnauthenticatedProposal: pP.u()}
+	inMsg = messageEvent{T: payloadPresent, Input: m}
+	inMsg = inMsg.AttachReceivedAt(time.Second, r, nil)
+	err, panicErr = pM.transition(inMsg)
+	require.NoError(t, err)
+	require.NoError(t, panicErr)
+	moveToRound(t, pWhite, pM, helper, r+1, p, pP, pV, m, protocol.ConsensusFuture)
+
+	// assert lowest vote validateAt time was recorded into payloadArrivals
+	require.NotNil(t, dynamicFilterCredentialArrivalHistory)
+	require.Equal(t, pWhite.lowestCredentialArrivals.writePtr, 1)
+	require.False(t, pWhite.lowestCredentialArrivals.isFull())
+	require.Equal(t, fixedClockDuration, pWhite.lowestCredentialArrivals.history[0])
+}
+
 func TestPlayerRetainsReceivedValidatedAtAVPPHistoryWindow(t *testing.T) {
 	partitiontest.PartitionTest(t)
 	const r = round(20239)

agreement/service.go

@@ -19,6 +19,7 @@ package agreement
 //go:generate dbgen -i agree.sql -p agreement -n agree -o agreeInstall.go -h ../scripts/LICENSE_HEADER
 import (
 	"context"
+	"time"
 
 	"github.com/algorand/go-algorand/config"
 	"github.com/algorand/go-algorand/logging"
@@ -57,7 +58,7 @@ type Service struct {
 	persistStatus  player
 	persistActions []action
 
-	historicalClocks map[round]timers.Clock[TimeoutType]
+	historicalClocks map[round]historicalClock
 }
 
 // Parameters holds the parameters necessary to run the agreement protocol.
@@ -86,6 +87,11 @@ type externalDemuxSignals struct {
 	CurrentRound         round
 }
 
+// an interface representing a clock from a previous round
+type historicalClock interface {
+	Since() time.Duration
+}
+
 // MakeService creates a new Agreement Service instance given a set of Parameters.
 //
 // Call Start to start execution and Shutdown to finish execution.
@@ -107,7 +113,7 @@ func MakeService(p Parameters) (*Service, error) {
 
 	s.persistenceLoop = makeAsyncPersistenceLoop(s.log, s.Accessor, s.Ledger)
 
-	s.historicalClocks = make(map[round]timers.Clock[TimeoutType])
+	s.historicalClocks = make(map[round]historicalClock)
 
 	return s, nil
 }