From c82b660143245e2dc563c25dd87304dcc39460e5 Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Sat, 15 Mar 2025 11:19:15 +0100 Subject: [PATCH] eth/filters: refactored filter logic --- common/range.go | 18 ++- eth/filters/filter.go | 308 +++++++++++++++++++++---------------- eth/filters/filter_test.go | 17 +- 3 files changed, 191 insertions(+), 152 deletions(-) diff --git a/common/range.go b/common/range.go index ca7af1b56d..a346998bb8 100644 --- a/common/range.go +++ b/common/range.go @@ -92,16 +92,18 @@ func (r *Range[T]) SetLast(v T) { } func (r Range[T]) Intersection(q Range[T]) Range[T] { - if r.first > q.first { - q.first = r.first - } - if r.afterLast < q.afterLast { - q.afterLast = r.afterLast - } - if q.first > q.afterLast { + i := Range[T]{first: max(r.first, q.first), afterLast: min(r.afterLast, q.afterLast)} + if i.first > i.afterLast { return Range[T]{} } - return q + return i +} + +func (r Range[T]) Union(q Range[T]) Range[T] { + if max(r.first, q.first) > min(r.afterLast, q.afterLast) { + panic("cannot create union; gap between ranges") + } + return Range[T]{first: min(r.first, q.first), afterLast: max(r.afterLast, q.afterLast)} } // Iter iterates all integers in the range. diff --git a/eth/filters/filter.go b/eth/filters/filter.go index 061b2d2229..2a95500ea5 100644 --- a/eth/filters/filter.go +++ b/eth/filters/filter.go @@ -146,6 +146,173 @@ type rangeLogsTestEvent struct { begin, end uint64 } +type searchSession struct { + ctx context.Context + filter *Filter + mb filtermaps.MatcherBackend + syncRange filtermaps.SyncRange // latest synchronized state with the matcher + firstBlock, lastBlock uint64 // specified search range; each can be MaxUint64 + searchRange common.Range[uint64] // actual search range; end trimmed to latest head + matchRange common.Range[uint64] // range in which we have results (subset of searchRange) + matches []*types.Log // valid set of matches in matchRange + forceUnindexed bool // revert to unindexed search +} + +func newSearchSession(ctx context.Context, filter *Filter, mb filtermaps.MatcherBackend, firstBlock, lastBlock uint64) (*searchSession, error) { + s := &searchSession{ + ctx: ctx, + filter: filter, + mb: mb, + firstBlock: firstBlock, + lastBlock: lastBlock, + } + // enforce a consistent state before starting the search in order to be able + // to determine valid range later + if err := s.syncMatcher(0); err != nil { + return nil, err + } + return s, nil +} + +func (s *searchSession) syncMatcher(trimTailThreshold uint64) error { + if s.filter.rangeLogsTestHook != nil && !s.matchRange.IsEmpty() { + s.filter.rangeLogsTestHook <- rangeLogsTestEvent{event: rangeLogsTestSync, begin: s.matchRange.First(), end: s.matchRange.Last()} + } + var err error + s.syncRange, err = s.mb.SyncLogIndex(s.ctx) + if err != nil { + return err + } + // update actual search range based on current head number + first := min(s.firstBlock, s.syncRange.HeadNumber) + last := min(s.lastBlock, s.syncRange.HeadNumber) + s.searchRange = common.NewRange[uint64](first, last+1-first) + // discard everything that is not needed or might be invalid + trimRange := s.syncRange.ValidBlocks + if trimRange.First() <= trimTailThreshold { + // everything before this point is already known to be valid; if this is + // valid then keep everything before + trimRange.SetFirst(0) + } + trimRange = trimRange.Intersection(s.searchRange) + s.trimMatches(trimRange) + if s.filter.rangeLogsTestHook != nil { + if !s.matchRange.IsEmpty() { + s.filter.rangeLogsTestHook <- rangeLogsTestEvent{event: rangeLogsTestTrimmed, begin: s.matchRange.First(), end: s.matchRange.Last()} + } else { + s.filter.rangeLogsTestHook <- rangeLogsTestEvent{event: rangeLogsTestTrimmed, begin: 0, end: 0} + } + } + return nil +} + +func (s *searchSession) trimMatches(trimRange common.Range[uint64]) { + s.matchRange = s.matchRange.Intersection(trimRange) + if s.matchRange.IsEmpty() { + s.matches = nil + return + } + for len(s.matches) > 0 && s.matches[0].BlockNumber < s.matchRange.First() { + s.matches = s.matches[1:] + } + for len(s.matches) > 0 && s.matches[len(s.matches)-1].BlockNumber > s.matchRange.Last() { + s.matches = s.matches[:len(s.matches)-1] + } +} + +func (s *searchSession) searchInRange(r common.Range[uint64], indexed bool) ([]*types.Log, error) { + first, last := r.First(), r.Last() + if indexed { + if s.filter.rangeLogsTestHook != nil { + s.filter.rangeLogsTestHook <- rangeLogsTestEvent{rangeLogsTestIndexed, first, last} + } + results, err := s.filter.indexedLogs(s.ctx, s.mb, first, last) + if err != filtermaps.ErrMatchAll { + return results, err + } + // "match all" filters are not supported by filtermaps; fall back to + // unindexed search which is the most efficient in this case + s.forceUnindexed = true + // fall through to unindexed case + } + if s.filter.rangeLogsTestHook != nil { + s.filter.rangeLogsTestHook <- rangeLogsTestEvent{rangeLogsTestUnindexed, first, last} + } + return s.filter.unindexedLogs(s.ctx, first, last) +} + +func (s *searchSession) doSearchIteration() error { + switch { + case s.syncRange.IndexedBlocks.IsEmpty(): + // indexer is not ready; fallback to completely unindexed search, do not check valid range + var err error + s.matchRange = s.searchRange + s.matches, err = s.searchInRange(s.searchRange, false) + return err + + case s.matchRange.IsEmpty(): + // no results yet; try search in entire range + indexedSearchRange := s.searchRange.Intersection(s.syncRange.IndexedBlocks) + var err error + if s.forceUnindexed = indexedSearchRange.IsEmpty(); !s.forceUnindexed { + // indexed search on the intersection of indexed and searched range + s.matchRange = indexedSearchRange + s.matches, err = s.searchInRange(indexedSearchRange, true) + if err != nil { + return err + } + return s.syncMatcher(0) // trim everything that the matcher considers potentially invalid + } else { + // no intersection of indexed and searched range; unindexed search on the whole searched range + s.matchRange = s.searchRange + s.matches, err = s.searchInRange(s.searchRange, false) + if err != nil { + return err + } + return s.syncMatcher(math.MaxUint64) // unindexed search is not affected by the tail of valid range + } + + case !s.matchRange.IsEmpty() && s.matchRange.First() > s.searchRange.First(): + // we have results but tail section is missing; do unindexed search for + // the tail part but still allow indexed search for missing head section + tailRange := common.NewRange[uint64](s.searchRange.First(), s.matchRange.First()-s.searchRange.First()) + tailMatches, err := s.searchInRange(tailRange, false) + if err != nil { + return err + } + s.matches = append(tailMatches, s.matches...) + s.matchRange = tailRange.Union(s.matchRange) + return s.syncMatcher(math.MaxUint64) // unindexed search is not affected by the tail of valid range + + case !s.matchRange.IsEmpty() && s.matchRange.First() == s.searchRange.First() && s.searchRange.AfterLast() > s.matchRange.AfterLast(): + // we have results but head section is missing + headRange := common.NewRange[uint64](s.matchRange.AfterLast(), s.searchRange.AfterLast()-s.matchRange.AfterLast()) + if !s.forceUnindexed { + indexedHeadRange := headRange.Intersection(s.syncRange.IndexedBlocks) + if !indexedHeadRange.IsEmpty() && indexedHeadRange.First() == headRange.First() { + // indexed head range search is possible + headRange = indexedHeadRange + } else { + s.forceUnindexed = true + } + } + headMatches, err := s.searchInRange(headRange, !s.forceUnindexed) + if err != nil { + return err + } + s.matches = append(s.matches, headMatches...) + s.matchRange = s.matchRange.Union(headRange) + if s.forceUnindexed { + return s.syncMatcher(math.MaxUint64) // unindexed search is not affected by the tail of valid range + } else { + return s.syncMatcher(headRange.First()) // trim if the tail of latest head search results might be invalid + } + + default: + panic("invalid search session state") + } +} + func (f *Filter) rangeLogs(ctx context.Context, firstBlock, lastBlock uint64) ([]*types.Log, error) { if f.rangeLogsTestHook != nil { defer func() { @@ -157,150 +324,19 @@ func (f *Filter) rangeLogs(ctx context.Context, firstBlock, lastBlock uint64) ([ if firstBlock > lastBlock { return nil, nil } - mb := f.sys.backend.NewMatcherBackend() defer mb.Close() - // enforce a consistent state before starting the search in order to be able - // to determine valid range later - syncRange, err := mb.SyncLogIndex(ctx) + session, err := newSearchSession(ctx, f, mb, firstBlock, lastBlock) if err != nil { return nil, err } - if syncRange.IndexedBlocks.IsEmpty() { - // fallback to completely unindexed search - headNum := syncRange.HeadNumber - firstBlock = min(firstBlock, headNum) - lastBlock = min(lastBlock, headNum) - if f.rangeLogsTestHook != nil { - f.rangeLogsTestHook <- rangeLogsTestEvent{rangeLogsTestUnindexed, firstBlock, lastBlock} - } - return f.unindexedLogs(ctx, firstBlock, lastBlock) - } - - headBlock := syncRange.HeadNumber - // if haveMatches == true then matches correspond to the block number range - // between matchFirst and matchLast - var ( - matches []*types.Log - haveMatches, forceUnindexed bool - matchFirst, matchLast uint64 - ) - trimMatches := func(trimFirst, trimLast uint64) { - if !haveMatches { - return - } - if trimLast < matchFirst || trimFirst > matchLast { - matches, haveMatches, matchFirst, matchLast = nil, false, 0, 0 - return - } - if trimFirst > matchFirst { - for len(matches) > 0 && matches[0].BlockNumber < trimFirst { - matches = matches[1:] - } - matchFirst = trimFirst - } - if trimLast < matchLast { - for len(matches) > 0 && matches[len(matches)-1].BlockNumber > trimLast { - matches = matches[:len(matches)-1] - } - matchLast = trimLast - } - } - - for { - // determine range to be searched; for simplicity we only extend the most - // recent end of the existing match set by matching between searchFirst - // and searchLast. - searchFirst := min(firstBlock, headBlock) - searchLast := min(lastBlock, headBlock) - trimMatches(searchFirst, searchLast) - if haveMatches && matchFirst == searchFirst && matchLast == searchLast { - // The entire search range was explored. - return matches, nil - } - var trimTailIfNotValid uint64 - if haveMatches && matchFirst > searchFirst { - // missing tail section; do unindexed search - if f.rangeLogsTestHook != nil { - f.rangeLogsTestHook <- rangeLogsTestEvent{rangeLogsTestUnindexed, searchFirst, matchFirst - 1} - } - tailMatches, err := f.unindexedLogs(ctx, searchFirst, matchFirst-1) - if err != nil { - return matches, err - } - matches = append(tailMatches, matches...) - matchFirst = searchFirst - // unindexed results are not affected by valid tail; do not trim tail - trimTailIfNotValid = math.MaxUint64 - } else { - // if we have matches, they start at searchFirst - if haveMatches { - searchFirst = matchLast + 1 - if syncRange.IndexedBlocks.IsEmpty() || syncRange.IndexedBlocks.First() > searchFirst { - forceUnindexed = true - } - } - var newMatches []*types.Log - if syncRange.IndexedBlocks.IsEmpty() || syncRange.IndexedBlocks.First() > searchLast || syncRange.IndexedBlocks.Last() < searchFirst { - forceUnindexed = true - } - if !forceUnindexed { - searchFirst = max(searchFirst, syncRange.IndexedBlocks.First()) - searchLast = min(searchLast, syncRange.IndexedBlocks.Last()) - if f.rangeLogsTestHook != nil { - f.rangeLogsTestHook <- rangeLogsTestEvent{rangeLogsTestIndexed, searchFirst, searchLast} - } - newMatches, err = f.indexedLogs(ctx, mb, searchFirst, searchLast) - // trim tail if it affects the indexed search range - trimTailIfNotValid = searchFirst - if err == filtermaps.ErrMatchAll { - // "match all" filters are not supported by filtermaps; fall back - // to unindexed search which is the most efficient in this case - forceUnindexed = true - } - } - if forceUnindexed { - if f.rangeLogsTestHook != nil { - f.rangeLogsTestHook <- rangeLogsTestEvent{rangeLogsTestUnindexed, searchFirst, searchLast} - } - newMatches, err = f.unindexedLogs(ctx, searchFirst, searchLast) - // unindexed results are not affected by valid tail; do not trim tail - trimTailIfNotValid = math.MaxUint64 - } - if err != nil { - return matches, err - } - if !haveMatches { - matches = newMatches - haveMatches, matchFirst, matchLast = true, searchFirst, searchLast - } else { - matches = append(matches, newMatches...) - matchLast = searchLast - } - } - - if f.rangeLogsTestHook != nil { - f.rangeLogsTestHook <- rangeLogsTestEvent{event: rangeLogsTestSync, begin: matchFirst, end: matchLast} - } - syncRange, err = mb.SyncLogIndex(ctx) - if err != nil { - return matches, err - } - headBlock = syncRange.HeadNumber // Head is guaranteed != nil - if syncRange.ValidBlocks.IsEmpty() { - matches, haveMatches, matchFirst, matchLast = nil, false, 0, 0 - } else { - if syncRange.ValidBlocks.First() > trimTailIfNotValid { - trimMatches(syncRange.ValidBlocks.First(), syncRange.ValidBlocks.Last()) - } else { - trimMatches(0, syncRange.ValidBlocks.Last()) - } - } - if f.rangeLogsTestHook != nil { - f.rangeLogsTestHook <- rangeLogsTestEvent{event: rangeLogsTestTrimmed, begin: matchFirst, end: matchLast} + for session.searchRange != session.matchRange { + if err := session.doSearchIteration(); err != nil { + return session.matches, err } } + return session.matches, nil } func (f *Filter) indexedLogs(ctx context.Context, mb filtermaps.MatcherBackend, begin, end uint64) ([]*types.Log, error) { diff --git a/eth/filters/filter_test.go b/eth/filters/filter_test.go index e3e33f994b..aaae9e29e5 100644 --- a/eth/filters/filter_test.go +++ b/eth/filters/filter_test.go @@ -453,6 +453,14 @@ func TestRangeLogs(t *testing.T) { addresses = []common.Address{common.Address{}} ) + expEvent := func(exp rangeLogsTestEvent) { + event++ + ev := <-filter.rangeLogsTestHook + if ev != exp { + t.Fatalf("Test case #%d: wrong test event #%d received (got %v, expected %v)", testCase, event, ev, exp) + } + } + newFilter := func(begin, end int64) { testCase++ event = 0 @@ -464,14 +472,7 @@ func TestRangeLogs(t *testing.T) { for range filter.rangeLogsTestHook { } }(filter) - } - - expEvent := func(exp rangeLogsTestEvent) { - event++ - ev := <-filter.rangeLogsTestHook - if ev != exp { - t.Fatalf("Test case #%d: wrong test event #%d received (got %v, expected %v)", testCase, event, ev, exp) - } + expEvent(rangeLogsTestEvent{rangeLogsTestTrimmed, 0, 0}) } updateHead := func() {