Skip to main content

ferrum_interfaces/vnext/resource/
dynamic_pool.rs

1use super::{
2    invalid_resource, AllocationKind, AllocationLifetime, Arc, AtomicU64, AtomicU8, BTreeMap,
3    BackingChunkIdentity, BackingSegment, BufferDescriptor, BufferUsage, CapacityDomainId,
4    CapacityEntry, CapacityEpochs, CapacityUnits, CapacityVector, CapacityWaitCondition,
5    DeviceBufferRetention, DeviceCapacityAvailabilitySnapshot, DeviceCapacityGrant, DeviceRuntime,
6    DynamicBackingPoolId, DynamicBackingPoolSpec, DynamicResourceDescriptor, DynamicResourceShape,
7    DynamicStorageAllocator, DynamicStorageProfile, ElementType, ExecutionLaneId, FreeExtentIndex,
8    InvocationLivenessMode, LogicalAdmissionCoordinator, LogicalAdmissionCoordinatorId, Mutex,
9    Ordering, PlanNode, ResourceId, Serialize, StateInitialization, VNextError,
10};
11use crate::vnext::{
12    DeviceCapacityPressure, DeviceReusableAddressScope, DynamicPoolProvisioningPolicy,
13    DynamicResourceDemand, PoolCompatibilityKey, ReusableExecutionBucketId,
14    ReusableExecutionMemoryPlan,
15};
16use sha2::{Digest, Sha256};
17
18pub(super) static NEXT_DYNAMIC_POOL_INSTANCE_ID: AtomicU64 = AtomicU64::new(1);
19
20pub(super) fn align_up_resource(value: u64, alignment: u64) -> Result<u64, VNextError> {
21    if alignment == 0 || !alignment.is_power_of_two() {
22        return Err(invalid_resource(
23            "dynamic pool alignment is not a non-zero power of two",
24        ));
25    }
26    value
27        .checked_add(alignment - 1)
28        .map(|rounded| rounded & !(alignment - 1))
29        .ok_or_else(|| invalid_resource("dynamic pool aligned bytes overflow u64"))
30}
31
32pub(super) fn free_extent_layout_fingerprint(allocator: &FreeExtentIndex) -> String {
33    let mut hasher = Sha256::new();
34    for (&(chunk_ordinal, offset_bytes), extent) in &allocator.by_offset {
35        hasher.update(chunk_ordinal.to_be_bytes());
36        hasher.update(extent.chunk_generation.to_be_bytes());
37        hasher.update(offset_bytes.to_be_bytes());
38        hasher.update(extent.length_bytes.to_be_bytes());
39    }
40    format!("sha256/{:x}", hasher.finalize())
41}
42
43fn unused_simulation_chunk_ordinal(allocator: &FreeExtentIndex) -> Result<u32, VNextError> {
44    let mut candidate = u32::MAX;
45    loop {
46        if !allocator
47            .by_offset
48            .keys()
49            .any(|(chunk_ordinal, _)| *chunk_ordinal == candidate)
50        {
51            return Ok(candidate);
52        }
53        candidate = candidate.checked_sub(1).ok_or_else(|| {
54            invalid_resource("contiguous packing simulation exhausted chunk identities")
55        })?;
56    }
57}
58
59/// Returns one additional chunk size that makes the canonical
60/// best-fit-decreasing transaction packable. This only runs after physical
61/// deferral; successful hot-path claims still touch the allocator once.
62pub(super) fn contiguous_packing_growth_bytes(
63    allocator: &FreeExtentIndex,
64    pool_id: &DynamicBackingPoolId,
65    claim_bytes_descending: &[u64],
66) -> Result<u64, VNextError> {
67    if claim_bytes_descending.is_empty()
68        || claim_bytes_descending.iter().any(|bytes| *bytes == 0)
69        || claim_bytes_descending
70            .windows(2)
71            .any(|pair| pair[0] < pair[1])
72    {
73        return Err(invalid_resource(
74            "contiguous packing demand is empty, zero-sized, or non-canonical",
75        ));
76    }
77    let maximum_growth = claim_bytes_descending
78        .iter()
79        .try_fold(0_u64, |total, bytes| total.checked_add(*bytes))
80        .ok_or_else(|| invalid_resource("contiguous packing demand overflows u64"))?;
81    let synthetic_chunk = unused_simulation_chunk_ordinal(allocator)?;
82    let mut growth_bytes = 0_u64;
83    loop {
84        let mut simulation = allocator.clone();
85        if growth_bytes != 0 {
86            simulation.insert_extent(synthetic_chunk, u64::MAX, 0, growth_bytes)?;
87        }
88        let mut failed_claim = None;
89        for &claim_bytes in claim_bytes_descending {
90            if simulation
91                .allocate_contiguous(pool_id, claim_bytes)?
92                .is_none()
93            {
94                failed_claim = Some(claim_bytes);
95                break;
96            }
97        }
98        let Some(failed_claim) = failed_claim else {
99            return Ok(growth_bytes);
100        };
101        growth_bytes = growth_bytes
102            .checked_add(failed_claim)
103            .ok_or_else(|| invalid_resource("contiguous packing growth overflows u64"))?;
104        if growth_bytes > maximum_growth {
105            return Err(invalid_resource(
106                "contiguous packing planner exceeded its guaranteed growth bound",
107            ));
108        }
109    }
110}
111
112#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
113pub(super) struct DynamicPoolDomainSpec {
114    pub(super) domain_id: CapacityDomainId,
115    pub(super) pool: DynamicBackingPoolSpec,
116    pub(super) descriptors: Vec<DynamicResourceDescriptor>,
117}
118
119impl DynamicPoolDomainSpec {
120    pub const fn domain_id(&self) -> CapacityDomainId {
121        self.domain_id
122    }
123
124    pub(super) fn pool_id(&self) -> &DynamicBackingPoolId {
125        self.pool.pool_id()
126    }
127}
128
129#[derive(Debug)]
130pub(super) struct SubmissionWaveProjectionLayout {
131    pub(super) descriptor_index: usize,
132    pub(super) projection_index: usize,
133}
134
135#[derive(Debug)]
136pub(super) struct SubmissionWaveRowLayout {
137    pub(super) projections: Vec<SubmissionWaveProjectionLayout>,
138}
139
140#[derive(Debug)]
141pub(super) struct SubmissionWaveDomainLayout {
142    pub(super) rows: Vec<SubmissionWaveRowLayout>,
143    pub(super) claim_identity: PhysicalBackingClaimIdentity,
144    pub(super) projection_count: usize,
145}
146
147#[derive(Debug)]
148pub(super) struct SubmissionWaveProjectionCapacity {
149    pub(super) physical_offset_bytes: u64,
150    pub(super) capacity_size_bytes: u64,
151}
152
153#[derive(Debug)]
154pub(super) struct SubmissionWaveDomainCapacityLayout {
155    pub(super) physical_size_bytes: u64,
156    pub(super) projections: Vec<SubmissionWaveProjectionCapacity>,
157}
158
159pub(super) fn compile_submission_wave_domain_layout(
160    domain: &DynamicPoolDomainSpec,
161    nodes: &[PlanNode],
162) -> Result<Option<SubmissionWaveDomainLayout>, VNextError> {
163    if domain.pool.invocation_liveness_mode() == InvocationLivenessMode::NoInvocationResources {
164        return Ok(None);
165    }
166
167    let canonical_projections = domain
168        .descriptors
169        .iter()
170        .enumerate()
171        .filter(|(_, descriptor)| descriptor.lifetime() == AllocationLifetime::Invocation)
172        .collect::<Vec<_>>();
173    let projection_by_resource = canonical_projections
174        .iter()
175        .enumerate()
176        .map(|(projection_index, (descriptor_index, descriptor))| {
177            (
178                descriptor.base_resource_id(),
179                (*descriptor_index, projection_index),
180            )
181        })
182        .collect::<BTreeMap<_, _>>();
183    if canonical_projections.is_empty()
184        || projection_by_resource.len() != canonical_projections.len()
185    {
186        return Err(invalid_resource(
187            "submission wave layout requires unique invocation descriptors",
188        ));
189    }
190
191    let liveness = domain.pool.invocation_liveness();
192    let mut covered_projections = std::collections::BTreeSet::new();
193    let mut rows = Vec::with_capacity(liveness.len());
194    for node in nodes {
195        let Ok(row_index) = liveness.binary_search_by(|row| row.node_id().cmp(node.id())) else {
196            continue;
197        };
198        let row = &liveness[row_index];
199        let projections = row
200            .resource_ids()
201            .iter()
202            .map(|resource_id| {
203                let &(descriptor_index, projection_index) =
204                    projection_by_resource.get(resource_id).ok_or_else(|| {
205                        invalid_resource(
206                            "submission wave liveness references a descriptor outside its pool",
207                        )
208                    })?;
209                if !covered_projections.insert(projection_index) {
210                    return Err(invalid_resource(
211                        "submission wave liveness repeats one invocation descriptor",
212                    ));
213                }
214                Ok(SubmissionWaveProjectionLayout {
215                    descriptor_index,
216                    projection_index,
217                })
218            })
219            .collect::<Result<Vec<_>, VNextError>>()?;
220        rows.push(SubmissionWaveRowLayout { projections });
221    }
222    if rows.len() != liveness.len()
223        || covered_projections.len() != canonical_projections.len()
224        || covered_projections
225            .iter()
226            .copied()
227            .ne(0..canonical_projections.len())
228    {
229        return Err(invalid_resource(
230            "submission wave layout does not cover immutable plan invocation resources exactly",
231        ));
232    }
233
234    Ok(Some(SubmissionWaveDomainLayout {
235        rows,
236        claim_identity: PhysicalBackingClaimIdentity::new(
237            domain.pool_id().clone(),
238            canonical_projections
239                .iter()
240                .map(|(_, descriptor)| descriptor.base_resource_id().clone())
241                .collect(),
242        )?,
243        projection_count: canonical_projections.len(),
244    }))
245}
246
247pub(super) fn compile_submission_wave_reusable_capacity_layouts(
248    domains: &[DynamicPoolDomainSpec],
249    layouts: &[Option<SubmissionWaveDomainLayout>],
250    reusable_execution: Option<&ReusableExecutionMemoryPlan>,
251) -> Result<
252    BTreeMap<ReusableExecutionBucketId, Vec<Option<SubmissionWaveDomainCapacityLayout>>>,
253    VNextError,
254> {
255    let Some(reusable_execution) = reusable_execution else {
256        return Ok(BTreeMap::new());
257    };
258    if domains.len() != layouts.len() {
259        return Err(invalid_resource(
260            "submission wave reusable capacity layout count differs from dynamic pool domains",
261        ));
262    }
263
264    reusable_execution
265        .buckets()
266        .iter()
267        .map(|resolved| {
268            let bucket = resolved.bucket();
269            let capacity = bucket.capacity();
270            let capacity_shape = DynamicResourceShape::from_validated(
271                capacity.maximum_sequences(),
272                capacity.maximum_tokens(),
273                capacity.maximum_pages(),
274            );
275            let compiled = domains
276                .iter()
277                .zip(layouts)
278                .map(|(domain, layout)| {
279                    let Some(layout) = layout else {
280                        return Ok(None);
281                    };
282                    let mode = domain.pool.invocation_liveness_mode();
283                    let mut projections = (0..layout.projection_count)
284                        .map(|_| None)
285                        .collect::<Vec<_>>();
286                    let mut physical_size_bytes = 0_u64;
287                    for row in &layout.rows {
288                        let row_base = match mode {
289                            InvocationLivenessMode::TotalOrderReuse => 0,
290                            InvocationLivenessMode::ConservativeConcurrent => physical_size_bytes,
291                            InvocationLivenessMode::NoInvocationResources => unreachable!(),
292                        };
293                        let mut row_bytes = 0_u64;
294                        for projection_layout in &row.projections {
295                            let descriptor = domain
296                                .descriptors
297                                .get(projection_layout.descriptor_index)
298                                .ok_or_else(|| {
299                                    invalid_resource(
300                                        "reusable submission layout references a descriptor outside its pool",
301                                    )
302                                })?;
303                            if descriptor.lifetime() != AllocationLifetime::Invocation {
304                                return Err(invalid_resource(
305                                    "reusable submission layout references a non-Invocation descriptor",
306                                ));
307                            }
308                            let capacity_size_bytes =
309                                descriptor.evaluate_request_bytes_for_shape(capacity_shape)?;
310                            let physical_offset_bytes =
311                                row_base.checked_add(row_bytes).ok_or_else(|| {
312                                    invalid_resource(
313                                        "reusable submission projection offset overflows u64",
314                                    )
315                                })?;
316                            row_bytes =
317                                row_bytes.checked_add(capacity_size_bytes).ok_or_else(|| {
318                                    invalid_resource(
319                                        "reusable submission row capacity overflows u64",
320                                    )
321                                })?;
322                            if projections[projection_layout.projection_index]
323                                .replace(SubmissionWaveProjectionCapacity {
324                                    physical_offset_bytes,
325                                    capacity_size_bytes,
326                                })
327                                .is_some()
328                            {
329                                return Err(invalid_resource(
330                                    "reusable submission layout repeats one canonical projection",
331                                ));
332                            }
333                        }
334                        physical_size_bytes = match mode {
335                            InvocationLivenessMode::TotalOrderReuse => {
336                                physical_size_bytes.max(row_bytes)
337                            }
338                            InvocationLivenessMode::ConservativeConcurrent => physical_size_bytes
339                                .checked_add(row_bytes)
340                                .ok_or_else(|| {
341                                    invalid_resource(
342                                        "reusable submission pool capacity overflows u64",
343                                    )
344                                })?,
345                            InvocationLivenessMode::NoInvocationResources => unreachable!(),
346                        };
347                    }
348                    let projections =
349                        projections
350                            .into_iter()
351                            .collect::<Option<Vec<_>>>()
352                            .ok_or_else(|| {
353                                invalid_resource(
354                                    "reusable submission layout left a projection uncompiled",
355                                )
356                            })?;
357                    if projections.is_empty() || physical_size_bytes == 0 {
358                        return Err(invalid_resource(
359                            "reusable submission layout compiled empty capacity",
360                        ));
361                    }
362                    Ok(Some(SubmissionWaveDomainCapacityLayout {
363                        physical_size_bytes,
364                        projections,
365                    }))
366                })
367                .collect::<Result<Vec<_>, VNextError>>()?;
368            Ok((bucket.bucket_id().clone(), compiled))
369        })
370        .collect()
371}
372
373pub(super) struct ResidentChunkBacking<B> {
374    // Buffer must drop before its physical capacity grant is returned.
375    pub(super) buffer: B,
376    pub(super) _grant: DeviceCapacityGrant,
377    pub(super) identity: BackingChunkIdentity,
378    pub(super) descriptor: BufferDescriptor,
379}
380
381pub(super) struct ResidentChunkState<B> {
382    pub(super) backing: Arc<ResidentChunkBacking<B>>,
383    pub(super) live_segments: u64,
384}
385
386#[derive(Debug, Clone, Copy, PartialEq, Eq)]
387pub(super) struct DynamicBackingClaimOccupancy {
388    pub(super) scope: DynamicBackingClaimScope,
389    pub(super) residency: DynamicBackingClaimResidency,
390    pub(super) physical_bytes: u64,
391    pub(super) segment_count: u64,
392}
393
394pub(super) fn rollback_free_extent_journal<B>(
395    states: &mut [std::sync::MutexGuard<'_, DynamicBackingPoolState<B>>],
396    journals: &[Vec<Vec<BackingSegment>>],
397) -> Result<(), VNextError> {
398    for group_index in (0..journals.len()).rev() {
399        for segments in journals[group_index].iter().rev() {
400            for segment in segments.iter().rev() {
401                if let Err(error) = states[group_index].allocator.release(segment) {
402                    states[group_index].poisoned = true;
403                    return Err(invalid_resource(format!(
404                        "dynamic backing rollback failed and poisoned its pool: {error}"
405                    )));
406                }
407            }
408        }
409    }
410    Ok(())
411}
412
413pub(super) struct DynamicBackingPoolState<B> {
414    pub(super) resident_bytes: u64,
415    pub(super) pending_growth_bytes: u64,
416    pub(super) next_chunk_ordinal: u32,
417    pub(super) next_chunk_generation: u64,
418    pub(super) chunks: BTreeMap<u32, ResidentChunkState<B>>,
419    pub(super) allocator: FreeExtentIndex,
420    pub(super) live_occupancy: DynamicPoolLiveOccupancyStatus,
421    pub(super) quarantined: Vec<QuarantinedDynamicChunk<B>>,
422    pub(super) poisoned: bool,
423}
424
425#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
426#[serde(rename_all = "snake_case")]
427pub enum DynamicChunkQuarantineReason {
428    DescriptorMismatch,
429    PublicationRejected,
430}
431
432pub(super) struct QuarantinedDynamicChunk<B> {
433    pub(super) backing: Arc<ResidentChunkBacking<B>>,
434    pub(super) reason: DynamicChunkQuarantineReason,
435}
436
437pub(super) struct DynamicBackingPool<R>
438where
439    R: DeviceRuntime,
440{
441    pub(super) instance_id: u64,
442    pub(super) domain: DynamicPoolDomainSpec,
443    pub(super) logical_admission: LogicalAdmissionCoordinator,
444    pub(super) maintenance: Mutex<()>,
445    pub(super) next_extent_generation: AtomicU64,
446    pub(super) state: Mutex<DynamicBackingPoolState<R::Buffer>>,
447}
448
449pub(super) struct PendingGrowthGuard<R>
450where
451    R: DeviceRuntime,
452{
453    pub(super) pool: Arc<DynamicBackingPool<R>>,
454    pub(super) bytes: u64,
455    pub(super) armed: bool,
456}
457
458impl<R> PendingGrowthGuard<R>
459where
460    R: DeviceRuntime,
461{
462    pub(super) fn disarm(&mut self) {
463        self.armed = false;
464    }
465}
466
467impl<R> Drop for PendingGrowthGuard<R>
468where
469    R: DeviceRuntime,
470{
471    fn drop(&mut self) {
472        if self.armed {
473            self.pool.cancel_pending_growth(self.bytes);
474        }
475    }
476}
477
478pub(super) trait BackingExtentOwner: Send + Sync {
479    fn instance_id(&self) -> u64;
480    fn release_segments(
481        &self,
482        claim_identity: &PhysicalBackingClaimIdentity,
483        occupancy: DynamicBackingClaimOccupancy,
484        segments: &[BackingSegment],
485    );
486}
487
488pub(super) struct BackingSegmentLease {
489    pub(super) owner: Arc<dyn BackingExtentOwner>,
490    pub(super) owner_instance_id: u64,
491    pub(super) claim_identity: PhysicalBackingClaimIdentity,
492    pub(super) segment_generation: u64,
493    pub(super) occupancy: DynamicBackingClaimOccupancy,
494    pub(super) segments: Vec<BackingSegment>,
495    pub(super) size_bytes: u64,
496    pub(super) initialization: Option<Arc<BackingInitializationCell>>,
497    pub(super) released: bool,
498}
499
500#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
501#[serde(rename_all = "snake_case")]
502pub enum BackingInitializationStatus {
503    Pending,
504    Prepared,
505    InFlight,
506    Initialized,
507    Poisoned,
508}
509
510const BACKING_INITIALIZATION_PENDING: u8 = 0;
511const BACKING_INITIALIZATION_PREPARED: u8 = 1;
512const BACKING_INITIALIZATION_IN_FLIGHT: u8 = 2;
513const BACKING_INITIALIZATION_INITIALIZED: u8 = 3;
514const BACKING_INITIALIZATION_POISONED: u8 = 4;
515
516#[derive(Debug)]
517enum BackingInitializationState {
518    Pending,
519    Prepared { wave_fingerprint: String },
520    InFlight { wave_fingerprint: String },
521    Initialized,
522    Poisoned,
523}
524
525#[derive(Debug)]
526pub(super) struct BackingInitializationCell {
527    target_fingerprint: String,
528    status: AtomicU8,
529    state: Mutex<BackingInitializationState>,
530}
531
532impl BackingInitializationCell {
533    fn new(target_fingerprint: String) -> Self {
534        Self {
535            target_fingerprint,
536            status: AtomicU8::new(BACKING_INITIALIZATION_PENDING),
537            state: Mutex::new(BackingInitializationState::Pending),
538        }
539    }
540
541    pub(super) fn target_fingerprint(&self) -> &str {
542        &self.target_fingerprint
543    }
544
545    pub(super) fn status(&self) -> Result<BackingInitializationStatus, VNextError> {
546        match self.status.load(Ordering::Acquire) {
547            BACKING_INITIALIZATION_PENDING => Ok(BackingInitializationStatus::Pending),
548            BACKING_INITIALIZATION_PREPARED => Ok(BackingInitializationStatus::Prepared),
549            BACKING_INITIALIZATION_IN_FLIGHT => Ok(BackingInitializationStatus::InFlight),
550            BACKING_INITIALIZATION_INITIALIZED => Ok(BackingInitializationStatus::Initialized),
551            BACKING_INITIALIZATION_POISONED => Ok(BackingInitializationStatus::Poisoned),
552            _ => Err(invalid_resource(
553                "backing initialization status contains an invalid value",
554            )),
555        }
556    }
557
558    pub(super) fn prepare(&self, wave_fingerprint: &str) -> Result<bool, VNextError> {
559        let mut state = match self.state.lock() {
560            Ok(state) => state,
561            Err(poisoned) => {
562                let mut state = poisoned.into_inner();
563                *state = BackingInitializationState::Poisoned;
564                self.status
565                    .store(BACKING_INITIALIZATION_POISONED, Ordering::Release);
566                return Err(invalid_resource("backing initialization state is poisoned"));
567            }
568        };
569        match &*state {
570            BackingInitializationState::Pending => {
571                *state = BackingInitializationState::Prepared {
572                    wave_fingerprint: wave_fingerprint.to_owned(),
573                };
574                self.status
575                    .store(BACKING_INITIALIZATION_PREPARED, Ordering::Release);
576                Ok(true)
577            }
578            BackingInitializationState::Initialized => Ok(false),
579            BackingInitializationState::Prepared {
580                wave_fingerprint: current,
581            } if current == wave_fingerprint => Ok(true),
582            BackingInitializationState::Prepared { .. }
583            | BackingInitializationState::InFlight { .. } => Err(invalid_resource(
584                "backing initialization is owned by another submission wave",
585            )),
586            BackingInitializationState::Poisoned => Err(invalid_resource(
587                "backing initialization authority is fail-closed",
588            )),
589        }
590    }
591
592    pub(super) fn mark_in_flight(&self, wave_fingerprint: &str) -> Result<(), VNextError> {
593        let mut state = match self.state.lock() {
594            Ok(state) => state,
595            Err(poisoned) => {
596                let mut state = poisoned.into_inner();
597                *state = BackingInitializationState::Poisoned;
598                self.status
599                    .store(BACKING_INITIALIZATION_POISONED, Ordering::Release);
600                return Err(invalid_resource("backing initialization state is poisoned"));
601            }
602        };
603        match &*state {
604            BackingInitializationState::Prepared {
605                wave_fingerprint: current,
606            } if current == wave_fingerprint => {
607                *state = BackingInitializationState::InFlight {
608                    wave_fingerprint: wave_fingerprint.to_owned(),
609                };
610                self.status
611                    .store(BACKING_INITIALIZATION_IN_FLIGHT, Ordering::Release);
612                Ok(())
613            }
614            _ => {
615                *state = BackingInitializationState::Poisoned;
616                self.status
617                    .store(BACKING_INITIALIZATION_POISONED, Ordering::Release);
618                Err(invalid_resource(
619                    "backing initialization fence was installed from an invalid state",
620                ))
621            }
622        }
623    }
624
625    pub(super) fn finish(&self, wave_fingerprint: &str, succeeded: bool) -> Result<(), VNextError> {
626        let mut state = match self.state.lock() {
627            Ok(state) => state,
628            Err(poisoned) => {
629                let mut state = poisoned.into_inner();
630                *state = BackingInitializationState::Poisoned;
631                self.status
632                    .store(BACKING_INITIALIZATION_POISONED, Ordering::Release);
633                return Err(invalid_resource("backing initialization state is poisoned"));
634            }
635        };
636        match &*state {
637            BackingInitializationState::InFlight {
638                wave_fingerprint: current,
639            } if current == wave_fingerprint => {
640                *state = if succeeded {
641                    BackingInitializationState::Initialized
642                } else {
643                    BackingInitializationState::Poisoned
644                };
645                self.status.store(
646                    if succeeded {
647                        BACKING_INITIALIZATION_INITIALIZED
648                    } else {
649                        BACKING_INITIALIZATION_POISONED
650                    },
651                    Ordering::Release,
652                );
653                Ok(())
654            }
655            _ => {
656                *state = BackingInitializationState::Poisoned;
657                self.status
658                    .store(BACKING_INITIALIZATION_POISONED, Ordering::Release);
659                Err(invalid_resource(
660                    "backing initialization completed from an invalid state",
661                ))
662            }
663        }
664    }
665
666    pub(super) fn rollback_prepared(&self, wave_fingerprint: &str) {
667        let mut state = self
668            .state
669            .lock()
670            .unwrap_or_else(std::sync::PoisonError::into_inner);
671        match &*state {
672            BackingInitializationState::Prepared {
673                wave_fingerprint: current,
674            } if current == wave_fingerprint => {
675                *state = BackingInitializationState::Pending;
676                self.status
677                    .store(BACKING_INITIALIZATION_PENDING, Ordering::Release);
678            }
679            BackingInitializationState::Initialized | BackingInitializationState::Pending => {}
680            _ => {
681                *state = BackingInitializationState::Poisoned;
682                self.status
683                    .store(BACKING_INITIALIZATION_POISONED, Ordering::Release);
684            }
685        }
686    }
687
688    pub(super) fn mark_indeterminate(&self) {
689        let mut state = self
690            .state
691            .lock()
692            .unwrap_or_else(std::sync::PoisonError::into_inner);
693        *state = BackingInitializationState::Poisoned;
694        self.status
695            .store(BACKING_INITIALIZATION_POISONED, Ordering::Release);
696    }
697}
698
699impl Drop for BackingSegmentLease {
700    fn drop(&mut self) {
701        if !self.released {
702            self.owner
703                .release_segments(&self.claim_identity, self.occupancy, &self.segments);
704            self.released = true;
705        }
706    }
707}
708
709impl<R> BackingExtentOwner for DynamicBackingPool<R>
710where
711    R: DeviceRuntime,
712{
713    fn instance_id(&self) -> u64 {
714        self.instance_id
715    }
716
717    fn release_segments(
718        &self,
719        claim_identity: &PhysicalBackingClaimIdentity,
720        occupancy: DynamicBackingClaimOccupancy,
721        segments: &[BackingSegment],
722    ) {
723        let mut state = match self.state.lock() {
724            Ok(state) => state,
725            Err(poisoned) => {
726                let mut state = poisoned.into_inner();
727                state.poisoned = true;
728                return;
729            }
730        };
731        if state.poisoned {
732            return;
733        }
734        let segment_count = u64::try_from(segments.len()).ok();
735        let physical_bytes = segments.iter().try_fold(0_u64, |total, segment| {
736            total.checked_add(segment.length_bytes())
737        });
738        if claim_identity.pool_id() != self.domain.pool_id()
739            || Some(occupancy.segment_count) != segment_count
740            || Some(occupancy.physical_bytes) != physical_bytes
741        {
742            state.poisoned = true;
743            return;
744        }
745        let next_occupancy = match state.live_occupancy.checked_without_claim(occupancy) {
746            Ok(next) => next,
747            Err(_) => {
748                state.poisoned = true;
749                return;
750            }
751        };
752        for segment in segments {
753            if segment.pool_id() != self.domain.pool_id() {
754                state.poisoned = true;
755                return;
756            }
757            let Some(chunk) = state.chunks.get_mut(&segment.chunk_ordinal()) else {
758                state.poisoned = true;
759                return;
760            };
761            if chunk.backing.identity != *segment.chunk() || chunk.live_segments == 0 {
762                state.poisoned = true;
763                return;
764            }
765        }
766        for segment in segments {
767            if state.allocator.release(segment).is_err() {
768                state.poisoned = true;
769                return;
770            }
771            let chunk = state
772                .chunks
773                .get_mut(&segment.chunk_ordinal())
774                .expect("validated released chunk remains installed");
775            chunk.live_segments -= 1;
776        }
777        state.live_occupancy = next_occupancy;
778        drop(state);
779        if self
780            .logical_admission
781            .notify_domain_availability_changed(self.domain.domain_id)
782            .is_err()
783        {
784            let mut state = self
785                .state
786                .lock()
787                .unwrap_or_else(std::sync::PoisonError::into_inner);
788            state.poisoned = true;
789        }
790    }
791}
792
793#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
794pub struct DynamicPoolGrowthReceipt {
795    pub(super) pool_id: DynamicBackingPoolId,
796    pub(super) chunk: BackingChunkIdentity,
797    pub(super) chunk_bytes: u64,
798    pub(super) published_capacity_bytes: u64,
799    pub(super) capacity_epoch: u64,
800}
801
802#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
803pub struct DynamicPoolResourceContract {
804    pub(super) resource_id: ResourceId,
805    pub(super) demand: DynamicResourceDemand,
806    pub(super) lifetime: AllocationLifetime,
807    pub(super) kind: AllocationKind,
808    pub(super) physical_allocation_quantum_bytes: u64,
809    pub(super) initialization: StateInitialization,
810}
811
812impl DynamicPoolResourceContract {
813    fn from_descriptor(descriptor: &DynamicResourceDescriptor) -> Self {
814        Self {
815            resource_id: descriptor.base_resource_id().clone(),
816            demand: descriptor.demand().clone(),
817            lifetime: descriptor.lifetime(),
818            kind: descriptor.kind().clone(),
819            physical_allocation_quantum_bytes: descriptor.physical_allocation_quantum_bytes(),
820            initialization: descriptor.initialization(),
821        }
822    }
823
824    pub fn resource_id(&self) -> &ResourceId {
825        &self.resource_id
826    }
827
828    pub fn demand(&self) -> &DynamicResourceDemand {
829        &self.demand
830    }
831
832    pub const fn lifetime(&self) -> AllocationLifetime {
833        self.lifetime
834    }
835
836    pub fn kind(&self) -> &AllocationKind {
837        &self.kind
838    }
839
840    pub const fn physical_allocation_quantum_bytes(&self) -> u64 {
841        self.physical_allocation_quantum_bytes
842    }
843
844    pub const fn initialization(&self) -> StateInitialization {
845        self.initialization
846    }
847}
848
849#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
850pub struct DynamicPoolContractStatus {
851    pub(super) compatibility: PoolCompatibilityKey,
852    pub(super) resources: Vec<DynamicPoolResourceContract>,
853    pub(super) minimum_request_bytes: u64,
854    pub(super) minimum_sequence_bytes: u64,
855    pub(super) minimum_step_bytes: u64,
856    pub(super) minimum_invocation_peak_bytes: u64,
857    pub(super) reusable_workspace_ceiling_bytes: u64,
858    pub(super) provisioning: DynamicPoolProvisioningPolicy,
859    pub(super) invocation_liveness_mode: InvocationLivenessMode,
860}
861
862impl DynamicPoolContractStatus {
863    pub(super) fn from_domain(domain: &DynamicPoolDomainSpec) -> Self {
864        Self {
865            compatibility: domain.pool.compatibility().clone(),
866            resources: domain
867                .descriptors
868                .iter()
869                .map(DynamicPoolResourceContract::from_descriptor)
870                .collect(),
871            minimum_request_bytes: domain.pool.minimum_request_bytes(),
872            minimum_sequence_bytes: domain.pool.minimum_sequence_bytes(),
873            minimum_step_bytes: domain.pool.minimum_step_bytes(),
874            minimum_invocation_peak_bytes: domain.pool.minimum_invocation_peak_bytes(),
875            reusable_workspace_ceiling_bytes: domain.pool.reusable_workspace_ceiling_bytes(),
876            provisioning: domain.pool.provisioning().clone(),
877            invocation_liveness_mode: domain.pool.invocation_liveness_mode(),
878        }
879    }
880
881    pub fn compatibility(&self) -> &PoolCompatibilityKey {
882        &self.compatibility
883    }
884
885    pub fn resources(&self) -> &[DynamicPoolResourceContract] {
886        &self.resources
887    }
888
889    pub const fn minimum_request_bytes(&self) -> u64 {
890        self.minimum_request_bytes
891    }
892
893    pub const fn minimum_sequence_bytes(&self) -> u64 {
894        self.minimum_sequence_bytes
895    }
896
897    pub const fn minimum_step_bytes(&self) -> u64 {
898        self.minimum_step_bytes
899    }
900
901    pub const fn minimum_invocation_peak_bytes(&self) -> u64 {
902        self.minimum_invocation_peak_bytes
903    }
904
905    pub const fn reusable_workspace_ceiling_bytes(&self) -> u64 {
906        self.reusable_workspace_ceiling_bytes
907    }
908
909    pub fn provisioning(&self) -> &DynamicPoolProvisioningPolicy {
910        &self.provisioning
911    }
912
913    pub const fn invocation_liveness_mode(&self) -> InvocationLivenessMode {
914        self.invocation_liveness_mode
915    }
916}
917
918#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
919pub struct DynamicPoolStatus {
920    pub(super) pool_id: DynamicBackingPoolId,
921    pub(super) domain_id: CapacityDomainId,
922    pub(super) contract: DynamicPoolContractStatus,
923    pub(super) storage_profile: DynamicStorageProfile,
924    pub(super) resident_bytes: u64,
925    pub(super) pending_growth_bytes: u64,
926    pub(super) free_bytes: u64,
927    pub(super) largest_contiguous_bytes: u64,
928    pub(super) resident_chunks: usize,
929    pub(super) live_segments: u64,
930    pub(super) live_occupancy: DynamicPoolLiveOccupancyStatus,
931    pub(super) quarantined_chunks: usize,
932    pub(super) quarantined_bytes: u64,
933    pub(super) descriptor_mismatch_chunks: usize,
934    pub(super) publication_rejected_chunks: usize,
935    pub(super) poisoned: bool,
936}
937
938impl DynamicPoolStatus {
939    pub fn pool_id(&self) -> &DynamicBackingPoolId {
940        &self.pool_id
941    }
942
943    pub const fn domain_id(&self) -> CapacityDomainId {
944        self.domain_id
945    }
946
947    pub fn contract(&self) -> &DynamicPoolContractStatus {
948        &self.contract
949    }
950
951    pub const fn storage_profile(&self) -> DynamicStorageProfile {
952        self.storage_profile
953    }
954
955    pub const fn resident_bytes(&self) -> u64 {
956        self.resident_bytes
957    }
958
959    pub const fn pending_growth_bytes(&self) -> u64 {
960        self.pending_growth_bytes
961    }
962
963    pub const fn free_bytes(&self) -> u64 {
964        self.free_bytes
965    }
966
967    pub const fn largest_contiguous_bytes(&self) -> u64 {
968        self.largest_contiguous_bytes
969    }
970
971    pub const fn resident_chunks(&self) -> usize {
972        self.resident_chunks
973    }
974
975    pub const fn live_segments(&self) -> u64 {
976        self.live_segments
977    }
978
979    pub const fn live_occupancy(&self) -> &DynamicPoolLiveOccupancyStatus {
980        &self.live_occupancy
981    }
982
983    pub const fn quarantined_chunks(&self) -> usize {
984        self.quarantined_chunks
985    }
986
987    pub const fn quarantined_bytes(&self) -> u64 {
988        self.quarantined_bytes
989    }
990
991    pub const fn descriptor_mismatch_chunks(&self) -> usize {
992        self.descriptor_mismatch_chunks
993    }
994
995    pub const fn publication_rejected_chunks(&self) -> usize {
996        self.publication_rejected_chunks
997    }
998
999    pub const fn poisoned(&self) -> bool {
1000        self.poisoned
1001    }
1002}
1003
1004#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1005pub struct DynamicPoolIdleReclaim {
1006    pub(super) pool_id: DynamicBackingPoolId,
1007    pub(super) chunks: Vec<BackingChunkIdentity>,
1008    pub(super) reclaimed_bytes: u64,
1009    pub(super) published_capacity_bytes: u64,
1010}
1011
1012impl DynamicPoolIdleReclaim {
1013    pub fn pool_id(&self) -> &DynamicBackingPoolId {
1014        &self.pool_id
1015    }
1016
1017    pub fn chunks(&self) -> &[BackingChunkIdentity] {
1018        &self.chunks
1019    }
1020
1021    pub const fn reclaimed_bytes(&self) -> u64 {
1022        self.reclaimed_bytes
1023    }
1024
1025    pub const fn published_capacity_bytes(&self) -> u64 {
1026        self.published_capacity_bytes
1027    }
1028}
1029
1030#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1031pub struct DynamicPoolRebalanceReceipt {
1032    pub(super) pools: Vec<DynamicPoolIdleReclaim>,
1033    pub(super) reclaimed_chunks: usize,
1034    pub(super) reclaimed_bytes: u64,
1035    pub(super) logical_capacity_epoch: u64,
1036    pub(super) plan_device_capacity_epoch: u64,
1037    pub(super) process_device_capacity_epoch: u64,
1038}
1039
1040impl DynamicPoolRebalanceReceipt {
1041    pub fn pools(&self) -> &[DynamicPoolIdleReclaim] {
1042        &self.pools
1043    }
1044
1045    pub const fn reclaimed_chunks(&self) -> usize {
1046        self.reclaimed_chunks
1047    }
1048
1049    pub const fn reclaimed_bytes(&self) -> u64 {
1050        self.reclaimed_bytes
1051    }
1052
1053    pub const fn logical_capacity_epoch(&self) -> u64 {
1054        self.logical_capacity_epoch
1055    }
1056
1057    pub const fn plan_device_capacity_epoch(&self) -> u64 {
1058        self.plan_device_capacity_epoch
1059    }
1060
1061    pub const fn process_device_capacity_epoch(&self) -> u64 {
1062        self.process_device_capacity_epoch
1063    }
1064}
1065
1066pub(super) struct DynamicDeviceCapacityBlocked {
1067    pub(super) pressure: DeviceCapacityPressure,
1068    pub(super) availability: DeviceCapacityAvailabilitySnapshot,
1069    pub(super) planned_domains: Vec<CapacityDomainId>,
1070}
1071
1072impl DynamicPoolGrowthReceipt {
1073    pub fn pool_id(&self) -> &DynamicBackingPoolId {
1074        &self.pool_id
1075    }
1076
1077    pub fn chunk(&self) -> &BackingChunkIdentity {
1078        &self.chunk
1079    }
1080
1081    pub const fn chunk_bytes(&self) -> u64 {
1082        self.chunk_bytes
1083    }
1084
1085    pub const fn published_capacity_bytes(&self) -> u64 {
1086        self.published_capacity_bytes
1087    }
1088
1089    pub const fn capacity_epoch(&self) -> u64 {
1090        self.capacity_epoch
1091    }
1092}
1093
1094#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1095pub struct DynamicPoolGrowthRequest {
1096    pool_id: DynamicBackingPoolId,
1097    requested_bytes: u64,
1098}
1099
1100impl DynamicPoolGrowthRequest {
1101    pub fn new(pool_id: DynamicBackingPoolId, requested_bytes: u64) -> Result<Self, VNextError> {
1102        if requested_bytes == 0 {
1103            return Err(invalid_resource(
1104                "dynamic pool growth must request non-zero bytes",
1105            ));
1106        }
1107        Ok(Self {
1108            pool_id,
1109            requested_bytes,
1110        })
1111    }
1112
1113    pub fn pool_id(&self) -> &DynamicBackingPoolId {
1114        &self.pool_id
1115    }
1116
1117    pub const fn requested_bytes(&self) -> u64 {
1118        self.requested_bytes
1119    }
1120}
1121
1122#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
1123#[serde(rename_all = "snake_case")]
1124pub enum DynamicBackingDeferralReason {
1125    GrowthRequired,
1126    FragmentedContiguous,
1127}
1128
1129/// Semantic ownership boundary for one atomic physical backing attempt.
1130/// `InitialSequenceBundle` is the only scope allowed to combine Request and
1131/// Sequence descriptors; it publishes neither lifetime unless both can commit.
1132#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
1133#[serde(rename_all = "snake_case")]
1134pub enum DynamicBackingClaimScope {
1135    Plan,
1136    Request,
1137    Sequence,
1138    /// Independent immutable storage of Sequence state, without a live
1139    /// request/sequence execution slot or a sequence parent authority.
1140    Checkpoint,
1141    Step,
1142    Invocation,
1143    InitialSequenceBundle,
1144}
1145
1146impl DynamicBackingClaimScope {
1147    pub(super) const fn accepts(self, lifetime: AllocationLifetime) -> bool {
1148        match self {
1149            Self::Plan => matches!(lifetime, AllocationLifetime::Plan),
1150            Self::Request => matches!(lifetime, AllocationLifetime::Request),
1151            Self::Sequence => matches!(lifetime, AllocationLifetime::Sequence),
1152            Self::Checkpoint => matches!(lifetime, AllocationLifetime::Sequence),
1153            Self::Step => matches!(lifetime, AllocationLifetime::Step),
1154            Self::Invocation => matches!(lifetime, AllocationLifetime::Invocation),
1155            Self::InitialSequenceBundle => matches!(
1156                lifetime,
1157                AllocationLifetime::Request | AllocationLifetime::Sequence
1158            ),
1159        }
1160    }
1161
1162    pub const fn lifetime(self) -> Option<AllocationLifetime> {
1163        match self {
1164            Self::Plan => Some(AllocationLifetime::Plan),
1165            Self::Request => Some(AllocationLifetime::Request),
1166            Self::Sequence => Some(AllocationLifetime::Sequence),
1167            Self::Checkpoint => None,
1168            Self::Step => Some(AllocationLifetime::Step),
1169            Self::Invocation => Some(AllocationLifetime::Invocation),
1170            Self::InitialSequenceBundle => None,
1171        }
1172    }
1173}
1174
1175impl From<AllocationLifetime> for DynamicBackingClaimScope {
1176    fn from(lifetime: AllocationLifetime) -> Self {
1177        match lifetime {
1178            AllocationLifetime::Plan => Self::Plan,
1179            AllocationLifetime::Request => Self::Request,
1180            AllocationLifetime::Sequence => Self::Sequence,
1181            AllocationLifetime::Step => Self::Step,
1182            AllocationLifetime::Invocation => Self::Invocation,
1183        }
1184    }
1185}
1186
1187#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
1188#[serde(rename_all = "snake_case")]
1189pub(super) enum DynamicBackingClaimResidency {
1190    Transient,
1191    LaneStable,
1192}
1193
1194#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)]
1195pub struct DynamicPoolOccupancyCounter {
1196    pub(super) claim_count: u64,
1197    pub(super) segment_count: u64,
1198    pub(super) physical_bytes: u64,
1199}
1200
1201impl DynamicPoolOccupancyCounter {
1202    pub const fn claim_count(&self) -> u64 {
1203        self.claim_count
1204    }
1205
1206    pub const fn segment_count(&self) -> u64 {
1207        self.segment_count
1208    }
1209
1210    pub const fn physical_bytes(&self) -> u64 {
1211        self.physical_bytes
1212    }
1213
1214    fn checked_add_claim(&mut self, claim: DynamicBackingClaimOccupancy) -> Result<(), VNextError> {
1215        self.claim_count = self
1216            .claim_count
1217            .checked_add(1)
1218            .ok_or_else(|| invalid_resource("dynamic live claim count overflows u64"))?;
1219        self.segment_count = self
1220            .segment_count
1221            .checked_add(claim.segment_count)
1222            .ok_or_else(|| invalid_resource("dynamic live claim segment count overflows u64"))?;
1223        self.physical_bytes = self
1224            .physical_bytes
1225            .checked_add(claim.physical_bytes)
1226            .ok_or_else(|| invalid_resource("dynamic live claim physical bytes overflow u64"))?;
1227        Ok(())
1228    }
1229
1230    fn checked_remove_claim(
1231        &mut self,
1232        claim: DynamicBackingClaimOccupancy,
1233    ) -> Result<(), VNextError> {
1234        self.claim_count = self
1235            .claim_count
1236            .checked_sub(1)
1237            .ok_or_else(|| invalid_resource("dynamic live claim count underflows u64"))?;
1238        self.segment_count = self
1239            .segment_count
1240            .checked_sub(claim.segment_count)
1241            .ok_or_else(|| invalid_resource("dynamic live claim segment count underflows u64"))?;
1242        self.physical_bytes = self
1243            .physical_bytes
1244            .checked_sub(claim.physical_bytes)
1245            .ok_or_else(|| invalid_resource("dynamic live claim physical bytes underflow u64"))?;
1246        Ok(())
1247    }
1248}
1249
1250#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)]
1251pub struct DynamicPoolResidencyOccupancyStatus {
1252    pub(super) total: DynamicPoolOccupancyCounter,
1253    pub(super) plan: DynamicPoolOccupancyCounter,
1254    pub(super) request: DynamicPoolOccupancyCounter,
1255    pub(super) sequence: DynamicPoolOccupancyCounter,
1256    pub(super) checkpoint: DynamicPoolOccupancyCounter,
1257    pub(super) step: DynamicPoolOccupancyCounter,
1258    pub(super) invocation: DynamicPoolOccupancyCounter,
1259    pub(super) initial_sequence_bundle: DynamicPoolOccupancyCounter,
1260}
1261
1262impl DynamicPoolResidencyOccupancyStatus {
1263    pub const fn total(&self) -> &DynamicPoolOccupancyCounter {
1264        &self.total
1265    }
1266
1267    pub const fn plan(&self) -> &DynamicPoolOccupancyCounter {
1268        &self.plan
1269    }
1270
1271    pub const fn request(&self) -> &DynamicPoolOccupancyCounter {
1272        &self.request
1273    }
1274
1275    pub const fn sequence(&self) -> &DynamicPoolOccupancyCounter {
1276        &self.sequence
1277    }
1278
1279    pub const fn checkpoint(&self) -> &DynamicPoolOccupancyCounter {
1280        &self.checkpoint
1281    }
1282
1283    pub const fn step(&self) -> &DynamicPoolOccupancyCounter {
1284        &self.step
1285    }
1286
1287    pub const fn invocation(&self) -> &DynamicPoolOccupancyCounter {
1288        &self.invocation
1289    }
1290
1291    pub const fn initial_sequence_bundle(&self) -> &DynamicPoolOccupancyCounter {
1292        &self.initial_sequence_bundle
1293    }
1294
1295    fn counter_mut_for_scope(
1296        &mut self,
1297        scope: DynamicBackingClaimScope,
1298    ) -> &mut DynamicPoolOccupancyCounter {
1299        match scope {
1300            DynamicBackingClaimScope::Plan => &mut self.plan,
1301            DynamicBackingClaimScope::Request => &mut self.request,
1302            DynamicBackingClaimScope::Sequence => &mut self.sequence,
1303            DynamicBackingClaimScope::Checkpoint => &mut self.checkpoint,
1304            DynamicBackingClaimScope::Step => &mut self.step,
1305            DynamicBackingClaimScope::Invocation => &mut self.invocation,
1306            DynamicBackingClaimScope::InitialSequenceBundle => &mut self.initial_sequence_bundle,
1307        }
1308    }
1309
1310    fn checked_with_claim(&self, claim: DynamicBackingClaimOccupancy) -> Result<Self, VNextError> {
1311        let mut next = *self;
1312        next.total.checked_add_claim(claim)?;
1313        next.counter_mut_for_scope(claim.scope)
1314            .checked_add_claim(claim)?;
1315        Ok(next)
1316    }
1317
1318    fn checked_without_claim(
1319        &self,
1320        claim: DynamicBackingClaimOccupancy,
1321    ) -> Result<Self, VNextError> {
1322        let mut next = *self;
1323        next.total.checked_remove_claim(claim)?;
1324        next.counter_mut_for_scope(claim.scope)
1325            .checked_remove_claim(claim)?;
1326        Ok(next)
1327    }
1328}
1329
1330#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)]
1331pub struct DynamicPoolLiveOccupancyStatus {
1332    pub(super) total: DynamicPoolOccupancyCounter,
1333    pub(super) transient: DynamicPoolResidencyOccupancyStatus,
1334    pub(super) lane_stable: DynamicPoolResidencyOccupancyStatus,
1335}
1336
1337impl DynamicPoolLiveOccupancyStatus {
1338    pub const fn total(&self) -> &DynamicPoolOccupancyCounter {
1339        &self.total
1340    }
1341
1342    pub const fn transient(&self) -> &DynamicPoolResidencyOccupancyStatus {
1343        &self.transient
1344    }
1345
1346    pub const fn lane_stable(&self) -> &DynamicPoolResidencyOccupancyStatus {
1347        &self.lane_stable
1348    }
1349
1350    fn residency_mut(
1351        &mut self,
1352        residency: DynamicBackingClaimResidency,
1353    ) -> &mut DynamicPoolResidencyOccupancyStatus {
1354        match residency {
1355            DynamicBackingClaimResidency::Transient => &mut self.transient,
1356            DynamicBackingClaimResidency::LaneStable => &mut self.lane_stable,
1357        }
1358    }
1359
1360    pub(super) fn checked_with_claim(
1361        &self,
1362        claim: DynamicBackingClaimOccupancy,
1363    ) -> Result<Self, VNextError> {
1364        let mut next = *self;
1365        next.total.checked_add_claim(claim)?;
1366        let updated = (*next.residency_mut(claim.residency)).checked_with_claim(claim)?;
1367        *next.residency_mut(claim.residency) = updated;
1368        Ok(next)
1369    }
1370
1371    pub(super) fn checked_without_claim(
1372        &self,
1373        claim: DynamicBackingClaimOccupancy,
1374    ) -> Result<Self, VNextError> {
1375        let mut next = *self;
1376        next.total.checked_remove_claim(claim)?;
1377        let updated = (*next.residency_mut(claim.residency)).checked_without_claim(claim)?;
1378        *next.residency_mut(claim.residency) = updated;
1379        Ok(next)
1380    }
1381}
1382
1383#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1384pub struct DynamicBackingBlocker {
1385    pub(super) pool_id: DynamicBackingPoolId,
1386    pub(super) domain_id: CapacityDomainId,
1387    pub(super) reason: DynamicBackingDeferralReason,
1388    pub(super) requested_bytes: u64,
1389    pub(super) free_bytes: u64,
1390    pub(super) largest_contiguous_bytes: u64,
1391    pub(super) free_extent_layout_fingerprint: String,
1392    #[serde(skip_serializing_if = "Option::is_none")]
1393    pub(super) contiguous_claim_bytes_descending: Option<Vec<u64>>,
1394}
1395
1396impl DynamicBackingBlocker {
1397    pub fn pool_id(&self) -> &DynamicBackingPoolId {
1398        &self.pool_id
1399    }
1400
1401    pub const fn domain_id(&self) -> CapacityDomainId {
1402        self.domain_id
1403    }
1404
1405    pub const fn reason(&self) -> DynamicBackingDeferralReason {
1406        self.reason
1407    }
1408
1409    pub const fn requested_bytes(&self) -> u64 {
1410        self.requested_bytes
1411    }
1412
1413    pub const fn free_bytes(&self) -> u64 {
1414        self.free_bytes
1415    }
1416
1417    pub const fn largest_contiguous_bytes(&self) -> u64 {
1418        self.largest_contiguous_bytes
1419    }
1420
1421    pub fn free_extent_layout_fingerprint(&self) -> &str {
1422        &self.free_extent_layout_fingerprint
1423    }
1424
1425    pub fn contiguous_claim_bytes_descending(&self) -> Option<&[u64]> {
1426        self.contiguous_claim_bytes_descending.as_deref()
1427    }
1428}
1429
1430#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1431pub struct DynamicBackingPackingEnvelope {
1432    pub(super) pool_id: DynamicBackingPoolId,
1433    pub(super) domain_id: CapacityDomainId,
1434    pub(super) claim_bytes_descending: Vec<u64>,
1435}
1436
1437impl DynamicBackingPackingEnvelope {
1438    pub(super) fn new(
1439        pool_id: DynamicBackingPoolId,
1440        domain_id: CapacityDomainId,
1441        mut claim_bytes_descending: Vec<u64>,
1442    ) -> Result<Self, VNextError> {
1443        if claim_bytes_descending.is_empty() || claim_bytes_descending.contains(&0) {
1444            return Err(invalid_resource(
1445                "dynamic backing packing envelope contains empty or zero-sized demand",
1446            ));
1447        }
1448        claim_bytes_descending.sort_unstable_by(|left, right| right.cmp(left));
1449        claim_bytes_descending
1450            .iter()
1451            .try_fold(0_u64, |total, bytes| {
1452                total.checked_add(*bytes).ok_or_else(|| {
1453                    invalid_resource("dynamic backing packing envelope bytes overflow u64")
1454                })
1455            })?;
1456        Ok(Self {
1457            pool_id,
1458            domain_id,
1459            claim_bytes_descending,
1460        })
1461    }
1462
1463    pub fn pool_id(&self) -> &DynamicBackingPoolId {
1464        &self.pool_id
1465    }
1466
1467    pub const fn domain_id(&self) -> CapacityDomainId {
1468        self.domain_id
1469    }
1470
1471    pub fn claim_bytes_descending(&self) -> &[u64] {
1472        &self.claim_bytes_descending
1473    }
1474
1475    pub(super) fn total_bytes(&self) -> Result<u64, VNextError> {
1476        self.claim_bytes_descending
1477            .iter()
1478            .try_fold(0_u64, |total, bytes| {
1479                total.checked_add(*bytes).ok_or_else(|| {
1480                    invalid_resource("dynamic backing packing envelope bytes overflow u64")
1481                })
1482            })
1483    }
1484}
1485
1486#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1487pub struct DynamicBackingDeferred {
1488    pub(super) blockers: Vec<DynamicBackingBlocker>,
1489    pub(super) epochs: CapacityEpochs,
1490    pub(super) wait_condition: CapacityWaitCondition,
1491    pub(super) scope: DynamicBackingClaimScope,
1492    pub(super) protected_immediate: CapacityVector,
1493    pub(super) protected_packing_envelopes: Vec<DynamicBackingPackingEnvelope>,
1494}
1495
1496impl DynamicBackingDeferred {
1497    pub fn blockers(&self) -> &[DynamicBackingBlocker] {
1498        &self.blockers
1499    }
1500
1501    pub const fn release_epoch(&self) -> u64 {
1502        self.epochs.release_epoch()
1503    }
1504
1505    pub const fn capacity_epoch(&self) -> u64 {
1506        self.epochs.capacity_epoch()
1507    }
1508
1509    pub const fn epochs(&self) -> CapacityEpochs {
1510        self.epochs
1511    }
1512
1513    pub fn wait_condition(&self) -> &CapacityWaitCondition {
1514        &self.wait_condition
1515    }
1516
1517    pub const fn scope(&self) -> DynamicBackingClaimScope {
1518        self.scope
1519    }
1520
1521    pub const fn lifetime(&self) -> Option<AllocationLifetime> {
1522        self.scope.lifetime()
1523    }
1524
1525    /// Exact uncommitted physical demand that must remain simultaneously
1526    /// runnable while maintenance rebalances other pools.
1527    pub fn protected_immediate(&self) -> &CapacityVector {
1528        &self.protected_immediate
1529    }
1530
1531    /// Exact per-pool physical claim shapes that must remain packable while
1532    /// maintenance rebalances unreferenced chunks for this deferred bundle.
1533    pub fn protected_packing_envelopes(&self) -> &[DynamicBackingPackingEnvelope] {
1534        &self.protected_packing_envelopes
1535    }
1536}
1537
1538#[derive(Clone)]
1539pub(super) enum DynamicPoolGrowthIntent {
1540    Additional(DynamicPoolGrowthRequest),
1541    Minimum(DynamicBackingPoolId),
1542    RevalidatedDeferral(DynamicBackingBlocker),
1543    /// Logical admission has not acquired this demand. Recheck both ledgers
1544    /// under the pool maintenance authority before deciding any growth.
1545    RevalidatedAdmissionPressure {
1546        pool_id: DynamicBackingPoolId,
1547        required_free_bytes: u64,
1548    },
1549}
1550
1551impl DynamicPoolGrowthIntent {
1552    pub(super) fn pool_id(&self) -> &DynamicBackingPoolId {
1553        match self {
1554            Self::Additional(request) => request.pool_id(),
1555            Self::Minimum(pool_id) => pool_id,
1556            Self::RevalidatedDeferral(blocker) => blocker.pool_id(),
1557            Self::RevalidatedAdmissionPressure { pool_id, .. } => pool_id,
1558        }
1559    }
1560}
1561
1562pub(super) struct PlannedDynamicGrowth<R>
1563where
1564    R: DeviceRuntime,
1565{
1566    pub(super) pool: Arc<DynamicBackingPool<R>>,
1567    pub(super) chunk: BackingChunkIdentity,
1568    pub(super) expected_resource_id: ResourceId,
1569    pub(super) chunk_bytes: u64,
1570}
1571
1572pub(super) struct AllocatedDynamicGrowth<B> {
1573    pub(super) backing: Arc<ResidentChunkBacking<B>>,
1574}
1575
1576#[derive(Clone)]
1577pub(super) struct IdleChunkReclaimCandidate {
1578    pub(super) pool_index: usize,
1579    pub(super) chunk: BackingChunkIdentity,
1580    pub(super) chunk_bytes: u64,
1581}
1582
1583#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize)]
1584pub struct PhysicalBackingClaimIdentity {
1585    pool_id: DynamicBackingPoolId,
1586    resource_ids: Vec<ResourceId>,
1587}
1588
1589impl PhysicalBackingClaimIdentity {
1590    pub(super) fn new(
1591        pool_id: DynamicBackingPoolId,
1592        mut resource_ids: Vec<ResourceId>,
1593    ) -> Result<Self, VNextError> {
1594        resource_ids.sort();
1595        if resource_ids.is_empty() || resource_ids.windows(2).any(|pair| pair[0] == pair[1]) {
1596            return Err(invalid_resource(
1597                "physical backing claim identity requires unique logical resources",
1598            ));
1599        }
1600        Ok(Self {
1601            pool_id,
1602            resource_ids,
1603        })
1604    }
1605
1606    pub fn pool_id(&self) -> &DynamicBackingPoolId {
1607        &self.pool_id
1608    }
1609
1610    pub fn resource_ids(&self) -> &[ResourceId] {
1611        &self.resource_ids
1612    }
1613
1614    pub const fn is_shared(&self) -> bool {
1615        self.resource_ids.len() > 1
1616    }
1617}
1618
1619#[derive(Clone)]
1620pub(super) struct EvaluatedBackingProjection<'a> {
1621    pub(super) descriptor: &'a DynamicResourceDescriptor,
1622    pub(super) physical_offset_bytes: u64,
1623    pub(super) logical_size_bytes: u64,
1624    pub(super) capacity_size_bytes: u64,
1625}
1626
1627#[derive(Clone)]
1628pub(super) struct EvaluatedBackingRequest<'a> {
1629    pub(super) domain: &'a DynamicPoolDomainSpec,
1630    pub(super) claim_identity: PhysicalBackingClaimIdentity,
1631    pub(super) capacity_size_bytes: u64,
1632    pub(super) reusable_execution_bucket_id: Option<ReusableExecutionBucketId>,
1633    pub(super) projections: Vec<EvaluatedBackingProjection<'a>>,
1634}
1635
1636pub(super) struct PreparedBackingExtent<R>
1637where
1638    R: DeviceRuntime,
1639{
1640    pub(super) pool: Arc<DynamicBackingPool<R>>,
1641    pub(super) claim_identity: PhysicalBackingClaimIdentity,
1642    pub(super) segment_generation: u64,
1643    pub(super) occupancy: DynamicBackingClaimOccupancy,
1644    pub(super) segments: Vec<BackingSegment>,
1645    pub(super) capacity_size_bytes: u64,
1646    pub(super) projections: Vec<LogicalBackingSliceEvidence>,
1647}
1648
1649pub(super) struct PreparedBackingClaim<R>
1650where
1651    R: DeviceRuntime,
1652{
1653    pub(super) extents: Vec<PreparedBackingExtent<R>>,
1654    pub(super) committed: bool,
1655}
1656
1657impl<R> PreparedBackingClaim<R>
1658where
1659    R: DeviceRuntime,
1660{
1661    pub(super) fn empty() -> Self {
1662        Self {
1663            extents: Vec::new(),
1664            committed: false,
1665        }
1666    }
1667
1668    pub(super) fn commit(mut self) -> Vec<LogicalBackingSliceAuthority> {
1669        let mut slices = Vec::new();
1670        for extent in std::mem::take(&mut self.extents) {
1671            let initialization = extent
1672                .projections
1673                .iter()
1674                .any(|projection| projection.initialization == StateInitialization::Zero)
1675                .then(|| {
1676                    Arc::new(BackingInitializationCell::new(
1677                        backing_initialization_target_fingerprint(&extent),
1678                    ))
1679                });
1680            let owner: Arc<dyn BackingExtentOwner> = extent.pool;
1681            let segment_lease = Arc::new(BackingSegmentLease {
1682                owner_instance_id: owner.instance_id(),
1683                owner,
1684                claim_identity: extent.claim_identity,
1685                segment_generation: extent.segment_generation,
1686                occupancy: extent.occupancy,
1687                segments: extent.segments,
1688                size_bytes: extent.capacity_size_bytes,
1689                initialization,
1690                released: false,
1691            });
1692            slices.extend(extent.projections.into_iter().map(|evidence| {
1693                LogicalBackingSliceAuthority {
1694                    evidence,
1695                    segment_lease: Arc::clone(&segment_lease),
1696                    reusable_lane: None,
1697                }
1698            }));
1699        }
1700        slices.sort_by(|left, right| left.resource_id().cmp(right.resource_id()));
1701        self.committed = true;
1702        slices
1703    }
1704}
1705
1706fn backing_initialization_target_fingerprint<R>(extent: &PreparedBackingExtent<R>) -> String
1707where
1708    R: DeviceRuntime,
1709{
1710    let mut hasher = Sha256::new();
1711    hasher.update(b"ferrum.runtime-vnext.backing-initialization-target.v1\0");
1712    hasher.update(extent.pool.instance_id.to_be_bytes());
1713    hasher.update(extent.segment_generation.to_be_bytes());
1714    hasher.update(extent.claim_identity.pool_id().as_str().as_bytes());
1715    for resource_id in extent.claim_identity.resource_ids() {
1716        hasher.update([0]);
1717        hasher.update(resource_id.as_str().as_bytes());
1718    }
1719    for segment in &extent.segments {
1720        hasher.update(segment.chunk_ordinal().to_be_bytes());
1721        hasher.update(segment.chunk_generation().to_be_bytes());
1722        hasher.update(segment.offset_bytes().to_be_bytes());
1723        hasher.update(segment.length_bytes().to_be_bytes());
1724    }
1725    for projection in extent
1726        .projections
1727        .iter()
1728        .filter(|projection| projection.initialization == StateInitialization::Zero)
1729    {
1730        hasher.update([1]);
1731        hasher.update(projection.resource_id.as_str().as_bytes());
1732        hasher.update(projection.physical_offset_bytes.to_be_bytes());
1733        hasher.update(projection.capacity_size_bytes.to_be_bytes());
1734    }
1735    format!("sha256/{:x}", hasher.finalize())
1736}
1737
1738impl<R> Drop for PreparedBackingClaim<R>
1739where
1740    R: DeviceRuntime,
1741{
1742    fn drop(&mut self) {
1743        if self.committed {
1744            return;
1745        }
1746        for extent in self.extents.iter().rev() {
1747            extent.pool.rollback_prepared(
1748                &extent.claim_identity,
1749                extent.occupancy,
1750                &extent.segments,
1751            );
1752        }
1753    }
1754}
1755
1756pub(super) enum BackingPrepareDecision<R>
1757where
1758    R: DeviceRuntime,
1759{
1760    Prepared(PreparedBackingClaim<R>),
1761    Deferred(DynamicBackingDeferred),
1762}
1763
1764/// Cold physical-layout proof for one committed backing claim.
1765///
1766/// Lane-stable arenas build this once when a slot is published. Reusing the
1767/// slot only binds the current logical projection sizes, so hot admission does
1768/// not need to reconstruct physical-claim maps or re-encode allocation
1769/// evidence on every wave.
1770#[derive(Debug)]
1771pub(super) struct BackingClaimCertificate {
1772    allocations: Box<[Arc<LogicalBackingSliceAllocationEvidence>]>,
1773    physical_capacity: CapacityVector,
1774    reusable_execution_bucket_id: Option<ReusableExecutionBucketId>,
1775    physical_claim_count: usize,
1776    has_shared_physical_claims: bool,
1777    fingerprint: String,
1778}
1779
1780#[derive(Debug)]
1781pub(super) struct BoundBackingClaimCertificate {
1782    fingerprint: String,
1783    physical_claim_count: usize,
1784    has_shared_physical_claims: bool,
1785}
1786
1787impl BoundBackingClaimCertificate {
1788    pub(super) fn fingerprint(&self) -> &str {
1789        &self.fingerprint
1790    }
1791
1792    pub(super) const fn physical_claim_count(&self) -> usize {
1793        self.physical_claim_count
1794    }
1795
1796    pub(super) const fn has_shared_physical_claims(&self) -> bool {
1797        self.has_shared_physical_claims
1798    }
1799}
1800
1801impl BackingClaimCertificate {
1802    pub(super) fn from_slices(
1803        backing_slices: &[LogicalBackingSliceAuthority],
1804    ) -> Result<Self, VNextError> {
1805        if backing_slices
1806            .windows(2)
1807            .any(|pair| pair[0].resource_id() >= pair[1].resource_id())
1808        {
1809            return Err(invalid_resource(
1810                "backing claim certificate requires canonical unique logical projections",
1811            ));
1812        }
1813        let reusable_execution_bucket_id = backing_slices
1814            .first()
1815            .and_then(|slice| slice.evidence().reusable_execution_bucket_id())
1816            .cloned();
1817        if backing_slices.iter().any(|slice| {
1818            slice.evidence().reusable_execution_bucket_id() != reusable_execution_bucket_id.as_ref()
1819        }) {
1820            return Err(invalid_resource(
1821                "one backing certificate cannot mix reusable execution buckets",
1822            ));
1823        }
1824
1825        let mut backing_by_domain = BTreeMap::<CapacityDomainId, u64>::new();
1826        let mut physical_claims = BTreeMap::<
1827            PhysicalBackingClaimIdentity,
1828            (Arc<BackingSegmentLease>, CapacityDomainId, u64),
1829        >::new();
1830        let mut has_shared_physical_claims = false;
1831        for slice in backing_slices {
1832            let evidence = slice.evidence();
1833            let claim_identity = evidence.physical_claim_identity();
1834            if claim_identity.pool_id() != evidence.pool_id()
1835                || claim_identity
1836                    .resource_ids()
1837                    .binary_search(evidence.resource_id())
1838                    .is_err()
1839                || slice.segment_lease.claim_identity != *claim_identity
1840                || slice.segment_lease.segment_generation != evidence.segment_generation()
1841                || slice.segment_lease.size_bytes != evidence.physical_size_bytes()
1842                || evidence.size_bytes() == 0
1843                || evidence.size_bytes() > evidence.capacity_size_bytes()
1844                || evidence
1845                    .physical_offset_bytes()
1846                    .checked_add(evidence.capacity_size_bytes())
1847                    .is_none_or(|end| end > evidence.physical_size_bytes())
1848            {
1849                return Err(invalid_resource(
1850                    "logical backing projection differs from its physical claim authority",
1851                ));
1852            }
1853            has_shared_physical_claims |= claim_identity.is_shared();
1854            match physical_claims.entry(claim_identity.clone()) {
1855                std::collections::btree_map::Entry::Vacant(entry) => {
1856                    let total = backing_by_domain.entry(slice.domain_id()).or_default();
1857                    *total = total
1858                        .checked_add(evidence.physical_size_bytes())
1859                        .ok_or_else(|| {
1860                            invalid_resource("certified backing domain bytes overflow u64")
1861                        })?;
1862                    entry.insert((
1863                        Arc::clone(&slice.segment_lease),
1864                        slice.domain_id(),
1865                        evidence.physical_size_bytes(),
1866                    ));
1867                }
1868                std::collections::btree_map::Entry::Occupied(entry) => {
1869                    let (lease, domain_id, size_bytes) = entry.get();
1870                    if !Arc::ptr_eq(lease, &slice.segment_lease)
1871                        || *domain_id != slice.domain_id()
1872                        || *size_bytes != evidence.physical_size_bytes()
1873                    {
1874                        return Err(invalid_resource(
1875                            "shared logical projections do not retain one physical claim",
1876                        ));
1877                    }
1878                }
1879            }
1880        }
1881        let physical_capacity = if backing_by_domain.is_empty() {
1882            CapacityVector::empty()
1883        } else {
1884            CapacityVector::new(
1885                backing_by_domain
1886                    .into_iter()
1887                    .map(|(domain, bytes)| CapacityEntry::new(domain, CapacityUnits::new(bytes)))
1888                    .collect::<Result<Vec<_>, _>>()?,
1889            )?
1890        };
1891        let allocations = backing_slices
1892            .iter()
1893            .map(|slice| Arc::clone(&slice.evidence.allocation))
1894            .collect::<Vec<_>>()
1895            .into_boxed_slice();
1896        let mut hasher = Sha256::new();
1897        hasher.update(b"ferrum.runtime-vnext.backing-claim-certificate.v1\0");
1898        for allocation in &allocations {
1899            let fingerprint = allocation.fingerprint.as_bytes();
1900            hasher.update(
1901                u64::try_from(fingerprint.len())
1902                    .map_err(|_| {
1903                        invalid_resource(
1904                            "backing allocation fingerprint length exceeds portable range",
1905                        )
1906                    })?
1907                    .to_be_bytes(),
1908            );
1909            hasher.update(fingerprint);
1910        }
1911        Ok(Self {
1912            allocations,
1913            physical_capacity,
1914            reusable_execution_bucket_id,
1915            physical_claim_count: physical_claims.len(),
1916            has_shared_physical_claims,
1917            fingerprint: format!("{:x}", hasher.finalize()),
1918        })
1919    }
1920
1921    pub(super) fn bind(
1922        &self,
1923        backing_slices: &[LogicalBackingSliceAuthority],
1924        demand: &super::AdmissionDemand,
1925    ) -> Result<BoundBackingClaimCertificate, VNextError> {
1926        if backing_slices.len() != self.allocations.len() {
1927            return Err(invalid_resource(
1928                "bound backing projection count differs from its physical certificate",
1929            ));
1930        }
1931        let mut hasher = Sha256::new();
1932        hasher.update(b"ferrum.runtime-vnext.bound-backing-claim.v1\0");
1933        hasher.update(self.fingerprint.as_bytes());
1934        for (slice, allocation) in backing_slices.iter().zip(&self.allocations) {
1935            if !Arc::ptr_eq(&slice.evidence.allocation, allocation)
1936                || slice.evidence.size_bytes() == 0
1937                || slice.evidence.size_bytes() > allocation.capacity_size_bytes
1938            {
1939                return Err(invalid_resource(
1940                    "bound logical projection differs from its certified allocation",
1941                ));
1942            }
1943            hasher.update(slice.evidence.size_bytes().to_be_bytes());
1944        }
1945        let physical_covers_logical = self.physical_capacity.entries().len()
1946            == demand.immediate_claim().entries().len()
1947            && self.physical_capacity.entries().iter().all(|physical| {
1948                demand
1949                    .immediate_claim()
1950                    .units_for(physical.domain())
1951                    .is_some_and(|logical| physical.units().get() >= logical.get())
1952            });
1953        let claim_matches = if self.reusable_execution_bucket_id.is_some() {
1954            physical_covers_logical
1955        } else {
1956            self.physical_capacity == *demand.immediate_claim()
1957        };
1958        if !claim_matches {
1959            return Err(invalid_resource(
1960                "certified physical backing does not cover the evaluated logical demand",
1961            ));
1962        }
1963        Ok(BoundBackingClaimCertificate {
1964            fingerprint: format!("{:x}", hasher.finalize()),
1965            physical_claim_count: self.physical_claim_count,
1966            has_shared_physical_claims: self.has_shared_physical_claims,
1967        })
1968    }
1969}
1970
1971#[doc(hidden)]
1972#[derive(Debug, PartialEq, Eq, Serialize)]
1973pub struct LogicalBackingSliceAllocationEvidence {
1974    pub(in crate::vnext::resource) domain_id: CapacityDomainId,
1975    pub(in crate::vnext::resource) pool_id: DynamicBackingPoolId,
1976    pub(in crate::vnext::resource) resource_id: ResourceId,
1977    pub(in crate::vnext::resource) pool_instance_id: u64,
1978    pub(in crate::vnext::resource) physical_claim_identity: PhysicalBackingClaimIdentity,
1979    #[serde(skip_serializing_if = "Option::is_none")]
1980    pub(in crate::vnext::resource) reusable_execution_bucket_id: Option<ReusableExecutionBucketId>,
1981    pub(in crate::vnext::resource) segment_generation: u64,
1982    pub(in crate::vnext::resource) segments: Vec<BackingSegment>,
1983    pub(in crate::vnext::resource) physical_offset_bytes: u64,
1984    pub(in crate::vnext::resource) capacity_size_bytes: u64,
1985    pub(in crate::vnext::resource) physical_size_bytes: u64,
1986    pub(in crate::vnext::resource) alignment_bytes: u64,
1987    pub(in crate::vnext::resource) usage: BufferUsage,
1988    pub(in crate::vnext::resource) element_type: ElementType,
1989    pub(in crate::vnext::resource) storage_profile: DynamicStorageProfile,
1990    pub(in crate::vnext::resource) initialization: StateInitialization,
1991    #[serde(skip)]
1992    pub(in crate::vnext::resource) fingerprint: String,
1993}
1994
1995#[derive(Debug, Clone, PartialEq, Eq)]
1996pub struct LogicalBackingSliceEvidence {
1997    pub(super) allocation: Arc<LogicalBackingSliceAllocationEvidence>,
1998    pub(in crate::vnext::resource) logical_size_bytes: u64,
1999}
2000
2001impl std::ops::Deref for LogicalBackingSliceEvidence {
2002    type Target = LogicalBackingSliceAllocationEvidence;
2003
2004    fn deref(&self) -> &Self::Target {
2005        self.allocation.as_ref()
2006    }
2007}
2008
2009impl Serialize for LogicalBackingSliceEvidence {
2010    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
2011    where
2012        S: serde::Serializer,
2013    {
2014        #[derive(Serialize)]
2015        struct Wire<'a> {
2016            domain_id: CapacityDomainId,
2017            pool_id: &'a DynamicBackingPoolId,
2018            resource_id: &'a ResourceId,
2019            pool_instance_id: u64,
2020            physical_claim_identity: &'a PhysicalBackingClaimIdentity,
2021            #[serde(skip_serializing_if = "Option::is_none")]
2022            reusable_execution_bucket_id: Option<&'a ReusableExecutionBucketId>,
2023            segment_generation: u64,
2024            segments: &'a [BackingSegment],
2025            physical_offset_bytes: u64,
2026            #[serde(rename = "size_bytes")]
2027            logical_size_bytes: u64,
2028            capacity_size_bytes: u64,
2029            physical_size_bytes: u64,
2030            alignment_bytes: u64,
2031            usage: BufferUsage,
2032            element_type: ElementType,
2033            storage_profile: DynamicStorageProfile,
2034            initialization: StateInitialization,
2035        }
2036
2037        Wire {
2038            domain_id: self.domain_id,
2039            pool_id: &self.pool_id,
2040            resource_id: &self.resource_id,
2041            pool_instance_id: self.pool_instance_id,
2042            physical_claim_identity: &self.physical_claim_identity,
2043            reusable_execution_bucket_id: self.reusable_execution_bucket_id.as_ref(),
2044            segment_generation: self.segment_generation,
2045            segments: &self.segments,
2046            physical_offset_bytes: self.physical_offset_bytes,
2047            logical_size_bytes: self.logical_size_bytes,
2048            capacity_size_bytes: self.capacity_size_bytes,
2049            physical_size_bytes: self.physical_size_bytes,
2050            alignment_bytes: self.alignment_bytes,
2051            usage: self.usage,
2052            element_type: self.element_type,
2053            storage_profile: self.storage_profile,
2054            initialization: self.initialization,
2055        }
2056        .serialize(serializer)
2057    }
2058}
2059
2060impl LogicalBackingSliceEvidence {
2061    pub fn domain_id(&self) -> CapacityDomainId {
2062        self.domain_id
2063    }
2064
2065    pub fn resource_id(&self) -> &ResourceId {
2066        &self.resource_id
2067    }
2068
2069    pub fn pool_id(&self) -> &DynamicBackingPoolId {
2070        &self.pool_id
2071    }
2072
2073    pub fn pool_instance_id(&self) -> u64 {
2074        self.pool_instance_id
2075    }
2076
2077    pub fn segment_generation(&self) -> u64 {
2078        self.segment_generation
2079    }
2080
2081    pub fn physical_claim_identity(&self) -> &PhysicalBackingClaimIdentity {
2082        &self.physical_claim_identity
2083    }
2084
2085    pub fn reusable_execution_bucket_id(&self) -> Option<&ReusableExecutionBucketId> {
2086        self.reusable_execution_bucket_id.as_ref()
2087    }
2088
2089    pub fn segments(&self) -> &[BackingSegment] {
2090        &self.segments
2091    }
2092
2093    pub fn physical_offset_bytes(&self) -> u64 {
2094        self.physical_offset_bytes
2095    }
2096
2097    pub const fn size_bytes(&self) -> u64 {
2098        self.logical_size_bytes
2099    }
2100
2101    pub fn capacity_size_bytes(&self) -> u64 {
2102        self.capacity_size_bytes
2103    }
2104
2105    pub fn physical_size_bytes(&self) -> u64 {
2106        self.physical_size_bytes
2107    }
2108
2109    pub fn alignment_bytes(&self) -> u64 {
2110        self.alignment_bytes
2111    }
2112
2113    pub fn usage(&self) -> BufferUsage {
2114        self.usage
2115    }
2116
2117    pub fn element_type(&self) -> ElementType {
2118        self.element_type
2119    }
2120
2121    pub fn storage_profile(&self) -> DynamicStorageProfile {
2122        self.storage_profile
2123    }
2124
2125    pub fn initialization(&self) -> StateInitialization {
2126        self.initialization
2127    }
2128}
2129
2130#[must_use = "a logical backing authority owns its physical arena extents"]
2131pub struct LogicalBackingSliceAuthority {
2132    pub(in crate::vnext::resource) evidence: LogicalBackingSliceEvidence,
2133    pub(in crate::vnext::resource) segment_lease: Arc<BackingSegmentLease>,
2134    pub(super) reusable_lane: Option<ExecutionLaneId>,
2135}
2136
2137impl LogicalBackingSliceAuthority {
2138    pub fn evidence(&self) -> &LogicalBackingSliceEvidence {
2139        &self.evidence
2140    }
2141
2142    pub(in crate::vnext::resource) fn retained(&self) -> Self {
2143        Self {
2144            evidence: self.evidence.clone(),
2145            segment_lease: Arc::clone(&self.segment_lease),
2146            reusable_lane: self.reusable_lane,
2147        }
2148    }
2149
2150    pub(in crate::vnext::resource) fn retained_for_lane(&self, lane_id: ExecutionLaneId) -> Self {
2151        Self {
2152            evidence: self.evidence.clone(),
2153            segment_lease: Arc::clone(&self.segment_lease),
2154            reusable_lane: Some(lane_id),
2155        }
2156    }
2157
2158    pub(crate) const fn reusable_address_scope(&self) -> Option<DeviceReusableAddressScope> {
2159        match self.reusable_lane {
2160            Some(lane_id) => Some(DeviceReusableAddressScope::ExecutionLane(lane_id)),
2161            None => None,
2162        }
2163    }
2164
2165    pub fn domain_id(&self) -> CapacityDomainId {
2166        self.evidence.domain_id
2167    }
2168
2169    pub fn resource_id(&self) -> &ResourceId {
2170        &self.evidence.resource_id
2171    }
2172
2173    pub const fn size_bytes(&self) -> u64 {
2174        self.evidence.logical_size_bytes
2175    }
2176
2177    pub fn capacity_size_bytes(&self) -> u64 {
2178        self.evidence.capacity_size_bytes
2179    }
2180
2181    pub fn initialization_status(&self) -> Result<Option<BackingInitializationStatus>, VNextError> {
2182        self.segment_lease
2183            .initialization
2184            .as_ref()
2185            .map(|cell| cell.status())
2186            .transpose()
2187    }
2188
2189    pub(in crate::vnext::resource) fn initialization_cell(
2190        &self,
2191    ) -> Option<&Arc<BackingInitializationCell>> {
2192        self.segment_lease.initialization.as_ref()
2193    }
2194}
2195
2196pub struct LogicalBackingBufferView<'a, B> {
2197    pub(in crate::vnext::resource) bindings: Vec<LogicalBackingSegmentBinding<B>>,
2198    pub(super) authorities: &'a [LogicalBackingSliceAuthority],
2199    pub(super) logical_size_bytes: u64,
2200    pub(super) capacity_size_bytes: u64,
2201    pub(super) alignment_bytes: u64,
2202    pub(super) usage: BufferUsage,
2203    pub(super) element_type: ElementType,
2204    pub(super) storage_profile: DynamicStorageProfile,
2205}
2206
2207pub(crate) struct LogicalBackingSegmentBinding<B> {
2208    pub(in crate::vnext::resource) segment: BackingSegment,
2209    pub(in crate::vnext::resource) chunk: Arc<ResidentChunkBacking<B>>,
2210    pub(in crate::vnext::resource) retention: DeviceBufferRetention,
2211}
2212
2213impl<B> LogicalBackingSegmentBinding<B> {
2214    pub(crate) fn segment(&self) -> &BackingSegment {
2215        &self.segment
2216    }
2217
2218    pub(crate) fn chunk(&self) -> &BackingChunkIdentity {
2219        self.segment.chunk()
2220    }
2221
2222    pub(crate) fn buffer(&self) -> &B {
2223        &self.chunk.buffer
2224    }
2225
2226    pub(crate) fn descriptor(&self) -> &BufferDescriptor {
2227        &self.chunk.descriptor
2228    }
2229
2230    pub(crate) fn retention(&self) -> DeviceBufferRetention {
2231        self.retention.clone()
2232    }
2233}
2234
2235impl<'a, B> LogicalBackingBufferView<'a, B> {
2236    pub(crate) fn segment_bindings(&self) -> &[LogicalBackingSegmentBinding<B>] {
2237        &self.bindings
2238    }
2239
2240    pub const fn size_bytes(&self) -> u64 {
2241        self.logical_size_bytes
2242    }
2243
2244    pub const fn capacity_size_bytes(&self) -> u64 {
2245        self.capacity_size_bytes
2246    }
2247
2248    pub const fn alignment_bytes(&self) -> u64 {
2249        self.alignment_bytes
2250    }
2251
2252    pub const fn usage(&self) -> BufferUsage {
2253        self.usage
2254    }
2255
2256    pub const fn element_type(&self) -> ElementType {
2257        self.element_type
2258    }
2259
2260    pub const fn storage_profile(&self) -> DynamicStorageProfile {
2261        self.storage_profile
2262    }
2263
2264    pub fn committed_evidence_segments(&self) -> impl Iterator<Item = &BackingSegment> {
2265        self.authorities
2266            .iter()
2267            .flat_map(|authority| authority.evidence.segments())
2268    }
2269
2270    /// Compatibility accessor for callers that construct a single-slice view.
2271    /// Multi-extent callers must use the aggregate metadata and segment iterator.
2272    pub fn slice(&self) -> &'a LogicalBackingSliceEvidence {
2273        &self
2274            .authorities
2275            .first()
2276            .expect("logical backing views always contain an authority")
2277            .evidence
2278    }
2279}
2280
2281impl<R> DynamicBackingPool<R>
2282where
2283    R: DeviceRuntime,
2284{
2285    pub(super) fn allocation_quantum(&self) -> u64 {
2286        match self.domain.pool.compatibility().profile().allocator() {
2287            DynamicStorageAllocator::LinearArena => {
2288                self.domain.pool.compatibility().alignment_bytes()
2289            }
2290            DynamicStorageAllocator::FixedBlockArena { block_bytes } => {
2291                block_bytes.max(self.domain.pool.compatibility().alignment_bytes())
2292            }
2293        }
2294    }
2295
2296    fn cancel_pending_growth(&self, bytes: u64) {
2297        let mut state = match self.state.lock() {
2298            Ok(state) => state,
2299            Err(poisoned) => poisoned.into_inner(),
2300        };
2301        if state.pending_growth_bytes < bytes {
2302            state.poisoned = true;
2303            return;
2304        }
2305        state.pending_growth_bytes -= bytes;
2306    }
2307
2308    fn rollback_prepared(
2309        &self,
2310        claim_identity: &PhysicalBackingClaimIdentity,
2311        occupancy: DynamicBackingClaimOccupancy,
2312        segments: &[BackingSegment],
2313    ) {
2314        let mut state = match self.state.lock() {
2315            Ok(state) => state,
2316            Err(poisoned) => {
2317                let mut state = poisoned.into_inner();
2318                state.poisoned = true;
2319                return;
2320            }
2321        };
2322        if state.poisoned {
2323            return;
2324        }
2325        let segment_count = u64::try_from(segments.len()).ok();
2326        let physical_bytes = segments.iter().try_fold(0_u64, |total, segment| {
2327            total.checked_add(segment.length_bytes())
2328        });
2329        if claim_identity.pool_id() != self.domain.pool_id()
2330            || Some(occupancy.segment_count) != segment_count
2331            || Some(occupancy.physical_bytes) != physical_bytes
2332        {
2333            state.poisoned = true;
2334            return;
2335        }
2336        let next_occupancy = match state.live_occupancy.checked_without_claim(occupancy) {
2337            Ok(next) => next,
2338            Err(_) => {
2339                state.poisoned = true;
2340                return;
2341            }
2342        };
2343        for segment in segments.iter().rev() {
2344            let valid = state
2345                .chunks
2346                .get(&segment.chunk_ordinal())
2347                .is_some_and(|chunk| {
2348                    chunk.backing.identity == *segment.chunk() && chunk.live_segments != 0
2349                });
2350            if !valid || state.allocator.release(segment).is_err() {
2351                state.poisoned = true;
2352                return;
2353            }
2354            state
2355                .chunks
2356                .get_mut(&segment.chunk_ordinal())
2357                .expect("validated prepared chunk remains installed")
2358                .live_segments -= 1;
2359        }
2360        state.live_occupancy = next_occupancy;
2361        drop(state);
2362        if self
2363            .logical_admission
2364            .notify_domain_availability_changed(self.domain.domain_id)
2365            .is_err()
2366        {
2367            let mut state = self
2368                .state
2369                .lock()
2370                .unwrap_or_else(std::sync::PoisonError::into_inner);
2371            state.poisoned = true;
2372        }
2373    }
2374}