clean-up

Author: xiam

ethreceipts/ethreceipts.go

@@ -316,10 +316,6 @@ 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
 
 	exhausted := make(chan struct{})
@@ -356,7 +352,6 @@ 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)
@@ -370,7 +365,7 @@ func (l *ReceiptsListener) FetchTransactionReceiptWithFilter(ctx context.Context
 
 		for {
 			select {
-			case <-internalCtx.Done():
+			case <-ctx.Done():
 				return
 
 			case <-sub.Done():
@@ -429,11 +424,9 @@ 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:
@@ -443,7 +436,6 @@ 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 {
@@ -464,17 +456,19 @@ func (l *ReceiptsListener) FetchTransactionReceiptWithFilter(ctx context.Context
 // this transaction hash.
 func (l *ReceiptsListener) fetchTransactionReceipt(ctx context.Context, txnHash common.Hash, forceFetch bool) (*types.Receipt, error) {
 
+	// TODO: this might block forever if all workers are busy, should we have a
+	// timeout?
 	l.fetchSem <- struct{}{}
 
-	resultCh := make(chan *types.Receipt)
-	errCh := make(chan error)
+	// channels to receive result or error: it could be difficult to coordinate
+	// closing them manually because we're also selecting on ctx.Done(), so we
+	// use buffered channels of size 1 and let them be garbage collected after
+	// this function returns.
+	resultCh := make(chan *types.Receipt, 1)
+	errCh := make(chan error, 1)
 
 	go func() {
-
 		defer func() {
-			close(resultCh)
-			close(errCh)
-
 			<-l.fetchSem
 		}()
 
@@ -953,6 +947,7 @@ func (l *ReceiptsListener) latestBlockNum() *big.Int {
 	}
 
 	l.latestBlockNumTime.Store(time.Now().Unix())
+
 	latestBlockNum := l.fetchLatestBlockNum()
 	if latestBlockNum != nil && latestBlockNum.Cmp(big.NewInt(0)) > 0 {
 		l.latestBlockNumCache.Store(latestBlockNum.Uint64())