only emit events on a change of state

src/agent/onefuzz-supervisor/src/agent.rs

@@ -21,6 +21,7 @@ pub struct Agent {
     work_queue: Box<dyn IWorkQueue>,
     worker_runner: Box<dyn IWorkerRunner>,
     _heartbeat: Option<AgentHeartbeatClient>,
+    previous_state: NodeState,
 }
 
 impl Agent {
@@ -34,6 +35,7 @@ impl Agent {
         heartbeat: Option<AgentHeartbeatClient>,
     ) -> Self {
         let scheduler = Some(scheduler);
+        let previous_state = NodeState::Init;
 
         Self {
             coordinator,
@@ -43,6 +45,7 @@ impl Agent {
             work_queue,
             worker_runner,
             _heartbeat: heartbeat,
+            previous_state,
         }
     }
 
@@ -92,16 +95,33 @@ impl Agent {
             Scheduler::Done(s) => self.done(s).await?,
         };
 
+        self.previous_state = NodeState::from(&next);
         let done = matches!(next, Scheduler::Done(..));
 
         self.scheduler = Some(next);
 
         Ok(done)
     }
 
+    async fn emit_state_update_if_changed(&mut self, event: StateUpdateEvent) -> Result<()> {
+        match (&event, self.previous_state) {
+            (StateUpdateEvent::Free, NodeState::Free)
+            | (StateUpdateEvent::Busy, NodeState::Busy)
+            | (StateUpdateEvent::SettingUp{..}, NodeState::SettingUp)
+            | (StateUpdateEvent::Rebooting, NodeState::Rebooting)
+            | (StateUpdateEvent::Ready, NodeState::Ready)
+            | (StateUpdateEvent::Done{..}, NodeState::Done) => {}
+            _ => {
+                self.coordinator.emit_event(event.into()).await?;
+            }
+        }
+
+        Ok(())
+    }
+
     async fn free(&mut self, state: State<Free>) -> Result<Scheduler> {
-        let event = StateUpdateEvent::Free.into();
-        self.coordinator.emit_event(event).await?;
+        self.emit_state_update_if_changed(StateUpdateEvent::Free)
+            .await?;
 
         let msg = self.work_queue.poll().await?;
 
@@ -173,8 +193,8 @@ impl Agent {
         verbose!("agent setting up");
 
         let tasks = state.work_set().task_ids();
-        let event = StateUpdateEvent::SettingUp { tasks };
-        self.coordinator.emit_event(event.into()).await?;
+        self.emit_state_update_if_changed(StateUpdateEvent::SettingUp { tasks })
+            .await?;
 
         let scheduler = match state.finish(self.setup_runner.as_mut()).await? {
             SetupDone::Ready(s) => s.into(),
@@ -187,9 +207,8 @@ impl Agent {
 
     async fn pending_reboot(&mut self, state: State<PendingReboot>) -> Result<Scheduler> {
         verbose!("agent pending reboot");
-
-        let event = StateUpdateEvent::Rebooting.into();
-        self.coordinator.emit_event(event).await?;
+        self.emit_state_update_if_changed(StateUpdateEvent::Rebooting)
+            .await?;
 
         let ctx = state.reboot_context();
         self.reboot.save_context(ctx).await?;
@@ -200,17 +219,14 @@ impl Agent {
 
     async fn ready(&mut self, state: State<Ready>) -> Result<Scheduler> {
         verbose!("agent ready");
-
-        let event = StateUpdateEvent::Ready.into();
-        self.coordinator.emit_event(event).await?;
-
+        self.emit_state_update_if_changed(StateUpdateEvent::Ready)
+            .await?;
         Ok(state.run().await?.into())
     }
 
     async fn busy(&mut self, state: State<Busy>) -> Result<Scheduler> {
-        let event = StateUpdateEvent::Busy.into();
-        self.coordinator.emit_event(event).await?;
-
+        self.emit_state_update_if_changed(StateUpdateEvent::Busy)
+            .await?;
         let mut events = vec![];
         let updated = state
             .update(&mut events, self.worker_runner.as_mut())
@@ -241,9 +257,7 @@ impl Agent {
             },
         };
 
-        let event = event.into();
-        self.coordinator.emit_event(event).await?;
-
+        self.emit_state_update_if_changed(event).await?;
         // `Done` is a final state.
         Ok(state.into())
     }

src/agent/onefuzz-supervisor/src/scheduler.rs

@@ -6,7 +6,7 @@ use std::fmt;
 use anyhow::Result;
 use onefuzz::process::Output;
 
-use crate::coordinator::NodeCommand;
+use crate::coordinator::{NodeCommand, NodeState};
 use crate::reboot::RebootContext;
 use crate::setup::ISetupRunner;
 use crate::work::*;
@@ -21,6 +21,19 @@ pub enum Scheduler {
     Done(State<Done>),
 }
 
+impl From<&Scheduler> for NodeState {
+    fn from(value: &Scheduler) -> Self {
+        match value {
+            Scheduler::Free(_) => Self::Free,
+            Scheduler::SettingUp(_) => Self::SettingUp,
+            Scheduler::PendingReboot(_) => Self::Rebooting,
+            Scheduler::Ready(_) => Self::Ready,
+            Scheduler::Busy(_) => Self::Busy,
+            Scheduler::Done(_) => Self::Done,
+        }
+    }
+}
+
 impl fmt::Display for Scheduler {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         let s = match self {