clawft_kernel/workload_ctl/
host_service.rs

1//! The target side: a node's `workload-host` service (ADR-099 section 7).
2//!
3//! It verifies each signed request, then for `place` / `load`:
4//!
5//! 1. picks the runtime adapter for the chosen route (`native`,
6//!    `container`, ...); no adapter is an admission refusal;
7//! 2. fetches the package before loading, over the same connection, with
8//!    the artifact piece protocol (card 11) and verifies its signatures
9//!    against this node's own trust anchors;
10//! 3. rebuilds the workload from the fetched bytes and hands it to the
11//!    adapter through [`WorkloadHost`], whose admission self-check (binary
12//!    arch, OS, runtime) runs before governance and load (card 09).
13//!
14//! Every refusal is chained as `workload.refuse` (source
15//! [`HOST_CHAIN_SOURCE`]); a placement is chained as `workload.place`.
16//! Unknown `workload.*` methods go to the governance gate, which denies
17//! and chains them (default deny, ADR-099 section 4).
18
19use std::collections::{BTreeMap, HashMap, HashSet};
20use std::net::SocketAddr;
21use std::sync::{Arc, Mutex};
22
23use ed25519_dalek::SigningKey;
24use serde::{Deserialize, Serialize};
25use serde_json::{Value, json};
26
27use crate::chain::{self, ChainManager};
28use crate::cog_ingest::{IngestError, IngestHooks, IngestLease, InstanceBinding};
29use crate::gate::{GateBackend, GateDecision};
30use crate::ipc::GlobalPid;
31use crate::mesh_artifact::ArtifactExchange;
32use crate::mesh_artifact_pkg::PackageExchangeError;
33use crate::mesh_artifact_transfer::PeerSet;
34use crate::mesh_service_adv::ServiceAdvertisement;
35use crate::node_facts_advert::SignedNodeFacts;
36use crate::node_registry::node_id_from_pubkey;
37use crate::workload_pkg::TrustAnchors;
38use crate::workload_pkg::manifest::valid_token;
39use crate::workload_pkg::store::StoreSource;
40use crate::workload_runtime::{
41    HostContract, InstanceHandle, RunEvidence, RunMode, RuntimeError, VerifiedWorkload,
42    WorkloadConfig, WorkloadHost,
43};
44
45use super::cog_kind::route_of;
46use super::host_instances::InFlight;
47use super::msg::{
48    CTL_VERSION, ControllerPolicy, CtlOutcome, CtlRequest, CtlResponse, NonceGuard, Refusal,
49    RefusalCode, SignedCtl, WORKLOAD_HOST_SERVICE, method, verify_request,
50};
51use super::refusal_budget::RefusalBudget;
52
53/// Chain source for target-side placement records.
54pub const HOST_CHAIN_SOURCE: &str = "workload.host";
55
56/// Place-time configuration sent with `place` / `load`.
57#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
58#[serde(deny_unknown_fields)]
59pub struct CtlConfig {
60    /// Run mode (`once`, `interval`, `listener`).
61    pub mode: RunMode,
62    /// Extra cog arguments (validated by the adapter).
63    #[serde(default)]
64    pub args: Vec<String>,
65    /// UDP port for the sensor feed (0.0.0.0:<port>).
66    #[serde(default = "default_csi_port")]
67    pub csi_port: u16,
68}
69
70fn default_csi_port() -> u16 {
71    crate::workload_runtime::host_contract::DEFAULT_CSI_PORT
72}
73
74/// Body of `place` / `load`.
75#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
76#[serde(deny_unknown_fields)]
77pub struct PlaceBody {
78    /// Workload name (cog id).
79    pub name: String,
80    /// BLAKE3 of the signed manifest (fetch key).
81    pub manifest_hash: String,
82    /// Chosen variant (`aarch64-native`).
83    pub variant: String,
84    /// Instance configuration.
85    pub config: CtlConfig,
86    /// Start after loading (place only).
87    #[serde(default)]
88    pub start: bool,
89    /// Project that placed the workload (ingest bridge destination).
90    #[serde(default, skip_serializing_if = "Option::is_none")]
91    pub project_id: Option<String>,
92}
93
94/// Body of the instance verbs.
95#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
96#[serde(deny_unknown_fields)]
97pub struct InstanceBody {
98    /// Instance id (`status` without one lists every instance).
99    #[serde(default)]
100    pub instance_id: Option<String>,
101    /// Stop grace period.
102    #[serde(default)]
103    pub grace_ms: Option<u64>,
104    /// With a listing: also report instances taken down on purpose
105    /// (`departed_instance` rows), which a controller needs to tell a
106    /// teardown from an instance lost with a restart.
107    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
108    pub include_departed: bool,
109}
110
111pub(super) struct Placed {
112    pub(super) handle: InstanceHandle,
113    pub(super) route: String,
114    pub(super) name: String,
115    pub(super) variant: String,
116    pub(super) decision_id: Option<String>,
117    pub(super) last: Option<RunEvidence>,
118    /// The instance's ingest registration (cogs, when ingest is wired).
119    pub(super) ingest: Option<IngestLease>,
120    /// `enabled`, `disabled` (bridge down, no token issued) or `none`.
121    pub(super) ingest_state: &'static str,
122    /// State machine, health misses and restart history (see `host_supervise`).
123    pub(super) life: super::lifecycle::InstanceLife,
124    /// It runs until stopped (listener or interval mode), so an exit is a
125    /// failure, not the end of a one-shot run.
126    pub(super) continuous: bool,
127    /// The operator wants it running: set by `start`, cleared by `stop`.
128    /// A supervisor restarts only what is meant to run.
129    pub(super) desired_running: bool,
130    /// Stopped because the controller lease expired (see `host_supervise`);
131    /// the controller restarts it if it comes back still owning it.
132    pub(super) lease_stopped: bool,
133    /// Who placed it (a controller's node id). The lease applies only to
134    /// instances placed by another node: what this node controls itself
135    /// is exempt.
136    pub(super) requester: String,
137    /// A Cognitum-origin cog in a Seed-bound mesh: the run gate is asked
138    /// again on every start (ADR-106, phase 3).
139    pub(super) licence: Option<super::host_licence::LicensedRun>,
140}
141
142/// Fresh signed facts on demand (a daemon re-probes before the TTL ends).
143pub type FactsSource = Arc<dyn Fn() -> Option<SignedNodeFacts> + Send + Sync>;
144
145/// A node's `workload-host`.
146pub struct WorkloadHostService {
147    pub(super) node_id: String,
148    key: SigningKey,
149    anchors: TrustAnchors,
150    controllers: Box<dyn ControllerPolicy>,
151    nonces: NonceGuard,
152    pub(super) routes: BTreeMap<String, Arc<WorkloadHost>>,
153    exchange: Arc<ArtifactExchange>,
154    gate: Arc<dyn GateBackend>,
155    chain: Option<Arc<ChainManager>>,
156    facts: Mutex<Option<SignedNodeFacts>>,
157    facts_source: Option<FactsSource>,
158    address: Option<String>,
159    pub(super) instances: tokio::sync::Mutex<HashMap<String, Placed>>,
160    /// Decision ids of `place` / `load` requests being handled. A
161    /// controller that lost a response reconciles against this and the
162    /// instance list (see `plane_reconcile`).
163    pub(super) in_flight: Mutex<HashSet<String>>,
164    /// Bounds chain writes for requests that failed verification.
165    verify_budget: RefusalBudget,
166    /// Ingest bridge wiring: a token per placed cog, revoked on stop.
167    pub(super) ingest: Option<IngestHooks>,
168    /// The node's subject revocation list, for the placement race check and
169    /// the forced unload (the gate holds its own handle to the same list).
170    pub(super) revocations: std::sync::OnceLock<Arc<crate::revocation::RevocationList>>,
171    /// Kinds this node accepts for a fetched package and their health
172    /// definitions (see `host_supervise`); builtin by default.
173    pub(super) kinds: crate::workload_kind::KindRegistry,
174    /// Bounds on supervisor restarts.
175    pub(super) restart: super::lifecycle::RestartPolicy,
176    /// Opt-in node-side lease: continuous instances stop once no other
177    /// controller has been heard from for this long (default off).
178    pub(super) lease: Option<std::time::Duration>,
179    /// Instances taken down on purpose (revoked, unloaded), newest last, so
180    /// a controller can tell them from instances lost with a restart (this
181    /// list is in memory: a restart empties it, which is the signal).
182    pub(super) departed: Mutex<std::collections::VecDeque<(String, &'static str)>>,
183    /// Last request from a controller other than this node itself (ms).
184    pub(super) last_contact_ms: std::sync::atomic::AtomicU64,
185    /// The ADR-106 run gate for Cognitum-origin cogs (unset: not consulted).
186    pub(super) licence_gate: std::sync::OnceLock<Arc<dyn crate::licence::CognitumRunGate>>,
187    /// Answers the node-admin methods (`dashboard.*`); unset: they are refused.
188    pub(super) node_admin: std::sync::OnceLock<Arc<dyn super::host_node_admin::NodeAdmin>>,
189}
190
191fn now_ms() -> u64 {
192    chrono::Utc::now().timestamp_millis().max(0) as u64
193}
194
195pub(super) fn refuse(code: RefusalCode, reason: impl Into<String>) -> Refusal {
196    Refusal::new(code, reason)
197}
198
199pub(super) fn runtime_refusal(e: &RuntimeError) -> Refusal {
200    let code = match e {
201        RuntimeError::AdmissionRefused(_) => RefusalCode::Admission,
202        RuntimeError::Governance(_) => RefusalCode::Governance,
203        RuntimeError::UnknownInstance(_) => RefusalCode::UnknownInstance,
204        RuntimeError::InvalidConfig(_) => RefusalCode::InvalidRequest,
205        _ => RefusalCode::Runtime,
206    };
207    Refusal::new(code, e.to_string())
208}
209
210impl WorkloadHostService {
211    /// Host for the node owning `key`. Serves nobody until controllers are
212    /// added, and loads nothing until a route adapter is added.
213    pub fn new(
214        key: SigningKey,
215        exchange: Arc<ArtifactExchange>,
216        anchors: TrustAnchors,
217        gate: Arc<dyn GateBackend>,
218    ) -> Self {
219        Self {
220            node_id: node_id_from_pubkey(&key.verifying_key().to_bytes()),
221            key,
222            anchors,
223            controllers: Box::new(Vec::<[u8; 32]>::new()),
224            nonces: NonceGuard::new(),
225            routes: BTreeMap::new(),
226            exchange,
227            gate,
228            chain: None,
229            facts: Mutex::new(None),
230            facts_source: None,
231            address: None,
232            instances: tokio::sync::Mutex::new(HashMap::new()),
233            in_flight: Mutex::new(HashSet::new()),
234            verify_budget: RefusalBudget::default(),
235            ingest: None,
236            revocations: std::sync::OnceLock::new(),
237            kinds: crate::workload_kind::KindRegistry::builtin(),
238            restart: super::lifecycle::RestartPolicy::default(),
239            lease: None,
240            departed: Mutex::new(std::collections::VecDeque::new()),
241            last_contact_ms: std::sync::atomic::AtomicU64::new(now_ms()),
242            licence_gate: std::sync::OnceLock::new(),
243            node_admin: std::sync::OnceLock::new(),
244        }
245    }
246
247    /// Accept these workload kinds for a package fetched from a peer. The
248    /// default is [`KindRegistry::builtin`]; a node that places a registered
249    /// non-cog kind across nodes must pass the same registry as its control
250    /// plane, or the fetch refuses the kind.
251    ///
252    /// [`KindRegistry::builtin`]: crate::workload_kind::KindRegistry::builtin
253    pub fn with_kinds(mut self, kinds: crate::workload_kind::KindRegistry) -> Self {
254        self.kinds = kinds;
255        self
256    }
257
258    /// The revocation list this node enforces (first call wins). With it a
259    /// `place` that races a revocation is caught and torn down, and
260    /// [`Self::enforce_revocations`] callers share one source of truth.
261    pub fn set_revocations(&self, list: Arc<crate::revocation::RevocationList>) -> bool {
262        self.revocations.set(list).is_ok()
263    }
264
265    /// The list set by [`Self::set_revocations`], if any.
266    pub fn revocations(&self) -> Option<&Arc<crate::revocation::RevocationList>> {
267        self.revocations.get()
268    }
269
270    /// Wire the ingest bridge: placed cogs get a per-instance token,
271    /// bound to the placing project and revoked at stop and unload.
272    pub fn with_ingest(mut self, hooks: IngestHooks) -> Self {
273        self.ingest = Some(hooks);
274        self
275    }
276
277    /// The ingest wiring, if any.
278    pub fn ingest(&self) -> Option<&IngestHooks> {
279        self.ingest.as_ref()
280    }
281
282    /// Replace the bound on chained refusals of unverified requests.
283    pub fn with_refusal_budget(mut self, b: RefusalBudget) -> Self {
284        self.verify_budget = b;
285        self
286    }
287
288    /// Adapter (under governance) for one route kind (`native`, ...).
289    pub fn with_route(mut self, route: &str, host: Arc<WorkloadHost>) -> Self {
290        self.routes.insert(route.to_string(), host);
291        self
292    }
293
294    /// Who may send requests.
295    pub fn with_controllers(mut self, c: impl ControllerPolicy + 'static) -> Self {
296        self.controllers = Box::new(c);
297        self
298    }
299
300    /// Chain for placement and refusal records.
301    pub fn with_chain(mut self, cm: Arc<ChainManager>) -> Self {
302        self.chain = Some(cm);
303        self
304    }
305
306    /// Mesh address advertised for this service.
307    pub fn with_address(mut self, addr: impl Into<String>) -> Self {
308        self.address = Some(addr.into());
309        self
310    }
311
312    /// Replace the signed facts `describe` returns.
313    pub fn set_facts(&self, facts: SignedNodeFacts) {
314        if let Ok(mut f) = self.facts.lock() {
315            *f = Some(facts);
316        }
317    }
318
319    /// Read the signed facts from `source` on every `describe` (falling
320    /// back to [`Self::set_facts`] when it has none).
321    pub fn with_facts_source(mut self, source: FactsSource) -> Self {
322        self.facts_source = Some(source);
323        self
324    }
325
326    fn current_facts(&self) -> Option<SignedNodeFacts> {
327        self.facts_source
328            .as_ref()
329            .and_then(|f| f())
330            .or_else(|| self.facts.lock().ok().and_then(|f| f.clone()))
331    }
332
333    /// This node's id.
334    pub fn node_id(&self) -> &str {
335        &self.node_id
336    }
337
338    /// The `workload-host` advertisement (ADR-099 section 7).
339    pub fn advertisement(&self) -> ServiceAdvertisement {
340        let mut metadata = HashMap::new();
341        let routes: Vec<&str> = self.routes.keys().map(String::as_str).collect();
342        metadata.insert("routes".to_string(), routes.join(","));
343        if let Some(a) = &self.address {
344            metadata.insert("addr".to_string(), a.clone());
345        }
346        metadata.insert(
347            "ingest".to_string(),
348            self.ingest.as_ref().map_or("none", |h| h.state()).to_string(),
349        );
350        ServiceAdvertisement {
351            name: WORKLOAD_HOST_SERVICE.to_string(),
352            methods: method::ALL.iter().map(|m| m.to_string()).collect(),
353            node_id: self.node_id.clone(),
354            global_pid: GlobalPid::new(self.node_id.clone(), 0),
355            version: CTL_VERSION.to_string(),
356            contract_hash: None,
357            metadata,
358            last_updated: chrono::Utc::now().timestamp().max(0) as u64,
359        }
360    }
361
362    /// Remember that `instance_id` was taken down on purpose (`why`:
363    /// `revoked` or `unloaded`); the last 256 are reported by `status`.
364    pub(super) fn note_departed(&self, instance_id: &str, why: &'static str) {
365        if let Ok(mut d) = self.departed.lock() {
366            if d.len() >= 256 {
367                d.pop_front();
368            }
369            d.push_back((instance_id.to_string(), why));
370        }
371    }
372
373    pub(super) fn record(&self, kind: &str, payload: Value) {
374        if let Some(cm) = &self.chain {
375            cm.append(HOST_CHAIN_SOURCE, kind, Some(payload));
376        }
377    }
378
379    fn refused(&self, req: Option<&CtlRequest>, phase: &str, r: &Refusal) {
380        // Unauthenticated refusals are chained within a budget only.
381        let suppressed = match req {
382            Some(_) => 0,
383            None => match self.verify_budget.take() {
384                Some(n) => n,
385                None => return,
386            },
387        };
388        self.record(
389            chain::EVENT_KIND_WORKLOAD_REFUSE,
390            json!({
391                "node": self.node_id, "phase": phase, "code": r.code.as_str(),
392                "reason": r.reason, "method": req.map(|q| q.method.as_str()),
393                "requester": req.map(|q| q.requester.as_str()),
394                "decision_id": req.and_then(|q| q.decision_id.as_deref()),
395                "suppressed": suppressed,
396            }),
397        );
398    }
399
400    /// Handle one signed request (`fetch` is the connection back to the
401    /// controller, used to fetch the payload before loading).
402    pub async fn handle(
403        &self,
404        header_method: &str,
405        signed: &SignedCtl,
406        fetch: Option<&mut PeerSet>,
407    ) -> SignedCtl {
408        self.handle_checked(header_method, signed, fetch).await.0
409    }
410
411    /// [`Self::handle`], also saying whether the request was authenticated
412    /// (a session drops a peer after an unauthenticated request).
413    pub async fn handle_checked(
414        &self,
415        header_method: &str,
416        signed: &SignedCtl,
417        fetch: Option<&mut PeerSet>,
418    ) -> (SignedCtl, bool) {
419        let verified = verify_request(
420            signed,
421            &self.node_id,
422            now_ms(),
423            self.controllers.as_ref(),
424            &self.nonces,
425        )
426        .and_then(|r| {
427            if r.method == header_method {
428                Ok(r)
429            } else {
430                Err(refuse(
431                    RefusalCode::InvalidRequest,
432                    "envelope method differs from signed method",
433                ))
434            }
435        });
436        let authenticated = verified.is_ok();
437        let (nonce, method_name, outcome) = match verified {
438            Err(r) => {
439                self.refused(None, "verify", &r);
440                // Unauthenticated nonce, only to bind the refusal.
441                let claimed: Option<CtlRequest> = serde_json::from_str(&signed.payload).ok();
442                let nonce = claimed.map(|c| c.nonce).unwrap_or_default();
443                (nonce, header_method.to_string(), Err(r))
444            }
445            Ok(req) => {
446                if req.requester != self.node_id {
447                    self.last_contact_ms
448                        .store(now_ms(), std::sync::atomic::Ordering::Relaxed);
449                }
450                let out = self.dispatch(&req, fetch).await;
451                if let Err(r) = &out {
452                    self.refused(Some(&req), "handle", r);
453                }
454                (req.nonce, req.method, out)
455            }
456        };
457        let resp = CtlResponse {
458            version: CTL_VERSION,
459            method: method_name,
460            responder: self.node_id.clone(),
461            request_nonce: nonce,
462            outcome: match outcome {
463                Ok(result) => CtlOutcome::Ok { result },
464                Err(refusal) => CtlOutcome::Refused { refusal },
465            },
466        }
467        .sign(&self.key);
468        (resp, authenticated)
469    }
470
471    async fn dispatch(
472        &self,
473        req: &CtlRequest,
474        fetch: Option<&mut PeerSet>,
475    ) -> Result<Value, Refusal> {
476        // A decision id is the chain hash of the controller's decision.
477        let valid_id = req
478            .decision_id
479            .as_deref()
480            .is_some_and(|d| crate::workload_pkg::codec::is_lower_hex(d, 64));
481        if method::mutates(&req.method) && !valid_id {
482            return Err(refuse(
483                RefusalCode::InvalidRequest,
484                "mutations carry a decision id (64 hex chain hash)",
485            ));
486        }
487        match req.method.as_str() {
488            method::DESCRIBE => {
489                Ok(json!({ "facts": self.current_facts(), "advertisement": self.advertisement() }))
490            }
491            m if method::is_node_admin(m) => self.node_admin_op(req).await,
492            method::PLACE | method::LOAD => self.place(req, fetch).await,
493            method::START | method::STOP | method::UNLOAD | method::STATUS | method::LOGS => {
494                self.instance_op(req).await
495            }
496            m if m.starts_with(crate::workload_governance::policy::WORKLOAD_ACTION_PREFIX) => {
497                // Not in the message set: ask the gate so the refusal is on
498                // its audit trail, and refuse whatever it says.
499                let ctx = json!({ "requester": req.requester, "method": m });
500                let why = match self.gate.check(&req.requester, m, &ctx) {
501                    GateDecision::Deny { reason, .. } => reason,
502                    _ => "not a workload.ctl method".to_string(),
503                };
504                Err(refuse(RefusalCode::UnknownMethod, format!("{m}: {why}")))
505            }
506            m => Err(refuse(
507                RefusalCode::InvalidRequest,
508                format!("{m} is not a workload.ctl method"),
509            )),
510        }
511    }
512
513    async fn place(&self, req: &CtlRequest, fetch: Option<&mut PeerSet>) -> Result<Value, Refusal> {
514        let _in_flight = InFlight::enter(&self.in_flight, req.decision_id.clone());
515        // A revocation that lands after the gate looked at the list but
516        // before the instance is listed would be swept past: remember where
517        // the list stood, and look again once the instance is listed.
518        let list_at_start = self.revocations.get().map(|l| l.generation());
519        let b: PlaceBody = serde_json::from_value(req.body.clone())
520            .map_err(|e| refuse(RefusalCode::InvalidRequest, format!("place body: {e}")))?;
521        if !valid_token(&b.name, 64)
522            || !crate::workload_pkg::codec::is_lower_hex(&b.manifest_hash, 64)
523        {
524            return Err(refuse(
525                RefusalCode::InvalidRequest,
526                "bad name or manifest hash",
527            ));
528        }
529        let route = route_of(&b.variant).to_string();
530        let host = self.routes.get(&route).cloned().ok_or_else(|| {
531            refuse(
532                RefusalCode::Admission,
533                format!("no {route} adapter on this node for {}", b.variant),
534            )
535        })?;
536        let peers = fetch.ok_or_else(|| refuse(RefusalCode::Fetch, "no payload source"))?;
537        let pkg = self
538            .exchange
539            .fetch_package_in(peers, &b.manifest_hash, &self.anchors, &self.kinds)
540            .await
541            .map_err(|e| match e {
542                PackageExchangeError::Verify(v) => refuse(RefusalCode::Verify, v.to_string()),
543                other => refuse(RefusalCode::Fetch, other.to_string()),
544            })?;
545        let w =
546            VerifiedWorkload::from_package(&pkg.verified, &StoreSource::new(self.exchange.store()))
547                .map_err(|e| refuse(RefusalCode::Verify, e.to_string()))?;
548        if w.id != b.name {
549            return Err(refuse(
550                RefusalCode::Verify,
551                "package id differs from the placed name",
552            ));
553        }
554        // ADR-106 run gate: before anything of a Cognitum-origin cog is installed.
555        let licence = self.licence_check_place(&pkg.verified, &w, &b.variant, &host, req).await?;
556        if let Some(p) = &b.project_id
557            && !crate::cog_ingest::valid_project_id(p)
558        {
559            return Err(refuse(
560                RefusalCode::InvalidRequest,
561                "project_id must be a 26-character project id",
562            ));
563        }
564        let mut contract = HostContract::new(SocketAddr::from(([0, 0, 0, 0], b.config.csi_port)));
565        let mut listener = None;
566        let hooks = self.ingest.as_ref().filter(|_| w.kind == "cog");
567        // `none`: no ingest wiring on this node. `disabled`: wired, but the
568        // bridge could not start; the cog runs with no token and no URL.
569        let ingest_state: &'static str = hooks.map_or("none", |h| h.state());
570        // The project check runs whether or not the bridge is up, so a
571        // degraded placement never records an unverified project id.
572        if let Some(hk) = hooks {
573            // Instance id unknown yet: the binding only carries the project
574            // and the controller for the check.
575            let probe = InstanceBinding::new("", b.project_id.clone(), &req.requester);
576            hk.authorize(&probe).map_err(|e| match e {
577                IngestError::Forbidden => refuse(
578                    RefusalCode::Unauthorized,
579                    "this controller may not place cogs for that project",
580                ),
581                IngestError::NotRouted(m) => refuse(RefusalCode::Admission, m),
582                other => refuse(RefusalCode::Runtime, format!("ingest bridge: {other}")),
583            })?;
584        }
585        let hooks = hooks.filter(|h| h.is_enabled());
586        if ingest_state == "disabled" {
587            contract = contract.without_ingest();
588        }
589        if let Some(hk) = hooks {
590            (contract, listener) = hk
591                .prepare(&route, contract)
592                .await
593                .map_err(|e| refuse(RefusalCode::Runtime, format!("ingest bridge: {e}")))?;
594        }
595        let cfg = WorkloadConfig {
596            mode: b.config.mode.clone(),
597            args: b.config.args.clone(),
598            host: contract.clone(),
599            node_id: self.node_id.clone(),
600        };
601        let h = host.load(&w, &cfg).await.map_err(|e| runtime_refusal(&e))?;
602        let iid = h.instance_id.clone();
603        let lease = match hooks {
604            Some(hk) => {
605                let binding = InstanceBinding::new(&iid, b.project_id.clone(), &req.requester);
606                match hk.lease(binding, contract, listener) {
607                    Ok(l) => Some(l),
608                    Err(e) => {
609                        let mut r = refuse(RefusalCode::Runtime, format!("ingest bridge: {e}"));
610                        if let Err(u) = host.unload(h.clone()).await {
611                            r.reason = format!("{}; unload of {iid} failed too ({u})", r.reason);
612                        }
613                        return Err(r);
614                    }
615                }
616            }
617            None => None,
618        };
619        // A node derives instance ids from the workload: a fresh placement
620        // under an id this node reported as departed is no longer departed.
621        if let Ok(mut d) = self.departed.lock() {
622            d.retain(|(i, _)| i != &iid);
623        }
624        let continuous = !matches!(b.config.mode, crate::workload_runtime::RunMode::Once);
625        let will_start = req.method == method::PLACE && b.start;
626        let mut life = super::lifecycle::InstanceLife::new(
627            super::lifecycle::LifecycleState::Loaded,
628            now_ms(),
629        );
630        if will_start {
631            let _ = life.transition(super::lifecycle::LifecycleState::Running, now_ms());
632        }
633        self.instances.lock().await.insert(
634            iid.clone(),
635            Placed {
636                handle: h.clone(),
637                route: route.clone(),
638                name: b.name.clone(),
639                variant: b.variant.clone(),
640                decision_id: req.decision_id.clone(),
641                last: None,
642                ingest: lease,
643                ingest_state,
644                life,
645                continuous,
646                desired_running: will_start,
647                lease_stopped: false,
648                requester: req.requester.clone(),
649                licence,
650            },
651        );
652        if let (Some(list), Some(g0)) = (self.revocations.get(), list_at_start)
653            && list.generation() != g0
654        {
655            self.enforce_revocations(list).await;
656            if !self.instances.lock().await.contains_key(&iid) {
657                return Err(refuse(
658                    RefusalCode::Governance,
659                    format!("revoked while it was being placed: instance {iid} torn down"),
660                ));
661            }
662        }
663        let started = req.method == method::PLACE && b.start;
664        if started && let Err(e) = host.start(&h).await {
665            // A start the gate refused because the package was revoked is
666            // taken down by the revocation itself: the gated unload would
667            // need an unload permit the operator may not have written.
668            let revoked = match (self.revocations.get(), host.workload_for(&h).await) {
669                (Some(l), Some(w)) => {
670                    let (p, k, a) = w.revocation_refs();
671                    l.first_revoked(Some(&p), &k, &a)
672                }
673                _ => None,
674            };
675            return Err(self.roll_back(&host, &h, &e, revoked).await);
676        }
677        let status = host.status(&h).await;
678        self.record(
679            chain::EVENT_KIND_WORKLOAD_PLACE,
680            json!({
681                "phase": "placed", "node": self.node_id, "requester": req.requester,
682                "decision_id": req.decision_id, "workload": b.name, "variant": b.variant,
683                "runtime": h.runtime, "instance_id": iid, "package_id": pkg.package_id,
684                "started": started,
685            }),
686        );
687        Ok(json!({
688            "instance_id": iid, "runtime": h.runtime, "variant": b.variant,
689            "package_id": pkg.package_id, "status": status, "node": self.node_id,
690            "ingest": ingest_state,
691        }))
692    }
693
694    /// A start that failed after load: unload, so the target keeps no
695    /// instance the controller was told was refused. If the unload fails
696    /// too, the instance stays listed (and reconcilable) and the refusal
697    /// says so.
698    async fn roll_back(
699        &self,
700        host: &WorkloadHost,
701        h: &InstanceHandle,
702        e: &RuntimeError,
703        revoked: Option<crate::revocation::RevokedSubject>,
704    ) -> Refusal {
705        let mut r = runtime_refusal(e);
706        let iid = h.instance_id.clone();
707        if let (Some(hk), Some(p)) = (&self.ingest, self.instances.lock().await.get(&iid))
708            && let Some(l) = &p.ingest
709        {
710            hk.deactivate(l);
711        }
712        let unloaded = match &revoked {
713            Some(subject) => {
714                host.revoke_teardown(h, std::time::Duration::from_secs(2), json!(subject))
715                    .await
716            }
717            None => host.unload(h.clone()).await,
718        };
719        match unloaded {
720            Ok(()) => {
721                self.instances.lock().await.remove(&iid);
722                r.reason = format!("start failed: {}; loaded instance {iid} unloaded", r.reason);
723            }
724            Err(u) => {
725                r.reason = format!(
726                    "start failed: {}; unload of {iid} failed too ({u}), it stays listed",
727                    r.reason
728                );
729            }
730        }
731        r
732    }
733}
clawft_kernel/workload_ctl/host_service.rs

clawft_kernel/workload_ctl/
host_service.rs
