WIP endpoint abstractions + pruning integration / pruner rewrite

ZFSListChan returns (*DatasetDoesNotExist) for the case mentioned in the comment

.circleci/config.yml

@@ -60,10 +60,10 @@ jobs:
 
       - run: make zrepl-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
@@ -92,11 +92,21 @@ 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 -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:

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:
-	$(GO_BUILD) -o "$(ARTIFACTDIR)/platformtest-$(ZREPL_TARGET_TUPLE)" ./platformtest/harness
+generate-platform-test-list:
+	$(GO_BUILD) -o $(ARTIFACTDIR)/generate-platform-test-list ./platformtest/tests/gen
 
-##################### DEV TARGETS #####################
-# not part of the build, must do that manually
-.PHONY: generate format platformtest
-
-generate:
-	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')
+test-platform-bin:
+	$(GO_ENV_VARS) $(GO) test $(GO_BUILDFLAGS) \
+		-c -o "$(ARTIFACTDIR)/platformtest-$(ZREPL_TARGET_TUPLE)" \
+		-covermode=atomic -cover -coverpkg github.com/zrepl/zrepl/... \
+		./platformtest/harness
 
 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

@@ -309,6 +309,11 @@ type PruneKeepNotReplicated struct {
 	KeepSnapshotAtCursor bool   `yaml:"keep_snapshot_at_cursor,optional,default=true"`
 }
 
+type PruneKeepStepHolds struct {
+	Type             string   `yaml:"type"`
+	AdditionalJobIds []string `yaml:"additional_job_ids,optional"`
+}
+
 type PruneKeepLastN struct {
 	Type  string `yaml:"type"`
 	Count int    `yaml:"count"`
@@ -480,6 +485,7 @@ func (t *ServeEnum) UnmarshalYAML(u func(interface{}, bool) error) (err error) {
 
 func (t *PruningEnum) UnmarshalYAML(u func(interface{}, bool) error) (err error) {
 	t.Ret, err = enumUnmarshal(u, map[string]interface{}{
+		"step_holds":     &PruneKeepStepHolds{},
 		"not_replicated": &PruneKeepNotReplicated{},
 		"last_n":         &PruneKeepLastN{},
 		"grid":           &PruneGrid{},

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/daemon.go

@@ -13,6 +13,7 @@ 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/config"
 	"github.com/zrepl/zrepl/daemon/job"
@@ -50,6 +51,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 +96,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")
 

daemon/job/active.go

@@ -376,9 +376,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/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

@@ -102,9 +102,9 @@ import (
 )
 
 var metrics struct {
-	activeTasks                          prometheus.Gauge
-	uniqueConcurrentTaskNameBitvecLength *prometheus.GaugeVec
+	activeTasks prometheus.Gauge
 }
+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 {
-		this.parentTask.mtx.HoldWhile(func() {
-			if !this.parentTask.endedAt.IsZero() {
-				panic(ErrParentTaskAlreadyEnded)
-			}
-			this.parentTask.activeChildTasks++
-		})
+	if parentTask != nil {
+		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,17 +310,47 @@ 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
+	symbolizeSpan func(s *traceNode) string
+}
 
-	task := this.parentTask
-	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 {
@@ -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 task.annotation, spanIdStack
+	return fmt.Sprintf("%s$%s", taskStack, strings.Join(spanIdsRev, "."))
 }
 
-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 spanStack
+		return n.TaskAndSpanStack(StackKindId)
 	} 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,
+		active: make(map[string]*bitset.BitSet),
 	}
 }
 
@@ -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)

daemon/pruner.v2/fsstate_enumer.go

@@ -0,0 +1,72 @@
+// Code generated by "enumer -type=FSState -json"; DO NOT EDIT.
+
+//
+package pruner
+
+import (
+	"encoding/json"
+	"fmt"
+)
+
+const _FSStateName = "FSStateInitializedFSStatePlanningFSStatePlanErrFSStateExecutingFSStateExecuteErrFSStateExecuteSuccess"
+
+var _FSStateIndex = [...]uint8{0, 18, 33, 47, 63, 80, 101}
+
+func (i FSState) String() string {
+	if i < 0 || i >= FSState(len(_FSStateIndex)-1) {
+		return fmt.Sprintf("FSState(%d)", i)
+	}
+	return _FSStateName[_FSStateIndex[i]:_FSStateIndex[i+1]]
+}
+
+var _FSStateValues = []FSState{0, 1, 2, 3, 4, 5}
+
+var _FSStateNameToValueMap = map[string]FSState{
+	_FSStateName[0:18]:   0,
+	_FSStateName[18:33]:  1,
+	_FSStateName[33:47]:  2,
+	_FSStateName[47:63]:  3,
+	_FSStateName[63:80]:  4,
+	_FSStateName[80:101]: 5,
+}
+
+// FSStateString retrieves an enum value from the enum constants string name.
+// Throws an error if the param is not part of the enum.
+func FSStateString(s string) (FSState, error) {
+	if val, ok := _FSStateNameToValueMap[s]; ok {
+		return val, nil
+	}
+	return 0, fmt.Errorf("%s does not belong to FSState values", s)
+}
+
+// FSStateValues returns all values of the enum
+func FSStateValues() []FSState {
+	return _FSStateValues
+}
+
+// IsAFSState returns "true" if the value is listed in the enum definition. "false" otherwise
+func (i FSState) IsAFSState() bool {
+	for _, v := range _FSStateValues {
+		if i == v {
+			return true
+		}
+	}
+	return false
+}
+
+// MarshalJSON implements the json.Marshaler interface for FSState
+func (i FSState) MarshalJSON() ([]byte, error) {
+	return json.Marshal(i.String())
+}
+
+// UnmarshalJSON implements the json.Unmarshaler interface for FSState
+func (i *FSState) UnmarshalJSON(data []byte) error {
+	var s string
+	if err := json.Unmarshal(data, &s); err != nil {
+		return fmt.Errorf("FSState should be a string, got %s", data)
+	}
+
+	var err error
+	*i, err = FSStateString(s)
+	return err
+}

daemon/pruner.v2/pruner.go

@@ -0,0 +1,454 @@
+package pruner
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+
+	"github.com/pkg/errors"
+	"github.com/zrepl/zrepl/daemon/logging"
+	"github.com/zrepl/zrepl/daemon/logging/trace"
+	"github.com/zrepl/zrepl/endpoint"
+	"github.com/zrepl/zrepl/pruning"
+	"github.com/zrepl/zrepl/zfs"
+)
+
+type Pruner struct {
+	fsfilter  endpoint.FSFilter
+	jid       endpoint.JobID
+	side      Side
+	keepRules []pruning.KeepRule
+
+	// all channels consumed by the run loop
+	reportReqs chan reportRequest
+	stopReqs   chan stopRequest
+	done       chan struct{}
+	fsListRes  chan fsListRes
+
+	state State
+
+	listFilesystemsError error       // only in state StateListFilesystemsError
+	fsPruners            []*FSPruner // only in state StateFanOutFilesystems
+}
+
+//go:generate enumer -type=State -json
+type State int
+
+const (
+	StateInitialized State = iota
+	StateListFilesystems
+	StateListFilesystemsError
+	StateFanOutFilesystems
+	StateDone
+)
+
+type Report struct {
+	State                State
+	ListFilesystemsError error       // only valid in StateListFilesystemsError
+	Filesystems          []*FSReport // valid from StateFanOutFilesystems
+}
+
+type reportRequest struct {
+	ctx   context.Context
+	reply chan *Report
+}
+
+type runRequest struct {
+	complete chan struct{}
+}
+
+type stopRequest struct {
+	complete chan struct{}
+}
+
+type fsListRes struct {
+	filesystems []*zfs.DatasetPath
+	err         error
+}
+
+type Side interface {
+	// may return both nil, indicating there is no replication position
+	GetReplicationPosition(ctx context.Context, fs string) (*zfs.FilesystemVersion, error)
+	isSide() Side
+}
+
+func NewPruner(fsfilter endpoint.FSFilter, jid endpoint.JobID, side Side, keepRules []pruning.KeepRule) *Pruner {
+	return &Pruner{
+		fsfilter,
+		jid,
+		side,
+		keepRules,
+		make(chan reportRequest),
+		make(chan stopRequest),
+		make(chan struct{}),
+		make(chan fsListRes),
+		StateInitialized,
+		nil,
+		nil,
+	}
+}
+
+func (p *Pruner) Run(ctx context.Context) *Report {
+
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	if p.state != StateInitialized {
+		panic("Run can onl[y be called once")
+	}
+
+	go func() {
+		fss, err := zfs.ZFSListMapping(ctx, p.fsfilter)
+		p.fsListRes <- fsListRes{fss, err}
+	}()
+
+	for {
+		select {
+		case res := <-p.fsListRes:
+			if res.err != nil {
+				p.state = StateListFilesystemsError
+				p.listFilesystemsError = res.err
+				close(p.done)
+				continue
+			}
+
+			p.state = StateFanOutFilesystems
+
+			p.fsPruners = make([]*FSPruner, len(res.filesystems))
+			_, add, end := trace.WithTaskGroup(ctx, "pruner-fan-out-fs")
+			for i, fs := range res.filesystems {
+				p.fsPruners[i] = NewFSPruner(p.jid, p.side, p.keepRules, fs)
+				add(func(ctx context.Context) {
+					p.fsPruners[i].Run(ctx)
+				})
+			}
+			go func() {
+				end()
+				close(p.done)
+			}()
+
+		case req := <-p.stopReqs:
+			cancel()
+			go func() {
+				<-p.done
+				close(req.complete)
+			}()
+		case req := <-p.reportReqs:
+			req.reply <- p.report(req.ctx)
+		case <-p.done:
+			p.state = StateDone
+			return p.report(ctx)
+		}
+	}
+}
+
+func (p *Pruner) Report(ctx context.Context) *Report {
+	req := reportRequest{
+		ctx:   ctx,
+		reply: make(chan *Report, 1),
+	}
+	select {
+	case p.reportReqs <- req:
+		return <-req.reply
+	case <-ctx.Done():
+		return nil
+	case <-p.done:
+		return nil
+	}
+}
+
+func (p *Pruner) report(ctx context.Context) *Report {
+	fsreports := make([]*FSReport, len(p.fsPruners))
+	for i := range fsreports {
+		fsreports[i] = p.fsPruners[i].report()
+	}
+	return &Report{
+		State:                p.state,
+		ListFilesystemsError: p.listFilesystemsError,
+		Filesystems:          fsreports,
+	}
+}
+
+// implements pruning.Snapshot
+type snapshot struct {
+	replicated bool
+	stepHolds  []pruning.StepHold
+	zfs.FilesystemVersion
+
+	state     SnapState
+	destroyOp *zfs.DestroySnapOp
+}
+
+//go:generate enumer -type=SnapState -json
+type SnapState int
+
+const (
+	SnapStateInitialized SnapState = iota
+	SnapStateKeeping
+	SnapStateDeletePending
+	SnapStateDeleteAttempted
+)
+
+// implements pruning.StepHold
+type stepHold struct {
+	endpoint.Abstraction
+}
+
+func (s snapshot) Replicated() bool              { return s.replicated }
+func (s snapshot) StepHolds() []pruning.StepHold { return s.stepHolds }
+
+func (s stepHold) GetJobID() endpoint.JobID { return *s.Abstraction.GetJobID() }
+
+type FSPruner struct {
+	jid       endpoint.JobID
+	side      Side
+	keepRules []pruning.KeepRule
+	fsp       *zfs.DatasetPath
+
+	state FSState
+
+	// all channels consumed by the run loop
+	planned    chan fsPlanRes
+	executed   chan fsExecuteRes
+	done       chan struct{}
+	reportReqs chan fsReportReq
+
+	keepList    []*snapshot // valid in FSStateExecuting and forward
+	destroyList []*snapshot // valid in FSStateExecuting and forward, field .destroyOp is invalid until FSStateExecuting is left
+
+}
+
+type fsPlanRes struct {
+	keepList    []*snapshot
+	destroyList []*snapshot
+	err         error
+}
+
+type fsExecuteRes struct {
+	completedDestroyOps []*zfs.DestroySnapOp // same len() as FSPruner.destroyList
+}
+
+type fsReportReq struct {
+	res chan *FSReport
+}
+
+type FSReport struct {
+	State    FSState
+	KeepList []*SnapReport
+	Destroy  []*SnapReport
+}
+
+type SnapReport struct {
+	State         SnapState
+	Name          string
+	Replicated    bool
+	StepHoldCount int
+	DestroyError  error
+}
+
+//go:generate enumer -type=FSState -json
+type FSState int
+
+const (
+	FSStateInitialized FSState = iota
+	FSStatePlanning
+	FSStatePlanErr
+	FSStateExecuting
+	FSStateExecuteErr
+	FSStateExecuteSuccess
+)
+
+func (s FSState) IsTerminal() bool {
+	return s == FSStatePlanErr || s == FSStateExecuteErr || s == FSStateExecuteSuccess
+}
+
+func NewFSPruner(jid endpoint.JobID, side Side, keepRules []pruning.KeepRule, fsp *zfs.DatasetPath) *FSPruner {
+	return &FSPruner{
+		jid, side, keepRules, fsp,
+		FSStateInitialized,
+		make(chan fsPlanRes),
+		make(chan fsExecuteRes),
+		make(chan struct{}),
+		make(chan fsReportReq),
+		nil, nil,
+	}
+}
+
+func (p *FSPruner) Run(ctx context.Context) *FSReport {
+
+	defer func() {
+	}()
+
+	p.state = FSStatePlanning
+
+	go func() { p.planned <- p.plan(ctx) }()
+
+out:
+	for !p.state.IsTerminal() {
+		select {
+		case res := <-p.planned:
+
+			if res.err != nil {
+				p.state = FSStatePlanErr
+				continue
+			}
+			p.state = FSStateExecuting
+			p.keepList = res.keepList
+			p.destroyList = res.destroyList
+
+			go func() { p.executed <- p.execute(ctx, p.destroyList) }()
+
+		case res := <-p.executed:
+
+			if len(res.completedDestroyOps) != len(p.destroyList) {
+				panic("impl error: completedDestroyOps is a vector corresponding to entries in p.destroyList")
+			}
+
+			var erronous []*zfs.DestroySnapOp
+			for i, op := range res.completedDestroyOps {
+				if *op.ErrOut != nil {
+					erronous = append(erronous, op)
+				}
+				p.destroyList[i].destroyOp = op
+				p.destroyList[i].state = SnapStateDeleteAttempted
+			}
+			if len(erronous) > 0 {
+				p.state = FSStateExecuteErr
+			} else {
+				p.state = FSStateExecuteSuccess
+			}
+
+			close(p.done)
+
+		case <-p.reportReqs:
+			panic("unimp")
+		case <-p.done:
+			break out
+		}
+	}
+
+	// TODO render last FS report
+	return nil
+}
+
+func (p *FSPruner) plan(ctx context.Context) fsPlanRes {
+	fs := p.fsp.ToString()
+	vs, err := zfs.ZFSListFilesystemVersions(ctx, p.fsp, zfs.ListFilesystemVersionsOptions{})
+	if err != nil {
+		return fsPlanRes{err: errors.Wrap(err, "list filesystem versions")}
+	}
+
+	allJobsStepHolds, absErrs, err := endpoint.ListAbstractions(ctx, endpoint.ListZFSHoldsAndBookmarksQuery{
+		FS: endpoint.ListZFSHoldsAndBookmarksQueryFilesystemFilter{
+			FS: &fs,
+		},
+		What: endpoint.AbstractionTypeSet{
+			endpoint.AbstractionStepHold: true,
+		},
+		Concurrency: 1,
+	})
+	if err != nil {
+		return fsPlanRes{err: errors.Wrap(err, "list abstractions")}
+	}
+	if len(absErrs) > 0 {
+		logging.GetLogger(ctx, logging.SubsysPruning).WithError(endpoint.ListAbstractionsErrors(absErrs)).
+			Error("error listing some step holds, prune attempt might fail with 'dataset is busy' errors")
+	}
+
+	repPos, err := p.side.GetReplicationPosition(ctx, p.fsp.ToString())
+	if err != nil {
+		return fsPlanRes{err: errors.Wrap(err, "get replication position")}
+	}
+
+	vsAsSnaps := make([]pruning.Snapshot, len(vs))
+	for i := range vs {
+		var repPosCreateTxgOrZero uint64
+		if repPos != nil {
+			repPosCreateTxgOrZero = repPos.GetCreateTXG()
+		}
+		s := &snapshot{
+			state:             SnapStateInitialized,
+			FilesystemVersion: vs[i],
+			replicated:        vs[i].GetCreateTXG() <= repPosCreateTxgOrZero,
+		}
+		for _, h := range allJobsStepHolds {
+			if zfs.FilesystemVersionEqualIdentity(vs[i], h.GetFilesystemVersion()) {
+				s.stepHolds = append(s.stepHolds, stepHold{h})
+			}
+		}
+		vsAsSnaps[i] = s
+	}
+
+	downcastToSnapshots := func(l []pruning.Snapshot) (r []*snapshot) {
+		r = make([]*snapshot, len(l))
+		for i, e := range l {
+			r[i] = e.(*snapshot)
+		}
+		return r
+	}
+	pruningResult := pruning.PruneSnapshots(vsAsSnaps, p.keepRules)
+	remove, keep := downcastToSnapshots(pruningResult.Remove), downcastToSnapshots(pruningResult.Keep)
+	if len(remove)+len(keep) != len(vsAsSnaps) {
+		for _, s := range vsAsSnaps {
+			r, _ := json.MarshalIndent(s.(*snapshot).report(), "", "  ")
+			fmt.Fprintf(os.Stderr, "%s\n", string(r))
+		}
+		panic("indecisive")
+	}
+
+	for _, s := range remove {
+		s.state = SnapStateDeletePending
+	}
+	for _, s := range keep {
+		s.state = SnapStateKeeping
+	}
+
+	return fsPlanRes{keepList: keep, destroyList: remove, err: nil}
+}
+
+func (p *FSPruner) execute(ctx context.Context, destroyList []*snapshot) fsExecuteRes {
+	ops := make([]*zfs.DestroySnapOp, len(destroyList))
+	for i, fsv := range p.destroyList {
+		ops[i] = &zfs.DestroySnapOp{
+			Filesystem: p.fsp.ToString(),
+			Name:       fsv.GetName(),
+			ErrOut:     new(error),
+		}
+	}
+	zfs.ZFSDestroyFilesystemVersions(ctx, ops)
+
+	return fsExecuteRes{completedDestroyOps: ops}
+}
+
+func (p *FSPruner) report() *FSReport {
+	return &FSReport{
+		State:    p.state,
+		KeepList: p.reportRenderSnapReports(p.keepList),
+		Destroy:  p.reportRenderSnapReports(p.destroyList),
+	}
+}
+
+func (p *FSPruner) reportRenderSnapReports(l []*snapshot) (r []*SnapReport) {
+	r = make([]*SnapReport, len(l))
+	for i := range l {
+		r[i] = l[i].report()
+	}
+	return r
+}
+
+func (s *snapshot) report() *SnapReport {
+	var snapErr error
+	if s.state == SnapStateDeleteAttempted {
+		if *s.destroyOp.ErrOut != nil {
+			snapErr = (*s.destroyOp.ErrOut)
+		}
+	}
+	return &SnapReport{
+		State:         s.state,
+		Name:          s.Name,
+		Replicated:    s.Replicated(),
+		StepHoldCount: len(s.stepHolds),
+		DestroyError:  snapErr,
+	}
+}

daemon/pruner.v2/pruner_side.go

@@ -0,0 +1,27 @@
+package pruner
+
+import (
+	"context"
+
+	"github.com/zrepl/zrepl/endpoint"
+	"github.com/zrepl/zrepl/zfs"
+)
+
+type SideSender struct {
+	jobID endpoint.JobID
+}
+
+func NewSideSender(jid endpoint.JobID) *SideSender {
+	return &SideSender{jid}
+}
+
+func (s *SideSender) isSide() Side { return nil }
+
+var _ Side = (*SideSender)(nil)
+
+func (s *SideSender) GetReplicationPosition(ctx context.Context, fs string) (*zfs.FilesystemVersion, error) {
+	if fs == "" {
+		panic("must not pass zero value for fs")
+	}
+	return endpoint.GetMostRecentReplicationCursorOfJob(ctx, fs, s.jobID)
+}

daemon/pruner.v2/snapstate_enumer.go

@@ -0,0 +1,70 @@
+// Code generated by "enumer -type=SnapState -json"; DO NOT EDIT.
+
+//
+package pruner
+
+import (
+	"encoding/json"
+	"fmt"
+)
+
+const _SnapStateName = "SnapStateInitializedSnapStateKeepingSnapStateDeletePendingSnapStateDeleteAttempted"
+
+var _SnapStateIndex = [...]uint8{0, 20, 36, 58, 82}
+
+func (i SnapState) String() string {
+	if i < 0 || i >= SnapState(len(_SnapStateIndex)-1) {
+		return fmt.Sprintf("SnapState(%d)", i)
+	}
+	return _SnapStateName[_SnapStateIndex[i]:_SnapStateIndex[i+1]]
+}
+
+var _SnapStateValues = []SnapState{0, 1, 2, 3}
+
+var _SnapStateNameToValueMap = map[string]SnapState{
+	_SnapStateName[0:20]:  0,
+	_SnapStateName[20:36]: 1,
+	_SnapStateName[36:58]: 2,
+	_SnapStateName[58:82]: 3,
+}
+
+// SnapStateString retrieves an enum value from the enum constants string name.
+// Throws an error if the param is not part of the enum.
+func SnapStateString(s string) (SnapState, error) {
+	if val, ok := _SnapStateNameToValueMap[s]; ok {
+		return val, nil
+	}
+	return 0, fmt.Errorf("%s does not belong to SnapState values", s)
+}
+
+// SnapStateValues returns all values of the enum
+func SnapStateValues() []SnapState {
+	return _SnapStateValues
+}
+
+// IsASnapState returns "true" if the value is listed in the enum definition. "false" otherwise
+func (i SnapState) IsASnapState() bool {
+	for _, v := range _SnapStateValues {
+		if i == v {
+			return true
+		}
+	}
+	return false
+}
+
+// MarshalJSON implements the json.Marshaler interface for SnapState
+func (i SnapState) MarshalJSON() ([]byte, error) {
+	return json.Marshal(i.String())
+}
+
+// UnmarshalJSON implements the json.Unmarshaler interface for SnapState
+func (i *SnapState) UnmarshalJSON(data []byte) error {
+	var s string
+	if err := json.Unmarshal(data, &s); err != nil {
+		return fmt.Errorf("SnapState should be a string, got %s", data)
+	}
+
+	var err error
+	*i, err = SnapStateString(s)
+	return err
+}

daemon/pruner.v2/state_enumer.go

@@ -0,0 +1,71 @@
+// Code generated by "enumer -type=State -json"; DO NOT EDIT.
+
+//
+package pruner
+
+import (
+	"encoding/json"
+	"fmt"
+)
+
+const _StateName = "StateInitializedStateListFilesystemsStateListFilesystemsErrorStateFanOutFilesystemsStateDone"
+
+var _StateIndex = [...]uint8{0, 16, 36, 61, 83, 92}
+
+func (i State) String() string {
+	if i < 0 || i >= State(len(_StateIndex)-1) {
+		return fmt.Sprintf("State(%d)", i)
+	}
+	return _StateName[_StateIndex[i]:_StateIndex[i+1]]
+}
+
+var _StateValues = []State{0, 1, 2, 3, 4}
+
+var _StateNameToValueMap = map[string]State{
+	_StateName[0:16]:  0,
+	_StateName[16:36]: 1,
+	_StateName[36:61]: 2,
+	_StateName[61:83]: 3,
+	_StateName[83:92]: 4,
+}
+
+// StateString retrieves an enum value from the enum constants string name.
+// Throws an error if the param is not part of the enum.
+func StateString(s string) (State, error) {
+	if val, ok := _StateNameToValueMap[s]; ok {
+		return val, nil
+	}
+	return 0, fmt.Errorf("%s does not belong to State values", s)
+}
+
+// StateValues returns all values of the enum
+func StateValues() []State {
+	return _StateValues
+}
+
+// IsAState returns "true" if the value is listed in the enum definition. "false" otherwise
+func (i State) IsAState() bool {
+	for _, v := range _StateValues {
+		if i == v {
+			return true
+		}
+	}
+	return false
+}
+
+// MarshalJSON implements the json.Marshaler interface for State
+func (i State) MarshalJSON() ([]byte, error) {
+	return json.Marshal(i.String())
+}
+
+// UnmarshalJSON implements the json.Unmarshaler interface for State
+func (i *State) UnmarshalJSON(data []byte) error {
+	var s string
+	if err := json.Unmarshal(data, &s); err != nil {
+		return fmt.Errorf("State should be a string, got %s", data)
+	}
+
+	var err error
+	*i, err = StateString(s)
+	return err
+}

daemon/pruner/pruner.go

@@ -20,15 +20,14 @@ import (
 )
 
 // Try to keep it compatible with github.com/zrepl/zrepl/endpoint.Endpoint
-type History interface {
-	ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error)
+type Endpoint interface {
 	ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error)
+	ListFilesystemVersions(ctx context.Context, req *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error)
 }
 
 // Try to keep it compatible with github.com/zrepl/zrepl/endpoint.Endpoint
 type Target interface {
-	ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error)
-	ListFilesystemVersions(ctx context.Context, req *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error)
+	Endpoint
 	DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshotsReq) (*pdu.DestroySnapshotsRes, error)
 }
 
