Don't let error recovery wait forever behind a writer stalled on the failed flush (#15605)

Summary:
Pull Request resolved: https://github.com/meta/rocksdb/pull/15605

**Problem:** automatic error recovery can hang forever, with all writes stopped, when a writer is waiting on a write stop (for example the `max_write_buffer_number` memtable limit) as the flush that would end it fails. Only a restart clears it.

**Why:** the stalled writer leads the write queue. It waits in `DelayWrite()` for the stop to end, and keeps waiting while recovery is in progress, so any recovery step that enters the write queue waits behind it. Two such steps exist:
1. A retryable SST write error is a soft error, and recovery retries the flush of the immutable memtables, which does not need the write queue. But with `two_write_queues = true`, `ResumeImpl()` first enters both write queues to sync `LastSequence()` with the allocated sequence numbers.
2. A hard error with auto recovery, for example a retryable error syncing the closed WALs during the flush of a DB with two column families, is recovered by an atomic flush of all column families. Its memtable switch enters the write queue in every configuration. New writes already fail with the background error at this point (`PreprocessWrite()` checks `IsDBStopped()`); only a writer that was already waiting keeps waiting.

**Fix:**
- The sequence sync at recovery start no longer waits for a write queue whose leader is in a write stall. That leader stalls before it allocates sequence numbers, and it cannot leave the stall while recovery holds the DB mutex, so the sync stays exact. If the stall ends while recovery waits for the other queue, the sync retries. Queues without a stall are still drained first, and the recovery flushes that switch memtables keep their own full fence.
- `DelayWrite()` stops waiting on a write stop once the DB is stopped, and returns the background error, like a new write does.

**Tests:**
- In `error_handler_fs_test`:
  - `RetryableFlushErrorWithWriterStalledOnFlush` (with and without `two_write_queues`): the first flush fails with a retryable SST write error after the writer stops on its memtables; the writer finishes after recovery.
  - `FlushWALSyncErrorWithWriterStalledOnFlush`: with two column families, the closed WAL sync fails after the writer stops; the writer gets the hard error, recovery completes, and the next write succeeds.
- In `write_prepared_transaction_seqno_test`:
  - `RecoverySeqnoSyncRetriesWhenStallEndsDuringNonmemJoin`: the main queue stall ends while recovery joins the nonmem queue, and the writer allocates before recovery syncs; recovery retries and syncs only after the writer publishes.
  - `RecoverySeqnoSyncSkipsStalledWalOnlyCommit`: a WAL-only commit is stalled on the nonmem queue; recovery syncs and finishes while the commit still waits.

Reviewed By: archang19

Differential Revision: D127129804

fbshipit-source-id: 88ed4938be39d7222ceb7c28437187989885b5d0

HISTORY.md

@@ -75,6 +75,7 @@ Keep cached custom-index Slice views intact when a reader consumes its input. Re
 * With `ReadOptions::pin_data`, iterators over read-only DBs opened with `max_open_files = -1` and `allow_mmap_reads = true` no longer accumulate memory as they move across SST files.
 * On a read-write primary, `DB::GetSnapshot()` and `DB::ReleaseSnapshot()` no longer acquire the DB mutex in the common case. Secondary and follower snapshot creation retains the DB mutex needed to synchronize with view installation, while release remains sharded. The snapshot list is now split into 16 per-CPU shards, each with its own lock, so primary threads creating and releasing snapshots no longer serialize on one lock or wait behind flush, compaction, or other work holding the DB mutex. Snapshot-release maintenance is deferred to existing flush, compaction, and periodic maintenance paths. Primary snapshot creation serializes with the DB-mutex-held phases of external file ingestion but can proceed while the MANIFEST write temporarily releases that mutex.
 
+
 ## 11.13.0 (10/21/2026)
 ### New Features
 * Added `FileSystem::IsPreallocationSupported()`. A `FileSystem` that never preallocates file space can return false so that DB open skips reopening and truncating the last WAL file.

db/compaction/compaction_outputs.cc

@@ -235,6 +235,32 @@ uint64_t CompactionOutputs::GetCurrentKeyGrandparentOverlappedBytes(
   return overlapped_bytes;
 }
 
