DAOS-19616 object: retry rebuilding I/O at stable epoch boundary

src/dtx/dtx_common.c

@@ -959,7 +959,7 @@ dtx_handle_init(struct dtx_id *dti, daos_handle_t xoh, struct dtx_epoch *epoch,
 	dth->dth_dkey_hash       = 0;
 
 	if (!(flags & DTX_LOCAL)) {
-		if (daos_is_zero_dti(dti))
+		if (daos_is_zero_dti(dti) && !(flags & DTX_FOR_MIGRATION))
 			return 0;
 
 		if (!dtx_epoch_chosen(epoch)) {
@@ -970,6 +970,8 @@ dtx_handle_init(struct dtx_id *dti, daos_handle_t xoh, struct dtx_epoch *epoch,
 		}
 		dth->dth_epoch       = epoch->oe_value;
 		dth->dth_epoch_bound = dtx_epoch_bound(epoch);
+		if (daos_is_zero_dti(dti))
+			return 0;
 	}
 
 	rc = vos_dtx_rsrvd_init(dth);

src/object/cli_coll.c

@@ -706,6 +706,15 @@ dc_obj_coll_punch(tse_task_t *task, struct dc_object *obj, struct dtx_epoch *epo
 	if (rc != 0)
 		goto out;
 
+	D_RWLOCK_RDLOCK(&obj->cob_lock);
+	for (i = 0; i < obj->cob_shards_nr; i++) {
+		if (obj->cob_shards->do_shards[i].do_rebuilding) {
+			auxi->rebuilding = 1;
+			break;
+		}
+	}
+	D_RWLOCK_UNLOCK(&obj->cob_lock);
+
 	for (i = 0; i < obj->cob_shards_nr; i++) {
 		rc = obj_coll_prep_one(coa, obj, map_ver, i);
 		if (rc != 0)
@@ -801,6 +810,8 @@ dc_obj_coll_punch(tse_task_t *task, struct dc_object *obj, struct dtx_epoch *epo
 		goto out;
 
 	auxi->flags = ORF_LEADER;
+	if (auxi->rebuilding)
+		auxi->flags |= ORF_REBUILDING_IO;
 	if (auxi->io_retry) {
 		auxi->flags |= ORF_RESEND;
 		/* Reset @enqueue_id if resend to new leader. */

src/object/cli_shard.c

@@ -1422,6 +1422,8 @@ dc_obj_shard_punch(struct dc_obj_shard *shard, enum obj_rpc_opc opc,
 	uuid_copy(opi->opi_co_uuid, shard->do_co->dc_uuid);
 	daos_dti_copy(&opi->opi_dti, &args->pa_dti);
 	opi->opi_flags = args->pa_auxi.flags;
+	if (args->pa_auxi.obj_auxi->rebuilding)
+		opi->opi_flags |= ORF_REBUILDING_IO;
 	opi->opi_dti_cos.ca_count = 0;
 	opi->opi_dti_cos.ca_arrays = NULL;
 	if (opc_get_rpc_ver(req->cr_opc) >= 10) {

src/object/obj_tx.c

@@ -84,7 +84,7 @@ struct dc_tx {
 	uint32_t                 tx_fixed_epoch : 1,                      /** epoch is specified. */
 	    tx_retry : 1, /** Retry the commit RPC. */ tx_set_resend : 1, /** Set 'resend' flag. */
 	    tx_for_convert : 1, tx_has_cond : 1, tx_renew : 1, tx_closed : 1, tx_reintegrating : 1,
-	    tx_maybe_starve : 1;
+	    tx_rebuilding : 1, tx_maybe_starve : 1;
 	/** Transaction status (OPEN, COMMITTED, etc.), see dc_tx_status. */
 	enum dc_tx_status	 tx_status;
 	/** The rank for the server on which the TX leader resides. */
@@ -1348,6 +1348,8 @@ dc_tx_classify_common(struct dc_tx *tx, struct daos_cpd_sub_req *dcsr,
 
 		if (shard->do_reintegrating)
 			tx->tx_reintegrating = 1;
+		if (!read && shard->do_rebuilding)
+			tx->tx_rebuilding = 1;
 		/*
 		 * NOTE: It is possible that more than one shards locate on the same DAOS target
 		 *	 under OSA mode, then the shard_idx may be not equal to "shard->do_shard".
@@ -2320,6 +2322,8 @@ dc_tx_commit_trigger(tse_task_t *task, struct dc_tx *tx, daos_tx_commit_t *args)
 	tx->tx_renew = 0;
 	if (tx->tx_reintegrating)
 		oci->oci_flags |= ORF_REINTEGRATING_IO;
+	if (tx->tx_rebuilding)
+		oci->oci_flags |= ORF_REBUILDING_IO;
 	if (tx->tx_write_cnt == 0)
 		oci->oci_flags |= ORF_CPD_RDONLY;
 
@@ -2601,6 +2605,7 @@ dc_tx_restart_end(struct dc_tx *tx)
 	tx->tx_status = TX_OPEN;
 	tx->tx_pm_ver = 0;
 	tx->tx_epoch.oe_value = 0;
+	tx->tx_rebuilding     = 0;
 }
 
 /**

src/object/srv_obj.c

@@ -2468,6 +2468,40 @@ obj_ioc_init_oca(struct obj_io_context *ioc, daos_obj_id_t oid, bool for_modify)
 	return 0;
 }
 
+/*
+ * The IO targets some rebuilding shard(s). The rebuild scan uses the rebuild stable epoch as
+ * its snapshot boundary, so such an IO has to be stamped with an epoch that is strictly newer
+ * than the stable epoch, otherwise it may be missed by the rebuild and lost on the rebuilding
+ * target. Ask the client to retry (with a newer epoch) until that is the case.
+ *
+ * NOTE: It must be called after the epoch has been fixed, and after the resend (if any) has been
+ *	 resolved, otherwise the checked epoch is not the one that will be used for the write.
+ */
+static int
+obj_rebuilding_io_check(struct ds_cont_child *child, daos_epoch_t epoch, uint32_t flags)
+{
+	struct ds_pool *pool = child->sc_pool->spc_pool;
+	daos_epoch_t    stable_epoch;
+	uint32_t        version;
+
+	if (!(flags & ORF_REBUILDING_IO) || !atomic_load(&pool->sp_rebuilding))
+		return 0;
+
+	ds_rebuild_running_query(child->sc_pool_uuid, RB_OP_REBUILD, &version, &stable_epoch, NULL);
+	if (version == 0)
+		return 0;
+
+	if (stable_epoch == 0 || epoch <= stable_epoch) {
+		D_DEBUG(DB_IO,
+			DF_UUID " retry rebuilding IO epoch " DF_X64
+				", rebuilding %u stable " DF_X64 "\n",
+			DP_UUID(child->sc_pool_uuid), epoch, version, stable_epoch);
+		return -DER_UPDATE_AGAIN;
+	}
+
+	return 0;
+}
+
 static int
 obj_inflight_io_check(struct ds_cont_child *child, uint32_t opc,
 		      uint32_t rpc_map_ver, uint32_t flags)
@@ -3059,6 +3093,7 @@ ds_obj_rw_handler(crt_rpc_t *rpc)
 	int                              rc;
 	int                              retry      = 0;
 	bool                             need_abort = false;
+	bool                             prepared   = false;
 
 	D_ASSERT(orw != NULL);
 	D_ASSERT(orwo != NULL);
@@ -3193,11 +3228,23 @@ ds_obj_rw_handler(crt_rpc_t *rpc)
 			goto out;
 		if (rc == ORS_DONE)
 			D_GOTO(out, rc = 0);
+		prepared = (rc == ORS_PREPARED);
 	} else if (DAOS_FAIL_CHECK(DAOS_DTX_LOST_RPC_REQUEST)) {
 		ioc.ioc_lost_reply = 1;
 		D_GOTO(out, rc);
 	}
 
+	/*
+	 * Only gate the modifications that are really going to be executed: an already prepared
+	 * DTX carries a fixed epoch that cannot be advanced by a retry, rejecting it would loop
+	 * forever.
+	 */
+	if (!prepared) {
+		rc = obj_rebuilding_io_check(ioc.ioc_coc, orw->orw_epoch, orw->orw_flags);
+		if (rc != 0)
+			goto out;
+	}
+
 	/* For leader case, we need to find out the potential conflict
 	 * (or share the same non-committed object/dkey) DTX(s) in the
 	 * CoS (committable) cache, piggyback them via the dispdatched
@@ -3506,6 +3553,10 @@ obj_local_enum(struct obj_io_context *ioc, crt_rpc_t *rpc,
 		 */
 		atomic_store(&ioc->ioc_coc->sc_pool->spc_pool->sp_rebuild_enum, 1);
 		flags = DTX_FOR_MIGRATION;
+		if (!(oei->oei_flags & ORF_ENUM_WITHOUT_EPR)) {
+			epoch.oe_value = oei->oei_epr.epr_hi;
+			epoch.oe_first = oei->oei_epr.epr_hi;
+		}
 	}
 
 	rc = dtx_begin(ioc->ioc_vos_coh, &oei->oei_dti, &epoch, 0,
@@ -4059,6 +4110,7 @@ ds_obj_punch_handler(crt_rpc_t *rpc)
 	int                              rc;
 	int                              retry      = 0;
 	bool                             need_abort = false;
+	bool                             prepared   = false;
 
 	opi = crt_req_get(rpc);
 	D_ASSERT(opi != NULL);
@@ -4121,11 +4173,19 @@ ds_obj_punch_handler(crt_rpc_t *rpc)
 			goto out;
 		if (rc == ORS_DONE)
 			D_GOTO(out, rc = 0);
+		prepared = (rc == ORS_PREPARED);
 	} else if (DAOS_FAIL_CHECK(DAOS_DTX_LOST_RPC_REQUEST) ||
 		   DAOS_FAIL_CHECK(DAOS_DTX_LONG_TIME_RESEND)) {
 		goto cleanup;
 	}
 
+	/* See the comment in ds_obj_rw_handler(). */
+	if (!prepared) {
+		rc = obj_rebuilding_io_check(ioc.ioc_coc, opi->opi_epoch, opi->opi_flags);
+		if (rc != 0)
+			goto out;
+	}
+
 	/* For leader case, we need to find out the potential conflict
 	 * (or share the same non-committed object/dkey) DTX(s) in the
 	 * CoS (committable) cache, piggyback them via the dispdatched
@@ -5303,6 +5363,7 @@ ds_obj_dtx_leader(struct daos_cpd_args *dca)
 	int				 req_cnt = 0;
 	int				 rc = 0;
 	bool				 need_abort = false;
+	bool                             prepared   = false;
 
 	dcsh = ds_obj_cpd_get_head(dca->dca_rpc, dca->dca_idx);
 
@@ -5341,10 +5402,19 @@ ds_obj_dtx_leader(struct daos_cpd_args *dca)
 			goto out;
 		if (rc == ORS_DONE)
 			D_GOTO(out, rc = 0);
+		prepared = (rc == ORS_PREPARED);
 	} else if (DAOS_FAIL_CHECK(DAOS_DTX_LOST_RPC_REQUEST)) {
 		D_GOTO(out, rc = 0);
 	}
 
+	/* See the comment in ds_obj_rw_handler(). */
+	if (!prepared) {
+		rc = obj_rebuilding_io_check(dca->dca_ioc->ioc_coc, dcsh->dcsh_epoch.oe_value,
+					     oci->oci_flags);
+		if (rc != 0)
+			goto out;
+	}
+
 	dcde = ds_obj_cpd_get_ents(dca->dca_rpc, dca->dca_idx, 0);
 	dcsrs = ds_obj_cpd_get_reqs(dca->dca_rpc, dca->dca_idx);
 	tgts = ds_obj_cpd_get_tgts(dca->dca_rpc, dca->dca_idx);
@@ -5392,7 +5462,8 @@ ds_obj_dtx_leader(struct daos_cpd_args *dca)
 	rc = dtx_leader_end(dlh, dca->dca_ioc->ioc_coc, rc);
 
 out:
-	DL_CDEBUG(rc != 0 && rc != -DER_INPROGRESS && rc != -DER_TX_RESTART && rc != -DER_AGAIN,
+	DL_CDEBUG(rc != 0 && rc != -DER_INPROGRESS && rc != -DER_TX_RESTART && rc != -DER_AGAIN &&
+		      rc != -DER_UPDATE_AGAIN,
 		  DLOG_ERR, DB_IO, rc, "Handled DTX " DF_DTI " on leader, idx %u",
 		  DP_DTI(&dcsh->dcsh_xid), dca->dca_idx);
 
@@ -5872,6 +5943,7 @@ ds_obj_coll_punch_handler(crt_rpc_t *rpc)
 	int				 i;
 	bool				 need_abort = false;
 	bool                             leader;
+	bool                             prepared = false;
 
 	if (ocpi->ocpi_flags & ORF_LEADER)
 		leader = true;
@@ -5934,6 +6006,14 @@ ds_obj_coll_punch_handler(crt_rpc_t *rpc)
 			goto out;
 		if (rc == ORS_DONE)
 			D_GOTO(out, rc = 0);
+		prepared = (rc == ORS_PREPARED);
+	}
+
+	/* See the comment in ds_obj_rw_handler(). */
+	if (leader && !prepared) {
+		rc = obj_rebuilding_io_check(ioc.ioc_coc, ocpi->ocpi_epoch, ocpi->ocpi_flags);
+		if (rc != 0)
+			goto out;
 	}
 
 	epoch.oe_value = ocpi->ocpi_epoch;

src/object/srv_obj_migrate.c

@@ -919,6 +919,11 @@ mrone_obj_fetch(struct migrate_one *mrone, daos_handle_t oh, d_sg_list_t *sgls,
 	}
 
 out:
+	if (rc == -DER_VOS_PARTIAL_UPDATE) {
+		D_WARN(DF_RB ": " DF_UOID " restart rebuild after migration DTX conflict\n",
+		       DP_RB_MRO(mrone), DP_UOID(mrone->mo_oid));
+		rc = -DER_STALE;
+	}
 	return rc;
 }
 
@@ -3286,6 +3291,11 @@ migrate_obj_epoch(struct migrate_pool_tls *tls, struct iter_obj_arg *arg, daos_e
 				break;
 			}
 			continue;
+		} else if (rc == -DER_VOS_PARTIAL_UPDATE) {
+			D_WARN(DF_RB ": " DF_UOID
+				     " restart rebuild after migration enumeration DTX conflict\n",
+			       DP_RB_MPT(tls), DP_UOID(arg->oid));
+			break;
 		} else if (rc && rc != -DER_SHUTDOWN &&
 			   daos_anchor_get_flags(&dkey_anchor) & DIOF_TO_LEADER) {
 			if (rc != -DER_INPROGRESS) {

src/rebuild/scan.c

@@ -1011,8 +1011,8 @@ rebuild_container_scan_cb(daos_handle_t ih, vos_iter_entry_t *entry,
 	}
 
 	epoch.oe_value = rpt->rt_stable_epoch;
-	rc = dtx_begin(coh, &dti, &epoch, 0, rpt->rt_rebuild_ver,
-		       &oid, NULL, 0, DTX_IGNORE_UNCOMMITTED, NULL, &dth);
+	rc             = dtx_begin(coh, &dti, &epoch, 0, rpt->rt_rebuild_ver, &oid, NULL, 0,
+				   DTX_FOR_MIGRATION | DTX_IGNORE_UNCOMMITTED, NULL, &dth);
 	D_ASSERT(rc == 0);
 	memset(&param, 0, sizeof(param));
 	param.ip_hdl = coh;

src/vos/vos_dtx.c

@@ -1449,16 +1449,22 @@ vos_dtx_check_availability(daos_handle_t coh, uint32_t entry,
 		return ALB_UNAVAILABLE;
 	}
 
-	/*
-	 * Up layer rebuild logic guarantees that the rebuild scan will not be
-	 * triggered until DTX resync has been done on all related targets. So
-	 * here, if rebuild logic hits non-committed DTX entry, it must be for
-	 * new IO that version is not older than rebuild, then it is invisible
-	 * to rebuild. Related new IO corresponding to such non-committed DTX
-	 * has already been sent to the in-rebuilding target.
-	 */
-	if (intent == DAOS_INTENT_MIGRATION)
+	if (intent == DAOS_INTENT_MIGRATION) {
+		/*
+		 * A non-ready DTX inside the migration snapshot may still be modifying the
+		 * rebuilding shard. Restart rebuild with a new stable epoch rather than
+		 * migrating a view that can change after this check.
+		 */
+		if (dth != NULL && DAE_EPOCH(dae) <= dth->dth_epoch) {
+			D_WARN("Non-ready DTX " DF_DTI " at " DF_X64 " (version %u)"
+			       " conflicts with migration boundary " DF_X64 " (version %u)\n",
+			       DP_DTI(&DAE_XID(dae)), DAE_EPOCH(dae), DAE_VER(dae), dth->dth_epoch,
+			       dth->dth_ver);
+			return -DER_VOS_PARTIAL_UPDATE;
+		}
+
 		return ALB_UNAVAILABLE;
+	}
 
 	if (intent == DAOS_INTENT_DEFAULT) {
 		if (DAOS_FAIL_CHECK(DAOS_VOS_NON_LEADER))