batpak 0.9.0

//! Deterministic Batpak Substrate Closure subscriber frontier observations.
//!
//! This module reports structural delivery/frontier observations for lossy
//! subscribers and cursor-backed pull paths without imposing policy semantics.

use crate::store::Store;
use serde::{Deserialize, Serialize};

/// Report-body schema version for subscriber frontier evidence.
pub const SUBSCRIBER_FRONTIER_REPORT_SCHEMA_VERSION: u16 = 1;

/// Fixed-width hash used by subscriber frontier reports.
pub type SubscriberFrontierHash = [u8; 32];

/// Subscriber source lane being observed.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub enum SubscriberFrontierSource {
    /// Push-based lossy subscription.
    LossyPush,
    /// Cursor-backed pull path.
    CursorBacked,
}

/// Coarse delivery state for subscriber frontier observations.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub enum SubscriberDeliveryState {
    /// Subscriber is actively consuming.
    Active,
    /// Subscriber is behind available frontier.
    Lagging,
    /// Subscriber was dropped from a lossy path.
    Dropped,
    /// Delivery path disconnected.
    Disconnected,
    /// Caller could not determine the delivery state.
    Unknown,
}

/// Precision class for observed delivery loss.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub enum LossPrecision {
    /// Exact dropped ranges are available.
    ExactRange,
    /// Loss can be localized only after a consumed frontier.
    LossAfterFrontier,
    /// Subscriber was dropped from the lossy fanout path.
    SubscriberDropped,
    /// Caller could not determine loss precision.
    Unknown,
}

/// Request input for subscriber frontier observation.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct SubscriberFrontierRequest {
    /// Observed source lane.
    pub source: SubscriberFrontierSource,
    /// Last consumed global sequence if known.
    pub consumed_frontier_sequence: Option<u64>,
    /// Delivery state observed by caller.
    pub delivery_state: SubscriberDeliveryState,
    /// Loss precision observed by caller.
    pub loss_precision: LossPrecision,
    /// Exact dropped ranges when `loss_precision == ExactRange`.
    pub exact_dropped_ranges: Vec<(u64, u64)>,
}

impl SubscriberFrontierRequest {
    /// Build a lossy push observation request.
    #[must_use]
    pub fn lossy_push(
        consumed_frontier_sequence: Option<u64>,
        delivery_state: SubscriberDeliveryState,
        loss_precision: LossPrecision,
    ) -> Self {
        Self {
            source: SubscriberFrontierSource::LossyPush,
            consumed_frontier_sequence,
            delivery_state,
            loss_precision,
            exact_dropped_ranges: Vec::new(),
        }
    }

    /// Build a cursor-backed observation request.
    #[must_use]
    pub fn cursor_backed(
        consumed_frontier_sequence: Option<u64>,
        delivery_state: SubscriberDeliveryState,
        loss_precision: LossPrecision,
    ) -> Self {
        Self {
            source: SubscriberFrontierSource::CursorBacked,
            consumed_frontier_sequence,
            delivery_state,
            loss_precision,
            exact_dropped_ranges: Vec::new(),
        }
    }
}

/// Deterministic structural findings from subscriber frontier observation.
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub enum SubscriberFrontierFinding {
    /// Consumed frontier was not provided.
    ConsumedFrontierUnknown,
    /// Delivery state was unknown.
    DeliveryStateUnknown,
    /// Subscriber lag is observed in event-sequence units.
    LagObserved {
        /// Number of events between available and consumed frontiers.
        lag_events: u64,
    },
    /// Caller-reported consumed frontier is ahead of the available frontier.
    ConsumedFrontierAheadOfAvailable {
        /// Caller-reported consumed sequence.
        consumed_sequence: u64,
        /// Available sequence for the observed source lane.
        available_sequence: u64,
    },
    /// Delivery path reported dropped subscriber state.
    DeliveryDropped,
    /// Delivery path reported disconnection.
    DeliveryDisconnected,
    /// Loss is observed with the given precision class.
    LossObserved {
        /// Precision class for loss observations.
        precision: LossPrecision,
    },
    /// Exact dropped range when precision is available.
    ExactDroppedRange {
        /// Inclusive range start.
        start_sequence: u64,
        /// Exclusive range end.
        end_sequence: u64,
    },
}

/// Deterministic report body for subscriber frontier evidence.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct SubscriberFrontierReportBody {
    /// Report-body schema version.
    pub schema_version: u16,
    /// Observed source lane.
    pub source: SubscriberFrontierSource,
    /// Last consumed frontier sequence.
    pub consumed_frontier_sequence: Option<u64>,
    /// Available frontier sequence relevant to the source lane.
    pub available_frontier_sequence: u64,
    /// Lag in events if consumed frontier is known.
    pub lag_events: Option<u64>,
    /// Coarse delivery state.
    pub delivery_state: SubscriberDeliveryState,
    /// Loss precision class.
    pub loss_precision: LossPrecision,
    /// Deterministic structural findings.
    pub findings: Vec<SubscriberFrontierFinding>,
}

