Prevent normal flushes from starving error recovery (#15132)

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

This fixes a liveness bug in background-error recovery. Retryable soft errors allow foreground writes to continue, but recovery must wait for every scheduled background flush worker to exit before it can submit the recovery flush.

While recovery was active, normal flush requests could still enter `flush_queue_`. Atomic flush requests bypass per-column-family deduplication, so writes could accumulate a backlog of normal requests.

For each request, the background worker:

  1. Pops and drops the normal request because BG work is stopped
  2. Sleeps for the one-second background-error backoff
  3. Decrements `bg_flush_scheduled_`: 1 -> 0
  4. Calls `MaybeScheduleFlushOrCompaction()`
  5. Schedules the next queued request: 0 -> 1
  6. Only then signals the recovery thread

The rescheduling happens under the DB mutex before recovery is signaled:

  Queued requests: R1, R2, R3, ...

  Flush worker                      Recovery thread
  ------------                      ---------------
  drop R1
  sleep 1 second                    waiting for scheduled == 0
  scheduled: 1 -> 0
  schedule R2: 0 -> 1
  signal -------------------------> wake, observe scheduled == 1
                                    wait again

  drop R2
  sleep 1 second
  scheduled: 1 -> 0
  schedule R3: 0 -> 1
  signal -------------------------> wake, observe scheduled == 1
                                    wait again

The dropped request is not requeued. Instead, the worker exit path guarantees that the next queued request is scheduled while `unscheduled_flushes_ > 0`.

A finite backlog therefore delays recovery by roughly one second per request. If writes keep producing requests as quickly as workers discard them, the queue may never drain and recovery is starved indefinitely. Recovery can complete after writes become quiescent, so this is a liveness failure rather than a lock deadlock or data-corruption bug.

There is also a queue-accounting edge. Scheduling consumes `unscheduled_flushes_`, but a worker does not own its `FlushRequest` until it pops `flush_queue_`. A worker can exit on the background error before popping, leaving:

  flush_queue_.size()  > 0
  unscheduled_flushes_ == 0

Clearing only unscheduled requests therefore misses scheduled-but-unclaimed requests.

The fix:

- Reject new non-recovery requests while background work is stopped.
- Allow error-recovery flush reasons to bypass that guard.
- Clear the entire flush queue before recovery waits for scheduled workers.
- Retain the worker-side reason check for requests queued before the error.
- Rebuild recovery flushes and schedule a catch-up flush for writes made during recovery.

Reviewed By: pdillinger

Differential Revision: D116708546

fbshipit-source-id: 6e29fd3653ef72d047c64ca89d311a7bee51ab84

db/db_flush_test.cc

@@ -3725,6 +3725,108 @@ TEST_F(DBFlushTest, NonAtomicNormalFlushAbortWhenBGError) {
   SyncPoint::GetInstance()->DisableProcessing();
 }
 
+TEST_F(DBFlushTest, NonRecoveryFlushDoesNotStarveErrorRecovery) {
+  constexpr uint64_t kRecoveryTimeoutMicros = 15 * 1000 * 1000;
+  constexpr int kLateWriteCount = 32;
+
+  Options opts = CurrentOptions();
+  opts.atomic_flush = true;
+  opts.memtable_factory.reset(test::NewSpecialSkipListFactory(1));
+  opts.max_write_buffer_number = 64;
+  opts.max_background_flushes = 1;
+  DestroyAndReopen(opts);
+  env_->SetBackgroundThreads(1, Env::HIGH);
+
+  std::atomic_int flush_write_table_count{0};
+  port::Mutex wait_mutex;
+  port::CondVar wait_cv(&wait_mutex);
+  bool recovery_wait_started = false;
+  bool recovery_succeeded = false;
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->SetCallBack(
+      "FlushJob::WriteLevel0Table:s", [&](void* s_ptr) {
+        if (flush_write_table_count.fetch_add(1) == 0) {
+          Status* s = static_cast<Status*>(s_ptr);
+          IOStatus io_error = IOStatus::IOError("injected foobar");
+          io_error.SetRetryable(true);
+          *s = io_error;
+          TEST_SYNC_POINT(
+              "DBFlushTest::NonRecoveryFlushDoesNotStarveErrorRecovery:"
+              "FirstFlushFailed");
+          TEST_SYNC_POINT(
+              "DBFlushTest::NonRecoveryFlushDoesNotStarveErrorRecovery:"
+              "ContinueFirstFlush");
+        }
+      });
+  SyncPoint::GetInstance()->SetCallBack(
+      "DBImpl::ResumeImpl:BeforeWaitForBackgroundWork", [&](void*) {
+        MutexLock l(&wait_mutex);
+        recovery_wait_started = true;
+        wait_cv.SignalAll();
+      });
+  SyncPoint::GetInstance()->SetCallBack(
+      "RecoverFromRetryableBGIOError:RecoverSuccess", [&](void*) {
+        MutexLock l(&wait_mutex);
+        recovery_succeeded = true;
+        wait_cv.SignalAll();
+      });
+  SyncPoint::GetInstance()->LoadDependency(
+      {{"DBFlushTest::NonRecoveryFlushDoesNotStarveErrorRecovery:"
+        "FirstFlushFailed",
+        "DBFlushTest::NonRecoveryFlushDoesNotStarveErrorRecovery:"
+        "FailureObserved"},
+       {"DBFlushTest::NonRecoveryFlushDoesNotStarveErrorRecovery:"
+        "ReleaseFirstFlush",
+        "DBFlushTest::NonRecoveryFlushDoesNotStarveErrorRecovery:"
+        "ContinueFirstFlush"}});
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  auto wait_until = [&](const auto& predicate) {
+    const uint64_t abs_time =
+        SystemClock::Default()->NowMicros() + kRecoveryTimeoutMicros;
+    MutexLock l(&wait_mutex);
+    while (!predicate()) {
+      if (wait_cv.TimedWait(abs_time)) {
+        return false;
+      }
+    }
+    return true;
+  };
+
+  ASSERT_OK(Put(Key(1), "val1"));
+  ASSERT_OK(Put(Key(2), "val2"));
+  TEST_SYNC_POINT(
+      "DBFlushTest::NonRecoveryFlushDoesNotStarveErrorRecovery:"
+      "FailureObserved");
+  TEST_SYNC_POINT(
+      "DBFlushTest::NonRecoveryFlushDoesNotStarveErrorRecovery:"
+      "ReleaseFirstFlush");
+
+  const bool recovery_wait_reached =
+      wait_until([&] { return recovery_wait_started; });
+  Status late_write_status;
+  if (recovery_wait_reached) {
+    for (int key = 3; key < 3 + kLateWriteCount && late_write_status.ok();
+         ++key) {
+      late_write_status = Put(Key(key), "val");
+    }
+  }
+  const bool recovered = wait_until([&] { return recovery_succeeded; });
+
+  if (!recovered) {
+    IOStatus cleanup_error = IOStatus::IOError("stop stuck error recovery");
+    dbfull()->TEST_SetBGError(cleanup_error, BackgroundErrorReason::kFlush);
+    dbfull()->TEST_SignalAllBgCv();
+  }
+
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+  SyncPoint::GetInstance()->DisableProcessing();
+  ASSERT_TRUE(recovery_wait_reached);
+  ASSERT_OK(late_write_status);
+  ASSERT_TRUE(recovered);
+}
+
 TEST_F(DBFlushTest, DBStuckAfterAtomicFlushError) {
   // Test for a bug with atomic flush where DB can become stuck
   // after a flush error. A repro timeline:

db/db_impl/db_impl.cc

@@ -347,6 +347,22 @@ Status DBImpl::ResumeImpl(DBRecoverContext context,
   const ReadOptions read_options(io_activity);
   const WriteOptions write_options(io_activity);
 
+  assert(static_cast<size_t>(unscheduled_flushes_) <= flush_queue_.size());
+  // Recovery rebuilds flush requests for every column family with pending
+  // immutable data after scheduled workers exit, so every existing request is
+  // redundant and safe to remove. A worker can exit on the background error
+  // without popping a request, leaving it out of unscheduled_flushes_.
+  while (!flush_queue_.empty()) {
+    FlushRequest flush_req = PopFirstFromFlushQueue();
+    for (const auto& item : flush_req.cfd_to_max_mem_id_to_persist) {
+      ColumnFamilyData* cfd = item.first;
+      assert(cfd);
+      cfd->UnrefAndTryDelete();
+    }
+  }
+  unscheduled_flushes_ = 0;
+
+  TEST_SYNC_POINT("DBImpl::ResumeImpl:BeforeWaitForBackgroundWork");
   WaitForBackgroundWork();
 
   TEST_SYNC_POINT("DBImpl::ResumeImpl:Start");
@@ -474,9 +490,8 @@ Status DBImpl::ResumeImpl(DBRecoverContext context,
     s = Status::ShutdownInProgress();
   }
   if (s.ok() && context.flush_after_recovery) {
-    // Since we drop all non-recovery flush requests during recovery,
-    // and new memtable may fill up during recovery,
-    // schedule one more round of flush.
+    // Normal flush requests are discarded during recovery, and a new memtable
+    // may fill while recovery releases the DB mutex. Schedule a catch-up flush.
     Status status = RetryFlushesForErrorRecovery(
         FlushReason::kCatchUpAfterErrorRecovery, false /* wait */);
     if (!status.ok()) {

db/db_impl/db_impl.h

@@ -2785,7 +2785,11 @@ class DBImpl : public DB
     // equal to this per-column-family specified value, this flush request is
     // considered to have completed its work of flushing this column family.
     // After completing the work for all column families in this request, this
-    // flush is considered complete.
+    // flush is considered complete. EnqueuePendingFlush() acquires one
+    // reference for each CFD when it successfully queues this request.
+    // PopFirstFromFlushQueue() transfers responsibility for those references
+    // to its caller, which must release each one with UnrefAndTryDelete() after
+    // processing or discarding the request.
     std::unordered_map<ColumnFamilyData*, uint64_t>
         cfd_to_max_mem_id_to_persist;
 

db/db_impl/db_impl_compaction_flush.cc

@@ -3512,14 +3512,9 @@ void DBImpl::MaybeScheduleFlushOrCompaction() {
     return;
   } else if (error_handler_.IsBGWorkStopped() &&
              !error_handler_.IsRecoveryInProgress()) {
-    // There has been a hard error and this call is not part of the recovery
-    // sequence. Bail out here so we don't get into an endless loop of
-    // scheduling BG work which will again call this function
-    //
-    // Note that a non-recovery flush can still be scheduled if
-    // error_handler_.IsRecoveryInProgress() returns true. We rely on
-    // BackgroundCallFlush() to check flush reason and drop non-recovery
-    // flushes.
+    // Recovery must be able to schedule its flushes while background work is
+    // stopped. New non-recovery requests are rejected by EnqueuePendingFlush,
+    // and BackgroundFlush drops any request queued before the error.
     return;
   } else if (shutting_down_.load(std::memory_order_acquire)) {
     // DB is being deleted; no more background compactions
@@ -3817,6 +3812,10 @@ bool DBImpl::EnqueuePendingFlush(const FlushRequest& flush_req) {
   if (reject_new_background_jobs_) {
     return enqueued;
   }
+  if (error_handler_.IsBGWorkStopped() &&
+      !IsRecoveryFlush(flush_req.flush_reason)) {
+    return enqueued;
+  }
   if (flush_req.cfd_to_max_mem_id_to_persist.empty()) {
     return enqueued;
   }
@@ -3973,8 +3972,8 @@ Status DBImpl::BackgroundFlush(bool* made_progress, JobContext* job_context,
 
   Status status;
   *reason = FlushReason::kOthers;
-  // If BG work is stopped due to an error, but a recovery is in progress,
-  // that means this flush is part of the recovery. So allow it to go through
+  // During recovery, allow flush workers to inspect the queue. Non-recovery
+  // requests are dropped below.
   if (!error_handler_.IsBGWorkStopped()) {
     if (shutting_down_.load(std::memory_order_acquire)) {
       status = Status::ShutdownInProgress();
@@ -3996,12 +3995,9 @@ Status DBImpl::BackgroundFlush(bool* made_progress, JobContext* job_context,
     FlushRequest flush_req = PopFirstFromFlushQueue();
     FlushReason flush_reason = flush_req.flush_reason;
     if (!error_handler_.GetBGError().ok() && error_handler_.IsBGWorkStopped() &&
-        flush_reason != FlushReason::kErrorRecovery &&
-        flush_reason != FlushReason::kErrorRecoveryRetryFlush) {
-      // Stop non-recovery flush when bg work is stopped
-      // Note that we drop the flush request here.
-      // Recovery thread should schedule further flushes after bg error
-      // is cleared.
+        !IsRecoveryFlush(flush_reason)) {
+      // A request queued before the error can reach a worker before recovery
+      // clears the queue. Drop it; recovery rebuilds flush work for all CFs.
       status = error_handler_.GetBGError();
       assert(!status.ok());
       ROCKS_LOG_BUFFER(log_buffer,

db/error_handler.cc

@@ -637,10 +637,6 @@ Status ErrorHandler::RecoverFromBGError(bool is_manual) {
       return Status::Busy("Recovery already in progress");
     }
     recovery_in_prog_ = true;
-
-    // In manual resume, we allow the bg work to run. If it is a auto resume,
-    // the bg work should follow this tag.
-    soft_error_no_bg_work_ = false;
   }
 
   // kErrorRecovery can be the default context for a generic soft error that
@@ -659,7 +655,9 @@ Status ErrorHandler::RecoverFromBGError(bool is_manual) {
   // can generate background errors should be the flush operations
   recovery_error_ = IOStatus::OK();
   recovery_error_.PermitUncheckedError();
-  Status s = db_->ResumeImpl(recover_context_, Env::IOActivity::kFlush);
+  DBRecoverContext context = recover_context_;
+  context.flush_after_recovery = true;
+  Status s = db_->ResumeImpl(context, Env::IOActivity::kFlush);
   if (s.ok()) {
     soft_error_no_bg_work_ = false;
   } else {

unreleased_history/bug_fixes/background_error_recovery_flush_starvation.md

@@ -0,0 +1 @@
+Fixed a liveness bug where RocksDB could remain indefinitely in recovery from a soft background error when concurrent writes continuously generated normal flush requests.