Revert "DROP THIS don't run tests in CI"

This reverts commit a3c6173b85dede625f51ca7172a53a5aff024fae.

.circleci/config.yml

@@ -59,11 +59,12 @@ jobs:
       - run: ./lazy.sh devsetup
 
       - run: make zrepl-bin
+      - run: make test-platform-bin
       - run: make vet
-      - run: make test
       - run: make lint
       - run: make release
-
+      - run: make test-go
+      # cannot run test-platform because circle-ci runs in linux containers
 
       - store_artifacts:
           path: ./artifacts/release
@@ -85,6 +86,7 @@ jobs:
             mc config host add --api s3v4 zrepl-minio https://minio.cschwarz.com ${MINIO_ACCESS_KEY} ${MINIO_SECRET_KEY}
             
             # Upload artifacts
+            mkdir -p artifacts/release
             echo "$CIRCLE_BUILD_URL" > ./artifacts/release/cirlceci_build_url
             mc cp -r artifacts/release "zrepl-minio/zrepl-ci-artifacts/${CIRCLE_SHA1}/${CIRCLE_JOB}/"
 
@@ -92,11 +94,45 @@ jobs:
             REPO="zrepl/zrepl"
             COMMIT="${CIRCLE_SHA1}"
             JOB_NAME="${CIRCLE_JOB}"
-            curl "https://api.github.com/repos/$REPO/statuses/$COMMIT?access_token=$GITHUB_COMMIT_STATUS_TOKEN" \
+            curl "https://api.github.com/repos/$REPO/statuses/$COMMIT" \
               -H "Content-Type: application/json" \
+              -H "Authorization: token $GITHUB_COMMIT_STATUS_TOKEN" \
               -X POST \
               -d '{"context":"zrepl/publish-ci-artifacts", "state": "success", "description":"CI Build Artifacts for '"$JOB_NAME"'", "target_url":"https://minio.cschwarz.com/minio/zrepl-ci-artifacts/'"$COMMIT"'/"}'
 
+      - run:
+          shell: /bin/bash -eo pipefail
+          when: always
+          command: |
+            if [ -n "$CIRCLE_PR_NUMBER" ]; then # CIRCLE_PR_NUMBER is guaranteed to be only present in forked PRs (external)
+              echo "Forked PR detected. Sry, can't trust you with credentials to external artifact store, use CircleCI's instead."
+              exit 0
+            fi
+            set -u # from now on
+
+            # Trigger Platform Tests
+            # ZREPL_BOT_GIT_PRWN_EU_ACCESS_TOKEN from secrets
+            ## Sync this commit to the GitLab
+            git remote add prwn-gitlab "https://zrepl-bot:$ZREPL_BOT_GIT_PRWN_EU_ACCESS_TOKEN@git.prwn.eu/problame/zrepl.git"
+            REMOTE_REF=circleci/$CIRCLE_BRANCH/$CIRCLE_BUILD_NUM
+            git push -f prwn-gitlab "HEAD:$REMOTE_REF"
+            ## Trigger Tests
+            curl -v --request POST \
+              --header "PRIVATE-TOKEN: $ZREPL_BOT_GIT_PRWN_EU_ACCESS_TOKEN" \
+              --header "Content-Type: application/json" \
+              --data '{ "ref":"'"$REMOTE_REF"'", "variables": [ {"key":"ZREPL_MAIN_REPO_COMMIT", "value": "'"$CIRCLE_SHA1"'" }, { "key":"ZREPL_CI_BUILDJOB", "value": "'"$CIRCLE_JOB"'" } ] } ' \
+              "https://git.prwn.eu/api/v4/projects/13/pipeline"
+
+
+      - run:
+          shell: /bin/bash -euo pipefail
+          command: |
+            # Trigger Debian Package Build
+            curl -v -X POST https://api.github.com/repos/zrepl/debian-binary-packaging/dispatches \
+              -H 'Accept: application/vnd.github.v3+json' \
+              -H "Authorization: token $ZREPL_DEBIAN_BINARYPACKAGIN_TRIGGER_BUILD_GITHUB_TOKEN" \
+              --data '{"event_type": "push", "client_payload": { "zrepl_main_repo_commit": "'"$CIRCLE_SHA1"'", "go_version": "'"${CIRCLE_JOB##build-}"'" }}'
+
   build-1.11:
     <<: *build-latest
     docker:

.gitlab-ci.yml

@@ -0,0 +1,17 @@
+build-and-run-platformtests:
+  variables:
+    GIT_STRATEGY: none
+  script:
+    - mkdir artifacts
+    # platformtests are installed by the CI environment
+    - /opt/platformtest-linux-amd64
+      -test.coverprofile artifacts/platformtest.cover
+      -test.v
+      __DEVEL--i-heard-you-like-tests
+      -poolname zreplplatformtest
+      -imagepath /tmp/zreplplatformtest.pool.img
+      -mountpoint /tmp/zreplplatformtest.pool
+  artifacts:
+    paths:
+      - artifacts
+  image: "images:ubuntu/focal/cloud"

.platformtest-ci/download_ci_file.bash

@@ -0,0 +1,16 @@
+#!/usr/bin/env bash
+set -euo pipefail
+cd "$(dirname "${BASH_SOURCE[0]}")"
+
+commit="$1"
+build="$2"
+src="$3"
+dst="$4"
+
+[ -n "$commit" -a -n "$build" -a -n "$src" -a -n "$dst" ] || ( echo "arguments must not be empty"; exit 1 )
+
+aws="aws --endpoint-url https://minio.cschwarz.com --no-sign-request"
+
+
+path="s3://zrepl-ci-artifacts/$commit/$build/$src"
+$aws s3 cp "$path" "$dst"

Makefile

@@ -27,6 +27,8 @@ GO_EXTRA_BUILDFLAGS :=
 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
+
 ifneq ($(GOARM),)
 	ZREPL_TARGET_TUPLE := $(GOOS)-$(GOARCH)v$(GOARM)
 else
@@ -45,7 +47,7 @@ printvars:
 
 release: clean
 	# no cross-platform support for target test
-	$(MAKE) test
+	$(MAKE) test-go
 	$(MAKE) bins-all
 	$(MAKE) noarch
 	$(MAKE) wrapup-and-checksum
@@ -99,9 +101,9 @@ clean: docs-clean
 	rm -rf "$(ARTIFACTDIR)"
 
 ##################### BINARIES #####################
-.PHONY: bins-all lint test vet zrepl-bin platformtest-bin
+.PHONY: bins-all lint test-go test-platform cover-merge cover-html vet zrepl-bin test-platform-bin generate-platform-test-list
 
-BINS_ALL_TARGETS := zrepl-bin platformtest-bin vet lint
+BINS_ALL_TARGETS := zrepl-bin test-platform-bin vet lint
 GO_SUPPORTS_ILLUMOS := $(shell $(GO) version | gawk -F '.' '/^go version /{split($$0, comps, " "); split(comps[3], v, "."); if (v[1] == "go1" && v[2] >= 13) { print "illumos"; } else { print "noillumos"; }}')
 bins-all:
 	$(MAKE) $(BINS_ALL_TARGETS) GOOS=freebsd   GOARCH=amd64
@@ -122,27 +124,34 @@ endif
 
 lint:
 	$(GO_ENV_VARS) $(GOLANGCI_LINT) run ./...
-test:
+
-	$(GO_ENV_VARS) $(GO) test $(GO_BUILDFLAGS) ./...
 vet:
 	$(GO_ENV_VARS) $(GO) vet $(GO_BUILDFLAGS) ./...
 
+test-go: $(ARTIFACTDIR)
+	rm -f "$(ARTIFACTDIR)/gotest.cover"
+ifeq ($(COVER),1)
+	$(GO_ENV_VARS) $(GO) test $(GO_BUILDFLAGS) \
+		-coverpkg github.com/zrepl/zrepl/... \
+		-covermode atomic \
+		-coverprofile "$(ARTIFACTDIR)/gotest.cover" \
+		./...
+else
+	$(GO_ENV_VARS) $(GO) test $(GO_BUILDFLAGS) \
+		./...
+endif
+
 zrepl-bin:
 	$(GO_BUILD) -o "$(ARTIFACTDIR)/zrepl-$(ZREPL_TARGET_TUPLE)"
 
-platformtest-bin:
+generate-platform-test-list:
-	$(GO_BUILD) -o "$(ARTIFACTDIR)/platformtest-$(ZREPL_TARGET_TUPLE)" ./platformtest/harness
+	$(GO_BUILD) -o $(ARTIFACTDIR)/generate-platform-test-list ./platformtest/tests/gen
 
-##################### DEV TARGETS #####################
+test-platform-bin:
-# not part of the build, must do that manually
+	$(GO_ENV_VARS) $(GO) test $(GO_BUILDFLAGS) \
-.PHONY: generate format platformtest
+		-c -o "$(ARTIFACTDIR)/platformtest-$(ZREPL_TARGET_TUPLE)" \
-
+		-covermode=atomic -cover -coverpkg github.com/zrepl/zrepl/... \
-generate:
+		./platformtest/harness
-	protoc -I=replication/logic/pdu --go_out=plugins=grpc:replication/logic/pdu replication/logic/pdu/pdu.proto
-	$(GO_ENV_VARS) $(GO) generate $(GO_BUILDFLAGS) -x ./...
-
-format:
-	goimports -srcdir . -local 'github.com/zrepl/zrepl' -w $(shell find . -type f -name '*.go' -not -path "./vendor/*" -not -name '*.pb.go' -not -name '*_enumer.go')
 
 ZREPL_PLATFORMTEST_POOLNAME := zreplplatformtest
 ZREPL_PLATFORMTEST_IMAGEPATH := /tmp/zreplplatformtest.pool.img
@@ -150,16 +159,42 @@ ZREPL_PLATFORMTEST_MOUNTPOINT := /tmp/zreplplatformtest.pool
 ZREPL_PLATFORMTEST_ZFS_LOG := /tmp/zreplplatformtest.zfs.log
 # ZREPL_PLATFORMTEST_STOP_AND_KEEP := -failure.stop-and-keep-pool
 ZREPL_PLATFORMTEST_ARGS := 
-platformtest: # do not track dependency on platformtest-bin to allow build of platformtest outside of test VM
+test-platform: $(ARTIFACTDIR) # do not track dependency on test-platform-bin to allow build of platformtest outside of test VM
 	rm -f "$(ZREPL_PLATFORMTEST_ZFS_LOG)"
+	rm -f "$(ARTIFACTDIR)/platformtest.cover"
 	platformtest/logmockzfs/logzfsenv "$(ZREPL_PLATFORMTEST_ZFS_LOG)" `which zfs` \
 	"$(ARTIFACTDIR)/platformtest-$(ZREPL_TARGET_TUPLE)" \
+		-test.coverprofile "$(ARTIFACTDIR)/platformtest.cover" \
+		-test.v \
+		__DEVEL--i-heard-you-like-tests \
 		-poolname "$(ZREPL_PLATFORMTEST_POOLNAME)" \
 		-imagepath "$(ZREPL_PLATFORMTEST_IMAGEPATH)" \
 		-mountpoint "$(ZREPL_PLATFORMTEST_MOUNTPOINT)" \
 		$(ZREPL_PLATFORMTEST_STOP_AND_KEEP) \
 		$(ZREPL_PLATFORMTEST_ARGS)
 
+cover-merge: $(ARTIFACTDIR)
+	$(GOCOVMERGE) $(ARTIFACTDIR)/platformtest.cover $(ARTIFACTDIR)/gotest.cover > $(ARTIFACTDIR)/merged.cover
+cover-html: cover-merge
+	$(GO) tool cover -html "$(ARTIFACTDIR)/merged.cover" -o "$(ARTIFACTDIR)/merged.cover.html"
+
+test-full:
+	test "$$(id -u)" = "0" || echo "MUST RUN AS ROOT" 1>&2
+	$(MAKE) test-go COVER=1
+	$(MAKE) test-platform
+	$(MAKE) cover-html
+
+##################### DEV TARGETS #####################
+# not part of the build, must do that manually
+.PHONY: generate format
+
+generate: generate-platform-test-list
+	protoc -I=replication/logic/pdu --go_out=plugins=grpc:replication/logic/pdu replication/logic/pdu/pdu.proto
+	$(GO_ENV_VARS) $(GO) generate $(GO_BUILDFLAGS) -x ./...
+
+format:
+	goimports -srcdir . -local 'github.com/zrepl/zrepl' -w $(shell find . -type f -name '*.go' -not -path "./vendor/*" -not -name '*.pb.go' -not -name '*_enumer.go')
+
 ##################### NOARCH #####################
 .PHONY: noarch $(ARTIFACTDIR)/bash_completion $(ARTIFACTDIR)/_zrepl.zsh_completion $(ARTIFACTDIR)/go_env.txt docs docs-clean
 

README.md

@@ -39,7 +39,7 @@ Check out the *Coding Workflow* section below for details.
 * Ship other material provided in `./dist`, e.g. in `/usr/share/zrepl/`.
 * Use `make release ZREPL_VERSION='mydistro-1.2.3_1'`
     * Your distro's name and any versioning supplemental to zrepl's (e.g. package revision) should be in this string
-* Use `make platformtest` **on a test system** to validate that zrepl's abstractions on top of ZFS work with the system ZFS.
+* Use `sudo make test-platform` **on a test system** to validate that zrepl's abstractions on top of ZFS work with the system ZFS.
 * Make sure you are informed about new zrepl versions, e.g. by subscribing to GitHub's release RSS feed.
 
 ## Developer Documentation
