diff --git a/core/chains/evm/logpoller/orm.go b/core/chains/evm/logpoller/orm.go index 22870efccf3..eb09459c2eb 100644 --- a/core/chains/evm/logpoller/orm.go +++ b/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) diff --git a/core/chains/evm/logpoller/orm_test.go b/core/chains/evm/logpoller/orm_test.go index ab8d126e10a..ee7f408061d 100644 --- a/core/chains/evm/logpoller/orm_test.go +++ b/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)