fix(gateway): remove deleted Characters from parties

gateway/src/main.rs

@@ -523,14 +523,24 @@ mod character_gone_relay_tripwires {
     }
 
     #[test]
-    fn the_on_delete_callback_leaves_the_realm_core_party() {
+    fn the_on_delete_callback_moves_party_cleanup_off_the_coordinator_pump() {
         let src = include_str!("stdb/subscriptions.rs");
         let body = code_of(src, "fn arm_character_gone_relay(&self) {");
         assert!(
-            body.contains("realm_op::LEAVE") && body.contains("character_anywhere("),
-            "`arm_character_gone_relay` no longer checks every Shard and leaves on realm-core — \
-             either a moved Character is kicked mid-Transfer, or a gone one is never removed. Body \
-             was:\n{body}"
+            body.contains("store.request_deleted_character_party_reconciliation()")
+                && !body.contains("party::cleanup_deleted_character("),
+            "the row-delete callback must leave the Coordinator pump before cleanup sends Durable \
+             Requests back to the deleting Shard. Body was:\n{body}"
+        );
+        let worker = code_of(
+            src,
+            "fn request_deleted_character_party_reconciliation(&self) {",
+        );
+        assert!(
+            worker.contains("std::thread::Builder::new()")
+                && worker.contains("party_reconciliation_running")
+                && worker.contains("party::reconcile_deleted_character_parties(&store)"),
+            "deleted Character cleanup must use one coalescing worker thread. Body was:\n{worker}"
         );
     }
 
@@ -539,9 +549,23 @@ mod character_gone_relay_tripwires {
         let src = include_str!("stdb/subscriptions.rs");
         let body = code_of(src, "pub fn spawn_character_gone_relay(&self) {");
         assert!(
-            body.contains("on_reconnect") && body.contains("arm_character_gone_relay();"),
+            body.contains("on_reconnect")
+                && body.contains("arm_character_gone_relay();")
+                && body.contains("self.realm_core()")
+                && body.contains("request_deleted_character_party_reconciliation();"),
             "`spawn_character_gone_relay` no longer installs the per-shard `on_reconnect` re-arm; \
-             it goes silent after the first module republish. Body was:\n{body}"
+             both World Shard and Realm-core reconnects must schedule deferred cleanup off the \
+             parked Coordinator pump. Body was:\n{body}"
+        );
+        let worker = code_of(
+            src,
+            "fn request_deleted_character_party_reconciliation(&self) {",
+        );
+        assert!(
+            worker.contains("std::thread::Builder::new()")
+                && worker.contains("Duration::from_secs(5)")
+                && worker.contains("party::reconcile_deleted_character_parties(&store)"),
+            "reconciliation must run on its worker thread. Body was:\n{worker}"
         );
     }
 }

gateway/src/stdb/connection.rs

@@ -15,6 +15,15 @@ use super::account_sessions::SessionEpochs;
 use super::bindings::*;
 use super::movement_batch::MovementBatch;
 
+static LIVE_CONN_GENERATION: AtomicU64 = AtomicU64::new(1);
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) struct CharacterPresenceSnapshot {
+    pub(crate) generation: u64,
+    pub(crate) revision: u64,
+    pub(crate) present: bool,
+}
+
 /// Shared handle to the privileged coordination connection **of one shard**.
 ///
 /// Field 0 is the shard this handle talks to — every read, every reducer call, and every
@@ -46,6 +55,9 @@ pub(crate) struct ShardSet {
     /// unique across databases (realm-core hands each shard a disjoint guid range), so one index
     /// spans the whole realm.
     world: Arc<super::world_view::WorldView>,
+    /// Coalesce row-delete and reconnect requests behind one off-pump reconciliation worker.
+    pub(crate) party_reconciliation_requested: AtomicBool,
+    pub(crate) party_reconciliation_running: AtomicBool,
 }
 
 /// One live SDK connection generation and the handles that keep its pump and subscription alive.
