Merge branch 'develop' into re-3199/updater-root-module-go.mod-ref

smartcontractkit · Dec 10, 2024 · 45efc1a · 45efc1a
2 parents 737c8db + 4fa0013
commit 45efc1a
Show file tree

Hide file tree

Showing 41 changed files with 852 additions and 730 deletions.
diff --git a/.github/e2e-tests.yml b/.github/e2e-tests.yml
@@ -947,6 +947,7 @@ runner-test-matrix:
     test_env_vars:
       E2E_TEST_SELECTED_NETWORK: SIMULATED_1,SIMULATED_2
       E2E_JD_VERSION: 0.6.0
+      CCIP_V16_TEST_ENV: docker
 
   - id: smoke/ccip/ccip_token_price_updates_test.go:*
     path: integration-tests/smoke/ccip/ccip_token_price_updates_test.go
@@ -960,6 +961,7 @@ runner-test-matrix:
     test_env_vars:
       E2E_TEST_SELECTED_NETWORK: SIMULATED_1,SIMULATED_2
       E2E_JD_VERSION: 0.6.0
+      CCIP_V16_TEST_ENV: docker
 
   - id: smoke/ccip/ccip_gas_price_updates_test.go:*
     path: integration-tests/smoke/ccip/ccip_gas_price_updates_test.go
@@ -973,6 +975,7 @@ runner-test-matrix:
     test_env_vars:
       E2E_TEST_SELECTED_NETWORK: SIMULATED_1,SIMULATED_2
       E2E_JD_VERSION: 0.6.0
+      CCIP_V16_TEST_ENV: docker
 
   - id: smoke/ccip/ccip_rmn_test.go:^TestRMN_TwoMessagesOnTwoLanesIncludingBatching$
     path: integration-tests/smoke/ccip/ccip_rmn_test.go
@@ -988,6 +991,7 @@ runner-test-matrix:
       E2E_JD_VERSION: 0.6.0
       E2E_RMN_RAGEPROXY_VERSION: master-f461a9e
       E2E_RMN_AFN2PROXY_VERSION: master-f461a9e
+      CCIP_V16_TEST_ENV: docker
 
   - id: smoke/ccip/ccip_rmn_test.go:^TestRMN_MultipleMessagesOnOneLaneNoWaitForExec$
     path: integration-tests/smoke/ccip/ccip_rmn_test.go
@@ -1003,6 +1007,7 @@ runner-test-matrix:
       E2E_JD_VERSION: 0.6.0
       E2E_RMN_RAGEPROXY_VERSION: master-f461a9e
       E2E_RMN_AFN2PROXY_VERSION: master-f461a9e
+      CCIP_V16_TEST_ENV: docker
 
   - id: smoke/ccip/ccip_rmn_test.go:^TestRMN_NotEnoughObservers$
     path: integration-tests/smoke/ccip/ccip_rmn_test.go
@@ -1018,6 +1023,7 @@ runner-test-matrix:
       E2E_JD_VERSION: 0.6.0
       E2E_RMN_RAGEPROXY_VERSION: master-f461a9e
       E2E_RMN_AFN2PROXY_VERSION: master-f461a9e
+      CCIP_V16_TEST_ENV: docker
 
   - id: smoke/ccip/ccip_rmn_test.go:^TestRMN_DifferentSigners$
     path: integration-tests/smoke/ccip/ccip_rmn_test.go
@@ -1033,6 +1039,7 @@ runner-test-matrix:
       E2E_JD_VERSION: 0.6.0
       E2E_RMN_RAGEPROXY_VERSION: master-f461a9e
       E2E_RMN_AFN2PROXY_VERSION: master-f461a9e
+      CCIP_V16_TEST_ENV: docker
 
   - id: smoke/ccip/ccip_rmn_test.go:^TestRMN_NotEnoughSigners$
     path: integration-tests/smoke/ccip/ccip_rmn_test.go
