fix(ethmonitor): reject malformed logs before caching

Author: pkieltyka

ethmonitor/ethmonitor.go

@@ -851,6 +851,8 @@ func (m *Monitor) logTopics() [][]common.Hash {
 }
 
 func (m *Monitor) filterLogs(ctx context.Context, blockHash common.Hash, topics [][]common.Hash, blockBloom types.Bloom) ([]types.Log, []byte, error) {
+	var fetchedLogs []types.Log
+
 	getter := func(ctx context.Context, _ string) ([]byte, error) {
 		if m.options.DebugLogging {
 			m.log.Debug(fmt.Sprintf("ethmonitor: filterLogs is calling origin for block hash %s", blockHash))
@@ -873,6 +875,12 @@ func (m *Monitor) filterLogs(ctx context.Context, blockHash common.Hash, topics
 			// ensure we do not incorrectly cache an empty block-logs response as valid.
 			return nil, fmt.Errorf("ethmonitor: filterLogs detected empty block-logs response but block bloom is set, ignoring node response")
 		}
+		// Validate before caching so a malformed response cannot block log
+		// backfilling until cache expiry.
+		fetchedLogs, err = m.unmarshalLogs(logsPayload)
+		if err != nil {
+			return nil, err
+		}
 		return logsPayload, nil
 	}
 
@@ -881,16 +889,25 @@ func (m *Monitor) filterLogs(ctx context.Context, blockHash common.Hash, topics
 		if err != nil {
 			return nil, resp, err
 		}
-		logs, err := m.unmarshalLogs(resp)
-		return logs, resp, err
+		return fetchedLogs, resp, nil
 	}
 
 	key := CacheKeyBlockLogs(m.chainID, blockHash, topics)
 	resp, err := m.cache.GetOrSetWithLockEx(ctx, key, getter, m.options.CacheExpiry)
 	if err != nil {
 		return nil, resp, err
 	}
+	if fetchedLogs != nil {
+		return fetchedLogs, resp, nil
+	}
 	logs, err := m.unmarshalLogs(resp)
+	if err != nil {
+		// Recover entries cached by peers or older monitors that did not
+		// validate logs before writing them.
+		if deleteErr := m.cache.Delete(ctx, key); deleteErr != nil {
+			m.log.Warn(fmt.Sprintf("ethmonitor: error deleting invalid logs cache for block hash %s due to: '%v'", blockHash.Hex(), deleteErr))
+		}
+	}
 	return logs, resp, err
 }
 

ethmonitor/prefetch_internal_test.go

@@ -5,10 +5,14 @@ import (
 	"encoding/json"
 	"fmt"
 	"math/big"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/0xsequence/ethkit/ethmonitor/internal/mocks"
+	"github.com/0xsequence/ethkit/go-ethereum"
+	"github.com/0xsequence/ethkit/go-ethereum/common"
+	"github.com/0xsequence/ethkit/go-ethereum/core/types"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/mock/gomock"
 )
@@ -35,6 +39,137 @@ func TestPrefetchRejectsInvalidBlockBeforeCaching(t *testing.T) {
 	}
 }
 
