refactor(compaction): centralize the max_blob_id high-water mark in the catalog

- update_catalog_file now keeps max_blob_id monotonically non-decreasing, so the
  invariant "blob IDs are never reused" lives in one place
- online compaction drops its now-redundant explicit std::max
- offline compaction (tglogutil) carries over the existing catalog and its backup
  into the work directory, then loads+updates it like online compaction does,
  preserving the high-water mark through the shared update_catalog_file
- check the error_code (other than ENOENT) when probing/copying the catalog files,
  and abort with a clear message when the carried-over catalog is unrecoverable,
  instead of silently losing the high-water mark
- add compaction_catalog::get_catalog_backup_filename()
- tests: update_catalog_file never lowers max_blob_id; offline compaction
  preserves the high-water mark and recovers the catalog from its backup

Author: umegane

src/limestone/compaction_catalog.cpp

@@ -15,6 +15,7 @@
  */
 
 #include <glog/logging.h>
+#include <algorithm>
 #include <fstream>
 #include <stdexcept>
 #include <sstream>
@@ -180,9 +181,12 @@ void compaction_catalog::parse_catalog_entry(const std::string& line, bool& max_
 
 // Method to update the compaction catalog
 void compaction_catalog::update_catalog_file(epoch_id_type max_epoch_id, blob_id_type max_blob_id, const std::set<compacted_file_info>& compacted_files, const std::set<std::string>& detached_pwals) {
-    // Update internal state
+    // Update internal state. The maximum blob ID is a monotonically non-decreasing
+    // high-water mark: blob IDs must never be reused, so it must not drop below the
+    // value already recorded even when the caller passes a smaller value (e.g. a
+    // compaction that observed only the blobs still referenced by live entries).
     max_epoch_id_ = max_epoch_id;
-    max_blob_id_ = max_blob_id;
+    max_blob_id_ = std::max(max_blob_id, max_blob_id_);
     compacted_files_ = compacted_files;
     detached_pwals_ = detached_pwals;
 

src/limestone/compaction_catalog.h

@@ -137,8 +137,15 @@ class compaction_catalog {
      * This method updates the catalog with new compacted files, detached PWALs, the maximum epoch ID,
      * and the maximum blob ID, then writes the updated catalog to a file.
      * 
+     * The maximum blob ID is monotonically non-decreasing: the recorded value never
+     * drops below the value the catalog already holds, even if @p max_blob_id is
+     * smaller. Blob IDs must never be reused, so a compaction (online or offline)
+     * that observes only the blobs still referenced by live entries must not lower
+     * this high-water mark.
+     *
      * @param max_epoch_id The maximum epoch ID to be recorded in the catalog.
-     * @param max_blob_id The maximum blob ID to be recorded in the catalog.
+     * @param max_blob_id The candidate maximum blob ID; the recorded value is the
+     *                    maximum of this and the current value (see above).
      * @param compacted_files Set of compacted files to be included in the catalog.
      * @param detached_pwals Set of detached PWALs to be included in the catalog.
      */
@@ -180,6 +187,13 @@ class compaction_catalog {
      */
     [[nodiscard]] static inline std::string get_catalog_filename() { return COMPACTION_CATALOG_FILENAME; }
 
+    /**
+     * @brief Returns the filename of the compaction catalog backup.
+     *
+     * @return The filename of the compaction catalog backup.
+     */
+    [[nodiscard]] static inline std::string get_catalog_backup_filename() { return COMPACTION_CATALOG_BACKUP_FILENAME; }
+
     /**
      * @brief Retrieves the name of the compaction temporary directory.
      *

src/limestone/datastore.cpp

@@ -1014,9 +1014,11 @@ void datastore::compact_with_online() {
 
 
     // update compaction catalog
+    // update_catalog_file keeps max_blob_id monotonically non-decreasing, so the
+    // high-water mark recorded by previous compactions is preserved even though
+    // max_blob_id here reflects only the freshly compacted files.
     compacted_file_info compacted_file_info{compacted_file.filename().string(), 1};
     detached_pwals.erase(compacted_file.filename().string());
-    max_blob_id = std::max(max_blob_id, compaction_catalog_->get_max_blob_id());
     compaction_catalog_->update_catalog_file(result.get_epoch_id(), max_blob_id, {compacted_file_info}, detached_pwals);
     add_file(compacted_file);
 

src/limestone/dblogutil/dblogutil.cpp

@@ -252,8 +252,51 @@ void compaction(dblog_scan &ds, std::optional<epoch_id_type> epoch) {
     // Without this, a subsequent startup treats the directory as if no compaction
     // had been performed, and remove entries are dropped from the snapshot,
     // resurrecting deleted records (see tsurugi-issues #1498).
+    //
+    // tmp will replace from_dir, so carry over the existing catalog (and its backup)
+    // before updating it, exactly like online compaction updates the live catalog.
+    // This preserves the high-water marks recorded by previous compactions: in
+    // particular update_catalog_file keeps max_blob_id monotonically non-decreasing,
+    // so the freshly computed max_blob_id (which reflects only blobs still referenced
+    // by live entries) never lowers it and blob IDs are never reused. If from_dir has
+    // no catalog (a directory that was never compacted), the empty catalog created by
+    // setup_initial_logdir(tmp) is used instead.
+    auto copy_catalog_file = [&](const std::string& filename) {
+        boost::filesystem::path src = from_dir / filename;
+        boost::system::error_code ec;
+        bool present = boost::filesystem::exists(src, ec);
+        // A non-existent file is the normal case (boost reports it via ec as ENOENT).
+        // Any other error (permission, I/O, ...) must not be silently ignored: skipping
+        // the copy would drop the catalog carry-over and let later processing run on an
+        // inconsistent state.
+        if (ec && ec != boost::system::errc::no_such_file_or_directory) {
+            LOG_AND_THROW_IO_EXCEPTION("failed to check existence of compaction catalog file: " + src.string(), ec);
+        }
+        if (present) {
+            boost::filesystem::copy_file(src, tmp / filename, boost::filesystem::copy_options::overwrite_existing, ec);
+            if (ec) {
+                LOG_AND_THROW_IO_EXCEPTION("failed to copy compaction catalog file: " + src.string(), ec);
+            }
+        }
+    };
+    copy_catalog_file(compaction_catalog::get_catalog_filename());
+    copy_catalog_file(compaction_catalog::get_catalog_backup_filename());
+
     VLOG_LP(log_info) << "updating compaction catalog in " << tmp;
-    compaction_catalog catalog{tmp};
+    // tmp always has a catalog here (setup_initial_logdir created an empty one, possibly
+    // overwritten by the carry-over above), so loading only fails when the carried-over
+    // catalog and its backup are both unreadable. In that case abort with a clear message
+    // rather than silently proceeding, which would lose the blob-id high-water mark.
+    compaction_catalog catalog = [&]() {
+        try {
+            return compaction_catalog::from_catalog_file(tmp);
+        } catch (const limestone_exception& ex) {
+            LOG_AND_THROW_EXCEPTION(
+                "the existing compaction catalog in " + from_dir.string() +
+                " is unreadable and could not be recovered; offline compaction was aborted to"
+                " avoid losing the blob-id high-water mark (cause: " + ex.what() + ")");
+        }
+    }();
     compacted_file_info compacted_file{compaction_catalog::get_compacted_filename(), 1};
     catalog.update_catalog_file(ld_epoch, max_blob_id, {compacted_file}, {});
 

test/limestone/compaction/compaction_catalog_test.cpp

@@ -873,6 +873,31 @@ TEST_F(compaction_catalog_test, restore_from_backup_exceptions) {
         limestone_io_exception);
 }
 
+// Verifies that max_blob_id is a monotonically non-decreasing high-water mark:
+// update_catalog_file must never lower it (blob IDs must never be reused), while
+// max_epoch_id is set directly.
+TEST_F(compaction_catalog_test, max_blob_id_is_monotonically_non_decreasing) {
+    compaction_catalog catalog(test_dir);
+
+    catalog.update_catalog_file(10, 1000, {}, {});
+    EXPECT_EQ(catalog.get_max_blob_id(), 1000);
+    EXPECT_EQ(catalog.get_max_epoch_id(), 10);
+
+    // A smaller value must not lower the high-water mark.
+    catalog.update_catalog_file(20, 500, {}, {});
+    EXPECT_EQ(catalog.get_max_blob_id(), 1000);
+    EXPECT_EQ(catalog.get_max_epoch_id(), 20);  // epoch is set directly, not maxed
+
+    // A larger value updates it.
+    catalog.update_catalog_file(30, 2000, {}, {});
+    EXPECT_EQ(catalog.get_max_blob_id(), 2000);
+
+    // The preserved value survives a reload from the catalog file.
+    compaction_catalog reloaded = compaction_catalog::from_catalog_file(test_dir);
+    EXPECT_EQ(reloaded.get_max_blob_id(), 2000);
+    EXPECT_EQ(reloaded.get_max_epoch_id(), 30);
+}
+
 
 
 }  // namespace limestone::testing

test/limestone/compaction/offline_compaction_test.cpp

@@ -18,6 +18,7 @@
 #include <cerrno>
 #include <cstdio>
 #include <cstring>
+#include <fstream>
 #include <map>
 #include <set>
 #include <sstream>
@@ -191,4 +192,64 @@ TEST_F(offline_compaction_test, detects_inconsistent_compaction_catalog_at_start
     }
 }
 
+// Offline compaction must preserve the max_blob_id high-water mark recorded in the
+// existing catalog, even though the freshly computed value (no blobs here) is lower.
+// Lowering it could lead to blob-id reuse.
+TEST_F(offline_compaction_test, offline_compaction_preserves_blob_id_high_water_mark) {
+    gen_datastore();
+    datastore_->switch_epoch(1);
+    lc0_->begin_session();
+    lc0_->add_entry(1, "key1", "value1", {1, 0});  // no blob -> computed max_blob_id is 0
+    lc0_->end_session();
+    datastore_->switch_epoch(2);
+    datastore_->shutdown();
+    datastore_ = nullptr;
+
+    // Simulate a high-water mark left by earlier blob allocations / online compactions.
+    {
+        compaction_catalog catalog = compaction_catalog::from_catalog_file(location);
+        catalog.update_catalog_file(catalog.get_max_epoch_id(), 9999, {}, {});
+    }
+
+    run_offline_compaction();
+
+    // The high-water mark must survive offline compaction (not lowered to the computed 0).
+    compaction_catalog catalog = compaction_catalog::from_catalog_file(location);
+    EXPECT_EQ(catalog.get_max_blob_id(), 9999);
+}
+
+// Offline compaction must recover the existing catalog from its backup when the main
+// catalog file is unreadable, and still preserve the high-water mark.
+TEST_F(offline_compaction_test, offline_compaction_recovers_catalog_from_backup) {
+    gen_datastore();
+    datastore_->switch_epoch(1);
+    lc0_->begin_session();
+    lc0_->add_entry(1, "key1", "value1", {1, 0});
+    lc0_->end_session();
+    datastore_->switch_epoch(2);
+    datastore_->shutdown();
+    datastore_ = nullptr;
+
+    {
+        compaction_catalog catalog = compaction_catalog::from_catalog_file(location);
+        catalog.update_catalog_file(catalog.get_max_epoch_id(), 7777, {}, {});
+    }
+
+    // Leave the valid content only in the backup, then corrupt the main catalog file.
+    boost::filesystem::path dir{location};
+    boost::filesystem::path main_catalog = dir / compaction_catalog::get_catalog_filename();
+    boost::filesystem::path backup_catalog = dir / compaction_catalog::get_catalog_backup_filename();
+    boost::filesystem::copy_file(main_catalog, backup_catalog, boost::filesystem::copy_options::overwrite_existing);
+    {
+        std::ofstream ofs(main_catalog.string(), std::ios::trunc);
+        ofs << "GARBAGE";
+    }
+
+    run_offline_compaction();
+
+    // The catalog must have been recovered from the backup and the high-water mark kept.
+    compaction_catalog catalog = compaction_catalog::from_catalog_file(location);
+    EXPECT_EQ(catalog.get_max_blob_id(), 7777);
+}
+
 }  // namespace limestone::testing