periodically sync in-memory member state to foca state (#410)

Signed-off-by: Somtochi Onyekwere <[email protected]>

Author: somtochiama

crates/corro-agent/src/broadcast/mod.rs

@@ -147,7 +147,7 @@ pub fn runtime_loop(
         bounded(agent.config().perf.schedule_channel_len, "to_schedule");
 
     let mut runtime: DispatchRuntime<Actor> =
-        DispatchRuntime::new(to_send_tx, to_schedule_tx, notifications_tx);
+        DispatchRuntime::new(to_send_tx, to_schedule_tx, notifications_tx.clone());
 
     let (timer_tx, mut timer_rx) = mpsc::channel(10);
     let timer_spawner = TimerSpawner::new(timer_tx);
@@ -322,7 +322,7 @@ pub fn runtime_loop(
                         gauge!("corro.gossip.cluster_size").set(last_cluster_size.get() as f64);
                     }
                     Branch::DiffMembers => {
-                        diff_member_states(&agent, &foca, &mut last_states);
+                        diff_member_states(&agent, &foca, &mut last_states, &notifications_tx);
                     }
                 }
 
@@ -375,7 +375,9 @@ pub fn runtime_loop(
                 }
             }
 
-            if let Some(handle) = diff_member_states(&agent, &foca, &mut last_states) {
+            if let Some(handle) =
+                diff_member_states(&agent, &foca, &mut last_states, &notifications_tx)
+            {
                 info!("Waiting on task to update member states...");
                 if let Err(e) = handle.await {
                     error!("could not await task to update member states: {e}");
@@ -820,6 +822,7 @@ fn diff_member_states(
     agent: &Agent,
     foca: &Foca<Actor, BincodeCodec<bincode::config::Configuration>, StdRng, NoCustomBroadcast>,
     last_states: &mut HashMap<ActorId, (foca::Member<Actor>, Option<u64>)>,
+    notifications_tx: &CorroSender<foca::OwnedNotification<Actor>>,
 ) -> Option<tokio::task::JoinHandle<()>> {
     let mut foca_states = HashMap::new();
     for member in foca.iter_membership_state() {
@@ -885,10 +888,51 @@ fn diff_member_states(
         }
     });
 
+    let mut foca_notifications = vec![];
+    foca_notifications.extend(foca_states.iter().filter_map(|(id, member)| {
+        let member = *member;
+        if foca_state_is_active(&member.state()) && members.get(id).is_none() {
+            Some(foca::OwnedNotification::MemberUp(member.id().clone()))
+        } else {
+            None
+        }
+    }));
+
+    foca_notifications.extend(members.states.iter().filter_map(|(id, member_state)| {
+        match foca_states.get(id) {
+            Some(foca_state) => {
+                if !foca_state_is_active(&foca_state.state()) {
+                    Some(foca::OwnedNotification::MemberDown(foca_state.id().clone()))
+                } else {
+                    None
+                }
+            }
+            None => Some(foca::OwnedNotification::MemberDown(
+                member_state.to_actor(*id),
+            )),
+        }
+    }));
+
+    drop(members);
+
+    if !foca_notifications.is_empty() {
+        info!(
+            "Sending out {} foca notifications for members",
+            foca_notifications.len()
+        );
+        // best effort update since we'd retry this function.
+        for notification in foca_notifications {
+            if let Err(e) = notifications_tx.try_send(notification) {
+                error!("error dispatching notifications from diff_member_states: {e}");
+                counter!("corro.channel.error", "type" => "full", "name" => "dispatch.notifications")
+                    .increment(1);
+            }
+        }
+    }
+
     if to_update.is_empty() && to_delete.is_empty() {
         return None;
     }
-
     info!(
         "Scheduling cluster membership state update for {} members (delete: {})",
         to_update.len(),
@@ -968,6 +1012,10 @@ fn diff_member_states(
     }))
 }
 
+fn foca_state_is_active(state: &foca::State) -> bool {
+    matches!(*state, foca::State::Alive | foca::State::Suspect)
+}
+
 fn make_foca_config(cluster_size: NonZeroU32) -> foca::Config {
     let mut config = foca::Config::new_wan(cluster_size);
     config.remove_down_after = Duration::from_secs(2 * 24 * 3600);

crates/corro-types/src/members.rs

@@ -40,6 +40,10 @@ impl MemberState {
     pub fn is_ring0(&self) -> bool {
         self.ring == Some(0)
     }
+
+    pub fn to_actor(&self, id: ActorId) -> Actor {
+        Actor::new(id, self.addr, self.ts, self.cluster_id, self.member_id)
+    }
 }
 
 const RING_BUCKETS: [Range<u64>; 6] = [0..6, 6..15, 15..50, 50..100, 100..200, 200..300];