Update version and HISTORY.md for 11.2.4 release

Summary:
Pull Request resolved: https://github.com/facebook/rocksdb/pull/14881

Read-only DB instances can share a directory with a live writer, so their view of live files is only a snapshot from the time they opened. After read-only DB open began setting `opened_successfully_` for its intended open-success semantics, that state had an unintended side effect: the common close path treated a successful read-only open like a read-write open and ran obsolete-file cleanup. If the writer created or made files live after the read-only handle opened, that cleanup could use the stale read-only live set and delete files still needed by the writer.

This change records whether a `DBImpl` was opened read-only and skips close-time obsolete-file cleanup for read-only DBs instead of simply keeping `opened_successfully_=false`, which could be confusing and easily mistaken in the future again.

For completeness I also updated other callsites of `opened_successfully_`, but those should not be real bugs as ReadOnly DBs do not run flushes/compactions or write to WALs.

Reviewed By: xingbowang

Differential Revision: D109622445

fbshipit-source-id: d7be4b20fce86ccb218a63ac6f5b707b316aac79

HISTORY.md

@@ -1,6 +1,40 @@
 # Rocksdb Change Log
 > NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
 
+## 11.2.4 (06/24/2026)
+### Bug Fixes
+* Fixed a bug where closing a read-only DB instance could delete live SST files created by a concurrent read-write DB sharing the same directory.
+
+## 11.2.3 (05/07/2026)
+### New Features
+* Add experimental mutable DBOption `optimize_manifest_for_recovery` (default false). When enabled, RocksDB can reduce recovery work after a clean shutdown, which may lower DB::Open latency on warm reopens.
+* Add experimental `reuse_manifest_on_open` DBOption (default false). When enabled, DB::Open reuses the existing MANIFEST file for append instead of creating a fresh one, avoiding the cost of serializing the entire database state into a new MANIFEST on the first post-open write.
+
+## 11.2.2 (05/07/2026)
+## 11.2.1 (04/29/2026)
+### Bug Fixes
+* Fix bug in range tombstone synthesis that covers live keys added during an IngestExternalFile
+
+## 11.2.0 (04/18/2026)
+### New Features
+* Added experimental `DBOptions::fast_sst_open` option. When enabled, RocksDB retrieves opaque file system metadata for SST files after flush, compaction, and external file ingestion, persists it in the MANIFEST, and passes it back to the file system on subsequent file opens to accelerate DB open time.
+* Added new option `min_tombstones_for_range_conversion` in `AdvancedColumnFamilyOptions`. When set to a non-zero value N, forward or reverse iteration will convert N or more contiguous point tombstones into a range tombstone in the mutable memtable. Future read operations will then be able to benefit from range tombstone optimizations. There are some limitations when it comes to table_filters, prefix_filters, and UDTs. See header comments for more details.
+* Added read-triggered compaction: a new column family option `read_triggered_compaction_threshold` (default 0, disabled) that marks SST files for compaction when their read frequency (`num_collapsible_entry_reads_sampled / file_size`) exceeds the threshold. This helps reduce read amplification for frequently-read ("hot") keys. A new DB option `max_compaction_trigger_wakeup_seconds` (default 43200s / 12 hours) controls the maximum interval for periodic compaction score re-evaluation, which is necessary for this feature to work on quiet (no-write) databases.
+
+### Public API Changes
+* Added new `WideColumnBlobResolver` interface and `CompactionFilter::FilterV4()` method, including `ResolveColumn()` / `ResolveColumns()` helpers for lazy loading blob column values in wide-column entities during compaction. This allows compaction filters to resolve blob values on-demand, avoiding unnecessary I/O for blob columns they don't need to access.
+* Changed experimental feature `ExternalTableReader::Get` and `ExternalTableReader::MultiGet` to use `PinnableSlice` instead of `std::string` for output values, enabling zero-copy pinning. This will break existing implementations.
+* Added `SstFileReader::Get` and `SstFileReader::MultiGet` overloads that accept `PinnableSlice`/`std::vector<PinnableSlice>*`, enabling zero-copy reads when the underlying `TableReader` supports pinning.
+
+### Behavior Changes
+* Prefix filter changes - when seeking to a key that is out of domain, and total_order_seek is false, total_order_seek is treated as if it were true. When prefix_same_as_start = true, now iterating past a key that is out of domain invalidates the iterator. The existing behavior does not check for InDomain in the DBIter, so Transform() can produce undefined behavior (e.g. key of size 3 on FixedPrefixTransform(4)).
+* Wide-column entities with blob-backed columns now use a new V2 on-disk encoding; older RocksDB versions that do not support wide-column blob separation will reject DBs or SSTs containing those entities.
+
+### Bug Fixes
+* Fix blob garbage accounting for blob direct-write flushes so flush-time filtering and overwrite elision correctly register obsolete blob bytes in blob metadata.
+* Fix a memory accounting leak in IODispatcher where ReadIndex() moved block values out of ReadSet without releasing the associated prefetch memory, causing subsequent prefetches to be blocked when max_prefetch_memory_bytes was set.
+* Fix MultiScan to fall back to synchronous coalesced reads when async I/O is unsupported at runtime.
+
 ## 11.1.0 (03/23/2026)
 ### New Features
 * Add a new option `open_files_async`. The existing behavior is on DB open, we open all sst files and do basic validations. For very large DBs on remote filesystems with many ssts, this may take very long. This option performs these validations instead in the background. Open errors found by this async background task are surfaced as a new background error kAsyncFileOpen.

db/column_family.cc

@@ -612,7 +612,7 @@ ColumnFamilyData::ColumnFamilyData(
     const FileOptions* file_options, ColumnFamilySet* column_family_set,
     BlockCacheTracer* const block_cache_tracer,
     const std::shared_ptr<IOTracer>& io_tracer, const std::string& db_id,
-    const std::string& db_session_id, bool read_only)
+    const std::string& db_session_id, bool read_only, bool fast_sst_open)
     : id_(id),
       name_(name),
       dummy_versions_(_dummy_versions),
@@ -672,7 +672,7 @@ ColumnFamilyData::ColumnFamilyData(
         new InternalStats(ioptions_.num_levels, ioptions_.clock, this));
     table_cache_.reset(new TableCache(ioptions_, file_options, _table_cache,
                                       block_cache_tracer, io_tracer,
-                                      db_session_id));
+                                      db_session_id, fast_sst_open));
     blob_file_cache_.reset(
         new BlobFileCache(_table_cache, &ioptions(), soptions(), id_,
                           internal_stats_->GetBlobFileReadHist(), io_tracer));
@@ -1809,16 +1809,14 @@ void ColumnFamilyData::RecoverEpochNumbers() {
   vstorage->RecoverEpochNumbers(this);
 }
 