@@ -61,6 +73,9 @@ pub(crate) struct LiveConn {
     /// Keeps the SDK message-pump thread alive for the connection's lifetime.
     _pump: Option<std::thread::JoinHandle<()>>,
     pump_commands: tokio::sync::mpsc::UnboundedSender<PumpCommand>,
+    /// Identity of this cache plus the Character-presence revision maintained by row callbacks.
+    cache_generation: u64,
+    character_revision: Arc<AtomicU64>,
     /// Keeps this role's subscription active for the connection's lifetime.
     _sub: SubscriptionHandle,
 }
@@ -134,11 +149,73 @@ fn start_pump(
 impl LiveConn {
     /// A connection needs its transport, subscription, and pump. A Module republish can
     /// invalidate the subscription while the socket remains connected.
-    fn is_healthy(&self) -> bool {
+    pub(super) fn is_healthy(&self) -> bool {
         self.conn.is_active()
             && self._sub.is_active()
             && self._pump.as_ref().is_some_and(|pump| !pump.is_finished())
     }
+
+    /// Ask the server for a fresh, filtered subscription before sampling Character presence.
+    ///
+    /// The ordinary subscription can lag a commit made through another Gateway. `on_applied`
+    /// runs after the server's snapshot has updated this connection's cache, so the returned sample
+    /// is current through a server-acknowledged boundary. The main subscription continues to own
+    /// every Character row after this temporary query is removed.
+    pub(super) fn durable_character_presence_snapshot(
+        &self,
+        guid: u64,
+        shard_name: &str,
+    ) -> Result<CharacterPresenceSnapshot> {
+        if !self.is_healthy() {
+            return Err(anyhow!(
+                "World Shard {shard_name} has no healthy Coordinator subscription"
+            ));
+        }
+        let generation = self.cache_generation;
+        let revision = self.character_revision.clone();
+        let (tx, rx) = std::sync::mpsc::channel::<std::result::Result<_, String>>();
+        let tx_error = tx.clone();
+        let query = format!("SELECT * FROM game_character WHERE guid = {guid}");
+        let subscription = self
+            .conn
+            .subscription_builder()
+            .on_applied(move |ctx| {
+                let _ = tx.send(Ok(CharacterPresenceSnapshot {
+                    generation,
+                    revision: revision.load(Ordering::Acquire),
+                    present: ctx.db.game_character().guid().find(&guid).is_some(),
+                }));
+            })
+            .on_error(move |_ctx, error| {
+                let _ = tx_error.send(Err(error.to_string()));
+            })
+            .subscribe(query);
+        let snapshot = match rx.recv_timeout(Duration::from_secs(15)) {
+            Ok(Ok(snapshot)) => Ok(snapshot),
+            Ok(Err(error)) => Err(anyhow!(
+                "World Shard {shard_name} Character presence subscription failed: {error}"
+            )),
+            Err(error) => Err(anyhow!(
+                "World Shard {shard_name} Character presence subscription did not apply: {error}"
+            )),
+        };
+        if let Err(error) = subscription.unsubscribe() {
+            log::warn!(
+                "World Shard {shard_name} Character presence subscription could not stop: {error}"
+            );
+            if snapshot.is_ok() {
+                return Err(anyhow!(
+                    "World Shard {shard_name} Character presence subscription could not stop: \
+                     {error}"
+                ));
+            }
+        }
+        snapshot
+    }
+
+    pub(super) fn character_revision(&self) -> Arc<AtomicU64> {
+        self.character_revision.clone()
+    }
 }
 
 /// Retire callbacks before publishing a fresh resource, then run setup and release the old resource.
@@ -621,6 +698,8 @@ fn connect_subscribed(
         reducer_completion,
         _pump: Some(pump),
         pump_commands,
+        cache_generation: LIVE_CONN_GENERATION.fetch_add(1, Ordering::Relaxed),
+        character_revision: Arc::new(AtomicU64::new(0)),
         _sub: sub,
     })
 }
