WIP: fix encrypt-on-recv

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

.circleci/config.yml

@@ -150,7 +150,7 @@ parameters:
 
   release_docker_baseimage_tag:
     type: string
-    default: "1.17"
+    default: "1.16"
 
 workflows:
   version: 2
@@ -160,15 +160,15 @@ workflows:
     jobs:
       - quickcheck-docs
       - quickcheck-go: &quickcheck-go-smoketest
-          name: quickcheck-go-amd64-linux-1.17
+          name: quickcheck-go-amd64-linux-1.16
-          goversion: &latest-go-release "1.17"
+          goversion: &latest-go-release "1.16"
           goos: linux
           goarch: amd64
       - test-go-on-latest-go-release:
           goversion: *latest-go-release
       - quickcheck-go:
           requires:
-            - quickcheck-go-amd64-linux-1.17 #quickcheck-go-smoketest.name
+            - quickcheck-go-amd64-linux-1.16 #quickcheck-go-smoketest.name
           matrix: &quickcheck-go-matrix
             alias: quickcheck-go-matrix
             parameters:
@@ -231,6 +231,8 @@ jobs:
       - install-docdep
       - run: make docs
 
+      - store_artifacts:
+          path: artifacts
       - download-and-install-minio-client
       - upload-minio:
           src: artifacts

Makefile

@@ -28,7 +28,7 @@ GO_BUILDFLAGS := $(GO_MOD_READONLY) $(GO_EXTRA_BUILDFLAGS)
 GO_BUILD := $(GO_ENV_VARS) $(GO) build $(GO_BUILDFLAGS) -ldflags $(GO_LDFLAGS)
 GOLANGCI_LINT := golangci-lint
 GOCOVMERGE := gocovmerge
-RELEASE_DOCKER_BASEIMAGE_TAG ?= 1.17
+RELEASE_DOCKER_BASEIMAGE_TAG ?= 1.16
 RELEASE_DOCKER_BASEIMAGE ?= golang:$(RELEASE_DOCKER_BASEIMAGE_TAG)
 
 ifneq ($(GOARM),)
@@ -189,8 +189,6 @@ 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
 
-* 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.
+* 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.
 * 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.

build/tools.go

@@ -1,4 +1,3 @@
-//go:build tools
 // +build tools
 
 package main

client/status/viewmodel/bytecountbinary.go

@@ -2,22 +2,14 @@ package viewmodel
 
 import (
 	"fmt"
-	"math"
 )
 
