[CLD-150]: fix: migration operations package

Taskfile.yml

@@ -0,0 +1,10 @@
+# https://taskfile.dev
+
+version: "3"
+
+includes:
+  lint:
+    taskfile: ./taskfiles/lint/Taskfile.yml
+
+  test:
+    taskfile: ./taskfiles/test/Taskfile.yml

go.mod

@@ -4,19 +4,79 @@ go 1.24.1
 
 require (
 	github.com/Masterminds/semver/v3 v3.3.1
+	github.com/avast/retry-go/v4 v4.6.1
 	github.com/ethereum/go-ethereum v1.15.7
+	github.com/google/uuid v1.6.0
 	github.com/pkg/errors v0.9.1
 	github.com/smartcontractkit/chain-selectors v1.0.49
+	github.com/smartcontractkit/chainlink-common v0.6.0
+	github.com/smartcontractkit/mcms v0.16.1
 	github.com/stretchr/testify v1.10.0
+	go.uber.org/zap v1.27.0
 	golang.org/x/exp v0.0.0-20250408133849-7e4ce0ab07d0
 )
 
 require (
-	github.com/davecgh/go-spew v1.1.1 // indirect
+	filippo.io/edwards25519 v1.0.0-rc.1 // indirect
+	github.com/Microsoft/go-winio v0.6.2 // indirect
+	github.com/andres-erbsen/clock v0.0.0-20160526145045-9e14626cd129 // indirect
+	github.com/bits-and-blooms/bitset v1.17.0 // indirect
+	github.com/blendle/zapdriver v1.3.1 // indirect
+	github.com/consensys/bavard v0.1.22 // indirect
+	github.com/consensys/gnark-crypto v0.14.0 // indirect
+	github.com/crate-crypto/go-ipa v0.0.0-20240724233137-53bbb0ceb27a // indirect
+	github.com/crate-crypto/go-kzg-4844 v1.1.0 // indirect
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
+	github.com/deckarep/golang-set/v2 v2.6.0 // indirect
+	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.3.0 // indirect
+	github.com/ethereum/c-kzg-4844 v1.0.3 // indirect
+	github.com/ethereum/go-verkle v0.2.2 // indirect
+	github.com/fatih/color v1.17.0 // indirect
+	github.com/fsnotify/fsnotify v1.7.0 // indirect
+	github.com/gabriel-vasile/mimetype v1.4.8 // indirect
+	github.com/gagliardetto/binary v0.8.0 // indirect
+	github.com/gagliardetto/solana-go v1.12.0 // indirect
+	github.com/gagliardetto/treeout v0.1.4 // indirect
+	github.com/go-ole/go-ole v1.3.0 // indirect
+	github.com/go-playground/locales v0.14.1 // indirect
+	github.com/go-playground/universal-translator v0.18.1 // indirect
+	github.com/go-playground/validator/v10 v10.24.0 // indirect
+	github.com/gorilla/websocket v1.5.0 // indirect
 	github.com/holiman/uint256 v1.3.2 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/karalabe/hid v1.0.1-0.20240306101548-573246063e52 // indirect
+	github.com/klauspost/compress v1.17.11 // indirect
+	github.com/leodido/go-urn v1.4.0 // indirect
+	github.com/logrusorgru/aurora v2.0.3+incompatible // indirect
+	github.com/mattn/go-colorable v0.1.13 // indirect
+	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/mitchellh/go-testing-interface v1.14.1 // indirect
+	github.com/mmcloughlin/addchain v0.4.0 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/mostynb/zstdpool-freelist v0.0.0-20201229113212-927304c0c3b1 // indirect
 	github.com/mr-tron/base58 v1.2.0 // indirect
-	github.com/pmezard/go-difflib v1.0.0 // indirect
-	golang.org/x/crypto v0.35.0 // indirect
-	golang.org/x/sys v0.30.0 // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
+	github.com/shirou/gopsutil v3.21.11+incompatible // indirect
+	github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20250226104101-11778f2ead98 // indirect
+	github.com/smartcontractkit/libocr v0.0.0-20250220133800-f3b940c4f298 // indirect
+	github.com/spf13/cast v1.7.1 // indirect
+	github.com/streamingfast/logging v0.0.0-20230608130331-f22c91403091 // indirect
+	github.com/supranational/blst v0.3.14 // indirect
+	github.com/tklauser/go-sysconf v0.3.12 // indirect
+	github.com/tklauser/numcpus v0.6.1 // indirect
+	github.com/yusufpapurcu/wmi v1.2.3 // indirect
+	go.mongodb.org/mongo-driver v1.12.2 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/ratelimit v0.2.0 // indirect
+	golang.org/x/crypto v0.37.0 // indirect
+	golang.org/x/net v0.39.0 // indirect
+	golang.org/x/sync v0.13.0 // indirect
+	golang.org/x/sys v0.32.0 // indirect
+	golang.org/x/term v0.31.0 // indirect
+	golang.org/x/text v0.24.0 // indirect
+	golang.org/x/time v0.9.0 // indirect
+	golang.org/x/tools v0.32.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
+	rsc.io/tmplfunc v0.0.3 // indirect
 )

