ccip: Return an error from ParseLogs. (#408)

core/services/ocr2/plugins/ccip/internal/ccipdata/reader.go

@@ -1,6 +1,7 @@
 package ccipdata
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -43,23 +44,26 @@ func LogsConfirmations(finalized bool) logpoller.Confirmations {
 
 func ParseLogs[T any](logs []logpoller.Log, lggr logger.Logger, parseFunc func(log types.Log) (*T, error)) ([]Event[T], error) {
 	reqs := make([]Event[T], 0, len(logs))
+
 	for _, log := range logs {
 		data, err := parseFunc(log.ToGethLog())
-		if err == nil {
-			reqs = append(reqs, Event[T]{
-				Data: *data,
-				Meta: Meta{
-					BlockTimestamp: log.BlockTimestamp,
-					BlockNumber:    log.BlockNumber,
-					TxHash:         log.TxHash,
-					LogIndex:       uint(log.LogIndex),
-				},
-			})
+		if err != nil {
+			lggr.Errorw("Unable to parse log", "err", err)
+			continue
 		}
+		reqs = append(reqs, Event[T]{
+			Data: *data,
+			Meta: Meta{
+				BlockTimestamp: log.BlockTimestamp,
+				BlockNumber:    log.BlockNumber,
+				TxHash:         log.TxHash,
+				LogIndex:       uint(log.LogIndex),
+			},
+		})
 	}
 
 	if len(logs) != len(reqs) {
-		lggr.Warnw("Some logs were not parsed", "logs", len(logs), "requests", len(reqs))
+		return nil, fmt.Errorf("%d logs were not parsed", len(logs)-len(reqs))
 	}
 	return reqs, nil
 }

core/services/ocr2/plugins/ccip/internal/ccipdata/reader_test.go

@@ -7,6 +7,8 @@ import (
 
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap/zapcore"
 
 	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/logpoller"
 	"github.com/smartcontractkit/chainlink/v2/core/logger"
@@ -22,16 +24,12 @@ func Test_parseLogs(t *testing.T) {
 	}
 
 	parseFn := func(log types.Log) (*uint, error) {
-		// Simulate some random error
-		if log.Index == 100 {
-			return nil, fmt.Errorf("some error")
-		}
 		return &log.Index, nil
 	}
 
 	parsedEvents, err := ParseLogs[uint](logs, logger.TestLogger(t), parseFn)
-	assert.NoError(t, err)
-	assert.Len(t, parsedEvents, 99)
+	require.NoError(t, err)
+	assert.Len(t, parsedEvents, 100)
 
 	// Make sure everything is parsed according to the parse func
 	for i, ev := range parsedEvents {
@@ -40,3 +38,35 @@ func Test_parseLogs(t *testing.T) {
 		assert.Greater(t, ev.BlockTimestamp, time.Now().Add(-time.Minute))
 	}
 }
+
+func Test_parseLogs_withErrors(t *testing.T) {
+	// generate 50 valid logs and 50 errors
+	actualErrorCount := 50
+	logs := make([]logpoller.Log, actualErrorCount*2)
+	for i := range logs {
+		logs[i].LogIndex = int64(i + 1)
+	}
+
+	// return an error for half of the logs.
+	parseFn := func(log types.Log) (*uint, error) {
+		if log.Index%2 == 0 {
+			return nil, fmt.Errorf("cannot parse %d", log.Index)
+		}
+		return &log.Index, nil
+	}
+
+	log, observed := logger.TestLoggerObserved(t, zapcore.DebugLevel)
+	parsedEvents, err := ParseLogs[uint](logs, log, parseFn)
+	assert.ErrorContains(t, err, fmt.Sprintf("%d logs were not parsed", len(logs)/2))
+	assert.Nil(t, parsedEvents, "No events are returned if there was an error.")
+
+	// logs are written for errors.
+	require.Equal(t, actualErrorCount, observed.Len(), "Expect 51 warnings: one for each error and a summary.")
+	for i, entry := range observed.All() {
+		assert.Equal(t, zapcore.ErrorLevel, entry.Level)
+		assert.Contains(t, entry.Message, "Unable to parse log")
+		contextMap := entry.ContextMap()
+		require.Contains(t, contextMap, "err")
+		assert.Contains(t, contextMap["err"], fmt.Sprintf("cannot parse %d", (i+1)*2), "each error should be logged as a warning")
+	}
+}