From 1ce80bed2bda1485d462fe0b866c19afe2e6a1af Mon Sep 17 00:00:00 2001 From: Dmytro Haidashenko Date: Wed, 8 Jul 2026 18:56:26 +0200 Subject: [PATCH 01/12] Refactor LP to simplify additon of method to skip blocks --- pkg/logpoller/log_poller.go | 138 +++++++++++++++++++++--------------- 1 file changed, 79 insertions(+), 59 deletions(-) diff --git a/pkg/logpoller/log_poller.go b/pkg/logpoller/log_poller.go index 4bb2275a7e..b1bb3b6cda 100644 --- a/pkg/logpoller/log_poller.go +++ b/pkg/logpoller/log_poller.go @@ -136,11 +136,17 @@ type logPoller struct { filterDirty bool cachedAddresses []common.Address cachedEventSigs []common.Hash + filtersLoaded bool replayStart chan int64 replayComplete chan error stopCh services.StopChan wg sync.WaitGroup + // ensures that jobs outside main loop do not run concurrently with it. + // Mainly needed to simplify mental model and exclude concurrent changes to the blocks table, while handling reorgs. + // Pruning of old blocks is intentionally excluded from this lock, because they are much older + // than a block range that could be affected by a reorg. + runMu sync.Mutex // 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. @@ -675,72 +681,82 @@ func (lp *logPoller) run() { lp.lggr.Infow("Backup log poller disabled") } - filtersLoaded := false - for { select { case <-ctx.Done(): return case fromBlockReq := <-lp.replayStart: - lp.handleReplayRequest(ctx, fromBlockReq, filtersLoaded) + lp.handleReplayRequest(ctx, fromBlockReq) case <-logPollTicker.C: - if !filtersLoaded { - if err := lp.loadFilters(ctx); err != nil { - lp.lggr.Errorw("Failed loading filters in main logpoller loop, retrying later", "err", err) - continue - } - filtersLoaded = true - } - - // Always start from the latest block in the db. - var start int64 - lastProcessed, err := lp.orm.SelectLatestBlock(ctx) - if err != nil { - if !pkgerrors.Is(err, sql.ErrNoRows) { - // Assume transient db reading issue, retry forever. - lp.lggr.Errorw("unable to get starting block", "err", err) - continue - } - // Otherwise this is the first poll _ever_ on a new chain. - // Only safe thing to do is to start at the first finalized block. - _, latestFinalizedBlockNumber, err := lp.latestAndFinalizedBlocks(ctx) - if err != nil { - lp.lggr.Warnw("Unable to get latest for first poll", "err", err) - continue - } - // Starting at the first finalized block. We do not backfill the first finalized block. - start = latestFinalizedBlockNumber - } else { - lp.metrics.RecordLastProcessedBlock(ctx, lastProcessed.BlockNumber) - start = lastProcessed.BlockNumber + 1 - } - lp.PollAndSaveLogs(ctx, start, false) + lp.tickPoll(ctx) case <-backupLogPollCh: - // Backup log poller: this serves as an emergency backup to protect against eventual-consistency behavior - // of an rpc node (seen occasionally on optimism, but possibly could happen on other chains?). If the first - // time we request a block, no logs or incomplete logs come back, this ensures that every log is eventually - // re-requested after it is finalized. This doesn't add much overhead, because we can request all of them - // in one shot, since we don't need to worry about re-orgs after finality depth, and it runs far less - // frequently than the primary log poller (instead of roughly once per block it runs once roughly once every - // lp.backupPollerDelay blocks--with default settings about 100x less frequently). - - if !filtersLoaded { - lp.lggr.Warnw("Backup log poller ran before filters loaded, skipping") - continue - } - err := lp.BackupPollAndSaveLogs(ctx) - switch { - case errors.Is(err, commontypes.ErrFinalityViolated): - // BackupPoll only declares finality violation and does not resolve it, as it's possible that processed range - // does not contain the violation. - lp.lggr.Criticalw("Backup poll failed due to finality violation", "err", err) - lp.finalityViolated.Store(true) - lp.SvcErrBuffer.Append(err) - case err != nil: - lp.lggr.Errorw("Backup poller failed, retrying later", "err", err) - } + lp.tickBackupPoll(ctx) + } + } +} + +func (lp *logPoller) tickBackupPoll(ctx context.Context) { + lp.runMu.Lock() + defer lp.runMu.Unlock() + // Backup log poller: this serves as an emergency backup to protect against eventual-consistency behavior + // of an rpc node (seen occasionally on optimism, but possibly could happen on other chains?). If the first + // time we request a block, no logs or incomplete logs come back, this ensures that every log is eventually + // re-requested after it is finalized. This doesn't add much overhead, because we can request all of them + // in one shot, since we don't need to worry about re-orgs after finality depth, and it runs far less + // frequently than the primary log poller (instead of roughly once per block it runs once roughly once every + // lp.backupPollerDelay blocks--with default settings about 100x less frequently). + + if !lp.filtersLoaded { + lp.lggr.Warnw("Backup log poller ran before filters loaded, skipping") + return + } + err := lp.BackupPollAndSaveLogs(ctx) + switch { + case errors.Is(err, commontypes.ErrFinalityViolated): + // BackupPoll only declares finality violation and does not resolve it, as it's possible that processed range + // does not contain the violation. + lp.lggr.Criticalw("Backup poll failed due to finality violation", "err", err) + lp.finalityViolated.Store(true) + lp.SvcErrBuffer.Append(err) + case err != nil: + lp.lggr.Errorw("Backup poller failed, retrying later", "err", err) + } +} + +func (lp *logPoller) tickPoll(ctx context.Context) { + lp.runMu.Lock() + defer lp.runMu.Unlock() + if !lp.filtersLoaded { + if err := lp.loadFilters(ctx); err != nil { + lp.lggr.Errorw("Failed loading filters in main logpoller loop, retrying later", "err", err) + return + } + lp.filtersLoaded = true + } + + // Always start from the latest block in the db. + var start int64 + lastProcessed, err := lp.orm.SelectLatestBlock(ctx) + if err != nil { + if !pkgerrors.Is(err, sql.ErrNoRows) { + // Assume transient db reading issue, retry forever. + lp.lggr.Errorw("unable to get starting block", "err", err) + return } + // Otherwise this is the first poll _ever_ on a new chain. + // Only safe thing to do is to start at the first finalized block. + _, latestFinalizedBlockNumber, err := lp.latestAndFinalizedBlocks(ctx) + if err != nil { + lp.lggr.Warnw("Unable to get latest for first poll", "err", err) + return + } + // Starting at the first finalized block. We do not backfill the first finalized block. + start = latestFinalizedBlockNumber + } else { + lp.metrics.RecordLastProcessedBlock(ctx, lastProcessed.BlockNumber) + start = lastProcessed.BlockNumber + 1 } + lp.PollAndSaveLogs(ctx, start, false) } func (lp *logPoller) backgroundWorkerRun() { @@ -809,13 +825,17 @@ func (lp *logPoller) backgroundWorkerRun() { } } -func (lp *logPoller) handleReplayRequest(ctx context.Context, fromBlockReq int64, filtersLoaded bool) { +func (lp *logPoller) handleReplayRequest(ctx context.Context, fromBlockReq int64) { + lp.runMu.Lock() + defer lp.runMu.Unlock() fromBlock, err := lp.GetReplayFromBlock(ctx, fromBlockReq) if err == nil { - if !filtersLoaded { + if !lp.filtersLoaded { lp.lggr.Warnw("Received replayReq before filters loaded", "fromBlock", fromBlock, "requested", fromBlockReq) if err = lp.loadFilters(ctx); err != nil { lp.lggr.Errorw("Failed loading filters during Replay", "err", err, "fromBlock", fromBlock) + } else { + lp.filtersLoaded = true } } if err == nil { From eb2dbf9f52083e30518f56995b0690e92c865fc5 Mon Sep 17 00:00:00 2001 From: Dmytro Haidashenko Date: Thu, 9 Jul 2026 18:59:49 +0200 Subject: [PATCH 02/12] SkipToBlock implementation --- pkg/logpoller/disabled.go | 4 + pkg/logpoller/log_poller.go | 92 +++++++++++-- pkg/logpoller/log_poller_internal_test.go | 149 ++++++++++++++++++++++ pkg/logpoller/mock_orm_test.go | 48 +++++++ pkg/logpoller/mocks/log_poller.go | 47 +++++++ pkg/logpoller/observability.go | 6 + pkg/logpoller/orm.go | 63 ++++++--- pkg/logpoller/orm_test.go | 143 ++++++++++++++++++++- 8 files changed, 519 insertions(+), 33 deletions(-) diff --git a/pkg/logpoller/disabled.go b/pkg/logpoller/disabled.go index 806e11bb74..429c799b4f 100644 --- a/pkg/logpoller/disabled.go +++ b/pkg/logpoller/disabled.go @@ -129,3 +129,7 @@ func (d disabled) FindLCA(ctx context.Context) (*Block, error) { func (d disabled) DeleteLogsAndBlocksAfter(ctx context.Context, start int64) error { return ErrDisabled } + +func (d disabled) SkipToBlock(ctx context.Context, blockNumber int64) error { + return ErrDisabled +} diff --git a/pkg/logpoller/log_poller.go b/pkg/logpoller/log_poller.go index b1bb3b6cda..c662c6ef5b 100644 --- a/pkg/logpoller/log_poller.go +++ b/pkg/logpoller/log_poller.go @@ -30,6 +30,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/timeutil" commontypes "github.com/smartcontractkit/chainlink-common/pkg/types" "github.com/smartcontractkit/chainlink-common/pkg/types/query" + "github.com/smartcontractkit/chainlink-common/pkg/types/query/primitives" "github.com/smartcontractkit/chainlink-common/pkg/utils/mathutil" "github.com/smartcontractkit/chainlink-evm/pkg/client" @@ -55,6 +56,10 @@ type LogPoller interface { GetBlocksRange(ctx context.Context, numbers []uint64) ([]Block, error) FindLCA(ctx context.Context) (*Block, error) DeleteLogsAndBlocksAfter(ctx context.Context, start int64) error + // SkipToBlock repositions the poller to start processing from blockNumber (T). + // Block T-1 is fetched from RPC and saved as the anchor; the next poll resumes from T. + // Returns an error if T-1 is not finalized on chain. + SkipToBlock(ctx context.Context, blockNumber int64) error // General querying Logs(ctx context.Context, start, end int64, eventSig common.Hash, address common.Address) ([]Log, error) @@ -91,6 +96,7 @@ type LogPollerTest interface { type Client interface { HeadByNumber(ctx context.Context, n *big.Int) (*evmtypes.Head, error) + HeaderByNumberWithOpts(ctx context.Context, number *big.Int, opts evmtypes.HeaderByNumberOpts) (*evmtypes.Header, error) HeadByHash(ctx context.Context, n common.Hash) (*evmtypes.Head, error) BatchCallContext(ctx context.Context, b []rpc.BatchElem) error FilterLogs(ctx context.Context, q ethereum.FilterQuery) ([]types.Log, error) @@ -1870,15 +1876,7 @@ func (lp *logPoller) batchFetchBlocks(ctx context.Context, blocksRequested []uin } for _, head := range fetched1 { - lpBlock := Block{ - EVMChainID: head.EVMChainID, - BlockHash: head.Hash, - BlockNumber: head.Number, - BlockTimestamp: head.Timestamp, - FinalizedBlockNumber: head.Number, // always finalized; only matters if this block is returned by LatestBlock() - SafeBlockNumber: head.Number, - CreatedAt: head.CreatedAt, - } + lpBlock := finalizedHeadToBlock(head) logPollerBlocks[uint64(head.Number)] = lpBlock //nolint:gosec // G115 if chainValidationHead == nil || chainValidationHead.BlockNumber < lpBlock.BlockNumber { chainValidationHead = &lpBlock @@ -1889,6 +1887,18 @@ func (lp *logPoller) batchFetchBlocks(ctx context.Context, blocksRequested []uin return logPollerBlocks, nil } +func finalizedHeadToBlock(head *evmtypes.Head) Block { + return Block{ + EVMChainID: head.EVMChainID, + BlockHash: head.Hash, + BlockNumber: head.Number, + BlockTimestamp: head.Timestamp, + FinalizedBlockNumber: head.Number, // always finalized; only matters if this block is returned by LatestBlock() + SafeBlockNumber: head.Number, + CreatedAt: head.CreatedAt, + } +} + func ensureIdenticalBlocksBatches(fetched1, fetched2 map[uint64]*evmtypes.Head) error { if len(fetched1) != len(fetched2) { return fmt.Errorf("invariant violation: expected size of batches to be identical. Fetched1: %d, Fetched2: %d", len(fetched1), len(fetched2)) @@ -1918,16 +1928,27 @@ func validateBlockResponse(r rpc.BatchElem) (*evmtypes.Head, error) { if !is { return nil, pkgerrors.Errorf("expected result to be a %T, got %T", &evmtypes.Head{}, r.Result) } + + err := validateHead(block) + if err != nil { + return nil, err + } + + return block, nil +} + +func validateHead(block *evmtypes.Head) error { if block == nil { - return nil, pkgerrors.New("invariant violation: got nil block") + return pkgerrors.New("invariant violation: got nil block") } if block.Hash == (common.Hash{}) { - return nil, pkgerrors.Errorf("missing block hash for block number: %d", block.Number) + return pkgerrors.Errorf("missing block hash for block number: %d", block.Number) } if block.Number < 0 { - return nil, pkgerrors.Errorf("expected block number to be >= to 0, got %d", block.Number) + return pkgerrors.Errorf("expected block number to be >= to 0, got %d", block.Number) } - return block, nil + + return nil } // IndexedLogsWithSigsExcluding returns the set difference(A-B) of logs with signature sigA and sigB, matching is done on the topics index @@ -1943,6 +1964,51 @@ func (lp *logPoller) DeleteLogsAndBlocksAfter(ctx context.Context, start int64) return lp.orm.DeleteLogsAndBlocksAfter(ctx, start) } +// SkipToBlock repositions the poller to start processing from blockNumber (T). +// Block T-1 is fetched from RPC and saved as the anchor; the next poll resumes from T. +// Returns an error if T-1 is not finalized on chain. +func (lp *logPoller) SkipToBlock(ctx context.Context, blockNumber int64) error { + if blockNumber < 2 { + return fmt.Errorf("invalid skip block number %d, must be >= 2", blockNumber) + } + + lp.runMu.Lock() + defer lp.runMu.Unlock() + + anchorBlockNumber := blockNumber - 1 + + anchorHead, err := lp.ec.HeaderByNumberWithOpts(ctx, big.NewInt(anchorBlockNumber), evmtypes.HeaderByNumberOpts{ConfidenceLevel: primitives.Finalized}) + if err != nil { + return fmt.Errorf("failed to fetch anchor block %d from RPC: %w", anchorBlockNumber, err) + } + + err = validateHead((*evmtypes.Head)(anchorHead)) + if err != nil { + return fmt.Errorf("failed to validate anchor block %d: %w", anchorBlockNumber, err) + } + + anchorBlock := finalizedHeadToBlock((*evmtypes.Head)(anchorHead)) + + latest, err := lp.orm.SelectLatestBlock(ctx) + if err != nil && !pkgerrors.Is(err, sql.ErrNoRows) { + return fmt.Errorf("failed to select latest block: %w", err) + } + + minBlockToPrune := anchorBlockNumber + if latest != nil && latest.FinalizedBlockNumber < minBlockToPrune { + minBlockToPrune = latest.FinalizedBlockNumber + 1 + } + + err = lp.orm.StoreNewFinalizedCheckpoint(ctx, anchorBlock, minBlockToPrune) + if err != nil { + return fmt.Errorf("failed to store new finalized checkpoint: %w", err) + } + lp.backupPollerNextBlock = anchorBlockNumber + + lp.lggr.Infof("Skipped to block %d. Next poll starts at %d", anchorBlockNumber, blockNumber) + return nil +} + func (lp *logPoller) FindLCA(ctx context.Context) (*Block, error) { latest, err := lp.orm.SelectLatestBlock(ctx) if err != nil { diff --git a/pkg/logpoller/log_poller_internal_test.go b/pkg/logpoller/log_poller_internal_test.go index 6bb7db2ee2..d571b2fa15 100644 --- a/pkg/logpoller/log_poller_internal_test.go +++ b/pkg/logpoller/log_poller_internal_test.go @@ -31,6 +31,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/services/servicetest" "github.com/smartcontractkit/chainlink-common/pkg/sqlutil" commontypes "github.com/smartcontractkit/chainlink-common/pkg/types" + "github.com/smartcontractkit/chainlink-common/pkg/types/query/primitives" "github.com/smartcontractkit/chainlink-evm/pkg/client/clienttest" "github.com/smartcontractkit/chainlink-evm/pkg/heads/headstest" "github.com/smartcontractkit/chainlink-evm/pkg/logpoller/internal/log_emitter" @@ -2147,3 +2148,151 @@ func BenchmarkFilter100_10(b *testing.B) { func BenchmarkFilter1000_100(b *testing.B) { benchmarkFilter(b, 1000, 100, 100) } + +func skipToBlockOpts() Opts { + return Opts{ + UseFinalityTag: false, + FinalityDepth: 2, + BackfillBatchSize: 3, + RPCBatchSize: 2, + KeepFinalizedBlocksDepth: 1000, + PollPeriod: time.Hour, + } +} + +func expectFinalizedAnchorHeader(t *testing.T, ec *clienttest.Client, anchorBlockNumber int64) { + t.Helper() + ec.EXPECT().HeaderByNumberWithOpts( + mock.Anything, + big.NewInt(anchorBlockNumber), + evmtypes.HeaderByNumberOpts{ConfidenceLevel: primitives.Finalized}, + ).RunAndReturn(func(_ context.Context, _ *big.Int, _ evmtypes.HeaderByNumberOpts) (*evmtypes.Header, error) { + head := newHeadVal(anchorBlockNumber) + hdr := evmtypes.Header(head) + return &hdr, nil + }).Once() +} + +func expectStoreNewFinalizedCheckpoint(t *testing.T, orm *MockORM, anchorBlockNumber, minBlockToPrune int64) { + t.Helper() + orm.EXPECT().StoreNewFinalizedCheckpoint( + mock.Anything, + mock.MatchedBy(func(b Block) bool { + return b.BlockNumber == anchorBlockNumber && + b.BlockHash == hashOf(anchorBlockNumber) && + b.FinalizedBlockNumber == anchorBlockNumber && + b.SafeBlockNumber == anchorBlockNumber + }), + minBlockToPrune, + ).Return(nil).Once() +} + +func Test_skipToBlock(t *testing.T) { + t.Parallel() + ctx := testutils.Context(t) + + t.Run("invalid block number", func(t *testing.T) { + t.Parallel() + m := newMockedLP(t, skipToBlockOpts()) + err := m.LP.SkipToBlock(ctx, 1) + require.ErrorContains(t, err, "must be >= 2") + }) + + t.Run("disabled log poller returns error", func(t *testing.T) { + t.Parallel() + err := LogPollerDisabled.SkipToBlock(ctx, 2) + require.ErrorIs(t, err, ErrDisabled) + }) + + t.Run("RPC fetch error", func(t *testing.T) { + t.Parallel() + m := newMockedLP(t, skipToBlockOpts()) + m.Client.EXPECT().HeaderByNumberWithOpts( + mock.Anything, + big.NewInt(2), + evmtypes.HeaderByNumberOpts{ConfidenceLevel: primitives.Finalized}, + ).Return(nil, errors.New("rpc unavailable")).Once() + + err := m.LP.SkipToBlock(ctx, 3) + require.ErrorContains(t, err, "failed to fetch anchor block 2 from RPC") + require.ErrorContains(t, err, "rpc unavailable") + }) + + t.Run("invalid anchor header", func(t *testing.T) { + t.Parallel() + m := newMockedLP(t, skipToBlockOpts()) + m.Client.EXPECT().HeaderByNumberWithOpts( + mock.Anything, + big.NewInt(2), + evmtypes.HeaderByNumberOpts{ConfidenceLevel: primitives.Finalized}, + ).RunAndReturn(func(_ context.Context, _ *big.Int, _ evmtypes.HeaderByNumberOpts) (*evmtypes.Header, error) { + head := newHeadVal(2) + head.Hash = common.Hash{} + return (*evmtypes.Header)(&head), nil + }).Once() + + err := m.LP.SkipToBlock(ctx, 3) + require.ErrorContains(t, err, "failed to validate anchor block 2") + }) + + t.Run("forward skip on empty DB", func(t *testing.T) { + t.Parallel() + m := newMockedLP(t, skipToBlockOpts()) + expectFinalizedAnchorHeader(t, m.Client, 2) + m.ORM.EXPECT().SelectLatestBlock(mock.Anything).Return(nil, sql.ErrNoRows).Once() + expectStoreNewFinalizedCheckpoint(t, m.ORM, 2, 2) + + require.NoError(t, m.LP.SkipToBlock(ctx, 3)) + assert.Equal(t, int64(2), m.LP.backupPollerNextBlock) + }) + + t.Run("Happy path", func(t *testing.T) { + t.Parallel() + m := newMockedLP(t, skipToBlockOpts()) + expectFinalizedAnchorHeader(t, m.Client, 5) + m.ORM.EXPECT().SelectLatestBlock(mock.Anything).Return(&Block{ + BlockNumber: 4, + FinalizedBlockNumber: 2, + SafeBlockNumber: 2, + }, nil).Once() + expectStoreNewFinalizedCheckpoint(t, m.ORM, 5, 3) + + require.NoError(t, m.LP.SkipToBlock(ctx, 6)) + }) + + t.Run("backward skip removes blocks from anchor onward", func(t *testing.T) { + t.Parallel() + m := newMockedLP(t, skipToBlockOpts()) + expectFinalizedAnchorHeader(t, m.Client, 2) + m.ORM.EXPECT().SelectLatestBlock(mock.Anything).Return(&Block{ + BlockNumber: 4, + FinalizedBlockNumber: 3, + SafeBlockNumber: 3, + }, nil).Once() + expectStoreNewFinalizedCheckpoint(t, m.ORM, 2, 2) + + require.NoError(t, m.LP.SkipToBlock(ctx, 3)) + }) + + t.Run("select latest block error", func(t *testing.T) { + t.Parallel() + m := newMockedLP(t, skipToBlockOpts()) + expectFinalizedAnchorHeader(t, m.Client, 2) + m.ORM.EXPECT().SelectLatestBlock(mock.Anything).Return(nil, errors.New("db unavailable")).Once() + + err := m.LP.SkipToBlock(ctx, 3) + require.ErrorContains(t, err, "failed to select latest block") + }) + + t.Run("store checkpoint error", func(t *testing.T) { + t.Parallel() + m := newMockedLP(t, skipToBlockOpts()) + expectFinalizedAnchorHeader(t, m.Client, 2) + m.ORM.EXPECT().SelectLatestBlock(mock.Anything).Return(nil, sql.ErrNoRows).Once() + m.ORM.EXPECT().StoreNewFinalizedCheckpoint(mock.Anything, mock.Anything, int64(2)). + Return(errors.New("db write failed")).Once() + + err := m.LP.SkipToBlock(ctx, 3) + require.ErrorContains(t, err, "failed to store new finalized checkpoint") + }) +} diff --git a/pkg/logpoller/mock_orm_test.go b/pkg/logpoller/mock_orm_test.go index 419bc87470..b7a6ad742a 100644 --- a/pkg/logpoller/mock_orm_test.go +++ b/pkg/logpoller/mock_orm_test.go @@ -1460,6 +1460,54 @@ func (_c *MockORM_SelectLatestFinalizedBlock_Call) RunAndReturn(run func(context return _c } +// StoreNewFinalizedCheckpoint provides a mock function with given fields: ctx, block, minBlockToPrune +func (_m *MockORM) StoreNewFinalizedCheckpoint(ctx context.Context, block Block, minBlockToPrune int64) error { + ret := _m.Called(ctx, block, minBlockToPrune) + + if len(ret) == 0 { + panic("no return value specified for StoreNewFinalizedCheckpoint") + } + + var r0 error + if rf, ok := ret.Get(0).(func(context.Context, Block, int64) error); ok { + r0 = rf(ctx, block, minBlockToPrune) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// MockORM_StoreNewFinalizedCheckpoint_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'StoreNewFinalizedCheckpoint' +type MockORM_StoreNewFinalizedCheckpoint_Call struct { + *mock.Call +} + +// StoreNewFinalizedCheckpoint is a helper method to define mock.On call +// - ctx context.Context +// - block Block +// - minBlockToPrune int64 +func (_e *MockORM_Expecter) StoreNewFinalizedCheckpoint(ctx interface{}, block interface{}, minBlockToPrune interface{}) *MockORM_StoreNewFinalizedCheckpoint_Call { + return &MockORM_StoreNewFinalizedCheckpoint_Call{Call: _e.mock.On("StoreNewFinalizedCheckpoint", ctx, block, minBlockToPrune)} +} + +func (_c *MockORM_StoreNewFinalizedCheckpoint_Call) Run(run func(ctx context.Context, block Block, minBlockToPrune int64)) *MockORM_StoreNewFinalizedCheckpoint_Call { + _c.Call.Run(func(args mock.Arguments) { + run(args[0].(context.Context), args[1].(Block), args[2].(int64)) + }) + return _c +} + +func (_c *MockORM_StoreNewFinalizedCheckpoint_Call) Return(_a0 error) *MockORM_StoreNewFinalizedCheckpoint_Call { + _c.Call.Return(_a0) + return _c +} + +func (_c *MockORM_StoreNewFinalizedCheckpoint_Call) RunAndReturn(run func(context.Context, Block, int64) error) *MockORM_StoreNewFinalizedCheckpoint_Call { + _c.Call.Return(run) + return _c +} + // SelectLatestLogByEventSigWithConfs provides a mock function with given fields: ctx, eventSig, address, confs func (_m *MockORM) SelectLatestLogByEventSigWithConfs(ctx context.Context, eventSig common.Hash, address common.Address, confs types.Confirmations) (*Log, error) { ret := _m.Called(ctx, eventSig, address, confs) diff --git a/pkg/logpoller/mocks/log_poller.go b/pkg/logpoller/mocks/log_poller.go index 1ea8acabcd..d411ca24bb 100644 --- a/pkg/logpoller/mocks/log_poller.go +++ b/pkg/logpoller/mocks/log_poller.go @@ -1758,6 +1758,53 @@ func (_c *LogPoller_ReplayAsync_Call) RunAndReturn(run func(int64)) *LogPoller_R return _c } +// SkipToBlock provides a mock function with given fields: ctx, blockNumber +func (_m *LogPoller) SkipToBlock(ctx context.Context, blockNumber int64) error { + ret := _m.Called(ctx, blockNumber) + + if len(ret) == 0 { + panic("no return value specified for SkipToBlock") + } + + var r0 error + if rf, ok := ret.Get(0).(func(context.Context, int64) error); ok { + r0 = rf(ctx, blockNumber) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// LogPoller_SkipToBlock_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'SkipToBlock' +type LogPoller_SkipToBlock_Call struct { + *mock.Call +} + +// SkipToBlock is a helper method to define mock.On call +// - ctx context.Context +// - blockNumber int64 +func (_e *LogPoller_Expecter) SkipToBlock(ctx interface{}, blockNumber interface{}) *LogPoller_SkipToBlock_Call { + return &LogPoller_SkipToBlock_Call{Call: _e.mock.On("SkipToBlock", ctx, blockNumber)} +} + +func (_c *LogPoller_SkipToBlock_Call) Run(run func(ctx context.Context, blockNumber int64)) *LogPoller_SkipToBlock_Call { + _c.Call.Run(func(args mock.Arguments) { + run(args[0].(context.Context), args[1].(int64)) + }) + return _c +} + +func (_c *LogPoller_SkipToBlock_Call) Return(_a0 error) *LogPoller_SkipToBlock_Call { + _c.Call.Return(_a0) + return _c +} + +func (_c *LogPoller_SkipToBlock_Call) RunAndReturn(run func(context.Context, int64) error) *LogPoller_SkipToBlock_Call { + _c.Call.Return(run) + return _c +} + // Start provides a mock function with given fields: _a0 func (_m *LogPoller) Start(_a0 context.Context) error { ret := _m.Called(_a0) diff --git a/pkg/logpoller/observability.go b/pkg/logpoller/observability.go index a1d4fdd3a3..6bbf2eae70 100644 --- a/pkg/logpoller/observability.go +++ b/pkg/logpoller/observability.go @@ -102,6 +102,12 @@ func (o *ObservedORM) DeleteExpiredLogs(ctx context.Context, limit int64) (int64 }) } +func (o *ObservedORM) StoreNewFinalizedCheckpoint(ctx context.Context, block Block, minBlockToPrune int64) error { + return withObservedExec(ctx, o, "StoreNewFinalizedCheckpoint", metrics.Create, func() error { + return o.ORM.StoreNewFinalizedCheckpoint(ctx, block, minBlockToPrune) + }) +} + func (o *ObservedORM) SelectUnmatchedLogIDs(ctx context.Context, limit int64) (ids []uint64, err error) { return withObservedQueryAndResults[uint64](ctx, o, "SelectUnmatchedLogIDs", func() ([]uint64, error) { return o.ORM.SelectUnmatchedLogIDs(ctx, limit) diff --git a/pkg/logpoller/orm.go b/pkg/logpoller/orm.go index ca0bf0899c..25d40ba939 100644 --- a/pkg/logpoller/orm.go +++ b/pkg/logpoller/orm.go @@ -67,6 +67,8 @@ type ORM interface { // FilteredLogs accepts chainlink-common filtering DSL. FilteredLogs(ctx context.Context, filter []query.Expression, limitAndSort query.LimitAndSort, queryName string) ([]Log, error) + + StoreNewFinalizedCheckpoint(ctx context.Context, block Block, minBlockToPrune int64) error } type DSORM struct { @@ -114,6 +116,10 @@ func (o *DSORM) InsertBlock(ctx context.Context, blockHash common.Hash, blockNum } func (o *DSORM) InsertBlocks(ctx context.Context, blocks []Block) error { + return insertBlocks(ctx, o, blocks) +} + +func insertBlocks(ctx context.Context, o *DSORM, blocks []Block) error { const q = `INSERT INTO evm.log_poller_blocks (evm_chain_id, block_hash, block_number, block_timestamp, finalized_block_number, created_at, safe_block_number) VALUES (:evm_chain_id, :block_hash, :block_number, :block_timestamp, :finalized_block_number, NOW(), :safe_block_number) @@ -411,38 +417,57 @@ func (o *DSORM) DeleteBlocksBefore(ctx context.Context, end int64, limit int64) return q.ExecPagedQuery(ctx, limit, end) } +func (o *DSORM) StoreNewFinalizedCheckpoint(ctx context.Context, block Block, minBlockToPrune int64) error { + return o.Transact(ctx, func(orm *DSORM) error { + err := deleteLogsAndBlockAfter(ctx, orm, minBlockToPrune) + if err != nil { + return fmt.Errorf("failed to delete logs and block after minBlockToPrune %d: %w", minBlockToPrune, err) + } + + err = insertBlocks(ctx, orm, []Block{block}) + if err != nil { + return fmt.Errorf("failed to insert new finalized checkpoint block %d: %w", block.BlockNumber, err) + } + return nil + }) +} + func (o *DSORM) DeleteLogsAndBlocksAfter(ctx context.Context, start int64) error { // These deletes are bounded by reorg depth, so they are // fast and should not slow down the log readers. return o.Transact(ctx, func(orm *DSORM) error { - // Applying upper bound filter is critical for Postgres performance (especially for evm.logs table) - // because it allows the planner to properly estimate the number of rows to be scanned. - // If not applied, these queries can become very slow. After some critical number - // of logs, Postgres will try to scan all the logs in the index by block_number. - // Latency without upper bound filter can be orders of magnitude higher for large number of logs. - _, err := orm.ds.ExecContext(ctx, `DELETE FROM evm.log_poller_blocks + return deleteLogsAndBlockAfter(ctx, orm, start) + }) +} + +func deleteLogsAndBlockAfter(ctx context.Context, orm *DSORM, start int64) error { + // Applying upper bound filter is critical for Postgres performance (especially for evm.logs table) + // because it allows the planner to properly estimate the number of rows to be scanned. + // If not applied, these queries can become very slow. After some critical number + // of logs, Postgres will try to scan all the logs in the index by block_number. + // Latency without upper bound filter can be orders of magnitude higher for large number of logs. + _, err := orm.ds.ExecContext(ctx, `DELETE FROM evm.log_poller_blocks WHERE evm_chain_id = $1 AND block_number >= $2 AND block_number <= (SELECT MAX(block_number) FROM evm.log_poller_blocks WHERE evm_chain_id = $1)`, - sqlutil.New(o.chainID), start) - if err != nil { - o.lggr.Warnw("Unable to clear reorged blocks, retrying", "err", err) - return err - } + sqlutil.New(orm.chainID), start) + if err != nil { + orm.lggr.Warnw("Unable to clear reorged blocks, retrying", "err", err) + return err + } - _, err = orm.ds.ExecContext(ctx, `DELETE FROM evm.logs + _, err = orm.ds.ExecContext(ctx, `DELETE FROM evm.logs WHERE evm_chain_id = $1 AND block_number >= $2 AND block_number <= (SELECT MAX(block_number) FROM evm.logs WHERE evm_chain_id = $1)`, - sqlutil.New(o.chainID), start) - if err != nil { - o.lggr.Warnw("Unable to clear reorged logs, retrying", "err", err) - return err - } - return nil - }) + sqlutil.New(orm.chainID), start) + if err != nil { + orm.lggr.Warnw("Unable to clear reorged logs, retrying", "err", err) + return err + } + return nil } type Exp struct { diff --git a/pkg/logpoller/orm_test.go b/pkg/logpoller/orm_test.go index ac0ee97113..a5ab5ce762 100644 --- a/pkg/logpoller/orm_test.go +++ b/pkg/logpoller/orm_test.go @@ -2505,7 +2505,8 @@ func TestSelectLatestFinalizedBlock(t *testing.T) { type execCountingSQLxDB struct { *sqlx.DB - poolExecContext atomic.Int32 + poolExecContext atomic.Int32 + poolNamedExecContext atomic.Int32 } func (d *execCountingSQLxDB) PoolExecContextCount() int32 { @@ -2517,6 +2518,15 @@ func (d *execCountingSQLxDB) ExecContext(ctx context.Context, query string, args return d.DB.ExecContext(ctx, query, args...) } +func (d *execCountingSQLxDB) NamedExecContext(ctx context.Context, query string, arg any) (sql.Result, error) { + d.poolNamedExecContext.Add(1) + return d.DB.NamedExecContext(ctx, query, arg) +} + +func (d *execCountingSQLxDB) PoolDirectOpCount() int32 { + return d.poolExecContext.Load() + d.poolNamedExecContext.Load() +} + var _ sqlutil.DataSource = (*execCountingSQLxDB)(nil) func TestDSORM_DeleteLogsAndBlocksAfter_usesTransactionalDataSource(t *testing.T) { @@ -2537,3 +2547,134 @@ func TestDSORM_DeleteLogsAndBlocksAfter_usesTransactionalDataSource(t *testing.T "DeleteLogsAndBlocksAfter must run DELETEs on the transaction DataSource (orm.ds), not the outer pool; "+ "otherwise the two deletes are not atomic and this counter would be 2") } + +func insertBlockRangeWithLogs( + t *testing.T, + ctx context.Context, + th TestHarness, + orm logpoller.ORM, + start, end int64, +) { + t.Helper() + + var blocks []logpoller.Block + var logs []logpoller.Log + for i := start; i <= end; i++ { + hash := common.HexToHash(fmt.Sprintf("0x%04x", i)) + blocks = append(blocks, logpoller.Block{ + EVMChainID: sqlutil.New(th.ChainID), + BlockHash: hash, + BlockNumber: i, + BlockTimestamp: time.Unix(i*10, 0), + FinalizedBlockNumber: i - 2, + SafeBlockNumber: i - 2, + }) + logs = append(logs, GenLog(th.ChainID, 0, i, hash.String(), EmitterABI.Events["Log1"].ID.Bytes(), th.EmitterAddress1)) + } + require.NoError(t, orm.InsertLogsWithBlocks(ctx, logs, blocks)) +} + +func finalizedCheckpointBlock(chainID *big.Int, blockNumber int64, hash common.Hash) logpoller.Block { + return logpoller.Block{ + EVMChainID: sqlutil.New(chainID), + BlockHash: hash, + BlockNumber: blockNumber, + BlockTimestamp: time.Unix(blockNumber*10, 0), + FinalizedBlockNumber: blockNumber, + SafeBlockNumber: blockNumber, + } +} + +func requireNoBlock(t *testing.T, ctx context.Context, orm logpoller.ORM, blockNumber int64) { + t.Helper() + _, err := orm.SelectBlockByNumber(ctx, blockNumber) + require.Error(t, err) + require.True(t, pkgerrors.Is(err, sql.ErrNoRows)) +} + +func TestDSORM_StoreNewFinalizedCheckpoint(t *testing.T) { + t.Parallel() + testutils.SkipShortDB(t) + + ctx := testutils.Context(t) + + t.Run("forward skip prunes from minBlockToPrune and stores anchor", func(t *testing.T) { + th := SetupTH(t, lpOpts) + o1 := th.ORM + insertBlockRangeWithLogs(t, ctx, th, o1, 1, 4) + + anchor := finalizedCheckpointBlock(th.ChainID, 5, common.HexToHash("0x0005")) + require.NoError(t, o1.StoreNewFinalizedCheckpoint(ctx, anchor, 3)) + + b, err := o1.SelectBlockByNumber(ctx, 5) + require.NoError(t, err) + require.Equal(t, int64(5), b.BlockNumber) + require.Equal(t, common.HexToHash("0x0005"), b.BlockHash) + require.Equal(t, int64(5), b.FinalizedBlockNumber) + + for _, n := range []int64{1, 2} { + _, err := o1.SelectBlockByNumber(ctx, n) + require.NoError(t, err) + } + for _, n := range []int64{3, 4} { + requireNoBlock(t, ctx, o1, n) + } + + logs, err := o1.SelectLogsByBlockRange(ctx, 1, 5) + require.NoError(t, err) + require.Len(t, logs, 2) + for _, lg := range logs { + require.Less(t, lg.BlockNumber, int64(3)) + } + }) + + t.Run("empty DB stores anchor only", func(t *testing.T) { + th := SetupTH(t, lpOpts) + o1 := th.ORM + anchor := finalizedCheckpointBlock(th.ChainID, 5, common.HexToHash("0x0005")) + require.NoError(t, o1.StoreNewFinalizedCheckpoint(ctx, anchor, 5)) + + b, err := o1.SelectBlockByNumber(ctx, 5) + require.NoError(t, err) + require.Equal(t, int64(5), b.BlockNumber) + + logs, err := o1.SelectLogsByBlockRange(ctx, 1, 5) + require.NoError(t, err) + require.Empty(t, logs) + }) + + t.Run("does not affect other chain", func(t *testing.T) { + th := SetupTH(t, lpOpts) + o1 := th.ORM + o2 := th.ORM2 + insertBlockRangeWithLogs(t, ctx, th, o1, 1, 4) + require.NoError(t, o2.InsertBlock(ctx, common.HexToHash("0x9999"), 10, time.Now(), 10, 10)) + + anchor := finalizedCheckpointBlock(th.ChainID, 2, common.HexToHash("0x0002")) + require.NoError(t, o1.StoreNewFinalizedCheckpoint(ctx, anchor, 2)) + + b, err := o2.SelectBlockByNumber(ctx, 10) + require.NoError(t, err) + require.Equal(t, int64(10), b.BlockNumber) + }) +} + +func TestDSORM_StoreNewFinalizedCheckpoint_usesTransactionalDataSource(t *testing.T) { + t.Parallel() + testutils.SkipShortDB(t) + + ctx := testutils.Context(t) + base := testutils.NewSqlxDB(t) + require.NotNil(t, base) + wrapped := &execCountingSQLxDB{DB: base} + + chainID := testutils.NewRandomEVMChainID() + orm := logpoller.NewORM(chainID, wrapped, logger.Test(t)) + + anchor := finalizedCheckpointBlock(chainID, 5, common.HexToHash("0x0005")) + require.NoError(t, orm.StoreNewFinalizedCheckpoint(ctx, anchor, 5)) + + require.Zero(t, wrapped.PoolDirectOpCount(), + "StoreNewFinalizedCheckpoint must run DELETEs and INSERT on the transaction DataSource (orm.ds), not the outer pool; "+ + "otherwise the prune and checkpoint write are not atomic and this counter would be 3") +} From 559439dd0d3f7eba543e9c016424e670fdc89311 Mon Sep 17 00:00:00 2001 From: Dmytro Haidashenko Date: Mon, 13 Jul 2026 18:14:38 +0200 Subject: [PATCH 03/12] Implement EVM service --- pkg/relay/evm_service.go | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/pkg/relay/evm_service.go b/pkg/relay/evm_service.go index 7b84112dda..d9e4defdfd 100644 --- a/pkg/relay/evm_service.go +++ b/pkg/relay/evm_service.go @@ -373,6 +373,13 @@ func (e *evmService) GetLatestLPBlock(ctx context.Context) (*evm.LPBlock, error) }, nil } +func (e *evmService) LPSkipToBlock(ctx context.Context, blockNumber int64) error { + if err := e.chain.LogPoller().SkipToBlock(ctx, blockNumber); err != nil { + return fmt.Errorf("failed to skip log poller to block %d: %w", blockNumber, err) + } + return nil +} + func (r *Relayer) GetForwarderForEOA(ctx context.Context, eoa, ocr2AggregatorID evm.Address, pluginType string) (forwarder evm.Address, err error) { if pluginType == string(commontypes.Median) { return r.chain.TxManager().GetForwarderForEOAOCR2Feeds(ctx, eoa, ocr2AggregatorID) From f12b12e11d300d91c92ac2d3351f062503e9f958 Mon Sep 17 00:00:00 2001 From: Dmytro Haidashenko Date: Mon, 13 Jul 2026 19:37:21 +0200 Subject: [PATCH 04/12] Improve error handling --- pkg/logpoller/log_poller.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/pkg/logpoller/log_poller.go b/pkg/logpoller/log_poller.go index c662c6ef5b..d93b29ca62 100644 --- a/pkg/logpoller/log_poller.go +++ b/pkg/logpoller/log_poller.go @@ -760,6 +760,7 @@ func (lp *logPoller) tickPoll(ctx context.Context) { start = latestFinalizedBlockNumber } else { lp.metrics.RecordLastProcessedBlock(ctx, lastProcessed.BlockNumber) + lp.lggr.Debugw("New poll tick", "last_processed", lastProcessed.BlockNumber, "finalized_block", lastProcessed.FinalizedBlockNumber) start = lastProcessed.BlockNumber + 1 } lp.PollAndSaveLogs(ctx, start, false) @@ -1974,6 +1975,9 @@ func (lp *logPoller) SkipToBlock(ctx context.Context, blockNumber int64) error { lp.runMu.Lock() defer lp.runMu.Unlock() + if ctx.Err() != nil { + return fmt.Errorf("aborted, SkipToBlock request cancelled: %w", ctx.Err()) + } anchorBlockNumber := blockNumber - 1 From 85ee8c48d7652ca6f4baa14ff52ed6deea8739ba Mon Sep 17 00:00:00 2001 From: Dmytro Haidashenko Date: Mon, 13 Jul 2026 19:54:16 +0200 Subject: [PATCH 05/12] make generate --- pkg/logpoller/mock_orm_test.go | 96 +++++++++++++++++----------------- 1 file changed, 48 insertions(+), 48 deletions(-) diff --git a/pkg/logpoller/mock_orm_test.go b/pkg/logpoller/mock_orm_test.go index b7a6ad742a..aaab838864 100644 --- a/pkg/logpoller/mock_orm_test.go +++ b/pkg/logpoller/mock_orm_test.go @@ -1460,54 +1460,6 @@ func (_c *MockORM_SelectLatestFinalizedBlock_Call) RunAndReturn(run func(context return _c } -// StoreNewFinalizedCheckpoint provides a mock function with given fields: ctx, block, minBlockToPrune -func (_m *MockORM) StoreNewFinalizedCheckpoint(ctx context.Context, block Block, minBlockToPrune int64) error { - ret := _m.Called(ctx, block, minBlockToPrune) - - if len(ret) == 0 { - panic("no return value specified for StoreNewFinalizedCheckpoint") - } - - var r0 error - if rf, ok := ret.Get(0).(func(context.Context, Block, int64) error); ok { - r0 = rf(ctx, block, minBlockToPrune) - } else { - r0 = ret.Error(0) - } - - return r0 -} - -// MockORM_StoreNewFinalizedCheckpoint_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'StoreNewFinalizedCheckpoint' -type MockORM_StoreNewFinalizedCheckpoint_Call struct { - *mock.Call -} - -// StoreNewFinalizedCheckpoint is a helper method to define mock.On call -// - ctx context.Context -// - block Block -// - minBlockToPrune int64 -func (_e *MockORM_Expecter) StoreNewFinalizedCheckpoint(ctx interface{}, block interface{}, minBlockToPrune interface{}) *MockORM_StoreNewFinalizedCheckpoint_Call { - return &MockORM_StoreNewFinalizedCheckpoint_Call{Call: _e.mock.On("StoreNewFinalizedCheckpoint", ctx, block, minBlockToPrune)} -} - -func (_c *MockORM_StoreNewFinalizedCheckpoint_Call) Run(run func(ctx context.Context, block Block, minBlockToPrune int64)) *MockORM_StoreNewFinalizedCheckpoint_Call { - _c.Call.Run(func(args mock.Arguments) { - run(args[0].(context.Context), args[1].(Block), args[2].(int64)) - }) - return _c -} - -func (_c *MockORM_StoreNewFinalizedCheckpoint_Call) Return(_a0 error) *MockORM_StoreNewFinalizedCheckpoint_Call { - _c.Call.Return(_a0) - return _c -} - -func (_c *MockORM_StoreNewFinalizedCheckpoint_Call) RunAndReturn(run func(context.Context, Block, int64) error) *MockORM_StoreNewFinalizedCheckpoint_Call { - _c.Call.Return(run) - return _c -} - // SelectLatestLogByEventSigWithConfs provides a mock function with given fields: ctx, eventSig, address, confs func (_m *MockORM) SelectLatestLogByEventSigWithConfs(ctx context.Context, eventSig common.Hash, address common.Address, confs types.Confirmations) (*Log, error) { ret := _m.Called(ctx, eventSig, address, confs) @@ -2245,6 +2197,54 @@ func (_c *MockORM_SelectUnmatchedLogIDs_Call) RunAndReturn(run func(context.Cont return _c } +// StoreNewFinalizedCheckpoint provides a mock function with given fields: ctx, block, minBlockToPrune +func (_m *MockORM) StoreNewFinalizedCheckpoint(ctx context.Context, block Block, minBlockToPrune int64) error { + ret := _m.Called(ctx, block, minBlockToPrune) + + if len(ret) == 0 { + panic("no return value specified for StoreNewFinalizedCheckpoint") + } + + var r0 error + if rf, ok := ret.Get(0).(func(context.Context, Block, int64) error); ok { + r0 = rf(ctx, block, minBlockToPrune) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// MockORM_StoreNewFinalizedCheckpoint_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'StoreNewFinalizedCheckpoint' +type MockORM_StoreNewFinalizedCheckpoint_Call struct { + *mock.Call +} + +// StoreNewFinalizedCheckpoint is a helper method to define mock.On call +// - ctx context.Context +// - block Block +// - minBlockToPrune int64 +func (_e *MockORM_Expecter) StoreNewFinalizedCheckpoint(ctx interface{}, block interface{}, minBlockToPrune interface{}) *MockORM_StoreNewFinalizedCheckpoint_Call { + return &MockORM_StoreNewFinalizedCheckpoint_Call{Call: _e.mock.On("StoreNewFinalizedCheckpoint", ctx, block, minBlockToPrune)} +} + +func (_c *MockORM_StoreNewFinalizedCheckpoint_Call) Run(run func(ctx context.Context, block Block, minBlockToPrune int64)) *MockORM_StoreNewFinalizedCheckpoint_Call { + _c.Call.Run(func(args mock.Arguments) { + run(args[0].(context.Context), args[1].(Block), args[2].(int64)) + }) + return _c +} + +func (_c *MockORM_StoreNewFinalizedCheckpoint_Call) Return(_a0 error) *MockORM_StoreNewFinalizedCheckpoint_Call { + _c.Call.Return(_a0) + return _c +} + +func (_c *MockORM_StoreNewFinalizedCheckpoint_Call) RunAndReturn(run func(context.Context, Block, int64) error) *MockORM_StoreNewFinalizedCheckpoint_Call { + _c.Call.Return(run) + return _c +} + // NewMockORM creates a new instance of MockORM. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. // The first argument is typically a *testing.T value. func NewMockORM(t interface { From 8b39b14aa585a516b72af8154be90731c74e8870 Mon Sep 17 00:00:00 2001 From: Dmytro Haidashenko Date: Mon, 13 Jul 2026 20:07:13 +0200 Subject: [PATCH 06/12] linter fixes --- pkg/logpoller/log_poller_internal_test.go | 5 ++-- pkg/logpoller/orm_test.go | 35 +++++++++++------------ 2 files changed, 18 insertions(+), 22 deletions(-) diff --git a/pkg/logpoller/log_poller_internal_test.go b/pkg/logpoller/log_poller_internal_test.go index d571b2fa15..a9f2f64f39 100644 --- a/pkg/logpoller/log_poller_internal_test.go +++ b/pkg/logpoller/log_poller_internal_test.go @@ -2167,9 +2167,8 @@ func expectFinalizedAnchorHeader(t *testing.T, ec *clienttest.Client, anchorBloc big.NewInt(anchorBlockNumber), evmtypes.HeaderByNumberOpts{ConfidenceLevel: primitives.Finalized}, ).RunAndReturn(func(_ context.Context, _ *big.Int, _ evmtypes.HeaderByNumberOpts) (*evmtypes.Header, error) { - head := newHeadVal(anchorBlockNumber) - hdr := evmtypes.Header(head) - return &hdr, nil + head := newHead(anchorBlockNumber) + return (*evmtypes.Header)(head), nil }).Once() } diff --git a/pkg/logpoller/orm_test.go b/pkg/logpoller/orm_test.go index a5ab5ce762..5d83601780 100644 --- a/pkg/logpoller/orm_test.go +++ b/pkg/logpoller/orm_test.go @@ -2550,7 +2550,6 @@ func TestDSORM_DeleteLogsAndBlocksAfter_usesTransactionalDataSource(t *testing.T func insertBlockRangeWithLogs( t *testing.T, - ctx context.Context, th TestHarness, orm logpoller.ORM, start, end int64, @@ -2571,7 +2570,7 @@ func insertBlockRangeWithLogs( }) logs = append(logs, GenLog(th.ChainID, 0, i, hash.String(), EmitterABI.Events["Log1"].ID.Bytes(), th.EmitterAddress1)) } - require.NoError(t, orm.InsertLogsWithBlocks(ctx, logs, blocks)) + require.NoError(t, orm.InsertLogsWithBlocks(t.Context(), logs, blocks)) } func finalizedCheckpointBlock(chainID *big.Int, blockNumber int64, hash common.Hash) logpoller.Block { @@ -2585,9 +2584,9 @@ func finalizedCheckpointBlock(chainID *big.Int, blockNumber int64, hash common.H } } -func requireNoBlock(t *testing.T, ctx context.Context, orm logpoller.ORM, blockNumber int64) { +func requireNoBlock(t *testing.T, orm logpoller.ORM, blockNumber int64) { t.Helper() - _, err := orm.SelectBlockByNumber(ctx, blockNumber) + _, err := orm.SelectBlockByNumber(t.Context(), blockNumber) require.Error(t, err) require.True(t, pkgerrors.Is(err, sql.ErrNoRows)) } @@ -2596,31 +2595,29 @@ func TestDSORM_StoreNewFinalizedCheckpoint(t *testing.T) { t.Parallel() testutils.SkipShortDB(t) - ctx := testutils.Context(t) - t.Run("forward skip prunes from minBlockToPrune and stores anchor", func(t *testing.T) { th := SetupTH(t, lpOpts) o1 := th.ORM - insertBlockRangeWithLogs(t, ctx, th, o1, 1, 4) + insertBlockRangeWithLogs(t, th, o1, 1, 4) anchor := finalizedCheckpointBlock(th.ChainID, 5, common.HexToHash("0x0005")) - require.NoError(t, o1.StoreNewFinalizedCheckpoint(ctx, anchor, 3)) + require.NoError(t, o1.StoreNewFinalizedCheckpoint(t.Context(), anchor, 3)) - b, err := o1.SelectBlockByNumber(ctx, 5) + b, err := o1.SelectBlockByNumber(t.Context(), 5) require.NoError(t, err) require.Equal(t, int64(5), b.BlockNumber) require.Equal(t, common.HexToHash("0x0005"), b.BlockHash) require.Equal(t, int64(5), b.FinalizedBlockNumber) for _, n := range []int64{1, 2} { - _, err := o1.SelectBlockByNumber(ctx, n) + _, err = o1.SelectBlockByNumber(t.Context(), n) require.NoError(t, err) } for _, n := range []int64{3, 4} { - requireNoBlock(t, ctx, o1, n) + requireNoBlock(t, o1, n) } - logs, err := o1.SelectLogsByBlockRange(ctx, 1, 5) + logs, err := o1.SelectLogsByBlockRange(t.Context(), 1, 5) require.NoError(t, err) require.Len(t, logs, 2) for _, lg := range logs { @@ -2632,13 +2629,13 @@ func TestDSORM_StoreNewFinalizedCheckpoint(t *testing.T) { th := SetupTH(t, lpOpts) o1 := th.ORM anchor := finalizedCheckpointBlock(th.ChainID, 5, common.HexToHash("0x0005")) - require.NoError(t, o1.StoreNewFinalizedCheckpoint(ctx, anchor, 5)) + require.NoError(t, o1.StoreNewFinalizedCheckpoint(t.Context(), anchor, 5)) - b, err := o1.SelectBlockByNumber(ctx, 5) + b, err := o1.SelectBlockByNumber(t.Context(), 5) require.NoError(t, err) require.Equal(t, int64(5), b.BlockNumber) - logs, err := o1.SelectLogsByBlockRange(ctx, 1, 5) + logs, err := o1.SelectLogsByBlockRange(t.Context(), 1, 5) require.NoError(t, err) require.Empty(t, logs) }) @@ -2647,13 +2644,13 @@ func TestDSORM_StoreNewFinalizedCheckpoint(t *testing.T) { th := SetupTH(t, lpOpts) o1 := th.ORM o2 := th.ORM2 - insertBlockRangeWithLogs(t, ctx, th, o1, 1, 4) - require.NoError(t, o2.InsertBlock(ctx, common.HexToHash("0x9999"), 10, time.Now(), 10, 10)) + insertBlockRangeWithLogs(t, th, o1, 1, 4) + require.NoError(t, o2.InsertBlock(t.Context(), common.HexToHash("0x9999"), 10, time.Now(), 10, 10)) anchor := finalizedCheckpointBlock(th.ChainID, 2, common.HexToHash("0x0002")) - require.NoError(t, o1.StoreNewFinalizedCheckpoint(ctx, anchor, 2)) + require.NoError(t, o1.StoreNewFinalizedCheckpoint(t.Context(), anchor, 2)) - b, err := o2.SelectBlockByNumber(ctx, 10) + b, err := o2.SelectBlockByNumber(t.Context(), 10) require.NoError(t, err) require.Equal(t, int64(10), b.BlockNumber) }) From 182fc8d10dbe400ef1d760bdf9ae70a81beea9dd Mon Sep 17 00:00:00 2001 From: Dmytro Haidashenko Date: Mon, 13 Jul 2026 20:39:34 +0200 Subject: [PATCH 07/12] Fix tests --- pkg/logpoller/orm_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pkg/logpoller/orm_test.go b/pkg/logpoller/orm_test.go index 5d83601780..32aca8f25d 100644 --- a/pkg/logpoller/orm_test.go +++ b/pkg/logpoller/orm_test.go @@ -2565,8 +2565,8 @@ func insertBlockRangeWithLogs( BlockHash: hash, BlockNumber: i, BlockTimestamp: time.Unix(i*10, 0), - FinalizedBlockNumber: i - 2, - SafeBlockNumber: i - 2, + FinalizedBlockNumber: max(i-2, 0), + SafeBlockNumber: max(i-2, 0), }) logs = append(logs, GenLog(th.ChainID, 0, i, hash.String(), EmitterABI.Events["Log1"].ID.Bytes(), th.EmitterAddress1)) } From 62074374254abd58723aa75a32cf61d128965652 Mon Sep 17 00:00:00 2001 From: Dmytro Haidashenko Date: Tue, 14 Jul 2026 18:43:28 +0200 Subject: [PATCH 08/12] review improvements --- pkg/logpoller/log_poller.go | 19 +++++++++++-------- pkg/logpoller/log_poller_internal_test.go | 8 +++----- 2 files changed, 14 insertions(+), 13 deletions(-) diff --git a/pkg/logpoller/log_poller.go b/pkg/logpoller/log_poller.go index d93b29ca62..b24eeb89dd 100644 --- a/pkg/logpoller/log_poller.go +++ b/pkg/logpoller/log_poller.go @@ -1973,12 +1973,6 @@ func (lp *logPoller) SkipToBlock(ctx context.Context, blockNumber int64) error { return fmt.Errorf("invalid skip block number %d, must be >= 2", blockNumber) } - lp.runMu.Lock() - defer lp.runMu.Unlock() - if ctx.Err() != nil { - return fmt.Errorf("aborted, SkipToBlock request cancelled: %w", ctx.Err()) - } - anchorBlockNumber := blockNumber - 1 anchorHead, err := lp.ec.HeaderByNumberWithOpts(ctx, big.NewInt(anchorBlockNumber), evmtypes.HeaderByNumberOpts{ConfidenceLevel: primitives.Finalized}) @@ -1993,13 +1987,22 @@ func (lp *logPoller) SkipToBlock(ctx context.Context, blockNumber int64) error { anchorBlock := finalizedHeadToBlock((*evmtypes.Head)(anchorHead)) + lp.runMu.Lock() + defer lp.runMu.Unlock() + if ctx.Err() != nil { + return fmt.Errorf("aborted, SkipToBlock request cancelled: %w", ctx.Err()) + } + latest, err := lp.orm.SelectLatestBlock(ctx) if err != nil && !pkgerrors.Is(err, sql.ErrNoRows) { return fmt.Errorf("failed to select latest block: %w", err) } minBlockToPrune := anchorBlockNumber - if latest != nil && latest.FinalizedBlockNumber < minBlockToPrune { + if latest != nil { + if latest.FinalizedBlockNumber >= minBlockToPrune { + return fmt.Errorf("skipping to a block %d that is older than previously finalized %d is not allowed. Use explicit remove-blocks to delete blocks", anchorBlockNumber, latest.FinalizedBlockNumber) + } minBlockToPrune = latest.FinalizedBlockNumber + 1 } @@ -2007,7 +2010,7 @@ func (lp *logPoller) SkipToBlock(ctx context.Context, blockNumber int64) error { if err != nil { return fmt.Errorf("failed to store new finalized checkpoint: %w", err) } - lp.backupPollerNextBlock = anchorBlockNumber + lp.backupPollerNextBlock = blockNumber lp.lggr.Infof("Skipped to block %d. Next poll starts at %d", anchorBlockNumber, blockNumber) return nil diff --git a/pkg/logpoller/log_poller_internal_test.go b/pkg/logpoller/log_poller_internal_test.go index a9f2f64f39..7190791bdb 100644 --- a/pkg/logpoller/log_poller_internal_test.go +++ b/pkg/logpoller/log_poller_internal_test.go @@ -2242,7 +2242,7 @@ func Test_skipToBlock(t *testing.T) { expectStoreNewFinalizedCheckpoint(t, m.ORM, 2, 2) require.NoError(t, m.LP.SkipToBlock(ctx, 3)) - assert.Equal(t, int64(2), m.LP.backupPollerNextBlock) + assert.Equal(t, int64(3), m.LP.backupPollerNextBlock) }) t.Run("Happy path", func(t *testing.T) { @@ -2259,7 +2259,7 @@ func Test_skipToBlock(t *testing.T) { require.NoError(t, m.LP.SkipToBlock(ctx, 6)) }) - t.Run("backward skip removes blocks from anchor onward", func(t *testing.T) { + t.Run("backward skip is not allowed", func(t *testing.T) { t.Parallel() m := newMockedLP(t, skipToBlockOpts()) expectFinalizedAnchorHeader(t, m.Client, 2) @@ -2268,9 +2268,7 @@ func Test_skipToBlock(t *testing.T) { FinalizedBlockNumber: 3, SafeBlockNumber: 3, }, nil).Once() - expectStoreNewFinalizedCheckpoint(t, m.ORM, 2, 2) - - require.NoError(t, m.LP.SkipToBlock(ctx, 3)) + require.ErrorContains(t, m.LP.SkipToBlock(ctx, 3), "skipping to a block 2 that is older than previously finalized 3 is not allowed. Use explicit remove-blocks to delete blocks") }) t.Run("select latest block error", func(t *testing.T) { From c361787e9ae7fb85208e7c4bc601d2af9603bb20 Mon Sep 17 00:00:00 2001 From: Dmytro Haidashenko Date: Wed, 15 Jul 2026 14:57:50 +0200 Subject: [PATCH 09/12] use t.Context --- pkg/logpoller/log_poller_internal_test.go | 19 +++++++++---------- pkg/logpoller/orm_test.go | 3 +-- pkg/testutils/testutils.go | 1 + 3 files changed, 11 insertions(+), 12 deletions(-) diff --git a/pkg/logpoller/log_poller_internal_test.go b/pkg/logpoller/log_poller_internal_test.go index 7190791bdb..f36d94e2fa 100644 --- a/pkg/logpoller/log_poller_internal_test.go +++ b/pkg/logpoller/log_poller_internal_test.go @@ -2188,18 +2188,17 @@ func expectStoreNewFinalizedCheckpoint(t *testing.T, orm *MockORM, anchorBlockNu func Test_skipToBlock(t *testing.T) { t.Parallel() - ctx := testutils.Context(t) t.Run("invalid block number", func(t *testing.T) { t.Parallel() m := newMockedLP(t, skipToBlockOpts()) - err := m.LP.SkipToBlock(ctx, 1) + err := m.LP.SkipToBlock(t.Context(), 1) require.ErrorContains(t, err, "must be >= 2") }) t.Run("disabled log poller returns error", func(t *testing.T) { t.Parallel() - err := LogPollerDisabled.SkipToBlock(ctx, 2) + err := LogPollerDisabled.SkipToBlock(t.Context(), 2) require.ErrorIs(t, err, ErrDisabled) }) @@ -2212,7 +2211,7 @@ func Test_skipToBlock(t *testing.T) { evmtypes.HeaderByNumberOpts{ConfidenceLevel: primitives.Finalized}, ).Return(nil, errors.New("rpc unavailable")).Once() - err := m.LP.SkipToBlock(ctx, 3) + err := m.LP.SkipToBlock(t.Context(), 3) require.ErrorContains(t, err, "failed to fetch anchor block 2 from RPC") require.ErrorContains(t, err, "rpc unavailable") }) @@ -2230,7 +2229,7 @@ func Test_skipToBlock(t *testing.T) { return (*evmtypes.Header)(&head), nil }).Once() - err := m.LP.SkipToBlock(ctx, 3) + err := m.LP.SkipToBlock(t.Context(), 3) require.ErrorContains(t, err, "failed to validate anchor block 2") }) @@ -2241,7 +2240,7 @@ func Test_skipToBlock(t *testing.T) { m.ORM.EXPECT().SelectLatestBlock(mock.Anything).Return(nil, sql.ErrNoRows).Once() expectStoreNewFinalizedCheckpoint(t, m.ORM, 2, 2) - require.NoError(t, m.LP.SkipToBlock(ctx, 3)) + require.NoError(t, m.LP.SkipToBlock(t.Context(), 3)) assert.Equal(t, int64(3), m.LP.backupPollerNextBlock) }) @@ -2256,7 +2255,7 @@ func Test_skipToBlock(t *testing.T) { }, nil).Once() expectStoreNewFinalizedCheckpoint(t, m.ORM, 5, 3) - require.NoError(t, m.LP.SkipToBlock(ctx, 6)) + require.NoError(t, m.LP.SkipToBlock(t.Context(), 6)) }) t.Run("backward skip is not allowed", func(t *testing.T) { @@ -2268,7 +2267,7 @@ func Test_skipToBlock(t *testing.T) { FinalizedBlockNumber: 3, SafeBlockNumber: 3, }, nil).Once() - require.ErrorContains(t, m.LP.SkipToBlock(ctx, 3), "skipping to a block 2 that is older than previously finalized 3 is not allowed. Use explicit remove-blocks to delete blocks") + require.ErrorContains(t, m.LP.SkipToBlock(t.Context(), 3), "skipping to a block 2 that is older than previously finalized 3 is not allowed. Use explicit remove-blocks to delete blocks") }) t.Run("select latest block error", func(t *testing.T) { @@ -2277,7 +2276,7 @@ func Test_skipToBlock(t *testing.T) { expectFinalizedAnchorHeader(t, m.Client, 2) m.ORM.EXPECT().SelectLatestBlock(mock.Anything).Return(nil, errors.New("db unavailable")).Once() - err := m.LP.SkipToBlock(ctx, 3) + err := m.LP.SkipToBlock(t.Context(), 3) require.ErrorContains(t, err, "failed to select latest block") }) @@ -2289,7 +2288,7 @@ func Test_skipToBlock(t *testing.T) { m.ORM.EXPECT().StoreNewFinalizedCheckpoint(mock.Anything, mock.Anything, int64(2)). Return(errors.New("db write failed")).Once() - err := m.LP.SkipToBlock(ctx, 3) + err := m.LP.SkipToBlock(t.Context(), 3) require.ErrorContains(t, err, "failed to store new finalized checkpoint") }) } diff --git a/pkg/logpoller/orm_test.go b/pkg/logpoller/orm_test.go index 32aca8f25d..b38c8eb137 100644 --- a/pkg/logpoller/orm_test.go +++ b/pkg/logpoller/orm_test.go @@ -2660,7 +2660,6 @@ func TestDSORM_StoreNewFinalizedCheckpoint_usesTransactionalDataSource(t *testin t.Parallel() testutils.SkipShortDB(t) - ctx := testutils.Context(t) base := testutils.NewSqlxDB(t) require.NotNil(t, base) wrapped := &execCountingSQLxDB{DB: base} @@ -2669,7 +2668,7 @@ func TestDSORM_StoreNewFinalizedCheckpoint_usesTransactionalDataSource(t *testin orm := logpoller.NewORM(chainID, wrapped, logger.Test(t)) anchor := finalizedCheckpointBlock(chainID, 5, common.HexToHash("0x0005")) - require.NoError(t, orm.StoreNewFinalizedCheckpoint(ctx, anchor, 5)) + require.NoError(t, orm.StoreNewFinalizedCheckpoint(t.Context(), anchor, 5)) require.Zero(t, wrapped.PoolDirectOpCount(), "StoreNewFinalizedCheckpoint must run DELETEs and INSERT on the transaction DataSource (orm.ds), not the outer pool; "+ diff --git a/pkg/testutils/testutils.go b/pkg/testutils/testutils.go index b410360d6a..3504031289 100644 --- a/pkg/testutils/testutils.go +++ b/pkg/testutils/testutils.go @@ -11,6 +11,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/utils/tests" ) +// Deprecated: use [*testing.T.Context] // Context returns a context with the test's deadline, if available. func Context(tb testing.TB) context.Context { return tests.Context(tb) From aee7d426897e038072f8ed1be3e329e7c4eaa323 Mon Sep 17 00:00:00 2001 From: Dmytro Haidashenko Date: Wed, 15 Jul 2026 16:46:05 +0200 Subject: [PATCH 10/12] eliminate second cast --- pkg/logpoller/log_poller.go | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/pkg/logpoller/log_poller.go b/pkg/logpoller/log_poller.go index b24eeb89dd..5d1dd0ddd8 100644 --- a/pkg/logpoller/log_poller.go +++ b/pkg/logpoller/log_poller.go @@ -1965,6 +1965,15 @@ func (lp *logPoller) DeleteLogsAndBlocksAfter(ctx context.Context, start int64) return lp.orm.DeleteLogsAndBlocksAfter(ctx, start) } +func (lp *logPoller) headerByNumberWithOpts(ctx context.Context, number *big.Int, opts evmtypes.HeaderByNumberOpts) (*evmtypes.Head, error) { + anchorHead, err := lp.ec.HeaderByNumberWithOpts(ctx, number, opts) + if err != nil { + return nil, err + } + + return (*evmtypes.Head)(anchorHead), nil +} + // SkipToBlock repositions the poller to start processing from blockNumber (T). // Block T-1 is fetched from RPC and saved as the anchor; the next poll resumes from T. // Returns an error if T-1 is not finalized on chain. @@ -1975,17 +1984,17 @@ func (lp *logPoller) SkipToBlock(ctx context.Context, blockNumber int64) error { anchorBlockNumber := blockNumber - 1 - anchorHead, err := lp.ec.HeaderByNumberWithOpts(ctx, big.NewInt(anchorBlockNumber), evmtypes.HeaderByNumberOpts{ConfidenceLevel: primitives.Finalized}) + anchorHead, err := lp.headerByNumberWithOpts(ctx, big.NewInt(anchorBlockNumber), evmtypes.HeaderByNumberOpts{ConfidenceLevel: primitives.Finalized}) if err != nil { return fmt.Errorf("failed to fetch anchor block %d from RPC: %w", anchorBlockNumber, err) } - err = validateHead((*evmtypes.Head)(anchorHead)) + err = validateHead(anchorHead) if err != nil { return fmt.Errorf("failed to validate anchor block %d: %w", anchorBlockNumber, err) } - anchorBlock := finalizedHeadToBlock((*evmtypes.Head)(anchorHead)) + anchorBlock := finalizedHeadToBlock(anchorHead) lp.runMu.Lock() defer lp.runMu.Unlock() From 26df33cb35889510243718fa2aca62b478566ea4 Mon Sep 17 00:00:00 2001 From: Dmytro Haidashenko Date: Wed, 15 Jul 2026 17:11:04 +0200 Subject: [PATCH 11/12] bump common --- go.mod | 20 ++++++++++---------- go.sum | 40 ++++++++++++++++++---------------------- 2 files changed, 28 insertions(+), 32 deletions(-) diff --git a/go.mod b/go.mod index 4446297e3d..3c8ecd0d9b 100644 --- a/go.mod +++ b/go.mod @@ -30,7 +30,7 @@ require ( github.com/shopspring/decimal v1.4.0 github.com/smartcontractkit/chain-selectors v1.0.100 github.com/smartcontractkit/chainlink-automation v0.8.1 - github.com/smartcontractkit/chainlink-common v0.11.2-0.20260518100439-9564f35fd264 + github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715145851-8219609496a4 github.com/smartcontractkit/chainlink-common/keystore v1.1.0 github.com/smartcontractkit/chainlink-data-streams v0.1.15-0.20260522094612-5f9f748bd87a github.com/smartcontractkit/chainlink-evm/gethwrappers v0.0.0-20260512150409-b4068bf735e6 @@ -38,7 +38,7 @@ require ( github.com/smartcontractkit/chainlink-framework/chains v0.0.0-20260423135514-5b1a7565a99c github.com/smartcontractkit/chainlink-framework/metrics v0.0.0-20260521164805-26d78d5e1243 github.com/smartcontractkit/chainlink-framework/multinode v0.0.0-20260521164805-26d78d5e1243 - github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260514104516-a827acdffe43 + github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260622152157-c8e129347b8b github.com/smartcontractkit/chainlink-protos/svr v1.2.0 github.com/smartcontractkit/chainlink-tron/relayer v0.0.11-0.20250815105909-75499abc4335 github.com/smartcontractkit/freeport v0.1.3-0.20250828155247-add56fa28aad @@ -52,7 +52,7 @@ require ( go.opentelemetry.io/otel/metric v1.43.0 go.uber.org/multierr v1.11.0 go.uber.org/zap v1.28.0 - golang.org/x/crypto v0.51.0 + golang.org/x/crypto v0.52.0 golang.org/x/exp v0.0.0-20260508232706-74f9aab9d74a golang.org/x/sync v0.20.0 google.golang.org/protobuf v1.36.11 @@ -100,6 +100,7 @@ require ( github.com/fsnotify/fsnotify v1.10.1 // indirect github.com/gabriel-vasile/mimetype v1.4.13 // indirect github.com/getsentry/sentry-go v0.27.0 // indirect + github.com/go-errors/errors v1.5.1 // indirect github.com/go-json-experiment/json v0.0.0-20260505212615-e40f80bf6836 // indirect github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect @@ -168,15 +169,13 @@ require ( github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 // indirect github.com/scylladb/go-reflectx v1.0.1 // indirect github.com/shirou/gopsutil v3.21.11+incompatible // indirect - github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.10 // indirect + github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62 // indirect github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4 // indirect github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305 // indirect - github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305 // indirect - github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20260323124644-faea187e6997 // indirect + github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16 // indirect + github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20260528173149-f5b8336b19d9 // indirect github.com/smartcontractkit/grpc-proxy v0.0.0-20240830132753-a7e17fec5ab7 // indirect github.com/smartcontractkit/wsrpc v0.8.5-0.20250502134807-c57d3d995945 // indirect - github.com/stellar/go-stellar-sdk v0.5.0 // indirect - github.com/stellar/go-xdr v0.0.0-20260423131911-a87d4d0789c3 // indirect github.com/stretchr/objx v0.5.3 // indirect github.com/supranational/blst v0.3.16 // indirect github.com/syndtr/goleveldb v1.0.1-0.20220721030215-126854af5e6d // indirect @@ -195,6 +194,7 @@ require ( go.dedis.ch/kyber/v3 v3.1.0 // indirect go.mongodb.org/mongo-driver v1.17.9 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/contrib/bridges/prometheus v0.68.0 // indirect go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.68.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.19.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.19.0 // indirect @@ -216,8 +216,8 @@ require ( go.yaml.in/yaml/v2 v2.4.4 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect go.yaml.in/yaml/v4 v4.0.0-rc.4 // indirect - golang.org/x/net v0.54.0 // indirect - golang.org/x/sys v0.44.0 // indirect + golang.org/x/net v0.55.0 // indirect + golang.org/x/sys v0.45.0 // indirect golang.org/x/text v0.37.0 // indirect golang.org/x/time v0.15.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260511170946-3700d4141b60 // indirect diff --git a/go.sum b/go.sum index fd09504af8..92c401d00f 100644 --- a/go.sum +++ b/go.sum @@ -660,12 +660,12 @@ github.com/smartcontractkit/chain-selectors v1.0.100 h1:wpiSpmI/eFjY+wx/nPr5VuNF github.com/smartcontractkit/chain-selectors v1.0.100/go.mod h1:qy7whtgG5g+7z0jt0nRyii9bLND9m15NZTzuQPkMZ5w= github.com/smartcontractkit/chainlink-automation v0.8.1 h1:sTc9LKpBvcKPc1JDYAmgBc2xpDKBco/Q4h4ydl6+UUU= github.com/smartcontractkit/chainlink-automation v0.8.1/go.mod h1:Iij36PvWZ6blrdC5A/nrQUBuf3MH3JvsBB9sSyc9W08= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260518100439-9564f35fd264 h1:r4sH0uvOoSXegOQPVaMEsu27q1dfWplIwO4WvdcEtKM= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260518100439-9564f35fd264/go.mod h1:B+eYJSQmOc28kzs7OwJjwo0DEV2f01HnUk89r9R1d/Y= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715145851-8219609496a4 h1:LzDSHQ5bM4pp8Wrr0mFjg60A4J47+IFPIuC/rHNF1qA= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715145851-8219609496a4/go.mod h1:snfVBRRQTpC2x5O3bQHZe9SvJX5yv/SbG8oHkJTKLtE= github.com/smartcontractkit/chainlink-common/keystore v1.1.0 h1:2wzySccgk2fpWusPKO0bpeAZzfSU9eq6CS5U+JwYaVo= github.com/smartcontractkit/chainlink-common/keystore v1.1.0/go.mod h1:6JexOOhPhknQ0QMuppFIlOpm6wCp54yZMxai+tWugwY= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.10 h1:FJAFgXS9oqASnkS03RE1HQwYQQxrO4l46O5JSzxqLgg= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.10/go.mod h1:oiDa54M0FwxevWwyAX773lwdWvFYYlYHHQV1LQ5HpWY= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62 h1:o7vfwNQjQbMKQ9YsZFQOxvU7RMXD/wKnZsX5N9sDS3w= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62/go.mod h1:HmUyH2oD9m+GRpKq7q3vuRnm1F2Uczf/Nd1v3ipMSK8= github.com/smartcontractkit/chainlink-data-streams v0.1.15-0.20260522094612-5f9f748bd87a h1:8bIqv4r7SgDWkXL2Qz/Ijw+YjZY1uroIte3E2v2keVk= github.com/smartcontractkit/chainlink-data-streams v0.1.15-0.20260522094612-5f9f748bd87a/go.mod h1:dF5JiHWueHjYguUUUrFeb03MkcDqha/tssEkqTkgzp4= github.com/smartcontractkit/chainlink-evm/gethwrappers v0.0.0-20260512150409-b4068bf735e6 h1:JFo7C3FilwhfwGBLAyj2umbL+P4QxGmVi/b8yt9kqvI= @@ -680,16 +680,16 @@ github.com/smartcontractkit/chainlink-framework/multinode v0.0.0-20260521164805- github.com/smartcontractkit/chainlink-framework/multinode v0.0.0-20260521164805-26d78d5e1243/go.mod h1:7ketk4ischPQW/JQgmyHz6zdzLUJv1VC29SiSgosydQ= github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4 h1:GCzrxDWn3b7jFfEA+WiYRi8CKoegsayiDoJBCjYkneE= github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4/go.mod h1:HHGeDUpAsPa0pmOx7wrByCitjQ0mbUxf0R9v+g67uCA= -github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260514104516-a827acdffe43 h1:9vjqB+iNqwyazVoVjR1rozHXTeRYyeggavt3Q4sbNrg= -github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260514104516-a827acdffe43/go.mod h1:Jqt53s27Tr0jDl8mdBXg1xhu6F8Fci8JOuq43tgHOM8= +github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260622152157-c8e129347b8b h1:VDgJWDipihV9f7M5+d21d1RzSsg5rEv+iI12oN1VQbo= +github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260622152157-c8e129347b8b/go.mod h1:vTFHTCbLui4Vn8fTmAadfE3rdnvfrDwOmMujmW857D0= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305 h1:NJdGFhzT6zMaTod4QkBqVD2sg0I25iw1boOYtTpEwRo= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305/go.mod h1:qSTSwX3cBP3FKQwQacdjArqv0g6QnukjV4XuzO6UyoY= -github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305 h1:bnSl5p3mFekSJ6QcbZ1TmHn2ffYiX8xk6hNzVmyhstQ= -github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305/go.mod h1:dkR2uYg9XYJuT1JASkPzWE51jjFkVb86P7a/yXe5/GM= +github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16 h1:/vkKPJoweLkRd56V4YHGRAtTG4+/JAlgklGEfvH6l4c= +github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16/go.mod h1:dkR2uYg9XYJuT1JASkPzWE51jjFkVb86P7a/yXe5/GM= github.com/smartcontractkit/chainlink-protos/svr v1.2.0 h1:7jjgqRgORQS/ikL3z0ZgJy95pzjhR9LuU1TVWg4BZ78= github.com/smartcontractkit/chainlink-protos/svr v1.2.0/go.mod h1:TcOliTQU6r59DwG4lo3U+mFM9WWyBHGuFkkxQpvSujo= -github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20260323124644-faea187e6997 h1:W0HKHO8eE8BckTRnhSdqjHKbJcnk068nEWYnWRu6tJY= -github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20260323124644-faea187e6997/go.mod h1:GTpDgyK0OObf7jpch6p8N281KxN92wbB8serZhU9yRc= +github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20260528173149-f5b8336b19d9 h1:LQy2j2+TdKLSWsUTUYuqmQPn8kjqCLjGI3ZJYGtDc08= +github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20260528173149-f5b8336b19d9/go.mod h1:GTpDgyK0OObf7jpch6p8N281KxN92wbB8serZhU9yRc= github.com/smartcontractkit/chainlink-tron/relayer v0.0.11-0.20250815105909-75499abc4335 h1:7bxYNrPpygn8PUSBiEKn8riMd7CXMi/4bjTy0fHhcrY= github.com/smartcontractkit/chainlink-tron/relayer v0.0.11-0.20250815105909-75499abc4335/go.mod h1:ccjEgNeqOO+bjPddnL4lUrNLzyCvGCxgBjJdhFX3wa8= github.com/smartcontractkit/chainlink-tron/relayer/gotron-sdk v0.0.5-0.20250528121202-292529af39df h1:36e3ROIZyV/qE8SvFOACXtXfMOMd9vG4+zY2v2ScXkI= @@ -712,10 +712,6 @@ github.com/spf13/cobra v1.2.1/go.mod h1:ExllRjgxM/piMAM+3tAZvg8fsklGAf3tPfi+i8t6 github.com/spf13/jwalterweatherman v1.1.0/go.mod h1:aNWZUN0dPAAO/Ljvb5BEdw96iTZ0EXowPYD95IqWIGo= github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= github.com/spf13/viper v1.8.1/go.mod h1:o0Pch8wJ9BVSWGQMbra6iw0oQ5oktSIBaujf1rJH9Ns= -github.com/stellar/go-stellar-sdk v0.5.0 h1:xpOO+ZTyvGz54wTm7pwl2Gf1e6lZl0ExrJ/tKb+Roj4= -github.com/stellar/go-stellar-sdk v0.5.0/go.mod h1:tLKAQPxa2I5UvGMabBbUXcY3fmgYnfDudrMeK7CDX4w= -github.com/stellar/go-xdr v0.0.0-20260423131911-a87d4d0789c3 h1:/NnsQpfVYPFr5Xm/V7EOblI8P05zewwkJLOlLItsW7U= -github.com/stellar/go-xdr v0.0.0-20260423131911-a87d4d0789c3/go.mod h1:If+U9Z1W5xU97VrOgJandQT+2dN7/iOpkCrxBJEyF80= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.2.0/go.mod h1:qt09Ya8vawLte6SNmTgCsAVtYtaKzEcn8ATUoHMkEqE= @@ -770,8 +766,6 @@ github.com/valyala/fastjson v1.4.1/go.mod h1:nV6MsjxL2IMJQUoHDIrjEI7oLyeqK6aBD7E github.com/wlynxg/anet v0.0.3/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA= github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= -github.com/xdrpp/goxdr v0.1.1 h1:E1B2c6E8eYhOVyd7yEpOyopzTPirUeF6mVOfXfGyJyc= -github.com/xdrpp/goxdr v0.1.1/go.mod h1:dXo1scL/l6s7iME1gxHWo2XCppbHEKZS7m/KyYWkNzA= github.com/xrash/smetrics v0.0.0-20240521201337-686a1a2994c1 h1:gEOO8jv9F4OT7lGCjxCBTO/36wtF6j2nSip77qHd4x4= github.com/xrash/smetrics v0.0.0-20240521201337-686a1a2994c1/go.mod h1:Ohn+xnUBiLI6FVj/9LpzZWtj1/D6lUovWYBkxHVV3aM= github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= @@ -810,6 +804,8 @@ go.opencensus.io v0.22.5/go.mod h1:5pWMHQbX5EPX2/62yrJeAkowc+lfs/XD7Uxpq3pI6kk= go.opencensus.io v0.23.0/go.mod h1:XItmlyltB5F7CS4xOC1DcqMoFqwtC6OG2xF7mCv7P7E= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/contrib/bridges/prometheus v0.68.0 h1:w3zlHYETbDwXyWHZlyyR58ZC39XGi8rAhkBgUgJ9d5w= +go.opentelemetry.io/contrib/bridges/prometheus v0.68.0/go.mod h1:GR/mClR2nn7vE8RLwxKjoBNg+QtgdDhRzxVa93koy5o= go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.68.0 h1:0Qx7VGBacMm9ZENQ7TnNObTYI4ShC+lHI16seduaxZo= go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.68.0/go.mod h1:Sje3i3MjSPKTSPvVWCaL8ugBzJwik3u4smCjUeuupqg= go.opentelemetry.io/otel v1.21.0/go.mod h1:QZzNPQPm1zLX4gZK4cMi+71eaorMSGT3A4znnUvNNEo= @@ -901,8 +897,8 @@ golang.org/x/crypto v0.12.0/go.mod h1:NF0Gs7EO5K4qLn+Ylc+fih8BSTeIjAP05siRnAh98y golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg= golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDfU= golang.org/x/crypto v0.20.0/go.mod h1:Xwo95rrVNIoSMx9wa1JroENMToLWn3RNVrTBpLHgZPQ= -golang.org/x/crypto v0.51.0 h1:IBPXwPfKxY7cWQZ38ZCIRPI50YLeevDLlLnyC5wRGTI= -golang.org/x/crypto v0.51.0/go.mod h1:8AdwkbraGNABw2kOX6YFPs3WM22XqI4EXEd8g+x7Oc8= +golang.org/x/crypto v0.52.0 h1:RMs7fP2rXdep0CftQlK8Uf+kibLm7qkCcradZWYz988= +golang.org/x/crypto v0.52.0/go.mod h1:1QgfPxDqh0T2M/elOJtp9RvuR95kVjir0e6/BvEmGbc= golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8= @@ -994,8 +990,8 @@ golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg= golang.org/x/net v0.14.0/go.mod h1:PpSgVXXLK0OxS0F31C1/tv6XNguvCrnXIDrFMspZIUI= golang.org/x/net v0.20.0/go.mod h1:z8BVo6PvndSri0LbOE3hAn0apkU+1YvI6E70E9jsnvY= golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44= -golang.org/x/net v0.54.0 h1:2zJIZAxAHV/OHCDTCOHAYehQzLfSXuf/5SoL/Dv6w/w= -golang.org/x/net v0.54.0/go.mod h1:Sj4oj8jK6XmHpBZU/zWHw3BV3abl4Kvi+Ut7cQcY+cQ= +golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= +golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= @@ -1096,8 +1092,8 @@ golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.14.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= -golang.org/x/sys v0.44.0 h1:ildZl3J4uzeKP07r2F++Op7E9B29JRUy+a27EibtBTQ= -golang.org/x/sys v0.44.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY= +golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= From 9a4c63d868c9107dee1bb97ba5586f6eac93b4ac Mon Sep 17 00:00:00 2001 From: Dmytro Haidashenko Date: Wed, 15 Jul 2026 17:17:30 +0200 Subject: [PATCH 12/12] make generate --- go.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/go.md b/go.md index 62eb0824e0..654bc8ae77 100644 --- a/go.md +++ b/go.md @@ -10,6 +10,7 @@ flowchart LR chainlink-common --> chainlink-protos/billing/go chainlink-common --> chainlink-protos/cre/go chainlink-common --> chainlink-protos/linking-service/go + chainlink-common --> chainlink-protos/metering/go chainlink-common --> chainlink-protos/node-platform chainlink-common --> chainlink-protos/storage-service chainlink-common --> freeport @@ -54,6 +55,8 @@ flowchart LR click chainlink-protos/cre/go href "https://github.com/smartcontractkit/chainlink-protos" chainlink-protos/linking-service/go click chainlink-protos/linking-service/go href "https://github.com/smartcontractkit/chainlink-protos" + chainlink-protos/metering/go + click chainlink-protos/metering/go href "https://github.com/smartcontractkit/chainlink-protos" chainlink-protos/node-platform click chainlink-protos/node-platform href "https://github.com/smartcontractkit/chainlink-protos" chainlink-protos/storage-service @@ -106,6 +109,7 @@ flowchart LR chainlink-protos/billing/go chainlink-protos/cre/go chainlink-protos/linking-service/go + chainlink-protos/metering/go chainlink-protos/node-platform chainlink-protos/storage-service chainlink-protos/svr