-func ByteCountBinaryUint(b uint64) string {
-	if b > math.MaxInt64 {
-		panic(b)
-	}
-	return ByteCountBinary(int64(b))
-}
-
 func ByteCountBinary(b int64) string {
 	const unit = 1024
 	if b < unit {
 		return fmt.Sprintf("%d B", b)
 	}
-	div, exp := unit, 0
+	div, exp := int64(unit), 0
 	for n := b / unit; n >= unit; n /= unit {
 		div *= unit
 		exp++

client/status/viewmodel/bytesprogresshistory.go

@@ -4,16 +4,17 @@ import "time"
 
 type byteProgressMeasurement struct {
 	time time.Time
-	val  uint64
+	val  int64
 }
 
 type bytesProgressHistory struct {
 	last        *byteProgressMeasurement // pointer as poor man's optional
 	changeCount int
 	lastChange  time.Time
+	bpsAvg      float64
 }
 
-func (p *bytesProgressHistory) Update(currentVal uint64) (bytesPerSecondAvg int64, changeCount int) {
+func (p *bytesProgressHistory) Update(currentVal int64) (bytesPerSecondAvg int64, changeCount int) {
 
 	if p.last == nil {
 		p.last = &byteProgressMeasurement{
@@ -33,17 +34,15 @@ func (p *bytesProgressHistory) Update(currentVal uint64) (bytesPerSecondAvg int6
 		return 0, 0
 	}
 
-	var deltaV int64
+	deltaV := currentVal - p.last.val
-	if currentVal >= p.last.val {
-		deltaV = int64(currentVal - p.last.val)
-	} else {
-		deltaV = -int64(p.last.val - currentVal)
-	}
 	deltaT := time.Since(p.last.time)
 	rate := float64(deltaV) / deltaT.Seconds()
 
+	factor := 0.3
+	p.bpsAvg = (1-factor)*p.bpsAvg + factor*rate
+
 	p.last.time = time.Now()
 	p.last.val = currentVal
 
-	return int64(rate), p.changeCount
+	return int64(p.bpsAvg), p.changeCount
 }

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, maxFS int) {
+func printFilesystemStatus(t *stringbuilder.B, rep *report.FilesystemReport, active bool, maxFS int) {
 
 	expected, replicated, containsInvalidSizeEstimates := rep.BytesSum()
 	sizeEstimationImpreciseNotice := ""
@@ -227,20 +227,15 @@ func printFilesystemStatus(t *stringbuilder.B, rep *report.FilesystemReport, max
 		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)),
-		userVisisbleCurrentStep, userVisibleTotalSteps,
+		rep.CurrentStep, len(rep.Steps),
-		ByteCountBinaryUint(replicated), ByteCountBinaryUint(expected),
+		ByteCountBinary(replicated), ByteCountBinary(expected),
 		sizeEstimationImpreciseNotice,
 	)
 
 	activeIndicator := " "
-	if rep.BlockedOn == report.FsBlockedOnNothing &&
+	if active {
-		(rep.State == report.FilesystemPlanning || rep.State == report.FilesystemStepping) {
 		activeIndicator = "*"
 	}
 	t.AddIndent(1)
@@ -363,12 +358,12 @@ func renderReplicationReport(t *stringbuilder.B, rep *report.Report, history *by
 		rate, changeCount := history.Update(replicated)
 		eta := time.Duration(0)
 		if rate > 0 {
-			eta = time.Duration((float64(expected)-float64(replicated))/float64(rate)) * time.Second
+			eta = time.Duration((expected-replicated)/rate) * time.Second
 		}
 
 		t.Write("Progress: ")
 		t.DrawBar(50, replicated, expected, changeCount)
-		t.Write(fmt.Sprintf(" %s / %s @ %s/s", ByteCountBinaryUint(replicated), ByteCountBinaryUint(expected), ByteCountBinary(rate)))
+		t.Write(fmt.Sprintf(" %s / %s @ %s/s", ByteCountBinary(replicated), ByteCountBinary(expected), ByteCountBinary(rate)))
 		if eta != 0 {
 			t.Write(fmt.Sprintf(" (%s remaining)", humanizeDuration(eta)))
 		}
@@ -390,7 +385,7 @@ func renderReplicationReport(t *stringbuilder.B, rep *report.Report, history *by
 			}
 		}
 		for _, fs := range latest.Filesystems {
-			printFilesystemStatus(t, fs, maxFSLen)
+			printFilesystemStatus(t, fs, false, maxFSLen) // FIXME bring 'active' flag back
 		}
 
 	}

client/status/viewmodel/stringbuilder/stringbuilder.go

@@ -99,11 +99,11 @@ func RightPad(str string, length int, pad string) string {
 }
 
 // changeCount = 0 indicates stall / no progress
-func (w *B) DrawBar(length int, bytes, totalBytes uint64, changeCount int) {
+func (w *B) DrawBar(length int, bytes, totalBytes int64, changeCount int) {
 	const arrowPositions = `>\|/`
 	var completedLength int
 	if totalBytes > 0 {
-		completedLength = int(uint64(length) * bytes / totalBytes)
+		completedLength = int(int64(length) * bytes / totalBytes)
 		if completedLength > length {
 			completedLength = length
 		}

client/testcmd.go

@@ -139,11 +139,9 @@ 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")
@@ -156,7 +154,6 @@ 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
@@ -174,12 +171,7 @@ func runTestPlaceholder(ctx context.Context, subcommand *cli.Subcommand, args []
 			return errors.Wrap(err, "cannot get placeholder state")
 		}
 		if !ph.FSExists {
-			if datasetWasExplicitArgument {
+			panic("placeholder state inconsistent: filesystem " + ph.FS + " must exist in this context")
-				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

@@ -13,7 +13,6 @@ import (
 	"github.com/pkg/errors"
 	"github.com/zrepl/yaml-config"
 
-	"github.com/zrepl/zrepl/util/datasizeunit"
 	zfsprop "github.com/zrepl/zrepl/zfs/property"
 )
 
@@ -86,10 +85,8 @@ 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:"embedded_data,optional,default=false"`
+	EmbeddedData     bool `yaml:"embbeded_data,optional,default=false"`
 	Saved            bool `yaml:"saved,optional,default=false"`
-
-	BandwidthLimit *BandwidthLimit `yaml:"bandwidth_limit,optional,fromdefaults"`
 }
 
 type RecvOptions struct {
@@ -99,17 +96,6 @@ type RecvOptions struct {
 	// Reencrypt bool `yaml:"reencrypt"`
 
 	Properties *PropertyRecvOptions `yaml:"properties,fromdefaults"`
-
-	BandwidthLimit *BandwidthLimit `yaml:"bandwidth_limit,optional,fromdefaults"`
-
-	Placeholder *PlaceholderRecvOptions `yaml:"placeholder,fromdefaults"`
-}
-
-var _ yaml.Unmarshaler = &datasizeunit.Bits{}
-
-type BandwidthLimit struct {
-	Max            datasizeunit.Bits `yaml:"max,default=-1 B"`
-	BucketCapacity datasizeunit.Bits `yaml:"bucket_capacity,default=128 KiB"`
 }
 
 type Replication struct {
@@ -127,15 +113,15 @@ type ReplicationOptionsConcurrency struct {
 	SizeEstimates int `yaml:"size_estimates,optional,default=4"`
 }
 
+func (l *RecvOptions) SetDefault() {
+	*l = RecvOptions{Properties: &PropertyRecvOptions{}}
+}
+
 type PropertyRecvOptions struct {
 	Inherit  []zfsprop.Property          `yaml:"inherit,optional"`
 	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/bandwidth_limit.yml

@@ -1,41 +0,0 @@
-
-jobs:
-  - type: sink
-    name: "limited_sink"
-    root_fs: "fs0"
-    recv:
-      bandwidth_limit:
-        max: 12345 B
-    serve:
-      type: local
-      listener_name: localsink
-
-  - type: push
-    name: "limited_push"
-    connect:
-      type: local
-      listener_name: localsink
-      client_identity: local_backup
-    filesystems: {
-      "root<": true,
-    }
-    send:
-      bandwidth_limit:
-        max: 54321 B
-        bucket_capacity: 1024 B
-    snapshotting:
-      type: manual
-    pruning:
-      keep_sender:
-        - type: last_n
-          count: 1
-      keep_receiver:
-        - type: last_n
-          count: 1
-
-  - type: sink
-    name: "nolimit_sink"
-    root_fs: "fs1"
-    serve:
-      type: local
-      listener_name: localsink

config/samples/quickstart_continuous_server_backup_receiver.yml

@@ -9,7 +9,4 @@ 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,7 +8,6 @@ import (
 	"io"
 	"net"
 	"net/http"
-	"os"
 	"time"
 
 	"github.com/pkg/errors"
@@ -125,9 +124,8 @@ 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
 		}})
@@ -158,8 +156,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: envconst.Duration("ZREPL_DAEMON_CONTROL_SERVER_WRITE_TIMEOUT", 1*time.Second),
+		WriteTimeout: 1 * time.Second,
-		ReadTimeout:  envconst.Duration("ZREPL_DAEMON_CONTROL_SERVER_READ_TIMEOUT", 1*time.Second),
+		ReadTimeout:  1 * time.Second,
 	}
 
 outer:

daemon/daemon.go

@@ -160,9 +160,8 @@ 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,7 +42,6 @@ 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
@@ -322,6 +321,7 @@ 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,13 +329,6 @@ 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 {
@@ -367,7 +360,6 @@ 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() }
@@ -502,11 +494,7 @@ func (j *ActiveSide) do(ctx context.Context) {
 		repCancel()   // always cancel to free up context resources
 
 		replicationReport := j.tasks.replicationReport()
-		var numErrors = replicationReport.GetFailedFilesystemsCountInLatestAttempt()
+		j.promReplicationErrors.Set(float64(replicationReport.GetFailedFilesystemsCountInLatestAttempt()))
-		j.promReplicationErrors.Set(float64(numErrors))
-		if numErrors == 0 {
-			j.promLastSuccessful.SetToCurrentTime()
-		}
 
 		endSpan()
 	}

daemon/job/build_jobs.go

@@ -8,7 +8,6 @@ import (
 	"github.com/pkg/errors"
 
 	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/util/bandwidthlimit"
 )
 
 func JobsFromConfig(c *config.Config) ([]Job, error) {
@@ -108,13 +107,3 @@ func validateReceivingSidesDoNotOverlap(receivingRootFSs []string) error {
 	}
 	return nil
 }
-
-func buildBandwidthLimitConfig(in *config.BandwidthLimit) (c bandwidthlimit.Config, _ error) {
-	if in.Max.ToBytes() > 0 && int64(in.Max.ToBytes()) == 0 {
-		return c, fmt.Errorf("bandwidth limit `max` is too small, must at least specify one byte")
-	}
-	return bandwidthlimit.Config{
-		Max:            int64(in.Max.ToBytes()),
-		BucketCapacity: int64(in.BucketCapacity.ToBytes()),
-	}, nil
-}

daemon/job/build_jobs_sendrecvoptions.go

@@ -22,12 +22,7 @@ func buildSenderConfig(in SendingJobConfig, jobID endpoint.JobID) (*endpoint.Sen
 		return nil, errors.Wrap(err, "cannot build filesystem filter")
 	}
 	sendOpts := in.GetSendOptions()
-	bwlim, err := buildBandwidthLimitConfig(sendOpts.BandwidthLimit)
+	return &endpoint.SenderConfig{
-	if err != nil {
-		return nil, errors.Wrap(err, "cannot build bandwith limit config")
-	}
-
-	sc := &endpoint.SenderConfig{
 		FSF:   fsf,
 		JobID: jobID,
 
@@ -39,15 +34,7 @@ func buildSenderConfig(in SendingJobConfig, jobID endpoint.JobID) (*endpoint.Sen
 		SendCompressed:       sendOpts.Compressed,
 		SendEmbeddedData:     sendOpts.EmbeddedData,
 		SendSaved:            sendOpts.Saved,
-
+	}, nil
-		BandwidthLimit: bwlim,
-	}
-
-	if err := sc.Validate(); err != nil {
-		return nil, errors.Wrap(err, "cannot build sender config")
-	}
-
-	return sc, nil
 }
 
 type ReceivingJobConfig interface {
@@ -66,22 +53,6 @@ func buildReceiverConfig(in ReceivingJobConfig, jobID endpoint.JobID) (rc endpoi
 	}
 
 	recvOpts := in.GetRecvOptions()
-
-	bwlim, err := buildBandwidthLimitConfig(recvOpts.BandwidthLimit)
-	if err != nil {
-		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,
@@ -89,10 +60,6 @@ func buildReceiverConfig(in ReceivingJobConfig, jobID endpoint.JobID) (rc endpoi
 
 		InheritProperties:  recvOpts.Properties.Inherit,
 		OverrideProperties: recvOpts.Properties.Override,
-
-		BandwidthLimit: bwlim,
-
-		PlaceholderEncryption: placeholderEncryption,
 	}
 	if err := rc.Validate(); err != nil {
 		return rc, errors.Wrap(err, "cannot build receiver config")

daemon/job/build_jobs_test.go

@@ -119,14 +119,6 @@ func TestSampleConfigsAreBuiltWithoutErrors(t *testing.T) {
 		t.Errorf("glob failed: %+v", err)
 	}
 
-	type additionalCheck struct {
-		state int
-		test  func(t *testing.T, jobs []Job)
-	}
-	additionalChecks := map[string]*additionalCheck{
-		"bandwidth_limit.yml": {test: testSampleConfig_BandwidthLimit},
-	}
-
 	for _, p := range paths {
 
 		if path.Ext(p) != ".yml" {
@@ -134,20 +126,10 @@ func TestSampleConfigsAreBuiltWithoutErrors(t *testing.T) {
 			continue
 		}
 
-		filename := path.Base(p)
-		t.Logf("checking for presence additonal checks for file %q", filename)
-		additionalCheck := additionalChecks[filename]
-		if additionalCheck == nil {
-			t.Logf("no additional checks")
-		} else {
-			t.Logf("additional check present")
-			additionalCheck.state = 1
-		}
-
 		t.Run(p, func(t *testing.T) {
 			c, err := config.ParseConfig(p)
 			if err != nil {
-				t.Fatalf("error parsing %s:\n%+v", p, err)
+				t.Errorf("error parsing %s:\n%+v", p, err)
 			}
 
 			t.Logf("file: %s", p)
@@ -156,57 +138,11 @@ func TestSampleConfigsAreBuiltWithoutErrors(t *testing.T) {
 			tls.FakeCertificateLoading(t)
 			jobs, err := JobsFromConfig(c)
 			t.Logf("jobs: %#v", jobs)
-			require.NoError(t, err)
+			assert.NoError(t, err)
-
-			if additionalCheck != nil {
-				additionalCheck.test(t, jobs)
-				additionalCheck.state = 2
-			}
 		})
 
 	}
 
-	for basename, c := range additionalChecks {
-		if c.state == 0 {
-			panic("univisited additional check " + basename)
-		}
-	}
-
-}
-
-func testSampleConfig_BandwidthLimit(t *testing.T, jobs []Job) {
-	require.Len(t, jobs, 3)
-
-	{
-		limitedSink, ok := jobs[0].(*PassiveSide)
-		require.True(t, ok, "%T", jobs[0])
-		limitedSinkMode, ok := limitedSink.mode.(*modeSink)
-		require.True(t, ok, "%T", limitedSink)
-
-		assert.Equal(t, int64(12345), limitedSinkMode.receiverConfig.BandwidthLimit.Max)
-		assert.Equal(t, int64(1<<17), limitedSinkMode.receiverConfig.BandwidthLimit.BucketCapacity)
-	}
-
-	{
-		limitedPush, ok := jobs[1].(*ActiveSide)
-		require.True(t, ok, "%T", jobs[1])
-		limitedPushMode, ok := limitedPush.mode.(*modePush)
-		require.True(t, ok, "%T", limitedPush)
-
-		assert.Equal(t, int64(54321), limitedPushMode.senderConfig.BandwidthLimit.Max)
-		assert.Equal(t, int64(1024), limitedPushMode.senderConfig.BandwidthLimit.BucketCapacity)
-	}
-
-	{
-		unlimitedSink, ok := jobs[2].(*PassiveSide)
-		require.True(t, ok, "%T", jobs[2])
-		unlimitedSinkMode, ok := unlimitedSink.mode.(*modeSink)
-		require.True(t, ok, "%T", unlimitedSink)
-
-		max := unlimitedSinkMode.receiverConfig.BandwidthLimit.Max
-		assert.Less(t, max, int64(0), max, "unlimited mode <=> negative value for .Max, see bandwidthlimit.Config")
-	}
-
 }
 
 func TestReplicationOptions(t *testing.T) {

daemon/job/snapjob.go

@@ -10,7 +10,6 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/zrepl/zrepl/daemon/logging/trace"
-	"github.com/zrepl/zrepl/util/bandwidthlimit"
 	"github.com/zrepl/zrepl/util/nodefault"
 
 	"github.com/zrepl/zrepl/config"
@@ -147,7 +146,7 @@ type alwaysUpToDateReplicationCursorHistory struct {
 	target pruner.Target
 }
 
-var _ pruner.Sender = (*alwaysUpToDateReplicationCursorHistory)(nil)
+var _ pruner.History = (*alwaysUpToDateReplicationCursorHistory)(nil)
 
 func (h alwaysUpToDateReplicationCursorHistory) ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error) {
 	fsvReq := &pdu.ListFilesystemVersionsReq{
@@ -180,11 +179,8 @@ func (j *SnapJob) doPrune(ctx context.Context) {
 	sender := endpoint.NewSender(endpoint.SenderConfig{
 		JobID: j.name,
 		FSF:   j.fsfilter,
-		// FIXME the following config fields are irrelevant for SnapJob
+		// FIXME encryption setting is irrelevant for SnapJob because the endpoint is only used as pruner.Target
-		// because the endpoint is only used as pruner.Target.
+		Encrypt: &nodefault.Bool{B: true},
-		// However, the implementation requires them to be set.
-		Encrypt:        &nodefault.Bool{B: true},
-		BandwidthLimit: bandwidthlimit.NoLimitConfig(),
 	})
 	j.prunerMtx.Lock()
 	j.pruner = j.prunerFactory.BuildLocalPruner(ctx, sender, alwaysUpToDateReplicationCursorHistory{sender})

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 (run daemon with env var %s=1 for more details)\n", this.activeChildTasks, debugEnabledEnvVar))
+				panic(errors.WithMessagef(ErrTaskStillHasActiveChildTasks, "end task: %v active child tasks\n", this.activeChildTasks))
 			}
 
 			// support idempotent task ends

daemon/logging/trace/trace_debug.go

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

daemon/pruner/pruner.go

@@ -19,20 +19,12 @@ import (
 	"github.com/zrepl/zrepl/util/envconst"
 )
 
-// The sender in the replication setup.
-// The pruner uses the Sender to determine which of the Target's filesystems need to be pruned.
-// Also, it asks the Sender about the replication cursor of each filesystem
-// to enable the 'not_replicated' pruning rule.
-//
 // Try to keep it compatible with github.com/zrepl/zrepl/endpoint.Endpoint
-type Sender interface {
+type History interface {
 	ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error)
 	ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error)
 }
 
-// The pruning target, i.e., on which snapshots are destroyed.
-// This can be a replication sender or receiver.
-//
 // Try to keep it compatible with github.com/zrepl/zrepl/endpoint.Endpoint
 type Target interface {
 	ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error)
@@ -56,7 +48,7 @@ func GetLogger(ctx context.Context) Logger {
 type args struct {
 	ctx                            context.Context
 	target                         Target
-	sender                         Sender
+	receiver                       History
 	rules                          []pruning.KeepRule
 	retryWait                      time.Duration
 	considerSnapAtCursorReplicated bool
@@ -140,12 +132,12 @@ func NewPrunerFactory(in config.PruningSenderReceiver, promPruneSecs *prometheus
 	return f, nil
 }
 
-func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, target Target, sender Sender) *Pruner {
+func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, target Target, receiver History) *Pruner {
 	p := &Pruner{
 		args: args{
 			context.WithValue(ctx, contextKeyPruneSide, "sender"),
 			target,
-			sender,
+			receiver,
 			f.senderRules,
 			f.retryWait,
 			f.considerSnapAtCursorReplicated,
@@ -156,12 +148,12 @@ func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, target Target, se
 	return p
 }
 
-func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target, sender Sender) *Pruner {
+func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target, receiver History) *Pruner {
 	p := &Pruner{
 		args: args{
 			context.WithValue(ctx, contextKeyPruneSide, "receiver"),
 			target,
-			sender,
+			receiver,
 			f.receiverRules,
 			f.retryWait,
 			false, // senseless here anyways
@@ -172,12 +164,12 @@ func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target,
 	return p
 }
 
-func (f *LocalPrunerFactory) BuildLocalPruner(ctx context.Context, target Target, history Sender) *Pruner {
+func (f *LocalPrunerFactory) BuildLocalPruner(ctx context.Context, target Target, receiver History) *Pruner {
 	p := &Pruner{
 		args: args{
 			context.WithValue(ctx, contextKeyPruneSide, "local"),
 			target,
-			history,
+			receiver,
 			f.keepRules,
 			f.retryWait,
 			false, // considerSnapAtCursorReplicated is not relevant for local pruning
@@ -351,9 +343,9 @@ func (s snapshot) Date() time.Time { return s.date }
 
 func doOneAttempt(a *args, u updater) {
 
-	ctx, target, sender := a.ctx, a.target, a.sender
+	ctx, target, receiver := a.ctx, a.target, a.receiver
 
-	sfssres, err := sender.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
+	sfssres, err := receiver.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
 	if err != nil {
 		u(func(p *Pruner) {
 			p.state = PlanErr
@@ -418,7 +410,7 @@ tfss_loop:
 		rcReq := &pdu.ReplicationCursorReq{
 			Filesystem: tfs.Path,
 		}
-		rc, err := sender.ReplicationCursor(ctx, rcReq)
+		rc, err := receiver.ReplicationCursor(ctx, rcReq)
 		if err != nil {
 			pfsPlanErrAndLog(err, "cannot get replication cursor bookmark")
 			continue tfss_loop
@@ -464,7 +456,7 @@ tfss_loop:
 			})
 		}
 		if preCursor {
-			pfsPlanErrAndLog(fmt.Errorf("prune target has no snapshot that corresponds to sender replication cursor bookmark"), "")
+			pfsPlanErrAndLog(fmt.Errorf("replication cursor not found in prune target filesystem versions"), "")
 			continue tfss_loop
 		}
 

docs/changelog.rst

@@ -16,44 +16,18 @@ 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.
 
-0.5
+We use the following annotations for classifying changes:
----
-
-* |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
 -----
@@ -79,6 +53,13 @@ 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

@@ -67,9 +67,8 @@ Policy ``not_replicated``
      ...
 
 ``not_replicated`` keeps all snapshots that have not been replicated to the receiving side.
-It only makes sense to specify this rule for the ``keep_sender``.
+It only makes sense to specify this rule on a sender (source or push job).
-The reason is that, by definition, all snapshots on the receiver have already been replicated to there from the sender.
+The state required to evaluate this rule is stored in the :ref:`replication cursor bookmark <replication-cursor-and-last-received-hold>` on the sending side.
-To determine whether a sender-side snapshot has already been replicated, zrepl uses the :ref:`replication cursor bookmark <replication-cursor-and-last-received-hold>` which corresponds to the most recent successfully replicated snapshot.
 
 .. _prune-keep-retention-grid:
 
@@ -107,7 +106,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 are **not kept** by the grid specification.
+#. Snapshots older than the rightmost bucket **not kept** by this gridspec.
 #. For each bucket, we only keep the ``keep`` oldest snapshots.
 
 The syntax to describe the bucket list is as follows:
@@ -125,51 +124,43 @@ The syntax to describe the bucket list is as follows:
 
 ::
 
-   Assume the following grid specification:
+          This grid spec produces the following list of adjacent buckets. For the sake of simplicity,
+          we subject all snapshots to the grid pruning policy by settings `regex: .*`.
 
-      grid: 1x1h(keep=all) | 2x2h | 1x3h
+          `
+           grid: 1x1h(keep=all) | 2x2h | 1x3h
+           regex: .*
+          `
 
-   This grid specification produces the following constellation of buckets:
+          0h        1h        2h        3h        4h        5h        6h        7h        8h        9h
-
+          |         |         |         |         |         |         |         |         |         |
-   0h        1h        2h        3h        4h        5h        6h        7h        8h        9h
+          |-Bucket1-|-----Bucket 2------|------Bucket 3-----|-----------Bucket 4----------|
-   |         |         |         |         |         |         |         |         |         |
+          | keep=all|      keep=1       |       keep=1      |            keep=1           |
-   |-Bucket1-|-----Bucket2-------|------Bucket3------|-----------Bucket4-----------|
-   | keep=all|      keep=1       |       keep=1      |            keep=1           |
 
 
 
-   Now assume that we have a set of snapshots @a, @b, ..., @D.
+          Let us consider the following set of snapshots @a-zA-C:
-   Snapshot @a is the most recent snapshot.
-   Snapshot @D is the oldest snapshot, it is almost 9 hours older than snapshot @a.
-   We place the snapshots on the same timeline as the buckets:
 
 
-   0h        1h        2h        3h        4h        5h        6h        7h        8h        9h
+          |  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        |
-   |         |         |         |         |         |         |         |         |         |
-   |-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
 
-   We obtain the following mapping of snapshots to buckets:
+          The `grid` algorithm maps them to their respective buckets:
 
-   Bucket1:   a,b,c
+          Bucket 1: a, b, c
-   Bucket2:   d,e,f,g,h,i
+          Bucket 2: d,e,f,g,h,i,j
-   Bucket3:   j,k,l,m,n,o,p
+          Bucket 3: k,l,m,n,o,p
-   Bucket4:   q,r,s,t,u,v,w,x,y,z
+          Bucket 4: q,r, q,r,s,t,u,v,w,x,y,z
-   No bucket: A,B,C,D
+          None:     A,B,C,D
 
-   For each bucket, we now prune snapshots until it only contains `keep` snapshots.
+          It then applies the per-bucket pruning logic described above which resulting in the
-   Newer snapshots are destroyed first.
+          following list of remaining snapshots.
-   Snapshots that do not fall into a bucket are always destroyed.
 
-   Result after pruning:
+          |  a  b  c                    j                 p                             z                    |
+
+          Note that it only makes sense to grow (not shorten) the interval duration for buckets
+          further in the past since each bucket acts like a low-pass filter for incoming snapshots
+          and adding a less-low-pass-filter after a low-pass one has no effect.
 
-   0h        1h        2h        3h        4h        5h        6h        7h        8h        9h
-   |         |         |         |         |         |         |         |         |         |
-   |-Bucket1-|-----Bucket2-------|------Bucket3------|-----------Bucket4-----------|
-   |         |                   |                   |                             |
-   | a  b  c |                i  |                 p |                           z |
 
 .. _prune-keep-last-n:
 

docs/configuration/sendrecvoptions.rst

@@ -36,12 +36,9 @@ See the `upstream man page <https://openzfs.github.io/openzfs-docs/man/8/zfs-sen
     * - ``encrypted``
       -
       - Specific to zrepl, :ref:`see below <job-send-options-encrypted>`.
-    * - ``bandwidth_limit``
-      -
-      - Specific to zrepl, :ref:`see below <job-send-recv-options--bandwidth-limit>`.
     * - ``raw``
       - ``-w``
-      - Use ``encrypted`` to only allow encrypted sends. Mixed sends are not supported.
+      - Use ``encrypted`` to only allow encrypted sends.
     * - ``send_properties``
       - ``-p``
       - **Be careful**, read the :ref:`note on property replication below <job-note-property-replication>`.
@@ -54,7 +51,7 @@ See the `upstream man page <https://openzfs.github.io/openzfs-docs/man/8/zfs-sen
     * - ``compressed``
       - ``-c``
       -
-    * - ``embedded_data``
+    * - ``embbeded_data``
       - ``-e``
       -
     * - ``saved``
@@ -141,16 +138,8 @@ Recv Options
          override: {
            "org.openzfs.systemd:ignore": "on"
          }
-       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``
@@ -223,55 +212,3 @@ and property replication is enabled, the receiver must :ref:`inherit the followi
 * ``keylocation``
 * ``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:
-
-Bandwidth Limit (send & recv)
------------------------------
-
-::
-
-   bandwidth_limit:
-     max: 23.5 MiB # -1 is the default and disabled rate limiting
-     bucket_capacity: # token bucket capacity in bytes; defaults to 128KiB
-
-Both ``send`` and ``recv`` can be limited to a maximum bandwidth through ``bandwidth_limit``.
-For most users, it should be sufficient to just set ``bandwidth_limit.max``.
-The ``bandwidth_limit.bucket_capacity`` refers to the `token bucket size <https://github.com/juju/ratelimit>`_.
-
-The bandwidth limit only applies to the payload data, i.e., the ZFS send stream.
-It does not account for transport protocol overheads.
-The scope is the job level, i.e., all :ref:`concurrent <replication-option-concurrency>` sends or incoming receives of a job share the bandwidth limit.

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,7 +66,6 @@ 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,29 +9,18 @@ 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; \
-    set -ex
+    ARCH="$(dpkg --print-architecture)"; \
-    zrepl_apt_key_url=https://zrepl.cschwarz.com/apt/apt-key.asc
+    CODENAME="$(lsb_release -i -s | tr '[:upper:]' '[:lower:]') $(lsb_release -c -s | tr '[:upper:]' '[:lower:]')"; \
-    zrepl_apt_key_dst=/usr/share/keyrings/zrepl.gpg
+    echo "Using Distro and Codename: $CODENAME"; \
-    zrepl_apt_repo_file=/etc/apt/sources.list.d/zrepl.list
+    (curl https://zrepl.cschwarz.com/apt/apt-key.asc | sudo apt-key add -) && \
-
+    (echo "deb [arch=$ARCH] https://zrepl.cschwarz.com/apt/$CODENAME main" | sudo tee /etc/apt/sources.list.d/zrepl.list) && \
-    # 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,7 +93,6 @@ 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
 ######
@@ -135,18 +134,7 @@ 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,19 +51,6 @@ 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/supporters.rst

@@ -32,8 +32,6 @@ We would like to thank the following people and organizations for supporting zre
 
    <div class="fa fa-code" style="width: 1em;"></div>
 
-* |supporter-gold| Prominic.NET, Inc.
-* |supporter-std| Torsten Blum
 * |supporter-gold| Cyberiada GmbH
 * |supporter-std| `Gordon Schulz <https://github.com/azmodude>`_
 * |supporter-std| `@jwittlincohen <https://github.com/jwittlincohen>`_

docs/usage.rst

@@ -78,73 +78,3 @@ 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`.
-
-
-
-============
-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

@@ -1,36 +0,0 @@
-
-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

@@ -14,7 +14,6 @@ import (
 	"github.com/zrepl/zrepl/daemon/logging/trace"
 
 	"github.com/zrepl/zrepl/replication/logic/pdu"
-	"github.com/zrepl/zrepl/util/bandwidthlimit"
 	"github.com/zrepl/zrepl/util/chainedio"
 	"github.com/zrepl/zrepl/util/chainlock"
 	"github.com/zrepl/zrepl/util/envconst"
@@ -35,8 +34,6 @@ type SenderConfig struct {
 	SendCompressed       bool
 	SendEmbeddedData     bool
 	SendSaved            bool
-
-	BandwidthLimit bandwidthlimit.Config
 }
 
 func (c *SenderConfig) Validate() error {
@@ -47,9 +44,6 @@ func (c *SenderConfig) Validate() error {
 	if _, err := StepHoldTag(c.JobID); err != nil {
 		return fmt.Errorf("JobID cannot be used for hold tag: %s", err)
 	}
-	if err := bandwidthlimit.ValidateConfig(c.BandwidthLimit); err != nil {
-		return errors.Wrap(err, "`BandwidthLimit` field invalid")
-	}
 	return nil
 }
 
@@ -60,21 +54,16 @@ type Sender struct {
 	FSFilter zfs.DatasetFilter
 	jobId    JobID
 	config   SenderConfig
-	bwLimit  bandwidthlimit.Wrapper
 }
 
 func NewSender(conf SenderConfig) *Sender {
 	if err := conf.Validate(); err != nil {
 		panic("invalid config" + err.Error())
 	}
-
-	ratelimiter := bandwidthlimit.WrapperFromConfig(conf.BandwidthLimit)
-
 	return &Sender{
 		FSFilter: conf.FSF,
 		jobId:    conf.JobID,
 		config:   conf,
-		bwLimit:  ratelimiter,
 	}
 }
 
@@ -159,11 +148,12 @@ func sendArgsFromPDUAndValidateExistsAndGetVersion(ctx context.Context, fs strin
 	return version, nil
 }
 
-func (s *Sender) sendMakeArgs(ctx context.Context, r *pdu.SendReq) (sendArgs zfs.ZFSSendArgsValidated, _ error) {
+func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 
 	_, err := s.filterCheckFS(r.Filesystem)
 	if err != nil {
-		return sendArgs, err
+		return nil, nil, err
 	}
 	switch r.Encrypted {
 	case pdu.Tri_DontCare:
@@ -171,16 +161,16 @@ func (s *Sender) sendMakeArgs(ctx context.Context, r *pdu.SendReq) (sendArgs zfs
 		// ok, fallthrough outer
 	case pdu.Tri_False:
 		if s.config.Encrypt.B {
-			return sendArgs, errors.New("only encrypted sends allowed (send -w + encryption!= off), but unencrypted send requested")
+			return nil, nil, errors.New("only encrypted sends allowed (send -w + encryption!= off), but unencrypted send requested")
 		}
 		// fallthrough outer
 	case pdu.Tri_True:
 		if !s.config.Encrypt.B {
-			return sendArgs, errors.New("only unencrypted sends allowed, but encrypted send requested")
+			return nil, nil, errors.New("only unencrypted sends allowed, but encrypted send requested")
 		}
 		// fallthrough outer
 	default:
-		return sendArgs, fmt.Errorf("unknown pdu.Tri variant %q", r.Encrypted)
+		return nil, nil, fmt.Errorf("unknown pdu.Tri variant %q", r.Encrypted)
 	}
 
 	sendArgsUnvalidated := zfs.ZFSSendArgsUnvalidated{
@@ -200,19 +190,30 @@ func (s *Sender) sendMakeArgs(ctx context.Context, r *pdu.SendReq) (sendArgs zfs
 		},
 	}
 
-	sendArgs, err = sendArgsUnvalidated.Validate(ctx)
+	sendArgs, err := sendArgsUnvalidated.Validate(ctx)
 	if err != nil {
-		return sendArgs, errors.Wrap(err, "validate send arguments")
+		return nil, nil, errors.Wrap(err, "validate send arguments")
 	}
-	return sendArgs, nil
-}
 
-func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
+	si, err := zfs.ZFSSendDry(ctx, sendArgs)
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-
-	sendArgs, err := s.sendMakeArgs(ctx, r)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, errors.Wrap(err, "zfs send dry failed")
+	}
+
+	// From now on, assume that sendArgs has been validated by ZFSSendDry
+	// (because validation involves shelling out, it's actually a little expensive)
+
+	var expSize int64 = 0      // protocol says 0 means no estimate
+	if si.SizeEstimate != -1 { // but si returns -1 for no size estimate
+		expSize = si.SizeEstimate
+	}
+	res := &pdu.SendRes{
+		ExpectedSize:    expSize,
+		UsedResumeToken: r.ResumeToken != "",
+	}
+
+	if r.DryRun {
+		return res, nil, nil
 	}
 
 	// create holds or bookmarks of `From` and `To` to guarantee one of the following:
@@ -300,47 +301,15 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
 		abstractionsCacheSingleton.TryBatchDestroy(ctx, s.jobId, sendArgs.FS, destroyTypes, keep, check)
 	}()
 
-	var sendStream io.ReadCloser
+	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
-	sendStream, err = zfs.ZFSSend(ctx, sendArgs)
 	if err != nil {
 		// it's ok to not destroy the abstractions we just created here, a new send attempt will take care of it
 		return nil, nil, errors.Wrap(err, "zfs send failed")
 	}
 
-	// apply rate limit
-	sendStream = s.bwLimit.WrapReadCloser(sendStream)
-
-	res := &pdu.SendRes{
-		ExpectedSize:    0,
-		UsedResumeToken: r.ResumeToken != "",
-	}
-
 	return res, sendStream, nil
 }
 
-func (s *Sender) SendDry(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-
-	sendArgs, err := s.sendMakeArgs(ctx, r)
-	if err != nil {
-		return nil, err
-	}
-
-	si, err := zfs.ZFSSendDry(ctx, sendArgs)
-	if err != nil {
-		return nil, errors.Wrap(err, "zfs send dry failed")
-	}
-
-	// From now on, assume that sendArgs has been validated by ZFSSendDry
-	// (because validation involves shelling out, it's actually a little expensive)
-
-	res := &pdu.SendRes{
-		ExpectedSize:    si.SizeEstimate,
-		UsedResumeToken: r.ResumeToken != "",
-	}
-	return res, nil
-}
-
 func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
 	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 
@@ -470,22 +439,8 @@ type ReceiverConfig struct {
 
 	InheritProperties  []zfsprop.Property
 	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()
 
@@ -520,15 +475,6 @@ func (c *ReceiverConfig) Validate() error {
 	if c.RootWithoutClientComponent.Length() <= 0 {
 		return errors.New("RootWithoutClientComponent must not be an empty dataset path")
 	}
-
-	if err := bandwidthlimit.ValidateConfig(c.BandwidthLimit); err != nil {
-		return errors.Wrap(err, "`BandwidthLimit` field invalid")
-	}
-
-	if !c.PlaceholderEncryption.IsAPlaceholderCreationEncryptionProperty() {
-		return errors.Errorf("`PlaceholderEncryption` field is invalid")
-	}
-
 	return nil
 }
 
@@ -538,11 +484,7 @@ type Receiver struct {
 
 	conf ReceiverConfig // validated
 
-	bwLimit bandwidthlimit.Wrapper
-
 	recvParentCreationMtx *chainlock.L
-
-	Test_OverrideClientIdentityFunc func() string // for use by platformtest
 }
 
 func NewReceiver(config ReceiverConfig) *Receiver {
@@ -553,7 +495,6 @@ func NewReceiver(config ReceiverConfig) *Receiver {
 	return &Receiver{
 		conf:                  config,
 		recvParentCreationMtx: chainlock.New(),
-		bwLimit:               bandwidthlimit.WrapperFromConfig(config.BandwidthLimit),
 	}
 }
 
@@ -580,15 +521,9 @@ func (s *Receiver) clientRootFromCtx(ctx context.Context) *zfs.DatasetPath {
 		return s.conf.RootWithoutClientComponent.Copy()
 	}
 
-	var clientIdentity string
+	clientIdentity, ok := ctx.Value(ClientIdentityKey).(string)
-	if s.Test_OverrideClientIdentityFunc != nil {
+	if !ok {
-		clientIdentity = s.Test_OverrideClientIdentityFunc()
+		panic("ClientIdentityKey context value must be set")
-	} 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)
@@ -629,8 +564,7 @@ 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 {
-		getLogger(ctx).WithField("root_fs", s.conf.RootWithoutClientComponent).Error("root_fs does not exist")
+		return nil, errors.New("root_fs does not exist")
-		return nil, errors.Errorf("root_fs does not exist")
 	}
 
 	root := s.clientRootFromCtx(ctx)
@@ -734,38 +668,6 @@ func (s *Receiver) Send(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, io
 	return nil, nil, fmt.Errorf("receiver does not implement Send()")
 }
 
-func (s *Receiver) SendDry(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-	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)()
 
@@ -806,18 +708,15 @@ 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)
-			l.WithField("placeholder_state", fmt.Sprintf("%#v", ph)).
+			getLogger(ctx).
+				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("get placeholder state for filesystem")
+				Debug("placeholder state for filesystem")
 			if err != nil {
-				visitErr = errors.Wrapf(err, "cannot get placeholder state of %s", v.Path.ToString())
+				visitErr = err
 				return false
 			}
 
@@ -828,34 +727,21 @@ 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())
 					}
-					l.WithError(visitErr).Error("placeholders are only created automatically below root_fs")
+					getLogger(ctx).WithError(visitErr).Error("placeholders are only created automatically below root_fs")
 					return false
 				}
-
+				l := getLogger(ctx).WithField("placeholder_fs", v.Path)
-				// compute the value lazily so that users who don't rely on
+				l.Debug("create placeholder filesystem")
-				// placeholders can use the default value PlaceholderCreationEncryptionPropertyUnspecified
+				err := zfs.ZFSCreatePlaceholderFilesystem(ctx, v.Path, v.Parent.Path)
-				placeholderEncryption, err := s.receive_GetPlaceholderCreationEncryptionValue(root, v.Path)
 				if err != nil {
-					l.WithError(err).Error("cannot create placeholder filesystem") // logger already contains path
+					l.WithError(err).Error("cannot create placeholder filesystem")
-					visitErr = errors.Wrapf(err, "cannot create placeholder filesystem %s", v.Path.ToString())
+					visitErr = err
 					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")
@@ -901,9 +787,6 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.
 		return nil, errors.Wrap(err, "cannot determine whether we can use resumable send & recv")
 	}
 
-	// apply rate limit
-	receive = s.bwLimit.WrapReadCloser(receive)
-
 	var peek bytes.Buffer
 	var MaxPeek = envconst.Int64("ZREPL_ENDPOINT_RECV_PEEK_SIZE", 1<<20)
 	log.WithField("max_peek_bytes", MaxPeek).Info("peeking incoming stream")

endpoint/placeholdercreationencryptionproperty_enumer.go

@@ -1,62 +0,0 @@
-// 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
-}

go.mod

@@ -8,13 +8,13 @@ require (
 	github.com/gdamore/tcell/v2 v2.2.0
 	github.com/gitchander/permutation v0.0.0-20181107151852-9e56b92e9909
 	github.com/go-logfmt/logfmt v0.4.0
+	github.com/go-playground/universal-translator v0.17.0 // indirect
 	github.com/go-playground/validator v9.31.0+incompatible
 	github.com/go-playground/validator/v10 v10.4.1
 	github.com/go-sql-driver/mysql v1.4.1-0.20190907122137-b2c03bcae3d4
 	github.com/golang/protobuf v1.4.3
 	github.com/google/uuid v1.1.2
 	github.com/jinzhu/copier v0.0.0-20170922082739-db4671f3a9b8
-	github.com/juju/ratelimit v1.0.1
 	github.com/kisielk/gotool v1.0.0 // indirect
 	github.com/kr/pretty v0.1.0
 	github.com/leodido/go-urn v1.2.1 // indirect
@@ -43,6 +43,7 @@ require (
 	golang.org/x/net v0.0.0-20210119194325-5f4716e94777
 	golang.org/x/sync v0.0.0-20190423024810-112230192c58
 	golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c
+	golang.org/x/text v0.3.5 // indirect
 	golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135
 	gonum.org/v1/gonum v0.7.0 // indirect
 	google.golang.org/genproto v0.0.0-20210122163508-8081c04a3579 // indirect

go.sum

@@ -164,8 +164,6 @@ github.com/jinzhu/copier v0.0.0-20170922082739-db4671f3a9b8 h1:+dKzeuiDYbD/Cfi/s
 github.com/jinzhu/copier v0.0.0-20170922082739-db4671f3a9b8/go.mod h1:yL958EeXv8Ylng6IfnvG4oflryUi3vgA3xPs9hmII1s=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/json-iterator/go v1.1.7/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
-github.com/juju/ratelimit v1.0.1 h1:+7AIFJVQ0EQgq/K9+0Krm7m530Du7tIz0METWzN0RgY=
-github.com/juju/ratelimit v1.0.1/go.mod h1:qapgC/Gy+xNh9UxzV13HGGl/6UXNN+ct+vwSgWNm/qk=
 github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
 github.com/jung-kurt/gofpdf v1.0.3-0.20190309125859-24315acbbda5/go.mod h1:7Id9E/uU8ce6rXgefFLlgrJj/GYY22cpxn+r32jIOes=
 github.com/k0kubun/colorstring v0.0.0-20150214042306-9440f1994b88 h1:uC1QfSlInpQF+M0ao65imhwqKnz3Q2z/d8PWZRMQvDM=

packaging/deb/Dockerfile

@@ -1,13 +1,9 @@
 FROM debian:latest
 
-# binutils are for cross-compilation to work in bullseye
 RUN apt-get update && apt-get install -y \
 	build-essential \
     devscripts \
-    dh-exec \
+    dh-exec
-	binutils-aarch64-linux-gnu \
-	binutils-arm-linux-gnueabihf  \
-	binutils-i686-linux-gnu
 
 RUN mkdir -p /build/src && chmod -R 0777 /build
 

platformtest/harness/harness.go

@@ -37,7 +37,6 @@ 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,17 +5,12 @@ 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
 }
@@ -98,23 +93,8 @@ func (p *Zpool) Name() string { return p.args.PoolName }
 
 func (p *Zpool) Destroy(ctx context.Context, e Execer) error {
 
-	exportDeadline := time.Now().Add(ZpoolExportTimeout)
+	if err := e.RunExpectSuccessNoOutput(ctx, "zpool", "export", p.args.PoolName); err != nil {
-
+		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,9 +24,6 @@ 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,
@@ -37,10 +34,5 @@ var Cases = []Case{BatchDestroy,
 	SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden__EncryptionSupported_true,
 	SendArgsValidationResumeTokenDifferentFilesystemForbidden,
 	SendArgsValidationResumeTokenEncryptionMismatchForbidden,
-	SendStreamCloseAfterBlockedOnPipeWrite,
-	SendStreamCloseAfterEOFRead,
-	SendStreamMultipleCloseAfterEOF,
-	SendStreamMultipleCloseBeforeEOF,
-	SendStreamNonEOFReadErrorHandling,
 	UndestroyableSnapshotParsing,
 }

platformtest/tests/helpers.go

@@ -10,7 +10,6 @@ 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"
@@ -175,8 +174,3 @@ 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/recvRollback.go

@@ -40,7 +40,6 @@ func ReceiveForceRollbackWorksUnencrypted(ctx *platformtest.Context) {
 
 	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
 	require.NoError(ctx, err)
-	defer sendStream.Close()
 
 	recvOpts := zfs.RecvOptions{
 		RollbackAndForceRecv: true,

platformtest/tests/replication.go

@@ -7,7 +7,6 @@ import (
 	"os"
 	"path"
 	"sort"
-	"strings"
 	"time"
 
 	"github.com/kr/pretty"
@@ -21,7 +20,6 @@ import (
 	"github.com/zrepl/zrepl/replication/logic"
 	"github.com/zrepl/zrepl/replication/logic/pdu"
 	"github.com/zrepl/zrepl/replication/report"
-	"github.com/zrepl/zrepl/util/bandwidthlimit"
 	"github.com/zrepl/zrepl/util/limitio"
 	"github.com/zrepl/zrepl/util/nodefault"
 	"github.com/zrepl/zrepl/zfs"
@@ -65,10 +63,9 @@ func (i replicationInvocation) Do(ctx *platformtest.Context) *report.Report {
 	}
 
 	senderConfig := endpoint.SenderConfig{
-		FSF:            i.sfilter.AsFilter(),
+		FSF:     i.sfilter.AsFilter(),
-		Encrypt:        &nodefault.Bool{B: false},
+		Encrypt: &nodefault.Bool{B: false},
-		JobID:          i.sjid,
+		JobID:   i.sjid,
-		BandwidthLimit: bandwidthlimit.NoLimitConfig(),
 	}
 	if i.senderConfigHook != nil {
 		i.senderConfigHook(&senderConfig)
@@ -78,8 +75,6 @@ func (i replicationInvocation) Do(ctx *platformtest.Context) *report.Report {
 		JobID:                      i.rjid,
 		AppendClientIdentity:       false,
 		RootWithoutClientComponent: mustDatasetPath(i.rfsRoot),
-		BandwidthLimit:             bandwidthlimit.NoLimitConfig(),
-		PlaceholderEncryption:      endpoint.PlaceholderCreationEncryptionPropertyUnspecified,
 	}
 	if i.receiverConfigHook != nil {
 		i.receiverConfigHook(&receiverConfig)
@@ -95,7 +90,6 @@ func (i replicationInvocation) Do(ctx *platformtest.Context) *report.Report {
 		ReplicationConfig: &pdu.ReplicationConfig{
 			Protection: i.guarantee,
 		},
-		SizeEstimationConcurrency: 1,
 	}
 
 	report, wait := replication.Do(
@@ -810,22 +804,13 @@ type NeverEndingSender struct {
 	*endpoint.Sender
 }
 
-func (s *NeverEndingSender) SendDry(ctx context.Context, req *pdu.SendReq) (r *pdu.SendRes, err error) {
-	r, _, err = s.sendImpl(ctx, req, true)
-	return r, err
-}
-
 func (s *NeverEndingSender) Send(ctx context.Context, req *pdu.SendReq) (r *pdu.SendRes, stream io.ReadCloser, _ error) {
-	return s.sendImpl(ctx, req, false)
-}
-
-func (s *NeverEndingSender) sendImpl(ctx context.Context, req *pdu.SendReq, dry bool) (r *pdu.SendRes, stream io.ReadCloser, _ error) {
 	stream = nil
 	r = &pdu.SendRes{
 		UsedResumeToken: false,
 		ExpectedSize:    1 << 30,
 	}
-	if dry {
+	if req.DryRun {
 		return r, stream, nil
 	}
 	dz, err := os.Open("/dev/zero")
@@ -904,9 +889,13 @@ func ReplicationFailingInitialParentProhibitsChildReplication(ctx *platformtest.
 	fsAA := ctx.RootDataset + "/sender/aa"
 
 	sfilter := filters.NewDatasetMapFilter(3, true)
-	mustAddToSFilter(ctx, sfilter, fsA)
+	mustAddToSFilter := func(fs string) {
-	mustAddToSFilter(ctx, sfilter, fsAChild)
+		err := sfilter.Add(fs, "ok")
-	mustAddToSFilter(ctx, sfilter, fsAA)
+		require.NoError(ctx, err)
+	}
+	mustAddToSFilter(fsA)
+	mustAddToSFilter(fsAChild)
+	mustAddToSFilter(fsAA)
 	rfsRoot := ctx.RootDataset + "/receiver"
 
 	mockRecvErr := fmt.Errorf("yifae4ohPhaquaes0hohghiep9oufie4roo7quoWooluaj2ee8")
@@ -962,7 +951,6 @@ 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")
@@ -972,8 +960,12 @@ func ReplicationPropertyReplicationWorks(ctx *platformtest.Context) {
 	fsAChild := ctx.RootDataset + "/sender/a/child"
 
 	sfilter := filters.NewDatasetMapFilter(2, true)
-	mustAddToSFilter(ctx, sfilter, fsA)
+	mustAddToSFilter := func(fs string) {
-	mustAddToSFilter(ctx, sfilter, fsAChild)
+		err := sfilter.Add(fs, "ok")
+		require.NoError(ctx, err)
+	}
+	mustAddToSFilter(fsA)
+	mustAddToSFilter(fsAChild)
 	rfsRoot := ctx.RootDataset + "/receiver"
 
 	type testPropExpectation struct {
@@ -1062,9 +1054,6 @@ func ReplicationPropertyReplicationWorks(ctx *platformtest.Context) {
 		rep := fsByName[fs]
 		require.Len(ctx, rep.Steps, 1)
 		require.Nil(ctx, rep.PlanError)
-		if rep.StepError != nil && strings.Contains(rep.StepError.Error(), "invalid option 'x'") {
-			ctx.SkipNow() // XXX feature detection
-		}
 		require.Nil(ctx, rep.StepError)
 		require.Len(ctx, rep.Steps, 1)
 		require.Equal(ctx, 1, rep.CurrentStep)
@@ -1101,208 +1090,3 @@ 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"))
-}

platformtest/tests/sendStream.go

@@ -1,186 +0,0 @@
-package tests
-
-import (
-	"fmt"
-	"io"
-	"io/ioutil"
-	"os"
-	"os/exec"
-	"path"
-	"syscall"
-	"time"
-
-	"github.com/stretchr/testify/require"
-	"golang.org/x/sys/unix"
-
-	"github.com/zrepl/zrepl/platformtest"
-	"github.com/zrepl/zrepl/util/nodefault"
-	"github.com/zrepl/zrepl/zfs"
-)
-
-func sendStreamTest(ctx *platformtest.Context) *zfs.SendStream {
-
-	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
-		DESTROYROOT
-		CREATEROOT
-		+  "sender"
-	`)
-
-	fs := fmt.Sprintf("%s/sender", ctx.RootDataset)
-
-	fsmpo, err := zfs.ZFSGetMountpoint(ctx, fs)
-	require.NoError(ctx, err)
-
-	writeDummyData(path.Join(fsmpo.Mountpoint, "dummy.data"), 1<<26)
-	mustSnapshot(ctx, fs+"@1")
-	snap := fsversion(ctx, fs, "@1")
-	snapSendArg := snap.ToSendArgVersion()
-
-	sendArgs, err := zfs.ZFSSendArgsUnvalidated{
-		FS:   fs,
-		From: nil,
-		To:   &snapSendArg,
-		ZFSSendFlags: zfs.ZFSSendFlags{
-			Encrypted: &nodefault.Bool{B: false},
-		},
-	}.Validate(ctx)
-	require.NoError(ctx, err)
-
-	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
-	require.NoError(ctx, err)
-
-	return sendStream
-
-}
-
-func SendStreamCloseAfterBlockedOnPipeWrite(ctx *platformtest.Context) {
-
-	sendStream := sendStreamTest(ctx)
-
-	// let the pipe buffer fill and the zfs process block uninterruptibly
-
-	ctx.Logf("waiting for pipe write to block")
-	time.Sleep(5 * time.Second) // XXX need a platform-neutral way to detect that the pipe is full and the writer is blocked
-
-	ctx.Logf("closing send stream")
-	err := sendStream.Close() // this is what this test case is about
-	ctx.Logf("close error: %T %s", err, err)
-	require.NoError(ctx, err)
-
-	exitErrZfsError := sendStream.TestOnly_ExitErr()
-	require.Contains(ctx, exitErrZfsError.Error(), "signal")
-	require.Error(ctx, exitErrZfsError)
-	exitErr, ok := exitErrZfsError.WaitErr.(*exec.ExitError)
-	require.True(ctx, ok)
-	if exitErr.Exited() {
-		// some ZFS impls (FreeBSD 12) behaves that way
-		return
-	}
-
-	// ProcessState is only available after exit
-	// => use as proxy that the process was wait()ed upon and is gone
-	ctx.Logf("%#v", exitErr.ProcessState)
-	require.NotNil(ctx, exitErr.ProcessState)
-	// and let's verify that the process got killed, so that we know it was the call to .Close() above
-	waitStatus := exitErr.ProcessState.Sys().(syscall.WaitStatus)
-	ctx.Logf("wait status: %#v", waitStatus)
-	ctx.Logf("exit status: %v", waitStatus.ExitStatus())
-	require.True(ctx, waitStatus.Signaled())
-	switch waitStatus.Signal() {
-	case unix.SIGKILL:
-		fallthrough
-	case unix.SIGPIPE:
-		// ok
-
-	default:
-		ctx.Errorf("%T %s\n%v", waitStatus.Signal(), waitStatus.Signal(), waitStatus.Signal())
-		ctx.FailNow()
-	}
-}
-
-func SendStreamCloseAfterEOFRead(ctx *platformtest.Context) {
-
-	sendStream := sendStreamTest(ctx)
-
-	_, err := io.Copy(ioutil.Discard, sendStream)
-	require.NoError(ctx, err)
-
-	var buf [128]byte
-	n, err := sendStream.Read(buf[:])
-	require.Zero(ctx, n)
-	require.Equal(ctx, io.EOF, err)
-
-	err = sendStream.Close()
-	require.NoError(ctx, err)
-
-	n, err = sendStream.Read(buf[:])
-	require.Zero(ctx, n)
-	require.Equal(ctx, os.ErrClosed, err, "same read error should be returned")
-}
-
-func SendStreamMultipleCloseAfterEOF(ctx *platformtest.Context) {
-
-	sendStream := sendStreamTest(ctx)
-
-	_, err := io.Copy(ioutil.Discard, sendStream)
-	require.NoError(ctx, err)
-
-	var buf [128]byte
-	n, err := sendStream.Read(buf[:])
-	require.Zero(ctx, n)
-	require.Equal(ctx, io.EOF, err)
-
-	err = sendStream.Close()
-	require.NoError(ctx, err)
-
-	err = sendStream.Close()
-	require.Equal(ctx, os.ErrClosed, err)
-}
-
-func SendStreamMultipleCloseBeforeEOF(ctx *platformtest.Context) {
-
-	sendStream := sendStreamTest(ctx)
-
-	err := sendStream.Close()
-	require.NoError(ctx, err)
-
-	err = sendStream.Close()
-	require.Equal(ctx, os.ErrClosed, err)
-}
-
-type failingReadCloser struct {
-	err error
-}
-
-var _ io.ReadCloser = &failingReadCloser{}
-
-func (c *failingReadCloser) Read(p []byte) (int, error) { return 0, c.err }
-func (c *failingReadCloser) Close() error               { return c.err }
-
-func SendStreamNonEOFReadErrorHandling(ctx *platformtest.Context) {
-
-	sendStream := sendStreamTest(ctx)
-
-	var buf [128]byte
-	n, err := sendStream.Read(buf[:])
-	require.Equal(ctx, len(buf), n)
-	require.NoError(ctx, err)
-
-	var mockError = fmt.Errorf("taeghaefow4piesahwahjocu7ul5tiachaiLipheijae8ooZ8Pies8shohGee9feeTeirai5aiFeiyaecai4kiaLoh4azeih0tea")
-	mock := &failingReadCloser{err: mockError}
-	orig := sendStream.TestOnly_ReplaceStdoutReader(mock)
-
-	n, err = sendStream.Read(buf[:])
-	require.Equal(ctx, 0, n)
-	require.Equal(ctx, mockError, err)
-
-	if sendStream.TestOnly_ReplaceStdoutReader(orig) != mock {
-		panic("incorrect test impl")
-	}
-
-	err = sendStream.Close()
-	require.NoError(ctx, err) // if we can't kill the child then this will be a flaky test, but let's assume we can kill the child
-
-	err = sendStream.Close()
-	require.Equal(ctx, os.ErrClosed, err)
-}

replication/driver/replication_driver.go

@@ -151,8 +151,6 @@ type fs struct {
 
 	l *chainlock.L
 
-	blockedOn report.FsBlockedOn
-
 	// ordering relationship that must be maintained for initial replication
 	initialRepOrd struct {
 		parents, children []*fs
@@ -160,9 +158,8 @@ type fs struct {
 	}
 
 	planning struct {
-		waitingForStepQueue bool
+		done bool
-		done                bool
+		err  *timedError
-		err                 *timedError
 	}
 
 	// valid iff planning.done && planning.err == nil
@@ -340,9 +337,8 @@ 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)
@@ -452,10 +448,6 @@ 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() {
@@ -494,16 +486,11 @@ 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
 	})
@@ -558,7 +545,6 @@ 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)
@@ -627,6 +613,7 @@ 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
 			}
@@ -644,9 +631,7 @@ 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()
@@ -740,7 +725,6 @@ 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,12 +14,11 @@ 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
@@ -91,8 +90,6 @@ 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/logic/pdu/pdu.pb.go

@@ -518,7 +518,8 @@ type SendReq struct {
 	// encoded in the ResumeToken. Otherwise, the Sender MUST return an error.
 	ResumeToken       string             `protobuf:"bytes,4,opt,name=ResumeToken,proto3" json:"ResumeToken,omitempty"`
 	Encrypted         Tri                `protobuf:"varint,5,opt,name=Encrypted,proto3,enum=Tri" json:"Encrypted,omitempty"`
-	ReplicationConfig *ReplicationConfig `protobuf:"bytes,6,opt,name=ReplicationConfig,proto3" json:"ReplicationConfig,omitempty"`
+	DryRun            bool               `protobuf:"varint,6,opt,name=DryRun,proto3" json:"DryRun,omitempty"`
+	ReplicationConfig *ReplicationConfig `protobuf:"bytes,7,opt,name=ReplicationConfig,proto3" json:"ReplicationConfig,omitempty"`
 }
 
 func (x *SendReq) Reset() {
@@ -588,6 +589,13 @@ func (x *SendReq) GetEncrypted() Tri {
 	return Tri_DontCare
 }
 
+func (x *SendReq) GetDryRun() bool {
+	if x != nil {
+		return x.DryRun
+	}
+	return false
+}
+
 func (x *SendReq) GetReplicationConfig() *ReplicationConfig {
 	if x != nil {
 		return x.ReplicationConfig
@@ -758,11 +766,12 @@ type SendRes struct {
 	unknownFields protoimpl.UnknownFields
 
 	// Whether the resume token provided in the request has been used or not.
-	// If the SendReq.ResumeToken == "", this field MUST be false.
+	// If the SendReq.ResumeToken == "", this field has no meaning.
-	UsedResumeToken bool `protobuf:"varint,1,opt,name=UsedResumeToken,proto3" json:"UsedResumeToken,omitempty"`
+	UsedResumeToken bool `protobuf:"varint,2,opt,name=UsedResumeToken,proto3" json:"UsedResumeToken,omitempty"`
 	// Expected stream size determined by dry run, not exact.
 	// 0 indicates that for the given SendReq, no size estimate could be made.
-	ExpectedSize uint64 `protobuf:"varint,2,opt,name=ExpectedSize,proto3" json:"ExpectedSize,omitempty"`
+	ExpectedSize int64       `protobuf:"varint,3,opt,name=ExpectedSize,proto3" json:"ExpectedSize,omitempty"`
+	Properties   []*Property `protobuf:"bytes,4,rep,name=Properties,proto3" json:"Properties,omitempty"`
 }
 
 func (x *SendRes) Reset() {
@@ -804,13 +813,20 @@ func (x *SendRes) GetUsedResumeToken() bool {
 	return false
 }
 
-func (x *SendRes) GetExpectedSize() uint64 {
+func (x *SendRes) GetExpectedSize() int64 {
 	if x != nil {
 		return x.ExpectedSize
 	}
 	return 0
 }
 
+func (x *SendRes) GetProperties() []*Property {
+	if x != nil {
+		return x.Properties
+	}
+	return nil
+}
+
 type SendCompletedReq struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -1427,7 +1443,7 @@ var file_pdu_proto_rawDesc = []byte{
 	0x72, 0x65, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x22, 0x29, 0x0a, 0x0b, 0x56, 0x65, 0x72, 0x73, 0x69,
 	0x6f, 0x6e, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0c, 0x0a, 0x08, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68,
 	0x6f, 0x74, 0x10, 0x00, 0x12, 0x0c, 0x0a, 0x08, 0x42, 0x6f, 0x6f, 0x6b, 0x6d, 0x61, 0x72, 0x6b,
-	0x10, 0x01, 0x22, 0xfd, 0x01, 0x0a, 0x07, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65, 0x71, 0x12, 0x1e,
+	0x10, 0x01, 0x22, 0x95, 0x02, 0x0a, 0x07, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65, 0x71, 0x12, 0x1e,
 	0x0a, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x18, 0x01, 0x20, 0x01,
 	0x28, 0x09, 0x52, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x12, 0x26,
 	0x0a, 0x04, 0x46, 0x72, 0x6f, 0x6d, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x46,
@@ -1439,119 +1455,122 @@ var file_pdu_proto_rawDesc = []byte{
 	0x0b, 0x52, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x54, 0x6f, 0x6b, 0x65, 0x6e, 0x12, 0x22, 0x0a, 0x09,
 	0x45, 0x6e, 0x63, 0x72, 0x79, 0x70, 0x74, 0x65, 0x64, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0e, 0x32,
 	0x04, 0x2e, 0x54, 0x72, 0x69, 0x52, 0x09, 0x45, 0x6e, 0x63, 0x72, 0x79, 0x70, 0x74, 0x65, 0x64,
-	0x12, 0x40, 0x0a, 0x11, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43,
+	0x12, 0x16, 0x0a, 0x06, 0x44, 0x72, 0x79, 0x52, 0x75, 0x6e, 0x18, 0x06, 0x20, 0x01, 0x28, 0x08,
-	0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x52, 0x65,
+	0x52, 0x06, 0x44, 0x72, 0x79, 0x52, 0x75, 0x6e, 0x12, 0x40, 0x0a, 0x11, 0x52, 0x65, 0x70, 0x6c,
-	0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52,
+	0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x07, 0x20,
+	0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f,
+	0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52, 0x11, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61,
+	0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x22, 0x51, 0x0a, 0x11, 0x52, 0x65,
+	0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12,
+	0x3c, 0x0a, 0x0a, 0x70, 0x72, 0x6f, 0x74, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20,
+	0x01, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f,
+	0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x50, 0x72, 0x6f, 0x74, 0x65, 0x63, 0x74, 0x69, 0x6f,
+	0x6e, 0x52, 0x0a, 0x70, 0x72, 0x6f, 0x74, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x22, 0x8f, 0x01,
+	0x0a, 0x1b, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e,
+	0x66, 0x69, 0x67, 0x50, 0x72, 0x6f, 0x74, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x33, 0x0a,
+	0x07, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x6c, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x19,
+	0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x47, 0x75, 0x61, 0x72,
+	0x61, 0x6e, 0x74, 0x65, 0x65, 0x4b, 0x69, 0x6e, 0x64, 0x52, 0x07, 0x49, 0x6e, 0x69, 0x74, 0x69,
+	0x61, 0x6c, 0x12, 0x3b, 0x0a, 0x0b, 0x49, 0x6e, 0x63, 0x72, 0x65, 0x6d, 0x65, 0x6e, 0x74, 0x61,
+	0x6c, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x19, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63,
+	0x61, 0x74, 0x69, 0x6f, 0x6e, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x4b, 0x69,
+	0x6e, 0x64, 0x52, 0x0b, 0x49, 0x6e, 0x63, 0x72, 0x65, 0x6d, 0x65, 0x6e, 0x74, 0x61, 0x6c, 0x22,
+	0x34, 0x0a, 0x08, 0x50, 0x72, 0x6f, 0x70, 0x65, 0x72, 0x74, 0x79, 0x12, 0x12, 0x0a, 0x04, 0x4e,
+	0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x4e, 0x61, 0x6d, 0x65, 0x12,
+	0x14, 0x0a, 0x05, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05,
+	0x56, 0x61, 0x6c, 0x75, 0x65, 0x22, 0x82, 0x01, 0x0a, 0x07, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65,
+	0x73, 0x12, 0x28, 0x0a, 0x0f, 0x55, 0x73, 0x65, 0x64, 0x52, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x54,
+	0x6f, 0x6b, 0x65, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x08, 0x52, 0x0f, 0x55, 0x73, 0x65, 0x64,
+	0x52, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x54, 0x6f, 0x6b, 0x65, 0x6e, 0x12, 0x22, 0x0a, 0x0c, 0x45,
+	0x78, 0x70, 0x65, 0x63, 0x74, 0x65, 0x64, 0x53, 0x69, 0x7a, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28,
+	0x03, 0x52, 0x0c, 0x45, 0x78, 0x70, 0x65, 0x63, 0x74, 0x65, 0x64, 0x53, 0x69, 0x7a, 0x65, 0x12,
+	0x29, 0x0a, 0x0a, 0x50, 0x72, 0x6f, 0x70, 0x65, 0x72, 0x74, 0x69, 0x65, 0x73, 0x18, 0x04, 0x20,
+	0x03, 0x28, 0x0b, 0x32, 0x09, 0x2e, 0x50, 0x72, 0x6f, 0x70, 0x65, 0x72, 0x74, 0x79, 0x52, 0x0a,
+	0x50, 0x72, 0x6f, 0x70, 0x65, 0x72, 0x74, 0x69, 0x65, 0x73, 0x22, 0x3e, 0x0a, 0x10, 0x53, 0x65,
+	0x6e, 0x64, 0x43, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x71, 0x12, 0x2a,
+	0x0a, 0x0b, 0x4f, 0x72, 0x69, 0x67, 0x69, 0x6e, 0x61, 0x6c, 0x52, 0x65, 0x71, 0x18, 0x02, 0x20,
+	0x01, 0x28, 0x0b, 0x32, 0x08, 0x2e, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65, 0x71, 0x52, 0x0b, 0x4f,
+	0x72, 0x69, 0x67, 0x69, 0x6e, 0x61, 0x6c, 0x52, 0x65, 0x71, 0x22, 0x12, 0x0a, 0x10, 0x53, 0x65,
+	0x6e, 0x64, 0x43, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x73, 0x22, 0xbe,
+	0x01, 0x0a, 0x0a, 0x52, 0x65, 0x63, 0x65, 0x69, 0x76, 0x65, 0x52, 0x65, 0x71, 0x12, 0x1e, 0x0a,
+	0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28,
+	0x09, 0x52, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x12, 0x22, 0x0a,
+	0x02, 0x54, 0x6f, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x46, 0x69, 0x6c, 0x65,
+	0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x02, 0x54,
+	0x6f, 0x12, 0x2a, 0x0a, 0x10, 0x43, 0x6c, 0x65, 0x61, 0x72, 0x52, 0x65, 0x73, 0x75, 0x6d, 0x65,
+	0x54, 0x6f, 0x6b, 0x65, 0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x08, 0x52, 0x10, 0x43, 0x6c, 0x65,
+	0x61, 0x72, 0x52, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x54, 0x6f, 0x6b, 0x65, 0x6e, 0x12, 0x40, 0x0a,
 	0x11, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66,
-	0x69, 0x67, 0x22, 0x51, 0x0a, 0x11, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f,
+	0x69, 0x67, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69,
-	0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x3c, 0x0a, 0x0a, 0x70, 0x72, 0x6f, 0x74, 0x65,
+	0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52, 0x11, 0x52, 0x65,
-	0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x52, 0x65,
+	0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x22,
-	0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x50,
+	0x0c, 0x0a, 0x0a, 0x52, 0x65, 0x63, 0x65, 0x69, 0x76, 0x65, 0x52, 0x65, 0x73, 0x22, 0x67, 0x0a,
-	0x72, 0x6f, 0x74, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x0a, 0x70, 0x72, 0x6f, 0x74, 0x65,
+	0x13, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74,
-	0x63, 0x74, 0x69, 0x6f, 0x6e, 0x22, 0x8f, 0x01, 0x0a, 0x1b, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63,
+	0x73, 0x52, 0x65, 0x71, 0x12, 0x1e, 0x0a, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74,
-	0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x50, 0x72, 0x6f, 0x74, 0x65,
+	0x65, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79,
-	0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x33, 0x0a, 0x07, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x6c,
+	0x73, 0x74, 0x65, 0x6d, 0x12, 0x30, 0x0a, 0x09, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74,
-	0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x19, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61,
+	0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79,
-	0x74, 0x69, 0x6f, 0x6e, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x4b, 0x69, 0x6e,
+	0x73, 0x74, 0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x09, 0x53, 0x6e, 0x61,
-	0x64, 0x52, 0x07, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x6c, 0x12, 0x3b, 0x0a, 0x0b, 0x49, 0x6e,
+	0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x22, 0x5a, 0x0a, 0x12, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f,
-	0x63, 0x72, 0x65, 0x6d, 0x65, 0x6e, 0x74, 0x61, 0x6c, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0e, 0x32,
+	0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x52, 0x65, 0x73, 0x12, 0x2e, 0x0a, 0x08,
-	0x19, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x47, 0x75, 0x61,
+	0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12,
-	0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x4b, 0x69, 0x6e, 0x64, 0x52, 0x0b, 0x49, 0x6e, 0x63, 0x72,
+	0x2e, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69,
-	0x65, 0x6d, 0x65, 0x6e, 0x74, 0x61, 0x6c, 0x22, 0x34, 0x0a, 0x08, 0x50, 0x72, 0x6f, 0x70, 0x65,
+	0x6f, 0x6e, 0x52, 0x08, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x12, 0x14, 0x0a, 0x05,
-	0x72, 0x74, 0x79, 0x12, 0x12, 0x0a, 0x04, 0x4e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28,
+	0x45, 0x72, 0x72, 0x6f, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x45, 0x72, 0x72,
-	0x09, 0x52, 0x04, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x56, 0x61, 0x6c, 0x75, 0x65,
+	0x6f, 0x72, 0x22, 0x44, 0x0a, 0x13, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61,
-	0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x22, 0x57, 0x0a,
+	0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x52, 0x65, 0x73, 0x12, 0x2d, 0x0a, 0x07, 0x52, 0x65, 0x73,
-	0x07, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65, 0x73, 0x12, 0x28, 0x0a, 0x0f, 0x55, 0x73, 0x65, 0x64,
+	0x75, 0x6c, 0x74, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x44, 0x65, 0x73,
-	0x52, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x54, 0x6f, 0x6b, 0x65, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28,
+	0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x52, 0x65, 0x73, 0x52,
-	0x08, 0x52, 0x0f, 0x55, 0x73, 0x65, 0x64, 0x52, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x54, 0x6f, 0x6b,
+	0x07, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x73, 0x22, 0x36, 0x0a, 0x14, 0x52, 0x65, 0x70, 0x6c,
-	0x65, 0x6e, 0x12, 0x22, 0x0a, 0x0c, 0x45, 0x78, 0x70, 0x65, 0x63, 0x74, 0x65, 0x64, 0x53, 0x69,
+	0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x75, 0x72, 0x73, 0x6f, 0x72, 0x52, 0x65, 0x71,
-	0x7a, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0c, 0x45, 0x78, 0x70, 0x65, 0x63, 0x74,
-	0x65, 0x64, 0x53, 0x69, 0x7a, 0x65, 0x22, 0x3e, 0x0a, 0x10, 0x53, 0x65, 0x6e, 0x64, 0x43, 0x6f,
-	0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x71, 0x12, 0x2a, 0x0a, 0x0b, 0x4f, 0x72,
-	0x69, 0x67, 0x69, 0x6e, 0x61, 0x6c, 0x52, 0x65, 0x71, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32,
-	0x08, 0x2e, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65, 0x71, 0x52, 0x0b, 0x4f, 0x72, 0x69, 0x67, 0x69,
-	0x6e, 0x61, 0x6c, 0x52, 0x65, 0x71, 0x22, 0x12, 0x0a, 0x10, 0x53, 0x65, 0x6e, 0x64, 0x43, 0x6f,
-	0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x73, 0x22, 0xbe, 0x01, 0x0a, 0x0a, 0x52,
-	0x65, 0x63, 0x65, 0x69, 0x76, 0x65, 0x52, 0x65, 0x71, 0x12, 0x1e, 0x0a, 0x0a, 0x46, 0x69, 0x6c,
-	0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x46,
-	0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x12, 0x22, 0x0a, 0x02, 0x54, 0x6f, 0x18,
-	0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74,
-	0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x02, 0x54, 0x6f, 0x12, 0x2a, 0x0a,
-	0x10, 0x43, 0x6c, 0x65, 0x61, 0x72, 0x52, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x54, 0x6f, 0x6b, 0x65,
-	0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x08, 0x52, 0x10, 0x43, 0x6c, 0x65, 0x61, 0x72, 0x52, 0x65,
-	0x73, 0x75, 0x6d, 0x65, 0x54, 0x6f, 0x6b, 0x65, 0x6e, 0x12, 0x40, 0x0a, 0x11, 0x52, 0x65, 0x70,
-	0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x04,
-	0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69,
-	0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52, 0x11, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63,
-	0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x22, 0x0c, 0x0a, 0x0a, 0x52,
-	0x65, 0x63, 0x65, 0x69, 0x76, 0x65, 0x52, 0x65, 0x73, 0x22, 0x67, 0x0a, 0x13, 0x44, 0x65, 0x73,
-	0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x52, 0x65, 0x71,
 	0x12, 0x1e, 0x0a, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x18, 0x01,
 	0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d,
-	0x12, 0x30, 0x0a, 0x09, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x18, 0x02, 0x20,
+	0x22, 0x54, 0x0a, 0x14, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43,
-	0x03, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d,
+	0x75, 0x72, 0x73, 0x6f, 0x72, 0x52, 0x65, 0x73, 0x12, 0x14, 0x0a, 0x04, 0x47, 0x75, 0x69, 0x64,
-	0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x09, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f,
+	0x18, 0x01, 0x20, 0x01, 0x28, 0x04, 0x48, 0x00, 0x52, 0x04, 0x47, 0x75, 0x69, 0x64, 0x12, 0x1c,
-	0x74, 0x73, 0x22, 0x5a, 0x0a, 0x12, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61,
+	0x0a, 0x08, 0x4e, 0x6f, 0x74, 0x65, 0x78, 0x69, 0x73, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x08,
-	0x70, 0x73, 0x68, 0x6f, 0x74, 0x52, 0x65, 0x73, 0x12, 0x2e, 0x0a, 0x08, 0x53, 0x6e, 0x61, 0x70,
+	0x48, 0x00, 0x52, 0x08, 0x4e, 0x6f, 0x74, 0x65, 0x78, 0x69, 0x73, 0x74, 0x42, 0x08, 0x0a, 0x06,
-	0x73, 0x68, 0x6f, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x46, 0x69, 0x6c,
+	0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x22, 0x23, 0x0a, 0x07, 0x50, 0x69, 0x6e, 0x67, 0x52, 0x65,
-	0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x08,
+	0x71, 0x12, 0x18, 0x0a, 0x07, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x01, 0x20, 0x01,
-	0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x12, 0x14, 0x0a, 0x05, 0x45, 0x72, 0x72, 0x6f,
+	0x28, 0x09, 0x52, 0x07, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x1d, 0x0a, 0x07, 0x50,
-	0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x22, 0x44,
+	0x69, 0x6e, 0x67, 0x52, 0x65, 0x73, 0x12, 0x12, 0x0a, 0x04, 0x45, 0x63, 0x68, 0x6f, 0x18, 0x01,
-	0x0a, 0x13, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f,
+	0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x45, 0x63, 0x68, 0x6f, 0x2a, 0x28, 0x0a, 0x03, 0x54, 0x72,
-	0x74, 0x73, 0x52, 0x65, 0x73, 0x12, 0x2d, 0x0a, 0x07, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x73,
+	0x69, 0x12, 0x0c, 0x0a, 0x08, 0x44, 0x6f, 0x6e, 0x74, 0x43, 0x61, 0x72, 0x65, 0x10, 0x00, 0x12,
-	0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79,
+	0x09, 0x0a, 0x05, 0x46, 0x61, 0x6c, 0x73, 0x65, 0x10, 0x01, 0x12, 0x08, 0x0a, 0x04, 0x54, 0x72,
-	0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x52, 0x65, 0x73, 0x52, 0x07, 0x52, 0x65, 0x73,
+	0x75, 0x65, 0x10, 0x02, 0x2a, 0x86, 0x01, 0x0a, 0x18, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61,
-	0x75, 0x6c, 0x74, 0x73, 0x22, 0x36, 0x0a, 0x14, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74,
+	0x74, 0x69, 0x6f, 0x6e, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x4b, 0x69, 0x6e,
-	0x69, 0x6f, 0x6e, 0x43, 0x75, 0x72, 0x73, 0x6f, 0x72, 0x52, 0x65, 0x71, 0x12, 0x1e, 0x0a, 0x0a,
+	0x64, 0x12, 0x14, 0x0a, 0x10, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x49, 0x6e,
-	0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09,
+	0x76, 0x61, 0x6c, 0x69, 0x64, 0x10, 0x00, 0x12, 0x19, 0x0a, 0x15, 0x47, 0x75, 0x61, 0x72, 0x61,
-	0x52, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x22, 0x54, 0x0a, 0x14,
+	0x6e, 0x74, 0x65, 0x65, 0x52, 0x65, 0x73, 0x75, 0x6d, 0x61, 0x62, 0x69, 0x6c, 0x69, 0x74, 0x79,
-	0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x75, 0x72, 0x73, 0x6f,
+	0x10, 0x01, 0x12, 0x23, 0x0a, 0x1f, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x49,
-	0x72, 0x52, 0x65, 0x73, 0x12, 0x14, 0x0a, 0x04, 0x47, 0x75, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01,
+	0x6e, 0x63, 0x72, 0x65, 0x6d, 0x65, 0x6e, 0x74, 0x61, 0x6c, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63,
-	0x28, 0x04, 0x48, 0x00, 0x52, 0x04, 0x47, 0x75, 0x69, 0x64, 0x12, 0x1c, 0x0a, 0x08, 0x4e, 0x6f,
+	0x61, 0x74, 0x69, 0x6f, 0x6e, 0x10, 0x02, 0x12, 0x14, 0x0a, 0x10, 0x47, 0x75, 0x61, 0x72, 0x61,
-	0x74, 0x65, 0x78, 0x69, 0x73, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x08, 0x48, 0x00, 0x52, 0x08,
+	0x6e, 0x74, 0x65, 0x65, 0x4e, 0x6f, 0x74, 0x68, 0x69, 0x6e, 0x67, 0x10, 0x03, 0x32, 0xf0, 0x02,
-	0x4e, 0x6f, 0x74, 0x65, 0x78, 0x69, 0x73, 0x74, 0x42, 0x08, 0x0a, 0x06, 0x52, 0x65, 0x73, 0x75,
+	0x0a, 0x0b, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x1a, 0x0a,
-	0x6c, 0x74, 0x22, 0x23, 0x0a, 0x07, 0x50, 0x69, 0x6e, 0x67, 0x52, 0x65, 0x71, 0x12, 0x18, 0x0a,
+	0x04, 0x50, 0x69, 0x6e, 0x67, 0x12, 0x08, 0x2e, 0x50, 0x69, 0x6e, 0x67, 0x52, 0x65, 0x71, 0x1a,
-	0x07, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07,
+	0x08, 0x2e, 0x50, 0x69, 0x6e, 0x67, 0x52, 0x65, 0x73, 0x12, 0x39, 0x0a, 0x0f, 0x4c, 0x69, 0x73,
-	0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x1d, 0x0a, 0x07, 0x50, 0x69, 0x6e, 0x67, 0x52,
+	0x74, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x73, 0x12, 0x12, 0x2e, 0x4c,
-	0x65, 0x73, 0x12, 0x12, 0x0a, 0x04, 0x45, 0x63, 0x68, 0x6f, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09,
+	0x69, 0x73, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x52, 0x65, 0x71,
-	0x52, 0x04, 0x45, 0x63, 0x68, 0x6f, 0x2a, 0x28, 0x0a, 0x03, 0x54, 0x72, 0x69, 0x12, 0x0c, 0x0a,
+	0x1a, 0x12, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65,
-	0x08, 0x44, 0x6f, 0x6e, 0x74, 0x43, 0x61, 0x72, 0x65, 0x10, 0x00, 0x12, 0x09, 0x0a, 0x05, 0x46,
+	0x6d, 0x52, 0x65, 0x73, 0x12, 0x50, 0x0a, 0x16, 0x4c, 0x69, 0x73, 0x74, 0x46, 0x69, 0x6c, 0x65,
-	0x61, 0x6c, 0x73, 0x65, 0x10, 0x01, 0x12, 0x08, 0x0a, 0x04, 0x54, 0x72, 0x75, 0x65, 0x10, 0x02,
+	0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x1a,
-	0x2a, 0x86, 0x01, 0x0a, 0x18, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e,
+	0x2e, 0x4c, 0x69, 0x73, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x56,
-	0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x4b, 0x69, 0x6e, 0x64, 0x12, 0x14, 0x0a,
+	0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x52, 0x65, 0x71, 0x1a, 0x1a, 0x2e, 0x4c, 0x69, 0x73,
-	0x10, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x49, 0x6e, 0x76, 0x61, 0x6c, 0x69,
-	0x64, 0x10, 0x00, 0x12, 0x19, 0x0a, 0x15, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65,
-	0x52, 0x65, 0x73, 0x75, 0x6d, 0x61, 0x62, 0x69, 0x6c, 0x69, 0x74, 0x79, 0x10, 0x01, 0x12, 0x23,
-	0x0a, 0x1f, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x49, 0x6e, 0x63, 0x72, 0x65,
-	0x6d, 0x65, 0x6e, 0x74, 0x61, 0x6c, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f,
-	0x6e, 0x10, 0x02, 0x12, 0x14, 0x0a, 0x10, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65,
-	0x4e, 0x6f, 0x74, 0x68, 0x69, 0x6e, 0x67, 0x10, 0x03, 0x32, 0x8f, 0x03, 0x0a, 0x0b, 0x52, 0x65,
-	0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x1a, 0x0a, 0x04, 0x50, 0x69, 0x6e,
-	0x67, 0x12, 0x08, 0x2e, 0x50, 0x69, 0x6e, 0x67, 0x52, 0x65, 0x71, 0x1a, 0x08, 0x2e, 0x50, 0x69,
-	0x6e, 0x67, 0x52, 0x65, 0x73, 0x12, 0x39, 0x0a, 0x0f, 0x4c, 0x69, 0x73, 0x74, 0x46, 0x69, 0x6c,
-	0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x73, 0x12, 0x12, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x46,
-	0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x52, 0x65, 0x71, 0x1a, 0x12, 0x2e, 0x4c,
-	0x69, 0x73, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x52, 0x65, 0x73,
-	0x12, 0x50, 0x0a, 0x16, 0x4c, 0x69, 0x73, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74,
-	0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x1a, 0x2e, 0x4c, 0x69, 0x73,
 	0x74, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69,
-	0x6f, 0x6e, 0x73, 0x52, 0x65, 0x71, 0x1a, 0x1a, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x46, 0x69, 0x6c,
+	0x6f, 0x6e, 0x73, 0x52, 0x65, 0x73, 0x12, 0x3e, 0x0a, 0x10, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f,
-	0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x52,
+	0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x12, 0x14, 0x2e, 0x44, 0x65, 0x73,
-	0x65, 0x73, 0x12, 0x3e, 0x0a, 0x10, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61,
+	0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x52, 0x65, 0x71,
-	0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x12, 0x14, 0x2e, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79,
+	0x1a, 0x14, 0x2e, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68,
-	0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x52, 0x65, 0x71, 0x1a, 0x14, 0x2e, 0x44,
+	0x6f, 0x74, 0x73, 0x52, 0x65, 0x73, 0x12, 0x41, 0x0a, 0x11, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63,
-	0x65, 0x73, 0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x52,
+	0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x75, 0x72, 0x73, 0x6f, 0x72, 0x12, 0x15, 0x2e, 0x52, 0x65,
-	0x65, 0x73, 0x12, 0x41, 0x0a, 0x11, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f,
+	0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x75, 0x72, 0x73, 0x6f, 0x72, 0x52,
-	0x6e, 0x43, 0x75, 0x72, 0x73, 0x6f, 0x72, 0x12, 0x15, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63,
+	0x65, 0x71, 0x1a, 0x15, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e,
-	0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x75, 0x72, 0x73, 0x6f, 0x72, 0x52, 0x65, 0x71, 0x1a, 0x15,
+	0x43, 0x75, 0x72, 0x73, 0x6f, 0x72, 0x52, 0x65, 0x73, 0x12, 0x35, 0x0a, 0x0d, 0x53, 0x65, 0x6e,
-	0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x75, 0x72, 0x73,
+	0x64, 0x43, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x12, 0x11, 0x2e, 0x53, 0x65, 0x6e,
-	0x6f, 0x72, 0x52, 0x65, 0x73, 0x12, 0x1d, 0x0a, 0x07, 0x53, 0x65, 0x6e, 0x64, 0x44, 0x72, 0x79,
+	0x64, 0x43, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x71, 0x1a, 0x11, 0x2e,
-	0x12, 0x08, 0x2e, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65, 0x71, 0x1a, 0x08, 0x2e, 0x53, 0x65, 0x6e,
+	0x53, 0x65, 0x6e, 0x64, 0x43, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x73,
-	0x64, 0x52, 0x65, 0x73, 0x12, 0x35, 0x0a, 0x0d, 0x53, 0x65, 0x6e, 0x64, 0x43, 0x6f, 0x6d, 0x70,
+	0x42, 0x07, 0x5a, 0x05, 0x2e, 0x3b, 0x70, 0x64, 0x75, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f,
-	0x6c, 0x65, 0x74, 0x65, 0x64, 0x12, 0x11, 0x2e, 0x53, 0x65, 0x6e, 0x64, 0x43, 0x6f, 0x6d, 0x70,
+	0x33,
-	0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x71, 0x1a, 0x11, 0x2e, 0x53, 0x65, 0x6e, 0x64, 0x43,
-	0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x73, 0x42, 0x07, 0x5a, 0x05, 0x2e,
-	0x3b, 0x70, 0x64, 0x75, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -1606,31 +1625,30 @@ var file_pdu_proto_depIdxs = []int32{
 	11, // 7: ReplicationConfig.protection:type_name -> ReplicationConfigProtection
 	1,  // 8: ReplicationConfigProtection.Initial:type_name -> ReplicationGuaranteeKind
 	1,  // 9: ReplicationConfigProtection.Incremental:type_name -> ReplicationGuaranteeKind
-	9,  // 10: SendCompletedReq.OriginalReq:type_name -> SendReq
+	12, // 10: SendRes.Properties:type_name -> Property
-	8,  // 11: ReceiveReq.To:type_name -> FilesystemVersion
+	9,  // 11: SendCompletedReq.OriginalReq:type_name -> SendReq
-	10, // 12: ReceiveReq.ReplicationConfig:type_name -> ReplicationConfig
+	8,  // 12: ReceiveReq.To:type_name -> FilesystemVersion
-	8,  // 13: DestroySnapshotsReq.Snapshots:type_name -> FilesystemVersion
+	10, // 13: ReceiveReq.ReplicationConfig:type_name -> ReplicationConfig
-	8,  // 14: DestroySnapshotRes.Snapshot:type_name -> FilesystemVersion
+	8,  // 14: DestroySnapshotsReq.Snapshots:type_name -> FilesystemVersion
-	19, // 15: DestroySnapshotsRes.Results:type_name -> DestroySnapshotRes
+	8,  // 15: DestroySnapshotRes.Snapshot:type_name -> FilesystemVersion
-	23, // 16: Replication.Ping:input_type -> PingReq
+	19, // 16: DestroySnapshotsRes.Results:type_name -> DestroySnapshotRes
-	3,  // 17: Replication.ListFilesystems:input_type -> ListFilesystemReq
+	23, // 17: Replication.Ping:input_type -> PingReq
-	6,  // 18: Replication.ListFilesystemVersions:input_type -> ListFilesystemVersionsReq
+	3,  // 18: Replication.ListFilesystems:input_type -> ListFilesystemReq
-	18, // 19: Replication.DestroySnapshots:input_type -> DestroySnapshotsReq
+	6,  // 19: Replication.ListFilesystemVersions:input_type -> ListFilesystemVersionsReq
-	21, // 20: Replication.ReplicationCursor:input_type -> ReplicationCursorReq
+	18, // 20: Replication.DestroySnapshots:input_type -> DestroySnapshotsReq
-	9,  // 21: Replication.SendDry:input_type -> SendReq
+	21, // 21: Replication.ReplicationCursor:input_type -> ReplicationCursorReq
 	14, // 22: Replication.SendCompleted:input_type -> SendCompletedReq
 	24, // 23: Replication.Ping:output_type -> PingRes
 	4,  // 24: Replication.ListFilesystems:output_type -> ListFilesystemRes
 	7,  // 25: Replication.ListFilesystemVersions:output_type -> ListFilesystemVersionsRes
 	20, // 26: Replication.DestroySnapshots:output_type -> DestroySnapshotsRes
 	22, // 27: Replication.ReplicationCursor:output_type -> ReplicationCursorRes
-	13, // 28: Replication.SendDry:output_type -> SendRes
+	15, // 28: Replication.SendCompleted:output_type -> SendCompletedRes
-	15, // 29: Replication.SendCompleted:output_type -> SendCompletedRes
+	23, // [23:29] is the sub-list for method output_type
-	23, // [23:30] is the sub-list for method output_type
+	17, // [17:23] is the sub-list for method input_type
-	16, // [16:23] is the sub-list for method input_type
+	17, // [17:17] is the sub-list for extension type_name
-	16, // [16:16] is the sub-list for extension type_name
+	17, // [17:17] is the sub-list for extension extendee
-	16, // [16:16] is the sub-list for extension extendee
+	0,  // [0:17] is the sub-list for field type_name
-	0,  // [0:16] is the sub-list for field type_name
 }
 
 func init() { file_pdu_proto_init() }

replication/logic/pdu/pdu.proto

@@ -8,7 +8,6 @@ service Replication {
       returns (ListFilesystemVersionsRes);
   rpc DestroySnapshots(DestroySnapshotsReq) returns (DestroySnapshotsRes);
   rpc ReplicationCursor(ReplicationCursorReq) returns (ReplicationCursorRes);
-  rpc SendDry(SendReq) returns (SendRes);
   rpc SendCompleted(SendCompletedReq) returns (SendCompletedRes);
   // for Send and Recv, see package rpc
 }
@@ -61,7 +60,9 @@ message SendReq {
   string ResumeToken = 4;
   Tri Encrypted = 5;
 
-  ReplicationConfig ReplicationConfig = 6;
+  bool DryRun = 6;
+
+  ReplicationConfig ReplicationConfig = 7;
 }
 
 message ReplicationConfig {
@@ -88,12 +89,14 @@ message Property {
 
 message SendRes {
   // Whether the resume token provided in the request has been used or not.
-  // If the SendReq.ResumeToken == "", this field MUST be false.
+  // If the SendReq.ResumeToken == "", this field has no meaning.
-  bool UsedResumeToken = 1;
+  bool UsedResumeToken = 2;
 
   // Expected stream size determined by dry run, not exact.
   // 0 indicates that for the given SendReq, no size estimate could be made.
-  uint64 ExpectedSize = 2;
+  int64 ExpectedSize = 3;
+
+  repeated Property Properties = 4;
 }
 
 message SendCompletedReq {

replication/logic/pdu/pdu_grpc.pb.go

@@ -23,7 +23,6 @@ type ReplicationClient interface {
 	ListFilesystemVersions(ctx context.Context, in *ListFilesystemVersionsReq, opts ...grpc.CallOption) (*ListFilesystemVersionsRes, error)
 	DestroySnapshots(ctx context.Context, in *DestroySnapshotsReq, opts ...grpc.CallOption) (*DestroySnapshotsRes, error)
 	ReplicationCursor(ctx context.Context, in *ReplicationCursorReq, opts ...grpc.CallOption) (*ReplicationCursorRes, error)
-	SendDry(ctx context.Context, in *SendReq, opts ...grpc.CallOption) (*SendRes, error)
 	SendCompleted(ctx context.Context, in *SendCompletedReq, opts ...grpc.CallOption) (*SendCompletedRes, error)
 }
 
@@ -80,15 +79,6 @@ func (c *replicationClient) ReplicationCursor(ctx context.Context, in *Replicati
 	return out, nil
 }
 
-func (c *replicationClient) SendDry(ctx context.Context, in *SendReq, opts ...grpc.CallOption) (*SendRes, error) {
-	out := new(SendRes)
-	err := c.cc.Invoke(ctx, "/Replication/SendDry", in, out, opts...)
-	if err != nil {
-		return nil, err
-	}
-	return out, nil
-}
-
 func (c *replicationClient) SendCompleted(ctx context.Context, in *SendCompletedReq, opts ...grpc.CallOption) (*SendCompletedRes, error) {
 	out := new(SendCompletedRes)
 	err := c.cc.Invoke(ctx, "/Replication/SendCompleted", in, out, opts...)
@@ -107,7 +97,6 @@ type ReplicationServer interface {
 	ListFilesystemVersions(context.Context, *ListFilesystemVersionsReq) (*ListFilesystemVersionsRes, error)
 	DestroySnapshots(context.Context, *DestroySnapshotsReq) (*DestroySnapshotsRes, error)
 	ReplicationCursor(context.Context, *ReplicationCursorReq) (*ReplicationCursorRes, error)
-	SendDry(context.Context, *SendReq) (*SendRes, error)
 	SendCompleted(context.Context, *SendCompletedReq) (*SendCompletedRes, error)
 	mustEmbedUnimplementedReplicationServer()
 }
@@ -131,9 +120,6 @@ func (UnimplementedReplicationServer) DestroySnapshots(context.Context, *Destroy
 func (UnimplementedReplicationServer) ReplicationCursor(context.Context, *ReplicationCursorReq) (*ReplicationCursorRes, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method ReplicationCursor not implemented")
 }
-func (UnimplementedReplicationServer) SendDry(context.Context, *SendReq) (*SendRes, error) {
-	return nil, status.Errorf(codes.Unimplemented, "method SendDry not implemented")
-}
 func (UnimplementedReplicationServer) SendCompleted(context.Context, *SendCompletedReq) (*SendCompletedRes, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method SendCompleted not implemented")
 }
@@ -240,24 +226,6 @@ func _Replication_ReplicationCursor_Handler(srv interface{}, ctx context.Context
 	return interceptor(ctx, in, info, handler)
 }
 
-func _Replication_SendDry_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := new(SendReq)
-	if err := dec(in); err != nil {
-		return nil, err
-	}
-	if interceptor == nil {
-		return srv.(ReplicationServer).SendDry(ctx, in)
-	}
-	info := &grpc.UnaryServerInfo{
-		Server:     srv,
-		FullMethod: "/Replication/SendDry",
-	}
-	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(ReplicationServer).SendDry(ctx, req.(*SendReq))
-	}
-	return interceptor(ctx, in, info, handler)
-}
-
 func _Replication_SendCompleted_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(SendCompletedReq)
 	if err := dec(in); err != nil {
@@ -303,10 +271,6 @@ var Replication_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "ReplicationCursor",
 			Handler:    _Replication_ReplicationCursor_Handler,
 		},
-		{
-			MethodName: "SendDry",
-			Handler:    _Replication_SendDry_Handler,
-		},
 		{
 			MethodName: "SendCompleted",
 			Handler:    _Replication_SendCompleted_Handler,

replication/logic/replication_logic.go

@@ -41,7 +41,6 @@ type Sender interface {
 	// any next call to the parent github.com/zrepl/zrepl/replication.Endpoint.
 	// If the send request is for dry run the io.ReadCloser will be nil
 	Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error)
-	SendDry(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, error)
 	SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error)
 	ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error)
 }
@@ -159,7 +158,7 @@ type Step struct {
 	encrypt     tri
 	resumeToken string // empty means no resume token shall be used
 
-	expectedSize uint64 // 0 means no size estimate present / possible
+	expectedSize int64 // 0 means no size estimate present / possible
 
 	// byteCounter is nil initially, and set later in Step.doReplication
 	// => concurrent read of that pointer from Step.ReportInfo must be protected
@@ -190,7 +189,7 @@ func (s *Step) Step(ctx context.Context) error {
 func (s *Step) ReportInfo() *report.StepInfo {
 
 	// get current byteCounter value
-	var byteCounter uint64
+	var byteCounter int64
 	s.byteCounterMtx.Lock()
 	if s.byteCounter != nil {
 		byteCounter = s.byteCounter.Count()
@@ -547,10 +546,10 @@ func (s *Step) updateSizeEstimate(ctx context.Context) error {
 
 	log := getLogger(ctx)
 
-	sr := s.buildSendRequest()
+	sr := s.buildSendRequest(true)
 
 	log.Debug("initiate dry run send request")
-	sres, err := s.sender.SendDry(ctx, sr)
+	sres, _, err := s.sender.Send(ctx, sr)
 	if err != nil {
 		log.WithError(err).Error("dry run send request failed")
 		return err
@@ -564,7 +563,7 @@ func (s *Step) updateSizeEstimate(ctx context.Context) error {
 	return nil
 }
 
-func (s *Step) buildSendRequest() (sr *pdu.SendReq) {
+func (s *Step) buildSendRequest(dryRun bool) (sr *pdu.SendReq) {
 	fs := s.parent.Path
 	sr = &pdu.SendReq{
 		Filesystem:        fs,
@@ -572,6 +571,7 @@ func (s *Step) buildSendRequest() (sr *pdu.SendReq) {
 		To:                s.to,
 		Encrypted:         s.encrypt.ToPDU(),
 		ResumeToken:       s.resumeToken,
+		DryRun:            dryRun,
 		ReplicationConfig: s.parent.policy.ReplicationConfig,
 	}
 	return sr
@@ -582,7 +582,7 @@ func (s *Step) doReplication(ctx context.Context) error {
 	fs := s.parent.Path
 
 	log := getLogger(ctx).WithField("filesystem", fs)
-	sr := s.buildSendRequest()
+	sr := s.buildSendRequest(false)
 
 	log.Debug("initiate send request")
 	sres, stream, err := s.sender.Send(ctx, sr)

replication/report/replication_report.go

@@ -62,23 +62,11 @@ 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
@@ -109,11 +97,11 @@ type StepInfo struct {
 	From, To        string
 	Resumed         bool
 	Encrypted       EncryptedEnum
-	BytesExpected   uint64
+	BytesExpected   int64
-	BytesReplicated uint64
+	BytesReplicated int64
 }
 
-func (a *AttemptReport) BytesSum() (expected, replicated uint64, containsInvalidSizeEstimates bool) {
+func (a *AttemptReport) BytesSum() (expected, replicated int64, containsInvalidSizeEstimates bool) {
 	for _, fs := range a.Filesystems {
 		e, r, fsContainsInvalidEstimate := fs.BytesSum()
 		containsInvalidSizeEstimates = containsInvalidSizeEstimates || fsContainsInvalidEstimate
@@ -123,7 +111,7 @@ func (a *AttemptReport) BytesSum() (expected, replicated uint64, containsInvalid
 	return expected, replicated, containsInvalidSizeEstimates
 }
 
-func (f *FilesystemReport) BytesSum() (expected, replicated uint64, containsInvalidSizeEstimates bool) {
+func (f *FilesystemReport) BytesSum() (expected, replicated int64, containsInvalidSizeEstimates bool) {
 	for _, step := range f.Steps {
 		expected += step.Info.BytesExpected
 		replicated += step.Info.BytesReplicated

rpc/dataconn/dataconn_client.go

@@ -114,6 +114,12 @@ func (c *Client) ReqSend(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, i
 	if err != nil {
 		return nil, nil, err
 	}
+	putWireOnReturn := true
+	defer func() {
+		if putWireOnReturn {
+			c.putWire(conn)
+		}
+	}()
 
 	if err := c.send(ctx, conn, EndpointSend, req, nil); err != nil {
 		return nil, nil, err
@@ -125,10 +131,14 @@ func (c *Client) ReqSend(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, i
 	}
 
 	var stream io.ReadCloser
-	stream, err = conn.ReadStream(ZFSStream, true) // no shadow
+	if !req.DryRun {
-	if err != nil {
+		putWireOnReturn = false
-		return nil, nil, err
+		stream, err = conn.ReadStream(ZFSStream, true) // no shadow
+		if err != nil {
+			return nil, nil, err
+		}
 	}
+
 	return &res, stream, nil
 }
 

rpc/dataconn/dataconn_server.go

@@ -172,15 +172,6 @@ func (s *Server) serveConnRequest(ctx context.Context, endpoint string, c *strea
 			return
 		}
 		res, sendStream, handlerErr = s.h.Send(ctx, &req) // SHADOWING
-		// ensure that we always close the sendStream
-		if sendStream != nil {
-			defer func() {
-				err := sendStream.Close()
-				if err != nil {
-					s.log.WithError(err).Error("cannot close send stream")
-				}
-			}()
-		}
 	case EndpointRecv:
 		var req pdu.ReceiveReq
 		if err := proto.Unmarshal(reqStructured, &req); err != nil {
@@ -248,9 +239,12 @@ func (s *Server) serveConnRequest(ctx context.Context, endpoint string, c *strea
 
 	if sendStream != nil {
 		err := c.SendStream(ctx, sendStream, ZFSStream)
+		closeErr := sendStream.Close()
+		if closeErr != nil {
+			s.log.WithError(err).Error("cannot close send stream")
+		}
 		if err != nil {
 			s.log.WithError(err).Error("cannot write send stream")
 		}
-		// sendStream.Close() done via defer above
 	}
 }

rpc/dataconn/timeoutconn/timeoutconn_readv_unsupported.go

@@ -1,4 +1,3 @@
-//go:build illumos || solaris
 // +build illumos solaris
 
 package timeoutconn

rpc/dataconn/timeoutconn/timeoutconn_test.go

@@ -14,7 +14,6 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/zrepl/zrepl/util/socketpair"
-	"github.com/zrepl/zrepl/util/zreplcircleci"
 )
 
 func TestReadTimeout(t *testing.T) {
@@ -82,8 +81,6 @@ 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()
@@ -154,7 +151,6 @@ 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()

rpc/dataconn/timeoutconn/timoutconn_readv.go

@@ -1,5 +1,5 @@
-//go:build !illumos && !solaris
+// +build !illumos
-// +build !illumos,!solaris
+// +build !solaris
 
 package timeoutconn
 

rpc/rpc_client.go

@@ -108,13 +108,6 @@ func (c *Client) Receive(ctx context.Context, req *pdu.ReceiveReq, stream io.Rea
 	return c.dataClient.ReqRecv(ctx, req, stream)
 }
 
-func (c *Client) SendDry(ctx context.Context, in *pdu.SendReq) (*pdu.SendRes, error) {
-	ctx, endSpan := trace.WithSpan(ctx, "rpc.client.SendDry")
-	defer endSpan()
-
-	return c.controlClient.SendDry(ctx, in)
-}
-
 func (c *Client) ListFilesystems(ctx context.Context, in *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error) {
 	ctx, endSpan := trace.WithSpan(ctx, "rpc.client.ListFilesystems")
 	defer endSpan()

rpc/versionhandshake/versionhandshake.go

@@ -152,7 +152,7 @@ func (m *HandshakeMessage) DecodeReader(r io.Reader, maxLen int) error {
 
 func DoHandshakeCurrentVersion(conn net.Conn, deadline time.Time) *HandshakeError {
 	// current protocol version is hardcoded here
-	return DoHandshakeVersion(conn, deadline, 6)
+	return DoHandshakeVersion(conn, deadline, 5)
 }
 
 const HandshakeMessageMaxLen = 16 * 4096

transport/tcp/serve_tcp_ipmap_test.go

@@ -91,12 +91,15 @@ func TestIPMap(t *testing.T) {
 				"fde4:8dba:82e1::/64": "sub64-*",
 			},
 			expect: map[string]testCaseExpect{
-				"10.1.2.3":      {expectNoMapping: true},
+				"10.1.2.3":        {expectNoMapping: true},
-				"192.168.23.1":  {expectIdent: "db-192.168.23.1"},
+				"192.168.23.1":    {expectIdent: "db-192.168.23.1"},
-				"192.168.42.1":  {expectIdent: "web-192.168.42.1"},
+				"192.168.23.23":   {expectIdent: "db-twentythree"},
-				"192.168.23.23": {expectIdent: "db-twentythree"},
+				"192.168.023.001": {expectIdent: "db-192.168.23.1"},
-				"10.1.4.5":      {expectIdent: "my-10.1.4.5-server"},
+				"10.1.4.5":        {expectIdent: "my-10.1.4.5-server"},
 
+				// normalization
+				"192.168.42.1":    {expectIdent: "web-192.168.42.1"},
+				"192.168.042.001": {expectIdent: "web-192.168.42.1"},
 				// v6 matching
 				"fe80::23:42%eth1": {expectIdent: "san-fe80::23:42-eth1"},
 				"fe80::23:42%eth2": {expectNoMapping: true},
@@ -176,8 +179,7 @@ func TestIPMap(t *testing.T) {
 			for input, expect := range c.expect {
 				// reuse newIPMapEntry to parse test case input
 				// "test" is not used during testing but must not be empty.
-				ipMapEntry, err := newIPMapEntry(input, "test")
+				ipMapEntry, _ := newIPMapEntry(input, "test")
-				require.NoError(t, err)
 				ones, bits := ipMapEntry.subnet.Mask.Size()
 				require.Equal(t, bits, net.IPv6len*8, "and we know ipMapEntry always expands its IPs to 16bytes")
 				require.Equal(t, ones, net.IPv6len*8, "test case addresses must be fully specified")

util/bandwidthlimit/bandwidthlimit.go

@@ -1,79 +0,0 @@
-package bandwidthlimit
-
-import (
-	"errors"
-	"io"
-
-	"github.com/juju/ratelimit"
-)
-
-type Wrapper interface {
-	WrapReadCloser(io.ReadCloser) io.ReadCloser
-}
-
-type Config struct {
-	// Units in this struct are in _bytes_.
-
-	Max            int64 // < 0 means no limit, BucketCapacity is irrelevant then
-	BucketCapacity int64
-}
-
-func NoLimitConfig() Config {
-	return Config{
-		Max:            -1,
-		BucketCapacity: -1,
-	}
-}
-
-func ValidateConfig(conf Config) error {
-	if conf.BucketCapacity == 0 {
-		return errors.New("BucketCapacity must not be zero")
-	}
-	return nil
-}
-
-func WrapperFromConfig(conf Config) Wrapper {
-	if err := ValidateConfig(conf); err != nil {
-		panic(err)
-	}
-
-	if conf.Max < 0 {
-		return noLimit{}
-	}
-
-	return &withLimit{
-		bucket: ratelimit.NewBucketWithRate(float64(conf.Max), conf.BucketCapacity),
-	}
-}
-
-type noLimit struct{}
-
-func (_ noLimit) WrapReadCloser(rc io.ReadCloser) io.ReadCloser { return rc }
-
-type withLimit struct {
-	bucket *ratelimit.Bucket
-}
-
-func (l *withLimit) WrapReadCloser(rc io.ReadCloser) io.ReadCloser {
-	return WrapReadCloser(rc, l.bucket)
-}
-
-type withLimitReadCloser struct {
-	orig    io.Closer
-	limited io.Reader
-}
-
-func (r *withLimitReadCloser) Read(buf []byte) (int, error) {
-	return r.limited.Read(buf)
-}
-
-func (r *withLimitReadCloser) Close() error {
-	return r.orig.Close()
-}
-
-func WrapReadCloser(rc io.ReadCloser, bucket *ratelimit.Bucket) io.ReadCloser {
-	return &withLimitReadCloser{
-		limited: ratelimit.Reader(rc, bucket),
-		orig:    rc,
-	}
-}

util/bandwidthlimit/bandwidthlimit_test.go

@@ -1,19 +0,0 @@
-package bandwidthlimit
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/require"
-)
-
-func TestNoLimitConfig(t *testing.T) {
-
-	conf := NoLimitConfig()
-
-	err := ValidateConfig(conf)
-	require.NoError(t, err)
-
-	require.NotPanics(t, func() {
-		_ = WrapperFromConfig(conf)
-	})
-}

util/bytecounter/bytecounter_readcloser.go

@@ -9,7 +9,7 @@ import (
 // its interface and counting the bytes written to during copying.
 type ReadCloser interface {
 	io.ReadCloser
-	Count() uint64
+	Count() int64
 }
 
 // NewReadCloser wraps rc.
@@ -19,11 +19,11 @@ func NewReadCloser(rc io.ReadCloser) ReadCloser {
 
 type readCloser struct {
 	rc    io.ReadCloser
-	count uint64
+	count int64
 }
 
-func (r *readCloser) Count() uint64 {
+func (r *readCloser) Count() int64 {
-	return atomic.LoadUint64(&r.count)
+	return atomic.LoadInt64(&r.count)
 }
 
 var _ io.ReadCloser = &readCloser{}
@@ -34,9 +34,6 @@ func (r *readCloser) Close() error {
 
 func (r *readCloser) Read(p []byte) (int, error) {
 	n, err := r.rc.Read(p)
-	if n < 0 {
+	atomic.AddInt64(&r.count, int64(n))
-		panic("expecting n >= 0")
-	}
-	atomic.AddUint64(&r.count, uint64(n))
 	return n, err
 }

util/datasizeunit/datasizeunit.go

@@ -1,86 +0,0 @@
-package datasizeunit
-
-import (
-	"errors"
-	"fmt"
-	"math"
-	"regexp"
-	"strconv"
-	"strings"
-)
-
-type Bits struct {
-	bits float64
-}
-
-func (b Bits) ToBits() float64    { return b.bits }
-func (b Bits) ToBytes() float64   { return b.bits / 8 }
-func FromBytesInt64(i int64) Bits { return Bits{float64(i) * 8} }
-
-var datarateRegex = regexp.MustCompile(`^([-0-9\.]*)\s*(bit|(|K|Ki|M|Mi|G|Gi|T|Ti)([bB]))$`)
-
-func (r *Bits) UnmarshalYAML(u func(interface{}, bool) error) (_ error) {
-
-	var s string
-	err := u(&s, false)
-	if err != nil {
-		return err
-	}
-
-	genericErr := func(err error) error {
-		var buf strings.Builder
-		fmt.Fprintf(&buf, "cannot parse %q using regex %s", s, datarateRegex)
-		if err != nil {
-			fmt.Fprintf(&buf, ": %s", err)
-		}
-		return errors.New(buf.String())
-	}
-
-	match := datarateRegex.FindStringSubmatch(s)
-	if match == nil {
-		return genericErr(nil)
-	}
-
-	bps, err := strconv.ParseFloat(match[1], 64)
-	if err != nil {
-		return genericErr(err)
-	}
-
-	if match[2] == "bit" {
-		if math.Round(bps) != bps {
-			return genericErr(fmt.Errorf("unit bit must be an integer value"))
-		}
-		r.bits = bps
-		return nil
-	}
-
-	factorMap := map[string]uint64{
-		"": 1,
-
-		"K": 1e3,
-		"M": 1e6,
-		"G": 1e9,
-		"T": 1e12,
-
-		"Ki": 1 << 10,
-		"Mi": 1 << 20,
-		"Gi": 1 << 30,
-		"Ti": 1 << 40,
-	}
-	factor, ok := factorMap[match[3]]
-	if !ok {
-		panic(match)
-	}
-
-	baseUnitFactorMap := map[string]uint64{
-		"b": 1,
-		"B": 8,
-	}
-	baseUnitFactor, ok := baseUnitFactorMap[match[4]]
-	if !ok {
-		panic(match)
-	}
-
-	r.bits = bps * float64(factor) * float64(baseUnitFactor)
-	return nil
-}

util/datasizeunit/datasizeunit_test.go

@@ -1,57 +0,0 @@
-package datasizeunit
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"github.com/zrepl/yaml-config"
-)
-
-func TestBits(t *testing.T) {
-
-	tcs := []struct {
-		input      string
-		expectRate float64
-		expectErr  string
-	}{
-		{`23 bit`, 23, ""}, // bit special case works
-		{`23bit`, 23, ""},  // also without space
-
-		{`10MiB`, 10 * (1 << 20) * 8, ""},  // integer unit without space
-		{`10 MiB`, 8 * 10 * (1 << 20), ""}, // integer unit with space
-
-		{`10.5 Kib`, 10.5 * (1 << 10), ""}, // floating point with bit unit works with space
-		{`10.5Kib`, 10.5 * (1 << 10), ""},  // floating point with bit unit works without space
-
-		// unit checks
-		{`1 bit`, 1, ""},
-		{`1 B`, 1 * 8, ""},
-		{`1 Kb`, 1e3, ""},
-		{`1 Kib`, 1 << 10, ""},
-		{`1 Mb`, 1e6, ""},
-		{`1 Mib`, 1 << 20, ""},
-		{`1 Gb`, 1e9, ""},
-		{`1 Gib`, 1 << 30, ""},
-		{`1 Tb`, 1e12, ""},
-		{`1 Tib`, 1 << 40, ""},
-	}
-
-	for _, tc := range tcs {
-		t.Run(tc.input, func(t *testing.T) {
-
-			var bits Bits
-			err := yaml.Unmarshal([]byte(tc.input), &bits)
-			if tc.expectErr != "" {
-				assert.Error(t, err)
-				assert.Regexp(t, tc.expectErr, err.Error())
-				assert.Zero(t, bits.bits)
-			} else {
-				require.NoError(t, err)
-				assert.Equal(t, tc.expectRate, bits.bits)
-			}
-		})
-
-	}
-
-}

util/envconst/envconst.go

@@ -73,24 +73,6 @@ func Int64(varname string, def int64) (d int64) {
 	return d
 }
 
-func Uint64(varname string, def uint64) (d uint64) {
-	var err error
-	if v, ok := cache.Load(varname); ok {
-		return v.(uint64)
-	}
-	e := os.Getenv(varname)
-	if e == "" {
-		d = def
-	} else {
-		d, err = strconv.ParseUint(e, 10, 64)
-		if err != nil {
-			panic(err)
-		}
-	}
-	cache.Store(varname, d)
-	return d
-}
-
 func Bool(varname string, def bool) (d bool) {
 	var err error
 	if v, ok := cache.Load(varname); ok {

util/tcpsock/tcpsock_freebind_freebsd.go

@@ -1,4 +1,3 @@
-//go:build freebsd
 // +build freebsd
 
 package tcpsock

util/tcpsock/tcpsock_freebind_linux.go

@@ -1,4 +1,3 @@
-//go:build linux
 // +build linux
 
 package tcpsock

util/tcpsock/tcpsock_freebind_unsupported.go

@@ -1,4 +1,3 @@
-//go:build !linux && !freebsd
 // +build !linux,!freebsd
 
 package tcpsock

util/zreplcircleci/zreplcircleci.go

@@ -1,13 +0,0 @@
-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

@@ -1,51 +0,0 @@
-// 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,15 +80,7 @@ func ZFSGetFilesystemPlaceholderState(ctx context.Context, p *DatasetPath) (stat
 	return state, nil
 }
 
-//go:generate enumer -type=FilesystemPlaceholderCreateEncryptionValue -trimprefix=FilesystemPlaceholderCreateEncryption
+func ZFSCreatePlaceholderFilesystem(ctx context.Context, fs *DatasetPath, parent *DatasetPath) (err error) {
-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())
 	}
@@ -99,18 +91,44 @@ func ZFSCreatePlaceholderFilesystem(ctx context.Context, fs *DatasetPath, parent
 		"-o", "mountpoint=none",
 	}
 
-	if !encryption.IsAFilesystemPlaceholderCreateEncryptionValue() {
+	// xxx handle encryption not supported
-		panic(encryption)
+	props, err := zfsGet(ctx, parent.ToString(), []string{"keystatus"}, SourceAny)
+	if err != nil {
+		return errors.Wrap(err, "cannot determine key status")
 	}
-	switch encryption {
+	keystatus := props.Get("keystatus") // xxx ability to distringuish `-` from ``
-	case FilesystemPlaceholderCreateEncryptionInherit:
+	if keystatus == "" {
-		// no-op
+		// parent is unencrypted => placeholder inherits encryption
-	case FilesystemPlaceholderCreateEncryptionOff:
+	} else if keystatus == "available" {
-		cmdline = append(cmdline, "-o", "encryption=off")
+		// parent is encrypted but since the key is loaded we can create an encrypted placeholder dataset
-	default:
+		// without `-o encryption=off`
-		panic(encryption)
+	} else if keystatus == "unavailable" {
+		// parent is encrypted but keys are not loaded, either because
+		//  1) it's a send-encrypted dataset, or because
+		//  2) the user forgot to zfs load-key the root_fs or above
+		// In both cases we can't create an encrypted placeholder dataset.
+		// In case 1), we want to create an unencrypted placeholder through `-o encryption=off`.
+		// In case 2), `-o encryption=off` is harmful security-wise because it breaks the encrypt-on-receiver use case (https://github.com/zrepl/zrepl/issues/504)
+		// 			   I.e., all children of the placeholder won't be encrypted because they inherit encryption=off
+		//
+		// => we could attempt to distinguish the cases by being more context sensitive
+		//    (i.e., check whether the encryption root is root_fs or a parent thereof)
+		//    However, that's always going to be imprecise, and a wrong decision there is harmful security-wise.
+		//
+		// => thus the safe choice is to never use `-o encryption=off` by default
+		//    only if we know for sure that the stream is encrypted should we create placeholders with `-o encryption=off`
+		//    => this knowledge can be achieved through one of the following means:
+		//		a) have the sender indicate it to us in the RPC request (it's ok to trust them in this particular case since lying only hurts _their_ data's confidentiality)
+		// 		b) have the user acknowledge it in the receiver config
+	} else {
+		return errors.Errorf("unknown keystatus value %q for dataset %q", keystatus, parent.ToString())
 	}
 
+	if parentEncrypted, err := ZFSGetEncryptionEnabled(ctx, parent.ToString()); err != nil {
+		return errors.Wrap(err, "cannot determine encryption support")
+	} else if parentEncrypted {
+		cmdline = append(cmdline, "-o", "encryption=off")
+	}
 	cmdline = append(cmdline, fs.ToString())
 	cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, cmdline...)
 
@@ -164,34 +182,3 @@ 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

@@ -7,7 +7,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
-	"math"
 	"os"
 	"os/exec"
 	"regexp"
@@ -15,6 +14,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
@@ -334,122 +334,60 @@ func pipeWithCapacityHint(capacity int) (r, w *os.File, err error) {
 	return stdoutReader, stdoutWriter, nil
 }
 
-type sendStreamState int
-
-const (
-	sendStreamOpen sendStreamState = iota
-	sendStreamClosed
-)
-
 type SendStream struct {
-	cmd          *zfscmd.Cmd
+	cmd  *zfscmd.Cmd
-	kill         context.CancelFunc
+	kill context.CancelFunc
-	stdoutReader io.ReadCloser // not *os.File for mocking during platformtest
-	stderrBuf    *circlog.CircularLog
 
-	mtx     sync.Mutex
+	closeMtx     sync.Mutex
-	state   sendStreamState
+	stdoutReader *os.File
-	exitErr *ZFSError
+	stderrBuf    *circlog.CircularLog
+	opErr        error
 }
 
-func (s *SendStream) Read(p []byte) (n int, _ error) {
+func (s *SendStream) Read(p []byte) (n int, err error) {
-	s.mtx.Lock()
+	s.closeMtx.Lock()
-	defer s.mtx.Unlock()
+	opErr := s.opErr
-
+	s.closeMtx.Unlock()
-	switch s.state {
+	if opErr != nil {
-	case sendStreamClosed:
+		return 0, opErr
-		return 0, os.ErrClosed
-
-	case sendStreamOpen:
-		n, readErr := s.stdoutReader.Read(p)
-		if readErr != nil {
-			debug("sendStream: read: readErr=%T %s", readErr, readErr)
-			if readErr == io.EOF {
-				// io.EOF must be bubbled up as is so that consumers can handle it properly.
-				return n, readErr
-			}
-			// Assume that the error is not retryable.
-			// Try to kill now so that we can return a nice *ZFSError with captured stderr.
-			// If the kill doesn't work, it doesn't matter because the caller must by contract call Close() anyways.
-			killErr := s.killAndWait()
-			debug("sendStream: read: killErr=%T %s", killErr, killErr)
-			if killErr == nil {
-				s.state = sendStreamClosed
-				return n, s.exitErr // return the nice error
-			} else {
-				// we remain open so that we retry
-				return n, readErr // return the normal error
-			}
-		}
-		return n, readErr
-
-	default:
-		panic("unreachable")
 	}
+
+	n, err = s.stdoutReader.Read(p)
+	if err != nil {
+		debug("sendStream: read err: %T %s", err, err)
+		// TODO we assume here that any read error is permanent
+		// which is most likely the case for a local zfs send
+		kwerr := s.killAndWait(err)
+		debug("sendStream: killAndWait n=%v err= %T %s", n, kwerr, kwerr)
+		// TODO we assume here that any read error is permanent
+		return n, kwerr
+	}
+	return n, err
 }
 
 func (s *SendStream) Close() error {
 	debug("sendStream: close called")
-	s.mtx.Lock()
+	return s.killAndWait(nil)
-	defer s.mtx.Unlock()
-
-	switch s.state {
-	case sendStreamOpen:
-		err := s.killAndWait()
-		if err != nil {
-			return err
-		} else {
-			s.state = sendStreamClosed
-			return nil
-		}
-	case sendStreamClosed:
-		return os.ErrClosed
-	default:
-		panic("unreachable")
-	}
 }
 
-// returns nil iff the child process is gone (has been successfully waited upon)
+func (s *SendStream) killAndWait(precedingReadErr error) error {
-// in that case, s.exitErr is set
-func (s *SendStream) killAndWait() error {
 
 	debug("sendStream: killAndWait enter")
 	defer debug("sendStream: killAndWait leave")
+	if precedingReadErr == io.EOF {
+		// give the zfs process a little bit of time to terminate itself
+		// if it holds this deadline, exitErr will be nil
+		time.AfterFunc(200*time.Millisecond, s.kill)
+	} else {
+		s.kill()
+	}
 
-	// send SIGKILL
+	// allow async kills from Close(), that's why we only take the mutex here
-	s.kill()
+	s.closeMtx.Lock()
+	defer s.closeMtx.Unlock()
 
-	// Close our read-end of the pipe.
+	if s.opErr != nil {
-	//
+		return s.opErr
-	// We must do this before .Wait() because in some (not all) versions/build configs of ZFS,
-	// `zfs send` uses a separate kernel thread (taskq) to write the send stream (function `dump_bytes`).
-	// The `zfs send` thread then waits uinterruptably for the taskq thread to finish the write.
-	// And signalling the `zfs send` thread doesn't propagate to the taskq thread.
-	// So we end up in a state where we .Wait() forever.
-	// (See https://github.com/openzfs/zfs/issues/12500 and
-	//  https://github.com/zrepl/zrepl/issues/495#issuecomment-902530043)
-	//
-	// By closing our read end of the pipe before .Wait(), we unblock the taskq thread if there is any.
-	// If there is no separate taskq thread, the SIGKILL to `zfs end` would suffice and be most precise,
-	// but due to the circumstances above, there is no other portable & robust way.
-	//
-	// However, the fallout from closing the pipe is that (in non-taskq builds) `zfs sends` will get a SIGPIPE.
-	// And on Linux, that SIGPIPE appears to win over the previously issued SIGKILL.
-	// And thus, on Linux, the `zfs send` will be killed by the default SIGPIPE handler.
-	// We can observe this in the WaitStatus below.
-	// This behavior is slightly annoying because the *exec.ExitError's message ("signal: broken pipe")
-	// isn't as clear as ("signal: killed").
-	// However, it seems like we just have to live with that. (covered by platformtest)
-	var closePipeErr error
-	if s.stdoutReader != nil {
-		closePipeErr = s.stdoutReader.Close()
-		if closePipeErr == nil {
-			// avoid double-closes in case waiting below doesn't work
-			// and someone attempts Close again
-			s.stdoutReader = nil
-		} else {
-			return closePipeErr
-		}
 	}
 
 	waitErr := s.cmd.Wait()
@@ -464,30 +402,39 @@ func (s *SendStream) killAndWait() error {
 		}
 	}
 
-	// invariant: at this point, the child is gone and we cleaned up everything related to the SendStream
+	// now, after we know the program exited do we close the pipe
+	var closePipeErr error
+	if s.stdoutReader != nil {
+		closePipeErr = s.stdoutReader.Close()
+		if closePipeErr == nil {
+			// avoid double-closes in case anything below doesn't work
+			// and someone calls Close again
+			s.stdoutReader = nil
+		} else {
+			return closePipeErr
+		}
+	}
 
+	// we managed to tear things down, no let's give the user some pretty *ZFSError
 	if exitErr != nil {
-		// zfs send exited with an error or was killed by a signal.
+		s.opErr = &ZFSError{
-		s.exitErr = &ZFSError{
 			Stderr:  []byte(s.stderrBuf.String()),
 			WaitErr: exitErr,
 		}
 	} else {
-		// zfs send exited successfully (we know that since waitErr was either nil or wasn't an *exec.ExitError)
+		s.opErr = precedingReadErr
-		s.exitErr = nil
 	}
 
-	return nil
+	// detect the edge where we're called from s.Read
-}
+	// after the pipe EOFed and zfs send exited without errors
+	// this is actually the "hot" / nice path
+	if exitErr == nil && precedingReadErr == io.EOF {
+		return precedingReadErr
+	}
 
-func (s *SendStream) TestOnly_ReplaceStdoutReader(f io.ReadCloser) (prev io.ReadCloser) {
+	return s.opErr
-	prev = s.stdoutReader
-	s.stdoutReader = f
-	return prev
 }
 
-func (s *SendStream) TestOnly_ExitErr() *ZFSError { return s.exitErr }
-
 // NOTE: When updating this struct, make sure to update funcs Validate ValidateCorrespondsToResumeToken
 type ZFSSendArgVersion struct {
 	RelName string
@@ -961,7 +908,7 @@ type DrySendInfo struct {
 	Type         DrySendType
 	Filesystem   string // parsed from To field
 	From, To     string // direct copy from ZFS output
-	SizeEstimate uint64 // 0 if size estimate is not possible
+	SizeEstimate int64  // -1 if size estimate is not possible
 }
 
 var (
@@ -1034,10 +981,11 @@ func (s *DrySendInfo) unmarshalInfoLine(l string) (regexMatched bool, err error)
 		// see https://github.com/zrepl/zrepl/issues/289
 		fields["size"] = "0"
 	}
-	s.SizeEstimate, err = strconv.ParseUint(fields["size"], 10, 64)
+	s.SizeEstimate, err = strconv.ParseInt(fields["size"], 10, 64)
 	if err != nil {
 		return true, fmt.Errorf("cannot not parse size: %s", err)
 	}
+
 	return true, nil
 }
 
@@ -1047,7 +995,6 @@ func ZFSSendDry(ctx context.Context, sendArgs ZFSSendArgsValidated) (_ *DrySendI
 
 	if sendArgs.From != nil && strings.Contains(sendArgs.From.RelName, "#") {
 		/* TODO:
-		 * XXX feature check & support this as well
 		 * ZFS at the time of writing does not support dry-run send because size-estimation
 		 * uses fromSnap's deadlist. However, for a bookmark, that deadlist no longer exists.
 		 * Redacted send & recv will bring this functionality, see
@@ -1066,7 +1013,7 @@ func ZFSSendDry(ctx context.Context, sendArgs ZFSSendArgsValidated) (_ *DrySendI
 			Filesystem:   sendArgs.FS,
 			From:         fromAbs,
 			To:           toAbs,
-			SizeEstimate: 0}, nil
+			SizeEstimate: -1}, nil
 	}
 
 	args := make([]string, 0)
@@ -1086,19 +1033,6 @@ func ZFSSendDry(ctx context.Context, sendArgs ZFSSendArgsValidated) (_ *DrySendI
 	if err := si.unmarshalZFSOutput(output); err != nil {
 		return nil, fmt.Errorf("could not parse zfs send -n output: %s", err)
 	}
-
-	// There is a bug in OpenZFS where it estimates the size incorrectly.
-	// - zrepl: https://github.com/zrepl/zrepl/issues/463
-	// - resulting upstream bug: https://github.com/openzfs/zfs/issues/12265
-	//
-	// The wrong estimates are easy to detect because they are absurdly large.
-	// NB: we're doing the workaround for this late so that the test cases are not affected.
-	sizeEstimateThreshold := envconst.Uint64("ZREPL_ZFS_SEND_SIZE_ESTIMATE_INCORRECT_THRESHOLD", math.MaxInt64)
-	if sizeEstimateThreshold != 0 && si.SizeEstimate >= sizeEstimateThreshold {
-		debug("size estimate exceeds threshold %v, working around it: %#v %q", sizeEstimateThreshold, si, args)
-		si.SizeEstimate = 0
-	}
-
 	return &si, nil
 }
 
@@ -1567,18 +1501,8 @@ func (s PropertySource) zfsGetSourceFieldPrefixes() []string {
 	return prefixes
 }
 
-func zfsGetRecursive(ctx context.Context, path string, depth int, dstypes []string, props []string, allowedSources PropertySource) (map[string]*ZFSProperties, error) {
+func zfsGet(ctx context.Context, path string, props []string, allowedSources PropertySource) (*ZFSProperties, error) {
-	args := []string{"get", "-Hp", "-o", "name,property,value,source"}
+	args := []string{"get", "-Hp", "-o", "property,value,source", strings.Join(props, ","), path}
-	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 {
@@ -1598,67 +1522,36 @@ func zfsGetRecursive(ctx context.Context, path string, depth int, dstypes []stri
 	}
 	o := string(stdout)
 	lines := strings.Split(o, "\n")
-	propsByFS := make(map[string]*ZFSProperties)
+	if len(lines) < 1 || // account for newlines
+		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] { // last line is an empty line due to how strings.Split works
+	for _, line := range lines[:len(lines)-1] {
 		fields := strings.FieldsFunc(line, func(r rune) bool {
 			return r == '\t'
 		})
-		if len(fields) != 4 {
+		if len(fields) != 3 {
-			return nil, fmt.Errorf("zfs get did not return name,property,value,source tuples")
+			return nil, fmt.Errorf("zfs get did not return property,value,source tuples")
 		}
 		for _, p := range allowedPrefixes {
 			// prefix-match so that SourceAny (= "") works
-			if strings.HasPrefix(fields[3], p) {
+			if strings.HasPrefix(fields[2], p) {
-				source, err := parsePropertySource(fields[3])
+				source, err := parsePropertySource(fields[2])
 				if err != nil {
 					return nil, errors.Wrap(err, "parse property source")
 				}
-				fsProps, ok := propsByFS[fields[0]]
+				res.m[fields[0]] = PropertyValue{
-				if !ok {
+					Value:  fields[1],
-					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
 }
 

zfs/zfs_pipe.go

@@ -1,4 +1,3 @@
-//go:build !linux
 // +build !linux
 
 package zfs

zfs/zfscmd/zfscmd.go

@@ -209,7 +209,3 @@ func (c *Cmd) Runtime() time.Duration {
 	}
 	return c.waitReturnedAt.Sub(c.startedAt)
 }
-
-func (c *Cmd) TestOnly_ExecCmd() *exec.Cmd {
-	return c.cmd
-}