1 files changed, 31 insertions, 11 deletions

diff --git a/src/state.rs b/src/state.rs
index f9da5e0..81663f3 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -606,25 +606,45 @@ impl Host for AppState {
         }
     }
 
-    fn is_cluster_member(&self, peer: &str) -> impl Future<Output = bool> + Send {
+    fn suspend(&self) -> impl Future<Output = bool> + Send {
         let state = self.clone();
-        let peer = peer.to_owned();
         async move {
-            let Some(ours) = state.cluster_id().await else {
+            let mut lifecycle = state.lifecycle.write().await;
+            let NodeLifecycle::Ready { cluster_id } = &*lifecycle else {
                 return false;
             };
-            crate::peer::peer_cluster(&state, &peer).await.as_deref() == Some(ours.as_str())
+            let cluster_id = cluster_id.clone();
+            debug!(cluster_id = %cluster_id, "Node suspended (rejoining)");
+            *lifecycle = NodeLifecycle::Joining {
+                discovered_cluster_id: Some(cluster_id),
+            };
+            true
         }
     }
 
-    fn send_message(
-        &self,
-        peer: &str,
-        message: transport::ParsedMessage,
-    ) -> impl Future<Output = anyhow::Result<()>> + Send {
+    fn resume(&self) -> impl Future<Output = ()> + Send {
         let state = self.clone();
-        let peer = peer.to_owned();
-        async move { state.transport.send_message(&peer, &message).await }
+        async move {
+            let mut lifecycle = state.lifecycle.write().await;
+            if let NodeLifecycle::Joining {
+                discovered_cluster_id: Some(cluster_id),
+            } = &*lifecycle
+            {
+                let cluster_id = cluster_id.clone();
+                debug!(cluster_id = %cluster_id, "Node resumed");
+                *lifecycle = NodeLifecycle::Ready { cluster_id };
+            }
+        }
+    }
+
+    fn initial_sync_done(&self) -> bool {
+        self.initial_sync_done
+            .load(std::sync::atomic::Ordering::Acquire)
+    }
+
+    fn reset_initial_sync(&self) {
+        self.initial_sync_done
+            .store(false, std::sync::atomic::Ordering::Release);
     }
 
     fn subscribe_changes(&self) -> broadcast::Receiver<Arc<ChangeSet>> {