@@ -46,13 +45,14 @@ func GetLogger(ctx context.Context) Logger {
 }
 
 type args struct {
-	ctx                            context.Context
-	target                         Target
-	receiver                       History
-	rules                          []pruning.KeepRule
-	retryWait                      time.Duration
-	considerSnapAtCursorReplicated bool
-	promPruneSecs                  prometheus.Observer
+	ctx                              context.Context
+	target                           Target
+	sender, receiver                 Endpoint
+	rules                            []pruning.KeepRule
+	retryWait                        time.Duration
+	considerSnapAtCursorReplicated   bool
+	convertAnyStepHoldToStepBookmark bool
+	promPruneSecs                    prometheus.Observer
 }
 
 type Pruner struct {
@@ -70,11 +70,12 @@ type Pruner struct {
 }
 
 type PrunerFactory struct {
-	senderRules                    []pruning.KeepRule
-	receiverRules                  []pruning.KeepRule
-	retryWait                      time.Duration
-	considerSnapAtCursorReplicated bool
-	promPruneSecs                  *prometheus.HistogramVec
+	senderRules                      []pruning.KeepRule
+	receiverRules                    []pruning.KeepRule
+	retryWait                        time.Duration
+	considerSnapAtCursorReplicated   bool
+	convertAnyStepHoldToStepBookmark bool
+	promPruneSecs                    *prometheus.HistogramVec
 }
 
 type LocalPrunerFactory struct {
@@ -122,25 +123,35 @@ func NewPrunerFactory(in config.PruningSenderReceiver, promPruneSecs *prometheus
 		}
 		considerSnapAtCursorReplicated = considerSnapAtCursorReplicated || !knr.KeepSnapshotAtCursor
 	}
+
+	convertAnyStepHoldToStepBookmark := false
+	for _, r := range in.KeepSender {
+		_, ok := r.Ret.(*config.PruneKeepStepHolds)
+		convertAnyStepHoldToStepBookmark = convertAnyStepHoldToStepBookmark || ok
+	}
+
 	f := &PrunerFactory{
-		senderRules:                    keepRulesSender,
-		receiverRules:                  keepRulesReceiver,
-		retryWait:                      envconst.Duration("ZREPL_PRUNER_RETRY_INTERVAL", 10*time.Second),
-		considerSnapAtCursorReplicated: considerSnapAtCursorReplicated,
-		promPruneSecs:                  promPruneSecs,
+		senderRules:                      keepRulesSender,
+		receiverRules:                    keepRulesReceiver,
+		retryWait:                        envconst.Duration("ZREPL_PRUNER_RETRY_INTERVAL", 10*time.Second),
+		considerSnapAtCursorReplicated:   considerSnapAtCursorReplicated,
+		convertAnyStepHoldToStepBookmark: convertAnyStepHoldToStepBookmark,
+		promPruneSecs:                    promPruneSecs,
 	}
 	return f, nil
 }
 
