#[cfg(feature = "runtime-adapter")]
use super::*;
#[cfg(feature = "runtime-adapter")]
use meerkat_core::comms::TrustedPeerDescriptor;
#[cfg(feature = "runtime-adapter")]
use meerkat_core::ops_lifecycle::{
OperationId, OperationKind, OperationLifecycleAction, OperationLifecycleSnapshot,
OperationPeerHandle, OperationProgressUpdate, OperationSource, OperationSpec, OperationStatus,
OpsLifecycleError, OpsLifecycleRegistry,
};
#[cfg(feature = "runtime-adapter")]
use meerkat_core::types::SessionId;
#[cfg(feature = "runtime-adapter")]
use std::collections::HashMap;
#[cfg(feature = "runtime-adapter")]
use std::sync::Mutex;
#[cfg(feature = "runtime-adapter")]
#[derive(Clone)]
struct MemberOpsBinding {
owner_bridge_session_id: SessionId,
registry: Arc<dyn OpsLifecycleRegistry>,
display_name: Option<String>,
operation_source: OperationSource,
}
#[cfg(feature = "runtime-adapter")]
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
enum MemberOpsKey {
Session(SessionId),
BackendPeer { peer_id: String, address: String },
}
#[cfg(feature = "runtime-adapter")]
impl MemberOpsKey {
fn from_member_ref(member_ref: &MemberRef) -> Option<Self> {
match member_ref {
MemberRef::Session { session_id } => Some(Self::Session(session_id.clone())),
MemberRef::BackendPeer {
session_id: Some(session_id),
..
} => Some(Self::Session(session_id.clone())),
MemberRef::BackendPeer {
peer_id,
address,
session_id: None,
..
} => Some(Self::BackendPeer {
peer_id: peer_id.clone(),
address: address.clone(),
}),
}
}
fn fallback_display_name(&self) -> String {
match self {
Self::Session(session_id) => format!("mob_member/{session_id}"),
Self::BackendPeer { peer_id, address } => {
format!("mob_member/backend_peer/{peer_id}@{address}")
}
}
}
fn child_session_id(&self) -> Option<SessionId> {
match self {
Self::Session(session_id) => Some(session_id.clone()),
Self::BackendPeer { .. } => None,
}
}
}
#[cfg(feature = "runtime-adapter")]
pub(crate) struct MobOpsAdapter {
member_bindings: Mutex<HashMap<MemberOpsKey, MemberOpsBinding>>,
}
#[cfg(feature = "runtime-adapter")]
impl MobOpsAdapter {
pub(crate) fn new() -> Self {
Self {
member_bindings: Mutex::new(HashMap::new()),
}
}
pub(crate) fn bind_session_registry(
&self,
child_session_id: SessionId,
owner_bridge_session_id: SessionId,
registry: Arc<dyn OpsLifecycleRegistry>,
) {
self.member_bindings
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.insert(
MemberOpsKey::Session(child_session_id.clone()),
MemberOpsBinding {
owner_bridge_session_id,
registry,
display_name: None,
operation_source: OperationSource::session_child(child_session_id),
},
);
}
pub(crate) fn bind_member_registry(
&self,
member_ref: &MemberRef,
owner_bridge_session_id: SessionId,
registry: Arc<dyn OpsLifecycleRegistry>,
display_name: impl Into<String>,
operation_source: OperationSource,
) -> Result<(), MobError> {
let member_key = MemberOpsKey::from_member_ref(member_ref).ok_or_else(|| {
MobError::Internal(format!(
"mob ops adapter cannot bind registry for member without canonical key: {member_ref:?}",
))
})?;
self.member_bindings
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.insert(
member_key,
MemberOpsBinding {
owner_bridge_session_id,
registry,
display_name: Some(display_name.into()),
operation_source,
},
);
Ok(())
}
fn binding_for_key(&self, member_key: &MemberOpsKey) -> Option<MemberOpsBinding> {
self.member_bindings
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.get(member_key)
.cloned()
}
fn require_binding_for_key(
&self,
member_key: &MemberOpsKey,
operation: &str,
) -> Result<MemberOpsBinding, MobError> {
self.binding_for_key(member_key).ok_or_else(|| {
MobError::Internal(format!(
"mob ops adapter cannot {operation} for member '{member_key:?}': missing generated owner operation binding"
))
})
}
fn clear_member_binding(&self, member_key: &MemberOpsKey) {
self.member_bindings
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.remove(member_key);
}
pub(crate) async fn active_operation_id_for_session(
&self,
session_id: &SessionId,
) -> Option<OperationId> {
self.active_operation_id_for_key(&MemberOpsKey::Session(session_id.clone()))
.await
}
pub(crate) async fn active_operation_id_for_member(
&self,
member_ref: &MemberRef,
) -> Option<OperationId> {
let member_key = MemberOpsKey::from_member_ref(member_ref)?;
self.active_operation_id_for_key(&member_key).await
}
async fn active_operation_id_for_key(&self, member_key: &MemberOpsKey) -> Option<OperationId> {
let binding = self.binding_for_key(member_key)?;
let snapshots = match Self::matching_operations_for_binding(member_key, &binding) {
Ok(snapshots) => snapshots,
Err(error) => {
tracing::error!(
member_key = ?member_key,
error = %error,
"mob ops adapter failed to project operations through generated authority",
);
return None;
}
};
let active_ids = Self::active_operation_ids(&snapshots);
match active_ids.len() {
1 => {
let operation_id = active_ids[0].clone();
if let Err(error) = self.ensure_running_before_live_update(
member_key,
&operation_id,
"resolve active operation",
) {
tracing::error!(
member_key = ?member_key,
operation_id = %operation_id,
error = %error,
"mob ops adapter failed to canonicalize active operation before returning it",
);
return None;
}
Some(operation_id)
}
0 => None,
_ => {
tracing::error!(
member_key = ?member_key,
operation_ids = %Self::format_operation_ids(&active_ids),
"mob ops adapter found ambiguous active mob child operations; refusing to pick one",
);
None
}
}
}
fn require_member_key(
member_ref: &MemberRef,
operation: &str,
) -> Result<MemberOpsKey, MobError> {
MemberOpsKey::from_member_ref(member_ref).ok_or_else(|| {
MobError::Internal(format!(
"mob ops adapter cannot {operation} member without canonical lifecycle key: {member_ref:?}",
))
})
}
fn matching_operations_for_key(
&self,
member_key: &MemberOpsKey,
) -> Result<Vec<OperationLifecycleSnapshot>, MobError> {
let Some(binding) = self.binding_for_key(member_key) else {
return Ok(Vec::new());
};
Self::matching_operations_for_binding(member_key, &binding)
}
fn snapshot_matches_binding(
binding: &MemberOpsBinding,
snapshot: &OperationLifecycleSnapshot,
) -> bool {
snapshot.kind == OperationKind::MobMemberChild
&& snapshot.operation_source.as_ref() == Some(&binding.operation_source)
}
fn matching_operations_for_binding(
_member_key: &MemberOpsKey,
binding: &MemberOpsBinding,
) -> Result<Vec<OperationLifecycleSnapshot>, MobError> {
Ok(binding
.registry
.list_operations()
.map_err(|error| MobError::Internal(error.to_string()))?
.into_iter()
.filter(|snapshot| Self::snapshot_matches_binding(binding, snapshot))
.collect())
}
fn require_operation_for_binding(
member_key: &MemberOpsKey,
binding: &MemberOpsBinding,
operation_id: &OperationId,
operation: &str,
) -> Result<OperationLifecycleSnapshot, MobError> {
let Some(snapshot) = binding
.registry
.snapshot(operation_id)
.map_err(|error| MobError::Internal(error.to_string()))?
else {
return Err(MobError::Internal(format!(
"mob ops adapter cannot {operation}: operation '{operation_id}' not found",
)));
};
if !Self::snapshot_matches_binding(binding, &snapshot) {
return Err(MobError::Internal(format!(
"mob ops adapter cannot {operation} for member '{member_key:?}': operation '{operation_id}' is not owned by generated operation source authority"
)));
}
Ok(snapshot)
}
fn newest_operation_snapshot(
snapshots: &[OperationLifecycleSnapshot],
) -> Option<&OperationLifecycleSnapshot> {
snapshots.iter().max_by(|lhs, rhs| {
lhs.created_at_ms
.cmp(&rhs.created_at_ms)
.then_with(|| lhs.id.0.cmp(&rhs.id.0))
})
}
fn snapshot_is_terminal(snapshot: &OperationLifecycleSnapshot) -> bool {
snapshot.terminal
}
fn active_operation_ids(snapshots: &[OperationLifecycleSnapshot]) -> Vec<OperationId> {
let mut active_ids = Vec::new();
for snapshot in snapshots {
if !Self::snapshot_is_terminal(snapshot) {
active_ids.push(snapshot.id.clone());
}
}
active_ids
}
fn invalid_transition_is_idempotent(
registry: &Arc<dyn OpsLifecycleRegistry>,
action: OperationLifecycleAction,
error: &OpsLifecycleError,
) -> Result<bool, MobError> {
match error {
OpsLifecycleError::InvalidTransition { id, .. } => registry
.classify_operation_transition_idempotence(id, action)
.map_err(|error| MobError::Internal(error.to_string())),
_ => Ok(false),
}
}
fn require_exact_existing_peer_handle(
registry: &Arc<dyn OpsLifecycleRegistry>,
operation_id: &OperationId,
requested_peer_handle: &OperationPeerHandle,
) -> Result<(), MobError> {
let snapshot = registry
.snapshot(operation_id)
.map_err(|error| MobError::Internal(error.to_string()))?
.ok_or_else(|| {
MobError::Internal(format!(
"operation '{operation_id}' reported already peer-ready but has no lifecycle snapshot"
))
})?;
match snapshot.peer_handle {
Some(existing) if &existing == requested_peer_handle => Ok(()),
Some(existing) => Err(MobError::Internal(format!(
"operation '{operation_id}' is already peer-ready with a different handle: existing={existing:?}, requested={requested_peer_handle:?}"
))),
None => Err(MobError::Internal(format!(
"operation '{operation_id}' reported already peer-ready without a persisted peer handle"
))),
}
}
fn format_operation_ids(ids: &[OperationId]) -> String {
ids.iter()
.map(ToString::to_string)
.collect::<Vec<_>>()
.join(", ")
}
pub(crate) fn operation_status_with_terminality(
&self,
session_id: &SessionId,
operation_id: &OperationId,
) -> Result<Option<(OperationStatus, bool)>, MobError> {
let Some(binding) = self.binding_for_key(&MemberOpsKey::Session(session_id.clone())) else {
return Ok(None);
};
let Some(snapshot) = binding
.registry
.snapshot(operation_id)
.map_err(|error| MobError::Internal(error.to_string()))?
.filter(|snapshot| Self::snapshot_matches_binding(&binding, snapshot))
else {
return Ok(None);
};
let terminal = binding
.registry
.classify_operation_terminality(operation_id, snapshot.status)
.map_err(|error| MobError::Internal(error.to_string()))?;
Ok(Some((snapshot.status, terminal)))
}
fn display_name_for_key(
&self,
member_key: &MemberOpsKey,
operation: &str,
) -> Result<String, MobError> {
let binding = self.require_binding_for_key(member_key, operation)?;
Ok(binding
.display_name
.unwrap_or_else(|| member_key.fallback_display_name()))
}
fn registry_for_existing_binding(
&self,
member_key: &MemberOpsKey,
operation: &str,
) -> Result<Arc<dyn OpsLifecycleRegistry>, MobError> {
Ok(self
.require_binding_for_key(member_key, operation)?
.registry)
}
async fn resolve_or_register_active_operation_id_for_key(
&self,
member_key: &MemberOpsKey,
operation: &str,
display_name: &str,
) -> Result<OperationId, MobError> {
let binding = self.require_binding_for_key(member_key, operation)?;
let snapshots = Self::matching_operations_for_binding(member_key, &binding)?;
let active_ids = Self::active_operation_ids(&snapshots);
match active_ids.len() {
1 => Ok(active_ids[0].clone()),
0 => {
self.ensure_operation_for_key(member_key, display_name)
.await
}
_ => Err(MobError::Internal(format!(
"mob ops adapter cannot {operation} for member '{member_key:?}': multiple active mob child operations [{}]",
Self::format_operation_ids(&active_ids),
))),
}
}
async fn ensure_operation_for_key(
&self,
member_key: &MemberOpsKey,
display_name: &str,
) -> Result<OperationId, MobError> {
let binding = self.require_binding_for_key(member_key, "register operation")?;
let snapshots = Self::matching_operations_for_binding(member_key, &binding)?;
let active_ids = Self::active_operation_ids(&snapshots);
match active_ids.len() {
1 => return Ok(active_ids[0].clone()),
0 => {}
_ => {
return Err(MobError::Internal(format!(
"cannot provision member '{member_key:?}' with ambiguous active mob child operations [{}]",
Self::format_operation_ids(&active_ids)
)));
}
}
let operation_id = OperationId::new();
let registry = binding.registry;
registry
.register_operation(OperationSpec {
id: operation_id.clone(),
kind: OperationKind::MobMemberChild,
owner_session_id: binding.owner_bridge_session_id,
display_name: display_name.to_string(),
source_label: "mob_member".to_string(),
operation_source: Some(binding.operation_source.clone()),
child_session_id: member_key.child_session_id(),
expect_peer_channel: true,
})
.map_err(|error| MobError::Internal(error.to_string()))?;
Ok(operation_id)
}
pub(crate) async fn mark_member_provisioned(
&self,
session_id: &SessionId,
display_name: &str,
) -> Result<OperationId, MobError> {
self.mark_member_provisioned_for_key(
&MemberOpsKey::Session(session_id.clone()),
display_name,
)
.await
}
pub(crate) async fn mark_member_provisioned_for_member(
&self,
member_ref: &MemberRef,
display_name: &str,
) -> Result<OperationId, MobError> {
let member_key = Self::require_member_key(member_ref, "mark provisioned for")?;
self.mark_member_provisioned_for_key(&member_key, display_name)
.await
}
async fn mark_member_provisioned_for_key(
&self,
member_key: &MemberOpsKey,
display_name: &str,
) -> Result<OperationId, MobError> {
let operation_id = self
.ensure_operation_for_key(member_key, display_name)
.await?;
let registry = self.registry_for_existing_binding(member_key, "mark provisioned")?;
match registry.provisioning_succeeded(&operation_id) {
Ok(()) => {}
Err(error) => {
if !Self::invalid_transition_is_idempotent(
®istry,
OperationLifecycleAction::Start,
&error,
)? {
return Err(MobError::Internal(error.to_string()));
}
}
}
Ok(operation_id)
}
pub(crate) async fn report_member_progress(
&self,
member_ref: &MemberRef,
message: impl Into<String>,
) -> Result<(), MobError> {
tracing::debug!(
member_ref = ?member_ref,
"MobOpsAdapter::report_member_progress resolving member key"
);
let member_key = Self::require_member_key(member_ref, "report progress for")?;
tracing::debug!(
member_key = ?member_key,
"MobOpsAdapter::report_member_progress resolving display name"
);
let display_name = self.display_name_for_key(&member_key, "report progress")?;
tracing::debug!(
member_key = ?member_key,
display_name,
"MobOpsAdapter::report_member_progress resolving active operation"
);
let operation_id = self
.resolve_or_register_active_operation_id_for_key(
&member_key,
"report progress",
&display_name,
)
.await?;
tracing::debug!(
member_key = ?member_key,
operation_id = %operation_id,
"MobOpsAdapter::report_member_progress resolved active operation"
);
self.ensure_running_before_live_update(&member_key, &operation_id, "report progress")?;
tracing::debug!(
member_key = ?member_key,
operation_id = %operation_id,
"MobOpsAdapter::report_member_progress operation running"
);
let registry = self.registry_for_existing_binding(&member_key, "report progress")?;
tracing::debug!(
member_key = ?member_key,
operation_id = %operation_id,
"MobOpsAdapter::report_member_progress applying progress"
);
match registry.report_progress(
&operation_id,
OperationProgressUpdate {
message: message.into(),
percent: None,
},
) {
Ok(()) => Ok(()),
Err(error) => {
if Self::invalid_transition_is_idempotent(
®istry,
OperationLifecycleAction::ProgressReported,
&error,
)? {
Ok(())
} else {
Err(MobError::Internal(error.to_string()))
}
}
}?;
tracing::debug!(
member_key = ?member_key,
operation_id = %operation_id,
"MobOpsAdapter::report_member_progress applied progress"
);
Ok(())
}
pub(crate) fn report_member_progress_for_operation(
&self,
member_ref: &MemberRef,
operation_id: &OperationId,
message: impl Into<String>,
) -> Result<(), MobError> {
tracing::debug!(
member_ref = ?member_ref,
operation_id = %operation_id,
"MobOpsAdapter::report_member_progress_for_operation resolving member key"
);
let member_key = Self::require_member_key(member_ref, "report progress for")?;
let registry = self.registry_for_existing_binding(&member_key, "report progress")?;
tracing::debug!(
member_key = ?member_key,
operation_id = %operation_id,
"MobOpsAdapter::report_member_progress_for_operation applying progress"
);
match registry.report_progress(
operation_id,
OperationProgressUpdate {
message: message.into(),
percent: None,
},
) {
Ok(()) => Ok(()),
Err(error) => {
if Self::invalid_transition_is_idempotent(
®istry,
OperationLifecycleAction::ProgressReported,
&error,
)? {
Ok(())
} else {
Err(MobError::Internal(error.to_string()))
}
}
}?;
tracing::debug!(
member_key = ?member_key,
operation_id = %operation_id,
"MobOpsAdapter::report_member_progress_for_operation applied progress"
);
Ok(())
}
pub(crate) async fn mark_member_peer_ready(
&self,
member_ref: &MemberRef,
peer_name: &str,
trusted_peer: TrustedPeerDescriptor,
) -> Result<(), MobError> {
tracing::debug!(
peer_name,
member_ref = ?member_ref,
"MobOpsAdapter::mark_member_peer_ready resolving member key"
);
let member_key = Self::require_member_key(member_ref, "mark peer ready for")?;
tracing::debug!(
peer_name,
member_key = ?member_key,
"MobOpsAdapter::mark_member_peer_ready resolving active operation"
);
let operation_id = self
.resolve_or_register_active_operation_id_for_key(
&member_key,
"mark peer ready",
peer_name,
)
.await?;
tracing::debug!(
peer_name,
member_key = ?member_key,
operation_id = %operation_id,
"MobOpsAdapter::mark_member_peer_ready resolved active operation"
);
self.ensure_running_before_live_update(&member_key, &operation_id, "mark peer ready")?;
tracing::debug!(
peer_name,
member_key = ?member_key,
operation_id = %operation_id,
"MobOpsAdapter::mark_member_peer_ready operation running"
);
let registry = self.registry_for_existing_binding(&member_key, "mark peer ready")?;
tracing::debug!(
peer_name,
member_key = ?member_key,
operation_id = %operation_id,
"MobOpsAdapter::mark_member_peer_ready applying peer ready"
);
let requested_peer_handle = OperationPeerHandle {
peer_name: meerkat_core::comms::PeerName::new(peer_name)
.map_err(|e| MobError::Internal(format!("invalid peer name: {e}")))?,
trusted_peer,
};
match registry.peer_ready(&operation_id, requested_peer_handle.clone()) {
Ok(()) => Ok(()),
Err(OpsLifecycleError::AlreadyPeerReady(_)) => {
Self::require_exact_existing_peer_handle(
®istry,
&operation_id,
&requested_peer_handle,
)
}
Err(error) => {
if Self::invalid_transition_is_idempotent(
®istry,
OperationLifecycleAction::PeerReady,
&error,
)? {
Self::require_exact_existing_peer_handle(
®istry,
&operation_id,
&requested_peer_handle,
)
} else {
Err(MobError::Internal(error.to_string()))
}
}
}?;
tracing::debug!(
peer_name,
member_key = ?member_key,
operation_id = %operation_id,
"MobOpsAdapter::mark_member_peer_ready applied peer ready"
);
Ok(())
}
pub(crate) fn mark_member_peer_ready_for_operation(
&self,
member_ref: &MemberRef,
operation_id: &OperationId,
peer_name: &str,
trusted_peer: TrustedPeerDescriptor,
) -> Result<(), MobError> {
tracing::debug!(
peer_name,
operation_id = %operation_id,
member_ref = ?member_ref,
"MobOpsAdapter::mark_member_peer_ready_for_operation resolving member key"
);
let member_key = Self::require_member_key(member_ref, "mark peer ready for")?;
tracing::debug!(
peer_name,
operation_id = %operation_id,
member_key = ?member_key,
"MobOpsAdapter::mark_member_peer_ready_for_operation getting registry"
);
let registry = self.registry_for_existing_binding(&member_key, "mark peer ready")?;
tracing::debug!(
peer_name,
operation_id = %operation_id,
member_key = ?member_key,
"MobOpsAdapter::mark_member_peer_ready_for_operation applying peer ready"
);
let requested_peer_handle = OperationPeerHandle {
peer_name: meerkat_core::comms::PeerName::new(peer_name)
.map_err(|e| MobError::Internal(format!("invalid peer name: {e}")))?,
trusted_peer,
};
match registry.peer_ready(operation_id, requested_peer_handle.clone()) {
Ok(()) => Ok(()),
Err(OpsLifecycleError::AlreadyPeerReady(_)) => {
Self::require_exact_existing_peer_handle(
®istry,
operation_id,
&requested_peer_handle,
)
}
Err(error) => {
if Self::invalid_transition_is_idempotent(
®istry,
OperationLifecycleAction::PeerReady,
&error,
)? {
Self::require_exact_existing_peer_handle(
®istry,
operation_id,
&requested_peer_handle,
)
} else {
Err(MobError::Internal(error.to_string()))
}
}
}?;
tracing::debug!(
peer_name,
operation_id = %operation_id,
member_key = ?member_key,
"MobOpsAdapter::mark_member_peer_ready_for_operation applied peer ready"
);
Ok(())
}
pub(crate) async fn mark_member_retired(&self, member_ref: &MemberRef) -> Result<(), MobError> {
let member_key = Self::require_member_key(member_ref, "mark retired for")?;
let binding = self.require_binding_for_key(&member_key, "mark retired")?;
let snapshots = Self::matching_operations_for_binding(&member_key, &binding)?;
let active_ids = Self::active_operation_ids(&snapshots);
let operation_id = match active_ids.len() {
1 => active_ids[0].clone(),
0 => {
if let Some(latest) = Self::newest_operation_snapshot(&snapshots)
&& Self::snapshot_is_terminal(latest)
{
return Ok(());
}
return Ok(());
}
_ => {
return Err(MobError::Internal(format!(
"cannot retire member '{member_key:?}': multiple active mob child operations [{}]",
Self::format_operation_ids(&active_ids)
)));
}
};
let registry = binding.registry;
match registry.request_retire(&operation_id) {
Ok(()) => {}
Err(error) => {
if !Self::invalid_transition_is_idempotent(
®istry,
OperationLifecycleAction::RetireRequested,
&error,
)? {
return Err(MobError::Internal(error.to_string()));
}
}
}
let result = match registry.mark_retired(&operation_id) {
Ok(()) => Ok(()),
Err(error) => {
if Self::invalid_transition_is_idempotent(
®istry,
OperationLifecycleAction::RetireCompleted,
&error,
)? {
Ok(())
} else {
Err(MobError::Internal(error.to_string()))
}
}
};
if result.is_ok() {
self.clear_member_binding(&member_key);
}
result
}
pub(crate) async fn abort_member_provision(
&self,
session_id: &SessionId,
operation_id: &OperationId,
reason: Option<String>,
) -> Result<(), MobError> {
self.abort_member_provision_for_key(
&MemberOpsKey::Session(session_id.clone()),
operation_id,
reason,
)
.await
}
pub(crate) async fn abort_member_provision_for_member(
&self,
member_ref: &MemberRef,
operation_id: &OperationId,
reason: Option<String>,
) -> Result<(), MobError> {
let member_key = Self::require_member_key(member_ref, "abort provision for")?;
self.abort_member_provision_for_key(&member_key, operation_id, reason)
.await
}
async fn abort_member_provision_for_key(
&self,
member_key: &MemberOpsKey,
operation_id: &OperationId,
reason: Option<String>,
) -> Result<(), MobError> {
let binding = self.require_binding_for_key(member_key, "abort provision")?;
let registry = Arc::clone(&binding.registry);
Self::require_operation_for_binding(member_key, &binding, operation_id, "abort provision")?;
let result = match registry.abort_provisioning(operation_id, reason) {
Ok(()) => Ok(()),
Err(OpsLifecycleError::NotFound(_)) => Ok(()),
Err(error) => {
if Self::invalid_transition_is_idempotent(
®istry,
OperationLifecycleAction::Abort,
&error,
)? {
Ok(())
} else {
Err(MobError::Internal(error.to_string()))
}
}
};
if result.is_ok() {
self.clear_member_binding(member_key);
}
result
}
fn ensure_running_before_live_update(
&self,
member_key: &MemberOpsKey,
operation_id: &OperationId,
operation: &str,
) -> Result<(), MobError> {
let binding = self.require_binding_for_key(member_key, operation)?;
let registry = Arc::clone(&binding.registry);
let snapshot =
Self::require_operation_for_binding(member_key, &binding, operation_id, operation)?;
if snapshot.status != OperationStatus::Provisioning {
return Ok(());
}
match registry.provisioning_succeeded(operation_id) {
Ok(()) => Ok(()),
Err(error) => {
if Self::invalid_transition_is_idempotent(
®istry,
OperationLifecycleAction::Start,
&error,
)? {
Ok(())
} else {
Err(MobError::Internal(format!(
"mob ops adapter cannot {operation}: failed to canonicalize provisioning operation '{operation_id}' into running state: {error}",
)))
}
}
}
}
}
#[cfg(test)]
#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
mod tests {
use super::*;
use meerkat_core::comms::{PeerAddress, PeerId, TrustedPeerDescriptor};
use meerkat_core::ops_lifecycle::OperationStatus;
use meerkat_runtime::RuntimeOpsLifecycleRegistry;
#[tokio::test]
async fn ops_registry_integration_red_ok_member_adapter_tracks_peer_ready_and_retire() {
let adapter = MobOpsAdapter::new();
let owner_bridge_session_id = SessionId::new();
let session_id = SessionId::new();
let member_ref = MemberRef::from_bridge_session_id(session_id.clone());
let registry = Arc::new(RuntimeOpsLifecycleRegistry::new());
adapter.bind_session_registry(
session_id.clone(),
owner_bridge_session_id,
Arc::clone(®istry) as Arc<dyn OpsLifecycleRegistry>,
);
let operation_id = adapter
.mark_member_provisioned(&session_id, "mob-a/orchestrator/member-alpha")
.await
.expect("register member");
adapter
.report_member_progress(&member_ref, "turn dispatched")
.await
.expect("progress");
adapter
.mark_member_peer_ready(
&member_ref,
"mob-a/orchestrator/member-alpha",
TrustedPeerDescriptor::test_only_unsigned(
"mob-a/orchestrator/member-alpha",
"00000000-0000-4000-8000-000000000002",
"inproc://member-alpha",
)
.expect("trusted peer"),
)
.await
.expect("peer ready");
let running_snapshot = registry
.snapshot(&operation_id)
.expect("snapshot projection")
.expect("snapshot");
assert_eq!(running_snapshot.status, OperationStatus::Running);
assert!(running_snapshot.peer_ready);
assert_eq!(running_snapshot.progress_count, 1);
adapter
.mark_member_retired(&member_ref)
.await
.expect("retire member");
let retired_snapshot = registry
.snapshot(&operation_id)
.expect("snapshot projection")
.expect("snapshot");
assert_eq!(retired_snapshot.status, OperationStatus::Retired);
}
#[tokio::test]
async fn operation_peer_ready_retry_requires_the_exact_existing_handle() {
let adapter = MobOpsAdapter::new();
let owner_bridge_session_id = SessionId::new();
let session_id = SessionId::new();
let member_ref = MemberRef::from_bridge_session_id(session_id.clone());
let registry = Arc::new(RuntimeOpsLifecycleRegistry::new());
adapter.bind_session_registry(
session_id.clone(),
owner_bridge_session_id,
Arc::clone(®istry) as Arc<dyn OpsLifecycleRegistry>,
);
let operation_id = adapter
.mark_member_provisioned(&session_id, "mob-a/orchestrator/member-alpha")
.await
.expect("register member");
let peer_name = "mob-a/orchestrator/member-alpha";
let exact_peer = TrustedPeerDescriptor::test_only_unsigned(
peer_name,
"00000000-0000-4000-8000-000000000002",
"inproc://member-alpha",
)
.expect("exact trusted peer");
adapter
.mark_member_peer_ready_for_operation(
&member_ref,
&operation_id,
peer_name,
exact_peer.clone(),
)
.expect("first peer-ready publication");
adapter
.mark_member_peer_ready_for_operation(
&member_ref,
&operation_id,
peer_name,
exact_peer.clone(),
)
.expect("same exact peer handle is idempotent");
adapter
.mark_member_peer_ready(&member_ref, peer_name, exact_peer.clone())
.await
.expect("resolved-operation path also accepts the exact existing handle");
let different_peer = TrustedPeerDescriptor::test_only_unsigned(
peer_name,
"00000000-0000-4000-8000-000000000003",
"inproc://member-alpha-rebound",
)
.expect("different trusted peer");
let resolved_error = adapter
.mark_member_peer_ready(&member_ref, peer_name, different_peer.clone())
.await
.expect_err("resolved-operation path must reject a different peer handle");
assert!(
resolved_error
.to_string()
.contains("already peer-ready with a different handle"),
"unexpected resolved-operation mismatch error: {resolved_error}"
);
let error = adapter
.mark_member_peer_ready_for_operation(
&member_ref,
&operation_id,
peer_name,
different_peer,
)
.expect_err("different peer handle must fail closed");
assert!(
error
.to_string()
.contains("already peer-ready with a different handle"),
"unexpected mismatch error: {error}"
);
let snapshot = registry
.snapshot(&operation_id)
.expect("snapshot projection")
.expect("snapshot");
assert_eq!(
snapshot.peer_handle,
Some(OperationPeerHandle {
peer_name: meerkat_core::comms::PeerName::new(peer_name)
.expect("canonical peer name"),
trusted_peer: exact_peer,
}),
"failed retry must not replace the operation's exact peer handle"
);
}
#[tokio::test]
async fn bound_session_registry_owns_child_operation_ids() {
let adapter = MobOpsAdapter::new();
let owner_bridge_session_id = SessionId::new();
let child_session_id = SessionId::new();
let bound_registry = Arc::new(RuntimeOpsLifecycleRegistry::new());
adapter.bind_session_registry(
child_session_id.clone(),
owner_bridge_session_id,
Arc::clone(&bound_registry) as Arc<dyn OpsLifecycleRegistry>,
);
let operation_id = adapter
.mark_member_provisioned(&child_session_id, "mob/member-bound")
.await
.expect("register bound child op");
let bound_snapshot = bound_registry
.snapshot(&operation_id)
.expect("snapshot projection")
.expect("bound registry should own exported child operation ids");
assert_eq!(bound_snapshot.kind, OperationKind::MobMemberChild);
assert_eq!(bound_snapshot.status, OperationStatus::Running);
assert_eq!(bound_snapshot.child_session_id, Some(child_session_id));
}
#[tokio::test]
async fn repeated_member_provision_uses_generated_idempotence_classifier() {
let adapter = MobOpsAdapter::new();
let owner_bridge_session_id = SessionId::new();
let child_session_id = SessionId::new();
let registry = Arc::new(RuntimeOpsLifecycleRegistry::new());
adapter.bind_session_registry(
child_session_id.clone(),
owner_bridge_session_id,
Arc::clone(®istry) as Arc<dyn OpsLifecycleRegistry>,
);
let first = adapter
.mark_member_provisioned(&child_session_id, "mob/member-idempotent")
.await
.expect("first provision");
let second = adapter
.mark_member_provisioned(&child_session_id, "mob/member-idempotent")
.await
.expect("generated idempotent start classification");
assert_eq!(first, second);
let snapshot = registry
.snapshot(&first)
.expect("snapshot projection")
.expect("snapshot");
assert_eq!(snapshot.status, OperationStatus::Running);
}
#[tokio::test]
async fn already_retiring_member_retire_uses_generated_idempotence_classifier() {
let adapter = MobOpsAdapter::new();
let owner_bridge_session_id = SessionId::new();
let child_session_id = SessionId::new();
let member_ref = MemberRef::from_bridge_session_id(child_session_id.clone());
let registry = Arc::new(RuntimeOpsLifecycleRegistry::new());
adapter.bind_session_registry(
child_session_id.clone(),
owner_bridge_session_id,
Arc::clone(®istry) as Arc<dyn OpsLifecycleRegistry>,
);
let operation_id = adapter
.mark_member_provisioned(&child_session_id, "mob/member-retiring")
.await
.expect("provision member");
registry
.request_retire(&operation_id)
.expect("generated retire request");
adapter
.mark_member_retired(&member_ref)
.await
.expect("generated idempotent retire-request classification");
let snapshot = registry
.snapshot(&operation_id)
.expect("snapshot projection")
.expect("snapshot");
assert_eq!(snapshot.status, OperationStatus::Retired);
}
#[tokio::test]
async fn mark_member_retired_without_existing_operation_is_noop() {
let adapter = MobOpsAdapter::new();
let owner_bridge_session_id = SessionId::new();
let session_id = SessionId::new();
let member_ref = MemberRef::from_bridge_session_id(session_id.clone());
let registry = Arc::new(RuntimeOpsLifecycleRegistry::new());
adapter.bind_session_registry(
session_id,
owner_bridge_session_id,
Arc::clone(®istry) as Arc<dyn OpsLifecycleRegistry>,
);
adapter
.mark_member_retired(&member_ref)
.await
.expect("missing lifecycle entry should retire as no-op");
assert!(
registry.list_operations().unwrap().is_empty(),
"retire must not fabricate a provisioning operation"
);
}
#[tokio::test]
async fn member_operation_registration_without_owner_binding_fails_closed() {
let adapter = MobOpsAdapter::new();
let session_id = SessionId::new();
let err = adapter
.mark_member_provisioned(&session_id, "mob/member-unbound")
.await
.expect_err("missing generated owner binding must fail closed");
assert!(
err.to_string()
.contains("missing generated owner operation binding"),
"unexpected error: {err}"
);
let member_ref = MemberRef::BackendPeer {
peer_id: "peer-a".into(),
address: "inproc://peer-a".into(),
pubkey: [7u8; 32],
bootstrap_token: None,
session_id: None,
};
let err = adapter
.mark_member_provisioned_for_member(&member_ref, "mob/backend-peer")
.await
.expect_err("peer-only member without generated owner binding must fail closed");
assert!(
err.to_string()
.contains("missing generated owner operation binding"),
"unexpected error: {err}"
);
assert!(
adapter
.active_operation_id_for_member(&member_ref)
.await
.is_none(),
"unbound peer-only member must not create fallback operation state"
);
}
#[tokio::test]
async fn peer_only_operation_identity_uses_generated_operation_source() {
let adapter = MobOpsAdapter::new();
let owner_bridge_session_id = SessionId::new();
let registry = Arc::new(RuntimeOpsLifecycleRegistry::new());
let peer_id = PeerId::new();
let address = PeerAddress::parse("inproc://peer-a").expect("peer address");
let operation_source = OperationSource::backend_peer(peer_id, address.clone());
let member_ref = MemberRef::BackendPeer {
peer_id: peer_id.to_string(),
address: address.to_string(),
pubkey: [7u8; 32],
bootstrap_token: None,
session_id: None,
};
adapter
.bind_member_registry(
&member_ref,
owner_bridge_session_id.clone(),
Arc::clone(®istry) as Arc<dyn OpsLifecycleRegistry>,
"display-before",
operation_source.clone(),
)
.expect("bind peer-only member registry");
let operation_id = adapter
.mark_member_provisioned_for_member(&member_ref, "display-before")
.await
.expect("register peer-only child operation");
let snapshot = registry
.snapshot(&operation_id)
.expect("snapshot projection")
.expect("snapshot");
assert_eq!(snapshot.display_name, "display-before");
assert_eq!(snapshot.operation_source, Some(operation_source.clone()));
adapter
.bind_member_registry(
&member_ref,
owner_bridge_session_id,
Arc::clone(®istry) as Arc<dyn OpsLifecycleRegistry>,
"display-after",
operation_source,
)
.expect("rebinding display name must not change lifecycle identity");
assert_eq!(
adapter.active_operation_id_for_member(&member_ref).await,
Some(operation_id)
);
}
#[tokio::test]
async fn mark_member_retired_without_owner_binding_fails_closed() {
let adapter = MobOpsAdapter::new();
let member_ref = MemberRef::from_bridge_session_id(SessionId::new());
let err = adapter
.mark_member_retired(&member_ref)
.await
.expect_err("missing generated owner binding must fail closed");
assert!(
err.to_string()
.contains("missing generated owner operation binding"),
"unexpected error: {err}"
);
}
#[tokio::test]
async fn abort_member_provision_marks_provisioning_operation_aborted() {
let adapter = MobOpsAdapter::new();
let owner_bridge_session_id = SessionId::new();
let session_id = SessionId::new();
let operation_id = OperationId::new();
let operation_source = OperationSource::session_child(session_id.clone());
let registry = Arc::new(RuntimeOpsLifecycleRegistry::new());
adapter.bind_session_registry(
session_id.clone(),
owner_bridge_session_id,
Arc::clone(®istry) as Arc<dyn OpsLifecycleRegistry>,
);
registry
.register_operation(OperationSpec {
id: operation_id.clone(),
kind: OperationKind::MobMemberChild,
owner_session_id: session_id.clone(),
display_name: "mob/member-abort".into(),
source_label: "mob_member".into(),
operation_source: Some(operation_source),
child_session_id: Some(session_id.clone()),
expect_peer_channel: true,
})
.expect("register provisioning operation");
adapter
.abort_member_provision(&session_id, &operation_id, Some("mob is stopping".into()))
.await
.expect("abort provisioning should succeed");
let snapshot = registry
.snapshot(&operation_id)
.expect("snapshot projection")
.expect("snapshot");
assert_eq!(snapshot.status, OperationStatus::Aborted);
}
#[tokio::test]
async fn abort_member_provision_without_owner_binding_fails_closed() {
let adapter = MobOpsAdapter::new();
let session_id = SessionId::new();
let operation_id = OperationId::new();
let err = adapter
.abort_member_provision(&session_id, &operation_id, Some("mob is stopping".into()))
.await
.expect_err("missing generated owner binding must fail closed");
assert!(
err.to_string()
.contains("missing generated owner operation binding"),
"unexpected error: {err}"
);
}
}