clawft_kernel/workload_ctl/
plane.rs

1//! The controller side of placement (ADR-099 sections 3 and 7).
2//!
3//! [`PlacementControlPlane`] learns targets over signed `describe` round
4//! trips (their signed facts go into a [`NodeFactsCache`] under the trust
5//! tier the operator assigned, and their `workload-host` advertisement
6//! into a [`ClusterServiceRegistry`]), runs the pure engine on the cached
7//! facts, chains the decision, and dispatches signed `workload.ctl`
8//! requests. Placement itself is in [`super::plane_place`].
9
10use std::collections::BTreeMap;
11use std::sync::{Arc, Mutex, RwLock};
12use std::time::Duration;
13
14use clawft_types::placement::TrustTier as FactsTier;
15use clawft_types::placement::engine::{Liveness, ScoringWeights};
16use ed25519_dalek::SigningKey;
17use serde::{Deserialize, Serialize};
18use serde_json::{Value, json};
19
20use crate::chain::ChainManager;
21use crate::cluster::ClusterMembership;
22use crate::gate::GateBackend;
23use crate::mesh_artifact::ArtifactExchange;
24use crate::mesh_service_adv::{ClusterServiceRegistry, ServiceAdvertisement};
25use crate::node_facts::NodeFactsCache;
26use crate::node_facts_advert::SignedNodeFacts;
27use crate::node_registry::node_id_from_pubkey;
28use crate::workload_kind::KindRegistry;
29use crate::workload_pkg::TrustAnchors;
30use crate::workload_runtime::WorkloadHost;
31
32use super::facts::{LiveNodeFacts, placement_view};
33use super::msg::{ANY_TARGET, CtlOutcome, CtlRequest, Refusal, method, verify_response};
34use super::session::CtlConnection;
35use super::transport::CtlConnector;
36
37/// Chain source for controller-side placement records.
38pub const PLANE_CHAIN_SOURCE: &str = "workload.placement";
39
40/// Tunables.
41#[derive(Debug, Clone)]
42pub struct PlaneConfig {
43    /// Timeout for read-only calls.
44    pub call_timeout: Duration,
45    /// Timeout for `place` (includes the payload fetch).
46    pub place_timeout: Duration,
47    /// Request lifetime.
48    pub request_ttl_ms: u64,
49    /// Scoring weights (governance config).
50    pub weights: ScoringWeights,
51    /// A target unreachable for this long counts as `Dead` (membership
52    /// peer state, when known, decides first). Its instances become `Lost`
53    /// and are rescheduled by [`PlacementControlPlane::lifecycle_tick`].
54    pub dead_after: Duration,
55    /// Timeout for the `place` of a reschedule (shorter than
56    /// `place_timeout`: a replacement that cannot be fetched in this time
57    /// is retried on a later tick rather than holding this one).
58    pub reschedule_timeout: Duration,
59    /// First wait before retrying a reschedule that found no node, or an
60    /// orphan unload that failed; doubles per attempt up to five minutes.
61    pub retry_backoff: Duration,
62}
63
64impl Default for PlaneConfig {
65    fn default() -> Self {
66        Self {
67            call_timeout: Duration::from_secs(20),
68            place_timeout: Duration::from_secs(280),
69            request_ttl_ms: 290_000,
70            weights: ScoringWeights::default(),
71            dead_after: Duration::from_secs(30),
72            reschedule_timeout: Duration::from_secs(60),
73            retry_backoff: Duration::from_secs(10),
74        }
75    }
76}
77
78/// Why a call to a target failed.
79#[derive(Debug, Clone, PartialEq, thiserror::Error)]
80pub enum CallFailure {
81    /// The target answered with a signed refusal.
82    #[error("{0}")]
83    Refused(Refusal),
84    /// No valid answer, and the request never left this node (connect
85    /// failed), or it did not change state. Safe to try elsewhere.
86    #[error("unreachable: {0}")]
87    Unreachable(String),
88    /// A state-changing request was sent but no valid answer came back
89    /// (timeout, dropped connection, bad signature): the target may have
90    /// acted on it. Never retried elsewhere before reconciling.
91    #[error("indeterminate: {0}")]
92    Indeterminate(String),
93}
94
95/// Controller-level errors.
96#[derive(Debug, Clone, PartialEq, thiserror::Error)]
97pub enum PlaneError {
98    /// Bad input.
99    #[error("invalid: {0}")]
100    Invalid(String),
101    /// Package could not be verified or seeded.
102    #[error("package: {0}")]
103    Package(String),
104    /// Unknown node or instance.
105    #[error("unknown: {0}")]
106    Unknown(String),
107    /// A call failed.
108    #[error(transparent)]
109    Call(#[from] CallFailure),
110    /// Governance refused.
111    #[error("governance: {0}")]
112    Governance(String),
113    /// The engine refused the request (bad pin, invalid spec).
114    #[error("placement: {0}")]
115    Placement(String),
116    /// The package's workload kind is not registered.
117    #[error(transparent)]
118    UnknownKind(#[from] crate::workload_kind::UnknownKind),
119}
120
121/// A known target.
122#[derive(Debug, Clone, Serialize, Deserialize)]
123pub struct TargetInfo {
124    /// Node id.
125    pub node_id: String,
126    /// Where its `workload-host` listens.
127    pub addr: String,
128    /// Operator-assigned trust tier.
129    pub tier: FactsTier,
130    /// Signing key (hex) from its signed facts.
131    pub public_key: String,
132    /// Last `describe` succeeded.
133    pub reachable: bool,
134    /// When this key was first learned (ms): an operator peer listed
135    /// without a key trusts the first key learned at its address only.
136    #[serde(default)]
137    pub learned_ms: u64,
138}
139
140/// One placed instance, as the controller knows it.
141#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
142pub struct PlacementRecord {
143    /// Instance id on the target.
144    pub instance_id: String,
145    /// Target node.
146    pub node_id: String,
147    /// Workload kind.
148    pub kind: String,
149    /// Workload name.
150    pub workload: String,
151    /// Chosen variant.
152    pub variant: String,
153    /// Chained decision.
154    pub decision_id: String,
155    /// Package manifest hash.
156    pub manifest_hash: String,
157    /// Project that placed it, when placement went through a project; the
158    /// cog ingest bridge delivers the instance's vectors to that project's
159    /// store (ADR-100, Decision 5).
160    #[serde(default, skip_serializing_if = "Option::is_none")]
161    pub project_id: Option<String>,
162    /// Package id the instance was placed from (what a revocation names).
163    /// Absent on records written before revocation tracking: the manifest
164    /// hash stands in for it.
165    #[serde(default, skip_serializing_if = "Option::is_none")]
166    pub package_id: Option<String>,
167    /// Hex keys of the package's signers, for revocation.
168    #[serde(default, skip_serializing_if = "Vec::is_empty")]
169    pub signer_keys: Vec<String>,
170    /// BLAKE3 hex hashes of the package's artifacts, for revocation.
171    #[serde(default, skip_serializing_if = "Vec::is_empty")]
172    pub artifact_hashes: Vec<String>,
173}
174
175impl PlacementRecord {
176    /// What the revocation list is checked against for this placement:
177    /// the recorded refs, or for an older record the manifest hash.
178    pub(super) fn revocation_refs(&self) -> (Option<String>, Vec<String>, Vec<String>) {
179        let mut hashes = self.artifact_hashes.clone();
180        if self.package_id.is_none() && self.signer_keys.is_empty() && hashes.is_empty() {
181            let h = self.manifest_hash.to_ascii_lowercase();
182            if h.len() == 64 && h.chars().all(|c| c.is_ascii_hexdigit()) {
183                hashes.push(h);
184            }
185        }
186        (self.package_id.clone(), self.signer_keys.clone(), hashes)
187    }
188}
189
190/// One signed round trip.
191pub(super) struct Call<'a> {
192    pub addr: &'a str,
193    pub target: &'a str,
194    pub expect: Option<[u8; 32]>,
195    pub method: &'a str,
196    pub decision_id: Option<String>,
197    pub body: Value,
198    /// Serve the payload to the target during the call.
199    pub serve: bool,
200    /// Replace the method's default timeout (a reschedule must not hold
201    /// the lifecycle tick for the full `place_timeout`).
202    pub timeout: Option<Duration>,
203}
204
205/// The controller.
206pub struct PlacementControlPlane {
207    pub(super) node_id: String,
208    pub(super) key: SigningKey,
209    pub(super) gate: Arc<dyn GateBackend>,
210    pub(super) chain: Arc<ChainManager>,
211    pub(super) exchange: Arc<ArtifactExchange>,
212    pub(super) anchors: TrustAnchors,
213    /// Registered workload kinds (`cog` by default).
214    pub(super) kinds: KindRegistry,
215    pub(super) connector: Arc<dyn CtlConnector>,
216    pub(super) facts: NodeFactsCache,
217    pub(super) targets: RwLock<BTreeMap<String, TargetInfo>>,
218    pub(super) services: Mutex<ClusterServiceRegistry>,
219    pub(super) membership: Option<Arc<ClusterMembership>>,
220    pub(super) placements: Mutex<BTreeMap<String, PlacementRecord>>,
221    /// Decisions whose `place` answer was lost and could not be reconciled,
222    /// by decision id (instance id empty until adopted).
223    pub(super) unsettled: Mutex<BTreeMap<String, PlacementRecord>>,
224    /// Seed adapters by operator-assigned node id (card 09's `remote.api`),
225    /// with the trust tier the operator assigned.
226    pub(super) seeds: RwLock<BTreeMap<String, (Arc<WorkloadHost>, FactsTier)>>,
227    /// Handles of instances placed on Seeds.
228    pub(super) seed_handles: Mutex<BTreeMap<String, super::plane_seed::SeedEntry>>,
229    pub(super) cfg: PlaneConfig,
230    /// Lifecycle of each placed instance (state machine, stored order and
231    /// policy for rescheduling), by instance id. See `plane_lifecycle`.
232    pub(super) lives: Mutex<BTreeMap<String, super::plane_lifecycle::ControllerLife>>,
233    /// When each target was first seen unreachable (ms), cleared on contact.
234    pub(super) down_since: Mutex<BTreeMap<String, u64>>,
235    /// Where targets and placements are persisted (see `plane_state`).
236    pub(super) state_file: Option<std::path::PathBuf>,
237    pub(super) state_lock: Mutex<()>,
238}
239
240pub(super) fn now_ms() -> u64 {
241    chrono::Utc::now().timestamp_millis().max(0) as u64
242}
243
244impl PlacementControlPlane {
245    /// Replace the workload-kind registry (default: [`KindRegistry::builtin`]).
246    pub fn with_kind_registry(mut self, kinds: KindRegistry) -> Self {
247        self.kinds = kinds;
248        self
249    }
250
251    /// Controller for the node owning `key`. Every decision is chained.
252    pub fn new(
253        key: SigningKey,
254        gate: Arc<dyn GateBackend>,
255        chain: Arc<ChainManager>,
256        exchange: Arc<ArtifactExchange>,
257        anchors: TrustAnchors,
258        connector: Arc<dyn CtlConnector>,
259    ) -> Self {
260        Self {
261            node_id: node_id_from_pubkey(&key.verifying_key().to_bytes()),
262            key,
263            gate,
264            chain,
265            exchange,
266            anchors,
267            kinds: KindRegistry::builtin(),
268            connector,
269            facts: NodeFactsCache::new(),
270            targets: RwLock::new(BTreeMap::new()),
271            services: Mutex::new(ClusterServiceRegistry::new()),
272            membership: None,
273            placements: Mutex::new(BTreeMap::new()),
274            unsettled: Mutex::new(BTreeMap::new()),
275            seeds: RwLock::new(BTreeMap::new()),
276            seed_handles: Mutex::new(BTreeMap::new()),
277            cfg: PlaneConfig::default(),
278            lives: Mutex::new(BTreeMap::new()),
279            down_since: Mutex::new(BTreeMap::new()),
280            state_file: None,
281            state_lock: Mutex::new(()),
282        }
283    }
284
285    /// Use membership peer state for liveness.
286    pub fn with_membership(mut self, m: Arc<ClusterMembership>) -> Self {
287        self.membership = Some(m);
288        self
289    }
290
291    /// Replace the tunables.
292    pub fn with_config(mut self, cfg: PlaneConfig) -> Self {
293        self.cfg = cfg;
294        self
295    }
296
297    /// This controller's node id.
298    pub fn node_id(&self) -> &str {
299        &self.node_id
300    }
301
302    /// Known targets.
303    pub fn targets(&self) -> Vec<TargetInfo> {
304        self.targets
305            .read()
306            .map(|t| t.values().cloned().collect())
307            .unwrap_or_default()
308    }
309
310    /// Nodes advertising `workload-host`.
311    pub fn hosts(&self) -> Vec<ServiceAdvertisement> {
312        self.services
313            .lock()
314            .map(|s| {
315                s.resolve(super::msg::WORKLOAD_HOST_SERVICE)
316                    .into_iter()
317                    .cloned()
318                    .collect()
319            })
320            .unwrap_or_default()
321    }
322
323    /// Placed instances.
324    pub fn placements(&self) -> Vec<PlacementRecord> {
325        self.placements
326            .lock()
327            .map(|p| p.values().cloned().collect())
328            .unwrap_or_default()
329    }
330
331    /// Adapter routes each `workload-host` advertises, by node id.
332    fn advertised_routes(&self) -> BTreeMap<String, Vec<String>> {
333        self.hosts()
334            .into_iter()
335            .map(|a| {
336                let routes = a
337                    .metadata
338                    .get("routes")
339                    .map(|r| {
340                        r.split(',')
341                            .filter(|x| !x.is_empty())
342                            .map(str::to_owned)
343                            .collect()
344                    })
345                    .unwrap_or_default();
346                (a.node_id, routes)
347            })
348            .collect()
349    }
350
351    /// The verified facts the engine will see, with liveness. Runtimes a
352    /// node has no adapter route for are hidden (a node without a
353    /// `workload-host` advertisement offers none).
354    pub fn view(&self) -> Vec<LiveNodeFacts> {
355        let targets = self.targets.read().map(|t| t.clone()).unwrap_or_default();
356        let contact = |id: &str| {
357            targets.get(id).map(|t| {
358                if t.reachable {
359                    Liveness::Alive
360                } else {
361                    Liveness::Suspect
362                }
363            })
364        };
365        let routes = self.advertised_routes();
366        let mut view = placement_view(
367            &self.facts,
368            now_ms() / 1000,
369            None,
370            self.membership.as_deref(),
371            &contact,
372        );
373        for v in &mut view {
374            let r = routes.get(v.cached.node_id()).map(Vec::as_slice);
375            v.restrict_to_routes(r.unwrap_or(&[]));
376        }
377        view
378    }
379
380    /// One signed round trip to `addr`. `expect` is the target's key when
381    /// known. `serve` lets the target fetch the payload over the call.
382    pub(super) async fn round_trip(&self, c: Call<'_>) -> Result<(Value, [u8; 32]), CallFailure> {
383        let Call {
384            addr,
385            target,
386            expect,
387            method: m,
388            decision_id,
389            body,
390            serve,
391            timeout: timeout_override,
392        } = c;
393        let timeout = timeout_override.unwrap_or(if m == method::PLACE || m == method::LOAD {
394            self.cfg.place_timeout
395        } else {
396            self.cfg.call_timeout
397        });
398        let req = CtlRequest::new(
399            &self.key,
400            m,
401            target,
402            now_ms(),
403            self.cfg.request_ttl_ms,
404            decision_id,
405            body,
406        );
407        let signed = req.sign(&self.key).map_err(CallFailure::Refused)?;
408        let stream = self
409            .connector
410            .connect(addr)
411            .await
412            .map_err(|e| CallFailure::Unreachable(e.to_string()))?;
413        let mut conn = CtlConnection::new(stream, self.node_id.clone());
414        let exchange = serve.then_some(self.exchange.as_ref());
415        let r = conn.call(target, m, &signed, exchange, timeout).await;
416        conn.close().await;
417        // Past this point the request may have reached the target.
418        let lost = |why: String| {
419            if method::mutates(m) {
420                CallFailure::Indeterminate(why)
421            } else {
422                CallFailure::Unreachable(why)
423            }
424        };
425        let resp_signed = r.map_err(|e| lost(e.to_string()))?;
426        let (resp, pk) = verify_response(&resp_signed, &req, expect.as_ref())
427            .map_err(|e| lost(format!("bad response: {e}")))?;
428        match resp.outcome {
429            CtlOutcome::Ok { result } => Ok((result, pk)),
430            CtlOutcome::Refused { refusal } => Err(CallFailure::Refused(refusal)),
431        }
432    }
433
434    /// Learn (or refresh) the target at `addr`: signed `describe`, verify
435    /// and cache its signed facts under `tier`, merge its advertisement.
436    /// Returns its node id. Whatever key answers is accepted (first use);
437    /// see [`Self::add_target_expecting`] to require a key.
438    pub async fn add_target(&self, addr: &str, tier: FactsTier) -> Result<String, PlaneError> {
439        self.add_target_expecting(addr, tier, None).await
440    }
441
442    /// [`Self::add_target`], refusing any answer not signed by `expect`
443    /// when given. Trust is bound to the node key: an operator tier is only
444    /// ever given to the key the operator pinned, or to the key first seen
445    /// at that address, never to whoever answers there later.
446    pub async fn add_target_expecting(
447        &self,
448        addr: &str,
449        tier: FactsTier,
450        expect: Option<[u8; 32]>,
451    ) -> Result<String, PlaneError> {
452        let (result, pk) = self
453            .round_trip(Call {
454                addr,
455                target: ANY_TARGET,
456                expect,
457                method: method::DESCRIBE,
458                decision_id: None,
459                body: json!({}),
460                serve: false,
461                timeout: None,
462            })
463            .await?;
464        let node_id = node_id_from_pubkey(&pk);
465        let facts: SignedNodeFacts = serde_json::from_value(result["facts"].clone())
466            .map_err(|e| PlaneError::Invalid(format!("{node_id} sent no signed facts: {e}")))?;
467        if facts.public_key != pk.to_vec() {
468            return Err(PlaneError::Invalid(format!(
469                "{node_id}: facts signed by another key"
470            )));
471        }
472        let now = now_ms() / 1000;
473        match self.facts.insert(facts, tier, now) {
474            Ok(_) | Err(crate::node_facts::CacheError::Stale { .. }) => {}
475            Err(e) => return Err(PlaneError::Invalid(format!("{node_id} facts refused: {e}"))),
476        }
477        // Unchanged facts keep the held tier; the tier given here wins.
478        self.facts.set_trust_tier(&node_id, tier);
479        if let Ok(adv) =
480            serde_json::from_value::<ServiceAdvertisement>(result["advertisement"].clone())
481            && adv.node_id == node_id
482            && let Ok(mut s) = self.services.lock()
483        {
484            s.merge(adv);
485        }
486        if let Ok(mut t) = self.targets.write() {
487            let learned_ms = t.get(&node_id).map_or_else(now_ms, |e| e.learned_ms);
488            t.insert(
489                node_id.clone(),
490                TargetInfo {
491                    node_id: node_id.clone(),
492                    addr: addr.to_string(),
493                    tier,
494                    public_key: crate::workload_pkg::codec::hex_encode(&pk),
495                    reachable: true,
496                    learned_ms,
497                },
498            );
499        }
500        self.persist();
501        Ok(node_id)
502    }
503
504    /// Re-describe every known target, requiring the key it was learned
505    /// with; unreachable ones are marked so the engine sees them as not
506    /// alive. A node whose address now answers with another key (it was
507    /// replaced, or someone else took the address) is marked unreachable
508    /// and the new key is not learned: it gets no tier until the operator
509    /// pins it (`OperatorPeer::key`).
510    pub async fn refresh(&self) {
511        for t in self.targets() {
512            let pk = crate::workload_pkg::codec::hex_decode_exact::<32>(&t.public_key);
513            let same = pk.is_some()
514                && matches!(self.add_target_expecting(&t.addr, t.tier, pk).await,
515                            Ok(id) if id == t.node_id);
516            if !same
517                && let Ok(mut map) = self.targets.write()
518                && let Some(e) = map.get_mut(&t.node_id)
519            {
520                e.reachable = false;
521            }
522        }
523    }
524
525    pub(super) fn target(&self, node_id: &str) -> Result<(TargetInfo, [u8; 32]), PlaneError> {
526        let t = self
527            .targets
528            .read()
529            .ok()
530            .and_then(|m| m.get(node_id).cloned())
531            .ok_or_else(|| PlaneError::Unknown(format!("no workload-host known for {node_id}")))?;
532        let pk = crate::workload_pkg::codec::hex_decode_exact::<32>(&t.public_key)
533            .ok_or_else(|| PlaneError::Invalid("bad stored key".into()))?;
534        Ok((t, pk))
535    }
536
537    /// A signed call to a known node (no payload serving).
538    pub async fn call(
539        &self,
540        node_id: &str,
541        m: &str,
542        decision_id: Option<String>,
543        body: Value,
544    ) -> Result<Value, PlaneError> {
545        let (t, pk) = self.target(node_id)?;
546        Ok(self
547            .round_trip(Call {
548                addr: &t.addr,
549                target: node_id,
550                expect: Some(pk),
551                method: m,
552                decision_id,
553                body,
554                serve: false,
555                timeout: None,
556            })
557            .await?
558            .0)
559    }
560}
clawft_kernel/workload_ctl/plane.rs

clawft_kernel/workload_ctl/
plane.rs