/// Subscriber frontier evidence report.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct SubscriberFrontierEvidenceReport {
    /// Deterministic report body.
    pub body: SubscriberFrontierReportBody,
    /// Canonical hash of `body`.
    pub body_hash: SubscriberFrontierHash,
    /// Optional generation timestamp metadata outside deterministic identity.
    pub generated_at_unix_ms: Option<u64>,
    /// Optional producer version metadata outside deterministic identity.
    pub batpak_version: Option<String>,
    /// Optional diagnostics outside deterministic identity.
    pub diagnostics: Vec<String>,
}

/// Error returned when subscriber frontier report generation fails.
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum SubscriberFrontierReportError {
    /// Canonical report-body encoding failed.
    BodyEncoding {
        /// Human-readable encoding failure.
        message: String,
    },
}

impl std::fmt::Display for SubscriberFrontierReportError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::BodyEncoding { message } => {
                write!(
                    f,
                    "subscriber frontier report body encoding failed: {message}"
                )
            }
        }
    }
}

impl std::error::Error for SubscriberFrontierReportError {}

impl<State: crate::store::StoreState> Store<State> {
    /// Build a deterministic subscriber frontier evidence report.
    ///
    /// # Errors
    /// Returns [`SubscriberFrontierReportError::BodyEncoding`] when canonical
    /// encoding of the deterministic report body fails.
    pub fn subscriber_frontier_observation(
        &self,
        request: &SubscriberFrontierRequest,
    ) -> Result<SubscriberFrontierEvidenceReport, SubscriberFrontierReportError> {
        let frontier = self.frontier();
        let available_frontier_sequence = match request.source {
            SubscriberFrontierSource::LossyPush => frontier.emitted_hlc.global_sequence,
            SubscriberFrontierSource::CursorBacked => frontier.visible_hlc.global_sequence,
        };

        let lag_events = request
            .consumed_frontier_sequence
            .map(|consumed| available_frontier_sequence.saturating_sub(consumed));

        let mut findings = Vec::new();
        if request.consumed_frontier_sequence.is_none() {
            findings.push(SubscriberFrontierFinding::ConsumedFrontierUnknown);
        }
        if request.delivery_state == SubscriberDeliveryState::Unknown {
            findings.push(SubscriberFrontierFinding::DeliveryStateUnknown);
        }
        if let Some(lag) = lag_events {
            if lag > 0 {
                findings.push(SubscriberFrontierFinding::LagObserved { lag_events: lag });
            }
        }
        if let Some(consumed_sequence) = request.consumed_frontier_sequence {
            if consumed_sequence > available_frontier_sequence {
                findings.push(
                    SubscriberFrontierFinding::ConsumedFrontierAheadOfAvailable {
                        consumed_sequence,
                        available_sequence: available_frontier_sequence,
                    },
                );
            }
        }

        match request.delivery_state {
            SubscriberDeliveryState::Dropped => {
                findings.push(SubscriberFrontierFinding::DeliveryDropped)
            }
            SubscriberDeliveryState::Disconnected => {
                findings.push(SubscriberFrontierFinding::DeliveryDisconnected)
            }
            SubscriberDeliveryState::Active
            | SubscriberDeliveryState::Lagging
            | SubscriberDeliveryState::Unknown => {}
        }

        if request.loss_precision != LossPrecision::Unknown {
            findings.push(SubscriberFrontierFinding::LossObserved {
                precision: request.loss_precision,
            });
        }
        if request.loss_precision == LossPrecision::ExactRange {
            let mut ranges = request.exact_dropped_ranges.clone();
            ranges.sort_unstable();
            for (start_sequence, end_sequence) in ranges {
                findings.push(SubscriberFrontierFinding::ExactDroppedRange {
                    start_sequence,
                    end_sequence,
                });
            }
        }

        crate::evidence::sort_findings(&mut findings);

        let body = SubscriberFrontierReportBody {
            schema_version: SUBSCRIBER_FRONTIER_REPORT_SCHEMA_VERSION,
            source: request.source,
            consumed_frontier_sequence: request.consumed_frontier_sequence,
            available_frontier_sequence,
            lag_events,
            delivery_state: request.delivery_state,
            loss_precision: request.loss_precision,
            findings,
        };
        let body_hash = report_body_hash(&body)?;
        Ok(SubscriberFrontierEvidenceReport {
            body,
            body_hash,
            generated_at_unix_ms: None,
            batpak_version: None,
            diagnostics: Vec::new(),
        })
    }
}

fn report_body_hash(
    body: &SubscriberFrontierReportBody,
) -> Result<SubscriberFrontierHash, SubscriberFrontierReportError> {
    crate::evidence::report_body_hash(body, |message| {
        SubscriberFrontierReportError::BodyEncoding { message }
    })
}

#[cfg(test)]
mod lag_boundary_tests {
    use super::{
        LossPrecision, SubscriberDeliveryState, SubscriberFrontierFinding,
        SubscriberFrontierRequest,
    };
    use crate::coordinate::Coordinate;
    use crate::event::EventKind;
    use crate::store::{Store, StoreConfig};

