feltdb 0.2.0

Embedded durable state store: an append-only log with atomic multi-record transactions, version-conditional writes, and idempotent retries
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
//! Authority Failover Protocol
//!
//! This module implements automatic authority election, failover, and recovery
//! for distributed FeltDB clusters. The design follows these principles:
//!
//! 1. **Single Authority**: At any time, exactly one node holds authority.
//!    Operations flow through the authority, which coordinates replication.
//!
//! 2. **Fencing**: A stale authority cannot accept writes after a new authority
//!    is elected. Fencing tokens prevent split-brain scenarios.
//!
//! 3. **Automatic Failover**: When the authority fails, surviving nodes detect
//!    the failure and elect a new authority without application intervention.
//!
//! 4. **Recovery**: A previously-failed authority can rejoin as a replica and
//!    catch up to the current state.
//!
//! ## Consistency Contract
//!
//! - A write is **committed** when the authority has persisted it durably
//!   and at least one other active replica has acknowledged receipt.
//! - A committed write survives any single node failure.
//! - During failover, a brief period of unavailability occurs while
//!   the new authority is elected.
//! - The protocol prevents two nodes from believing they are authority
//!   simultaneously through monotonic fencing tokens.
//!
//! ## Failover Sequence
//!
//! ```text
//! Authority A fails
//!      │
//!      â–¼
//! Replicas detect failure (heartbeat timeout)
//!      │
//!      â–¼
//! Election: highest-priority surviving replica
//!      │
//!      â–¼
//! New authority B claims with fencing token
//!      │
//!      â–¼
//! B broadcasts authority change
//!      │
//!      â–¼
//! Replicas acknowledge new authority
//!      │
//!      â–¼
//! B begins accepting writes
//!      │
//!      â–¼
//! Old authority A rejoins as replica
//! ```

use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, HashSet};
use std::path::{Path, PathBuf};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

fn now_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|elapsed| elapsed.as_millis() as u64)
        .unwrap_or(0)
}

/// Authority role of a node in the cluster.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
pub enum AuthorityRole {
    /// This node is the authority and accepts writes.
    Authority,
    /// This node is a replica and forwards writes to the authority.
    Replica,
    /// This node is a candidate seeking to become authority.
    Candidate,
    /// This node has been fenced (stale authority).
    Fenced,
}

/// A fencing token that prevents stale authorities from operating.
///
/// Monotonically increasing: a higher token always supersedes a lower one.
/// The fencing token is durably stored and survives restarts.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub struct FencingToken(pub u64);

impl FencingToken {
    pub fn initial() -> Self {
        Self(1)
    }

    pub fn next(&self) -> Self {
        Self(self.0.saturating_add(1))
    }

    pub fn supersedes(&self, other: &FencingToken) -> bool {
        self.0 > other.0
    }
}

impl std::fmt::Display for FencingToken {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "FT:{}", self.0)
    }
}

/// Record of an authority election.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct AuthorityRecord {
    /// The node ID of the authority.
    pub node_id: String,
    /// The fencing token for this authority term.
    pub fencing_token: FencingToken,
    /// When this authority was elected.
    pub elected_at_ms: u64,
    /// The previous authority (if any).
    pub previous_authority: Option<String>,
    /// Nodes that acknowledged this authority.
    pub acknowledgements: HashSet<String>,
}

impl AuthorityRecord {
    pub fn new(node_id: String, fencing_token: FencingToken, previous: Option<String>) -> Self {
        Self {
            node_id,
            fencing_token,
            elected_at_ms: now_ms(),
            previous_authority: previous,
            acknowledgements: HashSet::new(),
        }
    }
}

/// Heartbeat tracking for failure detection.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct HeartbeatState {
    /// Last heartbeat received from each peer.
    pub last_seen: BTreeMap<String, u64>,
    /// Heartbeat interval in milliseconds.
    pub interval_ms: u64,
    /// How many missed heartbeats before a node is considered failed.
    pub failure_threshold: u32,
}

impl HeartbeatState {
    pub fn new(interval_ms: u64, failure_threshold: u32) -> Self {
        Self {
            last_seen: BTreeMap::new(),
            interval_ms,
            failure_threshold,
        }
    }

