Merge pull request #14294 from mszeszko-meta/cherry_pick_14280_to_10_10

10.10.1 Patch Release

.github/workflows/nightly.yml

@@ -110,6 +110,20 @@ jobs:
       - run: sudo apt-get update && sudo apt-get install -y build-essential libgflags-dev
       - run: make V=1 J=4 -j4 check
       - uses: "./.github/actions/post-steps"
+  build-linux-arm-crashtest:
+    if: ${{ github.repository_owner == 'facebook' }}
+    runs-on:
+      labels: 4-core-ubuntu-arm
+    steps:
+    - uses: actions/checkout@v4.1.0
+    - uses: "./.github/actions/pre-steps"
+    - run: sudo apt-get update && sudo apt-get install -y build-essential libgflags-dev libsnappy-dev zlib1g-dev libbz2-dev liblz4-dev libzstd-dev
+    - run: sudo mount -o remount,size=16G /dev/shm
+    - run: sudo dd bs=1048576 count=4096 if=/dev/zero of=/swapfile && sudo chmod 600 /swapfile && sudo mkswap /swapfile && sudo swapon /swapfile
+    - run: ulimit -S -n `ulimit -H -n` && make V=1 -j8 CRASH_TEST_EXT_ARGS='--duration=1800 --max_key=2500000' blackbox_crash_test_with_atomic_flush
+    - run: rm -rf /dev/shm/rocksdb.*
+    - run: ulimit -S -n `ulimit -H -n` && make V=1 -j8 CRASH_TEST_EXT_ARGS='--duration=1800 --max_key=2500000' blackbox_crash_test_with_multiops_wc_txn
+    - uses: "./.github/actions/post-steps"
   build-examples:
     if: ${{ github.repository_owner == 'facebook' }}
     runs-on:

CMakeLists.txt

@@ -219,6 +219,10 @@ if(MSVC)
     set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /Zi /nologo /EHsc /GS /Gd /GR /GF /fp:precise /Zc:wchar_t /Zc:forScope /errorReport:queue")
     set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /FC /d2Zi+ /W4 /wd4127 /wd4996 /wd4100 /wd4324")
   endif()
+  if(CMAKE_BUILD_TYPE STREQUAL "Release")
+    set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /DNDEBUG")
+    message(STATUS "Setting /DNDEBUG as CMAKE_BUILD_TYPE is set to ${CMAKE_BUILD_TYPE}")
+  endif()
 else()
   set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -W -Wextra -Wall -pthread")
   set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wsign-compare -Wshadow -Wno-unused-parameter -Wno-unused-variable -Woverloaded-virtual -Wnon-virtual-dtor -Wno-missing-field-initializers -Wno-strict-aliasing -Wno-invalid-offsetof")

HISTORY.md

@@ -1,9 +1,15 @@
 # Rocksdb Change Log
 > NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
 
-## 10.9.1 (12/11/2025)
+## 10.10.1 (02/02/2026)
 ### Bug Fixes
+* Fix Windows VS 2022 build errors.
+
+## 10.10.0 (12/16/2025)
+### Bug Fixes
+* Fixed a bug in best-efforts recovery that causes use-after-free crashes when accessing SST files that were cached during the recovery.
 * Fix resumable compaction incorrectly allowing resumption from a truncated range deletion that is not well handled currently.
+* Fixed a bug in `PosixRandomFileAccess` IO uring submission queue ownership & management. Fix eliminates the false positive 'Bad cqe data' IO errors in `PosixRandomFileAccess::MultiRead` when interleaved with `PosixRandomFileAccess::ReadAsync` on the same thread.
 
 ## 10.9.0 (11/21/2025)
 ### New Features

Makefile

@@ -296,6 +296,28 @@ $(info $(shell $(CC) --version))
 $(info $(shell $(CXX) --version))
 endif
 
