fix(ci): race conditions in tests (#119)

* fix: race condition in TestFindProviders

the test was incorrectly using require.Eventually on already-read results.
also, parallelRouter doesn't deduplicate, so peers[0] can appear twice
(once with addresses from mr1, once without from mr2). fixed by searching
for the record with addresses before verifying filtration.

* fix: deadlock in manyIter.Close()

manyIter.Close() would deadlock when goroutines were blocked
on unbuffered channel sends. fixed with synchronous drain that
matches http streaming semantics from routing v1 spec:
- idempotent close (safe to call multiple times)
- drain channel to unblock senders (expected for early termination)
- proper cleanup of child iterators

also added missing defer it.Close() in tests.

* fix: deadlock in cacheFallbackIter

dispatchFindPeer goroutines blocked forever trying to send to
unbuffered channel when iterator closed early.

fixed with buffered channel + non-blocking sends that drop
results if nobody is reading (acceptable for background lookups).
also added context cancellation in Close() to stop goroutines.

* fix: add missing iterator Close() calls in tests

added defer it.Close() to all test iterators to prevent
resource leaks that could cause intermittent hangs.

* fix: infinite recursion in cacheFallbackIter.Next()

replaced recursion with loop to avoid stack overflow.
maintains original logic: after dispatching lookup for
peer without addresses, continues trying source iterator
or waiting for lookup results.

also fixed deadlock when ongoingLookups reaches 0 between
check and channel read by using periodic timeout.

* refactor: simplify cacheFallbackIter.Next()

- removed unnecessary handleRecord closure
- cache ongoingLookups value to avoid race in logging
- use defer timer.Stop() for proper cleanup

* fix: resource leak and double-close issues

- fix iterator leak in dispatchFindPeer by adding defer Close()
- remove redundant goroutine in manyIter.Close() that could
  cause double-close panic. the channel is already closed by
  the goroutine in newManyIter.

Author: lidel

server_cached_router.go

@@ -117,56 +117,70 @@ type cacheFallbackIter struct {
 	findPeersResult chan types.PeerRecord
 	router          cachedRouter
 	ctx             context.Context
+	cancel          context.CancelFunc
 	ongoingLookups  atomic.Int32
 }
 
 // NewCacheFallbackIter is a wrapper around a results iterator that will resolve peers with no addresses from cache and if no cached addresses, will look them up via FindPeers.
 // It's a bit complex because it ensures we continue iterating without blocking on the FindPeers call.
 func NewCacheFallbackIter(sourceIter iter.ResultIter[types.Record], router cachedRouter, ctx context.Context) *cacheFallbackIter {
+	// Create a cancellable context for this iterator
+	iterCtx, cancel := context.WithCancel(ctx)
+
 	iter := &cacheFallbackIter{
 		sourceIter:      sourceIter,
 		router:          router,
-		ctx:             ctx,
-		findPeersResult: make(chan types.PeerRecord),
+		ctx:             iterCtx,
+		cancel:          cancel,
+		findPeersResult: make(chan types.PeerRecord, 100), // Buffer to avoid drops in typical cases
 		ongoingLookups:  atomic.Int32{},
 	}
 
 	return iter
 }
 
 func (it *cacheFallbackIter) Next() bool {
-	// Try to get the next value from the source iterator first
-	if it.sourceIter.Next() {
-		val := it.sourceIter.Val()
-		handleRecord := func(id *peer.ID, record *types.PeerRecord) bool {
-			record.Addrs = it.router.withAddrsFromCache(addrQueryOriginProviders, *id, record.Addrs)
-			if len(record.Addrs) > 0 {
-				it.current = iter.Result[types.Record]{Val: record}
-				return true
+	for {
+		// Try to get the next value from the source iterator first
+		if it.sourceIter.Next() {
+			val := it.sourceIter.Val()
+
+			switch val.Val.GetSchema() {
+			case types.SchemaPeer:
+				if record, ok := val.Val.(*types.PeerRecord); ok {
+					record.Addrs = it.router.withAddrsFromCache(addrQueryOriginProviders, *record.ID, record.Addrs)
+					if len(record.Addrs) > 0 {
+						it.current = iter.Result[types.Record]{Val: record}
+						return true
+					}
+
+					logger.Infow("no cached addresses found in cacheFallbackIter, dispatching find peers", "peer", record.ID)
+					if it.router.cachedAddrBook.ShouldProbePeer(*record.ID) {
+						it.ongoingLookups.Add(1) // important to increment before dispatchFindPeer
+						// If a record has no addrs, we dispatch a lookup to find addresses
+						go it.dispatchFindPeer(*record)
+					}
+					// Continue to try next source item
+					continue
+				}
 			}
-			logger.Infow("no cached addresses found in cacheFallbackIter, dispatching find peers", "peer", id)
-
-			if it.router.cachedAddrBook.ShouldProbePeer(*id) {
-				it.ongoingLookups.Add(1) // important to increment before dispatchFindPeer
-				// If a record has no addrs, we dispatch a lookup to find addresses
-				go it.dispatchFindPeer(*record)
-			}
-			return it.Next() // Recursively call Next() to either read from sourceIter or wait for lookup result
+			it.current = val // pass through unknown schemas
+			return true
 		}
 
-		switch val.Val.GetSchema() {
-		case types.SchemaPeer:
-			if record, ok := val.Val.(*types.PeerRecord); ok {
-				return handleRecord(record.ID, record)
-			}
+		// No more source items. If there are still ongoing lookups, wait for them
+		ongoing := it.ongoingLookups.Load()
+		if ongoing == 0 {
+			// No more lookups, we're done
+			return false
 		}
-		it.current = val // pass through unknown schemas
-		return true
-	}
 
-	// If there are still ongoing lookups, wait for them
-	if it.ongoingLookups.Load() > 0 {
-		logger.Infow("waiting for ongoing find peers result")
+		logger.Infow("waiting for ongoing find peers result", "ongoing", ongoing)
+
+		// Use a timeout to recheck ongoingLookups periodically
+		// This prevents deadlock if ongoingLookups becomes 0 after we check
+		timer := time.NewTimer(100 * time.Millisecond)
+		defer timer.Stop() // Ensure cleanup even if we return early
 		select {
 		case result, ok := <-it.findPeersResult:
 			if !ok {
@@ -175,15 +189,14 @@ func (it *cacheFallbackIter) Next() bool {
 			if len(result.Addrs) > 0 { // Only if the lookup returned a result and it has addrs
 				it.current = iter.Result[types.Record]{Val: &result}
 				return true
-			} else {
-				return it.Next() // recursively call Next() in case there are more ongoing lookups
 			}
+			// If no addresses, continue the loop to check for more source items or results
 		case <-it.ctx.Done():
 			return false
+		case <-timer.C:
+			// Timeout expired, loop back to try source iterator again and recheck ongoingLookups
 		}
 	}
-
-	return false
 }
 
 func (it *cacheFallbackIter) Val() iter.Result[types.Record] {
@@ -194,6 +207,12 @@ func (it *cacheFallbackIter) Val() iter.Result[types.Record] {
 }
 
 func (it *cacheFallbackIter) Close() error {
+	// Cancel the context to stop any ongoing lookups
+	if it.cancel != nil {
+		it.cancel()
+	}
+
+	// Close the source iterator
 	return it.sourceIter.Close()
 }
 
@@ -207,25 +226,42 @@ func (it *cacheFallbackIter) dispatchFindPeer(record types.PeerRecord) {
 	defer cancel()
 
 	peersIt, err := it.router.FindPeers(ctx, *record.ID, 1)
+	if err == nil {
+		defer peersIt.Close() // Ensure cleanup of the iterator
+	}
 
 	// Check if the parent context is done before sending
 	if it.ctx.Err() != nil {
 		return // Exit early if the parent context is done
 	}
 
+	// Helper to send result without blocking
+	sendResult := func(r types.PeerRecord) {
+		select {
+		case it.findPeersResult <- r:
+			// Sent successfully
+		case <-it.ctx.Done():
+			// Context cancelled, exit
+		default:
+			// Channel full or nobody listening anymore, drop the result
+			// This is OK - these are best-effort background lookups
+			logger.Debugw("dropping find peers result, nobody listening", "peer", r.ID)
+		}
+	}
+
 	if err != nil {
-		it.findPeersResult <- record // pass back the record with no addrs
+		sendResult(record) // pass back the record with no addrs
 		return
 	}
 	peers, err := iter.ReadAllResults(peersIt)
 	if err != nil {
-		it.findPeersResult <- record // pass back the record with no addrs
+		sendResult(record) // pass back the record with no addrs
 		return
 	}
 	if len(peers) > 0 {
 		// If we found the peer, pass back the result
-		it.findPeersResult <- *peers[0]
+		sendResult(*peers[0])
 	} else {
-		it.findPeersResult <- record // pass back the record with no addrs
+		sendResult(record) // pass back the record with no addrs
 	}
 }

server_routers.go

@@ -195,9 +195,20 @@ func (mi *manyIter[T]) Val() iter.Result[T] {
 }
 
 func (mi *manyIter[T]) Close() error {
+	if mi.done {
+		return nil // Already closed, idempotent
+	}
 	mi.done = true
-	mi.cancel()
-	mi.wg.Wait()
+	mi.cancel() // Signal goroutines to stop
+
+	// The channel will be closed by the goroutine in newManyIter once all workers finish
+	// We just need to drain it to unblock any pending sends
+	// This is expected behavior when client terminates early (per HTTP routing spec)
+	for range mi.ch {
+		// Discard remaining values
+	}
+
+	// Now close child iterators
 	var err error
 	for _, it := range mi.its {
 		err = errors.Join(err, it.Close())

server_routers_test.go

@@ -296,8 +296,8 @@ func TestFindProviders(t *testing.T) {
 		var d router
 		d = parallelRouter{}
 		it, err := d.FindProviders(ctx, c, 10)
-
 		require.NoError(t, err)
+		defer it.Close()
 		require.False(t, it.Next())
 
 		mr1 := &mockRouter{}
@@ -338,15 +338,31 @@ func TestFindProviders(t *testing.T) {
 
 		it, err = d.FindProviders(ctx, c, 10)
 		require.NoError(t, err)
+		defer it.Close()
 
 		results, err := iter.ReadAllResults(it)
 		require.NoError(t, err)
 		require.Len(t, results, 5)
 
-		require.Eventually(t, func() bool {
-			return len(results[0].(*types.PeerRecord).Addrs) == 1
-		}, time.Second*3, time.Millisecond*100)
-		require.Equal(t, publicAddr.String(), results[0].(*types.PeerRecord).Addrs[0].String())
+		// The parallelRouter uses manyIter which merges results from multiple routers concurrently.
+		// Both mr1 and mr2 send a record for peers[0]:
+		// - mr1 sends peers[0] WITH addresses (private, loopback, public)
+		// - mr2 sends peers[0] WITHOUT addresses
+		// Due to concurrent execution, either record could arrive first in the results.
+		// The parallelRouter doesn't deduplicate, so both records are included.
+		// We need to find the record that has addresses to verify the sanitizeRouter
+		// correctly filtered out private/loopback addresses, keeping only public ones.
+		var peerWithAddrs *types.PeerRecord
+		for _, r := range results {
+			pr := r.(*types.PeerRecord)
+			if *pr.ID == peers[0] && len(pr.Addrs) > 0 {
+				peerWithAddrs = pr
+				break
+			}
+		}
+		require.NotNil(t, peerWithAddrs, "should have found peer[0] with addresses")
+		require.Len(t, peerWithAddrs.Addrs, 1)
+		require.Equal(t, publicAddr.String(), peerWithAddrs.Addrs[0].String())
 	})
 
 	t.Run("Failed to Create All Iterators", func(t *testing.T) {
@@ -389,6 +405,7 @@ func TestFindProviders(t *testing.T) {
 
 		it, err := d.FindProviders(ctx, c, 10)
 		require.NoError(t, err)
+		defer it.Close()
 
 		results, err := iter.ReadAllResults(it)
 		require.NoError(t, err)
@@ -405,8 +422,8 @@ func TestFindPeers(t *testing.T) {
 
 		d := parallelRouter{}
 		it, err := d.FindPeers(ctx, pid, 10)
-
 		require.NoError(t, err)
+		defer it.Close()
 		require.False(t, it.Next())
 
 		mr1 := &mockRouter{}
@@ -439,6 +456,7 @@ func TestFindPeers(t *testing.T) {
 
 		it, err = d.FindPeers(ctx, pid, 10)
 		require.NoError(t, err)
+		defer it.Close()
 
 		results, err := iter.ReadAllResults(it)
 		require.NoError(t, err)
@@ -484,6 +502,7 @@ func TestFindPeers(t *testing.T) {
 
 		it, err := d.FindPeers(ctx, pid, 10)
 		require.NoError(t, err)
+		defer it.Close()
 
 		results, err := iter.ReadAllResults(it)
 		require.NoError(t, err)