forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-26 14:46:42 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

eth/filters: safe chain view update

Browse source
This commit is contained in:
Zsolt Felfoldi 2025-04-19 00:30:20 +02:00
parent 03e88a8505
commit d0967568a4
10 changed files with 229 additions and 156 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
core/filtermaps/matcher.go
View file

@ -57,7 +57,7 @@ type MatcherBackend interface {
// all states of the chain since the previous SyncLogIndex or the creation of // all states of the chain since the previous SyncLogIndex or the creation of
// the matcher backend. // the matcher backend.
type SyncRange struct { type SyncRange struct {
HeadNumber uint64 IndexedView *ChainView
// block range where the index has not changed since the last matcher sync // block range where the index has not changed since the last matcher sync
// and therefore the set of matches found in this region is guaranteed to // and therefore the set of matches found in this region is guaranteed to
// be valid and complete. // be valid and complete.

6
core/filtermaps/matcher_backend.go
View file

@ -128,7 +128,7 @@ func (fm *FilterMapsMatcherBackend) synced() {
indexedBlocks.SetAfterLast(indexedBlocks.Last()) // remove partially indexed last block indexedBlocks.SetAfterLast(indexedBlocks.Last()) // remove partially indexed last block
} }
fm.syncCh <- SyncRange{ fm.syncCh <- SyncRange{
HeadNumber: fm.f.targetView.HeadNumber(), IndexedView: fm.f.indexedView,
ValidBlocks: fm.validBlocks, ValidBlocks: fm.validBlocks,
IndexedBlocks: indexedBlocks, IndexedBlocks: indexedBlocks,
} }
@ -154,7 +154,7 @@ func (fm *FilterMapsMatcherBackend) SyncLogIndex(ctx context.Context) (SyncRange
case <-ctx.Done(): case <-ctx.Done():
return SyncRange{}, ctx.Err() return SyncRange{}, ctx.Err()
case <-fm.f.disabledCh: case <-fm.f.disabledCh:
return SyncRange{HeadNumber: fm.f.targetView.HeadNumber()}, nil return SyncRange{IndexedView: fm.f.indexedView}, nil
} }
select { select {
case vr := <-syncCh: case vr := <-syncCh:
@ -162,7 +162,7 @@ func (fm *FilterMapsMatcherBackend) SyncLogIndex(ctx context.Context) (SyncRange
case <-ctx.Done(): case <-ctx.Done():
return SyncRange{}, ctx.Err() return SyncRange{}, ctx.Err()
case <-fm.f.disabledCh: case <-fm.f.disabledCh:
return SyncRange{HeadNumber: fm.f.targetView.HeadNumber()}, nil return SyncRange{IndexedView: fm.f.indexedView}, nil
} }
} }

8
eth/api_backend.go
View file

@ -443,6 +443,14 @@ func (b *EthAPIBackend) RPCTxFeeCap() float64 {
return b.eth.config.RPCTxFeeCap return b.eth.config.RPCTxFeeCap
} }
func (b *EthAPIBackend) CurrentView() *filtermaps.ChainView {
head := b.eth.blockchain.CurrentBlock()
if head == nil {
return nil
}
return filtermaps.NewChainView(b.eth.blockchain, head.Number.Uint64(), head.Hash())
}
func (b *EthAPIBackend) NewMatcherBackend() filtermaps.MatcherBackend { func (b *EthAPIBackend) NewMatcherBackend() filtermaps.MatcherBackend {
return b.eth.filterMaps.NewMatcherBackend() return b.eth.filterMaps.NewMatcherBackend()
} }

221
eth/filters/filter.go
View file

