appcore_control_plane/
leadership.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: leadership.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/07/22 15:41:18 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/07/24 16:07:49 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11use super::*;
12
13/// In-memory service lease guard for embedded hosts and tests.
14#[derive(Debug, Clone, Default)]
15pub struct StaticServiceLeadershipGuard {
16    leases: Arc<Mutex<BTreeMap<ServiceId, ServiceLeaderLease>>>,
17}
18
19impl StaticServiceLeadershipGuard {
20    /// Creates a guard from zero or more independently scoped leases.
21    pub fn new(leases: impl IntoIterator<Item = ServiceLeaderLease>) -> Self {
22        Self {
23            leases: Arc::new(Mutex::new(
24                leases
25                    .into_iter()
26                    .map(|lease| (lease.service_id.clone(), lease))
27                    .collect(),
28            )),
29        }
30    }
31
32    /// Adds, renews, or removes the lease for one service.
33    pub fn set_service_lease(
34        &self,
35        service_id: ServiceId,
36        lease: Option<ServiceLeaderLease>,
37    ) -> ControlPlaneResult<()> {
38        if lease
39            .as_ref()
40            .is_some_and(|lease| lease.service_id != service_id)
41        {
42            return Err(ControlPlaneError::Conflict(
43                "service lease identity mismatch".to_string(),
44            ));
45        }
46        let mut leases = self
47            .leases
48            .lock()
49            .map_err(|_| ControlPlaneError::Transport("service lease lock poisoned".to_string()))?;
50        if let (Some(existing), Some(replacement)) = (leases.get(&service_id), lease.as_ref()) {
51            let same_scope = existing.tenant_id == replacement.tenant_id
52                && existing.cluster_id == replacement.cluster_id;
53            if same_scope && replacement.epoch < existing.epoch {
54                return Err(ControlPlaneError::Conflict("stale lease epoch".to_string()));
55            }
56            if same_scope
57                && replacement.epoch == existing.epoch
58                && replacement.holder_core_id != existing.holder_core_id
59            {
60                return Err(ControlPlaneError::Conflict(
61                    "lease epoch holder conflict".to_string(),
62                ));
63            }
64        }
65        match lease {
66            Some(lease) => {
67                leases.insert(service_id, lease);
68            }
69            None => {
70                leases.remove(&service_id);
71            }
72        }
73        Ok(())
74    }
75}
76
77impl ServiceLeadershipGuard for StaticServiceLeadershipGuard {
78    fn current_service_lease(&self, service_id: &ServiceId) -> Option<ServiceLeaderLease> {
79        self.leases
80            .lock()
81            .ok()
82            .and_then(|leases| leases.get(service_id).cloned())
83    }
84
85    fn check_service_write_permission(
86        &self,
87        service_id: &ServiceId,
88        tenant_id: &TenantId,
89        cluster_id: &ClusterId,
90        core_id: &CoreId,
91        min_epoch: Option<u64>,
92        now_ms: u64,
93    ) -> LeadershipDecision {
94        let Some(lease) = self.current_service_lease(service_id) else {
95            return LeadershipDecision::NoLease;
96        };
97        if lease.expires_at_ms <= now_ms {
98            return LeadershipDecision::Expired;
99        }
100        if lease.service_id != *service_id
101            || lease.tenant_id != *tenant_id
102            || lease.cluster_id != *cluster_id
103            || lease.holder_core_id != *core_id
104        {
105            return LeadershipDecision::WrongHolder;
106        }
107        if min_epoch.map(|epoch| lease.epoch < epoch).unwrap_or(false) {
108            return LeadershipDecision::StaleEpoch;
109        }
110        LeadershipDecision::Allowed
111    }
112}
appcore_control_plane/leadership.rs

appcore_control_plane/
leadership.rs
