Skip to main content

fxfs/object_store/
data_object_handle.rs

1// Copyright 2021 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use crate::errors::FxfsError;
6use crate::log::*;
7use crate::lsm_tree::Query;
8use crate::lsm_tree::types::{ItemRef, LayerIterator};
9use crate::object_handle::{
10    ObjectHandle, ObjectProperties, ReadObjectHandle, WriteBytes, WriteObjectHandle,
11};
12use crate::object_store::extent_record::{ExtentMode, ExtentValue};
13use crate::object_store::object_manager::ObjectManager;
14use crate::object_store::object_record::{
15    AttributeKey, DirType, FsverityMetadata, ObjectAttributes, ObjectItem, ObjectKey,
16    ObjectKeyData, ObjectKind, ObjectValue, Timestamp,
17};
18use crate::object_store::store_object_handle::{MaybeChecksums, NeedsTrim};
19use crate::object_store::transaction::{
20    self, AssocObj, AssociatedObject, LockKey, Mutation, ObjectStoreMutation, Operation, Options,
21    Transaction, lock_keys,
22};
23use crate::object_store::{
24    AttributeId, Extent, HandleOptions, HandleOwner, RootDigest, StoreObjectHandle,
25    TRANSACTION_MUTATION_THRESHOLD, TrimMode, TrimResult,
26};
27use crate::range::RangeExt;
28use crate::round::{round_down, round_up};
29use anyhow::{Context, Error, anyhow, bail, ensure};
30use async_trait::async_trait;
31use fidl_fuchsia_io as fio;
32use fsverity_merkle::{
33    FsVerityDescriptor, FsVerityDescriptorRaw, FsVerityHash, FsVerityHasher, FsVerityHasherOptions,
34    MerkleTree, MerkleTreeBuilder, Sha256Hash, Sha512Hash,
35};
36use fuchsia_sync::Mutex;
37use futures::TryStreamExt;
38use futures::stream::FuturesUnordered;
39use fxfs_trace::trace;
40use std::cmp::min;
41use std::ops::{Deref, Range};
42use std::sync::Arc;
43use std::sync::atomic::{self, AtomicU64, Ordering};
44use storage_device::buffer::{Buffer, BufferFuture, BufferRef, MutableBufferRef};
45use storage_ptr_slice::PtrByteSlice;
46use zerocopy::FromBytes;
47
48mod allocated_ranges;
49pub use allocated_ranges::{AllocatedRanges, RangeType};
50
51/// How much data each transaction will cover when writing an attribute across batches. Pulled from
52/// `FLUSH_BATCH_SIZE` in paged_object_handle.rs.
53pub const WRITE_ATTR_BATCH_SIZE: usize = 524_288;
54
55/// DataObjectHandle is a typed handle for file-like objects that store data in the default data
56/// attribute. In addition to traditional files, this means things like the journal, superblocks,
57/// and layer files.
58///
59/// It caches the content size of the data attribute it was configured for, and has helpers for
60/// complex extent manipulation, as well as implementations of ReadObjectHandle and
61/// WriteObjectHandle.
62pub struct DataObjectHandle<S: HandleOwner> {
63    handle: StoreObjectHandle<S>,
64    attribute_id: AttributeId,
65    content_size: AtomicU64,
66    state: Mutex<DataObjectState>,
67}
68
69/// Represents the mapping of a file's contents to the physical storage backing it.
70#[derive(Debug, Clone)]
71pub struct FileExtent {
72    logical_offset: u64,
73    device_range: Range<u64>,
74}
75
76impl FileExtent {
77    pub fn new(logical_offset: u64, device_range: Range<u64>) -> Result<Self, Error> {
78        // Ensure `device_range` is valid.
79        let length = device_range.length()?;
80        // Ensure no overflow when we calculate the end of the logical range.
81        let _ = logical_offset.checked_add(length).ok_or(FxfsError::OutOfRange)?;
82        Ok(Self { logical_offset, device_range })
83    }
84}
85
86impl FileExtent {
87    pub fn length(&self) -> u64 {
88        // SAFETY: We verified that the device_range's length is valid in Self::new.
89        unsafe { self.device_range.unchecked_length() }
90    }
91
92    pub fn logical_offset(&self) -> u64 {
93        self.logical_offset
94    }
95
96    pub fn logical_range(&self) -> Range<u64> {
97        // SAFETY: We verified logical_offset plus device_range length won't overflow in Self::new.
98        unsafe { self.logical_offset..self.logical_offset.unchecked_add(self.length()) }
99    }
100
101    pub fn device_range(&self) -> &Range<u64> {
102        &self.device_range
103    }
104}
105
106#[derive(Debug)]
107pub enum DataObjectState {
108    Standard(AllocatedRanges),
109    VerityStarted,
110    VerityPending(FsverityStateInner),
111    Verity(FsverityStateInner),
112}
113
114#[derive(Debug)]
115pub struct FsverityStateInner {
116    root_digest: RootDigest,
117    salt: Vec<u8>,
118    // TODO(b/309656632): This should store the entire merkle tree and not just the leaf nodes.
119    // Potentially store a pager-backed vmo instead of passing around a boxed array.
120    merkle_tree: Box<[u8]>,
121}
122
123#[derive(Debug, Default)]
124pub struct OverwriteOptions {
125    // If false, then all the extents for the overwrite range must have been preallocated using
126    // preallocate_range or from existing writes.
127    pub allow_allocations: bool,
128    pub barrier_on_first_write: bool,
129}
130
131impl FsverityStateInner {
132    pub fn new(root_digest: RootDigest, salt: Vec<u8>, merkle_tree: Box<[u8]>) -> Self {
133        FsverityStateInner { root_digest, salt, merkle_tree }
134    }
135
136    fn get_hasher_for_block_size(&self, block_size: usize) -> FsVerityHasher {
137        match self.root_digest {
138            RootDigest::Sha256(_) => {
139                FsVerityHasher::Sha256(FsVerityHasherOptions::new(self.salt.clone(), block_size))
140            }
141            RootDigest::Sha512(_) => {
142                FsVerityHasher::Sha512(FsVerityHasherOptions::new(self.salt.clone(), block_size))
143            }
144        }
145    }
146
147    fn from_ptr_slice(
148        data: PtrByteSlice<'_>,
149        block_size: usize,
150    ) -> Result<(Self, FsVerityHasher), Error> {
151        let descriptor = FsVerityDescriptor::new(data, block_size)
152            .map_err(|e| anyhow!(FxfsError::IntegrityError).context(e))?;
153
154        let root_digest = match descriptor.digest_algorithm() {
155            fio::HashAlgorithm::Sha256 => {
156                RootDigest::Sha256(descriptor.root_digest().try_into().unwrap())
157            }
158            fio::HashAlgorithm::Sha512 => RootDigest::Sha512(descriptor.root_digest().to_vec()),
159            _ => return Err(anyhow!(FxfsError::NotSupported).context("Unsupported hash algorithm")),
160        };
161        let hasher = descriptor.hasher();
162        let leaves =
163            descriptor.leaf_digests().map_err(|e| anyhow!(FxfsError::IntegrityError).context(e))?;
164
165        Ok((Self::new(root_digest, descriptor.salt().to_vec(), leaves.into_boxed_slice()), hasher))
166    }
167}
168
169impl<S: HandleOwner> Deref for DataObjectHandle<S> {
170    type Target = StoreObjectHandle<S>;
171    fn deref(&self) -> &Self::Target {
172        &self.handle
173    }
174}
175
176impl<S: HandleOwner> DataObjectHandle<S> {
177    pub fn new(
178        owner: Arc<S>,
179        object_id: u64,
180        permanent_keys: bool,
181        attribute_id: AttributeId,
182        size: u64,
183        options: HandleOptions,
184        trace: bool,
185        overwrite_ranges: &[Range<u64>],
186    ) -> Self {
187        Self {
188            handle: StoreObjectHandle::new(owner, object_id, permanent_keys, options, trace),
189            attribute_id,
190            content_size: AtomicU64::new(size),
191            state: Mutex::new(DataObjectState::Standard(AllocatedRanges::new(overwrite_ranges))),
192        }
193    }
194
195    pub fn attribute_id(&self) -> AttributeId {
196        self.attribute_id
197    }
198
199    /// Consumes the `DataObjectHandle` and returns the `StoreObjectHandle` that it contained.
200    pub fn into_store_object_handle(self) -> StoreObjectHandle<S> {
201        self.handle
202    }
203
204    pub fn overwrite_ranges_is_empty(&self) -> bool {
205        match &*self.state.lock() {
206            DataObjectState::Standard(ranges) => ranges.is_empty(),
207            _ => true,
208        }
209    }
210
211    pub fn with_overwrite_ranges<R>(&self, f: impl FnOnce(Option<&AllocatedRanges>) -> R) -> R {
212        let state = self.state.lock();
213        match &*state {
214            DataObjectState::Standard(ranges) => f(Some(ranges)),
215            _ => f(None),
216        }
217    }
218
219    pub fn with_overwrite_ranges_mut<R>(
220        &self,
221        f: impl FnOnce(Option<&mut AllocatedRanges>) -> R,
222    ) -> R {
223        let mut state = self.state.lock();
224        match &mut *state {
225            DataObjectState::Standard(ranges) => f(Some(ranges)),
226            _ => f(None),
227        }
228    }
229
230    pub fn is_verified_file(&self) -> bool {
231        matches!(*self.state.lock(), DataObjectState::Verity(_))
232    }
233
234    /// Sets `self.state` to DataObjectState::VerityStarted. Called at the top of `enable_verity`.
235    /// If another caller has already started but not completed `enable_verity`, returns
236    /// FxfsError::AlreadyBound. If another caller has already completed `enable_verity`, returns
237    /// FxfsError::AlreadyExists.
238    ///
239    /// Note: This is called before `enable_verity` acquires its object transaction lock. Any
240    /// ongoing transaction holding the object lock (such as `allocate`) will proceed to commit
241    /// its mutations to disk while `enable_verity` waits for the lock, and `will_apply_mutation`
242    /// will gracefully discard in-memory range updates for the discarded `AllocatedRanges`.
243    pub fn set_fsverity_state_started(&self) -> Result<(), Error> {
244        let mut state = self.state.lock();
245        match *state {
246            DataObjectState::Standard(_) => {
247                *state = DataObjectState::VerityStarted;
248                Ok(())
249            }
250            DataObjectState::VerityStarted | DataObjectState::VerityPending(_) => {
251                Err(anyhow!(FxfsError::Unavailable))
252            }
253            DataObjectState::Verity(_) => Err(anyhow!(FxfsError::AlreadyExists)),
254        }
255    }
256
257    /// Sets `self.state` to VerityPending. Must be called before `finalize_fsverity_state()`.
258    ///
259    /// # Panics
260    ///
261    /// Panics if the prior state was not `DataObjectState::VerityStarted`.
262    pub fn set_fsverity_state_pending(&self, descriptor: FsverityStateInner) {
263        let mut state = self.state.lock();
264        assert!(matches!(*state, DataObjectState::VerityStarted));
265        *state = DataObjectState::VerityPending(descriptor);
266    }
267
268    /// Sets `self.state` to Verity.
269    ///
270    /// # Panics
271    ///
272    /// Panics if the prior state was not `DataObjectState::VerityPending(_)`.
273    pub fn finalize_fsverity_state(&self) {
274        let mut state = self.state.lock();
275        let old_state =
276            std::mem::replace(&mut *state, DataObjectState::Standard(AllocatedRanges::empty()));
277        match old_state {
278            DataObjectState::VerityPending(inner) => *state = DataObjectState::Verity(inner),
279            _ => panic!("Cannot finalize verity state from {old_state:?}"),
280        }
281    }
282
283    /// Sets `self.state` directly to Verity without going through the entire state machine.
284    /// Used to set `self.state` on open of a verified file. The merkle tree data is
285    /// verified against the root digest here, and will return an error if the tree is not correct.
286    ///
287    /// # Panics
288    ///
289    /// Panics if the prior state was not `DataObjectState::Standard(_)`.
290    pub async fn set_fsverity_state_some(&self, descriptor: FsverityMetadata) -> Result<(), Error> {
291        let (metadata, hasher) = match descriptor {
292            FsverityMetadata::Internal(root_digest, salt) => {
293                let merkle_tree = self
294                    .read_attr(AttributeId::FSVERITY_MERKLE)
295                    .await?
296                    .ok_or_else(|| anyhow!(FxfsError::Inconsistent))?;
297                let metadata = FsverityStateInner { root_digest, salt, merkle_tree };
298                let hasher = metadata.get_hasher_for_block_size(self.block_size() as usize);
299                (metadata, hasher)
300            }
301            FsverityMetadata::F2fs(verity_range) => {
302                let expected_length = verity_range.length()? as usize;
303                let mut buffer = self
304                    .allocate_buffer(expected_length.next_multiple_of(self.block_size() as usize))
305                    .await;
306                ensure!(
307                    expected_length
308                        == self
309                            .handle
310                            .read(AttributeId::FSVERITY_MERKLE, verity_range.start, buffer.as_mut())
311                            .await?,
312                    FxfsError::Inconsistent
313                );
314                let data = buffer.as_ptr_slice().subslice(0..expected_length);
315                FsverityStateInner::from_ptr_slice(data, self.block_size() as usize)?
316            }
317        };
318        // Validate the merkle tree data against the root before applying it.
319        ensure!(metadata.merkle_tree.len() % hasher.hash_size() == 0, FxfsError::Inconsistent);
320        let leaf_chunks = metadata.merkle_tree.chunks_exact(hasher.hash_size());
321
322        let root_hash = match &metadata.root_digest {
323            RootDigest::Sha256(root_hash) => root_hash.as_slice(),
324            RootDigest::Sha512(root_hash) => root_hash.as_slice(),
325        };
326
327        let tree = match hasher {
328            FsVerityHasher::Sha256(_) => {
329                let mut builder = MerkleTreeBuilder::<Sha256Hash>::new(hasher);
330                for leaf in leaf_chunks {
331                    let hash = Sha256Hash::read_from_bytes(leaf).unwrap();
332                    builder.push_data_hash(hash);
333                }
334                builder.finish()
335            }
336            FsVerityHasher::Sha512(_) => {
337                let mut builder = MerkleTreeBuilder::<Sha512Hash>::new(hasher);
338                for leaf in leaf_chunks {
339                    let hash = Sha512Hash::read_from_bytes(leaf).unwrap();
340                    builder.push_data_hash(hash);
341                }
342                builder.finish()
343            }
344        };
345
346        ensure!(root_hash == tree.root(), FxfsError::IntegrityError);
347
348        let mut state = self.state.lock();
349        assert!(matches!(*state, DataObjectState::Standard(_)));
350        *state = DataObjectState::Verity(metadata);
351
352        Ok(())
353    }
354
355    /// Verifies contents of `buffer` against the corresponding hashes in the stored merkle tree.
356    /// `offset` is the logical offset in the file that `buffer` starts at. Fails on non
357    /// fsverity-enabled files.
358    ///
359    /// # Panics
360    ///
361    /// Panics if `offset` is not block-aligned.
362    fn verify_data(&self, mut offset: usize, buffer: PtrByteSlice<'_>) -> Result<(), Error> {
363        let block_size = self.block_size() as usize;
364        assert!(offset % block_size == 0);
365        let state = self.state.lock();
366        match &*state {
367            DataObjectState::Standard(_) => {
368                Err(anyhow!("Tried to verify read on a non verity-enabled file"))
369            }
370            DataObjectState::VerityStarted | DataObjectState::VerityPending(_) => {
371                Err(anyhow!("Enable verity has not yet completed, state: {state:?}"))
372            }
373            DataObjectState::Verity(metadata) => {
374                let hasher = metadata.get_hasher_for_block_size(block_size);
375                let leaf_nodes: Vec<&[u8]> =
376                    metadata.merkle_tree.chunks(hasher.hash_size()).collect();
377                fxfs_trace::duration!("fsverity-verify", "len" => buffer.len());
378                // TODO(b/318880297): Consider parallelizing computation.
379                for chunk in buffer.chunks(block_size) {
380                    // SAFETY: Ideally we wouldn't be creating references here as technically this is
381                    // Rust undefined behaviour, but it's difficult for us to fix and mitigated
382                    // because these pointers end up being passed directly to non-Rust code (Mundane).
383                    let b = unsafe { &*chunk.as_raw_slice_ptr() };
384
385                    ensure!(
386                        hasher.hash_block(b) == leaf_nodes[offset / block_size],
387                        anyhow!(FxfsError::Inconsistent).context("Hash mismatch")
388                    );
389                    offset += block_size;
390                }
391                Ok(())
392            }
393        }
394    }
395
396    /// Extend the file with the given extent.  The only use case for this right now is for files
397    /// that must exist at certain offsets on the device, such as super-blocks.
398    pub async fn extend<'a>(
399        &'a self,
400        transaction: &mut Transaction<'a>,
401        device_range: Range<u64>,
402    ) -> Result<(), Error> {
403        let old_end =
404            round_up(self.txn_get_size(transaction), self.block_size()).ok_or(FxfsError::TooBig)?;
405        let new_size = old_end + device_range.end - device_range.start;
406        self.store().allocator().mark_allocated(
407            transaction,
408            self.store().store_object_id(),
409            device_range.clone(),
410        )?;
411        self.txn_update_size(transaction, new_size, None).await?;
412        let key_id = self.get_key(None).await?.0;
413        transaction.add(
414            self.store().store_object_id,
415            Mutation::merge_object(
416                ObjectKey::extent(self.object_id(), self.attribute_id(), old_end..new_size),
417                ObjectValue::Extent(ExtentValue::new_raw(device_range.start, key_id)),
418            ),
419        );
420        self.update_allocated_size(transaction, device_range.end - device_range.start, 0).await
421    }
422
423    // Returns a new aligned buffer (reading the head and tail blocks if necessary) with a copy of
424    // the data from `buf`.
425    async fn align_buffer(
426        &self,
427        offset: u64,
428        buf: BufferRef<'_>,
429    ) -> Result<(std::ops::Range<u64>, Buffer<'_>), Error> {
430        self.handle.align_buffer(self.attribute_id(), offset, buf).await
431    }
432
433    // Writes potentially unaligned data at `device_offset` and returns checksums if requested. The
434    // data will be encrypted if necessary.
435    // `buf` is mutable as an optimization, since the write may require encryption, we can encrypt
436    // the buffer in-place rather than copying to another buffer if the write is already aligned.
437    async fn write_at(
438        &self,
439        offset: u64,
440        buf: MutableBufferRef<'_>,
441        device_offset: u64,
442    ) -> Result<MaybeChecksums, Error> {
443        self.handle.write_at(self.attribute_id(), offset, buf, None, device_offset).await
444    }
445
446    /// Verifies that the entire range in the file is zeroes, as either uninitialized overwrite
447    /// range, or no extent at all. If a single allocated and written extent is found, this returns
448    /// false.
449    pub async fn check_unwritten_zero(&self, range: Range<u64>) -> Result<bool, Error> {
450        let tree = &self.store().tree();
451        let layer_set = tree.layer_set();
452        let key = Extent(range);
453        let lower_bound = ObjectKey::attribute(
454            self.object_id(),
455            self.attribute_id,
456            AttributeKey::Extent(key.search_key()),
457        );
458        let mut merger = layer_set.merger();
459        let mut iter = merger.query(Query::FullRange(&lower_bound)).await?;
460        while let Some(ItemRef {
461            key:
462                ObjectKey {
463                    object_id,
464                    data: ObjectKeyData::Attribute(attr_id, AttributeKey::Extent(extent_key)),
465                },
466            value: ObjectValue::Extent(value),
467            ..
468        }) = iter.get()
469            && *object_id == self.object_id()
470            && *attr_id == self.attribute_id
471        {
472            if let ExtentValue::Some { mode, .. } = value {
473                if let Some(overlap) = key.overlap(extent_key) {
474                    if let ExtentMode::OverwritePartial(bits) = mode {
475                        let starting_index = (overlap.start - extent_key.start) / self.block_size();
476                        for initialized in bits
477                            .iter()
478                            .skip(starting_index as usize)
479                            .take((overlap.length().unwrap() / self.block_size()) as usize)
480                        {
481                            if initialized {
482                                return Ok(false);
483                            }
484                        }
485                    } else {
486                        return Ok(false);
487                    }
488                } else {
489                    break;
490                }
491            }
492            iter.advance().await?;
493        }
494        Ok(true)
495    }
496
497    /// Zeroes the given range.  The range must be aligned.  Returns the amount of data deallocated.
498    pub async fn zero(
499        &self,
500        transaction: &mut Transaction<'_>,
501        range: Range<u64>,
502    ) -> Result<(), Error> {
503        self.handle.zero(transaction, self.attribute_id(), range).await
504    }
505
506    /// The cached value for `self.fsverity_state` is set either in `open_object` or on
507    /// `enable_verity`. If set, translates `self.fsverity_state.descriptor` into an
508    /// fio::VerificationOptions instance and a root hash. Otherwise, returns None.
509    pub fn get_descriptor(&self) -> Option<(fio::VerificationOptions, Vec<u8>)> {
510        let state = self.state.lock();
511        match &*state {
512            DataObjectState::Verity(metadata) => {
513                let (options, root_hash) = match &metadata.root_digest {
514                    RootDigest::Sha256(root_hash) => (
515                        fio::VerificationOptions {
516                            hash_algorithm: Some(fio::HashAlgorithm::Sha256),
517                            salt: Some(metadata.salt.clone()),
518                            ..Default::default()
519                        },
520                        root_hash.to_vec(),
521                    ),
522                    RootDigest::Sha512(root_hash) => (
523                        fio::VerificationOptions {
524                            hash_algorithm: Some(fio::HashAlgorithm::Sha512),
525                            salt: Some(metadata.salt.clone()),
526                            ..Default::default()
527                        },
528                        root_hash.clone(),
529                    ),
530                };
531                Some((options, root_hash))
532            }
533            _ => None,
534        }
535    }
536
537    async fn build_verity_tree(
538        &self,
539        hasher: FsVerityHasher,
540        hash_alg: fio::HashAlgorithm,
541        salt: &[u8],
542    ) -> Result<(MerkleTree, Vec<u8>), Error> {
543        match hasher {
544            FsVerityHasher::Sha256(_) => {
545                self.build_verity_tree_impl::<Sha256Hash>(hasher, hash_alg, salt).await
546            }
547            FsVerityHasher::Sha512(_) => {
548                self.build_verity_tree_impl::<Sha512Hash>(hasher, hash_alg, salt).await
549            }
550        }
551    }
552
553    async fn build_verity_tree_impl<D: FsVerityHash>(
554        &self,
555        hasher: FsVerityHasher,
556        hash_alg: fio::HashAlgorithm,
557        salt: &[u8],
558    ) -> Result<(MerkleTree, Vec<u8>), Error> {
559        let hash_len = hasher.hash_size();
560        let mut builder = MerkleTreeBuilder::<D>::new(hasher);
561        let mut offset = 0;
562        let size = self.get_size();
563        // TODO(b/314836822): Consider further tuning the buffer size to optimize
564        // performance. Experimentally, most verity-enabled files are <256K.
565        let mut buf = self.allocate_buffer(64 * self.block_size() as usize).await;
566        while offset < size {
567            // TODO(b/314842875): Consider optimizations for sparse files.
568            let read = self.read(offset, buf.as_mut()).await? as u64;
569            assert!(offset + read <= size);
570            let slice = buf.as_ptr_slice().subslice(0..read as usize);
571
572            // SAFETY: Ideally we wouldn't be creating references here as technically this is
573            // Rust undefined behaviour, but it's difficult for us to fix and mitigated
574            // because these pointers end up being passed directly to non-Rust code (Mundane).
575            let chunk = unsafe { &*slice.as_raw_slice_ptr() };
576
577            builder.write(chunk);
578            offset += read;
579        }
580        let tree = builder.finish();
581        // This will include a block for the root layer, which will be used to house the descriptor.
582        let tree_data_len = tree
583            .levels()
584            .iter()
585            .map(|layer| layer.len().next_multiple_of(self.block_size() as usize))
586            .sum();
587        let mut merkle_tree_data = Vec::<u8>::with_capacity(tree_data_len);
588        // Iterating from the top layers down to the leaves.
589        for layer in tree.levels().iter().rev() {
590            // Skip the root layer.
591            if layer.len() <= hash_len {
592                continue;
593            }
594            merkle_tree_data.extend_from_slice(layer);
595            // Pad to the end of the block.
596            let padded_size = merkle_tree_data.len().next_multiple_of(self.block_size() as usize);
597            merkle_tree_data.resize(padded_size, 0);
598        }
599
600        // Zero the last block, then write the descriptor to the start of it.
601        let descriptor_offset = merkle_tree_data.len();
602        merkle_tree_data.resize(descriptor_offset + self.block_size() as usize, 0);
603        let descriptor = FsVerityDescriptorRaw::new(
604            hash_alg,
605            self.block_size(),
606            self.get_size(),
607            tree.root(),
608            salt,
609        )?;
610        descriptor.write_to_slice(&mut merkle_tree_data[descriptor_offset..])?;
611
612        Ok((tree, merkle_tree_data))
613    }
614
615    /// Reads the data attribute and computes a merkle tree from the data. The values of the
616    /// parameters required to build the merkle tree are supplied by `descriptor` (i.e. salt,
617    /// hash_algorithm, etc.) Writes the leaf nodes of the merkle tree to an attribute with id
618    /// `AttributeId::FSVERITY_MERKLE`. Updates the root_hash of the `descriptor` according to the
619    /// computed merkle tree and then replaces the ObjectValue of the data attribute with
620    /// ObjectValue::VerifiedAttribute, which stores the `descriptor` inline.
621    #[trace]
622    pub async fn enable_verity(&self, options: fio::VerificationOptions) -> Result<(), Error> {
623        self.set_fsverity_state_started()?;
624        // If the merkle attribute was tombstoned in the last attempt of `enable_verity`, flushing
625        // the graveyard should process the tombstone before we start rewriting the attribute.
626        if self
627            .store()
628            .tree()
629            .find(&ObjectKey::graveyard_attribute_entry(
630                self.store().graveyard_directory_object_id(),
631                self.object_id(),
632                AttributeId::FSVERITY_MERKLE,
633            ))
634            .await?
635            .is_some()
636        {
637            self.store().filesystem().graveyard().flush().await;
638        }
639        let mut transaction = self.new_transaction().await?;
640        let hash_alg =
641            options.hash_algorithm.ok_or_else(|| anyhow!("No hash algorithm provided"))?;
642        let salt = options.salt.ok_or_else(|| anyhow!("No salt provided"))?;
643        let (root_digest, merkle_tree) = match hash_alg {
644            fio::HashAlgorithm::Sha256 => {
645                let hasher = FsVerityHasher::Sha256(FsVerityHasherOptions::new(
646                    salt.clone(),
647                    self.block_size() as usize,
648                ));
649                let (tree, merkle_tree_data) =
650                    self.build_verity_tree(hasher, hash_alg, &salt).await?;
651                let root: [u8; 32] = tree.root().try_into().unwrap();
652                (RootDigest::Sha256(root), merkle_tree_data)
653            }
654            fio::HashAlgorithm::Sha512 => {
655                let hasher = FsVerityHasher::Sha512(FsVerityHasherOptions::new(
656                    salt.clone(),
657                    self.block_size() as usize,
658                ));
659                let (tree, merkle_tree_data) =
660                    self.build_verity_tree(hasher, hash_alg, &salt).await?;
661                (RootDigest::Sha512(tree.root().to_vec()), merkle_tree_data)
662            }
663            _ => {
664                bail!(
665                    anyhow!(FxfsError::NotSupported)
666                        .context(format!("hash algorithm not supported"))
667                );
668            }
669        };
670        // TODO(b/314194485): Eventually want streaming writes.
671        // The merkle tree attribute should not require trimming because it should not
672        // exist.
673        self.handle
674            .write_new_attr_in_batches(
675                &mut transaction,
676                AttributeId::FSVERITY_MERKLE,
677                &merkle_tree,
678                WRITE_ATTR_BATCH_SIZE,
679            )
680            .await?;
681        if merkle_tree.len() > WRITE_ATTR_BATCH_SIZE {
682            self.store().remove_attribute_from_graveyard(
683                &mut transaction,
684                self.object_id(),
685                AttributeId::FSVERITY_MERKLE,
686            );
687        };
688        let descriptor_decoded =
689            FsVerityDescriptor::new(&merkle_tree[..], self.block_size() as usize)?;
690        let descriptor = FsverityStateInner {
691            root_digest,
692            salt,
693            merkle_tree: descriptor_decoded.leaf_digests()?.into(),
694        };
695        self.set_fsverity_state_pending(descriptor);
696        transaction.add_with_object(
697            self.store().store_object_id(),
698            Mutation::replace_or_insert_object(
699                ObjectKey::attribute(self.object_id(), AttributeId::DATA, AttributeKey::Attribute),
700                ObjectValue::verified_attribute(
701                    self.get_size(),
702                    FsverityMetadata::F2fs(0..merkle_tree.len() as u64),
703                ),
704            ),
705            AssocObj::Borrowed(self),
706        );
707        transaction.commit().await?;
708        Ok(())
709    }
710
711    /// Pre-allocate disk space for the given logical file range. If any part of the allocation
712    /// range is beyond the end of the file, the file size is updated.
713    pub async fn allocate(&self, range: Range<u64>) -> Result<(), Error> {
714        debug_assert!(range.start < range.end);
715
716        // It's not required that callers of allocate use block aligned ranges, but we need to make
717        // the extents block aligned. Luckily, fallocate in posix is allowed to allocate more than
718        // what was asked for for block alignment purposes. We just need to make sure that the size
719        // of the file is still the non-block-aligned end of the range if the size was changed.
720        let mut new_range = range.clone();
721        new_range.start = round_down(new_range.start, self.block_size());
722        // NB: FxfsError::TooBig turns into EFBIG when passed through starnix, which is the
723        // required error code when the requested range is larger than the file size.
724        new_range.end = round_up(new_range.end, self.block_size()).ok_or(FxfsError::TooBig)?;
725
726        let mut transaction = self.new_transaction().await?;
727        // It's safe to check state after acquiring the transaction lock. Note that `enable_verity`
728        // calls `set_fsverity_state_started` before creating its own transaction, which could
729        // transition `self.state` to `VerityStarted` concurrently while this transaction is held.
730        // If that happens, `enable_verity`'s subsequent `new_transaction` call will block until
731        // this allocation transaction finishes, and `will_apply_mutation` will gracefully discard
732        // updates to the in-memory allocated ranges.
733        {
734            let state = self.state.lock();
735            match &*state {
736                DataObjectState::Standard(_) => {}
737                _ => bail!(
738                    anyhow!(FxfsError::AccessDenied).context("Cannot allocate on verity file")
739                ),
740            }
741        }
742        let mut to_allocate = Vec::new();
743        let mut to_switch = Vec::new();
744        let key_id = self.get_key(None).await?.0;
745
746        {
747            let tree = &self.store().tree;
748            let layer_set = tree.layer_set();
749            let offset_key = ObjectKey::attribute(
750                self.object_id(),
751                self.attribute_id(),
752                AttributeKey::Extent(Extent::search_key_from_offset(new_range.start)),
753            );
754            let mut merger = layer_set.merger();
755            let mut iter = merger.query(Query::FullRange(&offset_key)).await?;
756
757            loop {
758                match iter.get() {
759                    Some(ItemRef {
760                        key:
761                            ObjectKey {
762                                object_id,
763                                data:
764                                    ObjectKeyData::Attribute(
765                                        attribute_id,
766                                        AttributeKey::Extent(extent_key),
767                                    ),
768                            },
769                        value: ObjectValue::Extent(extent_value),
770                        ..
771                    }) if *object_id == self.object_id()
772                        && *attribute_id == self.attribute_id() =>
773                    {
774                        // If the start of this extent is beyond the end of the range we are
775                        // allocating, we don't have any more work to do.
776                        if new_range.end <= extent_key.start {
777                            break;
778                        }
779                        // Add any prefix we might need to allocate.
780                        if new_range.start < extent_key.start {
781                            to_allocate.push(new_range.start..extent_key.start);
782                            new_range.start = extent_key.start;
783                        }
784                        let device_offset = match extent_value {
785                            ExtentValue::None => {
786                                // If the extent value is None, it indicates a deleted extent. In
787                                // that case, we just skip it entirely. By keeping the new_range
788                                // where it is, this section will get included in the new
789                                // allocations.
790                                iter.advance().await?;
791                                continue;
792                            }
793                            ExtentValue::Some { mode: ExtentMode::OverwritePartial(_), .. }
794                            | ExtentValue::Some { mode: ExtentMode::Overwrite, .. } => {
795                                // If this extent is already in overwrite mode, we can skip it.
796                                if extent_key.end < new_range.end {
797                                    new_range.start = extent_key.end;
798                                    iter.advance().await?;
799                                    continue;
800                                } else {
801                                    new_range.start = new_range.end;
802                                    break;
803                                }
804                            }
805                            ExtentValue::Some { device_offset, .. } => *device_offset,
806                        };
807
808                        // Figure out how we have to break up the ranges.
809                        let device_offset = device_offset + (new_range.start - extent_key.start);
810                        if extent_key.end < new_range.end {
811                            to_switch.push((new_range.start..extent_key.end, device_offset));
812                            new_range.start = extent_key.end;
813                        } else {
814                            to_switch.push((new_range.start..new_range.end, device_offset));
815                            new_range.start = new_range.end;
816                            break;
817                        }
818                    }
819                    // The records are sorted so if we find something that isn't an extent or
820                    // doesn't match the object id then there are no more extent records for this
821                    // object.
822                    _ => break,
823                }
824                iter.advance().await?;
825            }
826        }
827
828        if new_range.start < new_range.end {
829            to_allocate.push(new_range.clone());
830        }
831
832        // We can update the size in the first transaction because even if subsequent transactions
833        // don't get replayed, the data between the current and new end of the file will be zero
834        // (either sparse zero or allocated zero). On the other hand, if we don't update the size
835        // in the first transaction, overwrite extents may be written past the end of the file
836        // which is an fsck error.
837        //
838        // The potential new size needs to be the non-block-aligned range end - we round up to the
839        // nearest block size for the actual allocation, but shouldn't do that for the file size.
840        let new_size = std::cmp::max(range.end, self.get_size());
841        // Make sure the mutation that flips the has_overwrite_extents advisory flag is in the
842        // first transaction, in case we split transactions. This makes it okay to only replay the
843        // first transaction if power loss occurs - the file will be in an unusual state, but not
844        // an invalid one, if only part of the allocate goes through.
845        transaction.add_with_object(
846            self.store().store_object_id(),
847            Mutation::replace_or_insert_object(
848                ObjectKey::attribute(
849                    self.object_id(),
850                    self.attribute_id(),
851                    AttributeKey::Attribute,
852                ),
853                ObjectValue::Attribute { size: new_size, has_overwrite_extents: true },
854            ),
855            AssocObj::Borrowed(self),
856        );
857
858        // The maximum number of mutations we are going to allow per transaction in allocate. This
859        // is probably quite a bit lower than the actual limit, but it should be large enough to
860        // handle most non-edge-case versions of allocate without splitting the transaction.
861        const MAX_TRANSACTION_SIZE: usize = 256;
862        for (switch_range, device_offset) in to_switch {
863            transaction.add_with_object(
864                self.store().store_object_id(),
865                Mutation::merge_object(
866                    ObjectKey::extent(self.object_id(), self.attribute_id(), switch_range),
867                    ObjectValue::Extent(ExtentValue::initialized_overwrite_extent(
868                        device_offset,
869                        key_id,
870                    )),
871                ),
872                AssocObj::Borrowed(self),
873            );
874            if transaction.mutations().len() >= MAX_TRANSACTION_SIZE {
875                transaction.commit_and_continue().await?;
876            }
877        }
878
879        let mut allocated = 0;
880        let allocator = self.store().allocator();
881        for mut allocate_range in to_allocate {
882            while allocate_range.start < allocate_range.end {
883                let device_range = allocator
884                    .allocate(
885                        &mut transaction,
886                        self.store().store_object_id(),
887                        allocate_range.end - allocate_range.start,
888                    )
889                    .await
890                    .context("allocation failed")?;
891                let device_range_len = device_range.end - device_range.start;
892
893                transaction.add_with_object(
894                    self.store().store_object_id(),
895                    Mutation::merge_object(
896                        ObjectKey::extent(
897                            self.object_id(),
898                            self.attribute_id(),
899                            allocate_range.start..allocate_range.start + device_range_len,
900                        ),
901                        ObjectValue::Extent(ExtentValue::blank_overwrite_extent(
902                            device_range.start,
903                            (device_range_len / self.block_size()) as usize,
904                            key_id,
905                        )),
906                    ),
907                    AssocObj::Borrowed(self),
908                );
909
910                allocate_range.start += device_range_len;
911                allocated += device_range_len;
912
913                if transaction.mutations().len() >= MAX_TRANSACTION_SIZE {
914                    self.update_allocated_size(&mut transaction, allocated, 0).await?;
915                    transaction.commit_and_continue().await?;
916                    allocated = 0;
917                }
918            }
919        }
920
921        self.update_allocated_size(&mut transaction, allocated, 0).await?;
922        transaction.commit().await?;
923
924        Ok(())
925    }
926
927    /// Return information on a contiguous set of extents that has the same allocation status,
928    /// starting from `start_offset`. The information returned is if this set of extents are marked
929    /// allocated/not allocated and also the size of this set (in bytes). This is used when
930    /// querying slices for volumes.
931    /// This function expects `start_offset` to be aligned to block size
932    pub async fn is_allocated(&self, start_offset: u64) -> Result<(bool, u64), Error> {
933        let block_size = self.block_size();
934        assert_eq!(start_offset % block_size, 0);
935
936        if start_offset > self.get_size() {
937            bail!(FxfsError::OutOfRange)
938        }
939
940        if start_offset == self.get_size() {
941            return Ok((false, 0));
942        }
943
944        let tree = &self.store().tree;
945        let layer_set = tree.layer_set();
946        let offset_key = ObjectKey::attribute(
947            self.object_id(),
948            self.attribute_id(),
949            AttributeKey::Extent(Extent::search_key_from_offset(start_offset)),
950        );
951        let mut merger = layer_set.merger();
952        let mut iter = merger.query(Query::FullRange(&offset_key)).await?;
953
954        let mut allocated = None;
955        let mut end = start_offset;
956
957        loop {
958            // Iterate through the extents, each time setting `end` as the end of the previous
959            // extent
960            match iter.get() {
961                Some(ItemRef {
962                    key:
963                        ObjectKey {
964                            object_id,
965                            data:
966                                ObjectKeyData::Attribute(attribute_id, AttributeKey::Extent(extent_key)),
967                        },
968                    value: ObjectValue::Extent(extent_value),
969                    ..
970                }) => {
971                    // Equivalent of getting no extents back
972                    if *object_id != self.object_id() || *attribute_id != self.attribute_id() {
973                        if allocated == Some(false) || allocated.is_none() {
974                            end = self.get_size();
975                            allocated = Some(false);
976                        }
977                        break;
978                    }
979                    ensure!(extent_key.is_aligned(block_size), FxfsError::Inconsistent);
980                    if extent_key.start > end {
981                        // If a previous extent has already been visited and we are tracking an
982                        // allocated set, we are only interested in an extent where the range of the
983                        // current extent follows immediately after the previous one.
984                        if allocated == Some(true) {
985                            break;
986                        } else {
987                            // The gap between the previous `end` and this extent is not allocated
988                            end = extent_key.start;
989                            allocated = Some(false);
990                            // Continue this iteration, except now the `end` is set to the end of
991                            // the "previous" extent which is this gap between the start_offset
992                            // and the current extent
993                        }
994                    }
995
996                    // We can assume that from here, the `end` points to the end of a previous
997                    // extent.
998                    match extent_value {
999                        // The current extent has been allocated
1000                        ExtentValue::Some { .. } => {
1001                            // Stop searching if previous extent was marked deleted
1002                            if allocated == Some(false) {
1003                                break;
1004                            }
1005                            allocated = Some(true);
1006                        }
1007                        // This extent has been marked deleted
1008                        ExtentValue::None => {
1009                            // Stop searching if previous extent was marked allocated
1010                            if allocated == Some(true) {
1011                                break;
1012                            }
1013                            allocated = Some(false);
1014                        }
1015                    }
1016                    end = extent_key.end;
1017                }
1018                // This occurs when there are no extents left
1019                None => {
1020                    if allocated == Some(false) || allocated.is_none() {
1021                        end = self.get_size();
1022                        allocated = Some(false);
1023                    }
1024                    // Otherwise, we were monitoring extents that were allocated, so just exit.
1025                    break;
1026                }
1027                // Non-extent records (Object, Child, GraveyardEntry) are ignored.
1028                Some(_) => {}
1029            }
1030            iter.advance().await?;
1031        }
1032
1033        Ok((allocated.unwrap(), end - start_offset))
1034    }
1035
1036    pub async fn txn_write<'a>(
1037        &'a self,
1038        transaction: &mut Transaction<'a>,
1039        offset: u64,
1040        buf: BufferRef<'_>,
1041    ) -> Result<(), Error> {
1042        if buf.is_empty() {
1043            return Ok(());
1044        }
1045        let (aligned, mut transfer_buf) = self.align_buffer(offset, buf).await?;
1046        self.multi_write(
1047            transaction,
1048            self.attribute_id(),
1049            std::slice::from_ref(&aligned),
1050            transfer_buf.as_mut(),
1051        )
1052        .await?;
1053        if offset + buf.len() as u64 > self.txn_get_size(transaction) {
1054            self.txn_update_size(transaction, offset + buf.len() as u64, None).await?;
1055        }
1056        Ok(())
1057    }
1058
1059    // Writes to multiple ranges with data provided in `buf`.  The buffer can be modified in place
1060    // if encryption takes place.  The ranges must all be aligned and no change to content size is
1061    // applied; the caller is responsible for updating size if required.
1062    pub async fn multi_write<'a>(
1063        &'a self,
1064        transaction: &mut Transaction<'a>,
1065        attribute_id: AttributeId,
1066        ranges: &[Range<u64>],
1067        buf: MutableBufferRef<'_>,
1068    ) -> Result<(), Error> {
1069        self.handle.multi_write(transaction, attribute_id, None, ranges, buf).await
1070    }
1071
1072    // `buf` is mutable as an optimization, since the write may require encryption, we can
1073    // encrypt the buffer in-place rather than copying to another buffer if the write is
1074    // already aligned.
1075    //
1076    // Note: in the event of power failure during an overwrite() call, it is possible that
1077    // old data (which hasn't been overwritten with new bytes yet) may be exposed to the user.
1078    // Since the old data should be encrypted, it is probably safe to expose, although not ideal.
1079    pub async fn overwrite(
1080        &self,
1081        mut offset: u64,
1082        mut buf: MutableBufferRef<'_>,
1083        options: OverwriteOptions,
1084    ) -> Result<(), Error> {
1085        ensure!((buf.len() as u32) % self.store().device.block_size() == 0, FxfsError::InvalidArgs);
1086        let end = offset + buf.len() as u64;
1087
1088        let key_id = self.get_key(None).await?.0;
1089
1090        // The transaction only ends up being used if allow_allocations is true
1091        let mut transaction =
1092            if options.allow_allocations { Some(self.new_transaction().await?) } else { None };
1093
1094        // We build up a list of writes to perform later
1095        let writes = FuturesUnordered::new();
1096
1097        if options.barrier_on_first_write {
1098            self.store().device.barrier();
1099        }
1100
1101        // We create a new scope here, so that the merger iterator will get dropped before we try to
1102        // commit our transaction. Otherwise the transaction commit would block.
1103        {
1104            let store = self.store();
1105            let store_object_id = store.store_object_id;
1106            let allocator = store.allocator();
1107            let tree = &store.tree;
1108            let layer_set = tree.layer_set();
1109            let mut merger = layer_set.merger();
1110            let mut iter = merger
1111                .query(Query::FullRange(&ObjectKey::attribute(
1112                    self.object_id(),
1113                    self.attribute_id(),
1114                    AttributeKey::Extent(Extent::search_key_from_offset(offset)),
1115                )))
1116                .await?;
1117            let block_size = self.block_size();
1118
1119            loop {
1120                let (device_offset, bytes_to_write, should_advance) = match iter.get() {
1121                    Some(ItemRef {
1122                        key:
1123                            ObjectKey {
1124                                object_id,
1125                                data:
1126                                    ObjectKeyData::Attribute(attribute_id, AttributeKey::Extent(extent)),
1127                            },
1128                        value: ObjectValue::Extent(ExtentValue::Some { .. }),
1129                        ..
1130                    }) if *object_id == self.object_id()
1131                        && *attribute_id == self.attribute_id()
1132                        && extent.end == offset =>
1133                    {
1134                        iter.advance().await?;
1135                        continue;
1136                    }
1137                    Some(ItemRef {
1138                        key:
1139                            ObjectKey {
1140                                object_id,
1141                                data:
1142                                    ObjectKeyData::Attribute(attribute_id, AttributeKey::Extent(extent)),
1143                            },
1144                        value,
1145                        ..
1146                    }) if *object_id == self.object_id()
1147                        && *attribute_id == self.attribute_id()
1148                        && extent.start <= offset =>
1149                    {
1150                        match value {
1151                            ObjectValue::Extent(ExtentValue::Some {
1152                                device_offset,
1153                                mode: ExtentMode::Raw,
1154                                ..
1155                            }) => {
1156                                ensure!(
1157                                    extent.is_aligned(block_size)
1158                                        && device_offset % block_size == 0,
1159                                    FxfsError::Inconsistent
1160                                );
1161                                let offset_within_extent = offset - extent.start;
1162                                let remaining_length_of_extent = (extent
1163                                    .end
1164                                    .checked_sub(offset)
1165                                    .ok_or(FxfsError::Inconsistent)?)
1166                                    as usize;
1167                                // Yields (device_offset, bytes_to_write, should_advance)
1168                                (
1169                                    device_offset + offset_within_extent,
1170                                    min(buf.len(), remaining_length_of_extent),
1171                                    true,
1172                                )
1173                            }
1174                            ObjectValue::Extent(ExtentValue::Some { .. }) => {
1175                                // TODO(https://fxbug.dev/42066056): Maybe we should create
1176                                // a new extent without checksums?
1177                                bail!(
1178                                    "extent from ({},{}) which overlaps offset \
1179                                        {} has the wrong extent mode",
1180                                    extent.start,
1181                                    extent.end,
1182                                    offset
1183                                )
1184                            }
1185                            _ => {
1186                                bail!(
1187                                    "overwrite failed: extent overlapping offset {} has \
1188                                      unexpected ObjectValue",
1189                                    offset
1190                                )
1191                            }
1192                        }
1193                    }
1194                    maybe_item_ref => {
1195                        if let Some(transaction) = transaction.as_mut() {
1196                            assert_eq!(options.allow_allocations, true);
1197                            assert_eq!(offset % self.block_size(), 0);
1198
1199                            // We are going to make a new extent, but let's check if there is an
1200                            // extent after us. If there is an extent after us, then we don't want
1201                            // our new extent to bump into it...
1202                            let mut bytes_to_allocate =
1203                                round_up(buf.len() as u64, self.block_size())
1204                                    .ok_or(FxfsError::TooBig)?;
1205                            if let Some(ItemRef {
1206                                key:
1207                                    ObjectKey {
1208                                        object_id,
1209                                        data:
1210                                            ObjectKeyData::Attribute(
1211                                                attribute_id,
1212                                                AttributeKey::Extent(extent),
1213                                            ),
1214                                    },
1215                                ..
1216                            }) = maybe_item_ref
1217                            {
1218                                if *object_id == self.object_id()
1219                                    && *attribute_id == self.attribute_id()
1220                                    && offset < extent.start
1221                                {
1222                                    let bytes_until_next_extent = extent.start - offset;
1223                                    bytes_to_allocate =
1224                                        min(bytes_to_allocate, bytes_until_next_extent);
1225                                }
1226                            }
1227
1228                            let device_range = allocator
1229                                .allocate(transaction, store_object_id, bytes_to_allocate)
1230                                .await?;
1231                            let device_range_len = device_range.end - device_range.start;
1232                            transaction.add(
1233                                store_object_id,
1234                                Mutation::insert_object(
1235                                    ObjectKey::extent(
1236                                        self.object_id(),
1237                                        self.attribute_id(),
1238                                        offset..offset + device_range_len,
1239                                    ),
1240                                    ObjectValue::Extent(ExtentValue::new_raw(
1241                                        device_range.start,
1242                                        key_id,
1243                                    )),
1244                                ),
1245                            );
1246
1247                            self.update_allocated_size(transaction, device_range_len, 0).await?;
1248
1249                            // Yields (device_offset, bytes_to_write, should_advance)
1250                            (device_range.start, min(buf.len(), device_range_len as usize), false)
1251                        } else {
1252                            bail!(
1253                                "no extent overlapping offset {}, \
1254                                and new allocations are not allowed",
1255                                offset
1256                            )
1257                        }
1258                    }
1259                };
1260                let (current_buf, remaining_buf) = buf.split_at_mut(bytes_to_write);
1261                writes.push(self.write_at(offset, current_buf, device_offset));
1262                if remaining_buf.len() == 0 {
1263                    break;
1264                } else {
1265                    buf = remaining_buf;
1266                    offset += bytes_to_write as u64;
1267                    if should_advance {
1268                        iter.advance().await?;
1269                    }
1270                }
1271            }
1272        }
1273
1274        self.store().logical_write_ops.fetch_add(1, Ordering::Relaxed);
1275        // The checksums are being ignored here, but we don't need to know them
1276        writes.try_collect::<Vec<MaybeChecksums>>().await?;
1277
1278        if let Some(mut transaction) = transaction {
1279            assert_eq!(options.allow_allocations, true);
1280            if !transaction.is_empty() {
1281                if end > self.get_size() {
1282                    self.grow(&mut transaction, self.get_size(), end).await?;
1283                }
1284                transaction.commit().await?;
1285            }
1286        }
1287
1288        Ok(())
1289    }
1290
1291    // Within a transaction, the size of the object might have changed, so get the size from there
1292    // if it exists, otherwise, fall back on the cached size.
1293    fn txn_get_size(&self, transaction: &Transaction<'_>) -> u64 {
1294        transaction
1295            .get_object_mutation(
1296                self.store().store_object_id,
1297                ObjectKey::attribute(
1298                    self.object_id(),
1299                    self.attribute_id(),
1300                    AttributeKey::Attribute,
1301                ),
1302            )
1303            .and_then(|m| {
1304                if let ObjectItem { value: ObjectValue::Attribute { size, .. }, .. } = m.item {
1305                    Some(size)
1306                } else {
1307                    None
1308                }
1309            })
1310            .unwrap_or_else(|| self.get_size())
1311    }
1312
1313    pub async fn txn_update_size<'a>(
1314        &'a self,
1315        transaction: &mut Transaction<'a>,
1316        new_size: u64,
1317        // Allow callers to update the has_overwrite_extents metadata if they want. If this is
1318        // Some it is set to the value, if None it is left unchanged.
1319        update_has_overwrite_extents: Option<bool>,
1320    ) -> Result<(), Error> {
1321        let key =
1322            ObjectKey::attribute(self.object_id(), self.attribute_id(), AttributeKey::Attribute);
1323        let mut mutation = if let Some(mutation) =
1324            transaction.get_object_mutation(self.store().store_object_id(), key.clone())
1325        {
1326            mutation.clone()
1327        } else {
1328            ObjectStoreMutation {
1329                item: self.store().tree().find(&key).await?.ok_or(FxfsError::NotFound)?,
1330                op: Operation::ReplaceOrInsert,
1331            }
1332        };
1333        if let ObjectValue::Attribute { size, has_overwrite_extents } = &mut mutation.item.value {
1334            *size = new_size;
1335            if let Some(update_has_overwrite_extents) = update_has_overwrite_extents {
1336                *has_overwrite_extents = update_has_overwrite_extents;
1337            }
1338        } else {
1339            bail!(anyhow!(FxfsError::Inconsistent).context("Unexpected object value"));
1340        }
1341        transaction.add_with_object(
1342            self.store().store_object_id(),
1343            Mutation::ObjectStore(mutation),
1344            AssocObj::Borrowed(self),
1345        );
1346        Ok(())
1347    }
1348
1349    async fn update_allocated_size(
1350        &self,
1351        transaction: &mut Transaction<'_>,
1352        allocated: u64,
1353        deallocated: u64,
1354    ) -> Result<(), Error> {
1355        self.handle.update_allocated_size(transaction, allocated, deallocated).await
1356    }
1357
1358    pub fn truncate_overwrite_ranges(&self, size: u64) -> Result<Option<bool>, Error> {
1359        let cutoff = round_up(size, self.block_size()).ok_or(FxfsError::TooBig)?;
1360        if self.with_overwrite_ranges_mut(|ranges| ranges.map_or(false, |r| r.truncate(cutoff))) {
1361            // This returns true if there were ranges, but this truncate removed them all, which
1362            // indicates that we need to flip the has_overwrite_extents metadata flag to false.
1363            Ok(Some(false))
1364        } else {
1365            Ok(None)
1366        }
1367    }
1368
1369    pub async fn shrink<'a>(
1370        &'a self,
1371        transaction: &mut Transaction<'a>,
1372        size: u64,
1373        update_has_overwrite_extents: Option<bool>,
1374    ) -> Result<NeedsTrim, Error> {
1375        let needs_trim = self.handle.shrink(transaction, self.attribute_id(), size).await?;
1376        self.txn_update_size(transaction, size, update_has_overwrite_extents).await?;
1377        Ok(needs_trim)
1378    }
1379
1380    pub async fn grow<'a>(
1381        &'a self,
1382        transaction: &mut Transaction<'a>,
1383        old_size: u64,
1384        size: u64,
1385    ) -> Result<(), Error> {
1386        // Before growing the file, we must make sure that a previous trim has completed.
1387        let store = self.store();
1388        while matches!(
1389            store
1390                .trim_some(
1391                    transaction,
1392                    self.object_id(),
1393                    self.attribute_id(),
1394                    TrimMode::FromOffset(old_size)
1395                )
1396                .await?,
1397            TrimResult::Incomplete
1398        ) {
1399            transaction.commit_and_continue().await?;
1400        }
1401        // We might need to zero out the tail of the old last block.
1402        let block_size = self.block_size();
1403        if old_size % block_size != 0 {
1404            let layer_set = store.tree.layer_set();
1405            let mut merger = layer_set.merger();
1406            let aligned_old_size = round_down(old_size, block_size);
1407            let iter = merger
1408                .query(Query::FullRange(&ObjectKey::attribute(
1409                    self.object_id(),
1410                    self.attribute_id(),
1411                    AttributeKey::Extent(Extent::search_key_from_offset(aligned_old_size)),
1412                )))
1413                .await?;
1414            if let Some(ItemRef {
1415                key:
1416                    ObjectKey {
1417                        object_id,
1418                        data:
1419                            ObjectKeyData::Attribute(attribute_id, AttributeKey::Extent(extent_key)),
1420                    },
1421                value: ObjectValue::Extent(ExtentValue::Some { device_offset, key_id, .. }),
1422                ..
1423            }) = iter.get()
1424            {
1425                if *object_id == self.object_id() && *attribute_id == self.attribute_id() {
1426                    let device_offset = device_offset
1427                        .checked_add(aligned_old_size - extent_key.start)
1428                        .ok_or(FxfsError::Inconsistent)?;
1429                    ensure!(device_offset % block_size == 0, FxfsError::Inconsistent);
1430                    let mut buf = self.allocate_buffer(block_size as usize).await;
1431                    // In the case that this extent is in OverwritePartial mode, there is a
1432                    // possibility that the last block is allocated, but not initialized yet, in
1433                    // which case we don't actually need to bother zeroing out the tail. However,
1434                    // it's not strictly incorrect to change uninitialized data, so we skip the
1435                    // check and blindly do it to keep it simpler here.
1436                    self.read_and_decrypt(device_offset, aligned_old_size, buf.as_mut(), *key_id)
1437                        .await?;
1438                    buf.subslice_mut((old_size % block_size) as usize..buf.len()).fill(0);
1439                    self.multi_write(
1440                        transaction,
1441                        *attribute_id,
1442                        &[aligned_old_size..aligned_old_size + block_size],
1443                        buf.as_mut(),
1444                    )
1445                    .await?;
1446                }
1447            }
1448        }
1449        self.txn_update_size(transaction, size, None).await?;
1450        Ok(())
1451    }
1452
1453    /// Attempts to pre-allocate a `file_range` of bytes for this object.
1454    /// Returns a set of device ranges (i.e. potentially multiple extents).
1455    ///
1456    /// It may not be possible to preallocate the entire requested range in one request
1457    /// due to limitations on transaction size. In such cases, we will preallocate as much as
1458    /// we can up to some (arbitrary, internal) limit on transaction size.
1459    ///
1460    /// `file_range.start` is modified to point at the end of the logical range
1461    /// that was preallocated such that repeated calls to `preallocate_range` with new
1462    /// transactions can be used to preallocate ranges of any size.
1463    ///
1464    /// Requested range must be a multiple of block size.
1465    pub async fn preallocate_range<'a>(
1466        &'a self,
1467        transaction: &mut Transaction<'a>,
1468        file_range: &mut Range<u64>,
1469    ) -> Result<Vec<Range<u64>>, Error> {
1470        let block_size = self.block_size();
1471        ensure!(file_range.is_aligned(block_size), FxfsError::InvalidArgs);
1472        ensure!(!self.handle.is_encrypted(), FxfsError::NotSupported);
1473        let mut ranges = Vec::new();
1474        let tree = &self.store().tree;
1475        let layer_set = tree.layer_set();
1476        let mut merger = layer_set.merger();
1477        let mut iter = merger
1478            .query(Query::FullRange(&ObjectKey::attribute(
1479                self.object_id(),
1480                self.attribute_id(),
1481                AttributeKey::Extent(Extent::search_key_from_offset(file_range.start)),
1482            )))
1483            .await?;
1484        let mut allocated = 0;
1485        let key_id = self.get_key(None).await?.0;
1486        'outer: while file_range.start < file_range.end {
1487            let allocate_end = loop {
1488                match iter.get() {
1489                    // Case for allocated extents for the same object that overlap with file_range.
1490                    Some(ItemRef {
1491                        key:
1492                            ObjectKey {
1493                                object_id,
1494                                data:
1495                                    ObjectKeyData::Attribute(attribute_id, AttributeKey::Extent(extent)),
1496                            },
1497                        value: ObjectValue::Extent(ExtentValue::Some { device_offset, .. }),
1498                        ..
1499                    }) if *object_id == self.object_id()
1500                        && *attribute_id == self.attribute_id()
1501                        && extent.start < file_range.end =>
1502                    {
1503                        ensure!(
1504                            extent.is_valid()
1505                                && extent.is_aligned(block_size)
1506                                && device_offset % block_size == 0,
1507                            FxfsError::Inconsistent
1508                        );
1509                        // If the start of the requested file_range overlaps with an existing extent...
1510                        if extent.start <= file_range.start {
1511                            // Record the existing extent and move on.
1512                            let device_range = device_offset
1513                                .checked_add(file_range.start - extent.start)
1514                                .ok_or(FxfsError::Inconsistent)?
1515                                ..device_offset
1516                                    .checked_add(min(extent.end, file_range.end) - extent.start)
1517                                    .ok_or(FxfsError::Inconsistent)?;
1518                            file_range.start += device_range.end - device_range.start;
1519                            ranges.push(device_range);
1520                            if file_range.start >= file_range.end {
1521                                break 'outer;
1522                            }
1523                            iter.advance().await?;
1524                            continue;
1525                        } else {
1526                            // There's nothing allocated between file_range.start and the beginning
1527                            // of this extent.
1528                            break extent.start;
1529                        }
1530                    }
1531                    // Case for deleted extents eclipsed by file_range.
1532                    Some(ItemRef {
1533                        key:
1534                            ObjectKey {
1535                                object_id,
1536                                data:
1537                                    ObjectKeyData::Attribute(attribute_id, AttributeKey::Extent(extent)),
1538                            },
1539                        value: ObjectValue::Extent(ExtentValue::None),
1540                        ..
1541                    }) if *object_id == self.object_id()
1542                        && *attribute_id == self.attribute_id()
1543                        && extent.end < file_range.end =>
1544                    {
1545                        iter.advance().await?;
1546                    }
1547                    _ => {
1548                        // We can just preallocate the rest.
1549                        break file_range.end;
1550                    }
1551                }
1552            };
1553            let device_range = self
1554                .store()
1555                .allocator()
1556                .allocate(
1557                    transaction,
1558                    self.store().store_object_id(),
1559                    allocate_end - file_range.start,
1560                )
1561                .await
1562                .context("Allocation failed")?;
1563            allocated += device_range.end - device_range.start;
1564            let this_file_range =
1565                file_range.start..file_range.start + device_range.end - device_range.start;
1566            file_range.start = this_file_range.end;
1567            transaction.add(
1568                self.store().store_object_id,
1569                Mutation::merge_object(
1570                    ObjectKey::extent(self.object_id(), self.attribute_id(), this_file_range),
1571                    ObjectValue::Extent(ExtentValue::new_raw(device_range.start, key_id)),
1572                ),
1573            );
1574            ranges.push(device_range);
1575            // If we didn't allocate all that we requested, we'll loop around and try again.
1576            // ... unless we have filled the transaction. The caller should check file_range.
1577            if transaction.mutations().len() > TRANSACTION_MUTATION_THRESHOLD {
1578                break;
1579            }
1580        }
1581        // Update the file size if it changed.
1582        if file_range.start > round_up(self.txn_get_size(transaction), block_size).unwrap() {
1583            self.txn_update_size(transaction, file_range.start, None).await?;
1584        }
1585        self.update_allocated_size(transaction, allocated, 0).await?;
1586        Ok(ranges)
1587    }
1588
1589    pub async fn update_attributes<'a>(
1590        &self,
1591        transaction: &mut Transaction<'a>,
1592        node_attributes: Option<&fio::MutableNodeAttributes>,
1593        change_time: Option<Timestamp>,
1594    ) -> Result<(), Error> {
1595        // This codepath is only called by files, whose wrapping key id users cannot directly set
1596        // as per fscrypt.
1597        ensure!(
1598            !matches!(
1599                node_attributes,
1600                Some(fio::MutableNodeAttributes { wrapping_key_id: Some(_), .. })
1601            ),
1602            FxfsError::BadPath
1603        );
1604        self.handle.update_attributes(transaction, node_attributes, change_time).await
1605    }
1606
1607    /// Get the default set of transaction options for this object. This is mostly the overall
1608    /// default, modified by any [`HandleOptions`] held by this handle.
1609    pub fn default_transaction_options<'b>(&self) -> Options<'b> {
1610        self.handle.default_transaction_options()
1611    }
1612
1613    pub async fn new_transaction<'b>(&self) -> Result<Transaction<'b>, Error> {
1614        self.new_transaction_with_options(self.default_transaction_options()).await
1615    }
1616
1617    pub async fn new_transaction_with_options<'b>(
1618        &self,
1619        options: Options<'b>,
1620    ) -> Result<Transaction<'b>, Error> {
1621        self.handle.new_transaction_with_options(self.attribute_id(), options).await
1622    }
1623
1624    /// Flushes the underlying device.  This is expensive and should be used sparingly.
1625    pub async fn flush_device(&self) -> Result<(), Error> {
1626        self.handle.flush_device().await
1627    }
1628
1629    /// Reads an entire attribute.
1630    pub async fn read_attr(&self, attribute_id: AttributeId) -> Result<Option<Box<[u8]>>, Error> {
1631        self.handle.read_attr(attribute_id).await
1632    }
1633
1634    /// Writes an entire attribute.  This *always* uses the volume data key.
1635    pub async fn write_attr(&self, attribute_id: AttributeId, data: &[u8]) -> Result<(), Error> {
1636        // Must be different attribute otherwise cached size gets out of date.
1637        assert_ne!(attribute_id, self.attribute_id());
1638        let store = self.store();
1639        let mut transaction = self.new_transaction().await?;
1640        if self.handle.write_attr(&mut transaction, attribute_id, data).await?.0 {
1641            transaction.commit_and_continue().await?;
1642            while matches!(
1643                store
1644                    .trim_some(
1645                        &mut transaction,
1646                        self.object_id(),
1647                        attribute_id,
1648                        TrimMode::FromOffset(data.len() as u64),
1649                    )
1650                    .await?,
1651                TrimResult::Incomplete
1652            ) {
1653                transaction.commit_and_continue().await?;
1654            }
1655        }
1656        transaction.commit().await?;
1657        Ok(())
1658    }
1659
1660    async fn read_and_decrypt(
1661        &self,
1662        device_offset: u64,
1663        file_offset: u64,
1664        buffer: MutableBufferRef<'_>,
1665        key_id: u64,
1666    ) -> Result<(), Error> {
1667        self.handle
1668            .read_and_decrypt(self.attribute_id, device_offset, file_offset, buffer, key_id)
1669            .await
1670    }
1671
1672    /// Truncates a file to a given size (growing/shrinking as required).
1673    ///
1674    /// Nb: Most code will want to call truncate() instead. This method is used
1675    /// to update the super block -- a case where we must borrow metadata space.
1676    pub async fn truncate_with_options(
1677        &self,
1678        options: Options<'_>,
1679        size: u64,
1680    ) -> Result<(), Error> {
1681        let mut transaction = self.new_transaction_with_options(options).await?;
1682        let old_size = self.get_size();
1683        if size == old_size {
1684            return Ok(());
1685        }
1686        if size < old_size {
1687            let update_has_overwrite_ranges = self.truncate_overwrite_ranges(size)?;
1688            if self.shrink(&mut transaction, size, update_has_overwrite_ranges).await?.0 {
1689                // The file needs to be trimmed.
1690                transaction.commit_and_continue().await?;
1691                let store = self.store();
1692                while matches!(
1693                    store
1694                        .trim_some(
1695                            &mut transaction,
1696                            self.object_id(),
1697                            self.attribute_id(),
1698                            TrimMode::FromOffset(size)
1699                        )
1700                        .await?,
1701                    TrimResult::Incomplete
1702                ) {
1703                    if let Err(error) = transaction.commit_and_continue().await {
1704                        warn!(error:?; "Failed to trim after truncate");
1705                        return Ok(());
1706                    }
1707                }
1708                if let Err(error) = transaction.commit().await {
1709                    warn!(error:?; "Failed to trim after truncate");
1710                }
1711                return Ok(());
1712            }
1713        } else {
1714            self.grow(&mut transaction, old_size, size).await?;
1715        }
1716        transaction.commit().await?;
1717        Ok(())
1718    }
1719
1720    pub async fn get_properties(&self) -> Result<ObjectProperties, Error> {
1721        // We don't take a read guard here since the object properties are contained in a single
1722        // object, which cannot be inconsistent with itself. The LSM tree does not return
1723        // intermediate states for a single object.
1724        let item = self
1725            .store()
1726            .tree
1727            .find(&ObjectKey::object(self.object_id()))
1728            .await?
1729            .expect("Unable to find object record");
1730        match item.value {
1731            ObjectValue::Object {
1732                kind: ObjectKind::File { refs, .. },
1733                attributes:
1734                    ObjectAttributes {
1735                        creation_time,
1736                        modification_time,
1737                        posix_attributes,
1738                        allocated_size,
1739                        access_time,
1740                        change_time,
1741                        ..
1742                    },
1743            } => Ok(ObjectProperties {
1744                refs,
1745                allocated_size,
1746                data_attribute_size: self.get_size(),
1747                creation_time,
1748                modification_time,
1749                access_time,
1750                change_time,
1751                sub_dirs: 0,
1752                posix_attributes,
1753                dir_type: DirType::Normal,
1754            }),
1755            _ => bail!(FxfsError::NotFile),
1756        }
1757    }
1758
1759    // Returns the contents of this object. This object must be < |limit| bytes in size.
1760    pub async fn contents(&self, limit: usize) -> Result<Box<[u8]>, Error> {
1761        let size = self.get_size();
1762        if size > limit as u64 {
1763            bail!("Object too big ({} > {})", size, limit);
1764        }
1765        let mut buf = self.allocate_buffer(size as usize).await;
1766        self.read(0u64, buf.as_mut()).await?;
1767        Ok(buf.to_vec().into_boxed_slice())
1768    }
1769
1770    /// Returns the set of file_offset->extent mappings for this file. The extents will be sorted by
1771    /// their logical offset within the file.
1772    ///
1773    /// *NOTE*: This operation is potentially expensive and should generally be avoided.
1774    pub async fn device_extents(&self) -> Result<Vec<FileExtent>, Error> {
1775        let tree = &self.store().tree;
1776        let layer_set = tree.layer_set();
1777        let mut merger = layer_set.merger();
1778        let stream = self.handle.extent_stream(&mut merger, self.attribute_id()).await?;
1779        let extents: Vec<FileExtent> = stream.try_collect().await?;
1780        Ok(extents)
1781    }
1782}
1783
1784impl<S: HandleOwner> AssociatedObject for DataObjectHandle<S> {
1785    fn will_apply_mutation(&self, mutation: &Mutation, _object_id: u64, _manager: &ObjectManager) {
1786        match mutation {
1787            Mutation::ObjectStore(ObjectStoreMutation {
1788                item: ObjectItem { value: ObjectValue::Attribute { size, .. }, .. },
1789                ..
1790            }) => self.content_size.store(*size, atomic::Ordering::Relaxed),
1791            Mutation::ObjectStore(ObjectStoreMutation {
1792                item: ObjectItem { value: ObjectValue::VerifiedAttribute { size, .. }, .. },
1793                ..
1794            }) => {
1795                debug_assert_eq!(
1796                    self.get_size(),
1797                    *size,
1798                    "size should be set when verity is enabled and must not change"
1799                );
1800                self.finalize_fsverity_state()
1801            }
1802            Mutation::ObjectStore(ObjectStoreMutation {
1803                item:
1804                    ObjectItem {
1805                        key:
1806                            ObjectKey {
1807                                object_id,
1808                                data:
1809                                    ObjectKeyData::Attribute(attr_id, AttributeKey::Extent(extent)),
1810                            },
1811                        value: ObjectValue::Extent(ExtentValue::Some { mode, .. }),
1812                        ..
1813                    },
1814                ..
1815            }) if self.object_id() == *object_id && self.attribute_id() == *attr_id => match mode {
1816                ExtentMode::Overwrite | ExtentMode::OverwritePartial(_) => {
1817                    // If `enable_verity` transitioned state to `VerityStarted` concurrently while a
1818                    // transaction was in progress, `with_overwrite_ranges_mut` will return `None`
1819                    // and safely discard in-memory range updates, since verity files do not track
1820                    // overwrite ranges.
1821                    self.with_overwrite_ranges_mut(|ranges| {
1822                        if let Some(ranges) = ranges {
1823                            ranges.apply_range(extent.clone().into());
1824                        }
1825                    });
1826                }
1827                ExtentMode::Raw | ExtentMode::Cow(_) => (),
1828            },
1829            _ => {}
1830        }
1831    }
1832}
1833
1834impl<S: HandleOwner> ObjectHandle for DataObjectHandle<S> {
1835    fn set_trace(&self, v: bool) {
1836        self.handle.set_trace(v)
1837    }
1838
1839    fn object_id(&self) -> u64 {
1840        self.handle.object_id()
1841    }
1842
1843    fn allocate_buffer(&self, size: usize) -> BufferFuture<'_> {
1844        self.handle.allocate_buffer(size)
1845    }
1846
1847    fn block_size(&self) -> u64 {
1848        self.handle.block_size()
1849    }
1850}
1851
1852#[async_trait]
1853impl<S: HandleOwner> ReadObjectHandle for DataObjectHandle<S> {
1854    async fn read(&self, offset: u64, mut buf: MutableBufferRef<'_>) -> Result<usize, Error> {
1855        let fs = self.store().filesystem();
1856        let guard = fs
1857            .lock_manager()
1858            .read_lock(lock_keys![LockKey::object_attribute(
1859                self.store().store_object_id,
1860                self.object_id(),
1861                self.attribute_id(),
1862            )])
1863            .await;
1864
1865        let size = self.get_size();
1866        if offset >= size {
1867            return Ok(0);
1868        }
1869        let length = min(buf.len() as u64, size - offset) as usize;
1870        buf = buf.subslice_mut(0..length);
1871        self.handle.read_unchecked(self.attribute_id(), offset, buf.reborrow(), &guard).await?;
1872        if self.is_verified_file() {
1873            self.verify_data(offset as usize, buf.as_ptr_slice())?;
1874        }
1875        Ok(length)
1876    }
1877
1878    fn get_size(&self) -> u64 {
1879        self.content_size.load(atomic::Ordering::Relaxed)
1880    }
1881}
1882
1883impl<S: HandleOwner> WriteObjectHandle for DataObjectHandle<S> {
1884    async fn write_or_append(&self, offset: Option<u64>, buf: BufferRef<'_>) -> Result<u64, Error> {
1885        let offset = offset.unwrap_or_else(|| self.get_size());
1886        let mut transaction = self.new_transaction().await?;
1887        self.txn_write(&mut transaction, offset, buf).await?;
1888        let new_size = self.txn_get_size(&transaction);
1889        transaction.commit().await?;
1890        Ok(new_size)
1891    }
1892
1893    async fn truncate(&self, size: u64) -> Result<(), Error> {
1894        self.truncate_with_options(self.default_transaction_options(), size).await
1895    }
1896
1897    async fn flush(&self) -> Result<(), Error> {
1898        Ok(())
1899    }
1900}
1901
1902/// Like object_handle::Writer, but allows custom transaction options to be set, and makes every
1903/// write go directly to the handle in a transaction.
1904pub struct DirectWriter<'a, S: HandleOwner> {
1905    handle: &'a DataObjectHandle<S>,
1906    options: transaction::Options<'a>,
1907    buffer: Buffer<'a>,
1908    offset: u64,
1909    buf_offset: usize,
1910}
1911
1912const BUFFER_SIZE: usize = 1_048_576;
1913
1914impl<S: HandleOwner> Drop for DirectWriter<'_, S> {
1915    fn drop(&mut self) {
1916        if self.buf_offset != 0 {
1917            warn!("DirectWriter: dropping data, did you forget to call complete?");
1918        }
1919    }
1920}
1921
1922impl<'a, S: HandleOwner> DirectWriter<'a, S> {
1923    pub async fn new(
1924        handle: &'a DataObjectHandle<S>,
1925        options: transaction::Options<'a>,
1926    ) -> DirectWriter<'a, S> {
1927        Self {
1928            handle,
1929            options,
1930            buffer: handle.allocate_buffer(BUFFER_SIZE).await,
1931            offset: 0,
1932            buf_offset: 0,
1933        }
1934    }
1935
1936    async fn flush(&mut self) -> Result<(), Error> {
1937        let mut transaction = self.handle.new_transaction_with_options(self.options).await?;
1938        self.handle
1939            .txn_write(&mut transaction, self.offset, self.buffer.subslice(0..self.buf_offset))
1940            .await?;
1941        transaction.commit().await?;
1942        self.offset += self.buf_offset as u64;
1943        self.buf_offset = 0;
1944        Ok(())
1945    }
1946}
1947
1948impl<'a, S: HandleOwner> WriteBytes for DirectWriter<'a, S> {
1949    fn block_size(&self) -> u64 {
1950        self.handle.block_size()
1951    }
1952
1953    async fn write_bytes(&mut self, mut buf: &[u8]) -> Result<(), Error> {
1954        while buf.len() > 0 {
1955            let to_do = std::cmp::min(buf.len(), BUFFER_SIZE - self.buf_offset);
1956            self.buffer
1957                .subslice_mut(self.buf_offset..self.buf_offset + to_do)
1958                .copy_from_slice(&buf[..to_do]);
1959            self.buf_offset += to_do;
1960            if self.buf_offset == BUFFER_SIZE {
1961                self.flush().await?;
1962            }
1963            buf = &buf[to_do..];
1964        }
1965        Ok(())
1966    }
1967
1968    async fn complete(mut self) -> Result<u64, Error> {
1969        self.flush().await?;
1970        Ok(self.offset + self.buf_offset as u64)
1971    }
1972
1973    async fn skip(&mut self, amount: u64) -> Result<(), Error> {
1974        if (BUFFER_SIZE - self.buf_offset) as u64 > amount {
1975            self.buffer.subslice_mut(self.buf_offset..self.buf_offset + amount as usize).fill(0);
1976            self.buf_offset += amount as usize;
1977        } else {
1978            self.flush().await?;
1979            self.offset += amount;
1980        }
1981        Ok(())
1982    }
1983}
1984
1985#[cfg(test)]
1986mod tests {
1987    use crate::errors::FxfsError;
1988    use crate::filesystem::{
1989        FxFilesystem, FxFilesystemBuilder, JournalingObject, OpenFxFilesystem, SyncOptions,
1990    };
1991    use crate::fsck::{
1992        FsckOptions, fsck, fsck_volume, fsck_volume_with_options, fsck_with_options,
1993    };
1994    use crate::lsm_tree::Query;
1995    use crate::lsm_tree::types::{ItemRef, LayerIterator};
1996    use crate::object_handle::{
1997        ObjectHandle, ObjectProperties, ReadObjectHandle, WriteObjectHandle,
1998    };
1999    use crate::object_store::data_object_handle::{OverwriteOptions, WRITE_ATTR_BATCH_SIZE};
2000    use crate::object_store::directory::replace_child;
2001    use crate::object_store::object_record::{FsverityMetadata, ObjectKey, ObjectValue, Timestamp};
2002    use crate::object_store::transaction::{Mutation, Options, lock_keys};
2003    use crate::object_store::volume::root_volume;
2004    use crate::object_store::{
2005        AttributeId, AttributeKey, DataObjectHandle, DirType, Directory, Extent, ExtentMode,
2006        ExtentValue, HandleOptions, LockKey, NewChildStoreOptions, ObjectKeyData, ObjectStore,
2007        PosixAttributes, StoreOptions, TRANSACTION_MUTATION_THRESHOLD,
2008    };
2009    use crate::range::RangeExt;
2010    use crate::round::{round_down, round_up};
2011    use assert_matches::assert_matches;
2012    use bit_vec::BitVec;
2013    use fidl_fuchsia_io as fio;
2014    use fsverity_merkle::{FsVerityDescriptor, FsVerityDescriptorRaw};
2015    use fuchsia_async as fasync;
2016    use fuchsia_sync::Mutex;
2017    use futures::FutureExt;
2018    use futures::channel::oneshot::channel;
2019    use futures::stream::{FuturesUnordered, StreamExt};
2020    use fxfs_crypto::{Crypt, EncryptionKey, KeyPurpose};
2021    use fxfs_insecure_crypto::new_insecure_crypt;
2022    use std::ops::Range;
2023    use std::sync::Arc;
2024    use std::time::Duration;
2025    use storage_device::DeviceHolder;
2026    use storage_device::fake_device::FakeDevice;
2027
2028    const TEST_DEVICE_BLOCK_SIZE: u32 = 512;
2029
2030    // Some tests (the preallocate_range ones) currently assume that the data only occupies a single
2031    // device block.
2032    const TEST_DATA_OFFSET: u64 = 5000;
2033    const TEST_DATA: &[u8] = b"hello";
2034    const TEST_OBJECT_SIZE: u64 = 5678;
2035    const TEST_OBJECT_ALLOCATED_SIZE: u64 = 4096;
2036    const TEST_OBJECT_NAME: &str = "foo";
2037
2038    async fn test_filesystem() -> OpenFxFilesystem {
2039        let device = DeviceHolder::new(FakeDevice::new(8192, TEST_DEVICE_BLOCK_SIZE));
2040        FxFilesystem::new_empty(device).await.expect("new_empty failed")
2041    }
2042
2043    async fn create_object_with_key(
2044        fs: Arc<FxFilesystem>,
2045        crypt: Option<&dyn Crypt>,
2046        write_object_test_data: bool,
2047    ) -> DataObjectHandle<ObjectStore> {
2048        let store = fs.root_store();
2049        let object;
2050
2051        let mut transaction = fs
2052            .root_store()
2053            .new_transaction(
2054                lock_keys![LockKey::object(
2055                    store.store_object_id(),
2056                    store.root_directory_object_id()
2057                )],
2058                Options::default(),
2059            )
2060            .await
2061            .expect("new_transaction failed");
2062
2063        object = if let Some(crypt) = crypt {
2064            let object_id = store.get_next_object_id().await.unwrap();
2065            let (key, unwrapped_key) =
2066                crypt.create_key(object_id.get(), KeyPurpose::Data).await.unwrap();
2067            ObjectStore::create_object_with_key(
2068                &store,
2069                &mut transaction,
2070                object_id,
2071                HandleOptions::default(),
2072                EncryptionKey::Fxfs(key),
2073                unwrapped_key,
2074            )
2075            .await
2076            .expect("create_object failed")
2077        } else {
2078            ObjectStore::create_object(&store, &mut transaction, HandleOptions::default(), None)
2079                .await
2080                .expect("create_object failed")
2081        };
2082
2083        let root_directory =
2084            Directory::open(&store, store.root_directory_object_id()).await.expect("open failed");
2085        root_directory
2086            .add_child_file(&mut transaction, TEST_OBJECT_NAME, &object)
2087            .await
2088            .expect("add_child_file failed");
2089
2090        if write_object_test_data {
2091            let align = TEST_DATA_OFFSET as usize % TEST_DEVICE_BLOCK_SIZE as usize;
2092            let mut buf = object.allocate_buffer(align + TEST_DATA.len()).await;
2093            buf.subslice_mut(align..buf.len()).copy_from_slice(TEST_DATA);
2094            object
2095                .txn_write(&mut transaction, TEST_DATA_OFFSET, buf.subslice(align..buf.len()))
2096                .await
2097                .expect("write failed");
2098        }
2099        transaction.commit().await.expect("commit failed");
2100        object.truncate(TEST_OBJECT_SIZE).await.expect("truncate failed");
2101        object
2102    }
2103
2104    async fn test_filesystem_and_object_with_key(
2105        crypt: Option<&dyn Crypt>,
2106        write_object_test_data: bool,
2107    ) -> (OpenFxFilesystem, DataObjectHandle<ObjectStore>) {
2108        let fs = test_filesystem().await;
2109        let object = create_object_with_key(fs.clone(), crypt, write_object_test_data).await;
2110        (fs, object)
2111    }
2112
2113    async fn test_filesystem_and_object() -> (OpenFxFilesystem, DataObjectHandle<ObjectStore>) {
2114        test_filesystem_and_object_with_key(Some(&new_insecure_crypt()), true).await
2115    }
2116
2117    async fn test_filesystem_and_empty_object() -> (OpenFxFilesystem, DataObjectHandle<ObjectStore>)
2118    {
2119        test_filesystem_and_object_with_key(Some(&new_insecure_crypt()), false).await
2120    }
2121
2122    #[fuchsia::test]
2123    async fn test_zero_buf_len_read() {
2124        let (fs, object) = test_filesystem_and_object().await;
2125        let mut buf = object.allocate_buffer(0).await;
2126        assert_eq!(object.read(0u64, buf.as_mut()).await.expect("read failed"), 0);
2127        fs.close().await.expect("Close failed");
2128    }
2129
2130    #[fuchsia::test]
2131    async fn test_beyond_eof_read() {
2132        let (fs, object) = test_filesystem_and_object().await;
2133        let offset = TEST_OBJECT_SIZE as usize - 2;
2134        let align = offset % fs.block_size() as usize;
2135        let len: usize = 2;
2136        let mut buf = object.allocate_buffer(align + len + 1).await;
2137        buf.fill(123u8);
2138        assert_eq!(
2139            object.read((offset - align) as u64, buf.as_mut()).await.expect("read failed"),
2140            align + len
2141        );
2142        assert_eq!(&buf.as_ptr_slice().subslice(align..align + len).to_vec()[..], &vec![0u8; len]);
2143        assert_eq!(
2144            &buf.as_ptr_slice().subslice(align + len..buf.len()).to_vec()[..],
2145            &vec![123u8; buf.len() - align - len]
2146        );
2147        fs.close().await.expect("Close failed");
2148    }
2149
2150    #[fuchsia::test]
2151    async fn test_beyond_eof_read_from() {
2152        let (fs, object) = test_filesystem_and_object().await;
2153        let handle = &*object;
2154        let offset = TEST_OBJECT_SIZE as usize - 2;
2155        let align = offset % fs.block_size() as usize;
2156        let len: usize = 2;
2157        let mut buf = object.allocate_buffer(align + len + 1).await;
2158        buf.fill(123u8);
2159        assert_eq!(
2160            handle
2161                .read(AttributeId::DATA, (offset - align) as u64, buf.as_mut())
2162                .await
2163                .expect("read failed"),
2164            align + len
2165        );
2166        assert_eq!(&buf.as_ptr_slice().subslice(align..align + len).to_vec()[..], &vec![0u8; len]);
2167        assert_eq!(
2168            &buf.as_ptr_slice().subslice(align + len..buf.len()).to_vec()[..],
2169            &vec![123u8; buf.len() - align - len]
2170        );
2171        fs.close().await.expect("Close failed");
2172    }
2173
2174    #[fuchsia::test]
2175    async fn test_beyond_eof_read_unchecked() {
2176        let (fs, object) = test_filesystem_and_object().await;
2177        let offset = TEST_OBJECT_SIZE as usize - 2;
2178        let align = offset % fs.block_size() as usize;
2179        let len: usize = 2;
2180        let mut buf = object.allocate_buffer(align + len + 1).await;
2181        buf.fill(123u8);
2182        let guard = fs
2183            .lock_manager()
2184            .read_lock(lock_keys![LockKey::object_attribute(
2185                object.store().store_object_id,
2186                object.object_id(),
2187                AttributeId::DATA,
2188            )])
2189            .await;
2190        object
2191            .read_unchecked(AttributeId::DATA, (offset - align) as u64, buf.as_mut(), &guard)
2192            .await
2193            .expect("read failed");
2194        assert_eq!(
2195            &buf.as_ptr_slice().subslice(align..buf.len()).to_vec()[..],
2196            &vec![0u8; len + 1]
2197        );
2198        fs.close().await.expect("Close failed");
2199    }
2200
2201    #[fuchsia::test]
2202    async fn test_read_sparse() {
2203        let (fs, object) = test_filesystem_and_object().await;
2204        // Deliberately read not right to eof.
2205        let len = TEST_OBJECT_SIZE as usize - 1;
2206        let mut buf = object.allocate_buffer(len).await;
2207        buf.fill(123u8);
2208        assert_eq!(object.read(0, buf.as_mut()).await.expect("read failed"), len);
2209        let mut expected = vec![0; len];
2210        let offset = TEST_DATA_OFFSET as usize;
2211        expected[offset..offset + TEST_DATA.len()].copy_from_slice(TEST_DATA);
2212        assert_eq!(&buf.as_ptr_slice().subslice(0..len).to_vec()[..], &expected[..]);
2213        fs.close().await.expect("Close failed");
2214    }
2215
2216    #[fuchsia::test]
2217    async fn test_read_after_writes_interspersed_with_flush() {
2218        let (fs, object) = test_filesystem_and_object().await;
2219
2220        object.owner().flush().await.expect("flush failed");
2221
2222        // Write more test data to the first block fo the file.
2223        let mut buf = object.allocate_buffer(TEST_DATA.len()).await;
2224        buf.copy_from_slice(TEST_DATA);
2225        object.write_or_append(Some(0u64), buf.as_ref()).await.expect("write failed");
2226
2227        let len = TEST_OBJECT_SIZE as usize - 1;
2228        let mut buf = object.allocate_buffer(len).await;
2229        buf.fill(123u8);
2230        assert_eq!(object.read(0, buf.as_mut()).await.expect("read failed"), len);
2231
2232        let mut expected = vec![0u8; len];
2233        let offset = TEST_DATA_OFFSET as usize;
2234        expected[offset..offset + TEST_DATA.len()].copy_from_slice(TEST_DATA);
2235        expected[..TEST_DATA.len()].copy_from_slice(TEST_DATA);
2236        assert_eq!(&buf.to_vec(), &expected);
2237        fs.close().await.expect("Close failed");
2238    }
2239
2240    #[fuchsia::test]
2241    async fn test_read_after_truncate_and_extend() {
2242        let (fs, object) = test_filesystem_and_object().await;
2243
2244        // Arrange for there to be <extent><deleted-extent><extent>.
2245        let mut buf = object.allocate_buffer(TEST_DATA.len()).await;
2246        buf.copy_from_slice(TEST_DATA);
2247        // This adds an extent at 0..512.
2248        object.write_or_append(Some(0), buf.as_ref()).await.expect("write failed");
2249        // This deletes 512..1024.
2250        object.truncate(3).await.expect("truncate failed");
2251        let data = b"foo";
2252        let offset = 1500u64;
2253        let align = (offset % fs.block_size() as u64) as usize;
2254        let mut buf = object.allocate_buffer(align + data.len()).await;
2255        buf.subslice_mut(align..buf.len()).copy_from_slice(data);
2256        // This adds 1024..1536.
2257        object
2258            .write_or_append(Some(1500), buf.subslice(align..buf.len()))
2259            .await
2260            .expect("write failed");
2261
2262        const LEN1: usize = 1503;
2263        let mut buf = object.allocate_buffer(LEN1).await;
2264        buf.fill(123u8);
2265        assert_eq!(object.read(0, buf.as_mut()).await.expect("read failed"), LEN1);
2266        let mut expected = [0; LEN1];
2267        expected[..3].copy_from_slice(&TEST_DATA[..3]);
2268        expected[1500..].copy_from_slice(b"foo");
2269        assert_eq!(&buf.to_vec(), &expected);
2270
2271        // Also test a read that ends midway through the deleted extent.
2272        const LEN2: usize = 601;
2273        let mut buf = object.allocate_buffer(LEN2).await;
2274        buf.fill(123u8);
2275        assert_eq!(object.read(0, buf.as_mut()).await.expect("read failed"), LEN2);
2276        assert_eq!(buf.to_vec(), &expected[..LEN2]);
2277        fs.close().await.expect("Close failed");
2278    }
2279
2280    #[fuchsia::test]
2281    async fn test_read_whole_blocks_with_multiple_objects() {
2282        let (fs, object) = test_filesystem_and_object().await;
2283        let block_size = object.block_size() as usize;
2284        let mut buffer = object.allocate_buffer(block_size).await;
2285        buffer.fill(0xaf);
2286        object.write_or_append(Some(0), buffer.as_ref()).await.expect("write failed");
2287
2288        let store = object.owner();
2289        let mut transaction = fs
2290            .root_store()
2291            .new_transaction(lock_keys![], Options::default())
2292            .await
2293            .expect("new_transaction failed");
2294        let object2 =
2295            ObjectStore::create_object(&store, &mut transaction, HandleOptions::default(), None)
2296                .await
2297                .expect("create_object failed");
2298        transaction.commit().await.expect("commit failed");
2299        let mut ef_buffer = object.allocate_buffer(block_size).await;
2300        ef_buffer.fill(0xef);
2301        object2.write_or_append(Some(0), ef_buffer.as_ref()).await.expect("write failed");
2302
2303        let mut buffer = object.allocate_buffer(block_size).await;
2304        buffer.fill(0xaf);
2305        object
2306            .write_or_append(Some(block_size as u64), buffer.as_ref())
2307            .await
2308            .expect("write failed");
2309        object.truncate(3 * block_size as u64).await.expect("truncate failed");
2310        object2
2311            .write_or_append(Some(block_size as u64), ef_buffer.as_ref())
2312            .await
2313            .expect("write failed");
2314
2315        let mut buffer = object.allocate_buffer(4 * block_size).await;
2316        buffer.fill(123);
2317        assert_eq!(object.read(0, buffer.as_mut()).await.expect("read failed"), 3 * block_size);
2318        assert_eq!(
2319            &buffer.as_ptr_slice().subslice(0..2 * block_size).to_vec()[..],
2320            &vec![0xaf; 2 * block_size]
2321        );
2322        assert_eq!(
2323            &buffer.as_ptr_slice().subslice(2 * block_size..3 * block_size).to_vec()[..],
2324            &vec![0; block_size]
2325        );
2326        assert_eq!(object2.read(0, buffer.as_mut()).await.expect("read failed"), 2 * block_size);
2327        assert_eq!(
2328            &buffer.as_ptr_slice().subslice(0..2 * block_size).to_vec()[..],
2329            &vec![0xef; 2 * block_size]
2330        );
2331        fs.close().await.expect("Close failed");
2332    }
2333
2334    #[fuchsia::test]
2335    async fn test_alignment() {
2336        let (fs, object) = test_filesystem_and_object().await;
2337
2338        struct AlignTest {
2339            fill: u8,
2340            object: DataObjectHandle<ObjectStore>,
2341            mirror: Vec<u8>,
2342        }
2343
2344        impl AlignTest {
2345            async fn new(object: DataObjectHandle<ObjectStore>) -> Self {
2346                let mirror = {
2347                    let mut buf = object.allocate_buffer(object.get_size() as usize).await;
2348                    assert_eq!(object.read(0, buf.as_mut()).await.expect("read failed"), buf.len());
2349                    buf.to_vec()
2350                };
2351                Self { fill: 0, object, mirror }
2352            }
2353
2354            // Fills |range| of self.object with a byte value (self.fill) and mirrors the same
2355            // operation to an in-memory copy of the object.
2356            // Each subsequent call bumps the value of fill.
2357            // It is expected that the object and its mirror maintain identical content.
2358            async fn test(&mut self, range: Range<u64>) {
2359                let mut buf = self.object.allocate_buffer((range.end - range.start) as usize).await;
2360                self.fill += 1;
2361                buf.fill(self.fill);
2362                self.object
2363                    .write_or_append(Some(range.start), buf.as_ref())
2364                    .await
2365                    .expect("write_or_append failed");
2366                if range.end > self.mirror.len() as u64 {
2367                    self.mirror.resize(range.end as usize, 0);
2368                }
2369                self.mirror[range.start as usize..range.end as usize].fill(self.fill);
2370                let mut buf = self.object.allocate_buffer(self.mirror.len() + 1).await;
2371                assert_eq!(
2372                    self.object.read(0, buf.as_mut()).await.expect("read failed"),
2373                    self.mirror.len()
2374                );
2375                assert_eq!(
2376                    &buf.as_ptr_slice().subslice(0..self.mirror.len()).to_vec()[..],
2377                    self.mirror.as_slice()
2378                );
2379            }
2380        }
2381
2382        let block_size = object.block_size() as u64;
2383        let mut align = AlignTest::new(object).await;
2384
2385        // Fill the object to start with (with 1).
2386        align.test(0..2 * block_size + 1).await;
2387
2388        // Unaligned head (fills with 2, overwrites that with 3).
2389        align.test(1..block_size).await;
2390        align.test(1..2 * block_size).await;
2391
2392        // Unaligned tail (fills with 4 and 5).
2393        align.test(0..block_size - 1).await;
2394        align.test(0..2 * block_size - 1).await;
2395
2396        // Both unaligned (fills with 6 and 7).
2397        align.test(1..block_size - 1).await;
2398        align.test(1..2 * block_size - 1).await;
2399
2400        fs.close().await.expect("Close failed");
2401    }
2402
2403    async fn test_preallocate_common(fs: &FxFilesystem, object: DataObjectHandle<ObjectStore>) {
2404        let allocator = fs.allocator();
2405        let allocated_before = allocator.get_allocated_bytes();
2406        let mut transaction = object.new_transaction().await.expect("new_transaction failed");
2407        object
2408            .preallocate_range(&mut transaction, &mut (0..fs.block_size() as u64))
2409            .await
2410            .expect("preallocate_range failed");
2411        transaction.commit().await.expect("commit failed");
2412        assert!(object.get_size() < 1048576);
2413        let mut transaction = object.new_transaction().await.expect("new_transaction failed");
2414        object
2415            .preallocate_range(&mut transaction, &mut (0..1048576))
2416            .await
2417            .expect("preallocate_range failed");
2418        transaction.commit().await.expect("commit failed");
2419        assert_eq!(object.get_size(), 1048576);
2420        // Check that it didn't reallocate the space for the existing extent
2421        let allocated_after = allocator.get_allocated_bytes();
2422        assert_eq!(allocated_after - allocated_before, 1048576 - fs.block_size() as u64);
2423
2424        let mut buf = object
2425            .allocate_buffer(round_up(TEST_DATA_OFFSET, fs.block_size()).unwrap() as usize)
2426            .await;
2427        buf.fill(47);
2428        object
2429            .write_or_append(Some(0), buf.subslice(0..TEST_DATA_OFFSET as usize))
2430            .await
2431            .expect("write failed");
2432        buf.fill(95);
2433        let offset = round_up(TEST_OBJECT_SIZE, fs.block_size()).unwrap();
2434        object
2435            .overwrite(offset, buf.as_mut(), OverwriteOptions::default())
2436            .await
2437            .expect("write failed");
2438
2439        // Make sure there were no more allocations.
2440        assert_eq!(allocator.get_allocated_bytes(), allocated_after);
2441
2442        // Read back the data and make sure it is what we expect.
2443        let mut buf = object.allocate_buffer(104876).await;
2444        assert_eq!(object.read(0, buf.as_mut()).await.expect("read failed"), buf.len());
2445        assert_eq!(
2446            &buf.as_ptr_slice().subslice(0..TEST_DATA_OFFSET as usize).to_vec()[..],
2447            &[47; TEST_DATA_OFFSET as usize]
2448        );
2449        assert_eq!(
2450            &buf.as_ptr_slice()
2451                .subslice(TEST_DATA_OFFSET as usize..TEST_DATA_OFFSET as usize + TEST_DATA.len())
2452                .to_vec()[..],
2453            TEST_DATA
2454        );
2455        assert_eq!(
2456            &buf.as_ptr_slice().subslice(offset as usize..offset as usize + 2048).to_vec()[..],
2457            &[95; 2048]
2458        );
2459    }
2460
2461    #[fuchsia::test]
2462    async fn test_unaligned_overwrite_returns_error() {
2463        let (fs, object) = test_filesystem_and_object().await;
2464        let mut buf = object.allocate_buffer(100).await;
2465        let res = object.overwrite(0, buf.as_mut(), OverwriteOptions::default()).await;
2466        assert!(matches!(res, Err(e) if FxfsError::InvalidArgs.matches(&e)));
2467        fs.close().await.expect("Close failed");
2468    }
2469
2470    #[fuchsia::test]
2471    async fn test_unaligned_preallocate_returns_error() {
2472        let (fs, object) = test_filesystem_and_object().await;
2473        let mut transaction = fs
2474            .root_store()
2475            .new_transaction(lock_keys![], Options::default())
2476            .await
2477            .expect("new failed");
2478        let res = object.preallocate_range(&mut transaction, &mut (0..100)).await;
2479        assert!(matches!(res, Err(e) if FxfsError::InvalidArgs.matches(&e)));
2480        fs.close().await.expect("Close failed");
2481    }
2482
2483    #[fuchsia::test]
2484    async fn test_encrypted_preallocate_returns_error() {
2485        let (fs, object) = test_filesystem_and_object().await;
2486        let mut transaction = fs
2487            .root_store()
2488            .new_transaction(lock_keys![], Options::default())
2489            .await
2490            .expect("new failed");
2491        let bs = fs.block_size();
2492        let res = object.preallocate_range(&mut transaction, &mut (0..bs)).await;
2493        assert!(matches!(res, Err(e) if FxfsError::NotSupported.matches(&e)));
2494        fs.close().await.expect("Close failed");
2495    }
2496
2497    #[fuchsia::test]
2498    async fn test_preallocate_range() {
2499        let (fs, object) = test_filesystem_and_object_with_key(None, true).await;
2500        test_preallocate_common(&fs, object).await;
2501        fs.close().await.expect("Close failed");
2502    }
2503
2504    // This is identical to the previous test except that we flush so that extents end up in
2505    // different layers.
2506    #[fuchsia::test]
2507    async fn test_preallocate_succeeds_when_extents_are_in_different_layers() {
2508        let (fs, object) = test_filesystem_and_object_with_key(None, true).await;
2509        object.owner().flush().await.expect("flush failed");
2510        test_preallocate_common(&fs, object).await;
2511        fs.close().await.expect("Close failed");
2512    }
2513
2514    #[fuchsia::test]
2515    async fn test_already_preallocated() {
2516        let (fs, object) = test_filesystem_and_object_with_key(None, true).await;
2517        let allocator = fs.allocator();
2518        let allocated_before = allocator.get_allocated_bytes();
2519        let mut transaction = object.new_transaction().await.expect("new_transaction failed");
2520        let offset = TEST_DATA_OFFSET - TEST_DATA_OFFSET % fs.block_size() as u64;
2521        object
2522            .preallocate_range(&mut transaction, &mut (offset..offset + fs.block_size() as u64))
2523            .await
2524            .expect("preallocate_range failed");
2525        transaction.commit().await.expect("commit failed");
2526        // Check that it didn't reallocate any new space.
2527        assert_eq!(allocator.get_allocated_bytes(), allocated_before);
2528        fs.close().await.expect("Close failed");
2529    }
2530
2531    #[fuchsia::test]
2532    async fn test_overwrite_when_preallocated_at_start_of_file() {
2533        // The standard test data we put in the test object would cause an extent with checksums
2534        // to be created, which overwrite() doesn't support. So we create an empty object instead.
2535        let (fs, object) = test_filesystem_and_empty_object().await;
2536
2537        let object = ObjectStore::open_object(
2538            object.owner(),
2539            object.object_id(),
2540            HandleOptions::default(),
2541            None,
2542        )
2543        .await
2544        .expect("open_object failed");
2545
2546        assert_eq!(fs.block_size(), 4096);
2547
2548        let mut write_buf = object.allocate_buffer(4096).await;
2549        write_buf.fill(95);
2550
2551        // First try to overwrite without allowing allocations
2552        // We expect this to fail, since nothing is allocated yet
2553        object
2554            .overwrite(0, write_buf.as_mut(), OverwriteOptions::default())
2555            .await
2556            .expect_err("overwrite succeeded");
2557
2558        // Now preallocate some space (exactly one block)
2559        let mut transaction = object.new_transaction().await.expect("new_transaction failed");
2560        object
2561            .preallocate_range(&mut transaction, &mut (0..4096 as u64))
2562            .await
2563            .expect("preallocate_range failed");
2564        transaction.commit().await.expect("commit failed");
2565
2566        // Now try the same overwrite command as before, it should work this time,
2567        // even with allocations disabled...
2568        {
2569            let mut read_buf = object.allocate_buffer(4096).await;
2570            object.read(0, read_buf.as_mut()).await.expect("read failed");
2571            assert_eq!(&read_buf.to_vec()[..], &[0; 4096]);
2572        }
2573        object
2574            .overwrite(0, write_buf.as_mut(), OverwriteOptions::default())
2575            .await
2576            .expect("overwrite failed");
2577        {
2578            let mut read_buf = object.allocate_buffer(4096).await;
2579            object.read(0, read_buf.as_mut()).await.expect("read failed");
2580            assert_eq!(&read_buf.to_vec()[..], &[95; 4096]);
2581        }
2582
2583        // Now try to overwrite at offset 4096. We expect this to fail, since we only preallocated
2584        // one block earlier at offset 0
2585        object
2586            .overwrite(4096, write_buf.as_mut(), OverwriteOptions::default())
2587            .await
2588            .expect_err("overwrite succeeded");
2589
2590        // We can't assert anything about the existing bytes, because they haven't been allocated
2591        // yet and they could contain any values
2592        object
2593            .overwrite(
2594                4096,
2595                write_buf.as_mut(),
2596                OverwriteOptions { allow_allocations: true, ..Default::default() },
2597            )
2598            .await
2599            .expect("overwrite failed");
2600        {
2601            let mut read_buf = object.allocate_buffer(4096).await;
2602            object.read(4096, read_buf.as_mut()).await.expect("read failed");
2603            assert_eq!(&read_buf.to_vec()[..], &[95; 4096]);
2604        }
2605
2606        // Check that the overwrites haven't messed up the filesystem state
2607        let fsck_options = FsckOptions {
2608            fail_on_warning: true,
2609            no_lock: true,
2610            on_error: Box::new(|err| println!("fsck error: {:?}", err)),
2611            ..Default::default()
2612        };
2613        fsck_with_options(fs.clone(), &fsck_options).await.expect("fsck failed");
2614
2615        fs.close().await.expect("Close failed");
2616    }
2617
2618    #[fuchsia::test]
2619    async fn test_overwrite_large_buffer_and_file_with_many_holes() {
2620        // The standard test data we put in the test object would cause an extent with checksums
2621        // to be created, which overwrite() doesn't support. So we create an empty object instead.
2622        let (fs, object) = test_filesystem_and_empty_object().await;
2623
2624        let object = ObjectStore::open_object(
2625            object.owner(),
2626            object.object_id(),
2627            HandleOptions::default(),
2628            None,
2629        )
2630        .await
2631        .expect("open_object failed");
2632
2633        assert_eq!(fs.block_size(), 4096);
2634        assert_eq!(object.get_size(), TEST_OBJECT_SIZE);
2635
2636        // Let's create some non-holes
2637        let mut transaction = object.new_transaction().await.expect("new_transaction failed");
2638        object
2639            .preallocate_range(&mut transaction, &mut (4096..8192 as u64))
2640            .await
2641            .expect("preallocate_range failed");
2642        object
2643            .preallocate_range(&mut transaction, &mut (16384..32768 as u64))
2644            .await
2645            .expect("preallocate_range failed");
2646        object
2647            .preallocate_range(&mut transaction, &mut (65536..131072 as u64))
2648            .await
2649            .expect("preallocate_range failed");
2650        object
2651            .preallocate_range(&mut transaction, &mut (262144..524288 as u64))
2652            .await
2653            .expect("preallocate_range failed");
2654        transaction.commit().await.expect("commit failed");
2655
2656        assert_eq!(object.get_size(), 524288);
2657
2658        let mut write_buf = object.allocate_buffer(4096).await;
2659        write_buf.fill(95);
2660
2661        // We shouldn't be able to overwrite in the holes if new allocations aren't enabled
2662        object
2663            .overwrite(0, write_buf.as_mut(), OverwriteOptions::default())
2664            .await
2665            .expect_err("overwrite succeeded");
2666        object
2667            .overwrite(8192, write_buf.as_mut(), OverwriteOptions::default())
2668            .await
2669            .expect_err("overwrite succeeded");
2670        object
2671            .overwrite(32768, write_buf.as_mut(), OverwriteOptions::default())
2672            .await
2673            .expect_err("overwrite succeeded");
2674        object
2675            .overwrite(131072, write_buf.as_mut(), OverwriteOptions::default())
2676            .await
2677            .expect_err("overwrite succeeded");
2678
2679        // But we should be able to overwrite in the prealloc'd areas without needing allocations
2680        {
2681            let mut read_buf = object.allocate_buffer(4096).await;
2682            object.read(4096, read_buf.as_mut()).await.expect("read failed");
2683            assert_eq!(&read_buf.to_vec()[..], &[0; 4096]);
2684        }
2685        object
2686            .overwrite(4096, write_buf.as_mut(), OverwriteOptions::default())
2687            .await
2688            .expect("overwrite failed");
2689        {
2690            let mut read_buf = object.allocate_buffer(4096).await;
2691            object.read(4096, read_buf.as_mut()).await.expect("read failed");
2692            assert_eq!(&read_buf.to_vec()[..], &[95; 4096]);
2693        }
2694        {
2695            let mut read_buf = object.allocate_buffer(4096).await;
2696            object.read(16384, read_buf.as_mut()).await.expect("read failed");
2697            assert_eq!(&read_buf.to_vec()[..], &[0; 4096]);
2698        }
2699        object
2700            .overwrite(16384, write_buf.as_mut(), OverwriteOptions::default())
2701            .await
2702            .expect("overwrite failed");
2703        {
2704            let mut read_buf = object.allocate_buffer(4096).await;
2705            object.read(16384, read_buf.as_mut()).await.expect("read failed");
2706            assert_eq!(&read_buf.to_vec()[..], &[95; 4096]);
2707        }
2708        {
2709            let mut read_buf = object.allocate_buffer(4096).await;
2710            object.read(65536, read_buf.as_mut()).await.expect("read failed");
2711            assert_eq!(&read_buf.to_vec()[..], &[0; 4096]);
2712        }
2713        object
2714            .overwrite(65536, write_buf.as_mut(), OverwriteOptions::default())
2715            .await
2716            .expect("overwrite failed");
2717        {
2718            let mut read_buf = object.allocate_buffer(4096).await;
2719            object.read(65536, read_buf.as_mut()).await.expect("read failed");
2720            assert_eq!(&read_buf.to_vec()[..], &[95; 4096]);
2721        }
2722        {
2723            let mut read_buf = object.allocate_buffer(4096).await;
2724            object.read(262144, read_buf.as_mut()).await.expect("read failed");
2725            assert_eq!(&read_buf.to_vec()[..], &[0; 4096]);
2726        }
2727        object
2728            .overwrite(262144, write_buf.as_mut(), OverwriteOptions::default())
2729            .await
2730            .expect("overwrite failed");
2731        {
2732            let mut read_buf = object.allocate_buffer(4096).await;
2733            object.read(262144, read_buf.as_mut()).await.expect("read failed");
2734            assert_eq!(&read_buf.to_vec()[..], &[95; 4096]);
2735        }
2736
2737        // Now let's try to do a huge overwrite, that spans over many holes and non-holes
2738        let mut huge_write_buf = object.allocate_buffer(524288).await;
2739        huge_write_buf.fill(96);
2740
2741        // With allocations disabled, the big overwrite should fail...
2742        object
2743            .overwrite(0, huge_write_buf.as_mut(), OverwriteOptions::default())
2744            .await
2745            .expect_err("overwrite succeeded");
2746        // ... but it should work when allocations are enabled
2747        object
2748            .overwrite(
2749                0,
2750                huge_write_buf.as_mut(),
2751                OverwriteOptions { allow_allocations: true, ..Default::default() },
2752            )
2753            .await
2754            .expect("overwrite failed");
2755        {
2756            let mut read_buf = object.allocate_buffer(524288).await;
2757            object.read(0, read_buf.as_mut()).await.expect("read failed");
2758            assert_eq!(&read_buf.to_vec()[..], &[96; 524288]);
2759        }
2760
2761        // Check that the overwrites haven't messed up the filesystem state
2762        let fsck_options = FsckOptions {
2763            fail_on_warning: true,
2764            no_lock: true,
2765            on_error: Box::new(|err| println!("fsck error: {:?}", err)),
2766            ..Default::default()
2767        };
2768        fsck_with_options(fs.clone(), &fsck_options).await.expect("fsck failed");
2769
2770        fs.close().await.expect("Close failed");
2771    }
2772
2773    #[fuchsia::test]
2774    async fn test_overwrite_when_unallocated_at_start_of_file() {
2775        // The standard test data we put in the test object would cause an extent with checksums
2776        // to be created, which overwrite() doesn't support. So we create an empty object instead.
2777        let (fs, object) = test_filesystem_and_empty_object().await;
2778
2779        let object = ObjectStore::open_object(
2780            object.owner(),
2781            object.object_id(),
2782            HandleOptions::default(),
2783            None,
2784        )
2785        .await
2786        .expect("open_object failed");
2787
2788        assert_eq!(fs.block_size(), 4096);
2789
2790        let mut write_buf = object.allocate_buffer(4096).await;
2791        write_buf.fill(95);
2792
2793        // First try to overwrite without allowing allocations
2794        // We expect this to fail, since nothing is allocated yet
2795        object
2796            .overwrite(0, write_buf.as_mut(), OverwriteOptions::default())
2797            .await
2798            .expect_err("overwrite succeeded");
2799
2800        // Now try the same overwrite command as before, but allow allocations
2801        object
2802            .overwrite(
2803                0,
2804                write_buf.as_mut(),
2805                OverwriteOptions { allow_allocations: true, ..Default::default() },
2806            )
2807            .await
2808            .expect("overwrite failed");
2809        {
2810            let mut read_buf = object.allocate_buffer(4096).await;
2811            object.read(0, read_buf.as_mut()).await.expect("read failed");
2812            assert_eq!(&read_buf.to_vec()[..], &[95; 4096]);
2813        }
2814
2815        // Now try to overwrite at the next block. This should fail if allocations are disabled
2816        object
2817            .overwrite(4096, write_buf.as_mut(), OverwriteOptions::default())
2818            .await
2819            .expect_err("overwrite succeeded");
2820
2821        // ... but it should work if allocations are enabled
2822        object
2823            .overwrite(
2824                4096,
2825                write_buf.as_mut(),
2826                OverwriteOptions { allow_allocations: true, ..Default::default() },
2827            )
2828            .await
2829            .expect("overwrite failed");
2830        {
2831            let mut read_buf = object.allocate_buffer(4096).await;
2832            object.read(4096, read_buf.as_mut()).await.expect("read failed");
2833            assert_eq!(&read_buf.to_vec()[..], &[95; 4096]);
2834        }
2835
2836        // Check that the overwrites haven't messed up the filesystem state
2837        let fsck_options = FsckOptions {
2838            fail_on_warning: true,
2839            no_lock: true,
2840            on_error: Box::new(|err| println!("fsck error: {:?}", err)),
2841            ..Default::default()
2842        };
2843        fsck_with_options(fs.clone(), &fsck_options).await.expect("fsck failed");
2844
2845        fs.close().await.expect("Close failed");
2846    }
2847
2848    #[fuchsia::test]
2849    async fn test_overwrite_can_extend_a_file() {
2850        // The standard test data we put in the test object would cause an extent with checksums
2851        // to be created, which overwrite() doesn't support. So we create an empty object instead.
2852        let (fs, object) = test_filesystem_and_empty_object().await;
2853
2854        let object = ObjectStore::open_object(
2855            object.owner(),
2856            object.object_id(),
2857            HandleOptions::default(),
2858            None,
2859        )
2860        .await
2861        .expect("open_object failed");
2862
2863        assert_eq!(fs.block_size(), 4096);
2864        assert_eq!(object.get_size(), TEST_OBJECT_SIZE);
2865
2866        let mut write_buf = object.allocate_buffer(4096).await;
2867        write_buf.fill(95);
2868
2869        // Let's try to fill up the last block, and increase the file size in doing so
2870        let last_block_offset = round_down(TEST_OBJECT_SIZE, 4096 as u32);
2871
2872        // Expected to fail with allocations disabled
2873        object
2874            .overwrite(last_block_offset, write_buf.as_mut(), OverwriteOptions::default())
2875            .await
2876            .expect_err("overwrite succeeded");
2877        // ... but expected to succeed with allocations enabled
2878        object
2879            .overwrite(
2880                last_block_offset,
2881                write_buf.as_mut(),
2882                OverwriteOptions { allow_allocations: true, ..Default::default() },
2883            )
2884            .await
2885            .expect("overwrite failed");
2886        {
2887            let mut read_buf = object.allocate_buffer(4096).await;
2888            object.read(last_block_offset, read_buf.as_mut()).await.expect("read failed");
2889            assert_eq!(&read_buf.to_vec()[..], &[95; 4096]);
2890        }
2891
2892        assert_eq!(object.get_size(), 8192);
2893
2894        // Let's try to write at the next block, too
2895        let next_block_offset = round_up(TEST_OBJECT_SIZE, 4096 as u32).unwrap();
2896
2897        // Expected to fail with allocations disabled
2898        object
2899            .overwrite(next_block_offset, write_buf.as_mut(), OverwriteOptions::default())
2900            .await
2901            .expect_err("overwrite succeeded");
2902        // ... but expected to succeed with allocations enabled
2903        object
2904            .overwrite(
2905                next_block_offset,
2906                write_buf.as_mut(),
2907                OverwriteOptions { allow_allocations: true, ..Default::default() },
2908            )
2909            .await
2910            .expect("overwrite failed");
2911        {
2912            let mut read_buf = object.allocate_buffer(4096).await;
2913            object.read(next_block_offset, read_buf.as_mut()).await.expect("read failed");
2914            assert_eq!(&read_buf.to_vec()[..], &[95; 4096]);
2915        }
2916
2917        assert_eq!(object.get_size(), 12288);
2918
2919        // Check that the overwrites haven't messed up the filesystem state
2920        let fsck_options = FsckOptions {
2921            fail_on_warning: true,
2922            no_lock: true,
2923            on_error: Box::new(|err| println!("fsck error: {:?}", err)),
2924            ..Default::default()
2925        };
2926        fsck_with_options(fs.clone(), &fsck_options).await.expect("fsck failed");
2927
2928        fs.close().await.expect("Close failed");
2929    }
2930
2931    #[fuchsia::test]
2932    async fn test_enable_verity() {
2933        let fs: OpenFxFilesystem = test_filesystem().await;
2934        let mut transaction = fs
2935            .root_store()
2936            .new_transaction(lock_keys![], Options::default())
2937            .await
2938            .expect("new_transaction failed");
2939        let store = fs.root_store();
2940        let object = Arc::new(
2941            ObjectStore::create_object(&store, &mut transaction, HandleOptions::default(), None)
2942                .await
2943                .expect("create_object failed"),
2944        );
2945
2946        transaction.commit().await.unwrap();
2947
2948        object
2949            .enable_verity(fio::VerificationOptions {
2950                hash_algorithm: Some(fio::HashAlgorithm::Sha256),
2951                salt: Some(vec![]),
2952                ..Default::default()
2953            })
2954            .await
2955            .expect("set verified file metadata failed");
2956
2957        let handle =
2958            ObjectStore::open_object(&store, object.object_id(), HandleOptions::default(), None)
2959                .await
2960                .expect("open_object failed");
2961
2962        assert!(handle.is_verified_file());
2963
2964        fs.close().await.expect("Close failed");
2965    }
2966
2967    #[fuchsia::test]
2968    async fn test_enable_verity_large_file() {
2969        // Need to make a large FakeDevice to create space for a 67 MB file.
2970        let device = DeviceHolder::new(FakeDevice::new(262144, TEST_DEVICE_BLOCK_SIZE));
2971        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
2972        let root_store = fs.root_store();
2973        let mut transaction = fs
2974            .root_store()
2975            .new_transaction(lock_keys![], Options::default())
2976            .await
2977            .expect("new_transaction failed");
2978
2979        let handle = ObjectStore::create_object(
2980            &root_store,
2981            &mut transaction,
2982            HandleOptions::default(),
2983            None,
2984        )
2985        .await
2986        .expect("failed to create object");
2987        transaction.commit().await.expect("commit failed");
2988        let mut offset = 0;
2989
2990        // Write a file big enough to trigger multiple transactions on enable_verity().
2991        let mut buf = handle.allocate_buffer(WRITE_ATTR_BATCH_SIZE).await;
2992        buf.fill(1);
2993        for _ in 0..130 {
2994            handle.write_or_append(Some(offset), buf.as_ref()).await.expect("write failed");
2995            offset += WRITE_ATTR_BATCH_SIZE as u64;
2996        }
2997
2998        handle
2999            .enable_verity(fio::VerificationOptions {
3000                hash_algorithm: Some(fio::HashAlgorithm::Sha256),
3001                salt: Some(vec![]),
3002                ..Default::default()
3003            })
3004            .await
3005            .expect("set verified file metadata failed");
3006
3007        let mut buf = handle.allocate_buffer(WRITE_ATTR_BATCH_SIZE).await;
3008        offset = 0;
3009        for _ in 0..130 {
3010            handle.read(offset, buf.as_mut()).await.expect("verification during read should fail");
3011            assert_eq!(buf.to_vec(), &[1; WRITE_ATTR_BATCH_SIZE]);
3012            offset += WRITE_ATTR_BATCH_SIZE as u64;
3013        }
3014
3015        fsck(fs.clone()).await.expect("fsck failed");
3016        fs.close().await.expect("Close failed");
3017    }
3018
3019    #[fuchsia::test]
3020    async fn test_retry_enable_verity_on_reboot() {
3021        let device = DeviceHolder::new(FakeDevice::new(8192, TEST_DEVICE_BLOCK_SIZE));
3022        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
3023        let root_store = fs.root_store();
3024        let mut transaction = fs
3025            .root_store()
3026            .new_transaction(lock_keys![], Options::default())
3027            .await
3028            .expect("new_transaction failed");
3029
3030        let handle = ObjectStore::create_object(
3031            &root_store,
3032            &mut transaction,
3033            HandleOptions::default(),
3034            None,
3035        )
3036        .await
3037        .expect("failed to create object");
3038        transaction.commit().await.expect("commit failed");
3039
3040        let object_id = {
3041            let mut transaction = handle.new_transaction().await.expect("new_transaction failed");
3042            transaction.add(
3043                root_store.store_object_id(),
3044                Mutation::replace_or_insert_object(
3045                    ObjectKey::graveyard_attribute_entry(
3046                        root_store.graveyard_directory_object_id(),
3047                        handle.object_id(),
3048                        AttributeId::FSVERITY_MERKLE,
3049                    ),
3050                    ObjectValue::Some,
3051                ),
3052            );
3053
3054            // This write should span three transactions. This test mimics the behavior when the
3055            // last transaction gets interrupted by a filesystem.close().
3056            handle
3057                .write_new_attr_in_batches(
3058                    &mut transaction,
3059                    AttributeId::FSVERITY_MERKLE,
3060                    &vec![0; 2 * WRITE_ATTR_BATCH_SIZE],
3061                    WRITE_ATTR_BATCH_SIZE,
3062                )
3063                .await
3064                .expect("failed to write merkle attribute");
3065
3066            handle.object_id()
3067            // Drop the transaction to simulate interrupting the merkle tree creation as well as to
3068            // release the transaction locks.
3069        };
3070
3071        fs.close().await.expect("failed to close filesystem");
3072        let device = fs.take_device().await;
3073        device.reopen(false);
3074
3075        let fs =
3076            FxFilesystemBuilder::new().read_only(true).open(device).await.expect("open failed");
3077        fsck(fs.clone()).await.expect("fsck failed");
3078        fs.close().await.expect("failed to close filesystem");
3079        let device = fs.take_device().await;
3080        device.reopen(false);
3081
3082        // On open, the filesystem will call initial_reap which will call queue_tombstone().
3083        let fs = FxFilesystem::open(device).await.expect("open failed");
3084        let root_store = fs.root_store();
3085        let handle =
3086            ObjectStore::open_object(&root_store, object_id, HandleOptions::default(), None)
3087                .await
3088                .expect("open_object failed");
3089        handle
3090            .enable_verity(fio::VerificationOptions {
3091                hash_algorithm: Some(fio::HashAlgorithm::Sha256),
3092                salt: Some(vec![]),
3093                ..Default::default()
3094            })
3095            .await
3096            .expect("set verified file metadata failed");
3097
3098        // `flush` will ensure that initial reap fully processes all the graveyard entries. This
3099        // isn't strictly necessary for the test to pass (the graveyard marker was already
3100        // processed during `enable_verity`), but it does help catch bugs, such as the attribute
3101        // graveyard entry not being removed upon processing.
3102        fs.graveyard().flush().await;
3103        let merkle_data = handle
3104            .read_attr(AttributeId::FSVERITY_MERKLE)
3105            .await
3106            .expect("read_attr failed")
3107            .expect("No attr found");
3108        assert!(FsVerityDescriptor::new(&merkle_data[..], handle.block_size() as usize).is_ok());
3109        fsck(fs.clone()).await.expect("fsck failed");
3110        fs.close().await.expect("Close failed");
3111    }
3112
3113    #[fuchsia::test]
3114    async fn test_verify_data_corrupt_file() {
3115        let fs: OpenFxFilesystem = test_filesystem().await;
3116        let mut transaction = fs
3117            .root_store()
3118            .new_transaction(lock_keys![], Options::default())
3119            .await
3120            .expect("new_transaction failed");
3121        let store = fs.root_store();
3122        let object = Arc::new(
3123            ObjectStore::create_object(&store, &mut transaction, HandleOptions::default(), None)
3124                .await
3125                .expect("create_object failed"),
3126        );
3127
3128        transaction.commit().await.unwrap();
3129
3130        let mut buf = object.allocate_buffer(5 * fs.block_size() as usize).await;
3131        buf.fill(123);
3132        object.write_or_append(Some(0), buf.as_ref()).await.expect("write failed");
3133
3134        object
3135            .enable_verity(fio::VerificationOptions {
3136                hash_algorithm: Some(fio::HashAlgorithm::Sha256),
3137                salt: Some(vec![]),
3138                ..Default::default()
3139            })
3140            .await
3141            .expect("set verified file metadata failed");
3142
3143        // Change file contents and ensure verification fails
3144        buf.fill(234);
3145        object.write_or_append(Some(0), buf.as_ref()).await.expect("write failed");
3146        object.read(0, buf.as_mut()).await.expect_err("verification during read should fail");
3147
3148        fs.close().await.expect("Close failed");
3149    }
3150
3151    // TODO(https://fxbug.dev/450398331): More tests to be added when this can support writing the
3152    // f2fs format natively. For now, relying on tests inside of the f2fs_reader to exercise more
3153    // paths.
3154    #[fuchsia::test]
3155    async fn test_parse_f2fs_verity() {
3156        let fs: OpenFxFilesystem = test_filesystem().await;
3157        let mut transaction = fs
3158            .root_store()
3159            .new_transaction(lock_keys![], Options::default())
3160            .await
3161            .expect("new_transaction failed");
3162        let store = fs.root_store();
3163        let object = Arc::new(
3164            ObjectStore::create_object(&store, &mut transaction, HandleOptions::default(), None)
3165                .await
3166                .expect("create_object failed"),
3167        );
3168
3169        transaction.commit().await.unwrap();
3170        let file_size = fs.block_size() * 2;
3171        // Write over one block to make there be leaf hashes.
3172        {
3173            let mut buf = object.allocate_buffer(file_size as usize).await;
3174            buf.fill(64);
3175            assert_eq!(
3176                object.write_or_append(None, buf.as_ref()).await.expect("Writing to file."),
3177                file_size
3178            );
3179        }
3180
3181        // Enable verity normally, then shift the type.
3182        object
3183            .enable_verity(fio::VerificationOptions {
3184                hash_algorithm: Some(fio::HashAlgorithm::Sha256),
3185                salt: Some(vec![]),
3186                ..Default::default()
3187            })
3188            .await
3189            .expect("set verified file metadata failed");
3190        let (verity_info, root_hash) = object.get_descriptor().unwrap();
3191
3192        let mut transaction = fs
3193            .root_store()
3194            .new_transaction(
3195                lock_keys![LockKey::Object {
3196                    store_object_id: store.store_object_id(),
3197                    object_id: object.object_id()
3198                }],
3199                Options::default(),
3200            )
3201            .await
3202            .expect("new_transaction failed");
3203        transaction.add(
3204            store.store_object_id(),
3205            Mutation::replace_or_insert_object(
3206                ObjectKey::attribute(
3207                    object.object_id(),
3208                    AttributeId::DATA,
3209                    AttributeKey::Attribute,
3210                ),
3211                ObjectValue::verified_attribute(
3212                    file_size,
3213                    FsverityMetadata::F2fs(0..(fs.block_size() * 2)),
3214                ),
3215            ),
3216        );
3217        transaction.add(
3218            store.store_object_id(),
3219            Mutation::replace_or_insert_object(
3220                ObjectKey::attribute(
3221                    object.object_id(),
3222                    AttributeId::FSVERITY_MERKLE,
3223                    AttributeKey::Attribute,
3224                ),
3225                ObjectValue::attribute(fs.block_size() * 2, false),
3226            ),
3227        );
3228        {
3229            let descriptor = FsVerityDescriptorRaw::new(
3230                fio::HashAlgorithm::Sha256,
3231                fs.block_size(),
3232                file_size,
3233                root_hash.as_slice(),
3234                match &verity_info.salt {
3235                    Some(salt) => salt.as_slice(),
3236                    None => [0u8; 0].as_slice(),
3237                },
3238            )
3239            .expect("Creating descriptor");
3240            let mut buf = object.allocate_buffer(fs.block_size() as usize).await;
3241            let mut temp = vec![0u8; fs.block_size() as usize];
3242            descriptor.write_to_slice(&mut temp).expect("Writing descriptor to buf");
3243            buf.copy_from_slice(&temp);
3244            object
3245                .multi_write(
3246                    &mut transaction,
3247                    AttributeId::FSVERITY_MERKLE,
3248                    &[fs.block_size()..(fs.block_size() * 2)],
3249                    buf.as_mut(),
3250                )
3251                .await
3252                .expect("Writing descriptor");
3253        }
3254        transaction.commit().await.unwrap();
3255
3256        let handle =
3257            ObjectStore::open_object(&store, object.object_id(), HandleOptions::default(), None)
3258                .await
3259                .expect("open_object failed");
3260
3261        assert!(handle.is_verified_file());
3262
3263        let mut buf = object.allocate_buffer(file_size as usize).await;
3264        assert_eq!(
3265            handle.read(0, buf.as_mut()).await.expect("Read whole file."),
3266            file_size as usize
3267        );
3268
3269        fs.close().await.expect("Close failed");
3270    }
3271
3272    #[fuchsia::test]
3273    async fn test_verify_data_corrupt_tree() {
3274        let fs: OpenFxFilesystem = test_filesystem().await;
3275        let object_id = {
3276            let store = fs.root_store();
3277            let mut transaction = fs
3278                .root_store()
3279                .new_transaction(lock_keys![], Options::default())
3280                .await
3281                .expect("new_transaction failed");
3282            let object = Arc::new(
3283                ObjectStore::create_object(
3284                    &store,
3285                    &mut transaction,
3286                    HandleOptions::default(),
3287                    None,
3288                )
3289                .await
3290                .expect("create_object failed"),
3291            );
3292            let object_id = object.object_id();
3293
3294            transaction.commit().await.unwrap();
3295
3296            let mut buf = object.allocate_buffer(5 * fs.block_size() as usize).await;
3297            buf.fill(123);
3298            object.write_or_append(Some(0), buf.as_ref()).await.expect("write failed");
3299
3300            object
3301                .enable_verity(fio::VerificationOptions {
3302                    hash_algorithm: Some(fio::HashAlgorithm::Sha256),
3303                    salt: Some(vec![]),
3304                    ..Default::default()
3305                })
3306                .await
3307                .expect("set verified file metadata failed");
3308            object.read(0, buf.as_mut()).await.expect("verified read");
3309
3310            // Corrupt the merkle tree before closing.
3311            let mut merkle = object
3312                .read_attr(AttributeId::FSVERITY_MERKLE)
3313                .await
3314                .unwrap()
3315                .expect("Reading merkle tree");
3316            merkle[0] = merkle[0].wrapping_add(1);
3317            object
3318                .write_attr(AttributeId::FSVERITY_MERKLE, &*merkle)
3319                .await
3320                .expect("Overwriting merkle");
3321
3322            object_id
3323        }; // Close object.
3324
3325        // Reopening the object should complain about the corrupted merkle tree.
3326        assert!(
3327            ObjectStore::open_object(&fs.root_store(), object_id, HandleOptions::default(), None)
3328                .await
3329                .is_err()
3330        );
3331        fs.close().await.expect("Close failed");
3332    }
3333
3334    #[fuchsia::test]
3335    async fn test_allocate_verity_file() {
3336        let fs: OpenFxFilesystem = test_filesystem().await;
3337        let mut transaction = fs
3338            .root_store()
3339            .new_transaction(lock_keys![], Options::default())
3340            .await
3341            .expect("new_transaction failed");
3342        let store = fs.root_store();
3343        let object = Arc::new(
3344            ObjectStore::create_object(&store, &mut transaction, HandleOptions::default(), None)
3345                .await
3346                .expect("create_object failed"),
3347        );
3348        transaction.commit().await.unwrap();
3349
3350        let mut buf = object.allocate_buffer(8192).await;
3351        buf.fill(0xAA);
3352        object.write_or_append(Some(0), buf.as_ref()).await.expect("write failed");
3353
3354        object
3355            .enable_verity(fio::VerificationOptions {
3356                hash_algorithm: Some(fio::HashAlgorithm::Sha256),
3357                salt: Some(vec![]),
3358                ..Default::default()
3359            })
3360            .await
3361            .expect("enable_verity failed");
3362
3363        assert!(object.is_verified_file());
3364
3365        // Calling allocate on a verity-enabled file should return an error.
3366        assert!(object.allocate(0..8192).await.is_err());
3367
3368        // Even after opening the object again (simulating cache eviction), it must remain verified.
3369        let reopened =
3370            ObjectStore::open_object(&store, object.object_id(), HandleOptions::default(), None)
3371                .await
3372                .expect("open_object failed");
3373        assert!(reopened.is_verified_file());
3374
3375        fs.close().await.expect("Close failed");
3376    }
3377
3378    #[fuchsia::test]
3379    async fn test_extend() {
3380        let fs = test_filesystem().await;
3381        let handle;
3382        let mut transaction = fs
3383            .root_store()
3384            .new_transaction(lock_keys![], Options::default())
3385            .await
3386            .expect("new_transaction failed");
3387        let store = fs.root_store();
3388        handle =
3389            ObjectStore::create_object(&store, &mut transaction, HandleOptions::default(), None)
3390                .await
3391                .expect("create_object failed");
3392
3393        // As of writing, an empty filesystem has two 512kiB superblock extents and a little over
3394        // 256kiB of additional allocations (journal, etc) so we start use a 'magic' starting point
3395        // of 2MiB here.
3396        const START_OFFSET: u64 = 2048 * 1024;
3397        handle
3398            .extend(&mut transaction, START_OFFSET..START_OFFSET + 5 * fs.block_size() as u64)
3399            .await
3400            .expect("extend failed");
3401        transaction.commit().await.expect("commit failed");
3402        let mut buf = handle.allocate_buffer(5 * fs.block_size() as usize).await;
3403        buf.fill(123);
3404        handle.write_or_append(Some(0), buf.as_ref()).await.expect("write failed");
3405        buf.fill(67);
3406        handle.read(0, buf.as_mut()).await.expect("read failed");
3407        assert_eq!(buf.to_vec(), vec![123; 5 * fs.block_size() as usize]);
3408        fs.close().await.expect("Close failed");
3409    }
3410
3411    #[fuchsia::test]
3412    async fn test_truncate_deallocates_old_extents() {
3413        let (fs, object) = test_filesystem_and_object().await;
3414        let mut buf = object.allocate_buffer(5 * fs.block_size() as usize).await;
3415        buf.fill(0xaa);
3416        object.write_or_append(Some(0), buf.as_ref()).await.expect("write failed");
3417
3418        let allocator = fs.allocator();
3419        let allocated_before = allocator.get_allocated_bytes();
3420        object.truncate(fs.block_size() as u64).await.expect("truncate failed");
3421        let allocated_after = allocator.get_allocated_bytes();
3422        assert!(
3423            allocated_after < allocated_before,
3424            "before = {} after = {}",
3425            allocated_before,
3426            allocated_after
3427        );
3428        fs.close().await.expect("Close failed");
3429    }
3430
3431    #[fuchsia::test]
3432    async fn test_truncate_zeroes_tail_block() {
3433        let (fs, object) = test_filesystem_and_object().await;
3434
3435        WriteObjectHandle::truncate(&object, TEST_DATA_OFFSET + 3).await.expect("truncate failed");
3436        WriteObjectHandle::truncate(&object, TEST_DATA_OFFSET + TEST_DATA.len() as u64)
3437            .await
3438            .expect("truncate failed");
3439
3440        let mut buf = object.allocate_buffer(fs.block_size() as usize).await;
3441        let offset = (TEST_DATA_OFFSET % fs.block_size()) as usize;
3442        object.read(TEST_DATA_OFFSET - offset as u64, buf.as_mut()).await.expect("read failed");
3443
3444        let mut expected = TEST_DATA.to_vec();
3445        expected[3..].fill(0);
3446        assert_eq!(
3447            &buf.as_ptr_slice().subslice(offset..offset + expected.len()).to_vec()[..],
3448            &expected
3449        );
3450    }
3451
3452    #[fuchsia::test]
3453    async fn test_trim() {
3454        // Format a new filesystem.
3455        let device = DeviceHolder::new(FakeDevice::new(8192, TEST_DEVICE_BLOCK_SIZE));
3456        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
3457        let block_size = fs.block_size();
3458        root_volume(fs.clone())
3459            .await
3460            .expect("root_volume failed")
3461            .new_volume("test", NewChildStoreOptions::default())
3462            .await
3463            .expect("volume failed");
3464        fs.close().await.expect("close failed");
3465        let device = fs.take_device().await;
3466        device.reopen(false);
3467
3468        // To test trim, we open the filesystem and set up a post commit hook that runs after every
3469        // transaction.  When the hook triggers, we can fsck the volume, take a snapshot of the
3470        // device and check that it gets replayed correctly on the snapshot.  We can check that the
3471        // graveyard trims the file as expected.
3472        #[derive(Default)]
3473        struct Context {
3474            store: Option<Arc<ObjectStore>>,
3475            object_id: Option<u64>,
3476        }
3477        let shared_context = Arc::new(Mutex::new(Context::default()));
3478
3479        let object_size = (TRANSACTION_MUTATION_THRESHOLD as u64 + 10) * 2 * block_size;
3480
3481        // Wait for an object to get tombstoned by the graveyard.
3482        async fn expect_tombstoned(store: &Arc<ObjectStore>, object_id: u64) {
3483            loop {
3484                if let Err(e) =
3485                    ObjectStore::open_object(store, object_id, HandleOptions::default(), None).await
3486                {
3487                    assert!(
3488                        FxfsError::NotFound.matches(&e),
3489                        "open_object didn't fail with NotFound: {:?}",
3490                        e
3491                    );
3492                    break;
3493                }
3494                // The graveyard should eventually tombstone the object.
3495                fasync::Timer::new(std::time::Duration::from_millis(100)).await;
3496            }
3497        }
3498
3499        // Checks to see if the object needs to be trimmed.
3500        async fn needs_trim(store: &Arc<ObjectStore>) -> Option<DataObjectHandle<ObjectStore>> {
3501            let root_directory = Directory::open(store, store.root_directory_object_id())
3502                .await
3503                .expect("open failed");
3504            let oid = root_directory.lookup("foo").await.expect("lookup failed");
3505            if let Some((oid, _, _)) = oid {
3506                let object = ObjectStore::open_object(store, oid, HandleOptions::default(), None)
3507                    .await
3508                    .expect("open_object failed");
3509                let props = object.get_properties().await.expect("get_properties failed");
3510                if props.allocated_size > 0 && props.data_attribute_size == 0 {
3511                    Some(object)
3512                } else {
3513                    None
3514                }
3515            } else {
3516                None
3517            }
3518        }
3519
3520        let shared_context_clone = shared_context.clone();
3521        let post_commit = move || {
3522            let store = shared_context_clone.lock().store.as_ref().cloned().unwrap();
3523            let shared_context = shared_context_clone.clone();
3524            async move {
3525                // First run fsck on the current filesystem.
3526                let options = FsckOptions {
3527                    fail_on_warning: true,
3528                    no_lock: true,
3529                    on_error: Box::new(|err| println!("fsck error: {:?}", err)),
3530                    ..Default::default()
3531                };
3532                let fs = store.filesystem();
3533
3534                fsck_with_options(fs.clone(), &options).await.expect("fsck_with_options failed");
3535                fsck_volume_with_options(fs.as_ref(), &options, store.store_object_id(), None)
3536                    .await
3537                    .expect("fsck_volume_with_options failed");
3538
3539                // Now check that we can replay this correctly.
3540                fs.sync(SyncOptions { flush_device: true, ..Default::default() })
3541                    .await
3542                    .expect("sync failed");
3543                let device = fs.device().snapshot().expect("snapshot failed");
3544
3545                let object_id = shared_context.lock().object_id.clone();
3546
3547                let fs2 = FxFilesystemBuilder::new()
3548                    .skip_initial_reap(object_id.is_none())
3549                    .open(device)
3550                    .await
3551                    .expect("open failed");
3552
3553                // If the "foo" file exists check that allocated size matches content size.
3554                let root_vol = root_volume(fs2.clone()).await.expect("root_volume failed");
3555                let store =
3556                    root_vol.volume("test", StoreOptions::default()).await.expect("volume failed");
3557
3558                if let Some(oid) = object_id {
3559                    // For the second pass, the object should get tombstoned.
3560                    expect_tombstoned(&store, oid).await;
3561                } else if let Some(object) = needs_trim(&store).await {
3562                    // Extend the file and make sure that it is correctly trimmed.
3563                    object.truncate(object_size).await.expect("truncate failed");
3564                    let mut buf = object.allocate_buffer(block_size as usize).await;
3565                    object
3566                        .read(object_size - block_size * 2, buf.as_mut())
3567                        .await
3568                        .expect("read failed");
3569                    assert_eq!(buf.to_vec(), vec![0; block_size as usize]);
3570
3571                    // Remount, this time with the graveyard performing an initial reap and the
3572                    // object should get trimmed.
3573                    let fs = FxFilesystem::open(fs.device().snapshot().expect("snapshot failed"))
3574                        .await
3575                        .expect("open failed");
3576                    let root_vol = root_volume(fs.clone()).await.expect("root_volume failed");
3577                    let store = root_vol
3578                        .volume("test", StoreOptions::default())
3579                        .await
3580                        .expect("volume failed");
3581                    while needs_trim(&store).await.is_some() {
3582                        // The object has been truncated, but still has some data allocated to
3583                        // it.  The graveyard should trim the object eventually.
3584                        fasync::Timer::new(std::time::Duration::from_millis(100)).await;
3585                    }
3586
3587                    // Run fsck.
3588                    fsck_with_options(fs.clone(), &options)
3589                        .await
3590                        .expect("fsck_with_options failed");
3591                    fsck_volume_with_options(fs.as_ref(), &options, store.store_object_id(), None)
3592                        .await
3593                        .expect("fsck_volume_with_options failed");
3594                    fs.close().await.expect("close failed");
3595                }
3596
3597                // Run fsck on fs2.
3598                fsck_with_options(fs2.clone(), &options).await.expect("fsck_with_options failed");
3599                fsck_volume_with_options(fs2.as_ref(), &options, store.store_object_id(), None)
3600                    .await
3601                    .expect("fsck_volume_with_options failed");
3602                fs2.close().await.expect("close failed");
3603            }
3604            .boxed()
3605        };
3606
3607        let fs = FxFilesystemBuilder::new()
3608            .post_commit_hook(post_commit)
3609            .open(device)
3610            .await
3611            .expect("open failed");
3612
3613        let root_vol = root_volume(fs.clone()).await.expect("root_volume failed");
3614        let store = root_vol.volume("test", StoreOptions::default()).await.expect("volume failed");
3615
3616        shared_context.lock().store = Some(store.clone());
3617
3618        let root_directory =
3619            Directory::open(&store, store.root_directory_object_id()).await.expect("open failed");
3620
3621        let object;
3622        let mut transaction = fs
3623            .root_store()
3624            .new_transaction(
3625                lock_keys![LockKey::object(
3626                    store.store_object_id(),
3627                    store.root_directory_object_id()
3628                )],
3629                Options::default(),
3630            )
3631            .await
3632            .expect("new_transaction failed");
3633        object = root_directory
3634            .create_child_file(&mut transaction, "foo")
3635            .await
3636            .expect("create_object failed");
3637        transaction.commit().await.expect("commit failed");
3638
3639        let mut transaction = fs
3640            .root_store()
3641            .new_transaction(
3642                lock_keys![LockKey::object(store.store_object_id(), object.object_id())],
3643                Options::default(),
3644            )
3645            .await
3646            .expect("new_transaction failed");
3647
3648        // Two passes: first with a regular object, and then with that object moved into the
3649        // graveyard.
3650        let mut pass = 0;
3651        loop {
3652            // Create enough extents in it such that when we truncate the object it will require
3653            // more than one transaction.
3654            let mut buf = object.allocate_buffer(5).await;
3655            buf.fill(1);
3656            // Write every other block.
3657            for offset in (0..object_size).into_iter().step_by(2 * block_size as usize) {
3658                object
3659                    .txn_write(&mut transaction, offset, buf.as_ref())
3660                    .await
3661                    .expect("write failed");
3662            }
3663            transaction.commit().await.expect("commit failed");
3664            // This should take up more than one transaction.
3665            WriteObjectHandle::truncate(&object, 0).await.expect("truncate failed");
3666
3667            if pass == 1 {
3668                break;
3669            }
3670
3671            // Store the object ID so that we can make sure the object is always tombstoned
3672            // after remount (see above).
3673            shared_context.lock().object_id = Some(object.object_id());
3674
3675            transaction = fs
3676                .root_store()
3677                .new_transaction(
3678                    lock_keys![
3679                        LockKey::object(store.store_object_id(), store.root_directory_object_id()),
3680                        LockKey::object(store.store_object_id(), object.object_id()),
3681                    ],
3682                    Options::default(),
3683                )
3684                .await
3685                .expect("new_transaction failed");
3686
3687            // Move the object into the graveyard.
3688            replace_child(&mut transaction, None, (&root_directory, "foo"))
3689                .await
3690                .expect("replace_child failed");
3691            store.add_to_graveyard(&mut transaction, object.object_id());
3692
3693            pass += 1;
3694        }
3695
3696        fs.close().await.expect("Close failed");
3697    }
3698
3699    #[fuchsia::test]
3700    async fn test_adjust_refs() {
3701        let (fs, object) = test_filesystem_and_object().await;
3702        let store = object.owner();
3703        let mut transaction = fs
3704            .root_store()
3705            .new_transaction(
3706                lock_keys![LockKey::object(store.store_object_id(), object.object_id())],
3707                Options::default(),
3708            )
3709            .await
3710            .expect("new_transaction failed");
3711        assert_eq!(
3712            store
3713                .adjust_refs(&mut transaction, object.object_id(), 1)
3714                .await
3715                .expect("adjust_refs failed"),
3716            false
3717        );
3718        transaction.commit().await.expect("commit failed");
3719
3720        let allocator = fs.allocator();
3721        let allocated_before = allocator.get_allocated_bytes();
3722        let mut transaction = fs
3723            .root_store()
3724            .new_transaction(
3725                lock_keys![LockKey::object(store.store_object_id(), object.object_id())],
3726                Options::default(),
3727            )
3728            .await
3729            .expect("new_transaction failed");
3730        assert_eq!(
3731            store
3732                .adjust_refs(&mut transaction, object.object_id(), -2)
3733                .await
3734                .expect("adjust_refs failed"),
3735            true
3736        );
3737        transaction.commit().await.expect("commit failed");
3738
3739        assert_eq!(allocator.get_allocated_bytes(), allocated_before);
3740
3741        store
3742            .tombstone_object(
3743                object.object_id(),
3744                Options { borrow_metadata_space: true, ..Default::default() },
3745            )
3746            .await
3747            .expect("purge failed");
3748
3749        assert_eq!(allocated_before - allocator.get_allocated_bytes(), fs.block_size() as u64);
3750
3751        // We need to remove the directory entry, too, otherwise fsck will complain
3752        {
3753            let mut transaction = fs
3754                .root_store()
3755                .new_transaction(
3756                    lock_keys![LockKey::object(
3757                        store.store_object_id(),
3758                        store.root_directory_object_id()
3759                    )],
3760                    Options::default(),
3761                )
3762                .await
3763                .expect("new_transaction failed");
3764            let root_directory = Directory::open(&store, store.root_directory_object_id())
3765                .await
3766                .expect("open failed");
3767            transaction.add(
3768                store.store_object_id(),
3769                Mutation::replace_or_insert_object(
3770                    ObjectKey::child(root_directory.object_id(), TEST_OBJECT_NAME, DirType::Normal),
3771                    ObjectValue::None,
3772                ),
3773            );
3774            transaction.commit().await.expect("commit failed");
3775        }
3776
3777        fsck_with_options(
3778            fs.clone(),
3779            &FsckOptions {
3780                fail_on_warning: true,
3781                on_error: Box::new(|err| println!("fsck error: {:?}", err)),
3782                ..Default::default()
3783            },
3784        )
3785        .await
3786        .expect("fsck_with_options failed");
3787
3788        fs.close().await.expect("Close failed");
3789    }
3790
3791    #[fuchsia::test]
3792    async fn test_locks() {
3793        let (fs, object) = test_filesystem_and_object().await;
3794        let (send1, recv1) = channel();
3795        let (send2, recv2) = channel();
3796        let (send3, recv3) = channel();
3797        let done = Mutex::new(false);
3798        let mut futures = FuturesUnordered::new();
3799        futures.push(
3800            async {
3801                let mut t = object.new_transaction().await.expect("new_transaction failed");
3802                send1.send(()).unwrap(); // Tell the next future to continue.
3803                send3.send(()).unwrap(); // Tell the last future to continue.
3804                recv2.await.unwrap();
3805                let mut buf = object.allocate_buffer(5).await;
3806                buf.copy_from_slice(b"hello");
3807                object.txn_write(&mut t, 0, buf.as_ref()).await.expect("write failed");
3808                // This is a halting problem so all we can do is sleep.
3809                fasync::Timer::new(Duration::from_millis(100)).await;
3810                assert!(!*done.lock());
3811                t.commit().await.expect("commit failed");
3812            }
3813            .boxed(),
3814        );
3815        futures.push(
3816            async {
3817                recv1.await.unwrap();
3818                // Reads should not block.
3819                let offset = TEST_DATA_OFFSET as usize;
3820                let align = offset % fs.block_size() as usize;
3821                let len = TEST_DATA.len();
3822                let mut buf = object.allocate_buffer(align + len).await;
3823                assert_eq!(
3824                    object.read((offset - align) as u64, buf.as_mut()).await.expect("read failed"),
3825                    align + TEST_DATA.len()
3826                );
3827                assert_eq!(&buf.as_ptr_slice().subslice(align..buf.len()).to_vec()[..], TEST_DATA);
3828                // Tell the first future to continue.
3829                send2.send(()).unwrap();
3830            }
3831            .boxed(),
3832        );
3833        futures.push(
3834            async {
3835                // This should block until the first future has completed.
3836                recv3.await.unwrap();
3837                let _t = object.new_transaction().await.expect("new_transaction failed");
3838                let mut buf = object.allocate_buffer(5).await;
3839                assert_eq!(object.read(0, buf.as_mut()).await.expect("read failed"), 5);
3840                assert_eq!(buf.to_vec(), b"hello");
3841            }
3842            .boxed(),
3843        );
3844        while let Some(()) = futures.next().await {}
3845        fs.close().await.expect("Close failed");
3846    }
3847
3848    #[fuchsia::test(threads = 10)]
3849    async fn test_racy_reads() {
3850        let fs = test_filesystem().await;
3851        let object;
3852        let mut transaction = fs
3853            .root_store()
3854            .new_transaction(lock_keys![], Options::default())
3855            .await
3856            .expect("new_transaction failed");
3857        let store = fs.root_store();
3858        object = Arc::new(
3859            ObjectStore::create_object(&store, &mut transaction, HandleOptions::default(), None)
3860                .await
3861                .expect("create_object failed"),
3862        );
3863        transaction.commit().await.expect("commit failed");
3864        for _ in 0..100 {
3865            let cloned_object = object.clone();
3866            let writer = fasync::Task::spawn(async move {
3867                let mut buf = cloned_object.allocate_buffer(10).await;
3868                buf.fill(123);
3869                cloned_object.write_or_append(Some(0), buf.as_ref()).await.expect("write failed");
3870            });
3871            let cloned_object = object.clone();
3872            let reader = fasync::Task::spawn(async move {
3873                let wait_time = rand::random_range(0..5);
3874                fasync::Timer::new(Duration::from_millis(wait_time)).await;
3875                let mut buf = cloned_object.allocate_buffer(10).await;
3876                buf.fill(23);
3877                let amount = cloned_object.read(0, buf.as_mut()).await.expect("write failed");
3878                // If we succeed in reading data, it must include the write; i.e. if we see the size
3879                // change, we should see the data too.  For this to succeed it requires locking on
3880                // the read size to ensure that when we read the size, we get the extents changed in
3881                // that same transaction.
3882                if amount != 0 {
3883                    assert_eq!(amount, 10);
3884                    assert_eq!(buf.to_vec(), &[123; 10]);
3885                }
3886            });
3887            writer.await;
3888            reader.await;
3889            object.truncate(0).await.expect("truncate failed");
3890        }
3891        fs.close().await.expect("Close failed");
3892    }
3893
3894    #[fuchsia::test]
3895    async fn test_allocated_size() {
3896        let (fs, object) = test_filesystem_and_object_with_key(None, true).await;
3897
3898        let before = object.get_properties().await.expect("get_properties failed").allocated_size;
3899        let mut buf = object.allocate_buffer(5).await;
3900        buf.copy_from_slice(b"hello");
3901        object.write_or_append(Some(0), buf.as_ref()).await.expect("write failed");
3902        let after = object.get_properties().await.expect("get_properties failed").allocated_size;
3903        assert_eq!(after, before + fs.block_size() as u64);
3904
3905        // Do the same write again and there should be no change.
3906        object.write_or_append(Some(0), buf.as_ref()).await.expect("write failed");
3907        assert_eq!(
3908            object.get_properties().await.expect("get_properties failed").allocated_size,
3909            after
3910        );
3911
3912        // extend...
3913        let mut transaction = object.new_transaction().await.expect("new_transaction failed");
3914        let offset = 1000 * fs.block_size() as u64;
3915        let before = after;
3916        object
3917            .extend(&mut transaction, offset..offset + fs.block_size() as u64)
3918            .await
3919            .expect("extend failed");
3920        transaction.commit().await.expect("commit failed");
3921        let after = object.get_properties().await.expect("get_properties failed").allocated_size;
3922        assert_eq!(after, before + fs.block_size() as u64);
3923
3924        // truncate...
3925        let before = after;
3926        let size = object.get_size();
3927        object.truncate(size - fs.block_size() as u64).await.expect("extend failed");
3928        let after = object.get_properties().await.expect("get_properties failed").allocated_size;
3929        assert_eq!(after, before - fs.block_size() as u64);
3930
3931        // preallocate_range...
3932        let mut transaction = object.new_transaction().await.expect("new_transaction failed");
3933        let before = after;
3934        let mut file_range = offset..offset + fs.block_size() as u64;
3935        object.preallocate_range(&mut transaction, &mut file_range).await.expect("extend failed");
3936        transaction.commit().await.expect("commit failed");
3937        let after = object.get_properties().await.expect("get_properties failed").allocated_size;
3938        assert_eq!(after, before + fs.block_size() as u64);
3939        fs.close().await.expect("Close failed");
3940    }
3941
3942    #[fuchsia::test(threads = 10)]
3943    async fn test_zero() {
3944        let (fs, object) = test_filesystem_and_object().await;
3945        let expected_size = object.get_size();
3946        let mut transaction = object.new_transaction().await.expect("new_transaction failed");
3947        object.zero(&mut transaction, 0..fs.block_size() as u64 * 10).await.expect("zero failed");
3948        transaction.commit().await.expect("commit failed");
3949        assert_eq!(object.get_size(), expected_size);
3950        let mut buf = object.allocate_buffer(fs.block_size() as usize * 10).await;
3951        assert_eq!(object.read(0, buf.as_mut()).await.expect("read failed") as u64, expected_size);
3952        assert_eq!(
3953            &buf.as_ptr_slice().subslice(0..expected_size as usize).to_vec()[..],
3954            vec![0u8; expected_size as usize].as_slice()
3955        );
3956        fs.close().await.expect("Close failed");
3957    }
3958
3959    #[fuchsia::test]
3960    async fn test_properties() {
3961        let (fs, object) = test_filesystem_and_object().await;
3962        const CRTIME: Timestamp = Timestamp::from_nanos(1234);
3963        const MTIME: Timestamp = Timestamp::from_nanos(5678);
3964        const CTIME: Timestamp = Timestamp::from_nanos(8765);
3965
3966        // ObjectProperties can be updated through `update_attributes`.
3967        // `get_properties` should reflect the latest changes.
3968        let mut transaction = object.new_transaction().await.expect("new_transaction failed");
3969        object
3970            .update_attributes(
3971                &mut transaction,
3972                Some(&fio::MutableNodeAttributes {
3973                    creation_time: Some(CRTIME.as_nanos()),
3974                    modification_time: Some(MTIME.as_nanos()),
3975                    mode: Some(111),
3976                    gid: Some(222),
3977                    ..Default::default()
3978                }),
3979                None,
3980            )
3981            .await
3982            .expect("update_attributes failed");
3983        const MTIME_NEW: Timestamp = Timestamp::from_nanos(12345678);
3984        object
3985            .update_attributes(
3986                &mut transaction,
3987                Some(&fio::MutableNodeAttributes {
3988                    modification_time: Some(MTIME_NEW.as_nanos()),
3989                    gid: Some(333),
3990                    rdev: Some(444),
3991                    ..Default::default()
3992                }),
3993                Some(CTIME),
3994            )
3995            .await
3996            .expect("update_timestamps failed");
3997        transaction.commit().await.expect("commit failed");
3998
3999        let properties = object.get_properties().await.expect("get_properties failed");
4000        assert_matches!(
4001            properties,
4002            ObjectProperties {
4003                refs: 1u64,
4004                allocated_size: TEST_OBJECT_ALLOCATED_SIZE,
4005                data_attribute_size: TEST_OBJECT_SIZE,
4006                creation_time: CRTIME,
4007                modification_time: MTIME_NEW,
4008                posix_attributes: Some(PosixAttributes { mode: 111, gid: 333, rdev: 444, .. }),
4009                change_time: CTIME,
4010                ..
4011            }
4012        );
4013        fs.close().await.expect("Close failed");
4014    }
4015
4016    #[fuchsia::test]
4017    async fn test_is_allocated() {
4018        let (fs, object) = test_filesystem_and_object().await;
4019
4020        // `test_filesystem_and_object()` wrote the buffer `TEST_DATA` to the device at offset
4021        // `TEST_DATA_OFFSET` where the length and offset are aligned to the block size.
4022        let aligned_offset = round_down(TEST_DATA_OFFSET, fs.block_size());
4023        let aligned_length = round_up(TEST_DATA.len() as u64, fs.block_size()).unwrap();
4024
4025        // Check for the case where where we have the following extent layout
4026        //       [ unallocated ][ `TEST_DATA` ]
4027        // The extents before `aligned_offset` should not be allocated
4028        let (allocated, count) = object.is_allocated(0).await.expect("is_allocated failed");
4029        assert_eq!(count, aligned_offset);
4030        assert_eq!(allocated, false);
4031
4032        let (allocated, count) =
4033            object.is_allocated(aligned_offset).await.expect("is_allocated failed");
4034        assert_eq!(count, aligned_length);
4035        assert_eq!(allocated, true);
4036
4037        // Check for the case where where we query out of range
4038        let end = aligned_offset + aligned_length;
4039        object
4040            .is_allocated(end)
4041            .await
4042            .expect_err("is_allocated should have returned ERR_OUT_OF_RANGE");
4043
4044        // Check for the case where where we start querying for allocation starting from
4045        // an allocated range to the end of the device
4046        let size = 50 * fs.block_size() as u64;
4047        object.truncate(size).await.expect("extend failed");
4048
4049        let (allocated, count) = object.is_allocated(end).await.expect("is_allocated failed");
4050        assert_eq!(count, size - end);
4051        assert_eq!(allocated, false);
4052
4053        // Check for the case where where we have the following extent layout
4054        //      [ unallocated ][ `buf` ][ `buf` ]
4055        let buf_length = 5 * fs.block_size();
4056        let mut buf = object.allocate_buffer(buf_length as usize).await;
4057        buf.fill(123);
4058        let new_offset = end + 20 * fs.block_size() as u64;
4059        object.write_or_append(Some(new_offset), buf.as_ref()).await.expect("write failed");
4060        object
4061            .write_or_append(Some(new_offset + buf_length), buf.as_ref())
4062            .await
4063            .expect("write failed");
4064
4065        let (allocated, count) = object.is_allocated(end).await.expect("is_allocated failed");
4066        assert_eq!(count, new_offset - end);
4067        assert_eq!(allocated, false);
4068
4069        let (allocated, count) =
4070            object.is_allocated(new_offset).await.expect("is_allocated failed");
4071        assert_eq!(count, 2 * buf_length);
4072        assert_eq!(allocated, true);
4073
4074        // Check the case where we query from the middle of an extent
4075        let (allocated, count) = object
4076            .is_allocated(new_offset + 4 * fs.block_size())
4077            .await
4078            .expect("is_allocated failed");
4079        assert_eq!(count, 2 * buf_length - 4 * fs.block_size());
4080        assert_eq!(allocated, true);
4081
4082        // Now, write buffer to a location already written to.
4083        // Check for the case when we the following extent layout
4084        //      [ unallocated ][ `other_buf` ][ (part of) `buf` ][ `buf` ]
4085        let other_buf_length = 3 * fs.block_size();
4086        let mut other_buf = object.allocate_buffer(other_buf_length as usize).await;
4087        other_buf.fill(231);
4088        object.write_or_append(Some(new_offset), other_buf.as_ref()).await.expect("write failed");
4089
4090        // We still expect that `is_allocated(..)` will return that  there are 2*`buf_length bytes`
4091        // allocated from `new_offset`
4092        let (allocated, count) =
4093            object.is_allocated(new_offset).await.expect("is_allocated failed");
4094        assert_eq!(count, 2 * buf_length);
4095        assert_eq!(allocated, true);
4096
4097        // Check for the case when we the following extent layout
4098        //   [ unallocated ][ deleted ][ unallocated ][ deleted ][ allocated ]
4099        // Mark TEST_DATA as deleted
4100        let mut transaction = object.new_transaction().await.expect("new_transaction failed");
4101        object
4102            .zero(&mut transaction, aligned_offset..aligned_offset + aligned_length)
4103            .await
4104            .expect("zero failed");
4105        // Mark `other_buf` as deleted
4106        object
4107            .zero(&mut transaction, new_offset..new_offset + buf_length)
4108            .await
4109            .expect("zero failed");
4110        transaction.commit().await.expect("commit transaction failed");
4111
4112        let (allocated, count) = object.is_allocated(0).await.expect("is_allocated failed");
4113        assert_eq!(count, new_offset + buf_length);
4114        assert_eq!(allocated, false);
4115
4116        let (allocated, count) =
4117            object.is_allocated(new_offset + buf_length).await.expect("is_allocated failed");
4118        assert_eq!(count, buf_length);
4119        assert_eq!(allocated, true);
4120
4121        let new_end = new_offset + buf_length + count;
4122
4123        // Check for the case where there are objects with different keys.
4124        // Case that we're checking for:
4125        //      [ unallocated ][ extent (object with different key) ][ unallocated ]
4126        let store = object.owner();
4127        let mut transaction = fs
4128            .root_store()
4129            .new_transaction(lock_keys![], Options::default())
4130            .await
4131            .expect("new_transaction failed");
4132        let object2 =
4133            ObjectStore::create_object(&store, &mut transaction, HandleOptions::default(), None)
4134                .await
4135                .expect("create_object failed");
4136        transaction.commit().await.expect("commit failed");
4137
4138        object2
4139            .write_or_append(Some(new_end + fs.block_size()), buf.as_ref())
4140            .await
4141            .expect("write failed");
4142
4143        // Expecting that the extent with a different key is treated like unallocated extent
4144        let (allocated, count) = object.is_allocated(new_end).await.expect("is_allocated failed");
4145        assert_eq!(count, size - new_end);
4146        assert_eq!(allocated, false);
4147
4148        fs.close().await.expect("close failed");
4149    }
4150
4151    #[fuchsia::test(threads = 10)]
4152    async fn test_read_write_attr() {
4153        let (_fs, object) = test_filesystem_and_object().await;
4154        let data = [0xffu8; 16_384];
4155        object.write_attr(AttributeId(20), &data).await.expect("write_attr failed");
4156        let rdata = object
4157            .read_attr(AttributeId(20))
4158            .await
4159            .expect("read_attr failed")
4160            .expect("no attribute data found");
4161        assert_eq!(&data[..], &rdata[..]);
4162
4163        assert_eq!(object.read_attr(AttributeId(21)).await.expect("read_attr failed"), None);
4164    }
4165
4166    #[fuchsia::test(threads = 10)]
4167    async fn test_allocate_basic() {
4168        let (fs, object) = test_filesystem_and_empty_object().await;
4169        let block_size = fs.block_size();
4170        let file_size = block_size * 10;
4171        object.truncate(file_size).await.unwrap();
4172
4173        let small_buf_size = 1024;
4174        let large_buf_aligned_size = block_size as usize * 2;
4175        let large_buf_size = block_size as usize * 2 + 1024;
4176
4177        let mut small_buf = object.allocate_buffer(small_buf_size).await;
4178        let mut large_buf_aligned = object.allocate_buffer(large_buf_aligned_size).await;
4179        let mut large_buf = object.allocate_buffer(large_buf_size).await;
4180
4181        assert_eq!(object.read(0, small_buf.as_mut()).await.unwrap(), small_buf_size);
4182        assert_eq!(small_buf.to_vec(), vec![0; small_buf_size]);
4183        assert_eq!(object.read(0, large_buf.as_mut()).await.unwrap(), large_buf_size);
4184        assert_eq!(large_buf.to_vec(), vec![0; large_buf_size]);
4185        assert_eq!(
4186            object.read(0, large_buf_aligned.as_mut()).await.unwrap(),
4187            large_buf_aligned_size
4188        );
4189        assert_eq!(large_buf_aligned.to_vec(), vec![0; large_buf_aligned_size]);
4190
4191        // Allocation succeeds, and without any writes to the location it shows up as zero.
4192        object.allocate(block_size..block_size * 3).await.unwrap();
4193
4194        // Test starting before, inside, and after the allocated section with every sized buffer.
4195        for (buf_index, buf) in [small_buf, large_buf, large_buf_aligned].iter_mut().enumerate() {
4196            for offset in 0..4 {
4197                assert_eq!(
4198                    object.read(block_size * offset, buf.as_mut()).await.unwrap(),
4199                    buf.len(),
4200                    "buf_index: {}, read offset: {}",
4201                    buf_index,
4202                    offset,
4203                );
4204                assert_eq!(
4205                    &buf.to_vec(),
4206                    &vec![0; buf.len()],
4207                    "buf_index: {}, read offset: {}",
4208                    buf_index,
4209                    offset,
4210                );
4211            }
4212        }
4213
4214        fs.close().await.expect("close failed");
4215    }
4216
4217    #[fuchsia::test(threads = 10)]
4218    async fn test_allocate_extends_file() {
4219        const BUF_SIZE: usize = 1024;
4220        let (fs, object) = test_filesystem_and_empty_object().await;
4221        let mut buf = object.allocate_buffer(BUF_SIZE).await;
4222        let block_size = fs.block_size();
4223
4224        assert_eq!(object.read(0, buf.as_mut()).await.unwrap(), buf.len());
4225        assert_eq!(buf.to_vec(), &[0; BUF_SIZE]);
4226
4227        assert!(TEST_OBJECT_SIZE < block_size * 4);
4228        // Allocation succeeds, and without any writes to the location it shows up as zero.
4229        object.allocate(0..block_size * 4).await.unwrap();
4230        assert_eq!(object.read(0, buf.as_mut()).await.unwrap(), buf.len());
4231        assert_eq!(buf.to_vec(), &[0; BUF_SIZE]);
4232        assert_eq!(object.read(block_size, buf.as_mut()).await.unwrap(), buf.len());
4233        assert_eq!(buf.to_vec(), &[0; BUF_SIZE]);
4234        assert_eq!(object.read(block_size * 3, buf.as_mut()).await.unwrap(), buf.len());
4235        assert_eq!(buf.to_vec(), &[0; BUF_SIZE]);
4236
4237        fs.close().await.expect("close failed");
4238    }
4239
4240    #[fuchsia::test(threads = 10)]
4241    async fn test_allocate_past_end() {
4242        const BUF_SIZE: usize = 1024;
4243        let (fs, object) = test_filesystem_and_empty_object().await;
4244        let mut buf = object.allocate_buffer(BUF_SIZE).await;
4245        let block_size = fs.block_size();
4246
4247        assert_eq!(object.read(0, buf.as_mut()).await.unwrap(), buf.len());
4248        assert_eq!(buf.to_vec(), &[0; BUF_SIZE]);
4249
4250        assert!(TEST_OBJECT_SIZE < block_size * 4);
4251        // Allocation succeeds, and without any writes to the location it shows up as zero.
4252        object.allocate(block_size * 4..block_size * 6).await.unwrap();
4253        assert_eq!(object.read(0, buf.as_mut()).await.unwrap(), buf.len());
4254        assert_eq!(buf.to_vec(), &[0; BUF_SIZE]);
4255        assert_eq!(object.read(block_size * 4, buf.as_mut()).await.unwrap(), buf.len());
4256        assert_eq!(buf.to_vec(), &[0; BUF_SIZE]);
4257        assert_eq!(object.read(block_size * 5, buf.as_mut()).await.unwrap(), buf.len());
4258        assert_eq!(buf.to_vec(), &[0; BUF_SIZE]);
4259
4260        fs.close().await.expect("close failed");
4261    }
4262
4263    #[fuchsia::test(threads = 10)]
4264    async fn test_allocate_read_attr() {
4265        let (fs, object) = test_filesystem_and_empty_object().await;
4266        let block_size = fs.block_size();
4267        let file_size = block_size * 4;
4268        object.truncate(file_size).await.unwrap();
4269
4270        let content = object
4271            .read_attr(object.attribute_id())
4272            .await
4273            .expect("failed to read attr")
4274            .expect("attr returned none");
4275        assert_eq!(content.as_ref(), &vec![0; file_size as usize]);
4276
4277        object.allocate(block_size..block_size * 3).await.unwrap();
4278
4279        let content = object
4280            .read_attr(object.attribute_id())
4281            .await
4282            .expect("failed to read attr")
4283            .expect("attr returned none");
4284        assert_eq!(content.as_ref(), &vec![0; file_size as usize]);
4285
4286        fs.close().await.expect("close failed");
4287    }
4288
4289    #[fuchsia::test(threads = 10)]
4290    async fn test_allocate_existing_data() {
4291        struct Case {
4292            written_ranges: Vec<Range<usize>>,
4293            allocate_range: Range<u64>,
4294        }
4295        let cases = [
4296            Case { written_ranges: vec![4..7], allocate_range: 4..7 },
4297            Case { written_ranges: vec![4..7], allocate_range: 3..8 },
4298            Case { written_ranges: vec![4..7], allocate_range: 5..6 },
4299            Case { written_ranges: vec![4..7], allocate_range: 5..8 },
4300            Case { written_ranges: vec![4..7], allocate_range: 3..5 },
4301            Case { written_ranges: vec![0..1, 2..3, 4..5, 6..7, 8..9], allocate_range: 0..10 },
4302            Case { written_ranges: vec![0..2, 4..6, 7..10], allocate_range: 1..8 },
4303        ];
4304
4305        for case in cases {
4306            let (fs, object) = test_filesystem_and_empty_object().await;
4307            let block_size = fs.block_size();
4308            let file_size = block_size * 10;
4309            object.truncate(file_size).await.unwrap();
4310
4311            for write in &case.written_ranges {
4312                let write_len = (write.end - write.start) * block_size as usize;
4313                let mut write_buf = object.allocate_buffer(write_len).await;
4314                write_buf.fill(0xff);
4315                assert_eq!(
4316                    object
4317                        .write_or_append(Some(block_size * write.start as u64), write_buf.as_ref())
4318                        .await
4319                        .unwrap(),
4320                    file_size
4321                );
4322            }
4323
4324            let mut expected_buf = object.allocate_buffer(file_size as usize).await;
4325            assert_eq!(object.read(0, expected_buf.as_mut()).await.unwrap(), expected_buf.len());
4326
4327            object
4328                .allocate(
4329                    case.allocate_range.start * block_size..case.allocate_range.end * block_size,
4330                )
4331                .await
4332                .unwrap();
4333
4334            let mut read_buf = object.allocate_buffer(file_size as usize).await;
4335            assert_eq!(object.read(0, read_buf.as_mut()).await.unwrap(), read_buf.len());
4336            assert_eq!(read_buf.to_vec(), expected_buf.to_vec());
4337
4338            fs.close().await.expect("close failed");
4339        }
4340    }
4341
4342    async fn get_modes(
4343        obj: &DataObjectHandle<ObjectStore>,
4344        mut search_range: Range<u64>,
4345    ) -> Vec<(Range<u64>, ExtentMode)> {
4346        let mut modes = Vec::new();
4347        let store = obj.store();
4348        let tree = store.tree();
4349        let layer_set = tree.layer_set();
4350        let mut merger = layer_set.merger();
4351        let mut iter = merger
4352            .query(Query::FullRange(&ObjectKey::attribute(
4353                obj.object_id(),
4354                AttributeId::DATA,
4355                AttributeKey::Extent(Extent::search_key_from_offset(search_range.start)),
4356            )))
4357            .await
4358            .unwrap();
4359        loop {
4360            match iter.get() {
4361                Some(ItemRef {
4362                    key:
4363                        ObjectKey {
4364                            object_id,
4365                            data:
4366                                ObjectKeyData::Attribute(
4367                                    AttributeId::DATA,
4368                                    AttributeKey::Extent(extent),
4369                                ),
4370                        },
4371                    value: ObjectValue::Extent(ExtentValue::Some { mode, .. }),
4372                    ..
4373                }) if *object_id == obj.object_id() => {
4374                    if search_range.end <= extent.start {
4375                        break;
4376                    }
4377                    let found_range = std::cmp::max(search_range.start, extent.start)
4378                        ..std::cmp::min(search_range.end, extent.end);
4379                    search_range.start = found_range.end;
4380                    modes.push((found_range, mode.clone()));
4381                    if search_range.start == search_range.end {
4382                        break;
4383                    }
4384                    iter.advance().await.unwrap();
4385                }
4386                x => panic!("looking for extent record, found this {:?}", x),
4387            }
4388        }
4389        modes
4390    }
4391
4392    async fn assert_all_overwrite(
4393        obj: &DataObjectHandle<ObjectStore>,
4394        mut search_range: Range<u64>,
4395    ) {
4396        let modes = get_modes(obj, search_range.clone()).await;
4397        for mode in modes {
4398            assert_eq!(
4399                mode.0.start, search_range.start,
4400                "missing mode in range {}..{}",
4401                search_range.start, mode.0.start
4402            );
4403            match mode.1 {
4404                ExtentMode::Overwrite | ExtentMode::OverwritePartial(_) => (),
4405                m => panic!("mode at range {:?} was not overwrite, instead found {:?}", mode.0, m),
4406            }
4407            assert!(
4408                mode.0.end <= search_range.end,
4409                "mode ends beyond search range (bug in test) - search_range: {:?}, mode: {:?}",
4410                search_range,
4411                mode,
4412            );
4413            search_range.start = mode.0.end;
4414        }
4415        assert_eq!(
4416            search_range.start, search_range.end,
4417            "missing mode in range {:?}",
4418            search_range
4419        );
4420    }
4421
4422    #[fuchsia::test(threads = 10)]
4423    async fn test_multi_overwrite() {
4424        #[derive(Debug)]
4425        struct Case {
4426            pre_writes: Vec<Range<usize>>,
4427            allocate_ranges: Vec<Range<u64>>,
4428            overwrites: Vec<Vec<Range<u64>>>,
4429        }
4430        let cases = [
4431            Case {
4432                pre_writes: Vec::new(),
4433                allocate_ranges: vec![1..3],
4434                overwrites: vec![vec![1..3]],
4435            },
4436            Case {
4437                pre_writes: Vec::new(),
4438                allocate_ranges: vec![0..1, 1..2, 2..3, 3..4],
4439                overwrites: vec![vec![0..4]],
4440            },
4441            Case {
4442                pre_writes: Vec::new(),
4443                allocate_ranges: vec![0..4],
4444                overwrites: vec![vec![0..1], vec![1..2], vec![3..4]],
4445            },
4446            Case {
4447                pre_writes: Vec::new(),
4448                allocate_ranges: vec![0..4],
4449                overwrites: vec![vec![3..4]],
4450            },
4451            Case {
4452                pre_writes: Vec::new(),
4453                allocate_ranges: vec![0..4],
4454                overwrites: vec![vec![3..4], vec![2..3], vec![1..2]],
4455            },
4456            Case {
4457                pre_writes: Vec::new(),
4458                allocate_ranges: vec![1..2, 5..6, 7..8],
4459                overwrites: vec![vec![5..6]],
4460            },
4461            Case {
4462                pre_writes: Vec::new(),
4463                allocate_ranges: vec![1..3],
4464                overwrites: vec![
4465                    vec![1..3],
4466                    vec![1..3],
4467                    vec![1..3],
4468                    vec![1..3],
4469                    vec![1..3],
4470                    vec![1..3],
4471                    vec![1..3],
4472                    vec![1..3],
4473                ],
4474            },
4475            Case {
4476                pre_writes: Vec::new(),
4477                allocate_ranges: vec![0..5],
4478                overwrites: vec![
4479                    vec![1..3],
4480                    vec![1..3],
4481                    vec![1..3],
4482                    vec![1..3],
4483                    vec![1..3],
4484                    vec![1..3],
4485                    vec![1..3],
4486                    vec![1..3],
4487                ],
4488            },
4489            Case {
4490                pre_writes: Vec::new(),
4491                allocate_ranges: vec![0..5],
4492                overwrites: vec![vec![0..2, 2..4, 4..5]],
4493            },
4494            Case {
4495                pre_writes: Vec::new(),
4496                allocate_ranges: vec![0..5, 5..10],
4497                overwrites: vec![vec![1..2, 2..3, 4..7, 7..8]],
4498            },
4499            Case {
4500                pre_writes: Vec::new(),
4501                allocate_ranges: vec![0..4, 6..10],
4502                overwrites: vec![vec![2..3, 7..9]],
4503            },
4504            Case {
4505                pre_writes: Vec::new(),
4506                allocate_ranges: vec![0..10],
4507                overwrites: vec![vec![1..2, 5..10], vec![0..1, 5..10], vec![0..5, 5..10]],
4508            },
4509            Case {
4510                pre_writes: Vec::new(),
4511                allocate_ranges: vec![0..10],
4512                overwrites: vec![vec![0..2, 2..4, 4..6, 6..8, 8..10], vec![0..5, 5..10]],
4513            },
4514            Case {
4515                pre_writes: vec![1..3],
4516                allocate_ranges: vec![1..3],
4517                overwrites: vec![vec![1..3]],
4518            },
4519            Case {
4520                pre_writes: vec![1..3],
4521                allocate_ranges: vec![4..6],
4522                overwrites: vec![vec![5..6]],
4523            },
4524            Case {
4525                pre_writes: vec![1..3],
4526                allocate_ranges: vec![0..4],
4527                overwrites: vec![vec![0..4]],
4528            },
4529            Case {
4530                pre_writes: vec![1..3],
4531                allocate_ranges: vec![2..4],
4532                overwrites: vec![vec![2..4]],
4533            },
4534            Case {
4535                pre_writes: vec![3..5],
4536                allocate_ranges: vec![1..3, 6..7],
4537                overwrites: vec![vec![1..3, 6..7]],
4538            },
4539            Case {
4540                pre_writes: vec![1..3, 5..7, 8..9],
4541                allocate_ranges: vec![0..5],
4542                overwrites: vec![vec![0..2, 2..5], vec![0..5]],
4543            },
4544            Case {
4545                pre_writes: Vec::new(),
4546                allocate_ranges: vec![0..10, 4..6],
4547                overwrites: Vec::new(),
4548            },
4549            Case {
4550                pre_writes: Vec::new(),
4551                allocate_ranges: vec![3..8, 5..10],
4552                overwrites: Vec::new(),
4553            },
4554            Case {
4555                pre_writes: Vec::new(),
4556                allocate_ranges: vec![5..10, 3..8],
4557                overwrites: Vec::new(),
4558            },
4559        ];
4560
4561        for (i, case) in cases.into_iter().enumerate() {
4562            log::info!("running case {} - {:?}", i, case);
4563            let (fs, object) = test_filesystem_and_empty_object().await;
4564            let block_size = fs.block_size();
4565            let file_size = block_size * 10;
4566            object.truncate(file_size).await.unwrap();
4567
4568            for write in case.pre_writes {
4569                let write_len = (write.end - write.start) * block_size as usize;
4570                let mut write_buf = object.allocate_buffer(write_len).await;
4571                write_buf.fill(0xff);
4572                assert_eq!(
4573                    object
4574                        .write_or_append(Some(block_size * write.start as u64), write_buf.as_ref())
4575                        .await
4576                        .unwrap(),
4577                    file_size
4578                );
4579            }
4580
4581            for allocate_range in &case.allocate_ranges {
4582                object
4583                    .allocate(allocate_range.start * block_size..allocate_range.end * block_size)
4584                    .await
4585                    .unwrap();
4586            }
4587
4588            for allocate_range in case.allocate_ranges {
4589                assert_all_overwrite(
4590                    &object,
4591                    allocate_range.start * block_size..allocate_range.end * block_size,
4592                )
4593                .await;
4594            }
4595
4596            for overwrite in case.overwrites {
4597                let mut write_len = 0;
4598                let overwrite = overwrite
4599                    .into_iter()
4600                    .map(|r| {
4601                        write_len += (r.end - r.start) * block_size;
4602                        r.start * block_size..r.end * block_size
4603                    })
4604                    .collect::<Vec<_>>();
4605                let mut write_buf = object.allocate_buffer(write_len as usize).await;
4606                let data = (0..20).cycle().take(write_len as usize).collect::<Vec<_>>();
4607                write_buf.copy_from_slice(&data);
4608
4609                let mut expected_buf = object.allocate_buffer(file_size as usize).await;
4610                assert_eq!(
4611                    object.read(0, expected_buf.as_mut()).await.unwrap(),
4612                    expected_buf.len()
4613                );
4614                let mut expected_buf_slice = expected_buf.as_mut_ptr_slice();
4615                let mut data_slice = data.as_slice();
4616                for r in &overwrite {
4617                    let len = r.length().unwrap() as usize;
4618                    let (copy_from, rest) = data_slice.split_at(len);
4619                    expected_buf_slice
4620                        .subslice_mut(r.start as usize..r.end as usize)
4621                        .copy_from_slice(&copy_from);
4622                    data_slice = rest;
4623                }
4624
4625                let mut transaction = object.new_transaction().await.unwrap();
4626                object
4627                    .multi_overwrite(
4628                        &mut transaction,
4629                        AttributeId::DATA,
4630                        &overwrite,
4631                        write_buf.as_mut(),
4632                    )
4633                    .await
4634                    .unwrap_or_else(|_| panic!("multi_overwrite error on case {}", i));
4635                // Double check the emitted checksums. We should have one u64 checksum for every
4636                // block we wrote to disk.
4637                let mut checksummed_range_length = 0;
4638                let mut num_checksums = 0;
4639                for (device_range, checksums, _) in transaction.checksums() {
4640                    let range_len = device_range.end - device_range.start;
4641                    let checksums_len = checksums.len() as u64;
4642                    assert_eq!(range_len / checksums_len, block_size);
4643                    checksummed_range_length += range_len;
4644                    num_checksums += checksums_len;
4645                }
4646                assert_eq!(checksummed_range_length, write_len);
4647                assert_eq!(num_checksums, write_len / block_size);
4648                transaction.commit().await.unwrap();
4649
4650                let mut buf = object.allocate_buffer(file_size as usize).await;
4651                assert_eq!(
4652                    object.read(0, buf.as_mut()).await.unwrap(),
4653                    buf.len(),
4654                    "failed length check on case {}",
4655                    i,
4656                );
4657                assert_eq!(buf.to_vec(), expected_buf.to_vec(), "failed on case {}", i);
4658            }
4659
4660            fsck_volume(&fs, object.store().store_object_id(), None).await.expect("fsck failed");
4661            fs.close().await.expect("close failed");
4662        }
4663    }
4664
4665    #[fuchsia::test(threads = 10)]
4666    async fn test_multi_overwrite_mode_updates() {
4667        let (fs, object) = test_filesystem_and_empty_object().await;
4668        let block_size = fs.block_size();
4669        let file_size = block_size * 10;
4670        object.truncate(file_size).await.unwrap();
4671
4672        let mut expected_bitmap = BitVec::from_elem(10, false);
4673
4674        object.allocate(0..10 * block_size).await.unwrap();
4675        assert_eq!(
4676            get_modes(&object, 0..10 * block_size).await,
4677            vec![(0..10 * block_size, ExtentMode::OverwritePartial(expected_bitmap.clone()))]
4678        );
4679
4680        let mut write_buf = object.allocate_buffer(2 * block_size as usize).await;
4681        let data = (0..20).cycle().take(write_buf.len()).collect::<Vec<_>>();
4682        write_buf.copy_from_slice(&data);
4683        let mut transaction = object.new_transaction().await.unwrap();
4684        object
4685            .multi_overwrite(
4686                &mut transaction,
4687                AttributeId::DATA,
4688                &[2 * block_size..4 * block_size],
4689                write_buf.as_mut(),
4690            )
4691            .await
4692            .unwrap();
4693        transaction.commit().await.unwrap();
4694
4695        expected_bitmap.set(2, true);
4696        expected_bitmap.set(3, true);
4697        assert_eq!(
4698            get_modes(&object, 0..10 * block_size).await,
4699            vec![(0..10 * block_size, ExtentMode::OverwritePartial(expected_bitmap.clone()))]
4700        );
4701
4702        let mut write_buf = object.allocate_buffer(3 * block_size as usize).await;
4703        let data = (0..20).cycle().take(write_buf.len()).collect::<Vec<_>>();
4704        write_buf.copy_from_slice(&data);
4705        let mut transaction = object.new_transaction().await.unwrap();
4706        object
4707            .multi_overwrite(
4708                &mut transaction,
4709                AttributeId::DATA,
4710                &[3 * block_size..5 * block_size, 6 * block_size..7 * block_size],
4711                write_buf.as_mut(),
4712            )
4713            .await
4714            .unwrap();
4715        transaction.commit().await.unwrap();
4716
4717        expected_bitmap.set(4, true);
4718        expected_bitmap.set(6, true);
4719        assert_eq!(
4720            get_modes(&object, 0..10 * block_size).await,
4721            vec![(0..10 * block_size, ExtentMode::OverwritePartial(expected_bitmap.clone()))]
4722        );
4723
4724        let mut write_buf = object.allocate_buffer(6 * block_size as usize).await;
4725        let data = (0..20).cycle().take(write_buf.len()).collect::<Vec<_>>();
4726        write_buf.copy_from_slice(&data);
4727        let mut transaction = object.new_transaction().await.unwrap();
4728        object
4729            .multi_overwrite(
4730                &mut transaction,
4731                AttributeId::DATA,
4732                &[
4733                    0..2 * block_size,
4734                    5 * block_size..6 * block_size,
4735                    7 * block_size..10 * block_size,
4736                ],
4737                write_buf.as_mut(),
4738            )
4739            .await
4740            .unwrap();
4741        transaction.commit().await.unwrap();
4742
4743        assert_eq!(
4744            get_modes(&object, 0..10 * block_size).await,
4745            vec![(0..10 * block_size, ExtentMode::Overwrite)]
4746        );
4747
4748        fs.close().await.expect("close failed");
4749    }
4750
4751    #[fuchsia::test(threads = 10)]
4752    async fn test_check_unwritten_zero() {
4753        let device = DeviceHolder::new(FakeDevice::new(256 * 1024, TEST_DEVICE_BLOCK_SIZE));
4754        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
4755        let object = create_object_with_key(fs.clone(), Some(&new_insecure_crypt()), false).await;
4756        let block_size = fs.block_size();
4757
4758        // Set up a file with eight blocks to look like this:
4759        // | None | COW | COW | None | Overwrite(unwritten) | Overwrite(written) | None |
4760        let file_size = block_size * 7;
4761        object.truncate(file_size).await.unwrap();
4762        assert!(object.check_unwritten_zero(0..file_size).await.unwrap());
4763
4764        let mut buffer = object.allocate_buffer(block_size as usize).await;
4765        buffer.fill(1);
4766        object.write_or_append(Some(block_size), buffer.as_ref()).await.expect("write failed");
4767        object.write_or_append(Some(block_size * 2), buffer.as_ref()).await.expect("write failed");
4768
4769        object.allocate((block_size * 4)..(block_size * 6)).await.expect("Allocate failed");
4770        let mut transaction = fs
4771            .root_store()
4772            .new_transaction(
4773                lock_keys![LockKey::object(object.store().store_object_id(), object.object_id(),)],
4774                Options::default(),
4775            )
4776            .await
4777            .expect("new_transaction failed");
4778        object
4779            .multi_overwrite(
4780                &mut transaction,
4781                AttributeId::DATA,
4782                &vec![(block_size * 5)..(block_size * 6)],
4783                buffer.as_mut(),
4784            )
4785            .await
4786            .expect("Multi overwrite");
4787        transaction.commit().await.expect("Committing overwrite");
4788
4789        // Anything touching the COW ranges should fail.
4790        assert!(!object.check_unwritten_zero(0..(block_size * 2)).await.unwrap());
4791        assert!(!object.check_unwritten_zero(block_size..(block_size * 3)).await.unwrap());
4792        assert!(!object.check_unwritten_zero((block_size * 2)..(block_size * 4)).await.unwrap());
4793
4794        // This should be fine, as the OverwritePartial should only touch the unwritten block.
4795        assert!(object.check_unwritten_zero((block_size * 3)..(block_size * 5)).await.unwrap());
4796
4797        // These should touch the written overwrite block and fail.
4798        assert!(!object.check_unwritten_zero((block_size * 4)..(block_size * 6)).await.unwrap());
4799        assert!(!object.check_unwritten_zero((block_size * 5)..(block_size * 7)).await.unwrap());
4800
4801        fs.close().await.expect("close failed");
4802    }
4803}