Remove reference to the iostack from info fetcher (#1287)

remove reference to the iostack from info fetcher

fix fmt

Merge branch 'main' into vianney/fix-info-fetcher-holding-ref-to-runtime

Add comment for drain

Co-authored-by: vianney.ruhlmann <vianney.ruhlmann@datadoghq.com>

Author: VianneyRuhlmann

data-pipeline/src/agent_info/fetcher.rs

@@ -166,6 +166,14 @@ impl AgentInfoFetcher {
 
         (fetcher, response_observer)
     }
+
+    /// Drain message from the trigger channel.
+    pub fn drain(&mut self) {
+        // We read only once as the channel has a capacity of 1
+        if let Some(rx) = &mut self.trigger_rx {
+            let _ = rx.try_recv();
+        }
+    }
 }
 
 impl Worker for AgentInfoFetcher {
@@ -272,6 +280,11 @@ impl ResponseObserver {
             }
         }
     }
+
+    /// Manually send a message to the trigger channel.
+    pub fn manual_trigger(&self) {
+        let _ = self.trigger_tx.try_send(());
+    }
 }
 
 #[cfg(test)]

data-pipeline/src/trace_exporter/mod.rs

@@ -314,7 +314,15 @@ impl TraceExporter {
                 };
             });
         }
-        // Drop runtime to shutdown all threads
+        // When the info fetcher is paused, the trigger channel keeps a reference to the runtime's
+        // IoStack as a waker. This prevents the IoStack from being dropped when shutting
+        // down runtime. By manually sending a message to the trigger channel we trigger the
+        // waker releasing the reference to the IoStack. Finally we drain the channel to
+        // avoid triggering a fetch when the info fetcher is restarted.
+        if let PausableWorker::Paused { worker } = &mut self.workers.lock_or_panic().info {
+            self.info_response_observer.manual_trigger();
+            worker.drain();
+        }
         drop(runtime);
     }
 
@@ -462,7 +470,7 @@ impl TraceExporter {
     /// 2) It's not guaranteed to not block forever, since the /info endpoint might not be
     ///    available.
     ///
-    /// The `send`` function will check agent_info when running, which will only be available if the
+    /// The `send` function will check agent_info when running, which will only be available if the
     /// fetcher had time to reach to the agent.
     /// Since agent_info can enable CSS computation, waiting for this during testing can make
     /// snapshots non-deterministic.