    /// Record a heartbeat from a peer.
    pub fn record_heartbeat(&mut self, node_id: &str) {
        self.last_seen.insert(node_id.to_string(), now_ms());
    }

    /// Check if a node appears to have failed.
    pub fn is_failed(&self, node_id: &str) -> bool {
        let Some(last) = self.last_seen.get(node_id) else {
            // Never seen = not yet known, not necessarily failed
            return false;
        };
        let deadline = self.interval_ms * self.failure_threshold as u64;
        now_ms().saturating_sub(*last) > deadline
    }

    /// Get all nodes that appear to have failed.
    pub fn failed_nodes(&self) -> Vec<String> {
        self.last_seen
            .keys()
            .filter(|node_id| self.is_failed(node_id))
            .cloned()
            .collect()
    }
}

/// Election priority for a node.
/// Higher priority wins the election when multiple candidates exist.
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub struct ElectionPriority {
    /// Operations applied (higher is better - more up-to-date).
    pub operations_applied: u64,
    /// Static priority assigned to the node (for tie-breaking).
    pub static_priority: u32,
    /// Node ID (for deterministic tie-breaking).
    pub node_id: String,
}

impl ElectionPriority {
    pub fn new(operations_applied: u64, static_priority: u32, node_id: String) -> Self {
        Self {
            operations_applied,
            static_priority,
            node_id,
        }
    }
}

/// The result of an authority claim attempt.
#[derive(Clone, Debug)]
pub enum ClaimResult {
    /// Successfully claimed authority with this fencing token.
    Granted { token: FencingToken },
    /// Another node already has a higher fencing token.
    Rejected {
        current_authority: String,
        current_token: FencingToken,
    },
    /// This node is fenced and cannot claim authority.
    Fenced { token: FencingToken },
}

/// Authority state that is durably persisted.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct AuthorityState {
    /// The cluster ID this authority state belongs to.
    pub cluster_id: String,
    /// Current authority record, if any.
    pub current_authority: Option<AuthorityRecord>,
    /// This node's role.
    pub local_role: AuthorityRole,
    /// This node's ID.
    pub local_node_id: String,
    /// The highest fencing token this node has seen.
    pub highest_seen_token: FencingToken,
    /// Election priority for this node.
    pub local_priority: u32,
}

impl AuthorityState {
    pub fn new(cluster_id: String, local_node_id: String, local_priority: u32) -> Self {
        Self {
            cluster_id,
            current_authority: None,
            local_role: AuthorityRole::Replica,
            local_node_id,
            highest_seen_token: FencingToken::initial(),
            local_priority,
        }
    }

    /// Check if this node is currently the authority.
    pub fn is_authority(&self) -> bool {
        self.local_role == AuthorityRole::Authority
    }

    /// Check if this node is fenced.
    pub fn is_fenced(&self) -> bool {
        self.local_role == AuthorityRole::Fenced
    }

    /// Get the current authority node ID.
    pub fn authority_node(&self) -> Option<&str> {
        self.current_authority.as_ref().map(|r| r.node_id.as_str())
    }

    /// Get the current fencing token.
    pub fn current_token(&self) -> FencingToken {
        self.current_authority
            .as_ref()
            .map(|r| r.fencing_token)
            .unwrap_or(FencingToken::initial())
    }
}

/// Persistent authority store.
pub struct AuthorityStore {
    path: PathBuf,
    state: AuthorityState,
    heartbeats: HeartbeatState,
}

#[derive(Debug)]
pub enum AuthorityError {
    NoCluster,
    AlreadyAuthority(String),
    NotAuthority,
    Fenced(FencingToken),
    StaleToken {
        provided: FencingToken,
        current: FencingToken,
    },
    QuorumUnavailable {
        live: usize,
        required: usize,
    },
    Io(String),
}

