added support for recycler calling recycle_pages with depth>0

src/llfs/page_recycler.cpp

@@ -335,18 +335,21 @@ StatusOr<slot_offset_type> PageRecycler::recycle_pages(
     return this->wal_device_->slot_range(LogReadMode::kDurable).upper_bound;
   }
 
-  // Check to see if we have already seen this or any newer request.
+  // Check to make sure request was never seen before. Note that we do unique identifier check only
+  // for depth=0 as that's coming from the external client side. For all recycler internal calls
+  // (with depth > 0) we will simply accept the request.
   //
-  if (this->largest_offset_as_unique_identifier_ >= offset_as_unique_identifier) {
+  if (this->largest_offset_as_unique_identifier_ < offset_as_unique_identifier || depth != 0) {
+    // Update the largest unique identifier and continue.
+    //
+    this->largest_offset_as_unique_identifier_ = offset_as_unique_identifier;
+  } else {
+    // Look like this is a repost of some old request thus update some status and ignore/return.
+    //
     this->metrics_export().page_id_deletion_reissue.fetch_add(1);
-
     return this->wal_device_->slot_range(LogReadMode::kDurable).upper_bound;
   }
 
-  // Update the largest unique identifier.
-  //
-  this->largest_offset_as_unique_identifier_ = offset_as_unique_identifier;
-
   Optional<slot_offset_type> sync_point = None;
 
   if (depth == 0) {

src/llfs/page_recycler.test.cpp

@@ -228,9 +228,11 @@ class PageRecyclerTest : public ::testing::Test
         });
   }
 
+  // Returns an ever increasing unique_offset value for page recycler calls.
+  //
   slot_offset_type get_and_incr_unique_offset()
   {
-    return this->unique_offset_++;
+    return ++this->unique_offset_;
   }
 
   //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
@@ -272,7 +274,12 @@ class PageRecyclerTest : public ::testing::Test
   //
   std::unique_ptr<llfs::PageRecycler> unique_page_recycler_;
 
-  slot_offset_type unique_offset_{1};
+  // This is to track unique_offset for PageRecycler tests.
+  //
+  slot_offset_type unique_offset_{0};
+
+  // This mutex is to ensure serialized access to unique_offset counter.
+  //
   std::mutex recycle_pages_mutex_;
 };
 
@@ -445,6 +452,9 @@ TEST_F(PageRecyclerTest, CrashRecovery)
 
 void PageRecyclerTest::run_crash_recovery_test()
 {
+  // Note that increasing the page_count here most likely impact the test execution as we might run
+  // out of grant space.
+  //
   const usize fake_page_count = 180;
   const u32 max_branching_factor = 8;
 
@@ -510,19 +520,21 @@ void PageRecyclerTest::run_crash_recovery_test()
         const std::array<PageId, 1> to_recycle = {root_id};
 
         BATT_DEBUG_INFO("Test - recycle_pages");
-        std::lock_guard lock(this->recycle_pages_mutex_);
-        StatusOr<slot_offset_type> recycle_status =
-            recycler.recycle_pages(to_recycle, this->get_and_incr_unique_offset());
-        if (!recycle_status.ok()) {
-          failed = true;
-          break;
-        }
+        {
+          std::lock_guard lock(this->recycle_pages_mutex_);
+          StatusOr<slot_offset_type> recycle_status =
+              recycler.recycle_pages(to_recycle, this->get_and_incr_unique_offset());
+          if (!recycle_status.ok()) {
+            failed = true;
+            break;
+          }
 
-        BATT_DEBUG_INFO("Test - await_flush");
-        Status flush_status = recycler.await_flush(*recycle_status);
-        if (!flush_status.ok()) {
-          failed = true;
-          break;
+          BATT_DEBUG_INFO("Test - await_flush");
+          Status flush_status = recycler.await_flush(*recycle_status);
+          if (!flush_status.ok()) {
+            failed = true;
+            break;
+          }
         }
 
         ++progress;