Batch object reallocation syncs in zfs receive

When an incremental stream reallocates an object at a different dnode
slot count, receive_object() must wait for the old dnode's free to
sync out before the object number can be claimed again, because
dnode_check_slots_free() only accepts slots whose final dirty txg has
synced.  The receive writer did this with one txg_wait_synced() per
reallocated object, so a dataset that ever changed dnodesize (for
example dnodesize=auto, as the root-on-ZFS guides suggest) receives
such incrementals at a few KB/s while the pool syncs mostly empty
txgs (#11353).

Batch those waits.  The frees are still issued immediately, since
many frees can share one txg, but the record whose claim would need
the sync is parked on a per-receive list and the affected dnode slot
ranges are tracked in a small AVL tree.  Later records that touch a
parked range park behind it in stream order while unrelated records
keep flowing.  Once the batch reaches zfs_recv_defer_batch_size
(default 32 MB, 0 restores the old behavior) or the stream ends, a
single txg_wait_synced() covers every free issued so far and the
parked records are replayed through the normal processing path, where
the claims now find their object numbers free on disk.  The wait
targets the highest txg a free was assigned to instead of the whole
open txg window, as #17434 did for dmu_offset_next().

Object records themselves are only parked from receive_object(), after
its multi-slot expansion loop has run: that loop is what discovers a
dnode beginning inside the claimed range but ending past it, frees it,
and records the extent it actually occupies, so the DRR_FREEOBJECTS
the sender emits for those trailing slots parks behind the claim
instead of being applied against an interior slot and failing the
receive.

While records are parked the saved resume state is pinned at the
first parked record, so an interrupted resumable receive resumes from
a point no later than one batch back.  Raw streams keep the
historical per-object sync: their DRR_OBJECT_RANGE encryption
parameters are consumed by the first claim after each range record,
and deferring claims would require snapshotting that state per
record.

Receiving an incremental that reallocates 4000 objects across a
dnodesize change went from 4.98 s and 6006 txgs to 0.10 s and 5 txgs
for small files, and from 5.31 s to 0.49 s for 128 KB files, with the
received data verified identical to the source.

Tests: new ZTS cases send_realloc_dnode_resume (double-interrupted
resume across dnode churn, in the common runfile) and
send_realloc_dnode_mixed (mixed dnode sizes reaching past a deferred
claim, failing before the expansion-loop ordering above); ZTS rsend,
zfs_receive, zfs_send and redacted_send groups pass; manual
double-interrupt resume of a dnodesize-churned incremental diffs
clean.

Reviewed-by: Brian Behlendorf <[email protected]>
Reviewed-by: Alek Pinchuk <[email protected]>
Signed-off-by: MorganaFuture <[email protected]>
Closes #18831

man/man4/zfs.4

@@ -2423,6 +2423,25 @@ This setting will not reduce the write size below a single block.
 Capped at a maximum of
 .Sy 32 MiB .
 .
+.It Sy zfs_recv_defer_batch_size Ns = Ns Sy 33554432 Ns B Po 32 MiB Pc Pq uint
+When a non-raw incremental stream reallocates an object at a different
+dnode size,
+.Nm zfs Cm receive
+must wait for the old dnode's free to be written out before the object
+number can be reused.
+Rather than forcing one transaction group sync per reallocated object,
+affected records are parked, up to this many bytes at a time, and applied
+together behind a single sync.
+Parked records hold their payloads in memory, so a receive may use up to
+this much memory in addition to
+.Sy zfs_recv_queue_length .
+While records are parked, the saved resume state is held back at the first
+parked record, so interrupting a receive can require resending up to this
+much of the stream.
+Setting this to
+.Sy 0
+restores the previous behavior of one sync per reallocated object.
+.
 .It Sy zfs_recv_best_effort_corrective Ns = Ns Sy 0 Pq int
 When this variable is set to non-zero a corrective receive:
 .Bl -enum -compact -offset 4n -width "1."

module/zfs/dmu_recv.c

@@ -75,6 +75,7 @@
 static uint_t zfs_recv_queue_length = SPA_MAXBLOCKSIZE;
 static uint_t zfs_recv_queue_ff = 20;
 static uint_t zfs_recv_write_batch_size = 1024 * 1024;
+static uint_t zfs_recv_defer_batch_size = 32 * 1024 * 1024;
 static int zfs_recv_best_effort_corrective = 0;
 
 static const void *const dmu_recv_tag = "dmu_recv_tag";
@@ -103,6 +104,16 @@ struct receive_record_arg {
 	bqueue_node_t node;
 };
 
+/*
+ * A range of dnode slots that must not be touched until the deferred
+ * frees covering it have synced out; see receive_defer_park().
+ */
+typedef struct receive_defer_range {
+	avl_node_t rdr_node;
+	uint64_t rdr_first;	/* first slot in the range */
+	uint64_t rdr_last;	/* last slot in the range (inclusive) */
+} receive_defer_range_t;
+
 struct receive_writer_arg {
 	objset_t *os;
 	boolean_t byteswap;
@@ -143,8 +154,27 @@ struct receive_writer_arg {
 
 	/* Keep track of DRR_FREEOBJECTS right after DRR_OBJECT_RANGE */
 	or_need_sync_t or_need_sync;
+
+	/*
+	 * Records whose application is deferred until the frees they
+	 * depend on have synced out, the dnode slot ranges they cover,
+	 * and the resume-state pin for the first deferred record.  See
+	 * receive_defer_park().
+	 */
+	list_t defer_records;
+	avl_tree_t defer_ranges;
+	uint64_t defer_bytes;
+	uint64_t defer_nrecords;
+	uint64_t defer_first_object;
+	uint64_t defer_first_bytes_read;
+	uint64_t defer_max_free_txg;
+	boolean_t defer_replaying;
 };
 
+static int receive_process_record(struct receive_writer_arg *rwa,
+    struct receive_record_arg *rrd);
+static int flush_write_batch(struct receive_writer_arg *rwa);
+
 typedef struct dmu_recv_begin_arg {
 	const char *drba_origin;
 	dmu_recv_cookie_t *drba_cookie;
@@ -1678,15 +1708,39 @@ save_resume_state(struct receive_writer_arg *rwa,
     uint64_t object, uint64_t offset, dmu_tx_t *tx)
 {
 	int txgoff = dmu_tx_get_txg(tx) & TXG_MASK;
+	uint64_t bytes = rwa->bytes_read;
 
 	if (!rwa->resumable)
 		return;
 
+	/*
+	 * While records are parked on the defer list (including while
+	 * they are being replayed), the resume point must not advance
+	 * past the first unapplied record.  Everything parked is at or
+	 * after that stream position, and re-receiving records that were
+	 * already applied is supported (see receive_object()), so pin
+	 * the saved state there until the defer list drains.
+	 *
+	 * A sender emits objects in ascending order, so the pin can only
+	 * sit at or above anything saved before it.  A malformed stream
+	 * that reorders objects could place it lower; keep the already
+	 * saved state in that case instead of moving resume backwards
+	 * past records this receive never parked.
+	 */
+	if (rwa->defer_replaying || !list_is_empty(&rwa->defer_records)) {
+		if (rwa->defer_first_object <
+		    rwa->os->os_dsl_dataset->ds_resume_object[txgoff])
+			return;
+		object = rwa->defer_first_object;
+		offset = 0;
+		bytes = rwa->defer_first_bytes_read;
+	}
+
 	/*
 	 * We use ds_resume_bytes[] != 0 to indicate that we need to
 	 * update this on disk, so it must not be 0.
 	 */
-	ASSERT(rwa->bytes_read != 0);
+	ASSERT(bytes != 0);
 
 	/*
 	 * We only resume from write records, which have a valid
@@ -1707,7 +1761,7 @@ save_resume_state(struct receive_writer_arg *rwa,
 
 	rwa->os->os_dsl_dataset->ds_resume_object[txgoff] = object;
 	rwa->os->os_dsl_dataset->ds_resume_offset[txgoff] = offset;
-	rwa->os->os_dsl_dataset->ds_resume_bytes[txgoff] = rwa->bytes_read;
+	rwa->os->os_dsl_dataset->ds_resume_bytes[txgoff] = bytes;
 }
 
 static int
@@ -2607,10 +2661,400 @@ recv_check_drr_write_embedded(const struct drr_write_embedded *drrwe,
 }
 
 static int
-receive_handle_existing_object(const struct receive_writer_arg *rwa,
+receive_defer_range_compare(const void *a, const void *b)
+{
+	const receive_defer_range_t *ra = a;
+	const receive_defer_range_t *rb = b;
+
+	return (TREE_CMP(ra->rdr_first, rb->rdr_first));
+}
+
+/*
+ * Reclaiming an object number at a different dnode slot count requires the
+ * old dnode's free to be synced out before the number can be claimed again
+ * (dnode_check_slots_free() only accepts slots whose final dirty txg has
+ * synced).  receive_object() historically enforced that with one
+ * txg_wait_synced() per reallocated object, which collapses receive
+ * throughput to a few forced txgs per object when an incremental stream
+ * reallocates many dnodes at a changed slot count (issue #11353).
+ *
+ * Instead of syncing per object, the receive writer defers such claims:
+ * the frees are issued immediately, the affected slot ranges are recorded
+ * in rwa->defer_ranges, and the whole record is parked on
+ * rwa->defer_records.  Any later record that touches a parked range is
+ * parked too, in stream order, while unrelated records keep flowing.  When
+ * the batch grows past zfs_recv_defer_batch_size (or the stream ends), a
+ * single txg_wait_synced() covers every deferred free and the parked
+ * records are replayed in order; the replayed DRR_OBJECT records then find
+ * their object numbers free on disk and take the ordinary allocation path.
+ *
+ * Raw streams are excluded: their DRR_OBJECT_RANGE encryption parameters
+ * are consumed by the first claim after each range record, so deferring
+ * claims would require snapshotting that state per record.  Raw receives
+ * take the historical sync-per-object path unchanged.
+ */
+static boolean_t
+receive_defer_enabled(const struct receive_writer_arg *rwa)
+{
+	return (!rwa->raw && !rwa->heal && !rwa->defer_replaying &&
+	    zfs_recv_defer_batch_size != 0);
+}
+
+static boolean_t
+receive_defer_active(struct receive_writer_arg *rwa)
+{
+	return (!list_is_empty(&rwa->defer_records));
+}
+
+/*
+ * dmu_free_long_object(), except the txg the dnode's free lands in is
+ * folded into rwa->defer_max_free_txg.  The frees a deferred claim
+ * depends on can never be in a later txg than that high-water mark
+ * (dmu_free_long_range() commits its transactions before the dnode
+ * free is assigned), so the defer flush can wait for exactly that txg
+ * instead of the full open-txg window.
+ */
+static int
+receive_defer_free_object(struct receive_writer_arg *rwa, uint64_t object)
+{
+	objset_t *os = rwa->os;
+	dmu_tx_t *tx;
+	int err;
+
+	err = dmu_free_long_range(os, object, 0, DMU_OBJECT_END);
+	if (err != 0)
+		return (err);
+
+	tx = dmu_tx_create(os);
+	dmu_tx_hold_bonus(tx, object);
+	dmu_tx_hold_free(tx, object, 0, DMU_OBJECT_END);
+	dmu_tx_mark_netfree(tx);
+	err = dmu_tx_assign(tx, DMU_TX_WAIT);
+	if (err == 0) {
+		err = dmu_object_free(os, object, tx);
+		rwa->defer_max_free_txg = MAX(rwa->defer_max_free_txg,
+		    dmu_tx_get_txg(tx));
+		dmu_tx_commit(tx);
+	} else {
+		dmu_tx_abort(tx);
+	}
+
+	return (err);
+}
+
+/*
+ * Ranges in the tree are kept disjoint: an insert that overlaps or abuts
+ * existing ranges merges them into one node.  receive_defer_overlaps()
+ * relies on this to give exact answers with one neighbor probe each way.
+ * A bare AVL tree fits better here than zfs_range_tree_t, which asserts
+ * on overlapping adds (the expansion loop inserts ranges that overlap
+ * the claim's) and carries segment accounting this tree, empty outside
+ * a realloc burst, has no use for.
+ */
+static void
+receive_defer_range_insert(struct receive_writer_arg *rwa, uint64_t first,
+    uint64_t last)
+{
+	receive_defer_range_t srch = { .rdr_first = first };
+	avl_index_t where;
+	receive_defer_range_t *rdr, *next;
+
+	rdr = avl_find(&rwa->defer_ranges, &srch, &where);
+	if (rdr == NULL) {
+		rdr = avl_nearest(&rwa->defer_ranges, where, AVL_BEFORE);
+		if (rdr != NULL && rdr->rdr_last + 1 < first)
+			rdr = NULL;
+	}
+	if (rdr == NULL) {
+		rdr = kmem_alloc(sizeof (*rdr), KM_SLEEP);
+		rdr->rdr_first = first;
+		rdr->rdr_last = last;
+		avl_insert(&rwa->defer_ranges, rdr, where);
+	} else {
+		rdr->rdr_last = MAX(rdr->rdr_last, last);
+	}
+
+	while ((next = AVL_NEXT(&rwa->defer_ranges, rdr)) != NULL &&
+	    next->rdr_first <= rdr->rdr_last + 1) {
+		rdr->rdr_last = MAX(rdr->rdr_last, next->rdr_last);
+		avl_remove(&rwa->defer_ranges, next);
+		kmem_free(next, sizeof (*next));
+	}
+}
+
+static boolean_t
+receive_defer_overlaps(struct receive_writer_arg *rwa, uint64_t first,
+    uint64_t last)
+{
+	receive_defer_range_t srch = { .rdr_first = first };
+	avl_index_t where;
+	receive_defer_range_t *rdr;
+
+	if (avl_find(&rwa->defer_ranges, &srch, &where) != NULL)
+		return (B_TRUE);
+
+	rdr = avl_nearest(&rwa->defer_ranges, where, AVL_BEFORE);
+	if (rdr != NULL && rdr->rdr_last >= first)
+		return (B_TRUE);
+
+	rdr = avl_nearest(&rwa->defer_ranges, where, AVL_AFTER);
+	if (rdr != NULL && rdr->rdr_first <= last)
+		return (B_TRUE);
+
+	return (B_FALSE);
+}
+
+static int receive_defer_flush(struct receive_writer_arg *rwa);
+
+/*
+ * Park a record for replay after the next defer flush.  The caller must
+ * return EAGAIN without touching the record again: once the batch cap is
+ * reached the flush below replays and frees it.  A flush failure is
+ * published through rwa->err (this always runs on the writer thread, the
+ * sole writer of that field).
+ */
+static void
+receive_defer_park(struct receive_writer_arg *rwa,
+    struct receive_record_arg *rrd, uint64_t object)
+{
+	/*
+	 * Nothing may defer during replay, or the list being drained
+	 * would grow and the record could be freed out from under the
+	 * replay loop.
+	 */
+	ASSERT(!rwa->defer_replaying);
+
+	if (!receive_defer_active(rwa)) {
+		/*
+		 * The resume pin depends on the first parked record being
+		 * an object record: resuming from (object, 0) is only a
+		 * valid stream position for a DRR_OBJECT.
+		 */
+		ASSERT3U(rrd->header.drr_type, ==, DRR_OBJECT);
+		rwa->defer_first_object = object;
+		rwa->defer_first_bytes_read = rrd->bytes_read;
+	}
+	list_insert_tail(&rwa->defer_records, rrd);
+	rwa->defer_nrecords++;
+	/* For WRITE and SPILL records the payload aliases the abd. */
+	rwa->defer_bytes += sizeof (*rrd) + (rrd->abd != NULL ?
+	    abd_get_size(rrd->abd) : rrd->payload_size);
+
+	if (rwa->defer_bytes >= zfs_recv_defer_batch_size) {
+		int err = receive_defer_flush(rwa);
+		if (err != 0 && rwa->err == 0)
+			rwa->err = err;
+	}
+}
+
+/*
+ * Sync out the deferred frees and replay the parked records in stream
+ * order.  Replay feeds each record back through receive_process_record();
+ * by now every deferred free has synced, so the replayed DRR_OBJECT
+ * records find their object numbers free and no longer defer.
+ */
+static int
+receive_defer_flush(struct receive_writer_arg *rwa)
+{
+	struct receive_record_arg *rrd;
+	receive_defer_range_t *rdr;
+	void *cookie = NULL;
+	int err;
+
+	if (!receive_defer_active(rwa))
+		return (0);
+
+	err = flush_write_batch(rwa);
+	if (err != 0)
+		return (err);
+
+	zfs_dbgmsg("recv defer flush: %llu records, %llu bytes, txg %llu",
+	    (u_longlong_t)rwa->defer_nrecords,
+	    (u_longlong_t)rwa->defer_bytes,
+	    (u_longlong_t)rwa->defer_max_free_txg);
+
+	/*
+	 * Every free a parked claim depends on was assigned no later
+	 * than defer_max_free_txg, so it is enough to wait for that txg
+	 * rather than for the whole open-txg window (typically one or
+	 * two txgs instead of up to five).
+	 */
+	txg_wait_synced(dmu_objset_pool(rwa->os), rwa->defer_max_free_txg);
+
+	while ((rdr = avl_destroy_nodes(&rwa->defer_ranges, &cookie)) != NULL)
+		kmem_free(rdr, sizeof (*rdr));
+
+	/*
+	 * Parked records are earlier in the stream than the writer's
+	 * current position, so drop the write-ordering cursor for the
+	 * replay; the parked records are in stream order among
+	 * themselves, and every record that follows the replay is at a
+	 * higher (object, offset) than anything replayed.  The
+	 * out-of-order check is correspondingly weakened until the
+	 * first replayed write raises the cursor again.
+	 */
+	rwa->last_object = 0;
+	rwa->last_offset = 0;
+	rwa->defer_replaying = B_TRUE;
+
+	while ((rrd = list_remove_head(&rwa->defer_records)) != NULL) {
+		if (err != 0) {
+			if (rrd->abd != NULL) {
+				abd_free(rrd->abd);
+				rrd->abd = NULL;
+				rrd->payload = NULL;
+			} else if (rrd->payload != NULL) {
+				vmem_free(rrd->payload, rrd->payload_size);
+				rrd->payload = NULL;
+			}
+			kmem_free(rrd, sizeof (*rrd));
+			continue;
+		}
+		int err2 = receive_process_record(rwa, rrd);
+		/*
+		 * As in receive_writer_thread(), EAGAIN means the record
+		 * was stashed on the write batch and must not be freed.
+		 */
+		if (err2 != EAGAIN) {
+			err = err2;
+			kmem_free(rrd, sizeof (*rrd));
+		}
+	}
+
+	if (err == 0)
+		err = flush_write_batch(rwa);
+
+	/*
+	 * Frees issued during the replay itself are not captured in
+	 * defer_max_free_txg once it resets below.  That stays safe:
+	 * any txg captured later is no earlier than theirs, and a batch
+	 * with nothing captured waits for everything (txg 0).
+	 */
+	rwa->defer_replaying = B_FALSE;
+	rwa->defer_bytes = 0;
+	rwa->defer_nrecords = 0;
+	rwa->defer_first_object = 0;
+	rwa->defer_first_bytes_read = 0;
+	rwa->defer_max_free_txg = 0;
+
+	return (err);
+}
+
+/*
+ * Free any records still parked after an error, along with the range
+ * tree.  No-op if a successful flush already drained both.
+ */
+static void
+receive_defer_cleanup(struct receive_writer_arg *rwa)
+{
+	struct receive_record_arg *rrd;
+	receive_defer_range_t *rdr;
+	void *cookie = NULL;
+
+	while ((rrd = list_remove_head(&rwa->defer_records)) != NULL) {
+		if (rrd->abd != NULL) {
+			abd_free(rrd->abd);
+			rrd->abd = NULL;
+			rrd->payload = NULL;
+		} else if (rrd->payload != NULL) {
+			vmem_free(rrd->payload, rrd->payload_size);
+			rrd->payload = NULL;
+		}
+		kmem_free(rrd, sizeof (*rrd));
+	}
+	while ((rdr = avl_destroy_nodes(&rwa->defer_ranges, &cookie)) != NULL)
+		kmem_free(rdr, sizeof (*rdr));
+	rwa->defer_bytes = 0;
+	rwa->defer_nrecords = 0;
+	rwa->defer_first_object = 0;
+	rwa->defer_first_bytes_read = 0;
+	rwa->defer_max_free_txg = 0;
+}
+
+/*
+ * If this record touches a dnode slot range with an unsynced free, park
+ * it for replay after the next defer flush.  Returns EAGAIN when the
+ * record was parked (the caller must not free or process it, and the
+ * park may already have flushed and freed it), or 0 when the record
+ * should be processed normally.
+ */
+static int
+receive_defer_check(struct receive_writer_arg *rwa,
+    struct receive_record_arg *rrd)
+{
+	uint64_t first, last;
+
+	switch (rrd->header.drr_type) {
+	case DRR_OBJECT:
+		/*
+		 * Object records are never parked here.  A multi-slot claim
+		 * can expand over a dnode that begins inside the claimed
+		 * range and ends past it, and only the expansion loop in
+		 * receive_object() finds that dnode, frees it and records
+		 * the extent it actually occupies.  Parking ahead of that
+		 * loop would leave the dnode allocated with its tail slots
+		 * outside the tree, and the DRR_FREEOBJECTS the sender
+		 * emits for those slots would then be applied against an
+		 * interior slot and fail the receive.  receive_object()
+		 * defers the claim itself, once the loop has run.
+		 */
+		return (0);
+	case DRR_WRITE:
+		first = last = rrd->header.drr_u.drr_write.drr_object;
+		break;
+	case DRR_WRITE_EMBEDDED:
+		first = last = rrd->header.drr_u.drr_write_embedded.drr_object;
+		break;
+	case DRR_FREE:
+		first = last = rrd->header.drr_u.drr_free.drr_object;
+		break;
+	case DRR_SPILL:
+		first = last = rrd->header.drr_u.drr_spill.drr_object;
+		break;
+	case DRR_REDACT:
+		first = last = rrd->header.drr_u.drr_redact.drr_object;
+		break;
+	case DRR_FREEOBJECTS:
+	{
+		struct drr_freeobjects *drrfo =
+		    &rrd->header.drr_u.drr_freeobjects;
+
+		/*
+		 * Senders routinely free the tail slots of an object
+		 * that was reallocated at a smaller dnode size, so these
+		 * ranges do overlap deferred ones.  Park the record; by
+		 * replay time the slots it covers were freed along with
+		 * the old dnode and the frees are no-ops.  The parked
+		 * range joins the tree so that anything a malformed
+		 * stream sends into it afterwards stays ordered behind
+		 * the free instead of racing it at replay.
+		 */
+		if (drrfo->drr_numobjs != 0 &&
+		    receive_defer_overlaps(rwa, drrfo->drr_firstobj,
+		    drrfo->drr_firstobj + drrfo->drr_numobjs - 1)) {
+			receive_defer_range_insert(rwa, drrfo->drr_firstobj,
+			    drrfo->drr_firstobj + drrfo->drr_numobjs - 1);
+			receive_defer_park(rwa, rrd, drrfo->drr_firstobj);
+			return (EAGAIN);
+		}
+		return (0);
+	}
+	default:
+		return (0);
+	}
+
+	if (receive_defer_overlaps(rwa, first, last)) {
+		receive_defer_park(rwa, rrd, first);
+		return (EAGAIN);
+	}
+	return (0);
+}
+
+static int
+receive_handle_existing_object(struct receive_writer_arg *rwa,
     const struct drr_object *drro, const dmu_object_info_t *doi,
     const void *bonus_data,
-    uint64_t *object_to_hold, uint32_t *new_blksz)
+    uint64_t *object_to_hold, uint32_t *new_blksz, boolean_t *deferp)
 {
 	uint32_t indblksz = drro->drr_indblkshift ?
 	    1ULL << drro->drr_indblkshift : 0;
@@ -2786,10 +3230,24 @@ receive_handle_existing_object(const struct receive_writer_arg *rwa,
 	    indblksz != doi->doi_metadata_block_size) ||
 	    drro->drr_nlevels < doi->doi_indirection)) ||
 	    dn_slots != doi->doi_dnodesize >> DNODE_SHIFT) {
-		err = dmu_free_long_object(rwa->os, drro->drr_object);
+		uint8_t old_slots = doi->doi_dnodesize >> DNODE_SHIFT;
+
+		err = receive_defer_free_object(rwa, drro->drr_object);
 		if (err != 0)
 			return (SET_ERROR(EINVAL));
 
+		if (receive_defer_enabled(rwa)) {
+			/*
+			 * Defer the claim instead of syncing; the caller
+			 * parks this record for replay after the batched
+			 * sync (see receive_defer_park()).
+			 */
+			receive_defer_range_insert(rwa, drro->drr_object,
+			    drro->drr_object + MAX(old_slots, dn_slots) - 1);
+			*deferp = B_TRUE;
+			return (0);
+		}
+
 		txg_wait_synced(dmu_objset_pool(rwa->os), 0);
 		*object_to_hold = DMU_NEW_OBJECT;
 	}
@@ -2875,10 +3333,32 @@ receive_object(struct receive_writer_arg *rwa, struct drr_object *drro,
 	 * Raw receives will also check that the indirect structure of the
 	 * dnode hasn't changed.
 	 */
-	uint64_t object_to_hold;
+	uint64_t object_to_hold = DMU_NEW_OBJECT;
+	boolean_t defer = B_FALSE;
+
+	/*
+	 * Claiming slots that a parked record already covers has to wait
+	 * for the batch in any case, but the expansion loop below still
+	 * has to run first: it is what discovers a dnode reaching past
+	 * the end of this claim, frees it, and puts the slots it really
+	 * occupies in the tree.  Without that, the frees the sender sends
+	 * for those trailing slots would not be parked behind this claim.
+	 */
+	if (receive_defer_enabled(rwa) && receive_defer_overlaps(rwa,
+	    drro->drr_object, drro->drr_object + dn_slots - 1)) {
+		receive_defer_range_insert(rwa, drro->drr_object,
+		    drro->drr_object + dn_slots - 1);
+		defer = B_TRUE;
+	}
+
 	if (err == 0) {
+		/*
+		 * When the claim is deferred, still fall through to the
+		 * multi-slot expansion below so that neighbor frees are
+		 * issued now and share the batched sync.
+		 */
 		err = receive_handle_existing_object(rwa, drro, &doi, data,
-		    &object_to_hold, &new_blksz);
+		    &object_to_hold, &new_blksz, &defer);
 		if (err != 0)
 			return (err);
 	} else if (err == EEXIST) {
@@ -2889,10 +3369,17 @@ receive_object(struct receive_writer_arg *rwa, struct drr_object *drro,
 		 * to free this slot when we freed the associated dnode
 		 * earlier in the stream.
 		 */
-		txg_wait_synced(dmu_objset_pool(rwa->os), 0);
+		if (receive_defer_enabled(rwa)) {
+			receive_defer_range_insert(rwa, drro->drr_object,
+			    drro->drr_object + dn_slots - 1);
+			defer = B_TRUE;
+		} else {
+			txg_wait_synced(dmu_objset_pool(rwa->os), 0);
 
-		if (dmu_object_info(rwa->os, drro->drr_object, NULL) != ENOENT)
-			return (SET_ERROR(EINVAL));
+			if (dmu_object_info(rwa->os, drro->drr_object,
+			    NULL) != ENOENT)
+				return (SET_ERROR(EINVAL));
+		}
 
 		/* object was freed and we are about to allocate a new one */
 		object_to_hold = DMU_NEW_OBJECT;
@@ -2933,17 +3420,40 @@ receive_object(struct receive_writer_arg *rwa, struct drr_object *drro,
 			else if (err != 0)
 				return (err);
 
-			err = dmu_free_long_object(rwa->os, slot);
+			err = receive_defer_free_object(rwa, slot);
 			if (err != 0)
 				return (err);
 
+			if (receive_defer_enabled(rwa)) {
+				/*
+				 * The freed neighbor may itself have been
+				 * a multi-slot dnode extending past the
+				 * range being claimed here.
+				 */
+				uint8_t slot_slots =
+				    slot_doi.doi_dnodesize >> DNODE_SHIFT;
+				receive_defer_range_insert(rwa, slot,
+				    slot + slot_slots - 1);
+			}
+
 			need_sync = B_TRUE;
 		}
 
-		if (need_sync)
-			txg_wait_synced(dmu_objset_pool(rwa->os), 0);
+		if (need_sync) {
+			if (receive_defer_enabled(rwa)) {
+				receive_defer_range_insert(rwa,
+				    drro->drr_object,
+				    drro->drr_object + dn_slots - 1);
+				defer = B_TRUE;
+			} else {
+				txg_wait_synced(dmu_objset_pool(rwa->os), 0);
+			}
+		}
 	}
 
+	if (defer)
+		return (EAGAIN);
+
 	tx = dmu_tx_create(rwa->os);
 	dmu_tx_hold_bonus(tx, object_to_hold);
 	dmu_tx_hold_write(tx, object_to_hold, 0, 0);
@@ -3115,7 +3625,7 @@ receive_freeobjects(struct receive_writer_arg *rwa,
 		else if (err != 0)
 			return (err);
 
-		err = dmu_free_long_object(rwa->os, obj);
+		err = receive_defer_free_object(rwa, obj);
 
 		if (err != 0)
 			return (err);
@@ -4116,9 +4626,15 @@ receive_process_record(struct receive_writer_arg *rwa,
 {
 	int err;
 
-	/* Processing in order, therefore bytes_read should be increasing. */
-	ASSERT3U(rrd->bytes_read, >=, rwa->bytes_read);
-	rwa->bytes_read = rrd->bytes_read;
+	/*
+	 * Processing in order, therefore bytes_read should be increasing.
+	 * Replayed records are earlier in the stream than the writer's
+	 * current position, so leave the high-water mark alone for them.
+	 */
+	if (!rwa->defer_replaying) {
+		ASSERT3U(rrd->bytes_read, >=, rwa->bytes_read);
+		rwa->bytes_read = rrd->bytes_read;
+	}
 
 	/* We can only heal write records; other ones get ignored */
 	if (rwa->heal && rrd->header.drr_type != DRR_WRITE) {
@@ -4132,6 +4648,16 @@ receive_process_record(struct receive_writer_arg *rwa,
 		return (0);
 	}
 
+	/*
+	 * Records that depend on a deferred object claim are parked in
+	 * stream order behind it; everything else keeps flowing.
+	 */
+	if (!rwa->heal && !rwa->defer_replaying && receive_defer_active(rwa)) {
+		err = receive_defer_check(rwa, rrd);
+		if (err != 0)
+			return (err);
+	}
+
 	if (!rwa->heal && rrd->header.drr_type != DRR_WRITE) {
 		err = flush_write_batch(rwa);
 		if (err != 0) {
@@ -4153,6 +4679,16 @@ receive_process_record(struct receive_writer_arg *rwa,
 	{
 		struct drr_object *drro = &rrd->header.drr_u.drr_object;
 		err = receive_object(rwa, drro, rrd->payload);
+		if (err == EAGAIN) {
+			/*
+			 * The object's claim was deferred behind a batched
+			 * txg sync; keep the record (and its bonus payload)
+			 * parked for replay.  The park may flush and free
+			 * the record, so it must not be touched again.
+			 */
+			receive_defer_park(rwa, rrd, drro->drr_object);
+			return (EAGAIN);
+		}
 		vmem_free(rrd->payload, rrd->payload_size);
 		rrd->payload = NULL;
 		break;
@@ -4264,9 +4800,11 @@ receive_writer_thread(void *arg)
 			rrd->payload = NULL;
 		}
 		/*
-		 * EAGAIN indicates that this record has been saved (on
-		 * raw->write_batch), and will be used again, so we don't
-		 * free it.
+		 * EAGAIN indicates that ownership of this record has
+		 * moved: it was either saved on rwa->write_batch, or
+		 * parked on the defer list to be replayed and freed by
+		 * the defer flush (or by the defer cleanup on error).
+		 * Either way we don't free it here.
 		 * When healing data we always need to free the record.
 		 */
 		if (err != EAGAIN || rwa->heal) {
@@ -4280,10 +4818,16 @@ receive_writer_thread(void *arg)
 	if (rwa->heal) {
 		zio_wait(rwa->heal_pio);
 	} else {
-		int err = flush_write_batch(rwa);
+		int err = 0;
+		if (rwa->err == 0)
+			err = receive_defer_flush(rwa);
+		if (rwa->err == 0)
+			rwa->err = err;
+		err = flush_write_batch(rwa);
 		if (rwa->err == 0)
 			rwa->err = err;
 	}
+	receive_defer_cleanup(rwa);
 	mutex_enter(&rwa->mutex);
 	rwa->done = B_TRUE;
 	cv_signal(&rwa->cv);
@@ -4435,6 +4979,11 @@ dmu_recv_stream(dmu_recv_cookie_t *drc, offset_t *voffp)
 	}
 	list_create(&rwa->write_batch, sizeof (struct receive_record_arg),
 	    offsetof(struct receive_record_arg, node.bqn_node));
+	list_create(&rwa->defer_records, sizeof (struct receive_record_arg),
+	    offsetof(struct receive_record_arg, node.bqn_node));
+	avl_create(&rwa->defer_ranges, receive_defer_range_compare,
+	    sizeof (receive_defer_range_t),
+	    offsetof(receive_defer_range_t, rdr_node));
 
 	(void) thread_create(NULL, 0, receive_writer_thread, rwa, 0, curproc,
 	    TS_RUN, minclsyspri);
@@ -4522,6 +5071,8 @@ dmu_recv_stream(dmu_recv_cookie_t *drc, offset_t *voffp)
 	mutex_destroy(&rwa->mutex);
 	bqueue_destroy(&rwa->q);
 	list_destroy(&rwa->write_batch);
+	list_destroy(&rwa->defer_records);
+	avl_destroy(&rwa->defer_ranges);
 	if (err == 0)
 		err = rwa->err;
 
@@ -4886,5 +5437,9 @@ ZFS_MODULE_PARAM(zfs_recv, zfs_recv_, queue_ff, UINT, ZMOD_RW,
 ZFS_MODULE_PARAM(zfs_recv, zfs_recv_, write_batch_size, UINT, ZMOD_RW,
 	"Maximum amount of writes to batch into one transaction");
 
+ZFS_MODULE_PARAM(zfs_recv, zfs_recv_, defer_batch_size, UINT, ZMOD_RW,
+	"Maximum bytes of records parked behind one txg sync while "
+	"receiving reallocated dnodes (0 to sync per object)");
+
 ZFS_MODULE_PARAM(zfs_recv, zfs_recv_, best_effort_corrective, INT, ZMOD_RW,
 	"Ignore errors during corrective receive");

tests/runfiles/common.run

@@ -1016,7 +1016,8 @@ tests = ['recv_dedup', 'recv_dedup_encrypted_zvol', 'rsend_001_pos',
     'send_encrypted_holds',
     'send_encrypted_props', 'send_encrypted_truncated_files',
     'send_freeobjects', 'send_realloc_files', 'send_realloc_encrypted_files',
-    'send_realloc_dnode_nblkptr',
+    'send_realloc_dnode_nblkptr', 'send_realloc_dnode_mixed',
+    'send_realloc_dnode_resume',
     'send_spill_block', 'send_split_large_block', 'send_holds',
     'send_hole_birth', 'send_mixed_raw',
     'send-wR_encrypted_zvol',

tests/runfiles/linux.run

@@ -200,7 +200,8 @@ tests = ['renameat2_noreplace', 'renameat2_exchange', 'renameat2_whiteout']
 tags = ['functional', 'renameat2']
 
 [tests/functional/rsend:Linux]
-tests = ['send_realloc_dnode_size', 'send_encrypted_files', 'send-c_longname']
+tests = ['send_realloc_dnode_size', 'send_encrypted_files',
+    'send-c_longname']
 tags = ['functional', 'rsend']
 
 [tests/functional/simd:Linux]

tests/zfs-tests/include/tunables.cfg

@@ -77,6 +77,7 @@ OVERRIDE_ESTIMATE_RECORDSIZE	send.override_estimate_recordsize	zfs_override_esti
 PREFETCH_DISABLE		prefetch.disable		zfs_prefetch_disable
 RAIDZ_EXPAND_MAX_REFLOW_BYTES	vdev.expand_max_reflow_bytes	raidz_expand_max_reflow_bytes
 READ_SIT_OUT_SECS		vdev.read_sit_out_secs		vdev_read_sit_out_secs
+RECV_DEFER_BATCH_SIZE		recv.defer_batch_size		zfs_recv_defer_batch_size
 SIT_OUT_CHECK_INTERVAL		vdev.raidz_outlier_check_interval_ms	vdev_raidz_outlier_check_interval_ms
 SIT_OUT_INSENSITIVITY		vdev.raidz_outlier_insensitivity	vdev_raidz_outlier_insensitivity
 REBUILD_SCRUB_ENABLED		rebuild_scrub_enabled		zfs_rebuild_scrub_enabled

tests/zfs-tests/tests/Makefile.am

@@ -2186,7 +2186,9 @@ nobase_dist_datadir_zfs_tests_tests_SCRIPTS += \
 	functional/rsend/send_raw_ashift.ksh \
 	functional/rsend/send_raw_spill_block.ksh \
 	functional/rsend/send_raw_large_blocks.ksh \
+	functional/rsend/send_realloc_dnode_mixed.ksh \
 	functional/rsend/send_realloc_dnode_nblkptr.ksh \
+	functional/rsend/send_realloc_dnode_resume.ksh \
 	functional/rsend/send_realloc_dnode_size.ksh \
 	functional/rsend/send_realloc_encrypted_files.ksh \
 	functional/rsend/send_realloc_files.ksh \

tests/zfs-tests/tests/functional/rsend/send_realloc_dnode_mixed.ksh

@@ -0,0 +1,113 @@
+#!/bin/ksh
+# SPDX-License-Identifier: CDDL-1.0
+
+#
+# This file and its contents are supplied under the terms of the
+# Common Development and Distribution License ("CDDL"), version 1.0.
+# You may only use this file in accordance with the terms of version
+# 1.0 of the CDDL.
+#
+# A full copy of the text of the CDDL should have accompanied this
+# source.  A copy of the CDDL is also available via the Internet at
+# http://www.illumos.org/license/CDDL.
+#
+
+#
+# Copyright (c) 2026 MorganaFuture
+#
+
+. $STF_SUITE/include/libtest.shlib
+. $STF_SUITE/tests/functional/rsend/rsend.kshlib
+
+#
+# Description:
+# Verify incremental receive handles a wide dnode claim that expands over
+# a dnode reaching past the end of the claimed slot range.
+#
+# send_realloc_dnode_size covers reallocation between uniform dnode sizes,
+# where every object of a snapshot starts on the same alignment and a wider
+# claim therefore always ends on a dnode boundary.  Mixing the sizes puts
+# multi-slot dnodes at odd offsets, so a claim can end in the middle of one
+# and only the expansion loop in receive_object() is left to free it.
+#
+# Strategy:
+# 1. Populate a dataset with the dnode size cycling through legacy, 1k and
+#    2k so that the multi-slot dnodes sit at differing alignments
+# 2. Remove the objects and refill at 2k, then remove part of what was
+#    written so that the reallocated objects are separated by holes and the
+#    sender frees the slots between them
+# 3. Generate the initial and incremental streams
+# 4. Verify both can be received and the result matches the source
+#
+
+verify_runnable "both"
+
+log_assert "Verify incremental receive handles claims over misaligned dnodes"
+
+function cleanup
+{
+	rm -f $BACKDIR/fs-mixed
+	rm -f $BACKDIR/fs-realloc
+
+	datasetexists $POOL/fs && destroy_dataset $POOL/fs -rR
+	datasetexists $POOL/newfs && destroy_dataset $POOL/newfs -rR
+}
+
+log_onexit cleanup
+
+typeset -i nfiles=150
+typeset -i i=0
+
+# 1. Populate a dataset with the dnode size cycling through legacy, 1k and
+#    2k so that the multi-slot dnodes sit at differing alignments
+log_must zfs create $POOL/fs
+
+while (( i < nfiles )); do
+	case $((i % 3)) in
+	0)	log_must zfs set dnodesize=legacy $POOL/fs ;;
+	1)	log_must zfs set dnodesize=1k $POOL/fs ;;
+	2)	log_must zfs set dnodesize=2k $POOL/fs ;;
+	esac
+
+	log_must touch /$POOL/fs/file.$i
+	(( i += 1 ))
+done
+
+log_must zfs snapshot $POOL/fs@a
+
+# 2. Remove the objects and refill at 2k, then remove part of what was
+#    written so that the reallocated objects are separated by holes and the
+#    sender frees the slots between them
+log_must eval "rm -f /$POOL/fs/file.*"
+
+log_must zfs unmount $POOL/fs
+log_must zfs set dnodesize=2k $POOL/fs
+log_must zfs mount $POOL/fs
+
+i=0
+while (( i < nfiles )); do
+	log_must touch /$POOL/fs/new.$i
+	(( i += 1 ))
+done
+
+i=0
+while (( i < nfiles )); do
+	if (( i % 3 == 0 )); then
+		log_must rm -f /$POOL/fs/new.$i
+	fi
+	(( i += 1 ))
+done
+
+log_must zfs snapshot $POOL/fs@b
+
+# 3. Generate the initial and incremental streams
+log_must eval "zfs send $POOL/fs@a > $BACKDIR/fs-mixed"
+log_must eval "zfs send -i $POOL/fs@a $POOL/fs@b > $BACKDIR/fs-realloc"
+
+# 4. Verify both can be received and the result matches the source
+log_must eval "zfs recv $POOL/newfs < $BACKDIR/fs-mixed"
+log_must eval "zfs recv $POOL/newfs < $BACKDIR/fs-realloc"
+
+log_must directory_diff /$POOL/fs /$POOL/newfs
+
+log_pass "Verify incremental receive handles claims over misaligned dnodes"

tests/zfs-tests/tests/functional/rsend/send_realloc_dnode_resume.ksh

@@ -0,0 +1,95 @@
+#!/bin/ksh -p
+# SPDX-License-Identifier: CDDL-1.0
+
+#
+# This file and its contents are supplied under the terms of the
+# Common Development and Distribution License ("CDDL"), version 1.0.
+# You may only use this file in accordance with the terms of version
+# 1.0 of the CDDL.
+#
+# A full copy of the text of the CDDL should have accompanied this
+# source.  A copy of the CDDL is also available via the Internet at
+# http://www.illumos.org/license/CDDL.
+#
+
+#
+# Copyright (c) 2026 MorganaFuture
+#
+
+. $STF_SUITE/tests/functional/rsend/rsend.kshlib
+
+#
+# Description:
+# Verify resumability of incremental send/receive when the incremental
+# stream reallocates objects with a changed dnode slot count.
+#
+# Strategy:
+# 1. Populate a dataset with 1k byte dnodes and snapshot
+# 2. Remove objects, set dnodesize=legacy, and remount dataset so new
+#    objects get recycled numbers at a smaller slot count
+# 3. Remove objects, set dnodesize=2k, and remount dataset so new objects
+#    expand over recently recycled slots
+# 4. Send each stream into a corrupted state file, verify the receive
+#    fails, then resume from the receive_resume_token
+# 5. Verify the received snapshots match the sent ones
+#
+
+verify_runnable "both"
+
+sendfs=$POOL/sendfs
+recvfs=$POOL2/recvfs
+streamfs=$POOL/stream
+
+typeset saved_defer_batch=$(get_tunable RECV_DEFER_BATCH_SIZE)
+
+function cleanup
+{
+	log_must set_tunable32 RECV_DEFER_BATCH_SIZE $saved_defer_batch
+	resume_cleanup $sendfs $streamfs
+}
+
+log_assert "Verify resume of interrupted receive with changed dnode size"
+log_onexit cleanup
+
+# Shrink the defer batch so the receive flushes parked records mid-stream
+# instead of only at end-of-stream or interruption.
+log_must set_tunable32 RECV_DEFER_BATCH_SIZE $((1024 * 1024))
+
+# 1. Populate a dataset with 1k byte dnodes and snapshot
+log_must zfs create -o dnodesize=1k $sendfs
+log_must zfs create $streamfs
+log_must mk_files 200 262144 0 $sendfs
+log_must zfs snapshot $sendfs@a
+
+# 2. Remove objects, set dnodesize=legacy, and remount dataset so new
+#    objects get recycled numbers at a smaller slot count
+log_must eval "rm -f /$sendfs/*"
+log_must zfs unmount $sendfs
+log_must zfs set dnodesize=legacy $sendfs
+log_must zfs mount $sendfs
+log_must mk_files 200 262144 0 $sendfs
+log_must zfs snapshot $sendfs@b
+
+# 3. Remove objects, set dnodesize=2k, and remount dataset so new objects
+#    expand over recently recycled slots
+log_must eval "rm -f /$sendfs/*"
+log_must zfs unmount $sendfs
+log_must zfs set dnodesize=2k $sendfs
+log_must zfs mount $sendfs
+log_must mk_files 200 262144 0 $sendfs
+log_must zfs snapshot $sendfs@c
+
+# 4. Send each stream into a corrupted state file, verify the receive
+#    fails, then resume from the receive_resume_token
+resume_test "zfs send -v $sendfs@a" $streamfs $recvfs
+resume_test "zfs send -v -i @a $sendfs@b" $streamfs $recvfs
+resume_test "zfs send -v -i @b $sendfs@c" $streamfs $recvfs
+
+# 5. Verify the received snapshots match the sent ones.  The receives
+#    above use -u, so mount the destination first or the .zfs snapshot
+#    directories the diffs need would not exist.
+log_must zfs mount $recvfs
+file_check $sendfs $recvfs
+log_must directory_diff /$recvfs/.zfs/snapshot/c /$sendfs/.zfs/snapshot/c
+
+log_pass "Resume of interrupted receive with changed dnode size works"