clawft_kernel/workload_ctl/
plane_reconcile.rs

1//! Reconciling a `place` whose answer was lost (ADR-099 section 7).
2//!
3//! A signed `place` that was sent but got no valid answer (timeout,
4//! dropped connection) may have been carried out. Before any other
5//! candidate is tried, the controller asks the target for its instances
6//! and in-flight decisions:
7//!
8//! - the decision is still in flight: unsettled, stop;
9//! - it produced instances: unload each (signed, carrying the decision
10//!   id), then settled;
11//! - neither: settled (the target fetches the payload over the call's own
12//!   connection, which is closed by now, so it cannot load it later).
13//!
14//! An unreachable target is unsettled: placement stops rather than risk a
15//! second instance. The decision is remembered, and once the target
16//! answers ([`PlacementControlPlane::settle_unsettled`], run by status
17//! listings and by instance verbs on an unknown id) any instance it
18//! produced is adopted as a placement the controller manages.
19
20use serde_json::{Value, json};
21
22use super::msg::method;
23use super::plane::{PlacementControlPlane, PlacementRecord};
24use crate::chain;
25
26impl PlacementControlPlane {
27    pub(super) fn remember_unsettled(&self, template: PlacementRecord) {
28        if let Ok(mut u) = self.unsettled.lock() {
29            u.insert(template.decision_id.clone(), template);
30        }
31        self.persist();
32    }
33
34    /// Decisions still unsettled.
35    pub fn unsettled(&self) -> Vec<PlacementRecord> {
36        self.unsettled
37            .lock()
38            .map(|u| u.values().cloned().collect())
39            .unwrap_or_default()
40    }
41
42    /// Ask each target holding an unsettled decision what became of it:
43    /// adopt the instances it produced (chained as `adopted`), forget it if
44    /// it produced none, keep it while in flight or unreachable. Returns
45    /// the number of instances adopted.
46    pub async fn settle_unsettled(&self) -> usize {
47        let mut adopted = 0;
48        for t in self.unsettled() {
49            let rows = match self.call(&t.node_id, method::STATUS, None, json!({})).await {
50                Ok(Value::Array(rows)) => rows,
51                _ => continue,
52            };
53            let mine: Vec<&Value> = rows
54                .iter()
55                .filter(|r| r["decision_id"].as_str() == Some(t.decision_id.as_str()))
56                .collect();
57            if mine.iter().any(|r| r["in_flight"] == true) {
58                continue;
59            }
60            for iid in mine.iter().filter_map(|r| r["instance_id"].as_str()) {
61                let rec = PlacementRecord {
62                    instance_id: iid.to_string(),
63                    ..t.clone()
64                };
65                self.chain_event(
66                    chain::EVENT_KIND_WORKLOAD_PLACE,
67                    json!({ "phase": "adopted", "decision_id": t.decision_id, "node": t.node_id,
68                            "variant": t.variant, "instance_id": iid }),
69                );
70                if let Ok(mut p) = self.placements.lock() {
71                    p.insert(iid.to_string(), rec);
72                }
73                adopted += 1;
74            }
75            if let Ok(mut u) = self.unsettled.lock() {
76                u.remove(&t.decision_id);
77            }
78            self.persist();
79        }
80        adopted
81    }
82
83    /// `(settled, note)` for `decision_id` on `node`.
84    pub(super) async fn reconcile(&self, node: &str, decision_id: &str) -> (bool, String) {
85        let rows = match self.call(node, method::STATUS, None, json!({})).await {
86            Ok(Value::Array(rows)) => rows,
87            Ok(_) => return (false, "not reconciled: unexpected status answer".into()),
88            Err(e) => return (false, format!("not reconciled: {e}")),
89        };
90        let mine: Vec<&Value> = rows
91            .iter()
92            .filter(|r| r["decision_id"].as_str() == Some(decision_id))
93            .collect();
94        if mine.iter().any(|r| r["in_flight"] == true) {
95            return (
96                false,
97                "not reconciled: the target is still handling this decision".into(),
98            );
99        }
100        let mut unloaded = 0usize;
101        for iid in mine.iter().filter_map(|r| r["instance_id"].as_str()) {
102            let body = json!({ "instance_id": iid });
103            if let Err(e) = self
104                .call(node, method::UNLOAD, Some(decision_id.to_string()), body)
105                .await
106            {
107                return (
108                    false,
109                    format!("not reconciled: unload of {iid} failed ({e})"),
110                );
111            }
112            unloaded += 1;
113        }
114        (
115            true,
116            format!("reconciled: unloaded {unloaded} instance(s) left by the lost answer"),
117        )
118    }
119}
clawft_kernel/workload_ctl/plane_reconcile.rs

clawft_kernel/workload_ctl/
plane_reconcile.rs
