[DO NOT MERGE] Simulate failed pipeline state

.github/workflows/go.yml

@@ -79,7 +79,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        plan: ["simplan_fast_check", "only_log_trigger"]
+        plan: ["simplan_fast_check", "only_log_trigger", "pipeline_failure_recovery"]
 
     name: Upkeep Simulation ${{ matrix.plan }}
 

tools/simulator/config/event.go

@@ -92,6 +92,27 @@ type GenerateUpkeepEvent struct {
 	// not perform due to some other network concerns such as too high network
 	// delay or something that might disable the OCR3 protocol.
 	Expected string `json:"expected,omitempty"`
+	// Overrides provides customizations on upkeeps including check pipeline behavior.
+	// Only one configuration per upkeep is supported.
+	Overrides []UpkeepOverrides `json:"overrides,omitempty"`
+}
+
+type UpkeepOverrides struct {
+	// UpkeepID references the upkeep id generated from StartID + Count. This
+	// value is NOT the final upkeep id referenced by the plugin.
+	UpkeepID *big.Int `json:"id"`
+	// Pattern defines the pipeline return behavior. A '1' indicates a failure
+	// and a '0' indicates a non-failure. The pattern is structured like a count
+	// where each pipeline run on the defined upkeep increments the position in
+	// the pattern. At the end of the pattern, it repeats from the beginning.
+	FailurePattern string `json:"pipelineFailurePattern,omitempty"`
+	// Retryable indicates whether or not the retryable flag should be set on the
+	// pipeline result. Setting this will always set the retryable flag for the
+	// provided upkeep.
+	RetryableOnFailure bool `json:"retryableOnFailure"`
+	// Expected overrides the container expectation configuration in the case that
+	// specific upkeep configurations make the upkeep fail.
+	Expected bool `json:"expected"`
 }
 
 // LogTriggerEvent is a configuration for simulating logs emitted from a chain

tools/simulator/plans/pipeline_failure_recovery.json

@@ -0,0 +1,111 @@
+{
+    "node": {
+        "totalNodeCount": 4,
+        "maxNodeServiceWorkers": 100,
+        "maxNodeServiceQueueSize": 1000
+    },
+    "p2pNetwork": {
+        "maxLatency": "100ms"
+    },
+    "rpc": {
+        "maxBlockDelay": 600,
+        "averageLatency": 300,
+        "errorRate": 0.02,
+        "rateLimitThreshold": 1000
+    },
+    "blocks": {
+        "genesisBlock": 128943862,
+        "blockCadence": "1s",
+        "durationInBlocks": 60,
+        "endPadding": 20
+    },
+    "events": [
+        {
+            "type": "ocr3config",
+            "eventBlockNumber": 128943863,
+            "comment": "initial ocr config (valid)",
+            "maxFaultyNodes": 1,
+            "encodedOffchainConfig": "{\"version\":\"v3\",\"performLockoutWindow\":100000,\"targetProbability\":\"0.999\",\"targetInRounds\":4,\"minConfirmations\":1,\"gasLimitPerReport\":1000000,\"gasOverheadPerUpkeep\":300000,\"maxUpkeepBatchSize\":10}",
+            "maxRoundsPerEpoch": 7,
+            "deltaProgress": "10s",
+            "deltaResend": "10s",
+            "deltaInitial": "300ms",
+            "deltaRound": "1100ms",
+            "deltaGrace": "300ms",
+            "deltaCertifiedCommitRequest": "200ms",
+            "deltaStage": "20s",
+            "maxQueryTime": "50ms",
+            "maxObservationTime": "100ms",
+            "maxShouldAcceptTime": "50ms",
+            "maxShouldTransmitTime": "50ms"
+        },
+        {
+            "type": "generateUpkeeps",
+            "eventBlockNumber": 128943862,
+            "comment": "single log triggered upkeep",
+            "count": 1,
+            "startID": 300,
+            "eligibilityFunc": "always",
+            "upkeepType": "logTrigger",
+            "logTriggeredBy": "test_trigger_event"
+        },
+        {
+            "type": "generateUpkeeps",
+            "eventBlockNumber": 128943864,
+            "comment": "5 log triggered upkeeps",
+            "count": 5,
+            "startID": 400,
+            "eligibilityFunc": "always",
+            "upkeepType": "logTrigger",
+            "logTriggeredBy": "test_trigger_event",
+            "overrides": [
+                {
+                    "id": 401,
+                    "pipelineFailurePattern": "111000000",
+                    "retryableOnFailure": true,
+                    "expected": true
+                },
+                {
+                    "id": 402,
+                    "pipelineFailurePattern": "1",
+                    "retryableOnFailure": true,
+                    "expected": false
+                },
+                {
+                    "id": 403,
+                    "pipelineFailurePattern": "1000100",
+                    "retryableOnFailure": true,
+                    "expected": true
+                }
+            ]
+        },
+        {
+            "type": "logTrigger",
+            "eventBlockNumber": 128943872,
+            "comment": "trigger 10 blocks after trigger upkeep created",
+            "triggerValue": "test_trigger_event"
+        },
+        {
+            "type": "generateUpkeeps",
+            "eventBlockNumber": 128943878,
+            "comment": "7 log triggered upkeeps",
+            "count": 7,
+            "startID": 500,
+            "eligibilityFunc": "always",
+            "upkeepType": "logTrigger",
+            "logTriggeredBy": "test_trigger_event"
+        },
+        {
+            "type": "logTrigger",
+            "eventBlockNumber": 128943882,
+            "comment": "trigger 10 blocks after trigger upkeep created",
+            "triggerValue": "test_trigger_event"
+        },
+        {
+            "type": "logTrigger",
+            "eventBlockNumber": 128943892,
+            "comment": "trigger 10 blocks after trigger upkeep created",
+            "triggerValue": "test_trigger_event"
+        }
+    ]
+}

