crimson/osd: make osd_op_params::at_version coherent with last log entry

Before this commit we were doing something like:

1. initialize `at_version` with PG::projected_last_update`
   **incremented by one**.
2. produce a log entry at such version.
3. increment `at_version` for the sake of a further production
   that may never come.

The problem is `osd_op_params::at_version` is higher by one
than the last log entry which hurts at later stages of
`osd_op_params` processing (I was hit in the shared EC code
by the assertion in `PG::op_applied`).

This patch changes the algorithm to:

A. initialize `at_version` with PG::projected_last_update`
   **incremented by one**.
B. increment `at_version` for the sake of the very next production.
C. produce a log entry at this version.

Co-authored-by: Matan Breizman <[email protected]>
Signed-off-by: Radoslaw Zarzynski <[email protected]>

Author: rzarzynski

src/crimson/osd/ops_executer.cc

@@ -819,6 +819,7 @@ std::vector<pg_log_entry_t> OpsExecuter::prepare_transaction(
   // entry.
   assert(obc->obs.oi.soid.snap >= CEPH_MAXSNAP);
   std::vector<pg_log_entry_t> log_entries;
+  osd_op_params->at_version.version++;
   log_entries.emplace_back(
     obc->obs.exists ?
       pg_log_entry_t::MODIFY : pg_log_entry_t::DELETE,
@@ -829,7 +830,6 @@ std::vector<pg_log_entry_t> OpsExecuter::prepare_transaction(
     osd_op_params->req_id,
     osd_op_params->mtime,
     op_info.allows_returnvec() && !ops.empty() ? ops.back().rval.code : 0);
-  osd_op_params->at_version.version++;
   if (op_info.allows_returnvec()) {
     // also the per-op values are recorded in the pg log
     log_entries.back().set_op_returns(ops);
@@ -1042,15 +1042,14 @@ ObjectContextRef OpsExecuter::prepare_clone(
   const hobject_t& coid)
 {
   ceph_assert(pg->is_primary());
+  osd_op_params->at_version.version++;
   ObjectState clone_obs{coid};
   clone_obs.exists = true;
   clone_obs.oi.version = osd_op_params->at_version;
   clone_obs.oi.prior_version = obc->obs.oi.version;
   clone_obs.oi.copy_user_bits(obc->obs.oi);
   clone_obs.oi.clear_flag(object_info_t::FLAG_WHITEOUT);
 
-  osd_op_params->at_version.version++;
-
   auto [clone_obc, existed] = pg->obc_registry.get_cached_obc(std::move(coid));
   ceph_assert(!existed);
 

src/crimson/osd/ops_executer.h

@@ -529,6 +529,9 @@ OpsExecuter::flush_changes_n_do_ops_effects(
       txn
     ).then_interruptible([mut_func=std::move(mut_func),
                           this](auto&& log_entries) mutable {
+      if (auto log_rit = log_entries.rbegin(); log_rit != log_entries.rend()) {
+        ceph_assert(log_rit->version == osd_op_params->at_version);
+      }
       auto [submitted, all_completed] =
         std::forward<MutFunc>(mut_func)(std::move(txn),
                                         std::move(obc),