@ -146,25 +146,29 @@ func (f *Filter) Logs(ctx context.Context) ([]*types.Log, error) {
} }
const ( const (
rangeLogsTestSync = iota rangeLogsTestDone = iota // zero range
rangeLogsTestTrimmed rangeLogsTestSync // before sync; zero range
rangeLogsTestIndexed rangeLogsTestSynced // after sync; valid blocks range
rangeLogsTestUnindexed rangeLogsTestIndexed // individual search range
rangeLogsTestDone rangeLogsTestUnindexed // individual search range
rangeLogsTestResults // results range after search iteration
rangeLogsTestReorg // results range trimmed by reorg
) )
type rangeLogsTestEvent struct { type rangeLogsTestEvent struct {
event int event int
begin, end uint64 blocks common.Range[uint64]
} }
// searchSession represents a single search session. // searchSession represents a single search session.
type searchSession struct { type searchSession struct {
ctx context.Context ctx context.Context
filter *Filter filter *Filter
mb filtermaps.MatcherBackend mb filtermaps.MatcherBackend
syncRange filtermaps.SyncRange // latest synchronized state with the matcher syncRange filtermaps.SyncRange // latest synchronized state with the matcher
firstBlock, lastBlock uint64 // specified search range; each can be MaxUint64 chainView *filtermaps.ChainView // can be more recent than the indexed view in syncRange
// block ranges always refer to the current chainView
firstBlock, lastBlock uint64 // specified search range; MaxUint64 means latest block
searchRange common.Range[uint64] // actual search range; end trimmed to latest head 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) matchRange common.Range[uint64] // range in which we have results (subset of searchRange)
matches []*types.Log // valid set of matches in matchRange matches []*types.Log // valid set of matches in matchRange
@ -182,84 +186,94 @@ func newSearchSession(ctx context.Context, filter *Filter, mb filtermaps.Matcher
} }
// enforce a consistent state before starting the search in order to be able // enforce a consistent state before starting the search in order to be able
// to determine valid range later // to determine valid range later
if err := s.syncMatcher(0); err != nil { var err error
s.syncRange, err = s.mb.SyncLogIndex(s.ctx)
if err != nil {
return nil, err
}
if err := s.updateChainView(); err != nil {
return nil, err return nil, err
} }
return s, nil return s, nil
} }
// syncMatcher performs a synchronization step with the matcher. The resulting // updateChainView updates to the latest view of the underlying chain and sets
// syncRange structure holds information about the latest range of indexed blocks // searchRange by replacing MaxUint64 (meaning latest block) with actual head
// and the guaranteed valid blocks whose log index have not been changed since // number in the specified search range.
// the previous synchronization. // If the session already had an existing chain view and set of matches then
// The function also performs trimming of the match set in order to always keep // it also trims part of the match set that a chain reorg might have invalidated.
// it consistent with the synced matcher state. func (s *searchSession) updateChainView() error {
// Tail trimming is only performed if the first block of the valid log index range // update chain view based on current chain head (might be more recent than
// is higher than trimTailThreshold. This is useful because unindexed log search // the indexed view of syncRange as the indexer updates it asynchronously
// is not affected by the valid tail (on the other hand, valid head is taken into // with some delay
// account in order to provide reorg safety, even though the log index is not used). newChainView := s.filter.sys.backend.CurrentView()
// In case of indexed search the tail is only trimmed if the first part of the if newChainView == nil {
// recently obtained results might be invalid. If guaranteed valid new results return errors.New("head block not available")
// have been added at the head of previously validated results then there is no
// need to discard those even if the index tail have been unindexed since that.
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
} }
head := newChainView.HeadNumber()
// update actual search range based on current head number // update actual search range based on current head number
first := min(s.firstBlock, s.syncRange.HeadNumber) firstBlock, lastBlock := s.firstBlock, s.lastBlock
last := min(s.lastBlock, s.syncRange.HeadNumber) if firstBlock == math.MaxUint64 {
s.searchRange = common.NewRange(first, last+1-first) firstBlock = head
// 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) if lastBlock == math.MaxUint64 {
s.trimMatches(trimRange) lastBlock = head
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}
}
} }
if firstBlock > lastBlock || lastBlock > head {
return errInvalidBlockRange
}
s.searchRange = common.NewRange(firstBlock, lastBlock+1-firstBlock)
if !s.matchRange.IsEmpty() {
// trim existing match set in case a reorg may have invalidated some results
trimRange := newChainView.SharedRange(s.chainView).Intersection(s.searchRange)
s.matchRange, s.matches = s.trimMatches(trimRange, s.matchRange, s.matches)
}
s.chainView = newChainView
return nil return nil
} }
// trimMatches removes any entries from the current set of matches that is outside // trimMatches removes any entries from the specified set of matches that is
// the given range. // outside the given range.
func (s *searchSession) trimMatches(trimRange common.Range[uint64]) { func (s *searchSession) trimMatches(trimRange, matchRange common.Range[uint64], matches []*types.Log) (common.Range[uint64], []*types.Log) {
s.matchRange = s.matchRange.Intersection(trimRange) newRange := matchRange.Intersection(trimRange)
if s.matchRange.IsEmpty() { if newRange == matchRange {
s.matches = nil return matchRange, matches
return
} }
for len(s.matches) > 0 && s.matches[0].BlockNumber < s.matchRange.First() { if newRange.IsEmpty() {
s.matches = s.matches[1:] return newRange, nil
} }
for len(s.matches) > 0 && s.matches[len(s.matches)-1].BlockNumber > s.matchRange.Last() { for len(matches) > 0 && matches[0].BlockNumber < newRange.First() {
s.matches = s.matches[:len(s.matches)-1] matches = matches[1:]
} }
for len(matches) > 0 && matches[len(matches)-1].BlockNumber > newRange.Last() {
matches = matches[:len(matches)-1]
}
return newRange, matches
} }
// searchInRange performs a single range search, either indexed or unindexed. // searchInRange performs a single range search, either indexed or unindexed.
func (s *searchSession) searchInRange(r common.Range[uint64], indexed bool) ([]*types.Log, error) { func (s *searchSession) searchInRange(r common.Range[uint64], indexed bool) (common.Range[uint64], []*types.Log, error) {
first, last := r.First(), r.Last()
if indexed { if indexed {
if s.filter.rangeLogsTestHook != nil { if s.filter.rangeLogsTestHook != nil {
s.filter.rangeLogsTestHook <- rangeLogsTestEvent{rangeLogsTestIndexed, first, last} s.filter.rangeLogsTestHook <- rangeLogsTestEvent{rangeLogsTestIndexed, r}
} }
results, err := s.filter.indexedLogs(s.ctx, s.mb, first, last) results, err := s.filter.indexedLogs(s.ctx, s.mb, r.First(), r.Last())
if err != filtermaps.ErrMatchAll { if err != filtermaps.ErrMatchAll {
return results, err // sync with filtermaps matcher
if s.filter.rangeLogsTestHook != nil {
s.filter.rangeLogsTestHook <- rangeLogsTestEvent{rangeLogsTestSync, common.Range[uint64]{}}
}
var syncErr error
if s.syncRange, syncErr = s.mb.SyncLogIndex(s.ctx); syncErr != nil {
return common.Range[uint64]{}, nil, syncErr
}
if s.filter.rangeLogsTestHook != nil {
s.filter.rangeLogsTestHook <- rangeLogsTestEvent{rangeLogsTestSynced, s.syncRange.ValidBlocks}
}
// discard everything that might be invalid
trimRange := s.syncRange.ValidBlocks.Intersection(s.chainView.SharedRange(s.syncRange.IndexedView))
matchRange, matches := s.trimMatches(trimRange, r, results)
return matchRange, matches, err
} }
// "match all" filters are not supported by filtermaps; fall back to // "match all" filters are not supported by filtermaps; fall back to
// unindexed search which is the most efficient in this case // unindexed search which is the most efficient in this case
@ -267,55 +281,41 @@ func (s *searchSession) searchInRange(r common.Range[uint64], indexed bool) ([]*
// fall through to unindexed case // fall through to unindexed case
} }
if s.filter.rangeLogsTestHook != nil { if s.filter.rangeLogsTestHook != nil {
s.filter.rangeLogsTestHook <- rangeLogsTestEvent{rangeLogsTestUnindexed, first, last} s.filter.rangeLogsTestHook <- rangeLogsTestEvent{rangeLogsTestUnindexed, r}
} }
return s.filter.unindexedLogs(s.ctx, first, last) matches, err := s.filter.unindexedLogs(s.ctx, s.chainView, r.First(), r.Last())
return r, matches, err
} }
// doSearchIteration performs a search on a range missing from an incomplete set // doSearchIteration performs a search on a range missing from an incomplete set
// of results, adds the new section and removes invalidated entries. // of results, adds the new section and removes invalidated entries.
func (s *searchSession) doSearchIteration() error { func (s *searchSession) doSearchIteration() error {
switch { 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(): case s.matchRange.IsEmpty():
// no results yet; try search in entire range // no results yet; try search in entire range
indexedSearchRange := s.searchRange.Intersection(s.syncRange.IndexedBlocks) indexedSearchRange := s.searchRange.Intersection(s.syncRange.IndexedBlocks)
var err error var err error
if s.forceUnindexed = indexedSearchRange.IsEmpty(); !s.forceUnindexed { if s.forceUnindexed = indexedSearchRange.IsEmpty(); !s.forceUnindexed {
// indexed search on the intersection of indexed and searched range // indexed search on the intersection of indexed and searched range
s.matchRange = indexedSearchRange s.matchRange, s.matches, err = s.searchInRange(indexedSearchRange, true)
s.matches, err = s.searchInRange(indexedSearchRange, true) return err
if err != nil {
return err
}
return s.syncMatcher(0) // trim everything that the matcher considers potentially invalid
} else { } else {
// no intersection of indexed and searched range; unindexed search on the whole searched range // no intersection of indexed and searched range; unindexed search on the whole searched range
s.matchRange = s.searchRange s.matchRange, s.matches, err = s.searchInRange(s.searchRange, false)
s.matches, err = s.searchInRange(s.searchRange, false) return err
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(): case !s.matchRange.IsEmpty() && s.matchRange.First() > s.searchRange.First():
// we have results but tail section is missing; do unindexed search for // we have results but tail section is missing; do unindexed search for
// the tail part but still allow indexed search for missing head section // the tail part but still allow indexed search for missing head section
tailRange := common.NewRange(s.searchRange.First(), s.matchRange.First()-s.searchRange.First()) tailRange := common.NewRange(s.searchRange.First(), s.matchRange.First()-s.searchRange.First())
tailMatches, err := s.searchInRange(tailRange, false) _, tailMatches, err := s.searchInRange(tailRange, false)
if err != nil { if err != nil {
return err return err
} }
s.matches = append(tailMatches, s.matches...) s.matches = append(tailMatches, s.matches...)
s.matchRange = tailRange.Union(s.matchRange) s.matchRange = tailRange.Union(s.matchRange)
return s.syncMatcher(math.MaxUint64) // unindexed search is not affected by the tail of valid range return nil
case !s.matchRange.IsEmpty() && s.matchRange.First() == s.searchRange.First() && s.searchRange.AfterLast() > s.matchRange.AfterLast(): case !s.matchRange.IsEmpty() && s.matchRange.First() == s.searchRange.First() && s.searchRange.AfterLast() > s.matchRange.AfterLast():
// we have results but head section is missing // we have results but head section is missing
@ -329,17 +329,15 @@ func (s *searchSession) doSearchIteration() error {
s.forceUnindexed = true s.forceUnindexed = true
} }
} }
headMatches, err := s.searchInRange(headRange, !s.forceUnindexed) headMatchRange, headMatches, err := s.searchInRange(headRange, !s.forceUnindexed)
if err != nil { if headMatchRange.First() != s.matchRange.AfterLast() {
// improbable corner case, first part of new head range invalidated by tail unindexing
s.matches, s.matchRange = headMatches, headMatchRange
return err return err
} }
s.matches = append(s.matches, headMatches...) s.matches = append(s.matches, headMatches...)
s.matchRange = s.matchRange.Union(headRange) s.matchRange = s.matchRange.Union(headMatchRange)
if s.forceUnindexed { return err
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: default:
panic("invalid search session state") panic("invalid search session state")
@ -349,7 +347,7 @@ func (s *searchSession) doSearchIteration() error {
func (f *Filter) rangeLogs(ctx context.Context, firstBlock, lastBlock uint64) ([]*types.Log, error) { func (f *Filter) rangeLogs(ctx context.Context, firstBlock, lastBlock uint64) ([]*types.Log, error) {
if f.rangeLogsTestHook != nil { if f.rangeLogsTestHook != nil {
defer func() { defer func() {
f.rangeLogsTestHook <- rangeLogsTestEvent{rangeLogsTestDone, 0, 0} f.rangeLogsTestHook <- rangeLogsTestEvent{rangeLogsTestDone, common.Range[uint64]{}}
close(f.rangeLogsTestHook) close(f.rangeLogsTestHook)
}() }()
} }
@ -368,6 +366,16 @@ func (f *Filter) rangeLogs(ctx context.Context, firstBlock, lastBlock uint64) ([
if err := session.doSearchIteration(); err != nil { if err := session.doSearchIteration(); err != nil {
return session.matches, err return session.matches, err
} }
if f.rangeLogsTestHook != nil {
f.rangeLogsTestHook <- rangeLogsTestEvent{rangeLogsTestResults, session.matchRange}
}
mr := session.matchRange
if err := session.updateChainView(); err != nil {
return session.matches, err
}
if f.rangeLogsTestHook != nil && session.matchRange != mr {
f.rangeLogsTestHook <- rangeLogsTestEvent{rangeLogsTestReorg, session.matchRange}
}
} }
return session.matches, nil return session.matches, nil
} }
@ -382,7 +390,7 @@ func (f *Filter) indexedLogs(ctx context.Context, mb filtermaps.MatcherBackend,
// unindexedLogs returns the logs matching the filter criteria based on raw block // unindexedLogs returns the logs matching the filter criteria based on raw block
// iteration and bloom matching. // iteration and bloom matching.
func (f *Filter) unindexedLogs(ctx context.Context, begin, end uint64) ([]*types.Log, error) { func (f *Filter) unindexedLogs(ctx context.Context, chainView *filtermaps.ChainView, begin, end uint64) ([]*types.Log, error) {
start := time.Now() start := time.Now()
log.Debug("Performing unindexed log search", "begin", begin, "end", end) log.Debug("Performing unindexed log search", "begin", begin, "end", end)
var matches []*types.Log var matches []*types.Log
@ -392,9 +400,14 @@ func (f *Filter) unindexedLogs(ctx context.Context, begin, end uint64) ([]*types
return matches, ctx.Err() return matches, ctx.Err()
default: default:
} }
header, err := f.sys.backend.HeaderByNumber(ctx, rpc.BlockNumber(blockNumber)) if blockNumber > chainView.HeadNumber() {
if header == nil || err != nil { // check here so that we can return matches up until head along with
return matches, err // the error
return matches, errInvalidBlockRange
}
header := chainView.Header(blockNumber)
if header == nil {
return matches, errors.New("header not found")
} }
found, err := f.blockLogs(ctx, header) found, err := f.blockLogs(ctx, header)
if err != nil { if err != nil {

1
eth/filters/filter_system.go
View file

@ -71,6 +71,7 @@ type Backend interface {
SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription
SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription
CurrentView() *filtermaps.ChainView
NewMatcherBackend() filtermaps.MatcherBackend NewMatcherBackend() filtermaps.MatcherBackend
} }

5
eth/filters/filter_system_test.go
View file

@ -154,6 +154,11 @@ func (b *testBackend) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subsc
return b.chainFeed.Subscribe(ch) return b.chainFeed.Subscribe(ch)
} }
func (b *testBackend) CurrentView() *filtermaps.ChainView {
head := b.CurrentBlock()
return filtermaps.NewChainView(b, head.Number.Uint64(), head.Hash())
}
func (b *testBackend) NewMatcherBackend() filtermaps.MatcherBackend { func (b *testBackend) NewMatcherBackend() filtermaps.MatcherBackend {
return b.fm.NewMatcherBackend() return b.fm.NewMatcherBackend()
} }

137
eth/filters/filter_test.go
View file

@ -453,7 +453,8 @@ func TestRangeLogs(t *testing.T) {
addresses = []common.Address{{}} addresses = []common.Address{{}}
) )
expEvent := func(exp rangeLogsTestEvent) { expEvent := func(expEvent int, expFirst, expAfterLast uint64) {
exp := rangeLogsTestEvent{expEvent, common.NewRange[uint64](expFirst, expAfterLast-expFirst)}
event++ event++
ev := <-filter.rangeLogsTestHook ev := <-filter.rangeLogsTestHook
if ev != exp { if ev != exp {
@ -472,7 +473,6 @@ func TestRangeLogs(t *testing.T) {
for range filter.rangeLogsTestHook { for range filter.rangeLogsTestHook {
} }
}(filter) }(filter)
expEvent(rangeLogsTestEvent{rangeLogsTestTrimmed, 0, 0})
} }
updateHead := func() { updateHead := func() {
@ -483,81 +483,122 @@ func TestRangeLogs(t *testing.T) {
// test case #1 // test case #1
newFilter(300, 500) newFilter(300, 500)
expEvent(rangeLogsTestEvent{rangeLogsTestIndexed, 401, 500}) expEvent(rangeLogsTestIndexed, 401, 501)
expEvent(rangeLogsTestEvent{rangeLogsTestSync, 401, 500}) expEvent(rangeLogsTestSync, 0, 0)
expEvent(rangeLogsTestEvent{rangeLogsTestTrimmed, 401, 500}) expEvent(rangeLogsTestSynced, 401, 601)
expEvent(rangeLogsTestEvent{rangeLogsTestUnindexed, 300, 400}) expEvent(rangeLogsTestResults, 401, 501)
expEvent(rangeLogsTestUnindexed, 300, 401)
if _, err := bc.InsertChain(chain[600:700]); err != nil { if _, err := bc.InsertChain(chain[600:700]); err != nil {
t.Fatal(err) t.Fatal(err)
} }
updateHead() updateHead()
expEvent(rangeLogsTestEvent{rangeLogsTestSync, 300, 500}) expEvent(rangeLogsTestResults, 300, 501)
expEvent(rangeLogsTestEvent{rangeLogsTestTrimmed, 300, 500}) // unindexed search is not affected by trimmed tail expEvent(rangeLogsTestDone, 0, 0)
expEvent(rangeLogsTestEvent{rangeLogsTestDone, 0, 0})
// test case #2 // test case #2
newFilter(400, int64(rpc.LatestBlockNumber)) newFilter(400, int64(rpc.LatestBlockNumber))
expEvent(rangeLogsTestEvent{rangeLogsTestIndexed, 501, 700}) expEvent(rangeLogsTestIndexed, 501, 701)
if _, err := bc.InsertChain(chain[700:800]); err != nil { if _, err := bc.InsertChain(chain[700:800]); err != nil {
t.Fatal(err) t.Fatal(err)
} }
updateHead() updateHead()
expEvent(rangeLogsTestEvent{rangeLogsTestSync, 501, 700}) expEvent(rangeLogsTestSync, 0, 0)
expEvent(rangeLogsTestEvent{rangeLogsTestTrimmed, 601, 698}) expEvent(rangeLogsTestSynced, 601, 699)
expEvent(rangeLogsTestEvent{rangeLogsTestUnindexed, 400, 600}) expEvent(rangeLogsTestResults, 601, 699)
expEvent(rangeLogsTestEvent{rangeLogsTestSync, 400, 698}) expEvent(rangeLogsTestUnindexed, 400, 601)
expEvent(rangeLogsTestEvent{rangeLogsTestTrimmed, 400, 698}) expEvent(rangeLogsTestResults, 400, 699)
expEvent(rangeLogsTestEvent{rangeLogsTestIndexed, 699, 800}) expEvent(rangeLogsTestIndexed, 699, 801)
if err := bc.SetHead(750); err != nil { if _, err := bc.SetCanonical(chain[749]); err != nil { // set head to block 750
t.Fatal(err) t.Fatal(err)
} }
updateHead() updateHead()
expEvent(rangeLogsTestEvent{rangeLogsTestSync, 400, 800}) expEvent(rangeLogsTestSync, 0, 0)
expEvent(rangeLogsTestEvent{rangeLogsTestTrimmed, 400, 748}) expEvent(rangeLogsTestSynced, 601, 749)
expEvent(rangeLogsTestEvent{rangeLogsTestIndexed, 749, 750}) expEvent(rangeLogsTestResults, 400, 749)
expEvent(rangeLogsTestEvent{rangeLogsTestSync, 400, 750}) expEvent(rangeLogsTestIndexed, 749, 751)
expEvent(rangeLogsTestEvent{rangeLogsTestTrimmed, 400, 750}) expEvent(rangeLogsTestSync, 0, 0)
expEvent(rangeLogsTestEvent{rangeLogsTestDone, 0, 0}) expEvent(rangeLogsTestSynced, 551, 751)
expEvent(rangeLogsTestResults, 400, 751)
expEvent(rangeLogsTestDone, 0, 0)
// test case #3 // test case #3
newFilter(int64(rpc.LatestBlockNumber), int64(rpc.LatestBlockNumber)) newFilter(int64(rpc.LatestBlockNumber), int64(rpc.LatestBlockNumber))
expEvent(rangeLogsTestEvent{rangeLogsTestIndexed, 750, 750}) expEvent(rangeLogsTestIndexed, 750, 751)
if err := bc.SetHead(740); err != nil { if _, err := bc.SetCanonical(chain[739]); err != nil {
t.Fatal(err) t.Fatal(err)
} }
updateHead() updateHead()
expEvent(rangeLogsTestEvent{rangeLogsTestSync, 750, 750}) expEvent(rangeLogsTestSync, 0, 0)
expEvent(rangeLogsTestEvent{rangeLogsTestTrimmed, 0, 0}) expEvent(rangeLogsTestSynced, 551, 739)
expEvent(rangeLogsTestEvent{rangeLogsTestIndexed, 740, 740}) expEvent(rangeLogsTestResults, 0, 0)
expEvent(rangeLogsTestIndexed, 740, 741)
if _, err := bc.InsertChain(chain[740:750]); err != nil { if _, err := bc.InsertChain(chain[740:750]); err != nil {
t.Fatal(err) t.Fatal(err)
} }
updateHead() updateHead()
expEvent(rangeLogsTestEvent{rangeLogsTestSync, 740, 740}) expEvent(rangeLogsTestSync, 0, 0)
expEvent(rangeLogsTestEvent{rangeLogsTestTrimmed, 0, 0}) expEvent(rangeLogsTestSynced, 551, 739)
expEvent(rangeLogsTestEvent{rangeLogsTestIndexed, 750, 750}) expEvent(rangeLogsTestResults, 0, 0)
expEvent(rangeLogsTestEvent{rangeLogsTestSync, 750, 750}) expEvent(rangeLogsTestIndexed, 750, 751)
expEvent(rangeLogsTestEvent{rangeLogsTestTrimmed, 750, 750}) expEvent(rangeLogsTestSync, 0, 0)
expEvent(rangeLogsTestEvent{rangeLogsTestDone, 0, 0}) expEvent(rangeLogsTestSynced, 551, 751)
expEvent(rangeLogsTestResults, 750, 751)
expEvent(rangeLogsTestDone, 0, 0)
// test case #4 // test case #4
if _, err := bc.SetCanonical(chain[499]); err != nil {
t.Fatal(err)
}
updateHead()
newFilter(400, int64(rpc.LatestBlockNumber)) newFilter(400, int64(rpc.LatestBlockNumber))
expEvent(rangeLogsTestEvent{rangeLogsTestIndexed, 551, 750}) expEvent(rangeLogsTestIndexed, 400, 501)
expEvent(rangeLogsTestEvent{rangeLogsTestSync, 551, 750}) if _, err := bc.InsertChain(chain[500:650]); err != nil {
expEvent(rangeLogsTestEvent{rangeLogsTestTrimmed, 551, 750}) t.Fatal(err)
expEvent(rangeLogsTestEvent{rangeLogsTestUnindexed, 400, 550}) }
updateHead()
expEvent(rangeLogsTestSync, 0, 0)
expEvent(rangeLogsTestSynced, 451, 499)
expEvent(rangeLogsTestResults, 451, 499)
expEvent(rangeLogsTestUnindexed, 400, 451)
expEvent(rangeLogsTestResults, 400, 499)
// indexed head extension seems possible
expEvent(rangeLogsTestIndexed, 499, 651)
// further head extension causes tail unindexing in searched range
if _, err := bc.InsertChain(chain[650:750]); err != nil {
t.Fatal(err)
}
updateHead()
expEvent(rangeLogsTestSync, 0, 0)
expEvent(rangeLogsTestSynced, 551, 649)
// tail trimmed to 551; cannot merge with existing results
expEvent(rangeLogsTestResults, 551, 649)
expEvent(rangeLogsTestUnindexed, 400, 551)
expEvent(rangeLogsTestResults, 400, 649)
expEvent(rangeLogsTestIndexed, 649, 751)
expEvent(rangeLogsTestSync, 0, 0)
expEvent(rangeLogsTestSynced, 551, 751)
expEvent(rangeLogsTestResults, 400, 751)
expEvent(rangeLogsTestDone, 0, 0)
// test case #5
newFilter(400, int64(rpc.LatestBlockNumber))
expEvent(rangeLogsTestIndexed, 551, 751)
expEvent(rangeLogsTestSync, 0, 0)
expEvent(rangeLogsTestSynced, 551, 751)
expEvent(rangeLogsTestResults, 551, 751)
expEvent(rangeLogsTestUnindexed, 400, 551)
if _, err := bc.InsertChain(chain[750:1000]); err != nil { if _, err := bc.InsertChain(chain[750:1000]); err != nil {
t.Fatal(err) t.Fatal(err)
} }
updateHead() updateHead()
expEvent(rangeLogsTestEvent{rangeLogsTestSync, 400, 750}) expEvent(rangeLogsTestResults, 400, 751)
// indexed range affected by tail pruning so we have to discard the entire // indexed tail already beyond results head; revert to unindexed head search
// match set expEvent(rangeLogsTestUnindexed, 751, 1001)
expEvent(rangeLogsTestEvent{rangeLogsTestTrimmed, 0, 0}) if _, err := bc.SetCanonical(chain[899]); err != nil {
expEvent(rangeLogsTestEvent{rangeLogsTestIndexed, 801, 1000}) t.Fatal(err)
expEvent(rangeLogsTestEvent{rangeLogsTestSync, 801, 1000}) }
expEvent(rangeLogsTestEvent{rangeLogsTestTrimmed, 801, 1000}) updateHead()
expEvent(rangeLogsTestEvent{rangeLogsTestUnindexed, 400, 800}) expEvent(rangeLogsTestResults, 400, 1001)
expEvent(rangeLogsTestEvent{rangeLogsTestSync, 400, 1000}) expEvent(rangeLogsTestReorg, 400, 901)
expEvent(rangeLogsTestEvent{rangeLogsTestTrimmed, 400, 1000}) expEvent(rangeLogsTestDone, 0, 0)
} }

3
internal/ethapi/api_test.go
View file

@ -619,6 +619,9 @@ func (b testBackend) SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent)
func (b testBackend) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription { func (b testBackend) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription {
panic("implement me") panic("implement me")
} }
func (b testBackend) CurrentView() *filtermaps.ChainView {
panic("implement me")
}
func (b testBackend) NewMatcherBackend() filtermaps.MatcherBackend { func (b testBackend) NewMatcherBackend() filtermaps.MatcherBackend {
panic("implement me") panic("implement me")
} }

1
internal/ethapi/backend.go
View file

@ -95,6 +95,7 @@ type Backend interface {
SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription
SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription
CurrentView() *filtermaps.ChainView
NewMatcherBackend() filtermaps.MatcherBackend NewMatcherBackend() filtermaps.MatcherBackend
} }

1
internal/ethapi/transaction_args_test.go
View file

@ -401,6 +401,7 @@ func (b *backendMock) SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent)
func (b *backendMock) Engine() consensus.Engine { return nil } func (b *backendMock) Engine() consensus.Engine { return nil }
func (b *backendMock) CurrentView() *filtermaps.ChainView { return nil }
func (b *backendMock) NewMatcherBackend() filtermaps.MatcherBackend { return nil } func (b *backendMock) NewMatcherBackend() filtermaps.MatcherBackend { return nil }
func (b *backendMock) HistoryPruningCutoff() uint64 { return 0 } func (b *backendMock) HistoryPruningCutoff() uint64 { return 0 }