|
| 1 | +//! Cluster generation/epoch — a monotonic, leader-bumped fence token |
| 2 | +//! stamped on every Raft RPC frame. |
| 3 | +//! |
| 4 | +//! # Purpose |
| 5 | +//! |
| 6 | +//! `cluster_epoch` is a process-wide `u64` that advances on every |
| 7 | +//! cluster-membership-shifting event the metadata-group leader observes |
| 8 | +//! (currently: becoming leader of the metadata group). Stamping it on every |
| 9 | +//! outbound RPC and observing it on every inbound RPC lets every peer keep |
| 10 | +//! a cluster-wide high-water mark and reject (or quarantine) frames from |
| 11 | +//! peers stuck on a strictly older epoch — i.e. peers that missed a topology |
| 12 | +//! transition and may be acting on stale state. |
| 13 | +//! |
| 14 | +//! # Mechanics |
| 15 | +//! |
| 16 | +//! * A process-global `AtomicU64`, [`LOCAL_CLUSTER_EPOCH`], holds the local |
| 17 | +//! high-water mark. On startup it is loaded from the cluster catalog |
| 18 | +//! (see [`ClusterCatalog::load_cluster_epoch`]). |
| 19 | +//! * [`current_local_cluster_epoch`] reads it for the encoder. |
| 20 | +//! * [`observe_peer_cluster_epoch`] is called by the decoder for every |
| 21 | +//! inbound frame; it bumps the local mark via `fetch_max` (monotonic). |
| 22 | +//! * [`bump_local_cluster_epoch`] is called by the metadata-group leader |
| 23 | +//! when leadership transitions to it; it advances the local mark and |
| 24 | +//! persists the new value. |
| 25 | +//! |
| 26 | +//! Persistence is best-effort: a bump is committed in-memory atomically; |
| 27 | +//! if the catalog write fails, the in-memory value is still advanced and |
| 28 | +//! the failure is logged. (After a crash, the persisted value is a lower |
| 29 | +//! bound — the new leader will re-bump beyond it.) |
| 30 | +//! |
| 31 | +//! # Why a global atomic |
| 32 | +//! |
| 33 | +//! Every encode/decode call site needs the current epoch. Threading it |
| 34 | +//! through the existing 19 encode/decode functions and their callers |
| 35 | +//! would touch hundreds of sites for what is, semantically, a single |
| 36 | +//! per-process value. An `AtomicU64` is the right shape for this kind |
| 37 | +//! of read-mostly, monotonic counter. |
| 38 | +
|
| 39 | +use std::sync::Arc; |
| 40 | +use std::sync::atomic::{AtomicU64, Ordering}; |
| 41 | + |
| 42 | +use crate::catalog::ClusterCatalog; |
| 43 | +use crate::error::Result; |
| 44 | + |
| 45 | +/// Process-global cluster epoch high-water mark. |
| 46 | +/// |
| 47 | +/// Initialized to 0 (genesis). Loaded from catalog at startup via |
| 48 | +/// [`init_local_cluster_epoch_from_catalog`]. |
| 49 | +static LOCAL_CLUSTER_EPOCH: AtomicU64 = AtomicU64::new(0); |
| 50 | + |
| 51 | +/// Read the current local cluster epoch (the value an outbound RPC |
| 52 | +/// should stamp). Cheap; lock-free. |
| 53 | +pub fn current_local_cluster_epoch() -> u64 { |
| 54 | + LOCAL_CLUSTER_EPOCH.load(Ordering::Acquire) |
| 55 | +} |
| 56 | + |
| 57 | +/// Set the local epoch directly. Only intended for use by |
| 58 | +/// [`init_local_cluster_epoch_from_catalog`] at startup and by tests. |
| 59 | +/// Production code paths use [`observe_peer_cluster_epoch`] (monotonic |
| 60 | +/// max) or [`bump_local_cluster_epoch`] (leader-side increment). |
| 61 | +pub fn set_local_cluster_epoch(value: u64) { |
| 62 | + LOCAL_CLUSTER_EPOCH.store(value, Ordering::Release); |
| 63 | +} |
| 64 | + |
| 65 | +/// Observe an epoch carried by an inbound RPC. Advances the local mark |
| 66 | +/// via `fetch_max` (so concurrent observations are safe and monotonic). |
| 67 | +/// |
| 68 | +/// Returns the new local high-water mark. |
| 69 | +pub fn observe_peer_cluster_epoch(peer_epoch: u64) -> u64 { |
| 70 | + let prev = LOCAL_CLUSTER_EPOCH.fetch_max(peer_epoch, Ordering::AcqRel); |
| 71 | + prev.max(peer_epoch) |
| 72 | +} |
| 73 | + |
| 74 | +/// Increment the local epoch by 1 and persist the new value to the |
| 75 | +/// cluster catalog. Called by the metadata-group leader on a leadership |
| 76 | +/// transition. |
| 77 | +/// |
| 78 | +/// Returns the new epoch. The persistence failure path advances the |
| 79 | +/// in-memory value anyway (so RPCs immediately reflect the bump) and |
| 80 | +/// returns the persistence error to the caller. |
| 81 | +pub fn bump_local_cluster_epoch(catalog: &ClusterCatalog) -> Result<u64> { |
| 82 | + let new_epoch = LOCAL_CLUSTER_EPOCH.fetch_add(1, Ordering::AcqRel) + 1; |
| 83 | + catalog.save_cluster_epoch(new_epoch)?; |
| 84 | + Ok(new_epoch) |
| 85 | +} |
| 86 | + |
| 87 | +/// Initialize the local epoch from the cluster catalog at process |
| 88 | +/// startup. Idempotent — safe to call multiple times during boot. |
| 89 | +pub fn init_local_cluster_epoch_from_catalog(catalog: &Arc<ClusterCatalog>) -> Result<u64> { |
| 90 | + let persisted = catalog.load_cluster_epoch()?.unwrap_or(0); |
| 91 | + // fetch_max so we don't regress past anything already observed |
| 92 | + // earlier in startup (e.g. an inbound frame on the join path). |
| 93 | + let prev = LOCAL_CLUSTER_EPOCH.fetch_max(persisted, Ordering::AcqRel); |
| 94 | + Ok(prev.max(persisted)) |
| 95 | +} |
| 96 | + |
| 97 | +#[cfg(test)] |
| 98 | +mod tests { |
| 99 | + use super::*; |
| 100 | + use std::sync::Mutex; |
| 101 | + |
| 102 | + /// All tests in this module mutate the same global atomic, so they |
| 103 | + /// must be serialised. The lock is taken at the top of each test. |
| 104 | + static TEST_LOCK: Mutex<()> = Mutex::new(()); |
| 105 | + |
| 106 | + fn reset() -> std::sync::MutexGuard<'static, ()> { |
| 107 | + let g = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner()); |
| 108 | + LOCAL_CLUSTER_EPOCH.store(0, Ordering::Release); |
| 109 | + g |
| 110 | + } |
| 111 | + |
| 112 | + #[test] |
| 113 | + fn observe_is_monotonic_max() { |
| 114 | + let _g = reset(); |
| 115 | + assert_eq!(observe_peer_cluster_epoch(5), 5); |
| 116 | + assert_eq!(current_local_cluster_epoch(), 5); |
| 117 | + // Older peer epoch must not regress the local mark. |
| 118 | + assert_eq!(observe_peer_cluster_epoch(3), 5); |
| 119 | + assert_eq!(current_local_cluster_epoch(), 5); |
| 120 | + // Newer advances. |
| 121 | + assert_eq!(observe_peer_cluster_epoch(7), 7); |
| 122 | + assert_eq!(current_local_cluster_epoch(), 7); |
| 123 | + } |
| 124 | + |
| 125 | + #[test] |
| 126 | + fn set_overrides_for_init() { |
| 127 | + let _g = reset(); |
| 128 | + set_local_cluster_epoch(42); |
| 129 | + assert_eq!(current_local_cluster_epoch(), 42); |
| 130 | + } |
| 131 | + |
| 132 | + #[test] |
| 133 | + fn observe_zero_is_noop() { |
| 134 | + let _g = reset(); |
| 135 | + set_local_cluster_epoch(9); |
| 136 | + assert_eq!(observe_peer_cluster_epoch(0), 9); |
| 137 | + assert_eq!(current_local_cluster_epoch(), 9); |
| 138 | + } |
| 139 | + |
| 140 | + #[test] |
| 141 | + fn bump_increments_and_persists() { |
| 142 | + let _g = reset(); |
| 143 | + let dir = tempfile::tempdir().unwrap(); |
| 144 | + let catalog = ClusterCatalog::open(&dir.path().join("cluster.redb")).unwrap(); |
| 145 | + set_local_cluster_epoch(10); |
| 146 | + let new_epoch = bump_local_cluster_epoch(&catalog).unwrap(); |
| 147 | + assert_eq!(new_epoch, 11); |
| 148 | + assert_eq!(current_local_cluster_epoch(), 11); |
| 149 | + assert_eq!(catalog.load_cluster_epoch().unwrap(), Some(11)); |
| 150 | + } |
| 151 | + |
| 152 | + #[test] |
| 153 | + fn init_from_catalog_loads_persisted_value() { |
| 154 | + let _g = reset(); |
| 155 | + let dir = tempfile::tempdir().unwrap(); |
| 156 | + let catalog = Arc::new(ClusterCatalog::open(&dir.path().join("cluster.redb")).unwrap()); |
| 157 | + catalog.save_cluster_epoch(123).unwrap(); |
| 158 | + let v = init_local_cluster_epoch_from_catalog(&catalog).unwrap(); |
| 159 | + assert_eq!(v, 123); |
| 160 | + assert_eq!(current_local_cluster_epoch(), 123); |
| 161 | + } |
| 162 | + |
| 163 | + #[test] |
| 164 | + fn init_with_no_persisted_value_starts_at_zero() { |
| 165 | + let _g = reset(); |
| 166 | + let dir = tempfile::tempdir().unwrap(); |
| 167 | + let catalog = Arc::new(ClusterCatalog::open(&dir.path().join("cluster.redb")).unwrap()); |
| 168 | + let v = init_local_cluster_epoch_from_catalog(&catalog).unwrap(); |
| 169 | + assert_eq!(v, 0); |
| 170 | + assert_eq!(current_local_cluster_epoch(), 0); |
| 171 | + } |
| 172 | +} |
0 commit comments