impl std::fmt::Display for AuthorityError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::NoCluster => write!(f, "no cluster has been created"),
            Self::AlreadyAuthority(id) => write!(f, "node {id} is already authority"),
            Self::NotAuthority => write!(f, "this node is not the authority"),
            Self::Fenced(token) => write!(f, "this node is fenced at {token}"),
            Self::StaleToken { provided, current } => {
                write!(f, "stale token {provided}, current is {current}")
            }
            Self::QuorumUnavailable { live, required } => {
                write!(
                    f,
                    "authority quorum unavailable: {live} live, {required} required"
                )
            }
            Self::Io(msg) => write!(f, "authority store: {msg}"),
        }
    }
}

impl AuthorityStore {
    /// Open or create an authority store.
    pub fn open<P: AsRef<Path>>(
        path: P,
        cluster_id: &str,
        local_node_id: &str,
        local_priority: u32,
    ) -> Result<Self, AuthorityError> {
        Self::open_with_failure_detector(path, cluster_id, local_node_id, local_priority, 1_000, 5)
    }

    /// Open with an explicitly configured failure detector. Production keeps
    /// the conservative default above; process-level acceptance tests use a
    /// shorter interval so failure is event-driven without multi-second sleeps.
    pub fn open_with_failure_detector<P: AsRef<Path>>(
        path: P,
        cluster_id: &str,
        local_node_id: &str,
        local_priority: u32,
        heartbeat_interval_ms: u64,
        failure_threshold: u32,
    ) -> Result<Self, AuthorityError> {
        let path = path.as_ref().to_path_buf();
        let state = if path.exists() {
            let bytes = std::fs::read(&path).map_err(|e| AuthorityError::Io(e.to_string()))?;
            serde_json::from_slice(&bytes).map_err(|e| AuthorityError::Io(e.to_string()))?
        } else {
            AuthorityState::new(
                cluster_id.to_string(),
                local_node_id.to_string(),
                local_priority,
            )
        };
        Ok(Self {
            path,
            state,
            heartbeats: HeartbeatState::new(heartbeat_interval_ms, failure_threshold),
        })
    }

    /// Persist the authority state durably.
    fn persist(&self) -> Result<(), AuthorityError> {
        if let Some(parent) = self.path.parent() {
            std::fs::create_dir_all(parent).map_err(|e| AuthorityError::Io(e.to_string()))?;
        }
        let temp = self.path.with_extension("tmp");
        let bytes = serde_json::to_vec_pretty(&self.state)
            .map_err(|e| AuthorityError::Io(e.to_string()))?;
        {
            use std::io::Write;
            let mut file =
                std::fs::File::create(&temp).map_err(|e| AuthorityError::Io(e.to_string()))?;
            file.write_all(&bytes)
                .map_err(|e| AuthorityError::Io(e.to_string()))?;
            file.sync_all()
                .map_err(|e| AuthorityError::Io(e.to_string()))?;
        }
        std::fs::rename(&temp, &self.path).map_err(|e| AuthorityError::Io(e.to_string()))?;
        if let Some(parent) = self.path.parent() {
            if let Ok(dir) = std::fs::File::open(parent) {
                let _ = dir.sync_all();
            }
        }
        Ok(())
    }

    /// Get the current authority state.
    pub fn state(&self) -> &AuthorityState {
        &self.state
    }

    /// Record a heartbeat from a peer.
    pub fn heartbeat(&mut self, node_id: &str) {
        self.heartbeats.record_heartbeat(node_id);
    }

    /// Check if the current authority has failed.
    pub fn authority_failed(&self) -> bool {
        if let Some(ref authority) = self.state.current_authority {
            if authority.node_id == self.state.local_node_id {
                return false; // We are authority, we haven't failed
            }
            return self.heartbeats.is_failed(&authority.node_id);
        }
        false
    }

    /// Get all nodes that appear to have failed.
    pub fn failed_nodes(&self) -> Vec<String> {
        self.heartbeats.failed_nodes()
    }

