diff --git a/crates/coa-core/src/portable/ids.rs b/crates/coa-core/src/portable/ids.rs index cae612b..6618d3f 100644 --- a/crates/coa-core/src/portable/ids.rs +++ b/crates/coa-core/src/portable/ids.rs @@ -96,6 +96,10 @@ uuid7_id!( /// One import of a portable character into a realm (the key of the import journal). ImportId ); +uuid7_id!( + /// One runtime session of a portable character on a realm: from its automatic baseline to its final checkpoint. + SessionId +); /// A reference to a piece of game content that cannot be mistaken for another namespace's numbering: /// `core:wotlk:item:19019`, `coa:wardrobe:10581`, `mod::item:7`. diff --git a/crates/coa-core/src/portable/migrations/005_sessions.sql b/crates/coa-core/src/portable/migrations/005_sessions.sql new file mode 100644 index 0000000..5dea8b1 --- /dev/null +++ b/crates/coa-core/src/portable/migrations/005_sessions.sql @@ -0,0 +1,69 @@ +-- Schema 5: runtime portable sessions (Phase 5). The OWNER side and the HOST side keep their own tables; one Manager may +-- play both roles on one file in a prototype, but a deployment uses two files and the roles never read each other's. +-- +-- owner_session / owner_checkpoint : what the Owner has offered, started and applied for one session of one character on one +-- realm. C0 and B0 are kept as sealed snapshots (the anchors of every checkpoint merge) and +-- every applied sequence is logged (idempotency, ordering, lost acknowledgements). +-- host_session / host_outbox : what the Host has armed on its realm, the sequence it is at, the state it has already +-- read from the realm, and the messages not yet acknowledged. + +CREATE TABLE owner_session ( + session_id TEXT PRIMARY KEY, + character_id TEXT NOT NULL REFERENCES character(character_id) ON DELETE CASCADE, + server_id TEXT NOT NULL, + state TEXT NOT NULL CHECK (state IN ('offered', 'open', 'closed', 'superseded')), + c0_revision INTEGER NOT NULL, + c0_hash BLOB NOT NULL CHECK (length(c0_hash) = 32), + c0_payload BLOB NOT NULL, + baseline_generation INTEGER, + b0_hash BLOB CHECK (b0_hash IS NULL OR length(b0_hash) = 32), + b0_payload BLOB, + -- the canonical revision this session produced last (starts at c0_revision); the next checkpoint must still find it + head_revision INTEGER NOT NULL, + last_sequence INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + CHECK (state NOT IN ('open', 'closed') OR b0_hash IS NOT NULL) +) STRICT; +CREATE UNIQUE INDEX owner_session_one_live ON owner_session(character_id, server_id) WHERE state IN ('offered', 'open'); + +CREATE TABLE owner_checkpoint ( + session_id TEXT NOT NULL REFERENCES owner_session(session_id) ON DELETE CASCADE, + sequence INTEGER NOT NULL, + content_hash BLOB NOT NULL CHECK (length(content_hash) = 32), + resulting_revision INTEGER NOT NULL, + final INTEGER NOT NULL CHECK (final IN (0, 1)), + next_session_id TEXT, + applied_at TEXT NOT NULL, + PRIMARY KEY (session_id, sequence) +) STRICT; + +CREATE TABLE host_session ( + session_id TEXT PRIMARY KEY, + character_id TEXT NOT NULL REFERENCES character(character_id) ON DELETE CASCADE, + server_id TEXT NOT NULL, + local_guid INTEGER, + base_revision INTEGER NOT NULL, + generation INTEGER NOT NULL DEFAULT 1, + -- armed: the realm holds the marker, B0 not read yet; open: B0 sent, checkpoints running; closed + state TEXT NOT NULL CHECK (state IN ('armed', 'open', 'closed')), + -- the sequence the next checkpoint will use; persisted BEFORE the realm is asked to save + next_sequence INTEGER NOT NULL DEFAULT 1, + pending_sequence INTEGER, + acked_sequence INTEGER NOT NULL DEFAULT 0, + owned_items TEXT NOT NULL DEFAULT '[]', + owned_pets TEXT NOT NULL DEFAULT '[]', + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +) STRICT; +CREATE UNIQUE INDEX host_session_one_live ON host_session(character_id, server_id) WHERE state IN ('armed', 'open'); + +CREATE TABLE host_outbox ( + session_id TEXT NOT NULL REFERENCES host_session(session_id) ON DELETE CASCADE, + sequence INTEGER NOT NULL, + kind TEXT NOT NULL CHECK (kind IN ('started', 'checkpoint')), + message BLOB NOT NULL, + state TEXT NOT NULL CHECK (state IN ('pending', 'acked')), + created_at TEXT NOT NULL, + PRIMARY KEY (session_id, sequence) +) STRICT; diff --git a/crates/coa-core/src/portable/mod.rs b/crates/coa-core/src/portable/mod.rs index 81e8880..7f7e9ce 100644 --- a/crates/coa-core/src/portable/mod.rs +++ b/crates/coa-core/src/portable/mod.rs @@ -13,6 +13,7 @@ pub mod ids; pub mod merge; pub mod model; pub mod realm; +pub mod session; pub mod snapshot; pub mod store; pub mod versions; @@ -22,6 +23,6 @@ pub(crate) mod fixtures; pub use collection::IdSet; pub use error::{PortableError, Result}; -pub use ids::{CharacterId, ContentId, ImportId, PortableItemId, PortablePetId, ProfileId}; +pub use ids::{CharacterId, ContentId, ImportId, PortableItemId, PortablePetId, ProfileId, SessionId}; pub use model::{PortableCharacter, Ruleset}; pub use store::Store; diff --git a/crates/coa-core/src/portable/session/bridge.rs b/crates/coa-core/src/portable/session/bridge.rs new file mode 100644 index 0000000..2e0740c --- /dev/null +++ b/crates/coa-core/src/portable/session/bridge.rs @@ -0,0 +1,78 @@ +//! What the Host needs from a realm to run a session: the core's session marker, the checkpoint/release operations and a +//! read-only export. `LiveBridge` (see `live.rs`) implements it with the Manager's `Db` and RA; tests implement it with a +//! simulated realm. + +use std::collections::HashMap; + +use super::super::error::Result; +use super::super::ids::{CharacterId, PortableItemId, PortablePetId, SessionId}; +use super::super::realm::Exported; + +/// The core's `coa_portable_session.state`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RowState { + /// Armed by the importer/updater: the character's first load will take the baseline. + WaitingBaseline, + /// The baseline transaction was written; the player is held until the Host releases the session. + BaselineReady, + /// Released: the session runs. + Active, + /// The logout save was written. + Ended, +} + +impl RowState { + pub fn from_code(code: u64) -> Option { + Some(match code { + 0 => RowState::WaitingBaseline, + 1 => RowState::BaselineReady, + 2 => RowState::Active, + 3 => RowState::Ended, + _ => return None, + }) + } +} + +/// One row of the core's session table. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SessionRow { + pub guid: u32, + pub session_id: SessionId, + pub character_id: CharacterId, + pub imported_revision: u64, + pub generation: u32, + pub state: RowState, + pub checkpoint_seq: u64, + pub save_seq: u64, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum CheckpointReply { + /// The save was queued. This is **not** persistence: the marker row has to be seen. + Queued, + /// The character is being teleported or otherwise cannot be saved right now: try again. + Busy, + NotOnline, + /// The core refused (wrong session, stale sequence, not a portable character). + Refused(String), +} + +/// A read of the realm: the character as committed, and the session row **of the same consistent snapshot**. +pub struct RealmRead { + pub exported: Exported, + pub session: Option, + pub online: bool, +} + +pub trait RealmBridge { + /// Cheap poll of one character's session row. + fn session_row(&mut self, local_guid: u32) -> Result>; + /// Read-only export of the committed character. Allowed while the character is online **if** it has a session row. + fn read(&mut self, local_guid: u32, prior_items: &HashMap, prior_pets: &HashMap) -> Result; + /// `portable checkpoint ` + fn request_checkpoint(&mut self, local_guid: u32, session: SessionId, sequence: u64) -> Result; + /// `portable release ` + fn release(&mut self, session: SessionId) -> Result<()>; + /// Put the character's marker back to "waiting for baseline" with a new session (the character is offline). + fn arm(&mut self, local_guid: u32, session: SessionId, character: CharacterId, revision: u64, generation: u32) -> Result<()>; +} diff --git a/crates/coa-core/src/portable/session/fake.rs b/crates/coa-core/src/portable/session/fake.rs new file mode 100644 index 0000000..9ad5d3d --- /dev/null +++ b/crates/coa-core/src/portable/session/fake.rs @@ -0,0 +1,259 @@ +//! A simulated realm for the session tests: the core's marker, gate, checkpoint and logout behaviour, and a realm that normalises +//! characters on its own. It is deliberately small; the real worldserver is exercised by `live` tests. + +use std::collections::HashMap; + +use super::super::error::{PortableError, Result}; +use super::super::identity::item_identity; +use super::super::ids::{CharacterId, ContentId, PortableItemId, PortablePetId, SessionId}; +use super::super::model::*; +use super::super::realm::Exported; +use super::super::store::{pet_identity, ImportAllocation, ItemObservation, PetObservation, PlannedItem, PlannedPet, Store}; +use super::bridge::*; + +pub struct FakeRealm { + pub guid: u32, + /// What the player is playing right now. + pub memory: PortableCharacter, + /// What the realm's tables hold (the last save). + pub db: PortableCharacter, + item_guid: HashMap, + pet_number: HashMap, + next_item_guid: u32, + next_pet_number: u32, + pub row: Option, + pub online: bool, + /// The core holds the player until the Host releases the session. + pub gated: bool, + /// Checkpoint requests that answer `Busy` before one succeeds. + pub busy: u32, + pub saves: u32, + pub released: Vec, + pub normalize: fn(&mut FakeRealm), +} + +pub fn new_internal_item(n: u64, slot: u8, entry: u64) -> PortableItem { + PortableItem { + id: PortableItemId::from_uuid(crate::portable::fixtures::id7(7_000_000 + n)).unwrap(), + container: None, + slot, + entry: ContentId::new("coa", "item", entry).unwrap(), + count: 1, + duration: 0, + charges: vec![], + flags: 0, + enchantments: vec![], + random_property_id: 0, + durability: 30, + played_time: 0, + text: None, + creator_name: None, + gift: None, + } +} + +impl FakeRealm { + pub fn new(guid: u32) -> Self { + let empty = crate::portable::fixtures::naked_level_one(); + Self { + guid, + memory: empty.clone(), + db: empty, + item_guid: HashMap::new(), + pet_number: HashMap::new(), + next_item_guid: 50_000, + next_pet_number: 900, + row: None, + online: false, + gated: false, + busy: 0, + saves: 0, + released: vec![], + normalize: |_| {}, + } + } + + /// The character arrives (an import): items and pets get local ids in plan order, like the offline importer does. + pub fn import(&mut self, store: &mut Store, character: CharacterId, server_id: &str, revision: u64, session: SessionId) -> Result<()> { + let model = store.load_snapshot(character, revision)?; + let items: Vec = model.items.iter().map(|i| PlannedItem { id: i.id, entry: i.entry.clone(), identity: item_identity(&i.entry, i.random_property_id) }).collect(); + let pets: Vec = model.pets.iter().map(|p| PlannedPet { id: p.id, entry: p.entry.clone(), identity: pet_identity(&p.entry, p.pet_type, p.created_by_spell) }).collect(); + let ticket = store.begin_import(character, server_id, revision, &items, &pets)?; + let (item_base, pet_base) = (self.next_item_guid, self.next_pet_number); + self.next_item_guid += items.len() as u32; + self.next_pet_number += pets.len() as u32; + for (i, item) in items.iter().enumerate() { + self.item_guid.insert(item.id, item_base + i as u32); + } + for (i, pet) in pets.iter().enumerate() { + self.pet_number.insert(pet.id, pet_base + i as u32); + } + self.db = model.clone(); + self.memory = model; + store.finish_import(ticket.import_id, ImportAllocation { local_guid: self.guid, item_base, pet_base })?; + self.row = Some(SessionRow { guid: self.guid, session_id: session, character_id: character, imported_revision: revision, generation: 1, state: RowState::WaitingBaseline, checkpoint_seq: 0, save_seq: 0 }); + Ok(()) + } + + fn free_slot(&self) -> u8 { + (23u8..=38).chain(39..=66).find(|s| !self.memory.items.iter().any(|i| i.container.is_none() && i.slot == *s)).expect("a free slot") + } + + /// The realm gives the character an item; it goes to the first free place. + pub fn give_item(&mut self, mut item: PortableItem) -> u32 { + item.slot = self.free_slot(); + let guid = self.next_item_guid; + self.next_item_guid += 1; + self.item_guid.insert(item.id, guid); + self.memory.items.push(item); + guid + } + + fn save(&mut self) { + self.db = self.memory.clone().normalized(); + self.saves += 1; + if let Some(r) = &mut self.row { + r.save_seq += 1; + } + } + + /// The player logs in: the realm normalises, and a portable character in `WaitingBaseline` is saved and held. + pub fn login(&mut self) { + assert!(!self.online); + self.online = true; + self.memory = self.db.clone(); + (self.normalize)(self); + let waiting = matches!(&self.row, Some(r) if matches!(r.state, RowState::WaitingBaseline | RowState::BaselineReady)); + if waiting { + self.save(); + let row = self.row.as_mut().unwrap(); + row.state = RowState::BaselineReady; + self.gated = true; + } else if let Some(r) = &mut self.row { + if r.state == RowState::Ended { + r.state = RowState::Active; + } + } + } + + /// Gameplay. It is a test failure to play before the baseline was released: that is the property the gate guarantees. + pub fn play(&mut self, f: impl FnOnce(&mut PortableCharacter)) { + assert!(self.online, "the player is not online"); + assert!(!self.gated, "gameplay happened before B0 was taken"); + f(&mut self.memory); + } + + pub fn autosave(&mut self) { + assert!(self.online); + self.save(); + } + + pub fn logout(&mut self) { + assert!(self.online); + self.save(); + self.online = false; + self.gated = false; + if let Some(r) = &mut self.row { + r.state = RowState::Ended; + } + } + + /// The worldserver or the client dies: the last save stays, nothing else. + pub fn crash(&mut self) { + self.online = false; + self.gated = false; + self.memory = self.db.clone(); + } + + /// The gate's timeout: the player is disconnected, never released. + pub fn gate_timeout(&mut self) { + assert!(self.gated); + self.online = false; + self.gated = false; + } +} + +impl RealmBridge for FakeRealm { + fn session_row(&mut self, local_guid: u32) -> Result> { + Ok(if local_guid == self.guid { self.row.clone() } else { None }) + } + + fn read(&mut self, local_guid: u32, prior_items: &HashMap, prior_pets: &HashMap) -> Result { + assert_eq!(local_guid, self.guid); + let mut model = self.db.clone(); + let mut ids: HashMap = HashMap::new(); + let mut observations = Vec::new(); + for item in &mut model.items { + let internal = item.id; + let guid = *self.item_guid.get(&internal).ok_or_else(|| PortableError::Invalid("the simulated realm lost track of an item".into()))?; + let identity = item_identity(&item.entry, item.random_property_id); + let id = match prior_items.get(&guid) { + Some((id, prior)) if *prior == identity => *id, + _ => PortableItemId::new(), + }; + ids.insert(internal, id); + item.id = id; + observations.push(ItemObservation { portable_item_id: id, local_item_guid: guid, entry: item.entry.clone(), identity }); + } + for item in &mut model.items { + item.container = item.container.and_then(|c| ids.get(&c).copied()); + } + let mut pet_observations = Vec::new(); + for pet in &mut model.pets { + let internal = pet.id; + let number = *self.pet_number.entry(internal).or_insert_with(|| { + self.next_pet_number += 1; + self.next_pet_number + }); + let identity = pet_identity(&pet.entry, pet.pet_type, pet.created_by_spell); + let id = match prior_pets.get(&number) { + Some((id, prior)) if *prior == identity => *id, + _ => PortablePetId::new(), + }; + pet.id = id; + pet_observations.push(PetObservation { portable_pet_id: id, local_pet_number: number, identity }); + } + // the realm's tables are keyed by guid: a later read finds the same rows under the same internal ids + let model = model.normalized(); + Ok(RealmRead { exported: Exported { model, local_guid: self.guid, account: 1, observations, pet_observations, warnings: vec![] }, session: self.row.clone(), online: self.online }) + } + + fn request_checkpoint(&mut self, local_guid: u32, session: SessionId, sequence: u64) -> Result { + if local_guid != self.guid || !self.online { + return Ok(CheckpointReply::NotOnline); + } + let Some(row) = &self.row else { return Ok(CheckpointReply::Refused("not a portable character".into())) }; + if row.session_id != session || row.state != RowState::Active { + return Ok(CheckpointReply::Refused("wrong session".into())); + } + if sequence <= row.checkpoint_seq { + return Ok(CheckpointReply::Refused("stale sequence".into())); + } + if self.busy > 0 { + self.busy -= 1; + return Ok(CheckpointReply::Busy); + } + self.save(); + self.row.as_mut().unwrap().checkpoint_seq = sequence; + Ok(CheckpointReply::Queued) + } + + fn release(&mut self, session: SessionId) -> Result<()> { + if let Some(r) = &mut self.row { + if r.session_id == session && r.state == RowState::BaselineReady { + r.state = RowState::Active; + self.gated = false; + self.released.push(session); + } + } + Ok(()) + } + + fn arm(&mut self, local_guid: u32, session: SessionId, character: CharacterId, revision: u64, generation: u32) -> Result<()> { + assert_eq!(local_guid, self.guid); + assert!(!self.online, "a character is re-armed while it is offline"); + let save_seq = self.row.as_ref().map_or(0, |r| r.save_seq); + self.row = Some(SessionRow { guid: local_guid, session_id: session, character_id: character, imported_revision: revision, generation, state: RowState::WaitingBaseline, checkpoint_seq: 0, save_seq }); + Ok(()) + } +} diff --git a/crates/coa-core/src/portable/session/host.rs b/crates/coa-core/src/portable/session/host.rs new file mode 100644 index 0000000..f9253af --- /dev/null +++ b/crates/coa-core/src/portable/session/host.rs @@ -0,0 +1,199 @@ +//! The Host's session driver: arms sessions, takes the automatic baseline, runs checkpoints and queues the messages the +//! Owner must receive. It never decides what the canonical character becomes and never opens the Owner's database. + +use std::collections::HashMap; + +use super::super::error::{PortableError, Result}; +use super::super::ids::{CharacterId, PortableItemId, PortablePetId, ProfileId, SessionId}; +use super::super::store::{AckEffect, HostSession, HostState, OutboxMessage, Store}; +use super::bridge::{CheckpointReply, RealmBridge, RealmRead, RowState}; +use super::protocol::*; + +#[derive(Debug, Clone)] +pub struct HostConfig { + /// Seconds between checkpoints of a running session. + pub checkpoint_interval_secs: u64, +} + +impl Default for HostConfig { + fn default() -> Self { + Self { checkpoint_interval_secs: 60 } + } +} + +/// What one tick did, for logs and tests. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum HostEvent { + /// `B0` was read and `PortableSessionStarted` queued; the player is released. + BaselineTaken { session: SessionId }, + /// The realm held the player but the baseline could not be taken yet. + BaselineRetry { session: SessionId, why: String }, + /// The realm released the session without the Host (nothing can be done: the baseline is lost). + BaselineMissed { session: SessionId }, + CheckpointRequested { session: SessionId, sequence: u64 }, + CheckpointBusy { session: SessionId, sequence: u64 }, + CheckpointQueued { session: SessionId, sequence: u64, final_checkpoint: bool }, + /// The realm shows another session than the Host armed: the Host's session is closed. + SessionReplaced { session: SessionId }, + Refused { session: SessionId, why: String }, +} + +pub struct HostService<'a> { + store: &'a mut Store, + server_id: String, + config: HostConfig, + /// When the last checkpoint of each session was requested (not persisted: after a restart the first tick checkpoints). + last_checkpoint: HashMap, +} + +impl<'a> HostService<'a> { + pub fn new(store: &'a mut Store, server_id: &str, config: HostConfig) -> Self { + Self { store, server_id: server_id.to_string(), config, last_checkpoint: HashMap::new() } + } + + pub fn store(&self) -> &Store { + self.store + } + + /// Hand the Host an offer: it keeps a copy of the character and prepares the session. The caller then imports the + /// character into the realm (offline importer or the core's import service) with this session id armed, and calls + /// [`HostService::bind`] with the realm's local guid. + pub fn accept_offer(&mut self, profile: ProfileId, offer: &SessionOffer) -> Result { + if offer.protocol_version != PROTOCOL_VERSION { + return Err(PortableError::UnsupportedFormat { found: offer.protocol_version, supported: PROTOCOL_VERSION }); + } + if offer.server_id != self.server_id { + return Err(PortableError::Invalid("the offer is for another realm".into())); + } + let model = offer.snapshot.open()?; + if model.character_id != offer.character_id { + return Err(PortableError::WrongCharacter { expected: offer.character_id, found: model.character_id }); + } + self.store.host_install_copy(profile, &model, offer.canonical_revision, "owner")?; + self.store.host_prepare_session(offer, &model) + } + + /// The character exists on the realm with this guid. + pub fn bind(&mut self, session: SessionId, local_guid: u32) -> Result<()> { + self.store.host_bind_session(session, local_guid) + } + + /// One pass over every live session of this realm. Safe to call as often as wanted; every step is idempotent and resumable. + pub fn tick(&mut self, bridge: &mut dyn RealmBridge, now_secs: u64) -> Result> { + let mut events = Vec::new(); + for s in self.store.host_live_sessions(&self.server_id)? { + let Some(guid) = s.local_guid else { continue }; + let Some(row) = bridge.session_row(guid)? else { continue }; + if row.session_id != s.session_id { + if s.state == HostState::Open { + events.push(HostEvent::SessionReplaced { session: s.session_id }); + } + continue; + } + match (s.state, row.state) { + (HostState::Armed, RowState::BaselineReady) => self.take_baseline(bridge, &s, guid, &mut events)?, + (HostState::Armed, RowState::Active | RowState::Ended) => events.push(HostEvent::BaselineMissed { session: s.session_id }), + (HostState::Open, RowState::Active) => self.run_checkpoint(bridge, &s, guid, row.checkpoint_seq, false, now_secs, &mut events)?, + (HostState::Open, RowState::Ended) => self.run_checkpoint(bridge, &s, guid, row.checkpoint_seq, true, now_secs, &mut events)?, + _ => {} + } + } + Ok(events) + } + + fn priors(&self, s: &HostSession) -> Result<(HashMap, HashMap)> { + Ok((self.store.active_item_lookup(s.character_id, &s.server_id)?, self.store.active_pet_lookup(s.character_id, &s.server_id)?)) + } + + fn take_baseline(&mut self, bridge: &mut dyn RealmBridge, s: &HostSession, guid: u32, events: &mut Vec) -> Result<()> { + let (items, pets) = self.priors(s)?; + let RealmRead { exported, session, .. } = bridge.read(guid, &items, &pets)?; + // the baseline counts only if the very snapshot it came from still shows the baseline marker of this session + let Some(row) = session.filter(|r| r.session_id == s.session_id && r.state == RowState::BaselineReady) else { + events.push(HostEvent::BaselineRetry { session: s.session_id, why: "the marker moved while the baseline was being read".into() }); + return Ok(()); + }; + let msg = PortableSessionStarted::new(s.session_id, s.character_id, &s.server_id, s.base_revision, row.generation, &exported.model)?; + self.store.host_queue_started(s.session_id, &msg, &exported.model, &exported.observations, &exported.pet_observations)?; + // the baseline is persisted (and queued): only now may the player move on + bridge.release(s.session_id)?; + events.push(HostEvent::BaselineTaken { session: s.session_id }); + Ok(()) + } + + #[allow(clippy::too_many_arguments)] + fn run_checkpoint(&mut self, bridge: &mut dyn RealmBridge, s: &HostSession, guid: u32, marker_seq: u64, final_checkpoint: bool, now_secs: u64, events: &mut Vec) -> Result<()> { + // one checkpoint at a time: the last one has to be queued before another is reserved + let due = s.pending_sequence.is_some() || final_checkpoint || self.last_checkpoint.get(&s.session_id).is_none_or(|t| now_secs >= *t + self.config.checkpoint_interval_secs); + if !due { + return Ok(()); + } + let sequence = self.store.host_begin_checkpoint(s.session_id)?; + if !final_checkpoint && marker_seq < sequence { + match bridge.request_checkpoint(guid, s.session_id, sequence)? { + CheckpointReply::Queued => { + self.last_checkpoint.insert(s.session_id, now_secs); + events.push(HostEvent::CheckpointRequested { session: s.session_id, sequence }); + } + CheckpointReply::Busy => events.push(HostEvent::CheckpointBusy { session: s.session_id, sequence }), + CheckpointReply::NotOnline => {} + CheckpointReply::Refused(why) => events.push(HostEvent::Refused { session: s.session_id, why }), + } + // the marker is read from the database, never assumed from the reply + return Ok(()); + } + let (items, pets) = self.priors(s)?; + let RealmRead { exported, session, .. } = bridge.read(guid, &items, &pets)?; + let Some(row) = session.filter(|r| r.session_id == s.session_id) else { return Ok(()) }; + if !final_checkpoint && row.checkpoint_seq < sequence { + return Ok(()); + } + if final_checkpoint && row.state != RowState::Ended { + return Ok(()); + } + let msg = PortableCheckpoint::new(s.session_id, s.character_id, &s.server_id, s.base_revision, sequence, final_checkpoint, &exported.model)?; + self.store.host_queue_checkpoint(s.session_id, &msg, &exported.observations, &exported.pet_observations)?; + self.last_checkpoint.insert(s.session_id, now_secs); + events.push(HostEvent::CheckpointQueued { session: s.session_id, sequence, final_checkpoint }); + Ok(()) + } + + /// Messages waiting for delivery, in order. + pub fn outbox(&self) -> Result> { + self.store.host_outbox_pending(&self.server_id) + } + + /// Apply an acknowledgement. When it ends a session, the realm's marker is re-armed for the session that follows. + pub fn receive_ack(&mut self, bridge: &mut dyn RealmBridge, ack: &OwnerAck) -> Result { + let session = self.store.host_session(ack.session_id)?; + let effect = self.store.host_receive_ack(ack)?; + if let (AckEffect::Finished { next_session, next_revision }, Some(old)) = (&effect, session) { + self.rearm(bridge, old.character_id, *next_session, *next_revision)?; + } + Ok(effect) + } + + /// Re-arm the realm for every armed session whose marker is not on the realm yet (after a crash between the local commit + /// of a final acknowledgement and the realm write). + pub fn rearm_pending(&mut self, bridge: &mut dyn RealmBridge) -> Result { + let mut n = 0; + for s in self.store.host_live_sessions(&self.server_id)? { + let (Some(guid), HostState::Armed) = (s.local_guid, s.state) else { continue }; + let current = bridge.session_row(guid)?; + if current.as_ref().is_some_and(|r| r.session_id == s.session_id) { + continue; + } + if s.generation > 1 { + bridge.arm(guid, s.session_id, s.character_id, s.base_revision, s.generation)?; + n += 1; + } + } + Ok(n) + } + + fn rearm(&mut self, bridge: &mut dyn RealmBridge, character: CharacterId, session: SessionId, revision: u64) -> Result<()> { + let s = self.store.host_session(session)?.ok_or_else(|| PortableError::Invalid("the next session was not recorded".into()))?; + let guid = s.local_guid.ok_or_else(|| PortableError::Invalid("the next session has no realm character".into()))?; + bridge.arm(guid, session, character, revision, s.generation) + } +} diff --git a/crates/coa-core/src/portable/session/mod.rs b/crates/coa-core/src/portable/session/mod.rs new file mode 100644 index 0000000..ddc8545 --- /dev/null +++ b/crates/coa-core/src/portable/session/mod.rs @@ -0,0 +1,12 @@ +pub mod protocol; +pub mod bridge; +pub mod host; +pub mod owner; + +pub use host::{HostConfig, HostEvent, HostService}; +pub use owner::OwnerService; + +#[cfg(test)] +mod fake; +#[cfg(test)] +mod tests; diff --git a/crates/coa-core/src/portable/session/owner.rs b/crates/coa-core/src/portable/session/owner.rs new file mode 100644 index 0000000..0ddf6cb --- /dev/null +++ b/crates/coa-core/src/portable/session/owner.rs @@ -0,0 +1,41 @@ +//! The Owner's entry points: the only place a message from a realm enters the canonical store. + +use super::super::error::Result; +use super::super::ids::{CharacterId, SessionId}; +use super::super::store::Store; +use super::protocol::*; + +pub struct OwnerService<'a> { + store: &'a mut Store, +} + +impl<'a> OwnerService<'a> { + pub fn new(store: &'a mut Store) -> Self { + Self { store } + } + + pub fn store(&self) -> &Store { + self.store + } + + /// Offer the character's current canonical revision to a realm and open the session that follows it. + pub fn offer(&mut self, character: CharacterId, server_id: &str) -> Result { + self.store.owner_offer_session(character, server_id) + } + + /// `PortableSessionStarted` as bytes from a transport. A message that cannot be parsed is an error and changes nothing. + pub fn handle_started(&mut self, bytes: &[u8]) -> Result { + let msg: PortableSessionStarted = from_json(bytes)?; + self.store.owner_session_started(&msg) + } + + /// `PortableCheckpoint` as bytes from a transport. + pub fn handle_checkpoint(&mut self, bytes: &[u8]) -> Result { + let msg: PortableCheckpoint = from_json(bytes)?; + self.store.owner_checkpoint(&msg) + } + + pub fn session(&self, session: SessionId) -> Result> { + self.store.owner_session_info(session) + } +} diff --git a/crates/coa-core/src/portable/session/protocol.rs b/crates/coa-core/src/portable/session/protocol.rs new file mode 100644 index 0000000..5f6df96 --- /dev/null +++ b/crates/coa-core/src/portable/session/protocol.rs @@ -0,0 +1,181 @@ +//! The messages of a runtime portable session. Transport-neutral: a Registry/Relay later only has to carry these bytes. +//! +//! Every message is canonical JSON with unknown fields refused; snapshots travel in the existing snapshot envelope +//! (zstd of canonical JSON, SHA-256 of the canonical JSON). Size limits are checked before anything is parsed. + +use base64::Engine; +use serde::de::DeserializeOwned; +use serde::{Deserialize, Serialize}; + +use super::super::error::{PortableError, Result}; +use super::super::ids::{CharacterId, PortableItemId, PortablePetId, SessionId}; +use super::super::model::{limits::MAX_COMPRESSED_BYTES, PortableCharacter}; +use super::super::snapshot; + +pub const PROTOCOL_VERSION: u32 = 1; +/// A message carries at most one compressed snapshot (base64: 4/3) plus a little structure. +pub const MAX_MESSAGE_BYTES: usize = MAX_COMPRESSED_BYTES * 4 / 3 + 64 * 1024; +/// The most item/pet ids an acknowledgement lists. +pub const MAX_ACK_IDS: usize = 20_000; + +/// A sealed snapshot: the hash of the canonical JSON and the compressed bytes. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Envelope { + pub content_hash: String, + pub payload: String, +} + +impl Envelope { + pub fn seal(model: &PortableCharacter) -> Result { + let encoded = snapshot::encode(model)?; + Ok(Envelope { content_hash: hex::encode(encoded.content_hash), payload: base64::engine::general_purpose::STANDARD.encode(&encoded.payload) }) + } + + pub fn from_encoded(encoded: &snapshot::EncodedSnapshot) -> Envelope { + Envelope { content_hash: hex::encode(encoded.content_hash), payload: base64::engine::general_purpose::STANDARD.encode(&encoded.payload) } + } + + pub fn hash(&self) -> Result<[u8; 32]> { + let bytes = hex::decode(&self.content_hash).map_err(|_| PortableError::Invalid("the content hash is not hex".into()))?; + <[u8; 32]>::try_from(bytes.as_slice()).map_err(|_| PortableError::Invalid("the content hash is not 32 bytes".into())) + } + + /// Decode and fully verify (size caps while inflating, hash, structure). + pub fn open(&self) -> Result { + if self.payload.len() > MAX_COMPRESSED_BYTES * 4 / 3 + 8 { + return Err(PortableError::LimitExceeded("the snapshot payload is over the limit".into())); + } + let bytes = base64::engine::general_purpose::STANDARD.decode(&self.payload).map_err(|_| PortableError::Invalid("the snapshot payload is not base64".into()))?; + snapshot::decode(&bytes, Some(&self.hash()?)) + } + + /// The raw compressed bytes, for storage next to their hash. + pub fn bytes(&self) -> Result> { + base64::engine::general_purpose::STANDARD.decode(&self.payload).map_err(|_| PortableError::Invalid("the snapshot payload is not base64".into())) + } +} + +/// Owner -> Host: a character to put on the realm, and the session that will follow it. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SessionOffer { + pub protocol_version: u32, + pub session_id: SessionId, + pub character_id: CharacterId, + pub server_id: String, + pub canonical_revision: u64, + pub snapshot: Envelope, +} + +/// Host -> Owner: the realm's own first load/save of the character was captured, before any gameplay. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct PortableSessionStarted { + pub protocol_version: u32, + pub session_id: SessionId, + pub character_id: CharacterId, + pub server_id: String, + pub base_canonical_revision: u64, + pub baseline_generation: u32, + pub b0: Envelope, + pub content_hash: String, +} + +/// Host -> Owner: the realm's state now. `sequence` is monotonic per session; `final_checkpoint` ends it. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct PortableCheckpoint { + pub protocol_version: u32, + pub session_id: SessionId, + pub character_id: CharacterId, + pub server_id: String, + pub base_canonical_revision: u64, + pub sequence: u64, + pub final_checkpoint: bool, + pub realm_snapshot: Envelope, + pub content_hash: String, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields, tag = "status", content = "reason", rename_all = "snake_case")] +pub enum AckOutcome { + Applied, + /// The same sequence with the same content was applied before; the original result is repeated. + Duplicate, + /// An older sequence than the Owner already holds: ignored. + StaleSequence, + /// The session was superseded (another session or revision moved the character): nothing is accepted. + StaleSession, + Rejected(String), +} + +impl AckOutcome { + /// The Owner holds this state (applied now or before). + pub fn accepted(&self) -> bool { + matches!(self, AckOutcome::Applied | AckOutcome::Duplicate) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct NextSession { + pub session_id: SessionId, + pub canonical_revision: u64, +} + +/// Owner -> Host. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct OwnerAck { + pub protocol_version: u32, + pub session_id: SessionId, + /// 0 acknowledges `PortableSessionStarted`. + pub sequence: u64, + pub outcome: AckOutcome, + pub canonical_revision: u64, + pub canonical_hash: String, + /// The items and pets the canonical character owns after this checkpoint (what the Host maps as `present`). + pub owned_items: Vec, + pub owned_pets: Vec, + /// Only on the acknowledgement of a final checkpoint: the canonical character, which the realm is now synchronised with. + pub canonical: Option, + /// Only on the acknowledgement of a final checkpoint: the session that follows. + pub next_session: Option, +} + +pub fn to_json(message: &T) -> Result> { + let bytes = serde_json::to_vec(message)?; + if bytes.len() > MAX_MESSAGE_BYTES { + return Err(PortableError::LimitExceeded(format!("a message is limited to {MAX_MESSAGE_BYTES} bytes"))); + } + Ok(bytes) +} + +/// Parse one message. The size is checked before the bytes are looked at, unknown fields are refused, the protocol version +/// is checked before anything else is trusted. +pub fn from_json(bytes: &[u8]) -> Result { + if bytes.len() > MAX_MESSAGE_BYTES { + return Err(PortableError::LimitExceeded(format!("a message is limited to {MAX_MESSAGE_BYTES} bytes"))); + } + let value: serde_json::Value = serde_json::from_slice(bytes)?; + let version = value.get("protocol_version").and_then(|v| v.as_u64()).ok_or_else(|| PortableError::Invalid("the message has no protocol version".into()))?; + if version != PROTOCOL_VERSION as u64 { + return Err(PortableError::UnsupportedFormat { found: version.min(u32::MAX as u64) as u32, supported: PROTOCOL_VERSION }); + } + Ok(serde_json::from_value(value)?) +} + +impl PortableSessionStarted { + pub fn new(session_id: SessionId, character_id: CharacterId, server_id: &str, base_canonical_revision: u64, baseline_generation: u32, b0: &PortableCharacter) -> Result { + let b0 = Envelope::seal(b0)?; + Ok(Self { protocol_version: PROTOCOL_VERSION, session_id, character_id, server_id: server_id.to_string(), base_canonical_revision, baseline_generation, content_hash: b0.content_hash.clone(), b0 }) + } +} + +impl PortableCheckpoint { + pub fn new(session_id: SessionId, character_id: CharacterId, server_id: &str, base_canonical_revision: u64, sequence: u64, final_checkpoint: bool, b1: &PortableCharacter) -> Result { + let realm_snapshot = Envelope::seal(b1)?; + Ok(Self { protocol_version: PROTOCOL_VERSION, session_id, character_id, server_id: server_id.to_string(), base_canonical_revision, sequence, final_checkpoint, content_hash: realm_snapshot.content_hash.clone(), realm_snapshot }) + } +} diff --git a/crates/coa-core/src/portable/session/tests.rs b/crates/coa-core/src/portable/session/tests.rs new file mode 100644 index 0000000..99d2ea4 --- /dev/null +++ b/crates/coa-core/src/portable/session/tests.rs @@ -0,0 +1,550 @@ +//! The Owner and the Host as two separate stores, a simulated realm between them, and the messages carried by hand. + +use super::bridge::*; +use super::fake::*; +use super::protocol::*; +use super::*; +use crate::portable::fixtures::geared_level_eighty; +use crate::portable::ids::{CharacterId, PortableItemId, SessionId}; +use crate::portable::model::*; +use crate::portable::store::{HostState, Store}; + +const SERVER: &str = "realm-b"; +const GUID: u32 = 4242; + +struct Setup { + owner: Store, + host: Store, + realm: FakeRealm, + id: CharacterId, + session: SessionId, + c0: PortableCharacter, +} + +/// What a realm does by itself when it loads a character: default spells, resets, items it holds back, items of its own. +fn normalizes(r: &mut FakeRealm) { + r.memory.build.spells.extend([(6654, 255), (23250, 255)]); + r.memory.progression.honor.today_honor = 0; + r.memory.progression.chosen_title = 0; + r.memory.progression.known_currencies |= 1 << 40; + let held: Vec = plain(&r.memory).into_iter().take(4).collect(); + r.memory.items.retain(|i| !held.contains(&i.id)); + r.give_item(new_internal_item(1, 30, 66_001)); + r.give_item(new_internal_item(2, 31, 66_002)); +} + +fn plain(m: &PortableCharacter) -> Vec { + let containers: std::collections::HashSet = m.items.iter().filter_map(|i| i.container).collect(); + m.items.iter().filter(|i| i.container.is_none() && !containers.contains(&i.id) && (23..40).contains(&i.slot)).map(|i| i.id).collect() +} + +/// Canonical revision 10 on the Owner; the realm has the character imported and armed, not yet logged in. +fn setup() -> Setup { + setup_in(Store::open_in_memory().unwrap(), Store::open_in_memory().unwrap()) +} + +fn setup_in(mut owner: Store, mut host: Store) -> Setup { + let profile = owner.default_profile().unwrap(); + let id = owner.create_character(profile, geared_level_eighty(), "realm-a").unwrap(); + for _ in 2..=10 { + let mut m = owner.load_current(id).unwrap(); + m.progression.xp += 1; + let rev = owner.character(id).unwrap().revision; + owner.commit_snapshot(id, rev, m, "realm-a", None).unwrap(); + } + assert_eq!(owner.character(id).unwrap().revision, 10); + let c0 = owner.load_current(id).unwrap(); + let offer = OwnerService::new(&mut owner).offer(id, SERVER).unwrap(); + let host_profile = host.default_profile().unwrap(); + let mut realm = FakeRealm::new(GUID); + realm.normalize = normalizes; + { + let mut h = HostService::new(&mut host, SERVER, HostConfig::default()); + h.accept_offer(host_profile, &offer).unwrap(); + realm.import(&mut host, id, SERVER, offer.canonical_revision, offer.session_id).unwrap(); + HostService::new(&mut host, SERVER, HostConfig::default()).bind(offer.session_id, GUID).unwrap(); + } + Setup { owner, host, realm, id, session: offer.session_id, c0 } +} + +/// Carry every queued message to the Owner and its answer back. +fn deliver(host: &mut HostService, owner: &mut OwnerService, realm: &mut FakeRealm) -> Vec { + let mut acks = Vec::new(); + for m in host.outbox().unwrap() { + let ack = if m.started { owner.handle_started(&m.bytes) } else { owner.handle_checkpoint(&m.bytes) }.unwrap(); + host.receive_ack(realm, &ack).unwrap(); + acks.push(ack); + } + acks +} + +fn revision(owner: &OwnerService, id: CharacterId) -> u64 { + owner.store().character(id).unwrap().revision +} + +/// One checkpoint: request, then wait for the marker and read. +fn checkpoint(host: &mut HostService, owner: &mut OwnerService, realm: &mut FakeRealm, now: u64) -> Vec { + let mut events = host.tick(realm, now).unwrap(); + events.extend(host.tick(realm, now + 1).unwrap()); + deliver(host, owner, realm); + events +} + +#[test] +fn the_whole_session_from_an_automatic_baseline_to_the_final_checkpoint() { + let Setup { mut owner, mut host, mut realm, id, session, c0 } = setup(); + let mut h = HostService::new(&mut host, SERVER, HostConfig::default()); + let mut o = OwnerService::new(&mut owner); + + // the player logs in; the realm normalises and holds the player; the Host takes B0 without anybody doing anything + realm.login(); + assert!(realm.gated, "the core holds the player until B0 exists"); + let events = h.tick(&mut realm, 0).unwrap(); + assert_eq!(events, vec![HostEvent::BaselineTaken { session }]); + assert!(!realm.gated, "released only after B0 was persisted"); + let acks = deliver(&mut h, &mut o, &mut realm); + assert_eq!((acks.len(), &acks[0].outcome), (1, &AckOutcome::Applied)); + assert_eq!(o.session(session).unwrap().unwrap().0, "open"); + assert_eq!(revision(&o, id), 10, "a baseline changes nothing canonical"); + + // play, checkpoint #1 -> revision 11 + realm.play(|m| { + m.progression.money += 5_000; + m.progression.honor.total_kills += 3; + m.build.spells.push((1111, 255)); + m.quests.rewarded.push(777); + m.pets[0].exp += 100; + }); + let bonus = new_internal_item(3, 32, 77_001); + realm.give_item(bonus.clone()); + realm.memory.items.retain(|i| i.entry.id() != 66_002); + let events = checkpoint(&mut h, &mut o, &mut realm, 100); + assert!(events.contains(&HostEvent::CheckpointRequested { session, sequence: 1 }) && events.contains(&HostEvent::CheckpointQueued { session, sequence: 1, final_checkpoint: false }), "{events:?}"); + assert_eq!(revision(&o, id), 11); + let c1 = o.store().load_current(id).unwrap(); + assert_eq!(c1.progression.money, c0.progression.money + 5_000); + assert_eq!(c1.progression.honor.total_kills, c0.progression.honor.total_kills + 3); + assert!(c1.build.spells.contains(&(1111, 255)) && c1.quests.rewarded.contains(&777)); + assert_eq!(c1.pets[0].exp, c0.pets[0].exp + 100); + assert!(c1.items.iter().any(|i| i.entry.id() == 77_001), "the new item is a session gain"); + assert!(c1.items.iter().all(|i| i.entry.id() != 66_001 && i.entry.id() != 66_002), "the realm's own items never become canonical"); + // realm normalisation did not propagate, and what the realm held back is still canonical + assert!(!c1.build.spells.contains(&(6654, 255)) && !c1.build.spells.contains(&(23250, 255))); + assert_eq!(c1.progression.honor.today_honor, c0.progression.honor.today_honor); + assert_eq!(c1.progression.chosen_title, c0.progression.chosen_title); + assert_eq!(c1.progression.known_currencies, c0.progression.known_currencies); + assert_eq!(c1.items.len(), c0.items.len() + 1, "four items the realm held back stay, one session gain"); + let first = h.outbox().unwrap(); + assert!(first.is_empty(), "the acknowledgement emptied the outbox"); + + // the same checkpoint delivered again: nothing happens + let sent = host_outbox_history(&h, session, 1); + let again = o.handle_checkpoint(&sent).unwrap(); + assert_eq!(again.outcome, AckOutcome::Duplicate); + assert_eq!(revision(&o, id), 11); + assert_eq!(again.canonical_revision, 11); + + // more play, checkpoint #2 -> revision 12; money counted once + realm.play(|m| m.progression.money += 250); + checkpoint(&mut h, &mut o, &mut realm, 200); + assert_eq!(revision(&o, id), 12); + assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 5_250); + + // the delayed #1 arrives after #2: ignored, the original answer is repeated + let late = o.handle_checkpoint(&sent).unwrap(); + assert!(matches!(late.outcome, AckOutcome::Duplicate) && late.canonical_revision == 11); + assert_eq!(revision(&o, id), 12); + + // logout -> final checkpoint -> the owner's final revision, the session closes, the next one is armed + realm.logout(); + let events = h.tick(&mut realm, 300).unwrap(); + assert!(events.contains(&HostEvent::CheckpointQueued { session, sequence: 3, final_checkpoint: true }), "{events:?}"); + let acks = deliver(&mut h, &mut o, &mut realm); + assert_eq!(acks.len(), 1); + let next = acks[0].next_session.clone().expect("the final acknowledgement names the next session"); + assert_eq!(o.session(session).unwrap().unwrap().0, "closed"); + assert!(revision(&o, id) >= 12); + let row = realm.row.clone().unwrap(); + assert_eq!((row.session_id, row.state, row.generation), (next.session_id, RowState::WaitingBaseline, 2), "the realm is armed for the next login"); + let hs = h.store().host_session(session).unwrap().unwrap(); + assert_eq!(hs.state, HostState::Closed); + let synced = h.store().synced_model(id, SERVER).unwrap().unwrap(); + assert_eq!(synced, o.store().load_current(id).unwrap(), "the Host's realm is synchronised with what the Owner now holds"); +} + +/// The bytes of a message the Host queued (kept in the outbox table after acknowledgement). +fn host_outbox_history(h: &HostService, session: SessionId, sequence: u64) -> Vec { + h.store().host_message(session, sequence).unwrap().expect("the message is kept") +} + +// ---- the matrix ---------------------------------------------------------------------------------------------------------------------- + +fn start(h: &mut HostService, o: &mut OwnerService, realm: &mut FakeRealm) { + realm.login(); + if realm.gated { + h.tick(realm, 0).unwrap(); + deliver(h, o, realm); + } +} + +fn tamper(bytes: &[u8], f: impl FnOnce(&mut serde_json::Value)) -> Vec { + let mut v: serde_json::Value = serde_json::from_slice(bytes).unwrap(); + f(&mut v); + serde_json::to_vec(&v).unwrap() +} + +fn sealed_checkpoint(host: &Store, session: SessionId, sequence: u64) -> Vec { + host.host_message(session, sequence).unwrap().expect("queued") +} + +#[test] +fn a_host_that_never_answers_never_lets_the_player_play_and_the_baseline_is_retaken_at_the_next_login() { + let Setup { mut owner, mut host, mut realm, id, session, .. } = setup(); + let mut h = HostService::new(&mut host, SERVER, HostConfig::default()); + let mut o = OwnerService::new(&mut owner); + realm.login(); + assert!(realm.gated); + realm.gate_timeout(); + assert!(!realm.online && realm.row.as_ref().unwrap().state == RowState::BaselineReady && realm.released.is_empty()); + realm.login(); + assert!(realm.gated); + assert_eq!(h.tick(&mut realm, 0).unwrap(), vec![HostEvent::BaselineTaken { session }]); + deliver(&mut h, &mut o, &mut realm); + realm.play(|m| m.progression.money += 1); + assert_eq!(revision(&o, id), 10); +} + +#[test] +fn a_failed_intermediate_checkpoint_changes_nothing_canonical_and_is_retried() { + let Setup { mut owner, mut host, mut realm, id, session, c0 } = setup(); + let mut h = HostService::new(&mut host, SERVER, HostConfig::default()); + let mut o = OwnerService::new(&mut owner); + start(&mut h, &mut o, &mut realm); + realm.play(|m| m.progression.money += 100); + realm.busy = 2; + let e1 = h.tick(&mut realm, 100).unwrap(); + let e2 = h.tick(&mut realm, 101).unwrap(); + assert!(e1.contains(&HostEvent::CheckpointBusy { session, sequence: 1 }) && e2.contains(&HostEvent::CheckpointBusy { session, sequence: 1 }), "{e1:?} {e2:?}"); + assert_eq!(revision(&o, id), 10, "a failed checkpoint changes nothing"); + assert!(h.outbox().unwrap().is_empty()); + h.tick(&mut realm, 102).unwrap(); + h.tick(&mut realm, 103).unwrap(); + deliver(&mut h, &mut o, &mut realm); + assert_eq!(revision(&o, id), 11); + assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 100); + assert_eq!(h.store().host_session(session).unwrap().unwrap().acked_sequence, 1, "the retry used the same sequence number"); +} + +#[test] +fn a_lost_acknowledgement_is_healed_by_redelivery_and_nothing_is_counted_twice() { + let Setup { mut owner, mut host, mut realm, id, c0, .. } = setup(); + let mut h = HostService::new(&mut host, SERVER, HostConfig::default()); + let mut o = OwnerService::new(&mut owner); + start(&mut h, &mut o, &mut realm); + realm.play(|m| m.progression.money += 700); + h.tick(&mut realm, 100).unwrap(); + h.tick(&mut realm, 101).unwrap(); + let msg = h.outbox().unwrap().remove(0); + let lost = o.handle_checkpoint(&msg.bytes).unwrap(); + assert_eq!((lost.outcome.clone(), revision(&o, id)), (AckOutcome::Applied, 11)); + assert_eq!(h.outbox().unwrap().len(), 1, "the Host still holds the message"); + let acks = deliver(&mut h, &mut o, &mut realm); + assert_eq!(acks[0].outcome, AckOutcome::Duplicate); + assert_eq!(revision(&o, id), 11); + assert!(h.outbox().unwrap().is_empty()); + assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 700); +} + +#[test] +fn the_host_manager_can_restart_between_the_request_and_the_read_and_between_checkpoints() { + let dir = tempfile::tempdir().unwrap(); + let host_file = dir.path().join("host.db"); + let Setup { mut owner, host, mut realm, id, session, c0 } = setup_in(Store::open_in_memory().unwrap(), Store::open_file(&host_file).unwrap()); + drop(host); + let mut o = OwnerService::new(&mut owner); + { + let mut hs = Store::open_file(&host_file).unwrap(); + let mut h = HostService::new(&mut hs, SERVER, HostConfig::default()); + start(&mut h, &mut o, &mut realm); + realm.play(|m| m.progression.money += 40); + h.tick(&mut realm, 100).unwrap(); + assert_eq!(realm.row.as_ref().unwrap().checkpoint_seq, 1, "the realm saved at the request"); + } + let mut hs = Store::open_file(&host_file).unwrap(); + assert_eq!(hs.host_session(session).unwrap().unwrap().pending_sequence, Some(1), "the reserved sequence survived"); + { + let mut h = HostService::new(&mut hs, SERVER, HostConfig::default()); + h.tick(&mut realm, 101).unwrap(); + deliver(&mut h, &mut o, &mut realm); + assert_eq!(revision(&o, id), 11); + realm.play(|m| m.progression.money += 60); + h.tick(&mut realm, 200).unwrap(); + h.tick(&mut realm, 201).unwrap(); + assert_eq!(h.outbox().unwrap().len(), 1); + } + drop(hs); + let mut hs = Store::open_file(&host_file).unwrap(); + let mut h = HostService::new(&mut hs, SERVER, HostConfig::default()); + assert_eq!(h.outbox().unwrap().len(), 1, "the outbox is persistent"); + deliver(&mut h, &mut o, &mut realm); + assert_eq!(revision(&o, id), 12); + assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 100); +} + +#[test] +fn the_owner_manager_can_restart_at_any_point() { + let dir = tempfile::tempdir().unwrap(); + let owner_file = dir.path().join("owner.db"); + let Setup { owner, mut host, mut realm, id, session, c0 } = setup_in(Store::open_file(&owner_file).unwrap(), Store::open_in_memory().unwrap()); + drop(owner); + let mut h = HostService::new(&mut host, SERVER, HostConfig::default()); + { + let mut os = Store::open_file(&owner_file).unwrap(); + let mut o = OwnerService::new(&mut os); + start(&mut h, &mut o, &mut realm); + } + realm.play(|m| m.progression.money += 5); + h.tick(&mut realm, 100).unwrap(); + h.tick(&mut realm, 101).unwrap(); + { + let mut os = Store::open_file(&owner_file).unwrap(); + let mut o = OwnerService::new(&mut os); + assert_eq!(o.session(session).unwrap().unwrap().0, "open", "the baseline survived the restart"); + deliver(&mut h, &mut o, &mut realm); + assert_eq!(revision(&o, id), 11); + } + realm.play(|m| m.progression.money += 6); + h.tick(&mut realm, 200).unwrap(); + h.tick(&mut realm, 201).unwrap(); + let mut os = Store::open_file(&owner_file).unwrap(); + let mut o = OwnerService::new(&mut os); + deliver(&mut h, &mut o, &mut realm); + assert_eq!(revision(&o, id), 12); + assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 11); +} + +#[test] +fn a_worldserver_restart_continues_the_same_session_without_a_new_baseline() { + let Setup { mut owner, mut host, mut realm, id, session, c0 } = setup(); + let mut h = HostService::new(&mut host, SERVER, HostConfig::default()); + let mut o = OwnerService::new(&mut owner); + start(&mut h, &mut o, &mut realm); + realm.play(|m| m.progression.money += 10); + checkpoint(&mut h, &mut o, &mut realm, 100); + assert_eq!(revision(&o, id), 11); + realm.play(|m| m.progression.money += 99); + realm.crash(); + assert!(h.tick(&mut realm, 150).unwrap().is_empty(), "an offline character with a running session needs nothing"); + assert_eq!(revision(&o, id), 11, "the last acknowledged checkpoint wins"); + realm.login(); + assert!(!realm.gated); + assert_eq!(realm.row.as_ref().unwrap().session_id, session); + realm.play(|m| m.progression.money += 5); + checkpoint(&mut h, &mut o, &mut realm, 200); + assert_eq!(revision(&o, id), 12); + assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 15, "the lost 99 stayed lost, the rest was counted once"); +} + +#[test] +fn a_superseded_session_is_refused_and_closed() { + let Setup { mut owner, mut host, mut realm, id, session, .. } = setup(); + let mut h = HostService::new(&mut host, SERVER, HostConfig::default()); + { + let mut o = OwnerService::new(&mut owner); + start(&mut h, &mut o, &mut realm); + } + let mut m = owner.load_current(id).unwrap(); + m.progression.money += 1; + owner.commit_snapshot(id, 10, m, "realm-c", None).unwrap(); + let mut o = OwnerService::new(&mut owner); + realm.play(|m| m.progression.money += 3); + h.tick(&mut realm, 100).unwrap(); + h.tick(&mut realm, 101).unwrap(); + let acks = deliver(&mut h, &mut o, &mut realm); + assert_eq!(acks[0].outcome, AckOutcome::StaleSession); + assert_eq!(revision(&o, id), 11, "nothing was merged over the newer state"); + assert_eq!(o.store().load_current(id).unwrap().progression.money, geared_level_eighty().progression.money + 1); + assert_eq!(h.store().host_session(session).unwrap().unwrap().state, HostState::Closed); + let msg = sealed_checkpoint(h.store(), session, 1); + assert_eq!(o.handle_checkpoint(&msg).unwrap().outcome, AckOutcome::StaleSession); +} + +#[test] +fn messages_for_the_wrong_character_session_or_realm_and_malformed_ones_change_nothing() { + let Setup { mut owner, mut host, mut realm, id, session, .. } = setup(); + let mut h = HostService::new(&mut host, SERVER, HostConfig::default()); + let mut o = OwnerService::new(&mut owner); + start(&mut h, &mut o, &mut realm); + realm.play(|m| m.progression.money += 9); + h.tick(&mut realm, 100).unwrap(); + h.tick(&mut realm, 101).unwrap(); + let good = h.outbox().unwrap().remove(0).bytes; + let before = revision(&o, id); + let other_character = CharacterId::new(); + let other_session = SessionId::new(); + + let cases: Vec<(&str, Vec)> = vec![ + ("another character", tamper(&good, |v| v["character_id"] = other_character.to_string().into())), + ("an unknown session", tamper(&good, |v| v["session_id"] = other_session.to_string().into())), + ("another realm", tamper(&good, |v| v["server_id"] = "realm-z".into())), + ("another base revision", tamper(&good, |v| v["base_canonical_revision"] = 3.into())), + ("a wrong content hash", tamper(&good, |v| v["content_hash"] = "00".repeat(32).into())), + ("sequence zero", tamper(&good, |v| v["sequence"] = 0.into())), + ("a snapshot of another character", { + let mut other = geared_level_eighty(); + other.character_id = other_character; + to_json(&PortableCheckpoint::new(session, id, SERVER, 10, 1, false, &other).unwrap()).unwrap() + }), + ("a snapshot that does not match its hash", tamper(&good, |v| { + let payload = v["realm_snapshot"]["payload"].as_str().unwrap().to_string(); + v["realm_snapshot"]["payload"] = format!("{}AA", &payload[..payload.len() - 4]).into(); + })), + ]; + for (name, bytes) in cases { + let ack = o.handle_checkpoint(&bytes).unwrap_or_else(|e| panic!("{name}: {e}")); + assert!(matches!(ack.outcome, AckOutcome::Rejected(_)), "{name}: {:?}", ack.outcome); + assert_eq!(revision(&o, id), before, "{name} must not change the canonical character"); + } + for (name, bytes) in [ + ("garbage", b"\x00\x01 not json".to_vec()), + ("an unknown field", tamper(&good, |v| v["extra"] = 1.into())), + ("a newer protocol", tamper(&good, |v| v["protocol_version"] = 99.into())), + ("no protocol version", tamper(&good, |v| { + v.as_object_mut().unwrap().remove("protocol_version"); + })), + ("an over-limit message", vec![b' '; MAX_MESSAGE_BYTES + 1]), + ] { + assert!(o.handle_checkpoint(&bytes).is_err(), "{name}"); + assert_eq!(revision(&o, id), before, "{name}"); + } + let bomb = zstd::stream::encode_all(vec![b' '; crate::portable::model::limits::MAX_SNAPSHOT_BYTES + 4096].as_slice(), 19).unwrap(); + let evil = tamper(&good, |v| { + v["realm_snapshot"]["payload"] = base64::Engine::encode(&base64::engine::general_purpose::STANDARD, &bomb).into(); + v["realm_snapshot"]["content_hash"] = "11".repeat(32).into(); + v["content_hash"] = "11".repeat(32).into(); + }); + assert!(matches!(o.handle_checkpoint(&evil).unwrap().outcome, AckOutcome::Rejected(_))); + assert_eq!(revision(&o, id), before); + assert_eq!(o.handle_checkpoint(&good).unwrap().outcome, AckOutcome::Applied); + assert_eq!(revision(&o, id), before + 1); +} + +#[test] +fn checkpoints_out_of_order_or_with_reused_sequences_are_refused_safely() { + let Setup { mut owner, mut host, mut realm, id, session, c0 } = setup(); + let mut h = HostService::new(&mut host, SERVER, HostConfig::default()); + let mut o = OwnerService::new(&mut owner); + start(&mut h, &mut o, &mut realm); + realm.play(|m| m.progression.money += 10); + h.tick(&mut realm, 100).unwrap(); + h.tick(&mut realm, 101).unwrap(); + realm.play(|m| m.progression.money += 20); + h.tick(&mut realm, 200).unwrap(); + h.tick(&mut realm, 201).unwrap(); + let (one, two) = (sealed_checkpoint(h.store(), session, 1), sealed_checkpoint(h.store(), session, 2)); + assert_eq!(o.handle_checkpoint(&two).unwrap().outcome, AckOutcome::Applied); + assert_eq!(revision(&o, id), 11); + assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 30); + assert_eq!(o.handle_checkpoint(&one).unwrap().outcome, AckOutcome::StaleSequence); + assert_eq!(revision(&o, id), 11); + assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 30); + let reused = tamper(&two, |v| { + let mut m = geared_level_eighty(); + m.progression.money += 123; + let other = PortableCheckpoint::new(session, id, SERVER, 10, 2, false, &m).unwrap(); + v["realm_snapshot"] = serde_json::to_value(&other.realm_snapshot).unwrap(); + v["content_hash"] = other.content_hash.into(); + }); + assert!(matches!(o.handle_checkpoint(&reused).unwrap().outcome, AckOutcome::Rejected(_))); + assert_eq!(revision(&o, id), 11); +} + +#[test] +fn a_checkpoint_that_changes_nothing_makes_no_revision_and_new_items_keep_their_ids() { + let Setup { mut owner, mut host, mut realm, id, c0, .. } = setup(); + let mut h = HostService::new(&mut host, SERVER, HostConfig::default()); + let mut o = OwnerService::new(&mut owner); + start(&mut h, &mut o, &mut realm); + realm.give_item(new_internal_item(9, 40, 88_001)); + realm.play(|m| m.progression.money += 1); + checkpoint(&mut h, &mut o, &mut realm, 100); + assert_eq!(revision(&o, id), 11); + let gained: Vec = o.store().load_current(id).unwrap().items.iter().filter(|i| i.entry.id() == 88_001).map(|i| i.id).collect(); + assert_eq!(gained.len(), 1); + checkpoint(&mut h, &mut o, &mut realm, 200); + assert_eq!(revision(&o, id), 11, "no new revision for an unchanged character"); + assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 1, "nothing was counted again"); + realm.play(|m| m.progression.money += 1); + checkpoint(&mut h, &mut o, &mut realm, 300); + let again: Vec = o.store().load_current(id).unwrap().items.iter().filter(|i| i.entry.id() == 88_001).map(|i| i.id).collect(); + assert_eq!(again, gained, "an item first seen after B0 keeps one identity"); + assert_eq!(revision(&o, id), 12); +} + +#[test] +fn a_logout_while_a_checkpoint_is_pending_turns_it_into_the_final_one() { + let Setup { mut owner, mut host, mut realm, id, session, c0 } = setup(); + let mut h = HostService::new(&mut host, SERVER, HostConfig::default()); + let mut o = OwnerService::new(&mut owner); + start(&mut h, &mut o, &mut realm); + realm.play(|m| m.progression.money += 8); + h.tick(&mut realm, 100).unwrap(); + realm.play(|m| m.progression.money += 2); + realm.logout(); + let events = h.tick(&mut realm, 101).unwrap(); + assert!(events.contains(&HostEvent::CheckpointQueued { session, sequence: 1, final_checkpoint: true }), "{events:?}"); + let acks = deliver(&mut h, &mut o, &mut realm); + assert!(acks[0].next_session.is_some()); + assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 10, "the final state includes the logout save"); +} + +#[test] +fn a_lost_final_acknowledgement_is_healed_and_the_realm_is_armed_again_afterwards() { + let Setup { mut owner, mut host, mut realm, id, session, .. } = setup(); + let mut h = HostService::new(&mut host, SERVER, HostConfig::default()); + let mut o = OwnerService::new(&mut owner); + start(&mut h, &mut o, &mut realm); + realm.play(|m| m.progression.money += 4); + realm.logout(); + h.tick(&mut realm, 100).unwrap(); + let msg = h.outbox().unwrap().remove(0); + let lost = o.handle_checkpoint(&msg.bytes).unwrap(); + let next = lost.next_session.clone().unwrap(); + assert_eq!(realm.row.as_ref().unwrap().session_id, session, "the Host never heard the answer: the realm is not re-armed yet"); + let acks = deliver(&mut h, &mut o, &mut realm); + assert_eq!(acks[0].outcome, AckOutcome::Duplicate); + assert_eq!(acks[0].next_session.as_ref().unwrap().session_id, next.session_id, "the same next session is named again"); + assert_eq!(realm.row.as_ref().unwrap().session_id, next.session_id); + assert_eq!(o.session(next.session_id).unwrap().unwrap().0, "offered"); + realm.login(); + assert!(realm.gated); + assert_eq!(h.tick(&mut realm, 200).unwrap(), vec![HostEvent::BaselineTaken { session: next.session_id }]); + deliver(&mut h, &mut o, &mut realm); + assert!(matches!(h.store().host_session(next.session_id).unwrap().unwrap().state, HostState::Open)); + assert_eq!(o.session(next.session_id).unwrap().unwrap().0, "open"); + assert_eq!(revision(&o, id), o.store().character(id).unwrap().revision); +} + +#[test] +fn a_realm_that_was_never_given_a_session_row_is_left_alone() { + let Setup { mut host, mut realm, .. } = setup(); + realm.row = None; + let mut h = HostService::new(&mut host, SERVER, HostConfig::default()); + realm.login(); + assert!(h.tick(&mut realm, 0).unwrap().is_empty()); + assert!(!realm.gated); +} + +#[test] +fn an_autosave_between_checkpoints_is_harmless() { + let Setup { mut owner, mut host, mut realm, id, c0, .. } = setup(); + let mut h = HostService::new(&mut host, SERVER, HostConfig::default()); + let mut o = OwnerService::new(&mut owner); + start(&mut h, &mut o, &mut realm); + realm.play(|m| m.progression.money += 7); + realm.autosave(); + realm.play(|m| m.progression.money += 3); + checkpoint(&mut h, &mut o, &mut realm, 100); + assert_eq!(revision(&o, id), 11); + assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 10); +} diff --git a/crates/coa-core/src/portable/store.rs b/crates/coa-core/src/portable/store.rs index 7db52b6..f5a64a8 100644 --- a/crates/coa-core/src/portable/store.rs +++ b/crates/coa-core/src/portable/store.rs @@ -18,7 +18,7 @@ use super::versions::{PORTABLE_COLLECTION_FORMAT_VERSION, SNAPSHOT_FORMAT_VERSIO pub const DATABASE_FILE: &str = "portable.db"; /// `PRAGMA application_id`: "COAP". Refuses to adopt an unrelated SQLite file. const APPLICATION_ID: i64 = 0x434F_4150; -const MIGRATIONS: &[&str] = &[include_str!("migrations/001_init.sql"), include_str!("migrations/002_item_mapping_lifecycle.sql"), include_str!("migrations/003_import_journal.sql"), include_str!("migrations/004_reconciliation.sql")]; +const MIGRATIONS: &[&str] = &[include_str!("migrations/001_init.sql"), include_str!("migrations/002_item_mapping_lifecycle.sql"), include_str!("migrations/003_import_journal.sql"), include_str!("migrations/004_reconciliation.sql"), include_str!("migrations/005_sessions.sql")]; pub const DEFAULT_HISTORY_KEEP: u32 = 20; pub const MAX_HISTORY_KEEP: u32 = 1_000; @@ -1018,8 +1018,11 @@ fn validate_observations(server_id: &str, observations: &[ItemObservation]) -> R Ok(()) } +mod host; mod journal; +mod owner; mod sync; +pub use host::*; pub use journal::*; pub use sync::*; diff --git a/crates/coa-core/src/portable/store/host.rs b/crates/coa-core/src/portable/store/host.rs new file mode 100644 index 0000000..e24fe73 --- /dev/null +++ b/crates/coa-core/src/portable/store/host.rs @@ -0,0 +1,332 @@ +//! The HOST side of a runtime portable session: what the realm's Manager remembers about the sessions it runs. +//! +//! The Host keeps a **copy** of the canonical character at the revision the Owner handed over (it needs it to import, to map +//! items and to know what the realm was synchronised with); it never produces a new canonical revision. It keeps its own +//! sequence numbers and an outbox of messages the Owner has not acknowledged. + +use rusqlite::{params, OptionalExtension}; + +use super::*; +use crate::portable::ids::{PortablePetId, SessionId}; +use crate::portable::session::protocol::*; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum HostState { + /// The realm holds the marker; `B0` has not been read yet. + Armed, + /// `B0` was read and sent; checkpoints run. + Open, + Closed, +} + +impl HostState { + fn parse(s: &str) -> Result { + Ok(match s { + "armed" => HostState::Armed, + "open" => HostState::Open, + "closed" => HostState::Closed, + other => return Err(PortableError::Invalid(format!("unknown host session state {other:?}"))), + }) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct HostSession { + pub session_id: SessionId, + pub character_id: CharacterId, + pub server_id: String, + pub local_guid: Option, + pub base_revision: u64, + pub generation: u32, + pub state: HostState, + pub next_sequence: u64, + pub pending_sequence: Option, + pub acked_sequence: u64, + pub owned_items: Vec, + pub owned_pets: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct OutboxMessage { + pub session_id: SessionId, + pub sequence: u64, + pub started: bool, + pub bytes: Vec, +} + +/// What an acknowledgement changed on the Host. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum AckEffect { + /// Nothing was waiting for it (an acknowledgement delivered twice). + Ignored, + Applied, + /// The final checkpoint was applied: the realm is now synchronised with the canonical character, and this session follows. + Finished { next_session: SessionId, next_revision: u64 }, + /// The Owner will never accept this message (stale or rejected); it was dropped from the outbox. + Dropped(String), +} + +fn read_host_session(conn: &Connection, session: SessionId) -> Result> { + type Raw = (String, String, Option, i64, i64, String, i64, Option, i64, String, String); + let raw: Option = conn + .query_row( + "SELECT character_id, server_id, local_guid, base_revision, generation, state, next_sequence, pending_sequence, acked_sequence, owned_items, owned_pets FROM host_session WHERE session_id = ?1", + [session.to_string()], + |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?, r.get(5)?, r.get(6)?, r.get(7)?, r.get(8)?, r.get(9)?, r.get(10)?)), + ) + .optional()?; + let Some((character, server_id, guid, base, generation, state, next, pending, acked, items, pets)) = raw else { return Ok(None) }; + Ok(Some(HostSession { + session_id: session, + character_id: character.parse()?, + server_id, + local_guid: guid.map(|g| g as u32), + base_revision: base as u64, + generation: generation as u32, + state: HostState::parse(&state)?, + next_sequence: next as u64, + pending_sequence: pending.map(|p| p as u64), + acked_sequence: acked as u64, + owned_items: serde_json::from_str(&items)?, + owned_pets: serde_json::from_str(&pets)?, + })) +} + +fn insert_host_session(tx: &Transaction<'_>, session: SessionId, id: CharacterId, server_id: &str, guid: Option, base: u64, generation: u32, model: &PortableCharacter) -> Result<()> { + let items: Vec = model.items.iter().map(|i| i.id).collect(); + let pets: Vec = model.pets.iter().map(|p| p.id).collect(); + let at = now(); + tx.execute( + "INSERT INTO host_session(session_id, character_id, server_id, local_guid, base_revision, generation, state, next_sequence, acked_sequence, owned_items, owned_pets, created_at, updated_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, 'armed', 1, 0, ?7, ?8, ?9, ?9)", + params![session.to_string(), id.to_string(), server_id, guid, base as i64, generation, serde_json::to_string(&items)?, serde_json::to_string(&pets)?, at], + )?; + Ok(()) +} + +/// Install (or advance to) the canonical character the Owner handed over, as the Host's copy. The revision is the Owner's. +fn install_copy_in_tx(tx: &Transaction<'_>, profile: ProfileId, model: &PortableCharacter, revision: u64, source_server_id: &str) -> Result<()> { + let encoded = snapshot::encode(model)?; + let id = model.character_id; + let at = now(); + let exists: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM character WHERE character_id = ?1)", [id.to_string()], |r| r.get(0))?; + if !exists { + tx.execute( + "INSERT INTO character(character_id, profile_id, ruleset, name, race, class, gender, level, revision, created_at, updated_at, archived) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?10, 0)", + params![id.to_string(), profile.to_string(), model.ruleset.as_str(), model.identity.name, model.identity.race.to_string(), model.identity.class.to_string(), model.identity.gender, model.progression.level, revision as i64, at], + )?; + insert_snapshot(tx, id, revision, &encoded, source_server_id, Some("received"), &at)?; + return Ok(()); + } + let record = read_character(tx, id)?; + if record.ruleset != model.ruleset { + return Err(PortableError::RulesetChange { from: record.ruleset.to_string(), to: model.ruleset.to_string() }); + } + if revision < record.revision { + return Err(PortableError::StaleRevision { expected: revision, current: record.revision }); + } + if revision == record.revision { + let (_, have) = read_snapshot_row(tx, id, revision)?; + if have.content_hash != encoded.content_hash { + return Err(PortableError::Invalid("the Host already holds a different canonical state at this revision".into())); + } + return Ok(()); + } + insert_snapshot(tx, id, revision, &encoded, source_server_id, Some("received"), &at)?; + update_character(tx, id, model, revision, &at)?; + Ok(()) +} + +/// Presence of every active mapping after the Owner said what the canonical character owns: owned and shown = present, owned and +/// not shown = filtered, shown and not owned = realm-local, neither = retired. +fn reclassify_in_tx(tx: &Transaction<'_>, id: CharacterId, server_id: &str, revision: u64, items: &[PortableItemId], pets: &[PortablePetId]) -> Result<()> { + let at = now(); + for (table, column, owned) in [("item_mapping", "portable_item_id", items.iter().map(|i| i.to_string()).collect::>()), ("pet_mapping", "portable_pet_id", pets.iter().map(|p| p.to_string()).collect())] { + let mut stmt = tx.prepare(&format!("SELECT mapping_id, {column}, presence FROM {table} WHERE character_id = ?1 AND server_id = ?2 AND state = 'active'"))?; + let rows: Vec<(i64, String, String)> = stmt.query_map(params![id.to_string(), server_id], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))?.collect::>()?; + drop(stmt); + for (mapping, portable, presence) in rows { + let observed = presence != "filtered"; + let is_owned = owned.contains(&portable); + match (is_owned, observed) { + (true, true) => tx.execute(&format!("UPDATE {table} SET presence = 'present', updated_at = ?2 WHERE mapping_id = ?1"), params![mapping, at])?, + (true, false) => 0, + (false, true) => tx.execute(&format!("UPDATE {table} SET presence = 'realm_local', updated_at = ?2 WHERE mapping_id = ?1"), params![mapping, at])?, + (false, false) => tx.execute(&format!("UPDATE {table} SET state = 'retired', retired_revision = ?3, retired_reason = 'absent', updated_at = ?2 WHERE mapping_id = ?1"), params![mapping, at, revision as i64])?, + }; + } + } + Ok(()) +} + +impl Store { + /// The Host's copy of the character the Owner handed over (creates it, or advances it to a newer revision). + pub fn host_install_copy(&mut self, profile: ProfileId, model: &PortableCharacter, revision: u64, owner_server_id: &str) -> Result<()> { + check_server_id(owner_server_id)?; + let tx = self.write_tx()?; + install_copy_in_tx(&tx, profile, model, revision, owner_server_id)?; + tx.commit()?; + Ok(()) + } + + /// Record that a session is about to be armed on the realm (before the import runs, so a crash in between is recoverable). + pub fn host_prepare_session(&mut self, offer: &SessionOffer, model: &PortableCharacter) -> Result { + let tx = self.write_tx()?; + if let Some(existing) = read_host_session(&tx, offer.session_id)? { + return Ok(existing); + } + insert_host_session(&tx, offer.session_id, offer.character_id, &offer.server_id, None, offer.canonical_revision, 1, model)?; + tx.commit()?; + Ok(read_host_session(&self.conn, offer.session_id)?.expect("just inserted")) + } + + /// The import committed: the session now belongs to this local character. + pub fn host_bind_session(&mut self, session: SessionId, local_guid: u32) -> Result<()> { + let changed = self.conn.execute("UPDATE host_session SET local_guid = ?2, updated_at = ?3 WHERE session_id = ?1 AND (local_guid IS NULL OR local_guid = ?2)", params![session.to_string(), local_guid, now()])?; + if changed == 0 { + return Err(PortableError::Invalid(format!("session {session} does not exist or is bound to another character"))); + } + Ok(()) + } + + pub fn host_session(&self, session: SessionId) -> Result> { + read_host_session(&self.conn, session) + } + + /// Armed and open sessions on one realm, oldest first. + pub fn host_live_sessions(&self, server_id: &str) -> Result> { + let mut stmt = self.conn.prepare("SELECT session_id FROM host_session WHERE server_id = ?1 AND state IN ('armed', 'open') ORDER BY session_id")?; + let ids: Vec = stmt.query_map([server_id], |r| r.get(0))?.collect::>()?; + ids.iter().map(|s| Ok(read_host_session(&self.conn, s.parse()?)?.expect("it was just listed"))).collect() + } + + /// Reserve the sequence of the next checkpoint **before** the realm is asked to save. Asking again returns the same number + /// until the checkpoint is queued, so a restart resumes the same checkpoint. + pub fn host_begin_checkpoint(&mut self, session: SessionId) -> Result { + let tx = self.write_tx()?; + let s = read_host_session(&tx, session)?.ok_or_else(|| PortableError::Invalid(format!("unknown session {session}")))?; + if s.state != HostState::Open { + return Err(PortableError::Invalid("a checkpoint needs an open session".into())); + } + let sequence = s.pending_sequence.unwrap_or(s.next_sequence); + tx.execute("UPDATE host_session SET pending_sequence = ?2, updated_at = ?3 WHERE session_id = ?1", params![session.to_string(), sequence as i64, now()])?; + tx.commit()?; + Ok(sequence) + } + + /// `B0` was read from the realm: persist the baseline (presence of items and pets), queue `PortableSessionStarted`, open. + pub fn host_queue_started(&mut self, session: SessionId, msg: &PortableSessionStarted, b0: &PortableCharacter, items: &[ItemObservation], pets: &[PetObservation]) -> Result<()> { + let tx = self.write_tx()?; + let s = read_host_session(&tx, session)?.ok_or_else(|| PortableError::Invalid(format!("unknown session {session}")))?; + match s.state { + HostState::Armed => {} + _ => return Err(PortableError::Invalid("this session already has its baseline".into())), + } + capture_baseline_in_tx(&tx, s.character_id, &s.server_id, &BaselineInput { b0, items, pets })?; + let bytes = to_json(msg)?; + tx.execute("INSERT INTO host_outbox(session_id, sequence, kind, message, state, created_at) VALUES (?1, 0, 'started', ?2, 'pending', ?3)", params![session.to_string(), bytes, now()])?; + tx.execute("UPDATE host_session SET state = 'open', generation = ?2, updated_at = ?3 WHERE session_id = ?1", params![session.to_string(), msg.baseline_generation, now()])?; + tx.commit()?; + Ok(()) + } + + /// `B1` was read after the realm's checkpoint marker appeared: register what the realm shows (so new items keep their ids in + /// the next checkpoint) and queue the message. + pub fn host_queue_checkpoint(&mut self, session: SessionId, msg: &PortableCheckpoint, items: &[ItemObservation], pets: &[PetObservation]) -> Result<()> { + let tx = self.write_tx()?; + let s = read_host_session(&tx, session)?.ok_or_else(|| PortableError::Invalid(format!("unknown session {session}")))?; + if s.state != HostState::Open || s.pending_sequence != Some(msg.sequence) { + return Err(PortableError::Invalid("this checkpoint was not reserved".into())); + } + let baseline = read_open_baseline(&tx, s.character_id, &s.server_id)?.ok_or(PortableError::NoBaseline)?; + let (c0_items, b0_items): (std::collections::HashSet<_>, std::collections::HashSet<_>) = (baseline.c0.items.iter().map(|i| i.id).collect(), baseline.b0.items.iter().map(|i| i.id).collect()); + let (c0_pets, b0_pets): (std::collections::HashSet<_>, std::collections::HashSet<_>) = (baseline.c0.pets.iter().map(|p| p.id).collect(), baseline.b0.pets.iter().map(|p| p.id).collect()); + // what the Owner is expected to own: what it owned at the last acknowledgement plus what is new since B0 + let mut canonical_items: std::collections::HashSet<_> = s.owned_items.iter().copied().collect(); + canonical_items.extend(items.iter().map(|o| o.portable_item_id).filter(|i| !b0_items.contains(i) && !c0_items.contains(i))); + let mut canonical_pets: std::collections::HashSet<_> = s.owned_pets.iter().copied().collect(); + canonical_pets.extend(pets.iter().map(|o| o.portable_pet_id).filter(|p| !b0_pets.contains(p) && !c0_pets.contains(p))); + let mapping_revision: i64 = tx.query_row("SELECT last_revision FROM character_server_mapping WHERE character_id = ?1 AND server_id = ?2", params![s.character_id.to_string(), s.server_id], |r| r.get(0))?; + let protection = Protection { realm_local_items: items.iter().map(|o| o.portable_item_id).filter(|i| !canonical_items.contains(i)).collect(), canonical_items }; + reconcile_with(&tx, s.character_id, &s.server_id, mapping_revision as u64, items, &protection)?; + let pet_protection = PetProtection { realm_local_pets: pets.iter().map(|o| o.portable_pet_id).filter(|p| !canonical_pets.contains(p)).collect(), canonical_pets }; + sync_pets_in_tx(&tx, s.character_id, &s.server_id, mapping_revision as u64, pets, &pet_protection)?; + let bytes = to_json(msg)?; + tx.execute("INSERT INTO host_outbox(session_id, sequence, kind, message, state, created_at) VALUES (?1, ?2, 'checkpoint', ?3, 'pending', ?4)", params![session.to_string(), msg.sequence as i64, bytes, now()])?; + tx.execute("UPDATE host_session SET next_sequence = ?2, pending_sequence = NULL, updated_at = ?3 WHERE session_id = ?1", params![session.to_string(), msg.sequence as i64 + 1, now()])?; + tx.commit()?; + Ok(()) + } + + /// The bytes of one queued message, acknowledged or not. + pub fn host_message(&self, session: SessionId, sequence: u64) -> Result>> { + Ok(self.conn.query_row("SELECT message FROM host_outbox WHERE session_id = ?1 AND sequence = ?2", params![session.to_string(), sequence as i64], |r| r.get(0)).optional()?) + } + + /// Messages the Owner has not acknowledged yet, in the order they must be delivered. + pub fn host_outbox_pending(&self, server_id: &str) -> Result> { + let mut stmt = self.conn.prepare( + "SELECT o.session_id, o.sequence, o.kind, o.message FROM host_outbox o JOIN host_session s ON s.session_id = o.session_id + WHERE s.server_id = ?1 AND o.state = 'pending' ORDER BY o.created_at, o.sequence", + )?; + let rows = stmt.query_map([server_id], |r| Ok((r.get::<_, String>(0)?, r.get::<_, i64>(1)?, r.get::<_, String>(2)?, r.get::<_, Vec>(3)?)))?; + let mut out = Vec::new(); + for row in rows { + let (session, sequence, kind, bytes) = row?; + out.push(OutboxMessage { session_id: session.parse()?, sequence: sequence as u64, started: kind == "started", bytes }); + } + Ok(out) + } + + /// Apply the Owner's acknowledgement. + pub fn host_receive_ack(&mut self, ack: &OwnerAck) -> Result { + let tx = self.write_tx()?; + let Some(s) = read_host_session(&tx, ack.session_id)? else { return Ok(AckEffect::Ignored) }; + let waiting: Option = tx + .query_row("SELECT state FROM host_outbox WHERE session_id = ?1 AND sequence = ?2", params![ack.session_id.to_string(), ack.sequence as i64], |r| r.get(0)) + .optional()?; + match waiting.as_deref() { + None => return Ok(AckEffect::Ignored), + Some("acked") => return Ok(AckEffect::Ignored), + Some(_) => {} + } + let at = now(); + if !ack.outcome.accepted() { + tx.execute("UPDATE host_outbox SET state = 'acked' WHERE session_id = ?1 AND sequence = ?2", params![ack.session_id.to_string(), ack.sequence as i64])?; + if matches!(ack.outcome, AckOutcome::StaleSession) { + tx.execute("UPDATE host_session SET state = 'closed', updated_at = ?2 WHERE session_id = ?1", params![ack.session_id.to_string(), at])?; + } + tx.commit()?; + return Ok(AckEffect::Dropped(format!("{:?}", ack.outcome))); + } + tx.execute("UPDATE host_outbox SET state = 'acked' WHERE session_id = ?1 AND sequence = ?2", params![ack.session_id.to_string(), ack.sequence as i64])?; + if ack.sequence > 0 { + tx.execute( + "UPDATE host_session SET acked_sequence = max(acked_sequence, ?2), owned_items = ?3, owned_pets = ?4, updated_at = ?5 WHERE session_id = ?1", + params![ack.session_id.to_string(), ack.sequence as i64, serde_json::to_string(&ack.owned_items)?, serde_json::to_string(&ack.owned_pets)?, at], + )?; + reclassify_in_tx(&tx, s.character_id, &s.server_id, ack.canonical_revision, &ack.owned_items, &ack.owned_pets)?; + } + let (Some(canonical), Some(next)) = (&ack.canonical, &ack.next_session) else { + tx.commit()?; + return Ok(AckEffect::Applied); + }; + + // the final checkpoint: the realm is now synchronised with the canonical character, and the next session follows + let model = canonical.open()?; + if hex::encode(canonical.hash()?) != ack.canonical_hash || model.character_id != s.character_id { + return Err(PortableError::Invalid("the final acknowledgement carries another character than its session".into())); + } + let profile: String = tx.query_row("SELECT profile_id FROM character WHERE character_id = ?1", [s.character_id.to_string()], |r| r.get(0))?; + install_copy_in_tx(&tx, profile.parse()?, &model, ack.canonical_revision, "owner")?; + let encoded = snapshot::encode(&model)?; + tx.execute("UPDATE character_server_mapping SET last_revision = ?3, state = 'synced', updated_at = ?4 WHERE character_id = ?1 AND server_id = ?2", params![s.character_id.to_string(), s.server_id, ack.canonical_revision as i64, at])?; + set_synced_in_tx(&tx, s.character_id, &s.server_id, &encoded)?; + tx.execute("UPDATE realm_baseline SET state = 'closed', updated_at = ?3 WHERE character_id = ?1 AND server_id = ?2 AND state = 'open'", params![s.character_id.to_string(), s.server_id, at])?; + tx.execute("UPDATE host_session SET state = 'closed', updated_at = ?2 WHERE session_id = ?1", params![ack.session_id.to_string(), at])?; + insert_host_session(&tx, next.session_id, s.character_id, &s.server_id, s.local_guid, next.canonical_revision, s.generation + 1, &model)?; + tx.commit()?; + Ok(AckEffect::Finished { next_session: next.session_id, next_revision: next.canonical_revision }) + } +} diff --git a/crates/coa-core/src/portable/store/owner.rs b/crates/coa-core/src/portable/store/owner.rs new file mode 100644 index 0000000..51a8743 --- /dev/null +++ b/crates/coa-core/src/portable/store/owner.rs @@ -0,0 +1,278 @@ +//! The OWNER side of a runtime portable session: offer a character to a realm, accept the realm's baseline and apply its +//! checkpoints to the canonical character. Nothing here reads a realm, and nothing here knows how the messages travel. +//! +//! Every checkpoint is `merge3(C0, B0, current B1)` with the session's own `C0` and `B0` as anchors, so a checkpoint that +//! is repeated, re-sent or superseded by a later one can never count progress twice. + +use rusqlite::{params, OptionalExtension}; + +use super::*; +use crate::portable::ids::SessionId; +use crate::portable::merge::{merge3, Mode}; +use crate::portable::session::protocol::*; + +struct OwnerRow { + session_id: SessionId, + character_id: CharacterId, + server_id: String, + state: String, + c0_revision: u64, + c0_hash: [u8; 32], + c0_payload: Vec, + generation: Option, + b0_hash: Option<[u8; 32]>, + b0_payload: Option>, + head_revision: u64, + last_sequence: u64, +} + +fn read_owner_row(conn: &Connection, session: SessionId) -> Result> { + type Raw = (String, String, String, i64, Vec, Vec, Option, Option>, Option>, i64, i64); + let raw: Option = conn + .query_row( + "SELECT character_id, server_id, state, c0_revision, c0_hash, c0_payload, baseline_generation, b0_hash, b0_payload, head_revision, last_sequence FROM owner_session WHERE session_id = ?1", + [session.to_string()], + |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?, r.get(5)?, r.get(6)?, r.get(7)?, r.get(8)?, r.get(9)?, r.get(10)?)), + ) + .optional()?; + let Some((character, server_id, state, c0_revision, c0_hash, c0_payload, generation, b0_hash, b0_payload, head, last)) = raw else { return Ok(None) }; + Ok(Some(OwnerRow { + session_id: session, + character_id: character.parse()?, + server_id, + state, + c0_revision: c0_revision as u64, + c0_hash: hash_from_blob(c0_hash)?, + c0_payload, + generation: generation.map(|g| g as u32), + b0_hash: b0_hash.map(hash_from_blob).transpose()?, + b0_payload, + head_revision: head as u64, + last_sequence: last as u64, + })) +} + +fn head_hash(conn: &Connection, id: CharacterId, revision: u64) -> Result<[u8; 32]> { + let blob: Vec = conn.query_row("SELECT content_hash FROM snapshot WHERE character_id = ?1 AND revision = ?2", params![id.to_string(), revision as i64], |r| r.get(0)).map_err(|_| PortableError::UnknownRevision { character: id, revision })?; + hash_from_blob(blob) +} + +fn ack(session: SessionId, sequence: u64, outcome: AckOutcome, revision: u64, hash: [u8; 32]) -> OwnerAck { + OwnerAck { protocol_version: PROTOCOL_VERSION, session_id: session, sequence, outcome, canonical_revision: revision, canonical_hash: hex::encode(hash), owned_items: vec![], owned_pets: vec![], canonical: None, next_session: None } +} + +fn owned_ids(model: &PortableCharacter) -> (Vec, Vec) { + let items: Vec<_> = model.items.iter().map(|i| i.id).take(MAX_ACK_IDS).collect(); + let pets: Vec<_> = model.pets.iter().map(|p| p.id).take(MAX_ACK_IDS).collect(); + (items, pets) +} + +fn offer_from_row(row: &OwnerRow) -> SessionOffer { + SessionOffer { + protocol_version: PROTOCOL_VERSION, + session_id: row.session_id, + character_id: row.character_id, + server_id: row.server_id.clone(), + canonical_revision: row.c0_revision, + snapshot: Envelope::from_encoded(&EncodedSnapshot { content_hash: row.c0_hash, uncompressed_size: 0, payload: row.c0_payload.clone() }), + } +} + +fn open_offer(tx: &Transaction<'_>, id: CharacterId, server_id: &str) -> Result { + let record = read_character(tx, id)?; + let (_, enc) = read_snapshot_row(tx, id, record.revision)?; + let session = SessionId::new(); + let at = now(); + tx.execute( + "INSERT INTO owner_session(session_id, character_id, server_id, state, c0_revision, c0_hash, c0_payload, head_revision, last_sequence, created_at, updated_at) + VALUES (?1, ?2, ?3, 'offered', ?4, ?5, ?6, ?4, 0, ?7, ?7)", + params![session.to_string(), id.to_string(), server_id, record.revision as i64, enc.content_hash.as_slice(), enc.payload, at], + )?; + Ok(offer_from_row(&read_owner_row(tx, session)?.expect("just inserted"))) +} + +impl Store { + /// Offer the character's current canonical revision to a realm and open the session that will follow it. An offer that was + /// made and not yet started is repeated as it was; one made against an older revision is superseded by a fresh one. + pub fn owner_offer_session(&mut self, id: CharacterId, server_id: &str) -> Result { + check_server_id(server_id)?; + let tx = self.write_tx()?; + let record = read_character(&tx, id)?; + let live: Option = tx + .query_row("SELECT session_id FROM owner_session WHERE character_id = ?1 AND server_id = ?2 AND state IN ('offered', 'open')", params![id.to_string(), server_id], |r| r.get(0)) + .optional()?; + if let Some(live) = live { + let row = read_owner_row(&tx, live.parse()?)?.expect("it was just found"); + if row.state == "open" { + return Err(PortableError::SessionOpen); + } + if row.c0_revision == record.revision { + return Ok(offer_from_row(&row)); + } + tx.execute("UPDATE owner_session SET state = 'superseded', updated_at = ?2 WHERE session_id = ?1", params![live, now()])?; + } + let offer = open_offer(&tx, id, server_id)?; + tx.commit()?; + Ok(offer) + } + + /// The Host captured `B0`. Idempotent for the same baseline. + pub fn owner_session_started(&mut self, msg: &PortableSessionStarted) -> Result { + let tx = self.write_tx()?; + let Some(row) = read_owner_row(&tx, msg.session_id)? else { + return Ok(ack(msg.session_id, 0, AckOutcome::Rejected("unknown session".into()), 0, [0; 32])); + }; + let current_hash = head_hash(&tx, row.character_id, row.head_revision)?; + let reject = |why: &str| ack(msg.session_id, 0, AckOutcome::Rejected(why.into()), row.head_revision, current_hash); + if msg.character_id != row.character_id || msg.server_id != row.server_id { + return Ok(reject("the message does not belong to this session")); + } + if msg.base_canonical_revision != row.c0_revision { + return Ok(reject("the baseline was taken against another canonical revision")); + } + let b0_hash = match msg.b0.hash() { + Ok(h) if hex::encode(h) == msg.content_hash => h, + _ => return Ok(reject("the content hash does not match the baseline")), + }; + match row.state.as_str() { + "open" => { + return Ok(if row.b0_hash == Some(b0_hash) && row.generation == Some(msg.baseline_generation) { + ack(msg.session_id, 0, AckOutcome::Duplicate, row.head_revision, current_hash) + } else { + reject("this session already has a different baseline") + }) + } + "offered" => {} + _ => return Ok(ack(msg.session_id, 0, AckOutcome::StaleSession, row.head_revision, current_hash)), + } + let b0 = match msg.b0.open() { + Ok(m) => m, + Err(e) => return Ok(reject(&format!("the baseline cannot be read: {e}"))), + }; + let c0 = snapshot::decode(&row.c0_payload, Some(&row.c0_hash))?; + if b0.character_id != row.character_id || b0.ruleset != c0.ruleset || b0.content_namespace != c0.content_namespace { + return Ok(reject("the baseline is another character or ruleset")); + } + tx.execute( + "UPDATE owner_session SET state = 'open', baseline_generation = ?2, b0_hash = ?3, b0_payload = ?4, updated_at = ?5 WHERE session_id = ?1", + params![msg.session_id.to_string(), msg.baseline_generation, b0_hash.as_slice(), msg.b0.bytes()?, now()], + )?; + tx.commit()?; + Ok(ack(msg.session_id, 0, AckOutcome::Applied, row.head_revision, current_hash)) + } + + /// Apply one checkpoint of the realm: `merge3(C0, B0, B1)`. + pub fn owner_checkpoint(&mut self, msg: &PortableCheckpoint) -> Result { + let keep = self.history_keep()?; + let tx = self.write_tx()?; + let Some(row) = read_owner_row(&tx, msg.session_id)? else { + return Ok(ack(msg.session_id, msg.sequence, AckOutcome::Rejected("unknown session".into()), 0, [0; 32])); + }; + let record = read_character(&tx, row.character_id)?; + let seen_hash = head_hash(&tx, row.character_id, record.revision)?; + let outcome = |o: AckOutcome| ack(msg.session_id, msg.sequence, o, record.revision, seen_hash); + let reject = |why: &str| outcome(AckOutcome::Rejected(why.into())); + + if msg.character_id != row.character_id || msg.server_id != row.server_id { + return Ok(reject("the message does not belong to this session")); + } + if msg.base_canonical_revision != row.c0_revision { + return Ok(reject("the checkpoint was taken against another canonical revision")); + } + let b1_hash = match msg.realm_snapshot.hash() { + Ok(h) if hex::encode(h) == msg.content_hash => h, + _ => return Ok(reject("the content hash does not match the snapshot")), + }; + if msg.sequence == 0 { + return Ok(reject("sequences start at 1")); + } + + // a sequence the Owner has applied before is answered from its log, whatever the session's state is now + let logged: Option<(Vec, i64, i64, Option)> = tx + .query_row("SELECT content_hash, resulting_revision, final, next_session_id FROM owner_checkpoint WHERE session_id = ?1 AND sequence = ?2", params![msg.session_id.to_string(), msg.sequence as i64], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?))) + .optional()?; + if let Some((hash, revision, was_final, next)) = logged { + if hash.as_slice() != b1_hash.as_slice() { + return Ok(reject("this sequence was applied with different content")); + } + let revision = revision as u64; + let mut a = ack(msg.session_id, msg.sequence, AckOutcome::Duplicate, revision, head_hash(&tx, row.character_id, revision)?); + if let Ok((model, _)) = read_snapshot_row(&tx, row.character_id, revision) { + (a.owned_items, a.owned_pets) = owned_ids(&model); + if was_final == 1 { + a.canonical = Some(Envelope::seal(&model)?); + if let Some(next) = next { + a.next_session = Some(NextSession { session_id: next.parse()?, canonical_revision: revision }); + } + } + } + return Ok(a); + } + match row.state.as_str() { + "open" => {} + "offered" => return Ok(reject("the session has not started")), + _ => return Ok(outcome(AckOutcome::StaleSession)), + } + if msg.sequence < row.last_sequence { + return Ok(outcome(AckOutcome::StaleSequence)); + } + if record.revision != row.head_revision { + tx.execute("UPDATE owner_session SET state = 'superseded', updated_at = ?2 WHERE session_id = ?1", params![msg.session_id.to_string(), now()])?; + tx.commit()?; + return Ok(ack(msg.session_id, msg.sequence, AckOutcome::StaleSession, record.revision, head_hash(&self.conn, row.character_id, record.revision)?)); + } + + let b1 = match msg.realm_snapshot.open() { + Ok(m) => m, + Err(e) => return Ok(reject(&format!("the snapshot cannot be read: {e}"))), + }; + let (c0, b0) = (snapshot::decode(&row.c0_payload, Some(&row.c0_hash))?, snapshot::decode(row.b0_payload.as_deref().expect("an open session has B0"), row.b0_hash.as_ref())?); + if b1.character_id != row.character_id || b1.ruleset != c0.ruleset || b1.content_namespace != c0.content_namespace { + return Ok(reject("the snapshot is another character or ruleset")); + } + let merged = match merge3(&c0, &b0, &b1, Mode::Lenient) { + Ok(m) => m.model, + Err(e) => return Ok(reject(&format!("the checkpoint cannot be merged: {e}"))), + }; + let encoded = match snapshot::encode(&merged) { + Ok(e) => e, + Err(e) => return Ok(reject(&format!("the merged character is not valid: {e}"))), + }; + + let at = now(); + let head = head_hash(&tx, row.character_id, record.revision)?; + let revision = if head == encoded.content_hash { + record.revision + } else { + let revision = record.revision + 1; + insert_snapshot(&tx, row.character_id, revision, &encoded, &row.server_id, Some(&format!("session {} checkpoint {}", msg.session_id, msg.sequence)), &at)?; + update_character(&tx, row.character_id, &merged, revision, &at)?; + prune_tx(&tx, row.character_id, revision, keep)?; + revision + }; + + let mut a = ack(msg.session_id, msg.sequence, AckOutcome::Applied, revision, encoded.content_hash); + (a.owned_items, a.owned_pets) = owned_ids(&merged); + let mut next_id = None; + if msg.final_checkpoint { + tx.execute("UPDATE owner_session SET state = 'closed', last_sequence = ?2, head_revision = ?3, updated_at = ?4 WHERE session_id = ?1", params![msg.session_id.to_string(), msg.sequence as i64, revision as i64, at])?; + let offer = open_offer(&tx, row.character_id, &row.server_id)?; + a.canonical = Some(Envelope::from_encoded(&encoded)); + a.next_session = Some(NextSession { session_id: offer.session_id, canonical_revision: revision }); + next_id = Some(offer.session_id.to_string()); + } else { + tx.execute("UPDATE owner_session SET last_sequence = ?2, head_revision = ?3, updated_at = ?4 WHERE session_id = ?1", params![msg.session_id.to_string(), msg.sequence as i64, revision as i64, at])?; + } + tx.execute( + "INSERT INTO owner_checkpoint(session_id, sequence, content_hash, resulting_revision, final, next_session_id, applied_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", + params![msg.session_id.to_string(), msg.sequence as i64, b1_hash.as_slice(), revision as i64, msg.final_checkpoint as i64, next_id, at], + )?; + tx.commit()?; + Ok(a) + } + + /// What the Owner knows of one session: (state, last applied sequence, head revision). + pub fn owner_session_info(&self, session: SessionId) -> Result> { + Ok(read_owner_row(&self.conn, session)?.map(|r| (r.state, r.last_sequence, r.head_revision))) + } +} diff --git a/crates/coa-core/src/portable/store/sync.rs b/crates/coa-core/src/portable/store/sync.rs index 77a7bd8..f9207ba 100644 --- a/crates/coa-core/src/portable/store/sync.rs +++ b/crates/coa-core/src/portable/store/sync.rs @@ -309,36 +309,10 @@ impl Store { /// = **filtered** (kept, mapping stays active); shown by the realm but not owned = **realm-local** (never merged). pub fn capture_baseline(&mut self, id: CharacterId, server_id: &str, input: BaselineInput<'_>) -> Result { check_server_id(server_id)?; - let c0 = self.synced_model(id, server_id)?.ok_or_else(|| PortableError::Invalid("the realm has no synchronised snapshot to use as the session base".into()))?; - if input.b0.character_id != id || input.b0.ruleset != c0.ruleset { - return Err(PortableError::WrongCharacter { expected: id, found: input.b0.character_id }); - } let tx = self.write_tx()?; - let record = read_character(&tx, id)?; - let mapping_revision: i64 = tx.query_row("SELECT last_revision FROM character_server_mapping WHERE character_id = ?1 AND server_id = ?2", params![id.to_string(), server_id], |r| r.get(0))?; - if record.revision != mapping_revision as u64 { - return Err(PortableError::StaleRevision { expected: mapping_revision as u64, current: record.revision }); - } - let open: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM realm_baseline WHERE character_id = ?1 AND server_id = ?2 AND state = 'open')", params![id.to_string(), server_id], |r| r.get(0))?; - if open { - return Err(PortableError::Invalid("this character already has an open session baseline on this realm".into())); - } - let c0_enc = snapshot::encode(&c0)?; - let b0_enc = snapshot::encode(input.b0)?; - let c0_items: HashSet = c0.items.iter().map(|i| i.id).collect(); - let c0_pets: HashSet = c0.pets.iter().map(|p| p.id).collect(); - let protection = Protection { realm_local_items: input.items.iter().map(|o| o.portable_item_id).filter(|i| !c0_items.contains(i)).collect(), canonical_items: c0_items }; - reconcile_with(&tx, id, server_id, record.revision, input.items, &protection)?; - let pet_protection = PetProtection { realm_local_pets: input.pets.iter().map(|o| o.portable_pet_id).filter(|p| !c0_pets.contains(p)).collect(), canonical_pets: c0_pets }; - sync_pets_in_tx(&tx, id, server_id, record.revision, input.pets, &pet_protection)?; - let at = now(); - tx.execute( - "INSERT INTO realm_baseline(character_id, server_id, c0_revision, head_revision, state, c0_hash, c0_payload, b0_hash, b0_payload, created_at, updated_at) - VALUES (?1, ?2, ?3, ?3, 'open', ?4, ?5, ?6, ?7, ?8, ?8)", - params![id.to_string(), server_id, record.revision as i64, c0_enc.content_hash.as_slice(), c0_enc.payload, b0_enc.content_hash.as_slice(), b0_enc.payload, at], - )?; + let baseline = capture_baseline_in_tx(&tx, id, server_id, &input)?; tx.commit()?; - Ok(Baseline { character_id: id, server_id: server_id.to_string(), c0_revision: record.revision, head_revision: record.revision, c0, b0: input.b0.clone(), created_at: at }) + Ok(baseline) } pub fn open_baseline(&self, id: CharacterId, server_id: &str) -> Result> { @@ -407,7 +381,52 @@ impl Store { } } -fn read_open_baseline(conn: &Connection, id: CharacterId, server_id: &str) -> Result> { +pub(super) fn synced_model_in(conn: &Connection, id: CharacterId, server_id: &str) -> Result> { + let row: Option<(Option>, Option>)> = conn + .query_row("SELECT synced_hash, synced_payload FROM character_server_mapping WHERE character_id = ?1 AND server_id = ?2", params![id.to_string(), server_id], |r| Ok((r.get(0)?, r.get(1)?))) + .optional()?; + match row { + Some((Some(hash), Some(payload))) => Ok(Some(snapshot::decode(&payload, Some(&hash_from_blob(hash)?))?)), + Some(_) => Ok(None), + None => Err(PortableError::Invalid(format!("character {id} is not bound to server {server_id}"))), + } +} + +/// Persist the baseline of a session inside a transaction: `C0` is the snapshot the realm was synchronised with, `B0` what the +/// realm shows after its own normalisation. Items and pets the canonical character owns but the realm does not show are +/// **filtered** (kept, mapping stays active); shown but not owned = **realm-local** (never merged). +pub(super) fn capture_baseline_in_tx(tx: &Transaction<'_>, id: CharacterId, server_id: &str, input: &BaselineInput<'_>) -> Result { + let c0 = synced_model_in(tx, id, server_id)?.ok_or_else(|| PortableError::Invalid("the realm has no synchronised snapshot to use as the session base".into()))?; + if input.b0.character_id != id || input.b0.ruleset != c0.ruleset { + return Err(PortableError::WrongCharacter { expected: id, found: input.b0.character_id }); + } + let record = read_character(tx, id)?; + let mapping_revision: i64 = tx.query_row("SELECT last_revision FROM character_server_mapping WHERE character_id = ?1 AND server_id = ?2", params![id.to_string(), server_id], |r| r.get(0))?; + if record.revision != mapping_revision as u64 { + return Err(PortableError::StaleRevision { expected: mapping_revision as u64, current: record.revision }); + } + let open: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM realm_baseline WHERE character_id = ?1 AND server_id = ?2 AND state = 'open')", params![id.to_string(), server_id], |r| r.get(0))?; + if open { + return Err(PortableError::Invalid("this character already has an open session baseline on this realm".into())); + } + let c0_enc = snapshot::encode(&c0)?; + let b0_enc = snapshot::encode(input.b0)?; + let c0_items: HashSet = c0.items.iter().map(|i| i.id).collect(); + let c0_pets: HashSet = c0.pets.iter().map(|p| p.id).collect(); + let protection = Protection { realm_local_items: input.items.iter().map(|o| o.portable_item_id).filter(|i| !c0_items.contains(i)).collect(), canonical_items: c0_items }; + reconcile_with(tx, id, server_id, record.revision, input.items, &protection)?; + let pet_protection = PetProtection { realm_local_pets: input.pets.iter().map(|o| o.portable_pet_id).filter(|p| !c0_pets.contains(p)).collect(), canonical_pets: c0_pets }; + sync_pets_in_tx(tx, id, server_id, record.revision, input.pets, &pet_protection)?; + let at = now(); + tx.execute( + "INSERT INTO realm_baseline(character_id, server_id, c0_revision, head_revision, state, c0_hash, c0_payload, b0_hash, b0_payload, created_at, updated_at) + VALUES (?1, ?2, ?3, ?3, 'open', ?4, ?5, ?6, ?7, ?8, ?8)", + params![id.to_string(), server_id, record.revision as i64, c0_enc.content_hash.as_slice(), c0_enc.payload, b0_enc.content_hash.as_slice(), b0_enc.payload, at], + )?; + Ok(Baseline { character_id: id, server_id: server_id.to_string(), c0_revision: record.revision, head_revision: record.revision, c0, b0: input.b0.clone(), created_at: at }) +} + +pub(super) fn read_open_baseline(conn: &Connection, id: CharacterId, server_id: &str) -> Result> { type Row = (i64, i64, Vec, Vec, Vec, Vec, String); let row: Option = conn .query_row( diff --git a/docs/PORTABLE_SESSIONS.md b/docs/PORTABLE_SESSIONS.md new file mode 100644 index 0000000..53695ac --- /dev/null +++ b/docs/PORTABLE_SESSIONS.md @@ -0,0 +1,201 @@ +# Runtime portable sessions (Phase 5): design and audit + +Code: core `src/server/coa/CoAPortableSession*.{h,cpp}` and `CoAPortableImport*.{h,cpp}` (branch `feat/portable-session-bridge` of the +core fork), Manager `crates/coa-core/src/portable/session/`. CoA only; Wildcard portable transfer stays disabled. + +Phase 4 proved the offline model with a real client. Its one fragile rule is that the baseline `B0` must be captured *after* +the realm's own normalisation and *before* the character is played. Phase 5 makes that rule a property of the running +server instead of a user procedure: + +```text +running worldserver -> automatic B0 -> 60-second checkpoints -> Owner canonical revisions -> final logout checkpoint +``` + +## 1. Roles (the API reflects the real deployment) + +| | OWNER (player's Manager) | HOST (the realm's Manager) | +|---|---|---| +| owns | canonical `PortableCharacter`, revision history, `C0` per session, applied-checkpoint log | realm binding, item/pet mappings, synced snapshot of what the realm was given, outbox | +| does | validates every message, runs `merge3(C0, B0, B1)`, writes revision N+1 | controls the local worldserver, captures `B0`/`B1` from the realm, emits messages | +| never | touches a realm | opens or mutates the Owner's database; holds canonical history | + +Both roles run on `Store` (one schema) in this phase, but the APIs are `OwnerService` and `HostService`, they exchange only +the messages of section 5, and every test runs them on two separate stores (two files, two processes' worth of state). +A later Registry/Relay carries the same messages unchanged. + +## 2. Audit of the login path (what the core actually does) + +`WorldSession::HandlePlayerLoginFromDB` (`Handlers/CharacterHandler.cpp`) is one synchronous function on the world thread: + +1. `Player::LoadFromDB` (the realm's normalisation: unknown spells deleted, default spells/skills/reputation added, honor and title + resets, items that cannot be placed mailed or dropped); +2. `SendInitialPacketsBeforeAddToMap`, `AddPlayerToMap`, `SendInitialPacketsAfterAddToMap`, social/guild/group announcements, + `LoadCorpse`, at-login flags (`resetSpells`, `resetTalents`, first-login spells), `LoadPet`; +3. **`sScriptMgr->OnPlayerLogin`** and **`OnPlayerFirstLogin`** (module normalisation, e.g. the starter-kit restore) as the + *last* statements. + +`OnPlayerLogin` is therefore **not** early (it is the end of the function) but the end of the function **is** the right point: +everything the realm does to a loading character has run, and nothing the player does can have run, because + +* client packets are handled in `WorldSession::Update`, which cannot run before this handler returns (same thread), and +* the map's `Update` (mobs, auras, regeneration, `Player::Update`) runs later in the same world tick. + +So the character state at the end of the handler is exactly "normalised, not yet played". Bots enter through the same function +(`BotMgr` calls `HandlePlayerLoginFromDB`), so the same code serves tests with bot-logged characters. The "login after +disconnect" shortcut (`HandlePlayerLoginToCharInWorld`, a player that never left the world) does not reload the character and +does not start a session. + +Other facts that shape the design: + +* `Player::SaveToDB(trans, create, logout)` is the single place that writes a character; manual save, autosave, logout save, + `.saveall` and checkpoints all go through it. It returns early (a delayed save) while `IsBeingTeleportedFar()`, and it saves the + pet through a **separate** transaction (`Pet::SavePetToDB`). With `CharacterDatabase.WorkerThreads = 1` (the default, and + enforced by the portable subsystem at startup) transactions commit in FIFO order, so the pet's commit precedes ours. +* `CommitTransaction` is asynchronous: an RA reply is not persistence. The Host Manager reads the database and treats a save + as having happened only when it **sees the marker row**. +* The save is a snapshot taken when `SaveToDB` runs, not when the commit lands. That is what makes `B0` exact. +* RA/console commands run on the world thread (`QueueCliCommand`), so a command can safely touch a `Player*`. +* The character list is stale on already connected clients after an import; `CharacterCache` and + `World::UpdateRealmCharCount` must be updated (the importer does, section 7). +* `PlayerDumpReader::LoadDump` is the precedent for the importer's runtime invariants (live generators advanced only after the + commit, name check, cache entry, realm character count). + +## 3. Session marker (core table `coa_portable_session`) + +One row per portable character on the realm (pending SQL `data/sql/updates/pending_db_characters/`): + +| column | meaning | +|---|---| +| `guid` | local character guid (primary key) | +| `session_id`, `character_id` | UUIDv7 strings: the session and the portable character | +| `imported_revision` | canonical revision `C0` that the realm holds (`base_canonical_revision`) | +| `baseline_generation` | increments each time the session is (re)armed | +| `state` | 0 `waiting_baseline`, 1 `baseline_ready` (gated), 2 `active`, 3 `ended` | +| `checkpoint_seq` | the last checkpoint sequence the core saved | +| `save_seq` | incremented by **every** save of this character | +| `updated_at` | unix time | + +The Host arms a session (state 0) in the same realm transaction that imports/updates the character (offline importer, in-place +update) or the core's import service does (online). That is the "pending portable-session marker" the character carries until +its first load. + +### Automatic B0 + +At the end of `HandlePlayerLoginFromDB`, if the character's row is in state 0, the core, in **one** transaction: +`SaveToDB(trans, false, false)` + `UPDATE coa_portable_session SET state=1, baseline_generation=..., save_seq=save_seq+1`, +commits, and **gates** the session: + +* while gated the session's packet queue is not processed (the client stays connected; `CMSG_PING` is answered by the socket + layer), the character is made non-attackable and rooted, nothing the client sends is lost - it is processed after release; +* the gate ends when the Host sends `portable release ` (the Host has seen `state=1`, exported `B0` in one read-only + consistent snapshot, verified inside that snapshot that the marker is still the baseline's, and sent `PortableSessionStarted`); +* if no release arrives within `PortableSession.GateTimeoutSeconds` (default 30) the character is **kicked**, never released: + no gameplay may happen before `B0` exists. The row stays in state 1; the next login re-runs the baseline. + +Characters whose row is in state 2 (a worldserver restart or a relog in the middle of a session) get no baseline, only the +in-memory registration; `B0` and `C0` are already held by the Owner. + +### Every save carries a marker + +`Player::SaveToDB(trans, ...)` appends, for a registered portable character, `UPDATE coa_portable_session SET +save_seq=save_seq+1, updated_at=...` to the **same** transaction (plus `checkpoint_seq=N` when the save is a checkpoint and +`state=3` when it is the logout save). So the rows of a character in a read snapshot always belong to exactly the save whose +marker is in that snapshot; an autosave between two checkpoints is visible as a higher `save_seq`. + +### Checkpoint (typed core operation) + +`portable checkpoint ` (console/RA only): the character is addressed by guid, never by name; +the command verifies the player is online, a portable session is registered, `session_id` matches, the state is 2, the +sequence is greater than `checkpoint_seq`, and the character is not mid-teleport (`BUSY`, retry). Then **one** `SaveToDB` +transaction with the checkpoint marker, committed. The reply is `QUEUED `; it is **not** an acknowledgement of +persistence. `.saveall` is never used, so one checkpoint writes one character (verified, section 9). + +Logout: the logout save itself carries `state=3`; the Host exports and sends the final checkpoint. + +## 4. Host flow + +```text +import / update of a portable character -> arm session (state 0), outbox: nothing yet +poll realm every ~1 s (cheap indexed read of coa_portable_session): + state 1, no PortableSessionStarted sent -> read B0 (consistent snapshot, marker re-checked) -> persist -> release (RA) + -> emit PortableSessionStarted + state 2, online, every 60 s -> RA checkpoint(seq) -> wait for checkpoint_seq == seq -> read B1 -> + persist in outbox -> emit PortableCheckpoint(seq) + state 3 (logout) or character offline -> read final B1 -> emit PortableCheckpoint(final) -> on ack: re-arm (state 0) +``` + +A failed intermediate checkpoint changes nothing canonical and is retried at the next tick. The sequence number is persisted +**before** the RA call, so a Host restart resumes with the same number (the marker tells whether the save happened). +Messages wait in the outbox until acknowledged; redelivery is harmless (section 6). + +## 5. Messages (transport-neutral, versioned, JSON) + +`protocol_version = 1`; each message is canonical JSON with unknown fields refused. Snapshots travel in the existing +snapshot envelope (zstd of canonical JSON, SHA-256 of the canonical JSON). + +```text +PortableSessionStarted { protocol_version, session_id, character_id, server_id, base_canonical_revision, + baseline_generation, b0: envelope, content_hash } +PortableCheckpoint { protocol_version, session_id, character_id, server_id, base_canonical_revision, + sequence, final_checkpoint, realm_snapshot: envelope, content_hash } +OwnerAck { protocol_version, session_id, sequence, outcome, canonical_revision, canonical_hash, + [final: canonical snapshot envelope] } +``` + +`outcome` is `applied`, `duplicate` (same sequence and hash, already applied), `stale_sequence`, `stale_session` +(superseded: another session or revision), or `rejected(reason)`. + +The Owner validates, **before** `merge3`: protocol version; message size and decompression limits; `content_hash` against the +payload; `character_id` exists and the ruleset is CoA; the session is known (opened by `PortableSessionStarted` whose +`base_canonical_revision` equals the revision the Owner recorded as `C0` for that `server_id`); `server_id` and +`character_id` match the session; the payload's own `character_id`/ruleset match; the sequence. + +## 6. Ordering, idempotency, no double counting + +* Per session the Owner keeps `last_sequence` and a log of `(sequence, content_hash, resulting_revision)`. +* sequence = `last_sequence + 1`: applied. Same sequence with the same hash: `duplicate`, nothing happens, the original result + is returned (lost acknowledgements are healed by redelivery). Same sequence, different hash: `rejected`. + `sequence <= last_sequence` otherwise: `stale_sequence`, ignored. A gap (`sequence > last_sequence + 1`) is accepted only when + the Owner holds a later state (checkpoints are snapshots, not deltas, so skipping one loses nothing). +* **Every checkpoint is `merge3(C0, B0, current B1)`**, anchored to the session's original `C0` and `B0` (Phase 4's + property), never `merge(previous canonical, previous B1, B1)`. A repeated or re-sent state therefore cannot add money, honor or xp + twice; the canonical revision can advance while the anchors stay. +* A superseded session (the character's canonical revision no longer equals the session head) refuses checkpoints + (`stale_session`); no divergent-state merging is invented here. +* Crash: the last acknowledged checkpoint is canonical. A worldserver crash leaves the row in state 2 and the character offline: + the Host marks nothing, the next login continues the same session (no new baseline). A final export after a crash is a + normal offline export of the last committed save. + +## 7. PortableImportService (production import without stopping the realm) + +Follows `PORTABLE_IMPORT_SERVICE.md` with one concrete decision: the **job file is canonical JSON, not zstd**, because the core +has no zstd or JSON dependency and a hand-rolled bounded strict parser is easier to audit than adding either. The size cap +(16 MiB, as the Manager's decompression cap) is enforced before parsing; the header carries the SHA-256 the file must match. + +```text +Manager writes /.job (temp name + rename) +-> RA: portable import (job_id = UUIDv7, nothing else) +-> world thread: validate -> allocate from live generators -> prepared statements -> ONE CharacterDatabaseTransaction + (character, items, spells, ..., session marker in state 0) -> commit +-> after the commit only: CharacterCache entry, generators advanced, UpdateRealmCharCount +-> /.result (temp + rename); RA reply "OK " or "ERR " +``` + +No path, SQL or snapshot travels through RA. Idempotency: the marker row carries the job nonce; re-running a finished job +rewrites the same result. Never touches a character that is online or already exists. + +## 8. What is not in this phase + +VPS Registry/Relay and any network transport, public server list, remote account provisioning, Wildcard, vanity/wardrobe +transfer, personal Ascension bank, equipment sets, level-cap projection. + +## 9. Verification plan + +* Owner/Host on two stores: canonical revision 10 -> running realm -> automatic `B0` before gameplay -> play -> checkpoint 1 + (rev 11) -> same checkpoint again (still 11) -> more play -> checkpoint 2 (rev 12) -> delayed checkpoint 1 refused -> + final checkpoint. +* Money/item/spell/quest/pet progression; realm normalisation not propagated; retry; lost acknowledgement; Host, Owner and + worldserver restart; stale session; wrong character or session; malformed message; over-limit payload; logout during a + checkpoint; disconnect/crash before the final checkpoint. +* Real worldserver with the character online and several checkpoints without logout; the number of `characters` rows written + during a checkpoint is counted on the database (general log) to show one character is saved, not all.