diff --git a/core/chains/evm/logpoller/disabled.go b/core/chains/evm/logpoller/disabled.go index 05d591042f..9bc04ff0ed 100644 --- a/core/chains/evm/logpoller/disabled.go +++ b/core/chains/evm/logpoller/disabled.go @@ -21,6 +21,10 @@ func (disabled) Name() string { return "disabledLogPoller" } func (disabled) Start(ctx context.Context) error { return ErrDisabled } +func (disabled) Healthy() error { + return ErrDisabled +} + func (disabled) Close() error { return ErrDisabled } func (disabled) Ready() error { return ErrDisabled } diff --git a/core/chains/evm/logpoller/log_poller.go b/core/chains/evm/logpoller/log_poller.go index 7006c1762e..922762467e 100644 --- a/core/chains/evm/logpoller/log_poller.go +++ b/core/chains/evm/logpoller/log_poller.go @@ -10,6 +10,7 @@ import ( "sort" "strings" "sync" + "sync/atomic" "time" "github.com/ethereum/go-ethereum" @@ -34,6 +35,7 @@ import ( //go:generate mockery --quiet --name LogPoller --output ./mocks/ --case=underscore --structname LogPoller --filename log_poller.go type LogPoller interface { services.Service + Healthy() error Replay(ctx context.Context, fromBlock int64) error ReplayAsync(fromBlock int64) RegisterFilter(filter Filter, qopts ...pg.QOpt) error @@ -70,6 +72,8 @@ const ( Unconfirmed = Confirmations(0) ) +var ErrFinalityViolated = errors.New("finality violated") + type LogPollerTest interface { LogPoller PollAndSaveLogs(ctx context.Context, currentBlockNumber int64) @@ -118,6 +122,12 @@ type logPoller struct { ctx context.Context cancel context.CancelFunc wg sync.WaitGroup + // This flag is raised whenever the log poller detects that the chain's finality has been violated. + // It can happen when reorg is deeper than the latest finalized block that LogPoller saw in a previous PollAndSave tick. + // Usually the only way to recover is to manually remove the offending logs and block from the database. + // LogPoller keeps running in infinite loop, so whenever the invalid state is removed from the database it should + // recover automatically without needing to restart the LogPoller. + finalityViolated *atomic.Bool } // NewLogPoller creates a log poller. Note there is an assumption @@ -149,6 +159,7 @@ func NewLogPoller(orm ORM, ec Client, lggr logger.Logger, pollPeriod time.Durati keepFinalizedBlocksDepth: keepFinalizedBlocksDepth, filters: make(map[string]Filter), filterDirty: true, // Always build Filter on first call to cache an empty filter if nothing registered yet. + finalityViolated: new(atomic.Bool), } } @@ -203,6 +214,13 @@ func (filter *Filter) Contains(other *Filter) bool { return true } +func (lp *logPoller) Healthy() error { + if lp.finalityViolated.Load() { + return ErrFinalityViolated + } + return nil +} + // RegisterFilter adds the provided EventSigs and Addresses to the log poller's log filter query. // If any eventSig is emitted from any address, it will be captured by the log poller. // If an event matching any of the given event signatures is emitted from any of the provided Addresses, @@ -758,9 +776,11 @@ func (lp *logPoller) getCurrentBlockMaybeHandleReorg(ctx context.Context, curren // We return an error here which will cause us to restart polling from lastBlockSaved + 1 return nil, err2 } + lp.finalityViolated.Store(false) return blockAfterLCA, nil } // No reorg, return current block. + lp.finalityViolated.Store(false) return currentBlock, nil } @@ -924,6 +944,7 @@ func (lp *logPoller) findBlockAfterLCA(ctx context.Context, current *evmtypes.He lp.lggr.Criticalw("Reorg greater than finality depth detected", "finalityTag", lp.useFinalityTag, "current", current.Number, "latestFinalized", latestFinalizedBlockNumber) rerr := errors.New("Reorg greater than finality depth") lp.SvcErrBuffer.Append(rerr) + lp.finalityViolated.Store(true) return nil, rerr } diff --git a/core/chains/evm/logpoller/log_poller_test.go b/core/chains/evm/logpoller/log_poller_test.go index 2508e676e6..4fec4b4fa0 100644 --- a/core/chains/evm/logpoller/log_poller_test.go +++ b/core/chains/evm/logpoller/log_poller_test.go @@ -984,6 +984,86 @@ func TestLogPoller_PollAndSaveLogs(t *testing.T) { } } +func TestLogPoller_ReorgDeeperThanFinality(t *testing.T) { + tests := []struct { + name string + finalityDepth int64 + finalityTag bool + }{ + { + name: "fixed finality depth without finality tag", + finalityDepth: 1, + finalityTag: false, + }, + { + name: "chain finality in use", + finalityDepth: 0, + finalityTag: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + th := SetupTH(t, tt.finalityTag, tt.finalityDepth, 3, 2, 1000) + // Set up a log poller listening for log emitter logs. + err := th.LogPoller.RegisterFilter(logpoller.Filter{ + Name: "Test Emitter", + EventSigs: []common.Hash{EmitterABI.Events["Log1"].ID}, + Addresses: []common.Address{th.EmitterAddress1}, + }) + require.NoError(t, err) + + // Test scenario + // Chain gen <- 1 <- 2 <- 3 (finalized) <- 4 (L1_1) + _, err = th.Emitter1.EmitLog1(th.Owner, []*big.Int{big.NewInt(1)}) + require.NoError(t, err) + th.Client.Commit() + th.Client.Commit() + th.Client.Commit() + markBlockAsFinalized(t, th, 3) + + // Polling should get us the L1 log. + firstPoll := th.PollAndSaveLogs(testutils.Context(t), 1) + assert.Equal(t, int64(5), firstPoll) + assert.NoError(t, th.LogPoller.Healthy()) + + // Fork deeper than finality depth + // Chain gen <- 1 <- 2 <- 3 (finalized) <- 4 (L1_1) + // \ 2' <- 3' <- 4' <- 5' <- 6' (finalized) <- 7' <- 8' <- 9' <- 10' (L1_2) + lca, err := th.Client.BlockByNumber(testutils.Context(t), big.NewInt(1)) + require.NoError(t, err) + require.NoError(t, th.Client.Fork(testutils.Context(t), lca.Hash())) + + // Create 2' + _, err = th.Emitter1.EmitLog1(th.Owner, []*big.Int{big.NewInt(2)}) + require.NoError(t, err) + th.Client.Commit() + + // Create 3-10 + for i := 3; i < 10; i++ { + _, err = th.Emitter1.EmitLog1(th.Owner, []*big.Int{big.NewInt(int64(i))}) + require.NoError(t, err) + th.Client.Commit() + } + markBlockAsFinalized(t, th, 6) + + secondPoll := th.PollAndSaveLogs(testutils.Context(t), firstPoll) + assert.Equal(t, firstPoll, secondPoll) + assert.Equal(t, logpoller.ErrFinalityViolated, th.LogPoller.Healthy()) + + // Manually remove latest block from the log poller to bring it back to life + // LogPoller should be healthy again after first poll + // Chain gen <- 1 + // \ 2' <- 3' <- 4' <- 5' <- 6' (finalized) <- 7' <- 8' <- 9' <- 10' (L1_2) + require.NoError(t, th.ORM.DeleteLogsAndBlocksAfter(2)) + // Poll from latest + recoveryPoll := th.PollAndSaveLogs(testutils.Context(t), 1) + assert.Equal(t, int64(10), recoveryPoll) + assert.NoError(t, th.LogPoller.Healthy()) + }) + } +} + func TestLogPoller_PollAndSaveLogsDeepReorg(t *testing.T) { t.Parallel() @@ -1027,6 +1107,7 @@ func TestLogPoller_PollAndSaveLogsDeepReorg(t *testing.T) { // Polling should get us the L1 log. newStart := th.PollAndSaveLogs(testutils.Context(t), 1) assert.Equal(t, int64(3), newStart) + assert.NoError(t, th.LogPoller.Healthy()) // Check that L1_1 has a proper data payload lgs, err := th.ORM.SelectLogsByBlockRange(2, 2) require.NoError(t, err) @@ -1052,6 +1133,7 @@ func TestLogPoller_PollAndSaveLogsDeepReorg(t *testing.T) { newStart = th.PollAndSaveLogs(testutils.Context(t), newStart) assert.Equal(t, int64(10), newStart) + assert.NoError(t, th.LogPoller.Healthy()) // Expect L1_2 to be properly updated lgs, err = th.ORM.SelectLogsByBlockRange(2, 2) diff --git a/core/chains/evm/logpoller/mocks/log_poller.go b/core/chains/evm/logpoller/mocks/log_poller.go index 65d808b98d..453eeac272 100644 --- a/core/chains/evm/logpoller/mocks/log_poller.go +++ b/core/chains/evm/logpoller/mocks/log_poller.go @@ -114,6 +114,24 @@ func (_m *LogPoller) HealthReport() map[string]error { return r0 } +// Healthy provides a mock function with given fields: +func (_m *LogPoller) Healthy() error { + ret := _m.Called() + + if len(ret) == 0 { + panic("no return value specified for Healthy") + } + + var r0 error + if rf, ok := ret.Get(0).(func() error); ok { + r0 = rf() + } else { + r0 = ret.Error(0) + } + + return r0 +} + // IndexedLogs provides a mock function with given fields: eventSig, address, topicIndex, topicValues, confs, qopts func (_m *LogPoller) IndexedLogs(eventSig common.Hash, address common.Address, topicIndex int, topicValues []common.Hash, confs logpoller.Confirmations, qopts ...pg.QOpt) ([]logpoller.Log, error) { _va := make([]interface{}, len(qopts))