[#33444] DocDB: Orchestrate the index-backfill ordering-generation lifecycle from the master

The persisted ordering generation (#33580) had no production lifecycle: nothing
activated it before marked backfill writes, nothing released it, and the
master-side split suppression protecting a backfill is in-memory only (lost on
failover). This part wires the full lifecycle and flips marked-write gating
fail-closed.

Activation rides ChangeMetadataOperation (a new
`ChangeMetadataRequestPB.index_backfill_ordering_generation` variant, applied
via `Tablet::UpdateIndexBackfillOrderingGeneration` exactly like
`mark_backfill_done`): Raft-replicated, follower-applied, WAL-replayed, and
`base_op_index` is the activation operation's own Raft index, so every replica
and every replay derives the same base with no master-side bookkeeping.
Re-activation (failover resume) is idempotent -- the base only moves up.

Master flow, SKIP_ALL jobs on PGSQL indexed tables only (the mode rides YSQL
chunk requests; YCQL backfill writes are never marked, so generations would
only fence their splits without protecting anything):

- Before the first chunk (`DoBackfill`), the job disables and drains
  index-table splitting, then fans out one waited
  `UpdateOrderingGenerationForTablet` task per index tablet (the
  `GetSafeTimeForTablet` join pattern). All acks -> chunks launch; any failure
  -> the job aborts before a single marked write exists, and CREATE INDEX
  fails cleanly. The drain closes the activation/split TOCTOU: consensus
  already rejects CHANGE_METADATA_OP while a split is pending, the generation
  fence rejects splits after activation applies, and a split appended in the
  append-to-apply window can at worst fail the activation task -- never
  corrupt a scan set that marked writes exist in.
- Release is fire-and-forget from the terminal funnel
  (`UpdateIndexPermissionsForIndexes`), which every success/failure/abort path
  traverses. The tserver metadata validator converges stragglers, mirroring
  the retain_delete_markers machinery: tablets with an active generation join
  its existing GetBackfillStatus poll, and the generation is released locally
  (no Raft) once the master reports a terminal state. `IndexStatusPB` gains
  `BACKFILL_FAILED` (removal-path index permissions map to it) so a failed
  SKIP_ALL job whose funnel release was lost cannot leave an orphaned invalid
  index permanently split-fenced and retention-pinned; the
  retain_delete_markers heal itself stays success-only.
- The tablet-split manager refuses to split an index whose indexed table has a
  durable SKIP_ALL backfill job. The manager's table validation takes no
  catalog locks itself -- the indexed table is a caller-resolved parameter,
  because the manual-split path enters through ValidateSplitCandidateUnlocked
  with the catalog mutex already held (a recursive shared acquisition is fatal
  under lock_debug) (`SysTablesEntryPB.backfill_jobs`, survives
  failover, cleared in the funnel), with an INFO-level skip reason -- today's
  two in-memory suppressions cover only the indexed table and evaporate on
  failover. The tablet-side generation fence stays as the fail-closed layer.

Gating flip: `WriteOperation::ValidateLeaderOpId` now rejects marked writes
with *no* active generation (previously deferred) -- a marked write outside a
generation would store versions nothing tracks or releases (e.g. a stale chunk
retry after the job's terminal state). Tests that drive marked writes directly
now activate a generation first.

`write_id_floor_version` numbering lands as
`kIndexBackfillWriteIdFloorVersion = 1`.

The mode selector remains hardcoded to CHECK_ALL (#33484), so this whole flow
stays production-unreachable; tests drive it through the master-side TEST
override.

Test Plan:
./yb_build.sh release --cxx-test tablet_peer-test --gtest_filter 'TabletPeerTest.OrderingGenerationChangeMetadataOpSetsBaseFromOwnRaftIndex'
./yb_build.sh release --cxx-test tablet_peer-test --gtest_filter 'TabletPeerTest.FixedHybridTimeWriteRejectedAtOrBelowGenerationBase'
./yb_build.sh release --cxx-test pg_index_backfill-test --gtest_filter 'PgIndexBackfillSkipAllRaftOrdering.OrderingGenerationActivatedAndReleased/0'
./yb_build.sh release --cxx-test pg_index_backfill-test --gtest_filter 'PgIndexBackfillSkipAllActivationFailure.ActivationFailureFailsCreateIndexCleanly/0'
./yb_build.sh release --cxx-test pg_index_backfill-test --gtest_filter 'PgIndexBackfillSkipAllBlocked.SplitFencedDuringBackfill/0'
Retrofitted (marked writes now require an active generation):
./yb_build.sh release --cxx-test tablet_peer-test --gtest_filter 'TabletPeerTest.FixedHybridTimeWrite*'
./yb_build.sh release --cxx-test fixed_hybrid_time_write_id-itest
Regressions: skip_all e2e + marker canary through the real activation flow,
uniq-idx-1 mode tests, DuplicatesExistBeforeBackfill, 3b-i generation tests,
0a/0b abort tests, and drop-path tests for the permissions mapping change
(PgIndexBackfillTest.Drop, PgIndexBackfillFastClientTimeout.DropWhileBackfilling)
-- all green.

Assisted-By: devx/08801cf6-2854-4c6f-a378-1d98d239b22d

---
_automated · Claude Fable 5 (opencode)_

Author: jmeehan16

src/yb/common/doc_hybrid_time.h

@@ -54,6 +54,12 @@ constexpr IntraTxnWriteId kBackfillWriteIdFloor = kIntraTxnWriteIdLimit;
 constexpr IntraTxnWriteId kBackfillWriteIdIndexMax = kMaxWriteId - kBackfillWriteIdFloor - 1;
 static_assert((kBackfillWriteIdFloor | kBackfillWriteIdIndexMax) == kMaxWriteId - 1);
 
+// Version of the floor scheme recorded in each ordering generation
+// (IndexBackfillOrderingGenerationPB.write_id_floor_version): 1 = the scheme above; 0 is
+// reserved as absent/unknown. Bump if the floor or the derivation ever changes, so marked data
+// written under an older scheme cannot be misinterpreted.
+constexpr uint32_t kIndexBackfillWriteIdFloorVersion = 1;
+
 // An aggressive upper bound on the length of a DocDB-encoded hybrid time with a write id.
 // This could happen in the degenerate case when all three VarInts in encoded representation of a
 // DocHybridTime take 10 bytes (the maximum length for a VarInt-encoded int64_t).

src/yb/integration-tests/fixed_hybrid_time_write_id-itest.cc

@@ -285,6 +285,7 @@ TEST_F(FixedHybridTimeWriteIdITest, LeaderChangePreservesDistinctWriteIds) {
   ASSERT_OK(WaitUntilTabletHasLeader(
       cluster_.get(), tablet_id, CoarseMonoClock::Now() + 10s * kTimeMultiplier,
       RequireLeaderIsReady::kTrue));
+  ASSERT_OK(SetOrderingGenerationOnAllReplicas(tablet_id, MakeActiveOrderingGeneration()));
 
   // W1 under the original leader: two distinct keys in one marked op share the op's Raft-index
   // write ID ("dup" is the key W2 collides with later; "stable" pins the shared-ID property).
@@ -344,6 +345,7 @@ TEST_F(FixedHybridTimeWriteIdITest, MasterSplitRefusedWhileOrderingGenerationAct
   ASSERT_OK(WaitUntilTabletHasLeader(
       cluster_.get(), tablet_id, CoarseMonoClock::Now() + 10s * kTimeMultiplier,
       RequireLeaderIsReady::kTrue));
+  ASSERT_OK(SetOrderingGenerationOnAllReplicas(tablet_id, MakeActiveOrderingGeneration()));
 
   // Give the tablet hash-spread data and SSTs so it is a viable split candidate.
   const auto op_id = ASSERT_RESULT(
@@ -355,9 +357,9 @@ TEST_F(FixedHybridTimeWriteIdITest, MasterSplitRefusedWhileOrderingGenerationAct
   const auto table_id = table_.table()->id();
   ASSERT_EQ(ListActiveTabletIdsForTable(cluster_.get(), table_id).size(), 1);
 
-  // With the generation active on every replica, the master accepts the split request but the
-  // tablet-side fence rejects the split operation before it is appended to Raft.
-  ASSERT_OK(SetOrderingGenerationOnAllReplicas(tablet_id, MakeActiveOrderingGeneration()));
+  // The generation has been active on every replica since before the writes; the master
+  // accepts the split request but the tablet-side fence rejects the split operation before it
+  // is appended to Raft.
   ASSERT_OK(InvokeSplitTabletRpc(
       cluster_.get(), tablet_id, MonoDelta::FromSeconds(30) * kTimeMultiplier));
   SleepFor(MonoDelta::FromSeconds(3) * kTimeMultiplier);
@@ -393,9 +395,9 @@ TEST_F(FixedHybridTimeWriteIdITest, OrderingGenerationSurvivesRemoteBootstrap) {
       cluster_.get(), tablet_id, CoarseMonoClock::Now() + 10s * kTimeMultiplier,
       RequireLeaderIsReady::kTrue));
 
+  ASSERT_OK(SetOrderingGenerationOnAllReplicas(tablet_id, MakeActiveOrderingGeneration()));
   const auto op_id = ASSERT_RESULT(SendMarkedWrite(tablet_id, kWriteHT, {{"row", "value"}}));
   ASSERT_OK(WaitAllReplicasApplied(tablet_id, op_id));
-  ASSERT_OK(SetOrderingGenerationOnAllReplicas(tablet_id, MakeActiveOrderingGeneration()));
 
   // Tombstone one follower; it stays in the Raft config, so the leader remote-bootstraps it
   // back. The recovered replica's superblock comes from the leader and must carry the
@@ -464,13 +466,13 @@ TEST_F(FixedHybridTimeWriteIdITest, SplitWithFenceBypassedPreservesPerKeyWriteId
   ASSERT_OK(WaitUntilTabletHasLeader(
       cluster_.get(), tablet_id, CoarseMonoClock::Now() + 10s * kTimeMultiplier,
       RequireLeaderIsReady::kTrue));
+  ASSERT_OK(SetOrderingGenerationOnAllReplicas(tablet_id, MakeActiveOrderingGeneration()));
 
   const auto op_id = ASSERT_RESULT(
       SendMarkedWriteSpreadHashes(tablet_id, kWriteHT, /* num_keys= */ 256));
   ASSERT_OK(WaitAllReplicasApplied(tablet_id, op_id));
   ASSERT_OK(cluster_->FlushTablets());
   ASSERT_OK(WaitForAnySstFiles(cluster_.get(), tablet_id));
-  ASSERT_OK(SetOrderingGenerationOnAllReplicas(tablet_id, MakeActiveOrderingGeneration()));
 
   const auto table_id = table_.table()->id();
   ASSERT_OK(InvokeSplitTabletRpc(

src/yb/integration-tests/tablet-split-itest.cc

@@ -970,10 +970,12 @@ TEST_F(TabletSplitITest, MaxCreateTabletsPerTs) {
   auto table = catalog_mgr->GetTableInfo(table_->id());
 
   ANNOTATE_UNPROTECTED_WRITE(FLAGS_max_create_tablets_per_ts) = 1;
-  ASSERT_NOK(master.tablet_split_manager().ValidateSplitCandidateTable(table));
+  ASSERT_NOK(master.tablet_split_manager().ValidateSplitCandidateTable(
+      table, /* indexed_table= */ nullptr));
 
   ANNOTATE_UNPROTECTED_WRITE(FLAGS_max_create_tablets_per_ts) = 2;
-  ASSERT_OK(master.tablet_split_manager().ValidateSplitCandidateTable(table));
+  ASSERT_OK(master.tablet_split_manager().ValidateSplitCandidateTable(
+      table, /* indexed_table= */ nullptr));
 }
 
 TEST_F(TabletSplitITest, SplitDuringReplicaOffline) {
@@ -2762,23 +2764,26 @@ TEST_F(TabletSplitSingleServerITest, AutoSplitNotValidOnceCheckedForTtl) {
   auto table_info = ASSERT_NOTNULL(catalog_mgr->GetTableInfo(table_->id()));
 
   // Candidate table should start as a valid split candidate.
-  ASSERT_OK(split_manager->ValidateSplitCandidateTable(table_info));
+  ASSERT_OK(split_manager->ValidateSplitCandidateTable(
+      table_info, /* indexed_table= */ nullptr));
 
   // State that table should not be split for the next 1 second.
   // Candidate table should no longer be valid.
   split_manager->DisableSplittingForTtlTable(table_->id());
-  ASSERT_NOK(split_manager->ValidateSplitCandidateTable(table_info));
+  ASSERT_NOK(split_manager->ValidateSplitCandidateTable(
+      table_info, /* indexed_table= */ nullptr));
 
   // After 2 seconds, table is a valid split candidate again.
   SleepFor(kSecondsBetweenChecks * 2s);
-  ASSERT_OK(split_manager->ValidateSplitCandidateTable(table_info));
+  ASSERT_OK(split_manager->ValidateSplitCandidateTable(
+      table_info, /* indexed_table= */ nullptr));
 
   // State again that table should not be split for the next 1 second.
   // Candidate table should still be a valid candidate if ignore_disabled_list
   // is true (e.g. in the case of manual tablet splitting).
   split_manager->DisableSplittingForTtlTable(table_->id());
-  ASSERT_OK(split_manager->ValidateSplitCandidateTable(table_info,
-      master::IgnoreDisabledList::kTrue));
+  ASSERT_OK(split_manager->ValidateSplitCandidateTable(
+      table_info, /* indexed_table= */ nullptr, master::IgnoreDisabledList::kTrue));
 }
 
 TEST_F(TabletSplitSingleServerITest, ScheduledFullCompactionsDoNotBlockSplit) {

src/yb/master/backfill_index.cc

@@ -32,6 +32,7 @@
 #include "yb/tserver/tserver_admin.proxy.h"
 
 #include "yb/common/common_util.h"
+#include "yb/common/doc_hybrid_time.h"
 #include "yb/common/wire_protocol.h"
 
 #include "yb/docdb/doc_rowwise_iterator.h"
@@ -1076,6 +1077,18 @@ Status BackfillTable::DoBackfill() {
     VLOG_WITH_PREFIX(1) << "starting backfill with timestamp: " << read_time_for_backfill();
   }
 
+  if (unique_index_backfill_mode() == UniqueIndexBackfillMode::UNIQUE_INDEX_BACKFILL_SKIP_ALL &&
+      !ordering_generation_activated_.exchange(true)) {
+    // SKIP_ALL writes are marked (Raft-index write IDs), which every index tablet only accepts
+    // under an active ordering generation -- activate before the first chunk. The job continues
+    // into LaunchBackfillTablets through OrderingGenerationUpdateDone once every tablet acks.
+    // On failover resume this re-runs: re-activation is idempotent (the base moves up).
+    return LaunchOrderingGenerationActivation();
+  }
+  return LaunchBackfillTablets();
+}
+
+Status BackfillTable::LaunchBackfillTablets() {
   auto tablets = VERIFY_RESULT(indexed_table_->GetTablets());
   num_tablets_.store(tablets.size(), std::memory_order_release);
   tablets_pending_.store(tablets.size(), std::memory_order_release);
@@ -1086,6 +1099,126 @@ Status BackfillTable::DoBackfill() {
   return Status::OK();
 }
 
+Result<std::vector<scoped_refptr<TableInfo>>> BackfillTable::GetUniqueIndexTables(
+    RestrictToIndexesToBuild restrict_to_indexes_to_build) const {
+  const auto to_build = indexes_to_build();
+  std::vector<scoped_refptr<TableInfo>> tables;
+  for (const auto& index_info : index_infos_) {
+    if (!index_info.is_unique()) {
+      continue;
+    }
+    if (restrict_to_indexes_to_build && to_build.count(index_info.table_id()) == 0) {
+      continue;
+    }
+    auto res = master_->catalog_manager()->FindTableById(index_info.table_id());
+    if (!res && res.status().IsNotFound()) {
+      // Concurrent DROP INDEX; the job will fail through the missing-IndexInfoPB path.
+      LOG_WITH_PREFIX(WARNING) << "Index " << index_info.table_id() << " was not found; "
+                               << "skipping for ordering-generation update: " << res.status();
+      continue;
+    }
+    tables.push_back(VERIFY_RESULT_PREPEND(
+        std::move(res), Format("Could not find the index table $0", index_info.table_id())));
+  }
+  return tables;
+}
+
+Status BackfillTable::LaunchOrderingGenerationActivation() {
+  if (indexed_table_->GetTableType() != TableType::PGSQL_TABLE_TYPE) {
+    // The SKIP_ALL write path is PGSQL-only (the backfill request mode rides the YSQL chunk
+    // requests; YCQL backfill writes are never marked), so generations would only fence
+    // splits without protecting anything. Reachable today only via the TEST mode override.
+    return LaunchBackfillTablets();
+  }
+  auto index_tables = VERIFY_RESULT(GetUniqueIndexTables(RestrictToIndexesToBuild::kTrue));
+  if (index_tables.empty()) {
+    return LaunchBackfillTablets();
+  }
+
+  // Fence and drain index-table splitting before activating. The tablet-side fences (the
+  // pending-split rejection of CHANGE_METADATA_OP and, once active, the generation split
+  // fence) close the append-time races; this master-side drain removes the split/activation
+  // TOCTOU window entirely under a single master leader, and the persisted fence in the
+  // tablet-split manager (backfill-job state) keeps splits excluded across failover.
+  auto& tablet_split_manager = master_->tablet_split_manager();
+  const CoarseTimePoint deadline =
+      CoarseMonoClock::Now() +
+      FLAGS_index_backfill_tablet_split_completion_timeout_sec * 1s * kTimeMultiplier;
+  for (const auto& index_table : index_tables) {
+    tablet_split_manager.DisableSplittingForBackfillingTable(index_table->id());
+    while (!tablet_split_manager.IsTabletSplittingComplete(
+        *index_table, false /* wait_for_parent_deletion */, deadline)) {
+      if (CoarseMonoClock::Now() > deadline) {
+        return STATUS(
+            TimedOut,
+            "Index-tablet splitting did not complete after being disabled; cannot safely "
+            "activate the ordering generation.");
+      }
+      SleepFor(FLAGS_index_backfill_tablet_split_completion_poll_freq_ms * 1ms * kTimeMultiplier);
+    }
+  }
+
+  std::vector<std::pair<TabletInfoPtr, TableId>> tablets;
+  for (const auto& index_table : index_tables) {
+    for (auto& tablet : VERIFY_RESULT(index_table->GetTablets())) {
+      tablets.emplace_back(std::move(tablet), index_table->id());
+    }
+  }
+  if (tablets.empty()) {
+    return LaunchBackfillTablets();
+  }
+
+  LOG_WITH_PREFIX(INFO) << "Activating index-backfill ordering generation on " << tablets.size()
+                        << " index tablet(s) before launching backfill chunks";
+  activation_tablets_pending_.store(tablets.size(), std::memory_order_release);
+  for (auto& [tablet, index_table_id] : tablets) {
+    auto task = std::make_shared<UpdateOrderingGenerationForTablet>(
+        shared_from_this(), tablet, index_table_id, tablet::ActivateGeneration::kTrue,
+        NotifyBackfillTable::kTrue, epoch_);
+    RETURN_NOT_OK(task->Launch());
+  }
+  return Status::OK();
+}
+
+void BackfillTable::OrderingGenerationUpdateDone(
+    const Status& status, const TabletId& tablet_id) {
+  if (done()) {
+    return;
+  }
+  if (!status.ok()) {
+    LOG_WITH_PREFIX(WARNING) << "Ordering-generation activation failed for tablet " << tablet_id
+                             << ": " << status << "; aborting backfill";
+    WARN_NOT_OK(Abort(), "Failed to abort backfill after activation failure");
+    return;
+  }
+  if (--activation_tablets_pending_ == 0) {
+    LOG_WITH_PREFIX(INFO) << "Ordering generation active on all index tablets; "
+                          << "launching backfill chunks";
+    Status s = LaunchBackfillTablets();
+    if (!s.ok()) {
+      LOG_WITH_PREFIX(WARNING) << "Failed to launch backfill after activation: " << s;
+      WARN_NOT_OK(Abort(), "Failed to abort backfill");
+    }
+  }
+}
+
+Status BackfillTable::SendRpcToReleaseOrderingGenerations() {
+  // Release on every unique index of the job, built or not: activation may have partially
+  // succeeded before a failure. Fire-and-forget -- the tserver metadata validator and master
+  // reload reconciliation converge any straggler.
+  auto index_tables = VERIFY_RESULT(GetUniqueIndexTables(RestrictToIndexesToBuild::kFalse));
+  for (const auto& index_table : index_tables) {
+    for (const auto& tablet : VERIFY_RESULT(index_table->GetTablets())) {
+      auto task = std::make_shared<UpdateOrderingGenerationForTablet>(
+          shared_from_this(), tablet, index_table->id(), tablet::ActivateGeneration::kFalse,
+          NotifyBackfillTable::kFalse, epoch_);
+      WARN_NOT_OK(task->Launch(), "Failed to send ordering-generation release");
+    }
+    master_->tablet_split_manager().ReenableSplittingForBackfillingTable(index_table->id());
+  }
+  return Status::OK();
+}
+
 Status BackfillTable::Done(const Status& s, const std::unordered_set<TableId>& failed_indexes) {
   if (!s.ok()) {
     LOG_WITH_PREFIX(WARNING) << "failed to backfill the index: " << AsString(failed_indexes)
@@ -1316,6 +1449,13 @@ Status BackfillTable::UpdateIndexPermissionsForIndexes() {
   RETURN_NOT_OK(ClearCheckpointStateInTablets());
   indexed_table_->ClearIsBackfilling();
   master_->tablet_split_manager().ReenableSplittingForBackfillingTable(indexed_table_->id());
+  if (unique_index_backfill_mode() == UniqueIndexBackfillMode::UNIQUE_INDEX_BACKFILL_SKIP_ALL) {
+    // Terminal funnel: every success/failure/abort path of a SKIP_ALL job passes through here,
+    // so the ordering generations are released (and index-table splitting re-enabled) on all of
+    // them. At-least-once; backstops converge anything this misses (e.g. master death here).
+    WARN_NOT_OK(
+        SendRpcToReleaseOrderingGenerations(), "Failed to release ordering generations");
+  }
 
   VLOG(1) << "Sending alter table requests to the Indexed table";
   RETURN_NOT_OK(master_->catalog_manager_impl()->SendAlterTableRequest(indexed_table_, epoch_));
@@ -1666,6 +1806,115 @@ void GetSafeTimeForTablet::UnregisterAsyncTaskCallback() {
     "Could not UpdateSafeTime");
 }
 
+UpdateOrderingGenerationForTablet::UpdateOrderingGenerationForTablet(
+    std::shared_ptr<BackfillTable> backfill_table,
+    const TabletInfoPtr& tablet,
+    const TableId& index_table_id,
+    tablet::ActivateGeneration activate,
+    NotifyBackfillTable notify_backfill_table,
+    LeaderEpoch epoch)
+    : RetryingTSRpcTaskWithTable(
+          backfill_table->master(), backfill_table->threadpool(),
+          std::unique_ptr<TSPicker>(new PickLeaderReplica(tablet)), tablet->table(),
+          std::move(epoch),
+          /* async_task_throttler */ nullptr),
+      backfill_table_(backfill_table),
+      tablet_(tablet),
+      index_table_id_(index_table_id),
+      activate_(activate),
+      notify_backfill_table_(notify_backfill_table) {
+  deadline_ = MonoTime::Max();  // Single-attempt deadline comes from ComputeDeadline().
+}
+
+Status UpdateOrderingGenerationForTablet::Launch() {
+  tablet_->table()->AddTask(shared_from_this());
+  RETURN_NOT_OK_PREPEND(
+      Run(),
+      Substitute("Failed to send UpdateOrderingGeneration request for $0. ",
+                 tablet_->ToString()));
+  VLOG(3) << "Started UpdateOrderingGenerationForTablet : " << this->description();
+  return Status::OK();
+}
+
+std::string UpdateOrderingGenerationForTablet::description() const {
+  return Format(
+      "$0 ordering generation for index tablet $1 of $2",
+      activate_ ? "Activate" : "Release", tablet_id(), index_table_id_);
+}
+
+TabletId UpdateOrderingGenerationForTablet::tablet_id() const { return tablet_->id(); }
+
+bool UpdateOrderingGenerationForTablet::SendRequest(int attempt) {
+  ADOPT_WAIT_STATE(backfill_table_->wait_state());
+  tablet::ChangeMetadataRequestPB req;
+  req.set_dest_uuid(permanent_uuid());
+  req.set_tablet_id(tablet_->tablet_id());
+  req.set_propagated_hybrid_time(master_->clock()->Now().ToUint64());
+  auto* generation_op = req.mutable_index_backfill_ordering_generation();
+  generation_op->set_table_id(index_table_id_);
+  generation_op->set_activate(activate_);
+  if (activate_) {
+    // History at and above backfill_read_time.Decremented() must survive until verification;
+    // the record carries the barrier, enforcement lands with the verification read fence.
+    generation_op->set_retention_barrier_ht(
+        backfill_table_->read_time_for_backfill().Decremented().ToUint64());
+    generation_op->set_write_id_floor_version(kIndexBackfillWriteIdFloorVersion);
+  }
+
+  ts_admin_proxy_->UpdateIndexBackfillOrderingGenerationAsync(
+      req, &resp_, &rpc_, BindRpcCallback());
+  VLOG(1) << "Send " << description() << " to " << permanent_uuid()
+          << " (attempt " << attempt << "):\n" << req.DebugString();
+  return true;
+}
+
+void UpdateOrderingGenerationForTablet::HandleResponse(int attempt) {
+  ADOPT_WAIT_STATE(backfill_table_->wait_state());
+  Status status = Status::OK();
+  if (resp_.has_error()) {
+    status = StatusFromPB(resp_.error().status());
+    switch (resp_.error().code()) {
+      case TabletServerErrorPB::TABLET_NOT_FOUND:
+      case TabletServerErrorPB::OPERATION_NOT_SUPPORTED:
+        LOG(WARNING) << "TS " << permanent_uuid() << ": " << description()
+                     << " failed, no further retry: " << status;
+        TransitionToFailedState(MonitoredTaskState::kRunning, status);
+        break;
+      default:
+        LOG(WARNING) << "TS " << permanent_uuid() << ": " << description() << " failed: "
+                     << status << " code " << resp_.error().code();
+        break;
+    }
+  } else {
+    TransitionToCompleteState();
+    VLOG(1) << "TS " << permanent_uuid() << ": " << description() << " complete";
+  }
+
+  server::UpdateClock(resp_, master_->clock());
+}
+
+void UpdateOrderingGenerationForTablet::UnregisterAsyncTaskCallback() {
+  ADOPT_WAIT_STATE(backfill_table_->wait_state());
+  if (state() == MonitoredTaskState::kAborted) {
+    // Deliberately no join notification (same shape as GetSafeTimeForTablet): external task
+    // aborts accompany leadership loss or table teardown, where the job object is dying with
+    // us; notifying could double-drive a job that is already unwinding.
+    VLOG(1) << " was aborted";
+    return;
+  }
+  if (!notify_backfill_table_) {
+    return;
+  }
+
+  Status status;
+  if (resp_.has_error()) {
+    status = StatusFromPB(resp_.error().status());
+  } else if (state() != MonitoredTaskState::kComplete) {
+    status = STATUS_FORMAT(InternalError, "$0 in state $1", description(), state());
+  }
+  backfill_table_->OrderingGenerationUpdateDone(status, tablet_->tablet_id());
+}
+
 BackfillChunk::BackfillChunk(std::shared_ptr<BackfillTablet> backfill_tablet,
                              const std::string& start_key,
                              LeaderEpoch epoch)