4986 receiving replication stream fails if any snapshot exceeds refquota

@@ -4088,10 +4088,59 @@
 next:
                 pair = next_pair;
         }
 }
 
+/*
+ * Extract properties that cannot be set PRIOR to the receipt of a dataset.
+ * For example, refquota cannot be set until after the receipt of a dataset,
+ * because a prior snapshot may exceed the refquota, and refquotas only apply
+ * to the current dataset.  The caller (libzfs) will manage these properties
+ * somewhat as well to make sure they only come down with the last dataset in
+ * a replication stream, but we still need to be safe about it here in
+ * kernel-land.
+ */
+static nvlist_t *
+extract_delay_props(nvlist_t *props)
+{
+        nvlist_t *delayprops;
+        nvpair_t *nvp, *tmp;
+        static const zfs_prop_t delayable[] =
+            { ZFS_PROP_REFQUOTA, ZFS_PROP_REFRESERVATION, 0 };
+        boolean_t dontbother = B_TRUE;
+        int i;
+
+        VERIFY(nvlist_alloc(&delayprops, NV_UNIQUE_NAME, KM_SLEEP) == 0);
+
+        for (nvp = nvlist_next_nvpair(props, NULL); nvp != NULL;
+            nvp = nvlist_next_nvpair(props, nvp)) {
+                /*
+                 * strcmp() is safe because zfs_prop_to_name() always returns
+                 * a bounded string.
+                 */
+                for (i = 0; delayable[i] != 0; i++) {
+                        if (strcmp(zfs_prop_to_name(delayable[i]),
+                            nvpair_name(nvp)) == 0) {
+                                break;
+                        }
+                }
+                if (delayable[i] != 0) {
+                        tmp = nvlist_prev_nvpair(props, nvp);
+                        VERIFY(nvlist_add_nvpair(delayprops, nvp) == 0);
+                        VERIFY(nvlist_remove_nvpair(props, nvp) == 0);
+                        nvp = tmp;
+                        dontbother = B_FALSE;  /* Actually, do bother! */
+                }
+        }
+
+        if (dontbother) {
+                nvlist_free(delayprops);
+                delayprops = NULL;
+        }
+        return (delayprops);
+}
+
 #ifdef  DEBUG
 static boolean_t zfs_ioc_recv_inject_err;
 #endif
 
 /*

@@ -4124,10 +4173,11 @@
         int props_error = 0;
         nvlist_t *errors;
         offset_t off;
         nvlist_t *props = NULL; /* sent properties */
         nvlist_t *origprops = NULL; /* existing properties */
+        nvlist_t *delayprops = NULL; /* sent properties applied post-receive */
         char *origin = NULL;
         char *tosnap;
         char tofs[ZFS_MAXNAMELEN];
         boolean_t first_recvd_props = B_FALSE;
 

@@ -4204,25 +4254,16 @@
 
         if (props != NULL) {
                 props_error = dsl_prop_set_hasrecvd(tofs);
 
                 if (props_error == 0) {
+                        delayprops = extract_delay_props(props);
                         (void) zfs_set_prop_nvlist(tofs, ZPROP_SRC_RECEIVED,
                             props, errors);
                 }
         }
 
-        if (zc->zc_nvlist_dst_size != 0 &&
-            (nvlist_smush(errors, zc->zc_nvlist_dst_size) != 0 ||
-            put_nvlist(zc, errors) != 0)) {
-                /*
-                 * Caller made zc->zc_nvlist_dst less than the minimum expected
-                 * size or supplied an invalid address.
-                 */
-                props_error = SET_ERROR(EINVAL);
-        }
-
         off = fp->f_offset;
         error = dmu_recv_stream(&drc, fp->f_vnode, &off, zc->zc_cleanup_fd,
             &zc->zc_action_handle);
 
         if (error == 0) {

@@ -4243,12 +4284,35 @@
                         error = error ? error : end_err;
                         VFS_RELE(zfsvfs->z_vfs);
                 } else {
                         error = dmu_recv_end(&drc, NULL);
                 }
+
+                /* Set delayed properties now, after we're done receiving. */
+                if (delayprops != NULL) {
+                        (void) zfs_set_prop_nvlist(tofs, ZPROP_SRC_RECEIVED,
+                            delayprops, errors);
         }
+        }
 
+        /* Merge delayprops back in with regular props, in case of errors. */
+        if (delayprops != NULL) {
+                VERIFY(nvlist_merge(props, delayprops, 0) == 0);
+                nvlist_free(delayprops);
+        }
+
+        /* Put the props error list into zc AFTER the delayprops. */
+        if (zc->zc_nvlist_dst_size != 0 &&
+            (nvlist_smush(errors, zc->zc_nvlist_dst_size) != 0 ||
+            put_nvlist(zc, errors) != 0)) {
+                /*
+                 * Caller made zc->zc_nvlist_dst less than the minimum expected
+                 * size or supplied an invalid address.
+                 */
+                props_error = SET_ERROR(EINVAL);
+        }
+
         zc->zc_cookie = off - fp->f_offset;
         if (VOP_SEEK(fp->f_vnode, fp->f_offset, &off, NULL) == 0)
                 fp->f_offset = off;
 
 #ifdef  DEBUG