From b61a1864701ffdcbdac04b35bd7e5271f60f37e1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=94=D0=BC=D0=B8=D1=82=D1=80=D0=B8=D0=B9?= Date: Sat, 24 Oct 2026 11:12:01 +0500 Subject: [PATCH] Restore remote portable progress and resilient relay connections --- changelog.d/20261024-relay-recovery.md | 7 + changelog.d/20261024-remote-progress.md | 7 + changelog.d/20261024-remote-refresh.md | 7 + crates/coa-control-proto/src/app.rs | 92 ++- crates/coa-core/src/control/backend.rs | 632 +++++++++++++++++- crates/coa-core/src/control/direct_route.rs | 16 +- crates/coa-core/src/control/host_manager.rs | 16 +- crates/coa-core/src/control/join.rs | 51 +- .../src/control/join/portable_sync.rs | 290 ++++++++ crates/coa-core/src/control/player.rs | 18 +- crates/coa-core/src/control/relay_link.rs | 181 ++++- crates/coa-core/src/control/service.rs | 238 ++++++- .../src/control/service/portable_sync.rs | 209 ++++++ crates/coa-core/src/control/store.rs | 113 ++++ .../coa-core/src/portable/service/engine.rs | 60 +- .../portable/service/engine/remote_sync.rs | 345 ++++++++++ crates/coa-core/src/portable/store.rs | 2 +- crates/coa-core/src/portable/store/owner.rs | 23 + crates/coa-core/src/portable/store/tests.rs | 28 +- docs/PORTABLE-CHARACTERS.md | 2 + docs/de/PORTABLE-CHARACTERS.md | 2 + docs/es/PORTABLE-CHARACTERS.md | 2 + docs/fr/PORTABLE-CHARACTERS.md | 2 + docs/ru/PORTABLE-CHARACTERS.md | 2 + docs/zh-CN/PORTABLE-CHARACTERS.md | 2 + .../coa-control-e2e/tests/live_control.rs | 325 +++++++-- src-tauri/src/control_cmds.rs | 52 +- 27 files changed, 2579 insertions(+), 145 deletions(-) create mode 100644 changelog.d/20261024-relay-recovery.md create mode 100644 changelog.d/20261024-remote-progress.md create mode 100644 changelog.d/20261024-remote-refresh.md create mode 100644 crates/coa-core/src/control/join/portable_sync.rs create mode 100644 crates/coa-core/src/control/service/portable_sync.rs create mode 100644 crates/coa-core/src/portable/service/engine/remote_sync.rs diff --git a/changelog.d/20261024-relay-recovery.md b/changelog.d/20261024-relay-recovery.md new file mode 100644 index 0000000..fbc3250 --- /dev/null +++ b/changelog.d/20261024-relay-recovery.md @@ -0,0 +1,7 @@ +--- +area: manager +type: fixed +audience: admins +title: Relay connections recover and use the ports of the installed server +--- +Dropped or unresponsive relay links reconnect automatically. Abandoned game connections close, and tunnels refresh when local ports become available or change. Custom authentication ports are respected. diff --git a/changelog.d/20261024-remote-progress.md b/changelog.d/20261024-remote-progress.md new file mode 100644 index 0000000..1965a12 --- /dev/null +++ b/changelog.d/20261024-remote-progress.md @@ -0,0 +1,7 @@ +--- +area: manager +type: fixed +audience: admins +title: Portable progress resumes after reconnecting or restarting the Manager +--- +The player Manager now saves remote baselines, checkpoints and logout progress durably. Repeated logins stay monitored, and transfer retries preserve the same character and session. diff --git a/changelog.d/20261024-remote-refresh.md b/changelog.d/20261024-remote-refresh.md new file mode 100644 index 0000000..498325d --- /dev/null +++ b/changelog.d/20261024-remote-refresh.md @@ -0,0 +1,7 @@ +--- +area: manager +type: known-issue +audience: admins +title: Older copies on another host may require maintenance before returning +--- +Experimental portable play blocks an outdated existing copy on another host. Automatic refresh while that remote worldserver runs is not available. Your canonical character remains saved in your Manager. diff --git a/crates/coa-control-proto/src/app.rs b/crates/coa-control-proto/src/app.rs index 3317138..6df8d63 100644 --- a/crates/coa-control-proto/src/app.rs +++ b/crates/coa-control-proto/src/app.rs @@ -15,6 +15,15 @@ pub const MAX_USERNAME: usize = 16; pub const MIN_PASSWORD: usize = 8; pub const MAX_PASSWORD: usize = 16; pub const MAX_CHARACTERS: usize = 200; +pub const MAX_SESSION_BYTES: usize = 12 * 1024 * 1024; + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum SessionMessageKind { + Started, + Checkpoint, + Collection, +} /// A game account name: 3 to 16 ASCII letters, digits or `_`, not beginning or ending with `_`. pub fn valid_username(s: &str) -> bool { @@ -109,6 +118,15 @@ pub enum Request { total_size: usize, collections: BTreeMap>, }, + OwnedTransferOffer { + transfer_id: Uuid, + character_id: Uuid, + ruleset: String, + canonical_revision: u64, + content_hash: String, + total_size: usize, + collections: BTreeMap>, + }, /// A chunk of the encoded portable character snapshot payload. TransferChunk { transfer_id: Uuid, @@ -128,6 +146,24 @@ pub enum Request { transfer_id: Uuid, character_id: Uuid, }, + SessionPoll { + character_id: Uuid, + session_id: Uuid, + }, + SessionRead { + character_id: Uuid, + session_id: Uuid, + message_hash: String, + offset: usize, + }, + SessionAck { + character_id: Uuid, + session_id: Uuid, + message_hash: String, + offset: usize, + total_size: usize, + data: String, + }, /// Phase 13: Request a short-lived relay allocation for a relayed JOIN. AllocateRelay { player_id: Uuid, @@ -172,6 +208,22 @@ pub struct CharacterEntry { #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] #[serde(tag = "op", rename_all = "snake_case")] pub enum Response { + SessionPending { + kind: SessionMessageKind, + total_size: usize, + content_hash: String, + }, + SessionIdle { + closed: bool, + }, + SessionData { + offset: usize, + data: String, + }, + SessionAcked { + received_size: usize, + complete: bool, + }, Welcome { realm_id: coa_registry_proto::RealmId, automatic: bool, @@ -247,7 +299,10 @@ pub fn decode_request(bytes: &[u8]) -> Result { .map_err(|_| ControlError::Invalid("a request is not valid".into()))?; if !matches!( &r, - Request::TransferOffer { .. } | Request::TransferChunk { .. } + Request::TransferOffer { .. } + | Request::OwnedTransferOffer { .. } + | Request::TransferChunk { .. } + | Request::SessionAck { .. } ) && bytes.len() > MAX_CONTROL_REQUEST_BYTES { return Err(ControlError::Limit( @@ -255,6 +310,34 @@ pub fn decode_request(bytes: &[u8]) -> Result { )); } match &r { + Request::SessionRead { + message_hash, + offset, + .. + } => { + if message_hash.len() != 64 + || !message_hash.bytes().all(|b| b.is_ascii_hexdigit()) + || *offset > MAX_SESSION_BYTES + { + return invalid("invalid portable session read"); + } + } + Request::SessionAck { + message_hash, + offset, + total_size, + data, + .. + } => { + if message_hash.len() != 64 + || !message_hash.bytes().all(|b| b.is_ascii_hexdigit()) + || *total_size > MAX_SESSION_BYTES + || *offset > *total_size + || data.len() > 48 * 1024 + { + return invalid("invalid portable session acknowledgement chunk"); + } + } Request::Provision { desired, password, .. } => { @@ -280,6 +363,13 @@ pub fn decode_request(bytes: &[u8]) -> Result { total_size, collections, .. + } + | Request::OwnedTransferOffer { + ruleset, + content_hash, + total_size, + collections, + .. } => { if !matches!(ruleset.as_str(), "coa" | "wildcard" | "reborn") { return invalid("transfer ruleset must be coa, wildcard, or reborn"); diff --git a/crates/coa-core/src/control/backend.rs b/crates/coa-core/src/control/backend.rs index 8d7eef9..2a8a527 100644 --- a/crates/coa-core/src/control/backend.rs +++ b/crates/coa-core/src/control/backend.rs @@ -65,10 +65,16 @@ impl LocalBackend { fn require_portable(&self, local_id: &str) -> Result<()> { let settings = settings::load(&self.settings_dir)?; - if settings.realms.get(local_id).is_some_and(|cfg| cfg.allow_portable) { + if settings + .realms + .get(local_id) + .is_some_and(|cfg| cfg.allow_portable) + { Ok(()) } else { - Err(Error::Invalid("The realm owner has disabled portable characters.".into())) + Err(Error::Invalid( + "The realm owner has disabled portable characters.".into(), + )) } } @@ -116,6 +122,38 @@ impl LocalBackend { } impl RealmBackend for LocalBackend { + fn session_message( + &self, + local_id: &str, + character: Uuid, + session: Uuid, + account: u32, + ) -> Result { + let id = local_id.to_string(); + self.portable + .call(move |service| service.remote_session_message(&id, character, session, account)) + .map_err(|error| unavailable("portable engine", error))? + .map_err(|error| unavailable("portable session", error)) + } + + fn session_ack( + &self, + local_id: &str, + character: Uuid, + session: Uuid, + account: u32, + kind: app::SessionMessageKind, + bytes: &[u8], + ) -> Result<()> { + let id = local_id.to_string(); + let bytes = bytes.to_vec(); + self.portable + .call(move |service| { + service.apply_remote_session_ack(&id, character, session, account, kind, &bytes) + }) + .map_err(|error| unavailable("portable engine", error))? + .map_err(|error| unavailable("portable session", error)) + } fn info(&self, local_id: &str) -> Result { let cfg = settings::load(&self.settings_dir)?.realms.remove(local_id); let access = self.access(local_id)?; @@ -310,6 +348,23 @@ impl RealmBackend for LocalBackend { .map_err(|e| unavailable("import remote", e)) } + fn auth_port(&self, local_id: &str) -> Result { + let access = self.access(local_id)?; + let root = access + .root + .ok_or_else(|| Error::Invalid("The realm's auth port is unresolved.".into()))?; + let settings: serde_json::Value = + crate::fsx::read_json(&root.join("Settings/repack.json"))?; + match settings.get("authPort") { + None => Ok(coa_control_proto::relay::LOCAL_AUTH_PORT), + Some(value) => value + .as_u64() + .and_then(|port| u16::try_from(port).ok()) + .filter(|port| *port != 0) + .ok_or_else(|| Error::Invalid("The realm's auth port is invalid.".into())), + } + } + fn world_port(&self, local_id: &str) -> Result { if let Ok(override_port) = std::env::var("COA_OVERRIDE_LOCAL_WORLD_PORT") { if let Ok(p) = override_port.parse::() { @@ -346,20 +401,560 @@ mod tests { #[test] fn portable_access_requires_owner_opt_in_and_revokes_immediately() { let dir = tempdir().unwrap(); - let rt = PortableRuntime::start(&dir.path().join("portable"), Box::new(Vec::new), None).unwrap(); + let rt = + PortableRuntime::start(&dir.path().join("portable"), Box::new(Vec::new), None).unwrap(); let backend = LocalBackend::new(Arc::new(rt), dir.path(), dir.path(), Arc::new(Vec::new)); assert!(backend.require_portable("srv-test").is_err()); let mut cfg = settings::RegistrySettings::default(); - cfg.realms.insert("srv-test".into(), settings::PublishConfig { - realm_id: None, enabled: true, display_name: "Test".into(), description: String::new(), - language: "en".into(), region: None, existing_only: false, allow_portable: true, route: None, - }); + cfg.realms.insert( + "srv-test".into(), + settings::PublishConfig { + realm_id: None, + enabled: true, + display_name: "Test".into(), + description: String::new(), + language: "en".into(), + region: None, + existing_only: false, + allow_portable: true, + route: None, + }, + ); settings::save(dir.path(), &cfg).unwrap(); assert!(backend.require_portable("srv-test").is_ok()); cfg.realms.get_mut("srv-test").unwrap().allow_portable = false; settings::save(dir.path(), &cfg).unwrap(); - assert!(backend.export("srv-test", 1, 1).unwrap_err().to_string().contains("disabled portable")); - assert!(backend.check_transfer("srv-test", Uuid::new_v4(), 1).unwrap_err().to_string().contains("disabled portable")); + assert!(backend + .export("srv-test", 1, 1) + .unwrap_err() + .to_string() + .contains("disabled portable")); + assert!(backend + .check_transfer("srv-test", Uuid::new_v4(), 1) + .unwrap_err() + .to_string() + .contains("disabled portable")); + } + + #[test] + #[ignore = "Requires disposable A/B databases, B worldserver configuration and REGISTRY_URL"] + fn live_vps_transfer_uses_real_backend_and_preserves_the_working_copy() { + use crate::control::{ + host_link::HostLink, + join::{PlayerControl, RealmTarget}, + secrets::MemoryStore, + service::HostService, + store::ControlStore, + transport::coordinator_url, + }; + use crate::portable::{ + extension::ExtensionRegistry, + realm::{ + live_import::{number, realms, reset_b}, + live_projection::fixture_character, + profile::probe_capabilities, + }, + session::live_session::{spec, Server}, + store::Store, + }; + use crate::realm_registry::{advert::advertise, client::RegistryClient}; + use coa_registry_proto::{ + AccountProvisioning, Listing, Population, Rates, RealmId, RegisterRequest, + REGISTRY_PROTOCOL_VERSION, + }; + use std::collections::BTreeMap; + use std::sync::Mutex; + use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + + let base = std::env::var("REGISTRY_URL").expect("REGISTRY_URL"); + let root = std::path::PathBuf::from( + std::env::var("COA_REAL_CONTROL_ROOT").expect("disposable root"), + ); + assert!(root.starts_with("C:/games/source/candidate-evidence-20261021")); + let databases = realms().expect("disposable databases"); + reset_b(&databases.b); + let server = Server::start(spec().expect("disposable worldserver")); + struct Auth(std::process::Child); + impl Drop for Auth { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } + } + let auth_port = crate::layout::read_ports(&root).auth; + assert_eq!( + auth_port, 57637, + "Only the disposable auth port may be used" + ); + assert!(std::net::TcpStream::connect(("127.0.0.1", auth_port)).is_err()); + let mut command = std::process::Command::new(root.join("Core/authserver.exe")); + command + .arg("-c") + .arg(root.join("Core/configs/authserver.conf")) + .current_dir(root.join("Core")) + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()); + #[cfg(windows)] + { + use std::os::windows::process::CommandExt; + command.creation_flags(0x0800_0000); + } + let mut auth = Auth(command.spawn().unwrap()); + let end = Instant::now() + Duration::from_secs(30); + while std::net::TcpStream::connect(("127.0.0.1", auth_port)).is_err() { + assert!( + auth.0.try_wait().unwrap().is_none(), + "Disposable authserver exited" + ); + assert!(Instant::now() < end, "Disposable authserver did not listen"); + std::thread::sleep(Duration::from_millis(100)); + } + let mut ra = crate::ra::Ra::connect_to( + server.spec.ra_port, + &server.spec.user, + &server.spec.password, + ) + .unwrap(); + let caps = probe_capabilities( + &databases.b, + Some(&root.join("Data")), + Some(&mut ra), + &ExtensionRegistry::new(), + ) + .unwrap(); + let advertised = advertise(&caps).unwrap(); + let realm = RealmId::new(); + let mut seed = [0u8; 32]; + seed[..16].copy_from_slice(Uuid::new_v4().as_bytes()); + seed[16..].copy_from_slice(Uuid::new_v4().as_bytes()); + let key = ed25519_dalek::SigningKey::from_bytes(&seed); + let clock = || { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs() as i64 + }; + let registry = RegistryClient::new(&base).unwrap(); + let listing = Listing { + display_name: "Real portable candidate gate".into(), + description: "Disposable database and real worldserver test".into(), + language: "en".into(), + region: None, + rates: Rates::default(), + modules: vec![], + account_provisioning: AccountProvisioning { + automatic: true, + existing_only: false, + }, + manager_version: crate::MANAGER_VERSION.into(), + }; + registry + .register( + &key, + &RegisterRequest { + protocol_version: REGISTRY_PROTOCOL_VERSION, + realm_id: realm, + public_key: coa_registry_proto::sign::encode_public_key(&key.verifying_key()), + listing_hash: listing.hash(), + listing, + capabilities_hash: advertised.advert_hash(), + capabilities: advertised, + population: Population { + players: 0, + bots: 0, + capacity: None, + }, + }, + clock(), + ) + .unwrap(); + // Always unpublish this isolated identity, including after a failed assertion. + struct Lease(RegistryClient, RealmId, ed25519_dalek::SigningKey); + impl Drop for Lease { + fn drop(&mut self) { + let ts = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs() as i64; + let _ = self.0.unpublish(&self.2, &self.1, ts); + } + } + let _lease = Lease(registry, realm, key.clone()); + let temporary = tempdir().unwrap(); + let settings_dir = temporary.path().join("registry"); + let local_id = "srv-live-candidate"; + let mut cfg = settings::RegistrySettings::default(); + cfg.realms.insert( + local_id.into(), + settings::PublishConfig { + realm_id: Some(realm), + enabled: true, + display_name: "Candidate".into(), + description: String::new(), + language: "en".into(), + region: None, + existing_only: false, + allow_portable: true, + route: None, + }, + ); + settings::save(&settings_dir, &cfg).unwrap(); + let installed = root.clone(); + let runtime = Arc::new( + PortableRuntime::start( + &temporary.path().join("portable"), + Box::new(move || vec![("live-candidate".into(), installed.clone())]), + None, + ) + .unwrap(), + ); + let installed = root.clone(); + let backend = LocalBackend::new( + runtime.clone(), + &settings_dir, + temporary.path().join("descriptors"), + Arc::new(move || vec![("live-candidate".into(), installed.clone())]), + ); + assert_eq!( + backend.info(local_id).unwrap().ruleset, + crate::portable::Ruleset::Coa + ); + let service = Arc::new(HostService::new( + local_id, + realm, + Arc::new(Mutex::new(ControlStore::open_in_memory().unwrap())), + Arc::new(backend), + )); + let relay = crate::control::relay_link::RelayLink::start_with_local_ports( + coordinator_url(&base, "/relay/v1/host").unwrap(), + realm, + key.clone(), + auth_port, + Some(57638), + ); + service.set_relay(relay.clone()); + let mut link = HostLink::start( + coordinator_url(&base, "/coord/v1/host").unwrap(), + realm, + key.clone(), + service, + Arc::new(clock), + ); + let deadline = Instant::now() + Duration::from_secs(20); + while !link.status().connected && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(100)); + } + assert!(link.status().connected, "{:?}", link.status()); + let player_secrets = Arc::new(MemoryStore::default()); + let player = Arc::new( + PlayerControl::open(&temporary.path().join("player"), player_secrets.clone()).unwrap(), + ); + let target = RealmTarget { + realm_id: realm, + key: key.verifying_key(), + coordinator: coordinator_url(&base, "/coord/v1/player").unwrap(), + }; + let mut channel = player.connect(&target).unwrap(); + let account = player + .ensure_account(&mut channel, &realm, Some("REALPORTABLEGATE")) + .unwrap(); + let until = Instant::now() + Duration::from_secs(20); + while !relay.status().connected && Instant::now() < until { + std::thread::sleep(Duration::from_millis(100)); + } + assert!(relay.status().connected, "{:?}", relay.status()); + let allocation = player.allocate_relay(&mut channel).unwrap(); + use std::io::{Read, Write}; + let mut auth_client = + std::net::TcpStream::connect((allocation.relay_host.as_str(), allocation.auth_port)) + .unwrap(); + auth_client + .set_read_timeout(Some(Duration::from_secs(10))) + .unwrap(); + let username = account.username.to_ascii_uppercase(); + let mut challenge = vec![0, 0]; + challenge.extend_from_slice(&((30 + username.len()) as u16).to_le_bytes()); + challenge.extend_from_slice(b"WoW\0"); + challenge.extend_from_slice(&[3, 3, 5]); + challenge.extend_from_slice(&12340u16.to_le_bytes()); + challenge.extend_from_slice(b"68x\0niW\0SUne"); + challenge.extend_from_slice(&0u32.to_le_bytes()); + challenge.extend_from_slice(&[127, 0, 0, 1]); + challenge.push(username.len() as u8); + challenge.extend_from_slice(username.as_bytes()); + auth_client.write_all(&challenge).unwrap(); + let mut response = [0u8; 35]; + auth_client.read_exact(&mut response).unwrap(); + assert_eq!( + &response[..3], + &[0, 0, 0], + "Real authserver must accept the relay challenge" + ); + assert!(response[3..].iter().any(|byte| *byte != 0)); + drop(auth_client); + let mut world_client = + std::net::TcpStream::connect((allocation.relay_host.as_str(), allocation.world_port)) + .unwrap(); + world_client + .set_read_timeout(Some(Duration::from_secs(10))) + .unwrap(); + let mut response = [0u8; 44]; + world_client.read_exact(&mut response).unwrap(); + assert_eq!( + &response[2..4], + &[0xec, 0x01], + "Real worldserver authentication challenge must pass through relay" + ); + drop(world_client); + + let mut owner = Store::open_in_memory().unwrap(); + let profile = owner.default_profile().unwrap(); + let character = fixture_character(&databases, &mut owner, profile); + let mut model = owner.load_current(character).unwrap(); + model.progression.level = 50; + model.progression.xp = 0; + let hidden_items = model.items.len(); + let owner_path = temporary.path().join("player-portable"); + let owner_runtime = + Arc::new(PortableRuntime::start(&owner_path, Box::new(Vec::new), None).unwrap()); + owner_runtime + .call(move |service| { + let profile = service.owner_profile(); + service + .owner_mut() + .host_install_copy(profile, &model, 1, "test-owner") + .unwrap(); + }) + .unwrap(); + let public_realm = realm.to_string(); + let (bundle, session, revision) = owner_runtime + .call(move |service| { + service.prepare_remote_transfer(&character.to_string(), &public_realm) + }) + .unwrap() + .unwrap(); + let transferred = player + .transfer_owned_character( + &mut channel, + session, + character.as_uuid(), + "coa", + revision, + &bundle.payload, + bundle.content_hash, + BTreeMap::new(), + ) + .unwrap(); + assert_eq!(transferred.session_id, session); + assert!(transferred.local_guid > 0); + assert_eq!( + number( + &databases.b, + &format!( + "SELECT level FROM acore_characters.characters WHERE guid = {}", + transferred.local_guid + ) + ), + 50 + ); + let retry = player + .transfer_owned_character( + &mut channel, + session, + character.as_uuid(), + "coa", + revision, + &bundle.payload, + bundle.content_hash, + BTreeMap::new(), + ) + .unwrap(); + assert_eq!( + (retry.local_guid, retry.session_id), + (transferred.local_guid, transferred.session_id) + ); + let pending = player.pending_remote_sessions().unwrap().pop().unwrap(); + let mut console = server.ra(); + console + .run(&format!("botcmd spawnbot {}", transferred.local_guid)) + .unwrap(); + let deadline = Instant::now() + Duration::from_secs(90); + let sid = crate::portable::SessionId::from_uuid(session).unwrap(); + let mut playing = false; + while Instant::now() < deadline { + player + .sync_remote_session(&pending, &owner_runtime) + .unwrap(); + let opened = owner_runtime + .call(move |service| { + service + .owner() + .owner_session_info(sid) + .unwrap() + .is_some_and(|row| row.0 == "open") + }) + .unwrap(); + if opened + && crate::portable::session::live_session::online( + &databases.b, + transferred.local_guid, + ) + && console + .run(&format!("portable status {}", transferred.local_guid)) + .unwrap_or_default() + .contains("gated 0") + { + playing = true; + break; + } + std::thread::sleep(Duration::from_millis(500)); + } + assert!( + playing, + "The actual player never received the Owner's durable baseline acknowledgement" + ); + let name = + crate::portable::session::live_session::name_of(&databases.b, transferred.local_guid); + console.run(&format!("character level {name} 51")).unwrap(); + // Reopen both Player databases before receiving progress. Recovery uses the pinned realm key and durable session. + owner_runtime.shutdown(); + drop(owner_runtime); + drop(player); + let owner_runtime = + Arc::new(PortableRuntime::start(&owner_path, Box::new(Vec::new), None).unwrap()); + let player = Arc::new( + PlayerControl::open(&temporary.path().join("player"), player_secrets).unwrap(), + ); + let pending = player.pending_remote_sessions().unwrap().pop().unwrap(); + let deadline = Instant::now() + Duration::from_secs(90); + let mut saved = false; + while Instant::now() < deadline { + player + .sync_remote_session(&pending, &owner_runtime) + .unwrap(); + saved = owner_runtime + .call(move |service| { + service + .owner() + .load_current(character) + .unwrap() + .progression + .level + == 51 + }) + .unwrap(); + if saved { + break; + } + std::thread::sleep(Duration::from_millis(500)); + } + assert!(saved, "Real in-game progress was not saved through the public coordinator after Manager restart"); + assert!(crate::portable::session::live_session::online( + &databases.b, + transferred.local_guid + )); + console + .run(&format!("botcmd despawn {}", transferred.local_guid)) + .unwrap(); + let deadline = Instant::now() + Duration::from_secs(90); + let mut completed = false; + while Instant::now() < deadline { + if player + .sync_remote_session(&pending, &owner_runtime) + .unwrap() + { + completed = true; + break; + } + std::thread::sleep(Duration::from_millis(500)); + } + assert!( + completed, + "The final progress acknowledgement never completed" + ); + let next = player.pending_remote_sessions().unwrap().pop().unwrap(); + assert_ne!( + next.session_id, session, + "The next login must remain monitored" + ); + assert!(owner_runtime + .call(move |service| service.remote_session_idle(next.session_id)) + .unwrap() + .unwrap()); + // A second login without pressing Play is still supervised and saved. + PlayerControl::start_remote_session_sync(&player, &owner_runtime); + console + .run(&format!("botcmd spawnbot {}", transferred.local_guid)) + .unwrap(); + let deadline = Instant::now() + Duration::from_secs(90); + let next_sid = crate::portable::SessionId::from_uuid(next.session_id).unwrap(); + let mut opened = false; + while Instant::now() < deadline { + opened = owner_runtime + .call(move |service| { + service + .owner() + .owner_session_info(next_sid) + .unwrap() + .is_some_and(|row| row.0 == "open") + }) + .unwrap(); + if opened { + break; + } + std::thread::sleep(Duration::from_millis(500)); + } + assert!( + opened + && crate::portable::session::live_session::online( + &databases.b, + transferred.local_guid + ) + ); + console.run(&format!("character level {name} 52")).unwrap(); + console + .run(&format!("botcmd despawn {}", transferred.local_guid)) + .unwrap(); + let deadline = Instant::now() + Duration::from_secs(90); + let mut completed = false; + while Instant::now() < deadline { + if player + .pending_remote_sessions() + .unwrap() + .iter() + .all(|pending| pending.session_id != next.session_id) + { + completed = true; + break; + } + std::thread::sleep(Duration::from_millis(500)); + } + assert!( + completed, + "The second login's final checkpoint was not acknowledged" + ); + player + .leave_idle_remote_session(character.as_uuid(), &owner_runtime) + .unwrap(); + assert!(player.pending_remote_sessions().unwrap().is_empty()); + owner_runtime + .call(move |service| { + assert!(service.remote_session_closed(session).unwrap()); + let current = service.owner().load_current(character).unwrap(); + assert_eq!(current.progression.level, 52); + assert!( + current.items.len() >= hidden_items, + "Hidden equipment was lost" + ); + }) + .unwrap(); + owner_runtime.shutdown(); + channel.close(); + link.stop(); + relay.stop(); + runtime.shutdown(); + drop(server); } #[test] @@ -414,6 +1009,25 @@ mod tests { Arc::new(move || vec![("test1".to_string(), root_clone.clone())]), ); + assert_eq!(backend.auth_port("srv-test1:coa").unwrap(), 3724); + fs::write( + root.join("Settings/repack.json"), + r#"{"worldPort":8085,"authPort":57324}"#, + ) + .unwrap(); + assert_eq!(backend.auth_port("srv-test1:coa").unwrap(), 57324); + assert_eq!(backend.auth_port("srv-test1:wildcard").unwrap(), 57324); + fs::write( + root.join("Settings/repack.json"), + r#"{"worldPort":8085,"authPort":0}"#, + ) + .unwrap(); + assert!(backend.auth_port("srv-test1:coa").is_err()); + fs::write( + root.join("Settings/repack.json"), + r#"{"worldPort":8085,"authPort":57324}"#, + ) + .unwrap(); assert_eq!(backend.world_port("srv-test1:coa").unwrap(), 8085); assert_eq!(backend.world_port("srv-test1:wildcard").unwrap(), 8086); diff --git a/crates/coa-core/src/control/direct_route.rs b/crates/coa-core/src/control/direct_route.rs index dd3e939..e40745d 100644 --- a/crates/coa-core/src/control/direct_route.rs +++ b/crates/coa-core/src/control/direct_route.rs @@ -303,7 +303,21 @@ fn attempt_route_discovery( let local_auth = std::env::var("COA_OVERRIDE_LOCAL_AUTH_PORT") .ok() .and_then(|p| p.parse().ok()) - .unwrap_or(coa_control_proto::relay::LOCAL_AUTH_PORT); + .map(Ok) + .unwrap_or_else(|| service.local_auth_port()); + let local_auth = match local_auth { + Ok(port) => port, + Err(error) => { + set_status( + status, + RouteStatus::RelayOnly { + reason: format!("Cannot resolve local auth port: {error}"), + }, + ); + service.set_direct_route(None); + return; + } + }; let local_world = match std::env::var("COA_OVERRIDE_LOCAL_WORLD_PORT") .ok() .and_then(|p| p.parse().ok()) diff --git a/crates/coa-core/src/control/host_manager.rs b/crates/coa-core/src/control/host_manager.rs index cdd6a65..801b199 100644 --- a/crates/coa-core/src/control/host_manager.rs +++ b/crates/coa-core/src/control/host_manager.rs @@ -36,6 +36,7 @@ pub struct HostControl { struct Running { realm_id: RealmId, url: String, + ports: Option<(u16, u16)>, link: HostLink, #[allow(dead_code)] relay_link: Option>, @@ -84,7 +85,7 @@ impl HostControl { running.retain(|id, r| { wanted .get(id) - .is_some_and(|(realm, url, _, _)| *realm == r.realm_id && *url == r.url) + .is_some_and(|(realm, url, _, _)| *realm == r.realm_id && *url == r.url && backend.auth_port(id).ok().zip(backend.world_port(id).ok()) == r.ports) }); for (local_id, (realm_id, url, base, key)) in wanted { if running.contains_key(&local_id) { @@ -96,15 +97,17 @@ impl HostControl { store.clone(), backend.clone(), )); - let relay_link = match service.local_world_port() { - Ok(world_port) => { + let ports = service.local_auth_port().ok().zip(service.local_world_port().ok()); + let relay_link = match ports { + Some((auth_port, world_port)) => { crate::control::transport::relay_url(&base, "/relay/v1/host").map( |r_url| { let rl = - crate::control::relay_link::RelayLink::start_with_ports( + crate::control::relay_link::RelayLink::start_with_local_ports( r_url, realm_id, key.clone(), + auth_port, Some(world_port), ); service.set_relay(rl.clone()); @@ -112,8 +115,8 @@ impl HostControl { }, ) } - Err(e) => { - eprintln!("Relay link unavailable for {local_id}: cannot resolve world port: {e}"); + None => { + tracing::warn!(%local_id, "relay is waiting for valid local auth and world ports"); None } }; @@ -129,6 +132,7 @@ impl HostControl { Running { realm_id, url, + ports, link, relay_link, direct_route, diff --git a/crates/coa-core/src/control/join.rs b/crates/coa-core/src/control/join.rs index 8103fc0..c13ebf9 100644 --- a/crates/coa-core/src/control/join.rs +++ b/crates/coa-core/src/control/join.rs @@ -20,6 +20,8 @@ use super::store::{AccountKind, ControlStore}; use super::transport::{self, LinkError}; use crate::{Error, Result}; +mod portable_sync; + /// Where and as whom to reach a realm. #[derive(Clone, Debug)] pub struct RealmTarget { @@ -412,9 +414,11 @@ impl PlayerControl { } /// Transfer a portable character to a remote realm over the verified control channel (Phase 12.1). - pub fn transfer_character( + fn transfer_with_id( &self, ch: &mut PlayerChannel, + transfer_id: Uuid, + owned_session: bool, character_id: Uuid, ruleset: &str, canonical_revision: u64, @@ -422,16 +426,27 @@ impl PlayerControl { content_hash: [u8; 32], collections: BTreeMap>, ) -> R { - let transfer_id = Uuid::now_v7(); let content_hash_str = hex::encode(content_hash); - let offer = Request::TransferOffer { - transfer_id, - character_id, - ruleset: ruleset.to_string(), - canonical_revision, - content_hash: content_hash_str.clone(), - total_size: payload.len(), - collections, + let offer = if owned_session { + Request::OwnedTransferOffer { + transfer_id, + character_id, + ruleset: ruleset.to_string(), + canonical_revision, + content_hash: content_hash_str.clone(), + total_size: payload.len(), + collections, + } + } else { + Request::TransferOffer { + transfer_id, + character_id, + ruleset: ruleset.to_string(), + canonical_revision, + content_hash: content_hash_str.clone(), + total_size: payload.len(), + collections, + } }; let r = ch.request(&offer)?; let mut offset = match r { @@ -456,7 +471,21 @@ impl PlayerControl { notes, }); } - Response::Error { code, message } => return Err(ControlFail::App { code, message }), + Response::Error { code, message } => { + if owned_session + && matches!( + code, + AppError::Incompatible + | AppError::NotYours + | AppError::AlreadyClaimed + | AppError::Invalid + ) + { + self.store()? + .player_remote_session_remove(character_id, transfer_id)?; + } + return Err(ControlFail::App { code, message }); + } _ => { return Err(ControlFail::Link(LinkError::Protocol( "unexpected response to transfer offer".into(), diff --git a/crates/coa-core/src/control/join/portable_sync.rs b/crates/coa-core/src/control/join/portable_sync.rs new file mode 100644 index 0000000..64d554f --- /dev/null +++ b/crates/coa-core/src/control/join/portable_sync.rs @@ -0,0 +1,290 @@ +use super::*; +use crate::control::store::PlayerRemoteSession; +use crate::portable::service::PortableRuntime; +use base64::engine::general_purpose::STANDARD as B64; +use sha2::{Digest, Sha256}; + +fn protocol_error(message: &str) -> ControlFail { + ControlFail::Link(LinkError::Protocol(message.into())) +} + +impl PlayerControl { + /// Compatibility helper for transport tests. Interactive play uses the Owner's durable session id. + pub fn transfer_character( + &self, + channel: &mut PlayerChannel, + character: Uuid, + ruleset: &str, + revision: u64, + payload: &[u8], + hash: [u8; 32], + collections: BTreeMap>, + ) -> R { + self.transfer_with_id( + channel, + Uuid::now_v7(), + false, + character, + ruleset, + revision, + payload, + hash, + collections, + ) + } + + pub fn transfer_owned_character( + &self, + channel: &mut PlayerChannel, + session: Uuid, + character: Uuid, + ruleset: &str, + revision: u64, + payload: &[u8], + hash: [u8; 32], + collections: BTreeMap>, + ) -> R { + self.store()? + .player_remote_session_set(&PlayerRemoteSession { + character_id: character, + realm_id: channel.realm_id.to_string(), + session_id: session, + public_key: coord::encode_public_key(&channel.realm_key), + coordinator: channel.coordinator.clone(), + })?; + let result = self.transfer_with_id( + channel, + session, + true, + character, + ruleset, + revision, + payload, + hash, + collections, + ); + if let Ok(outcome) = &result { + if outcome.session_id != session { + return Err(protocol_error( + "The Host returned another portable session.", + )); + } + } + result + } + + pub fn pending_remote_sessions(&self) -> R> { + Ok(self.store()?.player_remote_sessions()?) + } + + /// An idle, already-saved realm may be left; an uncertain or open session must finish first. + pub fn leave_idle_remote_session(&self, character: Uuid, runtime: &PortableRuntime) -> R<()> { + for pending in self + .pending_remote_sessions()? + .into_iter() + .filter(|pending| pending.character_id == character) + { + let sid = pending.session_id; + let idle = runtime + .call(move |service| service.remote_session_idle(sid)) + .map_err(|error| ControlFail::Local(error.to_string()))? + .map_err(|error| ControlFail::Local(error.message))?; + if !idle { + return Err(ControlFail::Local("Progress from the previous realm is still being saved. Keep both Managers running until it finishes.".into())); + } + self.store()?.player_remote_session_remove(character, sid)?; + } + Ok(()) + } + + /// Persist the Owner's acknowledgement before acknowledging it to the Host. A lost reply is retried without making another revision. + pub fn sync_remote_session( + &self, + pending: &PlayerRemoteSession, + runtime: &PortableRuntime, + ) -> R { + let mut channel = self.remote_session_channel(pending)?; + self.sync_remote_session_on_channel(pending, runtime, &mut channel) + } + + fn remote_session_channel(&self, pending: &PlayerRemoteSession) -> R { + let realm = RealmId::parse(&pending.realm_id) + .map_err(|error| ControlFail::Local(error.to_string()))?; + let key = coord::decode_public_key(&pending.public_key) + .map_err(|error| protocol_error(&error.to_string()))?; + let target = RealmTarget { + realm_id: realm, + key, + coordinator: pending.coordinator.clone(), + }; + self.connect(&target) + } + + fn sync_remote_session_on_channel( + &self, + pending: &PlayerRemoteSession, + runtime: &PortableRuntime, + channel: &mut PlayerChannel, + ) -> R { + let character_id = pending.character_id; + let session_id = pending.session_id; + for _ in 0..16 { + let (kind, total_size, content_hash) = match channel.request(&Request::SessionPoll { + character_id, + session_id, + })? { + Response::SessionIdle { closed } => { + if closed { + let locally_closed = runtime + .call(move |service| service.remote_session_closed(session_id)) + .map_err(|error| ControlFail::Local(error.to_string()))? + .map_err(|error| ControlFail::Local(error.message))?; + if !locally_closed { + return Err(protocol_error( + "The Host closed a session before the Owner saved it.", + )); + } + let realm = pending.realm_id.clone(); + let next = runtime + .call(move |service| service.next_remote_session(character_id, &realm)) + .map_err(|error| ControlFail::Local(error.to_string()))? + .map_err(|error| ControlFail::Local(error.message))?; + if let Some(next) = next { + self.store()?.player_remote_session_advance( + character_id, + session_id, + next, + )?; + } else { + self.store()? + .player_remote_session_remove(character_id, session_id)?; + } + return Ok(true); + } + return Ok(false); + } + Response::SessionPending { + kind, + total_size, + content_hash, + } if total_size > 0 + && total_size <= app::MAX_SESSION_BYTES + && content_hash.len() == 64 => + { + (kind, total_size, content_hash) + } + Response::Error { code, message } => { + return Err(ControlFail::App { code, message }) + } + _ => return Err(protocol_error("Invalid portable session metadata.")), + }; + let mut bytes = Vec::with_capacity(total_size); + while bytes.len() < total_size { + match channel.request(&Request::SessionRead { + character_id, + session_id, + message_hash: content_hash.clone(), + offset: bytes.len(), + })? { + Response::SessionData { offset, data } if offset == bytes.len() => { + let chunk = B64 + .decode(data) + .map_err(|_| protocol_error("Invalid portable session chunk."))?; + if chunk.is_empty() || chunk.len() > total_size - bytes.len() { + return Err(protocol_error("Invalid portable session chunk bounds.")); + } + bytes.extend_from_slice(&chunk); + } + Response::Error { code, message } => { + return Err(ControlFail::App { code, message }) + } + _ => return Err(protocol_error("Invalid portable session chunk offset.")), + } + } + if hex::encode(Sha256::digest(&bytes)) != content_hash { + return Err(protocol_error("The portable session message hash changed.")); + } + let remote_realm = pending.realm_id.clone(); + let ack = runtime + .call(move |service| { + service.apply_remote_session_message( + character_id, + session_id, + &remote_realm, + kind, + &bytes, + ) + }) + .map_err(|error| ControlFail::Local(error.to_string()))? + .map_err(|error| ControlFail::Local(error.message))?; + let mut offset = 0; + while offset < ack.len() { + let end = ack.len().min(offset + 24 * 1024); + match channel.request(&Request::SessionAck { + character_id, + session_id, + message_hash: content_hash.clone(), + offset, + total_size: ack.len(), + data: B64.encode(&ack[offset..end]), + })? { + Response::SessionAcked { + received_size, + complete, + } if received_size > offset + && received_size <= ack.len() + && (!complete || received_size == ack.len()) => + { + offset = received_size + } + Response::Error { code, message } => { + return Err(ControlFail::App { code, message }) + } + _ => return Err(protocol_error("Invalid portable acknowledgement offset.")), + } + } + } + Ok(false) + } + + /// Independent of the visible page and the game process. Durable sessions resume after a Manager restart. + pub fn start_remote_session_sync(player: &Arc, runtime: &Arc) { + let player = Arc::downgrade(player); + let runtime = Arc::downgrade(runtime); + let _ = std::thread::Builder::new().name("portable-remote-sync".into()).spawn(move || { + let mut errors = std::collections::HashMap::::new(); + let mut channels = std::collections::HashMap::::new(); + loop { + { + let (Some(player), Some(runtime)) = (player.upgrade(), runtime.upgrade()) else { return }; + if let Ok(sessions) = player.pending_remote_sessions() { + let wanted: std::collections::HashSet<_> = sessions.iter().map(|session| format!("{}|{}|{}", session.realm_id, session.public_key, session.coordinator)).collect(); + channels.retain(|key, _| wanted.contains(key)); + for session in sessions { + let key = format!("{}|{}|{}", session.realm_id, session.public_key, session.coordinator); + let result = (|| { + if !channels.contains_key(&key) { + if channels.len() >= 8 { channels.clear(); } + channels.insert(key.clone(), player.remote_session_channel(&session)?); + } + player.sync_remote_session_on_channel(&session, &runtime, channels.get_mut(&key).expect("inserted")) + })(); + match result { + Ok(_) => { errors.remove(&session.character_id); }, + Err(error) => { + channels.remove(&key); + let message = error.to_string(); + if errors.get(&session.character_id) != Some(&message) { + tracing::warn!(character = %session.character_id, error = %message, "remote portable progress is pending; retrying"); + errors.insert(session.character_id, message); + } + } + } + } + } + } + std::thread::sleep(std::time::Duration::from_secs(2)); + } + }); + } +} diff --git a/crates/coa-core/src/control/player.rs b/crates/coa-core/src/control/player.rs index fcbb2aa..e8de87e 100644 --- a/crates/coa-core/src/control/player.rs +++ b/crates/coa-core/src/control/player.rs @@ -13,6 +13,9 @@ use super::identity::PlayerIdentity; use super::transport::{self, LinkError, Ws}; pub struct PlayerChannel { + pub(crate) realm_id: RealmId, + pub(crate) realm_key: VerifyingKey, + pub(crate) coordinator: String, ws: Ws, channel: Channel, conn: u32, @@ -35,7 +38,11 @@ impl PlayerChannel { Duration::from_millis(500), )?; let nonce = match transport::next_frame(&mut ws, STEP)? { - Frame::Challenge { protocol, nonce } if protocol == coord::COORDINATOR_PROTOCOL_VERSION => nonce, + Frame::Challenge { protocol, nonce } + if protocol == coord::COORDINATOR_PROTOCOL_VERSION => + { + nonce + } _ => return Err(LinkError::Protocol("a challenge was expected".into())), }; transport::send_text( @@ -67,7 +74,14 @@ impl PlayerChannel { { transport::send_binary(&mut ws, f)?; } - Ok(PlayerChannel { ws, channel, conn }) + Ok(PlayerChannel { + ws, + channel, + conn, + realm_id: *realm, + realm_key: *realm_key, + coordinator: coordinator.into(), + }) } pub fn connection(&self) -> u32 { diff --git a/crates/coa-core/src/control/relay_link.rs b/crates/coa-core/src/control/relay_link.rs index 7ba366f..df6aa58 100644 --- a/crates/coa-core/src/control/relay_link.rs +++ b/crates/coa-core/src/control/relay_link.rs @@ -1,6 +1,6 @@ //! Outbound Host <-> Game Relay tunnel link (Phase 13). //! Maintains outbound connectivity from Host Manager to the Game Relay, -//! opening local TCP connections strictly to 127.0.0.1:3724 (Auth) and 127.0.0.1:8085 (World). +//! opening local TCP connections only to the installed realm's configured loopback Auth and World ports. use std::collections::HashMap; use std::io::{Read, Write}; @@ -79,6 +79,22 @@ impl RelayLink { realm_id: RealmId, key: SigningKey, local_world_port: Option, + ) -> Arc { + Self::start_with_local_ports( + url, + realm_id, + key, + RelayTarget::Auth.local_port(), + local_world_port, + ) + } + + pub fn start_with_local_ports( + url: String, + realm_id: RealmId, + key: SigningKey, + local_auth_port: u16, + local_world_port: Option, ) -> Arc { let stop = Arc::new(AtomicBool::new(false)); let status = Arc::new(Mutex::new(RelayStatus::default())); @@ -88,7 +104,18 @@ impl RelayLink { let st = status.clone(); let join = std::thread::Builder::new() .name("control-relay-host".into()) - .spawn(move || run(&url, realm_id, &key, alloc_rx, &s, &st, local_world_port)) + .spawn(move || { + run( + &url, + realm_id, + &key, + alloc_rx, + &s, + &st, + local_auth_port, + local_world_port, + ) + }) .ok(); Arc::new(RelayLink { @@ -136,12 +163,22 @@ fn run( )>, stop: &AtomicBool, status: &Mutex, + local_auth_port: u16, local_world_port: Option, ) { let mut failures = 0u32; while !stop.load(Ordering::SeqCst) { let began = Instant::now(); - let result = session(url, realm, key, &alloc_rx, stop, status, local_world_port); + let result = session( + url, + realm, + key, + &alloc_rx, + stop, + status, + local_auth_port, + local_world_port, + ); if let Err(error) = &result { tracing::warn!(%realm, %error, "game relay connection lost; retrying"); } @@ -176,6 +213,15 @@ fn run( struct StreamWorker { tx: Sender, + reset_on_drop: bool, +} + +impl Drop for StreamWorker { + fn drop(&mut self) { + if self.reset_on_drop { + let _ = self.tx.send(TunnelMsg::Reset { stream_id: 0 }); + } + } } fn session( @@ -189,6 +235,7 @@ fn session( )>, stop: &AtomicBool, status: &Mutex, + local_auth_port: u16, local_world_port: Option, ) -> Result<(), LinkError> { let mut ws = transport::connect(url, POLL)?; @@ -257,6 +304,7 @@ fn session( let (ws_out_tx, ws_out_rx) = channel::(); let mut last_ping = Instant::now(); + let mut last_pong = Instant::now(); while !stop.load(Ordering::SeqCst) { // Drain pending allocation requests @@ -277,10 +325,14 @@ fn session( } // Keepalive + if last_pong.elapsed() > Duration::from_secs(90) { + return Err(LinkError::Timeout); + } if last_ping.elapsed() > Duration::from_secs(30) { - let _ = ws.send(Message::Text( + ws.send(Message::Text( serde_json::to_string(&TunnelMsg::Ping).unwrap().into(), - )); + )) + .map_err(|error| LinkError::Io(error.to_string()))?; last_ping = Instant::now(); } @@ -321,7 +373,7 @@ fn session( } => { // Strictly predefined local targets: Auth (3724) or World (8085 or bound realm world port) let local_port = match target { - RelayTarget::Auth => target.local_port(), + RelayTarget::Auth => local_auth_port, RelayTarget::World => match local_world_port { Some(p) => p, None => { @@ -341,7 +393,13 @@ fn session( // Spawn worker for this stream spawn_local_tcp_bridge(stream_id, tcp, stream_rx, ws_tx); - streams.insert(stream_id, StreamWorker { tx: stream_tx }); + streams.insert( + stream_id, + StreamWorker { + tx: stream_tx, + reset_on_drop: true, + }, + ); // Reply ConnectOk let ok_msg = TunnelMsg::ConnectOk { stream_id }; @@ -367,7 +425,8 @@ fn session( } } TunnelMsg::Close { stream_id } => { - if let Some(w) = streams.remove(&stream_id) { + if let Some(mut w) = streams.remove(&stream_id) { + w.reset_on_drop = false; let _ = w.tx.send(TunnelMsg::Close { stream_id }); } set_status(status, |s| s.active_streams = streams.len()); @@ -378,7 +437,9 @@ fn session( } set_status(status, |s| s.active_streams = streams.len()); } - TunnelMsg::Pong => {} + TunnelMsg::Pong => { + last_pong = Instant::now(); + } _ => {} } } @@ -460,3 +521,105 @@ fn spawn_local_tcp_bridge( }) .ok(); } + +#[cfg(test)] +mod tests { + use super::*; + use std::net::TcpListener; + use std::sync::atomic::AtomicUsize; + + #[test] + fn relay_reconnects_after_a_dropped_link_and_clears_its_error() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let accepted = Arc::new(AtomicUsize::new(0)); + let seen = accepted.clone(); + let worker = std::thread::spawn(move || { + for turn in 0..2 { + let (tcp, _) = listener.accept().unwrap(); + tcp.set_read_timeout(Some(Duration::from_secs(15))).unwrap(); + let mut ws = tungstenite::accept(tcp).unwrap(); + ws.send(Message::Text( + serde_json::to_string(&RelayChallenge { + nonce: "relay-reconnect-test".into(), + }) + .unwrap() + .into(), + )) + .unwrap(); + let _: HostHello = + serde_json::from_str(ws.read().unwrap().to_text().unwrap()).unwrap(); + ws.send(Message::Text( + serde_json::to_string(&RelayWelcome { + ok: true, + message: String::new(), + }) + .unwrap() + .into(), + )) + .unwrap(); + seen.fetch_add(1, Ordering::SeqCst); + if turn == 0 { + continue; + } + while let Ok(message) = ws.read() { + let Ok(text) = message.to_text() else { + continue; + }; + if let Ok(TunnelMsg::Allocate { request_id, .. }) = serde_json::from_str(text) { + ws.send(Message::Text( + serde_json::to_string(&TunnelMsg::AllocateOk { + request_id, + token: "test-only".into(), + relay_host: "127.0.0.1".into(), + auth_port: 41001, + world_port: 41002, + expires_at: i64::MAX, + }) + .unwrap() + .into(), + )) + .unwrap(); + break; + } + } + } + }); + let link = RelayLink::start_with_local_ports( + format!("ws://{addr}"), + RealmId::new(), + SigningKey::from_bytes(&[43; 32]), + 57324, + Some(58085), + ); + let until = Instant::now() + Duration::from_secs(20); + while accepted.load(Ordering::SeqCst) < 2 || !link.status().connected { + assert!(Instant::now() < until, "{:?}", link.status()); + std::thread::sleep(Duration::from_millis(50)); + } + assert!(link.status().last_error.is_none()); + let allocation = link.allocate(&Uuid::now_v7(), None).unwrap(); + assert_eq!( + (allocation.auth_port, allocation.world_port), + (41001, 41002) + ); + link.stop(); + worker.join().unwrap(); + } + + #[test] + fn dropping_a_tunnel_closes_its_local_tcp_stream() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let tcp = TcpStream::connect(listener.local_addr().unwrap()).unwrap(); + let (mut peer, _) = listener.accept().unwrap(); + peer.set_read_timeout(Some(Duration::from_secs(3))).unwrap(); + let (tx, rx) = channel(); + let (out, _) = channel(); + spawn_local_tcp_bridge(7, tcp, rx, out); + drop(StreamWorker { + tx, + reset_on_drop: true, + }); + assert_eq!(peer.read(&mut [0u8; 1]).unwrap(), 0); + } +} diff --git a/crates/coa-core/src/control/service.rs b/crates/coa-core/src/control/service.rs index 32ac619..fbd96c0 100644 --- a/crates/coa-core/src/control/service.rs +++ b/crates/coa-core/src/control/service.rs @@ -20,6 +20,20 @@ use uuid::Uuid; use super::store::{AccountKind, ClaimState, ControlStore, HostTransfer, TransferState}; use crate::{Error, Result}; +mod portable_sync; + +#[derive(Clone, Debug)] +pub struct RemoteSessionMessage { + pub kind: app::SessionMessageKind, + pub bytes: Vec, +} + +#[derive(Clone, Debug)] +pub struct RemoteSessionState { + pub message: Option, + pub closed: bool, +} + #[derive(Clone, Debug, PartialEq, Eq)] pub struct RealmInfo { /// Create accounts for joining players (the default). When false the realm only links accounts that exist. @@ -64,6 +78,7 @@ pub struct ClaimExport { #[derive(Clone, Debug)] pub struct RemoteImportParams { + pub session_id: Uuid, pub character_id: Uuid, pub revision: u64, pub payload: Vec, @@ -89,6 +104,30 @@ pub enum PreflightVerdict { /// The realm itself: its login database, its console and the portable engine. pub trait RealmBackend: Send + Sync { + fn session_message( + &self, + _local_id: &str, + _character: Uuid, + _session: Uuid, + _account: u32, + ) -> Result { + Err(Error::Invalid( + "Portable session synchronization is unavailable.".into(), + )) + } + fn session_ack( + &self, + _local_id: &str, + _character: Uuid, + _session: Uuid, + _account: u32, + _kind: app::SessionMessageKind, + _bytes: &[u8], + ) -> Result<()> { + Err(Error::Invalid( + "Portable session synchronization is unavailable.".into(), + )) + } fn info(&self, local_id: &str) -> Result; fn account_by_name(&self, local_id: &str, username: &str) -> Result>; fn account_by_id(&self, local_id: &str, id: u32) -> Result>; @@ -121,6 +160,9 @@ pub trait RealmBackend: Send + Sync { params: RemoteImportParams, ) -> Result; /// Local world port for this realm. Must fail closed if the realm port cannot be resolved. + fn auth_port(&self, _local_id: &str) -> Result { + Ok(coa_control_proto::relay::LOCAL_AUTH_PORT) + } fn world_port(&self, local_id: &str) -> Result; } @@ -148,7 +190,7 @@ impl Session { } const MAX_REQUESTS_PER_SESSION: u32 = 60; -const MAX_TRANSFER_CHUNKS_PER_SESSION: u32 = 128; +const MAX_TRANSFER_CHUNKS_PER_SESSION: u32 = 2048; const LINK_FAILURES: usize = 5; const LINK_WINDOW: Duration = Duration::from_secs(600); @@ -222,6 +264,10 @@ impl HostService { self.realm_id } + pub fn local_auth_port(&self) -> Result { + self.backend.auth_port(&self.local_id) + } + pub fn local_world_port(&self) -> Result { self.backend.world_port(&self.local_id) } @@ -246,7 +292,12 @@ impl HostService { public_key: &str, request: Request, ) -> Response { - if matches!(request, Request::TransferChunk { .. }) { + if matches!( + request, + Request::TransferChunk { .. } + | Request::SessionRead { .. } + | Request::SessionAck { .. } + ) { session.transfer_chunks += 1; if session.transfer_chunks > MAX_TRANSFER_CHUNKS_PER_SESSION { return err( @@ -325,6 +376,26 @@ impl HostService { content_hash, total_size, collections, + false, + ), + Request::OwnedTransferOffer { + transfer_id, + character_id, + ruleset, + canonical_revision, + content_hash, + total_size, + collections, + } => self.transfer_offer( + player, + transfer_id, + character_id, + ruleset, + canonical_revision, + content_hash, + total_size, + collections, + true, ), Request::TransferChunk { transfer_id, @@ -337,6 +408,32 @@ impl HostService { transfer_id, character_id, } => self.transfer_ack(player, transfer_id, character_id), + Request::SessionPoll { + character_id, + session_id, + } => self.session_poll(player, character_id, session_id), + Request::SessionRead { + character_id, + session_id, + message_hash, + offset, + } => self.session_read(player, character_id, session_id, &message_hash, offset), + Request::SessionAck { + character_id, + session_id, + message_hash, + offset, + total_size, + data, + } => self.session_ack_chunk( + player, + character_id, + session_id, + &message_hash, + offset, + total_size, + &data, + ), Request::AllocateRelay { player_id } => { self.allocate_relay(session, player, &player_id) } @@ -742,6 +839,7 @@ impl HostService { content_hash: String, total_size: usize, collections: BTreeMap>, + owned_session: bool, ) -> Response { let Ok(_guard) = self.write.lock() else { return err(AppError::Unavailable, "Busy."); @@ -792,6 +890,41 @@ impl HostService { } } + // An exact committed retry returns its saved result before transient presence or compatibility checks. + if let Ok(Some(existing)) = self + .store + .lock() + .map_err(|_| Error::Invalid("lock".into())) + .and_then(|s| s.host_transfer_latest(&self.local_id, &character_id)) + { + if existing.player_id != *player { + return err( + AppError::NotYours, + "This character's transfer belongs to another player.", + ); + } + if existing.state != TransferState::Receiving + && (!owned_session || existing.transfer_id == transfer_id) + && existing.player_id == *player + && existing.ruleset == ruleset + && existing.total_size == total_size + && existing.collections == collections + && existing.revision == canonical_revision + && existing.content_hash == content_hash + { + if let (Some(guid), Some(sid)) = (existing.local_guid, existing.session_id) { + return Response::TransferCommitted { + transfer_id, + character_id, + local_guid: guid, + session_id: sid, + projected_level: existing.projected_level, + notes: existing.notes, + }; + } + } + } + // 2. Active session conflict: if account is online with active characters match self.backend.presence(&self.local_id, m.account_id) { Ok(p) if p.account_online && p.online_characters > 0 => { @@ -819,30 +952,6 @@ impl HostService { _ => {} } - // 4. Idempotency / already committed check - if let Ok(Some(existing)) = self - .store - .lock() - .map_err(|_| Error::Invalid("lock".into())) - .and_then(|s| s.host_transfer_latest(&self.local_id, &character_id)) - { - if existing.state != TransferState::Receiving - && existing.revision == canonical_revision - && existing.content_hash == content_hash - { - if let (Some(guid), Some(sid)) = (existing.local_guid, existing.session_id) { - return Response::TransferCommitted { - transfer_id, - character_id, - local_guid: guid, - session_id: sid, - projected_level: existing.projected_level, - notes: existing.notes, - }; - } - } - } - // 5. Resume check for this specific transfer_id if let Ok(Some(cur)) = self .store @@ -1099,6 +1208,7 @@ impl HostService { }; let params = RemoteImportParams { + session_id: t.transfer_id, character_id: t.character_id, revision: t.revision, payload, @@ -2258,6 +2368,82 @@ mod tests { ); } + #[test] + fn committed_transfer_retry_survives_presence_and_compatibility_changes() { + let (realm, service, player, account) = claimed_setup(); + let character = Uuid::now_v7(); + let session_id = Uuid::now_v7(); + let (payload, hash) = test_snapshot(character, crate::portable::Ruleset::Coa); + let transfer = HostTransfer { + realm_local_id: service.local_id.clone(), + transfer_id: Uuid::now_v7(), + character_id: character, + ruleset: "coa".into(), + player_id: player, + revision: 7, + content_hash: hash.clone(), + total_size: payload.len(), + received_size: payload.len(), + state: TransferState::Committed, + local_guid: Some(3012), + session_id: Some(session_id), + projected_level: Some(60), + notes: vec![], + collections: BTreeMap::new(), + }; + service + .store + .lock() + .unwrap() + .host_transfer_save(&transfer) + .unwrap(); + realm.incompatible.lock().unwrap().insert(character); + realm.online.lock().unwrap().insert( + account, + Presence { + account_online: true, + online_characters: 1, + }, + ); + let offer = || Request::TransferOffer { + transfer_id: Uuid::now_v7(), + character_id: character, + ruleset: "coa".into(), + canonical_revision: 7, + content_hash: hash.clone(), + total_size: payload.len(), + collections: BTreeMap::new(), + }; + assert!( + matches!(service.handle(&mut Session::default(), &player, "K", offer()), + Response::TransferCommitted { local_guid: 3012, session_id: returned, .. } if returned == session_id) + ); + let mut different = transfer; + different.player_id = Uuid::now_v7(); + different.transfer_id = Uuid::now_v7(); + different.character_id = Uuid::now_v7(); + service + .store + .lock() + .unwrap() + .host_transfer_save(&different) + .unwrap(); + let foreign_offer = Request::TransferOffer { + transfer_id: Uuid::now_v7(), + character_id: different.character_id, + ruleset: "coa".into(), + canonical_revision: 7, + content_hash: hash.clone(), + total_size: payload.len(), + collections: BTreeMap::new(), + }; + assert!(matches!( + service.handle(&mut Session::default(), &player, "K", foreign_offer), + Response::Error { .. } + )); + assert!(realm.imported.lock().unwrap().is_empty()); + } + #[test] fn wrong_player_identity_cannot_overwrite_another_players_realm_copy() { let (_realm, s, p, _acc) = claimed_setup(); diff --git a/crates/coa-core/src/control/service/portable_sync.rs b/crates/coa-core/src/control/service/portable_sync.rs new file mode 100644 index 0000000..7f89232 --- /dev/null +++ b/crates/coa-core/src/control/service/portable_sync.rs @@ -0,0 +1,209 @@ +//! Authenticated, bounded transport for the existing Owner/Host session messages. +use super::*; +use base64::{engine::general_purpose::STANDARD as B64, Engine}; +use sha2::{Digest, Sha256}; +use std::io::Write; + +fn digest(bytes: &[u8]) -> String { + hex::encode(Sha256::digest(bytes)) +} + +impl HostService { + fn session_account(&self, player: &Uuid, character: Uuid) -> Option { + let mapping = self.mapping(player)?; + let transfer = self + .store + .lock() + .ok()? + .host_transfer_latest(&self.local_id, &character) + .ok()??; + (transfer.player_id == *player && transfer.state != TransferState::Receiving) + .then_some(mapping.account_id) + } + + pub(super) fn session_poll(&self, player: &Uuid, character: Uuid, session: Uuid) -> Response { + let Some(account) = self.session_account(player, character) else { + return err( + AppError::NotYours, + "This portable session does not belong to this player.", + ); + }; + match self + .backend + .session_message(&self.local_id, character, session, account) + { + Ok(RemoteSessionState { + message: Some(message), + .. + }) => Response::SessionPending { + kind: message.kind, + total_size: message.bytes.len(), + content_hash: digest(&message.bytes), + }, + Ok(state) => Response::SessionIdle { + closed: state.closed, + }, + Err(error) => unavailable("reading portable session", &error), + } + } + + pub(super) fn session_read( + &self, + player: &Uuid, + character: Uuid, + session: Uuid, + expected: &str, + offset: usize, + ) -> Response { + let Some(account) = self.session_account(player, character) else { + return err( + AppError::NotYours, + "This portable session does not belong to this player.", + ); + }; + let message = + match self + .backend + .session_message(&self.local_id, character, session, account) + { + Ok(state) => match state.message { + Some(message) => message, + None => { + return err( + AppError::Conflict, + "The portable session message changed; retry.", + ) + } + }, + Err(error) => return unavailable("reading portable session", &error), + }; + if digest(&message.bytes) != expected || offset > message.bytes.len() { + return err( + AppError::Conflict, + "The portable session message changed; retry.", + ); + } + let end = message.bytes.len().min(offset.saturating_add(24 * 1024)); + Response::SessionData { + offset, + data: B64.encode(&message.bytes[offset..end]), + } + } + + pub(super) fn session_ack_chunk( + &self, + player: &Uuid, + character: Uuid, + session: Uuid, + expected: &str, + offset: usize, + total_size: usize, + data: &str, + ) -> Response { + let Ok(_guard) = self.write.lock() else { + return err(AppError::Unavailable, "Busy."); + }; + let Some(account) = self.session_account(player, character) else { + return err( + AppError::NotYours, + "This portable session does not belong to this player.", + ); + }; + if expected.len() != 64 + || !expected.bytes().all(|b| b.is_ascii_hexdigit()) + || total_size == 0 + || total_size > app::MAX_SESSION_BYTES + || offset > total_size + || data.len() > 48 * 1024 + { + return err(AppError::Invalid, "Invalid acknowledgement bounds."); + } + let bytes = match B64.decode(data) { + Ok(bytes) if bytes.len() <= total_size - offset => bytes, + _ => return err(AppError::Invalid, "Invalid acknowledgement chunk."), + }; + let directory = std::env::temp_dir() + .join("coa-session-acks") + .join(digest(self.local_id.as_bytes())); + if let Err(error) = std::fs::create_dir_all(&directory) { + return unavailable("preparing session acknowledgement", &Error::from(error)); + } + let path = directory.join(format!("{character}-{session}.part")); + let state = match self + .backend + .session_message(&self.local_id, character, session, account) + { + Ok(state) => state, + Err(error) => return unavailable("reading portable session", &error), + }; + let Some(message) = state.message else { + let _ = std::fs::remove_file(&path); + return Response::SessionAcked { + received_size: total_size, + complete: true, + }; + }; + if digest(&message.bytes) != expected { + return err( + AppError::Conflict, + "The portable session message changed; retry.", + ); + } + // Bind one partial file to the precise message and acknowledgement length. + let mut header = expected.as_bytes().to_vec(); + header.extend_from_slice(&(total_size as u64).to_le_bytes()); + let write = (|| -> Result { + if offset == 0 { + let mut file = std::fs::File::create(&path)?; + file.write_all(&header)?; + } + let current = std::fs::read(&path)?; + if !current.starts_with(&header) { + return Err(Error::Invalid( + "The acknowledgement changed; restart it.".into(), + )); + } + let received = current.len() - header.len(); + if received != offset { + return Ok(received); + } + let mut file = std::fs::OpenOptions::new().append(true).open(&path)?; + file.write_all(&bytes)?; + file.sync_all()?; + Ok(received + bytes.len()) + })(); + let received = match write { + Ok(received) => received, + Err(error) => return unavailable("saving portable acknowledgement", &error), + }; + if received != total_size { + return Response::SessionAcked { + received_size: received, + complete: false, + }; + } + let complete = match std::fs::read(&path) { + Ok(bytes) => bytes, + Err(error) => { + return unavailable("reading portable acknowledgement", &Error::from(error)) + } + }; + match self.backend.session_ack( + &self.local_id, + character, + session, + account, + message.kind, + &complete[header.len()..], + ) { + Ok(()) => { + let _ = std::fs::remove_file(&path); + Response::SessionAcked { + received_size: total_size, + complete: true, + } + } + Err(error) => unavailable("applying portable acknowledgement", &error), + } + } +} diff --git a/crates/coa-core/src/control/store.rs b/crates/coa-core/src/control/store.rs index 3ca386b..3995b73 100644 --- a/crates/coa-core/src/control/store.rs +++ b/crates/coa-core/src/control/store.rs @@ -72,6 +72,13 @@ CREATE TABLE IF NOT EXISTS host_transfer( updated_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS host_transfer_char ON host_transfer(realm_local_id, character_id); +CREATE TABLE IF NOT EXISTS player_remote_session( + character_id TEXT PRIMARY KEY, + realm_id TEXT NOT NULL, + session_id TEXT NOT NULL, + public_key TEXT NOT NULL, + coordinator TEXT NOT NULL +); "; fn now() -> String { @@ -177,6 +184,15 @@ pub struct PlayerRealm { pub kind: AccountKind, } +#[derive(Clone, Debug)] +pub struct PlayerRemoteSession { + pub character_id: Uuid, + pub realm_id: String, + pub session_id: Uuid, + pub public_key: String, + pub coordinator: String, +} + pub struct ControlStore { conn: Connection, } @@ -186,6 +202,66 @@ fn uuid(s: String) -> Result { } impl ControlStore { + pub fn player_remote_sessions(&self) -> Result> { + let mut statement = self.conn.prepare("SELECT character_id, realm_id, session_id, public_key, coordinator FROM player_remote_session ORDER BY character_id").map_err(db)?; + let rows = statement + .query_map([], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + row.get::<_, String>(3)?, + row.get::<_, String>(4)?, + )) + }) + .map_err(db)?; + rows.map(|row| { + let (character, realm_id, session, public_key, coordinator) = row.map_err(db)?; + Ok(PlayerRemoteSession { + character_id: uuid(character)?, + realm_id, + session_id: uuid(session)?, + public_key, + coordinator, + }) + }) + .collect() + } + + pub fn player_remote_session_set(&self, session: &PlayerRemoteSession) -> Result<()> { + if self.player_remote_sessions()?.iter().any(|existing| { + existing.character_id == session.character_id + && (existing.session_id != session.session_id + || existing.realm_id != session.realm_id + || existing.public_key != session.public_key) + }) { + return Err(Error::Invalid( + "The character's previous remote session must finish saving first.".into(), + )); + } + self.conn.execute("INSERT INTO player_remote_session(character_id, realm_id, session_id, public_key, coordinator) VALUES(?1,?2,?3,?4,?5) ON CONFLICT(character_id) DO NOTHING", params![session.character_id.to_string(), session.realm_id, session.session_id.to_string(), session.public_key, session.coordinator]).map_err(db)?; + Ok(()) + } + + pub fn player_remote_session_advance( + &self, + character: Uuid, + previous: Uuid, + next: Uuid, + ) -> Result<()> { + self.conn.execute("UPDATE player_remote_session SET session_id=?3 WHERE character_id=?1 AND session_id=?2", params![character.to_string(), previous.to_string(), next.to_string()]).map_err(db)?; + Ok(()) + } + + pub fn player_remote_session_remove(&self, character: Uuid, session: Uuid) -> Result<()> { + self.conn + .execute( + "DELETE FROM player_remote_session WHERE character_id=?1 AND session_id=?2", + params![character.to_string(), session.to_string()], + ) + .map_err(db)?; + Ok(()) + } pub fn open(dir: &Path) -> Result { std::fs::create_dir_all(dir)?; let conn = Connection::open(dir.join("control.sqlite")).map_err(db)?; @@ -574,6 +650,43 @@ fn row_transfer(r: &rusqlite::Row<'_>) -> rusqlite::Result> mod tests { use super::*; + #[test] + fn remote_session_survives_restart_and_cannot_be_replaced() { + let dir = tempfile::tempdir().unwrap(); + let row = PlayerRemoteSession { + character_id: Uuid::now_v7(), + realm_id: "realm".into(), + session_id: Uuid::now_v7(), + public_key: "pinned-key".into(), + coordinator: "wss://example.test/coord/v1/player".into(), + }; + { + let store = ControlStore::open(dir.path()).unwrap(); + store.player_remote_session_set(&row).unwrap(); + store.player_remote_session_set(&row).unwrap(); + let mut other = row.clone(); + other.session_id = Uuid::now_v7(); + assert!(store.player_remote_session_set(&other).is_err()); + other = row.clone(); + other.public_key = "changed-key".into(); + assert!(store.player_remote_session_set(&other).is_err()); + } + let store = ControlStore::open(dir.path()).unwrap(); + let recovered = store.player_remote_sessions().unwrap().pop().unwrap(); + assert_eq!( + (recovered.session_id, recovered.public_key), + (row.session_id, row.public_key) + ); + store + .player_remote_session_remove(row.character_id, Uuid::now_v7()) + .unwrap(); + assert_eq!(store.player_remote_sessions().unwrap().len(), 1); + store + .player_remote_session_remove(row.character_id, row.session_id) + .unwrap(); + assert!(store.player_remote_sessions().unwrap().is_empty()); + } + #[test] fn a_player_owns_one_account_per_realm_and_an_account_one_player() { let s = ControlStore::open_in_memory().unwrap(); diff --git a/crates/coa-core/src/portable/service/engine.rs b/crates/coa-core/src/portable/service/engine.rs index 24febb9..517a5b2 100644 --- a/crates/coa-core/src/portable/service/engine.rs +++ b/crates/coa-core/src/portable/service/engine.rs @@ -31,6 +31,8 @@ use serde::{Deserialize, Serialize}; use crate::db::Db; use crate::ra::Ra; +mod remote_sync; + use super::super::capabilities::RealmCapabilities; use super::super::compat::{self, CompatibilityReport, Inputs, Operation, Outcome, Topic}; use super::super::error::PortableError; @@ -729,6 +731,12 @@ impl PortableService { updated_at: at, }); } + let remote_active = self + .owner + .owner_active_remote(c.character_id) + .ok() + .flatten() + .is_some(); let active = copies .iter() .find(|x| { @@ -741,9 +749,9 @@ impl PortableService { ) }) .cloned(); - let status = active - .as_ref() - .map(|a| a.status) + let status = remote_active + .then_some(PlayStatus::Syncing) + .or_else(|| active.as_ref().map(|a| a.status)) .or_else(|| { copies .iter() @@ -1318,6 +1326,11 @@ impl PortableService { if events.is_ok() { delivery = (|| { for m in host.outbox()? { + if self.owner.owner_session_realm(m.session_id)?.as_deref() + != Some(access.id.as_str()) + { + continue; // The remote Owner must acknowledge its own baseline and checkpoints. + } let mut owner = OwnerService::new(&mut self.owner); let ack: OwnerAck = if m.started { owner.handle_started(&m.bytes)? @@ -1331,6 +1344,25 @@ impl PortableService { } } for m in host.collection_outbox()? { + let local_owner = + host.store() + .host_live_sessions(&access.id)? + .iter() + .any(|session| { + session.local_guid.is_some_and(|guid| { + realm::collections::account_of(&db, guid).ok().flatten() + == Some(m.account) + }) && self + .owner + .owner_session_realm(session.session_id) + .ok() + .flatten() + .as_deref() + == Some(access.id.as_str()) + }); + if !local_owner { + continue; + } let ack = OwnerService::new(&mut self.owner).handle_collection(&m.bytes)?; host.receive_collection_ack(&mut bridge, m.account, &ack)?; } @@ -1655,6 +1687,20 @@ impl PortableService { )); } + let existing_mapping = self + .host + .server_mappings(cid)? + .into_iter() + .find(|m| m.server_id == realm_id); + + // A live realm must never arm stale data as a new canonical revision. + if existing_mapping + .as_ref() + .is_some_and(|mapping| mapping.last_revision != params.revision) + { + return Err(fail("update_required", "The realm's existing copy needs an offline refresh by its owner before this revision can be played. Your canonical character is unchanged.")); + } + // Install or advance copy in host store self.host .host_install_copy(self.host_profile, &model, params.revision, "remote")?; @@ -1677,13 +1723,7 @@ impl PortableService { } } - let existing_mapping = self - .host - .server_mappings(cid)? - .into_iter() - .find(|m| m.server_id == realm_id); - - let session_id = SessionId::new(); + let session_id = SessionId::from_uuid(params.session_id)?; let envelope = super::super::session::protocol::Envelope::from_encoded( &super::super::snapshot::encode(&model)?, ); diff --git a/crates/coa-core/src/portable/service/engine/remote_sync.rs b/crates/coa-core/src/portable/service/engine/remote_sync.rs new file mode 100644 index 0000000..1ef7bf8 --- /dev/null +++ b/crates/coa-core/src/portable/service/engine/remote_sync.rs @@ -0,0 +1,345 @@ +use super::*; +use crate::control::service::{RemoteSessionMessage, RemoteSessionState}; +use crate::portable::session::protocol::{ + self, CollectionAck, CollectionObserved, OwnerAck, PortableCheckpoint, PortableSessionStarted, +}; +use coa_control_proto::app::SessionMessageKind; + +pub fn owner_realm(realm: &str) -> String { + format!("remote-{realm}") +} + +impl PortableService { + /// Open the Owner's durable session before its snapshot leaves this computer. + pub fn prepare_remote_transfer( + &mut self, + character: &str, + realm: &str, + ) -> Res<(ClaimBundle, uuid::Uuid, u64)> { + let cid = self.cid(character)?; + let offer = OwnerService::new(&mut self.owner).offer(cid, &owner_realm(realm))?; + let bundle = self.export_for_remote_transfer(character)?; + self.publish(); + Ok((bundle, offer.session_id.as_uuid(), offer.canonical_revision)) + } + + pub fn remote_session_closed(&self, session: uuid::Uuid) -> Res { + let sid = SessionId::from_uuid(session)?; + Ok(self + .owner + .owner_session_info(sid)? + .is_some_and(|row| row.0 == "closed")) + } + + pub fn remote_session_idle(&self, session: uuid::Uuid) -> Res { + let sid = SessionId::from_uuid(session)?; + Ok(self + .owner + .owner_session_info(sid)? + .is_some_and(|row| row.0 == "offered" || row.0 == "superseded")) + } + + pub fn next_remote_session( + &self, + character: uuid::Uuid, + realm: &str, + ) -> Res> { + Ok(self + .owner + .owner_next_session(CharacterId::from_uuid(character)?, &owner_realm(realm))? + .map(|sid| sid.as_uuid())) + } + + /// The channel authenticates the public realm. Its private local id is never an Owner-side namespace. + pub fn apply_remote_session_message( + &mut self, + character: uuid::Uuid, + session: uuid::Uuid, + realm: &str, + kind: SessionMessageKind, + bytes: &[u8], + ) -> Res> { + let cid = CharacterId::from_uuid(character)?; + let sid = SessionId::from_uuid(session)?; + let scope = owner_realm(realm); + if self.owner.owner_session_realm(sid)?.as_deref() != Some(scope.as_str()) { + return Err(fail( + "not_yours", + "This remote session was not opened by this Manager.", + )); + } + let ruleset = self.owner.character(cid)?.ruleset; + let answer = match kind { + SessionMessageKind::Started => { + let mut message: PortableSessionStarted = protocol::from_json(bytes)?; + if message.character_id != cid + || message.session_id != sid + || message.ruleset != ruleset + { + return Err(fail( + "not_yours", + "The baseline belongs to another character or session.", + )); + } + message.server_id = scope; + let ack = self.owner.owner_session_started(&message)?; + protocol::to_json(&ack)? + } + SessionMessageKind::Checkpoint => { + let mut message: PortableCheckpoint = protocol::from_json(bytes)?; + if message.character_id != cid + || message.session_id != sid + || message.ruleset != ruleset + { + return Err(fail( + "not_yours", + "The checkpoint belongs to another character or session.", + )); + } + message.server_id = scope; + let ack = self.owner.owner_checkpoint(&message)?; + protocol::to_json(&ack)? + } + SessionMessageKind::Collection => { + let mut message: CollectionObserved = protocol::collection_from_json(bytes)?; + if message.ruleset != ruleset { + return Err(fail( + "incompatible", + "The collection belongs to another ruleset.", + )); + } + message.server_id = scope; + let ack = OwnerService::new(&mut self.owner) + .handle_collection(&protocol::collection_to_json(&message)?)?; + protocol::collection_to_json(&ack)? + } + }; + self.publish(); + Ok(answer) + } + + fn checked_remote_session( + &self, + realm: &str, + character: uuid::Uuid, + session: uuid::Uuid, + account: u32, + ) -> Res<(RealmAccess, crate::portable::store::HostSession)> { + let access = self.access(realm)?; + let cid = CharacterId::from_uuid(character)?; + let sid = SessionId::from_uuid(session)?; + let row = self + .host + .host_session(sid)? + .ok_or_else(|| fail("not_yours", "Unknown portable session."))?; + if row.character_id != cid || row.server_id != realm { + return Err(fail( + "not_yours", + "The portable session belongs to another realm or character.", + )); + } + let guid = row + .local_guid + .ok_or_else(|| fail("not_yours", "The portable session has no realm character."))?; + if realm::collections::account_of(&access.db()?, guid)? != Some(account) { + return Err(fail( + "not_yours", + "The portable session belongs to another account.", + )); + } + Ok((access, row)) + } + + pub fn remote_session_message( + &mut self, + realm: &str, + character: uuid::Uuid, + session: uuid::Uuid, + account: u32, + ) -> Res { + let (_, row) = self.checked_remote_session(realm, character, session, account)?; + let host = HostService::new(&mut self.host, realm, self.host_config.clone()); + let collection = host + .collection_outbox()? + .into_iter() + .find(|message| message.account == account); + let message = if let Some(message) = collection { + Some(RemoteSessionMessage { + kind: SessionMessageKind::Collection, + bytes: message.bytes, + }) + } else { + host.outbox()? + .into_iter() + .find(|message| message.session_id == row.session_id) + .map(|message| RemoteSessionMessage { + kind: if message.started { + SessionMessageKind::Started + } else { + SessionMessageKind::Checkpoint + }, + bytes: message.bytes, + }) + }; + Ok(RemoteSessionState { + message, + closed: row.state == crate::portable::store::HostState::Closed, + }) + } + + pub fn apply_remote_session_ack( + &mut self, + realm: &str, + character: uuid::Uuid, + session: uuid::Uuid, + account: u32, + kind: SessionMessageKind, + bytes: &[u8], + ) -> Res<()> { + let (access, row) = self.checked_remote_session(realm, character, session, account)?; + let db = access.db()?; + let knowledge = RealmKnowledge::from_data_dir(&access.data_dir) + .ok() + .map(Arc::new); + let mut bridge = LiveBridge::new( + &db, + access + .ra() + .map_err(|error| fail("realm_offline", error.to_string()))?, + ) + .with_knowledge(knowledge); + let mut host = HostService::new(&mut self.host, realm, self.host_config.clone()); + match kind { + SessionMessageKind::Collection => { + let ack: CollectionAck = protocol::collection_from_json(bytes)?; + host.receive_collection_ack(&mut bridge, account, &ack)?; + } + _ => { + let ack: OwnerAck = protocol::from_json(bytes)?; + if ack.session_id != row.session_id { + return Err(fail( + "not_yours", + "The acknowledgement belongs to another session.", + )); + } + host.receive_ack(&mut bridge, &ack)?; + } + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::portable::{ + fixtures::geared_level_eighty, + session::protocol::{AckOutcome, PortableCheckpoint, PortableSessionStarted}, + }; + + #[test] + fn remote_progress_is_scoped_durable_and_idempotent() { + let dir = tempfile::tempdir().unwrap(); + let mut service = PortableService::open(dir.path(), vec![]).unwrap(); + let model = geared_level_eighty(); + let cid = model.character_id; + service + .owner + .host_install_copy(service.owner_profile, &model, 7, "fixture") + .unwrap(); + let (_, sid, revision) = service + .prepare_remote_transfer(&cid.to_string(), "public-realm") + .unwrap(); + assert_eq!(revision, 7); + let session = SessionId::from_uuid(sid).unwrap(); + let started = + PortableSessionStarted::new(session, cid, "private-host-id", revision, 1, &model, None) + .unwrap(); + let bytes = protocol::to_json(&started).unwrap(); + assert!(service + .apply_remote_session_message( + cid.as_uuid(), + sid, + "other-realm", + SessionMessageKind::Started, + &bytes + ) + .is_err()); + assert!(service + .apply_remote_session_message( + uuid::Uuid::now_v7(), + sid, + "public-realm", + SessionMessageKind::Started, + &bytes + ) + .is_err()); + let ack: OwnerAck = protocol::from_json( + &service + .apply_remote_session_message( + cid.as_uuid(), + sid, + "public-realm", + SessionMessageKind::Started, + &bytes, + ) + .unwrap(), + ) + .unwrap(); + assert_eq!(ack.outcome, AckOutcome::Applied); + assert_eq!(service.state().characters[0].status, PlayStatus::Syncing); + assert!(service + .prepare_remote_transfer(&cid.to_string(), "other-realm") + .is_err()); + drop(service); + let mut service = PortableService::open(dir.path(), vec![]).unwrap(); + let mut changed = model.clone(); + changed.progression.money += 123; + let checkpoint = PortableCheckpoint::new( + session, + cid, + "private-host-id", + revision, + 1, + true, + &changed, + None, + ) + .unwrap(); + let bytes = protocol::to_json(&checkpoint).unwrap(); + let first: OwnerAck = protocol::from_json( + &service + .apply_remote_session_message( + cid.as_uuid(), + sid, + "public-realm", + SessionMessageKind::Checkpoint, + &bytes, + ) + .unwrap(), + ) + .unwrap(); + assert_eq!(first.outcome, AckOutcome::Applied); + let again: OwnerAck = protocol::from_json( + &service + .apply_remote_session_message( + cid.as_uuid(), + sid, + "public-realm", + SessionMessageKind::Checkpoint, + &bytes, + ) + .unwrap(), + ) + .unwrap(); + assert_eq!(again.outcome, AckOutcome::Duplicate); + assert_eq!(first.canonical_revision, again.canonical_revision); + assert!(service.remote_session_closed(sid).unwrap()); + drop(service); + let service = PortableService::open(dir.path(), vec![]).unwrap(); + assert_eq!( + service.owner.load_current(cid).unwrap().progression.money, + changed.progression.money + ); + } +} diff --git a/crates/coa-core/src/portable/store.rs b/crates/coa-core/src/portable/store.rs index 09acf6e..d114e96 100644 --- a/crates/coa-core/src/portable/store.rs +++ b/crates/coa-core/src/portable/store.rs @@ -177,7 +177,7 @@ impl Store { fn init(conn: Connection) -> Result { conn.busy_timeout(std::time::Duration::from_secs(5))?; let _ = conn.pragma_update(None, "journal_mode", "WAL"); - let _ = conn.pragma_update(None, "synchronous", "NORMAL"); + conn.pragma_update(None, "synchronous", "FULL")?; let mut store = Store { conn }; store.migrate()?; store.conn.pragma_update(None, "foreign_keys", true)?; diff --git a/crates/coa-core/src/portable/store/owner.rs b/crates/coa-core/src/portable/store/owner.rs index 7e06862..f257913 100644 --- a/crates/coa-core/src/portable/store/owner.rs +++ b/crates/coa-core/src/portable/store/owner.rs @@ -181,6 +181,10 @@ impl Store { check_server_id(server_id)?; let tx = self.write_tx()?; let record = read_character(&tx, id)?; + let busy: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM owner_session WHERE character_id = ?1 AND server_id != ?2 AND state = 'open')", params![id.to_string(), server_id], |row| row.get(0))?; + if busy { + return Err(PortableError::SessionOpen); + } let live: Option = tx .query_row("SELECT session_id FROM owner_session WHERE character_id = ?1 AND server_id = ?2 AND state IN ('offered', 'open')", params![id.to_string(), server_id], |r| r.get(0)) .optional()?; @@ -194,6 +198,8 @@ impl Store { } tx.execute("UPDATE owner_session SET state = 'superseded', updated_at = ?2 WHERE session_id = ?1", params![live, now()])?; } + // Only the selected realm may start the next session. Other idle offers must not come alive later. + tx.execute("UPDATE owner_session SET state='superseded', updated_at=?3 WHERE character_id=?1 AND server_id!=?2 AND state='offered'", params![id.to_string(), server_id, now()])?; let offer = open_offer(&tx, id, server_id)?; tx.commit()?; Ok(offer) @@ -521,4 +527,21 @@ impl Store { Ok(read_owner_row(&self.conn, session)? .map(|r| (r.state, r.last_sequence, r.head_revision))) } + + pub fn owner_next_session( + &self, + character: CharacterId, + realm: &str, + ) -> Result> { + let sid: Option = self.conn.query_row("SELECT session_id FROM owner_session WHERE character_id=?1 AND server_id=?2 AND state IN ('offered','open') ORDER BY updated_at DESC LIMIT 1", params![character.to_string(), realm], |row| row.get(0)).optional()?; + sid.map(|sid| sid.parse()).transpose() + } + + pub fn owner_active_remote(&self, character: CharacterId) -> Result> { + self.conn.query_row("SELECT server_id, state FROM owner_session WHERE character_id=?1 AND server_id LIKE 'remote-%' AND state='open' ORDER BY updated_at DESC LIMIT 1", [character.to_string()], |row| Ok((row.get(0)?, row.get(1)?))).optional().map_err(Into::into) + } + + pub fn owner_session_realm(&self, session: SessionId) -> Result> { + Ok(read_owner_row(&self.conn, session)?.map(|row| row.server_id)) + } } diff --git a/crates/coa-core/src/portable/store/tests.rs b/crates/coa-core/src/portable/store/tests.rs index 6d54c46..bfee39d 100644 --- a/crates/coa-core/src/portable/store/tests.rs +++ b/crates/coa-core/src/portable/store/tests.rs @@ -43,7 +43,14 @@ fn schema_nine_upgrade_preserves_characters_and_isolates_collections() { assert_eq!(store.schema_version().unwrap(), 10); assert_eq!(store.load_current(id).unwrap(), before); assert_eq!(store.server_mappings(id).unwrap()[0].local_guid, 42); - assert_eq!(store.collection(profile, Ruleset::Coa, kind).unwrap().unwrap().1, set); + assert_eq!( + store + .collection(profile, Ruleset::Coa, kind) + .unwrap() + .unwrap() + .1, + set + ); assert_eq!( store .collection_info(profile, Ruleset::Coa, kind) @@ -61,7 +68,11 @@ fn schema_nine_upgrade_preserves_characters_and_isolates_collections() { .merge_collection(profile, Ruleset::Coa, kind, &IdSet::from_ids([30]).unwrap()) .unwrap(); assert_eq!( - store.collection(profile, Ruleset::Coa, kind).unwrap().unwrap().1, + store + .collection(profile, Ruleset::Coa, kind) + .unwrap() + .unwrap() + .1, IdSet::from_ids([10, 20, 30]).unwrap() ); let foreign: (i64, Vec) = store @@ -2578,3 +2589,16 @@ fn reborn_store_corruption_fails_closed() { "Loading reborn summary with Wildcard snapshot must fail closed" ); } + +#[test] +fn canonical_file_retains_full_sync_after_initialization_and_reopen() { + let dir = tempfile::tempdir().unwrap(); + for _ in 0..2 { + let store = Store::open(dir.path()).unwrap(); + let sync: i64 = store + .conn + .pragma_query_value(None, "synchronous", |row| row.get(0)) + .unwrap(); + assert_eq!(sync, 2); + } +} diff --git a/docs/PORTABLE-CHARACTERS.md b/docs/PORTABLE-CHARACTERS.md index 8e5670e..4c1fc1a 100644 --- a/docs/PORTABLE-CHARACTERS.md +++ b/docs/PORTABLE-CHARACTERS.md @@ -58,3 +58,5 @@ Use **Saved versions** to see when progress was saved. If the Manager reports a ## Portable access and relay In the realm's Public list card, enable “Allow portable characters on this realm” and save. It is off until the owner opts in. Turning it off blocks new exports and imports; it does not delete saved characters. Keep the host Manager running. The control connection and game relay have separate status indicators: a connected control channel does not establish a game tunnel. Characters stay within their original ruleset (CoA, Wildcard or Warcraft Reborn); changing servers does not convert their ruleset. + +Keep both Managers running. Saving resumes after reconnecting or restarting, and later logins stay monitored. Wait for saving before switching realms. Your installed servers use local preparation and recovery. Experimental limitation: an older existing copy on another host cannot yet be refreshed automatically while its worldserver runs. Returning may require host maintenance. The stale copy is blocked and your canonical character stays saved. Do not delete the character or reinstall the server to bypass this message. diff --git a/docs/de/PORTABLE-CHARACTERS.md b/docs/de/PORTABLE-CHARACTERS.md index 359164e..53dd519 100644 --- a/docs/de/PORTABLE-CHARACTERS.md +++ b/docs/de/PORTABLE-CHARACTERS.md @@ -49,3 +49,5 @@ Unter **Gespeicherte Versionen (Saved versions)** siehst du, wann dein Fortschri ## Portable Charaktere und Relay Aktiviere in der Karte zur öffentlichen Realmliste „Portable Charaktere auf diesem Realm erlauben“ und speichere. Ohne Zustimmung des Betreibers bleibt der Zugang gesperrt. Ausschalten verhindert neue Exporte und Importe, löscht aber keine gespeicherten Charaktere. Der Manager des Betreibers muss laufen. Steuerverbindung und Spiel-Relay haben getrennte Statusanzeigen; eine Steuerverbindung allein ist noch kein Spieltunnel. Charaktere bleiben in ihrem ursprünglichen Spielmodus: CoA, Wildcard oder Warcraft Reborn. + +Lass beide Manager laufen. Nach Verbindungsabbrüchen oder Neustarts wird ausstehender Fortschritt weiter gespeichert; auch spätere Anmeldungen werden überwacht. Warte vor dem Realmwechsel auf den Abschluss. Eigene installierte Server nutzen die lokale Vorbereitung und Wiederherstellung. Experimentelle Einschränkung: Eine ältere Kopie bei einem anderen Host lässt sich auf dessen laufendem Worldserver noch nicht automatisch aktualisieren. Die Rückkehr kann Wartungsarbeiten des Hosts erfordern. Die alte Kopie wird blockiert, dein kanonischer Charakter bleibt gespeichert. Lösche den Charakter nicht und installiere den Server nicht neu, um die Meldung zu umgehen. diff --git a/docs/es/PORTABLE-CHARACTERS.md b/docs/es/PORTABLE-CHARACTERS.md index e644fac..ff78391 100644 --- a/docs/es/PORTABLE-CHARACTERS.md +++ b/docs/es/PORTABLE-CHARACTERS.md @@ -49,3 +49,5 @@ En **Versiones guardadas (Saved versions)** puedes ver cuándo se guardó el pro ## Personajes portátiles y relay En la tarjeta de publicación del reino, activa «Permitir personajes portátiles en este reino» y guarda. El acceso está desactivado por defecto. Desactivarlo impide nuevas exportaciones e importaciones sin borrar personajes guardados. El Manager del anfitrión debe permanecer abierto. La conexión de control y el relay del juego muestran estados separados: un canal de control conectado no garantiza un túnel de juego. Los personajes conservan su modo original: CoA, Wildcard o Warcraft Reborn. + +Mantén abiertos ambos Managers. Tras una desconexión o reinicio, el progreso pendiente vuelve a guardarse; también se supervisan los siguientes inicios de sesión. Espera a que termine antes de cambiar de reino. Tus servidores instalados usan la preparación y recuperación locales. Limitación experimental: una copia antigua en otro anfitrión todavía no se actualiza automáticamente mientras su worldserver está en ejecución. Volver puede requerir mantenimiento del anfitrión. Esa copia se bloquea y tu personaje canónico queda guardado. No borres el personaje ni reinstales el servidor para evitar este mensaje. diff --git a/docs/fr/PORTABLE-CHARACTERS.md b/docs/fr/PORTABLE-CHARACTERS.md index 73fdbb6..661b686 100644 --- a/docs/fr/PORTABLE-CHARACTERS.md +++ b/docs/fr/PORTABLE-CHARACTERS.md @@ -49,3 +49,5 @@ Sur un royaume dont le niveau maximal est inférieur, votre personnage peut temp ## Personnages portables et relais Dans la carte de publication du royaume, activez « Autoriser les personnages portables sur ce royaume », puis enregistrez. Cet accès est désactivé par défaut. Le désactiver bloque les nouveaux exports et imports sans supprimer les personnages sauvegardés. Le Manager de l’hôte doit rester ouvert. La connexion de contrôle et le relais de jeu ont des états distincts : un canal de contrôle connecté ne suffit pas à établir un tunnel de jeu. Chaque personnage conserve son mode d’origine : CoA, Wildcard ou Warcraft Reborn. + +Gardez les deux Managers ouverts. Après une coupure ou un redémarrage, la sauvegarde reprend ; les connexions suivantes restent suivies. Attendez la fin avant de changer de royaume. Vos serveurs installés utilisent la préparation et la récupération locales. Limite expérimentale : une ancienne copie chez un autre hôte ne peut pas encore être actualisée automatiquement sur son worldserver en cours d’exécution. Le retour peut nécessiter une maintenance par l’hôte. Cette copie est bloquée et votre personnage canonique reste enregistré. Ne supprimez pas le personnage et ne réinstallez pas le serveur pour contourner ce message. diff --git a/docs/ru/PORTABLE-CHARACTERS.md b/docs/ru/PORTABLE-CHARACTERS.md index 5ef8302..0132f20 100644 --- a/docs/ru/PORTABLE-CHARACTERS.md +++ b/docs/ru/PORTABLE-CHARACTERS.md @@ -49,3 +49,5 @@ ## Доступ переносимых персонажей и relay В блоке «Публичный список» включите «Разрешить переносимых персонажей на этом реалме» и сохраните. По умолчанию доступ выключен. Отключение запрещает новые экспорты и импорты, но не удаляет сохранённых персонажей. Менеджер владельца должен оставаться запущенным. Подключение к службе управления и игровой relay имеют отдельные статусы: управляющий канал ещё не означает наличие игрового туннеля. Персонажи остаются в своём режиме (CoA, Wildcard или Warcraft Reborn); смена сервера не преобразует режим персонажа. + +Во время игры должны работать оба менеджера. После обрыва связи или перезапуска сохранение продолжается автоматически; повторные входы тоже отслеживаются. Перед сменой реалма дождитесь сохранения. Ваши установленные серверы используют локальную подготовку и восстановление. Ограничение экспериментальной функции: устаревшая копия у другого владельца пока не обновляется автоматически при работающем worldserver. Для возвращения может потребоваться обслуживание владельцем. Устаревшая копия блокируется, основной персонаж остаётся сохранённым. Не удаляйте персонажа и не переустанавливайте сервер для обхода сообщения. diff --git a/docs/zh-CN/PORTABLE-CHARACTERS.md b/docs/zh-CN/PORTABLE-CHARACTERS.md index e1b939d..03615b3 100644 --- a/docs/zh-CN/PORTABLE-CHARACTERS.md +++ b/docs/zh-CN/PORTABLE-CHARACTERS.md @@ -49,3 +49,5 @@ ## 便携角色访问与中继 在服务器的公开列表设置中，勾选“允许便携角色进入此服务器”并保存。默认不允许访问。关闭后会阻止新的导出和导入，但不会删除已保存的角色。主机端 Manager 必须保持运行。控制连接与游戏中继分别显示状态；控制连接成功不代表游戏隧道已经建立。角色始终保留原来的游戏模式（CoA、Wildcard 或 Warcraft Reborn），更换服务器不会转换角色模式。 + +请保持两个管理器运行。断线或重启后，待保存的进度会继续同步，再次登录也会继续跟踪。请等待保存完成后再切换服务器。自己安装的服务器使用本地准备和恢复流程。实验功能限制：其他主机上的旧角色副本目前无法在其 worldserver 运行时自动更新。返回该服务器可能需要主人进行维护。旧副本会被阻止，你的主角色仍然保存在管理器中。请勿通过删除角色或重装服务器绕过提示。 diff --git a/services/coa-control-e2e/tests/live_control.rs b/services/coa-control-e2e/tests/live_control.rs index cc8c18e..798ac73 100644 --- a/services/coa-control-e2e/tests/live_control.rs +++ b/services/coa-control-e2e/tests/live_control.rs @@ -28,7 +28,10 @@ fn transfer_snapshot(id: Uuid, money: u32) -> (Vec, [u8; 32]) { } fn now() -> i64 { - std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs() as i64 + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs() as i64 } fn random_key() -> SigningKey { @@ -45,26 +48,61 @@ fn base_url() -> Option { fn sample_caps(max_level: u32) -> AdvertisedCapabilities { let content = ContentProfile { ruleset: Ruleset::Coa, - character_formats: CharacterFormats { readable: FormatRange { min: 1, max: 2 }, writable: FormatRange { min: 2, max: 2 } }, + character_formats: CharacterFormats { + readable: FormatRange { min: 1, max: 2 }, + writable: FormatRange { min: 2, max: 2 }, + }, online_import_job_formats: vec![2], session_protocol: 2, collection_protocol: 1, - features: [Feature::RuntimeSessions, Feature::Wardrobe, Feature::LevelProjection].into_iter().collect(), + features: [ + Feature::RuntimeSessions, + Feature::Wardrobe, + Feature::LevelProjection, + ] + .into_iter() + .collect(), collection_kinds: ["coa:appearance".to_string()].into_iter().collect(), - extensions: vec![ExtensionSupport { namespace: "coa".into(), module_version: "1".into(), formats: FormatRange { min: 1, max: 1 } }], - client_catalog: [("Appearances.dbc".to_string(), CatalogEntry { sha256: "a".repeat(64), records: 5 })].into_iter().collect(), + extensions: vec![ExtensionSupport { + namespace: "coa".into(), + module_version: "1".into(), + formats: FormatRange { min: 1, max: 1 }, + }], + client_catalog: [( + "Appearances.dbc".to_string(), + CatalogEntry { + sha256: "a".repeat(64), + records: 5, + }, + )] + .into_iter() + .collect(), }; let content_profile_hash = content.hash(); AdvertisedCapabilities { profile_version: 2, - core: Some(CoreIdentity { commit: "abc123".into(), branch: "main".into(), date: "2026-10-01".into() }), + core: Some(CoreIdentity { + commit: "abc123".into(), + branch: "main".into(), + date: "2026-10-01".into(), + }), content, content_profile_hash, - progression: Some(Progression { max_player_level: max_level, projection_protocol: 1, projection_policy_version: 1, progression_signature: "b".repeat(64), scaling_enabled: false }), + progression: Some(Progression { + max_player_level: max_level, + projection_protocol: 1, + projection_policy_version: 1, + progression_signature: "b".repeat(64), + scaling_enabled: false, + }), } } -fn register_realm(client: &RegistryClient, realm: &RealmId, key: &SigningKey) -> std::result::Result<(), Box> { +fn register_realm( + client: &RegistryClient, + realm: &RealmId, + key: &SigningKey, +) -> std::result::Result<(), Box> { let listing = Listing { display_name: "Live Control Gate".into(), description: "TLS Phase 12 gate".into(), @@ -72,7 +110,10 @@ fn register_realm(client: &RegistryClient, realm: &RealmId, key: &SigningKey) -> region: Some("EU".into()), rates: Rates::default(), modules: vec![], - account_provisioning: AccountProvisioning { automatic: true, existing_only: false }, + account_provisioning: AccountProvisioning { + automatic: true, + existing_only: false, + }, manager_version: "0.7.0".into(), }; let capabilities = sample_caps(60); @@ -84,7 +125,11 @@ fn register_realm(client: &RegistryClient, realm: &RealmId, key: &SigningKey) -> listing, capabilities_hash: capabilities.advert_hash(), capabilities, - population: Population { players: 0, bots: 0, capacity: None }, + population: Population { + players: 0, + bots: 0, + capacity: None, + }, }; let ts = now(); client.register(key, &req, ts)?; @@ -116,23 +161,40 @@ fn gate_phase12_live_vps_control_channel() { let service = Arc::new(HostService::new("srv-gate", realm, store, fake.clone())); let host_ws_url = coordinator_url(&base, "/coord/v1/host").expect("coordinator host url"); - assert!(host_ws_url.starts_with("wss://"), "Coordinator URL must use WSS: {host_ws_url}"); + assert!( + host_ws_url.starts_with("wss://"), + "Coordinator URL must use WSS: {host_ws_url}" + ); - let mut link = HostLink::start(host_ws_url.clone(), realm, host_key.clone(), service, Arc::new(now)); + let mut link = HostLink::start( + host_ws_url.clone(), + realm, + host_key.clone(), + service, + Arc::new(now), + ); // Wait for Host to connect to Coordinator let deadline = Instant::now() + Duration::from_secs(15); while Instant::now() < deadline && !link.status().connected { std::thread::sleep(Duration::from_millis(200)); } - assert!(link.status().connected, "Host failed to connect to Coordinator via WSS at {host_ws_url}: {:?}", link.status()); + assert!( + link.status().connected, + "Host failed to connect to Coordinator via WSS at {host_ws_url}: {:?}", + link.status() + ); println!("GATE 12 PASS Coordinator Host connection established over WSS: {host_ws_url}"); // 3. Player connects to Host via Coordinator over WSS let player_dir = tempfile::tempdir().unwrap(); let pc = PlayerControl::open(player_dir.path(), Arc::new(MemoryStore::default())).unwrap(); - let player_ws_base = coordinator_url(&base, "/coord/v1/player").expect("coordinator player url"); - assert!(player_ws_base.starts_with("wss://"), "Player Coordinator URL must use WSS: {player_ws_base}"); + let player_ws_base = + coordinator_url(&base, "/coord/v1/player").expect("coordinator player url"); + assert!( + player_ws_base.starts_with("wss://"), + "Player Coordinator URL must use WSS: {player_ws_base}" + ); let target = RealmTarget { realm_id: realm, @@ -147,20 +209,29 @@ fn gate_phase12_live_vps_control_channel() { assert!(auto); assert!(!existing_only); assert!(route.is_none()); - println!("GATE 12 PASS handshake welcome received (auto={auto}, existing_only={existing_only})"); + println!( + "GATE 12 PASS handshake welcome received (auto={auto}, existing_only={existing_only})" + ); // 4. Automatic account provisioning over WSS - let outcome = pc.ensure_account(&mut ch, &realm, Some("GateWarrior")).expect("ensure_account"); + let outcome = pc + .ensure_account(&mut ch, &realm, Some("GateWarrior")) + .expect("ensure_account"); assert_eq!(outcome.username, "GATEWARRIOR"); assert!(outcome.created); - println!("GATE 12 PASS automatic account provisioning succeeded: {}", outcome.username); + println!( + "GATE 12 PASS automatic account provisioning succeeded: {}", + outcome.username + ); // 5. A second player tests existing-account linking over WSS let player2_dir = tempfile::tempdir().unwrap(); let pc2 = PlayerControl::open(player2_dir.path(), Arc::new(MemoryStore::default())).unwrap(); let mut ch2 = pc2.connect(&target).expect("Player 2 connect via WSS"); - let linked = pc2.link_existing(&mut ch2, &realm, "OLDHERO", "SECRET123").expect("link_existing"); + let linked = pc2 + .link_existing(&mut ch2, &realm, "OLDHERO", "SECRET123") + .expect("link_existing"); assert_eq!(linked, "OLDHERO"); println!("GATE 12 PASS existing-account linking succeeded: {linked}"); @@ -170,16 +241,24 @@ fn gate_phase12_live_vps_control_channel() { assert_eq!(chars[0].name, "LegacyChar"); assert!(chars[0].eligible); - let claimed = pc2.claim(&mut ch2, chars[0].token).expect("claim character"); + let claimed = pc2 + .claim(&mut ch2, chars[0].token) + .expect("claim character"); assert_eq!(claimed.payload, b"payload-42"); - pc2.acknowledge(&mut ch2, claimed.character_id, &realm, "LegacyChar").expect("acknowledge claim"); - println!("GATE 12 PASS character claim and acknowledge succeeded: character_id={}", claimed.character_id); + pc2.acknowledge(&mut ch2, claimed.character_id, &realm, "LegacyChar") + .expect("acknowledge claim"); + println!( + "GATE 12 PASS character claim and acknowledge succeeded: character_id={}", + claimed.character_id + ); // 7. Clean up ch.close(); ch2.close(); link.stop(); - client.unpublish(&host_key, &realm, now()).expect("unpublish realm"); + client + .unpublish(&host_key, &realm, now()) + .expect("unpublish realm"); println!("GATE 12 PASS realm unpublished and links closed cleanly"); } @@ -200,20 +279,36 @@ fn gate_phase12_1_live_vps_remote_transfer() { let fake = FakeRealm::new(true); let store = Arc::new(Mutex::new(ControlStore::open_in_memory().unwrap())); - let service = Arc::new(HostService::new("srv-gate-12-1", realm, store, fake.clone())); + let service = Arc::new(HostService::new( + "srv-gate-12-1", + realm, + store, + fake.clone(), + )); let host_ws_url = coordinator_url(&base, "/coord/v1/host").expect("coordinator host url"); - let mut link = HostLink::start(host_ws_url.clone(), realm, host_key.clone(), service, Arc::new(now)); + let mut link = HostLink::start( + host_ws_url.clone(), + realm, + host_key.clone(), + service, + Arc::new(now), + ); let deadline = Instant::now() + Duration::from_secs(15); while Instant::now() < deadline && !link.status().connected { std::thread::sleep(Duration::from_millis(200)); } - assert!(link.status().connected, "Host failed to connect to Coordinator via WSS: {:?}", link.status()); + assert!( + link.status().connected, + "Host failed to connect to Coordinator via WSS: {:?}", + link.status() + ); let player_dir = tempfile::tempdir().unwrap(); let pc = PlayerControl::open(player_dir.path(), Arc::new(MemoryStore::default())).unwrap(); - let player_ws_base = coordinator_url(&base, "/coord/v1/player").expect("coordinator player url"); + let player_ws_base = + coordinator_url(&base, "/coord/v1/player").expect("coordinator player url"); let target = RealmTarget { realm_id: realm, @@ -222,50 +317,81 @@ fn gate_phase12_1_live_vps_remote_transfer() { }; let mut ch = pc.connect(&target).expect("Player connect via WSS"); - pc.ensure_account(&mut ch, &realm, Some("VarianWrynn")).expect("ensure_account"); + pc.ensure_account(&mut ch, &realm, Some("VarianWrynn")) + .expect("ensure_account"); let cid = Uuid::now_v7(); let (payload, hash) = transfer_snapshot(cid, 1); let collections = BTreeMap::from([("coa:appearance".to_string(), vec![101, 202])]); // 1. Remote transfer over Coordinator WSS channel - let outcome = pc.transfer_character(&mut ch, cid, "coa", 1, &payload, hash, collections.clone()).expect("transfer_character"); + let outcome = pc + .transfer_character(&mut ch, cid, "coa", 1, &payload, hash, collections.clone()) + .expect("transfer_character"); assert!(outcome.local_guid > 0); assert_ne!(outcome.session_id, Uuid::nil()); - println!("GATE 12.1 PASS remote transfer succeeded over WSS: local_guid={}, session_id={}", outcome.local_guid, outcome.session_id); + println!( + "GATE 12.1 PASS remote transfer succeeded over WSS: local_guid={}, session_id={}", + outcome.local_guid, outcome.session_id + ); // 2. Idempotent retry with same payload discovers existing commit - let retry_outcome = pc.transfer_character(&mut ch, cid, "coa", 1, &payload, hash, collections.clone()).expect("retry transfer"); - assert_eq!(outcome.local_guid, retry_outcome.local_guid, "idempotent retry must return existing guid"); - assert_eq!(outcome.session_id, retry_outcome.session_id, "idempotent retry must return existing session"); + let retry_outcome = pc + .transfer_character(&mut ch, cid, "coa", 1, &payload, hash, collections.clone()) + .expect("retry transfer"); + assert_eq!( + outcome.local_guid, retry_outcome.local_guid, + "idempotent retry must return existing guid" + ); + assert_eq!( + outcome.session_id, retry_outcome.session_id, + "idempotent retry must return existing session" + ); println!("GATE 12.1 PASS idempotent retry returned existing commit without duplication"); // 3. Update with newer revision 2 updates working copy without duplicating character let (payload2, hash2) = transfer_snapshot(cid, 2); - let update_outcome = pc.transfer_character(&mut ch, cid, "coa", 2, &payload2, hash2, collections).expect("update transfer"); - assert_eq!(outcome.local_guid, update_outcome.local_guid, "revision update must preserve guid"); + let update_outcome = pc + .transfer_character(&mut ch, cid, "coa", 2, &payload2, hash2, collections) + .expect("update transfer"); + assert_eq!( + outcome.local_guid, update_outcome.local_guid, + "revision update must preserve guid" + ); println!("GATE 12.1 PASS revision 2 update succeeded without character duplication"); // 4. Incompatible realm rejects before mutation let incomp_id = Uuid::now_v7(); fake.incompatible.lock().unwrap().insert(incomp_id); let (incomp_payload, incomp_hash) = transfer_snapshot(incomp_id, 1); - let err = pc.transfer_character(&mut ch, incomp_id, "coa", 1, &incomp_payload, incomp_hash, BTreeMap::new()).expect_err("should reject"); + let err = pc + .transfer_character( + &mut ch, + incomp_id, + "coa", + 1, + &incomp_payload, + incomp_hash, + BTreeMap::new(), + ) + .expect_err("should reject"); assert_eq!(err.code(), "incompatible"); println!("GATE 12.1 PASS incompatible character rejected cleanly"); ch.close(); link.stop(); - client.unpublish(&host_key, &realm, now()).expect("unpublish realm"); + client + .unpublish(&host_key, &realm, now()) + .expect("unpublish realm"); println!("GATE 12.1 PASS remote transfer gate passed and cleaned up"); } #[test] #[ignore] fn gate_phase13_live_vps_relay() { + use coa_core::control::relay_link::RelayLink; use std::io::{Read, Write}; use std::net::{TcpListener, TcpStream}; - use coa_core::control::relay_link::RelayLink; let Some(base) = base_url() else { return }; @@ -290,7 +416,9 @@ fn gate_phase13_live_vps_relay() { // Spawn mock Auth worker: sends REALM_LIST packet when connection arrives std::thread::spawn(move || { - let Ok((mut stream, _)) = mock_auth_listener.accept() else { return }; + let Ok((mut stream, _)) = mock_auth_listener.accept() else { + return; + }; // Build mock REALM_LIST packet let mut pkt = Vec::new(); pkt.push(0x10); // Opcode @@ -318,7 +446,9 @@ fn gate_phase13_live_vps_relay() { // Spawn mock World worker: echoes back received data std::thread::spawn(move || { - let Ok((mut stream, _)) = mock_world_listener.accept() else { return }; + let Ok((mut stream, _)) = mock_world_listener.accept() else { + return; + }; let mut buf = [0u8; 128]; if let Ok(n) = stream.read(&mut buf) { let _ = stream.write_all(&buf[..n]); @@ -332,24 +462,75 @@ fn gate_phase13_live_vps_relay() { let host_ws_url = coordinator_url(&base, "/coord/v1/host").expect("coordinator host url"); let relay_ws_url = coordinator_url(&base, "/relay/v1/host").expect("relay host url"); - let relay_link = RelayLink::start_with_ports(relay_ws_url.clone(), realm, host_key.clone(), Some(mock_world_port)); + let mut relay_link = RelayLink::start_with_local_ports( + relay_ws_url.clone(), + realm, + host_key.clone(), + mock_auth_port, + Some(mock_world_port), + ); service.set_relay(relay_link.clone()); - let mut host_link = HostLink::start(host_ws_url.clone(), realm, host_key.clone(), service, Arc::new(now)); + let mut host_link = HostLink::start( + host_ws_url.clone(), + realm, + host_key.clone(), + service.clone(), + Arc::new(now), + ); // Wait for HostLink and RelayLink to connect let deadline = Instant::now() + Duration::from_secs(15); - while Instant::now() < deadline && (!host_link.status().connected || !relay_link.status().connected) { + while Instant::now() < deadline + && (!host_link.status().connected || !relay_link.status().connected) + { std::thread::sleep(Duration::from_millis(200)); } - assert!(host_link.status().connected, "Host failed to connect to Coordinator"); - assert!(relay_link.status().connected, "Host failed to connect to Game Relay: {:?}", relay_link.status()); + assert!( + host_link.status().connected, + "Host failed to connect to Coordinator" + ); + assert!( + relay_link.status().connected, + "Host failed to connect to Game Relay: {:?}", + relay_link.status() + ); println!("GATE 13 PASS Host connected to Coordinator and Game Relay via outbound WSS"); + // Reopening the same Manager identity must restore both outbound links and permit a new allocation. + host_link.stop(); + relay_link.stop(); + relay_link = RelayLink::start_with_local_ports( + relay_ws_url.clone(), + realm, + host_key.clone(), + mock_auth_port, + Some(mock_world_port), + ); + service.set_relay(relay_link.clone()); + host_link = HostLink::start( + host_ws_url.clone(), + realm, + host_key.clone(), + service.clone(), + Arc::new(now), + ); + let deadline = Instant::now() + Duration::from_secs(20); + while Instant::now() < deadline + && (!host_link.status().connected || !relay_link.status().connected) + { + std::thread::sleep(Duration::from_millis(200)); + } + assert!( + host_link.status().connected && relay_link.status().connected, + "Restarted Manager failed to restore outbound links" + ); + // Player connects let player_dir = tempfile::tempdir().unwrap(); let pc = PlayerControl::open(player_dir.path(), Arc::new(MemoryStore::default())).unwrap(); - let player_ws_base = coordinator_url(&base, "/coord/v1/player").expect("coordinator player url"); + let player_ws_base = + coordinator_url(&base, "/coord/v1/player").expect("coordinator player url"); let target = RealmTarget { realm_id: realm, @@ -358,39 +539,62 @@ fn gate_phase13_live_vps_relay() { }; let mut ch = pc.connect(&target).expect("Player connect via WSS"); - pc.ensure_account(&mut ch, &realm, Some("Thrall")).expect("ensure_account"); + pc.ensure_account(&mut ch, &realm, Some("Thrall")) + .expect("ensure_account"); // Allocate relay let alloc = pc.allocate_relay(&mut ch).expect("allocate_relay"); - println!("GATE 13 PASS allocated relay: host={}, auth_port={}, world_port={}, token={}", alloc.relay_host, alloc.auth_port, alloc.world_port, alloc.token); + println!( + "GATE 13 PASS allocated relay: host={}, auth_port={}, world_port={}", + alloc.relay_host, alloc.auth_port, alloc.world_port + ); assert_eq!(alloc.relay_host, "coa-manager.duckdns.org"); assert!(alloc.auth_port >= 40000 && alloc.auth_port <= 43999); assert!(alloc.world_port >= 40000 && alloc.world_port <= 43999); assert_ne!(alloc.auth_port, alloc.world_port); // Test AUTH traffic through VPS Relay: verify REALM_LIST address rewrite - let mut auth_client = TcpStream::connect((alloc.relay_host.as_str(), alloc.auth_port)).expect("connect to auth relay"); - auth_client.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + let mut auth_client = TcpStream::connect((alloc.relay_host.as_str(), alloc.auth_port)) + .expect("connect to auth relay"); + auth_client + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); let mut pkt_buf = vec![0u8; 512]; - let n = auth_client.read(&mut pkt_buf).expect("read REALM_LIST response from auth relay"); + let n = auth_client + .read(&mut pkt_buf) + .expect("read REALM_LIST response from auth relay"); assert!(n > 10, "expected realm list packet"); assert_eq!(pkt_buf[0], 0x10, "opcode must be 0x10 (REALM_LIST)"); let expected_world_addr = format!("{}:{}", alloc.relay_host, alloc.world_port); - assert!(pkt_buf[..n].windows(expected_world_addr.len()).any(|w| w == expected_world_addr.as_bytes()), - "REALM_LIST packet must have address rewritten to {expected_world_addr}"); + assert!( + pkt_buf[..n] + .windows(expected_world_addr.len()) + .any(|w| w == expected_world_addr.as_bytes()), + "REALM_LIST packet must have address rewritten to {expected_world_addr}" + ); println!("GATE 13 PASS AUTH stream rewrote REALM_LIST address to VPS world port {expected_world_addr}"); drop(auth_client); // Test WORLD traffic through VPS Relay: verify dumb byte tunneling - let mut world_client = TcpStream::connect((alloc.relay_host.as_str(), alloc.world_port)).expect("connect to world relay"); - world_client.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + let mut world_client = TcpStream::connect((alloc.relay_host.as_str(), alloc.world_port)) + .expect("connect to world relay"); + world_client + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); let test_payload = b"WORLD_TEST_PAYLOAD_FROM_CLIENT_THROUGH_VPS_RELAY"; - world_client.write_all(test_payload).expect("send data to world relay"); + world_client + .write_all(test_payload) + .expect("send data to world relay"); let mut echo_buf = vec![0u8; test_payload.len()]; - world_client.read_exact(&mut echo_buf).expect("read echo from world relay"); - assert_eq!(&echo_buf, test_payload, "world relay must tunnel bytes faithfully without alteration"); + world_client + .read_exact(&mut echo_buf) + .expect("read echo from world relay"); + assert_eq!( + &echo_buf, test_payload, + "world relay must tunnel bytes faithfully without alteration" + ); println!("GATE 13 PASS WORLD stream tunneled game bytes faithfully end-to-end"); drop(world_client); @@ -398,7 +602,8 @@ fn gate_phase13_live_vps_relay() { ch.close(); host_link.stop(); relay_link.stop(); - client.unpublish(&host_key, &realm, now()).expect("unpublish realm"); + client + .unpublish(&host_key, &realm, now()) + .expect("unpublish realm"); println!("GATE 13 PASS game relay gate passed and cleaned up"); } - diff --git a/src-tauri/src/control_cmds.rs b/src-tauri/src/control_cmds.rs index 1dcfe00..9e6b7cf 100644 --- a/src-tauri/src/control_cmds.rs +++ b/src-tauri/src/control_cmds.rs @@ -72,6 +72,9 @@ pub(crate) fn start( None } }; + if let (Some(player), Some(portable)) = (&player, &portable) { + PlayerControl::start_remote_session_sync(player, portable); + } let host = portable.and_then(|portable| { let store = match ControlStore::open(&base.join("host")) { Ok(s) => Arc::new(Mutex::new(s)), @@ -280,12 +283,46 @@ pub(crate) async fn join_realm( out.status = "incompatible".into(); return Ok(out); } - let here = pc_bg.claimed_here(&rid_bg).map_err(from_control)?; - if !here.iter().any(|(id, _)| id.to_string() == character) { - // Phase 12.1: Transfer canonical character to remote Host over the encrypted control channel + let owned_realm = + coa_core::realm_registry::settings::load(&data_dir().join("registry")) + .map_err(|e| fail("registry", e.to_string()))? + .realms + .into_iter() + .find(|(_, cfg)| cfg.enabled && cfg.realm_id == Some(rid_bg)) + .map(|(local_id, _)| local_id); + if let Some(local_id) = owned_realm { + if let Some(pending) = pc_bg + .pending_remote_sessions() + .map_err(from_control)? + .into_iter() + .find(|pending| pending.character_id.to_string() == character) + { + pc_bg + .leave_idle_remote_session(pending.character_id, &portable) + .map_err(from_control)?; + } + let username = out.username.clone(); + portable.call(move |service| { + service.play(&character, &local_id, username.as_deref()) + })??; + } else { + if let Some(pending) = pc_bg + .pending_remote_sessions() + .map_err(from_control)? + .into_iter() + .find(|pending| { + pending.character_id.to_string() == character + && pending.realm_id != rid_bg.to_string() + }) + { + pc_bg + .leave_idle_remote_session(pending.character_id, &portable) + .map_err(from_control)?; + } let char_id_str = character.clone(); - let bundle = - portable.call(move |s| s.export_for_remote_transfer(&char_id_str))??; + let remote_realm = rid_bg.to_string(); + let (bundle, session, revision) = portable + .call(move |s| s.prepare_remote_transfer(&char_id_str, &remote_realm))??; if bundle.ruleset.as_str() != detail.ruleset.as_str() { ch.close(); return Err(fail( @@ -299,11 +336,12 @@ pub(crate) async fn join_realm( } let cuuid = bundle.character_id.as_uuid(); let _outcome = pc_bg - .transfer_character( + .transfer_owned_character( &mut ch, + session, cuuid, bundle.ruleset.as_str(), - 1, + revision, &bundle.payload, bundle.content_hash, bundle.collections,