+# ccache support
+# Set USE_CCACHE=1 to enable ccache, or let it auto-detect
+ifndef USE_CCACHE
+  CCACHE := $(shell which ccache 2>/dev/null)
+  ifneq ($(CCACHE),)
+    USE_CCACHE := 1
+  else
+    USE_CCACHE := 0
+  endif
+endif
+
+ifeq ($(USE_CCACHE), 1)
+  CCACHE := $(shell which ccache 2>/dev/null)
+  ifneq ($(CCACHE),)
+    $(info Using ccache: $(CCACHE))
+    CC := $(CCACHE) $(CC)
+    CXX := $(CCACHE) $(CXX)
+  else
+    $(warning ccache requested but not found in PATH)
+  endif
+endif
+
 missing_make_config_paths := $(shell				\
 	grep "\./\S*\|/\S*" -o $(CURDIR)/make_config.mk | 	\
 	while read path;					\

cache/clock_cache.cc

@@ -1752,26 +1752,6 @@ inline void GetHomeIndexAndShift(uint64_t length_info, uint64_t hash,
   assert(*home < LengthInfoToUsedLength(length_info));
 }
 
-inline int GetShiftFromNextWithShift(uint64_t next_with_shift) {
-  return BitwiseAnd(next_with_shift,
-                    AutoHyperClockTable::HandleImpl::kShiftMask);
-}
-
-inline size_t GetNextFromNextWithShift(uint64_t next_with_shift) {
-  return static_cast<size_t>(next_with_shift >>
-                             AutoHyperClockTable::HandleImpl::kNextShift);
-}
-
-inline uint64_t MakeNextWithShift(size_t next, int shift) {
-  return (uint64_t{next} << AutoHyperClockTable::HandleImpl::kNextShift) |
-         static_cast<uint64_t>(shift);
-}
-
-inline uint64_t MakeNextWithShiftEnd(size_t head, int shift) {
-  return AutoHyperClockTable::HandleImpl::kNextEndFlags |
-         MakeNextWithShift(head, shift);
-}
-
 // Helper function for Lookup
 inline bool MatchAndRef(const UniqueId64x2* hashed_key, const ClockHandle& h,
                         int shift = 0, size_t home = 0,
@@ -1821,36 +1801,39 @@ inline bool MatchAndRef(const UniqueId64x2* hashed_key, const ClockHandle& h,
   }
 }
 
+using NextWithShift = AutoHyperClockTable::HandleImpl::NextWithShift;
+
 // Assumes a chain rewrite lock prevents concurrent modification of
 // these chain pointers
 void UpgradeShiftsOnRange(AutoHyperClockTable::HandleImpl* arr,
-                          size_t& frontier, uint64_t stop_before_or_new_tail,
-                          int old_shift, int new_shift) {
+                          size_t& frontier,
+                          NextWithShift stop_before_or_new_tail, int old_shift,
+                          int new_shift) {
   assert(frontier != SIZE_MAX);
   assert(new_shift == old_shift + 1);
   (void)old_shift;
   (void)new_shift;
-  using HandleImpl = AutoHyperClockTable::HandleImpl;
   for (;;) {
-    uint64_t next_with_shift = arr[frontier].chain_next_with_shift.Load();
-    assert(GetShiftFromNextWithShift(next_with_shift) == old_shift);
+    NextWithShift next_with_shift = arr[frontier].chain_next_with_shift.Load();
+    assert(next_with_shift.GetShift() == old_shift);
     if (next_with_shift == stop_before_or_new_tail) {
       // Stopping at entry with pointer matching "stop before"
-      assert(!HandleImpl::IsEnd(next_with_shift));
+      assert(!next_with_shift.IsEnd());
       return;
     }
-    if (HandleImpl::IsEnd(next_with_shift)) {
+    if (next_with_shift.IsEnd()) {
       // Also update tail to new tail
-      assert(HandleImpl::IsEnd(stop_before_or_new_tail));
+      assert(stop_before_or_new_tail.IsEnd());
       arr[frontier].chain_next_with_shift.Store(stop_before_or_new_tail);
       // Mark nothing left to upgrade
       frontier = SIZE_MAX;
       return;
     }
     // Next is another entry to process, so upgrade and advance frontier
-    arr[frontier].chain_next_with_shift.FetchAdd(1U);
-    assert(GetShiftFromNextWithShift(next_with_shift + 1) == new_shift);
-    frontier = GetNextFromNextWithShift(next_with_shift);
+    arr[frontier].chain_next_with_shift.Apply(
+        NextWithShift::Shift::PlusTransformPromiseNoOverflow(1U));
+    assert(next_with_shift.GetShift() + 1 == new_shift);
+    frontier = next_with_shift.GetNext();
   }
 }
 
@@ -1888,19 +1871,19 @@ class AutoHyperClockTable::ChainRewriteLock {
   // RAII wrap existing lock held (or end)
   explicit ChainRewriteLock(HandleImpl* h,
                             RelaxedAtomic<uint64_t>& /*yield_count*/,
-                            uint64_t already_locked_or_end)
+                            NextWithShift already_locked_or_end)
       : head_ptr_(&h->head_next_with_shift) {
     saved_head_ = already_locked_or_end;
     // already locked or end
-    assert(saved_head_ & HandleImpl::kHeadLocked);
+    assert(saved_head_.IsLocked());
   }
 
   ~ChainRewriteLock() {
     if (!IsEnd()) {
       // Release lock
-      uint64_t old = head_ptr_->FetchAnd(~HandleImpl::kHeadLocked);
-      (void)old;
-      assert((old & HandleImpl::kNextEndFlags) == HandleImpl::kHeadLocked);
+      NextWithShift old;
+      head_ptr_->Apply(NextWithShift::LockedFlag::ClearTransform(), &old);
+      assert(old.IsLockedNotEnd());
     }
   }
 
@@ -1910,12 +1893,13 @@ class AutoHyperClockTable::ChainRewriteLock {
   }
 
   // Expected current state, assuming no parallel updates.
-  uint64_t GetSavedHead() const { return saved_head_; }
+  NextWithShift GetSavedHead() const { return saved_head_; }
 
-  bool CasUpdate(uint64_t next_with_shift,
+  bool CasUpdate(NextWithShift next_with_shift,
                  RelaxedAtomic<uint64_t>& yield_count) {
-    uint64_t new_head = next_with_shift | HandleImpl::kHeadLocked;
-    uint64_t expected = GetSavedHead();
+    NextWithShift new_head =
+        next_with_shift.With<NextWithShift::LockedFlag>(true);
+    NextWithShift expected = GetSavedHead();
     bool success = head_ptr_->CasStrong(expected, new_head);
     if (success) {
       // Ensure IsEnd() is kept up-to-date, including for dtor
@@ -1924,7 +1908,7 @@ class AutoHyperClockTable::ChainRewriteLock {
       // Parallel update to head, such as Insert()
       if (IsEnd()) {
         // Didn't previously hold a lock
-        if (HandleImpl::IsEnd(expected)) {
+        if (expected.IsEnd()) {
           // Still don't need to
           saved_head_ = expected;
         } else {
@@ -1933,28 +1917,25 @@ class AutoHyperClockTable::ChainRewriteLock {
         }
       } else {
         // Parallel update must preserve our lock
-        assert((expected & HandleImpl::kNextEndFlags) ==
-               HandleImpl::kHeadLocked);
+        assert(expected.IsLockedNotEnd());
         saved_head_ = expected;
       }
     }
     return success;
   }
 
-  bool IsEnd() const { return HandleImpl::IsEnd(saved_head_); }
+  bool IsEnd() const { return saved_head_.IsEnd(); }
 
  private:
   void Acquire(RelaxedAtomic<uint64_t>& yield_count) {
     for (;;) {
       // Acquire removal lock on the chain
-      uint64_t old_head = head_ptr_->FetchOr(HandleImpl::kHeadLocked);
-      if ((old_head & HandleImpl::kNextEndFlags) != HandleImpl::kHeadLocked) {
+      NextWithShift old_head;
+      head_ptr_->Apply(NextWithShift::LockedFlag::SetTransform(), &old_head,
+                       &saved_head_);
+      if (!old_head.IsLockedNotEnd()) {
         // Either acquired the lock or lock not needed (end)
-        assert((old_head & HandleImpl::kNextEndFlags) == 0 ||
-               (old_head & HandleImpl::kNextEndFlags) ==
-                   HandleImpl::kNextEndFlags);
-
-        saved_head_ = old_head | HandleImpl::kHeadLocked;
+        assert(old_head.IsEnd() == old_head.IsLocked());
         break;
       }
       // NOTE: one of the few yield-wait loops, which is rare enough in practice
@@ -1965,8 +1946,8 @@ class AutoHyperClockTable::ChainRewriteLock {
     }
   }
 
-  AcqRelAtomic<uint64_t>* head_ptr_;
-  uint64_t saved_head_;
+  AcqRelBitFieldsAtomic<NextWithShift>* head_ptr_;
+  NextWithShift saved_head_;
 };
 
 AutoHyperClockTable::AutoHyperClockTable(
@@ -2021,9 +2002,9 @@ AutoHyperClockTable::AutoHyperClockTable(
 #endif
     if (major + i < used_length) {
       array_[i].head_next_with_shift.StoreRelaxed(
-          MakeNextWithShiftEnd(i, max_shift));
+          NextWithShift::MakeEnd(i, max_shift));
       array_[major + i].head_next_with_shift.StoreRelaxed(
-          MakeNextWithShiftEnd(major + i, max_shift));
+          NextWithShift::MakeEnd(major + i, max_shift));
 #ifndef NDEBUG  // Extra invariant checking
       GetHomeIndexAndShift(length_info, i, &home, &shift);
       assert(home == i);
@@ -2034,7 +2015,7 @@ AutoHyperClockTable::AutoHyperClockTable(
 #endif
     } else {
       array_[i].head_next_with_shift.StoreRelaxed(
-          MakeNextWithShiftEnd(i, min_shift));
+          NextWithShift::MakeEnd(i, min_shift));
 #ifndef NDEBUG  // Extra invariant checking
       GetHomeIndexAndShift(length_info, i, &home, &shift);
       assert(home == i);
@@ -2066,8 +2047,10 @@ AutoHyperClockTable::~AutoHyperClockTable() {
   // just a reasonable frontier past what we expect to have written.
 #ifdef MUST_FREE_HEAP_ALLOCATIONS
   for (size_t i = used_end; i < array_.Count() && i < used_end + 64U; i++) {
-    assert(array_[i].head_next_with_shift.LoadRelaxed() == 0);
-    assert(array_[i].chain_next_with_shift.LoadRelaxed() == 0);
+    assert(array_[i].head_next_with_shift.LoadRelaxed() ==
+           HandleImpl::kUnusedMarker);
+    assert(array_[i].chain_next_with_shift.LoadRelaxed() ==
+           HandleImpl::kUnusedMarker);
     assert(array_[i].meta.LoadRelaxed() == 0);
   }
 #endif          // MUST_FREE_HEAP_ALLOCATIONS
@@ -2089,11 +2072,9 @@ AutoHyperClockTable::~AutoHyperClockTable() {
         usage_.FetchSubRelaxed(h.total_charge);
         occupancy_.FetchSubRelaxed(1U);
         was_populated[i] = true;
-        if (!HandleImpl::IsEnd(h.chain_next_with_shift.LoadRelaxed())) {
-          assert((h.chain_next_with_shift.LoadRelaxed() &
-                  HandleImpl::kHeadLocked) == 0);
-          size_t next =
-              GetNextFromNextWithShift(h.chain_next_with_shift.LoadRelaxed());
+        if (!h.chain_next_with_shift.LoadRelaxed().IsEnd()) {
+          assert(!h.chain_next_with_shift.LoadRelaxed().IsLocked());
+          size_t next = h.chain_next_with_shift.LoadRelaxed().GetNext();
           assert(!was_pointed_to[next]);
           was_pointed_to[next] = true;
         }
@@ -2105,9 +2086,8 @@ AutoHyperClockTable::~AutoHyperClockTable() {
         break;
     }
 #ifndef NDEBUG  // Extra invariant checking
-    if (!HandleImpl::IsEnd(h.head_next_with_shift.LoadRelaxed())) {
-      size_t next =
-          GetNextFromNextWithShift(h.head_next_with_shift.LoadRelaxed());
+    if (!h.head_next_with_shift.LoadRelaxed().IsEnd()) {
+      size_t next = h.head_next_with_shift.LoadRelaxed().GetNext();
       assert(!was_pointed_to[next]);
       was_pointed_to[next] = true;
     }
@@ -2222,10 +2202,10 @@ bool AutoHyperClockTable::Grow(InsertState& state) {
   // chain rewrite lock has been released.
   size_t old_old_home = BottomNBits(grow_home, old_shift - 1);
   for (;;) {
-    uint64_t old_old_head = array_[old_old_home].head_next_with_shift.Load();
-    if (GetShiftFromNextWithShift(old_old_head) >= old_shift) {
-      if ((old_old_head & HandleImpl::kNextEndFlags) !=
-          HandleImpl::kHeadLocked) {
+    NextWithShift old_old_head =
+        array_[old_old_home].head_next_with_shift.Load();
+    if (old_old_head.GetShift() >= old_shift) {
+      if (!old_old_head.IsLockedNotEnd()) {
         break;
       }
     }
@@ -2285,8 +2265,7 @@ void AutoHyperClockTable::CatchUpLengthInfoNoWait(
     if (published_usable_size < known_usable_grow_home) {
       int old_shift = FloorLog2(next_usable_size - 1);
       size_t old_home = BottomNBits(published_usable_size, old_shift);
-      int shift = GetShiftFromNextWithShift(
-          array_[old_home].head_next_with_shift.Load());
+      int shift = array_[old_home].head_next_with_shift.Load().GetShift();
       if (shift <= old_shift) {
         // Not ready
         break;
@@ -2437,9 +2416,10 @@ void AutoHyperClockTable::SplitForGrow(size_t grow_home, size_t old_home,
   ChainRewriteLock zero_head_lock(&arr[old_home], yield_count_);
 
   // Used for locking the one chain below
-  uint64_t saved_one_head;
+  NextWithShift saved_one_head;
   // One head has not been written to
-  assert(arr[grow_home].head_next_with_shift.Load() == 0);
+  assert(arr[grow_home].head_next_with_shift.Load() ==
+         HandleImpl::kUnusedMarker);
 
   // old_home will also the head of the new "zero chain" -- all entries in the
   // "from" chain whose next hash bit is 0. grow_home will be head of the new
@@ -2461,7 +2441,7 @@ void AutoHyperClockTable::SplitForGrow(size_t grow_home, size_t old_home,
     assert(cur == SIZE_MAX);
     assert(chain_frontier_first == -1);
 
-    uint64_t next_with_shift = zero_head_lock.GetSavedHead();
+    NextWithShift next_with_shift = zero_head_lock.GetSavedHead();
 
     // Find a single representative for each target chain, or scan the whole
     // chain if some target chain has no representative.
@@ -2474,16 +2454,16 @@ void AutoHyperClockTable::SplitForGrow(size_t grow_home, size_t old_home,
       assert((cur == SIZE_MAX) == (zero_chain_frontier == SIZE_MAX &&
                                    one_chain_frontier == SIZE_MAX));
 
-      assert(GetShiftFromNextWithShift(next_with_shift) == old_shift);
+      assert(next_with_shift.GetShift() == old_shift);
 
       // Check for end of original chain
-      if (HandleImpl::IsEnd(next_with_shift)) {
+      if (next_with_shift.IsEnd()) {
         cur = SIZE_MAX;
         break;
       }
 
       // next_with_shift is not End
-      cur = GetNextFromNextWithShift(next_with_shift);
+      cur = next_with_shift.GetNext();
 
       if (BottomNBits(arr[cur].hashed_key[1], new_shift) == old_home) {
         // Entry for zero chain
@@ -2522,10 +2502,10 @@ void AutoHyperClockTable::SplitForGrow(size_t grow_home, size_t old_home,
            (zero_chain_frontier == SIZE_MAX && one_chain_frontier == SIZE_MAX));
 
     // Always update one chain's head first (safe), and mark it as locked
-    saved_one_head = HandleImpl::kHeadLocked |
-                     (one_chain_frontier != SIZE_MAX
-                          ? MakeNextWithShift(one_chain_frontier, new_shift)
-                          : MakeNextWithShiftEnd(grow_home, new_shift));
+    saved_one_head = one_chain_frontier != SIZE_MAX
+                         ? NextWithShift::Make(one_chain_frontier, new_shift)
+                         : NextWithShift::MakeEnd(grow_home, new_shift);
+    saved_one_head.Set<NextWithShift::LockedFlag>(true);
     arr[grow_home].head_next_with_shift.Store(saved_one_head);
 
     // Make sure length_info_ hasn't been updated too early, as we're about
@@ -2535,8 +2515,8 @@ void AutoHyperClockTable::SplitForGrow(size_t grow_home, size_t old_home,
     // Try to set zero's head.
     if (zero_head_lock.CasUpdate(
             zero_chain_frontier != SIZE_MAX
-                ? MakeNextWithShift(zero_chain_frontier, new_shift)
-                : MakeNextWithShiftEnd(old_home, new_shift),
+                ? NextWithShift::Make(zero_chain_frontier, new_shift)
+                : NextWithShift::MakeEnd(old_home, new_shift),
             yield_count_)) {
       // Both heads successfully updated to new shift
       break;
@@ -2570,10 +2550,10 @@ void AutoHyperClockTable::SplitForGrow(size_t grow_home, size_t old_home,
     size_t& other_frontier = chain_frontier_first != 0
                                  ? /*&*/ zero_chain_frontier
                                  : /*&*/ one_chain_frontier;
-    uint64_t stop_before_or_new_tail =
+    NextWithShift stop_before_or_new_tail =
         other_frontier != SIZE_MAX
-            ? /*stop before*/ MakeNextWithShift(other_frontier, old_shift)
-            : /*new tail*/ MakeNextWithShiftEnd(
+            ? /*stop before*/ NextWithShift::Make(other_frontier, old_shift)
+            : /*new tail*/ NextWithShift::MakeEnd(
                   chain_frontier_first == 0 ? old_home : grow_home, new_shift);
     UpgradeShiftsOnRange(arr, first_frontier, stop_before_or_new_tail,
                          old_shift, new_shift);
@@ -2599,20 +2579,19 @@ void AutoHyperClockTable::SplitForGrow(size_t grow_home, size_t old_home,
                                    ? /*&*/ zero_chain_frontier
                                    : /*&*/ one_chain_frontier;
       assert(cur != first_frontier);
-      assert(GetNextFromNextWithShift(
-                 arr[first_frontier].chain_next_with_shift.Load()) ==
+      assert(arr[first_frontier].chain_next_with_shift.Load().GetNext() ==
              other_frontier);
 
-      uint64_t next_with_shift = arr[cur].chain_next_with_shift.Load();
+      NextWithShift next_with_shift = arr[cur].chain_next_with_shift.Load();
 
       // Check for end of original chain
-      if (HandleImpl::IsEnd(next_with_shift)) {
+      if (next_with_shift.IsEnd()) {
         // Can set upgraded tail on first chain
-        uint64_t first_new_tail = MakeNextWithShiftEnd(
+        NextWithShift first_new_tail = NextWithShift::MakeEnd(
             chain_frontier_first == 0 ? old_home : grow_home, new_shift);
         arr[first_frontier].chain_next_with_shift.Store(first_new_tail);
         // And upgrade remainder of other chain
-        uint64_t other_new_tail = MakeNextWithShiftEnd(
+        NextWithShift other_new_tail = NextWithShift::MakeEnd(
             chain_frontier_first != 0 ? old_home : grow_home, new_shift);
         UpgradeShiftsOnRange(arr, other_frontier, other_new_tail, old_shift,
                              new_shift);
@@ -2621,7 +2600,7 @@ void AutoHyperClockTable::SplitForGrow(size_t grow_home, size_t old_home,
       }
 
       // next_with_shift is not End
-      cur = GetNextFromNextWithShift(next_with_shift);
+      cur = next_with_shift.GetNext();
 
       int target_chain;
       if (BottomNBits(arr[cur].hashed_key[1], new_shift) == old_home) {
@@ -2634,7 +2613,7 @@ void AutoHyperClockTable::SplitForGrow(size_t grow_home, size_t old_home,
       }
       if (target_chain == chain_frontier_first) {
         // Found next entry to skip to on the first chain
-        uint64_t skip_to = MakeNextWithShift(cur, new_shift);
+        NextWithShift skip_to = NextWithShift::Make(cur, new_shift);
         arr[first_frontier].chain_next_with_shift.Store(skip_to);
         first_frontier = cur;
         // Upgrade other chain up to entry before that one
@@ -2675,17 +2654,17 @@ void AutoHyperClockTable::PurgeImplLocked(OpData* op_data,
 
   HandleImpl* const arr = array_.Get();
 
-  uint64_t next_with_shift = rewrite_lock.GetSavedHead();
-  assert(!HandleImpl::IsEnd(next_with_shift));
-  int home_shift = GetShiftFromNextWithShift(next_with_shift);
+  NextWithShift next_with_shift = rewrite_lock.GetSavedHead();
+  assert(!next_with_shift.IsEnd());
+  int home_shift = next_with_shift.GetShift();
   (void)home;
   (void)home_shift;
-  size_t next = GetNextFromNextWithShift(next_with_shift);
+  size_t next = next_with_shift.GetNext();
   assert(next < array_.Count());
   HandleImpl* h = &arr[next];
   HandleImpl* prev_to_keep = nullptr;
 #ifndef NDEBUG
-  uint64_t prev_to_keep_next_with_shift = 0;
+  NextWithShift prev_to_keep_next_with_shift{};
 #endif
   // Whether there are entries between h and prev_to_keep that should be
   // purged from the chain.
@@ -2743,13 +2722,13 @@ void AutoHyperClockTable::PurgeImplLocked(OpData* op_data,
         // update any new entries just inserted in parallel.
         // Can simply restart (GetSavedHead() already updated from CAS failure).
         next_with_shift = rewrite_lock.GetSavedHead();
-        assert(!HandleImpl::IsEnd(next_with_shift));
-        next = GetNextFromNextWithShift(next_with_shift);
+        assert(!next_with_shift.IsEnd());
+        next = next_with_shift.GetNext();
         assert(next < array_.Count());
         h = &arr[next];
         pending_purge = false;
         assert(prev_to_keep == nullptr);
-        assert(GetShiftFromNextWithShift(next_with_shift) == home_shift);
+        assert(next_with_shift.GetShift() == home_shift);
         continue;
       }
       pending_purge = false;
@@ -2771,13 +2750,13 @@ void AutoHyperClockTable::PurgeImplLocked(OpData* op_data,
     }
 #endif
 
-    assert(GetShiftFromNextWithShift(next_with_shift) == home_shift);
+    assert(next_with_shift.GetShift() == home_shift);
 
     // Check for end marker
-    if (HandleImpl::IsEnd(next_with_shift)) {
+    if (next_with_shift.IsEnd()) {
       h = nullptr;
     } else {
-      next = GetNextFromNextWithShift(next_with_shift);
+      next = next_with_shift.GetNext();
       assert(next < array_.Count());
       h = &arr[next];
       assert(h != prev_to_keep);
@@ -2849,7 +2828,7 @@ void AutoHyperClockTable::PurgeImpl(OpData* op_data, size_t home,
     // Ensure we are at the correct home for the shift in effect for the
     // chain head.
     for (;;) {
-      int shift = GetShiftFromNextWithShift(rewrite_lock.GetSavedHead());
+      int shift = rewrite_lock.GetSavedHead().GetShift();
 
       if (shift > home_shift) {
         // Found a newer shift at candidate head, which must apply to us.
@@ -3045,14 +3024,14 @@ AutoHyperClockTable::HandleImpl* AutoHyperClockTable::DoInsert(
   }
 
   // Now insert into chain using head pointer
-  uint64_t next_with_shift;
+  NextWithShift next_with_shift;
   int home_shift = orig_home_shift;
 
   // Might need to retry
   for (int i = 0;; ++i) {
     CHECK_TOO_MANY_ITERATIONS(i);
     next_with_shift = arr[home].head_next_with_shift.Load();
-    int shift = GetShiftFromNextWithShift(next_with_shift);
+    int shift = next_with_shift.GetShift();
 
     if (UNLIKELY(shift != home_shift)) {
       // NOTE: shift increases with table growth
@@ -3079,15 +3058,14 @@ AutoHyperClockTable::HandleImpl* AutoHyperClockTable::DoInsert(
     }
 
     // Values to update to
-    uint64_t head_next_with_shift = MakeNextWithShift(idx, home_shift);
-    uint64_t chain_next_with_shift = next_with_shift;
+    NextWithShift head_next_with_shift = NextWithShift::Make(idx, home_shift);
+    NextWithShift chain_next_with_shift = next_with_shift;
 
     // Preserve the locked state in head, without propagating to chain next
     // where it is meaningless (and not allowed)
-    if (UNLIKELY((next_with_shift & HandleImpl::kNextEndFlags) ==
-                 HandleImpl::kHeadLocked)) {
-      head_next_with_shift |= HandleImpl::kHeadLocked;
-      chain_next_with_shift &= ~HandleImpl::kHeadLocked;
+    if (UNLIKELY(next_with_shift.IsLockedNotEnd())) {
+      head_next_with_shift.Set<NextWithShift::LockedFlag>(true);
+      chain_next_with_shift.Set<NextWithShift::LockedFlag>(false);
     }
 
     arr[idx].chain_next_with_shift.Store(chain_next_with_shift);
@@ -3156,9 +3134,9 @@ AutoHyperClockTable::HandleImpl* AutoHyperClockTable::Lookup(
   // of a loop as possible.
 
   HandleImpl* const arr = array_.Get();
-  uint64_t next_with_shift = arr[home].head_next_with_shift.LoadRelaxed();
-  for (size_t i = 0; !HandleImpl::IsEnd(next_with_shift) && i < 10; ++i) {
-    HandleImpl* h = &arr[GetNextFromNextWithShift(next_with_shift)];
+  NextWithShift next_with_shift = arr[home].head_next_with_shift.LoadRelaxed();
+  for (size_t i = 0; !next_with_shift.IsEnd() && i < 10; ++i) {
+    HandleImpl* h = &arr[next_with_shift.IsEnd()];
     // Attempt cheap key match without acquiring a read ref. This could give a
     // false positive, which is re-checked after acquiring read ref, or false
     // negative, which is re-checked in the full Lookup. Also, this is a
@@ -3203,7 +3181,7 @@ AutoHyperClockTable::HandleImpl* AutoHyperClockTable::Lookup(
     // Read head or chain pointer
     next_with_shift = h ? h->chain_next_with_shift.Load()
                         : arr[home].head_next_with_shift.Load();
-    int shift = GetShiftFromNextWithShift(next_with_shift);
+    int shift = next_with_shift.GetShift();
 
     // Make sure it's usable
     size_t effective_home = home;
@@ -3257,10 +3235,10 @@ AutoHyperClockTable::HandleImpl* AutoHyperClockTable::Lookup(
     }
 
     // Check for end marker
-    if (HandleImpl::IsEnd(next_with_shift)) {
+    if (next_with_shift.IsEnd()) {
       // To ensure we didn't miss anything in the chain, the end marker must
       // point back to the correct home.
-      if (LIKELY(GetNextFromNextWithShift(next_with_shift) == effective_home)) {
+      if (LIKELY(next_with_shift.GetNext() == effective_home)) {
         // Complete, clean iteration of the chain, not found.
         // Clean up.
         if (read_ref_on_chain) {
@@ -3276,7 +3254,7 @@ AutoHyperClockTable::HandleImpl* AutoHyperClockTable::Lookup(
     }
 
     // Follow the next and check for full key match, home match, or neither
-    h = &arr[GetNextFromNextWithShift(next_with_shift)];
+    h = &arr[next_with_shift.GetNext()];
     bool full_match_or_unknown = false;
     if (MatchAndRef(&hashed_key, *h, shift, effective_home,
                     &full_match_or_unknown)) {
@@ -3600,8 +3578,7 @@ size_t AutoHyperClockTable::CalcMaxUsableLength(
 
 namespace {
 bool IsHeadNonempty(const AutoHyperClockTable::HandleImpl& h) {
-  return !AutoHyperClockTable::HandleImpl::IsEnd(
-      h.head_next_with_shift.LoadRelaxed());
+  return !h.head_next_with_shift.LoadRelaxed().IsEnd();
 }
 bool IsEntryAtHome(const AutoHyperClockTable::HandleImpl& h, int shift,
                    size_t home) {

cache/clock_cache.h

@@ -775,6 +775,7 @@ class AutoHyperClockTable : public BaseClockTable {
     // chain--specifically the next entry in the chain.
     // * The end of a chain is given a special "end" marker and refers back
     // to the head of the chain.
+    // These decorated pointers use the NextWithShift bit field struct below.
     //
     // Why do we need shift on each pointer? To make Lookup wait-free, we need
     // to be able to query a chain without missing anything, and preferably
@@ -794,47 +795,63 @@ class AutoHyperClockTable : public BaseClockTable {
     // it is normal to see "under construction" entries on the chain, and it
     // is not safe to read their hashed key without either a read reference
     // on the entry or a rewrite lock on the chain.
+    struct NextWithShift : public BitFields<uint64_t, NextWithShift> {
+      // The "shift" associated with this decorated pointer (see description
+      // above).
+      using Shift = UnsignedBitField<NextWithShift, 6, NoPrevBitField>;
+      // Marker for the end of a chain. Must also (a) point back to the head of
+      // the chain (with end marker removed), and (b) set the LockedFlag
+      // (below), so that attempting to lock an empty chain has no effect (not
+      // needed, as the lock is only needed for removals).
+      using EndFlag = BoolBitField<NextWithShift, Shift>;
+      // Marker that some thread owning writes to the chain structure (except
+      // for inserts), but only if not an "end" pointer. Also called the
+      // "rewrite lock."
+      using LockedFlag = BoolBitField<NextWithShift, EndFlag>;
+      // The "next" associated with this decorated pointer, which is an index
+      // into the table's array_ (see description above).
+      using Next = UnsignedBitField<NextWithShift, 56, LockedFlag>;
 
-    // Marker in a "with_shift" head pointer for some thread owning writes
-    // to the chain structure (except for inserts), but only if not an
-    // "end" pointer. Also called the "rewrite lock."
-    static constexpr uint64_t kHeadLocked = uint64_t{1} << 7;
+      bool IsLocked() const { return Get<LockedFlag>(); }
+      bool IsEnd() const {
+        // End flag should imply locked flag
+        assert(!Get<EndFlag>() || Get<LockedFlag>());
+        return Get<EndFlag>();
+      }
+      bool IsLockedNotEnd() const {
+        // NOTE: helping GCC to optimize this simpler code:
+        // return IsLocked() && !IsEnd();
+        constexpr U kEndFlag = U{1} << EndFlag::kBitOffset;
+        constexpr U kLockedFlag = U{1} << LockedFlag::kBitOffset;
+        return (underlying & (kEndFlag | kLockedFlag)) == kLockedFlag;
+      }
+      auto GetNext() const { return Get<Next>(); }
+      auto GetShift() const { return Get<Shift>(); }
 
-    // Marker in a "with_shift" pointer for the end of a chain. Must also
-    // point back to the head of the chain (with end marker removed).
-    // Also includes the "locked" bit so that attempting to lock an empty
-    // chain has no effect (not needed, as the lock is only needed for
-    // removals).
-    static constexpr uint64_t kNextEndFlags = (uint64_t{1} << 6) | kHeadLocked;
+      static NextWithShift Make(size_t next, int shift) {
+        return NextWithShift{}.With<Next>(next).With<Shift>(
+            static_cast<uint8_t>(shift));
+      }
 
-    static inline bool IsEnd(uint64_t next_with_shift) {
-      // Assuming certain values never used, suffices to check this one bit
-      constexpr auto kCheckBit = kNextEndFlags ^ kHeadLocked;
-      return next_with_shift & kCheckBit;
-    }
-
-    // Bottom bits to right shift away to get an array index from a
-    // "with_shift" pointer.
-    static constexpr int kNextShift = 8;
-
-    // A bit mask for the "shift" associated with each "with_shift" pointer.
-    // Always bottommost bits.
-    static constexpr int kShiftMask = 63;
+      static NextWithShift MakeEnd(size_t next, int shift) {
+        return Make(next, shift).With<EndFlag>(true).With<LockedFlag>(true);
+      }
+    };
 
     // A marker for head_next_with_shift that indicates this HandleImpl is
     // heap allocated (standalone) rather than in the table.
-    static constexpr uint64_t kStandaloneMarker = UINT64_MAX;
+    static constexpr NextWithShift kStandaloneMarker{UINT64_MAX};
 
     // A marker for head_next_with_shift indicating the head is not yet part
     // of the usable table, or for chain_next_with_shift indicating that the
     // entry is not present or is not yet part of a chain (must not be
     // "shareable" state).
-    static constexpr uint64_t kUnusedMarker = 0;
+    static constexpr NextWithShift kUnusedMarker{0};
 
     // See above. The head pointer is logically independent of the rest of
     // the entry, including the chain next pointer.
-    AcqRelAtomic<uint64_t> head_next_with_shift{kUnusedMarker};
-    AcqRelAtomic<uint64_t> chain_next_with_shift{kUnusedMarker};
+    AcqRelBitFieldsAtomic<NextWithShift> head_next_with_shift{kUnusedMarker};
+    AcqRelBitFieldsAtomic<NextWithShift> chain_next_with_shift{kUnusedMarker};
 
     // For supporting CreateStandalone and some fallback cases.
     inline bool IsStandalone() const {

db/c.cc

@@ -71,6 +71,11 @@ using ROCKSDB_NAMESPACE::CompactionFilterFactory;
 using ROCKSDB_NAMESPACE::CompactionJobInfo;
 using ROCKSDB_NAMESPACE::CompactionOptionsFIFO;
 using ROCKSDB_NAMESPACE::CompactionReason;
+using ROCKSDB_NAMESPACE::CompactionService;
+using ROCKSDB_NAMESPACE::CompactionServiceJobInfo;
+using ROCKSDB_NAMESPACE::CompactionServiceJobStatus;
+using ROCKSDB_NAMESPACE::CompactionServiceOptionsOverride;
+using ROCKSDB_NAMESPACE::CompactionServiceScheduleResponse;
 using ROCKSDB_NAMESPACE::CompactRangeOptions;
 using ROCKSDB_NAMESPACE::Comparator;
 using ROCKSDB_NAMESPACE::CompressionType;
@@ -107,6 +112,7 @@ using ROCKSDB_NAMESPACE::NewCompactOnDeletionCollectorFactory;
 using ROCKSDB_NAMESPACE::NewGenericRateLimiter;
 using ROCKSDB_NAMESPACE::NewLRUCache;
 using ROCKSDB_NAMESPACE::NewRibbonFilterPolicy;
+using ROCKSDB_NAMESPACE::OpenAndCompactOptions;
 using ROCKSDB_NAMESPACE::OptimisticTransactionDB;
 using ROCKSDB_NAMESPACE::OptimisticTransactionOptions;
 using ROCKSDB_NAMESPACE::Options;
@@ -349,6 +355,22 @@ struct rocksdb_statistics_histogram_data_t {
   HistogramData rep;
 };
 
+struct rocksdb_compactionservice_scheduleresponse_t {
+  CompactionServiceScheduleResponse rep;
+};
+
+struct rocksdb_compactionservice_jobinfo_t {
+  CompactionServiceJobInfo rep;
+};
+
+struct rocksdb_compaction_service_options_override_t {
+  CompactionServiceOptionsOverride rep;
+};
+
+struct rocksdb_open_and_compact_options_t {
+  OpenAndCompactOptions rep;
+};
+
 struct rocksdb_compactionfilter_t : public CompactionFilter {
   void* state_;
   void (*destructor_)(void*);
@@ -643,6 +665,410 @@ static inline char* CopyString(const Slice& slice) {
   return result;
 }
 
+const char* rocksdb_compactionservice_jobinfo_t_get_db_name(
+    const rocksdb_compactionservice_jobinfo_t* info, size_t* len) {
+  *len = info->rep.db_name.size();
+  return info->rep.db_name.data();
+}
+
+const char* rocksdb_compactionservice_jobinfo_t_get_db_id(
+    const rocksdb_compactionservice_jobinfo_t* info, size_t* len) {
+  *len = info->rep.db_id.size();
+  return info->rep.db_id.data();
+}
+
+const char* rocksdb_compactionservice_jobinfo_t_get_db_session_id(
+    const rocksdb_compactionservice_jobinfo_t* info, size_t* len) {
+  *len = info->rep.db_session_id.size();
+  return info->rep.db_session_id.data();
+}
+
+const char* rocksdb_compactionservice_jobinfo_t_get_cf_name(
+    const rocksdb_compactionservice_jobinfo_t* info, size_t* len) {
+  *len = info->rep.cf_name.size();
+  return info->rep.cf_name.data();
+}
+
+uint32_t rocksdb_compactionservice_jobinfo_t_get_cf_id(
+    const rocksdb_compactionservice_jobinfo_t* info) {
+  return info->rep.cf_id;
+}
+
+uint64_t rocksdb_compactionservice_jobinfo_t_get_job_id(
+    const rocksdb_compactionservice_jobinfo_t* info) {
+  return info->rep.job_id;
+}
+
+int rocksdb_compactionservice_jobinfo_t_get_priority(
+    const rocksdb_compactionservice_jobinfo_t* info) {
+  return static_cast<int>(info->rep.priority);
+}
+
+int rocksdb_compactionservice_jobinfo_t_get_compaction_reason(
+    const rocksdb_compactionservice_jobinfo_t* info) {
+  return static_cast<int>(info->rep.compaction_reason);
+}
+
+int rocksdb_compactionservice_jobinfo_t_get_base_input_level(
+    const rocksdb_compactionservice_jobinfo_t* info) {
+  return info->rep.base_input_level;
+}
+
+int rocksdb_compactionservice_jobinfo_t_get_output_level(
+    const rocksdb_compactionservice_jobinfo_t* info) {
+  return info->rep.output_level;
+}
+
+unsigned char rocksdb_compactionservice_jobinfo_t_is_full_compaction(
+    const rocksdb_compactionservice_jobinfo_t* info) {
+  return info->rep.is_full_compaction;
+}
+
+unsigned char rocksdb_compactionservice_jobinfo_t_is_manual_compaction(
+    const rocksdb_compactionservice_jobinfo_t* info) {
+  return info->rep.is_manual_compaction;
+}
+
+unsigned char rocksdb_compactionservice_jobinfo_t_is_bottommost_level(
+    const rocksdb_compactionservice_jobinfo_t* info) {
+  return info->rep.bottommost_level;
+}
+
+// Helper function to validate compaction service job status
+static inline bool IsValidCompactionServiceJobStatus(int status) {
+  return status >= rocksdb_compactionservice_jobstatus_success &&
+         status <= rocksdb_compactionservice_jobstatus_use_local;
+}
+
+rocksdb_compactionservice_scheduleresponse_t*
+rocksdb_compactionservice_scheduleresponse_create(const char* scheduled_job_id,
+                                                  int status, char** errptr) {
+  // Validate status is in range [success=0, failure=1, aborted=2, use_local=3]
+  if (!IsValidCompactionServiceJobStatus(status)) {
+    SaveError(errptr,
+              Status::InvalidArgument("Invalid status value. Must be 0-3."));
+    return nullptr;
+  }
+
+  rocksdb_compactionservice_scheduleresponse_t* response =
+      new rocksdb_compactionservice_scheduleresponse_t{
+          CompactionServiceScheduleResponse(
+              scheduled_job_id ? std::string(scheduled_job_id) : "",
+              static_cast<CompactionServiceJobStatus>(status))};
+  return response;
+}
+
+rocksdb_compactionservice_scheduleresponse_t*
+rocksdb_compactionservice_scheduleresponse_create_with_status(int status,
+                                                              char** errptr) {
+  // Validate status is in range [success=0, failure=1, aborted=2, use_local=3]
+  if (!IsValidCompactionServiceJobStatus(status)) {
+    SaveError(errptr,
+              Status::InvalidArgument("Invalid status value. Must be 0-3."));
+    return nullptr;
+  }
+
+  rocksdb_compactionservice_scheduleresponse_t* response =
+      new rocksdb_compactionservice_scheduleresponse_t{
+          CompactionServiceScheduleResponse(
+              static_cast<CompactionServiceJobStatus>(status))};
+  return response;
+}
+
+void rocksdb_compactionservice_scheduleresponse_t_destroy(
+    rocksdb_compactionservice_scheduleresponse_t* response) {
+  if (response) {
+    delete response;
+  }
+}
+
+int rocksdb_compactionservice_scheduleresponse_getstatus(
+    const rocksdb_compactionservice_scheduleresponse_t* response) {
+  if (!response) {
+    return rocksdb_compactionservice_jobstatus_failure;
+  }
+  return static_cast<int>(response->rep.status);
+}
+
+const char* rocksdb_compactionservice_scheduleresponse_get_scheduled_job_id(
+    const rocksdb_compactionservice_scheduleresponse_t* response, size_t* len) {
+  if (!response || !len) {
+    if (len) {
+      *len = 0;
+    }
+    return "";
+  }
+  *len = response->rep.scheduled_job_id.size();
+  return response->rep.scheduled_job_id.data();
+}
+
+struct rocksdb_compactionservice_t : public CompactionService {
+  void* state_;
+  void (*destructor_)(void*);
+  rocksdb_compaction_service_schedule_cb schedule_;
+  std::string name_;
+  rocksdb_compaction_service_wait_cb wait_;
+  rocksdb_compaction_service_cancel_awaiting_jobs_cb cancel_awaiting_jobs_;
+  rocksdb_compaction_service_on_installation_cb on_installation_;
+
+  rocksdb_compactionservice_t(
+      void* state, void (*destructor)(void*),
+      rocksdb_compaction_service_schedule_cb
+          rocksdb_compaction_service_schedule_ptr,
+      const char* name, rocksdb_compaction_service_wait_cb wait,
+      rocksdb_compaction_service_cancel_awaiting_jobs_cb cancel_awaiting_jobs,
+      rocksdb_compaction_service_on_installation_cb on_installation)
+      : state_(state),
+        destructor_(destructor),
+        schedule_(rocksdb_compaction_service_schedule_ptr),
+        name_(name ? name : "CompactionService"),
+        wait_(wait),
+        cancel_awaiting_jobs_(cancel_awaiting_jobs),
+        on_installation_(on_installation) {}
+
+  ~rocksdb_compactionservice_t() override {
+    if (destructor_) {
+      (*destructor_)(state_);
+    }
+  }
+
+  const char* Name() const override { return name_.c_str(); }
+
+  CompactionServiceScheduleResponse Schedule(
+      const CompactionServiceJobInfo& info,
+      const std::string& compaction_service_input) override {
+    if (schedule_ == nullptr) {
+      return CompactionServiceScheduleResponse(
+          CompactionServiceJobStatus::kUseLocal);
+    }
+
+    rocksdb_compactionservice_scheduleresponse_t* c_response = (*schedule_)(
+        state_,
+        reinterpret_cast<const rocksdb_compactionservice_jobinfo_t*>(&info),
+        compaction_service_input.data(), compaction_service_input.size());
+
+    if (c_response == nullptr) {
+      return CompactionServiceScheduleResponse(
+          CompactionServiceJobStatus::kFailure);
+    }
+
+    CompactionServiceScheduleResponse response = std::move(c_response->rep);
+    delete c_response;
+    return response;
+  }
+
+  CompactionServiceJobStatus Wait(const std::string& scheduled_job_id,
+                                  std::string* result) override {
+    if (wait_ == nullptr) {
+      return CompactionServiceJobStatus::kUseLocal;
+    }
+
+    char* c_result = nullptr;
+    size_t result_len = 0;
+
+    int status =
+        (*wait_)(state_, scheduled_job_id.c_str(), &c_result, &result_len);
+
+    if (c_result != nullptr) {
+      if (result != nullptr) {
+        result->assign(c_result, result_len);
+      }
+      free(c_result);
+    }
+
+    return static_cast<CompactionServiceJobStatus>(status);
+  }
+
+  void CancelAwaitingJobs() override {
+    if (cancel_awaiting_jobs_ != nullptr) {
+      (*cancel_awaiting_jobs_)(state_);
+    }
+  }
+
+  void OnInstallation(const std::string& scheduled_job_id,
+                      CompactionServiceJobStatus status) override {
+    if (on_installation_ != nullptr) {
+      (*on_installation_)(state_, scheduled_job_id.c_str(),
+                          static_cast<int>(status));
+    }
+  }
+};
+
+rocksdb_compactionservice_t* rocksdb_compactionservice_create(
+    void* state, void (*destructor)(void*),
+    rocksdb_compaction_service_schedule_cb schedule, const char* name,
+    rocksdb_compaction_service_wait_cb wait,
+    rocksdb_compaction_service_cancel_awaiting_jobs_cb cancel_awaiting_jobs,
+    rocksdb_compaction_service_on_installation_cb on_installation) {
+  return new rocksdb_compactionservice_t(state, destructor, schedule, name,
+                                         wait, cancel_awaiting_jobs,
+                                         on_installation);
+}
+
+void rocksdb_options_set_compaction_service(
+    rocksdb_options_t* opt, rocksdb_compactionservice_t* service) {
+  if (!opt || !service) {
+    return;
+  }
+
+  opt->rep.compaction_service = std::shared_ptr<CompactionService>(service);
+}
+
+// CompactionServiceOptionsOverride functions
+rocksdb_compaction_service_options_override_t*
+rocksdb_compaction_service_options_override_create() {
+  return new rocksdb_compaction_service_options_override_t;
+}
+
+void rocksdb_compaction_service_options_override_destroy(
+    rocksdb_compaction_service_options_override_t* override_options) {
+  if (override_options) {
+    delete override_options;
+  }
+}
+
+void rocksdb_compaction_service_options_override_set_env(
+    rocksdb_compaction_service_options_override_t* override_options,
+    rocksdb_env_t* env) {
+  if (override_options && env) {
+    override_options->rep.env = env->rep;
+  }
+}
+
+void rocksdb_compaction_service_options_override_set_comparator(
+    rocksdb_compaction_service_options_override_t* override_options,
+    rocksdb_comparator_t* comparator) {
+  if (override_options && comparator) {
+    override_options->rep.comparator = comparator;
+  }
+}
+
+// Atomic bool management for cancellation
+unsigned char* rocksdb_open_and_compact_canceled_create() {
+  return reinterpret_cast<unsigned char*>(new std::atomic<bool>(false));
+}
+
+void rocksdb_open_and_compact_canceled_destroy(unsigned char* canceled) {
+  if (canceled) {
+    delete reinterpret_cast<std::atomic<bool>*>(canceled);
+  }
+}
+
+void rocksdb_open_and_compact_canceled_set(unsigned char* canceled,
+                                           unsigned char value) {
+  if (canceled) {
+    reinterpret_cast<std::atomic<bool>*>(canceled)->store(value != 0);
+  }
+}
+
+// OpenAndCompactOptions functions
+rocksdb_open_and_compact_options_t* rocksdb_open_and_compact_options_create() {
+  return new rocksdb_open_and_compact_options_t;
+}
+
+void rocksdb_open_and_compact_options_destroy(
+    rocksdb_open_and_compact_options_t* options) {
+  if (options) {
+    delete options;
+  }
+}
+
+void rocksdb_open_and_compact_options_set_canceled(
+    rocksdb_open_and_compact_options_t* options, unsigned char* canceled) {
+  if (options && canceled) {
+    options->rep.canceled = reinterpret_cast<std::atomic<bool>*>(canceled);
+  }
+}
+
+void rocksdb_open_and_compact_options_set_allow_resumption(
+    rocksdb_open_and_compact_options_t* options,
+    unsigned char allow_resumption) {
+  if (options) {
+    options->rep.allow_resumption = allow_resumption != 0;
+  }
+}
+
+// OpenAndCompact functions
+char* rocksdb_open_and_compact(
+    const char* db_path, const char* output_directory, const char* input,
+    size_t input_len, size_t* output_len,
+    const rocksdb_compaction_service_options_override_t* override_options,
+    char** errptr) {
+  if (!db_path || !output_directory || !input || !override_options) {
+    SaveError(errptr, Status::InvalidArgument("Invalid arguments"));
+    return nullptr;
+  }
+
+  std::string input_str(input, input_len);
+  std::string output_str;
+
+  Status s = DB::OpenAndCompact(db_path, output_directory, input_str,
+                                &output_str, override_options->rep);
+
+  if (!s.ok()) {
+    SaveError(errptr, s);
+    return nullptr;
+  }
+
+  // Allocate +1 for null terminator
+  char* result = static_cast<char*>(malloc(output_str.size() + 1));
+  if (!result) {
+    SaveError(errptr, Status::MemoryLimit("Failed to allocate output buffer"));
+    return nullptr;
+  }
+
+  memcpy(result, output_str.data(), output_str.size());
+  result[output_str.size()] = '\0';
+
+  // Only set output_len after successful allocation
+  if (output_len) {
+    *output_len = output_str.size();
+  }
+
+  return result;
+}
+
+char* rocksdb_open_and_compact_with_options(
+    const rocksdb_open_and_compact_options_t* options, const char* db_path,
+    const char* output_directory, const char* input, size_t input_len,
+    size_t* output_len,
+    const rocksdb_compaction_service_options_override_t* override_options,
+    char** errptr) {
+  if (!options || !db_path || !output_directory || !input ||
+      !override_options) {
+    SaveError(errptr, Status::InvalidArgument("Invalid arguments"));
+    return nullptr;
+  }
+
+  std::string input_str(input, input_len);
+  std::string output_str;
+
+  Status s = DB::OpenAndCompact(options->rep, db_path, output_directory,
+                                input_str, &output_str, override_options->rep);
+
+  if (!s.ok()) {
+    SaveError(errptr, s);
+    return nullptr;
+  }
+
+  // Allocate +1 for null terminator
+  char* result = static_cast<char*>(malloc(output_str.size() + 1));
+  if (!result) {
+    SaveError(errptr, Status::MemoryLimit("Failed to allocate output buffer"));
+    return nullptr;
+  }
+
+  memcpy(result, output_str.data(), output_str.size());
+  result[output_str.size()] = '\0';  // Null terminate
+
+  // Only set output_len after successful allocation
+  if (output_len) {
+    *output_len = output_str.size();
+  }
+
+  return result;
+}
+
 rocksdb_t* rocksdb_open(const rocksdb_options_t* options, const char* name,
                         char** errptr) {
   DB* db;
@@ -7841,7 +8267,7 @@ rocksdb_memory_usage_t* rocksdb_approximate_memory_usage_create(
     dbs.push_back(db->rep);
   }
 
-  unordered_set<const Cache*> cache_set;
+  std::unordered_set<const Cache*> cache_set;
   for (auto cache : consumers->caches) {
     cache_set.insert(const_cast<const Cache*>(cache->rep.get()));
   }

db/c_test.c

@@ -722,6 +722,88 @@ static void LoadAndCheckLatestOptions(const char* db_name, rocksdb_env_t* env,
                                       num_column_families);
 }
 
+// Global state for tracking remote compaction calls
+typedef struct {
+  int schedule_called;
+  int wait_called;
+  int cancel_called;
+  char last_scheduled_job_id[256];
+  char last_db_name[256];
+} RemoteCompactionState;
+
+// Schedule callback - gets called when compaction is scheduled
+static rocksdb_compactionservice_scheduleresponse_t* RemoteCompactionSchedule(
+    void* state, const rocksdb_compactionservice_jobinfo_t* info,
+    const char* input, size_t input_len) {
+  (void)input;
+  (void)input_len;
+  RemoteCompactionState* rcs = (RemoteCompactionState*)state;
+  rcs->schedule_called++;
+
+  // Extract job info
+  size_t db_name_len;
+  const char* db_name =
+      rocksdb_compactionservice_jobinfo_t_get_db_name(info, &db_name_len);
+  memcpy(rcs->last_db_name, db_name, db_name_len);
+  rcs->last_db_name[db_name_len] = '\0';
+
+  // Generate a job ID
+  snprintf(rcs->last_scheduled_job_id, sizeof(rcs->last_scheduled_job_id),
+           "job-%d", rcs->schedule_called);
+
+  // Create response with success status
+  char* err = NULL;
+  rocksdb_compactionservice_scheduleresponse_t* response =
+      rocksdb_compactionservice_scheduleresponse_create(
+          rcs->last_scheduled_job_id,
+          rocksdb_compactionservice_jobstatus_success, &err);
+  if (err) {
+    free(err);
+  }
+  return response;
+}
+
+// Wait callback - simulates waiting for remote compaction to complete
+static int RemoteCompactionWait(void* state, const char* scheduled_job_id,
+                                char** result, size_t* result_len) {
+  RemoteCompactionState* rcs = (RemoteCompactionState*)state;
+  rcs->wait_called++;
+
+  if (strcmp(scheduled_job_id, rcs->last_scheduled_job_id) != 0) {
+    return rocksdb_compactionservice_jobstatus_failure;
+  }
+
+  // For testing purposes, return kUseLocal to cause RocksDB to fall back to
+  // local compaction. This tests the callback mechanism without needing a fully
+  // serialized result. In a real scenario, this would communicate with a remote
+  // worker that calls rocksdb_open_and_compact() and returns a properly
+  // serialized CompactionServiceResult
+  *result = NULL;
+  *result_len = 0;
+
+  return rocksdb_compactionservice_jobstatus_use_local;
+}
+
+// Cancel callback - cancels pending jobs
+static void RemoteCompactionCancel(void* state) {
+  RemoteCompactionState* rcs = (RemoteCompactionState*)state;
+  rcs->cancel_called++;
+}
+
+// Destructor callback
+static void RemoteCompactionDestroy(void* state) { (void)state; }
+
+// NULL schedule callback for testing failure handling
+static rocksdb_compactionservice_scheduleresponse_t* NullSchedule(
+    void* state, const rocksdb_compactionservice_jobinfo_t* info,
+    const char* input, size_t input_len) {
+  (void)state;
+  (void)info;
+  (void)input;
+  (void)input_len;
+  return NULL;  // Return NULL to simulate failure
+}
+
 int main(int argc, char** argv) {
   (void)argc;
   (void)argv;
@@ -4483,6 +4565,219 @@ int main(int argc, char** argv) {
     rocksdb_cache_destroy(lru);
   }
 
+  StartPhase("remote_compaction_service");
+  {
+    RemoteCompactionState remote_state = {0, 0, 0, "", ""};
+
+    // Create compaction service
+    rocksdb_compactionservice_t* service = rocksdb_compactionservice_create(
+        &remote_state,             // state
+        RemoteCompactionDestroy,   // destructor
+        RemoteCompactionSchedule,  // schedule callback
+        "TestRemoteCompaction",    // name
+        RemoteCompactionWait,      // wait callback
+        RemoteCompactionCancel,    // cancel_awaiting_jobs
+        NULL);                     // on_installation
+
+    // Create options with remote compaction
+    rocksdb_options_t* remote_options = rocksdb_options_create();
+    rocksdb_options_set_create_if_missing(remote_options, 1);
+    rocksdb_options_set_level0_file_num_compaction_trigger(remote_options, 2);
+    rocksdb_options_set_write_buffer_size(remote_options,
+                                          64 * 1024);  // 64KB buffer
+    rocksdb_options_set_max_bytes_for_level_base(remote_options,
+                                                 256 * 1024);  // 256KB
+    rocksdb_options_set_target_file_size_base(
+        remote_options, 64 * 1024);  // 64KB target file size
+    // Disable automatic compactions to test manual compaction only
+    rocksdb_options_set_disable_auto_compactions(remote_options, 1);
+    rocksdb_options_set_compaction_service(remote_options, service);
+
+    // Destroy old DB and create new one
+    rocksdb_close(db);
+    rocksdb_destroy_db(remote_options, dbname, &err);
+    CheckNoError(err);
+
+    db = rocksdb_open(remote_options, dbname, &err);
+    CheckNoError(err);
+
+    // Create multiple SST files to trigger compaction
+    rocksdb_flushoptions_t* flush_opts = rocksdb_flushoptions_create();
+    rocksdb_flushoptions_set_wait(flush_opts, 1);
+
+    // Write and flush multiple times to create multiple L0 files
+    // Write more data with larger values to ensure files are substantial
+    for (int batch = 0; batch < 5; batch++) {
+      for (int i = 0; i < 200; i++) {
+        char key[20], val[1000];
+        snprintf(key, sizeof(key), "key%d_%d", batch, i);
+        // Fill value with repeated data to make it larger
+        memset(val, 'a' + (batch % 26), sizeof(val) - 1);
+        val[sizeof(val) - 1] = '\0';
+        rocksdb_put(db, woptions, key, strlen(key), val, strlen(val), &err);
+        CheckNoError(err);
+      }
+      rocksdb_flush(db, flush_opts, &err);
+      CheckNoError(err);
+    }
+    rocksdb_flushoptions_destroy(flush_opts);
+
+    // Trigger manual compaction to invoke remote compaction service
+    rocksdb_compact_range(db, NULL, 0, NULL, 0);
+
+    rocksdb_wait_for_compact_options_t* wco =
+        rocksdb_wait_for_compact_options_create();
+    rocksdb_wait_for_compact(db, wco, &err);
+    CheckNoError(err);
+    rocksdb_wait_for_compact_options_destroy(wco);
+
+    // Verify that callbacks were actually called
+    CheckCondition(remote_state.schedule_called > 0);
+    CheckCondition(remote_state.wait_called > 0);
+    CheckCondition(strlen(remote_state.last_db_name) > 0);
+    CheckCondition(strstr(remote_state.last_db_name, "rocksdb_c_test") != NULL);
+
+    // Verify data is still accessible after remote compaction
+    // Just check a few keys to verify data integrity
+    for (int batch = 0; batch < 5; batch++) {
+      char key[20];
+      snprintf(key, sizeof(key), "key%d_0", batch);
+      size_t vallen;
+      char* val = rocksdb_get(db, roptions, key, strlen(key), &vallen, &err);
+      CheckNoError(err);
+      CheckCondition(val != NULL);
+      CheckCondition(vallen == 999);  // strlen of 1000-byte string
+      free(val);
+    }
+
+    // Test cancellation API directly
+    RemoteCompactionCancel(&remote_state);
+    CheckCondition(remote_state.cancel_called > 0);
+
+    // Cleanup
+    rocksdb_close(db);
+    rocksdb_destroy_db(remote_options, dbname, &err);
+    CheckNoError(err);
+    rocksdb_options_destroy(remote_options);
+
+    // Reopen DB with original options for subsequent tests
+    db = rocksdb_open(options, dbname, &err);
+    CheckNoError(err);
+  }
+
+  StartPhase("remote_compaction_scheduleresponse");
+  {
+    // Test scheduleresponse creation and getters
+    rocksdb_compactionservice_scheduleresponse_t* response;
+
+    // Test success response
+    err = NULL;
+    response = rocksdb_compactionservice_scheduleresponse_create(
+        "test-job-123", rocksdb_compactionservice_jobstatus_success, &err);
+    CheckNoError(err);
+    CheckCondition(response != NULL);
+    CheckCondition(
+        rocksdb_compactionservice_scheduleresponse_getstatus(response) ==
+        rocksdb_compactionservice_jobstatus_success);
+
+    size_t job_id_len;
+    const char* job_id =
+        rocksdb_compactionservice_scheduleresponse_get_scheduled_job_id(
+            response, &job_id_len);
+    CheckCondition(job_id_len == strlen("test-job-123"));
+    CheckCondition(memcmp(job_id, "test-job-123", job_id_len) == 0);
+    rocksdb_compactionservice_scheduleresponse_t_destroy(response);
+
+    // Test failure response
+    response = rocksdb_compactionservice_scheduleresponse_create_with_status(
+        rocksdb_compactionservice_jobstatus_failure, &err);
+    CheckCondition(response != NULL);
+    CheckCondition(
+        rocksdb_compactionservice_scheduleresponse_getstatus(response) ==
+        rocksdb_compactionservice_jobstatus_failure);
+    rocksdb_compactionservice_scheduleresponse_t_destroy(response);
+
+    response = rocksdb_compactionservice_scheduleresponse_create_with_status(
+        999, &err);
+    CheckCondition(response == NULL);  // Invalid status
+    if (err) {
+      Free(&err);
+    }
+  }
+
+  StartPhase("remote_compaction_options_override");
+  {
+    // Test CompactionServiceOptionsOverride API
+    rocksdb_compaction_service_options_override_t* override_opts =
+        rocksdb_compaction_service_options_override_create();
+    CheckCondition(override_opts != NULL);
+
+    // Set up override options
+    rocksdb_compaction_service_options_override_set_env(override_opts, env);
+    rocksdb_compaction_service_options_override_set_comparator(override_opts,
+                                                               cmp);
+
+    rocksdb_compaction_service_options_override_destroy(override_opts);
+  }
+
+  StartPhase("remote_compaction_null_callback_handling");
+  {
+    // Test that NULL callback returns are handled gracefully
+    // This simulates a failure in the remote compaction service
+    rocksdb_compactionservice_t* null_service =
+        rocksdb_compactionservice_create(NULL, NULL, NullSchedule,
+                                         "NullTestService", NULL, NULL, NULL);
+
+    rocksdb_options_t* null_opts = rocksdb_options_create();
+    rocksdb_options_set_create_if_missing(null_opts, 1);
+    rocksdb_options_set_compaction_service(null_opts, null_service);
+
+    const char* null_db = "rocksdb_c_test_null_service";
+
+    rocksdb_t* null_db_handle = rocksdb_open(null_opts, null_db, &err);
+    CheckNoError(err);
+
+    // Write data and trigger compaction
+    for (int i = 0; i < 100; i++) {
+      char key[20], val[50];
+      snprintf(key, sizeof(key), "key%d", i);
+      snprintf(val, sizeof(val), "val%d", i);
+      rocksdb_put(null_db_handle, woptions, key, strlen(key), val, strlen(val),
+                  &err);
+      CheckNoError(err);
+    }
+
+    // This should fall back to local compaction (not crash)
+    rocksdb_compact_range(null_db_handle, NULL, 0, NULL, 0);
+
+    // Data should still be readable
+    CheckGet(null_db_handle, roptions, "key50", "val50");
+
+    rocksdb_close(null_db_handle);
+    rocksdb_destroy_db(null_opts, null_db, &err);
+    rocksdb_options_destroy(null_opts);
+  }
+
+  StartPhase("remote_compaction_canceled_flag");
+  {
+    // Test atomic cancellation flag API
+    unsigned char* canceled = rocksdb_open_and_compact_canceled_create();
+    CheckCondition(canceled != NULL);
+
+    // Set cancellation
+    rocksdb_open_and_compact_canceled_set(canceled, 1);
+
+    // Use with OpenAndCompactOptions
+    rocksdb_open_and_compact_options_t* oac_opts =
+        rocksdb_open_and_compact_options_create();
+    rocksdb_open_and_compact_options_set_canceled(oac_opts, canceled);
+    rocksdb_open_and_compact_options_set_allow_resumption(oac_opts, 1);
+
+    // Cleanup
+    rocksdb_open_and_compact_options_destroy(oac_opts);
+    rocksdb_open_and_compact_canceled_destroy(canceled);
+  }
+
   StartPhase("sst_file_manager");
   {
     rocksdb_sst_file_manager_t* sst_file_manager;

db/db_basic_test.cc

@@ -3868,6 +3868,75 @@ TEST_F(DBBasicTest, SkipWALIfMissingTableFiles) {
   ASSERT_OK(iter->status());
 }
 
+TEST_F(DBBasicTest, BestEffortRecoveryFailureWithTableCacheUseAfterFree) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.env = env_;
+  // Force multiple manifest files
+  options.max_manifest_file_size = 1;
+  options.max_manifest_space_amp_pct = 0;
+
+  DestroyAndReopen(options);
+
+  // Disable file deletions to preserve old manifest files for
+  // best-efforts recovery to succeed
+  ASSERT_OK(db_->DisableFileDeletions());
+
+  // Create multiple SST files to populate TableCache during
+  // best-efforts recovery
+  for (int i = 0; i < 10; i++) {
+    ASSERT_OK(Put("key" + std::to_string(i),
+                  std::string(1000, static_cast<char>('a' + i))));
+    ASSERT_OK(Flush());
+  }
+
+  // Verify we have multiple manifest files
+  std::vector<std::string> files;
+  ASSERT_OK(env_->GetChildren(dbname_, &files));
+  int manifest_count = 0;
+  for (const auto& file : files) {
+    if (file.find("MANIFEST") != std::string::npos) {
+      manifest_count++;
+    }
+  }
+  ASSERT_GE(manifest_count, 2);
+
+  // Inject corruption after TableCache is populated (count > 3), but only once
+  // (injected flag) to allow best-effort recovery to trigger retry and succeed.
+  // This coerce the bug: first recovery caches SSTs with reference to column
+  // family's options in table cache and retry deletes column family so the
+  // reference becomes dangling.
+  int count = 0;
+  bool injected = false;
+  SyncPoint::GetInstance()->SetCallBack(
+      "VersionBuilder::CheckConsistencyBeforeReturn", [&](void* arg) {
+        count++;
+        if (count > 3 && !injected) {
+          ASSERT_NE(nullptr, arg);
+          *(static_cast<Status*>(arg)) =
+              Status::Corruption("Injected corruption");
+          injected = true;
+        }
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  options.best_efforts_recovery = true;
+
+  Status s = TryReopen(options);
+  ASSERT_OK(s);
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+
+  for (int i = 0; i < 10; i++) {
+    std::string value;
+    // Without the fix, ASAN detects use-after-free when accessing cached SST
+    // files that hold dangling references to deleted ioptions.
+    s = db_->Get(ReadOptions(), "key" + std::to_string(i), &value);
+    ASSERT_TRUE(s.ok() || s.IsNotFound());
+  }
+}
+
 TEST_F(DBBasicTest, DisableTrackWal) {
   // If WAL tracking was enabled, and then disabled during reopen,
   // the previously tracked WALs should be removed from MANIFEST.

db/version_set.cc

@@ -5571,6 +5571,15 @@ void VersionSet::Reset() {
   if (column_family_set_) {
     WriteBufferManager* wbm = column_family_set_->write_buffer_manager();
     WriteController* wc = column_family_set_->write_controller();
+
+    // Clear TableCache to prevent use-after-free: Reset() deletes old
+    // ColumnFamilySet but reuses table_cache_, which may contain
+    // BlockBasedTable entries with dangling references to deleted CFD's
+    // ioptions.
+    if (table_cache_) {
+      table_cache_->EraseUnRefEntries();
+    }
+
     // db_id becomes the source of truth after DBImpl::Recover():
     // https://github.com/facebook/rocksdb/blob/v7.3.1/db/db_impl/db_impl_open.cc#L527
     // Note: we may not be able to recover db_id from MANIFEST if

env/env_test.cc

@@ -1655,42 +1655,6 @@ void GenerateFilesAndRequest(Env* env, const std::string& fname,
   }
 }
 
-TEST_F(EnvPosixTest, MultiReadIOUringError) {
-  // In this test we don't do aligned read, so we can't do direct I/O.
-  EnvOptions soptions;
-  soptions.use_direct_reads = soptions.use_direct_writes = false;
-  std::string fname = test::PerThreadDBPath(env_, "testfile");
-
-  std::vector<std::string> scratches;
-  std::vector<ReadRequest> reqs;
-  GenerateFilesAndRequest(env_, fname, &reqs, &scratches);
-  // Query the data
-  std::unique_ptr<RandomAccessFile> file;
-  ASSERT_OK(env_->NewRandomAccessFile(fname, &file, soptions));
-
-  bool io_uring_wait_cqe_called = false;
-  SyncPoint::GetInstance()->SetCallBack(
-      "PosixRandomAccessFile::MultiRead:io_uring_wait_cqe:return",
-      [&](void* arg) {
-        if (!io_uring_wait_cqe_called) {
-          io_uring_wait_cqe_called = true;
-          ssize_t& ret = *(static_cast<ssize_t*>(arg));
-          ret = 1;
-        }
-      });
-  SyncPoint::GetInstance()->EnableProcessing();
-
-  Status s = file->MultiRead(reqs.data(), reqs.size());
-  if (io_uring_wait_cqe_called) {
-    ASSERT_NOK(s);
-  } else {
-    s.PermitUncheckedError();
-  }
-
-  SyncPoint::GetInstance()->DisableProcessing();
-  SyncPoint::GetInstance()->ClearAllCallBacks();
-}
-
 TEST_F(EnvPosixTest, MultiReadIOUringError2) {
   // In this test we don't do aligned read, so we can't do direct I/O.
   EnvOptions soptions;
@@ -1706,19 +1670,20 @@ TEST_F(EnvPosixTest, MultiReadIOUringError2) {
 
   bool io_uring_submit_and_wait_called = false;
   SyncPoint::GetInstance()->SetCallBack(
-      "PosixRandomAccessFile::MultiRead:io_uring_submit_and_wait:return1",
+      "PosixRandomAccessFile::MultiRead:io_uring_sq_ready:return1",
       [&](void* arg) {
         io_uring_submit_and_wait_called = true;
-        ssize_t* ret = static_cast<ssize_t*>(arg);
-        (*ret)--;
+        unsigned* ret = static_cast<unsigned*>(arg);
+        *ret = 1;
       });
   SyncPoint::GetInstance()->SetCallBack(
       "PosixRandomAccessFile::MultiRead:io_uring_submit_and_wait:return2",
       [&](void* arg) {
         struct io_uring* iu = static_cast<struct io_uring*>(arg);
         struct io_uring_cqe* cqe;
-        assert(io_uring_wait_cqe(iu, &cqe) == 0);
-        io_uring_cqe_seen(iu, cqe);
+        // CQ should be empty after drain - peek should fail
+        int ret = io_uring_peek_cqe(iu, &cqe);
+        assert(-EAGAIN == ret);  // No CQEs available
       });
   SyncPoint::GetInstance()->EnableProcessing();
 
@@ -3640,6 +3605,126 @@ TEST_F(TestAsyncRead, ReadAsync) {
   }
 }
 
+// Test ReadAsync -> MultiRead -> Poll with real io_uring (not mock).
+// This verifies that MultiRead doesn't interfere with async read buffers.
+TEST_F(TestAsyncRead, InterleavingIOUringOperations) {
+#if defined(ROCKSDB_IOURING_PRESENT)
+  // Use the real filesystem directly (not the mock ReadAsyncFS).
+  std::shared_ptr<FileSystem> fs = env_->GetFileSystem();
+  std::string fname = test::PerThreadDBPath(env_, "testfile_iouring");
+
+  constexpr size_t kSectorSize = 4096;
+  constexpr size_t kNumSectors = 8;
+
+  // 1. Create & write to a file.
+  {
+    std::unique_ptr<FSWritableFile> wfile;
+    ASSERT_OK(
+        fs->NewWritableFile(fname, FileOptions(), &wfile, nullptr /*dbg*/));
+
+    for (size_t i = 0; i < kNumSectors; ++i) {
+      auto data = NewAligned(kSectorSize * 8, static_cast<char>(i + 1));
+      Slice slice(data.get(), kSectorSize);
+      ASSERT_OK(wfile->Append(slice, IOOptions(), nullptr));
+    }
+    ASSERT_OK(wfile->Close(IOOptions(), nullptr));
+  }
+
+  // 2. Test interleaved ReadAsync and MultiRead operations.
+  {
+    std::unique_ptr<FSRandomAccessFile> file;
+    ASSERT_OK(fs->NewRandomAccessFile(fname, FileOptions(), &file, nullptr));
+
+    IOOptions opts;
+    std::vector<void*> io_handles(kNumSectors);
+    std::vector<FSReadRequest> async_reqs(kNumSectors);
+    std::vector<std::unique_ptr<char, Deleter>> async_data;
+    std::vector<size_t> vals;
+    IOHandleDeleter del_fn;
+
+    // Initialize async read requests.
+    for (size_t i = 0; i < kNumSectors; i++) {
+      async_reqs[i].offset = i * kSectorSize;
+      async_reqs[i].len = kSectorSize;
+      async_data.emplace_back(NewAligned(kSectorSize, 0));
+      async_reqs[i].scratch = async_data.back().get();
+      vals.push_back(i);
+    }
+
+    // Callback function for async reads.
+    std::function<void(FSReadRequest&, void*)> callback =
+        [&](FSReadRequest& req, void* cb_arg) {
+          assert(cb_arg != nullptr);
+          size_t i = *(reinterpret_cast<size_t*>(cb_arg));
+          async_reqs[i].offset = req.offset;
+          async_reqs[i].result = req.result;
+          async_reqs[i].status = req.status;
+        };
+
+    // Submit asynchronous read requests.
+    for (size_t i = 0; i < kNumSectors; i++) {
+      void* cb_arg = static_cast<void*>(&(vals[i]));
+      IOStatus s = file->ReadAsync(async_reqs[i], opts, callback, cb_arg,
+                                   &(io_handles[i]), &del_fn, nullptr);
+      if (s.IsNotSupported()) {
+        // io_uring not supported on this system, skip the test.
+        fprintf(stderr, "Skipping test - io_uring not supported: %s\n",
+                s.ToString().c_str());
+        for (size_t j = 0; j < i; j++) {
+          if (io_handles[j] != nullptr) {
+            del_fn(io_handles[j]);
+          }
+        }
+        return;
+      }
+      // For any other error, fail the test.
+      ASSERT_OK(s);
+    }
+
+    // Do a MultiRead on same sectors while async reads are submitted.
+    std::vector<FSReadRequest> multi_reqs(kNumSectors);
+    std::vector<std::unique_ptr<char, Deleter>> multi_data;
+    for (size_t i = 0; i < kNumSectors; i++) {
+      multi_reqs[i].offset = i * kSectorSize;
+      multi_reqs[i].len = kSectorSize;
+      multi_data.emplace_back(NewAligned(kSectorSize, 0));
+      multi_reqs[i].scratch = multi_data.back().get();
+    }
+    ASSERT_OK(file->MultiRead(multi_reqs.data(), kNumSectors, opts, nullptr));
+
+    // Check the status of MultiRead requests (should all succeed).
+    for (size_t i = 0; i < kNumSectors; i++) {
+      auto buf = NewAligned(kSectorSize * 8, static_cast<char>(i + 1));
+      Slice expected_data(buf.get(), kSectorSize);
+
+      ASSERT_EQ(multi_reqs[i].offset, i * kSectorSize);
+      ASSERT_OK(multi_reqs[i].status);
+      ASSERT_EQ(expected_data.ToString(), multi_reqs[i].result.ToString());
+    }
+
+    // Poll for the submitted async requests.
+    ASSERT_OK(fs->Poll(io_handles, kNumSectors));
+
+    // Check the status of async read requests (should all succeed).
+    for (size_t i = 0; i < kNumSectors; i++) {
+      auto buf = NewAligned(kSectorSize * 8, static_cast<char>(i + 1));
+      Slice expected_data(buf.get(), kSectorSize);
+
+      ASSERT_EQ(async_reqs[i].offset, i * kSectorSize);
+      ASSERT_OK(async_reqs[i].status);
+      ASSERT_EQ(expected_data.ToString(), async_reqs[i].result.ToString());
+    }
+
+    // Delete io_handles.
+    for (size_t i = 0; i < io_handles.size(); i++) {
+      del_fn(io_handles[i]);
+    }
+  }
+#else
+  fprintf(stderr, "Skipping test - ROCKSDB_IOURING_PRESENT not defined\n");
+#endif
+}
+
 struct StaticDestructionTester {
   bool activated = false;
   ~StaticDestructionTester() {

env/fs_posix.cc

@@ -270,7 +270,10 @@ class PosixFileSystem : public FileSystem {
           options
 #if defined(ROCKSDB_IOURING_PRESENT)
           ,
-          !IsIOUringEnabled() ? nullptr : thread_local_io_urings_.get()
+          !IsIOUringEnabled() ? nullptr
+                              : thread_local_async_read_io_urings_.get(),
+          !IsIOUringEnabled() ? nullptr
+                              : thread_local_multi_read_io_urings_.get()
 #endif
               ));
     }
@@ -1087,8 +1090,9 @@ class PosixFileSystem : public FileSystem {
 #if defined(ROCKSDB_IOURING_PRESENT)
     // io_uring_queue_init.
     struct io_uring* iu = nullptr;
-    if (thread_local_io_urings_) {
-      iu = static_cast<struct io_uring*>(thread_local_io_urings_->Get());
+    if (thread_local_async_read_io_urings_) {
+      iu = static_cast<struct io_uring*>(
+          thread_local_async_read_io_urings_->Get());
     }
 
     // Init failed, platform doesn't support io_uring.
@@ -1161,8 +1165,9 @@ class PosixFileSystem : public FileSystem {
 #if defined(ROCKSDB_IOURING_PRESENT)
     // io_uring_queue_init.
     struct io_uring* iu = nullptr;
-    if (thread_local_io_urings_) {
-      iu = static_cast<struct io_uring*>(thread_local_io_urings_->Get());
+    if (thread_local_async_read_io_urings_) {
+      iu = static_cast<struct io_uring*>(
+          thread_local_async_read_io_urings_->Get());
     }
 
     // Init failed, platform doesn't support io_uring.
@@ -1277,7 +1282,8 @@ class PosixFileSystem : public FileSystem {
 
 #if defined(ROCKSDB_IOURING_PRESENT)
   // io_uring instance
-  std::unique_ptr<ThreadLocalPtr> thread_local_io_urings_;
+  std::unique_ptr<ThreadLocalPtr> thread_local_async_read_io_urings_;
+  std::unique_ptr<ThreadLocalPtr> thread_local_multi_read_io_urings_;
 #endif
 
   size_t page_size_;
@@ -1337,7 +1343,8 @@ PosixFileSystem::PosixFileSystem()
   // io_uring can be created.
   struct io_uring* new_io_uring = CreateIOUring();
   if (new_io_uring != nullptr) {
-    thread_local_io_urings_.reset(new ThreadLocalPtr(DeleteIOUring));
+    thread_local_async_read_io_urings_.reset(new ThreadLocalPtr(DeleteIOUring));
+    thread_local_multi_read_io_urings_.reset(new ThreadLocalPtr(DeleteIOUring));
     delete new_io_uring;
   }
 #endif

env/io_posix.cc

@@ -589,7 +589,8 @@ PosixRandomAccessFile::PosixRandomAccessFile(
     const EnvOptions& options
 #if defined(ROCKSDB_IOURING_PRESENT)
     ,
-    ThreadLocalPtr* thread_local_io_urings
+    ThreadLocalPtr* thread_local_async_read_io_urings,
+    ThreadLocalPtr* thread_local_multi_read_io_urings
 #endif
     )
     : filename_(fname),
@@ -598,7 +599,8 @@ PosixRandomAccessFile::PosixRandomAccessFile(
       logical_sector_size_(logical_block_size)
 #if defined(ROCKSDB_IOURING_PRESENT)
       ,
-      thread_local_io_urings_(thread_local_io_urings)
+      thread_local_async_read_io_urings_(thread_local_async_read_io_urings),
+      thread_local_multi_read_io_urings_(thread_local_multi_read_io_urings)
 #endif
 {
   assert(!options.use_direct_reads || !options.use_mmap_reads);
@@ -659,6 +661,83 @@ IOStatus PosixRandomAccessFile::Read(uint64_t offset, size_t n,
   return s;
 }
 
+// MultiRead: Perform multiple concurrent read requests using io_uring.
+//
+// OVERVIEW:
+// This function batches multiple read requests and submits them concurrently
+// to io_uring for improved I/O performance. It operates synchronously from the
+// caller's perspective (blocks until all reads complete) but uses io_uring's
+// async capabilities internally for parallel I/O execution.
+//
+// IO_URING LIFECYCLE:
+// 1. Preparation Phase:
+//    - Allocate SQEs (Submission Queue Entries) for read requests
+//    - Limited by: min(pending_work, io_uring_sq_space_left(), kIoUringDepth -
+//    inflight)
+//    - Uses io_uring_sq_space_left() to query available SQ slots
+//    - Each SQE is tracked in wrap_cache for completion matching
+//
+// 2. Submission Phase:
+//    - Loop: while io_uring_sq_ready() > 0 (SQEs pending submission)
+//    - Call io_uring_submit_and_wait() to submit SQEs and wait for CQEs
+//    - Handles retryable errors (EINTR, EAGAIN) by continuing
+//    - Breaks on terminal errors (logs error, sets err variable)
+//
+// 3. Completion Phase:
+//    - Non-blocking CQE reaping via io_uring_for_each_cqe()
+//    - Matches CQEs to requests using user_data pointer
+//    - Processes results: updates bytes read, handles partial reads
+//    - Removes completed requests from wrap_cache
+//
+// 4. Loop Iteration:
+//    - Repeats until: all requests submitted AND all completions reaped
+//    - Termination condition: (num_reqs == reqs_off) &&
+//    resubmit_rq_list.empty() && wrap_cache.empty()
+//
+// ERROR HANDLING STRATEGY:
+// - Retryable submission errors (-EINTR, -EAGAIN): Retry submission
+// - Memory pressure (-ENOMEM): Mark memory_pressure_on_submission, attempt
+// recovery
+// - Terminal submission errors: Break, enter teardown path
+// - Retryable CQE errors (-EINTR, -EAGAIN): Add to resubmit_rq_list for retry
+// - Terminal CQE errors: Set ios to IOError, continue processing other CQEs
+// - Teardown path: If SQEs remain unsubmitted after error, reap submitted CQEs,
+//   destroy io_uring instance, return error
+//
+// PARTIAL READ HANDLING:
+// - Short reads (bytes_read < requested): Request added to resubmit_rq_list
+// - finished_len tracks cumulative bytes read across resubmissions
+// - iov.iov_base/iov_len adjusted on each resubmission attempt
+// - UpdateResult() determines if read should be retried based on:
+//   * Direct I/O alignment requirements
+//   * EOF detection
+//   * Error conditions
+//
+// RESUBMISSION LOGIC:
+// - resubmit_rq_list: Requests needing retry (short reads, EINTR/EAGAIN errors)
+// - Prioritized in SQE allocation loop: resubmits before new requests
+// - List cleared after SQE preparation
+// - Requests remain in wrap_cache across resubmissions until fully complete
+//
+// CONCURRENCY CONTROL:
+// - wrap_cache.size(): Tracks total inflight requests (SQ + CQ)
+// - io_uring_sq_ready(): Queries SQEs prepared but not yet submitted
+// - io_uring_sq_space_left(): Queries available SQ slots
+// - Max concurrency: kIoUringDepth (256)
+//
+// ACCOUNTING CORRECTNESS:
+// - Uses io_uring native APIs (io_uring_sq_ready, io_uring_sq_space_left)
+//   instead of manual counters for robustness
+// - wrap_cache is the authoritative source for inflight request tracking
+// - Re-query io_uring_sq_ready() after submission loop to detect
+//   unsubmitted SQEs (indicates submission errors)
+//
+// THREAD SAFETY:
+// - Uses thread-local io_uring instance (thread_local_multi_read_io_urings_)
+// - IORING_SETUP_SINGLE_ISSUER: Only one thread submits to this ring
+// - IORING_SETUP_DEFER_TASKRUN: Task work runs in submitting thread
+// - No cross-thread coordination required
+//
 IOStatus PosixRandomAccessFile::MultiRead(FSReadRequest* reqs, size_t num_reqs,
                                           const IOOptions& options,
                                           IODebugContext* dbg) {
@@ -672,12 +751,16 @@ IOStatus PosixRandomAccessFile::MultiRead(FSReadRequest* reqs, size_t num_reqs,
 
 #if defined(ROCKSDB_IOURING_PRESENT)
   struct io_uring* iu = nullptr;
-  if (thread_local_io_urings_) {
-    iu = static_cast<struct io_uring*>(thread_local_io_urings_->Get());
+  if (thread_local_multi_read_io_urings_) {
+    iu = static_cast<struct io_uring*>(
+        thread_local_multi_read_io_urings_->Get());
     if (iu == nullptr) {
-      iu = CreateIOUring();
+      unsigned int flags = 0;
+      flags |= IORING_SETUP_SINGLE_ISSUER;
+      flags |= IORING_SETUP_DEFER_TASKRUN;
+      iu = CreateIOUring(flags);
       if (iu != nullptr) {
-        thread_local_io_urings_->Reset(iu);
+        thread_local_multi_read_io_urings_->Reset(iu);
       }
     }
   }
@@ -688,8 +771,6 @@ IOStatus PosixRandomAccessFile::MultiRead(FSReadRequest* reqs, size_t num_reqs,
     return FSRandomAccessFile::MultiRead(reqs, num_reqs, options, dbg);
   }
 
-  IOStatus ios = IOStatus::OK();
-
   struct WrappedReadRequest {
     FSReadRequest* req;
     struct iovec iov;
@@ -698,118 +779,199 @@ IOStatus PosixRandomAccessFile::MultiRead(FSReadRequest* reqs, size_t num_reqs,
   };
 
   autovector<WrappedReadRequest, 32> req_wraps;
-  autovector<WrappedReadRequest*, 4> incomplete_rq_list;
+  autovector<WrappedReadRequest*, 4> resubmit_rq_list;
   std::unordered_set<WrappedReadRequest*> wrap_cache;
 
   for (size_t i = 0; i < num_reqs; i++) {
     req_wraps.emplace_back(&reqs[i]);
   }
 
+  IOStatus ios = IOStatus::OK();
   size_t reqs_off = 0;
-  while (num_reqs > reqs_off || !incomplete_rq_list.empty()) {
-    size_t this_reqs = (num_reqs - reqs_off) + incomplete_rq_list.size();
-
-    // If requests exceed depth, split it into batches
-    if (this_reqs > kIoUringDepth) {
-      this_reqs = kIoUringDepth;
-    }
-
-    assert(incomplete_rq_list.size() <= this_reqs);
-    for (size_t i = 0; i < this_reqs; i++) {
-      WrappedReadRequest* rep_to_submit;
-      if (i < incomplete_rq_list.size()) {
-        rep_to_submit = incomplete_rq_list[i];
+  while ((num_reqs > reqs_off) || !resubmit_rq_list.empty() ||
+         !wrap_cache.empty()) {
+    assert(resubmit_rq_list.size() + wrap_cache.size() <= kIoUringDepth);
+    // Total number of requests that still need to be submitted, includes:
+    //
+    //  1) requests NOT yet submitted (num_reqs - reqs_off)
+    //  2) requests on resubmission list (resubmit_rq_list)
+    //
+    // capped by min of the # of remaining entries in IO ring submission queue
+    // and the max IO ring depth less the inflight requests.
+    size_t new_sqe_reqs_count = std::min({
+        num_reqs - reqs_off + resubmit_rq_list.size(),
+        static_cast<size_t>(io_uring_sq_space_left(iu)),
+        kIoUringDepth - wrap_cache.size()  // queue depth less inflight requests
+    });
+    for (size_t i = 0; i < new_sqe_reqs_count; i++) {
+      WrappedReadRequest* req;
+      if (i < resubmit_rq_list.size()) {
+        req = resubmit_rq_list[i];
       } else {
-        rep_to_submit = &req_wraps[reqs_off++];
+        req = &req_wraps[reqs_off++];
       }
-      assert(rep_to_submit->req->len > rep_to_submit->finished_len);
-      rep_to_submit->iov.iov_base =
-          rep_to_submit->req->scratch + rep_to_submit->finished_len;
-      rep_to_submit->iov.iov_len =
-          rep_to_submit->req->len - rep_to_submit->finished_len;
+      assert(req->req->len > req->finished_len);
+      req->iov.iov_base = req->req->scratch + req->finished_len;
+      req->iov.iov_len = req->req->len - req->finished_len;
 
       struct io_uring_sqe* sqe;
       sqe = io_uring_get_sqe(iu);
-      io_uring_prep_readv(
-          sqe, fd_, &rep_to_submit->iov, 1,
-          rep_to_submit->req->offset + rep_to_submit->finished_len);
-      io_uring_sqe_set_data(sqe, rep_to_submit);
-      wrap_cache.emplace(rep_to_submit);
+      // NULL is unexpected as we do maintain proper ring accounting.
+      assert(sqe);
+      io_uring_prep_readv(sqe, fd_, &req->iov, 1,
+                          req->req->offset + req->finished_len);
+      io_uring_sqe_set_data(sqe, req);
+      wrap_cache.emplace(req);
     }
-    incomplete_rq_list.clear();
+    resubmit_rq_list.clear();
 
-    ssize_t ret =
-        io_uring_submit_and_wait(iu, static_cast<unsigned int>(this_reqs));
-    TEST_SYNC_POINT_CALLBACK(
-        "PosixRandomAccessFile::MultiRead:io_uring_submit_and_wait:return1",
-        &ret);
-    TEST_SYNC_POINT_CALLBACK(
-        "PosixRandomAccessFile::MultiRead:io_uring_submit_and_wait:return2",
-        iu);
-
-    if (static_cast<size_t>(ret) != this_reqs) {
-      fprintf(stderr, "ret = %ld this_reqs: %ld\n", (long)ret, (long)this_reqs);
-      // If error happens and we submitted fewer than expected, it is an
-      // exception case and we don't retry here. We should still consume
-      // what is is submitted in the ring.
-      for (ssize_t i = 0; i < ret; i++) {
-        struct io_uring_cqe* cqe = nullptr;
-        io_uring_wait_cqe(iu, &cqe);
-        if (cqe != nullptr) {
-          io_uring_cqe_seen(iu, cqe);
+    struct io_uring_cqe* cqe = nullptr;
+    unsigned head;
+    ssize_t err = 0;
+    bool memory_pressure_on_submission = false;
+    unsigned reqs_pending_submission;
+    unsigned reqs_submitted = 0;
+    while ((reqs_pending_submission = io_uring_sq_ready(iu))) {
+      // MultiRead is synchronous in nature. io_uring_submit_and_wait provides
+      // batching semantics (submit + best effort wait in one syscall), while
+      // io_uring_submit enables async producer/consumer semantics (submit
+      // only, requires separate reaping). We chose batching approach to
+      // reduce the volume of syscalls and context switches.
+      ssize_t ret = io_uring_submit_and_wait(iu, reqs_pending_submission);
+      if (ret < 0) {
+        if (-EINTR == ret || -EAGAIN == ret) {
+          // Submission failed due to rare, retryable syscall error. Try again.
+          continue;
         }
-      }
-      return IOStatus::IOError("io_uring_submit_and_wait() requested " +
-                               std::to_string(this_reqs) + " but returned " +
-                               std::to_string(ret));
-    }
-
-    for (size_t i = 0; i < this_reqs; i++) {
-      struct io_uring_cqe* cqe = nullptr;
-      WrappedReadRequest* req_wrap;
-
-      // We could use the peek variant here, but this seems safer in terms
-      // of our initial wait not reaping all completions
-      ret = io_uring_wait_cqe(iu, &cqe);
-      TEST_SYNC_POINT_CALLBACK(
-          "PosixRandomAccessFile::MultiRead:io_uring_wait_cqe:return", &ret);
-      if (ret) {
-        ios = IOStatus::IOError("io_uring_wait_cqe() returns " +
-                                std::to_string(ret));
-
-        if (cqe != nullptr) {
-          io_uring_cqe_seen(iu, cqe);
+        if (-ENOMEM == ret) {
+          fprintf(stderr,
+                  "PosixRandomAccessFile::MultiRead: io_uring_submit_and_wait "
+                  "experienced terse memory condition.\n");
+          // Best effort to reclaim resources in terse condition.
+          memory_pressure_on_submission = true;
+        } else {
+          fprintf(stderr,
+                  "PosixRandomAccessFile::MultiRead: "
+                  "io_uring_submit_and_wait returned terminal error: %zd.\n",
+                  ret);
+          err = ret;
         }
-        continue;
+        break;
       }
-
-      req_wrap = static_cast<WrappedReadRequest*>(io_uring_cqe_get_data(cqe));
-      // Reset cqe data to catch any stray reuse of it
-      static_cast<struct io_uring_cqe*>(cqe)->user_data = 0xd5d5d5d5d5d5d5d5;
-      // Check that we got a valid unique cqe data
-      auto wrap_check = wrap_cache.find(req_wrap);
-      if (wrap_check == wrap_cache.end()) {
+      if (0 == ret) {
+        // This scenario is unexpected for any modern kernel!
+        // We deliberately error out to avoid bugs around infinite loops.
         fprintf(stderr,
                 "PosixRandomAccessFile::MultiRead: "
-                "Bad cqe data from IO uring - %p\n",
-                req_wrap);
-        port::PrintStack();
-        ios = IOStatus::IOError("io_uring_cqe_get_data() returned " +
-                                std::to_string((uint64_t)req_wrap));
-        continue;
+                "io_uring_submit_and_wait returned 0 submissions!\n");
+        break;
       }
-      wrap_cache.erase(wrap_check);
+      reqs_submitted += static_cast<unsigned int>(ret);
+    };
+    reqs_pending_submission = io_uring_sq_ready(iu);
 
-      FSReadRequest* req = req_wrap->req;
-      size_t bytes_read = 0;
-      bool read_again = false;
-      UpdateResult(cqe, filename_, req->len, req_wrap->iov.iov_len,
-                   false /*async_read*/, use_direct_io(),
-                   GetRequiredBufferAlignment(), req_wrap->finished_len, req,
-                   bytes_read, read_again);
-      int32_t res = cqe->res;
-      if (res >= 0) {
-        if (bytes_read == 0) {
+    TEST_SYNC_POINT_CALLBACK(
+        "PosixRandomAccessFile::MultiRead:io_uring_sq_ready:return1",
+        &reqs_pending_submission);
+
+    // Error occurred or IO uring stopped submitting outstanding requests.
+    if (reqs_pending_submission && !memory_pressure_on_submission) {
+      // IO ring is initialized once in thread-local variable and then reused
+      // to handle the consecutive MultiRead API calls. Therefore, it's crucial
+      // to reap all the submitted requests.
+      //
+      // NOTE: Loop will run indefinitely until we reap all the completions!!!
+      size_t nr = 0;
+      assert(reqs_pending_submission <= wrap_cache.size());
+      size_t nr_await_cqe = wrap_cache.size() - reqs_pending_submission;
+      while (nr < nr_await_cqe) {
+        // blocking
+        io_uring_wait_cqes(iu, &cqe,
+                           static_cast<unsigned int>(nr_await_cqe - nr),
+                           nullptr, nullptr);
+        size_t reaped_cqe_count = 0;
+        io_uring_for_each_cqe(iu, head, cqe) { reaped_cqe_count++; }
+        if (reaped_cqe_count > 0) {
+          io_uring_cq_advance(iu, static_cast<unsigned int>(reaped_cqe_count));
+          nr += reaped_cqe_count;
+        }
+      }
+
+      TEST_SYNC_POINT_CALLBACK(
+          "PosixRandomAccessFile::MultiRead:io_uring_submit_and_wait:return2",
+          iu);
+
+      // While all the submitted completions have been reaped successfully,
+      // IO ring submission queue still contains at least one non-submitted
+      // request. Destroy io_uring (discards unsubmitted SQEs).
+      //
+      // NOTE: This is a rare scenario and should not happen in normal cases.
+      //       Hence, this should NOT materially impact the performance metrics.
+      io_uring_queue_exit(iu);
+      delete iu;
+      thread_local_multi_read_io_urings_->Reset(nullptr);
+
+      if (err < 0) {
+        return IOStatus::IOError(
+            "io_uring_submit_and_wait() failed with an error " +
+            std::to_string(err));
+      }
+      return IOStatus::IOError(
+          "io_uring_submit_and_wait() requested " +
+          std::to_string(reqs_submitted + reqs_pending_submission) +
+          " but returned " + std::to_string(reqs_submitted));
+    }
+
+    if ((0 == reqs_submitted) && wrap_cache.size() > reqs_pending_submission) {
+      // If no requests have been submitted and there is at least one request
+      // pending completion, wait for at least one completion to arrive.
+      // This is a guardrail to prevent the busy CPU loops.
+      //
+      // NOTE: it's not really a tight CPU-burning loop in the traditional sense
+      // as it's naturally throttled by the io_uring_submit_and_wait() syscall.
+      io_uring_wait_cqe(iu, &cqe);
+    }
+
+    unsigned int nr = 0;
+    io_uring_for_each_cqe(iu, head, cqe) {  // non-blocking
+      if (cqe->user_data) {  // non-discarded, valid user data only!
+        nr++;
+        WrappedReadRequest* req_wrap =
+            static_cast<WrappedReadRequest*>(io_uring_cqe_get_data(cqe));
+        // Reset cqe data to catch any stray reuse of it
+        static_cast<struct io_uring_cqe*>(cqe)->user_data = 0xd5d5d5d5d5d5d5d5;
+        // Check that we got a valid unique cqe data
+        auto wrap_check = wrap_cache.find(req_wrap);
+        if (wrap_check == wrap_cache.end()) {
+          fprintf(stderr,
+                  "PosixRandomAccessFile::MultiRead: "
+                  "Bad cqe data from IO uring - %p\n",
+                  req_wrap);
+          port::PrintStack();
+          ios = IOStatus::IOError("io_uring_cqe_get_data() returned " +
+                                  std::to_string((uint64_t)req_wrap));
+          continue;
+        }
+        wrap_cache.erase(wrap_check);
+        if (cqe->res < 0) {
+          if (-EINTR == cqe->res || -EAGAIN == cqe->res) {
+            resubmit_rq_list.push_back(req_wrap);
+          } else {
+            ios = IOStatus::IOError("io_uring_for_each_cqe() returns " +
+                                    std::to_string(cqe->res));
+          }
+          continue;
+        }
+        // cqe->res >= 0
+        FSReadRequest* req = req_wrap->req;
+        size_t bytes_read = 0;
+        bool read_again = false;
+        UpdateResult(cqe, filename_, req->len, req_wrap->iov.iov_len,
+                     false /*async_read*/, use_direct_io(),
+                     GetRequiredBufferAlignment(), req_wrap->finished_len, req,
+                     bytes_read, read_again);
+
+        if (0 == bytes_read) {
           if (read_again) {
             Slice tmp_slice;
             req->status =
@@ -819,14 +981,15 @@ IOStatus PosixRandomAccessFile::MultiRead(FSReadRequest* reqs, size_t num_reqs,
             req->result =
                 Slice(req->scratch, req_wrap->finished_len + tmp_slice.size());
           }
-          // else It means EOF so no need to do anything.
+          // else it means EOF so no need to do anything.
         } else if (bytes_read < req_wrap->iov.iov_len) {
-          incomplete_rq_list.push_back(req_wrap);
+          resubmit_rq_list.push_back(req_wrap);
         }
       }
-      io_uring_cqe_seen(iu, cqe);
     }
-    wrap_cache.clear();
+    if (nr > 0) {
+      io_uring_cq_advance(iu, nr);
+    }
   }
   return ios;
 #else
@@ -923,12 +1086,16 @@ IOStatus PosixRandomAccessFile::ReadAsync(
 #if defined(ROCKSDB_IOURING_PRESENT)
   // io_uring_queue_init.
   struct io_uring* iu = nullptr;
-  if (thread_local_io_urings_) {
-    iu = static_cast<struct io_uring*>(thread_local_io_urings_->Get());
+  if (thread_local_async_read_io_urings_) {
+    iu = static_cast<struct io_uring*>(
+        thread_local_async_read_io_urings_->Get());
     if (iu == nullptr) {
-      iu = CreateIOUring();
+      unsigned int flags = 0;
+      flags |= IORING_SETUP_SINGLE_ISSUER;
+      flags |= IORING_SETUP_DEFER_TASKRUN;
+      iu = CreateIOUring(flags);
       if (iu != nullptr) {
-        thread_local_io_urings_->Reset(iu);
+        thread_local_async_read_io_urings_->Reset(iu);
       }
     }
   }
@@ -966,11 +1133,35 @@ IOStatus PosixRandomAccessFile::ReadAsync(
   io_uring_sqe_set_data(sqe, posix_handle);
 
   // Step 4: io_uring_submit
-  ssize_t ret = io_uring_submit(iu);
-  if (ret < 0) {
-    fprintf(stderr, "io_uring_submit error: %ld\n", long(ret));
-    return IOStatus::IOError("io_uring_submit() requested but returned " +
-                             std::to_string(ret));
+  ssize_t ret;
+  do {
+    ret = io_uring_submit(iu);
+    if (ret < 0) {
+      if (-EINTR == ret || -EAGAIN == ret) {
+        // Submission failed due to transient error. Try again.
+        continue;
+      }
+      fprintf(stderr,
+              "PosixRandomAccessFile::ReadAsync: "
+              "io_uring_submit returned terminal error = %zd\n",
+              ret);
+      break;
+    }
+    if (0 == ret) {
+      // Unexpected. Will be reported as error.
+      break;
+    }
+  } while (ret < 1);
+  if (ret <= 0) {
+    return IOStatus::IOError(
+        "PosixRandomAccessFile::ReadAsync: io_uring_submit() returned " +
+        std::to_string(ret));
+  }
+  if (ret > 1) {
+    fprintf(stderr,
+            "PosixRandomAccessFile::ReadAsync: "
+            "io_uring_submit() returned = %zd\n",
+            ret);
   }
   return IOStatus::OK();
 #else

env/io_posix.h

@@ -11,6 +11,16 @@
 #if defined(ROCKSDB_IOURING_PRESENT)
 #include <liburing.h>
 #include <sys/uio.h>
+
+// Compatibility defines for io_uring flags that may not be present in older
+// kernel headers. These values are fixed and won't change, so it's safe to
+// define them even if the running kernel doesn't support them.
+#ifndef IORING_SETUP_SINGLE_ISSUER
+#define IORING_SETUP_SINGLE_ISSUER (1U << 12)
+#endif
+#ifndef IORING_SETUP_DEFER_TASKRUN
+#define IORING_SETUP_DEFER_TASKRUN (1U << 13)
+#endif
 #endif
 #include <unistd.h>
 
@@ -297,9 +307,9 @@ inline void DeleteIOUring(void* p) {
   delete iu;
 }
 
-inline struct io_uring* CreateIOUring() {
+inline struct io_uring* CreateIOUring(unsigned int flags = 0) {
   struct io_uring* new_io_uring = new struct io_uring;
-  int ret = io_uring_queue_init(kIoUringDepth, new_io_uring, 0);
+  int ret = io_uring_queue_init(kIoUringDepth, new_io_uring, flags);
   if (ret) {
     delete new_io_uring;
     new_io_uring = nullptr;
@@ -315,7 +325,8 @@ class PosixRandomAccessFile : public FSRandomAccessFile {
   bool use_direct_io_;
   size_t logical_sector_size_;
 #if defined(ROCKSDB_IOURING_PRESENT)
-  ThreadLocalPtr* thread_local_io_urings_;
+  ThreadLocalPtr* thread_local_async_read_io_urings_;
+  ThreadLocalPtr* thread_local_multi_read_io_urings_;
 #endif
 
  public:
@@ -323,7 +334,8 @@ class PosixRandomAccessFile : public FSRandomAccessFile {
                         size_t logical_block_size, const EnvOptions& options
 #if defined(ROCKSDB_IOURING_PRESENT)
                         ,
-                        ThreadLocalPtr* thread_local_io_urings
+                        ThreadLocalPtr* thread_local_async_read_io_urings,
+                        ThreadLocalPtr* thread_local_multi_read_io_urings
 #endif
   );
   virtual ~PosixRandomAccessFile();

include/rocksdb/c.h

@@ -166,6 +166,31 @@ typedef struct rocksdb_writestallinfo_t rocksdb_writestallinfo_t;
 typedef struct rocksdb_writestallcondition_t rocksdb_writestallcondition_t;
 typedef struct rocksdb_memtableinfo_t rocksdb_memtableinfo_t;
 
+// Remote Compaction typedef
+typedef struct rocksdb_compactionservice_scheduleresponse_t
+    rocksdb_compactionservice_scheduleresponse_t;
+typedef struct rocksdb_compactionservice_jobinfo_t
+    rocksdb_compactionservice_jobinfo_t;
+typedef struct rocksdb_compactionservice_t rocksdb_compactionservice_t;
+typedef struct rocksdb_compaction_service_options_override_t
+    rocksdb_compaction_service_options_override_t;
+typedef struct rocksdb_open_and_compact_options_t
+    rocksdb_open_and_compact_options_t;
+typedef rocksdb_compactionservice_scheduleresponse_t* (
+    *rocksdb_compaction_service_schedule_cb)(
+    void* state, const rocksdb_compactionservice_jobinfo_t* info,
+    const char* compaction_service_input, size_t input_len);
+
+typedef int (*rocksdb_compaction_service_wait_cb)(void* state,
+                                                  const char* scheduled_job_id,
+                                                  char** result,
+                                                  size_t* result_len);
+
+typedef void (*rocksdb_compaction_service_cancel_awaiting_jobs_cb)(void* state);
+
+typedef void (*rocksdb_compaction_service_on_installation_cb)(
+    void* state, const char* scheduled_job_id, int status);
+
 /* DB operations */
 
 extern ROCKSDB_LIBRARY_API rocksdb_t* rocksdb_open(
@@ -3512,6 +3537,155 @@ extern ROCKSDB_LIBRARY_API unsigned char rocksdb_get_into_buffer_cf(
     size_t keylen, char* buffer, size_t buffer_size, size_t* vallen,
     unsigned char* found, char** errptr);
 
+// Remote compaction
+enum {
+  rocksdb_compactionservice_jobstatus_success = 0,
+  rocksdb_compactionservice_jobstatus_failure = 1,
+  rocksdb_compactionservice_jobstatus_aborted = 2,
+  rocksdb_compactionservice_jobstatus_use_local = 3,
+};
+
+extern ROCKSDB_LIBRARY_API rocksdb_compactionservice_scheduleresponse_t*
+rocksdb_compactionservice_scheduleresponse_create(const char* scheduled_job_id,
+                                                  int status, char** errptr);
+
+extern ROCKSDB_LIBRARY_API rocksdb_compactionservice_scheduleresponse_t*
+rocksdb_compactionservice_scheduleresponse_create_with_status(int status,
+                                                              char** errptr);
+
+extern ROCKSDB_LIBRARY_API int
+rocksdb_compactionservice_scheduleresponse_getstatus(
+    const rocksdb_compactionservice_scheduleresponse_t* response);
+
+extern ROCKSDB_LIBRARY_API const char*
+rocksdb_compactionservice_scheduleresponse_get_scheduled_job_id(
+    const rocksdb_compactionservice_scheduleresponse_t* response, size_t* len);
+
+extern ROCKSDB_LIBRARY_API void
+rocksdb_compactionservice_scheduleresponse_t_destroy(
+    rocksdb_compactionservice_scheduleresponse_t* response);
+
+extern ROCKSDB_LIBRARY_API const char*
+rocksdb_compactionservice_jobinfo_t_get_db_name(
+    const rocksdb_compactionservice_jobinfo_t* info, size_t* len);
+
+extern ROCKSDB_LIBRARY_API const char*
+rocksdb_compactionservice_jobinfo_t_get_db_id(
+    const rocksdb_compactionservice_jobinfo_t* info, size_t* len);
+
+extern ROCKSDB_LIBRARY_API const char*
+rocksdb_compactionservice_jobinfo_t_get_db_session_id(
+    const rocksdb_compactionservice_jobinfo_t* info, size_t* len);
+
+extern ROCKSDB_LIBRARY_API const char*
+rocksdb_compactionservice_jobinfo_t_get_cf_name(
+    const rocksdb_compactionservice_jobinfo_t* info, size_t* len);
+
+extern ROCKSDB_LIBRARY_API uint32_t
+rocksdb_compactionservice_jobinfo_t_get_cf_id(
+    const rocksdb_compactionservice_jobinfo_t* info);
+
+extern ROCKSDB_LIBRARY_API uint64_t
+rocksdb_compactionservice_jobinfo_t_get_job_id(
+    const rocksdb_compactionservice_jobinfo_t* info);
+
+extern ROCKSDB_LIBRARY_API int rocksdb_compactionservice_jobinfo_t_get_priority(
+    const rocksdb_compactionservice_jobinfo_t* info);
+
+extern ROCKSDB_LIBRARY_API int
+rocksdb_compactionservice_jobinfo_t_get_compaction_reason(
+    const rocksdb_compactionservice_jobinfo_t* info);
+
+extern ROCKSDB_LIBRARY_API int
+rocksdb_compactionservice_jobinfo_t_get_base_input_level(
+    const rocksdb_compactionservice_jobinfo_t* info);
+
+extern ROCKSDB_LIBRARY_API int
+rocksdb_compactionservice_jobinfo_t_get_output_level(
+    const rocksdb_compactionservice_jobinfo_t* info);
+
+extern ROCKSDB_LIBRARY_API unsigned char
+rocksdb_compactionservice_jobinfo_t_is_full_compaction(
+    const rocksdb_compactionservice_jobinfo_t* info);
+
+extern ROCKSDB_LIBRARY_API unsigned char
+rocksdb_compactionservice_jobinfo_t_is_manual_compaction(
+    const rocksdb_compactionservice_jobinfo_t* info);
+
+extern ROCKSDB_LIBRARY_API unsigned char
+rocksdb_compactionservice_jobinfo_t_is_bottommost_level(
+    const rocksdb_compactionservice_jobinfo_t* info);
+
+extern ROCKSDB_LIBRARY_API rocksdb_compactionservice_t*
+rocksdb_compactionservice_create(
+    void* state, void (*destructor)(void*),
+    rocksdb_compaction_service_schedule_cb schedule, const char* name,
+    rocksdb_compaction_service_wait_cb wait,
+    rocksdb_compaction_service_cancel_awaiting_jobs_cb cancel_awaiting_jobs,
+    rocksdb_compaction_service_on_installation_cb on_installation);
+
+extern ROCKSDB_LIBRARY_API void rocksdb_options_set_compaction_service(
+    rocksdb_options_t* options, rocksdb_compactionservice_t* service);
+
+// CompactionServiceOptionsOverride
+extern ROCKSDB_LIBRARY_API rocksdb_compaction_service_options_override_t*
+rocksdb_compaction_service_options_override_create(void);
+
+extern ROCKSDB_LIBRARY_API void
+rocksdb_compaction_service_options_override_destroy(
+    rocksdb_compaction_service_options_override_t* override_options);
+
+extern ROCKSDB_LIBRARY_API void
+rocksdb_compaction_service_options_override_set_env(
+    rocksdb_compaction_service_options_override_t* override_options,
+    rocksdb_env_t* env);
+
+extern ROCKSDB_LIBRARY_API void
+rocksdb_compaction_service_options_override_set_comparator(
+    rocksdb_compaction_service_options_override_t* override_options,
+    rocksdb_comparator_t* comparator);
+
+// Atomic bool management for cancellation
+// Creates an atomic bool that can be used for cancellation.
+// User must call rocksdb_open_and_compact_canceled_destroy() to free it.
+extern ROCKSDB_LIBRARY_API unsigned char*
+rocksdb_open_and_compact_canceled_create(void);
+
+extern ROCKSDB_LIBRARY_API void rocksdb_open_and_compact_canceled_destroy(
+    unsigned char* canceled);
+
+extern ROCKSDB_LIBRARY_API void rocksdb_open_and_compact_canceled_set(
+    unsigned char* canceled, unsigned char value);
+
+// OpenAndCompactOptions
+extern ROCKSDB_LIBRARY_API rocksdb_open_and_compact_options_t*
+rocksdb_open_and_compact_options_create(void);
+
+extern ROCKSDB_LIBRARY_API void rocksdb_open_and_compact_options_destroy(
+    rocksdb_open_and_compact_options_t* options);
+
+extern ROCKSDB_LIBRARY_API void rocksdb_open_and_compact_options_set_canceled(
+    rocksdb_open_and_compact_options_t* options, unsigned char* canceled);
+
+extern ROCKSDB_LIBRARY_API void
+rocksdb_open_and_compact_options_set_allow_resumption(
+    rocksdb_open_and_compact_options_t* options,
+    unsigned char allow_resumption);
+
+// OpenAndCompact - main functions
+extern ROCKSDB_LIBRARY_API char* rocksdb_open_and_compact(
+    const char* db_path, const char* output_directory, const char* input,
+    size_t input_len, size_t* output_len,
+    const rocksdb_compaction_service_options_override_t* override_options,
+    char** errptr);
+
+extern ROCKSDB_LIBRARY_API char* rocksdb_open_and_compact_with_options(
+    const rocksdb_open_and_compact_options_t* options, const char* db_path,
+    const char* output_directory, const char* input, size_t input_len,
+    size_t* output_len,
+    const rocksdb_compaction_service_options_override_t* override_options,
+    char** errptr);
+
 #ifdef __cplusplus
 } /* end extern "C" */
 #endif

include/rocksdb/db.h

@@ -1105,9 +1105,12 @@ class DB {
   // details. For optimal performance, ensure that either all entries in
   // scan_opts specify the range limit, or none of them do.
   //
-  // NOTE: iterate_upper_bound in ReadOptions will be ignored. Instead, the
-  // range.limit in ScanOptions is consulted to determine the upper bound key,
-  // if specified.
+  // NOTE: NOT YET SUPPORTED in DBs using user timestamp (see
+  // Comparator::timestamp_size())
+  //
+  // NOTE: iterate_upper_bound in ReadOptions will
+  // be ignored. Instead, the range.limit in ScanOptions is consulted to
+  // determine the upper bound key, if specified.
   //
   // Example usage -
   //  std::vector<ScanOptions> scans{{.start = Slice("bar")},

include/rocksdb/utilities/backup_engine.h

@@ -621,7 +621,14 @@ class BackupEngineAppendOnlyBase {
   // The backup will stop ASAP and the call to CreateNewBackup will
   // return Status::Incomplete(). It will not clean up after itself, but
   // the state will remain consistent. The state will be cleaned up the
-  // next time you call CreateNewBackup or GarbageCollect.
+  // next time you call CreateNewBackup or GarbageCollect for the same backup
+  // directory on a new BackupEngine object.
+  //
+  // NOTE: This is a one-way operation. Once StopBackup() is called on a
+  // BackupEngine instance, all subsequent backup requests (CreateNewBackup,
+  // CreateNewBackupWithMetadata) will fail with Status::Incomplete().
+  // To create new backups after calling StopBackup(), you must open a new
+  // BackupEngine instance.
   virtual void StopBackup() = 0;
 
   // Will delete any files left over from incomplete creation or deletion of

include/rocksdb/version.h

@@ -12,7 +12,7 @@
 // NOTE: in 'main' development branch, this should be the *next*
 // minor or major version number planned for release.
 #define ROCKSDB_MAJOR 10
-#define ROCKSDB_MINOR 9
+#define ROCKSDB_MINOR 10
 #define ROCKSDB_PATCH 1
 
 // Make it easy to do conditional compilation based on version checks, i.e.

table/block_based/block_based_table_iterator.h

@@ -652,8 +652,6 @@ class BlockBasedTableIterator : public InternalIteratorBase<Slice> {
             std::to_string(async_state->offset) + " and async callback " +
             std::to_string(req.offset));
       }
-    } else {
-      assert(async_state->status.IsAborted());
     }
   }
 

table/block_based/block_based_table_reader.cc

@@ -3232,7 +3232,8 @@ Status BlockBasedTable::GetKVPairsFromDataBlocks(
   return Status::OK();
 }
 
-Status BlockBasedTable::DumpTable(WritableFile* out_file) {
+Status BlockBasedTable::DumpTable(WritableFile* out_file,
+                                  bool show_sequence_number_type) {
   WritableFileStringStreamAdapter out_file_wrapper(out_file);
   std::ostream out_stream(&out_file_wrapper);
   // Output Footer
@@ -3325,15 +3326,15 @@ Status BlockBasedTable::DumpTable(WritableFile* out_file) {
       out_stream << "Range deletions:\n"
                     "--------------------------------------\n";
       for (; range_del_iter->Valid(); range_del_iter->Next()) {
-        DumpKeyValue(range_del_iter->key(), range_del_iter->value(),
-                     out_stream);
+        DumpKeyValue(range_del_iter->key(), range_del_iter->value(), out_stream,
+                     show_sequence_number_type);
       }
       out_stream << "\n";
     }
     delete range_del_iter;
   }
   // Output Data blocks
-  s = DumpDataBlocks(out_stream);
+  s = DumpDataBlocks(out_stream, show_sequence_number_type);
 
   if (!s.ok()) {
     return s;
@@ -3398,7 +3399,8 @@ Status BlockBasedTable::DumpIndexBlock(std::ostream& out_stream) {
   return Status::OK();
 }
 
-Status BlockBasedTable::DumpDataBlocks(std::ostream& out_stream) {
+Status BlockBasedTable::DumpDataBlocks(std::ostream& out_stream,
+                                       bool show_sequence_number_type) {
   // TODO: plumb Env::IOActivity, Env::IOPriority
   const ReadOptions read_options;
   std::unique_ptr<InternalIteratorBase<IndexValue>> blockhandles_iter(
@@ -3455,7 +3457,8 @@ Status BlockBasedTable::DumpDataBlocks(std::ostream& out_stream) {
         out_stream << "Error reading the block - Skipped \n";
         break;
       }
-      DumpKeyValue(datablock_iter->key(), datablock_iter->value(), out_stream);
+      DumpKeyValue(datablock_iter->key(), datablock_iter->value(), out_stream,
+                   show_sequence_number_type);
     }
     out_stream << "\n";
   }
@@ -3477,14 +3480,26 @@ Status BlockBasedTable::DumpDataBlocks(std::ostream& out_stream) {
 }
 
 void BlockBasedTable::DumpKeyValue(const Slice& key, const Slice& value,
-                                   std::ostream& out_stream) {
-  InternalKey ikey;
-  ikey.DecodeFrom(key);
+                                   std::ostream& out_stream,
+                                   bool show_sequence_number_type) {
+  ParsedInternalKey result;
+  auto s = ParseInternalKey(key, &result, true);
+  if (!s.ok()) {
+    out_stream << "Error parsing internal key - Skipped \n";
+    return;
+  }
 
-  out_stream << "  HEX    " << ikey.user_key().ToString(true) << ": "
-             << value.ToString(true) << "\n";
+  if (show_sequence_number_type) {
+    out_stream << "  HEX    " << result.user_key.ToString(true)
+               << "  seq: " << result.sequence
+               << "  type: " << std::to_string(result.type) << " : "
+               << value.ToString(true) << "\n";
+  } else {
+    out_stream << "  HEX    " << result.user_key.ToString(true) << ": "
+               << value.ToString(true) << "\n";
+  }
 
-  std::string str_key = ikey.user_key().ToString();
+  std::string str_key = result.user_key.ToString();
   std::string str_value = value.ToString();
   std::string res_key, res_value;
   char cspace = ' ';

table/block_based/block_based_table_reader.h

@@ -208,7 +208,8 @@ class BlockBasedTable : public TableReader {
   size_t ApproximateMemoryUsage() const override;
 
   // convert SST file to a human readable form
-  Status DumpTable(WritableFile* out_file) override;
+  Status DumpTable(WritableFile* out_file,
+                   bool show_sequence_number_type = false) override;
 
   Status VerifyChecksum(const ReadOptions& readOptions,
                         TableReaderCaller caller,
@@ -549,9 +550,11 @@ class BlockBasedTable : public TableReader {
 
   // Helper functions for DumpTable()
   Status DumpIndexBlock(std::ostream& out_stream);
-  Status DumpDataBlocks(std::ostream& out_stream);
+  Status DumpDataBlocks(std::ostream& out_stream,
+                        bool show_sequence_number_type = false);
   void DumpKeyValue(const Slice& key, const Slice& value,
-                    std::ostream& out_stream);
+                    std::ostream& out_stream,
+                    bool show_sequence_number_type = false);
 
   // Returns false if prefix_extractor exists and is compatible with that used
   // in building the table file, otherwise true.

table/sst_file_dumper.cc

@@ -47,12 +47,13 @@ SstFileDumper::SstFileDumper(const Options& options,
                              Temperature file_temp, size_t readahead_size,
                              bool verify_checksum, bool output_hex,
                              bool decode_blob_index, const EnvOptions& soptions,
-                             bool silent)
+                             bool silent, bool show_sequence_number_type)
     : file_name_(file_path),
       read_num_(0),
       file_temp_(file_temp),
       output_hex_(output_hex),
       decode_blob_index_(decode_blob_index),
+      show_sequence_number_type_(show_sequence_number_type),
       soptions_(soptions),
       silent_(silent),
       options_(options),
@@ -220,7 +221,7 @@ Status SstFileDumper::DumpTable(const std::string& out_filename) {
   Env* env = options_.env;
   Status s = env->NewWritableFile(out_filename, &out_file, soptions_);
   if (s.ok()) {
-    s = table_reader_->DumpTable(out_file.get());
+    s = table_reader_->DumpTable(out_file.get(), show_sequence_number_type_);
   }
   if (!s.ok()) {
     // close the file before return error, ignore the close error if there's any

table/sst_file_dumper.h

@@ -21,7 +21,8 @@ class SstFileDumper {
                          bool verify_checksum, bool output_hex,
                          bool decode_blob_index,
                          const EnvOptions& soptions = EnvOptions(),
-                         bool silent = false);
+                         bool silent = false,
+                         bool show_sequence_number_type = false);
 
   // read_num_limit limits the total number of keys read. If read_num_limit = 0,
   // then there is no limit. If read_num_limit = 0 or
@@ -79,6 +80,7 @@ class SstFileDumper {
   Temperature file_temp_;
   bool output_hex_;
   bool decode_blob_index_;
+  bool show_sequence_number_type_;
   EnvOptions soptions_;
   // less verbose in stdout/stderr
   bool silent_;

table/table_reader.h

@@ -179,7 +179,8 @@ class TableReader {
   }
 
   // convert db file to a human readable form
-  virtual Status DumpTable(WritableFile* /*out_file*/) {
+  virtual Status DumpTable(WritableFile* /*out_file*/,
+                           bool /*show_sequence_number_type*/ = false) {
     return Status::NotSupported("DumpTable() not supported");
   }
 

test_util/sync_point.cc

@@ -79,4 +79,8 @@ void SetupSyncPointsToMockDirectIO() {
   ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
 #endif
 }
+
+#ifndef NDEBUG
+std::atomic<int> g_throw_on_testable_assertion_failure{0};
+#endif  // NDEBUG
 }  // namespace ROCKSDB_NAMESPACE

test_util/sync_point.h

@@ -180,3 +180,35 @@ void SetupSyncPointsToMockDirectIO();
     }                                               \
   }
 #endif  // NDEBUG
+
+// An alternative to assert() that is more test-friendly than using
+// ASSERT_DEATH. Relies on exception propagation.
+#ifdef NDEBUG
+#define testable_assert(cond)
+#else
+namespace ROCKSDB_NAMESPACE {
+// Intentionally not based on std::exception to reduce places where this
+// would be caught
+struct TestableAssertionFailure {};
+extern std::atomic<int> g_throw_on_testable_assertion_failure;
+}  // namespace ROCKSDB_NAMESPACE
+#define testable_assert(cond)                                          \
+  do {                                                                 \
+    if (ROCKSDB_NAMESPACE::g_throw_on_testable_assertion_failure.load( \
+            std::memory_order_relaxed) > 0) {                          \
+      if (cond) {                                                      \
+      } else                                                           \
+        throw ROCKSDB_NAMESPACE::TestableAssertionFailure();           \
+    } else {                                                           \
+      assert(cond);                                                    \
+    }                                                                  \
+  } while (0)
+#define ASSERT_TESTABLE_FAILURE(expr)                                   \
+  do {                                                                  \
+    ROCKSDB_NAMESPACE::g_throw_on_testable_assertion_failure.fetch_add( \
+        1, std::memory_order_relaxed);                                  \
+    ASSERT_THROW(expr, ROCKSDB_NAMESPACE::TestableAssertionFailure);    \
+    ROCKSDB_NAMESPACE::g_throw_on_testable_assertion_failure.fetch_sub( \
+        1, std::memory_order_relaxed);                                  \
+  } while (0)
+#endif

tools/db_crashtest.py

@@ -5,6 +5,7 @@ import argparse
 import math
 import os
 import random
+import shlex
 import shutil
 import subprocess
 import sys
@@ -22,6 +23,18 @@ def get_random_seed(override):
         return override
 
 
+def quote_arg_for_display(arg):
+    """
+    Quote only the value after '=' for shell display.
+    This makes the printed command safe to copy/paste into a Unix shell.
+    Note: shlex is Unix-focused; Non-Unix shell users may need to adjust quoting after copying.
+    """
+    if "=" not in arg:
+        return arg
+    flag, value = arg.split("=", 1)
+    return f"{flag}={shlex.quote(value)}"
+
+
 def setup_random_seed_before_main():
     parser = argparse.ArgumentParser()
     parser.add_argument(
@@ -679,6 +692,8 @@ ts_params = {
     # Below flag is randomly picked once and kept consistent in following runs.
     "persist_user_defined_timestamps": random.choice([0, 1, 1]),
     "use_merge": 0,
+    # Causing failures and not yet compatible
+    "use_multiscan": 0,
     "use_full_merge_v1": 0,
     "use_txn": 0,
     "ingest_external_file_one_in": 0,
@@ -1324,7 +1339,10 @@ def gen_cmd(params, unknown_params):
 
 def execute_cmd(cmd, timeout=None, timeout_pstack=False):
     child = subprocess.Popen(cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE)
-    print("Running db_stress with pid=%d: %s\n\n" % (child.pid, " ".join(cmd)))
+    print(
+        "Running db_stress with pid=%d: %s\n\n"
+        % (child.pid, " ".join(quote_arg_for_display(arg) for arg in cmd))
+    )
     pid = child.pid
 
     try:

tools/sst_dump_tool.cc

@@ -63,6 +63,9 @@ void print_help(bool to_stderr) {
     --decode_blob_index
       Decode blob indexes and print them in a human-readable format during scans.
 
+    --show_sequence_number_type
+      Show sequence number and value type when executing raw command
+
     --from=<user_key>
       Key to start reading from when executing check|scan
 
@@ -177,6 +180,7 @@ int SSTDumpTool::Run(int argc, char const* const* argv, Options options) {
   bool verify_checksum = false;
   bool output_hex = false;
   bool decode_blob_index = false;
+  bool show_sequence_number_type = false;
   bool input_key_hex = false;
   bool has_from = false;
   bool has_to = false;
@@ -235,6 +239,8 @@ int SSTDumpTool::Run(int argc, char const* const* argv, Options options) {
       output_hex = true;
     } else if (strcmp(argv[i], "--decode_blob_index") == 0) {
       decode_blob_index = true;
+    } else if (strcmp(argv[i], "--show_sequence_number_type") == 0) {
+      show_sequence_number_type = true;
     } else if (strcmp(argv[i], "--input_key_hex") == 0) {
       input_key_hex = true;
     } else if (sscanf(argv[i], "--read_num=%lu%c", (unsigned long*)&n, &junk) ==
@@ -531,7 +537,8 @@ int SSTDumpTool::Run(int argc, char const* const* argv, Options options) {
 
     ROCKSDB_NAMESPACE::SstFileDumper dumper(
         options, filename, Temperature::kUnknown, readahead_size,
-        verify_checksum, output_hex, decode_blob_index);
+        verify_checksum, output_hex, decode_blob_index, EnvOptions(), false,
+        show_sequence_number_type);
 
     // Not a valid SST
     if (!dumper.getStatus().ok()) {

util/bit_fields.h

@@ -6,8 +6,10 @@
 #pragma once
 
 #include <atomic>
+#include <vector>
 
 #include "rocksdb/rocksdb_namespace.h"
+#include "test_util/sync_point.h"
 #include "util/math.h"
 
 namespace ROCKSDB_NAMESPACE {
@@ -144,8 +146,44 @@ struct AndTransform {
   }
 };
 
-// TODO: AddTransfrom, which is more complicated due to possible overflow into
-// other fields etc.
+// Can represent a combination of both subtractions and additions, representing
+// subtractions as the addition of a negated value. To ensure we don't create a
+// net overflow or underflow between fields, in debug builds we track the
+// corresponding preconditions. (NOTE that when representing a subtraction, we
+// rely on overflow of the unsigned representation.)
+template <typename BitFieldsT>
+struct AddTransform {
+  using U = typename BitFieldsT::U;
+  U to_add = 0;
+#ifndef NDEBUG
+  struct Precondition {
+    U mask;   // for bits of the target field
+    U piece;  // component of to_add for the target field
+  };
+  std::vector<Precondition> preconditions;
+#endif  // NDEBUG
+  void AssertPreconditions([[maybe_unused]] U from) {
+#ifndef NDEBUG
+    for (auto p : preconditions) {
+      U tmp = (from & p.mask) + p.piece;
+      // Assert no under/overflow (unless the field is at the top bits of the
+      // representation in U, which is allowed because it doesn't lead to
+      // leakage into other fields)
+      testable_assert((tmp & ~p.mask) == 0);
+    }
+#endif  // NDEBUG
+  }
+  // + for general combine
+  AddTransform<BitFieldsT> operator+(AddTransform<BitFieldsT> other) const {
+    AddTransform<BitFieldsT> rv{to_add + other.to_add};
+#ifndef NDEBUG
+    rv.preconditions = preconditions;
+    rv.preconditions.insert(rv.preconditions.end(), other.preconditions.begin(),
+                            other.preconditions.end());
+#endif  // NDEBUG
+    return rv;
+  }
+};
 
 // Placeholder for PrevField for the first field
 struct NoPrevBitField {
@@ -204,6 +242,7 @@ struct UnsignedBitField {
   static_assert(kBitCount >= 1);
   static_assert(kBitCount <= 64);
   static_assert(kBitOffset >= 0 && kEndBit <= BitFieldsT::kBitCount);
+  static constexpr bool kIncludesTopBit = (kEndBit == BitFieldsT::kBitCount);
 
   static constexpr V kMask = (V{1} << (kBitCount - 1) << 1) - 1;
 
@@ -219,9 +258,59 @@ struct UnsignedBitField {
     bf.underlying |= static_cast<U>(value & kMask) << kBitOffset;
   }
 
+  // Create a transfor for clearing this field to zero.
   static AndTransform<BitFieldsT> ClearTransform() {
     return AndTransform<BitFieldsT>{~(static_cast<U>(kMask) << kBitOffset)};
   }
+
+  // Create a transform for adding a particular value, but with the precondition
+  // that adding the value will not overflow the field. This applies for fields
+  // that do not include the top bit of the underlying representation. Can be
+  // combined with other additive transforms for other fields.
+  static AddTransform<BitFieldsT> PlusTransformPromiseNoOverflow(V value) {
+    static_assert(!kIncludesTopBit);
+    AddTransform<BitFieldsT> rv{static_cast<U>(value) << kBitOffset};
+#ifndef NDEBUG
+    rv.preconditions.push_back(
+        {static_cast<U>(kMask) << kBitOffset, rv.to_add});
+#endif  // NDEBUG
+    return rv;
+  }
+
+  // Create a transform for adding a particular value, but ignoring any overflow
+  // in that field. This applies for fields that include the top bit of the
+  // underlying representation. Can be combined with other additive transforms
+  // for other fields.
+  static AddTransform<BitFieldsT> PlusTransformIgnoreOverflow(V value) {
+    static_assert(kIncludesTopBit);
+    AddTransform<BitFieldsT> rv{static_cast<U>(value) << kBitOffset};
+    return rv;
+  }
+
+  // Create a transform for subtracting a particular value, but with the
+  // precondition that subtracting the value will not underflow the field. This
+  // applies for fields that do not include the top bit of the underlying
+  // representation. Can be combined with other additive transforms for other
+  // fields.
+  static AddTransform<BitFieldsT> MinusTransformPromiseNoUnderflow(V value) {
+    static_assert(!kIncludesTopBit);
+    AddTransform<BitFieldsT> rv{U{0} - (static_cast<U>(value) << kBitOffset)};
+#ifndef NDEBUG
+    rv.preconditions.push_back(
+        {static_cast<U>(kMask) << kBitOffset, rv.to_add});
+#endif  // NDEBUG
+    return rv;
+  }
+
+  // Create a transform for subtracting a particular value, but ignoring any
+  // underflow in that field. This applies for fields that include the top bit
+  // of the underlying representation. Can be combined with other additive
+  // transforms for other fields.
+  static AddTransform<BitFieldsT> MinusTransformIgnoreUnderflow(V value) {
+    static_assert(kIncludesTopBit);
+    AddTransform<BitFieldsT> rv{U{0} - (static_cast<U>(value) << kBitOffset)};
+    return rv;
+  }
 };
 
 // A handy wrapper for a relaxed atomic on some BitFields type (unlike
@@ -260,7 +349,22 @@ class RelaxedBitFieldsAtomic {
   }
   void ApplyRelaxed(OrTransform<BitFieldsT> transform,
                     BitFieldsT* before = nullptr, BitFieldsT* after = nullptr) {
-    U before_val = v_.fetch_or(transform.to_or, std::memory_order_relaxed);
+    ApplyImpl<std::memory_order_relaxed>(transform, before, after);
+  }
+  void ApplyRelaxed(AndTransform<BitFieldsT> transform,
+                    BitFieldsT* before = nullptr, BitFieldsT* after = nullptr) {
+    ApplyImpl<std::memory_order_relaxed>(transform, before, after);
+  }
+  void ApplyRelaxed(AddTransform<BitFieldsT> transform,
+                    BitFieldsT* before = nullptr, BitFieldsT* after = nullptr) {
+    ApplyImpl<std::memory_order_relaxed>(transform, before, after);
+  }
+
+ protected:  // fns
+  template <std::memory_order kOrder>
+  void ApplyImpl(OrTransform<BitFieldsT> transform,
+                 BitFieldsT* before = nullptr, BitFieldsT* after = nullptr) {
+    U before_val = v_.fetch_or(transform.to_or, kOrder);
     if (before) {
       before->underlying = before_val;
     }
@@ -268,9 +372,10 @@ class RelaxedBitFieldsAtomic {
       after->underlying = before_val | transform.to_or;
     }
   }
-  void ApplyRelaxed(AndTransform<BitFieldsT> transform,
-                    BitFieldsT* before = nullptr, BitFieldsT* after = nullptr) {
-    U before_val = v_.fetch_and(transform.to_and, std::memory_order_relaxed);
+  template <std::memory_order kOrder>
+  void ApplyImpl(AndTransform<BitFieldsT> transform,
+                 BitFieldsT* before = nullptr, BitFieldsT* after = nullptr) {
+    U before_val = v_.fetch_and(transform.to_and, kOrder);
     if (before) {
       before->underlying = before_val;
     }
@@ -278,8 +383,20 @@ class RelaxedBitFieldsAtomic {
       after->underlying = before_val & transform.to_and;
     }
   }
+  template <std::memory_order kOrder>
+  void ApplyImpl(AddTransform<BitFieldsT> transform,
+                 BitFieldsT* before = nullptr, BitFieldsT* after = nullptr) {
+    U before_val = v_.fetch_add(transform.to_add, kOrder);
+    transform.AssertPreconditions(before_val);
+    if (before) {
+      before->underlying = before_val;
+    }
+    if (after) {
+      after->underlying = before_val + transform.to_add;
+    }
+  }
 
- protected:
+ protected:  // data
   std::atomic<U> v_;
 };
 
@@ -313,25 +430,18 @@ class AcqRelBitFieldsAtomic : public RelaxedBitFieldsAtomic<BitFieldsT> {
   }
   void Apply(OrTransform<BitFieldsT> transform, BitFieldsT* before = nullptr,
              BitFieldsT* after = nullptr) {
-    U before_val =
-        Base::v_.fetch_or(transform.to_or, std::memory_order_acq_rel);
-    if (before) {
-      before->underlying = before_val;
-    }
-    if (after) {
-      after->underlying = before_val | transform.to_or;
-    }
+    Base::template ApplyImpl<std::memory_order_acq_rel>(transform, before,
+                                                        after);
   }
   void Apply(AndTransform<BitFieldsT> transform, BitFieldsT* before = nullptr,
              BitFieldsT* after = nullptr) {
-    U before_val =
-        Base::v_.fetch_and(transform.to_and, std::memory_order_acq_rel);
-    if (before) {
-      before->underlying = before_val;
-    }
-    if (after) {
-      after->underlying = before_val & transform.to_and;
-    }
+    Base::template ApplyImpl<std::memory_order_acq_rel>(transform, before,
+                                                        after);
+  }
+  void Apply(AddTransform<BitFieldsT> transform, BitFieldsT* before = nullptr,
+             BitFieldsT* after = nullptr) {
+    Base::template ApplyImpl<std::memory_order_acq_rel>(transform, before,
+                                                        after);
   }
 };
 

util/slice_test.cc

@@ -527,6 +527,67 @@ TEST(BitFieldsTest, BitFields) {
     ASSERT_EQ(before.Get<Field2>(), false);
     ASSERT_EQ(before.Get<Field3>(), false);
     ASSERT_EQ(after, state);
+
+    ASSERT_EQ(state.Get<Field1>(), 45U);
+    ASSERT_EQ(after.Get<Field2>(), true);
+    ASSERT_EQ(after.Get<Field3>(), true);
+    ASSERT_EQ(state.Get<Field4>(), 3U);
+
+    auto transform3 = Field1::PlusTransformPromiseNoOverflow(10000U) +
+                      Field4::MinusTransformPromiseNoUnderflow(3U);
+    relaxed.ApplyRelaxed(transform3, &before, &after);
+    ASSERT_EQ(before, state);
+    ASSERT_NE(after, state);
+    ASSERT_EQ(after.Get<Field1>(), 10045U);
+    ASSERT_EQ(after.Get<Field4>(), 0U);
+
+    auto transform4 = Field1::MinusTransformPromiseNoUnderflow(999U) +
+                      Field4::PlusTransformPromiseNoOverflow(31U);
+    relaxed.ApplyRelaxed(transform4, &before, &after);
+    ASSERT_EQ(after.Get<Field1>(), 9046U);
+    ASSERT_EQ(after.Get<Field4>(), 31U);
+
+    // Unmodified
+    ASSERT_EQ(after.Get<Field2>(), true);
+    ASSERT_EQ(after.Get<Field3>(), true);
+
+    // Test overflow/underflow detection
+    relaxed.StoreRelaxed(MyState{}.With<Field1>(65535U));  // Field1 max value
+    ASSERT_TESTABLE_FAILURE(
+        relaxed.ApplyRelaxed(Field1::PlusTransformPromiseNoOverflow(1U)));
+    relaxed.StoreRelaxed(MyState{}.With<Field4>(31U));  // Field4 max value
+    ASSERT_TESTABLE_FAILURE(
+        relaxed.ApplyRelaxed(Field4::PlusTransformPromiseNoOverflow(1U)));
+    relaxed.StoreRelaxed(MyState{}.With<Field1>(0U));
+    ASSERT_TESTABLE_FAILURE(
+        relaxed.ApplyRelaxed(Field1::MinusTransformPromiseNoUnderflow(1U)));
+    relaxed.StoreRelaxed(MyState{}.With<Field4>(0U));
+    ASSERT_TESTABLE_FAILURE(
+        relaxed.ApplyRelaxed(Field4::MinusTransformPromiseNoUnderflow(1U)));
+    ASSERT_TESTABLE_FAILURE(relaxed.ApplyRelaxed(
+        Field4::MinusTransformPromiseNoUnderflow(64U)));  // Too big
+    ASSERT_TESTABLE_FAILURE(relaxed.ApplyRelaxed(
+        Field4::PlusTransformPromiseNoOverflow(64U)));  // Too big
+
+    // Including combinations
+    relaxed.StoreRelaxed(MyState{}.With<Field4>(31U));  // Field4 max value
+    relaxed.StoreRelaxed(MyState{}.With<Field1>(0U));
+    ASSERT_TESTABLE_FAILURE(
+        relaxed.ApplyRelaxed(Field4::PlusTransformPromiseNoOverflow(1U) +
+                             Field1::MinusTransformPromiseNoUnderflow(1U)));
+
+    // But a field at the limit of upper bits is allowed to over/underflow
+    using Field5 = UnsignedBitField<MyState, 9, Field4>;
+    relaxed.StoreRelaxed(MyState{}.With<Field5>(0));  // Field5 max value
+    relaxed.ApplyRelaxed(Field5::MinusTransformIgnoreUnderflow(1U), &before,
+                         &after);  // "Safe" underflow
+    ASSERT_EQ(after.Get<Field5>(), 511U);
+    relaxed.ApplyRelaxed(Field5::PlusTransformIgnoreOverflow(1U), &before,
+                         &after);  // "Safe" overflow
+    ASSERT_EQ(after.Get<Field5>(), 0U);
+    relaxed.ApplyRelaxed(Field5::PlusTransformIgnoreOverflow(2048U), &before,
+                         &after);  // "Safe" overflow
+    ASSERT_EQ(after.Get<Field5>(), 0U);
   }
   {
     AcqRelBitFieldsAtomic<MyState> acqrel{state};
@@ -555,6 +616,37 @@ TEST(BitFieldsTest, BitFields) {
     ASSERT_EQ(before.Get<Field2>(), false);
     ASSERT_EQ(before.Get<Field3>(), false);
     ASSERT_EQ(after, state);
+
+    ASSERT_EQ(state.Get<Field1>(), 45U);
+    ASSERT_EQ(after.Get<Field2>(), true);
+    ASSERT_EQ(after.Get<Field3>(), true);
+    ASSERT_EQ(state.Get<Field4>(), 3U);
+
+    auto transform3 = Field1::PlusTransformPromiseNoOverflow(10000U) +
+                      Field4::MinusTransformPromiseNoUnderflow(3U);
+    acqrel.Apply(transform3, &before, &after);
+    ASSERT_EQ(before, state);
+    ASSERT_NE(after, state);
+    ASSERT_EQ(after.Get<Field1>(), 10045U);
+    ASSERT_EQ(after.Get<Field4>(), 0U);
+
+    auto transform4 = Field1::MinusTransformPromiseNoUnderflow(999U) +
+                      Field4::PlusTransformPromiseNoOverflow(31U);
+    acqrel.Apply(transform4, &before, &after);
+    ASSERT_EQ(after.Get<Field1>(), 9046U);
+    ASSERT_EQ(after.Get<Field4>(), 31U);
+
+    // Unmodified
+    ASSERT_EQ(after.Get<Field2>(), true);
+    ASSERT_EQ(after.Get<Field3>(), true);
+
+    // Test overflow/underflow detection
+    acqrel.Store(MyState{}.With<Field1>(65535U));
+    ASSERT_TESTABLE_FAILURE(
+        acqrel.Apply(Field1::PlusTransformPromiseNoOverflow(1U)));
+    acqrel.Store(MyState{}.With<Field4>(0U));
+    ASSERT_TESTABLE_FAILURE(
+        acqrel.Apply(Field4::MinusTransformPromiseNoUnderflow(1U)));
   }
 }
 

utilities/backup/backup_engine.cc

@@ -615,6 +615,10 @@ class BackupEngineImpl {
                                       std::string* checksum_hex,
                                       const Temperature src_temperature) const;
 
+  // Helper method to check if backup should be stopped. Can be overridden
+  // via sync points for testing.
+  bool ShouldStopBackup() const;
+
   // Obtain db_id and db_session_id from the table properties of file_path
   Status GetFileDbIdentities(Env* src_env, const EnvOptions& src_env_options,
                              const std::string& file_path,
@@ -2353,6 +2357,10 @@ IOStatus BackupEngineImpl::CopyOrCreateFile(
     Temperature dst_temperature, uint64_t* bytes_toward_next_callback,
     uint64_t* size, std::string* checksum_hex) {
   assert(src.empty() != contents.empty());
+  if (ShouldStopBackup()) {
+    return status_to_io_status(Status::Incomplete("Backup stopped"));
+  }
+
   IOStatus io_s;
   std::unique_ptr<FSWritableFile> dst_file;
   std::unique_ptr<FSSequentialFile> src_file;
@@ -2413,7 +2421,7 @@ IOStatus BackupEngineImpl::CopyOrCreateFile(
   Slice data;
   const IOOptions opts;
   do {
-    if (stop_backup_.load(std::memory_order_acquire)) {
+    if (ShouldStopBackup()) {
       return status_to_io_status(Status::Incomplete("Backup stopped"));
     }
     if (!src.empty()) {
@@ -2749,6 +2757,12 @@ IOStatus BackupEngineImpl::AddBackupFileWorkItem(
   return IOStatus::OK();
 }
 
+bool BackupEngineImpl::ShouldStopBackup() const {
+  bool should_stop = stop_backup_.load(std::memory_order_acquire);
+  TEST_SYNC_POINT_CALLBACK("BackupEngineImpl::ShouldStopBackup", &should_stop);
+  return should_stop;
+}
+
 IOStatus BackupEngineImpl::ReadFileAndComputeChecksum(
     const std::string& src, const std::shared_ptr<FileSystem>& src_fs,
     const EnvOptions& src_env_options, uint64_t size_limit,
@@ -2756,6 +2770,9 @@ IOStatus BackupEngineImpl::ReadFileAndComputeChecksum(
   if (checksum_hex == nullptr) {
     return status_to_io_status(Status::Aborted("Checksum pointer is null"));
   }
+  if (ShouldStopBackup()) {
+    return status_to_io_status(Status::Incomplete("Backup stopped"));
+  }
   uint32_t checksum_value = 0;
   if (size_limit == 0) {
     size_limit = std::numeric_limits<uint64_t>::max();
@@ -2783,7 +2800,7 @@ IOStatus BackupEngineImpl::ReadFileAndComputeChecksum(
   Slice data;
 
   do {
-    if (stop_backup_.load(std::memory_order_acquire)) {
+    if (ShouldStopBackup()) {
       return status_to_io_status(Status::Incomplete("Backup stopped"));
     }
     size_t buffer_to_read =

utilities/backup/backup_engine_test.cc

@@ -43,6 +43,7 @@
 #include "test_util/sync_point.h"
 #include "test_util/testharness.h"
 #include "test_util/testutil.h"
+#include "util/atomic.h"
 #include "util/cast_util.h"
 #include "util/mutexlock.h"
 #include "util/random.h"
@@ -4790,6 +4791,79 @@ TEST_F(BackupEngineTest, IOBufferSize) {
   ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks();
 }
 
+// Test stopping backup at different points in the backup lifecycle
+// Uses randomized stop points with geometric distribution to better catch
+// edge cases across multiple iterations.
+TEST_F(BackupEngineTest, StopBackupAtDifferentStages) {
+  const int keys_iteration = 5000;
+  const int num_iterations = 10;
+
+  // Enable multi-threaded backup
+  engine_options_->max_background_operations = 7;
+
+  // Generate DB once and reuse across iterations
+  OpenDBAndBackupEngine(true);
+  FillDB(db_.get(), 0, keys_iteration);
+
+  Random rnd(301);
+
+  for (int iteration = 0; iteration < num_iterations; iteration++) {
+    // Generate stop threshold using skewed distribution
+    // Smaller numbers are more likely, which is more interesting for testing
+    // Range: [0, 2^7-1] = [0, 127] with exponential bias towards 0
+    int stop_after_calls = rnd.Skewed(7);
+
+    RelaxedAtomic<int> call_count{0};
+    ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+        "BackupEngineImpl::ShouldStopBackup", [&](void* arg) {
+          call_count.FetchAddRelaxed(1);
+          if (call_count.LoadRelaxed() > stop_after_calls) {
+            bool* should_stop = static_cast<bool*>(arg);
+            *should_stop = true;
+          }
+        });
+    ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
+
+    // Create backup - it may complete successfully or be stopped
+    IOStatus s = backup_engine_->CreateNewBackup(db_.get());
+
+    // Verify that ShouldStopBackup was called
+    ASSERT_GT(call_count.LoadRelaxed(), 0);
+
+    if (s.IsIncomplete()) {
+      // Backup was stopped - verify it's the expected error
+      ASSERT_TRUE(s.ToString().find("Backup stopped") != std::string::npos)
+          << "Unexpected incomplete status for threshold " << stop_after_calls
+          << ": " << s.ToString();
+      ASSERT_GT(call_count.LoadRelaxed(), stop_after_calls)
+          << "Expected call_count > stop_after_calls";
+
+      // Verify that no valid backup was created
+      std::vector<BackupInfo> backup_info;
+      backup_engine_->GetBackupInfo(&backup_info);
+      ASSERT_EQ(0, backup_info.size());
+    } else {
+      // Backup completed successfully before reaching the stop threshold
+      ASSERT_OK(s) << "Unexpected error for threshold " << stop_after_calls;
+      ASSERT_LE(call_count.LoadRelaxed(), stop_after_calls)
+          << "Backup completed but call_count exceeded threshold";
+
+      // Verify a backup was created
+      std::vector<BackupInfo> backup_info;
+      backup_engine_->GetBackupInfo(&backup_info);
+      ASSERT_EQ(1, backup_info.size());
+
+      // Clean up the successful backup for next iteration
+      ASSERT_OK(backup_engine_->DeleteBackup(backup_info[0].backup_id));
+    }
+
+    ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
+    ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks();
+  }
+
+  CloseDBAndBackupEngine();
+}
+
 }  // namespace
 
 }  // namespace ROCKSDB_NAMESPACE