ferrum_interfaces/vnext/resource/

dynamic_pool_set.rs

//! Dynamic pool-set orchestration over backing owned by `dynamic_pool`.

use super::{
    align_up_resource, backing_segment_range, bind_lane_stable_slot_projections,
    compile_program_binding_layouts, compile_submission_wave_domain_layout,
    compile_submission_wave_reusable_capacity_layouts, contiguous_packing_growth_bytes,
    free_extent_layout_fingerprint, invalid_resource, lane_stable_layout_key,
    rollback_free_extent_journal, validate_runtime_descriptor_for_admission,
    AllocatedDynamicGrowth, AllocationLifetime, AllocationSeal, Arc, AtomicU64, BTreeMap,
    BackingChunkIdentity, BackingClaimCertificate, BackingPrepareDecision, BackingSegment,
    BufferRequest, CapacityAvailabilityEpoch, CapacityDomainId, CapacityEntry, CapacityEpochs,
    CapacityUnits, CapacityVector, DeviceAllocationPermit, DeviceBufferRetention,
    DeviceCapacityAvailabilitySnapshot, DeviceCapacityBudget, DeviceCapacityReservation,
    DeviceRuntime, Digest, DynamicBackingBlocker, DynamicBackingClaimOccupancy,
    DynamicBackingClaimResidency, DynamicBackingClaimScope, DynamicBackingDeferralReason,
    DynamicBackingDeferred, DynamicBackingPackingEnvelope, DynamicBackingPool,
    DynamicBackingPoolId, DynamicBackingPoolState, DynamicChunkQuarantineReason,
    DynamicDeviceCapacityBlocked, DynamicPoolDomainSpec, DynamicPoolGrowthIntent,
    DynamicPoolGrowthReceipt, DynamicPoolIdleReclaim, DynamicPoolLiveOccupancyStatus,
    DynamicPoolMaintenanceBoundaryChunk, DynamicPoolMaintenanceBoundaryPool,
    DynamicPoolMaintenanceBoundaryReceipt, DynamicPoolRebalanceReceipt, DynamicResourceShape,
    DynamicStorageView, EvaluatedBackingRequest, ExecutionLane, FreeExtentIndex,
    IdleChunkReclaimCandidate, InvocationLivenessMode, LaneBackingPrepareDecision,
    LaneStableArenaEntry, LaneStableArenaEvictionCandidate, LaneStableArenaLane,
    LaneStableArenaSlot, LaneStableArenaSlotLease, LaneStableArenaState,
    LogicalAdmissionCoordinator, LogicalAdmissionCoordinatorId, LogicalBackingBufferView,
    LogicalBackingSegmentBinding, LogicalBackingSliceAllocationEvidence,
    LogicalBackingSliceAuthority, LogicalBackingSliceEvidence, Mutex, Ordering, PendingGrowthGuard,
    PlanNode, PlannedDynamicGrowth, PreparedBackingClaim, PreparedBackingExtent,
    PreparedLaneBackingClaim, ProgramBindingLayout, QuarantinedDynamicChunk,
    RequestStateHazardCoordinator, ResidentChunkBacking, ResidentChunkState, ResourceId,
    ResourceReservation, ResourceRetentionPolicy, ResourceTransactionIdentity, RunId, Serialize,
    Sha256, StateInitialization, StaticProvisioningBinding, StepResourceSlotKind,
    SubmissionWaveDomainCapacityLayout, SubmissionWaveDomainLayout, TransactionId, VNextError,
    DYNAMIC_POOL_MAINTENANCE_BOUNDARY_SCHEMA_VERSION, NEXT_DYNAMIC_POOL_INSTANCE_ID,
};
use crate::vnext::{
    DeviceCapacityPressure, DynamicPoolResidentPressure, ReusableExecutionBucketId,
    ReusableExecutionMemoryPlan,
};

#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct DynamicPoolGrowthBatchReceipt {
    pub(super) coordinator_id: LogicalAdmissionCoordinatorId,
    pub(super) growths: Vec<DynamicPoolGrowthReceipt>,
    pub(super) capacity_epoch: u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(super) rebalance: Option<DynamicPoolRebalanceReceipt>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(super) maintenance_boundary: Option<DynamicPoolMaintenanceBoundaryReceipt>,
}

impl DynamicPoolGrowthBatchReceipt {
    pub const fn coordinator_id(&self) -> LogicalAdmissionCoordinatorId {
        self.coordinator_id
    }

    pub fn growths(&self) -> &[DynamicPoolGrowthReceipt] {
        &self.growths
    }

    pub const fn capacity_epoch(&self) -> u64 {
        self.capacity_epoch
    }

    pub const fn rebalance(&self) -> Option<&DynamicPoolRebalanceReceipt> {
        self.rebalance.as_ref()
    }

    pub const fn maintenance_boundary(&self) -> Option<&DynamicPoolMaintenanceBoundaryReceipt> {
        self.maintenance_boundary.as_ref()
    }
}

pub(in crate::vnext::resource) struct DynamicPoolSet<R>
where
    R: DeviceRuntime,
{
    pub(in crate::vnext::resource) pools:
        BTreeMap<DynamicBackingPoolId, Arc<DynamicBackingPool<R>>>,
    pub(in crate::vnext::resource) domains: Vec<DynamicPoolDomainSpec>,
    pub(in crate::vnext::resource) nodes: Arc<[PlanNode]>,
    pub(in crate::vnext::resource) submission_wave_layouts: Vec<Option<SubmissionWaveDomainLayout>>,
    pub(in crate::vnext::resource) submission_wave_reusable_capacity_layouts:
        BTreeMap<ReusableExecutionBucketId, Vec<Option<SubmissionWaveDomainCapacityLayout>>>,
    pub(in crate::vnext::resource) program_binding_layouts:
        BTreeMap<ReusableExecutionBucketId, Arc<ProgramBindingLayout>>,
    pub(in crate::vnext::resource) reusable_execution: Option<ReusableExecutionMemoryPlan>,
    pub(in crate::vnext::resource) request_state_hazards: Arc<RequestStateHazardCoordinator>,
    pub(in crate::vnext::resource) logical_admission: LogicalAdmissionCoordinator,
    pub(in crate::vnext::resource) budget: Arc<DeviceCapacityBudget>,
    lane_stable_arenas: Arc<Mutex<LaneStableArenaState>>,
    binding: StaticProvisioningBinding,
    // Backend context must outlive every resident/quarantined buffer above.
    runtime: Arc<R>,
}

pub(in crate::vnext::resource) struct DynamicPoolRebalanceAttempt {
    pub(in crate::vnext::resource) boundary: DynamicPoolMaintenanceBoundaryReceipt,
    pub(in crate::vnext::resource) rebalance: Option<DynamicPoolRebalanceReceipt>,
}

#[derive(Clone, Copy)]
enum ReclaimTargetPolicy {
    PreserveRunnableExtent,
    PreserveAll,
}

