pipecircuit 0.0.19

//! Consumer-side subscription identity and atomic reconnect installation.
use super::*;
use crate::storage::warp_table::{
    IndexKey,
    observed::{ViewApplyOutcome, ViewConnection},
};
use crate::{StateMachine, StepInputOf};
#[cfg(kani)]
mod proofs;

#[cfg(test)]
mod tests;

/// Local carrier generation, scoped to the SubscribedView that issued it.
/// The transport keeps this token with that connection's events. Do not stamp
/// late events with the token of a newer connection. This is not a wire revision
/// or authorization credential, and is not interchangeable between consumers.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct SubscriptionSession(u64);

/// Keep the session with this carrier; send the request through its pipe.
#[derive(Debug)]
pub struct SubscriptionStart<Query, Command, View> {
    pub session: SubscriptionSession,
    pub request: ContinuousRequest<Query, Command, View>,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum Phase {
    Disconnected,
    Subscribing(RequestId),
    Baseline(SubscriptionId),
    Live(SubscriptionId),
    Refresh(SubscriptionId),
    Resynchronizing {
        id: RequestId,
        subscription: SubscriptionId,
        baseline: Baseline,
    },
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum Baseline {
    Replace,
    Refresh,
}

pub struct SubscribedView<Id: IndexKey, Value: Clone + Eq> {
    copy: ContinuousView<Id, Value>,
    generation: u64,
    phase: Phase,
}

#[derive(Debug)]
pub enum SubscriptionReceiveError {
    WrongSession,
    Unexpected,
    Apply(ContinuousViewError),
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum SubscriptionProgress {
    Subscribed,
    Updated(ViewApplyOutcome),
    Resynchronizing,
    Refused(SubscriptionRefusal),
}

pub struct ReceiveSubscription<'a, Value, Outcome, Refusal, Id, Projection> {
    pub session: SubscriptionSession,
    pub reply: &'a ContinuousReply<Value, Outcome, Refusal, Id, Projection>,
}

impl<Id: IndexKey, Value: Clone + Eq> StateMachine for SubscribedView<Id, Value> {}
impl<Id: IndexKey, Value: Clone + Eq> SubscribedView<Id, Value> {
    pub fn new(mut copy: ContinuousView<Id, Value>) -> Self {
        copy.set_connection(ViewConnection::Disconnected);
        Self {
            copy,
            generation: 0,
            phase: Phase::Disconnected,
        }
    }
    pub const fn view(&self) -> &ContinuousView<Id, Value> {
        &self.copy
    }
    /// Only one carrier owns this subscription at a time. Rejection preserves
    /// the requested definition and the previous readable copy.
    pub fn subscribe<Query, Command, View>(
        &mut self,
        id: RequestId,
        view: View,
        guarantee: Guarantee,
    ) -> Result<SubscriptionStart<Query, Command, View>, (View, Error)> {
        if self.phase != Phase::Disconnected {
            return Err((view, Error::Busy));
        }
        let Some(generation) = self.generation.checked_add(1) else {
            return Err((view, Error::Exhausted));
        };
        self.generation = generation;
        self.phase = Phase::Subscribing(id);
        Ok(SubscriptionStart {
            session: SubscriptionSession(generation),
            request: ContinuousRequest::Subscribe {
                id,
                view,
                guarantee,
            },
        })
    }
    pub fn disconnect(
        &mut self,
        session: SubscriptionSession,
    ) -> Result<(), SubscriptionReceiveError> {
        if session.0 != self.generation {
            return Err(SubscriptionReceiveError::WrongSession);
        }
        self.phase = Phase::Disconnected;
        self.copy.set_connection(ViewConnection::Disconnected);
        Ok(())
    }
    /// Explicit gap recovery on the same carrier. Publications already ahead
    /// of the acknowledgement are refused unchanged; the acknowledgement gates
    /// the new baseline. Physical delivery must preserve per-pipe ordering.
    pub fn resynchronize<Query, Command, View>(
        &mut self,
        id: RequestId,
    ) -> Result<ContinuousRequest<Query, Command, View>, Error> {
        let (subscription, baseline) = match self.phase {
            Phase::Baseline(subscription) => (subscription, Baseline::Replace),
            Phase::Live(subscription) | Phase::Refresh(subscription) => {
                (subscription, Baseline::Refresh)
            }
            Phase::Disconnected | Phase::Subscribing(_) | Phase::Resynchronizing { .. } => {
                return Err(Error::Busy);
            }
        };
        self.phase = Phase::Resynchronizing {
            id,
            subscription,
            baseline,
        };
        self.copy.set_connection(ViewConnection::Disconnected);
        Ok(ContinuousRequest::Resynchronize { id, subscription })
    }
    pub fn into_view(mut self) -> ContinuousView<Id, Value> {
        self.copy.set_connection(ViewConnection::Disconnected);
        self.copy
    }
}

impl<Value, Outcome, Refusal, Id: IndexKey, Projection: Clone + Eq>
    StepInputOf<SubscribedView<Id, Projection>>
    for ReceiveSubscription<'_, Value, Outcome, Refusal, Id, Projection>
{
    type Outcome = Result<SubscriptionProgress, SubscriptionReceiveError>;
    fn step(self, subscriber: &mut SubscribedView<Id, Projection>) -> Self::Outcome {
        if self.session.0 != subscriber.generation {
            return Err(SubscriptionReceiveError::WrongSession);
        }
        match self.reply {
            ContinuousReply::Subscribed { id, subscription }
                if subscriber.phase == Phase::Subscribing(*id) =>
            {
                subscriber.phase = Phase::Baseline(*subscription);
                Ok(SubscriptionProgress::Subscribed)
            }
            ContinuousReply::Resynchronizing { id, subscription } => match subscriber.phase {
                Phase::Resynchronizing {
                    id: expected,
                    subscription: current,
                    baseline,
                } if expected == *id && current == *subscription => {
                    subscriber.phase = match baseline {
                        Baseline::Replace => Phase::Baseline(*subscription),
                        Baseline::Refresh => Phase::Refresh(*subscription),
                    };
                    Ok(SubscriptionProgress::Resynchronizing)
                }
                Phase::Disconnected
                | Phase::Subscribing(_)
                | Phase::Baseline(_)
                | Phase::Live(_)
                | Phase::Refresh(_)
                | Phase::Resynchronizing { .. } => Err(SubscriptionReceiveError::Unexpected),
            },
            ContinuousReply::SubscriptionRefused { id, reason } if matches!(subscriber.phase, Phase::Subscribing(request) | Phase::Resynchronizing { id: request, .. } if request == *id) =>
            {
                subscriber.phase = Phase::Disconnected;
                subscriber.copy.set_connection(ViewConnection::Disconnected);
                Ok(SubscriptionProgress::Refused(*reason))
            }
            ContinuousReply::View {
                subscription,
                update,
            } if matches!(subscriber.phase, Phase::Baseline(current) | Phase::Live(current) | Phase::Refresh(current) if current == *subscription) =>
            {
                let baseline = matches!(subscriber.phase, Phase::Baseline(_));
                if matches!(subscriber.phase, Phase::Refresh(_))
                    && !matches!(update, ViewUpdate::Baseline { .. })
                {
                    return Ok(SubscriptionProgress::Updated(
                        ViewApplyOutcome::BaselineRequired,
                    ));
                }
                let outcome = if baseline {
                    subscriber.copy.reconnect(update)
                } else {
                    subscriber.copy.receive(update)
                }
                .map_err(SubscriptionReceiveError::Apply)?;
                match outcome {
                    ViewApplyOutcome::Applied => {
                        subscriber.phase = Phase::Live(*subscription);
                        subscriber.copy.set_connection(ViewConnection::Connected);
                    }
                    ViewApplyOutcome::Stale => {}
                    ViewApplyOutcome::BaselineRequired => {
                        subscriber.copy.set_connection(ViewConnection::Disconnected);
                        // Keep Live for an existing subscription: only an
                        // explicit resynchronization may replace its identity.
                    }
                }
                Ok(SubscriptionProgress::Updated(outcome))
            }
            ContinuousReply::Command { .. }
            | ContinuousReply::ReplicationInterrupted { .. }
            | ContinuousReply::Subscribed { .. }
            | ContinuousReply::Unsubscribed { .. }
            | ContinuousReply::SubscriptionRefused { .. }
            | ContinuousReply::View { .. } => Err(SubscriptionReceiveError::Unexpected),
        }
    }
}