hydracache/grid/
convergence_staleness.rs

1use serde::{Deserialize, Serialize};
2
3use crate::grid::hardening::{MergePolicy, ReplicatedValueRecord};
4use crate::grid::session_context::{PartitionKey, SessionWatermark, VersionStamp};
5
6/// Explicit staleness budget for a session read.
7#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
8pub struct StalenessBound {
9    /// Maximum allowed lag in monotonic value versions.
10    pub max_version_lag: u64,
11}
12
13impl StalenessBound {
14    /// Create a version-lag bound.
15    pub const fn versions(max_version_lag: u64) -> Self {
16        Self { max_version_lag }
17    }
18}
19
20/// Session read freshness mode.
21#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
22#[serde(rename_all = "snake_case")]
23pub enum SessionReadMode {
24    /// Require the full session watermark before serving locally.
25    #[default]
26    Causal,
27    /// Allow bounded lag, but never below the causal dependency floor.
28    BoundedStaleness {
29        /// Explicit max staleness.
30        max: StalenessBound,
31    },
32}
33
34/// Why a bounded-staleness read had to escalate.
35#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
36#[serde(rename_all = "snake_case")]
37pub enum StalenessEscalationReason {
38    /// Candidate is below a W4 causal dependency floor.
39    BelowCausalFloor,
40    /// Strict causal mode requires the full session watermark.
41    BelowSessionWatermark,
42    /// Candidate is above the causal floor but outside the chosen staleness bound.
43    BeyondBound,
44}
45
46/// Bounded-staleness read decision.
47#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
48#[serde(rename_all = "snake_case")]
49pub enum StalenessDecision {
50    /// Strict causal mode can serve locally.
51    ServeCausal {
52        /// Observed version lag from the session watermark.
53        observed_version_lag: u64,
54    },
55    /// Bounded mode can serve locally within the explicit budget.
56    ServeFast {
57        /// Observed version lag from the session watermark.
58        observed_version_lag: u64,
59    },
60    /// The caller must escalate through W2.
61    Escalate {
62        /// Escalation reason.
63        reason: StalenessEscalationReason,
64        /// Observed version lag from the session watermark.
65        observed_version_lag: u64,
66    },
67}
68
69/// Bounded staleness metrics.
70#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
71pub struct BoundedStalenessMetrics {
72    /// Reads served locally by bounded staleness.
73    pub bounded_staleness_fast_serves_total: u64,
74    /// Bounded-staleness reads that escalated.
75    pub bounded_staleness_escalations_total: u64,
76}
77
78/// Reduce concurrent replicated records to one converged value through the merge policy.
79pub fn converge_replicated_values<P, I>(policy: &P, records: I) -> Option<ReplicatedValueRecord>
80where
81    P: MergePolicy + ?Sized,
82    I: IntoIterator<Item = ReplicatedValueRecord>,
83{
84    let mut winner = None;
85    for record in records {
86        winner = policy.merge(winner.as_ref(), &record);
87    }
88    winner
89}
90
91/// Resolve a read under either strict causal or bounded-staleness mode.
92pub fn resolve_session_read_mode(
93    watermark: &SessionWatermark,
94    key: &PartitionKey,
95    causal_floor: Option<VersionStamp>,
96    candidate: VersionStamp,
97    mode: SessionReadMode,
98) -> StalenessDecision {
99    let observed_version_lag = watermark
100        .highest_seen(key)
101        .map(|seen| seen.version_distance_from(candidate))
102        .unwrap_or_default();
103
104    if causal_floor.is_some_and(|floor| candidate < floor) {
105        return StalenessDecision::Escalate {
106            reason: StalenessEscalationReason::BelowCausalFloor,
107            observed_version_lag,
108        };
109    }
110
111    match mode {
112        SessionReadMode::Causal => {
113            if watermark
114                .highest_seen(key)
115                .is_some_and(|required| candidate < required)
116            {
117                StalenessDecision::Escalate {
118                    reason: StalenessEscalationReason::BelowSessionWatermark,
119                    observed_version_lag,
120                }
121            } else {
122                StalenessDecision::ServeCausal {
123                    observed_version_lag,
124                }
125            }
126        }
127        SessionReadMode::BoundedStaleness { max }
128            if observed_version_lag <= max.max_version_lag =>
129        {
130            StalenessDecision::ServeFast {
131                observed_version_lag,
132            }
133        }
134        SessionReadMode::BoundedStaleness { .. } => StalenessDecision::Escalate {
135            reason: StalenessEscalationReason::BeyondBound,
136            observed_version_lag,
137        },
138    }
139}
140
141/// Return whether the read can be served locally under the chosen mode.
142pub fn within_staleness_bound(
143    watermark: &SessionWatermark,
144    key: &PartitionKey,
145    causal_floor: Option<VersionStamp>,
146    candidate: VersionStamp,
147    mode: SessionReadMode,
148) -> bool {
149    matches!(
150        resolve_session_read_mode(watermark, key, causal_floor, candidate, mode),
151        StalenessDecision::ServeCausal { .. } | StalenessDecision::ServeFast { .. }
152    )
153}
hydracache/grid/convergence_staleness.rs

hydracache/grid/
convergence_staleness.rs
