ferrum_interfaces/vnext/
error.rs

1use thiserror::Error;
2
3use serde::{Deserialize, Serialize};
4use std::fmt;
5
6use crate::model_executor::PlanRuntimeResourceSnapshot;
7
8use super::execution::DynamicBackingPoolId;
9
10/// Maximum encoded size accepted by the untrusted failure-envelope decoder.
11pub const MAX_FAILURE_ENVELOPE_WIRE_BYTES: usize = 8 * 1024;
12
13#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
14#[serde(rename_all = "snake_case")]
15pub enum FailureDomain {
16    Device,
17    Operation,
18    Resource,
19    Planning,
20    ModelResolution,
21    Product,
22    Event,
23}
24
25/// Portable failure payload. Execution identity is carried by the surrounding
26/// event or resource receipt rather than flattened into this message.
27#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
28pub struct FailureEnvelope {
29    domain: FailureDomain,
30    code: String,
31    message: String,
32    retryable: bool,
33    #[serde(default, skip_serializing_if = "Option::is_none")]
34    resource_snapshot: Option<PlanRuntimeResourceSnapshot>,
35}
36
37#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
38pub struct UnvalidatedFailureEnvelope {
39    domain: FailureDomain,
40    code: String,
41    message: String,
42    retryable: bool,
43    #[serde(default, skip_serializing_if = "Option::is_none")]
44    resource_snapshot: Option<PlanRuntimeResourceSnapshot>,
45}
46
47#[derive(Deserialize)]
48#[serde(deny_unknown_fields)]
49pub(crate) struct FailureEnvelopeWire {
50    domain: FailureDomain,
51    code: String,
52    message: String,
53    retryable: bool,
54    #[serde(default)]
55    resource_snapshot: Option<PlanRuntimeResourceSnapshot>,
56}
57
58impl From<FailureEnvelopeWire> for UnvalidatedFailureEnvelope {
59    fn from(wire: FailureEnvelopeWire) -> Self {
60        Self {
61            domain: wire.domain,
62            code: wire.code,
63            message: wire.message,
64            retryable: wire.retryable,
65            resource_snapshot: wire.resource_snapshot,
66        }
67    }
68}
69
70impl UnvalidatedFailureEnvelope {
71    pub fn revalidate(self, expected_domain: FailureDomain) -> Result<FailureEnvelope, VNextError> {
72        if self.domain != expected_domain {
73            return Err(VNextError::InvalidExecutionPlan {
74                reason: format!(
75                    "failure domain `{:?}` differs from expected `{:?}`",
76                    self.domain, expected_domain
77                ),
78            });
79        }
80        let resource_snapshot = self.resource_snapshot;
81        let envelope = FailureEnvelope::new(self.domain, self.code, self.message, self.retryable)?;
82        match resource_snapshot {
83            Some(snapshot) => envelope.with_resource_snapshot(snapshot),
84            None => Ok(envelope),
85        }
86    }
87}
88
89impl FailureEnvelope {
90    pub fn new(
91        domain: FailureDomain,
92        code: impl Into<String>,
93        message: impl Into<String>,
94        retryable: bool,
95    ) -> Result<Self, VNextError> {
96        let envelope = Self {
97            domain,
98            code: code.into(),
99            message: message.into(),
100            retryable,
101            resource_snapshot: None,
102        };
103        envelope.validate()?;
104        Ok(envelope)
105    }
106
107    pub fn validate(&self) -> Result<(), VNextError> {
108        if self.code.is_empty()
109            || self.code.len() > 64
110            || !self
111                .code
112                .bytes()
113                .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-'))
114            || self.message.trim().is_empty()
115            || self.message.len() > 4096
116            || self
117                .message
118                .bytes()
119                .any(|byte| byte.is_ascii_control() && !matches!(byte, b'\n' | b'\t'))
120        {
121            return Err(VNextError::InvalidExecutionPlan {
122                reason: "failure code or message is empty, oversized, or non-portable".to_owned(),
123            });
124        }
125        if let Some(snapshot) = &self.resource_snapshot {
126            if self.domain != FailureDomain::Resource {
127                return Err(VNextError::InvalidExecutionPlan {
128                    reason:
129                        "only resource-domain failures may carry a plan runtime resource snapshot"
130                            .to_owned(),
131                });
132            }
133            snapshot
134                .validate()
135                .map_err(|error| VNextError::InvalidExecutionPlan {
136                    reason: format!("invalid plan runtime resource snapshot: {error}"),
137                })?;
138        }
139        Ok(())
140    }
141
142    pub fn with_resource_snapshot(
143        mut self,
144        snapshot: PlanRuntimeResourceSnapshot,
145    ) -> Result<Self, VNextError> {
146        self.resource_snapshot = Some(snapshot);
147        self.validate()?;
148        Ok(self)
149    }
150
151    pub const fn domain(&self) -> FailureDomain {
152        self.domain
153    }
154
155    pub fn code(&self) -> &str {
156        &self.code
157    }
158
159    pub fn message(&self) -> &str {
160        &self.message
161    }
162
163    pub const fn retryable(&self) -> bool {
164        self.retryable
165    }
166
167    pub const fn resource_snapshot(&self) -> Option<&PlanRuntimeResourceSnapshot> {
168        self.resource_snapshot.as_ref()
169    }
170
171    pub fn decode_untrusted(bytes: &[u8]) -> Result<UnvalidatedFailureEnvelope, VNextError> {
172        if bytes.len() > MAX_FAILURE_ENVELOPE_WIRE_BYTES {
173            return Err(VNextError::Serialization {
174                context: "decode untrusted failure envelope",
175                message: format!(
176                    "payload has {} bytes; maximum is {MAX_FAILURE_ENVELOPE_WIRE_BYTES}",
177                    bytes.len()
178                ),
179            });
180        }
181        serde_json::from_slice::<FailureEnvelopeWire>(bytes)
182            .map(Into::into)
183            .map_err(|error| VNextError::Serialization {
184                context: "decode untrusted failure envelope",
185                message: error.to_string(),
186            })
187    }
188}
189
190#[cfg(test)]
191mod failure_envelope_tests {
192    use super::{FailureDomain, FailureEnvelope};
193    use crate::model_executor::PlanRuntimeResourceSnapshot;
194
195    fn snapshot() -> PlanRuntimeResourceSnapshot {
196        PlanRuntimeResourceSnapshot::new(1_000, 900, 700, 700, 400, 300, 200, 0, 0).unwrap()
197    }
198
199    #[test]
200    fn resource_snapshot_round_trips_only_after_revalidation() {
201        let envelope = FailureEnvelope::new(
202            FailureDomain::Resource,
203            "diagnostic_resource_failure",
204            "injected resource failure",
205            false,
206        )
207        .unwrap()
208        .with_resource_snapshot(snapshot())
209        .unwrap();
210        let encoded = serde_json::to_vec(&envelope).unwrap();
211
212        let decoded = FailureEnvelope::decode_untrusted(&encoded)
213            .unwrap()
214            .revalidate(FailureDomain::Resource)
215            .unwrap();
216
217        assert_eq!(decoded, envelope);
218        assert_eq!(
219            decoded
220                .resource_snapshot()
221                .unwrap()
222                .available_bytes()
223                .unwrap(),
224            400
225        );
226    }
227
228    #[test]
229    fn non_resource_failure_rejects_resource_snapshot() {
230        let envelope = FailureEnvelope::new(
231            FailureDomain::Operation,
232            "operation_failure",
233            "operation failure",
234            false,
235        )
236        .unwrap();
237
238        assert!(envelope.with_resource_snapshot(snapshot()).is_err());
239    }
240
241    #[test]
242    fn untrusted_resource_snapshot_fails_closed_when_capacity_is_incoherent() {
243        let wire = serde_json::json!({
244            "domain": "resource",
245            "code": "forged_resource_failure",
246            "message": "forged resource failure",
247            "retryable": false,
248            "resource_snapshot": {
249                "device_capacity_bytes": 1_000,
250                "usable_capacity_bytes": 1_001,
251                "process_claimed_bytes": 700,
252                "plan_claimed_bytes": 700,
253                "static_bytes": 400,
254                "dynamic_resident_bytes": 300,
255                "dynamic_free_bytes": 200,
256                "pending_growth_bytes": 0,
257                "quarantined_bytes": 0
258            }
259        });
260
261        let unvalidated = FailureEnvelope::decode_untrusted(&serde_json::to_vec(&wire).unwrap())
262            .expect("wire shape should decode before trust validation");
263
264        assert!(unvalidated.revalidate(FailureDomain::Resource).is_err());
265    }
266}
267
268#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
269#[serde(rename_all = "snake_case")]
270pub enum DynamicAdmissionFaultKind {
271    InvalidContract,
272    UnknownDomain,
273    ForeignCoordinator,
274    Poisoned,
275    EpochExhausted,
276    EpochRegression,
277    AuthorityExhausted,
278    ArithmeticOverflow,
279    AllocationFailure,
280}
281
282/// Capacity boundary that rejected one otherwise valid dynamic growth.
283#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
284#[serde(rename_all = "snake_case")]
285pub enum DeviceCapacityPressureScope {
286    PlanBudget,
287    ProcessWide,
288}
289
290/// Exact device-wide pressure observed while trying to grow dynamic backing.
291///
292/// This is retry evidence, not an allocation authority. Plan-budget pressure
293/// can be paired with plan-local release epochs; process-wide pressure requires
294/// device-wide coordination. Contract and allocator failures remain terminal.
295#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
296pub struct DeviceCapacityPressure {
297    scope: DeviceCapacityPressureScope,
298    device_id: String,
299    requested_bytes: u64,
300    plan_claimed_bytes: u64,
301    plan_usable_bytes: u64,
302    process_claimed_bytes: u64,
303    process_usable_bytes: u64,
304}
305
306impl DeviceCapacityPressure {
307    pub fn new(
308        scope: DeviceCapacityPressureScope,
309        device_id: String,
310        requested_bytes: u64,
311        plan_claimed_bytes: u64,
312        plan_usable_bytes: u64,
313        process_claimed_bytes: u64,
314        process_usable_bytes: u64,
315    ) -> Result<Self, VNextError> {
316        let pressure = Self {
317            scope,
318            device_id,
319            requested_bytes,
320            plan_claimed_bytes,
321            plan_usable_bytes,
322            process_claimed_bytes,
323            process_usable_bytes,
324        };
325        let plan_available = pressure
326            .plan_usable_bytes
327            .checked_sub(pressure.plan_claimed_bytes);
328        let process_available = pressure
329            .process_usable_bytes
330            .checked_sub(pressure.process_claimed_bytes);
331        let scope_matches = match pressure.scope {
332            DeviceCapacityPressureScope::PlanBudget => {
333                plan_available.is_some_and(|available| available < pressure.requested_bytes)
334            }
335            DeviceCapacityPressureScope::ProcessWide => {
336                plan_available.is_some_and(|available| available >= pressure.requested_bytes)
337                    && process_available
338                        .is_some_and(|available| available < pressure.requested_bytes)
339            }
340        };
341        if pressure.device_id.trim().is_empty()
342            || pressure.requested_bytes == 0
343            || process_available.is_none()
344            || !scope_matches
345        {
346            return Err(VNextError::InvalidExecutionPlan {
347                reason: "device capacity pressure evidence is inconsistent".to_owned(),
348            });
349        }
350        Ok(pressure)
351    }
352
353    pub fn scope(&self) -> &DeviceCapacityPressureScope {
354        &self.scope
355    }
356
357    pub fn device_id(&self) -> &str {
358        &self.device_id
359    }
360
361    pub const fn requested_bytes(&self) -> u64 {
362        self.requested_bytes
363    }
364
365    pub const fn plan_claimed_bytes(&self) -> u64 {
366        self.plan_claimed_bytes
367    }
368
369    pub const fn plan_usable_bytes(&self) -> u64 {
370        self.plan_usable_bytes
371    }
372
373    pub const fn process_claimed_bytes(&self) -> u64 {
374        self.process_claimed_bytes
375    }
376
377    pub const fn process_usable_bytes(&self) -> u64 {
378        self.process_usable_bytes
379    }
380
381    pub const fn available_bytes(&self) -> u64 {
382        let plan_available = self
383            .plan_usable_bytes
384            .saturating_sub(self.plan_claimed_bytes);
385        let process_available = self
386            .process_usable_bytes
387            .saturating_sub(self.process_claimed_bytes);
388        if plan_available < process_available {
389            plan_available
390        } else {
391            process_available
392        }
393    }
394}
395
396impl fmt::Display for DeviceCapacityPressure {
397    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
398        write!(
399            formatter,
400            "device `{}` capacity is temporarily unavailable: requested {}, plan claimed {}/{}, process claimed {}/{}",
401            self.device_id,
402            self.requested_bytes,
403            self.plan_claimed_bytes,
404            self.plan_usable_bytes,
405            self.process_claimed_bytes,
406            self.process_usable_bytes
407        )
408    }
409}
410
411/// Exact pool-local resident ceiling observed while maintaining otherwise
412/// valid deferred backing.
413///
414/// Unlike device capacity pressure, this does not authorize a larger pool. It
415/// proves that existing resident owners must be released or a reusable cache
416/// entry must be evicted before the deferred transaction can be retried.
417#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
418pub struct DynamicPoolResidentPressure {
419    pool_id: DynamicBackingPoolId,
420    requested_bytes: u64,
421    resident_bytes: u64,
422    maximum_resident_bytes: u64,
423}
424
425impl DynamicPoolResidentPressure {
426    pub fn new(
427        pool_id: DynamicBackingPoolId,
428        requested_bytes: u64,
429        resident_bytes: u64,
430        maximum_resident_bytes: u64,
431    ) -> Result<Self, VNextError> {
432        let pressure = Self {
433            pool_id,
434            requested_bytes,
435            resident_bytes,
436            maximum_resident_bytes,
437        };
438        if pressure.requested_bytes == 0
439            || pressure.resident_bytes > pressure.maximum_resident_bytes
440            || pressure.available_bytes() >= pressure.requested_bytes
441        {
442            return Err(VNextError::InvalidExecutionPlan {
443                reason: "dynamic pool resident pressure evidence is inconsistent".to_owned(),
444            });
445        }
446        Ok(pressure)
447    }
448
449    pub fn pool_id(&self) -> &DynamicBackingPoolId {
450        &self.pool_id
451    }
452
453    pub const fn requested_bytes(&self) -> u64 {
454        self.requested_bytes
455    }
456
457    pub const fn resident_bytes(&self) -> u64 {
458        self.resident_bytes
459    }
460
461    pub const fn maximum_resident_bytes(&self) -> u64 {
462        self.maximum_resident_bytes
463    }
464
465    pub const fn available_bytes(&self) -> u64 {
466        self.maximum_resident_bytes
467            .saturating_sub(self.resident_bytes)
468    }
469}
470
471impl fmt::Display for DynamicPoolResidentPressure {
472    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
473        write!(
474            formatter,
475            "dynamic pool `{}` resident capacity is temporarily unavailable: requested {}, resident {}/{}",
476            self.pool_id.as_str(),
477            self.requested_bytes,
478            self.resident_bytes,
479            self.maximum_resident_bytes
480        )
481    }
482}
483
484/// Recoverable physical pressure returned by deferred backing maintenance.
485#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
486#[serde(tag = "kind", content = "evidence", rename_all = "snake_case")]
487pub enum DynamicBackingPressure {
488    DeviceCapacity(DeviceCapacityPressure),
489    PoolResident(DynamicPoolResidentPressure),
490}
491
492impl DynamicBackingPressure {
493    pub const fn requested_bytes(&self) -> u64 {
494        match self {
495            Self::DeviceCapacity(pressure) => pressure.requested_bytes(),
496            Self::PoolResident(pressure) => pressure.requested_bytes(),
497        }
498    }
499
500    pub const fn available_bytes(&self) -> u64 {
501        match self {
502            Self::DeviceCapacity(pressure) => pressure.available_bytes(),
503            Self::PoolResident(pressure) => pressure.available_bytes(),
504        }
505    }
506
507    pub const fn device_capacity(&self) -> Option<&DeviceCapacityPressure> {
508        match self {
509            Self::DeviceCapacity(pressure) => Some(pressure),
510            Self::PoolResident(_) => None,
511        }
512    }
513
514    pub const fn pool_resident(&self) -> Option<&DynamicPoolResidentPressure> {
515        match self {
516            Self::DeviceCapacity(_) => None,
517            Self::PoolResident(pressure) => Some(pressure),
518        }
519    }
520}
521
522impl From<DeviceCapacityPressure> for DynamicBackingPressure {
523    fn from(pressure: DeviceCapacityPressure) -> Self {
524        Self::DeviceCapacity(pressure)
525    }
526}
527
528impl From<DynamicPoolResidentPressure> for DynamicBackingPressure {
529    fn from(pressure: DynamicPoolResidentPressure) -> Self {
530        Self::PoolResident(pressure)
531    }
532}
533
534/// Structured, fail-closed errors produced by the vNext contracts.
535#[derive(Debug, Clone, PartialEq, Eq, Error)]
536pub enum VNextError {
537    #[error("invalid {kind} identity `{value}`: {reason}")]
538    InvalidIdentity {
539        kind: &'static str,
540        value: String,
541        reason: &'static str,
542    },
543    #[error("unknown model family `{family_id}`")]
544    UnknownModelFamily { family_id: String },
545    #[error("unknown external model metadata `{metadata_id}`")]
546    UnknownExternalModelMetadata { metadata_id: String },
547    #[error(
548        "ambiguous model family registration for {identity_kind} `{identity}`: {matches} matches"
549    )]
550    AmbiguousModelFamilyRegistration {
551        identity_kind: &'static str,
552        identity: String,
553        matches: usize,
554    },
555    #[error("invalid model config for `{family_id}` at `{field}`: {reason}")]
556    InvalidModelConfig {
557        family_id: String,
558        field: String,
559        reason: String,
560    },
561    #[error("unknown weight layout `{layout_id}` for model family `{family_id}`")]
562    UnknownWeightLayout {
563        family_id: String,
564        layout_id: String,
565    },
566    #[error(
567        "operation `{operation_id}` requires version {required_major}.{required_minor}; provider offers {available_major}.{available_minor}"
568    )]
569    IncompatibleOperationVersion {
570        node_id: Option<String>,
571        operation_id: String,
572        required_major: u16,
573        required_minor: u16,
574        available_major: u16,
575        available_minor: u16,
576    },
577    #[error("no provider for operation `{operation_id}` on device `{device_id}`: {reason}")]
578    UnsupportedOperation {
579        node_id: Option<String>,
580        operation_id: String,
581        device_id: String,
582        reason: String,
583    },
584    #[error("invalid execution plan: {reason}")]
585    InvalidExecutionPlan { reason: String },
586    #[error("dynamic admission {kind:?}: {reason}")]
587    DynamicAdmissionContract {
588        kind: DynamicAdmissionFaultKind,
589        reason: String,
590    },
591    #[error("{0}")]
592    DeviceCapacityUnavailable(DeviceCapacityPressure),
593    #[error("{0}")]
594    DynamicPoolResidentUnavailable(DynamicPoolResidentPressure),
595    #[error(
596        "dynamic resource admission is not connected: {descriptor_count} descriptors require at least {minimum_sequence_bytes} bytes for one runnable sequence"
597    )]
598    DynamicResourceAdmissionRequired {
599        descriptor_count: usize,
600        minimum_sequence_bytes: u64,
601    },
602    #[error(
603        "unsupported execution plan schema {actual_major}.{actual_minor}; expected {expected_major}.{expected_minor}"
604    )]
605    UnsupportedPlanSchema {
606        expected_major: u16,
607        expected_minor: u16,
608        actual_major: u16,
609        actual_minor: u16,
610    },
611    #[error("execution plan hash mismatch: expected `{expected}`, actual `{actual}`")]
612    PlanHashMismatch { expected: String, actual: String },
613    #[error("invalid resource transition for `{resource_id}`: {from} + {action}")]
614    InvalidResourceTransition {
615        resource_id: String,
616        from: &'static str,
617        action: &'static str,
618    },
619    #[error("invalid resource lease transition for `{lease_id}`: {from} + {action}")]
620    InvalidLeaseTransition {
621        lease_id: String,
622        from: &'static str,
623        action: &'static str,
624    },
625    #[error("invalid resolved model plan at `{field}`: {reason}")]
626    InvalidResolvedModelPlan { field: String, reason: String },
627    #[error("failed to {context}: {message}")]
628    Serialization {
629        context: &'static str,
630        message: String,
631    },
632}
ferrum_interfaces/vnext/error.rs

ferrum_interfaces/vnext/
error.rs
