Fix LP tests filters check and add to smoke tests in CI

.github/workflows/integration-tests.yml

@@ -328,7 +328,7 @@ jobs:
             run: -run TestOCRJobReplacement
             file: ocr
             pyroscope_env: ci-smoke-ocr-evm-simulated                         
-          - name: ocr2
+          - name: ocr2-replacement
             nodes: 1
             os: ubuntu-latest
             run: -run TestOCRv2JobReplacement
@@ -369,6 +369,24 @@ jobs:
             nodes: 1
             os: ubuntu-latest
             pyroscope_env: ci-smoke-forwarder-ocr-evm-simulated
+          - name: log-poller-finality-tag
+            nodes: 1
+            os: ubuntu-latest
+            run: -run TestLogPollerFewFiltersFinalityTag
+            file: log_poller
+            pyroscope_env: ""
+          - name: log-poller-chaos-fixed-depth
+            nodes: 1
+            os: ubuntu-latest
+            run: -run TestLogPollerWithChaosFixedDepth
+            file: log_poller
+            pyroscope_env: ""          
+          - name: log-poller-replay-finality-tag
+            nodes: 1
+            os: ubuntu-latest
+            run: -run TestLogPollerReplayFinalityTag
+            file: log_poller
+            pyroscope_env: ""              
     runs-on: ${{ matrix.product.os }}
     name: ETH Smoke Tests ${{ matrix.product.name }}${{ matrix.product.tag_suffix }}
     steps:

integration-tests/smoke/log_poller_test.go