    fn open_store() -> (tempfile::TempDir, Store) {
        let dir = tempfile::TempDir::new().expect("temp dir");
        let store = Store::open(StoreConfig::new(dir.path())).expect("open store");
        (dir, store)
    }

    fn cursor_backed(consumed: Option<u64>) -> SubscriberFrontierRequest {
        SubscriberFrontierRequest::cursor_backed(
            consumed,
            SubscriberDeliveryState::Active,
            LossPrecision::Unknown,
        )
    }

    #[test]
    fn positive_lag_emits_lag_observed_with_the_exact_event_delta() {
        // Kills subscriber_frontier.rs:228 `lag > 0` -> `<` and -> `==`. When the
        // consumed frontier trails the available frontier, a LagObserved finding
        // with the EXACT event delta must be emitted. `<` (never true for u64)
        // suppresses it entirely; `==` only fires it on a zero delta, so a real
        // positive lag would be dropped.
        let (_dir, store) = open_store();
        let coord = Coordinate::new("entity:lag-positive", "scope:test").expect("coord");
        let _receipt = store
            .append(
                &coord,
                EventKind::custom(0xF, 0x40),
                &serde_json::json!({ "n": 0 }),
            )
            .expect("append advances the visible frontier");

        let report = store
            .subscriber_frontier_observation(&cursor_backed(Some(0)))
            .expect("observation");
        let available = report.body.available_frontier_sequence;
        assert!(
            available >= 1,
            "sanity: an acked append advanced the available (visible) frontier, got {available}"
        );
        assert_eq!(
            report.body.lag_events,
            Some(available),
            "consumed 0 against available {available} is a lag of {available} events"
        );
        assert!(
            report.body.findings.iter().any(|finding| matches!(
                finding,
                SubscriberFrontierFinding::LagObserved { lag_events } if *lag_events == available
            )),
            "PROPERTY: a positive lag emits LagObserved{{{available}}}; the `<`/`==` \
             mutants of `lag > 0` drop it, got {:?}",
            report.body.findings
        );
        assert!(
            !report.body.findings.iter().any(|f| matches!(
                f,
                SubscriberFrontierFinding::ConsumedFrontierAheadOfAvailable { .. }
            )),
            "a trailing consumed frontier is not ahead of available"
        );
    }

    #[test]
    fn zero_lag_emits_neither_lag_observed_nor_ahead_of_available() {
        // Kills subscriber_frontier.rs:228 `lag > 0` -> `>=`/`==` AND
        // subscriber_frontier.rs:233 `consumed_sequence > available` -> `>=`. When
        // the consumed frontier EQUALS the available frontier, the lag is zero:
        // neither a LagObserved nor a ConsumedFrontierAheadOfAvailable finding may
        // appear. `>=` on either comparison fires a spurious finding at equality.
        let (_dir, store) = open_store();
        let available = store
            .subscriber_frontier_observation(&cursor_backed(None))
            .expect("probe observation")
            .body
            .available_frontier_sequence;

        let report = store
            .subscriber_frontier_observation(&cursor_backed(Some(available)))
            .expect("observation");
        assert_eq!(
            report.body.lag_events,
            Some(0),
            "consuming exactly up to available is a zero lag"
        );
        assert!(
            !report
                .body
                .findings
                .iter()
                .any(|f| matches!(f, SubscriberFrontierFinding::LagObserved { .. })),
            "PROPERTY: a zero lag emits NO LagObserved; the `>=`/`==` mutants of \
             `lag > 0` fabricate one, got {:?}",
            report.body.findings
        );
        assert!(
            !report.body.findings.iter().any(|f| matches!(
                f,
                SubscriberFrontierFinding::ConsumedFrontierAheadOfAvailable { .. }
            )),
            "PROPERTY: consumed == available is NOT ahead of available; the `>=` mutant \
             of `consumed > available` fabricates an ahead-of-available finding, got {:?}",
            report.body.findings
        );
    }
}

#[cfg(test)]
mod tests {
    use super::{LossPrecision, SubscriberFrontierFinding};

    #[test]
    fn subscriber_frontier_findings_are_sorted_structurally() {
        let mut findings = vec![
            SubscriberFrontierFinding::ExactDroppedRange {
                start_sequence: 20,
                end_sequence: 30,
            },
            SubscriberFrontierFinding::ExactDroppedRange {
                start_sequence: 10,
                end_sequence: 15,
            },
            SubscriberFrontierFinding::LossObserved {
                precision: LossPrecision::Unknown,
            },
        ];
        crate::evidence::sort_findings(&mut findings);

        assert_eq!(
            findings,
            vec![
                SubscriberFrontierFinding::LossObserved {
                    precision: LossPrecision::Unknown,
                },
                SubscriberFrontierFinding::ExactDroppedRange {
                    start_sequence: 10,
                    end_sequence: 15,
                },
                SubscriberFrontierFinding::ExactDroppedRange {
                    start_sequence: 20,
                    end_sequence: 30,
                },
            ],
            "PROPERTY: subscriber frontier findings must be sorted in deterministic structural order"
        );
    }
}