Fix for #30.

conan.lock

@@ -10,12 +10,11 @@
         "glog/0.7.1",
         "gtest/1.17.0",
         "keyvcr/0.2.2",
-        "libbacktrace/cci.20240730",
         "libfuse/3.16.2",
         "libpfm4/4.13.0",
         "libunwind/1.8.1",
         "liburing/2.11",
-        "llfs/0.47.1",
+        "llfs/0.48.1",
         "openssl/3.6.0",
         "pcg-cpp/cci.20220409",
         "protobuf/3.21.12",
@@ -41,9 +40,6 @@
         "zlib/[>=1.2.11 <2]": [
             "zlib/1.3.1"
         ],
-        "libbacktrace/cci.20210118": [
-            "libbacktrace/cci.20240730"
-        ],
         "zlib/[>=1.3.1 <2]": [
             "zlib/1.3.1"
         ],

conanfile.py

@@ -92,15 +92,11 @@ def requirements(self):
         self.requires("batteries/[>=0.71.1 <1]", **VISIBLE, **OVERRIDE)
         self.requires("boost/1.88.0", **VISIBLE, **OVERRIDE)
         self.requires("glog/0.7.1", **VISIBLE)
-        self.requires("llfs/[>=0.47.0 <1]", **VISIBLE)
+        self.requires("llfs/[>=0.48.0 <1]", **VISIBLE)
         self.requires("pcg-cpp/cci.20220409", **VISIBLE)
         self.requires("yaml-cpp/[>=0.9.0 <1]")
         self.requires("zlib/1.3.1", **OVERRIDE)
 
-        # boost/1.88.0 and ninja/1.13.2 depend (exactly) on libbacktrace/cci.20210118
-        #
-        self.requires("libbacktrace/[>=cci.20240730]", **OVERRIDE)
-
         if platform.system() == "Linux":
             if self.options.with_keyvcr:
                 self.requires("keyvcr/[>=0.2.2 <1]", **VISIBLE)

src/turtle_kv/change_log/change_log.test.cpp

@@ -827,4 +827,96 @@ TEST_F(ChangeLogTest, SyncStaggeredOffsets)
   LOG(INFO) << BATT_INSPECT(this->writer_->metrics().advance_sync_upper_bound_latency);
 }
 
+//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
+// Goal: Make sure that sync operates correctly when blocks are trimmed faster than they are
+// written.
+//
+TEST_F(ChangeLogTest, SyncAggressiveTrim)
+{
+  // Set the log size large enough to reliably trigger the bug, but small enough so the
+  // test runs quickly.
+  //
+  this->config_.set_log_size(2 * kMiB);
+
+  // How long to wait for sync at the end.
+  //
+  const i32 kSyncTimeoutSeconds =
+      batt::getenv_as<i32>("TURTLE_KV_CHANGE_LOG_TEST_SYNC_TIMEOUT_SECONDS").value_or(15);
+
+  // How many bytes to write.  Make it large enough that trim has to be working.
+  //
+  const usize kBytesToAppend = this->config_.log_size() * 3;
+
+  // How many bytes to write in a single slot; make it a weird size so we get some, but not a lot
+  // of, wasted space in the blocks.
+  //
+  const usize kSlotSize = this->config_.block_size / 5 - 17;
+
+  // Create the log + open a writer.
+  //
+  ASSERT_OK(this->create_writer(RemoveExisting{true}));
+
+  // Keep track of how many blocks were used.
+  //
+  usize blocks_seen = 0;
+  usize bytes_appended = 0;
+  {
+    ChangeLogWriter::Context context{*this->writer_};
+
+    while (bytes_appended < kBytesToAppend) {
+      Status status = context.append_slot(
+          /*min_edit_offset_lower_bound=*/EditOffset{0},
+          kSlotSize,
+          batt::WaitForResource::kTrue,
+          [&](FirstVisitToBlock first_visit,
+              ChangeLogBlock*,
+              MutableBuffer dst,
+              EditOffset edit_offset) {
+            if (first_visit) {
+              ++blocks_seen;
+            }
+            BATT_CHECK_EQ(edit_offset.value(), BATT_CHECKED_CAST(i64, bytes_appended));
+
+            std::memset(dst.data(), bytes_appended & 0xff, dst.size());
+            bytes_appended += dst.size();
+          });
+
+      ASSERT_OK(status);
+
+      // Immediately trim.
+      //
+      ASSERT_OK(this->writer_->trim(EditOffset{(i64)bytes_appended}));
+    }
+  }
+
+  // Create a background thread to eventually close the log writer if sync is taking too long.
+  //
+  std::atomic<bool> cancel_halt_timeout{false};
+  std::thread halt_thread{[&] {
+    for (i32 i = 0; i < kSyncTimeoutSeconds * 10; ++i) {
+      if (cancel_halt_timeout.load()) {
+        return;
+      }
+      std::this_thread::sleep_for(std::chrono::milliseconds{100});
+    }
+    this->writer_->halt();
+  }};
+
+  // The moment of truth: wait for everything to be flushed!
+  //
+  Status sync_status = this->writer_->sync(EditOffset{(i64)kBytesToAppend});
+
+  // Tell the background thread it can shut down now.
+  //
+  cancel_halt_timeout.store(true);
+  halt_thread.join();
+
+  const usize estimated_blocks = bytes_appended / this->config_.block_size;
+
+  EXPECT_GT(blocks_seen, estimated_blocks / 2);
+  EXPECT_LE(blocks_seen, estimated_blocks * 2);
+  EXPECT_EQ(this->writer_->durable_upper_bound().value(), BATT_CHECKED_CAST(i64, bytes_appended));
+  ASSERT_OK(sync_status);
+}
+
 }  // namespace turtle_kv