@@ -70,7 +70,7 @@ func TestLogPollerFewFiltersFinalityTag(t *testing.T) {
 
 // consistency test with no network disruptions with approximate emission of 1000-1100 logs per second for ~110-120 seconds
 // 900 filters are registered
-func TestLogManyFiltersPollerFixedDepth(t *testing.T) {
+func TestLogPollerManyFiltersFixedDepth(t *testing.T) {
 	cfg := logpoller.Config{
 		General: &logpoller.General{
 			Generator:      logpoller.GeneratorType_Looped,
@@ -99,7 +99,7 @@ func TestLogManyFiltersPollerFixedDepth(t *testing.T) {
 	logpoller.ExecuteBasicLogPollerTest(t, &cfg)
 }
 
-func TestLogManyFiltersPollerFinalityTag(t *testing.T) {
+func TestLogPollerManyFiltersFinalityTag(t *testing.T) {
 	cfg := logpoller.Config{
 		General: &logpoller.General{
 			Generator:      logpoller.GeneratorType_Looped,
@@ -141,15 +141,15 @@ func TestLogPollerWithChaosFixedDepth(t *testing.T) {
 		},
 		LoopedConfig: &logpoller.LoopedConfig{
 			ContractConfig: logpoller.ContractConfig{
-				ExecutionCount: 100,
+				ExecutionCount: 50,
 			},
 			FuzzConfig: logpoller.FuzzConfig{
-				MinEmitWaitTimeMs: 200,
-				MaxEmitWaitTimeMs: 500,
+				MinEmitWaitTimeMs: 100,
+				MaxEmitWaitTimeMs: 300,
 			},
 		},
 		ChaosConfig: &logpoller.ChaosConfig{
-			ExperimentCount: 10,
+			ExperimentCount: 5,
 		},
 	}
 
@@ -173,15 +173,15 @@ func TestLogPollerWithChaosFinalityTag(t *testing.T) {
 		},
 		LoopedConfig: &logpoller.LoopedConfig{
 			ContractConfig: logpoller.ContractConfig{
-				ExecutionCount: 100,
+				ExecutionCount: 50,
 			},
 			FuzzConfig: logpoller.FuzzConfig{
-				MinEmitWaitTimeMs: 200,
-				MaxEmitWaitTimeMs: 500,
+				MinEmitWaitTimeMs: 100,
+				MaxEmitWaitTimeMs: 300,
 			},
 		},
 		ChaosConfig: &logpoller.ChaosConfig{
-			ExperimentCount: 10,
+			ExperimentCount: 5,
 		},
 	}
 
@@ -236,7 +236,7 @@ func TestLogPollerReplayFinalityTag(t *testing.T) {
 			Generator:      logpoller.GeneratorType_Looped,
 			Contracts:      2,
 			EventsPerTx:    4,
-			UseFinalityTag: false,
+			UseFinalityTag: true,
 		},
 		LoopedConfig: &logpoller.LoopedConfig{
 			ContractConfig: logpoller.ContractConfig{

integration-tests/universal/log_poller/helpers.go

@@ -12,11 +12,13 @@ import (
 	"testing"
 	"time"
 
+	"cosmossdk.io/errors"
 	geth "github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/accounts/abi"
 	"github.com/ethereum/go-ethereum/common"
 	geth_types "github.com/ethereum/go-ethereum/core/types"
 	"github.com/jmoiron/sqlx"
+	"github.com/onsi/gomega"
 	"github.com/rs/zerolog"
 	"github.com/scylladb/go-reflectx"
 	"github.com/stretchr/testify/require"
@@ -28,6 +30,7 @@ import (
 	"github.com/smartcontractkit/chainlink-testing-framework/logging"
 	"github.com/smartcontractkit/chainlink-testing-framework/networks"
 	"github.com/smartcontractkit/chainlink-testing-framework/utils/ptr"
+	"github.com/smartcontractkit/chainlink-testing-framework/utils/testcontext"
 	evmcfg "github.com/smartcontractkit/chainlink/v2/core/chains/evm/config/toml"
 	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/logpoller"
 	cltypes "github.com/smartcontractkit/chainlink/v2/core/chains/evm/types"
@@ -227,14 +230,14 @@ func getStringSlice(length int) []string {
 var emitEvents = func(ctx context.Context, l zerolog.Logger, logEmitter *contracts.LogEmitter, cfg *Config, wg *sync.WaitGroup, results chan LogEmitterChannel) {
 	address := (*logEmitter).Address().String()
 	localCounter := 0
-	select {
-	case <-ctx.Done():
-		l.Warn().Str("Emitter address", address).Msg("Context cancelled, not emitting events")
-		return
-	default:
-		defer wg.Done()
-		for i := 0; i < cfg.LoopedConfig.ExecutionCount; i++ {
-			for _, event := range cfg.General.EventsToEmit {
+	defer wg.Done()
+	for i := 0; i < cfg.LoopedConfig.ExecutionCount; i++ {
+		for _, event := range cfg.General.EventsToEmit {
+			select {
+			case <-ctx.Done():
+				l.Warn().Str("Emitter address", address).Msg("Context cancelled, not emitting events")
+				return
+			default:
 				l.Debug().Str("Emitter address", address).Str("Event type", event.Name).Str("index", fmt.Sprintf("%d/%d", (i+1), cfg.LoopedConfig.ExecutionCount)).Msg("Emitting log from emitter")
 				var err error
 				switch event.Name {
@@ -244,14 +247,15 @@ var emitEvents = func(ctx context.Context, l zerolog.Logger, logEmitter *contrac
 					_, err = (*logEmitter).EmitLogIntsIndexed(getIntSlice(cfg.General.EventsPerTx))
 				case "Log3":
 					_, err = (*logEmitter).EmitLogStrings(getStringSlice(cfg.General.EventsPerTx))
+				case "Log4":
+					_, err = (*logEmitter).EmitLogIntMultiIndexed(1, 1, cfg.General.EventsPerTx)
 				default:
 					err = fmt.Errorf("unknown event name: %s", event.Name)
 				}
 
 				if err != nil {
 					results <- LogEmitterChannel{
-						logsEmitted: 0,
-						err:         err,
+						err: err,
 					}
 					return
 				}
@@ -264,35 +268,24 @@ var emitEvents = func(ctx context.Context, l zerolog.Logger, logEmitter *contrac
 				l.Info().Str("Emitter address", address).Str("Index", fmt.Sprintf("%d/%d", i+1, cfg.LoopedConfig.ExecutionCount)).Msg("Emitted all three events")
 			}
 		}
-
-		l.Info().Str("Emitter address", address).Int("Total logs emitted", localCounter).Msg("Finished emitting events")
-
-		results <- LogEmitterChannel{
-			logsEmitted: localCounter,
-			err:         nil,
-		}
-	}
-}
-
-var chainHasFinalisedEndBlock = func(l zerolog.Logger, evmClient blockchain.EVMClient, endBlock int64) (bool, error) {
-	effectiveEndBlock := endBlock + 1
-	lastFinalisedBlockHeader, err := evmClient.GetLatestFinalizedBlockHeader(context.Background())
-	if err != nil {
-		return false, err
 	}
 
-	l.Info().Int64("Last finalised block header", lastFinalisedBlockHeader.Number.Int64()).Int64("End block", effectiveEndBlock).Int64("Blocks left till end block", effectiveEndBlock-lastFinalisedBlockHeader.Number.Int64()).Msg("Waiting for the finalized block to move beyond end block")
+	l.Info().Str("Emitter address", address).Int("Total logs emitted", localCounter).Msg("Finished emitting events")
 
-	return lastFinalisedBlockHeader.Number.Int64() > effectiveEndBlock, nil
+	results <- LogEmitterChannel{
+		logsEmitted: localCounter,
+		err:         nil,
+	}
 }
 
 var logPollerHasFinalisedEndBlock = func(endBlock int64, chainID *big.Int, l zerolog.Logger, coreLogger core_logger.SugaredLogger, nodes *test_env.ClCluster) (bool, error) {
 	wg := &sync.WaitGroup{}
 
 	type boolQueryResult struct {
-		nodeName     string
-		hasFinalised bool
-		err          error
+		nodeName       string
+		hasFinalised   bool
+		finalizedBlock int64
+		err            error
 	}
 
 	endBlockCh := make(chan boolQueryResult, len(nodes.Nodes)-1)
@@ -328,9 +321,10 @@ var logPollerHasFinalisedEndBlock = func(endBlock int64, chainID *big.Int, l zer
 				}
 
 				r <- boolQueryResult{
-					nodeName:     clNode.ContainerName,
-					hasFinalised: latestBlock.FinalizedBlockNumber > endBlock,
-					err:          nil,
+					nodeName:       clNode.ContainerName,
+					finalizedBlock: latestBlock.FinalizedBlockNumber,
+					hasFinalised:   latestBlock.FinalizedBlockNumber > endBlock,
+					err:            nil,
 				}
 
 			}
@@ -353,7 +347,7 @@ var logPollerHasFinalisedEndBlock = func(endBlock int64, chainID *big.Int, l zer
 			if r.hasFinalised {
 				l.Info().Str("Node name", r.nodeName).Msg("CL node has finalised end block")
 			} else {
-				l.Warn().Str("Node name", r.nodeName).Msg("CL node has not finalised end block yet")
+				l.Warn().Int64("Has", r.finalizedBlock).Int64("Want", endBlock).Str("Node name", r.nodeName).Msg("CL node has not finalised end block yet")
 			}
 
 			if len(foundMap) == len(nodes.Nodes)-1 {
@@ -798,27 +792,32 @@ func runLoopedGenerator(t *testing.T, cfg *Config, logEmitters []*contracts.LogE
 	aggrChan := make(chan int, len(logEmitters))
 
 	go func() {
-		for emitter := range emitterCh {
-			if emitter.err != nil {
-				emitErr = emitter.err
-				cancelFn()
+		for {
+			select {
+			case <-ctx.Done():
 				return
+			case emitter := <-emitterCh:
+				if emitter.err != nil {
+					emitErr = emitter.err
+					cancelFn()
+					return
+				}
+				aggrChan <- emitter.logsEmitted
 			}
-			aggrChan <- emitter.logsEmitted
 		}
 	}()
 
 	wg.Wait()
 	close(emitterCh)
 
-	for i := 0; i < len(logEmitters); i++ {
-		total += <-aggrChan
-	}
-
 	if emitErr != nil {
 		return 0, emitErr
 	}
 
+	for i := 0; i < len(logEmitters); i++ {
+		total += <-aggrChan
+	}
+
 	return int(total), nil
 }
 
@@ -1043,9 +1042,9 @@ func setupLogPollerTestDocker(
 		WithConsensusType(ctf_test_env.ConsensusType_PoS).
 		WithConsensusLayer(ctf_test_env.ConsensusLayer_Prysm).
 		WithExecutionLayer(ctf_test_env.ExecutionLayer_Geth).
-		WithWaitingForFinalization().
+		// WithWaitingForFinalization().
 		WithEthereumChainConfig(ctf_test_env.EthereumChainConfig{
-			SecondsPerSlot: 8,
+			SecondsPerSlot: 4,
 			SlotsPerEpoch:  2,
 		}).
 		Build()
@@ -1118,3 +1117,97 @@ func setupLogPollerTestDocker(
 
 	return env.EVMClient, nodeClients, env.ContractDeployer, linkToken, registry, registrar, env
 }
+
+func uploadLogEmitterContractsAndWaitForFinalisation(l zerolog.Logger, t *testing.T, testEnv *test_env.CLClusterTestEnv, cfg *Config) []*contracts.LogEmitter {
+	logEmitters := make([]*contracts.LogEmitter, 0)
+	for i := 0; i < cfg.General.Contracts; i++ {
+		logEmitter, err := testEnv.ContractDeployer.DeployLogEmitterContract()
+		logEmitters = append(logEmitters, &logEmitter)
+		require.NoError(t, err, "Error deploying log emitter contract")
+		l.Info().Str("Contract address", logEmitter.Address().Hex()).Msg("Log emitter contract deployed")
+		time.Sleep(200 * time.Millisecond)
+	}
+	afterUploadBlock, err := testEnv.EVMClient.LatestBlockNumber(testcontext.Get(t))
+	require.NoError(t, err, "Error getting latest block number")
+
+	gom := gomega.NewGomegaWithT(t)
+	gom.Eventually(func(g gomega.Gomega) {
+		targetBlockNumber := int64(afterUploadBlock + 1)
+		finalized, err := testEnv.EVMClient.GetLatestFinalizedBlockHeader(testcontext.Get(t))
+		if err != nil {
+			l.Warn().Err(err).Msg("Error checking if contract were uploaded. Retrying...")
+			return
+		}
+		finalizedBlockNumber := finalized.Number.Int64()
+
+		if finalizedBlockNumber < targetBlockNumber {
+			l.Debug().Int64("Finalized block", finalized.Number.Int64()).Int64("After upload block", int64(afterUploadBlock+1)).Msg("Waiting for contract upload to finalise")
+		}
+
+		g.Expect(finalizedBlockNumber >= targetBlockNumber).To(gomega.BeTrue(), "Contract upload did not finalize in time")
+	}, "2m", "10s").Should(gomega.Succeed())
+
+	return logEmitters
+}
+
+func assertUpkeepIdsUniqueness(upkeepIDs []*big.Int) error {
+	upKeepIdSeen := make(map[int64]bool)
+	for _, upkeepID := range upkeepIDs {
+		if _, ok := upKeepIdSeen[upkeepID.Int64()]; ok {
+			return fmt.Errorf("Duplicate upkeep ID %d", upkeepID.Int64())
+		}
+		upKeepIdSeen[upkeepID.Int64()] = true
+	}
+
+	return nil
+}
+
+func assertContractAddressUniquneness(logEmitters []*contracts.LogEmitter) error {
+	contractAddressSeen := make(map[string]bool)
+	for _, logEmitter := range logEmitters {
+		address := (*logEmitter).Address().String()
+		if _, ok := contractAddressSeen[address]; ok {
+			return fmt.Errorf("Duplicate contract address %s", address)
+		}
+		contractAddressSeen[address] = true
+	}
+
+	return nil
+}
+
+func registerFiltersAndAssertUniquness(l zerolog.Logger, registry contracts.KeeperRegistry, upkeepIDs []*big.Int, logEmitters []*contracts.LogEmitter, cfg *Config, upKeepsNeeded int) error {
+	uniqueFilters := make(map[string]bool)
+
+	upkeepIdIndex := 0
+	for i := 0; i < len(logEmitters); i++ {
+		for j := 0; j < len(cfg.General.EventsToEmit); j++ {
+			emitterAddress := (*logEmitters[i]).Address()
+			topicId := cfg.General.EventsToEmit[j].ID
+
+			upkeepID := upkeepIDs[upkeepIdIndex]
+			l.Debug().Int("Upkeep id", int(upkeepID.Int64())).Str("Emitter address", emitterAddress.String()).Str("Topic", topicId.Hex()).Msg("Registering log trigger for log emitter")
+			err := registerSingleTopicFilter(registry, upkeepID, emitterAddress, topicId)
+			randomWait(150, 300)
+			if err != nil {
+				return errors.Wrapf(err, "Error registering log trigger for log emitter %s", emitterAddress.String())
+			}
+
+			if i%10 == 0 {
+				l.Info().Msgf("Registered log trigger for topic %d for log emitter %d/%d", j, i, len(logEmitters))
+			}
+
+			key := fmt.Sprintf("%s-%s", emitterAddress.String(), topicId.Hex())
+			if _, ok := uniqueFilters[key]; ok {
+				return fmt.Errorf("Duplicate filter %s", key)
+			}
+			uniqueFilters[key] = true
+			upkeepIdIndex++
+		}
+	}
+
+	if upKeepsNeeded != len(uniqueFilters) {
+		return fmt.Errorf("Number of unique filters should be equal to number of upkeeps. Expected %d. Got %d", upKeepsNeeded, len(uniqueFilters))
+	}
+
+	return nil
+}