    /// Claim authority with a new fencing token.
    ///
    /// This node attempts to become the authority. If successful:
    /// - A new fencing token is assigned (one higher than any seen)
    /// - This node becomes authority
    /// - The previous authority (if any) is recorded
    ///
    /// If another node has a higher token, the claim is rejected.
    pub fn claim_authority(&mut self) -> Result<ClaimResult, AuthorityError> {
        if self.state.is_fenced() {
            return Ok(ClaimResult::Fenced {
                token: self.state.highest_seen_token,
            });
        }

        // If there's already an authority with a higher token, reject
        if let Some(ref current) = self.state.current_authority {
            if current.node_id != self.state.local_node_id
                && !self.heartbeats.is_failed(&current.node_id)
            {
                return Ok(ClaimResult::Rejected {
                    current_authority: current.node_id.clone(),
                    current_token: current.fencing_token,
                });
            }
        }

        // Claim with a new token
        let new_token = self.state.highest_seen_token.next();
        let previous = self
            .state
            .current_authority
            .as_ref()
            .map(|r| r.node_id.clone());

        let mut record =
            AuthorityRecord::new(self.state.local_node_id.clone(), new_token, previous);
        record
            .acknowledgements
            .insert(self.state.local_node_id.clone());

        self.state.current_authority = Some(record);
        self.state.local_role = AuthorityRole::Authority;
        self.state.highest_seen_token = new_token;

        self.persist()?;

        Ok(ClaimResult::Granted { token: new_token })
    }

    /// Accept a new authority announcement.
    ///
    /// If the fencing token is higher than any we've seen, accept the new
    /// authority and update our role accordingly.
    pub fn accept_authority(
        &mut self,
        authority_node: &str,
        token: FencingToken,
    ) -> Result<(), AuthorityError> {
        // Check if token is valid
        if !token.supersedes(&self.state.highest_seen_token)
            && token != self.state.highest_seen_token
        {
            return Err(AuthorityError::StaleToken {
                provided: token,
                current: self.state.highest_seen_token,
            });
        }

        // Update our state
        self.state.highest_seen_token = token;

        // If we were authority with a lower token, we are now fenced
        if self.state.local_role == AuthorityRole::Authority
            && authority_node != self.state.local_node_id
        {
            self.state.local_role = AuthorityRole::Fenced;
        } else if authority_node != self.state.local_node_id {
            self.state.local_role = AuthorityRole::Replica;
        }

        // Record the new authority
        self.state.current_authority = Some(AuthorityRecord::new(
            authority_node.to_string(),
            token,
            self.state
                .current_authority
                .as_ref()
                .map(|r| r.node_id.clone()),
        ));

        // Start tracking heartbeats from the new authority.
        // This allows failure detection to work even if we've never received
        // a heartbeat yet - the clock starts now.
        if authority_node != self.state.local_node_id {
            self.heartbeats.record_heartbeat(authority_node);
        }

        self.persist()
    }

    /// Acknowledge the current authority.
    pub fn acknowledge_authority(&mut self, from_node: &str) -> Result<(), AuthorityError> {
        if let Some(ref mut authority) = self.state.current_authority {
            authority.acknowledgements.insert(from_node.to_string());
            self.persist()?;
        }
        Ok(())
    }

    /// Validate a write operation against the current fencing token.
    ///
    /// Returns Ok if the token is valid (current or not provided for local writes).
    /// Returns Err if the token is stale.
    pub fn validate_write(&self, token: Option<FencingToken>) -> Result<(), AuthorityError> {
        if self.state.is_fenced() {
            return Err(AuthorityError::Fenced(self.state.highest_seen_token));
        }

        if !self.state.is_authority() {
            return Err(AuthorityError::NotAuthority);
        }

        if let Some(provided) = token {
            if provided != self.state.current_token() {
                return Err(AuthorityError::StaleToken {
                    provided,
                    current: self.state.current_token(),
                });
            }
        }

        Ok(())
    }

    /// Validate an authoritative write against both its fencing term and a
    /// live quorum. A partitioned former authority may not have observed the
    /// replacement's newer token yet, so fencing alone is insufficient until
    /// healing; expiry of its quorum evidence closes that split-brain window.
    pub fn validate_quorum_write(
        &self,
        token: FencingToken,
        required_quorum: usize,
    ) -> Result<(), AuthorityError> {
        self.validate_write(Some(token))?;
        let live = 1 + self
            .heartbeats
            .last_seen
            .keys()
            .filter(|node_id| !self.heartbeats.is_failed(node_id))
            .count();
        if live < required_quorum {
            return Err(AuthorityError::QuorumUnavailable {
                live,
                required: required_quorum,
            });
        }
        Ok(())
    }

