From 1b19e3485af4c715c521f0d1d8099791074450bc Mon Sep 17 00:00:00 2001 From: T3 Code Date: Mon, 7 Sep 2026 17:25:39 +0000 Subject: [PATCH 1/4] fix(gateway): clean up parties after Character deletion --- gateway/src/main.rs | 38 ++- gateway/src/stdb/connection.rs | 114 +++++++- gateway/src/stdb/reads/party.rs | 24 ++ gateway/src/stdb/reducers.rs | 17 ++ gateway/src/stdb/subscriptions.rs | 428 +++++++++++++++++++++++++++--- gateway/src/stdb/world_store.rs | 175 +++++++++++- gateway/src/world/party.rs | 182 ++++++++++++- gateway/src/world/party_tests.rs | 369 +++++++++++++++++++++++++- gateway/src/world/store.rs | 52 ++++ gateway/src/world/tests.rs | 100 +++++++ 10 files changed, 1441 insertions(+), 58 deletions(-) diff --git a/gateway/src/main.rs b/gateway/src/main.rs index 8725d0799..eb6d02f60 100644 --- a/gateway/src/main.rs +++ b/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}" ); } } diff --git a/gateway/src/stdb/connection.rs b/gateway/src/stdb/connection.rs index 3f40315be..c347d2a01 100644 --- a/gateway/src/stdb/connection.rs +++ b/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, + /// 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>, pump_commands: tokio::sync::mpsc::UnboundedSender, + /// Identity of this cache plus the Character-presence revision maintained by row callbacks. + cache_generation: u64, + character_revision: Arc, /// Keeps this role's subscription active for the connection's lifetime. _sub: SubscriptionHandle, } @@ -134,11 +149,67 @@ 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 { + 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::>(); + 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() { + return Err(anyhow!( + "World Shard {shard_name} Character presence subscription could not stop: {error}" + )); + } + snapshot + } + + pub(super) fn character_revision(&self) -> Arc { + self.character_revision.clone() + } } /// Retire callbacks before publishing a fresh resource, then run setup and release the old resource. @@ -621,6 +692,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 +2190,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 +2436,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 +2664,26 @@ impl Coordinator { .collect() } + /// Every configured World Shard, only while each Coordinator subscription is healthy. + pub(crate) fn world_shards_for_absence(&self) -> Result> { + 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 { diff --git a/gateway/src/stdb/reads/party.rs b/gateway/src/stdb/reads/party.rs index 566d1fd6f..17086ab45 100644 --- a/gateway/src/stdb/reads/party.rs +++ b/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 { + 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 { + 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** diff --git a/gateway/src/stdb/reducers.rs b/gateway/src/stdb/reducers.rs index 393b302dc..8a8f460c5 100644 --- a/gateway/src/stdb/reducers.rs +++ b/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 { + 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). diff --git a/gateway/src/stdb/subscriptions.rs b/gateway/src/stdb/subscriptions.rs index 3891dd1f8..db65a7ea7 100644 --- a/gateway/src/stdb/subscriptions.rs +++ b/gateway/src/stdb/subscriptions.rs @@ -19,6 +19,7 @@ use crate::world::{Outbound, SessionTx}; use anyhow::Result; use spacetimedb_sdk::Table; use std::collections::HashSet; +use std::sync::atomic::Ordering; use std::sync::{Arc, Mutex}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use wow_world_messages::vanilla::opcodes::ServerOpcodeMessage; @@ -41,6 +42,293 @@ pub struct PlayerSubscriptions { view: Option>, } +#[cfg(test)] +mod character_gone_durable_tests { + use super::*; + use crate::accept::BlockingTaskCapacity; + use crate::config::GatewayConfig; + use crate::durable_test_support::{poll_until, Standalone, POLL_TIMEOUT}; + use crate::world::party::PartyOutcome; + use std::path::{Path, PathBuf}; + use std::process::{Command, Output}; + + struct PrivateCli { + config: PathBuf, + } + + impl PrivateCli { + fn new(owner_token: &str) -> Self { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let config = std::env::temp_dir().join(format!( + "lyracore-party-cleanup-cli-{}-{nonce}.toml", + std::process::id() + )); + std::fs::write(&config, format!("spacetimedb_token = {owner_token:?}\n")).unwrap(); + Self { config } + } + + fn command(&self) -> Command { + let binary = std::env::var_os("SPACETIME_BIN").unwrap_or_else(|| "spacetime".into()); + let mut command = Command::new(binary); + command.args(["--config-path", self.config.to_str().unwrap()]); + command + } + + fn publish(&self, server: &str, database: &str) -> Output { + let workspace = Path::new(env!("CARGO_MANIFEST_DIR")).parent().unwrap(); + self.command() + .current_dir(workspace) + .args([ + "publish", + "-s", + server, + "--module-path", + workspace.join("module").to_str().unwrap(), + "--build-options=--features=debug_reducers", + "-y", + database, + ]) + .output() + .unwrap() + } + + fn call(&self, server: &str, database: &str, reducer: &str, args: &[&str]) -> Output { + let mut command = self.command(); + command.args(["call", "-s", server, database, reducer, "--"]); + command.args(args); + command.output().unwrap() + } + } + + impl Drop for PrivateCli { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.config); + } + } + + struct TopologyEnv { + shard_map: Option, + realm_core: Option, + } + + impl TopologyEnv { + fn install(shard_map: &str, realm_core: &str) -> Self { + let previous = Self { + shard_map: std::env::var_os("LYRACORE_SHARD_MAP"), + realm_core: std::env::var_os("LYRACORE_REALM_CORE"), + }; + std::env::set_var("LYRACORE_SHARD_MAP", shard_map); + std::env::set_var("LYRACORE_REALM_CORE", realm_core); + previous + } + } + + impl Drop for TopologyEnv { + fn drop(&mut self) { + match &self.shard_map { + Some(value) => std::env::set_var("LYRACORE_SHARD_MAP", value), + None => std::env::remove_var("LYRACORE_SHARD_MAP"), + } + match &self.realm_core { + Some(value) => std::env::set_var("LYRACORE_REALM_CORE", value), + None => std::env::remove_var("LYRACORE_REALM_CORE"), + } + } + } + + fn reconciliation_is_idle(coordinator: &Coordinator) -> bool { + !coordinator + .1 + .party_reconciliation_requested + .load(Ordering::Acquire) + && !coordinator + .1 + .party_reconciliation_running + .load(Ordering::Acquire) + } + + fn reconciliation_is_running(coordinator: &Coordinator) -> bool { + coordinator + .1 + .party_reconciliation_running + .load(Ordering::Acquire) + } + + #[test] + #[ignore = "requires the SpacetimeDB 2.7.1 CLI and Wasm toolchain"] + fn another_gateway_waits_for_the_transfer_then_cleans_the_deleted_character() { + const INSTANCES: &str = "party-cleanup-instances"; + const OTHER: &str = "party-cleanup-other"; + const REALM: &str = "party-cleanup-realm"; + const INSTANCES_SURVIVOR: u64 = 2; + const OTHER_SURVIVOR: u64 = 1_000_000_001; + + let mut standalone = Standalone::start("party-cleanup-world"); + standalone.publish_module(); + let cli = PrivateCli::new(&standalone.owner_token()); + for database in [INSTANCES, OTHER, REALM] { + let output = cli.publish(standalone.server(), database); + standalone.assert_output_success(&output, &format!("publish {database}")); + } + standalone.assert_call("claim_operator", &[]); + for database in [INSTANCES, OTHER, REALM] { + let output = cli.call(standalone.server(), database, "claim_operator", &[]); + standalone.assert_output_success(&output, &format!("claim {database} operator")); + } + for database in [INSTANCES, OTHER] { + let output = cli.call( + standalone.server(), + database, + "debug_delete_character", + &["1"], + ); + standalone.assert_output_success(&output, &format!("clear Character 1 on {database}")); + } + for (database, base) in [(INSTANCES, "0"), (OTHER, "1000000000")] { + let output = cli.call(standalone.server(), database, "install_guid_range", &[base]); + standalone.assert_output_success(&output, &format!("install range on {database}")); + } + for (database, name) in [(INSTANCES, "PartyTwo"), (OTHER, "PartyOther")] { + let output = cli.call( + standalone.server(), + database, + "create_character", + &["1", name, "1", "1", "0", "0", "0", "0", "0", "0"], + ); + standalone.assert_output_success(&output, &format!("create {name}")); + } + standalone.assert_call("debug_spawn_player_entity", &["1"]); + + let _topology = TopologyEnv::install(&format!("36:*={INSTANCES}, 1:*={OTHER}"), REALM); + let cfg = GatewayConfig { + logon_bind: "127.0.0.1:0".into(), + world_bind: "127.0.0.1:0".into(), + stdb_uri: standalone.server().into(), + module_name: standalone.shard_name().into(), + coordinator_token: Some(standalone.owner_token()), + gateway_id: "party-cleanup-test".into(), + blocking_task_capacity: BlockingTaskCapacity::new(1), + }; + let runtime = tokio::runtime::Runtime::new().unwrap(); + let world = runtime.block_on(Coordinator::connect(&cfg)).unwrap(); + let instances = world.shard_handle(INSTANCES).unwrap(); + let realm = world.realm_core().unwrap(); + let observer = runtime.block_on(Coordinator::connect(&cfg)).unwrap(); + let observer_instances = observer.shard_handle(INSTANCES).unwrap(); + observer.spawn_character_gone_relay(); + assert!(poll_until(POLL_TIMEOUT, || reconciliation_is_idle( + &observer + ))); + + use lyracore_shared::group::realm_op; + assert_eq!( + realm + .realm_group_op(realm_op::INVITE, 1, INSTANCES_SURVIVOR, 0, 0) + .unwrap(), + PartyOutcome::Ran + ); + assert_eq!( + realm + .realm_group_op(realm_op::ACCEPT, INSTANCES_SURVIVOR, 0, 0, 0) + .unwrap(), + PartyOutcome::Ran + ); + assert_eq!( + realm + .realm_group_op(realm_op::INVITE, 1, OTHER_SURVIVOR, 0, 0) + .unwrap(), + PartyOutcome::Ran + ); + assert_eq!( + realm + .realm_group_op(realm_op::ACCEPT, OTHER_SURVIVOR, 0, 0, 0) + .unwrap(), + PartyOutcome::Ran + ); + assert!(poll_until(POLL_TIMEOUT, || realm + .group_roster(1) + .is_some_and( + |roster| roster.members == [1, INSTANCES_SURVIVOR, OTHER_SURVIVOR] + ))); + let roster = realm.group_roster(1).unwrap(); + for (_, shard) in world.world_shards() { + shard.sync_group_mirror(&roster).unwrap(); + } + assert!(poll_until(POLL_TIMEOUT, || world + .world_shards() + .into_iter() + .all(|(_, shard)| shard.group_roster(1).is_some_and( + |roster| roster.members == [1, INSTANCES_SURVIVOR, OTHER_SURVIVOR] + )))); + + let source_revision = observer + .0 + .coord() + .character_revision() + .load(Ordering::Acquire); + observer_instances.park_pump_for_test(); + let output = cli.call( + standalone.server(), + standalone.shard_name(), + "debug_bot_transfer", + &["1", "36", "1", "-11208", "1672", "24", "0", "party-test"], + ); + standalone.assert_output_success(&output, "transfer Character"); + crate::world::transfer::run_bot_transfer(&world, 1, 36, 1, "party-test").unwrap(); + assert!(poll_until(POLL_TIMEOUT, || { + world.character_by_guid(1).unwrap().is_none() + && instances.character_by_guid(1).unwrap().is_some() + && observer + .0 + .coord() + .character_revision() + .load(Ordering::Acquire) + > source_revision + })); + assert!(poll_until(POLL_TIMEOUT, || reconciliation_is_running( + &observer + ))); + std::thread::sleep(Duration::from_millis(250)); + assert!(realm.group_roster(1).is_some()); + observer_instances.resume_pump_for_test(); + assert!(poll_until(POLL_TIMEOUT, || reconciliation_is_idle( + &observer + ))); + assert!(realm.group_roster(1).is_some()); + + let destination_revision = observer_instances + .0 + .coord() + .character_revision() + .load(Ordering::Acquire); + let output = cli.call( + standalone.server(), + INSTANCES, + "debug_delete_character", + &["1"], + ); + standalone.assert_output_success(&output, "delete transferred Character"); + assert!(poll_until(POLL_TIMEOUT, || { + observer_instances + .0 + .coord() + .character_revision() + .load(Ordering::Acquire) + > destination_revision + && reconciliation_is_idle(&observer) + && realm.group_roster(1).is_none() + })); + let survivors = realm.group_roster(INSTANCES_SURVIVOR).unwrap(); + assert_eq!(survivors.members, [INSTANCES_SURVIVOR, OTHER_SURVIVOR]); + assert!(world.world_shards().into_iter().all(|(_, shard)| shard + .group_roster(INSTANCES_SURVIVOR) + .is_some_and(|roster| roster == survivors))); + } +} + impl PlayerSubscriptions { /// A no-op guard (used by the in-memory test store; drops harmlessly). #[cfg(test)] @@ -3650,14 +3938,10 @@ impl Coordinator { } } - /// Wire up the character-gone relay: one registration per connected WORLD SHARD, armed once at - /// gateway startup (`main.rs`), re-armed on reconnect like the two intent relays above. + /// Reconcile party membership when a World Shard deletes a Character. /// - /// realm-core owns party membership and holds no character rows, so a Character a Shard deletes - /// — a despawned playerbot, a deleted character — stays a member there until something asks. - /// This watches every Shard's `game_character` deletes and, when the Character exists on no - /// Shard at all, makes it leave its realm-core party. A Transfer deletes its source copy only - /// after the destination copy exists, so a moved Character is found and left alone. + /// Fresh Character subscriptions on every Shard distinguish deletion from a Transfer whose + /// destination update has not reached this Gateway yet. pub fn spawn_character_gone_relay(&self) { for shard in self.all_shards() { shard.arm_character_gone_relay(); @@ -3669,47 +3953,109 @@ impl Coordinator { .unwrap() .push(std::sync::Arc::new(move || { hook_shard.arm_character_gone_relay(); + hook_shard.request_deleted_character_party_reconciliation(); })); } + if self.is_sharded() { + if let Ok(realm) = self.realm_core() { + let reconciliation_store = self.clone(); + realm + .0 + .on_reconnect + .lock() + .unwrap() + .push(std::sync::Arc::new(move || { + reconciliation_store.request_deleted_character_party_reconciliation(); + })); + } + } + self.request_deleted_character_party_reconciliation(); } - fn arm_character_gone_relay(&self) { - use crate::world::WorldStore as _; + fn request_deleted_character_party_reconciliation(&self) { + self.1 + .party_reconciliation_requested + .store(true, Ordering::Release); + if self + .1 + .party_reconciliation_running + .swap(true, Ordering::AcqRel) + { + return; + } let store = self.clone(); - self.0 - .coord() - .conn - .db - .game_character() - .on_delete(move |_ctx, row| { - let Some(realm) = store.realm_store() else { - return; // one database: the Shard's own sweep already removed the member - }; - if crate::world::party::character_anywhere(&store, row.guid) - .ok() - .flatten() - .is_some() - { - return; - } - match realm.realm_group_op( - lyracore_shared::group::realm_op::LEAVE, - row.guid, - 0, - 0, - 0, - ) { - Ok(crate::world::party::PartyOutcome::Ran) => log::info!( - "party: character {} exists on no Shard and left its realm-core party", - row.guid - ), - outcome => log::debug!( - "party: character {} exists on no Shard; realm-core had no party to leave \ - ({outcome:?})", - row.guid - ), + let spawned = std::thread::Builder::new() + .name("party-reconcile-deleted".into()) + .spawn(move || { + let mut retry_delay = Duration::from_millis(100); + loop { + store + .1 + .party_reconciliation_requested + .store(false, Ordering::Release); + if let Err(error) = + crate::world::party::reconcile_deleted_character_parties(&store) + { + log::warn!( + "party: deleted Character reconciliation deferred ({error:#}); retrying in \ + {retry_delay:?}" + ); + store + .1 + .party_reconciliation_requested + .store(true, Ordering::Release); + std::thread::sleep(retry_delay); + retry_delay = (retry_delay * 2).min(Duration::from_secs(5)); + continue; + } else { + retry_delay = Duration::from_millis(100); + } + + if store + .1 + .party_reconciliation_requested + .load(Ordering::Acquire) + { + continue; + } + store + .1 + .party_reconciliation_running + .store(false, Ordering::Release); + if !store + .1 + .party_reconciliation_requested + .load(Ordering::Acquire) + || store + .1 + .party_reconciliation_running + .swap(true, Ordering::AcqRel) + { + break; + } } }); + if let Err(error) = spawned { + self.1 + .party_reconciliation_running + .store(false, Ordering::Release); + log::error!("party: could not start deleted Character reconciliation: {error}"); + } + } + + fn arm_character_gone_relay(&self) { + let live = self.0.coord(); + let inserted_revision = live.character_revision(); + live.conn.db.game_character().on_insert(move |_ctx, _row| { + inserted_revision.fetch_add(1, Ordering::Release); + }); + + let deleted_revision = live.character_revision(); + let store = self.clone(); + live.conn.db.game_character().on_delete(move |_ctx, _row| { + deleted_revision.fetch_add(1, Ordering::Release); + store.request_deleted_character_party_reconciliation(); + }); } /// One shard's half of [`spawn_bot_transfer_relay`], for the same reason diff --git a/gateway/src/stdb/world_store.rs b/gateway/src/stdb/world_store.rs index 4f99edd04..8481fad81 100644 --- a/gateway/src/stdb/world_store.rs +++ b/gateway/src/stdb/world_store.rs @@ -12,17 +12,49 @@ //! serializes the `MovementInfo` before enqueueing it on the shard's shared batch. use anyhow::{anyhow, Result}; +use spacetimedb_sdk::Table; use wow_world_messages::vanilla::MovementInfo; use crate::codec; use crate::realm_core::SessionKey; use crate::world::{SessionTx, WorldSession, WorldStore, MOVE_SUBMITTED}; -use super::bindings::GwMove; -use super::connection::Coordinator; +use super::bindings::{ + game_group_member_table::GameGroupMemberTableAccess, game_group_table::GameGroupTableAccess, + GwMove, +}; +use super::connection::{CharacterPresenceSnapshot, Coordinator}; use super::views::{AccountRow, RealmRow}; use super::PlayerSubscriptions; +fn stable_character_absence( + first: &[CharacterPresenceSnapshot], + second: &[CharacterPresenceSnapshot], +) -> bool { + first.len() == second.len() + && first.iter().zip(second).all(|(before, after)| { + !before.present + && !after.present + && before.generation == after.generation + && before.revision == after.revision + }) +} + +fn durable_character_presence_snapshots( + shards: &[(String, Coordinator)], + guid: u64, +) -> Result> { + shards + .iter() + .map(|(name, shard)| { + shard + .0 + .coord() + .durable_character_presence_snapshot(guid, name) + }) + .collect() +} + impl WorldStore for Coordinator { /// Multi-shard routing, driven by the realm-core character→shard index. /// @@ -403,6 +435,24 @@ impl WorldStore for Coordinator { self.character_by_guid(guid) } + fn character_exists_on_any_world_shard(&self, guid: u64) -> Result { + let shards = self.world_shards_for_absence()?; + let first = durable_character_presence_snapshots(&shards, guid)?; + if first.iter().any(|snapshot| snapshot.present) { + return Ok(true); + } + let second = durable_character_presence_snapshots(&shards, guid)?; + if second.iter().any(|snapshot| snapshot.present) { + return Ok(true); + } + if !stable_character_absence(&first, &second) { + anyhow::bail!( + "World Shard Character presence changed while checking {guid}; cleanup is deferred" + ); + } + Ok(false) + } + fn creature_template(&self, entry: u32) -> Result> { self.creature_template(entry) } @@ -996,6 +1046,14 @@ impl WorldStore for Coordinator { .map(|rc| std::sync::Arc::new(rc) as std::sync::Arc) } + fn party_cleanup_realm(&self) -> Result>> { + if !self.is_sharded() { + return Ok(None); + } + self.realm_core() + .map(|realm| Some(std::sync::Arc::new(realm) as std::sync::Arc)) + } + /// Every connected WORLD shard (realm-core excluded by `ShardMap::shards`, as always) — the /// mirror fan-out set. Empty when unsharded, so the push costs a single-database gateway nothing. fn world_stores(&self) -> Vec> { @@ -1023,6 +1081,13 @@ impl WorldStore for Coordinator { self.realm_group_op(op, actor_guid, target_guid, arg_a, arg_b) } + fn deleted_character_party_leave( + &self, + character_guid: u64, + ) -> Result { + self.deleted_character_party_leave(character_guid) + } + fn group_roster( &self, character_guid: u64, @@ -1037,6 +1102,62 @@ impl WorldStore for Coordinator { Ok(self.group_roster_by_id(group_id)) } + fn party_cleanup_group_roster_by_id( + &self, + group_id: u64, + ) -> Result> { + let live = self.0.coord(); + if !live.is_healthy() { + anyhow::bail!( + "{} has no healthy Coordinator subscription for party cleanup", + self.shard_name() + ); + } + let db = &live.conn.db; + let Some(group) = db + .game_group() + .iter() + .find(|group| group.group_id == group_id) + else { + return Ok(None); + }; + let mut rows: Vec<(u64, u64)> = db + .game_group_member() + .iter() + .filter(|member| member.group_id == group_id) + .map(|member| (member.id, member.character_guid)) + .collect(); + rows.sort_unstable(); + Ok(Some(crate::world::party::GroupRoster { + group_id, + leader_guid: group.leader_guid, + loot_method: group.loot_method, + loot_threshold: group.loot_threshold, + master_looter_guid: group.master_looter_guid, + members: rows.into_iter().map(|(_, guid)| guid).collect(), + })) + } + + fn party_member_guids(&self) -> Result> { + if !self.0.coord().is_healthy() { + anyhow::bail!( + "{} has no healthy Coordinator subscription for party cleanup", + self.shard_name() + ); + } + Ok(self.party_member_guids()) + } + + fn party_group_ids(&self) -> Result> { + if !self.0.coord().is_healthy() { + anyhow::bail!( + "{} has no healthy Coordinator subscription for party cleanup", + self.shard_name() + ); + } + Ok(self.party_group_ids()) + } + fn sync_group_mirror(&self, roster: &crate::world::party::GroupRoster) -> Result<()> { self.sync_group_mirror(roster) } @@ -1611,4 +1732,54 @@ mod routing_call_site_tests { unconfigured gateway nothing" ); } + + #[test] + fn deleted_character_absence_uses_two_durable_snapshots_from_every_shard() { + let read = code_of("character_exists_on_any_world_shard"); + assert!( + read.contains("self.world_shards_for_absence()?") + && read + .matches("durable_character_presence_snapshots(&shards, guid)?") + .count() + == 2 + && read.contains("stable_character_absence(&first, &second)") + ); + + let src = include_str!("connection.rs"); + let body = crate::test_scan::code_of(src, "pub(crate) fn world_shards_for_absence(&self)"); + assert!( + body.contains(".shards()") + && body.contains("self.1.conns.get(&db)") + && body.contains("inner.coord().is_healthy()"), + "absence can only be established from every configured World Shard's healthy \ + Coordinator subscription. Body was:\n{body}" + ); + + let realm = code_of("party_cleanup_realm"); + assert!( + realm.contains("self.realm_core()") && !realm.contains(".ok()"), + "deleted Character cleanup must fail when configured Realm-core is unavailable. Body \ + was:\n{realm}" + ); + } + + #[test] + fn a_transfer_between_durable_scans_cannot_prove_stable_absence() { + use super::{stable_character_absence, CharacterPresenceSnapshot}; + + let snapshot = |generation, revision, present| CharacterPresenceSnapshot { + generation, + revision, + present, + }; + let first = [snapshot(1, 0, false), snapshot(2, 1, false)]; + let destination_visible = [snapshot(1, 1, true), snapshot(2, 1, false)]; + assert!(!stable_character_absence(&first, &destination_visible)); + + let moved_again = [snapshot(1, 2, false), snapshot(2, 2, false)]; + assert!(!stable_character_absence(&first, &moved_again)); + + let unchanged = [snapshot(1, 0, false), snapshot(2, 1, false)]; + assert!(stable_character_absence(&first, &unchanged)); + } } diff --git a/gateway/src/world/party.rs b/gateway/src/world/party.rs index d8e30dd31..a06b2f1a2 100644 --- a/gateway/src/world/party.rs +++ b/gateway/src/world/party.rs @@ -513,15 +513,185 @@ pub(crate) fn run_bot_leave(store: &St, leaver_guid: u64) -> Res } None => store, }; + let leave = run_server_leave(store, realm, leaver_guid, 1, |realm, character_guid| { + realm.realm_group_op(realm_op::LEAVE, character_guid, 0, 0, 0) + })?; + if leave.outcome == PartyOutcome::Ran { + sync_mirrors(store, realm, leaver_guid, leave.previous_group_id); + } + Ok(leave.outcome) +} + +struct ServerLeave { + outcome: PartyOutcome, + previous_group_id: Option, +} + +fn run_server_leave( + store: &St, + realm: &dyn WorldStore, + leaver_guid: u64, + attempts: usize, + leave_party: impl Fn(&dyn WorldStore, u64) -> Result, +) -> Result { let before = realm.group_roster(leaver_guid)?.map(|r| r.group_id); crate::world::loot::flush_pending_promotions(store, realm); - if let PartyOutcome::Refused(refusal) = - realm.realm_group_op(realm_op::LEAVE, leaver_guid, 0, 0, 0)? - { - return Ok(PartyOutcome::Refused(refusal)); + let mut last_error = None; + for _ in 0..attempts { + match leave_party(realm, leaver_guid) { + Ok(PartyOutcome::Ran) => { + return Ok(ServerLeave { + outcome: PartyOutcome::Ran, + previous_group_id: before, + }); + } + Ok(PartyOutcome::Refused(GroupRefusal::NotInGroup)) + if before.is_some() && last_error.is_some() => + { + return Ok(ServerLeave { + outcome: PartyOutcome::Ran, + previous_group_id: before, + }); + } + Ok(PartyOutcome::Refused(refusal)) => { + return Ok(ServerLeave { + outcome: PartyOutcome::Refused(refusal), + previous_group_id: before, + }); + } + Err(error) => last_error = Some(error), + } + } + + let Some(error) = last_error else { + return Ok(ServerLeave { + outcome: PartyOutcome::Refused(GroupRefusal::NotInGroup), + previous_group_id: before, + }); + }; + if before.is_some() && realm.group_roster(leaver_guid)?.is_none() { + return Ok(ServerLeave { + outcome: PartyOutcome::Ran, + previous_group_id: before, + }); + } + Err(error) +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum DeletedCharacterPartyCleanup { + Preserved, + Removed, + AlreadyClean, +} + +const DELETED_CHARACTER_LEAVE_ATTEMPTS: usize = 3; + +/// Remove a deleted Character from its realm-core party after every World Shard confirms absence. +pub(crate) fn cleanup_deleted_character( + store: &St, + character_guid: u64, +) -> Result { + let Some(realm) = store.party_cleanup_realm()? else { + return Ok(DeletedCharacterPartyCleanup::AlreadyClean); + }; + if store.character_exists_on_any_world_shard(character_guid)? { + return Ok(DeletedCharacterPartyCleanup::Preserved); + } + let leave = run_server_leave( + store, + realm.as_ref(), + character_guid, + DELETED_CHARACTER_LEAVE_ATTEMPTS, + |realm, character_guid| realm.deleted_character_party_leave(character_guid), + )?; + match leave.outcome { + PartyOutcome::Ran => { + if let Some(group_id) = leave.previous_group_id { + sync_group_mirrors_required(store, realm.as_ref(), group_id)?; + } + Ok(DeletedCharacterPartyCleanup::Removed) + } + PartyOutcome::Refused(GroupRefusal::NotInGroup) => { + Ok(DeletedCharacterPartyCleanup::AlreadyClean) + } + PartyOutcome::Refused(refusal) => { + anyhow::bail!("realm-core refused deleted Character cleanup: {refusal:?}") + } + } +} + +/// Recheck authoritative party members after startup or a Coordinator reconnect. Row-delete +/// callbacks are not replayed, so this closes cleanup attempts deferred while a Shard was down. +pub(crate) fn reconcile_deleted_character_parties(store: &St) -> Result<()> { + let Some(realm) = store.party_cleanup_realm()? else { + return Ok(()); + }; + let mut failures = 0usize; + let mut last_error = None; + for character_guid in realm.party_member_guids()? { + if let Err(error) = cleanup_deleted_character(store, character_guid) { + failures += 1; + log::warn!( + "party: could not finish reconciling Character {character_guid} ({error:#}); retrying" + ); + last_error = Some(error); + } + } + let mut group_ids: std::collections::HashSet = + realm.party_group_ids()?.into_iter().collect(); + for shard in store.world_stores() { + group_ids.extend(shard.party_group_ids()?); + } + for group_id in group_ids { + if let Err(error) = sync_group_mirrors_required(store, realm.as_ref(), group_id) { + failures += 1; + log::warn!("party: could not reconcile group {group_id} mirrors ({error:#}); retrying"); + last_error = Some(error); + } + } + match last_error { + Some(error) => Err(error.context(format!( + "{failures} deleted Character party reconciliation attempt(s) were deferred" + ))), + None => Ok(()), + } +} + +const DELETED_CHARACTER_MIRROR_ATTEMPTS: usize = 3; + +fn sync_group_mirrors_required( + store: &St, + realm: &dyn WorldStore, + group_id: u64, +) -> Result<()> { + let roster = realm + .party_cleanup_group_roster_by_id(group_id)? + .unwrap_or_else(|| GroupRoster::disbanded(group_id)); + let mut failures = 0usize; + let mut last_error = None; + for shard in store.world_stores() { + let mut synced = false; + for _ in 0..DELETED_CHARACTER_MIRROR_ATTEMPTS { + match shard.sync_group_mirror(&roster) { + Ok(()) => { + synced = true; + break; + } + Err(error) => last_error = Some(error), + } + } + if !synced { + failures += 1; + } + } + match last_error { + Some(error) if failures > 0 => Err(error.context(format!( + "{failures} World Shard party mirror update(s) failed after \ + {DELETED_CHARACTER_MIRROR_ATTEMPTS} attempts" + ))), + _ => Ok(()), } - sync_mirrors(store, realm, leaver_guid, before); - Ok(PartyOutcome::Ran) } /// Push realm-core's roster for every party this op could have touched onto every connected world diff --git a/gateway/src/world/party_tests.rs b/gateway/src/world/party_tests.rs index 163890333..51b212004 100644 --- a/gateway/src/world/party_tests.rs +++ b/gateway/src/world/party_tests.rs @@ -9,7 +9,7 @@ //! A child module of `world::tests` so it can reach `InMemoryStore` without widening anything. use super::*; -use lyracore_shared::group::GroupRefusal; +use lyracore_shared::group::{realm_op, GroupRefusal}; pub(super) const GINGER: u64 = 1; // in the open world, on `world` pub(super) const VIM: u64 = 2; // inside the dungeon, on `instances` @@ -24,6 +24,50 @@ pub(super) const BOT: u64 = 5; /// standing on a different database than the inviting player is reachable in principle; this pins it. const FAR_BOT: u64 = 6; +fn topology_after_vim_is_deleted() -> ( + std::sync::Arc, + std::sync::Arc, + std::sync::Arc, + ShardCallLog, +) { + let (realm, world, instances, calls) = party_topology(); + form_split_party(&world, &instances); + party::run(world.as_ref(), 7, GINGER, party::Op::Invite(TRIN)).expect("invite the survivor"); + party::run(world.as_ref(), 9, TRIN, party::Op::Accept).expect("the survivor accepts"); + + let deleted_from_instances = std::sync::Arc::new(InMemoryStore { + shard: "instances".into(), + calls: calls.clone(), + realm: Some(realm.clone()), + mirror: std::sync::Mutex::new(instances.mirror.lock().unwrap().clone()), + ..Default::default() + }); + *world.peers.lock().unwrap() = vec![world.clone(), deleted_from_instances.clone()]; + *deleted_from_instances.peers.lock().unwrap() = + vec![world.clone(), deleted_from_instances.clone()]; + (realm, world, deleted_from_instances, calls) +} + +fn two_member_topology_after_vim_is_deleted() -> ( + std::sync::Arc, + std::sync::Arc, + std::sync::Arc, +) { + let (realm, world, instances, calls) = party_topology(); + form_split_party(&world, &instances); + let deleted_from_instances = std::sync::Arc::new(InMemoryStore { + shard: "instances".into(), + calls, + realm: Some(realm.clone()), + mirror: std::sync::Mutex::new(instances.mirror.lock().unwrap().clone()), + ..Default::default() + }); + *world.peers.lock().unwrap() = vec![world.clone(), deleted_from_instances.clone()]; + *deleted_from_instances.peers.lock().unwrap() = + vec![world.clone(), deleted_from_instances.clone()]; + (realm, world, deleted_from_instances) +} + pub(super) fn character(guid: u64, name: &str) -> codec::CharacterView { codec::CharacterView { guid, @@ -293,6 +337,329 @@ fn leaving_a_party_re_pushes_the_roster_of_the_group_the_leaver_left() { } } +#[test] +fn an_uncertain_character_read_preserves_realm_core_membership() { + let (realm, world, instances, _calls) = party_topology(); + form_split_party(&world, &instances); + let uncertain_instances = std::sync::Arc::new(InMemoryStore { + shard: "instances".into(), + realm: Some(realm.clone()), + character_read_error: Some("subscription unavailable".into()), + ..Default::default() + }); + *world.peers.lock().unwrap() = vec![world.clone(), uncertain_instances]; + + let error = party::cleanup_deleted_character(world.as_ref(), VIM) + .expect_err("unknown absence must stop cleanup"); + + assert!(error.to_string().contains("subscription unavailable")); + assert!( + realm.group_roster(VIM).unwrap().is_some(), + "an unreadable Shard cannot prove deletion, so Realm-core membership must stay unchanged" + ); +} + +#[test] +fn a_deleted_member_leaves_realm_core_and_both_shards_receive_the_surviving_roster() { + let (realm, world, instances, _calls) = topology_after_vim_is_deleted(); + + assert_eq!( + party::cleanup_deleted_character(world.as_ref(), VIM).unwrap(), + party::DeletedCharacterPartyCleanup::Removed + ); + + let survivors = realm.group_roster(GINGER).unwrap().unwrap(); + assert_eq!(survivors.members, vec![GINGER, TRIN]); + for shard in [&world, &instances] { + assert_eq!( + shard.mirror.lock().unwrap().as_slice(), + std::slice::from_ref(&survivors) + ); + } +} + +#[test] +fn a_character_visible_on_the_destination_shard_keeps_party_membership() { + let (realm, world, instances, calls) = party_topology(); + form_split_party(&world, &instances); + calls.lock().unwrap().clear(); + + assert_eq!( + party::cleanup_deleted_character(world.as_ref(), VIM).unwrap(), + party::DeletedCharacterPartyCleanup::Preserved + ); + assert!(realm.group_roster(VIM).unwrap().is_some()); + assert!(calls.lock().unwrap().is_empty()); +} + +#[test] +fn an_unavailable_configured_shard_preserves_realm_core_membership() { + let (realm, world, instances, calls) = party_topology(); + form_split_party(&world, &instances); + let incomplete = InMemoryStore { + shard: "world".into(), + calls, + realm: Some(realm.clone()), + world_shard_set_error: Some("instances has no healthy Coordinator subscription".into()), + ..Default::default() + }; + + let error = party::cleanup_deleted_character(&incomplete, VIM) + .expect_err("an incomplete Shard set cannot establish absence"); + + assert!(error.to_string().contains("no healthy Coordinator")); + assert!(realm.group_roster(VIM).unwrap().is_some()); +} + +#[test] +fn an_unavailable_realm_core_preserves_party_membership() { + let (realm, world, instances, calls) = party_topology(); + form_split_party(&world, &instances); + let unavailable = InMemoryStore { + shard: "world".into(), + calls, + realm: Some(realm.clone()), + party_cleanup_realm_error: Some("Realm-core is not connected".into()), + ..Default::default() + }; + + let error = party::cleanup_deleted_character(&unavailable, VIM) + .expect_err("unavailable Realm-core cannot become an unsharded cleanup"); + + assert!(error.to_string().contains("not connected")); + assert!(realm.group_roster(VIM).unwrap().is_some()); +} + +#[test] +fn repeated_deleted_character_cleanup_is_harmless() { + let (realm, world, _instances, _calls) = topology_after_vim_is_deleted(); + assert_eq!( + party::cleanup_deleted_character(world.as_ref(), VIM).unwrap(), + party::DeletedCharacterPartyCleanup::Removed + ); + + assert_eq!( + party::cleanup_deleted_character(world.as_ref(), VIM).unwrap(), + party::DeletedCharacterPartyCleanup::AlreadyClean + ); + assert_eq!( + realm.group_roster(GINGER).unwrap().unwrap().members, + vec![GINGER, TRIN] + ); +} + +#[test] +fn deleted_character_cleanup_retries_a_transient_realm_core_leave_failure() { + let (realm, world, _instances, _calls) = topology_after_vim_is_deleted(); + realm + .party_leave_failures + .store(2, std::sync::atomic::Ordering::SeqCst); + + assert_eq!( + party::cleanup_deleted_character(world.as_ref(), VIM).unwrap(), + party::DeletedCharacterPartyCleanup::Removed + ); + assert_eq!(realm.group_roster(VIM).unwrap(), None); + assert_eq!( + realm + .party + .lock() + .unwrap() + .ops + .iter() + .filter(|(op, actor, ..)| *op == realm_op::LEAVE && *actor == VIM) + .count(), + 3 + ); +} + +#[test] +fn a_lost_realm_core_leave_reply_still_refreshes_the_surviving_party() { + let (realm, world, instances, _calls) = topology_after_vim_is_deleted(); + realm + .party_leave_commit_then_error + .store(true, std::sync::atomic::Ordering::SeqCst); + + assert_eq!( + party::cleanup_deleted_character(world.as_ref(), VIM).unwrap(), + party::DeletedCharacterPartyCleanup::Removed + ); + + let survivors = realm.group_roster(GINGER).unwrap().unwrap(); + assert_eq!(survivors.members, vec![GINGER, TRIN]); + for shard in [&world, &instances] { + assert_eq!( + shard.mirror.lock().unwrap().as_slice(), + std::slice::from_ref(&survivors) + ); + } +} + +#[test] +fn deleted_character_cleanup_retries_a_transient_mirror_failure() { + let (realm, world, instances, calls) = topology_after_vim_is_deleted(); + instances + .mirror_failures + .store(2, std::sync::atomic::Ordering::SeqCst); + calls.lock().unwrap().clear(); + + assert_eq!( + party::cleanup_deleted_character(world.as_ref(), VIM).unwrap(), + party::DeletedCharacterPartyCleanup::Removed + ); + + let survivors = realm.group_roster(GINGER).unwrap().unwrap(); + assert_eq!( + instances.mirror.lock().unwrap().as_slice(), + std::slice::from_ref(&survivors) + ); + assert_eq!( + calls + .lock() + .unwrap() + .iter() + .filter(|(shard, call)| shard == "instances" && call == "sync_group_mirror") + .count(), + 3 + ); +} + +#[test] +fn reconciliation_repairs_a_mirror_left_stale_after_membership_cleanup() { + let (realm, world, instances) = two_member_topology_after_vim_is_deleted(); + instances + .mirror_failures + .store(7, std::sync::atomic::Ordering::SeqCst); + + let error = party::cleanup_deleted_character(world.as_ref(), VIM) + .expect_err("three failed mirror writes must keep cleanup pending"); + assert!(error.to_string().contains("3 attempts")); + assert_eq!(realm.group_roster(VIM).unwrap(), None); + assert_eq!(realm.party_group_ids().unwrap(), Vec::::new()); + assert!(instances + .mirror + .lock() + .unwrap() + .iter() + .any(|roster| roster.members.contains(&VIM))); + + party::reconcile_deleted_character_parties(world.as_ref()) + .expect_err("the first reconciliation exhausts its bounded mirror retries"); + assert!(!instances.mirror.lock().unwrap().is_empty()); + + party::reconcile_deleted_character_parties(world.as_ref()).unwrap(); + + assert!(instances.mirror.lock().unwrap().is_empty()); +} + +#[test] +fn deleted_character_cleanup_stops_after_three_realm_core_leave_failures() { + let (realm, world, _instances, _calls) = topology_after_vim_is_deleted(); + realm + .party_leave_failures + .store(4, std::sync::atomic::Ordering::SeqCst); + + let error = party::cleanup_deleted_character(world.as_ref(), VIM) + .expect_err("cleanup must report a persistent Realm-core failure"); + + assert!(error.to_string().contains("connection interrupted")); + assert!(realm.group_roster(VIM).unwrap().is_some()); + assert_eq!( + realm + .party + .lock() + .unwrap() + .ops + .iter() + .filter(|(op, actor, ..)| *op == realm_op::LEAVE && *actor == VIM) + .count(), + 3 + ); +} + +#[test] +fn unsharded_deleted_character_cleanup_stays_on_the_module_sweep() { + let store = InMemoryStore { + shard: "world".into(), + ..Default::default() + }; + + assert_eq!( + party::cleanup_deleted_character(&store, VIM).unwrap(), + party::DeletedCharacterPartyCleanup::AlreadyClean + ); + assert!(store.calls.lock().unwrap().is_empty()); +} + +#[test] +fn reconnect_reconciliation_removes_a_member_whose_delete_event_was_missed() { + let (realm, world, instances, _calls) = topology_after_vim_is_deleted(); + + party::reconcile_deleted_character_parties(world.as_ref()).unwrap(); + + let survivors = realm.group_roster(GINGER).unwrap().unwrap(); + assert_eq!(survivors.members, vec![GINGER, TRIN]); + assert_eq!( + instances.mirror.lock().unwrap().as_slice(), + std::slice::from_ref(&survivors) + ); +} + +#[test] +fn deleted_character_cleanup_flushes_pending_loot_before_leaving_realm_core() { + let (_realm, world, instances, calls) = topology_after_vim_is_deleted(); + *instances.pending_rolls.lock().unwrap() = vec![loot::PendingLootRoll { + roll_id: 40, + corpse_guid: 90, + slot: 1, + item_entry: 100, + deadline_micros: 500, + recipients: vec![GINGER, VIM, TRIN], + }]; + calls.lock().unwrap().clear(); + + party::cleanup_deleted_character(world.as_ref(), VIM).unwrap(); + + let names: Vec = calls + .lock() + .unwrap() + .iter() + .map(|(_, call)| call.clone()) + .collect(); + let promotion = names + .iter() + .position(|name| name == "realm_loot_op") + .unwrap(); + let leave = names + .iter() + .position(|name| name == "realm_group_op") + .unwrap(); + assert!( + promotion < leave, + "pending loot from the deleted Character's former Shard must reach Realm-core before a \ + disband-capable leave" + ); +} + +#[test] +fn deleted_character_leave_returns_with_the_committed_roster_visible() { + let src = include_str!("../stdb/reducers.rs"); + let body = crate::test_scan::code_of(src, "pub fn deleted_character_party_leave("); + assert!( + body.contains("let coordinator = self.0.visibility_pipe()") + && body.contains("coordinator.conn.reducers"), + "cleanup reads Realm-core immediately after LEAVE, so the Durable Request must return a \ + Coordinator visibility receipt. Body was:\n{body}" + ); + + let ordinary = crate::test_scan::code_of(src, "pub fn realm_group_op("); + assert!( + ordinary.contains("self.0.call_pipe().conn.reducers"), + "bot callbacks can invoke the ordinary party operation on their Coordinator pump, so it \ + must retain the independent call pipe. Body was:\n{ordinary}" + ); +} + /// **The invariant this batch has broken five times: unset config changes NOTHING.** /// /// A single-database gateway has no realm-core to route to, so every op takes the pre-realm-core diff --git a/gateway/src/world/store.rs b/gateway/src/world/store.rs index 04f648c86..4853c6731 100644 --- a/gateway/src/world/store.rs +++ b/gateway/src/world/store.rs @@ -181,6 +181,12 @@ pub trait WorldStore: None } + /// Realm-core for deleted Character cleanup. An unavailable configured Realm-core is an + /// infrastructure failure and cannot fall back to a World Shard. + fn party_cleanup_realm(&self) -> Result>> { + Ok(self.realm_store()) + } + /// Every connected WORLD shard's handle (realm-core excluded — it owns no gameplay reads). The /// fan-out set for the roster mirror; empty on a single-database gateway, which is what makes the /// mirror push a no-op there. @@ -205,6 +211,18 @@ pub trait WorldStore: Err(anyhow!("this store does not host realm-wide party state")) } + /// Realm-core LEAVE for a deleted Character. Production returns after its Coordinator cache + /// has the committed roster; Fakes may reuse the ordinary party operation. + fn deleted_character_party_leave(&self, character_guid: u64) -> Result { + self.realm_group_op( + lyracore_shared::group::realm_op::LEAVE, + character_guid, + 0, + 0, + 0, + ) + } + /// The party `character_guid` is in, as THIS handle's database sees it: authoritative on /// realm-core, a mirror on a world shard. `None` = not in a party there. fn group_roster(&self, _character_guid: u64) -> Result> { @@ -217,6 +235,26 @@ pub trait WorldStore: Ok(None) } + /// Authoritative roster read used to repair mirrors after deleted Character cleanup. A + /// configured but unhealthy Realm-core must return `Err`, never a false disband. + fn party_cleanup_group_roster_by_id( + &self, + group_id: u64, + ) -> Result> { + self.group_roster_by_id(group_id) + } + + /// Every Character with authoritative Realm-core party membership. + fn party_member_guids(&self) -> Result> { + Ok(Vec::new()) + } + + /// Every party id held by this store. Realm-core rows are authoritative; World Shard rows are + /// mirrors that reconnect reconciliation must also inspect for stale parties. + fn party_group_ids(&self) -> Result> { + Ok(Vec::new()) + } + /// `sync_group_mirror` — replace this shard's mirror of one party with realm-core's roster. /// An empty `roster.members` is the disband tombstone. fn sync_group_mirror(&self, _roster: &party::GroupRoster) -> Result<()> { @@ -386,6 +424,20 @@ pub trait WorldStore: /// usually a peer, so this is not account-scoped. fn character_by_guid(&self, guid: u64) -> Result>; + /// Does any World Shard hold this Character? `false` means every configured Shard was readable + /// and had no row. An incomplete, unhealthy, or changing Shard set must return `Err`. + fn character_exists_on_any_world_shard(&self, guid: u64) -> Result { + if self.character_by_guid(guid)?.is_some() { + return Ok(true); + } + for shard in self.world_stores() { + if shard.character_by_guid(guid)?.is_some() { + return Ok(true); + } + } + Ok(false) + } + /// Look up a creature template by entry to answer `CMSG_CREATURE_QUERY` (Tier 2 / NPCs). fn creature_template(&self, entry: u32) -> Result>; diff --git a/gateway/src/world/tests.rs b/gateway/src/world/tests.rs index 0047355b8..1654eabe2 100644 --- a/gateway/src/world/tests.rs +++ b/gateway/src/world/tests.rs @@ -275,6 +275,8 @@ struct InMemoryStore { username: String, session: Option, characters: Vec, + /// When set, `character_by_guid` cannot answer from this Shard. + character_read_error: Option, login_entity: Option, /// The auctioneer's house and faction verdict returned to the focused auction seam. auction_interaction: Option, @@ -549,6 +551,8 @@ struct InMemoryStore { /// the topology can be wired up AFTER every handle exists — production reads the shared /// `ShardSet`, which has the same shape and the same "includes this handle" membership. peers: std::sync::Mutex>>, + /// When set, the configured World Shard set is incomplete or unhealthy. + world_shard_set_error: Option, /// Unclaimed bot invite intent ids on this World Shard. Two concurrent consumers share this /// collection, matching the Module table both Gateways call into. bot_invite_intents: std::sync::Mutex>, @@ -571,6 +575,8 @@ struct InMemoryStore { /// When set, `sync_group_mirror` fails with this message — a world shard that cannot be /// mirrored (an unreachable database), which must not fail a party op realm-core already took. mirror_error: Option, + /// How many mirror writes fail before this Shard accepts one. + mirror_failures: std::sync::atomic::AtomicUsize, /// What `realm_whisper` was asked to deliver on THIS handle — /// `(sender_guid, target_guid, message, sender_is_ignored)`. The realm handle owns the list; a /// world shard's staying empty is how a test tells "the whisper went to the authority" from "it @@ -621,6 +627,12 @@ struct InMemoryStore { /// reachable in production — a concurrent op on another socket — and what it must not do is leave /// the invite dialog hanging. party_accept_error: Option, + /// How many Realm-core LEAVE calls fail before one reaches the party state. + party_leave_failures: std::sync::atomic::AtomicUsize, + /// Return a connection failure after the next Realm-core LEAVE commits. + party_leave_commit_then_error: std::sync::atomic::AtomicBool, + /// When set, deleted Character cleanup cannot reach Realm-core. + party_cleanup_realm_error: Option, /// The transfer step to fail at, simulating a gateway killed before that step's /// transaction committed. `None` = nothing fails. kill_at: Option, @@ -1307,8 +1319,26 @@ impl WorldStore for InMemoryStore { } } fn character_by_guid(&self, guid: u64) -> Result> { + if let Some(error) = &self.character_read_error { + return Err(anyhow!(error.clone())); + } Ok(self.characters.iter().find(|c| c.guid == guid).cloned()) } + + fn character_exists_on_any_world_shard(&self, guid: u64) -> Result { + if let Some(error) = &self.world_shard_set_error { + return Err(anyhow!(error.clone())); + } + if self.character_by_guid(guid)?.is_some() { + return Ok(true); + } + for shard in self.peers.lock().unwrap().iter() { + if shard.character_by_guid(guid)?.is_some() { + return Ok(true); + } + } + Ok(false) + } fn creature_template(&self, _entry: u32) -> Result> { Ok(None) } @@ -2386,6 +2416,13 @@ impl WorldStore for InMemoryStore { .map(|r| r as std::sync::Arc) } + fn party_cleanup_realm(&self) -> Result>> { + if let Some(error) = &self.party_cleanup_realm_error { + return Err(anyhow!(error.clone())); + } + Ok(self.realm_store()) + } + fn world_stores(&self) -> Vec> { self.peers .lock() @@ -2467,9 +2504,27 @@ impl WorldStore for InMemoryStore { p.events.push((inviter, kind::DECLINE)); } realm_op::LEAVE => { + if self + .party_leave_failures + .fetch_update( + std::sync::atomic::Ordering::SeqCst, + std::sync::atomic::Ordering::SeqCst, + |remaining| remaining.checked_sub(1), + ) + .is_ok() + { + return Err(anyhow!("Realm-core LEAVE connection interrupted")); + } if p.group_of(actor_guid).is_none() { return Ok(GroupRefusal::NotInGroup.into()); } + if self + .party_leave_commit_then_error + .swap(false, std::sync::atomic::Ordering::SeqCst) + { + p.remove_member(actor_guid); + return Err(anyhow!("Realm-core LEAVE reply was lost after commit")); + } p.remove_member(actor_guid); } realm_op::UNINVITE => { @@ -2530,8 +2585,53 @@ impl WorldStore for InMemoryStore { .cloned()) } + fn party_member_guids(&self) -> Result> { + if !self.is_realm { + return Ok(Vec::new()); + } + Ok(self + .party + .lock() + .unwrap() + .members + .iter() + .map(|(_, guid)| *guid) + .collect()) + } + + fn party_group_ids(&self) -> Result> { + if self.is_realm { + return Ok(self + .party + .lock() + .unwrap() + .groups + .iter() + .map(|(group_id, ..)| *group_id) + .collect()); + } + Ok(self + .mirror + .lock() + .unwrap() + .iter() + .map(|roster| roster.group_id) + .collect()) + } + fn sync_group_mirror(&self, roster: &super::party::GroupRoster) -> Result<()> { self.rec("sync_group_mirror"); + if self + .mirror_failures + .fetch_update( + std::sync::atomic::Ordering::SeqCst, + std::sync::atomic::Ordering::SeqCst, + |remaining| remaining.checked_sub(1), + ) + .is_ok() + { + return Err(anyhow!("World Shard mirror connection interrupted")); + } if let Some(e) = &self.mirror_error { return Err(anyhow!("{e}")); } From 08cb0b2fd1a9cdecef77ca41800880e44a7cd9ee Mon Sep 17 00:00:00 2001 From: T3 Code Date: Mon, 7 Sep 2026 18:12:16 +0000 Subject: [PATCH 2/4] fix(gateway): preserve party cleanup diagnostics --- gateway/src/stdb/connection.rs | 10 +- gateway/src/stdb/subscriptions.rs | 291 +----------------- ...scriptions_character_gone_durable_tests.rs | 283 +++++++++++++++++ gateway/src/stdb/world_store.rs | 37 +-- gateway/src/world/party_tests.rs | 1 + 5 files changed, 303 insertions(+), 319 deletions(-) create mode 100644 gateway/src/stdb/subscriptions_character_gone_durable_tests.rs diff --git a/gateway/src/stdb/connection.rs b/gateway/src/stdb/connection.rs index c347d2a01..803605697 100644 --- a/gateway/src/stdb/connection.rs +++ b/gateway/src/stdb/connection.rs @@ -200,9 +200,15 @@ impl LiveConn { )), }; if let Err(error) = subscription.unsubscribe() { - return Err(anyhow!( + 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 } diff --git a/gateway/src/stdb/subscriptions.rs b/gateway/src/stdb/subscriptions.rs index db65a7ea7..c2ba200d7 100644 --- a/gateway/src/stdb/subscriptions.rs +++ b/gateway/src/stdb/subscriptions.rs @@ -42,293 +42,6 @@ pub struct PlayerSubscriptions { view: Option>, } -#[cfg(test)] -mod character_gone_durable_tests { - use super::*; - use crate::accept::BlockingTaskCapacity; - use crate::config::GatewayConfig; - use crate::durable_test_support::{poll_until, Standalone, POLL_TIMEOUT}; - use crate::world::party::PartyOutcome; - use std::path::{Path, PathBuf}; - use std::process::{Command, Output}; - - struct PrivateCli { - config: PathBuf, - } - - impl PrivateCli { - fn new(owner_token: &str) -> Self { - let nonce = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap() - .as_nanos(); - let config = std::env::temp_dir().join(format!( - "lyracore-party-cleanup-cli-{}-{nonce}.toml", - std::process::id() - )); - std::fs::write(&config, format!("spacetimedb_token = {owner_token:?}\n")).unwrap(); - Self { config } - } - - fn command(&self) -> Command { - let binary = std::env::var_os("SPACETIME_BIN").unwrap_or_else(|| "spacetime".into()); - let mut command = Command::new(binary); - command.args(["--config-path", self.config.to_str().unwrap()]); - command - } - - fn publish(&self, server: &str, database: &str) -> Output { - let workspace = Path::new(env!("CARGO_MANIFEST_DIR")).parent().unwrap(); - self.command() - .current_dir(workspace) - .args([ - "publish", - "-s", - server, - "--module-path", - workspace.join("module").to_str().unwrap(), - "--build-options=--features=debug_reducers", - "-y", - database, - ]) - .output() - .unwrap() - } - - fn call(&self, server: &str, database: &str, reducer: &str, args: &[&str]) -> Output { - let mut command = self.command(); - command.args(["call", "-s", server, database, reducer, "--"]); - command.args(args); - command.output().unwrap() - } - } - - impl Drop for PrivateCli { - fn drop(&mut self) { - let _ = std::fs::remove_file(&self.config); - } - } - - struct TopologyEnv { - shard_map: Option, - realm_core: Option, - } - - impl TopologyEnv { - fn install(shard_map: &str, realm_core: &str) -> Self { - let previous = Self { - shard_map: std::env::var_os("LYRACORE_SHARD_MAP"), - realm_core: std::env::var_os("LYRACORE_REALM_CORE"), - }; - std::env::set_var("LYRACORE_SHARD_MAP", shard_map); - std::env::set_var("LYRACORE_REALM_CORE", realm_core); - previous - } - } - - impl Drop for TopologyEnv { - fn drop(&mut self) { - match &self.shard_map { - Some(value) => std::env::set_var("LYRACORE_SHARD_MAP", value), - None => std::env::remove_var("LYRACORE_SHARD_MAP"), - } - match &self.realm_core { - Some(value) => std::env::set_var("LYRACORE_REALM_CORE", value), - None => std::env::remove_var("LYRACORE_REALM_CORE"), - } - } - } - - fn reconciliation_is_idle(coordinator: &Coordinator) -> bool { - !coordinator - .1 - .party_reconciliation_requested - .load(Ordering::Acquire) - && !coordinator - .1 - .party_reconciliation_running - .load(Ordering::Acquire) - } - - fn reconciliation_is_running(coordinator: &Coordinator) -> bool { - coordinator - .1 - .party_reconciliation_running - .load(Ordering::Acquire) - } - - #[test] - #[ignore = "requires the SpacetimeDB 2.7.1 CLI and Wasm toolchain"] - fn another_gateway_waits_for_the_transfer_then_cleans_the_deleted_character() { - const INSTANCES: &str = "party-cleanup-instances"; - const OTHER: &str = "party-cleanup-other"; - const REALM: &str = "party-cleanup-realm"; - const INSTANCES_SURVIVOR: u64 = 2; - const OTHER_SURVIVOR: u64 = 1_000_000_001; - - let mut standalone = Standalone::start("party-cleanup-world"); - standalone.publish_module(); - let cli = PrivateCli::new(&standalone.owner_token()); - for database in [INSTANCES, OTHER, REALM] { - let output = cli.publish(standalone.server(), database); - standalone.assert_output_success(&output, &format!("publish {database}")); - } - standalone.assert_call("claim_operator", &[]); - for database in [INSTANCES, OTHER, REALM] { - let output = cli.call(standalone.server(), database, "claim_operator", &[]); - standalone.assert_output_success(&output, &format!("claim {database} operator")); - } - for database in [INSTANCES, OTHER] { - let output = cli.call( - standalone.server(), - database, - "debug_delete_character", - &["1"], - ); - standalone.assert_output_success(&output, &format!("clear Character 1 on {database}")); - } - for (database, base) in [(INSTANCES, "0"), (OTHER, "1000000000")] { - let output = cli.call(standalone.server(), database, "install_guid_range", &[base]); - standalone.assert_output_success(&output, &format!("install range on {database}")); - } - for (database, name) in [(INSTANCES, "PartyTwo"), (OTHER, "PartyOther")] { - let output = cli.call( - standalone.server(), - database, - "create_character", - &["1", name, "1", "1", "0", "0", "0", "0", "0", "0"], - ); - standalone.assert_output_success(&output, &format!("create {name}")); - } - standalone.assert_call("debug_spawn_player_entity", &["1"]); - - let _topology = TopologyEnv::install(&format!("36:*={INSTANCES}, 1:*={OTHER}"), REALM); - let cfg = GatewayConfig { - logon_bind: "127.0.0.1:0".into(), - world_bind: "127.0.0.1:0".into(), - stdb_uri: standalone.server().into(), - module_name: standalone.shard_name().into(), - coordinator_token: Some(standalone.owner_token()), - gateway_id: "party-cleanup-test".into(), - blocking_task_capacity: BlockingTaskCapacity::new(1), - }; - let runtime = tokio::runtime::Runtime::new().unwrap(); - let world = runtime.block_on(Coordinator::connect(&cfg)).unwrap(); - let instances = world.shard_handle(INSTANCES).unwrap(); - let realm = world.realm_core().unwrap(); - let observer = runtime.block_on(Coordinator::connect(&cfg)).unwrap(); - let observer_instances = observer.shard_handle(INSTANCES).unwrap(); - observer.spawn_character_gone_relay(); - assert!(poll_until(POLL_TIMEOUT, || reconciliation_is_idle( - &observer - ))); - - use lyracore_shared::group::realm_op; - assert_eq!( - realm - .realm_group_op(realm_op::INVITE, 1, INSTANCES_SURVIVOR, 0, 0) - .unwrap(), - PartyOutcome::Ran - ); - assert_eq!( - realm - .realm_group_op(realm_op::ACCEPT, INSTANCES_SURVIVOR, 0, 0, 0) - .unwrap(), - PartyOutcome::Ran - ); - assert_eq!( - realm - .realm_group_op(realm_op::INVITE, 1, OTHER_SURVIVOR, 0, 0) - .unwrap(), - PartyOutcome::Ran - ); - assert_eq!( - realm - .realm_group_op(realm_op::ACCEPT, OTHER_SURVIVOR, 0, 0, 0) - .unwrap(), - PartyOutcome::Ran - ); - assert!(poll_until(POLL_TIMEOUT, || realm - .group_roster(1) - .is_some_and( - |roster| roster.members == [1, INSTANCES_SURVIVOR, OTHER_SURVIVOR] - ))); - let roster = realm.group_roster(1).unwrap(); - for (_, shard) in world.world_shards() { - shard.sync_group_mirror(&roster).unwrap(); - } - assert!(poll_until(POLL_TIMEOUT, || world - .world_shards() - .into_iter() - .all(|(_, shard)| shard.group_roster(1).is_some_and( - |roster| roster.members == [1, INSTANCES_SURVIVOR, OTHER_SURVIVOR] - )))); - - let source_revision = observer - .0 - .coord() - .character_revision() - .load(Ordering::Acquire); - observer_instances.park_pump_for_test(); - let output = cli.call( - standalone.server(), - standalone.shard_name(), - "debug_bot_transfer", - &["1", "36", "1", "-11208", "1672", "24", "0", "party-test"], - ); - standalone.assert_output_success(&output, "transfer Character"); - crate::world::transfer::run_bot_transfer(&world, 1, 36, 1, "party-test").unwrap(); - assert!(poll_until(POLL_TIMEOUT, || { - world.character_by_guid(1).unwrap().is_none() - && instances.character_by_guid(1).unwrap().is_some() - && observer - .0 - .coord() - .character_revision() - .load(Ordering::Acquire) - > source_revision - })); - assert!(poll_until(POLL_TIMEOUT, || reconciliation_is_running( - &observer - ))); - std::thread::sleep(Duration::from_millis(250)); - assert!(realm.group_roster(1).is_some()); - observer_instances.resume_pump_for_test(); - assert!(poll_until(POLL_TIMEOUT, || reconciliation_is_idle( - &observer - ))); - assert!(realm.group_roster(1).is_some()); - - let destination_revision = observer_instances - .0 - .coord() - .character_revision() - .load(Ordering::Acquire); - let output = cli.call( - standalone.server(), - INSTANCES, - "debug_delete_character", - &["1"], - ); - standalone.assert_output_success(&output, "delete transferred Character"); - assert!(poll_until(POLL_TIMEOUT, || { - observer_instances - .0 - .coord() - .character_revision() - .load(Ordering::Acquire) - > destination_revision - && reconciliation_is_idle(&observer) - && realm.group_roster(1).is_none() - })); - let survivors = realm.group_roster(INSTANCES_SURVIVOR).unwrap(); - assert_eq!(survivors.members, [INSTANCES_SURVIVOR, OTHER_SURVIVOR]); - assert!(world.world_shards().into_iter().all(|(_, shard)| shard - .group_roster(INSTANCES_SURVIVOR) - .is_some_and(|roster| roster == survivors))); - } -} - impl PlayerSubscriptions { /// A no-op guard (used by the in-memory test store; drops harmlessly). #[cfg(test)] @@ -7362,3 +7075,7 @@ mod tests { ); } } + +#[cfg(test)] +#[path = "subscriptions_character_gone_durable_tests.rs"] +mod character_gone_durable_tests; diff --git a/gateway/src/stdb/subscriptions_character_gone_durable_tests.rs b/gateway/src/stdb/subscriptions_character_gone_durable_tests.rs new file mode 100644 index 000000000..d8ae06e29 --- /dev/null +++ b/gateway/src/stdb/subscriptions_character_gone_durable_tests.rs @@ -0,0 +1,283 @@ +use super::*; +use crate::accept::BlockingTaskCapacity; +use crate::config::GatewayConfig; +use crate::durable_test_support::{poll_until, Standalone, POLL_TIMEOUT}; +use crate::world::party::PartyOutcome; +use std::path::{Path, PathBuf}; +use std::process::{Command, Output}; + +struct PrivateCli { + config: PathBuf, +} + +impl PrivateCli { + fn new(owner_token: &str) -> Self { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let config = std::env::temp_dir().join(format!( + "lyracore-party-cleanup-cli-{}-{nonce}.toml", + std::process::id() + )); + std::fs::write(&config, format!("spacetimedb_token = {owner_token:?}\n")).unwrap(); + Self { config } + } + + fn command(&self) -> Command { + let binary = std::env::var_os("SPACETIME_BIN").unwrap_or_else(|| "spacetime".into()); + let mut command = Command::new(binary); + command.args(["--config-path", self.config.to_str().unwrap()]); + command + } + + fn publish(&self, server: &str, database: &str) -> Output { + let workspace = Path::new(env!("CARGO_MANIFEST_DIR")).parent().unwrap(); + self.command() + .current_dir(workspace) + .args([ + "publish", + "-s", + server, + "--module-path", + workspace.join("module").to_str().unwrap(), + "--build-options=--features=debug_reducers", + "-y", + database, + ]) + .output() + .unwrap() + } + + fn call(&self, server: &str, database: &str, reducer: &str, args: &[&str]) -> Output { + let mut command = self.command(); + command.args(["call", "-s", server, database, reducer, "--"]); + command.args(args); + command.output().unwrap() + } +} + +impl Drop for PrivateCli { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.config); + } +} + +struct TopologyEnv { + shard_map: Option, + realm_core: Option, +} + +impl TopologyEnv { + fn install(shard_map: &str, realm_core: &str) -> Self { + let previous = Self { + shard_map: std::env::var_os("LYRACORE_SHARD_MAP"), + realm_core: std::env::var_os("LYRACORE_REALM_CORE"), + }; + std::env::set_var("LYRACORE_SHARD_MAP", shard_map); + std::env::set_var("LYRACORE_REALM_CORE", realm_core); + previous + } +} + +impl Drop for TopologyEnv { + fn drop(&mut self) { + match &self.shard_map { + Some(value) => std::env::set_var("LYRACORE_SHARD_MAP", value), + None => std::env::remove_var("LYRACORE_SHARD_MAP"), + } + match &self.realm_core { + Some(value) => std::env::set_var("LYRACORE_REALM_CORE", value), + None => std::env::remove_var("LYRACORE_REALM_CORE"), + } + } +} + +fn reconciliation_is_idle(coordinator: &Coordinator) -> bool { + !coordinator + .1 + .party_reconciliation_requested + .load(Ordering::Acquire) + && !coordinator + .1 + .party_reconciliation_running + .load(Ordering::Acquire) +} + +fn reconciliation_is_running(coordinator: &Coordinator) -> bool { + coordinator + .1 + .party_reconciliation_running + .load(Ordering::Acquire) +} + +#[test] +#[ignore = "requires the SpacetimeDB 2.7.1 CLI and Wasm toolchain"] +fn another_gateway_waits_for_the_transfer_then_cleans_the_deleted_character() { + const INSTANCES: &str = "party-cleanup-instances"; + const OTHER: &str = "party-cleanup-other"; + const REALM: &str = "party-cleanup-realm"; + const INSTANCES_SURVIVOR: u64 = 2; + const OTHER_SURVIVOR: u64 = 1_000_000_001; + + let mut standalone = Standalone::start("party-cleanup-world"); + standalone.publish_module(); + let cli = PrivateCli::new(&standalone.owner_token()); + for database in [INSTANCES, OTHER, REALM] { + let output = cli.publish(standalone.server(), database); + standalone.assert_output_success(&output, &format!("publish {database}")); + } + standalone.assert_call("claim_operator", &[]); + for database in [INSTANCES, OTHER, REALM] { + let output = cli.call(standalone.server(), database, "claim_operator", &[]); + standalone.assert_output_success(&output, &format!("claim {database} operator")); + } + for database in [INSTANCES, OTHER] { + let output = cli.call( + standalone.server(), + database, + "debug_delete_character", + &["1"], + ); + standalone.assert_output_success(&output, &format!("clear Character 1 on {database}")); + } + for (database, base) in [(INSTANCES, "0"), (OTHER, "1000000000")] { + let output = cli.call(standalone.server(), database, "install_guid_range", &[base]); + standalone.assert_output_success(&output, &format!("install range on {database}")); + } + for (database, name) in [(INSTANCES, "PartyTwo"), (OTHER, "PartyOther")] { + let output = cli.call( + standalone.server(), + database, + "create_character", + &["1", name, "1", "1", "0", "0", "0", "0", "0", "0"], + ); + standalone.assert_output_success(&output, &format!("create {name}")); + } + standalone.assert_call("debug_spawn_player_entity", &["1"]); + + let _topology = TopologyEnv::install(&format!("36:*={INSTANCES}, 1:*={OTHER}"), REALM); + let cfg = GatewayConfig { + logon_bind: "127.0.0.1:0".into(), + world_bind: "127.0.0.1:0".into(), + stdb_uri: standalone.server().into(), + module_name: standalone.shard_name().into(), + coordinator_token: Some(standalone.owner_token()), + gateway_id: "party-cleanup-test".into(), + blocking_task_capacity: BlockingTaskCapacity::new(1), + }; + let runtime = tokio::runtime::Runtime::new().unwrap(); + let world = runtime.block_on(Coordinator::connect(&cfg)).unwrap(); + let instances = world.shard_handle(INSTANCES).unwrap(); + let realm = world.realm_core().unwrap(); + let observer = runtime.block_on(Coordinator::connect(&cfg)).unwrap(); + let observer_instances = observer.shard_handle(INSTANCES).unwrap(); + observer.spawn_character_gone_relay(); + assert!(poll_until(POLL_TIMEOUT, || reconciliation_is_idle( + &observer + ))); + + use lyracore_shared::group::realm_op; + assert_eq!( + realm + .realm_group_op(realm_op::INVITE, 1, INSTANCES_SURVIVOR, 0, 0) + .unwrap(), + PartyOutcome::Ran + ); + assert_eq!( + realm + .realm_group_op(realm_op::ACCEPT, INSTANCES_SURVIVOR, 0, 0, 0) + .unwrap(), + PartyOutcome::Ran + ); + assert_eq!( + realm + .realm_group_op(realm_op::INVITE, 1, OTHER_SURVIVOR, 0, 0) + .unwrap(), + PartyOutcome::Ran + ); + assert_eq!( + realm + .realm_group_op(realm_op::ACCEPT, OTHER_SURVIVOR, 0, 0, 0) + .unwrap(), + PartyOutcome::Ran + ); + assert!(poll_until(POLL_TIMEOUT, || realm + .group_roster(1) + .is_some_and( + |roster| roster.members == [1, INSTANCES_SURVIVOR, OTHER_SURVIVOR] + ))); + let roster = realm.group_roster(1).unwrap(); + for (_, shard) in world.world_shards() { + shard.sync_group_mirror(&roster).unwrap(); + } + assert!(poll_until(POLL_TIMEOUT, || world + .world_shards() + .into_iter() + .all(|(_, shard)| shard.group_roster(1).is_some_and( + |roster| roster.members == [1, INSTANCES_SURVIVOR, OTHER_SURVIVOR] + )))); + + let source_revision = observer + .0 + .coord() + .character_revision() + .load(Ordering::Acquire); + observer_instances.park_pump_for_test(); + let output = cli.call( + standalone.server(), + standalone.shard_name(), + "debug_bot_transfer", + &["1", "36", "1", "-11208", "1672", "24", "0", "party-test"], + ); + standalone.assert_output_success(&output, "transfer Character"); + crate::world::transfer::run_bot_transfer(&world, 1, 36, 1, "party-test").unwrap(); + assert!(poll_until(POLL_TIMEOUT, || { + world.character_by_guid(1).unwrap().is_none() + && instances.character_by_guid(1).unwrap().is_some() + && observer + .0 + .coord() + .character_revision() + .load(Ordering::Acquire) + > source_revision + })); + assert!(poll_until(POLL_TIMEOUT, || reconciliation_is_running( + &observer + ))); + std::thread::sleep(Duration::from_millis(250)); + assert!(realm.group_roster(1).is_some()); + observer_instances.resume_pump_for_test(); + assert!(poll_until(POLL_TIMEOUT, || reconciliation_is_idle( + &observer + ))); + assert!(realm.group_roster(1).is_some()); + + let destination_revision = observer_instances + .0 + .coord() + .character_revision() + .load(Ordering::Acquire); + let output = cli.call( + standalone.server(), + INSTANCES, + "debug_delete_character", + &["1"], + ); + standalone.assert_output_success(&output, "delete transferred Character"); + assert!(poll_until(POLL_TIMEOUT, || { + observer_instances + .0 + .coord() + .character_revision() + .load(Ordering::Acquire) + > destination_revision + && reconciliation_is_idle(&observer) + && realm.group_roster(1).is_none() + })); + let survivors = realm.group_roster(INSTANCES_SURVIVOR).unwrap(); + assert_eq!(survivors.members, [INSTANCES_SURVIVOR, OTHER_SURVIVOR]); + assert!(world.world_shards().into_iter().all(|(_, shard)| shard + .group_roster(INSTANCES_SURVIVOR) + .is_some_and(|roster| roster == survivors))); +} diff --git a/gateway/src/stdb/world_store.rs b/gateway/src/stdb/world_store.rs index 8481fad81..5f736a34f 100644 --- a/gateway/src/stdb/world_store.rs +++ b/gateway/src/stdb/world_store.rs @@ -12,17 +12,13 @@ //! serializes the `MovementInfo` before enqueueing it on the shard's shared batch. use anyhow::{anyhow, Result}; -use spacetimedb_sdk::Table; use wow_world_messages::vanilla::MovementInfo; use crate::codec; use crate::realm_core::SessionKey; use crate::world::{SessionTx, WorldSession, WorldStore, MOVE_SUBMITTED}; -use super::bindings::{ - game_group_member_table::GameGroupMemberTableAccess, game_group_table::GameGroupTableAccess, - GwMove, -}; +use super::bindings::GwMove; use super::connection::{CharacterPresenceSnapshot, Coordinator}; use super::views::{AccountRow, RealmRow}; use super::PlayerSubscriptions; @@ -1106,36 +1102,17 @@ impl WorldStore for Coordinator { &self, group_id: u64, ) -> Result> { - let live = self.0.coord(); - if !live.is_healthy() { + let healthy = { + let live = self.0.coord(); + live.is_healthy() + }; + if !healthy { anyhow::bail!( "{} has no healthy Coordinator subscription for party cleanup", self.shard_name() ); } - let db = &live.conn.db; - let Some(group) = db - .game_group() - .iter() - .find(|group| group.group_id == group_id) - else { - return Ok(None); - }; - let mut rows: Vec<(u64, u64)> = db - .game_group_member() - .iter() - .filter(|member| member.group_id == group_id) - .map(|member| (member.id, member.character_guid)) - .collect(); - rows.sort_unstable(); - Ok(Some(crate::world::party::GroupRoster { - group_id, - leader_guid: group.leader_guid, - loot_method: group.loot_method, - loot_threshold: group.loot_threshold, - master_looter_guid: group.master_looter_guid, - members: rows.into_iter().map(|(_, guid)| guid).collect(), - })) + Ok(self.group_roster_by_id(group_id)) } fn party_member_guids(&self) -> Result> { diff --git a/gateway/src/world/party_tests.rs b/gateway/src/world/party_tests.rs index 51b212004..24f4d7c0c 100644 --- a/gateway/src/world/party_tests.rs +++ b/gateway/src/world/party_tests.rs @@ -613,6 +613,7 @@ fn deleted_character_cleanup_flushes_pending_loot_before_leaving_realm_core() { corpse_guid: 90, slot: 1, item_entry: 100, + random_property_id: 0, deadline_micros: 500, recipients: vec![GINGER, VIM, TRIN], }]; From ed4171d52eca642ce6888ac502d85aac5a467e2f Mon Sep 17 00:00:00 2001 From: T3 Code Date: Mon, 7 Sep 2026 19:11:03 +0000 Subject: [PATCH 3/4] test(gateway): retain loot property in relay recorder --- gateway/src/world/loot_tests.rs | 10 ++++++---- gateway/src/world/tests.rs | 8 +++++--- 2 files changed, 11 insertions(+), 7 deletions(-) diff --git a/gateway/src/world/loot_tests.rs b/gateway/src/world/loot_tests.rs index 66a6a2644..6c550d414 100644 --- a/gateway/src/world/loot_tests.rs +++ b/gateway/src/world/loot_tests.rs @@ -65,7 +65,8 @@ fn a_sharded_store_routes_the_vote_to_realm_core_with_the_authenticated_guid() { GINGER, lyracore_shared::loot_roll::vote_kind::GREED, 0, - vec![] + vec![], + 0 )], "sharded, the vote must reach REALM-CORE, addressed by the AUTHENTICATED guid" ); @@ -124,7 +125,7 @@ fn relay_tick_promotes_a_staging_roll_and_clears_it() { item_entry: 1234, deadline_micros: 999_999, recipients: vec![GINGER, TRIN], - random_property_id: 0, + random_property_id: 0x1234_5678, }]; let mut watermark = 0u64; loot::relay_tick(world.as_ref(), &mut watermark); @@ -138,7 +139,8 @@ fn relay_tick_promotes_a_staging_roll_and_clears_it() { 0, 0, 999_999, - vec![GINGER, TRIN] + vec![GINGER, TRIN], + 0x1234_5678 )], "the promoted roll must carry the ORIGINAL deadline and the FULL recipient snapshot" ); @@ -217,7 +219,7 @@ fn a_leave_flushes_every_connected_shards_pending_rolls_before_the_disband_reach assert_eq!( realm.realm_loot_ops.lock().unwrap().clone(), - vec![(loot_op::START, 500, 2, 1234, 0, 0, 999, vec![GINGER])], + vec![(loot_op::START, 500, 2, 1234, 0, 0, 999, vec![GINGER], 0)], "the pending roll on the OTHER shard must be promoted before the disband, not left behind" ); assert_eq!( diff --git a/gateway/src/world/tests.rs b/gateway/src/world/tests.rs index 1654eabe2..c87c08eb2 100644 --- a/gateway/src/world/tests.rs +++ b/gateway/src/world/tests.rs @@ -662,10 +662,11 @@ struct InMemoryStore { // Test recorder: the tuple is `realm_loot_op`'s argument list verbatim. #[allow(clippy::type_complexity)] /// Recorded `realm_loot_op` calls — `(op, corpse_guid, slot, item_entry, actor_guid, vote, - /// deadline_micros, recipients)` — every arg the gateway's loot-roll routing/relay passed. The + /// deadline_micros, recipients, random_property_id)` — every arg the gateway's loot-roll + /// routing/relay passed. The /// realm handle owns this; a world shard's staying empty is how a test tells "the vote/promotion /// went to the authority" from "it stayed shard-local". - realm_loot_ops: std::sync::Mutex)>>, + realm_loot_ops: std::sync::Mutex, u32)>>, /// When set, `realm_loot_op` fails with this message. realm_loot_op_error: Option, /// This WORLD SHARD's staging rolls `pending_local_rolls` answers — the relay's promotion @@ -2721,7 +2722,7 @@ impl WorldStore for InMemoryStore { vote: u8, deadline_micros: i64, recipients: Vec, - _random_property_id: u32, + random_property_id: u32, ) -> Result<()> { self.rec("realm_loot_op"); self.realm_loot_ops.lock().unwrap().push(( @@ -2733,6 +2734,7 @@ impl WorldStore for InMemoryStore { vote, deadline_micros, recipients, + random_property_id, )); if let Some(e) = &self.realm_loot_op_error { return Err(anyhow!("{e}")); From d94bc1d5bf7365e5bec118cbd98fab1d04b165ce Mon Sep 17 00:00:00 2001 From: T3 Code Date: Mon, 7 Sep 2026 19:34:07 +0000 Subject: [PATCH 4/4] fix(gateway): preserve party cleanup retries --- gateway/src/world/party.rs | 33 ++++++++++++++++++++----- gateway/src/world/party_tests.rs | 41 ++++++++++++++++++++++++++++++++ gateway/src/world/tests.rs | 18 ++++++++++++++ 3 files changed, 86 insertions(+), 6 deletions(-) diff --git a/gateway/src/world/party.rs b/gateway/src/world/party.rs index a06b2f1a2..1d1f6997f 100644 --- a/gateway/src/world/party.rs +++ b/gateway/src/world/party.rs @@ -569,11 +569,22 @@ fn run_server_leave( previous_group_id: before, }); }; - if before.is_some() && realm.group_roster(leaver_guid)?.is_none() { - return Ok(ServerLeave { - outcome: PartyOutcome::Ran, - previous_group_id: before, - }); + if before.is_some() { + match realm.group_roster(leaver_guid) { + Ok(None) => { + return Ok(ServerLeave { + outcome: PartyOutcome::Ran, + previous_group_id: before, + }); + } + Ok(Some(_)) => {} + Err(confirmation_error) => { + log::warn!( + "party: could not confirm Character {leaver_guid} membership after Realm-core \ + LEAVE failed ({confirmation_error:#}); retrying" + ); + } + } } Err(error) } @@ -641,7 +652,17 @@ pub(crate) fn reconcile_deleted_character_parties(store: &St) -> let mut group_ids: std::collections::HashSet = realm.party_group_ids()?.into_iter().collect(); for shard in store.world_stores() { - group_ids.extend(shard.party_group_ids()?); + match shard.party_group_ids() { + Ok(ids) => group_ids.extend(ids), + Err(error) => { + failures += 1; + log::warn!( + "party: could not enumerate mirrored groups on {} ({error:#}); retrying", + shard.shard_name() + ); + last_error = Some(error); + } + } } for group_id in group_ids { if let Err(error) = sync_group_mirrors_required(store, realm.as_ref(), group_id) { diff --git a/gateway/src/world/party_tests.rs b/gateway/src/world/party_tests.rs index 24f4d7c0c..3b17f3413 100644 --- a/gateway/src/world/party_tests.rs +++ b/gateway/src/world/party_tests.rs @@ -577,6 +577,47 @@ fn deleted_character_cleanup_stops_after_three_realm_core_leave_failures() { ); } +#[test] +fn roster_confirmation_failure_preserves_the_realm_core_leave_error() { + let (realm, world, _instances, _calls) = topology_after_vim_is_deleted(); + realm + .party_leave_failures + .store(3, std::sync::atomic::Ordering::SeqCst); + *realm.group_roster_error_on_read.lock().unwrap() = + Some((2, "Realm-core roster confirmation unavailable".into())); + + let error = party::cleanup_deleted_character(world.as_ref(), VIM) + .expect_err("cleanup must retain the failed LEAVE as its primary error"); + + assert!(error.to_string().contains("LEAVE connection interrupted")); + assert!(!error + .to_string() + .contains("roster confirmation unavailable")); + assert!(realm.group_roster(VIM).unwrap().is_some()); +} + +#[test] +fn reconciliation_continues_when_one_shard_cannot_enumerate_mirrored_groups() { + let (realm, world, instances, _calls) = party_topology(); + form_split_party(&world, &instances); + world.mirror.lock().unwrap().clear(); + *instances.party_group_ids_error.lock().unwrap() = + Some("instances party subscription unavailable".into()); + + let error = party::reconcile_deleted_character_parties(world.as_ref()) + .expect_err("the unavailable Shard read must keep reconciliation pending"); + + let diagnostic = format!("{error:#}"); + assert!(diagnostic.contains("1 deleted Character party reconciliation attempt")); + assert!(diagnostic.contains("instances party subscription unavailable")); + let authoritative = realm.group_roster(GINGER).unwrap().unwrap(); + assert_eq!( + world.mirror.lock().unwrap().as_slice(), + std::slice::from_ref(&authoritative), + "the known realm-core group must still repair healthy Shard mirrors" + ); +} + #[test] fn unsharded_deleted_character_cleanup_stays_on_the_module_sweep() { let store = InMemoryStore { diff --git a/gateway/src/world/tests.rs b/gateway/src/world/tests.rs index c87c08eb2..211a54d51 100644 --- a/gateway/src/world/tests.rs +++ b/gateway/src/world/tests.rs @@ -631,6 +631,12 @@ struct InMemoryStore { party_leave_failures: std::sync::atomic::AtomicUsize, /// Return a connection failure after the next Realm-core LEAVE commits. party_leave_commit_then_error: std::sync::atomic::AtomicBool, + /// Fail one `group_roster` read by its one-based call number. + group_roster_error_on_read: std::sync::Mutex>, + /// Number of `group_roster` reads on this handle. + group_roster_reads: std::sync::atomic::AtomicUsize, + /// When set, this handle cannot enumerate its mirrored party ids. + party_group_ids_error: std::sync::Mutex>, /// When set, deleted Character cleanup cannot reach Realm-core. party_cleanup_realm_error: Option, /// The transfer step to fail at, simulating a gateway killed before that step's @@ -2560,6 +2566,15 @@ impl WorldStore for InMemoryStore { } fn group_roster(&self, character_guid: u64) -> Result> { + let read = self + .group_roster_reads + .fetch_add(1, std::sync::atomic::Ordering::SeqCst) + + 1; + if let Some((error_read, error)) = &*self.group_roster_error_on_read.lock().unwrap() { + if read == *error_read { + return Err(anyhow!(error.clone())); + } + } if self.is_realm { let p = self.party.lock().unwrap(); return Ok(p.group_of(character_guid).and_then(|g| p.roster(g))); @@ -2601,6 +2616,9 @@ impl WorldStore for InMemoryStore { } fn party_group_ids(&self) -> Result> { + if let Some(error) = &*self.party_group_ids_error.lock().unwrap() { + return Err(anyhow!(error.clone())); + } if self.is_realm { return Ok(self .party