ethreceipts: port #188

Author: xiam

ethreceipts/ethreceipts.go

@@ -35,6 +35,10 @@ var DefaultOptions = Options{
 	Alerter:                          util.NoopAlerter(),
 }
 
+const (
+	maxFiltersPerListener = 1000
+)
+
 type Options struct {
 	// ..
 	MaxConcurrentFetchReceiptWorkers int
@@ -145,7 +149,7 @@ func NewReceiptsListener(log logger.Logger, provider ethrpc.Interface, monitor *
 		pastReceipts:      pastReceipts,
 		notFoundTxnHashes: notFoundTxnHashes,
 		subscribers:       make([]*subscriber, 0),
-		registerFiltersCh: make(chan registerFilters, 1000),
+		registerFiltersCh: make(chan registerFilters, maxFiltersPerListener),
 		filterSem:         make(chan struct{}, opts.MaxConcurrentFilterWorkers),
 	}, nil
 }
@@ -382,11 +386,8 @@ func (l *ReceiptsListener) FetchTransactionReceiptWithFilter(ctx context.Context
 func (l *ReceiptsListener) fetchTransactionReceipt(ctx context.Context, txnHash common.Hash, forceFetch bool) (*types.Receipt, error) {
 	l.fetchSem <- struct{}{}
 
-	resultCh := make(chan *types.Receipt)
-	errCh := make(chan error)
-
-	defer close(resultCh)
-	defer close(errCh)
+	resultCh := make(chan *types.Receipt, 1)
+	errCh := make(chan error, 1)
 
 	go func() {
 		defer func() {
@@ -632,6 +633,12 @@ func (l *ReceiptsListener) listener() error {
 									panic("ethreceipts: unexpected")
 								}
 
+								// Do not exhaust a TxnHash filter that has never matched.
+								// Its historical search may have failed, and it should be allowed to keep waiting.
+								if f.Cond().TxnHash != nil && f.LastMatchBlockNum() == 0 {
+									continue
+								}
+
 								if (f.Options().LimitOne && f.LastMatchBlockNum() == 0) || !f.Options().LimitOne {
 									l.log.Debugf("filter exhausted! last block matched:%d maxWait:%d filterID:%d", filterer.LastMatchBlockNum(), maxWait, filterer.FilterID())
 
@@ -711,6 +718,11 @@ func (l *ReceiptsListener) processBlocks(blocks ethmonitor.Blocks, subscribers [
 					wg.Done()
 				}()
 
+				// retry pending receipts first
+				retryCtx, cancel := context.WithTimeout(l.ctx, 5*time.Second)
+				sub.retryPendingReceipts(retryCtx)
+				cancel()
+
 				// filter matcher
 				matched, err := sub.matchFilters(l.ctx, filterers[i], receipts)
 				if err != nil {
@@ -822,6 +834,8 @@ func (l *ReceiptsListener) latestBlockNum() *big.Int {
 
 func getChainID(ctx context.Context, provider ethrpc.Interface) (*big.Int, error) {
 	var chainID *big.Int
+
+	// provide plenty of time for breaker to succeed
 	err := breaker.Do(ctx, func() error {
 		ctx, cancel := context.WithTimeout(ctx, 4*time.Second)
 		defer cancel()
@@ -832,7 +846,7 @@ func getChainID(ctx context.Context, provider ethrpc.Interface) (*big.Int, error
 		}
 		chainID = id
 		return nil
-	}, nil, 1*time.Second, 2, 3)
+	}, nil, 1*time.Second, 2, 10)
 
 	if err != nil {
 		return nil, err