crimson: Create the shared promise before waited upon

RecoveryBackend::pushes map creates each shared_promise
in wait_for_pushes call. There can be a situation where
set_pushed is called due to handled push reply (handle_push_reply)
before the shared_promise was even constructed due to backfill progress
is stuck.

Fixes: https://tracker.ceph.com/issues/70502
Signed-off-by: Radoslaw Zarzynski <[email protected]>
Signed-off-by: Mohit Agrawal <[email protected]>
(cherry picked from commit 435f065)

Author: mohit84

src/crimson/osd/recovery_backend.h

@@ -220,11 +220,9 @@ class RecoveryBackend {
     }
     void set_pushed(pg_shard_t shard) {
       auto it = pushes.find(shard);
-      if (it != pushes.end()) {
-	auto &push_promise = it->second;
-	push_promise.set_value();
-	pushes.erase(it);
-      }
+      ceph_assert(it != pushes.end());
+      it->second.set_value();
+      pushes.erase(it);
     }
     void set_pulled() {
       if (pulled) {

src/crimson/osd/replicated_recovery_backend.cc

@@ -81,13 +81,14 @@ ReplicatedRecoveryBackend::maybe_push_shards(
         msg->min_epoch = pg.get_last_peering_reset();
         msg->pushes.push_back(std::move(push));
         msg->set_priority(pg.get_recovery_op_priority());
+        seastar::future<> push_future = get_recovering(soid).wait_for_pushes(shard);
         return interruptor::make_interruptible(
             shard_services.send_to_osd(shard.osd,
                                        std::move(msg),
                                        pg.get_osdmap_epoch()))
         .then_interruptible(
-          [this, soid, shard] {
-          return get_recovering(soid).wait_for_pushes(shard);
+          [push_future = std::move(push_future)]() mutable {
+          return std::move(push_future);
         });
       });
     });
@@ -184,11 +185,12 @@ ReplicatedRecoveryBackend::push_delete(
 	  pg.get_pg_whoami(), target_pg, pg.get_osdmap_epoch(), min_epoch);
       msg->set_priority(pg.get_recovery_op_priority());
       msg->objects.push_back(std::make_pair(soid, need));
+      seastar::future<> push_future = get_recovering(soid).wait_for_pushes(shard);
       return interruptor::make_interruptible(
 	  shard_services.send_to_osd(shard.osd, std::move(msg),
 				     pg.get_osdmap_epoch())).then_interruptible(
-	[this, soid, shard] {
-	return get_recovering(soid).wait_for_pushes(shard);
+        [push_future = std::move(push_future)]() mutable {
+        return std::move(push_future);
       });
     }
     return seastar::make_ready_future<>();