hydracache/grid/
active_active.rs

1use std::collections::BTreeMap;
2use std::fmt;
3
4use serde::{Deserialize, Serialize};
5
6use crate::cluster::{partition_for_key, ClusterEpoch, ClusterNodeId, PartitionId};
7use crate::grid::elasticity::RegionId;
8use crate::grid::hardening::{
9    quorum_read_your_writes, MergePolicy, ReplicatedValueRecord, ValueVersion, WriteWatermark,
10};
11
12/// Write-authority mode for a geo-distributed partition.
13#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
14#[serde(rename_all = "snake_case")]
15pub enum WriteAuthority {
16    /// The 0.43 default: only the home region may accept writes.
17    HomeRegionOnly { home: RegionId },
18    /// 0.45 opt-in: any region may accept a local write and converge via home.
19    ActiveActive { home: RegionId },
20}
21
22impl WriteAuthority {
23    /// Return the home region that still owns authoritative ordering.
24    pub fn home(&self) -> &RegionId {
25        match self {
26            Self::HomeRegionOnly { home } | Self::ActiveActive { home } => home,
27        }
28    }
29
30    /// Return whether remote regions may acknowledge local writes.
31    pub fn accepts_local_writes_in(&self, region: &RegionId) -> bool {
32        match self {
33            Self::HomeRegionOnly { home } => home == region,
34            Self::ActiveActive { .. } => true,
35        }
36    }
37}
38
39/// Explicit acknowledgement that active-active weakens cross-region consistency.
40#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
41#[serde(rename_all = "snake_case")]
42pub enum ActiveActiveAcknowledgement {
43    /// The caller did not acknowledge bounded cross-region staleness.
44    Missing,
45    /// The caller accepted the bounded-staleness contract explicitly.
46    BoundedStalenessAccepted,
47}
48
49/// Error returned when active-active is requested without the loud acknowledgement.
50#[derive(Debug, Clone, PartialEq, Eq)]
51pub struct ActiveActiveConfigError {
52    message: String,
53}
54
55impl ActiveActiveConfigError {
56    fn new(message: impl Into<String>) -> Self {
57        Self {
58            message: message.into(),
59        }
60    }
61}
62
63impl fmt::Display for ActiveActiveConfigError {
64    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
65        formatter.write_str(&self.message)
66    }
67}
68
69impl std::error::Error for ActiveActiveConfigError {}
70
71/// Active-active mode configuration for one cache/grid slice.
72#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
73pub struct ActiveActiveConfig {
74    /// Write authority mode.
75    pub authority: WriteAuthority,
76    /// Whether the weaker cross-region contract was acknowledged.
77    pub acknowledgement: ActiveActiveAcknowledgement,
78}
79
80impl ActiveActiveConfig {
81    /// Create the 0.43-compatible single-authority mode.
82    pub fn home_region_only(home: impl Into<RegionId>) -> Self {
83        Self {
84            authority: WriteAuthority::HomeRegionOnly { home: home.into() },
85            acknowledgement: ActiveActiveAcknowledgement::Missing,
86        }
87    }
88
89    /// Create active-active mode only when the caller acknowledges bounded staleness.
90    pub fn active_active(
91        home: impl Into<RegionId>,
92        acknowledgement: ActiveActiveAcknowledgement,
93    ) -> Result<Self, ActiveActiveConfigError> {
94        if acknowledgement != ActiveActiveAcknowledgement::BoundedStalenessAccepted {
95            return Err(ActiveActiveConfigError::new(
96                "active-active requires explicit bounded-staleness acknowledgement",
97            ));
98        }
99        Ok(Self {
100            authority: WriteAuthority::ActiveActive { home: home.into() },
101            acknowledgement,
102        })
103    }
104
105    /// Return whether this config is ready to accept active-active writes.
106    pub fn active_active_ready(&self) -> bool {
107        matches!(self.authority, WriteAuthority::ActiveActive { .. })
108            && self.acknowledgement == ActiveActiveAcknowledgement::BoundedStalenessAccepted
109    }
110}
111
112/// Hybrid logical clock timestamp used for deterministic cross-region tie-breaks.
113#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
114pub struct HybridLogicalClock {
115    wall: u64,
116    logical: u32,
117}
118
119impl HybridLogicalClock {
120    /// Create a timestamp from wall/logical components.
121    pub const fn new(wall: u64, logical: u32) -> Self {
122        Self { wall, logical }
123    }
124
125    /// Return the physical component observed by the node.
126    pub const fn wall(self) -> u64 {
127        self.wall
128    }
129
130    /// Return the logical component.
131    pub const fn logical(self) -> u32 {
132        self.logical
133    }
134
135    /// Advance the clock for a local event.
136    pub fn tick(&mut self, observed_wall: u64) -> Self {
137        if observed_wall > self.wall {
138            self.wall = observed_wall;
139            self.logical = 0;
140        } else {
141            self.logical = self.logical.saturating_add(1);
142        }
143        *self
144    }
145
146    /// Observe a remote timestamp and return the next local timestamp.
147    pub fn observe(&mut self, remote: Self, observed_wall: u64) -> Self {
148        let max_wall = self.wall.max(remote.wall).max(observed_wall);
149        self.logical = if max_wall == self.wall && max_wall == remote.wall {
150            self.logical.max(remote.logical).saturating_add(1)
151        } else if max_wall == self.wall {
152            self.logical.saturating_add(1)
153        } else if max_wall == remote.wall {
154            remote.logical.saturating_add(1)
155        } else {
156            0
157        };
158        self.wall = max_wall;
159        *self
160    }
161}
162
163/// Active-active write stamped at the accepting region.
164#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
165pub struct GeoWrite {
166    /// Cache key being written.
167    pub key: String,
168    /// Owning partition.
169    pub partition: PartitionId,
170    /// Monotonic value version.
171    pub version: ValueVersion,
172    /// Authority epoch.
173    pub epoch: ClusterEpoch,
174    /// HLC used only as a deterministic tie-break.
175    pub hlc: HybridLogicalClock,
176    /// Region that accepted the write locally.
177    pub origin_region: RegionId,
178    /// Node that accepted the write.
179    pub origin_node: ClusterNodeId,
180    /// Sealed value bytes.
181    pub value: Vec<u8>,
182}
183
184impl GeoWrite {
185    /// Convert a write into a replicated value record.
186    pub fn to_record(&self) -> ReplicatedValueRecord {
187        ReplicatedValueRecord::value(self.partition, self.version, self.epoch, self.value.clone())
188    }
189}
190
191/// Result of accepting a local write.
192#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
193pub struct GeoWriteAck {
194    /// Watermark visible inside the accepting region.
195    pub watermark: WriteWatermark,
196    /// Whether the acknowledgement path crossed the WAN.
197    pub crossed_wan: bool,
198    /// Regions that must receive the write asynchronously.
199    pub replication_targets: Vec<RegionId>,
200}
201
202/// Deterministic active-active state machine used by release gates and adapters.
203#[derive(Debug, Clone)]
204pub struct ActiveActiveState {
205    config: ActiveActiveConfig,
206    local_region: RegionId,
207    local_node: ClusterNodeId,
208    partition_count: u32,
209    epoch: ClusterEpoch,
210    next_version: ValueVersion,
211    clock: HybridLogicalClock,
212    peers: Vec<RegionId>,
213    records: BTreeMap<String, ReplicatedValueRecord>,
214    writes: BTreeMap<String, GeoWrite>,
215    pending: Vec<GeoWrite>,
216}
217
218impl ActiveActiveState {
219    /// Create a deterministic active-active state machine.
220    pub fn new(
221        config: ActiveActiveConfig,
222        local_region: impl Into<RegionId>,
223        local_node: impl Into<ClusterNodeId>,
224        partition_count: u32,
225        epoch: ClusterEpoch,
226        peers: Vec<RegionId>,
227    ) -> Self {
228        Self {
229            config,
230            local_region: local_region.into(),
231            local_node: local_node.into(),
232            partition_count: partition_count.max(1),
233            epoch,
234            next_version: 1,
235            clock: HybridLogicalClock::new(0, 0),
236            peers,
237            records: BTreeMap::new(),
238            writes: BTreeMap::new(),
239            pending: Vec::new(),
240        }
241    }
242
243    /// Accept a local write, apply it locally, and enqueue cross-region propagation.
244    pub fn accept_local_write(
245        &mut self,
246        key: impl Into<String>,
247        value: impl Into<Vec<u8>>,
248        observed_wall: u64,
249    ) -> Result<GeoWriteAck, ActiveActiveConfigError> {
250        if !self
251            .config
252            .authority
253            .accepts_local_writes_in(&self.local_region)
254        {
255            return Err(ActiveActiveConfigError::new(
256                "local region is not allowed to accept writes for this authority mode",
257            ));
258        }
259        if matches!(self.config.authority, WriteAuthority::ActiveActive { .. })
260            && !self.config.active_active_ready()
261        {
262            return Err(ActiveActiveConfigError::new(
263                "active-active write refused without bounded-staleness acknowledgement",
264            ));
265        }
266
267        let key = key.into();
268        let partition = partition_for_key(&key, self.partition_count);
269        let version = self.next_version;
270        self.next_version = self.next_version.saturating_add(1);
271        let hlc = self.clock.tick(observed_wall);
272        let write = GeoWrite {
273            key: key.clone(),
274            partition,
275            version,
276            epoch: self.epoch,
277            hlc,
278            origin_region: self.local_region.clone(),
279            origin_node: self.local_node.clone(),
280            value: value.into(),
281        };
282        self.records.insert(key, write.to_record());
283        self.writes.insert(write.key.clone(), write.clone());
284        let targets = self.replication_targets_for(&write.origin_region);
285        if matches!(self.config.authority, WriteAuthority::ActiveActive { .. }) {
286            self.pending.push(write);
287        }
288        Ok(GeoWriteAck {
289            watermark: WriteWatermark::new(partition, version, self.epoch),
290            crossed_wan: false,
291            replication_targets: targets,
292        })
293    }
294
295    /// Reconcile a remote write with local authoritative state.
296    pub fn reconcile_remote(&mut self, write: GeoWrite, policy: &dyn MergePolicy) {
297        self.clock.observe(write.hlc, write.hlc.wall());
298        let key = write.key.clone();
299        if let Some(existing) = self.writes.get(&key) {
300            let winner = choose_hlc_tiebreak(existing, &write).clone();
301            self.records.insert(key.clone(), winner.to_record());
302            self.writes.insert(key, winner);
303            return;
304        }
305
306        let incoming = write.to_record();
307        let merged = policy
308            .merge(self.records.get(&key), &incoming)
309            .unwrap_or(incoming);
310        self.records.insert(key, merged);
311        self.writes.insert(write.key.clone(), write);
312    }
313
314    /// Return a record by key.
315    pub fn record(&self, key: &str) -> Option<&ReplicatedValueRecord> {
316        self.records.get(key)
317    }
318
319    /// Drain pending cross-region writes in deterministic FIFO order.
320    pub fn drain_pending(&mut self) -> Vec<GeoWrite> {
321        std::mem::take(&mut self.pending)
322    }
323
324    /// Check the existing 0.42 in-region read-your-writes contract.
325    pub fn intra_region_read_your_writes_holds(
326        &self,
327        watermark: WriteWatermark,
328        read_quorum: usize,
329    ) -> bool {
330        let replicas = self
331            .records
332            .values()
333            .filter(|record| record.partition == watermark.partition)
334            .cloned()
335            .collect::<Vec<_>>();
336        !quorum_read_your_writes(watermark, replicas, read_quorum).requires_primary_fallback
337    }
338
339    fn replication_targets_for(&self, origin: &RegionId) -> Vec<RegionId> {
340        let mut targets = Vec::new();
341        if self.config.authority.home() != origin {
342            targets.push(self.config.authority.home().clone());
343        }
344        for peer in &self.peers {
345            if peer != origin && !targets.contains(peer) {
346                targets.push(peer.clone());
347            }
348        }
349        targets
350    }
351}
352
353/// Deterministically choose between equal `(version, epoch)` writes using HLC.
354pub fn choose_hlc_tiebreak<'a>(left: &'a GeoWrite, right: &'a GeoWrite) -> &'a GeoWrite {
355    match (left.version, left.epoch).cmp(&(right.version, right.epoch)) {
356        std::cmp::Ordering::Greater => left,
357        std::cmp::Ordering::Less => right,
358        std::cmp::Ordering::Equal => {
359            if (left.hlc, &left.origin_region, &left.origin_node)
360                >= (right.hlc, &right.origin_region, &right.origin_node)
361            {
362                left
363            } else {
364                right
365            }
366        }
367    }
368}
hydracache/grid/active_active.rs

hydracache/grid/
active_active.rs
