alopex_core/kv/
read_at.rs

1//! Fenced snapshot-read contract for distributed reads.
2//!
3//! This module intentionally distinguishes a globally issued data epoch from
4//! the local timestamp used by an ordinary storage transaction.  A backend is
5//! never eligible for strong or stale remote reads merely because it can open
6//! a local read-only transaction.
7
8use serde::{Deserialize, Serialize};
9
10/// A cluster-authorized, fenced snapshot cut requested from a storage backend.
11#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
12pub struct ReadAtPoint {
13    /// Monotonic user-data epoch issued by the cluster authority.
14    pub data_epoch: u64,
15    /// Committed metadata version that must be visible at this read.
16    pub metadata_version: u64,
17    /// Schema manifest identity required by the query plan.
18    pub schema_epoch: u64,
19    /// Index definition identity required by the query plan.
20    pub index_epoch: u64,
21}
22
23impl ReadAtPoint {
24    /// Creates a fenced read point from cluster-issued identities.
25    pub const fn new(
26        data_epoch: u64,
27        metadata_version: u64,
28        schema_epoch: u64,
29        index_epoch: u64,
30    ) -> Self {
31        Self {
32            data_epoch,
33            metadata_version,
34            schema_epoch,
35            index_epoch,
36        }
37    }
38}
39
40/// Backend evidence required before remote strong/stale reads may be enabled.
41#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
42pub enum ReadAtCapability {
43    /// The backend retains every epoch in this inclusive readable interval and
44    /// can bind a read-only session to the complete supplied fence.
45    Available {
46        /// Oldest retained data epoch.
47        readable_from_epoch: u64,
48        /// Newest data epoch proven readable.
49        readable_through_epoch: u64,
50    },
51    /// The backend must not be selected for remote strong/stale reads.
52    Unavailable {
53        /// Stable operator-visible reason.
54        reason: String,
55    },
56}
57
58impl ReadAtCapability {
59    /// Creates an explicit unavailable capability result.
60    pub fn unavailable(reason: impl Into<String>) -> Self {
61        Self::Unavailable {
62            reason: reason.into(),
63        }
64    }
65
66    /// Checks whether `point` is inside the proven retained interval.
67    pub fn validate(&self, point: &ReadAtPoint) -> ReadAtResult<()> {
68        match self {
69            Self::Available {
70                readable_from_epoch,
71                readable_through_epoch,
72            } if point.data_epoch < *readable_from_epoch => Err(ReadAtError::Expired {
73                requested_epoch: point.data_epoch,
74                readable_from_epoch: *readable_from_epoch,
75            }),
76            Self::Available {
77                readable_through_epoch,
78                ..
79            } if point.data_epoch > *readable_through_epoch => Err(ReadAtError::NotYetReadable {
80                requested_epoch: point.data_epoch,
81                readable_through_epoch: *readable_through_epoch,
82            }),
83            Self::Available { .. } => Ok(()),
84            Self::Unavailable { reason } => Err(ReadAtError::Unavailable {
85                requested_epoch: point.data_epoch,
86                reason: reason.clone(),
87            }),
88        }
89    }
90
91    /// Produces an unavailable result while retaining the requested epoch.
92    pub fn unavailable_error(&self, point: &ReadAtPoint, fallback_reason: &str) -> ReadAtError {
93        match self {
94            Self::Unavailable { reason } => ReadAtError::Unavailable {
95                requested_epoch: point.data_epoch,
96                reason: reason.clone(),
97            },
98            Self::Available { .. } => ReadAtError::Unavailable {
99                requested_epoch: point.data_epoch,
100                reason: fallback_reason.to_string(),
101            },
102        }
103    }
104}
105
106/// Classified failure to open a fenced storage snapshot.
107#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
108pub enum ReadAtError {
109    /// The requested epoch was compacted before a session could open.
110    #[error(
111        "read point expired: requested={requested_epoch}, readable_from={readable_from_epoch}"
112    )]
113    Expired {
114        /// Requested cluster data epoch.
115        requested_epoch: u64,
116        /// Oldest retained data epoch.
117        readable_from_epoch: u64,
118    },
119    /// The requested epoch has not been applied by the backend.
120    #[error("read point unavailable: requested={requested_epoch}, readable_through={readable_through_epoch}")]
121    NotYetReadable {
122        /// Requested cluster data epoch.
123        requested_epoch: u64,
124        /// Newest locally readable data epoch.
125        readable_through_epoch: u64,
126    },
127    /// The backend has no admissible read-at implementation.
128    #[error("read point unavailable at epoch {requested_epoch}: {reason}")]
129    Unavailable {
130        /// Requested cluster data epoch.
131        requested_epoch: u64,
132        /// Stable reason for ineligibility.
133        reason: String,
134    },
135}
136
137/// Result returned by fenced snapshot-read capability operations.
138pub type ReadAtResult<T> = std::result::Result<T, ReadAtError>;
139
140#[cfg(test)]
141mod tests {
142    use super::*;
143    use crate::kv::{AnyKV, KVStore};
144
145    fn point(epoch: u64) -> ReadAtPoint {
146        ReadAtPoint::new(epoch, 4, 5, 6)
147    }
148
149    #[test]
150    fn retained_interval_classifies_expired_and_unapplied_epochs() {
151        let capability = ReadAtCapability::Available {
152            readable_from_epoch: 10,
153            readable_through_epoch: 20,
154        };
155        assert!(matches!(
156            capability.validate(&point(9)),
157            Err(ReadAtError::Expired { .. })
158        ));
159        assert!(capability.validate(&point(10)).is_ok());
160        assert!(capability.validate(&point(20)).is_ok());
161        assert!(matches!(
162            capability.validate(&point(21)),
163            Err(ReadAtError::NotYetReadable { .. })
164        ));
165    }
166
167    #[test]
168    fn any_kv_does_not_treat_a_local_snapshot_as_a_cluster_read_point() {
169        let store = AnyKV::Memory(crate::kv::memory::MemoryKV::new());
170        assert!(matches!(
171            store.read_at_capability(),
172            ReadAtCapability::Unavailable { .. }
173        ));
174        assert!(matches!(
175            store.begin_read_at(&point(1)),
176            Err(ReadAtError::Unavailable { .. })
177        ));
178    }
179}
alopex_core/kv/read_at.rs

alopex_core/kv/
read_at.rs
