rpc/dataconn/timeoutconn: disable TestPartialWriteMockConn in CircleCI

fixes https://github.com/zrepl/zrepl/issues/540

Makefile

@@ -189,6 +189,8 @@ GO_SUPPORTS_ILLUMOS := $(shell $(GO) version | gawk -F '.' '/^go version /{split
 bins-all:
 	$(MAKE) $(BINS_ALL_TARGETS) GOOS=freebsd   GOARCH=amd64
 	$(MAKE) $(BINS_ALL_TARGETS) GOOS=freebsd   GOARCH=386
+	$(MAKE) $(BINS_ALL_TARGETS) GOOS=freebsd   GOARCH=arm	  GOARM=7
+	$(MAKE) $(BINS_ALL_TARGETS) GOOS=freebsd   GOARCH=arm64
 	$(MAKE) $(BINS_ALL_TARGETS) GOOS=linux     GOARCH=amd64
 	$(MAKE) $(BINS_ALL_TARGETS) GOOS=linux     GOARCH=arm64
 	$(MAKE) $(BINS_ALL_TARGETS) GOOS=linux     GOARCH=arm     GOARM=7

README.md

@@ -95,7 +95,7 @@ Downstream packagers can read the changelog to determine whether they want to pu
 
 ### Additional Notes to Distro Package Maintainers
 
-* Use `sudo make test-platform-bin && sudo make test-platform` **on a test system** to validate that zrepl's abstractions on top of ZFS work with the system ZFS.
+* Run the platform tests (Docs -> Usage -> Platform Tests) **on a test system** to validate that zrepl's abstractions on top of ZFS work with the system ZFS.
 * Ship a default config that adheres to your distro's `hier` and logging system.
 * Ship a service manager file and _please_ try to upstream it to this repository.
   * `dist/systemd` contains a Systemd unit template.

client/status/viewmodel/render.go

@@ -216,7 +216,7 @@ func drawJob(t *stringbuilder.B, name string, v *job.Status, history *bytesProgr
 	}
 }
 
