lazaret/crates/lazaret-core/src/records/lifecycle.rs

1454 lines
48 KiB
Rust

use super::{
BundleDigest, ClientId, ConfigurationFingerprint, ContainerId, CoordinatorId,
DOMAIN_STORE_SCHEMA_ID, DOMAIN_STORE_SCHEMA_VERSION, DerivedPolicyDigest, Deserialize,
DeviceId, DomainId, FileDescriptor, GenerationId, GuestProcessGroupId, GuestProcessId,
GuestProcessStartTimeTicks, HostPath, InodeId, MountIdentity, NamespaceIdentity,
NetworkReadinessEvidence, NetworkSidecarId, PidfsProcessIdentityV1, PreparedArtifactId,
ReadinessFingerprint, RecordInvariantError, RecordTransitionError, RequestId,
STORAGE_CATALOG_SCHEMA_ID, STORAGE_CATALOG_SCHEMA_VERSION, SandboxId, Serialize,
SessionAnchorId, SessionId, SourceIdentity, fmt,
};
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct StoreMetadata {
schema_id: String,
schema_version: u32,
}
impl StoreMetadata {
#[must_use]
pub fn domain_store_v1() -> Self {
Self {
schema_id: DOMAIN_STORE_SCHEMA_ID.into(),
schema_version: DOMAIN_STORE_SCHEMA_VERSION,
}
}
#[must_use]
pub fn storage_catalog_v1() -> Self {
Self {
schema_id: STORAGE_CATALOG_SCHEMA_ID.into(),
schema_version: STORAGE_CATALOG_SCHEMA_VERSION,
}
}
#[must_use]
pub fn schema_id(&self) -> &str {
&self.schema_id
}
#[must_use]
pub const fn schema_version(&self) -> u32 {
self.schema_version
}
fn is_domain_store(&self) -> bool {
self.schema_id == DOMAIN_STORE_SCHEMA_ID
&& self.schema_version == DOMAIN_STORE_SCHEMA_VERSION
}
fn is_storage_catalog(&self) -> bool {
self.schema_id == STORAGE_CATALOG_SCHEMA_ID
&& self.schema_version == STORAGE_CATALOG_SCHEMA_VERSION
}
}
impl<'de> Deserialize<'de> for StoreMetadata {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
#[derive(Deserialize)]
struct Fields {
schema_id: String,
schema_version: u32,
}
let fields = Fields::deserialize(deserializer)?;
let metadata = Self {
schema_id: fields.schema_id,
schema_version: fields.schema_version,
};
if metadata.is_domain_store() || metadata.is_storage_catalog() {
Ok(metadata)
} else {
Err(serde::de::Error::custom(
"unknown store schema identity or version",
))
}
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct DomainStoreMetadata(StoreMetadata);
impl DomainStoreMetadata {
#[must_use]
pub fn v1() -> Self {
Self(StoreMetadata::domain_store_v1())
}
#[must_use]
pub const fn as_metadata(&self) -> &StoreMetadata {
&self.0
}
}
impl<'de> Deserialize<'de> for DomainStoreMetadata {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let metadata = StoreMetadata::deserialize(deserializer)?;
if metadata.is_domain_store() {
Ok(Self(metadata))
} else {
Err(serde::de::Error::custom("expected Domain Store metadata"))
}
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct StorageCatalogMetadata(StoreMetadata);
impl StorageCatalogMetadata {
#[must_use]
pub fn v1() -> Self {
Self(StoreMetadata::storage_catalog_v1())
}
#[must_use]
pub const fn as_metadata(&self) -> &StoreMetadata {
&self.0
}
}
impl<'de> Deserialize<'de> for StorageCatalogMetadata {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let metadata = StoreMetadata::deserialize(deserializer)?;
if metadata.is_storage_catalog() {
Ok(Self(metadata))
} else {
Err(serde::de::Error::custom(
"expected Storage Catalog metadata",
))
}
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct Domain {
pub domain_id: DomainId,
pub profile_name: String,
pub canonical_workspace: HostPath,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub enum DurableDecision {
None,
Preparing,
Published,
Clean,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct Generation {
pub domain_id: DomainId,
pub generation_id: GenerationId,
pub configuration_fingerprint: ConfigurationFingerprint,
pub coordinator_id: CoordinatorId,
pub network_sidecar_id: NetworkSidecarId,
pub sandbox_id: SandboxId,
pub durable_decision: DurableDecision,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub enum ParticipantRole {
PreparedLower,
StateAnchor,
GitRouting,
Network,
Helper,
Sandbox,
Publication,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub enum ParticipantState {
Intent,
LeaseHeld,
Prepared,
Committed,
ReleaseIntent,
Clean,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct HostProcessIdentity(PidfsProcessIdentityV1);
impl HostProcessIdentity {
#[must_use]
pub const fn pidfs_v1(identity: PidfsProcessIdentityV1) -> Self {
Self(identity)
}
#[must_use]
pub const fn pidfs(&self) -> &PidfsProcessIdentityV1 {
&self.0
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub enum PersistedHostProcessOwnership {
Unassigned,
Staged(PidfsProcessIdentityV1),
Committed(HostProcessIdentity),
}
impl PersistedHostProcessOwnership {
#[must_use]
pub const fn unassigned() -> Self {
Self::Unassigned
}
#[must_use]
pub const fn staged(identity: PidfsProcessIdentityV1) -> Self {
Self::Staged(identity)
}
#[must_use]
pub const fn committed(identity: PidfsProcessIdentityV1) -> Self {
Self::Committed(HostProcessIdentity::pidfs_v1(identity))
}
#[must_use]
pub const fn is_staged(&self) -> bool {
matches!(self, Self::Staged(_))
}
#[must_use]
pub const fn is_unassigned(&self) -> bool {
matches!(self, Self::Unassigned)
}
#[must_use]
pub const fn is_current_committed(&self) -> bool {
matches!(self, Self::Committed(_))
}
const fn is_initial_acquisition(&self) -> bool {
matches!(self, Self::Unassigned | Self::Staged(_))
}
const fn is_initial_or_current_committed(&self) -> bool {
self.is_initial_acquisition() || self.is_current_committed()
}
/// # Errors
/// Unassigned ownership has no positive authority.
pub fn authoritative_pidfs(&self) -> Result<&PidfsProcessIdentityV1, ProcessAuthorityRefusal> {
match self {
Self::Unassigned => Err(ProcessAuthorityRefusal::Unassigned),
Self::Staged(identity) => Ok(identity),
Self::Committed(identity) => Ok(identity.pidfs()),
}
}
/// # Errors
/// Unassigned or staged ownership is not serviceable.
pub fn serviceable_pidfs(&self) -> Result<&PidfsProcessIdentityV1, ProcessOwnershipRefusal> {
match self {
Self::Unassigned => Err(ProcessOwnershipRefusal::Unassigned),
Self::Staged(_) => Err(ProcessOwnershipRefusal::Staged),
Self::Committed(identity) => Ok(identity.pidfs()),
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ProcessOwnershipRefusal {
Unassigned,
Staged,
}
impl fmt::Display for ProcessOwnershipRefusal {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Unassigned => formatter.write_str("process ownership is unassigned"),
Self::Staged => formatter.write_str("process ownership is staged, not committed"),
}
}
}
impl std::error::Error for ProcessOwnershipRefusal {}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ProcessAuthorityRefusal {
Unassigned,
}
impl fmt::Display for ProcessAuthorityRefusal {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Unassigned => formatter.write_str("process ownership is unassigned"),
}
}
}
impl std::error::Error for ProcessAuthorityRefusal {}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub enum HostProcessOwnershipSlot {
SandboxHostCoordinator,
SandboxRuntimeSupervisor,
SandboxAttachedRunsc,
StateAnchorDaemon,
GitRoutingDaemon,
NetworkRuntimeSupervisor,
HelperConnectivityHelper,
HelperRuntimeSupervisor,
SessionTerminalHostCoordinator(SessionId),
SessionTerminalExecClient(SessionId),
EscalationHostProcess(RequestId),
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub enum ProcessOwnershipMessageVersion {
V1,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct StagedProcessOwnership {
version: ProcessOwnershipMessageVersion,
domain_id: DomainId,
generation_id: GenerationId,
coordinator_id: CoordinatorId,
slot: HostProcessOwnershipSlot,
identity: PidfsProcessIdentityV1,
}
impl StagedProcessOwnership {
#[must_use]
pub const fn v1(
domain_id: DomainId,
generation_id: GenerationId,
coordinator_id: CoordinatorId,
slot: HostProcessOwnershipSlot,
identity: PidfsProcessIdentityV1,
) -> Self {
Self {
version: ProcessOwnershipMessageVersion::V1,
domain_id,
generation_id,
coordinator_id,
slot,
identity,
}
}
#[must_use]
pub const fn domain_id(self) -> DomainId {
self.domain_id
}
#[must_use]
pub const fn generation_id(self) -> GenerationId {
self.generation_id
}
#[must_use]
pub const fn coordinator_id(self) -> CoordinatorId {
self.coordinator_id
}
#[must_use]
pub const fn slot(self) -> HostProcessOwnershipSlot {
self.slot
}
#[must_use]
pub const fn identity(self) -> PidfsProcessIdentityV1 {
self.identity
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct ProcessOwnershipCommitAck {
version: ProcessOwnershipMessageVersion,
committed: StagedProcessOwnership,
}
impl ProcessOwnershipCommitAck {
#[must_use]
pub const fn v1(committed: StagedProcessOwnership) -> Self {
Self {
version: ProcessOwnershipMessageVersion::V1,
committed,
}
}
#[must_use]
pub fn acknowledges(&self, staged: &StagedProcessOwnership) -> bool {
self.version == ProcessOwnershipMessageVersion::V1 && self.committed == *staged
}
#[must_use]
pub const fn committed(self) -> StagedProcessOwnership {
self.committed
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct Participant {
generation_id: GenerationId,
role: ParticipantRole,
state: ParticipantState,
ownership: Option<ParticipantOwnership>,
}
impl Participant {
#[must_use]
pub const fn bare_intent(generation_id: GenerationId, role: ParticipantRole) -> Self {
Self {
generation_id,
role,
state: ParticipantState::Intent,
ownership: None,
}
}
/// # Errors
/// The snapshot must be compatible with its role and state.
pub fn snapshot(
generation_id: GenerationId,
role: ParticipantRole,
state: ParticipantState,
ownership: Option<ParticipantOwnership>,
) -> Result<Self, RecordInvariantError> {
match (state, ownership.as_ref()) {
(ParticipantState::Intent, None) => {}
(ParticipantState::Intent, Some(snapshot))
if snapshot.role() == role && !snapshot.is_empty() && snapshot.is_intent() => {}
(
ParticipantState::LeaseHeld
| ParticipantState::Prepared
| ParticipantState::Committed,
Some(snapshot),
) if snapshot.role() == role && snapshot.is_active_for(state, generation_id) => {}
(ParticipantState::ReleaseIntent, Some(snapshot))
if snapshot.role() == role
&& !snapshot.is_empty()
&& snapshot.is_release_intent() => {}
(ParticipantState::Clean, Some(snapshot))
if snapshot.role() == role && snapshot.is_empty() => {}
_ => return Err(RecordInvariantError::ParticipantSnapshotState),
}
Ok(Self {
generation_id,
role,
state,
ownership,
})
}
#[must_use]
pub const fn generation_id(&self) -> GenerationId {
self.generation_id
}
#[must_use]
pub const fn role(&self) -> ParticipantRole {
self.role
}
#[must_use]
pub const fn state(&self) -> ParticipantState {
self.state
}
#[must_use]
pub const fn ownership(&self) -> Option<&ParticipantOwnership> {
self.ownership.as_ref()
}
/// # Errors
/// The transition must preserve evidence and follow the lifecycle.
pub fn validate_transition_to(&self, next: &Self) -> Result<(), RecordTransitionError> {
self.validate_identity(next)?;
match (self.state, next.state) {
(ParticipantState::Intent, ParticipantState::Intent) => {
validate_participant_progress(self.ownership.as_ref(), next.ownership.as_ref())
}
(
ParticipantState::Intent,
ParticipantState::LeaseHeld
| ParticipantState::Prepared
| ParticipantState::Committed,
) => validate_participant_activation(self.ownership.as_ref(), next.ownership.as_ref()),
(
ParticipantState::Intent
| ParticipantState::LeaseHeld
| ParticipantState::Prepared
| ParticipantState::Committed,
ParticipantState::ReleaseIntent,
) => validate_participant_release(self.ownership.as_ref(), next.ownership.as_ref()),
(old_state, new_state)
if old_state == new_state
&& matches!(
old_state,
ParticipantState::LeaseHeld
| ParticipantState::Prepared
| ParticipantState::Committed
| ParticipantState::ReleaseIntent
| ParticipantState::Clean
)
&& self.ownership == next.ownership =>
{
Ok(())
}
(_, ParticipantState::Clean) => Err(RecordTransitionError::AbsenceProofRequired),
_ => Err(RecordTransitionError::StateRegression),
}
}
/// # Errors
/// Absence must permit an atomic transition to explicit Clean.
pub fn validate_transition_to_clean_after_absence(
&self,
next: &Self,
) -> Result<(), RecordTransitionError> {
self.validate_identity(next)?;
if next.state != ParticipantState::Clean {
return Err(RecordTransitionError::InvalidCleanTarget);
}
let expected = match (self.state, self.ownership.as_ref()) {
(ParticipantState::Intent, None) => ParticipantOwnership::empty(self.role),
(ParticipantState::ReleaseIntent, Some(ownership)) => ownership.cleared(),
_ => return Err(RecordTransitionError::StateRegression),
};
if next.ownership.as_ref() == Some(&expected) {
Ok(())
} else {
Err(RecordTransitionError::EvidenceChanged)
}
}
fn validate_identity(&self, next: &Self) -> Result<(), RecordTransitionError> {
if self.generation_id == next.generation_id && self.role == next.role {
Ok(())
} else {
Err(RecordTransitionError::RecordIdentityChanged)
}
}
}
impl<'de> Deserialize<'de> for Participant {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
#[derive(Deserialize)]
struct Fields {
generation_id: GenerationId,
role: ParticipantRole,
state: ParticipantState,
ownership: Option<ParticipantOwnership>,
}
let fields = Fields::deserialize(deserializer)?;
Self::snapshot(
fields.generation_id,
fields.role,
fields.state,
fields.ownership,
)
.map_err(serde::de::Error::custom)
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub enum SessionPhase {
New,
Waiting,
Registering,
Attached,
Prompting,
HostExecuting,
Closed,
Failed,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct Session {
pub generation_id: GenerationId,
pub session_id: SessionId,
pub cli_id: ClientId,
pub session_anchor_id: SessionAnchorId,
pub session_phase: SessionPhase,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub enum SessionTerminalState {
Intent,
Running,
ReleaseIntent,
Clean,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct PtyIdentity {
pub master_fd: FileDescriptor,
pub device: DeviceId,
pub inode: InodeId,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct GuestProcessIdentity {
pub leader_pid: GuestProcessId,
pub leader_start_time: GuestProcessStartTimeTicks,
pub process_group_id: GuestProcessGroupId,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct SessionTerminal {
generation_id: GenerationId,
session_id: SessionId,
state: SessionTerminalState,
container_id: Option<ContainerId>,
host_coordinator: PersistedHostProcessOwnership,
pty_master: Option<PtyIdentity>,
exec_client: PersistedHostProcessOwnership,
guest_process: Option<GuestProcessIdentity>,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct SessionTerminalFields {
pub generation_id: GenerationId,
pub session_id: SessionId,
pub state: SessionTerminalState,
pub container_id: Option<ContainerId>,
pub host_coordinator: PersistedHostProcessOwnership,
pub pty_master: Option<PtyIdentity>,
pub exec_client: PersistedHostProcessOwnership,
pub guest_process: Option<GuestProcessIdentity>,
}
impl SessionTerminal {
/// # Errors
/// The evidence must be compatible with the terminal state.
pub fn snapshot(fields: SessionTerminalFields) -> Result<Self, RecordInvariantError> {
let is_empty = fields.container_id.is_none()
&& fields.host_coordinator.is_unassigned()
&& fields.pty_master.is_none()
&& fields.exec_client.is_unassigned()
&& fields.guest_process.is_none();
let is_running = fields.container_id.is_some()
&& fields.host_coordinator.is_current_committed()
&& fields.pty_master.is_some()
&& fields.exec_client.is_current_committed()
&& fields.guest_process.is_some();
match fields.state {
SessionTerminalState::Intent => {}
SessionTerminalState::Running if is_running => {}
SessionTerminalState::ReleaseIntent if !is_empty => {}
SessionTerminalState::Clean if is_empty => {}
_ => return Err(RecordInvariantError::SessionTerminalSnapshotState),
}
Ok(Self {
generation_id: fields.generation_id,
session_id: fields.session_id,
state: fields.state,
container_id: fields.container_id,
host_coordinator: fields.host_coordinator,
pty_master: fields.pty_master,
exec_client: fields.exec_client,
guest_process: fields.guest_process,
})
}
#[must_use]
pub const fn generation_id(&self) -> GenerationId {
self.generation_id
}
#[must_use]
pub const fn session_id(&self) -> SessionId {
self.session_id
}
#[must_use]
pub const fn state(&self) -> SessionTerminalState {
self.state
}
#[must_use]
pub const fn host_coordinator(&self) -> &PersistedHostProcessOwnership {
&self.host_coordinator
}
#[must_use]
pub const fn container_id(&self) -> Option<ContainerId> {
self.container_id
}
#[must_use]
pub const fn pty_master(&self) -> Option<PtyIdentity> {
self.pty_master
}
#[must_use]
pub const fn exec_client(&self) -> &PersistedHostProcessOwnership {
&self.exec_client
}
#[must_use]
pub const fn guest_process(&self) -> Option<GuestProcessIdentity> {
self.guest_process
}
/// # Errors
/// The transition must preserve evidence and follow the lifecycle.
pub fn validate_transition_to(&self, next: &Self) -> Result<(), RecordTransitionError> {
self.validate_identity(next)?;
match (self.state, next.state) {
(
SessionTerminalState::Intent,
SessionTerminalState::Intent | SessionTerminalState::Running,
) => self.validate_progress(next),
(
SessionTerminalState::Intent
| SessionTerminalState::Running
| SessionTerminalState::ReleaseIntent,
SessionTerminalState::ReleaseIntent,
)
| (SessionTerminalState::Clean, SessionTerminalState::Clean)
| (SessionTerminalState::Running, SessionTerminalState::Running)
if self.has_same_evidence(next) =>
{
Ok(())
}
(_, SessionTerminalState::Clean) => Err(RecordTransitionError::AbsenceProofRequired),
_ => Err(RecordTransitionError::StateRegression),
}
}
/// # Errors
/// Absence must permit an atomic transition to Clean.
pub fn validate_transition_to_clean_after_absence(
&self,
next: &Self,
) -> Result<(), RecordTransitionError> {
self.validate_identity(next)?;
let source_is_bare = self.state == SessionTerminalState::Intent && self.is_empty();
if !(source_is_bare || self.state == SessionTerminalState::ReleaseIntent) {
return Err(RecordTransitionError::StateRegression);
}
if next.state == SessionTerminalState::Clean && next.is_empty() {
Ok(())
} else {
Err(RecordTransitionError::InvalidCleanTarget)
}
}
fn validate_identity(&self, next: &Self) -> Result<(), RecordTransitionError> {
if self.generation_id == next.generation_id && self.session_id == next.session_id {
Ok(())
} else {
Err(RecordTransitionError::RecordIdentityChanged)
}
}
fn validate_progress(&self, next: &Self) -> Result<(), RecordTransitionError> {
validate_option_progress(self.container_id.as_ref(), next.container_id.as_ref())?;
validate_existing_process_progress(&self.host_coordinator, &next.host_coordinator)?;
validate_option_progress(self.pty_master.as_ref(), next.pty_master.as_ref())?;
validate_process_progress(&self.exec_client, &next.exec_client)?;
validate_option_progress(self.guest_process.as_ref(), next.guest_process.as_ref())
}
fn has_same_evidence(&self, next: &Self) -> bool {
self.container_id == next.container_id
&& self.host_coordinator == next.host_coordinator
&& self.pty_master == next.pty_master
&& self.exec_client == next.exec_client
&& self.guest_process == next.guest_process
}
fn is_empty(&self) -> bool {
self.container_id.is_none()
&& self.host_coordinator.is_unassigned()
&& self.pty_master.is_none()
&& self.exec_client.is_unassigned()
&& self.guest_process.is_none()
}
}
impl<'de> Deserialize<'de> for SessionTerminal {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
#[derive(Deserialize)]
struct Fields {
generation_id: GenerationId,
session_id: SessionId,
state: SessionTerminalState,
container_id: Option<ContainerId>,
host_coordinator: PersistedHostProcessOwnership,
pty_master: Option<PtyIdentity>,
exec_client: PersistedHostProcessOwnership,
guest_process: Option<GuestProcessIdentity>,
}
let fields = Fields::deserialize(deserializer)?;
Self::snapshot(SessionTerminalFields {
generation_id: fields.generation_id,
session_id: fields.session_id,
state: fields.state,
container_id: fields.container_id,
host_coordinator: fields.host_coordinator,
pty_master: fields.pty_master,
exec_client: fields.exec_client,
guest_process: fields.guest_process,
})
.map_err(serde::de::Error::custom)
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct Mailbox {
pub generation_id: GenerationId,
pub identity_path: HostPath,
pub inbox_path: HostPath,
pub outbox_path: HostPath,
pub state: ParticipantState,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct DomainStore {
pub metadata: DomainStoreMetadata,
pub domain: Domain,
pub generations: Vec<Generation>,
pub participants: Vec<Participant>,
pub sessions: Vec<Session>,
pub session_terminals: Vec<SessionTerminal>,
pub mailboxes: Vec<Mailbox>,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub enum ParticipantOwnership {
PreparedLower(PreparedLowerOwnership),
StateAnchor(StateAnchorOwnership),
GitRouting(GitRoutingOwnership),
Network(Box<NetworkOwnership>),
Helper(HelperOwnership),
Sandbox(SandboxOwnership),
Publication(PublicationOwnership),
}
impl ParticipantOwnership {
#[must_use]
pub const fn role(&self) -> ParticipantRole {
match self {
Self::PreparedLower(_) => ParticipantRole::PreparedLower,
Self::StateAnchor(_) => ParticipantRole::StateAnchor,
Self::GitRouting(_) => ParticipantRole::GitRouting,
Self::Network(_) => ParticipantRole::Network,
Self::Helper(_) => ParticipantRole::Helper,
Self::Sandbox(_) => ParticipantRole::Sandbox,
Self::Publication(_) => ParticipantRole::Publication,
}
}
#[must_use]
pub fn empty(role: ParticipantRole) -> Self {
match role {
ParticipantRole::PreparedLower => Self::PreparedLower(PreparedLowerOwnership::empty()),
ParticipantRole::StateAnchor => Self::StateAnchor(StateAnchorOwnership::empty()),
ParticipantRole::GitRouting => Self::GitRouting(GitRoutingOwnership::empty()),
ParticipantRole::Network => Self::Network(Box::new(NetworkOwnership::empty())),
ParticipantRole::Helper => Self::Helper(HelperOwnership::empty()),
ParticipantRole::Sandbox => Self::Sandbox(SandboxOwnership::empty()),
ParticipantRole::Publication => Self::Publication(PublicationOwnership::empty()),
}
}
#[must_use]
pub fn is_empty(&self) -> bool {
match self {
Self::PreparedLower(ownership) => ownership.is_empty(),
Self::StateAnchor(ownership) => ownership.is_empty(),
Self::GitRouting(ownership) => ownership.is_empty(),
Self::Network(ownership) => ownership.is_empty(),
Self::Helper(ownership) => ownership.is_empty(),
Self::Sandbox(ownership) => ownership.is_empty(),
Self::Publication(ownership) => ownership.is_empty(),
}
}
fn is_intent(&self) -> bool {
match self {
Self::PreparedLower(ownership) => ownership
.lease_state
.is_none_or(|state| state == ParticipantState::Intent),
_ => true,
}
}
fn is_release_intent(&self) -> bool {
match self {
Self::PreparedLower(ownership) => ownership
.lease_state
.is_none_or(|state| state == ParticipantState::ReleaseIntent),
_ => true,
}
}
fn is_active_for(&self, state: ParticipantState, generation_id: GenerationId) -> bool {
match (self, state) {
(Self::PreparedLower(ownership), ParticipantState::LeaseHeld) => {
ownership.is_lease_held()
}
(Self::StateAnchor(ownership), ParticipantState::Prepared) => ownership.is_prepared(),
(Self::GitRouting(ownership), ParticipantState::Prepared) => ownership.is_prepared(),
(Self::Network(ownership), ParticipantState::Committed) => {
ownership.is_committed(generation_id)
}
(Self::Helper(ownership), ParticipantState::Committed) => ownership.is_committed(),
(Self::Sandbox(ownership), ParticipantState::Committed) => ownership.is_committed(),
(Self::Publication(ownership), ParticipantState::Committed) => ownership.is_committed(),
_ => false,
}
}
fn cleared(&self) -> Self {
Self::empty(self.role())
}
fn validate_progress_to(&self, next: &Self) -> Result<(), RecordTransitionError> {
match (self, next) {
(Self::PreparedLower(old), Self::PreparedLower(new)) => old.validate_progress_to(new),
(Self::StateAnchor(old), Self::StateAnchor(new)) => old.validate_progress_to(new),
(Self::GitRouting(old), Self::GitRouting(new)) => old.validate_progress_to(new),
(Self::Network(old), Self::Network(new)) => old.validate_progress_to(new),
(Self::Helper(old), Self::Helper(new)) => old.validate_progress_to(new),
(Self::Sandbox(old), Self::Sandbox(new)) => old.validate_progress_to(new),
(Self::Publication(old), Self::Publication(new)) => old.validate_progress_to(new),
_ => Err(RecordTransitionError::RecordIdentityChanged),
}
}
fn is_initial_progress(&self) -> bool {
match self {
Self::PreparedLower(_) | Self::Publication(_) => true,
Self::StateAnchor(ownership) => ownership.daemon.is_initial_acquisition(),
Self::GitRouting(ownership) => ownership.daemon.is_initial_acquisition(),
Self::Network(ownership) => ownership
.runtime_supervisor
.is_initial_or_current_committed(),
Self::Helper(ownership) => {
ownership.connectivity_helper.is_initial_acquisition()
&& ownership
.runtime_supervisor
.is_initial_or_current_committed()
}
Self::Sandbox(ownership) => {
ownership.host_coordinator.is_initial_or_current_committed()
&& ownership.runtime_supervisor.is_initial_acquisition()
&& ownership.attached_runsc.is_initial_acquisition()
}
}
}
fn validate_release_to(&self, next: &Self) -> Result<(), RecordTransitionError> {
match (self, next) {
(Self::PreparedLower(old), Self::PreparedLower(new)) => old.validate_release_to(new),
_ if self == next => Ok(()),
_ => Err(RecordTransitionError::EvidenceChanged),
}
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct PreparedLowerOwnership {
pub prepared_artifact_id: Option<PreparedArtifactId>,
pub source_identity: Option<SourceIdentity>,
pub lease_state: Option<ParticipantState>,
pub path: Option<HostPath>,
pub device: Option<DeviceId>,
pub inode: Option<InodeId>,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct StateAnchorOwnership {
pub lower_artifact_id: Option<PreparedArtifactId>,
pub upper_path: Option<HostPath>,
pub work_path: Option<HostPath>,
pub merged_path: Option<HostPath>,
pub daemon: PersistedHostProcessOwnership,
pub user_namespace: Option<NamespaceIdentity>,
pub mount_namespace: Option<NamespaceIdentity>,
pub mount: Option<MountIdentity>,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct GitRoutingOwnership {
pub mountpoint: Option<HostPath>,
pub host_root: Option<HostPath>,
pub isolated_root: Option<HostPath>,
pub daemon: PersistedHostProcessOwnership,
pub mount: Option<MountIdentity>,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct NetworkOwnership {
pub derived_policy_digest: Option<DerivedPolicyDigest>,
pub readiness_fingerprint: Option<ReadinessFingerprint>,
pub readiness_evidence: Option<NetworkReadinessEvidence>,
pub namespace: Option<NamespaceIdentity>,
pub runtime_supervisor: PersistedHostProcessOwnership,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct HelperOwnership {
pub network_sidecar_id: Option<NetworkSidecarId>,
pub connectivity_helper: PersistedHostProcessOwnership,
pub runtime_supervisor: PersistedHostProcessOwnership,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct SandboxOwnership {
pub container_id: Option<ContainerId>,
pub runtime_root: Option<HostPath>,
pub bundle_digest: Option<BundleDigest>,
pub host_coordinator: PersistedHostProcessOwnership,
pub runtime_supervisor: PersistedHostProcessOwnership,
pub attached_runsc: PersistedHostProcessOwnership,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct PublicationOwnership {
pub socket_path: Option<HostPath>,
pub identity_path: Option<HostPath>,
}
impl PreparedLowerOwnership {
fn empty() -> Self {
Self {
prepared_artifact_id: None,
source_identity: None,
lease_state: None,
path: None,
device: None,
inode: None,
}
}
fn is_empty(&self) -> bool {
self.prepared_artifact_id.is_none()
&& self.source_identity.is_none()
&& self.lease_state.is_none()
&& self.path.is_none()
&& self.device.is_none()
&& self.inode.is_none()
}
fn is_lease_held(&self) -> bool {
self.prepared_artifact_id.is_some()
&& self.source_identity.is_some()
&& self.lease_state == Some(ParticipantState::LeaseHeld)
&& self.path.is_some()
&& self.device.is_some()
&& self.inode.is_some()
}
fn validate_progress_to(&self, next: &Self) -> Result<(), RecordTransitionError> {
validate_option_progress(
self.prepared_artifact_id.as_ref(),
next.prepared_artifact_id.as_ref(),
)?;
validate_option_progress(self.source_identity.as_ref(), next.source_identity.as_ref())?;
validate_prepared_lower_lease_progress(self.lease_state, next.lease_state)?;
validate_option_progress(self.path.as_ref(), next.path.as_ref())?;
validate_option_progress(self.device.as_ref(), next.device.as_ref())?;
validate_option_progress(self.inode.as_ref(), next.inode.as_ref())
}
fn validate_release_to(&self, next: &Self) -> Result<(), RecordTransitionError> {
if self.prepared_artifact_id != next.prepared_artifact_id
|| self.source_identity != next.source_identity
|| self.path != next.path
|| self.device != next.device
|| self.inode != next.inode
{
return Err(RecordTransitionError::EvidenceChanged);
}
match (self.lease_state, next.lease_state) {
(None, None)
| (Some(ParticipantState::ReleaseIntent), Some(ParticipantState::ReleaseIntent)) => {
Ok(())
}
(Some(_), Some(ParticipantState::ReleaseIntent)) => Ok(()),
_ => Err(RecordTransitionError::EvidenceChanged),
}
}
}
impl StateAnchorOwnership {
fn empty() -> Self {
Self {
lower_artifact_id: None,
upper_path: None,
work_path: None,
merged_path: None,
daemon: PersistedHostProcessOwnership::unassigned(),
user_namespace: None,
mount_namespace: None,
mount: None,
}
}
fn is_empty(&self) -> bool {
self.lower_artifact_id.is_none()
&& self.upper_path.is_none()
&& self.work_path.is_none()
&& self.merged_path.is_none()
&& self.daemon.is_unassigned()
&& self.user_namespace.is_none()
&& self.mount_namespace.is_none()
&& self.mount.is_none()
}
fn is_prepared(&self) -> bool {
self.lower_artifact_id.is_some()
&& self.upper_path.is_some()
&& self.work_path.is_some()
&& self.merged_path.is_some()
&& self.daemon.is_current_committed()
&& self.user_namespace.is_some()
&& self.mount_namespace.is_some()
&& self.mount.is_some()
}
fn validate_progress_to(&self, next: &Self) -> Result<(), RecordTransitionError> {
validate_option_progress(
self.lower_artifact_id.as_ref(),
next.lower_artifact_id.as_ref(),
)?;
validate_option_progress(self.upper_path.as_ref(), next.upper_path.as_ref())?;
validate_option_progress(self.work_path.as_ref(), next.work_path.as_ref())?;
validate_option_progress(self.merged_path.as_ref(), next.merged_path.as_ref())?;
validate_process_progress(&self.daemon, &next.daemon)?;
validate_option_progress(self.user_namespace.as_ref(), next.user_namespace.as_ref())?;
validate_option_progress(self.mount_namespace.as_ref(), next.mount_namespace.as_ref())?;
validate_option_progress(self.mount.as_ref(), next.mount.as_ref())
}
}
impl GitRoutingOwnership {
fn empty() -> Self {
Self {
mountpoint: None,
host_root: None,
isolated_root: None,
daemon: PersistedHostProcessOwnership::unassigned(),
mount: None,
}
}
fn is_empty(&self) -> bool {
self.mountpoint.is_none()
&& self.host_root.is_none()
&& self.isolated_root.is_none()
&& self.daemon.is_unassigned()
&& self.mount.is_none()
}
fn is_prepared(&self) -> bool {
self.mountpoint.is_some()
&& self.host_root.is_some()
&& self.isolated_root.is_some()
&& self.daemon.is_current_committed()
&& self.mount.is_some()
}
fn validate_progress_to(&self, next: &Self) -> Result<(), RecordTransitionError> {
validate_option_progress(self.mountpoint.as_ref(), next.mountpoint.as_ref())?;
validate_option_progress(self.host_root.as_ref(), next.host_root.as_ref())?;
validate_option_progress(self.isolated_root.as_ref(), next.isolated_root.as_ref())?;
validate_process_progress(&self.daemon, &next.daemon)?;
validate_option_progress(self.mount.as_ref(), next.mount.as_ref())
}
}
impl NetworkOwnership {
fn empty() -> Self {
Self {
derived_policy_digest: None,
readiness_fingerprint: None,
readiness_evidence: None,
namespace: None,
runtime_supervisor: PersistedHostProcessOwnership::unassigned(),
}
}
fn is_empty(&self) -> bool {
self.derived_policy_digest.is_none()
&& self.readiness_fingerprint.is_none()
&& self.readiness_evidence.is_none()
&& self.namespace.is_none()
&& self.runtime_supervisor.is_unassigned()
}
fn is_committed(&self, generation_id: GenerationId) -> bool {
let (
Some(derived_policy_digest),
Some(readiness_fingerprint),
Some(readiness_evidence),
Some(namespace),
Ok(runtime_supervisor),
) = (
self.derived_policy_digest,
self.readiness_fingerprint,
self.readiness_evidence.as_ref(),
self.namespace,
self.runtime_supervisor.serviceable_pidfs(),
)
else {
return false;
};
readiness_evidence.matches(
generation_id,
derived_policy_digest,
readiness_fingerprint,
namespace,
runtime_supervisor,
)
}
fn validate_progress_to(&self, next: &Self) -> Result<(), RecordTransitionError> {
validate_option_progress(
self.derived_policy_digest.as_ref(),
next.derived_policy_digest.as_ref(),
)?;
validate_option_progress(
self.readiness_fingerprint.as_ref(),
next.readiness_fingerprint.as_ref(),
)?;
validate_option_progress(
self.readiness_evidence.as_ref(),
next.readiness_evidence.as_ref(),
)?;
validate_option_progress(self.namespace.as_ref(), next.namespace.as_ref())?;
validate_existing_process_progress(&self.runtime_supervisor, &next.runtime_supervisor)
}
}
impl HelperOwnership {
fn empty() -> Self {
Self {
network_sidecar_id: None,
connectivity_helper: PersistedHostProcessOwnership::unassigned(),
runtime_supervisor: PersistedHostProcessOwnership::unassigned(),
}
}
fn is_empty(&self) -> bool {
self.network_sidecar_id.is_none()
&& self.connectivity_helper.is_unassigned()
&& self.runtime_supervisor.is_unassigned()
}
fn is_committed(&self) -> bool {
self.network_sidecar_id.is_some()
&& self.connectivity_helper.is_current_committed()
&& self.runtime_supervisor.is_current_committed()
}
fn validate_progress_to(&self, next: &Self) -> Result<(), RecordTransitionError> {
validate_option_progress(
self.network_sidecar_id.as_ref(),
next.network_sidecar_id.as_ref(),
)?;
validate_process_progress(&self.connectivity_helper, &next.connectivity_helper)?;
validate_existing_process_progress(&self.runtime_supervisor, &next.runtime_supervisor)
}
}
impl SandboxOwnership {
fn empty() -> Self {
Self {
container_id: None,
runtime_root: None,
bundle_digest: None,
host_coordinator: PersistedHostProcessOwnership::unassigned(),
runtime_supervisor: PersistedHostProcessOwnership::unassigned(),
attached_runsc: PersistedHostProcessOwnership::unassigned(),
}
}
fn is_empty(&self) -> bool {
self.container_id.is_none()
&& self.runtime_root.is_none()
&& self.bundle_digest.is_none()
&& self.host_coordinator.is_unassigned()
&& self.runtime_supervisor.is_unassigned()
&& self.attached_runsc.is_unassigned()
}
fn is_committed(&self) -> bool {
self.container_id.is_some()
&& self.runtime_root.is_some()
&& self.bundle_digest.is_some()
&& self.host_coordinator.is_current_committed()
&& self.runtime_supervisor.is_current_committed()
&& self.attached_runsc.is_current_committed()
}
fn validate_progress_to(&self, next: &Self) -> Result<(), RecordTransitionError> {
validate_option_progress(self.container_id.as_ref(), next.container_id.as_ref())?;
validate_option_progress(self.runtime_root.as_ref(), next.runtime_root.as_ref())?;
validate_option_progress(self.bundle_digest.as_ref(), next.bundle_digest.as_ref())?;
validate_existing_process_progress(&self.host_coordinator, &next.host_coordinator)?;
validate_process_progress(&self.runtime_supervisor, &next.runtime_supervisor)?;
validate_process_progress(&self.attached_runsc, &next.attached_runsc)
}
}
impl PublicationOwnership {
fn empty() -> Self {
Self {
socket_path: None,
identity_path: None,
}
}
fn is_empty(&self) -> bool {
self.socket_path.is_none() && self.identity_path.is_none()
}
fn is_committed(&self) -> bool {
self.socket_path.is_some() && self.identity_path.is_some()
}
fn validate_progress_to(&self, next: &Self) -> Result<(), RecordTransitionError> {
validate_option_progress(self.socket_path.as_ref(), next.socket_path.as_ref())?;
validate_option_progress(self.identity_path.as_ref(), next.identity_path.as_ref())
}
}
fn validate_participant_progress(
old: Option<&ParticipantOwnership>,
next: Option<&ParticipantOwnership>,
) -> Result<(), RecordTransitionError> {
match (old, next) {
(None, None) => Ok(()),
(None, Some(next)) if next.is_initial_progress() => Ok(()),
(None, Some(_)) => Err(RecordTransitionError::ProcessStageSkipped),
(Some(_), None) => Err(RecordTransitionError::EvidenceLost),
(Some(old), Some(next)) => old.validate_progress_to(next),
}
}
fn validate_participant_release(
old: Option<&ParticipantOwnership>,
next: Option<&ParticipantOwnership>,
) -> Result<(), RecordTransitionError> {
match (old, next) {
(Some(old), Some(next)) => old.validate_release_to(next),
_ => Err(RecordTransitionError::EvidenceChanged),
}
}
fn validate_participant_activation(
old: Option<&ParticipantOwnership>,
next: Option<&ParticipantOwnership>,
) -> Result<(), RecordTransitionError> {
match (old, next) {
(Some(old), Some(next)) => old.validate_progress_to(next),
_ => Err(RecordTransitionError::PriorAcquisitionSnapshotRequired),
}
}
fn validate_prepared_lower_lease_progress(
old: Option<ParticipantState>,
next: Option<ParticipantState>,
) -> Result<(), RecordTransitionError> {
match (old, next) {
(None, None | Some(ParticipantState::Intent))
| (
Some(ParticipantState::Intent),
Some(ParticipantState::Intent | ParticipantState::LeaseHeld),
)
| (Some(ParticipantState::LeaseHeld), Some(ParticipantState::LeaseHeld)) => Ok(()),
(Some(_), None) => Err(RecordTransitionError::EvidenceLost),
_ => Err(RecordTransitionError::EvidenceChanged),
}
}
fn validate_option_progress<T: Eq>(
old: Option<&T>,
next: Option<&T>,
) -> Result<(), RecordTransitionError> {
match (old, next) {
(None, None | Some(_)) => Ok(()),
(Some(_), None) => Err(RecordTransitionError::EvidenceLost),
(Some(old), Some(next)) if old == next => Ok(()),
(Some(_), Some(_)) => Err(RecordTransitionError::EvidenceChanged),
}
}
fn validate_process_progress(
old: &PersistedHostProcessOwnership,
next: &PersistedHostProcessOwnership,
) -> Result<(), RecordTransitionError> {
match (old, next) {
(
PersistedHostProcessOwnership::Unassigned,
PersistedHostProcessOwnership::Unassigned | PersistedHostProcessOwnership::Staged(_),
) => Ok(()),
(
PersistedHostProcessOwnership::Unassigned,
PersistedHostProcessOwnership::Committed(_),
) => Err(RecordTransitionError::ProcessStageSkipped),
(
PersistedHostProcessOwnership::Staged(_) | PersistedHostProcessOwnership::Committed(_),
PersistedHostProcessOwnership::Unassigned,
) => Err(RecordTransitionError::EvidenceLost),
(PersistedHostProcessOwnership::Committed(_), PersistedHostProcessOwnership::Staged(_)) => {
Err(RecordTransitionError::EvidenceLost)
}
(
PersistedHostProcessOwnership::Staged(old),
PersistedHostProcessOwnership::Staged(next),
) if old == next => Ok(()),
(
PersistedHostProcessOwnership::Staged(old),
PersistedHostProcessOwnership::Committed(next),
) if old == next.pidfs() => Ok(()),
(
PersistedHostProcessOwnership::Committed(old),
PersistedHostProcessOwnership::Committed(next),
) if old == next => Ok(()),
_ => Err(RecordTransitionError::EvidenceChanged),
}
}
fn validate_existing_process_progress(
old: &PersistedHostProcessOwnership,
next: &PersistedHostProcessOwnership,
) -> Result<(), RecordTransitionError> {
if matches!(
(old, next),
(
PersistedHostProcessOwnership::Unassigned,
PersistedHostProcessOwnership::Committed(_)
)
) {
Ok(())
} else {
validate_process_progress(old, next)
}
}