-func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, target Target, receiver History) *Pruner {
+func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, sender Target, receiver Endpoint) *Pruner {
 	p := &Pruner{
 		args: args{
 			context.WithValue(ctx, contextKeyPruneSide, "sender"),
-			target,
+			sender,
+			sender,
 			receiver,
 			f.senderRules,
 			f.retryWait,
 			f.considerSnapAtCursorReplicated,
+			f.convertAnyStepHoldToStepBookmark,
 			f.promPruneSecs.WithLabelValues("sender"),
 		},
 		state: Plan,
@@ -148,15 +159,17 @@ func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, target Target, re
 	return p
 }
 
-func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target, receiver History) *Pruner {
+func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, receiver Target, sender Endpoint) *Pruner {
 	p := &Pruner{
 		args: args{
 			context.WithValue(ctx, contextKeyPruneSide, "receiver"),
-			target,
+			receiver,
+			sender,
 			receiver,
 			f.receiverRules,
 			f.retryWait,
 			false, // senseless here anyways
+			false, // senseless here anyways
 			f.promPruneSecs.WithLabelValues("receiver"),
 		},
 		state: Plan,
@@ -164,15 +177,17 @@ func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target,
 	return p
 }
 
-func (f *LocalPrunerFactory) BuildLocalPruner(ctx context.Context, target Target, receiver History) *Pruner {
+func (f *LocalPrunerFactory) BuildLocalPruner(ctx context.Context, target Target) *Pruner {
 	p := &Pruner{
 		args: args{
 			context.WithValue(ctx, contextKeyPruneSide, "local"),
 			target,
-			receiver,
+			target,
+			target,
 			f.keepRules,
 			f.retryWait,
 			false, // considerSnapAtCursorReplicated is not relevant for local pruning
+			false, // convertAnyStepHoldToStepBookmark is not relevant for local pruning
 			f.promPruneSecs.WithLabelValues("local"),
 		},
 		state: Plan,
@@ -341,11 +356,13 @@ func (s snapshot) Replicated() bool { return s.replicated }
 
 func (s snapshot) Date() time.Time { return s.date }
 
+func (s snapshot) CreateTXG() uint64 { return s.fsv.GetCreateTXG() }
+
 func doOneAttempt(a *args, u updater) {
 
-	ctx, target, receiver := a.ctx, a.target, a.receiver
+	ctx, sender, receiver, target := a.ctx, a.sender, a.receiver, a.target
 
-	sfssres, err := receiver.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
+	sfssres, err := sender.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
 	if err != nil {
 		u(func(p *Pruner) {
 			p.state = PlanErr
@@ -407,6 +424,10 @@ tfss_loop:
 
 		pfs.snaps = make([]pruning.Snapshot, 0, len(tfsvs))
 
+		receiver.ListFilesystemVersions(ctx, &pdu.ListFilesystemVersionsReq{
+			Filesystem: tfs.Path,
+		})
+
 		rcReq := &pdu.ReplicationCursorReq{
 			Filesystem: tfs.Path,
 		}
@@ -415,6 +436,7 @@ tfss_loop:
 			pfsPlanErrAndLog(err, "cannot get replication cursor bookmark")
 			continue tfss_loop
 		}
+
 		if rc.GetNotexist() {
 			err := errors.New("replication cursor bookmark does not exist (one successful replication is required before pruning works)")
 			pfsPlanErrAndLog(err, "")

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" : "mx01"
+          "192.168.122.123" :               "mysql01",
+          "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
-`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.
-However, until we get around documenting those setups, you will have to run zrepl as root or experiment yourself :)
-
-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/>`_
-    * - 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`.
+    installation/user-privileges
+    installation/packages
+    installation/apt-repos
+    installation/compile-from-source
+    installation/freebsd-jail-with-iocage
+    installation/what-next

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"
 
@@ -112,124 +112,20 @@ func (s *Sender) ListFilesystemVersions(ctx context.Context, r *pdu.ListFilesyst
 	for i := range fsvs {
 		rfsvs[i] = pdu.FilesystemVersionFromZFS(&fsvs[i])
 	}
+
+	sendAbstractions := sendAbstractionsCacheSingleton.GetByFS(ctx, lp.ToString())
+	rSabsInfo := make([]*pdu.SendAbstraction, len(sendAbstractions))
+	for i := range rSabsInfo {
+		rSabsInfo[i] = SendAbstractionToPDU(sendAbstractions[i])
+	}
+
 	res := &pdu.ListFilesystemVersionsRes{Versions: rfsvs}
 	return res, nil
 
 }
 
-func (p *Sender) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.HintMostRecentCommonAncestorReq) (*pdu.HintMostRecentCommonAncestorRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-
-	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))
+var maxConcurrentZFSSend = envconst.Int64("ZREPL_ENDPOINT_MAX_CONCURRENT_SEND", 10)
+var maxConcurrentZFSSendSemaphore = semaphore.New(maxConcurrentZFSSend)
 
 func uncheckedSendArgsFromPDU(fsv *pdu.FilesystemVersion) *zfs.ZFSSendArgVersion {
 	if fsv == nil {
@@ -319,10 +215,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 +228,10 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
 		}
 	}
 
+	var fromHold, toHold Abstraction
 	// make sure `From` doesn't go away in order to make this step resumable
 	if sendArgs.From != nil {
-		_, 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
@@ -342,17 +240,71 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
 		}
 	}
 	// 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.
+	func() {
+		ctx, endSpan := trace.WithSpan(ctx, "cleanup-stale-abstractions")
+		defer endSpan()
+
+		liveAbs := []Abstraction{fromHold, toHold, fromReplicationCursor}
+		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)
+	}()
+
+	if fromHold != nil {
+		sendAbstractionsCacheSingleton.Put(fromHold)
+	}
+	sendAbstractionsCacheSingleton.Put(toHold)
+	if fromReplicationCursor != nil {
+		sendAbstractionsCacheSingleton.Put(fromReplicationCursor)
+	}
 
 	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 +341,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")
-	destroyedCursors, err := MoveReplicationCursor(ctx, fs, to, p.jobId)
+	toReplicationCursor, err := CreateReplicationCursor(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 +349,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
-	// if we fail to release them, don't bother the caller:
-	// 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`")
-		}
-
-	}()
-	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()
+	keep := func(a Abstraction) bool {
+		return AbstractionEquals(a, toReplicationCursor)
+	}
+	sendAbstractionsCacheSingleton.TryBatchDestroy(ctx, p.jobId, fs, keep, nil)
 
 	return &pdu.SendCompletedRes{}, nil
+
 }
 
 func (p *Sender) DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshotsReq) (*pdu.DestroySnapshotsRes, error) {
@@ -858,9 +766,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 +882,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,232 @@
+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()
+
+}
+
+func (s *sendAbstractionsCache) GetByFS(ctx context.Context, fs string) (ret []Abstraction) {
+	defer s.mtx.Lock().Unlock()
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+	if fs == "" {
+		panic("must not pass zero-value fs")
+	}
+
+	s.tryLoadOnDiskSendAbstractions(ctx, fs)
+
+	for _, a := range s.abstractions {
+		if a.GetFS() == fs {
+			ret = append(ret, a)
+		}
+	}
+
+	return ret
+}
+
+// - 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,23 +614,23 @@ 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 {
+						emitCandidate(a)
+					}
 				}
 			}
-			if a != nil {
-				emitCandidate(a)
-			}
 		}
 	}
 }

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 {
-	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.
+// idempotently create a replication cursor targeting `target`
 //
 // 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) {
-
-	if !target.IsSnapshot() {
-		return nil, zfs.ErrBookmarkCloningNotSupported
-	}
+func CreateReplicationCursor(ctx context.Context, fs string, target zfs.FilesystemVersion, jobID JobID) (a Abstraction, err error) {
 
 	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
-	// cleanup the old one afterwards
+	if target.IsBookmark() && target.GetName() == bookmarkname {
+		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)
-	if err != nil {
-		return nil, errors.Wrap(err, "destroy obsolete replication cursors")
-	}
-
-	return destroyedCursors, nil
+	return &bookmarkBasedAbstraction{
+		Type:              AbstractionReplicationCursorBookmarkV2,
+		FS:                fs,
+		FilesystemVersion: cursorBookmark,
+		JobID:             jobID,
+	}, nil
 }
 
-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")
-	}
-}
+const (
+	ReplicationCursorBookmarkNamePrefix = "zrepl_last_received_J_"
+)
 
 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_pdu.go

@@ -0,0 +1,28 @@
+package endpoint
+
+import "github.com/zrepl/zrepl/replication/logic/pdu"
+
+func SendAbstractionToPDU(a Abstraction) *pdu.SendAbstraction {
+	var ty pdu.SendAbstraction_SendAbstractionType
+	switch a.GetType() {
+	case AbstractionLastReceivedHold:
+		panic(a)
+	case AbstractionReplicationCursorBookmarkV1:
+		panic(a)
+	case AbstractionReplicationCursorBookmarkV2:
+		ty = pdu.SendAbstraction_ReplicationCursorV2
+	case AbstractionStepHold:
+		ty = pdu.SendAbstraction_StepHold
+	case AbstractionStepBookmark:
+		ty = pdu.SendAbstraction_StepBookmark
+	default:
+		panic(a)
+	}
+
+	version := a.GetFilesystemVersion()
+	return &pdu.SendAbstraction{
+		Type:    ty,
+		JobID:   (*a.GetJobID()).String(),
+		Version: pdu.FilesystemVersionFromZFS(&version),
+	}
+}

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

@@ -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

@@ -268,6 +268,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 +351,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 {
-	createArgs            platformtest.ZpoolCreateArgs
-	stopAndKeepPoolOnFail bool
-
-	run   string
-	runRE *regexp.Regexp
-}
+const DefaultPoolImageSize = 200 * (1 << 20)
 
 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", "", "")
-	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()
+func HarnessRun(args HarnessArgs) error {
 
-	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
+		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/batchDestroy.go

@@ -2,8 +2,9 @@ package tests
 
 import (
 	"fmt"
-	"strings"
 
+	"github.com/kr/pretty"
+	"github.com/stretchr/testify/require"
 	"github.com/zrepl/zrepl/platformtest"
 	"github.com/zrepl/zrepl/zfs"
 )
@@ -17,7 +18,9 @@ func BatchDestroy(ctx *platformtest.Context) {
 		+  "foo bar@1"
 		+  "foo bar@2"
 		+  "foo bar@3"
+		+  "foo bar@4"
 		R  zfs hold zrepl_platformtest "${ROOTDS}/foo bar@2"
+		R  zfs hold zrepl_platformtest "${ROOTDS}/foo bar@4"
 	`)
 
 	reqs := []*zfs.DestroySnapOp{
@@ -31,24 +34,40 @@ func BatchDestroy(ctx *platformtest.Context) {
 			Filesystem: fmt.Sprintf("%s/foo bar", ctx.RootDataset),
 			Name:       "2",
 		},
+		&zfs.DestroySnapOp{
+			ErrOut:     new(error),
+			Filesystem: fmt.Sprintf("%s/foo bar", ctx.RootDataset),
+			Name:       "non existent",
+		},
+		&zfs.DestroySnapOp{
+			ErrOut:     new(error),
+			Filesystem: fmt.Sprintf("%s/foo bar", ctx.RootDataset),
+			Name:       "4",
+		},
 	}
 	zfs.ZFSDestroyFilesystemVersions(ctx, reqs)
+
+	pretty.Println(reqs)
+
 	if *reqs[0].ErrOut != nil {
 		panic("expecting no error")
 	}
-	err := (*reqs[1].ErrOut).Error()
-	if !strings.Contains(err, fmt.Sprintf("%s/foo bar@2", ctx.RootDataset)) {
-		panic(fmt.Sprintf("expecting error about being unable to destroy @2: %T\n%s", err, err))
-	}
+
+	eBusy, ok := (*reqs[1].ErrOut).(*zfs.ErrDestroySnapshotDatasetIsBusy)
+	require.True(ctx, ok)
+	require.Equal(ctx, reqs[1].Name, eBusy.Name)
+
+	require.Nil(ctx, *reqs[2].ErrOut, "destroying non-existent snap is not an error (idempotence)")
+
+	eBusy, ok = (*reqs[3].ErrOut).(*zfs.ErrDestroySnapshotDatasetIsBusy)
+	require.True(ctx, ok)
+	require.Equal(ctx, reqs[3].Name, eBusy.Name)
 
 	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
 	!N "foo bar@3"
 	!E "foo bar@1"
 	!E "foo bar@2"
-	R zfs release zrepl_platformtest "${ROOTDS}/foo bar@2"
-	-  "foo bar@2"
-	-  "foo bar@1"
-	-  "foo bar"
+	!E "foo bar@4"
 	`)
 
 }

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,
+	Pruner2NotReplicated,
+	ReceiveForceIntoEncryptedErr,
+	ReceiveForceRollbackWorksUnencrypted,
+	ReplicationIncrementalCleansUpStaleAbstractionsWithCacheOnSecondReplication,
+	ReplicationIncrementalCleansUpStaleAbstractionsWithoutCacheOnSecondReplication,
+	ReplicationIncrementalIsPossibleIfCommonSnapshotIsDestroyed,
+	ReplicationIsResumableFullSend,
+	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")
-	anotherSnap := sendArgVersion(ctx, fs, "@another snap")
+	asnap := fsversion(ctx, fs, "@a 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")
-	err := zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
+	asnap := fsversion(ctx, fs, "@a snap")
+	_, 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/pruner.go.deact

