#![cfg_attr(target_arch = "wasm32", allow(dead_code))]
pub mod memory;
#[cfg(feature = "sqlite-store")]
pub mod sqlite;
mod whole_blob_rewrite;
pub use meerkat_core::{HeadCanonicalProvisionalTailAuthority, WholeBlobProvisionalTailAuthority};
pub use whole_blob_rewrite::{
PreparedWholeBlobRewriteBoundary, PreparedWholeBlobRewriteStoreParts,
VerifiedCommittedWholeBlobPayload,
};
use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::Arc;
use meerkat_core::lifecycle::core_executor::BoundSessionCommit;
use meerkat_core::lifecycle::{InputId, RunBoundaryReceipt, RunId};
use sha2::{Digest, Sha256};
use crate::identifiers::{IdempotencyKey, LogicalRuntimeId};
use crate::input_state::{InputStatePersistenceRecord, StoredInputState};
use crate::runtime_state::RuntimeState;
const LEGACY_MACHINE_LIFECYCLE_STORE_RECORD_VERSION: u16 = 1;
const SUPERVISOR_MACHINE_LIFECYCLE_STORE_RECORD_VERSION: u16 = 2;
const UNREGISTER_MACHINE_LIFECYCLE_STORE_RECORD_VERSION: u16 = 3;
pub(crate) const MACHINE_LIFECYCLE_STORE_RECORD_VERSION: u16 = 4;
pub const MAX_INPUT_STATE_BATCH_CAS: usize = 256;
pub const MAX_PENDING_TERMINAL_OWNER_PAGE: usize = 256;
pub(crate) fn input_state_is_pending_terminal_owner(
state: &crate::input_state::InputState,
) -> bool {
let owns_pending_completion = state
.terminal_completion
.as_ref()
.is_some_and(|completion| {
completion.owner_input_id == state.input_id
&& matches!(
&completion.phase,
crate::input_state::InputTerminalCompletionPhase::Pending
)
});
let owns_unpublished_interaction =
state
.interaction_terminal_outbox
.as_ref()
.is_some_and(|outbox| {
outbox.candidate_owner_input_id == state.input_id
&& !matches!(
&outbox.phase,
crate::input_state::InteractionTerminalOutboxPhase::Published { .. }
)
});
owns_pending_completion || owns_unpublished_interaction
}
pub(crate) fn input_state_is_recovery_nonterminal(state: &StoredInputState) -> bool {
!matches!(
state.seed.phase,
crate::input_state::InputLifecycleState::Consumed
| crate::input_state::InputLifecycleState::Superseded
| crate::input_state::InputLifecycleState::Coalesced
| crate::input_state::InputLifecycleState::Abandoned
)
}
pub(crate) fn input_state_payload_is_retirable(state: &StoredInputState) -> bool {
let lifecycle_terminal = matches!(
state.seed.phase,
crate::input_state::InputLifecycleState::Consumed
| crate::input_state::InputLifecycleState::Superseded
| crate::input_state::InputLifecycleState::Coalesced
| crate::input_state::InputLifecycleState::Abandoned
) && state.seed.terminal_outcome.is_some();
let completion_closed = state
.state
.terminal_completion
.as_ref()
.is_none_or(|completion| {
matches!(
completion.phase,
crate::input_state::InputTerminalCompletionPhase::Finalized { .. }
)
});
let publication_closed =
state
.state
.interaction_terminal_outbox
.as_ref()
.is_none_or(|outbox| {
matches!(
outbox.phase,
crate::input_state::InteractionTerminalOutboxPhase::Published { .. }
)
});
let has_unmaterialized_directed_terminal =
state.state.persisted_input.as_ref().is_some_and(|input| {
crate::input::validated_directed_interaction_id(input)
.map(|interaction_id| {
interaction_id.is_some() && state.state.interaction_terminal_outbox.is_none()
})
.unwrap_or(true)
});
let has_compact_directed_attribution = state
.state
.interaction_terminal_outbox
.as_ref()
.is_none_or(|_| {
state
.state
.directed_run_started_attribution
.as_ref()
.is_some_and(|attribution| !attribution.content_digest().is_empty())
});
lifecycle_terminal
&& completion_closed
&& publication_closed
&& !has_unmaterialized_directed_terminal
&& has_compact_directed_attribution
}
#[cfg(test)]
mod terminal_payload_retirement_tests {
use super::*;
use crate::input::{Input, PromptInput};
use crate::input_state::{InputLifecycleState, InputTerminalOutcome, StoredInputState};
fn prompt_payload() -> Input {
Input::Prompt(PromptInput::new("large durable prompt", None))
}
fn with_terminal_seed(mut stored: StoredInputState) -> StoredInputState {
stored.seed.phase = InputLifecycleState::Consumed;
stored.seed.terminal_outcome = Some(InputTerminalOutcome::Consumed);
stored.seed.recovery_lane = None;
stored
}
#[test]
fn payload_retirement_requires_terminal_lifecycle_and_closed_obligations() {
let input_id = InputId::new();
let mut accepted = StoredInputState::new_accepted(input_id.clone());
accepted.state.persisted_input = Some(prompt_payload());
assert!(!input_state_payload_is_retirable(&accepted));
let mut staged = accepted.clone();
staged.seed.phase = InputLifecycleState::Staged;
assert!(!input_state_payload_is_retirable(&staged));
let terminal = with_terminal_seed(accepted);
assert!(input_state_payload_is_retirable(&terminal));
let (mut pending, _) = pending_terminal_owner_fixture(input_id.clone(), false);
pending.state.persisted_input = Some(prompt_payload());
pending = with_terminal_seed(pending);
assert!(!input_state_payload_is_retirable(&pending));
let (mut published, _) = pending_terminal_owner_fixture(input_id, true);
published.state.persisted_input = Some(prompt_payload());
published = with_terminal_seed(published);
assert!(input_state_payload_is_retirable(&published));
}
#[test]
fn retired_terminal_wire_image_omits_only_the_payload() {
let mut terminal = StoredInputState::new_accepted(InputId::new());
terminal.state.persisted_input = Some(prompt_payload());
terminal = with_terminal_seed(terminal);
let before_seed = terminal.seed.clone();
assert!(input_state_payload_is_retirable(&terminal));
terminal.state.persisted_input = None;
let encoded = serde_json::to_value(&terminal).unwrap();
let decoded: StoredInputState = serde_json::from_value(encoded.clone()).unwrap();
assert!(encoded.get("persisted_input").is_none());
assert_eq!(decoded.seed, before_seed);
assert_eq!(decoded.seed.phase, InputLifecycleState::Consumed);
assert_eq!(
decoded.seed.terminal_outcome,
Some(InputTerminalOutcome::Consumed)
);
}
}
pub(crate) fn validate_pending_terminal_owner_page(
after: Option<&InputId>,
limit: usize,
owner_input_ids: &[InputId],
) -> Result<(), RuntimeStoreError> {
if limit == 0 || limit > MAX_PENDING_TERMINAL_OWNER_PAGE {
return Err(RuntimeStoreError::InvalidInputStateBatchCas {
reason: format!(
"pending-terminal owner page limit {limit} is outside 1..={MAX_PENDING_TERMINAL_OWNER_PAGE}"
),
});
}
if owner_input_ids.len() > limit {
return Err(RuntimeStoreError::ReadFailed(format!(
"pending-terminal owner page returned {} ids for limit {limit}",
owner_input_ids.len()
)));
}
if owner_input_ids
.windows(2)
.any(|window| window[0].0 >= window[1].0)
{
return Err(RuntimeStoreError::ReadFailed(
"pending-terminal owner page is not strictly ordered".to_string(),
));
}
if let (Some(after), Some(first)) = (after, owner_input_ids.first())
&& first.0 <= after.0
{
return Err(RuntimeStoreError::ReadFailed(
"pending-terminal owner page did not advance its stable cursor".to_string(),
));
}
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum InputStateBatchCasOutcome {
Swapped,
Stale,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum InputStateBatchCasImplementationProfile {
Unsupported,
MultiWriter,
ExclusiveWriterFenced,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum FencedInputStateBatchCasOutcome {
Swapped,
Stale,
FenceConflict { reason: String },
FenceBackoff { reason: String },
}
#[derive(Debug)]
struct PreparedInputStateBatchCasRow {
input_id: InputId,
expected_json: Vec<u8>,
replacement: StoredInputState,
#[cfg_attr(not(feature = "sqlite-store"), allow(dead_code))]
replacement_json: Vec<u8>,
}
fn prepare_input_state_batch_cas(
expected: &[StoredInputState],
replacements: &[InputStatePersistenceRecord],
) -> Result<Vec<PreparedInputStateBatchCasRow>, RuntimeStoreError> {
if expected.len() != replacements.len() {
return Err(RuntimeStoreError::InvalidInputStateBatchCas {
reason: format!(
"expected row count {} does not match replacement row count {}",
expected.len(),
replacements.len()
),
});
}
if expected.len() > MAX_INPUT_STATE_BATCH_CAS {
return Err(RuntimeStoreError::InvalidInputStateBatchCas {
reason: format!(
"batch contains {} rows, exceeding the maximum of {MAX_INPUT_STATE_BATCH_CAS}",
expected.len()
),
});
}
let mut expected_ids = HashSet::with_capacity(expected.len());
for row in expected {
if !expected_ids.insert(row.state.input_id.clone()) {
return Err(RuntimeStoreError::InvalidInputStateBatchCas {
reason: format!("expected batch repeats input {}", row.state.input_id),
});
}
}
let mut replacement_by_id = HashMap::with_capacity(replacements.len());
for record in replacements {
let replacement = record.clone_stored();
let input_id = replacement.state.input_id.clone();
let replacement_json = serde_json::to_vec(&replacement)
.map_err(|error| RuntimeStoreError::WriteFailed(error.to_string()))?;
if replacement_by_id
.insert(input_id.clone(), (replacement, replacement_json))
.is_some()
{
return Err(RuntimeStoreError::InvalidInputStateBatchCas {
reason: format!("replacement batch repeats input {input_id}"),
});
}
}
let mut prepared = Vec::with_capacity(expected.len());
for expected_row in expected {
let input_id = expected_row.state.input_id.clone();
let Some((replacement, replacement_json)) = replacement_by_id.remove(&input_id) else {
return Err(RuntimeStoreError::InvalidInputStateBatchCas {
reason: format!("replacement batch does not contain expected input {input_id}"),
});
};
let expected_json = serde_json::to_vec(expected_row)
.map_err(|error| RuntimeStoreError::WriteFailed(error.to_string()))?;
prepared.push(PreparedInputStateBatchCasRow {
input_id,
expected_json,
replacement,
replacement_json,
});
}
if let Some(extra) = replacement_by_id.keys().next() {
return Err(RuntimeStoreError::InvalidInputStateBatchCas {
reason: format!("replacement batch contains unexpected input {extra}"),
});
}
Ok(prepared)
}
pub(crate) fn validate_input_state_batch_read_ids(
input_ids: &[InputId],
) -> Result<(), RuntimeStoreError> {
if input_ids.len() > MAX_INPUT_STATE_BATCH_CAS {
return Err(RuntimeStoreError::InvalidInputStateBatchCas {
reason: format!(
"batch read contains {} rows, exceeding the maximum of {MAX_INPUT_STATE_BATCH_CAS}",
input_ids.len()
),
});
}
let mut unique = HashSet::with_capacity(input_ids.len());
for input_id in input_ids {
if !unique.insert(input_id.clone()) {
return Err(RuntimeStoreError::InvalidInputStateBatchCas {
reason: format!("batch read repeats input {input_id}"),
});
}
}
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum RuntimeSessionPersistenceProfile {
WholeBlobV1,
HeadCanonicalV1,
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct RuntimeSessionCatalogEntry {
session_id: meerkat_core::types::SessionId,
persistence_profile: RuntimeSessionPersistenceProfile,
created_at: meerkat_core::time_compat::SystemTime,
updated_at: meerkat_core::time_compat::SystemTime,
message_count: usize,
total_tokens: u64,
labels: BTreeMap<String, String>,
lifecycle_terminal: Option<meerkat_core::SessionLifecycleTerminal>,
runtime_state: Option<RuntimeState>,
}
impl RuntimeSessionCatalogEntry {
const SESSION_LABELS_KEY: &'static str = "session_labels";
pub fn from_session(
session: &meerkat_core::Session,
persistence_profile: RuntimeSessionPersistenceProfile,
runtime_state: Option<RuntimeState>,
) -> Result<Self, RuntimeStoreError> {
let labels = session
.metadata()
.get(Self::SESSION_LABELS_KEY)
.map(|value| {
serde_json::from_value::<BTreeMap<String, String>>(value.clone()).map_err(|error| {
RuntimeStoreError::WriteFailed(format!(
"session {} has malformed catalog labels: {error}",
session.id()
))
})
})
.transpose()?
.unwrap_or_default();
let lifecycle_terminal = session.try_lifecycle_terminal().map_err(|error| {
RuntimeStoreError::WriteFailed(format!(
"session {} has malformed lifecycle-terminal metadata: {error}",
session.id()
))
})?;
Ok(Self {
session_id: session.id().clone(),
persistence_profile,
created_at: session.created_at(),
updated_at: session.updated_at(),
message_count: session.messages().len(),
total_tokens: session.total_tokens(),
labels,
lifecycle_terminal,
runtime_state,
})
}
pub fn from_head(
head: &meerkat_core::session_store::SessionHead,
persistence_profile: RuntimeSessionPersistenceProfile,
runtime_state: Option<RuntimeState>,
) -> Result<Self, RuntimeStoreError> {
let metadata = head.materialized_metadata().map_err(|error| {
RuntimeStoreError::WriteFailed(format!(
"session {} head has no exact catalog metadata projection: {error}",
head.id
))
})?;
let labels = metadata
.get(Self::SESSION_LABELS_KEY)
.map(|value| {
serde_json::from_value::<BTreeMap<String, String>>(value.clone()).map_err(|error| {
RuntimeStoreError::WriteFailed(format!(
"session {} head has malformed catalog labels: {error}",
head.id
))
})
})
.transpose()?
.unwrap_or_default();
let lifecycle_terminal =
meerkat_core::try_lifecycle_terminal_from_map(&metadata).map_err(|error| {
RuntimeStoreError::WriteFailed(format!(
"session {} head has malformed lifecycle-terminal metadata: {error}",
head.id
))
})?;
Self::from_head_facts(
head,
labels,
lifecycle_terminal,
persistence_profile,
runtime_state,
)
}
pub(crate) fn from_head_facts(
head: &meerkat_core::session_store::SessionHead,
labels: BTreeMap<String, String>,
lifecycle_terminal: Option<meerkat_core::SessionLifecycleTerminal>,
persistence_profile: RuntimeSessionPersistenceProfile,
runtime_state: Option<RuntimeState>,
) -> Result<Self, RuntimeStoreError> {
Ok(Self {
session_id: head.id.clone(),
persistence_profile,
created_at: head.created_at,
updated_at: head.updated_at,
message_count: usize::try_from(head.message_count).map_err(|_| {
RuntimeStoreError::WriteFailed(format!(
"session {} head message count exceeds host range",
head.id
))
})?,
total_tokens: head.usage.total_tokens(),
labels,
lifecycle_terminal,
runtime_state,
})
}
#[must_use]
pub fn session_id(&self) -> &meerkat_core::types::SessionId {
&self.session_id
}
#[must_use]
pub const fn persistence_profile(&self) -> RuntimeSessionPersistenceProfile {
self.persistence_profile
}
#[must_use]
pub const fn created_at(&self) -> meerkat_core::time_compat::SystemTime {
self.created_at
}
#[must_use]
pub const fn updated_at(&self) -> meerkat_core::time_compat::SystemTime {
self.updated_at
}
#[must_use]
pub const fn message_count(&self) -> usize {
self.message_count
}
#[must_use]
pub const fn total_tokens(&self) -> u64 {
self.total_tokens
}
#[must_use]
pub fn labels(&self) -> &BTreeMap<String, String> {
&self.labels
}
#[must_use]
pub const fn lifecycle_terminal(&self) -> Option<meerkat_core::SessionLifecycleTerminal> {
self.lifecycle_terminal
}
#[must_use]
pub const fn runtime_state(&self) -> Option<RuntimeState> {
self.runtime_state
}
pub(crate) fn set_runtime_state(&mut self, runtime_state: Option<RuntimeState>) {
self.runtime_state = runtime_state;
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RuntimeSessionAuthorityReadCost {
Bounded,
Unsupported,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WholeBlobStoreAuthority {
authority_version: u16,
session_id: meerkat_core::types::SessionId,
store_revision: u64,
blob_sha256: String,
}
fn is_canonical_row_sha256_token(token: &str) -> bool {
let Some(hex) = token.strip_prefix("row-sha256:") else {
return false;
};
hex.len() == 64
&& hex
.bytes()
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
}
impl WholeBlobStoreAuthority {
pub const VERSION: u16 = 1;
pub fn from_store_record(
authority_version: u16,
session_id: meerkat_core::types::SessionId,
store_revision: u64,
blob_sha256: String,
) -> Result<Self, RuntimeStoreError> {
if authority_version != Self::VERSION
|| store_revision == 0
|| !is_canonical_row_sha256_token(&blob_sha256)
{
return Err(RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: session_id.to_string(),
detail: "WholeBlob store authority requires the current version, nonzero \
revision, and canonical physical row digest"
.to_string(),
});
}
Ok(Self {
authority_version,
session_id,
store_revision,
blob_sha256,
})
}
pub(crate) fn issued(
session_id: meerkat_core::types::SessionId,
store_revision: u64,
blob_sha256: String,
) -> Result<Self, RuntimeStoreError> {
Self::from_store_record(Self::VERSION, session_id, store_revision, blob_sha256)
}
#[must_use]
pub const fn authority_version(&self) -> u16 {
self.authority_version
}
#[must_use]
pub fn session_id(&self) -> &meerkat_core::types::SessionId {
&self.session_id
}
#[must_use]
pub const fn store_revision(&self) -> u64 {
self.store_revision
}
#[must_use]
pub fn blob_sha256(&self) -> &str {
&self.blob_sha256
}
}
#[derive(Debug, Clone)]
pub struct CommittedWholeBlobSnapshot {
session: Arc<meerkat_core::Session>,
bytes: Arc<Vec<u8>>,
authority: WholeBlobStoreAuthority,
}
#[derive(Debug, Clone)]
pub struct PreparedWholeBlobSnapshotCas {
expected_authority: WholeBlobStoreAuthority,
candidate_session: Arc<meerkat_core::Session>,
candidate_bytes: Arc<Vec<u8>>,
candidate_blob_sha256: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum WholeBlobSnapshotCasOutcome {
Committed(WholeBlobStoreAuthority),
Conflict,
}
impl PreparedWholeBlobSnapshotCas {
pub fn prepare(
expected_authority: WholeBlobStoreAuthority,
candidate: BoundSessionCommit,
) -> Result<Self, RuntimeStoreError> {
let candidate_session = candidate.into_session_arc().ok_or_else(|| {
RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: expected_authority.session_id().to_string(),
detail: "WholeBlob snapshot CAS requires a sealed typed Session".to_string(),
}
})?;
if candidate_session.id() != expected_authority.session_id() {
return Err(RuntimeStoreError::SessionKeyMismatch {
expected: expected_authority.session_id().clone(),
actual: candidate_session.id().clone(),
});
}
let (candidate_bytes, candidate_blob_sha256) =
encode_whole_blob_session(candidate_session.as_ref())?;
Ok(Self {
expected_authority,
candidate_session,
candidate_bytes,
candidate_blob_sha256,
})
}
#[must_use]
pub fn expected_authority(&self) -> &WholeBlobStoreAuthority {
&self.expected_authority
}
#[must_use]
pub fn candidate_blob_sha256(&self) -> &str {
&self.candidate_blob_sha256
}
#[must_use]
pub fn accepts_committed_authority(&self, observed: &WholeBlobStoreAuthority) -> bool {
if observed.session_id() != self.expected_authority.session_id()
|| observed.blob_sha256() != self.candidate_blob_sha256
{
return false;
}
(observed.store_revision() == self.expected_authority.store_revision()
&& self.candidate_blob_sha256 == self.expected_authority.blob_sha256())
|| self
.expected_authority
.store_revision()
.checked_add(1)
.is_some_and(|successor| observed.store_revision() == successor)
}
pub(crate) fn into_parts(
self,
) -> (
WholeBlobStoreAuthority,
Arc<meerkat_core::Session>,
Arc<Vec<u8>>,
String,
) {
(
self.expected_authority,
self.candidate_session,
self.candidate_bytes,
self.candidate_blob_sha256,
)
}
}
struct WholeBlobSessionWriter {
bytes: Vec<u8>,
hasher: Sha256,
}
impl std::io::Write for WholeBlobSessionWriter {
fn write(&mut self, buffer: &[u8]) -> std::io::Result<usize> {
self.bytes.extend_from_slice(buffer);
self.hasher.update(buffer);
Ok(buffer.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
fn encode_whole_blob_session(
session: &meerkat_core::Session,
) -> Result<(Arc<Vec<u8>>, String), RuntimeStoreError> {
let mut writer = WholeBlobSessionWriter {
bytes: Vec::new(),
hasher: Sha256::new(),
};
serde_json::to_writer(&mut writer, session)
.map_err(|error| RuntimeStoreError::WriteFailed(error.to_string()))?;
let blob_sha256 = format!("row-sha256:{:x}", writer.hasher.finalize());
Ok((Arc::new(writer.bytes), blob_sha256))
}
#[derive(Debug, Clone)]
pub struct PreparedWholeBlobProvisionalTail {
authority: WholeBlobProvisionalTailAuthority,
candidate_artifact: meerkat_core::SerializedSessionArtifact,
conversation_digest: String,
message_count: u64,
catalog_entry: RuntimeSessionCatalogEntry,
compaction_projection_intents: Vec<meerkat_core::CompactionProjectionIntent>,
#[cfg(test)]
whole_blob_encode_count: Arc<std::sync::atomic::AtomicUsize>,
}
impl PreparedWholeBlobProvisionalTail {
pub fn prepare(
base: WholeBlobStoreAuthority,
run_id: RunId,
candidate_sequence: u64,
candidate: &BoundSessionCommit,
) -> Result<Self, RuntimeStoreError> {
let candidate_session = candidate.session_arc_cloned().ok_or_else(|| {
RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: base.session_id().to_string(),
detail: "WholeBlob provisional candidate requires a sealed typed Session"
.to_string(),
}
})?;
let artifact = candidate
.whole_blob_artifact()
.map_err(|error| {
RuntimeStoreError::WriteFailed(format!(
"failed to materialize WholeBlob provisional candidate: {error}"
))
})?
.clone();
Self::prepare_from_artifact(
base,
run_id,
candidate_sequence,
candidate_session.as_ref(),
artifact,
#[cfg(test)]
Arc::new(std::sync::atomic::AtomicUsize::new(0)),
)
}
pub fn prepare_from_session(
base: WholeBlobStoreAuthority,
run_id: RunId,
candidate_sequence: u64,
candidate: &meerkat_core::Session,
) -> Result<Self, RuntimeStoreError> {
#[cfg(test)]
let whole_blob_encode_count = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let artifact = candidate.to_persisted_artifact().map_err(|error| {
RuntimeStoreError::WriteFailed(format!(
"failed to materialize WholeBlob provisional candidate: {error}"
))
})?;
#[cfg(test)]
whole_blob_encode_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
Self::prepare_from_artifact(
base,
run_id,
candidate_sequence,
candidate,
artifact,
#[cfg(test)]
whole_blob_encode_count,
)
}
fn prepare_from_artifact(
base: WholeBlobStoreAuthority,
run_id: RunId,
candidate_sequence: u64,
candidate_session: &meerkat_core::Session,
candidate_artifact: meerkat_core::SerializedSessionArtifact,
#[cfg(test)] whole_blob_encode_count: Arc<std::sync::atomic::AtomicUsize>,
) -> Result<Self, RuntimeStoreError> {
if candidate_session.id() != base.session_id() {
return Err(RuntimeStoreError::SessionKeyMismatch {
expected: base.session_id().clone(),
actual: candidate_session.id().clone(),
});
}
let authority = WholeBlobProvisionalTailAuthority::issued(
base.session_id().clone(),
base.store_revision(),
base.blob_sha256().to_string(),
run_id,
candidate_artifact.row_sha256_token().to_string(),
candidate_sequence,
)
.map_err(
|error| RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: base.session_id().to_string(),
detail: error.to_string(),
},
)?;
let catalog_entry = RuntimeSessionCatalogEntry::from_session(
candidate_session,
RuntimeSessionPersistenceProfile::WholeBlobV1,
None,
)?;
let conversation_digest =
candidate_session
.transcript_content_digest()
.map_err(
|error| RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: base.session_id().to_string(),
detail: format!(
"failed to derive WholeBlob provisional conversation digest: {error}"
),
},
)?;
let message_count = u64::try_from(candidate_session.messages().len()).map_err(|_| {
RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: base.session_id().to_string(),
detail: "WholeBlob provisional message count exceeds the durable range".to_string(),
}
})?;
let compaction_projection_intents =
validated_compaction_projection_intents(candidate_session)?;
Ok(Self {
authority,
candidate_artifact,
conversation_digest,
message_count,
catalog_entry,
compaction_projection_intents,
#[cfg(test)]
whole_blob_encode_count,
})
}
#[must_use]
pub fn authority(&self) -> &WholeBlobProvisionalTailAuthority {
&self.authority
}
#[must_use]
pub fn conversation_digest(&self) -> &str {
&self.conversation_digest
}
#[must_use]
pub const fn message_count(&self) -> u64 {
self.message_count
}
#[cfg(test)]
fn whole_blob_encode_count(&self) -> usize {
self.whole_blob_encode_count
.load(std::sync::atomic::Ordering::Relaxed)
}
pub(crate) fn into_parts(
self,
) -> (
WholeBlobProvisionalTailAuthority,
meerkat_core::SerializedSessionArtifact,
String,
u64,
RuntimeSessionCatalogEntry,
Vec<meerkat_core::CompactionProjectionIntent>,
) {
(
self.authority,
self.candidate_artifact,
self.conversation_digest,
self.message_count,
self.catalog_entry,
self.compaction_projection_intents,
)
}
}
#[derive(Debug, Clone)]
pub struct CommittedWholeBlobProvisionalTail {
authority: WholeBlobProvisionalTailAuthority,
candidate_bytes: Arc<Vec<u8>>,
}
#[derive(Debug, Clone)]
pub struct PreparedWholeBlobProvisionalPromotion {
authority: WholeBlobProvisionalTailAuthority,
conversation_digest: String,
message_count: u64,
}
impl PreparedWholeBlobProvisionalPromotion {
pub fn prepare(
checkpoint: meerkat_core::RunCheckpointReceipt,
run_id: &RunId,
) -> Result<Self, RuntimeStoreError> {
let authority = checkpoint.whole_blob().cloned().ok_or_else(|| {
RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: checkpoint.session_id().to_string(),
detail: "HeadCanonical checkpoint cannot authorize WholeBlob promotion".to_string(),
}
})?;
if authority.run_id() != run_id {
return Err(RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: authority.session_id().to_string(),
detail: "WholeBlob promotion run differs from store-issued candidate run"
.to_string(),
});
}
Ok(Self {
authority,
conversation_digest: checkpoint.conversation_digest().to_string(),
message_count: checkpoint.message_count(),
})
}
#[must_use]
pub fn authority(&self) -> &WholeBlobProvisionalTailAuthority {
&self.authority
}
#[must_use]
pub(crate) fn into_parts(self) -> (WholeBlobProvisionalTailAuthority, String, u64) {
(self.authority, self.conversation_digest, self.message_count)
}
}
#[derive(Debug, Clone)]
pub(crate) struct PreparedWholeBlobRecoveryPromotion {
authority: WholeBlobProvisionalTailAuthority,
repaired_snapshot: Option<PreparedWholeBlobSnapshot>,
}
impl PreparedWholeBlobRecoveryPromotion {
fn prepare(
repaired_document: Option<&BoundSessionCommit>,
evidence: &PreparedRecoveryEvidence,
) -> Result<Self, RuntimeStoreError> {
let (
base_store_revision,
base_blob_sha256,
candidate_blob_sha256,
candidate_sequence,
recovered_blob_sha256,
) = evidence.whole_blob_authority_transition().ok_or_else(|| {
RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: evidence.session_id().to_string(),
detail: "HeadCanonical recovery evidence cannot authorize WholeBlob promotion"
.to_string(),
}
})?;
let authority = WholeBlobProvisionalTailAuthority::issued(
evidence.session_id().clone(),
base_store_revision,
base_blob_sha256.to_string(),
evidence.candidate_run_id().clone(),
candidate_blob_sha256.to_string(),
candidate_sequence,
)
.map_err(
|error| RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: evidence.session_id().to_string(),
detail: error.to_string(),
},
)?;
let repaired_snapshot = if recovered_blob_sha256 == candidate_blob_sha256 {
if repaired_document.is_some() {
return Err(RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: evidence.session_id().to_string(),
detail: "completed WholeBlob recovery must not carry a materialized body"
.to_string(),
});
}
None
} else {
let document = repaired_document.ok_or_else(|| {
RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: evidence.session_id().to_string(),
detail: "WholeBlob repair has no sealed successor artifact".to_string(),
}
})?;
let prepared = prepared_whole_blob_snapshot(document)?;
if prepared.session().id() != evidence.session_id()
|| prepared.blob_sha256() != recovered_blob_sha256
{
return Err(RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: evidence.session_id().to_string(),
detail: "WholeBlob repaired artifact differs from sealed recovery authority"
.to_string(),
});
}
Some(prepared)
};
Ok(Self {
authority,
repaired_snapshot,
})
}
pub(crate) fn into_parts(
self,
) -> (
WholeBlobProvisionalTailAuthority,
Option<PreparedWholeBlobSnapshot>,
) {
(self.authority, self.repaired_snapshot)
}
}
#[derive(Debug, Clone)]
pub struct PreparedHeadCanonicalProvisionalPromotion {
checkpoint: meerkat_core::RunCheckpointReceipt,
authority: HeadCanonicalProvisionalTailAuthority,
}
impl PreparedHeadCanonicalProvisionalPromotion {
pub fn prepare(
checkpoint: meerkat_core::RunCheckpointReceipt,
run_id: &RunId,
) -> Result<Self, RuntimeStoreError> {
let authority = checkpoint.head_canonical().cloned().ok_or_else(|| {
RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: checkpoint.session_id().to_string(),
detail: "HeadCanonical promotion received a WholeBlob checkpoint".to_string(),
}
})?;
if authority.run_id() != run_id {
return Err(RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: authority.session_id().to_string(),
detail: "HeadCanonical promotion run differs from store-issued physical tail"
.to_string(),
});
}
Ok(Self {
checkpoint,
authority,
})
}
#[must_use]
pub fn authority(&self) -> &HeadCanonicalProvisionalTailAuthority {
&self.authority
}
#[must_use]
pub fn checkpoint(&self) -> &meerkat_core::RunCheckpointReceipt {
&self.checkpoint
}
#[must_use]
pub(crate) fn into_parts(
self,
) -> (
meerkat_core::RunCheckpointReceipt,
HeadCanonicalProvisionalTailAuthority,
) {
(self.checkpoint, self.authority)
}
}
impl CommittedWholeBlobProvisionalTail {
pub(crate) fn new(
authority: WholeBlobProvisionalTailAuthority,
candidate_bytes: Arc<Vec<u8>>,
) -> Self {
Self {
authority,
candidate_bytes,
}
}
#[must_use]
pub fn authority(&self) -> &WholeBlobProvisionalTailAuthority {
&self.authority
}
#[must_use]
pub fn candidate_bytes(&self) -> &[u8] {
self.candidate_bytes.as_ref()
}
#[must_use]
pub fn candidate_bytes_arc(&self) -> Arc<Vec<u8>> {
Arc::clone(&self.candidate_bytes)
}
}
impl CommittedWholeBlobSnapshot {
pub(crate) fn new(
bytes: Arc<Vec<u8>>,
authority: WholeBlobStoreAuthority,
) -> Result<Self, RuntimeStoreError> {
let decoded =
meerkat_core::Session::decode_whole_blob_document(bytes.as_ref()).map_err(|error| {
RuntimeStoreError::ReadFailed(format!(
"WholeBlob body is not a valid current Session: {error}"
))
})?;
if decoded.row_sha256_token() != authority.blob_sha256() {
return Err(RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: authority.session_id().to_string(),
detail: "WholeBlob body digest differs from store authority".to_string(),
});
}
let session = Arc::new(decoded.into_session());
if session.id() != authority.session_id() {
return Err(RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: authority.session_id().to_string(),
detail: "WholeBlob body session differs from store authority".to_string(),
});
}
Ok(Self {
session,
bytes,
authority,
})
}
#[must_use]
pub fn session(&self) -> &meerkat_core::Session {
self.session.as_ref()
}
#[must_use]
pub fn session_arc(&self) -> Arc<meerkat_core::Session> {
Arc::clone(&self.session)
}
#[must_use]
pub fn bytes(&self) -> &[u8] {
self.bytes.as_ref()
}
#[must_use]
pub fn bytes_arc(&self) -> Arc<Vec<u8>> {
Arc::clone(&self.bytes)
}
#[must_use]
pub fn authority(&self) -> &WholeBlobStoreAuthority {
&self.authority
}
#[must_use]
pub fn into_parts(
self,
) -> (
Arc<meerkat_core::Session>,
Arc<Vec<u8>>,
WholeBlobStoreAuthority,
) {
(self.session, self.bytes, self.authority)
}
}
impl std::fmt::Display for RuntimeSessionPersistenceProfile {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::WholeBlobV1 => f.write_str("whole_blob_v1"),
Self::HeadCanonicalV1 => f.write_str("head_canonical_v1"),
}
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct HeadCanonicalStoreAuthority {
authority_version: u16,
session_id: meerkat_core::types::SessionId,
store_revision: u64,
boundary_head: meerkat_core::session_store::SessionHead,
committed_head_token: String,
}
impl HeadCanonicalStoreAuthority {
pub const VERSION: u16 = 1;
pub fn from_store_record(
authority_version: u16,
session_id: meerkat_core::types::SessionId,
store_revision: u64,
boundary_head: meerkat_core::session_store::SessionHead,
committed_head_token: String,
) -> Result<Self, RuntimeStoreError> {
let conflict = |detail: String| RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: session_id.to_string(),
detail,
};
if authority_version != Self::VERSION
|| store_revision == 0
|| committed_head_token.is_empty()
{
return Err(conflict(
"HeadCanonical authority requires the current version, nonzero store revision, \
and head token"
.to_string(),
));
}
if boundary_head.id != session_id {
return Err(conflict(format!(
"HeadCanonical boundary belongs to {}, not {session_id}",
boundary_head.id
)));
}
let row_prefix = boundary_head.message_row_prefix.as_ref().ok_or_else(|| {
conflict("HeadCanonical boundary has no exact message-row prefix".to_string())
})?;
if row_prefix.row_count() != boundary_head.message_count {
return Err(conflict(
"HeadCanonical boundary message count and row prefix differ".to_string(),
));
}
if boundary_head.rewrite_prefix.occurrence_count() != boundary_head.rewrite_count {
return Err(conflict(
"HeadCanonical boundary rewrite count and prefix differ".to_string(),
));
}
let derived = meerkat_core::session_head_cas_token(&boundary_head)
.map_err(|error| conflict(format!("HeadCanonical head token is invalid: {error}")))?;
if derived != committed_head_token {
return Err(conflict(
"store-issued HeadCanonical token differs from the exact boundary head".to_string(),
));
}
Ok(Self {
authority_version,
session_id,
store_revision,
boundary_head,
committed_head_token,
})
}
pub(crate) fn issued(
session_id: meerkat_core::types::SessionId,
store_revision: u64,
boundary_head: meerkat_core::session_store::SessionHead,
committed_head_token: String,
) -> Result<Self, RuntimeStoreError> {
Self::from_store_record(
Self::VERSION,
session_id,
store_revision,
boundary_head,
committed_head_token,
)
}
#[must_use]
pub const fn authority_version(&self) -> u16 {
self.authority_version
}
#[must_use]
pub fn session_id(&self) -> &meerkat_core::types::SessionId {
&self.session_id
}
#[must_use]
pub const fn store_revision(&self) -> u64 {
self.store_revision
}
#[must_use]
pub fn boundary_head(&self) -> &meerkat_core::session_store::SessionHead {
&self.boundary_head
}
#[must_use]
pub fn committed_head_token(&self) -> &str {
&self.committed_head_token
}
}
#[derive(Debug, Clone)]
pub struct PreparedHeadCanonicalProvisionalTail {
committed: HeadCanonicalStoreAuthority,
run_id: RunId,
successor_head: meerkat_core::session_store::SessionHead,
successor_head_token: String,
candidate_message_count: usize,
candidate_conversation_digest: String,
catalog_entry: RuntimeSessionCatalogEntry,
compaction_projection_intents: Vec<meerkat_core::CompactionProjectionIntent>,
}
impl PreparedHeadCanonicalProvisionalTail {
pub fn prepare(
committed: HeadCanonicalStoreAuthority,
run_id: RunId,
successor_head: &meerkat_core::session_store::SessionHead,
successor_head_token: &str,
candidate_session: &meerkat_core::Session,
) -> Result<Self, RuntimeStoreError> {
let session_id = committed.session_id().clone();
let conflict = |detail: String| RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: session_id.to_string(),
detail,
};
if successor_head.id != session_id || successor_head_token.is_empty() {
return Err(conflict(
"HeadCanonical provisional intent names the wrong session or an empty successor"
.to_string(),
));
}
let derived = meerkat_core::session_head_cas_token(successor_head).map_err(|error| {
conflict(format!(
"HeadCanonical provisional successor is invalid: {error}"
))
})?;
if derived != successor_head_token
|| successor_head_token == committed.committed_head_token()
{
return Err(conflict(
"HeadCanonical provisional successor token is not the exact distinct target head"
.to_string(),
));
}
if candidate_session.id() != &session_id
|| candidate_session.messages().len() as u64 != successor_head.message_count
|| candidate_session.version() != successor_head.version
|| candidate_session.created_at() != successor_head.created_at
|| candidate_session.updated_at() != successor_head.updated_at
|| candidate_session.total_usage() != successor_head.usage
|| !successor_head
.matches_session_metadata(candidate_session)
.map_err(|error| {
conflict(format!(
"HeadCanonical provisional candidate metadata is invalid: {error}"
))
})?
{
return Err(conflict(
"HeadCanonical provisional successor does not describe the exact candidate Session"
.to_string(),
));
}
let candidate_conversation_digest =
candidate_session
.transcript_content_digest()
.map_err(|error| {
conflict(format!(
"HeadCanonical provisional candidate transcript is invalid: {error}"
))
})?;
if candidate_conversation_digest != successor_head.head_revision {
return Err(conflict(
"HeadCanonical provisional candidate digest differs from its successor head"
.to_string(),
));
}
let catalog_entry = RuntimeSessionCatalogEntry::from_session(
candidate_session,
RuntimeSessionPersistenceProfile::HeadCanonicalV1,
None,
)?;
let compaction_projection_intents =
validated_compaction_projection_intents(candidate_session)?;
Ok(Self {
committed,
run_id,
successor_head: successor_head.clone(),
successor_head_token: successor_head_token.to_string(),
candidate_message_count: candidate_session.messages().len(),
candidate_conversation_digest,
catalog_entry,
compaction_projection_intents,
})
}
#[must_use]
pub(crate) fn committed(&self) -> &HeadCanonicalStoreAuthority {
&self.committed
}
#[must_use]
pub(crate) fn run_id(&self) -> &RunId {
&self.run_id
}
#[must_use]
pub(crate) fn successor_head(&self) -> &meerkat_core::session_store::SessionHead {
&self.successor_head
}
#[must_use]
pub(crate) fn successor_head_token(&self) -> &str {
&self.successor_head_token
}
#[must_use]
pub(crate) const fn candidate_message_count(&self) -> usize {
self.candidate_message_count
}
#[must_use]
pub(crate) fn candidate_conversation_digest(&self) -> &str {
&self.candidate_conversation_digest
}
#[must_use]
pub(crate) fn catalog_entry(&self) -> &RuntimeSessionCatalogEntry {
&self.catalog_entry
}
#[must_use]
pub(crate) fn compaction_projection_intents(
&self,
) -> &[meerkat_core::CompactionProjectionIntent] {
&self.compaction_projection_intents
}
}
#[derive(Debug, Clone, PartialEq)]
pub enum RuntimeSessionAuthority {
WholeBlob(WholeBlobStoreAuthority),
HeadCanonical(HeadCanonicalStoreAuthority),
}
impl RuntimeSessionAuthority {
#[must_use]
pub const fn profile(&self) -> RuntimeSessionPersistenceProfile {
match self {
Self::WholeBlob(_) => RuntimeSessionPersistenceProfile::WholeBlobV1,
Self::HeadCanonical(_) => RuntimeSessionPersistenceProfile::HeadCanonicalV1,
}
}
#[must_use]
pub fn session_id(&self) -> &meerkat_core::types::SessionId {
match self {
Self::WholeBlob(authority) => authority.session_id(),
Self::HeadCanonical(authority) => authority.session_id(),
}
}
#[must_use]
pub fn whole_blob(&self) -> Option<&WholeBlobStoreAuthority> {
match self {
Self::WholeBlob(authority) => Some(authority),
Self::HeadCanonical(_) => None,
}
}
#[must_use]
pub fn head_canonical(&self) -> Option<&HeadCanonicalStoreAuthority> {
match self {
Self::WholeBlob(_) => None,
Self::HeadCanonical(authority) => Some(authority),
}
}
}
#[derive(Debug, Clone)]
pub struct PreparedDurableTailRecoverySource {
runtime_authority: RuntimeSessionAuthority,
provisional_authority: Option<HeadCanonicalProvisionalTailAuthority>,
provisional_target_applied: bool,
committed_session: Arc<meerkat_core::Session>,
physical_head: meerkat_core::session_store::SessionHead,
physical_head_cas_token: String,
physical_session: Arc<meerkat_core::Session>,
}
impl PreparedDurableTailRecoverySource {
pub(crate) fn new(
runtime_authority: RuntimeSessionAuthority,
provisional_authority: Option<HeadCanonicalProvisionalTailAuthority>,
committed_materialization: meerkat_core::VerifiedSessionHeadMaterialization,
physical_materialization: meerkat_core::VerifiedSessionHeadMaterialization,
) -> Result<Self, RuntimeStoreError> {
let committed_session = Arc::clone(committed_materialization.session());
let physical_head = physical_materialization.head().clone();
let physical_session = Arc::clone(physical_materialization.session());
let runtime_id = runtime_authority.session_id().to_string();
let conflict = |detail: String| RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: runtime_id.clone(),
detail,
};
if runtime_authority.profile() != RuntimeSessionPersistenceProfile::HeadCanonicalV1 {
return Err(conflict(
"durable-tail source requires head-canonical session ownership".to_string(),
));
}
let committed_authority = runtime_authority
.head_canonical()
.ok_or_else(|| conflict("runtime authority is not HeadCanonical".to_string()))?;
let boundary_head = committed_authority.boundary_head();
if committed_materialization.head() != boundary_head {
return Err(conflict(
"verified committed materialization belongs to a different retained boundary head"
.to_string(),
));
}
let boundary_row_prefix = boundary_head.message_row_prefix.as_ref().ok_or_else(|| {
conflict("runtime boundary has no exact message-row prefix authority".to_string())
})?;
if physical_materialization
.exact_row_prefix_at(boundary_head.message_count)
.as_ref()
!= Some(boundary_row_prefix)
{
return Err(conflict(
"physical recovery materialization does not retain the runtime boundary's exact row prefix"
.to_string(),
));
}
if committed_session.id() != runtime_authority.session_id()
|| physical_session.id() != runtime_authority.session_id()
|| &physical_head.id != runtime_authority.session_id()
{
return Err(conflict(
"durable-tail source identities do not all match runtime authority".to_string(),
));
}
let committed_revision = committed_session
.transcript_content_digest()
.map_err(|error| conflict(format!("committed transcript is invalid: {error}")))?;
let committed_metadata_matches = boundary_head
.matches_session_metadata(&committed_session)
.map_err(|error| {
conflict(format!(
"committed recovery metadata identity is invalid: {error}"
))
})?;
if committed_session.messages().len() as u64 != boundary_head.message_count
|| committed_revision != boundary_head.head_revision
|| committed_session.version() != boundary_head.version
|| committed_session.created_at() != boundary_head.created_at
|| committed_session.updated_at() != boundary_head.updated_at
|| committed_session.total_usage() != boundary_head.usage
|| !committed_metadata_matches
{
return Err(conflict(format!(
"committed recovery materialization differs from the exact retained boundary envelope \
(message_count={}, revision={}, version={}, created_at={}, updated_at={}, usage={}, metadata={})",
committed_session.messages().len() as u64 == boundary_head.message_count,
committed_revision == boundary_head.head_revision,
committed_session.version() == boundary_head.version,
committed_session.created_at() == boundary_head.created_at,
committed_session.updated_at() == boundary_head.updated_at,
committed_session.total_usage() == boundary_head.usage,
committed_metadata_matches,
)));
}
let physical_revision = physical_session
.transcript_content_digest()
.map_err(|error| conflict(format!("physical transcript is invalid: {error}")))?;
let physical_metadata_matches = physical_head
.matches_session_metadata(&physical_session)
.map_err(|error| {
conflict(format!(
"physical recovery metadata identity is invalid: {error}"
))
})?;
if physical_session.messages().len() as u64 != physical_head.message_count
|| physical_revision != physical_head.head_revision
|| physical_session.version() != physical_head.version
|| physical_session.created_at() != physical_head.created_at
|| physical_session.updated_at() != physical_head.updated_at
|| physical_session.total_usage() != physical_head.usage
|| !physical_metadata_matches
{
return Err(conflict(
"physical recovery materialization differs from the exact current canonical envelope"
.to_string(),
));
}
let Some(physical_row_prefix) = physical_head.message_row_prefix.as_ref() else {
return Err(conflict(
"physical recovery head has no exact message-row prefix authority".to_string(),
));
};
if physical_row_prefix.row_count() != physical_head.message_count {
return Err(conflict(
"physical recovery head message count and exact row prefix differ".to_string(),
));
}
let physical_head_cas_token = meerkat_core::session_head_cas_token(&physical_head)
.map_err(|error| {
conflict(format!("physical recovery head token is invalid: {error}"))
})?;
if committed_authority.committed_head_token()
!= meerkat_core::session_head_cas_token(boundary_head)
.map_err(|error| conflict(format!("committed head token is invalid: {error}")))?
{
return Err(conflict(
"committed runtime authority token differs from its retained boundary head"
.to_string(),
));
}
let provisional_target_applied = match (
&provisional_authority,
physical_head == *boundary_head,
) {
(None, true) => false,
(None, false) => {
return Err(conflict(
"newer physical head has no store-issued provisional authority".to_string(),
));
}
(Some(provisional), aligned) => {
let first_provisional_revision = committed_authority
.store_revision()
.checked_add(1)
.ok_or_else(|| {
conflict("HeadCanonical store revision exhausted".to_string())
})?;
let target_applied = provisional.physical_head_token() == physical_head_cas_token;
if provisional.authority_version() != HeadCanonicalProvisionalTailAuthority::VERSION
|| provisional.session_id() != runtime_authority.session_id()
|| provisional.base_store_revision() != committed_authority.store_revision()
|| provisional.base_committed_head_token()
!= committed_authority.committed_head_token()
|| (aligned
&& (physical_head_cas_token != committed_authority.committed_head_token()
|| provisional.physical_store_revision() != first_provisional_revision))
|| (!aligned
&& !target_applied
&& provisional.physical_store_revision() <= first_provisional_revision)
{
return Err(conflict(
"provisional authority does not bind the exact committed parent and physical head"
.to_string(),
));
}
target_applied
}
};
Ok(Self {
runtime_authority,
provisional_authority,
provisional_target_applied,
committed_session,
physical_head,
physical_head_cas_token,
physical_session,
})
}
pub(crate) fn runtime_authority(&self) -> &RuntimeSessionAuthority {
&self.runtime_authority
}
pub(crate) fn committed_session(&self) -> &Arc<meerkat_core::Session> {
&self.committed_session
}
pub(crate) fn provisional_authority(&self) -> Option<&HeadCanonicalProvisionalTailAuthority> {
self.provisional_authority.as_ref()
}
pub(crate) const fn provisional_target_applied(&self) -> bool {
self.provisional_target_applied
}
pub(crate) fn physical_head(&self) -> &meerkat_core::session_store::SessionHead {
&self.physical_head
}
pub(crate) fn physical_head_cas_token(&self) -> &str {
&self.physical_head_cas_token
}
pub(crate) fn physical_session(&self) -> &Arc<meerkat_core::Session> {
&self.physical_session
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PreparedRecoveryReceiptSource {
receipt: RunBoundaryReceipt,
exact_row_token: String,
}
impl PreparedRecoveryReceiptSource {
pub(crate) fn from_serialized_row(bytes: &[u8]) -> Result<Self, RuntimeStoreError> {
let receipt = serde_json::from_slice(bytes).map_err(|error| {
RuntimeStoreError::ReadFailed(format!("invalid durable recovery receipt row: {error}"))
})?;
Ok(Self {
receipt,
exact_row_token: format!("receipt-row-sha256:{:x}", Sha256::digest(bytes)),
})
}
pub(crate) fn receipt(&self) -> &RunBoundaryReceipt {
&self.receipt
}
pub(crate) fn exact_row_token(&self) -> &str {
&self.exact_row_token
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PreparedRecoveryReceiptDigestEnrichment {
original_receipt: RunBoundaryReceipt,
original_exact_row_token: String,
derived_conversation_digest: String,
}
impl PreparedRecoveryReceiptDigestEnrichment {
pub(crate) fn new(
source: &PreparedRecoveryReceiptSource,
derived_conversation_digest: String,
) -> Result<Self, RuntimeStoreError> {
if source.receipt.conversation_digest.is_some() || derived_conversation_digest.is_empty() {
return Err(RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: source.receipt.run_id.to_string(),
detail: "receipt enrichment must replace exactly one missing digest".to_string(),
});
}
Ok(Self {
original_receipt: source.receipt.clone(),
original_exact_row_token: source.exact_row_token.clone(),
derived_conversation_digest,
})
}
pub(crate) fn original_receipt(&self) -> &RunBoundaryReceipt {
&self.original_receipt
}
pub(crate) fn original_exact_row_token(&self) -> &str {
&self.original_exact_row_token
}
pub(crate) fn derived_conversation_digest(&self) -> &str {
&self.derived_conversation_digest
}
pub(crate) fn enriched_receipt(&self) -> RunBoundaryReceipt {
let mut receipt = self.original_receipt.clone();
receipt.conversation_digest = Some(self.derived_conversation_digest.clone());
receipt
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PreparedRuntimeSessionCommitOutcome {
Applied,
AlreadyAppliedExact,
AlreadyAppliedReleasedEquivalent,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RecoveryCommitStatus {
Committed,
AlreadyCommittedExact,
}
#[derive(Debug, Clone, PartialEq)]
pub struct PreparedRuntimeSessionCommitResult {
profile: RuntimeSessionPersistenceProfile,
outcome: PreparedRuntimeSessionCommitOutcome,
recovery_status: Option<RecoveryCommitStatus>,
downstream_projection_required: bool,
authority: Option<RuntimeSessionAuthority>,
}
impl PreparedRuntimeSessionCommitResult {
#[must_use]
pub fn committed(authority: RuntimeSessionAuthority) -> Self {
let profile = authority.profile();
Self {
profile,
outcome: PreparedRuntimeSessionCommitOutcome::Applied,
recovery_status: None,
downstream_projection_required: false,
authority: Some(authority),
}
}
#[must_use]
pub const fn receipt_only(profile: RuntimeSessionPersistenceProfile) -> Self {
Self {
profile,
outcome: PreparedRuntimeSessionCommitOutcome::Applied,
recovery_status: None,
downstream_projection_required: false,
authority: None,
}
}
#[must_use]
pub fn recovery(authority: RuntimeSessionAuthority, status: RecoveryCommitStatus) -> Self {
let mut result = Self::committed(authority);
result.recovery_status = Some(status);
if status == RecoveryCommitStatus::AlreadyCommittedExact {
result.outcome = PreparedRuntimeSessionCommitOutcome::AlreadyAppliedExact;
}
result
}
#[must_use]
pub fn already_applied_exact(mut self) -> Self {
self.outcome = PreparedRuntimeSessionCommitOutcome::AlreadyAppliedExact;
self
}
#[must_use]
pub fn already_applied_released_equivalent(mut self) -> Self {
self.outcome = PreparedRuntimeSessionCommitOutcome::AlreadyAppliedReleasedEquivalent;
self
}
#[must_use]
pub const fn profile(&self) -> RuntimeSessionPersistenceProfile {
self.profile
}
#[must_use]
pub const fn outcome(&self) -> PreparedRuntimeSessionCommitOutcome {
self.outcome
}
#[must_use]
pub const fn recovery_status(&self) -> Option<RecoveryCommitStatus> {
self.recovery_status
}
#[must_use]
pub const fn downstream_projection_required(&self) -> bool {
self.downstream_projection_required
}
#[must_use]
pub fn authority(&self) -> Option<&RuntimeSessionAuthority> {
self.authority.as_ref()
}
}
#[derive(Debug, Clone, thiserror::Error)]
#[non_exhaustive]
pub enum RuntimeStoreError {
#[error("Store write failed: {0}")]
WriteFailed(String),
#[error("Store read failed: {0}")]
ReadFailed(String),
#[error("Session store key mismatch: expected {expected}, actual {actual}")]
SessionKeyMismatch {
expected: meerkat_core::types::SessionId,
actual: meerkat_core::types::SessionId,
},
#[error("Not found: {0}")]
NotFound(String),
#[error("Unsupported store operation: {0}")]
Unsupported(String),
#[error("runtime store profile '{profile}' must override commit_prepared_session_boundary")]
PreparedSessionBoundaryRequiresOverride {
profile: RuntimeSessionPersistenceProfile,
},
#[error(
"runtime store profile '{profile}' cannot atomically CAS the physical session head for prepared recovery"
)]
PreparedRecoveryRequiresAtomicPhysicalHeadCas {
profile: RuntimeSessionPersistenceProfile,
},
#[error(
"head-canonical profile activation is required for runtime '{runtime_id}' (state: {state})"
)]
HeadCanonicalActivationRequired {
runtime_id: String,
state: String,
},
#[error("session persistence authority conflict for runtime '{runtime_id}': {detail}")]
SessionPersistenceAuthorityConflict { runtime_id: String, detail: String },
#[error("Ops lifecycle epoch {epoch_id} for runtime {runtime_id} is retired")]
OpsLifecycleEpochRetired {
runtime_id: String,
epoch_id: meerkat_core::RuntimeEpochId,
},
#[error("Unregister finalization outcome is unknown: {0}")]
UnregisterFinalizationOutcomeUnknown(String),
#[error("Transcript revision conflict: expected {expected}, actual {actual}")]
TranscriptRevisionConflict { expected: String, actual: String },
#[error("Session snapshot for runtime '{runtime_id}' was superseded by the durable head")]
SessionSnapshotSuperseded { runtime_id: String },
#[error("Invalid input-state batch compare-and-swap: {reason}")]
InvalidInputStateBatchCas { reason: String },
#[error(
"input idempotency index for runtime '{runtime_id}' cannot prove key '{key}' while \
input row '{evidence_input_id}' is unindexable: {reason}"
)]
InputIdempotencyIndexUncertain {
runtime_id: String,
key: String,
evidence_input_id: String,
reason: String,
},
#[error("Machine lifecycle repair is blocked: {detail}")]
MachineLifecycleRepairBlocked {
evidence_digest: Option<String>,
detail: String,
},
#[error(
"schema for domain '{domain}' is from the future: file has version {found}, \
this binary supports up to {supported}"
)]
SchemaFromTheFuture {
domain: String,
found: i64,
supported: i64,
},
#[error("maintenance fence is held for '{path}'; storage is under offline maintenance")]
MaintenanceFenceHeld { path: String },
#[error(
"Input row version conflict for input '{input_id}': the stored row changed since it was observed"
)]
InputRowVersionConflict { input_id: String },
#[error(
"Recovery input-set conflict for runtime '{runtime_id}': the nonterminal input set changed since it was observed"
)]
RecoveryInputSetConflict { runtime_id: String },
#[error(
"Machine lifecycle version conflict for runtime '{runtime_id}': the stored row changed since it was observed"
)]
MachineLifecycleVersionConflict { runtime_id: String },
#[error("Internal error: {0}")]
Internal(String),
}
pub type AuthOAuthFlowSnapshotUpdate<'a> =
dyn FnMut(Option<&[u8]>) -> Result<Vec<u8>, RuntimeStoreError> + 'a;
#[derive(Debug, Clone)]
pub struct SerializedSessionSnapshot {
pub session_snapshot: std::sync::Arc<Vec<u8>>,
}
fn recovery_class_name(
class: crate::meerkat_machine::dsl::DurableTailRecoveryClass,
) -> &'static str {
use crate::meerkat_machine::dsl::DurableTailRecoveryClass;
match class {
DurableTailRecoveryClass::CompletedCandidate => "completed_candidate",
DurableTailRecoveryClass::InterruptedRepairableCandidate => {
"interrupted_repairable_candidate"
}
DurableTailRecoveryClass::Ambiguous => "ambiguous",
}
}
fn recovery_class_from_name(
name: &str,
) -> Result<crate::meerkat_machine::dsl::DurableTailRecoveryClass, RuntimeStoreError> {
use crate::meerkat_machine::dsl::DurableTailRecoveryClass;
match name {
"completed_candidate" => Ok(DurableTailRecoveryClass::CompletedCandidate),
"interrupted_repairable_candidate" => {
Ok(DurableTailRecoveryClass::InterruptedRepairableCandidate)
}
"ambiguous" => Ok(DurableTailRecoveryClass::Ambiguous),
other => Err(RuntimeStoreError::ReadFailed(format!(
"unknown committed recovery class '{other}'"
))),
}
}
fn recovery_disposition_name(
disposition: crate::meerkat_machine::dsl::DurableTailRecoveryDisposition,
) -> &'static str {
use crate::meerkat_machine::dsl::DurableTailRecoveryDisposition;
match disposition {
DurableTailRecoveryDisposition::RefuseRecovery => "refuse_recovery",
DurableTailRecoveryDisposition::CommitCompleted => "commit_completed",
DurableTailRecoveryDisposition::RepairAndCommitInterrupted => {
"repair_and_commit_interrupted"
}
DurableTailRecoveryDisposition::CommitCompletedRetainInputs => {
"commit_completed_retain_inputs"
}
DurableTailRecoveryDisposition::HoldIntact => "hold_intact",
}
}
fn recovery_disposition_from_name(
name: &str,
) -> Result<crate::meerkat_machine::dsl::DurableTailRecoveryDisposition, RuntimeStoreError> {
use crate::meerkat_machine::dsl::DurableTailRecoveryDisposition;
match name {
"refuse_recovery" => Ok(DurableTailRecoveryDisposition::RefuseRecovery),
"commit_completed" => Ok(DurableTailRecoveryDisposition::CommitCompleted),
"repair_and_commit_interrupted" => {
Ok(DurableTailRecoveryDisposition::RepairAndCommitInterrupted)
}
"commit_completed_retain_inputs" => {
Ok(DurableTailRecoveryDisposition::CommitCompletedRetainInputs)
}
"hold_intact" => Ok(DurableTailRecoveryDisposition::HoldIntact),
other => Err(RuntimeStoreError::ReadFailed(format!(
"unknown committed recovery disposition '{other}'"
))),
}
}
fn recovery_hash_part(hasher: &mut Sha256, label: &str, bytes: &[u8]) {
hasher.update((label.len() as u64).to_be_bytes());
hasher.update(label.as_bytes());
hasher.update((bytes.len() as u64).to_be_bytes());
hasher.update(bytes);
}
fn lifecycle_expected_version_token(
lifecycle: &MachineLifecycleCommit,
) -> Result<String, RuntimeStoreError> {
match lifecycle.expected_version() {
Some(MachineLifecycleExpectedVersion::Missing) => Ok("missing".to_string()),
Some(MachineLifecycleExpectedVersion::Version(version)) => {
Ok(format!("version:{}", version.as_str()))
}
None => Err(RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: "prepared-recovery".to_string(),
detail: "recovery lifecycle commit is not fenced on an exact observed row".to_string(),
}),
}
}
fn recovery_sha256_token(bytes: &[u8]) -> String {
format!("sha256:{:x}", Sha256::digest(bytes))
}
fn is_canonical_sha256_token(token: &str) -> bool {
let Some(hex) = token.strip_prefix("sha256:") else {
return false;
};
hex.len() == 64
&& hex
.bytes()
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
}
#[derive(Debug, Clone)]
pub struct ExactInputStateObservation {
state: StoredInputState,
exact_row_digest: String,
}
impl ExactInputStateObservation {
pub fn from_exact_stored_row(
state: StoredInputState,
exact_row_digest: String,
) -> Result<Self, RuntimeStoreError> {
if !is_canonical_sha256_token(&exact_row_digest) {
return Err(RuntimeStoreError::ReadFailed(format!(
"input {} has a malformed exact-row digest",
state.state.input_id
)));
}
Ok(Self {
state,
exact_row_digest,
})
}
#[must_use]
pub fn state(&self) -> &StoredInputState {
&self.state
}
#[must_use]
pub fn exact_row_digest(&self) -> &str {
&self.exact_row_digest
}
#[must_use]
pub fn into_parts(self) -> (StoredInputState, String) {
(self.state, self.exact_row_digest)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct RecoveryInputSetRevision(u64);
impl RecoveryInputSetRevision {
#[must_use]
pub fn from_store_generation(generation: u64) -> Self {
Self(generation)
}
#[must_use]
pub fn store_generation(self) -> u64 {
self.0
}
}
#[derive(Debug, Clone)]
pub struct PreparedRecoveryInputSnapshot {
runtime_id: LogicalRuntimeId,
input_set_revision: RecoveryInputSetRevision,
rows: Vec<(StoredInputState, String)>,
exact_set_token: String,
}
impl PreparedRecoveryInputSnapshot {
pub fn from_exact_nonterminal_rows(
runtime_id: LogicalRuntimeId,
input_set_revision: RecoveryInputSetRevision,
mut rows: Vec<(StoredInputState, String)>,
) -> Result<Self, RuntimeStoreError> {
if runtime_id.0.is_empty() {
return Err(RuntimeStoreError::ReadFailed(
"recovery input snapshot has an empty logical runtime id".to_string(),
));
}
rows.sort_by(|(left, _), (right, _)| {
left.state
.input_id
.to_string()
.cmp(&right.state.input_id.to_string())
});
for (index, (state, row_token)) in rows.iter().enumerate() {
if !input_state_is_recovery_nonterminal(state) {
return Err(RuntimeStoreError::ReadFailed(format!(
"recovery input snapshot includes terminal input {}",
state.state.input_id
)));
}
if !is_canonical_sha256_token(row_token) {
return Err(RuntimeStoreError::ReadFailed(format!(
"recovery input snapshot row {} has a malformed exact-row token",
state.state.input_id
)));
}
if index > 0 && rows[index - 1].0.state.input_id == state.state.input_id {
return Err(RuntimeStoreError::ReadFailed(format!(
"recovery input snapshot repeats input {}",
state.state.input_id
)));
}
}
let mut hasher = Sha256::new();
recovery_hash_part(&mut hasher, "domain", b"meerkat.recovery-input-set.v1");
recovery_hash_part(&mut hasher, "runtime_id", runtime_id.0.as_bytes());
recovery_hash_part(
&mut hasher,
"nonterminal_row_count",
&(rows.len() as u64).to_be_bytes(),
);
for (state, row_token) in &rows {
recovery_hash_part(
&mut hasher,
"input_id",
state.state.input_id.to_string().as_bytes(),
);
recovery_hash_part(&mut hasher, "exact_row_token", row_token.as_bytes());
}
let exact_set_token = format!("sha256:{:x}", hasher.finalize());
Ok(Self {
runtime_id,
input_set_revision,
rows,
exact_set_token,
})
}
#[must_use]
pub fn runtime_id(&self) -> &LogicalRuntimeId {
&self.runtime_id
}
#[must_use]
pub fn input_set_revision(&self) -> RecoveryInputSetRevision {
self.input_set_revision
}
#[must_use]
pub fn exact_set_token(&self) -> &str {
&self.exact_set_token
}
#[must_use]
pub fn into_parts(
self,
) -> (
Vec<(StoredInputState, String)>,
RecoveryInputSetRevision,
String,
) {
(self.rows, self.input_set_revision, self.exact_set_token)
}
}
#[derive(Debug, Clone)]
pub struct PreparedRecoveryInputDelete {
input_id: InputId,
expected_row_digest: String,
}
impl PreparedRecoveryInputDelete {
pub(crate) fn from_exact_observation(
input_id: InputId,
expected_row_digest: String,
) -> Result<Self, RuntimeStoreError> {
if !is_canonical_sha256_token(&expected_row_digest) {
return Err(RuntimeStoreError::InvalidInputStateBatchCas {
reason: format!(
"recovery delete for input {input_id} has a malformed predecessor digest"
),
});
}
Ok(Self {
input_id,
expected_row_digest,
})
}
#[must_use]
pub fn input_id(&self) -> &InputId {
&self.input_id
}
#[must_use]
pub fn expected_row_digest(&self) -> &str {
&self.expected_row_digest
}
}
#[derive(Debug, Clone)]
pub enum RecoveryInputStateMutation {
Upsert(InputStatePersistenceRecord),
Delete(PreparedRecoveryInputDelete),
}
impl RecoveryInputStateMutation {
pub(crate) fn delete(
input_id: InputId,
expected_row_digest: String,
) -> Result<Self, RuntimeStoreError> {
PreparedRecoveryInputDelete::from_exact_observation(input_id, expected_row_digest)
.map(Self::Delete)
}
}
#[derive(Debug, Clone)]
struct PreparedRecoveryInputUpdate {
record: InputStatePersistenceRecord,
input_id: InputId,
expected_row_digest: String,
target_bytes: Vec<u8>,
}
impl PartialEq for PreparedRecoveryInputUpdate {
fn eq(&self, other: &Self) -> bool {
self.input_id == other.input_id
&& self.expected_row_digest == other.expected_row_digest
&& self.target_bytes == other.target_bytes
}
}
impl Eq for PreparedRecoveryInputUpdate {}
impl PreparedRecoveryInputUpdate {
fn seal(record: InputStatePersistenceRecord) -> Result<Self, String> {
let input_id = record.as_stored().state.input_id.clone();
let expected_row_digest = record
.expected_row_digest()
.ok_or_else(|| {
format!("recovery input {input_id} is not fenced on an exact predecessor row")
})?
.to_string();
if !is_canonical_sha256_token(&expected_row_digest) {
return Err(format!(
"recovery input {input_id} has a malformed predecessor digest"
));
}
let target_bytes = serde_json::to_vec(record.as_stored()).map_err(|error| {
format!("failed to encode exact recovery input target {input_id}: {error}")
})?;
Ok(Self {
record,
input_id,
expected_row_digest,
target_bytes,
})
}
fn decode(
input_id: InputId,
expected_row_digest: String,
target_bytes: Vec<u8>,
) -> Result<Self, String> {
if !is_canonical_sha256_token(&expected_row_digest) {
return Err(format!(
"recovery input {input_id} has a malformed predecessor digest"
));
}
if target_bytes.is_empty() {
return Err(format!(
"recovery input {input_id} has an empty serialized target"
));
}
let bundle: StoredInputState = serde_json::from_slice(&target_bytes)
.map_err(|error| format!("recovery input {input_id} target is invalid: {error}"))?;
if bundle.state.input_id != input_id {
return Err(format!(
"recovery input target {} differs from sealed input {input_id}",
bundle.state.input_id
));
}
let canonical_target_bytes = serde_json::to_vec(&bundle).map_err(|error| {
format!("failed to canonicalize recovery input target {input_id}: {error}")
})?;
if canonical_target_bytes != target_bytes {
return Err(format!(
"recovery input {input_id} target is not in its canonical serialized form"
));
}
let record = InputStatePersistenceRecord::from_machine_snapshot(bundle)
.map_err(|error| {
format!("recovery input {input_id} target is not machine-authorized: {error}")
})?
.with_expected_row_digest(expected_row_digest.clone());
Ok(Self {
record,
input_id,
expected_row_digest,
target_bytes,
})
}
}
#[derive(Debug)]
pub(crate) enum PreparedRecoveryInputStateMutation {
Upsert {
replacement: StoredInputState,
expected_row_digest: String,
},
Delete {
input_id: InputId,
expected_row_digest: String,
},
}
impl PreparedRecoveryInputStateMutation {
pub(crate) fn input_id(&self) -> &InputId {
match self {
Self::Upsert { replacement, .. } => &replacement.state.input_id,
Self::Delete { input_id, .. } => input_id,
}
}
pub(crate) fn expected_row_digest(&self) -> &str {
match self {
Self::Upsert {
expected_row_digest,
..
}
| Self::Delete {
expected_row_digest,
..
} => expected_row_digest,
}
}
}
pub(crate) fn prepare_recovery_input_state_mutations(
mutations: &[RecoveryInputStateMutation],
) -> Result<Vec<PreparedRecoveryInputStateMutation>, RuntimeStoreError> {
let mut prepared = mutations
.iter()
.cloned()
.map(|mutation| match mutation {
RecoveryInputStateMutation::Upsert(record) => {
let update = PreparedRecoveryInputUpdate::seal(record)
.map_err(|reason| RuntimeStoreError::InvalidInputStateBatchCas { reason })?;
Ok(PreparedRecoveryInputStateMutation::Upsert {
replacement: update.record.clone_stored(),
expected_row_digest: update.expected_row_digest,
})
}
RecoveryInputStateMutation::Delete(delete) => {
if !is_canonical_sha256_token(&delete.expected_row_digest) {
return Err(RuntimeStoreError::InvalidInputStateBatchCas {
reason: format!(
"recovery delete for input {} has a malformed predecessor digest",
delete.input_id
),
});
}
Ok(PreparedRecoveryInputStateMutation::Delete {
input_id: delete.input_id,
expected_row_digest: delete.expected_row_digest,
})
}
})
.collect::<Result<Vec<_>, _>>()?;
prepared.sort_by(|left, right| {
left.input_id()
.to_string()
.cmp(&right.input_id().to_string())
});
if prepared
.windows(2)
.any(|pair| pair[0].input_id() == pair[1].input_id())
{
return Err(RuntimeStoreError::InvalidInputStateBatchCas {
reason: "recovery input mutations must have unique input ids".to_string(),
});
}
Ok(prepared)
}
fn validate_recovery_input_update_order(
input_updates: &[PreparedRecoveryInputUpdate],
) -> Result<(), String> {
if input_updates
.windows(2)
.any(|window| window[0].input_id.0 >= window[1].input_id.0)
{
return Err(
"recovery input updates must have unique input ids in canonical order".to_string(),
);
}
Ok(())
}
#[derive(Debug, Clone, PartialEq, Eq)]
enum PreparedRecoverySessionAuthority {
WholeBlob {
base_store_revision: u64,
base_blob_sha256: String,
provisional_candidate_blob_sha256: String,
provisional_candidate_sequence: u64,
recovered_blob_sha256: String,
},
HeadCanonical {
committed_store_revision: u64,
committed_head_token: String,
physical_store_revision: u64,
physical_head_token: String,
recovered_head_token: String,
},
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PreparedRecoveryEvidence {
session_id: meerkat_core::types::SessionId,
candidate_id: String,
candidate_run_id: RunId,
class: crate::meerkat_machine::dsl::DurableTailRecoveryClass,
disposition: crate::meerkat_machine::dsl::DurableTailRecoveryDisposition,
session_authority: PreparedRecoverySessionAuthority,
receipt_digest_enrichments: Vec<PreparedRecoveryReceiptDigestEnrichment>,
predecessor_nonterminal_input_set_revision: RecoveryInputSetRevision,
predecessor_nonterminal_input_set_token: String,
input_updates: Vec<PreparedRecoveryInputUpdate>,
lifecycle_target_token: String,
lifecycle_target_bytes: Vec<u8>,
exact_witness: String,
}
impl PreparedRecoveryEvidence {
#[allow(clippy::too_many_arguments)]
pub(crate) fn seal_head_canonical(
recovered: &meerkat_core::Session,
document: &BoundSessionCommit,
session_id: meerkat_core::types::SessionId,
candidate_id: String,
candidate_run_id: RunId,
class: crate::meerkat_machine::dsl::DurableTailRecoveryClass,
disposition: crate::meerkat_machine::dsl::DurableTailRecoveryDisposition,
committed_store_revision: u64,
committed_head_token: String,
physical_store_revision: u64,
physical_head_token: String,
recovered_head_token: String,
receipt_digest_enrichments: Vec<PreparedRecoveryReceiptDigestEnrichment>,
predecessor_nonterminal_input_set_revision: RecoveryInputSetRevision,
predecessor_nonterminal_input_set_token: String,
input_updates: Vec<InputStatePersistenceRecord>,
receipt: &RunBoundaryReceipt,
lifecycle: &MachineLifecycleCommit,
) -> Result<Self, RuntimeStoreError> {
let conflict = |detail: String| RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: session_id.to_string(),
detail,
};
if candidate_id.is_empty()
|| committed_store_revision == 0
|| physical_store_revision <= committed_store_revision
|| committed_head_token.is_empty()
|| physical_head_token.is_empty()
|| recovered_head_token.is_empty()
|| committed_head_token == physical_head_token
|| !is_canonical_sha256_token(&predecessor_nonterminal_input_set_token)
{
return Err(conflict(
"prepared recovery contains an invalid store-issued authority transition"
.to_string(),
));
}
let valid_disposition = matches!(
(class, disposition),
(
crate::meerkat_machine::dsl::DurableTailRecoveryClass::CompletedCandidate,
crate::meerkat_machine::dsl::DurableTailRecoveryDisposition::CommitCompleted
| crate::meerkat_machine::dsl::DurableTailRecoveryDisposition::CommitCompletedRetainInputs
) | (
crate::meerkat_machine::dsl::DurableTailRecoveryClass::InterruptedRepairableCandidate,
crate::meerkat_machine::dsl::DurableTailRecoveryDisposition::RepairAndCommitInterrupted
)
);
if !valid_disposition {
return Err(conflict(format!(
"recovery class {} cannot realize disposition {}",
recovery_class_name(class),
recovery_disposition_name(disposition)
)));
}
if recovered.id() != &session_id {
return Err(conflict(format!(
"prepared recovery document belongs to {}, not {session_id}",
recovered.id()
)));
}
let head_boundary = document.head_canonical().ok_or_else(|| {
conflict("prepared recovery has no sealed head-canonical mutation".to_string())
})?;
let successor_head = head_boundary.mutation().successor_head();
let recovered_message_count = u64::try_from(recovered.messages().len()).map_err(|_| {
conflict("recovered document message count exceeds u64 authority".to_string())
})?;
let conversation_digest = recovered.transcript_content_digest().map_err(|error| {
conflict(format!(
"failed to derive recovered conversation digest: {error}"
))
})?;
let metadata_matches =
successor_head
.matches_session_metadata(recovered)
.map_err(|error| {
conflict(format!(
"failed to compare recovered document metadata authority: {error}"
))
})?;
if &successor_head.id != recovered.id()
|| successor_head.version != recovered.version()
|| successor_head.head_revision != conversation_digest
|| successor_head.message_count != recovered_message_count
|| successor_head.created_at != recovered.created_at()
|| successor_head.updated_at != recovered.updated_at()
|| successor_head.usage != recovered.total_usage()
|| !metadata_matches
{
return Err(conflict(
"prepared recovered head differs from the exact recovered document".to_string(),
));
}
let derived_recovered_head_token = meerkat_core::session_head_cas_token(successor_head)
.map_err(|error| {
conflict(format!(
"failed to derive recovered HeadCanonical token: {error}"
))
})?;
if derived_recovered_head_token != recovered_head_token {
return Err(conflict(
"prepared recovered head differs from the sealed successor token".to_string(),
));
}
if receipt.run_id != candidate_run_id || receipt.message_count != recovered.messages().len()
{
return Err(conflict(
"recovery receipt does not bind the candidate run and exact message count"
.to_string(),
));
}
if receipt.conversation_digest.as_deref() != Some(conversation_digest.as_str()) {
return Err(conflict(
"recovery receipt does not bind the exact recovered conversation".to_string(),
));
}
let mut previous_enrichment_sequence = None;
for enrichment in &receipt_digest_enrichments {
let original = enrichment.original_receipt();
if original.run_id != candidate_run_id
|| original.conversation_digest.is_some()
|| previous_enrichment_sequence
.is_some_and(|previous| original.sequence <= previous)
|| original.message_count > recovered.messages().len()
{
return Err(conflict(
"prepared recovery receipt enrichment has an invalid run, sequence, count, or pre-migration shape"
.to_string(),
));
}
let derived = recovered
.transcript_prefix_digest(original.message_count)
.map_err(|error| {
conflict(format!(
"failed to verify recovery receipt enrichment prefix: {error}"
))
})?;
if derived != enrichment.derived_conversation_digest()
|| enrichment.original_exact_row_token().is_empty()
{
return Err(conflict(
"prepared recovery receipt enrichment differs from the exact recovered transcript prefix"
.to_string(),
));
}
previous_enrichment_sequence = Some(original.sequence);
}
let input_updates = input_updates
.into_iter()
.map(PreparedRecoveryInputUpdate::seal)
.collect::<Result<Vec<_>, _>>()
.map_err(&conflict)?;
validate_recovery_input_update_order(&input_updates).map_err(&conflict)?;
let lifecycle_target_bytes = lifecycle.store_record().encode()?;
let lifecycle_target_token = recovery_sha256_token(&lifecycle_target_bytes);
let _ = lifecycle_expected_version_token(lifecycle)?;
let session_authority = PreparedRecoverySessionAuthority::HeadCanonical {
committed_store_revision,
committed_head_token,
physical_store_revision,
physical_head_token,
recovered_head_token,
};
let mut evidence = Self {
session_id,
candidate_id,
candidate_run_id,
class,
disposition,
session_authority,
receipt_digest_enrichments,
predecessor_nonterminal_input_set_revision,
predecessor_nonterminal_input_set_token,
input_updates,
lifecycle_target_token,
lifecycle_target_bytes,
exact_witness: String::new(),
};
evidence.exact_witness = evidence.compute_exact_witness(receipt).map_err(|error| {
RuntimeStoreError::WriteFailed(format!(
"failed to encode exact recovery witness: {error}"
))
})?;
evidence.verify_head_canonical_boundary(document, receipt)?;
Ok(evidence)
}
#[allow(clippy::too_many_arguments)]
pub(crate) fn seal_whole_blob(
recovered: &meerkat_core::Session,
repaired_document: Option<&BoundSessionCommit>,
session_id: meerkat_core::types::SessionId,
candidate_id: String,
candidate_run_id: RunId,
class: crate::meerkat_machine::dsl::DurableTailRecoveryClass,
disposition: crate::meerkat_machine::dsl::DurableTailRecoveryDisposition,
base_store_revision: u64,
base_blob_sha256: String,
provisional_candidate_blob_sha256: String,
provisional_candidate_sequence: u64,
recovered_blob_sha256: String,
receipt_digest_enrichments: Vec<PreparedRecoveryReceiptDigestEnrichment>,
predecessor_nonterminal_input_set_revision: RecoveryInputSetRevision,
predecessor_nonterminal_input_set_token: String,
input_updates: Vec<InputStatePersistenceRecord>,
receipt: &RunBoundaryReceipt,
lifecycle: &MachineLifecycleCommit,
) -> Result<Self, RuntimeStoreError> {
let conflict = |detail: String| RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: session_id.to_string(),
detail,
};
if candidate_id.is_empty()
|| base_store_revision == 0
|| base_blob_sha256.is_empty()
|| provisional_candidate_blob_sha256.is_empty()
|| provisional_candidate_sequence == 0
|| recovered_blob_sha256.is_empty()
|| !is_canonical_sha256_token(&predecessor_nonterminal_input_set_token)
{
return Err(conflict(
"prepared WholeBlob recovery contains an invalid store-issued authority transition"
.to_string(),
));
}
if recovered.id() != &session_id {
return Err(conflict(format!(
"prepared recovery document belongs to {}, not {session_id}",
recovered.id()
)));
}
let valid_disposition = matches!(
(class, disposition),
(
crate::meerkat_machine::dsl::DurableTailRecoveryClass::CompletedCandidate,
crate::meerkat_machine::dsl::DurableTailRecoveryDisposition::CommitCompleted
| crate::meerkat_machine::dsl::DurableTailRecoveryDisposition::CommitCompletedRetainInputs
) | (
crate::meerkat_machine::dsl::DurableTailRecoveryClass::InterruptedRepairableCandidate,
crate::meerkat_machine::dsl::DurableTailRecoveryDisposition::RepairAndCommitInterrupted
)
);
if !valid_disposition {
return Err(conflict(format!(
"recovery class {} cannot realize disposition {}",
recovery_class_name(class),
recovery_disposition_name(disposition)
)));
}
match class {
crate::meerkat_machine::dsl::DurableTailRecoveryClass::CompletedCandidate => {
if recovered_blob_sha256 != provisional_candidate_blob_sha256
|| repaired_document.is_some()
{
return Err(conflict(
"completed WholeBlob recovery must promote the exact provisional candidate without a repaired artifact"
.to_string(),
));
}
}
crate::meerkat_machine::dsl::DurableTailRecoveryClass::InterruptedRepairableCandidate => {
if recovered_blob_sha256 == provisional_candidate_blob_sha256 {
return Err(conflict(
"interrupted WholeBlob recovery must install a distinct repaired artifact"
.to_string(),
));
}
let document = repaired_document.ok_or_else(|| {
conflict(
"interrupted WholeBlob recovery has no sealed repaired artifact"
.to_string(),
)
})?;
if document.head_canonical().is_some() {
return Err(conflict(
"WholeBlob recovery unexpectedly carries a HeadCanonical mutation"
.to_string(),
));
}
let artifact = document.whole_blob_artifact().map_err(|error| {
conflict(format!(
"failed to materialize recovered WholeBlob artifact: {error}"
))
})?;
if artifact.row_sha256_token() != recovered_blob_sha256 {
return Err(conflict(
"recovered WholeBlob bytes differ from the sealed successor digest"
.to_string(),
));
}
}
crate::meerkat_machine::dsl::DurableTailRecoveryClass::Ambiguous => {
return Err(conflict(
"ambiguous WholeBlob recovery cannot be sealed".to_string(),
));
}
}
if receipt.run_id != candidate_run_id || receipt.message_count != recovered.messages().len()
{
return Err(conflict(
"recovery receipt does not bind the candidate run and exact message count"
.to_string(),
));
}
let conversation_digest = recovered.transcript_content_digest().map_err(|error| {
conflict(format!(
"failed to derive recovered conversation digest: {error}"
))
})?;
if receipt.conversation_digest.as_deref() != Some(conversation_digest.as_str()) {
return Err(conflict(
"recovery receipt does not bind the exact recovered conversation".to_string(),
));
}
let mut previous_enrichment_sequence = None;
for enrichment in &receipt_digest_enrichments {
let original = enrichment.original_receipt();
if original.run_id != candidate_run_id
|| original.conversation_digest.is_some()
|| previous_enrichment_sequence
.is_some_and(|previous| original.sequence <= previous)
|| original.message_count > recovered.messages().len()
{
return Err(conflict(
"prepared recovery receipt enrichment has an invalid run, sequence, count, or pre-migration shape"
.to_string(),
));
}
let derived = recovered
.transcript_prefix_digest(original.message_count)
.map_err(|error| {
conflict(format!(
"failed to verify recovery receipt enrichment prefix: {error}"
))
})?;
if derived != enrichment.derived_conversation_digest()
|| enrichment.original_exact_row_token().is_empty()
{
return Err(conflict(
"prepared recovery receipt enrichment differs from the exact recovered transcript prefix"
.to_string(),
));
}
previous_enrichment_sequence = Some(original.sequence);
}
let input_updates = input_updates
.into_iter()
.map(PreparedRecoveryInputUpdate::seal)
.collect::<Result<Vec<_>, _>>()
.map_err(&conflict)?;
validate_recovery_input_update_order(&input_updates).map_err(&conflict)?;
let lifecycle_target_bytes = lifecycle.store_record().encode()?;
let lifecycle_target_token = recovery_sha256_token(&lifecycle_target_bytes);
let _ = lifecycle_expected_version_token(lifecycle)?;
let mut evidence = Self {
session_id,
candidate_id,
candidate_run_id,
class,
disposition,
session_authority: PreparedRecoverySessionAuthority::WholeBlob {
base_store_revision,
base_blob_sha256,
provisional_candidate_blob_sha256,
provisional_candidate_sequence,
recovered_blob_sha256,
},
receipt_digest_enrichments,
predecessor_nonterminal_input_set_revision,
predecessor_nonterminal_input_set_token,
input_updates,
lifecycle_target_token,
lifecycle_target_bytes,
exact_witness: String::new(),
};
evidence.exact_witness = evidence.compute_exact_witness(receipt).map_err(|error| {
RuntimeStoreError::WriteFailed(format!(
"failed to encode exact recovery witness: {error}"
))
})?;
Ok(evidence)
}
fn compute_exact_witness(
&self,
receipt: &RunBoundaryReceipt,
) -> Result<String, serde_json::Error> {
let receipt_json = serde_json::to_vec(receipt)?;
let mut hasher = Sha256::new();
recovery_hash_part(
&mut hasher,
"domain",
b"meerkat.prepared-recovery-evidence.v6",
);
recovery_hash_part(
&mut hasher,
"session_id",
self.session_id.to_string().as_bytes(),
);
recovery_hash_part(&mut hasher, "candidate_id", self.candidate_id.as_bytes());
recovery_hash_part(
&mut hasher,
"candidate_run_id",
self.candidate_run_id.to_string().as_bytes(),
);
recovery_hash_part(
&mut hasher,
"class",
recovery_class_name(self.class).as_bytes(),
);
recovery_hash_part(
&mut hasher,
"disposition",
recovery_disposition_name(self.disposition).as_bytes(),
);
match &self.session_authority {
PreparedRecoverySessionAuthority::WholeBlob {
base_store_revision,
base_blob_sha256,
provisional_candidate_blob_sha256,
provisional_candidate_sequence,
recovered_blob_sha256,
} => {
recovery_hash_part(&mut hasher, "profile", b"whole_blob_v1");
recovery_hash_part(
&mut hasher,
"base_store_revision",
&base_store_revision.to_be_bytes(),
);
recovery_hash_part(&mut hasher, "base_blob_sha256", base_blob_sha256.as_bytes());
recovery_hash_part(
&mut hasher,
"provisional_candidate_blob_sha256",
provisional_candidate_blob_sha256.as_bytes(),
);
recovery_hash_part(
&mut hasher,
"provisional_candidate_sequence",
&provisional_candidate_sequence.to_be_bytes(),
);
recovery_hash_part(
&mut hasher,
"recovered_blob_sha256",
recovered_blob_sha256.as_bytes(),
);
}
PreparedRecoverySessionAuthority::HeadCanonical {
committed_store_revision,
committed_head_token,
physical_store_revision,
physical_head_token,
recovered_head_token,
} => {
recovery_hash_part(&mut hasher, "profile", b"head_canonical_v1");
recovery_hash_part(
&mut hasher,
"committed_store_revision",
&committed_store_revision.to_be_bytes(),
);
recovery_hash_part(
&mut hasher,
"committed_head_token",
committed_head_token.as_bytes(),
);
recovery_hash_part(
&mut hasher,
"physical_store_revision",
&physical_store_revision.to_be_bytes(),
);
recovery_hash_part(
&mut hasher,
"physical_head_token",
physical_head_token.as_bytes(),
);
recovery_hash_part(
&mut hasher,
"recovered_head_token",
recovered_head_token.as_bytes(),
);
}
}
recovery_hash_part(
&mut hasher,
"receipt_digest_enrichment_count",
&(self.receipt_digest_enrichments.len() as u64).to_be_bytes(),
);
for enrichment in &self.receipt_digest_enrichments {
let original_json = serde_json::to_vec(enrichment.original_receipt())?;
recovery_hash_part(
&mut hasher,
"receipt_digest_enrichment_original",
&original_json,
);
recovery_hash_part(
&mut hasher,
"receipt_digest_enrichment_original_token",
enrichment.original_exact_row_token().as_bytes(),
);
recovery_hash_part(
&mut hasher,
"receipt_digest_enrichment_derived_digest",
enrichment.derived_conversation_digest().as_bytes(),
);
}
recovery_hash_part(
&mut hasher,
"predecessor_nonterminal_input_set_revision",
&self
.predecessor_nonterminal_input_set_revision
.store_generation()
.to_be_bytes(),
);
recovery_hash_part(
&mut hasher,
"predecessor_nonterminal_input_set_token",
self.predecessor_nonterminal_input_set_token.as_bytes(),
);
recovery_hash_part(
&mut hasher,
"input_update_count",
&(self.input_updates.len() as u64).to_be_bytes(),
);
for input_update in &self.input_updates {
recovery_hash_part(
&mut hasher,
"input_update_id",
input_update.input_id.to_string().as_bytes(),
);
recovery_hash_part(
&mut hasher,
"input_update_expected_row_digest",
input_update.expected_row_digest.as_bytes(),
);
recovery_hash_part(
&mut hasher,
"input_update_target",
&input_update.target_bytes,
);
}
recovery_hash_part(&mut hasher, "receipt", &receipt_json);
recovery_hash_part(
&mut hasher,
"lifecycle_target_token",
self.lifecycle_target_token.as_bytes(),
);
recovery_hash_part(
&mut hasher,
"lifecycle_target",
&self.lifecycle_target_bytes,
);
Ok(format!("sha256:{:x}", hasher.finalize()))
}
pub(crate) fn verify_head_canonical_boundary(
&self,
document: &BoundSessionCommit,
receipt: &RunBoundaryReceipt,
) -> Result<(), RuntimeStoreError> {
let conflict = |detail: String| RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: self.session_id.to_string(),
detail,
};
let boundary = document.head_canonical().ok_or_else(|| {
conflict("prepared recovery has no sealed head-canonical mutation".to_string())
})?;
let PreparedRecoverySessionAuthority::HeadCanonical {
physical_head_token,
recovered_head_token,
..
} = &self.session_authority
else {
return Err(conflict(
"WholeBlob recovery evidence cannot authorize a HeadCanonical mutation".to_string(),
));
};
let mutation = boundary.mutation();
if mutation.session_id() != &self.session_id
|| mutation.predecessor_head_token() != Some(physical_head_token.as_str())
{
return Err(conflict(
"prepared recovery head mutation differs from sealed source/successor authority"
.to_string(),
));
}
let successor = mutation.successor_head();
let successor_token = meerkat_core::session_head_cas_token(successor).map_err(|error| {
conflict(format!(
"prepared recovery successor token is invalid: {error}"
))
})?;
let receipt_count = u64::try_from(receipt.message_count).map_err(|_| {
conflict("recovery receipt message count does not fit head authority".to_string())
})?;
if successor_token != *recovered_head_token
|| successor.message_count != receipt_count
|| receipt.conversation_digest.as_deref() != Some(successor.head_revision.as_str())
{
return Err(conflict(
"prepared recovery head does not bind the receipt's exact transcript".to_string(),
));
}
Ok(())
}
pub(crate) fn verify_input_updates(
&self,
input_updates: &[InputStatePersistenceRecord],
) -> Result<(), RuntimeStoreError> {
let conflict = |detail: String| RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: self.session_id.to_string(),
detail,
};
let input_updates = input_updates
.iter()
.cloned()
.map(PreparedRecoveryInputUpdate::seal)
.collect::<Result<Vec<_>, _>>()
.map_err(&conflict)?;
validate_recovery_input_update_order(&input_updates).map_err(&conflict)?;
if input_updates != self.input_updates {
return Err(conflict(
"recovery input effects differ from sealed evidence".to_string(),
));
}
Ok(())
}
pub(crate) fn verify_request_effects(
&self,
receipt: &RunBoundaryReceipt,
lifecycle: &MachineLifecycleCommit,
) -> Result<(), RuntimeStoreError> {
let conflict = |detail: String| RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: self.session_id.to_string(),
detail,
};
let _ = lifecycle_expected_version_token(lifecycle)?;
let lifecycle_target_bytes = lifecycle.store_record().encode()?;
let lifecycle_target_token = recovery_sha256_token(&lifecycle_target_bytes);
if lifecycle_target_token != self.lifecycle_target_token
|| lifecycle_target_bytes != self.lifecycle_target_bytes
{
return Err(conflict(
"recovery lifecycle target differs from sealed evidence".to_string(),
));
}
let exact_witness = self.compute_exact_witness(receipt).map_err(|error| {
RuntimeStoreError::WriteFailed(format!(
"failed to re-encode exact recovery witness: {error}"
))
})?;
if exact_witness != self.exact_witness {
return Err(conflict(
"recovery receipt or sealed effects differ from exact evidence".to_string(),
));
}
Ok(())
}
pub(crate) fn cloned_input_updates(&self) -> Vec<InputStatePersistenceRecord> {
self.input_updates
.iter()
.map(|input_update| input_update.record.clone())
.collect()
}
pub(crate) fn session_id(&self) -> &meerkat_core::types::SessionId {
&self.session_id
}
pub(crate) fn candidate_id(&self) -> &str {
&self.candidate_id
}
pub(crate) fn candidate_run_id(&self) -> &RunId {
&self.candidate_run_id
}
pub(crate) fn disposition(
&self,
) -> crate::meerkat_machine::dsl::DurableTailRecoveryDisposition {
self.disposition
}
pub(crate) fn head_canonical_authority_transition(
&self,
) -> Option<(u64, &str, u64, &str, &str)> {
match &self.session_authority {
PreparedRecoverySessionAuthority::HeadCanonical {
committed_store_revision,
committed_head_token,
physical_store_revision,
physical_head_token,
recovered_head_token,
} => Some((
*committed_store_revision,
committed_head_token,
*physical_store_revision,
physical_head_token,
recovered_head_token,
)),
PreparedRecoverySessionAuthority::WholeBlob { .. } => None,
}
}
pub(crate) fn whole_blob_authority_transition(&self) -> Option<(u64, &str, &str, u64, &str)> {
match &self.session_authority {
PreparedRecoverySessionAuthority::WholeBlob {
base_store_revision,
base_blob_sha256,
provisional_candidate_blob_sha256,
provisional_candidate_sequence,
recovered_blob_sha256,
} => Some((
*base_store_revision,
base_blob_sha256,
provisional_candidate_blob_sha256,
*provisional_candidate_sequence,
recovered_blob_sha256,
)),
PreparedRecoverySessionAuthority::HeadCanonical { .. } => None,
}
}
pub(crate) fn receipt_digest_enrichments(&self) -> &[PreparedRecoveryReceiptDigestEnrichment] {
&self.receipt_digest_enrichments
}
pub(crate) fn predecessor_nonterminal_input_set_token(&self) -> &str {
&self.predecessor_nonterminal_input_set_token
}
pub(crate) fn predecessor_nonterminal_input_set_revision(&self) -> RecoveryInputSetRevision {
self.predecessor_nonterminal_input_set_revision
}
}
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct CommittedRecoveryReceiptDigestEnrichmentWire {
original_receipt: RunBoundaryReceipt,
original_exact_row_token: String,
derived_conversation_digest: String,
}
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct CommittedRecoveryInputUpdateWire {
input_id: InputId,
expected_row_digest: String,
target_bytes: Vec<u8>,
}
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(tag = "profile", rename_all = "snake_case", deny_unknown_fields)]
enum CommittedRecoverySessionAuthorityWire {
WholeBlobV1 {
base_store_revision: u64,
base_blob_sha256: String,
provisional_candidate_blob_sha256: String,
provisional_candidate_sequence: u64,
recovered_blob_sha256: String,
},
HeadCanonicalV1 {
committed_store_revision: u64,
committed_head_token: String,
physical_store_revision: u64,
physical_head_token: String,
recovered_head_token: String,
},
}
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct CommittedRecoveryBoundaryWire {
version: u16,
session_id: meerkat_core::types::SessionId,
candidate_id: String,
candidate_run_id: RunId,
class: String,
disposition: String,
session_authority: CommittedRecoverySessionAuthorityWire,
receipt_digest_enrichments: Vec<CommittedRecoveryReceiptDigestEnrichmentWire>,
predecessor_nonterminal_input_set_revision: u64,
predecessor_nonterminal_input_set_token: String,
input_updates: Vec<CommittedRecoveryInputUpdateWire>,
lifecycle_target_token: String,
lifecycle_target_bytes: Vec<u8>,
exact_witness: String,
receipt: RunBoundaryReceipt,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CommittedRecoveryBoundary {
evidence: PreparedRecoveryEvidence,
receipt: RunBoundaryReceipt,
}
impl CommittedRecoveryBoundary {
const VERSION: u16 = 6;
pub(crate) fn from_prepared(
evidence: &PreparedRecoveryEvidence,
receipt: &RunBoundaryReceipt,
) -> Self {
Self {
evidence: evidence.clone(),
receipt: receipt.clone(),
}
}
pub(crate) fn evidence(&self) -> &PreparedRecoveryEvidence {
&self.evidence
}
pub(crate) fn receipt(&self) -> &RunBoundaryReceipt {
&self.receipt
}
pub(crate) fn encode(&self) -> Result<Vec<u8>, RuntimeStoreError> {
serde_json::to_vec(&CommittedRecoveryBoundaryWire {
version: Self::VERSION,
session_id: self.evidence.session_id.clone(),
candidate_id: self.evidence.candidate_id.clone(),
candidate_run_id: self.evidence.candidate_run_id.clone(),
class: recovery_class_name(self.evidence.class).to_string(),
disposition: recovery_disposition_name(self.evidence.disposition).to_string(),
session_authority: match &self.evidence.session_authority {
PreparedRecoverySessionAuthority::WholeBlob {
base_store_revision,
base_blob_sha256,
provisional_candidate_blob_sha256,
provisional_candidate_sequence,
recovered_blob_sha256,
} => CommittedRecoverySessionAuthorityWire::WholeBlobV1 {
base_store_revision: *base_store_revision,
base_blob_sha256: base_blob_sha256.clone(),
provisional_candidate_blob_sha256: provisional_candidate_blob_sha256.clone(),
provisional_candidate_sequence: *provisional_candidate_sequence,
recovered_blob_sha256: recovered_blob_sha256.clone(),
},
PreparedRecoverySessionAuthority::HeadCanonical {
committed_store_revision,
committed_head_token,
physical_store_revision,
physical_head_token,
recovered_head_token,
} => CommittedRecoverySessionAuthorityWire::HeadCanonicalV1 {
committed_store_revision: *committed_store_revision,
committed_head_token: committed_head_token.clone(),
physical_store_revision: *physical_store_revision,
physical_head_token: physical_head_token.clone(),
recovered_head_token: recovered_head_token.clone(),
},
},
receipt_digest_enrichments: self
.evidence
.receipt_digest_enrichments
.iter()
.map(|enrichment| CommittedRecoveryReceiptDigestEnrichmentWire {
original_receipt: enrichment.original_receipt.clone(),
original_exact_row_token: enrichment.original_exact_row_token.clone(),
derived_conversation_digest: enrichment.derived_conversation_digest.clone(),
})
.collect(),
predecessor_nonterminal_input_set_revision: self
.evidence
.predecessor_nonterminal_input_set_revision
.store_generation(),
predecessor_nonterminal_input_set_token: self
.evidence
.predecessor_nonterminal_input_set_token
.clone(),
input_updates: self
.evidence
.input_updates
.iter()
.map(|input_update| CommittedRecoveryInputUpdateWire {
input_id: input_update.input_id.clone(),
expected_row_digest: input_update.expected_row_digest.clone(),
target_bytes: input_update.target_bytes.clone(),
})
.collect(),
lifecycle_target_token: self.evidence.lifecycle_target_token.clone(),
lifecycle_target_bytes: self.evidence.lifecycle_target_bytes.clone(),
exact_witness: self.evidence.exact_witness.clone(),
receipt: self.receipt.clone(),
})
.map_err(|error| {
RuntimeStoreError::WriteFailed(format!(
"failed to encode committed recovery boundary: {error}"
))
})
}
pub(crate) fn decode(bytes: &[u8]) -> Result<Self, RuntimeStoreError> {
let wire: CommittedRecoveryBoundaryWire =
serde_json::from_slice(bytes).map_err(|error| {
RuntimeStoreError::ReadFailed(format!(
"invalid committed recovery boundary: {error}"
))
})?;
if wire.version != Self::VERSION {
return Err(RuntimeStoreError::ReadFailed(format!(
"unsupported committed recovery boundary version {}",
wire.version
)));
}
let CommittedRecoveryBoundaryWire {
version: _,
session_id,
candidate_id,
candidate_run_id,
class,
disposition,
session_authority: session_authority_wire,
receipt_digest_enrichments: receipt_digest_enrichment_wires,
predecessor_nonterminal_input_set_revision,
predecessor_nonterminal_input_set_token,
input_updates: input_update_wires,
lifecycle_target_token,
lifecycle_target_bytes,
exact_witness,
receipt,
} = wire;
let session_authority = match session_authority_wire {
CommittedRecoverySessionAuthorityWire::WholeBlobV1 {
base_store_revision,
base_blob_sha256,
provisional_candidate_blob_sha256,
provisional_candidate_sequence,
recovered_blob_sha256,
} if base_store_revision != 0
&& !base_blob_sha256.is_empty()
&& !provisional_candidate_blob_sha256.is_empty()
&& provisional_candidate_sequence != 0
&& !recovered_blob_sha256.is_empty() =>
{
PreparedRecoverySessionAuthority::WholeBlob {
base_store_revision,
base_blob_sha256,
provisional_candidate_blob_sha256,
provisional_candidate_sequence,
recovered_blob_sha256,
}
}
CommittedRecoverySessionAuthorityWire::HeadCanonicalV1 {
committed_store_revision,
committed_head_token,
physical_store_revision,
physical_head_token,
recovered_head_token,
} if committed_store_revision != 0
&& physical_store_revision > committed_store_revision
&& !committed_head_token.is_empty()
&& !physical_head_token.is_empty()
&& !recovered_head_token.is_empty()
&& committed_head_token != physical_head_token =>
{
PreparedRecoverySessionAuthority::HeadCanonical {
committed_store_revision,
committed_head_token,
physical_store_revision,
physical_head_token,
recovered_head_token,
}
}
_ => {
return Err(RuntimeStoreError::ReadFailed(
"committed recovery boundary contains an invalid store authority transition"
.to_string(),
));
}
};
if candidate_id.is_empty()
|| !is_canonical_sha256_token(&predecessor_nonterminal_input_set_token)
|| lifecycle_target_bytes.is_empty()
|| !is_canonical_sha256_token(&lifecycle_target_token)
|| !is_canonical_sha256_token(&exact_witness)
{
return Err(RuntimeStoreError::ReadFailed(
"committed recovery boundary contains an empty exact identity".to_string(),
));
}
if recovery_sha256_token(&lifecycle_target_bytes) != lifecycle_target_token {
return Err(RuntimeStoreError::ReadFailed(
"committed recovery lifecycle target token does not match its exact bytes"
.to_string(),
));
}
let lifecycle_target_snapshot =
decode_machine_lifecycle_store_record(&lifecycle_target_bytes).map_err(|error| {
RuntimeStoreError::ReadFailed(format!(
"committed recovery lifecycle target is invalid: {error}"
))
})?;
let canonical_lifecycle_target_bytes =
MachineLifecycleStoreRecord::from_snapshot(&lifecycle_target_snapshot)
.encode()
.map_err(|error| {
RuntimeStoreError::ReadFailed(format!(
"failed to canonicalize committed recovery lifecycle target: {error}"
))
})?;
if canonical_lifecycle_target_bytes != lifecycle_target_bytes {
return Err(RuntimeStoreError::ReadFailed(
"committed recovery lifecycle target is not in canonical serialized form"
.to_string(),
));
}
if receipt.run_id != candidate_run_id {
return Err(RuntimeStoreError::ReadFailed(
"committed recovery receipt run differs from candidate run".to_string(),
));
}
let mut previous_enrichment_sequence = None;
let mut receipt_digest_enrichments =
Vec::with_capacity(receipt_digest_enrichment_wires.len());
for enrichment in receipt_digest_enrichment_wires {
if enrichment.original_receipt.run_id != candidate_run_id
|| enrichment.original_receipt.conversation_digest.is_some()
|| previous_enrichment_sequence
.is_some_and(|previous| enrichment.original_receipt.sequence <= previous)
|| enrichment.original_exact_row_token.is_empty()
|| enrichment.derived_conversation_digest.is_empty()
{
return Err(RuntimeStoreError::ReadFailed(
"committed recovery receipt enrichment is malformed".to_string(),
));
}
previous_enrichment_sequence = Some(enrichment.original_receipt.sequence);
receipt_digest_enrichments.push(PreparedRecoveryReceiptDigestEnrichment {
original_receipt: enrichment.original_receipt,
original_exact_row_token: enrichment.original_exact_row_token,
derived_conversation_digest: enrichment.derived_conversation_digest,
});
}
let input_updates = input_update_wires
.into_iter()
.map(|input_update| {
PreparedRecoveryInputUpdate::decode(
input_update.input_id,
input_update.expected_row_digest,
input_update.target_bytes,
)
})
.collect::<Result<Vec<_>, _>>()
.map_err(|detail| {
RuntimeStoreError::ReadFailed(format!(
"committed recovery input update is malformed: {detail}"
))
})?;
validate_recovery_input_update_order(&input_updates).map_err(|detail| {
RuntimeStoreError::ReadFailed(format!(
"committed recovery input update order is malformed: {detail}"
))
})?;
let class = recovery_class_from_name(&class)?;
let disposition = recovery_disposition_from_name(&disposition)?;
let mut evidence = PreparedRecoveryEvidence {
session_id,
candidate_id,
candidate_run_id,
class,
disposition,
session_authority,
receipt_digest_enrichments,
predecessor_nonterminal_input_set_revision:
RecoveryInputSetRevision::from_store_generation(
predecessor_nonterminal_input_set_revision,
),
predecessor_nonterminal_input_set_token,
input_updates,
lifecycle_target_token,
lifecycle_target_bytes,
exact_witness: String::new(),
};
let derived_exact_witness = evidence.compute_exact_witness(&receipt).map_err(|error| {
RuntimeStoreError::ReadFailed(format!(
"failed to verify committed recovery exact witness: {error}"
))
})?;
if derived_exact_witness != exact_witness {
return Err(RuntimeStoreError::ReadFailed(
"committed recovery exact witness does not match its serialized effects"
.to_string(),
));
}
evidence.exact_witness = exact_witness;
Ok(Self { evidence, receipt })
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PreparedRuntimeSessionCommitKind {
SnapshotOnly,
Success,
ServiceTurnTerminal,
MachineTerminal,
Recovery,
}
#[derive(Debug, Clone)]
pub(crate) enum PreparedRuntimeSessionCommitPayload {
SnapshotOnly {
session: BoundSessionCommit,
},
Success {
session: Option<BoundSessionCommit>,
receipt: RunBoundaryReceipt,
input_updates: Vec<InputStatePersistenceRecord>,
session_store_key: Option<meerkat_core::types::SessionId>,
},
PromoteWholeBlobSuccess {
promotion: PreparedWholeBlobProvisionalPromotion,
receipt: RunBoundaryReceipt,
input_updates: Vec<InputStatePersistenceRecord>,
session_store_key: meerkat_core::types::SessionId,
},
PromoteHeadCanonicalSuccess {
promotion: PreparedHeadCanonicalProvisionalPromotion,
receipt: RunBoundaryReceipt,
input_updates: Vec<InputStatePersistenceRecord>,
session_store_key: meerkat_core::types::SessionId,
},
ServiceTurnTerminal {
session: BoundSessionCommit,
receipt: RunBoundaryReceipt,
machine_lifecycle: MachineLifecycleCommit,
session_store_key: meerkat_core::types::SessionId,
},
PromoteWholeBlobServiceTurnTerminal {
promotion: PreparedWholeBlobProvisionalPromotion,
receipt: RunBoundaryReceipt,
machine_lifecycle: MachineLifecycleCommit,
session_store_key: meerkat_core::types::SessionId,
},
PromoteHeadCanonicalServiceTurnTerminal {
promotion: PreparedHeadCanonicalProvisionalPromotion,
receipt: RunBoundaryReceipt,
machine_lifecycle: MachineLifecycleCommit,
session_store_key: meerkat_core::types::SessionId,
},
MachineTerminal {
session: BoundSessionCommit,
receipt: RunBoundaryReceipt,
machine_lifecycle: MachineLifecycleCommit,
input_updates: Vec<InputStatePersistenceRecord>,
session_store_key: meerkat_core::types::SessionId,
},
PromoteWholeBlobMachineTerminal {
promotion: PreparedWholeBlobProvisionalPromotion,
receipt: RunBoundaryReceipt,
machine_lifecycle: MachineLifecycleCommit,
input_updates: Vec<InputStatePersistenceRecord>,
session_store_key: meerkat_core::types::SessionId,
},
PromoteHeadCanonicalMachineTerminal {
promotion: PreparedHeadCanonicalProvisionalPromotion,
receipt: RunBoundaryReceipt,
machine_lifecycle: MachineLifecycleCommit,
input_updates: Vec<InputStatePersistenceRecord>,
session_store_key: meerkat_core::types::SessionId,
},
Recovery {
session: BoundSessionCommit,
evidence: PreparedRecoveryEvidence,
receipt: RunBoundaryReceipt,
machine_lifecycle: MachineLifecycleCommit,
input_updates: Vec<InputStatePersistenceRecord>,
session_store_key: meerkat_core::types::SessionId,
},
PromoteWholeBlobRecovery {
promotion: PreparedWholeBlobRecoveryPromotion,
evidence: PreparedRecoveryEvidence,
receipt: RunBoundaryReceipt,
machine_lifecycle: MachineLifecycleCommit,
input_updates: Vec<InputStatePersistenceRecord>,
session_store_key: meerkat_core::types::SessionId,
},
}
#[derive(Debug, Clone)]
pub struct PreparedRuntimeSessionCommit {
payload: PreparedRuntimeSessionCommitPayload,
}
impl PreparedRuntimeSessionCommit {
fn validate_whole_blob_promotion_binding(
promotion: &PreparedWholeBlobProvisionalPromotion,
receipt: &RunBoundaryReceipt,
session_store_key: &meerkat_core::types::SessionId,
) -> Result<(), RuntimeStoreError> {
if promotion.authority().run_id() != &receipt.run_id {
return Err(RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: promotion.authority().session_id().to_string(),
detail: "WholeBlob promotion receipt run differs from provisional authority"
.to_string(),
});
}
if receipt.conversation_digest.as_deref() != Some(promotion.conversation_digest.as_str())
|| u64::try_from(receipt.message_count).ok() != Some(promotion.message_count)
{
return Err(RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: promotion.authority().session_id().to_string(),
detail: "WholeBlob final receipt differs from checkpoint candidate count/digest"
.to_string(),
});
}
if promotion.authority().session_id() != session_store_key {
return Err(RuntimeStoreError::SessionKeyMismatch {
expected: promotion.authority().session_id().clone(),
actual: session_store_key.clone(),
});
}
Ok(())
}
fn validate_head_canonical_promotion_binding(
promotion: &PreparedHeadCanonicalProvisionalPromotion,
receipt: &RunBoundaryReceipt,
session_store_key: &meerkat_core::types::SessionId,
) -> Result<(), RuntimeStoreError> {
if promotion.authority().run_id() != &receipt.run_id {
return Err(RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: promotion.authority().session_id().to_string(),
detail: "HeadCanonical promotion receipt run differs from provisional authority"
.to_string(),
});
}
if promotion.authority().session_id() != session_store_key {
return Err(RuntimeStoreError::SessionKeyMismatch {
expected: promotion.authority().session_id().clone(),
actual: session_store_key.clone(),
});
}
if receipt.conversation_digest.as_deref()
!= Some(promotion.checkpoint().conversation_digest())
|| u64::try_from(receipt.message_count).ok()
!= Some(promotion.checkpoint().message_count())
{
return Err(RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: promotion.authority().session_id().to_string(),
detail:
"HeadCanonical promotion terminal receipt differs from checkpoint digest/count"
.to_string(),
});
}
Ok(())
}
#[must_use]
pub fn snapshot_only(session: BoundSessionCommit) -> Self {
Self {
payload: PreparedRuntimeSessionCommitPayload::SnapshotOnly { session },
}
}
#[must_use]
pub fn success(
session: Option<BoundSessionCommit>,
receipt: RunBoundaryReceipt,
input_updates: Vec<InputStatePersistenceRecord>,
session_store_key: Option<meerkat_core::types::SessionId>,
) -> Self {
Self {
payload: PreparedRuntimeSessionCommitPayload::Success {
session,
receipt,
input_updates,
session_store_key,
},
}
}
pub fn promote_whole_blob_success(
promotion: PreparedWholeBlobProvisionalPromotion,
receipt: RunBoundaryReceipt,
input_updates: Vec<InputStatePersistenceRecord>,
session_store_key: meerkat_core::types::SessionId,
) -> Result<Self, RuntimeStoreError> {
Self::validate_whole_blob_promotion_binding(&promotion, &receipt, &session_store_key)?;
Ok(Self {
payload: PreparedRuntimeSessionCommitPayload::PromoteWholeBlobSuccess {
promotion,
receipt,
input_updates,
session_store_key,
},
})
}
pub fn promote_head_canonical_success(
promotion: PreparedHeadCanonicalProvisionalPromotion,
receipt: RunBoundaryReceipt,
input_updates: Vec<InputStatePersistenceRecord>,
session_store_key: meerkat_core::types::SessionId,
) -> Result<Self, RuntimeStoreError> {
Self::validate_head_canonical_promotion_binding(&promotion, &receipt, &session_store_key)?;
Ok(Self {
payload: PreparedRuntimeSessionCommitPayload::PromoteHeadCanonicalSuccess {
promotion,
receipt,
input_updates,
session_store_key,
},
})
}
#[must_use]
pub fn machine_terminal(
session: BoundSessionCommit,
receipt: RunBoundaryReceipt,
machine_lifecycle: MachineLifecycleCommit,
input_updates: Vec<InputStatePersistenceRecord>,
session_store_key: meerkat_core::types::SessionId,
) -> Self {
Self {
payload: PreparedRuntimeSessionCommitPayload::MachineTerminal {
session,
receipt,
machine_lifecycle,
input_updates,
session_store_key,
},
}
}
pub fn promote_whole_blob_machine_terminal(
promotion: PreparedWholeBlobProvisionalPromotion,
receipt: RunBoundaryReceipt,
machine_lifecycle: MachineLifecycleCommit,
input_updates: Vec<InputStatePersistenceRecord>,
session_store_key: meerkat_core::types::SessionId,
) -> Result<Self, RuntimeStoreError> {
Self::validate_whole_blob_promotion_binding(&promotion, &receipt, &session_store_key)?;
Ok(Self {
payload: PreparedRuntimeSessionCommitPayload::PromoteWholeBlobMachineTerminal {
promotion,
receipt,
machine_lifecycle,
input_updates,
session_store_key,
},
})
}
pub fn promote_head_canonical_machine_terminal(
promotion: PreparedHeadCanonicalProvisionalPromotion,
receipt: RunBoundaryReceipt,
machine_lifecycle: MachineLifecycleCommit,
input_updates: Vec<InputStatePersistenceRecord>,
session_store_key: meerkat_core::types::SessionId,
) -> Result<Self, RuntimeStoreError> {
Self::validate_head_canonical_promotion_binding(&promotion, &receipt, &session_store_key)?;
Ok(Self {
payload: PreparedRuntimeSessionCommitPayload::PromoteHeadCanonicalMachineTerminal {
promotion,
receipt,
machine_lifecycle,
input_updates,
session_store_key,
},
})
}
#[must_use]
pub fn service_turn_terminal(
session: BoundSessionCommit,
receipt: RunBoundaryReceipt,
machine_lifecycle: MachineLifecycleCommit,
session_store_key: meerkat_core::types::SessionId,
) -> Self {
Self {
payload: PreparedRuntimeSessionCommitPayload::ServiceTurnTerminal {
session,
receipt,
machine_lifecycle,
session_store_key,
},
}
}
pub fn promote_whole_blob_service_turn_terminal(
promotion: PreparedWholeBlobProvisionalPromotion,
receipt: RunBoundaryReceipt,
machine_lifecycle: MachineLifecycleCommit,
session_store_key: meerkat_core::types::SessionId,
) -> Result<Self, RuntimeStoreError> {
Self::validate_whole_blob_promotion_binding(&promotion, &receipt, &session_store_key)?;
Ok(Self {
payload: PreparedRuntimeSessionCommitPayload::PromoteWholeBlobServiceTurnTerminal {
promotion,
receipt,
machine_lifecycle,
session_store_key,
},
})
}
pub fn promote_head_canonical_service_turn_terminal(
promotion: PreparedHeadCanonicalProvisionalPromotion,
receipt: RunBoundaryReceipt,
machine_lifecycle: MachineLifecycleCommit,
session_store_key: meerkat_core::types::SessionId,
) -> Result<Self, RuntimeStoreError> {
Self::validate_head_canonical_promotion_binding(&promotion, &receipt, &session_store_key)?;
Ok(Self {
payload: PreparedRuntimeSessionCommitPayload::PromoteHeadCanonicalServiceTurnTerminal {
promotion,
receipt,
machine_lifecycle,
session_store_key,
},
})
}
pub(crate) fn machine_terminal_recovery(
session: BoundSessionCommit,
evidence: PreparedRecoveryEvidence,
receipt: RunBoundaryReceipt,
machine_lifecycle: MachineLifecycleCommit,
session_store_key: meerkat_core::types::SessionId,
) -> Result<Self, RuntimeStoreError> {
if &session_store_key != evidence.session_id() {
return Err(RuntimeStoreError::SessionKeyMismatch {
expected: evidence.session_id().clone(),
actual: session_store_key,
});
}
if &receipt.run_id != evidence.candidate_run_id() {
return Err(RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: evidence.session_id().to_string(),
detail: "recovery receipt run differs from sealed candidate run".to_string(),
});
}
evidence.verify_head_canonical_boundary(&session, &receipt)?;
evidence.verify_request_effects(&receipt, &machine_lifecycle)?;
let input_updates = evidence.cloned_input_updates();
Ok(Self {
payload: PreparedRuntimeSessionCommitPayload::Recovery {
session,
evidence,
receipt,
machine_lifecycle,
input_updates,
session_store_key,
},
})
}
pub(crate) fn machine_terminal_whole_blob_recovery(
repaired_document: Option<BoundSessionCommit>,
evidence: PreparedRecoveryEvidence,
receipt: RunBoundaryReceipt,
machine_lifecycle: MachineLifecycleCommit,
session_store_key: meerkat_core::types::SessionId,
) -> Result<Self, RuntimeStoreError> {
if &session_store_key != evidence.session_id() {
return Err(RuntimeStoreError::SessionKeyMismatch {
expected: evidence.session_id().clone(),
actual: session_store_key,
});
}
if &receipt.run_id != evidence.candidate_run_id() {
return Err(RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: evidence.session_id().to_string(),
detail: "recovery receipt run differs from sealed candidate run".to_string(),
});
}
evidence.verify_request_effects(&receipt, &machine_lifecycle)?;
let input_updates = evidence.cloned_input_updates();
evidence.verify_input_updates(&input_updates)?;
let promotion =
PreparedWholeBlobRecoveryPromotion::prepare(repaired_document.as_ref(), &evidence)?;
Ok(Self {
payload: PreparedRuntimeSessionCommitPayload::PromoteWholeBlobRecovery {
promotion,
evidence,
receipt,
machine_lifecycle,
input_updates,
session_store_key,
},
})
}
#[must_use]
pub fn kind(&self) -> PreparedRuntimeSessionCommitKind {
match &self.payload {
PreparedRuntimeSessionCommitPayload::SnapshotOnly { .. } => {
PreparedRuntimeSessionCommitKind::SnapshotOnly
}
PreparedRuntimeSessionCommitPayload::Success { .. }
| PreparedRuntimeSessionCommitPayload::PromoteWholeBlobSuccess { .. }
| PreparedRuntimeSessionCommitPayload::PromoteHeadCanonicalSuccess { .. } => {
PreparedRuntimeSessionCommitKind::Success
}
PreparedRuntimeSessionCommitPayload::ServiceTurnTerminal { .. }
| PreparedRuntimeSessionCommitPayload::PromoteWholeBlobServiceTurnTerminal { .. }
| PreparedRuntimeSessionCommitPayload::PromoteHeadCanonicalServiceTurnTerminal {
..
} => PreparedRuntimeSessionCommitKind::ServiceTurnTerminal,
PreparedRuntimeSessionCommitPayload::MachineTerminal { .. }
| PreparedRuntimeSessionCommitPayload::PromoteWholeBlobMachineTerminal { .. }
| PreparedRuntimeSessionCommitPayload::PromoteHeadCanonicalMachineTerminal { .. } => {
PreparedRuntimeSessionCommitKind::MachineTerminal
}
PreparedRuntimeSessionCommitPayload::Recovery { .. }
| PreparedRuntimeSessionCommitPayload::PromoteWholeBlobRecovery { .. } => {
PreparedRuntimeSessionCommitKind::Recovery
}
}
}
#[must_use]
pub fn session(&self) -> Option<&BoundSessionCommit> {
match &self.payload {
PreparedRuntimeSessionCommitPayload::SnapshotOnly { session, .. }
| PreparedRuntimeSessionCommitPayload::ServiceTurnTerminal { session, .. }
| PreparedRuntimeSessionCommitPayload::MachineTerminal { session, .. }
| PreparedRuntimeSessionCommitPayload::Recovery { session, .. } => Some(session),
PreparedRuntimeSessionCommitPayload::Success { session, .. } => session.as_ref(),
PreparedRuntimeSessionCommitPayload::PromoteWholeBlobSuccess { .. }
| PreparedRuntimeSessionCommitPayload::PromoteWholeBlobServiceTurnTerminal { .. }
| PreparedRuntimeSessionCommitPayload::PromoteWholeBlobMachineTerminal { .. }
| PreparedRuntimeSessionCommitPayload::PromoteHeadCanonicalSuccess { .. }
| PreparedRuntimeSessionCommitPayload::PromoteHeadCanonicalServiceTurnTerminal {
..
}
| PreparedRuntimeSessionCommitPayload::PromoteHeadCanonicalMachineTerminal { .. } => {
None
}
PreparedRuntimeSessionCommitPayload::PromoteWholeBlobRecovery { .. } => None,
}
}
#[must_use]
pub fn receipt(&self) -> Option<&RunBoundaryReceipt> {
match &self.payload {
PreparedRuntimeSessionCommitPayload::SnapshotOnly { .. } => None,
PreparedRuntimeSessionCommitPayload::Success { receipt, .. }
| PreparedRuntimeSessionCommitPayload::PromoteWholeBlobSuccess { receipt, .. }
| PreparedRuntimeSessionCommitPayload::PromoteHeadCanonicalSuccess {
receipt, ..
}
| PreparedRuntimeSessionCommitPayload::ServiceTurnTerminal { receipt, .. }
| PreparedRuntimeSessionCommitPayload::PromoteWholeBlobServiceTurnTerminal {
receipt,
..
}
| PreparedRuntimeSessionCommitPayload::PromoteHeadCanonicalServiceTurnTerminal {
receipt,
..
}
| PreparedRuntimeSessionCommitPayload::MachineTerminal { receipt, .. }
| PreparedRuntimeSessionCommitPayload::PromoteWholeBlobMachineTerminal {
receipt,
..
}
| PreparedRuntimeSessionCommitPayload::PromoteHeadCanonicalMachineTerminal {
receipt,
..
}
| PreparedRuntimeSessionCommitPayload::Recovery { receipt, .. }
| PreparedRuntimeSessionCommitPayload::PromoteWholeBlobRecovery { receipt, .. } => {
Some(receipt)
}
}
}
#[must_use]
pub fn input_updates(&self) -> Option<&[InputStatePersistenceRecord]> {
match &self.payload {
PreparedRuntimeSessionCommitPayload::SnapshotOnly { .. } => None,
PreparedRuntimeSessionCommitPayload::ServiceTurnTerminal { .. }
| PreparedRuntimeSessionCommitPayload::PromoteWholeBlobServiceTurnTerminal { .. }
| PreparedRuntimeSessionCommitPayload::PromoteHeadCanonicalServiceTurnTerminal {
..
} => Some(&[]),
PreparedRuntimeSessionCommitPayload::Success { input_updates, .. }
| PreparedRuntimeSessionCommitPayload::PromoteWholeBlobSuccess {
input_updates, ..
}
| PreparedRuntimeSessionCommitPayload::PromoteHeadCanonicalSuccess {
input_updates,
..
}
| PreparedRuntimeSessionCommitPayload::MachineTerminal { input_updates, .. }
| PreparedRuntimeSessionCommitPayload::PromoteWholeBlobMachineTerminal {
input_updates,
..
}
| PreparedRuntimeSessionCommitPayload::PromoteHeadCanonicalMachineTerminal {
input_updates,
..
}
| PreparedRuntimeSessionCommitPayload::Recovery { input_updates, .. }
| PreparedRuntimeSessionCommitPayload::PromoteWholeBlobRecovery {
input_updates, ..
} => Some(input_updates),
}
}
#[must_use]
pub fn session_store_key(&self) -> Option<&meerkat_core::types::SessionId> {
match &self.payload {
PreparedRuntimeSessionCommitPayload::SnapshotOnly { .. } => None,
PreparedRuntimeSessionCommitPayload::Success {
session_store_key, ..
} => session_store_key.as_ref(),
PreparedRuntimeSessionCommitPayload::PromoteWholeBlobSuccess {
session_store_key,
..
}
| PreparedRuntimeSessionCommitPayload::PromoteHeadCanonicalSuccess {
session_store_key,
..
}
| PreparedRuntimeSessionCommitPayload::ServiceTurnTerminal {
session_store_key, ..
}
| PreparedRuntimeSessionCommitPayload::PromoteWholeBlobServiceTurnTerminal {
session_store_key,
..
}
| PreparedRuntimeSessionCommitPayload::PromoteHeadCanonicalServiceTurnTerminal {
session_store_key,
..
}
| PreparedRuntimeSessionCommitPayload::MachineTerminal {
session_store_key, ..
}
| PreparedRuntimeSessionCommitPayload::PromoteWholeBlobMachineTerminal {
session_store_key,
..
}
| PreparedRuntimeSessionCommitPayload::PromoteHeadCanonicalMachineTerminal {
session_store_key,
..
}
| PreparedRuntimeSessionCommitPayload::Recovery {
session_store_key, ..
}
| PreparedRuntimeSessionCommitPayload::PromoteWholeBlobRecovery {
session_store_key,
..
} => Some(session_store_key),
}
}
#[must_use]
pub fn machine_lifecycle(&self) -> Option<&MachineLifecycleCommit> {
match &self.payload {
PreparedRuntimeSessionCommitPayload::ServiceTurnTerminal {
machine_lifecycle, ..
}
| PreparedRuntimeSessionCommitPayload::PromoteWholeBlobServiceTurnTerminal {
machine_lifecycle,
..
}
| PreparedRuntimeSessionCommitPayload::PromoteHeadCanonicalServiceTurnTerminal {
machine_lifecycle,
..
}
| PreparedRuntimeSessionCommitPayload::MachineTerminal {
machine_lifecycle, ..
}
| PreparedRuntimeSessionCommitPayload::PromoteWholeBlobMachineTerminal {
machine_lifecycle,
..
}
| PreparedRuntimeSessionCommitPayload::PromoteHeadCanonicalMachineTerminal {
machine_lifecycle,
..
}
| PreparedRuntimeSessionCommitPayload::Recovery {
machine_lifecycle, ..
}
| PreparedRuntimeSessionCommitPayload::PromoteWholeBlobRecovery {
machine_lifecycle,
..
} => Some(machine_lifecycle),
PreparedRuntimeSessionCommitPayload::SnapshotOnly { .. }
| PreparedRuntimeSessionCommitPayload::Success { .. }
| PreparedRuntimeSessionCommitPayload::PromoteWholeBlobSuccess { .. }
| PreparedRuntimeSessionCommitPayload::PromoteHeadCanonicalSuccess { .. } => None,
}
}
pub(crate) fn into_payload(self) -> PreparedRuntimeSessionCommitPayload {
self.payload
}
}
#[derive(Debug, Clone)]
pub(crate) struct PreparedWholeBlobSnapshot {
session: std::sync::Arc<meerkat_core::Session>,
serialized: SerializedSessionSnapshot,
blob_sha256: String,
}
impl PreparedWholeBlobSnapshot {
#[must_use]
pub(crate) fn session(&self) -> &meerkat_core::Session {
self.session.as_ref()
}
#[must_use]
pub(crate) fn blob_sha256(&self) -> &str {
&self.blob_sha256
}
#[must_use]
pub(crate) fn into_parts(
self,
) -> (
std::sync::Arc<meerkat_core::Session>,
SerializedSessionSnapshot,
String,
) {
(self.session, self.serialized, self.blob_sha256)
}
}
fn prepared_whole_blob_snapshot(
session: &BoundSessionCommit,
) -> Result<PreparedWholeBlobSnapshot, RuntimeStoreError> {
let typed_session = session.session_arc_cloned().ok_or_else(|| {
RuntimeStoreError::SessionPersistenceAuthorityConflict {
runtime_id: "<untyped-whole-blob-boundary>".to_string(),
detail: "prepared WholeBlob boundary requires a sealed typed Session".to_string(),
}
})?;
let artifact = session.whole_blob_artifact().map_err(|error| {
RuntimeStoreError::WriteFailed(format!(
"failed to materialize whole-blob session boundary: {error}"
))
})?;
Ok(PreparedWholeBlobSnapshot {
session: typed_session,
serialized: whole_blob_serialized_snapshot(artifact.bytes_arc()),
blob_sha256: artifact.row_sha256_token().to_string(),
})
}
fn parsed_whole_blob_snapshot(
serialized: SerializedSessionSnapshot,
) -> Result<PreparedWholeBlobSnapshot, RuntimeStoreError> {
let session = std::sync::Arc::new(
meerkat_core::Session::from_persisted_bytes(serialized.session_snapshot.as_ref()).map_err(
|error| {
RuntimeStoreError::WriteFailed(format!(
"whole-blob snapshot is not a valid Session payload: {error}"
))
},
)?,
);
let blob_sha256 = format!(
"row-sha256:{:x}",
sha2::Sha256::digest(serialized.session_snapshot.as_ref())
);
Ok(PreparedWholeBlobSnapshot {
session,
serialized,
blob_sha256,
})
}
fn whole_blob_serialized_snapshot(
session_snapshot: std::sync::Arc<Vec<u8>>,
) -> SerializedSessionSnapshot {
SerializedSessionSnapshot { session_snapshot }
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RuntimeDeliveryAuthorityRecord {
revision: u64,
state_json: Vec<u8>,
}
impl RuntimeDeliveryAuthorityRecord {
#[doc(hidden)]
pub fn from_parts(revision: u64, state_json: Vec<u8>) -> Self {
Self {
revision,
state_json,
}
}
pub fn revision(&self) -> u64 {
self.revision
}
pub fn state_json(&self) -> &[u8] {
&self.state_json
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RuntimeDeliveryStoreRecord {
delivery_id: String,
sequence: u64,
submission_json: Vec<u8>,
}
impl RuntimeDeliveryStoreRecord {
#[doc(hidden)]
pub fn from_parts(
delivery_id: impl Into<String>,
sequence: u64,
submission_json: Vec<u8>,
) -> Self {
Self {
delivery_id: delivery_id.into(),
sequence,
submission_json,
}
}
pub fn delivery_id(&self) -> &str {
&self.delivery_id
}
pub fn sequence(&self) -> u64 {
self.sequence
}
pub fn submission_json(&self) -> &[u8] {
&self.submission_json
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RuntimeDeliveryAuthorityCasOutcome {
Applied(RuntimeDeliveryAuthorityRecord),
Conflict(Option<RuntimeDeliveryAuthorityRecord>),
}
fn validated_compaction_projection_intents(
session: &meerkat_core::Session,
) -> Result<Vec<meerkat_core::CompactionProjectionIntent>, RuntimeStoreError> {
session
.validated_compaction_projection_intents()
.map_err(|error| RuntimeStoreError::WriteFailed(error.to_string()))
}
pub(crate) fn complete_compaction_projection_intent(
session: &mut meerkat_core::Session,
projection: &meerkat_core::CompactionProjectionId,
) -> Result<(), RuntimeStoreError> {
session
.complete_compaction_projection_intent(projection)
.map_err(|error| RuntimeStoreError::WriteFailed(error.to_string()))?;
Ok(())
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct MachineLifecycleBindingFacts {
agent_runtime_id: Option<String>,
fence_token: Option<u64>,
runtime_generation: Option<u64>,
runtime_epoch_id: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RevokedSupervisorReceipt {
peer_id: String,
signing_public_key: String,
epoch: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SupervisorBindingReceipt {
name: String,
peer_id: String,
address: String,
signing_public_key: String,
epoch: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SupervisorRevocationPendingReceipt {
name: String,
peer_id: String,
address: String,
signing_public_key: String,
epoch: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SupervisorRotationPersistencePhase {
PreviousRevokePending,
NextPublishPending,
Completed,
Rejected,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SupervisorRotationRejection {
OperationConflict,
NotBound,
SenderMismatch,
TargetEpochNotAdvanced,
InvalidTarget,
UnsupportedProtocolVersion,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SupervisorRotationReceipt {
operation_id: meerkat_contracts::wire::supervisor_bridge::SupervisorRotationOperationId,
phase: SupervisorRotationPersistencePhase,
rejection: Option<SupervisorRotationRejection>,
previous: SupervisorBindingReceipt,
next: SupervisorBindingReceipt,
}
impl SupervisorRotationReceipt {
pub(crate) fn new(
operation_id: meerkat_contracts::wire::supervisor_bridge::SupervisorRotationOperationId,
phase: SupervisorRotationPersistencePhase,
rejection: Option<SupervisorRotationRejection>,
previous: SupervisorBindingReceipt,
next: SupervisorBindingReceipt,
) -> Self {
Self {
operation_id,
phase,
rejection,
previous,
next,
}
}
pub fn operation_id(
&self,
) -> meerkat_contracts::wire::supervisor_bridge::SupervisorRotationOperationId {
self.operation_id
}
pub fn phase(&self) -> SupervisorRotationPersistencePhase {
self.phase
}
pub fn rejection(&self) -> Option<SupervisorRotationRejection> {
self.rejection
}
pub fn previous(&self) -> &SupervisorBindingReceipt {
&self.previous
}
pub fn next(&self) -> &SupervisorBindingReceipt {
&self.next
}
}
impl SupervisorBindingReceipt {
pub(crate) fn new(
name: String,
peer_id: String,
address: String,
signing_public_key: String,
epoch: u64,
) -> Self {
Self {
name,
peer_id,
address,
signing_public_key,
epoch,
}
}
pub fn name(&self) -> &str {
&self.name
}
pub fn peer_id(&self) -> &str {
&self.peer_id
}
pub fn address(&self) -> &str {
&self.address
}
pub fn signing_public_key(&self) -> &str {
&self.signing_public_key
}
pub fn epoch(&self) -> u64 {
self.epoch
}
}
impl RevokedSupervisorReceipt {
pub(crate) fn new(peer_id: String, signing_public_key: String, epoch: u64) -> Self {
Self {
peer_id,
signing_public_key,
epoch,
}
}
pub fn peer_id(&self) -> &str {
&self.peer_id
}
pub fn signing_public_key(&self) -> &str {
&self.signing_public_key
}
pub fn epoch(&self) -> u64 {
self.epoch
}
}
impl SupervisorRevocationPendingReceipt {
pub(crate) fn new(
name: String,
peer_id: String,
address: String,
signing_public_key: String,
epoch: u64,
) -> Self {
Self {
name,
peer_id,
address,
signing_public_key,
epoch,
}
}
pub fn name(&self) -> &str {
&self.name
}
pub fn peer_id(&self) -> &str {
&self.peer_id
}
pub fn address(&self) -> &str {
&self.address
}
pub fn signing_public_key(&self) -> &str {
&self.signing_public_key
}
pub fn epoch(&self) -> u64 {
self.epoch
}
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub enum SupervisorAuthoritySnapshot {
#[default]
UnboundNoReceipt,
Bound(SupervisorBindingReceipt),
RevocationPending(SupervisorRevocationPendingReceipt),
RotationOperation(SupervisorRotationReceipt),
RevokedReceipt(RevokedSupervisorReceipt),
WithRotationHistory {
current: Box<SupervisorAuthoritySnapshot>,
terminal_receipts: std::collections::BTreeMap<
meerkat_contracts::wire::supervisor_bridge::SupervisorRotationOperationId,
SupervisorRotationReceipt,
>,
},
}
impl MachineLifecycleBindingFacts {
pub(crate) fn new(
agent_runtime_id: Option<String>,
fence_token: Option<u64>,
runtime_generation: Option<u64>,
runtime_epoch_id: Option<String>,
) -> Self {
Self {
agent_runtime_id,
fence_token,
runtime_generation,
runtime_epoch_id,
}
}
pub fn agent_runtime_id(&self) -> Option<&str> {
self.agent_runtime_id.as_deref()
}
pub fn fence_token(&self) -> Option<u64> {
self.fence_token
}
pub fn runtime_generation(&self) -> Option<u64> {
self.runtime_generation
}
pub fn runtime_epoch_id(&self) -> Option<&str> {
self.runtime_epoch_id.as_deref()
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct MachineLifecycleObservationVersion(String);
impl MachineLifecycleObservationVersion {
pub fn from_raw_record(bytes: &[u8]) -> Self {
Self(format!("sha256:{:x}", Sha256::digest(bytes)))
}
#[must_use]
pub fn as_str(&self) -> &str {
&self.0
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum MachineLifecyclePreRunPhase {
Idle,
Attached,
Retired,
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct MachineLifecycleRunFacts {
current_run_id: Option<RunId>,
pre_run_phase: Option<MachineLifecyclePreRunPhase>,
}
impl MachineLifecycleRunFacts {
pub(crate) fn new(
current_run_id: Option<RunId>,
pre_run_phase: Option<MachineLifecyclePreRunPhase>,
) -> Self {
Self {
current_run_id,
pre_run_phase,
}
}
#[must_use]
pub fn current_run_id(&self) -> Option<&RunId> {
self.current_run_id.as_ref()
}
#[must_use]
pub fn pre_run_phase(&self) -> Option<MachineLifecyclePreRunPhase> {
self.pre_run_phase
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DecodedMachineLifecycleObservation {
record_version: u16,
runtime_state: Option<RuntimeState>,
binding: MachineLifecycleBindingFacts,
run: MachineLifecycleRunFacts,
supervisor_authority: SupervisorAuthoritySnapshot,
unregister_progress: Option<MachineUnregisterProgressSnapshot>,
}
impl DecodedMachineLifecycleObservation {
#[must_use]
pub fn record_version(&self) -> u16 {
self.record_version
}
#[must_use]
pub fn runtime_state(&self) -> Option<RuntimeState> {
self.runtime_state
}
#[must_use]
pub fn binding(&self) -> &MachineLifecycleBindingFacts {
&self.binding
}
#[must_use]
pub fn run(&self) -> &MachineLifecycleRunFacts {
&self.run
}
#[must_use]
pub fn supervisor_authority(&self) -> &SupervisorAuthoritySnapshot {
&self.supervisor_authority
}
#[must_use]
pub fn unregister_progress(&self) -> Option<&MachineUnregisterProgressSnapshot> {
self.unregister_progress.as_ref()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MachineLifecycleObservation {
Missing,
Decoded {
record: DecodedMachineLifecycleObservation,
version: MachineLifecycleObservationVersion,
},
Unsupported {
record_version: u64,
evidence_digest: String,
version: MachineLifecycleObservationVersion,
},
Malformed {
record_version: Option<u64>,
evidence_digest: String,
version: MachineLifecycleObservationVersion,
detail: String,
},
}
impl MachineLifecycleObservation {
#[must_use]
pub fn from_raw_record(bytes: &[u8]) -> Self {
classify_machine_lifecycle_record(bytes)
}
#[must_use]
pub fn version(&self) -> Option<&MachineLifecycleObservationVersion> {
match self {
Self::Missing => None,
Self::Decoded { version, .. }
| Self::Unsupported { version, .. }
| Self::Malformed { version, .. } => Some(version),
}
}
#[must_use]
pub fn evidence_digest(&self) -> Option<&str> {
match self {
Self::Unsupported {
evidence_digest, ..
}
| Self::Malformed {
evidence_digest, ..
} => Some(evidence_digest),
Self::Missing | Self::Decoded { .. } => None,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MachineLifecycleExpectedVersion {
Missing,
Version(MachineLifecycleObservationVersion),
}
impl MachineLifecycleObservation {
#[must_use]
pub fn expected_version(&self) -> MachineLifecycleExpectedVersion {
self.version()
.map_or(MachineLifecycleExpectedVersion::Missing, |version| {
MachineLifecycleExpectedVersion::Version(version.clone())
})
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RuntimeStoreWriteFenceOutcome {
Applied,
Conflict { reason: String },
Backoff { reason: String },
}
pub trait RuntimeStoreWriteFence: Send + Sync {
fn execute_if_current(
&self,
operation: Box<dyn FnOnce() -> Result<(), RuntimeStoreError> + '_>,
) -> Result<RuntimeStoreWriteFenceOutcome, RuntimeStoreError>;
}
pub(crate) fn execute_runtime_store_write_fence(
write_fence: &dyn RuntimeStoreWriteFence,
operation: impl FnOnce() -> Result<(), RuntimeStoreError>,
) -> Result<RuntimeStoreWriteFenceOutcome, RuntimeStoreError> {
let invoked = std::cell::Cell::new(false);
let operation_result = std::cell::RefCell::new(None);
let checked_operation = || {
invoked.set(true);
let result = operation();
*operation_result.borrow_mut() = Some(result.clone());
result
};
let outcome = write_fence.execute_if_current(Box::new(checked_operation))?;
if let Some(Err(error)) = operation_result.borrow_mut().take() {
return Err(error);
}
let shape_is_valid = matches!(
(&outcome, invoked.get()),
(RuntimeStoreWriteFenceOutcome::Applied, true)
| (
RuntimeStoreWriteFenceOutcome::Conflict { .. }
| RuntimeStoreWriteFenceOutcome::Backoff { .. },
false,
)
);
if !shape_is_valid {
return Err(RuntimeStoreError::Internal(
"runtime write fence returned an outcome inconsistent with operation execution"
.to_string(),
));
}
Ok(outcome)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum FencedMachineLifecycleCasOutcome {
Applied {
record: DecodedMachineLifecycleObservation,
version: MachineLifecycleObservationVersion,
},
AlreadyExact {
record: DecodedMachineLifecycleObservation,
version: MachineLifecycleObservationVersion,
},
Conflict {
current: MachineLifecycleObservation,
},
FenceConflict {
reason: String,
},
FenceBackoff {
reason: String,
},
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MachineLifecycleCasOutcome {
Applied {
version: MachineLifecycleObservationVersion,
},
Conflict {
current: MachineLifecycleObservation,
},
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MachineLifecycleSnapshot {
runtime_state: RuntimeState,
binding: MachineLifecycleBindingFacts,
run: MachineLifecycleRunFacts,
supervisor_authority: SupervisorAuthoritySnapshot,
unregister_progress: Option<MachineUnregisterProgressSnapshot>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MachineUnregisterProgressSnapshot {
runtime_loop_drain_pending: bool,
comms_drain_exit_pending: bool,
completion_waiter_drain_pending: bool,
runtime_loop_forced_abort: bool,
comms_drain_forced_abort: bool,
}
impl MachineUnregisterProgressSnapshot {
pub(crate) fn new(
runtime_loop_drain_pending: bool,
comms_drain_exit_pending: bool,
completion_waiter_drain_pending: bool,
runtime_loop_forced_abort: bool,
comms_drain_forced_abort: bool,
) -> Self {
Self {
runtime_loop_drain_pending,
comms_drain_exit_pending,
completion_waiter_drain_pending,
runtime_loop_forced_abort,
comms_drain_forced_abort,
}
}
pub(crate) fn runtime_loop_drain_pending(&self) -> bool {
self.runtime_loop_drain_pending
}
pub(crate) fn comms_drain_exit_pending(&self) -> bool {
self.comms_drain_exit_pending
}
pub(crate) fn completion_waiter_drain_pending(&self) -> bool {
self.completion_waiter_drain_pending
}
pub(crate) fn runtime_loop_forced_abort(&self) -> bool {
self.runtime_loop_forced_abort
}
pub(crate) fn comms_drain_forced_abort(&self) -> bool {
self.comms_drain_forced_abort
}
}
impl MachineLifecycleSnapshot {
pub(crate) fn new(
runtime_state: RuntimeState,
binding: MachineLifecycleBindingFacts,
supervisor_authority: SupervisorAuthoritySnapshot,
) -> Self {
Self::new_with_unregister_progress(runtime_state, binding, supervisor_authority, None)
}
pub(crate) fn new_with_unregister_progress(
runtime_state: RuntimeState,
binding: MachineLifecycleBindingFacts,
supervisor_authority: SupervisorAuthoritySnapshot,
unregister_progress: Option<MachineUnregisterProgressSnapshot>,
) -> Self {
Self::new_with_run_and_unregister_progress(
runtime_state,
binding,
MachineLifecycleRunFacts::default(),
supervisor_authority,
unregister_progress,
)
}
pub(crate) fn new_with_run_and_unregister_progress(
runtime_state: RuntimeState,
binding: MachineLifecycleBindingFacts,
run: MachineLifecycleRunFacts,
supervisor_authority: SupervisorAuthoritySnapshot,
unregister_progress: Option<MachineUnregisterProgressSnapshot>,
) -> Self {
Self {
runtime_state,
binding,
run,
supervisor_authority,
unregister_progress,
}
}
pub fn runtime_state(&self) -> RuntimeState {
self.runtime_state
}
pub fn binding(&self) -> &MachineLifecycleBindingFacts {
&self.binding
}
pub fn run(&self) -> &MachineLifecycleRunFacts {
&self.run
}
pub fn supervisor_authority(&self) -> &SupervisorAuthoritySnapshot {
&self.supervisor_authority
}
pub fn unregister_progress(&self) -> Option<&MachineUnregisterProgressSnapshot> {
self.unregister_progress.as_ref()
}
}
#[allow(
clippy::option_option,
reason = "serde distinguishes missing from explicit null"
)]
fn deserialize_present_nullable<'de, D, T>(deserializer: D) -> Result<Option<Option<T>>, D::Error>
where
D: serde::Deserializer<'de>,
T: serde::Deserialize<'de>,
{
<Option<T> as serde::Deserialize>::deserialize(deserializer).map(Some)
}
#[allow(
clippy::option_option,
reason = "serde distinguishes missing from explicit null"
)]
fn require_present_nullable<T>(
value: Option<Option<T>>,
field: &str,
) -> Result<Option<T>, RuntimeStoreError> {
value.ok_or_else(|| {
RuntimeStoreError::ReadFailed(format!(
"machine lifecycle field {field} is required (explicit null is allowed)"
))
})
}
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct MachineLifecycleBindingFactsStoreWire {
#[allow(
clippy::option_option,
reason = "serde distinguishes missing from explicit null"
)]
#[serde(default, deserialize_with = "deserialize_present_nullable")]
agent_runtime_id: Option<Option<String>>,
#[allow(
clippy::option_option,
reason = "serde distinguishes missing from explicit null"
)]
#[serde(default, deserialize_with = "deserialize_present_nullable")]
fence_token: Option<Option<u64>>,
#[allow(
clippy::option_option,
reason = "serde distinguishes missing from explicit null"
)]
#[serde(default, deserialize_with = "deserialize_present_nullable")]
runtime_generation: Option<Option<u64>>,
#[allow(
clippy::option_option,
reason = "serde distinguishes missing from explicit null"
)]
#[serde(default, deserialize_with = "deserialize_present_nullable")]
runtime_epoch_id: Option<Option<String>>,
}
#[derive(serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct MachineLifecycleBindingFactsStoreWireV1 {
agent_runtime_id: Option<String>,
fence_token: Option<u64>,
runtime_generation: Option<u64>,
runtime_epoch_id: Option<String>,
}
impl From<&MachineLifecycleBindingFacts> for MachineLifecycleBindingFactsStoreWire {
fn from(binding: &MachineLifecycleBindingFacts) -> Self {
Self {
agent_runtime_id: Some(binding.agent_runtime_id().map(ToOwned::to_owned)),
fence_token: Some(binding.fence_token()),
runtime_generation: Some(binding.runtime_generation()),
runtime_epoch_id: Some(binding.runtime_epoch_id().map(ToOwned::to_owned)),
}
}
}
impl TryFrom<MachineLifecycleBindingFactsStoreWire> for MachineLifecycleBindingFacts {
type Error = RuntimeStoreError;
fn try_from(binding: MachineLifecycleBindingFactsStoreWire) -> Result<Self, Self::Error> {
Ok(Self::new(
require_present_nullable(binding.agent_runtime_id, "binding.agent_runtime_id")?,
require_present_nullable(binding.fence_token, "binding.fence_token")?,
require_present_nullable(binding.runtime_generation, "binding.runtime_generation")?,
require_present_nullable(binding.runtime_epoch_id, "binding.runtime_epoch_id")?,
))
}
}
impl From<MachineLifecycleBindingFactsStoreWireV1> for MachineLifecycleBindingFacts {
fn from(binding: MachineLifecycleBindingFactsStoreWireV1) -> Self {
Self::new(
binding.agent_runtime_id,
binding.fence_token,
binding.runtime_generation,
binding.runtime_epoch_id,
)
}
}
#[derive(serde::Serialize)]
#[serde(deny_unknown_fields)]
struct MachineLifecycleSnapshotStoreWire {
record_version: u16,
runtime_state: RuntimeState,
binding: MachineLifecycleBindingFactsStoreWire,
current_run_id: Option<RunId>,
pre_run_phase: Option<MachineLifecyclePreRunPhase>,
supervisor_authority: SupervisorAuthoritySnapshotStoreWire,
unregister_progress: Option<MachineUnregisterProgressSnapshotStoreWire>,
}
#[derive(serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct MachineLifecycleObservationStoreWireV4 {
record_version: u16,
#[allow(
clippy::option_option,
reason = "serde distinguishes a missing phase from an explicitly absent observed phase"
)]
#[serde(default, deserialize_with = "deserialize_present_nullable")]
runtime_state: Option<Option<RuntimeState>>,
binding: MachineLifecycleBindingFactsStoreWire,
#[allow(
clippy::option_option,
reason = "serde distinguishes a missing run id from an explicitly absent run id"
)]
#[serde(default, deserialize_with = "deserialize_present_nullable")]
current_run_id: Option<Option<RunId>>,
#[allow(
clippy::option_option,
reason = "serde distinguishes a missing pre-run phase from an explicitly absent phase"
)]
#[serde(default, deserialize_with = "deserialize_present_nullable")]
pre_run_phase: Option<Option<MachineLifecyclePreRunPhase>>,
supervisor_authority: SupervisorAuthoritySnapshotStoreWire,
#[allow(
clippy::option_option,
reason = "serde distinguishes a missing v4 field from explicit null progress"
)]
#[serde(default, deserialize_with = "deserialize_present_nullable")]
unregister_progress: Option<Option<MachineUnregisterProgressSnapshotStoreWire>>,
}
#[derive(serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct MachineLifecycleSnapshotStoreWireV3 {
record_version: u16,
runtime_state: RuntimeState,
binding: MachineLifecycleBindingFactsStoreWire,
supervisor_authority: SupervisorAuthoritySnapshotStoreWire,
#[allow(
clippy::option_option,
reason = "serde distinguishes a missing v3 field from explicit null progress"
)]
#[serde(default, deserialize_with = "deserialize_present_nullable")]
unregister_progress: Option<Option<MachineUnregisterProgressSnapshotStoreWire>>,
}
#[derive(serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct MachineLifecycleSnapshotStoreWireV2 {
record_version: u16,
runtime_state: RuntimeState,
binding: MachineLifecycleBindingFactsStoreWire,
supervisor_authority: SupervisorAuthoritySnapshotStoreWire,
}
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct MachineUnregisterProgressSnapshotStoreWire {
runtime_loop_drain_pending: bool,
comms_drain_exit_pending: bool,
completion_waiter_drain_pending: bool,
runtime_loop_forced_abort: bool,
comms_drain_forced_abort: bool,
}
impl From<&MachineUnregisterProgressSnapshot> for MachineUnregisterProgressSnapshotStoreWire {
fn from(snapshot: &MachineUnregisterProgressSnapshot) -> Self {
Self {
runtime_loop_drain_pending: snapshot.runtime_loop_drain_pending(),
comms_drain_exit_pending: snapshot.comms_drain_exit_pending(),
completion_waiter_drain_pending: snapshot.completion_waiter_drain_pending(),
runtime_loop_forced_abort: snapshot.runtime_loop_forced_abort(),
comms_drain_forced_abort: snapshot.comms_drain_forced_abort(),
}
}
}
impl From<MachineUnregisterProgressSnapshotStoreWire> for MachineUnregisterProgressSnapshot {
fn from(snapshot: MachineUnregisterProgressSnapshotStoreWire) -> Self {
Self::new(
snapshot.runtime_loop_drain_pending,
snapshot.comms_drain_exit_pending,
snapshot.completion_waiter_drain_pending,
snapshot.runtime_loop_forced_abort,
snapshot.comms_drain_forced_abort,
)
}
}
#[derive(serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct MachineLifecycleSnapshotStoreWireV1 {
record_version: u16,
runtime_state: RuntimeState,
binding: MachineLifecycleBindingFactsStoreWireV1,
}
#[derive(serde::Deserialize)]
struct MachineLifecycleSnapshotStoreVersionProbe {
record_version: u16,
}
#[derive(Default, serde::Serialize, serde::Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
enum SupervisorAuthoritySnapshotStoreWire {
#[default]
UnboundNoReceipt,
Bound {
binding: SupervisorBindingReceiptStoreWire,
},
RevocationPending {
pending: SupervisorRevocationPendingReceiptStoreWire,
},
RotationOperation {
rotation: SupervisorRotationReceiptStoreWire,
},
RevokedReceipt {
receipt: RevokedSupervisorReceiptStoreWire,
},
WithRotationHistory {
current: Box<SupervisorAuthoritySnapshotStoreWire>,
terminal_receipts: Vec<SupervisorRotationReceiptStoreWire>,
},
}
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct SupervisorBindingReceiptStoreWire {
name: String,
peer_id: String,
address: String,
signing_public_key: String,
epoch: u64,
}
impl From<&SupervisorBindingReceipt> for SupervisorBindingReceiptStoreWire {
fn from(receipt: &SupervisorBindingReceipt) -> Self {
Self {
name: receipt.name().to_owned(),
peer_id: receipt.peer_id().to_owned(),
address: receipt.address().to_owned(),
signing_public_key: receipt.signing_public_key().to_owned(),
epoch: receipt.epoch(),
}
}
}
impl From<SupervisorBindingReceiptStoreWire> for SupervisorBindingReceipt {
fn from(receipt: SupervisorBindingReceiptStoreWire) -> Self {
Self::new(
receipt.name,
receipt.peer_id,
receipt.address,
receipt.signing_public_key,
receipt.epoch,
)
}
}
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct RevokedSupervisorReceiptStoreWire {
peer_id: String,
signing_public_key: String,
epoch: u64,
}
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct SupervisorRevocationPendingReceiptStoreWire {
name: String,
peer_id: String,
address: String,
signing_public_key: String,
epoch: u64,
}
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct SupervisorRotationReceiptStoreWire {
operation_id: meerkat_contracts::wire::supervisor_bridge::SupervisorRotationOperationId,
phase: SupervisorRotationPersistencePhase,
#[allow(
clippy::option_option,
reason = "serde distinguishes missing from explicit null"
)]
#[serde(default, deserialize_with = "deserialize_present_nullable")]
rejection: Option<Option<SupervisorRotationRejection>>,
previous: SupervisorBindingReceiptStoreWire,
next: SupervisorBindingReceiptStoreWire,
}
impl From<&SupervisorRotationReceipt> for SupervisorRotationReceiptStoreWire {
fn from(receipt: &SupervisorRotationReceipt) -> Self {
Self {
operation_id: receipt.operation_id(),
phase: receipt.phase(),
rejection: Some(receipt.rejection()),
previous: receipt.previous().into(),
next: receipt.next().into(),
}
}
}
impl TryFrom<SupervisorRotationReceiptStoreWire> for SupervisorRotationReceipt {
type Error = RuntimeStoreError;
fn try_from(receipt: SupervisorRotationReceiptStoreWire) -> Result<Self, Self::Error> {
Ok(Self::new(
receipt.operation_id,
receipt.phase,
require_present_nullable(receipt.rejection, "supervisor_authority.rotation.rejection")?,
receipt.previous.into(),
receipt.next.into(),
))
}
}
impl From<&SupervisorRevocationPendingReceipt> for SupervisorRevocationPendingReceiptStoreWire {
fn from(receipt: &SupervisorRevocationPendingReceipt) -> Self {
Self {
name: receipt.name().to_owned(),
peer_id: receipt.peer_id().to_owned(),
address: receipt.address().to_owned(),
signing_public_key: receipt.signing_public_key().to_owned(),
epoch: receipt.epoch(),
}
}
}
impl From<SupervisorRevocationPendingReceiptStoreWire> for SupervisorRevocationPendingReceipt {
fn from(receipt: SupervisorRevocationPendingReceiptStoreWire) -> Self {
Self::new(
receipt.name,
receipt.peer_id,
receipt.address,
receipt.signing_public_key,
receipt.epoch,
)
}
}
impl From<&RevokedSupervisorReceipt> for RevokedSupervisorReceiptStoreWire {
fn from(receipt: &RevokedSupervisorReceipt) -> Self {
Self {
peer_id: receipt.peer_id().to_owned(),
signing_public_key: receipt.signing_public_key().to_owned(),
epoch: receipt.epoch(),
}
}
}
impl From<RevokedSupervisorReceiptStoreWire> for RevokedSupervisorReceipt {
fn from(receipt: RevokedSupervisorReceiptStoreWire) -> Self {
Self::new(receipt.peer_id, receipt.signing_public_key, receipt.epoch)
}
}
impl From<&SupervisorAuthoritySnapshot> for SupervisorAuthoritySnapshotStoreWire {
fn from(snapshot: &SupervisorAuthoritySnapshot) -> Self {
match snapshot {
SupervisorAuthoritySnapshot::UnboundNoReceipt => Self::UnboundNoReceipt,
SupervisorAuthoritySnapshot::Bound(binding) => Self::Bound {
binding: binding.into(),
},
SupervisorAuthoritySnapshot::RevocationPending(pending) => Self::RevocationPending {
pending: pending.into(),
},
SupervisorAuthoritySnapshot::RotationOperation(rotation) => Self::RotationOperation {
rotation: rotation.into(),
},
SupervisorAuthoritySnapshot::RevokedReceipt(receipt) => Self::RevokedReceipt {
receipt: receipt.into(),
},
SupervisorAuthoritySnapshot::WithRotationHistory {
current,
terminal_receipts,
} => Self::WithRotationHistory {
current: Box::new(current.as_ref().into()),
terminal_receipts: terminal_receipts.values().map(Into::into).collect(),
},
}
}
}
fn supervisor_authority_read_error(
context: &str,
detail: impl std::fmt::Display,
) -> RuntimeStoreError {
RuntimeStoreError::ReadFailed(format!("{context}: {detail}"))
}
fn validate_supervisor_descriptor(
name: &str,
peer_id: &str,
address: &str,
signing_public_key: &str,
context: &str,
) -> Result<(), RuntimeStoreError> {
let pubkey = crate::comms_drain::decode_supervisor_signing_public_key(signing_public_key)
.map_err(|error| supervisor_authority_read_error(context, error))?;
let spec = meerkat_contracts::wire::supervisor_bridge::BridgePeerSpec {
name: name.to_owned(),
peer_id: peer_id.to_owned(),
address: address.to_owned(),
pubkey,
};
meerkat_core::comms::TrustedPeerDescriptor::try_from(&spec)
.map(|_| ())
.map_err(|error| supervisor_authority_read_error(context, error))
}
fn validate_supervisor_binding_receipt(
receipt: &SupervisorBindingReceipt,
context: &str,
) -> Result<(), RuntimeStoreError> {
validate_supervisor_descriptor(
receipt.name(),
receipt.peer_id(),
receipt.address(),
receipt.signing_public_key(),
context,
)
}
fn validate_revoked_supervisor_receipt(
receipt: &RevokedSupervisorReceipt,
context: &str,
) -> Result<(), RuntimeStoreError> {
let pubkey =
crate::comms_drain::decode_supervisor_signing_public_key(receipt.signing_public_key())
.map_err(|error| supervisor_authority_read_error(context, error))?;
if pubkey.iter().all(|byte| *byte == 0) {
return Err(supervisor_authority_read_error(
context,
"supervisor signing public key must be non-zero",
));
}
let peer_id = meerkat_core::comms::PeerId::parse(receipt.peer_id())
.map_err(|error| supervisor_authority_read_error(context, error))?;
let derived = meerkat_core::comms::PeerId::from_ed25519_pubkey(&pubkey);
if peer_id != derived {
return Err(supervisor_authority_read_error(
context,
format!("peer id {peer_id} does not match signing-key-derived id {derived}"),
));
}
Ok(())
}
fn validate_supervisor_rotation_receipt(
receipt: &SupervisorRotationReceipt,
terminal_history: bool,
) -> Result<(), RuntimeStoreError> {
let operation_id = receipt.operation_id();
if operation_id.as_uuid().is_nil() {
return Err(supervisor_authority_read_error(
"supervisor rotation operation",
"operation id must not be the nil UUID",
));
}
validate_supervisor_binding_receipt(
receipt.previous(),
&format!("supervisor rotation {operation_id} previous authority is invalid"),
)?;
let rejection_matches = matches!(
(receipt.phase(), receipt.rejection()),
(
SupervisorRotationPersistencePhase::PreviousRevokePending
| SupervisorRotationPersistencePhase::NextPublishPending
| SupervisorRotationPersistencePhase::Completed,
None
) | (SupervisorRotationPersistencePhase::Rejected, Some(_))
);
if !rejection_matches {
return Err(supervisor_authority_read_error(
"supervisor rotation operation",
format!("{operation_id} has inconsistent rejection state"),
));
}
if terminal_history
&& !matches!(
receipt.phase(),
SupervisorRotationPersistencePhase::Completed
| SupervisorRotationPersistencePhase::Rejected
)
{
return Err(supervisor_authority_read_error(
"supervisor rotation history",
format!("{operation_id} is not terminal"),
));
}
match receipt.phase() {
SupervisorRotationPersistencePhase::PreviousRevokePending
| SupervisorRotationPersistencePhase::NextPublishPending => {
validate_supervisor_binding_receipt(
receipt.next(),
&format!("supervisor rotation {operation_id} target is invalid"),
)?;
if receipt.next().epoch() <= receipt.previous().epoch() {
return Err(supervisor_authority_read_error(
"supervisor rotation operation",
format!(
"{operation_id} target epoch {} does not advance previous epoch {}",
receipt.next().epoch(),
receipt.previous().epoch()
),
));
}
}
SupervisorRotationPersistencePhase::Completed => {
validate_supervisor_binding_receipt(
receipt.next(),
&format!("supervisor rotation {operation_id} target is invalid"),
)?;
let exact_current_adoption = receipt.previous() == receipt.next();
if !exact_current_adoption && receipt.next().epoch() <= receipt.previous().epoch() {
return Err(supervisor_authority_read_error(
"supervisor rotation operation",
format!(
"{operation_id} completed target epoch {} does not advance previous epoch {}",
receipt.next().epoch(),
receipt.previous().epoch()
),
));
}
}
SupervisorRotationPersistencePhase::Rejected => {
let Some(rejection) = receipt.rejection() else {
return Err(supervisor_authority_read_error(
"supervisor rotation operation",
format!("{operation_id} rejected without a rejection class"),
));
};
match rejection {
SupervisorRotationRejection::InvalidTarget
| SupervisorRotationRejection::UnsupportedProtocolVersion => {
}
SupervisorRotationRejection::TargetEpochNotAdvanced => {
validate_supervisor_binding_receipt(
receipt.next(),
&format!("supervisor rotation {operation_id} rejected target is invalid"),
)?;
if receipt.next().epoch() > receipt.previous().epoch() {
return Err(supervisor_authority_read_error(
"supervisor rotation operation",
format!(
"{operation_id} rejected as non-advancing but target epoch {} advances previous epoch {}",
receipt.next().epoch(),
receipt.previous().epoch()
),
));
}
}
SupervisorRotationRejection::OperationConflict
| SupervisorRotationRejection::NotBound
| SupervisorRotationRejection::SenderMismatch => {
return Err(supervisor_authority_read_error(
"supervisor rotation operation",
format!(
"{operation_id} transient rejection {rejection:?} must not be persisted as a durable receipt"
),
));
}
}
}
}
Ok(())
}
type SupervisorEpochKeyIndex = std::collections::BTreeMap<u64, [u8; 32]>;
fn record_supervisor_epoch_key(
epochs: &mut SupervisorEpochKeyIndex,
epoch: u64,
signing_public_key: &str,
context: &str,
) -> Result<(), RuntimeStoreError> {
let key = crate::comms_drain::decode_supervisor_signing_public_key(signing_public_key)
.map_err(|error| supervisor_authority_read_error(context, error))?;
if let Some(existing) = epochs.get(&epoch) {
if existing != &key {
return Err(supervisor_authority_read_error(
context,
format!("epoch {epoch} is bound to conflicting supervisor signing keys"),
));
}
} else {
epochs.insert(epoch, key);
}
Ok(())
}
fn record_supervisor_binding_epoch(
epochs: &mut SupervisorEpochKeyIndex,
receipt: &SupervisorBindingReceipt,
context: &str,
) -> Result<(), RuntimeStoreError> {
record_supervisor_epoch_key(
epochs,
receipt.epoch(),
receipt.signing_public_key(),
context,
)
}
fn record_rotation_authoritative_epochs(
epochs: &mut SupervisorEpochKeyIndex,
receipt: &SupervisorRotationReceipt,
context: &str,
) -> Result<(), RuntimeStoreError> {
record_supervisor_binding_epoch(epochs, receipt.previous(), context)?;
if matches!(
receipt.phase(),
SupervisorRotationPersistencePhase::PreviousRevokePending
| SupervisorRotationPersistencePhase::NextPublishPending
| SupervisorRotationPersistencePhase::Completed
) {
record_supervisor_binding_epoch(epochs, receipt.next(), context)?;
}
Ok(())
}
fn record_current_authoritative_epochs(
epochs: &mut SupervisorEpochKeyIndex,
current: &SupervisorAuthoritySnapshot,
) -> Result<(), RuntimeStoreError> {
match current {
SupervisorAuthoritySnapshot::UnboundNoReceipt => Ok(()),
SupervisorAuthoritySnapshot::Bound(binding) => {
record_supervisor_binding_epoch(epochs, binding, "current supervisor authority")
}
SupervisorAuthoritySnapshot::RevocationPending(pending) => record_supervisor_epoch_key(
epochs,
pending.epoch(),
pending.signing_public_key(),
"current pending supervisor revocation authority",
),
SupervisorAuthoritySnapshot::RotationOperation(rotation) => {
record_rotation_authoritative_epochs(
epochs,
rotation,
"current supervisor rotation authority",
)
}
SupervisorAuthoritySnapshot::RevokedReceipt(receipt) => record_supervisor_epoch_key(
epochs,
receipt.epoch(),
receipt.signing_public_key(),
"current revoked supervisor authority",
),
SupervisorAuthoritySnapshot::WithRotationHistory { .. } => {
Err(RuntimeStoreError::ReadFailed(
"nested supervisor rotation history is not canonical".to_string(),
))
}
}
}
fn current_supervisor_epoch(current: &SupervisorAuthoritySnapshot) -> Option<u64> {
match current {
SupervisorAuthoritySnapshot::UnboundNoReceipt => None,
SupervisorAuthoritySnapshot::Bound(binding) => Some(binding.epoch()),
SupervisorAuthoritySnapshot::RevocationPending(pending) => Some(pending.epoch()),
SupervisorAuthoritySnapshot::RotationOperation(rotation) => Some(match rotation.phase() {
SupervisorRotationPersistencePhase::PreviousRevokePending
| SupervisorRotationPersistencePhase::Rejected => rotation.previous().epoch(),
SupervisorRotationPersistencePhase::NextPublishPending
| SupervisorRotationPersistencePhase::Completed => rotation.next().epoch(),
}),
SupervisorAuthoritySnapshot::RevokedReceipt(receipt) => Some(receipt.epoch()),
SupervisorAuthoritySnapshot::WithRotationHistory { .. } => None,
}
}
fn terminal_rotation_authority_epoch(receipt: &SupervisorRotationReceipt) -> u64 {
match receipt.phase() {
SupervisorRotationPersistencePhase::Completed => receipt.next().epoch(),
SupervisorRotationPersistencePhase::Rejected => receipt.previous().epoch(),
SupervisorRotationPersistencePhase::PreviousRevokePending
| SupervisorRotationPersistencePhase::NextPublishPending => receipt.previous().epoch(),
}
}
fn validate_supervisor_rotation_history_coherence(
current: &SupervisorAuthoritySnapshot,
terminal_receipts: &std::collections::BTreeMap<
meerkat_contracts::wire::supervisor_bridge::SupervisorRotationOperationId,
SupervisorRotationReceipt,
>,
) -> Result<(), RuntimeStoreError> {
let Some(current_epoch) = current_supervisor_epoch(current) else {
return Err(RuntimeStoreError::ReadFailed(
"supervisor rotation history requires a current authority epoch".to_string(),
));
};
let mut epochs = SupervisorEpochKeyIndex::new();
record_current_authoritative_epochs(&mut epochs, current)?;
let mut history_high_water = 0;
for receipt in terminal_receipts.values() {
record_rotation_authoritative_epochs(
&mut epochs,
receipt,
"supervisor rotation history authority",
)?;
history_high_water = history_high_water.max(terminal_rotation_authority_epoch(receipt));
}
if current_epoch < history_high_water {
return Err(RuntimeStoreError::ReadFailed(format!(
"current supervisor epoch {current_epoch} is below terminal rotation history high-water {history_high_water}"
)));
}
Ok(())
}
fn validate_supervisor_authority_snapshot(
snapshot: &SupervisorAuthoritySnapshot,
) -> Result<(), RuntimeStoreError> {
match snapshot {
SupervisorAuthoritySnapshot::UnboundNoReceipt => Ok(()),
SupervisorAuthoritySnapshot::Bound(binding) => {
validate_supervisor_binding_receipt(binding, "bound supervisor is invalid")
}
SupervisorAuthoritySnapshot::RevocationPending(pending) => validate_supervisor_descriptor(
pending.name(),
pending.peer_id(),
pending.address(),
pending.signing_public_key(),
"pending supervisor revocation authority is invalid",
),
SupervisorAuthoritySnapshot::RotationOperation(rotation) => {
validate_supervisor_rotation_receipt(rotation, false)
}
SupervisorAuthoritySnapshot::RevokedReceipt(receipt) => {
validate_revoked_supervisor_receipt(receipt, "revoked supervisor receipt is invalid")
}
SupervisorAuthoritySnapshot::WithRotationHistory {
current,
terminal_receipts,
} => {
if matches!(
current.as_ref(),
SupervisorAuthoritySnapshot::WithRotationHistory { .. }
) {
return Err(RuntimeStoreError::ReadFailed(
"nested supervisor rotation history is not canonical".to_string(),
));
}
if terminal_receipts.is_empty() {
return Err(RuntimeStoreError::ReadFailed(
"empty supervisor rotation history wrapper is not canonical".to_string(),
));
}
validate_supervisor_authority_snapshot(current)?;
for (operation_id, receipt) in terminal_receipts {
if operation_id != &receipt.operation_id() {
return Err(RuntimeStoreError::ReadFailed(format!(
"supervisor rotation history key {operation_id} does not match receipt id {}",
receipt.operation_id()
)));
}
validate_supervisor_rotation_receipt(receipt, true)?;
}
if let SupervisorAuthoritySnapshot::RotationOperation(active) = current.as_ref()
&& terminal_receipts.contains_key(&active.operation_id())
{
return Err(RuntimeStoreError::ReadFailed(
"active supervisor rotation is duplicated in terminal history".to_string(),
));
}
validate_supervisor_rotation_history_coherence(current, terminal_receipts)
}
}
}
impl TryFrom<SupervisorAuthoritySnapshotStoreWire> for SupervisorAuthoritySnapshot {
type Error = RuntimeStoreError;
fn try_from(snapshot: SupervisorAuthoritySnapshotStoreWire) -> Result<Self, Self::Error> {
match snapshot {
SupervisorAuthoritySnapshotStoreWire::UnboundNoReceipt => Ok(Self::UnboundNoReceipt),
SupervisorAuthoritySnapshotStoreWire::Bound { binding } => {
let binding = binding.into();
validate_supervisor_binding_receipt(&binding, "bound supervisor is invalid")?;
Ok(Self::Bound(binding))
}
SupervisorAuthoritySnapshotStoreWire::RevocationPending { pending } => {
let pending: SupervisorRevocationPendingReceipt = pending.into();
validate_supervisor_descriptor(
pending.name(),
pending.peer_id(),
pending.address(),
pending.signing_public_key(),
"pending supervisor revocation authority is invalid",
)?;
Ok(Self::RevocationPending(pending))
}
SupervisorAuthoritySnapshotStoreWire::RotationOperation { rotation } => {
let receipt: SupervisorRotationReceipt = rotation.try_into()?;
validate_supervisor_rotation_receipt(&receipt, false)?;
Ok(Self::RotationOperation(receipt))
}
SupervisorAuthoritySnapshotStoreWire::RevokedReceipt { receipt } => {
let receipt = receipt.into();
validate_revoked_supervisor_receipt(
&receipt,
"revoked supervisor receipt is invalid",
)?;
Ok(Self::RevokedReceipt(receipt))
}
SupervisorAuthoritySnapshotStoreWire::WithRotationHistory {
current,
terminal_receipts,
} => {
if terminal_receipts.is_empty() {
return Err(RuntimeStoreError::ReadFailed(
"empty supervisor rotation history wrapper is not canonical".to_string(),
));
}
let current = Self::try_from(*current)?;
if matches!(current, Self::WithRotationHistory { .. }) {
return Err(RuntimeStoreError::ReadFailed(
"nested supervisor rotation history is not canonical".to_string(),
));
}
let mut receipts = std::collections::BTreeMap::new();
for wire in terminal_receipts {
let receipt: SupervisorRotationReceipt = wire.try_into()?;
validate_supervisor_rotation_receipt(&receipt, true)?;
if receipts.insert(receipt.operation_id(), receipt).is_some() {
return Err(RuntimeStoreError::ReadFailed(
"supervisor rotation history contains a duplicate operation id"
.to_string(),
));
}
}
if let Self::RotationOperation(active) = ¤t
&& receipts.contains_key(&active.operation_id())
{
return Err(RuntimeStoreError::ReadFailed(
"active supervisor rotation is duplicated in terminal history".to_string(),
));
}
let snapshot = Self::WithRotationHistory {
current: Box::new(current),
terminal_receipts: receipts,
};
validate_supervisor_authority_snapshot(&snapshot)?;
Ok(snapshot)
}
}
}
}
impl From<&MachineLifecycleSnapshot> for MachineLifecycleSnapshotStoreWire {
fn from(snapshot: &MachineLifecycleSnapshot) -> Self {
Self {
record_version: MACHINE_LIFECYCLE_STORE_RECORD_VERSION,
runtime_state: snapshot.runtime_state(),
binding: snapshot.binding().into(),
current_run_id: snapshot.run().current_run_id().cloned(),
pre_run_phase: snapshot.run().pre_run_phase(),
supervisor_authority: snapshot.supervisor_authority().into(),
unregister_progress: snapshot.unregister_progress().map(Into::into),
}
}
}
fn validate_unregister_progress_snapshot(
progress: Option<&MachineUnregisterProgressSnapshot>,
) -> Result<(), RuntimeStoreError> {
if let Some(progress) = progress {
if progress.runtime_loop_drain_pending() && progress.runtime_loop_forced_abort() {
return Err(RuntimeStoreError::ReadFailed(
"unregister runtime-loop forced disposition cannot precede obligation closure"
.into(),
));
}
if progress.comms_drain_exit_pending() && progress.comms_drain_forced_abort() {
return Err(RuntimeStoreError::ReadFailed(
"unregister comms-drain forced disposition cannot precede obligation closure"
.into(),
));
}
}
Ok(())
}
impl TryFrom<MachineLifecycleSnapshotStoreWireV3> for MachineLifecycleSnapshot {
type Error = RuntimeStoreError;
fn try_from(record: MachineLifecycleSnapshotStoreWireV3) -> Result<Self, Self::Error> {
if record.record_version != UNREGISTER_MACHINE_LIFECYCLE_STORE_RECORD_VERSION {
return Err(RuntimeStoreError::ReadFailed(format!(
"unsupported machine lifecycle store record version {}",
record.record_version
)));
}
let unregister_progress =
require_present_nullable(record.unregister_progress, "unregister_progress")?
.map(Into::into);
validate_unregister_progress_snapshot(unregister_progress.as_ref())?;
Ok(Self::new_with_unregister_progress(
record.runtime_state,
record.binding.try_into()?,
record.supervisor_authority.try_into()?,
unregister_progress,
))
}
}
fn decode_machine_lifecycle_observation_v4(
bytes: &[u8],
) -> Result<DecodedMachineLifecycleObservation, RuntimeStoreError> {
let record = serde_json::from_slice::<MachineLifecycleObservationStoreWireV4>(bytes)
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?;
if record.record_version != MACHINE_LIFECYCLE_STORE_RECORD_VERSION {
return Err(RuntimeStoreError::ReadFailed(format!(
"unsupported machine lifecycle store record version {}",
record.record_version
)));
}
let runtime_state = require_present_nullable(record.runtime_state, "runtime_state")?;
let current_run_id = require_present_nullable(record.current_run_id, "current_run_id")?;
let pre_run_phase = require_present_nullable(record.pre_run_phase, "pre_run_phase")?;
let unregister_progress =
require_present_nullable(record.unregister_progress, "unregister_progress")?
.map(Into::into);
validate_unregister_progress_snapshot(unregister_progress.as_ref())?;
Ok(DecodedMachineLifecycleObservation {
record_version: record.record_version,
runtime_state,
binding: record.binding.try_into()?,
run: MachineLifecycleRunFacts::new(current_run_id, pre_run_phase),
supervisor_authority: record.supervisor_authority.try_into()?,
unregister_progress,
})
}
fn decoded_machine_lifecycle_from_snapshot(
record_version: u16,
snapshot: MachineLifecycleSnapshot,
) -> DecodedMachineLifecycleObservation {
DecodedMachineLifecycleObservation {
record_version,
runtime_state: Some(snapshot.runtime_state),
binding: snapshot.binding,
run: snapshot.run,
supervisor_authority: snapshot.supervisor_authority,
unregister_progress: snapshot.unregister_progress,
}
}
fn decode_machine_lifecycle_store_record(
bytes: &[u8],
) -> Result<MachineLifecycleSnapshot, RuntimeStoreError> {
let version = serde_json::from_slice::<MachineLifecycleSnapshotStoreVersionProbe>(bytes)
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?;
match version.record_version {
LEGACY_MACHINE_LIFECYCLE_STORE_RECORD_VERSION => {
let record = serde_json::from_slice::<MachineLifecycleSnapshotStoreWireV1>(bytes)
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?;
if record.record_version != LEGACY_MACHINE_LIFECYCLE_STORE_RECORD_VERSION {
return Err(RuntimeStoreError::ReadFailed(format!(
"unsupported machine lifecycle store record version {}",
record.record_version
)));
}
Ok(MachineLifecycleSnapshot::new(
record.runtime_state,
record.binding.into(),
SupervisorAuthoritySnapshot::UnboundNoReceipt,
))
}
SUPERVISOR_MACHINE_LIFECYCLE_STORE_RECORD_VERSION => {
let record = serde_json::from_slice::<MachineLifecycleSnapshotStoreWireV2>(bytes)
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?;
if record.record_version != SUPERVISOR_MACHINE_LIFECYCLE_STORE_RECORD_VERSION {
return Err(RuntimeStoreError::ReadFailed(format!(
"unsupported machine lifecycle store record version {}",
record.record_version
)));
}
Ok(MachineLifecycleSnapshot::new(
record.runtime_state,
record.binding.try_into()?,
record.supervisor_authority.try_into()?,
))
}
UNREGISTER_MACHINE_LIFECYCLE_STORE_RECORD_VERSION => {
let record = serde_json::from_slice::<MachineLifecycleSnapshotStoreWireV3>(bytes)
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?;
MachineLifecycleSnapshot::try_from(record)
}
MACHINE_LIFECYCLE_STORE_RECORD_VERSION => {
let record = decode_machine_lifecycle_observation_v4(bytes)?;
let runtime_state = record.runtime_state.ok_or_else(|| {
RuntimeStoreError::ReadFailed(
"machine lifecycle runtime_state cannot be null for strict recovery".into(),
)
})?;
Ok(
MachineLifecycleSnapshot::new_with_run_and_unregister_progress(
runtime_state,
record.binding,
record.run,
record.supervisor_authority,
record.unregister_progress,
),
)
}
unsupported => Err(RuntimeStoreError::ReadFailed(format!(
"unsupported machine lifecycle store record version {unsupported}"
))),
}
}
#[derive(serde::Deserialize)]
struct MachineLifecycleRawVersionProbe {
record_version: u64,
}
fn machine_lifecycle_record_version(bytes: &[u8]) -> Result<u64, String> {
serde_json::from_slice::<MachineLifecycleRawVersionProbe>(bytes)
.map(|probe| probe.record_version)
.map_err(|error| {
format!("machine lifecycle record_version is not uniquely readable: {error}")
})
}
fn classify_machine_lifecycle_record(bytes: &[u8]) -> MachineLifecycleObservation {
let version = MachineLifecycleObservationVersion::from_raw_record(bytes);
let evidence_digest = version.as_str().to_owned();
let record_version = match machine_lifecycle_record_version(bytes) {
Ok(record_version) => record_version,
Err(detail) => {
return MachineLifecycleObservation::Malformed {
record_version: None,
evidence_digest,
version,
detail,
};
}
};
let supported = [
u64::from(LEGACY_MACHINE_LIFECYCLE_STORE_RECORD_VERSION),
u64::from(SUPERVISOR_MACHINE_LIFECYCLE_STORE_RECORD_VERSION),
u64::from(UNREGISTER_MACHINE_LIFECYCLE_STORE_RECORD_VERSION),
u64::from(MACHINE_LIFECYCLE_STORE_RECORD_VERSION),
];
if !supported.contains(&record_version) {
return MachineLifecycleObservation::Unsupported {
record_version,
evidence_digest,
version,
};
}
let decoded = if record_version == u64::from(MACHINE_LIFECYCLE_STORE_RECORD_VERSION) {
decode_machine_lifecycle_observation_v4(bytes)
} else {
decode_machine_lifecycle_store_record(bytes).map(|snapshot| {
decoded_machine_lifecycle_from_snapshot(record_version as u16, snapshot)
})
};
match decoded {
Ok(record) => MachineLifecycleObservation::Decoded { record, version },
Err(error) => MachineLifecycleObservation::Malformed {
record_version: Some(record_version),
evidence_digest,
version,
detail: error.to_string(),
},
}
}
#[cfg(test)]
pub(crate) async fn assert_input_idempotency_final_image_contract(store: &dyn RuntimeStore) {
fn state_with_key(input_id: InputId, key: &str) -> StoredInputState {
let mut state = StoredInputState::new_accepted(input_id);
state.state.idempotency_key = Some(IdempotencyKey::new(key));
state
}
fn record(state: StoredInputState) -> InputStatePersistenceRecord {
InputStatePersistenceRecord::from_machine_snapshot(state).unwrap()
}
let runtime_id =
LogicalRuntimeId::new(format!("idempotency-final-image-{}", uuid::Uuid::now_v7()));
let left_id = InputId::new();
let right_id = InputId::new();
let left = state_with_key(left_id.clone(), "left-key");
let right = state_with_key(right_id.clone(), "right-key");
store
.persist_input_states_atomically(
&runtime_id,
&[record(left.clone()), record(right.clone())],
)
.await
.unwrap();
let swapped_left = state_with_key(left_id.clone(), "right-key");
let swapped_right = state_with_key(right_id.clone(), "left-key");
store
.persist_input_states_atomically(
&runtime_id,
&[record(swapped_left.clone()), record(swapped_right.clone())],
)
.await
.expect("complete-final-image persistence must permit a key swap");
let left_key_owner = store
.load_input_state_by_idempotency_key(&runtime_id, &IdempotencyKey::new("left-key"))
.await
.unwrap()
.expect("left key after swap");
let right_key_owner = store
.load_input_state_by_idempotency_key(&runtime_id, &IdempotencyKey::new("right-key"))
.await
.unwrap()
.expect("right key after swap");
assert_eq!(left_key_owner.state().state.input_id, right_id);
assert_eq!(right_key_owner.state().state.input_id, left_id);
assert_eq!(
store
.compare_and_swap_input_states_atomically(
&runtime_id,
&[swapped_left, swapped_right],
&[record(left), record(right)],
)
.await
.unwrap(),
InputStateBatchCasOutcome::Swapped,
"complete-final-image CAS must permit the reverse key swap"
);
}
#[cfg(test)]
pub(crate) fn pending_terminal_owner_fixture(
input_id: InputId,
published: bool,
) -> (StoredInputState, InputStatePersistenceRecord) {
use crate::input_state::{
InputState, InputStateSeed, InteractionTerminalBatchKey, InteractionTerminalCandidate,
InteractionTerminalOutbox, InteractionTerminalOutboxPhase, InteractionTerminalPublication,
interaction_terminal_payload_digest,
};
let candidate = InteractionTerminalCandidate::RuntimeTerminated {
reason: "indexed terminal recovery fixture".to_string(),
};
let recipients = vec![input_id.clone()];
let candidate_digest = interaction_terminal_payload_digest(&candidate).unwrap();
let completion_input_ids_digest = interaction_terminal_payload_digest(&recipients).unwrap();
let phase = if published {
InteractionTerminalOutboxPhase::Published {
finalization_failed: false,
publication: InteractionTerminalPublication {
terminal_seq: 1,
payload_digest: "published-payload".to_string(),
},
}
} else {
InteractionTerminalOutboxPhase::Candidate
};
let outbox = InteractionTerminalOutbox {
interaction_id: meerkat_core::interaction::InteractionId(input_id.0),
input_id: input_id.clone(),
batch_ordinal: 0,
batch_key: InteractionTerminalBatchKey::RuntimeTermination {
candidate_owner_input_id: input_id.clone(),
},
owner_session_id: meerkat_core::types::SessionId::new(),
owner_agent_runtime_id: Some("indexed-runtime".to_string()),
owner_fence_token: Some(1),
owner_runtime_generation: Some(1),
owner_runtime_epoch_id: Some("indexed-epoch".to_string()),
candidate_owner_input_id: input_id.clone(),
candidate: (!published).then_some(candidate),
candidate_digest,
completion_input_ids: (!published).then_some(recipients),
completion_input_ids_digest,
phase,
};
outbox.validate().unwrap();
let directed_input = crate::mob_adapter::create_tracked_flow_step_input(
"fixture-step",
meerkat_core::types::ContentInput::Text("fixture-directed-input".to_string()),
"fixture-run",
None,
&input_id.to_string(),
)
.unwrap();
let mut state = InputState::new_accepted(input_id);
state.directed_run_started_attribution =
crate::input_state::DirectedRunStartedAttribution::from_input(&directed_input).unwrap();
state.interaction_terminal_outbox = Some(outbox);
let stored = StoredInputState {
state,
seed: InputStateSeed::new_accepted(),
};
let record = InputStatePersistenceRecord::from_machine_snapshot(stored.clone()).unwrap();
(stored, record)
}
#[cfg(test)]
pub(crate) async fn assert_pending_terminal_owner_index_contract(store: &dyn RuntimeStore) {
let runtime_id =
LogicalRuntimeId::new(format!("pending-terminal-index-{}", uuid::Uuid::now_v7()));
let mut ids = [InputId::new(), InputId::new(), InputId::new()];
ids.sort_by_key(|input_id| input_id.0);
let fixtures = ids
.iter()
.cloned()
.map(|input_id| pending_terminal_owner_fixture(input_id, false))
.collect::<Vec<_>>();
for (_, record) in &fixtures {
store
.persist_input_state(&runtime_id, record)
.await
.unwrap();
}
let first_page = store
.load_pending_terminal_owner_ids_page(&runtime_id, None, 2)
.await
.unwrap();
assert_eq!(first_page, ids[..2]);
let second_page = store
.load_pending_terminal_owner_ids_page(&runtime_id, first_page.last(), 2)
.await
.unwrap();
assert_eq!(second_page, ids[2..]);
let (_, published) = pending_terminal_owner_fixture(ids[1].clone(), true);
assert_eq!(
store
.compare_and_swap_input_states_atomically(
&runtime_id,
std::slice::from_ref(&fixtures[1].0),
std::slice::from_ref(&published),
)
.await
.unwrap(),
InputStateBatchCasOutcome::Swapped
);
assert_eq!(
store
.load_pending_terminal_owner_ids_page(&runtime_id, None, 3)
.await
.unwrap(),
vec![ids[0].clone(), ids[2].clone()]
);
}
fn replacement_repair_blocked(
evidence_digest: Option<String>,
detail: impl Into<String>,
) -> RuntimeStoreError {
RuntimeStoreError::MachineLifecycleRepairBlocked {
evidence_digest,
detail: detail.into(),
}
}
fn validate_machine_lifecycle_replacement(
current: &MachineLifecycleObservation,
_current_raw: Option<&[u8]>,
_replacement: &MachineLifecycleSnapshot,
) -> Result<(), RuntimeStoreError> {
match current {
MachineLifecycleObservation::Missing | MachineLifecycleObservation::Decoded { .. } => {
Ok(())
}
MachineLifecycleObservation::Unsupported {
evidence_digest,
record_version,
..
} => Err(replacement_repair_blocked(
Some(evidence_digest.clone()),
format!(
"unsupported lifecycle record version {record_version} cannot prove fencing semantics"
),
)),
MachineLifecycleObservation::Malformed {
evidence_digest,
detail,
..
} => Err(replacement_repair_blocked(
Some(evidence_digest.clone()),
format!("malformed lifecycle evidence is not reclaimable: {detail}"),
)),
}
}
struct PreparedMachineLifecycleReplacement {
snapshot: MachineLifecycleSnapshot,
bytes: Vec<u8>,
version: MachineLifecycleObservationVersion,
}
impl PreparedMachineLifecycleReplacement {
fn preserve_observed_custody(
mut self,
current: &MachineLifecycleObservation,
) -> Result<Self, RuntimeStoreError> {
if let MachineLifecycleObservation::Decoded { record, .. } = current {
self.snapshot.supervisor_authority = record.supervisor_authority().clone();
self.snapshot.unregister_progress = record.unregister_progress().cloned();
self.bytes = MachineLifecycleStoreRecord::from_snapshot(&self.snapshot).encode()?;
self.version = MachineLifecycleObservationVersion::from_raw_record(&self.bytes);
}
Ok(self)
}
}
fn prepare_machine_lifecycle_replacement(
commit: MachineLifecycleCommit,
) -> Result<PreparedMachineLifecycleReplacement, RuntimeStoreError> {
let bytes = commit.store_record().encode()?;
let version = MachineLifecycleObservationVersion::from_raw_record(&bytes);
Ok(PreparedMachineLifecycleReplacement {
snapshot: commit.into_snapshot(),
bytes,
version,
})
}
fn decoded_prepared_machine_lifecycle_replacement(
replacement: &PreparedMachineLifecycleReplacement,
) -> Result<DecodedMachineLifecycleObservation, RuntimeStoreError> {
match classify_machine_lifecycle_record(&replacement.bytes) {
MachineLifecycleObservation::Decoded { record, .. } => Ok(record),
other => Err(RuntimeStoreError::Internal(format!(
"machine-authorized lifecycle replacement did not decode: {other:?}"
))),
}
}
pub async fn load_runtime_state(
store: &dyn RuntimeStore,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<RuntimeState>, RuntimeStoreError> {
Ok(load_machine_lifecycle(store, runtime_id)
.await?
.map(|snapshot| snapshot.runtime_state()))
}
pub(crate) async fn load_machine_lifecycle(
store: &dyn RuntimeStore,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<MachineLifecycleSnapshot>, RuntimeStoreError> {
store
.load_machine_lifecycle_record(runtime_id)
.await?
.map(|bytes| decode_machine_lifecycle_store_record(&bytes))
.transpose()
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MachineLifecycleStoreRecord {
snapshot: MachineLifecycleSnapshot,
}
impl MachineLifecycleStoreRecord {
pub(crate) fn from_snapshot(snapshot: &MachineLifecycleSnapshot) -> Self {
Self {
snapshot: snapshot.clone(),
}
}
#[must_use]
pub fn runtime_state(&self) -> RuntimeState {
self.snapshot.runtime_state()
}
pub fn encode(&self) -> Result<Vec<u8>, RuntimeStoreError> {
validate_supervisor_authority_snapshot(self.snapshot.supervisor_authority())
.map_err(|error| RuntimeStoreError::WriteFailed(error.to_string()))?;
validate_unregister_progress_snapshot(self.snapshot.unregister_progress())
.map_err(|error| RuntimeStoreError::WriteFailed(error.to_string()))?;
let wire = MachineLifecycleSnapshotStoreWire::from(&self.snapshot);
serde_json::to_vec(&wire).map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MachineLifecycleCommit {
snapshot: MachineLifecycleSnapshot,
expected_version: Option<MachineLifecycleExpectedVersion>,
}
impl MachineLifecycleCommit {
#[cfg(test)]
pub(crate) fn new_with_binding(
runtime_state: RuntimeState,
binding: MachineLifecycleBindingFacts,
supervisor_authority: SupervisorAuthoritySnapshot,
) -> Self {
Self::new_with_binding_and_unregister_progress(
runtime_state,
binding,
supervisor_authority,
None,
)
}
pub(crate) fn new_with_binding_and_unregister_progress(
runtime_state: RuntimeState,
binding: MachineLifecycleBindingFacts,
supervisor_authority: SupervisorAuthoritySnapshot,
unregister_progress: Option<MachineUnregisterProgressSnapshot>,
) -> Self {
Self::new_with_binding_run_and_unregister_progress(
runtime_state,
binding,
MachineLifecycleRunFacts::default(),
supervisor_authority,
unregister_progress,
)
}
pub(crate) fn new_with_binding_run_and_unregister_progress(
runtime_state: RuntimeState,
binding: MachineLifecycleBindingFacts,
run: MachineLifecycleRunFacts,
supervisor_authority: SupervisorAuthoritySnapshot,
unregister_progress: Option<MachineUnregisterProgressSnapshot>,
) -> Self {
Self {
snapshot: MachineLifecycleSnapshot::new_with_run_and_unregister_progress(
runtime_state,
binding,
run,
supervisor_authority,
unregister_progress,
),
expected_version: None,
}
}
pub(crate) fn with_expected_version(
mut self,
expected: MachineLifecycleExpectedVersion,
) -> Self {
self.expected_version = Some(expected);
self
}
pub fn runtime_state(&self) -> RuntimeState {
self.snapshot.runtime_state()
}
pub fn snapshot(&self) -> &MachineLifecycleSnapshot {
&self.snapshot
}
pub fn store_record(&self) -> MachineLifecycleStoreRecord {
MachineLifecycleStoreRecord::from_snapshot(&self.snapshot)
}
pub fn expected_version(&self) -> Option<&MachineLifecycleExpectedVersion> {
self.expected_version.as_ref()
}
pub(crate) fn into_snapshot(self) -> MachineLifecycleSnapshot {
self.snapshot
}
}
#[derive(Debug, Clone)]
pub struct UnregisterFinalizationCommit {
machine_lifecycle: MachineLifecycleCommit,
input_states: Vec<InputStatePersistenceRecord>,
retired_ops_epoch: meerkat_core::RuntimeEpochId,
}
impl UnregisterFinalizationCommit {
pub(crate) fn new(
machine_lifecycle: MachineLifecycleCommit,
input_states: Vec<InputStatePersistenceRecord>,
retired_ops_epoch: meerkat_core::RuntimeEpochId,
_authority: crate::meerkat_machine::DeleteOpsFinalizationAuthority,
) -> Self {
Self {
machine_lifecycle,
input_states,
retired_ops_epoch,
}
}
pub(crate) fn into_parts(
self,
) -> (
MachineLifecycleSnapshot,
Vec<InputStatePersistenceRecord>,
meerkat_core::RuntimeEpochId,
) {
(
self.machine_lifecycle.into_snapshot(),
self.input_states,
self.retired_ops_epoch,
)
}
pub fn lifecycle_store_record(&self) -> MachineLifecycleStoreRecord {
self.machine_lifecycle.store_record()
}
pub fn input_states(&self) -> &[InputStatePersistenceRecord] {
&self.input_states
}
pub fn retired_ops_epoch(&self) -> &meerkat_core::RuntimeEpochId {
&self.retired_ops_epoch
}
}
#[derive(Debug, Clone)]
pub enum InputStateRow {
Decoded(Box<StoredInputState>),
Corrupt {
input_id: String,
detail: String,
},
}
pub async fn load_input_states_for_recovery(
store: &dyn RuntimeStore,
runtime_id: &LogicalRuntimeId,
) -> Result<Vec<StoredInputState>, RuntimeStoreError> {
let mut states = Vec::new();
for row in store.load_input_states(runtime_id).await? {
match row {
InputStateRow::Decoded(state) => states.push(*state),
InputStateRow::Corrupt { input_id, detail } => {
tracing::error!(
runtime_id = %runtime_id.0,
input_id = %input_id,
detail = %detail,
"durable input row no longer decodes; recovering the runtime's remaining inputs without it"
);
}
}
}
Ok(states)
}
#[doc(hidden)]
#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)]
#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))]
pub trait RuntimeSessionAuthorityOps: Send + Sync {
fn session_persistence_profile(&self) -> RuntimeSessionPersistenceProfile;
fn session_boundary_authority_read_cost(&self) -> RuntimeSessionAuthorityReadCost;
async fn commit_prepared_session_boundary(
&self,
runtime_id: &LogicalRuntimeId,
request: PreparedRuntimeSessionCommit,
) -> Result<PreparedRuntimeSessionCommitResult, RuntimeStoreError>;
async fn load_session_boundary_authority(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<RuntimeSessionAuthority>, RuntimeStoreError>;
async fn load_whole_blob_store_authority(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<WholeBlobStoreAuthority>, RuntimeStoreError>;
async fn load_committed_whole_blob_snapshot(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<CommittedWholeBlobSnapshot>, RuntimeStoreError>;
async fn commit_prepared_whole_blob_snapshot_cas(
&self,
runtime_id: &LogicalRuntimeId,
prepared: PreparedWholeBlobSnapshotCas,
) -> Result<WholeBlobSnapshotCasOutcome, RuntimeStoreError>;
async fn delete_runtime_session_catalog_entry(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<(), RuntimeStoreError>;
async fn load_runtime_session_catalog_entry(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<RuntimeSessionCatalogEntry>, RuntimeStoreError>;
async fn list_runtime_session_catalog_entries(
&self,
filter: meerkat_core::SessionFilter,
) -> Result<Vec<RuntimeSessionCatalogEntry>, RuntimeStoreError>;
async fn write_prepared_whole_blob_provisional_tail(
&self,
runtime_id: &LogicalRuntimeId,
prepared: PreparedWholeBlobProvisionalTail,
) -> Result<WholeBlobProvisionalTailAuthority, RuntimeStoreError>;
async fn load_whole_blob_provisional_tail(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<CommittedWholeBlobProvisionalTail>, RuntimeStoreError>;
async fn discard_whole_blob_provisional_tail(
&self,
runtime_id: &LogicalRuntimeId,
expected: &WholeBlobProvisionalTailAuthority,
) -> Result<bool, RuntimeStoreError>;
async fn write_prepared_head_canonical_provisional_tail(
&self,
runtime_id: &LogicalRuntimeId,
prepared: PreparedHeadCanonicalProvisionalTail,
) -> Result<HeadCanonicalProvisionalTailAuthority, RuntimeStoreError>;
async fn load_head_canonical_provisional_tail(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<HeadCanonicalProvisionalTailAuthority>, RuntimeStoreError>;
async fn discard_head_canonical_provisional_tail(
&self,
runtime_id: &LogicalRuntimeId,
expected: &HeadCanonicalProvisionalTailAuthority,
) -> Result<bool, RuntimeStoreError>;
async fn load_durable_tail_recovery_source(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<PreparedDurableTailRecoverySource>, RuntimeStoreError>;
async fn load_durable_tail_recovery_receipts(
&self,
runtime_id: &LogicalRuntimeId,
run_id: &RunId,
) -> Result<Vec<PreparedRecoveryReceiptSource>, RuntimeStoreError>;
async fn load_committed_recovery_boundary(
&self,
runtime_id: &LogicalRuntimeId,
candidate_id: &str,
) -> Result<Option<CommittedRecoveryBoundary>, RuntimeStoreError>;
}
#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)]
#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))]
pub trait RuntimeStore: Send + Sync {
#[doc(hidden)]
fn session_authority_ops(&self) -> &dyn RuntimeSessionAuthorityOps;
fn session_persistence_profile(&self) -> RuntimeSessionPersistenceProfile {
self.session_authority_ops().session_persistence_profile()
}
fn session_boundary_authority_read_cost(&self) -> RuntimeSessionAuthorityReadCost {
self.session_authority_ops()
.session_boundary_authority_read_cost()
}
async fn commit_prepared_session_boundary(
&self,
runtime_id: &LogicalRuntimeId,
request: PreparedRuntimeSessionCommit,
) -> Result<PreparedRuntimeSessionCommitResult, RuntimeStoreError> {
self.session_authority_ops()
.commit_prepared_session_boundary(runtime_id, request)
.await
}
async fn load_session_boundary_authority(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<RuntimeSessionAuthority>, RuntimeStoreError> {
self.session_authority_ops()
.load_session_boundary_authority(runtime_id)
.await
}
async fn load_whole_blob_store_authority(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<WholeBlobStoreAuthority>, RuntimeStoreError> {
self.session_authority_ops()
.load_whole_blob_store_authority(runtime_id)
.await
}
async fn load_committed_whole_blob_snapshot(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<CommittedWholeBlobSnapshot>, RuntimeStoreError> {
self.session_authority_ops()
.load_committed_whole_blob_snapshot(runtime_id)
.await
}
async fn commit_prepared_whole_blob_snapshot_cas(
&self,
runtime_id: &LogicalRuntimeId,
prepared: PreparedWholeBlobSnapshotCas,
) -> Result<WholeBlobSnapshotCasOutcome, RuntimeStoreError> {
self.session_authority_ops()
.commit_prepared_whole_blob_snapshot_cas(runtime_id, prepared)
.await
}
async fn delete_runtime_session_catalog_entry(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<(), RuntimeStoreError> {
self.session_authority_ops()
.delete_runtime_session_catalog_entry(runtime_id)
.await
}
async fn load_runtime_session_catalog_entry(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<RuntimeSessionCatalogEntry>, RuntimeStoreError> {
self.session_authority_ops()
.load_runtime_session_catalog_entry(runtime_id)
.await
}
async fn list_runtime_session_catalog_entries(
&self,
filter: meerkat_core::SessionFilter,
) -> Result<Vec<RuntimeSessionCatalogEntry>, RuntimeStoreError> {
self.session_authority_ops()
.list_runtime_session_catalog_entries(filter)
.await
}
async fn write_prepared_whole_blob_provisional_tail(
&self,
runtime_id: &LogicalRuntimeId,
prepared: PreparedWholeBlobProvisionalTail,
) -> Result<WholeBlobProvisionalTailAuthority, RuntimeStoreError> {
self.session_authority_ops()
.write_prepared_whole_blob_provisional_tail(runtime_id, prepared)
.await
}
async fn load_whole_blob_provisional_tail(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<CommittedWholeBlobProvisionalTail>, RuntimeStoreError> {
self.session_authority_ops()
.load_whole_blob_provisional_tail(runtime_id)
.await
}
async fn discard_whole_blob_provisional_tail(
&self,
runtime_id: &LogicalRuntimeId,
expected: &WholeBlobProvisionalTailAuthority,
) -> Result<bool, RuntimeStoreError> {
self.session_authority_ops()
.discard_whole_blob_provisional_tail(runtime_id, expected)
.await
}
async fn write_prepared_head_canonical_provisional_tail(
&self,
runtime_id: &LogicalRuntimeId,
prepared: PreparedHeadCanonicalProvisionalTail,
) -> Result<HeadCanonicalProvisionalTailAuthority, RuntimeStoreError> {
self.session_authority_ops()
.write_prepared_head_canonical_provisional_tail(runtime_id, prepared)
.await
}
async fn load_head_canonical_provisional_tail(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<HeadCanonicalProvisionalTailAuthority>, RuntimeStoreError> {
self.session_authority_ops()
.load_head_canonical_provisional_tail(runtime_id)
.await
}
async fn discard_head_canonical_provisional_tail(
&self,
runtime_id: &LogicalRuntimeId,
expected: &HeadCanonicalProvisionalTailAuthority,
) -> Result<bool, RuntimeStoreError> {
self.session_authority_ops()
.discard_head_canonical_provisional_tail(runtime_id, expected)
.await
}
async fn load_durable_tail_recovery_source(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<PreparedDurableTailRecoverySource>, RuntimeStoreError> {
self.session_authority_ops()
.load_durable_tail_recovery_source(runtime_id)
.await
}
async fn load_durable_tail_recovery_receipts(
&self,
runtime_id: &LogicalRuntimeId,
run_id: &RunId,
) -> Result<Vec<PreparedRecoveryReceiptSource>, RuntimeStoreError> {
self.session_authority_ops()
.load_durable_tail_recovery_receipts(runtime_id, run_id)
.await
}
async fn load_committed_recovery_boundary(
&self,
runtime_id: &LogicalRuntimeId,
candidate_id: &str,
) -> Result<Option<CommittedRecoveryBoundary>, RuntimeStoreError> {
self.session_authority_ops()
.load_committed_recovery_boundary(runtime_id, candidate_id)
.await
}
fn supports_compaction_projection_outbox(&self) -> bool {
false
}
fn auth_authority_key(&self) -> Option<String> {
None
}
async fn load_runtime_delivery_authority(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<RuntimeDeliveryAuthorityRecord>, RuntimeStoreError> {
let _ = runtime_id;
Err(RuntimeStoreError::Unsupported(
"load_runtime_delivery_authority".into(),
))
}
async fn load_runtime_delivery_record(
&self,
runtime_id: &LogicalRuntimeId,
delivery_id: &str,
) -> Result<Option<RuntimeDeliveryStoreRecord>, RuntimeStoreError> {
let _ = (runtime_id, delivery_id);
Err(RuntimeStoreError::Unsupported(
"load_runtime_delivery_record".into(),
))
}
async fn compare_and_swap_runtime_delivery_authority(
&self,
runtime_id: &LogicalRuntimeId,
expected_revision: Option<u64>,
replacement: RuntimeDeliveryAuthorityRecord,
inserted_delivery: Option<RuntimeDeliveryStoreRecord>,
) -> Result<RuntimeDeliveryAuthorityCasOutcome, RuntimeStoreError> {
let _ = (
runtime_id,
expected_revision,
replacement,
inserted_delivery,
);
Err(RuntimeStoreError::Unsupported(
"compare_and_swap_runtime_delivery_authority".into(),
))
}
async fn list_runtime_delivery_records(
&self,
runtime_id: &LogicalRuntimeId,
after_sequence: u64,
limit: usize,
) -> Result<Vec<RuntimeDeliveryStoreRecord>, RuntimeStoreError> {
let _ = (runtime_id, after_sequence, limit);
Err(RuntimeStoreError::Unsupported(
"list_runtime_delivery_records".into(),
))
}
fn persist_auth_oauth_flow_snapshot(
&self,
snapshot_json: &[u8],
) -> Result<(), RuntimeStoreError> {
let _ = snapshot_json;
Err(RuntimeStoreError::Unsupported(
"persist_auth_oauth_flow_snapshot".into(),
))
}
fn load_auth_oauth_flow_snapshot(&self) -> Result<Option<Vec<u8>>, RuntimeStoreError> {
Err(RuntimeStoreError::Unsupported(
"load_auth_oauth_flow_snapshot".into(),
))
}
fn update_auth_oauth_flow_snapshot(
&self,
_update: &mut AuthOAuthFlowSnapshotUpdate<'_>,
) -> Result<(), RuntimeStoreError> {
Err(RuntimeStoreError::Unsupported(
"update_auth_oauth_flow_snapshot".into(),
))
}
async fn commit_session_snapshot(
&self,
runtime_id: &LogicalRuntimeId,
session_delta: SerializedSessionSnapshot,
) -> Result<(), RuntimeStoreError>;
async fn commit_prepared_whole_blob_rewrite_boundary(
&self,
runtime_id: &LogicalRuntimeId,
boundary: PreparedWholeBlobRewriteStoreParts,
) -> Result<WholeBlobStoreAuthority, RuntimeStoreError>;
async fn atomic_apply(
&self,
runtime_id: &LogicalRuntimeId,
session_delta: Option<SerializedSessionSnapshot>,
receipt: RunBoundaryReceipt,
input_updates: Vec<InputStatePersistenceRecord>,
session_store_key: Option<meerkat_core::types::SessionId>,
) -> Result<(), RuntimeStoreError>;
async fn load_pending_compaction_projections(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Vec<meerkat_core::CompactionProjectionIntent>, RuntimeStoreError> {
let _ = runtime_id;
Err(RuntimeStoreError::Unsupported(
"load_pending_compaction_projections".to_string(),
))
}
async fn mark_compaction_projection_finalized(
&self,
runtime_id: &LogicalRuntimeId,
projection: &meerkat_core::CompactionProjectionId,
) -> Result<(), RuntimeStoreError> {
let _ = (runtime_id, projection);
Err(RuntimeStoreError::Unsupported(
"mark_compaction_projection_finalized".to_string(),
))
}
async fn atomic_apply_with_machine_lifecycle(
&self,
runtime_id: &LogicalRuntimeId,
session_delta: SerializedSessionSnapshot,
receipt: RunBoundaryReceipt,
machine_lifecycle: MachineLifecycleCommit,
input_updates: Vec<InputStatePersistenceRecord>,
session_store_key: meerkat_core::types::SessionId,
) -> Result<(), RuntimeStoreError> {
let _ = (
runtime_id,
session_delta,
receipt,
machine_lifecycle,
input_updates,
session_store_key,
);
Err(RuntimeStoreError::Unsupported(
"atomic_apply_with_machine_lifecycle".to_string(),
))
}
async fn load_input_states(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Vec<InputStateRow>, RuntimeStoreError>;
async fn load_input_states_strict(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Vec<StoredInputState>, RuntimeStoreError> {
let mut states = Vec::new();
for row in self.load_input_states(runtime_id).await? {
match row {
InputStateRow::Decoded(state) => states.push(*state),
InputStateRow::Corrupt { input_id, detail } => {
return Err(RuntimeStoreError::ReadFailed(format!(
"input state row `{input_id}` failed to decode: {detail}"
)));
}
}
}
Ok(states)
}
async fn load_boundary_receipt(
&self,
runtime_id: &LogicalRuntimeId,
run_id: &RunId,
sequence: u64,
) -> Result<Option<RunBoundaryReceipt>, RuntimeStoreError>;
async fn load_committed_boundary_receipts(
&self,
runtime_id: &LogicalRuntimeId,
run_id: &RunId,
) -> Result<Vec<RunBoundaryReceipt>, RuntimeStoreError> {
const PROBE_CAP: u64 = 100_000;
let mut receipts = Vec::new();
for sequence in 1..=PROBE_CAP {
match self
.load_boundary_receipt(runtime_id, run_id, sequence)
.await?
{
Some(receipt) => receipts.push(receipt),
None => return Ok(receipts),
}
}
Err(RuntimeStoreError::ReadFailed(format!(
"run {run_id} has more than {PROBE_CAP} boundary receipts; refusing to probe further"
)))
}
async fn load_input_states_with_versions(
&self,
_runtime_id: &LogicalRuntimeId,
) -> Result<PreparedRecoveryInputSnapshot, RuntimeStoreError> {
Err(RuntimeStoreError::Unsupported(
"load_input_states_with_versions requires exact stored-row and complete-set tokens"
.to_string(),
))
}
async fn load_session_snapshot(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<std::sync::Arc<Vec<u8>>>, RuntimeStoreError>;
async fn clear_session_snapshot(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<(), RuntimeStoreError>;
async fn replace_session_snapshot_if_current(
&self,
runtime_id: &LogicalRuntimeId,
expected_current: &[u8],
replacement: Vec<u8>,
) -> Result<bool, RuntimeStoreError>;
async fn clear_session_snapshot_if_current(
&self,
runtime_id: &LogicalRuntimeId,
expected_current: &[u8],
) -> Result<bool, RuntimeStoreError>;
async fn is_runtime_projection_quarantined(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<bool, RuntimeStoreError> {
let _ = runtime_id;
Ok(false)
}
async fn persist_input_state(
&self,
runtime_id: &LogicalRuntimeId,
state: &InputStatePersistenceRecord,
) -> Result<(), RuntimeStoreError>;
async fn persist_input_states_atomically(
&self,
_runtime_id: &LogicalRuntimeId,
states: &[InputStatePersistenceRecord],
) -> Result<(), RuntimeStoreError> {
if states.is_empty() {
return Ok(());
}
Err(RuntimeStoreError::Unsupported(
"persist_input_states_atomically".to_string(),
))
}
fn input_state_batch_cas_implementation_profile(
&self,
) -> InputStateBatchCasImplementationProfile {
InputStateBatchCasImplementationProfile::Unsupported
}
async fn compare_and_swap_input_states_atomically(
&self,
_runtime_id: &LogicalRuntimeId,
expected: &[StoredInputState],
replacements: &[InputStatePersistenceRecord],
) -> Result<InputStateBatchCasOutcome, RuntimeStoreError> {
let prepared = prepare_input_state_batch_cas(expected, replacements)?;
if prepared.is_empty() {
return Ok(InputStateBatchCasOutcome::Swapped);
}
Err(RuntimeStoreError::Unsupported(
"compare_and_swap_input_states_atomically".to_string(),
))
}
async fn compare_and_swap_input_states_atomically_with_fence(
&self,
runtime_id: &LogicalRuntimeId,
expected: &[StoredInputState],
replacements: &[InputStatePersistenceRecord],
write_fence: std::sync::Arc<dyn RuntimeStoreWriteFence>,
) -> Result<FencedInputStateBatchCasOutcome, RuntimeStoreError> {
let prepared = prepare_input_state_batch_cas(expected, replacements)?;
if prepared.is_empty() {
return Ok(FencedInputStateBatchCasOutcome::Swapped);
}
let _ = (runtime_id, write_fence);
Err(RuntimeStoreError::Unsupported(
"compare_and_swap_input_states_atomically_with_fence".to_string(),
))
}
async fn compare_and_swap_recovery_input_states_atomically(
&self,
runtime_id: &LogicalRuntimeId,
expected_revision: RecoveryInputSetRevision,
mutations: &[RecoveryInputStateMutation],
) -> Result<InputStateBatchCasOutcome, RuntimeStoreError> {
let _ = prepare_recovery_input_state_mutations(mutations)?;
let _ = (runtime_id, expected_revision);
Err(RuntimeStoreError::Unsupported(
"compare_and_swap_recovery_input_states_atomically".to_string(),
))
}
async fn compare_and_swap_recovery_input_states_atomically_with_fence(
&self,
runtime_id: &LogicalRuntimeId,
expected_revision: RecoveryInputSetRevision,
mutations: &[RecoveryInputStateMutation],
write_fence: std::sync::Arc<dyn RuntimeStoreWriteFence>,
) -> Result<FencedInputStateBatchCasOutcome, RuntimeStoreError> {
let _ = prepare_recovery_input_state_mutations(mutations)?;
let _ = (runtime_id, expected_revision, write_fence);
Err(RuntimeStoreError::Unsupported(
"compare_and_swap_recovery_input_states_atomically_with_fence".to_string(),
))
}
async fn load_input_state(
&self,
runtime_id: &LogicalRuntimeId,
input_id: &InputId,
) -> Result<Option<StoredInputState>, RuntimeStoreError>;
async fn load_input_state_by_idempotency_key(
&self,
_runtime_id: &LogicalRuntimeId,
_key: &IdempotencyKey,
) -> Result<Option<ExactInputStateObservation>, RuntimeStoreError> {
Err(RuntimeStoreError::Unsupported(
"load_input_state_by_idempotency_key requires an exact maintained index".to_string(),
))
}
async fn load_input_states_by_ids(
&self,
_runtime_id: &LogicalRuntimeId,
input_ids: &[InputId],
) -> Result<Vec<Option<StoredInputState>>, RuntimeStoreError> {
validate_input_state_batch_read_ids(input_ids)?;
if input_ids.is_empty() {
return Ok(Vec::new());
}
Err(RuntimeStoreError::Unsupported(
"load_input_states_by_ids".to_string(),
))
}
async fn load_pending_terminal_owner_ids_page(
&self,
_runtime_id: &LogicalRuntimeId,
after: Option<&InputId>,
limit: usize,
) -> Result<Vec<InputId>, RuntimeStoreError> {
validate_pending_terminal_owner_page(after, limit, &[])?;
Err(RuntimeStoreError::Unsupported(
"load_pending_terminal_owner_ids_page".to_string(),
))
}
async fn observe_machine_lifecycle(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<MachineLifecycleObservation, RuntimeStoreError> {
let _ = runtime_id;
Err(RuntimeStoreError::Unsupported(
"observe_machine_lifecycle".to_string(),
))
}
async fn compare_and_swap_machine_lifecycle(
&self,
runtime_id: &LogicalRuntimeId,
expected: MachineLifecycleExpectedVersion,
replacement: MachineLifecycleCommit,
) -> Result<MachineLifecycleCasOutcome, RuntimeStoreError> {
let _ = (runtime_id, expected, replacement);
Err(RuntimeStoreError::Unsupported(
"compare_and_swap_machine_lifecycle".to_string(),
))
}
async fn compare_and_swap_machine_lifecycle_with_fence(
&self,
runtime_id: &LogicalRuntimeId,
expected: MachineLifecycleExpectedVersion,
replacement: MachineLifecycleCommit,
write_fence: std::sync::Arc<dyn RuntimeStoreWriteFence>,
) -> Result<FencedMachineLifecycleCasOutcome, RuntimeStoreError> {
let _ = (runtime_id, expected, replacement, write_fence);
Err(RuntimeStoreError::Unsupported(
"compare_and_swap_machine_lifecycle_with_fence".to_string(),
))
}
async fn load_machine_lifecycle_record(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<Vec<u8>>, RuntimeStoreError>;
async fn commit_machine_lifecycle(
&self,
runtime_id: &LogicalRuntimeId,
commit: MachineLifecycleCommit,
input_states: &[InputStatePersistenceRecord],
) -> Result<(), RuntimeStoreError>;
async fn commit_unregister_finalization(
&self,
runtime_id: &LogicalRuntimeId,
finalization: UnregisterFinalizationCommit,
) -> Result<(), RuntimeStoreError> {
let _ = (runtime_id, finalization);
Err(RuntimeStoreError::Unsupported(
"commit_unregister_finalization".into(),
))
}
async fn initialize_ops_lifecycle_if_absent(
&self,
runtime_id: &LogicalRuntimeId,
candidate: &crate::ops_lifecycle::PersistedOpsSnapshot,
) -> Result<crate::ops_lifecycle::PersistedOpsSnapshot, RuntimeStoreError> {
let _ = (runtime_id, candidate);
Err(RuntimeStoreError::Unsupported(
"initialize_ops_lifecycle_if_absent".into(),
))
}
async fn persist_ops_lifecycle(
&self,
runtime_id: &LogicalRuntimeId,
snapshot: &crate::ops_lifecycle::PersistedOpsSnapshot,
) -> Result<(), RuntimeStoreError> {
let _ = (runtime_id, snapshot);
Err(RuntimeStoreError::Unsupported(
"persist_ops_lifecycle".into(),
))
}
async fn load_ops_lifecycle(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<crate::ops_lifecycle::PersistedOpsSnapshot>, RuntimeStoreError> {
let _ = runtime_id;
Err(RuntimeStoreError::Unsupported("load_ops_lifecycle".into()))
}
async fn delete_ops_lifecycle(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<(), RuntimeStoreError> {
let _ = runtime_id;
Err(RuntimeStoreError::Unsupported(
"delete_ops_lifecycle".into(),
))
}
async fn load_mob_host_binding(
&self,
mob_id: &str,
) -> Result<Option<Vec<u8>>, RuntimeStoreError> {
let _ = mob_id;
Err(RuntimeStoreError::Unsupported(
"load_mob_host_binding".into(),
))
}
async fn list_mob_host_bindings(&self) -> Result<Vec<(String, Vec<u8>)>, RuntimeStoreError> {
Err(RuntimeStoreError::Unsupported(
"list_mob_host_bindings".into(),
))
}
async fn put_mob_host_binding_if_absent(
&self,
mob_id: &str,
record_json: &[u8],
) -> Result<bool, RuntimeStoreError> {
let _ = (mob_id, record_json);
Err(RuntimeStoreError::Unsupported(
"put_mob_host_binding_if_absent".into(),
))
}
async fn compare_and_put_mob_host_binding(
&self,
mob_id: &str,
expected_json: &[u8],
next_json: &[u8],
) -> Result<bool, RuntimeStoreError> {
let _ = (mob_id, expected_json, next_json);
Err(RuntimeStoreError::Unsupported(
"compare_and_put_mob_host_binding".into(),
))
}
async fn delete_mob_host_binding(
&self,
mob_id: &str,
expected_json: &[u8],
) -> Result<bool, RuntimeStoreError> {
let _ = (mob_id, expected_json);
Err(RuntimeStoreError::Unsupported(
"delete_mob_host_binding".into(),
))
}
async fn load_mob_host_revocation(
&self,
mob_id: &str,
) -> Result<Option<Vec<u8>>, RuntimeStoreError> {
let _ = mob_id;
Err(RuntimeStoreError::Unsupported(
"load_mob_host_revocation".into(),
))
}
async fn list_mob_host_revocations(&self) -> Result<Vec<(String, Vec<u8>)>, RuntimeStoreError> {
Err(RuntimeStoreError::Unsupported(
"list_mob_host_revocations".into(),
))
}
async fn revoke_mob_host_binding(
&self,
mob_id: &str,
expected_binding_json: &[u8],
receipt_json: &[u8],
) -> Result<bool, RuntimeStoreError> {
let _ = (mob_id, expected_binding_json, receipt_json);
Err(RuntimeStoreError::Unsupported(
"revoke_mob_host_binding".into(),
))
}
}
pub use memory::InMemoryRuntimeStore;
#[cfg(feature = "sqlite-store")]
pub use sqlite::SqliteRuntimeStore;
#[cfg(test)]
mod store_authority_record_tests {
use super::*;
use meerkat_core::session_store::PreparedHeadCanonicalMutation;
use meerkat_core::types::{Message, UserMessage};
fn row_digest(byte: char) -> String {
format!("row-sha256:{}", byte.to_string().repeat(64))
}
fn head_record() -> (
meerkat_core::types::SessionId,
meerkat_core::session_store::SessionHead,
String,
) {
let mut session = meerkat_core::Session::new();
session.push(Message::User(UserMessage::text("canonical head")));
let session_id = session.id().clone();
let mutation =
PreparedHeadCanonicalMutation::prepare(&session, None).expect("prepare canonical head");
(
session_id,
mutation.successor_head().clone(),
mutation.successor_head_token().to_string(),
)
}
#[test]
fn borrowed_whole_blob_provisional_prepare_encodes_once_and_retains_no_session() {
let mut session = meerkat_core::Session::new();
session.push(Message::User(UserMessage::text("candidate")));
let session_id = session.id().clone();
let base = WholeBlobStoreAuthority::issued(session_id.clone(), 7, row_digest('b'))
.expect("valid base authority");
let prepared =
PreparedWholeBlobProvisionalTail::prepare_from_session(base, RunId::new(), 1, &session)
.expect("prepare borrowed WholeBlob candidate");
assert_eq!(
prepared.whole_blob_encode_count(),
1,
"borrowed preparation must stream the Session exactly once"
);
let retained = prepared.clone();
drop(prepared);
drop(session);
assert_eq!(
retained.whole_blob_encode_count(),
1,
"cloning the bounded carrier must share bytes, not re-encode"
);
let (authority, artifact, digest, message_count, catalog, intents) = retained.into_parts();
assert_eq!(authority.session_id(), &session_id);
assert_eq!(
authority.candidate_blob_sha256(),
artifact.row_sha256_token()
);
assert_eq!(catalog.session_id(), &session_id);
assert_eq!(catalog.message_count(), 1);
assert_eq!(message_count, 1);
assert!(!digest.is_empty());
assert!(intents.is_empty());
let decoded = meerkat_core::Session::from_persisted_bytes(artifact.bytes())
.expect("carrier bytes remain independently usable after the Session is dropped");
assert_eq!(decoded.id(), &session_id);
assert_eq!(decoded.messages().len(), 1);
}
#[test]
fn committed_whole_blob_decode_installs_store_owned_rewrite_lineage() {
let mut session = meerkat_core::Session::new();
session.push(Message::User(UserMessage::text("original")));
let artifact = session
.to_persisted_artifact()
.expect("serialize WholeBlob document");
let authority = WholeBlobStoreAuthority::issued(
session.id().clone(),
1,
artifact.row_sha256_token().to_string(),
)
.expect("issue exact WholeBlob authority");
let committed = CommittedWholeBlobSnapshot::new(artifact.bytes_arc(), authority)
.expect("decode store-owned WholeBlob document");
let mut decoded = committed.session().clone();
let parent_revision = decoded.transcript_revision().expect("read parent revision");
decoded
.commit_transcript_rewrite(
meerkat_core::TranscriptRewriteSelection::MessageRange { start: 0, end: 1 },
vec![Message::User(UserMessage::text("edited"))],
meerkat_core::TranscriptRewriteReason::new("test"),
Some("runtime-store-test".to_string()),
Some(parent_revision),
)
.expect("store-owned WholeBlob decode must carry exact rewrite lineage");
}
#[test]
fn whole_blob_store_record_constructor_validates_every_fixed_field() {
let session_id = meerkat_core::types::SessionId::new();
let valid = WholeBlobStoreAuthority::from_store_record(
WholeBlobStoreAuthority::VERSION,
session_id.clone(),
7,
row_digest('a'),
)
.expect("valid WholeBlob record");
assert_eq!(valid.authority_version(), WholeBlobStoreAuthority::VERSION);
assert_eq!(valid.session_id(), &session_id);
assert_eq!(valid.store_revision(), 7);
assert_eq!(valid.blob_sha256(), row_digest('a'));
for (version, revision, digest) in [
(WholeBlobStoreAuthority::VERSION + 1, 7, row_digest('a')),
(WholeBlobStoreAuthority::VERSION, 0, row_digest('a')),
(WholeBlobStoreAuthority::VERSION, 7, String::new()),
(
WholeBlobStoreAuthority::VERSION,
7,
format!("sha256:{}", "a".repeat(64)),
),
(WholeBlobStoreAuthority::VERSION, 7, row_digest('A')),
(
WholeBlobStoreAuthority::VERSION,
7,
format!("row-sha256:{}", "a".repeat(63)),
),
] {
assert!(
WholeBlobStoreAuthority::from_store_record(
version,
session_id.clone(),
revision,
digest,
)
.is_err()
);
}
}
#[test]
fn head_canonical_store_record_constructor_validates_every_bound_fact() {
let (session_id, head, token) = head_record();
let valid = HeadCanonicalStoreAuthority::from_store_record(
HeadCanonicalStoreAuthority::VERSION,
session_id.clone(),
11,
head.clone(),
token.clone(),
)
.expect("valid HeadCanonical record");
assert_eq!(
valid.authority_version(),
HeadCanonicalStoreAuthority::VERSION
);
assert_eq!(valid.session_id(), &session_id);
assert_eq!(valid.store_revision(), 11);
assert_eq!(valid.boundary_head(), &head);
assert_eq!(valid.committed_head_token(), token);
assert!(
HeadCanonicalStoreAuthority::from_store_record(
HeadCanonicalStoreAuthority::VERSION + 1,
session_id.clone(),
11,
head.clone(),
token.clone(),
)
.is_err()
);
assert!(
HeadCanonicalStoreAuthority::from_store_record(
HeadCanonicalStoreAuthority::VERSION,
session_id.clone(),
0,
head.clone(),
token.clone(),
)
.is_err()
);
assert!(
HeadCanonicalStoreAuthority::from_store_record(
HeadCanonicalStoreAuthority::VERSION,
session_id.clone(),
11,
head.clone(),
String::new(),
)
.is_err()
);
assert!(
HeadCanonicalStoreAuthority::from_store_record(
HeadCanonicalStoreAuthority::VERSION,
session_id.clone(),
11,
head.clone(),
"head-cas:different".to_string(),
)
.is_err()
);
let mut wrong_session = head.clone();
wrong_session.id = meerkat_core::types::SessionId::new();
assert!(
HeadCanonicalStoreAuthority::from_store_record(
HeadCanonicalStoreAuthority::VERSION,
session_id.clone(),
11,
wrong_session,
token.clone(),
)
.is_err()
);
let mut missing_row_prefix = head.clone();
missing_row_prefix.message_row_prefix = None;
assert!(
HeadCanonicalStoreAuthority::from_store_record(
HeadCanonicalStoreAuthority::VERSION,
session_id.clone(),
11,
missing_row_prefix,
token.clone(),
)
.is_err()
);
let mut wrong_row_count = head.clone();
wrong_row_count.message_count = wrong_row_count.message_count.saturating_add(1);
assert!(
HeadCanonicalStoreAuthority::from_store_record(
HeadCanonicalStoreAuthority::VERSION,
session_id.clone(),
11,
wrong_row_count,
token.clone(),
)
.is_err()
);
let mut wrong_rewrite_count = head;
wrong_rewrite_count.rewrite_count = wrong_rewrite_count.rewrite_count.saturating_add(1);
assert!(
HeadCanonicalStoreAuthority::from_store_record(
HeadCanonicalStoreAuthority::VERSION,
session_id,
11,
wrong_rewrite_count,
token,
)
.is_err()
);
}
}
#[cfg(test)]
mod runtime_session_catalog_entry_tests {
use super::*;
use meerkat_core::session_store::PreparedHeadCanonicalMutation;
use meerkat_core::types::{Message, UserMessage};
fn labeled_session() -> meerkat_core::Session {
let mut session = meerkat_core::Session::new();
session.push(Message::User(UserMessage::text(
"transcript body must not enter the catalog",
)));
session.set_metadata(
RuntimeSessionCatalogEntry::SESSION_LABELS_KEY,
serde_json::json!({
"owner": "operations",
"tier": "production"
}),
);
session
}
#[test]
fn public_session_projection_is_validated_and_body_free() {
let session = labeled_session();
let entry = RuntimeSessionCatalogEntry::from_session(
&session,
RuntimeSessionPersistenceProfile::WholeBlobV1,
Some(RuntimeState::Idle),
)
.expect("typed Session projects to bounded catalog metadata");
assert_eq!(entry.session_id(), session.id());
assert_eq!(
entry.persistence_profile(),
RuntimeSessionPersistenceProfile::WholeBlobV1
);
assert_eq!(entry.created_at(), session.created_at());
assert_eq!(entry.updated_at(), session.updated_at());
assert_eq!(entry.message_count(), session.messages().len());
assert_eq!(entry.total_tokens(), session.total_tokens());
assert_eq!(
entry.labels(),
&BTreeMap::from([
("owner".to_string(), "operations".to_string()),
("tier".to_string(), "production".to_string()),
])
);
assert_eq!(entry.runtime_state(), Some(RuntimeState::Idle));
let encoded = serde_json::to_string(&entry).expect("catalog entry serializes");
assert!(
!encoded.contains("transcript body must not enter the catalog"),
"catalog projection must never carry transcript body data"
);
}
#[test]
fn public_head_projection_matches_session_catalog_facts() {
let session = labeled_session();
let mutation =
PreparedHeadCanonicalMutation::prepare(&session, None).expect("prepare canonical head");
let from_session = RuntimeSessionCatalogEntry::from_session(
&session,
RuntimeSessionPersistenceProfile::HeadCanonicalV1,
None,
)
.expect("Session catalog projection");
let from_head = RuntimeSessionCatalogEntry::from_head(
mutation.successor_head(),
RuntimeSessionPersistenceProfile::HeadCanonicalV1,
None,
)
.expect("SessionHead catalog projection");
assert_eq!(from_head, from_session);
}
#[test]
fn public_catalog_projections_reject_malformed_labels() {
let mut session = labeled_session();
session.set_metadata(
RuntimeSessionCatalogEntry::SESSION_LABELS_KEY,
serde_json::json!(["not", "a", "label", "map"]),
);
assert!(matches!(
RuntimeSessionCatalogEntry::from_session(
&session,
RuntimeSessionPersistenceProfile::WholeBlobV1,
None,
),
Err(RuntimeStoreError::WriteFailed(detail))
if detail.contains("malformed catalog labels")
));
let mutation =
PreparedHeadCanonicalMutation::prepare(&session, None).expect("prepare canonical head");
assert!(matches!(
RuntimeSessionCatalogEntry::from_head(
mutation.successor_head(),
RuntimeSessionPersistenceProfile::HeadCanonicalV1,
None,
),
Err(RuntimeStoreError::WriteFailed(detail))
if detail.contains("malformed catalog labels")
));
}
}
#[cfg(test)]
mod lifecycle_record_compatibility_tests {
use super::*;
fn operation_id(
value: u128,
) -> meerkat_contracts::wire::supervisor_bridge::SupervisorRotationOperationId {
meerkat_contracts::wire::supervisor_bridge::SupervisorRotationOperationId::from_uuid(
uuid::Uuid::from_u128(value),
)
}
fn binding(seed: u8, name: &str, epoch: u64) -> SupervisorBindingReceipt {
let pubkey = [seed; 32];
SupervisorBindingReceipt::new(
name.to_string(),
meerkat_core::comms::PeerId::from_ed25519_pubkey(&pubkey).as_str(),
format!("inproc://{name}"),
crate::comms_drain::encode_supervisor_signing_public_key(pubkey),
epoch,
)
}
fn rotation(
operation_id: meerkat_contracts::wire::supervisor_bridge::SupervisorRotationOperationId,
phase: SupervisorRotationPersistencePhase,
rejection: Option<SupervisorRotationRejection>,
previous: SupervisorBindingReceipt,
next: SupervisorBindingReceipt,
) -> SupervisorRotationReceipt {
SupervisorRotationReceipt::new(operation_id, phase, rejection, previous, next)
}
fn snapshot(authority: SupervisorAuthoritySnapshot) -> MachineLifecycleSnapshot {
MachineLifecycleSnapshot::new(
RuntimeState::Idle,
MachineLifecycleBindingFacts::new(None, None, None, None),
authority,
)
}
fn encode_snapshot(snapshot: &MachineLifecycleSnapshot) -> Vec<u8> {
MachineLifecycleStoreRecord::from_snapshot(snapshot)
.encode()
.expect("encode lifecycle snapshot")
}
fn encode_unvalidated_snapshot(snapshot: &MachineLifecycleSnapshot) -> Vec<u8> {
serde_json::to_vec(&MachineLifecycleSnapshotStoreWire::from(snapshot))
.expect("serialize deliberately corrupt lifecycle snapshot")
}
fn encoded_value(snapshot: &MachineLifecycleSnapshot) -> serde_json::Value {
serde_json::from_slice(&encode_snapshot(snapshot)).expect("decode encoded snapshot as JSON")
}
fn assert_decode_fails(value: serde_json::Value) {
let bytes = serde_json::to_vec(&value).expect("serialize corrupt lifecycle record");
assert!(
decode_machine_lifecycle_store_record(&bytes).is_err(),
"corrupt lifecycle record must fail closed: {value}"
);
}
#[test]
fn version_one_record_without_supervisor_authority_migrates_explicitly_to_unbound() {
let bytes = serde_json::to_vec(&serde_json::json!({
"record_version": LEGACY_MACHINE_LIFECYCLE_STORE_RECORD_VERSION,
"runtime_state": RuntimeState::Retired,
"binding": {
"agent_runtime_id": "rt:session:legacy-v1",
"fence_token": 19,
"runtime_generation": 4,
"runtime_epoch_id": "epoch-legacy-v1"
}
}))
.expect("serialize legacy v1 lifecycle record");
let decoded = decode_machine_lifecycle_store_record(&bytes)
.expect("valid v1 record without the additive field must decode");
assert_eq!(decoded.runtime_state(), RuntimeState::Retired);
assert_eq!(
decoded.supervisor_authority(),
&SupervisorAuthoritySnapshot::UnboundNoReceipt
);
}
#[test]
fn current_record_requires_supervisor_authority_and_unregister_progress_presence() {
assert_decode_fails(serde_json::json!({
"record_version": MACHINE_LIFECYCLE_STORE_RECORD_VERSION,
"runtime_state": RuntimeState::Idle,
"binding": {
"agent_runtime_id": null,
"fence_token": null,
"runtime_generation": null,
"runtime_epoch_id": null
},
"unregister_progress": null
}));
}
#[test]
fn current_nullable_fields_require_presence_but_accept_explicit_null() {
let unbound = snapshot(SupervisorAuthoritySnapshot::UnboundNoReceipt);
let encoded = encoded_value(&unbound);
assert_eq!(
decode_machine_lifecycle_store_record(
&serde_json::to_vec(&encoded).expect("serialize valid current record")
)
.expect("explicit-null current binding fields must decode"),
unbound
);
let mut missing_progress = encoded.clone();
missing_progress
.as_object_mut()
.expect("lifecycle record object")
.remove("unregister_progress");
assert_decode_fails(missing_progress);
for field in [
"agent_runtime_id",
"fence_token",
"runtime_generation",
"runtime_epoch_id",
] {
let mut partial = encoded.clone();
partial["binding"]
.as_object_mut()
.expect("binding object")
.remove(field);
assert_decode_fails(partial);
}
for field in ["current_run_id", "pre_run_phase"] {
let mut partial = encoded.clone();
partial
.as_object_mut()
.expect("lifecycle record object")
.remove(field);
assert_decode_fails(partial);
}
let completed = snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
operation_id(101),
SupervisorRotationPersistencePhase::Completed,
None,
binding(30, "required-null-previous", 4),
binding(31, "required-null-next", 5),
)));
let mut missing_rejection = encoded_value(&completed);
assert!(missing_rejection["supervisor_authority"]["rotation"]["rejection"].is_null());
missing_rejection["supervisor_authority"]["rotation"]
.as_object_mut()
.expect("rotation object")
.remove("rejection");
assert_decode_fails(missing_rejection);
}
#[test]
fn lossless_observation_preserves_partial_run_pair_and_nullable_lifecycle() {
let mut value = encoded_value(&snapshot(SupervisorAuthoritySnapshot::UnboundNoReceipt));
let run_id = RunId::new();
value["runtime_state"] = serde_json::Value::Null;
value["current_run_id"] = serde_json::to_value(&run_id).expect("serialize run id");
value["pre_run_phase"] = serde_json::Value::Null;
let bytes = serde_json::to_vec(&value).expect("serialize partial lifecycle row");
let MachineLifecycleObservation::Decoded { record, version } =
classify_machine_lifecycle_record(&bytes)
else {
panic!("explicitly nullable partial runtime tuple must remain decoded");
};
assert_eq!(
record.record_version(),
MACHINE_LIFECYCLE_STORE_RECORD_VERSION
);
assert_eq!(record.runtime_state(), None);
assert_eq!(record.run().current_run_id(), Some(&run_id));
assert_eq!(record.run().pre_run_phase(), None);
assert_eq!(
version.as_str(),
format!("sha256:{:x}", Sha256::digest(&bytes))
);
assert!(decode_machine_lifecycle_store_record(&bytes).is_err());
}
#[test]
fn lifecycle_observation_distinguishes_unsupported_and_malformed_raw_rows() {
let unsupported = br#"{"record_version":99,"opaque":"future"}"#;
assert!(matches!(
classify_machine_lifecycle_record(unsupported),
MachineLifecycleObservation::Unsupported {
record_version: 99,
..
}
));
let malformed = br#"{"record_version":4,"binding":"torn"}"#;
assert!(matches!(
classify_machine_lifecycle_record(malformed),
MachineLifecycleObservation::Malformed {
record_version: Some(4),
..
}
));
let undecodable = b"not-json";
assert!(matches!(
classify_machine_lifecycle_record(undecodable),
MachineLifecycleObservation::Malformed {
record_version: None,
..
}
));
}
#[test]
fn version_three_unregister_record_migrates_without_run_binding() {
let expected = snapshot(SupervisorAuthoritySnapshot::UnboundNoReceipt);
let mut value = encoded_value(&expected);
value["record_version"] =
serde_json::json!(UNREGISTER_MACHINE_LIFECYCLE_STORE_RECORD_VERSION);
value
.as_object_mut()
.expect("lifecycle record object")
.remove("current_run_id");
value
.as_object_mut()
.expect("lifecycle record object")
.remove("pre_run_phase");
let bytes = serde_json::to_vec(&value).expect("serialize v3 row");
let decoded = decode_machine_lifecycle_store_record(&bytes).expect("decode v3 row");
assert_eq!(decoded, expected);
assert_eq!(decoded.run(), &MachineLifecycleRunFacts::default());
}
#[test]
fn version_two_supervisor_record_migrates_with_no_unregister_progress() {
let bytes = serde_json::to_vec(&serde_json::json!({
"record_version": SUPERVISOR_MACHINE_LIFECYCLE_STORE_RECORD_VERSION,
"runtime_state": RuntimeState::Retired,
"binding": {
"agent_runtime_id": "rt:session:legacy-v2",
"fence_token": 23,
"runtime_generation": 5,
"runtime_epoch_id": "epoch-legacy-v2"
},
"supervisor_authority": { "kind": "unbound_no_receipt" }
}))
.expect("serialize v2 lifecycle record");
let decoded = decode_machine_lifecycle_store_record(&bytes)
.expect("valid v2 supervisor record must migrate");
assert_eq!(decoded.runtime_state(), RuntimeState::Retired);
assert_eq!(decoded.unregister_progress(), None);
}
#[test]
fn current_unregister_progress_rejects_forced_disposition_before_feedback() {
let mut value = encoded_value(&snapshot(SupervisorAuthoritySnapshot::UnboundNoReceipt));
value["unregister_progress"] = serde_json::json!({
"runtime_loop_drain_pending": true,
"comms_drain_exit_pending": false,
"completion_waiter_drain_pending": true,
"runtime_loop_forced_abort": true,
"comms_drain_forced_abort": false
});
assert_decode_fails(value);
}
#[test]
fn version_one_migration_rejects_current_authority_fields() {
assert_decode_fails(serde_json::json!({
"record_version": LEGACY_MACHINE_LIFECYCLE_STORE_RECORD_VERSION,
"runtime_state": RuntimeState::Idle,
"binding": {
"agent_runtime_id": null,
"fence_token": null,
"runtime_generation": null,
"runtime_epoch_id": null
},
"supervisor_authority": { "kind": "unbound_no_receipt" }
}));
}
#[test]
fn mixed_or_unknown_supervisor_authority_fields_fail_closed() {
let current = binding(1, "current-supervisor", 7);
let mut value = encoded_value(&snapshot(SupervisorAuthoritySnapshot::Bound(current)));
value["supervisor_authority"]["rotation"] = serde_json::json!({});
assert_decode_fails(value);
}
#[test]
fn completed_rotation_operation_receipt_round_trips_for_cold_observation() {
let snapshot = snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
operation_id(1),
SupervisorRotationPersistencePhase::Completed,
None,
binding(1, "previous-supervisor", 7),
binding(2, "next-supervisor", 8),
)));
let encoded = encode_snapshot(&snapshot);
let decoded = decode_machine_lifecycle_store_record(&encoded)
.expect("decode completed rotation receipt");
assert_eq!(decoded, snapshot);
}
#[test]
fn exact_current_completed_adoption_round_trips_but_other_equal_epoch_completion_fails() {
let current = binding(3, "already-rotated-supervisor", 9);
let adoption = snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
operation_id(2),
SupervisorRotationPersistencePhase::Completed,
None,
current.clone(),
current,
)));
assert_eq!(
decode_machine_lifecycle_store_record(&encode_snapshot(&adoption))
.expect("exact-current legacy adoption receipt must decode"),
adoption
);
let non_advancing = snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
operation_id(3),
SupervisorRotationPersistencePhase::Completed,
None,
binding(3, "previous-supervisor", 9),
binding(4, "different-supervisor", 9),
)));
assert!(
decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(&non_advancing))
.is_err()
);
}
#[test]
fn malformed_rotation_descriptors_epochs_and_operation_ids_fail_closed() {
let invalid_previous = SupervisorBindingReceipt::new(
String::new(),
"not-a-uuid".to_string(),
"not-an-address".to_string(),
"not-a-key".to_string(),
1,
);
let invalid_previous_receipt =
snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
operation_id(4),
SupervisorRotationPersistencePhase::Rejected,
Some(SupervisorRotationRejection::InvalidTarget),
invalid_previous,
binding(5, "raw-target", 2),
)));
assert!(
decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(
&invalid_previous_receipt,
))
.is_err()
);
let invalid_next = SupervisorBindingReceipt::new(
"invalid-target".to_string(),
"not-a-uuid".to_string(),
"not-an-address".to_string(),
"not-a-key".to_string(),
2,
);
let invalid_completed_target =
snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
operation_id(5),
SupervisorRotationPersistencePhase::Completed,
None,
binding(6, "previous-supervisor", 1),
invalid_next,
)));
assert!(
decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(
&invalid_completed_target,
))
.is_err()
);
let mut invalid_id = encoded_value(&snapshot(
SupervisorAuthoritySnapshot::RotationOperation(rotation(
operation_id(6),
SupervisorRotationPersistencePhase::PreviousRevokePending,
None,
binding(7, "previous-supervisor", 1),
binding(8, "next-supervisor", 2),
)),
));
invalid_id["supervisor_authority"]["rotation"]["operation_id"] =
serde_json::json!("not-a-uuid");
assert_decode_fails(invalid_id);
let nil_id = snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
operation_id(0),
SupervisorRotationPersistencePhase::PreviousRevokePending,
None,
binding(7, "previous-supervisor", 1),
binding(8, "next-supervisor", 2),
)));
assert!(
decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(&nil_id)).is_err()
);
let non_advancing_pending =
snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
operation_id(13),
SupervisorRotationPersistencePhase::PreviousRevokePending,
None,
binding(7, "previous-supervisor", 4),
binding(8, "next-supervisor", 4),
)));
assert!(
decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(
&non_advancing_pending,
))
.is_err()
);
}
#[test]
fn rejected_invalid_or_unsupported_target_preserves_raw_evidence() {
for (id, rejection) in [
(7, SupervisorRotationRejection::InvalidTarget),
(14, SupervisorRotationRejection::UnsupportedProtocolVersion),
] {
let raw_invalid_target = SupervisorBindingReceipt::new(
"".to_string(),
"not-a-peer-id".to_string(),
"not-an-address".to_string(),
"not-a-signing-key".to_string(),
0,
);
let snapshot = snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
operation_id(id),
SupervisorRotationPersistencePhase::Rejected,
Some(rejection),
binding(9, "retained-supervisor", 11),
raw_invalid_target,
)));
assert_eq!(
decode_machine_lifecycle_store_record(&encode_snapshot(&snapshot))
.expect("rejected raw target evidence must remain durable"),
snapshot
);
}
}
#[test]
fn only_raw_target_rejections_are_durable_and_epoch_rejection_must_be_genuine() {
for (id, rejection) in [
(102, SupervisorRotationRejection::OperationConflict),
(103, SupervisorRotationRejection::NotBound),
(104, SupervisorRotationRejection::SenderMismatch),
] {
let impossible = snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
operation_id(id),
SupervisorRotationPersistencePhase::Rejected,
Some(rejection),
binding(32, "retained-supervisor", 7),
binding(33, "requested-supervisor", 8),
)));
assert!(
MachineLifecycleStoreRecord::from_snapshot(&impossible)
.encode()
.is_err()
);
assert!(
decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(&impossible))
.is_err()
);
}
let advancing = snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
operation_id(105),
SupervisorRotationPersistencePhase::Rejected,
Some(SupervisorRotationRejection::TargetEpochNotAdvanced),
binding(34, "retained-supervisor", 9),
binding(35, "advancing-target", 10),
)));
assert!(
MachineLifecycleStoreRecord::from_snapshot(&advancing)
.encode()
.is_err()
);
assert!(
decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(&advancing))
.is_err()
);
let non_advancing = snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
operation_id(106),
SupervisorRotationPersistencePhase::Rejected,
Some(SupervisorRotationRejection::TargetEpochNotAdvanced),
binding(36, "retained-supervisor", 11),
binding(37, "non-advancing-target", 11),
)));
assert_eq!(
decode_machine_lifecycle_store_record(&encode_snapshot(&non_advancing))
.expect("genuine target-epoch rejection must remain durable"),
non_advancing
);
}
#[test]
fn malformed_current_authority_variants_fail_closed() {
let malformed = SupervisorBindingReceipt::new(
String::new(),
"not-a-peer-id".to_string(),
"not-an-address".to_string(),
"not-a-signing-key".to_string(),
1,
);
let bound = snapshot(SupervisorAuthoritySnapshot::Bound(malformed.clone()));
assert!(
decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(&bound)).is_err()
);
let pending = snapshot(SupervisorAuthoritySnapshot::RevocationPending(
SupervisorRevocationPendingReceipt::new(
malformed.name().to_owned(),
malformed.peer_id().to_owned(),
malformed.address().to_owned(),
malformed.signing_public_key().to_owned(),
malformed.epoch(),
),
));
assert!(
decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(&pending)).is_err()
);
let revoked = snapshot(SupervisorAuthoritySnapshot::RevokedReceipt(
RevokedSupervisorReceipt::new(
malformed.peer_id().to_owned(),
malformed.signing_public_key().to_owned(),
malformed.epoch(),
),
));
assert!(
decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(&revoked)).is_err()
);
}
#[test]
fn partial_and_nonterminal_history_records_fail_closed() {
let receipt = rotation(
operation_id(8),
SupervisorRotationPersistencePhase::Completed,
None,
binding(10, "history-previous", 1),
binding(11, "history-next", 2),
);
let history = std::collections::BTreeMap::from([(receipt.operation_id(), receipt)]);
let snapshot = snapshot(SupervisorAuthoritySnapshot::WithRotationHistory {
current: Box::new(SupervisorAuthoritySnapshot::Bound(binding(
12,
"current-supervisor",
3,
))),
terminal_receipts: history,
});
let mut partial = encoded_value(&snapshot);
partial["supervisor_authority"]["terminal_receipts"][0]
.as_object_mut()
.expect("history receipt object")
.remove("next");
assert_decode_fails(partial);
let mut nonterminal = encoded_value(&snapshot);
nonterminal["supervisor_authority"]["terminal_receipts"][0]["phase"] =
serde_json::json!("next_publish_pending");
assert_decode_fails(nonterminal);
}
#[test]
fn duplicate_nested_and_active_history_conflicts_fail_closed() {
let history_receipt = rotation(
operation_id(9),
SupervisorRotationPersistencePhase::Completed,
None,
binding(13, "history-previous", 1),
binding(14, "history-next", 2),
);
let history = std::collections::BTreeMap::from([(
history_receipt.operation_id(),
history_receipt.clone(),
)]);
let wrapper = snapshot(SupervisorAuthoritySnapshot::WithRotationHistory {
current: Box::new(SupervisorAuthoritySnapshot::Bound(binding(
15,
"current-supervisor",
3,
))),
terminal_receipts: history,
});
let mut duplicate = encoded_value(&wrapper);
let receipt = duplicate["supervisor_authority"]["terminal_receipts"][0].clone();
duplicate["supervisor_authority"]["terminal_receipts"]
.as_array_mut()
.expect("history receipt array")
.push(receipt);
assert_decode_fails(duplicate);
let mut nested = encoded_value(&wrapper);
let nested_current = nested["supervisor_authority"].clone();
nested["supervisor_authority"]["current"] = nested_current;
assert_decode_fails(nested);
let active_conflict = snapshot(SupervisorAuthoritySnapshot::WithRotationHistory {
current: Box::new(SupervisorAuthoritySnapshot::RotationOperation(
history_receipt.clone(),
)),
terminal_receipts: std::collections::BTreeMap::from([(
history_receipt.operation_id(),
history_receipt,
)]),
});
assert!(
MachineLifecycleStoreRecord::from_snapshot(&active_conflict)
.encode()
.is_err()
);
let empty_history = snapshot(SupervisorAuthoritySnapshot::WithRotationHistory {
current: Box::new(SupervisorAuthoritySnapshot::Bound(binding(
20,
"current-supervisor",
4,
))),
terminal_receipts: std::collections::BTreeMap::new(),
});
assert!(
MachineLifecycleStoreRecord::from_snapshot(&empty_history)
.encode()
.is_err()
);
assert!(
decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(&empty_history))
.is_err()
);
let mismatched_key = snapshot(SupervisorAuthoritySnapshot::WithRotationHistory {
current: Box::new(SupervisorAuthoritySnapshot::Bound(binding(
21,
"current-supervisor",
4,
))),
terminal_receipts: std::collections::BTreeMap::from([(
operation_id(99),
rotation(
operation_id(98),
SupervisorRotationPersistencePhase::Completed,
None,
binding(22, "history-previous", 2),
binding(23, "history-next", 3),
),
)]),
});
assert!(
MachineLifecycleStoreRecord::from_snapshot(&mismatched_key)
.encode()
.is_err()
);
}
#[test]
fn history_current_epoch_and_same_epoch_identity_must_cohere() {
let previous = binding(38, "history-previous", 12);
let next = binding(39, "history-next", 13);
let completed = rotation(
operation_id(107),
SupervisorRotationPersistencePhase::Completed,
None,
previous.clone(),
next.clone(),
);
let history =
std::collections::BTreeMap::from([(completed.operation_id(), completed.clone())]);
let stale_current = snapshot(SupervisorAuthoritySnapshot::WithRotationHistory {
current: Box::new(SupervisorAuthoritySnapshot::Bound(binding(
38,
"refreshed-history-previous",
12,
))),
terminal_receipts: history.clone(),
});
assert!(
MachineLifecycleStoreRecord::from_snapshot(&stale_current)
.encode()
.is_err()
);
assert!(
decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(&stale_current))
.is_err()
);
let conflicting_current = snapshot(SupervisorAuthoritySnapshot::WithRotationHistory {
current: Box::new(SupervisorAuthoritySnapshot::Bound(binding(
40,
"conflicting-current",
13,
))),
terminal_receipts: history.clone(),
});
assert!(
MachineLifecycleStoreRecord::from_snapshot(&conflicting_current)
.encode()
.is_err()
);
assert!(
decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(
&conflicting_current,
))
.is_err()
);
let route_refreshed_current = snapshot(SupervisorAuthoritySnapshot::WithRotationHistory {
current: Box::new(SupervisorAuthoritySnapshot::Bound(binding(
39,
"route-refreshed-history-next",
13,
))),
terminal_receipts: history,
});
assert_eq!(
decode_machine_lifecycle_store_record(&encode_snapshot(&route_refreshed_current))
.expect("same identity may refresh route metadata within one epoch"),
route_refreshed_current
);
}
#[test]
fn terminal_history_survives_later_rotation_and_recovery() {
let first = rotation(
operation_id(10),
SupervisorRotationPersistencePhase::Completed,
None,
binding(16, "first-supervisor", 1),
binding(17, "second-supervisor", 2),
);
let rejected = rotation(
operation_id(11),
SupervisorRotationPersistencePhase::Rejected,
Some(SupervisorRotationRejection::TargetEpochNotAdvanced),
binding(17, "second-supervisor", 2),
binding(18, "rejected-supervisor", 2),
);
let later = rotation(
operation_id(12),
SupervisorRotationPersistencePhase::Completed,
None,
binding(17, "second-supervisor", 2),
binding(19, "current-supervisor", 3),
);
let snapshot = snapshot(SupervisorAuthoritySnapshot::WithRotationHistory {
current: Box::new(SupervisorAuthoritySnapshot::RotationOperation(later)),
terminal_receipts: std::collections::BTreeMap::from([
(first.operation_id(), first),
(rejected.operation_id(), rejected),
]),
});
let decoded = decode_machine_lifecycle_store_record(&encode_snapshot(&snapshot))
.expect("later rotation and old terminal history must recover together");
assert_eq!(decoded, snapshot);
}
}