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
//! Card 03's verified, TTL-cached node facts as card 04's
//! [`PlacementFacts`] (ADR-099 sections 2-3).
//!
//! [`CachedNodeFacts`] carries the signed facts, the receiver-assigned
//! trust tier and live state (deltas). Liveness is not part of the facts:
//! it comes from `ClusterMembership` peer state (heartbeats) or, for a node
//! the control plane talks to directly, from the last signed round trip.
//! [`LiveNodeFacts`] pairs the two, so `place()` consumes exactly what the
//! cache verified.
use clawft_types::placement::engine::{self, Liveness, PlacementFacts};
use clawft_types::placement::{Capability, TrustTier as FactsTier};
use crate::cluster::{ClusterMembership, NodeState};
use crate::node_facts::{CachedNodeFacts, NodeFactsCache};
use crate::workload_governance::NodeTrustTier;
/// Cached facts plus liveness: what the placement engine reads.
#[derive(Debug, Clone)]
pub struct LiveNodeFacts {
/// Verified facts (base plus deltas) with the receiver-assigned tier.
pub cached: CachedNodeFacts,
/// Liveness from membership or the last direct contact.
pub liveness: Liveness,
/// The capabilities the engine sees: the verified ones, less any
/// runtime the node's `workload-host` has no adapter route for.
caps: Vec<Capability>,
}
/// The adapter route a `runtime.*` capability needs on the target
/// (`None` for capabilities that are not runtimes).
fn route_for(cap: &Capability) -> Option<&'static [&'static str]> {
let id = cap.id.as_str();
if id == "runtime.native" || id.starts_with("runtime.native.") {
Some(&["native"])
} else if id == "runtime.container" || id.starts_with("runtime.container.") {
Some(&["container", "emulated"])
} else {
None
}
}
impl LiveNodeFacts {
/// Facts as verified, with `liveness`.
pub fn new(cached: CachedNodeFacts, liveness: Liveness) -> Self {
let caps = cached.capabilities().to_vec();
Self {
cached,
liveness,
caps,
}
}
/// Hide every runtime capability whose adapter route is not in
/// `routes` (the routes the node's `workload-host` advertises). A node
/// that probes a container engine but serves no container adapter is
/// then not offered a container variant it would always refuse.
pub fn restrict_to_routes(&mut self, routes: &[String]) {
self.caps.retain(|c| match route_for(c) {
None => true,
Some(need) => need.iter().any(|r| routes.iter().any(|x| x == r)),
});
}
}
/// Receiver-assigned facts tier as the engine's tier.
pub fn engine_tier(t: FactsTier) -> engine::TrustTier {
match t {
FactsTier::Discovered => engine::TrustTier::Discovered,
FactsTier::Paired => engine::TrustTier::Paired,
FactsTier::Pinned => engine::TrustTier::Pinned,
}
}
/// Receiver-assigned facts tier as the governance effect's tier.
pub fn governance_tier(t: FactsTier) -> NodeTrustTier {
match t {
FactsTier::Discovered => NodeTrustTier::Discovered,
FactsTier::Paired => NodeTrustTier::Paired,
FactsTier::Pinned => NodeTrustTier::Pinned,
}
}
/// Membership peer state as liveness. Only `Active` is `Alive`.
pub fn liveness_of(state: &NodeState) -> Liveness {
match state {
NodeState::Active => Liveness::Alive,
NodeState::Suspect => Liveness::Suspect,
NodeState::Unreachable | NodeState::Leaving | NodeState::Left => Liveness::Dead,
_ => Liveness::Unknown,
}
}
impl PlacementFacts for LiveNodeFacts {
fn node_id(&self) -> &str {
self.cached.node_id()
}
fn capabilities(&self) -> &[Capability] {
&self.caps
}
fn liveness(&self) -> Liveness {
self.liveness
}
fn trust_tier(&self) -> engine::TrustTier {
engine_tier(self.cached.trust_tier())
}
fn facts_expire_at_ms(&self) -> Option<u64> {
Some(self.cached.facts.expires_at().saturating_mul(1000))
}
/// Share of capabilities busy or reserved; unknown for an empty block.
fn load(&self) -> Option<f64> {
let l = self.cached.load();
(l.total > 0).then(|| l.busy as f64 / l.total as f64)
}
}
/// Every fresh cached node with its liveness. `local` is always alive;
/// otherwise `membership` peer state wins, then `contact` (the control
/// plane's last direct round trip), else [`Liveness::Unknown`].
pub fn placement_view(
cache: &NodeFactsCache,
now_secs: u64,
local: Option<&str>,
membership: Option<&ClusterMembership>,
contact: &dyn Fn(&str) -> Option<Liveness>,
) -> Vec<LiveNodeFacts> {
cache
.list(now_secs)
.into_iter()
.map(|cached| {
let id = cached.node_id().to_string();
let liveness = if local == Some(id.as_str()) {
Liveness::Alive
} else if let Some(p) = membership.and_then(|m| m.get_peer(&id)) {
liveness_of(&p.state)
} else {
contact(&id).unwrap_or(Liveness::Unknown)
};
LiveNodeFacts::new(cached, liveness)
})
.collect()
}