swarm/storage/netstore: add fetcher cancellation on shutdown (#19049)

swarm/network/stream: remove netstore internal wg
swarm/network/stream: run individual tests with t.Run

(cherry picked from commit 3ee09ba03511ad9a49e37c58f0c35b9c9771dd6f)

1 files changed, 19 insertions, 1 deletions

diff --git a/swarm/storage/netstore.go b/swarm/storage/netstore.go
index a2595d9fa..202af2bf5 100644
--- a/swarm/storage/netstore.go
+++ b/swarm/storage/netstore.go
@@ -128,7 +128,25 @@ func (n *NetStore) FetchFunc(ctx context.Context, ref Address) func(context.Cont
 func (n *NetStore) Close() {
 	close(n.closeC)
 	n.store.Close()
-	// TODO: loop through fetchers to cancel them
+
+	wg := sync.WaitGroup{}
+	for _, key := range n.fetchers.Keys() {
+		if f, ok := n.fetchers.Get(key); ok {
+			if fetch, ok := f.(*fetcher); ok {
+				wg.Add(1)
+				go func(fetch *fetcher) {
+					defer wg.Done()
+					fetch.cancel()
+
+					select {
+					case <-fetch.deliveredC:
+					case <-fetch.cancelledC:
+					}
+				}(fetch)
+			}
+		}
+	}
+	wg.Wait()
 }
 
 // get attempts at retrieving the chunk from LocalStore