crimson/osd: fix backfill deadlock when only drop operations remain

BackfillState::Waiting requires an ObjectPushed event to transition
back to Enqueuing, but drop (remove) operations are fire-and-forget
and never generate ObjectPushed. When a backfill round enqueues only
drops and no pushes, the FSM permanently stalls in the Waiting state,
holding the backfill reservation and blocking all other PGs from
backfilling to that OSD.

Fix by adding a pending-pushes counter to ProgressTracker to
distinguish "only drops remain" from "pushes still in-flight". When
all pushes are complete and the namespace is exhausted, drain remaining
drop entries and advance last_backfill to MAX so that
OP_BACKFILL_FINISH triggers the normal completion path via
RequestDone.

The classic OSD avoids this by design: drops are never tracked as
in-flight operations in recover_backfill() and are drained inline.

Fixes: https://tracker.ceph.com/issues/77936
Assisted-by: Claude <claude.ai>
Signed-off-by: Ronen Friedman <rfriedma@redhat.com>

src/crimson/osd/backfill_state.cc

@@ -444,10 +444,11 @@ BackfillState::Enqueuing::Enqueuing(my_context ctx)
     post_event(RequestPrimaryScanning{});
     return;
   } else {
-    if (backfill_state().progress_tracker->tracked_objects_completed()
+    if (backfill_state().progress_tracker->tracked_pushes_completed()
 	&& Enqueuing::all_enqueued(peering_state(),
 				   backfill_state().backfill_info,
 				   backfill_state().peer_backfill_info)) {
+      backfill_state().progress_tracker->complete_drops();
       backfill_state().last_backfill_started = hobject_t::get_max();
       backfill_listener().update_peers_last_backfill(hobject_t::get_max());
     }
@@ -686,6 +687,10 @@ bool BackfillState::ProgressTracker::enqueue_push(const hobject_t& obj)
 {
   [[maybe_unused]] const auto [it, first_seen] = registry.try_emplace(
     obj, registry_item_t{op_stage_t::enqueued_push, std::nullopt});
+  if (first_seen) {
+    // multiple targets may enqueue the same object; only count it once
+    ++num_pending_pushes;
+  }
   return first_seen;
 }
 
@@ -704,6 +709,10 @@ void BackfillState::ProgressTracker::complete_to(
   DEBUGDPP("obj={}", pg(), obj);
   if (auto completion_iter = registry.find(obj);
       completion_iter != std::end(registry)) {
+    if (completion_iter->second.stage == op_stage_t::enqueued_push) {
+      ceph_assert(num_pending_pushes > 0);
+      --num_pending_pushes;
+    }
     completion_iter->second = \
       registry_item_t{ op_stage_t::completed_push, stats };
   } else {
@@ -734,6 +743,27 @@ void BackfillState::ProgressTracker::complete_to(
   }
 }
 
+void BackfillState::ProgressTracker::complete_drops()
+{
+  LOG_PREFIX(BackfillState::ProgressTracker::complete_drops);
+  ceph_assert(num_pending_pushes == 0);
+  auto new_last_backfill = peering_state().earliest_backfill();
+  for (auto it = std::begin(registry);
+       it != std::end(registry);
+       it = registry.erase(it)) {
+    const auto& [soid, item] = *it;
+    ceph_assert(item.stage == op_stage_t::enqueued_drop);
+    assert(item.stats);
+    DEBUGDPP("draining drop obj={}", pg(), soid);
+    peering_state().update_complete_backfill_object_stats(
+      soid,
+      *item.stats);
+    assert(soid > new_last_backfill);
+    new_last_backfill = soid;
+  }
+  backfill_listener().update_peers_last_backfill(new_last_backfill);
+}
+
 void BackfillState::enqueue_standalone_push(
   const hobject_t &obj,
   const eversion_t &v,

src/crimson/osd/backfill_state.h

@@ -446,6 +446,8 @@ class BackfillState::ProgressTracker {
 
   BackfillMachine& backfill_machine;
   std::map<hobject_t, registry_item_t> registry;
+  // count of registry entries at enqueued_push stage, awaiting ObjectPushed.
+  size_t num_pending_pushes = 0;
 
   BackfillState& backfill_state() {
     return backfill_machine.backfill_state;
@@ -467,9 +469,17 @@ public:
 
   bool tracked_objects_completed() const;
 
+  // true when no push operations are awaiting ObjectPushed completion.
+  bool tracked_pushes_completed() const {
+    return num_pending_pushes == 0;
+  }
+
   bool enqueue_push(const hobject_t&);
   void enqueue_drop(const hobject_t&);
   void complete_to(const hobject_t&, const pg_stat_t&, bool may_push_to_max);
+
+  // drain all remaining drop entries from the registry, updating stats.
+  void complete_drops();
 };
 
 } // namespace crimson::osd