Merge pull request #14462 from smartcontractkit/hotfix/BCFR-906-logpo…

…ller-pruner

v2.16.0 hot fix for LogPoller pruning issue

core/chains/evm/logpoller/orm.go

@@ -330,8 +330,9 @@ func (o *DSORM) DeleteExpiredLogs(ctx context.Context, limit int64) (int64, erro
 				FROM evm.log_poller_filters
 				WHERE evm_chain_id = $1
 				GROUP BY evm_chain_id, address, event
-			) r ON l.evm_chain_id = $1 AND l.address = r.address AND l.event_sig = r.event
-			WHERE r.retention IS NULL OR (r.retention != 0 AND l.block_timestamp <= STATEMENT_TIMESTAMP() - (r.retention / 10^9 * interval '1 second')) %s)`
+			) r ON l.address = r.address AND l.event_sig = r.event
+			WHERE l.evm_chain_id = $1 AND -- Must be WHERE rather than ON due to LEFT JOIN
+			      r.retention IS NULL OR (r.retention != 0 AND l.block_timestamp <= STATEMENT_TIMESTAMP() - (r.retention / 10^9 * interval '1 second')) %s)`
 
 	if limit > 0 {
 		result, err = o.ds.ExecContext(ctx, fmt.Sprintf(query, "LIMIT $2"), ubig.New(o.chainID), limit)

core/chains/evm/logpoller/orm_test.go

@@ -187,6 +187,7 @@ func TestORM_GetBlocks_From_Range_Recent_Blocks(t *testing.T) {
 }
 
 func TestORM(t *testing.T) {
+	t.Parallel()
 	th := SetupTH(t, lpOpts)
 	o1 := th.ORM
 	o2 := th.ORM2
@@ -334,6 +335,36 @@ func TestORM(t *testing.T) {
 		},
 	}))
 
+	// Insert a couple logs on a different chain, to make sure
+	// these aren't affected by any operations on the chain LogPoller
+	// is managing.
+	require.NoError(t, o2.InsertLogs(ctx, []logpoller.Log{
+		{
+			EvmChainId:     ubig.New(th.ChainID2),
+			LogIndex:       8,
+			BlockHash:      common.HexToHash("0x1238"),
+			BlockNumber:    int64(17),
+			EventSig:       topic2,
+			Topics:         [][]byte{topic2[:]},
+			Address:        common.HexToAddress("0x1236"),
+			TxHash:         common.HexToHash("0x1888"),
+			Data:           []byte("same log on unrelated chain"),
+			BlockTimestamp: time.Now(),
+		},
+		{
+			EvmChainId:     ubig.New(th.ChainID2),
+			LogIndex:       9,
+			BlockHash:      common.HexToHash("0x1999"),
+			BlockNumber:    int64(18),
+			EventSig:       topic,
+			Topics:         [][]byte{topic[:], topic2[:]},
+			Address:        common.HexToAddress("0x5555"),
+			TxHash:         common.HexToHash("0x1543"),
+			Data:           []byte("different log on unrelated chain"),
+			BlockTimestamp: time.Now(),
+		},
+	}))
+
 	t.Log(latest.BlockNumber)
 	logs, err := o1.SelectLogsByBlockRange(ctx, 1, 17)
 	require.NoError(t, err)
@@ -454,11 +485,24 @@ func TestORM(t *testing.T) {
 	require.NoError(t, err)
 	require.Len(t, logs, 8)
 
-	// Delete expired logs
+	// Delete expired logs with page limit
 	time.Sleep(2 * time.Millisecond) // just in case we haven't reached the end of the 1ms retention period
-	deleted, err := o1.DeleteExpiredLogs(ctx, 0)
+	deleted, err := o1.DeleteExpiredLogs(ctx, 2)
+	require.NoError(t, err)
+	assert.Equal(t, int64(2), deleted)
+
+	// Delete expired logs without page limit
+	deleted, err = o1.DeleteExpiredLogs(ctx, 0)
+	require.NoError(t, err)
+	assert.Equal(t, int64(2), deleted)
+
+	// Ensure that both of the logs from the second chain are still there
+	logs, err = o2.SelectLogs(ctx, 0, 100, common.HexToAddress("0x1236"), topic2)
+	require.NoError(t, err)
+	assert.Len(t, logs, 1)
+	logs, err = o2.SelectLogs(ctx, 0, 100, common.HexToAddress("0x5555"), topic)
 	require.NoError(t, err)
-	assert.Equal(t, int64(4), deleted)
+	assert.Len(t, logs, 1)
 
 	logs, err = o1.SelectLogsByBlockRange(ctx, 1, latest.BlockNumber)
 	require.NoError(t, err)