Adding upstream version 18.2.2.upstream/18.2.2

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 1544 insertions, 0 deletions

diff --git a/src/crimson/osd/pg.cc b/src/crimson/osd/pg.cc
new file mode 100644
index 000000000..7cf3b158c
--- /dev/null
+++ b/src/crimson/osd/pg.cc
@@ -0,0 +1,1544 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:nil -*-
+// vim: ts=8 sw=2 smarttab expandtab
+
+#include "pg.h"
+
+#include <functional>
+
+#include <boost/range/adaptor/filtered.hpp>
+#include <boost/range/adaptor/map.hpp>
+#include <boost/range/adaptor/transformed.hpp>
+#include <boost/range/algorithm/copy.hpp>
+#include <boost/range/algorithm/max_element.hpp>
+#include <boost/range/numeric.hpp>
+#include <fmt/format.h>
+#include <fmt/ostream.h>
+
+#include "common/hobject_fmt.h"
+
+#include "messages/MOSDOp.h"
+#include "messages/MOSDOpReply.h"
+#include "messages/MOSDRepOp.h"
+#include "messages/MOSDRepOpReply.h"
+
+#include "osd/OSDMap.h"
+#include "osd/osd_types_fmt.h"
+
+#include "os/Transaction.h"
+
+#include "crimson/common/exception.h"
+#include "crimson/net/Connection.h"
+#include "crimson/net/Messenger.h"
+#include "crimson/os/cyanstore/cyan_store.h"
+#include "crimson/os/futurized_collection.h"
+#include "crimson/osd/exceptions.h"
+#include "crimson/osd/pg_meta.h"
+#include "crimson/osd/pg_backend.h"
+#include "crimson/osd/ops_executer.h"
+#include "crimson/osd/osd_operations/osdop_params.h"
+#include "crimson/osd/osd_operations/peering_event.h"
+#include "crimson/osd/osd_operations/background_recovery.h"
+#include "crimson/osd/osd_operations/snaptrim_event.h"
+#include "crimson/osd/pg_recovery.h"
+#include "crimson/osd/replicated_recovery_backend.h"
+#include "crimson/osd/watch.h"
+
+using std::ostream;
+using std::set;
+using std::string;
+using std::vector;
+
+namespace {
+  seastar::logger& logger() {
+    return crimson::get_logger(ceph_subsys_osd);
+  }
+}
+
+namespace std::chrono {
+std::ostream& operator<<(std::ostream& out, const signedspan& d)
+{
+  auto s = std::chrono::duration_cast<std::chrono::seconds>(d).count();
+  auto ns = std::abs((d % 1s).count());
+  fmt::print(out, "{}{}s", s, ns ? fmt::format(".{:0>9}", ns) : "");
+  return out;
+}
+}
+
+template <typename T>
+struct fmt::formatter<std::optional<T>> : fmt::formatter<T> {
+  template <typename FormatContext>
+  auto format(const std::optional<T>& v, FormatContext& ctx) const {
+    if (v.has_value()) {
+      return fmt::formatter<T>::format(*v, ctx);
+    }
+    return fmt::format_to(ctx.out(), "<null>");
+  }
+};
+
+namespace crimson::osd {
+
+using crimson::common::local_conf;
+
+class RecoverablePredicate : public IsPGRecoverablePredicate {
+public:
+  bool operator()(const set<pg_shard_t> &have) const override {
+    return !have.empty();
+  }
+};
+
+class ReadablePredicate: public IsPGReadablePredicate {
+  pg_shard_t whoami;
+public:
+  explicit ReadablePredicate(pg_shard_t whoami) : whoami(whoami) {}
+  bool operator()(const set<pg_shard_t> &have) const override {
+    return have.count(whoami);
+  }
+};
+
+PG::PG(
+  spg_t pgid,
+  pg_shard_t pg_shard,
+  crimson::os::CollectionRef coll_ref,
+  pg_pool_t&& pool,
+  std::string&& name,
+  cached_map_t osdmap,
+  ShardServices &shard_services,
+  ec_profile_t profile)
+  : pgid{pgid},
+    pg_whoami{pg_shard},
+    coll_ref{coll_ref},
+    pgmeta_oid{pgid.make_pgmeta_oid()},
+    osdmap_gate("PG::osdmap_gate"),
+    shard_services{shard_services},
+    backend(
+      PGBackend::create(
+	pgid.pgid,
+	pg_shard,
+	pool,
+	coll_ref,
+	shard_services,
+	profile,
+	*this)),
+    recovery_backend(
+      std::make_unique<ReplicatedRecoveryBackend>(
+	*this, shard_services, coll_ref, backend.get())),
+    recovery_handler(
+      std::make_unique<PGRecovery>(this)),
+    peering_state(
+      shard_services.get_cct(),
+      pg_shard,
+      pgid,
+      PGPool(
+	osdmap,
+	pgid.pool(),
+	pool,
+	name),
+      osdmap,
+      this,
+      this),
+    obc_registry{
+      local_conf()},
+    obc_loader{
+      obc_registry,
+      *backend.get(),
+      *this},
+    osdriver(
+      &shard_services.get_store(),
+      coll_ref,
+      pgid.make_pgmeta_oid()),
+    snap_mapper(
+      this->shard_services.get_cct(),
+      &osdriver,
+      pgid.ps(),
+      pgid.get_split_bits(pool.get_pg_num()),
+      pgid.pool(),
+      pgid.shard),
+    wait_for_active_blocker(this)
+{
+  peering_state.set_backend_predicates(
+    new ReadablePredicate(pg_whoami),
+    new RecoverablePredicate());
+  osdmap_gate.got_map(osdmap->get_epoch());
+}
+
+PG::~PG() {}
+
+void PG::check_blocklisted_watchers()
+{
+  logger().debug("{}", __func__);
+  obc_registry.for_each([this](ObjectContextRef obc) {
+    assert(obc);
+    for (const auto& [key, watch] : obc->watchers) {
+      assert(watch->get_pg() == this);
+      const auto& ea = watch->get_peer_addr();
+      logger().debug("watch: Found {} cookie {}. Checking entity_add_t {}",
+                     watch->get_entity(), watch->get_cookie(), ea);
+      if (get_osdmap()->is_blocklisted(ea)) {
+        logger().info("watch: Found blocklisted watcher for {}", ea);
+        watch->do_watch_timeout();
+      }
+    }
+  });
+}
+
+bool PG::try_flush_or_schedule_async() {
+  logger().debug("PG::try_flush_or_schedule_async: flush ...");
+  (void)shard_services.get_store().flush(
+    coll_ref
+  ).then(
+    [this, epoch=get_osdmap_epoch()]() {
+      return shard_services.start_operation<LocalPeeringEvent>(
+	this,
+	pg_whoami,
+	pgid,
+	epoch,
+	epoch,
+	PeeringState::IntervalFlush());
+    });
+  return false;
+}
+
+void PG::publish_stats_to_osd()
+{
+  if (!is_primary())
+    return;
+  if (auto new_pg_stats = peering_state.prepare_stats_for_publish(
+        pg_stats,
+        object_stat_collection_t());
+      new_pg_stats.has_value()) {
+    pg_stats = std::move(new_pg_stats);
+  }
+}
+
+void PG::clear_publish_stats()
+{
+  pg_stats.reset();
+}
+
+pg_stat_t PG::get_stats() const
+{
+  return pg_stats.value_or(pg_stat_t{});
+}
+
+void PG::queue_check_readable(epoch_t last_peering_reset, ceph::timespan delay)
+{
+  // handle the peering event in the background
+  logger().debug(
+    "{}: PG::queue_check_readable lpr: {}, delay: {}",
+    *this, last_peering_reset, delay);
+  check_readable_timer.cancel();
+  check_readable_timer.set_callback([last_peering_reset, this] {
+    logger().debug(
+      "{}: PG::queue_check_readable callback lpr: {}",
+      *this, last_peering_reset);
+    (void) shard_services.start_operation<LocalPeeringEvent>(
+      this,
+      pg_whoami,
+      pgid,
+      last_peering_reset,
+      last_peering_reset,
+      PeeringState::CheckReadable{});
+    });
+  check_readable_timer.arm(
+    std::chrono::duration_cast<seastar::lowres_clock::duration>(delay));
+}
+
+void PG::recheck_readable()
+{
+  bool changed = false;
+  const auto mnow = shard_services.get_mnow();
+  if (peering_state.state_test(PG_STATE_WAIT)) {
+    auto prior_readable_until_ub = peering_state.get_prior_readable_until_ub();
+    if (mnow < prior_readable_until_ub) {
+      logger().info(
+	"{}: {} will wait (mnow {} < prior_readable_until_ub {})",
+	*this, __func__, mnow, prior_readable_until_ub);
+      queue_check_readable(
+	peering_state.get_last_peering_reset(),
+	prior_readable_until_ub - mnow);
+    } else {
+      logger().info(
+	"{}:{} no longer wait (mnow {} >= prior_readable_until_ub {})",
+	*this, __func__, mnow, prior_readable_until_ub);
+      peering_state.state_clear(PG_STATE_WAIT);
+      peering_state.clear_prior_readable_until_ub();
+      changed = true;
+    }
+  }
+  if (peering_state.state_test(PG_STATE_LAGGY)) {
+    auto readable_until = peering_state.get_readable_until();
+    if (readable_until == readable_until.zero()) {
+      logger().info(
+	"{}:{} still laggy (mnow {}, readable_until zero)",
+	*this, __func__, mnow);
+    } else if (mnow >= readable_until) {
+      logger().info(
+	"{}:{} still laggy (mnow {} >= readable_until {})",
+	*this, __func__, mnow, readable_until);
+    } else {
+      logger().info(
+	"{}:{} no longer laggy (mnow {} < readable_until {})",
+	*this, __func__, mnow, readable_until);
+      peering_state.state_clear(PG_STATE_LAGGY);
+      changed = true;
+    }
+  }
+  if (changed) {
+    publish_stats_to_osd();
+    if (!peering_state.state_test(PG_STATE_WAIT) &&
+	!peering_state.state_test(PG_STATE_LAGGY)) {
+      // TODO: requeue ops waiting for readable
+    }
+  }
+}
+
+unsigned PG::get_target_pg_log_entries() const
+{
+  const unsigned local_num_pgs = shard_services.get_num_local_pgs();
+  const unsigned local_target =
+    local_conf().get_val<uint64_t>("osd_target_pg_log_entries_per_osd") /
+    seastar::smp::count;
+  const unsigned min_pg_log_entries =
+    local_conf().get_val<uint64_t>("osd_min_pg_log_entries");
+  if (local_num_pgs > 0 && local_target > 0) {
+    // target an even spread of our budgeted log entries across all
+    // PGs.  note that while we only get to control the entry count
+    // for primary PGs, we'll normally be responsible for a mix of
+    // primary and replica PGs (for the same pool(s) even), so this
+    // will work out.
+    const unsigned max_pg_log_entries =
+      local_conf().get_val<uint64_t>("osd_max_pg_log_entries");
+    return std::clamp(local_target / local_num_pgs,
+		      min_pg_log_entries,
+		      max_pg_log_entries);
+  } else {
+    // fall back to a per-pg value.
+    return min_pg_log_entries;
+  }
+}
+
+void PG::on_removal(ceph::os::Transaction &t) {
+  t.register_on_commit(
+    new LambdaContext(
+      [this](int r) {
+      ceph_assert(r == 0);
+      (void)shard_services.start_operation<LocalPeeringEvent>(
+        this, pg_whoami, pgid, float(0.001), get_osdmap_epoch(),
+        get_osdmap_epoch(), PeeringState::DeleteSome());
+  }));
+}
+
+void PG::on_activate(interval_set<snapid_t> snaps)
+{
+  logger().debug("{}: {} snaps={}", *this, __func__, snaps);
+  snap_trimq = std::move(snaps);
+  projected_last_update = peering_state.get_info().last_update;
+}
+
+void PG::on_activate_complete()
+{
+  wait_for_active_blocker.unblock();
+
+  if (peering_state.needs_recovery()) {
+    logger().info("{}: requesting recovery",
+                  __func__);
+    (void) shard_services.start_operation<LocalPeeringEvent>(
+      this,
+      pg_whoami,
+      pgid,
+      float(0.001),
+      get_osdmap_epoch(),
+      get_osdmap_epoch(),
+      PeeringState::DoRecovery{});
+  } else if (peering_state.needs_backfill()) {
+    logger().info("{}: requesting backfill",
+                  __func__);
+    (void) shard_services.start_operation<LocalPeeringEvent>(
+      this,
+      pg_whoami,
+      pgid,
+      float(0.001),
+      get_osdmap_epoch(),
+      get_osdmap_epoch(),
+      PeeringState::RequestBackfill{});
+  } else {
+    logger().debug("{}: no need to recover or backfill, AllReplicasRecovered",
+		   " for pg: {}", __func__, pgid);
+    (void) shard_services.start_operation<LocalPeeringEvent>(
+      this,
+      pg_whoami,
+      pgid,
+      float(0.001),
+      get_osdmap_epoch(),
+      get_osdmap_epoch(),
+      PeeringState::AllReplicasRecovered{});
+  }
+  publish_stats_to_osd();
+}
+
+void PG::prepare_write(pg_info_t &info,
+		       pg_info_t &last_written_info,
+		       PastIntervals &past_intervals,
+		       PGLog &pglog,
+		       bool dirty_info,
+		       bool dirty_big_info,
+		       bool need_write_epoch,
+		       ceph::os::Transaction &t)
+{
+  std::map<string,bufferlist> km;
+  std::string key_to_remove;
+  if (dirty_big_info || dirty_info) {
+    int ret = prepare_info_keymap(
+      shard_services.get_cct(),
+      &km,
+      &key_to_remove,
+      get_osdmap_epoch(),
+      info,
+      last_written_info,
+      past_intervals,
+      dirty_big_info,
+      need_write_epoch,
+      true,
+      nullptr,
+      this);
+    ceph_assert(ret == 0);
+  }
+  pglog.write_log_and_missing(
+    t, &km, coll_ref->get_cid(), pgmeta_oid,
+    peering_state.get_pgpool().info.require_rollback());
+  if (!km.empty()) {
+    t.omap_setkeys(coll_ref->get_cid(), pgmeta_oid, km);
+  }
+  if (!key_to_remove.empty()) {
+    t.omap_rmkey(coll_ref->get_cid(), pgmeta_oid, key_to_remove);
+  }
+}
+
+std::pair<ghobject_t, bool>
+PG::do_delete_work(ceph::os::Transaction &t, ghobject_t _next)
+{
+  logger().info("removing pg {}", pgid);
+  auto fut = interruptor::make_interruptible(
+    shard_services.get_store().list_objects(
+      coll_ref,
+      _next,
+      ghobject_t::get_max(),
+      local_conf()->osd_target_transaction_size));
+
+  auto [objs_to_rm, next] = fut.get();
+  if (objs_to_rm.empty()) {
+    logger().info("all objs removed, removing coll for {}", pgid);
+    t.remove(coll_ref->get_cid(), pgmeta_oid);
+    t.remove_collection(coll_ref->get_cid());
+    (void) shard_services.get_store().do_transaction(
+      coll_ref, std::move(t)).then([this] {
+      return shard_services.remove_pg(pgid);
+    });
+    return {next, false};
+  } else {
+    for (auto &obj : objs_to_rm) {
+      if (obj == pgmeta_oid) {
+        continue;
+      }
+      logger().trace("pg {}, removing obj {}", pgid, obj);
+      t.remove(coll_ref->get_cid(), obj);
+    }
+    t.register_on_commit(
+      new LambdaContext([this](int r) {
+      ceph_assert(r == 0);
+      logger().trace("triggering more pg delete {}", pgid);
+      (void) shard_services.start_operation<LocalPeeringEvent>(
+        this,
+        pg_whoami,
+        pgid,
+        float(0.001),
+        get_osdmap_epoch(),
+        get_osdmap_epoch(),
+        PeeringState::DeleteSome{});
+    }));
+    return {next, true};
+  }
+}
+
+Context *PG::on_clean()
+{
+  // Not needed yet (will be needed for IO unblocking)
+  return nullptr;
+}
+
+void PG::on_active_actmap()
+{
+  logger().debug("{}: {} snap_trimq={}", *this, __func__, snap_trimq);
+  peering_state.state_clear(PG_STATE_SNAPTRIM_ERROR);
+  // loops until snap_trimq is empty or SNAPTRIM_ERROR.
+  std::ignore = seastar::do_until(
+    [this] { return snap_trimq.empty()
+                    || peering_state.state_test(PG_STATE_SNAPTRIM_ERROR);
+    },
+    [this] {
+      peering_state.state_set(PG_STATE_SNAPTRIM);
+      publish_stats_to_osd();
+      const auto to_trim = snap_trimq.range_start();
+      snap_trimq.erase(to_trim);
+      const auto needs_pause = !snap_trimq.empty();
+      return seastar::repeat([to_trim, needs_pause, this] {
+        logger().debug("{}: going to start SnapTrimEvent, to_trim={}",
+                       *this, to_trim);
+        return shard_services.start_operation<SnapTrimEvent>(
+          this,
+          snap_mapper,
+          to_trim,
+          needs_pause
+        ).second.handle_error(
+          crimson::ct_error::enoent::handle([this] {
+            logger().error("{}: ENOENT saw, trimming stopped", *this);
+            peering_state.state_set(PG_STATE_SNAPTRIM_ERROR);
+            publish_stats_to_osd();
+            return seastar::make_ready_future<seastar::stop_iteration>(
+              seastar::stop_iteration::yes);
+          }), crimson::ct_error::eagain::handle([this] {
+            logger().info("{}: EAGAIN saw, trimming restarted", *this);
+            return seastar::make_ready_future<seastar::stop_iteration>(
+              seastar::stop_iteration::no);
+          })
+        );
+      }).then([this, trimmed=to_trim] {
+        logger().debug("{}: trimmed snap={}", *this, trimmed);
+      });
+    }).finally([this] {
+      logger().debug("{}: PG::on_active_actmap() finished trimming",
+                     *this);
+      peering_state.state_clear(PG_STATE_SNAPTRIM);
+      peering_state.state_clear(PG_STATE_SNAPTRIM_ERROR);
+      publish_stats_to_osd();
+    });
+}
+
+void PG::on_active_advmap(const OSDMapRef &osdmap)
+{
+  const auto new_removed_snaps = osdmap->get_new_removed_snaps();
+  if (auto it = new_removed_snaps.find(get_pgid().pool());
+      it != new_removed_snaps.end()) {
+    bool bad = false;
+    for (auto j : it->second) {
+      if (snap_trimq.intersects(j.first, j.second)) {
+	decltype(snap_trimq) added, overlap;
+	added.insert(j.first, j.second);
+	overlap.intersection_of(snap_trimq, added);
+        logger().error("{}: {} removed_snaps already contains {}",
+                       *this, __func__, overlap);
+	bad = true;
+	snap_trimq.union_of(added);
+      } else {
+	snap_trimq.insert(j.first, j.second);
+      }
+    }
+    logger().info("{}: {} new removed snaps {}, snap_trimq now{}",
+                  *this, __func__, it->second, snap_trimq);
+    assert(!bad || !local_conf().get_val<bool>("osd_debug_verify_cached_snaps"));
+  }
+}
+
+void PG::scrub_requested(scrub_level_t scrub_level, scrub_type_t scrub_type)
+{
+  // TODO: should update the stats upon finishing the scrub
+  peering_state.update_stats([scrub_level, this](auto& history, auto& stats) {
+    const utime_t now = ceph_clock_now();
+    history.last_scrub = peering_state.get_info().last_update;
+    history.last_scrub_stamp = now;
+    history.last_clean_scrub_stamp = now;
+    if (scrub_level == scrub_level_t::deep) {
+      history.last_deep_scrub = history.last_scrub;
+      history.last_deep_scrub_stamp = now;
+    }
+    // yes, please publish the stats
+    return true;
+  });
+}
+
+void PG::log_state_enter(const char *state) {
+  logger().info("Entering state: {}", state);
+}
+
+void PG::log_state_exit(
+  const char *state_name, utime_t enter_time,
+  uint64_t events, utime_t event_dur) {
+  logger().info(
+    "Exiting state: {}, entered at {}, {} spent on {} events",
+    state_name,
+    enter_time,
+    event_dur,
+    events);
+}
+
+ceph::signedspan PG::get_mnow() const
+{
+  return shard_services.get_mnow();
+}
+
+HeartbeatStampsRef PG::get_hb_stamps(int peer)
+{
+  return shard_services.get_hb_stamps(peer);
+}
+
+void PG::schedule_renew_lease(epoch_t last_peering_reset, ceph::timespan delay)
+{
+  // handle the peering event in the background
+  renew_lease_timer.cancel();
+  renew_lease_timer.set_callback([last_peering_reset, this] {
+    (void) shard_services.start_operation<LocalPeeringEvent>(
+      this,
+      pg_whoami,
+      pgid,
+      last_peering_reset,
+      last_peering_reset,
+      RenewLease{});
+    });
+  renew_lease_timer.arm(
+    std::chrono::duration_cast<seastar::lowres_clock::duration>(delay));
+}
+
+
+void PG::init(
+  int role,
+  const vector<int>& newup, int new_up_primary,
+  const vector<int>& newacting, int new_acting_primary,
+  const pg_history_t& history,
+  const PastIntervals& pi,
+  ObjectStore::Transaction &t)
+{
+  peering_state.init(
+    role, newup, new_up_primary, newacting,
+    new_acting_primary, history, pi, t);
+}
+
+seastar::future<> PG::read_state(crimson::os::FuturizedStore::Shard* store)
+{
+  if (__builtin_expect(stopping, false)) {
+    return seastar::make_exception_future<>(
+	crimson::common::system_shutdown_exception());
+  }
+
+  return seastar::do_with(PGMeta(*store, pgid), [] (auto& pg_meta) {
+    return pg_meta.load();
+  }).then([this, store](auto&& ret) {
+    auto [pg_info, past_intervals] = std::move(ret);
+    return peering_state.init_from_disk_state(
+	std::move(pg_info),
+	std::move(past_intervals),
+	[this, store] (PGLog &pglog) {
+	  return pglog.read_log_and_missing_crimson(
+	    *store,
+	    coll_ref,
+	    peering_state.get_info(),
+	    pgmeta_oid);
+      });
+  }).then([this]() {
+    int primary, up_primary;
+    vector<int> acting, up;
+    peering_state.get_osdmap()->pg_to_up_acting_osds(
+	pgid.pgid, &up, &up_primary, &acting, &primary);
+    peering_state.init_primary_up_acting(
+	up,
+	acting,
+	up_primary,
+	primary);
+    int rr = OSDMap::calc_pg_role(pg_whoami, acting);
+    peering_state.set_role(rr);
+
+    epoch_t epoch = get_osdmap_epoch();
+    (void) shard_services.start_operation<LocalPeeringEvent>(
+	this,
+	pg_whoami,
+	pgid,
+	epoch,
+	epoch,
+	PeeringState::Initialize());
+
+    return seastar::now();
+  });
+}
+
+PG::interruptible_future<> PG::do_peering_event(
+  PGPeeringEvent& evt, PeeringCtx &rctx)
+{
+  if (peering_state.pg_has_reset_since(evt.get_epoch_requested()) ||
+      peering_state.pg_has_reset_since(evt.get_epoch_sent())) {
+    logger().debug("{} ignoring {} -- pg has reset", __func__, evt.get_desc());
+    return interruptor::now();
+  } else {
+    logger().debug("{} handling {} for pg: {}", __func__, evt.get_desc(), pgid);
+    // all peering event handling needs to be run in a dedicated seastar::thread,
+    // so that event processing can involve I/O reqs freely, for example: PG::on_removal,
+    // PG::on_new_interval
+    return interruptor::async([this, &evt, &rctx] {
+      peering_state.handle_event(
+        evt.get_event(),
+        &rctx);
+      peering_state.write_if_dirty(rctx.transaction);
+    });
+  }
+}
+
+seastar::future<> PG::handle_advance_map(
+  cached_map_t next_map, PeeringCtx &rctx)
+{
+  return seastar::async([this, next_map=std::move(next_map), &rctx] {
+    vector<int> newup, newacting;
+    int up_primary, acting_primary;
+    next_map->pg_to_up_acting_osds(
+      pgid.pgid,
+      &newup, &up_primary,
+      &newacting, &acting_primary);
+    peering_state.advance_map(
+      next_map,
+      peering_state.get_osdmap(),
+      newup,
+      up_primary,
+      newacting,
+      acting_primary,
+      rctx);
+    osdmap_gate.got_map(next_map->get_epoch());
+  });
+}
+
+seastar::future<> PG::handle_activate_map(PeeringCtx &rctx)
+{
+  return seastar::async([this, &rctx] {
+    peering_state.activate_map(rctx);
+  });
+}
+
+seastar::future<> PG::handle_initialize(PeeringCtx &rctx)
+{
+  return seastar::async([this, &rctx] {
+    peering_state.handle_event(PeeringState::Initialize{}, &rctx);
+  });
+}
+
+
+void PG::print(ostream& out) const
+{
+  out << peering_state << " ";
+}
+
+void PG::dump_primary(Formatter* f)
+{
+  peering_state.dump_peering_state(f);
+
+  f->open_array_section("recovery_state");
+  PeeringState::QueryState q(f);
+  peering_state.handle_event(q, 0);
+  f->close_section();
+
+  // TODO: snap_trimq
+  // TODO: scrubber state
+  // TODO: agent state
+}
+
+std::ostream& operator<<(std::ostream& os, const PG& pg)
+{
+  os << " pg_epoch " << pg.get_osdmap_epoch() << " ";
+  pg.print(os);
+  return os;
+}
+
+std::tuple<PG::interruptible_future<>,
+           PG::interruptible_future<>>
+PG::submit_transaction(
+  ObjectContextRef&& obc,
+  ceph::os::Transaction&& txn,
+  osd_op_params_t&& osd_op_p,
+  std::vector<pg_log_entry_t>&& log_entries)
+{
+  if (__builtin_expect(stopping, false)) {
+    return {seastar::make_exception_future<>(
+              crimson::common::system_shutdown_exception()),
+            seastar::now()};
+  }
+
+  epoch_t map_epoch = get_osdmap_epoch();
+  ceph_assert(!has_reset_since(osd_op_p.at_version.epoch));
+
+  peering_state.pre_submit_op(obc->obs.oi.soid, log_entries, osd_op_p.at_version);
+  peering_state.append_log_with_trim_to_updated(std::move(log_entries), osd_op_p.at_version,
+						txn, true, false);
+
+  auto [submitted, all_completed] = backend->mutate_object(
+      peering_state.get_acting_recovery_backfill(),
+      std::move(obc),
+      std::move(txn),
+      std::move(osd_op_p),
+      peering_state.get_last_peering_reset(),
+      map_epoch,
+      std::move(log_entries));
+  return std::make_tuple(std::move(submitted), all_completed.then_interruptible(
+    [this, last_complete=peering_state.get_info().last_complete,
+      at_version=osd_op_p.at_version](auto acked) {
+    for (const auto& peer : acked) {
+      peering_state.update_peer_last_complete_ondisk(
+        peer.shard, peer.last_complete_ondisk);
+    }
+    peering_state.complete_write(at_version, last_complete);
+    return seastar::now();
+  }));
+}
+
+PG::interruptible_future<> PG::repair_object(
+  const hobject_t& oid,
+  eversion_t& v) 
+{
+  // see also PrimaryLogPG::rep_repair_primary_object()
+  assert(is_primary());
+  logger().debug("{}: {} peers osd.{}", __func__, oid, get_acting_recovery_backfill());
+  // Add object to PG's missing set if it isn't there already
+  assert(!get_local_missing().is_missing(oid));
+  peering_state.force_object_missing(pg_whoami, oid, v);
+  auto [op, fut] = get_shard_services().start_operation<UrgentRecovery>(
+    oid, v, this, get_shard_services(), get_osdmap_epoch());
+  return std::move(fut);
+}
+
+template <class Ret, class SuccessFunc, class FailureFunc>
+PG::do_osd_ops_iertr::future<PG::pg_rep_op_fut_t<Ret>>
+PG::do_osd_ops_execute(
+  seastar::lw_shared_ptr<OpsExecuter> ox,
+  std::vector<OSDOp>& ops,
+  SuccessFunc&& success_func,
+  FailureFunc&& failure_func)
+{
+  assert(ox);
+  auto rollbacker = ox->create_rollbacker([this] (auto& obc) {
+    return obc_loader.reload_obc(obc).handle_error_interruptible(
+      load_obc_ertr::assert_all{"can't live with object state messed up"});
+  });
+  auto failure_func_ptr = seastar::make_lw_shared(std::move(failure_func));
+  return interruptor::do_for_each(ops, [ox](OSDOp& osd_op) {
+    logger().debug(
+      "do_osd_ops_execute: object {} - handling op {}",
+      ox->get_target(),
+      ceph_osd_op_name(osd_op.op.op));
+    return ox->execute_op(osd_op);
+  }).safe_then_interruptible([this, ox, &ops] {
+    logger().debug(
+      "do_osd_ops_execute: object {} all operations successful",
+      ox->get_target());
+    // check for full
+    if ((ox->delta_stats.num_bytes > 0 ||
+      ox->delta_stats.num_objects > 0) &&
+      get_pgpool().info.has_flag(pg_pool_t::FLAG_FULL)) {
+      const auto& m = ox->get_message();
+      if (m.get_reqid().name.is_mds() ||   // FIXME: ignore MDS for now
+        m.has_flag(CEPH_OSD_FLAG_FULL_FORCE)) {
+        logger().info(" full, but proceeding due to FULL_FORCE or MDS");
+      } else if (m.has_flag(CEPH_OSD_FLAG_FULL_TRY)) {
+        // they tried, they failed.
+        logger().info(" full, replying to FULL_TRY op");
+        if (get_pgpool().info.has_flag(pg_pool_t::FLAG_FULL_QUOTA))
+          return interruptor::make_ready_future<OpsExecuter::rep_op_fut_tuple>(
+            seastar::now(),
+            OpsExecuter::osd_op_ierrorator::future<>(
+              crimson::ct_error::edquot::make()));
+        else
+          return interruptor::make_ready_future<OpsExecuter::rep_op_fut_tuple>(
+            seastar::now(),
+            OpsExecuter::osd_op_ierrorator::future<>(
+              crimson::ct_error::enospc::make()));
+      } else {
+        // drop request
+        logger().info(" full, dropping request (bad client)");
+        return interruptor::make_ready_future<OpsExecuter::rep_op_fut_tuple>(
+          seastar::now(),
+          OpsExecuter::osd_op_ierrorator::future<>(
+            crimson::ct_error::eagain::make()));
+      }
+    }
+    return std::move(*ox).flush_changes_n_do_ops_effects(
+      ops,
+      snap_mapper,
+      osdriver,
+      [this] (auto&& txn,
+              auto&& obc,
+              auto&& osd_op_p,
+              auto&& log_entries) {
+	logger().debug(
+	  "do_osd_ops_execute: object {} submitting txn",
+	  obc->get_oid());
+	return submit_transaction(
+          std::move(obc),
+          std::move(txn),
+          std::move(osd_op_p),
+          std::move(log_entries));
+    });
+  }).safe_then_unpack_interruptible(
+    [success_func=std::move(success_func), rollbacker, this, failure_func_ptr]
+    (auto submitted_fut, auto all_completed_fut) mutable {
+    return PG::do_osd_ops_iertr::make_ready_future<pg_rep_op_fut_t<Ret>>(
+        std::move(submitted_fut),
+        all_completed_fut.safe_then_interruptible_tuple(
+          std::move(success_func),
+          crimson::ct_error::object_corrupted::handle(
+            [rollbacker, this] (const std::error_code& e) mutable {
+            // this is a path for EIO. it's special because we want to fix the obejct
+            // and try again. that is, the layer above `PG::do_osd_ops` is supposed to
+            // restart the execution.
+            return rollbacker.rollback_obc_if_modified(e).then_interruptible(
+              [obc=rollbacker.get_obc(), this] {
+              return repair_object(obc->obs.oi.soid,
+                                   obc->obs.oi.version).then_interruptible([] {
+                return do_osd_ops_iertr::future<Ret>{crimson::ct_error::eagain::make()};
+              });
+            });
+          }), OpsExecuter::osd_op_errorator::all_same_way(
+            [rollbacker, failure_func_ptr]
+            (const std::error_code& e) mutable {
+            return rollbacker.rollback_obc_if_modified(e).then_interruptible(
+              [e, failure_func_ptr] {
+              return (*failure_func_ptr)(e);
+            });
+          })
+        )
+      );
+  }, OpsExecuter::osd_op_errorator::all_same_way(
+    [rollbacker, failure_func_ptr]
+    (const std::error_code& e) mutable {
+    return PG::do_osd_ops_iertr::make_ready_future<pg_rep_op_fut_t<Ret>>(
+        seastar::now(),
+        e.value() == ENOENT ? (*failure_func_ptr)(e) :
+        rollbacker.rollback_obc_if_modified(e).then_interruptible(
+          [e, failure_func_ptr] {
+          return (*failure_func_ptr)(e);
+        }));
+  }));
+}
+seastar::future<> PG::submit_error_log(
+  Ref<MOSDOp> m,
+  const OpInfo &op_info,
+  ObjectContextRef obc,
+  const std::error_code e,
+  ceph_tid_t rep_tid,
+  eversion_t &version)
+{
+  const osd_reqid_t &reqid = m->get_reqid();
+  mempool::osd_pglog::list<pg_log_entry_t> log_entries;
+  log_entries.push_back(pg_log_entry_t(pg_log_entry_t::ERROR,
+                                       obc->obs.oi.soid,
+                                       next_version(),
+                                       eversion_t(), 0,
+                                       reqid, utime_t(),
+                                       -e.value()));
+  if (op_info.allows_returnvec()) {
+    log_entries.back().set_op_returns(m->ops);
+  }
+  ceph_assert(is_primary());
+  if (!log_entries.empty()) {
+    ceph_assert(log_entries.rbegin()->version >= projected_last_update);
+    version = projected_last_update = log_entries.rbegin()->version;
+  }
+  ceph::os::Transaction t;
+  peering_state.merge_new_log_entries(
+    log_entries, t, peering_state.get_pg_trim_to(),
+    peering_state.get_min_last_complete_ondisk());
+
+    set<pg_shard_t> waiting_on;
+    for (auto &i : get_acting_recovery_backfill()) {
+      pg_shard_t peer(i);
+      if (peer == pg_whoami) continue;
+      ceph_assert(peering_state.get_peer_missing().count(peer));
+      ceph_assert(peering_state.has_peer_info(peer));
+      auto log_m = crimson::make_message<MOSDPGUpdateLogMissing>(
+                   log_entries,
+                   spg_t(peering_state.get_info().pgid.pgid, i.shard),
+                   pg_whoami.shard,
+                   get_osdmap_epoch(),
+                   get_last_peering_reset(),
+                   rep_tid,
+                   peering_state.get_pg_trim_to(),
+                   peering_state.get_min_last_complete_ondisk());
+      send_cluster_message(peer.osd, std::move(log_m), get_osdmap_epoch());
+      waiting_on.insert(peer);
+    }
+    waiting_on.insert(pg_whoami);
+    log_entry_update_waiting_on.insert(
+      std::make_pair(rep_tid, log_update_t{std::move(waiting_on)}));
+    return shard_services.get_store().do_transaction(
+      get_collection_ref(), std::move(t))
+      .then([this] {
+        peering_state.update_trim_to();
+        return seastar::now();
+    });
+}
+
+PG::do_osd_ops_iertr::future<PG::pg_rep_op_fut_t<MURef<MOSDOpReply>>>
+PG::do_osd_ops(
+  Ref<MOSDOp> m,
+  crimson::net::ConnectionRef conn,
+  ObjectContextRef obc,
+  const OpInfo &op_info,
+  const SnapContext& snapc)
+{
+  if (__builtin_expect(stopping, false)) {
+    throw crimson::common::system_shutdown_exception();
+  }
+  return do_osd_ops_execute<MURef<MOSDOpReply>>(
+    seastar::make_lw_shared<OpsExecuter>(
+      Ref<PG>{this}, obc, op_info, *m, conn, snapc),
+    m->ops,
+    [this, m, obc, may_write = op_info.may_write(),
+     may_read = op_info.may_read(), rvec = op_info.allows_returnvec()] {
+      // TODO: should stop at the first op which returns a negative retval,
+      //       cmpext uses it for returning the index of first unmatched byte
+      int result = m->ops.empty() ? 0 : m->ops.back().rval.code;
+      if (may_read && result >= 0) {
+        for (auto &osdop : m->ops) {
+          if (osdop.rval < 0 && !(osdop.op.flags & CEPH_OSD_OP_FLAG_FAILOK)) {
+            result = osdop.rval.code;
+            break;
+          }
+        }
+      } else if (result > 0 && may_write && !rvec) {
+        result = 0;
+      } else if (result < 0 && (m->ops.empty() ?
+        0 : m->ops.back().op.flags & CEPH_OSD_OP_FLAG_FAILOK)) {
+        result = 0;
+      }
+      auto reply = crimson::make_message<MOSDOpReply>(m.get(),
+                                             result,
+                                             get_osdmap_epoch(),
+                                             0,
+                                             false);
+      reply->add_flags(CEPH_OSD_FLAG_ACK | CEPH_OSD_FLAG_ONDISK);
+      logger().debug(
+        "do_osd_ops: {} - object {} sending reply",
+        *m,
+        m->get_hobj());
+      if (obc->obs.exists) {
+        reply->set_reply_versions(peering_state.get_info().last_update,
+          obc->obs.oi.user_version);
+      } else {
+        reply->set_reply_versions(peering_state.get_info().last_update,
+          peering_state.get_info().last_user_version);
+      }
+      return do_osd_ops_iertr::make_ready_future<MURef<MOSDOpReply>>(
+        std::move(reply));
+    },
+    [m, &op_info, obc, this] (const std::error_code& e) {
+    return seastar::do_with(eversion_t(), [m, &op_info, obc, e, this](auto &version) {
+      auto fut = seastar::now();
+      epoch_t epoch = get_osdmap_epoch();
+      ceph_tid_t rep_tid = shard_services.get_tid();
+      auto last_complete = peering_state.get_info().last_complete;
+      if (op_info.may_write()) {
+        fut = submit_error_log(m, op_info, obc, e, rep_tid, version);
+      }
+      return fut.then([m, e, epoch, &op_info, rep_tid, &version, last_complete,  this] {
+        auto log_reply = [m, e, this] {
+          auto reply = crimson::make_message<MOSDOpReply>(
+            m.get(), -e.value(), get_osdmap_epoch(), 0, false);
+          if (m->ops.empty() ? 0 :
+            m->ops.back().op.flags & CEPH_OSD_OP_FLAG_FAILOK) {
+            reply->set_result(0);
+          }
+          // For all ops except for CMPEXT, the correct error value is encoded
+          // in e.value(). For CMPEXT, osdop.rval has the actual error value.
+          if (e.value() == ct_error::cmp_fail_error_value) {
+            assert(!m->ops.empty());
+            for (auto &osdop : m->ops) {
+              if (osdop.rval < 0) {
+                reply->set_result(osdop.rval);
+                break;
+              }
+            }
+          }
+          reply->set_enoent_reply_versions(
+          peering_state.get_info().last_update,
+          peering_state.get_info().last_user_version);
+          reply->add_flags(CEPH_OSD_FLAG_ACK | CEPH_OSD_FLAG_ONDISK);
+          return do_osd_ops_iertr::make_ready_future<MURef<MOSDOpReply>>(
+            std::move(reply));
+	};
+
+        if (!peering_state.pg_has_reset_since(epoch) && op_info.may_write()) {
+          auto it = log_entry_update_waiting_on.find(rep_tid);
+          ceph_assert(it != log_entry_update_waiting_on.end());
+          auto it2 = it->second.waiting_on.find(pg_whoami);
+          ceph_assert(it2 != it->second.waiting_on.end());
+          it->second.waiting_on.erase(it2);
+
+          if (it->second.waiting_on.empty()) {
+            log_entry_update_waiting_on.erase(it);
+            if (version != eversion_t()) {
+              peering_state.complete_write(version, last_complete);
+            }
+            return log_reply();
+          } else {
+            return it->second.all_committed.get_shared_future()
+              .then([this, &version, last_complete, log_reply = std::move(log_reply)] {
+              if (version != eversion_t()) {
+                peering_state.complete_write(version, last_complete);
+              }
+              return log_reply();
+            });
+          }
+        } else {
+          return log_reply();
+        }
+      });
+    });
+  });
+}
+
+PG::do_osd_ops_iertr::future<PG::pg_rep_op_fut_t<>>
+PG::do_osd_ops(
+  ObjectContextRef obc,
+  std::vector<OSDOp>& ops,
+  const OpInfo &op_info,
+  const do_osd_ops_params_t &&msg_params,
+  do_osd_ops_success_func_t success_func,
+  do_osd_ops_failure_func_t failure_func)
+{
+  // This overload is generally used for internal client requests,
+  // use an empty SnapContext.
+  return seastar::do_with(
+    std::move(msg_params),
+    [=, this, &ops, &op_info](auto &msg_params) {
+    return do_osd_ops_execute<void>(
+      seastar::make_lw_shared<OpsExecuter>(
+        Ref<PG>{this},
+        std::move(obc),
+        op_info,
+        msg_params,
+        msg_params.get_connection(),
+        SnapContext{}
+      ),
+      ops,
+      std::move(success_func),
+      std::move(failure_func));
+  });
+}
+
+PG::interruptible_future<MURef<MOSDOpReply>> PG::do_pg_ops(Ref<MOSDOp> m)
+{
+  if (__builtin_expect(stopping, false)) {
+    throw crimson::common::system_shutdown_exception();
+  }
+
+  auto ox = std::make_unique<PgOpsExecuter>(std::as_const(*this),
+                                            std::as_const(*m));
+  return interruptor::do_for_each(m->ops, [ox = ox.get()](OSDOp& osd_op) {
+    logger().debug("will be handling pg op {}", ceph_osd_op_name(osd_op.op.op));
+    return ox->execute_op(osd_op);
+  }).then_interruptible([m, this, ox = std::move(ox)] {
+    auto reply = crimson::make_message<MOSDOpReply>(m.get(), 0, get_osdmap_epoch(),
+                                           CEPH_OSD_FLAG_ACK | CEPH_OSD_FLAG_ONDISK,
+                                           false);
+    reply->claim_op_out_data(m->ops);
+    reply->set_reply_versions(peering_state.get_info().last_update,
+      peering_state.get_info().last_user_version);
+    return seastar::make_ready_future<MURef<MOSDOpReply>>(std::move(reply));
+  }).handle_exception_type_interruptible([=, this](const crimson::osd::error& e) {
+    auto reply = crimson::make_message<MOSDOpReply>(
+      m.get(), -e.code().value(), get_osdmap_epoch(), 0, false);
+    reply->set_enoent_reply_versions(peering_state.get_info().last_update,
+				     peering_state.get_info().last_user_version);
+    return seastar::make_ready_future<MURef<MOSDOpReply>>(std::move(reply));
+  });
+}
+
+hobject_t PG::get_oid(const hobject_t& hobj)
+{
+  return hobj.snap == CEPH_SNAPDIR ? hobj.get_head() : hobj;
+}
+
+RWState::State PG::get_lock_type(const OpInfo &op_info)
+{
+
+  if (op_info.rwordered() && op_info.may_read()) {
+    return RWState::RWEXCL;
+  } else if (op_info.rwordered()) {
+    return RWState::RWWRITE;
+  } else {
+    ceph_assert(op_info.may_read());
+    return RWState::RWREAD;
+  }
+}
+
+void PG::check_blocklisted_obc_watchers(
+  ObjectContextRef &obc)
+{
+  if (obc->watchers.empty()) {
+    for (auto &[src, winfo] : obc->obs.oi.watchers) {
+      auto watch = crimson::osd::Watch::create(
+        obc, winfo, src.second, this);
+      watch->disconnect();
+      auto [it, emplaced] = obc->watchers.emplace(src, std::move(watch));
+      assert(emplaced);
+      logger().debug("added watch for obj {}, client {}",
+        obc->get_oid(), src.second);
+    }
+  }
+}
+
+PG::load_obc_iertr::future<>
+PG::with_locked_obc(const hobject_t &hobj,
+                    const OpInfo &op_info,
+                    with_obc_func_t &&f)
+{
+  if (__builtin_expect(stopping, false)) {
+    throw crimson::common::system_shutdown_exception();
+  }
+  const hobject_t oid = get_oid(hobj);
+  auto wrapper = [f=std::move(f), this](auto obc) {
+    check_blocklisted_obc_watchers(obc);
+    return f(obc);
+  };
+  switch (get_lock_type(op_info)) {
+  case RWState::RWREAD:
+      return obc_loader.with_obc<RWState::RWREAD>(oid, std::move(wrapper));
+  case RWState::RWWRITE:
+      return obc_loader.with_obc<RWState::RWWRITE>(oid, std::move(wrapper));
+  case RWState::RWEXCL:
+      return obc_loader.with_obc<RWState::RWEXCL>(oid, std::move(wrapper));
+  default:
+    ceph_abort();
+  };
+}
+
+PG::interruptible_future<> PG::handle_rep_op(Ref<MOSDRepOp> req)
+{
+  if (__builtin_expect(stopping, false)) {
+    return seastar::make_exception_future<>(
+	crimson::common::system_shutdown_exception());
+  }
+
+  logger().debug("{}: {}", __func__, *req);
+  if (can_discard_replica_op(*req)) {
+    return seastar::now();
+  }
+
+  ceph::os::Transaction txn;
+  auto encoded_txn = req->get_data().cbegin();
+  decode(txn, encoded_txn);
+  auto p = req->logbl.cbegin();
+  std::vector<pg_log_entry_t> log_entries;
+  decode(log_entries, p);
+  log_operation(std::move(log_entries),
+                req->pg_trim_to,
+                req->version,
+                req->min_last_complete_ondisk,
+                !txn.empty(),
+                txn,
+                false);
+  logger().debug("PG::handle_rep_op: do_transaction...");
+  return interruptor::make_interruptible(shard_services.get_store().do_transaction(
+	coll_ref, std::move(txn))).then_interruptible(
+      [req, lcod=peering_state.get_info().last_complete, this] {
+      peering_state.update_last_complete_ondisk(lcod);
+      const auto map_epoch = get_osdmap_epoch();
+      auto reply = crimson::make_message<MOSDRepOpReply>(
+        req.get(), pg_whoami, 0,
+	map_epoch, req->get_min_epoch(), CEPH_OSD_FLAG_ONDISK);
+      reply->set_last_complete_ondisk(lcod);
+      return shard_services.send_to_osd(req->from.osd, std::move(reply), map_epoch);
+    });
+}
+
+void PG::log_operation(
+  std::vector<pg_log_entry_t>&& logv,
+  const eversion_t &trim_to,
+  const eversion_t &roll_forward_to,
+  const eversion_t &min_last_complete_ondisk,
+  bool transaction_applied,
+  ObjectStore::Transaction &txn,
+  bool async) {
+  logger().debug("{}", __func__);
+  if (is_primary()) {
+    ceph_assert(trim_to <= peering_state.get_last_update_ondisk());
+  }
+  /* TODO: when we add snap mapper and projected log support,
+   * we'll likely want to update them here.
+   *
+   * See src/osd/PrimaryLogPG.h:log_operation for how classic
+   * handles these cases.
+   */
+#if 0
+  if (transaction_applied) {
+    //TODO:
+    //update_snap_map(logv, t);
+  }
+  auto last = logv.rbegin();
+  if (is_primary() && last != logv.rend()) {
+    projected_log.skip_can_rollback_to_to_head();
+    projected_log.trim(cct, last->version, nullptr, nullptr, nullptr);
+  }
+#endif
+  if (!is_primary()) { // && !is_ec_pg()
+    replica_clear_repop_obc(logv);
+  }
+  peering_state.append_log(std::move(logv),
+                           trim_to,
+                           roll_forward_to,
+                           min_last_complete_ondisk,
+                           txn,
+                           !txn.empty(),
+                           false);
+}
+
+void PG::replica_clear_repop_obc(
+  const std::vector<pg_log_entry_t> &logv) {
+    logger().debug("{} clearing {} entries", __func__, logv.size());
+    for (auto &&e: logv) {
+      logger().debug(" {} get_object_boundary(from): {} "
+                     " head version(to): {}",
+                     e.soid,
+                     e.soid.get_object_boundary(),
+                     e.soid.get_head());
+    /* Have to blast all clones, they share a snapset */
+    obc_registry.clear_range(
+      e.soid.get_object_boundary(), e.soid.get_head());
+  }
+}
+
+void PG::handle_rep_op_reply(const MOSDRepOpReply& m)
+{
+  if (!can_discard_replica_op(m)) {
+    backend->got_rep_op_reply(m);
+  }
+}
+
+PG::interruptible_future<> PG::do_update_log_missing(
+  Ref<MOSDPGUpdateLogMissing> m,
+  crimson::net::ConnectionRef conn)
+{
+  if (__builtin_expect(stopping, false)) {
+    return seastar::make_exception_future<>(
+      crimson::common::system_shutdown_exception());
+  }
+
+  ceph_assert(m->get_type() == MSG_OSD_PG_UPDATE_LOG_MISSING);
+  ObjectStore::Transaction t;
+  std::optional<eversion_t> op_trim_to, op_roll_forward_to;
+  if (m->pg_trim_to != eversion_t())
+    op_trim_to = m->pg_trim_to;
+  if (m->pg_roll_forward_to != eversion_t())
+    op_roll_forward_to = m->pg_roll_forward_to;
+  logger().debug("op_trim_to = {}, op_roll_forward_to = {}",
+    op_trim_to, op_roll_forward_to);
+
+  peering_state.append_log_entries_update_missing(
+    m->entries, t, op_trim_to, op_roll_forward_to);
+
+  return interruptor::make_interruptible(shard_services.get_store().do_transaction(
+    coll_ref, std::move(t))).then_interruptible(
+    [m, conn, lcod=peering_state.get_info().last_complete, this] {
+    if (!peering_state.pg_has_reset_since(m->get_epoch())) {
+      peering_state.update_last_complete_ondisk(lcod);
+      auto reply =
+        crimson::make_message<MOSDPGUpdateLogMissingReply>(
+          spg_t(peering_state.get_info().pgid.pgid, get_primary().shard),
+          pg_whoami.shard,
+          m->get_epoch(),
+          m->min_epoch,
+          m->get_tid(),
+          lcod);
+      reply->set_priority(CEPH_MSG_PRIO_HIGH);
+      return conn->send(std::move(reply));
+    }
+    return seastar::now();
+  });
+}
+
+
+PG::interruptible_future<> PG::do_update_log_missing_reply(
+  Ref<MOSDPGUpdateLogMissingReply> m)
+{
+  logger().debug("{}: got reply from {}", __func__, m->get_from());
+
+  auto it = log_entry_update_waiting_on.find(m->get_tid());
+  if (it != log_entry_update_waiting_on.end()) {
+    if (it->second.waiting_on.count(m->get_from())) {
+      it->second.waiting_on.erase(m->get_from());
+      if (m->last_complete_ondisk != eversion_t()) {
+        peering_state.update_peer_last_complete_ondisk(
+	  m->get_from(), m->last_complete_ondisk);
+      }
+    } else {
+      logger().error("{} : {} got reply {} from shard we are not waiting for ",
+        __func__, peering_state.get_info().pgid, *m, m->get_from());
+    }
+
+    if (it->second.waiting_on.empty()) {
+      it->second.all_committed.set_value();
+      it->second.all_committed = {};
+      log_entry_update_waiting_on.erase(it);
+    }
+  } else {
+    logger().error("{} : {} got reply {} on unknown tid {}",
+      __func__, peering_state.get_info().pgid, *m, m->get_tid());
+  }
+  return seastar::now();
+}
+
+bool PG::old_peering_msg(
+  const epoch_t reply_epoch,
+  const epoch_t query_epoch) const
+{
+  if (const epoch_t lpr = peering_state.get_last_peering_reset();
+      lpr > reply_epoch || lpr > query_epoch) {
+    logger().debug("{}: pg changed {} lpr {}, reply_epoch {}, query_epoch {}",
+                   __func__, get_info().history, lpr, reply_epoch, query_epoch);
+    return true;
+  }
+  return false;
+}
+
+bool PG::can_discard_replica_op(const Message& m, epoch_t m_map_epoch) const
+{
+  // if a repop is replied after a replica goes down in a new osdmap, and
+  // before the pg advances to this new osdmap, the repop replies before this
+  // repop can be discarded by that replica OSD, because the primary resets the
+  // connection to it when handling the new osdmap marking it down, and also
+  // resets the messenger sesssion when the replica reconnects. to avoid the
+  // out-of-order replies, the messages from that replica should be discarded.
+  const auto osdmap = peering_state.get_osdmap();
+  const int from_osd = m.get_source().num();
+  if (osdmap->is_down(from_osd)) {
+    return true;
+  }
+  // Mostly, this overlaps with the old_peering_msg
+  // condition.  An important exception is pushes
+  // sent by replicas not in the acting set, since
+  // if such a replica goes down it does not cause
+  // a new interval.
+  if (osdmap->get_down_at(from_osd) >= m_map_epoch) {
+    return true;
+  }
+  // same pg?
+  //  if pg changes *at all*, we reset and repeer!
+  return old_peering_msg(m_map_epoch, m_map_epoch);
+}
+
+seastar::future<> PG::stop()
+{
+  logger().info("PG {} {}", pgid, __func__);
+  stopping = true;
+  cancel_local_background_io_reservation();
+  cancel_remote_recovery_reservation();
+  check_readable_timer.cancel();
+  renew_lease_timer.cancel();
+  return osdmap_gate.stop().then([this] {
+    return wait_for_active_blocker.stop();
+  }).then([this] {
+    return recovery_handler->stop();
+  }).then([this] {
+    return recovery_backend->stop();
+  }).then([this] {
+    return backend->stop();
+  });
+}
+
+void PG::on_change(ceph::os::Transaction &t) {
+  logger().debug("{} {}:", *this, __func__);
+  context_registry_on_change();
+  obc_loader.notify_on_change(is_primary());
+  recovery_backend->on_peering_interval_change(t);
+  backend->on_actingset_changed(is_primary());
+  wait_for_active_blocker.unblock();
+  if (is_primary()) {
+    logger().debug("{} {}: requeueing", *this, __func__);
+    client_request_orderer.requeue(shard_services, this);
+  } else {
+    logger().debug("{} {}: dropping requests", *this, __func__);
+    client_request_orderer.clear_and_cancel();
+  }
+}
+
+void PG::context_registry_on_change() {
+    obc_registry.for_each([](ObjectContextRef obc) {
+      assert(obc);
+      for (auto j = obc->watchers.begin();
+           j != obc->watchers.end();
+           j = obc->watchers.erase(j)) {
+        j->second->discard_state();
+      }
+  });
+}
+
+bool PG::can_discard_op(const MOSDOp& m) const {
+  if (m.get_map_epoch() <
+      peering_state.get_info().history.same_primary_since) {
+    logger().debug("{} changed after {} dropping {} ",
+                   __func__ , m.get_map_epoch(), m);
+    return true;
+  }
+
+  if ((m.get_flags() & (CEPH_OSD_FLAG_BALANCE_READS |
+                        CEPH_OSD_FLAG_LOCALIZE_READS))
+    && !is_primary()
+    && (m.get_map_epoch() <
+        peering_state.get_info().history.same_interval_since))
+    {
+      // Note: the Objecter will resend on interval change without the primary
+      // changing if it actually sent to a replica.  If the primary hasn't
+      // changed since the send epoch, we got it, and we're primary, it won't
+      // have resent even if the interval did change as it sent it to the primary
+      // (us).
+      return true;
+    }
+  return __builtin_expect(m.get_map_epoch()
+      < peering_state.get_info().history.same_primary_since, false);
+}
+
+bool PG::is_degraded_or_backfilling_object(const hobject_t& soid) const {
+  /* The conditions below may clear (on_local_recover, before we queue
+   * the transaction) before we actually requeue the degraded waiters
+   * in on_global_recover after the transaction completes.
+   */
+  if (peering_state.get_pg_log().get_missing().get_items().count(soid))
+    return true;
+  ceph_assert(!get_acting_recovery_backfill().empty());
+  for (auto& peer : get_acting_recovery_backfill()) {
+    if (peer == get_primary()) continue;
+    auto peer_missing_entry = peering_state.get_peer_missing().find(peer);
+    // If an object is missing on an async_recovery_target, return false.
+    // This will not block the op and the object is async recovered later.
+    if (peer_missing_entry != peering_state.get_peer_missing().end() &&
+	peer_missing_entry->second.get_items().count(soid)) {
+	return true;
+    }
+    // Object is degraded if after last_backfill AND
+    // we are backfilling it
+    if (is_backfill_target(peer) &&
+        peering_state.get_peer_info(peer).last_backfill <= soid &&
+        recovery_handler->backfill_state &&
+	recovery_handler->backfill_state->get_last_backfill_started() >= soid &&
+	recovery_backend->is_recovering(soid)) {
+      return true;
+    }
+  }
+  return false;
+}
+
+PG::interruptible_future<std::optional<PG::complete_op_t>>
+PG::already_complete(const osd_reqid_t& reqid)
+{
+  eversion_t version;
+  version_t user_version;
+  int ret;
+  std::vector<pg_log_op_return_item_t> op_returns;
+
+  if (peering_state.get_pg_log().get_log().get_request(
+	reqid, &version, &user_version, &ret, &op_returns)) {
+    complete_op_t dupinfo{
+      user_version,
+      version,
+      ret};
+    return backend->request_committed(reqid, version).then([dupinfo] {
+      return seastar::make_ready_future<std::optional<complete_op_t>>(dupinfo);
+    });
+  } else {
+    return seastar::make_ready_future<std::optional<complete_op_t>>(std::nullopt);
+  }
+}
+
+}