feat(operations): introduce WithIdempotencyKey (#1016)

Operation API cache/report reuse is based on the definition + input of
the operation, sometimes we have use cases where the input is the same
but the dependency is not the same and user would like the operation for
it to be executed instead of being reused with the existing
cache/reports. Eg when the chain client passed into dep can be for
different chains but implements the same interface.

Because of this, users have made
[workarounds](https://github.com/smartcontractkit/payments/blob/5462c054fd879a0da77fffd68fa7b0c069d36d2d/changeset/shared/operations/access_control.go#L180-L185)
by passing chain-selectors or custom uuid into the input purely for
cache breaking reasons.

This change introduce an extra hash key that user can provide as a
differentiator. The API will also consider this new key as part of the
hash key for reusing previous successful results. By providing this key,
user avoid having to introduce extra placeholder param in the input
struct.

```go
result, err = operations.ExecuteOperation(bundle, op, deps, input, operations.WithIdempotencyKey[InputType, DepsType](extra-key))
```

This change originated from discussing with Rayene and Rens.

JIRA: https://smartcontract-it.atlassian.net/browse/CLD-2579

Author: graham-chainlink

.changeset/curly-hoops-listen.md

@@ -0,0 +1,5 @@
+---
+"chainlink-deployments-framework": minor
+---
+
+feat(operations): introduce WithIdempotencyKey

operations/doc.go

@@ -55,6 +55,9 @@ Reporter:
 	// Force execution and ignore previous successful reports.
 	result, err = operations.ExecuteOperation(bundle, op, deps, input, operations.WithForceExecute[InputType, DepsType]())
 
+	// A different idempotency key runs again; the same key reuses the prior successful report.
+	result, err = operations.ExecuteOperation(bundle, op, deps, input, operations.WithIdempotencyKey[InputType, DepsType]("extra-key"))
+
 	// Execute a sequence.
 	_, err = operations.ExecuteSequence(bundle, sequence, deps, input)
 */

operations/execute.go

@@ -15,6 +15,8 @@ type ExecuteConfig[IN, DEP any] struct {
 	retryConfig RetryConfig[IN, DEP]
 	// forceExecute controls whether execution should skip execution when previous successful report is found (set by WithForceExecute).
 	forceExecute bool
+	// idempotencyKey scopes report reuse beyond operation definition and input (set by WithIdempotencyKey).
+	idempotencyKey string
 }
 
 type ExecuteOption[IN, DEP any] func(*ExecuteConfig[IN, DEP])
@@ -94,6 +96,15 @@ func WithForceExecute[IN, DEP any]() ExecuteOption[IN, DEP] {
 	}
 }
 
+// WithIdempotencyKey is an ExecuteOption that adds an extra component to the idempotency hash.
+// The hash will then be built from the operation definition, input, and this key.
+// Use it when the same operation input can legitimately produce different results, so a later run should not reuse an earlier result.
+func WithIdempotencyKey[IN, DEP any](idempotencyKey string) ExecuteOption[IN, DEP] {
+	return func(c *ExecuteConfig[IN, DEP]) {
+		c.idempotencyKey = idempotencyKey
+	}
+}
+
 // WithOperationNRetry is an ExecuteOperationNOption that enables the default retry for each run in ExecuteOperationN.
 func WithOperationNRetry[IN, DEP any]() ExecuteOperationNOption[IN, DEP] {
 	return func(c *ExecuteOperationNConfig[IN, DEP]) {
@@ -151,7 +162,7 @@ func ExecuteOperation[IN, OUT, DEP any](
 		opt(executeConfig)
 	}
 	if !executeConfig.forceExecute {
-		if previousReport, ok := loadPreviousSuccessfulReport[IN, OUT](b, operation.def, input); ok {
+		if previousReport, ok := loadPreviousSuccessfulReport[IN, OUT](b, operation.def, input, executeConfig.idempotencyKey); ok {
 			b.Logger.Infow("Operation already executed. Returning previous result", "id", operation.def.ID,
 				"version", operation.def.Version, "description", operation.def.Description)
 
@@ -173,6 +184,7 @@ func ExecuteOperation[IN, OUT, DEP any](
 	}
 
 	report := NewReport(operation.def, input, output, err)
+	report.IdempotencyKey = executeConfig.idempotencyKey
 	if err = b.reporter.AddReport(genericReport(report)); err != nil {
 		return Report[IN, OUT]{}, err
 	}
@@ -314,7 +326,7 @@ func ExecuteSequence[IN, OUT, DEP any](
 		return SequenceReport[IN, OUT]{}, fmt.Errorf("sequence %s input: %w", sequence.def.ID, ErrNotSerializable)
 	}
 
-	if previousReport, ok := loadPreviousSuccessfulReport[IN, OUT](b, sequence.def, input); ok {
+	if previousReport, ok := loadPreviousSuccessfulReport[IN, OUT](b, sequence.def, input, ""); ok {
 		executionReports, err := b.reporter.GetExecutionReports(previousReport.ID)
 		if err != nil {
 			return SequenceReport[IN, OUT]{}, err
@@ -382,14 +394,14 @@ func NewUnrecoverableError(err error) error {
 }
 
 func loadPreviousSuccessfulReport[IN, OUT any](
-	b Bundle, def Definition, input IN,
+	b Bundle, def Definition, input IN, idempotencyKey string,
 ) (Report[IN, OUT], bool) {
 	prevReports, err := b.reporter.GetReports()
 	if err != nil {
 		b.Logger.Errorw("Failed to get reports", "error", err)
 		return Report[IN, OUT]{}, false
 	}
-	currentHash, err := constructUniqueHashFrom(b.reportHashCache, def, input)
+	currentHash, err := constructUniqueHashFrom(b.reportHashCache, def, input, idempotencyKey)
 	if err != nil {
 		b.Logger.Errorw("Failed to construct unique hash", "error", err)
 		return Report[IN, OUT]{}, false
@@ -400,7 +412,7 @@ func loadPreviousSuccessfulReport[IN, OUT any](
 	for i := len(prevReports) - 1; i >= 0; i-- {
 		report := prevReports[i]
 		// Check if operation/sequence was run previously and return the report if successful
-		reportHash, err := constructUniqueHashFrom(b.reportHashCache, report.Def, report.Input)
+		reportHash, err := constructUniqueHashFrom(b.reportHashCache, report.Def, report.Input, report.IdempotencyKey)
 		if err != nil {
 			b.Logger.Errorw("Failed to construct unique hash for previous report", "error", err)
 			continue
@@ -430,7 +442,7 @@ func loadSuccessfulExecutionSeriesReports[IN, OUT any](
 		b.Logger.Errorw("Failed to get reports", "error", err)
 		return []Report[IN, OUT]{}, false
 	}
-	currentHash, err := constructUniqueHashFrom(b.reportHashCache, def, input)
+	currentHash, err := constructUniqueHashFrom(b.reportHashCache, def, input, "")
 	if err != nil {
 		b.Logger.Errorw("Failed to construct unique hash", "error", err)
 		return []Report[IN, OUT]{}, false
@@ -442,7 +454,7 @@ func loadSuccessfulExecutionSeriesReports[IN, OUT any](
 		if report.ExecutionSeries == nil || report.ExecutionSeries.ID != seriesID {
 			continue
 		}
-		reportHash, err := constructUniqueHashFrom(b.reportHashCache, report.Def, report.Input)
+		reportHash, err := constructUniqueHashFrom(b.reportHashCache, report.Def, report.Input, "")
 		if err != nil {
 			b.Logger.Errorw("Failed to construct unique hash for previous report", "error", err)
 			continue

operations/execute_test.go

@@ -212,13 +212,29 @@ func Test_ExecuteOperation_WithPreviousRun(t *testing.T) {
 	assert.Equal(t, 4, res.Output)
 	assert.Equal(t, 3, handlerCalledTimes)
 
+	// same input with a different hash key should execute again
+	res, err = ExecuteOperation(bundle, op, nil, 1, WithIdempotencyKey[int, any]("other-key"))
+	require.NoError(t, err)
+	require.Nil(t, res.Err)
+	assert.Equal(t, 2, res.Output)
+	assert.Equal(t, 4, handlerCalledTimes)
+	idempotencyKeyRunID := res.ID
+	assert.NotEqual(t, forcedRunID, idempotencyKeyRunID)
+
+	// same input and idempotency key should reuse that report
+	res, err = ExecuteOperation(bundle, op, nil, 1, WithIdempotencyKey[int, any]("other-key"))
+	require.NoError(t, err)
+	require.Nil(t, res.Err)
+	assert.Equal(t, idempotencyKeyRunID, res.ID)
+	assert.Equal(t, 4, handlerCalledTimes)
+
 	// new run with different op, should perform execution
 	op = NewOperation("plus1-v2", semver.MustParse("2.0.0"), "test operation", handler)
 	res, err = ExecuteOperation(bundle, op, nil, 1)
 	require.NoError(t, err)
 	require.Nil(t, res.Err)
 	assert.Equal(t, 2, res.Output)
-	assert.Equal(t, 4, handlerCalledTimes)
+	assert.Equal(t, 5, handlerCalledTimes)
 
 	// new run with op that returns error
 	res, err = ExecuteOperation(bundle, opWithError, nil, 1)
@@ -718,7 +734,7 @@ func Test_loadPreviousSuccessfulReport(t *testing.T) {
 				bundle.reporter = tt.setupReporter()
 			}
 
-			report, found := loadPreviousSuccessfulReport[float64, int](bundle, definition, tt.input)
+			report, found := loadPreviousSuccessfulReport[float64, int](bundle, definition, tt.input, "")
 			assert.Equal(t, tt.wantFound, found)
 
 			if tt.wantFound {

operations/hashing.go

@@ -9,24 +9,37 @@ import (
 	"sync"
 )
 
-func constructUniqueHashFrom(hashCache *sync.Map, def Definition, input any) (string, error) {
-	// convert def and input into string to be used as cachekey
-	// as input can be any type, some types cannot be used as cache argument for sync.map
-	// converting to string ensure argument is always valid
-
+func buildUniqueHashCacheKey(def Definition, input any, idempotencyKey string) ([]byte, error) {
+	// convert def and input into bytes used as the sync.Map cache key
+	// as input can be any type, some types cannot be used as cache argument for sync.Map
 	key, err := json.Marshal(def)
 	if err != nil {
-		return "", err
+		return nil, err
 	}
 
 	// convert to a canonical representation that's stable regardless of field order
-	// If the keys are not sorted, the hash generated can be different in maps.
 	inputBytes, err := canonicalizeJSON(input)
 	if err != nil {
-		return "", err
+		return nil, err
 	}
 
 	key = append(key, inputBytes...)
+	if idempotencyKey != "" {
+		// NUL separates canonical input from the idempotency key so concatenations cannot collide
+		// (for example, input "1" + key "23" vs input "12" + key "3").
+		key = append(key, 0)
+		key = append(key, []byte(idempotencyKey)...)
+	}
+
+	return key, nil
+}
+
+func constructUniqueHashFrom(hashCache *sync.Map, def Definition, input any, idempotencyKey string) (string, error) {
+	key, err := buildUniqueHashCacheKey(def, input, idempotencyKey)
+	if err != nil {
+		return "", err
+	}
+
 	if cached, ok := hashCache.Load(string(key)); ok {
 		return cached.(string), nil
 	}

operations/hashing_test.go

@@ -1,7 +1,6 @@
 package operations
 
 import (
-	"encoding/json"
 	"math"
 	"sync"
 	"testing"
@@ -12,8 +11,9 @@ import (
 )
 
 type argument struct {
-	def   Definition
-	input any
+	def            Definition
+	input          any
+	idempotencyKey string
 }
 
 func Test_constructUniqueHashFrom(t *testing.T) {
@@ -153,15 +153,45 @@ func Test_constructUniqueHashFrom(t *testing.T) {
 			},
 			wantSame: true,
 		},
+		{
+			name: "Same def and input with different idempotency keys should have different hash",
+			left: argument{
+				def:            definition,
+				input:          Input{A: 1, B: 2},
+				idempotencyKey: "a",
+			},
+			right: argument{
+				def:            definition,
+				input:          Input{A: 1, B: 2},
+				idempotencyKey: "b",
+			},
+			wantSame:       false,
+			skipCacheCheck: true,
+		},
+		{
+			name: "Input and idempotency key concatenation must not collide",
+			left: argument{
+				def:            definition,
+				input:          1,
+				idempotencyKey: "23",
+			},
+			right: argument{
+				def:            definition,
+				input:          12,
+				idempotencyKey: "3",
+			},
+			wantSame:       false,
+			skipCacheCheck: true,
+		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			t.Parallel()
 
 			cache := &sync.Map{}
-			hash1, err1 := constructUniqueHashFrom(cache, tt.left.def, tt.left.input)
-			hash2, err2 := constructUniqueHashFrom(&sync.Map{}, tt.right.def, tt.right.input)
+			hash1, err1 := constructUniqueHashFrom(cache, tt.left.def, tt.left.input, tt.left.idempotencyKey)
+			hash2, err2 := constructUniqueHashFrom(&sync.Map{}, tt.right.def, tt.right.input, tt.right.idempotencyKey)
 
 			if tt.wantErr != "" {
 				require.Error(t, err1)
@@ -182,19 +212,15 @@ func Test_constructUniqueHashFrom(t *testing.T) {
 			}
 
 			if !tt.skipCacheCheck {
-				// Verify cache behavior
-				defBytes1, err := json.Marshal(tt.left.def)
-				require.NoError(t, err)
-				inputBytes1, err := json.Marshal(tt.left.input)
+				cacheKey1, err := buildUniqueHashCacheKey(tt.left.def, tt.left.input, tt.left.idempotencyKey)
 				require.NoError(t, err)
-				cacheKey1 := string(append(defBytes1, inputBytes1...))
 
-				cached1, ok := cache.Load(cacheKey1)
+				cached1, ok := cache.Load(string(cacheKey1))
 				require.True(t, ok)
 				assert.Equal(t, hash1, cached1)
 
 				// Second call should use cache
-				cachedHash1, err := constructUniqueHashFrom(cache, tt.left.def, tt.left.input)
+				cachedHash1, err := constructUniqueHashFrom(cache, tt.left.def, tt.left.input, tt.left.idempotencyKey)
 				require.NoError(t, err)
 				assert.Equal(t, hash1, cachedHash1)
 			}

operations/report.go

@@ -24,6 +24,9 @@ type Report[IN, OUT any] struct {
 	ChildOperationReports []string `json:"childOperationReports"`
 	// ExecutionSeries is used to track the execution of an operation that was executed multiple times
 	ExecutionSeries *ExecutionSeries `json:"executionSeries,omitempty"`
+	// IdempotencyKey is an additional component of the idempotency hash. This is used when the same operation input
+	// can legitimately produce different results by providing different idempotency keys. Set via WithIdempotencyKey on ExecuteOperation.
+	IdempotencyKey string `json:"idempotencyKey,omitempty"`
 }
 
 // ExecutionSeries is used to track the execution of an operation that was executed multiple times.
@@ -272,6 +275,7 @@ func genericReport[IN, OUT any](r Report[IN, OUT]) Report[any, any] {
 		Err:                   r.Err,
 		ChildOperationReports: r.ChildOperationReports,
 		ExecutionSeries:       r.ExecutionSeries,
+		IdempotencyKey:        r.IdempotencyKey,
 	}
 }
 
@@ -310,6 +314,7 @@ func typeReport[IN, OUT any](r Report[any, any]) (Report[IN, OUT], bool) {
 		Err:                   r.Err,
 		ChildOperationReports: r.ChildOperationReports,
 		ExecutionSeries:       r.ExecutionSeries,
+		IdempotencyKey:        r.IdempotencyKey,
 	}, true
 }