fix(gc): file-GC snapshot UAF + MakeCowRoot error ordering in Apply

include/storage/root_meta.h

@@ -213,6 +213,14 @@ class RootMetaMgr
 
     KvError EvictIfNeeded();
 
+#ifndef NDEBUG
+    // Test-only: force-evict a specific table's root meta entry, mimicking the
+    // LRU victim path of EvictIfNeeded (EvictRootForCache + Dequeue + erase).
+    // Used to deterministically reproduce the GC-holds-snapshot-across-eviction
+    // UAF. Returns true iff the entry existed, was unpinned, and was erased.
+    bool ForceEvictForTest(const TableIdent &tbl_id);
+#endif
+
 private:
     PageManager *owner_;
     void EnqueueFront(Entry *entry);

src/storage/root_meta.cpp

@@ -442,6 +442,35 @@ KvError RootMetaMgr::EvictIfNeeded()
     return KvError::NoError;
 }
 
+#ifndef NDEBUG
+bool RootMetaMgr::ForceEvictForTest(const TableIdent &tbl_id)
+{
+    Entry *victim = Find(tbl_id);
+    // Only an unpinned entry that is currently on the LRU list (prev_ != null)
+    // can be evicted, matching the invariants EvictIfNeeded relies on.
+    if (victim == nullptr || victim->meta_.ref_cnt_ != 0 ||
+        victim->prev_ == nullptr)
+    {
+        return false;
+    }
+    if (!EvictRootForCache(victim))
+    {
+        return false;
+    }
+    Dequeue(victim);
+    if (used_bytes_ >= victim->bytes_)
+    {
+        used_bytes_ -= victim->bytes_;
+    }
+    else
+    {
+        used_bytes_ = 0;
+    }
+    entries_.erase(victim->tbl_id_);
+    return true;
+}
+#endif
+
 void RootMetaMgr::EnqueueFront(Entry *entry)
 {
     CHECK(entry->prev_ == nullptr && entry->next_ == nullptr)

src/tasks/batch_write_task.cpp

@@ -254,8 +254,8 @@ KvError BatchWriteTask::Apply()
     // directly go to low priority queue and wait for scheduling
     YieldToLowPQ();
     KvError err = MakeCowRoot();
-    cow_meta_.compression_->SampleAndBuildDictionaryIfNeeded(data_batch_);
     CHECK_KV_ERR(err);
+    cow_meta_.compression_->SampleAndBuildDictionaryIfNeeded(data_batch_);
     err = ApplyBatch(cow_meta_.root_id_, true);
     if (err != KvError::NoError)
     {

src/tasks/write_task.cpp

@@ -13,6 +13,7 @@
 #include "absl/container/flat_hash_set.h"
 #include "async_io_manager.h"
 #include "error.h"
+#include "fail_point.h"
 #include "file_gc.h"
 #include "storage/data_page.h"
 #include "storage/mem_cached_page.h"
@@ -948,6 +949,27 @@ void WriteTask::TriggerFileGC() const
 {
     assert(Options()->data_append_mode);
 
+    // Pin the RootMeta entry for the entire GC. BuildRetainedFiles captures
+    // MappingSnapshot::Refs whose tbl_ident_ points into this entry, and
+    // ExecuteLocalGC yields; without a held Handle another task could evict the
+    // entry from the RootMeta cache (BuildRetainedFiles' own per-call Handle is
+    // already released) and free it while the snapshots still reference it,
+    // producing a use-after-free when snapshot_array is destroyed below and
+    // FreeMappingSnapshot dereferences the dangling tbl_ident_. Declared before
+    // the snapshot arrays so it outlives their destruction at function exit.
+    //
+    // When the entry is absent (NotFound, e.g. after a Drop cleared it) there
+    // are no snapshots to dangle, so the handle is a no-op; GC must still run
+    // with empty retained sets to purge orphaned files / the partition dir.
+    auto [root_handle, find_err] = shard->IndexManager()->FindRoot(tbl_ident_);
+    if (find_err != KvError::NoError && find_err != KvError::NotFound)
+    {
+        LOG(ERROR) << "TriggerFileGC: FindRoot failed for table "
+                   << tbl_ident_.ToString()
+                   << " err=" << static_cast<int>(find_err);
+        return;
+    }
+
     RetainedFiles retained_files;
     std::vector<MappingSnapshot::Ref> snapshot_array;
     KvError build_err = BuildRetainedFiles(
@@ -977,6 +999,18 @@ void WriteTask::TriggerFileGC() const
         return;
     }
 
+    // Test-only fault injection for the GC-holds-snapshot-across-eviction UAF:
+    // with snapshot_array / seg_snapshot_array still holding refs, force-evict
+    // this partition's RootMeta entry so tbl_ident_ inside those snapshots
+    // dangles into the freed entry. FreeMappingSnapshot then reads it when the
+    // snapshot arrays are destroyed at function exit. Compiled out in release.
+#ifndef NDEBUG
+    if (FailPoint::GetInstance().ShouldFail("GcForceEvictRoot"))
+    {
+        shard->IndexManager()->RootMetaManager()->ForceEvictForTest(tbl_ident_);
+    }
+#endif
+
     // Check if we're in cloud mode or local mode
     if (eloq_store->Mode() == StoreMode::Cloud)
     {
@@ -1015,6 +1049,18 @@ void WriteTask::TriggerFileGC() const
 KvError WriteTask::TriggerLocalFileGC() const
 {
     assert(Options()->data_append_mode);
+
+    // Pin the RootMeta entry across the whole GC so it cannot be evicted (and
+    // freed) while snapshot_array holds MappingSnapshot::Refs into it; see the
+    // rationale in TriggerFileGC. Declared before snapshot_array so it outlives
+    // that array's destruction. A NotFound entry has no snapshots, so the
+    // handle is a no-op and GC still runs with empty retained sets.
+    auto [root_handle, find_err] = shard->IndexManager()->FindRoot(tbl_ident_);
+    if (find_err != KvError::NoError && find_err != KvError::NotFound)
+    {
+        return find_err;
+    }
+
     RetainedFiles retained_files;
     std::vector<MappingSnapshot::Ref> snapshot_array;
     KvError build_err = BuildRetainedFiles(

tests/gc.cpp

@@ -9,6 +9,7 @@
 #include <vector>
 
 #include "common.h"
+#include "fail_point.h"
 #include "kv_options.h"
 #include "test_utils.h"
 #include "utils.h"
@@ -492,3 +493,50 @@ TEST_CASE("local GC honors in-memory least_unflushed boundary",
     store->Stop();
     CleanupStore(opts);
 }
+
+// Regression test for the file-GC use-after-free: TriggerFileGC builds
+// snapshot_array (MappingSnapshot::Refs whose tbl_ident_ points into the
+// partition's RootMeta entry), then yields across ExecuteLocalGC. If the
+// RootMeta entry is evicted from the cache during that window (its Handle was
+// already released, so ref_cnt_==0 and it is an LRU victim), the entry is
+// freed while the snapshots still reference it. When snapshot_array is
+// destroyed at TriggerFileGC exit, FreeMappingSnapshot dereferences the
+// dangling tbl_ident_.
+//
+// The eviction/GC interleaving is a narrow race, so we drive it deterministic-
+// ally: the "GcForceEvictRoot" fault point (armed here, fired once inside
+// TriggerFileGC after both snapshot arrays are built) performs exactly the
+// eviction the LRU would do under cache pressure. Under ASAN this aborts with
+// heap-use-after-free in FreeMappingSnapshot on unfixed code.
+TEST_CASE("gc snapshot refs survive root meta eviction", "[gc][local][uaf]")
+{
+    eloqstore::KvOptions opts = local_gc_opts;
+    opts.store_path = {"/tmp/test-gc-uaf"};
+    opts.num_threads = 1;
+    CleanupStore(opts);
+
+    eloqstore::EloqStore *store = InitStore(opts);
+    eloqstore::TableIdent tbl{"gc_uaf_partition", 0};
+
+    // Populate the partition so its live mapping is non-empty: TriggerFileGC
+    // then builds a non-empty snapshot_array, and Compact() does not take its
+    // "nothing to compact" short-circuit.
+    {
+        MapVerifier v(tbl, store, false);
+        v.SetValueSize(400);
+        v.Upsert(0, 200);
+        v.Upsert(0, 200);  // overwrite -> dead pages -> real compaction work
+    }
+
+    // Arm the fault point, then force a compaction whose TriggerFileGC evicts
+    // this partition's root meta while its GC snapshots are still live.
+    eloqstore::FailPoint::GetInstance().ArmOnce("GcForceEvictRoot");
+    eloqstore::CompactRequest compact;
+    compact.SetTableId(tbl);
+    store->ExecSync(&compact);
+    eloqstore::FailPoint::GetInstance().Disarm();
+    REQUIRE(compact.Error() == eloqstore::KvError::NoError);
+
+    store->Stop();
+    CleanupStore(opts);
+}