feat: add idempotency key support (#244)

* feat: add idempotency key support to prevent duplicate CCV verification processing

Updates aggregator database schema, API validation, and storage layer to use idempotency_key for deduplication instead of verification_timestamp. Modifies verifier pipeline to generate and propagate idempotency keys through the CCVDataWithIdempotencyKey paired struct from source readers to storage operations.

* Fix merge issues

* Address PR comments

Author: carte7000

aggregator/migrations/postgres/00001_create_all_tables.sql

@@ -16,8 +16,9 @@ CREATE TABLE IF NOT EXISTS commit_verification_records (
     signature_s BYTEA NOT NULL DEFAULT '',
     ccv_node_data BYTEA NOT NULL,
     verification_timestamp BIGINT NOT NULL,
+    idempotency_key UUID NOT NULL,
     created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
-    CONSTRAINT unique_verification UNIQUE (message_id, committee_id, signer_address, verification_timestamp)
+    CONSTRAINT unique_verification UNIQUE (message_id, committee_id, signer_address, idempotency_key)
 );
 
 CREATE TABLE IF NOT EXISTS commit_aggregated_reports (

aggregator/pkg/handlers/validations.go

@@ -2,17 +2,22 @@ package handlers
 
 import (
 	"bytes"
+	"regexp"
 
 	"github.com/smartcontractkit/chainlink-ccv/aggregator/pkg/model"
 
 	validation "github.com/go-ozzo/ozzo-validation/v4"
 	pb "github.com/smartcontractkit/chainlink-protos/chainlink-ccv/go/v1"
 )
 
+// uuidRegex matches standard UUID format (with or without hyphens).
+var uuidRegex = regexp.MustCompile(`^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$`)
+
 func validateWriteRequest(req *pb.WriteCommitCCVNodeDataRequest) error {
 	err := validation.ValidateStruct(
 		req,
-		validation.Field(&req.CcvNodeData, validation.Required))
+		validation.Field(&req.CcvNodeData, validation.Required),
+		validation.Field(&req.IdempotencyKey, validation.Required, validation.Match(uuidRegex).Error("must be a valid UUID")))
 	if err != nil {
 		return err
 	}

aggregator/pkg/handlers/write_commit_ccv_data.go

@@ -3,6 +3,7 @@ package handlers
 import (
 	"context"
 
+	"github.com/google/uuid"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
@@ -67,12 +68,22 @@ func (h *WriteCommitCCVNodeDataHandler) Handle(ctx context.Context, req *pb.Writ
 		signerCtx := scope.WithAddress(ctx, signer.Address)
 		signerCtx = scope.WithParticipantID(signerCtx, signer.ParticipantID)
 		signerCtx = scope.WithCommitteeID(signerCtx, signer.CommitteeID)
+
+		// Parse the idempotency key as UUID
+		idempotencyUUID, err := uuid.Parse(req.GetIdempotencyKey())
+		if err != nil {
+			return &pb.WriteCommitCCVNodeDataResponse{
+				Status: pb.WriteStatus_FAILED,
+			}, status.Errorf(codes.InvalidArgument, "invalid idempotency key format: %v", err)
+		}
+
 		record := model.CommitVerificationRecord{
 			MessageWithCCVNodeData: *req.GetCcvNodeData(),
 			IdentifierSigner:       signer,
 			CommitteeID:            signer.CommitteeID,
+			IdempotencyKey:         idempotencyUUID,
 		}
-		err := h.storage.SaveCommitVerification(signerCtx, &record)
+		err = h.storage.SaveCommitVerification(signerCtx, &record)
 		if err != nil {
 			return &pb.WriteCommitCCVNodeDataResponse{
 				Status: pb.WriteStatus_FAILED,

aggregator/pkg/model/commit_verification_record.go

@@ -4,6 +4,8 @@ import (
 	"encoding/hex"
 	"fmt"
 
+	"github.com/google/uuid"
+
 	pb "github.com/smartcontractkit/chainlink-protos/chainlink-ccv/go/v1"
 )
 
@@ -26,7 +28,8 @@ func (c CommitVerificationRecordIdentifier) ToIdentifier() string {
 type CommitVerificationRecord struct {
 	IdentifierSigner *IdentifierSigner
 	pb.MessageWithCCVNodeData
-	CommitteeID CommitteeID
+	CommitteeID    CommitteeID
+	IdempotencyKey uuid.UUID
 }
 
 // GetID retrieves the unique identifier for the commit verification record.

aggregator/pkg/storage/postgres/database_storage.go

@@ -142,9 +142,9 @@ func (d *DatabaseStorage) SaveCommitVerification(ctx context.Context, record *mo
 
 	stmt := `INSERT INTO commit_verification_records 
 		(message_id, committee_id, participant_id, signer_address, source_chain_selector, dest_chain_selector, 
-		 onramp_address, offramp_address, signature_r, signature_s, ccv_node_data, verification_timestamp) 
-		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
-		ON CONFLICT (message_id, committee_id, signer_address, verification_timestamp) 
+		 onramp_address, offramp_address, signature_r, signature_s, ccv_node_data, verification_timestamp, idempotency_key) 
+		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
+		ON CONFLICT (message_id, committee_id, signer_address, idempotency_key) 
 		DO NOTHING`
 
 	ccvNodeData, err := proto.Marshal(&record.MessageWithCCVNodeData)
@@ -174,6 +174,7 @@ func (d *DatabaseStorage) SaveCommitVerification(ctx context.Context, record *mo
 		record.IdentifierSigner.SignatureS[:],
 		ccvNodeData,
 		record.Timestamp,
+		record.IdempotencyKey,
 	)
 	if err != nil {
 		return fmt.Errorf("failed to save commit verification record: %w", err)

aggregator/pkg/storage/postgres/database_storage_test.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/google/uuid"
 	"github.com/jmoiron/sqlx"
 	"github.com/stretchr/testify/require"
 	"github.com/testcontainers/testcontainers-go"
@@ -181,7 +182,8 @@ func createTestCommitVerificationRecord(msgWithCCV *pb.MessageWithCCVNodeData, s
 			SignatureS:  s32,
 			CommitteeID: committeeID,
 		},
-		CommitteeID: committeeID,
+		CommitteeID:    committeeID,
+		IdempotencyKey: uuid.New(), // Generate unique idempotency key for each record
 	}
 }