ive harnessed the harness
Something went wrong. Try again.
14 kB · 425 lines
Rust
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426//! lifecycle truth for remote namespaces.//!//! m2 deliberately advertises **connection-scoped persistence**. a remote//! namespace belongs to one live connection owner; losing that connection//! loses the namespace. reconnecting creates a new owner, namespace id, and//! generation. this module does not claim reattachment, and `ControlPersist`//! is not evidence of one.
use crate::{ protocol::{OperationId, SessionId}, remote::{ bootstrap::{ensure_target_readiness, BootstrapConfig}, bundle::BundleSpec, }, store::Store, target::{NamespaceGeneration, NamespaceId, RemoteTarget, TargetId},};use anyhow::{bail, ensure, Context, Result};use serde::{Deserialize, Serialize};use serde_json::json;use std::{ path::PathBuf, sync::{Arc, Mutex},};
/// Persistence offered by a remote workbench.////// The only M2 variant is [`PersistenceMode::ConnectionScoped`]. In/// particular, there is no reattach variant until owner proof and deduplicated/// operation receipts exist.#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]#[serde(rename_all = "snake_case")]pub enum PersistenceMode { ConnectionScoped,}
impl PersistenceMode { pub const ADVERTISED: Self = Self::ConnectionScoped;}
/// Whether the owner of a connection-scoped namespace can still speak for it.#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]pub enum ConnectionState { Connected, Lost { reason: String },}
/// A public snapshot suitable for an SDK or diagnostic response.#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]#[serde(deny_unknown_fields)]pub struct LifecycleSnapshot { pub mode: PersistenceMode, pub namespace_id: NamespaceId, pub namespace_generation: NamespaceGeneration, pub owner_id: String, pub state: ConnectionState,}
/// A narrow, typed diagnosis that permits the explicit bootstrap repair lane.////// A command's nonzero exit is deliberately absent. Callers must first have a/// bootstrap boundary identify one of these failures; strings and booleans are/// not repair authorization.#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]pub enum BootstrapFailure { MissingPrerequisite { detail: String }, CandidateRejected { detail: String }, VerificationFailed { detail: String }, IncompatibleEnvironment { detail: String },}
impl BootstrapFailure { fn detail(&self) -> &str { match self { Self::MissingPrerequisite { detail } | Self::CandidateRejected { detail } | Self::VerificationFailed { detail } | Self::IncompatibleEnvironment { detail } => detail, } }
fn validate(&self) -> Result<()> { ensure_reason(self.detail()) }}
/// An attributable request for the exceptional repair lane.#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]#[serde(deny_unknown_fields)]pub struct RepairRequest { pub target: TargetId, pub operation: OperationId, pub reason: BootstrapFailure, /// If set, this repair is adjacent to an operation whose remote outcome is /// unknown. The repair is recorded, then refused; it can never replay it. pub uncertain_operation: Option<OperationId>,}
impl RepairRequest { pub fn new(target: TargetId, operation: OperationId, reason: BootstrapFailure) -> Result<Self> { reason.validate()?; Ok(Self { target, operation, reason, uncertain_operation: None, }) }
pub fn with_uncertain_operation(mut self, operation: OperationId) -> Self { self.uncertain_operation = Some(operation); self }}
/// A typed refusal for a repair that could otherwise replay an uncertain cell.#[derive(Clone, Debug, PartialEq, Eq)]pub enum RepairError { UncertainMutation { operation: OperationId },}
impl std::fmt::Display for RepairError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { Self::UncertainMutation { operation } => write!( f, "repair refused: operation {operation} has unknown outcome; replay is forbidden" ), } }}impl std::error::Error for RepairError {}
/// Durable evidence that the explicit repair lane was admitted.#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]#[serde(deny_unknown_fields)]pub struct RepairReceipt { pub target: TargetId, pub operation: OperationId, pub event_id: i64, pub status: String, pub normal_kernel_required: bool,}
fn ensure_reason(detail: &str) -> Result<()> { ensure!( !detail.trim().is_empty() && detail.len() <= 4096, "bootstrap diagnosis is empty or too large" ); Ok(())}
/// Production repair action for a diagnosed bootstrap failure.////// This is deliberately a concrete bootstrap invocation, not a caller-supplied/// closure: repair uses the same target readiness path as first admission.#[derive(Clone, Debug)]pub struct BootstrapRepair { target: RemoteTarget, spec: BundleSpec, bundles_root: PathBuf, config: BootstrapConfig,}
impl BootstrapRepair { pub fn new( target: RemoteTarget, spec: BundleSpec, bundles_root: impl Into<PathBuf>, config: BootstrapConfig, ) -> Self { Self { target, spec, bundles_root: bundles_root.into(), config, } }
/// Runs the real content-addressed bootstrap path after recording the /// diagnosis. An uncertain operation is recorded as refused and never /// reaches the bootstrap action. pub async fn repair( &self, store: &Store, session: &SessionId, request: RepairRequest, ) -> Result<RepairReceipt> { request.reason.validate()?; ensure!( request.target == self.target.id, "repair target does not match configured bootstrap target" ); let uncertain = request.uncertain_operation.clone(); let status = if uncertain.is_some() { "refused_uncertain" } else { "started" }; let admission_event = store .append( session, "remote.repair", json!({ "target": request.target, "operation": request.operation, "reason": request.reason, "uncertain_operation": uncertain, "status": status, "replay": "forbidden", "mode": PersistenceMode::ConnectionScoped, }), ) .await?; if let Some(operation) = uncertain { return Err(RepairError::UncertainMutation { operation }.into()); }
let outcome = match ensure_target_readiness( &self.target, &self.spec, &self.bundles_root, &self.config, ) .await { Ok(outcome) => outcome, Err(error) => { store .append( session, "remote.repair", json!({ "target": self.target.id, "operation": request.operation, "status": "failed", "admission_event": admission_event, "reason": request.reason, "error": format!("{error:#}"), "replay": "forbidden", }), ) .await?; return Err(error).context("diagnosed bootstrap repair failed"); } }; let event_id = store .append( session, "remote.repair", json!({ "target": self.target.id, "operation": request.operation, "status": "succeeded", "admission_event": admission_event, "bundle_id": outcome.manifest().bundle_id, "warm_reuse": outcome.is_warm(), "replay": "forbidden", }), ) .await?; Ok(RepairReceipt { target: self.target.id.clone(), operation: request.operation, event_id, status: "succeeded".into(), normal_kernel_required: true, }) }}
/// Host-owned lease for one connection-scoped remote namespace.////// `reconnect` never mutates this lease back to `Connected`: callers must/// replace it with the newly returned lease, making accidental old-handle/// reuse visible in both the namespace id and generation.#[derive(Clone, Debug)]pub struct ConnectionScopedLifecycle { namespace_id: NamespaceId, namespace_generation: NamespaceGeneration, owner_id: String, state: Arc<Mutex<ConnectionState>>,}
impl ConnectionScopedLifecycle { /// Creates the initial lease for an already-admitted remote connection. pub fn new(namespace_id: NamespaceId, namespace_generation: NamespaceGeneration) -> Self { Self { namespace_id, namespace_generation, owner_id: uuid::Uuid::new_v4().to_string(), state: Arc::new(Mutex::new(ConnectionState::Connected)), } }
/// M2's advertised persistence contract. #[must_use] pub const fn mode(&self) -> PersistenceMode { PersistenceMode::ConnectionScoped }
#[must_use] pub fn snapshot(&self) -> LifecycleSnapshot { let state = self .state .lock() .map(|state| state.clone()) .unwrap_or_else(|_| ConnectionState::Lost { reason: "lifecycle state lock poisoned".into(), }); LifecycleSnapshot { mode: self.mode(), namespace_id: self.namespace_id.clone(), namespace_generation: self.namespace_generation.clone(), owner_id: self.owner_id.clone(), state, } }
#[must_use] pub fn namespace_id(&self) -> &NamespaceId { &self.namespace_id }
#[must_use] pub fn namespace_generation(&self) -> &NamespaceGeneration { &self.namespace_generation }
#[must_use] pub fn is_connected(&self) -> bool { self.state .lock() .map(|state| matches!(*state, ConnectionState::Connected)) .unwrap_or(false) }
/// Refuses host work after the connection owner has been lost. pub fn ensure_connected(&self) -> Result<()> { let state = self .state .lock() .map_err(|_| anyhow::anyhow!("remote lifecycle state lock poisoned"))?; if let ConnectionState::Lost { reason } = &*state { bail!( "remote namespace {}@{} is lost: {reason}; reconnect explicitly", self.namespace_id, self.namespace_generation ); } Ok(()) }
/// Records an uncertain transport loss exactly once. /// /// The first loss reason wins because a later observation cannot turn an /// unknown remote outcome into a known one. pub fn mark_lost(&self, reason: impl Into<String>) -> Result<bool> { let mut state = self .state .lock() .map_err(|_| anyhow::anyhow!("remote lifecycle state lock poisoned"))?; if matches!(*state, ConnectionState::Lost { .. }) { return Ok(false); } *state = ConnectionState::Lost { reason: reason.into(), }; Ok(true) }
/// Starts an independent owner and namespace after an explicit reconnect. /// The old lease remains `Lost`; it is never revived or reused. pub fn reconnect(&self) -> Result<Self> { self.ensure_lost()?; Ok(Self::new( fresh_namespace_id()?, NamespaceGeneration::fresh(), )) }
fn ensure_lost(&self) -> Result<()> { let state = self .state .lock() .map_err(|_| anyhow::anyhow!("remote lifecycle state lock poisoned"))?; if matches!(*state, ConnectionState::Connected) { bail!("remote namespace is still connected; disconnect before reconnect") } Ok(()) }}
/// Makes an opaque namespace id for an explicit new connection owner.pub fn fresh_namespace_id() -> Result<NamespaceId> { NamespaceId::parse(format!("ns-{}", uuid::Uuid::new_v4().simple()))}
#[cfg(test)]mod tests { use super::*;
#[test] fn connection_scoped_loss_is_terminal_and_reconnect_is_new() -> Result<()> { let old = ConnectionScopedLifecycle::new( NamespaceId::parse("old")?, NamespaceGeneration::parse("generation-old")?, ); assert_eq!(old.mode(), PersistenceMode::ConnectionScoped); assert!(old.is_connected()); assert!(old.mark_lost("relay closed")?); assert!(!old.mark_lost("later observation")?); assert!(old.ensure_connected().is_err());
let new = old.reconnect()?; assert!(new.is_connected()); assert_ne!(old.namespace_id(), new.namespace_id()); assert_ne!(old.namespace_generation(), new.namespace_generation()); assert_ne!(old.snapshot().owner_id, new.snapshot().owner_id); assert!(old.ensure_connected().is_err()); Ok(()) }}