operations/execute.go

@@ -0,0 +1,229 @@
+package operations
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/avast/retry-go/v4"
+)
+
+var ErrNotSerializable = errors.New("data cannot be safely written to disk without data lost, " +
+	"avoid type that can't be serialized")
+
+// ExecuteConfig is the configuration for the ExecuteOperation function.
+type ExecuteConfig[IN, DEP any] struct {
+	retryConfig RetryConfig[IN, DEP]
+}
+
+type ExecuteOption[IN, DEP any] func(*ExecuteConfig[IN, DEP])
+
+type RetryConfig[IN, DEP any] struct {
+	// DisableRetry disables the retry mechanism if set to true.
+	DisableRetry bool
+	// InputHook is a function that returns an updated input before retrying the operation.
+	// The operation when retried will use the input returned by this function.
+	// This is useful for scenarios like updating the gas limit.
+	// This will be ignored if DisableRetry is set to true.
+	InputHook func(input IN, deps DEP) IN
+}
+
+// WithRetryConfig is an ExecuteOption that sets the retry configuration.
+func WithRetryConfig[IN, DEP any](config RetryConfig[IN, DEP]) ExecuteOption[IN, DEP] {
+	return func(c *ExecuteConfig[IN, DEP]) {
+		c.retryConfig = config
+	}
+}
+
+// ExecuteOperation executes an operation with the given input and dependencies.
+// Execution will return the previous successful execution result and skip execution if there was a
+// previous successful run found in the Reports.
+// If previous unsuccessful execution was found, the execution will not be skipped.
+//
+// Note:
+// Operations that were skipped will not be added to the reporter.
+//
+// Retry:
+// By default, it retries the operation up to 10 times with exponential backoff if it fails.
+// Use WithRetryConfig to customize the retry behavior.
+// To cancel the retry early, return an error with NewUnrecoverableError.
+//
+// Input & Output:
+// The input and output must be JSON serializable. If the input is not serializable, it will return an error.
+// To be serializable, the input and output must be json.marshalable, or it must implement json.Marshaler and json.Unmarshaler.
+// IsSerializable can be used to check if the input or output is serializable.
+func ExecuteOperation[IN, OUT, DEP any](
+	b Bundle,
+	operation *Operation[IN, OUT, DEP],
+	deps DEP,
+	input IN,
+	opts ...ExecuteOption[IN, DEP],
+) (Report[IN, OUT], error) {
+	if !IsSerializable(b.Logger, input) {
+		return Report[IN, OUT]{}, fmt.Errorf("operation %s input: %w", operation.def.ID, ErrNotSerializable)
+	}
+
+	if previousReport, found := loadPreviousSuccessfulReport[IN, OUT](b, operation.def, input); found {
+		b.Logger.Infow("Operation already executed. Returning previous result", "id", operation.def.ID,
+			"version", operation.def.Version, "description", operation.def.Description)
+		return previousReport, nil
+	}
+
+	executeConfig := &ExecuteConfig[IN, DEP]{retryConfig: RetryConfig[IN, DEP]{}}
+	for _, opt := range opts {
+		opt(executeConfig)
+	}
+
+	var output OUT
+	var err error
+
+	if executeConfig.retryConfig.DisableRetry {
+		output, err = operation.execute(b, deps, input)
+	} else {
+		var inputTemp = input
+		output, err = retry.DoWithData(func() (OUT, error) {
+			return operation.execute(b, deps, inputTemp)
+		}, retry.OnRetry(func(attempt uint, err error) {
+			b.Logger.Infow("Operation failed. Retrying...",
+				"operation", operation.def.ID, "attempt", attempt, "error", err)
+
+			if executeConfig.retryConfig.InputHook != nil {
+				inputTemp = executeConfig.retryConfig.InputHook(inputTemp, deps)
+			}
+		}))
+	}
+
+	if err == nil && !IsSerializable(b.Logger, output) {
+		return Report[IN, OUT]{}, fmt.Errorf("operation %s output: %w", operation.def.ID, ErrNotSerializable)
+	}
+
+	report := NewReport(operation.def, input, output, err)
+	err = b.reporter.AddReport(genericReport(report))
+	if err != nil {
+		return Report[IN, OUT]{}, err
+	}
+	if report.Err != nil {
+		return report, report.Err
+	}
+	return report, nil
+}
+
+// ExecuteSequence executes a Sequence and returns a SequenceReport.
+// The SequenceReport contains a report for the Sequence and also the execution reports which are all
+// the operations that were executed as part of this sequence.
+// The latter is useful when we want to return all the executed reports to the changeset output.
+// Execution will return the previous successful execution result and skip execution if there was a
+// previous successful run found in the Reports.
+// If previous unsuccessful execution was found, the execution will not be skipped.
+//
+// Note:
+// Sequences or Operations that were skipped will not be added to the reporter.
+// The ExecutionReports do not include Sequences or Operations that were skipped.
+//
+// Input & Output:
+// The input and output must be JSON serializable. If the input is not serializable, it will return an error.
+// To be serializable, the input and output must be json.marshalable, or it must implement json.Marshaler and json.Unmarshaler.
+// IsSerializable can be used to check if the input or output is serializable.
+func ExecuteSequence[IN, OUT, DEP any](
+	b Bundle, sequence *Sequence[IN, OUT, DEP], deps DEP, input IN,
+) (SequenceReport[IN, OUT], error) {
+	if !IsSerializable(b.Logger, input) {
+		return SequenceReport[IN, OUT]{}, fmt.Errorf("sequence %s input: %w", sequence.def.ID, ErrNotSerializable)
+	}
+
+	if previousReport, found := loadPreviousSuccessfulReport[IN, OUT](b, sequence.def, input); found {
+		executionReports, err := b.reporter.GetExecutionReports(previousReport.ID)
+		if err != nil {
+			return SequenceReport[IN, OUT]{}, err
+		}
+		b.Logger.Infow("Sequence already executed. Returning previous result", "id", sequence.def.ID,
+			"version", sequence.def.Version, "description", sequence.def.Description)
+		return SequenceReport[IN, OUT]{previousReport, executionReports}, nil
+	}
+
+	b.Logger.Infow("Executing sequence", "id", sequence.def.ID,
+		"version", sequence.def.Version, "description", sequence.def.Description)
+	recentReporter := NewRecentMemoryReporter(b.reporter)
+	newBundle := Bundle{
+		Logger:          b.Logger,
+		GetContext:      b.GetContext,
+		reporter:        recentReporter,
+		reportHashCache: b.reportHashCache,
+	}
+	ret, err := sequence.handler(newBundle, deps, input)
+	if errors.Is(err, ErrNotSerializable) {
+		return SequenceReport[IN, OUT]{}, err
+	}
+
+	if err == nil && !IsSerializable(b.Logger, ret) {
+		return SequenceReport[IN, OUT]{}, fmt.Errorf("sequence %s output: %w", sequence.def.ID, ErrNotSerializable)
+	}
+
+	recentReports := recentReporter.GetRecentReports()
+	childReports := make([]string, 0, len(recentReports))
+	for _, rep := range recentReports {
+		childReports = append(childReports, rep.ID)
+	}
+
+	report := NewReport(
+		sequence.def,
+		input,
+		ret,
+		err,
+		childReports...,
+	)
+
+	err = b.reporter.AddReport(genericReport(report))
+	if err != nil {
+		return SequenceReport[IN, OUT]{}, err
+	}
+	executionReports, err := b.reporter.GetExecutionReports(report.ID)
+	if err != nil {
+		return SequenceReport[IN, OUT]{}, err
+	}
+	if report.Err != nil {
+		return SequenceReport[IN, OUT]{report, executionReports}, report.Err
+	}
+	return SequenceReport[IN, OUT]{report, executionReports}, nil
+}
+
+// NewUnrecoverableError creates an error that indicates an unrecoverable error.
+// If this error is returned inside an operation, the operation will no longer retry.
+// This allows the operation to fail fast if it encounters an unrecoverable error.
+func NewUnrecoverableError(err error) error {
+	return retry.Unrecoverable(err)
+}
+
+func loadPreviousSuccessfulReport[IN, OUT any](
+	b Bundle, def Definition, input IN,
+) (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)
+	if err != nil {
+		b.Logger.Errorw("Failed to construct unique hash", "error", err)
+		return Report[IN, OUT]{}, false
+	}
+
+	for _, report := range prevReports {
+		// Check if operation/sequence was run previously and return the report if successful
+		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
+		}
+		if reportHash == currentHash && report.Err == nil {
+			typedReport, ok := typeReport[IN, OUT](report)
+			if !ok {
+				b.Logger.Debugw(fmt.Sprintf("Previous %s execution found but couldn't find its matching Report", def.ID), "report_id", report.ID)
+				continue
+			}
+			b.Logger.Debugw(fmt.Sprintf("Previous %s execution found. Returning its result from Report storage", def.ID), "report_id", report.ID)
+			return typedReport, true
+		}
+	}
+	// No previous execution was found
+	return Report[IN, OUT]{}, false
+}