improve pending receipt handling

Author: xiam

ethreceipts/subscription.go

@@ -21,6 +21,9 @@ const (
 
 	// After this many attempts, we give up retrying a receipt fetch.
 	maxReceiptRetryAttempts = 20
+
+	// Maximum number of pending receipts to track for retries.
+	maxPendingReceipts = 5000
 )
 
 var (
@@ -56,7 +59,7 @@ type subscriber struct {
 	finalizer   *finalizer
 	mu          sync.Mutex
 
-	pendingReceipts map[common.Hash]pendingReceipt
+	pendingReceipts map[common.Hash]*pendingReceipt
 	retryMu         sync.Mutex
 }
 
@@ -150,7 +153,6 @@ func (s *subscriber) matchFilters(ctx context.Context, filterers []Filterer, rec
 		filterer    Filterer
 	}
 	var toFetch []matchedReceipt
-	var mu sync.Mutex
 
 	// First pass: find all matches
 	for _, receipt := range receipts {
@@ -238,9 +240,7 @@ func (s *subscriber) matchFilters(ctx context.Context, filterers []Filterer, rec
 			}
 
 			// Broadcast to subscribers (needs mutex as multiple goroutines may send)
-			mu.Lock()
 			s.ch.Send(item.receipt)
-			mu.Unlock()
 
 			return nil
 		})
@@ -298,15 +298,24 @@ func (s *subscriber) addPendingReceipt(receipt Receipt, filterer Filterer) {
 
 	if s.pendingReceipts == nil {
 		// lazy init
-		s.pendingReceipts = make(map[common.Hash]pendingReceipt)
+		s.pendingReceipts = make(map[common.Hash]*pendingReceipt)
+	}
+
+	if len(s.pendingReceipts) >= maxPendingReceipts {
+		s.listener.log.Error(
+			"Pending receipts queue is full, dropping new receipt",
+			"txHash", txHash.String(),
+			"queueSize", len(s.pendingReceipts),
+		)
+		return
 	}
 
 	if _, exists := s.pendingReceipts[txHash]; exists {
 		// already pending, skip
 		return
 	}
 
-	s.pendingReceipts[receipt.TransactionHash()] = pendingReceipt{
+	s.pendingReceipts[txHash] = &pendingReceipt{
 		receipt:     receipt,
 		filterer:    filterer,
 		attempts:    1,
@@ -320,11 +329,15 @@ func (s *subscriber) retryPendingReceipts(ctx context.Context) {
 	s.retryMu.Lock()
 
 	// Create a snapshot of receipts that are due for retry
-	var toRetry []pendingReceipt
+	var toRetry []*pendingReceipt
 	now := time.Now()
 
 	for _, pending := range s.pendingReceipts {
 		if now.After(pending.nextRetryAt) {
+			// Claim this item for retry by pushing the nextRetryAt into the future,
+			// this prevents other concurrent retryPendingReceipts calls from picking
+			// it up.
+			pending.nextRetryAt = time.Now().Add(10 * time.Minute)
 			toRetry = append(toRetry, pending)
 		}
 	}
@@ -342,13 +355,19 @@ func (s *subscriber) retryPendingReceipts(ctx context.Context) {
 
 	for _, pending := range toRetry {
 		wg.Add(1)
-		go func(p pendingReceipt) {
+		go func(p *pendingReceipt) {
 			defer wg.Done()
 
 			select {
 			case sem <- struct{}{}:
 				defer func() { <-sem }()
 			case <-ctx.Done():
+				// If context is cancelled, release the claim so the item can be retried later.
+				s.retryMu.Lock()
+				if current, ok := s.pendingReceipts[p.receipt.TransactionHash()]; ok && current == p {
+					current.nextRetryAt = time.Now() // Reschedule immediately.
+				}
+				s.retryMu.Unlock()
 				return
 			}
 
@@ -359,9 +378,10 @@ func (s *subscriber) retryPendingReceipts(ctx context.Context) {
 			s.retryMu.Lock()
 			defer s.retryMu.Unlock()
 
+			// Check if the item still exists and is the same one we claimed.
 			currentPending, exists := s.pendingReceipts[txHash]
-			if !exists {
-				s.listener.log.Warn("Pending receipt no longer exists in map, skipping", "txHash", txHash.String())
+			if !exists || currentPending != p {
+				s.listener.log.Debug("Pending receipt is stale or already processed, skipping retry", "txHash", txHash.String())
 				return
 			}
 
@@ -373,7 +393,7 @@ func (s *subscriber) retryPendingReceipts(ctx context.Context) {
 					return
 				}
 
-				// Provider error - update retry state from the current state in the map
+				// Provider error - update retry state directly on the pointer.
 				currentPending.attempts++
 				if currentPending.attempts >= maxReceiptRetryAttempts {
 					delete(s.pendingReceipts, txHash)
@@ -394,7 +414,6 @@ func (s *subscriber) retryPendingReceipts(ctx context.Context) {
 					backoff = maxWaitBetweenRetries
 				}
 				currentPending.nextRetryAt = time.Now().Add(backoff)
-				s.pendingReceipts[txHash] = currentPending
 
 				s.listener.log.Debug(
 					"Receipt fetch failed, will retry",