Aggregator message TTL

aggregator/aggregator.go

@@ -113,6 +113,7 @@ type Aggregator struct {
 	rollupBroadcaster    RollupBroadcasterer
 	httpClient           safeclient.SafeClient
 	wsClient             safeclient.SafeClient
+	clock                core.Clock
 
 	// TODO(edwin): once rpc & rest decouple from aggregator fome it with them
 	registry           *prometheus.Registry
@@ -231,6 +232,7 @@ func NewAggregator(
 		rollupBroadcaster:                      rollupBroadcaster,
 		httpClient:                             ethHttpClient,
 		wsClient:                               ethWsClient,
+		clock:                                  *core.SystemClock,
 		taskBlsAggregationService:              taskBlsAggregationService,
 		stateRootUpdateBlsAggregationService:   stateRootUpdateBlsAggregationService,
 		operatorSetUpdateBlsAggregationService: operatorSetUpdateBlsAggregationService,
@@ -393,7 +395,7 @@ func (agg *Aggregator) sendNewCheckpointTask() {
 		fromTimestamp = 0
 	}
 
-	toTimestamp := block.Time()
+	toTimestamp := block.Time() - uint64(types.MESSAGE_SUBMISSION_TIMEOUT.Seconds())
 
 	agg.logger.Info("Aggregator sending new task", "fromTimestamp", fromTimestamp, "toTimestamp", toTimestamp)
 	// Send checkpoint to the task manager contract
@@ -556,6 +558,13 @@ func (agg *Aggregator) ProcessSignedStateRootUpdateMessage(signedStateRootUpdate
 		return DigestError
 	}
 
+	timestamp := signedStateRootUpdateMessage.Message.Timestamp
+	err = agg.validateMessageTimestamp(timestamp)
+	if err != nil {
+		agg.logger.Error("Failed to validate message timestamp", "err", err, "timestamp", timestamp)
+		return err
+	}
+
 	err = agg.stateRootUpdateBlsAggregationService.InitializeMessageIfNotExists(
 		messageDigest,
 		coretypes.QUORUM_NUMBERS,
@@ -586,6 +595,13 @@ func (agg *Aggregator) ProcessSignedOperatorSetUpdateMessage(signedOperatorSetUp
 		return DigestError
 	}
 
+	timestamp := signedOperatorSetUpdateMessage.Message.Timestamp
+	err = agg.validateMessageTimestamp(timestamp)
+	if err != nil {
+		agg.logger.Error("Failed to validate message timestamp", "err", err, "timestamp", timestamp)
+		return err
+	}
+
 	blockNumber, err := agg.avsReader.GetOperatorSetUpdateBlock(context.Background(), signedOperatorSetUpdateMessage.Message.Id)
 	if err != nil {
 		agg.logger.Error("Failed to get operator set update block", "err", err)
@@ -675,3 +691,15 @@ func (agg *Aggregator) GetCheckpointMessages(fromTimestamp, toTimestamp uint64)
 		CheckpointMessages: *checkpointMessages,
 	}, nil
 }
+
+func (agg *Aggregator) validateMessageTimestamp(messageTimestamp uint64) error {
+	now := agg.clock.Now().Unix()
+	timeoutInSeconds := types.MESSAGE_SUBMISSION_TIMEOUT.Seconds()
+
+	// Prevent possible underflow (specially in testing)
+	if uint64(now) > uint64(timeoutInSeconds) && messageTimestamp < uint64(now)-uint64(timeoutInSeconds) {
+		return MessageExpiredError
+	}
+
+	return nil
+}

aggregator/aggregator_test.go

@@ -20,6 +20,7 @@ import (
 	aggmocks "github.com/NethermindEth/near-sffl/aggregator/mocks"
 	"github.com/NethermindEth/near-sffl/aggregator/types"
 	taskmanager "github.com/NethermindEth/near-sffl/contracts/bindings/SFFLTaskManager"
+	"github.com/NethermindEth/near-sffl/core"
 	chainiomocks "github.com/NethermindEth/near-sffl/core/chainio/mocks"
 	safeclientmocks "github.com/NethermindEth/near-sffl/core/safeclient/mocks"
 	coretypes "github.com/NethermindEth/near-sffl/core/types"
@@ -59,8 +60,8 @@ func TestSendNewTask(t *testing.T) {
 
 	var TASK_INDEX = uint32(0)
 	var BLOCK_NUMBER = uint32(100)
-	var FROM_TIMESTAMP = uint64(3)
-	var TO_TIMESTAMP = uint64(4)
+	var FROM_TIMESTAMP = uint64(30_000)
+	var TO_TIMESTAMP = uint64(40_000)
 
 	mockClient.EXPECT().BlockNumber(context.Background()).Return(uint64(BLOCK_NUMBER), nil)
 	mockClient.EXPECT().BlockByNumber(context.Background(), big.NewInt(int64(BLOCK_NUMBER))).Return(
@@ -69,7 +70,7 @@ func TestSendNewTask(t *testing.T) {
 	)
 
 	mockAvsWriterer.EXPECT().SendNewCheckpointTask(
-		context.Background(), FROM_TIMESTAMP, TO_TIMESTAMP, types.TASK_QUORUM_THRESHOLD, coretypes.QUORUM_NUMBERS,
+		context.Background(), FROM_TIMESTAMP, TO_TIMESTAMP-uint64(types.MESSAGE_SUBMISSION_TIMEOUT.Seconds()), types.TASK_QUORUM_THRESHOLD, coretypes.QUORUM_NUMBERS,
 	).Return(aggmocks.MockSendNewCheckpointTask(BLOCK_NUMBER, TASK_INDEX, FROM_TIMESTAMP, TO_TIMESTAMP))
 	mockAvsReaderer.EXPECT().GetLastCheckpointToTimestamp(context.Background()).Return(FROM_TIMESTAMP-1, nil)
 
@@ -149,6 +150,46 @@ func TestHandleOperatorSetUpdateAggregationReachedQuorum(t *testing.T) {
 	assert.NotContains(t, aggregator.operatorSetUpdates, msgDigest)
 }
 
+func TestExpiredStateRootUpdateMessage(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+
+	aggregator, _, _, _, _, _, _, _, _, err := createMockAggregator(mockCtrl, MOCK_OPERATOR_PUBKEY_DICT)
+	assert.NoError(t, err)
+
+	nowTimestamp := uint64(6000)
+	aggregator.clock = core.Clock{Now: func() time.Time { return time.Unix(int64(nowTimestamp), 0) }}
+	messageTimestamp := nowTimestamp - 60 - 1 // 60 seconds for message submission timeout and 1 second to be out of range
+
+	err = aggregator.ProcessSignedStateRootUpdateMessage(&messages.SignedStateRootUpdateMessage{
+		Message: messages.StateRootUpdateMessage{
+			Timestamp: messageTimestamp,
+		},
+	})
+
+	assert.Equal(t, MessageExpiredError, err)
+}
+
+func TestExpiredOperatorSetUpdate(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+
+	aggregator, _, _, _, _, _, _, _, _, err := createMockAggregator(mockCtrl, MOCK_OPERATOR_PUBKEY_DICT)
+	assert.NoError(t, err)
+
+	nowTimestamp := uint64(8000)
+	aggregator.clock = core.Clock{Now: func() time.Time { return time.Unix(int64(nowTimestamp), 0) }}
+	messageTimestamp := nowTimestamp - 60 - 1 // 60 seconds for message submission timeout and 1 second to be out of range
+
+	err = aggregator.ProcessSignedOperatorSetUpdateMessage(&messages.SignedOperatorSetUpdateMessage{
+		Message: messages.OperatorSetUpdateMessage{
+			Timestamp: messageTimestamp,
+		},
+	})
+
+	assert.Equal(t, MessageExpiredError, err)
+}
+
 func createMockAggregator(
 	mockCtrl *gomock.Controller, operatorPubkeyDict map[eigentypes.OperatorId]types.OperatorInfo,
 ) (*Aggregator, *chainiomocks.MockAvsReaderer, *chainiomocks.MockAvsWriterer, *blsaggservmock.MockBlsAggregationService, *aggmocks.MockMessageBlsAggregationService, *aggmocks.MockMessageBlsAggregationService, *dbmocks.MockDatabaser, *aggmocks.MockRollupBroadcasterer, *safeclientmocks.MockSafeClient, error) {
@@ -178,6 +219,7 @@ func createMockAggregator(
 		httpClient:                             mockClient,
 		wsClient:                               mockClient,
 		aggregatorListener:                     &SelectiveAggregatorListener{},
+		clock:                                  *core.SystemClock,
 	}
 	return aggregator, mockAvsReader, mockAvsWriter, mockTaskBlsAggregationService, mockStateRootUpdateBlsAggregationService, mockOperatorSetUpdateBlsAggregationService, mockMsgDb, mockRollupBroadcaster, mockClient, nil
 }

aggregator/rpc_server_test.go

@@ -5,6 +5,7 @@ import (
 	"encoding/binary"
 	"math/big"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/mock/gomock"
@@ -14,6 +15,7 @@ import (
 	eigentypes "github.com/Layr-Labs/eigensdk-go/types"
 
 	"github.com/NethermindEth/near-sffl/aggregator/types"
+	"github.com/NethermindEth/near-sffl/core"
 	coretypes "github.com/NethermindEth/near-sffl/core/types"
 	"github.com/NethermindEth/near-sffl/core/types/messages"
 )
@@ -56,10 +58,11 @@ func TestProcessSignedStateRootUpdateMessage(t *testing.T) {
 	aggregator, _, _, _, mockMessageBlsAggServ, _, _, _, _, err := createMockAggregator(mockCtrl, MOCK_OPERATOR_PUBKEY_DICT)
 	assert.Nil(t, err)
 
+	aggregator.clock = core.Clock{Now: func() time.Time { return time.Unix(10_000, 0) }}
 	message := messages.StateRootUpdateMessage{
 		RollupId:            1,
 		BlockHeight:         2,
-		Timestamp:           3,
+		Timestamp:           9_995,
 		NearDaCommitment:    keccak256(4),
 		NearDaTransactionId: keccak256(5),
 		StateRoot:           keccak256(6),
@@ -70,8 +73,7 @@ func TestProcessSignedStateRootUpdateMessage(t *testing.T) {
 	messageDigest, err := signedMessage.Message.Digest()
 	assert.Nil(t, err)
 
-	mockMessageBlsAggServ.EXPECT().ProcessNewSignature(context.Background(), messageDigest,
-		&signedMessage.BlsSignature, signedMessage.OperatorId)
+	mockMessageBlsAggServ.EXPECT().ProcessNewSignature(context.Background(), messageDigest, &signedMessage.BlsSignature, signedMessage.OperatorId)
 	mockMessageBlsAggServ.EXPECT().InitializeMessageIfNotExists(messageDigest, coretypes.QUORUM_NUMBERS, []eigentypes.QuorumThresholdPercentage{types.MESSAGE_AGGREGATION_QUORUM_THRESHOLD}, types.MESSAGE_TTL, types.MESSAGE_BLS_AGGREGATION_TIMEOUT, uint64(0))
 	err = aggregator.ProcessSignedStateRootUpdateMessage(signedMessage)
 	assert.Nil(t, err)
@@ -84,9 +86,10 @@ func TestProcessOperatorSetUpdateMessage(t *testing.T) {
 	aggregator, mockAvsReader, _, _, _, mockMessageBlsAggServ, _, _, _, err := createMockAggregator(mockCtrl, MOCK_OPERATOR_PUBKEY_DICT)
 	assert.Nil(t, err)
 
+	aggregator.clock = core.Clock{Now: func() time.Time { return time.Unix(10_000, 0) }}
 	message := messages.OperatorSetUpdateMessage{
 		Id:        1,
-		Timestamp: 2,
+		Timestamp: 9_995,
 		Operators: []coretypes.RollupOperator{
 			{Pubkey: bls.NewG1Point(big.NewInt(3), big.NewInt(4)), Weight: big.NewInt(5)},
 		},

aggregator/types/types.go

@@ -18,6 +18,7 @@ const QUERY_FILTER_FROM_BLOCK = uint64(1)
 
 const MESSAGE_TTL = 1 * time.Minute
 const MESSAGE_BLS_AGGREGATION_TIMEOUT = 30 * time.Second
+const MESSAGE_SUBMISSION_TIMEOUT = 1 * time.Minute
 
 type OperatorInfo struct {
 	OperatorPubkeys eigentypes.OperatorPubkeys

core/utils.go

@@ -2,6 +2,7 @@ package core
 
 import (
 	"math/big"
+	"time"
 
 	"github.com/Layr-Labs/eigensdk-go/crypto/bls"
 	eigentypes "github.com/Layr-Labs/eigensdk-go/types"
@@ -86,3 +87,9 @@ func ConvertBytesToQuorumNumbers(input []byte) []eigentypes.QuorumNum {
 	}
 	return output
 }
+
+type Clock struct {
+	Now func() time.Time
+}
+
+var SystemClock = &Clock{Now: time.Now}