Seal the replacement world owner

This commit is contained in:
Meta Vens 2026-09-01 16:18:15 +02:00
commit b53251f94f
3 changed files with 1323 additions and 0 deletions

View file

@ -34,5 +34,6 @@ pub mod vault;
#[cfg(windows)]
#[cfg(windows)]
pub mod windows_secret;
mod world;
pub use kernel::{KernelError, WorldKernel};

View file

@ -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<CommandId, WorldCommit>,
health: Cell<JournalHealth>,
#[cfg(test)]
fail_after_durable_commit: bool,
}
impl WorldJournal {
pub(super) fn open_for_creation(
path: &Path,
creation_id: CommandId,
creation_digest: &str,
) -> Result<CreationOpen, 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)?;
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(&current, 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<WorldJournal, JournalError> {
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<CultCacheEnvelope>,
expected_world_id: Option<WorldId>,
) -> Result<
(
CultCacheEnvelope,
WorldState,
BTreeMap<CommandId, WorldCommit>,
),
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<CommandId, WorldCommit>,
) -> 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<T: Serialize>(
row_type: &str,
schema: &str,
key: String,
value: &T,
) -> Result<CultCacheEnvelope, JournalError> {
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<T: DeserializeOwned>(row: &CultCacheEnvelope) -> Result<T, JournalError> {
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(_))
));
}
}

View file

@ -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<String>) -> 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<String>,
}
impl WorldState {
fn genesis(world_id: WorldId, owner: PrincipalId, title: String) -> Result<Self, KernelError> {
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<u64>,
resulting_revision: u64,
previous_state_digest: Option<String>,
resulting_state_digest: String,
previous_commit_digest: Option<String>,
effect: WorldEffect,
committed_at: DateTime<Utc>,
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<String>,
}
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<journal::JournalError> 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<Path>,
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<Path>,
expected_world_id: WorldId,
) -> Result<Self, KernelError> {
let (journal, state) = journal::WorldJournal::open(path.as_ref(), expected_world_id)?;
Ok(Self {
aggregate: WorldAggregate { state, journal },
})
}
pub(crate) fn snapshot(&self) -> Result<WorldSnapshot, KernelError> {
self.aggregate.journal.ensure_healthy()?;
Ok(WorldSnapshot::from(&self.aggregate.state))
}
pub(crate) fn submit(
&mut self,
command: CommandEnvelope,
authenticated: &AuthenticatedPrincipal,
) -> Result<SubmitReceipt, KernelError> {
self.aggregate.submit(command, authenticated, Utc::now())
}
}
impl WorldAggregate {
fn submit(
&mut self,
command: CommandEnvelope,
authenticated: &AuthenticatedPrincipal,
committed_at: DateTime<Utc>,
) -> Result<SubmitReceipt, KernelError> {
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<String, KernelError> {
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<T: Serialize>(value: &T) -> Result<String, KernelError> {
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<String, KernelError> {
let mut value = state.clone();
value.state_digest.clear();
value.last_commit_digest = None;
digest(&value)
}
fn commit_digest(commit: &WorldCommit) -> Result<String, KernelError> {
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);
}
}