fix(providerquerymanager): confirm cancel via channel

instead of reading from a closed channel to prevent blocks after cancel, wait for a confirm message
on cancel

providerquerymanager/providerquerymanager.go

@@ -51,6 +51,7 @@ type newProvideQueryMessage struct {
 }
 
 type cancelRequestMessage struct {
+	cancelConfirm     chan<- struct{}
 	incomingProviders chan peer.ID
 	k                 cid.Cid
 }
@@ -159,12 +160,15 @@ func (pqm *ProviderQueryManager) receiveProviders(sessionCtx context.Context, k
 		for len(receivedProviders) > 0 || incomingProviders != nil {
 			select {
 			case <-sessionCtx.Done():
+				cancelConfirm := make(chan struct{})
 				pqm.providerQueryMessages <- &cancelRequestMessage{
+					cancelConfirm:     cancelConfirm,
 					incomingProviders: incomingProviders,
 					k:                 k,
 				}
-				// clear out any remaining providers
-				for range incomingProviders {
+				select {
+				case <-pqm.ctx.Done():
+				case <-cancelConfirm:
 				}
 				return
 			case provider, ok := <-incomingProviders:
@@ -364,13 +368,17 @@ func (crm *cancelRequestMessage) handle(pqm *ProviderQueryManager) {
 	requestStatus, ok := pqm.inProgressRequestStatuses[crm.k]
 	if !ok {
 		log.Errorf("Attempt to cancel request for cid (%s) not in progress", crm.k.String())
-		return
+	} else {
+		_, ok = requestStatus.listeners[crm.incomingProviders]
+		if !ok {
+			log.Errorf("Attempt to cancel request for for cid (%s) this is not a listener", crm.k.String())
+		} else {
+			delete(requestStatus.listeners, crm.incomingProviders)
+		}
+		close(crm.incomingProviders)
 	}
-	listener := crm.incomingProviders
-	if !ok {
-		log.Errorf("Attempt to cancel request for for cid (%s) this is not a listener", crm.k.String())
-		return
+	select {
+	case <-pqm.ctx.Done():
+	case crm.cancelConfirm <- struct{}{}:
 	}
-	close(listener)
-	delete(requestStatus.listeners, listener)
 }