-ColumnFamilySet::ColumnFamilySet(const std::string& dbname,
-                                 const ImmutableDBOptions* db_options,
-                                 const FileOptions& file_options,
-                                 Cache* table_cache,
-                                 WriteBufferManager* _write_buffer_manager,
-                                 WriteController* _write_controller,
-                                 BlockCacheTracer* const block_cache_tracer,
-                                 const std::shared_ptr<IOTracer>& io_tracer,
-                                 const std::string& db_id,
-                                 const std::string& db_session_id)
+ColumnFamilySet::ColumnFamilySet(
+    const std::string& dbname, const ImmutableDBOptions* db_options,
+    const FileOptions& file_options, Cache* table_cache,
+    WriteBufferManager* _write_buffer_manager,
+    WriteController* _write_controller,
+    BlockCacheTracer* const block_cache_tracer,
+    const std::shared_ptr<IOTracer>& io_tracer, const std::string& db_id,
+    const std::string& db_session_id, bool fast_sst_open)
     : max_column_family_(0),
       file_options_(file_options),
       dummy_cfd_(new ColumnFamilyData(
@@ -1835,7 +1833,8 @@ ColumnFamilySet::ColumnFamilySet(const std::string& dbname,
       block_cache_tracer_(block_cache_tracer),
       io_tracer_(io_tracer),
       db_id_(db_id),
-      db_session_id_(db_session_id) {
+      db_session_id_(db_session_id),
+      fast_sst_open_(fast_sst_open) {
   // initialize linked list
   dummy_cfd_->prev_ = dummy_cfd_;
   dummy_cfd_->next_ = dummy_cfd_;
@@ -1902,7 +1901,7 @@ ColumnFamilyData* ColumnFamilySet::CreateColumnFamily(
   ColumnFamilyData* new_cfd = new ColumnFamilyData(
       id, name, dummy_versions, table_cache_, write_buffer_manager_, options,
       *db_options_, &file_options_, this, block_cache_tracer_, io_tracer_,
-      db_id_, db_session_id_, read_only);
+      db_id_, db_session_id_, read_only, fast_sst_open_);
   column_families_.insert({name, id});
   column_family_data_.insert({id, new_cfd});
   auto ucmp = new_cfd->user_comparator();

db/column_family.h

@@ -619,16 +619,23 @@ class ColumnFamilyData {
     return ioptions_.cf_allow_ingest_behind || ioptions_.allow_ingest_behind;
   }
 
+  // Per-CF reader-writer lock that serializes IngestExternalFiles with range
+  // tombstone conversion.
+  port::RWMutex& GetIngestSstLock() { return ingest_sst_lock_; }
+
  private:
   friend class ColumnFamilySet;
-  ColumnFamilyData(
-      uint32_t id, const std::string& name, Version* dummy_versions,
-      Cache* table_cache, WriteBufferManager* write_buffer_manager,
-      const ColumnFamilyOptions& options, const ImmutableDBOptions& db_options,
-      const FileOptions* file_options, ColumnFamilySet* column_family_set,
-      BlockCacheTracer* const block_cache_tracer,
-      const std::shared_ptr<IOTracer>& io_tracer, const std::string& db_id,
-      const std::string& db_session_id, bool read_only);
+  ColumnFamilyData(uint32_t id, const std::string& name,
+                   Version* dummy_versions, Cache* table_cache,
+                   WriteBufferManager* write_buffer_manager,
+                   const ColumnFamilyOptions& options,
+                   const ImmutableDBOptions& db_options,
+                   const FileOptions* file_options,
+                   ColumnFamilySet* column_family_set,
+                   BlockCacheTracer* const block_cache_tracer,
+                   const std::shared_ptr<IOTracer>& io_tracer,
+                   const std::string& db_id, const std::string& db_session_id,
+                   bool read_only, bool fast_sst_open = false);
 
   std::vector<std::string> GetDbPaths() const;
 
@@ -729,6 +736,10 @@ class ColumnFamilyData {
   bool mempurge_used_;
 
   std::atomic<uint64_t> next_epoch_number_;
+
+  // Used to synchronize IngestExternalFile with range tombstone conversion. See
+  // also Memtable::ingest_seqno_barrier_.
+  port::RWMutex ingest_sst_lock_;
 };
 
 // ColumnFamilySet has interesting thread-safety requirements
@@ -774,7 +785,8 @@ class ColumnFamilySet {
                   WriteController* _write_controller,
                   BlockCacheTracer* const block_cache_tracer,
                   const std::shared_ptr<IOTracer>& io_tracer,
-                  const std::string& db_id, const std::string& db_session_id);
+                  const std::string& db_id, const std::string& db_session_id,
+                  bool fast_sst_open = false);
   ~ColumnFamilySet();
 
   ColumnFamilyData* GetDefault() const;
@@ -810,6 +822,9 @@ class ColumnFamilySet {
 
   Cache* get_table_cache() { return table_cache_; }
 
+  bool GetFastSstOpen() const { return fast_sst_open_; }
+  void SetFastSstOpen(bool v) { fast_sst_open_ = v; }
+
   WriteBufferManager* write_buffer_manager() { return write_buffer_manager_; }
 
   WriteController* write_controller() { return write_controller_; }
@@ -858,6 +873,7 @@ class ColumnFamilySet {
   std::shared_ptr<IOTracer> io_tracer_;
   const std::string& db_id_;
   std::string db_session_id_;
+  bool fast_sst_open_;
 };
 
 // A wrapper for ColumnFamilySet that supports releasing DB mutex during each

db/compaction/compaction_service_job.cc

@@ -846,11 +846,16 @@ static std::unordered_map<std::string, OptionTypeInfo>
          {offsetof(struct InternalStats::CompactionStats, count),
           OptionType::kUInt64T, OptionVerificationType::kNormal,
           OptionTypeFlags::kNone}},
-        {"counts", OptionTypeInfo::Array<
-                       int, static_cast<int>(CompactionReason::kNumOfReasons)>(
-                       offsetof(struct InternalStats::CompactionStats, counts),
-                       OptionVerificationType::kNormal, OptionTypeFlags::kNone,
-                       {0, OptionType::kInt})},
+        {"counts",
+         OptionTypeInfo::Array<
+             /* In release 11.2, a new compaction reason was added. This broken
+              * the reader and writer. To unblock release 11.1, we temporarily
+              * reduce the count array size to the old one. TODO add a proper
+              * serialization and deserialization method. */
+             int, static_cast<int>(CompactionReason::kNumOfReasons) - 1>(
+             offsetof(struct InternalStats::CompactionStats, counts),
+             OptionVerificationType::kNormal, OptionTypeFlags::kNone,
+             {0, OptionType::kInt})},
 };
 
 static std::unordered_map<std::string, OptionTypeInfo>

db/db_basic_test.cc

@@ -102,6 +102,757 @@ TEST_F(DBBasicTest, OpenWhenOpen) {
   ASSERT_TRUE(strstr(s.getState(), "lock ") != nullptr);
 }
 
+namespace {
+// Helper that captures per-branch SkippedNoopEdit counts for tests.
+struct RecoveryOptimizationCounters {
+  std::atomic<int> setup_dbid{0};
+  std::atomic<int> per_cf{0};
+  std::atomic<int> wal_deletion{0};
+  std::atomic<int> next_file_number{0};
+
+  void Install() {
+    auto* sp = ROCKSDB_NAMESPACE::SyncPoint::GetInstance();
+    sp->SetCallBack("DBImpl::Recovery:SkippedNoopEdit:SetupDBId",
+                    [this](void*) { setup_dbid.fetch_add(1); });
+    sp->SetCallBack("DBImpl::Recovery:SkippedNoopEdit:PerCF",
+                    [this](void*) { per_cf.fetch_add(1); });
+    sp->SetCallBack("DBImpl::Recovery:SkippedNoopEdit:WalDeletion",
+                    [this](void*) { wal_deletion.fetch_add(1); });
+    sp->SetCallBack("DBImpl::Recovery:SkippedNoopEdit:NextFileNumber",
+                    [this](void*) { next_file_number.fetch_add(1); });
+    sp->EnableProcessing();
+  }
+  void Uninstall() {
+    auto* sp = ROCKSDB_NAMESPACE::SyncPoint::GetInstance();
+    sp->DisableProcessing();
+    sp->ClearAllCallBacks();
+  }
+};
+}  // namespace
+
+// optimize_manifest_for_recovery=true: a clean reopen of a flushed DB must
+// append fewer individual records to the MANIFEST than the default-off path.
+// Verified by counting AddRecord calls (one per VersionEdit written to the
+// MANIFEST log).
+TEST_F(DBBasicTest,
+       OptimizeManifestForRecoveryReducesManifestWritesOnCleanReopen) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  auto* sp = ROCKSDB_NAMESPACE::SyncPoint::GetInstance();
+
+  // Measure MANIFEST records with optimize_manifest_for_recovery OFF (default).
+  DestroyAndReopen(options);
+  ASSERT_OK(Put("k1", "v1"));
+  ASSERT_OK(Flush());
+  Close();
+
+  std::atomic<int> records_off{0};
+  std::atomic<int> next_file_skips_off{0};
+  sp->SetCallBack("VersionSet::ProcessManifestWrites:AddRecord",
+                  [&](void*) { records_off.fetch_add(1); });
+  sp->SetCallBack("DBImpl::Recovery:SkippedNoopEdit:NextFileNumber",
+                  [&](void*) { next_file_skips_off.fetch_add(1); });
+  sp->EnableProcessing();
+  Reopen(options);
+  sp->DisableProcessing();
+  sp->ClearAllCallBacks();
+  ASSERT_EQ("v1", Get("k1"));
+  int off_count = records_off.load();
+  ASSERT_GT(off_count, 0);
+  ASSERT_EQ(0, next_file_skips_off.load());
+  Close();
+
+  // Measure MANIFEST records with optimize_manifest_for_recovery ON.
+  options.optimize_manifest_for_recovery = true;
+  DestroyAndReopen(options);
+  ASSERT_OK(Put("k2", "v2"));
+  ASSERT_OK(Flush());
+  Close();
+
+  std::atomic<int> records_on{0};
+  std::atomic<int> next_file_skips_on{0};
+  sp->SetCallBack("VersionSet::ProcessManifestWrites:AddRecord",
+                  [&](void*) { records_on.fetch_add(1); });
+  sp->SetCallBack("DBImpl::Recovery:SkippedNoopEdit:NextFileNumber",
+                  [&](void*) { next_file_skips_on.fetch_add(1); });
+  sp->EnableProcessing();
+  Reopen(options);
+  sp->DisableProcessing();
+  sp->ClearAllCallBacks();
+  ASSERT_EQ("v2", Get("k2"));
+  int on_count = records_on.load();
+  ASSERT_GT(next_file_skips_on.load(), 0);
+  ASSERT_LT(on_count, off_count);
+}
+
+// When the per-CF log_number actually advances, the per-CF skip must NOT
+// fire — the edit carries real information and must be emitted.
+TEST_F(DBBasicTest, OptimizeManifestForRecoveryEmitsPerCFWhenLogAdvances) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.optimize_manifest_for_recovery = true;
+  DestroyAndReopen(options);
+  ASSERT_OK(Put("k", "v"));
+  // Write data and close WITHOUT flushing — the WAL has un-replayed
+  // records, so on reopen recovery flushes the memtable and the per-CF
+  // edit's log_number must advance past the prior WAL.
+  Close();
+
+  RecoveryOptimizationCounters counters;
+  counters.Install();
+  Reopen(options);
+  counters.Uninstall();
+
+  ASSERT_EQ(0, counters.per_cf.load());
+  ASSERT_EQ("v", Get("k"));
+}
+
+// With track_and_verify_wals_in_manifest=true, the wal_deletion edit
+// must STILL be emitted (the DeleteWalsBefore record is required).
+TEST_F(DBBasicTest, OptimizeManifestForRecoveryPreservesWalTracking) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.optimize_manifest_for_recovery = true;
+  options.track_and_verify_wals_in_manifest = true;
+  DestroyAndReopen(options);
+  ASSERT_OK(Put("k", "v"));
+  ASSERT_OK(Flush());
+  Close();
+
+  RecoveryOptimizationCounters counters;
+  counters.Install();
+  Reopen(options);
+  counters.Uninstall();
+
+  ASSERT_EQ(0, counters.wal_deletion.load());
+}
+
+// best_efforts_recovery requires a fresh MANIFEST + CURRENT to be
+// produced on every open (the salvage contract). Even with the option
+// on, none of the SkippedNoopEdit branches must fire under
+// best_efforts_recovery — otherwise CURRENT can be left missing or
+// stale (regression caught by DBBasicTest.RecoverWithNoCurrentFile and
+// DBTest2.BestEffortsRecoveryWithSstUniqueIdVerification under an
+// option-on default).
+TEST_F(DBBasicTest, OptimizeManifestForRecoveryDisabledByBestEffortsRecovery) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.optimize_manifest_for_recovery = true;
+  options.best_efforts_recovery = true;
+  DestroyAndReopen(options);
+  ASSERT_OK(Put("k", "v"));
+  ASSERT_OK(Flush());
+  Close();
+
+  RecoveryOptimizationCounters counters;
+  counters.Install();
+  Reopen(options);
+  counters.Uninstall();
+
+  ASSERT_EQ(0, counters.setup_dbid.load());
+  ASSERT_EQ(0, counters.per_cf.load());
+  ASSERT_EQ(0, counters.wal_deletion.load());
+  ASSERT_EQ(0, counters.next_file_number.load());
+  ASSERT_EQ("v", Get("k"));
+}
+
+// Multi-column-family coverage: with one CF flushed and another not,
+// the un-flushed CF's edit must still be emitted.
+TEST_F(DBBasicTest, OptimizeManifestForRecoveryMultiCF) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.optimize_manifest_for_recovery = true;
+  CreateAndReopenWithCF({"pikachu", "raichu"}, options);
+  ASSERT_OK(Put(0, "k", "v0"));
+  ASSERT_OK(Put(1, "k", "v1"));
+  ASSERT_OK(Put(2, "k", "v2"));
+  ASSERT_OK(Flush(0));
+  // CF 1 and 2 keep dirty memtables — recovery must emit their per-CF
+  // edits (log_number advance from WAL replay).
+  Close();
+
+  RecoveryOptimizationCounters counters;
+  counters.Install();
+  ReopenWithColumnFamilies({"default", "pikachu", "raichu"}, options);
+  counters.Uninstall();
+
+  ASSERT_EQ("v0", Get(0, "k"));
+  ASSERT_EQ("v1", Get(1, "k"));
+  ASSERT_EQ("v2", Get(2, "k"));
+}
+
+// MaybeUpdateNextFileNumber's seed value (next_file_number_) makes the
+// post-loop comparison trivially true on every clean recovery, causing a
+// no-op SetNextFile edit to be appended to the MANIFEST. With
+// optimize_manifest_for_recovery=true, that emission is gated and
+// the SkippedNoopEdit:NextFileNumber sync point fires in its place.
+TEST_F(DBBasicTest, OptimizeManifestForRecoverySkipsNextFileNumber) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.optimize_manifest_for_recovery = true;
+  DestroyAndReopen(options);
+  ASSERT_OK(Put("k", "v"));
+  ASSERT_OK(Flush());
+  Close();
+
+  RecoveryOptimizationCounters counters;
+  counters.Install();
+  Reopen(options);
+  counters.Uninstall();
+
+  ASSERT_EQ(1, counters.next_file_number.load());
+}
+
+// With option=true and a synthetic on-disk file whose number is at or
+// above next_file_number_, MaybeUpdateNextFileNumber MUST emit the
+// SetNextFile edit and advance the counter past the synthetic number.
+TEST_F(DBBasicTest,
+       OptimizeManifestForRecoveryEmitsNextFileNumberWhenJustified) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.optimize_manifest_for_recovery = true;
+  DestroyAndReopen(options);
+  ASSERT_OK(Put("k", "v"));
+  ASSERT_OK(Flush());
+
+  const uint64_t next_before_close =
+      dbfull()->GetVersionSet()->current_next_file_number();
+  Close();
+
+  // Drop an empty .sst with a number well above next_file_number into
+  // the DB dir. MaybeUpdateNextFileNumber must observe it and advance.
+  const uint64_t synthetic_number = next_before_close + 100;
+  const std::string synthetic_sst =
+      dbname_ + "/" + MakeTableFileName("", synthetic_number);
+  ASSERT_OK(WriteStringToFile(env_, "" /*data*/, synthetic_sst,
+                              /*should_sync=*/true));
+
+  RecoveryOptimizationCounters counters;
+  counters.Install();
+  Reopen(options);
+  counters.Uninstall();
+
+  ASSERT_EQ(0, counters.next_file_number.load());
+  ASSERT_GT(dbfull()->GetVersionSet()->current_next_file_number(),
+            synthetic_number);
+}
+
+// optimize_manifest_for_recovery=true: after a clean Put + Flush + Close, the
+// next Open's min_log_number_to_keep equals (max_wal_before_close + 1),
+// proving the close-time MANIFEST write took effect.
+TEST_F(DBBasicTest, OptimizeManifestForRecoveryAdvancesWalMarkersOnClose) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.optimize_manifest_for_recovery = true;
+  DestroyAndReopen(options);
+  ASSERT_OK(Put("k", "v"));
+  ASSERT_OK(Flush());
+
+  uint64_t max_wal_before_close = 0;
+  auto* sp = ROCKSDB_NAMESPACE::SyncPoint::GetInstance();
+  sp->SetCallBack("DBImpl::CloseHelper:CapturedMaxWal", [&](void* arg) {
+    max_wal_before_close = *static_cast<uint64_t*>(arg);
+  });
+  sp->EnableProcessing();
+  Close();
+  sp->DisableProcessing();
+  sp->ClearAllCallBacks();
+  ASSERT_GT(max_wal_before_close, 0u);
+
+  Reopen(options);
+  ASSERT_EQ(max_wal_before_close + 1,
+            dbfull()->GetVersionSet()->min_log_number_to_keep());
+  ASSERT_EQ("v", Get("k"));
+}
+
+// Shared-option matrix: default-off and disable-before-close should both fall
+// back to recovery-time MANIFEST work, while enable-through-close should avoid
+// MANIFEST appends on a clean reopen.
+TEST_F(DBBasicTest, OptimizeManifestForRecoveryCleanReopenMatrix) {
+  struct TestCase {
+    const char* name;
+    bool enable_on_open;
+    bool disable_before_close;
+    bool expect_close_write;
+  };
+  const TestCase test_cases[] = {
+      {"default_off", false, false, false},
+      {"enabled", true, false, true},
+      {"disabled_before_close", true, true, false},
+  };
+
+  auto* sp = ROCKSDB_NAMESPACE::SyncPoint::GetInstance();
+  for (const auto& test_case : test_cases) {
+    SCOPED_TRACE(test_case.name);
+    Options options = CurrentOptions();
+    options.create_if_missing = true;
+    options.optimize_manifest_for_recovery = test_case.enable_on_open;
+    DestroyAndReopen(options);
+    ASSERT_OK(Put("k", test_case.name));
+    ASSERT_OK(Flush());
+
+    if (test_case.disable_before_close) {
+      ASSERT_OK(dbfull()->SetDBOptions(
+          {{"optimize_manifest_for_recovery", "false"}}));
+    }
+
+    std::atomic<int> entered{0};
+    sp->SetCallBack("DBImpl::CloseHelper:WriteWalMarkersOnCloseEntered",
+                    [&](void*) { entered.fetch_add(1); });
+    sp->EnableProcessing();
+    Close();
+    sp->DisableProcessing();
+    sp->ClearAllCallBacks();
+
+    std::atomic<int> records{0};
+    RecoveryOptimizationCounters counters;
+    counters.Install();
+    sp->SetCallBack("VersionSet::ProcessManifestWrites:AddRecord",
+                    [&](void*) { records.fetch_add(1); });
+    Reopen(options);
+    counters.Uninstall();
+
+    if (test_case.expect_close_write) {
+      ASSERT_EQ(1, entered.load());
+      ASSERT_EQ(0, records.load());
+      ASSERT_GT(counters.per_cf.load(), 0);
+    } else {
+      ASSERT_EQ(0, entered.load());
+      ASSERT_GT(records.load(), 0);
+    }
+    ASSERT_EQ(test_case.name, Get("k"));
+    Close();
+  }
+}
+
+// allow_2pc=true: even with the option on, the close-time write must NOT
+// advance MinLogNumberToKeep / DeleteWalsBefore (2pc requires the WAL to
+// remain replayable for uncommitted prepared transactions).
+TEST_F(DBBasicTest, OptimizeManifestForRecoveryRespectsTwoPC) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.optimize_manifest_for_recovery = true;
+  options.allow_2pc = true;
+  DestroyAndReopen(options);
+  ASSERT_OK(Put("k", "v"));
+  ASSERT_OK(Flush());
+
+  const uint64_t min_log_before_close =
+      dbfull()->GetVersionSet()->min_log_number_to_keep();
+  Close();
+  Reopen(options);
+  ASSERT_EQ(min_log_before_close,
+            dbfull()->GetVersionSet()->min_log_number_to_keep());
+  ASSERT_EQ("v", Get("k"));
+}
+
+// Mixed-CF emptiness: on reopen, recovery should still have MANIFEST work to
+// do for the dirty CF while skipping per-CF recovery edits for the empty CFs
+// whose markers were persisted at close.
+TEST_F(DBBasicTest, OptimizeManifestForRecoverySkipsNonEmptyCF) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.optimize_manifest_for_recovery = true;
+  CreateAndReopenWithCF({"empty_cf", "dirty_cf"}, options);
+  ASSERT_OK(Put(1, "k", "v1"));
+  ASSERT_OK(Flush(1));
+  ASSERT_OK(Put(2, "k", "v2"));
+  Close();
+
+  RecoveryOptimizationCounters counters;
+  std::atomic<int> records{0};
+  auto* sp = ROCKSDB_NAMESPACE::SyncPoint::GetInstance();
+  counters.Install();
+  sp->SetCallBack("VersionSet::ProcessManifestWrites:AddRecord",
+                  [&](void*) { records.fetch_add(1); });
+  ReopenWithColumnFamilies({"default", "empty_cf", "dirty_cf"}, options);
+  counters.Uninstall();
+
+  ASSERT_GT(counters.per_cf.load(), 0);
+  ASSERT_GT(records.load(), 0);
+  ASSERT_EQ("v1", Get(1, "k"));
+  ASSERT_EQ("v2", Get(2, "k"));
+}
+
+// track_and_verify_wals_in_manifest=true: the close-time write must leave
+// recovery with only the mandatory WAL-tracking MANIFEST work. The per-CF
+// recovery edits should still skip.
+TEST_F(DBBasicTest, OptimizeManifestForRecoveryEmitsWalDeletionWhenTracking) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.optimize_manifest_for_recovery = true;
+  options.track_and_verify_wals_in_manifest = true;
+  DestroyAndReopen(options);
+  ASSERT_OK(Put("k", "v"));
+  ASSERT_OK(Flush());
+  Close();
+
+  RecoveryOptimizationCounters counters;
+  std::atomic<int> records{0};
+  auto* sp = ROCKSDB_NAMESPACE::SyncPoint::GetInstance();
+  counters.Install();
+  sp->SetCallBack("VersionSet::ProcessManifestWrites:AddRecord",
+                  [&](void*) { records.fetch_add(1); });
+  Reopen(options);
+  counters.Uninstall();
+
+  ASSERT_GT(counters.per_cf.load(), 0);
+  ASSERT_EQ(1, records.load());
+  ASSERT_GT(dbfull()->GetVersionSet()->GetWalSet().GetMinWalNumberToKeep(), 0u);
+  ASSERT_EQ("v", Get("k"));
+}
+
+// Regression for the close-time marker path: if a new WAL is created while an
+// otherwise-empty user CF stays empty, Close() must reserve the next file
+// number before persisting SetLogNumber(cur_wal + 1) for that CF.
+TEST_F(DBBasicTest, OptimizeManifestForRecoveryImmediateCloseAfterWarmReopen) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.optimize_manifest_for_recovery = true;
+  CreateAndReopenWithCF({"empty_cf"}, options);
+
+  ASSERT_OK(Put(0, "k", "v"));
+  const uint64_t wal_before_switch = dbfull()->TEST_LogfileNumber();
+  ASSERT_OK(dbfull()->TEST_SwitchMemtable());
+  ASSERT_GT(dbfull()->TEST_LogfileNumber(), wal_before_switch);
+
+  const uint64_t expected_new_log_num = dbfull()->TEST_LogfileNumber() + 1;
+  ASSERT_EQ(expected_new_log_num,
+            dbfull()->GetVersionSet()->current_next_file_number());
+
+  Close();
+  ReopenWithColumnFamilies({"default", "empty_cf"}, options);
+
+  ASSERT_GT(dbfull()->GetVersionSet()->current_next_file_number(),
+            expected_new_log_num);
+  ASSERT_EQ("v", Get(0, "k"));
+}
+
+// Dropped-CF safety: dropped column families must NOT have markers written for
+// them at close time (the !IsDropped() guard protects against attaching the
+// global edit to a dropped CF, which would later trip MANIFEST replay
+// assertions).
+TEST_F(DBBasicTest, OptimizeManifestForRecoverySkipsDroppedCF) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.optimize_manifest_for_recovery = true;
+  CreateAndReopenWithCF({"to_drop", "keeper"}, options);
+  ASSERT_OK(Put(1, "k", "v1"));
+  ASSERT_OK(Put(2, "k", "v2"));
+  ASSERT_OK(Flush(1));
+  ASSERT_OK(Flush(2));
+  ASSERT_OK(db_->DropColumnFamily(handles_[1]));
+  Close();
+
+  ReopenWithColumnFamilies({"default", "keeper"}, options);
+  ASSERT_EQ("v2", Get(1, "k"));
+}
+
+// reuse_manifest_on_open=true: the next LogAndApply after Recover must
+// append to the existing MANIFEST file instead of allocating a fresh
+// one. Force a write after reopen and verify the MANIFEST file number
+// stays unchanged.
+TEST_F(DBBasicTest, ReuseManifestOnOpenAppendsToExistingFile) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.reuse_manifest_on_open = true;
+  DestroyAndReopen(options);
+  ASSERT_OK(Put("k", "v"));
+  ASSERT_OK(Flush());
+  Close();
+
+  Reopen(options);
+  const uint64_t manifest_after_reopen =
+      dbfull()->TEST_Current_Manifest_FileNo();
+  ASSERT_OK(Put("k2", "v2"));
+  ASSERT_OK(Flush());
+  const uint64_t manifest_after_flush =
+      dbfull()->TEST_Current_Manifest_FileNo();
+
+  EXPECT_EQ(manifest_after_reopen, manifest_after_flush);
+  EXPECT_EQ("v", Get("k"));
+  EXPECT_EQ("v2", Get("k2"));
+}
+
+// Default off: ReopenManifestForAppend must NOT be invoked.
+TEST_F(DBBasicTest, ReuseManifestOnOpenDefaultOffSkipsReopen) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  // reuse_manifest_on_open defaults to false
+  DestroyAndReopen(options);
+  ASSERT_OK(Put("k", "v"));
+  ASSERT_OK(Flush());
+  Close();
+
+  std::atomic<int> reopened{0};
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "VersionSet::ReopenManifestForAppend:Reopened",
+      [&](void* /*arg*/) { reopened.fetch_add(1); });
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
+
+  Reopen(options);
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks();
+
+  ASSERT_EQ(0, reopened.load());
+}
+
+// Regression for the WritableFileWriter::filesize_=0 bug fixed via
+// constructor-time initial_file_size: after ReopenManifestForAppend binds a
+// writer to the existing MANIFEST, GetFileSize() must return the on-disk
+// size, not 0.
+// (If it returned 0, the size-limit check in ProcessManifestWrites would
+// compare against 0 and Close-time Truncate could shrink the file.)
+TEST_F(DBBasicTest, ReuseManifestOnOpenAdoptsOnDiskSize) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.reuse_manifest_on_open = true;
+  DestroyAndReopen(options);
+  ASSERT_OK(Put("k", "v"));
+  ASSERT_OK(Flush());
+  Close();
+
+  // Capture the on-disk MANIFEST size before reopen.
+  Reopen(options);
+  const uint64_t manifest_no = dbfull()->TEST_Current_Manifest_FileNo();
+  const std::string manifest_path = DescriptorFileName(dbname_, manifest_no);
+  uint64_t on_disk_size = 0;
+  ASSERT_OK(env_->GetFileSize(manifest_path, &on_disk_size));
+  ASSERT_GT(on_disk_size, 0u);
+
+  // Force a write to flush the writer's buffer; verify Close doesn't
+  // shrink the file (which would happen if Truncate(filesize_=0) ran).
+  ASSERT_OK(Put("k2", "v2"));
+  ASSERT_OK(Flush());
+  Close();
+
+  uint64_t after_close_size = 0;
+  ASSERT_OK(env_->GetFileSize(manifest_path, &after_close_size));
+  EXPECT_GE(after_close_size, on_disk_size);
+}
+
+// MANIFEST rotation continues to work under reuse: when the file grows
+// past tuned_max_manifest_file_size_, ProcessManifestWrites must rotate
+// to a fresh MANIFEST (same as legacy). Use the rotation SyncPoint to
+// observe rotation directly, since tuned_max_manifest_file_size_ is
+// auto-derived and not directly = max_manifest_file_size.
+TEST_F(DBBasicTest, ReuseManifestOnOpenStillRotatesOnSizeCap) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.reuse_manifest_on_open = true;
+  options.max_manifest_file_size = 1;
+  options.max_manifest_space_amp_pct = 0;  // disable amp-based tuning
+  DestroyAndReopen(options);
+  ASSERT_OK(Put("k", "v"));
+  ASSERT_OK(Flush());
+  Close();
+
+  std::atomic<int> rotations{0};
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "VersionSet::ProcessManifestWrites:BeforeNewManifest",
+      [&](void* /*arg*/) { rotations.fetch_add(1); });
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
+
+  Reopen(options);
+  ASSERT_OK(Put("k2", "v2"));
+  ASSERT_OK(Flush());
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks();
+
+  // The Flush after Reopen must have triggered a rotation given the
+  // tiny size cap — proves the size-driven rotation path still runs
+  // when descriptor_log_ is bound by reuse.
+  EXPECT_GT(rotations.load(), 0);
+  EXPECT_EQ("v", Get("k"));
+  EXPECT_EQ("v2", Get("k2"));
+}
+
+// Multi-CF reuse: append-mode MANIFEST must correctly handle edits
+// from multiple CFs after reopen.
+TEST_F(DBBasicTest, ReuseManifestOnOpenMultiCF) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.reuse_manifest_on_open = true;
+  CreateAndReopenWithCF({"alpha", "beta"}, options);
+  ASSERT_OK(Put(0, "k", "v0"));
+  ASSERT_OK(Put(1, "k", "v1"));
+  ASSERT_OK(Put(2, "k", "v2"));
+  ASSERT_OK(Flush(0));
+  ASSERT_OK(Flush(1));
+  ASSERT_OK(Flush(2));
+  Close();
+
+  ReopenWithColumnFamilies({"default", "alpha", "beta"}, options);
+  // Trigger more writes post-reopen on each CF — these get appended to
+  // the reused MANIFEST.
+  ASSERT_OK(Put(0, "k2", "v0b"));
+  ASSERT_OK(Put(1, "k2", "v1b"));
+  ASSERT_OK(Put(2, "k2", "v2b"));
+  ASSERT_OK(Flush(0));
+  ASSERT_OK(Flush(1));
+  ASSERT_OK(Flush(2));
+
+  EXPECT_EQ("v0", Get(0, "k"));
+  EXPECT_EQ("v1", Get(1, "k"));
+  EXPECT_EQ("v2", Get(2, "k"));
+  EXPECT_EQ("v0b", Get(0, "k2"));
+  EXPECT_EQ("v1b", Get(1, "k2"));
+  EXPECT_EQ("v2b", Get(2, "k2"));
+}
+
+// Disabled under best_efforts_recovery: that mode rebuilds CURRENT and
+// MANIFEST as the side-effect of LogAndApplyForRecovery emitting an
+// edit; reusing the prior MANIFEST contradicts the salvage contract.
+TEST_F(DBBasicTest, ReuseManifestOnOpenDisabledByBestEffortsRecovery) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.reuse_manifest_on_open = true;
+  options.best_efforts_recovery = true;
+  DestroyAndReopen(options);
+  ASSERT_OK(Put("k", "v"));
+  ASSERT_OK(Flush());
+  Close();
+
+  std::atomic<int> reopened{0};
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "VersionSet::ReopenManifestForAppend:Reopened",
+      [&](void* /*arg*/) { reopened.fetch_add(1); });
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
+
+  Reopen(options);
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks();
+
+  ASSERT_EQ(0, reopened.load());
+  ASSERT_EQ("v", Get("k"));
+}
+
+// Full-stack composition: optimize_manifest_for_recovery plus
+// reuse_manifest_on_open. Verifies that Close writes recovery markers,
+// Reopen skips clean-recovery MANIFEST edits, and the MANIFEST is reused
+// instead of recreated for the next metadata update.
+TEST_F(DBBasicTest, ReuseManifestOnOpenFullStackComposition) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.optimize_manifest_for_recovery = true;
+  options.reuse_manifest_on_open = true;
+  DestroyAndReopen(options);
+  ASSERT_OK(Put("k", "v"));
+  ASSERT_OK(Flush());
+  Close();
+
+  // Capture the MANIFEST file number before reopen.
+  std::atomic<int> reopened{0};
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "VersionSet::ReopenManifestForAppend:Reopened",
+      [&](void* /*arg*/) { reopened.fetch_add(1); });
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
+
+  Reopen(options);
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks();
+
+  // All three composing: reuse fired, data is intact, no fresh MANIFEST.
+  EXPECT_EQ(1, reopened.load());
+  EXPECT_EQ("v", Get("k"));
+}
+
+// Direct unit test for WritableFileWriter's initial_file_size parameter:
+// verifies the visible size accessors report the existing on-disk size
+// immediately, rather than the constructor's zero default.
+TEST_F(DBBasicTest, WritableFileWriterInitialFileSizeAdoptsExistingSize) {
+  Env* env = Env::Default();
+  std::string fname = test::PerThreadDBPath("set_file_size_test");
+  ASSERT_OK(env->CreateDirIfMissing(test::TmpDir(env)));
+
+  // Create a file with some bytes, close it.
+  {
+    std::unique_ptr<WritableFile> raw;
+    ASSERT_OK(env->NewWritableFile(fname, &raw, EnvOptions()));
+    ASSERT_OK(raw->Append("hello world"));
+    ASSERT_OK(raw->Close());
+  }
+
+  // Reopen and wrap in a WritableFileWriter without initial_file_size —
+  // GetFileSize() should be 0 (the constructor's default).
+  std::unique_ptr<FSWritableFile> fs_file;
+  ASSERT_OK(env->GetFileSystem()->ReopenWritableFile(
+      fname, FileOptions(), &fs_file, /*dbg=*/nullptr));
+  std::unique_ptr<WritableFileWriter> writer(
+      new WritableFileWriter(std::move(fs_file), fname, FileOptions()));
+  EXPECT_EQ(0u, writer->GetFileSize());
+
+  // Reopen again and seed the writer's size accounting from the existing
+  // bytes. GetFileSize and GetFlushedSize must reflect it immediately.
+  ASSERT_OK(env->GetFileSystem()->ReopenWritableFile(
+      fname, FileOptions(), &fs_file, /*dbg=*/nullptr));
+  writer.reset(new WritableFileWriter(
+      std::move(fs_file), fname, FileOptions(),
+      /*clock=*/nullptr, /*io_tracer=*/nullptr, /*stats=*/nullptr,
+      Histograms::HISTOGRAM_ENUM_MAX, /*listeners=*/{},
+      /*file_checksum_gen_factory=*/nullptr,
+      /*perform_data_verification=*/false,
+      /*buffered_data_with_checksum=*/false,
+      /*initial_file_size=*/11));
+  EXPECT_EQ(11u, writer->GetFileSize());
+  EXPECT_EQ(11u, writer->GetFlushedSize());
+
+  ASSERT_OK(env->DeleteFile(fname));
+}
+
+// Tail corruption: appending garbage bytes to the MANIFEST after a
+// clean close must prevent reuse — the physical size exceeds the
+// last valid record end.
+TEST_F(DBBasicTest, ReuseManifestOnOpenSkipsOnTailCorruption) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.reuse_manifest_on_open = true;
+  DestroyAndReopen(options);
+  ASSERT_OK(Put("k", "v"));
+  ASSERT_OK(Flush());
+  Close();
+
+  // Find the MANIFEST file and append garbage to it.
+  std::string manifest_path;
+  {
+    std::vector<std::string> files;
+    ASSERT_OK(env_->GetChildren(dbname_, &files));
+    for (const auto& f : files) {
+      uint64_t number;
+      FileType type;
+      if (ParseFileName(f, &number, &type) && type == kDescriptorFile) {
+        manifest_path = dbname_ + "/" + f;
+        break;
+      }
+    }
+  }
+  ASSERT_FALSE(manifest_path.empty());
+  {
+    std::string contents;
+    ASSERT_OK(ReadFileToString(env_, manifest_path, &contents));
+    contents.append("garbage!");
+    ASSERT_OK(WriteStringToFile(env_, contents, manifest_path));
+  }
+
+  std::atomic<int> reopened{0};
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "VersionSet::ReopenManifestForAppend:Reopened",
+      [&](void* /*arg*/) { reopened.fetch_add(1); });
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
+
+  Reopen(options);
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks();
+
+  ASSERT_EQ(0, reopened.load());
+  ASSERT_EQ("v", Get("k"));
+}
+
 TEST_F(DBBasicTest, EnableDirectIOWithZeroBuf) {
   if (!IsDirectIOSupported()) {
     ROCKSDB_GTEST_BYPASS("Direct IO not supported");

db/db_flush_test.cc

@@ -3946,8 +3946,12 @@ TEST_F(DBFlushTest, FlushAfterReadPathRangeTombstoneInsertion) {
         // MarkImmutable), this returns false because the memtable is
         // already immutable. Without the fix, this succeeds and creates
         // an entry count mismatch.
-        cfh->cfd()->mem()->AddLogicallyRedundantRangeTombstone(1 /* seq */, "b",
-                                                               "f");
+        // Try to insert a range tombstone. With the fix (Construct after
+        // MarkImmutable), this returns false because the memtable is
+        // already immutable. Without the fix, this succeeds and creates
+        // an entry count mismatch.
+        cfh->cfd()->mem()->AddLogicallyRedundantRangeTombstone(
+            1 /* seq */, "b", "f", cfh->cfd()->GetIngestSstLock());
       });
   SyncPoint::GetInstance()->EnableProcessing();
 

db/db_impl/db_impl.cc

@@ -15,6 +15,7 @@
 
 #include <cinttypes>
 #include <cstdio>
+#include <deque>
 #include <map>
 #include <memory>
 #include <optional>
@@ -173,6 +174,7 @@ DBImpl::DBImpl(const DBOptions& options, const std::string& dbname,
                const bool seq_per_batch, const bool batch_per_txn,
                bool read_only)
     : dbname_(dbname),
+      read_only_(read_only),
       own_info_log_(options.info_log == nullptr),
       initial_db_options_(SanitizeOptions(dbname, options, read_only,
                                           &init_logger_creation_s_)),
@@ -636,6 +638,106 @@ void DBImpl::UnregisterBlobDirectWriteColumnFamily() {
   assert(false);
 }
 
+Status DBImpl::MaybeWriteWalMarkersToManifestOnClose() {
+  mutex_.AssertHeld();
+  if (!mutable_db_options_.optimize_manifest_for_recovery ||
+      !opened_successfully_ || read_only_ || versions_ == nullptr ||
+      logs_.empty()) {
+    return Status::OK();
+  }
+
+  TEST_SYNC_POINT("DBImpl::CloseHelper:WriteWalMarkersOnCloseEntered");
+  const ReadOptions read_options(Env::IOActivity::kUnknown);
+  const WriteOptions write_options(Env::IOActivity::kUnknown);
+  uint64_t max_wal_number = logs_.back().number;
+  TEST_SYNC_POINT_CALLBACK("DBImpl::CloseHelper:CapturedMaxWal",
+                           &max_wal_number);
+  const uint64_t new_log_num = max_wal_number + 1;
+
+  // std::deque keeps VersionEdit pointers stable across emplace_back.
+  std::deque<VersionEdit> edits;
+  autovector<ColumnFamilyData*> cfds;
+  autovector<autovector<VersionEdit*>> edit_lists;
+  VersionEdit global_edit;
+  bool persist_global_edit = false;
+  bool reserved_log_number = false;
+
+  bool all_cfs_empty = true;
+  for (auto* cfd : *versions_->GetColumnFamilySet()) {
+    if (cfd->IsDropped()) {
+      continue;
+    }
+    // cfd->IsEmpty() checks both mem and imm. An immutable memtable queued for
+    // flush still pins an older WAL; advancing past it would drop unflushed
+    // records on the next open.
+    if (!cfd->IsEmpty()) {
+      all_cfs_empty = false;
+      continue;
+    }
+    if (new_log_num > cfd->GetLogNumber()) {
+      if (!reserved_log_number) {
+        const uint64_t next_file_number = versions_->current_next_file_number();
+        if (next_file_number <= new_log_num) {
+          versions_->FetchAddFileNumber(new_log_num + 1 - next_file_number);
+          global_edit.SetNextFile(versions_->current_next_file_number());
+          persist_global_edit = true;
+        }
+        reserved_log_number = true;
+      }
+      edits.emplace_back();
+      VersionEdit& edit = edits.back();
+      edit.SetLogNumber(new_log_num);
+      edit.SetColumnFamily(cfd->GetID());
+      cfds.push_back(cfd);
+      edit_lists.emplace_back();
+      edit_lists.back().push_back(&edit);
+    }
+  }
+
+  if (all_cfs_empty && !allow_2pc()) {
+    if (new_log_num > versions_->min_log_number_to_keep()) {
+      global_edit.SetMinLogNumberToKeep(new_log_num);
+      persist_global_edit = true;
+    }
+    // Mirror the monotonicity guard in db_impl_files.cc:
+    // PrecomputeWalDeletionEdit emits DeleteWalsBefore only when the
+    // marker actually advances the WalSet's existing min. Skipping
+    // this guard would let a stale-replay edit regress WalSet state.
+    if (immutable_db_options_.track_and_verify_wals_in_manifest &&
+        new_log_num > versions_->GetWalSet().GetMinWalNumberToKeep()) {
+      global_edit.DeleteWalsBefore(new_log_num);
+      persist_global_edit = true;
+    }
+  }
+
+  if (persist_global_edit) {
+    ColumnFamilyData* default_cfd =
+        versions_->GetColumnFamilySet()->GetDefault();
+    if (default_cfd && !default_cfd->IsDropped()) {
+      bool attached_to_existing = false;
+      for (size_t i = 0; i < cfds.size(); ++i) {
+        if (cfds[i] == default_cfd) {
+          edit_lists[i].push_back(&global_edit);
+          attached_to_existing = true;
+          break;
+        }
+      }
+      if (!attached_to_existing) {
+        cfds.push_back(default_cfd);
+        edit_lists.emplace_back();
+        edit_lists.back().push_back(&global_edit);
+      }
+    }
+  }
+
+  if (cfds.empty()) {
+    return Status::OK();
+  }
+  return versions_->LogAndApply(cfds, read_options, write_options, edit_lists,
+                                &mutex_, directories_.GetDbDir(),
+                                /*new_descriptor_log=*/false);
+}
+
 Status DBImpl::CloseHelper() {
   // Guarantee that there is no background error recovery in progress before
   // continuing with the shutdown
@@ -727,7 +829,7 @@ Status DBImpl::CloseHelper() {
   // manifest file), it is not able to identify live files correctly. As a
   // result, all "live" files can get deleted by accident. However, corrupted
   // manifest is recoverable by RepairDB().
-  if (opened_successfully_) {
+  if (opened_successfully_ && !read_only_) {
     JobContext job_context(next_job_id_.fetch_add(1));
     FindObsoleteFiles(&job_context, true);
 
@@ -740,6 +842,17 @@ Status DBImpl::CloseHelper() {
     job_context.Clean();
     mutex_.Lock();
   }
+  // Best-effort warm-reopen optimization: after obsolete-file cleanup and
+  // before logs_.clear(), persist close-time WAL markers that can reduce
+  // recovery work on the next open. This matters most when some CFs have no
+  // unflushed writes and/or the newest WAL is empty, since recovery would
+  // otherwise append marker-only MANIFEST edits on reopen.
+  Status manifest_s = MaybeWriteWalMarkersToManifestOnClose();
+  if (!manifest_s.ok()) {
+    ROCKS_LOG_WARN(immutable_db_options_.info_log,
+                   "optimize_manifest_for_recovery close-time write failed: %s",
+                   manifest_s.ToString().c_str());
+  }
   {
     InstrumentedMutexLock lock(&wal_write_mutex_);
     for (auto l : wals_to_free_) {
@@ -1665,9 +1778,12 @@ Status DBImpl::SetDBOptions(
                                           : new_options.max_open_files - 10);
       // Potential table cache capacity change requires updating if table
       // handles should get pinned.
+      versions_->GetColumnFamilySet()->SetFastSstOpen(
+          new_options.fast_sst_open);
       for (auto cfd : *versions_->GetColumnFamilySet()) {
         if (!cfd->IsDropped()) {
           cfd->table_cache()->UpdateShouldPinTableHandles();
+          cfd->table_cache()->SetFastSstOpen(new_options.fast_sst_open);
         }
       }
       wal_other_option_changed = mutable_db_options_.wal_bytes_per_sync !=
@@ -6490,6 +6606,19 @@ Status DBImpl::IngestExternalFiles(
   TEST_SYNC_POINT("DBImpl::IngestExternalFiles:BeforeJobsRun:0");
   TEST_SYNC_POINT("DBImpl::IngestExternalFiles:BeforeJobsRun:1");
   TEST_SYNC_POINT("DBImpl::AddFile:Start");
+
+  // Acquire per-CF ingest_sst_lock as ReadLocks (shared) BEFORE the DB
+  // mutex so the lock-acquisition order is ingest_sst_lock -> DB mutex
+  // throughout. Use readlock so we still allow concurrent ingestions.
+  std::vector<std::unique_ptr<ReadLock>> ingest_read_locks;
+  ingest_read_locks.reserve(num_cfs);
+  for (size_t i = 0; i != num_cfs; ++i) {
+    auto* cfd = ingestion_jobs[i].GetColumnFamilyData();
+    if (!cfd->IsDropped()) {
+      ingest_read_locks.emplace_back(
+          std::make_unique<ReadLock>(&cfd->GetIngestSstLock()));
+    }
+  }
   {
     InstrumentedMutexLock l(&mutex_);
     TEST_SYNC_POINT("DBImpl::AddFile:MutexLock");
@@ -6579,6 +6708,23 @@ Status DBImpl::IngestExternalFiles(
         ingestion_jobs[i].RegisterRange();
       }
     }
+    // Now that Run() has assigned the actual seqno for each ingested file,
+    // bump each affected memtable's ingest_seqno_barrier_ to that exact
+    // value. We still hold the per-CF ingest_sst_lock as a ReadLock; any
+    // concurrent conversion's TryWriteLock on the same lock fails, so no
+    // converter can be reading or about to read the barrier while we
+    // update it. After we release the ReadLocks at function exit, the
+    // next conversion observes the new barrier and refuses any insert
+    // with insert_seq < assigned.
+    if (status.ok()) {
+      for (size_t i = 0; i != num_cfs; ++i) {
+        auto* cfd = ingestion_jobs[i].GetColumnFamilyData();
+        SequenceNumber assigned = ingestion_jobs[i].MaxAssignedSequenceNumber();
+        if (assigned > 0) {
+          cfd->mem()->BumpIngestSeqnoBarrier(assigned);
+        }
+      }
+    }
     if (status.ok()) {
       ReadOptions read_options;
       read_options.fill_cache = args[0].options.fill_cache;

db/db_impl/db_impl.h

@@ -1371,6 +1371,7 @@ class DBImpl : public DB {
 
  protected:
   const std::string dbname_;
+  const bool read_only_;
   // TODO(peterd): unify with VersionSet::db_id_
   std::string db_id_;
   // db_session_id_ is an identifier that gets reset
@@ -2743,6 +2744,8 @@ class DBImpl : public DB {
 
   Status MaybeReleaseTimestampedSnapshotsAndCheck();
 
+  Status MaybeWriteWalMarkersToManifestOnClose();
+
   Status CloseHelper();
 
   void WaitForBackgroundWork();

db/db_impl/db_impl_compaction_flush.cc

@@ -3162,7 +3162,7 @@ void DBImpl::ResumeAllCompactions() {
 void DBImpl::MaybeScheduleFlushOrCompaction() {
   mutex_.AssertHeld();
   TEST_SYNC_POINT("DBImpl::MaybeScheduleFlushOrCompaction:Start");
-  if (!opened_successfully_) {
+  if (!opened_successfully_ || read_only_) {
     // Compaction may introduce data race to DB open
     return;
   }

db/db_impl/db_impl_files.cc

@@ -1186,8 +1186,9 @@ std::set<std::string> DBImpl::CollectAllDBPaths() {
 
 Status DBImpl::MaybeUpdateNextFileNumber(RecoveryContext* recovery_ctx) {
   mutex_.AssertHeld();
-  uint64_t next_file_number = versions_->current_next_file_number();
-  uint64_t largest_file_number = next_file_number;
+  const uint64_t prev_next_file_number = versions_->current_next_file_number();
+  uint64_t largest_file_number = prev_next_file_number;
+  bool on_disk_file_advanced_counter = false;
   Status s;
   for (const auto& path : CollectAllDBPaths()) {
     std::vector<std::string> files;
@@ -1201,7 +1202,15 @@ Status DBImpl::MaybeUpdateNextFileNumber(RecoveryContext* recovery_ctx) {
       if (!ParseFileName(fname, &number, &type)) {
         continue;
       }
-      const std::string normalized_fpath = path + kFilePathSeparator + fname;
+      std::string normalized_fpath = path;
+      normalized_fpath += kFilePathSeparator;
+      normalized_fpath.append(fname);
+      // Use >= so a crashed-mid-allocation file at exactly
+      // prev_next_file_number triggers the advance — otherwise the next
+      // NewFileNumber() would collide with it.
+      if (number >= prev_next_file_number) {
+        on_disk_file_advanced_counter = true;
+      }
       largest_file_number = std::max(largest_file_number, number);
       if ((type == kTableFile || type == kBlobFile)) {
         recovery_ctx->existing_data_files_.push_back(normalized_fpath);
@@ -1212,16 +1221,31 @@ Status DBImpl::MaybeUpdateNextFileNumber(RecoveryContext* recovery_ctx) {
     return s;
   }
 
-  if (largest_file_number >= next_file_number) {
-    versions_->next_file_number_.store(largest_file_number + 1);
-  }
+  // Legacy bug: initializing largest_file_number from prev_next_file_number
+  // made the post-loop check trivially true and emitted a noop SetNextFile
+  // on every recovery.
+  // Preserved verbatim by default (some tests pin file numbers); skipped
+  // only when the option is on AND we're not in salvage mode (which
+  // requires the MANIFEST rewrite this emit triggers).
+  const bool optimize_manifest_for_recovery =
+      mutable_db_options_.optimize_manifest_for_recovery &&
+      !immutable_db_options_.best_efforts_recovery;
 
-  VersionEdit edit;
-  edit.SetNextFile(versions_->next_file_number_.load());
-  assert(versions_->GetColumnFamilySet());
-  ColumnFamilyData* default_cfd = versions_->GetColumnFamilySet()->GetDefault();
-  assert(default_cfd);
-  recovery_ctx->UpdateVersionEdits(default_cfd, edit);
+  if (!optimize_manifest_for_recovery || on_disk_file_advanced_counter) {
+    if (largest_file_number >= prev_next_file_number) {
+      versions_->next_file_number_.store(largest_file_number + 1);
+    }
+    TEST_SYNC_POINT("DBImpl::MaybeUpdateNextFileNumber:EmitEdit");
+    VersionEdit edit;
+    edit.SetNextFile(versions_->next_file_number_.load());
+    assert(versions_->GetColumnFamilySet());
+    ColumnFamilyData* default_cfd =
+        versions_->GetColumnFamilySet()->GetDefault();
+    assert(default_cfd);
+    recovery_ctx->UpdateVersionEdits(default_cfd, edit);
+  } else {
+    TEST_SYNC_POINT("DBImpl::Recovery:SkippedNoopEdit:NextFileNumber");
+  }
   return s;
 }
 }  // namespace ROCKSDB_NAMESPACE

db/db_impl/db_impl_open.cc

@@ -680,8 +680,18 @@ Status DBImpl::Recover(
   } else if (immutable_db_options_.write_dbid_to_manifest && recovery_ctx) {
     VersionEdit edit;
     s = SetupDBId(write_options, read_only, is_new_db, is_retry, &edit);
-    recovery_ctx->UpdateVersionEdits(
-        versions_->GetColumnFamilySet()->GetDefault(), edit);
+    // best_efforts_recovery rebuilds CURRENT/MANIFEST as the side-effect
+    // of LogAndApplyForRecovery emitting an edit, so the optimization is
+    // disabled there (see optimize_manifest_for_recovery doc).
+    const bool optimize_manifest_for_recovery =
+        mutable_db_options_.optimize_manifest_for_recovery &&
+        !immutable_db_options_.best_efforts_recovery;
+    if (!optimize_manifest_for_recovery || edit.HasDbId()) {
+      recovery_ctx->UpdateVersionEdits(
+          versions_->GetColumnFamilySet()->GetDefault(), edit);
+    } else {
+      TEST_SYNC_POINT("DBImpl::Recovery:SkippedNoopEdit:SetupDBId");
+    }
   } else {
     s = SetupDBId(write_options, read_only, is_new_db, is_retry, nullptr);
   }
@@ -770,6 +780,10 @@ Status DBImpl::Recover(
       // otherwise, in the future, if WAL tracking is enabled again,
       // since the WALs deleted when WAL tracking is disabled are not persisted
       // into MANIFEST, WAL check may fail.
+      //
+      // Intentionally NOT gated by optimize_manifest_for_recovery:
+      // this edit is safety-critical for a future re-enable of WAL tracking,
+      // even though it can look noop-ish under the current session.
       VersionEdit edit;
       WalNumber max_wal_number =
           versions_->GetWalSet().GetWals().rbegin()->first;
@@ -983,7 +997,11 @@ Status DBImpl::InitPersistStatsColumnFamily() {
 
 Status DBImpl::LogAndApplyForRecovery(const RecoveryContext& recovery_ctx) {
   mutex_.AssertHeld();
-  assert(versions_->descriptor_log_ == nullptr);
+  // descriptor_log_ is normally null after Recover, but when
+  // reuse_manifest_on_open is set VersionSet::Recover may have already
+  // bound a log::Writer to the existing MANIFEST for append.
+  assert(versions_->descriptor_log_ == nullptr ||
+         immutable_db_options_.reuse_manifest_on_open);
   const ReadOptions read_options(Env::IOActivity::kDBOpen);
   const WriteOptions write_options(Env::IOActivity::kDBOpen);
 
@@ -1870,25 +1888,47 @@ Status DBImpl::MaybeFlushFinalMemtableOrRestoreActiveLogFiles(
       versions_->MarkFileNumberUsed(max_wal_number + 1);
       assert(recovery_ctx != nullptr);
 
+      const bool optimize_manifest_for_recovery =
+          mutable_db_options_.optimize_manifest_for_recovery &&
+          !immutable_db_options_.best_efforts_recovery;
       for (auto* cfd : *versions_->GetColumnFamilySet()) {
         auto iter = version_edits->find(cfd->GetID());
         assert(iter != version_edits->end());
-        recovery_ctx->UpdateVersionEdits(cfd, iter->second);
+        const VersionEdit& cf_edit = iter->second;
+        if (!optimize_manifest_for_recovery ||
+            cf_edit.ShouldEmitPerColumnFamilyRecoveryEdit(
+                cfd->GetLogNumber())) {
+          recovery_ctx->UpdateVersionEdits(cfd, cf_edit);
+        } else {
+          TEST_SYNC_POINT("DBImpl::Recovery:SkippedNoopEdit:PerCF");
+        }
       }
 
       if (flushed || !data_seen) {
+        const uint64_t new_min_log = max_wal_number + 1;
         VersionEdit wal_deletion;
+        bool emit_wal_deletion = false;
         if (immutable_db_options_.track_and_verify_wals_in_manifest) {
-          wal_deletion.DeleteWalsBefore(max_wal_number + 1);
+          // Determining whether DeleteWalsBefore actually shrinks WalSet
+          // membership requires WalSet state outside this site, so emit
+          // unconditionally (pre-existing behavior).
+          wal_deletion.DeleteWalsBefore(new_min_log);
+          emit_wal_deletion = true;
         }
-        if (!allow_2pc()) {
-          // In non-2pc mode, flushing the memtables of the column families
-          // means we can advance min_log_number_to_keep.
-          wal_deletion.SetMinLogNumberToKeep(max_wal_number + 1);
+        const bool min_log_advances =
+            new_min_log > versions_->min_log_number_to_keep();
+        if (!allow_2pc() &&
+            (!optimize_manifest_for_recovery || min_log_advances)) {
+          wal_deletion.SetMinLogNumberToKeep(new_min_log);
+          emit_wal_deletion = true;
         }
         assert(versions_->GetColumnFamilySet() != nullptr);
-        recovery_ctx->UpdateVersionEdits(
-            versions_->GetColumnFamilySet()->GetDefault(), wal_deletion);
+        if (!optimize_manifest_for_recovery || emit_wal_deletion) {
+          recovery_ctx->UpdateVersionEdits(
+              versions_->GetColumnFamilySet()->GetDefault(), wal_deletion);
+        } else {
+          TEST_SYNC_POINT("DBImpl::Recovery:SkippedNoopEdit:WalDeletion");
+        }
       }
     }
   }

db/db_iter.cc

@@ -1788,7 +1788,7 @@ void DBIter::MaybeInsertRangeTombstone(const Slice& end_key) {
     return;
   }
 
-  // Insert at the read sequence so the synthesized tombstone is visible only
+  // Insert at the read sequence so the converted tombstone is visible only
   // to readers that could already observe the deletion run.
   SequenceNumber insert_seq = sequence_;
 
@@ -1819,8 +1819,10 @@ void DBIter::MaybeInsertRangeTombstone(const Slice& end_key) {
     }
   }
 
+  assert(cfh_ != nullptr);
   if (active_mem_->AddLogicallyRedundantRangeTombstone(
-          insert_seq, range_tomb_first_key_.GetUserKey(), end_key)) {
+          insert_seq, range_tomb_first_key_.GetUserKey(), end_key,
+          cfh_->cfd()->GetIngestSstLock())) {
     RecordTick(statistics_, READ_PATH_RANGE_TOMBSTONES_INSERTED);
     ROCKS_LOG_DEBUG(logger_,
                     "Inserted range tombstone [%s, %s) @ seq %" PRIu64

db/db_iterator_test.cc

@@ -5583,13 +5583,13 @@ class ReadPathRangeTombstoneTest : public DBIteratorBaseTest,
   bool Forward() const { return GetParam(); }
 
   void SetUp() override {
-    inserted_ranges_.clear();
+    attempted_insert_ranges_.clear();
     SyncPoint::GetInstance()->SetCallBack(
         "MemTable::AddLogicallyRedundantRangeTombstone:AddRange",
         [this](void* arg) {
           auto* range = static_cast<std::pair<Slice, Slice>*>(arg);
-          inserted_ranges_.emplace_back(range->first.ToString(),
-                                        range->second.ToString());
+          attempted_insert_ranges_.emplace_back(range->first.ToString(),
+                                                range->second.ToString());
         });
     SyncPoint::GetInstance()->EnableProcessing();
   }
@@ -5633,9 +5633,9 @@ class ReadPathRangeTombstoneTest : public DBIteratorBaseTest,
 
   void AssertRange(size_t idx, const std::string& start,
                    const std::string& end) {
-    ASSERT_LT(idx, inserted_ranges_.size());
-    ASSERT_EQ(inserted_ranges_[idx].first, start);
-    ASSERT_EQ(inserted_ranges_[idx].second, end);
+    ASSERT_LT(idx, attempted_insert_ranges_.size());
+    ASSERT_EQ(attempted_insert_ranges_[idx].first, start);
+    ASSERT_EQ(attempted_insert_ranges_[idx].second, end);
   }
 
   Slice MaxTimestamp(std::string* storage) const {
@@ -5675,7 +5675,7 @@ class ReadPathRangeTombstoneTest : public DBIteratorBaseTest,
         << iter->status().ToString();
   }
 
-  std::vector<std::pair<std::string, std::string>> inserted_ranges_;
+  std::vector<std::pair<std::string, std::string>> attempted_insert_ranges_;
 };
 
 INSTANTIATE_TEST_CASE_P(ReadPathRangeTombstoneTest, ReadPathRangeTombstoneTest,
@@ -5698,19 +5698,28 @@ TEST_P(ReadPathRangeTombstoneTest, BasicInsertion) {
 
     if (flush_before_read) {
       ASSERT_OK(Flush());
-      // Memtable is empty after flush. AddLogicallyRedundantRangeTombstone
-      // skips empty memtables.
-      inserted_ranges_.clear();
+      // After dropping the IsEmpty() fast-path in conversion, the now-empty
+      // active memtable is a valid conversion target; the per-CF
+      // ingest_sst_lock (held shared by ingestion) is the mechanism that
+      // gates conversion vs ingestion, not memtable emptiness.
+      attempted_insert_ranges_.clear();
       VerifyIteration({"a", "g", "h", "n"});
-      ASSERT_EQ(inserted_ranges_.size(), 0);
+      ASSERT_EQ(attempted_insert_ranges_.size(), 2);
+      if (forward) {
+        AssertRange(0, "b", "g");
+        AssertRange(1, "i", "n");
+      } else {
+        AssertRange(0, "i", "n");
+        AssertRange(1, "b", "g");
+      }
       break;
     }
 
-    inserted_ranges_.clear();
+    attempted_insert_ranges_.clear();
 
     VerifyIteration({"a", "g", "h", "n"});
 
-    ASSERT_EQ(inserted_ranges_.size(), 2);
+    ASSERT_EQ(attempted_insert_ranges_.size(), 2);
     if (forward) {
       AssertRange(0, "b", "g");
       AssertRange(1, "i", "n");
@@ -5722,9 +5731,9 @@ TEST_P(ReadPathRangeTombstoneTest, BasicInsertion) {
         options.statistics->getTickerCount(READ_PATH_RANGE_TOMBSTONES_INSERTED),
         2);
 
-    inserted_ranges_.clear();
+    attempted_insert_ranges_.clear();
     VerifyIteration({"a", "g", "h", "n"});
-    ASSERT_EQ(inserted_ranges_.size(), 0);
+    ASSERT_EQ(attempted_insert_ranges_.size(), 0);
     ASSERT_EQ(
         options.statistics->getTickerCount(READ_PATH_RANGE_TOMBSTONES_INSERTED),
         2);
@@ -5761,13 +5770,13 @@ TEST_P(ReadPathRangeTombstoneTest, MemtableSwitch) {
                 /*flushed_point_dels=*/{},
                 /*memtable_point_dels=*/{"b", "c", "d", "e", "f"});
 
-  inserted_ranges_.clear();
+  attempted_insert_ranges_.clear();
   SyncPoint::GetInstance()->SetCallBack(
       "MemTable::AddLogicallyRedundantRangeTombstone:AddRange",
       [this](void* arg) {
         auto* range = static_cast<std::pair<Slice, Slice>*>(arg);
-        inserted_ranges_.emplace_back(range->first.ToString(),
-                                      range->second.ToString());
+        attempted_insert_ranges_.emplace_back(range->first.ToString(),
+                                              range->second.ToString());
         auto* cfh =
             static_cast<ColumnFamilyHandleImpl*>(db_->DefaultColumnFamily());
         cfh->cfd()->mem()->MarkImmutable();
@@ -5776,7 +5785,7 @@ TEST_P(ReadPathRangeTombstoneTest, MemtableSwitch) {
 
   VerifyIteration({"a", "g", "h"});
 
-  ASSERT_EQ(inserted_ranges_.size(), 1);
+  ASSERT_EQ(attempted_insert_ranges_.size(), 1);
   AssertRange(0, "b", "g");
   ASSERT_EQ(
       options.statistics->getTickerCount(READ_PATH_RANGE_TOMBSTONES_DISCARDED),
@@ -5809,7 +5818,7 @@ TEST_P(ReadPathRangeTombstoneTest, ExhaustedIteratorWithBounds) {
   VerifyIteration({"e", "f"}, ro);
 
   // Both directions encounter two tombstone runs (a-d and g-j).
-  ASSERT_EQ(inserted_ranges_.size(), 2);
+  ASSERT_EQ(attempted_insert_ranges_.size(), 2);
   if (Forward()) {
     // Forward: sees a-d tombstones first → [a, e), then g-j → [g, z).
     AssertRange(0, "a", "e");
@@ -5830,9 +5839,9 @@ TEST_P(ReadPathRangeTombstoneTest, ExhaustedIteratorWithBounds) {
   ASSERT_EQ(Get("j"), "NOT_FOUND");
 
   // Second read: range tombstones already in memtable, no new insertion.
-  inserted_ranges_.clear();
+  attempted_insert_ranges_.clear();
   VerifyIteration({"e", "f"}, ro);
-  ASSERT_EQ(inserted_ranges_.size(), 0);
+  ASSERT_EQ(attempted_insert_ranges_.size(), 0);
   ASSERT_EQ(
       options.statistics->getTickerCount(READ_PATH_RANGE_TOMBSTONES_INSERTED),
       2);
@@ -5853,7 +5862,7 @@ TEST_P(ReadPathRangeTombstoneTest, ExhaustedIteratorNoBounds) {
 
   // Without bounds, only the a-d run (which has a live key boundary) gets
   // inserted. The g-j run at the end has no upper bound → dropped.
-  ASSERT_EQ(inserted_ranges_.size(), 1);
+  ASSERT_EQ(attempted_insert_ranges_.size(), 1);
   AssertRange(0, "a", "e");
   ASSERT_EQ(
       options.statistics->getTickerCount(READ_PATH_RANGE_TOMBSTONES_INSERTED),
@@ -5871,7 +5880,7 @@ TEST_P(ReadPathRangeTombstoneTest, DirectionChange) {
                 /*flushed_point_dels=*/{"c", "d", "e"},
                 /*memtable_point_dels=*/{"f", "g", "j", "k", "l", "m", "n"});
 
-  inserted_ranges_.clear();
+  attempted_insert_ranges_.clear();
 
   {
     ReadOptions ro;
@@ -5918,7 +5927,7 @@ TEST_P(ReadPathRangeTombstoneTest, DirectionChange) {
     ASSERT_OK(iter->status());
   }
 
-  ASSERT_EQ(inserted_ranges_.size(), 2);
+  ASSERT_EQ(attempted_insert_ranges_.size(), 2);
   if (forward_first) {
     AssertRange(0, "c", "h");
     AssertRange(1, "j", "o");
@@ -5948,11 +5957,11 @@ TEST_P(ReadPathRangeTombstoneTest, MixedDeleteAndSingleDelete) {
   ASSERT_OK(SingleDelete("c"));
   ASSERT_OK(SingleDelete("e"));
 
-  inserted_ranges_.clear();
+  attempted_insert_ranges_.clear();
 
   VerifyIteration({"a", "g", "h"});
 
-  ASSERT_EQ(inserted_ranges_.size(), 1);
+  ASSERT_EQ(attempted_insert_ranges_.size(), 1);
   AssertRange(0, "b", "g");
   ASSERT_EQ(
       options.statistics->getTickerCount(READ_PATH_RANGE_TOMBSTONES_INSERTED),
@@ -5982,11 +5991,11 @@ TEST_P(ReadPathRangeTombstoneTest, SingleDeleteOnlyRun) {
   ASSERT_OK(SingleDelete("e"));
   ASSERT_OK(SingleDelete("f"));
 
-  inserted_ranges_.clear();
+  attempted_insert_ranges_.clear();
 
   VerifyIteration({"a", "g", "h"});
 
-  ASSERT_EQ(inserted_ranges_.size(), 1);
+  ASSERT_EQ(attempted_insert_ranges_.size(), 1);
   AssertRange(0, "b", "g");
   ASSERT_EQ(
       options.statistics->getTickerCount(READ_PATH_RANGE_TOMBSTONES_INSERTED),
@@ -5997,7 +6006,7 @@ TEST_P(ReadPathRangeTombstoneTest, PrefixFilterDefaultReadOptions) {
   // total_order_seek=false (default) with prefix extractor. Even when the
   // visible scan contains a valid in-prefix tombstone run [ba, bc), read-path
   // range conversion is disabled in this legacy prefix mode, so no
-  // synthesized memtable tombstone is inserted.
+  // converted memtable tombstone is inserted.
   Options options = CurrentOptions();
   options.min_tombstones_for_range_conversion = 2;
   options.statistics = CreateDBStatistics();
@@ -6017,7 +6026,7 @@ TEST_P(ReadPathRangeTombstoneTest, PrefixFilterDefaultReadOptions) {
   ASSERT_OK(Delete("ba"));
   ASSERT_OK(Delete("bb"));
 
-  inserted_ranges_.clear();
+  attempted_insert_ranges_.clear();
   auto it = std::unique_ptr<Iterator>(db_->NewIterator(ReadOptions()));
   if (Forward()) {
     it->Seek("b");
@@ -6032,7 +6041,7 @@ TEST_P(ReadPathRangeTombstoneTest, PrefixFilterDefaultReadOptions) {
     }
   }
   ASSERT_OK(it->status());
-  ASSERT_EQ(inserted_ranges_.size(), 0u);
+  ASSERT_EQ(attempted_insert_ranges_.size(), 0u);
   ASSERT_EQ(
       options.statistics->getTickerCount(READ_PATH_RANGE_TOMBSTONES_INSERTED),
       0u);
@@ -6088,7 +6097,7 @@ TEST_P(ReadPathRangeTombstoneTest, PrefixFilterTotalOrderSeek) {
     ASSERT_OK(del("ca"));
     ASSERT_OK(put("fa", "above"));
 
-    inserted_ranges_.clear();
+    attempted_insert_ranges_.clear();
     ReadOptions ro;
     ro.total_order_seek = true;
     if (use_udt) {
@@ -6109,7 +6118,7 @@ TEST_P(ReadPathRangeTombstoneTest, PrefixFilterTotalOrderSeek) {
     }
     ASSERT_OK(it->status());
     // Tombstone crosses from prefix 'b' into 'c', terminated by live "cb".
-    ASSERT_EQ(inserted_ranges_.size(), 1u);
+    ASSERT_EQ(attempted_insert_ranges_.size(), 1u);
     if (use_udt) {
       AssertRange(0, std::string("ba") + ts, std::string("cb") + ts);
     } else {
@@ -6121,7 +6130,7 @@ TEST_P(ReadPathRangeTombstoneTest, PrefixFilterTotalOrderSeek) {
 TEST_P(ReadPathRangeTombstoneTest, PrefixFilterPrefixSameAsStart) {
   // prefix_same_as_start=true: prefix filtering active, DBIter bounds the
   // scan to the seek prefix. An out-of-prefix tombstone ends the visible run,
-  // but the synthesized range still stays within prefix by flushing to the
+  // but the converted range still stays within prefix by flushing to the
   // last tracked in-prefix tombstone.
   // total_order_seek should not matter as we are guaranteed a total order view
   // within the prefix bounds.
@@ -6169,7 +6178,7 @@ TEST_P(ReadPathRangeTombstoneTest, PrefixFilterPrefixSameAsStart) {
       ASSERT_OK(del("ca"));
       ASSERT_OK(put("fa", "above"));
 
-      inserted_ranges_.clear();
+      attempted_insert_ranges_.clear();
       ReadOptions ro;
       ro.prefix_same_as_start = true;
       ro.total_order_seek = total_order;
@@ -6187,7 +6196,7 @@ TEST_P(ReadPathRangeTombstoneTest, PrefixFilterPrefixSameAsStart) {
         ASSERT_FALSE(it->Valid());
       }
       ASSERT_OK(it->status());
-      ASSERT_EQ(inserted_ranges_.size(), 1u);
+      ASSERT_EQ(attempted_insert_ranges_.size(), 1u);
       ASSERT_EQ(options.statistics->getTickerCount(
                     READ_PATH_RANGE_TOMBSTONES_INSERTED),
                 1u);
@@ -6235,7 +6244,7 @@ TEST_P(ReadPathRangeTombstoneTest, PrefixFilterOutOfDomainSeek) {
   ASSERT_OK(Put("bbdd", "v2"));
   ASSERT_OK(Put("cccc", "v3"));
 
-  inserted_ranges_.clear();
+  attempted_insert_ranges_.clear();
   ReadOptions ro;
   ro.prefix_same_as_start = true;
   auto it = std::unique_ptr<Iterator>(db_->NewIterator(ro));
@@ -6254,7 +6263,7 @@ TEST_P(ReadPathRangeTombstoneTest, PrefixFilterOutOfDomainSeek) {
     }
   }
   ASSERT_OK(it->status());
-  ASSERT_EQ(inserted_ranges_.size(), 1u);
+  ASSERT_EQ(attempted_insert_ranges_.size(), 1u);
   AssertRange(0, "bbbb", "bbdd");
 }
 
@@ -6279,10 +6288,10 @@ TEST_P(ReadPathRangeTombstoneTest, TableFilterNotAllowed) {
   ASSERT_OK(Flush());
 
   // Keep the active memtable non-empty so the read path has somewhere to store
-  // the synthesized memtable range tombstone.
+  // the converted memtable range tombstone.
   ASSERT_OK(Put("zz", "tail_mem"));
 
-  inserted_ranges_.clear();
+  attempted_insert_ranges_.clear();
   {
     // First iterator sees the full SST set, so converting [a, c) into a
     // memtable range tombstone is safe.
@@ -6295,7 +6304,7 @@ TEST_P(ReadPathRangeTombstoneTest, TableFilterNotAllowed) {
     ASSERT_OK(it->status());
   }
 
-  ASSERT_GE(inserted_ranges_.size(), 1u);
+  ASSERT_GE(attempted_insert_ranges_.size(), 1u);
   AssertRange(0, "a", "c");
 
   {
@@ -6304,7 +6313,7 @@ TEST_P(ReadPathRangeTombstoneTest, TableFilterNotAllowed) {
       return props.num_entries != 2;
     };
     // Hiding the two-delete SST would otherwise leave this iterator with a
-    // partial SST view plus the previously synthesized memtable tombstone,
+    // partial SST view plus the previously converted memtable tombstone,
     // allowing hidden SST state to affect the filtered read result.
     AssertTableFilterRangeConversionRejected(filtered_ro);
   }
@@ -6329,7 +6338,7 @@ TEST_P(ReadPathRangeTombstoneTest, SnapshotPredatesMemtable) {
   ASSERT_OK(Flush());
   ASSERT_OK(Put("y", "vy"));
 
-  inserted_ranges_.clear();
+  attempted_insert_ranges_.clear();
 
   ReadOptions ro;
   ro.snapshot = snap;
@@ -6402,7 +6411,7 @@ TEST_P(ReadPathRangeTombstoneTest, NoInsertionOnBlockCacheTierIncomplete) {
 
   // No range tombstone should have been inserted despite meeting threshold,
   // because the iterator terminated with Incomplete (cache miss).
-  ASSERT_EQ(inserted_ranges_.size(), 0);
+  ASSERT_EQ(attempted_insert_ranges_.size(), 0);
   ASSERT_EQ(
       options.statistics->getTickerCount(READ_PATH_RANGE_TOMBSTONES_INSERTED),
       0);
@@ -6433,7 +6442,7 @@ TEST_P(ReadPathRangeTombstoneTest, SkipInsertionWhenCoveredByExistingRange) {
     ASSERT_OK(Delete(std::string(1, c)));
   }
 
-  inserted_ranges_.clear();
+  attempted_insert_ranges_.clear();
   ReadOptions ro;
   Slice upper("z");
   ro.iterate_upper_bound = &upper;
@@ -6493,7 +6502,7 @@ TEST_P(ReadPathRangeTombstoneTest, UDTBasicScan) {
   ASSERT_EQ(keys, (std::vector<std::string>{"a", "g", "h"}));
 
   // Range covers [b+ts, g+ts).
-  ASSERT_EQ(inserted_ranges_.size(), 1);
+  ASSERT_EQ(attempted_insert_ranges_.size(), 1);
   AssertRange(0, std::string("b") + ts, std::string("g") + ts);
   ASSERT_EQ(
       options.statistics->getTickerCount(READ_PATH_RANGE_TOMBSTONES_INSERTED),
@@ -6502,7 +6511,7 @@ TEST_P(ReadPathRangeTombstoneTest, UDTBasicScan) {
 
 // Regression test: an older UDT read timestamp can hide newer live versions
 // inside a delete run. Range conversion must stay disabled in that case, or
-// the synthesized range tombstone will incorrectly hide those newer versions
+// the converted range tombstone will incorrectly hide those newer versions
 // for later max-timestamp reads.
 TEST_P(ReadPathRangeTombstoneTest, UDTOlderTimestampDisablesInsertion) {
   Options options = CurrentOptions();
@@ -6535,12 +6544,12 @@ TEST_P(ReadPathRangeTombstoneTest, UDTOlderTimestampDisablesInsertion) {
   ASSERT_OK(db_->Put(WriteOptions(), "b", ts3_slice, "vb3"));
   ASSERT_OK(db_->Put(WriteOptions(), "c", ts3_slice, "vc3"));
 
-  inserted_ranges_.clear();
+  attempted_insert_ranges_.clear();
   ReadOptions old_ro;
   old_ro.timestamp = &ts2_slice;
   VerifyIteration({"d"}, old_ro);
 
-  ASSERT_EQ(inserted_ranges_.size(), 0u);
+  ASSERT_EQ(attempted_insert_ranges_.size(), 0u);
   ASSERT_EQ(
       options.statistics->getTickerCount(READ_PATH_RANGE_TOMBSTONES_INSERTED),
       0);
@@ -6601,7 +6610,7 @@ TEST_P(ReadPathRangeTombstoneTest, ExhaustedWithUDT) {
   ASSERT_OK(iter->status());
   ASSERT_EQ(keys.size(), 4);
 
-  ASSERT_EQ(inserted_ranges_.size(), 1);
+  ASSERT_EQ(attempted_insert_ranges_.size(), 1);
   if (Forward()) {
     // Forward exhaustion: range is [e+ts, z+min_ts).
     AssertRange(0, std::string("e") + ts, std::string("z") + min_ts);
@@ -6641,7 +6650,7 @@ TEST_P(ReadPathRangeTombstoneTest, SeekForPrevTombstone) {
     }
     SetupTestData('a', 'h', /*flushed_point_dels=*/{},
                   /*memtable_point_dels=*/{"e", "f", "g", "h"}, ts_ptr);
-    inserted_ranges_.clear();
+    attempted_insert_ranges_.clear();
 
     ReadOptions ro;
     if (use_udt) {
@@ -6666,9 +6675,9 @@ TEST_P(ReadPathRangeTombstoneTest, SeekForPrevTombstone) {
 
     if (Forward()) {
       // No upper bound → trailing tombstone run has no end key → dropped.
-      ASSERT_EQ(inserted_ranges_.size(), 0);
+      ASSERT_EQ(attempted_insert_ranges_.size(), 0);
     } else {
-      ASSERT_EQ(inserted_ranges_.size(), 1);
+      ASSERT_EQ(attempted_insert_ranges_.size(), 1);
       if (use_udt) {
         std::string min_ts(sizeof(uint64_t), '\0');
         AssertRange(0, std::string("e") + ts, std::string("h") + min_ts);
@@ -6710,7 +6719,7 @@ TEST_P(ReadPathRangeTombstoneTest, UpperBoundTombstone) {
     }
     SetupTestData('a', 'h', /*flushed_point_dels=*/{},
                   /*memtable_point_dels=*/{"e", "f", "g", "h"}, ts_ptr);
-    inserted_ranges_.clear();
+    attempted_insert_ranges_.clear();
 
     ReadOptions ro;
     if (use_udt) {
@@ -6735,7 +6744,7 @@ TEST_P(ReadPathRangeTombstoneTest, UpperBoundTombstone) {
     ASSERT_OK(iter->status());
     ASSERT_EQ(keys, (std::vector<std::string>{"a", "b", "c", "d"}));
 
-    ASSERT_EQ(inserted_ranges_.size(), 1);
+    ASSERT_EQ(attempted_insert_ranges_.size(), 1);
     if (use_udt) {
       // Forward end key: upper bound padded with min_ts.
       // Reverse end key: upper bound padded with max_ts (via
@@ -6780,7 +6789,7 @@ TEST_P(ReadPathRangeTombstoneTest, LowerBoundTruncatesReverse) {
         'a', 'j', /*flushed_point_dels=*/{},
         /*memtable_point_dels=*/{"a", "b", "c", "d", "e", "f", "g", "h"},
         ts_ptr);
-    inserted_ranges_.clear();
+    attempted_insert_ranges_.clear();
 
     ReadOptions ro;
     if (use_udt) {
@@ -6809,7 +6818,7 @@ TEST_P(ReadPathRangeTombstoneTest, LowerBoundTruncatesReverse) {
     ASSERT_EQ(keys, (std::vector<std::string>{"i", "j"}));
 
     // Both directions produce one range covering tombstones e-h.
-    ASSERT_EQ(inserted_ranges_.size(), 1);
+    ASSERT_EQ(attempted_insert_ranges_.size(), 1);
     if (use_udt) {
       AssertRange(0, std::string("e") + ts, std::string("i") + ts);
     } else {
@@ -6860,11 +6869,11 @@ TEST_P(ReadPathRangeTombstoneTest,
   ASSERT_OK(Delete("c"));
 
   // Iterate at the latest sequence. Both b and c hit the reseek path and
-  // synthesize a range tombstone [b, d) for later readers at the same seq.
-  inserted_ranges_.clear();
+  // convert a range tombstone [b, d) for later readers at the same seq.
+  attempted_insert_ranges_.clear();
   VerifyIteration({"a", "d"});
 
-  ASSERT_EQ(inserted_ranges_.size(), 1);
+  ASSERT_EQ(attempted_insert_ranges_.size(), 1);
   AssertRange(0, "b", "d");
 
   // Read with the earlier snapshot. The point deletes (seq 11-12) are not
@@ -6906,17 +6915,71 @@ TEST_P(ReadPathRangeTombstoneTest, InvisibleKeysDontBreakTombstoneRun) {
 
   ReadOptions ro;
   ro.snapshot = snap;
-  inserted_ranges_.clear();
+  attempted_insert_ranges_.clear();
   VerifyIteration({"a", "f"}, ro);
 
   // Invisible keys c and e should not break the tombstone run.
   // 2 tombstones (b, d) ≥ threshold → range [b, f).
-  ASSERT_EQ(inserted_ranges_.size(), 1);
+  ASSERT_EQ(attempted_insert_ranges_.size(), 1);
   AssertRange(0, "b", "f");
 
   db_->ReleaseSnapshot(snap);
 }
 
+// Regression test: a converted tombstone can land in the current memtable at
+// an older snapshot sequence than a live point already ingested into an older
+// L0 file. Latest point lookups must continue searching that older file when
+// its sequence range can still contain a newer point version.
+TEST_P(ReadPathRangeTombstoneTest, NewerPointInOlderFileStillVisible) {
+  Options options = CurrentOptions();
+  options.min_tombstones_for_range_conversion = 2;
+  options.disable_auto_compactions = true;
+  options.statistics = CreateDBStatistics();
+  DestroyAndReopen(options);
+
+  ASSERT_OK(Put("b", "vb"));
+  ASSERT_OK(Put("c", "vc"));
+  ASSERT_OK(Put("d", "vd"));
+  ASSERT_OK(Flush());
+
+  ASSERT_OK(Delete("b"));
+  ASSERT_OK(Delete("c"));
+  ASSERT_OK(Flush());
+
+  // Keep the active memtable older than the snapshot so the read path is
+  // allowed to convert a tombstone into it later.
+  ASSERT_OK(Put("z", "vz_anchor"));
+  const Snapshot* snap = db_->GetSnapshot();
+
+  const std::string ingest_file = dbname_ + "_live_c.sst";
+  {
+    SstFileWriter writer(EnvOptions(), options);
+    ASSERT_OK(writer.Open(ingest_file));
+    ASSERT_OK(writer.Put("c", "vc_live"));
+    ASSERT_OK(writer.Finish());
+  }
+
+  IngestExternalFileOptions ifo;
+  ifo.allow_blocking_flush = false;
+  ASSERT_OK(db_->IngestExternalFile({ingest_file}, ifo));
+  ASSERT_EQ(Get("c"), "vc_live");
+
+  attempted_insert_ranges_.clear();
+  ReadOptions snap_ro;
+  snap_ro.snapshot = snap;
+  VerifyIteration({"d", "z"}, snap_ro);
+
+  ASSERT_EQ(attempted_insert_ranges_.size(), 1);
+  AssertRange(0, "b", "d");
+
+  const Snapshot* latest = db_->GetSnapshot();
+  ASSERT_EQ(Get("c", latest), "vc_live");
+  ASSERT_EQ(MultiGet({"c"}, latest), (std::vector<std::string>{"vc_live"}));
+
+  db_->ReleaseSnapshot(latest);
+  db_->ReleaseSnapshot(snap);
+}
+
 // Regression test: SeekToLast() after Seek() must clear stale saved_key_ to
 // avoid corrupting range tombstone tracking bounds.
 TEST_P(ReadPathRangeTombstoneTest, SeekToLastStaleSavedKey) {
@@ -6975,7 +7038,7 @@ TEST_P(ReadPathRangeTombstoneTest, SeekToLastTombstones) {
   ASSERT_OK(Delete("x"));
   ASSERT_OK(Delete("y"));
   auto iter = std::unique_ptr<Iterator>(db_->NewIterator(ReadOptions()));
-  inserted_ranges_.clear();
+  attempted_insert_ranges_.clear();
 
   iter->Seek("a");
   ASSERT_TRUE(iter->Valid());
@@ -6985,20 +7048,20 @@ TEST_P(ReadPathRangeTombstoneTest, SeekToLastTombstones) {
   ASSERT_TRUE(iter->Valid());
   ASSERT_OK(iter->status());
   ASSERT_EQ("z", iter->key().ToString());
-  ASSERT_EQ(inserted_ranges_.size(), 0u);
+  ASSERT_EQ(attempted_insert_ranges_.size(), 0u);
 
   // Reverse iteration skips deleted x and y.
   iter->Prev();
   ASSERT_TRUE(iter->Valid());
   ASSERT_OK(iter->status());
   ASSERT_EQ("w", iter->key().ToString());
-  ASSERT_EQ(inserted_ranges_.size(), 1u);
+  ASSERT_EQ(attempted_insert_ranges_.size(), 1u);
   AssertRange(0, "x", "z");
 }
 
 // Regression test for a crash-test pattern where the interior between two
 // point tombstones is hidden by a later DeleteRange. A latest iterator may
-// still synthesize a redundant range tombstone, but an older snapshot must
+// still convert a redundant range tombstone, but an older snapshot must
 // continue to see the pre-DeleteRange live key after iteration.
 TEST_P(ReadPathRangeTombstoneTest,
        RangeDeletedInteriorPreservesOlderSnapshots) {
@@ -7028,11 +7091,11 @@ TEST_P(ReadPathRangeTombstoneTest,
   ASSERT_OK(
       db_->DeleteRange(WriteOptions(), db_->DefaultColumnFamily(), "c", "m"));
 
-  inserted_ranges_.clear();
+  attempted_insert_ranges_.clear();
   VerifyIteration({"a", "n"});
 
   // Materialize the redundant tombstone at the latest read sequence only.
-  ASSERT_EQ(inserted_ranges_.size(), 1);
+  ASSERT_EQ(attempted_insert_ranges_.size(), 1);
   AssertRange(0, "b", "n");
 
   snap_value.clear();

db/db_test.cc

@@ -8071,6 +8071,41 @@ INSTANTIATE_TEST_CASE_P(OpenFilesAsync, OpenFilesAsyncTest,
                                            ::testing::Values(-1, 10),
                                            ::testing::Bool()));
 
+TEST_F(DBTest, ReadOnlyCloseDoesNotDeleteWriterFiles) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.disable_auto_compactions = true;
+  DestroyAndReopen(options);
+
+  ASSERT_OK(Put("before", "value"));
+  ASSERT_OK(Flush());
+
+  std::unique_ptr<DB> read_only_db;
+  ASSERT_OK(DB::OpenForReadOnly(options, dbname_, &read_only_db));
+
+  ASSERT_OK(Put("after", "value"));
+  ASSERT_OK(Flush());
+
+  std::vector<LiveFileMetaData> live_files;
+  db_->GetLiveFilesMetaData(&live_files);
+  ASSERT_GE(live_files.size(), 2);
+
+  std::string newest_sst_path;
+  uint64_t newest_file_number = 0;
+  for (const auto& live_file : live_files) {
+    if (live_file.file_number > newest_file_number) {
+      newest_file_number = live_file.file_number;
+      newest_sst_path = live_file.directory + "/" + live_file.relative_filename;
+    }
+  }
+  ASSERT_FALSE(newest_sst_path.empty());
+  ASSERT_OK(env_->FileExists(newest_sst_path));
+
+  read_only_db.reset();
+
+  ASSERT_OK(env_->FileExists(newest_sst_path));
+}
+
 // Test mix of races with async file open, reads, compactions
 TEST_P(OpenFilesAsyncTest, ConcurrentFileAccess) {
   Options options = CurrentOptions();

db/db_test2.cc

@@ -8149,6 +8149,47 @@ TEST_F(DBTest2, FastSstOpenIngestion) {
   ASSERT_OK(DestroyDB(test_dbname, options));
 }
 
+TEST_F(DBTest2, FastSstOpenDisableAfterMetadataPersisted) {
+  // Verify that disabling fast_sst_open prevents previously-persisted
+  // metadata from being passed to NewRandomAccessFile. This is critical
+  // for cases where stale metadata (e.g. expired credentials) would
+  // cause file open failures.
+  auto test_fs = std::make_shared<FastOpenTestFS>(env_->GetFileSystem());
+  std::unique_ptr<Env> test_env(new CompositeEnvWrapper(env_, test_fs));
+
+  Options options;
+  options.env = test_env.get();
+  options.fast_sst_open = true;
+  options.create_if_missing = true;
+
+  std::string test_dbname = test::PerThreadDBPath("fast_open_disable_after");
+  ASSERT_OK(DestroyDB(test_dbname, options));
+
+  // Open with fast_sst_open enabled, write and flush to persist metadata
+  std::unique_ptr<DB> db;
+  ASSERT_OK(DB::Open(options, test_dbname, &db));
+  ASSERT_OK(db->Put(WriteOptions(), "key1", "value1"));
+  ASSERT_OK(db->Flush(FlushOptions()));
+  ASSERT_EQ(1, test_fs->GetMetadataRetrievedCount());
+  db.reset();
+
+  // Reopen with fast_sst_open DISABLED — metadata is in the MANIFEST
+  // but should NOT be passed to the filesystem
+  options.fast_sst_open = false;
+  test_fs->ResetCounters();
+  ASSERT_OK(DB::Open(options, test_dbname, &db));
+
+  std::string value;
+  ASSERT_OK(db->Get(ReadOptions(), "key1", &value));
+  ASSERT_EQ("value1", value);
+
+  // No metadata should have been passed despite being in the MANIFEST
+  ASSERT_EQ(0, test_fs->GetMetadataPassedOnOpenCount());
+
+  db.reset();
+  ASSERT_OK(DestroyDB(test_dbname, options));
+}
+
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {

db/log_writer.cc

@@ -22,9 +22,10 @@ namespace ROCKSDB_NAMESPACE::log {
 
 Writer::Writer(std::unique_ptr<WritableFileWriter>&& dest, uint64_t log_number,
                bool recycle_log_files, bool manual_flush,
-               CompressionType compression_type, bool track_and_verify_wals)
+               CompressionType compression_type, bool track_and_verify_wals,
+               size_t initial_block_offset)
     : dest_(std::move(dest)),
-      block_offset_(0),
+      block_offset_(initial_block_offset),
       log_number_(log_number),
       recycle_log_files_(recycle_log_files),
       // Header size varies depending on whether we are recycling or not.

db/log_writer.h

@@ -75,15 +75,20 @@ namespace log {
 class Writer {
  public:
   // Create a writer that will append data to "*dest".
-  // "*dest" must be initially empty.
-  // "*dest" must remain live while this Writer is in use.
+  // "*dest" must remain live while this Writer is in use. By default
+  // "*dest" is expected to be empty (initial_block_offset = 0). When
+  // resuming append into an existing log file (e.g.
+  // VersionSet::ReopenManifestForAppend), pass the offset within the
+  // current 32 KiB block at which writes should resume so that record
+  // framing aligns to the existing block layout.
   // TODO(hx235): separate WAL related parameters from general `Reader`
   // parameters
   explicit Writer(std::unique_ptr<WritableFileWriter>&& dest,
                   uint64_t log_number, bool recycle_log_files,
                   bool manual_flush = false,
                   CompressionType compressionType = kNoCompression,
-                  bool track_and_verify_wals = false);
+                  bool track_and_verify_wals = false,
+                  size_t initial_block_offset = 0);
   // No copying allowed
   Writer(const Writer&) = delete;
   void operator=(const Writer&) = delete;

db/memtable.cc

@@ -918,12 +918,11 @@ void MemTable::ConstructFragmentedRangeTombstones() {
   }
 }
 
-bool MemTable::AddLogicallyRedundantRangeTombstone(SequenceNumber seq,
-                                                   const Slice& start_key,
-                                                   const Slice& end_key) {
-  // Fast path: skip if already immutable or empty. Some code paths (i.e.
-  // ExternalFileIngestion) rely ensuring memtable is empty after flushing.
-  if (is_immutable_.LoadRelaxed() || IsEmpty()) {
+bool MemTable::AddLogicallyRedundantRangeTombstone(
+    SequenceNumber seq, const Slice& start_key, const Slice& end_key,
+    port::RWMutex& ingest_sst_lock) {
+  // Fast path: skip if already immutable.
+  if (is_immutable_.LoadRelaxed()) {
     return false;
   }
 
@@ -939,6 +938,33 @@ bool MemTable::AddLogicallyRedundantRangeTombstone(SequenceNumber seq,
     return false;
   }
 
+  // Range tombstone reads have an assumption that all levels below it have a
+  // LOWER seqno than it, so it is safe to skip reading files. Normally, this is
+  // true, but range tombstone conversion creates an exception.
+  //
+  // The inserted range tombstone uses iterator seqno. There are guards to
+  // ensure that we only insert it if it is within current memtable's bounds,
+  // BUT an external file ingestion can break that still, as the newly ingested
+  // L0 file will be assigned a higher seqno than an earlier iterator.
+  //
+  // So the solution here is to use a RW lock + ingest seqno to gate range
+  // tombstone conversions. An added side effect is also we can no longer insert
+  // to memtables while a file ingestion is in progress, which is an expectation
+  // of file ingestion. Note we expect this insertion to be rare, and we do not
+  // want to limit concurrent external file ingestions so the conversion path
+  // uses a write lock while the ingestion path uses a read lock.
+  TryWriteLock ingest_wl(&ingest_sst_lock);
+  if (!ingest_wl.OwnsLock()) {
+    return false;
+  }
+  // After ingestion releases its WriteLock, an iterator with an older
+  // snapshot could still try to convert a tombstone whose seq sits
+  // below the just-ingested file's seq. The barrier persists past the end
+  // of the ingestion that bumped it and refuses such inserts.
+  if (seq < ingest_seqno_barrier_.LoadRelaxed()) {
+    return false;
+  }
+
   MemTablePostProcessInfo post_process_info;
   Status s = Add(seq, kTypeRangeDeletion, start_key, end_key,
                  nullptr /* kv_prot_info */, true /* allow_concurrent */,
@@ -950,6 +976,12 @@ bool MemTable::AddLogicallyRedundantRangeTombstone(SequenceNumber seq,
   return false;
 }
 
+void MemTable::BumpIngestSeqnoBarrier(SequenceNumber y) {
+  if (ingest_seqno_barrier_.LoadRelaxed() < y) {
+    ingest_seqno_barrier_.StoreRelaxed(y);
+  }
+}
+
 port::RWMutex* MemTable::GetLock(const Slice& key) {
   return &locks_[GetSliceRangedNPHash(key, locks_.size())];
 }

db/memtable.h

@@ -328,11 +328,13 @@ class ReadOnlyMemTable {
   // Adding a range tombstone may fail if
   // - memtable switches to immutable state
   // - a range tombstone with the same key+seq already exists (duplicate insert)
-  //
+  // - the per-memtable ingest seqno barrier already exceeds `seq` (an
+  //   ingestion has committed an L0 file at a seq that this converted
+  //   tombstone would shadow) or an ingestion is in progress.
   // Returns true if the range tombstone was inserted, false if skipped.
-  virtual bool AddLogicallyRedundantRangeTombstone(SequenceNumber /*seq*/,
-                                                   const Slice& /*start_key*/,
-                                                   const Slice& /*end_key*/) {
+  virtual bool AddLogicallyRedundantRangeTombstone(
+      SequenceNumber /*seq*/, const Slice& /*start_key*/,
+      const Slice& /*end_key*/, port::RWMutex& /*ingest_sst_lock*/) {
     return false;
   }
 
@@ -858,9 +860,19 @@ class MemTable final : public ReadOnlyMemTable {
   // SwitchMemtable() may fail.
   void ConstructFragmentedRangeTombstones();
 
-  bool AddLogicallyRedundantRangeTombstone(SequenceNumber seq,
-                                           const Slice& start_key,
-                                           const Slice& end_key) override;
+  bool AddLogicallyRedundantRangeTombstone(
+      SequenceNumber seq, const Slice& start_key, const Slice& end_key,
+      port::RWMutex& ingest_sst_lock) override;
+
+  // Monotonically raises ingest_seqno_barrier_ to `y` (no-op if `y` is not
+  // greater than the current value). The conversion's barrier check
+  // (`seq < ingest_seqno_barrier_.LoadRelaxed()`) refuses converted
+  // range tombstones that would shadow a just-installed L0 file.
+  //
+  // REQUIRES: DB mutex held by the caller. The DB mutex serializes all
+  // callers, so the load-then-store pattern is race-free without needing
+  // a CAS loop. Only IngestExternalFiles calls this.
+  void BumpIngestSeqnoBarrier(SequenceNumber y);
 
   bool IsFragmentedRangeTombstonesConstructed() const override {
     return fragmented_range_tombstone_list_.get() != nullptr ||
@@ -922,6 +934,10 @@ class MemTable final : public ReadOnlyMemTable {
   // if not set.
   std::atomic<SequenceNumber> earliest_seqno_;
 
+  // Seqno of the latest ingested external SST. See also
+  // ColumnFamilyData::ingest_sst_lock_.
+  RelaxedAtomic<SequenceNumber> ingest_seqno_barrier_{0};
+
   SequenceNumber creation_seq_;
 
   // the earliest log containing a prepared section

db/table_cache.cc

@@ -69,13 +69,14 @@ TableCache::TableCache(const ImmutableOptions& ioptions,
                        const FileOptions* file_options, Cache* const cache,
                        BlockCacheTracer* const block_cache_tracer,
                        const std::shared_ptr<IOTracer>& io_tracer,
-                       const std::string& db_session_id)
+                       const std::string& db_session_id, bool fast_sst_open)
     : ioptions_(ioptions),
       file_options_(*file_options),
       cache_(cache),
       immortal_tables_(false),
       should_pin_table_handles_(cache_.get()->GetCapacity() >=
                                 kInfiniteCapacity),
+      fast_sst_open_(fast_sst_open),
       block_cache_tracer_(block_cache_tracer),
       loader_mutex_(kLoadConcurency),
       io_tracer_(io_tracer),
@@ -107,8 +108,11 @@ Status TableCache::GetTableReader(
   fopts.file_checksum_func_name = file_meta.file_checksum_func_name;
   // Pass file open metadata for fast SST open. Use a local copy since
   // fopts.file_metadata is a non-owning pointer and file_meta is const.
+  // Only pass metadata when fast_sst_open is enabled; otherwise ignore
+  // previously persisted metadata (e.g. stale filesystem credentials).
   std::string file_open_metadata_copy;
-  if (!file_meta.file_open_metadata.empty()) {
+  if (fast_sst_open_.load(std::memory_order_relaxed) &&
+      !file_meta.file_open_metadata.empty()) {
     file_open_metadata_copy = file_meta.file_open_metadata;
     fopts.file_metadata = &file_open_metadata_copy;
     RecordTick(ioptions_.stats, FILE_OPEN_METADATA_PASSED);

db/table_cache.h

@@ -54,7 +54,7 @@ class TableCache {
              const FileOptions* storage_options, Cache* cache,
              BlockCacheTracer* const block_cache_tracer,
              const std::shared_ptr<IOTracer>& io_tracer,
-             const std::string& db_session_id);
+             const std::string& db_session_id, bool fast_sst_open = false);
   ~TableCache();
 
   // Cache interface for table cache
@@ -270,6 +270,10 @@ class TableCache {
         cache_.get()->GetCapacity() >= kInfiniteCapacity;
   }
 
+  void SetFastSstOpen(bool enabled) {
+    fast_sst_open_.store(enabled, std::memory_order_relaxed);
+  }
+
  private:
   // Build a table reader
   Status GetTableReader(const ReadOptions& ro, const FileOptions& file_options,
@@ -312,6 +316,7 @@ class TableCache {
   std::string row_cache_id_;
   bool immortal_tables_;
   bool should_pin_table_handles_;
+  std::atomic<bool> fast_sst_open_;
   BlockCacheTracer* const block_cache_tracer_;
   Striped<CacheAlignedWrapper<port::Mutex>> loader_mutex_;
   std::shared_ptr<IOTracer> io_tracer_;

db/version_edit.cc

@@ -109,6 +109,17 @@ Status FileMetaData::UpdateBoundaries(const Slice& key, const Slice& value,
 
 void VersionEdit::Clear() { *this = VersionEdit(); }
 
+bool VersionEdit::ShouldEmitPerColumnFamilyRecoveryEdit(
+    uint64_t current_log_number) const {
+  return (HasLogNumber() && GetLogNumber() > current_log_number) ||
+         NumEntries() > 0 || HasComparatorName() || HasPrevLogNumber() ||
+         HasNextFile() || HasMaxColumnFamily() || HasMinLogNumberToKeep() ||
+         HasLastSequence() || !GetCompactCursors().empty() || HasDbId() ||
+         IsColumnFamilyManipulation() || IsInAtomicGroup() ||
+         HasFullHistoryTsLow() || HasPersistUserDefinedTimestamps() ||
+         HasSubcompactionProgress();
+}
+
 bool VersionEdit::EncodeTo(std::string* dst,
                            std::optional<size_t> ts_sz) const {
   assert(!IsNoManifestWriteDummy());

db/version_edit.h

@@ -1016,6 +1016,10 @@ class VersionEdit {
     subcompaction_progress_.Clear();
   }
 
+  // Recovery-time per-column-family edits only need to be written when they
+  // advance the CF's log number or carry some other manifest state.
+  bool ShouldEmitPerColumnFamilyRecoveryEdit(uint64_t current_log_number) const;
+
   // return true on success.
   // `ts_sz` is the size in bytes for the user-defined timestamp contained in
   // a user key. This argument is optional because it's only required for

db/version_edit_handler.cc

@@ -35,6 +35,7 @@ void VersionEditHandlerBase::Iterate(log::Reader& reader,
     VersionEdit edit;
     s = edit.DecodeFrom(record);
     if (s.ok()) {
+      last_valid_record_end_ = reader.LastRecordEnd();
       s = read_buffer_.AddEdit(&edit);
     }
     if (s.ok()) {
@@ -460,6 +461,7 @@ void VersionEditHandler::CheckIterationResult(const log::Reader& reader,
   if (s->ok()) {
     version_set_->manifest_file_size_ = reader.GetReadOffset();
     assert(version_set_->manifest_file_size_ > 0);
+    version_set_->manifest_last_valid_record_end_ = last_valid_record_end_;
     version_set_->next_file_number_.store(version_edit_params_.GetNextFile() +
                                           1);
     SequenceNumber last_seq = version_edit_params_.GetLastSequence();

db/version_edit_handler.h

@@ -31,6 +31,8 @@ class VersionEditHandlerBase {
 
   AtomicGroupReadBuffer& GetReadBuffer() { return read_buffer_; }
 
+  uint64_t GetLastValidRecordEnd() const { return last_valid_record_end_; }
+
  protected:
   explicit VersionEditHandlerBase(const ReadOptions& read_options,
                                   uint64_t max_read_size)
@@ -52,6 +54,10 @@ class VersionEditHandlerBase {
 
   const ReadOptions& read_options_;
 
+  // File offset at the end of the last successfully completed and
+  // decoded logical record.
+  uint64_t last_valid_record_end_ = 0;
+
  private:
   AtomicGroupReadBuffer read_buffer_;
   const uint64_t max_manifest_read_size_;

db/version_set.cc

@@ -5599,7 +5599,7 @@ VersionSet::VersionSet(
     : column_family_set_(new ColumnFamilySet(
           dbname, _db_options, storage_options, table_cache,
           write_buffer_manager, write_controller, block_cache_tracer, io_tracer,
-          db_id, db_session_id)),
+          db_id, db_session_id, mutable_db_options.fast_sst_open)),
       table_cache_(table_cache),
       env_(_db_options->env),
       fs_(_db_options->fs, io_tracer),
@@ -5618,6 +5618,7 @@ VersionSet::VersionSet(
       prev_log_number_(0),
       current_version_number_(0),
       manifest_file_size_(0),
+      manifest_last_valid_record_end_(0),
       last_compacted_manifest_file_size_(0),
       file_options_(storage_options),
       block_cache_tracer_(block_cache_tracer),
@@ -5808,7 +5809,8 @@ void VersionSet::Reset() {
     // options.write_dbid_to_manifest is false (default).
     column_family_set_.reset(new ColumnFamilySet(
         dbname_, db_options_, file_options_, table_cache_, wbm, wc,
-        block_cache_tracer_, io_tracer_, db_id_, db_session_id_));
+        block_cache_tracer_, io_tracer_, db_id_, db_session_id_,
+        column_family_set_->GetFastSstOpen()));
   }
   db_id_.clear();
   next_file_number_.store(2);
@@ -5824,6 +5826,7 @@ void VersionSet::Reset() {
   current_version_number_ = 0;
   manifest_writers_.clear();
   manifest_file_size_ = 0;
+  manifest_last_valid_record_end_ = 0;
   last_compacted_manifest_file_size_ = 0;
   TuneMaxManifestFileSize();
   obsolete_files_.clear();
@@ -6158,11 +6161,7 @@ Status VersionSet::ProcessManifestWrites(
       }
     }
   } else {
-    FileOptions opt_file_opts = fs_->OptimizeForManifestWrite(file_options_);
-    // DB option (in file_options_) takes precedence when not kUnknown
-    if (file_options_.temperature != Temperature::kUnknown) {
-      opt_file_opts.temperature = file_options_.temperature;
-    }
+    FileOptions opt_file_opts = GetFileOptionsForManifestWrite();
     mu->Unlock();
     TEST_SYNC_POINT("VersionSet::LogAndApply:WriteManifestStart");
     TEST_SYNC_POINT_CALLBACK("VersionSet::LogAndApply:WriteManifest", nullptr);
@@ -6197,16 +6196,9 @@ Status VersionSet::ProcessManifestWrites(
       io_s = NewWritableFile(fs_.get(), descriptor_fname, &descriptor_file,
                              opt_file_opts);
       if (io_s.ok()) {
-        descriptor_file->SetPreallocationBlockSize(manifest_preallocation_size);
-        FileTypeSet tmp_set = db_options_->checksum_handoff_file_types;
-        std::unique_ptr<WritableFileWriter> file_writer(new WritableFileWriter(
-            std::move(descriptor_file), descriptor_fname, opt_file_opts, clock_,
-            io_tracer_, nullptr, Histograms::HISTOGRAM_ENUM_MAX /* hist_type */,
-            db_options_->listeners, nullptr,
-            tmp_set.Contains(FileType::kDescriptorFile),
-            tmp_set.Contains(FileType::kDescriptorFile)));
-        new_desc_log_ptr.reset(
-            new log::Writer(std::move(file_writer), 0, false));
+        new_desc_log_ptr =
+            CreateManifestWriter(std::move(descriptor_file), descriptor_fname,
+                                 opt_file_opts, manifest_preallocation_size);
         raw_desc_log_ptr = new_desc_log_ptr.get();
         s = WriteCurrentStateToManifest(write_options, curr_state,
                                         wal_additions, raw_desc_log_ptr, io_s);
@@ -6256,6 +6248,7 @@ Status VersionSet::ProcessManifestWrites(
         }
         ++idx;
 #endif /* !NDEBUG */
+        TEST_SYNC_POINT("VersionSet::ProcessManifestWrites:AddRecord");
         io_s = raw_desc_log_ptr->AddRecord(write_options, record);
         if (!io_s.ok()) {
           s = io_s;
@@ -6677,6 +6670,106 @@ Status VersionSet::LogAndApplyHelper(ColumnFamilyData* cfd,
   return builder ? builder->Apply(edit) : Status::OK();
 }
 
+FileOptions VersionSet::GetFileOptionsForManifestWrite() const {
+  FileOptions opt_file_opts = fs_->OptimizeForManifestWrite(file_options_);
+  // DB-configured temperature wins over the FS's default (when not kUnknown).
+  if (file_options_.temperature != Temperature::kUnknown) {
+    opt_file_opts.temperature = file_options_.temperature;
+  }
+  return opt_file_opts;
+}
+
+std::unique_ptr<log::Writer> VersionSet::CreateManifestWriter(
+    std::unique_ptr<FSWritableFile> file, const std::string& fname,
+    const FileOptions& opts, uint64_t preallocation_size,
+    uint64_t initial_file_size) const {
+  file->SetPreallocationBlockSize(preallocation_size);
+  FileTypeSet tmp_set = db_options_->checksum_handoff_file_types;
+  auto file_writer = std::make_unique<WritableFileWriter>(
+      std::move(file), fname, opts, clock_, io_tracer_,
+      /*stats=*/nullptr, Histograms::HISTOGRAM_ENUM_MAX, db_options_->listeners,
+      /*file_checksum_gen_factory=*/nullptr,
+      tmp_set.Contains(FileType::kDescriptorFile),
+      tmp_set.Contains(FileType::kDescriptorFile), initial_file_size);
+  const size_t block_offset = initial_file_size % log::kBlockSize;
+  return std::make_unique<log::Writer>(
+      std::move(file_writer), /*log_number=*/0, /*recycle_log_files=*/false,
+      /*manual_flush=*/false, kNoCompression,
+      /*track_and_verify_wals=*/false, block_offset);
+}
+
+Status VersionSet::ReopenManifestForAppend(const std::string& manifest_path) {
+  assert(db_options_->reuse_manifest_on_open);
+  assert(manifest_last_valid_record_end_ > 0);
+
+  // Disabled under best_efforts_recovery: that mode rebuilds the
+  // MANIFEST + CURRENT from scratch via the side-effect of
+  // LogAndApplyForRecovery emitting an edit. Reusing the (possibly
+  // stale) prior MANIFEST contradicts the salvage contract.
+  if (db_options_->best_efforts_recovery) {
+    return Status::OK();
+  }
+
+  FileOptions opt_file_opts = GetFileOptionsForManifestWrite();
+
+  // Bail if the on-disk size diverges from what Recover's Reader
+  // consumed — likely a torn tail from a prior crash; mid-block append
+  // would mis-frame the record stream.
+  uint64_t physical_size = 0;
+  IOStatus stat_s = fs_->GetFileSize(manifest_path, IOOptions(), &physical_size,
+                                     /*dbg=*/nullptr);
+  if (!stat_s.ok() || physical_size != manifest_last_valid_record_end_) {
+    ROCKS_LOG_WARN(db_options_->info_log,
+                   "reuse_manifest_on_open: physical size %" PRIu64
+                   " != last valid record end %" PRIu64
+                   " (tail corruption?); falling back to fresh MANIFEST",
+                   physical_size, manifest_last_valid_record_end_);
+    return Status::OK();
+  }
+
+  // Direct-writes path is hard to reason about for an already-populated
+  // file (alignment + tail-pad). POSIX's OptimizeForManifestWrite forces
+  // it off, but custom FileSystems may not.
+  if (opt_file_opts.use_direct_writes) {
+    ROCKS_LOG_WARN(db_options_->info_log,
+                   "reuse_manifest_on_open: direct writes enabled; "
+                   "falling back to fresh MANIFEST");
+    return Status::OK();
+  }
+
+  std::unique_ptr<FSWritableFile> manifest_file;
+  IOStatus io_s = fs_->ReopenWritableFile(manifest_path, opt_file_opts,
+                                          &manifest_file, /*dbg=*/nullptr);
+  if (!io_s.ok()) {
+    ROCKS_LOG_WARN(db_options_->info_log,
+                   "Failed to reopen MANIFEST for append: %s",
+                   io_s.ToString().c_str());
+    return Status::OK();
+  }
+
+  const uint64_t reopened_size =
+      manifest_file->GetFileSize(opt_file_opts.io_options, /*dbg=*/nullptr);
+  if (reopened_size != manifest_last_valid_record_end_) {
+    ROCKS_LOG_WARN(db_options_->info_log,
+                   "reuse_manifest_on_open: reopened handle size %" PRIu64
+                   " != last valid record end %" PRIu64
+                   "; falling back to fresh MANIFEST",
+                   reopened_size, manifest_last_valid_record_end_);
+    return Status::OK();
+  }
+
+  descriptor_log_ = CreateManifestWriter(
+      std::move(manifest_file), manifest_path, opt_file_opts,
+      manifest_preallocation_size_, reopened_size);
+
+  ROCKS_LOG_INFO(db_options_->info_log,
+                 "Reusing existing MANIFEST file: %s (valid data size: %" PRIu64
+                 ")",
+                 manifest_path.c_str(), manifest_last_valid_record_end_);
+  TEST_SYNC_POINT("VersionSet::ReopenManifestForAppend:Reopened");
+  return Status::OK();
+}
+
 Status VersionSet::Recover(
     const std::vector<ColumnFamilyDescriptor>& column_families, bool read_only,
     std::string* db_id, bool no_error_if_files_missing, bool is_retry,
@@ -6764,6 +6857,11 @@ Status VersionSet::Recover(
     }
   }
 
+  if (s.ok() && !read_only && db_options_->reuse_manifest_on_open &&
+      manifest_last_valid_record_end_ > 0) {
+    s = ReopenManifestForAppend(manifest_path);
+  }
+
   return s;
 }
 

db/version_set.h

@@ -1719,6 +1719,33 @@ class VersionSet {
       const std::unordered_map<uint32_t, MutableCFState>& curr_state,
       const VersionEdit& wal_additions, log::Writer* log, IOStatus& io_s);
 
+  // Reopen the existing MANIFEST file for append at the end of Recover()
+  // when reuse_manifest_on_open is set, so the next LogAndApply appends
+  // to it instead of creating a fresh MANIFEST. Falls back to OK
+  // (descriptor_log_ stays null, next LogAndApply creates a fresh
+  // MANIFEST as before) if ReopenWritableFile fails.
+  Status ReopenManifestForAppend(const std::string& manifest_path);
+
+  // FileOptions for MANIFEST writes — applies the FS's
+  // OptimizeForManifestWrite tuning, then re-applies the user-configured
+  // temperature so a custom FS can't override it.
+  FileOptions GetFileOptionsForManifestWrite() const;
+
+  // Create a log::Writer over the given FSWritableFile with all standard
+  // MANIFEST setup applied (preallocation block size, checksum-handoff
+  // classification, listeners, etc.). preallocation_size should be a
+  // snapshot of manifest_preallocation_size_ taken under the DB mutex
+  // (the fresh-path call site reads the field before releasing mu).
+  // When initial_file_size > 0 the writer is treated as resuming over
+  // an already-populated file: WritableFileWriter is constructed with the
+  // existing size so size accounting is correct, and the log writer's
+  // initial_block_offset aligns to the existing file's tail within the
+  // current 32 KiB block.
+  std::unique_ptr<log::Writer> CreateManifestWriter(
+      std::unique_ptr<FSWritableFile> file, const std::string& fname,
+      const FileOptions& opts, uint64_t preallocation_size,
+      uint64_t initial_file_size = 0) const;
+
   void AppendVersion(ColumnFamilyData* column_family_data, Version* v);
 
   ColumnFamilyData* CreateColumnFamily(const ColumnFamilyOptions& cf_options,
@@ -1788,6 +1815,17 @@ class VersionSet {
   // Current size of manifest file
   uint64_t manifest_file_size_;
 
+  // File offset at the end of the last successfully completed logical
+  // record during MANIFEST recovery. Unlike manifest_file_size_ (the
+  // reader's I/O high-water mark, which includes any tolerated tail
+  // garbage), this value points to the byte after the last valid record.
+  // Used by ReopenManifestForAppend to detect intra-block tail
+  // corruption that doesn't extend the physical file size.
+  // manifest_file_size_ is kept separate because it is used for
+  // rotation decisions (ProcessManifestWrites), close-time verification
+  // (Close), and backup metadata.
+  uint64_t manifest_last_valid_record_end_;
+
   // Size of the populated manifest file last time it was re-written from
   // scratch.
   uint64_t last_compacted_manifest_file_size_;

db_stress_tool/db_stress_common.h

@@ -279,7 +279,9 @@ DECLARE_bool(use_full_merge_v1);
 DECLARE_int32(sync_wal_one_in);
 DECLARE_bool(avoid_unnecessary_blocking_io);
 DECLARE_bool(write_dbid_to_manifest);
+DECLARE_bool(optimize_manifest_for_recovery);
 DECLARE_bool(write_identity_file);
+DECLARE_bool(reuse_manifest_on_open);
 DECLARE_bool(avoid_flush_during_recovery);
 DECLARE_bool(enforce_write_buffer_manager_during_recovery);
 DECLARE_uint64(max_write_batch_group_size_bytes);

db_stress_tool/db_stress_gflags.cc

@@ -1133,10 +1133,19 @@ DEFINE_bool(write_dbid_to_manifest,
             ROCKSDB_NAMESPACE::Options().write_dbid_to_manifest,
             "Write DB_ID to manifest");
 
+DEFINE_bool(optimize_manifest_for_recovery,
+            ROCKSDB_NAMESPACE::Options().optimize_manifest_for_recovery,
+            "Reduce recovery work after a clean shutdown");
+
 DEFINE_bool(write_identity_file,
             ROCKSDB_NAMESPACE::Options().write_identity_file,
             "Write DB_ID to IDENTITY file");
 
+DEFINE_bool(reuse_manifest_on_open,
+            ROCKSDB_NAMESPACE::Options().reuse_manifest_on_open,
+            "Reopen existing MANIFEST for append after recovery instead of "
+            "creating a fresh one");
+
 DEFINE_bool(avoid_flush_during_recovery,
             ROCKSDB_NAMESPACE::Options().avoid_flush_during_recovery,
             "Avoid flush during recovery");

db_stress_tool/db_stress_test_base.cc

@@ -3747,6 +3747,10 @@ void StressTest::PrintEnv() const {
           static_cast<int>(FLAGS_avoid_unnecessary_blocking_io));
   fprintf(stdout, "Write DB ID to manifest   : %d\n",
           static_cast<int>(FLAGS_write_dbid_to_manifest));
+  fprintf(stdout, "Optimize manifest recovery: %d\n",
+          static_cast<int>(FLAGS_optimize_manifest_for_recovery));
+  fprintf(stdout, "Reuse manifest on open    : %d\n",
+          static_cast<int>(FLAGS_reuse_manifest_on_open));
   fprintf(stdout, "Max Write Batch Group Size: %" PRIu64 "\n",
           FLAGS_max_write_batch_group_size_bytes);
   fprintf(stdout, "Use dynamic level         : %d\n",
@@ -4666,7 +4670,9 @@ void InitializeOptionsFromFlags(
   options.manual_wal_flush = FLAGS_manual_wal_flush_one_in > 0 ? true : false;
   options.avoid_unnecessary_blocking_io = FLAGS_avoid_unnecessary_blocking_io;
   options.write_dbid_to_manifest = FLAGS_write_dbid_to_manifest;
+  options.optimize_manifest_for_recovery = FLAGS_optimize_manifest_for_recovery;
   options.write_identity_file = FLAGS_write_identity_file;
+  options.reuse_manifest_on_open = FLAGS_reuse_manifest_on_open;
   options.avoid_flush_during_recovery = FLAGS_avoid_flush_during_recovery;
   options.enforce_write_buffer_manager_during_recovery =
       FLAGS_enforce_write_buffer_manager_during_recovery;

file/writable_file_writer.h

@@ -169,6 +169,9 @@ class WritableFileWriter {
   Temperature temperature_;
 
  public:
+  // `initial_file_size` seeds size accounting when wrapping a reopened file
+  // that will continue appending from its current end. Leave at 0 for newly
+  // created files and reuse-from-offset-0 paths.
   WritableFileWriter(
       std::unique_ptr<FSWritableFile>&& file, const std::string& _file_name,
       const FileOptions& options, SystemClock* clock = nullptr,
@@ -178,21 +181,21 @@ class WritableFileWriter {
       const std::vector<std::shared_ptr<EventListener>>& listeners = {},
       FileChecksumGenFactory* file_checksum_gen_factory = nullptr,
       bool perform_data_verification = false,
-      bool buffered_data_with_checksum = false)
+      bool buffered_data_with_checksum = false, uint64_t initial_file_size = 0)
       : file_name_(_file_name),
         writable_file_(std::move(file), io_tracer, _file_name),
         clock_(clock),
         buf_(),
         max_buffer_size_(options.writable_file_max_buffer_size),
-        filesize_(0),
-        flushed_size_(0),
-        next_write_offset_(0),
+        filesize_(initial_file_size),
+        flushed_size_(initial_file_size),
+        next_write_offset_(initial_file_size),
         pending_sync_(false),
         seen_error_(false),
 #ifndef NDEBUG
         seen_injected_error_(false),
 #endif  // NDEBUG
-        last_sync_size_(0),
+        last_sync_size_(initial_file_size),
         bytes_per_sync_(options.bytes_per_sync),
         rate_limiter_(options.rate_limiter),
         stats_(stats),

include/rocksdb/advanced_options.h

@@ -1419,8 +1419,18 @@ struct AdvancedColumnFamilyOptions {
   //   (ReadOptions::total_order_seek / ReadOptions::auto_prefix_mode) nor
   //   bounded by ReadOptions::prefix_same_as_start
   //
-  // It also requires an active mutable memtable, and insertion is skipped when
-  // that memtable is empty.
+  // Even if the above restrictions are met, there are still scenarios where a
+  // converted range tombstone may be discarded:
+  //   * The snapshot's active mutable memtable has already become immutable.
+  //   * The iterator's snapshot seq is below the active memtable's earliest
+  //     sequence number.
+  //   * A range tombstone covering [first_tombstone_key, next_live_key) is
+  //     already present in the memtable.
+  //   * A WritePrepared/WriteUnprepared transaction read callback is in use
+  //     and the snapshot seq is at or above its min uncommitted seq.
+  //   * An IngestExternalFile call is currently in flight on this column
+  //     family OR the inserted range tombstone seqno would be lower than the
+  //     ingested file seqno.
   //
   // Read-write iterators using ReadOptions::table_filter are rejected while
   // this option is enabled, see more details in ReadOptions::table_filter

include/rocksdb/options.h

@@ -1001,6 +1001,36 @@ struct DBOptions {
   // This option is mutable with SetDBOptions().
   bool verify_manifest_content_on_close = false;
 
+  // EXPERIMENTAL: If true, RocksDB can reduce recovery work after a clean
+  // shutdown, which may reduce DB::Open latency on warm reopens, especially on
+  // storage where metadata appends are expensive.
+  //
+  // Best-effort optimization: if it is disabled or unavailable, RocksDB falls
+  // back to the standard recovery path.
+  //
+  // Temporary rollout / kill switch for an optimization that is intended to be
+  // correct and eventually always enabled. Mutable via SetDBOptions().
+  bool optimize_manifest_for_recovery = false;
+
+  // EXPERIMENTAL: If true, DB::Open can try to reuse the existing MANIFEST
+  // for the first post-open metadata update instead of creating a fresh one.
+  // This can reduce warm-open latency for DBs whose MANIFEST is expensive to
+  // rebuild.
+  //
+  // Best-effort optimization: even when enabled, RocksDB may still create a
+  // fresh MANIFEST if the FileSystem does not support reopening the existing
+  // MANIFEST for append, or if RocksDB decides reuse is unsafe. That fallback
+  // is normal behavior.
+  //
+  // With very small `max_manifest_file_size` settings, the reused MANIFEST can
+  // still rotate earlier than expected after open, because RocksDB may keep a
+  // conservative auto-tuned rotation threshold until it later refreshes its
+  // compacted-size estimate.
+  //
+  // Temporary rollout / kill switch while this optimization is being
+  // validated.
+  bool reuse_manifest_on_open = false;
+
   // This option mostly replaces max_manifest_file_size to control an auto-tuned
   // balance of manifest write amplification and space amplification. A new
   // manifest file is created with the "compacted" contents of the old one when

include/rocksdb/version.h

@@ -13,7 +13,7 @@
 // minor or major version number planned for release.
 #define ROCKSDB_MAJOR 11
 #define ROCKSDB_MINOR 2
-#define ROCKSDB_PATCH 0
+#define ROCKSDB_PATCH 4
 
 // Make it easy to do conditional compilation based on version checks, i.e.
 // #if ROCKSDB_VERSION_GE(4, 5, 6)

options/db_options.cc

@@ -140,6 +140,10 @@ static std::unordered_map<std::string, OptionTypeInfo>
          {offsetof(struct MutableDBOptions, verify_manifest_content_on_close),
           OptionType::kBoolean, OptionVerificationType::kNormal,
           OptionTypeFlags::kMutable}},
+        {"optimize_manifest_for_recovery",
+         {offsetof(struct MutableDBOptions, optimize_manifest_for_recovery),
+          OptionType::kBoolean, OptionVerificationType::kNormal,
+          OptionTypeFlags::kMutable}},
         {"daily_offpeak_time_utc",
          {offsetof(struct MutableDBOptions, daily_offpeak_time_utc),
           OptionType::kString, OptionVerificationType::kNormal,
@@ -438,6 +442,10 @@ static std::unordered_map<std::string, OptionTypeInfo>
          {offsetof(struct ImmutableDBOptions, write_dbid_to_manifest),
           OptionType::kBoolean, OptionVerificationType::kNormal,
           OptionTypeFlags::kNone}},
+        {"reuse_manifest_on_open",
+         {offsetof(struct ImmutableDBOptions, reuse_manifest_on_open),
+          OptionType::kBoolean, OptionVerificationType::kNormal,
+          OptionTypeFlags::kNone}},
         {"write_identity_file",
          {offsetof(struct ImmutableDBOptions, write_identity_file),
           OptionType::kBoolean, OptionVerificationType::kNormal,
@@ -806,6 +814,7 @@ ImmutableDBOptions::ImmutableDBOptions(const DBOptions& options)
       prefix_seek_opt_in_only(options.prefix_seek_opt_in_only),
       persist_stats_to_disk(options.persist_stats_to_disk),
       write_dbid_to_manifest(options.write_dbid_to_manifest),
+      reuse_manifest_on_open(options.reuse_manifest_on_open),
       write_identity_file(options.write_identity_file),
       log_readahead_size(options.log_readahead_size),
       file_checksum_gen_factory(options.file_checksum_gen_factory),
@@ -985,6 +994,8 @@ void ImmutableDBOptions::Dump(Logger* log) const {
                    persist_stats_to_disk);
   ROCKS_LOG_HEADER(log, "                Options.write_dbid_to_manifest: %d",
                    write_dbid_to_manifest);
+  ROCKS_LOG_HEADER(log, "                Options.reuse_manifest_on_open: %d",
+                   reuse_manifest_on_open);
   ROCKS_LOG_HEADER(log, "                Options.write_identity_file: %d",
                    write_identity_file);
   ROCKS_LOG_HEADER(
@@ -1073,6 +1084,7 @@ MutableDBOptions::MutableDBOptions(const DBOptions& options)
       manifest_preallocation_size(options.manifest_preallocation_size),
       verify_manifest_content_on_close(
           options.verify_manifest_content_on_close),
+      optimize_manifest_for_recovery(options.optimize_manifest_for_recovery),
       fast_sst_open(options.fast_sst_open),
       daily_offpeak_time_utc(options.daily_offpeak_time_utc),
       max_compaction_trigger_wakeup_seconds(
@@ -1131,6 +1143,8 @@ void MutableDBOptions::Dump(Logger* log) const {
       manifest_preallocation_size);
   ROCKS_LOG_HEADER(log, "      Options.verify_manifest_content_on_close: %d",
                    verify_manifest_content_on_close);
+  ROCKS_LOG_HEADER(log, "         Options.optimize_manifest_for_recovery: %d",
+                   optimize_manifest_for_recovery);
   ROCKS_LOG_HEADER(log, "Options.daily_offpeak_time_utc: %s",
                    daily_offpeak_time_utc.c_str());
   ROCKS_LOG_HEADER(log,

options/db_options.h

@@ -88,6 +88,7 @@ struct ImmutableDBOptions {
   bool prefix_seek_opt_in_only;
   bool persist_stats_to_disk;
   bool write_dbid_to_manifest;
+  bool reuse_manifest_on_open;
   bool write_identity_file;
   size_t log_readahead_size;
   std::shared_ptr<FileChecksumGenFactory> file_checksum_gen_factory;
@@ -149,6 +150,7 @@ struct MutableDBOptions {
   int max_manifest_space_amp_pct;
   size_t manifest_preallocation_size;
   bool verify_manifest_content_on_close;
+  bool optimize_manifest_for_recovery;
   bool fast_sst_open;
   std::string daily_offpeak_time_utc;
   uint64_t max_compaction_trigger_wakeup_seconds;

options/options_helper.cc

@@ -173,6 +173,7 @@ void BuildDBOptions(const ImmutableDBOptions& immutable_db_options,
       immutable_db_options.avoid_unnecessary_blocking_io;
   options.write_dbid_to_manifest = immutable_db_options.write_dbid_to_manifest;
   options.write_identity_file = immutable_db_options.write_identity_file;
+  options.reuse_manifest_on_open = immutable_db_options.reuse_manifest_on_open;
   options.prefix_seek_opt_in_only =
       immutable_db_options.prefix_seek_opt_in_only;
   options.log_readahead_size = immutable_db_options.log_readahead_size;
@@ -192,6 +193,8 @@ void BuildDBOptions(const ImmutableDBOptions& immutable_db_options,
       immutable_db_options.enforce_single_del_contracts;
   options.verify_manifest_content_on_close =
       mutable_db_options.verify_manifest_content_on_close;
+  options.optimize_manifest_for_recovery =
+      mutable_db_options.optimize_manifest_for_recovery;
   options.daily_offpeak_time_utc = mutable_db_options.daily_offpeak_time_utc;
   options.max_compaction_trigger_wakeup_seconds =
       mutable_db_options.max_compaction_trigger_wakeup_seconds;

options/options_settable_test.cc

@@ -473,6 +473,7 @@ TEST_F(OptionsSettableTest, DBOptionsAllFieldsSettable) {
       "avoid_unnecessary_blocking_io=false;"
       "log_readahead_size=0;"
       "write_dbid_to_manifest=false;"
+      "optimize_manifest_for_recovery=false;"
       "best_efforts_recovery=false;"
       "max_bgerror_resume_count=2;"
       "bgerror_resume_retry_interval=1000000;"
@@ -489,10 +490,12 @@ TEST_F(OptionsSettableTest, DBOptionsAllFieldsSettable) {
       "wal_write_temperature=kHot;"
       "background_close_inactive_wals=true;"
       "write_dbid_to_manifest=true;"
+      "optimize_manifest_for_recovery=true;"
       "write_identity_file=true;"
       "verify_manifest_content_on_close=false;"
       "prefix_seek_opt_in_only=true;"
-      "fast_sst_open=true;",
+      "fast_sst_open=true;"
+      "reuse_manifest_on_open=true;",
       new_options));
 
   ASSERT_EQ(unset_bytes_base, NumUnsetBytes(new_options_ptr, sizeof(DBOptions),

port/port_posix.cc

@@ -161,10 +161,34 @@ void RWMutex::ReadLock() {
   PthreadCall("read lock", pthread_rwlock_rdlock(&mu_));
 }
 
+bool RWMutex::TryReadLock() {
+  int result = pthread_rwlock_tryrdlock(&mu_);
+  if (result == 0) {
+    return true;
+  }
+  if (result == EBUSY) {
+    return false;
+  }
+  PthreadCall("try read lock", result);
+  return false;
+}
+
 void RWMutex::WriteLock() {
   PthreadCall("write lock", pthread_rwlock_wrlock(&mu_));
 }
 
+bool RWMutex::TryWriteLock() {
+  int result = pthread_rwlock_trywrlock(&mu_);
+  if (result == 0) {
+    return true;
+  }
+  if (result == EBUSY) {
+    return false;
+  }
+  PthreadCall("try write lock", result);
+  return false;
+}
+
 void RWMutex::ReadUnlock() {
   PthreadCall("read unlock", pthread_rwlock_unlock(&mu_));
 }

port/port_posix.h

@@ -136,7 +136,9 @@ class RWMutex {
   ~RWMutex();
 
   void ReadLock();
+  bool TryReadLock();
   void WriteLock();
+  bool TryWriteLock();
   void ReadUnlock();
   void WriteUnlock();
   void AssertHeld() const {}

port/win/port_win.h

@@ -152,8 +152,12 @@ class RWMutex {
 
   void ReadLock() { AcquireSRWLockShared(&srwLock_); }
 
+  bool TryReadLock() { return TryAcquireSRWLockShared(&srwLock_) != 0; }
+
   void WriteLock() { AcquireSRWLockExclusive(&srwLock_); }
 
+  bool TryWriteLock() { return TryAcquireSRWLockExclusive(&srwLock_) != 0; }
+
   void ReadUnlock() { ReleaseSRWLockShared(&srwLock_); }
 
   void WriteUnlock() { ReleaseSRWLockExclusive(&srwLock_); }

tools/db_crashtest.py

@@ -482,8 +482,7 @@ default_params = {
     "auto_refresh_iterator_with_snapshot": lambda: random.choice([0, 1]),
     "memtable_op_scan_flush_trigger": lambda: random.choice([0, 10, 100, 1000]),
     "memtable_avg_op_scan_flush_trigger": lambda: random.choice([0, 2, 20, 200]),
-    # TODO(jkangs): Change back to [0, 2, 2, 4, 16] once range tombstone conversion stabilizes
-    "min_tombstones_for_range_conversion": lambda: random.choice([0]),
+    "min_tombstones_for_range_conversion": lambda: random.choice([0, 2, 2, 4, 16]),
     "ingest_wbwi_one_in": lambda: random.choice([0, 0, 100, 500]),
     "universal_reduce_file_locking": lambda: random.randint(0, 1),
     "compression_manager": lambda: random.choice(
@@ -930,6 +929,8 @@ def finalize_and_sanitize(src_params):
         # range tombstone conversion is enabled, because conversion must see
         # the full relevant SST set before synthesizing a memtable tombstone.
         dest_params["use_sqfc_for_range_queries"] = 0
+        # Delete range not compatible with inplace_update_support
+        dest_params["inplace_update_support"] = 0
     if (
         dest_params["use_direct_io_for_flush_and_compaction"] == 1
         or dest_params["use_direct_reads"] == 1
@@ -1710,9 +1711,7 @@ def collect_diagnostic_roots(base_paths, stdout, stderr):
     pruned_roots = []
     for root in sorted(roots, key=lambda path: (len(path), path)):
         if any(
-            root == existing
-            or existing == os.sep
-            or root.startswith(existing + os.sep)
+            root == existing or existing == os.sep or root.startswith(existing + os.sep)
             for existing in pruned_roots
         ):
             continue
@@ -1764,9 +1763,7 @@ def collect_directory_usage(root):
             with os.scandir(dirpath) as iterator:
                 children = sorted(list(iterator), key=lambda entry: entry.name)
         except OSError as exc:
-            errors.append(
-                (dirpath, f"failed to enumerate directory contents: {exc}")
-            )
+            errors.append((dirpath, f"failed to enumerate directory contents: {exc}"))
             return summary
 
         for child in children:
@@ -1774,9 +1771,7 @@ def collect_directory_usage(root):
                 if child.is_dir(follow_symlinks=False):
                     summary["local_dir_count"] += 1
                     child_summary = walk(child.path)
-                    summary["subtree_file_count"] += child_summary[
-                        "subtree_file_count"
-                    ]
+                    summary["subtree_file_count"] += child_summary["subtree_file_count"]
                     summary["subtree_bytes"] += child_summary["subtree_bytes"]
                     continue
 
@@ -1799,9 +1794,7 @@ def collect_directory_usage(root):
             local_usage["count"] += 1
             local_usage["bytes"] += file_size
 
-            global_usage = global_suffixes.setdefault(
-                suffix, {"count": 0, "bytes": 0}
-            )
+            global_usage = global_suffixes.setdefault(suffix, {"count": 0, "bytes": 0})
             global_usage["count"] += 1
             global_usage["bytes"] += file_size
 
@@ -1813,9 +1806,7 @@ def collect_directory_usage(root):
 
 
 def sorted_suffix_usage(suffixes):
-    return sorted(
-        suffixes.items(), key=lambda item: (-item[1]["bytes"], item[0])
-    )
+    return sorted(suffixes.items(), key=lambda item: (-item[1]["bytes"], item[0]))
 
 
 def format_directory_usage(root):

unreleased_history/behavior_changes/prefix_seek_range_tombstone_conversion.md

@@ -1 +0,0 @@
-Prefix filter changes - when seeking to a key that is out of domain, and total_order_seek is false, total_order_seek is treated as if it were true. When prefix_same_as_start = true, now iterating past a key that is out of domain invalidates the iterator. The existing behavior does not check for InDomain in the DBIter, so Transform() can produce undefined behavior (e.g. key of size 3 on FixedPrefixTransform(4)).

unreleased_history/behavior_changes/wide_column_blob_entity_v2_compatibility.md

@@ -1 +0,0 @@
-Wide-column entities with blob-backed columns now use a new V2 on-disk encoding; older RocksDB versions that do not support wide-column blob separation will reject DBs or SSTs containing those entities.

unreleased_history/bug_fixes/blob_direct_write_flush_blob_garbage.md

@@ -1 +0,0 @@
-Fix blob garbage accounting for blob direct-write flushes so flush-time filtering and overwrite elision correctly register obsolete blob bytes in blob metadata.

unreleased_history/bug_fixes/io_dispatcher_memory_leak.md

@@ -1 +0,0 @@
-Fix a memory accounting leak in IODispatcher where ReadIndex() moved block values out of ReadSet without releasing the associated prefetch memory, causing subsequent prefetches to be blocked when max_prefetch_memory_bytes was set.

unreleased_history/bug_fixes/multiscan_async_io_sync_fallback.md

@@ -1 +0,0 @@
-Fix MultiScan to fall back to synchronous coalesced reads when async I/O is unsupported at runtime.

unreleased_history/new_features/fast_sst_open.md

@@ -1 +0,0 @@
-Added experimental `DBOptions::fast_sst_open` option. When enabled, RocksDB retrieves opaque file system metadata for SST files after flush, compaction, and external file ingestion, persists it in the MANIFEST, and passes it back to the file system on subsequent file opens to accelerate DB open time.

unreleased_history/new_features/read_path_range_tombstone.md

@@ -1 +0,0 @@
-Added new option `min_tombstones_for_range_conversion` in `AdvancedColumnFamilyOptions`. When set to a non-zero value N, forward or reverse iteration will convert N or more contiguous point tombstones into a range tombstone in the mutable memtable. Future read operations will then be able to benefit from range tombstone optimizations. There are some limitations when it comes to table_filters, prefix_filters, and UDTs. See header comments for more details. 

unreleased_history/new_features/read_triggered_compaction.md

@@ -1 +0,0 @@
-Added read-triggered compaction: a new column family option `read_triggered_compaction_threshold` (default 0, disabled) that marks SST files for compaction when their read frequency (`num_collapsible_entry_reads_sampled / file_size`) exceeds the threshold. This helps reduce read amplification for frequently-read ("hot") keys. A new DB option `max_compaction_trigger_wakeup_seconds` (default 43200s / 12 hours) controls the maximum interval for periodic compaction score re-evaluation, which is necessary for this feature to work on quiet (no-write) databases.

unreleased_history/public_api_changes/compaction_filter_v4.md

@@ -1 +0,0 @@
-Added new `WideColumnBlobResolver` interface and `CompactionFilter::FilterV4()` method, including `ResolveColumn()` / `ResolveColumns()` helpers for lazy loading blob column values in wide-column entities during compaction. This allows compaction filters to resolve blob values on-demand, avoiding unnecessary I/O for blob columns they don't need to access.

unreleased_history/public_api_changes/external_table_reader_pinnable_slice.md

@@ -1 +0,0 @@
-Changed experimental feature `ExternalTableReader::Get` and `ExternalTableReader::MultiGet` to use `PinnableSlice` instead of `std::string` for output values, enabling zero-copy pinning. This will break existing implementations. 

unreleased_history/public_api_changes/sst_file_reader_pinnable_multiget.md

@@ -1 +0,0 @@
-Added `SstFileReader::Get` and `SstFileReader::MultiGet` overloads that accept `PinnableSlice`/`std::vector<PinnableSlice>*`, enabling zero-copy reads when the underlying `TableReader` supports pinning.

util/mutexlock.h

@@ -63,6 +63,30 @@ class ReadLock {
   port::RWMutex* const mu_;
 };
 
+//
+// Try to acquire a ReadLock on the specified RWMutex without blocking.
+//
+class TryReadLock {
+ public:
+  explicit TryReadLock(port::RWMutex* mu)
+      : mu_(mu), owns_(mu_->TryReadLock()) {}
+  // No copying allowed
+  TryReadLock(const TryReadLock&) = delete;
+  void operator=(const TryReadLock&) = delete;
+
+  ~TryReadLock() {
+    if (owns_) {
+      mu_->ReadUnlock();
+    }
+  }
+
+  bool OwnsLock() const { return owns_; }
+
+ private:
+  port::RWMutex* const mu_;
+  const bool owns_;
+};
+
 //
 // Automatically unlock a locked mutex when the object is destroyed
 //
@@ -97,6 +121,33 @@ class WriteLock {
   port::RWMutex* const mu_;
 };
 
+//
+// Try to acquire a WriteLock on the specified RWMutex without blocking.
+// If acquired, the lock is released when the object goes out of scope.
+// If not acquired, the destructor is a no-op. Use OwnsLock() to check
+// whether the lock was acquired.
+//
+class TryWriteLock {
+ public:
+  explicit TryWriteLock(port::RWMutex* mu)
+      : mu_(mu), owns_(mu_->TryWriteLock()) {}
+  // No copying allowed
+  TryWriteLock(const TryWriteLock&) = delete;
+  void operator=(const TryWriteLock&) = delete;
+
+  ~TryWriteLock() {
+    if (owns_) {
+      mu_->WriteUnlock();
+    }
+  }
+
+  bool OwnsLock() const { return owns_; }
+
+ private:
+  port::RWMutex* const mu_;
+  const bool owns_;
+};
+
 //
 // SpinMutex has very low overhead for low-contention cases.  Method names
 // are chosen so you can use std::unique_lock or std::lock_guard with it.