-func printFilesystemStatus(t *stringbuilder.B, rep *report.FilesystemReport, active bool, maxFS int) {
+func printFilesystemStatus(t *stringbuilder.B, rep *report.FilesystemReport, maxFS int) {
 
 	expected, replicated, containsInvalidSizeEstimates := rep.BytesSum()
 	sizeEstimationImpreciseNotice := ""
@@ -227,15 +227,20 @@ func printFilesystemStatus(t *stringbuilder.B, rep *report.FilesystemReport, act
 		sizeEstimationImpreciseNotice = " (step lacks size estimation)"
 	}
 
+	userVisisbleCurrentStep, userVisibleTotalSteps := rep.CurrentStep, len(rep.Steps)
+	if len(rep.Steps) > 0 {
+		userVisisbleCurrentStep = rep.CurrentStep + 1 // CurrentStep is an index that starts at 0
+	}
 	status := fmt.Sprintf("%s (step %d/%d, %s/%s)%s",
 		strings.ToUpper(string(rep.State)),
-		rep.CurrentStep, len(rep.Steps),
+		userVisisbleCurrentStep, userVisibleTotalSteps,
 		ByteCountBinaryUint(replicated), ByteCountBinaryUint(expected),
 		sizeEstimationImpreciseNotice,
 	)
 
 	activeIndicator := " "
-	if active {
+	if rep.BlockedOn == report.FsBlockedOnNothing &&
+		(rep.State == report.FilesystemPlanning || rep.State == report.FilesystemStepping) {
 		activeIndicator = "*"
 	}
 	t.AddIndent(1)
@@ -385,7 +390,7 @@ func renderReplicationReport(t *stringbuilder.B, rep *report.Report, history *by
 			}
 		}
 		for _, fs := range latest.Filesystems {
-			printFilesystemStatus(t, fs, false, maxFSLen) // FIXME bring 'active' flag back
+			printFilesystemStatus(t, fs, maxFSLen)
 		}
 
 	}

client/testcmd.go

@@ -139,9 +139,11 @@ var testPlaceholder = &cli.Subcommand{
 func runTestPlaceholder(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
 
 	var checkDPs []*zfs.DatasetPath
+	var datasetWasExplicitArgument bool
 
 	// all actions first
 	if testPlaceholderArgs.all {
+		datasetWasExplicitArgument = false
 		out, err := zfs.ZFSList(ctx, []string{"name"})
 		if err != nil {
 			return errors.Wrap(err, "could not list ZFS filesystems")
@@ -154,6 +156,7 @@ func runTestPlaceholder(ctx context.Context, subcommand *cli.Subcommand, args []
 			checkDPs = append(checkDPs, dp)
 		}
 	} else {
+		datasetWasExplicitArgument = true
 		dp, err := zfs.NewDatasetPath(testPlaceholderArgs.ds)
 		if err != nil {
 			return err
@@ -171,7 +174,12 @@ func runTestPlaceholder(ctx context.Context, subcommand *cli.Subcommand, args []
 			return errors.Wrap(err, "cannot get placeholder state")
 		}
 		if !ph.FSExists {
-			panic("placeholder state inconsistent: filesystem " + ph.FS + " must exist in this context")
+			if datasetWasExplicitArgument {
+				return errors.Errorf("filesystem %q does not exist", ph.FS)
+			} else {
+				// got deleted between ZFSList and ZFSGetFilesystemPlaceholderState
+				continue
+			}
 		}
 		is := "yes"
 		if !ph.IsPlaceholder {

config/config.go

@@ -86,7 +86,7 @@ type SendOptions struct {
 	BackupProperties bool `yaml:"backup_properties,optional,default=false"`
 	LargeBlocks      bool `yaml:"large_blocks,optional,default=false"`
 	Compressed       bool `yaml:"compressed,optional,default=false"`
-	EmbeddedData     bool `yaml:"embbeded_data,optional,default=false"`
+	EmbeddedData     bool `yaml:"embedded_data,optional,default=false"`
 	Saved            bool `yaml:"saved,optional,default=false"`
 
 	BandwidthLimit *BandwidthLimit `yaml:"bandwidth_limit,optional,fromdefaults"`
@@ -101,6 +101,8 @@ type RecvOptions struct {
 	Properties *PropertyRecvOptions `yaml:"properties,fromdefaults"`
 
 	BandwidthLimit *BandwidthLimit `yaml:"bandwidth_limit,optional,fromdefaults"`
+
+	Placeholder *PlaceholderRecvOptions `yaml:"placeholder,fromdefaults"`
 }
 
 var _ yaml.Unmarshaler = &datasizeunit.Bits{}
@@ -130,6 +132,10 @@ type PropertyRecvOptions struct {
 	Override map[zfsprop.Property]string `yaml:"override,optional"`
 }
 
+type PlaceholderRecvOptions struct {
+	Encryption string `yaml:"encryption,default=unspecified"`
+}
+
 type PushJob struct {
 	ActiveJob    `yaml:",inline"`
 	Snapshotting SnapshottingEnum  `yaml:"snapshotting"`

config/samples/quickstart_continuous_server_backup_receiver.yml

@@ -9,4 +9,7 @@ jobs:
       key: "/etc/zrepl/backups.key"
       client_cns:
         - "prod"
+  recv:
+    placeholder:
+      encryption: inherit # use 'off' if sender uses send.encrypted
   root_fs: "storage/zrepl/sink"

daemon/control.go

@@ -8,6 +8,7 @@ import (
 	"io"
 	"net"
 	"net/http"
+	"os"
 	"time"
 
 	"github.com/pkg/errors"
@@ -124,8 +125,9 @@ func (j *controlJob) Run(ctx context.Context) {
 			s := Status{
 				Jobs: jobs,
 				Global: GlobalStatus{
-					ZFSCmds:  globalZFS,
+					ZFSCmds:   globalZFS,
-					Envconst: envconstReport,
+					Envconst:  envconstReport,
+					OsEnviron: os.Environ(),
 				}}
 			return s, nil
 		}})
@@ -156,8 +158,8 @@ func (j *controlJob) Run(ctx context.Context) {
 	server := http.Server{
 		Handler: mux,
 		// control socket is local, 1s timeout should be more than sufficient, even on a loaded system
-		WriteTimeout: 1 * time.Second,
+		WriteTimeout: envconst.Duration("ZREPL_DAEMON_CONTROL_SERVER_WRITE_TIMEOUT", 1*time.Second),
-		ReadTimeout:  1 * time.Second,
+		ReadTimeout:  envconst.Duration("ZREPL_DAEMON_CONTROL_SERVER_READ_TIMEOUT", 1*time.Second),
 	}
 
 outer:

daemon/daemon.go

@@ -160,8 +160,9 @@ type Status struct {
 }
 
 type GlobalStatus struct {
-	ZFSCmds  *zfscmd.Report
+	ZFSCmds   *zfscmd.Report
-	Envconst *envconst.Report
+	Envconst  *envconst.Report
+	OsEnviron []string
 }
 
 func (s *jobs) status() map[string]*job.Status {

daemon/job/active.go

@@ -42,6 +42,7 @@ type ActiveSide struct {
 	promPruneSecs         *prometheus.HistogramVec // labels: prune_side
 	promBytesReplicated   *prometheus.CounterVec   // labels: filesystem
 	promReplicationErrors prometheus.Gauge
+	promLastSuccessful    prometheus.Gauge
 
 	tasksMtx sync.Mutex
 	tasks    activeSideTasks
@@ -321,7 +322,6 @@ func activeSide(g *config.Global, in *config.ActiveJob, configJob interface{}) (
 		Help:        "number of bytes replicated from sender to receiver per filesystem",
 		ConstLabels: prometheus.Labels{"zrepl_job": j.name.String()},
 	}, []string{"filesystem"})
-
 	j.promReplicationErrors = prometheus.NewGauge(prometheus.GaugeOpts{
 		Namespace:   "zrepl",
 		Subsystem:   "replication",
@@ -329,6 +329,13 @@ func activeSide(g *config.Global, in *config.ActiveJob, configJob interface{}) (
 		Help:        "number of filesystems that failed replication in the latest replication attempt, or -1 if the job failed before enumerating the filesystems",
 		ConstLabels: prometheus.Labels{"zrepl_job": j.name.String()},
 	})
+	j.promLastSuccessful = prometheus.NewGauge(prometheus.GaugeOpts{
+		Namespace:   "zrepl",
+		Subsystem:   "replication",
+		Name:        "last_successful",
+		Help:        "timestamp of last successful replication",
+		ConstLabels: prometheus.Labels{"zrepl_job": j.name.String()},
+	})
 
 	j.connecter, err = fromconfig.ConnecterFromConfig(g, in.Connect)
 	if err != nil {
@@ -360,6 +367,7 @@ func (j *ActiveSide) RegisterMetrics(registerer prometheus.Registerer) {
 	registerer.MustRegister(j.promPruneSecs)
 	registerer.MustRegister(j.promBytesReplicated)
 	registerer.MustRegister(j.promReplicationErrors)
+	registerer.MustRegister(j.promLastSuccessful)
 }
 
 func (j *ActiveSide) Name() string { return j.name.String() }
@@ -494,7 +502,11 @@ func (j *ActiveSide) do(ctx context.Context) {
 		repCancel()   // always cancel to free up context resources
 
 		replicationReport := j.tasks.replicationReport()
-		j.promReplicationErrors.Set(float64(replicationReport.GetFailedFilesystemsCountInLatestAttempt()))
+		var numErrors = replicationReport.GetFailedFilesystemsCountInLatestAttempt()
+		j.promReplicationErrors.Set(float64(numErrors))
+		if numErrors == 0 {
+			j.promLastSuccessful.SetToCurrentTime()
+		}
 
 		endSpan()
 	}

daemon/job/build_jobs_sendrecvoptions.go

@@ -72,6 +72,16 @@ func buildReceiverConfig(in ReceivingJobConfig, jobID endpoint.JobID) (rc endpoi
 		return rc, errors.Wrap(err, "cannot build bandwith limit config")
 	}
 
+	placeholderEncryption, err := endpoint.PlaceholderCreationEncryptionPropertyString(recvOpts.Placeholder.Encryption)
+	if err != nil {
+		options := []string{}
+		for _, v := range endpoint.PlaceholderCreationEncryptionPropertyValues() {
+			options = append(options, endpoint.PlaceholderCreationEncryptionProperty(v).String())
+		}
+		return rc, errors.Errorf("placeholder encryption value %q is invalid, must be one of %s",
+			recvOpts.Placeholder.Encryption, options)
+	}
+
 	rc = endpoint.ReceiverConfig{
 		JobID:                      jobID,
 		RootWithoutClientComponent: rootFs,
@@ -81,6 +91,8 @@ func buildReceiverConfig(in ReceivingJobConfig, jobID endpoint.JobID) (rc endpoi
 		OverrideProperties: recvOpts.Properties.Override,
 
 		BandwidthLimit: bwlim,
+
+		PlaceholderEncryption: placeholderEncryption,
 	}
 	if err := rc.Validate(); err != nil {
 		return rc, errors.Wrap(err, "cannot build receiver config")

daemon/logging/trace/trace.go

@@ -246,7 +246,7 @@ func WithTask(ctx context.Context, taskName string) (context.Context, DoneFunc)
 					// the debugString can be quite long and panic won't print it completely
 					fmt.Fprintf(os.Stderr, "going to panic due to activeChildTasks:\n%s\n", this.debugString())
 				}
-				panic(errors.WithMessagef(ErrTaskStillHasActiveChildTasks, "end task: %v active child tasks\n", this.activeChildTasks))
+				panic(errors.WithMessagef(ErrTaskStillHasActiveChildTasks, "end task: %v active child tasks (run daemon with env var %s=1 for more details)\n", this.activeChildTasks, debugEnabledEnvVar))
 			}
 
 			// support idempotent task ends

daemon/logging/trace/trace_debug.go

@@ -7,7 +7,9 @@ import (
 	"github.com/zrepl/zrepl/util/envconst"
 )
 
-var debugEnabled = envconst.Bool("ZREPL_TRACE_DEBUG_ENABLED", false)
+const debugEnabledEnvVar = "ZREPL_TRACE_DEBUG_ENABLED"
+
+var debugEnabled = envconst.Bool(debugEnabledEnvVar, false)
 
 func debug(format string, args ...interface{}) {
 	if !debugEnabled {

docs/changelog.rst

@@ -16,18 +16,44 @@ Changelog
 The changelog summarizes bugfixes that are deemed relevant for users and package maintainers.
 Developers should consult the git commit log or GitHub issue tracker.
 
-We use the following annotations for classifying changes:
+0.5
+---
+
+* |feature| :ref:`Bandwidth limiting <job-send-recv-options--bandwidth-limit>` (Thanks, Prominic.NET, Inc.)
+* |feature| zrepl status: use a ``*`` to indicate which filesystem is currently replicating
+* |feature| include daemon environment variables in zrepl status (currently only in ``--raw``)
+* |bugfix| **fix encrypt-on-receive + placeholders use case** (:issue:`504`)
+
+  * Before this fix, **plain sends** to a receiver with an encrypted ``root_fs`` **could be received unencrypted** if zrepl needed to create placeholders on the receiver.
+  * Existing zrepl users should :ref:`read the docs <job-recv-options--placeholder>` and check ``zfs get -r encryption,zrepl:placeholder PATH_TO_ROOTFS`` on the receiver.
+  * Thanks to `@mologie <https://github.com/mologie>`_ and `@razielgn <https://github.com/razielgn>`_ for reporting and testing!
+
+* |bugfix| Rename mis-spelled :ref:`send option <job-send-options>` ``embbeded_data`` to ``embedded_data``.
+* |bugfix| zrepl status: replication step numbers should start at 1
+* |bugfix| incorrect bandwidth averaging in ``zrepl status``.
+* |bugfix| FreeBSD with OpenZFS 2.0: zrepl would wait indefinitely for zfs send to exit on timeouts.
+* |bugfix| fix ``strconv.ParseInt: value out of range`` bug (and use the control RPCs).
+* |docs| improve description of multiple pruning rules.
+* |docs| document :ref:`platform tests <usage-platform-tests>`.
+* |docs| quickstart: make users aware that prune rules apply to all snapshots.
+* |maint| some platformtests were broken.
+* |maint| FreeBSD: release armv7 and arm64 binaries.
+* |maint| apt repo: update instructions due to ``apt-key`` deprecation.
+
+Note to all users: please read up on the following OpenZFS bugs, as you might be affected:
+
+* `ZFS send/recv with ashift 9->12 leads to data corruption <https://github.com/openzfs/zfs/issues/12762>`_.
+* Various bugs with encrypted send/recv (`Leadership meeting notes <https://openzfs.topicbox.com/groups/developer/T24bdaa2886c6cbf5-Mc039a11c3f1507ea0664817b/december-openzfs-leadership-meeting>`_)
+
+Finally, I'd like to point you to the `GitHub discussion <https://github.com/zrepl/zrepl/discussions/547>`_ about which bugfixes and features should be prioritized in zrepl 0.6 and beyond!
+
+.. NOTE::
+  |  zrepl is a spare-time project primarily developed by `Christian Schwarz <https://cschwarz.com>`_.
+  |  You can support maintenance and feature development through one of the following services:
+  |  |Donate via Patreon| |Donate via GitHub Sponsors| |Donate via Liberapay| |Donate via PayPal|
+  |  Note that PayPal processing fees are relatively high for small donations.
+  |  For SEPA wire transfer and **commercial support**, please `contact Christian directly <https://cschwarz.com>`_.
 
-* |break_config| Change that breaks the config.
-  As a package maintainer, make sure to warn your users about config breakage somehow.
-* |break| Change that breaks interoperability or persistent state representation with previous releases.
-  As a package maintainer, make sure to warn your users about config breakage somehow.
-  Note that even updating the package on both sides might not be sufficient, e.g. if persistent state needs to be migrated to a new format.
-* |mig| Migration that must be run by the user.
-* |feature| Change that introduces new functionality.
-* |bugfix| Change that fixes a bug, no regressions or incompatibilities expected.
-* |docs| Change to the documentation.
-* |maint| Maintenance changes.
 
 0.4.0
 -----
@@ -53,13 +79,6 @@ The following bugfix in 0.3.1 :issue:`caused problems for some users <400>`:
 
 * |bugfix| pruning: ``grid``:  add all snapshots that do not match the regex to the rule's destroy list.
 
-.. NOTE::
-  |  zrepl is a spare-time project primarily developed by `Christian Schwarz <https://cschwarz.com>`_.
-  |  You can support maintenance and feature development through one of the following services:
-  |  |Donate via Patreon| |Donate via GitHub Sponsors| |Donate via Liberapay| |Donate via PayPal|
-  |  Note that PayPal processing fees are relatively high for small donations.
-  |  For SEPA wire transfer and **commercial support**, please `contact Christian directly <https://cschwarz.com>`_.
-
 0.3.1
 -----
 

docs/configuration/prune.rst

@@ -107,7 +107,7 @@ The following procedure happens during pruning:
 #. All subsequent buckets are placed adjacent to their predecessor bucket.
 #. Now each snapshot on the axis either falls into one bucket or it is older than our rightmost bucket.
    Buckets are left-inclusive and right-exclusive which means that a snapshot on the edge of bucket will always 'fall into the right one'.
-#. Snapshots older than the rightmost bucket **not kept** by this gridspec.
+#. Snapshots older than the rightmost bucket are **not kept** by the grid specification.
 #. For each bucket, we only keep the ``keep`` oldest snapshots.
 
 The syntax to describe the bucket list is as follows:
@@ -125,11 +125,11 @@ The syntax to describe the bucket list is as follows:
 
 ::
 
-   Assume the following grid spec:
+   Assume the following grid specification:
 
       grid: 1x1h(keep=all) | 2x2h | 1x3h
 
-   This grid spec produces the following constellation of buckets:
+   This grid specification produces the following constellation of buckets:
 
    0h        1h        2h        3h        4h        5h        6h        7h        8h        9h
    |         |         |         |         |         |         |         |         |         |
@@ -149,13 +149,13 @@ The syntax to describe the bucket list is as follows:
    |-Bucket1-|-----Bucket2-------|------Bucket3------|-----------Bucket4-----------|
    | keep=all|      keep=1       |       keep=1      |            keep=1           |
    |         |                   |                   |                             |
-   |  a  b  c| d  e  f  g  h  i  j  k  l  m  n  o  p |q  r  s  t  u  v  w  x  y  z |A  B  C  D
+   | a  b  c | d  e  f  g  h  i  j  k  l  m  n  o  p |q  r  s  t  u  v  w  x  y  z |A  B  C  D
 
-   The result is the following mapping of snapshots to buckets:
+   We obtain the following mapping of snapshots to buckets:
 
-   Bucket1:   a, b, c
+   Bucket1:   a,b,c
-   Bucket2:   d,e,f,g,h,i,j
+   Bucket2:   d,e,f,g,h,i
-   Bucket3:   k,l,m,n,o,p
+   Bucket3:   j,k,l,m,n,o,p
    Bucket4:   q,r,s,t,u,v,w,x,y,z
    No bucket: A,B,C,D
 
@@ -169,7 +169,7 @@ The syntax to describe the bucket list is as follows:
    |         |         |         |         |         |         |         |         |         |
    |-Bucket1-|-----Bucket2-------|------Bucket3------|-----------Bucket4-----------|
    |         |                   |                   |                             |
-   |  a  b  c|                   j                 p |                           z  |
+   | a  b  c |                i  |                 p |                           z |
 
 .. _prune-keep-last-n:
 

docs/configuration/sendrecvoptions.rst

@@ -38,10 +38,10 @@ See the `upstream man page <https://openzfs.github.io/openzfs-docs/man/8/zfs-sen
       - Specific to zrepl, :ref:`see below <job-send-options-encrypted>`.
     * - ``bandwidth_limit``
       -
-      - Specific to zrepl, :ref:`see below <job-send-recv-options-bandwidth-limit>`.
+      - Specific to zrepl, :ref:`see below <job-send-recv-options--bandwidth-limit>`.
     * - ``raw``
       - ``-w``
-      - Use ``encrypted`` to only allow encrypted sends.
+      - Use ``encrypted`` to only allow encrypted sends. Mixed sends are not supported.
     * - ``send_properties``
       - ``-p``
       - **Be careful**, read the :ref:`note on property replication below <job-note-property-replication>`.
@@ -54,7 +54,7 @@ See the `upstream man page <https://openzfs.github.io/openzfs-docs/man/8/zfs-sen
     * - ``compressed``
       - ``-c``
       -
-    * - ``embbeded_data``
+    * - ``embedded_data``
       - ``-e``
       -
     * - ``saved``
@@ -141,9 +141,16 @@ Recv Options
          override: {
            "org.openzfs.systemd:ignore": "on"
          }
-       bandwidth_limit: ... # see below
+       bandwidth_limit: ...
+       placeholder:
+         encryption: unspecified | off | inherit
      ...
 
+Jump to
+:ref:`properties <job-recv-options--inherit-and-override>` ,
+:ref:`bandwidth_limit <job-send-recv-options--bandwidth-limit>` , and
+:ref:`bandwidth_limit <job-recv-options--placeholder>`.
+
 .. _job-recv-options--inherit-and-override:
 
 ``properties``
@@ -217,10 +224,40 @@ and property replication is enabled, the receiver must :ref:`inherit the followi
 * ``keyformat``
 * ``encryption``
 
+.. _job-recv-options--placeholder:
+
+Placeholders
+------------
+
+During replication, zrepl :ref:`creates placeholder datasets <replication-placeholder-property>` on the receiving side if the sending side's ``filesystems`` filter creates gaps in the dataset hierarchy.
+This is generally fully transparent to the user.
+However, with OpenZFS Native Encryption, placeholders require zrepl user attention.
+Specifically, the problem is that, when zrepl attempts to create the placeholder dataset on the receiver, and that placeholder's parent dataset is encrypted, ZFS wants to inherit encryption to the placeholder.
+This is relevant to two use cases that zrepl supports:
+
+1. **encrypted-send-to-untrusted-receiver** In this use case, the sender sends an :ref:`encrypted send stream <job-send-options-encrypted>` and the receiver doesn't have the key loaded.
+2. **send-plain-encrypt-on-receive** The receive-side ``root_fs`` dataset is encrypted, and the senders are unencrypted.
+   The key of ``root_fs`` is loaded, and the goal is that the plain sends (e.g., from production) are encrypted on-the-fly during receive, with ``root_fs``'s key.
+
+For **encrypted-send-to-untrusted-receiver**, the placeholder datasets need to be created with ``-o encryption=off``.
+Without it, creation would fail with an error, indicating that the placeholder's parent dataset's key needs to be loaded.
+But we don't trust the receiver, so we can't expect that to ever happen.
+
+However, for **send-plain-encrypt-on-receive**, we cannot set ``-o encryption=off``.
+The reason is that if we did, any of the (non-placeholder) child datasets below the placeholder would inherit ``encryption=off``, thereby silently breaking our encrypt-on-receive use case.
+So, to cover this use case, we need to create placeholders without specifying ``-o encryption``.
+This will make ``zfs create`` inherit the encryption mode from the parent dataset, and thereby transitively from ``root_fs``.
+
+The zrepl config provides the `recv.placeholder.encryption` knob to control this behavior.
+In ``undefined`` mode (default), placeholder creation bails out and asks the user to configure a behavior.
+In ``off`` mode, the placeholder is created with ``encryption=off``, i.e., **encrypted-send-to-untrusted-rceiver** use case.
+In ``inherit`` mode, the placeholder is created without specifying ``-o encryption`` at all, i.e., the **send-plain-encrypt-on-receive** use case.
+
+
 Common Options
 ~~~~~~~~~~~~~~
 
-.. _job-send-recv-options-bandwidth-limit:
+.. _job-send-recv-options--bandwidth-limit:
 
 Bandwidth Limit (send & recv)
 -----------------------------

docs/index.rst

@@ -25,10 +25,10 @@ zrepl - ZFS replication
          Progress: [=========================\----] 246.7 MiB / 264.7 MiB @ 11.5 MiB/s
            zroot              STEPPING (step 1/2, 624 B/1.2 KiB) next: @a => @b
            zroot/ROOT         DONE (step 2/2, 1.2 KiB/1.2 KiB)
-           zroot/ROOT/default STEPPING (step 1/2, 123.4 MiB/129.3 MiB) next: @a => @b
+         * zroot/ROOT/default STEPPING (step 1/2, 123.4 MiB/129.3 MiB) next: @a => @b
            zroot/tmp          STEPPING (step 1/2, 29.9 KiB/44.2 KiB) next: @a => @b
            zroot/usr          STEPPING (step 1/2, 624 B/1.2 KiB) next: @a => @b
-           zroot/usr/home     STEPPING (step 1/2, 123.3 MiB/135.3 MiB) next: @a => @b
+         * zroot/usr/home     STEPPING (step 1/2, 123.3 MiB/135.3 MiB) next: @a => @b
            zroot/var          STEPPING (step 1/2, 624 B/1.2 KiB) next: @a => @b
            zroot/var/audit    DONE (step 2/2, 1.2 KiB/1.2 KiB)
            zroot/var/crash    DONE (step 2/2, 1.2 KiB/1.2 KiB)
@@ -66,6 +66,7 @@ Main Features
     * [x] Large blocks send & receive
     * [x] Embedded data send & receive
     * [x] Resume state send & receive
+    * [x] Bandwidth limiting
 
 * **Automatic snapshot management**
 

docs/installation/apt-repos.rst

@@ -9,18 +9,29 @@ The fingerprint of the signing key is ``E101 418F D3D6 FBCB 9D65  A62D 7086 99FC
 It is available at `<https://zrepl.cschwarz.com/apt/apt-key.asc>`_ .
 Please open an issue in on GitHub if you encounter any issues with the repository.
 
-The following snippet configure the repository for your Debian or Ubuntu release:
-
 ::
 
-    sudo apt update && sudo apt install curl gnupg lsb-release; \
+    (
-    ARCH="$(dpkg --print-architecture)"; \
+    set -ex
-    CODENAME="$(lsb_release -i -s | tr '[:upper:]' '[:lower:]') $(lsb_release -c -s | tr '[:upper:]' '[:lower:]')"; \
+    zrepl_apt_key_url=https://zrepl.cschwarz.com/apt/apt-key.asc
-    echo "Using Distro and Codename: $CODENAME"; \
+    zrepl_apt_key_dst=/usr/share/keyrings/zrepl.gpg
-    (curl https://zrepl.cschwarz.com/apt/apt-key.asc | sudo apt-key add -) && \
+    zrepl_apt_repo_file=/etc/apt/sources.list.d/zrepl.list
-    (echo "deb [arch=$ARCH] https://zrepl.cschwarz.com/apt/$CODENAME main" | sudo tee /etc/apt/sources.list.d/zrepl.list) && \
-    sudo apt update
 
+    # Install dependencies for subsequent commands
+    sudo apt update && sudo apt install curl gnupg lsb-release
+
+    # Deploy the zrepl apt key.
+    curl -fsSL "$zrepl_apt_key_url" | tee | gpg --dearmor | sudo tee "$zrepl_apt_key_dst" > /dev/null
+
+    # Add the zrepl apt repo.
+    ARCH="$(dpkg --print-architecture)"
+    CODENAME="$(lsb_release -i -s | tr '[:upper:]' '[:lower:]') $(lsb_release -c -s | tr '[:upper:]' '[:lower:]')"
+    echo "Using Distro and Codename: $CODENAME"
+    echo "deb [arch=$ARCH signed-by=$zrepl_apt_key_dst] https://zrepl.cschwarz.com/apt/$CODENAME main" | sudo tee /etc/apt/sources.list.d/zrepl.list
+
+    # Update apt repos.
+    sudo apt update
+    )
 
 .. NOTE::
 

docs/installation/freebsd-jail-with-iocage.rst

@@ -93,6 +93,7 @@ Enable the zrepl daemon to start automatically at boot:
 
    sysrc zrepl_enable="YES"
 
+Now jump to :ref:`the summary <installation-freebsd-jail-summary>` below.
 
 Plugin
 ######
@@ -134,7 +135,18 @@ Now ``zrepl`` can be started.
 
    service zrepl start
 
+Now jump to :ref:`the summary <installation-freebsd-jail-summary>` below.
+
+.. _installation-freebsd-jail-summary:
+
 Summary
 -------
 
 Congratulations, you have a working jail!
+
+.. NOTE::
+
+   With FreeBSD 13's transition to OpenZFS 2.0, please ensure that your jail's FreeBSD version matches the one in the kernel module.
+   If you are getting cryptic errors such as
+   ``cannot receive new filesystem stream: invalid backup stream``
+   the instructions posted `here <https://github.com/zrepl/zrepl/issues/500#issuecomment-966215205>`_ might help.

docs/quickstart.rst

@@ -51,6 +51,19 @@ We hope that you have found a configuration that fits your use case.
 Use ``zrepl configcheck`` once again to make sure the config is correct (output indicates that everything is fine).
 Then restart the zrepl daemon on all systems involved in the replication, likely using ``service zrepl restart`` or ``systemctl restart zrepl``.
 
+.. WARNING::
+
+   Please :ref:`read up carefully <prune>` on the pruning rules before applying the config.
+   In particular, note that most example configs apply to all snapshots, not just zrepl-created snapshots.
+   Use the following keep rule on sender and receiver to prevent this:
+
+   ::
+      
+      - type: regex
+        negate: true
+        regex: "^zrepl_.*" # <- the 'prefix' specified in snapshotting.prefix
+
+
 Watch it Work
 =============
 

docs/usage.rst

@@ -77,4 +77,74 @@ Systemd Unit File
 
 A systemd service definition template is available in :repomasterlink:`dist/systemd`.
 Note that some of the options only work on recent versions of systemd.
-Any help & improvements are very welcome, see :issue:`145`.
+Any help & improvements are very welcome, see :issue:`145`.
+
+
+
+============
+Ops Runbooks
+============
+
+
+.. toctree::
+
+   usage/runbooks/migrating_sending_side_to_new_zpool.rst
+
+
+.. _usage-platform-tests:
+
+==============
+Platform Tests
+==============
+
+Along with the main ``zrepl`` binary, we release the ``platformtest`` binaries.
+The zrepl platform tests are an integration test suite that is complementary to the pure Go unit tests.
+Any test that needs to interact with ZFS is a platform test.
+
+The platform need to run as root.
+For each test, we create a fresh dummy zpool backed by a file-based vdev.
+The file path, and a root mountpoint for the dummy zpool, must be specified on the command line:
+
+::
+
+   mkdir -p /tmp/zreplplatformtest
+   ./platformtest \
+       -poolname 'zreplplatformtest' \  # <- name must contain zreplplatformtest
+       -imagepath /tmp/zreplplatformtest.img \ # <- zrepl will create the file
+       -mountpoint /tmp/zreplplatformtest # <- must exist
+
+
+.. WARNING::
+
+   ``platformtest`` will unconditionally overwrite the file at `imagepath`
+   and unconditionally ``zpool destroy $poolname``.
+   So, don't use a production poolname, and consider running the test in a VM.
+   It'll be a lot faster as well because the underlying operations, ``zfs list`` in particular, will be faster.
+
+
+While the platformtests are running, there will be a log of log output.
+After all tests have run, it prints a summary with a list of tests, grouped by result type (success, failure, skipped):
+
+::
+
+   PASSING TESTS:
+     github.com/zrepl/zrepl/platformtest/tests.BatchDestroy
+     github.com/zrepl/zrepl/platformtest/tests.CreateReplicationCursor
+     github.com/zrepl/zrepl/platformtest/tests.GetNonexistent
+     github.com/zrepl/zrepl/platformtest/tests.HoldsWork
+     ...
+     github.com/zrepl/zrepl/platformtest/tests.SendStreamNonEOFReadErrorHandling
+     github.com/zrepl/zrepl/platformtest/tests.UndestroyableSnapshotParsing
+   SKIPPED TESTS:
+     github.com/zrepl/zrepl/platformtest/tests.SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden__EncryptionSupported_false
+   FAILED TESTS: []
+
+
+If there is a failure, or a skipped test that you believe should be passing, re-run the test suite, capture stderr & stdout to a text file, and create an issue on GitHub.
+
+To run a specific test case, or a subset of tests matched by regex, use the ``-run REGEX`` command line flag.
+
+To stop test execution at the first failing test, and prevent cleanup of the dummy zpool, use the ``-failure.stop-and-keep-pool`` flag.
+
+To build the platformtests yourself, use ``make test-platform-bin``.
+There's also the ``make test-platform`` target to run the platform tests with a default command line.

docs/usage/runbooks/migrating_sending_side_to_new_zpool.rst

@@ -0,0 +1,36 @@
+
+Migrating Sending Side
+~~~~~~~~~~~~~~~~~~~~~~
+
+**Objective**:
+Move sending-side zpool to new hardware.
+Make the move fully transparent to the sending-side jobs.
+After the move is done, all sending-side zrepl jobs should continue to work as if the move had not happened.
+In particular, incremental replication should be able to pick up where it left before the move.
+
+Suppose we want to migrate all data from one zpool ``oldpool`` to another zpool ``newpool``.
+A possible reason might be that we want to change RAID levels, ``ashift``, or just migrate over to next-gen hardware.
+
+If the pool names are different, zrepl's matching between sender and receiver dataset will break becase the receive-side dataset names contain ``oldpool``.
+To avoid this, we will need the name of the new pool to match that of the old pool.
+The following steps will accomplish this:
+
+1. Stop zrepl.
+2. Create the new pool: ``zpool create newpool ...``
+3. Take a snapshot of the old pool so that you have something that you can ``zfs send``.
+   For example, run ``zfs snapshot -r oldpool@migration_oldpool_newpool``.
+4. Send all of the oldpool's datasets to the new pool:
+   ``zfs send -R oldpool@migration_oldpool_newpool | zfs recv -F newpool``
+5. Export the old pool: ``zpool export oldpool``
+6. Export the new pool: ``zpool export newpool``
+7. (Optional) Change the name of the old pool to something that does not conflict with the new pool.
+   We are going to use the name ``oldoldpool`` in this example.
+   Use ``zpool import`` with no arguments to see the pool id.
+   Then ``zpool import <id> oldoldpool && zpool export oldoldpool``.
+8. Import the new pool, while changing the name to match the old pool: ``zpool import newpool oldpool``
+9. Start zrepl again and wake up the relevant jobs.
+10. Use ``zrepl status`` or you monitoring to ensure that replication works.
+    The best test is an end-to-end test where you write some junk data on a sender dataset and wait until a snapshot with that data appears on the receiving side.
+11. Once you are confident that replication is working, you may dispose of the old pool.
+
+Note that, depending on pruning rules, it will not be possible to switch back to the old pool seamlessly, i.e., without a full re-replication.

endpoint/endpoint.go

@@ -472,8 +472,20 @@ type ReceiverConfig struct {
 	OverrideProperties map[zfsprop.Property]string
 
 	BandwidthLimit bandwidthlimit.Config
+
+	PlaceholderEncryption PlaceholderCreationEncryptionProperty
 }
 
+//go:generate enumer -type=PlaceholderCreationEncryptionProperty -transform=kebab -trimprefix=PlaceholderCreationEncryptionProperty
+type PlaceholderCreationEncryptionProperty int
+
+// Note: the constant names, transformed through enumer, are part of the config format!
+const (
+	PlaceholderCreationEncryptionPropertyUnspecified PlaceholderCreationEncryptionProperty = 1 << iota
+	PlaceholderCreationEncryptionPropertyInherit
+	PlaceholderCreationEncryptionPropertyOff
+)
+
 func (c *ReceiverConfig) copyIn() {
 	c.RootWithoutClientComponent = c.RootWithoutClientComponent.Copy()
 
@@ -513,6 +525,10 @@ func (c *ReceiverConfig) Validate() error {
 		return errors.Wrap(err, "`BandwidthLimit` field invalid")
 	}
 
+	if !c.PlaceholderEncryption.IsAPlaceholderCreationEncryptionProperty() {
+		return errors.Errorf("`PlaceholderEncryption` field is invalid")
+	}
+
 	return nil
 }
 
@@ -525,6 +541,8 @@ type Receiver struct {
 	bwLimit bandwidthlimit.Wrapper
 
 	recvParentCreationMtx *chainlock.L
+
+	Test_OverrideClientIdentityFunc func() string // for use by platformtest
 }
 
 func NewReceiver(config ReceiverConfig) *Receiver {
@@ -562,9 +580,15 @@ func (s *Receiver) clientRootFromCtx(ctx context.Context) *zfs.DatasetPath {
 		return s.conf.RootWithoutClientComponent.Copy()
 	}
 
-	clientIdentity, ok := ctx.Value(ClientIdentityKey).(string)
+	var clientIdentity string
-	if !ok {
+	if s.Test_OverrideClientIdentityFunc != nil {
-		panic("ClientIdentityKey context value must be set")
+		clientIdentity = s.Test_OverrideClientIdentityFunc()
+	} else {
+		var ok bool
+		clientIdentity, ok = ctx.Value(ClientIdentityKey).(string) // no shadow
+		if !ok {
+			panic("ClientIdentityKey context value must be set")
+		}
 	}
 
 	clientRoot, err := clientRoot(s.conf.RootWithoutClientComponent, clientIdentity)
@@ -605,7 +629,8 @@ func (s *Receiver) ListFilesystems(ctx context.Context, req *pdu.ListFilesystemR
 	if rphs, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, s.conf.RootWithoutClientComponent); err != nil {
 		return nil, errors.Wrap(err, "cannot determine whether root_fs exists")
 	} else if !rphs.FSExists {
-		return nil, errors.New("root_fs does not exist")
+		getLogger(ctx).WithField("root_fs", s.conf.RootWithoutClientComponent).Error("root_fs does not exist")
+		return nil, errors.Errorf("root_fs does not exist")
 	}
 
 	root := s.clientRootFromCtx(ctx)
@@ -714,6 +739,33 @@ func (s *Receiver) SendDry(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, e
 	return nil, fmt.Errorf("receiver does not implement SendDry()")
 }
 
+func (s *Receiver) receive_GetPlaceholderCreationEncryptionValue(client_root, path *zfs.DatasetPath) (zfs.FilesystemPlaceholderCreateEncryptionValue, error) {
+	if !s.conf.PlaceholderEncryption.IsAPlaceholderCreationEncryptionProperty() {
+		panic(s.conf.PlaceholderEncryption)
+	}
+
+	if client_root.Equal(path) && s.conf.PlaceholderEncryption == PlaceholderCreationEncryptionPropertyUnspecified {
+		// If our Receiver is configured to append a client component to s.conf.RootWithoutClientComponent
+		// then that dataset is always going to be a placeholder.
+		// We don't want to burden users with the concept of placeholders if their `filesystems` filter on the sender
+		// doesn't introduce any gaps.
+		// Since the dataset hierarchy up to and including that client component dataset is still fully controlled by us,
+		// using `inherit` is going to make it work in all expected use cases.
+		return zfs.FilesystemPlaceholderCreateEncryptionInherit, nil
+	}
+
+	switch s.conf.PlaceholderEncryption {
+	case PlaceholderCreationEncryptionPropertyUnspecified:
+		return 0, fmt.Errorf("placeholder filesystem encryption handling is unspecified in receiver config")
+	case PlaceholderCreationEncryptionPropertyInherit:
+		return zfs.FilesystemPlaceholderCreateEncryptionInherit, nil
+	case PlaceholderCreationEncryptionPropertyOff:
+		return zfs.FilesystemPlaceholderCreateEncryptionOff, nil
+	default:
+		panic(s.conf.PlaceholderEncryption)
+	}
+}
+
 func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.ReadCloser) (*pdu.ReceiveRes, error) {
 	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 
@@ -754,15 +806,18 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.
 			if v.Path.Equal(lp) {
 				return false
 			}
+
+			l := getLogger(ctx).
+				WithField("placeholder_fs", v.Path.ToString()).
+				WithField("receive_fs", lp.ToString())
+
 			ph, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, v.Path)
-			getLogger(ctx).
+			l.WithField("placeholder_state", fmt.Sprintf("%#v", ph)).
-				WithField("fs", v.Path.ToString()).
-				WithField("placeholder_state", fmt.Sprintf("%#v", ph)).
 				WithField("err", fmt.Sprintf("%s", err)).
 				WithField("errType", fmt.Sprintf("%T", err)).
-				Debug("placeholder state for filesystem")
+				Debug("get placeholder state for filesystem")
 			if err != nil {
-				visitErr = err
+				visitErr = errors.Wrapf(err, "cannot get placeholder state of %s", v.Path.ToString())
 				return false
 			}
 
@@ -773,21 +828,34 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.
 					} else {
 						visitErr = fmt.Errorf("root_fs %q does not exist", s.conf.RootWithoutClientComponent.ToString())
 					}
-					getLogger(ctx).WithError(visitErr).Error("placeholders are only created automatically below root_fs")
+					l.WithError(visitErr).Error("placeholders are only created automatically below root_fs")
 					return false
 				}
-				l := getLogger(ctx).WithField("placeholder_fs", v.Path)
+
-				l.Debug("create placeholder filesystem")
+				// compute the value lazily so that users who don't rely on
-				err := zfs.ZFSCreatePlaceholderFilesystem(ctx, v.Path, v.Parent.Path)
+				// placeholders can use the default value PlaceholderCreationEncryptionPropertyUnspecified
+				placeholderEncryption, err := s.receive_GetPlaceholderCreationEncryptionValue(root, v.Path)
 				if err != nil {
-					l.WithError(err).Error("cannot create placeholder filesystem")
+					l.WithError(err).Error("cannot create placeholder filesystem") // logger already contains path
-					visitErr = err
+					visitErr = errors.Wrapf(err, "cannot create placeholder filesystem %s", v.Path.ToString())
 					return false
 				}
+
+				l := l.WithField("encryption", placeholderEncryption)
+
+				l.Debug("creating placeholder filesystem")
+				err = zfs.ZFSCreatePlaceholderFilesystem(ctx, v.Path, v.Parent.Path, placeholderEncryption)
+				if err != nil {
+					l.WithError(err).Error("cannot create placeholder filesystem") // logger already contains path
+					visitErr = errors.Wrapf(err, "cannot create placeholder filesystem %s", v.Path.ToString())
+					return false
+				}
+				l.Info("created placeholder filesystem")
 				return true
+			} else {
+				l.Debug("filesystem exists")
+				return true // leave this fs as is
 			}
-			getLogger(ctx).WithField("filesystem", v.Path.ToString()).Debug("exists")
-			return true // leave this fs as is
 		})
 	}()
 	getLogger(ctx).WithField("visitErr", visitErr).Debug("complete tree-walk")

endpoint/placeholdercreationencryptionproperty_enumer.go

@@ -0,0 +1,62 @@
+// Code generated by "enumer -type=PlaceholderCreationEncryptionProperty -transform=kebab -trimprefix=PlaceholderCreationEncryptionProperty"; DO NOT EDIT.
+
+//
+package endpoint
+
+import (
+	"fmt"
+)
+
+const (
+	_PlaceholderCreationEncryptionPropertyName_0 = "unspecifiedinherit"
+	_PlaceholderCreationEncryptionPropertyName_1 = "off"
+)
+
+var (
+	_PlaceholderCreationEncryptionPropertyIndex_0 = [...]uint8{0, 11, 18}
+	_PlaceholderCreationEncryptionPropertyIndex_1 = [...]uint8{0, 3}
+)
+
+func (i PlaceholderCreationEncryptionProperty) String() string {
+	switch {
+	case 1 <= i && i <= 2:
+		i -= 1
+		return _PlaceholderCreationEncryptionPropertyName_0[_PlaceholderCreationEncryptionPropertyIndex_0[i]:_PlaceholderCreationEncryptionPropertyIndex_0[i+1]]
+	case i == 4:
+		return _PlaceholderCreationEncryptionPropertyName_1
+	default:
+		return fmt.Sprintf("PlaceholderCreationEncryptionProperty(%d)", i)
+	}
+}
+
+var _PlaceholderCreationEncryptionPropertyValues = []PlaceholderCreationEncryptionProperty{1, 2, 4}
+
+var _PlaceholderCreationEncryptionPropertyNameToValueMap = map[string]PlaceholderCreationEncryptionProperty{
+	_PlaceholderCreationEncryptionPropertyName_0[0:11]:  1,
+	_PlaceholderCreationEncryptionPropertyName_0[11:18]: 2,
+	_PlaceholderCreationEncryptionPropertyName_1[0:3]:   4,
+}
+
+// PlaceholderCreationEncryptionPropertyString retrieves an enum value from the enum constants string name.
+// Throws an error if the param is not part of the enum.
+func PlaceholderCreationEncryptionPropertyString(s string) (PlaceholderCreationEncryptionProperty, error) {
+	if val, ok := _PlaceholderCreationEncryptionPropertyNameToValueMap[s]; ok {
+		return val, nil
+	}
+	return 0, fmt.Errorf("%s does not belong to PlaceholderCreationEncryptionProperty values", s)
+}
+
+// PlaceholderCreationEncryptionPropertyValues returns all values of the enum
+func PlaceholderCreationEncryptionPropertyValues() []PlaceholderCreationEncryptionProperty {
+	return _PlaceholderCreationEncryptionPropertyValues
+}
+
+// IsAPlaceholderCreationEncryptionProperty returns "true" if the value is listed in the enum definition. "false" otherwise
+func (i PlaceholderCreationEncryptionProperty) IsAPlaceholderCreationEncryptionProperty() bool {
+	for _, v := range _PlaceholderCreationEncryptionPropertyValues {
+		if i == v {
+			return true
+		}
+	}
+	return false
+}

platformtest/harness/harness.go

@@ -37,6 +37,7 @@ func main() {
 	flag.StringVar(&args.CreateArgs.Mountpoint, "mountpoint", "", "")
 	flag.BoolVar(&args.StopAndKeepPoolOnFail, "failure.stop-and-keep-pool", false, "if a test case fails, stop test execution and keep pool as it was when the test failed")
 	flag.StringVar(&args.Run, "run", "", "")
+	flag.DurationVar(&platformtest.ZpoolExportTimeout, "zfs.zpool-export-timeout", platformtest.ZpoolExportTimeout, "")
 	flag.Parse()
 
 	if err := HarnessRun(args); err != nil {

platformtest/platformtest_zpool.go

@@ -5,12 +5,17 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"runtime"
+	"strings"
+	"time"
 
 	"github.com/pkg/errors"
 
 	"github.com/zrepl/zrepl/zfs"
 )
 
+var ZpoolExportTimeout time.Duration = 500 * time.Millisecond
+
 type Zpool struct {
 	args ZpoolCreateArgs
 }
@@ -93,8 +98,23 @@ func (p *Zpool) Name() string { return p.args.PoolName }
 
 func (p *Zpool) Destroy(ctx context.Context, e Execer) error {
 
-	if err := e.RunExpectSuccessNoOutput(ctx, "zpool", "export", p.args.PoolName); err != nil {
+	exportDeadline := time.Now().Add(ZpoolExportTimeout)
-		return errors.Wrapf(err, "export pool %q", p.args.PoolName)
+
+	for {
+		if time.Now().After(exportDeadline) {
+			return errors.Errorf("could not zpool export (got 'pool is busy'): %s", p.args.PoolName)
+		}
+		err := e.RunExpectSuccessNoOutput(ctx, "zpool", "export", p.args.PoolName)
+		if err == nil {
+			break
+		}
+		if strings.Contains(err.Error(), "pool is busy") {
+			runtime.Gosched()
+			continue
+		}
+		if err != nil {
+			return errors.Wrapf(err, "export pool %q", p.args.PoolName)
+		}
 	}
 
 	if err := os.Remove(p.args.ImagePath); err != nil {

platformtest/tests/generated_cases.go

@@ -24,6 +24,9 @@ var Cases = []Case{BatchDestroy,
 	ReplicationIsResumableFullSend__both_GuaranteeResumability,
 	ReplicationIsResumableFullSend__initial_GuaranteeIncrementalReplication_incremental_GuaranteeIncrementalReplication,
 	ReplicationIsResumableFullSend__initial_GuaranteeResumability_incremental_GuaranteeIncrementalReplication,
+	ReplicationPlaceholderEncryption__EncryptOnReceiverUseCase__WorksIfConfiguredWithInherit,
+	ReplicationPlaceholderEncryption__UnspecifiedIsOkForClientIdentityPlaceholder,
+	ReplicationPlaceholderEncryption__UnspecifiedLeadsToFailureAtRuntimeWhenCreatingPlaceholders,
 	ReplicationPropertyReplicationWorks,
 	ReplicationReceiverErrorWhileStillSending,
 	ReplicationStepCompletedLostBehavior__GuaranteeIncrementalReplication,

platformtest/tests/helpers.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	"github.com/zrepl/zrepl/daemon/filters"
 	"github.com/zrepl/zrepl/platformtest"
 	"github.com/zrepl/zrepl/util/limitio"
 	"github.com/zrepl/zrepl/zfs"
@@ -174,3 +175,8 @@ func datasetToStringSortedTrimPrefix(prefix *zfs.DatasetPath, paths []*zfs.Datas
 	sort.Strings(pstrs)
 	return pstrs
 }
+
+func mustAddToSFilter(ctx *platformtest.Context, f *filters.DatasetMapFilter, fs string) {
+	err := f.Add(fs, "ok")
+	require.NoError(ctx, err)
+}

platformtest/tests/replication.go

@@ -79,6 +79,7 @@ func (i replicationInvocation) Do(ctx *platformtest.Context) *report.Report {
 		AppendClientIdentity:       false,
 		RootWithoutClientComponent: mustDatasetPath(i.rfsRoot),
 		BandwidthLimit:             bandwidthlimit.NoLimitConfig(),
+		PlaceholderEncryption:      endpoint.PlaceholderCreationEncryptionPropertyUnspecified,
 	}
 	if i.receiverConfigHook != nil {
 		i.receiverConfigHook(&receiverConfig)
@@ -903,13 +904,9 @@ func ReplicationFailingInitialParentProhibitsChildReplication(ctx *platformtest.
 	fsAA := ctx.RootDataset + "/sender/aa"
 
 	sfilter := filters.NewDatasetMapFilter(3, true)
-	mustAddToSFilter := func(fs string) {
+	mustAddToSFilter(ctx, sfilter, fsA)
-		err := sfilter.Add(fs, "ok")
+	mustAddToSFilter(ctx, sfilter, fsAChild)
-		require.NoError(ctx, err)
+	mustAddToSFilter(ctx, sfilter, fsAA)
-	}
-	mustAddToSFilter(fsA)
-	mustAddToSFilter(fsAChild)
-	mustAddToSFilter(fsAA)
 	rfsRoot := ctx.RootDataset + "/receiver"
 
 	mockRecvErr := fmt.Errorf("yifae4ohPhaquaes0hohghiep9oufie4roo7quoWooluaj2ee8")
@@ -965,6 +962,7 @@ func ReplicationPropertyReplicationWorks(ctx *platformtest.Context) {
 		+  "sender/a/child"
 		+  "sender/a/child@1"
 		+  "receiver"
+		R  zfs create -p "${ROOTDS}/receiver/${ROOTDS}/sender"
 	`)
 
 	sjid := endpoint.MustMakeJobID("sender-job")
@@ -974,12 +972,8 @@ func ReplicationPropertyReplicationWorks(ctx *platformtest.Context) {
 	fsAChild := ctx.RootDataset + "/sender/a/child"
 
 	sfilter := filters.NewDatasetMapFilter(2, true)
-	mustAddToSFilter := func(fs string) {
+	mustAddToSFilter(ctx, sfilter, fsA)
-		err := sfilter.Add(fs, "ok")
+	mustAddToSFilter(ctx, sfilter, fsAChild)
-		require.NoError(ctx, err)
-	}
-	mustAddToSFilter(fsA)
-	mustAddToSFilter(fsAChild)
 	rfsRoot := ctx.RootDataset + "/receiver"
 
 	type testPropExpectation struct {
@@ -1107,3 +1101,208 @@ func ReplicationPropertyReplicationWorks(ctx *platformtest.Context) {
 		}
 	}
 }
+
+func ReplicationPlaceholderEncryption__UnspecifiedLeadsToFailureAtRuntimeWhenCreatingPlaceholders(ctx *platformtest.Context) {
+
+	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+		CREATEROOT
+		+  "sender"
+		+  "sender/a"
+		+  "sender/a/child"
+		+  "receiver"
+		R  zfs snapshot -r ${ROOTDS}/sender@initial
+	`)
+
+	sjid := endpoint.MustMakeJobID("sender-job")
+	rjid := endpoint.MustMakeJobID("receiver-job")
+
+	childfs := ctx.RootDataset + "/sender/a/child"
+
+	sfilter := filters.NewDatasetMapFilter(3, true)
+
+	mustAddToSFilter(ctx, sfilter, childfs)
+	rfsRoot := ctx.RootDataset + "/receiver"
+
+	rep := replicationInvocation{
+		sjid:      sjid,
+		rjid:      rjid,
+		sfilter:   sfilter,
+		rfsRoot:   rfsRoot,
+		guarantee: pdu.ReplicationConfigProtectionWithKind(pdu.ReplicationGuaranteeKind_GuaranteeResumability),
+		receiverConfigHook: func(rc *endpoint.ReceiverConfig) {
+			rc.PlaceholderEncryption = endpoint.PlaceholderCreationEncryptionPropertyUnspecified
+		},
+	}
+
+	r := rep.Do(ctx)
+	ctx.Logf("\n%s", pretty.Sprint(r))
+
+	require.Len(ctx, r.Attempts, 1)
+	attempt := r.Attempts[0]
+	require.Nil(ctx, attempt.PlanError)
+	require.Len(ctx, attempt.Filesystems, 1)
+
+	afs := attempt.Filesystems[0]
+	require.Equal(ctx, childfs, afs.Info.Name)
+
+	require.Equal(ctx, 1, len(afs.Steps))
+	require.Equal(ctx, 0, afs.CurrentStep)
+
+	require.Equal(ctx, report.FilesystemSteppingErrored, afs.State)
+
+	childfsFirstComponent := strings.Split(childfs, "/")[0]
+	require.Contains(ctx, afs.StepError.Err, "cannot create placeholder filesystem "+rfsRoot+"/"+childfsFirstComponent+": placeholder filesystem encryption handling is unspecified in receiver config")
+}
+
+type ClientIdentityReceiver struct {
+	clientIdentity string
+	*endpoint.Receiver
+}
+
+func (r *ClientIdentityReceiver) Receive(ctx context.Context, req *pdu.ReceiveReq, stream io.ReadCloser) (*pdu.ReceiveRes, error) {
+	ctx = context.WithValue(ctx, endpoint.ClientIdentityKey, r.clientIdentity)
+	return r.Receiver.Receive(ctx, req, stream)
+}
+
+func ReplicationPlaceholderEncryption__UnspecifiedIsOkForClientIdentityPlaceholder(ctx *platformtest.Context) {
+	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+		CREATEROOT
+		+  "receiver"
+	`)
+
+	sjid := endpoint.MustMakeJobID("sender-job")
+	rjid := endpoint.MustMakeJobID("receiver-job")
+
+	sfilter := filters.NewDatasetMapFilter(1, true)
+
+	// hacky...
+	comps := strings.Split(ctx.RootDataset, "/")
+	require.GreaterOrEqual(ctx, len(comps), 2)
+	pool := comps[0]
+	require.Contains(ctx, pool, "zreplplatformtest", "don't want to cause accidents")
+	poolchild := pool + "/" + comps[1]
+
+	err := zfs.ZFSSnapshot(ctx, mustDatasetPath(pool), "testsnap", false)
+	require.NoError(ctx, err)
+
+	err = zfs.ZFSSnapshot(ctx, mustDatasetPath(poolchild), "testsnap", false)
+	require.NoError(ctx, err)
+
+	mustAddToSFilter(ctx, sfilter, pool)
+	mustAddToSFilter(ctx, sfilter, poolchild)
+
+	clientIdentity := "testclientid"
+
+	rfsRoot := ctx.RootDataset + "/receiver"
+
+	rep := replicationInvocation{
+		sjid:      sjid,
+		rjid:      rjid,
+		sfilter:   sfilter,
+		rfsRoot:   rfsRoot,
+		guarantee: pdu.ReplicationConfigProtectionWithKind(pdu.ReplicationGuaranteeKind_GuaranteeResumability),
+		receiverConfigHook: func(rc *endpoint.ReceiverConfig) {
+			rc.PlaceholderEncryption = endpoint.PlaceholderCreationEncryptionPropertyUnspecified
+			rc.AppendClientIdentity = true
+		},
+		interceptReceiver: func(r *endpoint.Receiver) logic.Receiver {
+			r.Test_OverrideClientIdentityFunc = func() string { return clientIdentity }
+			return r
+		},
+	}
+
+	r := rep.Do(ctx)
+	ctx.Logf("\n%s", pretty.Sprint(r))
+
+	require.Len(ctx, r.Attempts, 1)
+	attempt := r.Attempts[0]
+	require.Nil(ctx, attempt.PlanError)
+	require.Len(ctx, attempt.Filesystems, 2)
+
+	filesystemsByName := make(map[string]*report.FilesystemReport)
+	for _, fs := range attempt.Filesystems {
+		filesystemsByName[fs.Info.Name] = fs
+	}
+	require.Len(ctx, filesystemsByName, len(attempt.Filesystems))
+
+	afs, ok := filesystemsByName[pool]
+	require.True(ctx, ok)
+	require.Nil(ctx, afs.PlanError)
+	require.Nil(ctx, afs.StepError)
+	require.Equal(ctx, report.FilesystemDone, afs.State)
+
+	afs, ok = filesystemsByName[poolchild]
+	require.True(ctx, ok)
+	require.Nil(ctx, afs.PlanError)
+	require.Nil(ctx, afs.StepError)
+	require.Equal(ctx, report.FilesystemDone, afs.State)
+
+	mustGetFilesystemVersion(ctx, rfsRoot+"/"+clientIdentity+"/"+pool+"@testsnap")
+	mustGetFilesystemVersion(ctx, rfsRoot+"/"+clientIdentity+"/"+poolchild+"@testsnap")
+
+}
+
+func replicationPlaceholderEncryption__EncryptOnReceiverUseCase__impl(ctx *platformtest.Context, placeholderEncryption endpoint.PlaceholderCreationEncryptionProperty) {
+
+	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+		CREATEROOT
+		+  "sender"
+		+  "sender/a"
+		+  "sender/a/child"
+		+  "receiver" encrypted
+		R  zfs snapshot -r ${ROOTDS}/sender@initial
+	`)
+
+	sjid := endpoint.MustMakeJobID("sender-job")
+	rjid := endpoint.MustMakeJobID("receiver-job")
+
+	childfs := ctx.RootDataset + "/sender/a/child"
+
+	sfilter := filters.NewDatasetMapFilter(3, true)
+
+	mustAddToSFilter(ctx, sfilter, childfs)
+	rfsRoot := ctx.RootDataset + "/receiver"
+
+	rep := replicationInvocation{
+		sjid:      sjid,
+		rjid:      rjid,
+		sfilter:   sfilter,
+		rfsRoot:   rfsRoot,
+		guarantee: pdu.ReplicationConfigProtectionWithKind(pdu.ReplicationGuaranteeKind_GuaranteeResumability),
+		receiverConfigHook: func(rc *endpoint.ReceiverConfig) {
+			rc.PlaceholderEncryption = placeholderEncryption
+			rc.AppendClientIdentity = false
+		},
+	}
+
+	r := rep.Do(ctx)
+	ctx.Logf("\n%s", pretty.Sprint(r))
+
+	require.Len(ctx, r.Attempts, 1)
+	attempt := r.Attempts[0]
+	require.Equal(ctx, report.AttemptDone, attempt.State)
+	require.Len(ctx, attempt.Filesystems, 1)
+	afs := attempt.Filesystems[0]
+	require.Equal(ctx, childfs, afs.Info.Name)
+
+	require.Equal(ctx, 1, len(afs.Steps))
+
+	rfs := mustDatasetPath(rfsRoot + "/" + childfs)
+	mustGetFilesystemVersion(ctx, rfs.ToString()+"@initial")
+}
+
+func ReplicationPlaceholderEncryption__EncryptOnReceiverUseCase__WorksIfConfiguredWithInherit(ctx *platformtest.Context) {
+	placeholderEncryption := endpoint.PlaceholderCreationEncryptionPropertyInherit
+
+	replicationPlaceholderEncryption__EncryptOnReceiverUseCase__impl(ctx, placeholderEncryption)
+	childfs := ctx.RootDataset + "/sender/a/child"
+	rfsRoot := ctx.RootDataset + "/receiver"
+	rfs := mustDatasetPath(rfsRoot + "/" + childfs)
+
+	// The leaf child dataset should be inhering from rfsRoot.
+	// If we had replicated with PlaceholderCreationEncryptionPropertyOff
+	// then it would be unencrypted and inherit from the placeholder.
+	props, err := zfs.ZFSGet(ctx, rfs, []string{"encryptionroot"})
+	require.NoError(ctx, err)
+	require.Equal(ctx, rfsRoot, props.Get("encryptionroot"))
+}

replication/driver/replication_driver.go

@@ -151,6 +151,8 @@ type fs struct {
 
 	l *chainlock.L
 
+	blockedOn report.FsBlockedOn
+
 	// ordering relationship that must be maintained for initial replication
 	initialRepOrd struct {
 		parents, children []*fs
@@ -158,8 +160,9 @@ type fs struct {
 	}
 
 	planning struct {
-		done bool
+		waitingForStepQueue bool
-		err  *timedError
+		done                bool
+		err                 *timedError
 	}
 
 	// valid iff planning.done && planning.err == nil
@@ -337,8 +340,9 @@ func (a *attempt) doGlobalPlanning(ctx context.Context, prev *attempt) map[*fs]*
 
 	for _, pfs := range pfss {
 		fs := &fs{
-			fs: pfs,
+			fs:        pfs,
-			l:  a.l,
+			l:         a.l,
+			blockedOn: report.FsBlockedOnNothing,
 		}
 		fs.initialRepOrd.parentDidUpdate = make(chan struct{}, 1)
 		a.fss = append(a.fss, fs)
@@ -448,6 +452,10 @@ func (a *attempt) doFilesystems(ctx context.Context, prevs map[*fs]*fs) {
 			ctx, endTask := trace.WithTaskAndSpan(ctx, "repl-fs", f.report().Info.Name)
 			defer endTask()
 			f.do(ctx, stepQueue, prevs[f])
+			f.l.HoldWhile(func() {
+				// every return from f means it's unblocked...
+				f.blockedOn = report.FsBlockedOnNothing
+			})
 		}(f)
 	}
 	a.l.DropWhile(func() {
@@ -486,11 +494,16 @@ func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 	var psteps []Step
 	var errTime time.Time
 	var err error
+	f.blockedOn = report.FsBlockedOnPlanningStepQueue
 	f.l.DropWhile(func() {
 		// TODO hacky
 		// choose target time that is earlier than any snapshot, so fs planning is always prioritized
 		targetDate := time.Unix(0, 0)
 		defer pq.WaitReady(ctx, f, targetDate)()
+		f.l.HoldWhile(func() {
+			// transition before we call PlanFS
+			f.blockedOn = report.FsBlockedOnNothing
+		})
 		psteps, err = f.fs.PlanFS(ctx) // no shadow
 		errTime = time.Now()           // no shadow
 	})
@@ -545,6 +558,7 @@ func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 	f.planning.done = true
 
 	// wait for parents' initial replication
+	f.blockedOn = report.FsBlockedOnParentInitialRepl
 	var parents []string
 	for _, p := range f.initialRepOrd.parents {
 		parents = append(parents, p.fs.ReportInfo().Name)
@@ -613,7 +627,6 @@ func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 			select {
 			case <-ctx.Done():
 				f.planned.stepErr = newTimedError(ctx.Err(), time.Now())
-				return
 			case <-f.initialRepOrd.parentDidUpdate:
 				// loop
 			}
@@ -631,7 +644,9 @@ func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 		f.l.DropWhile(func() {
 			// wait for parallel replication
 			targetDate := s.step.TargetDate()
+			f.l.HoldWhile(func() { f.blockedOn = report.FsBlockedOnReplStepQueue })
 			defer pq.WaitReady(ctx, f, targetDate)()
+			f.l.HoldWhile(func() { f.blockedOn = report.FsBlockedOnNothing })
 			// do the step
 			ctx, endSpan := trace.WithSpan(ctx, fmt.Sprintf("%#v", s.step.ReportInfo()))
 			defer endSpan()
@@ -725,6 +740,7 @@ func (f *fs) report() *report.FilesystemReport {
 	r := &report.FilesystemReport{
 		Info:        f.fs.ReportInfo(),
 		State:       state,
+		BlockedOn:   f.blockedOn,
 		PlanError:   f.planning.err.IntoReportError(),
 		StepError:   f.planned.stepErr.IntoReportError(),
 		Steps:       make([]*report.StepReport, len(f.planned.steps)),

replication/driver/replication_stepqueue_test.go

@@ -14,11 +14,12 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/zrepl/zrepl/daemon/logging/trace"
+	"github.com/zrepl/zrepl/util/zreplcircleci"
 )
 
-// FIXME: this test relies on timing and is thus rather flaky
-// (relies on scheduler responsiveness of < 500ms)
 func TestPqNotconcurrent(t *testing.T) {
+	zreplcircleci.SkipOnCircleCI(t, "because it relies on scheduler responsiveness < 500ms")
+
 	ctx, end := trace.WithTaskFromStack(context.Background())
 	defer end()
 	var ctr uint32
@@ -90,6 +91,8 @@ func (r record) String() string {
 // Hence, perform some statistics on the wakeup times and assert that the mean wakeup
 // times for each step are close together.
 func TestPqConcurrent(t *testing.T) {
+	zreplcircleci.SkipOnCircleCI(t, "because it relies on scheduler responsiveness < 500ms")
+
 	ctx, end := trace.WithTaskFromStack(context.Background())
 	defer end()
 

replication/report/replication_report.go

@@ -62,11 +62,23 @@ const (
 	FilesystemDone            FilesystemState = "done"
 )
 
+type FsBlockedOn string
+
+const (
+	FsBlockedOnNothing           FsBlockedOn = "nothing"
+	FsBlockedOnPlanningStepQueue FsBlockedOn = "plan-queue"
+	FsBlockedOnParentInitialRepl FsBlockedOn = "parent-initial-repl"
+	FsBlockedOnReplStepQueue     FsBlockedOn = "repl-queue"
+)
+
 type FilesystemReport struct {
 	Info *FilesystemInfo
 
 	State FilesystemState
 
+	// Always valid.
+	BlockedOn FsBlockedOn
+
 	// Valid in State = FilesystemPlanningErrored
 	PlanError *TimedError
 	// Valid in State = FilesystemSteppingErrored

rpc/dataconn/timeoutconn/timeoutconn_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/zrepl/zrepl/util/socketpair"
+	"github.com/zrepl/zrepl/util/zreplcircleci"
 )
 
 func TestReadTimeout(t *testing.T) {
@@ -81,6 +82,8 @@ func TestWriteTimeout(t *testing.T) {
 }
 
 func TestNoPartialReadsDueToDeadline(t *testing.T) {
+	zreplcircleci.SkipOnCircleCI(t, "needs predictable low scheduling latency")
+
 	a, b, err := socketpair.SocketPair()
 	require.NoError(t, err)
 	defer a.Close()
@@ -151,6 +154,7 @@ func (c *partialWriteMockConn) Write(p []byte) (int, error) {
 }
 
 func TestPartialWriteMockConn(t *testing.T) {
+	zreplcircleci.SkipOnCircleCI(t, "because it relies on scheduler responsiveness < 50ms")
 	mc := newPartialWriteMockConn(100*time.Millisecond, 5)
 	buf := []byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
 	begin := time.Now()

util/zreplcircleci/zreplcircleci.go

@@ -0,0 +1,13 @@
+package zreplcircleci
+
+import (
+	"fmt"
+	"os"
+	"testing"
+)
+
+func SkipOnCircleCI(t *testing.T, reasonFmt string, args ...interface{}) {
+	if os.Getenv("CIRCLECI") != "" {
+		t.Skipf("This test is skipped in CircleCI. Reason: %s", fmt.Sprintf(reasonFmt, args...))
+	}
+}

zfs/filesystemplaceholdercreateencryptionvalue_enumer.go

@@ -0,0 +1,51 @@
+// Code generated by "enumer -type=FilesystemPlaceholderCreateEncryptionValue -trimprefix=FilesystemPlaceholderCreateEncryption"; DO NOT EDIT.
+
+//
+package zfs
+
+import (
+	"fmt"
+)
+
+const _FilesystemPlaceholderCreateEncryptionValueName = "InheritOff"
+
+var _FilesystemPlaceholderCreateEncryptionValueIndex = [...]uint8{0, 7, 10}
+
+func (i FilesystemPlaceholderCreateEncryptionValue) String() string {
+	i -= 1
+	if i < 0 || i >= FilesystemPlaceholderCreateEncryptionValue(len(_FilesystemPlaceholderCreateEncryptionValueIndex)-1) {
+		return fmt.Sprintf("FilesystemPlaceholderCreateEncryptionValue(%d)", i+1)
+	}
+	return _FilesystemPlaceholderCreateEncryptionValueName[_FilesystemPlaceholderCreateEncryptionValueIndex[i]:_FilesystemPlaceholderCreateEncryptionValueIndex[i+1]]
+}
+
+var _FilesystemPlaceholderCreateEncryptionValueValues = []FilesystemPlaceholderCreateEncryptionValue{1, 2}
+
+var _FilesystemPlaceholderCreateEncryptionValueNameToValueMap = map[string]FilesystemPlaceholderCreateEncryptionValue{
+	_FilesystemPlaceholderCreateEncryptionValueName[0:7]:  1,
+	_FilesystemPlaceholderCreateEncryptionValueName[7:10]: 2,
+}
+
+// FilesystemPlaceholderCreateEncryptionValueString retrieves an enum value from the enum constants string name.
+// Throws an error if the param is not part of the enum.
+func FilesystemPlaceholderCreateEncryptionValueString(s string) (FilesystemPlaceholderCreateEncryptionValue, error) {
+	if val, ok := _FilesystemPlaceholderCreateEncryptionValueNameToValueMap[s]; ok {
+		return val, nil
+	}
+	return 0, fmt.Errorf("%s does not belong to FilesystemPlaceholderCreateEncryptionValue values", s)
+}
+
+// FilesystemPlaceholderCreateEncryptionValueValues returns all values of the enum
+func FilesystemPlaceholderCreateEncryptionValueValues() []FilesystemPlaceholderCreateEncryptionValue {
+	return _FilesystemPlaceholderCreateEncryptionValueValues
+}
+
+// IsAFilesystemPlaceholderCreateEncryptionValue returns "true" if the value is listed in the enum definition. "false" otherwise
+func (i FilesystemPlaceholderCreateEncryptionValue) IsAFilesystemPlaceholderCreateEncryptionValue() bool {
+	for _, v := range _FilesystemPlaceholderCreateEncryptionValueValues {
+		if i == v {
+			return true
+		}
+	}
+	return false
+}

zfs/placeholder.go

@@ -80,7 +80,15 @@ func ZFSGetFilesystemPlaceholderState(ctx context.Context, p *DatasetPath) (stat
 	return state, nil
 }
 
-func ZFSCreatePlaceholderFilesystem(ctx context.Context, fs *DatasetPath, parent *DatasetPath) (err error) {
+//go:generate enumer -type=FilesystemPlaceholderCreateEncryptionValue -trimprefix=FilesystemPlaceholderCreateEncryption
+type FilesystemPlaceholderCreateEncryptionValue int
+
+const (
+	FilesystemPlaceholderCreateEncryptionInherit FilesystemPlaceholderCreateEncryptionValue = 1 << iota
+	FilesystemPlaceholderCreateEncryptionOff
+)
+
+func ZFSCreatePlaceholderFilesystem(ctx context.Context, fs *DatasetPath, parent *DatasetPath, encryption FilesystemPlaceholderCreateEncryptionValue) (err error) {
 	if fs.Length() == 1 {
 		return fmt.Errorf("cannot create %q: pools cannot be created with zfs create", fs.ToString())
 	}
@@ -90,11 +98,19 @@ func ZFSCreatePlaceholderFilesystem(ctx context.Context, fs *DatasetPath, parent
 		"-o", fmt.Sprintf("%s=%s", PlaceholderPropertyName, placeholderPropertyOn),
 		"-o", "mountpoint=none",
 	}
-	if parentEncrypted, err := ZFSGetEncryptionEnabled(ctx, parent.ToString()); err != nil {
+
-		return errors.Wrap(err, "cannot determine encryption support")
+	if !encryption.IsAFilesystemPlaceholderCreateEncryptionValue() {
-	} else if parentEncrypted {
+		panic(encryption)
-		cmdline = append(cmdline, "-o", "encryption=off")
 	}
+	switch encryption {
+	case FilesystemPlaceholderCreateEncryptionInherit:
+		// no-op
+	case FilesystemPlaceholderCreateEncryptionOff:
+		cmdline = append(cmdline, "-o", "encryption=off")
+	default:
+		panic(encryption)
+	}
+
 	cmdline = append(cmdline, fs.ToString())
 	cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, cmdline...)
 
@@ -148,3 +164,34 @@ func ZFSMigrateHashBasedPlaceholderToCurrent(ctx context.Context, fs *DatasetPat
 	}
 	return &report, nil
 }
+
+func ZFSListPlaceholderFilesystemsWithAdditionalProps(ctx context.Context, root string, additionalProps []string) (map[string]*ZFSProperties, error) {
+
+	props := []string{PlaceholderPropertyName}
+	if len(additionalProps) > 0 {
+		props = append(props, additionalProps...)
+	}
+
+	propsByFS, err := zfsGetRecursive(ctx, root, -1, []string{"filesystem", "volume"}, props, SourceAny)
+	if err != nil {
+		return nil, errors.Wrapf(err, "cannot get placeholder filesystems under %q", root)
+	}
+
+	filtered := make(map[string]*ZFSProperties)
+	for fs, props := range propsByFS {
+		details := props.GetDetails(PlaceholderPropertyName)
+		if details.Source != SourceLocal {
+			continue
+		}
+		fsp, err := NewDatasetPath(fs)
+		if err != nil {
+			return nil, errors.Wrapf(err, "zfs get returned invalid dataset path %q", fs)
+		}
+		if !isLocalPlaceholderPropertyValuePlaceholder(fsp, details.Value) {
+			continue
+		}
+		filtered[fs] = props
+	}
+
+	return filtered, nil
+}

zfs/zfs.go

@@ -1567,8 +1567,18 @@ func (s PropertySource) zfsGetSourceFieldPrefixes() []string {
 	return prefixes
 }
 
-func zfsGet(ctx context.Context, path string, props []string, allowedSources PropertySource) (*ZFSProperties, error) {
+func zfsGetRecursive(ctx context.Context, path string, depth int, dstypes []string, props []string, allowedSources PropertySource) (map[string]*ZFSProperties, error) {
-	args := []string{"get", "-Hp", "-o", "property,value,source", strings.Join(props, ","), path}
+	args := []string{"get", "-Hp", "-o", "name,property,value,source"}
+	if depth != 0 {
+		args = append(args, "-r")
+		if depth != -1 {
+			args = append(args, "-d", fmt.Sprintf("%d", depth))
+		}
+	}
+	if len(dstypes) > 0 {
+		args = append(args, "-t", strings.Join(dstypes, ","))
+	}
+	args = append(args, strings.Join(props, ","), path)
 	cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, args...)
 	stdout, err := cmd.Output()
 	if err != nil {
@@ -1588,36 +1598,67 @@ func zfsGet(ctx context.Context, path string, props []string, allowedSources Pro
 	}
 	o := string(stdout)
 	lines := strings.Split(o, "\n")
-	if len(lines) < 1 || // account for newlines
+	propsByFS := make(map[string]*ZFSProperties)
-		len(lines)-1 != len(props) {
-		return nil, fmt.Errorf("zfs get did not return the number of expected property values")
-	}
-	res := &ZFSProperties{
-		make(map[string]PropertyValue, len(lines)),
-	}
 	allowedPrefixes := allowedSources.zfsGetSourceFieldPrefixes()
-	for _, line := range lines[:len(lines)-1] {
+	for _, line := range lines[:len(lines)-1] { // last line is an empty line due to how strings.Split works
 		fields := strings.FieldsFunc(line, func(r rune) bool {
 			return r == '\t'
 		})
-		if len(fields) != 3 {
+		if len(fields) != 4 {
-			return nil, fmt.Errorf("zfs get did not return property,value,source tuples")
+			return nil, fmt.Errorf("zfs get did not return name,property,value,source tuples")
 		}
 		for _, p := range allowedPrefixes {
 			// prefix-match so that SourceAny (= "") works
-			if strings.HasPrefix(fields[2], p) {
+			if strings.HasPrefix(fields[3], p) {
-				source, err := parsePropertySource(fields[2])
+				source, err := parsePropertySource(fields[3])
 				if err != nil {
 					return nil, errors.Wrap(err, "parse property source")
 				}
-				res.m[fields[0]] = PropertyValue{
+				fsProps, ok := propsByFS[fields[0]]
-					Value:  fields[1],
+				if !ok {
+					fsProps = &ZFSProperties{
+						make(map[string]PropertyValue),
+					}
+				}
+				if _, ok := fsProps.m[fields[1]]; ok {
+					return nil, errors.Errorf("duplicate property %q for dataset %q", fields[1], fields[0])
+				}
+				fsProps.m[fields[1]] = PropertyValue{
+					Value:  fields[2],
 					Source: source,
 				}
+				propsByFS[fields[0]] = fsProps
 				break
 			}
 		}
 	}
+	// validate we got expected output
+	for fs, fsProps := range propsByFS {
+		if len(fsProps.m) != len(props) {
+			return nil, errors.Errorf("zfs get did not return all requested values for dataset %q\noutput was:\n%s", fs, o)
+		}
+	}
+	return propsByFS, nil
+}
+
+func zfsGet(ctx context.Context, path string, props []string, allowedSources PropertySource) (*ZFSProperties, error) {
+	propMap, err := zfsGetRecursive(ctx, path, 0, nil, props, allowedSources)
+	if err != nil {
+		return nil, err
+	}
+	if len(propMap) == 0 {
+		// XXX callers expect to always get a result here
+		// They will observe props.Get("propname") == ""
+		// We should change .Get to return a tuple, or an error, or whatever.
+		return &ZFSProperties{make(map[string]PropertyValue)}, nil
+	}
+	if len(propMap) != 1 {
+		return nil, errors.Errorf("zfs get unexpectedly returned properties for multiple datasets")
+	}
+	res, ok := propMap[path]
+	if !ok {
+		return nil, errors.Errorf("zfs get returned properties for a different dataset that requested")
+	}
 	return res, nil
 }