fix: cross-node presence synchronization

The Presence GenServer now joins a pg group and forwards deltas to other
Presence servers on remote nodes. When a delta is received from a remote
node, it's applied to the local state, enabling presence data to sync
across the cluster.

Changes:
- Presence server joins pg group "presence:{name}" on startup
- broadcast_delta() forwards deltas to remote Presence servers via pg
- handle_info() applies incoming deltas from remote nodes to local state
- Channels subscribe to presence topic and handle presence delta messages

Author: scrogson

crates/starlang/src/presence/server.rs

@@ -2,6 +2,7 @@
 
 use super::types::{PresenceDiff, PresenceMessage, PresenceMeta, PresenceRef, PresenceState};
 use crate::core::{Pid, RawTerm};
+use crate::dist::pg;
 use crate::gen_server::{
     CallResult, CastResult, ContinueArg, ContinueResult, From, GenServer, InfoResult, InitResult,
     async_trait,
@@ -92,6 +93,8 @@ pub struct PresenceServerState {
     name: String,
     /// The PubSub server name.
     pubsub: String,
+    /// The pg group for presence servers.
+    pg_group: String,
     /// Presence state per topic: topic -> (key -> state).
     state: Arc<DashMap<String, HashMap<String, PresenceState>>>,
     /// Reverse lookup: PID -> set of (topic, key) pairs.
@@ -105,6 +108,7 @@ impl Default for PresenceServerState {
         Self {
             name: String::new(),
             pubsub: String::new(),
+            pg_group: String::new(),
             state: Arc::new(DashMap::new()),
             pid_to_presence: Arc::new(DashMap::new()),
             self_pid: None,
@@ -124,9 +128,11 @@ impl GenServer for Presence {
     type Reply = PresenceReply;
 
     async fn init(config: PresenceConfig) -> InitResult<PresenceServerState> {
+        let pg_group = format!("presence:{}", config.name);
         let state = PresenceServerState {
             name: config.name,
             pubsub: config.pubsub,
+            pg_group,
             state: Arc::new(DashMap::new()),
             pid_to_presence: Arc::new(DashMap::new()),
             self_pid: None,
@@ -165,6 +171,13 @@ impl GenServer for Presence {
                     let mut topic_state = state.state.entry(topic.clone()).or_default();
                     let key_state = topic_state.entry(key.clone()).or_default();
                     key_state.metas.push(presence_meta.clone());
+                    tracing::debug!(
+                        topic = %topic,
+                        key = %key,
+                        pid = ?pid,
+                        total_keys = topic_state.len(),
+                        "Presence::track - added to state"
+                    );
                 }
 
                 // Track reverse lookup
@@ -337,6 +350,12 @@ impl GenServer for Presence {
                     .get(&topic)
                     .map(|s| s.clone())
                     .unwrap_or_default();
+                tracing::debug!(
+                    topic = %topic,
+                    presence_count = presences.len(),
+                    keys = ?presences.keys().collect::<Vec<_>>(),
+                    "Presence::list called"
+                );
                 CallResult::Reply(PresenceReply::List(presences), std::mem::take(state))
             }
 
@@ -370,16 +389,36 @@ impl GenServer for Presence {
         msg: RawTerm,
         state: &mut PresenceServerState,
     ) -> InfoResult<PresenceServerState> {
-        // Handle presence messages from PubSub
+        // Handle presence messages from other Presence servers
         if let Ok(presence_msg) = postcard::from_bytes::<PresenceMessage>(msg.as_ref()) {
             match presence_msg {
                 PresenceMessage::Delta { topic, diff } => {
+                    tracing::debug!(
+                        topic = %topic,
+                        joins = diff.joins.len(),
+                        leaves = diff.leaves.len(),
+                        "Presence received delta from remote node, applying"
+                    );
                     apply_delta(&state.state, &topic, diff);
+
+                    // Log state after applying delta
+                    if let Some(topic_state) = state.state.get(&topic) {
+                        tracing::debug!(
+                            topic = %topic,
+                            keys = ?topic_state.keys().collect::<Vec<_>>(),
+                            "Presence state after applying delta"
+                        );
+                    }
                 }
                 PresenceMessage::StateSync {
                     topic,
                     state: remote_state,
                 } => {
+                    tracing::debug!(
+                        topic = %topic,
+                        remote_keys = remote_state.len(),
+                        "Presence received state sync from remote node, merging"
+                    );
                     merge_state(&state.state, &topic, remote_state);
                 }
             }
@@ -397,9 +436,13 @@ impl GenServer for Presence {
         // Register ourselves by name
         let _ = crate::register(&state.name, self_pid);
 
+        // Join the pg group so other Presence servers can find us
+        pg::join(&state.pg_group, self_pid);
+
         tracing::debug!(
             name = %state.name,
             pubsub = %state.pubsub,
+            pg_group = %state.pg_group,
             pid = ?self_pid,
             "Presence started"
         );
@@ -408,20 +451,41 @@ impl GenServer for Presence {
     }
 }
 
-/// Broadcast a delta to other nodes via PubSub.
+/// Broadcast a delta to other nodes via PubSub and to other Presence servers via pg.
 async fn broadcast_delta(pubsub: &str, topic: &str, diff: PresenceDiff) {
     if diff.is_empty() {
         return;
     }
 
     let msg = PresenceMessage::Delta {
         topic: topic.to_string(),
-        diff,
+        diff: diff.clone(),
     };
 
-    // Broadcast to all subscribers of the presence topic
+    // Broadcast to all subscribers of the presence topic (channels)
     let presence_topic = format!("presence:{}", topic);
     let _ = PubSub::broadcast(pubsub, &presence_topic, &msg).await;
+
+    // Also send to other Presence servers in the pg group so they can merge state
+    // The pg group name is "presence:{pubsub_name}"
+    let pg_group = format!("presence:{}", pubsub);
+    let members = pg::get_members(&pg_group);
+    let self_pid = crate::current_pid();
+    let my_node = crate::core::node::node_name_atom();
+
+    if let Ok(msg_bytes) = postcard::to_allocvec(&msg) {
+        for pid in members {
+            // Skip ourselves and only send to remote nodes
+            if pid != self_pid && pid.node() != my_node {
+                tracing::trace!(
+                    target_pid = ?pid,
+                    topic = %topic,
+                    "Forwarding presence delta to remote Presence server"
+                );
+                let _ = crate::send_raw(pid, msg_bytes.clone());
+            }
+        }
+    }
 }
 
 /// Apply a delta to local state.

examples/chat/src/channel.rs

@@ -12,10 +12,13 @@ use starlang::RawTerm;
 use starlang::channel::{
     Channel, ChannelReply, HandleResult, JoinError, JoinResult, Socket, broadcast_from, push,
 };
-use starlang::presence::Presence;
+use starlang::presence::{Presence, PresenceMessage};
+use starlang::pubsub::PubSub;
 
 /// The name of the chat presence server.
 const PRESENCE_NAME: &str = "chat_presence";
+/// The name of the chat pubsub server.
+const PUBSUB_NAME: &str = "chat_pubsub";
 use std::time::Duration;
 
 /// Custom state stored in each socket's assigns.
@@ -140,6 +143,13 @@ impl Channel for RoomChannel {
         }
         tracing::debug!(topic = %topic, key = %presence_key, "Tracked presence");
 
+        // Subscribe to presence updates for this room
+        let presence_topic = format!("presence:{}", topic);
+        if let Err(e) = PubSub::subscribe_pid(PUBSUB_NAME, &presence_topic, socket.pid).await {
+            tracing::warn!(error = %e, "Failed to subscribe to presence updates");
+        }
+        tracing::debug!(topic = %presence_topic, "Subscribed to presence updates");
+
         // Send ourselves an :after_join message to trigger presence sync and history push
         if let Ok(msg) = postcard::to_allocvec(&ChannelInfo::AfterJoin) {
             let _ = starlang::send_raw(socket.pid, msg);
@@ -331,8 +341,64 @@ impl Channel for RoomChannel {
             }
         }
 
-        // Note: Presence messages are now handled by the Presence GenServer
-        // via PubSub subscriptions, so we don't need to manually handle them here.
+        // Handle presence delta messages from PubSub
+        if let Some(presence_msg) = msg.decode::<PresenceMessage>() {
+            match presence_msg {
+                PresenceMessage::Delta { topic: _, diff } => {
+                    // Process joins - notify client of new users
+                    for (key, state) in &diff.joins {
+                        for meta in &state.metas {
+                            if let Some(user_meta) = meta.decode::<UserPresenceMeta>() {
+                                // Don't notify about ourselves
+                                if meta.pid != socket.pid {
+                                    tracing::debug!(
+                                        room = %socket.assigns.room_name,
+                                        nick = %user_meta.nick,
+                                        key = %key,
+                                        "Presence join received, notifying client"
+                                    );
+                                    push(
+                                        socket,
+                                        "user_joined",
+                                        &RoomOutEvent::UserJoined {
+                                            nick: user_meta.nick,
+                                        },
+                                    );
+                                }
+                            }
+                        }
+                    }
+
+                    // Process leaves - notify client of users leaving
+                    for (key, state) in &diff.leaves {
+                        for meta in &state.metas {
+                            if let Some(user_meta) = meta.decode::<UserPresenceMeta>() {
+                                // Don't notify about ourselves
+                                if meta.pid != socket.pid {
+                                    tracing::debug!(
+                                        room = %socket.assigns.room_name,
+                                        nick = %user_meta.nick,
+                                        key = %key,
+                                        "Presence leave received, notifying client"
+                                    );
+                                    push(
+                                        socket,
+                                        "user_left",
+                                        &RoomOutEvent::UserLeft {
+                                            nick: user_meta.nick,
+                                        },
+                                    );
+                                }
+                            }
+                        }
+                    }
+                    return HandleResult::NoReply;
+                }
+                PresenceMessage::StateSync { .. } => {
+                    // Full state sync - we handle this via PushPresenceState
+                }
+            }
+        }
 
         HandleResult::NoReply
     }
@@ -363,6 +429,12 @@ impl Channel for RoomChannel {
             tracing::warn!(error = %e, "Failed to untrack presence");
         }
         tracing::debug!(topic = %topic, key = %presence_key, "Untracked presence");
+
+        // Unsubscribe from presence updates
+        let presence_topic = format!("presence:{}", topic);
+        if let Err(e) = PubSub::unsubscribe_pid(PUBSUB_NAME, &presence_topic, socket.pid).await {
+            tracing::warn!(error = %e, "Failed to unsubscribe from presence updates");
+        }
     }
 }
 

examples/chat/tests/distribution_test.rs

@@ -270,15 +270,31 @@ async fn test_single_node_chat() {
     )
     .await;
 
-    // Bob should receive the message
-    let resp = timeout(Duration::from_secs(2), read_message(&mut bob)).await;
-    assert!(resp.is_ok(), "Bob should receive alice's message");
-    let resp = decode_event(&resp.unwrap());
-    assert!(
-        matches!(resp, ServerEvent::Message { ref from, ref text, .. } if from == "alice" && text == "hello from alice"),
-        "Expected Message, got {:?}",
-        resp
-    );
+    // Bob should receive the message (may receive UserJoined/UserList first due to presence sync)
+    let mut message_received = false;
+    for _ in 0..5 {
+        let resp = timeout(Duration::from_secs(2), read_message(&mut bob)).await;
+        if resp.is_err() {
+            break;
+        }
+        let event = decode_event(&resp.unwrap());
+        match event {
+            ServerEvent::Message {
+                ref from, ref text, ..
+            } if from == "alice" && text == "hello from alice" => {
+                message_received = true;
+                break;
+            }
+            ServerEvent::UserJoined { .. } | ServerEvent::UserList { .. } => {
+                // Skip presence-related events
+                continue;
+            }
+            other => {
+                panic!("Expected Message or presence event, got {:?}", other);
+            }
+        }
+    }
+    assert!(message_received, "Bob should have received alice's message");
 
     // Server is automatically cleaned up when _server goes out of scope (start_link)
 }