huskies: merge 1200 story Low-disk warning: the fleet tells the operator before the disk takes it down

server/src/gateway/tests.rs

@@ -513,6 +513,176 @@ async fn broadcaster_forwarder_resubscribes_on_lag() {
     );
 }
 
+/// Story 1200 AC4: identical disk-space warnings arriving from different
+/// sleds within the rate window must collapse into a single forwarded chat
+/// message.
+#[tokio::test]
+async fn broadcaster_forwarder_dedupes_identical_disk_warnings_from_different_sleds() {
+    use crate::chat::{ChatTransport, MessageId};
+    use crate::service::events::StoredEvent;
+    use async_trait::async_trait;
+
+    type CallLog = Arc<std::sync::Mutex<Vec<(String, String)>>>;
+
+    struct MockTransport {
+        calls: CallLog,
+    }
+
+    #[async_trait]
+    impl ChatTransport for MockTransport {
+        async fn send_message(
+            &self,
+            room_id: &str,
+            plain: &str,
+            _html: &str,
+        ) -> Result<MessageId, String> {
+            self.calls
+                .lock()
+                .unwrap()
+                .push((room_id.to_string(), plain.to_string()));
+            Ok("id".to_string())
+        }
+
+        async fn edit_message(
+            &self,
+            _room_id: &str,
+            _id: &str,
+            _plain: &str,
+            _html: &str,
+        ) -> Result<(), String> {
+            Ok(())
+        }
+
+        async fn send_typing(&self, _room_id: &str, _typing: bool) -> Result<(), String> {
+            Ok(())
+        }
+    }
+
+    let calls: CallLog = Arc::new(std::sync::Mutex::new(Vec::new()));
+    let transport = Arc::new(MockTransport {
+        calls: Arc::clone(&calls),
+    });
+
+    let (tx, rx) =
+        tokio::sync::broadcast::channel::<crate::service::gateway::GatewayStatusEvent>(16);
+    gateway::spawn_gateway_broadcaster_forwarder(
+        transport as Arc<dyn crate::chat::ChatTransport>,
+        vec!["!room:example.org".to_string()],
+        rx,
+    );
+
+    tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+
+    let warning = |host_id: &str| crate::service::gateway::GatewayStatusEvent {
+        project: host_id.to_string(),
+        event: StoredEvent::DiskSpaceWarning {
+            level: "warn".to_string(),
+            free_bytes: 45_000_000_000,
+            target_bytes: 10_000_000_000,
+            worktrees_bytes: 5_000_000_000,
+            host_id: host_id.to_string(),
+            timestamp_ms: 100,
+        },
+    };
+
+    // Two different sleds both observe the same "warn" level within the
+    // dedupe window — only the first should be forwarded.
+    tx.send(warning("sled-a")).unwrap();
+    tx.send(warning("sled-b")).unwrap();
+
+    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+
+    let messages = calls.lock().unwrap();
+    assert_eq!(
+        messages.len(),
+        1,
+        "Expected identical disk warnings from different sleds to dedupe to one message"
+    );
+}
+
+/// Non-disk-space events (e.g. stage transitions) must never be deduped, even
+/// when several arrive back-to-back — only disk-space warnings/recoveries
+/// share a dedupe key (story 1200 AC4).
+#[tokio::test]
+async fn broadcaster_forwarder_does_not_dedupe_non_disk_events() {
+    use crate::chat::{ChatTransport, MessageId};
+    use crate::service::events::StoredEvent;
+    use async_trait::async_trait;
+
+    type CallLog = Arc<std::sync::Mutex<Vec<(String, String)>>>;
+
+    struct MockTransport {
+        calls: CallLog,
+    }
+
+    #[async_trait]
+    impl ChatTransport for MockTransport {
+        async fn send_message(
+            &self,
+            room_id: &str,
+            plain: &str,
+            _html: &str,
+        ) -> Result<MessageId, String> {
+            self.calls
+                .lock()
+                .unwrap()
+                .push((room_id.to_string(), plain.to_string()));
+            Ok("id".to_string())
+        }
+
+        async fn edit_message(
+            &self,
+            _room_id: &str,
+            _id: &str,
+            _plain: &str,
+            _html: &str,
+        ) -> Result<(), String> {
+            Ok(())
+        }
+
+        async fn send_typing(&self, _room_id: &str, _typing: bool) -> Result<(), String> {
+            Ok(())
+        }
+    }
+
+    let calls: CallLog = Arc::new(std::sync::Mutex::new(Vec::new()));
+    let transport = Arc::new(MockTransport {
+        calls: Arc::clone(&calls),
+    });
+
+    let (tx, rx) =
+        tokio::sync::broadcast::channel::<crate::service::gateway::GatewayStatusEvent>(16);
+    gateway::spawn_gateway_broadcaster_forwarder(
+        transport as Arc<dyn crate::chat::ChatTransport>,
+        vec!["!room:example.org".to_string()],
+        rx,
+    );
+
+    tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+
+    let transition = |n: u64| crate::service::gateway::GatewayStatusEvent {
+        project: "p".to_string(),
+        event: StoredEvent::StageTransition {
+            story_id: format!("{n}_story"),
+            story_name: String::new(),
+            from_stage: "2_current".to_string(),
+            to_stage: "3_qa".to_string(),
+            timestamp_ms: n,
+        },
+    };
+    tx.send(transition(1)).unwrap();
+    tx.send(transition(2)).unwrap();
+
+    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+
+    let messages = calls.lock().unwrap();
+    assert_eq!(
+        messages.len(),
+        2,
+        "Non-disk events must not be deduped against each other"
+    );
+}
+
 // ── BotConfig tests ─────────────────────────────────────────────────
 
 #[test]