Skip to main content

fxfs/object_store/
object_manager.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::filesystem::{ApplyContext, ApplyMode, FlushReason, JournalingObject};
7use crate::log::*;
8use crate::metrics;
9use crate::object_handle::INVALID_OBJECT_ID;
10use crate::object_store::allocator::{Allocator, Reservation};
11use crate::object_store::directory::Directory;
12use crate::object_store::journal::{self, JournalCheckpoint};
13use crate::object_store::transaction::{
14    AssocObj, AssociatedObject, MetadataReservation, Mutation, ObjectMutationIterator, Transaction,
15    TxnMutation,
16};
17use crate::object_store::tree_cache::TreeCache;
18use crate::object_store::volume::{VOLUMES_DIRECTORY, list_volumes};
19use crate::object_store::{ObjectDescriptor, ObjectStore};
20use crate::round::round_div;
21use crate::serialized_types::{LATEST_VERSION, Version};
22use anyhow::{Context, Error, anyhow, bail, ensure};
23use fuchsia_inspect::{Property as _, UintProperty};
24use fuchsia_sync::RwLock;
25use futures::FutureExt as _;
26use rustc_hash::FxHashMap as HashMap;
27use std::collections::hash_map::Entry;
28use std::num::Saturating;
29use std::sync::{Arc, OnceLock};
30
31// Data written to the journal eventually needs to be flushed somewhere (typically into layer
32// files).  Here we conservatively assume that could take up to four times as much space as it does
33// in the journal.  In the layer file, it'll take up at least as much, but we must reserve the same
34// again that so that there's enough space for compactions, and then we need some spare for
35// overheads.
36//
37// TODO(https://fxbug.dev/42178158): We should come up with a better way of determining what the multiplier
38// should be here.  2x was too low, as it didn't cover any space for metadata.  4x might be too
39// much.
40pub const fn reserved_space_from_journal_usage(journal_usage: u64) -> u64 {
41    journal_usage * 4
42}
43
44/// ObjectManager is a global loading cache for object stores and other special objects.
45pub struct ObjectManager {
46    inner: RwLock<Inner>,
47    metadata_reservation: OnceLock<Reservation>,
48    volume_directory: OnceLock<Directory<ObjectStore>>,
49    on_new_store: Option<Box<dyn Fn(&ObjectStore) + Send + Sync>>,
50}
51
52// Whilst we are flushing we need to keep track of the old checkpoint that we are hoping to flush,
53// and a new one that should apply if we successfully finish the flush.
54#[derive(Debug)]
55enum Checkpoints {
56    Current(JournalCheckpoint),
57    Old(JournalCheckpoint),
58    Both(/* old: */ JournalCheckpoint, /* current: */ JournalCheckpoint),
59}
60
61impl Checkpoints {
62    // Returns the earliest checkpoint (which will always be the old one if present).
63    fn earliest(&self) -> &JournalCheckpoint {
64        match self {
65            Checkpoints::Old(x) | Checkpoints::Both(x, _) | Checkpoints::Current(x) => x,
66        }
67    }
68}
69
70// We currently maintain strong references to all stores that have been opened, but there's no
71// currently no mechanism for releasing stores that aren't being used.
72struct Inner {
73    stores: HashMap<u64, Arc<ObjectStore>>,
74    root_parent_store_object_id: u64,
75    root_store_object_id: u64,
76    allocator_object_id: u64,
77    allocator: Option<Arc<Allocator>>,
78
79    // Records dependencies on the journal for objects i.e. an entry for object ID 1, would mean it
80    // has a dependency on journal records from that offset.
81    journal_checkpoints: HashMap<u64, Checkpoints>,
82
83    // Mappings from object-id to a target reservation amount.  The object IDs here are from the
84    // root store namespace, so it can be associated with any object in the root store.  A
85    // reservation will be made to cover the *maximum* in this map, since it is assumed that any
86    // requirement is only temporary, for the duration of a compaction, and that once compaction has
87    // finished for a particular object, the space will be recovered.
88    reservations: HashMap<u64, u64>,
89
90    // The last journal end offset for a transaction that has been applied.  This is not necessarily
91    // the same as the start offset for the next transaction because of padding.
92    last_end_offset: u64,
93
94    // A running counter that tracks metadata space that has been borrowed on the understanding that
95    // eventually it will be recovered (potentially after a full compaction).
96    borrowed_metadata_space: u64,
97
98    // The maximum transaction size that has been encountered so far.
99    max_transaction_size: (u64, UintProperty),
100
101    // Extra temporary space that might be tied up in the journal that hasn't yet been deallocated.
102    reserved_space: u64,
103}
104
105impl Inner {
106    fn earliest_journal_offset(&self) -> Option<u64> {
107        self.journal_checkpoints.values().map(|c| c.earliest().file_offset).min()
108    }
109
110    // Returns the required size of the metadata reservation assuming that no space has been
111    // borrowed.  The invariant is: reservation-size + borrowed-space = required.
112    fn required_reservation(&self) -> u64 {
113        // Start with the maximum amount of temporary space we might need during compactions.
114        self.reservations.values().max().unwrap_or(&0)
115
116        // Account for data that has been written to the journal that will need to be written
117        // to layer files when flushed.
118            + self.earliest_journal_offset()
119            .map(|min| reserved_space_from_journal_usage(self.last_end_offset - min))
120            .unwrap_or(0)
121
122        // Extra reserved space
123            + self.reserved_space
124    }
125
126    fn journaling_object(&self, object_id: u64) -> Option<Arc<dyn JournalingObject>> {
127        if object_id == self.allocator_object_id {
128            Some(self.allocator.clone().unwrap() as Arc<dyn JournalingObject>)
129        } else {
130            self.stores.get(&object_id).map(|x| x.clone() as Arc<dyn JournalingObject>)
131        }
132    }
133}
134
135impl ObjectManager {
136    pub fn new(on_new_store: Option<Box<dyn Fn(&ObjectStore) + Send + Sync>>) -> ObjectManager {
137        ObjectManager {
138            inner: RwLock::new(Inner {
139                stores: HashMap::default(),
140                root_parent_store_object_id: INVALID_OBJECT_ID,
141                root_store_object_id: INVALID_OBJECT_ID,
142                allocator_object_id: INVALID_OBJECT_ID,
143                allocator: None,
144                journal_checkpoints: HashMap::default(),
145                reservations: HashMap::default(),
146                last_end_offset: 0,
147                borrowed_metadata_space: 0,
148                max_transaction_size: (0, metrics::detail().create_uint("max_transaction_size", 0)),
149                reserved_space: journal::RESERVED_SPACE,
150            }),
151            metadata_reservation: OnceLock::new(),
152            volume_directory: OnceLock::new(),
153            on_new_store,
154        }
155    }
156
157    pub fn required_reservation(&self) -> u64 {
158        self.inner.read().required_reservation()
159    }
160
161    pub fn root_parent_store_object_id(&self) -> u64 {
162        self.inner.read().root_parent_store_object_id
163    }
164
165    pub fn root_parent_store(&self) -> Arc<ObjectStore> {
166        let inner = self.inner.read();
167        inner.stores.get(&inner.root_parent_store_object_id).unwrap().clone()
168    }
169
170    pub fn set_root_parent_store(&self, store: Arc<ObjectStore>) {
171        if let Some(on_new_store) = &self.on_new_store {
172            on_new_store(&store);
173        }
174        let mut inner = self.inner.write();
175        let store_id = store.store_object_id();
176        inner.stores.insert(store_id, store);
177        inner.root_parent_store_object_id = store_id;
178    }
179
180    pub fn root_store_object_id(&self) -> u64 {
181        self.inner.read().root_store_object_id
182    }
183
184    pub fn root_store(&self) -> Arc<ObjectStore> {
185        let inner = self.inner.read();
186        inner.stores.get(&inner.root_store_object_id).unwrap().clone()
187    }
188
189    pub fn set_root_store(&self, store: Arc<ObjectStore>) {
190        if let Some(on_new_store) = &self.on_new_store {
191            on_new_store(&store);
192        }
193        let mut inner = self.inner.write();
194        let store_id = store.store_object_id();
195        inner.stores.insert(store_id, store);
196        inner.root_store_object_id = store_id;
197    }
198
199    pub fn is_system_store(&self, store_id: u64) -> bool {
200        let inner = self.inner.read();
201        store_id == inner.root_store_object_id || store_id == inner.root_parent_store_object_id
202    }
203
204    /// Returns the store which might or might not be locked.
205    pub fn store(&self, store_object_id: u64) -> Option<Arc<ObjectStore>> {
206        self.inner.read().stores.get(&store_object_id).cloned()
207    }
208
209    /// Returns the total bytes written to the LSM trees in the allocator and all object stores
210    /// during compaction operations.
211    pub fn compaction_bytes_written(&self) -> u64 {
212        let inner = self.inner.read();
213        let mut total = 0;
214        if let Some(allocator) = &inner.allocator {
215            total += allocator.tree().compaction_bytes_written();
216        }
217        for store in inner.stores.values() {
218            total += store.tree().compaction_bytes_written();
219        }
220        total
221    }
222
223    /// This is not thread-safe: it assumes that a store won't be forgotten whilst the loop is
224    /// running.  This is to be used after replaying the journal.
225    pub async fn on_replay_complete(&self) -> Result<(), Error> {
226        let root_store = self.root_store();
227
228        let root_directory = Directory::open(&root_store, root_store.root_directory_object_id())
229            .await
230            .context("Unable to open root volume directory")?;
231
232        match root_directory.lookup(VOLUMES_DIRECTORY).await? {
233            None => bail!("Root directory not found"),
234            Some((object_id, ObjectDescriptor::Directory, _)) => {
235                let volume_directory = Directory::open(&root_store, object_id)
236                    .await
237                    .context("Unable to open volumes directory")?;
238                self.volume_directory.set(volume_directory).unwrap();
239            }
240            _ => {
241                bail!(
242                    anyhow!(FxfsError::Inconsistent)
243                        .context("Unexpected type for volumes directory")
244                )
245            }
246        }
247
248        let object_ids = list_volumes(self.volume_directory.get().unwrap())
249            .await
250            .context("Failed to list volumes")?;
251
252        for store_id in object_ids {
253            self.open_store(&root_store, store_id).await?;
254        }
255
256        // This can fail if a filesystem is created and truncated to a size
257        // that doesn't leave enough free space for metadata reservations.
258        self.init_metadata_reservation()
259            .context("Insufficient free space for metadata reservation.")?;
260
261        Ok(())
262    }
263
264    pub fn volume_directory(&self) -> &Directory<ObjectStore> {
265        self.volume_directory.get().unwrap()
266    }
267
268    pub fn set_volume_directory(&self, volume_directory: Directory<ObjectStore>) {
269        self.volume_directory.set(volume_directory).unwrap();
270    }
271
272    pub fn add_store(&self, store: Arc<ObjectStore>) {
273        if let Some(on_new_store) = &self.on_new_store {
274            on_new_store(&store);
275        }
276        let mut inner = self.inner.write();
277        let store_object_id = store.store_object_id();
278        assert_ne!(store_object_id, inner.root_parent_store_object_id);
279        assert_ne!(store_object_id, inner.root_store_object_id);
280        assert_ne!(store_object_id, inner.allocator_object_id);
281        inner.stores.insert(store_object_id, store);
282    }
283
284    pub fn forget_store(&self, store_object_id: u64) {
285        let mut inner = self.inner.write();
286        assert_ne!(store_object_id, inner.allocator_object_id);
287        inner.stores.remove(&store_object_id);
288        inner.reservations.remove(&store_object_id);
289    }
290
291    pub fn set_allocator(&self, allocator: Arc<Allocator>) {
292        let mut inner = self.inner.write();
293        assert!(!inner.stores.contains_key(&allocator.object_id()));
294        inner.allocator_object_id = allocator.object_id();
295        inner.allocator = Some(allocator);
296    }
297
298    pub fn allocator(&self) -> Arc<Allocator> {
299        self.inner.read().allocator.clone().unwrap()
300    }
301
302    /// Applies `mutation` to `object` with `context`.
303    pub fn apply_mutation(
304        &self,
305        object_id: u64,
306        mutation: Mutation,
307        context: &ApplyContext<'_, '_>,
308        associated_object: AssocObj<'_>,
309    ) -> Result<(), Error> {
310        debug!(oid = object_id, mutation:?; "applying mutation");
311        let object = {
312            let mut inner = self.inner.write();
313            match mutation {
314                Mutation::BeginFlush => {
315                    if let Some(entry) = inner.journal_checkpoints.get_mut(&object_id) {
316                        match entry {
317                            Checkpoints::Current(x) | Checkpoints::Both(x, _) => {
318                                *entry = Checkpoints::Old(x.clone());
319                            }
320                            _ => {}
321                        }
322                    }
323                }
324                Mutation::EndFlush => {
325                    if let Entry::Occupied(mut o) = inner.journal_checkpoints.entry(object_id) {
326                        let entry = o.get_mut();
327                        match entry {
328                            Checkpoints::Old(_) => {
329                                o.remove();
330                            }
331                            Checkpoints::Both(_, x) => {
332                                *entry = Checkpoints::Current(x.clone());
333                            }
334                            _ => {}
335                        }
336                    }
337                }
338                Mutation::DeleteVolume => {
339                    if let Some(store) = inner.stores.remove(&object_id) {
340                        store.mark_deleted();
341                    }
342                    inner.reservations.remove(&object_id);
343                    inner.journal_checkpoints.remove(&object_id);
344                    return Ok(());
345                }
346                _ => {
347                    if object_id != inner.root_parent_store_object_id {
348                        inner
349                            .journal_checkpoints
350                            .entry(object_id)
351                            .and_modify(|entry| {
352                                if let Checkpoints::Old(x) = entry {
353                                    *entry =
354                                        Checkpoints::Both(x.clone(), context.checkpoint.clone());
355                                }
356                            })
357                            .or_insert_with(|| Checkpoints::Current(context.checkpoint.clone()));
358                    }
359                }
360            }
361            inner.journaling_object(object_id).unwrap()
362        };
363        associated_object.map(|o| o.will_apply_mutation(&mutation, object_id, self));
364        object.apply_mutation(mutation, context, associated_object)
365    }
366
367    /// Replays `mutations` for a single transaction.  `journal_offsets` contains the per-object
368    /// starting offsets; if the current transaction offset precedes an offset, the mutations for
369    /// that object are ignored.  `context` contains the location in the journal file for this
370    /// transaction and `end_offset` is the ending journal offset for this transaction.
371    pub async fn replay_mutations(
372        &self,
373        mutations: Vec<(u64, Mutation)>,
374        journal_offsets: &HashMap<u64, u64>,
375        context: &ApplyContext<'_, '_>,
376        end_offset: u64,
377    ) -> Result<(), Error> {
378        debug!(checkpoint = context.checkpoint.file_offset; "REPLAY");
379        let txn_size = {
380            let mut inner = self.inner.write();
381            if end_offset > inner.last_end_offset {
382                Some(end_offset - std::mem::replace(&mut inner.last_end_offset, end_offset))
383            } else {
384                None
385            }
386        };
387
388        let allocator_object_id = self.inner.read().allocator_object_id;
389
390        for (object_id, mutation) in mutations {
391            if let Mutation::UpdateBorrowed(borrowed) = mutation {
392                if let Some(txn_size) = txn_size {
393                    self.inner.write().borrowed_metadata_space = borrowed
394                        .checked_add(reserved_space_from_journal_usage(txn_size))
395                        .ok_or(FxfsError::Inconsistent)?;
396                }
397                continue;
398            }
399
400            // Don't replay mutations if the object doesn't want it.
401            if let Some(&offset) = journal_offsets.get(&object_id) {
402                if context.checkpoint.file_offset < offset {
403                    continue;
404                }
405            }
406
407            // If this is the first time we've encountered this store, we'll need to open it.
408            if object_id != allocator_object_id {
409                self.open_store(&self.root_store(), object_id).await?;
410            }
411
412            self.apply_mutation(object_id, mutation, context, AssocObj::None)?;
413        }
414        Ok(())
415    }
416
417    /// Opens the specified store if it isn't already.  This is *not* thread-safe.
418    async fn open_store(&self, parent: &Arc<ObjectStore>, object_id: u64) -> Result<(), Error> {
419        if self.inner.read().stores.contains_key(&object_id) {
420            return Ok(());
421        }
422        let store = ObjectStore::open(parent, object_id, Some(Box::new(TreeCache::new())))
423            .await
424            .with_context(|| format!("Failed to open store {object_id}"))?;
425        if let Some(on_new_store) = &self.on_new_store {
426            on_new_store(&store);
427        }
428        assert!(self.inner.write().stores.insert(object_id, store).is_none());
429        Ok(())
430    }
431
432    /// Called by the journaling system to apply a transaction.  `checkpoint` indicates the location
433    /// in the journal file for this transaction.  Returns an optional mutation to be written to be
434    /// included with the transaction.
435    pub fn apply_transaction(
436        &self,
437        transaction: &mut Transaction<'_>,
438        checkpoint: &JournalCheckpoint,
439    ) -> Result<Option<Mutation>, Error> {
440        // Record old values so we can see what changes as a result of this transaction.
441        let old_amount = self.metadata_reservation().amount();
442        let old_required = self.inner.read().required_reservation();
443
444        debug!(checkpoint = checkpoint.file_offset; "BEGIN TXN");
445        let mutations = transaction.take_mutations();
446        let context =
447            ApplyContext { mode: ApplyMode::Live(transaction), checkpoint: checkpoint.clone() };
448        for TxnMutation { object_id, mutation, associated_object, .. } in mutations {
449            self.apply_mutation(object_id, mutation, &context, associated_object)?;
450        }
451        debug!("END TXN");
452
453        Ok(if let MetadataReservation::Borrowed = transaction.metadata_reservation {
454            // If this transaction is borrowing metadata, figure out what has changed and return a
455            // mutation with the updated value for borrowed.  The transaction might have allocated
456            // or deallocated some data from the metadata reservation, or it might have made a
457            // change that means we need to reserve more or less space (e.g. we compacted).
458            let new_amount = self.metadata_reservation().amount();
459            let mut inner = self.inner.write();
460            let new_required = inner.required_reservation();
461            let add = old_amount + new_required;
462            let sub = new_amount + old_required;
463            if add >= sub {
464                inner.borrowed_metadata_space += add - sub;
465            } else {
466                inner.borrowed_metadata_space =
467                    inner.borrowed_metadata_space.saturating_sub(sub - add);
468            }
469            Some(Mutation::UpdateBorrowed(inner.borrowed_metadata_space))
470        } else {
471            // This transaction should have had no impact on the metadata reservation or the amount
472            // we need to reserve.
473            debug_assert_eq!(self.metadata_reservation().amount(), old_amount);
474            debug_assert_eq!(self.inner.read().required_reservation(), old_required);
475            None
476        })
477    }
478
479    /// Called by the journaling system after a transaction has been written providing the end
480    /// offset for the transaction so that we can adjust borrowed metadata space accordingly.
481    pub fn did_commit_transaction(
482        &self,
483        transaction: &mut Transaction<'_>,
484        _checkpoint: &JournalCheckpoint,
485        end_offset: u64,
486    ) {
487        let reservation = self.metadata_reservation();
488        let mut inner = self.inner.write();
489        let journal_usage = end_offset - std::mem::replace(&mut inner.last_end_offset, end_offset);
490
491        if journal_usage > inner.max_transaction_size.0 {
492            inner.max_transaction_size.0 = journal_usage;
493            inner.max_transaction_size.1.set(journal_usage);
494        }
495
496        let txn_space = reserved_space_from_journal_usage(journal_usage);
497        match &mut transaction.metadata_reservation {
498            MetadataReservation::None => unreachable!(),
499            MetadataReservation::Borrowed => {
500                // Account for the amount we need to borrow for the transaction itself now that we
501                // know the transaction size.
502                inner.borrowed_metadata_space += txn_space;
503
504                // This transaction borrowed metadata space, but it might have returned space to the
505                // transaction that we can now give back to the allocator.
506                let to_give_back = (reservation.amount() + inner.borrowed_metadata_space)
507                    .saturating_sub(inner.required_reservation());
508                if to_give_back > 0 {
509                    reservation.give_back(to_give_back);
510                }
511            }
512            MetadataReservation::Hold(hold_amount) => {
513                // Transfer reserved space into the metadata reservation.
514                let txn_reservation = transaction.allocator_reservation.unwrap();
515                assert_ne!(
516                    txn_reservation as *const _, reservation as *const _,
517                    "MetadataReservation::Borrowed should be used."
518                );
519                txn_reservation.commit(txn_space);
520                if txn_reservation.owner_object_id() != reservation.owner_object_id() {
521                    assert_eq!(
522                        reservation.owner_object_id(),
523                        None,
524                        "Should not be mixing attributed owners."
525                    );
526                    inner
527                        .allocator
528                        .as_ref()
529                        .unwrap()
530                        .disown_reservation(txn_reservation.owner_object_id(), txn_space);
531                }
532                if let Some(amount) = hold_amount.checked_sub(txn_space) {
533                    *hold_amount = amount;
534                } else {
535                    panic!("Transaction was larger than metadata reservation");
536                }
537                reservation.add(txn_space);
538            }
539            MetadataReservation::Reservation(txn_reservation) => {
540                // Transfer reserved space into the metadata reservation.
541                txn_reservation.move_to(reservation, txn_space);
542            }
543        }
544        // Check that our invariant holds true.
545        debug_assert_eq!(
546            reservation.amount() + inner.borrowed_metadata_space,
547            inner.required_reservation(),
548            "txn_space: {}, reservation_amount: {}, borrowed: {}, required: {}",
549            txn_space,
550            reservation.amount(),
551            inner.borrowed_metadata_space,
552            inner.required_reservation(),
553        );
554    }
555
556    /// Drops a transaction.  This is called automatically when a transaction is dropped.  If the
557    /// transaction has been committed, it should contain no mutations and so nothing will get rolled
558    /// back.  For each mutation, drop_mutation is called to allow for roll back (e.g. the allocator
559    /// will unreserve allocations).
560    pub fn drop_transaction(&self, transaction: &mut Transaction<'_>) {
561        for TxnMutation { object_id, mutation, .. } in transaction.take_mutations() {
562            self.journaling_object(object_id).map(|o| o.drop_mutation(mutation, transaction));
563        }
564    }
565
566    /// Returns the journal file offsets that each object depends on and the checkpoint for the
567    /// minimum offset.
568    pub fn journal_file_offsets(&self) -> (HashMap<u64, u64>, Option<JournalCheckpoint>) {
569        let inner = self.inner.read();
570        let mut min_checkpoint = None;
571        let mut offsets = HashMap::default();
572        for (&object_id, checkpoint) in &inner.journal_checkpoints {
573            let checkpoint = checkpoint.earliest();
574            match &mut min_checkpoint {
575                None => min_checkpoint = Some(checkpoint),
576                Some(min_checkpoint) => {
577                    if checkpoint.file_offset < min_checkpoint.file_offset {
578                        *min_checkpoint = checkpoint;
579                    }
580                }
581            }
582            offsets.insert(object_id, checkpoint.file_offset);
583        }
584        (offsets, min_checkpoint.cloned())
585    }
586
587    /// Returns the checkpoint into the journal that the object depends on, or None if the object
588    /// has no journaled updates.
589    pub fn journal_checkpoint(&self, object_id: u64) -> Option<JournalCheckpoint> {
590        self.inner
591            .read()
592            .journal_checkpoints
593            .get(&object_id)
594            .map(|checkpoints| checkpoints.earliest().clone())
595    }
596
597    /// Returns true if the object identified by `object_id` is known to have updates recorded in
598    /// the journal that the object depends upon.
599    pub fn needs_flush(&self, object_id: u64) -> bool {
600        self.inner.read().journal_checkpoints.contains_key(&object_id)
601    }
602
603    /// Flushes changes to device with the given reason.
604    /// Also returns the earliest known version of a struct on the filesystem.
605    pub async fn flush(&self, reason: FlushReason) -> Result<Version, Error> {
606        let objects = {
607            let inner = self.inner.read();
608            let mut object_ids = inner.journal_checkpoints.keys().cloned().collect::<Vec<_>>();
609            // Process objects in reverse sorted order because that will mean we compact the root
610            // object store last which will ensure we include the metadata from the compactions of
611            // other objects.
612            object_ids.sort_unstable();
613            object_ids
614                .iter()
615                .rev()
616                .filter_map(|oid| inner.journaling_object(*oid).map(|obj| (*oid, obj)))
617                .collect::<Vec<_>>()
618        };
619
620        // As we iterate, keep track of the earliest version used by structs in these objects
621        let mut earliest_version: Version = LATEST_VERSION;
622        for (object_id, object) in objects {
623            let object_earliest_version = object
624                .flush(reason)
625                .await
626                .with_context(|| format!("Failed to flush oid {object_id}"))?;
627            if object_earliest_version < earliest_version {
628                earliest_version = object_earliest_version;
629            }
630        }
631
632        Ok(earliest_version)
633    }
634
635    pub fn journaling_object(&self, object_id: u64) -> Option<Arc<dyn JournalingObject>> {
636        self.inner.read().journaling_object(object_id)
637    }
638
639    pub fn init_metadata_reservation(&self) -> Result<(), Error> {
640        if self.root_parent_store().filesystem().options().read_only {
641            // We don't need metadata reservations in read-only mode.
642            return Ok(());
643        }
644        let inner = self.inner.read();
645        let required = inner.required_reservation();
646        ensure!(required >= inner.borrowed_metadata_space, FxfsError::Inconsistent);
647        let allocator = inner.allocator.as_ref().cloned().unwrap();
648        self.metadata_reservation
649            .set(
650                allocator
651                    .clone()
652                    .reserve(None, inner.required_reservation() - inner.borrowed_metadata_space)
653                    .with_context(|| {
654                        format!(
655                            "Failed to reserve {} - {} = {} bytes, free={}, \
656                             owner_bytes={}",
657                            inner.required_reservation(),
658                            inner.borrowed_metadata_space,
659                            inner.required_reservation() - inner.borrowed_metadata_space,
660                            Saturating(allocator.get_disk_bytes()) - allocator.get_used_bytes(),
661                            allocator.owner_bytes_debug(),
662                        )
663                    })?,
664            )
665            .unwrap();
666        Ok(())
667    }
668
669    pub fn metadata_reservation(&self) -> &Reservation {
670        self.metadata_reservation.get().unwrap()
671    }
672
673    pub fn is_metadata_reservation(&self, reservation: &Reservation) -> bool {
674        self.metadata_reservation.get().is_some_and(|r| std::ptr::eq(r, reservation))
675    }
676
677    pub fn update_reservation(&self, object_id: u64, amount: u64) {
678        self.inner.write().reservations.insert(object_id, amount);
679    }
680
681    pub fn reservation(&self, object_id: u64) -> Option<u64> {
682        self.inner.read().reservations.get(&object_id).cloned()
683    }
684
685    /// Returns the maximum reservation held by any object (the amount budgeted for compaction).
686    pub fn max_store_reservation(&self) -> u64 {
687        self.inner.read().reservations.values().max().cloned().unwrap_or(0)
688    }
689
690    pub fn set_reserved_space(&self, amount: u64) {
691        self.inner.write().reserved_space = amount;
692    }
693
694    pub fn last_end_offset(&self) -> u64 {
695        self.inner.read().last_end_offset
696    }
697
698    pub fn set_last_end_offset(&self, v: u64) {
699        self.inner.write().last_end_offset = v;
700    }
701
702    pub fn borrowed_metadata_space(&self) -> u64 {
703        self.inner.read().borrowed_metadata_space
704    }
705
706    pub fn set_borrowed_metadata_space(&self, v: u64) {
707        self.inner.write().borrowed_metadata_space = v;
708    }
709
710    pub fn write_mutations(
711        &self,
712        object_id: u64,
713        mutations: ObjectMutationIterator<'_, '_>,
714        writer: journal::Writer<'_>,
715    ) {
716        self.journaling_object(object_id).unwrap().write_mutations(mutations, writer);
717    }
718
719    pub fn unlocked_stores(&self) -> Vec<Arc<ObjectStore>> {
720        let inner = self.inner.read();
721        let mut stores = Vec::new();
722        for store in inner.stores.values() {
723            if !store.is_locked() {
724                stores.push(store.clone());
725            }
726        }
727        stores
728    }
729
730    /// Creates a lazy inspect node named `str` under `parent` which will yield statistics for the
731    /// object manager when queried.
732    pub fn track_statistics(self: &Arc<Self>, parent: &fuchsia_inspect::Node, name: &str) {
733        let this = Arc::downgrade(self);
734        parent.record_lazy_child(name, move || {
735            let this_clone = this.clone();
736            async move {
737                let inspector = fuchsia_inspect::Inspector::default();
738                if let Some(this) = this_clone.upgrade() {
739                    let (required, borrowed, earliest_checkpoint) = {
740                        // TODO(https://fxbug.dev/42069513): Push-back or rate-limit to prevent DoS.
741                        let inner = this.inner.read();
742                        (
743                            inner.required_reservation(),
744                            inner.borrowed_metadata_space,
745                            inner.earliest_journal_offset(),
746                        )
747                    };
748                    let root = inspector.root();
749                    if let Some(reservation) = this.metadata_reservation.get() {
750                        root.record_uint("metadata_reservation", reservation.amount());
751                    }
752                    root.record_uint("required_reservation", required);
753                    root.record_uint("borrowed_reservation", borrowed);
754                    if let Some(earliest_checkpoint) = earliest_checkpoint {
755                        root.record_uint("earliest_checkpoint", earliest_checkpoint);
756                    }
757
758                    // TODO(https://fxbug.dev/42068224): Post-compute rather than manually computing metrics.
759                    if let Some(x) = round_div(100 * borrowed, required) {
760                        root.record_uint("borrowed_to_required_reservation_percent", x);
761                    }
762                }
763                Ok(inspector)
764            }
765            .boxed()
766        });
767    }
768
769    /// Normally, we make new transactions pay for overheads incurred by the journal, such as
770    /// checksums and padding, but if the journal has discarded a significant amount after a replay,
771    /// we run the risk of there not being enough reserved.  To handle this, if the amount is
772    /// significant, we force the journal to borrow the space (using a journal created transaction).
773    pub fn needs_borrow_for_journal(&self, checkpoint: u64) -> bool {
774        checkpoint.checked_sub(self.inner.read().last_end_offset).unwrap() > 256
775    }
776}
777
778/// ReservationUpdate is an associated object that sets the amount reserved for an object
779/// (overwriting any previous amount). Updates must be applied as part of a transaction before
780/// did_commit_transaction runs because it will reconcile the accounting for reserved metadata
781/// space.
782pub struct ReservationUpdate(u64);
783
784impl ReservationUpdate {
785    pub fn new(amount: u64) -> Self {
786        Self(amount)
787    }
788}
789
790impl AssociatedObject for ReservationUpdate {
791    fn will_apply_mutation(&self, _mutation: &Mutation, object_id: u64, manager: &ObjectManager) {
792        manager.update_reservation(object_id, self.0);
793    }
794}