src/turtle_kv/change_log/change_log_config.hpp

@@ -35,6 +35,22 @@ struct ChangeLogConfig {
 
   //+++++++++++-+-+--+----- --- -- -  -  -   -
 
+  /** \brief Returns the size of the log in bytes.  This does not include the meta-block.
+   */
+  i64 log_size() const noexcept
+  {
+    return this->block_size * this->block_count;
+  }
+
+  /** \brief Sets the size of the log to at least `target` by varying the block_count.  Does not
+   * modify block_size.
+   */
+  void set_log_size(i64 target) noexcept
+  {
+    this->block_count = BlockCount{(target + this->block_size - 1) / this->block_size};
+    BATT_CHECK_GE(this->log_size(), target);
+  }
+
   void pack_to(PackedChangeLogConfig* packed_config) const noexcept;
 
   /** \brief The maximum number of blocks which can be written in a single multi-chunk write.

src/turtle_kv/change_log/change_log_writer.cpp

@@ -856,6 +856,14 @@ Status ChangeLogWriter::activate_blocks(
             << BATT_INSPECT(next_block->edit_offset_lower_bound())
             << BATT_INSPECT(next_block->edit_offset_upper_bound());
 
+    // Collect blocks with slots for advancing the durable upper bound.  IMPORTANT: we must do this
+    // before short-circuiting the loop due to the trim point having moved beyond `next_block`
+    // below!
+    //
+    if (next_block->slot_count() > 0) {
+      newly_activated.emplace_back(next_block);
+    }
+
     // If there are no active blocks and the trim point is already past the next block, just
     // increment the block range and keep going.
     //
@@ -870,12 +878,6 @@ Status ChangeLogWriter::activate_blocks(
       continue;
     }
 
-    // Collect blocks with slots for advancing the durable upper bound.
-    //
-    if (next_block->slot_count() > 0) {
-      newly_activated.emplace_back(next_block);
-    }
-
     // Update active blocks edit offset upper bound.
     //
     output.block_upper_bounds[*input.block_index] = next_block->edit_offset_upper_bound().value();

src/turtle_kv/mem_table/mem_table.test.cpp

@@ -378,9 +378,9 @@ TEST_F(MemTableTest, PutGet)
 //
 TEST_F(MemTableTest, PutUntilFull)
 {
-  usize total_key_bytes = 0;
-  usize total_value_bytes = 0;
-  usize put_count = 0;
+  [[maybe_unused]] usize total_key_bytes = 0;
+  [[maybe_unused]] usize total_value_bytes = 0;
+  [[maybe_unused]] usize put_count = 0;
 
   for (;;) {
     KeyView key = this->make_random_key();