ferrum_interfaces/vnext/resource/
contracts.rs

1use serde::{Deserialize, Serialize};
2use std::collections::BTreeSet;
3use std::fmt;
4
5use super::{
6    AllocationKind, AllocationLifetime, BufferDescriptor, BufferUsage, CapacityDomainId,
7    DeviceDescriptor, DeviceId, ElementType, FailureDomain, FailureEnvelope, NodeId, PlanHash,
8    PlanId, RequestIdentity, ResourceAllocation, ResourceId, RunId, TransactionId, VNextError,
9};
10
11pub const MAX_RESOURCE_TRANSITION_RECEIPT_WIRE_BYTES: usize = 4 * 1024 * 1024;
12pub const MAX_RESOURCE_LEASE_RECEIPT_WIRE_BYTES: usize = 4 * 1024 * 1024;
13pub(super) const SEQUENCE_DISPATCH_POISONED_BIT: u64 = 1 << 63;
14
15#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
16#[serde(rename_all = "snake_case")]
17pub enum StateInitialization {
18    /// Core does not initialize the backing. The first consumer must fully
19    /// define every byte before it is read.
20    None,
21    /// Core zeroes each exact Sequence backing acquisition before its first
22    /// state consumer in the same ordered submission. Request-scope zeroing is
23    /// reserved until initialization dependencies can defer sibling sequences.
24    Zero,
25}
26
27pub(super) fn invalid_resource(reason: impl Into<String>) -> VNextError {
28    VNextError::InvalidExecutionPlan {
29        reason: reason.into(),
30    }
31}
32
33pub(super) fn core_resource_failure(
34    code: &'static str,
35    message: impl Into<String>,
36    retryable: bool,
37) -> FailureEnvelope {
38    FailureEnvelope::new(FailureDomain::Resource, code, message, retryable)
39        .expect("core-generated resource failure must be valid")
40}
41
42pub(crate) fn validate_runtime_descriptor_for_admission(
43    descriptor: &DeviceDescriptor,
44    admission: &StaticProvisioningBinding,
45    context: &'static str,
46) -> Result<(), VNextError> {
47    descriptor.validate()?;
48    if &descriptor.id != admission.device_id()
49        || descriptor.runtime_implementation_fingerprint
50            != admission.device_runtime_implementation_fingerprint()
51        || descriptor.total_memory_bytes != admission.device_capacity_bytes()
52    {
53        return Err(invalid_resource(format!(
54            "{context} runtime device, runtime implementation, or capacity differs from admission"
55        )));
56    }
57    Ok(())
58}
59
60#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
61pub struct ResourceTransactionIdentity {
62    pub(super) pool_id: ResourcePoolId,
63    pub(super) run_id: RunId,
64    pub(super) transaction_id: TransactionId,
65    pub(super) request_id: RequestIdentity,
66}
67
68impl ResourceTransactionIdentity {
69    pub fn for_admission(
70        admission: &StaticProvisioningBinding,
71        run_id: RunId,
72        transaction_id: TransactionId,
73    ) -> Self {
74        Self {
75            pool_id: admission.pool_id(),
76            run_id,
77            transaction_id,
78            request_id: admission.request_id().clone(),
79        }
80    }
81
82    pub const fn pool_id(&self) -> ResourcePoolId {
83        self.pool_id
84    }
85
86    pub fn run_id(&self) -> &RunId {
87        &self.run_id
88    }
89
90    pub fn transaction_id(&self) -> &TransactionId {
91        &self.transaction_id
92    }
93
94    pub fn request_id(&self) -> &RequestIdentity {
95        &self.request_id
96    }
97}
98
99#[derive(Debug, Clone, PartialEq, Eq)]
100pub struct ResourceDriverFailure {
101    failure: FailureEnvelope,
102}
103
104impl ResourceDriverFailure {
105    pub fn new(failure: FailureEnvelope) -> Result<Self, VNextError> {
106        failure.validate()?;
107        if failure.domain() != FailureDomain::Resource {
108            return Err(invalid_resource(
109                "resource driver failure must use the resource failure domain",
110            ));
111        }
112        Ok(Self { failure })
113    }
114
115    pub fn failure(&self) -> &FailureEnvelope {
116        &self.failure
117    }
118
119    pub fn into_failure(self) -> FailureEnvelope {
120        self.failure
121    }
122}
123
124#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
125#[serde(rename_all = "snake_case")]
126pub enum ResourceTransactionState {
127    New,
128    Reserved,
129    Committed,
130    RolledBack,
131    Released,
132    Quarantined,
133}
134
135impl ResourceTransactionState {
136    pub fn transition(
137        self,
138        resource_id: &ResourceId,
139        action: ResourceTransactionAction,
140    ) -> Result<Self, VNextError> {
141        expected_transition(action, self).ok_or_else(|| VNextError::InvalidResourceTransition {
142            resource_id: resource_id.to_string(),
143            from: self.as_str(),
144            action: action.as_str(),
145        })
146    }
147
148    pub const fn as_str(self) -> &'static str {
149        match self {
150            Self::New => "new",
151            Self::Reserved => "reserved",
152            Self::Committed => "committed",
153            Self::RolledBack => "rolled_back",
154            Self::Released => "released",
155            Self::Quarantined => "quarantined",
156        }
157    }
158
159    pub(super) const fn is_live(self) -> bool {
160        matches!(self, Self::New | Self::Reserved | Self::Committed)
161    }
162}
163
164#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
165#[serde(rename_all = "snake_case")]
166pub enum ResourceTransactionAction {
167    Reserve,
168    Commit,
169    Rollback,
170    Release,
171    Quarantine,
172}
173
174impl ResourceTransactionAction {
175    pub const fn as_str(self) -> &'static str {
176        match self {
177            Self::Reserve => "reserve",
178            Self::Commit => "commit",
179            Self::Rollback => "rollback",
180            Self::Release => "release",
181            Self::Quarantine => "quarantine",
182        }
183    }
184}
185
186pub(super) const fn expected_transition(
187    action: ResourceTransactionAction,
188    before: ResourceTransactionState,
189) -> Option<ResourceTransactionState> {
190    match (before, action) {
191        (ResourceTransactionState::New, ResourceTransactionAction::Reserve) => {
192            Some(ResourceTransactionState::Reserved)
193        }
194        (ResourceTransactionState::Reserved, ResourceTransactionAction::Commit) => {
195            Some(ResourceTransactionState::Committed)
196        }
197        (ResourceTransactionState::Reserved, ResourceTransactionAction::Rollback) => {
198            Some(ResourceTransactionState::RolledBack)
199        }
200        (ResourceTransactionState::Committed, ResourceTransactionAction::Release) => {
201            Some(ResourceTransactionState::Released)
202        }
203        (
204            ResourceTransactionState::New
205            | ResourceTransactionState::Reserved
206            | ResourceTransactionState::Committed,
207            ResourceTransactionAction::Quarantine,
208        ) => Some(ResourceTransactionState::Quarantined),
209        _ => None,
210    }
211}
212
213#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
214#[serde(rename_all = "snake_case")]
215pub enum ResourceCompensationAction {
216    UndoReserve,
217    UndoCommit,
218}
219
220impl ResourceCompensationAction {
221    pub(super) const fn for_prepare_action(action: ResourceTransactionAction) -> Option<Self> {
222        match action {
223            ResourceTransactionAction::Reserve => Some(Self::UndoReserve),
224            ResourceTransactionAction::Commit => Some(Self::UndoCommit),
225            _ => None,
226        }
227    }
228}
229
230#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
231#[serde(rename_all = "snake_case")]
232pub enum ResourceRecoveryStrategy {
233    ReverseCompensation,
234    ForwardCompletion,
235    ReconcileOrQuarantine,
236}
237
238/// Core-owned retention policy derived from `AllocationLifetime`. A backend or
239/// scheduler may decide when to act on it, but may not rewrite it after
240/// admission.
241#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
242#[serde(rename_all = "snake_case")]
243pub enum ResourceRetentionPolicy {
244    Plan,
245    Request,
246    Sequence,
247    Step,
248    Invocation,
249}
250
251impl From<AllocationLifetime> for ResourceRetentionPolicy {
252    fn from(lifetime: AllocationLifetime) -> Self {
253        match lifetime {
254            AllocationLifetime::Plan => Self::Plan,
255            AllocationLifetime::Request => Self::Request,
256            AllocationLifetime::Sequence => Self::Sequence,
257            AllocationLifetime::Step => Self::Step,
258            AllocationLifetime::Invocation => Self::Invocation,
259        }
260    }
261}
262
263#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
264#[serde(rename_all = "snake_case")]
265pub enum ResourceRetentionDecision {
266    Retain,
267    ReturnRequested,
268}
269
270/// Process-local identity of one provisioned resource pool. It is independent
271/// from both the request that provisioned the pool and requests that later use
272/// one of its active-sequence slots.
273#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
274#[serde(try_from = "u64", into = "u64")]
275pub struct ResourcePoolId(u64);
276
277impl ResourcePoolId {
278    pub(super) fn issue(generation: u64) -> Result<Self, VNextError> {
279        Self::try_from(generation)
280    }
281
282    pub const fn get(self) -> u64 {
283        self.0
284    }
285}
286
287impl TryFrom<u64> for ResourcePoolId {
288    type Error = VNextError;
289
290    fn try_from(value: u64) -> Result<Self, Self::Error> {
291        if value == 0 {
292            return Err(invalid_resource("resource pool id must be non-zero"));
293        }
294        Ok(Self(value))
295    }
296}
297
298impl From<ResourcePoolId> for u64 {
299    fn from(value: ResourcePoolId) -> Self {
300        value.0
301    }
302}
303
304impl fmt::Display for ResourcePoolId {
305    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
306        write!(formatter, "resource-pool:{}", self.0)
307    }
308}
309
310#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
311pub struct ResourcePoolIdentity {
312    pub(super) pool_id: ResourcePoolId,
313    pub(super) plan_id: PlanId,
314    pub(super) plan_hash: PlanHash,
315    pub(super) device_id: DeviceId,
316    pub(super) device_runtime_implementation_fingerprint: String,
317    pub(super) admission_generation: u64,
318}
319
320impl ResourcePoolIdentity {
321    pub const fn pool_id(&self) -> ResourcePoolId {
322        self.pool_id
323    }
324
325    pub fn plan_id(&self) -> &PlanId {
326        &self.plan_id
327    }
328
329    pub fn plan_hash(&self) -> &PlanHash {
330        &self.plan_hash
331    }
332
333    pub fn device_id(&self) -> &DeviceId {
334        &self.device_id
335    }
336
337    pub fn device_runtime_implementation_fingerprint(&self) -> &str {
338        &self.device_runtime_implementation_fingerprint
339    }
340
341    pub const fn admission_generation(&self) -> u64 {
342        self.admission_generation
343    }
344}
345
346/// Immutable identity and capacity envelope signed into an admission permit.
347/// This is trusted output and intentionally cannot be deserialized directly.
348#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
349pub struct StaticProvisioningBinding {
350    pub(super) pool_identity: ResourcePoolIdentity,
351    pub(super) plan_id: PlanId,
352    pub(super) plan_hash: PlanHash,
353    pub(super) request_id: RequestIdentity,
354    pub(super) device_id: DeviceId,
355    pub(super) device_runtime_implementation_fingerprint: String,
356    pub(super) device_capacity_bytes: u64,
357    pub(super) usable_capacity_bytes: u64,
358    pub(super) plan_static_bytes: u64,
359    pub(super) admitted_bytes: u64,
360    pub(super) maximum_active_sequences: u32,
361    pub(super) admission_generation: u64,
362}
363
364impl StaticProvisioningBinding {
365    pub fn pool_identity(&self) -> &ResourcePoolIdentity {
366        &self.pool_identity
367    }
368
369    pub const fn pool_id(&self) -> ResourcePoolId {
370        self.pool_identity.pool_id
371    }
372    pub fn plan_id(&self) -> &PlanId {
373        &self.plan_id
374    }
375
376    pub fn plan_hash(&self) -> &PlanHash {
377        &self.plan_hash
378    }
379
380    pub fn request_id(&self) -> &RequestIdentity {
381        &self.request_id
382    }
383
384    pub fn device_id(&self) -> &DeviceId {
385        &self.device_id
386    }
387
388    pub fn device_runtime_implementation_fingerprint(&self) -> &str {
389        &self.device_runtime_implementation_fingerprint
390    }
391
392    pub const fn device_capacity_bytes(&self) -> u64 {
393        self.device_capacity_bytes
394    }
395
396    pub const fn usable_capacity_bytes(&self) -> u64 {
397        self.usable_capacity_bytes
398    }
399
400    pub const fn plan_static_bytes(&self) -> u64 {
401        self.plan_static_bytes
402    }
403
404    pub const fn admitted_bytes(&self) -> u64 {
405        self.admitted_bytes
406    }
407
408    pub const fn maximum_active_sequences(&self) -> u32 {
409        self.maximum_active_sequences
410    }
411
412    pub const fn admission_generation(&self) -> u64 {
413        self.admission_generation
414    }
415}
416
417#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
418pub struct ResourceReservation {
419    pub(super) resource_id: ResourceId,
420    pub(super) request_id: RequestIdentity,
421    pub(super) owner_node_id: Option<NodeId>,
422    pub(super) size_bytes: u64,
423    pub(super) alignment_bytes: u64,
424    pub(super) usage: BufferUsage,
425    pub(super) element_type: ElementType,
426    pub(super) retention_policy: ResourceRetentionPolicy,
427    pub(super) backing_domain_id: Option<CapacityDomainId>,
428    pub(super) generation: u64,
429}
430
431impl ResourceReservation {
432    fn from_allocation(
433        allocation: &ResourceAllocation,
434        request_id: &RequestIdentity,
435        generation: u64,
436    ) -> Result<Self, VNextError> {
437        let owner_node_id = match allocation.kind() {
438            AllocationKind::Value | AllocationKind::InitializationScratch => None,
439            AllocationKind::Scratch { node_id, .. }
440            | AllocationKind::Binding { node_id, .. }
441            | AllocationKind::Persistent { node_id, .. } => Some(node_id.clone()),
442        };
443        if allocation.size_bytes() == 0
444            || allocation.alignment_bytes() == 0
445            || !allocation.alignment_bytes().is_power_of_two()
446            || generation == 0
447        {
448            return Err(invalid_resource(format!(
449                "allocation `{}` cannot be admitted",
450                allocation.resource_id()
451            )));
452        }
453        Ok(Self {
454            resource_id: allocation.resource_id().clone(),
455            request_id: request_id.clone(),
456            owner_node_id,
457            size_bytes: allocation.size_bytes(),
458            alignment_bytes: allocation.alignment_bytes(),
459            usage: allocation.usage(),
460            element_type: allocation.element_type(),
461            retention_policy: allocation.lifetime().into(),
462            backing_domain_id: None,
463            generation,
464        })
465    }
466
467    pub fn resource_id(&self) -> &ResourceId {
468        &self.resource_id
469    }
470
471    pub fn request_id(&self) -> &RequestIdentity {
472        &self.request_id
473    }
474
475    pub fn owner_node_id(&self) -> Option<&NodeId> {
476        self.owner_node_id.as_ref()
477    }
478
479    pub const fn size_bytes(&self) -> u64 {
480        self.size_bytes
481    }
482
483    pub const fn alignment_bytes(&self) -> u64 {
484        self.alignment_bytes
485    }
486
487    pub const fn usage(&self) -> BufferUsage {
488        self.usage
489    }
490
491    pub const fn element_type(&self) -> ElementType {
492        self.element_type
493    }
494
495    pub const fn retention_policy(&self) -> ResourceRetentionPolicy {
496        self.retention_policy
497    }
498
499    pub const fn backing_domain_id(&self) -> Option<CapacityDomainId> {
500        self.backing_domain_id
501    }
502
503    pub const fn generation(&self) -> u64 {
504        self.generation
505    }
506
507    pub(super) fn matches_descriptor(&self, descriptor: &BufferDescriptor) -> bool {
508        descriptor.resource_id == self.resource_id
509            && descriptor.size_bytes == self.size_bytes
510            && descriptor.alignment_bytes == self.alignment_bytes
511            && descriptor.usage == self.usage
512            && descriptor.element_type == self.element_type
513    }
514}
515
516#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
517pub struct ResourceReservationBatch {
518    request_id: RequestIdentity,
519    pub(super) reservations: Vec<ResourceReservation>,
520    plan_static_size_bytes: u64,
521    total_size_bytes: u64,
522}
523
524impl ResourceReservationBatch {
525    pub(super) fn from_allocations(
526        request_id: &RequestIdentity,
527        allocations: &[ResourceAllocation],
528        generation: u64,
529    ) -> Result<Self, VNextError> {
530        let mut ids = BTreeSet::new();
531        let mut reservations = Vec::with_capacity(allocations.len());
532        let mut plan_static_size_bytes = 0_u64;
533        for allocation in allocations {
534            if !ids.insert(allocation.resource_id().clone()) {
535                return Err(invalid_resource(format!(
536                    "resource `{}` is duplicated in admission",
537                    allocation.resource_id()
538                )));
539            }
540            plan_static_size_bytes = plan_static_size_bytes
541                .checked_add(allocation.size_bytes())
542                .ok_or_else(|| invalid_resource("admitted resource bytes overflow u64"))?;
543            reservations.push(ResourceReservation::from_allocation(
544                allocation, request_id, generation,
545            )?);
546        }
547        let total_size_bytes = plan_static_size_bytes;
548        Ok(Self {
549            request_id: request_id.clone(),
550            reservations,
551            plan_static_size_bytes,
552            total_size_bytes,
553        })
554    }
555
556    pub fn request_id(&self) -> &RequestIdentity {
557        &self.request_id
558    }
559
560    pub fn reservations(&self) -> &[ResourceReservation] {
561        &self.reservations
562    }
563
564    pub fn resource_ids(&self) -> impl Iterator<Item = &ResourceId> {
565        self.reservations
566            .iter()
567            .map(ResourceReservation::resource_id)
568    }
569
570    pub const fn total_size_bytes(&self) -> u64 {
571        self.total_size_bytes
572    }
573
574    pub const fn plan_static_size_bytes(&self) -> u64 {
575        self.plan_static_size_bytes
576    }
577}
ferrum_interfaces/vnext/resource/contracts.rs

ferrum_interfaces/vnext/resource/
contracts.rs
