fix(iroh::downloader): remove hash from providers in two missed c…

…ases (#1584)

## Description

We weren't removing the hash of a request that was scheduled, became
ready but couldn't be executed. This caused getting the next hash for a
provider to pop a hash no longer necessary.

I added an invariant check for the provider map and it seems to be we
missed a couple removals in #1480 so this fixes removal in cancellation
as well

## Notes & open questions
one small change, seemingly unrelated, is that the `debug_assert` now
uses the debug version of the hash instead of display. This is because
throughout the file we log the kind in debug, which in turn uses the
debug impl for the hash. This difference made it a bit harder to follow
the logs because the printed hash didn't match those in the rest of the
logs. The change makes this consistent

## Change checklist

- [x] Self-review.
- [ ] Documentation updates if relevant.
- [x] Tests if relevant.

iroh/src/downloader.rs

@@ -691,12 +691,15 @@ impl<G: Getter<Connection = D::Connection>, D: Dialer> Service<G, D> {
     /// This removes the registered download intent and, depending on its state, it will either
     /// remove it from the scheduled requests, or cancel the future.
     fn handle_cancel_download(&mut self, id: Id, kind: DownloadKind) {
+        let hash = *kind.hash();
+        let mut download_removed = false;
         if let Entry::Occupied(mut occupied_entry) = self.current_requests.entry(kind.clone()) {
             // remove the intent from the associated request
             let intents = &mut occupied_entry.get_mut().intents;
             intents.remove(&id);
             // if this was the last intent associated with the request cancel it
             if intents.is_empty() {
+                download_removed = true;
                 occupied_entry.remove().cancellation.cancel();
             }
         } else if let Entry::Occupied(mut occupied_entry) = self.scheduled_requests.entry(kind) {
@@ -708,8 +711,13 @@ impl<G: Getter<Connection = D::Connection>, D: Dialer> Service<G, D> {
             if intents.is_empty() {
                 let delay_key = occupied_entry.remove().delay_key;
                 self.scheduled_request_queue.remove(&delay_key);
+                download_removed = true;
             }
         }
+
+        if download_removed && !self.is_needed(hash) {
+            self.providers.remove(hash)
+        }
     }
 
     /// Handle a [`Message::PeersHave`].
@@ -791,7 +799,7 @@ impl<G: Getter<Connection = D::Connection>, D: Dialer> Service<G, D> {
         let Some((kind, info)) = self.unschedule(hash) else {
             debug_assert!(
                 false,
-                "invalid state: expected {hash} to be scheduled, but it wasn't"
+                "invalid state: expected {hash:?} to be scheduled, but it wasn't"
             );
             return;
         };
@@ -929,6 +937,11 @@ impl<G: Getter<Connection = D::Connection>, D: Dialer> Service<G, D> {
             remaining_retries -= 1;
             self.schedule_request(kind, remaining_retries, next_peer, intents);
         } else {
+            // check if this hash is needed in some form, otherwise remove it from providers
+            let hash = *kind.hash();
+            if !self.is_needed(hash) {
+                self.providers.remove(hash)
+            }
             // request can't be retried
             for sender in intents.into_values() {
                 let _ = sender.send(Err(anyhow::anyhow!("download ran out of attempts")));

iroh/src/downloader/invariants.rs

@@ -13,6 +13,7 @@ impl<G: Getter<Connection = D::Connection>, D: Dialer> Service<G, D> {
         self.check_scheduled_requests_consistency();
         self.check_idle_peer_consistency();
         self.chech_concurrency_limits();
+        self.check_provider_map_prunning();
     }
 
     /// Checks concurrency limits are maintained.
@@ -96,4 +97,15 @@ impl<G: Getter<Connection = D::Connection>, D: Dialer> Service<G, D> {
             "inconsistent count of idle peers"
         );
     }
+
+    /// Check that every hash in the provider map is needed.
+    #[track_caller]
+    fn check_provider_map_prunning(&self) {
+        for hash in self.providers.candidates.keys() {
+            assert!(
+                self.is_needed(*hash),
+                "provider map contains {hash:?} which should have been prunned"
+            );
+        }
+    }
 }