@@ -0,0 +1,331 @@
+package tests
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/require"
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon/filters"
+	"github.com/zrepl/zrepl/daemon/pruner"
+	"github.com/zrepl/zrepl/endpoint"
+	"github.com/zrepl/zrepl/platformtest"
+	"github.com/zrepl/zrepl/zfs"
+)
+
+func PrunerNotReplicated(ctx *platformtest.Context) {
+
+	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+        DESTROYROOT
+        CREATEROOT
+        +  "foo bar"
+        +  "foo bar@1"
+        +  "foo bar@2"
+        +  "foo bar@3"
+        +  "foo bar@4"
+        +  "foo bar@5"
+    `)
+
+	c, err := config.ParseConfigBytes([]byte(fmt.Sprintf(`
+jobs:
+- name: prunetest
+  type: push
+  filesystems: {
+      "%s/foo bar<": true
+  }
+  connect:
+    type: tcp
+    address: 255.255.255.255:255
+  snapshotting:
+    type: manual
+  pruning:
+    keep_sender:
+    - type: not_replicated
+    - type: last_n
+      count: 1
+    keep_receiver:
+    - type: last_n
+      count: 2
+    `, ctx.RootDataset)))
+	require.NoError(ctx, err)
+
+	pushJob := c.Jobs[0].Ret.(*config.PushJob)
+
+	dummyHistVec := prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: "foo",
+		Subsystem: "foo",
+		Name:      "foo",
+		Help:      "foo",
+	}, []string{"foo"})
+
+	prunerFactory, err := pruner.NewPrunerFactory(pushJob.Pruning, dummyHistVec)
+	require.NoError(ctx, err)
+
+	senderJid := endpoint.MustMakeJobID("sender-job")
+
+	fsfilter, err := filters.DatasetMapFilterFromConfig(pushJob.Filesystems)
+	require.NoError(ctx, err)
+
+	sender := endpoint.NewSender(endpoint.SenderConfig{
+		FSF: fsfilter,
+		Encrypt: &zfs.NilBool{
+			B: false,
+		},
+		JobID: senderJid,
+	})
+
+	fs := ctx.RootDataset + "/foo bar"
+
+	// create a replication cursor to make pruning work at all
+	_, err = endpoint.CreateReplicationCursor(ctx, fs, fsversion(ctx, fs, "@2"), senderJid)
+	require.NoError(ctx, err)
+
+	p := prunerFactory.BuildSenderPruner(ctx, sender, sender)
+
+	p.Prune()
+
+	report := p.Report()
+
+	reportJSON, err := json.MarshalIndent(report, "", "  ")
+	require.NoError(ctx, err)
+	ctx.Logf("%s\n", string(reportJSON))
+
+	require.Equal(ctx, pruner.Done.String(), report.State)
+	require.Len(ctx, report.Completed, 1)
+	fsReport := report.Completed[0]
+	require.Equal(ctx, fs, fsReport.Filesystem)
+	require.Empty(ctx, fsReport.SkipReason)
+	require.Empty(ctx, fsReport.LastError)
+	require.Len(ctx, fsReport.DestroyList, 1)
+	require.Equal(ctx, fsReport.DestroyList[0], pruner.SnapshotReport{
+		Name:       "1",
+		Replicated: true,
+		Date:       fsReport.DestroyList[0].Date,
+	})
+
+}
+
+func PrunerNoKeepNotReplicatedNoKeepStepHoldConvertsAnyStepHoldToBookmark(ctx *platformtest.Context) {
+
+	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+        DESTROYROOT
+        CREATEROOT
+        +  "foo bar"
+        +  "foo bar@1"
+        +  "foo bar@2"
+        +  "foo bar@3"
+        +  "foo bar@4"
+        +  "foo bar@5"
+    `)
+
+	c, err := config.ParseConfigBytes([]byte(fmt.Sprintf(`
+jobs:
+- name: prunetest
+  type: push
+  filesystems: {
+      "%s/foo bar<": true
+  }
+  connect:
+    type: tcp
+    address: 255.255.255.255:255
+  snapshotting:
+    type: manual
+  pruning:
+    keep_sender:
+    - type: last_n
+      count: 1
+    keep_receiver:
+    - type: last_n
+      count: 2
+    `, ctx.RootDataset)))
+	require.NoError(ctx, err)
+
+	pushJob := c.Jobs[0].Ret.(*config.PushJob)
+
+	dummyHistVec := prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: "foo",
+		Subsystem: "foo",
+		Name:      "foo",
+		Help:      "foo",
+	}, []string{"foo"})
+
+	prunerFactory, err := pruner.NewPrunerFactory(pushJob.Pruning, dummyHistVec)
+	require.NoError(ctx, err)
+
+	senderJid := endpoint.MustMakeJobID("sender-job")
+
+	fsfilter, err := filters.DatasetMapFilterFromConfig(pushJob.Filesystems)
+	require.NoError(ctx, err)
+
+	sender := endpoint.NewSender(endpoint.SenderConfig{
+		FSF: fsfilter,
+		Encrypt: &zfs.NilBool{
+			B: false,
+		},
+        JobID: senderJid,
+	})
+
+	fs := ctx.RootDataset + "/foo bar"
+
+	// create a replication cursor to make pruning work at all
+	_, err = endpoint.CreateReplicationCursor(ctx, fs, fsversion(ctx, fs, "@2"), senderJid)
+	require.NoError(ctx, err)
+
+	// create step holds for the incremental @2->@3
+	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@2"), senderJid)
+	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@3"), senderJid)
+	// create step holds for another job
+	otherJid := endpoint.MustMakeJobID("other-job")
+	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@2"), otherJid)
+	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@3"), otherJid)
+
+	p := prunerFactory.BuildSenderPruner(ctx, sender, sender)
+
+	p.Prune()
+
+	report := p.Report()
+
+	reportJSON, err := json.MarshalIndent(report, "", "  ")
+	require.NoError(ctx, err)
+	ctx.Logf("%s\n", string(reportJSON))
+
+	require.Equal(ctx, pruner.Done.String(), report.State)
+	require.Len(ctx, report.Completed, 1)
+	fsReport := report.Completed[0]
+	require.Equal(ctx, fs, fsReport.Filesystem)
+	require.Empty(ctx, fsReport.SkipReason)
+	require.Empty(ctx, fsReport.LastError)
+	expectDestroyList := []pruner.SnapshotReport{
+		{
+			Name:       "1",
+			Replicated: true,
+		},
+		{
+			Name:       "2",
+			Replicated: true,
+		},
+		{
+			Name:       "3",
+			Replicated: true,
+		},
+		{
+			Name:       "4",
+			Replicated: true,
+		},
+	}
+	for _, d := range fsReport.DestroyList {
+		d.Date = time.Time{}
+	}
+	require.Subset(ctx, fsReport.DestroyList, expectDestroyList)
+}
+
+
+
+func PrunerNoKeepNotReplicatedButKeepStepHold(ctx *platformtest.Context) {
+
+	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+        DESTROYROOT
+        CREATEROOT
+        +  "foo bar"
+        +  "foo bar@1"
+        +  "foo bar@2"
+        +  "foo bar@3"
+        +  "foo bar@4"
+        +  "foo bar@5"
+    `)
+
+	c, err := config.ParseConfigBytes([]byte(fmt.Sprintf(`
+jobs:
+- name: prunetest
+  type: push
+  filesystems: {
+      "%s/foo bar<": true
+  }
+  connect:
+    type: tcp
+    address: 255.255.255.255:255
+  snapshotting:
+    type: manual
+  pruning:
+    keep_sender:
+    - type: step_holds
+    - type: last_n
+      count: 1
+    keep_receiver:
+    - type: last_n
+      count: 2
+    `, ctx.RootDataset)))
+	require.NoError(ctx, err)
+
+	pushJob := c.Jobs[0].Ret.(*config.PushJob)
+
+	dummyHistVec := prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: "foo",
+		Subsystem: "foo",
+		Name:      "foo",
+		Help:      "foo",
+	}, []string{"foo"})
+
+	prunerFactory, err := pruner.NewPrunerFactory(pushJob.Pruning, dummyHistVec)
+	require.NoError(ctx, err)
+
+	senderJid := endpoint.MustMakeJobID("sender-job")
+
+	fsfilter, err := filters.DatasetMapFilterFromConfig(pushJob.Filesystems)
+	require.NoError(ctx, err)
+
+	sender := endpoint.NewSender(endpoint.SenderConfig{
+		FSF: fsfilter,
+		Encrypt: &zfs.NilBool{
+			B: false,
+		},
+        JobID: senderJid,
+	})
+
+	fs := ctx.RootDataset + "/foo bar"
+
+	// create a replication cursor to make pruning work at all
+	_, err = endpoint.CreateReplicationCursor(ctx, fs, fsversion(ctx, fs, "@2"), senderJid)
+	require.NoError(ctx, err)
+
+	// create step holds for the incremental @2->@3
+	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@2"), senderJid)
+	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@3"), senderJid)
+	// create step holds for another job
+	otherJid := endpoint.MustMakeJobID("other-job")
+	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@2"), otherJid)
+	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@3"), otherJid)
+
+	p := prunerFactory.BuildSenderPruner(ctx, sender, sender)
+
+	p.Prune()
+
+	report := p.Report()
+
+	reportJSON, err := json.MarshalIndent(report, "", "  ")
+	require.NoError(ctx, err)
+	ctx.Logf("%s\n", string(reportJSON))
+
+	require.Equal(ctx, pruner.Done.String(), report.State)
+	require.Len(ctx, report.Completed, 1)
+	fsReport := report.Completed[0]
+	require.Equal(ctx, fs, fsReport.Filesystem)
+	require.Empty(ctx, fsReport.SkipReason)
+	require.Empty(ctx, fsReport.LastError)
+	expectDestroyList := []pruner.SnapshotReport{
+		{
+			Name:       "1",
+			Replicated: true,
+		},
+		{
+			Name:       "4",
+			Replicated: true,
+		},
+	}
+	for _, d := range fsReport.DestroyList {
+		d.Date = time.Time{}
+	}
+	require.Subset(ctx, fsReport.DestroyList, expectDestroyList)
+}

platformtest/tests/pruner.v2.go

@@ -0,0 +1,137 @@
+package tests
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/kr/pretty"
+	"github.com/stretchr/testify/require"
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon/filters"
+	"github.com/zrepl/zrepl/daemon/pruner.v2"
+	"github.com/zrepl/zrepl/endpoint"
+	"github.com/zrepl/zrepl/platformtest"
+	"github.com/zrepl/zrepl/pruning"
+	"github.com/zrepl/zrepl/zfs"
+)
+
+func Pruner2NotReplicated(ctx *platformtest.Context) {
+
+	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+        DESTROYROOT
+        CREATEROOT
+        +  "foo bar"
+        +  "foo bar@1"
+        +  "foo bar@2"
+        +  "foo bar@3"
+        +  "foo bar@4"
+        +  "foo bar@5"
+	`)
+
+	fs := ctx.RootDataset + "/foo bar"
+	senderJid := endpoint.MustMakeJobID("sender-job")
+	otherJid1 := endpoint.MustMakeJobID("other-job-1")
+	otherJid2 := endpoint.MustMakeJobID("other-job-2")
+
+
+	// create step holds for the incremental @2->@3
+	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@2"), senderJid)
+	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@3"), senderJid)
+	// create step holds for other-job-1 @2 -> @3
+	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@2"), otherJid1)
+	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@3"), otherJid1)
+	// create step hold for other-job-2 @1 (will be pruned)
+	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@1"), otherJid2)
+
+	c, err := config.ParseConfigBytes([]byte(fmt.Sprintf(`
+jobs:
+- name: prunetest
+  type: push
+  filesystems: {
+      "%s/foo bar": true
+  }
+  connect:
+    type: tcp
+    address: 255.255.255.255:255
+  snapshotting:
+    type: manual
+  pruning:
+    keep_sender:
+    #- type: not_replicated
+    - type: step_holds
+    - type: last_n
+      count: 1
+    keep_receiver:
+    - type: last_n
+      count: 2
+    `, ctx.RootDataset)))
+	require.NoError(ctx, err)
+
+	pushJob := c.Jobs[0].Ret.(*config.PushJob)
+
+	require.NoError(ctx, err)
+
+	fsfilter, err := filters.DatasetMapFilterFromConfig(pushJob.Filesystems)
+	require.NoError(ctx, err)
+
+	matchedFilesystems, err := zfs.ZFSListMapping(ctx, fsfilter)
+	ctx.Logf("%s", pretty.Sprint(matchedFilesystems))
+	require.NoError(ctx, err)
+	require.Len(ctx, matchedFilesystems, 1)
+
+	sideSender := pruner.NewSideSender(senderJid)
+
+	keepRules, err := pruning.RulesFromConfig(senderJid, pushJob.Pruning.KeepSender)
+	require.NoError(ctx, err)
+	p := pruner.NewPruner(fsfilter, senderJid, sideSender, keepRules)
+
+	runDone := make(chan *pruner.Report)
+	go func() {
+		runDone <- p.Run(ctx)
+	}()
+
+	var report *pruner.Report
+	// concurrency stress
+out:
+	for {
+		select {
+		case <-time.After(10 * time.Millisecond):
+			p.Report(ctx)
+		case report = <-runDone:
+			break out
+		}
+	}
+	ctx.Logf("%s\n", pretty.Sprint(report))
+
+	reportJSON, err := json.MarshalIndent(report, "", "  ")
+	require.NoError(ctx, err)
+	ctx.Logf("%s\n", string(reportJSON))
+
+	ctx.FailNow()
+	// fs := ctx.RootDataset + "/foo bar"
+
+	// // create a replication cursor to make pruning work at all
+	// _, err = endpoint.CreateReplicationCursor(ctx, fs, fsversion(ctx, fs, "@2"), senderJid)
+	// require.NoError(ctx, err)
+
+	// p := prunerFactory.BuildSenderPruner(ctx, sender, sender)
+
+	// p.Prune()
+
+	// report := p.Report()
+
+	// require.Equal(ctx, pruner.Done.String(), report.State)
+	// require.Len(ctx, report.Completed, 1)
+	// fsReport := report.Completed[0]
+	// require.Equal(ctx, fs, fsReport.Filesystem)
+	// require.Empty(ctx, fsReport.SkipReason)
+	// require.Empty(ctx, fsReport.LastError)
+	// require.Len(ctx, fsReport.DestroyList, 1)
+	// require.Equal(ctx, fsReport.DestroyList[0], pruner.SnapshotReport{
+	// 	Name:       "1",
+	// 	Replicated: true,
+	// 	Date:       fsReport.DestroyList[0].Date,
+	// })
+
+}

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,396 @@
+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
+}
+
+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},
+		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,
+	}
+	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,
+	}
+	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(ctx *platformtest.Context) {
+
+	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}
+		},
+	}
+	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)
-	if err != nil {
-		panic(err)
+	checkCreateCursor := func(createErr error, c endpoint.Abstraction, references zfs.FilesystemVersion) {
+		assert.NoError(ctx, createErr)
+		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{
-	BatchDestroy,
-	UndestroyableSnapshotParsing,
-	GetNonexistent,
-	ReplicationCursor,
-	IdempotentHold,
-	IdempotentBookmark,
-	IdempotentDestroy,
-	ResumeTokenParsing,
-	ResumableRecvAndTokenHandling,
-	SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden,
-	SendArgsValidationResumeTokenEncryptionMismatchForbidden,
-	SendArgsValidationResumeTokenDifferentFilesystemForbidden,
-	ListFilesystemVersionsTypeFilteringAndPrefix,
-	ListFilesystemVersionsFilesystemNotExist,
-	ListFilesystemVersionsFilesystemNotExist,
-	ListFilesystemVersionsUserrefs,
-	ListFilesystemsNoFilter,
-}
+//go:generate ../../artifacts/generate-platform-test-list github.com/zrepl/zrepl/platformtest/tests

platformtest/tests/undestroyableSnapshotParsing.go

@@ -17,10 +17,12 @@ func UndestroyableSnapshotParsing(t *platformtest.Context) {
 		+  "foo bar@1 2 3"
 		+  "foo bar@4 5 6"
 		+  "foo bar@7 8 9"
+		+  "foo bar@10 11 12"
 		R  zfs hold zrepl_platformtest "${ROOTDS}/foo bar@4 5 6"
+		R  zfs hold zrepl_platformtest "${ROOTDS}/foo bar@7 8 9"
 	`)
 
-	err := zfs.ZFSDestroy(t, fmt.Sprintf("%s/foo bar@1 2 3,4 5 6,7 8 9", t.RootDataset))
+	err := zfs.ZFSDestroy(t, fmt.Sprintf("%s/foo bar@1 2 3,4 5 6,7 8 9,10 11 12", t.RootDataset))
 	if err == nil {
 		panic("expecting destroy error due to hold")
 	}
@@ -30,8 +32,9 @@ func UndestroyableSnapshotParsing(t *platformtest.Context) {
 		if dse.Filesystem != fmt.Sprintf("%s/foo bar", t.RootDataset) {
 			panic(dse.Filesystem)
 		}
-		require.Equal(t, []string{"4 5 6"}, dse.Undestroyable)
-		require.Equal(t, []string{"dataset is busy"}, dse.Reason)
+		expectUndestroyable := []string{"4 5 6", "7 8 9"}
+		require.Len(t, dse.Undestroyable, len(expectUndestroyable))
+		require.Subset(t, dse.Undestroyable, expectUndestroyable)
 	}
 
 }

pruning/keep_grid.go

@@ -66,24 +66,26 @@ type retentionGridAdaptor struct {
 	Snapshot
 }
 
+func (a retentionGridAdaptor) Date() time.Time { return a.Snapshot.GetCreation() }
+
 func (a retentionGridAdaptor) LessThan(b retentiongrid.Entry) bool {
 	return a.Date().Before(b.Date())
 }
 
 // Prune filters snapshots with the retention grid.
-func (p *KeepGrid) KeepRule(snaps []Snapshot) (destroyList []Snapshot) {
+func (p *KeepGrid) KeepRule(snaps []Snapshot) PruneSnapshotsResult {
 
-	snaps = filterSnapList(snaps, func(snapshot Snapshot) bool {
-		return p.re.MatchString(snapshot.Name())
+	reCandidates := partitionSnapList(snaps, func(snapshot Snapshot) bool {
+		return p.re.MatchString(snapshot.GetName())
 	})
-	if len(snaps) == 0 {
-		return nil
+	if len(reCandidates.Remove) == 0 {
+		return reCandidates
 	}
 
 	// Build adaptors for retention grid
 	adaptors := make([]retentiongrid.Entry, 0)
 	for i := range snaps {
-		adaptors = append(adaptors, retentionGridAdaptor{snaps[i]})
+		adaptors = append(adaptors, retentionGridAdaptor{reCandidates.Remove[i]})
 	}
 
 	// determine 'now' edge
@@ -93,12 +95,17 @@ func (p *KeepGrid) KeepRule(snaps []Snapshot) (destroyList []Snapshot) {
 	now := adaptors[len(adaptors)-1].Date()
 
 	// Evaluate retention grid
-	_, removea := p.retentionGrid.FitEntries(now, adaptors)
+	keepa, removea := p.retentionGrid.FitEntries(now, adaptors)
 
 	// Revert adaptors
-	destroyList = make([]Snapshot, len(removea))
+	destroyList := make([]Snapshot, len(removea))
 	for i := range removea {
 		destroyList[i] = removea[i].(retentionGridAdaptor).Snapshot
 	}
-	return destroyList
+	for _, a := range keepa {
+		reCandidates.Keep = append(reCandidates.Keep, a.(retentionGridAdaptor))
+	}
+	reCandidates.Remove = destroyList
+
+	return reCandidates
 }

pruning/keep_helpers.go

@@ -1,10 +1,13 @@
 package pruning
 
-func filterSnapList(snaps []Snapshot, predicate func(Snapshot) bool) []Snapshot {
-	r := make([]Snapshot, 0, len(snaps))
+func partitionSnapList(snaps []Snapshot, remove func(Snapshot) bool) (r PruneSnapshotsResult) {
+	r.Keep = make([]Snapshot, 0, len(snaps))
+	r.Remove = make([]Snapshot, 0, len(snaps))
 	for i := range snaps {
-		if predicate(snaps[i]) {
-			r = append(r, snaps[i])
+		if remove(snaps[i]) {
+			r.Remove = append(r.Remove, snaps[i])
+		} else {
+			r.Keep = append(r.Keep, snaps[i])
 		}
 	}
 	return r

pruning/keep_last_n.go

@@ -17,17 +17,17 @@ func NewKeepLastN(n int) (*KeepLastN, error) {
 	return &KeepLastN{n}, nil
 }
 
-func (k KeepLastN) KeepRule(snaps []Snapshot) (destroyList []Snapshot) {
+func (k KeepLastN) KeepRule(snaps []Snapshot) PruneSnapshotsResult {
 
 	if k.n > len(snaps) {
-		return []Snapshot{}
+		return PruneSnapshotsResult{Keep: snaps}
 	}
 
 	res := shallowCopySnapList(snaps)
 
 	sort.Slice(res, func(i, j int) bool {
-		return res[i].Date().After(res[j].Date())
+		return res[i].GetCreateTXG() > res[j].GetCreateTXG()
 	})
 
-	return res[k.n:]
+	return PruneSnapshotsResult{Remove: res[k.n:], Keep: res[:k.n]}
 }

pruning/keep_not_replicated.go

@@ -2,8 +2,8 @@ package pruning
 
 type KeepNotReplicated struct{}
 
-func (*KeepNotReplicated) KeepRule(snaps []Snapshot) (destroyList []Snapshot) {
-	return filterSnapList(snaps, func(snapshot Snapshot) bool {
+func (*KeepNotReplicated) KeepRule(snaps []Snapshot) PruneSnapshotsResult {
+	return partitionSnapList(snaps, func(snapshot Snapshot) bool {
 		return snapshot.Replicated()
 	})
 }

pruning/keep_regex.go

@@ -27,12 +27,12 @@ func MustKeepRegex(expr string, negate bool) *KeepRegex {
 	return k
 }
 
-func (k *KeepRegex) KeepRule(snaps []Snapshot) []Snapshot {
-	return filterSnapList(snaps, func(s Snapshot) bool {
+func (k *KeepRegex) KeepRule(snaps []Snapshot) PruneSnapshotsResult {
+	return partitionSnapList(snaps, func(s Snapshot) bool {
 		if k.negate {
-			return k.expr.FindStringIndex(s.Name()) != nil
+			return k.expr.FindStringIndex(s.GetName()) != nil
 		} else {
-			return k.expr.FindStringIndex(s.Name()) == nil
+			return k.expr.FindStringIndex(s.GetName()) == nil
 		}
 	})
 }

pruning/keep_step_holds.go

@@ -0,0 +1,44 @@
+package pruning
+
+import (
+	"github.com/pkg/errors"
+	"github.com/zrepl/zrepl/endpoint"
+)
+
+type KeepStepHolds struct {
+	keepJobIDs map[endpoint.JobID]bool
+}
+
+var _ KeepRule = (*KeepStepHolds)(nil)
+
+func NewKeepStepHolds(mainJobId endpoint.JobID, additionalJobIdsStrings []string) (_ *KeepStepHolds, err error) {
+	additionalJobIds := make(map[endpoint.JobID]bool, len(additionalJobIdsStrings))
+
+	mainJobId.MustValidate()
+	additionalJobIds[mainJobId] = true
+
+	for i := range additionalJobIdsStrings {
+		ajid, err := endpoint.MakeJobID(additionalJobIdsStrings[i])
+		if err != nil {
+			return nil, errors.WithMessagef(err, "cannot parse job id %q: %s", additionalJobIdsStrings[i])
+		}
+		if additionalJobIds[ajid] == true {
+			return nil, errors.Errorf("duplicate job id %q", ajid)
+		}
+	}
+	return &KeepStepHolds{additionalJobIds}, nil
+}
+
+func (h *KeepStepHolds) KeepRule(snaps []Snapshot) PruneSnapshotsResult {
+	return partitionSnapList(snaps, func(s Snapshot) bool {
+		holdingJobIDs := make(map[endpoint.JobID]bool)
+		for _, h := range s.StepHolds() {
+			holdingJobIDs[h.GetJobID()] = true
+		}
+		oneOrMoreOfOurJobIDsHoldsSnap := false
+		for kjid := range h.keepJobIDs {
+			oneOrMoreOfOurJobIDsHoldsSnap = oneOrMoreOfOurJobIDsHoldsSnap || holdingJobIDs[kjid]
+		}
+		return !oneOrMoreOfOurJobIDsHoldsSnap
+	})
+}

pruning/pruning.go

@@ -7,47 +7,93 @@ import (
 	"github.com/pkg/errors"
 
 	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/endpoint"
 )
 
 type KeepRule interface {
-	KeepRule(snaps []Snapshot) (destroyList []Snapshot)
+	KeepRule(snaps []Snapshot) PruneSnapshotsResult
 }
 
 type Snapshot interface {
-	Name() string
+	GetName() string
 	Replicated() bool
-	Date() time.Time
+	GetCreation() time.Time
+	GetCreateTXG() uint64
+	StepHolds() []StepHold
 }
 
-// The returned snapshot list is guaranteed to only contains elements of input parameter snaps
-func PruneSnapshots(snaps []Snapshot, keepRules []KeepRule) []Snapshot {
+type StepHold interface {
+	GetJobID() endpoint.JobID
+}
+
+type PruneSnapshotsResult struct {
+	Remove, Keep []Snapshot
+}
+
+// The returned snapshot results are a partition of the snaps argument.
+// That means than len(Remove) + len(Keep) == len(snaps)
+func PruneSnapshots(snapsI []Snapshot, keepRules []KeepRule) PruneSnapshotsResult {
 
 	if len(keepRules) == 0 {
-		return []Snapshot{}
+		return PruneSnapshotsResult{Remove: nil, Keep: snapsI}
+	}
+
+	type snapshot struct {
+		Snapshot
+		keepCount, removeCount int
+	}
+
+	// project down to snapshot
+	snaps := make([]Snapshot, len(snapsI))
+	for i := range snaps {
+		snaps[i] = &snapshot{snapsI[i], 0, 0}
 	}
 
-	remCount := make(map[Snapshot]int, len(snaps))
 	for _, r := range keepRules {
-		ruleRems := r.KeepRule(snaps)
-		for _, ruleRem := range ruleRems {
-			remCount[ruleRem]++
+
+		ruleImplCheckSet := make(map[Snapshot]int, len(snaps))
+		for _, s := range snaps {
+			ruleImplCheckSet[s] = ruleImplCheckSet[s] + 1
+		}
+
+		ruleResults := r.KeepRule(snaps)
+
+		for _, s := range snaps {
+			ruleImplCheckSet[s] = ruleImplCheckSet[s] - 1
+		}
+		for _, n := range ruleImplCheckSet {
+			if n != 0 {
+				panic(fmt.Sprintf("incorrect rule implementation: %T", r))
+			}
+		}
+
+		for _, s := range ruleResults.Remove {
+			s.(*snapshot).removeCount++
+		}
+		for _, s := range ruleResults.Keep {
+			s.(*snapshot).keepCount++
 		}
 	}
 
 	remove := make([]Snapshot, 0, len(snaps))
-	for snap, rc := range remCount {
-		if rc == len(keepRules) {
-			remove = append(remove, snap)
+	keep := make([]Snapshot, 0, len(snaps))
+	for _, sI := range snaps {
+		s := sI.(*snapshot)
+		if s.removeCount == len(keepRules) {
+			// all keep rules agree to remove the snap
+			remove = append(remove, s.Snapshot)
+		} else {
+			keep = append(keep, s.Snapshot)
 		}
 	}
 
-	return remove
+	return PruneSnapshotsResult{Remove: remove, Keep: keep}
 }
 
-func RulesFromConfig(in []config.PruningEnum) (rules []KeepRule, err error) {
+func RulesFromConfig(mainJobId endpoint.JobID, in []config.PruningEnum) (rules []KeepRule, err error) {
 	rules = make([]KeepRule, len(in))
 	for i := range in {
-		rules[i], err = RuleFromConfig(in[i])
+		rules[i], err = RuleFromConfig(mainJobId, in[i])
 		if err != nil {
 			return nil, errors.Wrapf(err, "cannot build rule #%d", i)
 		}
@@ -55,7 +101,7 @@ func RulesFromConfig(in []config.PruningEnum) (rules []KeepRule, err error) {
 	return rules, nil
 }
 
-func RuleFromConfig(in config.PruningEnum) (KeepRule, error) {
+func RuleFromConfig(mainJobId endpoint.JobID, in config.PruningEnum) (KeepRule, error) {
 	switch v := in.Ret.(type) {
 	case *config.PruneKeepNotReplicated:
 		return NewKeepNotReplicated(), nil
@@ -65,6 +111,8 @@ func RuleFromConfig(in config.PruningEnum) (KeepRule, error) {
 		return NewKeepRegex(v.Regex, v.Negate)
 	case *config.PruneGrid:
 		return NewKeepGrid(v)
+	case *config.PruneKeepStepHolds:
+		return NewKeepStepHolds(mainJobId, v.AdditionalJobIds)
 	default:
 		return nil, fmt.Errorf("unknown keep rule type %T", v)
 	}

replication/logic/pdu/pdu.pb.go

@@ -46,7 +46,36 @@ 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_2d84e8d7d278a80d, []int{0}
+}
+
+type SendAbstraction_SendAbstractionType int32
+
+const (
+	SendAbstraction_Undefined           SendAbstraction_SendAbstractionType = 0
+	SendAbstraction_ReplicationCursorV2 SendAbstraction_SendAbstractionType = 1
+	SendAbstraction_StepHold            SendAbstraction_SendAbstractionType = 2
+	SendAbstraction_StepBookmark        SendAbstraction_SendAbstractionType = 3
+)
+
+var SendAbstraction_SendAbstractionType_name = map[int32]string{
+	0: "Undefined",
+	1: "ReplicationCursorV2",
+	2: "StepHold",
+	3: "StepBookmark",
+}
+var SendAbstraction_SendAbstractionType_value = map[string]int32{
+	"Undefined":           0,
+	"ReplicationCursorV2": 1,
+	"StepHold":            2,
+	"StepBookmark":        3,
+}
+
+func (x SendAbstraction_SendAbstractionType) String() string {
+	return proto.EnumName(SendAbstraction_SendAbstractionType_name, int32(x))
+}
+func (SendAbstraction_SendAbstractionType) EnumDescriptor() ([]byte, []int) {
+	return fileDescriptor_pdu_2d84e8d7d278a80d, []int{5, 0}
 }
 
 type FilesystemVersion_VersionType int32
@@ -69,7 +98,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_2d84e8d7d278a80d, []int{6, 0}
 }
 
 type ListFilesystemReq struct {
@@ -82,7 +111,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_2d84e8d7d278a80d, []int{0}
 }
 func (m *ListFilesystemReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ListFilesystemReq.Unmarshal(m, b)
@@ -113,7 +142,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_2d84e8d7d278a80d, []int{1}
 }
 func (m *ListFilesystemRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ListFilesystemRes.Unmarshal(m, b)
@@ -154,7 +183,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_2d84e8d7d278a80d, []int{2}
 }
 func (m *Filesystem) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_Filesystem.Unmarshal(m, b)
@@ -213,7 +242,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_2d84e8d7d278a80d, []int{3}
 }
 func (m *ListFilesystemVersionsReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ListFilesystemVersionsReq.Unmarshal(m, b)
@@ -242,6 +271,7 @@ func (m *ListFilesystemVersionsReq) GetFilesystem() string {
 
 type ListFilesystemVersionsRes struct {
 	Versions             []*FilesystemVersion `protobuf:"bytes,1,rep,name=Versions,proto3" json:"Versions,omitempty"`
+	SendAbstractions     []*SendAbstraction   `protobuf:"bytes,2,rep,name=SendAbstractions,proto3" json:"SendAbstractions,omitempty"`
 	XXX_NoUnkeyedLiteral struct{}             `json:"-"`
 	XXX_unrecognized     []byte               `json:"-"`
 	XXX_sizecache        int32                `json:"-"`
@@ -251,7 +281,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_2d84e8d7d278a80d, []int{4}
 }
 func (m *ListFilesystemVersionsRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ListFilesystemVersionsRes.Unmarshal(m, b)
@@ -278,6 +308,67 @@ func (m *ListFilesystemVersionsRes) GetVersions() []*FilesystemVersion {
 	return nil
 }
 
+func (m *ListFilesystemVersionsRes) GetSendAbstractions() []*SendAbstraction {
+	if m != nil {
+		return m.SendAbstractions
+	}
+	return nil
+}
+
+type SendAbstraction struct {
+	Type                 SendAbstraction_SendAbstractionType `protobuf:"varint,1,opt,name=Type,proto3,enum=SendAbstraction_SendAbstractionType" json:"Type,omitempty"`
+	JobID                string                              `protobuf:"bytes,2,opt,name=JobID,proto3" json:"JobID,omitempty"`
+	Version              *FilesystemVersion                  `protobuf:"bytes,3,opt,name=Version,proto3" json:"Version,omitempty"`
+	XXX_NoUnkeyedLiteral struct{}                            `json:"-"`
+	XXX_unrecognized     []byte                              `json:"-"`
+	XXX_sizecache        int32                               `json:"-"`
+}
+
+func (m *SendAbstraction) Reset()         { *m = SendAbstraction{} }
+func (m *SendAbstraction) String() string { return proto.CompactTextString(m) }
+func (*SendAbstraction) ProtoMessage()    {}
+func (*SendAbstraction) Descriptor() ([]byte, []int) {
+	return fileDescriptor_pdu_2d84e8d7d278a80d, []int{5}
+}
+func (m *SendAbstraction) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_SendAbstraction.Unmarshal(m, b)
+}
+func (m *SendAbstraction) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_SendAbstraction.Marshal(b, m, deterministic)
+}
+func (dst *SendAbstraction) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_SendAbstraction.Merge(dst, src)
+}
+func (m *SendAbstraction) XXX_Size() int {
+	return xxx_messageInfo_SendAbstraction.Size(m)
+}
+func (m *SendAbstraction) XXX_DiscardUnknown() {
+	xxx_messageInfo_SendAbstraction.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_SendAbstraction proto.InternalMessageInfo
+
+func (m *SendAbstraction) GetType() SendAbstraction_SendAbstractionType {
+	if m != nil {
+		return m.Type
+	}
+	return SendAbstraction_Undefined
+}
+
+func (m *SendAbstraction) GetJobID() string {
+	if m != nil {
+		return m.JobID
+	}
+	return ""
+}
+
+func (m *SendAbstraction) GetVersion() *FilesystemVersion {
+	if m != nil {
+		return m.Version
+	}
+	return nil
+}
+
 type FilesystemVersion struct {
 	Type                 FilesystemVersion_VersionType `protobuf:"varint,1,opt,name=Type,proto3,enum=FilesystemVersion_VersionType" json:"Type,omitempty"`
 	Name                 string                        `protobuf:"bytes,2,opt,name=Name,proto3" json:"Name,omitempty"`
@@ -293,7 +384,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_2d84e8d7d278a80d, []int{6}
 }
 func (m *FilesystemVersion) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_FilesystemVersion.Unmarshal(m, b)
@@ -371,7 +462,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_2d84e8d7d278a80d, []int{7}
 }
 func (m *SendReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_SendReq.Unmarshal(m, b)
@@ -445,7 +536,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_2d84e8d7d278a80d, []int{8}
 }
 func (m *Property) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_Property.Unmarshal(m, b)
@@ -496,7 +587,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_2d84e8d7d278a80d, []int{9}
 }
 func (m *SendRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_SendRes.Unmarshal(m, b)
@@ -548,7 +639,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_2d84e8d7d278a80d, []int{10}
 }
 func (m *SendCompletedReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_SendCompletedReq.Unmarshal(m, b)
@@ -585,7 +676,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_2d84e8d7d278a80d, []int{11}
 }
 func (m *SendCompletedRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_SendCompletedRes.Unmarshal(m, b)
@@ -620,7 +711,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_2d84e8d7d278a80d, []int{12}
 }
 func (m *ReceiveReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ReceiveReq.Unmarshal(m, b)
@@ -671,7 +762,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_2d84e8d7d278a80d, []int{13}
 }
 func (m *ReceiveRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ReceiveRes.Unmarshal(m, b)
@@ -704,7 +795,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_2d84e8d7d278a80d, []int{14}
 }
 func (m *DestroySnapshotsReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_DestroySnapshotsReq.Unmarshal(m, b)
@@ -750,7 +841,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_2d84e8d7d278a80d, []int{15}
 }
 func (m *DestroySnapshotRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_DestroySnapshotRes.Unmarshal(m, b)
@@ -795,7 +886,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_2d84e8d7d278a80d, []int{16}
 }
 func (m *DestroySnapshotsRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_DestroySnapshotsRes.Unmarshal(m, b)
@@ -833,7 +924,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_2d84e8d7d278a80d, []int{17}
 }
 func (m *ReplicationCursorReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ReplicationCursorReq.Unmarshal(m, b)
@@ -874,7 +965,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_2d84e8d7d278a80d, []int{18}
 }
 func (m *ReplicationCursorRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ReplicationCursorRes.Unmarshal(m, b)
@@ -1010,7 +1101,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_2d84e8d7d278a80d, []int{19}
 }
 func (m *PingReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_PingReq.Unmarshal(m, b)
@@ -1049,7 +1140,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_2d84e8d7d278a80d, []int{20}
 }
 func (m *PingRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_PingRes.Unmarshal(m, b)
@@ -1076,97 +1167,13 @@ 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")
 	proto.RegisterType((*Filesystem)(nil), "Filesystem")
 	proto.RegisterType((*ListFilesystemVersionsReq)(nil), "ListFilesystemVersionsReq")
 	proto.RegisterType((*ListFilesystemVersionsRes)(nil), "ListFilesystemVersionsRes")
+	proto.RegisterType((*SendAbstraction)(nil), "SendAbstraction")
 	proto.RegisterType((*FilesystemVersion)(nil), "FilesystemVersion")
 	proto.RegisterType((*SendReq)(nil), "SendReq")
 	proto.RegisterType((*Property)(nil), "Property")
@@ -1182,9 +1189,8 @@ 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("SendAbstraction_SendAbstractionType", SendAbstraction_SendAbstractionType_name, SendAbstraction_SendAbstractionType_value)
 	proto.RegisterEnum("FilesystemVersion_VersionType", FilesystemVersion_VersionType_name, FilesystemVersion_VersionType_value)
 }
 
@@ -1206,7 +1212,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 +1276,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 +1284,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 +1398,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 +1426,72 @@ 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_2d84e8d7d278a80d) }
 
-var fileDescriptor_pdu_e59763dc61674a79 = []byte{
-	// 892 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x56, 0xdf, 0x6f, 0xdb, 0x36,
-	0x10, 0x8e, 0x6c, 0x39, 0x91, 0xcf, 0xe9, 0xea, 0x5c, 0xb2, 0x42, 0x13, 0xba, 0xce, 0xe3, 0x86,
-	0xc1, 0x0d, 0x30, 0x61, 0xc8, 0x7e, 0x60, 0xc3, 0x80, 0x02, 0x8d, 0x93, 0x34, 0xc5, 0xd6, 0xce,
-	0x60, 0xbc, 0x62, 0xe8, 0x9b, 0x6a, 0x1f, 0x12, 0x21, 0xb2, 0xa8, 0x90, 0xf4, 0x50, 0x6f, 0x7b,
-	0xda, 0xe3, 0xfe, 0xbd, 0xe5, 0x0f, 0x2a, 0x44, 0x4b, 0xb6, 0x6c, 0xc9, 0x89, 0x9f, 0xcc, 0xfb,
-	0x78, 0x14, 0xef, 0xbe, 0xfb, 0xee, 0x68, 0x68, 0x26, 0xa3, 0x89, 0x9f, 0x48, 0xa1, 0x05, 0xdb,
-	0x87, 0xbd, 0x5f, 0x43, 0xa5, 0xcf, 0xc2, 0x88, 0xd4, 0x54, 0x69, 0x1a, 0x73, 0xba, 0x61, 0xc7,
-	0x65, 0x50, 0xe1, 0xd7, 0xd0, 0x5a, 0x00, 0xca, 0xb5, 0x3a, 0xf5, 0x6e, 0xeb, 0xa8, 0xe5, 0x17,
-	0x9c, 0x8a, 0xfb, 0xec, 0x3f, 0x0b, 0x60, 0x61, 0x23, 0x82, 0xdd, 0x0f, 0xf4, 0x95, 0x6b, 0x75,
-	0xac, 0x6e, 0x93, 0x9b, 0x35, 0x76, 0xa0, 0xc5, 0x49, 0x4d, 0xc6, 0x34, 0x10, 0xd7, 0x14, 0xbb,
-	0x35, 0xb3, 0x55, 0x84, 0xf0, 0x4b, 0x78, 0xf0, 0x52, 0xf5, 0xa3, 0x60, 0x48, 0x57, 0x22, 0x1a,
-	0x91, 0x74, 0xeb, 0x1d, 0xab, 0xeb, 0xf0, 0x65, 0x30, 0xfd, 0xce, 0x4b, 0x75, 0x1a, 0x0f, 0xe5,
-	0x34, 0xd1, 0x34, 0x72, 0x6d, 0xe3, 0x53, 0x84, 0xd8, 0xcf, 0xf0, 0xc9, 0x72, 0x42, 0x6f, 0x48,
-	0xaa, 0x50, 0xc4, 0x8a, 0xd3, 0x0d, 0x3e, 0x29, 0x06, 0x9a, 0x05, 0x58, 0x40, 0xd8, 0x2f, 0xeb,
-	0x0f, 0x2b, 0xf4, 0xc1, 0xc9, 0xcd, 0x8c, 0x12, 0xf4, 0x4b, 0x9e, 0x7c, 0xee, 0xc3, 0x6e, 0x2d,
-	0xd8, 0x2b, 0xed, 0xe3, 0x11, 0xd8, 0x83, 0x69, 0x42, 0xe6, 0xf2, 0x8f, 0x8e, 0x9e, 0x94, 0xbf,
-	0xe0, 0x67, 0xbf, 0xa9, 0x17, 0x37, 0xbe, 0x29, 0xa3, 0xaf, 0x83, 0x31, 0x65, 0xb4, 0x99, 0x75,
-	0x8a, 0xbd, 0x98, 0x84, 0x23, 0x43, 0x93, 0xcd, 0xcd, 0x1a, 0x1f, 0x43, 0xb3, 0x27, 0x29, 0xd0,
-	0x34, 0xf8, 0xe3, 0x85, 0xe1, 0xc6, 0xe6, 0x0b, 0x00, 0x3d, 0x70, 0x8c, 0x11, 0x8a, 0xd8, 0x6d,
-	0x98, 0x2f, 0xcd, 0x6d, 0xf6, 0x14, 0x5a, 0x85, 0x6b, 0x71, 0x17, 0x9c, 0x8b, 0x38, 0x48, 0xd4,
-	0x95, 0xd0, 0xed, 0xad, 0xd4, 0x3a, 0x16, 0xe2, 0x7a, 0x1c, 0xc8, 0xeb, 0xb6, 0xc5, 0xfe, 0xb7,
-	0x60, 0xe7, 0x82, 0xe2, 0xd1, 0x06, 0x7c, 0xe2, 0x57, 0x60, 0x9f, 0x49, 0x31, 0x36, 0x81, 0x57,
-	0xd3, 0x65, 0xf6, 0x91, 0x41, 0x6d, 0x20, 0x4c, 0x2a, 0xd5, 0x5e, 0xb5, 0x81, 0x58, 0x95, 0x90,
-	0x5d, 0x96, 0x10, 0x83, 0xe6, 0x42, 0x1a, 0x0d, 0xc3, 0xaf, 0xed, 0x0f, 0x64, 0xc8, 0x17, 0x30,
-	0x3e, 0x82, 0xed, 0x13, 0x39, 0xe5, 0x93, 0xd8, 0xdd, 0x36, 0xda, 0xc9, 0x2c, 0xf6, 0x1d, 0x38,
-	0x7d, 0x29, 0x12, 0x92, 0x7a, 0x3a, 0xa7, 0xdb, 0x2a, 0xd0, 0x7d, 0x00, 0x8d, 0x37, 0x41, 0x34,
-	0xc9, 0x6b, 0x30, 0x33, 0xd8, 0xbf, 0x73, 0x2e, 0x14, 0x76, 0xe1, 0xe1, 0xef, 0x8a, 0x46, 0xab,
-	0x32, 0x77, 0xf8, 0x2a, 0x8c, 0x0c, 0x76, 0x4f, 0xdf, 0x27, 0x34, 0xd4, 0x34, 0xba, 0x08, 0xff,
-	0x22, 0x93, 0x77, 0x9d, 0x2f, 0x61, 0xf8, 0x14, 0x20, 0x8b, 0x27, 0x24, 0xe5, 0xda, 0x46, 0x6e,
-	0x4d, 0x3f, 0x0f, 0x91, 0x17, 0x36, 0xd9, 0x33, 0x68, 0xa7, 0x31, 0xf4, 0xc4, 0x38, 0x89, 0x48,
-	0x93, 0x29, 0xcc, 0x21, 0xb4, 0x7e, 0x93, 0xe1, 0x65, 0x18, 0x07, 0x11, 0xa7, 0x9b, 0x8c, 0x7f,
-	0xc7, 0xcf, 0xea, 0xc6, 0x8b, 0x9b, 0x0c, 0x4b, 0xe7, 0x15, 0xfb, 0x07, 0x80, 0xd3, 0x90, 0xc2,
-	0x3f, 0x69, 0x93, 0x32, 0xcf, 0xca, 0x57, 0xbb, 0xb3, 0x7c, 0x87, 0xd0, 0xee, 0x45, 0x14, 0xc8,
-	0x22, 0x3f, 0xb3, 0x16, 0x2f, 0xe1, 0x6c, 0xb7, 0x70, 0xbb, 0x62, 0x97, 0xb0, 0x7f, 0x42, 0x4a,
-	0x4b, 0x31, 0xcd, 0x35, 0xb9, 0x49, 0x2f, 0xe3, 0x37, 0xd0, 0x9c, 0xfb, 0xbb, 0xb5, 0xb5, 0xfd,
-	0xba, 0x70, 0x62, 0x6f, 0x01, 0x57, 0x2e, 0xca, 0xda, 0x3e, 0x37, 0xcd, 0x2d, 0x6b, 0xda, 0x3e,
-	0xf7, 0x49, 0x95, 0x72, 0x2a, 0xa5, 0x90, 0xb9, 0x52, 0x8c, 0xc1, 0x4e, 0xaa, 0x92, 0x48, 0x27,
-	0xed, 0x4e, 0x9a, 0x78, 0xa4, 0xf3, 0x91, 0xb2, 0xef, 0x97, 0x43, 0xe0, 0xb9, 0x0f, 0xfb, 0x01,
-	0x0e, 0x38, 0x25, 0x51, 0x38, 0x34, 0x5d, 0xdb, 0x9b, 0x48, 0x25, 0xe4, 0x26, 0x73, 0x6d, 0x50,
-	0x79, 0x4e, 0xe1, 0x41, 0x36, 0x44, 0xd2, 0x13, 0xf6, 0xf9, 0xd6, 0x7c, 0x8c, 0x38, 0xaf, 0x85,
-	0xa6, 0xf7, 0xa1, 0xd2, 0x33, 0x09, 0x9f, 0x6f, 0xf1, 0x39, 0x72, 0xec, 0xc0, 0xf6, 0x2c, 0x1c,
-	0xf6, 0x05, 0xec, 0xf4, 0xc3, 0xf8, 0x32, 0x0d, 0xc0, 0x85, 0x9d, 0x57, 0xa4, 0x54, 0x70, 0x99,
-	0x77, 0x4d, 0x6e, 0xb2, 0x4f, 0x73, 0x27, 0x95, 0xf6, 0xd5, 0xe9, 0xf0, 0x4a, 0xe4, 0x7d, 0x95,
-	0xae, 0xd9, 0xdf, 0xf0, 0xd9, 0x79, 0x18, 0xeb, 0x57, 0x42, 0xe9, 0xb4, 0xe4, 0xb1, 0xee, 0x89,
-	0xf1, 0x58, 0xc4, 0xcf, 0xe3, 0x21, 0x29, 0xbd, 0x51, 0x72, 0xf8, 0x23, 0x3c, 0x48, 0xf5, 0x4b,
-	0x32, 0xab, 0xc5, 0x1d, 0x42, 0x5c, 0x76, 0x64, 0x9f, 0xdf, 0x77, 0xb9, 0x3a, 0xec, 0x42, 0x7d,
-	0x20, 0xc3, 0x74, 0x04, 0x9e, 0x88, 0x58, 0xf7, 0x02, 0x49, 0xed, 0x2d, 0x6c, 0x42, 0xe3, 0x2c,
-	0x88, 0x14, 0xb5, 0x2d, 0x74, 0xc0, 0x1e, 0xc8, 0x09, 0xb5, 0x6b, 0x47, 0xb7, 0xf5, 0x74, 0x40,
-	0xcd, 0x49, 0x46, 0x0f, 0xec, 0x34, 0x71, 0x74, 0xfc, 0x8c, 0x24, 0x2f, 0x5f, 0x29, 0xfc, 0x09,
-	0x1e, 0x2e, 0xbf, 0x33, 0x0a, 0xd1, 0x2f, 0x3d, 0xce, 0x5e, 0x19, 0x53, 0xd8, 0x87, 0x47, 0xd5,
-	0x4f, 0x14, 0x7a, 0xfe, 0xda, 0x87, 0xcf, 0x5b, 0xbf, 0xa7, 0xf0, 0x19, 0xb4, 0x57, 0xa5, 0x89,
-	0x07, 0x7e, 0x45, 0xcb, 0x79, 0x55, 0xa8, 0xc2, 0xe7, 0xb0, 0x57, 0x12, 0x17, 0x7e, 0xec, 0x57,
-	0x09, 0xd5, 0xab, 0x84, 0x15, 0x7e, 0x3f, 0x2b, 0xe1, 0x7c, 0x04, 0xe1, 0x9e, 0xbf, 0x3a, 0xd2,
-	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,
+var fileDescriptor_pdu_2d84e8d7d278a80d = []byte{
+	// 940 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x56, 0x6d, 0x6f, 0xe3, 0xc4,
+	0x13, 0xaf, 0x13, 0xa7, 0x75, 0x26, 0xed, 0xbf, 0xee, 0x24, 0xff, 0x23, 0x44, 0x70, 0xaa, 0x96,
+	0x13, 0xca, 0x55, 0x60, 0xa1, 0xf0, 0x20, 0x10, 0xe8, 0xa4, 0x6b, 0xd2, 0x5e, 0x8b, 0xe0, 0x88,
+	0xb6, 0xb9, 0x0a, 0x9d, 0xc4, 0x0b, 0x37, 0x1e, 0x5a, 0xab, 0x8e, 0xd7, 0xdd, 0x75, 0xd0, 0x05,
+	0xf1, 0x8a, 0x77, 0xf0, 0xf5, 0xe0, 0x73, 0x20, 0x3e, 0x02, 0xf2, 0xc6, 0x4e, 0x1c, 0xdb, 0x3d,
+	0xf5, 0x55, 0x76, 0x7e, 0x33, 0xe3, 0x9d, 0xc7, 0xdf, 0x06, 0x9a, 0x91, 0x37, 0x77, 0x22, 0x29,
+	0x62, 0xc1, 0xda, 0x70, 0xf0, 0x9d, 0xaf, 0xe2, 0x53, 0x3f, 0x20, 0xb5, 0x50, 0x31, 0xcd, 0x38,
+	0xdd, 0xb1, 0xe3, 0x32, 0xa8, 0xf0, 0x63, 0x68, 0xad, 0x01, 0xd5, 0x35, 0x0e, 0xeb, 0xfd, 0xd6,
+	0xa0, 0xe5, 0xe4, 0x8c, 0xf2, 0x7a, 0xf6, 0xa7, 0x01, 0xb0, 0x96, 0x11, 0xc1, 0x1c, 0xbb, 0xf1,
+	0x4d, 0xd7, 0x38, 0x34, 0xfa, 0x4d, 0xae, 0xcf, 0x78, 0x08, 0x2d, 0x4e, 0x6a, 0x3e, 0xa3, 0x89,
+	0xb8, 0xa5, 0xb0, 0x5b, 0xd3, 0xaa, 0x3c, 0x84, 0x4f, 0x60, 0xef, 0x5c, 0x8d, 0x03, 0x77, 0x4a,
+	0x37, 0x22, 0xf0, 0x48, 0x76, 0xeb, 0x87, 0x46, 0xdf, 0xe2, 0x9b, 0x60, 0xf2, 0x9d, 0x73, 0x75,
+	0x12, 0x4e, 0xe5, 0x22, 0x8a, 0xc9, 0xeb, 0x9a, 0xda, 0x26, 0x0f, 0xb1, 0xaf, 0xe1, 0xdd, 0xcd,
+	0x84, 0x2e, 0x49, 0x2a, 0x5f, 0x84, 0x8a, 0xd3, 0x1d, 0x3e, 0xce, 0x07, 0x9a, 0x06, 0x98, 0x43,
+	0xd8, 0x1f, 0xc6, 0xfd, 0xde, 0x0a, 0x1d, 0xb0, 0x32, 0x31, 0xad, 0x09, 0x3a, 0x25, 0x4b, 0xbe,
+	0xb2, 0xc1, 0x6f, 0xc0, 0xbe, 0xa0, 0xd0, 0x7b, 0x7e, 0xa5, 0x62, 0xe9, 0x4e, 0x63, 0xed, 0x57,
+	0xd3, 0x7e, 0xb6, 0x53, 0x50, 0xf0, 0x92, 0x25, 0xfb, 0xc7, 0x80, 0xfd, 0x02, 0x88, 0x5f, 0x82,
+	0x39, 0x59, 0x44, 0xa4, 0x23, 0xff, 0xdf, 0xe0, 0x49, 0xf1, 0x2b, 0x45, 0x39, 0xb1, 0xe5, 0xda,
+	0x03, 0x3b, 0xd0, 0xf8, 0x56, 0x5c, 0x9d, 0x8f, 0xd2, 0xd2, 0x2f, 0x05, 0xfc, 0x08, 0x76, 0xd2,
+	0x68, 0x75, 0xb9, 0xab, 0x13, 0xca, 0x4c, 0xd8, 0x4f, 0xd0, 0xae, 0xb8, 0x00, 0xf7, 0xa0, 0xf9,
+	0x2a, 0xf4, 0xe8, 0x67, 0x3f, 0x24, 0xcf, 0xde, 0xc2, 0x77, 0xa0, 0xcd, 0x29, 0x0a, 0xfc, 0xa9,
+	0x9b, 0x58, 0x0c, 0xe7, 0x52, 0x09, 0x79, 0x39, 0xb0, 0x0d, 0xdc, 0x05, 0xeb, 0x22, 0xa6, 0xe8,
+	0x4c, 0x04, 0x9e, 0x5d, 0x43, 0x1b, 0x76, 0x13, 0xe9, 0x58, 0x88, 0xdb, 0x99, 0x2b, 0x6f, 0xed,
+	0x3a, 0xfb, 0xdb, 0x80, 0x83, 0xd2, 0xed, 0x38, 0xd8, 0x48, 0xf9, 0x71, 0x39, 0x3e, 0x27, 0xfd,
+	0xcd, 0x25, 0x8b, 0x60, 0xbe, 0x74, 0x67, 0x94, 0xe6, 0xaa, 0xcf, 0x09, 0xf6, 0x62, 0xee, 0x7b,
+	0x3a, 0x4f, 0x93, 0xeb, 0x33, 0xbe, 0x07, 0xcd, 0xa1, 0x24, 0x37, 0xa6, 0xc9, 0x8f, 0x2f, 0xf4,
+	0x2c, 0x99, 0x7c, 0x0d, 0x60, 0x0f, 0x2c, 0x2d, 0x24, 0xd5, 0x69, 0xe8, 0x2f, 0xad, 0x64, 0xf6,
+	0x14, 0x5a, 0xb9, 0x6b, 0x75, 0x6a, 0xa1, 0x1b, 0xa9, 0x1b, 0x11, 0xdb, 0x5b, 0x89, 0xb4, 0x4a,
+	0xcb, 0x60, 0x7f, 0x19, 0xb0, 0x93, 0x94, 0xed, 0x01, 0xf3, 0x87, 0x1f, 0x82, 0x79, 0x2a, 0xc5,
+	0x4c, 0x07, 0x5e, 0xdd, 0x0c, 0xad, 0x47, 0x06, 0xb5, 0x89, 0x78, 0x4b, 0xcb, 0x6a, 0x13, 0x51,
+	0x5c, 0x39, 0xb3, 0xbc, 0x72, 0x0c, 0x9a, 0xeb, 0x55, 0x6a, 0xe8, 0xfa, 0x9a, 0xce, 0x44, 0xfa,
+	0x7c, 0x0d, 0xe3, 0x23, 0xd8, 0x1e, 0xc9, 0x05, 0x9f, 0x87, 0xdd, 0x6d, 0xbd, 0x6b, 0xa9, 0xc4,
+	0x3e, 0x03, 0x6b, 0x2c, 0x45, 0x44, 0x32, 0x5e, 0xac, 0xca, 0x6d, 0xe4, 0xca, 0xdd, 0x81, 0xc6,
+	0xa5, 0x1b, 0xcc, 0xb3, 0x1e, 0x2c, 0x05, 0xf6, 0xfb, 0xaa, 0x16, 0x0a, 0xfb, 0xb0, 0xff, 0x4a,
+	0x91, 0x57, 0xa4, 0x05, 0x8b, 0x17, 0x61, 0x64, 0xb0, 0x7b, 0xf2, 0x26, 0xa2, 0x69, 0x4c, 0xde,
+	0x85, 0xff, 0x2b, 0xe9, 0xbc, 0xeb, 0x7c, 0x03, 0xc3, 0xa7, 0x00, 0x69, 0x3c, 0x3e, 0xa9, 0xae,
+	0xa9, 0xb7, 0xac, 0xe9, 0x64, 0x21, 0xf2, 0x9c, 0x92, 0x3d, 0x5b, 0xae, 0xe5, 0x50, 0xcc, 0xa2,
+	0x80, 0x62, 0xd2, 0x8d, 0x39, 0x82, 0xd6, 0x0f, 0xd2, 0xbf, 0xf6, 0x43, 0x37, 0xe0, 0x74, 0x97,
+	0xd6, 0xdf, 0x72, 0xd2, 0xbe, 0xf1, 0xbc, 0x92, 0x61, 0xc9, 0x5f, 0xb1, 0xdf, 0x00, 0x38, 0x4d,
+	0xc9, 0xff, 0x85, 0x1e, 0xd2, 0xe6, 0x65, 0xfb, 0x6a, 0x6f, 0x6d, 0xdf, 0x11, 0xd8, 0xc3, 0x80,
+	0x5c, 0x99, 0xaf, 0xcf, 0x92, 0x12, 0x4b, 0x38, 0xdb, 0xcd, 0xdd, 0xae, 0xd8, 0x35, 0xb4, 0x47,
+	0xa4, 0x62, 0x29, 0x16, 0xd9, 0x4c, 0x3e, 0x84, 0xfb, 0xf0, 0x13, 0x68, 0xae, 0xec, 0x53, 0x9a,
+	0xaa, 0x8a, 0x6d, 0x6d, 0xc4, 0x5e, 0x03, 0x16, 0x2e, 0x4a, 0x59, 0x32, 0x13, 0xf5, 0x2d, 0xf7,
+	0xb0, 0x64, 0x66, 0x93, 0x4c, 0xca, 0x89, 0x94, 0x42, 0x66, 0x93, 0xa2, 0x05, 0x36, 0xaa, 0x4a,
+	0x22, 0x79, 0x99, 0x76, 0x92, 0xc4, 0x83, 0x38, 0x63, 0xe0, 0xb6, 0x53, 0x0e, 0x81, 0x67, 0x36,
+	0xec, 0x0b, 0xe8, 0x94, 0xb8, 0xe8, 0x21, 0xef, 0xc0, 0xa4, 0xd2, 0x4f, 0x61, 0x27, 0x25, 0x91,
+	0xc4, 0xc3, 0x3c, 0xdb, 0x5a, 0xd1, 0x88, 0xf5, 0x52, 0xc4, 0xf4, 0xc6, 0x57, 0xf1, 0x72, 0x84,
+	0xcf, 0xb6, 0xf8, 0x0a, 0x39, 0xb6, 0x60, 0x7b, 0x19, 0x0e, 0xfb, 0x00, 0x76, 0xc6, 0x7e, 0x78,
+	0x9d, 0x04, 0xd0, 0x85, 0x9d, 0xef, 0x49, 0x29, 0xf7, 0x3a, 0xdb, 0x9a, 0x4c, 0x64, 0xef, 0x67,
+	0x46, 0x2a, 0xd9, 0xab, 0x93, 0xe9, 0x8d, 0xc8, 0xf6, 0x2a, 0x39, 0x1f, 0xf5, 0xa1, 0x3e, 0x91,
+	0x7e, 0x42, 0x31, 0x23, 0x11, 0xc6, 0x43, 0x57, 0x92, 0xbd, 0x85, 0x4d, 0x68, 0x9c, 0xba, 0x81,
+	0x22, 0xdb, 0x40, 0x0b, 0xcc, 0x89, 0x9c, 0x93, 0x5d, 0x1b, 0xfc, 0x5b, 0x4b, 0x08, 0x60, 0x95,
+	0x04, 0xf6, 0xc0, 0x4c, 0x3e, 0x8c, 0x96, 0x93, 0x06, 0xd1, 0xcb, 0x4e, 0x0a, 0xbf, 0x82, 0xfd,
+	0xcd, 0x67, 0x4f, 0x21, 0x3a, 0xa5, 0x3f, 0x0b, 0xbd, 0x32, 0xa6, 0x70, 0x0c, 0x8f, 0xaa, 0x5f,
+	0x4c, 0xec, 0x39, 0xf7, 0x3e, 0xc4, 0xbd, 0xfb, 0x75, 0x0a, 0x9f, 0x81, 0x5d, 0x6c, 0x3d, 0x76,
+	0x9c, 0x8a, 0x91, 0xee, 0x55, 0xa1, 0x0a, 0x9f, 0xc3, 0x41, 0xa9, 0x79, 0xf8, 0x7f, 0xa7, 0x6a,
+	0x10, 0x7a, 0x95, 0xb0, 0xc2, 0xcf, 0x61, 0x6f, 0x63, 0xc5, 0xf1, 0xc0, 0x29, 0x52, 0x46, 0xaf,
+	0x04, 0xa9, 0xe3, 0xc6, 0xeb, 0x7a, 0xe4, 0xcd, 0xaf, 0xb6, 0xf5, 0xff, 0xad, 0x4f, 0xff, 0x0b,
+	0x00, 0x00, 0xff, 0xff, 0x0a, 0xa0, 0x3f, 0xc2, 0x7c, 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
 }
 
@@ -26,7 +25,22 @@ message Filesystem {
 
 message ListFilesystemVersionsReq { string Filesystem = 1; }
 
-message ListFilesystemVersionsRes { repeated FilesystemVersion Versions = 1; }
+message ListFilesystemVersionsRes {
+  repeated FilesystemVersion Versions = 1;
+  repeated SendAbstraction SendAbstractions = 2;
+}
+
+message SendAbstraction {
+  enum SendAbstractionType {
+    Undefined = 0;
+    ReplicationCursorV2 = 1;
+    StepHold = 2;
+    StepBookmark = 3;
+  };
+  SendAbstractionType Type = 1;
+  string JobID = 2;
+  FilesystemVersion Version = 3;
+}
 
 message FilesystemVersion {
   enum VersionType {
@@ -81,9 +95,7 @@ message SendRes {
   repeated Property Properties = 4;
 }
 
-message SendCompletedReq {
-  SendReq OriginalReq = 2;
-}
+message SendCompletedReq { SendReq OriginalReq = 2; }
 
 message SendCompletedRes {}
 
@@ -99,7 +111,7 @@ message ReceiveReq {
 message ReceiveRes {}
 
 message DestroySnapshotsReq {
-  string Filesystem = 1;
+string Filesystem = 1;
   // Path to filesystem, snapshot or bookmark to be destroyed
   repeated FilesystemVersion Snapshots = 2;
 }
@@ -126,19 +138,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()
-		s.parent.promBytesReplicated.Add(float64(s.byteCounter.Count()))
+		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
-	clientIdent, err := f.clientMap.Get(clientIP)
+	clientAddr := &net.IPAddr{
+		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) {
-	path, err := zfs.NewDatasetPath(in)
+func ValidateClientIdentity(in string) error {
+	err := zfs.ComponentNamecheck(in)
 	if err != nil {
-		return err
-	}
-	if path.Length() != 1 {
-		return errors.New("client identity must be a single path component (not empty, no '/')")
+		return errors.Wrap(err, "client identity must be usable as a single dataset path component")
 	}
 	return nil
 }

zfs/versions.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"io"
 	"strconv"
 	"strings"
 	"time"
@@ -116,6 +115,15 @@ func (v FilesystemVersion) RelName() string {
 }
 func (v FilesystemVersion) String() string { return v.RelName() }
 
+func (v FilesystemVersion) GetCreation() time.Time { return v.Creation }
+
+// 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 +232,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 {
@@ -1498,6 +1500,30 @@ func tryParseDestroySnapshotsError(arg string, stderr []byte) *DestroySnapshotsE
 	}
 }
 
+type ErrDestroySnapshotDatasetIsBusy struct {
+	*DestroySnapshotsError
+	Name string
+}
+
+var _ error = (*ErrDestroySnapshotDatasetIsBusy)(nil)
+
+func tryErrDestroySnapshotDatasetIsBusy(arg string, stderr []byte) *ErrDestroySnapshotDatasetIsBusy {
+	dsne := tryParseDestroySnapshotsError(arg, stderr)
+	if dsne == nil {
+		return nil
+	}
+	if len(dsne.Reason) != 1 {
+		return nil
+	}
+	if dsne.Reason[0] == "dataset is busy" {
+		return &ErrDestroySnapshotDatasetIsBusy{
+			DestroySnapshotsError: dsne,
+			Name:                  dsne.Undestroyable[0],
+		}
+	}
+	return nil
+}
+
 func ZFSDestroy(ctx context.Context, arg string) (err error) {
 
 	var dstype, filesystem string
@@ -1531,6 +1557,8 @@ func ZFSDestroy(ctx context.Context, arg string) (err error) {
 			err = &DatasetDoesNotExist{arg}
 		} else if dsNotExistErr := tryDatasetDoesNotExist(filesystem, stdio); dsNotExistErr != nil {
 			err = dsNotExistErr
+		} else if dsBusy := tryErrDestroySnapshotDatasetIsBusy(arg, stdio); dsBusy != nil {
+			err = dsBusy
 		} else if dserr := tryParseDestroySnapshotsError(arg, stdio); dserr != nil {
 			err = dserr
 		}
@@ -1595,22 +1623,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 +1657,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 {
-				debug("bookmark: %q %q was idempotent: {snap,book}guid %d == %d", snapname, bookmarkname, v.GUID, bookGuid)
-				return nil
+			if v.Guid == bookGuid {
+				debug("bookmark: %q %q was idempotent: {snap,book}guid %d == %d", snapname, bookmarkname, v.Guid, bookGuid)
+				return bm, nil
 			}
-			return &BookmarkExists{
-				fs: fs, bookmarkOrigin: v, bookmark: bookmark,
+			return bm, &BookmarkExists{
+				fs: fs, bookmarkOrigin: v.ToSendArgVersion(), bookmark: bookmark,
 				zfsMsg:   string(stdio),
 				bookGuid: bookGuid,
 			}
 
 		} else {
-			return &ZFSError{
+			return bm, &ZFSError{
 				Stderr:  stdio,
 				WaitErr: err,
 			}
@@ -1647,8 +1685,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) {