1use super::allocation::{AllocationError, AllocationLedger};
4use super::extent::extent_page_encoded_length;
5use super::{
6 BlobBuildError, BlobBuildOptions, BlobReadError, CanonicalDecodeError, DecodeLimits, Extent,
7 ExtentCloneError, ExtentCloneRequest, ExtentKind, ExtentMutation, ExtentMutationError,
8 ExtentMutationOptions, ExtentPage, ExtentRangeRequest, ExtentReadError, FilePayload,
9 InlineFileData, InlineFileDataError, MAXIMUM_INLINE_FILE_BYTES, apply_extent_mutations_async,
10 build_blob_async, clone_extent_range_async, encode_extent_page, plan_extent_range_async,
11 read_blob_range_async,
12};
13use crate::async_storage::AsyncObjectStore;
14use crate::cancellation::CancellationToken;
15use crate::model::VolumeConfig;
16use crate::performance::{OperationFailure, WorkBudget, WorkCounters, WorkError};
17use crate::storage::{
18 ByteRange, OBJECT_DIGEST_ENVELOPE_BYTES, ObjectId, ObjectKind, ObjectStoreError, object_digest,
19};
20use bytes::Bytes;
21use std::io::Cursor;
22use std::mem::size_of;
23use thiserror::Error;
24
25#[derive(Clone, Copy, Debug, Eq, PartialEq)]
26pub(crate) enum RegularMutation {
27 Write {
28 offset: u64,
29 length: u64,
30 content: ObjectId,
31 content_offset: u64,
32 },
33 Resize {
34 logical_bytes: u64,
35 },
36 ZeroRange {
37 offset: u64,
38 length: u64,
39 allocated: bool,
40 extend: bool,
41 },
42 Preallocate {
43 offset: u64,
44 length: u64,
45 keep_size: bool,
46 },
47}
48
49#[derive(Clone, Copy, Debug, Eq, PartialEq)]
50pub(crate) struct RegularMutationReceipt {
51 pub(crate) payload: FilePayload,
52 pub(crate) work: WorkCounters,
53}
54
55#[derive(Clone, Copy, Debug, Eq, PartialEq)]
56pub(crate) struct RegularCloneReceipt {
57 pub(crate) destination: FilePayload,
58 pub(crate) work: WorkCounters,
59}
60
61#[derive(Clone, Copy, Debug, Eq, PartialEq)]
62enum RegularStorage {
63 Inline(InlineFileData),
64 Sparse {
65 logical_bytes: u64,
66 extents: ObjectId,
67 },
68}
69
70impl RegularStorage {
71 fn validate(payload: FilePayload) -> Result<Self, RegularMutationFailure> {
72 match payload {
73 FilePayload::InlineRegular(data) => Ok(Self::Inline(data)),
74 FilePayload::Regular {
75 logical_bytes,
76 extents,
77 } => Ok(Self::Sparse {
78 logical_bytes,
79 extents,
80 }),
81 FilePayload::Directory { .. }
82 | FilePayload::SymbolicLink { .. }
83 | FilePayload::Empty
84 | FilePayload::Device { .. }
85 | FilePayload::ReparsePoint { .. } => Err(OperationFailure::before_work(
86 RegularMutationError::NotRegular,
87 )),
88 }
89 }
90
91 fn logical_bytes(self) -> u64 {
92 match self {
93 Self::Inline(data) => u64::try_from(data.as_bytes().len()).unwrap_or(u64::MAX),
94 Self::Sparse { logical_bytes, .. } => logical_bytes,
95 }
96 }
97}
98
99#[derive(Clone, Copy, Debug, Eq, PartialEq)]
100struct PromotedRegularReceipt {
101 logical_bytes: u64,
102 extents: ObjectId,
103 work: WorkCounters,
104}
105
106pub(crate) type RegularMutationFailure = OperationFailure<RegularMutationError>;
107
108#[derive(Debug, Error)]
110pub enum RegularMutationError {
111 #[error("content mutation requires a regular-file payload")]
113 NotRegular,
114 #[error("regular-file mutation range overflowed")]
116 RangeOverflow,
117 #[error("regular-file content object has the wrong kind")]
119 WrongContentKind,
120 #[error(transparent)]
122 Inline(#[from] InlineFileDataError),
123 #[error(transparent)]
125 BlobBuild(#[from] BlobBuildError),
126 #[error(transparent)]
128 BlobRead(#[from] BlobReadError),
129 #[error(transparent)]
131 Extent(#[from] ExtentMutationError),
132 #[error(transparent)]
134 Range(#[from] ExtentReadError),
135 #[error(transparent)]
137 Clone(#[from] ExtentCloneError),
138 #[error("regular-file clone source range is outside the source file")]
140 SourceRangeOutOfBounds,
141 #[error("regular-file operation requires sparse volume semantics")]
143 SparseSemanticsDisabled,
144 #[error("keep-size preallocation beyond EOF requires a physical backend capability")]
146 KeepSizeBeyondEofUnsupported,
147 #[error(transparent)]
149 Codec(#[from] CanonicalDecodeError),
150 #[error(transparent)]
152 Storage(#[from] ObjectStoreError),
153 #[error("regular-file mutation allocation failed")]
155 AllocationFailed,
156 #[error(transparent)]
158 Work(#[from] WorkError),
159}
160
161pub(crate) async fn apply_regular_mutation_async<S: AsyncObjectStore>(
162 store: &S,
163 payload: FilePayload,
164 mutation: RegularMutation,
165 config: VolumeConfig,
166 budget: WorkBudget,
167 cancellation: &CancellationToken,
168) -> Result<RegularMutationReceipt, RegularMutationFailure> {
169 cancellation
170 .check()
171 .map_err(|_| OperationFailure::before_work(cancelled()))?;
172 validate_mutation(mutation)?;
173 if !config.sparse_files {
174 let current = regular_logical_bytes(payload)?;
175 let requires_sparse = match mutation {
176 RegularMutation::Write { offset, .. } => offset > current,
177 RegularMutation::Resize { logical_bytes } => logical_bytes > current,
178 RegularMutation::ZeroRange { .. } | RegularMutation::Preallocate { .. } => true,
179 };
180 if requires_sparse {
181 return Err(OperationFailure::before_work(
182 RegularMutationError::SparseSemanticsDisabled,
183 ));
184 }
185 }
186 if let FilePayload::InlineRegular(data) = payload {
187 if let Some(receipt) =
188 try_inline(store, data, mutation, config, budget, cancellation).await?
189 {
190 return Ok(receipt);
191 }
192 let promoted = promote_inline(store, data, config, budget, cancellation).await?;
193 return apply_sparse(
194 store,
195 FilePayload::Regular {
196 logical_bytes: promoted.logical_bytes,
197 extents: promoted.extents,
198 },
199 mutation,
200 config,
201 budget,
202 promoted.work,
203 cancellation,
204 )
205 .await;
206 }
207 apply_sparse(
208 store,
209 payload,
210 mutation,
211 config,
212 budget,
213 WorkCounters::default(),
214 cancellation,
215 )
216 .await
217}
218
219#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
220pub(crate) async fn apply_regular_clone_async<S: AsyncObjectStore>(
221 store: &S,
222 source: FilePayload,
223 source_offset: u64,
224 destination: FilePayload,
225 destination_offset: u64,
226 length: u64,
227 config: VolumeConfig,
228 budget: WorkBudget,
229 cancellation: &CancellationToken,
230) -> Result<RegularCloneReceipt, RegularMutationFailure> {
231 cancellation
232 .check()
233 .map_err(|_| OperationFailure::before_work(cancelled()))?;
234 if !config.sparse_files {
235 return Err(OperationFailure::before_work(
236 RegularMutationError::SparseSemanticsDisabled,
237 ));
238 }
239 let source = RegularStorage::validate(source)?;
240 let destination = RegularStorage::validate(destination)?;
241 let source_bytes = source.logical_bytes();
242 let destination_bytes = destination.logical_bytes();
243 let source_end = source_offset
244 .checked_add(length)
245 .ok_or_else(|| OperationFailure::before_work(RegularMutationError::RangeOverflow))?;
246 let destination_end = destination_offset
247 .checked_add(length)
248 .ok_or_else(|| OperationFailure::before_work(RegularMutationError::RangeOverflow))?;
249 if length == 0 || source_end > source_bytes {
250 return Err(OperationFailure::before_work(
251 RegularMutationError::SourceRangeOutOfBounds,
252 ));
253 }
254
255 if let (RegularStorage::Inline(source_data), RegularStorage::Inline(destination_data)) =
256 (source, destination)
257 && destination_offset <= destination_bytes
258 && destination_end <= u64::try_from(MAXIMUM_INLINE_FILE_BYTES).unwrap_or(u64::MAX)
259 {
260 let source_start = usize::try_from(source_offset)
261 .map_err(|_| OperationFailure::before_work(RegularMutationError::RangeOverflow))?;
262 let source_limit = usize::try_from(source_end)
263 .map_err(|_| OperationFailure::before_work(RegularMutationError::RangeOverflow))?;
264 let work = WorkCounters {
265 bytes_copied: length,
266 ..WorkCounters::default()
267 };
268 work.verify(budget)
269 .map_err(|error| OperationFailure::before_work(error.into()))?;
270 #[allow(
271 clippy::indexing_slicing,
272 reason = "`RegularStorage` is `Copy`, so `source_data` here is the same bytes that \
273 `source.logical_bytes()` measured into `source_bytes` above; the earlier \
274 `source_end > source_bytes` check (which returns \
275 SourceRangeOutOfBounds otherwise) guarantees `source_limit == source_end \
276 <= source_data.as_bytes().len()`, and `source_start == source_offset < \
277 source_end` since the same check rejects `length == 0`"
278 )]
279 let destination = destination_data
280 .replace_range(
281 usize::try_from(destination_offset).map_err(|_| {
282 OperationFailure::before_work(RegularMutationError::RangeOverflow)
283 })?,
284 &source_data.as_bytes()[source_start..source_limit],
285 usize::try_from(destination_bytes.max(destination_end)).map_err(|_| {
286 OperationFailure::before_work(RegularMutationError::RangeOverflow)
287 })?,
288 )
289 .map_err(|error| OperationFailure::new(error.into(), work))?;
290 return Ok(RegularCloneReceipt {
291 destination: FilePayload::InlineRegular(destination),
292 work,
293 });
294 }
295
296 let equal_payloads = source == destination;
297 let mut work = WorkCounters::default();
298 let (source_logical_bytes, source_root) = match source {
299 RegularStorage::Inline(data) => {
300 let receipt =
301 promote_inline(store, data, config, remaining(work, budget)?, cancellation)
302 .await
303 .map_err(|failure| failure.map_with_prior_work(work, std::convert::identity))?;
304 work = add(work, receipt.work)?;
305 (receipt.logical_bytes, receipt.extents)
306 }
307 RegularStorage::Sparse {
308 logical_bytes,
309 extents,
310 } => (logical_bytes, extents),
311 };
312 let (destination_logical_bytes, destination_root) = if equal_payloads {
313 (source_logical_bytes, source_root)
314 } else {
315 match destination {
316 RegularStorage::Inline(data) => {
317 let receipt =
318 promote_inline(store, data, config, remaining(work, budget)?, cancellation)
319 .await
320 .map_err(|failure| {
321 failure.map_with_prior_work(work, std::convert::identity)
322 })?;
323 work = add(work, receipt.work)?;
324 (receipt.logical_bytes, receipt.extents)
325 }
326 RegularStorage::Sparse {
327 logical_bytes,
328 extents,
329 } => (logical_bytes, extents),
330 }
331 };
332 let receipt = clone_extent_range_async(
333 store,
334 ExtentCloneRequest {
335 source_root,
336 source_logical_bytes,
337 source_range: ByteRange {
338 offset: source_offset,
339 length,
340 },
341 destination_root,
342 destination_logical_bytes,
343 destination_offset,
344 maximum_spans: config.limits.maximum_mutations_per_batch,
345 maximum_mutations: config.limits.maximum_mutations_per_batch,
346 },
347 decode_limits(config),
348 remaining(work, budget)?,
349 cancellation,
350 )
351 .await
352 .map_err(|failure| failure.map_with_prior_work(work, Into::into))?;
353 work = add(work, receipt.work)?;
354 if receipt.logical_bytes <= u64::try_from(MAXIMUM_INLINE_FILE_BYTES).unwrap_or(u64::MAX) {
355 let attempt = try_demote_sparse_async(
356 store,
357 receipt.root,
358 receipt.logical_bytes,
359 receipt.logical_bytes,
360 config,
361 remaining(work, budget)?,
362 cancellation,
363 )
364 .await
365 .map_err(|failure| failure.map_with_prior_work(work, std::convert::identity))?;
366 work = add(work, attempt.work)?;
367 if let Some(data) = attempt.data {
368 return Ok(RegularCloneReceipt {
369 destination: FilePayload::InlineRegular(data),
370 work,
371 });
372 }
373 }
374 Ok(RegularCloneReceipt {
375 destination: FilePayload::Regular {
376 logical_bytes: receipt.logical_bytes,
377 extents: receipt.root,
378 },
379 work,
380 })
381}
382
383fn regular_logical_bytes(payload: FilePayload) -> Result<u64, RegularMutationFailure> {
384 match payload {
385 FilePayload::InlineRegular(data) => {
386 Ok(crate::foundation::usize_to_u64(data.as_bytes().len()))
387 }
388 FilePayload::Regular { logical_bytes, .. } => Ok(logical_bytes),
389 FilePayload::Directory { .. }
390 | FilePayload::SymbolicLink { .. }
391 | FilePayload::Empty
392 | FilePayload::Device { .. }
393 | FilePayload::ReparsePoint { .. } => Err(OperationFailure::before_work(
394 RegularMutationError::NotRegular,
395 )),
396 }
397}
398
399fn validate_mutation(mutation: RegularMutation) -> Result<(), RegularMutationFailure> {
400 match mutation {
401 RegularMutation::Write {
402 offset,
403 length,
404 content,
405 content_offset,
406 } => {
407 if content.kind != ObjectKind::Blob {
408 return Err(OperationFailure::before_work(
409 RegularMutationError::WrongContentKind,
410 ));
411 }
412 if length == 0
413 || offset.checked_add(length).is_none()
414 || content_offset.checked_add(length).is_none()
415 {
416 return Err(OperationFailure::before_work(
417 RegularMutationError::RangeOverflow,
418 ));
419 }
420 }
421 RegularMutation::ZeroRange { offset, length, .. }
422 | RegularMutation::Preallocate { offset, length, .. } => {
423 if length == 0 || offset.checked_add(length).is_none() {
424 return Err(OperationFailure::before_work(
425 RegularMutationError::RangeOverflow,
426 ));
427 }
428 }
429 RegularMutation::Resize { .. } => {}
430 }
431 Ok(())
432}
433
434#[allow(clippy::too_many_lines)]
435async fn try_inline<S: AsyncObjectStore>(
436 store: &S,
437 data: InlineFileData,
438 mutation: RegularMutation,
439 config: VolumeConfig,
440 budget: WorkBudget,
441 cancellation: &CancellationToken,
442) -> Result<Option<RegularMutationReceipt>, RegularMutationFailure> {
443 match mutation {
444 RegularMutation::Resize { logical_bytes } => {
445 let current = u64::try_from(data.as_bytes().len()).unwrap_or(u64::MAX);
446 if logical_bytes > current {
447 return Ok(None);
448 }
449 let length = usize::try_from(logical_bytes)
450 .map_err(|_| OperationFailure::before_work(RegularMutationError::RangeOverflow))?;
451 Ok(Some(RegularMutationReceipt {
452 payload: FilePayload::InlineRegular(
453 data.truncate(length)
454 .map_err(|error| OperationFailure::before_work(error.into()))?,
455 ),
456 work: WorkCounters::default(),
457 }))
458 }
459 RegularMutation::ZeroRange {
460 offset,
461 length,
462 allocated,
463 extend,
464 } => {
465 let current = u64::try_from(data.as_bytes().len()).unwrap_or(u64::MAX);
466 let requested_end = offset.checked_add(length).ok_or_else(|| {
467 OperationFailure::before_work(RegularMutationError::RangeOverflow)
468 })?;
469 if !extend && offset >= current {
470 return Ok(Some(RegularMutationReceipt {
471 payload: FilePayload::InlineRegular(data),
472 work: WorkCounters::default(),
473 }));
474 }
475 if !allocated {
476 return Ok(None);
477 }
478 if extend && offset > current {
479 return Ok(None);
480 }
481 let logical_bytes = if extend {
482 requested_end.max(current)
483 } else {
484 current
485 };
486 if logical_bytes > u64::try_from(MAXIMUM_INLINE_FILE_BYTES).unwrap_or(u64::MAX) {
487 return Ok(None);
488 }
489 let end = requested_end.min(logical_bytes);
490 let copied = end.checked_sub(offset).ok_or_else(|| {
491 OperationFailure::before_work(RegularMutationError::RangeOverflow)
492 })?;
493 let work = WorkCounters {
494 bytes_copied: copied,
495 ..WorkCounters::default()
496 };
497 work.verify(budget)
498 .map_err(|error| OperationFailure::before_work(error.into()))?;
499 let zeros = [0_u8; MAXIMUM_INLINE_FILE_BYTES];
500 #[allow(
501 clippy::indexing_slicing,
502 reason = "`logical_bytes <= MAXIMUM_INLINE_FILE_BYTES` is checked just above (else \
503 this returns `Ok(None)`), `end = requested_end.min(logical_bytes)` so \
504 `end <= logical_bytes <= MAXIMUM_INLINE_FILE_BYTES`, and \
505 `copied = end.checked_sub(offset)` so `copied <= end`; hence `copied <= \
506 MAXIMUM_INLINE_FILE_BYTES == zeros.len()`, so this slice never exceeds \
507 the fixed-size `zeros` array"
508 )]
509 let replacement = &zeros[..usize::try_from(copied)
510 .map_err(|_| OperationFailure::before_work(RegularMutationError::RangeOverflow))?];
511 let payload = data
512 .replace_range(
513 usize::try_from(offset).map_err(|_| {
514 OperationFailure::before_work(RegularMutationError::RangeOverflow)
515 })?,
516 replacement,
517 usize::try_from(logical_bytes).map_err(|_| {
518 OperationFailure::before_work(RegularMutationError::RangeOverflow)
519 })?,
520 )
521 .map_err(|error| OperationFailure::new(error.into(), work))?;
522 Ok(Some(RegularMutationReceipt {
523 payload: FilePayload::InlineRegular(payload),
524 work,
525 }))
526 }
527 RegularMutation::Preallocate {
528 offset,
529 length,
530 keep_size,
531 } => {
532 let current = u64::try_from(data.as_bytes().len()).unwrap_or(u64::MAX);
533 let end = offset.checked_add(length).ok_or_else(|| {
534 OperationFailure::before_work(RegularMutationError::RangeOverflow)
535 })?;
536 if keep_size && end > current {
537 return Err(OperationFailure::before_work(
538 RegularMutationError::KeepSizeBeyondEofUnsupported,
539 ));
540 }
541 if end <= current {
542 return Ok(Some(RegularMutationReceipt {
543 payload: FilePayload::InlineRegular(data),
544 work: WorkCounters::default(),
545 }));
546 }
547 Ok(None)
548 }
549 RegularMutation::Write {
550 offset,
551 length,
552 content,
553 content_offset,
554 } => {
555 let current = u64::try_from(data.as_bytes().len()).unwrap_or(u64::MAX);
556 let end = offset.checked_add(length).ok_or_else(|| {
557 OperationFailure::before_work(RegularMutationError::RangeOverflow)
558 })?;
559 if offset > current
560 || end > u64::try_from(MAXIMUM_INLINE_FILE_BYTES).unwrap_or(u64::MAX)
561 {
562 return Ok(None);
563 }
564 let read = read_blob_range_async(
565 store,
566 content,
567 ByteRange {
568 offset: content_offset,
569 length,
570 },
571 decode_limits(config),
572 budget,
573 cancellation,
574 )
575 .await
576 .map_err(|failure| OperationFailure::new(failure.error.into(), *failure.work))?;
577 let mut work = read.work;
578 let copy = WorkCounters {
579 bytes_copied: length,
580 ..WorkCounters::default()
581 };
582 work = add(work, copy)?;
583 work.verify(budget)
584 .map_err(|error| OperationFailure::new(error.into(), read.work))?;
585 let logical_bytes = current.max(end);
586 let payload = data
587 .replace_range(
588 usize::try_from(offset).map_err(|_| {
589 OperationFailure::new(RegularMutationError::RangeOverflow, work)
590 })?,
591 &read.bytes,
592 usize::try_from(logical_bytes).map_err(|_| {
593 OperationFailure::new(RegularMutationError::RangeOverflow, work)
594 })?,
595 )
596 .map_err(|error| OperationFailure::new(error.into(), work))?;
597 Ok(Some(RegularMutationReceipt {
598 payload: FilePayload::InlineRegular(payload),
599 work,
600 }))
601 }
602 }
603}
604
605async fn promote_inline<S: AsyncObjectStore>(
606 store: &S,
607 data: InlineFileData,
608 config: VolumeConfig,
609 budget: WorkBudget,
610 cancellation: &CancellationToken,
611) -> Result<PromotedRegularReceipt, RegularMutationFailure> {
612 let logical_bytes = u64::try_from(data.as_bytes().len()).unwrap_or(u64::MAX);
613 let mut allocations = AllocationLedger::default();
614 let (extents, mut work, extent_bytes) = if logical_bytes == 0 {
615 (Vec::new(), WorkCounters::default(), 0)
616 } else {
617 let mut source = Cursor::new(data.as_bytes());
618 let blob = build_blob_async(
619 store,
620 &mut source,
621 BlobBuildOptions {
622 chunk_bytes: u32::try_from(MAXIMUM_INLINE_FILE_BYTES).unwrap_or(u32::MAX),
623 page_items: config.limits.maximum_directory_page_entries,
624 page_bytes: u32::try_from(config.limits.maximum_object_bytes).unwrap_or(u32::MAX),
625 maximum_blob_bytes: logical_bytes,
626 },
627 budget,
628 cancellation,
629 )
630 .await
631 .map_err(|failure| OperationFailure::new(failure.error.into(), *failure.work))?;
632 let mut work = blob.work;
633 let extent_bytes = allocations
634 .claim_elements::<Extent>(1, &mut work, budget)
635 .map_err(|error| allocation_error(error, work))?;
636 let mut extents = Vec::new();
637 if extents.try_reserve_exact(1).is_err() {
638 allocations
639 .release(extent_bytes)
640 .map_err(|error| allocation_error(error, work))?;
641 return Err(OperationFailure::new(
642 RegularMutationError::AllocationFailed,
643 work,
644 ));
645 }
646 extents.push(Extent {
647 offset: 0,
648 length: logical_bytes,
649 kind: ExtentKind::Content {
650 object: blob.root,
651 object_offset: 0,
652 },
653 });
654 (extents, work, extent_bytes)
655 };
656 let page = ExtentPage::Leaf(extents);
657 let mut page_budget = remaining(work, budget)?;
658 page_budget.peak_allocation_bytes = page_budget
659 .peak_allocation_bytes
660 .checked_sub(allocations.live_bytes())
661 .ok_or_else(|| OperationFailure::new(RegularMutationError::RangeOverflow, work))?;
662 let written = put_extent_page_async(
663 store,
664 &page,
665 decode_limits(config),
666 page_budget,
667 cancellation,
668 )
669 .await
670 .map_err(|failure| {
671 simultaneous_failure(work, *failure.work, allocations.live_bytes(), failure.error)
672 })?;
673 work = simultaneous(work, written.work, allocations.live_bytes(), budget)?;
674 allocations
675 .release(extent_bytes)
676 .map_err(|error| allocation_error(error, work))?;
677 Ok(PromotedRegularReceipt {
678 logical_bytes,
679 extents: written.object,
680 work,
681 })
682}
683
684#[allow(clippy::too_many_lines)]
685async fn apply_sparse<S: AsyncObjectStore>(
686 store: &S,
687 payload: FilePayload,
688 mutation: RegularMutation,
689 config: VolumeConfig,
690 budget: WorkBudget,
691 prior: WorkCounters,
692 cancellation: &CancellationToken,
693) -> Result<RegularMutationReceipt, RegularMutationFailure> {
694 let FilePayload::Regular {
695 logical_bytes,
696 extents,
697 } = payload
698 else {
699 return Err(OperationFailure::new(
700 RegularMutationError::NotRegular,
701 prior,
702 ));
703 };
704 let mut prior = prior;
705 let mut demotion_known_impossible = false;
706 if let RegularMutation::ZeroRange {
707 offset,
708 extend: false,
709 ..
710 } = mutation
711 && offset >= logical_bytes
712 {
713 return Ok(RegularMutationReceipt {
714 payload,
715 work: prior,
716 });
717 }
718 if let RegularMutation::Resize {
719 logical_bytes: target,
720 } = mutation
721 && target <= u64::try_from(MAXIMUM_INLINE_FILE_BYTES).unwrap_or(u64::MAX)
722 && target <= logical_bytes
723 {
724 let attempt = try_demote_sparse_async(
725 store,
726 extents,
727 logical_bytes,
728 target,
729 config,
730 remaining(prior, budget)?,
731 cancellation,
732 )
733 .await
734 .map_err(|failure| failure.map_with_prior_work(prior, std::convert::identity))?;
735 prior = add(prior, attempt.work)?;
736 if let Some(data) = attempt.data {
737 return Ok(RegularMutationReceipt {
738 payload: FilePayload::InlineRegular(data),
739 work: prior,
740 });
741 }
742 demotion_known_impossible = true;
743 }
744 let extent_mutation = match mutation {
745 RegularMutation::Write {
746 offset,
747 length,
748 content,
749 content_offset,
750 } => ExtentMutation::Replace {
751 offset,
752 length,
753 kind: ExtentKind::Content {
754 object: content,
755 object_offset: content_offset,
756 },
757 extend: true,
758 },
759 RegularMutation::Resize { logical_bytes } => ExtentMutation::Resize { logical_bytes },
760 RegularMutation::ZeroRange {
761 offset,
762 mut length,
763 allocated,
764 extend,
765 } => {
766 if !extend {
767 length = length.min(logical_bytes - offset);
768 }
769 ExtentMutation::Replace {
770 offset,
771 length,
772 kind: if allocated {
773 ExtentKind::AllocatedZero
774 } else {
775 ExtentKind::Hole
776 },
777 extend,
778 }
779 }
780 RegularMutation::Preallocate {
781 offset,
782 length,
783 keep_size,
784 } => {
785 return apply_sparse_preallocation(
786 store,
787 logical_bytes,
788 extents,
789 offset,
790 length,
791 keep_size,
792 config,
793 budget,
794 prior,
795 cancellation,
796 )
797 .await;
798 }
799 };
800 let receipt = apply_extent_mutations_async(
801 store,
802 extents,
803 logical_bytes,
804 &[extent_mutation],
805 ExtentMutationOptions {
806 maximum_mutations: 1,
807 limits: decode_limits(config),
808 budget: remaining(prior, budget)?,
809 },
810 cancellation,
811 )
812 .await
813 .map_err(|failure| failure.map_with_prior_work(prior, Into::into))?;
814 let work = add(prior, receipt.work)?;
815 if !demotion_known_impossible
816 && receipt.logical_bytes <= u64::try_from(MAXIMUM_INLINE_FILE_BYTES).unwrap_or(u64::MAX)
817 {
818 let attempt = try_demote_sparse_async(
819 store,
820 receipt.root,
821 receipt.logical_bytes,
822 receipt.logical_bytes,
823 config,
824 remaining(work, budget)?,
825 cancellation,
826 )
827 .await
828 .map_err(|failure| failure.map_with_prior_work(work, std::convert::identity))?;
829 let work = add(work, attempt.work)?;
830 if let Some(data) = attempt.data {
831 return Ok(RegularMutationReceipt {
832 payload: FilePayload::InlineRegular(data),
833 work,
834 });
835 }
836 return Ok(RegularMutationReceipt {
837 payload: FilePayload::Regular {
838 logical_bytes: receipt.logical_bytes,
839 extents: receipt.root,
840 },
841 work,
842 });
843 }
844 Ok(RegularMutationReceipt {
845 payload: FilePayload::Regular {
846 logical_bytes: receipt.logical_bytes,
847 extents: receipt.root,
848 },
849 work,
850 })
851}
852
853struct InlineDemotionAttempt {
854 data: Option<InlineFileData>,
855 work: WorkCounters,
856}
857
858struct SparsePreallocationPlan {
859 mutations: Vec<ExtentMutation>,
860 mutation_bytes: u64,
861 work: WorkCounters,
862}
863
864#[allow(clippy::too_many_arguments)]
865async fn apply_sparse_preallocation<S: AsyncObjectStore>(
866 store: &S,
867 logical_bytes: u64,
868 extents: ObjectId,
869 offset: u64,
870 length: u64,
871 keep_size: bool,
872 config: VolumeConfig,
873 budget: WorkBudget,
874 prior: WorkCounters,
875 cancellation: &CancellationToken,
876) -> Result<RegularMutationReceipt, RegularMutationFailure> {
877 let end = offset
878 .checked_add(length)
879 .ok_or_else(|| OperationFailure::new(RegularMutationError::RangeOverflow, prior))?;
880 if keep_size && end > logical_bytes {
881 return Err(OperationFailure::new(
882 RegularMutationError::KeepSizeBeyondEofUnsupported,
883 prior,
884 ));
885 }
886 let plan = plan_sparse_preallocation(
887 store,
888 extents,
889 logical_bytes,
890 offset,
891 end,
892 config,
893 budget,
894 prior,
895 cancellation,
896 )
897 .await?;
898 if plan.mutations.is_empty() {
899 return Ok(RegularMutationReceipt {
900 payload: FilePayload::Regular {
901 logical_bytes,
902 extents,
903 },
904 work: plan.work,
905 });
906 }
907 let mut nested_budget = remaining(plan.work, budget)?;
908 nested_budget.peak_allocation_bytes = nested_budget
909 .peak_allocation_bytes
910 .checked_sub(plan.mutation_bytes)
911 .ok_or_else(|| OperationFailure::new(RegularMutationError::RangeOverflow, plan.work))?;
912 let receipt = apply_extent_mutations_async(
913 store,
914 extents,
915 logical_bytes,
916 &plan.mutations,
917 ExtentMutationOptions {
918 maximum_mutations: config.limits.maximum_mutations_per_batch,
919 limits: decode_limits(config),
920 budget: nested_budget,
921 },
922 cancellation,
923 )
924 .await
925 .map_err(|failure| {
926 simultaneous_failure(
927 plan.work,
928 *failure.work,
929 plan.mutation_bytes,
930 failure.error.into(),
931 )
932 })?;
933 let work = simultaneous(plan.work, receipt.work, plan.mutation_bytes, budget)?;
934 Ok(RegularMutationReceipt {
935 payload: FilePayload::Regular {
936 logical_bytes: receipt.logical_bytes,
937 extents: receipt.root,
938 },
939 work,
940 })
941}
942
943#[allow(clippy::too_many_arguments)]
944#[allow(clippy::too_many_lines)]
945async fn plan_sparse_preallocation<S: AsyncObjectStore>(
946 store: &S,
947 extents: ObjectId,
948 logical_bytes: u64,
949 offset: u64,
950 end: u64,
951 config: VolumeConfig,
952 budget: WorkBudget,
953 prior: WorkCounters,
954 cancellation: &CancellationToken,
955) -> Result<SparsePreallocationPlan, RegularMutationFailure> {
956 let existing_end = end.min(logical_bytes);
957 let existing_length = existing_end.saturating_sub(offset.min(existing_end));
958 let (spans, retained, mut work) = if existing_length == 0 {
959 (Vec::new(), 0, prior)
960 } else {
961 let plan = plan_extent_range_async(
962 store,
963 ExtentRangeRequest {
964 root: extents,
965 file_size: logical_bytes,
966 range: ByteRange {
967 offset,
968 length: existing_length,
969 },
970 maximum_spans: config.limits.maximum_mutations_per_batch,
971 limits: decode_limits(config),
972 budget: remaining(prior, budget)?,
973 },
974 cancellation,
975 )
976 .await
977 .map_err(|failure| failure.map_with_prior_work(prior, Into::into))?;
978 let work = add(prior, plan.work)?;
979 (plan.spans, plan.retained_allocation_bytes, work)
980 };
981 let scan = WorkCounters {
982 items_examined: u64::try_from(spans.len()).unwrap_or(u64::MAX),
983 ..WorkCounters::default()
984 };
985 let scanned = work
986 .checked_add(scan)
987 .map_err(|error| OperationFailure::new(error.into(), work))?;
988 scanned
989 .verify(budget)
990 .map_err(|error| OperationFailure::new(error.into(), work))?;
991 work = scanned;
992 let hole_count = spans
993 .iter()
994 .filter(|span| matches!(span.kind, ExtentKind::Hole))
995 .count();
996 let grows_file = end > logical_bytes;
997 let extension_offset = logical_bytes.max(offset);
998 let merge_extension = grows_file
999 && spans.last().is_some_and(|span| {
1000 matches!(span.kind, ExtentKind::Hole)
1001 && span.offset.checked_add(span.length) == Some(extension_offset)
1002 });
1003 let mutation_count = hole_count
1004 .checked_add(usize::from(grows_file))
1005 .and_then(|count| count.checked_sub(usize::from(merge_extension)))
1006 .ok_or_else(|| OperationFailure::new(RegularMutationError::RangeOverflow, work))?;
1007 if u32::try_from(mutation_count).unwrap_or(u32::MAX) > config.limits.maximum_mutations_per_batch
1008 {
1009 return Err(OperationFailure::new(
1010 RegularMutationError::Extent(ExtentMutationError::TooManyMutations),
1011 work,
1012 ));
1013 }
1014 if mutation_count == 0 {
1015 return Ok(SparsePreallocationPlan {
1016 mutations: Vec::new(),
1017 mutation_bytes: 0,
1018 work,
1019 });
1020 }
1021 let mutation_bytes = mutation_count
1022 .checked_mul(size_of::<ExtentMutation>())
1023 .map(crate::foundation::usize_to_u64)
1024 .ok_or_else(|| OperationFailure::new(RegularMutationError::RangeOverflow, work))?;
1025 let simultaneous_bytes = retained
1026 .checked_add(mutation_bytes)
1027 .ok_or_else(|| OperationFailure::new(RegularMutationError::RangeOverflow, work))?;
1028 let allocation_attempt = work
1029 .checked_add(WorkCounters {
1030 allocation_operations: 1,
1031 ..WorkCounters::default()
1032 })
1033 .map_err(|error| OperationFailure::new(error.into(), work))?;
1034 let mut admitted = allocation_attempt;
1035 admitted.peak_allocation_bytes = admitted.peak_allocation_bytes.max(simultaneous_bytes);
1036 admitted
1037 .verify(budget)
1038 .map_err(|error| OperationFailure::new(error.into(), work))?;
1039 let mut mutations = Vec::new();
1040 if mutations.try_reserve_exact(mutation_count).is_err() {
1041 return Err(OperationFailure::new(
1042 RegularMutationError::AllocationFailed,
1043 allocation_attempt,
1044 ));
1045 }
1046 work = admitted;
1047 mutations.extend(
1048 spans
1049 .iter()
1050 .filter(|span| matches!(span.kind, ExtentKind::Hole))
1051 .map(|span| ExtentMutation::Replace {
1052 offset: span.offset,
1053 length: if merge_extension
1054 && span.offset.checked_add(span.length) == Some(extension_offset)
1055 {
1056 end - span.offset
1057 } else {
1058 span.length
1059 },
1060 kind: ExtentKind::AllocatedZero,
1061 extend: merge_extension
1062 && span.offset.checked_add(span.length) == Some(extension_offset),
1063 }),
1064 );
1065 if grows_file && !merge_extension {
1066 mutations.push(ExtentMutation::Replace {
1067 offset: extension_offset,
1068 length: end - extension_offset,
1069 kind: ExtentKind::AllocatedZero,
1070 extend: true,
1071 });
1072 }
1073 drop(spans);
1074 Ok(SparsePreallocationPlan {
1075 mutations,
1076 mutation_bytes,
1077 work,
1078 })
1079}
1080
1081#[allow(clippy::too_many_arguments)]
1082async fn try_demote_sparse_async<S: AsyncObjectStore>(
1083 store: &S,
1084 root: ObjectId,
1085 file_size: u64,
1086 target_size: u64,
1087 config: VolumeConfig,
1088 budget: WorkBudget,
1089 cancellation: &CancellationToken,
1090) -> Result<InlineDemotionAttempt, RegularMutationFailure> {
1091 if target_size == 0 {
1092 return Ok(InlineDemotionAttempt {
1093 data: Some(
1094 InlineFileData::new(&[])
1095 .map_err(|error| OperationFailure::before_work(error.into()))?,
1096 ),
1097 work: WorkCounters::default(),
1098 });
1099 }
1100 let plan = plan_extent_range_async(
1101 store,
1102 ExtentRangeRequest {
1103 root,
1104 file_size,
1105 range: ByteRange {
1106 offset: 0,
1107 length: target_size,
1108 },
1109 maximum_spans: config.limits.maximum_mutations_per_batch,
1110 limits: decode_limits(config),
1111 budget,
1112 },
1113 cancellation,
1114 )
1115 .await
1116 .map_err(|failure| OperationFailure::new(failure.error.into(), *failure.work))?;
1117 let retained = plan.retained_allocation_bytes;
1118 let mut work = plan.work;
1119 if plan
1120 .spans
1121 .iter()
1122 .any(|span| !matches!(span.kind, ExtentKind::Content { .. }))
1123 {
1124 return Ok(InlineDemotionAttempt { data: None, work });
1125 }
1126 let target = usize::try_from(target_size)
1127 .map_err(|_| OperationFailure::new(RegularMutationError::RangeOverflow, work))?;
1128 let mut data =
1129 InlineFileData::new(&[]).map_err(|error| OperationFailure::new(error.into(), work))?;
1130 for span in &plan.spans {
1131 let ExtentKind::Content {
1132 object,
1133 object_offset,
1134 } = span.kind
1135 else {
1136 return Ok(InlineDemotionAttempt { data: None, work });
1137 };
1138 let mut nested_budget = remaining(work, budget)?;
1139 nested_budget.peak_allocation_bytes = nested_budget
1140 .peak_allocation_bytes
1141 .checked_sub(retained)
1142 .ok_or_else(|| OperationFailure::new(RegularMutationError::RangeOverflow, work))?;
1143 let read = read_blob_range_async(
1144 store,
1145 object,
1146 ByteRange {
1147 offset: object_offset,
1148 length: span.length,
1149 },
1150 decode_limits(config),
1151 nested_budget,
1152 cancellation,
1153 )
1154 .await
1155 .map_err(|failure| {
1156 simultaneous_failure(work, *failure.work, retained, failure.error.into())
1157 })?;
1158 work = simultaneous(work, read.work, retained, budget)?;
1159 let start = usize::try_from(span.offset)
1160 .map_err(|_| OperationFailure::new(RegularMutationError::RangeOverflow, work))?;
1161 data = data
1162 .replace_range(start, &read.bytes, target)
1163 .map_err(|error| OperationFailure::new(error.into(), work))?;
1164 work = add(
1165 work,
1166 WorkCounters {
1167 bytes_copied: span.length,
1168 ..WorkCounters::default()
1169 },
1170 )?;
1171 work.verify(budget)
1172 .map_err(|error| OperationFailure::new(error.into(), work))?;
1173 }
1174 Ok(InlineDemotionAttempt {
1175 data: Some(data),
1176 work,
1177 })
1178}
1179
1180struct ExtentPageWrite {
1181 object: ObjectId,
1182 work: WorkCounters,
1183}
1184
1185async fn put_extent_page_async<S: AsyncObjectStore>(
1186 store: &S,
1187 page: &ExtentPage,
1188 limits: DecodeLimits,
1189 budget: WorkBudget,
1190 cancellation: &CancellationToken,
1191) -> Result<ExtentPageWrite, RegularMutationFailure> {
1192 let encoded_length = extent_page_encoded_length(page, limits.maximum_page_items)
1193 .map_err(|error| OperationFailure::before_work(error.into()))?;
1194 let encoded_bytes = crate::foundation::usize_to_u64(encoded_length);
1195 if encoded_bytes > limits.maximum_page_object_bytes() {
1196 return Err(OperationFailure::before_work(
1197 CanonicalDecodeError::ObjectTooLarge {
1198 observed: encoded_bytes,
1199 maximum: limits.maximum_page_object_bytes(),
1200 }
1201 .into(),
1202 ));
1203 }
1204 let mut work = WorkCounters {
1205 allocation_operations: u64::from(encoded_bytes != 0),
1206 peak_allocation_bytes: encoded_bytes,
1207 ..WorkCounters::default()
1208 };
1209 work.verify(budget)
1210 .map_err(|error| OperationFailure::before_work(error.into()))?;
1211 let encoded = encode_extent_page(page, limits.maximum_page_items)
1212 .map_err(|error| OperationFailure::new(error.into(), work))?;
1213 if u64::try_from(encoded.capacity()).unwrap_or(u64::MAX) != encoded_bytes {
1214 return Err(OperationFailure::new(
1215 RegularMutationError::AllocationFailed,
1216 work,
1217 ));
1218 }
1219 work = add(
1220 work,
1221 WorkCounters {
1222 bytes_encoded: encoded_bytes,
1223 ..WorkCounters::default()
1224 },
1225 )?;
1226 let hashed = work
1227 .checked_add(WorkCounters {
1228 bytes_hashed: encoded_bytes
1229 .checked_add(OBJECT_DIGEST_ENVELOPE_BYTES)
1230 .ok_or_else(|| OperationFailure::new(RegularMutationError::RangeOverflow, work))?,
1231 page_writes: 1,
1232 ..WorkCounters::default()
1233 })
1234 .map_err(|error| OperationFailure::new(error.into(), work))?;
1235 hashed
1236 .verify(budget)
1237 .map_err(|error| OperationFailure::new(error.into(), work))?;
1238 let object = ObjectId {
1239 kind: ObjectKind::ExtentPage,
1240 digest: object_digest(ObjectKind::ExtentPage, &encoded),
1241 };
1242 let mut backend_budget = remaining(hashed, budget)?;
1243 backend_budget.peak_allocation_bytes = backend_budget
1244 .peak_allocation_bytes
1245 .checked_sub(encoded_bytes)
1246 .ok_or_else(|| OperationFailure::new(RegularMutationError::RangeOverflow, hashed))?;
1247 let receipt = store
1248 .put(object, Bytes::from(encoded), backend_budget, cancellation)
1249 .await
1250 .map_err(|failure| {
1251 simultaneous_failure(hashed, *failure.work, encoded_bytes, failure.error.into())
1252 })?;
1253 Ok(ExtentPageWrite {
1254 object,
1255 work: simultaneous(hashed, receipt.work, encoded_bytes, budget)?,
1256 })
1257}
1258
1259fn decode_limits(config: VolumeConfig) -> DecodeLimits {
1260 DecodeLimits {
1261 maximum_object_bytes: config.limits.maximum_object_bytes,
1262 maximum_name_bytes: config.limits.maximum_component_bytes,
1263 maximum_page_items: config.limits.maximum_directory_page_entries,
1264 maximum_page_bytes: u32::try_from(config.limits.maximum_object_bytes).unwrap_or(u32::MAX),
1265 maximum_page_height: config.limits.maximum_page_height,
1266 maximum_visited_pages: u32::try_from(config.limits.maximum_objects_per_generation)
1267 .unwrap_or(u32::MAX),
1268 }
1269}
1270
1271fn add(prior: WorkCounters, next: WorkCounters) -> Result<WorkCounters, RegularMutationFailure> {
1272 prior
1273 .checked_add(next)
1274 .map_err(|error| OperationFailure::new(error.into(), prior))
1275}
1276
1277fn remaining(work: WorkCounters, budget: WorkBudget) -> Result<WorkBudget, RegularMutationFailure> {
1278 work.remaining(budget)
1279 .map_err(|error| OperationFailure::new(error.into(), work))
1280}
1281
1282fn simultaneous(
1283 prior: WorkCounters,
1284 mut nested: WorkCounters,
1285 live_bytes: u64,
1286 budget: WorkBudget,
1287) -> Result<WorkCounters, RegularMutationFailure> {
1288 let peak = live_bytes
1289 .checked_add(nested.peak_allocation_bytes)
1290 .ok_or_else(|| OperationFailure::new(RegularMutationError::RangeOverflow, prior))?;
1291 nested.peak_allocation_bytes = 0;
1292 let mut work = add(prior, nested)?;
1293 work.peak_allocation_bytes = work.peak_allocation_bytes.max(peak);
1294 work.verify(budget)
1295 .map_err(|error| OperationFailure::new(error.into(), work))?;
1296 Ok(work)
1297}
1298
1299fn simultaneous_failure(
1300 prior: WorkCounters,
1301 mut nested: WorkCounters,
1302 live_bytes: u64,
1303 error: RegularMutationError,
1304) -> RegularMutationFailure {
1305 let Some(peak) = live_bytes.checked_add(nested.peak_allocation_bytes) else {
1306 return OperationFailure::new(RegularMutationError::RangeOverflow, prior);
1307 };
1308 nested.peak_allocation_bytes = 0;
1309 let Ok(mut work) = prior.checked_add(nested) else {
1310 return OperationFailure::new(RegularMutationError::Work(WorkError::Overflow), prior);
1311 };
1312 work.peak_allocation_bytes = work.peak_allocation_bytes.max(peak);
1313 OperationFailure::new(error, work)
1314}
1315
1316fn cancelled() -> RegularMutationError {
1317 RegularMutationError::Storage(ObjectStoreError::Cancelled)
1318}
1319
1320fn allocation_error(error: AllocationError, work: WorkCounters) -> RegularMutationFailure {
1321 match error {
1322 AllocationError::Work(error) => OperationFailure::new(error.into(), work),
1323 AllocationError::Overflow | AllocationError::ReleaseInvariant => {
1324 OperationFailure::new(RegularMutationError::RangeOverflow, work)
1325 }
1326 AllocationError::InvalidCapacity
1327 | AllocationError::CapacityExceeded
1328 | AllocationError::AllocationFailed => {
1329 OperationFailure::new(RegularMutationError::AllocationFailed, work)
1330 }
1331 }
1332}
1333
1334#[cfg(all(test, feature = "memory"))]
1335#[path = "tests/regular_mutation.rs"]
1336mod tests;