fix(compaction): update compaction catalog on offline compaction

- tglogutil offline compaction now registers the compacted file in the
  compaction catalog (max epoch id, max blob id, pwal_0000.compacted)
- without this, a later startup generated the snapshot as if no compaction
  had been performed, dropping remove entries and resurrecting deleted
  records (tsurugi-issues #1498)
- detect at startup when pwal_0000.compacted exists on disk but is not
  registered in the catalog, and fail fast with initialization_failure
  instead of silently corrupting data
- finalize the expected error-message substring in the offline compaction
  test comment

Author: umegane

src/limestone/datastore.cpp

@@ -146,6 +146,44 @@ datastore::datastore(configuration const& conf) : location_(conf.data_location_)
         add_file(compaction_catalog_path);
         compaction_catalog_ = std::make_unique<compaction_catalog>(compaction_catalog::from_catalog_file(location_));
 
+        // Verify consistency between the compaction catalog and the WAL files.
+        // The compacted file is loaded at startup based purely on its presence on
+        // disk (see snapshot_impl). If it exists but is not registered in the
+        // catalog, the snapshot would be generated as if no compaction had been
+        // performed, dropping remove entries and resurrecting deleted records
+        // (see tsurugi-issues #1498). Fail fast instead of silently corrupting data.
+        // NOTE: only the single well-known compacted file is checked here. If
+        // multiple compacted files are ever registered, this check must be extended.
+        {
+            boost::filesystem::path compacted_file_path = location_ / compaction_catalog::get_compacted_filename();
+            // Use the error_code overload so that a filesystem error (permission, broken
+            // symlink, I/O error, ...) is funneled into a limestone_exception rather than
+            // escaping as a boost::filesystem::filesystem_error.
+            bool compacted_file_exists = boost::filesystem::exists(compacted_file_path, error);
+            if (error) {
+                std::string err_msg = "failed to check existence of the compacted file '"
+                    + compacted_file_path.string() + "': " + error.message();
+                LOG(ERROR) << "/:limestone:config:datastore " << err_msg;
+                throw limestone_exception(exception_type::initialization_failure, err_msg);
+            }
+            if (compacted_file_exists) {
+                bool registered = false;
+                for (auto const& info : compaction_catalog_->get_compacted_files()) {
+                    if (info.get_file_name() == compaction_catalog::get_compacted_filename()) {
+                        registered = true;
+                        break;
+                    }
+                }
+                if (!registered) {
+                    std::string err_msg = "compaction catalog is inconsistent: the compacted file '"
+                        + compaction_catalog::get_compacted_filename()
+                        + "' exists but is not registered in the catalog, log directory: " + location_.string();
+                    LOG(ERROR) << "/:limestone:config:datastore " << err_msg;
+                    throw limestone_exception(exception_type::initialization_failure, err_msg);
+                }
+            }
+        }
+
         epoch_file_path_ = location_ / std::string(limestone::internal::epoch_file_name);
         tmp_epoch_file_path_ = location_ / std::string(limestone::internal::tmp_epoch_file_name);
         const bool result = boost::filesystem::exists(epoch_file_path_, error);

src/limestone/dblogutil/dblogutil.cpp

@@ -26,6 +26,7 @@
 #include "log_entry.h"
 #include "limestone_exception_helper.h"
 #include "manifest.h"
+#include "compaction_catalog.h"
 
 // NOLINTBEGIN(performance-avoid-endl)
 
@@ -227,7 +228,7 @@ void compaction(dblog_scan &ds, std::optional<epoch_id_type> epoch) {
 
     VLOG_LP(log_info) << "making compact pwal file to " << tmp;
     compaction_options options{from_dir, tmp, FLAGS_thread_num};
-    create_compact_pwal_and_get_max_blob_id(options);
+    blob_id_type max_blob_id = create_compact_pwal_and_get_max_blob_id(options);
 
     // epoch file
     VLOG_LP(log_info) << "making compact epoch file to " << tmp;
@@ -247,6 +248,15 @@ void compaction(dblog_scan &ds, std::optional<epoch_id_type> epoch) {
         LOG_AND_THROW_IO_EXCEPTION("fclose failed", errno);
     }
 
+    // Update the compaction catalog so that the compacted file is registered.
+    // 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).
+    VLOG_LP(log_info) << "updating compaction catalog in " << tmp;
+    compaction_catalog catalog{tmp};
+    compacted_file_info compacted_file{compaction_catalog::get_compacted_filename(), 1};
+    catalog.update_catalog_file(ld_epoch, max_blob_id, {compacted_file}, {});
+
     if (FLAGS_dry_run) {
         std::cout << "compaction will be successfully completed (dry-run mode)" << std::endl;
         VLOG_LP(log_info) << "deleting work directory " << tmp;

test/limestone/compaction/offline_compaction_test.cpp

@@ -174,7 +174,7 @@ TEST_F(offline_compaction_test, detects_inconsistent_compaction_catalog_at_start
     // Startup must fail fast because the catalog is inconsistent. Verify both that a
     // limestone_exception is thrown and that its message is the intended one (so that an
     // unrelated failure that happens to throw the same type is not mistaken for success).
-    // TODO: replace this placeholder with the actual message once the detection is implemented.
+    // This substring must match the message produced by datastore startup.
     const std::string expected_message_substr = "compaction catalog is inconsistent";
     try {
         gen_datastore();