@@ -1048,6 +1055,7 @@ runner-test-matrix:
       E2E_JD_VERSION: 0.6.0
       E2E_RMN_RAGEPROXY_VERSION: master-f461a9e
       E2E_RMN_AFN2PROXY_VERSION: master-f461a9e
+      CCIP_V16_TEST_ENV: docker
 
   - id: smoke/ccip/ccip_rmn_test.go:^TestRMN_DifferentRmnNodesForDifferentChains$
     path: integration-tests/smoke/ccip/ccip_rmn_test.go
@@ -1063,6 +1071,7 @@ runner-test-matrix:
       E2E_JD_VERSION: 0.6.0
       E2E_RMN_RAGEPROXY_VERSION: master-f461a9e
       E2E_RMN_AFN2PROXY_VERSION: master-f461a9e
+      CCIP_V16_TEST_ENV: docker
 
   - id: smoke/ccip/ccip_rmn_test.go:^TestRMN_TwoMessagesOneSourceChainCursed$
     path: integration-tests/smoke/ccip/ccip_rmn_test.go
@@ -1078,6 +1087,7 @@ runner-test-matrix:
       E2E_JD_VERSION: 0.6.0
       E2E_RMN_RAGEPROXY_VERSION: master-f461a9e
       E2E_RMN_AFN2PROXY_VERSION: master-f461a9e
+      CCIP_V16_TEST_ENV: docker
 
   - id: smoke/ccip/ccip_rmn_test.go:^TestRMN_GlobalCurseTwoMessagesOnTwoLanes$
     path: integration-tests/smoke/ccip/ccip_rmn_test.go
@@ -1093,6 +1103,7 @@ runner-test-matrix:
       E2E_JD_VERSION: 0.6.0
       E2E_RMN_RAGEPROXY_VERSION: master-f461a9e
       E2E_RMN_AFN2PROXY_VERSION: master-f461a9e
+      CCIP_V16_TEST_ENV: docker
 
   # END: CCIPv1.6 tests
 

diff --git a/core/chains/evm/logpoller/helper_test.go b/core/chains/evm/logpoller/helper_test.go
@@ -118,11 +118,19 @@ func SetupTH(t testing.TB, opts logpoller.Opts) TestHarness {
 		opts.PollPeriod = 1 * time.Hour
 	}
 	lp := logpoller.NewLogPoller(o, esc, lggr, headTracker, opts)
+
+	pendingNonce, err := backend.Client().PendingNonceAt(testutils.Context(t), owner.From)
+	require.NoError(t, err)
+
+	owner.Nonce = big.NewInt(0).SetUint64(pendingNonce)
 	emitterAddress1, _, emitter1, err := log_emitter.DeployLogEmitter(owner, backend.Client())
 	require.NoError(t, err)
+
+	owner.Nonce.Add(owner.Nonce, big.NewInt(1)) // Avoid race where DeployLogEmitter returns before PendingNonce has been incremented
 	emitterAddress2, _, emitter2, err := log_emitter.DeployLogEmitter(owner, backend.Client())
 	require.NoError(t, err)
 	backend.Commit()
