Implement Checkpoint::recover

conanfile.py

@@ -60,7 +60,7 @@ def requirements(self):
         self.requires("gperftools/2.16", **VISIBLE)
         self.requires("llfs/0.42.0", **VISIBLE)
         self.requires("pcg-cpp/cci.20220409", **VISIBLE)
-        self.requires("vqf/0.2.5", **VISIBLE)
+        self.requires("vqf/0.2.5-devel", **VISIBLE)
         self.requires("zlib/1.3.1", **OVERRIDE)
 
         if platform.system() == "Linux":

src/turtle_kv/checkpoint.cpp

@@ -13,38 +13,40 @@
 
 namespace turtle_kv {
 
-#if 0  // TODO [tastolfi 2025-03-27] re-enable me!
 //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
 //
 /*static*/ StatusOr<Checkpoint> Checkpoint::recover(
     llfs::Volume& checkpoint_volume,
-    const llfs::SlotWithPayload<TabletCheckpoint>& packed_checkpoint) noexcept
+    llfs::SlotParse& slot,
+    const PackedCheckpoint& packed_checkpoint) noexcept
 {
-  const llfs::PageId tree_root_id = packed_checkpoint.payload.new_tree_root.as_page_id();
+  VLOG(1) << "Entering Checkpoint::recover";
 
-  BATT_ASSIGN_OK_RESULT(
-      std::shared_ptr<const TreeView> tree,
-      TreeView::from_page(
-          checkpoint_volume.cache().get_page(tree_root_id, llfs::OkIfNotFound{false})));
+  // TODO: [Gabe Bornstein 11/4/25] Consider, error handling for invalid checkpoint
+  //
 
-  if (static_cast<i16>(tree->height()) != packed_checkpoint.payload.new_tree_height) {
-    // return make_db_status(turtle_db::DBStatusCodes::kBadRecoveredTreeHeight);
-    return {batt::StatusCode::kDataLoss};  // TODO [tastolfi 2025-02-20]
-  }
+  const llfs::PageId tree_root_id = packed_checkpoint.new_tree_root.as_page_id();
+
+  Subtree tree = Subtree::from_page_id(tree_root_id);
 
+  batt::StatusOr<i32> height = tree.get_height(*(checkpoint_volume.new_job()));
+
+  BATT_REQUIRE_OK(height);
   BATT_ASSIGN_OK_RESULT(llfs::SlotReadLock slot_read_lock,
-                        checkpoint_volume.lock_slots(packed_checkpoint.slot_range,
+
+                        checkpoint_volume.lock_slots(slot.offset,
                                                      llfs::LogReadMode::kDurable,
                                                      /*lock_holder=*/"Checkpoint::recover"));
 
+  VLOG(1) << "Exiting Checkpoint::recover";
   return Checkpoint{
       tree_root_id,
-      std::move(tree),
-      DeltaBatchId::from_u64(packed_checkpoint.payload.slot_upper_bound),
+      std::make_shared<Subtree>(std::move(tree)),
+      *height,
+      DeltaBatchId::from_u64(packed_checkpoint.batch_upper_bound),
       CheckpointLock::make_durable(std::move(slot_read_lock)),
   };
 }
-#endif
 
 //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
 //

src/turtle_kv/checkpoint.hpp

@@ -2,6 +2,7 @@
 
 #include <turtle_kv/checkpoint_lock.hpp>
 #include <turtle_kv/delta_batch_id.hpp>
+#include <turtle_kv/packed_checkpoint.hpp>
 
 #include <turtle_kv/tree/subtree.hpp>
 
@@ -28,11 +29,9 @@ class Checkpoint
  public:
   //+++++++++++-+-+--+----- --- -- -  -  -   -
 
-#if 0  // TODO [tastolfi 2025-03-27] re-enable
-  static StatusOr<Checkpoint> recover(
-      llfs::Volume& checkpoint_volume,
-      const llfs::SlotWithPayload<TabletCheckpoint>& checkpoint) noexcept;
-#endif
+  static StatusOr<Checkpoint> recover(llfs::Volume& checkpoint_volume,
+                                      llfs::SlotParse& slot,
+                                      const PackedCheckpoint& checkpoint) noexcept;
 
   static Checkpoint empty_at_batch(DeltaBatchId batch_id) noexcept;
 

src/turtle_kv/checkpoint_log.cpp

@@ -55,7 +55,7 @@ Status create_checkpoint_log(llfs::StorageContext& storage_context,
   return OkStatus();
 }
 
-//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
+// ==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
 //
 StatusOr<std::unique_ptr<llfs::Volume>> open_checkpoint_log(
     llfs::StorageContext& storage_context,
@@ -85,7 +85,10 @@ StatusOr<std::unique_ptr<llfs::Volume>> open_checkpoint_log(
   return storage_context.recover_object(batt::StaticType<llfs::PackedVolumeConfig>{},
                                         uuid,
                                         llfs::VolumeRuntimeOptions{
-                                            .slot_visitor_fn = llfs::VolumeReader::SlotVisitorFn{},
+                                            .slot_visitor_fn =
+                                                [](auto&&...) {
+                                                  return OkStatus();
+                                                },
                                             .root_log_options = root_log_options,
                                             .recycler_log_options = recycler_log_options,
                                             .trim_control = nullptr,

src/turtle_kv/kv_store.cpp

@@ -862,6 +862,78 @@ Status KVStore::update_checkpoint(const State* observed_state)
   return OkStatus();
 }
 
+using CheckpointSlotPairs = std::vector<std::pair<llfs::SlotParse, turtle_kv::PackedCheckpoint>>;
+using CheckpointEvent = llfs::PackedVariant<turtle_kv::PackedCheckpoint>;
+
+//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
+//
+batt::StatusOr<CheckpointSlotPairs> recover_packed_checkpoints(llfs::Volume& volume)
+{
+  VLOG(1) << "Entering recover_packed_checkpoints";
+
+  llfs::StatusOr<llfs::TypedVolumeReader<CheckpointEvent>> reader =
+      volume.typed_reader<CheckpointEvent>(
+          llfs::SlotRangeSpec{
+              .lower_bound = llfs::None,
+              .upper_bound = llfs::None,
+          },
+          llfs::LogReadMode::kDurable);
+
+  BATT_REQUIRE_OK(reader);
+
+  CheckpointSlotPairs checkpoints;
+
+  for (;;) {
+    llfs::StatusOr<usize> n_slots_visited = reader->visit_typed_next(
+        batt::WaitForResource::kFalse,
+        [&checkpoints](const llfs::SlotParse& slot,
+                       const turtle_kv::PackedCheckpoint& packed_checkpoint) {
+          std::pair<llfs::SlotParse, turtle_kv::PackedCheckpoint> pair(slot, packed_checkpoint);
+          checkpoints.push_back(pair);
+          return llfs::OkStatus();
+        });
+
+    BATT_REQUIRE_OK(n_slots_visited);
+    VLOG(2) << "Visited n_slots_visited= " << *n_slots_visited << " checkpoints";
+    if (*n_slots_visited == 0) {
+      break;
+    }
+  }
+
+  VLOG(1) << "Exiting recover_packed_checkpoints";
+  return checkpoints;
+}
+
+//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
+//
+/*static*/ batt::StatusOr<turtle_kv::Checkpoint> KVStore::recover_latest_checkpoint(
+    llfs::Volume& checkpoint_log_volume,
+    std::filesystem::path checkpoint_log_dir)
+{
+  batt::StatusOr<CheckpointSlotPairs> packed_checkpoints =
+      recover_packed_checkpoints(checkpoint_log_volume);
+
+  BATT_REQUIRE_OK(packed_checkpoints);
+
+  if (packed_checkpoints->size() == 0) {
+    return Checkpoint::empty_at_batch(DeltaBatchId::from_u64(0));
+  }
+
+  // Validate that the checkpoints are in ascending order based on batch_upper_bound
+  //
+  std::pair<llfs::SlotParse, turtle_kv::PackedCheckpoint> prev_checkpoint =
+      packed_checkpoints->front();
+
+  for (auto checkpoint : *packed_checkpoints) {
+    BATT_CHECK_GE(checkpoint.second.batch_upper_bound, prev_checkpoint.second.batch_upper_bound);
+    prev_checkpoint = checkpoint;
+  }
+
+  return turtle_kv::Checkpoint::recover(checkpoint_log_volume,
+                                        packed_checkpoints->back().first,
+                                        packed_checkpoints->back().second);
+}
+
 //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
 //
 void KVStore::info_task_main() noexcept
@@ -946,6 +1018,8 @@ void KVStore::checkpoint_update_thread_main()
 StatusOr<std::unique_ptr<CheckpointJob>> KVStore::apply_batch_to_checkpoint(
     std::unique_ptr<DeltaBatch>&& delta_batch)
 {
+  // A MemTable has filled up.
+  //
   if (delta_batch) {
     // Apply the finalized MemTable to the current checkpoint (in-memory).
     //
@@ -961,7 +1035,7 @@ StatusOr<std::unique_ptr<CheckpointJob>> KVStore::apply_batch_to_checkpoint(
 
   // If the batch count is below the checkpoint distance, we are done.
   //
-  if (this->checkpoint_batch_count_ < this->checkpoint_distance_.load()) {
+  if (!this->should_create_checkpoint()) {
     return nullptr;
   }
 

src/turtle_kv/kv_store.hpp

@@ -165,11 +165,23 @@ class KVStore : public Table
 
   void set_checkpoint_distance(usize chi) noexcept;
 
+  static batt::StatusOr<turtle_kv::Checkpoint> recover_latest_checkpoint(
+      llfs::Volume& checkpoint_log_volume,
+      std::filesystem::path checkpoint_log_dir);
+
   usize get_checkpoint_distance() const noexcept
   {
     return this->checkpoint_distance_.load();
   }
 
+  bool should_create_checkpoint() const
+  {
+    // If the batch count is greater than or equal to the checkpoint distance, we need to create a
+    // checkpoint.
+    //
+    return this->checkpoint_batch_count_ >= this->checkpoint_distance_.load();
+  }
+
   Status force_checkpoint();
 
   std::function<void(std::ostream&)> debug_info() noexcept;
@@ -242,6 +254,9 @@ class KVStore : public Table
 
   std::unique_ptr<ChangeLogWriter> log_writer_;
 
+  // How frequently we take checkpoints, where the units of distance are number of MemTables.
+  // (i.e. if checkpoint_distance_ == 3, we take a checkpoint every time 3 MemTables are filled up)
+  //
   std::atomic<usize> checkpoint_distance_;
 
   absl::Mutex base_checkpoint_mutex_;