Skip to content

Commit b33c95f

Browse files
pkieltykapkieltyka
committed
refactor(ethreceipts): pass block processing contexts explicitly
1 parent 654c149 commit b33c95f

5 files changed

Lines changed: 24 additions & 58 deletions

File tree

‎FIXES.md‎

Lines changed: 0 additions & 26 deletions
This file was deleted.

‎ethreceipts/codegenie_regression_test.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -573,9 +573,9 @@ func TestReceiptsFixReadoptionAfterRetention(t *testing.T) {
573573
t.Helper()
574574
var err error
575575
if cached {
576-
_, err = l.processCachedBlocksContext(context.Background(), batch, []*subscriber{s}, [][]Filterer{s.filterers()})
576+
_, err = l.processCachedBlocks(context.Background(), batch, []*subscriber{s}, [][]Filterer{s.filterers()})
577577
} else {
578-
_, err = l.processBlocksContext(context.Background(), batch, []*subscriber{s}, [][]Filterer{s.filterers()})
578+
_, err = l.processBlocks(context.Background(), batch, []*subscriber{s}, [][]Filterer{s.filterers()})
579579
}
580580
if err != nil {
581581
t.Fatal(err)

‎ethreceipts/ethreceipts.go‎

Lines changed: 11 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -98,13 +98,12 @@ type ReceiptsListener struct {
9898
registerFiltersCh chan registerFilters
9999
filterSem chan struct{}
100100

101-
ctx context.Context
102101
ctxStop context.CancelFunc
103102
running int32
104-
lifecycleMu sync.Mutex
103+
lifecycleMu sync.Mutex // protects Run/Stop cancellation handle and running transitions
105104
receiptMu sync.Mutex
106105
blockStates map[common.Hash]blockState
107-
reorgRevision uint64
106+
reorgRevision uint64 // removal sequence used to reject hash queries started before invalidation
108107
mu sync.RWMutex
109108
}
110109

@@ -204,7 +203,7 @@ func (l *ReceiptsListener) Run(ctx context.Context) error {
204203
return fmt.Errorf("ethreceipts: already running")
205204
}
206205
runCtx, cancel := context.WithCancel(ctx)
207-
l.ctx, l.ctxStop = runCtx, cancel
206+
l.ctxStop = cancel
208207
atomic.StoreInt32(&l.running, 1)
209208
l.lifecycleMu.Unlock()
210209
defer func() {
@@ -702,7 +701,7 @@ func (l *ReceiptsListener) listener(runCtx context.Context) error {
702701

703702
// Search our local blocks cache from monitor retention list, and notify subscriber
704703
// of any matches found by publishing receipts.
705-
matchedList, err := l.processCachedBlocksContext(ctx, blocks, []*subscriber{reg.subscriber}, [][]Filterer{filters})
704+
matchedList, err := l.processCachedBlocks(ctx, blocks, []*subscriber{reg.subscriber}, [][]Filterer{filters})
706705
if err != nil {
707706
l.log.Warn(fmt.Sprintf("ethreceipts: failed to process blocks during new filter registration: %v", err))
708707
}
@@ -776,7 +775,7 @@ func (l *ReceiptsListener) listener(runCtx context.Context) error {
776775
}
777776

778777
// Match blocks against subscribers[i] X filters[i][..]
779-
matchedList, err := l.processBlocksContext(ctx, blocks, subscribers, filters)
778+
matchedList, err := l.processBlocks(ctx, blocks, subscribers, filters)
780779
if err != nil {
781780
l.log.Warn(fmt.Sprintf("ethreceipts: failed to process blocks: %v", err))
782781
}
@@ -838,20 +837,12 @@ func (l *ReceiptsListener) listener(runCtx context.Context) error {
838837

839838
// processBlocks attempts to match blocks against subscriber[i] X filterers[i].. list of filters. There is
840839
// a corresponding list of filters[i] for each subscriber[i].
841-
func (l *ReceiptsListener) processBlocks(blocks ethmonitor.Blocks, subscribers []*subscriber, filterers [][]Filterer) ([][]bool, error) {
842-
l.lifecycleMu.Lock()
843-
ctx := l.ctx
844-
l.lifecycleMu.Unlock()
845-
if ctx == nil {
846-
ctx = context.Background()
847-
}
848-
return l.processCachedBlocksContext(ctx, blocks, subscribers, filterers)
849-
}
850-
851-
func (l *ReceiptsListener) processBlocksContext(ctx context.Context, blocks ethmonitor.Blocks, subscribers []*subscriber, filterers [][]Filterer) ([][]bool, error) {
840+
func (l *ReceiptsListener) processBlocks(ctx context.Context, blocks ethmonitor.Blocks, subscribers []*subscriber, filterers [][]Filterer) ([][]bool, error) {
852841
return l.processBlockEvents(ctx, blocks, subscribers, filterers, true)
853842
}
854-
func (l *ReceiptsListener) processCachedBlocksContext(ctx context.Context, blocks ethmonitor.Blocks, subscribers []*subscriber, filterers [][]Filterer) ([][]bool, error) {
843+
844+
// Cached snapshots may match receipts, but cannot authorize canonical re-adoption.
845+
func (l *ReceiptsListener) processCachedBlocks(ctx context.Context, blocks ethmonitor.Blocks, subscribers []*subscriber, filterers [][]Filterer) ([][]bool, error) {
855846
return l.processBlockEvents(ctx, blocks, subscribers, filterers, false)
856847
}
857848
func (l *ReceiptsListener) processBlockEvents(ctx context.Context, blocks ethmonitor.Blocks, subscribers []*subscriber, filterers [][]Filterer, canonicalEvents bool) ([][]bool, error) {
@@ -1135,6 +1126,8 @@ func (l *ReceiptsListener) latestBlockNum(ctx context.Context) *big.Int {
11351126
}
11361127

11371128
func getChainID(ctx context.Context, provider ethrpc.Interface) (*big.Int, error) {
1129+
// breaker v0.2.0 sleeps without checking cancellation between attempts.
1130+
// Preserve its retry policy here while making startup stoppable during backoff.
11381131
delay := time.Second
11391132
for attempt := 0; ; attempt++ {
11401133
if ctx.Err() != nil {

‎ethreceipts/hardening_test.go‎

Lines changed: 9 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -82,7 +82,6 @@ func hardeningListener(t *testing.T, p *hardeningProvider, opts Options, blocks
8282
if err != nil {
8383
t.Fatal(err)
8484
}
85-
l.ctx = context.Background()
8685
l.br = breaker.New(log, 0, 1, 0)
8786
return l
8887
}
@@ -133,7 +132,7 @@ func hardeningNoReceipt(t *testing.T, s Subscription) {
133132
}
134133
func hardeningProcess(t *testing.T, l *ReceiptsListener, s *subscriber, blocks ...*ethmonitor.Block) {
135134
t.Helper()
136-
if _, err := l.processBlocks(blocks, []*subscriber{s}, [][]Filterer{s.Filters()}); err != nil {
135+
if _, err := l.processCachedBlocks(context.Background(), blocks, []*subscriber{s}, [][]Filterer{s.Filters()}); err != nil {
137136
t.Fatal(err)
138137
}
139138
}
@@ -365,7 +364,7 @@ func TestHardeningLimitOneSnapshots(t *testing.T) {
365364
s := l.Subscribe(FilterLogs(func([]*types.Log) bool { return true }).LimitOne(true)).(*subscriber)
366365
defer s.Unsubscribe()
367366
fs := s.Filters()
368-
if _, err := l.processBlocks(blocks, []*subscriber{s}, [][]Filterer{fs}); err != nil {
367+
if _, err := l.processCachedBlocks(context.Background(), blocks, []*subscriber{s}, [][]Filterer{fs}); err != nil {
369368
t.Fatal(err)
370369
}
371370
r := hardeningRead(t, s)
@@ -529,7 +528,7 @@ func TestHardeningFinalityWaitCancellation(t *testing.T) {
529528
func TestHardeningConcurrentRunStop(t *testing.T) {
530529
for i := 0; i < 5; i++ {
531530
l := hardeningListener(t, &hardeningProvider{}, hardeningOptions())
532-
l.ctx, l.ctxStop = context.WithCancel(context.Background())
531+
_, l.ctxStop = context.WithCancel(context.Background())
533532
ctx, cancel := context.WithCancel(context.Background())
534533
cancel()
535534
var wg sync.WaitGroup
@@ -575,7 +574,7 @@ func TestHardeningInflightRollbackRemining(t *testing.T) {
575574
snapshot := s.Filters()
576575
oldDone := make(chan error, 1)
577576
go func() {
578-
_, err := l.processBlocks(ethmonitor.Blocks{old}, []*subscriber{s}, [][]Filterer{snapshot})
577+
_, err := l.processCachedBlocks(context.Background(), ethmonitor.Blocks{old}, []*subscriber{s}, [][]Filterer{snapshot})
579578
oldDone <- err
580579
}()
581580
<-entered
@@ -610,7 +609,7 @@ func TestHardeningInflightRollbackRemining(t *testing.T) {
610609
t.Fatal("old fetch did not finish")
611610
}
612611
// A registration snapshot retained before rollback must also remain invalid.
613-
if _, err := l.processBlocks(ethmonitor.Blocks{old}, []*subscriber{s}, [][]Filterer{snapshot}); err != nil {
612+
if _, err := l.processCachedBlocks(context.Background(), ethmonitor.Blocks{old}, []*subscriber{s}, [][]Filterer{snapshot}); err != nil {
614613
t.Fatal(err)
615614
}
616615
hardeningNoReceipt(t, s)
@@ -658,11 +657,11 @@ func TestHardeningConcurrentLimitOne(t *testing.T) {
658657
fs := s.Filters()
659658
done := make(chan error, 1)
660659
go func() {
661-
_, err := l.processBlocks(ethmonitor.Blocks{b1}, []*subscriber{s}, [][]Filterer{fs})
660+
_, err := l.processCachedBlocks(context.Background(), ethmonitor.Blocks{b1}, []*subscriber{s}, [][]Filterer{fs})
662661
done <- err
663662
}()
664663
<-entered
665-
if _, err := l.processBlocks(ethmonitor.Blocks{b2}, []*subscriber{s}, [][]Filterer{fs}); err != nil {
664+
if _, err := l.processCachedBlocks(context.Background(), ethmonitor.Blocks{b2}, []*subscriber{s}, [][]Filterer{fs}); err != nil {
666665
t.Fatal(err)
667666
}
668667
close(release)
@@ -926,7 +925,7 @@ func TestHardeningSameHashReadoptionRejectsOldWork(t *testing.T) {
926925
fs := s.Filters()
927926
oldDone := make(chan error, 1)
928927
go func() {
929-
_, err := l.processBlocks(ethmonitor.Blocks{block}, []*subscriber{s}, [][]Filterer{fs})
928+
_, err := l.processCachedBlocks(context.Background(), ethmonitor.Blocks{block}, []*subscriber{s}, [][]Filterer{fs})
930929
oldDone <- err
931930
}()
932931
<-entered
@@ -944,7 +943,7 @@ func TestHardeningSameHashReadoptionRejectsOldWork(t *testing.T) {
944943
t.Errorf("stale snapshot cleared removal marker: %d fetches", calls.Load())
945944
}
946945
// The live block-processing entry point receives the canonical Added event.
947-
if _, err := l.processBlocksContext(context.Background(), ethmonitor.Blocks{block}, []*subscriber{s}, [][]Filterer{fs}); err != nil {
946+
if _, err := l.processBlocks(context.Background(), ethmonitor.Blocks{block}, []*subscriber{s}, [][]Filterer{fs}); err != nil {
948947
t.Fatal(err)
949948
}
950949
select {

‎ethreceipts/ownership_regression_test.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -185,7 +185,7 @@ func TestOwnershipReaddRejectsOldWorker(t *testing.T) {
185185
defer s.Unsubscribe()
186186
done := make(chan error, 1)
187187
go func() {
188-
_, err := l.processBlocks(ethmonitor.Blocks{b1}, []*subscriber{s}, [][]Filterer{s.Filters()})
188+
_, err := l.processCachedBlocks(context.Background(), ethmonitor.Blocks{b1}, []*subscriber{s}, [][]Filterer{s.Filters()})
189189
done <- err
190190
}()
191191
<-entered
@@ -348,7 +348,7 @@ func TestOwnershipStaleFetchPreservesCache(t *testing.T) {
348348
hardeningProcess(t, l, s, &removed)
349349
hardeningRead(t, s)
350350
if sameHash {
351-
if _, err := l.processBlocksContext(context.Background(), ethmonitor.Blocks{canonical}, []*subscriber{s}, [][]Filterer{s.Filters()}); err != nil {
351+
if _, err := l.processBlocks(context.Background(), ethmonitor.Blocks{canonical}, []*subscriber{s}, [][]Filterer{s.Filters()}); err != nil {
352352
t.Fatal(err)
353353
}
354354
} else {

0 commit comments

Comments
 (0)