impl<R> DynamicPoolSet<R>
where
    R: DeviceRuntime,
{
    pub(in crate::vnext::resource) fn new(
        runtime: Arc<R>,
        binding: StaticProvisioningBinding,
        budget: Arc<DeviceCapacityBudget>,
        logical_admission: LogicalAdmissionCoordinator,
        domains: Vec<DynamicPoolDomainSpec>,
        nodes: Arc<[PlanNode]>,
        reusable_execution: Option<ReusableExecutionMemoryPlan>,
    ) -> Result<Self, VNextError> {
        let request_state_hazards = RequestStateHazardCoordinator::compile(&nodes)?;
        let submission_wave_layouts = domains
            .iter()
            .map(|domain| compile_submission_wave_domain_layout(domain, &nodes))
            .collect::<Result<Vec<_>, _>>()?;
        let submission_wave_reusable_capacity_layouts =
            compile_submission_wave_reusable_capacity_layouts(
                &domains,
                &submission_wave_layouts,
                reusable_execution.as_ref(),
            )?;
        let program_binding_layouts = compile_program_binding_layouts(
            &domains,
            &nodes,
            &submission_wave_layouts,
            &submission_wave_reusable_capacity_layouts,
        )?
        .into_iter()
        .map(|(bucket_id, layout)| (bucket_id, Arc::new(layout)))
        .collect();
        let mut pools = BTreeMap::new();
        for domain in &domains {
            let instance_id = NEXT_DYNAMIC_POOL_INSTANCE_ID
                .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
                    current.checked_add(1)
                })
                .map_err(|_| invalid_resource("dynamic pool instance id space is exhausted"))?;
            let pool = Arc::new(DynamicBackingPool {
                instance_id,
                domain: domain.clone(),
                logical_admission: logical_admission.clone(),
                maintenance: Mutex::new(()),
                next_extent_generation: AtomicU64::new(1),
                state: Mutex::new(DynamicBackingPoolState {
                    resident_bytes: 0,
                    pending_growth_bytes: 0,
                    next_chunk_ordinal: 1,
                    next_chunk_generation: 1,
                    chunks: BTreeMap::new(),
                    allocator: FreeExtentIndex::default(),
                    live_occupancy: DynamicPoolLiveOccupancyStatus::default(),
                    quarantined: Vec::new(),
                    poisoned: false,
                }),
            });
            if pools.insert(domain.pool_id().clone(), pool).is_some() {
                return Err(invalid_resource(
                    "dynamic pool set contains a duplicate pool",
                ));
            }
        }
        Ok(Self {
            runtime,
            binding,
            budget,
            logical_admission,
            domains,
            pools,
            nodes,
            submission_wave_layouts,
            submission_wave_reusable_capacity_layouts,
            program_binding_layouts,
            reusable_execution,
            request_state_hazards,
            lane_stable_arenas: Arc::new(Mutex::new(LaneStableArenaState::default())),
        })
    }

    pub(in crate::vnext::resource) fn program_binding_layout(
        &self,
        bucket_id: &ReusableExecutionBucketId,
    ) -> Option<&Arc<ProgramBindingLayout>> {
        self.program_binding_layouts.get(bucket_id)
    }

    pub(in crate::vnext::resource) const fn maximum_active_sequences(&self) -> u32 {
        self.binding.maximum_active_sequences()
    }

    pub(in crate::vnext::resource) fn write_capacity_availability(
        &self,
        out: &mut Vec<CapacityAvailabilityEpoch>,
    ) -> Result<CapacityEpochs, VNextError> {
        let epochs = self.logical_admission.write_availability_epochs(out)?;
        self.budget.write_availability_epochs(out)?;
        debug_assert!(out
            .windows(2)
            .all(|pair| pair[0].source() < pair[1].source()));
        Ok(epochs)
    }

    /// Rebalances only whole, unreferenced chunks. An entirely idle contiguous
    /// target may replace redundant undersized chunks while retaining its
    /// largest runnable extent; other targets remain excluded. The
    /// batch is selected before mutation, logical totals publish atomically,
    /// and physical grants are returned only after every pool lock is dropped.
    pub(in crate::vnext::resource) fn reclaim_idle_chunks_for_pressure(
        &self,
        pressure: &DeviceCapacityPressure,
        capacity_availability: DeviceCapacityAvailabilitySnapshot,
        excluded_domains: &[CapacityDomainId],
        protected_immediate: &CapacityVector,
        protected_packing_envelopes: &[DynamicBackingPackingEnvelope],
    ) -> Result<DynamicPoolRebalanceAttempt, VNextError> {
        self.reclaim_idle_chunks(
            pressure,
            capacity_availability,
            excluded_domains,
            protected_immediate,
            protected_packing_envelopes,
            ReclaimTargetPolicy::PreserveRunnableExtent,
        )
    }

    fn reclaim_idle_chunks(
        &self,
        pressure: &DeviceCapacityPressure,
        capacity_availability: DeviceCapacityAvailabilitySnapshot,
        excluded_domains: &[CapacityDomainId],
        protected_immediate: &CapacityVector,
        protected_packing_envelopes: &[DynamicBackingPackingEnvelope],
        target_policy: ReclaimTargetPolicy,
    ) -> Result<DynamicPoolRebalanceAttempt, VNextError> {
        if pressure.device_id() != self.runtime.descriptor().id.to_string() {
            return Err(invalid_resource(
                "dynamic pool rebalance received pressure for another device",
            ));
        }
        let deficit = pressure
            .requested_bytes()
            .checked_sub(pressure.available_bytes())
            .ok_or_else(|| invalid_resource("dynamic pool pressure has no reclaimable deficit"))?;
        if deficit == 0 {
            return Err(invalid_resource(
                "dynamic pool pressure has an empty reclaimable deficit",
            ));
        }

        let excluded_domains = excluded_domains
            .iter()
            .copied()
            .collect::<std::collections::BTreeSet<_>>();
        let pools = self.pools.values().cloned().collect::<Vec<_>>();
        let maintenance = pools
            .iter()
            .map(|pool| {
                pool.maintenance
                    .lock()
                    .map_err(|_| invalid_resource("dynamic pool maintenance authority is poisoned"))
            })
            .collect::<Result<Vec<_>, _>>()?;
        let mut states = pools
            .iter()
            .map(|pool| {
                pool.state
                    .lock()
                    .map_err(|_| invalid_resource("dynamic backing pool is poisoned"))
            })
            .collect::<Result<Vec<_>, _>>()?;
        let logical = self.logical_admission.snapshot()?;
        let used_by_domain = logical
            .domains()
            .iter()
            .map(|domain| (domain.domain(), domain.used().get()))
            .collect::<BTreeMap<_, _>>();
        let protected_by_domain = protected_immediate
            .entries()
            .iter()
            .map(|entry| (entry.domain(), entry.units().get()))
            .collect::<BTreeMap<_, _>>();
        let protected_packing_by_domain = protected_packing_envelopes
            .iter()
            .map(|envelope| (envelope.domain_id(), envelope))
            .collect::<BTreeMap<_, _>>();
        if protected_packing_by_domain.len() != protected_packing_envelopes.len() {
            return Err(invalid_resource(
                "dynamic backing protection contains duplicate packing domains",
            ));
        }

        let mut candidates = Vec::new();
        let mut boundary_pools = Vec::with_capacity(pools.len());
        let mut reclaimable_by_pool = vec![0_u64; pools.len()];
        for (pool_index, (pool, state)) in pools.iter().zip(states.iter()).enumerate() {
            if state.poisoned {
                return Err(invalid_resource("dynamic backing pool is fail-closed"));
            }
            let growth_target = excluded_domains.contains(&pool.domain.domain_id);
            let used = used_by_domain
                .get(&pool.domain.domain_id)
                .copied()
                .ok_or_else(|| invalid_resource("dynamic pool domain is absent from admission"))?;
            let physically_occupied = state
                .resident_bytes
                .checked_sub(state.allocator.free_bytes)
                .ok_or_else(|| {
                    invalid_resource("dynamic pool free bytes exceed resident capacity")
                })?;
            if physically_occupied != state.live_occupancy.total().physical_bytes() {
                return Err(invalid_resource(
                    "dynamic pool allocator occupancy differs from its typed live occupancy",
                ));
            }
            let protected = protected_by_domain
                .get(&pool.domain.domain_id)
                .copied()
                .unwrap_or(0);
            let mut protected_packing_satisfied = true;
            let mut protected_chunks = match protected_packing_by_domain.get(&pool.domain.domain_id)
            {
                Some(envelope) => {
                    if envelope.pool_id() != pool.domain.pool_id()
                        || envelope.total_bytes()? != protected
                    {
                        return Err(invalid_resource(
                            "dynamic backing byte and packing protection diverged",
                        ));
                    }
                    let mut allocator = state.allocator.clone();
                    let mut chunks = std::collections::BTreeSet::new();
                    for &claim_bytes in envelope.claim_bytes_descending() {
                        let segments = match pool.domain.pool.compatibility().profile().view() {
                            DynamicStorageView::Contiguous => allocator
                                .allocate_contiguous(pool.domain.pool_id(), claim_bytes)?
                                .map(|segment| vec![segment]),
                            DynamicStorageView::PagedRegions { block_bytes } => allocator
                                .allocate_paged(pool.domain.pool_id(), claim_bytes, block_bytes)?,
                        };
                        let Some(segments) = segments else {
                            chunks.clear();
                            break;
                        };
                        chunks.extend(segments.iter().map(BackingSegment::chunk_ordinal));
                    }
                    if chunks.is_empty() && protected != 0 {
                        protected_packing_satisfied = false;
                    }
                    chunks
                }
                // Logical admission deferrals protect aggregate capacity before
                // an exact physical packing attempt exists.
                None => std::collections::BTreeSet::new(),
            };
            // A larger single contiguous claim cannot use any undersized old
            // chunk. Keeping all of them while allocating one more can trap
            // an otherwise idle pool at the device ceiling. Preserve the
            // largest old extent, including the runnable minimum, and only
            // offer redundant chunks when no physical or logical owner exists.
            let may_replace_target =
                matches!(target_policy, ReclaimTargetPolicy::PreserveRunnableExtent);
            let replacement_keeper = if growth_target
                && may_replace_target
                && !protected_packing_satisfied
                && used == 0
                && physically_occupied == 0
                && state.pending_growth_bytes == 0
                && matches!(
                    pool.domain.pool.compatibility().profile().view(),
                    DynamicStorageView::Contiguous
                )
                && protected_packing_by_domain
                    .get(&pool.domain.domain_id)
                    .is_some_and(|envelope| envelope.claim_bytes_descending() == [protected])
                && state.chunks.values().all(|chunk| {
                    let bytes = chunk.backing._grant.bytes();
                    chunk.live_segments == 0
                        && Arc::strong_count(&chunk.backing) == 1
                        && chunk.backing.descriptor.size_bytes == bytes
                        && state
                            .allocator
                            .by_offset
                            .get(&(chunk.backing.identity.ordinal(), 0))
                            .is_some_and(|extent| {
                                extent.chunk_generation == chunk.backing.identity.generation()
                                    && extent.length_bytes == bytes
                            })
                }) {
                state
                    .chunks
                    .values()
                    .max_by_key(|chunk| chunk.backing._grant.bytes())
                    .filter(|chunk| {
                        let bytes = chunk.backing._grant.bytes();
                        bytes >= pool.domain.pool.provisioning().minimum_resident_bytes()
                            && bytes < protected
                    })
                    .map(|chunk| (chunk.backing.identity.clone(), chunk.backing._grant.bytes()))
            } else {
                None
            };
            if let Some((keeper, _)) = &replacement_keeper {
                // Revalidate the exact packing need. Since every old extent
                // is smaller than the sole claim, removing any non-keeper
                // cannot increase this growth. This is no capacity permit:
                // the ordinary post-reclaim maintenance must reserve anew.
                if contiguous_packing_growth_bytes(
                    &state.allocator,
                    pool.domain.pool_id(),
                    &[protected],
                )? != protected
                {
                    return Err(invalid_resource(
                        "idle target replacement changed its packing need",
                    ));
                }
                protected_chunks.insert(keeper.ordinal());
            }
            let excluded_from_reclaim = growth_target && replacement_keeper.is_none();
            // Logical admission does not own lane-stable or not-yet-committed
            // physical extents. Reclaim must preserve whichever ownership view
            // is larger before adding this bundle's uncommitted demand.
            let owned = used.max(physically_occupied);
            let coherent_runnable_floor = match &replacement_keeper {
                // The pending large claim already cannot run. Keep the whole
                // previous largest extent runnable even if new growth fails.
                Some((_, bytes)) => *bytes,
                None => owned.checked_add(protected).ok_or_else(|| {
                    invalid_resource("dynamic pool protected runnable floor overflows u64")
                })?,
            };
            let resident_floor = pool
                .domain
                .pool
                .provisioning()
                .minimum_resident_bytes()
                .max(coherent_runnable_floor);
            let reclaimable = state.resident_bytes.saturating_sub(resident_floor);
            reclaimable_by_pool[pool_index] = reclaimable;
            let mut boundary_chunks = Vec::with_capacity(state.chunks.len());
            for (&ordinal, chunk) in &state.chunks {
                let chunk_bytes = chunk.backing._grant.bytes();
                let full_extent = state.allocator.by_offset.get(&(ordinal, 0));
                let external_references = Arc::strong_count(&chunk.backing).saturating_sub(1);
                let protected_packing = protected_chunks.contains(&ordinal);
                let full_extent_available = chunk.backing.descriptor.size_bytes == chunk_bytes
                    && full_extent.is_some_and(|extent| {
                        extent.chunk_generation == chunk.backing.identity.generation()
                            && extent.length_bytes == chunk_bytes
                    });
                let resident_floor_allows_reclaim = chunk_bytes <= reclaimable;
                let reclaim_candidate = !excluded_from_reclaim
                    && state.pending_growth_bytes == 0
                    && (protected_packing_satisfied || replacement_keeper.is_some())
                    && chunk.live_segments == 0
                    && external_references == 0
                    && !protected_packing
                    && resident_floor_allows_reclaim
                    && full_extent_available;
                boundary_chunks.push(DynamicPoolMaintenanceBoundaryChunk {
                    identity: chunk.backing.identity.clone(),
                    bytes: chunk_bytes,
                    live_segments: chunk.live_segments,
                    external_references,
                    protected_packing,
                    full_extent_available,
                    resident_floor_allows_reclaim,
                    reclaim_candidate,
                });
                if reclaim_candidate {
                    candidates.push(IdleChunkReclaimCandidate {
                        pool_index,
                        chunk: chunk.backing.identity.clone(),
                        chunk_bytes,
                    });
                }
            }
            boundary_pools.push(DynamicPoolMaintenanceBoundaryPool {
                pool_id: pool.domain.pool_id().clone(),
                domain_id: pool.domain.domain_id,
                excluded_from_reclaim,
                resident_bytes: state.resident_bytes,
                pending_growth_bytes: state.pending_growth_bytes,
                free_bytes: state.allocator.free_bytes,
                largest_contiguous_bytes: state.allocator.largest_contiguous_bytes(),
                free_extent_layout_fingerprint: free_extent_layout_fingerprint(&state.allocator),
                logical_used_bytes: used,
                live_occupancy: state.live_occupancy,
                minimum_resident_bytes: pool.domain.pool.provisioning().minimum_resident_bytes(),
                maximum_resident_bytes: pool.domain.pool.provisioning().maximum_resident_bytes(),
                protected_immediate_bytes: protected,
                protected_packing_satisfied,
                idle_target_replacement_kept_chunk: replacement_keeper.map(|(keeper, _)| keeper),
                coherent_runnable_floor_bytes: coherent_runnable_floor,
                resident_floor_bytes: resident_floor,
                reclaimable_bytes: reclaimable,
                chunks: boundary_chunks,
            });
        }

        let reclaim_candidate_chunks = candidates.len();
        let reclaim_candidate_bytes = candidates.iter().try_fold(0_u64, |total, candidate| {
            total
                .checked_add(candidate.chunk_bytes)
                .ok_or_else(|| invalid_resource("dynamic reclaim candidate bytes overflow u64"))
        })?;
        let mut selected = Vec::<IdleChunkReclaimCandidate>::new();
        let mut selected_by_pool = vec![0_u64; pools.len()];
        let mut reclaimed_bytes = 0_u64;
        let best_single = candidates
            .iter()
            .filter(|candidate| candidate.chunk_bytes >= deficit)
            .min_by(|left, right| {
                left.chunk_bytes
                    .cmp(&right.chunk_bytes)
                    .then_with(|| left.pool_index.cmp(&right.pool_index))
                    .then_with(|| right.chunk.ordinal().cmp(&left.chunk.ordinal()))
            })
            .cloned();
        if let Some(candidate) = best_single {
            selected_by_pool[candidate.pool_index] = candidate.chunk_bytes;
            reclaimed_bytes = candidate.chunk_bytes;
            selected.push(candidate);
        } else {
            candidates.sort_by(|left, right| {
                right
                    .chunk_bytes
                    .cmp(&left.chunk_bytes)
                    .then_with(|| left.pool_index.cmp(&right.pool_index))
                    .then_with(|| right.chunk.ordinal().cmp(&left.chunk.ordinal()))
            });
            for candidate in candidates {
                let next_pool_total = selected_by_pool[candidate.pool_index]
                    .checked_add(candidate.chunk_bytes)
                    .ok_or_else(|| invalid_resource("dynamic reclaim bytes overflow u64"))?;
                if next_pool_total > reclaimable_by_pool[candidate.pool_index] {
                    continue;
                }
                selected_by_pool[candidate.pool_index] = next_pool_total;
                reclaimed_bytes = reclaimed_bytes
                    .checked_add(candidate.chunk_bytes)
                    .ok_or_else(|| invalid_resource("dynamic reclaim bytes overflow u64"))?;
                selected.push(candidate);
                if reclaimed_bytes >= deficit {
                    break;
                }
            }
        }
        selected.sort_by(|left, right| {
            left.pool_index
                .cmp(&right.pool_index)
                .then_with(|| left.chunk.ordinal().cmp(&right.chunk.ordinal()))
        });
        let mut planned_domains = excluded_domains.iter().copied().collect::<Vec<_>>();
        planned_domains.sort_unstable();
        let mut protected_packing_envelopes = protected_packing_envelopes.to_vec();
        protected_packing_envelopes.sort_by_key(DynamicBackingPackingEnvelope::domain_id);
        let boundary = DynamicPoolMaintenanceBoundaryReceipt {
            schema_version: DYNAMIC_POOL_MAINTENANCE_BOUNDARY_SCHEMA_VERSION,
            coordinator_id: logical.coordinator_id(),
            logical_release_epoch: logical.release_epoch(),
            logical_capacity_epoch: logical.capacity_epoch(),
            plan_device_capacity_epoch: capacity_availability.plan_epoch(),
            process_device_capacity_epoch: capacity_availability.process_epoch(),
            pressure: pressure.clone(),
            planned_domains,
            protected_immediate: protected_immediate.clone(),
            protected_packing_envelopes,
            pools: boundary_pools,
            reclaim_candidate_chunks,
            reclaim_candidate_bytes,
            selected_chunks: selected
                .iter()
                .map(|candidate| candidate.chunk.clone())
                .collect(),
            selected_bytes: reclaimed_bytes,
            reclaim_sufficient: reclaimed_bytes >= deficit,
        };
        if reclaimed_bytes < deficit {
            return Ok(DynamicPoolRebalanceAttempt {
                boundary,
                rebalance: None,
            });
        }

        for candidate in &selected {
            let state = &states[candidate.pool_index];
            let chunk = state
                .chunks
                .get(&candidate.chunk.ordinal())
                .ok_or_else(|| invalid_resource("selected dynamic reclaim chunk disappeared"))?;
            let extent = state
                .allocator
                .by_offset
                .get(&(candidate.chunk.ordinal(), 0))
                .ok_or_else(|| invalid_resource("selected dynamic reclaim extent disappeared"))?;
            if chunk.live_segments != 0
                || Arc::strong_count(&chunk.backing) != 1
                || chunk.backing.identity != candidate.chunk
                || extent.chunk_generation != candidate.chunk.generation()
                || extent.length_bytes != candidate.chunk_bytes
            {
                return Err(invalid_resource(
                    "selected dynamic reclaim chunk changed before publication",
                ));
            }
        }

        let published_totals = states
            .iter()
            .zip(&selected_by_pool)
            .map(|(state, &selected_bytes)| {
                state
                    .resident_bytes
                    .checked_sub(selected_bytes)
                    .ok_or_else(|| invalid_resource("dynamic reclaim resident bytes underflow"))
            })
            .collect::<Result<Vec<_>, _>>()?;
        let mut removed = Vec::with_capacity(selected.len());
        for candidate in &selected {
            let state = &mut states[candidate.pool_index];
            let extent = state
                .allocator
                .remove_extent(candidate.chunk.ordinal(), 0)
                .expect("validated idle chunk retains its exact full extent");
            debug_assert_eq!(extent.chunk_generation, candidate.chunk.generation());
            debug_assert_eq!(extent.length_bytes, candidate.chunk_bytes);
            let chunk = state
                .chunks
                .remove(&candidate.chunk.ordinal())
                .expect("validated idle chunk remains resident");
            removed.push((candidate.clone(), chunk));
        }
        let updates = selected_by_pool
            .iter()
            .enumerate()
            .filter(|(_, selected_bytes)| **selected_bytes != 0)
            .map(|(pool_index, _)| {
                (
                    pools[pool_index].domain.domain_id,
                    CapacityUnits::new(published_totals[pool_index]),
                )
            })
            .collect::<Vec<_>>();
        let epochs = match self.logical_admission.set_domain_totals(&updates) {
            Ok(epochs) => epochs,
            Err(error) => {
                for (candidate, chunk) in removed.drain(..).rev() {
                    let state = &mut states[candidate.pool_index];
                    state
                        .allocator
                        .insert_extent(
                            candidate.chunk.ordinal(),
                            candidate.chunk.generation(),
                            0,
                            candidate.chunk_bytes,
                        )
                        .expect("unpublished idle chunk extent can be restored");
                    assert!(state
                        .chunks
                        .insert(candidate.chunk.ordinal(), chunk)
                        .is_none());
                }
                return Err(error);
            }
        };
        for (state, &published_total) in states.iter_mut().zip(&published_totals) {
            state.resident_bytes = published_total;
        }

        let mut pool_receipts = Vec::new();
        for (pool_index, &pool_reclaimed_bytes) in selected_by_pool.iter().enumerate() {
            if pool_reclaimed_bytes == 0 {
                continue;
            }
            pool_receipts.push(DynamicPoolIdleReclaim {
                pool_id: pools[pool_index].domain.pool_id().clone(),
                chunks: selected
                    .iter()
                    .filter(|candidate| candidate.pool_index == pool_index)
                    .map(|candidate| candidate.chunk.clone())
                    .collect(),
                reclaimed_bytes: pool_reclaimed_bytes,
                published_capacity_bytes: published_totals[pool_index],
            });
        }
        let reclaimed_chunks = selected.len();
        drop(states);
        drop(maintenance);
        drop(removed);
        let availability = self.budget.availability_snapshot()?;
        Ok(DynamicPoolRebalanceAttempt {
            boundary,
            rebalance: Some(DynamicPoolRebalanceReceipt {
                pools: pool_receipts,
                reclaimed_chunks,
                reclaimed_bytes,
                logical_capacity_epoch: epochs.capacity_epoch(),
                plan_device_capacity_epoch: availability.plan_epoch(),
                process_device_capacity_epoch: availability.process_epoch(),
            }),
        })
    }

    pub(in crate::vnext::resource) fn maintain_pools(
        &self,
        intents: Vec<DynamicPoolGrowthIntent>,
    ) -> Result<DynamicPoolGrowthBatchReceipt, VNextError> {
        let mut ignored_capacity_block = None;
        self.maintain_pools_observed(intents, &mut ignored_capacity_block)
    }

    pub(in crate::vnext::resource) fn maintain_pools_observed(
        &self,
        mut intents: Vec<DynamicPoolGrowthIntent>,
        capacity_blocked: &mut Option<DynamicDeviceCapacityBlocked>,
    ) -> Result<DynamicPoolGrowthBatchReceipt, VNextError> {
        intents.sort_by(|left, right| left.pool_id().cmp(right.pool_id()));
        if intents
            .windows(2)
            .any(|pair| pair[0].pool_id() == pair[1].pool_id())
        {
            return Err(invalid_resource(
                "dynamic maintenance batch contains a duplicate pool",
            ));
        }
        let pools = intents
            .iter()
            .map(|intent| {
                self.pools.get(intent.pool_id()).cloned().ok_or_else(|| {
                    invalid_resource("dynamic maintenance references an unknown pool")
                })
            })
            .collect::<Result<Vec<_>, _>>()?;
        let _maintenance = pools
            .iter()
            .map(|pool| {
                pool.maintenance
                    .lock()
                    .map_err(|_| invalid_resource("dynamic pool maintenance authority is poisoned"))
            })
            .collect::<Result<Vec<_>, _>>()?;

        let mut planned = Vec::with_capacity(intents.len());
        let mut pending = Vec::with_capacity(intents.len());
        for (intent, pool) in intents.iter().zip(&pools) {
            let requested_bytes = {
                let state = pool
                    .state
                    .lock()
                    .map_err(|_| invalid_resource("dynamic backing pool is poisoned"))?;
                if state.poisoned {
                    return Err(invalid_resource("dynamic backing pool is fail-closed"));
                }
                match intent {
                    DynamicPoolGrowthIntent::Additional(request) => request.requested_bytes(),
                    DynamicPoolGrowthIntent::Minimum(_) => {
                        let current = state
                            .resident_bytes
                            .checked_add(state.pending_growth_bytes)
                            .ok_or_else(|| {
                                invalid_resource("dynamic pool residency overflows u64")
                            })?;
                        let minimum = pool.domain.pool.provisioning().minimum_resident_bytes();
                        if current >= minimum {
                            continue;
                        }
                        minimum - current
                    }
                    DynamicPoolGrowthIntent::RevalidatedDeferral(blocker) => {
                        if let Some(claim_bytes) = blocker.contiguous_claim_bytes_descending() {
                            let required_growth = contiguous_packing_growth_bytes(
                                &state.allocator,
                                pool.domain.pool_id(),
                                claim_bytes,
                            )?;
                            if required_growth == 0 {
                                continue;
                            }
                            required_growth
                        } else {
                            match blocker.reason() {
                                DynamicBackingDeferralReason::GrowthRequired => {
                                    let required_free = blocker
                                        .free_bytes()
                                        .checked_add(blocker.requested_bytes())
                                        .ok_or_else(|| {
                                            invalid_resource(
                                            "dynamic backing deferred requirement overflows u64",
                                        )
                                        })?;
                                    if state.allocator.free_bytes >= required_free {
                                        continue;
                                    }
                                    required_free - state.allocator.free_bytes
                                }
                                DynamicBackingDeferralReason::FragmentedContiguous => {
                                    return Err(invalid_resource(
                                        "fragmented contiguous blocker lost its transaction demand",
                                    ));
                                }
                            }
                        }
                    }
                    DynamicPoolGrowthIntent::RevalidatedAdmissionPressure {
                        required_free_bytes,
                        ..
                    } => {
                        // A logical lease may precede its physical prepare;
                        // conversely physical release precedes logical release.
                        // Neither ledger alone proves that the demand now fits.
                        let logical = self.logical_admission.snapshot()?;
                        if logical.poisoned() {
                            return Err(invalid_resource("logical admission is fail-closed"));
                        }
                        let available = logical
                            .domains()
                            .iter()
                            .find(|domain| domain.domain() == pool.domain.domain_id)
                            .ok_or_else(|| invalid_resource("pool has no logical capacity domain"))?
                            .available()
                            .get();
                        let missing = required_free_bytes
                            .saturating_sub(available)
                            .max(required_free_bytes.saturating_sub(state.allocator.free_bytes));
                        if missing == 0 {
                            continue;
                        }
                        missing
                    }
                }
            };
            let chunk_bytes = align_up_resource(requested_bytes, pool.allocation_quantum())?;
            let chunk = {
                let mut state = pool
                    .state
                    .lock()
                    .map_err(|_| invalid_resource("dynamic backing pool is poisoned"))?;
                if state.poisoned {
                    return Err(invalid_resource("dynamic backing pool is fail-closed"));
                }
                let ordinal = state.next_chunk_ordinal;
                let generation = state.next_chunk_generation;
                let next_ordinal = ordinal
                    .checked_add(1)
                    .ok_or_else(|| invalid_resource("dynamic chunk ordinal space is exhausted"))?;
                let next_generation = generation.checked_add(1).ok_or_else(|| {
                    invalid_resource("dynamic chunk generation space is exhausted")
                })?;
                let next_pending = state
                    .pending_growth_bytes
                    .checked_add(chunk_bytes)
                    .ok_or_else(|| invalid_resource("pending dynamic growth bytes overflow u64"))?;
                state.next_chunk_ordinal = next_ordinal;
                state.next_chunk_generation = next_generation;
                state.pending_growth_bytes = next_pending;
                BackingChunkIdentity::from_parts(
                    pool.domain.pool_id().clone(),
                    ordinal,
                    generation,
                )?
            };
            pending.push(PendingGrowthGuard {
                pool: Arc::clone(pool),
                bytes: chunk_bytes,
                armed: true,
            });
            let expected_resource_id = ResourceId::new(format!(
                "{}/chunk/{}/{}",
                pool.domain.pool_id().as_str(),
                chunk.ordinal(),
                chunk.generation()
            ))?;
            planned.push(PlannedDynamicGrowth {
                pool: Arc::clone(pool),
                chunk,
                expected_resource_id,
                chunk_bytes,
            });
        }
        if planned.is_empty() {
            return Ok(DynamicPoolGrowthBatchReceipt {
                coordinator_id: self.logical_admission.id(),
                growths: Vec::new(),
                capacity_epoch: self.logical_admission.epochs()?.capacity_epoch(),
                rebalance: None,
                maintenance_boundary: None,
            });
        }

        let total_bytes = planned.iter().try_fold(0_u64, |total, growth| {
            total
                .checked_add(growth.chunk_bytes)
                .ok_or_else(|| invalid_resource("dynamic maintenance batch bytes overflow u64"))
        })?;
        let capacity_availability = self.budget.availability_snapshot()?;
        let reservation = match DeviceCapacityReservation::reserve(&self.budget, total_bytes) {
            Ok(reservation) => reservation,
            Err(VNextError::DeviceCapacityUnavailable(pressure)) => {
                *capacity_blocked = Some(DynamicDeviceCapacityBlocked {
                    pressure: pressure.clone(),
                    availability: capacity_availability,
                    planned_domains: planned
                        .iter()
                        .map(|growth| growth.pool.domain.domain_id)
                        .collect(),
                });
                return Err(VNextError::DeviceCapacityUnavailable(pressure));
            }
            Err(error) => return Err(error),
        };
        // Device-budget saturation is recoverable pressure even when the same
        // growth also crosses a pool's device-derived resident ceiling. Only a
        // growth that the authoritative device budget accepted can prove that
        // the remaining pool ceiling is a terminal theoretical-plan violation.
        for growth in &planned {
            let state = growth
                .pool
                .state
                .lock()
                .map_err(|_| invalid_resource("dynamic backing pool is poisoned"))?;
            let next_residency = state
                .resident_bytes
                .checked_add(state.pending_growth_bytes)
                .ok_or_else(|| invalid_resource("dynamic pool resident bytes overflow u64"))?;
            if next_residency
                > growth
                    .pool
                    .domain
                    .pool
                    .provisioning()
                    .maximum_resident_bytes()
            {
                return Err(VNextError::DynamicPoolResidentUnavailable(
                    DynamicPoolResidentPressure::new(
                        growth.pool.domain.pool_id().clone(),
                        growth.chunk_bytes,
                        state.resident_bytes,
                        growth
                            .pool
                            .domain
                            .pool
                            .provisioning()
                            .maximum_resident_bytes(),
                    )?,
                ));
            }
        }
        let grants = reservation.commit_split(
            &planned
                .iter()
                .map(|growth| growth.chunk_bytes)
                .collect::<Vec<_>>(),
        )?;
        let mut allocated = Vec::with_capacity(planned.len());
        for (growth, grant) in planned.iter().zip(grants) {
            let transaction_identity = ResourceTransactionIdentity {
                pool_id: self.binding.pool_id(),
                run_id: RunId::new(format!("dynamic-grow-{}", growth.chunk.generation()))?,
                transaction_id: TransactionId::new(format!(
                    "dynamic-grow-{}-{}",
                    growth.chunk.ordinal(),
                    growth.chunk.generation()
                ))?,
                request_id: self.binding.request_id().clone(),
            };
            let reservation_evidence = ResourceReservation {
                resource_id: growth.expected_resource_id.clone(),
                request_id: self.binding.request_id().clone(),
                owner_node_id: None,
                size_bytes: growth.chunk_bytes,
                alignment_bytes: growth.pool.domain.pool.compatibility().alignment_bytes(),
                usage: growth.pool.domain.pool.compatibility().usage(),
                element_type: growth.pool.domain.pool.compatibility().element_type(),
                retention_policy: ResourceRetentionPolicy::Plan,
                backing_domain_id: Some(growth.pool.domain.domain_id),
                generation: growth.chunk.generation(),
            };
            let request = BufferRequest::new(
                growth.expected_resource_id.clone(),
                growth.chunk_bytes,
                reservation_evidence.alignment_bytes,
                reservation_evidence.usage,
                reservation_evidence.element_type,
            )?;
            validate_runtime_descriptor_for_admission(
                self.runtime.descriptor(),
                &self.binding,
                "dynamic pool batch growth preflight",
            )?;
            let buffer = self
                .runtime
                .allocate(DeviceAllocationPermit {
                    identity: &transaction_identity,
                    binding: &self.binding,
                    reservation: &reservation_evidence,
                    request: &request,
                    seal: AllocationSeal,
                })
                .map_err(|error| {
                    invalid_resource(format!("dynamic pool device allocation failed: {error}"))
                })?;
            let actual_descriptor = self.runtime.buffer_descriptor(&buffer);
            validate_runtime_descriptor_for_admission(
                self.runtime.descriptor(),
                &self.binding,
                "dynamic pool batch growth completion",
            )?;
            if grant.bytes() != growth.chunk_bytes {
                return Err(invalid_resource(
                    "dynamic pool capacity grant differs from its chunk",
                ));
            }
            let backing = Arc::new(ResidentChunkBacking {
                buffer,
                _grant: grant,
                identity: growth.chunk.clone(),
                descriptor: actual_descriptor.clone(),
            });
            if !reservation_evidence.matches_descriptor(&actual_descriptor) {
                let mut state = growth
                    .pool
                    .state
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner);
                state.quarantined.push(QuarantinedDynamicChunk {
                    backing,
                    reason: DynamicChunkQuarantineReason::DescriptorMismatch,
                });
                return Err(invalid_resource(
                    "dynamic chunk descriptor mismatch was quarantined without capacity publication",
                ));
            }
            allocated.push(AllocatedDynamicGrowth { backing });
        }

        let mut states = planned
            .iter()
            .map(|growth| {
                growth.pool.state.lock().map_err(|_| {
                    invalid_resource("dynamic backing pool is poisoned after allocation")
                })
            })
            .collect::<Result<Vec<_>, _>>()?;
        let mut published_totals = Vec::with_capacity(planned.len());
        for ((growth, allocation), state) in planned.iter().zip(&allocated).zip(&states) {
            if state.poisoned
                || state.pending_growth_bytes < growth.chunk_bytes
                || state.chunks.contains_key(&growth.chunk.ordinal())
                || state
                    .allocator
                    .by_offset
                    .range((growth.chunk.ordinal(), 0)..=(growth.chunk.ordinal(), u64::MAX))
                    .next()
                    .is_some()
                || state
                    .allocator
                    .free_bytes
                    .checked_add(growth.chunk_bytes)
                    .is_none()
                || allocation.backing.identity != growth.chunk
            {
                return Err(invalid_resource(
                    "dynamic batch installation preconditions changed before publication",
                ));
            }
            published_totals.push(
                state
                    .resident_bytes
                    .checked_add(growth.chunk_bytes)
                    .ok_or_else(|| invalid_resource("published dynamic capacity overflows u64"))?,
            );
        }
        for index in 0..planned.len() {
            let growth = &planned[index];
            let state = &mut states[index];
            state.pending_growth_bytes -= growth.chunk_bytes;
            pending[index].disarm();
            state
                .allocator
                .insert_extent(
                    growth.chunk.ordinal(),
                    growth.chunk.generation(),
                    0,
                    growth.chunk_bytes,
                )
                .expect("validated new chunk has one disjoint free extent");
            state.chunks.insert(
                growth.chunk.ordinal(),
                ResidentChunkState {
                    backing: Arc::clone(&allocated[index].backing),
                    live_segments: 0,
                },
            );
        }
        let updates = planned
            .iter()
            .zip(&published_totals)
            .map(|(growth, &total)| (growth.pool.domain.domain_id, CapacityUnits::new(total)))
            .collect::<Vec<_>>();
        let epochs = match self.logical_admission.set_domain_totals(&updates) {
            Ok(epochs) => epochs,
            Err(error) => {
                for index in 0..planned.len() {
                    states[index]
                        .allocator
                        .remove_extent(planned[index].chunk.ordinal(), 0)
                        .expect("unpublished dynamic chunk free extent remains installed");
                    let removed = states[index]
                        .chunks
                        .remove(&planned[index].chunk.ordinal())
                        .expect("unpublished dynamic chunk remains installed");
                    states[index].quarantined.push(QuarantinedDynamicChunk {
                        backing: removed.backing,
                        reason: DynamicChunkQuarantineReason::PublicationRejected,
                    });
                }
                return Err(error);
            }
        };
        for (state, &published_total) in states.iter_mut().zip(&published_totals) {
            state.resident_bytes = published_total;
        }
        Ok(DynamicPoolGrowthBatchReceipt {
            coordinator_id: self.logical_admission.id(),
            growths: planned
                .iter()
                .zip(published_totals)
                .map(
                    |(growth, published_capacity_bytes)| DynamicPoolGrowthReceipt {
                        pool_id: growth.pool.domain.pool_id().clone(),
                        chunk: growth.chunk.clone(),
                        chunk_bytes: growth.chunk_bytes,
                        published_capacity_bytes,
                        capacity_epoch: epochs.capacity_epoch(),
                    },
                )
                .collect(),
            capacity_epoch: epochs.capacity_epoch(),
            rebalance: None,
            maintenance_boundary: None,
        })
    }

    pub(in crate::vnext::resource) fn prepare_claim(
        &self,
        requests: &[EvaluatedBackingRequest<'_>],
    ) -> Result<BackingPrepareDecision<R>, VNextError> {
        if requests.is_empty() {
            return Ok(BackingPrepareDecision::Prepared(
                PreparedBackingClaim::empty(),
            ));
        }
        let lifetime = requests
            .first()
            .and_then(|request| request.projections.first())
            .map(|projection| projection.descriptor.lifetime())
            .ok_or_else(|| invalid_resource("dynamic backing request has no projection"))?;
        self.prepare_claim_scoped(
            requests,
            DynamicBackingClaimScope::from(lifetime),
            DynamicBackingClaimResidency::Transient,
        )
    }

    fn reusable_capacity_shape_for_requests(
        &self,
        requests: &[EvaluatedBackingRequest<'_>],
    ) -> Result<Option<DynamicResourceShape>, VNextError> {
        let reusable_execution_bucket_id = requests
            .first()
            .and_then(|request| request.reusable_execution_bucket_id.as_ref());
        if requests.iter().any(|request| {
            request.reusable_execution_bucket_id.as_ref() != reusable_execution_bucket_id
        }) {
            return Err(invalid_resource(
                "one dynamic backing claim cannot mix reusable execution buckets",
            ));
        }
        reusable_execution_bucket_id
            .map(|bucket_id| {
                let bucket = self
                    .reusable_execution
                    .as_ref()
                    .and_then(|plan| plan.bucket(bucket_id))
                    .map(|resolved| resolved.bucket())
                    .ok_or_else(|| {
                        invalid_resource(
                            "dynamic backing claim references a reusable bucket outside its immutable plan",
                        )
                    })?;
                Ok(DynamicResourceShape::from_validated(
                    bucket.capacity().maximum_sequences(),
                    bucket.capacity().maximum_tokens(),
                    bucket.capacity().maximum_pages(),
                ))
            })
            .transpose()
    }

    pub(in crate::vnext::resource) fn prepare_lane_stable_claim(
        self: &Arc<Self>,
        lane: &Arc<ExecutionLane<R>>,
        requests: &[EvaluatedBackingRequest<'_>],
    ) -> Result<LaneBackingPrepareDecision, VNextError> {
        if !Arc::ptr_eq(&self.runtime, lane.runtime_arc())
            || lane.descriptor() != self.runtime.descriptor()
            || !lane.is_reusable()
        {
            return Err(invalid_resource(
                "lane-stable backing requires the reusable execution lane bound to this plan runtime",
            ));
        }
        if requests.is_empty() {
            return Ok(LaneBackingPrepareDecision::Prepared(
                PreparedLaneBackingClaim::new(Vec::new(), None)?,
            ));
        }
        let reusable_capacity_shape = self.reusable_capacity_shape_for_requests(requests)?;
        if reusable_capacity_shape.is_none() {
            return match self.prepare_claim(requests)? {
                BackingPrepareDecision::Prepared(prepared) => {
                    Ok(LaneBackingPrepareDecision::Prepared(
                        PreparedLaneBackingClaim::new(prepared.commit(), None)?,
                    ))
                }
                BackingPrepareDecision::Deferred(deferred) => {
                    Ok(LaneBackingPrepareDecision::Deferred(deferred))
                }
            };
        }
        let lifetime = requests
            .first()
            .and_then(|request| request.projections.first())
            .map(|projection| projection.descriptor.lifetime())
            .ok_or_else(|| invalid_resource("dynamic backing request has no projection"))?;
        if !matches!(
            lifetime,
            AllocationLifetime::Step | AllocationLifetime::Invocation
        ) || requests.iter().any(|request| {
            request.projections.is_empty()
                || request.projections.iter().any(|projection| {
                    projection.descriptor.lifetime() != lifetime
                        || projection.descriptor.initialization() != StateInitialization::None
                })
        }) {
            return Err(invalid_resource(
                "lane-stable backing accepts only non-initialized Step or Invocation resources",
            ));
        }

        let mut canonical_requests = requests.iter().collect::<Vec<_>>();
        canonical_requests
            .sort_unstable_by(|left, right| left.claim_identity.cmp(&right.claim_identity));
        let key = lane_stable_layout_key(lane.id(), lifetime, &canonical_requests)?;
        let lane_owner: Arc<dyn LaneStableArenaLane> =
            Arc::clone(lane) as Arc<dyn LaneStableArenaLane>;

        {
            {
                let mut arenas = self
                    .lane_stable_arenas
                    .lock()
                    .map_err(|_| invalid_resource("lane-stable arena registry is poisoned"))?;
                if arenas.poisoned {
                    return Err(invalid_resource(
                        "lane-stable arena registry is fail-closed",
                    ));
                }
                let now = arenas.tick();
                if let Some(entry) = arenas.entries.get_mut(&key) {
                    let owner = entry.lane.upgrade().ok_or_else(|| {
                        invalid_resource("lane-stable arena retained an expired execution lane")
                    })?;
                    if !Arc::ptr_eq(&owner, &lane_owner) {
                        return Err(invalid_resource(
                            "lane-stable arena identity aliases another execution lane",
                        ));
                    }
                    if let Some((slot_id, stable, certificate, slot_domains)) =
                        entry.claim_idle_slot(lane.id(), now, &canonical_requests)?
                    {
                        return Ok(LaneBackingPrepareDecision::Prepared(
                            PreparedLaneBackingClaim::certified(
                                stable,
                                certificate,
                                LaneStableArenaSlotLease {
                                    arenas: Arc::clone(&self.lane_stable_arenas),
                                    logical_admission: self.logical_admission.clone(),
                                    availability_domains: slot_domains,
                                    key: key.clone(),
                                    slot_id,
                                },
                            ),
                        ));
                    }
                }
            }

            match self.prepare_claim_scoped(
                requests,
                DynamicBackingClaimScope::from(lifetime),
                DynamicBackingClaimResidency::LaneStable,
            )? {
                BackingPrepareDecision::Prepared(prepared) => {
                    let mut arenas = self
                        .lane_stable_arenas
                        .lock()
                        .map_err(|_| invalid_resource("lane-stable arena registry is poisoned"))?;
                    if arenas.poisoned {
                        return Err(invalid_resource(
                            "lane-stable arena registry is fail-closed",
                        ));
                    }
                    let now = arenas.tick();
                    if let Some(entry) = arenas.entries.get_mut(&key) {
                        let owner = entry.lane.upgrade().ok_or_else(|| {
                            invalid_resource("lane-stable arena retained an expired execution lane")
                        })?;
                        if !Arc::ptr_eq(&owner, &lane_owner) {
                            return Err(invalid_resource(
                                "lane-stable arena identity aliases another execution lane",
                            ));
                        }
                        if let Some((slot_id, stable, certificate, slot_domains)) =
                            entry.claim_idle_slot(lane.id(), now, &canonical_requests)?
                        {
                            drop(arenas);
                            drop(prepared);
                            return Ok(LaneBackingPrepareDecision::Prepared(
                                PreparedLaneBackingClaim::certified(
                                    stable,
                                    certificate,
                                    LaneStableArenaSlotLease {
                                        arenas: Arc::clone(&self.lane_stable_arenas),
                                        logical_admission: self.logical_admission.clone(),
                                        availability_domains: slot_domains,
                                        key: key.clone(),
                                        slot_id,
                                    },
                                ),
                            ));
                        }
                    }
                    let authorities = prepared.commit();
                    let projection_bindings =
                        bind_lane_stable_slot_projections(&authorities, &canonical_requests)?;
                    let certificate = Arc::new(BackingClaimCertificate::from_slices(&authorities)?);
                    let stable = authorities
                        .iter()
                        .map(|authority| authority.retained_for_lane(lane.id()))
                        .collect();
                    let availability_domains = requests
                        .iter()
                        .map(|request| request.domain.domain_id())
                        .collect::<std::collections::BTreeSet<_>>()
                        .into_iter()
                        .collect::<Vec<_>>();
                    let slot_id = arenas.issue_slot_id()?;
                    let entry =
                        arenas
                            .entries
                            .entry(key.clone())
                            .or_insert_with(|| LaneStableArenaEntry {
                                lane: Arc::downgrade(&lane_owner),
                                slots: BTreeMap::new(),
                            });
                    if entry
                        .slots
                        .insert(
                            slot_id,
                            LaneStableArenaSlot {
                                slot_id,
                                authorities,
                                certificate: Arc::clone(&certificate),
                                projection_bindings,
                                availability_domains: availability_domains.clone(),
                                in_use: true,
                                last_used: now,
                            },
                        )
                        .is_some()
                    {
                        arenas.poisoned = true;
                        return Err(invalid_resource(
                            "lane-stable arena slot publication replaced an existing slot",
                        ));
                    }
                    return Ok(LaneBackingPrepareDecision::Prepared(
                        PreparedLaneBackingClaim::certified(
                            stable,
                            certificate,
                            LaneStableArenaSlotLease {
                                arenas: Arc::clone(&self.lane_stable_arenas),
                                logical_admission: self.logical_admission.clone(),
                                availability_domains: availability_domains.clone(),
                                key: key.clone(),
                                slot_id,
                            },
                        ),
                    ));
                }
                BackingPrepareDecision::Deferred(deferred) => {
                    let mut arenas = self
                        .lane_stable_arenas
                        .lock()
                        .map_err(|_| invalid_resource("lane-stable arena registry is poisoned"))?;
                    if arenas.poisoned {
                        return Err(invalid_resource(
                            "lane-stable arena registry is fail-closed",
                        ));
                    }
                    let now = arenas.tick();
                    if let Some(entry) = arenas.entries.get_mut(&key) {
                        let owner = entry.lane.upgrade().ok_or_else(|| {
                            invalid_resource("lane-stable arena retained an expired execution lane")
                        })?;
                        if !Arc::ptr_eq(&owner, &lane_owner) {
                            return Err(invalid_resource(
                                "lane-stable arena identity aliases another execution lane",
                            ));
                        }
                        if let Some((slot_id, stable, certificate, slot_domains)) =
                            entry.claim_idle_slot(lane.id(), now, &canonical_requests)?
                        {
                            drop(arenas);
                            drop(deferred);
                            return Ok(LaneBackingPrepareDecision::Prepared(
                                PreparedLaneBackingClaim::certified(
                                    stable,
                                    certificate,
                                    LaneStableArenaSlotLease {
                                        arenas: Arc::clone(&self.lane_stable_arenas),
                                        logical_admission: self.logical_admission.clone(),
                                        availability_domains: slot_domains,
                                        key: key.clone(),
                                        slot_id,
                                    },
                                ),
                            ));
                        }
                    }
                    return Ok(LaneBackingPrepareDecision::Deferred(deferred));
                }
            }
        }
    }

    pub(in crate::vnext::resource) fn try_reclaim_expired_lane_slots(
        &self,
    ) -> Result<bool, VNextError> {
        let expired_entries = {
            let mut arenas = self
                .lane_stable_arenas
                .lock()
                .map_err(|_| invalid_resource("lane-stable arena registry is poisoned"))?;
            if arenas.poisoned {
                return Err(invalid_resource(
                    "lane-stable arena registry is fail-closed",
                ));
            }
            arenas.take_expired_lanes()?
        };
        let reclaimed = !expired_entries.is_empty();
        // Releasing backing owners can enter backend/pool destruction paths.
        // Keep that work outside the arena registry's hot mutex.
        drop(expired_entries);
        Ok(reclaimed)
    }

    pub(in crate::vnext::resource) fn try_reclaim_one_idle_lane_slot(
        &self,
    ) -> Result<bool, VNextError> {
        if self.try_reclaim_expired_lane_slots()? {
            return Ok(true);
        }
        let mut candidates = {
            let arenas = self
                .lane_stable_arenas
                .lock()
                .map_err(|_| invalid_resource("lane-stable arena registry is poisoned"))?;
            if arenas.poisoned {
                return Err(invalid_resource(
                    "lane-stable arena registry is fail-closed",
                ));
            }
            arenas
                .entries
                .iter()
                .filter_map(|(key, entry)| entry.lane.upgrade().map(|lane| (key, entry, lane)))
                .flat_map(|(key, entry, lane)| {
                    entry
                        .slots
                        .values()
                        .filter(|slot| !slot.in_use)
                        .map(move |slot| LaneStableArenaEvictionCandidate {
                            key: key.clone(),
                            slot_id: slot.slot_id,
                            last_used: slot.last_used,
                            lane: Arc::clone(&lane),
                        })
                })
                .collect::<Vec<_>>()
        };
        candidates.sort_by_key(|candidate| candidate.last_used);

        for candidate in candidates {
            if !candidate.lane.try_trim_reusable_executables()? {
                continue;
            }
            let victim = {
                let mut arenas = self
                    .lane_stable_arenas
                    .lock()
                    .map_err(|_| invalid_resource("lane-stable arena registry is poisoned"))?;
                if arenas.poisoned {
                    return Err(invalid_resource(
                        "lane-stable arena registry is fail-closed",
                    ));
                }
                let removable = arenas
                    .entries
                    .get(&candidate.key)
                    .and_then(|entry| entry.slots.get(&candidate.slot_id))
                    .is_some_and(|slot| !slot.in_use && !slot.has_external_address_pins());
                if !removable {
                    None
                } else {
                    let (victim, remove_entry) = {
                        let entry = arenas.entries.get_mut(&candidate.key).ok_or_else(|| {
                            invalid_resource("lane-stable arena eviction lost its entry")
                        })?;
                        let victim = entry.slots.remove(&candidate.slot_id).ok_or_else(|| {
                            invalid_resource("lane-stable arena eviction lost its idle slot")
                        })?;
                        (victim, entry.slots.is_empty())
                    };
                    if remove_entry {
                        arenas.entries.remove(&candidate.key);
                    }
                    Some(victim)
                }
            };
            if let Some(victim) = victim {
                drop(victim);
                return Ok(true);
            }
        }
        Ok(false)
    }

    pub(in crate::vnext::resource) fn prepare_initial_sequence_claim(
        &self,
        requests: &[EvaluatedBackingRequest<'_>],
    ) -> Result<BackingPrepareDecision<R>, VNextError> {
        self.prepare_claim_scoped(
            requests,
            DynamicBackingClaimScope::InitialSequenceBundle,
            DynamicBackingClaimResidency::Transient,
        )
    }

    /// Optional capture may grow only its certified State pools. Reuse the
    /// same packing planner and atomic device-budget publication as ordinary
    /// growth, without the foreground path's cross-pool pressure reclamation.
    pub(in crate::vnext::resource) fn maintain_checkpoint_capacity(
        &self,
        requests: &[EvaluatedBackingRequest<'_>],
    ) -> Result<DynamicPoolGrowthBatchReceipt, VNextError> {
        self.maintain_checkpoint_capacity_observed(requests, &mut None)
    }

    pub(in crate::vnext::resource) fn maintain_checkpoint_capacity_with_idle_reclaim(
        &self,
        requests: &[EvaluatedBackingRequest<'_>],
    ) -> Result<DynamicPoolGrowthBatchReceipt, VNextError> {
        let mut capacity_blocked = None;
        match self.maintain_checkpoint_capacity_observed(requests, &mut capacity_blocked) {
            Err(VNextError::DeviceCapacityUnavailable(pressure)) => {
                let blocked = capacity_blocked.as_ref().ok_or_else(|| {
                    invalid_resource("checkpoint growth lost its device capacity observation")
                })?;
                // Use the entire certified byte plan, including target pools
                // whose current free extents already suffice. Reclaiming those
                // would invalidate capture's other claims on the next attempt.
                let targets = requests
                    .iter()
                    .map(|request| request.domain.domain_id)
                    .collect::<std::collections::BTreeSet<_>>()
                    .into_iter()
                    .collect::<Vec<_>>();
                let attempt = self.reclaim_idle_chunks(
                    &pressure,
                    blocked.availability,
                    &targets,
                    &CapacityVector::empty(),
                    &[],
                    ReclaimTargetPolicy::PreserveAll,
                )?;
                tracing::debug!(
                    coordinator_id = ?attempt.boundary.coordinator_id(),
                    ?pressure,
                    protected_target_domains = ?targets,
                    reclaim_sufficient = attempt.boundary.reclaim_sufficient(),
                    candidate_chunks = attempt.boundary.reclaim_candidate_chunks(),
                    candidate_bytes = attempt.boundary.reclaim_candidate_bytes(),
                    selected_bytes = attempt.boundary.selected_bytes(),
                    reclaimed_bytes = attempt.rebalance.as_ref().map_or(0, |r| r.reclaimed_bytes()),
                    reclaimed_chunks = attempt.rebalance.as_ref().map_or(0, |r| r.reclaimed_chunks()),
                    "optional checkpoint idle reclaim evaluated"
                );
                if tracing::enabled!(tracing::Level::DEBUG) {
                    for pool in attempt.boundary.pools() {
                        let reclaimed_bytes = attempt
                            .rebalance
                            .as_ref()
                            .and_then(|rebalance| {
                                rebalance
                                    .pools()
                                    .iter()
                                    .find(|reclaimed| reclaimed.pool_id() == pool.pool_id())
                            })
                            .map_or(0, |reclaimed| reclaimed.reclaimed_bytes());
                        tracing::debug!(
                            coordinator_id = ?attempt.boundary.coordinator_id(),
                            domain = ?pool.domain_id(),
                            pool_id = pool.pool_id().as_str(),
                            excluded = pool.excluded_from_reclaim(),
                            resident_bytes = pool.resident_bytes(),
                            free_bytes = pool.free_bytes(),
                            logical_used_bytes = pool.logical_used_bytes(),
                            physical_live_bytes = pool.live_occupancy().total().physical_bytes(),
                            pending_growth_bytes = pool.pending_growth_bytes(),
                            resident_floor_bytes = pool.resident_floor_bytes(),
                            bytes_above_floor = pool.reclaimable_bytes(),
                            reclaimed_bytes,
                            "optional checkpoint idle reclaim pool boundary"
                        );
                    }
                }
                let Some(rebalance) = attempt.rebalance else {
                    return Err(VNextError::DeviceCapacityUnavailable(pressure));
                };
                // Recompute every target's packing from current state and
                // reserve anew. A competitor or allocation failure can still
                // reject this single retry; reclamation grants no capacity.
                let mut receipt = self.maintain_checkpoint_capacity(requests)?;
                tracing::debug!(
                    coordinator_id = ?receipt.coordinator_id(),
                    reclaimed_bytes = rebalance.reclaimed_bytes(),
                    growths = ?receipt.growths(),
                    "optional checkpoint idle reclaim followed by fresh growth"
                );
                receipt.rebalance = Some(rebalance);
                receipt.maintenance_boundary = Some(attempt.boundary);
                Ok(receipt)
            }
            result => result,
        }
    }

    fn maintain_checkpoint_capacity_observed(
        &self,
        requests: &[EvaluatedBackingRequest<'_>],
        capacity_blocked: &mut Option<DynamicDeviceCapacityBlocked>,
    ) -> Result<DynamicPoolGrowthBatchReceipt, VNextError> {
        self.validate_checkpoint_requests(requests)?;
        let mut by_pool = BTreeMap::<DynamicBackingPoolId, Vec<u64>>::new();
        for request in requests {
            by_pool
                .entry(request.domain.pool_id().clone())
                .or_default()
                .push(request.capacity_size_bytes);
        }
        let mut intents = Vec::with_capacity(by_pool.len());
        for (pool_id, mut claims) in by_pool {
            claims.sort_unstable_by(|left, right| right.cmp(left));
            let pool = self
                .pools
                .get(&pool_id)
                .ok_or_else(|| invalid_resource("checkpoint pool is absent"))?;
            let state = pool
                .state
                .lock()
                .map_err(|_| invalid_resource("checkpoint pool is poisoned"))?;
            if state.poisoned {
                return Err(invalid_resource("checkpoint pool is fail-closed"));
            }
            let (growth, contiguous_claims) =
                match pool.domain.pool.compatibility().profile().view() {
                    DynamicStorageView::Contiguous => (
                        contiguous_packing_growth_bytes(&state.allocator, &pool_id, &claims)?,
                        Some(claims),
                    ),
                    DynamicStorageView::PagedRegions { .. } => (
                        claims
                            .iter()
                            .try_fold(0_u64, |total, bytes| total.checked_add(*bytes))
                            .ok_or_else(|| {
                                invalid_resource("checkpoint pool demand overflows u64")
                            })?
                            .saturating_sub(state.allocator.free_bytes),
                        None,
                    ),
                };
            if growth != 0 {
                intents.push(DynamicPoolGrowthIntent::RevalidatedDeferral(
                    DynamicBackingBlocker {
                        pool_id,
                        domain_id: pool.domain.domain_id,
                        reason: DynamicBackingDeferralReason::GrowthRequired,
                        requested_bytes: growth,
                        free_bytes: state.allocator.free_bytes,
                        largest_contiguous_bytes: state.allocator.largest_contiguous_bytes(),
                        free_extent_layout_fingerprint: free_extent_layout_fingerprint(
                            &state.allocator,
                        ),
                        contiguous_claim_bytes_descending: contiguous_claims,
                    },
                ));
            }
        }
        // RevalidatedDeferral recomputes demand after acquiring the existing
        // canonical maintenance locks. No stale free-space snapshot grants bytes.
        self.maintain_pools_observed(intents, capacity_blocked)
    }

    pub(in crate::vnext::resource) fn prepare_checkpoint_claim(
        &self,
        requests: &[EvaluatedBackingRequest<'_>],
    ) -> Result<BackingPrepareDecision<R>, VNextError> {
        self.validate_checkpoint_requests(requests)?;
        self.prepare_claim_scoped(
            requests,
            DynamicBackingClaimScope::Checkpoint,
            DynamicBackingClaimResidency::Transient,
        )
    }

    fn validate_checkpoint_requests(
        &self,
        requests: &[EvaluatedBackingRequest<'_>],
    ) -> Result<(), VNextError> {
        if requests.is_empty()
            || requests.iter().any(|request| {
                !self
                    .domains
                    .iter()
                    .any(|domain| std::ptr::eq(domain, request.domain))
                    || request.reusable_execution_bucket_id.is_some()
                    || request.projections.len() != 1
                    || request.projections.iter().any(|projection| {
                        projection.descriptor.lifetime() != AllocationLifetime::Sequence
                            || projection.descriptor.usage() != super::BufferUsage::State
                            || *projection.descriptor.kind() != super::AllocationKind::Value
                    })
            })
        {
            return Err(invalid_resource(
                "checkpoint backing requires this plan's non-empty Sequence state projections",
            ));
        }
        Ok(())
    }

    fn prepare_claim_scoped(
        &self,
        requests: &[EvaluatedBackingRequest<'_>],
        scope: DynamicBackingClaimScope,
        residency: DynamicBackingClaimResidency,
    ) -> Result<BackingPrepareDecision<R>, VNextError> {
        if requests.is_empty() {
            return Ok(BackingPrepareDecision::Prepared(
                PreparedBackingClaim::empty(),
            ));
        }
        let reusable_capacity_shape = self.reusable_capacity_shape_for_requests(requests)?;
        let mut grouped =
            BTreeMap::<DynamicBackingPoolId, Vec<&EvaluatedBackingRequest<'_>>>::new();
        for request in requests {
            grouped
                .entry(request.domain.pool_id().clone())
                .or_default()
                .push(request);
        }
        let mut groups = Vec::with_capacity(grouped.len());
        for (pool_id, mut requests) in grouped {
            requests.sort_by(|left, right| left.claim_identity.cmp(&right.claim_identity));
            if requests
                .windows(2)
                .any(|pair| pair[0].claim_identity == pair[1].claim_identity)
            {
                return Err(invalid_resource(
                    "dynamic backing reservation contains a duplicate physical claim",
                ));
            }
            let pool = self.pools.get(&pool_id).cloned().ok_or_else(|| {
                invalid_resource("dynamic backing reservation references an unknown pool")
            })?;
            groups.push((pool, requests));
        }
        let protected_packing_envelopes = groups
            .iter()
            .map(|(pool, requests)| {
                DynamicBackingPackingEnvelope::new(
                    pool.domain.pool_id().clone(),
                    pool.domain.domain_id,
                    requests
                        .iter()
                        .map(|request| request.capacity_size_bytes)
                        .collect(),
                )
            })
            .collect::<Result<Vec<_>, VNextError>>()?;
        let protected_immediate = CapacityVector::new(
            protected_packing_envelopes
                .iter()
                .map(|envelope| {
                    CapacityEntry::new(
                        envelope.domain_id(),
                        CapacityUnits::new(envelope.total_bytes()?),
                    )
                })
                .collect::<Result<Vec<_>, VNextError>>()?,
        )?;
        'prepare: loop {
            let mut states = groups
                .iter()
                .map(|(pool, _)| {
                    pool.state
                        .lock()
                        .map_err(|_| invalid_resource("dynamic backing pool is poisoned"))
                })
                .collect::<Result<Vec<_>, _>>()?;
            for (group_index, (pool, pool_requests)) in groups.iter().enumerate() {
                if states[group_index].poisoned {
                    return Err(invalid_resource("dynamic backing pool is fail-closed"));
                }
                let quantum = pool.allocation_quantum();
                for request in pool_requests {
                    let projection_ids = request
                        .projections
                        .iter()
                        .map(|projection| projection.descriptor.base_resource_id().clone())
                        .collect::<Vec<_>>();
                    let single_projection = request.projections.len() == 1
                        && request.projections[0].physical_offset_bytes == 0
                        && request.projections[0].capacity_size_bytes
                            == request.capacity_size_bytes;
                    let shared_step_slot = request.projections.len() > 1
                        && request
                            .projections
                            .iter()
                            .all(|projection| projection.physical_offset_bytes == 0)
                        && request
                            .projections
                            .iter()
                            .map(|projection| projection.capacity_size_bytes)
                            .max()
                            == Some(request.capacity_size_bytes)
                        && pool.domain.pool.step_resource_slots().iter().any(|slot| {
                            slot.kind() == StepResourceSlotKind::OrderedSingleFenceStepWave
                                && slot.resource_ids() == request.claim_identity.resource_ids()
                        });
                    let invocation_wave =
                        self.validate_invocation_wave_projection(pool, request)?;
                    if request.domain.pool_id() != pool.domain.pool_id()
                        || request.claim_identity.pool_id() != pool.domain.pool_id()
                        || request.claim_identity.resource_ids() != projection_ids
                        || request.projections.is_empty()
                        || request.projections.windows(2).any(|pair| {
                            pair[0].descriptor.base_resource_id()
                                >= pair[1].descriptor.base_resource_id()
                        })
                        || request.projections.iter().any(|projection| {
                            let capacity_matches_plan = match reusable_capacity_shape {
                                Some(shape) => {
                                    matches!(
                                        projection.descriptor.lifetime(),
                                        AllocationLifetime::Step | AllocationLifetime::Invocation
                                    ) && projection
                                        .descriptor
                                        .evaluate_request_bytes_for_shape(shape)
                                        .is_ok_and(|bytes| bytes == projection.capacity_size_bytes)
                                }
                                None => {
                                    scope == DynamicBackingClaimScope::Checkpoint
                                        || projection.logical_size_bytes
                                            == projection.capacity_size_bytes
                                }
                            };
                            projection.descriptor.pool_id() != pool.domain.pool_id()
                                || !scope.accepts(projection.descriptor.lifetime())
                                || !capacity_matches_plan
                                || projection.logical_size_bytes == 0
                                || projection.logical_size_bytes > projection.capacity_size_bytes
                                || projection.capacity_size_bytes == 0
                                || projection.capacity_size_bytes % quantum != 0
                                || projection.physical_offset_bytes % quantum != 0
                                || projection
                                    .physical_offset_bytes
                                    .checked_add(projection.capacity_size_bytes)
                                    .is_none_or(|end| end > request.capacity_size_bytes)
                                || !request
                                    .domain
                                    .descriptors
                                    .iter()
                                    .any(|descriptor| descriptor == projection.descriptor)
                        })
                        || request.capacity_size_bytes == 0
                        || request.capacity_size_bytes % quantum != 0
                        || !(single_projection || shared_step_slot || invocation_wave)
                    {
                        return Err(invalid_resource(
                        "dynamic backing request violates its physical claim, projection, pool, or allocation quantum",
                    ));
                    }
                }
            }
            let blockers = groups
                .iter()
                .enumerate()
                .map(|(group_index, (pool, pool_requests))| {
                    let requested_group_bytes =
                        pool_requests.iter().try_fold(0_u64, |total, request| {
                            total
                                .checked_add(request.capacity_size_bytes)
                                .ok_or_else(|| {
                                    invalid_resource("dynamic backing batch bytes overflow u64")
                                })
                        })?;
                    let state = &states[group_index];
                    if state.allocator.free_bytes >= requested_group_bytes {
                        return Ok(None);
                    }
                    let (reason, requested_bytes, contiguous_claim_bytes_descending) =
                        match pool.domain.pool.compatibility().profile().view() {
                            DynamicStorageView::Contiguous => {
                                let mut claim_bytes = pool_requests
                                    .iter()
                                    .map(|request| request.capacity_size_bytes)
                                    .collect::<Vec<_>>();
                                claim_bytes.sort_unstable_by(|left, right| right.cmp(left));
                                let growth = contiguous_packing_growth_bytes(
                                    &state.allocator,
                                    pool.domain.pool_id(),
                                    &claim_bytes,
                                )?;
                                if growth == 0 {
                                    return Err(invalid_resource(
                                    "insufficient contiguous capacity produced zero packing growth",
                                ));
                                }
                                (
                                    DynamicBackingDeferralReason::GrowthRequired,
                                    growth,
                                    Some(claim_bytes),
                                )
                            }
                            DynamicStorageView::PagedRegions { .. } => (
                                DynamicBackingDeferralReason::GrowthRequired,
                                requested_group_bytes - state.allocator.free_bytes,
                                None,
                            ),
                        };
                    Ok(Some(DynamicBackingBlocker {
                        pool_id: pool.domain.pool_id().clone(),
                        domain_id: pool.domain.domain_id,
                        reason,
                        requested_bytes,
                        free_bytes: state.allocator.free_bytes,
                        largest_contiguous_bytes: state.allocator.largest_contiguous_bytes(),
                        free_extent_layout_fingerprint: free_extent_layout_fingerprint(
                            &state.allocator,
                        ),
                        contiguous_claim_bytes_descending,
                    }))
                })
                .collect::<Result<Vec<_>, VNextError>>()?
                .into_iter()
                .flatten()
                .collect::<Vec<_>>();
            if !blockers.is_empty() {
                drop(states);
                if let Some(deferred) = self.confirm_backing_deferral(
                    blockers,
                    scope,
                    protected_immediate.clone(),
                    protected_packing_envelopes.clone(),
                )? {
                    return Ok(BackingPrepareDecision::Deferred(deferred));
                }
                continue 'prepare;
            }
            let segment_generations = groups
                .iter()
                .map(|(pool, requests)| {
                    (0..requests.len())
                        .map(|_| {
                            pool.next_extent_generation
                                .fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| {
                                    current.checked_add(1)
                                })
                                .map_err(|_| {
                                    invalid_resource("dynamic extent generation space is exhausted")
                                })
                        })
                        .collect::<Result<Vec<_>, _>>()
                })
                .collect::<Result<Vec<_>, _>>()?;
            let mut selections = groups
                .iter()
                .map(|_| {
                    Vec::<(
                        &EvaluatedBackingRequest<'_>,
                        u64,
                        Vec<BackingSegment>,
                        DynamicBackingClaimOccupancy,
                    )>::new()
                })
                .collect::<Vec<_>>();
            let mut journals = groups
                .iter()
                .map(|_| Vec::<Vec<BackingSegment>>::new())
                .collect::<Vec<_>>();
            for group_index in 0..groups.len() {
                let (pool, pool_requests) = &groups[group_index];
                let profile = pool.domain.pool.compatibility().profile();
                let mut allocation_requests = pool_requests.clone();
                allocation_requests.sort_by(|left, right| {
                    right
                        .capacity_size_bytes
                        .cmp(&left.capacity_size_bytes)
                        .then_with(|| left.claim_identity.cmp(&right.claim_identity))
                });
                for request in allocation_requests {
                    let reserved = match match profile.view() {
                        DynamicStorageView::Contiguous => states[group_index]
                            .allocator
                            .allocate_contiguous(pool.domain.pool_id(), request.capacity_size_bytes)
                            .map(|segment| segment.map(|segment| vec![segment])),
                        DynamicStorageView::PagedRegions { block_bytes } => {
                            states[group_index].allocator.allocate_paged(
                                pool.domain.pool_id(),
                                request.capacity_size_bytes,
                                block_bytes,
                            )
                        }
                    } {
                        Ok(reserved) => reserved,
                        Err(error) => {
                            states[group_index].poisoned = true;
                            rollback_free_extent_journal(&mut states, &journals)?;
                            return Err(error);
                        }
                    };
                    let Some(segments) = reserved else {
                        rollback_free_extent_journal(&mut states, &journals)?;
                        if !matches!(profile.view(), DynamicStorageView::Contiguous) {
                            states[group_index].poisoned = true;
                            return Err(invalid_resource(
                                "paged backing allocation failed after its aggregate fit check",
                            ));
                        }
                        let free_bytes = states[group_index].allocator.free_bytes;
                        let reason = DynamicBackingDeferralReason::FragmentedContiguous;
                        let mut claim_bytes_descending = pool_requests
                            .iter()
                            .map(|request| request.capacity_size_bytes)
                            .collect::<Vec<_>>();
                        claim_bytes_descending.sort_unstable_by(|left, right| right.cmp(left));
                        let requested_bytes = contiguous_packing_growth_bytes(
                            &states[group_index].allocator,
                            pool.domain.pool_id(),
                            &claim_bytes_descending,
                        )?;
                        if requested_bytes == 0 {
                            return Err(invalid_resource(
                                "contiguous packing failed without a progress-producing growth",
                            ));
                        }
                        let blocker = DynamicBackingBlocker {
                            pool_id: pool.domain.pool_id().clone(),
                            domain_id: pool.domain.domain_id,
                            reason,
                            requested_bytes,
                            free_bytes,
                            largest_contiguous_bytes: states[group_index]
                                .allocator
                                .largest_contiguous_bytes(),
                            free_extent_layout_fingerprint: free_extent_layout_fingerprint(
                                &states[group_index].allocator,
                            ),
                            contiguous_claim_bytes_descending: Some(claim_bytes_descending),
                        };
                        drop(states);
                        if let Some(deferred) = self.confirm_backing_deferral(
                            vec![blocker],
                            scope,
                            protected_immediate.clone(),
                            protected_packing_envelopes.clone(),
                        )? {
                            return Ok(BackingPrepareDecision::Deferred(deferred));
                        }
                        continue 'prepare;
                    };
                    journals[group_index].push(segments.clone());
                    let extent_bytes = match segments.iter().try_fold(0_u64, |total, segment| {
                        total.checked_add(segment.length_bytes()).ok_or_else(|| {
                            invalid_resource("dynamic backing extent bytes overflow u64")
                        })
                    }) {
                        Ok(bytes) => bytes,
                        Err(error) => {
                            rollback_free_extent_journal(&mut states, &journals)?;
                            return Err(error);
                        }
                    };
                    if extent_bytes != request.capacity_size_bytes {
                        rollback_free_extent_journal(&mut states, &journals)?;
                        return Err(invalid_resource(
                            "dynamic backing extents differ from their physical capacity claim",
                        ));
                    }
                    let generation =
                        segment_generations[group_index][selections[group_index].len()];
                    let segment_count = match u64::try_from(segments.len()) {
                        Ok(count) => count,
                        Err(_) => {
                            rollback_free_extent_journal(&mut states, &journals)?;
                            return Err(invalid_resource(
                                "dynamic backing segment count exceeds portable range",
                            ));
                        }
                    };
                    selections[group_index].push((
                        request,
                        generation,
                        segments,
                        DynamicBackingClaimOccupancy {
                            scope,
                            residency,
                            physical_bytes: request.capacity_size_bytes,
                            segment_count,
                        },
                    ));
                }
            }

            for group_selections in &selections {
                for (request, _, segments, _) in group_selections {
                    for projection in &request.projections {
                        if let Err(error) = backing_segment_range(
                            segments,
                            projection.physical_offset_bytes,
                            projection.capacity_size_bytes,
                        ) {
                            rollback_free_extent_journal(&mut states, &journals)?;
                            return Err(error);
                        }
                    }
                }
            }

            let accounting_updates = match (0..groups.len())
                .map(|group_index| {
                    let mut increments = BTreeMap::<u32, u64>::new();
                    for (_, _, segments, _) in &selections[group_index] {
                        for segment in segments {
                            let count = increments.entry(segment.chunk_ordinal()).or_default();
                            *count = count.checked_add(1).ok_or_else(|| {
                                invalid_resource(
                                    "dynamic chunk live extent increment overflows u64",
                                )
                            })?;
                        }
                    }
                    for (&ordinal, &increment) in &increments {
                        states[group_index]
                            .chunks
                            .get(&ordinal)
                            .ok_or_else(|| invalid_resource("reserved dynamic chunk disappeared"))?
                            .live_segments
                            .checked_add(increment)
                            .ok_or_else(|| {
                                invalid_resource("dynamic chunk live extent count overflowed")
                            })?;
                    }
                    let occupancy = selections[group_index].iter().try_fold(
                        states[group_index].live_occupancy,
                        |occupancy, (_, _, _, claim)| occupancy.checked_with_claim(*claim),
                    )?;
                    Ok((increments, occupancy))
                })
                .collect::<Result<Vec<_>, VNextError>>()
            {
                Ok(updates) => updates,
                Err(error) => {
                    rollback_free_extent_journal(&mut states, &journals)?;
                    return Err(error);
                }
            };
            for (group_index, (increments, occupancy)) in accounting_updates.into_iter().enumerate()
            {
                for (ordinal, increment) in increments {
                    states[group_index]
                        .chunks
                        .get_mut(&ordinal)
                        .expect("validated reserved dynamic chunk remains installed")
                        .live_segments += increment;
                }
                states[group_index].live_occupancy = occupancy;
            }
            drop(states);
            let mut extents = Vec::new();
            for ((pool, _), selections) in groups.into_iter().zip(selections) {
                for (request, segment_generation, segments, occupancy) in selections {
                    let projections = request
                        .projections
                        .iter()
                        .map(|projection| {
                            let mut allocation = LogicalBackingSliceAllocationEvidence {
                                domain_id: pool.domain.domain_id,
                                pool_id: pool.domain.pool_id().clone(),
                                resource_id: projection.descriptor.base_resource_id().clone(),
                                pool_instance_id: pool.instance_id,
                                physical_claim_identity: request.claim_identity.clone(),
                                reusable_execution_bucket_id: request
                                    .reusable_execution_bucket_id
                                    .clone(),
                                segment_generation,
                                segments: backing_segment_range(
                                    &segments,
                                    projection.physical_offset_bytes,
                                    projection.capacity_size_bytes,
                                )?,
                                physical_offset_bytes: projection.physical_offset_bytes,
                                capacity_size_bytes: projection.capacity_size_bytes,
                                physical_size_bytes: request.capacity_size_bytes,
                                alignment_bytes: projection.descriptor.alignment_bytes(),
                                usage: projection.descriptor.usage(),
                                element_type: projection.descriptor.element_type(),
                                storage_profile: pool.domain.pool.compatibility().profile(),
                                initialization: projection.descriptor.initialization(),
                                fingerprint: String::new(),
                            };
                            let bytes = serde_json::to_vec(&allocation).map_err(|error| {
                                invalid_resource(format!(
                                    "logical backing allocation evidence encode failed: {error}"
                                ))
                            })?;
                            allocation.fingerprint = format!("sha256/{:x}", Sha256::digest(bytes));
                            Ok(LogicalBackingSliceEvidence {
                                allocation: Arc::new(allocation),
                                logical_size_bytes: projection.logical_size_bytes,
                            })
                        })
                        .collect::<Result<Vec<_>, VNextError>>()?;
                    extents.push(PreparedBackingExtent {
                        pool: Arc::clone(&pool),
                        claim_identity: request.claim_identity.clone(),
                        segment_generation,
                        occupancy,
                        segments,
                        capacity_size_bytes: request.capacity_size_bytes,
                        projections,
                    });
                }
            }
            return Ok(BackingPrepareDecision::Prepared(PreparedBackingClaim {
                extents,
                committed: false,
            }));
        }
    }

    /// Publishes a physical deferral with the event-subscription ordering
    /// required to avoid a lost release:
    ///
    /// 1. observe the exact coordinator generations after the failed check;
    /// 2. recheck the physical allocator observations;
    /// 3. publish only if those observations are still current.
    ///
    /// A release before step 1 is visible to step 2. A release after step 1
    /// advances the returned predicate (or is visible to step 2 before its
    /// coordinator notification), so the scheduler cannot sleep forever on a
    /// stale blocker.
    fn confirm_backing_deferral(
        &self,
        blockers: Vec<DynamicBackingBlocker>,
        scope: DynamicBackingClaimScope,
        protected_immediate: CapacityVector,
        protected_packing_envelopes: Vec<DynamicBackingPackingEnvelope>,
    ) -> Result<Option<DynamicBackingDeferred>, VNextError> {
        let wait_snapshot = self
            .logical_admission
            .wait_snapshot_for_domains(blockers.iter().map(DynamicBackingBlocker::domain_id))?;
        for blocker in &blockers {
            let pool = self.pools.get(blocker.pool_id()).ok_or_else(|| {
                invalid_resource("dynamic backing blocker references an unknown pool")
            })?;
            let state = pool
                .state
                .lock()
                .map_err(|_| invalid_resource("dynamic backing pool is poisoned"))?;
            if state.poisoned {
                return Err(invalid_resource("dynamic backing pool is fail-closed"));
            }
            if state.allocator.free_bytes != blocker.free_bytes()
                || state.allocator.largest_contiguous_bytes() != blocker.largest_contiguous_bytes()
                || free_extent_layout_fingerprint(&state.allocator)
                    != blocker.free_extent_layout_fingerprint()
            {
                return Ok(None);
            }
        }
        Ok(Some(DynamicBackingDeferred {
            blockers,
            epochs: wait_snapshot.epochs(),
            wait_condition: wait_snapshot.wait_condition().clone(),
            scope,
            protected_immediate,
            protected_packing_envelopes,
        }))
    }

    fn validate_invocation_wave_projection(
        &self,
        pool: &DynamicBackingPool<R>,
        request: &EvaluatedBackingRequest<'_>,
    ) -> Result<bool, VNextError> {
        let mode = pool.domain.pool.invocation_liveness_mode();
        if mode == InvocationLivenessMode::NoInvocationResources
            || request.projections.len() < 2
            || request.projections.iter().any(|projection| {
                projection.descriptor.lifetime() != super::AllocationLifetime::Invocation
            })
        {
            return Ok(false);
        }
        let mut expected_resources = pool
            .domain
            .pool
            .invocation_liveness()
            .iter()
            .flat_map(|row| row.resource_ids().iter().cloned())
            .collect::<Vec<_>>();
        expected_resources.sort();
        if expected_resources.windows(2).any(|pair| pair[0] == pair[1])
            || expected_resources != request.claim_identity.resource_ids()
        {
            return Ok(false);
        }
        let projections = request
            .projections
            .iter()
            .map(|projection| (projection.descriptor.base_resource_id().clone(), projection))
            .collect::<BTreeMap<_, _>>();
        if projections.len() != request.projections.len() {
            return Ok(false);
        }
        let rows_by_node = pool
            .domain
            .pool
            .invocation_liveness()
            .iter()
            .map(|row| (row.node_id(), row))
            .collect::<BTreeMap<_, _>>();
        let rows = self
            .nodes
            .iter()
            .filter_map(|node| rows_by_node.get(node.id()).copied())
            .collect::<Vec<_>>();
        if rows.len() != rows_by_node.len() {
            return Ok(false);
        }

        let mut concurrent_cursor = 0_u64;
        let mut peak = 0_u64;
        for row in rows {
            let row_base = match mode {
                InvocationLivenessMode::TotalOrderReuse => 0,
                InvocationLivenessMode::ConservativeConcurrent => concurrent_cursor,
                InvocationLivenessMode::NoInvocationResources => unreachable!(),
            };
            let mut row_cursor = 0_u64;
            for resource_id in row.resource_ids() {
                let Some(projection) = projections.get(resource_id) else {
                    return Ok(false);
                };
                let expected_offset = row_base.checked_add(row_cursor).ok_or_else(|| {
                    invalid_resource("invocation wave projection offset overflows u64")
                })?;
                if projection.physical_offset_bytes != expected_offset {
                    return Ok(false);
                }
                row_cursor = row_cursor
                    .checked_add(projection.capacity_size_bytes)
                    .ok_or_else(|| invalid_resource("invocation wave row size overflows u64"))?;
            }
            peak = peak.max(row_cursor);
            if mode == InvocationLivenessMode::ConservativeConcurrent {
                concurrent_cursor = concurrent_cursor.checked_add(row_cursor).ok_or_else(|| {
                    invalid_resource("concurrent invocation wave size overflows u64")
                })?;
            }
        }
        Ok(request.capacity_size_bytes
            == match mode {
                InvocationLivenessMode::TotalOrderReuse => peak,
                InvocationLivenessMode::ConservativeConcurrent => concurrent_cursor,
                InvocationLivenessMode::NoInvocationResources => 0,
            })
    }

    pub(in crate::vnext::resource) fn view<'lease>(
        &'lease self,
        authority: &'lease LogicalBackingSliceAuthority,
    ) -> Result<LogicalBackingBufferView<'lease, R::Buffer>, VNextError> {
        self.view_many(std::slice::from_ref(authority))
    }

    pub(in crate::vnext::resource) fn view_many<'lease>(
        &'lease self,
        authorities: &'lease [LogicalBackingSliceAuthority],
    ) -> Result<LogicalBackingBufferView<'lease, R::Buffer>, VNextError> {
        let first = authorities
            .first()
            .ok_or_else(|| invalid_resource("logical backing view requires an authority"))?;
        let pool = self
            .pools
            .get(&first.evidence.pool_id)
            .ok_or_else(|| invalid_resource("logical backing authority has no dynamic pool"))?;
        let mut logical_size_bytes = 0_u64;
        let mut capacity_size_bytes = 0_u64;
        let mut segment_count = 0_usize;
        for (index, authority) in authorities.iter().enumerate() {
            if authority.evidence.pool_id != first.evidence.pool_id
                || authority.evidence.resource_id != first.evidence.resource_id
                || authority.evidence.storage_profile != first.evidence.storage_profile
                || authority.evidence.alignment_bytes != first.evidence.alignment_bytes
                || authority.evidence.usage != first.evidence.usage
                || authority.evidence.element_type != first.evidence.element_type
                || authority.evidence.initialization != first.evidence.initialization
            {
                return Err(invalid_resource(
                    "logical backing authorities have incompatible resource metadata",
                ));
            }
            Self::validate_authority(pool, authority)?;
            if index + 1 < authorities.len()
                && authority.evidence.logical_size_bytes != authority.evidence.capacity_size_bytes
            {
                return Err(invalid_resource(
                    "multi-extent logical backing cannot contain interior capacity slack",
                ));
            }
            logical_size_bytes = logical_size_bytes
                .checked_add(authority.evidence.logical_size_bytes)
                .ok_or_else(|| invalid_resource("logical backing view size overflows u64"))?;
            capacity_size_bytes = capacity_size_bytes
                .checked_add(authority.evidence.capacity_size_bytes)
                .ok_or_else(|| invalid_resource("logical backing capacity overflows u64"))?;
            segment_count = segment_count
                .checked_add(authority.evidence.segments.len())
                .ok_or_else(|| invalid_resource("logical backing segment count overflows usize"))?;
        }
        let state = pool
            .state
            .lock()
            .map_err(|_| invalid_resource("dynamic backing pool is poisoned"))?;
        if state.poisoned {
            return Err(invalid_resource("dynamic backing pool is fail-closed"));
        }
        let mut bindings = Vec::with_capacity(segment_count);
        for authority in authorities {
            for segment in &authority.evidence.segments {
                let chunk = state.chunks.get(&segment.chunk_ordinal()).ok_or_else(|| {
                    invalid_resource("logical backing references a missing chunk")
                })?;
                if segment.pool_id() != &authority.evidence.pool_id
                    || chunk.backing.identity != *segment.chunk()
                    || segment
                        .offset_bytes()
                        .checked_add(segment.length_bytes())
                        .is_none_or(|end| end > chunk.backing.descriptor.size_bytes)
                {
                    return Err(invalid_resource(
                        "logical backing references a stale or out-of-bounds chunk region",
                    ));
                }
                let retention = match authority.reusable_lane {
                    Some(lane_id) => DeviceBufferRetention::lane_pair(
                        lane_id,
                        Arc::clone(&authority.segment_lease),
                        Arc::clone(&chunk.backing),
                    ),
                    None => DeviceBufferRetention::pair(
                        Arc::clone(&authority.segment_lease),
                        Arc::clone(&chunk.backing),
                    ),
                };
                bindings.push(LogicalBackingSegmentBinding {
                    segment: segment.clone(),
                    chunk: Arc::clone(&chunk.backing),
                    retention,
                });
            }
        }
        drop(state);
        Ok(LogicalBackingBufferView {
            bindings,
            authorities,
            logical_size_bytes,
            capacity_size_bytes,
            alignment_bytes: first.evidence.alignment_bytes,
            usage: first.evidence.usage,
            element_type: first.evidence.element_type,
            storage_profile: first.evidence.storage_profile,
        })
    }

    fn validate_authority(
        pool: &DynamicBackingPool<R>,
        authority: &LogicalBackingSliceAuthority,
    ) -> Result<(), VNextError> {
        if pool.instance_id != authority.evidence.pool_instance_id
            || authority.segment_lease.owner_instance_id != pool.instance_id
            || authority.segment_lease.owner.instance_id() != pool.instance_id
            || authority.segment_lease.claim_identity != authority.evidence.physical_claim_identity
            || authority.segment_lease.segment_generation != authority.evidence.segment_generation
            || authority.segment_lease.size_bytes != authority.evidence.physical_size_bytes
            || authority.evidence.domain_id != pool.domain.domain_id
            || authority.evidence.physical_claim_identity.pool_id() != pool.domain.pool_id()
            || authority
                .evidence
                .physical_claim_identity
                .resource_ids()
                .binary_search(&authority.evidence.resource_id)
                .is_err()
            || authority.evidence.logical_size_bytes == 0
            || authority.evidence.logical_size_bytes > authority.evidence.capacity_size_bytes
            || authority
                .evidence
                .physical_offset_bytes
                .checked_add(authority.evidence.capacity_size_bytes)
                .is_none_or(|end| end > authority.evidence.physical_size_bytes)
            || authority.evidence.storage_profile != pool.domain.pool.compatibility().profile()
        {
            return Err(invalid_resource(
                "logical backing authority belongs to another dynamic pool instance",
            ));
        }
        let expected_projection = backing_segment_range(
            &authority.segment_lease.segments,
            authority.evidence.physical_offset_bytes,
            authority.evidence.capacity_size_bytes,
        )?;
        if expected_projection != authority.evidence.segments {
            return Err(invalid_resource(
                "logical backing projection differs from its shared physical extent",
            ));
        }
        Ok(())
    }
}