From 8ae0891f267601615c20e1cad2099acfd0f7bfb3 Mon Sep 17 00:00:00 2001 From: Mathih13 Date: Sat, 10 Oct 2026 23:14:27 +0200 Subject: [PATCH 1/3] fix(transfer): keep Kalimdor entry from ending the World Session --- .github/workflows/module-durable.yml | 1 + docs/character-transfer.md | 10 + gateway/src/realm_core.rs | 236 ++++++++++++++---- gateway/src/stdb/realm_db.rs | 11 - gateway/src/stdb/subscriptions.rs | 2 +- .../src/world/shard_routing_durable_tests.rs | 153 ++++++++++++ gateway/src/world/tests.rs | 2 + module/src/realm_core.rs | 37 ++- module/tests/realm_transfer_locator.rs | 109 ++++++++ 9 files changed, 486 insertions(+), 75 deletions(-) create mode 100644 gateway/src/world/shard_routing_durable_tests.rs create mode 100644 module/tests/realm_transfer_locator.rs diff --git a/.github/workflows/module-durable.yml b/.github/workflows/module-durable.yml index 13758367e..16de6bd68 100644 --- a/.github/workflows/module-durable.yml +++ b/.github/workflows/module-durable.yml @@ -99,6 +99,7 @@ jobs: cargo test -p lyracore-gateway --bin lyracore-gateway stdb::subscriptions::character_gone_durable_tests::another_gateway_waits_for_the_transfer_then_cleans_the_deleted_character -- --ignored --exact cargo test -p lyracore-gateway --bin lyracore-gateway stdb::subscriptions::roster_relay_durable_tests::a_roster_change_made_only_on_realm_core_reaches_every_world_shard_mirror -- --ignored --exact cargo test -p lyracore-gateway --bin lyracore-gateway stdb::subscriptions::meeting_stone_durable_tests::seekers_on_two_world_shards_form_one_party_that_reaches_every_mirror -- --ignored --exact + cargo test -p lyracore-gateway --bin lyracore-gateway world::tests::shard_routing_durable_tests -- --ignored --test-threads=1 cargo test -p lyracore-gateway --bin lyracore-gateway stdb::armor::tests::resistance_projection_adds_property_effects_only_while_the_item_is_worn_and_unbroken -- --ignored --exact cargo test -p lyracore-gateway --test provision_cli provision_writes_the_account_to_the_world_shard_and_realm_core -- --ignored --exact cargo test -p lyracore-gateway --test startup_relays -- --ignored --test-threads=1 diff --git a/docs/character-transfer.md b/docs/character-transfer.md index 082b24f9a..7b65c209a 100644 --- a/docs/character-transfer.md +++ b/docs/character-transfer.md @@ -47,5 +47,15 @@ receipt means the destination is unconsulted. The reaper holds the Escrow, and t forward recovery at the next world entry. A frozen Character is recoverable; deleting or duplicating one is not. +A Character's first cross-shard Transfer may have no Realm-core locator. The Gateway supplies its +source Character's location with predecessor revision zero. Realm-core creates revision one +directly as pending, so the destination is never advertised as settled before import. Only that +exact pending crossing can replay the initialization. Bot Transfers still require their bound +predecessor. + +While the locator is pending, login checks the World Shards for source Escrow before accepting a +destination copy. After the source finishes, login can settle and release the destination's exact +arrival fence. Login repair never overwrites a pending locator. + The Fakes execute the shared planners and protocol bodies. Durable tests exercise reducer shims, real table arms, and SpacetimeDB transaction rollback. These checks cover different boundaries. diff --git a/gateway/src/realm_core.rs b/gateway/src/realm_core.rs index 56ae890d7..9dff5cde5 100644 --- a/gateway/src/realm_core.rs +++ b/gateway/src/realm_core.rs @@ -68,8 +68,6 @@ pub(crate) trait RealmDb: Clone + Sized + Send + Sync { // --- the character→shard index /// Where THIS database's own rows say the character is. `None` = not here. fn character_location(&self, guid: u64) -> Option<(u32, u64)>; - /// This database's `game_character_shard` entry for `guid` — a HINT, never the authority. - fn character_shard(&self, guid: u64) -> Option<(u32, u64)>; /// Write `guid`'s location into this database's character→shard index. fn set_character_shard(&self, guid: u64, map_id: u32, instance_id: u64) -> Result<()>; /// Read the ordered Realm partition and any pending Transfer phase. @@ -77,7 +75,8 @@ pub(crate) trait RealmDb: Clone + Sized + Send + Sync { &self, guid: u64, ) -> Result>; - /// Begin one exact Realm Transfer phase from its observed predecessor. + /// Begin one exact Realm Transfer phase from its observed predecessor. Revision zero requires + /// an absent human locator and creates pending revision one. // The arguments preserve the Realm reducer's predecessor, destination, and crossing Gate. #[allow(clippy::too_many_arguments)] fn begin_character_shard_transfer( @@ -269,12 +268,10 @@ pub(crate) fn begin_shard_index_transfer( bot_intent: Option<(&crate::world::transfer::BotTransferIntent, u64)>, ) -> Result { let realm = db.realm_core()?; - let locator = realm - .realm_character_partition(plan.character_guid)? - .ok_or_else(|| anyhow!("Transfer source has no Realm locator"))?; + let locator = realm.realm_character_partition(plan.character_guid)?; let (source_map, source_instance, source_revision, source_identity, intent_id, generation) = - match bot_intent { - Some((intent, _claim_token)) => { + match (locator, bot_intent) { + (Some(locator), Some((intent, _claim_token))) => { if intent.source_locator_revision == 0 { anyhow::bail!("bot Transfer has no bound Realm locator predecessor"); } @@ -322,7 +319,7 @@ pub(crate) fn begin_shard_index_transfer( intent.controller_generation, ) } - None => ( + (Some(locator), None) => ( locator.map_id, locator.instance_id, locator.revision, @@ -330,6 +327,20 @@ pub(crate) fn begin_shard_index_transfer( 0, 0, ), + (None, None) => { + let (map_id, instance_id) = db + .character_location(plan.character_guid) + .ok_or_else(|| anyhow!("Transfer source has no Character"))?; + ( + map_id, + instance_id, + 0, + spacetimedb_sdk::Identity::ZERO, + 0, + 0, + ) + } + (None, Some(_)) => anyhow::bail!("Transfer source has no Realm locator"), }; realm.begin_character_shard_transfer( source_map, @@ -344,21 +355,22 @@ pub(crate) fn begin_shard_index_transfer( )?; let deadline = std::time::Instant::now() + std::time::Duration::from_secs(3); loop { - let observed = realm + if let Some(observed) = realm .realm_character_partition(plan.character_guid)? - .ok_or_else(|| anyhow!("Transfer source Realm locator disappeared"))?; - if observed.transfer_pending - && (observed.map_id, observed.instance_id, observed.revision) - == (source_map, source_instance, source_revision) - && ( - observed.pending_destination_map, - observed.pending_destination_instance, - ) == (plan.dest_map_id, plan.dest_instance_id) - && ( - observed.bot_source_identity, - observed.bot_transfer_intent_id, - observed.bot_controller_generation, - ) == (source_identity, intent_id, generation) + .filter(|row| { + row.transfer_pending + && (row.map_id, row.instance_id, row.revision) + == (source_map, source_instance, source_revision.max(1)) + && ( + row.pending_destination_map, + row.pending_destination_instance, + ) == (plan.dest_map_id, plan.dest_instance_id) + && ( + row.bot_source_identity, + row.bot_transfer_intent_id, + row.bot_controller_generation, + ) == (source_identity, intent_id, generation) + }) { return Ok(observed); } @@ -547,12 +559,18 @@ pub(crate) fn locate_home_shard(db: &D, guid: u64) -> Option { let shards = db.world_shards(); let is_connected = |d: &str| shards.iter().any(|(n, _)| n == d); - // Fast path: trust the hint only far enough to know which ONE shard to ask. + // A pending locator may already name the destination map. Scan for source Escrow before + // accepting an imported destination copy. if let Some(rc) = &index { - if let Some(hint) = rc.character_shard(guid) { - let hinted_name = db - .shard_map() - .resolve_connected(hint.0, hint.1, is_connected); + if let Some(hint) = rc + .realm_character_partition(guid) + .ok() + .flatten() + .filter(|row| !row.transfer_pending) + { + let hinted_name = + db.shard_map() + .resolve_connected(hint.map_id, hint.instance_id, is_connected); if let Some((_, hinted)) = shards.iter().find(|(n, _)| n == hinted_name) { if hinted.has_escrow(guid) || hinted.character_location(guid).is_some() { heal_shard_entry(rc, guid, hinted); @@ -637,7 +655,10 @@ fn heal_shard_entry(rc: &D, guid: u64, shard: &D) { if shard.shard_map().resolve(actual.0, actual.1) != shard.shard_name() { return; } - if rc.character_shard(guid) == Some(actual) { + let Ok(locator) = rc.realm_character_partition(guid) else { + return; + }; + if locator.is_some_and(|row| row.transfer_pending || (row.map_id, row.instance_id) == actual) { return; } if let Err(e) = rc.set_character_shard(guid, actual.0, actual.1) { @@ -1077,12 +1098,6 @@ pub(crate) mod fake { .get(&guid) .map(|(_, loc)| *loc) } - fn character_shard(&self, guid: u64) -> Option<(u32, u64)> { - let db = self.store(); - db.note(&format!("character_shard({guid})")); - self.stored_partition(guid) - .map(|partition| (partition.map_id, partition.instance_id)) - } fn set_character_shard(&self, guid: u64, map_id: u32, instance_id: u64) -> Result<()> { let db = self.store(); db.note(&format!("set_character_shard({guid})")); @@ -1152,8 +1167,11 @@ pub(crate) mod fake { ) -> Result<()> { let db = self.store(); db.note(&format!("begin_character_shard_transfer({character_guid})")); - if source_revision == 0 - || (source_module_identity == spacetimedb_sdk::Identity::ZERO) != (intent_id == 0) + if (source_module_identity == spacetimedb_sdk::Identity::ZERO) != (intent_id == 0) + || (source_revision == 0 + && (source_module_identity != spacetimedb_sdk::Identity::ZERO + || intent_id != 0 + || controller_generation != 0)) { return Err(anyhow!("Transfer locator identity is incomplete")); } @@ -1161,12 +1179,35 @@ pub(crate) mod fake { let current = phases .get(&character_guid) .copied() - .or_else(|| self.indexed_partition(character_guid)) - .ok_or_else(|| anyhow!("Transfer source has no Realm locator"))?; + .or_else(|| self.indexed_partition(character_guid)); + let Some(current) = current else { + if source_revision != 0 { + return Err(anyhow!("Transfer source has no Realm locator")); + } + phases.insert( + character_guid, + crate::world::party::RealmCharacterPartition { + map_id: source_map, + instance_id: source_instance, + revision: 1, + transfer_pending: true, + pending_destination_map: destination_map, + pending_destination_instance: destination_instance, + bot_source_identity: source_module_identity, + bot_transfer_intent_id: intent_id, + bot_controller_generation: controller_generation, + }, + ); + db.shard_index + .lock() + .unwrap() + .insert(character_guid, (source_map, source_instance)); + return Ok(()); + }; let crossing = (source_module_identity, intent_id, controller_generation); if current.transfer_pending && (current.map_id, current.instance_id, current.revision) - == (source_map, source_instance, source_revision) + == (source_map, source_instance, source_revision.max(1)) && ( current.pending_destination_map, current.pending_destination_instance, @@ -1179,7 +1220,8 @@ pub(crate) mod fake { { return Ok(()); } - if current.transfer_pending + if source_revision == 0 + || current.transfer_pending || (current.map_id, current.instance_id, current.revision) != (source_map, source_instance, source_revision) { @@ -1892,11 +1934,7 @@ mod tests { ); } - /// The read this whole issue is about: a correct index entry must be TRUSTED, not just - /// consulted-then-ignored. If `locate_home_shard` stopped reading the index (falling straight - /// through to the scan), this still "routes correctly" by accident — the scan finds the same - /// answer — so the pin is the ACCESS LOG, not the routing answer: deleting the index - /// consultation makes both assertions below fail. + /// A settled locator avoids reads on unrelated World Shards. #[test] fn locate_home_shard_trusts_a_correct_index_entry_without_scanning_the_other_shard() { let h = routed_realm(); @@ -1911,14 +1949,6 @@ mod tests { INSTANCES, "must route to the shard the index names" ); - assert!( - h.db_at(CORE) - .touched() - .iter() - .any(|c| c.starts_with("character_shard(100)")), - "the realm-core index was never consulted — `settle_home_shard`'s production path \ - would be back to the unconditional scan this lookup exists to replace" - ); assert!( h.db_at(WORLD).touched().is_empty(), "a correct index entry must not cost a scan of the OTHER connected shard — reads \ @@ -1960,6 +1990,99 @@ mod tests { ); } + #[test] + fn a_first_kalimdor_login_can_begin_transfer_without_a_realm_locator() { + let h = realm( + &[WORLD, "kalimdor", INSTANCES, CORE], + "1:*=kalimdor,36:*=instances", + Some(CORE), + ); + h.db_at(WORLD) + .characters + .lock() + .unwrap() + .insert(100, (1, (1, 0))); + let holder = locate_home_shard(&h, 100).expect("the Character remains on its Home Shard"); + assert_eq!(holder.shard_name(), WORLD); + assert!(h.at(CORE).realm_character_partition(100).unwrap().is_none()); + + let pending = begin_shard_index_transfer(&holder, &kalimdor_plan(), None) + .expect("the first Kalimdor login must begin its Transfer"); + assert!(pending.transfer_pending); + assert_eq!(pending.revision, 1); + assert_eq!(pending.pending_destination_map, 1); + assert_eq!(pending.pending_destination_instance, 0); + } + + fn kalimdor_plan() -> crate::world::transfer::TransferPlan { + crate::world::transfer::TransferPlan { + transfer_id: 100, + character_guid: 100, + dest_map_id: 1, + dest_instance_id: 0, + dest_x: 0.0, + dest_y: 0.0, + dest_z: 0.0, + dest_o: 0.0, + } + } + + #[test] + fn an_in_world_transfer_can_begin_without_a_realm_locator() { + let h = realm(&[WORLD, INSTANCES, CORE], "36:*=instances", Some(CORE)); + h.store() + .characters + .lock() + .unwrap() + .insert(100, (1, (0, 0))); + let pending = begin_shard_index_transfer(&h, &kalimdor_plan(), None).unwrap(); + assert_eq!( + (pending.map_id, pending.instance_id, pending.revision), + (0, 0, 1) + ); + assert!(pending.transfer_pending); + assert_eq!(pending.pending_destination_map, 1); + } + + #[test] + fn a_missing_character_cannot_create_a_realm_transfer_locator() { + let h = realm(&[WORLD, INSTANCES, CORE], "36:*=instances", Some(CORE)); + assert!(begin_shard_index_transfer(&h, &kalimdor_plan(), None).is_err()); + assert!(h.at(CORE).realm_character_partition(100).unwrap().is_none()); + } + + #[test] + fn a_pending_destination_hint_keeps_routing_to_source_escrow_until_source_finish() { + let h = realm( + &[WORLD, "kalimdor", INSTANCES, CORE], + "1:*=kalimdor,36:*=instances", + Some(CORE), + ); + for shard in [WORLD, "kalimdor"] { + h.db_at(shard) + .characters + .lock() + .unwrap() + .insert(100, (1, (1, 0))); + } + h.store().escrows.lock().unwrap().insert(100); + let pending = begin_shard_index_transfer(&h, &kalimdor_plan(), None).unwrap(); + + assert_eq!(locate_home_shard(&h, 100).unwrap().shard_name(), WORLD); + assert_eq!( + h.at(CORE).realm_character_partition(100).unwrap(), + Some(pending) + ); + + h.store().characters.lock().unwrap().remove(&100); + h.store().escrows.lock().unwrap().remove(&100); + assert_eq!(locate_home_shard(&h, 100).unwrap().shard_name(), "kalimdor"); + assert_eq!( + h.at(CORE).realm_character_partition(100).unwrap(), + Some(pending) + ); + } + #[test] fn a_correct_index_entry_is_not_rewritten_by_locate_home_shard() { let h = routed_realm(); @@ -2166,7 +2289,10 @@ mod tests { realm .set_character_shard(100, 0, 0) .expect("a later settled repair succeeds"); - assert_eq!(realm.character_shard(100), Some((0, 0))); + assert_eq!( + realm.store().shard_index.lock().unwrap().get(&100), + Some(&(0, 0)) + ); let partition = realm .realm_character_partition(100) .expect("partition read") diff --git a/gateway/src/stdb/realm_db.rs b/gateway/src/stdb/realm_db.rs index 928ec90de..df985dfd2 100644 --- a/gateway/src/stdb/realm_db.rs +++ b/gateway/src/stdb/realm_db.rs @@ -65,17 +65,6 @@ impl RealmDb for Coordinator { self.character_location(guid) } - fn character_shard(&self, guid: u64) -> Option<(u32, u64)> { - self.0 - .coord() - .conn - .db - .game_character_shard() - .character_guid() - .find(&guid) - .map(|s| (s.map_id, s.instance_id)) - } - /// Call it on the Realm-core handle: on a World Shard `finish_transfer` maintains the index /// in the same transaction. fn set_character_shard(&self, guid: u64, map_id: u32, instance_id: u64) -> Result<()> { diff --git a/gateway/src/stdb/subscriptions.rs b/gateway/src/stdb/subscriptions.rs index cc7b8a4ac..976bfdfae 100644 --- a/gateway/src/stdb/subscriptions.rs +++ b/gateway/src/stdb/subscriptions.rs @@ -36,7 +36,7 @@ use super::world_index::{CellKey, EntityLayer}; use super::world_view::{self, Viewer, WorldView}; #[cfg(test)] -static DURABLE_TOPOLOGY_ENV_LOCK: Mutex<()> = Mutex::new(()); +pub(crate) static DURABLE_TOPOLOGY_ENV_LOCK: Mutex<()> = Mutex::new(()); /// RAII guard for one world session's shared-view registration, held by the world connection. /// Dropping the guard makes the session unreachable from every shard-level dispatcher. diff --git a/gateway/src/world/shard_routing_durable_tests.rs b/gateway/src/world/shard_routing_durable_tests.rs new file mode 100644 index 000000000..5696849e8 --- /dev/null +++ b/gateway/src/world/shard_routing_durable_tests.rs @@ -0,0 +1,153 @@ +use crate::accept::BlockingTaskCapacity; +use crate::config::GatewayConfig; +use crate::durable_test_support::{module_bytes, poll_until, Standalone, POLL_TIMEOUT}; +use crate::stdb::subscriptions::DURABLE_TOPOLOGY_ENV_LOCK; +use crate::stdb::Coordinator; +use crate::world::{ShardRoutingStore, TransferStore}; + +struct TopologyEnv { + previous: Vec<(&'static str, Option)>, + _guard: std::sync::MutexGuard<'static, ()>, +} + +impl TopologyEnv { + fn install(kalimdor: &str, instances: &str, realm: &str) -> Self { + let guard = DURABLE_TOPOLOGY_ENV_LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let map = format!("1:*={kalimdor},36:*={instances}"); + let values = [ + ("LYRACORE_SHARD_MAP", Some(map.as_str())), + ("LYRACORE_SHARD_MAP_FILE", None), + ("LYRACORE_REALM_CORE", Some(realm)), + ("LYRACORE_CALL_PIPES", Some("1")), + ]; + let previous = values + .into_iter() + .map(|(key, value)| { + let previous = std::env::var_os(key); + match value { + Some(value) => std::env::set_var(key, value), + None => std::env::remove_var(key), + } + (key, previous) + }) + .collect(); + Self { + previous, + _guard: guard, + } + } +} + +impl Drop for TopologyEnv { + fn drop(&mut self) { + for (key, value) in self.previous.drain(..) { + match value { + Some(value) => std::env::set_var(key, value), + None => std::env::remove_var(key), + } + } + } +} + +#[test] +#[ignore = "requires the SpacetimeDB 2.7.1 CLI and Wasm toolchain"] +fn a_first_kalimdor_transfer_arrives_and_recovers_after_import_or_source_finish() { + const KALIMDOR: &str = "first-transfer-kalimdor"; + const INSTANCES: &str = "first-transfer-instances"; + const REALM: &str = "first-transfer-realm"; + let mut node = Standalone::start_persistent("first-transfer-world"); + node.publish_module(); + node.assert_call("claim_operator", &[]); + node.assert_call("install_guid_range", &["0"]); + for (shard, base) in [ + (KALIMDOR, "1000000000"), + (INSTANCES, "2000000000"), + (REALM, "3000000000"), + ] { + node.publish_named_module_bytes(shard, module_bytes()); + node.assert_call_database(shard, "claim_operator", &[]); + node.assert_call_database(shard, "install_guid_range", &[base]); + } + + let _topology = TopologyEnv::install(KALIMDOR, INSTANCES, REALM); + let config = GatewayConfig { + logon_bind: "127.0.0.1:0".into(), + world_bind: "127.0.0.1:0".into(), + stdb_uri: node.server().into(), + module_name: node.shard_name().into(), + coordinator_token: Some(node.owner_token()), + gateway_id: "first-transfer-test".into(), + blocking_task_capacity: BlockingTaskCapacity::new(2), + }; + let runtime = tokio::runtime::Runtime::new().unwrap(); + let gateway = runtime.block_on(Coordinator::connect(&config)).unwrap(); + let destination = gateway.shard_handle(KALIMDOR).unwrap(); + + for (name, interruption) in [ + ("Firstarrival", None), + ("Afterimport", Some("import_character_blob")), + ("Afterfinish", Some("finish_transfer")), + ] { + node.assert_call( + "create_character", + &["1", name, "4", "1", "0", "0", "0", "0", "0", "0"], + ); + let rows = node.query_rows(&format!( + "SELECT guid FROM game_character WHERE name = '{name}'" + )); + let guid = rows[0]["guid"].parse::().unwrap(); + assert!(poll_until(POLL_TIMEOUT, || gateway + .character_location(guid) + == Some((1, 0)))); + assert!(gateway + .realm_core() + .unwrap() + .realm_character_partition(guid) + .unwrap() + .is_none()); + + if let Some(step) = interruption { + let plan = gateway.character_destination(guid).unwrap(); + let interrupted = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + crate::world::transfer::run_transfer_injected( + &gateway, + &destination, + &plan, + Some(step), + ) + })); + assert!( + interrupted.is_err(), + "Transfer must reach the committed {step} boundary" + ); + } + + // A fresh Coordinator reads the committed state as a replacement Gateway would. + let replacement = runtime.block_on(Coordinator::connect(&config)).unwrap(); + let holder = replacement.settle_home_shard(guid).unwrap().unwrap(); + assert_eq!(holder.shard_name(), KALIMDOR); + let character_query = format!("SELECT guid FROM game_character WHERE guid = {guid}"); + assert!(node.query_rows(&character_query).is_empty()); + assert_eq!( + node.query_database_rows(KALIMDOR, &character_query).len(), + 1 + ); + assert!(node + .query_rows(&format!( + "SELECT transfer_id FROM game_transfer_out WHERE character_guid = {guid}" + )) + .is_empty()); + assert!(node + .query_database_rows( + KALIMDOR, + &format!("SELECT transfer_id FROM game_transfer_in WHERE character_guid = {guid}") + ) + .is_empty()); + let locator = node.query_database_rows(REALM, &format!("SELECT revision, map_id, transfer_pending FROM game_character_shard WHERE character_guid = {guid}")); + assert_eq!(locator[0]["map_id"], "1"); + assert_eq!(locator[0]["revision"], "2"); + assert_eq!(locator[0]["transfer_pending"], "false"); + } +} diff --git a/gateway/src/world/tests.rs b/gateway/src/world/tests.rs index 2a1355ca8..41f8478ad 100644 --- a/gateway/src/world/tests.rs +++ b/gateway/src/world/tests.rs @@ -151,6 +151,8 @@ mod presence_tests; mod quest_tests; #[path = "session_tests.rs"] mod session_tests; +#[path = "shard_routing_durable_tests.rs"] +mod shard_routing_durable_tests; #[path = "shard_routing_tests.rs"] mod shard_routing_tests; #[path = "social_tests.rs"] diff --git a/module/src/realm_core.rs b/module/src/realm_core.rs index eaba3f90e..1907594fe 100644 --- a/module/src/realm_core.rs +++ b/module/src/realm_core.rs @@ -137,6 +137,8 @@ pub(crate) fn record_shard( /// Mark one crossing before the source is frozen. Party partition certification treats this Realm /// phase as pending without trusting a World Shard cache that can change after it is sampled. +/// A human crossing with source revision zero requires an absent locator and inserts pending +/// revision one. Replays must name that same pending crossing; they cannot replace a settled row. #[reducer] #[allow(clippy::too_many_arguments)] // Exact predecessor, destination, crossing, and Actor are the wire Gate. pub fn begin_character_shard_transfer( @@ -154,16 +156,34 @@ pub fn begin_character_shard_transfer( ) -> Result<(), String> { crate::helpers::require_operator(ctx)?; crate::account_ownership::require_actor_for(ctx, request_actor, character_guid)?; - if source_revision == 0 - || (source_module_identity == Identity::ZERO) != (transfer_intent_id == 0) + if (source_module_identity == Identity::ZERO) != (transfer_intent_id == 0) + || (source_revision == 0 + && (source_module_identity != Identity::ZERO + || transfer_intent_id != 0 + || controller_generation != 0)) { return Err("Transfer locator identity is incomplete".to_string()); } let table = ctx.db.game_character_shard(); - let current = table - .character_guid() - .find(character_guid) - .ok_or_else(|| "Transfer source has no Realm locator".to_string())?; + let Some(current) = table.character_guid().find(character_guid) else { + if source_revision != 0 { + return Err("Transfer source has no Realm locator".to_string()); + } + table.insert(CharacterShard { + character_guid, + map_id: source_map_id, + instance_id: source_instance_id, + updated_micros: ctx.timestamp.to_micros_since_unix_epoch(), + revision: 1, + bot_source_identity: source_module_identity, + bot_transfer_intent_id: transfer_intent_id, + bot_controller_generation: controller_generation, + transfer_pending: true, + pending_destination_map: destination_map_id, + pending_destination_instance: destination_instance_id, + }); + return Ok(()); + }; let crossing = ( source_module_identity, transfer_intent_id, @@ -171,7 +191,7 @@ pub fn begin_character_shard_transfer( ); if current.transfer_pending && (current.map_id, current.instance_id, current.revision) - == (source_map_id, source_instance_id, source_revision) + == (source_map_id, source_instance_id, source_revision.max(1)) && ( current.pending_destination_map, current.pending_destination_instance, @@ -184,7 +204,8 @@ pub fn begin_character_shard_transfer( { return Ok(()); } - if current.transfer_pending + if source_revision == 0 + || current.transfer_pending || (current.map_id, current.instance_id, current.revision) != (source_map_id, source_instance_id, source_revision) { diff --git a/module/tests/realm_transfer_locator.rs b/module/tests/realm_transfer_locator.rs new file mode 100644 index 000000000..10e696de8 --- /dev/null +++ b/module/tests/realm_transfer_locator.rs @@ -0,0 +1,109 @@ +mod support; + +use support::{Row, Standalone}; + +const ACTOR: &str = r#"{"guid":1,"ownership":null}"#; +const ZERO_IDENTITY: &str = "0x0000000000000000000000000000000000000000000000000000000000000000"; + +fn realm(name: &str) -> Standalone { + let mut realm = Standalone::start(name); + realm.publish_module(); + realm.assert_call("claim_operator", &[]); + assert!(locator(&realm).is_empty()); + realm +} + +fn locator(realm: &Standalone) -> Vec { + realm.query_rows("SELECT * FROM game_character_shard WHERE character_guid = 1") +} + +fn begin(revision: &str) -> [&str; 10] { + [ + "1", + "1", + "0", + revision, + "1", + "0", + ZERO_IDENTITY, + "0", + "0", + ACTOR, + ] +} + +fn assert_begin_refused(realm: &Standalone, args: &[&str]) { + let before = locator(realm); + let output = realm.call("begin_character_shard_transfer", args); + assert!( + !output.status.success(), + "the competing Transfer must be refused" + ); + assert_eq!( + locator(realm), + before, + "a Refusal must leave the locator intact" + ); +} + +#[test] +#[ignore = "requires the SpacetimeDB 2.7.1 CLI and Wasm toolchain"] +fn a_first_human_transfer_creates_only_a_pending_locator_and_replays_exactly() { + let realm = realm("first-human-transfer-locator"); + realm.assert_call("begin_character_shard_transfer", &begin("0")); + let pending = locator(&realm); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0]["map_id"], "1"); + assert_eq!(pending[0]["instance_id"], "0"); + assert_eq!(pending[0]["revision"], "1"); + assert_eq!(pending[0]["transfer_pending"], "true"); + assert_eq!(pending[0]["pending_destination_map"], "1"); + assert_eq!(pending[0]["pending_destination_instance"], "0"); + + realm.assert_call("begin_character_shard_transfer", &begin("0")); + realm.assert_call("begin_character_shard_transfer", &begin("1")); + assert_eq!(locator(&realm), pending); + + realm.assert_call("finish_character_shard_transfer", &begin("1")); + let settled = locator(&realm); + assert_eq!(settled[0]["revision"], "2"); + assert_eq!(settled[0]["transfer_pending"], "false"); + assert_begin_refused(&realm, &begin("0")); +} + +#[test] +#[ignore = "requires the SpacetimeDB 2.7.1 CLI and Wasm toolchain"] +fn first_transfer_initialization_cannot_replace_a_competing_pending_locator() { + let realm = realm("competing-first-transfer-locator"); + realm.assert_call("begin_character_shard_transfer", &begin("0")); + for (column, value) in [(1, "0"), (2, "7"), (3, "2"), (4, "36"), (5, "7")] { + let mut competing = begin("0"); + competing[column] = value; + assert_begin_refused(&realm, &competing); + } +} + +#[test] +#[ignore = "requires the SpacetimeDB 2.7.1 CLI and Wasm toolchain"] +fn first_transfer_initialization_cannot_replace_a_settled_locator() { + let realm = realm("settled-first-transfer-locator"); + realm.assert_call("set_character_shard", &["1", "1", "0", ACTOR]); + assert_eq!(locator(&realm)[0]["revision"], "1"); + assert_begin_refused(&realm, &begin("0")); +} + +#[test] +#[ignore = "requires the SpacetimeDB 2.7.1 CLI and Wasm toolchain"] +fn a_bot_cannot_create_a_missing_transfer_locator() { + let realm = realm("bot-missing-transfer-locator"); + for revision in ["0", "1"] { + let mut bot = begin(revision); + bot[6] = "0x0101010101010101010101010101010101010101010101010101010101010101"; + bot[7] = "10"; + bot[8] = "4"; + assert_begin_refused(&realm, &bot); + } + let mut incomplete = begin("0"); + incomplete[8] = "4"; + assert_begin_refused(&realm, &incomplete); +} From ab4da6f54426aa142fecddda262f1cf2936e19bb Mon Sep 17 00:00:00 2001 From: Mathih13 Date: Sat, 10 Oct 2026 23:22:47 +0200 Subject: [PATCH 2/3] test(transfer): verify Kalimdor login stays connected --- gateway/src/realm_core.rs | 36 ++++++----- .../src/world/shard_routing_durable_tests.rs | 61 ++++++++++++++++++- 2 files changed, 80 insertions(+), 17 deletions(-) diff --git a/gateway/src/realm_core.rs b/gateway/src/realm_core.rs index 9dff5cde5..ab0d176d6 100644 --- a/gateway/src/realm_core.rs +++ b/gateway/src/realm_core.rs @@ -542,16 +542,7 @@ fn wait_for_settled_locator( /// A missing or stale hint falls back to a census. An escrow holder takes precedence over /// a destination copy so a resumed transfer can complete from its source. pub(crate) fn locate_home_shard(db: &D, guid: u64) -> Option { - // This function's own copy of the short-circuit. `settle_home_shard` (`stdb::world_store`) already - // guards its call to this function with an identical `if !self.is_sharded() { return Ok(None); - // }` — that one is production's, stays exactly where it is, and cannot be exercised by any test - // without a live SpacetimeDB connection (`Coordinator` cannot be constructed offline). This copy - // is redundant on the happy path (never fires when the caller's guard already returned), but it - // is the only layer of "an unconfigured gateway reads NOTHING while resolving a home shard" that - // a test can actually reach — `fake::Handle` runs this function directly. Without it, a single - // connected database (unsharded) still pays a `character_shard` probe, a `character_location` - // scan and a `set_character_shard` heal write to reach the one answer "stay put" gives for - // free — exactly the reads an unset `LYRACORE_REALM_CORE`/`LYRACORE_SHARD_MAP` must never pay. + // An unsharded Gateway needs no locator read or repair to find its own Character. if !db.is_sharded() { return None; } @@ -741,6 +732,8 @@ pub(crate) mod fake { pub shard_index: Mutex>, /// Ordered `game_character_shard` Transfer phases used by the crossing tests. pub shard_phases: Mutex>, + /// Reads before a Realm locator subscription observes the committed row. + pub unobserved_locator_reads: Mutex, /// The node-issued identity of this database's per-account player connection. pub identities: Mutex>, /// How many identities this database has ever minted. Stamped into every identity so @@ -1151,6 +1144,11 @@ pub(crate) mod fake { ) -> Result> { self.store() .note(&format!("realm_character_partition({guid})")); + let mut remaining = self.store().unobserved_locator_reads.lock().unwrap(); + if *remaining > 0 { + *remaining -= 1; + return Ok(None); + } Ok(self.stored_partition(guid)) } fn begin_character_shard_transfer( @@ -2051,6 +2049,20 @@ mod tests { assert!(h.at(CORE).realm_character_partition(100).unwrap().is_none()); } + #[test] + fn a_first_transfer_waits_for_its_new_locator_subscription() { + let h = realm(&[WORLD, INSTANCES, CORE], "36:*=instances", Some(CORE)); + h.store() + .characters + .lock() + .unwrap() + .insert(100, (1, (1, 0))); + *h.db_at(CORE).unobserved_locator_reads.lock().unwrap() = 2; + let pending = begin_shard_index_transfer(&h, &kalimdor_plan(), None).unwrap(); + assert!(pending.transfer_pending); + assert_eq!(pending.revision, 1); + } + #[test] fn a_pending_destination_hint_keeps_routing_to_source_escrow_until_source_finish() { let h = realm( @@ -2289,10 +2301,6 @@ mod tests { realm .set_character_shard(100, 0, 0) .expect("a later settled repair succeeds"); - assert_eq!( - realm.store().shard_index.lock().unwrap().get(&100), - Some(&(0, 0)) - ); let partition = realm .realm_character_partition(100) .expect("partition read") diff --git a/gateway/src/world/shard_routing_durable_tests.rs b/gateway/src/world/shard_routing_durable_tests.rs index 5696849e8..95084117d 100644 --- a/gateway/src/world/shard_routing_durable_tests.rs +++ b/gateway/src/world/shard_routing_durable_tests.rs @@ -3,7 +3,55 @@ use crate::config::GatewayConfig; use crate::durable_test_support::{module_bytes, poll_until, Standalone, POLL_TIMEOUT}; use crate::stdb::subscriptions::DURABLE_TOPOLOGY_ENV_LOCK; use crate::stdb::Coordinator; -use crate::world::{ShardRoutingStore, TransferStore}; +use crate::world::TransferStore; + +fn login_on_kalimdor(gateway: Coordinator, runtime: &tokio::runtime::Runtime, guid: u64) { + use crate::world::test_support::{ + client_handshake, drain_world_entry, world_session_socket_pair, + }; + use wow_world_messages::vanilla::opcodes::ServerOpcodeMessage; + use wow_world_messages::vanilla::{ + ClientMessage, Map, CMSG_PING, CMSG_PLAYER_LOGIN, SMSG_PONG, + }; + use wow_world_messages::{Guid, Message}; + + for shard in [&gateway, &gateway.realm_core().unwrap()] { + let account = shard.account_by_username("TEST").unwrap().unwrap().id; + shard + .establish_session(account, &[7; 40], shard.bound_identity(account).unwrap()) + .unwrap(); + } + let (mut client, socket) = world_session_socket_pair(); + let handle = runtime.handle().clone(); + let session = std::thread::spawn(move || { + let _entered = handle.enter(); + crate::world::run_world_session(socket, std::sync::Arc::new(gateway)) + }); + let (mut encrypt, mut decrypt) = client_handshake(&mut client, "TEST", [7; 40]); + CMSG_PLAYER_LOGIN { + guid: Guid::new(guid), + } + .write_encrypted_client(&mut client, &mut encrypt) + .unwrap(); + let entry = drain_world_entry(&mut client, &mut decrypt); + assert!(entry.iter().any(|packet| matches!(packet, ServerOpcodeMessage::SMSG_LOGIN_VERIFY_WORLD(world) if world.map == Map::Kalimdor))); + CMSG_PING { + sequence_id: 7, + round_time_in_ms: 0, + } + .write_encrypted_client(&mut client, &mut encrypt) + .unwrap(); + let answered = (0..64).any(|_| { + let (opcode, body) = super::read_raw_frame(&mut client, &mut decrypt); + opcode == SMSG_PONG::OPCODE as u16 && body == 7u32.to_le_bytes() + }); + assert!( + answered, + "the Character must remain connected after Kalimdor entry" + ); + drop(client); + session.join().unwrap().unwrap(); +} struct TopologyEnv { previous: Vec<(&'static str, Option)>, @@ -61,6 +109,7 @@ fn a_first_kalimdor_transfer_arrives_and_recovers_after_import_or_source_finish( node.publish_module(); node.assert_call("claim_operator", &[]); node.assert_call("install_guid_range", &["0"]); + node.assert_sql("INSERT INTO game_start_position (race_class, race, class, map_id, zone_id, x, y, z, orientation, display_id) VALUES (1025, 4, 1, 1, 141, 0, 0, 0, 0, 55)"); for (shard, base) in [ (KALIMDOR, "1000000000"), (INSTANCES, "2000000000"), @@ -83,6 +132,13 @@ fn a_first_kalimdor_transfer_arrives_and_recovers_after_import_or_source_finish( }; let runtime = tokio::runtime::Runtime::new().unwrap(); let gateway = runtime.block_on(Coordinator::connect(&config)).unwrap(); + for shard in [node.shard_name(), KALIMDOR, INSTANCES, REALM] { + node.assert_call_database(shard, "gw_heartbeat", &[]); + } + { + let _entered = runtime.enter(); + gateway.spawn_gateway_heartbeat(); + } let destination = gateway.shard_handle(KALIMDOR).unwrap(); for (name, interruption) in [ @@ -126,8 +182,7 @@ fn a_first_kalimdor_transfer_arrives_and_recovers_after_import_or_source_finish( // A fresh Coordinator reads the committed state as a replacement Gateway would. let replacement = runtime.block_on(Coordinator::connect(&config)).unwrap(); - let holder = replacement.settle_home_shard(guid).unwrap().unwrap(); - assert_eq!(holder.shard_name(), KALIMDOR); + login_on_kalimdor(replacement, &runtime, guid); let character_query = format!("SELECT guid FROM game_character WHERE guid = {guid}"); assert!(node.query_rows(&character_query).is_empty()); assert_eq!( From 0a486ee876b4b55cb266f0d236678ca64883f058 Mon Sep 17 00:00:00 2001 From: Mathih13 Date: Sat, 10 Oct 2026 23:23:38 +0200 Subject: [PATCH 3/3] test(transfer): close the headless session without a reset --- gateway/src/world/shard_routing_durable_tests.rs | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/gateway/src/world/shard_routing_durable_tests.rs b/gateway/src/world/shard_routing_durable_tests.rs index 95084117d..25d1c1dcd 100644 --- a/gateway/src/world/shard_routing_durable_tests.rs +++ b/gateway/src/world/shard_routing_durable_tests.rs @@ -23,9 +23,11 @@ fn login_on_kalimdor(gateway: Coordinator, runtime: &tokio::runtime::Runtime, gu } let (mut client, socket) = world_session_socket_pair(); let handle = runtime.handle().clone(); + let (done, finished) = std::sync::mpsc::channel(); let session = std::thread::spawn(move || { let _entered = handle.enter(); - crate::world::run_world_session(socket, std::sync::Arc::new(gateway)) + let result = crate::world::run_world_session(socket, std::sync::Arc::new(gateway)); + let _ = done.send(result); }); let (mut encrypt, mut decrypt) = client_handshake(&mut client, "TEST", [7; 40]); CMSG_PLAYER_LOGIN { @@ -49,8 +51,12 @@ fn login_on_kalimdor(gateway: Coordinator, runtime: &tokio::runtime::Runtime, gu answered, "the Character must remain connected after Kalimdor entry" ); - drop(client); - session.join().unwrap().unwrap(); + client.shutdown(std::net::Shutdown::Write).unwrap(); + finished + .recv_timeout(std::time::Duration::from_secs(10)) + .expect("the World Session must stop after the Client closes its write side") + .unwrap(); + session.join().unwrap(); } struct TopologyEnv {