+func TestMonitorRecoversInvalidPrefetchedLogs(t *testing.T) {
+	for _, payload := range []string{`{}`, `[{}]`} {
+		for _, cached := range []bool{false, true} {
+			t.Run(fmt.Sprintf("payload=%s/cached=%v", payload, cached), func(t *testing.T) {
+				const first, target, last = uint64(1000), uint64(1002), uint64(1004)
+				targetHash := common.BigToHash(new(big.Int).SetUint64(target))
+				expectedLogs := []types.Log{{
+					Address:     common.HexToAddress("0x1234"),
+					Topics:      []common.Hash{common.HexToHash("0xabcd")},
+					Data:        []byte{1, 2},
+					TxHash:      common.HexToHash("0x5678"),
+					BlockHash:   targetHash,
+					BlockNumber: target,
+				}}
+				validPayload, err := json.Marshal(expectedLogs)
+				require.NoError(t, err)
+
+				provider := mocks.NewMockRawInterface(gomock.NewController(t))
+				provider.EXPECT().ChainID(gomock.Any()).Return(big.NewInt(1), nil).AnyTimes()
+				provider.EXPECT().IsStreamingEnabled().Return(false).AnyTimes()
+				provider.EXPECT().BlockNumber(gomock.Any()).Return(last, nil).AnyTimes()
+				provider.EXPECT().RawBlockByNumber(gomock.Any(), gomock.Any()).DoAndReturn(
+					func(ctx context.Context, number *big.Int) (json.RawMessage, error) {
+						if err := ctx.Err(); err != nil {
+							return nil, err
+						}
+						if number.Uint64() > last {
+							return nil, ethereum.NotFound
+						}
+						var bloom types.Bloom
+						if number.Uint64() == target {
+							bloom = types.BytesToBloom([]byte{1})
+						}
+						header := &types.Header{
+							Number:     number,
+							ParentHash: common.BigToHash(new(big.Int).Sub(number, big.NewInt(1))),
+							Difficulty: big.NewInt(0),
+							GasLimit:   30_000_000,
+							Bloom:      bloom,
+						}
+						header.SetHash(common.BigToHash(number))
+						return json.Marshal(header)
+					},
+				).AnyTimes()
+				var originCalls atomic.Int64
+				provider.EXPECT().RawFilterLogs(gomock.Any(), gomock.Any()).DoAndReturn(
+					func(ctx context.Context, query ethereum.FilterQuery) (json.RawMessage, error) {
+						if err := ctx.Err(); err != nil {
+							return nil, err
+						}
+						if *query.BlockHash != targetHash {
+							return json.RawMessage(`[]`), nil
+						}
+						if originCalls.Add(1) == 1 && !cached {
+							return json.RawMessage(payload), nil
+						}
+						return json.RawMessage(validPayload), nil
+					},
+				).AnyTimes()
+
+				opts := DefaultOptions
+				opts.WithLogs = true
+				opts.PrefetchConcurrency = 1
+				opts.StartBlockNumber = new(big.Int).SetUint64(first)
+				opts.PollingInterval = 10 * time.Millisecond
+				opts.CacheExpiry = time.Hour
+				// Leave CacheBackend unset to exercise automatic memory caching.
+				monitor, err := NewMonitor(provider, opts)
+				require.NoError(t, err)
+				monitor.chainID = big.NewInt(1)
+				key := CacheKeyBlockLogs(monitor.chainID, targetHash, monitor.logTopics())
+				if cached {
+					require.NoError(t, monitor.cache.SetEx(context.Background(), key, []byte(payload), time.Hour))
+				} else {
+					// Complete the speculative fetch before starting the serial loop,
+					// so the worker deterministically receives the malformed response.
+					monitor.prefetch.fetch(context.Background(), prefetchJob{num: target})
+					require.Equal(t, int64(1), originCalls.Load())
+					_, found, err := monitor.cache.Get(context.Background(), key)
+					require.NoError(t, err)
+					require.False(t, found, "failed prefetch must not cache malformed logs")
+				}
+
+				sub := monitor.Subscribe("TestMonitorRecoversInvalidPrefetchedLogs")
+				defer sub.Unsubscribe()
+				ctx, cancel := context.WithCancel(context.Background())
+				done := make(chan error, 1)
+				go func() { done <- monitor.Run(ctx) }()
+				defer func() {
+					cancel()
+					select {
+					case err := <-done:
+						require.NoError(t, err)
+					case <-time.After(5 * time.Second):
+						t.Error("monitor did not stop")
+					}
+				}()
+
+				timer := time.NewTimer(3 * time.Second)
+				defer timer.Stop()
+				next := first
+				for next <= last {
+					select {
+					case blocks := <-sub.Blocks():
+						for _, block := range blocks {
+							require.Equal(t, Added, block.Event)
+							require.True(t, block.OK)
+							require.Equal(t, next, block.NumberU64(), "blocks must publish in order")
+							if next == target {
+								require.Equal(t, expectedLogs, block.Logs)
+							}
+							next++
+						}
+					case <-timer.C:
+						t.Fatalf("publication stalled at block %d; origin calls: %d", next, originCalls.Load())
+					}
+				}
+				if !cached {
+					require.GreaterOrEqual(t, originCalls.Load(), int64(2), "malformed prefetch must allow an origin retry")
+				} else {
+					require.Positive(t, originCalls.Load(), "invalid cache entry must allow an origin retry")
+				}
+				stored, found, err := monitor.cache.Get(context.Background(), key)
+				require.NoError(t, err)
+				require.True(t, found)
+				require.JSONEq(t, string(validPayload), string(stored))
+			})
+		}
+	}
+}
+
 func TestPrefetchPanicStopsWorkers(t *testing.T) {
 	provider := mocks.NewMockRawInterface(gomock.NewController(t))
 	provider.EXPECT().BlockNumber(gomock.Any()).DoAndReturn(func(context.Context) (uint64, error) {