Merge branch 'develop' into fix/inject-df-decoder

Author: silaslenihan

go.mod

@@ -21,8 +21,8 @@ require (
 	github.com/prometheus/client_model v0.6.1
 	github.com/prometheus/common v0.63.0
 	github.com/shopspring/decimal v1.4.0
-	github.com/smartcontractkit/chainlink-common v0.7.1-0.20250606180235-5981ca5ebf20
-	github.com/smartcontractkit/chainlink-framework/capabilities v0.0.0-20250606184316-1aa3b177e1ed
+	github.com/smartcontractkit/chainlink-common v0.7.1-0.20250609122728-51de5127f783
+	github.com/smartcontractkit/chainlink-framework/capabilities v0.0.0-20250609122841-af2e3861cd5b
 	github.com/smartcontractkit/chainlink-framework/chains v0.0.0-20250522110034-65c54665034a
 	github.com/smartcontractkit/chainlink-framework/metrics v0.0.0-20250522110034-65c54665034a
 	github.com/smartcontractkit/chainlink-framework/multinode v0.0.0-20250522110034-65c54665034a

go.sum

@@ -636,10 +636,10 @@ github.com/shurcooL/vfsgen v0.0.0-20200824052919-0d455de96546/go.mod h1:TrYk7fJV
 github.com/sirupsen/logrus v1.4.1/go.mod h1:ni0Sbl8bgC9z8RoU9G6nDWqqs/fq4eDPysMBDgk/93Q=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/sirupsen/logrus v1.8.1/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
-github.com/smartcontractkit/chainlink-common v0.7.1-0.20250606180235-5981ca5ebf20 h1:S2QoB+l2K1QUH5c/Y/1bfxQO+Qct0eRcyYg8JYuTpK0=
-github.com/smartcontractkit/chainlink-common v0.7.1-0.20250606180235-5981ca5ebf20/go.mod h1:H7gOuN4Jzf+DWllfP5Pb7AiCWBMQrDX1D1KYXAEhdnw=
-github.com/smartcontractkit/chainlink-framework/capabilities v0.0.0-20250606184316-1aa3b177e1ed h1:RFyeTr0i47HduHjmzQOFdnDIckpaERZ47TpjGxM++IM=
-github.com/smartcontractkit/chainlink-framework/capabilities v0.0.0-20250606184316-1aa3b177e1ed/go.mod h1:zT66e+8i4OifSvNqpR6yCB2lbciq35vPGxzFDkMWdnQ=
+github.com/smartcontractkit/chainlink-common v0.7.1-0.20250609122728-51de5127f783 h1:59Dqm+KykI4aeuCOl5JlQP3gv1u3ACGScXadakjIPjM=
+github.com/smartcontractkit/chainlink-common v0.7.1-0.20250609122728-51de5127f783/go.mod h1:H7gOuN4Jzf+DWllfP5Pb7AiCWBMQrDX1D1KYXAEhdnw=
+github.com/smartcontractkit/chainlink-framework/capabilities v0.0.0-20250609122841-af2e3861cd5b h1:nA2SI6TP532lk+cX3YG+XaRhqY9MwpGgOKBpKlbBjsI=
+github.com/smartcontractkit/chainlink-framework/capabilities v0.0.0-20250609122841-af2e3861cd5b/go.mod h1:3xH9a4lc4UYUWk/7CTfPgtZCbhEosqme1KOW8Fs0CaE=
 github.com/smartcontractkit/chainlink-framework/chains v0.0.0-20250522110034-65c54665034a h1:1CrY+turGOYaSEs4efA4zJprX0G2Uk+0dcDO1eCsHhk=
 github.com/smartcontractkit/chainlink-framework/chains v0.0.0-20250522110034-65c54665034a/go.mod h1:X+a4k2a+2G2/yeAaRQMCTLmlhNdQYAeN6v+ZpLzRZww=
 github.com/smartcontractkit/chainlink-framework/metrics v0.0.0-20250522110034-65c54665034a h1:bFYBcW0cmhq0G8NSjPxSFfL/fVODuhEGluyWOxJTqqk=

pkg/logpoller/log_poller.go

@@ -879,7 +879,7 @@ func (lp *logPoller) blocksFromFinalizedLogs(ctx context.Context, logs []types.L
 	for _, log := range logs {
 		numbers = append(numbers, log.BlockNumber)
 	}
-	if numbers[len(numbers)-1] != endBlockNumber {
+	if len(numbers) == 0 || numbers[len(numbers)-1] != endBlockNumber {
 		numbers = append(numbers, endBlockNumber)
 	}
 	blocks, err = lp.GetBlocksRange(ctx, numbers)
@@ -930,21 +930,19 @@ func (lp *logPoller) backfill(ctx context.Context, start, end int64) error {
 			continue
 		}
 		lp.missingBlocksErrorCount.Store(0) // clear unhealthy node state in case we were missing blocks and just found them
-		if len(gethLogs) == 0 {
-			continue
-		}
+
 		blocks, err := lp.blocksFromFinalizedLogs(ctx, gethLogs, uint64(to)) //nolint:gosec // G115
 		if err != nil {
 			return err
 		}
 
 		endblock := blocks[len(blocks)-1]
-		if gethLogs[len(gethLogs)-1].BlockNumber != uint64(to) {
+		if len(gethLogs) == 0 || gethLogs[len(gethLogs)-1].BlockNumber != uint64(to) { //nolint:gosec // G115
 			// Pop endblock if there were no logs for it, so that length of blocks & gethLogs are the same to pass to convertLogs
 			blocks = blocks[:len(blocks)-1]
 		}
 
-		lp.lggr.Debugw("Backfill found logs", "from", from, "to", to, "logs", len(gethLogs), "blocks", blocks)
+		lp.lggr.Debugw("Inserting backfilled logs with batch endblock", "from", from, "to", to, "logs", len(gethLogs), "blocks", blocks)
 		err = lp.orm.InsertLogsWithBlock(ctx, convertLogs(gethLogs, blocks, lp.lggr, lp.ec.ConfiguredChainID()), endblock)
 		if err != nil {
 			lp.lggr.Warnw("Unable to insert logs, retrying", "err", err, "from", from, "to", to)
@@ -1208,7 +1206,13 @@ func (lp *logPoller) PruneOldBlocks(ctx context.Context) (bool, error) {
 		// No blocks saved yet.
 		return true, nil
 	}
-	if latestBlock.FinalizedBlockNumber <= lp.keepFinalizedBlocksDepth {
+
+	// If the latest block we have in the db was saved during a backfill, then the latest finalized
+	// block number stored with it will be larger than its block number. Instead of risking deleting
+	// all blocks from the db, we should still keep the latest keepFinalizedBlocksDepth blocks
+	referenceBlockNumber := mathutil.Min(latestBlock.FinalizedBlockNumber, latestBlock.BlockNumber)
+
+	if referenceBlockNumber <= lp.keepFinalizedBlocksDepth {
 		// No-op, keep all blocks
 		return true, nil
 	}

pkg/logpoller/log_poller_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/ethclient/simulated"
+	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/leanovate/gopter"
 	"github.com/leanovate/gopter/gen"
 	"github.com/leanovate/gopter/prop"
@@ -1562,15 +1563,42 @@ func TestTooManyLogResults(t *testing.T) {
 		if blockNumber == nil {
 			require.FailNow(t, "unexpected call to get current head")
 		}
-		return &evmtypes.Head{Number: blockNumber.Int64()}, nil
+		return &evmtypes.Head{Number: blockNumber.Int64(), ParentHash: common.HexToHash(fmt.Sprintf("0x%x", blockNumber.Int64()-1))}, nil
 	})
 
 	t.Run("halves size until small enough, then succeeds", func(t *testing.T) {
-		// Simulate currentBlock = 300
+		// Simulate latestBlock = 300
 		head.Number = 300
+		head.Hash = common.HexToHash("0x1234") // needed to satisfy validation in fetchBlocks()
 		finalized.Number = head.Number - lpOpts.FinalityDepth
+
 		headTracker.On("LatestAndFinalizedBlock", mock.Anything).Return(head, finalized, nil).Once()
 
+		headByHash := ec.On("HeadByHash", mock.Anything, mock.Anything).Return(func(ctx context.Context, blockHash common.Hash) (*evmtypes.Head, error) {
+			return &evmtypes.Head{Hash: blockHash}, nil
+		})
+
+		batchCallContext := ec.On("BatchCallContext", mock.Anything, mock.Anything).Return(
+			func(ctx context.Context, calls []rpc.BatchElem) error {
+				for i := range calls {
+					blockNumberHex := calls[i].Args[0].(string)
+					if blockNumberHex == "latest" {
+						calls[i].Result = head
+						continue
+					}
+					blockNumber, ok := new(big.Int).SetString(blockNumberHex[2:], 16)
+					require.True(t, ok, blockNumberHex)
+
+					calls[i].Result = &evmtypes.Head{
+						Number:     blockNumber.Int64(),
+						Hash:       common.HexToHash(fmt.Sprintf("0x%x", blockNumber.Int64())),
+						ParentHash: common.HexToHash(fmt.Sprintf("0x%x", blockNumber.Int64()-1)),
+					}
+				}
+				return nil
+			},
+		)
+
 		filterLogsCall = ec.On("FilterLogs", mock.Anything, mock.Anything).Return(func(ctx context.Context, fq ethereum.FilterQuery) (logs []types.Log, err error) {
 			if fq.BlockHash != nil {
 				return []types.Log{}, nil // succeed when single block requested
@@ -1605,6 +1633,8 @@ func TestTooManyLogResults(t *testing.T) {
 			assert.Equal(t, s, logs[i].ContextMap()["newBatchSize"])
 		}
 		filterLogsCall.Unset()
+		batchCallContext.Unset()
+		headByHash.Unset()
 	})
 
 	t.Run("Halves size until single block, then reports critical error", func(t *testing.T) {

pkg/report/por/processor/processor.go

@@ -22,7 +22,7 @@ func NewPORProcessor(metrics *registry.Metrics, emitter beholder.ProtoEmitter) *
 
 func GetPORSchema() abi.Arguments {
 	return registry.GetSchema("tuple(bytes32,uint32,bytes)[]", "", []abi.ArgumentMarshaling{
-		{Name: "dataId", Type: "bytes32"},
+		{Name: "dataID", Type: "bytes32"},
 		{Name: "timestamp", Type: "uint32"},
 		{Name: "bundle", Type: "bytes"},
 	})