    /// Step down from authority role (voluntary).
    pub fn step_down(&mut self) -> Result<(), AuthorityError> {
        if self.state.local_role == AuthorityRole::Authority {
            self.state.local_role = AuthorityRole::Replica;
            self.persist()?;
        }
        Ok(())
    }

    /// Recover from fenced state (after catching up).
    pub fn recover_from_fenced(&mut self) -> Result<(), AuthorityError> {
        if self.state.local_role == AuthorityRole::Fenced {
            self.state.local_role = AuthorityRole::Replica;
            self.persist()?;
        }
        Ok(())
    }

    /// Compute election priority for this node.
    pub fn election_priority(&self, operations_applied: u64) -> ElectionPriority {
        ElectionPriority::new(
            operations_applied,
            self.state.local_priority,
            self.state.local_node_id.clone(),
        )
    }

    /// Determine the winner of an election given a set of candidates.
    pub fn elect_authority(candidates: &[ElectionPriority]) -> Option<&ElectionPriority> {
        candidates.iter().max()
    }
}

/// Authority announcement message.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct AuthorityAnnouncement {
    pub authority_node: String,
    pub fencing_token: FencingToken,
    pub elected_at_ms: u64,
    pub cluster_id: String,
}

/// Heartbeat message.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Heartbeat {
    pub from_node: String,
    pub is_authority: bool,
    pub fencing_token: FencingToken,
    pub operations_applied: u64,
    pub timestamp_ms: u64,
}

impl Heartbeat {
    pub fn new(
        from_node: String,
        is_authority: bool,
        fencing_token: FencingToken,
        operations_applied: u64,
    ) -> Self {
        Self {
            from_node,
            is_authority,
            fencing_token,
            operations_applied,
            timestamp_ms: now_ms(),
        }
    }
}

/// Election request message.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct ElectionRequest {
    pub candidate_node: String,
    pub priority: ElectionPriority,
    pub proposed_token: FencingToken,
    pub timestamp_ms: u64,
}

