1use crate::errors::FxfsError;
8use crate::log::*;
9use crate::lsm_tree::types::{ItemRef, LayerIterator};
10use crate::lsm_tree::{LSMTree, layers_from_handles};
11use crate::object_handle::{INVALID_OBJECT_ID, ObjectHandle, ReadObjectHandle};
12use crate::object_store::extent_record::ExtentValue;
13use crate::object_store::object_manager::{ObjectManager, ReservationUpdate};
14use crate::object_store::object_record::{ObjectKey, ObjectValue};
15use crate::object_store::transaction::{AssociatedObject, LockKey, Mutation, lock_keys};
16use crate::object_store::{
17 AssocObj, DirectWriter, EncryptedMutations, HandleOptions, LastObjectId, LastObjectIdInfo,
18 LockState, MAX_ENCRYPTED_MUTATIONS_SIZE, ObjectStore, Options, ReservedId, StoreInfo,
19 layer_size_from_encrypted_mutations_size, tree,
20};
21use crate::serialized_types::{LATEST_VERSION, Version, VersionedLatest};
22use anyhow::{Context, Error, anyhow};
23use std::sync::OnceLock;
24use std::sync::atomic::Ordering;
25
26#[derive(Copy, Clone, Debug, PartialEq, Eq)]
27pub enum Reason {
28 Journal,
30
31 PostMount,
34}
35
36#[fxfs_trace::trace]
37impl ObjectStore {
38 #[trace("store_object_id" => self.store_object_id)]
39 pub async fn flush_with_reason(&self, reason: Reason) -> Result<Version, Error> {
40 if self.parent_store.is_none() {
41 return Ok(self.tree.get_earliest_version());
43 }
44 let filesystem = self.filesystem();
45 let object_manager = filesystem.object_manager();
46
47 let keys = lock_keys![LockKey::flush(self.store_object_id())];
48 let _guard = Some(filesystem.lock_manager().write_lock(keys).await);
49
50 if matches!(*self.lock_state.lock(), LockState::Deleted) {
52 return Ok(LATEST_VERSION);
55 }
56
57 match reason {
58 Reason::PostMount => {
59 if self.store_info().unwrap().encrypted_mutations_object_id == INVALID_OBJECT_ID
65 && self.tree.get_earliest_version() == LATEST_VERSION
66 {
67 return Ok(self.tree.get_earliest_version());
70 }
71 }
72 Reason::Journal => {
73 let earliest_version = self.tree.get_earliest_version();
76 if !object_manager.needs_flush(self.store_object_id)
77 && earliest_version == LATEST_VERSION
78 {
79 return Ok(earliest_version);
82 }
83 }
84 }
85
86 let trace = self.trace.load(Ordering::Relaxed);
87 if trace {
88 info!(store_id = self.store_object_id(); "OS: begin flush");
89 }
90
91 if matches!(&*self.lock_state.lock(), LockState::Locked) {
92 self.flush_locked().await.with_context(|| {
93 format!("Failed to flush object store {}", self.store_object_id)
94 })?;
95 } else {
96 self.flush_unlocked(reason).await.with_context(|| {
97 format!("Failed to flush object store {}", self.store_object_id)
98 })?;
99 }
100
101 if trace {
102 info!(store_id = self.store_object_id(); "OS: end flush");
103 }
104 if let Some(callback) = &*self.flush_callback.lock() {
105 callback(self);
106 }
107
108 let mut counters = self.counters.lock();
111 counters.num_flushes += 1;
112 counters.last_flush_time = Some(std::time::SystemTime::now());
113 Ok(self.tree.get_earliest_version())
115 }
116
117 async fn flush_unlocked(&self, reason: Reason) -> Result<Vec<u64>, Error> {
119 struct StoreInfoSnapshot<'a> {
120 store: &'a ObjectStore,
121 store_info: OnceLock<StoreInfo>,
122 }
123 impl AssociatedObject for StoreInfoSnapshot<'_> {
124 fn will_apply_mutation(
125 &self,
126 _mutation: &Mutation,
127 _object_id: u64,
128 _manager: &ObjectManager,
129 ) {
130 let mut store_info = self.store.store_info().unwrap();
131
132 let mutations_cipher = self.store.mutations_cipher.lock();
134 if let Some(cipher) = mutations_cipher.as_ref() {
135 store_info.mutations_cipher_offset = cipher.offset();
136 }
137
138 self.store_info.set(store_info).unwrap();
139 }
140 }
141
142 let store_info_snapshot = StoreInfoSnapshot { store: self, store_info: OnceLock::new() };
143
144 let filesystem = self.filesystem();
145 let object_manager = filesystem.object_manager();
146 let reservation = object_manager.metadata_reservation();
147 let txn_options = Options {
148 skip_journal_checks: true,
149 skip_key_roll: true,
150 borrow_metadata_space: true,
151 allocator_reservation: Some(reservation),
152 ..Default::default()
153 };
154
155 let mut transaction = self.new_transaction(lock_keys![], txn_options).await?;
158 transaction.add_with_object(
159 self.store_object_id(),
160 Mutation::BeginFlush,
161 AssocObj::Borrowed(&store_info_snapshot),
162 );
163 transaction.commit().await?;
164
165 let mut new_store_info = store_info_snapshot.store_info.into_inner().unwrap();
166
167 let mut transaction = self.new_transaction(lock_keys![], txn_options).await?;
172
173 let parent_store = self.parent_store.as_ref().unwrap();
175 let handle_options = HandleOptions { skip_journal_checks: true, ..Default::default() };
176 let id_and_key = {
177 let mut lock_state = self.lock_state.lock();
178 match &mut *lock_state {
179 LockState::Unlocked { cached_keys, .. } => {
180 if let Some(item) = cached_keys.pop() {
181 Ok(Some(item))
182 } else {
183 log::warn!("No cached keys for flush for store {}", self.store_object_id());
184 Err(anyhow!(FxfsError::Internal).context("No cached keys for flush"))
185 }
186 }
187 LockState::UnlockedReadOnly(..) => {
188 Err(anyhow!(FxfsError::Internal).context("Flush on read-only store"))
189 }
190 LockState::Unencrypted => Ok(None),
191 _ => Err(anyhow!(FxfsError::Internal))
192 .with_context(|| format!("Invalid lock state ({:?}) for flush", *lock_state)),
193 }
194 }?;
195
196 let new_object_tree_layer = if let Some((raw_id, key, unwrapped_key)) = id_and_key {
197 let object_id = ReservedId::new(parent_store, raw_id);
198 ObjectStore::create_object_with_key(
199 parent_store,
200 &mut transaction,
201 object_id,
202 handle_options,
203 key,
204 unwrapped_key,
205 )
206 .await?
207 } else {
208 ObjectStore::create_object(parent_store, &mut transaction, handle_options, None).await?
209 };
210 let writer = DirectWriter::new(&new_object_tree_layer, txn_options).await;
211 let new_object_tree_layer_object_id = new_object_tree_layer.object_id();
212 parent_store.add_to_graveyard(&mut transaction, new_object_tree_layer_object_id);
213
214 transaction.commit().await?;
215
216 let (layers_to_keep, old_layers) = tree::flush(
218 &self.tree,
219 writer,
220 (reason == Reason::Journal).then(|| filesystem.journal().get_compaction_yielder()),
221 )
222 .await
223 .context("Failed to flush tree")?;
224
225 let mut new_layers = layers_from_handles([new_object_tree_layer]).await?;
227 new_layers.extend(layers_to_keep.iter().map(|l| (*l).clone()));
228
229 new_store_info.layers = Vec::new();
230 for layer in &new_layers {
231 if let Some(handle) = layer.handle() {
232 new_store_info.layers.push(handle.object_id());
233 }
234 }
235
236 let reservation_update: ReservationUpdate; let mut end_transaction = parent_store
238 .new_transaction(
239 lock_keys![LockKey::object(
240 self.parent_store.as_ref().unwrap().store_object_id(),
241 self.store_info_handle_object_id().unwrap(),
242 )],
243 txn_options,
244 )
245 .await?;
246
247 parent_store.remove_from_graveyard(&mut end_transaction, new_object_tree_layer_object_id);
248
249 for layer in &old_layers {
251 if let Some(handle) = layer.handle() {
252 parent_store.add_to_graveyard(&mut end_transaction, handle.object_id());
253 }
254 }
255
256 let old_encrypted_mutations_object_id =
257 std::mem::replace(&mut new_store_info.encrypted_mutations_object_id, INVALID_OBJECT_ID);
258 if old_encrypted_mutations_object_id != INVALID_OBJECT_ID {
259 parent_store.add_to_graveyard(&mut end_transaction, old_encrypted_mutations_object_id);
260 }
261
262 match &mut new_store_info.last_object_id {
270 LastObjectIdInfo::Unencrypted { id } => {
271 let LastObjectId::Unencrypted { id: in_memory_value } =
272 &*self.last_object_id.lock()
273 else {
274 unreachable!()
275 };
276 *id = *in_memory_value;
277 }
278 LastObjectIdInfo::Encrypted { id, key } => {
279 let LastObjectId::Encrypted { id: in_memory_value, .. } =
280 &*self.last_object_id.lock()
281 else {
282 unreachable!()
283 };
284 *id = *in_memory_value;
285 let guard = self.store_info.lock();
286 let current_store_info = guard.as_ref().unwrap();
287 let LastObjectIdInfo::Encrypted { key: in_memory_value, .. } =
288 ¤t_store_info.last_object_id
289 else {
290 unreachable!()
291 };
292 *key = in_memory_value.clone();
293 }
294 LastObjectIdInfo::Low32Bit => {}
295 }
296
297 self.write_store_info(&mut end_transaction, &new_store_info).await?;
298
299 let layer_file_sizes = new_layers
300 .iter()
301 .map(|l| l.handle().map(ReadObjectHandle::get_size).unwrap_or(0))
302 .collect::<Vec<u64>>();
303
304 let total_layer_size = layer_file_sizes.iter().sum();
305 reservation_update =
306 ReservationUpdate::new(tree::reservation_amount_from_layer_size(total_layer_size));
307
308 end_transaction.add_with_object(
309 self.store_object_id(),
310 Mutation::EndFlush,
311 AssocObj::Borrowed(&reservation_update),
312 );
313
314 if self.trace.load(Ordering::Relaxed) {
315 info!(
316 store_id = self.store_object_id(),
317 old_layer_count = old_layers.len(),
318 new_layer_count = new_layers.len(),
319 total_layer_size,
320 new_store_info:?;
321 "OS: compacting"
322 );
323 }
324
325 end_transaction
326 .commit_with_callback(|_| {
327 let mut store_info = self.store_info.lock();
328 let info = store_info.as_mut().unwrap();
329 info.layers = new_store_info.layers;
330 info.encrypted_mutations_object_id = new_store_info.encrypted_mutations_object_id;
331 info.mutations_cipher_offset = new_store_info.mutations_cipher_offset;
332 self.tree.set_layers(new_layers);
333 })
334 .await?;
335
336 for layer in old_layers {
338 let object_id = layer.handle().map(|h| h.object_id());
339 layer.close_layer().await;
340 if let Some(object_id) = object_id {
341 parent_store.tombstone_object(object_id, txn_options).await?;
342 }
343 }
344
345 if old_encrypted_mutations_object_id != INVALID_OBJECT_ID {
346 parent_store.tombstone_object(old_encrypted_mutations_object_id, txn_options).await?;
347 }
348
349 Ok(layer_file_sizes)
350 }
351
352 async fn flush_locked(&self) -> Result<(), Error> {
354 let filesystem = self.filesystem();
355 let object_manager = filesystem.object_manager();
356 let reservation = object_manager.metadata_reservation();
357 let txn_options = Options {
358 skip_journal_checks: true,
359 skip_key_roll: true,
360 borrow_metadata_space: true,
361 allocator_reservation: Some(reservation),
362 ..Default::default()
363 };
364
365 let mut transaction = self.new_transaction(lock_keys![], txn_options).await?;
366 transaction.add(self.store_object_id(), Mutation::BeginFlush);
367 transaction.commit().await?;
368
369 let mut new_store_info = self.load_store_info().await?;
370
371 let mut transaction = self.new_transaction(lock_keys![], txn_options).await?;
376
377 let reservation_update: ReservationUpdate; let handle; let mut end_transaction;
380
381 let parent_store = self.parent_store.as_ref().unwrap();
384 handle = if new_store_info.encrypted_mutations_object_id == INVALID_OBJECT_ID {
385 let handle = ObjectStore::create_object(
386 parent_store,
387 &mut transaction,
388 HandleOptions { skip_journal_checks: true, ..Default::default() },
389 None,
390 )
391 .await?;
392 let oid = handle.object_id();
393 end_transaction = parent_store
394 .new_transaction(
395 lock_keys![
396 LockKey::object(parent_store.store_object_id(), oid),
397 LockKey::object(
398 parent_store.store_object_id(),
399 self.store_info_handle_object_id().unwrap(),
400 ),
401 ],
402 txn_options,
403 )
404 .await?;
405 new_store_info.encrypted_mutations_object_id = oid;
406 parent_store.add_to_graveyard(&mut transaction, oid);
407 parent_store.remove_from_graveyard(&mut end_transaction, oid);
408 handle
409 } else {
410 end_transaction = parent_store
411 .new_transaction(
412 lock_keys![
413 LockKey::object(
414 parent_store.store_object_id(),
415 new_store_info.encrypted_mutations_object_id,
416 ),
417 LockKey::object(
418 parent_store.store_object_id(),
419 self.store_info_handle_object_id().unwrap(),
420 ),
421 ],
422 txn_options,
423 )
424 .await?;
425 ObjectStore::open_object(
426 parent_store,
427 new_store_info.encrypted_mutations_object_id,
428 HandleOptions { skip_journal_checks: true, ..Default::default() },
429 None,
430 )
431 .await?
432 };
433 transaction.commit().await?;
434
435 let journaled = filesystem
438 .journal()
439 .read_transactions_for_object(self.store_object_id)
440 .await
441 .context("Failed to read encrypted mutations from journal")?;
442 let mut buffer = handle.allocate_buffer(MAX_ENCRYPTED_MUTATIONS_SIZE).await;
443 let mut cursor = std::io::Cursor::new(buffer.as_mut_slice());
444 EncryptedMutations::from_replayed_mutations(self.store_object_id, journaled)
445 .serialize_with_version(&mut cursor)?;
446 let len = cursor.position() as usize;
447 handle.txn_write(&mut end_transaction, handle.get_size(), buffer.subslice(..len)).await?;
448
449 self.write_store_info(&mut end_transaction, &new_store_info).await?;
450
451 let mut total_layer_size = 0;
452 for &oid in &new_store_info.layers {
453 total_layer_size += parent_store.get_file_size(oid).await?;
454 }
455 total_layer_size +=
456 layer_size_from_encrypted_mutations_size(handle.get_size() + len as u64);
457
458 reservation_update =
459 ReservationUpdate::new(tree::reservation_amount_from_layer_size(total_layer_size));
460
461 end_transaction.add_with_object(
462 self.store_object_id(),
463 Mutation::EndFlush,
464 AssocObj::Borrowed(&reservation_update),
465 );
466
467 end_transaction.commit().await?;
468
469 Ok(())
470 }
471}
472
473#[cfg(test)]
474mod tests {
475 use crate::filesystem::{FxFilesystem, FxFilesystemBuilder, JournalingObject, SyncOptions};
476 use crate::object_handle::{INVALID_OBJECT_ID, ObjectHandle};
477 use crate::object_store::directory::Directory;
478 use crate::object_store::transaction::{Options, lock_keys};
479 use crate::object_store::volume::root_volume;
480 use crate::object_store::{
481 HandleOptions, LockKey, NewChildStoreOptions, ObjectStore, StoreOptions,
482 layer_size_from_encrypted_mutations_size, tree,
483 };
484 use fxfs_insecure_crypto::new_insecure_crypt;
485 use std::sync::Arc;
486 use storage_device::DeviceHolder;
487 use storage_device::fake_device::FakeDevice;
488
489 async fn run_key_roll_test(flush_before_unlock: bool) {
490 let device = DeviceHolder::new(FakeDevice::new(8192, 1024));
491 let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
492 let store_id = {
493 let root_volume = root_volume(fs.clone()).await.expect("root_volume failed");
494 root_volume
495 .new_volume(
496 "test",
497 NewChildStoreOptions {
498 options: StoreOptions {
499 crypt: Some(Arc::new(new_insecure_crypt())),
500 ..StoreOptions::default()
501 },
502 ..NewChildStoreOptions::default()
503 },
504 )
505 .await
506 .expect("new_volume failed")
507 .store_object_id()
508 };
509
510 fs.close().await.expect("close failed");
511 let device = fs.take_device().await;
512 device.reopen(false);
513
514 let fs = FxFilesystemBuilder::new()
515 .roll_metadata_key_byte_count(512 * 1024)
516 .open(device)
517 .await
518 .expect("open failed");
519
520 let (first_filename, last_filename) = {
521 let store = fs.object_manager().store(store_id).expect("store not found");
522 store.unlock(Arc::new(new_insecure_crypt())).await.expect("unlock failed");
523
524 let root_dir = Directory::open(&store, store.root_directory_object_id())
526 .await
527 .expect("open failed");
528
529 let mut last_mutations_cipher_offset = 0;
530 let mut i = 0;
531 let first_filename = format!("{:<200}", i);
532 loop {
533 let mut transaction = store
534 .new_transaction(
535 lock_keys![LockKey::object(store_id, root_dir.object_id())],
536 Options::default(),
537 )
538 .await
539 .expect("new_transaction failed");
540 root_dir
541 .create_child_file(&mut transaction, &format!("{:<200}", i))
542 .await
543 .expect("create_child_file failed");
544 i += 1;
545 transaction.commit().await.expect("commit failed");
546 let cipher_offset = store.mutations_cipher.lock().as_ref().unwrap().offset();
547 if cipher_offset < last_mutations_cipher_offset {
548 break;
549 }
550 last_mutations_cipher_offset = cipher_offset;
551 }
552
553 fs.sync(SyncOptions::default()).await.expect("sync failed");
556
557 let mut transaction = store
559 .new_transaction(
560 lock_keys![LockKey::object(store_id, root_dir.object_id())],
561 Options::default(),
562 )
563 .await
564 .expect("new_transaction failed");
565 let last_filename = format!("{:<200}", i);
566 root_dir
567 .create_child_file(&mut transaction, &last_filename)
568 .await
569 .expect("create_child_file failed");
570 transaction.commit().await.expect("commit failed");
571 (first_filename, last_filename)
572 };
573
574 fs.close().await.expect("close failed");
575
576 let device = fs.take_device().await;
578 device.reopen(false);
579 let fs = FxFilesystemBuilder::new()
580 .roll_metadata_key_byte_count(512 * 1024)
581 .open(device)
582 .await
583 .expect("open failed");
584
585 if flush_before_unlock {
586 fs.object_manager().flush().await.expect("flush failed");
589 }
590
591 {
592 let store = fs.object_manager().store(store_id).expect("store not found");
593 store.unlock(Arc::new(new_insecure_crypt())).await.expect("unlock failed");
594
595 assert_eq!(store.mutations_cipher.lock().as_ref().unwrap().offset(), 0);
597
598 let root_dir = Directory::open(&store, store.root_directory_object_id())
599 .await
600 .expect("open failed");
601 root_dir
602 .lookup(&first_filename)
603 .await
604 .expect("Lookup failed")
605 .expect("First created file wasn't present");
606 root_dir
607 .lookup(&last_filename)
608 .await
609 .expect("Lookup failed")
610 .expect("Last created file wasn't present");
611 }
612 }
613
614 #[fuchsia::test(threads = 10)]
615 async fn test_metadata_key_roll() {
616 run_key_roll_test(false).await;
617 }
618
619 #[fuchsia::test(threads = 10)]
620 async fn test_metadata_key_roll_with_flush_before_unlock() {
621 run_key_roll_test(true).await;
622 }
623
624 #[fuchsia::test]
625 async fn test_flush_when_locked() {
626 let device = DeviceHolder::new(FakeDevice::new(8192, 1024));
627 let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
628 let root_volume = root_volume(fs.clone()).await.expect("root_volume failed");
629 let crypt = Arc::new(new_insecure_crypt());
630 let store = root_volume
631 .new_volume(
632 "test",
633 NewChildStoreOptions {
634 options: StoreOptions { crypt: Some(crypt.clone()), ..StoreOptions::default() },
635 ..NewChildStoreOptions::default()
636 },
637 )
638 .await
639 .expect("new_volume failed");
640 let root_dir =
641 Directory::open(&store, store.root_directory_object_id()).await.expect("open failed");
642 let mut transaction = fs
643 .root_store()
644 .new_transaction(
645 lock_keys![LockKey::object(store.store_object_id(), root_dir.object_id())],
646 Options::default(),
647 )
648 .await
649 .expect("new_transaction failed");
650 let foo = root_dir
651 .create_child_file(&mut transaction, "foo")
652 .await
653 .expect("create_child_file failed");
654 transaction.commit().await.expect("commit failed");
655
656 store.flush().await.expect("flush failed");
660
661 let mut transaction = fs
662 .root_store()
663 .new_transaction(
664 lock_keys![LockKey::object(store.store_object_id(), root_dir.object_id())],
665 Options::default(),
666 )
667 .await
668 .expect("new_transaction failed");
669 let bar = root_dir
670 .create_child_file(&mut transaction, "bar")
671 .await
672 .expect("create_child_file failed");
673 transaction.commit().await.expect("commit failed");
674
675 store.lock().await.expect("lock failed");
676
677 store.flush().await.expect("flush failed");
679
680 let info = store.load_store_info().await.unwrap();
682 let parent_store = store.parent_store().unwrap();
683 let mut total_layer_size = 0;
684 for &oid in &info.layers {
685 total_layer_size +=
686 parent_store.get_file_size(oid).await.expect("get_file_size failed");
687 }
688 assert_ne!(info.encrypted_mutations_object_id, INVALID_OBJECT_ID);
689 total_layer_size += layer_size_from_encrypted_mutations_size(
690 parent_store
691 .get_file_size(info.encrypted_mutations_object_id)
692 .await
693 .expect("get_file_size failed"),
694 );
695 assert_eq!(
696 fs.object_manager().reservation(store.store_object_id()),
697 Some(tree::reservation_amount_from_layer_size(total_layer_size))
698 );
699
700 store.unlock(crypt).await.expect("unlock failed");
702
703 ObjectStore::open_object(&store, foo.object_id(), HandleOptions::default(), None)
704 .await
705 .expect("open_object failed");
706
707 ObjectStore::open_object(&store, bar.object_id(), HandleOptions::default(), None)
708 .await
709 .expect("open_object failed");
710
711 fs.close().await.expect("close failed");
712 }
713}
714
715impl tree::MajorCompactable<ObjectKey, ObjectValue> for LSMTree<ObjectKey, ObjectValue> {
716 async fn major_iter(
717 iter: impl LayerIterator<ObjectKey, ObjectValue>,
718 ) -> Result<impl LayerIterator<ObjectKey, ObjectValue>, Error> {
719 iter.filter(|item: ItemRef<'_, _, _>| match item {
720 ItemRef { value: ObjectValue::None, .. } => false,
722 ItemRef { value: ObjectValue::Extent(ExtentValue::None), .. } => false,
724 _ => true,
725 })
726 .await
727 }
728}