diff --git a/crates/huginn-mind/src/mind.rs b/crates/huginn-mind/src/mind.rs index ba34555..926b92d 100644 --- a/crates/huginn-mind/src/mind.rs +++ b/crates/huginn-mind/src/mind.rs @@ -11,9 +11,7 @@ use std::path::{Path, PathBuf}; -use cultcache_rs::{ - CacheBackingStore, CultCache, CultCacheEnvelope, DatabaseEntry, OwnedRedbMessagePackBackingStore, -}; +use cultcache_rs::{CultCache, CultCacheEnvelope, DatabaseEntry, OwnedRedbMessagePackBackingStore}; use epiphany_pipeline::{ PIPELINE_SCHEMA_EPOCH, PipelineDocument, PipelineKind, Slug, register_pipeline_document_types, }; @@ -246,6 +244,7 @@ mod tests { use super::*; use crate::fixtures::{INSTANCE, epoch, instance, prepare, slug}; use crate::store::test_stores::MemoryStore; + use cultcache_rs::CacheBackingStore; fn planted(rows: Vec) -> MemoryStore { let store = MemoryStore::new(); diff --git a/crates/huginn-mind/src/receipt.rs b/crates/huginn-mind/src/receipt.rs index a74731b..3680585 100644 --- a/crates/huginn-mind/src/receipt.rs +++ b/crates/huginn-mind/src/receipt.rs @@ -13,12 +13,16 @@ //! payload bytes and their SHA-256. `committed_at` and `provenance` are not //! digested. +use chrono::{DateTime, SecondsFormat, Utc}; use cultcache_rs::{CultCacheEnvelope, DatabaseEntry}; +use epiphany_pipeline::{PipelineKind, PipelineRef, Short, Slug}; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; +use crate::mind::{Mind, unavailable}; use crate::refusal::MindRefusal; +use crate::store::MindStore; pub const RECEIPT_SCHEMA_VERSION: &str = "huginn.mind_commit_receipt.v1"; @@ -138,6 +142,108 @@ impl HuginnCommitReceipt { } } +/// How a commit ended. An exact replay is decided by admission before the +/// commit is attempted, so it is not an exit of the primitive. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) enum CommitOutcome { + /// The batch and its receipt landed whole. + Committed(HuginnCommitReceipt), + /// The swap lost: a strong read changed underneath, or a write's identity + /// already existed. Named where the kind is a pipeline kind. + Conflict(Vec), +} + +/// The receipt a batch would land with: its id is the digest of the exact +/// bytes read and written, computed before anything is stored so admission +/// can look up an exact replay. This is the one construction site of a +/// receipt. +pub(crate) fn candidate( + instance: &Slug, + provenance: PipelineProvenance, + strong_reads: &[CultCacheEnvelope], + writes: &[CultCacheEnvelope], + now: DateTime, +) -> Result { + let mut receipt = HuginnCommitReceipt { + schema_version: RECEIPT_SCHEMA_VERSION.to_string(), + receipt_id: String::new(), + instance: instance.0.clone(), + provenance, + strong_reads: versions(strong_reads), + writes: versions(writes), + committed_at: now.to_rfc3339_opts(SecondsFormat::Secs, true), + }; + receipt.receipt_id = receipt.digest()?; + Ok(receipt) +} + +/// The versions of a set of envelopes in identity order, so the digest does +/// not depend on the order a batch was assembled in. +fn versions(envelopes: &[CultCacheEnvelope]) -> Vec { + let mut versions = envelopes.iter().map(DocumentVersion::from_envelope).collect::>(); + versions.sort_by(|left, right| left.identity().cmp(&right.identity())); + versions +} + +/// The stored receipt with the candidate's id, if the batch already landed: +/// its reads and writes must equal the candidate's, or the store holds a +/// receipt this binary cannot vouch for. +pub(crate) fn replay( + mind: &Mind, + candidate: &HuginnCommitReceipt, +) -> Result, MindRefusal> { + let Some(envelope) = mind.raw_envelope(HuginnCommitReceipt::TYPE, &candidate.receipt_id) else { + return Ok(None); + }; + let existing: HuginnCommitReceipt = rmp_serde::from_slice(&envelope.payload) + .map_err(|error| integrity(&format!("receipt {} does not decode: {error}", candidate.receipt_id)))?; + existing.validate()?; + if existing.strong_reads != candidate.strong_reads || existing.writes != candidate.writes { + return Err(integrity(&format!("receipt {} is stored with other content", candidate.receipt_id))); + } + Ok(Some(existing)) +} + +/// Lands a batch whole, or not at all: one compare-and-swap with the strong +/// reads as its expectation and the writes plus the receipt as its +/// replacements. The image is re-pulled either way; a lost swap is diffed +/// against the fresh image and typed `Conflict`. +pub(crate) fn commit( + mind: &mut Mind, + receipt: HuginnCommitReceipt, + strong_reads: Vec, + writes: Vec, +) -> Result { + receipt.validate()?; + let receipt_envelope = mind + .cache() + .prepare_entry_named(&receipt.receipt_id, &receipt) + .map(|(envelope, _)| envelope) + .map_err(unavailable)?; + // The store requires every expected identity to be replaced, so each + // strong read is re-inserted unchanged beside the writes. + let mut replacements = writes.clone(); + replacements.extend(strong_reads.iter().cloned()); + replacements.push(receipt_envelope); + let expected: &[CultCacheEnvelope] = &strong_reads; + let landed = MindStore::compare_and_swap_batch(mind.store(), expected, replacements).map_err(unavailable)?; + mind.refresh()?; + if !landed { + let changed = strong_reads + .iter() + .filter(|read| mind.raw_envelope(&read.r#type, &read.key) != Some(read)) + .chain(writes.iter().filter(|write| mind.raw_envelope(&write.r#type, &write.key).is_some())); + let identities = changed + .filter_map(|envelope| { + let kind = PipelineKind::ALL.iter().find(|kind| kind.type_id() == envelope.r#type)?; + Some(PipelineRef { kind: *kind, id: Short(envelope.key.clone()) }) + }) + .collect(); + return Ok(CommitOutcome::Conflict(identities)); + } + Ok(CommitOutcome::Committed(receipt)) +} + fn unique_identities(versions: &[DocumentVersion]) -> bool { let identities = versions.iter().map(DocumentVersion::identity).collect::>(); identities.len() == versions.len()