fix(transfer): keep Kalimdor entry from ending the World Session

.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

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.

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<()> {

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.

gateway/src/world/shard_routing_durable_tests.rs

@@ -0,0 +1,214 @@
+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::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 (done, finished) = std::sync::mpsc::channel();
+    let session = std::thread::spawn(move || {
+        let _entered = handle.enter();
+        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 {
+        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"
+    );
+    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 {
+    previous: Vec<(&'static str, Option<std::ffi::OsString>)>,
+    _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"]);
+    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"),
+        (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();
+    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 [
+        ("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::<u64>().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();
+        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!(
+            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");
+    }
+}

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"]

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,24 +156,42 @@ 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,
         controller_generation,
     );
     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)
     {