Evict ended telemetry streams from the client actor cache

The per-component stream task exits once it sees no receivers and reports
`StreamStatus::Ended`, but the actor only removed the component from the
retry map — the now-dead `broadcast::Sender` stayed cached in
`component_streams`. Any later subscription for that component was handed
`tx.subscribe()` on that dead sender without a new tonic stream being
started, so it never received telemetry. With the telemetry trackers now
exiting cleanly instead of leaking (which used to hold receivers forever
and mask this), every drop-pool/rebuild cycle hit it: the rebuilt pool saw
its components as permanently silent.

On `Ended`, drop the cached sender so the next subscription starts a fresh
stream. If a subscriber arrived in the window between the stream task's
no-receivers check and the actor processing `Ended`, the entry still has
receivers — restart the stream for them instead of evicting it, since the
old task is already gone either way.

Fixes #42.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>

Author: shsms

src/client/microgrid_client_actor.rs

@@ -82,6 +82,29 @@ impl<T: MicrogridApiClient> MicrogridClientActor<T> {
                         }
                         Some(StreamStatus::Ended(component_id)) => {
                             components_to_retry.remove(&component_id);
+                            match component_streams.get(&component_id) {
+                                // A subscriber arrived between the stream
+                                // task's no-receivers check and this message;
+                                // the task is gone, so restart the stream for
+                                // the new subscriber.
+                                Some(tx) if tx.receiver_count() > 0 => {
+                                    let tx = tx.clone();
+                                    start_electrical_component_telemetry_stream(
+                                        &mut self.client,
+                                        component_id,
+                                        tx,
+                                        stream_status_tx.clone(),
+                                    )
+                                    .await;
+                                }
+                                // Drop the cached sender: nothing writes to it
+                                // anymore, so handing out further subscriptions
+                                // on it would never yield data. The next
+                                // subscription starts a fresh stream instead.
+                                _ => {
+                                    component_streams.remove(&component_id);
+                                }
+                            }
                         }
                         None => {
                             tracing::error!("MicrogridClientActor: Stream status channel closed, exiting.");

src/client/microgrid_client_handle.rs

@@ -405,4 +405,32 @@ mod tests {
             ]
         );
     }
+
+    #[tokio::test(start_paused = true)]
+    async fn test_resubscribing_after_all_receivers_dropped_restarts_stream() {
+        let handle = new_client_handle();
+
+        let mut telemetry_rx = handle
+            .receive_electrical_component_telemetry_stream(2)
+            .await
+            .unwrap();
+        telemetry_rx.recv().await.unwrap();
+        drop(telemetry_rx);
+
+        // Let the stream task notice that all receivers are gone (it checks
+        // before each sample) and the actor process its Ended status, which
+        // must evict the cached sender.
+        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+
+        // A fresh subscription must get a live stream again, not a
+        // subscription on the dead cached channel.
+        let mut telemetry_rx = handle
+            .receive_electrical_component_telemetry_stream(2)
+            .await
+            .unwrap();
+        tokio::time::timeout(std::time::Duration::from_secs(10), telemetry_rx.recv())
+            .await
+            .expect("no telemetry after resubscribing; stale stream cache?")
+            .unwrap();
+    }
 }