tools/simulator/simulate/chain/block.go

@@ -1,7 +1,10 @@
 package chain
 
 import (
+	"fmt"
 	"math/big"
+	"strconv"
+	"sync"
 
 	"github.com/smartcontractkit/libocr/offchainreporting2plus/types"
 )
@@ -50,6 +53,54 @@ type SimulatedUpkeep struct {
 	TriggeredBy    string
 	CheckData      []byte
 	Expected       bool
+	Retryable      bool
+	States         *CheckPipelineStateManager
+}
+
+type CheckPipelineStateManager struct {
+	nextPosition int
+	states       []int
+	mu           sync.Mutex
+}
+
+func NewCheckPipelineStateManager(pattern string) (*CheckPipelineStateManager, error) {
+	states := make([]int, 0, len(pattern))
+
+	for _, rne := range pattern {
+		flag, err := strconv.Atoi(string(rne))
+		if err != nil {
+			return nil, err
+		}
+
+		if flag > 1 || flag < 0 {
+			return nil, fmt.Errorf("only 0 and 1 allowed")
+		}
+
+		states = append(states, flag)
+	}
+
+	return &CheckPipelineStateManager{
+		states: states,
+	}, nil
+}
+
+func (m *CheckPipelineStateManager) GetNextState() int {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if len(m.states) == 0 {
+		return 0
+	}
+
+	nextState := m.states[m.nextPosition]
+
+	m.nextPosition++
+
+	if m.nextPosition >= len(m.states) {
+		m.nextPosition = 0
+	}
+
+	return nextState
 }
 
 type SimulatedLog struct {

tools/simulator/simulate/chain/block_test.go

@@ -0,0 +1,32 @@
+package chain_test
+
+import (
+	"testing"
+
+	"github.com/smartcontractkit/chainlink-automation/tools/simulator/simulate/chain"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestGetNextState(t *testing.T) {
+	tests := []struct {
+		pattern string
+		err     error
+		states  []int
+	}{
+		{"0", nil, []int{0, 0, 0}},
+		{"1", nil, []int{1, 1, 1}},
+		{"", nil, []int{0, 0, 0}},
+		{"0001000101", nil, []int{0, 0, 0, 1, 0, 0, 0, 1, 0, 1, 0, 0, 0, 1}},
+	}
+
+	for _, test := range tests {
+		manager, err := chain.NewCheckPipelineStateManager(test.pattern)
+
+		require.ErrorIs(t, err, test.err)
+
+		for idx, expected := range test.states {
+			assert.Equal(t, expected, manager.GetNextState(), "state should match for %d", idx)
+		}
+	}
+}

tools/simulator/simulate/chain/generate.go

@@ -75,6 +75,28 @@ func generateBasicSimulatedUpkeeps(event config.GenerateUpkeepEvent, alwaysEligi
 
 	for y := 1; y <= event.Count; y++ {
 		id := new(big.Int).Add(event.StartID, big.NewInt(int64(y)))
+		retryable := true
+		expected := event.Expected == config.AllExpected
+
+		states, err := NewCheckPipelineStateManager("")
+		if err != nil {
+			return nil, err
+		}
+
+		for _, override := range event.Overrides {
+			if id.Cmp(override.UpkeepID) == 0 {
+				states, err = NewCheckPipelineStateManager(override.FailurePattern)
+				if err != nil {
+					return nil, err
+				}
+
+				retryable = override.RetryableOnFailure
+				expected = override.Expected
+
+				break
+			}
+		}
+
 		simulated := SimulatedUpkeep{
 			ID:             id,
 			Type:           simulationType,
@@ -83,7 +105,9 @@ func generateBasicSimulatedUpkeeps(event config.GenerateUpkeepEvent, alwaysEligi
 			AlwaysEligible: alwaysEligible,
 			EligibleAt:     make([]*big.Int, 0),
 			TriggeredBy:    event.LogTriggeredBy,
-			Expected:       event.Expected == config.AllExpected,
+			Expected:       expected,
+			Retryable:      retryable,
+			States:         states,
 		}
 
 		generated = append(generated, simulated)
@@ -113,7 +137,6 @@ func generateEligibilityFuncSimulatedUpkeeps(event config.GenerateUpkeepEvent, s
 			AlwaysEligible: false,
 			EligibleAt:     make([]*big.Int, 0),
 			TriggeredBy:    event.LogTriggeredBy,
-			Expected:       event.Expected == config.AllExpected,
 		}
 
 		var genesis *big.Int
@@ -134,6 +157,32 @@ func generateEligibilityFuncSimulatedUpkeeps(event config.GenerateUpkeepEvent, s
 			return nil, err
 		}
 
+		retryable := true
+		expected := event.Expected == config.AllExpected
+
+		states, err := NewCheckPipelineStateManager("")
+		if err != nil {
+			return nil, err
+		}
+
+		for _, override := range event.Overrides {
+			if id.Cmp(override.UpkeepID) == 0 {
+				states, err = NewCheckPipelineStateManager(override.FailurePattern)
+				if err != nil {
+					return nil, err
+				}
+
+				retryable = override.RetryableOnFailure
+				expected = override.Expected
+
+				break
+			}
+		}
+
+		sym.Retryable = retryable
+		sym.States = states
+		sym.Expected = expected
+
 		generated = append(generated, sym)
 	}