use super::{
MergedSourcePosition, ObserverProgressLineage, ObserverProgressOccurrence,
ObserverProgressSourceIdentity, ObserverProgressSourceMetadata, ObserverProgressSourceWitness,
ObserverProgressWitnessState, validate_metadata,
};
use crate::server::participant::production::restore_checkpoint::{
CheckpointError, CheckpointPositions,
};
use crate::server::participant::production::state::checkpoint_body::invalid;
use liminal_protocol::lifecycle::{ObserverProgressProjection, OwnerCheckpointRestoreError};
use liminal_protocol::lifecycle::{ObserverProjectionCheckpointKind, RestoredLifecycle};
use liminal_protocol::wire::{ConversationId, DeliverySeq};
use std::collections::{BTreeMap, BTreeSet};
#[derive(Debug, thiserror::Error)]
#[error("native observer checkpoint source refused: {0:?}")]
struct NativeObserverSourceError(OwnerCheckpointRestoreError);
fn native_error(source: OwnerCheckpointRestoreError) -> CheckpointError {
CheckpointError::LiveState {
source: Box::new(NativeObserverSourceError(source)),
}
}
#[derive(serde::Serialize, serde::Deserialize)]
struct SavedWitness {
projection: Vec<u8>,
source: Vec<u8>,
merged_position: MergedSourcePosition,
metadata: ObserverProgressSourceMetadata,
}
#[derive(serde::Serialize, serde::Deserialize)]
struct SavedState {
version: u8,
conversation_id: ConversationId,
witnesses: Vec<SavedWitness>,
pruned_sources: Vec<SavedWitness>,
occurrences: Vec<ObserverProgressOccurrence>,
lineage_progress: Vec<(ObserverProgressLineage, DeliverySeq)>,
last_merged_position: Option<MergedSourcePosition>,
pruned_through: DeliverySeq,
}
impl ObserverProgressSourceWitness {
fn saved(&self) -> Result<SavedWitness, CheckpointError> {
self.validate_source(self.metadata.occurrence.conversation_id())?;
Ok(SavedWitness {
projection: self.projection.encode_checkpoint().to_vec(),
source: self
.projection
.encode_checkpoint_source()
.map_err(|_| self.missing_source())?,
merged_position: self.merged_position,
metadata: self.metadata,
})
}
const fn missing_source(&self) -> CheckpointError {
CheckpointError::MissingObserverProgressProjectionSource {
conversation_id: self.metadata.occurrence.conversation_id(),
participant_id: Some(self.metadata.occurrence.participant_id()),
base_log_head: Some(self.merged_position.base_log_head),
extension_sequence: Some(self.merged_position.extension_sequence),
witness_progress: Some(self.progress()),
pruned_through: None,
encoded_section_bytes: None,
}
}
fn validate_source(&self, conversation_id: ConversationId) -> Result<(), CheckpointError> {
validate_metadata(self.progress(), self.metadata)
.map_err(|_| invalid("observer witness metadata"))?;
if self.projection.conversation_id() != conversation_id
|| self.metadata.occurrence.conversation_id() != conversation_id
|| self.merged_position != self.metadata.source.merged_position()
|| self.projection.checkpoint_source_participant_id()
!= Some(self.metadata.occurrence.participant_id())
{
return Err(self.missing_source());
}
let matches = matches!(
(
self.projection.checkpoint_source_kind(),
self.metadata.occurrence,
),
(
Some(
ObserverProjectionCheckpointKind::DetachedTerminal
| ObserverProjectionCheckpointKind::DiedTerminal,
),
ObserverProgressOccurrence::AttachedTerminal { .. }
| ObserverProgressOccurrence::DetachedTerminal { .. }
| ObserverProgressOccurrence::DiedTerminal { .. },
) | (
Some(ObserverProjectionCheckpointKind::ParticipantAck),
ObserverProgressOccurrence::ParticipantAck { .. },
) | (
Some(ObserverProjectionCheckpointKind::MarkerAck),
ObserverProgressOccurrence::MarkerAck { .. },
) | (
Some(ObserverProjectionCheckpointKind::Leave),
ObserverProgressOccurrence::Leave { .. },
) | (
Some(ObserverProjectionCheckpointKind::RecoveredFate),
ObserverProgressOccurrence::RecoveredFate { .. },
)
);
if !matches {
return Err(self.missing_source());
}
let associated = match self.metadata.occurrence {
ObserverProgressOccurrence::DetachedTerminal { .. } => {
self.projection.checkpoint_source_kind()
== Some(ObserverProjectionCheckpointKind::DetachedTerminal)
}
ObserverProgressOccurrence::DiedTerminal { .. } => {
self.projection.checkpoint_source_kind()
== Some(ObserverProjectionCheckpointKind::DiedTerminal)
}
ObserverProgressOccurrence::RecoveredFate { binding_epoch, .. } => {
self.projection.checkpoint_source_binding_epoch() == Some(binding_epoch)
}
_ => true,
};
if !associated {
return Err(self.missing_source());
}
self.projection
.encode_checkpoint_source()
.map_err(|_| self.missing_source())?;
Ok(())
}
}
impl SavedWitness {
fn restore(
self,
positions: &CheckpointPositions,
lifecycle: &RestoredLifecycle,
) -> Result<ObserverProgressSourceWitness, CheckpointError> {
let projection = lifecycle
.restore_checkpoint_observer_projection(
&self.source,
self.metadata.occurrence.participant_id(),
)
.map_err(native_error)?;
let projection =
ObserverProgressProjection::restore_checkpoint(&self.projection, projection)
.map_err(native_error)?;
let witness = ObserverProgressSourceWitness {
projection,
merged_position: self.merged_position,
metadata: self.metadata,
};
witness.validate_source(positions.conversation_id)?;
let valid = match witness.metadata.source {
ObserverProgressSourceIdentity::Base { sequence, .. } => {
sequence < positions.next_log_sequence
}
ObserverProgressSourceIdentity::Extension {
base_log_head,
extension_sequence,
..
} => {
base_log_head <= positions.next_log_sequence
&& extension_sequence < positions.extension_sequence
}
};
if !valid || witness.progress() > positions.observer_progress {
return Err(invalid("observer witness checkpoint coverage"));
}
Ok(witness)
}
}
impl ObserverProgressWitnessState {
pub(in crate::server::participant::production) fn validate_checkpoint_sources(
&self,
conversation_id: ConversationId,
) -> Result<(), CheckpointError> {
if self.active_visit {
return Err(invalid("observer active source visit"));
}
if self.pruned_through != 0 && self.pruned_sources.is_empty() {
return Err(CheckpointError::MissingObserverProgressProjectionSource {
conversation_id,
participant_id: None,
base_log_head: None,
extension_sequence: None,
witness_progress: None,
pruned_through: Some(self.pruned_through),
encoded_section_bytes: None,
});
}
for witness in self.witnesses.iter().chain(self.pruned_sources.values()) {
witness.validate_source(conversation_id)?;
}
self.validate_checkpoint_bookkeeping()
}
pub(in crate::server::participant::production) fn validate_checkpoint_lifecycle(
&self,
lifecycle: &RestoredLifecycle,
) -> Result<(), CheckpointError> {
for witness in self.witnesses.iter().chain(self.pruned_sources.values()) {
let source = witness
.projection
.encode_checkpoint_source()
.map_err(|_| witness.missing_source())?;
let restored = lifecycle
.restore_checkpoint_observer_projection(
&source,
witness.metadata.occurrence.participant_id(),
)
.map_err(native_error)?;
if restored.encode_checkpoint() != witness.projection.encode_checkpoint() {
return Err(invalid("observer capture projection association"));
}
}
Ok(())
}
fn validate_checkpoint_bookkeeping(&self) -> Result<(), CheckpointError> {
if !self
.witnesses
.windows(2)
.all(|pair| pair[0].merged_position < pair[1].merged_position)
{
return Err(invalid("observer witness source order"));
}
let mut lineage = BTreeMap::new();
let mut occurrences = BTreeSet::new();
let mut recovered = BTreeMap::new();
let mut positions = BTreeSet::new();
for witness in self.pruned_sources.values().chain(self.witnesses.iter()) {
if !positions.insert(witness.merged_position) {
return Err(invalid("observer duplicate merged source"));
}
let previous = lineage.insert(
witness.metadata.lineage,
(witness.progress(), witness.merged_position),
);
if previous.is_some_and(|(progress, position)| {
progress > witness.progress() || position >= witness.merged_position
}) {
return Err(invalid("observer source lineage regression"));
}
if witness.metadata.occurrence.prunable() {
if witness.progress() > self.pruned_through {
occurrences.insert(witness.metadata.occurrence);
}
} else {
let ObserverProgressOccurrence::RecoveredFate {
participant_id,
binding_epoch,
..
} = witness.metadata.occurrence
else {
return Err(invalid("observer recovered occurrence"));
};
if recovered
.insert(participant_id, witness.metadata.occurrence)
.is_some_and(|previous: ObserverProgressOccurrence| match previous {
ObserverProgressOccurrence::RecoveredFate {
binding_epoch: held,
..
} => binding_epoch.capability_generation <= held.capability_generation,
_ => true,
})
{
return Err(invalid("observer recovered source generation"));
}
}
}
occurrences.extend(recovered.into_values());
let lineage: BTreeMap<_, _> = lineage
.into_iter()
.map(|(key, (progress, _))| (key, progress))
.collect();
if lineage != self.lineage_progress || occurrences != self.occurrences {
return Err(invalid("observer bookkeeping source association"));
}
if positions.last().copied() != self.last_merged_position {
return Err(invalid("observer last merged source"));
}
let pruned_maximum = self
.pruned_sources
.values()
.map(ObserverProgressSourceWitness::progress)
.max()
.unwrap_or(0);
if pruned_maximum != self.pruned_through {
return Err(invalid("observer pruned source association"));
}
self.validate_pruned_source_lineages()
}
fn validate_pruned_source_lineages(&self) -> Result<(), CheckpointError> {
for (lineage, source) in &self.pruned_sources {
if source.metadata.lineage != *lineage || source.progress() > self.pruned_through {
return Err(invalid("observer pruned lineage source"));
}
}
if !self.pruned_sources.is_empty()
&& self
.witnesses
.iter()
.any(|witness| witness.progress() <= self.pruned_through)
{
return Err(invalid("observer witness below prune boundary"));
}
Ok(())
}
fn checkpoint_maximum(&self) -> DeliverySeq {
self.witnesses
.iter()
.chain(self.pruned_sources.values())
.map(ObserverProgressSourceWitness::progress)
.max()
.unwrap_or(0)
}
pub(in crate::server::participant::production) fn encode_checkpoint(
&self,
conversation_id: ConversationId,
observer_progress: DeliverySeq,
) -> Result<Vec<u8>, CheckpointError> {
self.validate_checkpoint_sources(conversation_id)?;
if self.checkpoint_maximum() != observer_progress {
return Err(invalid("observer capture maximum"));
}
if self.checkpoint_is_pristine() {
return Ok(Vec::new());
}
let state = SavedState {
version: 1,
conversation_id,
witnesses: self
.witnesses
.iter()
.map(ObserverProgressSourceWitness::saved)
.collect::<Result<_, _>>()?,
pruned_sources: self
.pruned_sources
.values()
.map(ObserverProgressSourceWitness::saved)
.collect::<Result<_, _>>()?,
occurrences: self.occurrences.iter().copied().collect(),
lineage_progress: self
.lineage_progress
.iter()
.map(|(&lineage, &progress)| (lineage, progress))
.collect(),
last_merged_position: self.last_merged_position,
pruned_through: self.pruned_through,
};
serde_json::to_vec(&state).map_err(|_| invalid("observer witness encoding"))
}
pub(in crate::server::participant::production) fn decode_checkpoint(
bytes: &[u8],
positions: &CheckpointPositions,
lifecycle: &RestoredLifecycle,
) -> Result<Self, CheckpointError> {
if bytes.is_empty() {
if positions.observer_progress != 0 {
return Err(invalid("missing observer witness state"));
}
return Ok(Self::new());
}
let saved: SavedState =
serde_json::from_slice(bytes).map_err(|_| invalid("observer witness framing"))?;
if saved.version != 1
|| saved.conversation_id != positions.conversation_id
|| serde_json::to_vec(&saved)
.map_err(|_| invalid("observer witness canonical encoding"))?
!= bytes
{
return Err(invalid("observer witness canonical association"));
}
let mut state = Self::new();
state.witnesses = saved
.witnesses
.into_iter()
.map(|witness| witness.restore(positions, lifecycle))
.collect::<Result<_, _>>()?;
for saved in saved.pruned_sources {
let witness = saved.restore(positions, lifecycle)?;
if state
.pruned_sources
.last_key_value()
.is_some_and(|(last, _)| *last >= witness.metadata.lineage)
{
return Err(invalid("observer pruned source order"));
}
state
.pruned_sources
.insert(witness.metadata.lineage, witness);
}
if !saved.occurrences.windows(2).all(|pair| pair[0] < pair[1])
|| !saved
.lineage_progress
.windows(2)
.all(|pair| pair[0].0 < pair[1].0)
{
return Err(invalid("observer bookkeeping order"));
}
state.occurrences = saved.occurrences.into_iter().collect();
state.lineage_progress = saved.lineage_progress.into_iter().collect();
state.last_merged_position = saved.last_merged_position;
state.pruned_through = saved.pruned_through;
state.validate_checkpoint_sources(positions.conversation_id)?;
let maximum = state
.witnesses
.iter()
.chain(state.pruned_sources.values())
.map(ObserverProgressSourceWitness::progress)
.max()
.unwrap_or(0);
if maximum != positions.observer_progress {
return Err(invalid("observer checkpoint maximum"));
}
if state.encode_checkpoint(positions.conversation_id, positions.observer_progress)? != bytes
{
return Err(invalid("observer checkpoint recapture"));
}
Ok(state)
}
}
pub(super) mod epoch {
use liminal_protocol::wire::{BindingEpoch, ConnectionIncarnation, Generation};
use serde::{Deserialize, Serialize};
pub(in crate::server::participant::production::observer_progress) fn serialize<
S: serde::Serializer,
>(
epoch: &BindingEpoch,
serializer: S,
) -> Result<S::Ok, S::Error> {
(
epoch.connection_incarnation.server_incarnation,
epoch.connection_incarnation.connection_ordinal,
epoch.capability_generation.get(),
)
.serialize(serializer)
}
pub(in crate::server::participant::production::observer_progress) fn deserialize<
'de,
D: serde::Deserializer<'de>,
>(
deserializer: D,
) -> Result<BindingEpoch, D::Error> {
let (server, ordinal, generation): (u64, u64, u64) =
Deserialize::deserialize(deserializer)?;
let generation = Generation::new(generation)
.ok_or_else(|| serde::de::Error::custom("observer binding generation is zero"))?;
Ok(BindingEpoch::new(
ConnectionIncarnation::new(server, ordinal),
generation,
))
}
}