use crate::model::{canonicalize_options, require_text};
use crate::{
DecisionAction, DecisionAttemptOutcome, DecisionAttemptRecord, DecisionControllerBinding,
DecisionError, DecisionErrorCode, DecisionMutation, DecisionOutcome, DecisionPolicy,
DecisionPolicyIdentity, DecisionTicket, DecisionTicketState, DecisionTrace, PolicyDecision,
};
use canwu_core::{CommandRequestId, DecisionTicketId, DecisionTraceId};
use canwu_time::SimTime;
use im::{OrdMap, OrdSet};
use serde::{Deserialize, Serialize};
use std::cell::RefCell;
use std::collections::BTreeMap as StdBTreeMap;
use std::fmt::Write as _;
use std::mem::size_of;
use std::ops::Index;
use std::sync::Arc;
mod persistent_decision_map_serde {
use im::OrdMap;
use serde::ser::SerializeMap as _;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use std::collections::BTreeMap;
use std::sync::Arc;
pub fn serialize<S, K, V>(entries: &OrdMap<K, Arc<V>>, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
K: Clone + Ord + Serialize,
V: Serialize,
{
let mut map = serializer.serialize_map(Some(entries.len()))?;
for (ordinal, entry) in entries {
map.serialize_entry(ordinal, entry.as_ref())?;
}
map.end()
}
pub fn deserialize<'de, D, K, V>(deserializer: D) -> Result<OrdMap<K, Arc<V>>, D::Error>
where
D: Deserializer<'de>,
K: Clone + Deserialize<'de> + Ord,
V: Deserialize<'de>,
{
Ok(BTreeMap::<K, V>::deserialize(deserializer)?
.into_iter()
.map(|(ordinal, entry)| (ordinal, Arc::new(entry)))
.collect())
}
}
mod decision_archive_receipts_serde {
use super::{
CompactDecisionArchiveReceipt, DecisionArchivePageKey, DecisionArchiveReceipt,
DecisionHistoryKey,
};
use im::OrdMap;
use serde::de::Error as _;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
pub fn serialize<S>(
buckets: &OrdMap<
DecisionArchivePageKey,
OrdMap<DecisionHistoryKey, CompactDecisionArchiveReceipt>,
>,
serializer: S,
) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
buckets
.values()
.flat_map(OrdMap::iter)
.map(|(key, receipt)| receipt.to_receipt(key))
.collect::<Vec<_>>()
.serialize(serializer)
}
pub fn deserialize<'de, D>(
deserializer: D,
) -> Result<
OrdMap<DecisionArchivePageKey, OrdMap<DecisionHistoryKey, CompactDecisionArchiveReceipt>>,
D::Error,
>
where
D: Deserializer<'de>,
{
let entries = Vec::<DecisionArchiveReceipt>::deserialize(deserializer)?;
let mut buckets = OrdMap::<
DecisionArchivePageKey,
OrdMap<DecisionHistoryKey, CompactDecisionArchiveReceipt>,
>::new();
for receipt in entries {
let key = receipt.key.clone();
let bucket = super::decision_history_page_key(&key).map_err(D::Error::custom)?;
let value =
CompactDecisionArchiveReceipt::from_receipt(&receipt).map_err(D::Error::custom)?;
if buckets
.entry(bucket)
.or_default()
.insert(key, value)
.is_some()
{
return Err(D::Error::custom(
"decision archive receipt map contains a duplicate key",
));
}
}
Ok(buckets)
}
}
mod decision_archive_page_directory_serde {
use super::DecisionArchivePageKey;
use im::OrdMap;
use serde::de::Error as _;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
pub fn serialize<S>(
pages: &OrdMap<DecisionArchivePageKey, String>,
serializer: S,
) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
pages
.iter()
.map(|(key, id)| (*key, id))
.collect::<Vec<_>>()
.serialize(serializer)
}
pub fn deserialize<'de, D>(
deserializer: D,
) -> Result<OrdMap<DecisionArchivePageKey, String>, D::Error>
where
D: Deserializer<'de>,
{
let entries = Vec::<(DecisionArchivePageKey, String)>::deserialize(deserializer)?;
let mut pages = OrdMap::new();
let mut previous = None;
for (key, id) in entries {
if previous.is_some_and(|previous| previous >= key) || pages.insert(key, id).is_some() {
return Err(D::Error::custom(
"decision archive page directory is not strictly ordered",
));
}
previous = Some(key);
}
Ok(pages)
}
}
#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case", tag = "kind", content = "id")]
pub enum DecisionHistoryKey {
Ticket(canwu_core::DecisionTicketId),
Attempt(canwu_core::DecisionRequestId),
Trace(canwu_core::DecisionTraceId),
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case", tag = "location")]
pub enum DecisionHistoryLocation {
Hot,
Archived {
locator: String,
},
Unresolved {
bucket: u16,
segment: u8,
},
Absent,
}
pub const DECISION_ARCHIVE_FORMAT_VERSION: u32 = 1;
pub const MAX_DECISION_ARCHIVE_BATCH_ENTRIES: usize = 4_096;
pub const MAX_DECISION_HISTORY_PAGE_SIZE: usize = 512;
pub const MAX_DECISION_HISTORY_PAGE_BYTES: u64 = 16 * 1024 * 1024;
pub const DECISION_HISTORY_BUCKET_BITS: u8 = 12;
pub const DECISION_HISTORY_BUCKET_COUNT: u16 = 1 << DECISION_HISTORY_BUCKET_BITS;
pub const DECISION_ARCHIVE_BUCKET_PAGE_FORMAT_VERSION: u32 = 1;
pub const MAX_DECISION_ARCHIVE_BUCKET_PAGE_ENTRIES: usize = 64;
pub const MAX_DECISION_ARCHIVE_BUCKET_PAGE_BYTES: usize = 1024 * 1024;
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DecisionArchivePageKey {
pub bucket: u16,
pub segment: u8,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DecisionHistoryQueryBudget {
pub max_results: usize,
pub max_provider_calls: usize,
pub max_decoded_bytes: u64,
}
impl Default for DecisionHistoryQueryBudget {
fn default() -> Self {
Self {
max_results: 128,
max_provider_calls: 128,
max_decoded_bytes: 4 * 1024 * 1024,
}
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DecisionHistoryCursor {
pub archive_root: String,
pub bucket: u16,
pub segment: u8,
pub after: DecisionHistoryKey,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DecisionHistoryPage {
pub archive_root: String,
pub records: Vec<DecisionArchiveRecord>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub next_cursor: Option<DecisionHistoryCursor>,
pub provider_calls: u64,
pub decoded_bytes: u64,
}
#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DecisionArchiveReachability {
pub bucket_page_ids: OrdSet<String>,
pub blob_locators: OrdSet<String>,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DecisionArchiveBucketPage {
pub format_version: u32,
pub bucket: u16,
pub segment: u8,
pub receipts: Vec<DecisionArchiveReceipt>,
}
impl DecisionArchiveBucketPage {
pub fn validate(&self) -> Result<(), DecisionError> {
if self.format_version != DECISION_ARCHIVE_BUCKET_PAGE_FORMAT_VERSION
|| self.bucket >= DECISION_HISTORY_BUCKET_COUNT
|| self.receipts.is_empty()
|| self.receipts.len() > MAX_DECISION_ARCHIVE_BUCKET_PAGE_ENTRIES
|| self
.receipts
.windows(2)
.any(|pair| pair[0].key >= pair[1].key)
|| self.receipts.iter().any(|receipt| {
decision_history_page_key(&receipt.key).ok()
!= Some(DecisionArchivePageKey {
bucket: self.bucket,
segment: self.segment,
})
|| CompactDecisionArchiveReceipt::from_receipt(receipt).is_err()
})
{
return Err(archive_error(
"decision archive bucket page is malformed or non-canonical",
));
}
let encoded = serde_json::to_vec(self).map_err(|error| {
archive_error(format!(
"cannot encode decision archive bucket page: {error}"
))
})?;
if encoded.len() > MAX_DECISION_ARCHIVE_BUCKET_PAGE_BYTES {
return Err(archive_error(
"decision archive bucket page exceeds the hard byte limit",
));
}
Ok(())
}
pub fn state_page_id(&self) -> Result<String, DecisionError> {
self.validate()?;
let bytes = serde_json::to_vec(self).map_err(|error| {
archive_error(format!(
"cannot encode decision archive bucket page: {error}"
))
})?;
let mut hasher = blake3::Hasher::new();
hasher.update(b"canwu.state-page.v1");
hasher.update(&[0]);
hasher.update(&bytes);
Ok(hasher.finalize().to_hex().to_string())
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DecisionLocatorScaleMetrics {
pub entries: u64,
pub locator_pages: u64,
pub max_page_entries: u64,
pub max_page_encoded_bytes: u64,
pub archive_batches: u64,
pub exact_restart_queries: u64,
pub reachable_blob_locators: u64,
pub estimated_resident_structural_bytes: u64,
pub root_hash: String,
}
#[doc(hidden)]
pub struct DecisionLocatorScaleFixture {
pub state: DecisionState,
pub archive_blobs: Vec<DecisionArchiveBlob>,
pub metrics: DecisionLocatorScaleMetrics,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct TraceLocatorScaleMetrics {
pub hot_trace_entries: u64,
pub indexed_lookup_samples: u64,
pub archive_commit_entries: u64,
pub target_archived: bool,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "record", rename_all = "snake_case")]
pub enum DecisionArchiveRecord {
Ticket { ticket: DecisionTicket },
Attempt { attempt: DecisionAttemptRecord },
Trace { trace: DecisionTrace },
}
impl DecisionArchiveRecord {
#[must_use]
pub fn key(&self) -> DecisionHistoryKey {
match self {
Self::Ticket { ticket } => DecisionHistoryKey::Ticket(ticket.id),
Self::Attempt { attempt } => DecisionHistoryKey::Attempt(attempt.request_id),
Self::Trace { trace } => DecisionHistoryKey::Trace(trace.id),
}
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DecisionArchiveBlob {
pub format_version: u32,
pub key: DecisionHistoryKey,
pub record: DecisionArchiveRecord,
}
impl DecisionArchiveBlob {
pub fn validate(&self) -> Result<(), DecisionError> {
if self.format_version != DECISION_ARCHIVE_FORMAT_VERSION || self.key != self.record.key() {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
"decision archive blob format or typed identity is invalid",
));
}
Ok(())
}
pub fn content_id(&self) -> Result<String, DecisionError> {
self.validate()?;
decision_archive_hash("canwu.decision.archive-blob.v1", self)
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DecisionArchiveReceipt {
pub format_version: u32,
pub key: DecisionHistoryKey,
pub locator: String,
pub content_id: String,
pub encoded_bytes: u64,
}
#[derive(Clone, Debug, Eq, PartialEq)]
struct CompactDecisionArchiveReceipt {
content_id: [u8; 32],
encoded_bytes: u64,
}
impl CompactDecisionArchiveReceipt {
fn from_receipt(receipt: &DecisionArchiveReceipt) -> Result<Self, DecisionError> {
if receipt.format_version != DECISION_ARCHIVE_FORMAT_VERSION
|| receipt.locator != receipt.content_id
|| receipt.encoded_bytes == 0
{
return Err(archive_error("decision archive receipt is not canonical"));
}
Ok(Self {
content_id: decode_archive_hash(&receipt.content_id)?,
encoded_bytes: receipt.encoded_bytes,
})
}
fn to_receipt(&self, key: &DecisionHistoryKey) -> DecisionArchiveReceipt {
let content_id = encode_archive_hash(self.content_id);
DecisionArchiveReceipt {
format_version: DECISION_ARCHIVE_FORMAT_VERSION,
key: key.clone(),
locator: content_id.clone(),
content_id,
encoded_bytes: self.encoded_bytes,
}
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct PreparedDecisionArchive {
pub source_root: String,
pub token: String,
pub blobs: Vec<DecisionArchiveBlob>,
pub receipts: Vec<DecisionArchiveReceipt>,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct VerifiedDecisionArchiveCommit {
format_version: u32,
source_root: String,
token: String,
receipts: Vec<DecisionArchiveReceipt>,
page_replacements: Vec<VerifiedDecisionArchivePageReplacement>,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
struct VerifiedDecisionArchivePageReplacement {
page_key: DecisionArchivePageKey,
previous_page_id: Option<String>,
page: DecisionArchiveBucketPage,
}
impl VerifiedDecisionArchiveCommit {
#[must_use]
pub fn token(&self) -> &str {
&self.token
}
#[must_use]
pub fn source_root(&self) -> &str {
&self.source_root
}
#[must_use]
pub fn has_current_nonempty_shape(&self) -> bool {
self.format_version == DECISION_ARCHIVE_FORMAT_VERSION
&& !self.receipts.is_empty()
&& !self.page_replacements.is_empty()
}
pub fn archive_locators(&self) -> impl Iterator<Item = &str> {
self.receipts.iter().map(|receipt| receipt.locator.as_str())
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DecisionArchiveStoreOutcome {
Stored,
AlreadyStored,
}
pub trait DecisionArchiveProvider {
fn load_decision_archive(
&self,
locator: &str,
) -> Result<Option<DecisionArchiveBlob>, DecisionError>;
fn load_decision_archive_bucket_page(
&self,
_page_id: &str,
) -> Result<Option<DecisionArchiveBucketPage>, DecisionError> {
Ok(None)
}
}
pub trait DecisionArchiveStore: DecisionArchiveProvider {
fn store_decision_archive(
&self,
blob: &DecisionArchiveBlob,
) -> Result<DecisionArchiveStoreOutcome, DecisionError>;
}
fn archive_error(message: impl Into<String>) -> DecisionError {
DecisionError::new(DecisionErrorCode::InvalidDecision, message)
}
fn history_unavailable(message: impl Into<String>) -> DecisionError {
DecisionError::new(DecisionErrorCode::DecisionHistoryUnavailable, message)
}
fn history_budget_error(message: impl Into<String>) -> DecisionError {
DecisionError::new(DecisionErrorCode::QueryBudgetExceeded, message)
}
fn decision_archive_hash(domain: &str, value: &impl Serialize) -> Result<String, DecisionError> {
let encoded = serde_json::to_vec(&(domain, value)).map_err(|error| {
archive_error(format!(
"cannot encode canonical decision archive content: {error}"
))
})?;
Ok(blake3::hash(&encoded).to_hex().to_string())
}
fn canonical_archive_hash(value: &str) -> bool {
value.len() == 64
&& value
.bytes()
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
}
fn decode_archive_hash(value: &str) -> Result<[u8; 32], DecisionError> {
if !canonical_archive_hash(value) {
return Err(archive_error(
"decision archive hash must be lower-case 32-byte hexadecimal",
));
}
let mut decoded = [0_u8; 32];
let (pairs, remainder) = value.as_bytes().as_chunks::<2>();
debug_assert!(remainder.is_empty());
for (index, pair) in pairs.iter().enumerate() {
let digit = |byte| match byte {
b'0'..=b'9' => Some(byte - b'0'),
b'a'..=b'f' => Some(byte - b'a' + 10),
_ => None,
};
decoded[index] = digit(pair[0])
.zip(digit(pair[1]))
.map(|(high, low)| (high << 4) | low)
.ok_or_else(|| archive_error("decision archive hash contains invalid hexadecimal"))?;
}
Ok(decoded)
}
fn encode_archive_hash(value: [u8; 32]) -> String {
let mut encoded = String::with_capacity(64);
for byte in value {
write!(&mut encoded, "{byte:02x}").expect("writing to a string cannot fail");
}
encoded
}
#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
pub struct DecisionHotState {
pub ticket_count: u64,
pub attempt_count: u64,
pub trace_count: u64,
}
#[derive(Clone, Debug, Default, Eq, PartialEq)]
struct DecisionHotHistoryAccumulator {
count: u64,
xor: [u8; 32],
sum: [u8; 32],
}
impl DecisionHotHistoryAccumulator {
fn insert(
&mut self,
key: &DecisionHistoryKey,
record: &DecisionArchiveRecord,
) -> Result<(), DecisionError> {
let digest = decision_hot_leaf_hash(key, record)?;
self.count = self
.count
.checked_add(1)
.ok_or_else(|| archive_error("decision hot-history count overflowed"))?;
for (target, byte) in self.xor.iter_mut().zip(digest) {
*target ^= byte;
}
add_digest_mod_256(&mut self.sum, digest);
Ok(())
}
fn remove(
&mut self,
key: &DecisionHistoryKey,
record: &DecisionArchiveRecord,
) -> Result<(), DecisionError> {
let digest = decision_hot_leaf_hash(key, record)?;
self.count = self
.count
.checked_sub(1)
.ok_or_else(|| archive_error("decision hot-history count underflowed"))?;
for (target, byte) in self.xor.iter_mut().zip(digest) {
*target ^= byte;
}
subtract_digest_mod_256(&mut self.sum, digest);
Ok(())
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(bound(serialize = "T: Serialize", deserialize = "T: Deserialize<'de>"))]
pub struct PersistentDecisionLog<T: Clone> {
next_ordinal: u64,
#[serde(with = "persistent_decision_map_serde")]
entries: OrdMap<u64, Arc<T>>,
}
impl<T: Clone> Default for PersistentDecisionLog<T> {
fn default() -> Self {
Self {
next_ordinal: 1,
entries: OrdMap::new(),
}
}
}
impl<T: Clone> PersistentDecisionLog<T> {
#[must_use]
pub fn is_empty(&self) -> bool {
self.entries.is_empty()
}
fn is_default_empty(&self) -> bool {
self.entries.is_empty() && self.next_ordinal == 1
}
#[must_use]
pub fn len(&self) -> usize {
self.entries.len()
}
#[must_use]
pub fn get(&self, index: usize) -> Option<&T> {
self.entries.values().nth(index).map(Arc::as_ref)
}
#[must_use]
pub fn last(&self) -> Option<&T> {
self.entries.get_max().map(|(_, value)| value.as_ref())
}
#[must_use]
pub fn iter(&self) -> impl DoubleEndedIterator<Item = &T> + ExactSizeIterator {
self.entries.values().map(Arc::as_ref)
}
pub fn push(&mut self, value: T) -> u64 {
let ordinal = self.next_ordinal;
self.next_ordinal = self
.next_ordinal
.checked_add(1)
.expect("decision log ordinal range is exhausted");
self.entries.insert(ordinal, Arc::new(value));
ordinal
}
#[must_use]
pub fn get_ordinal(&self, ordinal: u64) -> Option<&T> {
self.entries.get(&ordinal).map(Arc::as_ref)
}
pub fn remove_ordinal(&mut self, ordinal: u64) -> Option<T> {
self.entries.remove(&ordinal).map(Arc::unwrap_or_clone)
}
#[must_use]
pub fn ordinals(&self) -> impl DoubleEndedIterator<Item = u64> + ExactSizeIterator + '_ {
self.entries.keys().copied()
}
#[must_use]
pub const fn next_ordinal(&self) -> u64 {
self.next_ordinal
}
}
impl<T: Clone> Index<usize> for PersistentDecisionLog<T> {
type Output = T;
fn index(&self, index: usize) -> &Self::Output {
self.get(index)
.expect("persistent decision log index is out of bounds")
}
}
pub struct PersistentDecisionLogIter<'a, T: Clone> {
inner: im::ordmap::Values<'a, u64, Arc<T>>,
}
impl<'a, T: Clone> Iterator for PersistentDecisionLogIter<'a, T> {
type Item = &'a T;
fn next(&mut self) -> Option<Self::Item> {
self.inner.next().map(Arc::as_ref)
}
fn size_hint(&self) -> (usize, Option<usize>) {
self.inner.size_hint()
}
}
impl<T: Clone> DoubleEndedIterator for PersistentDecisionLogIter<'_, T> {
fn next_back(&mut self) -> Option<Self::Item> {
self.inner.next_back().map(Arc::as_ref)
}
}
impl<T: Clone> ExactSizeIterator for PersistentDecisionLogIter<'_, T> {}
impl<'a, T: Clone> IntoIterator for &'a PersistentDecisionLog<T> {
type Item = &'a T;
type IntoIter = PersistentDecisionLogIter<'a, T>;
fn into_iter(self) -> Self::IntoIter {
PersistentDecisionLogIter {
inner: self.entries.values(),
}
}
}
#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize)]
pub struct DecisionState {
#[serde(
default,
skip_serializing_if = "OrdMap::is_empty",
with = "persistent_decision_map_serde"
)]
pub controllers: OrdMap<String, Arc<DecisionControllerBinding>>,
#[serde(
default,
skip_serializing_if = "OrdMap::is_empty",
with = "persistent_decision_map_serde"
)]
pub tickets: OrdMap<DecisionTicketId, Arc<DecisionTicket>>,
#[serde(
default,
skip_serializing_if = "PersistentDecisionLog::is_default_empty"
)]
pub traces: PersistentDecisionLog<DecisionTrace>,
#[serde(
default,
skip_serializing_if = "PersistentDecisionLog::is_default_empty"
)]
attempts: PersistentDecisionLog<DecisionAttemptRecord>,
#[serde(skip)]
attempts_by_request: OrdMap<canwu_core::DecisionRequestId, u64>,
#[serde(skip)]
deadline_index: OrdMap<SimTime, OrdSet<DecisionTicketId>>,
#[serde(
default,
skip_serializing_if = "OrdMap::is_empty",
with = "decision_archive_receipts_serde"
)]
archive_receipt_buckets:
OrdMap<DecisionArchivePageKey, OrdMap<DecisionHistoryKey, CompactDecisionArchiveReceipt>>,
#[serde(
default,
skip_serializing_if = "OrdMap::is_empty",
with = "decision_archive_page_directory_serde"
)]
archive_bucket_page_ids: OrdMap<DecisionArchivePageKey, String>,
#[serde(default, skip_serializing_if = "is_zero_u64")]
archive_receipt_count: u64,
#[serde(skip)]
hot_history_accumulator: DecisionHotHistoryAccumulator,
}
#[allow(clippy::trivially_copy_pass_by_ref)]
const fn is_zero_u64(value: &u64) -> bool {
*value == 0
}
impl<'de> Deserialize<'de> for DecisionState {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
#[derive(Deserialize)]
struct PersistedDecisionState {
#[serde(default, with = "persistent_decision_map_serde")]
controllers: OrdMap<String, Arc<DecisionControllerBinding>>,
#[serde(default, with = "persistent_decision_map_serde")]
tickets: OrdMap<canwu_core::DecisionTicketId, Arc<DecisionTicket>>,
#[serde(default)]
traces: PersistentDecisionLog<DecisionTrace>,
#[serde(default)]
attempts: PersistentDecisionLog<DecisionAttemptRecord>,
#[serde(default, with = "decision_archive_receipts_serde")]
archive_receipt_buckets: OrdMap<
DecisionArchivePageKey,
OrdMap<DecisionHistoryKey, CompactDecisionArchiveReceipt>,
>,
#[serde(default, with = "decision_archive_page_directory_serde")]
archive_bucket_page_ids: OrdMap<DecisionArchivePageKey, String>,
#[serde(default)]
archive_receipt_count: u64,
}
let persisted = PersistedDecisionState::deserialize(deserializer)?;
let attempts_by_request = persisted
.attempts
.ordinals()
.zip(persisted.attempts.iter())
.map(|(ordinal, attempt)| (attempt.request_id, ordinal))
.collect();
let mut deadline_index = OrdMap::<SimTime, OrdSet<DecisionTicketId>>::new();
for ticket in persisted.tickets.values().filter(|ticket| ticket.is_open()) {
if let Some(deadline) = ticket.deadline {
deadline_index
.entry(deadline)
.or_default()
.insert(ticket.id);
}
}
Self {
controllers: persisted.controllers,
tickets: persisted.tickets,
traces: persisted.traces,
attempts: persisted.attempts,
attempts_by_request,
deadline_index,
archive_receipt_buckets: persisted.archive_receipt_buckets,
archive_bucket_page_ids: persisted.archive_bucket_page_ids,
archive_receipt_count: persisted.archive_receipt_count,
hot_history_accumulator: DecisionHotHistoryAccumulator::default(),
}
.with_rebuilt_hot_history_accumulator()
.map_err(serde::de::Error::custom)?
.with_rebuilt_archive_bucket_pages()
.map_err(serde::de::Error::custom)
}
}
impl DecisionState {
fn with_rebuilt_hot_history_accumulator(mut self) -> Result<Self, DecisionError> {
self.rebuild_hot_history_accumulator()?;
Ok(self)
}
fn rebuild_hot_history_accumulator(&mut self) -> Result<(), DecisionError> {
let mut rebuilt = DecisionHotHistoryAccumulator::default();
for ticket in self.tickets.values() {
let key = DecisionHistoryKey::Ticket(ticket.id);
rebuilt.insert(
&key,
&DecisionArchiveRecord::Ticket {
ticket: ticket.as_ref().clone(),
},
)?;
}
for attempt in self.attempts.iter() {
let key = DecisionHistoryKey::Attempt(attempt.request_id);
rebuilt.insert(
&key,
&DecisionArchiveRecord::Attempt {
attempt: attempt.clone(),
},
)?;
}
for trace in self.traces.iter() {
let key = DecisionHistoryKey::Trace(trace.id);
rebuilt.insert(
&key,
&DecisionArchiveRecord::Trace {
trace: trace.clone(),
},
)?;
}
self.hot_history_accumulator = rebuilt;
Ok(())
}
fn insert_hot_history_record(
&mut self,
record: &DecisionArchiveRecord,
) -> Result<(), DecisionError> {
self.hot_history_accumulator.insert(&record.key(), record)
}
fn replace_hot_history_record(
&mut self,
previous: &DecisionArchiveRecord,
next: &DecisionArchiveRecord,
) -> Result<(), DecisionError> {
if previous.key() != next.key() {
return Err(archive_error(
"decision hot-history replacement changed record identity",
));
}
self.hot_history_accumulator
.remove(&previous.key(), previous)?;
self.hot_history_accumulator.insert(&next.key(), next)
}
fn remove_hot_history_record(
&mut self,
record: &DecisionArchiveRecord,
) -> Result<(), DecisionError> {
self.hot_history_accumulator.remove(&record.key(), record)
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.controllers.is_empty()
&& self.tickets.is_empty()
&& self.traces.is_empty()
&& self.attempts.is_empty()
&& self.archive_receipt_count == 0
}
fn compact_archive_receipt(
&self,
key: &DecisionHistoryKey,
) -> Option<&CompactDecisionArchiveReceipt> {
let bucket = decision_history_page_key(key).ok()?;
self.archive_receipt_buckets.get(&bucket)?.get(key)
}
fn contains_archived_key(&self, key: &DecisionHistoryKey) -> bool {
self.compact_archive_receipt(key).is_some()
}
fn insert_archive_receipt(
&mut self,
receipt: &DecisionArchiveReceipt,
) -> Result<Option<CompactDecisionArchiveReceipt>, DecisionError> {
let bucket = decision_history_page_key(&receipt.key)?;
let previous = self
.archive_receipt_buckets
.entry(bucket)
.or_default()
.insert(
receipt.key.clone(),
CompactDecisionArchiveReceipt::from_receipt(receipt)?,
);
self.refresh_archive_bucket_page_id(bucket)?;
if previous.is_none() {
self.archive_receipt_count = self
.archive_receipt_count
.checked_add(1)
.ok_or_else(|| archive_error("decision archive receipt count overflowed"))?;
}
Ok(previous)
}
fn with_rebuilt_archive_bucket_pages(mut self) -> Result<Self, DecisionError> {
let resident_count =
self.archive_receipt_buckets
.values()
.try_fold(0_u64, |total, receipts| {
total
.checked_add(receipts.len() as u64)
.ok_or_else(|| archive_error("decision archive receipt count overflowed"))
})?;
if resident_count > 0 {
let buckets = self
.archive_receipt_buckets
.keys()
.copied()
.collect::<Vec<_>>();
if self.archive_bucket_page_ids.is_empty() {
self.archive_receipt_count = resident_count;
for bucket in buckets {
self.refresh_archive_bucket_page_id(bucket)?;
}
} else {
for bucket in buckets {
let resident_page_id = self
.decision_archive_bucket_page(bucket)?
.ok_or_else(|| archive_error("resident decision archive bucket vanished"))?
.state_page_id()?;
if self.archive_bucket_page_ids.get(&bucket) != Some(&resident_page_id) {
return Err(archive_error(
"decision archive bucket directory disagrees with resident receipts",
));
}
}
}
}
self.validate_archive_directory_shape()?;
Ok(self)
}
fn refresh_archive_bucket_page_id(
&mut self,
bucket: DecisionArchivePageKey,
) -> Result<(), DecisionError> {
let page = self
.decision_archive_bucket_page(bucket)?
.ok_or_else(|| archive_error("decision archive bucket unexpectedly disappeared"))?;
self.archive_bucket_page_ids
.insert(bucket, page.state_page_id()?);
Ok(())
}
pub fn decision_archive_bucket_page(
&self,
bucket: DecisionArchivePageKey,
) -> Result<Option<DecisionArchiveBucketPage>, DecisionError> {
let Some(receipts) = self.archive_receipt_buckets.get(&bucket) else {
return Ok(None);
};
let page = DecisionArchiveBucketPage {
format_version: DECISION_ARCHIVE_BUCKET_PAGE_FORMAT_VERSION,
bucket: bucket.bucket,
segment: bucket.segment,
receipts: receipts
.iter()
.map(|(key, receipt)| receipt.to_receipt(key))
.collect(),
};
page.validate()?;
Ok(Some(page))
}
#[must_use]
pub fn decision_archive_bucket_page_ids(&self) -> &OrdMap<DecisionArchivePageKey, String> {
&self.archive_bucket_page_ids
}
#[must_use]
pub fn paged_checkpoint_hot_state(&self) -> Self {
let mut hot = self.clone();
hot.archive_receipt_buckets.clear();
hot.archive_bucket_page_ids.clear();
hot.archive_receipt_count = 0;
hot
}
#[must_use]
pub fn archived_history_count(&self) -> usize {
usize::try_from(self.archive_receipt_count).unwrap_or(usize::MAX)
}
#[must_use]
pub fn controller(&self, id: &str) -> Option<&DecisionControllerBinding> {
self.controllers.get(id).map(Arc::as_ref)
}
#[must_use]
pub fn ticket(&self, id: DecisionTicketId) -> Option<&DecisionTicket> {
self.tickets.get(&id).map(Arc::as_ref)
}
#[must_use]
pub fn trace(&self, id: DecisionTraceId) -> Option<&DecisionTrace> {
self.traces.get_ordinal(id.get())
}
#[must_use]
pub fn attempt(&self, id: canwu_core::DecisionRequestId) -> Option<&DecisionAttemptRecord> {
self.attempts_by_request
.get(&id)
.and_then(|ordinal| self.attempts.get_ordinal(*ordinal))
}
#[must_use]
pub const fn attempts(&self) -> &PersistentDecisionLog<DecisionAttemptRecord> {
&self.attempts
}
#[must_use]
pub fn decision_locator(&self, key: &DecisionHistoryKey) -> DecisionHistoryLocation {
let hot = match key {
DecisionHistoryKey::Ticket(id) => self.tickets.contains_key(id),
DecisionHistoryKey::Attempt(id) => self.attempts_by_request.contains_key(id),
DecisionHistoryKey::Trace(id) => self.traces.get_ordinal(id.get()).is_some(),
};
if hot {
DecisionHistoryLocation::Hot
} else if let Some(receipt) = self.compact_archive_receipt(key) {
DecisionHistoryLocation::Archived {
locator: encode_archive_hash(receipt.content_id),
}
} else if let Ok(page) = decision_history_page_key(key)
&& self.archive_bucket_page_ids.contains_key(&page)
{
DecisionHistoryLocation::Unresolved {
bucket: page.bucket,
segment: page.segment,
}
} else {
DecisionHistoryLocation::Absent
}
}
pub fn decision_locator_with_provider(
&self,
key: &DecisionHistoryKey,
provider: &dyn DecisionArchiveProvider,
) -> Result<DecisionHistoryLocation, DecisionError> {
let local = self.decision_locator(key);
match local {
DecisionHistoryLocation::Hot
| DecisionHistoryLocation::Archived { .. }
| DecisionHistoryLocation::Absent => return Ok(local),
DecisionHistoryLocation::Unresolved { .. } => {}
}
let bucket = decision_history_page_key(key)?;
let Some(page_id) = self.archive_bucket_page_ids.get(&bucket) else {
return Ok(DecisionHistoryLocation::Absent);
};
let page = provider
.load_decision_archive_bucket_page(page_id)?
.ok_or_else(|| history_unavailable("decision locator bucket page is unavailable"))?;
page.validate()?;
if page.bucket != bucket.bucket
|| page.segment != bucket.segment
|| page.state_page_id()? != *page_id
{
return Err(history_unavailable(
"decision locator provider returned a mismatched bucket page",
));
}
match page
.receipts
.binary_search_by(|receipt| receipt.key.cmp(key))
{
Ok(index) => Ok(DecisionHistoryLocation::Archived {
locator: page.receipts[index].locator.clone(),
}),
Err(_) => Ok(DecisionHistoryLocation::Absent),
}
}
#[must_use]
pub fn archive_receipt(&self, key: &DecisionHistoryKey) -> Option<DecisionArchiveReceipt> {
self.compact_archive_receipt(key)
.map(|receipt| receipt.to_receipt(key))
}
fn archive_receipt_with_provider(
&self,
key: &DecisionHistoryKey,
provider: &dyn DecisionArchiveProvider,
) -> Result<Option<DecisionArchiveReceipt>, DecisionError> {
if let Some(receipt) = self.archive_receipt(key) {
return Ok(Some(receipt));
}
let bucket = decision_history_page_key(key)?;
let Some(page_id) = self.archive_bucket_page_ids.get(&bucket) else {
return Ok(None);
};
let page = provider
.load_decision_archive_bucket_page(page_id)?
.ok_or_else(|| history_unavailable("decision locator bucket page is unavailable"))?;
page.validate()?;
if page.bucket != bucket.bucket
|| page.segment != bucket.segment
|| page.state_page_id()? != *page_id
{
return Err(history_unavailable(
"decision locator provider returned a mismatched bucket page",
));
}
Ok(page
.receipts
.binary_search_by(|receipt| receipt.key.cmp(key))
.ok()
.map(|index| page.receipts[index].clone()))
}
pub fn archived_history_keys(&self) -> impl Iterator<Item = &DecisionHistoryKey> {
self.archive_receipt_buckets.values().flat_map(OrdMap::keys)
}
pub fn required_archived_dependency_page_keys(
&self,
) -> Result<OrdSet<DecisionArchivePageKey>, DecisionError> {
let mut pages = OrdSet::new();
for trace in self.traces.iter() {
if !self.tickets.contains_key(&trace.ticket_id) {
pages.insert(decision_history_page_key(&DecisionHistoryKey::Ticket(
trace.ticket_id,
))?);
}
}
for ticket in self.tickets.values() {
if let DecisionTicketState::Resolved { trace_id, .. } = ticket.state
&& self.traces.get_ordinal(trace_id.get()).is_none()
{
pages.insert(decision_history_page_key(&DecisionHistoryKey::Trace(
trace_id,
))?);
}
if let Some(parent_id) = ticket.parent_ticket
&& !self.tickets.contains_key(&parent_id)
{
pages.insert(decision_history_page_key(&DecisionHistoryKey::Ticket(
parent_id,
))?);
}
}
for attempt in self.attempts.iter() {
if let DecisionAttemptOutcome::Accepted {
trace_id: Some(trace_id),
..
} = &attempt.outcome
&& self.traces.get_ordinal(trace_id.get()).is_none()
{
pages.insert(decision_history_page_key(&DecisionHistoryKey::Trace(
*trace_id,
))?);
}
}
Ok(pages)
}
pub fn paged_checkpoint_parts(
&self,
) -> Result<
(
Self,
std::collections::BTreeMap<DecisionArchivePageKey, Vec<DecisionArchiveReceipt>>,
),
DecisionError,
> {
let hot = self.paged_checkpoint_hot_state();
let mut buckets =
std::collections::BTreeMap::<DecisionArchivePageKey, Vec<DecisionArchiveReceipt>>::new(
);
for (bucket, receipts) in &self.archive_receipt_buckets {
let decoded = receipts
.iter()
.map(|(key, receipt)| receipt.to_receipt(key))
.collect();
buckets.insert(*bucket, decoded);
}
Ok((hot, buckets))
}
pub fn from_paged_checkpoint_parts(
mut hot: Self,
buckets: impl IntoIterator<Item = (DecisionArchivePageKey, Vec<DecisionArchiveReceipt>)>,
) -> Result<Self, DecisionError> {
if !hot.archive_receipt_buckets.is_empty() {
return Err(archive_error(
"paged decision hot state must not duplicate archive receipts",
));
}
hot.archive_bucket_page_ids.clear();
hot.archive_receipt_count = 0;
for (bucket, receipts) in buckets {
if bucket.bucket >= DECISION_HISTORY_BUCKET_COUNT
|| receipts.windows(2).any(|pair| pair[0].key >= pair[1].key)
{
return Err(archive_error(
"paged decision archive bucket is out of range or not strictly ordered",
));
}
for receipt in receipts {
if decision_history_page_key(&receipt.key)? != bucket
|| hot.insert_archive_receipt(&receipt)?.is_some()
{
return Err(archive_error(
"paged decision archive bucket contains a misplaced or duplicate receipt",
));
}
}
}
hot.validate()?;
Ok(hot)
}
pub fn from_paged_checkpoint_root(
hot: Self,
archive_bucket_page_ids: OrdMap<DecisionArchivePageKey, String>,
archive_receipt_count: u64,
archive_receipt_root: &str,
) -> Result<Self, DecisionError> {
Self::from_paged_checkpoint_root_with_resident_pages(
hot,
archive_bucket_page_ids,
archive_receipt_count,
archive_receipt_root,
std::iter::empty(),
)
}
pub fn from_paged_checkpoint_root_with_resident_pages(
mut hot: Self,
archive_bucket_page_ids: OrdMap<DecisionArchivePageKey, String>,
archive_receipt_count: u64,
archive_receipt_root: &str,
resident_pages: impl IntoIterator<Item = DecisionArchiveBucketPage>,
) -> Result<Self, DecisionError> {
if !hot.archive_receipt_buckets.is_empty()
|| !hot.archive_bucket_page_ids.is_empty()
|| hot.archive_receipt_count != 0
{
return Err(archive_error(
"paged decision hot state must not contain archive receipts or archive metadata",
));
}
hot.archive_bucket_page_ids = archive_bucket_page_ids;
hot.archive_receipt_count = archive_receipt_count;
for page in resident_pages {
page.validate()?;
let page_key = DecisionArchivePageKey {
bucket: page.bucket,
segment: page.segment,
};
let page_id = page.state_page_id()?;
if hot.archive_bucket_page_ids.get(&page_key) != Some(&page_id)
|| hot.archive_receipt_buckets.contains_key(&page_key)
{
return Err(archive_error(
"resident decision archive page is absent from or disagrees with the committed directory",
));
}
let mut receipts = OrdMap::new();
for receipt in page.receipts {
if receipts
.insert(
receipt.key.clone(),
CompactDecisionArchiveReceipt::from_receipt(&receipt)?,
)
.is_some()
{
return Err(archive_error(
"resident decision archive page contains duplicate receipts",
));
}
}
hot.archive_receipt_buckets.insert(page_key, receipts);
}
hot.validate()?;
if hot.archive_receipt_root()? != archive_receipt_root {
return Err(archive_error(
"paged decision archive directory root is inconsistent",
));
}
Ok(hot)
}
pub fn archive_receipt_commitment(&self) -> Result<String, DecisionError> {
self.archive_receipt_root()
}
pub fn authoritative_commitment(&self) -> Result<String, DecisionError> {
decision_archive_hash(
"canwu.commitment.decisions.v2",
&(
&self.controllers,
self.hot_history_root()?,
self.archive_receipt_root()?,
),
)
}
pub fn archived_decision_history_page(
&self,
cursor: Option<&DecisionHistoryCursor>,
budget: DecisionHistoryQueryBudget,
provider: &dyn DecisionArchiveProvider,
) -> Result<DecisionHistoryPage, DecisionError> {
if budget.max_results == 0
|| budget.max_results > MAX_DECISION_HISTORY_PAGE_SIZE
|| budget.max_provider_calls == 0
|| budget.max_decoded_bytes == 0
|| budget.max_decoded_bytes > MAX_DECISION_HISTORY_PAGE_BYTES
{
return Err(history_budget_error(
"decision history page budget is zero, inconsistent, or exceeds the hard limit",
));
}
let archive_root = self.archive_receipt_root()?;
if cursor.is_some_and(|cursor| cursor.archive_root != archive_root) {
return Err(history_unavailable(
"decision history cursor belongs to a different archive generation",
));
}
if cursor.is_some_and(|cursor| {
cursor.bucket >= DECISION_HISTORY_BUCKET_COUNT
|| decision_history_page_key(&cursor.after).ok()
!= Some(DecisionArchivePageKey {
bucket: cursor.bucket,
segment: cursor.segment,
})
}) {
return Err(history_unavailable(
"decision history cursor has an invalid bucket position",
));
}
let start_bucket = cursor.map_or(
DecisionArchivePageKey {
bucket: 0,
segment: 0,
},
|cursor| DecisionArchivePageKey {
bucket: cursor.bucket,
segment: cursor.segment,
},
);
let mut selected = Vec::<(DecisionArchivePageKey, DecisionArchiveReceipt)>::new();
let mut provider_calls = 0_usize;
for (bucket, page_id) in self.archive_bucket_page_ids.range(start_bucket..) {
let loaded;
let receipts = if let Some(receipts) = self.archive_receipt_buckets.get(bucket) {
receipts
.iter()
.map(|(key, receipt)| receipt.to_receipt(key))
.collect::<Vec<_>>()
} else {
provider_calls = provider_calls.checked_add(1).ok_or_else(|| {
history_budget_error("decision history provider-call count overflowed")
})?;
if provider_calls > budget.max_provider_calls {
return Err(history_budget_error(
"decision history page exceeded its provider-call budget",
));
}
loaded = provider
.load_decision_archive_bucket_page(page_id)?
.ok_or_else(|| {
history_unavailable("decision locator bucket page is unavailable")
})?;
loaded.validate()?;
if loaded.bucket != bucket.bucket
|| loaded.segment != bucket.segment
|| loaded.state_page_id()? != *page_id
{
return Err(history_unavailable(
"decision locator provider returned a mismatched bucket page",
));
}
loaded.receipts
};
if let Some(cursor) = cursor
.filter(|cursor| cursor.bucket == bucket.bucket && cursor.segment == bucket.segment)
{
let after = &cursor.after;
for receipt in receipts.iter().filter(|receipt| receipt.key > *after) {
selected.push((*bucket, receipt.clone()));
if selected.len() > budget.max_results {
break;
}
}
} else {
for receipt in receipts {
selected.push((*bucket, receipt));
if selected.len() > budget.max_results {
break;
}
}
}
if selected.len() > budget.max_results {
break;
}
}
let has_more = selected.len() > budget.max_results;
let selected = &selected[..selected.len().min(budget.max_results)];
let mut records = Vec::with_capacity(selected.len());
let mut decoded_bytes = 0_u64;
for (_, receipt) in selected {
provider_calls = provider_calls.checked_add(1).ok_or_else(|| {
history_budget_error("decision history provider-call count overflowed")
})?;
if provider_calls > budget.max_provider_calls {
return Err(history_budget_error(
"decision history page exceeded its provider-call budget",
));
}
let blob = provider
.load_decision_archive(&receipt.locator)?
.ok_or_else(|| {
history_unavailable("decision archive provider omitted a committed page member")
})?;
blob.validate()?;
if blob.key != receipt.key || blob.content_id()? != receipt.content_id {
return Err(history_unavailable(
"decision archive provider returned a mismatched page member",
));
}
decoded_bytes = decoded_bytes
.checked_add(receipt.encoded_bytes)
.ok_or_else(|| history_budget_error("decision history byte count overflowed"))?;
if decoded_bytes > budget.max_decoded_bytes {
return Err(history_budget_error(
"decision history page exceeded its decoded-byte budget",
));
}
records.push(blob.record);
}
let next_cursor = if has_more {
selected
.last()
.map(|(bucket, receipt)| DecisionHistoryCursor {
archive_root: archive_root.clone(),
bucket: bucket.bucket,
segment: bucket.segment,
after: receipt.key.clone(),
})
} else {
None
};
Ok(DecisionHistoryPage {
archive_root,
records,
next_cursor,
provider_calls: provider_calls as u64,
decoded_bytes,
})
}
pub fn archive_reachability(
&self,
provider: &dyn DecisionArchiveProvider,
) -> Result<DecisionArchiveReachability, DecisionError> {
let mut reachable = DecisionArchiveReachability::default();
let mut receipt_count = 0_u64;
for (bucket, page_id) in &self.archive_bucket_page_ids {
let page = provider
.load_decision_archive_bucket_page(page_id)?
.ok_or_else(|| {
history_unavailable("decision archive bucket page is unavailable")
})?;
page.validate()?;
if page.bucket != bucket.bucket
|| page.segment != bucket.segment
|| page.state_page_id()? != *page_id
{
return Err(history_unavailable(
"decision archive reachability provider returned a mismatched page",
));
}
reachable.bucket_page_ids.insert(page_id.clone());
for receipt in page.receipts {
receipt_count = receipt_count
.checked_add(1)
.ok_or_else(|| archive_error("decision archive receipt count overflowed"))?;
reachable.blob_locators.insert(receipt.locator);
}
}
if receipt_count != self.archive_receipt_count {
return Err(archive_error(
"decision archive reachability count disagrees with its directory",
));
}
Ok(reachable)
}
pub fn load_decision_history(
&self,
key: &DecisionHistoryKey,
provider: &dyn DecisionArchiveProvider,
) -> Result<Option<DecisionArchiveRecord>, DecisionError> {
if let Some(record) = self.hot_archive_record(key) {
return Ok(Some(record));
}
let Some(receipt) = self.archive_receipt_with_provider(key, provider)? else {
return Ok(None);
};
let blob = provider
.load_decision_archive(&receipt.locator)?
.ok_or_else(|| archive_error("decision archive provider omitted a committed blob"))?;
blob.validate()?;
let content_id = blob.content_id()?;
if blob.key != *key || content_id != receipt.content_id {
return Err(archive_error(
"decision archive provider returned content that disagrees with its receipt",
));
}
Ok(Some(blob.record))
}
pub fn prepare_decision_archive(
&self,
keys: &[DecisionHistoryKey],
) -> Result<PreparedDecisionArchive, DecisionError> {
if keys.is_empty() || keys.len() > MAX_DECISION_ARCHIVE_BATCH_ENTRIES {
return Err(archive_error(
"decision archive batch is empty or exceeds the bounded entry limit",
));
}
let mut canonical_keys = keys.to_vec();
canonical_keys.sort();
canonical_keys.dedup();
if canonical_keys.len() != keys.len() {
return Err(archive_error(
"decision archive batch contains duplicate keys",
));
}
let source_root = self.hot_history_root()?;
let mut blobs = Vec::with_capacity(canonical_keys.len());
let mut receipts = Vec::with_capacity(canonical_keys.len());
for key in canonical_keys {
if self.contains_archived_key(&key) {
return Err(archive_error("decision history is already archived"));
}
let record = self
.hot_archive_record(&key)
.ok_or_else(|| archive_error("decision archive key is not resident"))?;
self.validate_archive_eligibility(&record)?;
let blob = DecisionArchiveBlob {
format_version: DECISION_ARCHIVE_FORMAT_VERSION,
key: key.clone(),
record,
};
let content_id = blob.content_id()?;
let encoded_bytes = u64::try_from(
serde_json::to_vec(&blob)
.map_err(|error| {
archive_error(format!("cannot encode decision archive: {error}"))
})?
.len(),
)
.map_err(|_| {
archive_error("decision archive blob exceeds the persistent byte range")
})?;
receipts.push(DecisionArchiveReceipt {
format_version: DECISION_ARCHIVE_FORMAT_VERSION,
key,
locator: content_id.clone(),
content_id,
encoded_bytes,
});
blobs.push(blob);
}
let token = decision_archive_hash(
"canwu.decision.archive-token.v1",
&(&source_root, &receipts),
)?;
Ok(PreparedDecisionArchive {
source_root,
token,
blobs,
receipts,
})
}
pub fn commit_decision_archive(
&self,
prepared: &PreparedDecisionArchive,
provider: &dyn DecisionArchiveProvider,
) -> Result<Self, DecisionError> {
let verified = self.verify_decision_archive(prepared, provider)?;
self.commit_verified_decision_archive(&verified)
}
pub fn verify_decision_archive(
&self,
prepared: &PreparedDecisionArchive,
provider: &dyn DecisionArchiveProvider,
) -> Result<VerifiedDecisionArchiveCommit, DecisionError> {
let keys = prepared
.receipts
.iter()
.map(|receipt| receipt.key.clone())
.collect::<Vec<_>>();
let expected = self.prepare_decision_archive(&keys)?;
if expected != *prepared {
return Err(archive_error(
"prepared decision archive is stale or altered",
));
}
for (blob, receipt) in prepared.blobs.iter().zip(&prepared.receipts) {
let stored = provider
.load_decision_archive(&receipt.locator)?
.ok_or_else(|| {
archive_error("prepared decision archive is not durably readable")
})?;
if stored != *blob || stored.content_id()? != receipt.content_id {
return Err(archive_error(
"stored decision archive blob failed verification",
));
}
}
let mut additions = std::collections::BTreeMap::<
DecisionArchivePageKey,
Vec<&DecisionArchiveReceipt>,
>::new();
for receipt in &prepared.receipts {
additions
.entry(decision_history_page_key(&receipt.key)?)
.or_default()
.push(receipt);
}
let mut page_replacements = Vec::with_capacity(additions.len());
for (page_key, receipts) in additions {
let previous_page_id = self.archive_bucket_page_ids.get(&page_key).cloned();
let mut merged = if let Some(resident) = self.archive_receipt_buckets.get(&page_key) {
resident
.iter()
.map(|(key, receipt)| (key.clone(), receipt.to_receipt(key)))
.collect::<std::collections::BTreeMap<_, _>>()
} else if let Some(page_id) = previous_page_id.as_deref() {
let page = provider
.load_decision_archive_bucket_page(page_id)?
.ok_or_else(|| {
history_unavailable(
"decision archive locator page is unavailable for authenticated merge",
)
})?;
page.validate()?;
if page.bucket != page_key.bucket
|| page.segment != page_key.segment
|| page.state_page_id()? != page_id
{
return Err(history_unavailable(
"decision archive locator provider returned a mismatched merge page",
));
}
page.receipts
.into_iter()
.map(|receipt| (receipt.key.clone(), receipt))
.collect::<std::collections::BTreeMap<_, _>>()
} else {
std::collections::BTreeMap::new()
};
for receipt in receipts {
if merged
.insert(receipt.key.clone(), receipt.clone())
.is_some()
{
return Err(archive_error(
"decision archive locator merge would replace an existing receipt",
));
}
}
let page = DecisionArchiveBucketPage {
format_version: DECISION_ARCHIVE_BUCKET_PAGE_FORMAT_VERSION,
bucket: page_key.bucket,
segment: page_key.segment,
receipts: merged.into_values().collect(),
};
page.validate()?;
page_replacements.push(VerifiedDecisionArchivePageReplacement {
page_key,
previous_page_id,
page,
});
}
Ok(VerifiedDecisionArchiveCommit {
format_version: DECISION_ARCHIVE_FORMAT_VERSION,
source_root: prepared.source_root.clone(),
token: prepared.token.clone(),
receipts: prepared.receipts.clone(),
page_replacements,
})
}
pub fn commit_verified_decision_archive(
&self,
verified: &VerifiedDecisionArchiveCommit,
) -> Result<Self, DecisionError> {
if verified.format_version != DECISION_ARCHIVE_FORMAT_VERSION {
return Err(archive_error(
"verified decision archive uses an unsupported format",
));
}
let mut replacement_ids = std::collections::BTreeMap::new();
for replacement in &verified.page_replacements {
replacement.page.validate()?;
if replacement.page_key.bucket != replacement.page.bucket
|| replacement.page_key.segment != replacement.page.segment
|| replacement_ids
.insert(replacement.page_key, replacement.page.state_page_id()?)
.is_some()
|| replacement
.previous_page_id
.as_ref()
.is_some_and(|page_id| !canonical_archive_hash(page_id))
{
return Err(archive_error(
"verified decision archive page replacement is malformed",
));
}
}
let mut receipts_by_page = std::collections::BTreeMap::<
DecisionArchivePageKey,
Vec<&DecisionArchiveReceipt>,
>::new();
for receipt in &verified.receipts {
receipts_by_page
.entry(decision_history_page_key(&receipt.key)?)
.or_default()
.push(receipt);
}
if receipts_by_page.len() != replacement_ids.len()
|| receipts_by_page
.keys()
.any(|page_key| !replacement_ids.contains_key(page_key))
{
return Err(archive_error(
"verified decision archive does not replace every touched locator page",
));
}
let already_applied = verified.receipts.iter().all(|receipt| {
!matches!(
self.decision_locator(&receipt.key),
DecisionHistoryLocation::Hot
)
}) && replacement_ids
.iter()
.all(|(page_key, page_id)| self.archive_bucket_page_ids.get(page_key) == Some(page_id));
if already_applied {
return Ok(self.clone());
}
for replacement in &verified.page_replacements {
if self.archive_bucket_page_ids.get(&replacement.page_key)
!= replacement.previous_page_id.as_ref()
{
return Err(archive_error(
"verified decision archive locator base changed before commit",
));
}
let receipts = receipts_by_page.get(&replacement.page_key).ok_or_else(|| {
archive_error("verified decision archive locator page has no touched receipts")
})?;
for receipt in receipts {
if replacement
.page
.receipts
.binary_search_by(|candidate| candidate.key.cmp(&receipt.key))
.ok()
.is_none_or(|index| replacement.page.receipts[index] != **receipt)
{
return Err(archive_error(
"verified decision archive locator page omits a touched receipt",
));
}
}
}
let keys = verified
.receipts
.iter()
.map(|receipt| receipt.key.clone())
.collect::<Vec<_>>();
let expected = self.prepare_decision_archive(&keys)?;
if expected.source_root != verified.source_root
|| expected.token != verified.token
|| expected.receipts != verified.receipts
{
return Err(archive_error(
"verified decision archive is stale or altered",
));
}
let mut next = self.clone();
for receipt in &verified.receipts {
let hot_record = next
.hot_archive_record(&receipt.key)
.ok_or_else(|| archive_error("decision history disappeared during commit"))?;
next.remove_hot_history_record(&hot_record)?;
match receipt.key {
DecisionHistoryKey::Ticket(id) => {
next.tickets.remove(&id);
}
DecisionHistoryKey::Attempt(id) => {
let ordinal = next.attempts_by_request.remove(&id).ok_or_else(|| {
archive_error("decision attempt disappeared during commit")
})?;
next.attempts.remove_ordinal(ordinal);
}
DecisionHistoryKey::Trace(id) => {
next.traces.remove_ordinal(id.get());
}
}
}
next.archive_receipt_count = next
.archive_receipt_count
.checked_add(verified.receipts.len() as u64)
.ok_or_else(|| archive_error("decision archive receipt count overflowed"))?;
for replacement in &verified.page_replacements {
let receipts = replacement
.page
.receipts
.iter()
.map(|receipt| {
Ok((
receipt.key.clone(),
CompactDecisionArchiveReceipt::from_receipt(receipt)?,
))
})
.collect::<Result<OrdMap<_, _>, DecisionError>>()?;
next.archive_receipt_buckets
.insert(replacement.page_key, receipts);
next.archive_bucket_page_ids
.insert(replacement.page_key, replacement.page.state_page_id()?);
}
next.validate_archive_directory_shape()?;
for receipt in &verified.receipts {
if matches!(
next.decision_locator(&receipt.key),
DecisionHistoryLocation::Hot
) {
return Err(archive_error(
"decision archive commit left a touched key in hot state",
));
}
}
Ok(next)
}
fn hot_archive_record(&self, key: &DecisionHistoryKey) -> Option<DecisionArchiveRecord> {
match key {
DecisionHistoryKey::Ticket(id) => self
.tickets
.get(id)
.map(|ticket| ticket.as_ref().clone())
.map(|ticket| DecisionArchiveRecord::Ticket { ticket }),
DecisionHistoryKey::Attempt(id) => self
.attempt(*id)
.cloned()
.map(|attempt| DecisionArchiveRecord::Attempt { attempt }),
DecisionHistoryKey::Trace(id) => self
.traces
.get_ordinal(id.get())
.cloned()
.map(|trace| DecisionArchiveRecord::Trace { trace }),
}
}
fn validate_archive_eligibility(
&self,
record: &DecisionArchiveRecord,
) -> Result<(), DecisionError> {
match record {
DecisionArchiveRecord::Ticket { ticket } if ticket.is_open() => Err(archive_error(
"open decision tickets cannot leave the hot mutation path",
)),
DecisionArchiveRecord::Trace { trace } => {
let terminal = self
.tickets
.get(&trace.ticket_id)
.is_some_and(|ticket| !ticket.is_open())
|| self.contains_archived_key(&DecisionHistoryKey::Ticket(trace.ticket_id));
if terminal {
Ok(())
} else {
Err(archive_error(
"decision traces can archive only after their ticket is terminal",
))
}
}
DecisionArchiveRecord::Ticket { .. } | DecisionArchiveRecord::Attempt { .. } => Ok(()),
}
}
fn hot_history_root(&self) -> Result<String, DecisionError> {
let resident_count = (self.tickets.len() as u64)
.checked_add(self.traces.len() as u64)
.and_then(|count| count.checked_add(self.attempts.len() as u64))
.ok_or_else(|| archive_error("decision hot-history count overflowed"))?;
if resident_count != self.hot_history_accumulator.count {
return Err(archive_error(
"decision hot-history accumulator count is inconsistent",
));
}
decision_archive_hash(
"canwu.decision.hot-history-root.v2",
&(
self.hot_history_accumulator.count,
self.hot_history_accumulator.xor,
self.hot_history_accumulator.sum,
),
)
}
pub fn hot_history_commitment(&self) -> Result<String, DecisionError> {
self.hot_history_root()
}
fn archive_receipt_root(&self) -> Result<String, DecisionError> {
let mut hasher = blake3::Hasher::new();
hasher.update(b"canwu.decision.archive-receipt-root.v3");
hasher.update(&[0]);
hasher.update(&self.archive_receipt_count.to_be_bytes());
hasher.update(&(self.archive_bucket_page_ids.len() as u64).to_be_bytes());
for (bucket, page_id) in &self.archive_bucket_page_ids {
hasher.update(&bucket.bucket.to_be_bytes());
hasher.update(&[bucket.segment]);
hasher.update(&decode_archive_hash(page_id)?);
}
Ok(hasher.finalize().to_hex().to_string())
}
#[must_use]
pub fn decision_hot_state(&self) -> DecisionHotState {
DecisionHotState {
ticket_count: self.tickets.len() as u64,
attempt_count: self.attempts.len() as u64,
trace_count: self.traces.len() as u64,
}
}
pub fn append_attempt(&mut self, attempt: DecisionAttemptRecord) -> Result<(), DecisionError> {
validate_attempt_shape(self, &attempt)?;
if self.attempts_by_request.contains_key(&attempt.request_id) {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
"decision attempts must use unique request IDs",
));
}
let request_id = attempt.request_id;
self.insert_hot_history_record(&DecisionArchiveRecord::Attempt {
attempt: attempt.clone(),
})?;
let ordinal = self.attempts.push(attempt);
self.attempts_by_request.insert(request_id, ordinal);
Ok(())
}
pub fn open_tickets(&self) -> impl Iterator<Item = &DecisionTicket> {
self.tickets
.values()
.map(Arc::as_ref)
.filter(|ticket| ticket.is_open())
}
pub fn validate(&self) -> Result<(), DecisionError> {
self.validate_archive_directory_shape()?;
if self.attempts_by_request.len() != self.attempts.len() {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
"decision attempt request index is inconsistent",
));
}
let mut expected_deadlines = OrdMap::<SimTime, OrdSet<DecisionTicketId>>::new();
for ticket in self.tickets.values().filter(|ticket| ticket.is_open()) {
if let Some(deadline) = ticket.deadline {
expected_deadlines
.entry(deadline)
.or_default()
.insert(ticket.id);
}
}
if expected_deadlines != self.deadline_index {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
"decision deadline index is inconsistent",
));
}
for (id, controller) in &self.controllers {
if id != &controller.id {
return Err(DecisionError::new(
DecisionErrorCode::InvalidController,
"controller map key does not match its persisted identity",
));
}
controller.validate()?;
}
for (bucket, receipts) in &self.archive_receipt_buckets {
if bucket.bucket >= DECISION_HISTORY_BUCKET_COUNT || receipts.is_empty() {
return Err(archive_error("decision archive receipt bucket is invalid"));
}
for (key, receipt) in receipts {
if decision_history_page_key(key)? != *bucket
|| receipt.encoded_bytes == 0
|| matches!(self.decision_locator(key), DecisionHistoryLocation::Hot)
{
return Err(archive_error(
"decision archive receipt is invalid or overlaps hot history",
));
}
}
let expected_page_id = self
.decision_archive_bucket_page(*bucket)?
.ok_or_else(|| archive_error("nonempty archive bucket has no page"))?
.state_page_id()?;
if self.archive_bucket_page_ids.get(bucket) != Some(&expected_page_id) {
return Err(archive_error(
"decision archive bucket page commitment is inconsistent",
));
}
}
if self.attempts.iter().any(|attempt| {
attempt.request_commitment.len() != 64
|| !attempt
.request_commitment
.bytes()
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
}) {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
"decision attempts require a canonical request commitment",
));
}
for (id, ticket) in &self.tickets {
if id != &ticket.id || !self.controllers.contains_key(&ticket.assigned_controller) {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
"ticket identity or assigned controller is invalid",
));
}
ticket.validate()?;
if let Some(parent_id) = ticket.parent_ticket {
let lineage_valid = match self.tickets.get(&parent_id) {
Some(parent) => {
!parent.is_open()
&& self.lineage_links(
parent,
&ticket.decision_maker,
&ticket.assigned_controller,
)
&& parent.updated_at <= ticket.opened_at
}
None => self.contains_archived_key(&DecisionHistoryKey::Ticket(parent_id)),
};
if !lineage_valid {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
"decision ticket lineage references an open, foreign, later, or unavailable parent",
));
}
}
}
for (ordinal, trace) in self.traces.ordinals().zip(self.traces.iter()) {
if trace.id.get() != ordinal || ordinal >= self.traces.next_ordinal() {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
"decision trace identity disagrees with its persistent log ordinal",
));
}
let ticket_version_valid = self.tickets.get(&trace.ticket_id).is_some_and(|ticket| {
trace.ticket_version != 0
&& trace.ticket_version <= ticket.version
&& trace.parent_ticket == ticket.parent_ticket
}) || self
.contains_archived_key(&DecisionHistoryKey::Ticket(trace.ticket_id));
if !ticket_version_valid || !self.controllers.contains_key(&trace.controller_id) {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
"decision trace ticket, version, or controller is invalid",
));
}
}
for ticket in self.tickets.values() {
if let DecisionTicketState::Resolved { trace_id, .. } = ticket.state {
let hot_trace_matches = self
.traces
.get_ordinal(trace_id.get())
.is_some_and(|trace| trace.id == trace_id && trace.ticket_id == ticket.id);
let archived_trace_exists =
self.contains_archived_key(&DecisionHistoryKey::Trace(trace_id));
if !hot_trace_matches && !archived_trace_exists {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
"resolved ticket does not reference hot or archived trace evidence",
));
}
}
}
let mut request_ids = std::collections::BTreeSet::new();
for (ordinal, attempt) in self.attempts.ordinals().zip(self.attempts.iter()) {
if attempt.request_id.get() == 0 || !request_ids.insert(attempt.request_id) {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
"decision attempts must use unique nonzero request IDs",
));
}
if self.attempts_by_request.get(&attempt.request_id) != Some(&ordinal) {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
"decision attempt request index is inconsistent",
));
}
match &attempt.outcome {
DecisionAttemptOutcome::Accepted {
trace_id,
command_request_id,
} => {
if command_request_id.is_some() && trace_id.is_none() {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
"accepted decision commands require a decision trace",
));
}
if trace_id.is_some_and(|trace_id| {
self.traces.get_ordinal(trace_id.get()).is_none()
&& !self.contains_archived_key(&DecisionHistoryKey::Trace(trace_id))
}) {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
"accepted decision attempt references unavailable trace evidence",
));
}
}
DecisionAttemptOutcome::Rejected { message, .. } => {
require_text(message, "decision rejection message")?;
}
}
}
Ok(())
}
fn validate_archive_directory_shape(&self) -> Result<(), DecisionError> {
if self
.archive_bucket_page_ids
.iter()
.any(|(bucket, page_id)| {
bucket.bucket >= DECISION_HISTORY_BUCKET_COUNT || !canonical_archive_hash(page_id)
})
{
return Err(archive_error(
"decision archive bucket directory is malformed",
));
}
if (self.archive_receipt_count == 0) != self.archive_bucket_page_ids.is_empty() {
return Err(archive_error(
"decision archive receipt count and page directory disagree",
));
}
if !self.archive_receipt_buckets.is_empty() {
let resident_count =
self.archive_receipt_buckets
.values()
.try_fold(0_u64, |total, receipts| {
total.checked_add(receipts.len() as u64).ok_or_else(|| {
archive_error("decision archive receipt count overflowed")
})
})?;
if resident_count > self.archive_receipt_count {
return Err(archive_error(
"resident decision archive receipt count exceeds the committed total",
));
}
}
Ok(())
}
pub fn apply(
&mut self,
mutation: DecisionMutation,
at: SimTime,
trace_id: Option<DecisionTraceId>,
) -> Result<PreparedDecision, DecisionError> {
let prepared = match mutation {
DecisionMutation::RegisterController { controller } => {
controller.validate()?;
if self.controllers.contains_key(&controller.id) {
return Err(DecisionError::new(
DecisionErrorCode::DuplicateController,
format!(
"decision controller {} is already registered",
controller.id
),
));
}
self.controllers
.insert(controller.id.clone(), Arc::new(controller));
PreparedDecision::default()
}
DecisionMutation::Open { mut ticket } => {
ticket.validate()?;
if self.tickets.contains_key(&ticket.id) {
return Err(DecisionError::new(
DecisionErrorCode::DuplicateTicket,
format!("decision ticket {} is already present", ticket.id),
));
}
if !self.controllers.contains_key(&ticket.assigned_controller) {
return Err(DecisionError::new(
DecisionErrorCode::InvalidController,
format!(
"decision ticket {} names unknown controller {}",
ticket.id, ticket.assigned_controller
),
));
}
if ticket.deadline.is_some_and(|deadline| deadline < at) {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
"decision deadline precedes its admission time",
));
}
if let Some(parent_id) = ticket.parent_ticket {
self.validate_parent_admission(
parent_id,
&ticket.decision_maker,
&ticket.assigned_controller,
)?;
}
let persisted = DecisionTicket {
id: ticket.id,
definition: ticket.definition,
decision_maker: ticket.decision_maker,
assigned_controller: ticket.assigned_controller,
summary: ticket.summary,
context: ticket.context,
options: std::mem::take(&mut ticket.options),
opened_at: at,
updated_at: at,
deadline: ticket.deadline,
version: 1,
state: DecisionTicketState::Open,
parent_ticket: ticket.parent_ticket,
};
if let Some(deadline) = persisted.deadline {
self.deadline_index
.entry(deadline)
.or_default()
.insert(persisted.id);
}
self.insert_hot_history_record(&DecisionArchiveRecord::Ticket {
ticket: persisted.clone(),
})?;
self.tickets.insert(persisted.id, Arc::new(persisted));
PreparedDecision::default()
}
DecisionMutation::ReplaceOptions {
ticket_id,
expected_version,
context,
mut options,
} => {
context.validate()?;
canonicalize_options(&mut options)?;
let previous = self
.tickets
.get(&ticket_id)
.map(|ticket| ticket.as_ref().clone())
.ok_or_else(|| {
DecisionError::new(
DecisionErrorCode::TicketNotFound,
format!("decision ticket {ticket_id} was not found"),
)
})?;
let ticket = self.open_ticket_mut(ticket_id, expected_version, at)?;
ticket.context = context;
ticket.options = options;
ticket.updated_at = at;
ticket.version = ticket.version.checked_add(1).ok_or_else(|| {
DecisionError::new(
DecisionErrorCode::InvalidDecision,
"decision ticket version is exhausted",
)
})?;
ticket.validate()?;
let updated = ticket.clone();
let _ = ticket;
self.replace_hot_history_record(
&DecisionArchiveRecord::Ticket { ticket: previous },
&DecisionArchiveRecord::Ticket { ticket: updated },
)?;
PreparedDecision::default()
}
DecisionMutation::Resolve {
ticket_id,
expected_version,
controller_id,
policy,
decision,
command_request_id,
} => self.resolve(
ticket_id,
expected_version,
&controller_id,
policy,
decision,
command_request_id,
at,
trace_id.ok_or_else(|| {
DecisionError::new(
DecisionErrorCode::InvalidDecision,
"decision resolution requires a claimed trace ID",
)
})?,
)?,
DecisionMutation::Cancel {
ticket_id,
expected_version,
reason,
} => {
require_text(&reason, "decision cancellation reason")?;
let previous = self
.tickets
.get(&ticket_id)
.map(|ticket| ticket.as_ref().clone())
.ok_or_else(|| {
DecisionError::new(
DecisionErrorCode::TicketNotFound,
format!("decision ticket {ticket_id} was not found"),
)
})?;
let ticket = self.open_ticket_mut(ticket_id, expected_version, at)?;
ticket.updated_at = at;
ticket.version = ticket.version.checked_add(1).ok_or_else(|| {
DecisionError::new(
DecisionErrorCode::InvalidDecision,
"decision ticket version is exhausted",
)
})?;
ticket.state = DecisionTicketState::Cancelled { reason };
ticket.validate()?;
let deadline = ticket.deadline;
let updated = ticket.clone();
let _ = ticket;
self.replace_hot_history_record(
&DecisionArchiveRecord::Ticket { ticket: previous },
&DecisionArchiveRecord::Ticket { ticket: updated },
)?;
self.remove_deadline(ticket_id, deadline);
PreparedDecision::default()
}
};
self.advance_time(at)?;
Ok(prepared)
}
fn lineage_links(
&self,
parent: &DecisionTicket,
decision_maker: &canwu_core::EntityRef,
assigned_controller: &str,
) -> bool {
if &parent.decision_maker == decision_maker {
return true;
}
let seat = |controller: &str| {
self.controllers
.get(controller)
.and_then(|binding| binding.seat_id.as_deref())
};
seat(&parent.assigned_controller)
.is_some_and(|parent_seat| seat(assigned_controller) == Some(parent_seat))
}
fn validate_parent_admission(
&self,
parent_id: DecisionTicketId,
decision_maker: &canwu_core::EntityRef,
assigned_controller: &str,
) -> Result<(), DecisionError> {
let Some(parent) = self.ticket(parent_id) else {
return Err(DecisionError::new(
DecisionErrorCode::TicketNotFound,
format!("decision ticket parent {parent_id} is not in hot decision history"),
));
};
if parent.is_open() {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
format!("decision ticket parent {parent_id} is still open"),
));
}
if !self.lineage_links(parent, decision_maker, assigned_controller) {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
format!(
"decision ticket parent {parent_id} belongs to a different decision maker and controller seat"
),
));
}
Ok(())
}
fn open_ticket_mut(
&mut self,
ticket_id: DecisionTicketId,
expected_version: u64,
at: SimTime,
) -> Result<&mut DecisionTicket, DecisionError> {
let ticket = self.tickets.get_mut(&ticket_id).ok_or_else(|| {
DecisionError::new(
DecisionErrorCode::TicketNotFound,
format!("decision ticket {ticket_id} was not found"),
)
})?;
let ticket = Arc::make_mut(ticket);
if !ticket.is_open() || ticket.deadline.is_some_and(|deadline| deadline < at) {
return Err(DecisionError::new(
DecisionErrorCode::ClosedTicket,
format!("decision ticket {ticket_id} is not open"),
));
}
if ticket.version != expected_version {
return Err(DecisionError::new(
DecisionErrorCode::VersionConflict,
format!(
"decision ticket {ticket_id} is at version {}, expected {expected_version}",
ticket.version
),
));
}
Ok(ticket)
}
#[allow(clippy::too_many_arguments)]
fn resolve(
&mut self,
ticket_id: DecisionTicketId,
expected_version: u64,
controller_id: &str,
policy: DecisionPolicyIdentity,
decision: PolicyDecision,
command_request_id: Option<CommandRequestId>,
at: SimTime,
trace_id: DecisionTraceId,
) -> Result<PreparedDecision, DecisionError> {
let controller = self.controllers.get(controller_id).ok_or_else(|| {
DecisionError::new(
DecisionErrorCode::InvalidController,
format!("decision controller {controller_id} was not found"),
)
})?;
if controller.policy != policy {
return Err(DecisionError::new(
DecisionErrorCode::PolicyMismatch,
"decision resolution policy does not match the persisted controller binding",
));
}
let random_evidence_matches_policy = match controller.policy.kind {
crate::DecisionPolicyKind::Random => {
decision.random.is_some() && decision.stage.is_none()
}
crate::DecisionPolicyKind::Utility => {
decision.random.is_some() == decision.is_random_tie_break()
&& (controller.random_tie_break || decision.random.is_none())
}
crate::DecisionPolicyKind::Rule
| crate::DecisionPolicyKind::Human
| crate::DecisionPolicyKind::External
| crate::DecisionPolicyKind::Llm => {
decision.random.is_none() && !decision.is_random_tie_break()
}
};
if !random_evidence_matches_policy {
return Err(DecisionError::new(
DecisionErrorCode::PolicyMismatch,
"random decision controllers require random draw evidence, utility controllers accept it only for a random tie-break, and other controllers reject it",
));
}
let previous_ticket = self
.tickets
.get(&ticket_id)
.map(|ticket| ticket.as_ref().clone())
.ok_or_else(|| {
DecisionError::new(
DecisionErrorCode::TicketNotFound,
format!("decision ticket {ticket_id} was not found"),
)
})?;
let ticket = self.open_ticket_mut(ticket_id, expected_version, at)?;
if ticket.assigned_controller != controller_id {
return Err(DecisionError::new(
DecisionErrorCode::InvalidController,
"decision resolution came from a controller not assigned to the ticket",
));
}
decision.validate(ticket)?;
let action = match &decision.outcome {
DecisionOutcome::Selected { option_id } => {
ticket.option(option_id).map(|option| option.action.clone())
}
DecisionOutcome::Deferred { .. } => None,
DecisionOutcome::Pending { .. } | DecisionOutcome::PendingRandom { .. } => {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
"pending policy outcomes are not authoritative decision mutations",
));
}
};
if matches!(action, Some(DecisionAction::Command { .. })) != command_request_id.is_some() {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
"command actions require exactly one command request ID",
));
}
let trace = DecisionTrace {
id: trace_id,
ticket_id,
ticket_version: ticket.version,
controller_id: controller_id.to_owned(),
policy,
decided_at: at,
outcome: decision.outcome.clone(),
summary: decision.summary,
evaluations: decision.evaluations,
external: decision.external,
random: decision.random,
command_request_id,
stage: decision.stage,
fired_guards: decision.fired_guards,
parent_ticket: ticket.parent_ticket,
};
ticket.updated_at = at;
ticket.version = ticket.version.checked_add(1).ok_or_else(|| {
DecisionError::new(
DecisionErrorCode::InvalidDecision,
"decision ticket version is exhausted",
)
})?;
if let DecisionOutcome::Selected { option_id } = &trace.outcome {
ticket.state = DecisionTicketState::Resolved {
option_id: option_id.clone(),
trace_id,
};
}
ticket.validate()?;
let deadline = ticket.deadline;
let updated_ticket = ticket.clone();
let _ = ticket;
self.replace_hot_history_record(
&DecisionArchiveRecord::Ticket {
ticket: previous_ticket,
},
&DecisionArchiveRecord::Ticket {
ticket: updated_ticket,
},
)?;
self.remove_deadline(ticket_id, deadline);
self.insert_hot_history_record(&DecisionArchiveRecord::Trace {
trace: trace.clone(),
})?;
let ordinal = self.traces.push(trace.clone());
debug_assert_eq!(ordinal, trace.id.get());
Ok(PreparedDecision {
trace: Some(trace),
action,
})
}
pub fn advance_time(&mut self, at: SimTime) -> Result<(), DecisionError> {
let due = self
.deadline_index
.range(..at)
.map(|(deadline, tickets)| (*deadline, tickets.iter().copied().collect::<Vec<_>>()))
.collect::<Vec<_>>();
for (deadline, ticket_ids) in due {
self.deadline_index.remove(&deadline);
for ticket_id in ticket_ids {
let previous = self
.tickets
.get(&ticket_id)
.map(|ticket| ticket.as_ref().clone())
.ok_or_else(|| {
DecisionError::new(
DecisionErrorCode::InvalidDecision,
"decision ticket index changed during time advancement",
)
})?;
let Some(ticket) = self.tickets.get_mut(&ticket_id) else {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
"decision ticket index changed during time advancement",
));
};
let ticket = Arc::make_mut(ticket);
let updated =
if ticket.is_open() && ticket.deadline.is_some_and(|deadline| deadline < at) {
ticket.updated_at = at;
ticket.version = ticket.version.checked_add(1).ok_or_else(|| {
DecisionError::new(
DecisionErrorCode::InvalidDecision,
"decision ticket version is exhausted",
)
})?;
ticket.state = DecisionTicketState::Expired;
ticket.validate()?;
Some(ticket.clone())
} else {
None
};
let _ = ticket;
if let Some(updated) = updated {
self.replace_hot_history_record(
&DecisionArchiveRecord::Ticket { ticket: previous },
&DecisionArchiveRecord::Ticket { ticket: updated },
)?;
}
}
}
Ok(())
}
fn remove_deadline(&mut self, ticket_id: DecisionTicketId, deadline: Option<SimTime>) {
let Some(deadline) = deadline else {
return;
};
if let Some(mut tickets) = self.deadline_index.get(&deadline).cloned() {
tickets.remove(&ticket_id);
if tickets.is_empty() {
self.deadline_index.remove(&deadline);
} else {
self.deadline_index.insert(deadline, tickets);
}
}
}
}
fn validate_attempt_shape(
state: &DecisionState,
attempt: &DecisionAttemptRecord,
) -> Result<(), DecisionError> {
if attempt.request_id.get() == 0 || !canonical_archive_hash(&attempt.request_commitment) {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
"decision attempt identity or request commitment is invalid",
));
}
match &attempt.outcome {
DecisionAttemptOutcome::Accepted {
trace_id,
command_request_id,
} => {
if command_request_id.is_some() && trace_id.is_none() {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
"accepted decision commands require a decision trace",
));
}
if trace_id.is_some_and(|trace_id| {
state.traces.get_ordinal(trace_id.get()).is_none()
&& !state.contains_archived_key(&DecisionHistoryKey::Trace(trace_id))
}) {
return Err(DecisionError::new(
DecisionErrorCode::InvalidDecision,
"accepted decision attempt references unavailable trace evidence",
));
}
}
DecisionAttemptOutcome::Rejected { message, .. } => {
require_text(message, "decision rejection message")?;
}
}
Ok(())
}
fn decision_hot_leaf_hash(
key: &DecisionHistoryKey,
record: &DecisionArchiveRecord,
) -> Result<[u8; 32], DecisionError> {
if record.key() != *key {
return Err(archive_error(
"decision hot-history leaf identity is inconsistent",
));
}
let bytes = serde_json::to_vec(&(key, record)).map_err(|error| {
archive_error(format!("cannot encode decision hot-history leaf: {error}"))
})?;
let mut hasher = blake3::Hasher::new();
hasher.update(b"canwu.decision.hot-history-leaf.v2");
hasher.update(&[0]);
hasher.update(&bytes);
Ok(*hasher.finalize().as_bytes())
}
fn add_digest_mod_256(target: &mut [u8; 32], digest: [u8; 32]) {
let mut carry = 0_u16;
for index in (0..target.len()).rev() {
let value = u16::from(target[index]) + u16::from(digest[index]) + carry;
target[index] = u8::try_from(value % 256).expect("modulo 256 always fits in u8");
carry = value >> 8;
}
}
fn subtract_digest_mod_256(target: &mut [u8; 32], digest: [u8; 32]) {
let mut borrow = 0_i16;
for index in (0..target.len()).rev() {
let value = i16::from(target[index]) - i16::from(digest[index]) - borrow;
if value < 0 {
target[index] =
u8::try_from(value + 256).expect("normalized byte subtraction fits in u8");
borrow = 1;
} else {
target[index] = u8::try_from(value).expect("non-negative byte subtraction fits in u8");
borrow = 0;
}
}
}
pub fn decision_history_bucket(key: &DecisionHistoryKey) -> Result<u16, DecisionError> {
Ok(decision_history_page_key(key)?.bucket)
}
pub fn decision_history_page_key(
key: &DecisionHistoryKey,
) -> Result<DecisionArchivePageKey, DecisionError> {
let encoded = serde_json::to_vec(key)
.map_err(|error| archive_error(format!("cannot encode decision history key: {error}")))?;
let digest = blake3::hash(&encoded);
let bytes = digest.as_bytes();
Ok(DecisionArchivePageKey {
bucket: (u16::from(bytes[0]) << 4) | u16::from(bytes[1] >> 4),
segment: bytes[1] & 0x0f,
})
}
#[derive(Default)]
struct DecisionScaleArchive {
blobs: RefCell<StdBTreeMap<String, DecisionArchiveBlob>>,
pages: RefCell<StdBTreeMap<String, DecisionArchiveBucketPage>>,
}
impl DecisionArchiveProvider for DecisionScaleArchive {
fn load_decision_archive(
&self,
locator: &str,
) -> Result<Option<DecisionArchiveBlob>, DecisionError> {
Ok(self.blobs.borrow().get(locator).cloned())
}
fn load_decision_archive_bucket_page(
&self,
page_id: &str,
) -> Result<Option<DecisionArchiveBucketPage>, DecisionError> {
Ok(self.pages.borrow().get(page_id).cloned())
}
}
impl DecisionArchiveStore for DecisionScaleArchive {
fn store_decision_archive(
&self,
blob: &DecisionArchiveBlob,
) -> Result<DecisionArchiveStoreOutcome, DecisionError> {
let locator = blob.content_id()?;
let mut blobs = self.blobs.borrow_mut();
if let Some(existing) = blobs.get(&locator) {
if existing != blob {
return Err(archive_error(
"decision scale archive locator contains different content",
));
}
return Ok(DecisionArchiveStoreOutcome::AlreadyStored);
}
blobs.insert(locator, blob.clone());
Ok(DecisionArchiveStoreOutcome::Stored)
}
}
#[doc(hidden)]
pub fn format8_decision_locator_scale_fixture(
key_count: usize,
) -> Result<DecisionLocatorScaleFixture, DecisionError> {
let mut state = DecisionState::default();
let archive = DecisionScaleArchive::default();
let mut archive_batches = 0_u64;
let mut next_ordinal = 1_usize;
while next_ordinal <= key_count {
let batch_end = next_ordinal
.saturating_add(MAX_DECISION_ARCHIVE_BATCH_ENTRIES - 1)
.min(key_count);
let mut keys = Vec::with_capacity(batch_end - next_ordinal + 1);
for ordinal in next_ordinal..=batch_end {
let ordinal = u64::try_from(ordinal)
.map_err(|_| archive_error("decision locator scale key exceeds u64"))?;
let request_id = canwu_core::DecisionRequestId::new(ordinal);
let request_commitment = blake3::hash(&ordinal.to_be_bytes()).to_hex().to_string();
state.append_attempt(DecisionAttemptRecord {
request_id,
request_commitment,
at: SimTime::from_minutes(
i64::try_from(ordinal)
.map_err(|_| archive_error("decision scale time exceeds i64"))?,
),
revision_before: ordinal - 1,
expected_revision: ordinal - 1,
outcome: DecisionAttemptOutcome::Rejected {
code: crate::DecisionAttemptErrorCode::InvalidDecision,
message: "format8 production-path scale attempt".to_owned(),
},
})?;
keys.push(DecisionHistoryKey::Attempt(request_id));
}
let prepared = state.prepare_decision_archive(&keys)?;
for blob in &prepared.blobs {
let _ = archive.store_decision_archive(blob)?;
}
let verified = state.verify_decision_archive(&prepared, &archive)?;
state = state.commit_verified_decision_archive(&verified)?;
let touched_pages = keys
.iter()
.map(decision_history_page_key)
.collect::<Result<OrdSet<_>, _>>()?;
for page_key in touched_pages {
let page = state
.decision_archive_bucket_page(page_key)?
.ok_or_else(|| archive_error("committed decision locator page is missing"))?;
let page_id = page.state_page_id()?;
archive.pages.borrow_mut().insert(page_id, page);
}
archive_batches = archive_batches
.checked_add(1)
.ok_or_else(|| archive_error("decision archive batch count overflowed"))?;
next_ordinal = batch_end.saturating_add(1);
}
let archive_root = state.archive_receipt_root()?;
let hot_bytes = serde_json::to_vec(&state.paged_checkpoint_hot_state()).map_err(|error| {
archive_error(format!(
"cannot persist root-only decision hot state: {error}"
))
})?;
let hot = serde_json::from_slice::<DecisionState>(&hot_bytes).map_err(|error| {
archive_error(format!(
"cannot restart root-only decision hot state: {error}"
))
})?;
let restarted = DecisionState::from_paged_checkpoint_root(
hot,
state.archive_bucket_page_ids.clone(),
state.archive_receipt_count,
&archive_root,
)?;
let samples: OrdSet<usize> = [1_usize, key_count.saturating_div(2).max(1), key_count]
.into_iter()
.filter(|ordinal| *ordinal <= key_count)
.collect::<OrdSet<_>>();
for ordinal in &samples {
let key = DecisionHistoryKey::Attempt(canwu_core::DecisionRequestId::new(
u64::try_from(*ordinal)
.map_err(|_| archive_error("decision scale sample exceeds u64"))?,
));
if restarted.load_decision_history(&key, &archive)?.is_none() {
return Err(archive_error(
"root-only decision restart lost exact archived history",
));
}
}
let reachable = restarted.archive_reachability(&archive)?;
let mut max_page_entries = 0_u64;
let mut max_page_encoded_bytes = 0_u64;
for page_id in restarted.archive_bucket_page_ids.values() {
let page = archive
.load_decision_archive_bucket_page(page_id)?
.ok_or_else(|| archive_error("decision scale locator page is unavailable"))?;
max_page_entries = max_page_entries.max(page.receipts.len() as u64);
max_page_encoded_bytes = max_page_encoded_bytes.max(
serde_json::to_vec(&page)
.map_err(|error| archive_error(format!("cannot size locator page: {error}")))?
.len() as u64,
);
}
let entry_bytes = size_of::<DecisionHistoryKey>()
.saturating_add(size_of::<CompactDecisionArchiveReceipt>())
.saturating_add(48);
let metrics = DecisionLocatorScaleMetrics {
entries: restarted.archived_history_count() as u64,
locator_pages: restarted.archive_bucket_page_ids.len() as u64,
max_page_entries,
max_page_encoded_bytes,
archive_batches,
exact_restart_queries: samples.len() as u64,
reachable_blob_locators: reachable.blob_locators.len() as u64,
estimated_resident_structural_bytes: (restarted.archived_history_count() as u64)
.saturating_mul(entry_bytes as u64),
root_hash: archive_root,
};
Ok(DecisionLocatorScaleFixture {
state,
archive_blobs: archive.blobs.into_inner().into_values().collect(),
metrics,
})
}
pub fn format8_decision_locator_scale_probe(
key_count: usize,
) -> Result<DecisionLocatorScaleMetrics, DecisionError> {
Ok(format8_decision_locator_scale_fixture(key_count)?.metrics)
}
pub fn format8_trace_locator_scale_probe(
trace_count: usize,
) -> Result<TraceLocatorScaleMetrics, DecisionError> {
if trace_count == 0 {
return Err(archive_error(
"trace locator scale probe requires at least one trace",
));
}
let mut state = DecisionState::default();
let controller_id = "format8-trace-controller".to_owned();
let policy =
DecisionPolicyIdentity::new(crate::DecisionPolicyKind::Rule, "format8-trace-policy", "1");
state.controllers.insert(
controller_id.clone(),
Arc::new(DecisionControllerBinding::new(
controller_id.clone(),
policy.clone(),
crate::DecisionAuthority::NoResponsibleActor {
reason: "Format-8 trace scale fixture".to_owned(),
},
)),
);
let ticket_id = DecisionTicketId::new(1);
let ticket = DecisionTicket {
id: ticket_id,
definition: "format8-trace-scale".to_owned(),
decision_maker: canwu_core::EntityRef::Person(canwu_core::PersonId::new(1)),
assigned_controller: controller_id.clone(),
summary: "Format-8 trace scale ticket".to_owned(),
context: crate::DecisionContext::new("format8-trace-scale", serde_json::json!({})),
options: vec![crate::DecisionOption::new("defer", "Defer")],
opened_at: SimTime::EPOCH,
updated_at: SimTime::EPOCH,
deadline: None,
version: 1,
state: DecisionTicketState::Cancelled {
reason: "Scale fixture is terminal".to_owned(),
},
parent_ticket: None,
};
ticket.validate()?;
state.insert_hot_history_record(&DecisionArchiveRecord::Ticket {
ticket: ticket.clone(),
})?;
state.tickets.insert(ticket_id, Arc::new(ticket));
for ordinal in 1..=trace_count {
let ordinal = u64::try_from(ordinal)
.map_err(|_| archive_error("trace locator scale ordinal exceeds u64"))?;
let trace = DecisionTrace {
id: DecisionTraceId::new(ordinal),
ticket_id,
ticket_version: 1,
controller_id: controller_id.clone(),
policy: policy.clone(),
decided_at: SimTime::from_minutes(
i64::try_from(ordinal)
.map_err(|_| archive_error("trace locator scale time exceeds i64"))?,
),
outcome: DecisionOutcome::Deferred {
reason: "trace-scale".to_owned(),
},
summary: "trace-scale".to_owned(),
evaluations: Vec::new(),
external: None,
random: None,
command_request_id: None,
stage: None,
fired_guards: Vec::new(),
parent_ticket: None,
};
state.insert_hot_history_record(&DecisionArchiveRecord::Trace {
trace: trace.clone(),
})?;
let inserted = state.traces.push(trace);
if inserted != ordinal {
return Err(archive_error(
"trace locator scale ordinal insertion is inconsistent",
));
}
}
state.validate()?;
let samples: OrdSet<usize> = [1_usize, trace_count.saturating_div(2).max(1), trace_count]
.into_iter()
.filter(|ordinal| *ordinal <= trace_count)
.collect::<OrdSet<_>>();
for ordinal in &samples {
let key = DecisionHistoryKey::Trace(DecisionTraceId::new(
u64::try_from(*ordinal)
.map_err(|_| archive_error("trace locator scale sample exceeds u64"))?,
));
if state.decision_locator(&key) != DecisionHistoryLocation::Hot {
return Err(archive_error(
"trace locator ordinal index lost a retained hot trace",
));
}
}
let target = DecisionHistoryKey::Trace(DecisionTraceId::new(
u64::try_from(trace_count)
.map_err(|_| archive_error("trace locator scale target exceeds u64"))?,
));
let archive = DecisionScaleArchive::default();
let prepared = state.prepare_decision_archive(std::slice::from_ref(&target))?;
for blob in &prepared.blobs {
let _ = archive.store_decision_archive(blob)?;
}
let verified = state.verify_decision_archive(&prepared, &archive)?;
let committed = state.commit_verified_decision_archive(&verified)?;
let target_archived = matches!(
committed.decision_locator(&target),
DecisionHistoryLocation::Archived { .. }
);
if !target_archived {
return Err(archive_error(
"trace locator scale archive commit did not release its target",
));
}
Ok(TraceLocatorScaleMetrics {
hot_trace_entries: trace_count as u64,
indexed_lookup_samples: samples.len() as u64,
archive_commit_entries: verified.receipts.len() as u64,
target_archived,
})
}
#[cfg(test)]
mod archive_restart_tests {
use super::*;
fn linked_decision_state(resolved_ticket: bool, accepted_attempt: bool) -> DecisionState {
let mut state = DecisionState::default();
let controller_id = "restart-controller".to_owned();
let policy =
DecisionPolicyIdentity::new(crate::DecisionPolicyKind::Rule, "restart-policy", "1");
state.controllers.insert(
controller_id.clone(),
Arc::new(DecisionControllerBinding::new(
controller_id.clone(),
policy.clone(),
crate::DecisionAuthority::NoResponsibleActor {
reason: "restart dependency fixture".to_owned(),
},
)),
);
let ticket_id = DecisionTicketId::new(91);
let trace_id = DecisionTraceId::new(1);
let ticket = DecisionTicket {
id: ticket_id,
definition: "restart-dependency".to_owned(),
decision_maker: canwu_core::EntityRef::Person(canwu_core::PersonId::new(9)),
assigned_controller: controller_id.clone(),
summary: "Restart dependency fixture".to_owned(),
context: crate::DecisionContext::new("restart-dependency", serde_json::json!({})),
options: vec![crate::DecisionOption::new("accept", "Accept")],
opened_at: SimTime::EPOCH,
updated_at: SimTime::from_minutes(1),
deadline: None,
version: 2,
state: if resolved_ticket {
DecisionTicketState::Resolved {
option_id: "accept".to_owned(),
trace_id,
}
} else {
DecisionTicketState::Cancelled {
reason: "terminal fixture".to_owned(),
}
},
parent_ticket: None,
};
state
.insert_hot_history_record(&DecisionArchiveRecord::Ticket {
ticket: ticket.clone(),
})
.expect("insert fixture ticket history");
state.tickets.insert(ticket_id, Arc::new(ticket));
let trace = DecisionTrace {
id: trace_id,
ticket_id,
ticket_version: 1,
controller_id,
policy,
decided_at: SimTime::from_minutes(1),
outcome: if resolved_ticket {
DecisionOutcome::Selected {
option_id: "accept".to_owned(),
}
} else {
DecisionOutcome::Deferred {
reason: "terminal fixture".to_owned(),
}
},
summary: "Restart dependency trace".to_owned(),
evaluations: Vec::new(),
external: None,
random: None,
command_request_id: None,
stage: None,
fired_guards: Vec::new(),
parent_ticket: None,
};
state
.insert_hot_history_record(&DecisionArchiveRecord::Trace {
trace: trace.clone(),
})
.expect("insert fixture trace history");
assert_eq!(state.traces.push(trace), trace_id.get());
if accepted_attempt {
state
.append_attempt(DecisionAttemptRecord {
request_id: canwu_core::DecisionRequestId::new(92),
request_commitment: "a".repeat(64),
at: SimTime::from_minutes(1),
revision_before: 1,
expected_revision: 1,
outcome: DecisionAttemptOutcome::Accepted {
trace_id: Some(trace_id),
command_request_id: None,
},
})
.expect("append fixture attempt");
}
state.validate().expect("fixture state validates");
state
}
fn restart_with_exact_dependency_pages(state: &DecisionState) -> DecisionState {
let hot_bytes = serde_json::to_vec(&state.paged_checkpoint_hot_state())
.expect("encode paged hot state");
let hot: DecisionState =
serde_json::from_slice(&hot_bytes).expect("decode paged hot state");
let required_pages = hot
.required_archived_dependency_page_keys()
.expect("derive exact dependency pages");
assert!(!required_pages.is_empty());
let resident_pages = required_pages
.iter()
.map(|page_key| {
state
.decision_archive_bucket_page(*page_key)
.expect("encode dependency page")
.expect("dependency page is resident")
})
.collect::<Vec<_>>();
let restarted = DecisionState::from_paged_checkpoint_root_with_resident_pages(
hot,
state.archive_bucket_page_ids.clone(),
state.archive_receipt_count,
&state.archive_receipt_root().expect("archive root"),
resident_pages,
)
.expect("restart with exact dependency pages");
let encoded = serde_json::to_vec(&restarted).expect("encode sparse restarted state");
serde_json::from_slice(&encoded).expect("sparse restarted state remains restartable")
}
#[test]
fn paged_checkpoint_root_rejects_hot_archive_metadata() {
let hot = DecisionState {
archive_receipt_count: 1,
..DecisionState::default()
};
let error = DecisionState::from_paged_checkpoint_root(
hot,
OrdMap::new(),
0,
&decision_archive_hash(
"canwu.decision.archive-receipts.v3",
&(0_u64, OrdMap::<DecisionArchivePageKey, String>::new()),
)
.expect("empty archive root"),
)
.expect_err("hot archive metadata must be rejected");
assert_eq!(error.code, DecisionErrorCode::InvalidDecision);
}
fn append_terminal_attempt(state: &mut DecisionState, ordinal: u64) {
state
.append_attempt(DecisionAttemptRecord {
request_id: canwu_core::DecisionRequestId::new(ordinal),
request_commitment: blake3::hash(&ordinal.to_be_bytes()).to_hex().to_string(),
at: SimTime::from_minutes(i64::try_from(ordinal).expect("test time fits i64")),
revision_before: ordinal.saturating_sub(1),
expected_revision: ordinal.saturating_sub(1),
outcome: DecisionAttemptOutcome::Rejected {
code: crate::DecisionAttemptErrorCode::InvalidDecision,
message: "terminal test attempt".to_owned(),
},
})
.expect("append terminal attempt");
}
fn archive_keys(
state: &DecisionState,
archive: &DecisionScaleArchive,
keys: &[DecisionHistoryKey],
) -> (DecisionState, VerifiedDecisionArchiveCommit) {
let prepared = state
.prepare_decision_archive(keys)
.expect("prepare archive");
for blob in &prepared.blobs {
archive.store_decision_archive(blob).expect("store blob");
}
let verified = state
.verify_decision_archive(&prepared, archive)
.expect("verify archive");
let next = state
.commit_verified_decision_archive(&verified)
.expect("commit archive");
for page_key in keys
.iter()
.map(decision_history_page_key)
.collect::<Result<OrdSet<_>, _>>()
.expect("derive touched pages")
{
let page = next
.decision_archive_bucket_page(page_key)
.expect("build page")
.expect("page is resident");
archive
.pages
.borrow_mut()
.insert(page.state_page_id().expect("page id"), page);
}
(next, verified)
}
#[test]
fn root_only_restart_can_archive_again_into_the_same_segment() {
let first = 1_u64;
let first_key = DecisionHistoryKey::Attempt(canwu_core::DecisionRequestId::new(first));
let page_key = decision_history_page_key(&first_key).expect("first page key");
let second = (2_u64..1_000_000)
.find(|ordinal| {
decision_history_page_key(&DecisionHistoryKey::Attempt(
canwu_core::DecisionRequestId::new(*ordinal),
))
.ok()
== Some(page_key)
})
.expect("find same-segment request id");
let second_key = DecisionHistoryKey::Attempt(canwu_core::DecisionRequestId::new(second));
let archive = DecisionScaleArchive::default();
let mut state = DecisionState::default();
append_terminal_attempt(&mut state, first);
let (state, _) = archive_keys(&state, &archive, std::slice::from_ref(&first_key));
let archive_root = state.archive_receipt_root().expect("archive root");
let hot = state.paged_checkpoint_hot_state();
assert!(hot.archive_receipt_buckets.is_empty());
assert!(hot.archive_bucket_page_ids.is_empty());
assert_eq!(hot.archive_receipt_count, 0);
let mut restarted = DecisionState::from_paged_checkpoint_root(
hot,
state.archive_bucket_page_ids.clone(),
state.archive_receipt_count,
&archive_root,
)
.expect("root-only restart");
append_terminal_attempt(&mut restarted, second);
let (restarted, verified) =
archive_keys(&restarted, &archive, std::slice::from_ref(&second_key));
assert_eq!(restarted.archived_history_count(), 2);
assert!(
restarted
.load_decision_history(&first_key, &archive)
.expect("load first")
.is_some()
);
assert!(
restarted
.load_decision_history(&second_key, &archive)
.expect("load second")
.is_some()
);
assert_eq!(
restarted
.archive_reachability(&archive)
.expect("enumerate reachability")
.blob_locators
.len(),
2
);
assert_eq!(
restarted
.commit_verified_decision_archive(&verified)
.expect("replay commit"),
restarted
);
}
#[test]
fn paged_restart_loads_archived_ticket_referenced_by_hot_trace() {
let archive = DecisionScaleArchive::default();
let state = linked_decision_state(true, false);
let ticket_id = DecisionTicketId::new(91);
let (archived, _) =
archive_keys(&state, &archive, &[DecisionHistoryKey::Ticket(ticket_id)]);
let restarted = restart_with_exact_dependency_pages(&archived);
assert!(restarted.trace(DecisionTraceId::new(1)).is_some());
assert!(matches!(
restarted.decision_locator(&DecisionHistoryKey::Ticket(ticket_id)),
DecisionHistoryLocation::Archived { .. }
));
restarted
.validate()
.expect("hot trace dependency validates");
}
#[test]
fn paged_restart_loads_archived_trace_referenced_by_resolved_hot_ticket() {
let archive = DecisionScaleArchive::default();
let state = linked_decision_state(true, false);
let trace_id = DecisionTraceId::new(1);
let (archived, _) = archive_keys(&state, &archive, &[DecisionHistoryKey::Trace(trace_id)]);
let restarted = restart_with_exact_dependency_pages(&archived);
assert!(restarted.ticket(DecisionTicketId::new(91)).is_some());
assert!(matches!(
restarted.decision_locator(&DecisionHistoryKey::Trace(trace_id)),
DecisionHistoryLocation::Archived { .. }
));
restarted
.validate()
.expect("resolved hot ticket dependency validates");
}
#[test]
fn paged_restart_loads_archived_trace_referenced_by_accepted_hot_attempt() {
let archive = DecisionScaleArchive::default();
let state = linked_decision_state(false, true);
let trace_id = DecisionTraceId::new(1);
let (archived, _) = archive_keys(&state, &archive, &[DecisionHistoryKey::Trace(trace_id)]);
let restarted = restart_with_exact_dependency_pages(&archived);
assert!(
restarted
.attempt(canwu_core::DecisionRequestId::new(92))
.is_some()
);
assert!(matches!(
restarted.decision_locator(&DecisionHistoryKey::Trace(trace_id)),
DecisionHistoryLocation::Archived { .. }
));
restarted
.validate()
.expect("accepted hot attempt dependency validates");
}
}
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct PreparedDecision {
pub trace: Option<DecisionTrace>,
pub action: Option<DecisionAction>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum ControllerDecision {
Authoritative {
decision: PolicyDecision,
action: Option<DecisionAction>,
},
Pending(PolicyDecision),
}
pub struct DecisionController;
impl DecisionController {
pub fn evaluate(
ticket: &DecisionTicket,
controller: &DecisionControllerBinding,
policy: &dyn DecisionPolicy,
) -> Result<ControllerDecision, DecisionError> {
if !ticket.is_open() {
return Err(DecisionError::new(
DecisionErrorCode::ClosedTicket,
"only open tickets can be evaluated",
));
}
if ticket.assigned_controller != controller.id || policy.identity() != &controller.policy {
return Err(DecisionError::new(
DecisionErrorCode::PolicyMismatch,
"runtime policy identity does not match the ticket controller binding",
));
}
let decision = policy.decide(ticket)?;
decision.validate(ticket)?;
if decision.random.is_some() {
return Err(DecisionError::new(
DecisionErrorCode::PolicyMismatch,
"policies cannot supply random draw evidence; draws come from a boundary resolution",
));
}
if matches!(decision.outcome, DecisionOutcome::PendingRandom { .. })
&& !controller.random_tie_break
{
return Err(DecisionError::new(
DecisionErrorCode::PolicyMismatch,
"the controller binding does not permit random tie-breaks",
));
}
if decision.outcome.is_pending() {
return Ok(ControllerDecision::Pending(decision));
}
let action = match &decision.outcome {
DecisionOutcome::Selected { option_id } => {
ticket.option(option_id).map(|option| option.action.clone())
}
DecisionOutcome::Deferred { .. }
| DecisionOutcome::Pending { .. }
| DecisionOutcome::PendingRandom { .. } => None,
};
Ok(ControllerDecision::Authoritative { action, decision })
}
}