4986 receiving replication stream fails if any snapshot exceeds refquota
Reviewed by: John Kennedy <john.kennedy@delphix.com>
Reviewed by: Matthew Ahrens <mahrens@delphix.com>

@@ -20,10 +20,11 @@
  */
 
 /*
  * Copyright (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.
  * Portions Copyright 2011 Martin Matuska
+ * Copyright 2015, OmniTI Computer Consulting, Inc. All rights reserved.
  * Copyright 2015 Nexenta Systems, Inc.  All rights reserved.
  * Copyright (c) 2014, Joyent, Inc. All rights reserved.
  * Copyright (c) 2011, 2015 by Delphix. All rights reserved.
  * Copyright (c) 2013 by Saso Kiselkov. All rights reserved.
  * Copyright (c) 2013 Steven Hartland. All rights reserved.

@@ -4088,10 +4089,60 @@
 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 in replication streams, an older/earlier snapshot may exceed the
+ * refquota.  We want to receive the older/earlier snapshot, but setting
+ * refquota pre-receipt will set the dsl's ACTUAL quota, which will prevent
+ * the older/earlier snapshot from being received (with EDQUOT).
+ *
+ * The ZFS test "zfs_receive_011_pos" demonstrates such a scenario.
+ *
+ * libzfs will need to be judicious handling errors encountered by props
+ * extracted by this function.
+ */
+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, 0 };
+        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;
+                }
+        }
+
+        if (nvlist_empty(delayprops)) {
+                nvlist_free(delayprops);
+                delayprops = NULL;
+        }
+        return (delayprops);
+}
+
 #ifdef  DEBUG
 static boolean_t zfs_ioc_recv_inject_err;
 #endif
 
 /*

@@ -4124,10 +4175,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 +4256,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 +4286,46 @@
                         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 && error == 0) {
+                        (void) zfs_set_prop_nvlist(tofs, ZPROP_SRC_RECEIVED,
+                            delayprops, errors);
         }
+        }
 
+        if (delayprops != NULL) {
+                /*
+                 * Merge delayed props back in with initial props, in case
+                 * we're DEBUG and zfs_ioc_recv_inject_err is set (which means
+                 * we have to make sure clear_received_props() includes
+                 * the delayed properties).
+                 *
+                 * Since zfs_ioc_recv_inject_err is only in DEBUG kernels,
+                 * using ASSERT() will be just like a VERIFY.
+                 */
+                ASSERT(nvlist_merge(props, delayprops, 0) == 0);
+                nvlist_free(delayprops);
+        }
+
+        /*
+         * Now that all props, initial and delayed, are set, report the prop
+         * errors to the caller.
+         */
+        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