Fixes client downlink runtime attachment bug

Author: SirCipher

swimos_client/src/pending.rs

@@ -79,14 +79,6 @@ impl<'f> PendingConnections<'f> {
         )
     }
 
-    fn map_waiters(
-        col: Option<FnvHashMap<Key, Vec<PendingDownlink>>>,
-    ) -> impl Iterator<Item = (Key, PendingDownlink)> {
-        col.unwrap_or_default()
-            .into_iter()
-            .flat_map(|(key, downlinks)| downlinks.into_iter().map(move |dl| (key.clone(), dl)))
-    }
-
     pub fn feed_task(&self, task: BoxFuture<'f, Either<PendingDns, PendingHandshake>>) {
         self.tasks.push(task)
     }
@@ -110,14 +102,18 @@ impl<'f> PendingConnections<'f> {
         &mut self,
         host: Text,
     ) -> impl Iterator<Item = (Key, PendingDownlink)> {
-        Self::map_waiters(self.waiters.remove(&WaiterKey::Connection(host)))
+        self.waiters
+            .remove(&WaiterKey::Connection(host))
+            .unwrap_or_default()
+            .into_iter()
+            .flat_map(|(key, downlinks)| downlinks.into_iter().map(move |dl| (key.clone(), dl)))
     }
 
-    pub fn drain_runtime_queue(
-        &mut self,
-        addr: SocketAddr,
-    ) -> impl Iterator<Item = (Key, PendingDownlink)> {
-        Self::map_waiters(self.waiters.remove(&WaiterKey::Runtime(addr)))
+    pub fn drain_runtime_queue(&mut self, addr: SocketAddr, key: &Key) -> Vec<PendingDownlink> {
+        match self.waiters.get_mut(&WaiterKey::Runtime(addr)) {
+            Some(waiters) => waiters.remove(key).unwrap_or_default(),
+            None => Vec::new(),
+        }
     }
 
     pub fn waiting_on(&self, addr: SocketAddr, key: &Key) -> bool {

swimos_client/src/runtime.rs

@@ -516,7 +516,7 @@ async fn runtime_task<Net, Ws, Provider>(
                             .boxed(),
                     );
 
-                    for (_key, pending_downlink) in pending.drain_runtime_queue(sock) {
+                    for pending_downlink in pending.drain_runtime_queue(sock, &peer_key) {
                         attachment_tasks
                             .push(attach_downlink(attach.clone(), pending_downlink).boxed());
                     }
@@ -526,7 +526,7 @@ async fn runtime_task<Net, Ws, Provider>(
                 None => {
                     let error =
                         Err(DownlinkRuntimeError::new(DownlinkErrorKind::RemoteStopped).shared());
-                    for (_key, pending_downlink) in pending.drain_runtime_queue(sock) {
+                    for pending_downlink in pending.drain_runtime_queue(sock, &key) {
                         let PendingDownlink {
                             callback,
                             address,
@@ -543,7 +543,7 @@ async fn runtime_task<Net, Ws, Provider>(
             RuntimeEvent::DownlinkRuntimeStarted {
                 sock,
                 result: Err((cause, host)),
-                ..
+                key,
             } => {
                 error!(error = %cause, host = %host, "Failed to start a downlink runtime to host: ");
 
@@ -552,7 +552,7 @@ async fn runtime_task<Net, Ws, Provider>(
                     cause,
                 )
                 .shared());
-                for (_key, pending_downlink) in pending.drain_runtime_queue(sock) {
+                for pending_downlink in pending.drain_runtime_queue(sock, &key) {
                     let PendingDownlink {
                         callback,
                         address,