//! 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(())
}
}