added smoke test & load test for log poller (#11110)

* added smoke test & load test for log poller

* read CL nodes logs in parallel and compare them in parallel with EVM node logs (has a big impact on execution, when we emit 100k+ logs)

* add simple config validation

* add smoke tests for backup process and replay

* run replay test for 15m instead of 5m (for debuggin)

* do not use hardcoded postgres values

* added support for chaos experiments (pausing containers) + a smoke test that uses them

* streamline log poller tests

* remove backup poller test -- way to test it reliably in e2e tests

* don't skip replay test

* add go.work* to .gitignore

* add tests that can easier run in CI + some changes after testing with live testnets

* wait for LP to finalise endblock + on demand workflow in GH

* rename on demand workflow

* fix typo in workflow name

.github/log_poller_on_demand.yml

@@ -0,0 +1,66 @@
+name: On Demand Log Poller Consistency Test
+on:
+  workflow_dispatch:
+    inputs:
+      contracts:
+        description: Number of test contracts
+        default: "2"
+        required: true
+      eventsPerTx:
+        description: Number of events to emit per transaction
+        default: "10"
+        required: true
+      useFinalityTag:
+        description: Use finality tag
+        default: "false"
+        required: true
+      loadDuration:
+        description: Load duration (e.g. 10s, 10m, 1h)
+        default: "10m"
+        required: true
+      chainlinkImage:
+        description: Chainlink image to use
+        default: "public.ecr.aws/chainlink/chainlink"
+        required: true
+      chainlinkVersion:
+        description: Chainlink version to use
+        default: "v2.7.0-beta0"
+        required: true
+      selectedNetworks:
+        description: Network to use (only Sepolia or Mumbai)
+        default: "Sepolia"
+        required: true
+      fundingKey:
+        description: Private key used to fund the contracts
+        required: true
+      rpcURL:
+        description: RPC URL to use
+        required: true
+      wsURL:
+        description: WS URL to use
+        required: true
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Setup Go
+        uses: actions/setup-go@v3
+        with:
+          go-version-file: "integration-tests/go.mod"
+          cache: true
+      - name: Show overrides
+        env:
+          CONTRACTS: ${{ inputs.contracts }}
+          EVENTS_PER_TX: ${{ inputs.eventsPerTx }}
+          LOAD_DURATION: ${{ inputs.loadDuration }}
+          USE_FINALITY_TAG: ${{ inputs.useFinalityTag }}
+          CHAINLINK_IMAGE: ${{ inputs.chainlinkImage }}
+          CHAINLINK_VERSION: ${{ inputs.chainlinkVersion }}
+          SELECTED_NETWORKS: ${{ inputs.selectedNetworks }}
+          EVM_KEYS: ${{ inputs.fundingKey }}
+          EVM_HTTP_URLS: ${{ inputs.rpcURL }}
+          EVM_URLS: ${{ inputs.wsURL }}
+        run: |
+          go test -v -timeout 5h -run=TestLogPollerFromEnv integration-tests/reorg/log_poller_maybe_reorg_test.go

.gitignore

@@ -65,7 +65,7 @@ tests-*.xml
 tmp-manifest-*.yaml
 ztarrepo.tar.gz
 **/test-ledger/*
-__debug_bin
+__debug_bin*
 
 # goreleaser builds
 cosign.*
@@ -82,3 +82,4 @@ contracts/yarn.lock
 
 # Ignore DevSpace cache and log folder
 .devspace/
+go.work*

core/chains/evm/logpoller/log_poller.go

@@ -466,6 +466,7 @@ func (lp *logPoller) run() {
 					// Serially process replay requests.
 					lp.lggr.Infow("Executing replay", "fromBlock", fromBlock, "requested", fromBlockReq)
 					lp.PollAndSaveLogs(lp.ctx, fromBlock)
+					lp.lggr.Infow("Executing replay finished", "fromBlock", fromBlock, "requested", fromBlockReq)
 				}
 			} else {
 				lp.lggr.Errorw("Error executing replay, could not get fromBlock", "err", err)
@@ -574,13 +575,14 @@ func (lp *logPoller) BackupPollAndSaveLogs(ctx context.Context, backupPollerBloc
 
 	lastSafeBackfillBlock := latestFinalizedBlockNumber - 1
 	if lastSafeBackfillBlock >= lp.backupPollerNextBlock {
-		lp.lggr.Infow("Backup poller backfilling logs", "start", lp.backupPollerNextBlock, "end", lastSafeBackfillBlock)
+		lp.lggr.Infow("Backup poller started backfilling logs", "start", lp.backupPollerNextBlock, "end", lastSafeBackfillBlock)
 		if err = lp.backfill(ctx, lp.backupPollerNextBlock, lastSafeBackfillBlock); err != nil {
 			// If there's an error backfilling, we can just return and retry from the last block saved
 			// since we don't save any blocks on backfilling. We may re-insert the same logs but thats ok.
 			lp.lggr.Warnw("Backup poller failed", "err", err)
 			return
 		}
+		lp.lggr.Infow("Backup poller finished backfilling", "start", lp.backupPollerNextBlock, "end", lastSafeBackfillBlock)
 		lp.backupPollerNextBlock = lastSafeBackfillBlock + 1
 	}
 }

integration-tests/client/chainlink.go

@@ -1213,3 +1213,23 @@ func (c *ChainlinkClient) GetForwarders() (*Forwarders, *http.Response, error) {
 	}
 	return response, resp.RawResponse, err
 }
+
+// Replays log poller from block number
+func (c *ChainlinkClient) ReplayLogPollerFromBlock(fromBlock, evmChainID int64) (*ReplayResponse, *http.Response, error) {
+	specObj := &ReplayResponse{}
+	c.l.Info().Str(NodeURL, c.Config.URL).Int64("From block", fromBlock).Int64("EVM chain ID", evmChainID).Msg("Replaying Log Poller from block")
+	resp, err := c.APIClient.R().
+		SetResult(&specObj).
+		SetQueryParams(map[string]string{
+			"evmChainID": fmt.Sprint(evmChainID),
+		}).
+		SetPathParams(map[string]string{
+			"fromBlock": fmt.Sprint(fromBlock),
+		}).
+		Post("/v2/replay_from_block/{fromBlock}")
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return specObj, resp.RawResponse, err
+}

integration-tests/client/chainlink_models.go

@@ -9,6 +9,7 @@ import (
 	"gopkg.in/guregu/null.v4"
 
 	"github.com/smartcontractkit/chainlink/v2/core/services/job"
+	"github.com/smartcontractkit/chainlink/v2/core/utils"
 )
 
 // EIServiceConfig represents External Initiator service config
@@ -1407,3 +1408,16 @@ type ForwarderAttributes struct {
 	CreatedAt time.Time `json:"createdAt"`
 	UpdatedAt time.Time `json:"updatedAt"`
 }
+
+type ReplayResponse struct {
+	Data ReplayResponseData `json:"data"`
+}
+
+type ReplayResponseData struct {
+	Attributes ReplayResponseAttributes `json:"attributes"`
+}
+
+type ReplayResponseAttributes struct {
+	Message    string     `json:"message"`
+	EVMChainID *utils.Big `json:"evmChainID"`
+}

integration-tests/contracts/contract_deployer.go

@@ -45,6 +45,7 @@ import (
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/generated/keeper_registry_wrapper2_0"
 	registry21 "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/generated/keeper_registry_wrapper_2_1"
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/generated/link_token_interface"
+	le "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/generated/log_emitter"
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/generated/log_triggered_streams_lookup_wrapper"
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/generated/log_upkeep_counter_wrapper"
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/generated/mock_aggregator_proxy"
@@ -138,6 +139,7 @@ type ContractDeployer interface {
 	DeployMercuryVerifierProxyContract(accessControllerAddr common.Address) (MercuryVerifierProxy, error)
 	DeployMercuryFeeManager(linkAddress common.Address, nativeAddress common.Address, proxyAddress common.Address, rewardManagerAddress common.Address) (MercuryFeeManager, error)
 	DeployMercuryRewardManager(linkAddress common.Address) (MercuryRewardManager, error)
+	DeployLogEmitterContract() (LogEmitter, error)
 }
 
 // NewContractDeployer returns an instance of a contract deployer based on the client type
@@ -1613,3 +1615,21 @@ func (e *EthereumContractDeployer) DeployWERC20Mock() (WERC20Mock, error) {
 		l:        e.l,
 	}, err
 }
+
+func (e *EthereumContractDeployer) DeployLogEmitterContract() (LogEmitter, error) {
+	address, _, instance, err := e.client.DeployContract("Log Emitter", func(
+		auth *bind.TransactOpts,
+		backend bind.ContractBackend,
+	) (common.Address, *types.Transaction, interface{}, error) {
+		return le.DeployLogEmitter(auth, backend)
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &LogEmitterContract{
+		client:   e.client,
+		instance: instance.(*le.LogEmitter),
+		address:  *address,
+		l:        e.l,
+	}, err
+}

integration-tests/contracts/contract_models.go

@@ -400,3 +400,13 @@ type WERC20Mock interface {
 	Transfer(to string, amount *big.Int) error
 	Mint(account common.Address, amount *big.Int) (*types.Transaction, error)
 }
+
+type LogEmitter interface {
+	Address() common.Address
+	EmitLogInts(ints []int) (*types.Transaction, error)
+	EmitLogIntsIndexed(ints []int) (*types.Transaction, error)
+	EmitLogStrings(strings []string) (*types.Transaction, error)
+	EmitLogInt(payload int) (*types.Transaction, error)
+	EmitLogIntIndexed(payload int) (*types.Transaction, error)
+	EmitLogString(strings string) (*types.Transaction, error)
+}

integration-tests/contracts/test_contracts.go

@@ -0,0 +1,79 @@
+package contracts
+
+import (
+	"math/big"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/rs/zerolog"
+	"github.com/smartcontractkit/chainlink-testing-framework/blockchain"
+
+	le "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/generated/log_emitter"
+)
+
+type LogEmitterContract struct {
+	address  common.Address
+	client   blockchain.EVMClient
+	instance *le.LogEmitter
+	l        zerolog.Logger
+}
+
+func (e *LogEmitterContract) Address() common.Address {
+	return e.address
+}
+
+func (e *LogEmitterContract) EmitLogInts(ints []int) (*types.Transaction, error) {
+	opts, err := e.client.TransactionOpts(e.client.GetDefaultWallet())
+	if err != nil {
+		return nil, err
+	}
+	bigInts := make([]*big.Int, len(ints))
+	for i, v := range ints {
+		bigInts[i] = big.NewInt(int64(v))
+	}
+	tx, err := e.instance.EmitLog1(opts, bigInts)
+	if err != nil {
+		return nil, err
+	}
+	return tx, e.client.ProcessTransaction(tx)
+}
+
+func (e *LogEmitterContract) EmitLogIntsIndexed(ints []int) (*types.Transaction, error) {
+	opts, err := e.client.TransactionOpts(e.client.GetDefaultWallet())
+	if err != nil {
+		return nil, err
+	}
+	bigInts := make([]*big.Int, len(ints))
+	for i, v := range ints {
+		bigInts[i] = big.NewInt(int64(v))
+	}
+	tx, err := e.instance.EmitLog2(opts, bigInts)
+	if err != nil {
+		return nil, err
+	}
+	return tx, e.client.ProcessTransaction(tx)
+}
+
+func (e *LogEmitterContract) EmitLogStrings(strings []string) (*types.Transaction, error) {
+	opts, err := e.client.TransactionOpts(e.client.GetDefaultWallet())
+	if err != nil {
+		return nil, err
+	}
+	tx, err := e.instance.EmitLog3(opts, strings)
+	if err != nil {
+		return nil, err
+	}
+	return tx, e.client.ProcessTransaction(tx)
+}
+
+func (e *LogEmitterContract) EmitLogInt(payload int) (*types.Transaction, error) {
+	return e.EmitLogInts([]int{payload})
+}
+
+func (e *LogEmitterContract) EmitLogIntIndexed(payload int) (*types.Transaction, error) {
+	return e.EmitLogIntsIndexed([]int{payload})
+}
+
+func (e *LogEmitterContract) EmitLogString(strings string) (*types.Transaction, error) {
+	return e.EmitLogStrings([]string{strings})
+}

integration-tests/docker/cmd/test_env.go

@@ -50,6 +50,7 @@ func main() {
 			return nil
 		},
 	}
+
 	startEnvCmd.AddCommand(startFullEnvCmd)
 
 	// Set default log level for non-testcontainer code

integration-tests/docker/test_env/test_env.go

@@ -166,6 +166,7 @@ func (te *CLClusterTestEnv) FundChainlinkNodes(amount *big.Float) error {
 		if err := cl.Fund(te.EVMClient, amount); err != nil {
 			return errors.Wrap(err, ErrFundCLNode)
 		}
+		time.Sleep(5 * time.Second)
 	}
 	return te.EVMClient.WaitForEvents()
 }