Restore remote portable progress and resilient relay connections

This commit is contained in:
Дмитрий committed 2026-10-24 11:12:01 +05:00
1 parent 9e9778af74
commit b61a186470
27 files changed
+2579 -145

No files matched your search

+7
View File
@@ -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.
+7
View File
@@ -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.
+7
View File
@@ -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.
+91 -1
View File
@@ -15,6 +15,15 @@ pub const MAX_USERNAME: usize = 16;
pub const MIN_PASSWORD: usize = 8; pub const MIN_PASSWORD: usize = 8;
pub const MAX_PASSWORD: usize = 16; pub const MAX_PASSWORD: usize = 16;
pub const MAX_CHARACTERS: usize = 200; 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 `_`. /// A game account name: 3 to 16 ASCII letters, digits or `_`, not beginning or ending with `_`.
pub fn valid_username(s: &str) -> bool { pub fn valid_username(s: &str) -> bool {
@@ -109,6 +118,15 @@ pub enum Request {
total_size: usize, total_size: usize,
collections: BTreeMap<String, Vec<u32>>, collections: BTreeMap<String, Vec<u32>>,
}, },
OwnedTransferOffer {
transfer_id: Uuid,
character_id: Uuid,
ruleset: String,
canonical_revision: u64,
content_hash: String,
total_size: usize,
collections: BTreeMap<String, Vec<u32>>,
},
/// A chunk of the encoded portable character snapshot payload. /// A chunk of the encoded portable character snapshot payload.
TransferChunk { TransferChunk {
transfer_id: Uuid, transfer_id: Uuid,
@@ -128,6 +146,24 @@ pub enum Request {
transfer_id: Uuid, transfer_id: Uuid,
character_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. /// Phase 13: Request a short-lived relay allocation for a relayed JOIN.
AllocateRelay { AllocateRelay {
player_id: Uuid, player_id: Uuid,
@@ -172,6 +208,22 @@ pub struct CharacterEntry {
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "op", rename_all = "snake_case")] #[serde(tag = "op", rename_all = "snake_case")]
pub enum Response { 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 { Welcome {
realm_id: coa_registry_proto::RealmId, realm_id: coa_registry_proto::RealmId,
automatic: bool, automatic: bool,
@@ -247,7 +299,10 @@ pub fn decode_request(bytes: &[u8]) -> Result<Request> {
.map_err(|_| ControlError::Invalid("a request is not valid".into()))?; .map_err(|_| ControlError::Invalid("a request is not valid".into()))?;
if !matches!( if !matches!(
&r, &r,
Request::TransferOffer { .. } | Request::TransferChunk { .. } Request::TransferOffer { .. }
| Request::OwnedTransferOffer { .. }
| Request::TransferChunk { .. }
| Request::SessionAck { .. }
) && bytes.len() > MAX_CONTROL_REQUEST_BYTES ) && bytes.len() > MAX_CONTROL_REQUEST_BYTES
{ {
return Err(ControlError::Limit( return Err(ControlError::Limit(
@@ -255,6 +310,34 @@ pub fn decode_request(bytes: &[u8]) -> Result<Request> {
)); ));
} }
match &r { 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 { Request::Provision {
desired, password, .. desired, password, ..
} => { } => {
@@ -280,6 +363,13 @@ pub fn decode_request(bytes: &[u8]) -> Result<Request> {
total_size, total_size,
collections, collections,
.. ..
}
| Request::OwnedTransferOffer {
ruleset,
content_hash,
total_size,
collections,
..
} => { } => {
if !matches!(ruleset.as_str(), "coa" | "wildcard" | "reborn") { if !matches!(ruleset.as_str(), "coa" | "wildcard" | "reborn") {
return invalid("transfer ruleset must be coa, wildcard, or reborn"); return invalid("transfer ruleset must be coa, wildcard, or reborn");
+623 -9
View File
@@ -65,10 +65,16 @@ impl LocalBackend {
fn require_portable(&self, local_id: &str) -> Result<()> { fn require_portable(&self, local_id: &str) -> Result<()> {
let settings = settings::load(&self.settings_dir)?; 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(()) Ok(())
} else { } 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 { impl RealmBackend for LocalBackend {
fn session_message(
&self,
local_id: &str,
character: Uuid,
session: Uuid,
account: u32,
) -> Result<super::service::RemoteSessionState> {
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<RealmInfo> { fn info(&self, local_id: &str) -> Result<RealmInfo> {
let cfg = settings::load(&self.settings_dir)?.realms.remove(local_id); let cfg = settings::load(&self.settings_dir)?.realms.remove(local_id);
let access = self.access(local_id)?; let access = self.access(local_id)?;
@@ -310,6 +348,23 @@ impl RealmBackend for LocalBackend {
.map_err(|e| unavailable("import remote", e)) .map_err(|e| unavailable("import remote", e))
} }
fn auth_port(&self, local_id: &str) -> Result<u16> {
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<u16> { fn world_port(&self, local_id: &str) -> Result<u16> {
if let Ok(override_port) = std::env::var("COA_OVERRIDE_LOCAL_WORLD_PORT") { if let Ok(override_port) = std::env::var("COA_OVERRIDE_LOCAL_WORLD_PORT") {
if let Ok(p) = override_port.parse::<u16>() { if let Ok(p) = override_port.parse::<u16>() {
@@ -346,20 +401,560 @@ mod tests {
#[test] #[test]
fn portable_access_requires_owner_opt_in_and_revokes_immediately() { fn portable_access_requires_owner_opt_in_and_revokes_immediately() {
let dir = tempdir().unwrap(); 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)); let backend = LocalBackend::new(Arc::new(rt), dir.path(), dir.path(), Arc::new(Vec::new));
assert!(backend.require_portable("srv-test").is_err()); assert!(backend.require_portable("srv-test").is_err());
let mut cfg = settings::RegistrySettings::default(); let mut cfg = settings::RegistrySettings::default();
cfg.realms.insert("srv-test".into(), settings::PublishConfig { cfg.realms.insert(
realm_id: None, enabled: true, display_name: "Test".into(), description: String::new(), "srv-test".into(),
language: "en".into(), region: None, existing_only: false, allow_portable: true, route: None, 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(); settings::save(dir.path(), &cfg).unwrap();
assert!(backend.require_portable("srv-test").is_ok()); assert!(backend.require_portable("srv-test").is_ok());
cfg.realms.get_mut("srv-test").unwrap().allow_portable = false; cfg.realms.get_mut("srv-test").unwrap().allow_portable = false;
settings::save(dir.path(), &cfg).unwrap(); settings::save(dir.path(), &cfg).unwrap();
assert!(backend.export("srv-test", 1, 1).unwrap_err().to_string().contains("disabled portable")); assert!(backend
assert!(backend.check_transfer("srv-test", Uuid::new_v4(), 1).unwrap_err().to_string().contains("disabled portable")); .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] #[test]
@@ -414,6 +1009,25 @@ mod tests {
Arc::new(move || vec![("test1".to_string(), root_clone.clone())]), 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:coa").unwrap(), 8085);
assert_eq!(backend.world_port("srv-test1:wildcard").unwrap(), 8086); assert_eq!(backend.world_port("srv-test1:wildcard").unwrap(), 8086);
+15 -1
View File
@@ -303,7 +303,21 @@ fn attempt_route_discovery(
let local_auth = std::env::var("COA_OVERRIDE_LOCAL_AUTH_PORT") let local_auth = std::env::var("COA_OVERRIDE_LOCAL_AUTH_PORT")
.ok() .ok()
.and_then(|p| p.parse().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") let local_world = match std::env::var("COA_OVERRIDE_LOCAL_WORLD_PORT")
.ok() .ok()
.and_then(|p| p.parse().ok()) .and_then(|p| p.parse().ok())
+10 -6
View File
@@ -36,6 +36,7 @@ pub struct HostControl {
struct Running { struct Running {
realm_id: RealmId, realm_id: RealmId,
url: String, url: String,
ports: Option<(u16, u16)>,
link: HostLink, link: HostLink,
#[allow(dead_code)] #[allow(dead_code)]
relay_link: Option<Arc<crate::control::relay_link::RelayLink>>, relay_link: Option<Arc<crate::control::relay_link::RelayLink>>,
@@ -84,7 +85,7 @@ impl HostControl {
running.retain(|id, r| { running.retain(|id, r| {
wanted wanted
.get(id) .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 { for (local_id, (realm_id, url, base, key)) in wanted {
if running.contains_key(&local_id) { if running.contains_key(&local_id) {
@@ -96,15 +97,17 @@ impl HostControl {
store.clone(), store.clone(),
backend.clone(), backend.clone(),
)); ));
let relay_link = match service.local_world_port() { let ports = service.local_auth_port().ok().zip(service.local_world_port().ok());
Ok(world_port) => { let relay_link = match ports {
Some((auth_port, world_port)) => {
crate::control::transport::relay_url(&base, "/relay/v1/host").map( crate::control::transport::relay_url(&base, "/relay/v1/host").map(
|r_url| { |r_url| {
let rl = let rl =
crate::control::relay_link::RelayLink::start_with_ports( crate::control::relay_link::RelayLink::start_with_local_ports(
r_url, r_url,
realm_id, realm_id,
key.clone(), key.clone(),
auth_port,
Some(world_port), Some(world_port),
); );
service.set_relay(rl.clone()); service.set_relay(rl.clone());
@@ -112,8 +115,8 @@ impl HostControl {
}, },
) )
} }
Err(e) => { None => {
eprintln!("Relay link unavailable for {local_id}: cannot resolve world port: {e}"); tracing::warn!(%local_id, "relay is waiting for valid local auth and world ports");
None None
} }
}; };
@@ -129,6 +132,7 @@ impl HostControl {
Running { Running {
realm_id, realm_id,
url, url,
ports,
link, link,
relay_link, relay_link,
direct_route, direct_route,
+40 -11
View File
@@ -20,6 +20,8 @@ use super::store::{AccountKind, ControlStore};
use super::transport::{self, LinkError}; use super::transport::{self, LinkError};
use crate::{Error, Result}; use crate::{Error, Result};
mod portable_sync;
/// Where and as whom to reach a realm. /// Where and as whom to reach a realm.
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
pub struct RealmTarget { 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). /// 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, &self,
ch: &mut PlayerChannel, ch: &mut PlayerChannel,
transfer_id: Uuid,
owned_session: bool,
character_id: Uuid, character_id: Uuid,
ruleset: &str, ruleset: &str,
canonical_revision: u64, canonical_revision: u64,
@@ -422,16 +426,27 @@ impl PlayerControl {
content_hash: [u8; 32], content_hash: [u8; 32],
collections: BTreeMap<String, Vec<u32>>, collections: BTreeMap<String, Vec<u32>>,
) -> R<RemoteTransferOutcome> { ) -> R<RemoteTransferOutcome> {
let transfer_id = Uuid::now_v7();
let content_hash_str = hex::encode(content_hash); let content_hash_str = hex::encode(content_hash);
let offer = Request::TransferOffer { let offer = if owned_session {
transfer_id, Request::OwnedTransferOffer {
character_id, transfer_id,
ruleset: ruleset.to_string(), character_id,
canonical_revision, ruleset: ruleset.to_string(),
content_hash: content_hash_str.clone(), canonical_revision,
total_size: payload.len(), content_hash: content_hash_str.clone(),
collections, 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 r = ch.request(&offer)?;
let mut offset = match r { let mut offset = match r {
@@ -456,7 +471,21 @@ impl PlayerControl {
notes, 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( return Err(ControlFail::Link(LinkError::Protocol(
"unexpected response to transfer offer".into(), "unexpected response to transfer offer".into(),
@@ -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<String, Vec<u32>>,
) -> R<RemoteTransferOutcome> {
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<String, Vec<u32>>,
) -> R<RemoteTransferOutcome> {
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<Vec<PlayerRemoteSession>> {
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<bool> {
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<PlayerChannel> {
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<bool> {
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<Self>, runtime: &Arc<PortableRuntime>) {
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::<Uuid, String>::new();
let mut channels = std::collections::HashMap::<String, PlayerChannel>::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));
}
});
}
}
+16 -2
View File
@@ -13,6 +13,9 @@ use super::identity::PlayerIdentity;
use super::transport::{self, LinkError, Ws}; use super::transport::{self, LinkError, Ws};
pub struct PlayerChannel { pub struct PlayerChannel {
pub(crate) realm_id: RealmId,
pub(crate) realm_key: VerifyingKey,
pub(crate) coordinator: String,
ws: Ws, ws: Ws,
channel: Channel, channel: Channel,
conn: u32, conn: u32,
@@ -35,7 +38,11 @@ impl PlayerChannel {
Duration::from_millis(500), Duration::from_millis(500),
)?; )?;
let nonce = match transport::next_frame(&mut ws, STEP)? { 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())), _ => return Err(LinkError::Protocol("a challenge was expected".into())),
}; };
transport::send_text( transport::send_text(
@@ -67,7 +74,14 @@ impl PlayerChannel {
{ {
transport::send_binary(&mut ws, f)?; 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 { pub fn connection(&self) -> u32 {
+172 -9
View File
@@ -1,6 +1,6 @@
//! Outbound Host <-> Game Relay tunnel link (Phase 13). //! Outbound Host <-> Game Relay tunnel link (Phase 13).
//! Maintains outbound connectivity from Host Manager to the Game Relay, //! 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::collections::HashMap;
use std::io::{Read, Write}; use std::io::{Read, Write};
@@ -79,6 +79,22 @@ impl RelayLink {
realm_id: RealmId, realm_id: RealmId,
key: SigningKey, key: SigningKey,
local_world_port: Option<u16>, local_world_port: Option<u16>,
) -> Arc<Self> {
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<u16>,
) -> Arc<Self> { ) -> Arc<Self> {
let stop = Arc::new(AtomicBool::new(false)); let stop = Arc::new(AtomicBool::new(false));
let status = Arc::new(Mutex::new(RelayStatus::default())); let status = Arc::new(Mutex::new(RelayStatus::default()));
@@ -88,7 +104,18 @@ impl RelayLink {
let st = status.clone(); let st = status.clone();
let join = std::thread::Builder::new() let join = std::thread::Builder::new()
.name("control-relay-host".into()) .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(); .ok();
Arc::new(RelayLink { Arc::new(RelayLink {
@@ -136,12 +163,22 @@ fn run(
)>, )>,
stop: &AtomicBool, stop: &AtomicBool,
status: &Mutex<RelayStatus>, status: &Mutex<RelayStatus>,
local_auth_port: u16,
local_world_port: Option<u16>, local_world_port: Option<u16>,
) { ) {
let mut failures = 0u32; let mut failures = 0u32;
while !stop.load(Ordering::SeqCst) { while !stop.load(Ordering::SeqCst) {
let began = Instant::now(); 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 { if let Err(error) = &result {
tracing::warn!(%realm, %error, "game relay connection lost; retrying"); tracing::warn!(%realm, %error, "game relay connection lost; retrying");
} }
@@ -176,6 +213,15 @@ fn run(
struct StreamWorker { struct StreamWorker {
tx: Sender<TunnelMsg>, tx: Sender<TunnelMsg>,
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( fn session(
@@ -189,6 +235,7 @@ fn session(
)>, )>,
stop: &AtomicBool, stop: &AtomicBool,
status: &Mutex<RelayStatus>, status: &Mutex<RelayStatus>,
local_auth_port: u16,
local_world_port: Option<u16>, local_world_port: Option<u16>,
) -> Result<(), LinkError> { ) -> Result<(), LinkError> {
let mut ws = transport::connect(url, POLL)?; let mut ws = transport::connect(url, POLL)?;
@@ -257,6 +304,7 @@ fn session(
let (ws_out_tx, ws_out_rx) = channel::<TunnelMsg>(); let (ws_out_tx, ws_out_rx) = channel::<TunnelMsg>();
let mut last_ping = Instant::now(); let mut last_ping = Instant::now();
let mut last_pong = Instant::now();
while !stop.load(Ordering::SeqCst) { while !stop.load(Ordering::SeqCst) {
// Drain pending allocation requests // Drain pending allocation requests
@@ -277,10 +325,14 @@ fn session(
} }
// Keepalive // Keepalive
if last_pong.elapsed() > Duration::from_secs(90) {
return Err(LinkError::Timeout);
}
if last_ping.elapsed() > Duration::from_secs(30) { 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(), serde_json::to_string(&TunnelMsg::Ping).unwrap().into(),
)); ))
.map_err(|error| LinkError::Io(error.to_string()))?;
last_ping = Instant::now(); last_ping = Instant::now();
} }
@@ -321,7 +373,7 @@ fn session(
} => { } => {
// Strictly predefined local targets: Auth (3724) or World (8085 or bound realm world port) // Strictly predefined local targets: Auth (3724) or World (8085 or bound realm world port)
let local_port = match target { let local_port = match target {
RelayTarget::Auth => target.local_port(), RelayTarget::Auth => local_auth_port,
RelayTarget::World => match local_world_port { RelayTarget::World => match local_world_port {
Some(p) => p, Some(p) => p,
None => { None => {
@@ -341,7 +393,13 @@ fn session(
// Spawn worker for this stream // Spawn worker for this stream
spawn_local_tcp_bridge(stream_id, tcp, stream_rx, ws_tx); 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 // Reply ConnectOk
let ok_msg = TunnelMsg::ConnectOk { stream_id }; let ok_msg = TunnelMsg::ConnectOk { stream_id };
@@ -367,7 +425,8 @@ fn session(
} }
} }
TunnelMsg::Close { stream_id } => { 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 }); let _ = w.tx.send(TunnelMsg::Close { stream_id });
} }
set_status(status, |s| s.active_streams = streams.len()); set_status(status, |s| s.active_streams = streams.len());
@@ -378,7 +437,9 @@ fn session(
} }
set_status(status, |s| s.active_streams = streams.len()); 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(); .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);
}
}
+212 -26
View File
@@ -20,6 +20,20 @@ use uuid::Uuid;
use super::store::{AccountKind, ClaimState, ControlStore, HostTransfer, TransferState}; use super::store::{AccountKind, ClaimState, ControlStore, HostTransfer, TransferState};
use crate::{Error, Result}; use crate::{Error, Result};
mod portable_sync;
#[derive(Clone, Debug)]
pub struct RemoteSessionMessage {
pub kind: app::SessionMessageKind,
pub bytes: Vec<u8>,
}
#[derive(Clone, Debug)]
pub struct RemoteSessionState {
pub message: Option<RemoteSessionMessage>,
pub closed: bool,
}
#[derive(Clone, Debug, PartialEq, Eq)] #[derive(Clone, Debug, PartialEq, Eq)]
pub struct RealmInfo { pub struct RealmInfo {
/// Create accounts for joining players (the default). When false the realm only links accounts that exist. /// 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)] #[derive(Clone, Debug)]
pub struct RemoteImportParams { pub struct RemoteImportParams {
pub session_id: Uuid,
pub character_id: Uuid, pub character_id: Uuid,
pub revision: u64, pub revision: u64,
pub payload: Vec<u8>, pub payload: Vec<u8>,
@@ -89,6 +104,30 @@ pub enum PreflightVerdict {
/// The realm itself: its login database, its console and the portable engine. /// The realm itself: its login database, its console and the portable engine.
pub trait RealmBackend: Send + Sync { pub trait RealmBackend: Send + Sync {
fn session_message(
&self,
_local_id: &str,
_character: Uuid,
_session: Uuid,
_account: u32,
) -> Result<RemoteSessionState> {
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<RealmInfo>; fn info(&self, local_id: &str) -> Result<RealmInfo>;
fn account_by_name(&self, local_id: &str, username: &str) -> Result<Option<AccountRow>>; fn account_by_name(&self, local_id: &str, username: &str) -> Result<Option<AccountRow>>;
fn account_by_id(&self, local_id: &str, id: u32) -> Result<Option<AccountRow>>; fn account_by_id(&self, local_id: &str, id: u32) -> Result<Option<AccountRow>>;
@@ -121,6 +160,9 @@ pub trait RealmBackend: Send + Sync {
params: RemoteImportParams, params: RemoteImportParams,
) -> Result<RemoteImportOutcome>; ) -> Result<RemoteImportOutcome>;
/// Local world port for this realm. Must fail closed if the realm port cannot be resolved. /// Local world port for this realm. Must fail closed if the realm port cannot be resolved.
fn auth_port(&self, _local_id: &str) -> Result<u16> {
Ok(coa_control_proto::relay::LOCAL_AUTH_PORT)
}
fn world_port(&self, local_id: &str) -> Result<u16>; fn world_port(&self, local_id: &str) -> Result<u16>;
} }
@@ -148,7 +190,7 @@ impl Session {
} }
const MAX_REQUESTS_PER_SESSION: u32 = 60; 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_FAILURES: usize = 5;
const LINK_WINDOW: Duration = Duration::from_secs(600); const LINK_WINDOW: Duration = Duration::from_secs(600);
@@ -222,6 +264,10 @@ impl HostService {
self.realm_id self.realm_id
} }
pub fn local_auth_port(&self) -> Result<u16> {
self.backend.auth_port(&self.local_id)
}
pub fn local_world_port(&self) -> Result<u16> { pub fn local_world_port(&self) -> Result<u16> {
self.backend.world_port(&self.local_id) self.backend.world_port(&self.local_id)
} }
@@ -246,7 +292,12 @@ impl HostService {
public_key: &str, public_key: &str,
request: Request, request: Request,
) -> Response { ) -> Response {
if matches!(request, Request::TransferChunk { .. }) { if matches!(
request,
Request::TransferChunk { .. }
| Request::SessionRead { .. }
| Request::SessionAck { .. }
) {
session.transfer_chunks += 1; session.transfer_chunks += 1;
if session.transfer_chunks > MAX_TRANSFER_CHUNKS_PER_SESSION { if session.transfer_chunks > MAX_TRANSFER_CHUNKS_PER_SESSION {
return err( return err(
@@ -325,6 +376,26 @@ impl HostService {
content_hash, content_hash,
total_size, total_size,
collections, 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 { Request::TransferChunk {
transfer_id, transfer_id,
@@ -337,6 +408,32 @@ impl HostService {
transfer_id, transfer_id,
character_id, character_id,
} => self.transfer_ack(player, 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 } => { Request::AllocateRelay { player_id } => {
self.allocate_relay(session, player, &player_id) self.allocate_relay(session, player, &player_id)
} }
@@ -742,6 +839,7 @@ impl HostService {
content_hash: String, content_hash: String,
total_size: usize, total_size: usize,
collections: BTreeMap<String, Vec<u32>>, collections: BTreeMap<String, Vec<u32>>,
owned_session: bool,
) -> Response { ) -> Response {
let Ok(_guard) = self.write.lock() else { let Ok(_guard) = self.write.lock() else {
return err(AppError::Unavailable, "Busy."); 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 // 2. Active session conflict: if account is online with active characters
match self.backend.presence(&self.local_id, m.account_id) { match self.backend.presence(&self.local_id, m.account_id) {
Ok(p) if p.account_online && p.online_characters > 0 => { 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 // 5. Resume check for this specific transfer_id
if let Ok(Some(cur)) = self if let Ok(Some(cur)) = self
.store .store
@@ -1099,6 +1208,7 @@ impl HostService {
}; };
let params = RemoteImportParams { let params = RemoteImportParams {
session_id: t.transfer_id,
character_id: t.character_id, character_id: t.character_id,
revision: t.revision, revision: t.revision,
payload, 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] #[test]
fn wrong_player_identity_cannot_overwrite_another_players_realm_copy() { fn wrong_player_identity_cannot_overwrite_another_players_realm_copy() {
let (_realm, s, p, _acc) = claimed_setup(); let (_realm, s, p, _acc) = claimed_setup();
@@ -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<u32> {
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<usize> {
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),
}
}
}
+113
View File
@@ -72,6 +72,13 @@ CREATE TABLE IF NOT EXISTS host_transfer(
updated_at TEXT NOT NULL updated_at TEXT NOT NULL
); );
CREATE INDEX IF NOT EXISTS host_transfer_char ON host_transfer(realm_local_id, character_id); 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 { fn now() -> String {
@@ -177,6 +184,15 @@ pub struct PlayerRealm {
pub kind: AccountKind, 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 { pub struct ControlStore {
conn: Connection, conn: Connection,
} }
@@ -186,6 +202,66 @@ fn uuid(s: String) -> Result<Uuid> {
} }
impl ControlStore { impl ControlStore {
pub fn player_remote_sessions(&self) -> Result<Vec<PlayerRemoteSession>> {
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<Self> { pub fn open(dir: &Path) -> Result<Self> {
std::fs::create_dir_all(dir)?; std::fs::create_dir_all(dir)?;
let conn = Connection::open(dir.join("control.sqlite")).map_err(db)?; let conn = Connection::open(dir.join("control.sqlite")).map_err(db)?;
@@ -574,6 +650,43 @@ fn row_transfer(r: &rusqlite::Row<'_>) -> rusqlite::Result<Result<HostTransfer>>
mod tests { mod tests {
use super::*; 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] #[test]
fn a_player_owns_one_account_per_realm_and_an_account_one_player() { fn a_player_owns_one_account_per_realm_and_an_account_one_player() {
let s = ControlStore::open_in_memory().unwrap(); let s = ControlStore::open_in_memory().unwrap();
+50 -10
View File
@@ -31,6 +31,8 @@ use serde::{Deserialize, Serialize};
use crate::db::Db; use crate::db::Db;
use crate::ra::Ra; use crate::ra::Ra;
mod remote_sync;
use super::super::capabilities::RealmCapabilities; use super::super::capabilities::RealmCapabilities;
use super::super::compat::{self, CompatibilityReport, Inputs, Operation, Outcome, Topic}; use super::super::compat::{self, CompatibilityReport, Inputs, Operation, Outcome, Topic};
use super::super::error::PortableError; use super::super::error::PortableError;
@@ -729,6 +731,12 @@ impl PortableService {
updated_at: at, updated_at: at,
}); });
} }
let remote_active = self
.owner
.owner_active_remote(c.character_id)
.ok()
.flatten()
.is_some();
let active = copies let active = copies
.iter() .iter()
.find(|x| { .find(|x| {
@@ -741,9 +749,9 @@ impl PortableService {
) )
}) })
.cloned(); .cloned();
let status = active let status = remote_active
.as_ref() .then_some(PlayStatus::Syncing)
.map(|a| a.status) .or_else(|| active.as_ref().map(|a| a.status))
.or_else(|| { .or_else(|| {
copies copies
.iter() .iter()
@@ -1318,6 +1326,11 @@ impl PortableService {
if events.is_ok() { if events.is_ok() {
delivery = (|| { delivery = (|| {
for m in host.outbox()? { 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 mut owner = OwnerService::new(&mut self.owner);
let ack: OwnerAck = if m.started { let ack: OwnerAck = if m.started {
owner.handle_started(&m.bytes)? owner.handle_started(&m.bytes)?
@@ -1331,6 +1344,25 @@ impl PortableService {
} }
} }
for m in host.collection_outbox()? { 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)?; let ack = OwnerService::new(&mut self.owner).handle_collection(&m.bytes)?;
host.receive_collection_ack(&mut bridge, m.account, &ack)?; 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 // Install or advance copy in host store
self.host self.host
.host_install_copy(self.host_profile, &model, params.revision, "remote")?; .host_install_copy(self.host_profile, &model, params.revision, "remote")?;
@@ -1677,13 +1723,7 @@ impl PortableService {
} }
} }
let existing_mapping = self let session_id = SessionId::from_uuid(params.session_id)?;
.host
.server_mappings(cid)?
.into_iter()
.find(|m| m.server_id == realm_id);
let session_id = SessionId::new();
let envelope = super::super::session::protocol::Envelope::from_encoded( let envelope = super::super::session::protocol::Envelope::from_encoded(
&super::super::snapshot::encode(&model)?, &super::super::snapshot::encode(&model)?,
); );
@@ -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<bool> {
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<bool> {
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<Option<uuid::Uuid>> {
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<Vec<u8>> {
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<RemoteSessionState> {
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
);
}
}
+1 -1
View File
@@ -177,7 +177,7 @@ impl Store {
fn init(conn: Connection) -> Result<Store> { fn init(conn: Connection) -> Result<Store> {
conn.busy_timeout(std::time::Duration::from_secs(5))?; conn.busy_timeout(std::time::Duration::from_secs(5))?;
let _ = conn.pragma_update(None, "journal_mode", "WAL"); 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 }; let mut store = Store { conn };
store.migrate()?; store.migrate()?;
store.conn.pragma_update(None, "foreign_keys", true)?; store.conn.pragma_update(None, "foreign_keys", true)?;
@@ -181,6 +181,10 @@ impl Store {
check_server_id(server_id)?; check_server_id(server_id)?;
let tx = self.write_tx()?; let tx = self.write_tx()?;
let record = read_character(&tx, id)?; 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<String> = tx let live: Option<String> = 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)) .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()?; .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()])?; 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)?; let offer = open_offer(&tx, id, server_id)?;
tx.commit()?; tx.commit()?;
Ok(offer) Ok(offer)
@@ -521,4 +527,21 @@ impl Store {
Ok(read_owner_row(&self.conn, session)? Ok(read_owner_row(&self.conn, session)?
.map(|r| (r.state, r.last_sequence, r.head_revision))) .map(|r| (r.state, r.last_sequence, r.head_revision)))
} }
pub fn owner_next_session(
&self,
character: CharacterId,
realm: &str,
) -> Result<Option<SessionId>> {
let sid: Option<String> = 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<Option<(String, String)>> {
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<Option<String>> {
Ok(read_owner_row(&self.conn, session)?.map(|row| row.server_id))
}
} }
+26 -2
View File
@@ -43,7 +43,14 @@ fn schema_nine_upgrade_preserves_characters_and_isolates_collections() {
assert_eq!(store.schema_version().unwrap(), 10); assert_eq!(store.schema_version().unwrap(), 10);
assert_eq!(store.load_current(id).unwrap(), before); assert_eq!(store.load_current(id).unwrap(), before);
assert_eq!(store.server_mappings(id).unwrap()[0].local_guid, 42); 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!( assert_eq!(
store store
.collection_info(profile, Ruleset::Coa, kind) .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()) .merge_collection(profile, Ruleset::Coa, kind, &IdSet::from_ids([30]).unwrap())
.unwrap(); .unwrap();
assert_eq!( 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() IdSet::from_ids([10, 20, 30]).unwrap()
); );
let foreign: (i64, Vec<u8>) = store let foreign: (i64, Vec<u8>) = store
@@ -2578,3 +2589,16 @@ fn reborn_store_corruption_fails_closed() {
"Loading reborn summary with Wildcard snapshot must fail 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);
}
}
+2
View File
@@ -58,3 +58,5 @@ Use **Saved versions** to see when progress was saved. If the Manager reports a
## Portable access and relay ## 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. 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.
+2
View File
@@ -49,3 +49,5 @@ Unter **Gespeicherte Versionen (Saved versions)** siehst du, wann dein Fortschri
## Portable Charaktere und Relay ## 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. 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.
+2
View File
@@ -49,3 +49,5 @@ En **Versiones guardadas (Saved versions)** puedes ver cuándo se guardó el pro
## Personajes portátiles y relay ## 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. 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.
+2
View File
@@ -49,3 +49,5 @@ Sur un royaume dont le niveau maximal est inférieur, votre personnage peut temp
## Personnages portables et relais ## 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. 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.
+2
View File
@@ -49,3 +49,5 @@
## Доступ переносимых персонажей и relay ## Доступ переносимых персонажей и relay
В блоке «Публичный список» включите «Разрешить переносимых персонажей на этом реалме» и сохраните. По умолчанию доступ выключен. Отключение запрещает новые экспорты и импорты, но не удаляет сохранённых персонажей. Менеджер владельца должен оставаться запущенным. Подключение к службе управления и игровой relay имеют отдельные статусы: управляющий канал ещё не означает наличие игрового туннеля. Персонажи остаются в своём режиме (CoA, Wildcard или Warcraft Reborn); смена сервера не преобразует режим персонажа. В блоке «Публичный список» включите «Разрешить переносимых персонажей на этом реалме» и сохраните. По умолчанию доступ выключен. Отключение запрещает новые экспорты и импорты, но не удаляет сохранённых персонажей. Менеджер владельца должен оставаться запущенным. Подключение к службе управления и игровой relay имеют отдельные статусы: управляющий канал ещё не означает наличие игрового туннеля. Персонажи остаются в своём режиме (CoA, Wildcard или Warcraft Reborn); смена сервера не преобразует режим персонажа.
Во время игры должны работать оба менеджера. После обрыва связи или перезапуска сохранение продолжается автоматически; повторные входы тоже отслеживаются. Перед сменой реалма дождитесь сохранения. Ваши установленные серверы используют локальную подготовку и восстановление. Ограничение экспериментальной функции: устаревшая копия у другого владельца пока не обновляется автоматически при работающем worldserver. Для возвращения может потребоваться обслуживание владельцем. Устаревшая копия блокируется, основной персонаж остаётся сохранённым. Не удаляйте персонажа и не переустанавливайте сервер для обхода сообщения.
+2
View File
@@ -49,3 +49,5 @@
## 便携角色访问与中继 ## 便携角色访问与中继
在服务器的公开列表设置中,勾选“允许便携角色进入此服务器”并保存。默认不允许访问。关闭后会阻止新的导出和导入,但不会删除已保存的角色。主机端 Manager 必须保持运行。控制连接与游戏中继分别显示状态;控制连接成功不代表游戏隧道已经建立。角色始终保留原来的游戏模式(CoA、Wildcard 或 Warcraft Reborn),更换服务器不会转换角色模式。 在服务器的公开列表设置中,勾选“允许便携角色进入此服务器”并保存。默认不允许访问。关闭后会阻止新的导出和导入,但不会删除已保存的角色。主机端 Manager 必须保持运行。控制连接与游戏中继分别显示状态;控制连接成功不代表游戏隧道已经建立。角色始终保留原来的游戏模式(CoA、Wildcard 或 Warcraft Reborn),更换服务器不会转换角色模式。
请保持两个管理器运行。断线或重启后,待保存的进度会继续同步,再次登录也会继续跟踪。请等待保存完成后再切换服务器。自己安装的服务器使用本地准备和恢复流程。实验功能限制:其他主机上的旧角色副本目前无法在其 worldserver 运行时自动更新。返回该服务器可能需要主人进行维护。旧副本会被阻止,你的主角色仍然保存在管理器中。请勿通过删除角色或重装服务器绕过提示。
+265 -60
View File
@@ -28,7 +28,10 @@ fn transfer_snapshot(id: Uuid, money: u32) -> (Vec<u8>, [u8; 32]) {
} }
fn now() -> i64 { 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 { fn random_key() -> SigningKey {
@@ -45,26 +48,61 @@ fn base_url() -> Option<String> {
fn sample_caps(max_level: u32) -> AdvertisedCapabilities { fn sample_caps(max_level: u32) -> AdvertisedCapabilities {
let content = ContentProfile { let content = ContentProfile {
ruleset: Ruleset::Coa, 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], online_import_job_formats: vec![2],
session_protocol: 2, session_protocol: 2,
collection_protocol: 1, 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(), 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 } }], extensions: vec![ExtensionSupport {
client_catalog: [("Appearances.dbc".to_string(), CatalogEntry { sha256: "a".repeat(64), records: 5 })].into_iter().collect(), 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(); let content_profile_hash = content.hash();
AdvertisedCapabilities { AdvertisedCapabilities {
profile_version: 2, 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,
content_profile_hash, 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<dyn std::error::Error>> { fn register_realm(
client: &RegistryClient,
realm: &RealmId,
key: &SigningKey,
) -> std::result::Result<(), Box<dyn std::error::Error>> {
let listing = Listing { let listing = Listing {
display_name: "Live Control Gate".into(), display_name: "Live Control Gate".into(),
description: "TLS Phase 12 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()), region: Some("EU".into()),
rates: Rates::default(), rates: Rates::default(),
modules: vec![], modules: vec![],
account_provisioning: AccountProvisioning { automatic: true, existing_only: false }, account_provisioning: AccountProvisioning {
automatic: true,
existing_only: false,
},
manager_version: "0.7.0".into(), manager_version: "0.7.0".into(),
}; };
let capabilities = sample_caps(60); let capabilities = sample_caps(60);
@@ -84,7 +125,11 @@ fn register_realm(client: &RegistryClient, realm: &RealmId, key: &SigningKey) ->
listing, listing,
capabilities_hash: capabilities.advert_hash(), capabilities_hash: capabilities.advert_hash(),
capabilities, capabilities,
population: Population { players: 0, bots: 0, capacity: None }, population: Population {
players: 0,
bots: 0,
capacity: None,
},
}; };
let ts = now(); let ts = now();
client.register(key, &req, ts)?; 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 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"); 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 // Wait for Host to connect to Coordinator
let deadline = Instant::now() + Duration::from_secs(15); let deadline = Instant::now() + Duration::from_secs(15);
while Instant::now() < deadline && !link.status().connected { while Instant::now() < deadline && !link.status().connected {
std::thread::sleep(Duration::from_millis(200)); 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}"); println!("GATE 12 PASS Coordinator Host connection established over WSS: {host_ws_url}");
// 3. Player connects to Host via Coordinator over WSS // 3. Player connects to Host via Coordinator over WSS
let player_dir = tempfile::tempdir().unwrap(); let player_dir = tempfile::tempdir().unwrap();
let pc = PlayerControl::open(player_dir.path(), Arc::new(MemoryStore::default())).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 =
assert!(player_ws_base.starts_with("wss://"), "Player Coordinator URL must use WSS: {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 { let target = RealmTarget {
realm_id: realm, realm_id: realm,
@@ -147,20 +209,29 @@ fn gate_phase12_live_vps_control_channel() {
assert!(auto); assert!(auto);
assert!(!existing_only); assert!(!existing_only);
assert!(route.is_none()); 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 // 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_eq!(outcome.username, "GATEWARRIOR");
assert!(outcome.created); 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 // 5. A second player tests existing-account linking over WSS
let player2_dir = tempfile::tempdir().unwrap(); let player2_dir = tempfile::tempdir().unwrap();
let pc2 = PlayerControl::open(player2_dir.path(), Arc::new(MemoryStore::default())).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 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"); assert_eq!(linked, "OLDHERO");
println!("GATE 12 PASS existing-account linking succeeded: {linked}"); 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_eq!(chars[0].name, "LegacyChar");
assert!(chars[0].eligible); 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"); assert_eq!(claimed.payload, b"payload-42");
pc2.acknowledge(&mut ch2, claimed.character_id, &realm, "LegacyChar").expect("acknowledge claim"); pc2.acknowledge(&mut ch2, claimed.character_id, &realm, "LegacyChar")
println!("GATE 12 PASS character claim and acknowledge succeeded: character_id={}", claimed.character_id); .expect("acknowledge claim");
println!(
"GATE 12 PASS character claim and acknowledge succeeded: character_id={}",
claimed.character_id
);
// 7. Clean up // 7. Clean up
ch.close(); ch.close();
ch2.close(); ch2.close();
link.stop(); 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"); 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 fake = FakeRealm::new(true);
let store = Arc::new(Mutex::new(ControlStore::open_in_memory().unwrap())); 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 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); let deadline = Instant::now() + Duration::from_secs(15);
while Instant::now() < deadline && !link.status().connected { while Instant::now() < deadline && !link.status().connected {
std::thread::sleep(Duration::from_millis(200)); 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 player_dir = tempfile::tempdir().unwrap();
let pc = PlayerControl::open(player_dir.path(), Arc::new(MemoryStore::default())).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 { let target = RealmTarget {
realm_id: realm, 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"); 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 cid = Uuid::now_v7();
let (payload, hash) = transfer_snapshot(cid, 1); let (payload, hash) = transfer_snapshot(cid, 1);
let collections = BTreeMap::from([("coa:appearance".to_string(), vec![101, 202])]); let collections = BTreeMap::from([("coa:appearance".to_string(), vec![101, 202])]);
// 1. Remote transfer over Coordinator WSS channel // 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!(outcome.local_guid > 0);
assert_ne!(outcome.session_id, Uuid::nil()); 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 // 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"); let retry_outcome = pc
assert_eq!(outcome.local_guid, retry_outcome.local_guid, "idempotent retry must return existing guid"); .transfer_character(&mut ch, cid, "coa", 1, &payload, hash, collections.clone())
assert_eq!(outcome.session_id, retry_outcome.session_id, "idempotent retry must return existing session"); .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"); println!("GATE 12.1 PASS idempotent retry returned existing commit without duplication");
// 3. Update with newer revision 2 updates working copy without duplicating character // 3. Update with newer revision 2 updates working copy without duplicating character
let (payload2, hash2) = transfer_snapshot(cid, 2); let (payload2, hash2) = transfer_snapshot(cid, 2);
let update_outcome = pc.transfer_character(&mut ch, cid, "coa", 2, &payload2, hash2, collections).expect("update transfer"); let update_outcome = pc
assert_eq!(outcome.local_guid, update_outcome.local_guid, "revision update must preserve guid"); .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"); println!("GATE 12.1 PASS revision 2 update succeeded without character duplication");
// 4. Incompatible realm rejects before mutation // 4. Incompatible realm rejects before mutation
let incomp_id = Uuid::now_v7(); let incomp_id = Uuid::now_v7();
fake.incompatible.lock().unwrap().insert(incomp_id); fake.incompatible.lock().unwrap().insert(incomp_id);
let (incomp_payload, incomp_hash) = transfer_snapshot(incomp_id, 1); 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"); assert_eq!(err.code(), "incompatible");
println!("GATE 12.1 PASS incompatible character rejected cleanly"); println!("GATE 12.1 PASS incompatible character rejected cleanly");
ch.close(); ch.close();
link.stop(); 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"); println!("GATE 12.1 PASS remote transfer gate passed and cleaned up");
} }
#[test] #[test]
#[ignore] #[ignore]
fn gate_phase13_live_vps_relay() { fn gate_phase13_live_vps_relay() {
use coa_core::control::relay_link::RelayLink;
use std::io::{Read, Write}; use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream}; use std::net::{TcpListener, TcpStream};
use coa_core::control::relay_link::RelayLink;
let Some(base) = base_url() else { return }; 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 // Spawn mock Auth worker: sends REALM_LIST packet when connection arrives
std::thread::spawn(move || { 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 // Build mock REALM_LIST packet
let mut pkt = Vec::new(); let mut pkt = Vec::new();
pkt.push(0x10); // Opcode pkt.push(0x10); // Opcode
@@ -318,7 +446,9 @@ fn gate_phase13_live_vps_relay() {
// Spawn mock World worker: echoes back received data // Spawn mock World worker: echoes back received data
std::thread::spawn(move || { 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]; let mut buf = [0u8; 128];
if let Ok(n) = stream.read(&mut buf) { if let Ok(n) = stream.read(&mut buf) {
let _ = stream.write_all(&buf[..n]); 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 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_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()); 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 // Wait for HostLink and RelayLink to connect
let deadline = Instant::now() + Duration::from_secs(15); 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)); std::thread::sleep(Duration::from_millis(200));
} }
assert!(host_link.status().connected, "Host failed to connect to Coordinator"); assert!(
assert!(relay_link.status().connected, "Host failed to connect to Game Relay: {:?}", relay_link.status()); 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"); 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 // Player connects
let player_dir = tempfile::tempdir().unwrap(); let player_dir = tempfile::tempdir().unwrap();
let pc = PlayerControl::open(player_dir.path(), Arc::new(MemoryStore::default())).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 { let target = RealmTarget {
realm_id: realm, realm_id: realm,
@@ -358,39 +539,62 @@ fn gate_phase13_live_vps_relay() {
}; };
let mut ch = pc.connect(&target).expect("Player connect via WSS"); 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 // Allocate relay
let alloc = pc.allocate_relay(&mut ch).expect("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_eq!(alloc.relay_host, "coa-manager.duckdns.org");
assert!(alloc.auth_port >= 40000 && alloc.auth_port <= 43999); assert!(alloc.auth_port >= 40000 && alloc.auth_port <= 43999);
assert!(alloc.world_port >= 40000 && alloc.world_port <= 43999); assert!(alloc.world_port >= 40000 && alloc.world_port <= 43999);
assert_ne!(alloc.auth_port, alloc.world_port); assert_ne!(alloc.auth_port, alloc.world_port);
// Test AUTH traffic through VPS Relay: verify REALM_LIST address rewrite // 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"); let mut auth_client = TcpStream::connect((alloc.relay_host.as_str(), alloc.auth_port))
auth_client.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); .expect("connect to auth relay");
auth_client
.set_read_timeout(Some(Duration::from_secs(5)))
.unwrap();
let mut pkt_buf = vec![0u8; 512]; 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!(n > 10, "expected realm list packet");
assert_eq!(pkt_buf[0], 0x10, "opcode must be 0x10 (REALM_LIST)"); assert_eq!(pkt_buf[0], 0x10, "opcode must be 0x10 (REALM_LIST)");
let expected_world_addr = format!("{}:{}", alloc.relay_host, alloc.world_port); 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()), assert!(
"REALM_LIST packet must have address rewritten to {expected_world_addr}"); 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}"); println!("GATE 13 PASS AUTH stream rewrote REALM_LIST address to VPS world port {expected_world_addr}");
drop(auth_client); drop(auth_client);
// Test WORLD traffic through VPS Relay: verify dumb byte tunneling // 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"); let mut world_client = TcpStream::connect((alloc.relay_host.as_str(), alloc.world_port))
world_client.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); .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"; 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()]; let mut echo_buf = vec![0u8; test_payload.len()];
world_client.read_exact(&mut echo_buf).expect("read echo from world relay"); world_client
assert_eq!(&echo_buf, test_payload, "world relay must tunnel bytes faithfully without alteration"); .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"); println!("GATE 13 PASS WORLD stream tunneled game bytes faithfully end-to-end");
drop(world_client); drop(world_client);
@@ -398,7 +602,8 @@ fn gate_phase13_live_vps_relay() {
ch.close(); ch.close();
host_link.stop(); host_link.stop();
relay_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"); println!("GATE 13 PASS game relay gate passed and cleaned up");
} }
+45 -7
View File
@@ -72,6 +72,9 @@ pub(crate) fn start(
None None
} }
}; };
if let (Some(player), Some(portable)) = (&player, &portable) {
PlayerControl::start_remote_session_sync(player, portable);
}
let host = portable.and_then(|portable| { let host = portable.and_then(|portable| {
let store = match ControlStore::open(&base.join("host")) { let store = match ControlStore::open(&base.join("host")) {
Ok(s) => Arc::new(Mutex::new(s)), Ok(s) => Arc::new(Mutex::new(s)),
@@ -280,12 +283,46 @@ pub(crate) async fn join_realm(
out.status = "incompatible".into(); out.status = "incompatible".into();
return Ok(out); return Ok(out);
} }
let here = pc_bg.claimed_here(&rid_bg).map_err(from_control)?; let owned_realm =
if !here.iter().any(|(id, _)| id.to_string() == character) { coa_core::realm_registry::settings::load(&data_dir().join("registry"))
// Phase 12.1: Transfer canonical character to remote Host over the encrypted control channel .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 char_id_str = character.clone();
let bundle = let remote_realm = rid_bg.to_string();
portable.call(move |s| s.export_for_remote_transfer(&char_id_str))??; 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() { if bundle.ruleset.as_str() != detail.ruleset.as_str() {
ch.close(); ch.close();
return Err(fail( return Err(fail(
@@ -299,11 +336,12 @@ pub(crate) async fn join_realm(
} }
let cuuid = bundle.character_id.as_uuid(); let cuuid = bundle.character_id.as_uuid();
let _outcome = pc_bg let _outcome = pc_bg
.transfer_character( .transfer_owned_character(
&mut ch, &mut ch,
session,
cuuid, cuuid,
bundle.ruleset.as_str(), bundle.ruleset.as_str(),
1, revision,
&bundle.payload, &bundle.payload,
bundle.content_hash, bundle.content_hash,
bundle.collections, bundle.collections,