+	owner.Nonce = nil // Just use pending nonce after this
 
 	return TestHarness{
 		Lggr:            lggr,

diff --git a/core/scripts/go.mod b/core/scripts/go.mod
@@ -33,7 +33,7 @@ require (
 	github.com/prometheus/client_golang v1.20.5
 	github.com/shopspring/decimal v1.4.0
 	github.com/smartcontractkit/chainlink-automation v0.8.1
-	github.com/smartcontractkit/chainlink-common v0.3.1-0.20241209151352-70300ddcc776
+	github.com/smartcontractkit/chainlink-common v0.3.1-0.20241210192653-a9c706f99e83
 	github.com/smartcontractkit/libocr v0.0.0-20241007185508-adbe57025f12
 	github.com/spf13/cobra v1.8.1
 	github.com/spf13/viper v1.19.0
@@ -303,7 +303,7 @@ require (
 	github.com/shirou/gopsutil/v3 v3.24.3 // indirect
 	github.com/smartcontractkit/ccip-owner-contracts v0.0.0-20240926212305-a6deabdfce86 // indirect
 	github.com/smartcontractkit/chain-selectors v1.0.34 // indirect
-	github.com/smartcontractkit/chainlink-ccip v0.0.0-20241204015713-8956bb614e9e // indirect
+	github.com/smartcontractkit/chainlink-ccip v0.0.0-20241210163448-e683c0b91a44 // indirect
 	github.com/smartcontractkit/chainlink-cosmos v0.5.2-0.20241202195413-82468150ac1e // indirect
 	github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20241202141438-a90db35252db // indirect
 	github.com/smartcontractkit/chainlink-feeds v0.1.1 // indirect

diff --git a/core/scripts/go.sum b/core/scripts/go.sum
@@ -1140,10 +1140,10 @@ github.com/smartcontractkit/chain-selectors v1.0.34 h1:MJ17OGu8+jjl426pcKrJkCf3f
 github.com/smartcontractkit/chain-selectors v1.0.34/go.mod h1:xsKM0aN3YGcQKTPRPDDtPx2l4mlTN1Djmg0VVXV40b8=
 github.com/smartcontractkit/chainlink-automation v0.8.1 h1:sTc9LKpBvcKPc1JDYAmgBc2xpDKBco/Q4h4ydl6+UUU=
 github.com/smartcontractkit/chainlink-automation v0.8.1/go.mod h1:Iij36PvWZ6blrdC5A/nrQUBuf3MH3JvsBB9sSyc9W08=
-github.com/smartcontractkit/chainlink-ccip v0.0.0-20241204015713-8956bb614e9e h1:GnM6ZWV6vlk2+n6c6o+v/R1LtXzBGVVx7r37nt/h6Uc=
-github.com/smartcontractkit/chainlink-ccip v0.0.0-20241204015713-8956bb614e9e/go.mod h1:80vGBbOfertJig0xFKsRfm+i17FkjdKkk1dAaGE45Os=
-github.com/smartcontractkit/chainlink-common v0.3.1-0.20241209151352-70300ddcc776 h1:NATQA1LfrEPXCdtEed9/G4SxaVuF8EZp5O2ucOK5C98=
-github.com/smartcontractkit/chainlink-common v0.3.1-0.20241209151352-70300ddcc776/go.mod h1:bQktEJf7sJ0U3SmIcXvbGUox7SmXcnSEZ4kUbT8R5Nk=
+github.com/smartcontractkit/chainlink-ccip v0.0.0-20241210163448-e683c0b91a44 h1:+3Uc4x1tDFCddjhmgkphDqWr1N+mzP7NQbXD8Bby6Ck=
+github.com/smartcontractkit/chainlink-ccip v0.0.0-20241210163448-e683c0b91a44/go.mod h1:F8xQAIW0ymb2BZhqn89sWZLXreJhM5KDVF6Qb4y44N0=
+github.com/smartcontractkit/chainlink-common v0.3.1-0.20241210192653-a9c706f99e83 h1:NjrU7KOn3Tk+C6QFo9tQBqeotPKytpBwhn/J1s+yiiY=
+github.com/smartcontractkit/chainlink-common v0.3.1-0.20241210192653-a9c706f99e83/go.mod h1:bQktEJf7sJ0U3SmIcXvbGUox7SmXcnSEZ4kUbT8R5Nk=
 github.com/smartcontractkit/chainlink-cosmos v0.5.2-0.20241202195413-82468150ac1e h1:PRoeby6ZlTuTkv2f+7tVU4+zboTfRzI+beECynF4JQ0=
 github.com/smartcontractkit/chainlink-cosmos v0.5.2-0.20241202195413-82468150ac1e/go.mod h1:mUh5/woemsVaHgTorA080hrYmO3syBCmPdnWc/5dOqk=
 github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20241202141438-a90db35252db h1:N1RH1hSr2ACzOFc9hkCcjE8pRBTdcU3p8nsTJByaLes=

diff --git a/core/services/workflows/syncer/handler.go b/core/services/workflows/syncer/handler.go
@@ -438,6 +438,11 @@ func (h *eventHandler) workflowRegisteredEvent(
 		return fmt.Errorf("workflowID mismatch: %x != %x", hash, payload.WorkflowID)
 	}
 
+	// Ensure that there is no running workflow engine for the given workflow ID.
+	if h.engineRegistry.IsRunning(hex.EncodeToString(payload.WorkflowID[:])) {
+		return fmt.Errorf("workflow is already running, so not starting it : %s", hex.EncodeToString(payload.WorkflowID[:]))
+	}
+
 	// Save the workflow secrets
 	urlHash, err := h.orm.GetSecretsURLHash(payload.WorkflowOwner, []byte(payload.SecretsURL))
 	if err != nil {

diff --git a/core/services/workflows/syncer/handler_test.go b/core/services/workflows/syncer/handler_test.go
@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/custmsg"
+	"github.com/smartcontractkit/chainlink-common/pkg/services"
 	pkgworkflows "github.com/smartcontractkit/chainlink-common/pkg/workflows"
 	"github.com/smartcontractkit/chainlink-common/pkg/workflows/secrets"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities"
@@ -47,6 +48,28 @@ func newMockFetcher(m map[string]mockFetchResp) FetcherFunc {
 	return (&mockFetcher{responseMap: m}).Fetch
 }
 
+type mockEngine struct {
+	CloseErr error
+	ReadyErr error
+	StartErr error
+}
+
+func (m *mockEngine) Ready() error {
+	return m.ReadyErr
+}
+
+func (m *mockEngine) Close() error {
+	return m.CloseErr
+}
+
+func (m *mockEngine) Start(_ context.Context) error {
+	return m.StartErr
+}
+
+func (m *mockEngine) HealthReport() map[string]error { return nil }
+
+func (m *mockEngine) Name() string { return "mockEngine" }
+
 func Test_Handler(t *testing.T) {
 	lggr := logger.TestLogger(t)
 	emitter := custmsg.NewLabeler()
@@ -181,7 +204,11 @@ func Test_workflowRegisteredHandler(t *testing.T) {
 	var wfOwner = []byte("0xOwner")
 	var binary = wasmtest.CreateTestBinary(binaryCmd, binaryLocation, true, t)
 	var encodedBinary = []byte(base64.StdEncoding.EncodeToString(binary))
-	defaultValidationFn := func(t *testing.T, ctx context.Context, h *eventHandler, wfOwner []byte, wfName string, wfID string) {
+
+	defaultValidationFn := func(t *testing.T, ctx context.Context, event WorkflowRegistryWorkflowRegisteredV1, h *eventHandler, wfOwner []byte, wfName string, wfID string) {
+		err := h.workflowRegisteredEvent(ctx, event)
+		require.NoError(t, err)
+
 		// Verify the record is updated in the database
 		dbSpec, err := h.orm.GetWorkflowSpec(ctx, hex.EncodeToString(wfOwner), "workflow-name")
 		require.NoError(t, err)
@@ -204,6 +231,9 @@ func Test_workflowRegisteredHandler(t *testing.T) {
 				configURL:  {Body: config, Err: nil},
 				secretsURL: {Body: []byte("secrets"), Err: nil},
 			}),
+			engineFactoryFn: func(ctx context.Context, wfid string, owner string, name string, config []byte, binary []byte) (services.Service, error) {
+				return &mockEngine{}, nil
+			},
 			GiveConfig: config,
 			ConfigURL:  configURL,
 			SecretsURL: secretsURL,
@@ -223,6 +253,71 @@ func Test_workflowRegisteredHandler(t *testing.T) {
 			},
 			validationFn: defaultValidationFn,
 		},
+		{
+			Name: "fails to start engine",
+			fetcher: newMockFetcher(map[string]mockFetchResp{
+				binaryURL:  {Body: encodedBinary, Err: nil},
+				configURL:  {Body: config, Err: nil},
+				secretsURL: {Body: []byte("secrets"), Err: nil},
+			}),
+			engineFactoryFn: func(ctx context.Context, wfid string, owner string, name string, config []byte, binary []byte) (services.Service, error) {
+				return &mockEngine{StartErr: assert.AnError}, nil
+			},
+			GiveConfig: config,
+			ConfigURL:  configURL,
+			SecretsURL: secretsURL,
+			BinaryURL:  binaryURL,
+			GiveBinary: binary,
+			WFOwner:    wfOwner,
+			Event: func(wfID []byte) WorkflowRegistryWorkflowRegisteredV1 {
+				return WorkflowRegistryWorkflowRegisteredV1{
+					Status:        uint8(0),
+					WorkflowID:    [32]byte(wfID),
+					WorkflowOwner: wfOwner,
+					WorkflowName:  "workflow-name",
+					BinaryURL:     binaryURL,
+					ConfigURL:     configURL,
+					SecretsURL:    secretsURL,
+				}
+			},
+			validationFn: func(t *testing.T, ctx context.Context, event WorkflowRegistryWorkflowRegisteredV1, h *eventHandler, wfOwner []byte, wfName string, wfID string) {
+				err := h.workflowRegisteredEvent(ctx, event)
+				require.Error(t, err)
+				require.ErrorIs(t, err, assert.AnError)
+			},
+		},
+		{
+			Name: "fails if running engine exists",
+			fetcher: newMockFetcher(map[string]mockFetchResp{
+				binaryURL:  {Body: encodedBinary, Err: nil},
+				configURL:  {Body: config, Err: nil},
+				secretsURL: {Body: []byte("secrets"), Err: nil},
+			}),
+			GiveConfig: config,
+			ConfigURL:  configURL,
+			SecretsURL: secretsURL,
+			BinaryURL:  binaryURL,
+			GiveBinary: binary,
+			WFOwner:    wfOwner,
+			Event: func(wfID []byte) WorkflowRegistryWorkflowRegisteredV1 {
+				return WorkflowRegistryWorkflowRegisteredV1{
+					Status:        uint8(0),
+					WorkflowID:    [32]byte(wfID),
+					WorkflowOwner: wfOwner,
+					WorkflowName:  "workflow-name",
+					BinaryURL:     binaryURL,
+					ConfigURL:     configURL,
+					SecretsURL:    secretsURL,
+				}
+			},
+			validationFn: func(t *testing.T, ctx context.Context, event WorkflowRegistryWorkflowRegisteredV1, h *eventHandler, wfOwner []byte, wfName string, wfID string) {
+				me := &mockEngine{}
+				h.engineRegistry.Add(wfID, me)
+				err := h.workflowRegisteredEvent(ctx, event)
+				require.Error(t, err)
+				require.ErrorContains(t, err, "workflow is already running")
+			},
+		},
 		{
 			Name: "success with paused workflow registered",
 			fetcher: newMockFetcher(map[string]mockFetchResp{
@@ -247,7 +342,10 @@ func Test_workflowRegisteredHandler(t *testing.T) {
 					SecretsURL:    secretsURL,
 				}
 			},
-			validationFn: func(t *testing.T, ctx context.Context, h *eventHandler, wfOwner []byte, wfName string, wfID string) {
+			validationFn: func(t *testing.T, ctx context.Context, event WorkflowRegistryWorkflowRegisteredV1, h *eventHandler, wfOwner []byte, wfName string, wfID string) {
+				err := h.workflowRegisteredEvent(ctx, event)
+				require.NoError(t, err)
+
 				// Verify the record is updated in the database
 				dbSpec, err := h.orm.GetWorkflowSpec(ctx, hex.EncodeToString(wfOwner), "workflow-name")
 				require.NoError(t, err)
@@ -315,62 +413,58 @@ func Test_workflowRegisteredHandler(t *testing.T) {
 }
 
 type testCase struct {
-	Name         string
-	SecretsURL   string
-	BinaryURL    string
-	GiveBinary   []byte
-	GiveConfig   []byte
-	ConfigURL    string
-	WFOwner      []byte
-	fetcher      FetcherFunc
-	Event        func([]byte) WorkflowRegistryWorkflowRegisteredV1
-	validationFn func(t *testing.T, ctx context.Context, h *eventHandler, wfOwner []byte, wfName string, wfID string)
+	Name            string
+	SecretsURL      string
+	BinaryURL       string
+	GiveBinary      []byte
+	GiveConfig      []byte
+	ConfigURL       string
+	WFOwner         []byte
+	fetcher         FetcherFunc
+	Event           func([]byte) WorkflowRegistryWorkflowRegisteredV1
+	validationFn    func(t *testing.T, ctx context.Context, event WorkflowRegistryWorkflowRegisteredV1, h *eventHandler, wfOwner []byte, wfName string, wfID string)
+	engineFactoryFn func(ctx context.Context, wfid string, owner string, name string, config []byte, binary []byte) (services.Service, error)
 }
 
-func testRunningWorkflow(t *testing.T, cmd testCase) {
+func testRunningWorkflow(t *testing.T, tc testCase) {
 	t.Helper()
-	t.Run(cmd.Name, func(t *testing.T) {
+	t.Run(tc.Name, func(t *testing.T) {
 		var (
 			ctx     = testutils.Context(t)
 			lggr    = logger.TestLogger(t)
 			db      = pgtest.NewSqlxDB(t)
 			orm     = NewWorkflowRegistryDS(db, lggr)
 			emitter = custmsg.NewLabeler()
 
-			binary     = cmd.GiveBinary
-			config     = cmd.GiveConfig
-			secretsURL = cmd.SecretsURL
-			wfOwner    = cmd.WFOwner
+			binary     = tc.GiveBinary
+			config     = tc.GiveConfig
+			secretsURL = tc.SecretsURL
+			wfOwner    = tc.WFOwner
 
-			fetcher = cmd.fetcher
+			fetcher = tc.fetcher
 		)
 
 		giveWFID, err := pkgworkflows.GenerateWorkflowID(wfOwner, binary, config, secretsURL)
 		require.NoError(t, err)
 
 		wfID := hex.EncodeToString(giveWFID[:])
 
-		event := cmd.Event(giveWFID[:])
+		event := tc.Event(giveWFID[:])
 
 		er := NewEngineRegistry()
+		opts := []func(*eventHandler){
+			WithEngineRegistry(er),
+		}
+		if tc.engineFactoryFn != nil {
+			opts = append(opts, WithEngineFactoryFn(tc.engineFactoryFn))
+		}
 		store := wfstore.NewDBStore(db, lggr, clockwork.NewFakeClock())
 		registry := capabilities.NewRegistry(lggr)
 		registry.SetLocalRegistry(&capabilities.TestMetadataRegistry{})
-		h := NewEventHandler(
-			lggr,
-			orm,
-			fetcher,
-			store,
-			registry,
-			emitter,
-			clockwork.NewFakeClock(),
-			workflowkey.Key{},
-			WithEngineRegistry(er),
-		)
-		err = h.workflowRegisteredEvent(ctx, event)
-		require.NoError(t, err)
+		h := NewEventHandler(lggr, orm, fetcher, store, registry, emitter, clockwork.NewFakeClock(),
+			workflowkey.Key{}, opts...)
 
-		cmd.validationFn(t, ctx, h, wfOwner, "workflow-name", wfID)
+		tc.validationFn(t, ctx, event, h, wfOwner, "workflow-name", wfID)
 	})
 }