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