@@ -2117,6 +2196,23 @@ macro_rules! call_reducer {
 pub(crate) use call_reducer;
 
 impl Coordinator {
+    #[cfg(test)]
+    pub(crate) fn park_pump_for_test(&self) {
+        let commands = self.0.coord().pump_commands.clone();
+        let (tx, rx) = std::sync::mpsc::channel();
+        commands.send(PumpCommand::Park(tx)).unwrap();
+        rx.recv_timeout(Duration::from_secs(15)).unwrap().unwrap();
+    }
+
+    #[cfg(test)]
+    pub(crate) fn resume_pump_for_test(&self) {
+        self.0
+            .coord()
+            .pump_commands
+            .send(PumpCommand::Resume)
+            .unwrap();
+    }
+
     /// Live player-session count on THIS shard: player entities (`account_id != 0`) in the
     /// shard's coordinator cache. Shard truth, so horizontally-scaled gateways report the same
     /// number instead of each undercounting to its own connections. Good enough for an ops gauge.
@@ -2346,6 +2442,8 @@ impl Coordinator {
                 conns,
                 sessions: SessionEpochs::default(),
                 world,
+                party_reconciliation_requested: AtomicBool::new(false),
+                party_reconciliation_running: AtomicBool::new(false),
             }),
         );
         // Arm the shared spatial, broadcast, private, and owner dispatch — one callback set per
@@ -2572,6 +2670,26 @@ impl Coordinator {
             .collect()
     }
 
+    /// Every configured World Shard, only while each Coordinator subscription is healthy.
+    pub(crate) fn world_shards_for_absence(&self) -> Result<Vec<(String, Coordinator)>> {
+        self.1
+            .map
+            .shards()
+            .into_iter()
+            .map(|db| {
+                let inner = self.1.conns.get(&db).cloned().ok_or_else(|| {
+                    anyhow!("World Shard {db} is configured but has no Coordinator connection")
+                })?;
+                if !inner.coord().is_healthy() {
+                    return Err(anyhow!(
+                        "World Shard {db} has no healthy Coordinator subscription"
+                    ));
+                }
+                Ok((db, Coordinator(inner, self.1.clone())))
+            })
+            .collect()
+    }
+
     /// The routing table this handle resolves through (read-only; the pure resolvers in
     /// `crate::config` take it by reference).
     pub(crate) fn shard_map(&self) -> &ShardMap {

gateway/src/stdb/reads/party.rs

@@ -8,6 +8,30 @@ use super::super::bindings::*;
 use super::super::connection::Coordinator;
 
 impl Coordinator {
+    /// Every party in this Realm-core or World Shard cache.
+    pub fn party_group_ids(&self) -> Vec<u64> {
+        self.0
+            .coord()
+            .conn
+            .db
+            .game_group()
+            .iter()
+            .map(|group| group.group_id)
+            .collect()
+    }
+
+    /// Every Character with party membership in this Realm-core cache.
+    pub fn party_member_guids(&self) -> Vec<u64> {
+        self.0
+            .coord()
+            .conn
+            .db
+            .game_group_member()
+            .iter()
+            .map(|member| member.character_guid)
+            .collect()
+    }
+
     /// The party `character_guid` belongs to, read from THIS handle's database.
     ///
     /// Which database the handle points at is the whole meaning of the answer: on the **realm-core**

gateway/src/stdb/reducers.rs

@@ -2613,6 +2613,23 @@ impl Coordinator {
         ))
     }
 
+    /// Remove a deleted Character from a party and return only after the Coordinator cache holds
+    /// the committed Realm-core roster. The caller always runs off the Coordinator pump.
+    pub fn deleted_character_party_leave(&self, character_guid: u64) -> Result<PartyOutcome> {
+        let coordinator = self.0.visibility_pipe();
+        party_outcome(call_reducer!(
+            coordinator.conn.reducers,
+            "realm_group_op",
+            realm_group_op_then(
+                lyracore_shared::group::realm_op::LEAVE,
+                character_guid,
+                0,
+                0,
+                0
+            )
+        ))
+    }
+
     /// `realm_whisper` — deliver one whisper against the database THIS handle points at. The
     /// gateway calls it on the **realm-core** handle, the only database that can
     /// address both parties of a cross-shard whisper (a guid is realm-wide; an identity is not).