fix: cancel parallel routers

compparallel.go

@@ -95,7 +95,10 @@ func (r *composableParallel) FindProvidersAsync(ctx context.Context, cid cid.Cid
 			return r.FindProvidersAsync(ctx, cid, count), nil
 		},
 		func() bool {
-			return atomic.AddInt64(&totalCount, 1) > int64(count) && count != 0
+			if count == 0 {
+				return false
+			}
+			return atomic.AddInt64(&totalCount, 1) >= int64(count)
 		},
 	)
 
@@ -369,23 +372,24 @@ func getChannelOrErrorParallel[T any](
 							return
 						}
 
-						if shouldStop() {
-							log.Debug("getChannelOrErrorParallel: stopping channel iteration for router ", r.Router,
+						select {
+						case <-ctx.Done():
+							log.Debug("getChannelOrErrorParallel: stopping execution on router on context done inside select for router ", r.Router,
 								" with timeout ", r.Timeout,
 								" and ignore errors ", r.IgnoreError,
-								" closed channel: ", ok,
 							)
 							return
+						case outCh <- val:
 						}
 
-						select {
-						case <-ctx.Done():
-							log.Debug("getChannelOrErrorParallel: stopping execution on router on context done inside select for router ", r.Router,
+						if shouldStop() {
+							log.Debug("getChannelOrErrorParallel: stopping channel iteration for router ", r.Router,
 								" with timeout ", r.Timeout,
 								" and ignore errors ", r.IgnoreError,
+								" closed channel: ", ok,
 							)
+							cancelAll()
 							return
-						case outCh <- val:
 						}
 					}
 				}
@@ -402,21 +406,5 @@ func getChannelOrErrorParallel[T any](
 		log.Debug("getChannelOrErrorParallel: finished executing all routers ", len(routers))
 	}()
 
-	select {
-	case err, ok := <-errCh:
-		if !ok {
-			log.Debug("getChannelOrErrorParallel: closed error channel")
-			return outCh, routing.ErrNotFound
-		}
-		log.Debug("getChannelOrErrorParallel: error on method execution: ", err)
-
-		return outCh, err
-	case <-ctx.Done():
-		err := ctx.Err()
-		log.Debug("getChannelOrErrorParallel: context done: ", err)
-		return outCh, err
-	default:
-		log.Debug("getChannelOrErrorParallel: returning channel")
-		return outCh, nil
-	}
+	return outCh, nil
 }