+bool CompactionOutputs::PartitionerRequiresCut(
+    const CompactionIterator& c_iter) const {
+  assert(HasBuilder());
+  if (!partitioner_) {
+    return false;
+  }
+  if (partitioner_prefix_valid_) {
+    // Fast path: no per-key copy or virtual call. A partition is requested at
+    // the first key that no longer shares the established prefix. An empty
+    // prefix means no further partitions (every key "starts with" it).
+    return UNLIKELY(!c_iter.user_key().starts_with(partitioner_prefix_));
+  }
+  // Slow path: per-key callback (and copy elsewhere)
+  return UNLIKELY(partitioner_->ShouldPartition(PartitionerRequest(
+                      last_key_for_partitioner_, c_iter.user_key(),
+                      current_output_file_size_)) == kRequired);
+}
+
+bool CompactionOutputs::LargestUserKeyEquals(const Slice& user_key) {
+  const InternalKey& largest = current_output().meta.largest;
+  return largest.size() > 0 &&
+         compaction_->column_family_data()
+             ->user_comparator()
+             ->EqualWithoutTimestamp(largest.user_key(), user_key);
+}
+
 bool CompactionOutputs::ShouldStopBefore(const CompactionIterator& c_iter) {
   assert(c_iter.Valid());
   const Slice& internal_key = c_iter.key();
@@ -270,22 +296,8 @@ bool CompactionOutputs::ShouldStopBefore(const CompactionIterator& c_iter) {
     return true;
   }
 
-  if (partitioner_) {
-    if (partitioner_prefix_valid_) {
-      // Fast path: no per-key copy or virtual call. A partition is requested at
-      // the first key that no longer shares the established prefix. An empty
-      // prefix means no further partitions (every key "starts with" it).
-      if (UNLIKELY(!c_iter.user_key().starts_with(partitioner_prefix_))) {
-        return true;
-      }
-    } else {
-      // Slow path: per-key callback (and copy elsewhere)
-      if (UNLIKELY(partitioner_->ShouldPartition(PartitionerRequest(
-                       last_key_for_partitioner_, c_iter.user_key(),
-                       current_output_file_size_)) == kRequired)) {
-        return true;
-      }
-    }
+  if (PartitionerRequiresCut(c_iter)) {
+    return true;
   }
 
   // files output to Level 0 won't be split
@@ -400,13 +412,24 @@ Status CompactionOutputs::AddToOutput(
     const ParsedInternalKey& prev_iter_output_internal_key) {
   Status s;
   bool is_range_del = c_iter.IsDeleteRangeSentinelKey();
+  bool cut_before;
   if (is_range_del && compaction_->bottommost_level()) {
-    // We don't consider range tombstone for bottommost level since:
-    // 1. there is no grandparent and hence no overlap to consider
-    // 2. range tombstone may be dropped at bottommost level.
-    return s;
+    // At the bottommost level, a range tombstone is not by itself a reason to
+    // cut: there are no grandparents to limit overlap with, and the tombstone
+    // might be dropped. However, a boundary required by the SstPartitioner
+    // must still be honored; otherwise AddRangeDels() would extend the
+    // current file across it. No boundary can fall inside one user key, and
+    // a point key there may have its seqno zeroed, so that it sorts after a
+    // tombstone starting at the same user key.
+    if (!HasBuilder() || LargestUserKeyEquals(c_iter.user_key()) ||
+        !PartitionerRequiresCut(c_iter)) {
+      return s;
+    }
+    cut_before = true;
+  } else {
+    cut_before = ShouldStopBefore(c_iter) && HasBuilder();
   }
-  if (ShouldStopBefore(c_iter) && HasBuilder()) {
+  if (cut_before) {
     const Slice next_table_min_key = c_iter.key();
     s = close_file_func(c_iter.InputStatus(), prev_iter_output_internal_key,
                         next_table_min_key, &c_iter,

db/compaction/compaction_outputs.h

@@ -245,6 +245,15 @@ class CompactionOutputs {
   // before processing the current key in compaction iterator.
   bool ShouldStopBefore(const CompactionIterator& c_iter);
 
+  // Returns true iff the SST partitioner requires a new output file to start
+  // at the current key. Only meaningful while an output file is open: the
+  // partitioner state describes the current file.
+  bool PartitionerRequiresCut(const CompactionIterator& c_iter) const;
+
+  // Returns true iff the current output file has a largest key and its user
+  // key equals `user_key`, ignoring timestamps.
+  bool LargestUserKeyEquals(const Slice& user_key);
+
   // Re-establishes per-file state that depends on the first key accepted by a
   // freshly opened builder. This keeps caller-requested file cuts equivalent to
   // RocksDB's built-in file-boundary decisions for partition tracking.

db/compaction/tiered_compaction_test.cc

@@ -843,19 +843,20 @@ TEST_F(TieredCompactionTest, LevelOutofBoundaryRangeDelete) {
   ASSERT_OK(Flush());
 
   ASSERT_OK(db_->CompactRange(cro, nullptr, nullptr));
-  ASSERT_EQ("0,2,10",
-            FilesPerLevel());  // one file is at the proximal level
-                               // which only contains a range delete
+  ASSERT_EQ("0,3,10",
+            FilesPerLevel());  // the partitioner cuts before and after the
+                               // range delete, so it has a file of its own
+                               // at the proximal level
   std::vector<LiveFileMetaData> live_file_meta;
   db_->GetLiveFilesMetaData(&live_file_meta);
   bool found_sst_with_del = false;
   uint64_t sst_with_del_num = 0;
   for (const auto& meta : live_file_meta) {
     if (meta.num_deletions > 0) {
-      // found SST with del, which has 2 entries, one for data one for range del
+      // found SST with del, which only contains the range del
       ASSERT_EQ(meta.level,
                 kNumLevels - 2);  // output to proximal level
-      ASSERT_EQ(meta.num_entries, 2);
+      ASSERT_EQ(meta.num_entries, 1);
       ASSERT_EQ(meta.num_deletions, 1);
       found_sst_with_del = true;
       sst_with_del_num = meta.file_number;
@@ -974,7 +975,9 @@ TEST_F(TieredCompactionTest, UniversalRangeDelete) {
   ASSERT_OK(Flush());
 
   ASSERT_OK(db_->CompactRange(cro, nullptr, nullptr));
-  ASSERT_EQ("0,0,0,0,0,2,8", FilesPerLevel());
+  // the partitioner cuts before and after the range del, so it has a file of
+  // its own at the proximal level
+  ASSERT_EQ("0,0,0,0,0,3,8", FilesPerLevel());
   // find the SST with range del
   std::vector<LiveFileMetaData> live_file_meta;
   db_->GetLiveFilesMetaData(&live_file_meta);
@@ -982,10 +985,10 @@ TEST_F(TieredCompactionTest, UniversalRangeDelete) {
   uint64_t sst_with_del_num = 0;
   for (const auto& meta : live_file_meta) {
     if (meta.num_deletions > 0) {
-      // found SST with del, which has 2 entries, one for data one for range del
+      // found SST with del, which only contains the range del
       ASSERT_EQ(meta.level,
                 kNumLevels - 2);  // output_to_proximal_level level
-      ASSERT_EQ(meta.num_entries, 2);
+      ASSERT_EQ(meta.num_entries, 1);
       ASSERT_EQ(meta.num_deletions, 1);
       found_sst_with_del = true;
       sst_with_del_num = meta.file_number;

db/db_impl/db_impl.cc

@@ -354,6 +354,47 @@ Status DBImpl::Resume() {
   return s;
 }
 
+void DBImpl::SyncLastSequenceWithAllocatedAtRecoveryStart() {
+  mutex_.AssertHeld();
+  assert(two_write_queues_);
+  for (;;) {
+    WriteThread::Writer w;
+    const bool entered = write_thread_.EnterUnbatchedNonBlocking(&w, &mutex_);
+    const uint64_t stall =
+        entered ? 0 : write_thread_.GetBegunCountOfOutstandingStall();
+    WriteThread::Writer nonmem_w;
+    const bool nonmem_entered =
+        nonmem_write_thread_.EnterUnbatchedNonBlocking(&nonmem_w, &mutex_);
+    const uint64_t nonmem_stall =
+        nonmem_entered ? 0
+                       : nonmem_write_thread_.GetBegunCountOfOutstandingStall();
+    WaitForPendingWrites();
+
+    // A queue that was not entered must still be in the same stall, so that
+    // no writer in it can have allocated since.
+    const bool no_writer_in_flight =
+        (entered ||
+         (stall != 0 &&
+          write_thread_.GetBegunCountOfOutstandingStall() == stall)) &&
+        (nonmem_entered ||
+         (nonmem_stall != 0 &&
+          nonmem_write_thread_.GetBegunCountOfOutstandingStall() ==
+              nonmem_stall));
+    if (no_writer_in_flight) {
+      versions_->SyncLastSequenceWithAllocated();
+    }
+    if (nonmem_entered) {
+      nonmem_write_thread_.ExitUnbatched(&nonmem_w);
+    }
+    if (entered) {
+      write_thread_.ExitUnbatched(&w);
+    }
+    if (no_writer_in_flight) {
+      return;
+    }
+  }
+}
+
 // This function implements both manual and automatic background-error
 // recovery. It discards stale queued flush requests, waits for scheduled work
 // to exit, rebuilds and runs the flush work required by the recovery context,
@@ -425,16 +466,7 @@ Status DBImpl::ResumeImpl(DBRecoverContext context,
   // memtable-switch fence, after FlushAllColumnFamilies() has dropped and
   // re-acquired the DB mutex.
   if (two_write_queues_) {
-    WriteThread::Writer w;
-    write_thread_.EnterUnbatched(&w, &mutex_);
-    WriteThread::Writer nonmem_w;
-    nonmem_write_thread_.EnterUnbatched(&nonmem_w, &mutex_);
-    WaitForPendingWrites();
-
-    versions_->SyncLastSequenceWithAllocated();
-
-    nonmem_write_thread_.ExitUnbatched(&nonmem_w);
-    write_thread_.ExitUnbatched(&w);
+    SyncLastSequenceWithAllocatedAtRecoveryStart();
   }
 
   TEST_SYNC_POINT("DBImpl::ResumeImpl:AfterSyncSeq");
@@ -8037,6 +8069,7 @@ class FileIngestionHandleImpl : public FileIngestionHandle {
   FileIngestionHandleImpl& operator=(FileIngestionHandleImpl&&) = delete;
 
   Status Abort() override;
+  std::vector<SstFileMetaData> GetPreparedFiles() const override;
 
   DBImpl* const db_;
   const uint64_t first_file_number_;
@@ -8083,6 +8116,44 @@ Status FileIngestionHandleImpl::Abort() {
   return Status::OK();
 }
 
+std::vector<SstFileMetaData> FileIngestionHandleImpl::GetPreparedFiles() const {
+  std::vector<SstFileMetaData> files;
+  if (consumed_) {
+    return files;
+  }
+  for (const auto& job : jobs_) {
+    const auto& cf_paths = job->GetColumnFamilyData()->ioptions().cf_paths;
+    // Without allow_db_generated_files, commit assigns each file its sequence
+    // numbers, so they are unknown here.
+    const bool seqnos_known =
+        job->GetIngestionOptions().allow_db_generated_files;
+    for (const auto& file : job->files_to_ingest()) {
+      SstFileMetaData meta;
+      meta.directory = file.fd.GetPathId() < cf_paths.size()
+                           ? cf_paths[file.fd.GetPathId()].path
+                           : cf_paths.back().path;
+      meta.db_path = meta.directory;
+      meta.name = MakeTableFileName("", file.fd.GetNumber());
+      meta.relative_filename = meta.name.substr(1);
+      if (seqnos_known) {
+        meta.smallest_seqno = file.smallest_seqno;
+        meta.largest_seqno = file.largest_seqno;
+      }
+      meta.file_number = file.fd.GetNumber();
+      meta.size = file.fd.GetFileSize();
+      meta.temperature = file.file_temperature;
+      meta.file_checksum = file.file_checksum;
+      meta.file_checksum_func_name = file.file_checksum_func_name;
+      meta.smallestkey = file.smallest_internal_key.user_key().ToString();
+      meta.largestkey = file.largest_internal_key.user_key().ToString();
+      meta.num_entries = file.num_entries;
+      meta.num_deletions = file.table_properties.num_deletions;
+      files.push_back(std::move(meta));
+    }
+  }
+  return files;
+}
+
 Status DBImpl::PrepareFileIngestion(
     const std::vector<IngestExternalFileArg>& args,
     std::unique_ptr<FileIngestionHandle>* handle) {

db/db_impl/db_impl.h

@@ -2900,6 +2900,12 @@ class DBImpl : public DB
   // fence that is about to switch memtables.
   void MaybeSyncLastSequenceWithAllocatedForRecovery(FlushReason flush_reason);
 
+  // Like the memtable-switch fence, but does not wait for a write queue whose
+  // leader is in a write stall: that leader has not allocated sequence numbers
+  // yet, and it can be waiting for this recovery.
+  // REQUIRES: mutex locked, two_write_queues
+  void SyncLastSequenceWithAllocatedAtRecoveryStart();
+
   Status RetryFlushesForErrorRecovery(FlushReason flush_reason, bool wait);
 
   // Wait until flushing this column family won't stall writes

db/db_impl/db_impl_write.cc

@@ -2945,9 +2945,12 @@ Status DBImpl::DelayWrite(uint64_t num_bytes, WriteThread& write_thread,
     }
 
     // Don't wait if there's a background error that is not pending recovery
-    // since recovery might never be attempted.
+    // since recovery might never be attempted. Don't wait once the DB is
+    // stopped either: its recovery flush needs the write queue this writer
+    // holds.
     while ((error_handler_.GetBGError().ok() ||
-            error_handler_.IsRecoveryInProgress()) &&
+            (error_handler_.IsRecoveryInProgress() &&
+             !error_handler_.IsDBStopped())) &&
            write_controller_.IsStopped() &&
            !shutting_down_.load(std::memory_order_relaxed)) {
       if (write_options.no_slowdown) {

db/db_range_del_test.cc

@@ -1817,6 +1817,148 @@ TEST_F(DBRangeDelTest, CompactionEmitRangeTombstoneToSSTPartitioner) {
   ASSERT_EQ(2, NumTableFilesAtLevel(1));
 }
 
+namespace {
+// Records the compaction output files that were started, and how many of them
+// were empty and not kept.
+class CompactionOutputListener : public EventListener {
+ public:
+  void OnTableFileCreationStarted(
+      const TableFileCreationBriefInfo& info) override {
+    if (info.reason == TableFileCreationReason::kCompaction) {
+      std::lock_guard<std::mutex> lock(mutex_);
+      started_paths_.push_back(info.file_path);
+    }
+  }
+
+  void OnTableFileCreated(const TableFileCreationInfo& info) override {
+    if (info.reason == TableFileCreationReason::kCompaction &&
+        info.status.IsAborted()) {
+      std::lock_guard<std::mutex> lock(mutex_);
+      ++num_not_kept_;
+    }
+  }
+
+  std::vector<std::string> TakeStartedPaths() {
+    std::lock_guard<std::mutex> lock(mutex_);
+    return std::exchange(started_paths_, {});
+  }
+
+  int TakeNumNotKept() {
+    std::lock_guard<std::mutex> lock(mutex_);
+    return std::exchange(num_not_kept_, 0);
+  }
+
+ private:
+  std::mutex mutex_;
+  std::vector<std::string> started_paths_;
+  int num_not_kept_ = 0;
+};
+}  // namespace
+
+TEST_F(DBRangeDelTest, BottommostCompactionEmitRangeTombstoneToSSTPartitioner) {
+  // A range tombstone after the last point key must reach the SSTPartitioner
+  // at the bottommost level too. Otherwise the tombstone is appended to the
+  // file holding the point key and extends that file across the partition
+  // boundary, so no file starting at the boundary fits in the level.
+  // The fixed-prefix factory exercises the partitioner's prefix fast path,
+  // and SingleKeySstPartitionerFactory its per-key ShouldPartition() path.
+  // Each Key(i) is 9 bytes, so a 9-byte prefix puts every key in its own
+  // partition.
+  for (const auto& factory :
+       std::vector<std::shared_ptr<SstPartitionerFactory>>{
+           NewSstPartitionerFixedPrefixFactory(9),
+           std::make_shared<SingleKeySstPartitionerFactory>()}) {
+    SCOPED_TRACE(factory->Name());
+    Options options = CurrentOptions();
+    options.sst_partitioner_factory = factory;
+    options.disable_auto_compactions = true;
+    options.level_compaction_dynamic_level_bytes = true;
+    auto listener = std::make_shared<CompactionOutputListener>();
+    options.listeners.push_back(listener);
+    DestroyAndReopen(options);
+    const int bottommost_level = options.num_levels - 1;
+
+    ASSERT_OK(Put(Key(1), "val"));
+    // Keeps the tombstone from being dropped at the bottommost level
+    const Snapshot* snapshot = db_->GetSnapshot();
+    ASSERT_OK(db_->DeleteRange(WriteOptions(), db_->DefaultColumnFamily(),
+                               Key(2), Key(5)));
+    ASSERT_OK(Flush());
+
+    CompactRangeOptions cro;
+    cro.bottommost_level_compaction = BottommostLevelCompaction::kForce;
+    ASSERT_OK(db_->CompactRange(cro, nullptr, nullptr));
+
+    ColumnFamilyMetaData cf_meta;
+    db_->GetColumnFamilyMetaData(&cf_meta);
+    const std::vector<SstFileMetaData>& files =
+        cf_meta.levels[bottommost_level].files;
+    ASSERT_EQ(2, files.size());
+    ASSERT_EQ(Key(1), files[0].largestkey);
+    ASSERT_EQ(Key(2), files[1].smallestkey);
+    ASSERT_EQ(0, listener->TakeNumNotKept());
+    listener->TakeStartedPaths();
+
+    // Once nothing pins it, the tombstone is dropped. The file cut for it is
+    // still created, but it is empty, so the compaction deletes it before
+    // installing its outputs: it is never part of the LSM.
+    db_->ReleaseSnapshot(snapshot);
+    ASSERT_OK(db_->CompactRange(cro, nullptr, nullptr));
+    ASSERT_EQ(1, NumTableFilesAtLevel(bottommost_level));
+    ASSERT_EQ(1, listener->TakeNumNotKept());
+    const std::vector<std::string> started = listener->TakeStartedPaths();
+    ASSERT_EQ(2, started.size());
+    int num_on_disk = 0;
+    for (const auto& path : started) {
+      const Status exists = env_->FileExists(path);
+      if (exists.ok()) {
+        ++num_on_disk;
+      } else {
+        ASSERT_TRUE(exists.IsNotFound()) << exists.ToString();
+      }
+    }
+    ASSERT_EQ(1, num_on_disk);
+  }
+}
+
+TEST_F(DBRangeDelTest, BottommostRangeTombstoneAtSameUserKeyAsPointKey) {
+  // At the bottommost level a point key's seqno can be zeroed, so it sorts
+  // after a range tombstone that starts at the same user key. The partitioner
+  // must not cut between them, even one that requires a cut for every key.
+  Options options = CurrentOptions();
+  options.sst_partitioner_factory =
+      std::make_shared<SingleKeySstPartitionerFactory>();
+  options.disable_auto_compactions = true;
+  options.level_compaction_dynamic_level_bytes = true;
+  CompactRangeOptions cro;
+  cro.bottommost_level_compaction = BottommostLevelCompaction::kForce;
+
+  // The tombstone is dropped.
+  DestroyAndReopen(options);
+  ASSERT_OK(Put(Key(1), "val"));
+  ASSERT_OK(db_->DeleteRange(WriteOptions(), db_->DefaultColumnFamily(), Key(2),
+                             Key(5)));
+  ASSERT_OK(Put(Key(2), "val"));
+  ASSERT_OK(Flush());
+  ASSERT_OK(db_->CompactRange(cro, nullptr, nullptr));
+  ASSERT_EQ("val", Get(Key(2)));
+
+  // A newer tombstone covering the same user key is kept by a snapshot, so it
+  // is written to every output file that overlaps it.
+  DestroyAndReopen(options);
+  ASSERT_OK(db_->DeleteRange(WriteOptions(), db_->DefaultColumnFamily(), Key(2),
+                             Key(5)));
+  ASSERT_OK(Put(Key(2), "val"));
+  ASSERT_OK(Flush());
+  const Snapshot* snapshot = db_->GetSnapshot();
+  ASSERT_OK(db_->DeleteRange(WriteOptions(), db_->DefaultColumnFamily(), Key(1),
+                             Key(4)));
+  ASSERT_OK(db_->CompactRange(cro, nullptr, nullptr));
+  ASSERT_EQ("NOT_FOUND", Get(Key(2)));
+  ASSERT_EQ("val", Get(Key(2), snapshot));
+  db_->ReleaseSnapshot(snapshot);
+}
+
 TEST_F(DBRangeDelTest, OversizeCompactionGapBetweenPointKeyAndTombstone) {
   // L2 has 2 files
   // L2_0: 0, 1, 2, 3, 4

db/error_handler_fs_test.cc

@@ -413,6 +413,113 @@ TEST_F(DBErrorHandlingFSTest, FlushWriteFileScopeError) {
   }
 }
 
+// A retryable SST write error during a flush is a soft error that recovery
+// fixes by retrying the flush. A writer that waits for that flush must finish
+// once recovery is done.
+TEST_F(DBErrorHandlingFSTest, RetryableFlushErrorWithWriterStalledOnFlush) {
+  for (bool two_write_queues : {false, true}) {
+    SCOPED_TRACE("two_write_queues=" + std::to_string(two_write_queues));
+    Options options = GetDefaultOptions();
+    options.env = fault_env_.get();
+    options.create_if_missing = true;
+    options.two_write_queues = two_write_queues;
+    options.write_buffer_size = 64 << 10;
+    options.max_write_buffer_number = 2;
+    options.min_write_buffer_number_to_merge = 1;
+    options.bgerror_resume_retry_interval = 100000;
+    DestroyAndReopen(options);
+
+    IOStatus error_msg = IOStatus::IOError("Retryable IO Error");
+    error_msg.SetRetryable(true);
+    std::atomic<bool> injected{false};
+    // The first flush fails only after the writer stops on its memtables.
+    SyncPoint::GetInstance()->LoadDependency(
+        {{"DBImpl::DelayWrite:Wait", "BuildTable:BeforeFinishBuildTable"}});
+    SyncPoint::GetInstance()->SetCallBack(
+        "BuildTable:BeforeFinishBuildTable", [&](void*) {
+          if (!injected.exchange(true)) {
+            fault_fs_->SetFilesystemActive(false, error_msg);
+          }
+        });
+    SyncPoint::GetInstance()->SetCallBack(
+        "RecoverFromRetryableBGIOError:BeforeStart",
+        [&](void*) { fault_fs_->SetFilesystemActive(true); });
+    SyncPoint::GetInstance()->EnableProcessing();
+
+    const std::string value(40 << 10, 'v');
+    for (int i = 0; i < 8; ++i) {
+      ASSERT_OK(Put(Key(i), value));
+    }
+    ASSERT_TRUE(injected);
+    SyncPoint::GetInstance()->DisableProcessing();
+    SyncPoint::GetInstance()->ClearAllCallBacks();
+    SyncPoint::GetInstance()->LoadDependency({});
+    for (int i = 0; i < 8; ++i) {
+      ASSERT_EQ(value, Get(Key(i)));
+    }
+  }
+}
+
+// A retryable error syncing closed WALs during a flush is a hard error. A
+// writer that waits for that flush gets the error, and recovery then flushes
+// and resumes writes.
+TEST_F(DBErrorHandlingFSTest, FlushWALSyncErrorWithWriterStalledOnFlush) {
+  Options options = GetDefaultOptions();
+  options.env = fault_env_.get();
+  options.create_if_missing = true;
+  options.write_buffer_size = 64 << 10;
+  options.max_write_buffer_number = 2;
+  options.min_write_buffer_number_to_merge = 1;
+  options.bgerror_resume_retry_interval = 100000;
+  // With two column families, a flush syncs the closed WALs.
+  CreateAndReopenWithCF({"pikachu"}, options);
+
+  IOStatus error_msg = IOStatus::IOError("Retryable IO Error");
+  error_msg.SetRetryable(true);
+  std::atomic<bool> injected{false};
+  // The first flush fails only after the writer stops on its memtables.
+  SyncPoint::GetInstance()->LoadDependency(
+      {{"DBImpl::DelayWrite:Wait", "DBImpl::SyncClosedWals:Start"},
+       {"RecoverFromRetryableBGIOError:RecoverSuccess",
+        "DBErrorHandlingFSTest::FlushWALSyncErrorWithWriterStalledOnFlush:"
+        "Recovered"}});
+  SyncPoint::GetInstance()->SetCallBack(
+      "DBImpl::SyncClosedWals:Start", [&](void*) {
+        if (!injected.exchange(true)) {
+          fault_fs_->SetFilesystemActive(false, error_msg);
+        }
+      });
+  SyncPoint::GetInstance()->SetCallBack(
+      "RecoverFromRetryableBGIOError:BeforeStart",
+      [&](void*) { fault_fs_->SetFilesystemActive(true); });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  const std::string value(40 << 10, 'v');
+  Status s;
+  int num_written = 0;
+  while (num_written < 8) {
+    s = Put(Key(num_written), value);
+    if (!s.ok()) {
+      break;
+    }
+    ++num_written;
+  }
+  ASSERT_TRUE(injected);
+  ASSERT_TRUE(s.IsIOError()) << s.ToString();
+  ASSERT_EQ(s.severity(), Status::Severity::kHardError);
+
+  TEST_SYNC_POINT(
+      "DBErrorHandlingFSTest::FlushWALSyncErrorWithWriterStalledOnFlush:"
+      "Recovered");
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+  SyncPoint::GetInstance()->LoadDependency({});
+  ASSERT_OK(Put(Key(num_written), value));
+  for (int i = 0; i <= num_written; ++i) {
+    ASSERT_EQ(value, Get(Key(i)));
+  }
+}
+
 TEST_F(DBErrorHandlingFSTest, FlushWALWriteRetryableError) {
   std::shared_ptr<ErrorHandlerFSListener> listener =
       std::make_shared<ErrorHandlerFSListener>();

db/external_sst_file_ingestion_job.h

@@ -297,6 +297,10 @@ class ExternalSstFileIngestionJob {
 
   ColumnFamilyData* GetColumnFamilyData() const { return cfd_; }
 
+  const IngestExternalFileOptions& GetIngestionOptions() const {
+    return ingestion_options_;
+  }
+
   // Prepare the job by copying external files into the DB.
   Status Prepare(const std::vector<std::string>& external_files_paths,
                  const std::vector<std::string>& files_checksums,

db/external_sst_file_test.cc

@@ -8,6 +8,7 @@
 #include <atomic>
 #include <functional>
 #include <memory>
+#include <set>
 #include <sstream>
 
 #include "db/db_test_util.h"
@@ -846,6 +847,189 @@ TEST_F(ExternalSSTFileTest, AbortPreparedIngestion) {
   ASSERT_EQ("1", Get("x"));
 }
 
+static std::string DescribeFile(const SstFileMetaData& meta) {
+  return meta.directory + "/" + meta.relative_filename + " #" +
+         std::to_string(meta.file_number) + " " + std::to_string(meta.size) +
+         "B " + meta.file_checksum_func_name + ":" +
+         Slice(meta.file_checksum).ToString(/*hex=*/true) + " [" +
+         meta.smallestkey + ", " + meta.largestkey + "] temperature " +
+         std::to_string(static_cast<int>(meta.temperature));
+}
+
+static std::set<std::string> DescribeFiles(
+    const std::vector<SstFileMetaData>& files) {
+  std::set<std::string> described;
+  for (const auto& meta : files) {
+    described.insert(DescribeFile(meta));
+  }
+  return described;
+}
+
+static std::set<std::string> DescribeLiveFiles(DB* db) {
+  std::vector<LiveFileMetaData> live;
+  db->GetLiveFilesMetaData(&live);
+  return DescribeFiles({live.begin(), live.end()});
+}
+
+TEST_F(ExternalSSTFileTest, PreparedFilesAreTheFilesThatGoLive) {
+  Options options = CurrentOptions();
+  options.file_checksum_gen_factory = GetFileChecksumGenCrc32cFactory();
+  DestroyAndReopen(options);
+
+  std::string file_a;
+  ASSERT_OK(
+      GenerateExternalFileOnly(options, {{"a1", "1"}, {"a2", "2"}}, &file_a));
+  const std::string file_b = sst_files_dir_ + "put_and_delete.sst";
+  SstFileWriter writer(EnvOptions(), options);
+  ASSERT_OK(writer.Open(file_b));
+  ASSERT_OK(writer.Put("b", "2"));
+  ASSERT_OK(writer.Delete("c"));
+  ASSERT_OK(writer.Finish());
+  IngestExternalFileArg arg;
+  arg.column_family = db_->DefaultColumnFamily();
+  arg.external_files = {file_a, file_b};
+
+  std::unique_ptr<FileIngestionHandle> handle;
+  ASSERT_OK(db_->PrepareFileIngestion({arg}, &handle));
+  const std::vector<SstFileMetaData> prepared = handle->GetPreparedFiles();
+
+  ASSERT_EQ(2, prepared.size());
+  for (const auto& meta : prepared) {
+    const std::string path = meta.directory + "/" + meta.relative_filename;
+    uint64_t size_on_disk = 0;
+    ASSERT_OK(env_->GetFileSize(path, &size_on_disk));
+    ASSERT_EQ(size_on_disk, meta.size);
+    ASSERT_EQ("FileChecksumCrc32c", meta.file_checksum_func_name);
+    ASSERT_FALSE(meta.file_checksum.empty());
+  }
+  std::set<std::string> key_ranges;
+  for (const auto& meta : prepared) {
+    key_ranges.insert(meta.smallestkey + ".." + meta.largestkey + " " +
+                      std::to_string(meta.num_entries) + " entries " +
+                      std::to_string(meta.num_deletions) + " deletions");
+  }
+  ASSERT_EQ((std::set<std::string>{"a1..a2 2 entries 0 deletions",
+                                   "b..c 2 entries 1 deletions"}),
+            key_ranges);
+  ASSERT_EQ("NOT_FOUND", Get("a1"));
+
+  ASSERT_OK(db_->CommitFileIngestionHandle(std::move(handle)));
+
+  ASSERT_EQ(DescribeFiles(prepared), DescribeLiveFiles(db_.get()));
+}
+
+TEST_F(ExternalSSTFileTest, PreparedDbGeneratedFilesCarryTheirSeqnos) {
+  Options options = CurrentOptions();
+  DestroyAndReopen(options);
+
+  const std::string source_dbname = dbname_ + "_source";
+  ASSERT_OK(DestroyDB(source_dbname, options));
+  std::unique_ptr<DB> source_db;
+  ASSERT_OK(DB::Open(options, source_dbname, &source_db));
+  ASSERT_OK(source_db->Put(WriteOptions(), Key(1), "v1"));
+  ASSERT_OK(source_db->Put(WriteOptions(), Key(2), "v2"));
+  ASSERT_OK(source_db->Flush(FlushOptions()));
+  std::vector<LiveFileMetaData> source_meta;
+  source_db->GetLiveFilesMetaData(&source_meta);
+  ASSERT_EQ(1, source_meta.size());
+  ASSERT_LT(source_meta[0].smallest_seqno, source_meta[0].largest_seqno);
+
+  IngestExternalFileArg arg;
+  arg.column_family = db_->DefaultColumnFamily();
+  arg.external_files = {source_meta[0].directory + "/" +
+                        source_meta[0].relative_filename};
+  arg.options.allow_db_generated_files = true;
+  arg.options.snapshot_consistency = false;
+
+  std::unique_ptr<FileIngestionHandle> handle;
+  ASSERT_OK(db_->PrepareFileIngestion({arg}, &handle));
+  const std::vector<SstFileMetaData> prepared = handle->GetPreparedFiles();
+  ASSERT_EQ(1, prepared.size());
+  ASSERT_EQ(source_meta[0].smallest_seqno, prepared[0].smallest_seqno);
+  ASSERT_EQ(source_meta[0].largest_seqno, prepared[0].largest_seqno);
+
+  ASSERT_OK(db_->CommitFileIngestionHandle(std::move(handle)));
+
+  std::vector<LiveFileMetaData> live;
+  db_->GetLiveFilesMetaData(&live);
+  ASSERT_EQ(1, live.size());
+  ASSERT_EQ(live[0].smallest_seqno, prepared[0].smallest_seqno);
+  ASSERT_EQ(live[0].largest_seqno, prepared[0].largest_seqno);
+  ASSERT_EQ(live[0].name, prepared[0].name);
+  ASSERT_EQ(live[0].db_path, prepared[0].db_path);
+
+  ASSERT_OK(source_db->Close());
+  source_db.reset();
+  ASSERT_OK(DestroyDB(source_dbname, options));
+}
+
+TEST_F(ExternalSSTFileTest, PreparedFilesAreGoneAfterAbort) {
+  Options options = CurrentOptions();
+  DestroyAndReopen(options);
+
+  std::string file_path;
+  ASSERT_OK(GenerateExternalFileOnly(options, {{"x", "1"}}, &file_path));
+  IngestExternalFileArg arg;
+  arg.column_family = db_->DefaultColumnFamily();
+  arg.external_files = {file_path};
+
+  std::unique_ptr<FileIngestionHandle> handle;
+  ASSERT_OK(db_->PrepareFileIngestion({arg}, &handle));
+  const std::vector<SstFileMetaData> prepared = handle->GetPreparedFiles();
+  ASSERT_EQ(1, prepared.size());
+  const std::string path =
+      prepared[0].directory + "/" + prepared[0].relative_filename;
+  ASSERT_OK(env_->FileExists(path));
+
+  ASSERT_OK(handle->Abort());
+
+  ASSERT_TRUE(handle->GetPreparedFiles().empty());
+  ASSERT_TRUE(env_->FileExists(path).IsNotFound());
+}
+
+TEST_F(ExternalSSTFileTest, PreparedFilesOfMergedHandlesGoLive) {
+  Options options = CurrentOptions();
+  CreateAndReopenWithCF({"cf1", "cf2", "cf3"}, options);
+
+  std::string f_cf1;
+  std::string f_cf2_h1;
+  std::string f_cf2_h2;
+  std::string f_cf3;
+  ASSERT_OK(GenerateExternalFileOnly(options, {{"a", "1"}}, &f_cf1));
+  ASSERT_OK(GenerateExternalFileOnly(options, {{"b", "h1"}}, &f_cf2_h1));
+  ASSERT_OK(GenerateExternalFileOnly(options, {{"b", "h2"}}, &f_cf2_h2));
+  ASSERT_OK(GenerateExternalFileOnly(options, {{"c", "3"}}, &f_cf3));
+
+  // Each handle spans two column families, and both write cf2, so commit
+  // merges the two cf2 jobs.
+  std::vector<IngestExternalFileArg> args1(2);
+  args1[0].column_family = handles_[1];
+  args1[0].external_files = {f_cf1};
+  args1[1].column_family = handles_[2];
+  args1[1].external_files = {f_cf2_h1};
+  std::vector<IngestExternalFileArg> args2(2);
+  args2[0].column_family = handles_[2];
+  args2[0].external_files = {f_cf2_h2};
+  args2[1].column_family = handles_[3];
+  args2[1].external_files = {f_cf3};
+
+  std::unique_ptr<FileIngestionHandle> h1;
+  std::unique_ptr<FileIngestionHandle> h2;
+  ASSERT_OK(db_->PrepareFileIngestion(args1, &h1));
+  ASSERT_OK(db_->PrepareFileIngestion(args2, &h2));
+  std::set<std::string> prepared = DescribeFiles(h1->GetPreparedFiles());
+  prepared.merge(DescribeFiles(h2->GetPreparedFiles()));
+  ASSERT_EQ(4, prepared.size());
+
+  std::vector<std::unique_ptr<FileIngestionHandle>> batch;
+  batch.push_back(std::move(h1));
+  batch.push_back(std::move(h2));
+  ASSERT_OK(db_->CommitFileIngestionHandles(std::move(batch)));
+
+  ASSERT_EQ("h2", Get(2, "b"));
+  ASSERT_EQ(prepared, DescribeLiveFiles(db_.get()));
+}
+
 TEST_F(ExternalSSTFileTest, MultiHandleAtomicCommit) {
   Options options = CurrentOptions();
   CreateAndReopenWithCF({"cf1", "cf2", "cf3"}, options);

include/rocksdb/db.h

@@ -192,6 +192,37 @@ class FileIngestionHandle {
   // returns a Status.
   virtual Status Abort() = 0;
 
+  // Metadata of the ingested files this handle will make visible on commit.
+  // The files exist from prepare until commit or abort and keep their path,
+  // number, size and checksum when committed, so a caller can read them
+  // before they go live. Empty once the handle is spent.
+  //
+  // Set fields:
+  //   directory
+  //   relative_filename
+  //   name
+  //   db_path
+  //   file_number
+  //   size
+  //   temperature
+  //   file_checksum
+  //   file_checksum_func_name
+  //   smallestkey
+  //   largestkey
+  //   num_entries
+  //   num_deletions
+  //   smallest_seqno, largest_seqno: only with allow_db_generated_files;
+  //     otherwise commit assigns them
+  //
+  // Left at defaults (commit decides them):
+  //   epoch_number
+  //   file_creation_time
+  //   oldest_ancestor_time
+  //
+  // Does not include the range-tombstone file that an atomic_replace_range
+  // ingestion may add at commit.
+  virtual std::vector<SstFileMetaData> GetPreparedFiles() const = 0;
+
   FileIngestionHandle(const FileIngestionHandle&) = delete;
   FileIngestionHandle& operator=(const FileIngestionHandle&) = delete;
   FileIngestionHandle(FileIngestionHandle&&) = delete;

include/rocksdb/version.h

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

table/table_test.cc

@@ -8680,11 +8680,15 @@ TEST_F(ExternalTableTest, FactoryWithoutTableBuilderIsNeverCompacted) {
     return names;
   };
   std::vector<std::string> ingested[3];
+  // Otherwise the FIFO compaction of handles_[1] can delete its file before
+  // the check below sees it.
+  ASSERT_OK(db_->PauseBackgroundWork());
   for (int cf : {1, 2}) {
     ASSERT_OK(db_->IngestExternalFile(handles_[cf], {file}, ingest_options));
     ingested[cf] = live_files(cf);
     ASSERT_EQ(ingested[cf].size(), 1);
   }
+  ASSERT_OK(db_->ContinueBackgroundWork());
   ASSERT_OK(dbfull()->TEST_WaitForCompact());
   ASSERT_TRUE(live_files(1).empty());
   ASSERT_EQ(live_files(2), ingested[2]);

tools/check_format_compatible.sh

@@ -161,7 +161,7 @@ EOF
 
 # To check for DB forward compatibility with loading options (old version
 # reading data from new), as well as backward compatibility
-declare -a db_forward_with_options_refs=("10.4.fb" "10.5.fb" "10.6.fb" "10.7.fb" "10.8.fb" "10.9.fb" "10.10.fb" "10.11.fb" "11.0.fb" "11.1.fb" "11.2.fb" "11.3.fb" "11.4.fb" "11.5.fb" "11.6.fb" "11.7.fb" "11.8.fb" "11.9.fb" "11.10.fb" "11.11.fb" "11.12.fb" "11.13.fb")
+declare -a db_forward_with_options_refs=("10.4.fb" "10.5.fb" "10.6.fb" "10.7.fb" "10.8.fb" "10.9.fb" "10.10.fb" "10.11.fb" "11.0.fb" "11.1.fb" "11.2.fb" "11.3.fb" "11.4.fb" "11.5.fb" "11.6.fb" "11.7.fb" "11.8.fb" "11.9.fb" "11.10.fb" "11.11.fb" "11.12.fb" "11.13.fb" "11.14.fb")
 # To check for DB forward compatibility without loading options (in addition
 # to the "with loading options" set), as well as backward compatibility
 declare -a db_forward_no_options_refs=() # N/A at the moment

unreleased_history/bug_fixes/bottommost_range_del_partitioner_cut.md

@@ -0,0 +1 @@
+Fixed a bug where a bottommost-level compaction could ignore an `SstPartitioner` boundary between the last point key of an output file and a following range tombstone kept alive by a snapshot. The tombstone extended that file across the boundary, so an `IngestExternalFile()` with `fail_if_not_bottommost_level=true` into the range starting at the boundary could fail with `TryAgain`.

unreleased_history/bug_fixes/error_recovery_behind_stalled_writer.md

@@ -0,0 +1 @@
+Fixed automatic error recovery hanging forever, with all writes stopped, when a writer was waiting on a write stall (for example the `max_write_buffer_number` memtable limit) as a flush failed. This happened for a retryable SST write error with `two_write_queues = true`, and for a hard error such as a failed WAL sync during the flush in any configuration. A writer that is waiting when the DB stops now returns the background error, like a new write does.

unreleased_history/new_features/prepared_files.md

@@ -0,0 +1 @@
+Added `FileIngestionHandle::GetPreparedFiles()`: the metadata (path, file number, size, checksum, key range, entry and deletion counts) of a prepared ingestion's files. The files keep their path, number, size and checksum when the handle is committed, so a caller can read the files before they become visible. The list does not include the range-tombstone file that an `atomic_replace_range` ingestion may add at commit.

utilities/transactions/write_prepared_transaction_seqno_test.cc

@@ -238,6 +238,160 @@ TEST_F(WritePreparedTransactionSeqnoTest,
   Close();
 }
 
+// Recovery does not wait for a main write queue whose leader is in a write
+// stall. If that stall ends while recovery is joining the nonmem queue, the
+// leader can allocate a sequence range before recovery syncs. Recovery must
+// then retry and wait for the leader to publish, instead of publishing the
+// leader's allocation for it.
+TEST_F(WritePreparedTransactionSeqnoTest,
+       RecoverySeqnoSyncRetriesWhenStallEndsDuringNonmemJoin) {
+  txn_db_options_.write_policy = TxnDBWritePolicy::WRITE_UNPREPARED;
+  ASSERT_OK(Open());
+  TransactionDB& db = AssertNotNull(db_);
+  DBImpl& db_impl =
+      AssertNotNull(static_cast_with_check<DBImpl>(db.GetRootDB()));
+  VersionSet& versions = AssertNotNull(db_impl.GetVersionSet());
+  SyncPoint& sync_point = AssertNotNull(SyncPoint::GetInstance());
+
+  const std::string test_prefix =
+      "RecoverySeqnoSyncRetriesWhenStallEndsDuringNonmemJoin";
+  sync_point.LoadDependency(
+      {{"DBImpl::DelayWrite:Wait", test_prefix + ":writer_stalled"},
+       {"DBImpl::WriteImpl:AfterWBWIIngestBeforeSetLastSequence:pause",
+        test_prefix + ":writer_paused"},
+       {test_prefix + ":release_writer",
+        "DBImpl::WriteImpl:AfterWBWIIngestBeforeSetLastSequence:resume"}});
+
+  std::unique_ptr<WriteControllerToken> stop_token =
+      db_impl.TEST_write_controler().GetStopToken();
+  std::atomic<int> non_blocking_joins{0};
+  std::atomic<bool> release_writer_called{false};
+  std::atomic<bool> recovery_waited_for_writer{false};
+  std::atomic<bool> writer_published_before_sync{false};
+  auto release_writer = [&]() {
+    if (!release_writer_called.exchange(true)) {
+      sync_point.Process(test_prefix + ":release_writer");
+    }
+  };
+  // The first non-blocking join is refused by the main queue stall. The second
+  // one joins the nonmem queue without the DB mutex. End the stall there and
+  // let the stalled writer allocate and pause before it publishes.
+  sync_point.SetCallBack("WriteThread::EnterUnbatchedNonBlocking:BeforeLock",
+                         [&](void*) {
+                           if (non_blocking_joins.fetch_add(1) == 1) {
+                             stop_token.reset();
+                             db_impl.TEST_SignalAllBgCv();
+                             sync_point.Process(test_prefix + ":writer_paused");
+                           }
+                         });
+  // On retry, recovery queues behind the paused writer in the main queue.
+  sync_point.SetCallBack("WriteThread::EnterUnbatched:Wait", [&](void*) {
+    recovery_waited_for_writer.store(true);
+    release_writer();
+  });
+  // Fallback so a path that syncs without retrying fails instead of hanging.
+  sync_point.SetCallBack("DBImpl::ResumeImpl:AfterSyncSeq", [&](void*) {
+    writer_published_before_sync.store(release_writer_called.load());
+    release_writer();
+  });
+  sync_point.EnableProcessing();
+
+  WriteOptions write_opts;
+  TransactionOptions txn_opts;
+  Status prepare_status;
+  std::unique_ptr<Transaction> stalled_txn(
+      db.BeginTransaction(write_opts, txn_opts));
+  Transaction& stalled_txn_ref = AssertNotNull(stalled_txn.get());
+  ASSERT_OK(stalled_txn_ref.SetName("stalled_txn"));
+  ASSERT_OK(stalled_txn_ref.Put("stalled_key", "stalled_value"));
+  std::thread writer([&]() { prepare_status = stalled_txn_ref.Prepare(); });
+
+  // The writer leads the main queue and waits on the write stop.
+  TEST_SYNC_POINT(test_prefix + ":writer_stalled");
+  DBRecoverContext context(FlushReason::kErrorRecoveryRetryFlush);
+  Status resume_status = db_impl.TEST_ResumeImpl(context);
+  writer.join();
+  sync_point.DisableProcessing();
+  sync_point.ClearAllCallBacks();
+  sync_point.LoadDependency({});
+
+  ASSERT_OK(resume_status);
+  ASSERT_OK(prepare_status);
+  ASSERT_GE(non_blocking_joins.load(), 3);
+  ASSERT_TRUE(recovery_waited_for_writer.load());
+  ASSERT_TRUE(writer_published_before_sync.load());
+  ASSERT_EQ(versions.LastSequence(), versions.LastAllocatedSequence());
+  ASSERT_OK(stalled_txn_ref.Commit());
+
+  stalled_txn.reset();
+  Close();
+}
+
+// Recovery does not wait for a nonmem write queue whose leader, a WAL-only
+// commit, is in a write stall. The stalled commit has not allocated, so
+// recovery syncs and finishes while the commit is still waiting.
+TEST_F(WritePreparedTransactionSeqnoTest,
+       RecoverySeqnoSyncSkipsStalledWalOnlyCommit) {
+  txn_db_options_.write_policy = TxnDBWritePolicy::WRITE_UNPREPARED;
+  ASSERT_OK(Open());
+  TransactionDB& db = AssertNotNull(db_);
+  DBImpl& db_impl =
+      AssertNotNull(static_cast_with_check<DBImpl>(db.GetRootDB()));
+  VersionSet& versions = AssertNotNull(db_impl.GetVersionSet());
+  SyncPoint& sync_point = AssertNotNull(SyncPoint::GetInstance());
+
+  WriteOptions write_opts;
+  TransactionOptions txn_opts;
+  // The data goes to the memtable at Prepare(), so Commit() writes only a WAL
+  // marker on the nonmem write queue.
+  std::unique_ptr<Transaction> wal_only_commit_txn(
+      db.BeginTransaction(write_opts, txn_opts));
+  Transaction& wal_only_commit_txn_ref =
+      AssertNotNull(wal_only_commit_txn.get());
+  ASSERT_OK(wal_only_commit_txn_ref.SetName("wal_only_commit_txn"));
+  ASSERT_OK(wal_only_commit_txn_ref.Put("commit_key", "commit_value"));
+  ASSERT_OK(wal_only_commit_txn_ref.Prepare());
+
+  const std::string test_prefix = "RecoverySeqnoSyncSkipsStalledWalOnlyCommit";
+  sync_point.LoadDependency(
+      {{"DBImpl::DelayWrite:NonmemWait", test_prefix + ":commit_stalled"}});
+  sync_point.EnableProcessing();
+
+  std::unique_ptr<WriteControllerToken> stop_token =
+      db_impl.TEST_write_controler().GetStopToken();
+  std::atomic<bool> commit_done{false};
+  Status commit_status;
+  std::thread committer([&]() {
+    commit_status = wal_only_commit_txn_ref.Commit();
+    commit_done.store(true);
+  });
+
+  // The commit leads the nonmem queue and waits on the write stop.
+  TEST_SYNC_POINT(test_prefix + ":commit_stalled");
+  DBRecoverContext context(FlushReason::kErrorRecoveryRetryFlush);
+  Status resume_status = db_impl.TEST_ResumeImpl(context);
+  const bool commit_done_at_recovery_end = commit_done.load();
+  const bool synced =
+      versions.LastSequence() == versions.LastAllocatedSequence();
+
+  stop_token.reset();
+  db_impl.TEST_SignalAllBgCv();
+  committer.join();
+  sync_point.DisableProcessing();
+  sync_point.LoadDependency({});
+
+  ASSERT_OK(resume_status);
+  ASSERT_FALSE(commit_done_at_recovery_end);
+  ASSERT_TRUE(synced);
+  ASSERT_OK(commit_status);
+  std::string value;
+  ASSERT_OK(db.Get(ReadOptions(), "commit_key", &value));
+  ASSERT_EQ("commit_value", value);
+
+  wal_only_commit_txn.reset();
+  Close();
+}
+
 // Regression test for the gap left by syncing only at the start of
 // ResumeImpl(). Recovery drops mutex_ in FlushAllColumnFamilies() before it
 // enters AtomicFlushMemTables(), so a WriteUnprepared WAL-only commit can