@@ -47,10 +47,18 @@ Check out the *Coding Workflow* section below for details.
 zrepl is written in [Go](https://golang.org) and uses [Go modules](https://github.com/golang/go/wiki/Modules) to manage dependencies.
 The documentation is written in [ReStructured Text](http://docutils.sourceforge.net/rst.html) using the [Sphinx](https://www.sphinx-doc.org) framework.
 
-To get started, run `./lazy.sh devsetup` to easily install build dependencies and read `docs/installation.rst -> Compiling from Source`.
+Install **build dependencies** using  `./lazy.sh devsetup`.
 `lazy.sh` uses `python3-pip` to fetch the build dependencies for the docs - you might want to use a [venv](https://docs.python.org/3/library/venv.html).
 If you just want to install the Go dependencies, run `./lazy.sh godep`.
 
+The **test suite** is split into pure **Go tests** (`make test-go`) and **platform tests** that interact with ZFS and thus generally **require root privileges** (`sudo make test-platform`).
+Platform tests run on their own pool with the name `zreplplatformtest`, which is created using the file vdev in `/tmp`.
+
+For a full **code coverage** profile, run `make test-go COVER=1 && sudo make test-platform && make cover-merge`.
+An HTML report can be generated using `make cover-html`.
+
+**Code generation** is triggered by `make generate`. Generated code is committed to the source tree.
+
 ### Project Structure
 
 ```

build/go.mod

@@ -28,6 +28,7 @@ require (
 	github.com/spf13/jwalterweatherman v1.1.0 // indirect
 	github.com/spf13/viper v1.3.2 // indirect
 	github.com/stretchr/testify v1.4.0 // indirect
+	github.com/wadey/gocovmerge v0.0.0-20160331181800-b5bfa59ec0ad // indirect
 	golang.org/x/net v0.0.0-20190613194153-d28f0bde5980 // indirect
 	golang.org/x/sys v0.0.0-20191010194322-b09406accb47 // indirect
 	golang.org/x/tools v0.0.0-20190524210228-3d17549cdc6b

build/go.sum

@@ -214,6 +214,8 @@ github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyC
 github.com/valyala/fasthttp v1.2.0/go.mod h1:4vX61m6KN+xDduDNwXrhIAVZaZaZiQ1luJk8LWSxF3s=
 github.com/valyala/quicktemplate v1.1.1/go.mod h1:EH+4AkTd43SvgIbQHYu59/cJyxDoOVRUAfrukLPuGJ4=
 github.com/valyala/tcplisten v0.0.0-20161114210144-ceec8f93295a/go.mod h1:v3UYOV9WzVtRmSR+PDvWpU/qWl4Wa5LApYYX4ZtKbio=
+github.com/wadey/gocovmerge v0.0.0-20160331181800-b5bfa59ec0ad h1:W0LEBv82YCGEtcmPA3uNZBI33/qF//HAAs3MawDjRa0=
+github.com/wadey/gocovmerge v0.0.0-20160331181800-b5bfa59ec0ad/go.mod h1:Hy8o65+MXnS6EwGElrSRjUzQDLXreJlzYLlWiHtt8hM=
 github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q=
 golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=

build/tools.go

@@ -6,6 +6,7 @@ import (
 	_ "github.com/alvaroloes/enumer"
 	_ "github.com/golang/protobuf/protoc-gen-go"
 	_ "github.com/golangci/golangci-lint/cmd/golangci-lint"
+	_ "github.com/wadey/gocovmerge"
 	_ "golang.org/x/tools/cmd/goimports"
 	_ "golang.org/x/tools/cmd/stringer"
 )

config/config.go

@@ -77,6 +77,11 @@ type SnapJob struct {
 
 type SendOptions struct {
 	Encrypted bool                 `yaml:"encrypted"`
+	StepHolds SendOptionsStepHolds `yaml:"step_holds,optional"`
+}
+
+type SendOptionsStepHolds struct {
+	DisableIncremental bool `yaml:"disable_incremental,optional"`
 }
 
 var _ yaml.Defaulter = (*SendOptions)(nil)

config/samples/source.yml

@@ -5,7 +5,11 @@ jobs:
     type: tcp
     listen: "0.0.0.0:8888"
     clients: {
-      "192.168.122.123" : "client1"
+      "192.168.122.123" :               "mysql01",
+      "192.168.122.42" :                "mx01",
+      "2001:0db8:85a3::8a2e:0370:7334": "gateway",
+      "10.23.42.0/24":       "cluster-*",
+      "fde4:8dba:82e1::/64": "san-*",
     }
   filesystems: {
     "<": true,

daemon/control.go

@@ -120,7 +120,13 @@ func (j *controlJob) Run(ctx context.Context) {
 		jsonResponder{log, func() (interface{}, error) {
 			jobs := j.jobs.status()
 			globalZFS := zfscmd.GetReport()
-			s := Status{Jobs: jobs, Global: GlobalStatus{ZFSCmds: globalZFS}}
+			envconstReport := envconst.GetReport()
+			s := Status{
+				Jobs: jobs,
+				Global: GlobalStatus{
+					ZFSCmds:  globalZFS,
+					Envconst: envconstReport,
+				}}
 			return s, nil
 		}})
 

daemon/daemon.go

@@ -13,6 +13,8 @@ import (
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/zrepl/zrepl/daemon/logging/trace"
+	"github.com/zrepl/zrepl/endpoint"
+	"github.com/zrepl/zrepl/util/envconst"
 
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/job"
@@ -50,6 +52,15 @@ func Run(ctx context.Context, conf *config.Config) error {
 	log.Info(version.NewZreplVersionInformation().String())
 
 	ctx = logging.WithLoggers(ctx, logging.SubsystemLoggersWithUniversalLogger(log))
+	trace.RegisterCallback(trace.Callback{
+		OnBegin: func(ctx context.Context) { logging.GetLogger(ctx, logging.SubsysTraceData).Debug("begin span") },
+		OnEnd: func(ctx context.Context, spanInfo trace.SpanInfo) {
+			logging.
+				GetLogger(ctx, logging.SubsysTraceData).
+				WithField("duration_s", spanInfo.EndedAt().Sub(spanInfo.StartedAt()).Seconds()).
+				Debug("finished span " + spanInfo.TaskAndSpanStack(trace.SpanStackKindAnnotation))
+		},
+	})
 
 	for _, job := range confJobs {
 		if IsInternalJobName(job.Name()) {
@@ -86,6 +97,7 @@ func Run(ctx context.Context, conf *config.Config) error {
 	// register global (=non job-local) metrics
 	zfscmd.RegisterMetrics(prometheus.DefaultRegisterer)
 	trace.RegisterMetrics(prometheus.DefaultRegisterer)
+	endpoint.RegisterMetrics(prometheus.DefaultRegisterer)
 
 	log.Info("starting daemon")
 
@@ -140,6 +152,7 @@ type Status struct {
 
 type GlobalStatus struct {
 	ZFSCmds  *zfscmd.Report
+	Envconst *envconst.Report
 }
 
 func (s *jobs) status() map[string]*job.Status {

daemon/job/active.go

@@ -154,6 +154,7 @@ func modePushFromConfig(g *config.Global, in *config.PushJob, jobID endpoint.Job
 	m.senderConfig = &endpoint.SenderConfig{
 		FSF:                         fsf,
 		Encrypt:                     &zfs.NilBool{B: in.Send.Encrypted},
+		DisableIncrementalStepHolds: in.Send.StepHolds.DisableIncremental,
 		JobID:                       jobID,
 	}
 	m.plannerPolicy = &logic.PlannerPolicy{
@@ -376,9 +377,21 @@ func (j *ActiveSide) SenderConfig() *endpoint.SenderConfig {
 	return push.senderConfig
 }
 
+// The active side of a replication uses one end (sender or receiver)
+// directly by method invocation, without going through a transport that
+// provides a client identity.
+// However, in order to avoid the need to distinguish between direct-method-invocating
+// clients and RPC client, we use an invalid client identity as a sentinel value.
+func FakeActiveSideDirectMethodInvocationClientIdentity(jobId endpoint.JobID) string {
+	return fmt.Sprintf("<local><active><job><client><identity><job=%q>", jobId.String())
+}
+
 func (j *ActiveSide) Run(ctx context.Context) {
 	ctx, endTask := trace.WithTaskAndSpan(ctx, "active-side-job", j.Name())
 	defer endTask()
+
+	ctx = context.WithValue(ctx, endpoint.ClientIdentityKey, FakeActiveSideDirectMethodInvocationClientIdentity(j.name))
+
 	log := GetLogger(ctx)
 
 	defer log.Info("job exiting")

daemon/job/active_test.go

@@ -0,0 +1,19 @@
+package job
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/zrepl/zrepl/endpoint"
+	"github.com/zrepl/zrepl/transport"
+)
+
+func TestFakeActiveSideDirectMethodInvocationClientIdentityDoesNotPassValidityTest(t *testing.T) {
+	jobid, err := endpoint.MakeJobID("validjobname")
+	require.NoError(t, err)
+	clientIdentity := FakeActiveSideDirectMethodInvocationClientIdentity(jobid)
+	t.Logf("%v", clientIdentity)
+	err = transport.ValidateClientIdentity(clientIdentity)
+	assert.Error(t, err)
+}

daemon/job/passive.go

@@ -81,6 +81,7 @@ func modeSourceFromConfig(g *config.Global, in *config.SourceJob, jobID endpoint
 	m.senderConfig = &endpoint.SenderConfig{
 		FSF:                         fsf,
 		Encrypt:                     &zfs.NilBool{B: in.Send.Encrypted},
+		DisableIncrementalStepHolds: in.Send.StepHolds.DisableIncremental,
 		JobID:                       jobID,
 	}
 

daemon/job/snapjob.go

@@ -175,6 +175,8 @@ func (j *SnapJob) doPrune(ctx context.Context) {
 		FSF:   j.fsfilter,
 		// FIXME encryption setting is irrelevant for SnapJob because the endpoint is only used as pruner.Target
 		Encrypt: &zfs.NilBool{B: true},
+		// FIXME DisableIncrementalStepHolds setting is irrelevant for SnapJob because the endpoint is only used as pruner.Target
+		DisableIncrementalStepHolds: false,
 	})
 	j.pruner = j.prunerFactory.BuildLocalPruner(ctx, sender, alwaysUpToDateReplicationCursorHistory{sender})
 	log.Info("start pruning")

daemon/logging/build_logging.go

@@ -73,6 +73,8 @@ const (
 	SubsysRPCControl   Subsystem = "rpc.ctrl"
 	SubsysRPCData      Subsystem = "rpc.data"
 	SubsysZFSCmd       Subsystem = "zfs.cmd"
+	SubsysTraceData    Subsystem = "trace.data"
+	SubsysPlatformtest Subsystem = "platformtest"
 )
 
 var AllSubsystems = []Subsystem{
@@ -89,6 +91,8 @@ var AllSubsystems = []Subsystem{
 	SubsysRPCControl,
 	SubsysRPCData,
 	SubsysZFSCmd,
+	SubsysTraceData,
+	SubsysPlatformtest,
 }
 
 type injectedField struct {
@@ -157,7 +161,7 @@ func getLoggerImpl(ctx context.Context, subsys Subsystem, panicIfEnded bool) log
 
 	l = l.WithField(SubsysField, subsys)
 
-	l = l.WithField(SpanField, trace.GetSpanStackOrDefault(ctx, "NOSPAN"))
+	l = l.WithField(SpanField, trace.GetSpanStackOrDefault(ctx, *trace.StackKindId, "NOSPAN"))
 
 	fields := make(logger.Fields)
 	iterInjectedFields(ctx, func(field string, value interface{}) {

daemon/logging/trace/trace.go

@@ -103,8 +103,8 @@ import (
 
 var metrics struct {
 	activeTasks prometheus.Gauge
-	uniqueConcurrentTaskNameBitvecLength *prometheus.GaugeVec
 }
+var taskNamer *uniqueConcurrentTaskNamer = newUniqueTaskNamer()
 
 func init() {
 	metrics.activeTasks = prometheus.NewGauge(prometheus.GaugeOpts{
@@ -113,21 +113,12 @@ func init() {
 		Name:      "active_tasks",
 		Help:      "number of active (tracing-level) tasks in the daemon",
 	})
-	metrics.uniqueConcurrentTaskNameBitvecLength = prometheus.NewGaugeVec(prometheus.GaugeOpts{
-		Namespace: "zrepl",
-		Subsystem: "trace",
-		Name:      "unique_concurrent_task_name_bitvec_length",
-		Help:      "length of the bitvec used to find unique names for concurrent tasks",
-	}, []string{"task_name"})
 }
 
 func RegisterMetrics(r prometheus.Registerer) {
 	r.MustRegister(metrics.activeTasks)
-	r.MustRegister(metrics.uniqueConcurrentTaskNameBitvecLength)
 }
 
-var taskNamer = newUniqueTaskNamer(metrics.uniqueConcurrentTaskNameBitvecLength)
-
 type traceNode struct {
 	id         string
 	annotation string
@@ -143,6 +134,9 @@ type traceNode struct {
 	endedAt   time.Time
 }
 
+func (s *traceNode) StartedAt() time.Time { return s.startedAt }
+func (s *traceNode) EndedAt() time.Time   { return s.endedAt }
+
 // Returned from WithTask or WithSpan.
 // Must be called once the task or span ends.
 // See package-level docs for nesting rules.
@@ -150,7 +144,6 @@ type traceNode struct {
 type DoneFunc func()
 
 var ErrTaskStillHasActiveChildTasks = fmt.Errorf("end task: task still has active child tasks")
-var ErrParentTaskAlreadyEnded = fmt.Errorf("create task: parent task already ended")
 
 // Start a new root task or create a child task of an existing task.
 //
@@ -174,6 +167,20 @@ func WithTask(ctx context.Context, taskName string) (context.Context, DoneFunc)
 			parentTask = node
 		}
 	}
+	// find the first ancestor that hasn't ended yet (nil if need be)
+	if parentTask != nil {
+		parentTask.mtx.Lock()
+	}
+	for parentTask != nil && !parentTask.endedAt.IsZero() {
+		thisParent := parentTask
+		parentTask = parentTask.parentTask
+		// lock parent first such that it isn't modified by other callers to thisParent
+		if parentTask != nil {
+			parentTask.mtx.Lock()
+		}
+		thisParent.mtx.Unlock()
+	}
+	// invariant: either parentTask != nil and we hold the lock on parentTask, or parentTask is nil
 
 	taskName, taskNameDone := taskNamer.UniqueConcurrentTaskName(taskName)
 
@@ -189,13 +196,9 @@ func WithTask(ctx context.Context, taskName string) (context.Context, DoneFunc)
 		endedAt:   time.Time{},
 	}
 
-	if this.parentTask != nil {
+	if parentTask != nil {
-		this.parentTask.mtx.HoldWhile(func() {
-			if !this.parentTask.endedAt.IsZero() {
-				panic(ErrParentTaskAlreadyEnded)
-			}
 		this.parentTask.activeChildTasks++
-		})
+		parentTask.mtx.Unlock()
 	}
 
 	ctx = context.WithValue(ctx, contextKeyTraceNode, this)
@@ -215,7 +218,7 @@ func WithTask(ctx context.Context, taskName string) (context.Context, DoneFunc)
 			defer this.mtx.Lock().Unlock()
 
 			if this.activeChildTasks != 0 {
-				panic(errors.Wrapf(ErrTaskStillHasActiveChildTasks, "end task: %v active child tasks", this.activeChildSpan))
+				panic(errors.Wrapf(ErrTaskStillHasActiveChildTasks, "end task: %v active child tasks", this.activeChildTasks))
 			}
 
 			// support idempotent task ends
@@ -285,6 +288,7 @@ func WithSpan(ctx context.Context, annotation string) (context.Context, DoneFunc
 
 	ctx = context.WithValue(ctx, contextKeyTraceNode, this)
 	chrometraceBeginSpan(this)
+	callbackEndSpan := callbackBeginSpan(ctx)
 
 	endTaskFunc := func() {
 
@@ -306,18 +310,48 @@ func WithSpan(ctx context.Context, annotation string) (context.Context, DoneFunc
 		this.endedAt = time.Now()
 
 		chrometraceEndSpan(this)
+		callbackEndSpan(this)
 	}
 
 	return ctx, endTaskFunc
 }
 
-func currentTaskNameAndSpanStack(this *traceNode) (taskName string, spanIdStack string) {
+type StackKind struct {
-
+	symbolizeTask func(t *traceNode) string
-	task := this.parentTask
+	symbolizeSpan func(s *traceNode) string
-	if this.parentSpan == nil {
-		task = this
 }
 
+var (
+	StackKindId = &StackKind{
+		symbolizeTask: func(t *traceNode) string { return t.id },
+		symbolizeSpan: func(s *traceNode) string { return s.id },
+	}
+	SpanStackKindCombined = &StackKind{
+		symbolizeTask: func(t *traceNode) string { return fmt.Sprintf("(%s %q)", t.id, t.annotation) },
+		symbolizeSpan: func(s *traceNode) string { return fmt.Sprintf("(%s %q)", s.id, s.annotation) },
+	}
+	SpanStackKindAnnotation = &StackKind{
+		symbolizeTask: func(t *traceNode) string { return t.annotation },
+		symbolizeSpan: func(s *traceNode) string { return s.annotation },
+	}
+)
+
+func (n *traceNode) task() *traceNode {
+	task := n.parentTask
+	if n.parentSpan == nil {
+		task = n
+	}
+	return task
+}
+
+func (n *traceNode) TaskName() string {
+	task := n.task()
+	return task.annotation
+}
+
+func (this *traceNode) TaskAndSpanStack(kind *StackKind) (spanIdStack string) {
+	task := this.task()
+
 	var spansInTask []*traceNode
 	for s := this; s != nil; s = s.parentSpan {
 		spansInTask = append(spansInTask, s)
@@ -330,25 +364,22 @@ func currentTaskNameAndSpanStack(this *traceNode) (taskName string, spanIdStack
 
 	var taskIdsRev []string
 	for i := len(tasks) - 1; i >= 0; i-- {
-		taskIdsRev = append(taskIdsRev, tasks[i].id)
+		taskIdsRev = append(taskIdsRev, kind.symbolizeTask(tasks[i]))
 	}
 
 	var spanIdsRev []string
 	for i := len(spansInTask) - 1; i >= 0; i-- {
-		spanIdsRev = append(spanIdsRev, spansInTask[i].id)
+		spanIdsRev = append(spanIdsRev, kind.symbolizeSpan(spansInTask[i]))
 	}
 
 	taskStack := strings.Join(taskIdsRev, "$")
-	spanIdStack = fmt.Sprintf("%s$%s", taskStack, strings.Join(spanIdsRev, "."))
+	return fmt.Sprintf("%s$%s", taskStack, strings.Join(spanIdsRev, "."))
-
-	return task.annotation, spanIdStack
 }
 
-func GetSpanStackOrDefault(ctx context.Context, def string) string {
+func GetSpanStackOrDefault(ctx context.Context, kind StackKind, def string) string {
 	if nI := ctx.Value(contextKeyTraceNode); nI != nil {
 		n := nI.(*traceNode)
-		_, spanStack := currentTaskNameAndSpanStack(n)
+		return n.TaskAndSpanStack(StackKindId)
-		return spanStack
 	} else {
 		return def
 	}

daemon/logging/trace/trace_callback.go

@@ -0,0 +1,54 @@
+package trace
+
+import (
+	"context"
+	"time"
+
+	"github.com/zrepl/zrepl/util/chainlock"
+)
+
+type SpanInfo interface {
+	StartedAt() time.Time
+	EndedAt() time.Time
+	TaskAndSpanStack(kind *StackKind) string
+}
+
+type Callback struct {
+	OnBegin func(ctx context.Context)
+	OnEnd   func(ctx context.Context, spanInfo SpanInfo)
+}
+
+var callbacks struct {
+	mtx chainlock.L
+	cs  []Callback
+}
+
+func RegisterCallback(c Callback) {
+	callbacks.mtx.HoldWhile(func() {
+		callbacks.cs = append(callbacks.cs, c)
+	})
+}
+
+func callbackBeginSpan(ctx context.Context) func(SpanInfo) {
+	// capture the current state of callbacks into a local variable
+	// this is safe because the slice is append-only and immutable
+
+	// (it is important that a callback registered _after_ callbackBeginSpin is called does not get called on OnEnd)
+
+	var cbs []Callback
+	callbacks.mtx.HoldWhile(func() {
+		cbs = callbacks.cs
+	})
+	for _, cb := range cbs {
+		if cb.OnBegin != nil {
+			cb.OnBegin(ctx)
+		}
+	}
+	return func(spanInfo SpanInfo) {
+		for _, cb := range cbs {
+			if cb.OnEnd != nil {
+				cb.OnEnd(ctx, spanInfo)
+			}
+		}
+	}
+}

daemon/logging/trace/trace_chrometrace.go

@@ -51,7 +51,7 @@ type chrometraceEvent struct {
 }
 
 func chrometraceBeginSpan(s *traceNode) {
-	taskName, _ := currentTaskNameAndSpanStack(s)
+	taskName := s.TaskName()
 	chrometraceWrite(chrometraceEvent{
 		Name:                      s.annotation,
 		Phase:                     "B",
@@ -62,7 +62,7 @@ func chrometraceBeginSpan(s *traceNode) {
 }
 
 func chrometraceEndSpan(s *traceNode) {
-	taskName, _ := currentTaskNameAndSpanStack(s)
+	taskName := s.TaskName()
 	chrometraceWrite(chrometraceEvent{
 		Name:                      s.annotation,
 		Phase:                     "E",
@@ -86,7 +86,7 @@ func chrometraceBeginTask(s *traceNode) {
 	flowId := atomic.AddUint64(&chrometraceFlowId, 1)
 	flowIdStr := fmt.Sprintf("%x", flowId)
 
-	parentTask, _ := currentTaskNameAndSpanStack(s.parentTask)
+	parentTask := s.parentTask.TaskName()
 	chrometraceWrite(chrometraceEvent{
 		Cat:                       "task", // seems to be necessary, otherwise the GUI shows some `indexOf` JS error
 		Name:                      "child-task",
@@ -97,7 +97,7 @@ func chrometraceBeginTask(s *traceNode) {
 		Id:                        flowIdStr,
 	})
 
-	childTask, _ := currentTaskNameAndSpanStack(s)
+	childTask := s.TaskName()
 	if parentTask == childTask {
 		panic(parentTask)
 	}

daemon/logging/trace/trace_test.go

@@ -112,12 +112,6 @@ func TestIdempotentEndTask(t *testing.T) {
 	require.NotPanics(t, func() { end() })
 }
 
-func TestCannotReuseEndedTask(t *testing.T) {
-	root, end := WithTask(context.Background(), "root")
-	end()
-	require.PanicsWithValue(t, ErrParentTaskAlreadyEnded, func() { WithTask(root, "child-after-parent-ended") })
-}
-
 func TestSpansPanicIfNoParentTask(t *testing.T) {
 	require.Panics(t, func() { WithSpan(context.Background(), "taskless-span") })
 }
@@ -170,3 +164,42 @@ func TestWhiteboxHierachy(t *testing.T) {
 	e4()
 	e1()
 }
+
+func TestOrphanTasksBecomeNewRootWhenAllAncestorsAreDead(t *testing.T) {
+	parent, e1 := WithTask(context.Background(), "parent")
+	_ = logAndGetTraceNode(t, "parent-task", parent)
+	e1()
+
+	var child context.Context
+	var e2 DoneFunc
+	require.NotPanics(t, func() {
+		child, e2 = WithTask(parent, "child")
+	})
+	childN := logAndGetTraceNode(t, "child-task", child)
+	assert.Nil(t, childN.parentTask)
+
+	grandchild, _ := WithTask(child, "grandchild")
+	grandchildN := logAndGetTraceNode(t, "grandchild-task", grandchild)
+	assert.Equal(t, childN, grandchildN.parentTask)
+
+	require.Panics(t, func() { e2() }, "if the parent was alive at creation of child task, wrong termination order remains a panicable offense though")
+}
+
+func TestOrphanTaskBecomesChildToNearestLiveAncestor(t *testing.T) {
+	parent, e1 := WithTask(context.Background(), "parent")
+	parentN := logAndGetTraceNode(t, "parent-task", parent)
+
+	child, e2 := WithTask(parent, "child")
+	childN := logAndGetTraceNode(t, "child-task", child)
+	assert.Equal(t, parentN, childN.parentTask)
+	e2()
+
+	grandchild, e3 := WithTask(child, "grandchild")
+	grandchildN := logAndGetTraceNode(t, "grandchild-task", grandchild)
+	assert.Equal(t, parentN, grandchildN.parentTask) // child is already dead
+
+	require.NotPanics(t, func() {
+		e3()
+		e1()
+	})
+}

daemon/logging/trace/trace_unique_concurrent_task_namer.go

@@ -5,22 +5,18 @@ import (
 	"strings"
 	"sync"
 
-	"github.com/prometheus/client_golang/prometheus"
 	"github.com/willf/bitset"
 )
 
 type uniqueConcurrentTaskNamer struct {
 	mtx    sync.Mutex
 	active map[string]*bitset.BitSet
-
-	bitvecLengthGauge *prometheus.GaugeVec
 }
 
 // bitvecLengthGauge may be nil
-func newUniqueTaskNamer(bitvecLengthGauge *prometheus.GaugeVec) *uniqueConcurrentTaskNamer {
+func newUniqueTaskNamer() *uniqueConcurrentTaskNamer {
 	return &uniqueConcurrentTaskNamer{
 		active: make(map[string]*bitset.BitSet),
-		bitvecLengthGauge: bitvecLengthGauge,
 	}
 }
 
@@ -45,10 +41,6 @@ func (namer *uniqueConcurrentTaskNamer) UniqueConcurrentTaskName(name string) (u
 	act.Set(id)
 	namer.mtx.Unlock()
 
-	if namer.bitvecLengthGauge != nil {
-		namer.bitvecLengthGauge.WithLabelValues(name).Set(float64(act.Len()))
-	}
-
 	return fmt.Sprintf("%s#%d", name, id), func() {
 		namer.mtx.Lock()
 		defer namer.mtx.Unlock()

daemon/logging/trace/trace_unique_concurrent_task_namer_test.go

@@ -29,7 +29,7 @@ func TestBitsetFeaturesForUniqueConcurrentTaskNamer(t *testing.T) {
 }
 
 func TestUniqueConcurrentTaskNamer(t *testing.T) {
-	namer := newUniqueTaskNamer(nil)
+	namer := newUniqueTaskNamer()
 
 	var wg sync.WaitGroup
 

daemon/pprof.go

@@ -8,10 +8,11 @@ import (
 	"net"
 	"net/http/pprof"
 
-	"github.com/zrepl/zrepl/daemon/logging/trace"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"golang.org/x/net/websocket"
 
 	"github.com/zrepl/zrepl/daemon/job"
+	"github.com/zrepl/zrepl/daemon/logging/trace"
 )
 
 type pprofServer struct {
@@ -67,6 +68,7 @@ outer:
 			mux.Handle("/debug/pprof/profile", http.HandlerFunc(pprof.Profile))
 			mux.Handle("/debug/pprof/symbol", http.HandlerFunc(pprof.Symbol))
 			mux.Handle("/debug/pprof/trace", http.HandlerFunc(pprof.Trace))
+			mux.Handle("/metrics", promhttp.Handler())
 			mux.Handle("/debug/zrepl/activity-trace", websocket.Handler(trace.ChrometraceClientWebsocketHandler))
 			go func() {
 				err := http.Serve(s.listener, mux)

docs/changelog.rst

@@ -35,18 +35,23 @@ This is a big one! Headlining features:
 * **Resumable Send & Recv Support**
   No knobs required, automatically used where supported.
 * **Hold-Protected Send & Recv**
-  Automatic ZFS holds to ensure that we can always resume a replication step.
+  Automatic ZFS holds to ensure that we can always use resumable send&recv for a replication step.
-* **Encrypted Send & Recv Support** for OpenZFS native encryption.
+* **Encrypted Send & Recv Support** for OpenZFS native encryption,
-  :ref:`Configurable <job-send-options>` at the job level, i.e., for all filesystems a job is responsible for.
+  :ref:`configurable <job-send-options>` at the job level, i.e., for all filesystems a job is responsible for.
 * **Receive-side hold on last received dataset**
   The counterpart to the replication cursor bookmark on the send-side.
   Ensures that incremental replication will always be possible between a sender and receiver.
 
-Actual changelog:
+.. TIP::
 
-* |break_config| **more restrictive job names than in prior zrepl versions**
+   We highly recommend studying the :ref:`overview section of the configuration chapter <overview-how-replication-works>` to understand how replication works.
-  Starting with this version, job names are going to be embedded into ZFS holds and bookmark names.
 
+Additional changelog:
+
+* |break| |break_config| **more restrictive job names than in prior zrepl versions**
+  Starting with this version, job names are going to be embedded into ZFS holds and bookmark names (see :ref:`here<replication-cursor-and-last-received-hold>` and :ref:`here<step-holds-and-bookmarks>`).
+  Therefore you might need to adjust your job names.
+  **Note that jobs** :issue:`cannot be renamed easily<327>` **once you start using zrepl 0.3.**
 * |break| |mig| replication cursor representation changed
 
   * zrepl now manages the :ref:`replication cursor bookmark <replication-cursor-and-last-received-hold>` per job-filesystem tuple instead of a single replication cursor per filesystem.
@@ -56,16 +61,29 @@ Actual changelog:
   * Run ``zrepl migrate replication-cursor:v1-v2`` to safely destroy old-format cursors.
     The migration will ensure that only those old-format cursors are destroyed that have been superseeded by new-format cursors.
 
+* |feature| New option ``listen_freebind`` (tcp, tls, prometheus listener)
+* |feature| :issue:`<265>` transport/tcp: support for CIDR masks in client IP whitelist
+* |feature| documented subcommand to generate ``bash`` and ``zsh`` completions
+* |feature| :issue:`<307>` ``chrome://trace`` -compatible activity tracing of zrepl daemon activity
+* |feature| logging: trace IDs for better log entry correlation with concurrent replication jobs
+* |feature| experimental environment variable for parallel replication (see :issue:`<306>` )
 * |bugfix| missing logger context vars in control connection handlers
 * |bugfix| improved error messages on ``zfs send`` errors
-* |feature| New option ``listen_freebind`` (tcp, tls, prometheus listener)
 * |bugfix| |docs| snapshotting: clarify sync-up behavior and warn about filesystems
+* |bugfix| transport/ssh: do not leak zombie ssh process on connection failures
   that will not be snapshotted until the sync-up phase is over
+* |docs| Installation: :ref:`FreeBSD jail with iocage <installation-freebsd-jail-with-iocage>`
 * |docs| Document new replication features in the :ref:`config overview <overview-how-replication-works>` and :repomasterlink:`replication/design.md`.
-* |feature| documented subcommand to generate ``bash`` and ``zsh`` completions
 * **[MAINTAINER NOTICE]** New platform tests in this version, please make sure you run them for your distro!
 * **[MAINTAINER NOTICE]** Please add the shell completions to the zrepl packages.
 
+.. 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 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.2.1
 -----
 
@@ -102,14 +120,6 @@ Actual changelog:
   The pool name ``zreplplatformtest`` is reserved for this use case.
   Only run ``make platformtest`` on test systems, e.g. a FreeBSD VM image.
 
-.. 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 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.1.1
 -----
 

docs/configuration/jobs.rst

@@ -19,7 +19,7 @@ Job Type ``push``
     * - ``type``
       - = ``push``
     * - ``name``
-      - unique name of the job
+      - unique name of the job :issue:`(must not change)<327>`
     * - ``connect``
       - |connect-transport|
     * - ``filesystems``
@@ -47,7 +47,7 @@ Job Type ``sink``
     * - ``type``
       - = ``sink``
     * - ``name``
-      - unique name of the job
+      - unique name of the job :issue:`(must not change)<327>`
     * - ``serve``
       - |serve-transport|
     * - ``root_fs``
@@ -70,7 +70,7 @@ Job Type ``pull``
     * - ``type``
       - = ``pull``
     * - ``name``
-      - unique name of the job
+      - unique name of the job :issue:`(must not change)<327>`
     * - ``connect``
       - |connect-transport|
     * - ``root_fs``
@@ -98,7 +98,7 @@ Job Type ``source``
     * - ``type``
       - = ``source``
     * - ``name``
-      - unique name of the job
+      - unique name of the job :issue:`(must not change)<327>`
     * - ``serve``
       - |serve-transport|
     * - ``filesystems``
@@ -137,7 +137,7 @@ Job type that only takes snapshots and performs pruning on the local machine.
     * - ``type``
       - = ``snap``
     * - ``name``
-      - unique name of the job
+      - unique name of the job :issue:`(must not change)<327>`
     * - ``filesystems``
       - |filter-spec| for filesystems to be snapshotted
     * - ``snapshotting``

docs/configuration/overview.rst

@@ -39,6 +39,10 @@ A *job* is the unit of activity tracked by the zrepl daemon.
 The ``type`` of a job determines its role in a replication setup and in snapshot management.
 Jobs are identified by their ``name``, both in log files and the ``zrepl status`` command.
 
+.. NOTE::
+   The job name is persisted in several places on disk and thus :issue:`cannot be changed easily<327>`.
+
+
 Replication always happens between a pair of jobs: one is the **active side**, and one the **passive side**.
 The active side connects to the passive side using a :ref:`transport <transport>` and starts executing the replication logic.
 The passive side responds to requests from the active side after checking its permissions.
@@ -99,7 +103,7 @@ One of the major design goals of the replication module is to avoid any duplicat
 As such, the code works on abstract senders and receiver **endpoints**, where typically one will be implemented by a local program object and the other is an RPC client instance.
 Regardless of push- or pull-style setup, the logic executes on the active side, i.e. in the ``push`` or ``pull`` job.
 
-The following steps take place during replication and can be monitored using the ``zrepl status`` subcommand:
+The following high-level steps take place during replication and can be monitored using the ``zrepl status`` subcommand:
 
 * Plan the replication:
 
@@ -107,9 +111,9 @@ The following steps take place during replication and can be monitored using the
   * Build the **replication plan**
 
     * Per filesystem, compute a diff between sender and receiver snapshots
-    * Build a list of replication steps
+    * Build a list of **replication steps**
 
-      * If possible, use incremental and resumable sends (``zfs send -i``)
+      * If possible, use incremental and resumable sends
       * Otherwise, use full send of most recent snapshot on sender
 
   * Retry on errors that are likely temporary (i.e. network failures).
@@ -128,16 +132,8 @@ The following steps take place during replication and can be monitored using the
    
 The idea behind the execution order of replication steps is that if the sender snapshots all filesystems simultaneously at fixed intervals, the receiver will have all filesystems snapshotted at time ``T1`` before the first snapshot at ``T2 = T1 + $interval`` is replicated.
 
-.. _replication-cursor-and-last-received-hold:
+Placeholder Filesystems
-
+^^^^^^^^^^^^^^^^^^^^^^^
-**Replication cursor** bookmark and **last-received-hold** are managed by zrepl to ensure that future replications can always be done incrementally:
-the replication cursor is a send-side bookmark of the most recent successfully replicated snapshot,
-and the last-received-hold is a hold of that snapshot on the receiving side.
-The replication cursor has the format ``#zrepl_CUSOR_G_<GUID>_J_<JOBNAME>``.
-The last-received-hold tag has the format ``#zrepl_last_received_J_<JOBNAME>``.
-Encoding the job name in the names ensures that multiple sending jobs can replicate the same filesystem to different receivers without interference.
-The ``zrepl holds list`` provides a listing of all bookmarks and holds managed by zrepl.
-
 .. _replication-placeholder-property:
 
 **Placeholder filesystems** on the receiving side are regular ZFS filesystems with the placeholder property ``zrepl:placeholder=on``.
@@ -148,16 +144,83 @@ Thus, zrepl creates the parent filesystems as placeholders on the receiving side
 If at some point ``S/H`` and ``S`` shall be replicated, the receiving side invalidates the placeholder flag automatically.
 The ``zrepl test placeholder`` command can be used to check whether a filesystem is a placeholder.
 
+ZFS Background Knowledge
+^^^^^^^^^^^^^^^^^^^^^^^^
+
+This section gives some background knowledge about ZFS features that zrepl uses to guarantee that
+**incremental replication is always possible and that started replication steps can always be resumed if they are interrupted.**
+
+**ZFS Send Modes & Bookmarks**
+ZFS supports full sends (``zfs send fs@to``) and incremental sends (``zfs send -i @from fs@to``).
+Full sends are used to create a new filesystem on the receiver with the send-side state of ``fs@to``.
+Incremental sends only transfer the delta between ``@from`` and ``@to``.
+Incremental sends require that ``@from`` be present on the receiving side when receiving the incremental stream.
+Incremental sends can also use a ZFS bookmark as *from* on the sending side (``zfs send -i #bm_from fs@to``), where ``#bm_from`` was created using ``zfs bookmark fs@from fs#bm_from``.
+The receiving side must always have the actual snapshot ``@from``, regardless of whether the sending side uses ``@from`` or a bookmark of it.
+
+**Resumable Send & Recv**
+The ``-s`` flag for ``zfs recv`` tells zfs to save the partially received send stream in case it is interrupted.
+To resume the replication, the receiving side filesystem's ``receive_resume_token`` must be passed to a new ``zfs send -t <value> | zfs recv`` command.
+A full send can only be resumed if ``@to`` still exists.
+An incremental send can only be resumed if ``@to`` still exists *and* either ``@from`` still exists *or* a bookmark ``#fbm`` of ``@from`` still exists.
+
+**ZFS Holds**
+ZFS holds prevent a snapshot from being deleted through ``zfs destroy``, letting the destroy fail with a ``datset is busy`` error.
+Holds are created and referred to by a user-defined *tag*. They can be thought of as a named, persistent lock on the snapshot.
+
+.. _replication-cursor-and-last-received-hold:
+
+Guaranteeing That Incremental Sends Are Always Possible
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+**Replication cursor** bookmark and **last-received-hold** are managed by zrepl to ensure that future replications can always be done incrementally.
+The replication cursor is a send-side bookmark of the most recent successfully replicated snapshot,
+and the last-received-hold is a hold of that snapshot on the receiving side.
+Both are moved aomically after the receiving side has confirmed that a replication step is complete.
+
+The replication cursor has the format ``#zrepl_CUSOR_G_<GUID>_J_<JOBNAME>``.
+The last-received-hold tag has the format ``zrepl_last_received_J_<JOBNAME>``.
+Encoding the job name in the names ensures that multiple sending jobs can replicate the same filesystem to different receivers without interference.
+
+The ``zrepl zfs-abstraction list`` command provides a listing of all bookmarks and holds managed by zrepl.
+
+.. _step-holds-and-bookmarks:
+
+Guaranteeing That Sends Are Always Resumable
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+**Step holds** are zfs holds managed by zrepl to ensure that a replication step can always be resumed if it is interrupted, e.g., due to network outage.
+zrepl creates step holds before it attempts a replication step and releases them after the receiver confirms that the replication step is complete.
+For an initial replication ``full @initial_snap``, zrepl puts a zfs hold on ``@initial_snap``.
+For an incremental send ``@from -> @to``, zrepl puts a zfs hold on both ``@from`` and ``@to``.
+Note that ``@from`` is not strictly necessary for resumability -- a bookmark on the sending side would be sufficient --, but size-estimation in currently used OpenZFS versions only works if ``@from`` is a snapshot.
+
+The hold tag has the format ``zrepl_STEP_J_<JOBNAME>``.
+A job only ever has one active send per filesystem.
+Thus, there are never more than two step holds for a given pair of ``(job,filesystem)``.
+
+**Step bookmarks** are zrepl's equivalent for holds on bookmarks (ZFS does not support putting holds on bookmarks).
+They are intended for a situation where a replication step uses a bookmark ``#bm`` as incremental ``from`` that is not managed by zrepl.
+To ensure resumability, zrepl copies ``#bm`` to step bookmark ``#zrepl_STEP_G_<GUID>_J_<JOBNAME>``.
+If the replication is interrupted and ``#bm`` is deleted by the user, the step bookmark remains as an incremental source for the resumable send.
+Note that zrepl does not yet support creating step bookmarks because the `corresponding ZFS feature for copying bookmarks <https://github.com/openzfs/zfs/pull/9571>`_ is not yet widely available .
+Subscribe to zrepl :issue:`326` for details.
+
+The ``zrepl zfs-abstraction list`` command provides a listing of all bookmarks and holds managed by zrepl.
+
+.. NOTE::
+
+    More details can be found in the design document :repomasterlink:`replication/design.md`.
+
+Limitations
+^^^^^^^^^^^
+
 .. ATTENTION::
 
     Currently, zrepl does not replicate filesystem properties.
     When receiving a filesystem, it is never mounted (`-u` flag)  and `mountpoint=none` is set.
     This is temporary and being worked on :issue:`24`.
 
-.. NOTE::
-
-    More details can be found in the design document :repomasterlink:`replication/design.md`.
-
 
 .. _jobs-multiple-jobs:
 
@@ -177,7 +240,7 @@ Multiple Jobs & More than 2 Machines
 If you would like to see improvements to multi-job setups, please `open an issue on GitHub <https://github.com/zrepl/zrepl/issues/new>`_.
 
 No Overlapping
-~~~~~~~~~~~~~~
+^^^^^^^^^^^^^^
 
 Jobs run independently of each other.
 If two jobs match the same filesystem with their ``filesystems`` filter, they will operate on that filesystem independently and potentially in parallel.
@@ -185,14 +248,14 @@ For example, if job A prunes snapshots that job B is planning to replicate, the
 However, the next replication attempt will re-examine the situation from scratch and should work.
 
 N push jobs to 1 sink
-~~~~~~~~~~~~~~~~~~~~~
+^^^^^^^^^^^^^^^^^^^^^
 
 The :ref:`sink job <job-sink>` namespaces by client identity.
 It is thus safe to push to one sink job with different client identities.
 If the push jobs have the same client identity, the filesystems matched by the push jobs must be disjoint to avoid races.
 
 N pull jobs from 1 source
-~~~~~~~~~~~~~~~~~~~~~~~~~
+^^^^^^^^^^^^^^^^^^^^^^^^^
 
 Multiple pull jobs pulling from the same source have potential for race conditions during pruning:
 each pull job prunes the source side independently, causing replication-prune and prune-prune races.

docs/configuration/sendrecvoptions.rst

@@ -16,6 +16,8 @@ Send Options
      filesystems: ...
      send:
        encrypted: true
+       step_holds:
+         disable_incremental: false
      ...
 
 :ref:`Source<job-source>` and :ref:`push<job-push>` jobs have an optional ``send`` configuration section.
@@ -34,6 +36,17 @@ Filesystems matched by ``filesystems`` that are not encrypted are not sent and w
 
 If ``encryption=false``, zrepl expects that filesystems matching ``filesystems`` are not encrypted or have loaded encryption keys.
 
+
+``step_holds.disable_incremental`` option
+-----------------------------------------
+
+The ``step_holds.disable_incremental`` variable controls whether the creation of :ref:`step holds <step-holds-and-bookmarks>` should be disabled for incremental replication.
+The default value is ``false``.
+
+Disabling step holds has the disadvantage that steps :ref:`might not be resumable <step-holds-and-bookmarks>` if interrupted.
+However, the advantage and :issue:`reason for existence <288>` of this flag is that it allows the pruner to delete snapshots of interrupted replication steps
+which is useful if replication happens so rarely (or fails so frequently) that the amount of disk space exclusively referenced by the step's snapshots becomes intolerable.
+
 .. _job-recv-options:
 
 Recv Options

docs/configuration/transports.rst

@@ -52,8 +52,15 @@ Serve
         listen: ":8888"
         listen_freebind: true # optional, default false
         clients: {
-          "192.168.122.123" : "mysql01"
+          "192.168.122.123" :               "mysql01",
-          "192.168.122.123" : "mx01"
+          "192.168.122.42" :                "mx01",
+          "2001:0db8:85a3::8a2e:0370:7334": "gateway",
+
+          # CIDR masks require a '*' in the client identity string
+          # that is expanded to the client's IP address
+
+          "10.23.42.0/24":       "cluster-*"
+          "fde4:8dba:82e1::/64": "san-*"
         }
       ...
 

docs/installation.rst

@@ -1,138 +1,19 @@
-.. _binary releases: https://github.com/zrepl/zrepl/releases
 
-.. _installation:
+.. _installation_toc:
 
+************
 Installation
-============
+************
 
 .. TIP::
 
     Note: check out the :ref:`tutorial` if you want a first impression of zrepl.
 
-User Privileges
+.. toctree::
----------------
 
-It is possible to run zrepl as an unprivileged user in combination with
+    installation/user-privileges
-`ZFS delegation <https://www.freebsd.org/doc/handbook/zfs-zfs-allow.html>`_.
+    installation/packages
-Also, there is the possibility to run it in a jail on FreeBSD by delegating a dataset to the jail.
+    installation/apt-repos
-However, until we get around documenting those setups, you will have to run zrepl as root or experiment yourself :)
+    installation/compile-from-source
-
+    installation/freebsd-jail-with-iocage
-Packages
+    installation/what-next
---------
-
-zrepl source releases are signed & tagged by the author in the git repository.
-Your OS vendor may provide binary packages of zrepl through the package manager.
-Additionally, `binary releases`_ are provided on GitHub.
-The following list may be incomplete, feel free to submit a PR with an update:
-
-.. list-table::
-    :header-rows: 1
-
-    * - OS / Distro
-      - Install Command
-      - Link
-    * - FreeBSD
-      - ``pkg install zrepl``
-      - `<https://www.freshports.org/sysutils/zrepl/>`_
-    * - MacOS
-      - ``brew install zrepl``
-      - Available on `homebrew <https://brew.sh>`_
-    * - Arch Linux
-      - ``yay install zrepl``
-      - Available on `AUR <https://aur.archlinux.org/packages/zrepl>`_
-    * - Fedora
-      - ``dnf install zrepl``
-      - Available on `COPR <https://copr.fedorainfracloud.org/coprs/poettlerric/zrepl/>`_
-    * - CentOS/RHEL
-      - ``yum install zrepl``
-      - Available on `COPR <https://copr.fedorainfracloud.org/coprs/poettlerric/zrepl/>`_
-    * - Debian + Ubuntu
-      - ``apt install zrepl``
-      - APT repository config :ref:`see below <installation-apt-repos>`
-    * - OmniOS
-      - ``pkg install zrepl``
-      - Available since `r151030 <https://pkg.omniosce.org/r151030/extra/en/search.shtml?token=zrepl&action=Search>`_
-    * - Void Linux
-      - ``xbps-install zrepl``
-      - Available since `a88a2a4 <https://github.com/void-linux/void-packages/commit/a88a2a4d7bf56072dadf61ab56b8424e39155890>`_
-    * - Others
-      -
-      - Use `binary releases`_ or build from source.
-
-.. _installation-apt-repos:
-
-Debian / Ubuntu APT repositories
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-
-We maintain APT repositories for Debian, Ubuntu and derivatives.
-The fingerprint of the signing key is ``E101 418F D3D6 FBCB 9D65  A62D 7086 99FC 5F2E BF16``.
-It is available at `<https://zrepl.cschwarz.com/apt/apt-key.asc>`_ .
-Please open an issue `in the packaging repository <https://github.com/zrepl/debian-binary-packaging>`_ if you encounter any issues with the repository.
-
-The following snippet configure the repository for your Debian or Ubuntu release:
-
-::
-
-    apt update && apt install curl gnupg lsb-release; \
-    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"; \
-    (curl https://zrepl.cschwarz.com/apt/apt-key.asc | apt-key add -) && \
-    (echo "deb [arch=$ARCH] https://zrepl.cschwarz.com/apt/$CODENAME main" > /etc/apt/sources.list.d/zrepl.list) && \
-    apt update
-
-
-.. NOTE::
-
-   Until zrepl reaches 1.0, all APT repositories will be updated to the latest zrepl release immediately.
-   This includes breaking changes between zrepl versions.
-   Use ``apt-mark hold zrepl`` to prevent upgrades of zrepl.
-
-Compile From Source
-~~~~~~~~~~~~~~~~~~~
-
-Producing a release requires **Go 1.11** or newer and **Python 3** + **pip3** + ``docs/requirements.txt`` for the Sphinx documentation.
-A tutorial to install Go is available over at `golang.org <https://golang.org/doc/install>`_.
-Python and pip3 should probably be installed via your distro's package manager.
-
-Alternatively, you can use the Docker build process:
-it is used to produce the official zrepl `binary releases`_
-and serves as a reference for build dependencies and procedure:
-
-::
-
-    git clone https://github.com/zrepl/zrepl.git && \
-    cd zrepl && \
-    sudo docker build -t zrepl_build -f build.Dockerfile . && \
-    sudo docker run -it --rm \
-        -v "${PWD}:/src" \
-        --user "$(id -u):$(id -g)" \
-        zrepl_build make release
-
-Alternatively, you can install build dependencies on your local system and then build in your ``$GOPATH``:
-
-::
-
-    mkdir -p "${GOPATH}/src/github.com/zrepl/zrepl"
-    git clone https://github.com/zrepl/zrepl.git "${GOPATH}/src/github.com/zrepl/zrepl"
-    cd "${GOPATH}/src/github.com/zrepl/zrepl"
-    python3 -m venv3
-    source venv3/bin/activate
-    ./lazy.sh devsetup
-    make release
-
-The Python venv is used for the documentation build dependencies.
-If you just want to build the zrepl binary, leave it out and use `./lazy.sh godep` instead.
-Either way, all build results are located in the ``artifacts/`` directory.
-
-.. NOTE::
-
-    It is your job to install the appropriate binary in the zrepl users's ``$PATH``, e.g. ``/usr/local/bin/zrepl``.
-    Otherwise, the examples in the :ref:`tutorial` may need to be adjusted.
-
-What next?
-----------
-
-Read the :ref:`configuration chapter<configuration_toc>` and then continue with the :ref:`usage chapter<usage>`.
-
-**Reminder**: If you want a quick introduction, please read the :ref:`tutorial`.

docs/installation/apt-repos.rst

@@ -0,0 +1,29 @@
+
+.. _installation-apt-repos:
+
+Debian / Ubuntu APT repositories
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+We maintain APT repositories for Debian, Ubuntu and derivatives.
+The fingerprint of the signing key is ``E101 418F D3D6 FBCB 9D65  A62D 7086 99FC 5F2E BF16``.
+It is available at `<https://zrepl.cschwarz.com/apt/apt-key.asc>`_ .
+Please open an issue `in the packaging repository <https://github.com/zrepl/debian-binary-packaging>`_ if you encounter any issues with the repository.
+
+The following snippet configure the repository for your Debian or Ubuntu release:
+
+::
+
+    apt update && apt install curl gnupg lsb-release; \
+    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"; \
+    (curl https://zrepl.cschwarz.com/apt/apt-key.asc | apt-key add -) && \
+    (echo "deb [arch=$ARCH] https://zrepl.cschwarz.com/apt/$CODENAME main" > /etc/apt/sources.list.d/zrepl.list) && \
+    apt update
+
+
+.. NOTE::
+
+   Until zrepl reaches 1.0, all APT repositories will be updated to the latest zrepl release immediately.
+   This includes breaking changes between zrepl versions.
+   Use ``apt-mark hold zrepl`` to prevent upgrades of zrepl.

docs/installation/compile-from-source.rst

@@ -0,0 +1,45 @@
+.. _binary releases: https://github.com/zrepl/zrepl/releases
+
+.. _installation-compile-from-source:
+
+Compile From Source
+~~~~~~~~~~~~~~~~~~~
+
+Producing a release requires **Go 1.11** or newer and **Python 3** + **pip3** + ``docs/requirements.txt`` for the Sphinx documentation.
+A tutorial to install Go is available over at `golang.org <https://golang.org/doc/install>`_.
+Python and pip3 should probably be installed via your distro's package manager.
+
+Alternatively, you can use the Docker build process:
+it is used to produce the official zrepl `binary releases`_
+and serves as a reference for build dependencies and procedure:
+
+::
+
+    git clone https://github.com/zrepl/zrepl.git && \
+    cd zrepl && \
+    sudo docker build -t zrepl_build -f build.Dockerfile . && \
+    sudo docker run -it --rm \
+        -v "${PWD}:/src" \
+        --user "$(id -u):$(id -g)" \
+        zrepl_build make release
+
+Alternatively, you can install build dependencies on your local system and then build in your ``$GOPATH``:
+
+::
+
+    mkdir -p "${GOPATH}/src/github.com/zrepl/zrepl"
+    git clone https://github.com/zrepl/zrepl.git "${GOPATH}/src/github.com/zrepl/zrepl"
+    cd "${GOPATH}/src/github.com/zrepl/zrepl"
+    python3 -m venv3
+    source venv3/bin/activate
+    ./lazy.sh devsetup
+    make release
+
+The Python venv is used for the documentation build dependencies.
+If you just want to build the zrepl binary, leave it out and use `./lazy.sh godep` instead.
+Either way, all build results are located in the ``artifacts/`` directory.
+
+.. NOTE::
+
+    It is your job to install the appropriate binary in the zrepl users's ``$PATH``, e.g. ``/usr/local/bin/zrepl``.
+    Otherwise, the examples in the :ref:`tutorial` may need to be adjusted.

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

@@ -0,0 +1,140 @@
+.. include:: ../global.rst.inc
+
+.. _installation-freebsd-jail-with-iocage:
+
+FreeBSD Jail With iocage
+========================
+
+
+This tutorial shows how zrepl can be installed on FreeBSD, or FreeNAS in a jail using iocage.
+While this tutorial focuses on using iocage, much of the setup would be similar
+using a different jail manager.
+
+.. NOTE::
+
+   From a security perspective, just keep in mind that ``zfs send``/``recv`` was never designed with
+   jails in mind, an attacker could probably crash the receive-side kernel or worse induce stateful
+   damage to the receive-side pool if they were able to get access to the jail.
+
+   The jail doesn't provide security benefits, but only management ones.
+
+Requirements
+------------
+
+A dataset that will be delegated to the jail needs to be created if one does not already exist.
+For the tutorial ``tank/zrepl`` will be used.
+
+.. code-block:: bash
+
+   zfs create -o mountpoint=none tank/zrepl
+
+The only software requirements on the host system are ``iocage``, which can be installed
+from ports or packages.
+
+.. code-block:: bash
+
+   pkg install py37-iocage
+
+.. NOTE::
+
+   By default ``iocage`` will "activate" on first use which will set up some defaults such as
+   which pool will be used. To activate ``iocage`` manually the ``iocage activate`` command can be used.
+
+Jail Creation
+-------------
+
+There are two options for jail creation using FreeBSD.
+
+1. Manually set up the jail from scratch
+2. Create the jail using the ``zrepl`` plugin. On FreeNAS this is possible from the user interface using the community index.
+
+Manual Jail
+###########
+
+Create a jail, using the same release as the host, called ``zrepl`` that will be automatically started at boot.
+The jail will have ``tank/zrepl`` delegated into it.
+
+.. code-block:: bash
+
+   iocage create --release "$(freebsd-version -k | cut -d '-' -f '1,2')" --name zrepl \
+          boot=on nat=1 \
+          jail_zfs=on \
+          jail_zfs_dataset=zrepl \
+          jail_zfs_mountpoint='none'
+
+Enter the jail:
+
+.. code-block:: bash
+
+   iocage console zrepl
+
+Install ``zrepl``
+
+.. code-block:: bash
+
+   pkg update && pkg upgrade
+   pkg install zrepl
+
+Create the log file ``/var/log/zrepl.log``
+
+.. code-block:: bash
+
+   touch /var/log/zrepl.log && service newsyslog restart
+
+Tell syslogd to redirect facility local0 to the ``zrepl.log`` file:
+
+.. code-block:: bash
+
+   service syslogd reload
+
+Enable the zrepl daemon to start automatically at boot:
+
+.. code-block:: bash
+
+   sysrc zrepl_enable="YES"
+
+
+Plugin
+######
+
+When using the plugin, ``zrepl`` will be installed for you in a jail using the following ``iocage`` properties.
+
+* ``nat=1``
+* ``jail_zfs=on``
+* ``jail_zfs_mountpoint=none``
+
+Additionally the delegated dataset should be specified upon creation, and optionally start on boot can be set.
+This can also be done from the FreeNAS webui.
+
+.. code-block:: bash
+
+   fetch https://raw.githubusercontent.com/ix-plugin-hub/iocage-plugin-index/master/zrepl.json -o /tmp/zrepl.json
+   iocage fetch -P /tmp/zrepl.json --name zrepl jail_zfs_dataset=zrepl boot=on
+
+Configuration
+-------------
+
+Now ``zrepl`` can be configured.
+
+Enter the jail.
+
+.. code-block:: bash
+
+   iocage console zrepl
+
+Modify the ``/usr/local/etc/zrepl/zrepl.yml`` configuration file.
+
+.. TIP::
+
+    Note: check out the :ref:`tutorial` for examples of a ``sink`` job.
+
+Now ``zrepl`` can be started.
+
+.. code-block:: bash
+
+   service zrepl start
+
+Summary
+-------
+
+Congratulations, you have a working jail!

docs/installation/packages.rst

@@ -0,0 +1,50 @@
+.. _installation-packages:
+
+.. _binary releases: https://github.com/zrepl/zrepl/releases
+
+Packages
+--------
+
+zrepl source releases are signed & tagged by the author in the git repository.
+Your OS vendor may provide binary packages of zrepl through the package manager.
+Additionally, `binary releases`_ are provided on GitHub.
+The following list may be incomplete, feel free to submit a PR with an update:
+
+.. list-table::
+    :header-rows: 1
+
+    * - OS / Distro
+      - Install Command
+      - Link
+    * - FreeBSD
+      - ``pkg install zrepl``
+      - `<https://www.freshports.org/sysutils/zrepl/>`_
+      
+        :ref:`installation-freebsd-jail-with-iocage`
+    * - FreeNAS
+      -
+      - :ref:`installation-freebsd-jail-with-iocage`
+    * - MacOS
+      - ``brew install zrepl``
+      - Available on `homebrew <https://brew.sh>`_
+    * - Arch Linux
+      - ``yay install zrepl``
+      - Available on `AUR <https://aur.archlinux.org/packages/zrepl>`_
+    * - Fedora
+      - ``dnf install zrepl``
+      - Available on `COPR <https://copr.fedorainfracloud.org/coprs/poettlerric/zrepl/>`_
+    * - CentOS/RHEL
+      - ``yum install zrepl``
+      - Available on `COPR <https://copr.fedorainfracloud.org/coprs/poettlerric/zrepl/>`_
+    * - Debian + Ubuntu
+      - ``apt install zrepl``
+      - APT repository config :ref:`see below <installation-apt-repos>`
+    * - OmniOS
+      - ``pkg install zrepl``
+      - Available since `r151030 <https://pkg.omniosce.org/r151030/extra/en/search.shtml?token=zrepl&action=Search>`_
+    * - Void Linux
+      - ``xbps-install zrepl``
+      - Available since `a88a2a4 <https://github.com/void-linux/void-packages/commit/a88a2a4d7bf56072dadf61ab56b8424e39155890>`_
+    * - Others
+      -
+      - Use `binary releases`_ or build from source.

docs/installation/user-privileges.rst

@@ -0,0 +1,12 @@
+.. _installation-user-privileges:
+
+User Privileges
+---------------
+
+It is possible to run zrepl as an unprivileged user in combination with
+`ZFS delegation <https://www.freebsd.org/doc/handbook/zfs-zfs-allow.html>`_.
+Also, there is the possibility to run it in a jail on FreeBSD by delegating a dataset to the jail.
+
+.. TIP::
+
+    Note: check out the :ref:`installation-freebsd-jail-with-iocage` for FreeBSD jail setup instructions.

docs/installation/what-next.rst

@@ -0,0 +1,8 @@
+.. _installation-what-next:
+
+What next?
+----------
+
+Read the :ref:`configuration chapter<configuration_toc>` and then continue with the :ref:`usage chapter<usage>`.
+
+**Reminder**: If you want a quick introduction, please read the :ref:`tutorial`.

docs/supporters.rst

@@ -32,6 +32,9 @@ We would like to thank the following people and organizations for supporting zre
 
    <div class="fa fa-code" style="width: 1em;"></div>
 
+* |supporter-std| Henning Kessler
+* |supporter-std| `John Ramsden <https://github.com/johnramsden>`_
+* |supporter-std| `DrLuke <https://github.com/drluke>`_
 * |supporter-gold| `Mateusz Kwiatkowski (runhyve.app) <https://runhyve.app>`_
 * |supporter-std| `Gaelan D'costa <https://www.robot-disco.net>`_
 * |supporter-std| `Tenzin Lhakhang <https://confluence.tenzin.io>`_

docs/tutorial.rst

@@ -47,7 +47,7 @@ We can model this situation as two jobs:
 Install zrepl
 -------------
 
-Follow the :ref:`OS-specific installation instructions <installation>` and come back here.
+Follow the :ref:`OS-specific installation instructions <installation_toc>` and come back here.
 
 Generate TLS Certificates
 -------------------------

endpoint/endpoint.go

@@ -7,8 +7,8 @@ import (
 	"fmt"
 	"io"
 	"path"
-	"sync"
 
+	"github.com/kr/pretty"
 	"github.com/pkg/errors"
 	"github.com/zrepl/zrepl/daemon/logging/trace"
 
@@ -23,6 +23,7 @@ import (
 type SenderConfig struct {
 	FSF                         zfs.DatasetFilter
 	Encrypt                     *zfs.NilBool
+	DisableIncrementalStepHolds bool
 	JobID                       JobID
 }
 
@@ -41,6 +42,7 @@ func (c *SenderConfig) Validate() error {
 type Sender struct {
 	FSFilter                    zfs.DatasetFilter
 	encrypt                     *zfs.NilBool
+	disableIncrementalStepHolds bool
 	jobId                       JobID
 }
 
@@ -51,6 +53,7 @@ func NewSender(conf SenderConfig) *Sender {
 	return &Sender{
 		FSFilter:                    conf.FSF,
 		encrypt:                     conf.Encrypt,
+		disableIncrementalStepHolds: conf.DisableIncrementalStepHolds,
 		jobId:                       conf.JobID,
 	}
 }
@@ -117,119 +120,8 @@ func (s *Sender) ListFilesystemVersions(ctx context.Context, r *pdu.ListFilesyst
 
 }
 
-func (p *Sender) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.HintMostRecentCommonAncestorReq) (*pdu.HintMostRecentCommonAncestorRes, error) {
+var maxConcurrentZFSSend = envconst.Int64("ZREPL_ENDPOINT_MAX_CONCURRENT_SEND", 10)
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+var maxConcurrentZFSSendSemaphore = semaphore.New(maxConcurrentZFSSend)
-
-	fsp, err := p.filterCheckFS(r.GetFilesystem())
-	if err != nil {
-		return nil, err
-	}
-	fs := fsp.ToString()
-
-	log := getLogger(ctx).WithField("fs", fs).WithField("hinted_most_recent", fmt.Sprintf("%#v", r.GetSenderVersion()))
-
-	log.WithField("full_hint", r).Debug("full hint")
-
-	if r.GetSenderVersion() == nil {
-		// no common ancestor found, likely due to failed prior replication attempt
-		// => release stale step holds to prevent them from accumulating
-		//    (they can accumulate on initial replication because each inital replication step might hold a different `to`)
-		// => replication cursors cannot accumulate because we always _move_ the replication cursor
-		log.Debug("releasing all step holds on the filesystem")
-		TryReleaseStepStaleFS(ctx, fs, p.jobId)
-		return &pdu.HintMostRecentCommonAncestorRes{}, nil
-	}
-
-	// we were hinted a specific common ancestor
-
-	mostRecentVersion, err := sendArgsFromPDUAndValidateExistsAndGetVersion(ctx, fs, r.GetSenderVersion())
-	if err != nil {
-		msg := "HintMostRecentCommonAncestor rpc with nonexistent most recent version"
-		log.Warn(msg)
-		return nil, errors.Wrap(err, msg)
-	}
-
-	// move replication cursor to this position
-	destroyedCursors, err := MoveReplicationCursor(ctx, fs, mostRecentVersion, p.jobId)
-	if err == zfs.ErrBookmarkCloningNotSupported {
-		log.Debug("not creating replication cursor from bookmark because ZFS does not support it")
-		// fallthrough
-	} else if err != nil {
-		return nil, errors.Wrap(err, "cannot set replication cursor to hinted version")
-	}
-
-	// take care of stale step holds
-	log.WithField("step-holds-cleanup-mode", senderHintMostRecentCommonAncestorStepCleanupMode).
-		Debug("taking care of possibly stale step holds")
-	doStepCleanup := false
-	var stepCleanupSince *CreateTXGRangeBound
-	switch senderHintMostRecentCommonAncestorStepCleanupMode {
-	case StepCleanupNoCleanup:
-		doStepCleanup = false
-	case StepCleanupRangeSinceUnbounded:
-		doStepCleanup = true
-		stepCleanupSince = nil
-	case StepCleanupRangeSinceReplicationCursor:
-		doStepCleanup = true
-		// Use the destroyed replication cursors as indicator how far the previous replication got.
-		// To be precise: We limit the amount of visisted snapshots to exactly those snapshots
-		// created since the last successful replication cursor movement (i.e. last successful replication step)
-		//
-		// If we crash now, we'll leak the step we are about to release, but the performance gain
-		// of limiting the amount of snapshots we visit makes up for that.
-		// Users have the `zrepl holds release-stale` command to cleanup leaked step holds.
-		for _, destroyed := range destroyedCursors {
-			if stepCleanupSince == nil {
-				stepCleanupSince = &CreateTXGRangeBound{
-					CreateTXG: destroyed.GetCreateTXG(),
-					Inclusive: &zfs.NilBool{B: true},
-				}
-			} else if destroyed.GetCreateTXG() < stepCleanupSince.CreateTXG {
-				stepCleanupSince.CreateTXG = destroyed.GetCreateTXG()
-			}
-		}
-	default:
-		panic(senderHintMostRecentCommonAncestorStepCleanupMode)
-	}
-	if !doStepCleanup {
-		log.Info("skipping cleanup of prior invocations' step holds due to environment variable setting")
-	} else {
-		if err := ReleaseStepCummulativeInclusive(ctx, fs, stepCleanupSince, mostRecentVersion, p.jobId); err != nil {
-			return nil, errors.Wrap(err, "cannot cleanup prior invocation's step holds and bookmarks")
-		} else {
-			log.Info("step hold cleanup done")
-		}
-	}
-
-	return &pdu.HintMostRecentCommonAncestorRes{}, nil
-}
-
-type HintMostRecentCommonAncestorStepCleanupMode struct{ string }
-
-var (
-	StepCleanupRangeSinceReplicationCursor = HintMostRecentCommonAncestorStepCleanupMode{"range-since-replication-cursor"}
-	StepCleanupRangeSinceUnbounded         = HintMostRecentCommonAncestorStepCleanupMode{"range-since-unbounded"}
-	StepCleanupNoCleanup                   = HintMostRecentCommonAncestorStepCleanupMode{"no-cleanup"}
-)
-
-func (m HintMostRecentCommonAncestorStepCleanupMode) String() string { return string(m.string) }
-func (m *HintMostRecentCommonAncestorStepCleanupMode) Set(s string) error {
-	switch s {
-	case StepCleanupRangeSinceReplicationCursor.String():
-		*m = StepCleanupRangeSinceReplicationCursor
-	case StepCleanupRangeSinceUnbounded.String():
-		*m = StepCleanupRangeSinceUnbounded
-	case StepCleanupNoCleanup.String():
-		*m = StepCleanupNoCleanup
-	default:
-		return fmt.Errorf("unknown step cleanup mode %q", s)
-	}
-	return nil
-}
-
-var senderHintMostRecentCommonAncestorStepCleanupMode = *envconst.Var("ZREPL_ENDPOINT_SENDER_HINT_MOST_RECENT_STEP_HOLD_CLEANUP_MODE", &StepCleanupRangeSinceReplicationCursor).(*HintMostRecentCommonAncestorStepCleanupMode)
-
-var maxConcurrentZFSSendSemaphore = semaphore.New(envconst.Int64("ZREPL_ENDPOINT_MAX_CONCURRENT_SEND", 10))
 
 func uncheckedSendArgsFromPDU(fsv *pdu.FilesystemVersion) *zfs.ZFSSendArgVersion {
 	if fsv == nil {
@@ -319,10 +211,11 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
 		return res, nil, nil
 	}
 
-	// update replication cursor
+	// create a replication cursor for `From` (usually an idempotent no-op because SendCompleted already created it before)
+	var fromReplicationCursor Abstraction
 	if sendArgs.From != nil {
 		// For all but the first replication, this should always be a no-op because SendCompleted already moved the cursor
-		_, err = MoveReplicationCursor(ctx, sendArgs.FS, sendArgs.FromVersion, s.jobId)
+		fromReplicationCursor, err = CreateReplicationCursor(ctx, sendArgs.FS, *sendArgs.FromVersion, s.jobId) // no shadow
 		if err == zfs.ErrBookmarkCloningNotSupported {
 			getLogger(ctx).Debug("not creating replication cursor from bookmark because ZFS does not support it")
 			// fallthrough
@@ -331,9 +224,12 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
 		}
 	}
 
+	takeStepHolds := sendArgs.FromVersion == nil || !s.disableIncrementalStepHolds
+
+	var fromHold, toHold Abstraction
 	// make sure `From` doesn't go away in order to make this step resumable
-	if sendArgs.From != nil {
+	if sendArgs.From != nil && takeStepHolds {
-		_, err := HoldStep(ctx, sendArgs.FS, *sendArgs.FromVersion, s.jobId)
+		fromHold, err = HoldStep(ctx, sendArgs.FS, *sendArgs.FromVersion, s.jobId) // no shadow
 		if err == zfs.ErrBookmarkCloningNotSupported {
 			getLogger(ctx).Debug("not creating step bookmark because ZFS does not support it")
 			// fallthrough
@@ -341,18 +237,72 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
 			return nil, nil, errors.Wrapf(err, "cannot hold `from` version %q before starting send", *sendArgs.FromVersion)
 		}
 	}
+	if takeStepHolds {
 		// make sure `To` doesn't go away in order to make this step resumable
-	_, err = HoldStep(ctx, sendArgs.FS, sendArgs.ToVersion, s.jobId)
+		toHold, err = HoldStep(ctx, sendArgs.FS, sendArgs.ToVersion, s.jobId)
 		if err != nil {
 			return nil, nil, errors.Wrapf(err, "cannot hold `to` version %q before starting send", sendArgs.ToVersion)
 		}
+	}
 
-	// step holds & replication cursor released / moved forward in s.SendCompleted => s.moveCursorAndReleaseSendHolds
+	// cleanup the mess that _this function_ might have created in prior failed attempts:
+	//
+	// In summary, we delete every endpoint ZFS abstraction created on this filesystem for this job id,
+	// except for the ones we just created above.
+	//
+	// This is the most robust approach to avoid leaking (= forgetting to clean up) endpoint ZFS abstractions,
+	// all under the assumption that there will only ever be one send for a (jobId,fs) combination at any given time.
+	//
+	// Note that the SendCompleted rpc can't be relied upon for this purpose:
+	// - it might be lost due to network errors,
+	// - or never be sent by a potentially malicious or buggy client,
+	// - or never be send because the replication step failed at some point
+	//   (potentially leaving a resumable state on the receiver, which is the case where we really do not want to blow away the step holds too soon.)
+	//
+	// Note further that a resuming send, due to the idempotent nature of func CreateReplicationCursor and HoldStep,
+	// will never lose its step holds because we just (idempotently re-)created them above, before attempting the cleanup.
+	liveAbs := []Abstraction{fromHold, toHold, fromReplicationCursor}
+	func() {
+		ctx, endSpan := trace.WithSpan(ctx, "cleanup-stale-abstractions")
+		defer endSpan()
+
+		keep := func(a Abstraction) (keep bool) {
+			keep = false
+			for _, k := range liveAbs {
+				keep = keep || AbstractionEquals(a, k)
+			}
+			return keep
+		}
+		check := func(obsoleteAbs []Abstraction) {
+			// last line of defense: check that we don't destroy the incremental `from` and `to`
+			// if we did that, we might be about to blow away the last common filesystem version between sender and receiver
+			mustLiveVersions := []zfs.FilesystemVersion{sendArgs.ToVersion}
+			if sendArgs.FromVersion != nil {
+				mustLiveVersions = append(mustLiveVersions, *sendArgs.FromVersion)
+			}
+			for _, staleVersion := range obsoleteAbs {
+				for _, mustLiveVersion := range mustLiveVersions {
+					if zfs.FilesystemVersionEqualIdentity(mustLiveVersion, staleVersion.GetFilesystemVersion()) {
+						panic(fmt.Sprintf("impl error: %q would be destroyed because it is considered stale but it is part of of sendArgs=%s", mustLiveVersion.String(), pretty.Sprint(sendArgs)))
+					}
+				}
+			}
+		}
+		sendAbstractionsCacheSingleton.TryBatchDestroy(ctx, s.jobId, sendArgs.FS, keep, check)
+	}()
+	// now add the newly created abstractions to the cleaned-up cache
+	for _, a := range liveAbs {
+		if a != nil {
+			sendAbstractionsCacheSingleton.Put(a)
+		}
+	}
 
 	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")
 	}
+
 	return res, sendStream, nil
 }
 
@@ -389,8 +339,7 @@ func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*p
 		return log
 	}
 
-	log(ctx).Debug("move replication cursor to most recent common version")
+	toReplicationCursor, err := CreateReplicationCursor(ctx, fs, to, p.jobId)
-	destroyedCursors, err := MoveReplicationCursor(ctx, fs, to, p.jobId)
 	if err != nil {
 		if err == zfs.ErrBookmarkCloningNotSupported {
 			log(ctx).Debug("not setting replication cursor, bookmark cloning not supported")
@@ -398,63 +347,20 @@ func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*p
 			msg := "cannot move replication cursor, keeping hold on `to` until successful"
 			log(ctx).WithError(err).Error(msg)
 			err = errors.Wrap(err, msg)
-			// it is correct to not release the hold if we can't move the cursor!
+			// it is correct to not destroy from and to step holds if we can't move the cursor!
 			return &pdu.SendCompletedRes{}, err
 		}
 	} else {
-		log(ctx).Info("successfully moved replication cursor")
+		log(ctx).WithField("to_cursor", toReplicationCursor.String()).Info("successfully created `to` replication cursor")
 	}
 
-	// kick off releasing of step holds / bookmarks
+	keep := func(a Abstraction) bool {
-	// if we fail to release them, don't bother the caller:
+		return AbstractionEquals(a, toReplicationCursor)
-	// they are merely an implementation detail on the sender for better resumability
-	var wg sync.WaitGroup
-	wg.Add(2)
-	go func() {
-		defer wg.Done()
-		ctx, endTask := trace.WithTask(ctx, "release-step-hold-to")
-		defer endTask()
-
-		log(ctx).Debug("release step-hold of or step-bookmark on `to`")
-		err = ReleaseStep(ctx, fs, to, p.jobId)
-		if err != nil {
-			log(ctx).WithError(err).Error("cannot release step-holds on or destroy step-bookmark of `to`")
-		} else {
-			log(ctx).Info("successfully released step-holds on or destroyed step-bookmark of `to`")
 	}
-
+	sendAbstractionsCacheSingleton.TryBatchDestroy(ctx, p.jobId, fs, keep, nil)
-	}()
-	go func() {
-		defer wg.Done()
-		ctx, endTask := trace.WithTask(ctx, "release-step-hold-from")
-		defer endTask()
-
-		if from == nil {
-			return
-		}
-		log(ctx).Debug("release step-hold of or step-bookmark on `from`")
-		err := ReleaseStep(ctx, fs, *from, p.jobId)
-		if err != nil {
-			if dne, ok := err.(*zfs.DatasetDoesNotExist); ok {
-				// If bookmark cloning is not supported, `from` might be the old replication cursor
-				// and thus have already been destroyed by MoveReplicationCursor above
-				// In that case, nonexistence of `from` is not an error, otherwise it is.
-				for _, c := range destroyedCursors {
-					if c.GetFullPath() == dne.Path {
-						log(ctx).Info("`from` was a replication cursor and has already been destroyed")
-						return
-					}
-				}
-				// fallthrough
-			}
-			log(ctx).WithError(err).Error("cannot release step-holds on or destroy step-bookmark of `from`")
-		} else {
-			log(ctx).Info("successfully released step-holds on or destroyed step-bookmark of `from`")
-		}
-	}()
-	wg.Wait()
 
 	return &pdu.SendCompletedRes{}, nil
+
 }
 
 func (p *Sender) DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshotsReq) (*pdu.DestroySnapshotsRes, error) {
@@ -858,9 +764,9 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.
 	}
 	defer guard.Release()
 
-	log.Info("peeking 1M ahead")
 	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")
 	if _, err := io.Copy(&peek, io.LimitReader(receive, MaxPeek)); err != nil {
 		log.WithError(err).Error("cannot read peek-buffer from send stream")
 	}
@@ -974,16 +880,6 @@ func (s *Receiver) DestroySnapshots(ctx context.Context, req *pdu.DestroySnapsho
 	return doDestroySnapshots(ctx, lp, req.Snapshots)
 }
 
-func (p *Receiver) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.HintMostRecentCommonAncestorReq) (*pdu.HintMostRecentCommonAncestorRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-	// we don't move last-received-hold as part of this hint
-	// because that wouldn't give us any benefit wrt resumability.
-	//
-	// Other reason: the replication logic that issues this RPC would require refactoring
-	// to include the receiver's FilesystemVersion in the request)
-	return &pdu.HintMostRecentCommonAncestorRes{}, nil
-}
-
 func (p *Receiver) SendCompleted(ctx context.Context, _ *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
 	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 

endpoint/endpoint_metrics.go

@@ -0,0 +1,7 @@
+package endpoint
+
+import "github.com/prometheus/client_golang/prometheus"
+
+func RegisterMetrics(r prometheus.Registerer) {
+	r.MustRegister(sendAbstractionsCacheMetrics.count)
+}

endpoint/endpoint_send_abstractions_cache.go

@@ -0,0 +1,214 @@
+package endpoint
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/zrepl/zrepl/daemon/logging/trace"
+	"github.com/zrepl/zrepl/util/chainlock"
+)
+
+var sendAbstractionsCacheMetrics struct {
+	count prometheus.Gauge
+}
+
+func init() {
+	sendAbstractionsCacheMetrics.count = prometheus.NewGauge(prometheus.GaugeOpts{
+		Namespace: "zrepl",
+		Subsystem: "endpoint",
+		Name:      "send_abstractions_cache_entry_count",
+		Help:      "number of send abstractions tracked in the sendAbstractionsCache data structure",
+	})
+}
+
+var sendAbstractionsCacheSingleton = newSendAbstractionsCache()
+
+func SendAbstractionsCacheInvalidate(fs string) {
+	sendAbstractionsCacheSingleton.InvalidateFSCache(fs)
+}
+
+type sendAbstractionsCacheDidLoadFSState int
+
+const (
+	sendAbstractionsCacheDidLoadFSStateNo sendAbstractionsCacheDidLoadFSState = iota // 0-value has meaning
+	sendAbstractionsCacheDidLoadFSStateInProgress
+	sendAbstractionsCacheDidLoadFSStateDone
+)
+
+type sendAbstractionsCache struct {
+	mtx              chainlock.L
+	abstractions     []Abstraction
+	didLoadFS        map[string]sendAbstractionsCacheDidLoadFSState
+	didLoadFSChanged *sync.Cond
+}
+
+func newSendAbstractionsCache() *sendAbstractionsCache {
+	c := &sendAbstractionsCache{
+		didLoadFS: make(map[string]sendAbstractionsCacheDidLoadFSState),
+	}
+	c.didLoadFSChanged = c.mtx.NewCond()
+	return c
+}
+
+func (s *sendAbstractionsCache) Put(a Abstraction) {
+	defer s.mtx.Lock().Unlock()
+
+	var zeroJobId JobID
+	if a.GetJobID() == nil {
+		panic("abstraction must not have nil job id")
+	} else if *a.GetJobID() == zeroJobId {
+		panic(fmt.Sprintf("abstraction must not have zero-value job id: %s", a))
+	}
+
+	s.abstractions = append(s.abstractions, a)
+	sendAbstractionsCacheMetrics.count.Set(float64(len(s.abstractions)))
+}
+
+func (s *sendAbstractionsCache) InvalidateFSCache(fs string) {
+	// FIXME: O(n)
+	newAbs := make([]Abstraction, 0, len(s.abstractions))
+	for _, a := range s.abstractions {
+		if a.GetFS() != fs {
+			newAbs = append(newAbs, a)
+		}
+	}
+	s.abstractions = newAbs
+	sendAbstractionsCacheMetrics.count.Set(float64(len(s.abstractions)))
+
+	s.didLoadFS[fs] = sendAbstractionsCacheDidLoadFSStateNo
+	s.didLoadFSChanged.Broadcast()
+
+}
+
+// - logs errors in getting on-disk abstractions
+// - only fetches on-disk abstractions once, but every time from the in-memory store
+//
+// That means that for precise results, all abstractions created by the endpoint must be .Put into this cache.
+func (s *sendAbstractionsCache) GetAndDeleteByJobIDAndFS(ctx context.Context, jobID JobID, fs string) (ret []Abstraction) {
+	defer s.mtx.Lock().Unlock()
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+	var zeroJobId JobID
+	if jobID == zeroJobId {
+		panic("must not pass zero-value job id")
+	}
+	if fs == "" {
+		panic("must not pass zero-value fs")
+	}
+
+	s.tryLoadOnDiskSendAbstractions(ctx, fs)
+
+	// FIXME O(n)
+	var remaining []Abstraction
+	for _, a := range s.abstractions {
+		aJobId := *a.GetJobID()
+		aFS := a.GetFS()
+		if aJobId == jobID && aFS == fs {
+			ret = append(ret, a)
+		} else {
+			remaining = append(remaining, a)
+		}
+	}
+	s.abstractions = remaining
+	sendAbstractionsCacheMetrics.count.Set(float64(len(s.abstractions)))
+
+	return ret
+}
+
+// caller must hold s.mtx
+func (s *sendAbstractionsCache) tryLoadOnDiskSendAbstractions(ctx context.Context, fs string) {
+	for s.didLoadFS[fs] != sendAbstractionsCacheDidLoadFSStateDone {
+		if s.didLoadFS[fs] == sendAbstractionsCacheDidLoadFSStateInProgress {
+			s.didLoadFSChanged.Wait()
+			continue
+		}
+		if s.didLoadFS[fs] != sendAbstractionsCacheDidLoadFSStateNo {
+			panic(fmt.Sprintf("unreachable: %v", s.didLoadFS[fs]))
+		}
+
+		s.didLoadFS[fs] = sendAbstractionsCacheDidLoadFSStateInProgress
+		defer s.didLoadFSChanged.Broadcast()
+
+		var onDiskAbs []Abstraction
+		var err error
+		s.mtx.DropWhile(func() {
+			onDiskAbs, err = s.tryLoadOnDiskSendAbstractionsImpl(ctx, fs) // no shadow
+		})
+
+		if err != nil {
+			s.didLoadFS[fs] = sendAbstractionsCacheDidLoadFSStateNo
+			getLogger(ctx).WithField("fs", fs).WithError(err).Error("cannot list send step abstractions for filesystem")
+		} else {
+			s.didLoadFS[fs] = sendAbstractionsCacheDidLoadFSStateDone
+			s.abstractions = append(s.abstractions, onDiskAbs...)
+			getLogger(ctx).WithField("fs", fs).WithField("abstractions", onDiskAbs).Debug("loaded step abstractions for filesystem")
+		}
+		return
+	}
+}
+
+// caller should _not hold s.mtx
+func (s *sendAbstractionsCache) tryLoadOnDiskSendAbstractionsImpl(ctx context.Context, fs string) ([]Abstraction, error) {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+
+	q := ListZFSHoldsAndBookmarksQuery{
+		FS: ListZFSHoldsAndBookmarksQueryFilesystemFilter{
+			FS: &fs,
+		},
+		JobID: nil,
+		What: AbstractionTypeSet{
+			AbstractionStepHold:                    true,
+			AbstractionStepBookmark:                true,
+			AbstractionReplicationCursorBookmarkV2: true,
+		},
+		Concurrency: 1,
+	}
+	abs, absErrs, err := ListAbstractions(ctx, q)
+	if err != nil {
+		return nil, err
+	}
+	// safe to ignore absErrs here, this is best-effort cleanup
+	if len(absErrs) > 0 {
+		return nil, ListAbstractionsErrors(absErrs)
+	}
+	return abs, nil
+}
+
+func (s *sendAbstractionsCache) TryBatchDestroy(ctx context.Context, jobId JobID, fs string, keep func(a Abstraction) bool, check func(willDestroy []Abstraction)) {
+	// no s.mtx, we only use the public interface in this function
+
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+
+	allSendStepAbstractions := s.GetAndDeleteByJobIDAndFS(ctx, jobId, fs)
+
+	var obsoleteAbs []Abstraction
+	for _, a := range allSendStepAbstractions {
+		if !keep(a) {
+			obsoleteAbs = append(obsoleteAbs, a)
+		}
+	}
+
+	if check != nil {
+		check(obsoleteAbs)
+	}
+
+	hadErr := false
+	for res := range BatchDestroy(ctx, obsoleteAbs) {
+		if res.DestroyErr != nil {
+			hadErr = true
+			getLogger(ctx).
+				WithField("abstraction", res.Abstraction).
+				WithError(res.DestroyErr).
+				Error("cannot destroy stale send step abstraction")
+		} else {
+			getLogger(ctx).
+				WithField("abstraction", res.Abstraction).
+				Info("destroyed stale send step abstraction")
+		}
+	}
+	if hadErr {
+		s.InvalidateFSCache(fs)
+	}
+
+}

endpoint/endpoint_zfs_abstraction.go

@@ -54,6 +54,30 @@ type Abstraction interface {
 	json.Marshaler
 }
 
+func AbstractionEquals(a, b Abstraction) bool {
+	if (a != nil) != (b != nil) {
+		return false
+	}
+	if a == nil && b == nil {
+		return true
+	}
+	var aJobId, bJobId JobID
+	if aJid := a.GetJobID(); aJid != nil {
+		aJobId = *aJid
+	}
+	if bJid := b.GetJobID(); bJid != nil {
+		bJobId = *bJid
+	}
+	return a.GetType() == b.GetType() &&
+		a.GetFS() == b.GetFS() &&
+		a.GetName() == b.GetName() &&
+		a.GetFullPath() == b.GetFullPath() &&
+		aJobId == bJobId &&
+		a.GetCreateTXG() == b.GetCreateTXG() &&
+		zfs.FilesystemVersionEqualIdentity(a.GetFilesystemVersion(), b.GetFilesystemVersion()) &&
+		a.String() == b.String()
+}
+
 func (t AbstractionType) Validate() error {
 	switch t {
 	case AbstractionStepBookmark:
@@ -590,26 +614,26 @@ func listAbstractionsImplFS(ctx context.Context, fs string, query *ListZFSHoldsA
 			panic("implementation error: extractors misconfigured for " + at)
 		}
 		for _, v := range fsvs {
-			var a Abstraction
 			if v.Type == zfs.Bookmark && bmE != nil {
-				a = bmE(fsp, v)
+				if a := bmE(fsp, v); a != nil {
+					emitCandidate(a)
 				}
-			if v.Type == zfs.Snapshot && holdE != nil && query.CreateTXG.Contains(v.GetCreateTXG()) && (!v.UserRefs.Valid || v.UserRefs.Value > 0) {
+			}
+			if v.Type == zfs.Snapshot && holdE != nil && query.CreateTXG.Contains(v.GetCreateTXG()) && (!v.UserRefs.Valid || v.UserRefs.Value > 0) { // FIXME review v.UserRefsValid
 				holds, err := zfs.ZFSHolds(ctx, fsp.ToString(), v.Name)
 				if err != nil {
 					errCb(err, v.ToAbsPath(fsp), "get hold on snap")
 					continue
 				}
 				for _, tag := range holds {
-					a = holdE(fsp, v, tag)
+					if a := holdE(fsp, v, tag); a != nil {
-				}
-			}
-			if a != nil {
 						emitCandidate(a)
 					}
 				}
 			}
 		}
+	}
+}
 
 type BatchDestroyResult struct {
 	Abstraction

endpoint/endpoint_zfs_abstraction_cursor_and_last_received_hold.go

@@ -118,31 +118,32 @@ func GetReplicationCursors(ctx context.Context, dp *zfs.DatasetPath, jobID JobID
 	return candidates, nil
 }
 
-type ReplicationCursorTarget interface {
+// idempotently create a replication cursor targeting `target`
-	IsSnapshot() bool
-	GetGuid() uint64
-	GetCreateTXG() uint64
-	ToSendArgVersion() zfs.ZFSSendArgVersion
-}
-
-// `target` is validated before replication cursor is set. if validation fails, the cursor is not moved.
 //
 // returns ErrBookmarkCloningNotSupported if version is a bookmark and bookmarking bookmarks is not supported by ZFS
-func MoveReplicationCursor(ctx context.Context, fs string, target ReplicationCursorTarget, jobID JobID) (destroyedCursors []Abstraction, err error) {
+func CreateReplicationCursor(ctx context.Context, fs string, target zfs.FilesystemVersion, jobID JobID) (a Abstraction, err error) {
-
-	if !target.IsSnapshot() {
-		return nil, zfs.ErrBookmarkCloningNotSupported
-	}
 
 	bookmarkname, err := ReplicationCursorBookmarkName(fs, target.GetGuid(), jobID)
 	if err != nil {
 		return nil, errors.Wrap(err, "determine replication cursor name")
 	}
 
-	// idempotently create bookmark (guid is encoded in it, hence we'll most likely add a new one
+	if target.IsBookmark() && target.GetName() == bookmarkname {
-	// cleanup the old one afterwards
+		return &bookmarkBasedAbstraction{
+			Type:              AbstractionReplicationCursorBookmarkV2,
+			FS:                fs,
+			FilesystemVersion: target,
+			JobID:             jobID,
+		}, nil
+	}
 
-	err = zfs.ZFSBookmark(ctx, fs, target.ToSendArgVersion(), bookmarkname)
+	if !target.IsSnapshot() {
+		return nil, zfs.ErrBookmarkCloningNotSupported
+	}
+
+	// idempotently create bookmark (guid is encoded in it)
+
+	cursorBookmark, err := zfs.ZFSBookmark(ctx, fs, target, bookmarkname)
 	if err != nil {
 		if err == zfs.ErrBookmarkCloningNotSupported {
 			return nil, err // TODO go1.13 use wrapping
@@ -150,65 +151,17 @@ func MoveReplicationCursor(ctx context.Context, fs string, target ReplicationCur
 		return nil, errors.Wrapf(err, "cannot create bookmark")
 	}
 
-	destroyedCursors, err = DestroyObsoleteReplicationCursors(ctx, fs, target, jobID)
+	return &bookmarkBasedAbstraction{
-	if err != nil {
+		Type:              AbstractionReplicationCursorBookmarkV2,
-		return nil, errors.Wrap(err, "destroy obsolete replication cursors")
+		FS:                fs,
+		FilesystemVersion: cursorBookmark,
+		JobID:             jobID,
+	}, nil
 }
 
-	return destroyedCursors, nil
+const (
-}
+	ReplicationCursorBookmarkNamePrefix = "zrepl_last_received_J_"
-
+)
-type ReplicationCursor interface {
-	GetCreateTXG() uint64
-}
-
-func DestroyObsoleteReplicationCursors(ctx context.Context, fs string, current ReplicationCursor, jobID JobID) (_ []Abstraction, err error) {
-
-	q := ListZFSHoldsAndBookmarksQuery{
-		FS: ListZFSHoldsAndBookmarksQueryFilesystemFilter{
-			FS: &fs,
-		},
-		What: AbstractionTypeSet{
-			AbstractionReplicationCursorBookmarkV2: true,
-		},
-		JobID: &jobID,
-		CreateTXG: CreateTXGRange{
-			Since: nil,
-			Until: &CreateTXGRangeBound{
-				CreateTXG: current.GetCreateTXG(),
-				Inclusive: &zfs.NilBool{B: false},
-			},
-		},
-		Concurrency: 1,
-	}
-	abs, absErr, err := ListAbstractions(ctx, q)
-	if err != nil {
-		return nil, errors.Wrap(err, "list abstractions")
-	}
-	if len(absErr) > 0 {
-		return nil, errors.Wrap(ListAbstractionsErrors(absErr), "list abstractions")
-	}
-
-	var destroyed []Abstraction
-	var errs []error
-	for res := range BatchDestroy(ctx, abs) {
-		log := getLogger(ctx).
-			WithField("replication_cursor_bookmark", res.Abstraction)
-		if res.DestroyErr != nil {
-			errs = append(errs, res.DestroyErr)
-			log.WithError(err).
-				Error("cannot destroy obsolete replication cursor bookmark")
-		} else {
-			destroyed = append(destroyed, res.Abstraction)
-			log.Info("destroyed obsolete replication cursor bookmark")
-		}
-	}
-	if len(errs) == 0 {
-		return destroyed, nil
-	} else {
-		return destroyed, errorarray.Wrap(errs, "destroy obsolete replication cursor")
-	}
-}
 
 var lastReceivedHoldTagRE = regexp.MustCompile("^zrepl_last_received_J_(.+)$")
 
@@ -230,22 +183,22 @@ func LastReceivedHoldTag(jobID JobID) (string, error) {
 }
 
 func lastReceivedHoldImpl(jobid string) (string, error) {
-	tag := fmt.Sprintf("zrepl_last_received_J_%s", jobid)
+	tag := fmt.Sprintf("%s%s", ReplicationCursorBookmarkNamePrefix, jobid)
 	if err := zfs.ValidHoldTag(tag); err != nil {
 		return "", err
 	}
 	return tag, nil
 }
 
-func MoveLastReceivedHold(ctx context.Context, fs string, to zfs.FilesystemVersion, jobID JobID) error {
+func CreateLastReceivedHold(ctx context.Context, fs string, to zfs.FilesystemVersion, jobID JobID) (Abstraction, error) {
 
 	if !to.IsSnapshot() {
-		return errors.Errorf("last-received-hold: target must be a snapshot: %s", to.FullPath(fs))
+		return nil, errors.Errorf("last-received-hold: target must be a snapshot: %s", to.FullPath(fs))
 	}
 
 	tag, err := LastReceivedHoldTag(jobID)
 	if err != nil {
-		return errors.Wrap(err, "last-received-hold: hold tag")
+		return nil, errors.Wrap(err, "last-received-hold: hold tag")
 	}
 
 	// we never want to be without a hold
@@ -253,7 +206,23 @@ func MoveLastReceivedHold(ctx context.Context, fs string, to zfs.FilesystemVersi
 
 	err = zfs.ZFSHold(ctx, fs, to, tag)
 	if err != nil {
-		return errors.Wrap(err, "last-received-hold: hold newly received")
+		return nil, errors.Wrap(err, "last-received-hold: hold newly received")
+	}
+
+	return &holdBasedAbstraction{
+		Type:              AbstractionLastReceivedHold,
+		FS:                fs,
+		FilesystemVersion: to,
+		JobID:             jobID,
+		Tag:               tag,
+	}, nil
+}
+
+func MoveLastReceivedHold(ctx context.Context, fs string, to zfs.FilesystemVersion, jobID JobID) error {
+
+	_, err := CreateLastReceivedHold(ctx, fs, to, jobID)
+	if err != nil {
+		return err
 	}
 
 	q := ListZFSHoldsAndBookmarksQuery{

endpoint/endpoint_zfs_abstraction_step.go

@@ -7,7 +7,6 @@ import (
 
 	"github.com/pkg/errors"
 
-	"github.com/zrepl/zrepl/util/errorarray"
 	"github.com/zrepl/zrepl/zfs"
 )
 
@@ -93,7 +92,7 @@ func HoldStep(ctx context.Context, fs string, v zfs.FilesystemVersion, jobID Job
 		return nil, errors.Wrap(err, "create step bookmark: determine bookmark name")
 	}
 	// idempotently create bookmark
-	err = zfs.ZFSBookmark(ctx, fs, v.ToSendArgVersion(), bmname)
+	stepBookmark, err := zfs.ZFSBookmark(ctx, fs, v, bmname)
 	if err != nil {
 		if err == zfs.ErrBookmarkCloningNotSupported {
 			// TODO we could actually try to find a local snapshot that has the requested GUID
@@ -108,137 +107,11 @@ func HoldStep(ctx context.Context, fs string, v zfs.FilesystemVersion, jobID Job
 	return &bookmarkBasedAbstraction{
 		Type:              AbstractionStepBookmark,
 		FS:                fs,
-		FilesystemVersion: v,
+		FilesystemVersion: stepBookmark,
 		JobID:             jobID,
 	}, nil
 }
 
-// idempotently release the step-hold on v if v is a snapshot
-// or idempotently destroy the step-bookmark  of v if v is a bookmark
-//
-// note that this operation leaves v itself untouched, unless v is the step-bookmark itself, in which case v is destroyed
-//
-// returns an instance of *zfs.DatasetDoesNotExist if `v` does not exist
-func ReleaseStep(ctx context.Context, fs string, v zfs.FilesystemVersion, jobID JobID) error {
-
-	if v.IsSnapshot() {
-		tag, err := StepHoldTag(jobID)
-		if err != nil {
-			return errors.Wrap(err, "step release tag")
-		}
-
-		if err := zfs.ZFSRelease(ctx, tag, v.FullPath(fs)); err != nil {
-			return errors.Wrap(err, "step release: zfs")
-		}
-
-		return nil
-	}
-	if !v.IsBookmark() {
-		panic(fmt.Sprintf("impl error: expecting version to be a bookmark, got %#v", v))
-	}
-
-	bmname, err := StepBookmarkName(fs, v.Guid, jobID)
-	if err != nil {
-		return errors.Wrap(err, "step release: determine bookmark name")
-	}
-	// idempotently destroy bookmark
-
-	if err := zfs.ZFSDestroyIdempotent(ctx, bmname); err != nil {
-		return errors.Wrap(err, "step release: bookmark destroy: zfs")
-	}
-
-	return nil
-}
-
-// release {step holds, step bookmarks} earlier and including `mostRecent`
-func ReleaseStepCummulativeInclusive(ctx context.Context, fs string, since *CreateTXGRangeBound, mostRecent zfs.FilesystemVersion, jobID JobID) error {
-	q := ListZFSHoldsAndBookmarksQuery{
-		What: AbstractionTypeSet{
-			AbstractionStepHold:     true,
-			AbstractionStepBookmark: true,
-		},
-		FS: ListZFSHoldsAndBookmarksQueryFilesystemFilter{
-			FS: &fs,
-		},
-		JobID: &jobID,
-		CreateTXG: CreateTXGRange{
-			Since: since,
-			Until: &CreateTXGRangeBound{
-				CreateTXG: mostRecent.CreateTXG,
-				Inclusive: &zfs.NilBool{B: true},
-			},
-		},
-		Concurrency: 1,
-	}
-	abs, absErrs, err := ListAbstractions(ctx, q)
-	if err != nil {
-		return errors.Wrap(err, "step release cummulative: list")
-	}
-	if len(absErrs) > 0 {
-		return errors.Wrap(ListAbstractionsErrors(absErrs), "step release cummulative: list")
-	}
-
-	getLogger(ctx).WithField("step_holds_and_bookmarks", fmt.Sprintf("%s", abs)).Debug("releasing step holds and bookmarks")
-
-	var errs []error
-	for res := range BatchDestroy(ctx, abs) {
-		log := getLogger(ctx).
-			WithField("step_hold_or_bookmark", res.Abstraction)
-		if res.DestroyErr != nil {
-			errs = append(errs, res.DestroyErr)
-			log.WithError(err).
-				Error("cannot release step hold or bookmark")
-		} else {
-			log.Info("released step hold or bookmark")
-		}
-	}
-	if len(errs) == 0 {
-		return nil
-	} else {
-		return errorarray.Wrap(errs, "step release cummulative: release")
-	}
-}
-
-func TryReleaseStepStaleFS(ctx context.Context, fs string, jobID JobID) {
-
-	q := ListZFSHoldsAndBookmarksQuery{
-		FS: ListZFSHoldsAndBookmarksQueryFilesystemFilter{
-			FS: &fs,
-		},
-		JobID: &jobID,
-		What: AbstractionTypeSet{
-			AbstractionStepHold:                    true,
-			AbstractionStepBookmark:                true,
-			AbstractionReplicationCursorBookmarkV2: true,
-		},
-		Concurrency: 1,
-	}
-	staleness, err := ListStale(ctx, q)
-	if _, ok := err.(*ListStaleQueryError); ok {
-		panic(err)
-	} else if err != nil {
-		getLogger(ctx).WithError(err).Error("cannot list stale step holds and bookmarks")
-		return
-	}
-	for _, s := range staleness.Stale {
-		getLogger(ctx).WithField("stale_step_hold_or_bookmark", s).Info("batch-destroying stale step hold or bookmark")
-	}
-	for res := range BatchDestroy(ctx, staleness.Stale) {
-		if res.DestroyErr != nil {
-			getLogger(ctx).
-				WithField("stale_step_hold_or_bookmark", res.Abstraction).
-				WithError(res.DestroyErr).
-				Error("cannot destroy stale step-hold or bookmark")
-		} else {
-			getLogger(ctx).
-				WithField("stale_step_hold_or_bookmark", res.Abstraction).
-				WithError(res.DestroyErr).
-				Info("destroyed stale step-hold or bookmark")
-		}
-	}
-
-}
-
 var _ BookmarkExtractor = StepBookmarkExtractor
 
 func StepBookmarkExtractor(fs *zfs.DatasetPath, v zfs.FilesystemVersion) (_ Abstraction) {

go.mod

@@ -22,7 +22,7 @@ require (
 	github.com/onsi/gomega v1.7.0 // indirect
 	github.com/pkg/errors v0.8.1
 	github.com/pkg/profile v1.2.1
-	github.com/problame/go-netssh v0.0.0-20191209123953-18d8aa6923c7
+	github.com/problame/go-netssh v0.0.0-20200601114649-26439f9f0dc5
 	github.com/prometheus/client_golang v1.2.1
 	github.com/prometheus/common v0.7.0
 	github.com/sergi/go-diff v1.0.1-0.20180205163309-da645544ed44 // go1.12 thinks it needs this
@@ -36,6 +36,7 @@ require (
 	golang.org/x/net v0.0.0-20190613194153-d28f0bde5980
 	golang.org/x/sync v0.0.0-20190423024810-112230192c58
 	golang.org/x/sys v0.0.0-20191026070338-33540a1f6037
+	golang.org/x/tools v0.0.0-20190206041539-40960b6deb8e
 	gonum.org/v1/gonum v0.7.0 // indirect
 	google.golang.org/grpc v1.17.0
 )

go.sum

@@ -212,6 +212,8 @@ github.com/problame/go-netssh v0.0.0-20191026123024-f34099f4f6b1 h1:HH8yzlaZq/A8
 github.com/problame/go-netssh v0.0.0-20191026123024-f34099f4f6b1/go.mod h1:JRs3nyBjvOv/9YHC+Vlw9z1Jfzl5s5t0BNnTzmclj1c=
 github.com/problame/go-netssh v0.0.0-20191209123953-18d8aa6923c7 h1:Oik8tLrLhoMq4fduDRuNNOAQ+q0M0dXkjAYLUf4+mAc=
 github.com/problame/go-netssh v0.0.0-20191209123953-18d8aa6923c7/go.mod h1:Ba6RaFpK1+IHWs1wLZ6sCBuXkt4iAVLeOG5GinXHusM=
+github.com/problame/go-netssh v0.0.0-20200601114649-26439f9f0dc5 h1:1eSrO2WAeSXjMol8WRKW9LekL1252VIMaKQwWkmEdvs=
+github.com/problame/go-netssh v0.0.0-20200601114649-26439f9f0dc5/go.mod h1:Ba6RaFpK1+IHWs1wLZ6sCBuXkt4iAVLeOG5GinXHusM=
 github.com/prometheus/client_golang v0.0.0-20180410130117-e11c6ff8170b/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
 github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
 github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=
@@ -268,6 +270,7 @@ github.com/spf13/viper v1.0.2/go.mod h1:A8kyI5cUJhb8N+3pkfONlcEcZbueH6nhAm0Fq7Sr
 github.com/spf13/viper v1.3.2/go.mod h1:ZiWeW+zYFKm7srdB9IoDzzZXaJaI5eL9QjNiN/DMA2s=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.2.1/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
@@ -350,6 +353,7 @@ golang.org/x/tools v0.0.0-20181117154741-2ddaf7f79a09/go.mod h1:n7NCudcB/nEzxVGm
 golang.org/x/tools v0.0.0-20181205014116-22934f0fdb62/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190110163146-51295c7ec13a/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190121143147-24cd39ecf745/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20190206041539-40960b6deb8e h1:Io7mpb+aUAGF0MKxbyQ7HQl1VgB+cL6ZJZUFaFNqVV4=
 golang.org/x/tools v0.0.0-20190206041539-40960b6deb8e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190213192042-740235f6c0d8/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=

lazy.sh

@@ -36,9 +36,10 @@ godep() {
     go build -v -mod=readonly -o "$GOPATH/bin/enumer"        github.com/alvaroloes/enumer
     go build -v -mod=readonly -o "$GOPATH/bin/goimports"     golang.org/x/tools/cmd/goimports
     go build -v -mod=readonly -o "$GOPATH/bin/golangci-lint" github.com/golangci/golangci-lint/cmd/golangci-lint
+    go build -v -mod=readonly -o "$GOPATH/bin/gocovmerge"    github.com/wadey/gocovmerge
     set +x
     popd
-    if ! type stringer || ! type protoc-gen-go || ! type enumer || ! type goimports || ! type golangci-lint; then
+    if ! type stringer || ! type protoc-gen-go || ! type enumer || ! type goimports || ! type golangci-lint || ! type gocovmerge; then
         echo "Installed dependencies but can't find them in \$PATH, adjust it to contain \$GOPATH/bin" 1>&2
         exit 1
     fi

platformtest/harness/harness.go

@@ -24,33 +24,36 @@ var bold = color.New(color.Bold)
 var boldRed = color.New(color.Bold, color.FgHiRed)
 var boldGreen = color.New(color.Bold, color.FgHiGreen)
 
-var args struct {
+const DefaultPoolImageSize = 200 * (1 << 20)
-	createArgs            platformtest.ZpoolCreateArgs
-	stopAndKeepPoolOnFail bool
-
-	run   string
-	runRE *regexp.Regexp
-}
 
 func main() {
-	if err := doMain(); err != nil {
+
+	var args HarnessArgs
+
+	flag.StringVar(&args.CreateArgs.PoolName, "poolname", "", "")
+	flag.StringVar(&args.CreateArgs.ImagePath, "imagepath", "", "")
+	flag.Int64Var(&args.CreateArgs.ImageSize, "imagesize", DefaultPoolImageSize, "")
+	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.Parse()
+
+	if err := HarnessRun(args); err != nil {
 		os.Exit(1)
 	}
 }
 
 var exitWithErr = fmt.Errorf("exit with error")
 
-func doMain() error {
+type HarnessArgs struct {
+	CreateArgs            platformtest.ZpoolCreateArgs
+	StopAndKeepPoolOnFail bool
+	Run                   string
+}
 
-	flag.StringVar(&args.createArgs.PoolName, "poolname", "", "")
+func HarnessRun(args HarnessArgs) error {
-	flag.StringVar(&args.createArgs.ImagePath, "imagepath", "", "")
-	flag.Int64Var(&args.createArgs.ImageSize, "imagesize", 200*(1<<20), "")
-	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.Parse()
 
-	args.runRE = regexp.MustCompile(args.run)
+	runRE := regexp.MustCompile(args.Run)
 
 	outlets := logger.NewOutlets()
 	outlet, level, err := logging.ParseOutlet(config.LoggingOutletEnum{Ret: &config.StdoutLoggingOutlet{
@@ -65,13 +68,13 @@ func doMain() error {
 	outlets.Add(outlet, level)
 	logger := logger.NewLogger(outlets, 1*time.Second)
 
-	if err := args.createArgs.Validate(); err != nil {
+	if err := args.CreateArgs.Validate(); err != nil {
 		logger.Error(err.Error())
 		panic(err)
 	}
 	ctx := context.Background()
 	defer trace.WithTaskFromStackUpdateCtx(&ctx)()
-	ctx = platformtest.WithLogger(ctx, logger)
+	ctx = logging.WithLoggers(ctx, logging.SubsystemLoggersWithUniversalLogger(logger))
 	ex := platformtest.NewEx(logger)
 
 	type invocation struct {
@@ -81,7 +84,7 @@ func doMain() error {
 
 	invocations := make([]*invocation, 0, len(tests.Cases))
 	for _, c := range tests.Cases {
-		if args.runRE.MatchString(c.String()) {
+		if runRE.MatchString(c.String()) {
 			invocations = append(invocations, &invocation{runFunc: c})
 		}
 	}
@@ -90,7 +93,7 @@ func doMain() error {
 
 		bold.Printf("BEGIN TEST CASE %s\n", inv.runFunc.String())
 
-		pool, err := platformtest.CreateOrReplaceZpool(ctx, ex, args.createArgs)
+		pool, err := platformtest.CreateOrReplaceZpool(ctx, ex, args.CreateArgs)
 		if err != nil {
 			panic(errors.Wrap(err, "create test pool"))
 		}
@@ -106,7 +109,7 @@ func doMain() error {
 			fmt.Printf("%+v\n", res.failedStack) // print with stack trace
 		}
 
-		if res.failed && args.stopAndKeepPoolOnFail {
+		if res.failed && args.StopAndKeepPoolOnFail {
 			boldRed.Printf("STOPPING TEST RUN AT FAILING TEST PER USER REQUEST\n")
 			return exitWithErr
 		}

platformtest/harness/harness_test.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"fmt"
 	"os"
 	"strings"
 	"testing"
@@ -21,22 +22,29 @@ import (
 //  https://github.com/wadey/gocovmerge
 
 func TestMain(t *testing.T) {
+	fmt.Println("incoming args: ", os.Args)
+
 	var (
 		args             []string
 		run              bool
+		startCaptureArgs bool
 	)
 
-	for _, arg := range os.Args {
+	for i, arg := range os.Args {
 		switch {
 		case arg == "__DEVEL--i-heard-you-like-tests":
 			run = true
+			startCaptureArgs = true
 		case strings.HasPrefix(arg, "-test"):
 		case strings.HasPrefix(arg, "__DEVEL"):
-		default:
+		case i == 0:
+			args = append(args, arg)
+		case startCaptureArgs:
 			args = append(args, arg)
 		}
 	}
 	os.Args = args
+	fmt.Println("using args: ", os.Args)
 
 	if run {
 		main()

platformtest/platformtest.go

@@ -27,6 +27,7 @@ func (c *Context) Logf(format string, args ...interface{}) {
 
 func (c *Context) Errorf(format string, args ...interface{}) {
 	GetLog(c).Printf(format, args...)
+	c.FailNow()
 }
 
 func (c *Context) FailNow() {

platformtest/platformtest_logging.go

@@ -3,22 +3,12 @@ package platformtest
 import (
 	"context"
 
+	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 )
 
 type Logger = logger.Logger
 
-type contextKey int
-
-const (
-	contextKeyLogger contextKey = iota
-)
-
-func WithLogger(ctx context.Context, logger Logger) context.Context {
-	ctx = context.WithValue(ctx, contextKeyLogger, logger)
-	return ctx
-}
-
 func GetLog(ctx context.Context) Logger {
-	return ctx.Value(contextKeyLogger).(Logger)
+	return logging.GetLogger(ctx, logging.SubsysPlatformtest)
 }

platformtest/platformtest_zpool.go

@@ -58,6 +58,14 @@ func CreateOrReplaceZpool(ctx context.Context, e Execer, args ZpoolCreateArgs) (
 		}
 	}
 
+	// clear the mountpoint dir
+	if err := os.RemoveAll(args.Mountpoint); err != nil {
+		return nil, errors.Wrapf(err, "remove mountpoint dir %q", args.Mountpoint)
+	}
+	if err := os.Mkdir(args.Mountpoint, 0700); err != nil {
+		return nil, errors.Wrapf(err, "create mountpoint dir %q", args.Mountpoint)
+	}
+
 	// idempotently (re)create the pool image
 	image, err := os.OpenFile(args.ImagePath, os.O_CREATE|os.O_RDWR, 0600)
 	if err != nil {
@@ -93,5 +101,9 @@ func (p *Zpool) Destroy(ctx context.Context, e Execer) error {
 		return errors.Wrapf(err, "remove pool image")
 	}
 
+	if err := os.RemoveAll(p.args.Mountpoint); err != nil {
+		return errors.Wrapf(err, "remove mountpoint dir %q", p.args.Mountpoint)
+	}
+
 	return nil
 }

platformtest/tests/gen/gen.go

@@ -0,0 +1,140 @@
+package main
+
+import (
+	"bytes"
+	"go/ast"
+	"go/format"
+	"go/parser"
+	"go/token"
+	"io/ioutil"
+	"os"
+	"sort"
+	"strings"
+	"text/template"
+
+	"golang.org/x/tools/go/packages"
+)
+
+func check(err error) {
+	if err != nil {
+		panic(err)
+	}
+}
+
+type platformtestFuncDeclFinder struct {
+	pkg       *packages.Package
+	testFuncs []*ast.FuncDecl
+}
+
+func isPlatformtestFunc(n *ast.FuncDecl) bool {
+	if !n.Name.IsExported() {
+		return false
+	}
+	if n.Recv != nil {
+		return false
+	}
+	if n.Type.Results.NumFields() != 0 {
+		return false
+	}
+	if n.Type.Params.NumFields() != 1 {
+		return false
+	}
+	se, ok := n.Type.Params.List[0].Type.(*ast.StarExpr)
+	if !ok {
+		return false
+	}
+	sel, ok := se.X.(*ast.SelectorExpr)
+	if !ok {
+		return false
+	}
+	x, ok := sel.X.(*ast.Ident)
+	if !ok {
+		return false
+	}
+	if x.Name != "platformtest" || sel.Sel.Name != "Context" {
+		return false
+	}
+	return true
+}
+
+func (e *platformtestFuncDeclFinder) Visit(n2 ast.Node) ast.Visitor {
+	switch n := n2.(type) {
+	case *ast.File:
+		return e
+	case *ast.FuncDecl:
+		if isPlatformtestFunc(n) {
+			e.testFuncs = append(e.testFuncs, n)
+		}
+		return nil
+	default:
+		return nil
+	}
+}
+
+func main() {
+
+	// TODO safeguards that prevent us from deleting non-generated generated_cases.go
+	os.Remove("generated_cases.go")
+	// (no error handling to easily cover the case where the file doesn't exist)
+
+	pkgs, err := packages.Load(
+		&packages.Config{
+			Mode:  packages.LoadFiles,
+			Tests: false,
+		},
+		os.Args[1],
+	)
+	check(err)
+
+	if len(pkgs) != 1 {
+		panic(pkgs)
+	}
+
+	p := pkgs[0]
+
+	var tests []*ast.FuncDecl
+
+	for _, f := range p.GoFiles {
+		s := token.NewFileSet()
+		a, err := parser.ParseFile(s, f, nil, parser.AllErrors)
+		check(err)
+		finder := &platformtestFuncDeclFinder{
+			pkg: p,
+		}
+		ast.Walk(finder, a)
+		tests = append(tests, finder.testFuncs...)
+	}
+
+	sort.Slice(tests, func(i, j int) bool {
+		return strings.Compare(tests[i].Name.Name, tests[j].Name.Name) < 0
+	})
+
+	{
+		casesTemplate := `
+// Code generated by zrepl tooling; DO NOT EDIT.
+
+package tests
+
+var Cases = []Case {
+{{- range . -}}
+	{{ .Name }},
+{{ end -}}
+}
+
+	`
+		t, err := template.New("CaseFunc").Parse(casesTemplate)
+		check(err)
+
+		var buf bytes.Buffer
+		err = t.Execute(&buf, tests)
+		check(err)
+
+		formatted, err := format.Source(buf.Bytes())
+		check(err)
+
+		err = ioutil.WriteFile("generated_cases.go", formatted, 0664)
+		check(err)
+
+	}
+
+}

platformtest/tests/generated_cases.go

@@ -0,0 +1,30 @@
+// Code generated by zrepl tooling; DO NOT EDIT.
+
+package tests
+
+var Cases = []Case{BatchDestroy,
+	CreateReplicationCursor,
+	GetNonexistent,
+	HoldsWork,
+	IdempotentBookmark,
+	IdempotentDestroy,
+	IdempotentHold,
+	ListFilesystemVersionsFilesystemNotExist,
+	ListFilesystemVersionsTypeFilteringAndPrefix,
+	ListFilesystemVersionsUserrefs,
+	ListFilesystemVersionsZeroExistIsNotAnError,
+	ListFilesystemsNoFilter,
+	ReceiveForceIntoEncryptedErr,
+	ReceiveForceRollbackWorksUnencrypted,
+	ReplicationIncrementalCleansUpStaleAbstractionsWithCacheOnSecondReplication,
+	ReplicationIncrementalCleansUpStaleAbstractionsWithoutCacheOnSecondReplication,
+	ReplicationIncrementalIsPossibleIfCommonSnapshotIsDestroyed,
+	ReplicationIsResumableFullSend__DisableIncrementalStepHolds_False,
+	ReplicationIsResumableFullSend__DisableIncrementalStepHolds_True,
+	ResumableRecvAndTokenHandling,
+	ResumeTokenParsing,
+	SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden,
+	SendArgsValidationResumeTokenDifferentFilesystemForbidden,
+	SendArgsValidationResumeTokenEncryptionMismatchForbidden,
+	UndestroyableSnapshotParsing,
+}

platformtest/tests/holds.go

@@ -0,0 +1,32 @@
+package tests
+
+import (
+	"path"
+
+	"github.com/stretchr/testify/require"
+	"github.com/zrepl/zrepl/platformtest"
+	"github.com/zrepl/zrepl/zfs"
+)
+
+func HoldsWork(ctx *platformtest.Context) {
+	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+		DESTROYROOT
+		CREATEROOT
+		+  "foo bar"
+		+  "foo bar@snap name"
+	`)
+
+	fs := path.Join(ctx.RootDataset, "foo bar")
+
+	err := zfs.ZFSHold(ctx, fs, fsversion(ctx, fs, "@snap name"), "tag 1")
+	require.NoError(ctx, err)
+
+	err = zfs.ZFSHold(ctx, fs, fsversion(ctx, fs, "@snap name"), "tag 2")
+	require.NoError(ctx, err)
+
+	holds, err := zfs.ZFSHolds(ctx, fs, "snap name")
+	require.NoError(ctx, err)
+	require.Len(ctx, holds, 2)
+	require.Contains(ctx, holds, "tag 1")
+	require.Contains(ctx, holds, "tag 2")
+}

platformtest/tests/idempotentBookmark.go

@@ -3,6 +3,7 @@ package tests
 import (
 	"fmt"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/zrepl/zrepl/platformtest"
 	"github.com/zrepl/zrepl/zfs"
 )
@@ -19,22 +20,23 @@ func IdempotentBookmark(ctx *platformtest.Context) {
 
 	fs := fmt.Sprintf("%s/foo bar", ctx.RootDataset)
 
-	asnap := sendArgVersion(ctx, fs, "@a snap")
+	asnap := fsversion(ctx, fs, "@a snap")
-	anotherSnap := sendArgVersion(ctx, fs, "@another snap")
+	anotherSnap := fsversion(ctx, fs, "@another snap")
 
-	err := zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
+	aBookmark, err := zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
 	if err != nil {
 		panic(err)
 	}
 
 	// do it again, should be idempotent
-	err = zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
+	aBookmarkIdemp, err := zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
 	if err != nil {
 		panic(err)
 	}
+	assert.Equal(ctx, aBookmark, aBookmarkIdemp)
 
 	// should fail for another snapshot
-	err = zfs.ZFSBookmark(ctx, fs, anotherSnap, "a bookmark")
+	_, err = zfs.ZFSBookmark(ctx, fs, anotherSnap, "a bookmark")
 	if err == nil {
 		panic(err)
 	}
@@ -48,7 +50,7 @@ func IdempotentBookmark(ctx *platformtest.Context) {
 	}
 
 	// do it again, should fail with special error type
-	err = zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
+	_, err = zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
 	if err == nil {
 		panic(err)
 	}

platformtest/tests/idempotentDestroy.go

@@ -18,8 +18,8 @@ func IdempotentDestroy(ctx *platformtest.Context) {
 	`)
 
 	fs := fmt.Sprintf("%s/foo bar", ctx.RootDataset)
-	asnap := sendArgVersion(ctx, fs, "@a snap")
+	asnap := fsversion(ctx, fs, "@a snap")
-	err := zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
+	_, err := zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
 	if err != nil {
 		panic(err)
 	}

platformtest/tests/listFilesystemVersions.go

@@ -85,9 +85,6 @@ func ListFilesystemVersionsZeroExistIsNotAnError(t *platformtest.Context) {
 	vs, err := zfs.ZFSListFilesystemVersions(t, mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
 	require.Empty(t, vs)
 	require.NoError(t, err)
-	dsne, ok := err.(*zfs.DatasetDoesNotExist)
-	require.True(t, ok)
-	require.Equal(t, fs, dsne.Path)
 }
 
 func ListFilesystemVersionsFilesystemNotExist(t *platformtest.Context) {

platformtest/tests/recvForceIntoEncryptedErr.go

@@ -0,0 +1,45 @@
+package tests
+
+import (
+	"fmt"
+
+	"github.com/stretchr/testify/require"
+	"github.com/zrepl/zrepl/platformtest"
+	"github.com/zrepl/zrepl/zfs"
+)
+
+func ReceiveForceIntoEncryptedErr(ctx *platformtest.Context) {
+	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+		DESTROYROOT
+		CREATEROOT
+		+  "foo bar" encrypted
+		+  "sender" encrypted
+		+  "sender@1"
+	`)
+
+	rfs := fmt.Sprintf("%s/foo bar", ctx.RootDataset)
+	sfs := fmt.Sprintf("%s/sender", ctx.RootDataset)
+	sfsSnap1 := sendArgVersion(ctx, sfs, "@1")
+
+	sendArgs, err := zfs.ZFSSendArgsUnvalidated{
+		FS:          sfs,
+		Encrypted:   &zfs.NilBool{B: false},
+		From:        nil,
+		To:          &sfsSnap1,
+		ResumeToken: "",
+	}.Validate(ctx)
+	require.NoError(ctx, err)
+
+	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
+	require.NoError(ctx, err)
+
+	recvOpts := zfs.RecvOptions{
+		RollbackAndForceRecv: true,
+		SavePartialRecvState: false,
+	}
+	err = zfs.ZFSRecv(ctx, rfs, &zfs.ZFSSendArgVersion{RelName: "@1", GUID: sfsSnap1.GUID}, sendStream, recvOpts)
+	require.Error(ctx, err)
+	re, ok := err.(*zfs.RecvDestroyOrOverwriteEncryptedErr)
+	require.True(ctx, ok)
+	require.Contains(ctx, re.Error(), "zfs receive -F cannot be used to destroy an encrypted filesystem or overwrite an unencrypted one with an encrypted on")
+}

platformtest/tests/recvRollback.go

@@ -0,0 +1,54 @@
+package tests
+
+import (
+	"fmt"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/zrepl/zrepl/platformtest"
+	"github.com/zrepl/zrepl/zfs"
+)
+
+func ReceiveForceRollbackWorksUnencrypted(ctx *platformtest.Context) {
+	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+		DESTROYROOT
+		CREATEROOT
+		+  "foo bar"
+		+  "foo bar@a snap"
+		+  "foo bar@another snap"
+		+  "foo bar@snap3"
+		+  "sender"
+		+  "sender@1"
+	`)
+
+	rfs := fmt.Sprintf("%s/foo bar", ctx.RootDataset)
+	sfs := fmt.Sprintf("%s/sender", ctx.RootDataset)
+	sfsSnap1 := sendArgVersion(ctx, sfs, "@1")
+
+	sendArgs, err := zfs.ZFSSendArgsUnvalidated{
+		FS:          sfs,
+		Encrypted:   &zfs.NilBool{B: false},
+		From:        nil,
+		To:          &sfsSnap1,
+		ResumeToken: "",
+	}.Validate(ctx)
+	require.NoError(ctx, err)
+
+	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
+	require.NoError(ctx, err)
+
+	recvOpts := zfs.RecvOptions{
+		RollbackAndForceRecv: true,
+		SavePartialRecvState: false,
+	}
+	err = zfs.ZFSRecv(ctx, rfs, &zfs.ZFSSendArgVersion{RelName: "@1", GUID: sfsSnap1.GUID}, sendStream, recvOpts)
+	require.NoError(ctx, err)
+
+	// assert exists on receiver
+	rfsSnap1 := fsversion(ctx, rfs, "@1")
+	// assert it's the only one (rollback and force-recv should be blowing away the other filesystems)
+	rfsVersions, err := zfs.ZFSListFilesystemVersions(ctx, mustDatasetPath(rfs), zfs.ListFilesystemVersionsOptions{})
+	require.NoError(ctx, err)
+	assert.Len(ctx, rfsVersions, 1)
+	assert.Equal(ctx, rfsVersions[0], rfsSnap1)
+}

platformtest/tests/replication.go

@@ -0,0 +1,409 @@
+package tests
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"path"
+
+	"github.com/kr/pretty"
+	"github.com/stretchr/testify/require"
+	"github.com/zrepl/zrepl/daemon/filters"
+	"github.com/zrepl/zrepl/endpoint"
+	"github.com/zrepl/zrepl/platformtest"
+	"github.com/zrepl/zrepl/replication"
+	"github.com/zrepl/zrepl/replication/logic"
+	"github.com/zrepl/zrepl/replication/logic/pdu"
+	"github.com/zrepl/zrepl/replication/report"
+	"github.com/zrepl/zrepl/util/limitio"
+	"github.com/zrepl/zrepl/zfs"
+)
+
+// mimics the replication invocations of an active-side job
+// for a single sender-receiver filesystem pair
+//
+// each invocation of method Do results in the construction
+// of a new sender and receiver instance and one blocking invocation
+// of the replication engine without encryption
+type replicationInvocation struct {
+	sjid, rjid                  endpoint.JobID
+	sfs                         string
+	rfsRoot                     string
+	interceptSender             func(e *endpoint.Sender) logic.Sender
+	disableIncrementalStepHolds bool
+}
+
+func (i replicationInvocation) Do(ctx *platformtest.Context) *report.Report {
+
+	if i.interceptSender == nil {
+		i.interceptSender = func(e *endpoint.Sender) logic.Sender { return e }
+	}
+
+	sfilter := filters.NewDatasetMapFilter(1, true)
+	err := sfilter.Add(i.sfs, "ok")
+	require.NoError(ctx, err)
+	sender := i.interceptSender(endpoint.NewSender(endpoint.SenderConfig{
+		FSF:                         sfilter.AsFilter(),
+		Encrypt:                     &zfs.NilBool{B: false},
+		DisableIncrementalStepHolds: i.disableIncrementalStepHolds,
+		JobID:                       i.sjid,
+	}))
+	receiver := endpoint.NewReceiver(endpoint.ReceiverConfig{
+		JobID:                      i.rjid,
+		AppendClientIdentity:       false,
+		RootWithoutClientComponent: mustDatasetPath(i.rfsRoot),
+		UpdateLastReceivedHold:     true,
+	})
+	plannerPolicy := logic.PlannerPolicy{
+		EncryptedSend: logic.TriFromBool(false),
+	}
+
+	report, wait := replication.Do(
+		ctx,
+		logic.NewPlanner(nil, nil, sender, receiver, plannerPolicy),
+	)
+	wait(true)
+	return report()
+}
+
+func (i replicationInvocation) ReceiveSideFilesystem() string {
+	return path.Join(i.rfsRoot, i.sfs)
+}
+
+func ReplicationIncrementalIsPossibleIfCommonSnapshotIsDestroyed(ctx *platformtest.Context) {
+
+	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+		CREATEROOT
+		+  "sender"
+		+  "sender@1"
+		+  "receiver"
+		R  zfs create -p "${ROOTDS}/receiver/${ROOTDS}"
+	`)
+
+	sjid := endpoint.MustMakeJobID("sender-job")
+	rjid := endpoint.MustMakeJobID("receiver-job")
+
+	sfs := ctx.RootDataset + "/sender"
+	rfsRoot := ctx.RootDataset + "/receiver"
+	snap1 := fsversion(ctx, sfs, "@1")
+
+	rep := replicationInvocation{
+		sjid:                        sjid,
+		rjid:                        rjid,
+		sfs:                         sfs,
+		rfsRoot:                     rfsRoot,
+		disableIncrementalStepHolds: false,
+	}
+	rfs := rep.ReceiveSideFilesystem()
+
+	// first replication
+	report := rep.Do(ctx)
+	ctx.Logf("\n%s", pretty.Sprint(report))
+
+	// assert @1 exists on receiver
+	_ = fsversion(ctx, rfs, "@1")
+
+	// cut off the common base between sender and receiver
+	// (replication engine guarantees resumability through bookmarks)
+	err := zfs.ZFSDestroy(ctx, snap1.FullPath(sfs))
+	require.NoError(ctx, err)
+
+	// assert that the replication cursor has been created
+	snap1CursorName, err := endpoint.ReplicationCursorBookmarkName(sfs, snap1.Guid, sjid)
+	require.NoError(ctx, err)
+	snap1CursorInfo, err := zfs.ZFSGetFilesystemVersion(ctx, sfs+"#"+snap1CursorName)
+	require.NoError(ctx, err)
+	require.True(ctx, snap1CursorInfo.IsBookmark())
+
+	// second replication of a new snapshot, should use the cursor
+	mustSnapshot(ctx, sfs+"@2")
+	report = rep.Do(ctx)
+	ctx.Logf("\n%s", pretty.Sprint(report))
+	_ = fsversion(ctx, rfs, "@2")
+
+}
+
+func ReplicationIncrementalCleansUpStaleAbstractionsWithCacheOnSecondReplication(ctx *platformtest.Context) {
+	implReplicationIncrementalCleansUpStaleAbstractions(ctx, true)
+}
+
+func ReplicationIncrementalCleansUpStaleAbstractionsWithoutCacheOnSecondReplication(ctx *platformtest.Context) {
+	implReplicationIncrementalCleansUpStaleAbstractions(ctx, false)
+}
+
+func implReplicationIncrementalCleansUpStaleAbstractions(ctx *platformtest.Context, invalidateCacheBeforeSecondReplication bool) {
+
+	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+		CREATEROOT
+		+  "sender"
+		+  "sender@1"
+		+  "sender@2"
+		+  "sender#2" "sender@2"
+		+  "sender@3"
+		+  "receiver"
+		R  zfs create -p "${ROOTDS}/receiver/${ROOTDS}"
+	`)
+
+	sjid := endpoint.MustMakeJobID("sender-job")
+	ojid := endpoint.MustMakeJobID("other-job")
+	rjid := endpoint.MustMakeJobID("receiver-job")
+
+	sfs := ctx.RootDataset + "/sender"
+	rfsRoot := ctx.RootDataset + "/receiver"
+
+	rep := replicationInvocation{
+		sjid:                        sjid,
+		rjid:                        rjid,
+		sfs:                         sfs,
+		rfsRoot:                     rfsRoot,
+		disableIncrementalStepHolds: false,
+	}
+	rfs := rep.ReceiveSideFilesystem()
+
+	// first replication
+	report := rep.Do(ctx)
+	ctx.Logf("\n%s", pretty.Sprint(report))
+
+	// assert most recent send-side version @3 exists on receiver (=replication succeeded)
+	rSnap3 := fsversion(ctx, rfs, "@3")
+	// assert the source-side versions not managed by zrepl still exist
+	snap1 := fsversion(ctx, sfs, "@1")
+	snap2 := fsversion(ctx, sfs, "@2")
+	_ = fsversion(ctx, sfs, "#2") // non-replicationc-cursor bookmarks should not be affected
+	snap3 := fsversion(ctx, sfs, "@3")
+	// assert a replication cursor is in place
+	snap3CursorName, err := endpoint.ReplicationCursorBookmarkName(sfs, snap3.Guid, sjid)
+	require.NoError(ctx, err)
+	_ = fsversion(ctx, sfs, "#"+snap3CursorName)
+	// assert a last-received hold is in place
+	expectRjidHoldTag, err := endpoint.LastReceivedHoldTag(rjid)
+	require.NoError(ctx, err)
+	holds, err := zfs.ZFSHolds(ctx, rfs, rSnap3.Name)
+	require.NoError(ctx, err)
+	require.Contains(ctx, holds, expectRjidHoldTag)
+
+	// create artifical stale replication cursors
+	createArtificalStaleAbstractions := func(jobId endpoint.JobID) []endpoint.Abstraction {
+		snap2Cursor, err := endpoint.CreateReplicationCursor(ctx, sfs, snap2, jobId) // no shadow
+		require.NoError(ctx, err)
+		// create artifical stale step holds jobId
+		snap1Hold, err := endpoint.HoldStep(ctx, sfs, snap1, jobId) // no shadow
+		require.NoError(ctx, err)
+		snap2Hold, err := endpoint.HoldStep(ctx, sfs, snap2, jobId) // no shadow
+		require.NoError(ctx, err)
+		return []endpoint.Abstraction{snap2Cursor, snap1Hold, snap2Hold}
+	}
+	createArtificalStaleAbstractions(sjid)
+	ojidSendAbstractions := createArtificalStaleAbstractions(ojid)
+
+	snap3ojidLastReceivedHold, err := endpoint.CreateLastReceivedHold(ctx, rfs, fsversion(ctx, rfs, "@3"), ojid)
+	require.NoError(ctx, err)
+	require.True(ctx, zfs.FilesystemVersionEqualIdentity(fsversion(ctx, rfs, "@3"), snap3ojidLastReceivedHold.GetFilesystemVersion()))
+
+	// take another 2 snapshots
+	mustSnapshot(ctx, sfs+"@4")
+	mustSnapshot(ctx, sfs+"@5")
+	snap5 := fsversion(ctx, sfs, "@5")
+
+	if invalidateCacheBeforeSecondReplication {
+		endpoint.SendAbstractionsCacheInvalidate(sfs)
+	}
+
+	// do another replication
+	// - ojid's abstractions should not be affected on either side
+	// - stale abstractions of sjid and rjid should be cleaned up
+	// - 1 replication cursors and 1 last-received hold should be present
+
+	checkOjidAbstractionsExist := func() {
+		var expectedOjidAbstractions []endpoint.Abstraction
+		expectedOjidAbstractions = append(expectedOjidAbstractions, ojidSendAbstractions...)
+		expectedOjidAbstractions = append(expectedOjidAbstractions, snap3ojidLastReceivedHold)
+
+		sfsAndRfsFilter := filters.NewDatasetMapFilter(2, true)
+		require.NoError(ctx, sfsAndRfsFilter.Add(sfs, "ok"))
+		require.NoError(ctx, sfsAndRfsFilter.Add(rfs, "ok"))
+		rAbs, rAbsErrs, err := endpoint.ListAbstractions(ctx, endpoint.ListZFSHoldsAndBookmarksQuery{
+			FS:          endpoint.ListZFSHoldsAndBookmarksQueryFilesystemFilter{Filter: sfsAndRfsFilter},
+			JobID:       &ojid,
+			What:        endpoint.AbstractionTypesAll,
+			Concurrency: 1,
+		})
+		require.NoError(ctx, err)
+		require.Len(ctx, rAbsErrs, 0)
+		ctx.Logf("rAbs=%s", rAbs)
+		ctx.Logf("expectedOjidAbstractions=%s", expectedOjidAbstractions)
+		require.Equal(ctx, len(expectedOjidAbstractions), len(rAbs))
+		for _, ea := range expectedOjidAbstractions {
+			ctx.Logf("looking for %s %#v", ea, ea.GetFilesystemVersion())
+			found := false
+			for _, a := range rAbs {
+				eq := endpoint.AbstractionEquals(ea, a)
+				ctx.Logf("comp=%v for %s %#v", eq, a, a.GetFilesystemVersion())
+				found = found || eq
+			}
+			require.True(ctx, found, "%s", ea)
+		}
+	}
+	checkOjidAbstractionsExist()
+
+	report = rep.Do(ctx)
+	ctx.Logf("\n%s", pretty.Sprint(report))
+
+	checkOjidAbstractionsExist()
+
+	_ = fsversion(ctx, sfs, "@1")
+	_ = fsversion(ctx, sfs, "@2")
+	_ = fsversion(ctx, sfs, "#2")
+	_ = fsversion(ctx, sfs, "@3")
+	_ = fsversion(ctx, sfs, "@4")
+	_ = fsversion(ctx, sfs, "@5")
+
+	_ = fsversion(ctx, rfs, "@3")
+	_ = fsversion(ctx, rfs, "@4")
+	_ = fsversion(ctx, rfs, "@5")
+
+	// check bookmark situation
+	{
+		sBms, err := zfs.ZFSListFilesystemVersions(ctx, mustDatasetPath(sfs), zfs.ListFilesystemVersionsOptions{
+			Types: zfs.Bookmarks,
+		})
+		ctx.Logf("sbms=%s", sBms)
+		require.NoError(ctx, err)
+
+		snap5SjidCursorName, err := endpoint.ReplicationCursorBookmarkName(sfs, snap5.Guid, sjid)
+		require.NoError(ctx, err)
+		snap2SjidCursorName, err := endpoint.ReplicationCursorBookmarkName(sfs, snap2.Guid, sjid)
+		require.NoError(ctx, err)
+		snap2OjidCursorName, err := endpoint.ReplicationCursorBookmarkName(sfs, snap2.Guid, ojid)
+		require.NoError(ctx, err)
+		var bmNames []string
+		for _, bm := range sBms {
+			bmNames = append(bmNames, bm.Name)
+		}
+
+		if invalidateCacheBeforeSecondReplication {
+			require.Len(ctx, sBms, 3)
+			require.Contains(ctx, bmNames, snap5SjidCursorName)
+			require.Contains(ctx, bmNames, snap2OjidCursorName)
+			require.Contains(ctx, bmNames, "2")
+		} else {
+			require.Len(ctx, sBms, 4)
+			require.Contains(ctx, bmNames, snap5SjidCursorName)
+			require.Contains(ctx, bmNames, snap2SjidCursorName)
+			require.Contains(ctx, bmNames, snap2OjidCursorName)
+			require.Contains(ctx, bmNames, "2")
+		}
+	}
+
+	// check last-received hold moved
+	{
+		rAbs, rAbsErrs, err := endpoint.ListAbstractions(ctx, endpoint.ListZFSHoldsAndBookmarksQuery{
+			FS:          endpoint.ListZFSHoldsAndBookmarksQueryFilesystemFilter{FS: &rfs},
+			JobID:       &rjid,
+			What:        endpoint.AbstractionTypesAll,
+			Concurrency: 1,
+		})
+		require.NoError(ctx, err)
+		require.Len(ctx, rAbsErrs, 0)
+		require.Len(ctx, rAbs, 1)
+		require.Equal(ctx, rAbs[0].GetType(), endpoint.AbstractionLastReceivedHold)
+		require.Equal(ctx, *rAbs[0].GetJobID(), rjid)
+		require.Equal(ctx, rAbs[0].GetFilesystemVersion().GetGuid(), snap5.GetGuid())
+	}
+
+}
+
+type PartialSender struct {
+	*endpoint.Sender
+	failAfterByteCount int64
+}
+
+var _ logic.Sender = (*PartialSender)(nil)
+
+func (s *PartialSender) Send(ctx context.Context, r *pdu.SendReq) (r1 *pdu.SendRes, r2 io.ReadCloser, r3 error) {
+	r1, r2, r3 = s.Sender.Send(ctx, r)
+	r2 = limitio.ReadCloser(r2, s.failAfterByteCount)
+	return r1, r2, r3
+}
+
+func ReplicationIsResumableFullSend__DisableIncrementalStepHolds_False(ctx *platformtest.Context) {
+	implReplicationIsResumableFullSend(ctx, false)
+}
+
+func ReplicationIsResumableFullSend__DisableIncrementalStepHolds_True(ctx *platformtest.Context) {
+	implReplicationIsResumableFullSend(ctx, true)
+}
+
+func implReplicationIsResumableFullSend(ctx *platformtest.Context, disableIncrementalStepHolds bool) {
+
+	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+		CREATEROOT
+		+  "sender"
+		+  "receiver"
+		R  zfs create -p "${ROOTDS}/receiver/${ROOTDS}"
+	`)
+
+	sjid := endpoint.MustMakeJobID("sender-job")
+	rjid := endpoint.MustMakeJobID("receiver-job")
+
+	sfs := ctx.RootDataset + "/sender"
+	rfsRoot := ctx.RootDataset + "/receiver"
+
+	sfsmp, err := zfs.ZFSGetMountpoint(ctx, sfs)
+	require.NoError(ctx, err)
+	require.True(ctx, sfsmp.Mounted)
+
+	writeDummyData(path.Join(sfsmp.Mountpoint, "dummy.data"), 1<<22)
+	mustSnapshot(ctx, sfs+"@1")
+	snap1 := fsversion(ctx, sfs, "@1")
+
+	rep := replicationInvocation{
+		sjid:    sjid,
+		rjid:    rjid,
+		sfs:     sfs,
+		rfsRoot: rfsRoot,
+		interceptSender: func(e *endpoint.Sender) logic.Sender {
+			return &PartialSender{Sender: e, failAfterByteCount: 1 << 20}
+		},
+		disableIncrementalStepHolds: disableIncrementalStepHolds,
+	}
+	rfs := rep.ReceiveSideFilesystem()
+
+	for i := 2; i < 10; i++ {
+		report := rep.Do(ctx)
+		ctx.Logf("\n%s", pretty.Sprint(report))
+
+		// always attempt to destroy the incremental source
+		err := zfs.ZFSDestroy(ctx, snap1.FullPath(sfs))
+		if i < 4 {
+			// we configured the PartialSender to fail after 1<<20 bytes
+			// and we wrote dummy data 1<<22 bytes, thus at least
+			// for the first 4 times this should not be possible
+			// due to step holds
+			require.Error(ctx, err)
+			require.Contains(ctx, err.Error(), "dataset is busy")
+		}
+
+		// and create some additional snapshots that could
+		// confuse a naive implementation that doesn't take into
+		// account resume state when planning replication
+		if i == 2 || i == 3 {
+			// no significant size to avoid making this test run longer than necessary
+			mustSnapshot(ctx, fmt.Sprintf("%s@%d", sfs, i))
+		}
+
+		require.Len(ctx, report.Attempts, 1)
+		require.Nil(ctx, report.Attempts[0].PlanError)
+		require.Len(ctx, report.Attempts[0].Filesystems, 1)
+		if len(report.Attempts[0].Filesystems[0].Steps) == 0 {
+			break
+		}
+	}
+
+	// make sure all the filesystem versions we created
+	// were replicated by the replication loop
+	_ = fsversion(ctx, rfs, "@1")
+	_ = fsversion(ctx, rfs, "@2")
+	_ = fsversion(ctx, rfs, "@3")
+
+}

platformtest/tests/replicationCursor.go

@@ -11,7 +11,7 @@ import (
 	"github.com/zrepl/zrepl/zfs"
 )
 
-func ReplicationCursor(ctx *platformtest.Context) {
+func CreateReplicationCursor(ctx *platformtest.Context) {
 
 	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
 		CREATEROOT
@@ -19,9 +19,9 @@ func ReplicationCursor(ctx *platformtest.Context) {
 		+  "foo bar@1 with space"
 		R  zfs bookmark "${ROOTDS}/foo bar@1 with space" "${ROOTDS}/foo bar#1 with space"
 		+  "foo bar@2 with space"
-		R  zfs bookmark "${ROOTDS}/foo bar@1 with space" "${ROOTDS}/foo bar#2 with space"
+		R  zfs bookmark "${ROOTDS}/foo bar@2 with space" "${ROOTDS}/foo bar#2 with space"
 		+  "foo bar@3 with space"
-		R  zfs bookmark "${ROOTDS}/foo bar@1 with space" "${ROOTDS}/foo bar#3 with space"
+		R  zfs bookmark "${ROOTDS}/foo bar@3 with space" "${ROOTDS}/foo bar#3 with space"
 	`)
 
 	jobid := endpoint.MustMakeJobID("zreplplatformtest")
@@ -32,13 +32,29 @@ func ReplicationCursor(ctx *platformtest.Context) {
 	}
 
 	fs := ds.ToString()
-	snap := fsversion(ctx, fs, "@1 with space")
 
-	destroyed, err := endpoint.MoveReplicationCursor(ctx, fs, &snap, jobid)
+	checkCreateCursor := func(createErr error, c endpoint.Abstraction, references zfs.FilesystemVersion) {
-	if err != nil {
+		assert.NoError(ctx, createErr)
-		panic(err)
+		expectName, err := endpoint.ReplicationCursorBookmarkName(fs, references.Guid, jobid)
+		assert.NoError(ctx, err)
+		require.Equal(ctx, expectName, c.GetFilesystemVersion().Name)
 	}
-	assert.Empty(ctx, destroyed)
+
+	snap := fsversion(ctx, fs, "@1 with space")
+	book := fsversion(ctx, fs, "#1 with space")
+
+	// create first cursor
+	cursorOfSnap, err := endpoint.CreateReplicationCursor(ctx, fs, snap, jobid)
+	checkCreateCursor(err, cursorOfSnap, snap)
+	// check CreateReplicationCursor is idempotent (for snapshot target)
+	cursorOfSnapIdemp, err := endpoint.CreateReplicationCursor(ctx, fs, snap, jobid)
+	checkCreateCursor(err, cursorOfSnap, snap)
+	// ... for target = non-cursor bookmark
+	_, err = endpoint.CreateReplicationCursor(ctx, fs, book, jobid)
+	assert.Equal(ctx, zfs.ErrBookmarkCloningNotSupported, err)
+	// ... for target = replication cursor bookmark to be created
+	cursorOfCursor, err := endpoint.CreateReplicationCursor(ctx, fs, cursorOfSnapIdemp.GetFilesystemVersion(), jobid)
+	checkCreateCursor(err, cursorOfCursor, cursorOfSnap.GetFilesystemVersion())
 
 	snapProps, err := zfs.ZFSGetFilesystemVersion(ctx, snap.FullPath(fs))
 	if err != nil {
@@ -56,14 +72,4 @@ func ReplicationCursor(ctx *platformtest.Context) {
 		panic(fmt.Sprintf("guids do not match: %v != %v", bm.Guid, snapProps.Guid))
 	}
 
-	// try moving
-	cursor1BookmarkName, err := endpoint.ReplicationCursorBookmarkName(fs, snap.Guid, jobid)
-	require.NoError(ctx, err)
-
-	snap2 := fsversion(ctx, fs, "@2 with space")
-	destroyed, err = endpoint.MoveReplicationCursor(ctx, fs, &snap2, jobid)
-	require.NoError(ctx, err)
-	require.Equal(ctx, 1, len(destroyed))
-	require.Equal(ctx, endpoint.AbstractionReplicationCursorBookmarkV2, destroyed[0].GetType())
-	require.Equal(ctx, cursor1BookmarkName, destroyed[0].GetName())
 }

platformtest/tests/tests.go

@@ -13,22 +13,4 @@ func (c Case) String() string {
 	return runtime.FuncForPC(reflect.ValueOf(c).Pointer()).Name()
 }
 
-var Cases = []Case{
+//go:generate ../../artifacts/generate-platform-test-list github.com/zrepl/zrepl/platformtest/tests
-	BatchDestroy,
-	UndestroyableSnapshotParsing,
-	GetNonexistent,
-	ReplicationCursor,
-	IdempotentHold,
-	IdempotentBookmark,
-	IdempotentDestroy,
-	ResumeTokenParsing,
-	ResumableRecvAndTokenHandling,
-	SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden,
-	SendArgsValidationResumeTokenEncryptionMismatchForbidden,
-	SendArgsValidationResumeTokenDifferentFilesystemForbidden,
-	ListFilesystemVersionsTypeFilteringAndPrefix,
-	ListFilesystemVersionsFilesystemNotExist,
-	ListFilesystemVersionsFilesystemNotExist,
-	ListFilesystemVersionsUserrefs,
-	ListFilesystemsNoFilter,
-}

replication/logic/pdu/pdu.pb.go

@@ -46,7 +46,7 @@ func (x Tri) String() string {
 	return proto.EnumName(Tri_name, int32(x))
 }
 func (Tri) EnumDescriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{0}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{0}
 }
 
 type FilesystemVersion_VersionType int32
@@ -69,7 +69,7 @@ func (x FilesystemVersion_VersionType) String() string {
 	return proto.EnumName(FilesystemVersion_VersionType_name, int32(x))
 }
 func (FilesystemVersion_VersionType) EnumDescriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{5, 0}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{5, 0}
 }
 
 type ListFilesystemReq struct {
@@ -82,7 +82,7 @@ func (m *ListFilesystemReq) Reset()         { *m = ListFilesystemReq{} }
 func (m *ListFilesystemReq) String() string { return proto.CompactTextString(m) }
 func (*ListFilesystemReq) ProtoMessage()    {}
 func (*ListFilesystemReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{0}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{0}
 }
 func (m *ListFilesystemReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ListFilesystemReq.Unmarshal(m, b)
@@ -113,7 +113,7 @@ func (m *ListFilesystemRes) Reset()         { *m = ListFilesystemRes{} }
 func (m *ListFilesystemRes) String() string { return proto.CompactTextString(m) }
 func (*ListFilesystemRes) ProtoMessage()    {}
 func (*ListFilesystemRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{1}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{1}
 }
 func (m *ListFilesystemRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ListFilesystemRes.Unmarshal(m, b)
@@ -154,7 +154,7 @@ func (m *Filesystem) Reset()         { *m = Filesystem{} }
 func (m *Filesystem) String() string { return proto.CompactTextString(m) }
 func (*Filesystem) ProtoMessage()    {}
 func (*Filesystem) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{2}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{2}
 }
 func (m *Filesystem) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_Filesystem.Unmarshal(m, b)
@@ -213,7 +213,7 @@ func (m *ListFilesystemVersionsReq) Reset()         { *m = ListFilesystemVersion
 func (m *ListFilesystemVersionsReq) String() string { return proto.CompactTextString(m) }
 func (*ListFilesystemVersionsReq) ProtoMessage()    {}
 func (*ListFilesystemVersionsReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{3}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{3}
 }
 func (m *ListFilesystemVersionsReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ListFilesystemVersionsReq.Unmarshal(m, b)
@@ -251,7 +251,7 @@ func (m *ListFilesystemVersionsRes) Reset()         { *m = ListFilesystemVersion
 func (m *ListFilesystemVersionsRes) String() string { return proto.CompactTextString(m) }
 func (*ListFilesystemVersionsRes) ProtoMessage()    {}
 func (*ListFilesystemVersionsRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{4}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{4}
 }
 func (m *ListFilesystemVersionsRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ListFilesystemVersionsRes.Unmarshal(m, b)
@@ -293,7 +293,7 @@ func (m *FilesystemVersion) Reset()         { *m = FilesystemVersion{} }
 func (m *FilesystemVersion) String() string { return proto.CompactTextString(m) }
 func (*FilesystemVersion) ProtoMessage()    {}
 func (*FilesystemVersion) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{5}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{5}
 }
 func (m *FilesystemVersion) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_FilesystemVersion.Unmarshal(m, b)
@@ -371,7 +371,7 @@ func (m *SendReq) Reset()         { *m = SendReq{} }
 func (m *SendReq) String() string { return proto.CompactTextString(m) }
 func (*SendReq) ProtoMessage()    {}
 func (*SendReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{6}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{6}
 }
 func (m *SendReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_SendReq.Unmarshal(m, b)
@@ -445,7 +445,7 @@ func (m *Property) Reset()         { *m = Property{} }
 func (m *Property) String() string { return proto.CompactTextString(m) }
 func (*Property) ProtoMessage()    {}
 func (*Property) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{7}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{7}
 }
 func (m *Property) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_Property.Unmarshal(m, b)
@@ -496,7 +496,7 @@ func (m *SendRes) Reset()         { *m = SendRes{} }
 func (m *SendRes) String() string { return proto.CompactTextString(m) }
 func (*SendRes) ProtoMessage()    {}
 func (*SendRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{8}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{8}
 }
 func (m *SendRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_SendRes.Unmarshal(m, b)
@@ -548,7 +548,7 @@ func (m *SendCompletedReq) Reset()         { *m = SendCompletedReq{} }
 func (m *SendCompletedReq) String() string { return proto.CompactTextString(m) }
 func (*SendCompletedReq) ProtoMessage()    {}
 func (*SendCompletedReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{9}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{9}
 }
 func (m *SendCompletedReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_SendCompletedReq.Unmarshal(m, b)
@@ -585,7 +585,7 @@ func (m *SendCompletedRes) Reset()         { *m = SendCompletedRes{} }
 func (m *SendCompletedRes) String() string { return proto.CompactTextString(m) }
 func (*SendCompletedRes) ProtoMessage()    {}
 func (*SendCompletedRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{10}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{10}
 }
 func (m *SendCompletedRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_SendCompletedRes.Unmarshal(m, b)
@@ -620,7 +620,7 @@ func (m *ReceiveReq) Reset()         { *m = ReceiveReq{} }
 func (m *ReceiveReq) String() string { return proto.CompactTextString(m) }
 func (*ReceiveReq) ProtoMessage()    {}
 func (*ReceiveReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{11}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{11}
 }
 func (m *ReceiveReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ReceiveReq.Unmarshal(m, b)
@@ -671,7 +671,7 @@ func (m *ReceiveRes) Reset()         { *m = ReceiveRes{} }
 func (m *ReceiveRes) String() string { return proto.CompactTextString(m) }
 func (*ReceiveRes) ProtoMessage()    {}
 func (*ReceiveRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{12}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{12}
 }
 func (m *ReceiveRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ReceiveRes.Unmarshal(m, b)
@@ -704,7 +704,7 @@ func (m *DestroySnapshotsReq) Reset()         { *m = DestroySnapshotsReq{} }
 func (m *DestroySnapshotsReq) String() string { return proto.CompactTextString(m) }
 func (*DestroySnapshotsReq) ProtoMessage()    {}
 func (*DestroySnapshotsReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{13}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{13}
 }
 func (m *DestroySnapshotsReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_DestroySnapshotsReq.Unmarshal(m, b)
@@ -750,7 +750,7 @@ func (m *DestroySnapshotRes) Reset()         { *m = DestroySnapshotRes{} }
 func (m *DestroySnapshotRes) String() string { return proto.CompactTextString(m) }
 func (*DestroySnapshotRes) ProtoMessage()    {}
 func (*DestroySnapshotRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{14}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{14}
 }
 func (m *DestroySnapshotRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_DestroySnapshotRes.Unmarshal(m, b)
@@ -795,7 +795,7 @@ func (m *DestroySnapshotsRes) Reset()         { *m = DestroySnapshotsRes{} }
 func (m *DestroySnapshotsRes) String() string { return proto.CompactTextString(m) }
 func (*DestroySnapshotsRes) ProtoMessage()    {}
 func (*DestroySnapshotsRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{15}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{15}
 }
 func (m *DestroySnapshotsRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_DestroySnapshotsRes.Unmarshal(m, b)
@@ -833,7 +833,7 @@ func (m *ReplicationCursorReq) Reset()         { *m = ReplicationCursorReq{} }
 func (m *ReplicationCursorReq) String() string { return proto.CompactTextString(m) }
 func (*ReplicationCursorReq) ProtoMessage()    {}
 func (*ReplicationCursorReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{16}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{16}
 }
 func (m *ReplicationCursorReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ReplicationCursorReq.Unmarshal(m, b)
@@ -874,7 +874,7 @@ func (m *ReplicationCursorRes) Reset()         { *m = ReplicationCursorRes{} }
 func (m *ReplicationCursorRes) String() string { return proto.CompactTextString(m) }
 func (*ReplicationCursorRes) ProtoMessage()    {}
 func (*ReplicationCursorRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{17}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{17}
 }
 func (m *ReplicationCursorRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ReplicationCursorRes.Unmarshal(m, b)
@@ -1010,7 +1010,7 @@ func (m *PingReq) Reset()         { *m = PingReq{} }
 func (m *PingReq) String() string { return proto.CompactTextString(m) }
 func (*PingReq) ProtoMessage()    {}
 func (*PingReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{18}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{18}
 }
 func (m *PingReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_PingReq.Unmarshal(m, b)
@@ -1049,7 +1049,7 @@ func (m *PingRes) Reset()         { *m = PingRes{} }
 func (m *PingRes) String() string { return proto.CompactTextString(m) }
 func (*PingRes) ProtoMessage()    {}
 func (*PingRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{19}
+	return fileDescriptor_pdu_483c6918b7b3d747, []int{19}
 }
 func (m *PingRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_PingRes.Unmarshal(m, b)
@@ -1076,91 +1076,6 @@ func (m *PingRes) GetEcho() string {
 	return ""
 }
 
-type HintMostRecentCommonAncestorReq struct {
-	Filesystem string `protobuf:"bytes,1,opt,name=Filesystem,proto3" json:"Filesystem,omitempty"`
-	// A copy of the FilesystemVersion on the sending side that the replication
-	// algorithm identified as a shared most recent common version between sending
-	// and receiving side.
-	//
-	// If nil, this is an indication that the replication algorithm could not
-	// find a common ancestor between the two sides.
-	// NOTE: nilness does not mean that replication never happened - there could
-	// as well be a replication conflict. thus, dont' jump to conclusions too
-	// rapidly here.
-	SenderVersion        *FilesystemVersion `protobuf:"bytes,2,opt,name=SenderVersion,proto3" json:"SenderVersion,omitempty"`
-	XXX_NoUnkeyedLiteral struct{}           `json:"-"`
-	XXX_unrecognized     []byte             `json:"-"`
-	XXX_sizecache        int32              `json:"-"`
-}
-
-func (m *HintMostRecentCommonAncestorReq) Reset()         { *m = HintMostRecentCommonAncestorReq{} }
-func (m *HintMostRecentCommonAncestorReq) String() string { return proto.CompactTextString(m) }
-func (*HintMostRecentCommonAncestorReq) ProtoMessage()    {}
-func (*HintMostRecentCommonAncestorReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{20}
-}
-func (m *HintMostRecentCommonAncestorReq) XXX_Unmarshal(b []byte) error {
-	return xxx_messageInfo_HintMostRecentCommonAncestorReq.Unmarshal(m, b)
-}
-func (m *HintMostRecentCommonAncestorReq) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
-	return xxx_messageInfo_HintMostRecentCommonAncestorReq.Marshal(b, m, deterministic)
-}
-func (dst *HintMostRecentCommonAncestorReq) XXX_Merge(src proto.Message) {
-	xxx_messageInfo_HintMostRecentCommonAncestorReq.Merge(dst, src)
-}
-func (m *HintMostRecentCommonAncestorReq) XXX_Size() int {
-	return xxx_messageInfo_HintMostRecentCommonAncestorReq.Size(m)
-}
-func (m *HintMostRecentCommonAncestorReq) XXX_DiscardUnknown() {
-	xxx_messageInfo_HintMostRecentCommonAncestorReq.DiscardUnknown(m)
-}
-
-var xxx_messageInfo_HintMostRecentCommonAncestorReq proto.InternalMessageInfo
-
-func (m *HintMostRecentCommonAncestorReq) GetFilesystem() string {
-	if m != nil {
-		return m.Filesystem
-	}
-	return ""
-}
-
-func (m *HintMostRecentCommonAncestorReq) GetSenderVersion() *FilesystemVersion {
-	if m != nil {
-		return m.SenderVersion
-	}
-	return nil
-}
-
-type HintMostRecentCommonAncestorRes struct {
-	XXX_NoUnkeyedLiteral struct{} `json:"-"`
-	XXX_unrecognized     []byte   `json:"-"`
-	XXX_sizecache        int32    `json:"-"`
-}
-
-func (m *HintMostRecentCommonAncestorRes) Reset()         { *m = HintMostRecentCommonAncestorRes{} }
-func (m *HintMostRecentCommonAncestorRes) String() string { return proto.CompactTextString(m) }
-func (*HintMostRecentCommonAncestorRes) ProtoMessage()    {}
-func (*HintMostRecentCommonAncestorRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{21}
-}
-func (m *HintMostRecentCommonAncestorRes) XXX_Unmarshal(b []byte) error {
-	return xxx_messageInfo_HintMostRecentCommonAncestorRes.Unmarshal(m, b)
-}
-func (m *HintMostRecentCommonAncestorRes) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
-	return xxx_messageInfo_HintMostRecentCommonAncestorRes.Marshal(b, m, deterministic)
-}
-func (dst *HintMostRecentCommonAncestorRes) XXX_Merge(src proto.Message) {
-	xxx_messageInfo_HintMostRecentCommonAncestorRes.Merge(dst, src)
-}
-func (m *HintMostRecentCommonAncestorRes) XXX_Size() int {
-	return xxx_messageInfo_HintMostRecentCommonAncestorRes.Size(m)
-}
-func (m *HintMostRecentCommonAncestorRes) XXX_DiscardUnknown() {
-	xxx_messageInfo_HintMostRecentCommonAncestorRes.DiscardUnknown(m)
-}
-
-var xxx_messageInfo_HintMostRecentCommonAncestorRes proto.InternalMessageInfo
-
 func init() {
 	proto.RegisterType((*ListFilesystemReq)(nil), "ListFilesystemReq")
 	proto.RegisterType((*ListFilesystemRes)(nil), "ListFilesystemRes")
@@ -1182,8 +1097,6 @@ func init() {
 	proto.RegisterType((*ReplicationCursorRes)(nil), "ReplicationCursorRes")
 	proto.RegisterType((*PingReq)(nil), "PingReq")
 	proto.RegisterType((*PingRes)(nil), "PingRes")
-	proto.RegisterType((*HintMostRecentCommonAncestorReq)(nil), "HintMostRecentCommonAncestorReq")
-	proto.RegisterType((*HintMostRecentCommonAncestorRes)(nil), "HintMostRecentCommonAncestorRes")
 	proto.RegisterEnum("Tri", Tri_name, Tri_value)
 	proto.RegisterEnum("FilesystemVersion_VersionType", FilesystemVersion_VersionType_name, FilesystemVersion_VersionType_value)
 }
@@ -1206,7 +1119,6 @@ type ReplicationClient interface {
 	DestroySnapshots(ctx context.Context, in *DestroySnapshotsReq, opts ...grpc.CallOption) (*DestroySnapshotsRes, error)
 	ReplicationCursor(ctx context.Context, in *ReplicationCursorReq, opts ...grpc.CallOption) (*ReplicationCursorRes, error)
 	SendCompleted(ctx context.Context, in *SendCompletedReq, opts ...grpc.CallOption) (*SendCompletedRes, error)
-	HintMostRecentCommonAncestor(ctx context.Context, in *HintMostRecentCommonAncestorReq, opts ...grpc.CallOption) (*HintMostRecentCommonAncestorRes, error)
 }
 
 type replicationClient struct {
@@ -1271,15 +1183,6 @@ func (c *replicationClient) SendCompleted(ctx context.Context, in *SendCompleted
 	return out, nil
 }
 
-func (c *replicationClient) HintMostRecentCommonAncestor(ctx context.Context, in *HintMostRecentCommonAncestorReq, opts ...grpc.CallOption) (*HintMostRecentCommonAncestorRes, error) {
-	out := new(HintMostRecentCommonAncestorRes)
-	err := c.cc.Invoke(ctx, "/Replication/HintMostRecentCommonAncestor", in, out, opts...)
-	if err != nil {
-		return nil, err
-	}
-	return out, nil
-}
-
 // ReplicationServer is the server API for Replication service.
 type ReplicationServer interface {
 	Ping(context.Context, *PingReq) (*PingRes, error)
@@ -1288,7 +1191,6 @@ type ReplicationServer interface {
 	DestroySnapshots(context.Context, *DestroySnapshotsReq) (*DestroySnapshotsRes, error)
 	ReplicationCursor(context.Context, *ReplicationCursorReq) (*ReplicationCursorRes, error)
 	SendCompleted(context.Context, *SendCompletedReq) (*SendCompletedRes, error)
-	HintMostRecentCommonAncestor(context.Context, *HintMostRecentCommonAncestorReq) (*HintMostRecentCommonAncestorRes, error)
 }
 
 func RegisterReplicationServer(s *grpc.Server, srv ReplicationServer) {
@@ -1403,24 +1305,6 @@ func _Replication_SendCompleted_Handler(srv interface{}, ctx context.Context, de
 	return interceptor(ctx, in, info, handler)
 }
 
-func _Replication_HintMostRecentCommonAncestor_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := new(HintMostRecentCommonAncestorReq)
-	if err := dec(in); err != nil {
-		return nil, err
-	}
-	if interceptor == nil {
-		return srv.(ReplicationServer).HintMostRecentCommonAncestor(ctx, in)
-	}
-	info := &grpc.UnaryServerInfo{
-		Server:     srv,
-		FullMethod: "/Replication/HintMostRecentCommonAncestor",
-	}
-	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(ReplicationServer).HintMostRecentCommonAncestor(ctx, req.(*HintMostRecentCommonAncestorReq))
-	}
-	return interceptor(ctx, in, info, handler)
-}
-
 var _Replication_serviceDesc = grpc.ServiceDesc{
 	ServiceName: "Replication",
 	HandlerType: (*ReplicationServer)(nil),
@@ -1449,73 +1333,66 @@ var _Replication_serviceDesc = grpc.ServiceDesc{
 			MethodName: "SendCompleted",
 			Handler:    _Replication_SendCompleted_Handler,
 		},
-		{
-			MethodName: "HintMostRecentCommonAncestor",
-			Handler:    _Replication_HintMostRecentCommonAncestor_Handler,
-		},
 	},
 	Streams:  []grpc.StreamDesc{},
 	Metadata: "pdu.proto",
 }
 
-func init() { proto.RegisterFile("pdu.proto", fileDescriptor_pdu_e59763dc61674a79) }
+func init() { proto.RegisterFile("pdu.proto", fileDescriptor_pdu_483c6918b7b3d747) }
 
-var fileDescriptor_pdu_e59763dc61674a79 = []byte{
+var fileDescriptor_pdu_483c6918b7b3d747 = []byte{
-	// 892 bytes of a gzipped FileDescriptorProto
+	// 833 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x56, 0xdf, 0x6f, 0xdb, 0x36,
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x56, 0x5f, 0x6f, 0xe3, 0x44,
-	0x10, 0x8e, 0x6c, 0x39, 0x91, 0xcf, 0xe9, 0xea, 0x5c, 0xb2, 0x42, 0x13, 0xba, 0xce, 0xe3, 0x86,
+	0x10, 0xaf, 0x13, 0xa7, 0x75, 0x26, 0x3d, 0x2e, 0x9d, 0x96, 0x93, 0xb1, 0xe0, 0x54, 0x2d, 0x08,
-	0xc1, 0x0d, 0x30, 0x61, 0xc8, 0x7e, 0x60, 0xc3, 0x80, 0x02, 0x8d, 0x93, 0x34, 0xc5, 0xd6, 0xce,
+	0xe5, 0x2a, 0x61, 0xa1, 0xf2, 0x47, 0x42, 0x48, 0x27, 0xd1, 0xb4, 0xbd, 0x3b, 0x01, 0x47, 0xb4,
-	0x60, 0xbc, 0x62, 0xe8, 0x9b, 0x6a, 0x1f, 0x12, 0x21, 0xb2, 0xa8, 0x90, 0xf4, 0x50, 0x6f, 0x7b,
+	0x35, 0x27, 0x74, 0x6f, 0x26, 0x19, 0xb5, 0x56, 0x1d, 0xaf, 0xbb, 0xe3, 0xa0, 0x0b, 0xe2, 0x89,
-	0xda, 0xe3, 0xfe, 0xbd, 0xe5, 0x0f, 0x2a, 0x44, 0x4b, 0xb6, 0x6c, 0xc9, 0x89, 0x9f, 0xcc, 0xfb,
+	0x47, 0xbe, 0x1e, 0x7c, 0x10, 0x3e, 0x02, 0xf2, 0xc6, 0x4e, 0x9c, 0xd8, 0x41, 0x79, 0xca, 0xce,
-	0x78, 0x14, 0xef, 0xbe, 0xfb, 0xee, 0x68, 0x68, 0x26, 0xa3, 0x89, 0x9f, 0x48, 0xa1, 0x05, 0xdb,
+	0x6f, 0x66, 0x77, 0x67, 0x7f, 0xf3, 0x9b, 0x71, 0xa0, 0x9b, 0x4e, 0x66, 0x7e, 0xaa, 0x55, 0xa6,
-	0x87, 0xbd, 0x5f, 0x43, 0xa5, 0xcf, 0xc2, 0x88, 0xd4, 0x54, 0x69, 0x1a, 0x73, 0xba, 0x61, 0xc7,
+	0xc4, 0x31, 0x1c, 0xfd, 0x10, 0x71, 0x76, 0x1d, 0xc5, 0xc4, 0x73, 0xce, 0x68, 0x2a, 0xe9, 0x41,
-	0x65, 0x50, 0xe1, 0xd7, 0xd0, 0x5a, 0x00, 0xca, 0xb5, 0x3a, 0xf5, 0x6e, 0xeb, 0xa8, 0xe5, 0x17,
+	0x5c, 0xd4, 0x41, 0xc6, 0xcf, 0xa0, 0xb7, 0x02, 0xd8, 0xb5, 0x4e, 0xdb, 0x83, 0xde, 0x79, 0xcf,
-	0x9c, 0x8a, 0xfb, 0xec, 0x3f, 0x0b, 0x60, 0x61, 0x23, 0x82, 0xdd, 0x0f, 0xf4, 0x95, 0x6b, 0x75,
+	0xaf, 0x04, 0x55, 0xfd, 0xe2, 0x2f, 0x0b, 0x60, 0x65, 0x23, 0x82, 0x3d, 0x0a, 0xb3, 0x3b, 0xd7,
-	0xac, 0x6e, 0x93, 0x9b, 0x35, 0x76, 0xa0, 0xc5, 0x49, 0x4d, 0xc6, 0x34, 0x10, 0xd7, 0x14, 0xbb,
+	0x3a, 0xb5, 0x06, 0x5d, 0x69, 0xd6, 0x78, 0x0a, 0x3d, 0x49, 0x3c, 0x9b, 0x52, 0xa0, 0xee, 0x29,
-	0x35, 0xb3, 0x55, 0x84, 0xf0, 0x4b, 0x78, 0xf0, 0x52, 0xf5, 0xa3, 0x60, 0x48, 0x57, 0x22, 0x1a,
+	0x71, 0x5b, 0xc6, 0x55, 0x85, 0xf0, 0x13, 0x78, 0xf4, 0x8a, 0x47, 0x71, 0x38, 0xa6, 0x3b, 0x15,
-	0x91, 0x74, 0xeb, 0x1d, 0xab, 0xeb, 0xf0, 0x65, 0x30, 0xfd, 0xce, 0x4b, 0x75, 0x1a, 0x0f, 0xe5,
+	0x4f, 0x48, 0xbb, 0xed, 0x53, 0x6b, 0xe0, 0xc8, 0x75, 0x30, 0x3f, 0xe7, 0x15, 0x5f, 0x25, 0x63,
-	0x34, 0xd1, 0x34, 0x72, 0x6d, 0xe3, 0x53, 0x84, 0xd8, 0xcf, 0xf0, 0xc9, 0x72, 0x42, 0x6f, 0x48,
+	0x3d, 0x4f, 0x33, 0x9a, 0xb8, 0xb6, 0x89, 0xa9, 0x42, 0xe2, 0x5b, 0xf8, 0x60, 0xfd, 0x41, 0x6f,
-	0xaa, 0x50, 0xc4, 0x8a, 0xd3, 0x0d, 0x3e, 0x29, 0x06, 0x9a, 0x05, 0x58, 0x40, 0xd8, 0x2f, 0xeb,
+	0x48, 0x73, 0xa4, 0x12, 0x96, 0xf4, 0x80, 0x4f, 0xab, 0x89, 0x16, 0x09, 0x56, 0x10, 0xf1, 0xfd,
-	0x0f, 0x2b, 0xf4, 0xc1, 0xc9, 0xcd, 0x8c, 0x12, 0xf4, 0x4b, 0x9e, 0x7c, 0xee, 0xc3, 0x6e, 0x2d,
+	0xf6, 0xcd, 0x8c, 0x3e, 0x38, 0xa5, 0x59, 0x50, 0x82, 0x7e, 0x2d, 0x52, 0x2e, 0x63, 0xc4, 0x3f,
-	0xd8, 0x2b, 0xed, 0xe3, 0x11, 0xd8, 0x83, 0x69, 0x42, 0xe6, 0xf2, 0x8f, 0x8e, 0x9e, 0x94, 0xbf,
+	0x16, 0x1c, 0xd5, 0xfc, 0x78, 0x0e, 0x76, 0x30, 0x4f, 0xc9, 0x5c, 0xfe, 0xde, 0xf9, 0xd3, 0xfa,
-	0xe0, 0x67, 0xbf, 0xa9, 0x17, 0x37, 0xbe, 0x29, 0xa3, 0xaf, 0x83, 0x31, 0x65, 0xb4, 0x99, 0x75,
+	0x09, 0x7e, 0xf1, 0x9b, 0x47, 0x49, 0x13, 0x9b, 0x33, 0xfa, 0x3a, 0x9c, 0x52, 0x41, 0x9b, 0x59,
-	0x8a, 0xbd, 0x98, 0x84, 0x23, 0x43, 0x93, 0xcd, 0xcd, 0x1a, 0x1f, 0x43, 0xb3, 0x27, 0x29, 0xd0,
+	0xe7, 0xd8, 0x8b, 0x59, 0x34, 0x31, 0x34, 0xd9, 0xd2, 0xac, 0xf1, 0x43, 0xe8, 0x0e, 0x35, 0x85,
-	0x34, 0xf8, 0xe3, 0x85, 0xe1, 0xc6, 0xe6, 0x0b, 0x00, 0x3d, 0x70, 0x8c, 0x11, 0x8a, 0xd8, 0x6d,
+	0x19, 0x05, 0xbf, 0xbc, 0x30, 0xdc, 0xd8, 0x72, 0x05, 0xa0, 0x07, 0x8e, 0x31, 0x22, 0x95, 0xb8,
-	0x98, 0x2f, 0xcd, 0x6d, 0xf6, 0x14, 0x5a, 0x85, 0x6b, 0x71, 0x17, 0x9c, 0x8b, 0x38, 0x48, 0xd4,
+	0x1d, 0x73, 0xd2, 0xd2, 0x16, 0xcf, 0xa0, 0x57, 0xb9, 0x16, 0x0f, 0xc1, 0xb9, 0x49, 0xc2, 0x94,
-	0x95, 0xd0, 0xed, 0xad, 0xd4, 0x3a, 0x16, 0xe2, 0x7a, 0x1c, 0xc8, 0xeb, 0xb6, 0xc5, 0xfe, 0xb7,
+	0xef, 0x54, 0xd6, 0xdf, 0xcb, 0xad, 0x0b, 0xa5, 0xee, 0xa7, 0xa1, 0xbe, 0xef, 0x5b, 0xe2, 0x6f,
-	0x60, 0xe7, 0x82, 0xe2, 0xd1, 0x06, 0x7c, 0xe2, 0x57, 0x60, 0x9f, 0x49, 0x31, 0x36, 0x81, 0x57,
+	0x0b, 0x0e, 0x6e, 0x28, 0x99, 0xec, 0xc0, 0x27, 0x7e, 0x0a, 0xf6, 0xb5, 0x56, 0x53, 0x93, 0x78,
-	0xd3, 0x65, 0xf6, 0x91, 0x41, 0x6d, 0x20, 0x4c, 0x2a, 0xd5, 0x5e, 0xb5, 0x81, 0x58, 0x95, 0x90,
+	0x33, 0x5d, 0xc6, 0x8f, 0x02, 0x5a, 0x81, 0x32, 0x4f, 0x69, 0x8e, 0x6a, 0x05, 0x6a, 0x53, 0x42,
-	0x5d, 0x96, 0x10, 0x83, 0xe6, 0x42, 0x1a, 0x0d, 0xc3, 0xaf, 0xed, 0x0f, 0x64, 0xc8, 0x17, 0x30,
+	0x76, 0x5d, 0x42, 0x02, 0xba, 0x2b, 0x69, 0x74, 0x0c, 0xbf, 0xb6, 0x1f, 0xe8, 0x48, 0xae, 0x60,
-	0x3e, 0x82, 0xed, 0x13, 0x39, 0xe5, 0x93, 0xd8, 0xdd, 0x36, 0xda, 0xc9, 0x2c, 0xf6, 0x1d, 0x38,
+	0x7c, 0x02, 0xfb, 0x97, 0x7a, 0x2e, 0x67, 0x89, 0xbb, 0x6f, 0xb4, 0x53, 0x58, 0xe2, 0x4b, 0x70,
-	0x7d, 0x29, 0x12, 0x92, 0x7a, 0x3a, 0xa7, 0xdb, 0x2a, 0xd0, 0x7d, 0x00, 0x8d, 0x37, 0x41, 0x34,
+	0x46, 0x5a, 0xa5, 0xa4, 0xb3, 0xf9, 0x92, 0x6e, 0xab, 0x42, 0xf7, 0x09, 0x74, 0xde, 0x84, 0xf1,
-	0xc9, 0x6b, 0x30, 0x33, 0xd8, 0xbf, 0x73, 0x2e, 0x14, 0x76, 0xe1, 0xe1, 0xef, 0x8a, 0x46, 0xab,
+	0xac, 0xac, 0xc1, 0xc2, 0x10, 0x7f, 0x2e, 0xb9, 0x60, 0x1c, 0xc0, 0xe3, 0x9f, 0x99, 0x26, 0x9b,
-	0x32, 0x77, 0xf8, 0x2a, 0x8c, 0x0c, 0x76, 0x4f, 0xdf, 0x27, 0x34, 0xd4, 0x34, 0xba, 0x08, 0xff,
+	0x32, 0x77, 0xe4, 0x26, 0x8c, 0x02, 0x0e, 0xaf, 0xde, 0xa5, 0x34, 0xce, 0x68, 0x72, 0x13, 0xfd,
-	0x22, 0x93, 0x77, 0x9d, 0x2f, 0x61, 0xf8, 0x14, 0x20, 0x8b, 0x27, 0x24, 0xe5, 0xda, 0x46, 0x6e,
+	0x4e, 0xe6, 0xdd, 0x6d, 0xb9, 0x86, 0xe1, 0x33, 0x80, 0x22, 0x9f, 0x88, 0xd8, 0xb5, 0x8d, 0xdc,
-	0x4d, 0x3f, 0x0f, 0x91, 0x17, 0x36, 0xd9, 0x33, 0x68, 0xa7, 0x31, 0xf4, 0xc4, 0x38, 0x89, 0x48,
+	0xba, 0x7e, 0x99, 0xa2, 0xac, 0x38, 0xc5, 0x73, 0xe8, 0xe7, 0x39, 0x0c, 0xd5, 0x34, 0x8d, 0x29,
-	0x93, 0x29, 0xcc, 0x21, 0xb4, 0x7e, 0x93, 0xe1, 0x65, 0x18, 0x07, 0x11, 0xa7, 0x9b, 0x8c, 0x7f,
+	0x23, 0x53, 0x98, 0x33, 0xe8, 0xfd, 0xa4, 0xa3, 0xdb, 0x28, 0x09, 0x63, 0x49, 0x0f, 0x05, 0xff,
-	0xc7, 0xcf, 0xea, 0xc6, 0x8b, 0x9b, 0x0c, 0x4b, 0xe7, 0x15, 0xfb, 0x07, 0x80, 0xd3, 0x90, 0xc2,
+	0x8e, 0x5f, 0xd4, 0x4d, 0x56, 0x9d, 0x02, 0x6b, 0xfb, 0x59, 0xfc, 0x01, 0x20, 0x69, 0x4c, 0xd1,
-	0x3f, 0x69, 0x93, 0x32, 0xcf, 0xca, 0x57, 0xbb, 0xb3, 0x7c, 0x87, 0xd0, 0xee, 0x45, 0x14, 0xc8,
+	0x6f, 0xb4, 0x4b, 0x99, 0x17, 0xe5, 0x6b, 0xfd, 0x6f, 0xf9, 0xce, 0xa0, 0x3f, 0x8c, 0x29, 0xd4,
-	0x22, 0x3f, 0xb3, 0x16, 0x2f, 0xe1, 0x6c, 0xb7, 0x70, 0xbb, 0x62, 0x97, 0xb0, 0x7f, 0x42, 0x4a,
+	0x55, 0x7e, 0x16, 0x2d, 0x5e, 0xc3, 0xc5, 0x61, 0xe5, 0x76, 0x16, 0xb7, 0x70, 0x7c, 0x49, 0x9c,
-	0x4b, 0x31, 0xcd, 0x35, 0xb9, 0x49, 0x2f, 0xe3, 0x37, 0xd0, 0x9c, 0xfb, 0xbb, 0xb5, 0xb5, 0xfd,
+	0x69, 0x35, 0x2f, 0x35, 0xb9, 0x4b, 0x2f, 0xe3, 0xe7, 0xd0, 0x5d, 0xc6, 0xbb, 0xad, 0xad, 0xfd,
-	0xba, 0x70, 0x62, 0x6f, 0x01, 0x57, 0x2e, 0xca, 0xda, 0x3e, 0x37, 0xcd, 0x2d, 0x6b, 0xda, 0x3e,
+	0xba, 0x0a, 0x12, 0x6f, 0x01, 0x37, 0x2e, 0x2a, 0xda, 0xbe, 0x34, 0xcd, 0x2d, 0x5b, 0xda, 0xbe,
-	0xf7, 0x49, 0x95, 0x72, 0x2a, 0xa5, 0x90, 0xb9, 0x52, 0x8c, 0xc1, 0x4e, 0xaa, 0x92, 0x48, 0x27,
+	0x8c, 0xc9, 0x95, 0x72, 0xa5, 0xb5, 0xd2, 0xa5, 0x52, 0x8c, 0x21, 0x2e, 0x9b, 0x1e, 0x91, 0x4f,
-	0xed, 0x4e, 0x9a, 0x78, 0xa4, 0xf3, 0x91, 0xb2, 0xef, 0x97, 0x43, 0xe0, 0xb9, 0x0f, 0xfb, 0x01,
+	0xda, 0x83, 0xfc, 0xe1, 0x71, 0x56, 0x8e, 0x94, 0x63, 0xbf, 0x9e, 0x82, 0x2c, 0x63, 0xc4, 0xd7,
-	0x0e, 0x38, 0x25, 0x51, 0x38, 0x34, 0x5d, 0xdb, 0x9b, 0x48, 0x25, 0xe4, 0x26, 0x73, 0x6d, 0x50,
+	0x70, 0x22, 0x29, 0x8d, 0xa3, 0xb1, 0xe9, 0xda, 0xe1, 0x4c, 0xb3, 0xd2, 0xbb, 0xcc, 0xb5, 0xa0,
-	0x79, 0x4e, 0xe1, 0x41, 0x36, 0x44, 0xd2, 0x13, 0xf6, 0xf9, 0xd6, 0x7c, 0x8c, 0x38, 0xaf, 0x85,
+	0x71, 0x1f, 0xe3, 0x49, 0x31, 0x44, 0xf2, 0x1d, 0xf6, 0xcb, 0xbd, 0xe5, 0x18, 0x71, 0x5e, 0xab,
-	0xa6, 0xf7, 0xa1, 0xd2, 0x33, 0x09, 0x9f, 0x6f, 0xf1, 0x39, 0x72, 0xec, 0xc0, 0xf6, 0x2c, 0x1c,
+	0x8c, 0xde, 0x45, 0x9c, 0x2d, 0x24, 0xfc, 0x72, 0x4f, 0x2e, 0x91, 0x0b, 0x07, 0xf6, 0x17, 0xe9,
-	0xf6, 0x05, 0xec, 0xf4, 0xc3, 0xf8, 0x32, 0x0d, 0xc0, 0x85, 0x9d, 0x57, 0xa4, 0x54, 0x70, 0x99,
+	0x88, 0x8f, 0xe1, 0x60, 0x14, 0x25, 0xb7, 0x79, 0x02, 0x2e, 0x1c, 0xfc, 0x48, 0xcc, 0xe1, 0x6d,
-	0x77, 0x4d, 0x6e, 0xb2, 0x4f, 0x73, 0x27, 0x95, 0xf6, 0xd5, 0xe9, 0xf0, 0x4a, 0xe4, 0x7d, 0x95,
+	0xd9, 0x35, 0xa5, 0x29, 0x3e, 0x2a, 0x83, 0x38, 0xef, 0xab, 0xab, 0xf1, 0x9d, 0x2a, 0xfb, 0x2a,
-	0xae, 0xd9, 0xdf, 0xf0, 0xd9, 0x79, 0x18, 0xeb, 0x57, 0x42, 0xe9, 0xb4, 0xe4, 0xb1, 0xee, 0x89,
+	0x5f, 0x9f, 0x0d, 0xa0, 0x1d, 0xe8, 0x28, 0x1f, 0x31, 0x97, 0x2a, 0xc9, 0x86, 0xa1, 0xa6, 0xfe,
-	0xf1, 0x58, 0xc4, 0xcf, 0xe3, 0x21, 0x29, 0xbd, 0x51, 0x72, 0xf8, 0x23, 0x3c, 0x48, 0xf5, 0x4b,
+	0x1e, 0x76, 0xa1, 0x73, 0x1d, 0xc6, 0x4c, 0x7d, 0x0b, 0x1d, 0xb0, 0x03, 0x3d, 0xa3, 0x7e, 0xeb,
-	0x32, 0xab, 0xc5, 0x1d, 0x42, 0x5c, 0x76, 0x64, 0x9f, 0xdf, 0x77, 0xb9, 0x3a, 0xec, 0x42, 0x7d,
+	0xfc, 0xdf, 0x56, 0x3e, 0x00, 0x96, 0x8f, 0x40, 0x0f, 0xec, 0xfc, 0x60, 0x74, 0xfc, 0x22, 0x09,
-	0x20, 0xc3, 0x74, 0x04, 0x9e, 0x88, 0x58, 0xf7, 0x02, 0x49, 0xed, 0x2d, 0x6c, 0x42, 0xe3, 0x2c,
+	0xaf, 0x5c, 0x31, 0x7e, 0x03, 0x8f, 0xd7, 0xe7, 0x38, 0x23, 0xfa, 0xb5, 0x8f, 0x9f, 0x57, 0xc7,
-	0x88, 0x14, 0xb5, 0x2d, 0x74, 0xc0, 0x1e, 0xc8, 0x09, 0xb5, 0x6b, 0x47, 0xb7, 0xf5, 0x74, 0x40,
+	0x18, 0x47, 0xf0, 0xa4, 0xf9, 0x13, 0x80, 0x9e, 0xbf, 0xf5, 0xc3, 0xe2, 0x6d, 0xf7, 0x31, 0x3e,
-	0xcd, 0x49, 0x46, 0x0f, 0xec, 0x34, 0x71, 0x74, 0xfc, 0x8c, 0x24, 0x2f, 0x5f, 0x29, 0xfc, 0x09,
+	0x87, 0xfe, 0x66, 0xe9, 0xf1, 0xc4, 0x6f, 0x90, 0xb4, 0xd7, 0x84, 0x32, 0x7e, 0x07, 0x47, 0xb5,
-	0x1e, 0x2e, 0xbf, 0x33, 0x0a, 0xd1, 0x2f, 0x3d, 0xce, 0x5e, 0x19, 0x53, 0xd8, 0x87, 0x47, 0xd5,
+	0xe2, 0xe1, 0xfb, 0x7e, 0x93, 0x10, 0xbc, 0x46, 0x98, 0xf1, 0x2b, 0x78, 0xb4, 0xd6, 0xe2, 0x78,
-	0x4f, 0x14, 0x7a, 0xfe, 0xda, 0x87, 0xcf, 0x5b, 0xbf, 0xa7, 0xf0, 0x19, 0xb4, 0x57, 0xa5, 0x89,
+	0xe4, 0x6f, 0x8e, 0x0c, 0xaf, 0x06, 0xf1, 0x45, 0xe7, 0x6d, 0x3b, 0x9d, 0xcc, 0x7e, 0xdd, 0x37,
-	0x07, 0x7e, 0x45, 0xcb, 0x79, 0x55, 0xa8, 0xc2, 0xe7, 0xb0, 0x57, 0x12, 0x17, 0x7e, 0xec, 0x57,
+	0xff, 0x1f, 0xbe, 0xf8, 0x2f, 0x00, 0x00, 0xff, 0xff, 0x27, 0x95, 0xc1, 0x78, 0x4c, 0x08, 0x00,
-	0x09, 0xd5, 0xab, 0x84, 0x15, 0x7e, 0x3f, 0x2b, 0xe1, 0x7c, 0x04, 0xe1, 0x9e, 0xbf, 0x3a, 0xd2,
+	0x00,
-	0xbc, 0x12, 0xa4, 0xf0, 0x1d, 0x3c, 0xbe, 0xab, 0x7e, 0xd8, 0xf1, 0xef, 0xd1, 0x96, 0x77, 0x9f,
-	0x87, 0x3a, 0x6e, 0xbc, 0xad, 0x27, 0xa3, 0xc9, 0xbb, 0x6d, 0xf3, 0x1f, 0xea, 0xdb, 0x0f, 0x01,
-	0x00, 0x00, 0xff, 0xff, 0xad, 0x4e, 0x98, 0x29, 0x50, 0x09, 0x00, 0x00,
 }

replication/logic/pdu/pdu.proto

@@ -9,7 +9,6 @@ service Replication {
   rpc DestroySnapshots(DestroySnapshotsReq) returns (DestroySnapshotsRes);
   rpc ReplicationCursor(ReplicationCursorReq) returns (ReplicationCursorRes);
   rpc SendCompleted(SendCompletedReq) returns (SendCompletedRes);
-  rpc HintMostRecentCommonAncestor(HintMostRecentCommonAncestorReq) returns (HintMostRecentCommonAncestorRes);
   // for Send and Recv, see package rpc
 }
 
@@ -126,19 +125,3 @@ message PingRes {
   // Echo must be PingReq.Message
   string Echo = 1;
 }
-
-message HintMostRecentCommonAncestorReq {
-  string Filesystem = 1;
-
-  // A copy of the FilesystemVersion on the sending side that the replication
-  // algorithm identified as a shared most recent common version between sending
-  // and receiving side.
-  //
-  // If nil, this is an indication that the replication algorithm could not
-  // find a common ancestor between the two sides.
-  // NOTE: nilness does not mean that replication never happened - there could
-  // as well be a replication conflict. thus, dont' jump to conclusions too
-  // rapidly here.
-  FilesystemVersion SenderVersion = 2;
-}
-message HintMostRecentCommonAncestorRes {}

replication/logic/replication_logic.go

@@ -33,7 +33,6 @@ type Endpoint interface {
 	ListFilesystemVersions(ctx context.Context, req *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error)
 	DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshotsReq) (*pdu.DestroySnapshotsRes, error)
 	WaitForConnectivity(ctx context.Context) error
-	HintMostRecentCommonAncestor(context.Context, *pdu.HintMostRecentCommonAncestorReq) (*pdu.HintMostRecentCommonAncestorRes, error)
 }
 
 type Sender interface {
@@ -288,7 +287,10 @@ func (p *Planner) doPlanning(ctx context.Context) ([]*Filesystem, error) {
 			}
 		}
 
-		ctr := p.promBytesReplicated.WithLabelValues(fs.Path)
+		var ctr prometheus.Counter
+		if p.promBytesReplicated != nil {
+			ctr = p.promBytesReplicated.WithLabelValues(fs.Path)
+		}
 
 		q = append(q, &Filesystem{
 			sender:                 p.sender,
@@ -359,43 +361,6 @@ func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
 		log(ctx).WithField("token", resumeToken).Debug("decode resume token")
 	}
 
-	// give both sides a hint about how far prior replication attempts got
-	// This serves as a cummulative variant of SendCompleted and can be useful
-	// for example to release stale holds from an earlier (interrupted) replication.
-	// TODO FIXME: enqueue this as a replication step instead of doing it here during planning
-	//             then again, the step should run regardless of planning success
-	//             so maybe a separate phase before PLANNING, then?
-	path, conflict := IncrementalPath(rfsvs, sfsvs)
-	var sender_mrca *pdu.FilesystemVersion
-	if conflict == nil && len(path) > 0 {
-		sender_mrca = path[0] // shadow
-	}
-	// yes, sender_mrca may be nil, indicating that we do not have an mrca
-	{
-		var wg sync.WaitGroup
-		doHint := func(ep Endpoint, name string) {
-			defer wg.Done()
-			ctx, endTask := trace.WithTask(ctx, "hint-mrca-"+name)
-			defer endTask()
-
-			log := log(ctx).WithField("to_side", name).
-				WithField("sender_mrca", sender_mrca.String())
-			log.Debug("hint most recent common ancestor")
-			hint := &pdu.HintMostRecentCommonAncestorReq{
-				Filesystem:    fs.Path,
-				SenderVersion: sender_mrca,
-			}
-			_, err := ep.HintMostRecentCommonAncestor(ctx, hint)
-			if err != nil {
-				log.WithError(err).Error("error hinting most recent common ancestor")
-			}
-		}
-		wg.Add(2)
-		go doHint(fs.sender, "sender")
-		go doHint(fs.receiver, "receiver")
-		wg.Wait()
-	}
-
 	var steps []*Step
 	// build the list of replication steps
 	//
@@ -632,7 +597,9 @@ func (s *Step) doReplication(ctx context.Context) error {
 	s.byteCounterMtx.Unlock()
 	defer func() {
 		defer s.byteCounterMtx.Lock().Unlock()
+		if s.parent.promBytesReplicated != nil {
 			s.parent.promBytesReplicated.Add(float64(s.byteCounter.Count()))
+		}
 	}()
 
 	rr := &pdu.ReceiveReq{

rpc/dataconn/timeoutconn/timeoutconn_test.go

@@ -117,7 +117,7 @@ func TestNoPartialReadsDueToDeadline(t *testing.T) {
 		// 50ms for the second read, 100ms after that one for the deadline
 		// allow for some jitter
 		assert.True(t, readDuration > 140*time.Millisecond)
-		assert.True(t, readDuration < 160*time.Millisecond)
+		assert.True(t, readDuration < 200*time.Millisecond)
 	}()
 
 	wg.Wait()

rpc/rpc_client.go

@@ -142,13 +142,6 @@ func (c *Client) SendCompleted(ctx context.Context, in *pdu.SendCompletedReq) (*
 	return c.controlClient.SendCompleted(ctx, in)
 }
 
-func (c *Client) HintMostRecentCommonAncestor(ctx context.Context, in *pdu.HintMostRecentCommonAncestorReq) (*pdu.HintMostRecentCommonAncestorRes, error) {
-	ctx, endSpan := trace.WithSpan(ctx, "rpc.client.HintMostRecentCommonAncestor")
-	defer endSpan()
-
-	return c.controlClient.HintMostRecentCommonAncestor(ctx, in)
-}
-
 func (c *Client) WaitForConnectivity(ctx context.Context) error {
 	ctx, endSpan := trace.WithSpan(ctx, "rpc.client.WaitForConnectivity")
 	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, 3)
+	return DoHandshakeVersion(conn, deadline, 4)
 }
 
 const HandshakeMessageMaxLen = 16 * 4096

transport/tcp/serve_tcp.go

@@ -11,39 +11,6 @@ import (
 	"github.com/zrepl/zrepl/util/tcpsock"
 )
 
-type ipMapEntry struct {
-	ip    net.IP
-	ident string
-}
-
-type ipMap struct {
-	entries []ipMapEntry
-}
-
-func ipMapFromConfig(clients map[string]string) (*ipMap, error) {
-	entries := make([]ipMapEntry, 0, len(clients))
-	for clientIPString, clientIdent := range clients {
-		clientIP := net.ParseIP(clientIPString)
-		if clientIP == nil {
-			return nil, errors.Errorf("cannot parse client IP %q", clientIPString)
-		}
-		if err := transport.ValidateClientIdentity(clientIdent); err != nil {
-			return nil, errors.Wrapf(err, "invalid client identity for IP %q", clientIPString)
-		}
-		entries = append(entries, ipMapEntry{clientIP, clientIdent})
-	}
-	return &ipMap{entries: entries}, nil
-}
-
-func (m *ipMap) Get(ip net.IP) (string, error) {
-	for _, e := range m.entries {
-		if e.ip.Equal(ip) {
-			return e.ident, nil
-		}
-	}
-	return "", errors.Errorf("no identity mapping for client IP %s", ip)
-}
-
 func TCPListenerFactoryFromConfig(c *config.Global, in *config.TCPServe) (transport.AuthenticatedListenerFactory, error) {
 	clientMap, err := ipMapFromConfig(in.Clients)
 	if err != nil {
@@ -75,10 +42,13 @@ func (f *TCPAuthListener) Accept(ctx context.Context) (*transport.AuthConn, erro
 	if err != nil {
 		return nil, err
 	}
-	clientIP := nc.RemoteAddr().(*net.TCPAddr).IP
+	clientAddr := &net.IPAddr{
-	clientIdent, err := f.clientMap.Get(clientIP)
+		IP:   nc.RemoteAddr().(*net.TCPAddr).IP,
+		Zone: nc.RemoteAddr().(*net.TCPAddr).Zone,
+	}
+	clientIdent, err := f.clientMap.Get(clientAddr)
 	if err != nil {
-		transport.GetLogger(ctx).WithField("ip", clientIP).Error("client IP not in client map")
+		transport.GetLogger(ctx).WithField("ipaddr", clientAddr).Error("client IP not in client map")
 		nc.Close()
 		return nil, err
 	}

transport/tcp/serve_tcp_ipmap.go

@@ -0,0 +1,166 @@
+package tcp
+
+import (
+	"bytes"
+	"fmt"
+	"net"
+	"sort"
+	"strings"
+
+	"github.com/pkg/errors"
+	"github.com/zrepl/zrepl/transport"
+	"golang.org/x/sys/unix"
+)
+
+type ipMapEntry struct {
+	subnet *net.IPNet
+	// ident is always not empty
+	ident string
+	// zone may be empty (e.g. for IPv4)
+	zone string
+}
+
+type ipMap struct {
+	entries []*ipMapEntry
+}
+
+func ipMapFromConfig(clients map[string]string) (*ipMap, error) {
+
+	entries := make([]*ipMapEntry, 0, len(clients))
+	for clientInput, clientIdent := range clients {
+		userIPMapEntry, err := newIPMapEntry(clientInput, clientIdent)
+		if err != nil {
+			return nil, errors.Wrapf(err, "cannot not parse %q:%q", clientInput, clientIdent)
+		}
+
+		entries = append(entries, userIPMapEntry)
+	}
+
+	sort.Sort(byPrefixlen(entries))
+
+	return &ipMap{entries: entries}, nil
+}
+
+func (m *ipMap) Get(ipAddr *net.IPAddr) (string, error) {
+	for _, e := range m.entries {
+		if e.zone != ipAddr.Zone {
+			continue
+		}
+		if e.subnet.Contains(ipAddr.IP) {
+			return zfsDatasetPathComponentCompatibleRepresentation(e.ident, ipAddr), nil
+		}
+	}
+	return "", errors.Errorf("no identity mapping for client IP: %s%%%s", ipAddr.IP, ipAddr.Zone)
+}
+
+var ipv6FullySpecifiedMask = bytes.Repeat([]byte{0xff}, net.IPv6len)
+
+func newIPMapEntry(input string, ident string) (*ipMapEntry, error) {
+	ip := input
+	var zone string
+
+	ipZoneSplit := strings.SplitN(input, "%", 2)
+	if len(ipZoneSplit) > 1 {
+		ip = ipZoneSplit[0]
+		zone = ipZoneSplit[1]
+	}
+
+	_, subnet, err := net.ParseCIDR(ip)
+	if err != nil {
+		// expect full IP, no '*' placeholder expansion
+
+		if strings.Count(ident, "*") != 0 {
+			return nil, fmt.Errorf("non-CIDR matches must not contain '*' placeholder")
+		}
+
+		if err := transport.ValidateClientIdentity(ident); err != nil {
+			return nil, errors.Wrapf(err, "invalid client identity %q for IP %q", ident, ip)
+		}
+
+		parsedIP := net.ParseIP(ip)
+		if parsedIP == nil {
+			return nil, errors.Wrapf(err, "invalid client address %q", ip)
+		}
+		parsedIP = parsedIP.To16()
+
+		return &ipMapEntry{
+			subnet: &net.IPNet{
+				IP:   parsedIP,
+				Mask: ipv6FullySpecifiedMask,
+			},
+			zone:  zone,
+			ident: ident,
+		}, nil
+	}
+
+	// expect CIDR and '*' placeholder expansion
+
+	if strings.Count(ident, "*") != 1 {
+		err = fmt.Errorf("CIDRs require 1 IP placeholder")
+		return nil, errors.Wrapf(err, "invalid client identity %q for IP %q", ident, ip)
+	}
+
+	longestIPAddr := net.IPAddr{
+		IP:   net.IP(bytes.Repeat([]byte{0xff}, net.IPv6len)),
+		Zone: strings.Repeat("i", unix.IFNAMSIZ),
+	}
+	longestIdent := zfsDatasetPathComponentCompatibleRepresentation(ident, &longestIPAddr)
+
+	if err := transport.ValidateClientIdentity(longestIdent); err != nil {
+		return nil, errors.Wrapf(err, "invalid client identity for IP %q", ip)
+	}
+
+	ones, _ := subnet.Mask.Size()
+	preExpansionAddrlen := len(subnet.IP) * 8
+	expanded := subnet.IP.To16()
+	postExpansionAddrlen := len(expanded) * 8
+	return &ipMapEntry{
+		subnet: &net.IPNet{
+			IP:   expanded,
+			Mask: net.CIDRMask(postExpansionAddrlen-preExpansionAddrlen+ones, postExpansionAddrlen),
+		},
+		zone:  zone,
+		ident: ident,
+	}, nil
+}
+
+func zfsDatasetPathComponentCompatibleRepresentation(identityWithWildcard string, addr *net.IPAddr) string {
+	// If a Zone exists we append it after the IP using a "-" because "%"
+	// is a zfs dataset forbidden char.
+	if addr.Zone != "" {
+		return strings.Replace(identityWithWildcard, "*", addr.IP.String()+"-"+addr.Zone, 1)
+	}
+	// newIPMapEntry validates that the line contains exactly one "*"
+	return strings.Replace(identityWithWildcard, "*", addr.IP.String(), 1)
+}
+
+func (e *ipMapEntry) String() string {
+	return fmt.Sprintf("&ipMapEntry{subnet=%q, ident=%q, zone=%q}", e.subnet.String(), e.ident, e.zone)
+}
+
+func (e *ipMapEntry) PrefixLen() int {
+	ones, bits := e.subnet.Mask.Size()
+	if bits != net.IPv6len*8 {
+		panic(fmt.Sprintf("impl error: we represent all addresses as 16byte internally ones=%v bits=%v: %s", ones, bits, e))
+	}
+	return ones
+}
+
+// newtype to support sorting by prefixlength
+type byPrefixlen []*ipMapEntry
+
+func (m byPrefixlen) Len() int { return len(m) }
+func (m byPrefixlen) Less(i, j int) bool {
+
+	if m[i].PrefixLen() != m[j].PrefixLen() {
+		return m[i].PrefixLen() > m[j].PrefixLen()
+	}
+
+	addrCmp := bytes.Compare(m[i].subnet.IP.To16(), m[j].subnet.IP.To16())
+	if addrCmp != 0 {
+		return addrCmp < 0
+	}
+
+	return strings.Compare(m[i].zone, m[j].zone) < 0
+}
+func (m byPrefixlen) Swap(i, j int) { m[i], m[j] = m[j], m[i] }

transport/tcp/serve_tcp_ipmap_test.go

@@ -0,0 +1,204 @@
+package tcp
+
+import (
+	"net"
+	"os"
+	"testing"
+
+	"github.com/kr/pretty"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestIPMap(t *testing.T) {
+	type testCaseExpect struct {
+		expectNoMapping bool
+		expectIdent     string
+	}
+	type testCase struct {
+		name          string
+		config        map[string]string
+		expectInitErr bool
+		expect        map[string]testCaseExpect
+	}
+
+	cases := []testCase{
+		{
+			name:          "regular ips",
+			expectInitErr: false,
+			config: map[string]string{
+				"192.168.123.234": "ident1",
+				"192.168.20.23":   "ident2",
+			},
+			expect: map[string]testCaseExpect{
+				"192.168.123.234": {expectIdent: "ident1"},
+				"192.168.20.23":   {expectIdent: "ident2"},
+				"192.168.20.10":   {expectNoMapping: true},
+			},
+		},
+		{
+			name:          "full wildcard",
+			expectInitErr: false,
+			config: map[string]string{
+				"0.0.0.0/0": "*",
+				"0::/0":     "*",
+			},
+			expect: map[string]testCaseExpect{
+				"10.123.234.24": {expectIdent: "10.123.234.24"},
+				// '[' and ']' are forbiddenin dataset names
+				"fe80::23:42": {expectIdent: "fe80::23:42"},
+			},
+		},
+		{
+			name:          "longest prefix matching",
+			expectInitErr: false,
+			config: map[string]string{
+				"10.1.2.3":              "specific-host",
+				"10.1.2.0/24":           "subnet-one-two-*",
+				"10.1.1.0/24":           "subnet-one-one-*",
+				"10.1.0.0/24":           "subnet-one-zero-*",
+				"10.1.0.0/16":           "subnet-one-*",
+				"fde4:8dba:82e1:1::1":   "v6-48-1-specialhost",
+				"fde4:8dba:82e1::/48":   "v6-48-*",
+				"fde4:8dba:82e1:1::/64": "v6-64-1-*",
+				"fde4:8dba:82e1:2::/64": "v6-64-2-*",
+			},
+			expect: map[string]testCaseExpect{
+				"10.1.2.3":             {expectIdent: "specific-host"},
+				"10.1.2.1":             {expectIdent: "subnet-one-two-10.1.2.1"},
+				"10.1.1.1":             {expectIdent: "subnet-one-one-10.1.1.1"},
+				"10.1.0.23":            {expectIdent: "subnet-one-zero-10.1.0.23"},
+				"10.1.3.1":             {expectIdent: "subnet-one-10.1.3.1"},
+				"10.2.1.1":             {expectNoMapping: true},
+				"fde4:8dba:82e1:1::1":  {expectIdent: "v6-48-1-specialhost"},
+				"fde4:8dba:82e1:23::1": {expectIdent: "v6-48-fde4:8dba:82e1:23::1"},
+				"fde4:8dba:82e1:1::2":  {expectIdent: "v6-64-1-fde4:8dba:82e1:1::2"},
+				"fde4:8dba:82e1:2::1":  {expectIdent: "v6-64-2-fde4:8dba:82e1:2::1"},
+				"fde4:8dba:82e2::1":    {expectNoMapping: true},
+			},
+		},
+		{
+			name:          "different prefixes, mixed ipv4 ipv6, with interface ids",
+			expectInitErr: false,
+			config: map[string]string{
+				"192.168.23.0/24": "db-*",
+				"192.168.23.23":   "db-twentythree",
+				"192.168.42.0/24": "web-*",
+				"10.1.4.0/24":     "my-*-server",
+				"2001:0db8:85a3:0000:0000:8a2e:0370:7334":      "aspecifichost",
+				"2001:0db8:85a3:0000:0000:8a2e:0370:7334%eth1": "aspecifichost",
+				"fe80::/16%eth1":      "san-*",
+				"fde4:8dba:82e1::/64": "sub64-*",
+			},
+			expect: map[string]testCaseExpect{
+				"10.1.2.3":        {expectNoMapping: true},
+				"192.168.23.1":    {expectIdent: "db-192.168.23.1"},
+				"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"},
+
+				// 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},
+
+				// v6 subnet matching
+				"fde4:8dba:82e1::1": {expectIdent: "sub64-fde4:8dba:82e1::1"},
+				// v6 subnet matching with suffix that matches another allowed IPv4
+				"fde4:8dba:82e1::c0a8:1717": {expectIdent: "sub64-fde4:8dba:82e1::c0a8:1717"},
+
+				"2001:0db8:85a3:0000:0000:8a2e:0370:7334":      {expectIdent: "aspecifichost"},
+				"2001:0db8:85a3:0000:0000:8a2e:0370:7334%eth1": {expectIdent: "aspecifichost"},
+				"2001:0db8:85a3:0000:0000:8a2e:0370:7334%eth2": {expectNoMapping: true},
+			},
+		},
+		{
+			name:          "invalid user input: non ip or cidr",
+			expectInitErr: true,
+			config: map[string]string{
+				"jimmy": "db",
+			},
+		},
+		{
+			name:          "invalid user input: v4 ip with an identity containing *",
+			expectInitErr: true,
+			config: map[string]string{
+				"192.168.1.2": "db-*",
+			},
+		},
+		{
+			name:          "invalid user input: v4 subnet without an identity containing *",
+			expectInitErr: true,
+			config: map[string]string{
+				"192.168.1.0/24": "db-",
+			},
+		},
+		{
+			name:          "invalid user input: v6 ip with an identity containing *",
+			expectInitErr: true,
+			config: map[string]string{
+				"2001:0db8:85a3:0000:0000:8a2e:0370:7334": "aspecifichost*",
+			},
+		},
+		{
+			name:          "invalid user input: v6 subnet without identity containing *",
+			expectInitErr: true,
+			config: map[string]string{
+				"fe80::/16%eth1": "db-",
+			},
+		},
+		{
+			name:          "invalid user input with subnet match: client identity with forbidden zfs dataset name char @",
+			expectInitErr: true,
+			config: map[string]string{
+				"fe80::/16": "db@-*",
+			},
+		},
+		{
+			name:          "invalid user input with IP match: client identity with forbidden zfs dataset name char @",
+			expectInitErr: true,
+			config: map[string]string{
+				"fe80::1": "db@foo",
+			},
+		},
+	}
+
+	for i := range cases {
+		c := cases[i]
+		t.Run(c.name, func(t *testing.T) {
+			pretty.Fprintf(os.Stderr, "running %#v\n", c)
+			m, err := ipMapFromConfig(c.config)
+			if c.expectInitErr {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.NotNil(t, m)
+			}
+			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, _ := newIPMapEntry(input, "test")
+				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")
+				require.NotNil(t, ipMapEntry)
+				ident, err := m.Get(&net.IPAddr{
+					IP:   ipMapEntry.subnet.IP,
+					Zone: ipMapEntry.zone,
+				})
+				if expect.expectNoMapping {
+					assert.Empty(t, ident)
+				} else {
+					assert.NoError(t, err)
+					assert.Equal(t, expect.expectIdent, ident)
+				}
+			}
+		})
+	}
+}
+
+func TestPackageNetAssumptions(t *testing.T) {
+
+}

transport/transport.go

@@ -4,10 +4,10 @@ package transport
 
 import (
 	"context"
-	"errors"
 	"net"
 	"syscall"
 
+	"github.com/pkg/errors"
 	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/rpc/dataconn/timeoutconn"
@@ -56,13 +56,10 @@ type Connecter interface {
 }
 
 // A client identity must be a single component in a ZFS filesystem path
-func ValidateClientIdentity(in string) (err error) {
+func ValidateClientIdentity(in string) error {
-	path, err := zfs.NewDatasetPath(in)
+	err := zfs.ComponentNamecheck(in)
 	if err != nil {
-		return err
+		return errors.Wrap(err, "client identity must be usable as a single dataset path component")
-	}
-	if path.Length() != 1 {
-		return errors.New("client identity must be a single path component (not empty, no '/')")
 	}
 	return nil
 }

util/envconst/envconst.go

@@ -12,84 +12,100 @@ import (
 
 var cache sync.Map
 
-func Duration(varname string, def time.Duration) time.Duration {
+func Reset() {
+	cache.Range(func(key, _ interface{}) bool {
+		cache.Delete(key)
+		return true
+	})
+}
+
+func Duration(varname string, def time.Duration) (d time.Duration) {
+	var err error
 	if v, ok := cache.Load(varname); ok {
 		return v.(time.Duration)
 	}
 	e := os.Getenv(varname)
 	if e == "" {
-		return def
+		d = def
-	}
+	} else {
-	d, err := time.ParseDuration(e)
+		d, err = time.ParseDuration(e)
 		if err != nil {
 			panic(err)
 		}
+	}
 	cache.Store(varname, d)
 	return d
 }
 
-func Int(varname string, def int) int {
+func Int(varname string, def int) (d int) {
 	if v, ok := cache.Load(varname); ok {
 		return v.(int)
 	}
 	e := os.Getenv(varname)
 	if e == "" {
-		return def
+		d = def
-	}
+	} else {
 		d64, err := strconv.ParseInt(e, 10, strconv.IntSize)
 		if err != nil {
 			panic(err)
 		}
-	d := int(d64)
+		d = int(d64)
+	}
 	cache.Store(varname, d)
 	return d
 }
 
-func Int64(varname string, def int64) int64 {
+func Int64(varname string, def int64) (d int64) {
+	var err error
 	if v, ok := cache.Load(varname); ok {
 		return v.(int64)
 	}
 	e := os.Getenv(varname)
 	if e == "" {
-		return def
+		d = def
-	}
+	} else {
-	d, err := strconv.ParseInt(e, 10, 64)
+		d, err = strconv.ParseInt(e, 10, 64)
 		if err != nil {
 			panic(err)
 		}
+	}
 	cache.Store(varname, d)
 	return d
 }
 
-func Bool(varname string, def bool) bool {
+func Bool(varname string, def bool) (d bool) {
+	var err error
 	if v, ok := cache.Load(varname); ok {
 		return v.(bool)
 	}
 	e := os.Getenv(varname)
 	if e == "" {
-		return def
+		d = def
-	}
+	} else {
-	d, err := strconv.ParseBool(e)
+		d, err = strconv.ParseBool(e)
 		if err != nil {
 			panic(err)
 		}
+	}
 	cache.Store(varname, d)
 	return d
 }
 
-func String(varname string, def string) string {
+func String(varname string, def string) (d string) {
 	if v, ok := cache.Load(varname); ok {
 		return v.(string)
 	}
 	e := os.Getenv(varname)
 	if e == "" {
-		return def
+		d = def
+	} else {
+		d = e
 	}
-	cache.Store(varname, e)
+	cache.Store(varname, d)
-	return e
+	return d
 }
 
-func Var(varname string, def flag.Value) interface{} {
+func Var(varname string, def flag.Value) (d interface{}) {
 
 	// use def's type to instantiate a new object of that same type
 	// and call flag.Value.Set() on it
@@ -100,19 +116,43 @@ func Var(varname string, def flag.Value) interface{} {
 	defElemType := defType.Elem()
 
 	if v, ok := cache.Load(varname); ok {
-		return v.(string)
+		return v
-	}
-	e := os.Getenv(varname)
-	if e == "" {
-		return def
 	}
 
+	e := os.Getenv(varname)
+	if e == "" {
+		d = def
+	} else {
 		newInstance := reflect.New(defElemType)
 		if err := newInstance.Interface().(flag.Value).Set(e); err != nil {
 			panic(err)
 		}
-
+		d = newInstance.Interface()
-	res := newInstance.Interface()
+	}
-	cache.Store(varname, res)
+
-	return res
+	cache.Store(varname, d)
+	return d
+}
+
+type Report struct {
+	Entries []EntryReport
+}
+
+type EntryReport struct {
+	Var         string
+	Value       string
+	ValueGoType string
+}
+
+func GetReport() *Report {
+	var r Report
+	cache.Range(func(key, value interface{}) bool {
+		r.Entries = append(r.Entries, EntryReport{
+			Var:         key.(string),
+			Value:       fmt.Sprintf("%v", value),
+			ValueGoType: fmt.Sprintf("%T", value),
+		})
+		return true
+	})
+	return &r
 }

util/envconst/envconst_test.go

@@ -32,19 +32,27 @@ func (m *ExampleVarType) Set(s string) error {
 
 const EnvVarName = "ZREPL_ENVCONST_UNIT_TEST_VAR"
 
-func TestVar(t *testing.T) {
+func TestVarDefaultValue(t *testing.T) {
+	envconst.Reset()
 	_, set := os.LookupEnv(EnvVarName)
 	require.False(t, set)
 	defer os.Unsetenv(EnvVarName)
 
 	val := envconst.Var(EnvVarName, &Var1)
 	if &Var1 != val {
-		t.Errorf("default value shut be same address")
+		t.Errorf("default value should be same address")
 	}
+}
+
+func TestVarOverriddenValue(t *testing.T) {
+	envconst.Reset()
+	_, set := os.LookupEnv(EnvVarName)
+	require.False(t, set)
+	defer os.Unsetenv(EnvVarName)
 
 	err := os.Setenv(EnvVarName, "var2")
 	require.NoError(t, err)
 
-	val = envconst.Var(EnvVarName, &Var1)
+	val := envconst.Var(EnvVarName, &Var1)
 	require.Equal(t, &Var2, val, "only structural identity is required for non-default vars")
 }

zfs/versions.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"io"
 	"strconv"
 	"strings"
 	"time"
@@ -116,6 +115,13 @@ func (v FilesystemVersion) RelName() string {
 }
 func (v FilesystemVersion) String() string { return v.RelName() }
 
+// Only takes into account those attributes of FilesystemVersion that
+// are immutable over time in ZFS.
+func FilesystemVersionEqualIdentity(a, b FilesystemVersion) bool {
+	// .Name is mutable
+	return a.Guid == b.Guid && a.CreateTXG == b.CreateTXG && a.Creation == b.Creation
+}
+
 func (v FilesystemVersion) ToAbsPath(p *DatasetPath) string {
 	var b bytes.Buffer
 	b.WriteString(p.ToString())
@@ -224,10 +230,6 @@ func ZFSListFilesystemVersions(ctx context.Context, fs *DatasetPath, options Lis
 	res = make([]FilesystemVersion, 0)
 	for listResult := range listResults {
 		if listResult.Err != nil {
-			if listResult.Err == io.ErrUnexpectedEOF {
-				// Since we specified the fs on the command line, we'll treat this like the filesystem doesn't exist
-				return []FilesystemVersion{}, nil
-			}
 			return nil, listResult.Err
 		}
 

zfs/zfs.go

@@ -454,7 +454,7 @@ func (s *SendStream) killAndWait(precedingReadErr error) error {
 			WaitErr: exitErr,
 		}
 	} else {
-		s.opErr = fmt.Errorf("zfs send exited with status code 0")
+		s.opErr = precedingReadErr
 	}
 
 	// detect the edge where we're called from s.Read
@@ -727,13 +727,13 @@ func (a ZFSSendArgsUnvalidated) validateCorrespondsToResumeToken(ctx context.Con
 
 	if (a.From != nil) != t.HasFromGUID { // existence must be same
 		if t.HasFromGUID {
-			return gen.fmt("resume token not expected to be incremental, but `fromguid` = %q", t.FromGUID)
+			return gen.fmt("resume token not expected to be incremental, but `fromguid` = %v", t.FromGUID)
 		} else {
 			return gen.fmt("resume token expected to be incremental, but `fromguid` not present")
 		}
 	} else if t.HasFromGUID { // if exists (which is same, we checked above), they must match
 		if t.FromGUID != a.From.GUID {
-			return gen.fmt("resume token `fromguid` != expected: %q != %q", t.FromGUID, a.From.GUID)
+			return gen.fmt("resume token `fromguid` != expected: %v != %v", t.FromGUID, a.From.GUID)
 		}
 	} else {
 		_ = struct{}{} // both empty, ok
@@ -744,7 +744,7 @@ func (a ZFSSendArgsUnvalidated) validateCorrespondsToResumeToken(ctx context.Con
 		return gen.fmt("resume token does not have `toguid`")
 	}
 	if t.ToGUID != a.To.GUID { // a.To != nil because Validate checks for that
-		return gen.fmt("resume token `toguid` != expected: %q != %q", t.ToGUID, a.To.GUID)
+		return gen.fmt("resume token `toguid` != expected: %v != %v", t.ToGUID, a.To.GUID)
 	}
 
 	if a.Encrypted.B {
@@ -1014,7 +1014,9 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, stream io.Rea
 		snaps, err := ZFSListFilesystemVersions(ctx, fsdp, ListFilesystemVersionsOptions{
 			Types: Snapshots,
 		})
-		if err != nil {
+		if _, ok := err.(*DatasetDoesNotExist); ok {
+			snaps = []FilesystemVersion{}
+		} else if err != nil {
 			return fmt.Errorf("cannot list versions for rollback for forced receive: %s", err)
 		}
 		sort.Slice(snaps, func(i, j int) bool {
@@ -1595,22 +1597,32 @@ var ErrBookmarkCloningNotSupported = fmt.Errorf("bookmark cloning feature is not
 //
 // does not destroy an existing bookmark, returns
 //
-func ZFSBookmark(ctx context.Context, fs string, v ZFSSendArgVersion, bookmark string) (err error) {
+func ZFSBookmark(ctx context.Context, fs string, v FilesystemVersion, bookmark string) (bm FilesystemVersion, err error) {
+
+	bm = FilesystemVersion{
+		Type:     Bookmark,
+		Name:     bookmark,
+		UserRefs: OptionUint64{Valid: false},
+		// bookmarks have the same createtxg, guid and creation as their origin
+		CreateTXG: v.CreateTXG,
+		Guid:      v.Guid,
+		Creation:  v.Creation,
+	}
 
 	promTimer := prometheus.NewTimer(prom.ZFSBookmarkDuration.WithLabelValues(fs))
 	defer promTimer.ObserveDuration()
 
 	if !v.IsSnapshot() {
-		return ErrBookmarkCloningNotSupported // TODO This is work in progress: https://github.com/zfsonlinux/zfs/pull/9571
+		return bm, ErrBookmarkCloningNotSupported // TODO This is work in progress: https://github.com/zfsonlinux/zfs/pull/9571
 	}
 
 	snapname := v.FullPath(fs)
 	if err := EntityNamecheck(snapname, EntityTypeSnapshot); err != nil {
-		return err
+		return bm, err
 	}
 	bookmarkname := fmt.Sprintf("%s#%s", fs, bookmark)
 	if err := EntityNamecheck(bookmarkname, EntityTypeBookmark); err != nil {
-		return err
+		return bm, err
 	}
 
 	debug("bookmark: %q %q", snapname, bookmarkname)
@@ -1619,27 +1631,27 @@ func ZFSBookmark(ctx context.Context, fs string, v ZFSSendArgVersion, bookmark s
 	stdio, err := cmd.CombinedOutput()
 	if err != nil {
 		if ddne := tryDatasetDoesNotExist(snapname, stdio); ddne != nil {
-			return ddne
+			return bm, ddne
 		} else if zfsBookmarkExistsRegex.Match(stdio) {
 
 			// check if this was idempotent
 			bookGuid, err := ZFSGetGUID(ctx, fs, "#"+bookmark)
 			if err != nil {
-				return errors.Wrap(err, "bookmark idempotency check") // guid error expressive enough
+				return bm, errors.Wrap(err, "bookmark idempotency check") // guid error expressive enough
 			}
 
-			if v.GUID == bookGuid {
+			if v.Guid == bookGuid {
-				debug("bookmark: %q %q was idempotent: {snap,book}guid %d == %d", snapname, bookmarkname, v.GUID, bookGuid)
+				debug("bookmark: %q %q was idempotent: {snap,book}guid %d == %d", snapname, bookmarkname, v.Guid, bookGuid)
-				return nil
+				return bm, nil
 			}
-			return &BookmarkExists{
+			return bm, &BookmarkExists{
-				fs: fs, bookmarkOrigin: v, bookmark: bookmark,
+				fs: fs, bookmarkOrigin: v.ToSendArgVersion(), bookmark: bookmark,
 				zfsMsg:   string(stdio),
 				bookGuid: bookGuid,
 			}
 
 		} else {
-			return &ZFSError{
+			return bm, &ZFSError{
 				Stderr:  stdio,
 				WaitErr: err,
 			}
@@ -1647,8 +1659,7 @@ func ZFSBookmark(ctx context.Context, fs string, v ZFSSendArgVersion, bookmark s
 
 	}
 
-	return nil
+	return bm, nil
-
 }
 
 func ZFSRollback(ctx context.Context, fs *DatasetPath, snapshot FilesystemVersion, rollbackArgs ...string) (err error) {