dspinner: RecursiveKeys(): do not hang on cancellations

Per ipfs/kubo#10593, if no one
is reading from the channel returned by RecursiveKeys() and the context is cancelled, streamIndex will hang indefinitely.

Proposed fix is to always select when attempting to write to the `out` channel. If the context is done and there is no one to read, we can abort.

Author: hsanjuan

pinning/pinner/dspinner/pin.go

@@ -707,19 +707,27 @@ func (p *pinner) streamIndex(ctx context.Context, index dsindex.Indexer, detaile
 		defer p.lock.RUnlock()
 
 		cidSet := cid.NewSet()
+		send := func(sp ipfspinner.StreamedPin) (ok bool) {
+			select {
+			case <-ctx.Done():
+				return false
+			case out <- sp:
+				return true
+			}
+		}
 
 		err := index.ForEach(ctx, "", func(key, value string) bool {
 			c, err := cid.Cast([]byte(key))
 			if err != nil {
-				out <- ipfspinner.StreamedPin{Err: err}
+				send(ipfspinner.StreamedPin{Err: err})
 				return false
 			}
 
 			var pin ipfspinner.Pinned
 			if detailed {
 				pp, err := p.loadPin(ctx, value)
 				if err != nil {
-					out <- ipfspinner.StreamedPin{Err: err}
+					send(ipfspinner.StreamedPin{Err: err})
 					return false
 				}
 
@@ -731,17 +739,16 @@ func (p *pinner) streamIndex(ctx context.Context, index dsindex.Indexer, detaile
 			}
 
 			if !cidSet.Has(c) {
-				select {
-				case <-ctx.Done():
+				if !send(ipfspinner.StreamedPin{Pin: pin}) {
 					return false
-				case out <- ipfspinner.StreamedPin{Pin: pin}:
 				}
 				cidSet.Add(c)
 			}
 			return true
 		})
 		if err != nil {
-			out <- ipfspinner.StreamedPin{Err: err}
+			send(ipfspinner.StreamedPin{Err: err})
+			return
 		}
 	}()