use super::{
state::{Event, Phase, State},
wire::{self, Binding, ContainerState, Operation, Outcome, Request, Response, Target},
};
use crate::{Invalid, Result};
use serde::{Deserialize, Serialize};
#[cfg(unix)]
mod storage;
#[cfg(unix)]
pub use storage::{
Dispatch, Journal, OperatorRecovery, Reservation, ResolutionDispatch, ServiceSlot,
};
mod resolution;
pub use resolution::CreatePlan;
use resolution::ResolutionProof;
const SCHEMA: &str = "gatehouse-lifecycle-lease/v1";
const MAX_ATTEMPTS: usize = 128;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum Purpose {
Service,
ServiceCleanup,
OperatorCleanup,
Reconcile,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Pending {
request: Request,
purpose: Purpose,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Attempt {
request: Request,
purpose: Purpose,
response: Option<Response>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
enum AbsenceReason {
OperatorObservedAbsence,
PostRemoveAbsence,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct AbsenceProof {
reason: AbsenceReason,
remove_request_sha256: Option<String>,
status_request_sha256: String,
response: Response,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Record {
schema: String,
create: Request,
resolution: Option<ResolutionProof>,
state: State,
revision: u64,
pending: Option<Pending>,
attempts: Vec<Attempt>,
absence: Option<AbsenceProof>,
recoveries: Vec<u64>,
}
fn same_authority(a: &Binding, b: &Binding) -> bool {
let mut a = a.clone();
a.invocation = b.invocation.clone();
a.operation_key = b.operation_key.clone();
a.generation = b.generation;
a == *b
}
impl Record {
fn new(create: &Request) -> Result<Self> {
Self::new_resolved(create, None)
}
fn new_resolved(create: &Request, resolution: Option<ResolutionProof>) -> Result<Self> {
create.validate()?;
if !matches!(create.operation, Operation::Create { .. })
|| create.binding.generation != if resolution.is_some() { 2 } else { 1 }
{
return Err(Invalid("journal requires first create generation"));
}
if let Some(proof) = &resolution {
proof.validate_create(create)?;
}
Ok(Self {
schema: SCHEMA.into(),
create: create.clone(),
resolution,
state: State::pending_create(),
revision: create.binding.generation,
pending: Some(Pending {
request: create.clone(),
purpose: Purpose::Service,
}),
attempts: vec![],
absence: None,
recoveries: vec![],
})
}
fn marker(&self) -> &str {
self.resolution
.as_ref()
.map(|p| p.request.binding.operation_key.as_str())
.unwrap_or(&self.create.binding.operation_key)
}
fn target(&self) -> Result<Target> {
let Operation::Create {
name,
labels,
container_port,
..
} = &self.create.operation
else {
return Err(Invalid("invalid journal create"));
};
Ok(Target {
object_id: self
.state
.object_id()
.ok_or(Invalid("journal object unresolved"))?
.into(),
name: name.clone(),
labels: labels.clone(),
container_port: *container_port,
})
}
fn check_request(&self, r: &Request) -> Result<()> {
r.validate()?;
if !same_authority(&r.binding, &self.create.binding) {
return Err(Invalid("journal authority mismatch"));
}
let target = match &r.operation {
Operation::Status { target }
| Operation::Start { target }
| Operation::Stop { target, .. }
| Operation::Remove { target }
| Operation::HealthExec { target, .. } => target,
_ => return Err(Invalid("journal operation unsupported after creation")),
};
if *target != self.target()? {
return Err(Invalid("journal target mismatch"));
}
Ok(())
}
fn validate(&self) -> Result<()> {
if self.schema != SCHEMA || self.attempts.len() > MAX_ATTEMPTS {
return Err(Invalid("invalid lifecycle journal"));
}
self.create.validate()?;
self.state.validate()?;
if !matches!(self.create.operation, Operation::Create { .. })
|| self.create.binding.generation != if self.resolution.is_some() { 2 } else { 1 }
{
return Err(Invalid("invalid journal origin"));
}
self.validate_history()?;
self.validate_absence()?;
self.replay()?;
Ok(())
}
fn validate_history(&self) -> Result<()> {
let base = self.create.binding.generation - 1;
let mut revision = base;
let mut nonces = std::collections::BTreeSet::new();
if let Some(proof) = &self.resolution {
proof.validate_create(&self.create)?;
nonces.insert(&proof.request.binding.invocation);
}
for a in &self.attempts {
if a.request.binding.generation != revision + 1
|| !nonces.insert(&a.request.binding.invocation)
{
return Err(Invalid("journal generation or nonce replay"));
}
a.request.validate()?;
if revision == base {
if a.request != self.create || a.purpose != Purpose::Service {
return Err(Invalid("journal creation history mismatch"));
}
match a.response.as_ref().map(|r| &r.outcome) {
Some(Outcome::Created { object })
if self.state.object_id() == Some(object.object_id.as_str()) => {}
None | Some(Outcome::Unknown { .. } | Outcome::RejectedBeforeEffect { .. })
if self.state.object_id().is_none() => {}
_ => return Err(Invalid("journal object has no creation evidence")),
}
} else {
self.check_request(&a.request)?;
}
if !same_authority(&a.request.binding, &self.create.binding) {
return Err(Invalid("journal historical authority mismatch"));
}
if let Some(r) = &a.response {
r.validate(&a.request)?;
}
revision += 1;
}
revision = self.validate_pending(base, revision, &mut nonces)?;
if revision != self.revision {
return Err(Invalid("journal revision mismatch"));
}
Ok(())
}
fn validate_pending<'a>(
&'a self,
base: u64,
mut revision: u64,
nonces: &mut std::collections::BTreeSet<&'a String>,
) -> Result<u64> {
if let Some(p) = &self.pending {
if p.request.binding.generation != revision + 1
|| !nonces.insert(&p.request.binding.invocation)
{
return Err(Invalid("journal pending replay"));
}
p.request.validate()?;
if revision == base {
if p.request != self.create
|| p.purpose != Purpose::Service
|| self.state.phase() != Phase::PendingCreate
{
return Err(Invalid("journal initial intent mismatch"));
}
} else {
self.check_request(&p.request)?;
}
if !same_authority(&p.request.binding, &self.create.binding) {
return Err(Invalid("journal pending authority mismatch"));
}
revision += 1;
}
Ok(revision)
}
fn validate_absence(&self) -> Result<()> {
if self.state.phase() == Phase::VerifiedAbsent {
let a = self
.attempts
.last()
.ok_or(Invalid("missing absence attempt"))?;
if a.purpose == Purpose::Service
|| (a.purpose == Purpose::Reconcile
&& !self
.attempts
.iter()
.any(|old| matches!(old.request.operation, Operation::Remove { .. })))
|| self.pending.is_some()
|| self.absence.as_ref().map(|p| &p.response) != a.response.as_ref()
|| !matches!(
self.absence.as_ref().map(|p| &p.response.outcome),
Some(Outcome::Absent { .. })
)
{
return Err(Invalid("missing journal absence evidence"));
}
} else if self.absence.is_some() {
return Err(Invalid("premature absence evidence"));
}
Ok(())
}
fn replay(&self) -> Result<()> {
if self.recoveries.windows(2).any(|w| w[0] >= w[1])
|| self.recoveries.iter().any(|g| {
*g < self.create.binding.generation
|| *g > self.attempts.len() as u64 + self.create.binding.generation - 1
})
{
return Err(Invalid("invalid journal recovery history"));
}
let mut replay = Self::new_resolved(&self.create, self.resolution.clone())?;
let mut fresh = false;
for (index, a) in self.attempts.iter().enumerate() {
if index > 0 {
replay = replay.begin_inner(&a.request, a.purpose, fresh)?;
}
let bytes = a
.response
.as_ref()
.map(|r| {
super::canonical::encode(
&serde_json::to_value(r).map_err(|_| Invalid("invalid stored response"))?,
)
})
.transpose()?;
replay = replay.observe_inner(bytes.as_deref())?;
fresh = matches!(
a.response.as_ref().map(|r| (&r.operation, &r.outcome)),
Some((Operation::Status { .. }, Outcome::Status { .. }))
);
if self.recoveries.contains(&replay.revision) {
replay.state = replay.state.recover()?;
replay.recoveries.push(replay.revision);
fresh = false;
}
}
if !self.attempts.is_empty() {
if let Some(p) = &self.pending {
replay = replay.begin_inner(&p.request, p.purpose, fresh)?;
}
}
if replay != *self {
return Err(Invalid(
"journal state does not follow its evidence history",
));
}
Ok(())
}
fn begin(&self, r: &Request, purpose: Purpose, fresh_status: bool) -> Result<Self> {
self.validate()?;
let next = self.begin_inner(r, purpose, fresh_status)?;
next.validate()?;
Ok(next)
}
fn normal_cleanup(&self) -> bool {
self.state.phase() == Phase::Running
|| (self.state.phase() == Phase::Stopped
&& self
.attempts
.iter()
.rev()
.find(|a| !matches!(a.request.operation, Operation::Status { .. }))
.is_some_and(|a| {
a.purpose == Purpose::ServiceCleanup
&& matches!(
a.response.as_ref().map(|r| &r.outcome),
Some(Outcome::Stopped { .. })
)
}))
}
fn begin_inner(&self, r: &Request, purpose: Purpose, fresh_status: bool) -> Result<Self> {
self.check_request(r)?;
if self.pending.is_some()
|| self.attempts.len() >= MAX_ATTEMPTS
|| self.state.phase() == Phase::VerifiedAbsent
|| r.binding.generation
!= self
.revision
.checked_add(1)
.ok_or(Invalid("generation overflow"))?
|| self
.attempts
.iter()
.any(|a| a.request.binding.invocation == r.binding.invocation)
{
return Err(Invalid("journal request replay or unfinished attempt"));
}
let mut next = self.clone();
next.state = self.begin_state(r, purpose, fresh_status)?;
next.revision = r.binding.generation;
next.pending = Some(Pending {
request: r.clone(),
purpose,
});
Ok(next)
}
fn cleanup_authorized(&self, purpose: Purpose) -> bool {
purpose == Purpose::OperatorCleanup
|| (purpose == Purpose::ServiceCleanup && self.normal_cleanup())
}
fn begin_state(&self, r: &Request, purpose: Purpose, fresh_status: bool) -> Result<State> {
let status = self
.attempts
.last()
.filter(|_| fresh_status)
.and_then(|a| a.response.as_ref())
.filter(|r| matches!(r.operation, Operation::Status { .. }));
let observed = match status.map(|r| &r.outcome) {
Some(Outcome::Status { object }) => Some(object.state),
_ => None,
};
let state = match &r.operation {
Operation::Status { .. } => {
if matches!(
self.state.phase(),
Phase::Retained | Phase::ReconcilingRemove
) && matches!(purpose, Purpose::Service | Purpose::ServiceCleanup)
{
return Err(Invalid("recovered lease requires explicit reconciliation"));
}
self.state.clone()
}
Operation::Start { .. }
if purpose == Purpose::Service && observed == Some(ContainerState::Created) =>
{
self.state.transition(Event::BeginStart)?
}
Operation::HealthExec { .. }
if purpose == Purpose::Service && observed == Some(ContainerState::Running) =>
{
self.state.transition(Event::BeginHealth)?
}
Operation::Stop { .. }
if self.cleanup_authorized(purpose)
&& observed == Some(ContainerState::Running) =>
{
let ev = if self.state.phase() == Phase::ReconcilingRemove {
Event::OperatorCleanupRunningVerified
} else {
Event::CleanupRunningVerified
};
self.state.transition(ev)?
}
Operation::Remove { .. }
if self.cleanup_authorized(purpose)
&& matches!(
observed,
Some(ContainerState::Created | ContainerState::Exited)
) =>
{
let stopped = if self.state.phase() == Phase::Stopped {
self.state.clone()
} else {
self.state
.transition(if self.state.phase() == Phase::ReconcilingRemove {
Event::OperatorCleanupTerminalVerified
} else {
Event::CleanupTerminalVerified
})?
};
stopped.transition(Event::BeginRemove)?
}
_ => {
return Err(Invalid(
"journal mutation lacks fresh evidence or authorization",
))
}
};
Ok(state)
}
fn observe(&self, bytes: Option<&[u8]>) -> Result<Self> {
self.validate()?;
let next = self.observe_inner(bytes)?;
next.validate()?;
Ok(next)
}
fn observe_inner(&self, bytes: Option<&[u8]>) -> Result<Self> {
let pending = self
.pending
.as_ref()
.ok_or(Invalid("no journal request pending"))?;
let response = bytes
.map(|b| wire::parse_response(b, &pending.request))
.transpose()?;
let mut next = self.clone();
let outcome = response.as_ref().map(|r| &r.outcome);
self.validate_observed_image(outcome)?;
next.state = match (&pending.request.operation, outcome) {
(_, None | Some(Outcome::Unknown { .. } | Outcome::RejectedBeforeEffect { .. })) => {
self.state.recover()?
}
(Operation::Create { .. }, Some(Outcome::Created { object })) => self
.state
.transition(Event::Created(object.object_id.clone()))?,
(Operation::Status { .. }, Some(Outcome::Status { .. })) => {
if matches!(self.state.phase(), Phase::AwaitingAbsence | Phase::Removing) {
if !matches!(
pending.purpose,
Purpose::OperatorCleanup | Purpose::Reconcile
) {
return Err(Invalid("present removal requires reconciliation purpose"));
}
self.state.recover()?
} else {
self.state.clone()
}
}
(Operation::Status { .. }, Some(Outcome::Absent { .. })) => {
let (state, proof) = self.observe_absence(
pending,
response
.clone()
.ok_or(Invalid("missing absence response"))?,
)?;
next.absence = Some(proof);
state
}
(Operation::Start { .. }, Some(Outcome::Started { .. })) => {
self.state.transition(Event::RunningVerified)?
}
(Operation::Stop { .. }, Some(Outcome::Stopped { .. })) => {
self.state.transition(Event::StoppedVerified)?
}
(Operation::Remove { .. }, Some(Outcome::Removed)) => {
self.state.transition(Event::RemoveClaimed)?
}
(Operation::HealthExec { .. }, Some(Outcome::Health { exit_code: 0, .. })) => {
self.state.transition(Event::HealthVerified)?
}
(Operation::HealthExec { .. }, Some(Outcome::Health { .. })) => {
self.state.transition(Event::Failed)?
}
_ => return Err(Invalid("journal outcome cannot advance state")),
};
next.attempts.push(Attempt {
request: pending.request.clone(),
purpose: pending.purpose,
response,
});
next.pending = None;
Ok(next)
}
fn observe_absence(
&self,
pending: &Pending,
response: Response,
) -> Result<(State, AbsenceProof)> {
let ev = match (self.state.phase(), pending.purpose) {
(
Phase::AwaitingAbsence | Phase::ReconcilingRemove,
Purpose::Reconcile | Purpose::OperatorCleanup,
) => Event::AbsenceVerified,
(Phase::Retained, Purpose::OperatorCleanup) => Event::OperatorAbsenceVerified,
_ => return Err(Invalid("absence lacks reconciliation authorization")),
};
let remove = self
.attempts
.iter()
.rev()
.find(|a| matches!(a.request.operation, Operation::Remove { .. }))
.map(|a| a.request.binding.operation_key.clone());
let proof = AbsenceProof {
reason: if remove.is_some() {
AbsenceReason::PostRemoveAbsence
} else {
AbsenceReason::OperatorObservedAbsence
},
remove_request_sha256: remove,
status_request_sha256: pending.request.binding.operation_key.clone(),
response,
};
Ok((self.state.transition(ev)?, proof))
}
fn validate_observed_image(&self, outcome: Option<&Outcome>) -> Result<()> {
if let Some(
Outcome::Created { object }
| Outcome::Status { object }
| Outcome::Started { object }
| Outcome::Stopped { object }
| Outcome::Health { object, .. },
) = outcome
{
if let Operation::Create { image_id, .. } = &self.create.operation {
if &object.image_id != image_id {
return Err(Invalid("journal image changed"));
}
}
}
Ok(())
}
fn recover(&self) -> Result<Self> {
let mut next = if self.pending.is_some() {
self.observe(None)?
} else {
self.clone()
};
let recovered = next.state.recover()?;
if recovered != next.state {
next.recoveries.push(next.revision);
next.state = recovered;
}
next.validate()?;
Ok(next)
}
}