appcore_scheduler/
placement.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: placement.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/07/21 23:21:21 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/07/23 23:50:45 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11//! Explainable capability placement for standalone and cluster runtimes.
12
13use appcore_contracts::{
14    CapabilityId, CoreId, CoreProfile, RuntimeHealthStatus, RuntimeMode, RuntimeOperationalMode,
15    ServiceId, WorkloadClass,
16};
17use std::collections::BTreeSet;
18
19/// Minimum resources requested by one placement operation.
20#[derive(Debug, Clone, Default, PartialEq, Eq)]
21pub struct ResourceRequest {
22    /// Minimum logical CPU cores.
23    pub cpu_cores: Option<u16>,
24    /// Minimum memory in bytes.
25    pub memory_bytes: Option<u64>,
26    /// Minimum GPU devices.
27    pub gpu_count: u16,
28}
29
30/// Inputs required to place one capability invocation.
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub struct PlacementRequest {
33    /// Capability required by the operation.
34    pub capability: CapabilityId,
35    /// Independently coordinated service.
36    pub service_id: ServiceId,
37    /// Required standalone or cluster mode.
38    pub runtime_mode: RuntimeMode,
39    /// Whether the operation writes state.
40    pub requires_write: bool,
41    /// Whether service leadership is required.
42    pub requires_leader: bool,
43    /// Preferred workload class.
44    pub workload: WorkloadClass,
45    /// Required scheduling affinity labels.
46    pub affinity: BTreeSet<String>,
47    /// Minimum resources.
48    pub resources: ResourceRequest,
49}
50
51/// Runtime state advertised by one placement candidate.
52#[derive(Debug, Clone, PartialEq, Eq)]
53pub struct PlacementCandidate {
54    /// Candidate Core identity.
55    pub core_id: CoreId,
56    /// Candidate Runtime mode.
57    pub runtime_mode: RuntimeMode,
58    /// Current operational mode.
59    pub operational_mode: RuntimeOperationalMode,
60    /// Current health.
61    pub health: RuntimeHealthStatus,
62    /// Current active-work count.
63    pub current_load: u32,
64    /// Services for which the Core holds leadership.
65    pub leader_services: BTreeSet<ServiceId>,
66    /// Declared Core profile.
67    pub profile: CoreProfile,
68}
69
70/// Stable reason that made a placement candidate ineligible.
71#[derive(Debug, Clone, Copy, PartialEq, Eq)]
72pub enum PlacementRejection {
73    /// Required capability is absent.
74    CapabilityUnavailable,
75    /// Standalone/cluster mode differs.
76    RuntimeModeMismatch,
77    /// Candidate is unhealthy.
78    Unhealthy,
79    /// Operational mode disallows the requested work.
80    OperationalMode,
81    /// Scheduling profile marks the candidate unavailable.
82    Unavailable,
83    /// Required service lease is absent.
84    LeadershipRequired,
85    /// Concurrency capacity is exhausted.
86    AtCapacity,
87    /// Required affinity labels are absent.
88    AffinityMismatch,
89    /// CPU request exceeds the profile.
90    InsufficientCpu,
91    /// Memory request exceeds the profile.
92    InsufficientMemory,
93    /// GPU request exceeds the profile.
94    InsufficientGpu,
95}
96
97/// Explainable evaluation of one candidate.
98#[derive(Debug, Clone, PartialEq, Eq)]
99pub struct PlacementEvaluation {
100    /// Evaluated Core.
101    pub core_id: CoreId,
102    /// Whether all hard constraints passed.
103    pub eligible: bool,
104    /// Deterministic score for an eligible candidate.
105    pub score: i64,
106    /// First hard-constraint failure.
107    pub rejection: Option<PlacementRejection>,
108}
109
110/// Complete placement result, including rejected alternatives.
111#[derive(Debug, Clone, PartialEq, Eq)]
112pub struct PlacementDecision {
113    /// Highest-scoring eligible Core.
114    pub selected_core_id: Option<CoreId>,
115    /// Evaluation of every candidate.
116    pub evaluations: Vec<PlacementEvaluation>,
117}
118
119/// Deterministic scheduler policy that evaluates all foundation signals.
120#[derive(Debug, Clone, Copy, Default)]
121pub struct PlacementEngine;
122
123impl PlacementEngine {
124    /// Evaluates candidates and selects the highest-scoring eligible core.
125    pub fn select(
126        &self,
127        request: &PlacementRequest,
128        candidates: &[PlacementCandidate],
129    ) -> PlacementDecision {
130        let mut evaluations = candidates
131            .iter()
132            .map(|candidate| evaluate_candidate(request, candidate))
133            .collect::<Vec<_>>();
134        evaluations.sort_by(|left, right| left.core_id.cmp(&right.core_id));
135        let selected_core_id = evaluations
136            .iter()
137            .filter(|evaluation| evaluation.eligible)
138            .max_by(|left, right| {
139                left.score
140                    .cmp(&right.score)
141                    .then_with(|| right.core_id.cmp(&left.core_id))
142            })
143            .map(|evaluation| evaluation.core_id.clone());
144        PlacementDecision {
145            selected_core_id,
146            evaluations,
147        }
148    }
149}
150
151fn evaluate_candidate(
152    request: &PlacementRequest,
153    candidate: &PlacementCandidate,
154) -> PlacementEvaluation {
155    let rejection = rejection_reason(request, candidate);
156    if let Some(rejection) = rejection {
157        return PlacementEvaluation {
158            core_id: candidate.core_id.clone(),
159            eligible: false,
160            score: i64::MIN,
161            rejection: Some(rejection),
162        };
163    }
164    let scheduling = candidate.profile.scheduling();
165    let health_score = match candidate.health {
166        RuntimeHealthStatus::Healthy => 2_000,
167        RuntimeHealthStatus::Degraded => 250,
168        RuntimeHealthStatus::Unhealthy => 0,
169    };
170    let workload_score = if scheduling.workload() == request.workload {
171        1_000
172    } else {
173        0
174    };
175    let affinity_score = i64::try_from(request.affinity.len()).unwrap_or(i64::MAX / 100) * 100;
176    let spare_capacity = scheduling
177        .max_concurrency()
178        .saturating_sub(candidate.current_load);
179    let score = i64::from(scheduling.priority()) * 10_000
180        + i64::from(scheduling.weight()) * 100
181        + i64::from(spare_capacity) * 10
182        + health_score
183        + workload_score
184        + affinity_score;
185    PlacementEvaluation {
186        core_id: candidate.core_id.clone(),
187        eligible: true,
188        score,
189        rejection: None,
190    }
191}
192
193fn rejection_reason(
194    request: &PlacementRequest,
195    candidate: &PlacementCandidate,
196) -> Option<PlacementRejection> {
197    if !candidate
198        .profile
199        .capabilities()
200        .contains(&request.capability)
201    {
202        return Some(PlacementRejection::CapabilityUnavailable);
203    }
204    if candidate.runtime_mode != request.runtime_mode {
205        return Some(PlacementRejection::RuntimeModeMismatch);
206    }
207    if candidate.health == RuntimeHealthStatus::Unhealthy {
208        return Some(PlacementRejection::Unhealthy);
209    }
210    if !candidate.operational_mode.allows_local_queries()
211        || (request.requires_write && !candidate.operational_mode.allows_writes())
212    {
213        return Some(PlacementRejection::OperationalMode);
214    }
215    let scheduling = candidate.profile.scheduling();
216    if !scheduling.is_available() {
217        return Some(PlacementRejection::Unavailable);
218    }
219    if request.requires_leader && !candidate.leader_services.contains(&request.service_id) {
220        return Some(PlacementRejection::LeadershipRequired);
221    }
222    if candidate.current_load >= scheduling.max_concurrency() {
223        return Some(PlacementRejection::AtCapacity);
224    }
225    if !request.affinity.is_subset(scheduling.affinity()) {
226        return Some(PlacementRejection::AffinityMismatch);
227    }
228    let resources = candidate.profile.resources();
229    if request.resources.cpu_cores.is_some_and(|required| {
230        resources
231            .cpu_cores()
232            .is_none_or(|available| available < required)
233    }) {
234        return Some(PlacementRejection::InsufficientCpu);
235    }
236    if request.resources.memory_bytes.is_some_and(|required| {
237        resources
238            .memory_bytes()
239            .is_none_or(|available| available < required)
240    }) {
241        return Some(PlacementRejection::InsufficientMemory);
242    }
243    if resources.gpu_count() < request.resources.gpu_count {
244        return Some(PlacementRejection::InsufficientGpu);
245    }
246    None
247}
248
249#[cfg(test)]
250mod tests {
251    use super::*;
252    use appcore_contracts::{
253        CoreRole, LeadershipMode, LeadershipRequirement, ResourceProfile, SchedulingProfile,
254    };
255
256    fn request() -> PlacementRequest {
257        PlacementRequest {
258            capability: CapabilityId::new("document.extract").unwrap(),
259            service_id: ServiceId::new("document.extract").unwrap(),
260            runtime_mode: RuntimeMode::Cluster,
261            requires_write: true,
262            requires_leader: true,
263            workload: WorkloadClass::Compute,
264            affinity: BTreeSet::from(["region.local".to_string()]),
265            resources: ResourceRequest {
266                cpu_cores: Some(4),
267                memory_bytes: Some(8_000),
268                gpu_count: 1,
269            },
270        }
271    }
272
273    fn candidate(core: &str, weight: u16, load: u32) -> PlacementCandidate {
274        let service = ServiceId::new("document.extract").unwrap();
275        let scheduling = SchedulingProfile::new(weight, 1, 8, WorkloadClass::Compute)
276            .unwrap()
277            .with_affinity("region.local")
278            .unwrap();
279        let profile = CoreProfile::new(
280            CoreRole::Compute,
281            service.clone(),
282            [CapabilityId::new("document.extract").unwrap()],
283            LeadershipRequirement::new(service.clone(), LeadershipMode::Required, 30_000).unwrap(),
284            ResourceProfile::new(Some(8), Some(16_000), 1),
285            scheduling,
286        )
287        .unwrap();
288        PlacementCandidate {
289            core_id: CoreId::new(core).unwrap(),
290            runtime_mode: RuntimeMode::Cluster,
291            operational_mode: RuntimeOperationalMode::ReadWrite,
292            health: RuntimeHealthStatus::Healthy,
293            current_load: load,
294            leader_services: BTreeSet::from([service]),
295            profile,
296        }
297    }
298
299    #[test]
300    fn placement_uses_weight_and_current_load() {
301        let decision = PlacementEngine.select(
302            &request(),
303            &[candidate("core-a", 10, 7), candidate("core-b", 20, 1)],
304        );
305        assert_eq!(decision.selected_core_id.unwrap().as_str(), "core-b");
306    }
307
308    #[test]
309    fn placement_rejects_missing_leadership() {
310        let mut candidate = candidate("core-a", 10, 0);
311        candidate.leader_services.clear();
312        let decision = PlacementEngine.select(&request(), &[candidate]);
313        assert!(decision.selected_core_id.is_none());
314        assert_eq!(
315            decision.evaluations[0].rejection,
316            Some(PlacementRejection::LeadershipRequired)
317        );
318    }
319
320    #[test]
321    fn placement_rejects_resource_and_affinity_mismatches() {
322        let mut resources = candidate("core-a", 10, 0);
323        let mut affinity = candidate("core-b", 10, 0);
324        resources.profile = CoreProfile::new(
325            CoreRole::Compute,
326            ServiceId::new("document.extract").unwrap(),
327            [CapabilityId::new("document.extract").unwrap()],
328            LeadershipRequirement::new(
329                ServiceId::new("document.extract").unwrap(),
330                LeadershipMode::Required,
331                30_000,
332            )
333            .unwrap(),
334            ResourceProfile::new(Some(2), Some(16_000), 1),
335            SchedulingProfile::new(10, 1, 8, WorkloadClass::Compute)
336                .unwrap()
337                .with_affinity("region.local")
338                .unwrap(),
339        )
340        .unwrap();
341        affinity.profile = CoreProfile::new(
342            CoreRole::Compute,
343            ServiceId::new("document.extract").unwrap(),
344            [CapabilityId::new("document.extract").unwrap()],
345            LeadershipRequirement::new(
346                ServiceId::new("document.extract").unwrap(),
347                LeadershipMode::Required,
348                30_000,
349            )
350            .unwrap(),
351            ResourceProfile::new(Some(8), Some(16_000), 1),
352            SchedulingProfile::new(10, 1, 8, WorkloadClass::Compute).unwrap(),
353        )
354        .unwrap();
355        let decision = PlacementEngine.select(&request(), &[resources, affinity]);
356        assert_eq!(
357            decision.evaluations[0].rejection,
358            Some(PlacementRejection::InsufficientCpu)
359        );
360        assert_eq!(
361            decision.evaluations[1].rejection,
362            Some(PlacementRejection::AffinityMismatch)
363        );
364    }
365}
appcore_scheduler/placement.rs

appcore_scheduler/
placement.rs
