fix race conditions

Author: xiam

ethreceipts/ethreceipts.go

@@ -316,6 +316,9 @@ func (l *ReceiptsListener) FetchTransactionReceiptWithFilter(ctx context.Context
 
 	sub := l.Subscribe(query)
 
+	// Use an internal context to coordinate cancellation and cleanup
+	internalCtx, internalCancel := context.WithCancel(ctx)
+
 	// Use a WaitGroup to ensure the goroutine cleans up before the function returns
 	var wg sync.WaitGroup
 
@@ -353,6 +356,7 @@ func (l *ReceiptsListener) FetchTransactionReceiptWithFilter(ctx context.Context
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
+		defer internalCancel()
 		defer sub.Unsubscribe()
 		defer close(mined)
 		defer close(finalized)
@@ -366,7 +370,7 @@ func (l *ReceiptsListener) FetchTransactionReceiptWithFilter(ctx context.Context
 
 		for {
 			select {
-			case <-ctx.Done():
+			case <-internalCtx.Done():
 				return
 
 			case <-sub.Done():
@@ -425,9 +429,11 @@ func (l *ReceiptsListener) FetchTransactionReceiptWithFilter(ctx context.Context
 	// Wait for the first mined receipt or an exit signal
 	select {
 	case <-ctx.Done():
+		internalCancel()
 		wg.Wait() // Ensure cleanup
 		return nil, nil, ctx.Err()
 	case <-sub.Done():
+		internalCancel()
 		wg.Wait() // Ensure cleanup
 		return nil, nil, ErrSubscriptionClosed
 	case <-exhausted:
@@ -437,6 +443,7 @@ func (l *ReceiptsListener) FetchTransactionReceiptWithFilter(ctx context.Context
 	case receipt, ok := <-mined:
 		if !ok {
 			// Mined channel closed without sending, implies goroutine exited early.
+			internalCancel()
 			wg.Wait() // Ensure cleanup
 			// Check if exhaustion occurred
 			select {
@@ -667,9 +674,9 @@ func (l *ReceiptsListener) listener() error {
 				if reorg {
 					for _, list := range filters {
 						for _, filterer := range list {
-							if f, _ := filterer.(*filter); f != nil {
-								f.startBlockNum = latestBlockNum
-								f.lastMatchBlockNum = 0
+							if f, ok := filterer.(*filter); ok {
+								f.setStartBlockNum(latestBlockNum)
+								f.setLastMatchBlockNum(0)
 							}
 						}
 					}
@@ -686,12 +693,12 @@ func (l *ReceiptsListener) listener() error {
 					for y, matched := range list {
 						filterer := filters[x][y]
 						if matched || filterer.StartBlockNum() == 0 {
-							if f, _ := filterer.(*filter); f != nil {
-								if f.startBlockNum == 0 {
-									f.startBlockNum = latestBlockNum
+							if f, ok := filterer.(*filter); ok {
+								if f.StartBlockNum() == 0 {
+									f.setStartBlockNum(latestBlockNum)
 								}
 								if matched {
-									f.lastMatchBlockNum = latestBlockNum
+									f.setLastMatchBlockNum(latestBlockNum)
 								}
 							}
 						} else {
@@ -713,10 +720,8 @@ func (l *ReceiptsListener) listener() error {
 									subscriber := subscribers[x]
 									subscriber.RemoveFilter(filterer)
 
-									select {
-									case <-f.Exhausted():
-									default:
-										close(f.exhausted)
+									if f, ok := filterer.(*filter); ok {
+										f.closeExhausted()
 									}
 								}
 							}
@@ -883,7 +888,7 @@ func (l *ReceiptsListener) searchFilterOnChain(ctx context.Context, subscriber *
 			}
 
 			if f, ok := item.filterer.(*filter); ok {
-				f.lastMatchBlockNum = r.BlockNumber.Uint64()
+				f.setLastMatchBlockNum(r.BlockNumber.Uint64())
 			}
 
 			receipt := Receipt{

ethreceipts/filterer.go

@@ -2,6 +2,7 @@ package ethreceipts
 
 import (
 	"context"
+	"sync"
 
 	"github.com/0xsequence/ethkit"
 	"github.com/0xsequence/ethkit/go-ethereum/core/types"
@@ -158,14 +159,16 @@ type filter struct {
 	options FilterOptions
 	cond    FilterCond
 
+	mu sync.RWMutex
 	// startBlockNum is the first block number observed once filter is active
 	startBlockNum uint64
 
 	// lastMatchBlockNum is the block number where a last match occured
 	lastMatchBlockNum uint64
 
 	// exhausted signals if the filter hit MaxWait
-	exhausted chan struct{}
+	exhausted     chan struct{}
+	exhaustedOnce sync.Once
 }
 
 var (
@@ -254,13 +257,35 @@ func (f *filter) Match(ctx context.Context, receipt Receipt) (bool, error) {
 }
 
 func (f *filter) StartBlockNum() uint64 {
+	f.mu.RLock()
+	defer f.mu.RUnlock()
 	return f.startBlockNum
 }
 
 func (f *filter) LastMatchBlockNum() uint64 {
+	f.mu.RLock()
+	defer f.mu.RUnlock()
 	return f.lastMatchBlockNum
 }
 
+func (f *filter) setStartBlockNum(num uint64) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.startBlockNum = num
+}
+
+func (f *filter) setLastMatchBlockNum(num uint64) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.lastMatchBlockNum = num
+}
+
+func (f *filter) closeExhausted() {
+	f.exhaustedOnce.Do(func() {
+		close(f.exhausted)
+	})
+}
+
 func (f *filter) Exhausted() <-chan struct{} {
 	return f.exhausted
 }

ethreceipts/subscription.go

@@ -319,7 +319,7 @@ func (s *subscriber) addPendingReceipt(receipt Receipt, filterer Filterer) {
 func (s *subscriber) retryPendingReceipts(ctx context.Context) {
 	s.retryMu.Lock()
 
-	// Collect receipts that are due for retry
+	// Create a snapshot of receipts that are due for retry
 	var toRetry []pendingReceipt
 	now := time.Now()
 
@@ -336,7 +336,7 @@ func (s *subscriber) retryPendingReceipts(ctx context.Context) {
 
 	s.listener.log.Info(fmt.Sprintf("ethreceipts: retrying %d pending receipts", len(toRetry)))
 
-	// Process retries concurrently with bounded parallelism
+	// Collect receipts that are due for retry
 	sem := make(chan struct{}, maxConcurrentReceiptRetries)
 	var wg sync.WaitGroup
 
@@ -359,46 +359,44 @@ func (s *subscriber) retryPendingReceipts(ctx context.Context) {
 			s.retryMu.Lock()
 			defer s.retryMu.Unlock()
 
-			if err != nil {
+			currentPending, exists := s.pendingReceipts[txHash]
+			if !exists {
+				s.listener.log.Warn("Pending receipt no longer exists in map, skipping", "txHash", txHash.String())
+				return
+			}
 
+			if err != nil {
 				if errors.Is(err, ethereum.NotFound) {
 					// Transaction genuinely doesn't exist - remove from queue
 					delete(s.pendingReceipts, txHash)
-					s.listener.log.Debug("Receipt not found after retry, removing from queue",
-						"txHash", txHash.String())
+					s.listener.log.Debug("Receipt not found after retry, removing from queue", "txHash", txHash.String())
 					return
 				}
 
-				// Provider error - update retry state
-				p.attempts++
-
-				// Check if max attempts reached
-				if p.attempts >= maxReceiptRetryAttempts {
-					// This will definitely remove the pending receipt, if it arrives
-					// later subscribers won't get notified.
+				// Provider error - update retry state from the current state in the map
+				currentPending.attempts++
+				if currentPending.attempts >= maxReceiptRetryAttempts {
 					delete(s.pendingReceipts, txHash)
 					s.listener.log.Error("Failed to fetch receipt after max retries",
 						"txHash", txHash.String(),
-						"attempts", p.attempts,
+						"attempts", currentPending.attempts,
 						"error", err)
-
 					// TODO: perhaps we should close the subscription here as we failed
 					// to deliver a receipt after many attempts?
 					return
 				}
 
 				// Exponential backoff for next retry
-				backoff := time.Duration(1<<uint(p.attempts)) * time.Second
+				backoff := time.Duration(1<<uint(currentPending.attempts)) * time.Second
 				if backoff > maxWaitBetweenRetries {
 					backoff = maxWaitBetweenRetries
 				}
-
-				p.nextRetryAt = time.Now().Add(backoff)
-				s.pendingReceipts[txHash] = p
+				currentPending.nextRetryAt = time.Now().Add(backoff)
+				s.pendingReceipts[txHash] = currentPending
 
 				s.listener.log.Debug("Receipt fetch failed, will retry",
 					"txHash", txHash.String(),
-					"attempt", p.attempts,
+					"attempt", currentPending.attempts,
 					"nextRetryIn", backoff,
 				)
 				return
@@ -433,7 +431,7 @@ func (s *subscriber) retryPendingReceipts(ctx context.Context) {
 
 			s.listener.log.Info("Successfully fetched receipt after retry",
 				"txHash", txHash.String(),
-				"attempts", p.attempts)
+				"attempts", currentPending.attempts)
 		}(pending)
 	}