Skip to main content

ferrum_interfaces/vnext/resource/
dynamic_pool_set.rs

1//! Dynamic pool-set orchestration over backing owned by `dynamic_pool`.
2
3use super::{
4    align_up_resource, backing_segment_range, bind_lane_stable_slot_projections,
5    compile_program_binding_layouts, compile_submission_wave_domain_layout,
6    compile_submission_wave_reusable_capacity_layouts, contiguous_packing_growth_bytes,
7    free_extent_layout_fingerprint, invalid_resource, lane_stable_layout_key,
8    rollback_free_extent_journal, validate_runtime_descriptor_for_admission,
9    AllocatedDynamicGrowth, AllocationLifetime, AllocationSeal, Arc, AtomicU64, BTreeMap,
10    BackingChunkIdentity, BackingClaimCertificate, BackingPrepareDecision, BackingSegment,
11    BufferRequest, CapacityAvailabilityEpoch, CapacityDomainId, CapacityEntry, CapacityEpochs,
12    CapacityUnits, CapacityVector, DeviceAllocationPermit, DeviceBufferRetention,
13    DeviceCapacityAvailabilitySnapshot, DeviceCapacityBudget, DeviceCapacityReservation,
14    DeviceRuntime, Digest, DynamicBackingBlocker, DynamicBackingClaimOccupancy,
15    DynamicBackingClaimResidency, DynamicBackingClaimScope, DynamicBackingDeferralReason,
16    DynamicBackingDeferred, DynamicBackingPackingEnvelope, DynamicBackingPool,
17    DynamicBackingPoolId, DynamicBackingPoolState, DynamicChunkQuarantineReason,
18    DynamicDeviceCapacityBlocked, DynamicPoolDomainSpec, DynamicPoolGrowthIntent,
19    DynamicPoolGrowthReceipt, DynamicPoolIdleReclaim, DynamicPoolLiveOccupancyStatus,
20    DynamicPoolMaintenanceBoundaryChunk, DynamicPoolMaintenanceBoundaryPool,
21    DynamicPoolMaintenanceBoundaryReceipt, DynamicPoolRebalanceReceipt, DynamicResourceShape,
22    DynamicStorageView, EvaluatedBackingRequest, ExecutionLane, FreeExtentIndex,
23    IdleChunkReclaimCandidate, InvocationLivenessMode, LaneBackingPrepareDecision,
24    LaneStableArenaEntry, LaneStableArenaEvictionCandidate, LaneStableArenaLane,
25    LaneStableArenaSlot, LaneStableArenaSlotLease, LaneStableArenaState,
26    LogicalAdmissionCoordinator, LogicalAdmissionCoordinatorId, LogicalBackingBufferView,
27    LogicalBackingSegmentBinding, LogicalBackingSliceAllocationEvidence,
28    LogicalBackingSliceAuthority, LogicalBackingSliceEvidence, Mutex, Ordering, PendingGrowthGuard,
29    PlanNode, PlannedDynamicGrowth, PreparedBackingClaim, PreparedBackingExtent,
30    PreparedLaneBackingClaim, ProgramBindingLayout, QuarantinedDynamicChunk,
31    RequestStateHazardCoordinator, ResidentChunkBacking, ResidentChunkState, ResourceId,
32    ResourceReservation, ResourceRetentionPolicy, ResourceTransactionIdentity, RunId, Serialize,
33    Sha256, StateInitialization, StaticProvisioningBinding, StepResourceSlotKind,
34    SubmissionWaveDomainCapacityLayout, SubmissionWaveDomainLayout, TransactionId, VNextError,
35    DYNAMIC_POOL_MAINTENANCE_BOUNDARY_SCHEMA_VERSION, NEXT_DYNAMIC_POOL_INSTANCE_ID,
36};
37use crate::vnext::{
38    DeviceCapacityPressure, DynamicPoolResidentPressure, ReusableExecutionBucketId,
39    ReusableExecutionMemoryPlan,
40};
41
42#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
43pub struct DynamicPoolGrowthBatchReceipt {
44    pub(super) coordinator_id: LogicalAdmissionCoordinatorId,
45    pub(super) growths: Vec<DynamicPoolGrowthReceipt>,
46    pub(super) capacity_epoch: u64,
47    #[serde(skip_serializing_if = "Option::is_none")]
48    pub(super) rebalance: Option<DynamicPoolRebalanceReceipt>,
49    #[serde(skip_serializing_if = "Option::is_none")]
50    pub(super) maintenance_boundary: Option<DynamicPoolMaintenanceBoundaryReceipt>,
51}
52
53impl DynamicPoolGrowthBatchReceipt {
54    pub const fn coordinator_id(&self) -> LogicalAdmissionCoordinatorId {
55        self.coordinator_id
56    }
57
58    pub fn growths(&self) -> &[DynamicPoolGrowthReceipt] {
59        &self.growths
60    }
61
62    pub const fn capacity_epoch(&self) -> u64 {
63        self.capacity_epoch
64    }
65
66    pub const fn rebalance(&self) -> Option<&DynamicPoolRebalanceReceipt> {
67        self.rebalance.as_ref()
68    }
69
70    pub const fn maintenance_boundary(&self) -> Option<&DynamicPoolMaintenanceBoundaryReceipt> {
71        self.maintenance_boundary.as_ref()
72    }
73}
74
75pub(in crate::vnext::resource) struct DynamicPoolSet<R>
76where
77    R: DeviceRuntime,
78{
79    pub(in crate::vnext::resource) pools:
80        BTreeMap<DynamicBackingPoolId, Arc<DynamicBackingPool<R>>>,
81    pub(in crate::vnext::resource) domains: Vec<DynamicPoolDomainSpec>,
82    pub(in crate::vnext::resource) nodes: Arc<[PlanNode]>,
83    pub(in crate::vnext::resource) submission_wave_layouts: Vec<Option<SubmissionWaveDomainLayout>>,
84    pub(in crate::vnext::resource) submission_wave_reusable_capacity_layouts:
85        BTreeMap<ReusableExecutionBucketId, Vec<Option<SubmissionWaveDomainCapacityLayout>>>,
86    pub(in crate::vnext::resource) program_binding_layouts:
87        BTreeMap<ReusableExecutionBucketId, Arc<ProgramBindingLayout>>,
88    pub(in crate::vnext::resource) reusable_execution: Option<ReusableExecutionMemoryPlan>,
89    pub(in crate::vnext::resource) request_state_hazards: Arc<RequestStateHazardCoordinator>,
90    pub(in crate::vnext::resource) logical_admission: LogicalAdmissionCoordinator,
91    pub(in crate::vnext::resource) budget: Arc<DeviceCapacityBudget>,
92    lane_stable_arenas: Arc<Mutex<LaneStableArenaState>>,
93    binding: StaticProvisioningBinding,
94    // Backend context must outlive every resident/quarantined buffer above.
95    runtime: Arc<R>,
96}
97
98pub(in crate::vnext::resource) struct DynamicPoolRebalanceAttempt {
99    pub(in crate::vnext::resource) boundary: DynamicPoolMaintenanceBoundaryReceipt,
100    pub(in crate::vnext::resource) rebalance: Option<DynamicPoolRebalanceReceipt>,
101}
102
103impl<R> DynamicPoolSet<R>
104where
105    R: DeviceRuntime,
106{
107    pub(in crate::vnext::resource) fn new(
108        runtime: Arc<R>,
109        binding: StaticProvisioningBinding,
110        budget: Arc<DeviceCapacityBudget>,
111        logical_admission: LogicalAdmissionCoordinator,
112        domains: Vec<DynamicPoolDomainSpec>,
113        nodes: Arc<[PlanNode]>,
114        reusable_execution: Option<ReusableExecutionMemoryPlan>,
115    ) -> Result<Self, VNextError> {
116        let request_state_hazards = RequestStateHazardCoordinator::compile(&nodes)?;
117        let submission_wave_layouts = domains
118            .iter()
119            .map(|domain| compile_submission_wave_domain_layout(domain, &nodes))
120            .collect::<Result<Vec<_>, _>>()?;
121        let submission_wave_reusable_capacity_layouts =
122            compile_submission_wave_reusable_capacity_layouts(
123                &domains,
124                &submission_wave_layouts,
125                reusable_execution.as_ref(),
126            )?;
127        let program_binding_layouts = compile_program_binding_layouts(
128            &domains,
129            &nodes,
130            &submission_wave_layouts,
131            &submission_wave_reusable_capacity_layouts,
132        )?
133        .into_iter()
134        .map(|(bucket_id, layout)| (bucket_id, Arc::new(layout)))
135        .collect();
136        let mut pools = BTreeMap::new();
137        for domain in &domains {
138            let instance_id = NEXT_DYNAMIC_POOL_INSTANCE_ID
139                .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
140                    current.checked_add(1)
141                })
142                .map_err(|_| invalid_resource("dynamic pool instance id space is exhausted"))?;
143            let pool = Arc::new(DynamicBackingPool {
144                instance_id,
145                domain: domain.clone(),
146                logical_admission: logical_admission.clone(),
147                maintenance: Mutex::new(()),
148                next_extent_generation: AtomicU64::new(1),
149                state: Mutex::new(DynamicBackingPoolState {
150                    resident_bytes: 0,
151                    pending_growth_bytes: 0,
152                    next_chunk_ordinal: 1,
153                    next_chunk_generation: 1,
154                    chunks: BTreeMap::new(),
155                    allocator: FreeExtentIndex::default(),
156                    live_occupancy: DynamicPoolLiveOccupancyStatus::default(),
157                    quarantined: Vec::new(),
158                    poisoned: false,
159                }),
160            });
161            if pools.insert(domain.pool_id().clone(), pool).is_some() {
162                return Err(invalid_resource(
163                    "dynamic pool set contains a duplicate pool",
164                ));
165            }
166        }
167        Ok(Self {
168            runtime,
169            binding,
170            budget,
171            logical_admission,
172            domains,
173            pools,
174            nodes,
175            submission_wave_layouts,
176            submission_wave_reusable_capacity_layouts,
177            program_binding_layouts,
178            reusable_execution,
179            request_state_hazards,
180            lane_stable_arenas: Arc::new(Mutex::new(LaneStableArenaState::default())),
181        })
182    }
183
184    pub(in crate::vnext::resource) fn program_binding_layout(
185        &self,
186        bucket_id: &ReusableExecutionBucketId,
187    ) -> Option<&Arc<ProgramBindingLayout>> {
188        self.program_binding_layouts.get(bucket_id)
189    }
190
191    pub(in crate::vnext::resource) const fn maximum_active_sequences(&self) -> u32 {
192        self.binding.maximum_active_sequences()
193    }
194
195    pub(in crate::vnext::resource) fn write_capacity_availability(
196        &self,
197        out: &mut Vec<CapacityAvailabilityEpoch>,
198    ) -> Result<CapacityEpochs, VNextError> {
199        let epochs = self.logical_admission.write_availability_epochs(out)?;
200        self.budget.write_availability_epochs(out)?;
201        debug_assert!(out
202            .windows(2)
203            .all(|pair| pair[0].source() < pair[1].source()));
204        Ok(epochs)
205    }
206
207    /// Rebalances only whole, unreferenced chunks from non-target pools. The
208    /// batch is selected before mutation, logical totals publish atomically,
209    /// and physical grants are returned only after every pool lock is dropped.
210    pub(in crate::vnext::resource) fn reclaim_idle_chunks_for_pressure(
211        &self,
212        pressure: &DeviceCapacityPressure,
213        capacity_availability: DeviceCapacityAvailabilitySnapshot,
214        excluded_domains: &[CapacityDomainId],
215        protected_immediate: &CapacityVector,
216        protected_packing_envelopes: &[DynamicBackingPackingEnvelope],
217    ) -> Result<DynamicPoolRebalanceAttempt, VNextError> {
218        if pressure.device_id() != self.runtime.descriptor().id.to_string() {
219            return Err(invalid_resource(
220                "dynamic pool rebalance received pressure for another device",
221            ));
222        }
223        let deficit = pressure
224            .requested_bytes()
225            .checked_sub(pressure.available_bytes())
226            .ok_or_else(|| invalid_resource("dynamic pool pressure has no reclaimable deficit"))?;
227        if deficit == 0 {
228            return Err(invalid_resource(
229                "dynamic pool pressure has an empty reclaimable deficit",
230            ));
231        }
232
233        let excluded_domains = excluded_domains
234            .iter()
235            .copied()
236            .collect::<std::collections::BTreeSet<_>>();
237        let pools = self.pools.values().cloned().collect::<Vec<_>>();
238        let maintenance = pools
239            .iter()
240            .map(|pool| {
241                pool.maintenance
242                    .lock()
243                    .map_err(|_| invalid_resource("dynamic pool maintenance authority is poisoned"))
244            })
245            .collect::<Result<Vec<_>, _>>()?;
246        let mut states = pools
247            .iter()
248            .map(|pool| {
249                pool.state
250                    .lock()
251                    .map_err(|_| invalid_resource("dynamic backing pool is poisoned"))
252            })
253            .collect::<Result<Vec<_>, _>>()?;
254        let logical = self.logical_admission.snapshot()?;
255        let used_by_domain = logical
256            .domains()
257            .iter()
258            .map(|domain| (domain.domain(), domain.used().get()))
259            .collect::<BTreeMap<_, _>>();
260        let protected_by_domain = protected_immediate
261            .entries()
262            .iter()
263            .map(|entry| (entry.domain(), entry.units().get()))
264            .collect::<BTreeMap<_, _>>();
265        let protected_packing_by_domain = protected_packing_envelopes
266            .iter()
267            .map(|envelope| (envelope.domain_id(), envelope))
268            .collect::<BTreeMap<_, _>>();
269        if protected_packing_by_domain.len() != protected_packing_envelopes.len() {
270            return Err(invalid_resource(
271                "dynamic backing protection contains duplicate packing domains",
272            ));
273        }
274
275        let mut candidates = Vec::new();
276        let mut boundary_pools = Vec::with_capacity(pools.len());
277        let mut reclaimable_by_pool = vec![0_u64; pools.len()];
278        for (pool_index, (pool, state)) in pools.iter().zip(states.iter()).enumerate() {
279            if state.poisoned {
280                return Err(invalid_resource("dynamic backing pool is fail-closed"));
281            }
282            let excluded_from_reclaim = excluded_domains.contains(&pool.domain.domain_id);
283            let used = used_by_domain
284                .get(&pool.domain.domain_id)
285                .copied()
286                .ok_or_else(|| invalid_resource("dynamic pool domain is absent from admission"))?;
287            let physically_occupied = state
288                .resident_bytes
289                .checked_sub(state.allocator.free_bytes)
290                .ok_or_else(|| {
291                    invalid_resource("dynamic pool free bytes exceed resident capacity")
292                })?;
293            if physically_occupied != state.live_occupancy.total().physical_bytes() {
294                return Err(invalid_resource(
295                    "dynamic pool allocator occupancy differs from its typed live occupancy",
296                ));
297            }
298            let protected = protected_by_domain
299                .get(&pool.domain.domain_id)
300                .copied()
301                .unwrap_or(0);
302            let mut protected_packing_satisfied = true;
303            let protected_chunks = match protected_packing_by_domain.get(&pool.domain.domain_id) {
304                Some(envelope) => {
305                    if envelope.pool_id() != pool.domain.pool_id()
306                        || envelope.total_bytes()? != protected
307                    {
308                        return Err(invalid_resource(
309                            "dynamic backing byte and packing protection diverged",
310                        ));
311                    }
312                    let mut allocator = state.allocator.clone();
313                    let mut chunks = std::collections::BTreeSet::new();
314                    for &claim_bytes in envelope.claim_bytes_descending() {
315                        let segments = match pool.domain.pool.compatibility().profile().view() {
316                            DynamicStorageView::Contiguous => allocator
317                                .allocate_contiguous(pool.domain.pool_id(), claim_bytes)?
318                                .map(|segment| vec![segment]),
319                            DynamicStorageView::PagedRegions { block_bytes } => allocator
320                                .allocate_paged(pool.domain.pool_id(), claim_bytes, block_bytes)?,
321                        };
322                        let Some(segments) = segments else {
323                            chunks.clear();
324                            break;
325                        };
326                        chunks.extend(segments.iter().map(BackingSegment::chunk_ordinal));
327                    }
328                    if chunks.is_empty() && protected != 0 {
329                        protected_packing_satisfied = false;
330                    }
331                    chunks
332                }
333                // Logical admission deferrals protect aggregate capacity before
334                // an exact physical packing attempt exists.
335                None => std::collections::BTreeSet::new(),
336            };
337            // Logical admission does not own lane-stable or not-yet-committed
338            // physical extents. Reclaim must preserve whichever ownership view
339            // is larger before adding this bundle's uncommitted demand.
340            let owned = used.max(physically_occupied);
341            let coherent_runnable_floor = owned.checked_add(protected).ok_or_else(|| {
342                invalid_resource("dynamic pool protected runnable floor overflows u64")
343            })?;
344            let resident_floor = pool
345                .domain
346                .pool
347                .provisioning()
348                .minimum_resident_bytes()
349                .max(coherent_runnable_floor);
350            let reclaimable = state.resident_bytes.saturating_sub(resident_floor);
351            reclaimable_by_pool[pool_index] = reclaimable;
352            let mut boundary_chunks = Vec::with_capacity(state.chunks.len());
353            for (&ordinal, chunk) in &state.chunks {
354                let chunk_bytes = chunk.backing._grant.bytes();
355                let full_extent = state.allocator.by_offset.get(&(ordinal, 0));
356                let external_references = Arc::strong_count(&chunk.backing).saturating_sub(1);
357                let protected_packing = protected_chunks.contains(&ordinal);
358                let full_extent_available = chunk.backing.descriptor.size_bytes == chunk_bytes
359                    && full_extent.is_some_and(|extent| {
360                        extent.chunk_generation == chunk.backing.identity.generation()
361                            && extent.length_bytes == chunk_bytes
362                    });
363                let resident_floor_allows_reclaim = chunk_bytes <= reclaimable;
364                let reclaim_candidate = !excluded_from_reclaim
365                    && state.pending_growth_bytes == 0
366                    && protected_packing_satisfied
367                    && chunk.live_segments == 0
368                    && external_references == 0
369                    && !protected_packing
370                    && resident_floor_allows_reclaim
371                    && full_extent_available;
372                boundary_chunks.push(DynamicPoolMaintenanceBoundaryChunk {
373                    identity: chunk.backing.identity.clone(),
374                    bytes: chunk_bytes,
375                    live_segments: chunk.live_segments,
376                    external_references,
377                    protected_packing,
378                    full_extent_available,
379                    resident_floor_allows_reclaim,
380                    reclaim_candidate,
381                });
382                if reclaim_candidate {
383                    candidates.push(IdleChunkReclaimCandidate {
384                        pool_index,
385                        chunk: chunk.backing.identity.clone(),
386                        chunk_bytes,
387                    });
388                }
389            }
390            boundary_pools.push(DynamicPoolMaintenanceBoundaryPool {
391                pool_id: pool.domain.pool_id().clone(),
392                domain_id: pool.domain.domain_id,
393                excluded_from_reclaim,
394                resident_bytes: state.resident_bytes,
395                pending_growth_bytes: state.pending_growth_bytes,
396                free_bytes: state.allocator.free_bytes,
397                largest_contiguous_bytes: state.allocator.largest_contiguous_bytes(),
398                free_extent_layout_fingerprint: free_extent_layout_fingerprint(&state.allocator),
399                logical_used_bytes: used,
400                live_occupancy: state.live_occupancy,
401                minimum_resident_bytes: pool.domain.pool.provisioning().minimum_resident_bytes(),
402                maximum_resident_bytes: pool.domain.pool.provisioning().maximum_resident_bytes(),
403                protected_immediate_bytes: protected,
404                protected_packing_satisfied,
405                coherent_runnable_floor_bytes: coherent_runnable_floor,
406                resident_floor_bytes: resident_floor,
407                reclaimable_bytes: reclaimable,
408                chunks: boundary_chunks,
409            });
410        }
411
412        let reclaim_candidate_chunks = candidates.len();
413        let reclaim_candidate_bytes = candidates.iter().try_fold(0_u64, |total, candidate| {
414            total
415                .checked_add(candidate.chunk_bytes)
416                .ok_or_else(|| invalid_resource("dynamic reclaim candidate bytes overflow u64"))
417        })?;
418        let mut selected = Vec::<IdleChunkReclaimCandidate>::new();
419        let mut selected_by_pool = vec![0_u64; pools.len()];
420        let mut reclaimed_bytes = 0_u64;
421        let best_single = candidates
422            .iter()
423            .filter(|candidate| candidate.chunk_bytes >= deficit)
424            .min_by(|left, right| {
425                left.chunk_bytes
426                    .cmp(&right.chunk_bytes)
427                    .then_with(|| left.pool_index.cmp(&right.pool_index))
428                    .then_with(|| right.chunk.ordinal().cmp(&left.chunk.ordinal()))
429            })
430            .cloned();
431        if let Some(candidate) = best_single {
432            selected_by_pool[candidate.pool_index] = candidate.chunk_bytes;
433            reclaimed_bytes = candidate.chunk_bytes;
434            selected.push(candidate);
435        } else {
436            candidates.sort_by(|left, right| {
437                right
438                    .chunk_bytes
439                    .cmp(&left.chunk_bytes)
440                    .then_with(|| left.pool_index.cmp(&right.pool_index))
441                    .then_with(|| right.chunk.ordinal().cmp(&left.chunk.ordinal()))
442            });
443            for candidate in candidates {
444                let next_pool_total = selected_by_pool[candidate.pool_index]
445                    .checked_add(candidate.chunk_bytes)
446                    .ok_or_else(|| invalid_resource("dynamic reclaim bytes overflow u64"))?;
447                if next_pool_total > reclaimable_by_pool[candidate.pool_index] {
448                    continue;
449                }
450                selected_by_pool[candidate.pool_index] = next_pool_total;
451                reclaimed_bytes = reclaimed_bytes
452                    .checked_add(candidate.chunk_bytes)
453                    .ok_or_else(|| invalid_resource("dynamic reclaim bytes overflow u64"))?;
454                selected.push(candidate);
455                if reclaimed_bytes >= deficit {
456                    break;
457                }
458            }
459        }
460        selected.sort_by(|left, right| {
461            left.pool_index
462                .cmp(&right.pool_index)
463                .then_with(|| left.chunk.ordinal().cmp(&right.chunk.ordinal()))
464        });
465        let mut planned_domains = excluded_domains.iter().copied().collect::<Vec<_>>();
466        planned_domains.sort_unstable();
467        let mut protected_packing_envelopes = protected_packing_envelopes.to_vec();
468        protected_packing_envelopes.sort_by_key(DynamicBackingPackingEnvelope::domain_id);
469        let boundary = DynamicPoolMaintenanceBoundaryReceipt {
470            schema_version: DYNAMIC_POOL_MAINTENANCE_BOUNDARY_SCHEMA_VERSION,
471            coordinator_id: logical.coordinator_id(),
472            logical_release_epoch: logical.release_epoch(),
473            logical_capacity_epoch: logical.capacity_epoch(),
474            plan_device_capacity_epoch: capacity_availability.plan_epoch(),
475            process_device_capacity_epoch: capacity_availability.process_epoch(),
476            pressure: pressure.clone(),
477            planned_domains,
478            protected_immediate: protected_immediate.clone(),
479            protected_packing_envelopes,
480            pools: boundary_pools,
481            reclaim_candidate_chunks,
482            reclaim_candidate_bytes,
483            selected_chunks: selected
484                .iter()
485                .map(|candidate| candidate.chunk.clone())
486                .collect(),
487            selected_bytes: reclaimed_bytes,
488            reclaim_sufficient: reclaimed_bytes >= deficit,
489        };
490        if reclaimed_bytes < deficit {
491            return Ok(DynamicPoolRebalanceAttempt {
492                boundary,
493                rebalance: None,
494            });
495        }
496
497        for candidate in &selected {
498            let state = &states[candidate.pool_index];
499            let chunk = state
500                .chunks
501                .get(&candidate.chunk.ordinal())
502                .ok_or_else(|| invalid_resource("selected dynamic reclaim chunk disappeared"))?;
503            let extent = state
504                .allocator
505                .by_offset
506                .get(&(candidate.chunk.ordinal(), 0))
507                .ok_or_else(|| invalid_resource("selected dynamic reclaim extent disappeared"))?;
508            if chunk.live_segments != 0
509                || Arc::strong_count(&chunk.backing) != 1
510                || chunk.backing.identity != candidate.chunk
511                || extent.chunk_generation != candidate.chunk.generation()
512                || extent.length_bytes != candidate.chunk_bytes
513            {
514                return Err(invalid_resource(
515                    "selected dynamic reclaim chunk changed before publication",
516                ));
517            }
518        }
519
520        let published_totals = states
521            .iter()
522            .zip(&selected_by_pool)
523            .map(|(state, &selected_bytes)| {
524                state
525                    .resident_bytes
526                    .checked_sub(selected_bytes)
527                    .ok_or_else(|| invalid_resource("dynamic reclaim resident bytes underflow"))
528            })
529            .collect::<Result<Vec<_>, _>>()?;
530        let mut removed = Vec::with_capacity(selected.len());
531        for candidate in &selected {
532            let state = &mut states[candidate.pool_index];
533            let extent = state
534                .allocator
535                .remove_extent(candidate.chunk.ordinal(), 0)
536                .expect("validated idle chunk retains its exact full extent");
537            debug_assert_eq!(extent.chunk_generation, candidate.chunk.generation());
538            debug_assert_eq!(extent.length_bytes, candidate.chunk_bytes);
539            let chunk = state
540                .chunks
541                .remove(&candidate.chunk.ordinal())
542                .expect("validated idle chunk remains resident");
543            removed.push((candidate.clone(), chunk));
544        }
545        let updates = selected_by_pool
546            .iter()
547            .enumerate()
548            .filter(|(_, selected_bytes)| **selected_bytes != 0)
549            .map(|(pool_index, _)| {
550                (
551                    pools[pool_index].domain.domain_id,
552                    CapacityUnits::new(published_totals[pool_index]),
553                )
554            })
555            .collect::<Vec<_>>();
556        let epochs = match self.logical_admission.set_domain_totals(&updates) {
557            Ok(epochs) => epochs,
558            Err(error) => {
559                for (candidate, chunk) in removed.drain(..).rev() {
560                    let state = &mut states[candidate.pool_index];
561                    state
562                        .allocator
563                        .insert_extent(
564                            candidate.chunk.ordinal(),
565                            candidate.chunk.generation(),
566                            0,
567                            candidate.chunk_bytes,
568                        )
569                        .expect("unpublished idle chunk extent can be restored");
570                    assert!(state
571                        .chunks
572                        .insert(candidate.chunk.ordinal(), chunk)
573                        .is_none());
574                }
575                return Err(error);
576            }
577        };
578        for (state, &published_total) in states.iter_mut().zip(&published_totals) {
579            state.resident_bytes = published_total;
580        }
581
582        let mut pool_receipts = Vec::new();
583        for (pool_index, &pool_reclaimed_bytes) in selected_by_pool.iter().enumerate() {
584            if pool_reclaimed_bytes == 0 {
585                continue;
586            }
587            pool_receipts.push(DynamicPoolIdleReclaim {
588                pool_id: pools[pool_index].domain.pool_id().clone(),
589                chunks: selected
590                    .iter()
591                    .filter(|candidate| candidate.pool_index == pool_index)
592                    .map(|candidate| candidate.chunk.clone())
593                    .collect(),
594                reclaimed_bytes: pool_reclaimed_bytes,
595                published_capacity_bytes: published_totals[pool_index],
596            });
597        }
598        let reclaimed_chunks = selected.len();
599        drop(states);
600        drop(maintenance);
601        drop(removed);
602        let availability = self.budget.availability_snapshot()?;
603        Ok(DynamicPoolRebalanceAttempt {
604            boundary,
605            rebalance: Some(DynamicPoolRebalanceReceipt {
606                pools: pool_receipts,
607                reclaimed_chunks,
608                reclaimed_bytes,
609                logical_capacity_epoch: epochs.capacity_epoch(),
610                plan_device_capacity_epoch: availability.plan_epoch(),
611                process_device_capacity_epoch: availability.process_epoch(),
612            }),
613        })
614    }
615
616    pub(in crate::vnext::resource) fn maintain_pools(
617        &self,
618        intents: Vec<DynamicPoolGrowthIntent>,
619    ) -> Result<DynamicPoolGrowthBatchReceipt, VNextError> {
620        let mut ignored_capacity_block = None;
621        self.maintain_pools_observed(intents, &mut ignored_capacity_block)
622    }
623
624    pub(in crate::vnext::resource) fn maintain_pools_observed(
625        &self,
626        mut intents: Vec<DynamicPoolGrowthIntent>,
627        capacity_blocked: &mut Option<DynamicDeviceCapacityBlocked>,
628    ) -> Result<DynamicPoolGrowthBatchReceipt, VNextError> {
629        intents.sort_by(|left, right| left.pool_id().cmp(right.pool_id()));
630        if intents
631            .windows(2)
632            .any(|pair| pair[0].pool_id() == pair[1].pool_id())
633        {
634            return Err(invalid_resource(
635                "dynamic maintenance batch contains a duplicate pool",
636            ));
637        }
638        let pools = intents
639            .iter()
640            .map(|intent| {
641                self.pools.get(intent.pool_id()).cloned().ok_or_else(|| {
642                    invalid_resource("dynamic maintenance references an unknown pool")
643                })
644            })
645            .collect::<Result<Vec<_>, _>>()?;
646        let _maintenance = pools
647            .iter()
648            .map(|pool| {
649                pool.maintenance
650                    .lock()
651                    .map_err(|_| invalid_resource("dynamic pool maintenance authority is poisoned"))
652            })
653            .collect::<Result<Vec<_>, _>>()?;
654
655        let mut planned = Vec::with_capacity(intents.len());
656        let mut pending = Vec::with_capacity(intents.len());
657        for (intent, pool) in intents.iter().zip(&pools) {
658            let requested_bytes = {
659                let state = pool
660                    .state
661                    .lock()
662                    .map_err(|_| invalid_resource("dynamic backing pool is poisoned"))?;
663                if state.poisoned {
664                    return Err(invalid_resource("dynamic backing pool is fail-closed"));
665                }
666                match intent {
667                    DynamicPoolGrowthIntent::Additional(request) => request.requested_bytes(),
668                    DynamicPoolGrowthIntent::Minimum(_) => {
669                        let current = state
670                            .resident_bytes
671                            .checked_add(state.pending_growth_bytes)
672                            .ok_or_else(|| {
673                                invalid_resource("dynamic pool residency overflows u64")
674                            })?;
675                        let minimum = pool.domain.pool.provisioning().minimum_resident_bytes();
676                        if current >= minimum {
677                            continue;
678                        }
679                        minimum - current
680                    }
681                    DynamicPoolGrowthIntent::RevalidatedDeferral(blocker) => {
682                        if let Some(claim_bytes) = blocker.contiguous_claim_bytes_descending() {
683                            let required_growth = contiguous_packing_growth_bytes(
684                                &state.allocator,
685                                pool.domain.pool_id(),
686                                claim_bytes,
687                            )?;
688                            if required_growth == 0 {
689                                continue;
690                            }
691                            required_growth
692                        } else {
693                            match blocker.reason() {
694                                DynamicBackingDeferralReason::GrowthRequired => {
695                                    let required_free = blocker
696                                        .free_bytes()
697                                        .checked_add(blocker.requested_bytes())
698                                        .ok_or_else(|| {
699                                            invalid_resource(
700                                            "dynamic backing deferred requirement overflows u64",
701                                        )
702                                        })?;
703                                    if state.allocator.free_bytes >= required_free {
704                                        continue;
705                                    }
706                                    required_free - state.allocator.free_bytes
707                                }
708                                DynamicBackingDeferralReason::FragmentedContiguous => {
709                                    return Err(invalid_resource(
710                                        "fragmented contiguous blocker lost its transaction demand",
711                                    ));
712                                }
713                            }
714                        }
715                    }
716                }
717            };
718            let chunk_bytes = align_up_resource(requested_bytes, pool.allocation_quantum())?;
719            let chunk = {
720                let mut state = pool
721                    .state
722                    .lock()
723                    .map_err(|_| invalid_resource("dynamic backing pool is poisoned"))?;
724                if state.poisoned {
725                    return Err(invalid_resource("dynamic backing pool is fail-closed"));
726                }
727                let ordinal = state.next_chunk_ordinal;
728                let generation = state.next_chunk_generation;
729                let next_ordinal = ordinal
730                    .checked_add(1)
731                    .ok_or_else(|| invalid_resource("dynamic chunk ordinal space is exhausted"))?;
732                let next_generation = generation.checked_add(1).ok_or_else(|| {
733                    invalid_resource("dynamic chunk generation space is exhausted")
734                })?;
735                let next_pending = state
736                    .pending_growth_bytes
737                    .checked_add(chunk_bytes)
738                    .ok_or_else(|| invalid_resource("pending dynamic growth bytes overflow u64"))?;
739                state.next_chunk_ordinal = next_ordinal;
740                state.next_chunk_generation = next_generation;
741                state.pending_growth_bytes = next_pending;
742                BackingChunkIdentity::from_parts(
743                    pool.domain.pool_id().clone(),
744                    ordinal,
745                    generation,
746                )?
747            };
748            pending.push(PendingGrowthGuard {
749                pool: Arc::clone(pool),
750                bytes: chunk_bytes,
751                armed: true,
752            });
753            let expected_resource_id = ResourceId::new(format!(
754                "{}/chunk/{}/{}",
755                pool.domain.pool_id().as_str(),
756                chunk.ordinal(),
757                chunk.generation()
758            ))?;
759            planned.push(PlannedDynamicGrowth {
760                pool: Arc::clone(pool),
761                chunk,
762                expected_resource_id,
763                chunk_bytes,
764            });
765        }
766        if planned.is_empty() {
767            return Ok(DynamicPoolGrowthBatchReceipt {
768                coordinator_id: self.logical_admission.id(),
769                growths: Vec::new(),
770                capacity_epoch: self.logical_admission.epochs()?.capacity_epoch(),
771                rebalance: None,
772                maintenance_boundary: None,
773            });
774        }
775
776        let total_bytes = planned.iter().try_fold(0_u64, |total, growth| {
777            total
778                .checked_add(growth.chunk_bytes)
779                .ok_or_else(|| invalid_resource("dynamic maintenance batch bytes overflow u64"))
780        })?;
781        let capacity_availability = self.budget.availability_snapshot()?;
782        let reservation = match DeviceCapacityReservation::reserve(&self.budget, total_bytes) {
783            Ok(reservation) => reservation,
784            Err(VNextError::DeviceCapacityUnavailable(pressure)) => {
785                *capacity_blocked = Some(DynamicDeviceCapacityBlocked {
786                    pressure: pressure.clone(),
787                    availability: capacity_availability,
788                    planned_domains: planned
789                        .iter()
790                        .map(|growth| growth.pool.domain.domain_id)
791                        .collect(),
792                });
793                return Err(VNextError::DeviceCapacityUnavailable(pressure));
794            }
795            Err(error) => return Err(error),
796        };
797        // Device-budget saturation is recoverable pressure even when the same
798        // growth also crosses a pool's device-derived resident ceiling. Only a
799        // growth that the authoritative device budget accepted can prove that
800        // the remaining pool ceiling is a terminal theoretical-plan violation.
801        for growth in &planned {
802            let state = growth
803                .pool
804                .state
805                .lock()
806                .map_err(|_| invalid_resource("dynamic backing pool is poisoned"))?;
807            let next_residency = state
808                .resident_bytes
809                .checked_add(state.pending_growth_bytes)
810                .ok_or_else(|| invalid_resource("dynamic pool resident bytes overflow u64"))?;
811            if next_residency
812                > growth
813                    .pool
814                    .domain
815                    .pool
816                    .provisioning()
817                    .maximum_resident_bytes()
818            {
819                return Err(VNextError::DynamicPoolResidentUnavailable(
820                    DynamicPoolResidentPressure::new(
821                        growth.pool.domain.pool_id().clone(),
822                        growth.chunk_bytes,
823                        state.resident_bytes,
824                        growth
825                            .pool
826                            .domain
827                            .pool
828                            .provisioning()
829                            .maximum_resident_bytes(),
830                    )?,
831                ));
832            }
833        }
834        let grants = reservation.commit_split(
835            &planned
836                .iter()
837                .map(|growth| growth.chunk_bytes)
838                .collect::<Vec<_>>(),
839        )?;
840        let mut allocated = Vec::with_capacity(planned.len());
841        for (growth, grant) in planned.iter().zip(grants) {
842            let transaction_identity = ResourceTransactionIdentity {
843                pool_id: self.binding.pool_id(),
844                run_id: RunId::new(format!("dynamic-grow-{}", growth.chunk.generation()))?,
845                transaction_id: TransactionId::new(format!(
846                    "dynamic-grow-{}-{}",
847                    growth.chunk.ordinal(),
848                    growth.chunk.generation()
849                ))?,
850                request_id: self.binding.request_id().clone(),
851            };
852            let reservation_evidence = ResourceReservation {
853                resource_id: growth.expected_resource_id.clone(),
854                request_id: self.binding.request_id().clone(),
855                owner_node_id: None,
856                size_bytes: growth.chunk_bytes,
857                alignment_bytes: growth.pool.domain.pool.compatibility().alignment_bytes(),
858                usage: growth.pool.domain.pool.compatibility().usage(),
859                element_type: growth.pool.domain.pool.compatibility().element_type(),
860                retention_policy: ResourceRetentionPolicy::Plan,
861                backing_domain_id: Some(growth.pool.domain.domain_id),
862                generation: growth.chunk.generation(),
863            };
864            let request = BufferRequest::new(
865                growth.expected_resource_id.clone(),
866                growth.chunk_bytes,
867                reservation_evidence.alignment_bytes,
868                reservation_evidence.usage,
869                reservation_evidence.element_type,
870            )?;
871            validate_runtime_descriptor_for_admission(
872                self.runtime.descriptor(),
873                &self.binding,
874                "dynamic pool batch growth preflight",
875            )?;
876            let buffer = self
877                .runtime
878                .allocate(DeviceAllocationPermit {
879                    identity: &transaction_identity,
880                    binding: &self.binding,
881                    reservation: &reservation_evidence,
882                    request: &request,
883                    seal: AllocationSeal,
884                })
885                .map_err(|error| {
886                    invalid_resource(format!("dynamic pool device allocation failed: {error}"))
887                })?;
888            let actual_descriptor = self.runtime.buffer_descriptor(&buffer);
889            validate_runtime_descriptor_for_admission(
890                self.runtime.descriptor(),
891                &self.binding,
892                "dynamic pool batch growth completion",
893            )?;
894            if grant.bytes() != growth.chunk_bytes {
895                return Err(invalid_resource(
896                    "dynamic pool capacity grant differs from its chunk",
897                ));
898            }
899            let backing = Arc::new(ResidentChunkBacking {
900                buffer,
901                _grant: grant,
902                identity: growth.chunk.clone(),
903                descriptor: actual_descriptor.clone(),
904            });
905            if !reservation_evidence.matches_descriptor(&actual_descriptor) {
906                let mut state = growth
907                    .pool
908                    .state
909                    .lock()
910                    .unwrap_or_else(std::sync::PoisonError::into_inner);
911                state.quarantined.push(QuarantinedDynamicChunk {
912                    backing,
913                    reason: DynamicChunkQuarantineReason::DescriptorMismatch,
914                });
915                return Err(invalid_resource(
916                    "dynamic chunk descriptor mismatch was quarantined without capacity publication",
917                ));
918            }
919            allocated.push(AllocatedDynamicGrowth { backing });
920        }
921
922        let mut states = planned
923            .iter()
924            .map(|growth| {
925                growth.pool.state.lock().map_err(|_| {
926                    invalid_resource("dynamic backing pool is poisoned after allocation")
927                })
928            })
929            .collect::<Result<Vec<_>, _>>()?;
930        let mut published_totals = Vec::with_capacity(planned.len());
931        for ((growth, allocation), state) in planned.iter().zip(&allocated).zip(&states) {
932            if state.poisoned
933                || state.pending_growth_bytes < growth.chunk_bytes
934                || state.chunks.contains_key(&growth.chunk.ordinal())
935                || state
936                    .allocator
937                    .by_offset
938                    .range((growth.chunk.ordinal(), 0)..=(growth.chunk.ordinal(), u64::MAX))
939                    .next()
940                    .is_some()
941                || state
942                    .allocator
943                    .free_bytes
944                    .checked_add(growth.chunk_bytes)
945                    .is_none()
946                || allocation.backing.identity != growth.chunk
947            {
948                return Err(invalid_resource(
949                    "dynamic batch installation preconditions changed before publication",
950                ));
951            }
952            published_totals.push(
953                state
954                    .resident_bytes
955                    .checked_add(growth.chunk_bytes)
956                    .ok_or_else(|| invalid_resource("published dynamic capacity overflows u64"))?,
957            );
958        }
959        for index in 0..planned.len() {
960            let growth = &planned[index];
961            let state = &mut states[index];
962            state.pending_growth_bytes -= growth.chunk_bytes;
963            pending[index].disarm();
964            state
965                .allocator
966                .insert_extent(
967                    growth.chunk.ordinal(),
968                    growth.chunk.generation(),
969                    0,
970                    growth.chunk_bytes,
971                )
972                .expect("validated new chunk has one disjoint free extent");
973            state.chunks.insert(
974                growth.chunk.ordinal(),
975                ResidentChunkState {
976                    backing: Arc::clone(&allocated[index].backing),
977                    live_segments: 0,
978                },
979            );
980        }
981        let updates = planned
982            .iter()
983            .zip(&published_totals)
984            .map(|(growth, &total)| (growth.pool.domain.domain_id, CapacityUnits::new(total)))
985            .collect::<Vec<_>>();
986        let epochs = match self.logical_admission.set_domain_totals(&updates) {
987            Ok(epochs) => epochs,
988            Err(error) => {
989                for index in 0..planned.len() {
990                    states[index]
991                        .allocator
992                        .remove_extent(planned[index].chunk.ordinal(), 0)
993                        .expect("unpublished dynamic chunk free extent remains installed");
994                    let removed = states[index]
995                        .chunks
996                        .remove(&planned[index].chunk.ordinal())
997                        .expect("unpublished dynamic chunk remains installed");
998                    states[index].quarantined.push(QuarantinedDynamicChunk {
999                        backing: removed.backing,
1000                        reason: DynamicChunkQuarantineReason::PublicationRejected,
1001                    });
1002                }
1003                return Err(error);
1004            }
1005        };
1006        for (state, &published_total) in states.iter_mut().zip(&published_totals) {
1007            state.resident_bytes = published_total;
1008        }
1009        Ok(DynamicPoolGrowthBatchReceipt {
1010            coordinator_id: self.logical_admission.id(),
1011            growths: planned
1012                .iter()
1013                .zip(published_totals)
1014                .map(
1015                    |(growth, published_capacity_bytes)| DynamicPoolGrowthReceipt {
1016                        pool_id: growth.pool.domain.pool_id().clone(),
1017                        chunk: growth.chunk.clone(),
1018                        chunk_bytes: growth.chunk_bytes,
1019                        published_capacity_bytes,
1020                        capacity_epoch: epochs.capacity_epoch(),
1021                    },
1022                )
1023                .collect(),
1024            capacity_epoch: epochs.capacity_epoch(),
1025            rebalance: None,
1026            maintenance_boundary: None,
1027        })
1028    }
1029
1030    pub(in crate::vnext::resource) fn prepare_claim(
1031        &self,
1032        requests: &[EvaluatedBackingRequest<'_>],
1033    ) -> Result<BackingPrepareDecision<R>, VNextError> {
1034        if requests.is_empty() {
1035            return Ok(BackingPrepareDecision::Prepared(
1036                PreparedBackingClaim::empty(),
1037            ));
1038        }
1039        let lifetime = requests
1040            .first()
1041            .and_then(|request| request.projections.first())
1042            .map(|projection| projection.descriptor.lifetime())
1043            .ok_or_else(|| invalid_resource("dynamic backing request has no projection"))?;
1044        self.prepare_claim_scoped(
1045            requests,
1046            DynamicBackingClaimScope::from(lifetime),
1047            DynamicBackingClaimResidency::Transient,
1048        )
1049    }
1050
1051    fn reusable_capacity_shape_for_requests(
1052        &self,
1053        requests: &[EvaluatedBackingRequest<'_>],
1054    ) -> Result<Option<DynamicResourceShape>, VNextError> {
1055        let reusable_execution_bucket_id = requests
1056            .first()
1057            .and_then(|request| request.reusable_execution_bucket_id.as_ref());
1058        if requests.iter().any(|request| {
1059            request.reusable_execution_bucket_id.as_ref() != reusable_execution_bucket_id
1060        }) {
1061            return Err(invalid_resource(
1062                "one dynamic backing claim cannot mix reusable execution buckets",
1063            ));
1064        }
1065        reusable_execution_bucket_id
1066            .map(|bucket_id| {
1067                let bucket = self
1068                    .reusable_execution
1069                    .as_ref()
1070                    .and_then(|plan| plan.bucket(bucket_id))
1071                    .map(|resolved| resolved.bucket())
1072                    .ok_or_else(|| {
1073                        invalid_resource(
1074                            "dynamic backing claim references a reusable bucket outside its immutable plan",
1075                        )
1076                    })?;
1077                Ok(DynamicResourceShape::from_validated(
1078                    bucket.capacity().maximum_sequences(),
1079                    bucket.capacity().maximum_tokens(),
1080                    bucket.capacity().maximum_pages(),
1081                ))
1082            })
1083            .transpose()
1084    }
1085
1086    pub(in crate::vnext::resource) fn prepare_lane_stable_claim(
1087        self: &Arc<Self>,
1088        lane: &Arc<ExecutionLane<R>>,
1089        requests: &[EvaluatedBackingRequest<'_>],
1090    ) -> Result<LaneBackingPrepareDecision, VNextError> {
1091        if !Arc::ptr_eq(&self.runtime, lane.runtime_arc())
1092            || lane.descriptor() != self.runtime.descriptor()
1093            || !lane.is_reusable()
1094        {
1095            return Err(invalid_resource(
1096                "lane-stable backing requires the reusable execution lane bound to this plan runtime",
1097            ));
1098        }
1099        if requests.is_empty() {
1100            return Ok(LaneBackingPrepareDecision::Prepared(
1101                PreparedLaneBackingClaim::new(Vec::new(), None)?,
1102            ));
1103        }
1104        let reusable_capacity_shape = self.reusable_capacity_shape_for_requests(requests)?;
1105        if reusable_capacity_shape.is_none() {
1106            return match self.prepare_claim(requests)? {
1107                BackingPrepareDecision::Prepared(prepared) => {
1108                    Ok(LaneBackingPrepareDecision::Prepared(
1109                        PreparedLaneBackingClaim::new(prepared.commit(), None)?,
1110                    ))
1111                }
1112                BackingPrepareDecision::Deferred(deferred) => {
1113                    Ok(LaneBackingPrepareDecision::Deferred(deferred))
1114                }
1115            };
1116        }
1117        let lifetime = requests
1118            .first()
1119            .and_then(|request| request.projections.first())
1120            .map(|projection| projection.descriptor.lifetime())
1121            .ok_or_else(|| invalid_resource("dynamic backing request has no projection"))?;
1122        if !matches!(
1123            lifetime,
1124            AllocationLifetime::Step | AllocationLifetime::Invocation
1125        ) || requests.iter().any(|request| {
1126            request.projections.is_empty()
1127                || request.projections.iter().any(|projection| {
1128                    projection.descriptor.lifetime() != lifetime
1129                        || projection.descriptor.initialization() != StateInitialization::None
1130                })
1131        }) {
1132            return Err(invalid_resource(
1133                "lane-stable backing accepts only non-initialized Step or Invocation resources",
1134            ));
1135        }
1136
1137        let mut canonical_requests = requests.iter().collect::<Vec<_>>();
1138        canonical_requests
1139            .sort_unstable_by(|left, right| left.claim_identity.cmp(&right.claim_identity));
1140        let key = lane_stable_layout_key(lane.id(), lifetime, &canonical_requests)?;
1141        let lane_owner: Arc<dyn LaneStableArenaLane> =
1142            Arc::clone(lane) as Arc<dyn LaneStableArenaLane>;
1143
1144        loop {
1145            {
1146                let mut arenas = self
1147                    .lane_stable_arenas
1148                    .lock()
1149                    .map_err(|_| invalid_resource("lane-stable arena registry is poisoned"))?;
1150                if arenas.poisoned {
1151                    return Err(invalid_resource(
1152                        "lane-stable arena registry is fail-closed",
1153                    ));
1154                }
1155                let now = arenas.tick();
1156                if let Some(entry) = arenas.entries.get_mut(&key) {
1157                    let owner = entry.lane.upgrade().ok_or_else(|| {
1158                        invalid_resource("lane-stable arena retained an expired execution lane")
1159                    })?;
1160                    if !Arc::ptr_eq(&owner, &lane_owner) {
1161                        return Err(invalid_resource(
1162                            "lane-stable arena identity aliases another execution lane",
1163                        ));
1164                    }
1165                    if let Some((slot_id, stable, certificate, slot_domains)) =
1166                        entry.claim_idle_slot(lane.id(), now, &canonical_requests)?
1167                    {
1168                        return Ok(LaneBackingPrepareDecision::Prepared(
1169                            PreparedLaneBackingClaim::certified(
1170                                stable,
1171                                certificate,
1172                                LaneStableArenaSlotLease {
1173                                    arenas: Arc::clone(&self.lane_stable_arenas),
1174                                    logical_admission: self.logical_admission.clone(),
1175                                    availability_domains: slot_domains,
1176                                    key: key.clone(),
1177                                    slot_id,
1178                                },
1179                            ),
1180                        ));
1181                    }
1182                }
1183            }
1184
1185            match self.prepare_claim_scoped(
1186                requests,
1187                DynamicBackingClaimScope::from(lifetime),
1188                DynamicBackingClaimResidency::LaneStable,
1189            )? {
1190                BackingPrepareDecision::Prepared(prepared) => {
1191                    let mut arenas = self
1192                        .lane_stable_arenas
1193                        .lock()
1194                        .map_err(|_| invalid_resource("lane-stable arena registry is poisoned"))?;
1195                    if arenas.poisoned {
1196                        return Err(invalid_resource(
1197                            "lane-stable arena registry is fail-closed",
1198                        ));
1199                    }
1200                    let now = arenas.tick();
1201                    if let Some(entry) = arenas.entries.get_mut(&key) {
1202                        let owner = entry.lane.upgrade().ok_or_else(|| {
1203                            invalid_resource("lane-stable arena retained an expired execution lane")
1204                        })?;
1205                        if !Arc::ptr_eq(&owner, &lane_owner) {
1206                            return Err(invalid_resource(
1207                                "lane-stable arena identity aliases another execution lane",
1208                            ));
1209                        }
1210                        if let Some((slot_id, stable, certificate, slot_domains)) =
1211                            entry.claim_idle_slot(lane.id(), now, &canonical_requests)?
1212                        {
1213                            drop(arenas);
1214                            drop(prepared);
1215                            return Ok(LaneBackingPrepareDecision::Prepared(
1216                                PreparedLaneBackingClaim::certified(
1217                                    stable,
1218                                    certificate,
1219                                    LaneStableArenaSlotLease {
1220                                        arenas: Arc::clone(&self.lane_stable_arenas),
1221                                        logical_admission: self.logical_admission.clone(),
1222                                        availability_domains: slot_domains,
1223                                        key: key.clone(),
1224                                        slot_id,
1225                                    },
1226                                ),
1227                            ));
1228                        }
1229                    }
1230                    let authorities = prepared.commit();
1231                    let projection_bindings =
1232                        bind_lane_stable_slot_projections(&authorities, &canonical_requests)?;
1233                    let certificate = Arc::new(BackingClaimCertificate::from_slices(&authorities)?);
1234                    let stable = authorities
1235                        .iter()
1236                        .map(|authority| authority.retained_for_lane(lane.id()))
1237                        .collect();
1238                    let availability_domains = requests
1239                        .iter()
1240                        .map(|request| request.domain.domain_id())
1241                        .collect::<std::collections::BTreeSet<_>>()
1242                        .into_iter()
1243                        .collect::<Vec<_>>();
1244                    let slot_id = arenas.issue_slot_id()?;
1245                    let entry =
1246                        arenas
1247                            .entries
1248                            .entry(key.clone())
1249                            .or_insert_with(|| LaneStableArenaEntry {
1250                                lane: Arc::downgrade(&lane_owner),
1251                                slots: BTreeMap::new(),
1252                            });
1253                    if entry
1254                        .slots
1255                        .insert(
1256                            slot_id,
1257                            LaneStableArenaSlot {
1258                                slot_id,
1259                                authorities,
1260                                certificate: Arc::clone(&certificate),
1261                                projection_bindings,
1262                                availability_domains: availability_domains.clone(),
1263                                in_use: true,
1264                                last_used: now,
1265                            },
1266                        )
1267                        .is_some()
1268                    {
1269                        arenas.poisoned = true;
1270                        return Err(invalid_resource(
1271                            "lane-stable arena slot publication replaced an existing slot",
1272                        ));
1273                    }
1274                    return Ok(LaneBackingPrepareDecision::Prepared(
1275                        PreparedLaneBackingClaim::certified(
1276                            stable,
1277                            certificate,
1278                            LaneStableArenaSlotLease {
1279                                arenas: Arc::clone(&self.lane_stable_arenas),
1280                                logical_admission: self.logical_admission.clone(),
1281                                availability_domains: availability_domains.clone(),
1282                                key: key.clone(),
1283                                slot_id,
1284                            },
1285                        ),
1286                    ));
1287                }
1288                BackingPrepareDecision::Deferred(deferred) => {
1289                    let mut arenas = self
1290                        .lane_stable_arenas
1291                        .lock()
1292                        .map_err(|_| invalid_resource("lane-stable arena registry is poisoned"))?;
1293                    if arenas.poisoned {
1294                        return Err(invalid_resource(
1295                            "lane-stable arena registry is fail-closed",
1296                        ));
1297                    }
1298                    let now = arenas.tick();
1299                    if let Some(entry) = arenas.entries.get_mut(&key) {
1300                        let owner = entry.lane.upgrade().ok_or_else(|| {
1301                            invalid_resource("lane-stable arena retained an expired execution lane")
1302                        })?;
1303                        if !Arc::ptr_eq(&owner, &lane_owner) {
1304                            return Err(invalid_resource(
1305                                "lane-stable arena identity aliases another execution lane",
1306                            ));
1307                        }
1308                        if let Some((slot_id, stable, certificate, slot_domains)) =
1309                            entry.claim_idle_slot(lane.id(), now, &canonical_requests)?
1310                        {
1311                            drop(arenas);
1312                            drop(deferred);
1313                            return Ok(LaneBackingPrepareDecision::Prepared(
1314                                PreparedLaneBackingClaim::certified(
1315                                    stable,
1316                                    certificate,
1317                                    LaneStableArenaSlotLease {
1318                                        arenas: Arc::clone(&self.lane_stable_arenas),
1319                                        logical_admission: self.logical_admission.clone(),
1320                                        availability_domains: slot_domains,
1321                                        key: key.clone(),
1322                                        slot_id,
1323                                    },
1324                                ),
1325                            ));
1326                        }
1327                    }
1328                    return Ok(LaneBackingPrepareDecision::Deferred(deferred));
1329                }
1330            }
1331        }
1332    }
1333
1334    pub(in crate::vnext::resource) fn try_reclaim_expired_lane_slots(
1335        &self,
1336    ) -> Result<bool, VNextError> {
1337        let expired_entries = {
1338            let mut arenas = self
1339                .lane_stable_arenas
1340                .lock()
1341                .map_err(|_| invalid_resource("lane-stable arena registry is poisoned"))?;
1342            if arenas.poisoned {
1343                return Err(invalid_resource(
1344                    "lane-stable arena registry is fail-closed",
1345                ));
1346            }
1347            arenas.take_expired_lanes()?
1348        };
1349        let reclaimed = !expired_entries.is_empty();
1350        // Releasing backing owners can enter backend/pool destruction paths.
1351        // Keep that work outside the arena registry's hot mutex.
1352        drop(expired_entries);
1353        Ok(reclaimed)
1354    }
1355
1356    pub(in crate::vnext::resource) fn try_reclaim_one_idle_lane_slot(
1357        &self,
1358    ) -> Result<bool, VNextError> {
1359        if self.try_reclaim_expired_lane_slots()? {
1360            return Ok(true);
1361        }
1362        let mut candidates = {
1363            let arenas = self
1364                .lane_stable_arenas
1365                .lock()
1366                .map_err(|_| invalid_resource("lane-stable arena registry is poisoned"))?;
1367            if arenas.poisoned {
1368                return Err(invalid_resource(
1369                    "lane-stable arena registry is fail-closed",
1370                ));
1371            }
1372            arenas
1373                .entries
1374                .iter()
1375                .filter_map(|(key, entry)| entry.lane.upgrade().map(|lane| (key, entry, lane)))
1376                .flat_map(|(key, entry, lane)| {
1377                    entry
1378                        .slots
1379                        .values()
1380                        .filter(|slot| !slot.in_use)
1381                        .map(move |slot| LaneStableArenaEvictionCandidate {
1382                            key: key.clone(),
1383                            slot_id: slot.slot_id,
1384                            last_used: slot.last_used,
1385                            lane: Arc::clone(&lane),
1386                        })
1387                })
1388                .collect::<Vec<_>>()
1389        };
1390        candidates.sort_by_key(|candidate| candidate.last_used);
1391
1392        for candidate in candidates {
1393            if !candidate.lane.try_trim_reusable_executables()? {
1394                continue;
1395            }
1396            let victim = {
1397                let mut arenas = self
1398                    .lane_stable_arenas
1399                    .lock()
1400                    .map_err(|_| invalid_resource("lane-stable arena registry is poisoned"))?;
1401                if arenas.poisoned {
1402                    return Err(invalid_resource(
1403                        "lane-stable arena registry is fail-closed",
1404                    ));
1405                }
1406                let removable = arenas
1407                    .entries
1408                    .get(&candidate.key)
1409                    .and_then(|entry| entry.slots.get(&candidate.slot_id))
1410                    .is_some_and(|slot| !slot.in_use && !slot.has_external_address_pins());
1411                if !removable {
1412                    None
1413                } else {
1414                    let (victim, remove_entry) = {
1415                        let entry = arenas.entries.get_mut(&candidate.key).ok_or_else(|| {
1416                            invalid_resource("lane-stable arena eviction lost its entry")
1417                        })?;
1418                        let victim = entry.slots.remove(&candidate.slot_id).ok_or_else(|| {
1419                            invalid_resource("lane-stable arena eviction lost its idle slot")
1420                        })?;
1421                        (victim, entry.slots.is_empty())
1422                    };
1423                    if remove_entry {
1424                        arenas.entries.remove(&candidate.key);
1425                    }
1426                    Some(victim)
1427                }
1428            };
1429            if let Some(victim) = victim {
1430                drop(victim);
1431                return Ok(true);
1432            }
1433        }
1434        Ok(false)
1435    }
1436
1437    pub(in crate::vnext::resource) fn prepare_initial_sequence_claim(
1438        &self,
1439        requests: &[EvaluatedBackingRequest<'_>],
1440    ) -> Result<BackingPrepareDecision<R>, VNextError> {
1441        self.prepare_claim_scoped(
1442            requests,
1443            DynamicBackingClaimScope::InitialSequenceBundle,
1444            DynamicBackingClaimResidency::Transient,
1445        )
1446    }
1447
1448    fn prepare_claim_scoped(
1449        &self,
1450        requests: &[EvaluatedBackingRequest<'_>],
1451        scope: DynamicBackingClaimScope,
1452        residency: DynamicBackingClaimResidency,
1453    ) -> Result<BackingPrepareDecision<R>, VNextError> {
1454        if requests.is_empty() {
1455            return Ok(BackingPrepareDecision::Prepared(
1456                PreparedBackingClaim::empty(),
1457            ));
1458        }
1459        let reusable_capacity_shape = self.reusable_capacity_shape_for_requests(requests)?;
1460        let mut grouped =
1461            BTreeMap::<DynamicBackingPoolId, Vec<&EvaluatedBackingRequest<'_>>>::new();
1462        for request in requests {
1463            grouped
1464                .entry(request.domain.pool_id().clone())
1465                .or_default()
1466                .push(request);
1467        }
1468        let mut groups = Vec::with_capacity(grouped.len());
1469        for (pool_id, mut requests) in grouped {
1470            requests.sort_by(|left, right| left.claim_identity.cmp(&right.claim_identity));
1471            if requests
1472                .windows(2)
1473                .any(|pair| pair[0].claim_identity == pair[1].claim_identity)
1474            {
1475                return Err(invalid_resource(
1476                    "dynamic backing reservation contains a duplicate physical claim",
1477                ));
1478            }
1479            let pool = self.pools.get(&pool_id).cloned().ok_or_else(|| {
1480                invalid_resource("dynamic backing reservation references an unknown pool")
1481            })?;
1482            groups.push((pool, requests));
1483        }
1484        let protected_packing_envelopes = groups
1485            .iter()
1486            .map(|(pool, requests)| {
1487                DynamicBackingPackingEnvelope::new(
1488                    pool.domain.pool_id().clone(),
1489                    pool.domain.domain_id,
1490                    requests
1491                        .iter()
1492                        .map(|request| request.capacity_size_bytes)
1493                        .collect(),
1494                )
1495            })
1496            .collect::<Result<Vec<_>, VNextError>>()?;
1497        let protected_immediate = CapacityVector::new(
1498            protected_packing_envelopes
1499                .iter()
1500                .map(|envelope| {
1501                    CapacityEntry::new(
1502                        envelope.domain_id(),
1503                        CapacityUnits::new(envelope.total_bytes()?),
1504                    )
1505                })
1506                .collect::<Result<Vec<_>, VNextError>>()?,
1507        )?;
1508        'prepare: loop {
1509            let mut states = groups
1510                .iter()
1511                .map(|(pool, _)| {
1512                    pool.state
1513                        .lock()
1514                        .map_err(|_| invalid_resource("dynamic backing pool is poisoned"))
1515                })
1516                .collect::<Result<Vec<_>, _>>()?;
1517            for (group_index, (pool, pool_requests)) in groups.iter().enumerate() {
1518                if states[group_index].poisoned {
1519                    return Err(invalid_resource("dynamic backing pool is fail-closed"));
1520                }
1521                let quantum = pool.allocation_quantum();
1522                for request in pool_requests {
1523                    let projection_ids = request
1524                        .projections
1525                        .iter()
1526                        .map(|projection| projection.descriptor.base_resource_id().clone())
1527                        .collect::<Vec<_>>();
1528                    let single_projection = request.projections.len() == 1
1529                        && request.projections[0].physical_offset_bytes == 0
1530                        && request.projections[0].capacity_size_bytes
1531                            == request.capacity_size_bytes;
1532                    let shared_step_slot = request.projections.len() > 1
1533                        && request
1534                            .projections
1535                            .iter()
1536                            .all(|projection| projection.physical_offset_bytes == 0)
1537                        && request
1538                            .projections
1539                            .iter()
1540                            .map(|projection| projection.capacity_size_bytes)
1541                            .max()
1542                            == Some(request.capacity_size_bytes)
1543                        && pool.domain.pool.step_resource_slots().iter().any(|slot| {
1544                            slot.kind() == StepResourceSlotKind::OrderedSingleFenceStepWave
1545                                && slot.resource_ids() == request.claim_identity.resource_ids()
1546                        });
1547                    let invocation_wave =
1548                        self.validate_invocation_wave_projection(pool, request)?;
1549                    if request.domain.pool_id() != pool.domain.pool_id()
1550                        || request.claim_identity.pool_id() != pool.domain.pool_id()
1551                        || request.claim_identity.resource_ids() != projection_ids
1552                        || request.projections.is_empty()
1553                        || request.projections.windows(2).any(|pair| {
1554                            pair[0].descriptor.base_resource_id()
1555                                >= pair[1].descriptor.base_resource_id()
1556                        })
1557                        || request.projections.iter().any(|projection| {
1558                            let capacity_matches_plan = match reusable_capacity_shape {
1559                                Some(shape) => {
1560                                    matches!(
1561                                        projection.descriptor.lifetime(),
1562                                        AllocationLifetime::Step | AllocationLifetime::Invocation
1563                                    ) && projection
1564                                        .descriptor
1565                                        .evaluate_request_bytes_for_shape(shape)
1566                                        .is_ok_and(|bytes| bytes == projection.capacity_size_bytes)
1567                                }
1568                                None => {
1569                                    projection.logical_size_bytes == projection.capacity_size_bytes
1570                                }
1571                            };
1572                            projection.descriptor.pool_id() != pool.domain.pool_id()
1573                                || !scope.accepts(projection.descriptor.lifetime())
1574                                || !capacity_matches_plan
1575                                || projection.logical_size_bytes == 0
1576                                || projection.logical_size_bytes > projection.capacity_size_bytes
1577                                || projection.capacity_size_bytes == 0
1578                                || projection.capacity_size_bytes % quantum != 0
1579                                || projection.physical_offset_bytes % quantum != 0
1580                                || projection
1581                                    .physical_offset_bytes
1582                                    .checked_add(projection.capacity_size_bytes)
1583                                    .is_none_or(|end| end > request.capacity_size_bytes)
1584                                || !request
1585                                    .domain
1586                                    .descriptors
1587                                    .iter()
1588                                    .any(|descriptor| descriptor == projection.descriptor)
1589                        })
1590                        || request.capacity_size_bytes == 0
1591                        || request.capacity_size_bytes % quantum != 0
1592                        || !(single_projection || shared_step_slot || invocation_wave)
1593                    {
1594                        return Err(invalid_resource(
1595                        "dynamic backing request violates its physical claim, projection, pool, or allocation quantum",
1596                    ));
1597                    }
1598                }
1599            }
1600            let blockers = groups
1601                .iter()
1602                .enumerate()
1603                .map(|(group_index, (pool, pool_requests))| {
1604                    let requested_group_bytes =
1605                        pool_requests.iter().try_fold(0_u64, |total, request| {
1606                            total
1607                                .checked_add(request.capacity_size_bytes)
1608                                .ok_or_else(|| {
1609                                    invalid_resource("dynamic backing batch bytes overflow u64")
1610                                })
1611                        })?;
1612                    let state = &states[group_index];
1613                    if state.allocator.free_bytes >= requested_group_bytes {
1614                        return Ok(None);
1615                    }
1616                    let (reason, requested_bytes, contiguous_claim_bytes_descending) =
1617                        match pool.domain.pool.compatibility().profile().view() {
1618                            DynamicStorageView::Contiguous => {
1619                                let mut claim_bytes = pool_requests
1620                                    .iter()
1621                                    .map(|request| request.capacity_size_bytes)
1622                                    .collect::<Vec<_>>();
1623                                claim_bytes.sort_unstable_by(|left, right| right.cmp(left));
1624                                let growth = contiguous_packing_growth_bytes(
1625                                    &state.allocator,
1626                                    pool.domain.pool_id(),
1627                                    &claim_bytes,
1628                                )?;
1629                                if growth == 0 {
1630                                    return Err(invalid_resource(
1631                                    "insufficient contiguous capacity produced zero packing growth",
1632                                ));
1633                                }
1634                                (
1635                                    DynamicBackingDeferralReason::GrowthRequired,
1636                                    growth,
1637                                    Some(claim_bytes),
1638                                )
1639                            }
1640                            DynamicStorageView::PagedRegions { .. } => (
1641                                DynamicBackingDeferralReason::GrowthRequired,
1642                                requested_group_bytes - state.allocator.free_bytes,
1643                                None,
1644                            ),
1645                        };
1646                    Ok(Some(DynamicBackingBlocker {
1647                        pool_id: pool.domain.pool_id().clone(),
1648                        domain_id: pool.domain.domain_id,
1649                        reason,
1650                        requested_bytes,
1651                        free_bytes: state.allocator.free_bytes,
1652                        largest_contiguous_bytes: state.allocator.largest_contiguous_bytes(),
1653                        free_extent_layout_fingerprint: free_extent_layout_fingerprint(
1654                            &state.allocator,
1655                        ),
1656                        contiguous_claim_bytes_descending,
1657                    }))
1658                })
1659                .collect::<Result<Vec<_>, VNextError>>()?
1660                .into_iter()
1661                .flatten()
1662                .collect::<Vec<_>>();
1663            if !blockers.is_empty() {
1664                drop(states);
1665                if let Some(deferred) = self.confirm_backing_deferral(
1666                    blockers,
1667                    scope,
1668                    protected_immediate.clone(),
1669                    protected_packing_envelopes.clone(),
1670                )? {
1671                    return Ok(BackingPrepareDecision::Deferred(deferred));
1672                }
1673                continue 'prepare;
1674            }
1675            let segment_generations = groups
1676                .iter()
1677                .map(|(pool, requests)| {
1678                    (0..requests.len())
1679                        .map(|_| {
1680                            pool.next_extent_generation
1681                                .fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| {
1682                                    current.checked_add(1)
1683                                })
1684                                .map_err(|_| {
1685                                    invalid_resource("dynamic extent generation space is exhausted")
1686                                })
1687                        })
1688                        .collect::<Result<Vec<_>, _>>()
1689                })
1690                .collect::<Result<Vec<_>, _>>()?;
1691            let mut selections = groups
1692                .iter()
1693                .map(|_| {
1694                    Vec::<(
1695                        &EvaluatedBackingRequest<'_>,
1696                        u64,
1697                        Vec<BackingSegment>,
1698                        DynamicBackingClaimOccupancy,
1699                    )>::new()
1700                })
1701                .collect::<Vec<_>>();
1702            let mut journals = groups
1703                .iter()
1704                .map(|_| Vec::<Vec<BackingSegment>>::new())
1705                .collect::<Vec<_>>();
1706            for group_index in 0..groups.len() {
1707                let (pool, pool_requests) = &groups[group_index];
1708                let profile = pool.domain.pool.compatibility().profile();
1709                let mut allocation_requests = pool_requests.clone();
1710                allocation_requests.sort_by(|left, right| {
1711                    right
1712                        .capacity_size_bytes
1713                        .cmp(&left.capacity_size_bytes)
1714                        .then_with(|| left.claim_identity.cmp(&right.claim_identity))
1715                });
1716                for request in allocation_requests {
1717                    let reserved = match match profile.view() {
1718                        DynamicStorageView::Contiguous => states[group_index]
1719                            .allocator
1720                            .allocate_contiguous(pool.domain.pool_id(), request.capacity_size_bytes)
1721                            .map(|segment| segment.map(|segment| vec![segment])),
1722                        DynamicStorageView::PagedRegions { block_bytes } => {
1723                            states[group_index].allocator.allocate_paged(
1724                                pool.domain.pool_id(),
1725                                request.capacity_size_bytes,
1726                                block_bytes,
1727                            )
1728                        }
1729                    } {
1730                        Ok(reserved) => reserved,
1731                        Err(error) => {
1732                            states[group_index].poisoned = true;
1733                            rollback_free_extent_journal(&mut states, &journals)?;
1734                            return Err(error);
1735                        }
1736                    };
1737                    let Some(segments) = reserved else {
1738                        rollback_free_extent_journal(&mut states, &journals)?;
1739                        if !matches!(profile.view(), DynamicStorageView::Contiguous) {
1740                            states[group_index].poisoned = true;
1741                            return Err(invalid_resource(
1742                                "paged backing allocation failed after its aggregate fit check",
1743                            ));
1744                        }
1745                        let free_bytes = states[group_index].allocator.free_bytes;
1746                        let reason = DynamicBackingDeferralReason::FragmentedContiguous;
1747                        let mut claim_bytes_descending = pool_requests
1748                            .iter()
1749                            .map(|request| request.capacity_size_bytes)
1750                            .collect::<Vec<_>>();
1751                        claim_bytes_descending.sort_unstable_by(|left, right| right.cmp(left));
1752                        let requested_bytes = contiguous_packing_growth_bytes(
1753                            &states[group_index].allocator,
1754                            pool.domain.pool_id(),
1755                            &claim_bytes_descending,
1756                        )?;
1757                        if requested_bytes == 0 {
1758                            return Err(invalid_resource(
1759                                "contiguous packing failed without a progress-producing growth",
1760                            ));
1761                        }
1762                        let blocker = DynamicBackingBlocker {
1763                            pool_id: pool.domain.pool_id().clone(),
1764                            domain_id: pool.domain.domain_id,
1765                            reason,
1766                            requested_bytes,
1767                            free_bytes,
1768                            largest_contiguous_bytes: states[group_index]
1769                                .allocator
1770                                .largest_contiguous_bytes(),
1771                            free_extent_layout_fingerprint: free_extent_layout_fingerprint(
1772                                &states[group_index].allocator,
1773                            ),
1774                            contiguous_claim_bytes_descending: Some(claim_bytes_descending),
1775                        };
1776                        drop(states);
1777                        if let Some(deferred) = self.confirm_backing_deferral(
1778                            vec![blocker],
1779                            scope,
1780                            protected_immediate.clone(),
1781                            protected_packing_envelopes.clone(),
1782                        )? {
1783                            return Ok(BackingPrepareDecision::Deferred(deferred));
1784                        }
1785                        continue 'prepare;
1786                    };
1787                    journals[group_index].push(segments.clone());
1788                    let extent_bytes = match segments.iter().try_fold(0_u64, |total, segment| {
1789                        total.checked_add(segment.length_bytes()).ok_or_else(|| {
1790                            invalid_resource("dynamic backing extent bytes overflow u64")
1791                        })
1792                    }) {
1793                        Ok(bytes) => bytes,
1794                        Err(error) => {
1795                            rollback_free_extent_journal(&mut states, &journals)?;
1796                            return Err(error);
1797                        }
1798                    };
1799                    if extent_bytes != request.capacity_size_bytes {
1800                        rollback_free_extent_journal(&mut states, &journals)?;
1801                        return Err(invalid_resource(
1802                            "dynamic backing extents differ from their physical capacity claim",
1803                        ));
1804                    }
1805                    let generation =
1806                        segment_generations[group_index][selections[group_index].len()];
1807                    let segment_count = match u64::try_from(segments.len()) {
1808                        Ok(count) => count,
1809                        Err(_) => {
1810                            rollback_free_extent_journal(&mut states, &journals)?;
1811                            return Err(invalid_resource(
1812                                "dynamic backing segment count exceeds portable range",
1813                            ));
1814                        }
1815                    };
1816                    selections[group_index].push((
1817                        request,
1818                        generation,
1819                        segments,
1820                        DynamicBackingClaimOccupancy {
1821                            scope,
1822                            residency,
1823                            physical_bytes: request.capacity_size_bytes,
1824                            segment_count,
1825                        },
1826                    ));
1827                }
1828            }
1829
1830            for group_selections in &selections {
1831                for (request, _, segments, _) in group_selections {
1832                    for projection in &request.projections {
1833                        if let Err(error) = backing_segment_range(
1834                            segments,
1835                            projection.physical_offset_bytes,
1836                            projection.capacity_size_bytes,
1837                        ) {
1838                            rollback_free_extent_journal(&mut states, &journals)?;
1839                            return Err(error);
1840                        }
1841                    }
1842                }
1843            }
1844
1845            let accounting_updates = match (0..groups.len())
1846                .map(|group_index| {
1847                    let mut increments = BTreeMap::<u32, u64>::new();
1848                    for (_, _, segments, _) in &selections[group_index] {
1849                        for segment in segments {
1850                            let count = increments.entry(segment.chunk_ordinal()).or_default();
1851                            *count = count.checked_add(1).ok_or_else(|| {
1852                                invalid_resource(
1853                                    "dynamic chunk live extent increment overflows u64",
1854                                )
1855                            })?;
1856                        }
1857                    }
1858                    for (&ordinal, &increment) in &increments {
1859                        states[group_index]
1860                            .chunks
1861                            .get(&ordinal)
1862                            .ok_or_else(|| invalid_resource("reserved dynamic chunk disappeared"))?
1863                            .live_segments
1864                            .checked_add(increment)
1865                            .ok_or_else(|| {
1866                                invalid_resource("dynamic chunk live extent count overflowed")
1867                            })?;
1868                    }
1869                    let occupancy = selections[group_index].iter().try_fold(
1870                        states[group_index].live_occupancy,
1871                        |occupancy, (_, _, _, claim)| occupancy.checked_with_claim(*claim),
1872                    )?;
1873                    Ok((increments, occupancy))
1874                })
1875                .collect::<Result<Vec<_>, VNextError>>()
1876            {
1877                Ok(updates) => updates,
1878                Err(error) => {
1879                    rollback_free_extent_journal(&mut states, &journals)?;
1880                    return Err(error);
1881                }
1882            };
1883            for (group_index, (increments, occupancy)) in accounting_updates.into_iter().enumerate()
1884            {
1885                for (ordinal, increment) in increments {
1886                    states[group_index]
1887                        .chunks
1888                        .get_mut(&ordinal)
1889                        .expect("validated reserved dynamic chunk remains installed")
1890                        .live_segments += increment;
1891                }
1892                states[group_index].live_occupancy = occupancy;
1893            }
1894            drop(states);
1895            let mut extents = Vec::new();
1896            for ((pool, _), selections) in groups.into_iter().zip(selections) {
1897                for (request, segment_generation, segments, occupancy) in selections {
1898                    let projections = request
1899                        .projections
1900                        .iter()
1901                        .map(|projection| {
1902                            let mut allocation = LogicalBackingSliceAllocationEvidence {
1903                                domain_id: pool.domain.domain_id,
1904                                pool_id: pool.domain.pool_id().clone(),
1905                                resource_id: projection.descriptor.base_resource_id().clone(),
1906                                pool_instance_id: pool.instance_id,
1907                                physical_claim_identity: request.claim_identity.clone(),
1908                                reusable_execution_bucket_id: request
1909                                    .reusable_execution_bucket_id
1910                                    .clone(),
1911                                segment_generation,
1912                                segments: backing_segment_range(
1913                                    &segments,
1914                                    projection.physical_offset_bytes,
1915                                    projection.capacity_size_bytes,
1916                                )?,
1917                                physical_offset_bytes: projection.physical_offset_bytes,
1918                                capacity_size_bytes: projection.capacity_size_bytes,
1919                                physical_size_bytes: request.capacity_size_bytes,
1920                                alignment_bytes: projection.descriptor.alignment_bytes(),
1921                                usage: projection.descriptor.usage(),
1922                                element_type: projection.descriptor.element_type(),
1923                                storage_profile: pool.domain.pool.compatibility().profile(),
1924                                initialization: projection.descriptor.initialization(),
1925                                fingerprint: String::new(),
1926                            };
1927                            let bytes = serde_json::to_vec(&allocation).map_err(|error| {
1928                                invalid_resource(format!(
1929                                    "logical backing allocation evidence encode failed: {error}"
1930                                ))
1931                            })?;
1932                            allocation.fingerprint = format!("sha256/{:x}", Sha256::digest(bytes));
1933                            Ok(LogicalBackingSliceEvidence {
1934                                allocation: Arc::new(allocation),
1935                                logical_size_bytes: projection.logical_size_bytes,
1936                            })
1937                        })
1938                        .collect::<Result<Vec<_>, VNextError>>()?;
1939                    extents.push(PreparedBackingExtent {
1940                        pool: Arc::clone(&pool),
1941                        claim_identity: request.claim_identity.clone(),
1942                        segment_generation,
1943                        occupancy,
1944                        segments,
1945                        capacity_size_bytes: request.capacity_size_bytes,
1946                        projections,
1947                    });
1948                }
1949            }
1950            return Ok(BackingPrepareDecision::Prepared(PreparedBackingClaim {
1951                extents,
1952                committed: false,
1953            }));
1954        }
1955    }
1956
1957    /// Publishes a physical deferral with the event-subscription ordering
1958    /// required to avoid a lost release:
1959    ///
1960    /// 1. observe the exact coordinator generations after the failed check;
1961    /// 2. recheck the physical allocator observations;
1962    /// 3. publish only if those observations are still current.
1963    ///
1964    /// A release before step 1 is visible to step 2. A release after step 1
1965    /// advances the returned predicate (or is visible to step 2 before its
1966    /// coordinator notification), so the scheduler cannot sleep forever on a
1967    /// stale blocker.
1968    fn confirm_backing_deferral(
1969        &self,
1970        blockers: Vec<DynamicBackingBlocker>,
1971        scope: DynamicBackingClaimScope,
1972        protected_immediate: CapacityVector,
1973        protected_packing_envelopes: Vec<DynamicBackingPackingEnvelope>,
1974    ) -> Result<Option<DynamicBackingDeferred>, VNextError> {
1975        let wait_snapshot = self
1976            .logical_admission
1977            .wait_snapshot_for_domains(blockers.iter().map(DynamicBackingBlocker::domain_id))?;
1978        for blocker in &blockers {
1979            let pool = self.pools.get(blocker.pool_id()).ok_or_else(|| {
1980                invalid_resource("dynamic backing blocker references an unknown pool")
1981            })?;
1982            let state = pool
1983                .state
1984                .lock()
1985                .map_err(|_| invalid_resource("dynamic backing pool is poisoned"))?;
1986            if state.poisoned {
1987                return Err(invalid_resource("dynamic backing pool is fail-closed"));
1988            }
1989            if state.allocator.free_bytes != blocker.free_bytes()
1990                || state.allocator.largest_contiguous_bytes() != blocker.largest_contiguous_bytes()
1991                || free_extent_layout_fingerprint(&state.allocator)
1992                    != blocker.free_extent_layout_fingerprint()
1993            {
1994                return Ok(None);
1995            }
1996        }
1997        Ok(Some(DynamicBackingDeferred {
1998            blockers,
1999            epochs: wait_snapshot.epochs(),
2000            wait_condition: wait_snapshot.wait_condition().clone(),
2001            scope,
2002            protected_immediate,
2003            protected_packing_envelopes,
2004        }))
2005    }
2006
2007    fn validate_invocation_wave_projection(
2008        &self,
2009        pool: &DynamicBackingPool<R>,
2010        request: &EvaluatedBackingRequest<'_>,
2011    ) -> Result<bool, VNextError> {
2012        let mode = pool.domain.pool.invocation_liveness_mode();
2013        if mode == InvocationLivenessMode::NoInvocationResources
2014            || request.projections.len() < 2
2015            || request.projections.iter().any(|projection| {
2016                projection.descriptor.lifetime() != super::AllocationLifetime::Invocation
2017            })
2018        {
2019            return Ok(false);
2020        }
2021        let mut expected_resources = pool
2022            .domain
2023            .pool
2024            .invocation_liveness()
2025            .iter()
2026            .flat_map(|row| row.resource_ids().iter().cloned())
2027            .collect::<Vec<_>>();
2028        expected_resources.sort();
2029        if expected_resources.windows(2).any(|pair| pair[0] == pair[1])
2030            || expected_resources != request.claim_identity.resource_ids()
2031        {
2032            return Ok(false);
2033        }
2034        let projections = request
2035            .projections
2036            .iter()
2037            .map(|projection| (projection.descriptor.base_resource_id().clone(), projection))
2038            .collect::<BTreeMap<_, _>>();
2039        if projections.len() != request.projections.len() {
2040            return Ok(false);
2041        }
2042        let rows_by_node = pool
2043            .domain
2044            .pool
2045            .invocation_liveness()
2046            .iter()
2047            .map(|row| (row.node_id(), row))
2048            .collect::<BTreeMap<_, _>>();
2049        let rows = self
2050            .nodes
2051            .iter()
2052            .filter_map(|node| rows_by_node.get(node.id()).copied())
2053            .collect::<Vec<_>>();
2054        if rows.len() != rows_by_node.len() {
2055            return Ok(false);
2056        }
2057
2058        let mut concurrent_cursor = 0_u64;
2059        let mut peak = 0_u64;
2060        for row in rows {
2061            let row_base = match mode {
2062                InvocationLivenessMode::TotalOrderReuse => 0,
2063                InvocationLivenessMode::ConservativeConcurrent => concurrent_cursor,
2064                InvocationLivenessMode::NoInvocationResources => unreachable!(),
2065            };
2066            let mut row_cursor = 0_u64;
2067            for resource_id in row.resource_ids() {
2068                let Some(projection) = projections.get(resource_id) else {
2069                    return Ok(false);
2070                };
2071                let expected_offset = row_base.checked_add(row_cursor).ok_or_else(|| {
2072                    invalid_resource("invocation wave projection offset overflows u64")
2073                })?;
2074                if projection.physical_offset_bytes != expected_offset {
2075                    return Ok(false);
2076                }
2077                row_cursor = row_cursor
2078                    .checked_add(projection.capacity_size_bytes)
2079                    .ok_or_else(|| invalid_resource("invocation wave row size overflows u64"))?;
2080            }
2081            peak = peak.max(row_cursor);
2082            if mode == InvocationLivenessMode::ConservativeConcurrent {
2083                concurrent_cursor = concurrent_cursor.checked_add(row_cursor).ok_or_else(|| {
2084                    invalid_resource("concurrent invocation wave size overflows u64")
2085                })?;
2086            }
2087        }
2088        Ok(request.capacity_size_bytes
2089            == match mode {
2090                InvocationLivenessMode::TotalOrderReuse => peak,
2091                InvocationLivenessMode::ConservativeConcurrent => concurrent_cursor,
2092                InvocationLivenessMode::NoInvocationResources => 0,
2093            })
2094    }
2095
2096    pub(in crate::vnext::resource) fn view<'lease>(
2097        &'lease self,
2098        authority: &'lease LogicalBackingSliceAuthority,
2099    ) -> Result<LogicalBackingBufferView<'lease, R::Buffer>, VNextError> {
2100        self.view_many(std::slice::from_ref(authority))
2101    }
2102
2103    pub(in crate::vnext::resource) fn view_many<'lease>(
2104        &'lease self,
2105        authorities: &'lease [LogicalBackingSliceAuthority],
2106    ) -> Result<LogicalBackingBufferView<'lease, R::Buffer>, VNextError> {
2107        let first = authorities
2108            .first()
2109            .ok_or_else(|| invalid_resource("logical backing view requires an authority"))?;
2110        let pool = self
2111            .pools
2112            .get(&first.evidence.pool_id)
2113            .ok_or_else(|| invalid_resource("logical backing authority has no dynamic pool"))?;
2114        let mut logical_size_bytes = 0_u64;
2115        let mut capacity_size_bytes = 0_u64;
2116        let mut segment_count = 0_usize;
2117        for (index, authority) in authorities.iter().enumerate() {
2118            if authority.evidence.pool_id != first.evidence.pool_id
2119                || authority.evidence.resource_id != first.evidence.resource_id
2120                || authority.evidence.storage_profile != first.evidence.storage_profile
2121                || authority.evidence.alignment_bytes != first.evidence.alignment_bytes
2122                || authority.evidence.usage != first.evidence.usage
2123                || authority.evidence.element_type != first.evidence.element_type
2124                || authority.evidence.initialization != first.evidence.initialization
2125            {
2126                return Err(invalid_resource(
2127                    "logical backing authorities have incompatible resource metadata",
2128                ));
2129            }
2130            Self::validate_authority(pool, authority)?;
2131            if index + 1 < authorities.len()
2132                && authority.evidence.logical_size_bytes != authority.evidence.capacity_size_bytes
2133            {
2134                return Err(invalid_resource(
2135                    "multi-extent logical backing cannot contain interior capacity slack",
2136                ));
2137            }
2138            logical_size_bytes = logical_size_bytes
2139                .checked_add(authority.evidence.logical_size_bytes)
2140                .ok_or_else(|| invalid_resource("logical backing view size overflows u64"))?;
2141            capacity_size_bytes = capacity_size_bytes
2142                .checked_add(authority.evidence.capacity_size_bytes)
2143                .ok_or_else(|| invalid_resource("logical backing capacity overflows u64"))?;
2144            segment_count = segment_count
2145                .checked_add(authority.evidence.segments.len())
2146                .ok_or_else(|| invalid_resource("logical backing segment count overflows usize"))?;
2147        }
2148        let state = pool
2149            .state
2150            .lock()
2151            .map_err(|_| invalid_resource("dynamic backing pool is poisoned"))?;
2152        if state.poisoned {
2153            return Err(invalid_resource("dynamic backing pool is fail-closed"));
2154        }
2155        let mut bindings = Vec::with_capacity(segment_count);
2156        for authority in authorities {
2157            for segment in &authority.evidence.segments {
2158                let chunk = state.chunks.get(&segment.chunk_ordinal()).ok_or_else(|| {
2159                    invalid_resource("logical backing references a missing chunk")
2160                })?;
2161                if segment.pool_id() != &authority.evidence.pool_id
2162                    || chunk.backing.identity != *segment.chunk()
2163                    || segment
2164                        .offset_bytes()
2165                        .checked_add(segment.length_bytes())
2166                        .is_none_or(|end| end > chunk.backing.descriptor.size_bytes)
2167                {
2168                    return Err(invalid_resource(
2169                        "logical backing references a stale or out-of-bounds chunk region",
2170                    ));
2171                }
2172                let retention = match authority.reusable_lane {
2173                    Some(lane_id) => DeviceBufferRetention::lane_pair(
2174                        lane_id,
2175                        Arc::clone(&authority.segment_lease),
2176                        Arc::clone(&chunk.backing),
2177                    ),
2178                    None => DeviceBufferRetention::pair(
2179                        Arc::clone(&authority.segment_lease),
2180                        Arc::clone(&chunk.backing),
2181                    ),
2182                };
2183                bindings.push(LogicalBackingSegmentBinding {
2184                    segment: segment.clone(),
2185                    chunk: Arc::clone(&chunk.backing),
2186                    retention,
2187                });
2188            }
2189        }
2190        drop(state);
2191        Ok(LogicalBackingBufferView {
2192            bindings,
2193            authorities,
2194            logical_size_bytes,
2195            capacity_size_bytes,
2196            alignment_bytes: first.evidence.alignment_bytes,
2197            usage: first.evidence.usage,
2198            element_type: first.evidence.element_type,
2199            storage_profile: first.evidence.storage_profile,
2200        })
2201    }
2202
2203    fn validate_authority(
2204        pool: &DynamicBackingPool<R>,
2205        authority: &LogicalBackingSliceAuthority,
2206    ) -> Result<(), VNextError> {
2207        if pool.instance_id != authority.evidence.pool_instance_id
2208            || authority.segment_lease.owner_instance_id != pool.instance_id
2209            || authority.segment_lease.owner.instance_id() != pool.instance_id
2210            || authority.segment_lease.claim_identity != authority.evidence.physical_claim_identity
2211            || authority.segment_lease.segment_generation != authority.evidence.segment_generation
2212            || authority.segment_lease.size_bytes != authority.evidence.physical_size_bytes
2213            || authority.evidence.domain_id != pool.domain.domain_id
2214            || authority.evidence.physical_claim_identity.pool_id() != pool.domain.pool_id()
2215            || authority
2216                .evidence
2217                .physical_claim_identity
2218                .resource_ids()
2219                .binary_search(&authority.evidence.resource_id)
2220                .is_err()
2221            || authority.evidence.logical_size_bytes == 0
2222            || authority.evidence.logical_size_bytes > authority.evidence.capacity_size_bytes
2223            || authority
2224                .evidence
2225                .physical_offset_bytes
2226                .checked_add(authority.evidence.capacity_size_bytes)
2227                .is_none_or(|end| end > authority.evidence.physical_size_bytes)
2228            || authority.evidence.storage_profile != pool.domain.pool.compatibility().profile()
2229        {
2230            return Err(invalid_resource(
2231                "logical backing authority belongs to another dynamic pool instance",
2232            ));
2233        }
2234        let expected_projection = backing_segment_range(
2235            &authority.segment_lease.segments,
2236            authority.evidence.physical_offset_bytes,
2237            authority.evidence.capacity_size_bytes,
2238        )?;
2239        if expected_projection != authority.evidence.segments {
2240            return Err(invalid_resource(
2241                "logical backing projection differs from its shared physical extent",
2242            ));
2243        }
2244        Ok(())
2245    }
2246}