diff --git a/crates/ghostlight-dungeon/src/lib.rs b/crates/ghostlight-dungeon/src/lib.rs index ae16f73..765e328 100644 --- a/crates/ghostlight-dungeon/src/lib.rs +++ b/crates/ghostlight-dungeon/src/lib.rs @@ -34,5 +34,6 @@ pub mod vault; #[cfg(windows)] #[cfg(windows)] pub mod windows_secret; +mod world; pub use kernel::{KernelError, WorldKernel}; diff --git a/crates/ghostlight-dungeon/src/world/journal.rs b/crates/ghostlight-dungeon/src/world/journal.rs new file mode 100644 index 0000000..d0a056e --- /dev/null +++ b/crates/ghostlight-dungeon/src/world/journal.rs @@ -0,0 +1,539 @@ +use super::{ + COMMIT_SCHEMA, CommandId, KernelError, STATE_SCHEMA, WorldCommit, WorldEffect, WorldId, + WorldState, apply_effect, commit_digest, state_digest, validate_principal, +}; +use chrono::Utc; +use cultcache_legacy::{CacheBackingStore, CultCacheEnvelope, OwnedRedbMessagePackBackingStore}; +use serde::{Serialize, de::DeserializeOwned}; +use std::{cell::Cell, collections::BTreeMap, path::Path}; +use thiserror::Error; + +const STATE_ROW: &str = "world_state.foundation.v0"; +const COMMIT_ROW: &str = "world_commit.foundation.v0"; + +#[derive(Debug, Error)] +pub(super) enum JournalError { + #[error("world store is not empty")] + NotEmpty, + #[error("opened store contains another world")] + WorldMismatch, + #[error("world creation ID was reused with different content")] + CreationConflict, + #[error("world ownership is uncertain after command {command_id:?}; drop and reopen")] + RecoveryRequired { command_id: CommandId }, + #[error("world store path no longer names the owned database")] + OwnershipLost, + #[error("world store operation failed: {0}")] + Store(String), + #[error("world journal is corrupt: {0}")] + Corrupt(String), +} + +pub(super) enum CreationOpen { + Empty(EmptyWorldJournal), + Existing { + journal: WorldJournal, + state: WorldState, + }, +} + +pub(super) struct EmptyWorldJournal { + store: OwnedRedbMessagePackBackingStore, +} + +#[derive(Clone, Copy)] +enum JournalHealth { + Healthy, + RecoveryRequired(CommandId), + OwnershipLost, +} + +pub(super) struct WorldJournal { + store: OwnedRedbMessagePackBackingStore, + state_row: CultCacheEnvelope, + commits: BTreeMap, + health: Cell, + #[cfg(test)] + fail_after_durable_commit: bool, +} + +impl WorldJournal { + pub(super) fn open_for_creation( + path: &Path, + creation_id: CommandId, + creation_digest: &str, + ) -> Result { + let store = OwnedRedbMessagePackBackingStore::new(path) + .map_err(|error| JournalError::Store(error.to_string()))?; + store + .validate_path_identity() + .map_err(|_| JournalError::OwnershipLost)?; + let rows = store + .pull_all() + .map_err(|error| JournalError::Store(error.to_string()))?; + store + .validate_path_identity() + .map_err(|_| JournalError::OwnershipLost)?; + if rows.is_empty() { + return Ok(CreationOpen::Empty(EmptyWorldJournal { store })); + } + let (state_row, state, commits) = recover(rows, None)?; + let Some(genesis) = commits.get(&creation_id) else { + return Err(JournalError::NotEmpty); + }; + if genesis.command_digest != creation_digest + || !matches!(&genesis.effect, WorldEffect::WorldCreated { .. }) + { + return Err(JournalError::CreationConflict); + } + Ok(CreationOpen::Existing { + journal: Self { + store, + state_row, + commits, + health: Cell::new(JournalHealth::Healthy), + #[cfg(test)] + fail_after_durable_commit: false, + }, + state, + }) + } + + pub(super) fn open( + path: &Path, + expected_world_id: WorldId, + ) -> Result<(Self, WorldState), JournalError> { + let store = OwnedRedbMessagePackBackingStore::new(path) + .map_err(|error| JournalError::Store(error.to_string()))?; + store + .validate_path_identity() + .map_err(|_| JournalError::OwnershipLost)?; + let rows = store + .pull_all() + .map_err(|error| JournalError::Store(error.to_string()))?; + store + .validate_path_identity() + .map_err(|_| JournalError::OwnershipLost)?; + let (state_row, state, commits) = recover(rows, Some(expected_world_id))?; + Ok(( + Self { + store, + state_row, + commits, + health: Cell::new(JournalHealth::Healthy), + #[cfg(test)] + fail_after_durable_commit: false, + }, + state, + )) + } + + pub(super) fn ensure_healthy(&self) -> Result<(), JournalError> { + match self.health.get() { + JournalHealth::RecoveryRequired(command_id) => { + return Err(JournalError::RecoveryRequired { command_id }); + } + JournalHealth::OwnershipLost => return Err(JournalError::OwnershipLost), + JournalHealth::Healthy => {} + } + if self.store.validate_path_identity().is_err() { + self.health.set(JournalHealth::OwnershipLost); + return Err(JournalError::OwnershipLost); + } + Ok(()) + } + + pub(super) fn commit( + &mut self, + next: &WorldState, + commit: &WorldCommit, + ) -> Result<(), JournalError> { + self.ensure_healthy()?; + if self.commits.contains_key(&commit.command_id) { + return Err(JournalError::Corrupt( + "single-owner append attempted to duplicate a committed command".into(), + )); + } + let current: WorldState = decode(&self.state_row)?; + verify_append(¤t, next, commit)?; + let next_row = envelope(STATE_ROW, STATE_SCHEMA, next.world_id.key(), next)?; + let commit_row = envelope(COMMIT_ROW, COMMIT_SCHEMA, commit.command_id.key(), commit)?; + self.health + .set(JournalHealth::RecoveryRequired(commit.command_id)); + let swapped = self.store.compare_and_swap_batch( + std::slice::from_ref(&self.state_row), + vec![next_row.clone(), commit_row], + ); + match swapped { + Ok(true) => {} + Ok(false) | Err(_) => { + return Err(JournalError::RecoveryRequired { + command_id: commit.command_id, + }); + } + } + #[cfg(test)] + if self.fail_after_durable_commit { + return Err(JournalError::RecoveryRequired { + command_id: commit.command_id, + }); + } + if self.store.validate_path_identity().is_err() { + return Err(JournalError::RecoveryRequired { + command_id: commit.command_id, + }); + } + self.state_row = next_row; + self.commits.insert(commit.command_id, commit.clone()); + self.health.set(JournalHealth::Healthy); + Ok(()) + } + + pub(super) fn commit_for(&self, command_id: CommandId) -> Option<&WorldCommit> { + self.commits.get(&command_id) + } + + #[cfg(test)] + pub(super) fn commit_count(&self) -> usize { + self.commits.len() + } + + #[cfg(test)] + pub(super) fn fail_after_durable_commit_for_test(&mut self) { + self.fail_after_durable_commit = true; + } +} + +impl EmptyWorldJournal { + pub(super) fn initialize( + self, + state: &WorldState, + genesis: &WorldCommit, + ) -> Result { + self.store + .validate_path_identity() + .map_err(|_| JournalError::OwnershipLost)?; + let mut commits = BTreeMap::new(); + commits.insert(genesis.command_id, genesis.clone()); + verify_history(state, &commits)?; + let state_row = envelope(STATE_ROW, STATE_SCHEMA, state.world_id.key(), state)?; + let commit_row = envelope(COMMIT_ROW, COMMIT_SCHEMA, genesis.command_id.key(), genesis)?; + let inserted = self + .store + .append_if_snapshot_unchanged(&[], vec![state_row.clone(), commit_row]) + .map_err(|error| JournalError::Store(error.to_string()))?; + if !inserted { + return Err(JournalError::NotEmpty); + } + if self.store.validate_path_identity().is_err() { + return Err(JournalError::RecoveryRequired { + command_id: genesis.command_id, + }); + } + Ok(WorldJournal { + store: self.store, + state_row, + commits, + health: Cell::new(JournalHealth::Healthy), + #[cfg(test)] + fail_after_durable_commit: false, + }) + } +} + +fn recover( + rows: Vec, + expected_world_id: Option, +) -> Result< + ( + CultCacheEnvelope, + WorldState, + BTreeMap, + ), + JournalError, +> { + let mut state_rows = Vec::new(); + let mut commits = BTreeMap::new(); + for row in rows { + match row.r#type.as_str() { + STATE_ROW => { + require_schema(&row, STATE_SCHEMA)?; + state_rows.push(row); + } + COMMIT_ROW => { + require_schema(&row, COMMIT_SCHEMA)?; + let commit: WorldCommit = decode(&row)?; + if row.key != commit.command_id.key() { + return Err(JournalError::Corrupt( + "commit row key does not match command ID".into(), + )); + } + if commits.insert(commit.command_id, commit).is_some() { + return Err(JournalError::Corrupt("duplicate command commit".into())); + } + } + other => { + return Err(JournalError::Corrupt(format!( + "unadmitted row type {other}" + ))); + } + } + } + if state_rows.len() != 1 { + return Err(JournalError::Corrupt(format!( + "expected one state row, found {}", + state_rows.len() + ))); + } + let state_row = state_rows.pop().expect("length checked"); + let state: WorldState = decode(&state_row)?; + if state_row.key != state.world_id.key() { + return Err(JournalError::Corrupt( + "state row key does not match its world ID".into(), + )); + } + if expected_world_id.is_some_and(|expected| state.world_id != expected) { + return Err(JournalError::WorldMismatch); + } + verify_history(&state, &commits)?; + Ok((state_row, state, commits)) +} + +fn verify_append( + current: &WorldState, + next: &WorldState, + commit: &WorldCommit, +) -> Result<(), JournalError> { + verify_state_shape(current)?; + verify_state_shape(next)?; + if current.world_id != next.world_id + || commit.world_id != current.world_id + || commit.schema != COMMIT_SCHEMA + || commit.previous_revision != Some(current.revision) + || commit.resulting_revision != next.revision + || next.revision != current.revision.saturating_add(1) + || commit.previous_state_digest.as_deref() != Some(current.state_digest.as_str()) + || commit.resulting_state_digest != next.state_digest + || commit.previous_commit_digest != current.last_commit_digest + || next.last_commit_digest.as_deref() != Some(commit.digest.as_str()) + { + return Err(JournalError::Corrupt( + "commit does not bind the exact state transition".into(), + )); + } + if state_digest(current).map_err(kernel_error)? != current.state_digest + || state_digest(next).map_err(kernel_error)? != next.state_digest + || commit_digest(commit).map_err(kernel_error)? != commit.digest + { + return Err(JournalError::Corrupt("digest mismatch".into())); + } + let mut replay = current.clone(); + apply_effect(&mut replay, &commit.effect).map_err(kernel_error)?; + replay.revision = next.revision; + replay.state_digest = state_digest(&replay).map_err(kernel_error)?; + replay.last_commit_digest = Some(commit.digest.clone()); + if &replay != next { + return Err(JournalError::Corrupt( + "commit effect does not reproduce next state".into(), + )); + } + Ok(()) +} + +fn verify_history( + state: &WorldState, + commits: &BTreeMap, +) -> Result<(), JournalError> { + verify_state_shape(state)?; + let expected_commits = usize::try_from(state.revision) + .map_err(|_| JournalError::Corrupt("revision does not fit this runtime".into()))?; + let expected_commits = expected_commits + .checked_add(1) + .ok_or_else(|| JournalError::Corrupt("commit count overflow".into()))?; + if commits.len() != expected_commits { + return Err(JournalError::Corrupt( + "commit count does not equal genesis plus head revision".into(), + )); + } + let mut ordered: Vec<_> = commits.values().collect(); + ordered.sort_by_key(|commit| commit.resulting_revision); + let genesis = ordered + .first() + .ok_or_else(|| JournalError::Corrupt("world has no genesis commit".into()))?; + let WorldEffect::WorldCreated { owner, title } = &genesis.effect else { + return Err(JournalError::Corrupt( + "first commit is not immutable world genesis".into(), + )); + }; + if genesis.schema != COMMIT_SCHEMA + || genesis.world_id != state.world_id + || genesis.previous_revision.is_some() + || genesis.resulting_revision != 0 + || genesis.previous_state_digest.is_some() + || genesis.previous_commit_digest.is_some() + || commit_digest(genesis).map_err(kernel_error)? != genesis.digest + { + return Err(JournalError::Corrupt( + "genesis commit is not canonical or verifiable".into(), + )); + } + let mut replay = + WorldState::genesis(state.world_id, owner.clone(), title.clone()).map_err(kernel_error)?; + verify_state_shape(&replay)?; + if replay.state_digest != genesis.resulting_state_digest { + return Err(JournalError::Corrupt( + "genesis resulting state digest is invalid".into(), + )); + } + replay.last_commit_digest = Some(genesis.digest.clone()); + + for commit in ordered.into_iter().skip(1) { + if commit.schema != COMMIT_SCHEMA + || commit.world_id != state.world_id + || commit.previous_revision != Some(replay.revision) + || commit.resulting_revision != replay.revision.saturating_add(1) + || commit.previous_state_digest.as_deref() != Some(replay.state_digest.as_str()) + || commit.previous_commit_digest != replay.last_commit_digest + || commit_digest(commit).map_err(kernel_error)? != commit.digest + { + return Err(JournalError::Corrupt( + "commit chain is not contiguous or verifiable".into(), + )); + } + apply_effect(&mut replay, &commit.effect).map_err(kernel_error)?; + replay.revision = commit.resulting_revision; + replay.state_digest = state_digest(&replay).map_err(kernel_error)?; + if replay.state_digest != commit.resulting_state_digest { + return Err(JournalError::Corrupt( + "commit resulting state digest is invalid".into(), + )); + } + replay.last_commit_digest = Some(commit.digest.clone()); + } + if &replay != state { + return Err(JournalError::Corrupt( + "head state does not equal replayed history".into(), + )); + } + Ok(()) +} + +fn verify_state_shape(state: &WorldState) -> Result<(), JournalError> { + if state.schema != STATE_SCHEMA { + return Err(JournalError::Corrupt("invalid state schema".into())); + } + if state.title.trim().is_empty() || state.title.trim() != state.title { + return Err(JournalError::Corrupt( + "world title is empty or noncanonical".into(), + )); + } + validate_principal(&state.owner).map_err(kernel_error)?; + Ok(()) +} + +fn require_schema(row: &CultCacheEnvelope, schema: &str) -> Result<(), JournalError> { + if row.schema_id.as_deref() != Some(schema) { + Err(JournalError::Corrupt(format!( + "row {}/{} has the wrong schema", + row.r#type, row.key + ))) + } else { + Ok(()) + } +} + +fn envelope( + row_type: &str, + schema: &str, + key: String, + value: &T, +) -> Result { + Ok(CultCacheEnvelope { + key, + r#type: row_type.into(), + payload: rmp_serde::to_vec_named(value) + .map_err(|error| JournalError::Store(error.to_string()))?, + stored_at: Utc::now().to_rfc3339(), + schema_id: Some(schema.into()), + }) +} + +fn decode(row: &CultCacheEnvelope) -> Result { + rmp_serde::from_slice(&row.payload).map_err(|error| { + JournalError::Corrupt(format!( + "could not decode row {}/{}: {error}", + row.r#type, row.key + )) + }) +} + +fn kernel_error(error: KernelError) -> JournalError { + JournalError::Corrupt(error.to_string()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::world::PrincipalId; + + #[test] + fn a_later_effect_cannot_repair_noncanonical_genesis() { + let world_id = WorldId::issue(); + let owner = PrincipalId::new("owner@example.test"); + let creation_id = CommandId::new(); + let mut genesis_state = + WorldState::genesis(world_id, owner.clone(), " Bad Genesis ".into()).unwrap(); + let mut genesis = WorldCommit { + schema: COMMIT_SCHEMA.into(), + world_id, + command_id: creation_id, + command_digest: "sha256:create".into(), + previous_revision: None, + resulting_revision: 0, + previous_state_digest: None, + resulting_state_digest: genesis_state.state_digest.clone(), + previous_commit_digest: None, + effect: WorldEffect::WorldCreated { + owner, + title: genesis_state.title.clone(), + }, + committed_at: Utc::now(), + digest: String::new(), + }; + genesis.digest = commit_digest(&genesis).unwrap(); + genesis_state.last_commit_digest = Some(genesis.digest.clone()); + + let effect = WorldEffect::TitleChanged { + previous_title: genesis_state.title.clone(), + resulting_title: "Canonical".into(), + }; + let mut head = genesis_state.clone(); + apply_effect(&mut head, &effect).unwrap(); + head.revision = 1; + head.state_digest = state_digest(&head).unwrap(); + let mut correction = WorldCommit { + schema: COMMIT_SCHEMA.into(), + world_id, + command_id: CommandId::new(), + command_digest: "sha256:correct".into(), + previous_revision: Some(0), + resulting_revision: 1, + previous_state_digest: Some(genesis_state.state_digest.clone()), + resulting_state_digest: head.state_digest.clone(), + previous_commit_digest: Some(genesis.digest.clone()), + effect, + committed_at: Utc::now(), + digest: String::new(), + }; + correction.digest = commit_digest(&correction).unwrap(); + head.last_commit_digest = Some(correction.digest.clone()); + + let mut commits = BTreeMap::new(); + commits.insert(genesis.command_id, genesis); + commits.insert(correction.command_id, correction); + assert!(matches!( + verify_history(&head, &commits), + Err(JournalError::Corrupt(_)) + )); + } +} diff --git a/crates/ghostlight-dungeon/src/world/mod.rs b/crates/ghostlight-dungeon/src/world/mod.rs new file mode 100644 index 0000000..8a472fc --- /dev/null +++ b/crates/ghostlight-dungeon/src/world/mod.rs @@ -0,0 +1,783 @@ +//! Sealed replacement world owner under construction. +//! +//! Nothing in this module is a public runtime path yet. The old kernel remains +//! the live crate export until this owner proves persistence and recovery, then +//! the old authority is deleted and this facade takes its name. + +mod journal; + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use std::path::Path; +use thiserror::Error; +use uuid::Uuid; + +const STATE_SCHEMA: &str = "ghostlight.world_state.foundation.v0"; +const COMMIT_SCHEMA: &str = "ghostlight.world_commit.foundation.v0"; + +macro_rules! opaque_uuid { + ($name:ident) => { + #[derive( + Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord, Hash, + )] + #[serde(transparent)] + pub(crate) struct $name(Uuid); + + impl $name { + fn issue() -> Self { + Self(Uuid::new_v4()) + } + + fn key(self) -> String { + self.0.to_string() + } + } + }; +} + +opaque_uuid!(WorldId); +opaque_uuid!(CommandId); + +impl CommandId { + pub(crate) fn new() -> Self { + Self::issue() + } +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)] +#[serde(transparent)] +pub(crate) struct PrincipalId(String); + +impl PrincipalId { + pub(crate) fn new(value: impl Into) -> Self { + Self(value.into()) + } +} + +#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub(crate) enum WorldPhase { + Draft, + Active, + Archived, +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +pub(crate) struct CreateWorld { + pub(crate) id: CommandId, + pub(crate) owner: PrincipalId, + pub(crate) title: String, +} + +/// Sealed identity evidence. There is deliberately no production constructor +/// until the Heimdall/app-session owner moves inside the replacement ingress. +pub(crate) struct AuthenticatedPrincipal { + principal: PrincipalId, +} + +impl AuthenticatedPrincipal { + #[cfg(test)] + fn fixture(principal: PrincipalId) -> Self { + Self { principal } + } +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +pub(crate) struct CommandEnvelope { + pub(crate) id: CommandId, + pub(crate) world_id: WorldId, + pub(crate) expected_revision: u64, + pub(crate) principal: PrincipalId, + pub(crate) body: CommandBody, +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(tag = "type", rename_all = "snake_case")] +pub(crate) enum CommandBody { + SetTitle { title: String }, +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +struct WorldState { + schema: String, + world_id: WorldId, + revision: u64, + phase: WorldPhase, + owner: PrincipalId, + title: String, + state_digest: String, + last_commit_digest: Option, +} + +impl WorldState { + fn genesis(world_id: WorldId, owner: PrincipalId, title: String) -> Result { + let mut state = Self { + schema: STATE_SCHEMA.into(), + world_id, + revision: 0, + phase: WorldPhase::Draft, + owner, + title, + state_digest: String::new(), + last_commit_digest: None, + }; + state.state_digest = state_digest(&state)?; + Ok(state) + } +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(tag = "type", rename_all = "snake_case")] +enum WorldEffect { + WorldCreated { + owner: PrincipalId, + title: String, + }, + TitleChanged { + previous_title: String, + resulting_title: String, + }, +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +struct WorldCommit { + schema: String, + world_id: WorldId, + command_id: CommandId, + command_digest: String, + previous_revision: Option, + resulting_revision: u64, + previous_state_digest: Option, + resulting_state_digest: String, + previous_commit_digest: Option, + effect: WorldEffect, + committed_at: DateTime, + digest: String, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct WorldSnapshot { + pub(crate) world_id: WorldId, + pub(crate) revision: u64, + pub(crate) phase: WorldPhase, + pub(crate) title: String, + pub(crate) state_digest: String, + pub(crate) last_commit_digest: Option, +} + +impl From<&WorldState> for WorldSnapshot { + fn from(state: &WorldState) -> Self { + Self { + world_id: state.world_id, + revision: state.revision, + phase: state.phase, + title: state.title.clone(), + state_digest: state.state_digest.clone(), + last_commit_digest: state.last_commit_digest.clone(), + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct CommitReceipt { + pub(crate) command_id: CommandId, + pub(crate) resulting_revision: u64, + pub(crate) resulting_state_digest: String, + pub(crate) commit_digest: String, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct CreationReceipt { + pub(crate) command_id: CommandId, + pub(crate) world_id: WorldId, + pub(crate) resulting_state_digest: String, + pub(crate) commit_digest: String, +} + +impl From<&WorldCommit> for CreationReceipt { + fn from(commit: &WorldCommit) -> Self { + Self { + command_id: commit.command_id, + world_id: commit.world_id, + resulting_state_digest: commit.resulting_state_digest.clone(), + commit_digest: commit.digest.clone(), + } + } +} + +impl From<&WorldCommit> for CommitReceipt { + fn from(commit: &WorldCommit) -> Self { + Self { + command_id: commit.command_id, + resulting_revision: commit.resulting_revision, + resulting_state_digest: commit.resulting_state_digest.clone(), + commit_digest: commit.digest.clone(), + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) enum SubmitReceipt { + Applied(CommitReceipt), + AlreadyApplied(CommitReceipt), + NoEffect(WorldSnapshot), +} + +#[derive(Debug, Error)] +pub(crate) enum KernelError { + #[error("world title must not be empty")] + EmptyTitle, + #[error("world owner principal must not be empty")] + EmptyPrincipal, + #[error("command targets another world")] + WorldMismatch, + #[error("authenticated principal does not match the command")] + AuthenticationMismatch, + #[error("principal does not own this world")] + Unauthorized, + #[error("expected revision {expected}, current revision {actual}")] + RevisionMismatch { expected: u64, actual: u64 }, + #[error("command ID was reused with different content")] + CommandIdConflict, + #[error("world creation ID was reused with different content")] + CreationConflict, + #[error("world creation target already contains another world")] + CreationTargetOccupied, + #[error("opened store contains another world")] + OpenedWorldMismatch, + #[error("world ownership is uncertain after command {command_id:?}; drop and reopen")] + RecoveryRequired { command_id: CommandId }, + #[error("world store path no longer names the owned database; drop and reopen explicitly")] + OwnershipLost, + #[error("canonical serialization failed: {0}")] + Serialization(String), + #[error("world store operation failed: {0}")] + Store(String), + #[error("world journal is corrupt: {0}")] + CorruptJournal(String), + #[error("private reducer invariant failed: {0}")] + Invariant(String), +} + +impl From for KernelError { + fn from(error: journal::JournalError) -> Self { + match error { + journal::JournalError::NotEmpty => Self::CreationTargetOccupied, + journal::JournalError::WorldMismatch => Self::OpenedWorldMismatch, + journal::JournalError::CreationConflict => Self::CreationConflict, + journal::JournalError::RecoveryRequired { command_id } => { + Self::RecoveryRequired { command_id } + } + journal::JournalError::OwnershipLost => Self::OwnershipLost, + journal::JournalError::Store(detail) => Self::Store(detail), + journal::JournalError::Corrupt(detail) => Self::CorruptJournal(detail), + } + } +} + +struct WorldAggregate { + state: WorldState, + journal: journal::WorldJournal, +} + +pub(crate) struct WorldKernel { + aggregate: WorldAggregate, +} + +impl WorldKernel { + pub(crate) fn create( + path: impl AsRef, + input: CreateWorld, + authenticated: &AuthenticatedPrincipal, + ) -> Result<(Self, CreationReceipt), KernelError> { + validate_principal(&input.owner)?; + if input.owner != authenticated.principal { + return Err(KernelError::AuthenticationMismatch); + } + let title = normalize_title(&input.title)?; + let creation_digest = digest(&input)?; + let (journal, state) = match journal::WorldJournal::open_for_creation( + path.as_ref(), + input.id, + &creation_digest, + )? { + journal::CreationOpen::Existing { journal, state } => (journal, state), + journal::CreationOpen::Empty(empty) => { + let mut state = WorldState::genesis(WorldId::issue(), input.owner, title)?; + let mut genesis = WorldCommit { + schema: COMMIT_SCHEMA.into(), + world_id: state.world_id, + command_id: input.id, + command_digest: creation_digest, + previous_revision: None, + resulting_revision: 0, + previous_state_digest: None, + resulting_state_digest: state.state_digest.clone(), + previous_commit_digest: None, + effect: WorldEffect::WorldCreated { + owner: state.owner.clone(), + title: state.title.clone(), + }, + committed_at: Utc::now(), + digest: String::new(), + }; + genesis.digest = commit_digest(&genesis)?; + state.last_commit_digest = Some(genesis.digest.clone()); + let journal = empty.initialize(&state, &genesis)?; + (journal, state) + } + }; + let receipt = journal + .commit_for(input.id) + .map(CreationReceipt::from) + .ok_or_else(|| KernelError::Invariant("world genesis receipt is missing".into()))?; + Ok(( + Self { + aggregate: WorldAggregate { state, journal }, + }, + receipt, + )) + } + + pub(crate) fn open( + path: impl AsRef, + expected_world_id: WorldId, + ) -> Result { + let (journal, state) = journal::WorldJournal::open(path.as_ref(), expected_world_id)?; + Ok(Self { + aggregate: WorldAggregate { state, journal }, + }) + } + + pub(crate) fn snapshot(&self) -> Result { + self.aggregate.journal.ensure_healthy()?; + Ok(WorldSnapshot::from(&self.aggregate.state)) + } + + pub(crate) fn submit( + &mut self, + command: CommandEnvelope, + authenticated: &AuthenticatedPrincipal, + ) -> Result { + self.aggregate.submit(command, authenticated, Utc::now()) + } +} + +impl WorldAggregate { + fn submit( + &mut self, + command: CommandEnvelope, + authenticated: &AuthenticatedPrincipal, + committed_at: DateTime, + ) -> Result { + self.journal.ensure_healthy()?; + if command.world_id != self.state.world_id { + return Err(KernelError::WorldMismatch); + } + if command.principal != authenticated.principal { + return Err(KernelError::AuthenticationMismatch); + } + if command.principal != self.state.owner { + return Err(KernelError::Unauthorized); + } + let command_digest = digest(&command)?; + if let Some(commit) = self.aggregate_commit(command.id) { + return if commit.command_digest == command_digest { + Ok(SubmitReceipt::AlreadyApplied(CommitReceipt::from(commit))) + } else { + Err(KernelError::CommandIdConflict) + }; + } + if command.expected_revision != self.state.revision { + return Err(KernelError::RevisionMismatch { + expected: command.expected_revision, + actual: self.state.revision, + }); + } + let effect = match &command.body { + CommandBody::SetTitle { title } => { + let title = normalize_title(title)?; + if title == self.state.title { + return Ok(SubmitReceipt::NoEffect(WorldSnapshot::from(&self.state))); + } + WorldEffect::TitleChanged { + previous_title: self.state.title.clone(), + resulting_title: title, + } + } + }; + + let mut candidate = self.state.clone(); + apply_effect(&mut candidate, &effect)?; + candidate.revision = candidate + .revision + .checked_add(1) + .ok_or_else(|| KernelError::Serialization("world revision overflow".into()))?; + candidate.state_digest = state_digest(&candidate)?; + + let mut commit = WorldCommit { + schema: COMMIT_SCHEMA.into(), + world_id: self.state.world_id, + command_id: command.id, + command_digest, + previous_revision: Some(self.state.revision), + resulting_revision: candidate.revision, + previous_state_digest: Some(self.state.state_digest.clone()), + resulting_state_digest: candidate.state_digest.clone(), + previous_commit_digest: self.state.last_commit_digest.clone(), + effect, + committed_at, + digest: String::new(), + }; + commit.digest = commit_digest(&commit)?; + candidate.last_commit_digest = Some(commit.digest.clone()); + + self.journal.commit(&candidate, &commit)?; + self.state = candidate; + Ok(SubmitReceipt::Applied(CommitReceipt::from(&commit))) + } + + fn aggregate_commit(&self, command_id: CommandId) -> Option<&WorldCommit> { + self.journal.commit_for(command_id) + } +} + +fn normalize_title(value: &str) -> Result { + let value = value.trim(); + if value.is_empty() { + Err(KernelError::EmptyTitle) + } else { + Ok(value.to_owned()) + } +} + +fn validate_principal(value: &PrincipalId) -> Result<(), KernelError> { + if value.0.trim().is_empty() || value.0.trim() != value.0 { + Err(KernelError::EmptyPrincipal) + } else { + Ok(()) + } +} + +fn apply_effect(state: &mut WorldState, effect: &WorldEffect) -> Result<(), KernelError> { + match effect { + WorldEffect::WorldCreated { .. } => { + return Err(KernelError::Invariant( + "world genesis cannot be applied as a mutable effect".into(), + )); + } + WorldEffect::TitleChanged { + previous_title, + resulting_title, + } => { + if &state.title != previous_title { + return Err(KernelError::Invariant( + "title-change effect does not match prior state".into(), + )); + } + if previous_title == resulting_title + || normalize_title(resulting_title)? != *resulting_title + { + return Err(KernelError::Invariant( + "title-change effect is empty or noncanonical".into(), + )); + } + state.title = resulting_title.clone(); + } + } + Ok(()) +} + +fn digest(value: &T) -> Result { + let bytes = rmp_serde::to_vec_named(value) + .map_err(|error| KernelError::Serialization(error.to_string()))?; + Ok(format!("sha256:{:x}", Sha256::digest(bytes))) +} + +fn state_digest(state: &WorldState) -> Result { + let mut value = state.clone(); + value.state_digest.clear(); + value.last_commit_digest = None; + digest(&value) +} + +fn commit_digest(commit: &WorldCommit) -> Result { + let mut value = commit.clone(); + value.digest.clear(); + digest(&value) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn owner() -> PrincipalId { + PrincipalId::new("owner@example.test") + } + + fn auth() -> AuthenticatedPrincipal { + AuthenticatedPrincipal::fixture(owner()) + } + + fn creation(id: CommandId, title: &str) -> CreateWorld { + CreateWorld { + id, + owner: owner(), + title: title.into(), + } + } + + fn command(snapshot: &WorldSnapshot, id: CommandId, title: &str) -> CommandEnvelope { + CommandEnvelope { + id, + world_id: snapshot.world_id, + expected_revision: snapshot.revision, + principal: owner(), + body: CommandBody::SetTitle { + title: title.into(), + }, + } + } + + #[test] + fn create_submit_restart_and_replay_are_exact() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("world.cc"); + let creation = creation(CommandId::new(), " First Title "); + let (mut kernel, creation_receipt) = + WorldKernel::create(&path, creation.clone(), &auth()).unwrap(); + let genesis = kernel.snapshot().unwrap(); + assert_eq!(genesis.title, "First Title"); + assert_eq!(creation_receipt.command_id, creation.id); + assert_eq!(creation_receipt.world_id, genesis.world_id); + assert_eq!( + creation_receipt.resulting_state_digest, + genesis.state_digest + ); + let command = command(&genesis, CommandId::new(), "Second Title"); + let applied = kernel.submit(command.clone(), &auth()).unwrap(); + let SubmitReceipt::Applied(receipt) = applied else { + panic!("expected applied receipt") + }; + let accepted = kernel.snapshot().unwrap(); + assert_eq!(accepted.revision, 1); + drop(kernel); + + let (mut reopened, retried_creation_receipt) = + WorldKernel::create(&path, creation, &auth()).unwrap(); + assert_eq!(retried_creation_receipt, creation_receipt); + assert_eq!(reopened.snapshot().unwrap(), accepted); + assert_eq!( + reopened.submit(command, &auth()).unwrap(), + SubmitReceipt::AlreadyApplied(receipt) + ); + assert_eq!(reopened.snapshot().unwrap(), accepted); + drop(reopened); + + let reopened = WorldKernel::open(&path, genesis.world_id).unwrap(); + assert_eq!(reopened.snapshot().unwrap(), accepted); + } + + #[test] + fn no_effect_invalid_and_stale_commands_do_not_commit() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("world.cc"); + let (mut kernel, _) = + WorldKernel::create(&path, creation(CommandId::new(), "Still"), &auth()).unwrap(); + let genesis = kernel.snapshot().unwrap(); + let no_effect = command(&genesis, CommandId::new(), " Still "); + assert!(matches!( + kernel.submit(no_effect, &auth()).unwrap(), + SubmitReceipt::NoEffect(_) + )); + let invalid = command(&genesis, CommandId::new(), " "); + assert!(matches!( + kernel.submit(invalid, &auth()), + Err(KernelError::EmptyTitle) + )); + assert_eq!(kernel.snapshot().unwrap(), genesis); + assert_eq!(kernel.aggregate.journal.commit_count(), 1); + drop(kernel); + + let mut kernel = WorldKernel::open(&path, genesis.world_id).unwrap(); + assert_eq!(kernel.snapshot().unwrap(), genesis); + assert_eq!(kernel.aggregate.journal.commit_count(), 1); + + let accepted = command(&genesis, CommandId::new(), "Changed"); + kernel.submit(accepted, &auth()).unwrap(); + let after = kernel.snapshot().unwrap(); + let stale = command(&genesis, CommandId::new(), "Stale"); + assert!(matches!( + kernel.submit(stale, &auth()), + Err(KernelError::RevisionMismatch { .. }) + )); + assert_eq!(kernel.snapshot().unwrap(), after); + assert_eq!(kernel.aggregate.journal.commit_count(), 2); + drop(kernel); + + let kernel = WorldKernel::open(&path, genesis.world_id).unwrap(); + assert_eq!(kernel.snapshot().unwrap(), after); + assert_eq!(kernel.aggregate.journal.commit_count(), 2); + } + + #[test] + fn durable_idempotency_precedes_revision_and_detects_conflicts() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("world.cc"); + let (mut kernel, _) = + WorldKernel::create(&path, creation(CommandId::new(), "A"), &auth()).unwrap(); + let genesis = kernel.snapshot().unwrap(); + let id = CommandId::new(); + let first = command(&genesis, id, "B"); + let first_receipt = kernel.submit(first.clone(), &auth()).unwrap(); + let second = command(&kernel.snapshot().unwrap(), CommandId::new(), "C"); + kernel.submit(second, &auth()).unwrap(); + let after_second = kernel.snapshot().unwrap(); + drop(kernel); + let mut kernel = WorldKernel::open(&path, genesis.world_id).unwrap(); + assert_eq!(kernel.snapshot().unwrap(), after_second); + assert_eq!( + kernel.submit(first, &auth()).unwrap(), + match first_receipt { + SubmitReceipt::Applied(value) => SubmitReceipt::AlreadyApplied(value), + _ => unreachable!(), + } + ); + let conflict = command(&genesis, id, "Different"); + assert!(matches!( + kernel.submit(conflict, &auth()), + Err(KernelError::CommandIdConflict) + )); + assert_eq!(kernel.snapshot().unwrap().revision, 2); + } + + #[test] + fn a_second_live_owner_is_refused() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("world.cc"); + let creation = creation(CommandId::new(), "Held"); + let (kernel, _) = WorldKernel::create(&path, creation.clone(), &auth()).unwrap(); + let world_id = kernel.snapshot().unwrap().world_id; + assert!(matches!( + WorldKernel::open(&path, world_id), + Err(KernelError::Store(_)) + )); + assert!(matches!( + WorldKernel::create(&path, creation.clone(), &auth()), + Err(KernelError::Store(_)) + )); + drop(kernel); + let kernel = WorldKernel::open(&path, world_id).unwrap(); + drop(kernel); + assert!(matches!( + WorldKernel::open(&path, WorldId::issue()), + Err(KernelError::OpenedWorldMismatch) + )); + } + + #[test] + fn creation_id_is_exact_and_conflicts_do_not_rewrite_genesis() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("world.cc"); + let creation_id = CommandId::new(); + let (kernel, receipt) = + WorldKernel::create(&path, creation(creation_id, "Original"), &auth()).unwrap(); + let genesis = kernel.snapshot().unwrap(); + drop(kernel); + + assert!(matches!( + WorldKernel::create(&path, creation(creation_id, "Different"), &auth()), + Err(KernelError::CreationConflict) + )); + let (kernel, retried) = + WorldKernel::create(&path, creation(creation_id, "Original"), &auth()).unwrap(); + assert_eq!(retried, receipt); + assert_eq!(kernel.snapshot().unwrap(), genesis); + drop(kernel); + assert!(matches!( + WorldKernel::create( + &path, + creation(CommandId::new(), "Unrelated creation"), + &auth() + ), + Err(KernelError::CreationTargetOccupied) + )); + } + + #[test] + fn lost_post_commit_ack_poisons_until_reopen_and_exact_retry() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("world.cc"); + let (mut kernel, _) = + WorldKernel::create(&path, creation(CommandId::new(), "Certain"), &auth()).unwrap(); + let genesis = kernel.snapshot().unwrap(); + let uncertain = CommandId::new(); + let attempted = command(&genesis, uncertain, "Durably changed"); + kernel + .aggregate + .journal + .fail_after_durable_commit_for_test(); + + assert!(matches!( + kernel.submit(attempted.clone(), &auth()), + Err(KernelError::RecoveryRequired { command_id }) if command_id == uncertain + )); + assert!(matches!( + kernel.snapshot(), + Err(KernelError::RecoveryRequired { command_id }) if command_id == uncertain + )); + drop(kernel); + + let mut reopened = WorldKernel::open(&path, genesis.world_id).unwrap(); + let durable = reopened.snapshot().unwrap(); + assert_eq!(durable.revision, 1); + assert_eq!(durable.title, "Durably changed"); + assert!(matches!( + reopened.submit(attempted, &auth()).unwrap(), + SubmitReceipt::AlreadyApplied(_) + )); + assert_eq!(reopened.snapshot().unwrap(), durable); + } + + #[test] + fn replacing_the_store_path_revokes_the_live_owner() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("world.cc"); + let displaced = dir.path().join("displaced.cc"); + let (mut kernel, _) = + WorldKernel::create(&path, creation(CommandId::new(), "Pinned"), &auth()).unwrap(); + let snapshot = kernel.snapshot().unwrap(); + + std::fs::rename(&path, &displaced).unwrap(); + std::fs::File::create(&path).unwrap(); + assert!(matches!(kernel.snapshot(), Err(KernelError::OwnershipLost))); + assert!(matches!( + kernel.submit(command(&snapshot, CommandId::new(), "Detached"), &auth()), + Err(KernelError::OwnershipLost) + )); + std::fs::remove_file(&path).unwrap(); + std::fs::rename(&displaced, &path).unwrap(); + assert!(matches!(kernel.snapshot(), Err(KernelError::OwnershipLost))); + drop(kernel); + + let reopened = WorldKernel::open(&path, snapshot.world_id).unwrap(); + assert_eq!(reopened.snapshot().unwrap(), snapshot); + } + + #[test] + fn sealed_authentication_cannot_be_replaced_by_a_command_claim() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("world.cc"); + let (mut kernel, _) = + WorldKernel::create(&path, creation(CommandId::new(), "Owned"), &auth()).unwrap(); + let snapshot = kernel.snapshot().unwrap(); + let mut forged = command(&snapshot, CommandId::new(), "Stolen"); + forged.principal = PrincipalId::new("attacker@example.test"); + assert!(matches!( + kernel.submit(forged, &auth()), + Err(KernelError::AuthenticationMismatch) + )); + assert_eq!(kernel.snapshot().unwrap(), snapshot); + } +}