/// Election response message.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct ElectionResponse {
    pub from_node: String,
    pub vote_granted: bool,
    pub reason: String,
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn fencing_token_ordering() {
        let t1 = FencingToken::initial();
        let t2 = t1.next();
        let t3 = t2.next();

        assert!(t2.supersedes(&t1));
        assert!(t3.supersedes(&t2));
        assert!(t3.supersedes(&t1));
        assert!(!t1.supersedes(&t2));
    }

    #[test]
    fn authority_claim_and_fencing() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("authority.json");

        // Node A claims authority
        let mut store_a =
            AuthorityStore::open(&path, "test-cluster", "A", 10).expect("open store A");
        let result = store_a.claim_authority().expect("claim");
        match result {
            ClaimResult::Granted { token } => {
                assert_eq!(token, FencingToken(2)); // initial is 1, next is 2
            }
            _ => panic!("expected Granted"),
        }
        assert!(store_a.state().is_authority());

        // Node B sees A's authority
        let path_b = dir.path().join("authority_b.json");
        let mut store_b =
            AuthorityStore::open(&path_b, "test-cluster", "B", 5).expect("open store B");
        store_b
            .accept_authority("A", FencingToken(2))
            .expect("accept");
        assert!(!store_b.state().is_authority());
        assert_eq!(store_b.state().authority_node(), Some("A"));

        // Node B tries to claim - should be rejected since A is still up
        let result_b = store_b.claim_authority().expect("claim B");
        match result_b {
            ClaimResult::Rejected {
                current_authority,
                current_token,
            } => {
                assert_eq!(current_authority, "A");
                assert_eq!(current_token, FencingToken(2));
            }
            _ => panic!("expected Rejected"),
        }
    }

    #[test]
    fn failover_when_authority_dies() {
        let dir = tempfile::tempdir().unwrap();

        // Node A is authority
        let path_a = dir.path().join("authority_a.json");
        let mut store_a =
            AuthorityStore::open(&path_a, "test-cluster", "A", 10).expect("open store A");
        store_a.claim_authority().expect("claim A");

        // Node B knows about A
        let path_b = dir.path().join("authority_b.json");
        let mut store_b =
            AuthorityStore::open(&path_b, "test-cluster", "B", 5).expect("open store B");
        store_b
            .accept_authority("A", FencingToken(2))
            .expect("accept A");

        // Simulate A dying - mark heartbeat as very old
        store_b.heartbeats.last_seen.insert("A".to_string(), 0);

        // Now B should be able to claim
        assert!(store_b.authority_failed());
        let result_b = store_b.claim_authority().expect("claim B after failover");
        match result_b {
            ClaimResult::Granted { token } => {
                assert_eq!(token, FencingToken(3)); // higher than A's token
            }
            _ => panic!("expected Granted after failover"),
        }
    }

    #[test]
    fn stale_authority_is_fenced() {
        let dir = tempfile::tempdir().unwrap();

        // A is authority with token 2
        let path_a = dir.path().join("authority_a.json");
        let mut store_a =
            AuthorityStore::open(&path_a, "test-cluster", "A", 10).expect("open store A");
        store_a.claim_authority().expect("claim A");
        assert!(store_a.state().is_authority());

        // A receives announcement that B is now authority with token 3
        store_a
            .accept_authority("B", FencingToken(3))
            .expect("accept B");

        // A should now be fenced
        assert!(store_a.state().is_fenced());
        assert!(!store_a.state().is_authority());

        // A's writes should be rejected
        let result = store_a.validate_write(None);
        assert!(matches!(result, Err(AuthorityError::Fenced(_))));
    }

    #[test]
    fn election_priority_ordering() {
        let p1 = ElectionPriority::new(100, 5, "A".to_string());
        let p2 = ElectionPriority::new(100, 10, "B".to_string());
        let p3 = ElectionPriority::new(150, 5, "C".to_string());

        // Higher operations wins
        assert!(p3 > p1);
        assert!(p3 > p2);

        // Same operations, higher priority wins
        assert!(p2 > p1);

        // Election selects highest priority
        let candidates = vec![p1.clone(), p2.clone(), p3.clone()];
        let winner = AuthorityStore::elect_authority(&candidates);
        assert_eq!(winner.map(|w| &w.node_id), Some(&"C".to_string()));
    }

    #[test]
    fn heartbeat_failure_detection() {
        let mut heartbeats = HeartbeatState::new(1000, 3); // 1s interval, 3 missed = failure

        // Record a heartbeat
        heartbeats.record_heartbeat("A");
        assert!(!heartbeats.is_failed("A"));

        // Manually set an old timestamp
        heartbeats.last_seen.insert("A".to_string(), 0);

        // Should now be detected as failed (current time is way past 3 intervals)
        assert!(heartbeats.is_failed("A"));
    }

    #[test]
    fn partitioned_authority_loses_write_quorum() {
        let dir = tempfile::tempdir().unwrap();
        let mut store = AuthorityStore::open_with_failure_detector(
            dir.path().join("authority.json"),
            "test-cluster",
            "A",
            10,
            1,
            1,
        )
        .expect("open authority store");
        let token = match store.claim_authority().expect("claim") {
            ClaimResult::Granted { token } => token,
            _ => panic!("authority claim was not granted"),
        };
        store.heartbeat("B");
        assert!(store.validate_quorum_write(token, 2).is_ok());
        store.heartbeats.last_seen.insert("B".to_string(), 0);
        assert!(matches!(
            store.validate_quorum_write(token, 2),
            Err(AuthorityError::QuorumUnavailable {
                live: 1,
                required: 2
            })
        ));
    }

    #[test]
    fn recovery_from_fenced() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("authority.json");

        let mut store = AuthorityStore::open(&path, "test-cluster", "A", 10).expect("open store");
        store.claim_authority().expect("claim");

        // Get fenced
        store
            .accept_authority("B", FencingToken(3))
            .expect("accept B");
        assert!(store.state().is_fenced());

        // Recover
        store.recover_from_fenced().expect("recover");
        assert!(!store.state().is_fenced());
        assert_eq!(store.state().local_role, AuthorityRole::Replica);
    }
}