[#289 ] zfs: workaround for OpenZFS 0.7 dry send info with zero estimated size

fixes #289
[#345 ] fix broken identification of parent-fs for initial replication ordering
2020-07-26 20:32:35 +02:00 · 2020-07-26 20:32:35 +02:00 · 2020-07-26 20:32:35 +02:00 · 2020-07-26 20:32:35 +02:00 · 2020-07-26 20:32:35 +02:00 · 2020-07-26 20:32:35 +02:00
227 changed files with 12161 additions and 4418 deletions
@@ -6,6 +6,7 @@ workflows:
      - build-1.11
      - build-1.12
      - build-1.13
+      - build-1.14
      - build-latest
      - test-build-in-docker
 jobs:
@@ -59,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
@@ -91,11 +92,23 @@ 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"'/"}'

+      # kick off binary packaging workflow
+      - run:
+          shell: /bin/bash -eo pipefail
+          command: |
+            if [ -n "$CIRCLE_PR_NUMBER" ]; then # CIRCLE_PR_NUMBER is guaranteed to be only present in forked PRs (external)
+              echo "Forked PR detected. Sry, can't trust you with credentials."
+              exit 0
+            fi
+            set -u # from now on
+            GITHUB_ACCESS_TOKEN="$ZREPL_DEBIAN_BINARYPACKAGIN_TRIGGER_BUILD_GITHUB_TOKEN" .circleci/trigger_debian_binary_packaging_workflow.bash "$CIRCLE_SHA1" "${CIRCLE_JOB##build-}"
+
  build-1.11:
    <<: *build-latest
    docker:
@@ -111,6 +124,11 @@ jobs:
    docker:
    - image: circleci/golang:1.13

+  build-1.14:
+    <<: *build-latest
+    docker:
+    - image: circleci/golang:1.14
+
  # this job tries to mimic the build-in-docker instructions
  # given in docs/installation.rst
  #
@@ -0,0 +1,10 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+COMMIT="$1"
+GO_VERSION="$2"
+
+curl -v -X POST https://api.github.com/repos/zrepl/debian-binary-packaging/dispatches \
+    -H 'Accept: application/vnd.github.v3+json' \
+    -H "Authorization: token $GITHUB_ACCESS_TOKEN" \
+    --data '{"event_type": "push", "client_payload": { "zrepl_main_repo_commit": "'"$COMMIT"'", "go_version": "'"$GO_VERSION"'" }}'
@@ -1,3 +1,4 @@
+github: problame
 patreon: zrepl
 liberapay: zrepl
 custom: https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=R5QSXJVYHGX96
@@ -23,9 +23,12 @@ GOHOSTARCH ?= $(shell bash -c 'source <($(GO) env) && echo "$$GOHOSTARCH"')
 GO_ENV_VARS := GO111MODULE=on
 GO_LDFLAGS := "-X github.com/zrepl/zrepl/version.zreplVersion=$(_ZREPL_VERSION)"
 GO_MOD_READONLY := -mod=readonly
-GO_BUILDFLAGS := $(GO_MOD_READONLY)
+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
@@ -44,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
@@ -65,6 +68,7 @@ wrapup-and-checksum:
 		-acf $(NOARCH_TARBALL) \
 		$(ARTIFACTDIR)/docs/html \
 		$(ARTIFACTDIR)/bash_completion \
+		$(ARTIFACTDIR)/_zrepl.zsh_completion \
 		$(ARTIFACTDIR)/go_env.txt \
 		dist \
 		config/samples
@@ -97,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
@@ -120,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
@@ -148,18 +159,44 @@ 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)/go_env.txt docs docs-clean
+.PHONY: noarch $(ARTIFACTDIR)/bash_completion $(ARTIFACTDIR)/_zrepl.zsh_completion $(ARTIFACTDIR)/go_env.txt docs docs-clean


 $(ARTIFACTDIR):
@@ -167,12 +204,16 @@ $(ARTIFACTDIR):
 $(ARTIFACTDIR)/docs: $(ARTIFACTDIR)
 	mkdir -p "$@"

-noarch: $(ARTIFACTDIR)/bash_completion $(ARTIFACTDIR)/go_env.txt docs
+noarch: $(ARTIFACTDIR)/bash_completion $(ARTIFACTDIR)/_zrepl.zsh_completion $(ARTIFACTDIR)/go_env.txt docs
 	# pass

 $(ARTIFACTDIR)/bash_completion:
 	$(MAKE) zrepl-bin GOOS=$(GOHOSTOS) GOARCH=$(GOHOSTARCH)
-	artifacts/zrepl-$(GOHOSTOS)-$(GOHOSTARCH) bashcomp "$@"
+	artifacts/zrepl-$(GOHOSTOS)-$(GOHOSTARCH) gencompletion bash "$@"
+
+$(ARTIFACTDIR)/_zrepl.zsh_completion:
+	$(MAKE) zrepl-bin GOOS=$(GOHOSTOS) GOARCH=$(GOHOSTARCH)
+	artifacts/zrepl-$(GOHOSTOS)-$(GOHOSTARCH) gencompletion zsh "$@"

 $(ARTIFACTDIR)/go_env.txt:
 	$(GO_ENV_VARS) $(GO) env > $@
@@ -1,8 +1,10 @@
 [![GitHub license](https://img.shields.io/github/license/zrepl/zrepl.svg)](https://github.com/zrepl/zrepl/blob/master/LICENSE)
 [![Language: Go](https://img.shields.io/badge/language-Go-6ad7e5.svg)](https://golang.org/)
 [![User Docs](https://img.shields.io/badge/docs-web-blue.svg)](https://zrepl.github.io)
+[![Donate via Patreon](https://img.shields.io/endpoint.svg?url=https%3A%2F%2Fshieldsio-patreon.herokuapp.com%2Fzrepl%2Fpledges&style=flat&color=yellow)](https://www.patreon.com/zrepl)
+[![Donate via GitHub Sponsors](https://img.shields.io/static/v1?label=Sponsor&message=%E2%9D%A4&logo=GitHub&style=flat&color=yellow)](https://github.com/sponsors/problame)
+[![Donate via Liberapay](https://img.shields.io/liberapay/patrons/zrepl.svg?logo=liberapay)](https://liberapay.com/zrepl/donate)
 [![Donate via PayPal](https://img.shields.io/badge/donate-paypal-yellow.svg)](https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=R5QSXJVYHGX96)
-[![Donate via Liberapay](https://img.shields.io/liberapay/receives/zrepl.svg?logo=liberapay)](https://liberapay.com/zrepl/donate)
 [![Twitter](https://img.shields.io/twitter/url/https/github.com/zrepl/zrepl.svg?style=social)](https://twitter.com/intent/tweet?text=Wow:&url=https%3A%2F%2Fgithub.com%2Fzrepl%2Fzrepl)

 # zrepl
@@ -23,6 +25,7 @@ zrepl is a one-stop ZFS backup & replication solution.
   If so, think of an expressive configuration example.
 2. Think of at least one use case that generalizes from your concrete application.
 3. Open an issue on GitHub with example conf & use case attached.
+4. **Optional**: [Post a bounty](https://www.bountysource.com/teams/zrepl) on the issue, or [contact Christian Schwarz](https://cschwarz.com) for contract work.

 The above does not apply if you already implemented everything.
 Check out the *Coding Workflow* section below for details.
@@ -37,7 +40,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
@@ -45,12 +48,17 @@ 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`.

-### Overall Architecture
+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`.

-The application architecture is documented as part of the user docs in the *Implementation* section (`docs/content/impl`).
-Make sure to develop an understanding how zrepl is typically used by studying the user docs first.
+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

@@ -62,14 +70,15 @@ Make sure to develop an understanding how zrepl is typically used by studying th
 │   └── samples
 ├── daemon                  # the implementation of `zrepl daemon` subcommand
 │   ├── filters
+│   ├── hooks               # snapshot hooks
 │   ├── job                 # job implementations
 │   ├── logging             # logging outlets + formatters
 │   ├── nethelpers
 │   ├── prometheus
 │   ├── pruner              # pruner implementation
 │   ├── snapper             # snapshotter implementation
-├── docs                    # sphinx-based documentation
 ├── dist                    # supplemental material for users & package maintainers
+├── docs                    # sphinx-based documentation
 │   ├── **/*.rst            # documentation in reStructuredText
 │   ├── sphinxconf
 │   │   └── conf.py         # sphinx config (see commit 445a280 why its not in docs/)
@@ -78,6 +87,7 @@ Make sure to develop an understanding how zrepl is typically used by studying th
 │   └── public_git          # checkout of zrepl.github.io managed by above shell script
 ├── endpoint                # implementation of replication endpoints (=> package replication)
 ├── logger                  # our own logger package
+├── platformtest            # test suite for our zfs abstractions (error classification, etc)
 ├── pruning                 # pruning rules (the logic, not the actual execution)
 │   └── retentiongrid
 ├── replication
@@ -92,14 +102,13 @@ Make sure to develop an understanding how zrepl is typically used by studying th
 │   ├── transportmux        # TCP connecter and listener used to split control & data traffic
 │   └── versionhandshake    # replication protocol version handshake perfomed on newly established connections
 ├── tlsconf                 # abstraction for Go TLS server + client config
-├── transport               # transports implementation
+├── transport               # transport implementations
 │   ├── fromconfig
 │   ├── local
 │   ├── ssh
 │   ├── tcp
 │   └── tls
 ├── util
-├── vendor                  # managed by dep
 ├── version                 # abstraction for versions (filled during build by Makefile)
 └── zfs                     # zfs(8) wrappers
 ```
@@ -134,3 +143,15 @@ There will not be a big refactoring (an attempt was made, but it's destroying to

 However, new contributions & patches should fix naming without further notice in the commit message.

+### RPC debugging
+
+Optionally, there are various RPC-related environment variables, that if set to something != `""` will produce additional debug output on stderr:
+
+https://github.com/zrepl/zrepl/blob/master/rpc/rpc_debug.go#L11
+
+https://github.com/zrepl/zrepl/blob/master/rpc/dataconn/dataconn_debug.go#L11
+
+https://github.com/zrepl/zrepl/blob/master/rpc/dataconn/stream/stream_debug.go#L11
+
+https://github.com/zrepl/zrepl/blob/master/rpc/dataconn/heartbeatconn/heartbeatconn_debug.go#L11
+
@@ -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
@@ -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=
@@ -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"
 )
@@ -1,11 +1,13 @@
 package cli

 import (
+	"context"
 	"fmt"
 	"os"

 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
+	"github.com/zrepl/zrepl/daemon/logging/trace"

 	"github.com/zrepl/zrepl/config"
 )
@@ -19,29 +21,55 @@ var rootCmd = &cobra.Command{
 	Short: "One-stop ZFS replication solution",
 }

-var bashcompCmd = &cobra.Command{
-	Use:   "bashcomp path/to/out/file",
-	Short: "generate bash completions",
-	Run: func(cmd *cobra.Command, args []string) {
-		if len(args) != 1 {
-			fmt.Fprintf(os.Stderr, "specify exactly one positional agument\n")
-			err := cmd.Usage()
-			if err != nil {
-				panic(err)
-			}
-			os.Exit(1)
-		}
-		if err := rootCmd.GenBashCompletionFile(args[0]); err != nil {
-			fmt.Fprintf(os.Stderr, "error generating bash completion: %s", err)
-			os.Exit(1)
-		}
+func init() {
+	rootCmd.PersistentFlags().StringVar(&rootArgs.configPath, "config", "", "config file path")
+}
+
+var genCompletionCmd = &cobra.Command{
+	Use:   "gencompletion",
+	Short: "generate shell auto-completions",
+}
+
+type completionCmdInfo struct {
+	genFunc func(outpath string) error
+	help    string
+}
+
+var completionCmdMap = map[string]completionCmdInfo{
+	"zsh": {
+		rootCmd.GenZshCompletionFile,
+		"  save to file `_zrepl` in your zsh's $fpath",
+	},
+	"bash": {
+		rootCmd.GenBashCompletionFile,
+		"  save to a path and source that path in your .bashrc",
 	},
-	Hidden: true,
 }

 func init() {
-	rootCmd.PersistentFlags().StringVar(&rootArgs.configPath, "config", "", "config file path")
-	rootCmd.AddCommand(bashcompCmd)
+	for sh, info := range completionCmdMap {
+		sh, info := sh, info
+		genCompletionCmd.AddCommand(&cobra.Command{
+			Use:     fmt.Sprintf("%s path/to/out/file", sh),
+			Short:   fmt.Sprintf("generate %s completions", sh),
+			Example: info.help,
+			Run: func(cmd *cobra.Command, args []string) {
+				if len(args) != 1 {
+					fmt.Fprintf(os.Stderr, "specify exactly one positional agument\n")
+					err := cmd.Usage()
+					if err != nil {
+						panic(err)
+					}
+					os.Exit(1)
+				}
+				if err := info.genFunc(args[0]); err != nil {
+					fmt.Fprintf(os.Stderr, "error generating %s completion: %s", sh, err)
+					os.Exit(1)
+				}
+			},
+		})
+	}
+	rootCmd.AddCommand(genCompletionCmd)
 }

 type Subcommand struct {
@@ -49,7 +77,7 @@ type Subcommand struct {
 	Short            string
 	Example          string
 	NoRequireConfig  bool
-	Run              func(subcommand *Subcommand, args []string) error
+	Run              func(ctx context.Context, subcommand *Subcommand, args []string) error
 	SetupFlags       func(f *pflag.FlagSet)
 	SetupSubcommands func() []*Subcommand

@@ -70,7 +98,11 @@ func (s *Subcommand) Config() *config.Config {

 func (s *Subcommand) run(cmd *cobra.Command, args []string) {
 	s.tryParseConfig()
-	err := s.Run(s, args)
+	ctx := context.Background()
+	endTask := trace.WithTaskFromStackUpdateCtx(&ctx)
+	defer endTask()
+	err := s.Run(ctx, s, args)
+	endTask()
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "%s\n", err)
 		os.Exit(1)
@@ -1,6 +1,7 @@
 package client

 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"os"
@@ -29,7 +30,7 @@ var ConfigcheckCmd = &cli.Subcommand{
 		f.StringVar(&configcheckArgs.format, "format", "", "dump parsed config object [pretty|yaml|json]")
 		f.StringVar(&configcheckArgs.what, "what", "all", "what to print [all|config|jobs|logging]")
 	},
-	Run: func(subcommand *cli.Subcommand, args []string) error {
+	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
 		formatMap := map[string]func(interface{}){
 			"": func(i interface{}) {},
 			"pretty": func(i interface{}) {
@@ -1,87 +0,0 @@
-package client
-
-import (
-	"context"
-	"encoding/json"
-	"fmt"
-	"os"
-	"strings"
-
-	"github.com/pkg/errors"
-
-	"github.com/zrepl/zrepl/cli"
-	"github.com/zrepl/zrepl/daemon/filters"
-	"github.com/zrepl/zrepl/endpoint"
-	"github.com/zrepl/zrepl/zfs"
-)
-
-var (
-	HoldsCmd = &cli.Subcommand{
-		Use:   "holds",
-		Short: "manage holds & step bookmarks",
-		SetupSubcommands: func() []*cli.Subcommand {
-			return holdsList
-		},
-	}
-)
-
-var holdsList = []*cli.Subcommand{
-	&cli.Subcommand{
-		Use:             "list [FSFILTER]",
-		Run:             doHoldsList,
-		NoRequireConfig: true,
-		Short: `
-FSFILTER SYNTAX:
-representation of a 'filesystems' filter statement on the command line
-		`,
-	},
-}
-
-func fsfilterFromCliArg(arg string) (zfs.DatasetFilter, error) {
-	mappings := strings.Split(arg, ",")
-	f := filters.NewDatasetMapFilter(len(mappings), true)
-	for _, m := range mappings {
-		thisMappingErr := fmt.Errorf("expecting comma-separated list of <dataset-pattern>:<ok|!> pairs, got %q", m)
-		lhsrhs := strings.SplitN(m, ":", 2)
-		if len(lhsrhs) != 2 {
-			return nil, thisMappingErr
-		}
-		err := f.Add(lhsrhs[0], lhsrhs[1])
-		if err != nil {
-			return nil, fmt.Errorf("%s: %s", thisMappingErr, err)
-		}
-	}
-	return f.AsFilter(), nil
-}
-
-func doHoldsList(sc *cli.Subcommand, args []string) error {
-	var err error
-	ctx := context.Background()
-
-	if len(args) > 1 {
-		return errors.New("this subcommand takes at most one argument")
-	}
-
-	var filter zfs.DatasetFilter
-	if len(args) == 0 {
-		filter = zfs.NoFilter()
-	} else {
-		filter, err = fsfilterFromCliArg(args[0])
-		if err != nil {
-			return errors.Wrap(err, "cannot parse filesystem filter args")
-		}
-	}
-
-	listing, err := endpoint.ListZFSHoldsAndBookmarks(ctx, filter)
-	if err != nil {
-		return err // context clear by invocation of command
-	}
-
-	enc := json.NewEncoder(os.Stdout)
-	enc.SetIndent("  ", "  ")
-	if err := enc.Encode(listing); err != nil {
-		panic(err)
-	}
-
-	return nil
-}
@@ -48,14 +48,13 @@ var migratePlaceholder0_1Args struct {
 	dryRun bool
 }

-func doMigratePlaceholder0_1(sc *cli.Subcommand, args []string) error {
+func doMigratePlaceholder0_1(ctx context.Context, sc *cli.Subcommand, args []string) error {
 	if len(args) != 0 {
 		return fmt.Errorf("migration does not take arguments, got %v", args)
 	}

 	cfg := sc.Config()

-	ctx := context.Background()
 	allFSS, err := zfs.ZFSListMapping(ctx, zfs.NoFilter())
 	if err != nil {
 		return errors.Wrap(err, "cannot list filesystems")
@@ -99,7 +98,7 @@ func doMigratePlaceholder0_1(sc *cli.Subcommand, args []string) error {
 		}
 		for _, fs := range wi.fss {
 			fmt.Printf("\t%q ... ", fs.ToString())
-			r, err := zfs.ZFSMigrateHashBasedPlaceholderToCurrent(fs, migratePlaceholder0_1Args.dryRun)
+			r, err := zfs.ZFSMigrateHashBasedPlaceholderToCurrent(ctx, fs, migratePlaceholder0_1Args.dryRun)
 			if err != nil {
 				fmt.Printf("error: %s\n", err)
 			} else if !r.NeedsModification {
@@ -124,7 +123,7 @@ var fail = color.New(color.FgRed)

 var migrateReplicationCursorSkipSentinel = fmt.Errorf("skipping this filesystem")

-func doMigrateReplicationCursor(sc *cli.Subcommand, args []string) error {
+func doMigrateReplicationCursor(ctx context.Context, sc *cli.Subcommand, args []string) error {
 	if len(args) != 0 {
 		return fmt.Errorf("migration does not take arguments, got %v", args)
 	}
@@ -137,8 +136,6 @@ func doMigrateReplicationCursor(sc *cli.Subcommand, args []string) error {
 		return fmt.Errorf("exiting migration after error")
 	}

-	ctx := context.Background()
-
 	v1cursorJobs := make([]job.Job, 0, len(cfg.Jobs))
 	for i, j := range cfg.Jobs {
 		if jobs[i].Name() != j.Name() {
@@ -165,7 +162,7 @@ func doMigrateReplicationCursor(sc *cli.Subcommand, args []string) error {
 	var hadError bool
 	for _, fs := range fss {

-		bold.Printf("INSPECT FILESYTEM %q\n", fs.ToString())
+		bold.Printf("INSPECT FILESYSTEM %q\n", fs.ToString())

 		err := doMigrateReplicationCursorFS(ctx, v1cursorJobs, fs)
 		if err == migrateReplicationCursorSkipSentinel {
@@ -211,17 +208,15 @@ func doMigrateReplicationCursorFS(ctx context.Context, v1CursorJobs []job.Job, f
 	}
 	fmt.Printf("identified owning job %q\n", owningJob.Name())

-	versions, err := zfs.ZFSListFilesystemVersions(fs, nil)
+	bookmarks, err := zfs.ZFSListFilesystemVersions(ctx, fs, zfs.ListFilesystemVersionsOptions{
+		Types: zfs.Bookmarks,
+	})
 	if err != nil {
 		return errors.Wrapf(err, "list filesystem versions of %q", fs.ToString())
 	}

 	var oldCursor *zfs.FilesystemVersion
-	for i, fsv := range versions {
-		if fsv.Type != zfs.Bookmark {
-			continue
-		}
-
+	for i, fsv := range bookmarks {
 		_, _, err := endpoint.ParseReplicationCursorBookmarkName(fsv.ToAbsPath(fs))
 		if err != endpoint.ErrV1ReplicationCursor {
 			continue
@@ -232,7 +227,7 @@ func doMigrateReplicationCursorFS(ctx context.Context, v1CursorJobs []job.Job, f
 			return errors.Wrap(err, "multiple filesystem versions identified as v1 replication cursors")
 		}

-		oldCursor = &versions[i]
+		oldCursor = &bookmarks[i]

 	}

@@ -264,7 +259,7 @@ func doMigrateReplicationCursorFS(ctx context.Context, v1CursorJobs []job.Job, f
 	if migrateReplicationCursorArgs.dryRun {
 		succ.Printf("DRY RUN\n")
 	} else {
-		if err := zfs.ZFSDestroyFilesystemVersion(fs, oldCursor); err != nil {
+		if err := zfs.ZFSDestroyFilesystemVersion(ctx, fs, oldCursor); err != nil {
 			return err
 		}
 	}
@@ -1,67 +1,10 @@
 package client

-import (
-	"errors"
-	"log"
-	"os"
-
-	"github.com/zrepl/zrepl/cli"
-	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon"
-)
-
-var pprofArgs struct {
-	daemon.PprofServerControlMsg
-}
+import "github.com/zrepl/zrepl/cli"

 var PprofCmd = &cli.Subcommand{
-	Use:   "pprof off | [on TCP_LISTEN_ADDRESS]",
-	Short: "start a http server exposing go-tool-compatible profiling endpoints at TCP_LISTEN_ADDRESS",
-	Run: func(subcommand *cli.Subcommand, args []string) error {
-		if len(args) < 1 {
-			goto enargs
-		}
-		switch args[0] {
-		case "on":
-			pprofArgs.Run = true
-			if len(args) != 2 {
-				return errors.New("must specify TCP_LISTEN_ADDRESS as second positional argument")
-			}
-			pprofArgs.HttpListenAddress = args[1]
-		case "off":
-			if len(args) != 1 {
-				goto enargs
-			}
-			pprofArgs.Run = false
-		}
-
-		RunPProf(subcommand.Config())
-		return nil
-	enargs:
-		return errors.New("invalid number of positional arguments")
-
+	Use: "pprof",
+	SetupSubcommands: func() []*cli.Subcommand {
+		return []*cli.Subcommand{PprofListenCmd, pprofActivityTraceCmd}
 	},
 }
-
-func RunPProf(conf *config.Config) {
-	log := log.New(os.Stderr, "", 0)
-
-	die := func() {
-		log.Printf("exiting after error")
-		os.Exit(1)
-	}
-
-	log.Printf("connecting to zrepl daemon")
-
-	httpc, err := controlHttpClient(conf.Global.Control.SockPath)
-	if err != nil {
-		log.Printf("error creating http client: %s", err)
-		die()
-	}
-	err = jsonRequestResponse(httpc, daemon.ControlJobEndpointPProf, pprofArgs.PprofServerControlMsg, struct{}{})
-	if err != nil {
-		log.Printf("error sending control message: %s", err)
-		die()
-	}
-	log.Printf("finished")
-}
@@ -0,0 +1,44 @@
+package client
+
+import (
+	"context"
+	"io"
+	"log"
+	"os"
+
+	"golang.org/x/net/websocket"
+
+	"github.com/zrepl/zrepl/cli"
+)
+
+var pprofActivityTraceCmd = &cli.Subcommand{
+	Use:   "activity-trace ZREPL_PPROF_HOST:ZREPL_PPROF_PORT",
+	Short: "attach to zrepl daemon with activated pprof listener and dump an activity-trace to stdout",
+	Run:   runPProfActivityTrace,
+}
+
+func runPProfActivityTrace(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
+	log := log.New(os.Stderr, "", 0)
+
+	die := func() {
+		log.Printf("exiting after error")
+		os.Exit(1)
+	}
+
+	if len(args) != 1 {
+		log.Printf("exactly one positional argument is required")
+		die()
+	}
+
+	url := "ws://" + args[0] + "/debug/zrepl/activity-trace" // FIXME dont' repeat that
+
+	log.Printf("attaching to activity trace stream %s", url)
+	ws, err := websocket.Dial(url, "", url)
+	if err != nil {
+		log.Printf("error: %s", err)
+		die()
+	}
+
+	_, err = io.Copy(os.Stdout, ws)
+	return err
+}
@@ -0,0 +1,68 @@
+package client
+
+import (
+	"context"
+	"errors"
+	"log"
+	"os"
+
+	"github.com/zrepl/zrepl/cli"
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon"
+)
+
+var pprofListenCmd struct {
+	daemon.PprofServerControlMsg
+}
+
+var PprofListenCmd = &cli.Subcommand{
+	Use:   "listen off | [on TCP_LISTEN_ADDRESS]",
+	Short: "start a http server exposing go-tool-compatible profiling endpoints at TCP_LISTEN_ADDRESS",
+	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
+		if len(args) < 1 {
+			goto enargs
+		}
+		switch args[0] {
+		case "on":
+			pprofListenCmd.Run = true
+			if len(args) != 2 {
+				return errors.New("must specify TCP_LISTEN_ADDRESS as second positional argument")
+			}
+			pprofListenCmd.HttpListenAddress = args[1]
+		case "off":
+			if len(args) != 1 {
+				goto enargs
+			}
+			pprofListenCmd.Run = false
+		}
+
+		RunPProf(subcommand.Config())
+		return nil
+	enargs:
+		return errors.New("invalid number of positional arguments")
+
+	},
+}
+
+func RunPProf(conf *config.Config) {
+	log := log.New(os.Stderr, "", 0)
+
+	die := func() {
+		log.Printf("exiting after error")
+		os.Exit(1)
+	}
+
+	log.Printf("connecting to zrepl daemon")
+
+	httpc, err := controlHttpClient(conf.Global.Control.SockPath)
+	if err != nil {
+		log.Printf("error creating http client: %s", err)
+		die()
+	}
+	err = jsonRequestResponse(httpc, daemon.ControlJobEndpointPProf, pprofListenCmd.PprofServerControlMsg, struct{}{})
+	if err != nil {
+		log.Printf("error sending control message: %s", err)
+		die()
+	}
+	log.Printf("finished")
+}
@@ -1,6 +1,8 @@
 package client

 import (
+	"context"
+
 	"github.com/pkg/errors"

 	"github.com/zrepl/zrepl/cli"
@@ -9,21 +11,16 @@ import (
 )

 var SignalCmd = &cli.Subcommand{
-	Use:   "signal [wakeup|reset] JOB [DATA]",
+	Use:   "signal [wakeup|reset] JOB",
 	Short: "wake up a job from wait state or abort its current invocation",
-	Run: func(subcommand *cli.Subcommand, args []string) error {
+	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
 		return runSignalCmd(subcommand.Config(), args)
 	},
 }

 func runSignalCmd(config *config.Config, args []string) error {
-	if len(args) < 2 || len(args) > 3 {
-		return errors.Errorf("Expected 2 arguments: [wakeup|reset|set-concurrency] JOB [DATA]")
-	}
-
-	var data string
-	if len(args) == 3 {
-		data = args[2]
+	if len(args) != 2 {
+		return errors.Errorf("Expected 2 arguments: [wakeup|reset] JOB")
 	}

 	httpc, err := controlHttpClient(config.Global.Control.SockPath)
@@ -35,11 +32,9 @@ func runSignalCmd(config *config.Config, args []string) error {
 		struct {
 			Name string
 			Op   string
-			Data string
 		}{
 			Name: args[1],
 			Op:   args[0],
-			Data: data,
 		},
 		struct{}{},
 	)
@@ -1,6 +1,7 @@
 package client

 import (
+	"context"
 	"fmt"
 	"io"
 	"math"
@@ -11,7 +12,7 @@ import (
 	"sync"
 	"time"

-	// tcell is the termbox-compatbile library for abstracting away escape sequences, etc.
+	// tcell is the termbox-compatible library for abstracting away escape sequences, etc.
 	// as of tcell#252, the number of default distributed terminals is relatively limited
 	// additional terminal definitions can be included via side-effect import
 	// See https://github.com/gdamore/tcell/blob/master/terminfo/base/base.go
@@ -81,7 +82,7 @@ type tui struct {
 	indent int

 	lock   sync.Mutex //For report and error
-	report map[string]job.Status
+	report map[string]*job.Status
 	err    error

 	jobFilter string
@@ -180,7 +181,7 @@ var StatusCmd = &cli.Subcommand{
 	Run: runStatus,
 }

-func runStatus(s *cli.Subcommand, args []string) error {
+func runStatus(ctx context.Context, s *cli.Subcommand, args []string) error {
 	httpc, err := controlHttpClient(s.Config().Global.Control.SockPath)
 	if err != nil {
 		return err
@@ -219,7 +220,7 @@ func runStatus(s *cli.Subcommand, args []string) error {
 	defer termbox.Close()

 	update := func() {
-		m := make(map[string]job.Status)
+		var m daemon.Status

 		err2 := jsonRequestResponse(httpc, daemon.ControlJobEndpointStatus,
 			struct{}{},
@@ -228,7 +229,7 @@ func runStatus(s *cli.Subcommand, args []string) error {

 		t.lock.Lock()
 		t.err = err2
-		t.report = m
+		t.report = m.Jobs
 		t.lock.Unlock()
 		t.draw()
 	}
@@ -264,7 +265,7 @@ loop:

 }

-func (t *tui) getReplicationProgresHistory(jobName string) *bytesProgressHistory {
+func (t *tui) getReplicationProgressHistory(jobName string) *bytesProgressHistory {
 	p, ok := t.replicationProgress[jobName]
 	if !ok {
 		p = &bytesProgressHistory{}
@@ -329,7 +330,7 @@ func (t *tui) draw() {
 				t.printf("Replication:")
 				t.newline()
 				t.addIndent(1)
-				t.renderReplicationReport(activeStatus.Replication, t.getReplicationProgresHistory(k))
+				t.renderReplicationReport(activeStatus.Replication, t.getReplicationProgressHistory(k))
 				t.addIndent(-1)

 				t.printf("Pruning Sender:")
@@ -697,7 +698,7 @@ func rightPad(str string, length int, pad string) string {

 var arrowPositions = `>\|/`

-// changeCount = 0 indicates stall / no progresss
+// changeCount = 0 indicates stall / no progress
 func (t *tui) drawBar(length int, bytes, totalBytes int64, changeCount int) {
 	var completedLength int
 	if totalBytes > 0 {
@@ -17,7 +17,7 @@ import (
 var StdinserverCmd = &cli.Subcommand{
 	Use:   "stdinserver CLIENT_IDENTITY",
 	Short: "stdinserver transport mode (started from authorized_keys file as forced command)",
-	Run: func(subcommand *cli.Subcommand, args []string) error {
+	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
 		return runStdinserver(subcommand.Config(), args)
 	},
 }
@@ -39,7 +39,7 @@ var testFilter = &cli.Subcommand{
 	Run: runTestFilterCmd,
 }

-func runTestFilterCmd(subcommand *cli.Subcommand, args []string) error {
+func runTestFilterCmd(ctx context.Context, subcommand *cli.Subcommand, args []string) error {

 	if testFilterArgs.job == "" {
 		return fmt.Errorf("must specify --job flag")
@@ -60,6 +60,8 @@ func runTestFilterCmd(subcommand *cli.Subcommand, args []string) error {
 		confFilter = j.Filesystems
 	case *config.PushJob:
 		confFilter = j.Filesystems
+	case *config.SnapJob:
+		confFilter = j.Filesystems
 	default:
 		return fmt.Errorf("job type %T does not have filesystems filter", j)
 	}
@@ -73,7 +75,7 @@ func runTestFilterCmd(subcommand *cli.Subcommand, args []string) error {
 	if testFilterArgs.input != "" {
 		fsnames = []string{testFilterArgs.input}
 	} else {
-		out, err := zfs.ZFSList([]string{"name"})
+		out, err := zfs.ZFSList(ctx, []string{"name"})
 		if err != nil {
 			return fmt.Errorf("could not list ZFS filesystems: %s", err)
 		}
@@ -134,13 +136,13 @@ var testPlaceholder = &cli.Subcommand{
 	Run: runTestPlaceholder,
 }

-func runTestPlaceholder(subcommand *cli.Subcommand, args []string) error {
+func runTestPlaceholder(ctx context.Context, subcommand *cli.Subcommand, args []string) error {

 	var checkDPs []*zfs.DatasetPath

 	// all actions first
 	if testPlaceholderArgs.all {
-		out, err := zfs.ZFSList([]string{"name"})
+		out, err := zfs.ZFSList(ctx, []string{"name"})
 		if err != nil {
 			return errors.Wrap(err, "could not list ZFS filesystems")
 		}
@@ -164,7 +166,7 @@ func runTestPlaceholder(subcommand *cli.Subcommand, args []string) error {

 	fmt.Printf("IS_PLACEHOLDER\tDATASET\tzrepl:placeholder\n")
 	for _, dp := range checkDPs {
-		ph, err := zfs.ZFSGetFilesystemPlaceholderState(dp)
+		ph, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, dp)
 		if err != nil {
 			return errors.Wrap(err, "cannot get placeholder state")
 		}
@@ -193,11 +195,11 @@ var testDecodeResumeToken = &cli.Subcommand{
 	Run: runTestDecodeResumeTokenCmd,
 }

-func runTestDecodeResumeTokenCmd(subcommand *cli.Subcommand, args []string) error {
+func runTestDecodeResumeTokenCmd(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
 	if testDecodeResumeTokenArgs.token == "" {
 		return fmt.Errorf("token argument must be specified")
 	}
-	token, err := zfs.ParseResumeToken(context.Background(), testDecodeResumeTokenArgs.token)
+	token, err := zfs.ParseResumeToken(ctx, testDecodeResumeTokenArgs.token)
 	if err != nil {
 		return err
 	}
@@ -1,6 +1,7 @@
 package client

 import (
+	"context"
 	"fmt"
 	"os"

@@ -25,7 +26,7 @@ var VersionCmd = &cli.Subcommand{
 	SetupFlags: func(f *pflag.FlagSet) {
 		f.StringVar(&versionArgs.Show, "show", "", "version info to show (client|daemon)")
 	},
-	Run: func(subcommand *cli.Subcommand, args []string) error {
+	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
 		versionArgs.Config = subcommand.Config()
 		versionArgs.ConfigErr = subcommand.ConfigParsingError()
 		return runVersionCmd()
@@ -0,0 +1,147 @@
+package client
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+
+	"github.com/spf13/pflag"
+
+	"github.com/zrepl/zrepl/cli"
+	"github.com/zrepl/zrepl/daemon/filters"
+	"github.com/zrepl/zrepl/endpoint"
+	"github.com/zrepl/zrepl/zfs"
+)
+
+var (
+	ZFSAbstractionsCmd = &cli.Subcommand{
+		Use:   "zfs-abstraction",
+		Short: "manage abstractions that zrepl builds on top of ZFS",
+		SetupSubcommands: func() []*cli.Subcommand {
+			return []*cli.Subcommand{
+				zabsCmdList,
+				zabsCmdReleaseAll,
+				zabsCmdReleaseStale,
+				zabsCmdCreate,
+			}
+		},
+	}
+)
+
+// a common set of CLI flags that map to the fields of an
+// endpoint.ListZFSHoldsAndBookmarksQuery
+type zabsFilterFlags struct {
+	Filesystems FilesystemsFilterFlag
+	Job         JobIDFlag
+	Types       AbstractionTypesFlag
+	Concurrency int64
+}
+
+// produce a query from the CLI flags
+func (f zabsFilterFlags) Query() (endpoint.ListZFSHoldsAndBookmarksQuery, error) {
+	q := endpoint.ListZFSHoldsAndBookmarksQuery{
+		FS:          f.Filesystems.FlagValue(),
+		What:        f.Types.FlagValue(),
+		JobID:       f.Job.FlagValue(),
+		Concurrency: f.Concurrency,
+	}
+	return q, q.Validate()
+}
+
+func (f *zabsFilterFlags) registerZabsFilterFlags(s *pflag.FlagSet, verb string) {
+	// Note: the default value is defined in the .FlagValue methods
+	s.Var(&f.Filesystems, "fs", fmt.Sprintf("only %s holds on the specified filesystem [default: all filesystems] [comma-separated list of <dataset-pattern>:<ok|!> pairs]", verb))
+	s.Var(&f.Job, "job", fmt.Sprintf("only %s holds created by the specified job [default: any job]", verb))
+
+	variants := make([]string, 0, len(endpoint.AbstractionTypesAll))
+	for v := range endpoint.AbstractionTypesAll {
+		variants = append(variants, string(v))
+	}
+	variants = sort.StringSlice(variants)
+	variantsJoined := strings.Join(variants, "|")
+	s.Var(&f.Types, "type", fmt.Sprintf("only %s holds of the specified type [default: all] [comma-separated list of %s]", verb, variantsJoined))
+
+	s.Int64VarP(&f.Concurrency, "concurrency", "p", 1, "number of concurrently queried filesystems")
+}
+
+type JobIDFlag struct{ J *endpoint.JobID }
+
+func (f *JobIDFlag) Set(s string) error {
+	if len(s) == 0 {
+		*f = JobIDFlag{J: nil}
+		return nil
+	}
+
+	jobID, err := endpoint.MakeJobID(s)
+	if err != nil {
+		return err
+	}
+	*f = JobIDFlag{J: &jobID}
+	return nil
+}
+func (f JobIDFlag) Type() string               { return "job-ID" }
+func (f JobIDFlag) String() string             { return fmt.Sprint(f.J) }
+func (f JobIDFlag) FlagValue() *endpoint.JobID { return f.J }
+
+type AbstractionTypesFlag map[endpoint.AbstractionType]bool
+
+func (f *AbstractionTypesFlag) Set(s string) error {
+	ats, err := endpoint.AbstractionTypeSetFromStrings(strings.Split(s, ","))
+	if err != nil {
+		return err
+	}
+	*f = AbstractionTypesFlag(ats)
+	return nil
+}
+func (f AbstractionTypesFlag) Type() string { return "abstraction-type" }
+func (f AbstractionTypesFlag) String() string {
+	return endpoint.AbstractionTypeSet(f).String()
+}
+func (f AbstractionTypesFlag) FlagValue() map[endpoint.AbstractionType]bool {
+	if len(f) > 0 {
+		return f
+	}
+	return endpoint.AbstractionTypesAll
+}
+
+type FilesystemsFilterFlag struct {
+	F endpoint.ListZFSHoldsAndBookmarksQueryFilesystemFilter
+}
+
+func (flag *FilesystemsFilterFlag) Set(s string) error {
+	mappings := strings.Split(s, ",")
+	if len(mappings) == 1 && !strings.Contains(mappings[0], ":") {
+		flag.F = endpoint.ListZFSHoldsAndBookmarksQueryFilesystemFilter{
+			FS: &mappings[0],
+		}
+		return nil
+	}
+
+	f := filters.NewDatasetMapFilter(len(mappings), true)
+	for _, m := range mappings {
+		thisMappingErr := fmt.Errorf("expecting comma-separated list of <dataset-pattern>:<ok|!> pairs, got %q", m)
+		lhsrhs := strings.SplitN(m, ":", 2)
+		if len(lhsrhs) != 2 {
+			return thisMappingErr
+		}
+		err := f.Add(lhsrhs[0], lhsrhs[1])
+		if err != nil {
+			return fmt.Errorf("%s: %s", thisMappingErr, err)
+		}
+	}
+	flag.F = endpoint.ListZFSHoldsAndBookmarksQueryFilesystemFilter{
+		Filter: f,
+	}
+	return nil
+}
+func (flag FilesystemsFilterFlag) Type() string { return "filesystem filter spec" }
+func (flag FilesystemsFilterFlag) String() string {
+	return fmt.Sprintf("%v", flag.F)
+}
+func (flag FilesystemsFilterFlag) FlagValue() endpoint.ListZFSHoldsAndBookmarksQueryFilesystemFilter {
+	var z FilesystemsFilterFlag
+	if flag == z {
+		return endpoint.ListZFSHoldsAndBookmarksQueryFilesystemFilter{Filter: zfs.NoFilter()}
+	}
+	return flag.F
+}
@@ -0,0 +1,14 @@
+package client
+
+import "github.com/zrepl/zrepl/cli"
+
+var zabsCmdCreate = &cli.Subcommand{
+	Use:             "create",
+	NoRequireConfig: true,
+	Short:           `create zrepl ZFS abstractions (mostly useful for debugging & development, users should not need to use this command)`,
+	SetupSubcommands: func() []*cli.Subcommand {
+		return []*cli.Subcommand{
+			zabsCmdCreateStepHold,
+		}
+	},
+}
@@ -0,0 +1,58 @@
+package client
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/pkg/errors"
+	"github.com/spf13/pflag"
+
+	"github.com/zrepl/zrepl/cli"
+	"github.com/zrepl/zrepl/endpoint"
+	"github.com/zrepl/zrepl/zfs"
+)
+
+var zabsCreateStepHoldFlags struct {
+	target string
+	jobid  JobIDFlag
+}
+
+var zabsCmdCreateStepHold = &cli.Subcommand{
+	Use:             "step",
+	Run:             doZabsCreateStep,
+	NoRequireConfig: true,
+	Short:           `create a step hold or bookmark`,
+	SetupFlags: func(f *pflag.FlagSet) {
+		f.StringVarP(&zabsCreateStepHoldFlags.target, "target", "t", "", "snapshot to be held / bookmark to be held")
+		f.VarP(&zabsCreateStepHoldFlags.jobid, "jobid", "j", "jobid for which the hold is installed")
+	},
+}
+
+func doZabsCreateStep(ctx context.Context, sc *cli.Subcommand, args []string) error {
+	if len(args) > 0 {
+		return errors.New("subcommand takes no arguments")
+	}
+
+	f := &zabsCreateStepHoldFlags
+
+	fs, _, _, err := zfs.DecomposeVersionString(f.target)
+	if err != nil {
+		return errors.Wrapf(err, "%q invalid target", f.target)
+	}
+
+	if f.jobid.FlagValue() == nil {
+		return errors.Errorf("jobid must be set")
+	}
+
+	v, err := zfs.ZFSGetFilesystemVersion(ctx, f.target)
+	if err != nil {
+		return errors.Wrapf(err, "get info about target %q", f.target)
+	}
+
+	step, err := endpoint.HoldStep(ctx, fs, v, *f.jobid.FlagValue())
+	if err != nil {
+		return errors.Wrap(err, "create step hold")
+	}
+	fmt.Println(step.String())
+	return nil
+}
@@ -0,0 +1,99 @@
+package client
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"sync"
+
+	"github.com/fatih/color"
+	"github.com/pkg/errors"
+	"github.com/spf13/pflag"
+
+	"github.com/zrepl/zrepl/cli"
+	"github.com/zrepl/zrepl/endpoint"
+	"github.com/zrepl/zrepl/util/chainlock"
+)
+
+var zabsListFlags struct {
+	Filter zabsFilterFlags
+	Json   bool
+}
+
+var zabsCmdList = &cli.Subcommand{
+	Use:             "list",
+	Short:           `list zrepl ZFS abstractions`,
+	Run:             doZabsList,
+	NoRequireConfig: true,
+	SetupFlags: func(f *pflag.FlagSet) {
+		zabsListFlags.Filter.registerZabsFilterFlags(f, "list")
+		f.BoolVar(&zabsListFlags.Json, "json", false, "emit JSON")
+	},
+}
+
+func doZabsList(ctx context.Context, sc *cli.Subcommand, args []string) error {
+	var err error
+
+	if len(args) > 0 {
+		return errors.New("this subcommand takes no positional arguments")
+	}
+
+	q, err := zabsListFlags.Filter.Query()
+	if err != nil {
+		return errors.Wrap(err, "invalid filter specification on command line")
+	}
+
+	abstractions, errors, err := endpoint.ListAbstractionsStreamed(ctx, q)
+	if err != nil {
+		return err // context clear by invocation of command
+	}
+
+	var line chainlock.L
+	var wg sync.WaitGroup
+	defer wg.Wait()
+	wg.Add(1)
+
+	// print results
+	go func() {
+		defer wg.Done()
+		enc := json.NewEncoder(os.Stdout)
+		for a := range abstractions {
+			func() {
+				defer line.Lock().Unlock()
+				if zabsListFlags.Json {
+					enc.SetIndent("", "  ")
+					if err := enc.Encode(a); err != nil {
+						panic(err)
+					}
+					fmt.Println()
+				} else {
+					fmt.Println(a)
+				}
+			}()
+		}
+	}()
+
+	// print errors to stderr
+	errorColor := color.New(color.FgRed)
+	var errorsSlice []endpoint.ListAbstractionsError
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for err := range errors {
+			func() {
+				defer line.Lock().Unlock()
+				errorsSlice = append(errorsSlice, err)
+				errorColor.Fprintf(os.Stderr, "%s\n", err)
+			}()
+		}
+	}()
+	wg.Wait()
+	if len(errorsSlice) > 0 {
+		errorColor.Add(color.Bold).Fprintf(os.Stderr, "there were errors in listing the abstractions")
+		return fmt.Errorf("")
+	} else {
+		return nil
+	}
+
+}
@@ -0,0 +1,143 @@
+package client
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+
+	"github.com/fatih/color"
+	"github.com/pkg/errors"
+	"github.com/spf13/pflag"
+
+	"github.com/zrepl/zrepl/cli"
+	"github.com/zrepl/zrepl/endpoint"
+)
+
+// shared between release-all and release-step
+var zabsReleaseFlags struct {
+	Filter zabsFilterFlags
+	Json   bool
+	DryRun bool
+}
+
+func registerZabsReleaseFlags(s *pflag.FlagSet) {
+	zabsReleaseFlags.Filter.registerZabsFilterFlags(s, "release")
+	s.BoolVar(&zabsReleaseFlags.Json, "json", false, "emit json instead of pretty-printed")
+	s.BoolVar(&zabsReleaseFlags.DryRun, "dry-run", false, "do a dry-run")
+}
+
+var zabsCmdReleaseAll = &cli.Subcommand{
+	Use:             "release-all",
+	Run:             doZabsReleaseAll,
+	NoRequireConfig: true,
+	Short:           `(DANGEROUS) release ALL zrepl ZFS abstractions (mostly useful for uninstalling zrepl completely or for "de-zrepl-ing" a filesystem)`,
+	SetupFlags:      registerZabsReleaseFlags,
+}
+
+var zabsCmdReleaseStale = &cli.Subcommand{
+	Use:             "release-stale",
+	Run:             doZabsReleaseStale,
+	NoRequireConfig: true,
+	Short:           `release stale zrepl ZFS abstractions (useful if zrepl has a bug and does not do it by itself)`,
+	SetupFlags:      registerZabsReleaseFlags,
+}
+
+func doZabsReleaseAll(ctx context.Context, sc *cli.Subcommand, args []string) error {
+	var err error
+
+	if len(args) > 0 {
+		return errors.New("this subcommand takes no positional arguments")
+	}
+
+	q, err := zabsReleaseFlags.Filter.Query()
+	if err != nil {
+		return errors.Wrap(err, "invalid filter specification on command line")
+	}
+
+	abstractions, listErrors, err := endpoint.ListAbstractions(ctx, q)
+	if err != nil {
+		return err // context clear by invocation of command
+	}
+	if len(listErrors) > 0 {
+		color.New(color.FgRed).Fprintf(os.Stderr, "there were errors in listing the abstractions:\n%s\n", listErrors)
+		// proceed anyways with rest of abstractions
+	}
+
+	return doZabsRelease_Common(ctx, abstractions)
+}
+
+func doZabsReleaseStale(ctx context.Context, sc *cli.Subcommand, args []string) error {
+
+	var err error
+
+	if len(args) > 0 {
+		return errors.New("this subcommand takes no positional arguments")
+	}
+
+	q, err := zabsReleaseFlags.Filter.Query()
+	if err != nil {
+		return errors.Wrap(err, "invalid filter specification on command line")
+	}
+
+	stalenessInfo, err := endpoint.ListStale(ctx, q)
+	if err != nil {
+		return err // context clear by invocation of command
+	}
+
+	return doZabsRelease_Common(ctx, stalenessInfo.Stale)
+}
+
+func doZabsRelease_Common(ctx context.Context, destroy []endpoint.Abstraction) error {
+
+	if zabsReleaseFlags.DryRun {
+		if zabsReleaseFlags.Json {
+			m, err := json.MarshalIndent(destroy, "", "  ")
+			if err != nil {
+				panic(err)
+			}
+			if _, err := os.Stdout.Write(m); err != nil {
+				panic(err)
+			}
+			fmt.Println()
+		} else {
+			for _, a := range destroy {
+				fmt.Printf("would destroy %s\n", a)
+			}
+		}
+		return nil
+	}
+
+	outcome := endpoint.BatchDestroy(ctx, destroy)
+	hadErr := false
+
+	enc := json.NewEncoder(os.Stdout)
+	enc.SetIndent("", "  ")
+	colorErr := color.New(color.FgRed)
+	printfSuccess := color.New(color.FgGreen).FprintfFunc()
+	printfSection := color.New(color.Bold).FprintfFunc()
+
+	for res := range outcome {
+		hadErr = hadErr || res.DestroyErr != nil
+		if zabsReleaseFlags.Json {
+			err := enc.Encode(res)
+			if err != nil {
+				colorErr.Fprintf(os.Stderr, "cannot marshal there were errors in destroying the abstractions")
+			}
+		} else {
+			printfSection(os.Stdout, "destroy %s ...", res.Abstraction)
+			if res.DestroyErr != nil {
+				colorErr.Fprintf(os.Stdout, " failed:\n%s\n", res.DestroyErr)
+			} else {
+				printfSuccess(os.Stdout, " OK\n")
+			}
+		}
+	}
+
+	if hadErr {
+		colorErr.Add(color.Bold).Fprintf(os.Stderr, "there were errors in destroying the abstractions")
+		return fmt.Errorf("")
+	} else {
+		return nil
+	}
+}
@@ -52,11 +52,12 @@ func (j JobEnum) Name() string {
 }

 type ActiveJob struct {
-	Type    string                `yaml:"type"`
-	Name    string                `yaml:"name"`
-	Connect ConnectEnum           `yaml:"connect"`
-	Pruning PruningSenderReceiver `yaml:"pruning"`
-	Debug   JobDebugSettings      `yaml:"debug,optional"`
+	Type        string                `yaml:"type"`
+	Name        string                `yaml:"name"`
+	Connect     ConnectEnum           `yaml:"connect"`
+	Pruning     PruningSenderReceiver `yaml:"pruning"`
+	Debug       JobDebugSettings      `yaml:"debug,optional"`
+	Replication *Replication          `yaml:"replication,optional,fromdefaults"`
 }

 type PassiveJob struct {
@@ -76,13 +77,7 @@ type SnapJob struct {
 }

 type SendOptions struct {
-	Encrypted bool `yaml:"encrypted"`
-}
-
-var _ yaml.Defaulter = (*SendOptions)(nil)
-
-func (l *SendOptions) SetDefault() {
-	*l = SendOptions{Encrypted: false}
+	Encrypted bool `yaml:"encrypted,optional,default=false"`
 }

 type RecvOptions struct {
@@ -93,10 +88,13 @@ type RecvOptions struct {
 	// Reencrypt bool `yaml:"reencrypt"`
 }

-var _ yaml.Defaulter = (*RecvOptions)(nil)
+type Replication struct {
+	Protection *ReplicationOptionsProtection `yaml:"protection,optional,fromdefaults"`
+}

-func (l *RecvOptions) SetDefault() {
-	*l = RecvOptions{}
+type ReplicationOptionsProtection struct {
+	Initial     string `yaml:"initial,optional,default=guarantee_resumability"`
+	Incremental string `yaml:"incremental,optional,default=guarantee_resumability"`
 }

 type PushJob struct {
@@ -106,6 +104,9 @@ type PushJob struct {
 	Send         *SendOptions      `yaml:"send,fromdefaults,optional"`
 }

+func (j *PushJob) GetFilesystems() FilesystemsFilter { return j.Filesystems }
+func (j *PushJob) GetSendOptions() *SendOptions      { return j.Send }
+
 type PullJob struct {
 	ActiveJob `yaml:",inline"`
 	RootFS    string                   `yaml:"root_fs"`
@@ -113,6 +114,10 @@ type PullJob struct {
 	Recv      *RecvOptions             `yaml:"recv,fromdefaults,optional"`
 }

+func (j *PullJob) GetRootFS() string             { return j.RootFS }
+func (j *PullJob) GetAppendClientIdentity() bool { return false }
+func (j *PullJob) GetRecvOptions() *RecvOptions  { return j.Recv }
+
 type PositiveDurationOrManual struct {
 	Interval time.Duration
 	Manual   bool
@@ -150,6 +155,10 @@ type SinkJob struct {
 	Recv       *RecvOptions `yaml:"recv,optional,fromdefaults"`
 }

+func (j *SinkJob) GetRootFS() string             { return j.RootFS }
+func (j *SinkJob) GetAppendClientIdentity() bool { return true }
+func (j *SinkJob) GetRecvOptions() *RecvOptions  { return j.Recv }
+
 type SourceJob struct {
 	PassiveJob   `yaml:",inline"`
 	Snapshotting SnapshottingEnum  `yaml:"snapshotting"`
@@ -157,6 +166,9 @@ type SourceJob struct {
 	Send         *SendOptions      `yaml:"send,optional,fromdefaults"`
 }

+func (j *SourceJob) GetFilesystems() FilesystemsFilter { return j.Filesystems }
+func (j *SourceJob) GetSendOptions() *SendOptions      { return j.Send }
+
 type FilesystemsFilter map[string]bool

 type SnapshottingEnum struct {
@@ -620,7 +632,7 @@ func ParseConfigBytes(bytes []byte) (*Config, error) {

 var durationStringRegex *regexp.Regexp = regexp.MustCompile(`^\s*(\d+)\s*(s|m|h|d|w)\s*$`)

-func parsePostitiveDuration(e string) (d time.Duration, err error) {
+func parsePositiveDuration(e string) (d time.Duration, err error) {
 	comps := durationStringRegex.FindStringSubmatch(e)
 	if len(comps) != 3 {
 		err = fmt.Errorf("does not match regex: %s %#v", e, comps)
@@ -0,0 +1,90 @@
+package config
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/zrepl/yaml-config"
+)
+
+type A struct {
+	B  *B     `yaml:"b,optional,fromdefaults"`
+	A1 string `yaml:"a1,optional"`
+}
+
+type B struct {
+	C *C     `yaml:"c,optional,fromdefaults"`
+	D string `yaml:"d,default=ddd"`
+	E string `yaml:"e,optional"`
+}
+
+type C struct {
+	Q string `yaml:"q,optional"`
+	R string `yaml:"r,default=r"`
+}
+
+func TestDepFromDefaults(t *testing.T) {
+
+	type testcase struct {
+		name   string
+		yaml   string
+		expect *A
+	}
+
+	tcs := []testcase{
+		{
+			name: "empty",
+			yaml: `{}`,
+			expect: &A{
+				B: &B{
+					C: &C{
+						R: "r",
+					},
+					D: "ddd",
+				},
+			},
+		},
+		{
+			name: "a1 set",
+			yaml: `{"a1":"blah"}`,
+			expect: &A{
+				A1: "blah",
+				B: &B{
+					C: &C{
+						R: "r",
+					},
+					D: "ddd",
+				},
+			},
+		},
+		{
+			name: "D set",
+			yaml: `
+b:
+  d: 4d
+`,
+			expect: &A{
+				B: &B{
+					D: "4d",
+					C: &C{
+						R: "r",
+					},
+				},
+			},
+		},
+	}
+
+	for tci := range tcs {
+		t.Run(fmt.Sprintf("%d-%s", tci, tcs[tci].name), func(t *testing.T) {
+			tc := tcs[tci]
+
+			var a A
+			err := yaml.UnmarshalStrict([]byte(tc.yaml), &a)
+			require.NoError(t, err)
+
+			require.Equal(t, tc.expect, &a)
+		})
+	}
+
+}
@@ -21,7 +21,7 @@ jobs:
    clients: {
      "10.0.0.1":"foo"
    }
-  root_fs: zoot/foo
+  root_fs: zroot/foo
 `
 	_, err := ParseConfigBytes([]byte(jobdef))
 	require.NoError(t, err)
@@ -60,9 +60,9 @@ jobs:
 	})

 	t.Run("encrypted_unspecified", func(t *testing.T) {
-		c, err := testConfig(t, fill(encrypted_unspecified))
-		assert.Error(t, err)
-		assert.Nil(t, c)
+		c = testValidConfig(t, fill(encrypted_unspecified))
+		encrypted := c.Jobs[0].Ret.(*PushJob).Send.Encrypted
+		assert.Equal(t, false, encrypted)
 	})

 	t.Run("send_not_specified", func(t *testing.T) {
@@ -42,7 +42,7 @@ func TestSampleConfigsAreParsedWithoutErrors(t *testing.T) {

 }

-// template must be a template/text template with a single '{{ . }}' as placehodler for val
+// template must be a template/text template with a single '{{ . }}' as placeholder for val
 //nolint[:deadcode,unused]
 func testValidConfigTemplate(t *testing.T, tmpl string, val string) *Config {
 	tmp, err := template.New("master").Parse(tmpl)
@@ -64,7 +64,7 @@ func parseRetentionGridIntervalString(e string) (intervals []RetentionInterval,
 		return nil, fmt.Errorf("contains factor <= 0")
 	}

-	duration, err := parsePostitiveDuration(comps[2])
+	duration, err := parsePositiveDuration(comps[2])
 	if err != nil {
 		return nil, err
 	}
@@ -0,0 +1,88 @@
+# This config serves as an example for a local zrepl installation that
+# backups the entire zpool `system` to `backuppool/zrepl/sink`
+#
+# The requirements covered by this setup are described in the zrepl documentation's
+# quick start section which inlines this example.
+#
+# CUSTOMIZATIONS YOU WILL LIKELY WANT TO APPLY:
+# - adjust the name of the production pool `system` in the `filesystems` filter of jobs `snapjob` and `push_to_derive`
+# - adjust the name of the backup pool `backuppool` in the `backuppool_sink` job
+# - adjust the occurences of `myhostname` to the name of the system you are backing up (cannot be easily changed once you start replicating)
+# - make sure the `zrepl_` prefix is not being used by any other zfs tools you might have installed (it likely isn't)
+
+jobs:
+
+# this job takes care of snapshot creation + pruning
+- name: snapjob
+  type: snap
+  filesystems: {
+      "system<": true,
+  }
+  # create snapshots with prefix `zrepl_` every 15 minutes
+  snapshotting:
+    type: periodic
+    interval: 15m
+    prefix: zrepl_
+  pruning:
+    keep:
+    # fade-out scheme for snapshots starting with `zrepl_`
+    # - keep all created in the last hour
+    # - then destroy snapshots such that we keep 24 each 1 hour apart
+    # - then destroy snapshots such that we keep 14 each 1 day apart
+    # - then destroy all older snapshots
+    - type: grid
+      grid: 1x1h(keep=all) | 24x1h | 14x1d
+      regex: "^zrepl_.*"
+    # keep all snapshots that don't have the `zrepl_` prefix
+    - type: regex
+      negate: true
+      regex: "^zrepl_.*"
+
+# This job pushes to the local sink defined in job `backuppool_sink`.
+# We trigger replication manually from the command line / udev rules using
+#  `zrepl signal wakeup push_to_drive`
+- type: push
+  name: push_to_drive
+  connect:
+    type: local
+    listener_name: backuppool_sink
+    client_identity: myhostname
+  filesystems: {
+      "system<": true
+  }
+  send:
+    encrypted: true
+  replication:
+    protection:
+      initial: guarantee_resumability
+      # Downgrade protection to guarantee_incremental which uses zfs bookmarks instead of zfs holds.
+      # Thus, when we yank out the backup drive during replication
+      # - we might not be able to resume the interrupted replication step because the partially received `to` snapshot of a `from`->`to` step may be pruned any time
+      # - but in exchange we get back the disk space allocated by `to` when we prune it
+      # - and because we still have the bookmarks created by `guarantee_incremental`, we can still do incremental replication of `from`->`to2` in the future
+      incremental: guarantee_incremental
+  snapshotting:
+    type: manual
+  pruning:
+    # no-op prune rule on sender (keep all snapshots), job `snapshot` takes care of this
+    keep_sender:
+    - type: regex
+      regex: ".*"
+    # retain 
+    keep_receiver:
+    # longer retention on the backup drive, we have more space there
+    - type: grid
+      grid: 1x1h(keep=all) | 24x1h | 360x1d
+      regex: "^zrepl_.*"
+    # retain all non-zrepl snapshots on the backup drive
+    - type: regex
+      negate: true
+      regex: "^zrepl_.*"
+
+# This job receives from job `push_to_drive` into `backuppool/zrepl/sink/myhostname`
+- type: sink
+  name: backuppool_sink
+  root_fs: "backuppool/zrepl/sink"
+  serve:
+    type: local
+    listener_name: backuppool_sink
@@ -0,0 +1,12 @@
+jobs:
+- name: sink
+  type: sink
+  serve:
+      type: tls
+      listen: ":8888"
+      ca: "/etc/zrepl/prod.crt"
+      cert: "/etc/zrepl/backups.crt"
+      key: "/etc/zrepl/backups.key"
+      client_cns:
+        - "prod"
+  root_fs: "storage/zrepl/sink"
@@ -0,0 +1,28 @@
+jobs:
+- name: prod_to_backups
+  type: push
+  connect:
+    type: tls
+    address: "backups.example.com:8888"
+    ca: /etc/zrepl/backups.crt
+    cert: /etc/zrepl/prod.crt
+    key:  /etc/zrepl/prod.key
+    server_cn: "backups"
+  filesystems: {
+    "zroot/var/db": true,
+    "zroot/usr/home<": true,
+    "zroot/usr/home/paranoid": false
+  }
+  snapshotting:
+    type: periodic
+    prefix: zrepl_
+    interval: 10m
+  pruning:
+    keep_sender:
+    - type: not_replicated
+    - type: last_n
+      count: 10
+    keep_receiver:
+    - type: grid
+      grid: 1x1h(keep=all) | 24x1h | 30x1d | 6x30d
+      regex: "^zrepl_"
@@ -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,
@@ -8,7 +8,6 @@ import (
 	"io"
 	"net"
 	"net/http"
-	"strconv"
 	"time"

 	"github.com/pkg/errors"
@@ -21,6 +20,7 @@ import (
 	"github.com/zrepl/zrepl/util/envconst"
 	"github.com/zrepl/zrepl/version"
 	"github.com/zrepl/zrepl/zfs"
+	"github.com/zrepl/zrepl/zfs/zfscmd"
 )

 type controlJob struct {
@@ -48,8 +48,6 @@ func (j *controlJob) OwnedDatasetSubtreeRoot() (p *zfs.DatasetPath, ok bool) { r

 func (j *controlJob) SenderConfig() *endpoint.SenderConfig { return nil }

-func (j *controlJob) SetConcurrency(concurrency int) error { return errors.Errorf("not supported") }
-
 var promControl struct {
 	requestBegin    *prometheus.CounterVec
 	requestFinished *prometheus.HistogramVec
@@ -67,7 +65,7 @@ func (j *controlJob) RegisterMetrics(registerer prometheus.Registerer) {
 		Namespace: "zrepl",
 		Subsystem: "control",
 		Name:      "request_finished",
-		Help:      "time it took a request to finih",
+		Help:      "time it took a request to finish",
 		Buckets:   []float64{1e-6, 10e-6, 100e-6, 500e-6, 1e-3, 10e-3, 100e-3, 200e-3, 400e-3, 800e-3, 1, 10, 20},
 	}, []string{"endpoint"})
 	registerer.MustRegister(promControl.requestBegin)
@@ -120,7 +118,15 @@ func (j *controlJob) Run(ctx context.Context) {
 	mux.Handle(ControlJobEndpointStatus,
 		// don't log requests to status endpoint, too spammy
 		jsonResponder{log, func() (interface{}, error) {
-			s := j.jobs.status()
+			jobs := j.jobs.status()
+			globalZFS := zfscmd.GetReport()
+			envconstReport := envconst.GetReport()
+			s := Status{
+				Jobs: jobs,
+				Global: GlobalStatus{
+					ZFSCmds:  globalZFS,
+					Envconst: envconstReport,
+				}}
 			return s, nil
 		}})

@@ -129,7 +135,6 @@ func (j *controlJob) Run(ctx context.Context) {
 			type reqT struct {
 				Name string
 				Op   string
-				Data string
 			}
 			var req reqT
 			if decoder(&req) != nil {
@@ -142,14 +147,6 @@ func (j *controlJob) Run(ctx context.Context) {
 				err = j.jobs.wakeup(req.Name)
 			case "reset":
 				err = j.jobs.reset(req.Name)
-			case "set-concurrency":
-				var concurrency int
-				concurrency, err = strconv.Atoi(req.Data) // shadow
-				if err != nil {
-					// fallthrough outer
-				} else {
-					err = j.jobs.setConcurrency(req.Name, concurrency)
-				}
 			default:
 				err = fmt.Errorf("operation %q is invalid", req.Op)
 			}
@@ -262,7 +259,7 @@ func (j jsonRequestResponder) ServeHTTP(w http.ResponseWriter, r *http.Request)
 	var buf bytes.Buffer
 	encodeErr := json.NewEncoder(&buf).Encode(res)
 	if encodeErr != nil {
-		j.log.WithError(producerErr).Error("control handler json marhsal error")
+		j.log.WithError(producerErr).Error("control handler json marshal error")
 		w.WriteHeader(http.StatusInternalServerError)
 		_, err := io.WriteString(w, encodeErr.Error())
 		logIoErr(err)
@@ -12,6 +12,9 @@ import (

 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/zrepl/zrepl/daemon/logging/trace"
+	"github.com/zrepl/zrepl/endpoint"
+	"github.com/zrepl/zrepl/util/envconst"

 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/job"
@@ -20,11 +23,11 @@ import (
 	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/version"
+	"github.com/zrepl/zrepl/zfs/zfscmd"
 )

-func Run(conf *config.Config) error {
-
-	ctx, cancel := context.WithCancel(context.Background())
+func Run(ctx context.Context, conf *config.Config) error {
+	ctx, cancel := context.WithCancel(ctx)

 	defer cancel()
 	sigChan := make(chan os.Signal, 1)
@@ -38,6 +41,7 @@ func Run(conf *config.Config) error {
 	if err != nil {
 		return errors.Wrap(err, "cannot build logging from config")
 	}
+	outlets.Add(newPrometheusLogOutlet(), logger.Debug)

 	confJobs, err := job.JobsFromConfig(conf)
 	if err != nil {
@@ -47,14 +51,23 @@ func Run(conf *config.Config) error {
 	log := logger.NewLogger(outlets, 1*time.Second)
 	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()) {
 			panic(fmt.Sprintf("internal job name used for config job '%s'", job.Name())) //FIXME
 		}
 	}

-	ctx = job.WithLogger(ctx, log)
-
 	jobs := newJobs()

 	// start control socket
@@ -81,6 +94,12 @@ func Run(conf *config.Config) error {
 		jobs.start(ctx, job, true)
 	}

+	// register global (=non job-local) metrics
+	version.PrometheusRegister(prometheus.DefaultRegisterer)
+	zfscmd.RegisterMetrics(prometheus.DefaultRegisterer)
+	trace.RegisterMetrics(prometheus.DefaultRegisterer)
+	endpoint.RegisterMetrics(prometheus.DefaultRegisterer)
+
 	log.Info("starting daemon")

 	// start regular jobs
@@ -94,6 +113,8 @@ func Run(conf *config.Config) error {
 	case <-ctx.Done():
 		log.WithError(ctx.Err()).Info("context finished")
 	}
+	log.Info("waiting for jobs to finish")
+	<-jobs.wait()
 	log.Info("daemon exiting")
 	return nil
 }
@@ -116,18 +137,25 @@ func newJobs() *jobs {
 	}
 }

-const (
-	logJobField string = "job"
-)
-
 func (s *jobs) wait() <-chan struct{} {
 	ch := make(chan struct{})
 	go func() {
 		s.wg.Wait()
+		close(ch)
 	}()
 	return ch
 }

+type Status struct {
+	Jobs   map[string]*job.Status
+	Global GlobalStatus
+}
+
+type GlobalStatus struct {
+	ZFSCmds  *zfscmd.Report
+	Envconst *envconst.Report
+}
+
 func (s *jobs) status() map[string]*job.Status {
 	s.m.RLock()
 	defer s.m.RUnlock()
@@ -176,18 +204,6 @@ func (s *jobs) reset(job string) error {
 	return wu()
 }

-func (s *jobs) setConcurrency(jobName string, concurrency int) error {
-	s.m.RLock()
-	defer s.m.RUnlock()
-
-	job, ok := s.jobs[jobName]
-	if !ok {
-		return errors.Errorf("Job %q does not exist", job)
-	}
-
-	return job.SetConcurrency(concurrency)
-}
-
 const (
 	jobNamePrometheus = "_prometheus"
 	jobNameControl    = "_control"
@@ -201,9 +217,8 @@ func (s *jobs) start(ctx context.Context, j job.Job, internal bool) {
 	s.m.Lock()
 	defer s.m.Unlock()

-	jobLog := job.GetLogger(ctx).
-		WithField(logJobField, j.Name()).
-		WithOutlet(newPrometheusLogOutlet(j.Name()), logger.Debug)
+	ctx = logging.WithInjectedField(ctx, logging.JobField, j.Name())
+
 	jobName := j.Name()
 	if !internal && IsInternalJobName(jobName) {
 		panic(fmt.Sprintf("internal job name used for non-internal job %s", jobName))
@@ -218,7 +233,7 @@ func (s *jobs) start(ctx context.Context, j job.Job, internal bool) {
 	j.RegisterMetrics(prometheus.DefaultRegisterer)

 	s.jobs[jobName] = j
-	ctx = job.WithLogger(ctx, jobLog)
+	ctx = zfscmd.WithJobID(ctx, j.Name())
 	ctx, wakeup := wakeup.Context(ctx)
 	ctx, resetFunc := reset.Context(ctx)
 	s.wakeups[jobName] = wakeup
@@ -227,8 +242,8 @@ func (s *jobs) start(ctx context.Context, j job.Job, internal bool) {
 	s.wg.Add(1)
 	go func() {
 		defer s.wg.Done()
-		jobLog.Info("starting job")
-		defer jobLog.Info("job exited")
+		job.GetLogger(ctx).Info("starting job")
+		defer job.GetLogger(ctx).Info("job exited")
 		j.Run(ctx)
 	}()
 }
@@ -161,7 +161,7 @@ func (m DatasetMapFilter) Filter(p *zfs.DatasetPath) (pass bool, err error) {
 }

 // Construct a new filter-only DatasetMapFilter from a mapping
-// The new filter allows excactly those paths that were not forbidden by the mapping.
+// The new filter allows exactly those paths that were not forbidden by the mapping.
 func (m DatasetMapFilter) InvertedFilter() (inv *DatasetMapFilter, err error) {

 	if m.filterMode {
@@ -1,41 +0,0 @@
-package filters
-
-import (
-	"strings"
-
-	"github.com/zrepl/zrepl/zfs"
-)
-
-type AnyFSVFilter struct{}
-
-func NewAnyFSVFilter() AnyFSVFilter {
-	return AnyFSVFilter{}
-}
-
-var _ zfs.FilesystemVersionFilter = AnyFSVFilter{}
-
-func (AnyFSVFilter) Filter(t zfs.VersionType, name string) (accept bool, err error) {
-	return true, nil
-}
-
-type PrefixFilter struct {
-	prefix    string
-	fstype    zfs.VersionType
-	fstypeSet bool // optionals anyone?
-}
-
-var _ zfs.FilesystemVersionFilter = &PrefixFilter{}
-
-func NewPrefixFilter(prefix string) *PrefixFilter {
-	return &PrefixFilter{prefix: prefix}
-}
-
-func NewTypedPrefixFilter(prefix string, versionType zfs.VersionType) *PrefixFilter {
-	return &PrefixFilter{prefix, versionType, true}
-}
-
-func (f *PrefixFilter) Filter(t zfs.VersionType, name string) (accept bool, err error) {
-	fstypeMatches := (!f.fstypeSet || t == f.fstype)
-	prefixMatches := strings.HasPrefix(name, f.prefix)
-	return fstypeMatches && prefixMatches, nil
-}
@@ -6,29 +6,17 @@ import (
 	"context"
 	"sync"

+	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/util/envconst"
 )

-type contextKey int
-
-const (
-	contextKeyLog contextKey = 0
-)
-
 type Logger = logger.Logger

-func WithLogger(ctx context.Context, log Logger) context.Context {
-	return context.WithValue(ctx, contextKeyLog, log)
-}
-
 func GetLogger(ctx context.Context) Logger { return getLogger(ctx) }

 func getLogger(ctx context.Context) Logger {
-	if log, ok := ctx.Value(contextKeyLog).(Logger); ok {
-		return log
-	}
-	return logger.NewNullLogger()
+	return logging.GetLogger(ctx, logging.SubsysHooks)
 }

 const MAX_HOOK_LOG_SIZE_DEFAULT int = 1 << 20
@@ -10,9 +10,11 @@ import (
 	"text/template"

 	"github.com/stretchr/testify/require"
+	"github.com/zrepl/zrepl/daemon/logging/trace"

 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/hooks"
+	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/zfs"
 )
@@ -69,6 +71,9 @@ func curry(f comparisonAssertionFunc, expected interface{}, right bool) (ret val
 }

 func TestHooks(t *testing.T) {
+	ctx, end := trace.WithTaskFromStack(context.Background())
+	defer end()
+
 	testFSName := "testpool/testdataset"
 	testSnapshotName := "testsnap"

@@ -418,9 +423,8 @@ jobs:

 			cbReached = false

-			ctx := context.Background()
 			if testing.Verbose() && !tt.SuppressOutput {
-				ctx = hooks.WithLogger(ctx, log)
+				ctx = logging.WithLoggers(ctx, logging.SubsystemLoggersWithUniversalLogger(log))
 			}
 			plan.Run(ctx, false)
 			report := plan.Report()
@@ -8,12 +8,12 @@ import (

 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/common/log"
+	"github.com/zrepl/zrepl/daemon/logging/trace"

 	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon/filters"
 	"github.com/zrepl/zrepl/daemon/job/reset"
 	"github.com/zrepl/zrepl/daemon/job/wakeup"
-	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/daemon/pruner"
 	"github.com/zrepl/zrepl/daemon/snapper"
 	"github.com/zrepl/zrepl/endpoint"
@@ -55,12 +55,9 @@ const (
 type activeSideTasks struct {
 	state ActiveSideState

-	concurrency int
-
 	// valid for state ActiveSideReplicating, ActiveSidePruneSender, ActiveSidePruneReceiver, ActiveSideDone
-	replicationReport         driver.ReportFunc
-	replicationCancel         context.CancelFunc
-	replicationSetConcurrency driver.SetConcurrencyFunc
+	replicationReport driver.ReportFunc
+	replicationCancel context.CancelFunc

 	// valid for state ActiveSidePruneSender, ActiveSidePruneReceiver, ActiveSideDone
 	prunerSender, prunerReceiver *pruner.Pruner
@@ -82,7 +79,7 @@ func (a *ActiveSide) updateTasks(u func(*activeSideTasks)) activeSideTasks {
 }

 type activeMode interface {
-	ConnectEndpoints(rpcLoggers rpc.Loggers, connecter transport.Connecter)
+	ConnectEndpoints(ctx context.Context, connecter transport.Connecter)
 	DisconnectEndpoints()
 	SenderReceiver() (logic.Sender, logic.Receiver)
 	Type() Type
@@ -101,14 +98,14 @@ type modePush struct {
 	snapper       *snapper.PeriodicOrManual
 }

-func (m *modePush) ConnectEndpoints(loggers rpc.Loggers, connecter transport.Connecter) {
+func (m *modePush) ConnectEndpoints(ctx context.Context, connecter transport.Connecter) {
 	m.setupMtx.Lock()
 	defer m.setupMtx.Unlock()
 	if m.receiver != nil || m.sender != nil {
 		panic("inconsistent use of ConnectEndpoints and DisconnectEndpoints")
 	}
 	m.sender = endpoint.NewSender(*m.senderConfig)
-	m.receiver = rpc.NewClient(connecter, loggers)
+	m.receiver = rpc.NewClient(connecter, rpc.GetLoggersOrPanic(ctx))
 }

 func (m *modePush) DisconnectEndpoints() {
@@ -147,22 +144,24 @@ func (m *modePush) ResetConnectBackoff() {

 func modePushFromConfig(g *config.Global, in *config.PushJob, jobID endpoint.JobID) (*modePush, error) {
 	m := &modePush{}
+	var err error

-	fsf, err := filters.DatasetMapFilterFromConfig(in.Filesystems)
+	m.senderConfig, err = buildSenderConfig(in, jobID)
 	if err != nil {
-		return nil, errors.Wrap(err, "cannnot build filesystem filter")
+		return nil, errors.Wrap(err, "sender config")
 	}

-	m.senderConfig = &endpoint.SenderConfig{
-		FSF:     fsf,
-		Encrypt: &zfs.NilBool{B: in.Send.Encrypted},
-		JobID:   jobID,
+	replicationConfig, err := logic.ReplicationConfigFromConfig(in.Replication)
+	if err != nil {
+		return nil, errors.Wrap(err, "field `replication`")
 	}
+
 	m.plannerPolicy = &logic.PlannerPolicy{
-		EncryptedSend: logic.TriFromBool(in.Send.Encrypted),
+		EncryptedSend:     logic.TriFromBool(in.Send.Encrypted),
+		ReplicationConfig: *replicationConfig,
 	}

-	if m.snapper, err = snapper.FromConfig(g, fsf, in.Snapshotting); err != nil {
+	if m.snapper, err = snapper.FromConfig(g, m.senderConfig.FSF, in.Snapshotting); err != nil {
 		return nil, errors.Wrap(err, "cannot build snapper")
 	}

@@ -174,19 +173,18 @@ type modePull struct {
 	receiver       *endpoint.Receiver
 	receiverConfig endpoint.ReceiverConfig
 	sender         *rpc.Client
-	rootFS         *zfs.DatasetPath
 	plannerPolicy  *logic.PlannerPolicy
 	interval       config.PositiveDurationOrManual
 }

-func (m *modePull) ConnectEndpoints(loggers rpc.Loggers, connecter transport.Connecter) {
+func (m *modePull) ConnectEndpoints(ctx context.Context, connecter transport.Connecter) {
 	m.setupMtx.Lock()
 	defer m.setupMtx.Unlock()
 	if m.receiver != nil || m.sender != nil {
 		panic("inconsistent use of ConnectEndpoints and DisconnectEndpoints")
 	}
 	m.receiver = endpoint.NewReceiver(m.receiverConfig)
-	m.sender = rpc.NewClient(connecter, loggers)
+	m.sender = rpc.NewClient(connecter, rpc.GetLoggersOrPanic(ctx))
 }

 func (m *modePull) DisconnectEndpoints() {
@@ -248,26 +246,19 @@ func modePullFromConfig(g *config.Global, in *config.PullJob, jobID endpoint.Job
 	m = &modePull{}
 	m.interval = in.Interval

-	m.rootFS, err = zfs.NewDatasetPath(in.RootFS)
+	replicationConfig, err := logic.ReplicationConfigFromConfig(in.Replication)
 	if err != nil {
-		return nil, errors.New("RootFS is not a valid zfs filesystem path")
-	}
-	if m.rootFS.Length() <= 0 {
-		return nil, errors.New("RootFS must not be empty") // duplicates error check of receiver
+		return nil, errors.Wrap(err, "field `replication`")
 	}

 	m.plannerPolicy = &logic.PlannerPolicy{
-		EncryptedSend: logic.DontCare,
+		EncryptedSend:     logic.DontCare,
+		ReplicationConfig: *replicationConfig,
 	}

-	m.receiverConfig = endpoint.ReceiverConfig{
-		JobID:                      jobID,
-		RootWithoutClientComponent: m.rootFS,
-		AppendClientIdentity:       false, // !
-		UpdateLastReceivedHold:     true,
-	}
-	if err := m.receiverConfig.Validate(); err != nil {
-		return nil, errors.Wrap(err, "cannot build receiver config")
+	m.receiverConfig, err = buildReceiverConfig(in, jobID)
+	if err != nil {
+		return nil, err
 	}

 	return m, nil
@@ -281,8 +272,6 @@ func activeSide(g *config.Global, in *config.ActiveJob, configJob interface{}) (
 		return nil, errors.Wrap(err, "invalid job name")
 	}

-	j.tasks.concurrency = 1 // FIXME
-
 	switch v := configJob.(type) {
 	case *config.PushJob:
 		j.mode, err = modePushFromConfig(g, v, j.name) // shadow
@@ -368,7 +357,7 @@ func (j *ActiveSide) OwnedDatasetSubtreeRoot() (rfs *zfs.DatasetPath, ok bool) {
 		_ = j.mode.(*modePush) // make sure we didn't introduce a new job type
 		return nil, false
 	}
-	return pull.rootFS.Copy(), true
+	return pull.receiverConfig.RootWithoutClientComponent.Copy(), true
 }

 func (j *ActiveSide) SenderConfig() *endpoint.SenderConfig {
@@ -380,32 +369,31 @@ func (j *ActiveSide) SenderConfig() *endpoint.SenderConfig {
 	return push.senderConfig
 }

-func (j *ActiveSide) SetConcurrency(concurrency int) (err error) {
-	j.updateTasks(func(tasks *activeSideTasks) {
-		if tasks.replicationSetConcurrency != nil {
-			err = tasks.replicationSetConcurrency(concurrency) // no shadow
-			if err == nil {
-				tasks.concurrency = concurrency
-			}
-		} else {
-			// FIXME this is not great, should always be able to set it
-			err = errors.Errorf("cannot set while not replicating")
-		}
-
-	})
-	return err
+// 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)
-	ctx = logging.WithSubsystemLoggers(ctx, log)

 	defer log.Info("job exiting")

 	periodicDone := make(chan struct{})
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
-	go j.mode.RunPeriodic(ctx, periodicDone)
+	periodicCtx, endTask := trace.WithTask(ctx, "periodic")
+	defer endTask()
+	go j.mode.RunPeriodic(periodicCtx, periodicDone)

 	invocationCount := 0
 outer:
@@ -421,17 +409,15 @@ outer:
 		case <-periodicDone:
 		}
 		invocationCount++
-		invLog := log.WithField("invocation", invocationCount)
-		j.do(WithLogger(ctx, invLog))
+		invocationCtx, endSpan := trace.WithSpan(ctx, fmt.Sprintf("invocation-%d", invocationCount))
+		j.do(invocationCtx)
+		endSpan()
 	}
 }

 func (j *ActiveSide) do(ctx context.Context) {

-	log := GetLogger(ctx)
-	ctx = logging.WithSubsystemLoggers(ctx, log)
-	loggers := rpc.GetLoggersOrPanic(ctx) // filled by WithSubsystemLoggers
-	j.mode.ConnectEndpoints(loggers, j.connecter)
+	j.mode.ConnectEndpoints(ctx, j.connecter)
 	defer j.mode.DisconnectEndpoints()

 	// allow cancellation of an invocation (this function)
@@ -454,23 +440,22 @@ func (j *ActiveSide) do(ctx context.Context) {
 			return
 		default:
 		}
+		ctx, endSpan := trace.WithSpan(ctx, "replication")
 		ctx, repCancel := context.WithCancel(ctx)
 		var repWait driver.WaitFunc
 		j.updateTasks(func(tasks *activeSideTasks) {
-			// reset it (almost)
-			old := *tasks
-			*tasks = activeSideTasks{
-				concurrency: old.concurrency,
-			}
-			tasks.replicationCancel = repCancel
-			tasks.replicationReport, repWait, tasks.replicationSetConcurrency = replication.Do(
-				ctx, tasks.concurrency, logic.NewPlanner(j.promRepStateSecs, j.promBytesReplicated, sender, receiver, j.mode.PlannerPolicy()),
+			// reset it
+			*tasks = activeSideTasks{}
+			tasks.replicationCancel = func() { repCancel(); endSpan() }
+			tasks.replicationReport, repWait = replication.Do(
+				ctx, logic.NewPlanner(j.promRepStateSecs, j.promBytesReplicated, sender, receiver, j.mode.PlannerPolicy()),
 			)
 			tasks.state = ActiveSideReplicating
 		})
-		log.Info("start replication")
+		GetLogger(ctx).Info("start replication")
 		repWait(true) // wait blocking
 		repCancel()   // always cancel to free up context resources
+		endSpan()
 	}

 	{
@@ -479,16 +464,18 @@ func (j *ActiveSide) do(ctx context.Context) {
 			return
 		default:
 		}
+		ctx, endSpan := trace.WithSpan(ctx, "prune_sender")
 		ctx, senderCancel := context.WithCancel(ctx)
 		tasks := j.updateTasks(func(tasks *activeSideTasks) {
 			tasks.prunerSender = j.prunerFactory.BuildSenderPruner(ctx, sender, sender)
-			tasks.prunerSenderCancel = senderCancel
+			tasks.prunerSenderCancel = func() { senderCancel(); endSpan() }
 			tasks.state = ActiveSidePruneSender
 		})
-		log.Info("start pruning sender")
+		GetLogger(ctx).Info("start pruning sender")
 		tasks.prunerSender.Prune()
-		log.Info("finished pruning sender")
+		GetLogger(ctx).Info("finished pruning sender")
 		senderCancel()
+		endSpan()
 	}
 	{
 		select {
@@ -496,16 +483,18 @@ func (j *ActiveSide) do(ctx context.Context) {
 			return
 		default:
 		}
+		ctx, endSpan := trace.WithSpan(ctx, "prune_recever")
 		ctx, receiverCancel := context.WithCancel(ctx)
 		tasks := j.updateTasks(func(tasks *activeSideTasks) {
 			tasks.prunerReceiver = j.prunerFactory.BuildReceiverPruner(ctx, receiver, sender)
-			tasks.prunerReceiverCancel = receiverCancel
+			tasks.prunerReceiverCancel = func() { receiverCancel(); endSpan() }
 			tasks.state = ActiveSidePruneReceiver
 		})
-		log.Info("start pruning receiver")
+		GetLogger(ctx).Info("start pruning receiver")
 		tasks.prunerReceiver.Prune()
-		log.Info("finished pruning receiver")
+		GetLogger(ctx).Info("finished pruning receiver")
 		receiverCancel()
+		endSpan()
 	}

 	j.updateTasks(func(tasks *activeSideTasks) {
@@ -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)
+}
@@ -0,0 +1,55 @@
+package job
+
+import (
+	"github.com/pkg/errors"
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon/filters"
+	"github.com/zrepl/zrepl/endpoint"
+	"github.com/zrepl/zrepl/zfs"
+)
+
+type SendingJobConfig interface {
+	GetFilesystems() config.FilesystemsFilter
+	GetSendOptions() *config.SendOptions // must not be nil
+}
+
+func buildSenderConfig(in SendingJobConfig, jobID endpoint.JobID) (*endpoint.SenderConfig, error) {
+
+	fsf, err := filters.DatasetMapFilterFromConfig(in.GetFilesystems())
+	if err != nil {
+		return nil, errors.Wrap(err, "cannot build filesystem filter")
+	}
+
+	return &endpoint.SenderConfig{
+		FSF:     fsf,
+		Encrypt: &zfs.NilBool{B: in.GetSendOptions().Encrypted},
+		JobID:   jobID,
+	}, nil
+}
+
+type ReceivingJobConfig interface {
+	GetRootFS() string
+	GetAppendClientIdentity() bool
+	GetRecvOptions() *config.RecvOptions
+}
+
+func buildReceiverConfig(in ReceivingJobConfig, jobID endpoint.JobID) (rc endpoint.ReceiverConfig, err error) {
+	rootFs, err := zfs.NewDatasetPath(in.GetRootFS())
+	if err != nil {
+		return rc, errors.New("root_fs is not a valid zfs filesystem path")
+	}
+	if rootFs.Length() <= 0 {
+		return rc, errors.New("root_fs must not be empty") // duplicates error check of receiver
+	}
+
+	rc = endpoint.ReceiverConfig{
+		JobID:                      jobID,
+		RootWithoutClientComponent: rootFs,
+		AppendClientIdentity:       in.GetAppendClientIdentity(),
+	}
+	if err := rc.Validate(); err != nil {
+		return rc, errors.Wrap(err, "cannot build receiver config")
+	}
+
+	return rc, nil
+}
@@ -2,12 +2,16 @@ package job

 import (
 	"fmt"
+	"path"
+	"path/filepath"
 	"testing"

+	"github.com/kr/pretty"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"

 	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/transport/tls"
 )

 func TestValidateReceivingSidesDoNotOverlap(t *testing.T) {
@@ -108,3 +112,35 @@ jobs:
 	}

 }
+
+func TestSampleConfigsAreBuiltWithoutErrors(t *testing.T) {
+	paths, err := filepath.Glob("../../config/samples/*")
+	if err != nil {
+		t.Errorf("glob failed: %+v", err)
+	}
+
+	for _, p := range paths {
+
+		if path.Ext(p) != ".yml" {
+			t.Logf("skipping file %s", p)
+			continue
+		}
+
+		t.Run(p, func(t *testing.T) {
+			c, err := config.ParseConfig(p)
+			if err != nil {
+				t.Errorf("error parsing %s:\n%+v", p, err)
+			}
+
+			t.Logf("file: %s", p)
+			t.Log(pretty.Sprint(c))
+
+			tls.FakeCertificateLoading(t)
+			jobs, err := JobsFromConfig(c)
+			t.Logf("jobs: %#v", jobs)
+			assert.NoError(t, err)
+		})
+
+	}
+
+}
@@ -7,6 +7,7 @@ import (

 	"github.com/prometheus/client_golang/prometheus"

+	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/endpoint"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/zfs"
@@ -14,21 +15,8 @@ import (

 type Logger = logger.Logger

-type contextKey int
-
-const (
-	contextKeyLog contextKey = iota
-)
-
 func GetLogger(ctx context.Context) Logger {
-	if l, ok := ctx.Value(contextKeyLog).(Logger); ok {
-		return l
-	}
-	return logger.NewNullLogger()
-}
-
-func WithLogger(ctx context.Context, l Logger) context.Context {
-	return context.WithValue(ctx, contextKeyLog, l)
+	return logging.GetLogger(ctx, logging.SubsysJob)
 }

 type Job interface {
@@ -40,7 +28,6 @@ type Job interface {
 	// must return the root of that subtree as rfs and ok = true
 	OwnedDatasetSubtreeRoot() (rfs *zfs.DatasetPath, ok bool)
 	SenderConfig() *endpoint.SenderConfig
-	SetConcurrency(concurrency int) error
 }

 type Type string
@@ -6,9 +6,9 @@ import (

 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/zrepl/zrepl/daemon/logging/trace"

 	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon/filters"
 	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/daemon/snapper"
 	"github.com/zrepl/zrepl/endpoint"
@@ -47,19 +47,9 @@ func (m *modeSink) SnapperReport() *snapper.Report { return nil }
 func modeSinkFromConfig(g *config.Global, in *config.SinkJob, jobID endpoint.JobID) (m *modeSink, err error) {
 	m = &modeSink{}

-	rootDataset, err := zfs.NewDatasetPath(in.RootFS)
+	m.receiverConfig, err = buildReceiverConfig(in, jobID)
 	if err != nil {
-		return nil, errors.New("root dataset is not a valid zfs filesystem path")
-	}
-
-	m.receiverConfig = endpoint.ReceiverConfig{
-		JobID:                      jobID,
-		RootWithoutClientComponent: rootDataset,
-		AppendClientIdentity:       true, // !
-		UpdateLastReceivedHold:     true,
-	}
-	if err := m.receiverConfig.Validate(); err != nil {
-		return nil, errors.Wrap(err, "cannot build receiver config")
+		return nil, err
 	}

 	return m, nil
@@ -73,17 +63,13 @@ type modeSource struct {
 func modeSourceFromConfig(g *config.Global, in *config.SourceJob, jobID endpoint.JobID) (m *modeSource, err error) {
 	// FIXME exact dedup of modePush
 	m = &modeSource{}
-	fsf, err := filters.DatasetMapFilterFromConfig(in.Filesystems)
+
+	m.senderConfig, err = buildSenderConfig(in, jobID)
 	if err != nil {
-		return nil, errors.Wrap(err, "cannnot build filesystem filter")
-	}
-	m.senderConfig = &endpoint.SenderConfig{
-		FSF:     fsf,
-		Encrypt: &zfs.NilBool{B: in.Send.Encrypted},
-		JobID:   jobID,
+		return nil, errors.Wrap(err, "send options")
 	}

-	if m.snapper, err = snapper.FromConfig(g, fsf, in.Snapshotting); err != nil {
+	if m.snapper, err = snapper.FromConfig(g, m.senderConfig.FSF, in.Snapshotting); err != nil {
 		return nil, errors.Wrap(err, "cannot build snapper")
 	}

@@ -163,15 +149,15 @@ func (j *PassiveSide) SenderConfig() *endpoint.SenderConfig {

 func (*PassiveSide) RegisterMetrics(registerer prometheus.Registerer) {}

-func (*PassiveSide) SetConcurrency(concurrency int) error { return errors.Errorf("not supported") }
-
 func (j *PassiveSide) Run(ctx context.Context) {
-
+	ctx, endTask := trace.WithTaskAndSpan(ctx, "passive-side-job", j.Name())
+	defer endTask()
 	log := GetLogger(ctx)
 	defer log.Info("job exiting")
-	ctx = logging.WithSubsystemLoggers(ctx, log)
 	{
-		ctx, cancel := context.WithCancel(ctx) // shadowing
+		ctx, endTask := trace.WithTask(ctx, "periodic") // shadowing
+		defer endTask()
+		ctx, cancel := context.WithCancel(ctx)
 		defer cancel()
 		go j.mode.RunPeriodic(ctx)
 	}
@@ -181,8 +167,14 @@ func (j *PassiveSide) Run(ctx context.Context) {
 		panic(fmt.Sprintf("implementation error: j.mode.Handler() returned nil: %#v", j))
 	}

-	ctxInterceptor := func(handlerCtx context.Context) context.Context {
-		return logging.WithSubsystemLoggers(handlerCtx, log)
+	ctxInterceptor := func(handlerCtx context.Context, info rpc.HandlerContextInterceptorData, handler func(ctx context.Context)) {
+		// the handlerCtx is clean => need to inherit logging and tracing config from job context
+		handlerCtx = logging.WithInherit(handlerCtx, ctx)
+		handlerCtx = trace.WithInherit(handlerCtx, ctx)
+
+		handlerCtx, endTask := trace.WithTaskAndSpan(handlerCtx, "handler", fmt.Sprintf("job=%q client=%q method=%q", j.Name(), info.ClientIdentity(), info.FullMethod()))
+		defer endTask()
+		handler(handlerCtx)
 	}

 	rpcLoggers := rpc.GetLoggersOrPanic(ctx) // WithSubsystemLoggers above
@@ -2,15 +2,16 @@ package job

 import (
 	"context"
+	"fmt"
 	"sort"

 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/zrepl/zrepl/daemon/logging/trace"

 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/filters"
 	"github.com/zrepl/zrepl/daemon/job/wakeup"
-	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/daemon/pruner"
 	"github.com/zrepl/zrepl/daemon/snapper"
 	"github.com/zrepl/zrepl/endpoint"
@@ -89,15 +90,18 @@ func (j *SnapJob) OwnedDatasetSubtreeRoot() (rfs *zfs.DatasetPath, ok bool) {
 func (j *SnapJob) SenderConfig() *endpoint.SenderConfig { return nil }

 func (j *SnapJob) Run(ctx context.Context) {
+	ctx, endTask := trace.WithTaskAndSpan(ctx, "snap-job", j.Name())
+	defer endTask()
 	log := GetLogger(ctx)
-	ctx = logging.WithSubsystemLoggers(ctx, log)

 	defer log.Info("job exiting")

 	periodicDone := make(chan struct{})
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
-	go j.snapper.Run(ctx, periodicDone)
+	periodicCtx, endTask := trace.WithTask(ctx, "snapshotting")
+	defer endTask()
+	go j.snapper.Run(periodicCtx, periodicDone)

 	invocationCount := 0
 outer:
@@ -112,13 +116,13 @@ outer:
 		case <-periodicDone:
 		}
 		invocationCount++
-		invLog := log.WithField("invocation", invocationCount)
-		j.doPrune(WithLogger(ctx, invLog))
+
+		invocationCtx, endSpan := trace.WithSpan(ctx, fmt.Sprintf("invocation-%d", invocationCount))
+		j.doPrune(invocationCtx)
+		endSpan()
 	}
 }

-func (*SnapJob) SetConcurrency(concurrency int) error { return errors.Errorf("not supported") }
-
 // Adaptor that implements pruner.History around a pruner.Target.
 // The ReplicationCursor method is Get-op only and always returns
 // the filesystem's most recent version's GUID.
@@ -163,8 +167,9 @@ func (h alwaysUpToDateReplicationCursorHistory) ListFilesystems(ctx context.Cont
 }

 func (j *SnapJob) doPrune(ctx context.Context) {
+	ctx, endSpan := trace.WithSpan(ctx, "snap-job-do-prune")
+	defer endSpan()
 	log := GetLogger(ctx)
-	ctx = logging.WithSubsystemLoggers(ctx, log)
 	sender := endpoint.NewSender(endpoint.SenderConfig{
 		JobID: j.name,
 		FSF:   j.fsfilter,
@@ -9,19 +9,10 @@ import (

 	"github.com/mattn/go-isatty"
 	"github.com/pkg/errors"
-
 	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon/hooks"
-	"github.com/zrepl/zrepl/daemon/pruner"
-	"github.com/zrepl/zrepl/daemon/snapper"
-	"github.com/zrepl/zrepl/endpoint"
+	"github.com/zrepl/zrepl/daemon/logging/trace"
 	"github.com/zrepl/zrepl/logger"
-	"github.com/zrepl/zrepl/replication/driver"
-	"github.com/zrepl/zrepl/replication/logic"
-	"github.com/zrepl/zrepl/rpc"
-	"github.com/zrepl/zrepl/rpc/transportmux"
 	"github.com/zrepl/zrepl/tlsconf"
-	"github.com/zrepl/zrepl/transport"
 )

 func OutletsFromConfig(in config.LoggingOutletEnumList) (*logger.Outlets, error) {
@@ -69,8 +60,10 @@ func OutletsFromConfig(in config.LoggingOutletEnumList) (*logger.Outlets, error)
 type Subsystem string

 const (
+	SubsysMeta         Subsystem = "meta"
+	SubsysJob          Subsystem = "job"
 	SubsysReplication  Subsystem = "repl"
-	SubsyEndpoint      Subsystem = "endpoint"
+	SubsysEndpoint     Subsystem = "endpoint"
 	SubsysPruning      Subsystem = "pruning"
 	SubsysSnapshot     Subsystem = "snapshot"
 	SubsysHooks        Subsystem = "hook"
@@ -79,29 +72,104 @@ const (
 	SubsysRPC          Subsystem = "rpc"
 	SubsysRPCControl   Subsystem = "rpc.ctrl"
 	SubsysRPCData      Subsystem = "rpc.data"
+	SubsysZFSCmd       Subsystem = "zfs.cmd"
+	SubsysTraceData    Subsystem = "trace.data"
+	SubsysPlatformtest Subsystem = "platformtest"
 )

-func WithSubsystemLoggers(ctx context.Context, log logger.Logger) context.Context {
-	ctx = logic.WithLogger(ctx, log.WithField(SubsysField, SubsysReplication))
-	ctx = driver.WithLogger(ctx, log.WithField(SubsysField, SubsysReplication))
-	ctx = endpoint.WithLogger(ctx, log.WithField(SubsysField, SubsyEndpoint))
-	ctx = pruner.WithLogger(ctx, log.WithField(SubsysField, SubsysPruning))
-	ctx = snapper.WithLogger(ctx, log.WithField(SubsysField, SubsysSnapshot))
-	ctx = hooks.WithLogger(ctx, log.WithField(SubsysField, SubsysHooks))
-	ctx = transport.WithLogger(ctx, log.WithField(SubsysField, SubsysTransport))
-	ctx = transportmux.WithLogger(ctx, log.WithField(SubsysField, SubsysTransportMux))
-	ctx = rpc.WithLoggers(ctx,
-		rpc.Loggers{
-			General: log.WithField(SubsysField, SubsysRPC),
-			Control: log.WithField(SubsysField, SubsysRPCControl),
-			Data:    log.WithField(SubsysField, SubsysRPCData),
-		},
-	)
-	return ctx
+var AllSubsystems = []Subsystem{
+	SubsysMeta,
+	SubsysJob,
+	SubsysReplication,
+	SubsysEndpoint,
+	SubsysPruning,
+	SubsysSnapshot,
+	SubsysHooks,
+	SubsysTransport,
+	SubsysTransportMux,
+	SubsysRPC,
+	SubsysRPCControl,
+	SubsysRPCData,
+	SubsysZFSCmd,
+	SubsysTraceData,
+	SubsysPlatformtest,
 }

-func LogSubsystem(log logger.Logger, subsys Subsystem) logger.Logger {
-	return log.ReplaceField(SubsysField, subsys)
+type injectedField struct {
+	field  string
+	value  interface{}
+	parent *injectedField
+}
+
+func WithInjectedField(ctx context.Context, field string, value interface{}) context.Context {
+	var parent *injectedField
+	parentI := ctx.Value(contextKeyInjectedField)
+	if parentI != nil {
+		parent = parentI.(*injectedField)
+	}
+	// TODO sanity-check `field` now
+	this := &injectedField{field, value, parent}
+	return context.WithValue(ctx, contextKeyInjectedField, this)
+}
+
+func iterInjectedFields(ctx context.Context, cb func(field string, value interface{})) {
+	injI := ctx.Value(contextKeyInjectedField)
+	if injI == nil {
+		return
+	}
+	inj := injI.(*injectedField)
+	for ; inj != nil; inj = inj.parent {
+		cb(inj.field, inj.value)
+	}
+}
+
+type SubsystemLoggers map[Subsystem]logger.Logger
+
+func SubsystemLoggersWithUniversalLogger(l logger.Logger) SubsystemLoggers {
+	loggers := make(SubsystemLoggers)
+	for _, s := range AllSubsystems {
+		loggers[s] = l
+	}
+	return loggers
+}
+
+func WithLoggers(ctx context.Context, loggers SubsystemLoggers) context.Context {
+	return context.WithValue(ctx, contextKeyLoggers, loggers)
+}
+
+func GetLoggers(ctx context.Context) SubsystemLoggers {
+	loggers, ok := ctx.Value(contextKeyLoggers).(SubsystemLoggers)
+	if !ok {
+		return nil
+	}
+	return loggers
+}
+
+func GetLogger(ctx context.Context, subsys Subsystem) logger.Logger {
+	return getLoggerImpl(ctx, subsys, true)
+}
+
+func getLoggerImpl(ctx context.Context, subsys Subsystem, panicIfEnded bool) logger.Logger {
+	loggers, ok := ctx.Value(contextKeyLoggers).(SubsystemLoggers)
+	if !ok || loggers == nil {
+		return logger.NewNullLogger()
+	}
+	l, ok := loggers[subsys]
+	if !ok {
+		return logger.NewNullLogger()
+	}
+
+	l = l.WithField(SubsysField, subsys)
+
+	l = l.WithField(SpanField, trace.GetSpanStackOrDefault(ctx, *trace.StackKindId, "NOSPAN"))
+
+	fields := make(logger.Fields)
+	iterInjectedFields(ctx, func(field string, value interface{}) {
+		fields[field] = value
+	})
+	l = l.WithFields(fields)
+
+	return l
 }

 func parseLogFormat(i interface{}) (f EntryFormatter, err error) {
@@ -0,0 +1,24 @@
+package logging
+
+import "context"
+
+type contextKey int
+
+const (
+	contextKeyLoggers contextKey = 1 + iota
+	contextKeyInjectedField
+)
+
+var contextKeys = []contextKey{
+	contextKeyLoggers,
+	contextKeyInjectedField,
+}
+
+func WithInherit(ctx, inheritFrom context.Context) context.Context {
+	for _, k := range contextKeys {
+		if v := inheritFrom.Value(k); v != nil {
+			ctx = context.WithValue(ctx, k, v) // no shadow
+		}
+	}
+	return ctx
+}
@@ -22,6 +22,7 @@ const (
 const (
 	JobField    string = "job"
 	SubsysField string = "subsystem"
+	SpanField   string = "span"
 )

 type MetadataFlags int64
@@ -85,7 +86,7 @@ func (f *HumanFormatter) Format(e *logger.Entry) (out []byte, err error) {
 		fmt.Fprintf(&line, "[%s]", col.Sprint(e.Level.Short()))
 	}

-	prefixFields := []string{JobField, SubsysField}
+	prefixFields := []string{JobField, SubsysField, SpanField}
 	prefixed := make(map[string]bool, len(prefixFields)+2)
 	for _, field := range prefixFields {
 		val, ok := e.Fields[field]
@@ -174,8 +175,8 @@ func (f *LogfmtFormatter) Format(e *logger.Entry) ([]byte, error) {
 	}

 	// at least try and put job and task in front
-	prefixed := make(map[string]bool, 2)
-	prefix := []string{JobField, SubsysField}
+	prefixed := make(map[string]bool, 3)
+	prefix := []string{JobField, SubsysField, SpanField}
 	for _, pf := range prefix {
 		v, ok := e.Fields[pf]
 		if !ok {
@@ -211,7 +212,7 @@ func logfmtTryEncodeKeyval(enc *logfmt.Encoder, field, value interface{}) error
 	case logfmt.ErrUnsupportedValueType:
 		err := enc.EncodeKeyval(field, fmt.Sprintf("<%T>", value))
 		if err != nil {
-			return errors.Wrap(err, "cannot encode unsuuported value type Go type")
+			return errors.Wrap(err, "cannot encode unsupported value type Go type")
 		}
 		return nil
 	}
@@ -63,7 +63,7 @@ func NewTCPOutlet(formatter EntryFormatter, network, address string, tlsConfig *
 		return
 	}

-	entryChan := make(chan *bytes.Buffer, 1) // allow one message in flight while previos is in io.Copy()
+	entryChan := make(chan *bytes.Buffer, 1) // allow one message in flight while previous is in io.Copy()

 	o := &TCPOutlet{
 		formatter: formatter,
@@ -0,0 +1,386 @@
+// package trace provides activity tracing via ctx through Tasks and Spans
+//
+// Basic Concepts
+//
+// Tracing can be used to identify where a piece of code spends its time.
+//
+// The Go standard library provides package runtime/trace which is useful to identify CPU bottlenecks or
+// to understand what happens inside the Go runtime.
+// However, it is not ideal for application level tracing, in particular if those traces should be understandable
+// to tech-savvy users (albeit not developers).
+//
+// This package provides the concept of Tasks and Spans to express what activity is happening within an application:
+//
+//  - Neither task nor span is really tangible but instead contained within the context.Context tree
+//  - Tasks represent concurrent activity (i.e. goroutines).
+//  - Spans represent a semantic stack trace within a task.
+//
+// As a consequence, whenever a context is propagated across goroutine boundary, you need to create a child task:
+//
+//   go func(ctx context.Context) {
+//     ctx, endTask = WithTask(ctx, "what-happens-inside-the-child-task")
+//     defer endTask()
+//     // ...
+//   }(ctx)
+//
+// Within the task, you can open up a hierarchy of spans.
+// In contrast to tasks, which have can multiple concurrently running child tasks,
+// spans must nest and not cross the goroutine boundary.
+//
+//  ctx, endSpan = WithSpan(ctx, "copy-dir")
+//  defer endSpan()
+//  for _, f := range dir.Files() {
+//    func() {
+//      ctx, endSpan := WithSpan(ctx, fmt.Sprintf("copy-file %q", f))
+//      defer endspan()
+//      b, _ := ioutil.ReadFile(f)
+//      _ = ioutil.WriteFile(f + ".copy", b, 0600)
+//    }()
+//  }
+//
+// In combination:
+//  ctx, endTask = WithTask(ctx, "copy-dirs")
+//  defer endTask()
+//  for i := range dirs {
+//    go func(dir string) {
+//      ctx, endTask := WithTask(ctx, "copy-dir")
+//      defer endTask()
+//      for _, f := range filesIn(dir) {
+//        func() {
+//          ctx, endSpan := WithSpan(ctx, fmt.Sprintf("copy-file %q", f))
+//          defer endspan()
+//          b, _ := ioutil.ReadFile(f)
+//          _ = ioutil.WriteFile(f + ".copy", b, 0600)
+//        }()
+//      }
+//    }()
+//  }
+//
+// Note that a span ends at the time you call endSpan - not before and not after that.
+// If you violate the stack-like nesting of spans by forgetting an endSpan() invocation,
+// the out-of-order endSpan() will panic.
+//
+// A similar rule applies to the endTask closure returned by WithTask:
+// If a task has live child tasks at the time you call endTask(), the call will panic.
+//
+// Recovering from endSpan() or endTask() panics will corrupt the trace stack and lead to corrupt tracefile output.
+//
+//
+// Best Practices For Naming Tasks And Spans
+//
+// Tasks should always have string constants as names, and must not contain the `#` character. WHy?
+// First, the visualization by chrome://tracing draws a horizontal bar for each task in the trace.
+// Also, the package appends `#NUM` for each concurrently running instance of a task name.
+// Note that the `#NUM` suffix will be reused if a task has ended, in order to avoid an
+// infinite number of horizontal bars in the visualization.
+//
+//
+// Chrome-compatible Tracefile Support
+//
+// The activity trace generated by usage of WithTask and WithSpan can be rendered to a JSON output file
+// that can be loaded into chrome://tracing .
+// Apart from function GetSpanStackOrDefault, this is the main benefit of this package.
+//
+// First, there is a convenience environment variable 'ZREPL_ACTIVITY_TRACE' that can be set to an output path.
+// From process start onward, a trace is written to that path.
+//
+// More consumers can attach to the activity trace through the ChrometraceClientWebsocketHandler websocket handler.
+//
+// If a write error is encountered with any consumer (including the env-var based one), the consumer is closed and
+// will not receive further trace output.
+package trace
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/zrepl/zrepl/util/chainlock"
+)
+
+var metrics struct {
+	activeTasks prometheus.Gauge
+}
+var taskNamer *uniqueConcurrentTaskNamer = newUniqueTaskNamer()
+
+func init() {
+	metrics.activeTasks = prometheus.NewGauge(prometheus.GaugeOpts{
+		Namespace: "zrepl",
+		Subsystem: "trace",
+		Name:      "active_tasks",
+		Help:      "number of active (tracing-level) tasks in the daemon",
+	})
+}
+
+func RegisterMetrics(r prometheus.Registerer) {
+	r.MustRegister(metrics.activeTasks)
+}
+
+type traceNode struct {
+	id         string
+	annotation string
+	parentTask *traceNode
+
+	mtx chainlock.L
+
+	activeChildTasks int32 // only for task nodes, insignificant for span nodes
+	parentSpan       *traceNode
+	activeChildSpan  *traceNode // nil if task or span doesn't have an active child span
+
+	startedAt time.Time
+	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.
+// Wrong call order / forgetting to call it will result in panics.
+type DoneFunc func()
+
+var ErrTaskStillHasActiveChildTasks = fmt.Errorf("end task: task still has active child tasks")
+
+// Start a new root task or create a child task of an existing task.
+//
+// This is required when starting a new goroutine and
+// passing an existing task context to it.
+//
+// taskName should be a constantand must not contain '#'
+//
+// The implementation ensures that,
+// if multiple tasks with the same name exist simultaneously,
+// a unique suffix is appended to uniquely identify the task opened with this function.
+func WithTask(ctx context.Context, taskName string) (context.Context, DoneFunc) {
+
+	var parentTask *traceNode
+	nodeI := ctx.Value(contextKeyTraceNode)
+	if nodeI != nil {
+		node := nodeI.(*traceNode)
+		if node.parentSpan != nil {
+			parentTask = node.parentTask
+		} else {
+			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)
+
+	this := &traceNode{
+		id:               genID(),
+		annotation:       taskName,
+		parentTask:       parentTask,
+		activeChildTasks: 0,
+		parentSpan:       nil,
+		activeChildSpan:  nil,
+
+		startedAt: time.Now(),
+		endedAt:   time.Time{},
+	}
+
+	if parentTask != nil {
+		this.parentTask.activeChildTasks++
+		parentTask.mtx.Unlock()
+	}
+
+	ctx = context.WithValue(ctx, contextKeyTraceNode, this)
+
+	chrometraceBeginTask(this)
+
+	metrics.activeTasks.Inc()
+
+	endTaskFunc := func() {
+
+		// only hold locks while manipulating the tree
+		// (trace writer might block too long and unlike spans, tasks are updated concurrently)
+		alreadyEnded := func() (alreadyEnded bool) {
+			if this.parentTask != nil {
+				defer this.parentTask.mtx.Lock().Unlock()
+			}
+			defer this.mtx.Lock().Unlock()
+
+			if this.activeChildTasks != 0 {
+				panic(errors.Wrapf(ErrTaskStillHasActiveChildTasks, "end task: %v active child tasks", this.activeChildTasks))
+			}
+
+			// support idempotent task ends
+			if !this.endedAt.IsZero() {
+				return true
+			}
+			this.endedAt = time.Now()
+
+			if this.parentTask != nil {
+				this.parentTask.activeChildTasks--
+				if this.parentTask.activeChildTasks < 0 {
+					panic("impl error: parent task with negative activeChildTasks count")
+				}
+			}
+			return false
+		}()
+		if alreadyEnded {
+			return
+		}
+
+		chrometraceEndTask(this)
+
+		metrics.activeTasks.Dec()
+
+		taskNameDone()
+	}
+
+	return ctx, endTaskFunc
+}
+
+var ErrAlreadyActiveChildSpan = fmt.Errorf("create child span: span already has an active child span")
+var ErrSpanStillHasActiveChildSpan = fmt.Errorf("end span: span still has active child spans")
+
+// Start a new span.
+// Important: ctx must have an active task (see WithTask)
+func WithSpan(ctx context.Context, annotation string) (context.Context, DoneFunc) {
+	var parentSpan, parentTask *traceNode
+	nodeI := ctx.Value(contextKeyTraceNode)
+	if nodeI != nil {
+		parentSpan = nodeI.(*traceNode)
+		if parentSpan.parentSpan == nil {
+			parentTask = parentSpan
+		} else {
+			parentTask = parentSpan.parentTask
+		}
+	} else {
+		panic("must be called from within a task")
+	}
+
+	this := &traceNode{
+		id:              genID(),
+		annotation:      annotation,
+		parentTask:      parentTask,
+		parentSpan:      parentSpan,
+		activeChildSpan: nil,
+
+		startedAt: time.Now(),
+		endedAt:   time.Time{},
+	}
+
+	parentSpan.mtx.HoldWhile(func() {
+		if parentSpan.activeChildSpan != nil {
+			panic(ErrAlreadyActiveChildSpan)
+		}
+		parentSpan.activeChildSpan = this
+	})
+
+	ctx = context.WithValue(ctx, contextKeyTraceNode, this)
+	chrometraceBeginSpan(this)
+	callbackEndSpan := callbackBeginSpan(ctx)
+
+	endTaskFunc := func() {
+
+		defer parentSpan.mtx.Lock().Unlock()
+		if parentSpan.activeChildSpan != this && this.endedAt.IsZero() {
+			panic("impl error: activeChildSpan should not change while != nil because there can only be one")
+		}
+
+		defer this.mtx.Lock().Unlock()
+		if this.activeChildSpan != nil {
+			panic(ErrSpanStillHasActiveChildSpan)
+		}
+
+		if !this.endedAt.IsZero() {
+			return // support idempotent span ends
+		}
+
+		parentSpan.activeChildSpan = nil
+		this.endedAt = time.Now()
+
+		chrometraceEndSpan(this)
+		callbackEndSpan(this)
+	}
+
+	return ctx, endTaskFunc
+}
+
+type StackKind struct {
+	symbolizeTask func(t *traceNode) string
+	symbolizeSpan func(s *traceNode) string
+}
+
+var (
+	StackKindId = &StackKind{
+		symbolizeTask: func(t *traceNode) string { return t.id },
+		symbolizeSpan: func(s *traceNode) string { return s.id },
+	}
+	SpanStackKindCombined = &StackKind{
+		symbolizeTask: func(t *traceNode) string { return fmt.Sprintf("(%s %q)", t.id, t.annotation) },
+		symbolizeSpan: func(s *traceNode) string { return fmt.Sprintf("(%s %q)", s.id, s.annotation) },
+	}
+	SpanStackKindAnnotation = &StackKind{
+		symbolizeTask: func(t *traceNode) string { return t.annotation },
+		symbolizeSpan: func(s *traceNode) string { return s.annotation },
+	}
+)
+
+func (n *traceNode) task() *traceNode {
+	task := n.parentTask
+	if n.parentSpan == nil {
+		task = n
+	}
+	return task
+}
+
+func (n *traceNode) TaskName() string {
+	task := n.task()
+	return task.annotation
+}
+
+func (this *traceNode) TaskAndSpanStack(kind *StackKind) (spanIdStack string) {
+	task := this.task()
+
+	var spansInTask []*traceNode
+	for s := this; s != nil; s = s.parentSpan {
+		spansInTask = append(spansInTask, s)
+	}
+
+	var tasks []*traceNode
+	for t := task; t != nil; t = t.parentTask {
+		tasks = append(tasks, t)
+	}
+
+	var taskIdsRev []string
+	for i := len(tasks) - 1; i >= 0; i-- {
+		taskIdsRev = append(taskIdsRev, kind.symbolizeTask(tasks[i]))
+	}
+
+	var spanIdsRev []string
+	for i := len(spansInTask) - 1; i >= 0; i-- {
+		spanIdsRev = append(spanIdsRev, kind.symbolizeSpan(spansInTask[i]))
+	}
+
+	taskStack := strings.Join(taskIdsRev, "$")
+	return fmt.Sprintf("%s$%s", taskStack, strings.Join(spanIdsRev, "."))
+}
+
+func GetSpanStackOrDefault(ctx context.Context, kind StackKind, def string) string {
+	if nI := ctx.Value(contextKeyTraceNode); nI != nil {
+		n := nI.(*traceNode)
+		return n.TaskAndSpanStack(StackKindId)
+	} else {
+		return def
+	}
+}
@@ -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)
+			}
+		}
+	}
+}
@@ -0,0 +1,230 @@
+package trace
+
+// The functions in this file are concerned with the generation
+// of trace files based on the information from WithTask and WithSpan.
+//
+// The emitted trace files are open-ended array of JSON objects
+// that follow the Chrome trace file format:
+//   https://docs.google.com/document/d/1CvAClvFfyA5R-PhYUmn5OOQtYMH4h6I0nSsKchNAySU/preview
+//
+// The emitted JSON can be loaded into Chrome's chrome://tracing view.
+//
+// The trace file can be written to a file whose path is specified in an env file,
+// and be written to web sockets established on ChrometraceHttpHandler
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+	"sync"
+	"sync/atomic"
+
+	"golang.org/x/net/websocket"
+
+	"github.com/zrepl/zrepl/util/envconst"
+)
+
+var chrometracePID string
+
+func init() {
+	var err error
+	chrometracePID, err = os.Hostname()
+	if err != nil {
+		panic(err)
+	}
+	chrometracePID = fmt.Sprintf("%q", chrometracePID)
+}
+
+type chrometraceEvent struct {
+	Cat                       string   `json:"cat,omitempty"`
+	Name                      string   `json:"name"`
+	Stack                     []string `json:"stack,omitempty"`
+	Phase                     string   `json:"ph"`
+	TimestampUnixMicroseconds int64    `json:"ts"`
+	DurationMicroseconds      int64    `json:"dur,omitempty"`
+	Pid                       string   `json:"pid"`
+	Tid                       string   `json:"tid"`
+	Id                        string   `json:"id,omitempty"`
+}
+
+func chrometraceBeginSpan(s *traceNode) {
+	taskName := s.TaskName()
+	chrometraceWrite(chrometraceEvent{
+		Name:                      s.annotation,
+		Phase:                     "B",
+		TimestampUnixMicroseconds: s.startedAt.UnixNano() / 1000,
+		Pid:                       chrometracePID,
+		Tid:                       taskName,
+	})
+}
+
+func chrometraceEndSpan(s *traceNode) {
+	taskName := s.TaskName()
+	chrometraceWrite(chrometraceEvent{
+		Name:                      s.annotation,
+		Phase:                     "E",
+		TimestampUnixMicroseconds: s.endedAt.UnixNano() / 1000,
+		Pid:                       chrometracePID,
+		Tid:                       taskName,
+	})
+}
+
+var chrometraceFlowId uint64
+
+func chrometraceBeginTask(s *traceNode) {
+	chrometraceBeginSpan(s)
+
+	if s.parentTask == nil {
+		return
+	}
+	// beginning of a task that has a parent
+	// => use flow events to link parent and child
+
+	flowId := atomic.AddUint64(&chrometraceFlowId, 1)
+	flowIdStr := fmt.Sprintf("%x", flowId)
+
+	parentTask := s.parentTask.TaskName()
+	chrometraceWrite(chrometraceEvent{
+		Cat:                       "task", // seems to be necessary, otherwise the GUI shows some `indexOf` JS error
+		Name:                      "child-task",
+		Phase:                     "s",
+		TimestampUnixMicroseconds: s.startedAt.UnixNano() / 1000, // yes, the child's timestamp (=> from-point of the flow line is at right x-position of parent's bar)
+		Pid:                       chrometracePID,
+		Tid:                       parentTask,
+		Id:                        flowIdStr,
+	})
+
+	childTask := s.TaskName()
+	if parentTask == childTask {
+		panic(parentTask)
+	}
+	chrometraceWrite(chrometraceEvent{
+		Cat:                       "task", // seems to be necessary, otherwise the GUI shows some `indexOf` JS error
+		Name:                      "child-task",
+		Phase:                     "f",
+		TimestampUnixMicroseconds: s.startedAt.UnixNano() / 1000,
+		Pid:                       chrometracePID,
+		Tid:                       childTask,
+		Id:                        flowIdStr,
+	})
+}
+
+func chrometraceEndTask(s *traceNode) {
+	chrometraceEndSpan(s)
+}
+
+type chrometraceConsumerRegistration struct {
+	w io.Writer
+	// errored must have capacity 1, the writer thread will send to it non-blocking, then close it
+	errored chan error
+}
+
+var chrometraceConsumers struct {
+	register  chan chrometraceConsumerRegistration
+	consumers map[chrometraceConsumerRegistration]bool
+	write     chan []byte
+}
+
+func init() {
+	chrometraceConsumers.register = make(chan chrometraceConsumerRegistration)
+	chrometraceConsumers.consumers = make(map[chrometraceConsumerRegistration]bool)
+	chrometraceConsumers.write = make(chan []byte)
+	go func() {
+		kickConsumer := func(c chrometraceConsumerRegistration, err error) {
+			debug("chrometrace kicking consumer %#v after error %v", c, err)
+			select {
+			case c.errored <- err:
+			default:
+			}
+			close(c.errored)
+			delete(chrometraceConsumers.consumers, c)
+		}
+		for {
+			select {
+			case reg := <-chrometraceConsumers.register:
+				debug("registered chrometrace consumer %#v", reg)
+				chrometraceConsumers.consumers[reg] = true
+				n, err := reg.w.Write([]byte("[\n"))
+				if err != nil {
+					kickConsumer(reg, err)
+				} else if n != 2 {
+					kickConsumer(reg, fmt.Errorf("short write: %v", n))
+				}
+				// successfully registered
+
+			case buf := <-chrometraceConsumers.write:
+				debug("chrometrace write request: %s", string(buf))
+				var r bytes.Reader
+				for c := range chrometraceConsumers.consumers {
+					r.Reset(buf)
+					n, err := io.Copy(c.w, &r)
+					debug("chrometrace wrote n=%v bytes to consumer %#v", n, c)
+					if err != nil {
+						kickConsumer(c, err)
+					}
+				}
+			}
+		}
+	}()
+}
+
+func chrometraceWrite(i interface{}) {
+	var buf bytes.Buffer
+	err := json.NewEncoder(&buf).Encode(i)
+	if err != nil {
+		panic(err)
+	}
+	buf.WriteString(",")
+	chrometraceConsumers.write <- buf.Bytes()
+}
+
+func ChrometraceClientWebsocketHandler(conn *websocket.Conn) {
+	defer conn.Close()
+
+	var wg sync.WaitGroup
+	defer wg.Wait()
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		r := bufio.NewReader(conn)
+		_, _, _ = r.ReadLine() // ignore errors
+		conn.Close()
+	}()
+
+	errored := make(chan error, 1)
+	chrometraceConsumers.register <- chrometraceConsumerRegistration{
+		w:       conn,
+		errored: errored,
+	}
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		<-errored
+		conn.Close()
+	}()
+}
+
+var chrometraceFileConsumerPath = envconst.String("ZREPL_ACTIVITY_TRACE", "")
+
+func init() {
+	if chrometraceFileConsumerPath != "" {
+		var err error
+		f, err := os.Create(chrometraceFileConsumerPath)
+		if err != nil {
+			panic(err)
+		}
+		errored := make(chan error, 1)
+		chrometraceConsumers.register <- chrometraceConsumerRegistration{
+			w:       f,
+			errored: errored,
+		}
+		go func() {
+			<-errored
+			f.Close()
+		}()
+	}
+}
@@ -0,0 +1,27 @@
+package trace
+
+import "context"
+
+type contextKey int
+
+const (
+	contextKeyTraceNode contextKey = 1 + iota
+)
+
+var contextKeys = []contextKey{
+	contextKeyTraceNode,
+}
+
+// WithInherit inherits the task hierarchy from inheritFrom into ctx.
+// The returned context is a child of ctx, but its task and span are those of inheritFrom.
+//
+// Note that in most use cases, callers most likely want to call WithTask since it will most likely
+// be in some sort of connection handler context.
+func WithInherit(ctx, inheritFrom context.Context) context.Context {
+	for _, k := range contextKeys {
+		if v := inheritFrom.Value(k); v != nil {
+			ctx = context.WithValue(ctx, k, v) // no shadow
+		}
+	}
+	return ctx
+}
@@ -0,0 +1,74 @@
+package trace
+
+import (
+	"context"
+	"fmt"
+	"runtime"
+	"strings"
+	"sync"
+)
+
+// use like this:
+//
+//	defer WithSpanFromStackUpdateCtx(&existingCtx)()
+//
+//
+func WithSpanFromStackUpdateCtx(ctx *context.Context) DoneFunc {
+	childSpanCtx, end := WithSpan(*ctx, getMyCallerOrPanic())
+	*ctx = childSpanCtx
+	return end
+}
+
+// derive task name from call stack (caller's name)
+func WithTaskFromStack(ctx context.Context) (context.Context, DoneFunc) {
+	return WithTask(ctx, getMyCallerOrPanic())
+}
+
+// derive task name from call stack (caller's name) and update *ctx
+// to point to be the child task ctx
+func WithTaskFromStackUpdateCtx(ctx *context.Context) DoneFunc {
+	child, end := WithTask(*ctx, getMyCallerOrPanic())
+	*ctx = child
+	return end
+}
+
+// create a task and a span within it in one call
+func WithTaskAndSpan(ctx context.Context, task string, span string) (context.Context, DoneFunc) {
+	ctx, endTask := WithTask(ctx, task)
+	ctx, endSpan := WithSpan(ctx, fmt.Sprintf("%s %s", task, span))
+	return ctx, func() {
+		endSpan()
+		endTask()
+	}
+}
+
+// create a span during which several child tasks are spawned using the `add` function
+func WithTaskGroup(ctx context.Context, taskGroup string) (_ context.Context, add func(f func(context.Context)), waitEnd DoneFunc) {
+	var wg sync.WaitGroup
+	ctx, endSpan := WithSpan(ctx, taskGroup)
+	add = func(f func(context.Context)) {
+		wg.Add(1)
+		defer wg.Done()
+		ctx, endTask := WithTask(ctx, taskGroup)
+		defer endTask()
+		f(ctx)
+	}
+	waitEnd = func() {
+		wg.Wait()
+		endSpan()
+	}
+	return ctx, add, waitEnd
+}
+
+func getMyCallerOrPanic() string {
+	pc, _, _, ok := runtime.Caller(2)
+	if !ok {
+		panic("cannot get caller")
+	}
+	details := runtime.FuncForPC(pc)
+	if ok && details != nil {
+		const prefix = "github.com/zrepl/zrepl"
+		return strings.TrimPrefix(strings.TrimPrefix(details.Name(), prefix), "/")
+	}
+	return ""
+}
@@ -0,0 +1,16 @@
+package trace
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGetCallerOrPanic(t *testing.T) {
+	withStackFromCtxMock := func() string {
+		return getMyCallerOrPanic()
+	}
+	ret := withStackFromCtxMock()
+	// zrepl prefix is stripped
+	assert.Equal(t, "daemon/logging/trace.TestGetCallerOrPanic", ret)
+}
@@ -0,0 +1,15 @@
+package trace
+
+import (
+	"fmt"
+	"os"
+)
+
+const debugEnabled = false
+
+func debug(format string, args ...interface{}) {
+	if !debugEnabled {
+		return
+	}
+	fmt.Fprintf(os.Stderr, format+"\n", args...)
+}
@@ -0,0 +1,47 @@
+package trace
+
+import (
+	"encoding/base64"
+	"math/rand"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/zrepl/zrepl/util/envconst"
+)
+
+var genIdPRNG = rand.New(rand.NewSource(1))
+
+func init() {
+	genIdPRNG.Seed(time.Now().UnixNano())
+	genIdPRNG.Seed(int64(os.Getpid()))
+}
+
+var genIdNumBytes = envconst.Int("ZREPL_TRACE_ID_NUM_BYTES", 3)
+
+func init() {
+	if genIdNumBytes < 1 {
+		panic("trace node id byte length must be at least 1")
+	}
+}
+
+func genID() string {
+	var out strings.Builder
+	enc := base64.NewEncoder(base64.RawStdEncoding, &out)
+	buf := make([]byte, genIdNumBytes)
+	for i := 0; i < len(buf); {
+		n, err := genIdPRNG.Read(buf[i:])
+		if err != nil {
+			panic(err)
+		}
+		i += n
+	}
+	n, err := enc.Write(buf[:])
+	if err != nil || n != len(buf) {
+		panic(err)
+	}
+	if err := enc.Close(); err != nil {
+		panic(err)
+	}
+	return out.String()
+}
@@ -0,0 +1,205 @@
+package trace
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/gitchander/permutation"
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRegularSpanUsage(t *testing.T) {
+	root, endRoot := WithTask(context.Background(), "root")
+	defer endRoot()
+
+	s1, endS1 := WithSpan(root, "parent")
+	s2, endS2 := WithSpan(s1, "child")
+	_, endS3 := WithSpan(s2, "grand-child")
+	require.NotPanics(t, func() { endS3() })
+	require.NotPanics(t, func() { endS2() })
+
+	// reuse
+	_, endS4 := WithSpan(s1, "child-2")
+	require.NotPanics(t, func() { endS4() })
+
+	// close parent
+	require.NotPanics(t, func() { endS1() })
+}
+
+func TestMultipleActiveChildSpansNotAllowed(t *testing.T) {
+	root, endRoot := WithTask(context.Background(), "root")
+	defer endRoot()
+
+	s1, _ := WithSpan(root, "s1")
+	_, endS2 := WithSpan(s1, "s1-child1")
+
+	require.PanicsWithValue(t, ErrAlreadyActiveChildSpan, func() {
+		_, _ = WithSpan(s1, "s1-child2")
+	})
+
+	endS2()
+
+	require.NotPanics(t, func() {
+		_, _ = WithSpan(s1, "s1-child2")
+	})
+}
+
+func TestForkingChildSpansNotAllowed(t *testing.T) {
+	root, endRoot := WithTask(context.Background(), "root")
+	defer endRoot()
+
+	s1, _ := WithSpan(root, "s1")
+	sc, endSC := WithSpan(s1, "s1-child")
+	_, _ = WithSpan(sc, "s1-child-child")
+
+	require.PanicsWithValue(t, ErrSpanStillHasActiveChildSpan, func() {
+		endSC()
+	})
+}
+
+func TestRegularTaskUsage(t *testing.T) {
+	// assert concurrent activities on different tasks can end in any order
+	closeOrder := []int{0, 1, 2}
+	closeOrders := permutation.New(permutation.IntSlice(closeOrder))
+	for closeOrders.Next() {
+		t.Run(fmt.Sprintf("%v", closeOrder), func(t *testing.T) {
+			root, endRoot := WithTask(context.Background(), "root")
+			defer endRoot()
+
+			c1, endC1 := WithTask(root, "c1")
+			defer endC1()
+			c2, endC2 := WithTask(root, "c2")
+			defer endC2()
+
+			// begin 3 concurrent activities
+			_, endAR := WithSpan(root, "aR")
+			_, endAC1 := WithSpan(c1, "aC1")
+			_, endAC2 := WithSpan(c2, "aC2")
+
+			endFuncs := []DoneFunc{endAR, endAC1, endAC2}
+			for _, i := range closeOrder {
+				require.NotPanics(t, func() {
+					endFuncs[i]()
+				}, "%v", i)
+			}
+		})
+	}
+}
+
+func TestTaskEndWithActiveChildTaskNotAllowed(t *testing.T) {
+	root, _ := WithTask(context.Background(), "root")
+	c, endC := WithTask(root, "child")
+	_, _ = WithTask(c, "grand-child")
+	func() {
+		defer func() {
+			r := recover()
+			require.NotNil(t, r)
+			err, ok := r.(error)
+			require.True(t, ok)
+			require.Equal(t, ErrTaskStillHasActiveChildTasks, errors.Cause(err))
+		}()
+		endC()
+	}()
+
+}
+
+func TestIdempotentEndTask(t *testing.T) {
+	_, end := WithTask(context.Background(), "root")
+	end()
+	require.NotPanics(t, func() { end() })
+}
+
+func TestSpansPanicIfNoParentTask(t *testing.T) {
+	require.Panics(t, func() { WithSpan(context.Background(), "taskless-span") })
+}
+
+func TestIdempotentEndSpan(t *testing.T) {
+	root, _ := WithTask(context.Background(), "root")
+	_, end := WithSpan(root, "span")
+	end()
+	require.NotPanics(t, func() { end() })
+}
+
+func logAndGetTraceNode(t *testing.T, descr string, ctx context.Context) *traceNode {
+	n, ok := ctx.Value(contextKeyTraceNode).(*traceNode)
+	require.True(t, ok)
+	t.Logf("% 20s %p %#v", descr, n, n)
+	return n
+}
+
+func TestWhiteboxHierachy(t *testing.T) {
+	root, e1 := WithTask(context.Background(), "root")
+	rootN := logAndGetTraceNode(t, "root", root)
+	assert.Nil(t, rootN.parentTask)
+	assert.Nil(t, rootN.parentSpan)
+
+	child, e2 := WithSpan(root, "child")
+	childN := logAndGetTraceNode(t, "child", child)
+	assert.Equal(t, rootN, childN.parentTask)
+	assert.Equal(t, rootN, childN.parentSpan)
+
+	grandchild, e3 := WithSpan(child, "grandchild")
+	grandchildN := logAndGetTraceNode(t, "grandchild", grandchild)
+	assert.Equal(t, rootN, grandchildN.parentTask)
+	assert.Equal(t, childN, grandchildN.parentSpan)
+
+	gcTask, e4 := WithTask(grandchild, "grandchild-task")
+	gcTaskN := logAndGetTraceNode(t, "grandchild-task", gcTask)
+	assert.Equal(t, rootN, gcTaskN.parentTask)
+	assert.Nil(t, gcTaskN.parentSpan)
+
+	// it is allowed that a child task outlives the _span_ in which it was created
+	// (albeit not its parent task)
+	e3()
+	e2()
+	gcTaskSpan, e5 := WithSpan(gcTask, "granschild-task-span")
+	gcTaskSpanN := logAndGetTraceNode(t, "granschild-task-span", gcTaskSpan)
+	assert.Equal(t, gcTaskN, gcTaskSpanN.parentTask)
+	assert.Equal(t, gcTaskN, gcTaskSpanN.parentSpan)
+	e5()
+
+	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()
+	})
+}
@@ -0,0 +1,53 @@
+package trace
+
+import (
+	"fmt"
+	"strings"
+	"sync"
+
+	"github.com/willf/bitset"
+)
+
+type uniqueConcurrentTaskNamer struct {
+	mtx    sync.Mutex
+	active map[string]*bitset.BitSet
+}
+
+// bitvecLengthGauge may be nil
+func newUniqueTaskNamer() *uniqueConcurrentTaskNamer {
+	return &uniqueConcurrentTaskNamer{
+		active: make(map[string]*bitset.BitSet),
+	}
+}
+
+// appends `#%d` to `name` such that until `done` is called,
+// it is guaranteed that `#%d` is not returned a second time for the same `name`
+func (namer *uniqueConcurrentTaskNamer) UniqueConcurrentTaskName(name string) (uniqueName string, done func()) {
+	if strings.Contains(name, "#") {
+		panic(name)
+	}
+	namer.mtx.Lock()
+	act, ok := namer.active[name]
+	if !ok {
+		act = bitset.New(64) // FIXME magic const
+		namer.active[name] = act
+	}
+	id, ok := act.NextClear(0)
+	if !ok {
+		// if !ok, all bits are 1 and act.Len() returns the next bit
+		id = act.Len()
+		// FIXME unbounded growth without reclamation
+	}
+	act.Set(id)
+	namer.mtx.Unlock()
+
+	return fmt.Sprintf("%s#%d", name, id), func() {
+		namer.mtx.Lock()
+		defer namer.mtx.Unlock()
+		act, ok := namer.active[name]
+		if !ok {
+			panic("must be initialized upon entry")
+		}
+		act.Clear(id)
+	}
+}
@@ -0,0 +1,58 @@
+package trace
+
+import (
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/willf/bitset"
+)
+
+func TestBitsetFeaturesForUniqueConcurrentTaskNamer(t *testing.T) {
+	var b bitset.BitSet
+	require.Equal(t, uint(0), b.Len())
+	require.Equal(t, uint(0), b.Count())
+
+	b.Set(0)
+	require.Equal(t, uint(1), b.Len())
+	require.Equal(t, uint(1), b.Count())
+
+	b.Set(8)
+	require.Equal(t, uint(9), b.Len())
+	require.Equal(t, uint(2), b.Count())
+
+	b.Set(1)
+	require.Equal(t, uint(9), b.Len())
+	require.Equal(t, uint(3), b.Count())
+}
+
+func TestUniqueConcurrentTaskNamer(t *testing.T) {
+	namer := newUniqueTaskNamer()
+
+	var wg sync.WaitGroup
+
+	const N = 8128
+	const Q = 23
+
+	var fails uint32
+	var m sync.Map
+	wg.Add(N)
+	for i := 0; i < N; i++ {
+		go func(i int) {
+			defer wg.Done()
+			name := fmt.Sprintf("%d", i/Q)
+			uniqueName, done := namer.UniqueConcurrentTaskName(name)
+			act, _ := m.LoadOrStore(uniqueName, i)
+			if act.(int) != i {
+				atomic.AddUint32(&fails, 1)
+			}
+			m.Delete(uniqueName)
+			done()
+		}(i)
+	}
+	wg.Wait()
+
+	require.Equal(t, uint32(0), fails)
+}
@@ -1,6 +1,8 @@
 package daemon

 import (
+	"context"
+
 	"github.com/zrepl/zrepl/cli"
 	"github.com/zrepl/zrepl/logger"
 )
@@ -10,7 +12,7 @@ type Logger = logger.Logger
 var DaemonCmd = &cli.Subcommand{
 	Use:   "daemon",
 	Short: "run the zrepl daemon",
-	Run: func(subcommand *cli.Subcommand, args []string) error {
-		return Run(subcommand.Config())
+	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
+		return Run(ctx, subcommand.Config())
 	},
 }
@@ -8,7 +8,11 @@ import (
 	"net"
 	"net/http/pprof"

+	"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 {
@@ -64,6 +68,8 @@ 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)
 				if ctx.Err() != nil {
@@ -5,12 +5,12 @@ import (
 	"net"
 	"net/http"

-	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"

 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/job"
+	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/endpoint"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/rpc/dataconn/frameconn"
@@ -52,8 +52,6 @@ func (j *prometheusJob) OwnedDatasetSubtreeRoot() (p *zfs.DatasetPath, ok bool)

 func (j *prometheusJob) SenderConfig() *endpoint.SenderConfig { return nil }

-func (j *prometheusJob) SetConcurrency(concurrency int) error { return errors.Errorf("not supported") }
-
 func (j *prometheusJob) RegisterMetrics(registerer prometheus.Registerer) {}

 func (j *prometheusJob) Run(ctx context.Context) {
@@ -89,16 +87,19 @@ func (j *prometheusJob) Run(ctx context.Context) {
 }

 type prometheusJobOutlet struct {
-	jobName string
 }

 var _ logger.Outlet = prometheusJobOutlet{}

-func newPrometheusLogOutlet(jobName string) prometheusJobOutlet {
-	return prometheusJobOutlet{jobName}
+func newPrometheusLogOutlet() prometheusJobOutlet {
+	return prometheusJobOutlet{}
 }

 func (o prometheusJobOutlet) WriteEntry(entry logger.Entry) error {
-	prom.taskLogEntries.WithLabelValues(o.jobName, entry.Level.String()).Inc()
+	jobFieldVal, ok := entry.Fields[logging.JobField].(string)
+	if !ok {
+		jobFieldVal = "_nojobid"
+	}
+	prom.taskLogEntries.WithLabelValues(jobFieldVal, entry.Level.String()).Inc()
 	return nil
 }
@@ -12,19 +12,20 @@ import (
 	"github.com/prometheus/client_golang/prometheus"

 	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/pruning"
 	"github.com/zrepl/zrepl/replication/logic/pdu"
 	"github.com/zrepl/zrepl/util/envconst"
 )

-// Try to keep it compatible with gitub.com/zrepl/zrepl/endpoint.Endpoint
+// 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)
 	ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error)
 }

-// Try to keep it compatible with gitub.com/zrepl/zrepl/endpoint.Endpoint
+// 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)
@@ -35,17 +36,13 @@ type Logger = logger.Logger

 type contextKey int

-const contextKeyLogger contextKey = 0
-
-func WithLogger(ctx context.Context, log Logger) context.Context {
-	return context.WithValue(ctx, contextKeyLogger, log)
-}
+const (
+	contextKeyPruneSide contextKey = 1 + iota
+)

 func GetLogger(ctx context.Context) Logger {
-	if l, ok := ctx.Value(contextKeyLogger).(Logger); ok {
-		return l
-	}
-	return logger.NewNullLogger()
+	pruneSide := ctx.Value(contextKeyPruneSide).(string)
+	return logging.GetLogger(ctx, logging.SubsysPruning).WithField("prune_side", pruneSide)
 }

 type args struct {
@@ -138,7 +135,7 @@ func NewPrunerFactory(in config.PruningSenderReceiver, promPruneSecs *prometheus
 func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, target Target, receiver History) *Pruner {
 	p := &Pruner{
 		args: args{
-			WithLogger(ctx, GetLogger(ctx).WithField("prune_side", "sender")),
+			context.WithValue(ctx, contextKeyPruneSide, "sender"),
 			target,
 			receiver,
 			f.senderRules,
@@ -154,7 +151,7 @@ func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, target Target, re
 func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target, receiver History) *Pruner {
 	p := &Pruner{
 		args: args{
-			WithLogger(ctx, GetLogger(ctx).WithField("prune_side", "receiver")),
+			context.WithValue(ctx, contextKeyPruneSide, "receiver"),
 			target,
 			receiver,
 			f.receiverRules,
@@ -170,7 +167,7 @@ func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target,
 func (f *LocalPrunerFactory) BuildLocalPruner(ctx context.Context, target Target, receiver History) *Pruner {
 	p := &Pruner{
 		args: args{
-			ctx,
+			context.WithValue(ctx, contextKeyPruneSide, "local"),
 			target,
 			receiver,
 			f.keepRules,
@@ -8,10 +8,11 @@ import (
 	"time"

 	"github.com/pkg/errors"
+	"github.com/zrepl/zrepl/daemon/logging/trace"

 	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon/filters"
 	"github.com/zrepl/zrepl/daemon/hooks"
+	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/util/envconst"
 	"github.com/zrepl/zrepl/zfs"
@@ -45,10 +46,9 @@ type snapProgress struct {

 type args struct {
 	ctx            context.Context
-	log            Logger
 	prefix         string
 	interval       time.Duration
-	fsf            *filters.DatasetMapFilter
+	fsf            zfs.DatasetFilter
 	snapshotsTaken chan<- struct{}
 	hooks          *hooks.List
 	dryRun         bool
@@ -102,26 +102,13 @@ func (s State) sf() state {
 type updater func(u func(*Snapper)) State
 type state func(a args, u updater) state

-type contextKey int
-
-const (
-	contextKeyLog contextKey = 0
-)
-
 type Logger = logger.Logger

-func WithLogger(ctx context.Context, log Logger) context.Context {
-	return context.WithValue(ctx, contextKeyLog, log)
-}
-
 func getLogger(ctx context.Context) Logger {
-	if log, ok := ctx.Value(contextKeyLog).(Logger); ok {
-		return log
-	}
-	return logger.NewNullLogger()
+	return logging.GetLogger(ctx, logging.SubsysSnapshot)
 }

-func PeriodicFromConfig(g *config.Global, fsf *filters.DatasetMapFilter, in *config.SnapshottingPeriodic) (*Snapper, error) {
+func PeriodicFromConfig(g *config.Global, fsf zfs.DatasetFilter, in *config.SnapshottingPeriodic) (*Snapper, error) {
 	if in.Prefix == "" {
 		return nil, errors.New("prefix must not be empty")
 	}
@@ -146,13 +133,12 @@ func PeriodicFromConfig(g *config.Global, fsf *filters.DatasetMapFilter, in *con
 }

 func (s *Snapper) Run(ctx context.Context, snapshotsTaken chan<- struct{}) {
-
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 	getLogger(ctx).Debug("start")
 	defer getLogger(ctx).Debug("stop")

 	s.args.snapshotsTaken = snapshotsTaken
 	s.args.ctx = ctx
-	s.args.log = getLogger(ctx)
 	s.args.dryRun = false // for future expansion

 	u := func(u func(*Snapper)) State {
@@ -190,7 +176,7 @@ func onErr(err error, u updater) state {
 		case Snapshotting:
 			s.state = ErrorWait
 		}
-		s.args.log.WithError(err).WithField("pre_state", preState).WithField("post_state", s.state).Error("snapshotting error")
+		getLogger(s.args.ctx).WithError(err).WithField("pre_state", preState).WithField("post_state", s.state).Error("snapshotting error")
 	}).sf()
 }

@@ -209,7 +195,7 @@ func syncUp(a args, u updater) state {
 	if err != nil {
 		return onErr(err, u)
 	}
-	syncPoint, err := findSyncPoint(a.log, fss, a.prefix, a.interval)
+	syncPoint, err := findSyncPoint(a.ctx, fss, a.prefix, a.interval)
 	if err != nil {
 		return onErr(err, u)
 	}
@@ -266,18 +252,18 @@ func snapshot(a args, u updater) state {
 		suffix := time.Now().In(time.UTC).Format("20060102_150405_000")
 		snapname := fmt.Sprintf("%s%s", a.prefix, suffix)

-		l := a.log.
-			WithField("fs", fs.ToString()).
-			WithField("snap", snapname)
+		ctx := logging.WithInjectedField(a.ctx, "fs", fs.ToString())
+		ctx = logging.WithInjectedField(ctx, "snap", snapname)

 		hookEnvExtra := hooks.Env{
 			hooks.EnvFS:       fs.ToString(),
 			hooks.EnvSnapshot: snapname,
 		}

-		jobCallback := hooks.NewCallbackHookForFilesystem("snapshot", fs, func(_ context.Context) (err error) {
+		jobCallback := hooks.NewCallbackHookForFilesystem("snapshot", fs, func(ctx context.Context) (err error) {
+			l := getLogger(ctx)
 			l.Debug("create snapshot")
-			err = zfs.ZFSSnapshot(fs, snapname, false) // TODO propagagte context to ZFSSnapshot
+			err = zfs.ZFSSnapshot(ctx, fs, snapname, false) // TODO propagate context to ZFSSnapshot
 			if err != nil {
 				l.WithError(err).Error("cannot create snapshot")
 			}
@@ -290,7 +276,7 @@ func snapshot(a args, u updater) state {
 		{
 			filteredHooks, err := a.hooks.CopyFilteredForFilesystem(fs)
 			if err != nil {
-				l.WithError(err).Error("unexpected filter error")
+				getLogger(ctx).WithError(err).Error("unexpected filter error")
 				fsHadErr = true
 				goto updateFSState
 			}
@@ -303,7 +289,7 @@ func snapshot(a args, u updater) state {
 			plan, planErr = hooks.NewPlan(&filteredHooks, hooks.PhaseSnapshot, jobCallback, hookEnvExtra)
 			if planErr != nil {
 				fsHadErr = true
-				l.WithError(planErr).Error("cannot create job hook plan")
+				getLogger(ctx).WithError(planErr).Error("cannot create job hook plan")
 				goto updateFSState
 			}
 		}
@@ -314,15 +300,14 @@ func snapshot(a args, u updater) state {
 			progress.state = SnapStarted
 		})
 		{
-			l := hooks.GetLogger(a.ctx).WithField("fs", fs.ToString()).WithField("snap", snapname)
-			l.WithField("report", plan.Report().String()).Debug("begin run job plan")
-			plan.Run(hooks.WithLogger(a.ctx, l), a.dryRun)
+			getLogger(ctx).WithField("report", plan.Report().String()).Debug("begin run job plan")
+			plan.Run(ctx, a.dryRun)
 			planReport = plan.Report()
 			fsHadErr = planReport.HadError() // not just fatal errors
 			if fsHadErr {
-				l.WithField("report", planReport.String()).Error("end run job plan with error")
+				getLogger(ctx).WithField("report", planReport.String()).Error("end run job plan with error")
 			} else {
-				l.WithField("report", planReport.String()).Info("end run job plan successful")
+				getLogger(ctx).WithField("report", planReport.String()).Info("end run job plan successful")
 			}
 		}

@@ -342,7 +327,7 @@ func snapshot(a args, u updater) state {
 	case a.snapshotsTaken <- struct{}{}:
 	default:
 		if a.snapshotsTaken != nil {
-			a.log.Warn("callback channel is full, discarding snapshot update event")
+			getLogger(a.ctx).Warn("callback channel is full, discarding snapshot update event")
 		}
 	}

@@ -355,7 +340,7 @@ func snapshot(a args, u updater) state {
 					break
 				}
 			}
-			a.log.WithField("hook", h.String()).WithField("hook_number", hookIdx+1).Warn("hook did not match any snapshotted filesystems")
+			getLogger(a.ctx).WithField("hook", h.String()).WithField("hook_number", hookIdx+1).Warn("hook did not match any snapshotted filesystems")
 		}
 	}

@@ -376,7 +361,7 @@ func wait(a args, u updater) state {
 		lastTick := snapper.lastInvocation
 		snapper.sleepUntil = lastTick.Add(a.interval)
 		sleepUntil = snapper.sleepUntil
-		log := a.log.WithField("sleep_until", sleepUntil).WithField("duration", a.interval)
+		log := getLogger(a.ctx).WithField("sleep_until", sleepUntil).WithField("duration", a.interval)
 		logFunc := log.Debug
 		if snapper.state == ErrorWait || snapper.state == SyncUpErrWait {
 			logFunc = log.Error
@@ -397,14 +382,14 @@ func wait(a args, u updater) state {
 	}
 }

-func listFSes(ctx context.Context, mf *filters.DatasetMapFilter) (fss []*zfs.DatasetPath, err error) {
+func listFSes(ctx context.Context, mf zfs.DatasetFilter) (fss []*zfs.DatasetPath, err error) {
 	return zfs.ZFSListMapping(ctx, mf)
 }

 var syncUpWarnNoSnapshotUntilSyncupMinDuration = envconst.Duration("ZREPL_SNAPPER_SYNCUP_WARN_MIN_DURATION", 1*time.Second)

 // see docs/snapshotting.rst
-func findSyncPoint(log Logger, fss []*zfs.DatasetPath, prefix string, interval time.Duration) (syncPoint time.Time, err error) {
+func findSyncPoint(ctx context.Context, fss []*zfs.DatasetPath, prefix string, interval time.Duration) (syncPoint time.Time, err error) {

 	const (
 		prioHasVersions int = iota
@@ -426,10 +411,10 @@ func findSyncPoint(log Logger, fss []*zfs.DatasetPath, prefix string, interval t

 	now := time.Now()

-	log.Debug("examine filesystem state to find sync point")
+	getLogger(ctx).Debug("examine filesystem state to find sync point")
 	for _, d := range fss {
-		l := log.WithField("fs", d.ToString())
-		syncPoint, err := findSyncPointFSNextOptimalSnapshotTime(l, now, interval, prefix, d)
+		ctx := logging.WithInjectedField(ctx, "fs", d.ToString())
+		syncPoint, err := findSyncPointFSNextOptimalSnapshotTime(ctx, now, interval, prefix, d)
 		if err == findSyncPointFSNoFilesystemVersionsErr {
 			snaptimes = append(snaptimes, snapTime{
 				ds:   d,
@@ -438,9 +423,9 @@ func findSyncPoint(log Logger, fss []*zfs.DatasetPath, prefix string, interval t
 			})
 		} else if err != nil {
 			hardErrs++
-			l.WithError(err).Error("cannot determine optimal sync point for this filesystem")
+			getLogger(ctx).WithError(err).Error("cannot determine optimal sync point for this filesystem")
 		} else {
-			l.WithField("syncPoint", syncPoint).Debug("found optimal sync point for this filesystem")
+			getLogger(ctx).WithField("syncPoint", syncPoint).Debug("found optimal sync point for this filesystem")
 			snaptimes = append(snaptimes, snapTime{
 				ds:   d,
 				prio: prioHasVersions,
@@ -467,7 +452,7 @@ func findSyncPoint(log Logger, fss []*zfs.DatasetPath, prefix string, interval t
 	})

 	winnerSyncPoint := snaptimes[0].time
-	l := log.WithField("syncPoint", winnerSyncPoint.String())
+	l := getLogger(ctx).WithField("syncPoint", winnerSyncPoint.String())
 	l.Info("determined sync point")
 	if winnerSyncPoint.Sub(now) > syncUpWarnNoSnapshotUntilSyncupMinDuration {
 		for _, st := range snaptimes {
@@ -483,9 +468,12 @@ func findSyncPoint(log Logger, fss []*zfs.DatasetPath, prefix string, interval t

 var findSyncPointFSNoFilesystemVersionsErr = fmt.Errorf("no filesystem versions")

-func findSyncPointFSNextOptimalSnapshotTime(l Logger, now time.Time, interval time.Duration, prefix string, d *zfs.DatasetPath) (time.Time, error) {
+func findSyncPointFSNextOptimalSnapshotTime(ctx context.Context, now time.Time, interval time.Duration, prefix string, d *zfs.DatasetPath) (time.Time, error) {

-	fsvs, err := zfs.ZFSListFilesystemVersions(d, filters.NewTypedPrefixFilter(prefix, zfs.Snapshot))
+	fsvs, err := zfs.ZFSListFilesystemVersions(ctx, d, zfs.ListFilesystemVersionsOptions{
+		Types:           zfs.Snapshots,
+		ShortnamePrefix: prefix,
+	})
 	if err != nil {
 		return time.Time{}, errors.Wrap(err, "list filesystem versions")
 	}
@@ -499,7 +487,7 @@ func findSyncPointFSNextOptimalSnapshotTime(l Logger, now time.Time, interval ti
 	})

 	latest := fsvs[len(fsvs)-1]
-	l.WithField("creation", latest.Creation).Debug("found latest snapshot")
+	getLogger(ctx).WithField("creation", latest.Creation).Debug("found latest snapshot")

 	since := now.Sub(latest.Creation)
 	if since < 0 {
@@ -5,7 +5,7 @@ import (
 	"fmt"

 	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon/filters"
+	"github.com/zrepl/zrepl/zfs"
 )

 // FIXME: properly abstract snapshotting:
@@ -32,7 +32,7 @@ func (s *PeriodicOrManual) Report() *Report {
 	return nil
 }

-func FromConfig(g *config.Global, fsf *filters.DatasetMapFilter, in config.SnapshottingEnum) (*PeriodicOrManual, error) {
+func FromConfig(g *config.Global, fsf zfs.DatasetFilter, in config.SnapshottingEnum) (*PeriodicOrManual, error) {
 	switch v := in.Ret.(type) {
 	case *config.SnapshottingPeriodic:
 		snapper, err := PeriodicFromConfig(g, fsf, v)
@@ -1,595 +0,0 @@
-{
-  "annotations": {
-    "list": [
-      {
-        "$$hashKey": "object:3351",
-        "builtIn": 1,
-        "datasource": "-- Grafana --",
-        "enable": true,
-        "hide": true,
-        "iconColor": "rgba(0, 211, 255, 1)",
-        "name": "Annotations & Alerts",
-        "type": "dashboard"
-      }
-    ]
-  },
-  "editable": true,
-  "gnetId": null,
-  "graphTooltip": 1,
-  "id": 7,
-  "iteration": 1552746418826,
-  "links": [],
-  "panels": [
-    {
-      "content": "# zrepl Prometheus Metrics\n\nzrepl exposes Prometheus metrics and ships with this Grafana dashboard.\nThe exported metrics are suitable for health checks:\n\n* The log should generally be warning & error-free\n  * The `Log Messages that require attention` graph visualizes log message counts indicating problems.\n* The number of goroutines should not grow unboundedly over time.\n  * During replication, the number of goroutines can be way higher than during idle time.\n  * If the goroutine count grows with each replication, there is clearly a goroutine leak. Please open a bug report.\n* The sys memory consumption should not grow unboundedly over time.\n  * Note that the Go runtime pre-allocates some of its heap from the OS.\n  * zrepl actually uses much less memory than allocated from the OS.\n  * Since Go 1.11, Go pre-allocates more aggressively.\n* Monitor that some data is replicated, although that metric does not guarantee that replication was successful.\n\n**In general, note that the exported metrics are not stable unless declared otherwise.**",
-      "gridPos": {
-        "h": 9,
-        "w": 24,
-        "x": 0,
-        "y": 0
-      },
-      "id": 35,
-      "mode": "markdown",
-      "title": "Panel Title",
-      "type": "text"
-    },
-    {
-      "aliasColors": {},
-      "bars": false,
-      "dashLength": 10,
-      "dashes": false,
-      "datasource": null,
-      "fill": 1,
-      "gridPos": {
-        "h": 9,
-        "w": 12,
-        "x": 0,
-        "y": 9
-      },
-      "id": 15,
-      "legend": {
-        "avg": false,
-        "current": false,
-        "max": false,
-        "min": false,
-        "show": true,
-        "total": false,
-        "values": false
-      },
-      "lines": true,
-      "linewidth": 1,
-      "links": [],
-      "nullPointMode": "null",
-      "percentage": false,
-      "pointradius": 5,
-      "points": false,
-      "renderer": "flot",
-      "seriesOverrides": [],
-      "spaceLength": 10,
-      "stack": true,
-      "steppedLine": false,
-      "targets": [
-        {
-          "$$hashKey": "object:3436",
-          "expr": "up{job='$prom_job_name'}",
-          "format": "time_series",
-          "intervalFactor": 1,
-          "refId": "A"
-        }
-      ],
-      "thresholds": [],
-      "timeFrom": null,
-      "timeShift": null,
-      "title": "zrepl Instances Up",
-      "tooltip": {
-        "shared": true,
-        "sort": 0,
-        "value_type": "individual"
-      },
-      "type": "graph",
-      "xaxis": {
-        "buckets": null,
-        "mode": "time",
-        "name": null,
-        "show": true,
-        "values": []
-      },
-      "yaxes": [
-        {
-          "format": "short",
-          "label": null,
-          "logBase": 1,
-          "max": "5",
-          "min": "0",
-          "show": true
-        },
-        {
-          "format": "short",
-          "label": null,
-          "logBase": 1,
-          "max": null,
-          "min": null,
-          "show": true
-        }
-      ]
-    },
-    {
-      "aliasColors": {},
-      "bars": false,
-      "dashLength": 10,
-      "dashes": false,
-      "datasource": null,
-      "fill": 1,
-      "gridPos": {
-        "h": 9,
-        "w": 12,
-        "x": 12,
-        "y": 9
-      },
-      "id": 22,
-      "legend": {
-        "avg": false,
-        "current": false,
-        "max": false,
-        "min": false,
-        "show": true,
-        "total": false,
-        "values": false
-      },
-      "lines": true,
-      "linewidth": 1,
-      "links": [],
-      "nullPointMode": "null",
-      "percentage": false,
-      "pointradius": 5,
-      "points": false,
-      "renderer": "flot",
-      "seriesOverrides": [],
-      "spaceLength": 10,
-      "stack": false,
-      "steppedLine": false,
-      "targets": [
-        {
-          "expr": "increase(zrepl_daemon_log_entries{job='$prom_job_name',level=~'warn|error'}[$__interval])",
-          "format": "time_series",
-          "intervalFactor": 1,
-          "refId": "A"
-        }
-      ],
-      "thresholds": [],
-      "timeFrom": null,
-      "timeShift": null,
-      "title": "Log Messages that require attention",
-      "tooltip": {
-        "shared": true,
-        "sort": 0,
-        "value_type": "individual"
-      },
-      "type": "graph",
-      "xaxis": {
-        "buckets": null,
-        "mode": "time",
-        "name": null,
-        "show": true,
-        "values": []
-      },
-      "yaxes": [
-        {
-          "format": "short",
-          "label": null,
-          "logBase": 1,
-          "max": null,
-          "min": "0",
-          "show": true
-        },
-        {
-          "format": "short",
-          "label": null,
-          "logBase": 1,
-          "max": null,
-          "min": null,
-          "show": true
-        }
-      ]
-    },
-    {
-      "aliasColors": {},
-      "bars": false,
-      "dashLength": 10,
-      "dashes": false,
-      "datasource": null,
-      "fill": 1,
-      "gridPos": {
-        "h": 9,
-        "w": 12,
-        "x": 0,
-        "y": 18
-      },
-      "id": 33,
-      "legend": {
-        "avg": false,
-        "current": false,
-        "max": false,
-        "min": false,
-        "show": true,
-        "total": false,
-        "values": false
-      },
-      "lines": true,
-      "linewidth": 1,
-      "links": [],
-      "nullPointMode": "null",
-      "percentage": false,
-      "pointradius": 5,
-      "points": false,
-      "renderer": "flot",
-      "seriesOverrides": [
-        {
-          "alias": "/replicated bytes in last.*/",
-          "yaxis": 2
-        }
-      ],
-      "spaceLength": 10,
-      "stack": false,
-      "steppedLine": false,
-      "targets": [
-        {
-          "expr": "sum(rate(zrepl_replication_bytes_replicated{job='$prom_job_name'}[$__interval])) by (zrepl_job)",
-          "format": "time_series",
-          "hide": false,
-          "interval": "",
-          "intervalFactor": 1,
-          "legendFormat": "replication data rate zrepl_job={{zrepl_job}}",
-          "refId": "A"
-        },
-        {
-          "expr": "sum(increase(zrepl_replication_bytes_replicated{job='$prom_job_name'}[10m])) by (zrepl_job)",
-          "format": "time_series",
-          "hide": false,
-          "interval": "",
-          "intervalFactor": 1,
-          "legendFormat": "replicated bytes in last 10min zrepl_job={{zrepl_job}}",
-          "refId": "B"
-        }
-      ],
-      "thresholds": [],
-      "timeFrom": null,
-      "timeShift": null,
-      "title": "Replication Data Rate and Volume(integrates last 10min)",
-      "tooltip": {
-        "shared": true,
-        "sort": 0,
-        "value_type": "individual"
-      },
-      "type": "graph",
-      "xaxis": {
-        "buckets": null,
-        "mode": "time",
-        "name": null,
-        "show": true,
-        "values": []
-      },
-      "yaxes": [
-        {
-          "format": "Bps",
-          "label": null,
-          "logBase": 1,
-          "max": null,
-          "min": null,
-          "show": true
-        },
-        {
-          "format": "bytes",
-          "label": null,
-          "logBase": 1,
-          "max": null,
-          "min": null,
-          "show": true
-        }
-      ]
-    },
-    {
-      "aliasColors": {},
-      "bars": false,
-      "dashLength": 10,
-      "dashes": false,
-      "datasource": null,
-      "fill": 1,
-      "gridPos": {
-        "h": 9,
-        "w": 12,
-        "x": 12,
-        "y": 18
-      },
-      "id": 23,
-      "legend": {
-        "avg": false,
-        "current": false,
-        "max": false,
-        "min": false,
-        "show": true,
-        "total": false,
-        "values": false
-      },
-      "lines": true,
-      "linewidth": 1,
-      "links": [],
-      "nullPointMode": "null",
-      "percentage": false,
-      "pointradius": 5,
-      "points": false,
-      "renderer": "flot",
-      "seriesOverrides": [],
-      "spaceLength": 10,
-      "stack": false,
-      "steppedLine": false,
-      "targets": [
-        {
-          "expr": "sum(increase(zrepl_daemon_log_entries{job='$prom_job_name',zrepl_job=~\"^[^_].*\"}[$__interval])) by (instance,zrepl_job)",
-          "format": "time_series",
-          "intervalFactor": 1,
-          "refId": "A"
-        }
-      ],
-      "thresholds": [],
-      "timeFrom": null,
-      "timeShift": null,
-      "title": "Log Activity (without internal jobs)",
-      "tooltip": {
-        "shared": true,
-        "sort": 0,
-        "value_type": "individual"
-      },
-      "type": "graph",
-      "xaxis": {
-        "buckets": null,
-        "mode": "time",
-        "name": null,
-        "show": true,
-        "values": []
-      },
-      "yaxes": [
-        {
-          "format": "short",
-          "label": null,
-          "logBase": 1,
-          "max": null,
-          "min": "0",
-          "show": true
-        },
-        {
-          "format": "short",
-          "label": null,
-          "logBase": 1,
-          "max": null,
-          "min": null,
-          "show": true
-        }
-      ]
-    },
-    {
-      "aliasColors": {},
-      "bars": false,
-      "dashLength": 10,
-      "dashes": false,
-      "datasource": null,
-      "fill": 1,
-      "gridPos": {
-        "h": 9,
-        "w": 12,
-        "x": 0,
-        "y": 27
-      },
-      "id": 17,
-      "legend": {
-        "avg": false,
-        "current": false,
-        "max": false,
-        "min": false,
-        "show": true,
-        "total": false,
-        "values": false
-      },
-      "lines": true,
-      "linewidth": 1,
-      "links": [],
-      "nullPointMode": "null",
-      "percentage": false,
-      "pointradius": 5,
-      "points": false,
-      "renderer": "flot",
-      "seriesOverrides": [],
-      "spaceLength": 10,
-      "stack": false,
-      "steppedLine": false,
-      "targets": [
-        {
-          "$$hashKey": "object:3535",
-          "expr": "go_memstats_sys_bytes{job='$prom_job_name'}",
-          "format": "time_series",
-          "hide": false,
-          "intervalFactor": 1,
-          "refId": "A"
-        }
-      ],
-      "thresholds": [],
-      "timeFrom": null,
-      "timeShift": null,
-      "title": "Memory Allocated by the Go runtime from the OS (should not grow unboundedly)",
-      "tooltip": {
-        "shared": true,
-        "sort": 0,
-        "value_type": "individual"
-      },
-      "type": "graph",
-      "xaxis": {
-        "buckets": null,
-        "mode": "time",
-        "name": null,
-        "show": true,
-        "values": []
-      },
-      "yaxes": [
-        {
-          "format": "bytes",
-          "label": null,
-          "logBase": 1,
-          "max": null,
-          "min": "0",
-          "show": true
-        },
-        {
-          "format": "short",
-          "label": null,
-          "logBase": 1,
-          "max": null,
-          "min": null,
-          "show": true
-        }
-      ]
-    },
-    {
-      "aliasColors": {},
-      "bars": false,
-      "dashLength": 10,
-      "dashes": false,
-      "datasource": null,
-      "fill": 1,
-      "gridPos": {
-        "h": 9,
-        "w": 12,
-        "x": 12,
-        "y": 27
-      },
-      "id": 19,
-      "legend": {
-        "avg": false,
-        "current": false,
-        "max": false,
-        "min": false,
-        "show": true,
-        "total": false,
-        "values": false
-      },
-      "lines": true,
-      "linewidth": 1,
-      "links": [],
-      "nullPointMode": "null",
-      "percentage": false,
-      "pointradius": 5,
-      "points": false,
-      "renderer": "flot",
-      "seriesOverrides": [],
-      "spaceLength": 10,
-      "stack": false,
-      "steppedLine": false,
-      "targets": [
-        {
-          "expr": "go_goroutines{job='$prom_job_name'}",
-          "format": "time_series",
-          "intervalFactor": 1,
-          "refId": "A"
-        }
-      ],
-      "thresholds": [],
-      "timeFrom": null,
-      "timeShift": null,
-      "title": "number of goroutines (should not grow unboundedly)",
-      "tooltip": {
-        "shared": true,
-        "sort": 0,
-        "value_type": "individual"
-      },
-      "type": "graph",
-      "xaxis": {
-        "buckets": null,
-        "mode": "time",
-        "name": null,
-        "show": true,
-        "values": []
-      },
-      "yaxes": [
-        {
-          "format": "short",
-          "label": null,
-          "logBase": 1,
-          "max": null,
-          "min": "0",
-          "show": true
-        },
-        {
-          "format": "short",
-          "label": null,
-          "logBase": 1,
-          "max": null,
-          "min": null,
-          "show": true
-        }
-      ]
-    }
-  ],
-  "refresh": "1m",
-  "schemaVersion": 16,
-  "style": "dark",
-  "tags": [],
-  "templating": {
-    "list": [
-      {
-        "allValue": null,
-        "current": {
-          "text": "zrepl",
-          "value": "zrepl"
-        },
-        "datasource": "prometheus",
-        "hide": 0,
-        "includeAll": false,
-        "label": "Prometheus Job Name",
-        "multi": false,
-        "name": "prom_job_name",
-        "options": [],
-        "query": "label_values(up, job)",
-        "refresh": 1,
-        "regex": "",
-        "sort": 1,
-        "tagValuesQuery": "",
-        "tags": [],
-        "tagsQuery": "",
-        "type": "query",
-        "useTags": false
-      }
-    ]
-  },
-  "time": {
-    "from": "now-2d",
-    "to": "now"
-  },
-  "timepicker": {
-    "refresh_intervals": [
-      "5s",
-      "10s",
-      "30s",
-      "1m",
-      "5m",
-      "15m",
-      "30m",
-      "1h",
-      "2h",
-      "1d"
-    ],
-    "time_options": [
-      "5m",
-      "15m",
-      "1h",
-      "6h",
-      "12h",
-      "24h",
-      "2d",
-      "7d",
-      "30d"
-    ]
-  },
-  "timezone": "",
-  "title": "zrepl 0.1",
-  "uid": "xTljn4qmk",
-  "version": 6
-}
@@ -34,35 +34,58 @@ This is a big one! Headlining features:

 * **Resumable Send & Recv Support**
  No knobs required, automatically used where supported.
-* **Hold-Protected Send & Recv**
-  Automatic ZFS holds to ensure that we can always resume a replication step.
-* **Encrypted Send & Recv Support** for OpenZFS native encryption.
-  :ref:`Configurable <job-send-options>` at the job level, i.e., for all filesystems a job is responsible for.
-* **Receive-side hold on last received dataset**
-  The counterpart to the replication cursor bookmark on the send-side.
-  Ensures that incremental replication will always be possible between a sender and receiver.
+* **Encrypted Send & Recv Support** for OpenZFS native encryption,
+  :ref:`configurable <job-send-options>` at the job level, i.e., for all filesystems a job is responsible for.
+* **Replication Guarantees**
+  Automatic use of ZFS holds and bookmarks to protect a replicated filesystem from losing synchronization between sender and receiver.
+  By default, zrepl guarantees that incremental replication will always be possible and interrupted steps will always be resumable.

-Actual changelog:
+.. TIP::

-* |break_config| **more restrictive job names than in prior zrepl versions**
-  Starting with this version, job names are going to be embedded into ZFS holds and bookmark names.
+   We highly recommend studying the updated :ref:`overview section of the configuration chapter <overview-how-replication-works>` to understand how replication works.

+Quick-start guides:
+
+* We have added :ref:`another quick-start guide for a typical workstation use case for zrepl <quickstart-backup-to-external-disk>`.
+  Check it out to learn how you can use zrepl to back up your workstation's OpenZFS natively-encrypted root filesystem to an external disk.
+
+Additional changelog:
+
+* |break| |break_config| **more restrictive job names than in prior zrepl versions**
+  Starting with this version, job names are going to be embedded into ZFS holds and bookmark names (see :ref:`this section for details <zrepl-zfs-abstractions>`).
+  Therefore you might need to adjust your job names.
+  **Note that jobs** cannot be renamed easily **once you start using zrepl 0.3.**
 * |break| |mig| replication cursor representation changed

-  * zrepl now manages the :ref:`replication cursor bookmark <replication-cursor-and-last-received-hold>` per job-filesystem tuple instead of a single replication cursor per filesystem.
+  * zrepl now manages the :ref:`replication cursor bookmark <zrepl-zfs-abstractions>` per job-filesystem tuple instead of a single replication cursor per filesystem.
    In the future, this will permit multiple sending jobs to send from the same filesystems.
  * ZFS does not allow bookmark renaming, thus we cannot migrate the old replication cursors.
  * zrepl 0.3 will automatically create cursors in the new format for new replications, and warn if it still finds ones in the old format.
  * Run ``zrepl migrate replication-cursor:v1-v2`` to safely destroy old-format cursors.
    The migration will ensure that only those old-format cursors are destroyed that have been superseeded by new-format cursors.

+* |feature| New option ``listen_freebind`` (tcp, tls, prometheus listener)
+* |feature| :issue:`265` transport/tcp: support for CIDR masks in client IP whitelist
+* |feature| documented subcommand to generate ``bash`` and ``zsh`` completions
+* |feature| :issue:`307` ``chrome://trace`` -compatible activity tracing of zrepl daemon activity
+* |feature| logging: trace IDs for better log entry correlation with concurrent replication jobs
+* |feature| experimental environment variable for parallel replication (see :issue:`306` )
 * |bugfix| missing logger context vars in control connection handlers
 * |bugfix| improved error messages on ``zfs send`` errors
-* |feature| New option ``listen_freebind`` (tcp, tls, prometheus listener)
 * |bugfix| |docs| snapshotting: clarify sync-up behavior and warn about filesystems
+* |bugfix| transport/ssh: do not leak zombie ssh process on connection failures
  that will not be snapshotted until the sync-up phase is over
+* |docs| Installation: :ref:`FreeBSD jail with iocage <installation-freebsd-jail-with-iocage>`
 * |docs| Document new replication features in the :ref:`config overview <overview-how-replication-works>` and :repomasterlink:`replication/design.md`.
 * **[MAINTAINER NOTICE]** New platform tests in this version, please make sure you run them for your distro!
+* **[MAINTAINER NOTICE]** Please add the shell completions to the zrepl packages.
+
+.. NOTE::
+   |  zrepl is a spare-time project primarily developed by `Christian Schwarz <https://cschwarz.com>`_.
+   |  You can support maintenance and feature development through one of the following services:
+   |  |Donate via Patreon| |Donate via GitHub Sponsors| |Donate via Liberapay| |Donate via PayPal|
+   |  Note that PayPal processing fees are relatively high for small donations.
+   |  For SEPA wire transfer and **commercial support**, please `contact Christian directly <https://cschwarz.com>`_.

 0.2.1
 -----
@@ -100,14 +123,6 @@ Actual changelog:
  The pool name ``zreplplatformtest`` is reserved for this use case.
  Only run ``make platformtest`` on test systems, e.g. a FreeBSD VM image.

-.. NOTE::
-   |  zrepl is a spare-time project primarily developed by `Christian Schwarz <https://cschwarz.com>`_.
-   |  You can support maintenance and feature development through one of the following services:
-   |  |Donate via Patreon| |Donate via Liberapay| |Donate via PayPal|
-   |  Note that PayPal processing fees are relatively high for small donations.
-   |  For SEPA wire transfer and **commerical support**, please `contact Christian directly <https://cschwarz.com>`_.
-
-
 0.1.1
 -----

@@ -12,6 +12,7 @@ Configuration
    configuration/transports
    configuration/filter_syntax
    configuration/sendrecvoptions
+    configuration/replication
    configuration/snapshotting
    configuration/prune
    configuration/logging
@@ -19,7 +19,7 @@ Job Type ``push``
    * - ``type``
      - = ``push``
    * - ``name``
-      - unique name of the job
+      - unique name of the job :issue:`(must not change)<327>`
    * - ``connect``
      - |connect-transport|
    * - ``filesystems``
@@ -47,7 +47,7 @@ Job Type ``sink``
    * - ``type``
      - = ``sink``
    * - ``name``
-      - unique name of the job
+      - unique name of the job :issue:`(must not change)<327>`
    * - ``serve``
      - |serve-transport|
    * - ``root_fs``
@@ -70,7 +70,7 @@ Job Type ``pull``
    * - ``type``
      - = ``pull``
    * - ``name``
-      - unique name of the job
+      - unique name of the job :issue:`(must not change)<327>`
    * - ``connect``
      - |connect-transport|
    * - ``root_fs``
@@ -98,7 +98,7 @@ Job Type ``source``
    * - ``type``
      - = ``source``
    * - ``name``
-      - unique name of the job
+      - unique name of the job :issue:`(must not change)<327>`
    * - ``serve``
      - |serve-transport|
    * - ``filesystems``
@@ -137,7 +137,7 @@ Job type that only takes snapshots and performs pruning on the local machine.
    * - ``type``
      - = ``snap``
    * - ``name``
-      - unique name of the job
+      - unique name of the job :issue:`(must not change)<327>`
    * - ``filesystems``
      - |filter-spec| for filesystems to be snapshotted
    * - ``snapshotting``
@@ -202,7 +202,7 @@ The latter is particularly useful in combination with log aggregation services.
 .. WARNING::

    zrepl drops log messages to the TCP outlet if the underlying connection is not fast enough.
-    Note that TCP buffering in the kernel must first run full becfore messages are dropped.
+    Note that TCP buffering in the kernel must first run full before messages are dropped.

    Make sure to always configure a ``stdout`` outlet as the special error outlet to be informed about problems
    with the TCP outlet (see :ref:`above <logging-error-outlet>` ).
@@ -15,7 +15,7 @@ Prometheus & Grafana
 zrepl can expose `Prometheus metrics <https://prometheus.io/docs/instrumenting/exposition_formats/>`_ via HTTP.
 The ``listen`` attribute is a `net.Listen <https://golang.org/pkg/net/#Listen>`_  string for tcp, e.g. ``:9091`` or ``127.0.0.1:9091``.
 The ``listen_freebind`` attribute is :ref:`explained here <listen-freebind-explanation>`.
-The Prometheues monitoring job appears in the ``zrepl control`` job list and may be specified **at most once**.
+The Prometheus monitoring job appears in the ``zrepl control`` job list and may be specified **at most once**.

 zrepl also ships with an importable `Grafana <https://grafana.com>`_ dashboard that consumes the Prometheus metrics:
 see :repomasterlink:`dist/grafana`.
@@ -12,7 +12,7 @@ The following paths are considered:
 The ``zrepl configcheck`` subcommand can be used to validate the configuration.
 The command will output nothing and exit with zero status code if the configuration is valid.
 The error messages vary in quality and usefulness: please report confusing config errors to the tracking :issue:`155`.
-Full example configs such as in the :ref:`tutorial` or the :sampleconf:`/` directory might also be helpful.
+Full example configs such as in the :ref:`quick-start guides <quickstart-toc>` or the :sampleconf:`/` directory might also be helpful.
 However, copy-pasting examples is no substitute for reading documentation!

 Config File Structure
@@ -26,9 +26,9 @@ Config File Structure
     type: push
   - ...

-zrepl is confgured using a single YAML configuration file with two main sections: ``global`` and ``jobs``.
+zrepl is configured using a single YAML configuration file with two main sections: ``global`` and ``jobs``.
 The ``global`` section is filled with sensible defaults and is covered later in this chapter.
-The ``jobs`` section is a list of jobs which we are goind to explain now.
+The ``jobs`` section is a list of jobs which we are going to explain now.

 .. _job-overview:

@@ -39,9 +39,13 @@ A *job* is the unit of activity tracked by the zrepl daemon.
 The ``type`` of a job determines its role in a replication setup and in snapshot management.
 Jobs are identified by their ``name``, both in log files and the ``zrepl status`` command.

+.. NOTE::
+   The job name is persisted in several places on disk and thus :issue:`cannot be changed easily<327>`.
+
+
 Replication always happens between a pair of jobs: one is the **active side**, and one the **passive side**.
 The active side connects to the passive side using a :ref:`transport <transport>` and starts executing the replication logic.
-The passive side responds to requests from the active side after checking its persmissions.
+The passive side responds to requests from the active side after checking its permissions.

 The following table shows how different job types can be combined to achieve **both push and pull mode setups**.
 Note that snapshot-creation denoted by "(snap)" is orthogonal to whether a job is active or passive.
@@ -99,7 +103,7 @@ One of the major design goals of the replication module is to avoid any duplicat
 As such, the code works on abstract senders and receiver **endpoints**, where typically one will be implemented by a local program object and the other is an RPC client instance.
 Regardless of push- or pull-style setup, the logic executes on the active side, i.e. in the ``push`` or ``pull`` job.

-The following steps take place during replication and can be monitored using the ``zrepl status`` subcommand:
+The following high-level steps take place during replication and can be monitored using the ``zrepl status`` subcommand:

 * Plan the replication:

@@ -107,9 +111,9 @@ The following steps take place during replication and can be monitored using the
  * Build the **replication plan**

    * Per filesystem, compute a diff between sender and receiver snapshots
-    * Build a list of replication steps
+    * Build a list of **replication steps**

-      * If possible, use incremental and resumable sends (``zfs send -i``)
+      * If possible, use incremental and resumable sends
      * Otherwise, use full send of most recent snapshot on sender

  * Retry on errors that are likely temporary (i.e. network failures).
@@ -120,7 +124,7 @@ The following steps take place during replication and can be monitored using the
  * Perform replication steps in the following order:
    Among all filesystems with pending replication steps, pick the filesystem whose next replication step's snapshot is the oldest.
  * Create placeholder filesystems on the receiving side to mirror the dataset paths on the sender to ``root_fs/${client_identity}``.
-  * Aquire send-side step-holds on the step's `from` and `to` snapshots.
+  * Acquire send-side *step-holds* on the step's `from` and `to` snapshots.
  * Perform the replication step.
  * Move the **replication cursor** bookmark on the sending side (see below).
  * Move the **last-received-hold** on the receiving side (see below).
@@ -128,36 +132,97 @@ The following steps take place during replication and can be monitored using the
   
 The idea behind the execution order of replication steps is that if the sender snapshots all filesystems simultaneously at fixed intervals, the receiver will have all filesystems snapshotted at time ``T1`` before the first snapshot at ``T2 = T1 + $interval`` is replicated.

-.. _replication-cursor-and-last-received-hold:
+ZFS Background Knowledge
+^^^^^^^^^^^^^^^^^^^^^^^^
+This section gives some background knowledge about ZFS features that zrepl uses to provide guarantees for a replication filesystem.
+Specifically, zrepl guarantees by default that **incremental replication is always possible and that started replication steps can always be resumed if they are interrupted.**

-**Replication cursor** bookmark and **last-received-hold** are managed by zrepl to ensure that future replications can always be done incrementally:
-the replication cursor is a send-side bookmark of the most recent successfully replicated snapshot,
-and the last-received-hold is a hold of that snapshot on the receiving side.
-The replication cursor has the format ``#zrepl_CUSOR_G_<GUID>_J_<JOBNAME>``.
-The last-received-hold tag has the format ``#zrepl_last_received_J_<JOBNAME>``.
-Encoding the job name in the names ensures that multiple sending jobs can replicate the same filesystem to different receivers without interference.
-The ``zrepl holds list`` provides a listing of all bookmarks and holds managed by zrepl.
+**ZFS Send Modes & Bookmarks**
+ZFS supports full sends (``zfs send fs@to``) and incremental sends (``zfs send -i @from fs@to``).
+Full sends are used to create a new filesystem on the receiver with the send-side state of ``fs@to``.
+Incremental sends only transfer the delta between ``@from`` and ``@to``.
+Incremental sends require that ``@from`` be present on the receiving side when receiving the incremental stream.
+Incremental sends can also use a ZFS bookmark as *from* on the sending side (``zfs send -i #bm_from fs@to``), where ``#bm_from`` was created using ``zfs bookmark fs@from fs#bm_from``.
+The receiving side must always have the actual snapshot ``@from``, regardless of whether the sending side uses ``@from`` or a bookmark of it.
+
+**Resumable Send & Recv**
+The ``-s`` flag for ``zfs recv`` tells zfs to save the partially received send stream in case it is interrupted.
+To resume the replication, the receiving side filesystem's ``receive_resume_token`` must be passed to a new ``zfs send -t <value> | zfs recv`` command.
+A full send can only be resumed if ``@to`` still exists.
+An incremental send can only be resumed if ``@to`` still exists *and* either ``@from`` still exists *or* a bookmark ``#fbm`` of ``@from`` still exists.
+
+**ZFS Holds**
+ZFS holds prevent a snapshot from being deleted through ``zfs destroy``, letting the destroy fail with a ``datset is busy`` error.
+Holds are created and referred to by a *tag*. They can be thought of as a named, persistent lock on the snapshot.
+
+
+.. _zrepl-zfs-abstractions:
+
+ZFS Abstractions Managed By zrepl
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+With the background knowledge from the previous paragraph, we now summarize the different on-disk ZFS objects that zrepl manages to provide its functionality.

 .. _replication-placeholder-property:

 **Placeholder filesystems** on the receiving side are regular ZFS filesystems with the placeholder property ``zrepl:placeholder=on``.
-Placeholders allow the receiving side to mirror the sender's ZFS dataset hierachy without replicating every filesystem at every intermediary dataset path component.
+Placeholders allow the receiving side to mirror the sender's ZFS dataset hierarchy without replicating every filesystem at every intermediary dataset path component.
 Consider the following example: ``S/H/J`` shall be replicated to ``R/sink/job/S/H/J``, but neither ``S/H`` nor ``S`` shall be replicated.
 ZFS requires the existence of ``R/sink/job/S`` and ``R/sink/job/S/H`` in order to receive into ``R/sink/job/S/H/J``.
 Thus, zrepl creates the parent filesystems as placeholders on the receiving side.
 If at some point ``S/H`` and ``S`` shall be replicated, the receiving side invalidates the placeholder flag automatically.
 The ``zrepl test placeholder`` command can be used to check whether a filesystem is a placeholder.

+.. _replication-cursor-and-last-received-hold:
+
+The **replication cursor** bookmark and **last-received-hold** are managed by zrepl to ensure that future replications can always be done incrementally.
+The replication cursor is a send-side bookmark of the most recent successfully replicated snapshot,
+and the last-received-hold is a hold of that snapshot on the receiving side.
+Both are moved atomically after the receiving side has confirmed that a replication step is complete.
+
+The replication cursor has the format ``#zrepl_CUSOR_G_<GUID>_J_<JOBNAME>``.
+The last-received-hold tag has the format ``zrepl_last_received_J_<JOBNAME>``.
+Encoding the job name in the names ensures that multiple sending jobs can replicate the same filesystem to different receivers without interference.
+
+.. _tentative-replication-cursor-bookmarks:
+
+**Tentative replication cursor bookmarks** are short-lived boomkarks that protect the atomic moving-forward of the replication cursor and last-received-hold (see :issue:`this issue <340>`).
+They are only necessary if step holds are not used as per the :ref:`replication.protection <replication-option-protection>` setting.
+The tentative replication cursor has the format ``#zrepl_CUSORTENTATIVE_G_<GUID>_J_<JOBNAME>``.
+The ``zrepl zfs-abstraction list`` command provides a listing of all bookmarks and holds managed by zrepl.
+
+.. _step-holds:
+
+**Step holds** are zfs holds managed by zrepl to ensure that a replication step can always be resumed if it is interrupted, e.g., due to network outage.
+zrepl creates step holds before it attempts a replication step and releases them after the receiver confirms that the replication step is complete.
+For an initial replication ``full @initial_snap``, zrepl puts a zfs hold on ``@initial_snap``.
+For an incremental send ``@from -> @to``, zrepl puts a zfs hold on both ``@from`` and ``@to``.
+Note that ``@from`` is not strictly necessary for resumability -- a bookmark on the sending side would be sufficient --, but size-estimation in currently used OpenZFS versions only works if ``@from`` is a snapshot.
+The hold tag has the format ``zrepl_STEP_J_<JOBNAME>``.
+A job only ever has one active send per filesystem.
+Thus, there are never more than two step holds for a given pair of ``(job,filesystem)``.
+
+**Step bookmarks** are zrepl's equivalent for holds on bookmarks (ZFS does not support putting holds on bookmarks).
+They are intended for a situation where a replication step uses a bookmark ``#bm`` as incremental ``from`` where ``#bm`` is not managed by zrepl.
+To ensure resumability, zrepl copies ``#bm`` to step bookmark ``#zrepl_STEP_G_<GUID>_J_<JOBNAME>``.
+If the replication is interrupted and ``#bm`` is deleted by the user, the step bookmark remains as an incremental source for the resumable send.
+Note that zrepl does not yet support creating step bookmarks because the `corresponding ZFS feature for copying bookmarks <https://github.com/openzfs/zfs/pull/9571>`_ is not yet widely available .
+Subscribe to zrepl :issue:`326` for details.
+
+The ``zrepl zfs-abstraction list`` command provides a listing of all bookmarks and holds managed by zrepl.
+
+.. NOTE::
+
+    More details can be found in the design document :repomasterlink:`replication/design.md`.
+
+Limitations
+^^^^^^^^^^^
+
 .. ATTENTION::

    Currently, zrepl does not replicate filesystem properties.
    When receiving a filesystem, it is never mounted (`-u` flag)  and `mountpoint=none` is set.
    This is temporary and being worked on :issue:`24`.

-.. NOTE::
-
-    More details can be found in the design document :repomasterlink:`replication/design.md`.
-

 .. _jobs-multiple-jobs:

@@ -177,26 +242,26 @@ Multiple Jobs & More than 2 Machines
 If you would like to see improvements to multi-job setups, please `open an issue on GitHub <https://github.com/zrepl/zrepl/issues/new>`_.

 No Overlapping
-~~~~~~~~~~~~~~
+^^^^^^^^^^^^^^

 Jobs run independently of each other.
 If two jobs match the same filesystem with their ``filesystems`` filter, they will operate on that filesystem independently and potentially in parallel.
-For example, if job A prunes snapshots that job B is planning to replicate, the replication will fail because B asssumed the snapshot to still be present.
+For example, if job A prunes snapshots that job B is planning to replicate, the replication will fail because B assumed the snapshot to still be present.
 However, the next replication attempt will re-examine the situation from scratch and should work.

 N push jobs to 1 sink
-~~~~~~~~~~~~~~~~~~~~~
+^^^^^^^^^^^^^^^^^^^^^

 The :ref:`sink job <job-sink>` namespaces by client identity.
 It is thus safe to push to one sink job with different client identities.
 If the push jobs have the same client identity, the filesystems matched by the push jobs must be disjoint to avoid races.

 N pull jobs from 1 source
-~~~~~~~~~~~~~~~~~~~~~~~~~
+^^^^^^^^^^^^^^^^^^^^^^^^^

 Multiple pull jobs pulling from the same source have potential for race conditions during pruning:
 each pull job prunes the source side independently, causing replication-prune and prune-prune races.

 There is currently no way for a pull job to filter which snapshots it should attempt to replicate.
-Thus, it is not possibe to just manually assert that the prune rules of all pull jobs are disjoint to avoid replication-prune and prune-prune races.
+Thus, it is not possible to just manually assert that the prune rules of all pull jobs are disjoint to avoid replication-prune and prune-prune races.

@@ -46,6 +46,8 @@ Example Configuration:
             grid: 1x1h(keep=all) | 24x1h | 35x1d | 6x30d
             regex: "^zrepl_.*"
           # manually created snapshots will be kept forever on receiver
+           - type: regex
+             regex: "^manual_.*"

 .. DANGER::
    You might have **existing snapshots** of filesystems affected by pruning which you want to keep, i.e. not be destroyed by zrepl.
@@ -154,7 +156,7 @@ Policy ``regex``
           negate: true
           regex: "^zrepl_.*"

-``regex`` keeps all snapshots whose names are matched by the regular expressionin ``regex``.
+``regex`` keeps all snapshots whose names are matched by the regular expression in ``regex``.
 Like all other regular expression fields in prune policies, zrepl uses Go's `regexp.Regexp <https://golang.org/pkg/regexp/#Compile>`_ Perl-compatible regular expressions (`Syntax <https://golang.org/pkg/regexp/syntax>`_).
 The optional `negate` boolean field inverts the semantics: Use it if you want to keep all snapshots that *do not* match the given regex.

@@ -0,0 +1,47 @@
+.. include:: ../global.rst.inc
+
+
+Replication Options
+===================
+
+
+::
+   
+   jobs:
+   - type: push
+     filesystems: ...
+     replication:
+       protection:
+         initial:     guarantee_resumability # guarantee_{resumability,incremental,nothing}
+         incremental: guarantee_resumability # guarantee_{resumability,incremental,nothing}
+     ...
+
+.. _replication-option-protection:
+
+``protection`` option
+--------------------------
+
+The ``protection`` variable controls the degree to which a replicated filesystem is protected from getting out of sync through a zrepl pruner or external tools that destroy snapshots.
+zrepl can guarantee :ref:`resumability <step-holds>` or just :ref:`incremental replication <replication-cursor-and-last-received-hold>`.
+
+``guarantee_resumability`` is the **default** value and guarantees that a replication step is always resumable and that incremental replication will always be possible.
+The implementation uses replication cursors, last-received-hold and step holds.
+
+``guarantee_incremental`` only guarantees that incremental replication will always be possible.
+If a step ``from -> to`` is interrupted and its `to` snapshot is destroyed, zrepl will remove the half-received ``to``'s resume state and start a new step ``from -> to2``.
+The implementation uses replication cursors, tentative replication cursors and last-received-hold.
+
+``guarantee_nothing`` does not make any guarantees with regards to keeping sending and receiving side in sync.
+No bookmarks or holds are created to protect sender and receiver from diverging.
+
+**Tradeoffs**
+
+Using ``guarantee_incremental`` instead of ``guarantee_resumability`` obviously removes the resumability guarantee.
+This means that replication progress is no longer monotonic which might lead to a replication setup that never makes progress if mid-step interruptions are too frequent (e.g. frequent network outages).
+However, the advantage and :issue:`reason for existence <288>` of the ``incremental`` mode is that it allows the pruner to delete snapshots of interrupted replication steps
+which is useful if replication happens so rarely (or fails so frequently) that the amount of disk space exclusively referenced by the step's snapshots becomes intolerable.
+
+.. NOTE::
+
+   When changing this flag, obsoleted zrepl-managed bookmarks and holds will be destroyed on the next replication step that is attempted for each filesystem.
+
@@ -24,7 +24,7 @@ Send Options
 ---------------------

 The ``encryption`` variable controls whether the matched filesystems are sent as `OpenZFS native encryption <http://open-zfs.org/wiki/ZFS-Native_Encryption>`_ raw sends.
-More specificially, if ``encryption=true``, zrepl
+More specifically, if ``encryption=true``, zrepl

 * checks for any of the filesystems matched by ``filesystems`` whether the ZFS ``encryption`` property indicates that the filesystem is actually encrypted with ZFS native encryption and
 * invokes the ``zfs send`` subcommand with the ``-w`` option (raw sends) and
@@ -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-*"
        }
      ...

@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+
+import subprocess
+import re
+
+output = subprocess.run(["git", "tag", "-l"], capture_output=True, check=True, text=True)
+tagRE = re.compile(r"^v(?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)(-rc(?P<rc>\d+))?$")
+class Tag:
+    orig: str
+    major: int
+    minor: int
+    patch: int
+    rc: int
+
+    def __str__(self):
+        return self.orig
+
+    def __repr__(self):
+        return self.orig
+
+tags = []
+for line in output.stdout.split("\n"):
+    m = tagRE.match(line)
+    if not m:
+        continue
+    m = m.groupdict()
+
+    t = Tag()
+    t.orig = line
+    t.major = int(m["major"])   
+    t.minor = int(m["minor"])   
+    t.patch = int(m["patch"])
+    t.rc = int(m["rc"] if m["rc"] is not None else 0)
+
+    tags.append(t)
+
+by_major_minor = {}
+
+for tag in tags:
+    key = (tag.major, tag.minor)
+    l = by_major_minor.get(key, [])
+    l.append(tag)
+    by_major_minor[key] = l
+
+latest_by_major_minor = []
+for (mm, l) in by_major_minor.items():
+    # sort ascending by patch-level (rc's weigh less)
+    l.sort(key=lambda tag: (tag.patch, int(tag.rc == 0), tag.rc))
+    latest_by_major_minor.append(l[-1])
+latest_by_major_minor.sort(key=lambda tag: (tag.major, tag.minor))
+
+# print(by_major_minor)
+# print(latest_by_major_minor)
+
+cmdline = []
+
+for latest_patch in latest_by_major_minor:
+    cmdline.append("--whitelist-tags")
+    cmdline.append(f"^{re.escape(latest_patch.orig)}$")
+
+# we want to render the latest non-rc version as the default page
+# (latest_by_major_minor is already sorted)
+default_version = latest_by_major_minor[-1]
+for tag in reversed(latest_by_major_minor):
+    if tag.rc == 0:
+        default_version = tag
+        break
+
+cmdline.extend(["--root-ref", f"{default_version}"])
+cmdline.extend(["--banner-main-ref", f"{default_version}"])
+cmdline.extend(["--show-banner"])
+cmdline.extend(["--sort", "semver"])
+
+cmdline.extend(["--whitelist-branches", "master"])
+
+print(" ".join(cmdline))
@@ -11,10 +11,12 @@
   :target: https://zrepl.github.io
 .. |Donate via PayPal| image:: https://img.shields.io/badge/donate-paypal-yellow.svg
   :target: https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=R5QSXJVYHGX96
-.. |Donate via Liberapay| image:: https://img.shields.io/liberapay/receives/zrepl.svg?logo=liberapay
+.. |Donate via Liberapay| image:: https://img.shields.io/liberapay/patrons/zrepl.svg?logo=liberapay
   :target: https://liberapay.com/zrepl/donate
 .. |Donate via Patreon| image:: https://img.shields.io/endpoint.svg?url=https%3A%2F%2Fshieldsio-patreon.herokuapp.com%2Fzrepl%2Fpledges&style=flat&color=yellow
   :target: https://www.patreon.com/zrepl
+.. |Donate via GitHub Sponsors| image:: https://img.shields.io/static/v1?label=Sponsor&message=%E2%9D%A4&logo=GitHub&style=flat&color=yellow
+   :target: https://github.com/sponsors/problame
 .. |Twitter| image:: https://img.shields.io/twitter/url/https/github.com/zrepl/zrepl.svg?style=social
   :target: https://twitter.com/intent/tweet?text=Wow:&url=https%3A%2F%2Fgithub.com%2Fzrepl%2Fzrepl

@@ -25,4 +27,8 @@
 .. |recv-options| replace:: :ref:`recv options<job-recv-options>`
 .. |snapshotting-spec| replace:: :ref:`snapshotting specification <job-snapshotting-spec>`
 .. |pruning-spec| replace:: :ref:`pruning specification <prune>`
-.. |filter-spec| replace:: :ref:`filter specification<pattern-filter>`
+.. |filter-spec| replace:: :ref:`filter specification<pattern-filter>`
+
+.. |br| raw:: html
+
+   <br/>
@@ -5,7 +5,7 @@

 .. include:: global.rst.inc

-|GitHub license| |Language: Go| |Twitter| |Donate via Patreon| |Donate via Liberapay| |Donate via PayPal| 
+|GitHub license| |Language: Go| |Twitter| |Donate via Patreon| |Donate via GitHub Sponsors| |Donate via Liberapay| |Donate via PayPal|


 zrepl - ZFS replication
@@ -44,7 +44,7 @@ zrepl - ZFS replication
 Getting started
 ~~~~~~~~~~~~~~~

-The :ref:`10 minutes tutorial setup <tutorial>` gives you a first impression.
+The :ref:`10 minute quick-start guides <quickstart-toc>` give you a first impression.

 Main Features
 ~~~~~~~~~~~~~
@@ -125,7 +125,7 @@ Table of Contents
   :maxdepth: 2
   :caption: Contents:

-   tutorial
+   quickstart
   installation
   configuration
   usage
@@ -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.
+    Note: check out the :ref:`quick-start guides <quickstart-toc>` 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 apropriate 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
@@ -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.
@@ -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:`quick-start guides <quickstart-toc>` may need to be adjusted.
@@ -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:`quick-start guides <quickstart-toc>` for examples of a ``sink`` job.
+
+Now ``zrepl`` can be started.
+
+.. code-block:: bash
+
+   service zrepl start
+
+Summary
+-------
+
+Congratulations, you have a working jail!
@@ -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.
@@ -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.
@@ -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:`quick-start guides <quickstart-toc>`.
@@ -51,11 +51,10 @@ git rm -rf .
 popd

 echo "building site"
+
 set -e
 sphinx-versioning build \
-   --show-banner \
-   --whitelist-branches '^master$' \
-   --whitelist-tags '(v)?\d+\.[1-9][0-9]*.\d+$' \
+   $(python3 gen-sphinx-versioning-flags.py) \
   docs ./public_git \
   -- -c sphinxconf # older conf.py throw errors because they used
                    # version = subprocess.show_output(["git", "describe"])
@@ -0,0 +1,87 @@
+.. include:: global.rst.inc
+
+.. _quickstart-toc:
+
+***********************
+Quick Start by Use Case
+***********************
+
+The goal of this quick-start guide is to give you an impression of how zrepl can accomodate your use case.
+
+Install zrepl
+=============
+
+Follow the :ref:`OS-specific installation instructions <installation_toc>` and come back here.
+
+Overview Of How zrepl Works
+============================
+
+Check out the :ref:`overview section <job-overview>` to get a rough idea of what you are going to configure in the next step, then come back here.
+
+Configuration Examples
+======================
+
+zrepl is configured through a YAML configuration file in ``/etc/zrepl/zrepl.yml``.
+We have prepared example use cases that show-case typical deployments and different functionality of zrepl.
+We encourage you to read through all of the examples to get an idea of what zrepl has to offer, and how you can mix-and-match configurations for your use case.
+Keep the :ref:`full config documentation <configuration_toc>` handy if a config snippet is unclear.
+
+**Example Use Cases**
+
+.. toctree::
+   :titlesonly:
+
+   quickstart/continuous_server_backup
+   quickstart/backup_to_external_disk
+
+Use ``zrepl configcheck`` to validate your configuration.
+No output indicates that everything is fine.
+
+.. NOTE::
+
+   Please open an issue on GitHub if your use case for zrepl is significantly different from those listed above.
+   Or even better, write it up in the same style as above and open a PR!
+
+.. _quickstart-apply-config:
+
+Apply Configuration Changes
+===========================
+
+We hope that you have found a configuration that fits your use case.
+Use ``zrepl configcheck`` once again to make sure the config is correct (output indicates that everything is fine).
+Then restart the zrepl daemon on all systems involved in the replication, likely using ``service zrepl restart`` or ``systemctl restart zrepl``.
+
+Watch it Work
+=============
+
+Run ``zrepl status`` on the active side of the replication setup to monitor snaphotting, replication and pruning activity.
+To re-trigger replication (snapshots are separate!), use ``zrepl signal wakeup JOBNAME``.
+(refer to the example use case document if you are uncertain which job you want to wake up).
+
+You can also use basic UNIX tools to inspect see what's going on.
+If you like tmux, here is a handy script that works on FreeBSD: ::
+
+    pkg install gnu-watch tmux
+    tmux new -s zrepl -d
+    tmux split-window -t zrepl "tail -f /var/log/messages"
+    tmux split-window -t zrepl "gnu-watch 'zfs list -t snapshot -o name,creation -s creation'"
+    tmux split-window -t zrepl "zrepl status"
+    tmux select-layout -t zrepl tiled
+    tmux attach -t zrepl
+
+The Linux equivalent might look like this: ::
+
+    # make sure tmux is installed & let's assume you use systemd + journald
+    tmux new -s zrepl -d
+    tmux split-window -t zrepl  "journalctl -f -u zrepl.service"
+    tmux split-window -t zrepl "watch 'zfs list -t snapshot -o name,creation -s creation'"
+    tmux split-window -t zrepl "zrepl status"
+    tmux select-layout -t zrepl tiled
+    tmux attach -t zrepl
+
+What Next?
+==========
+
+* Read more about :ref:`configuration format, options & job types <configuration_toc>`
+* Configure :ref:`logging <logging>` \& :ref:`monitoring <monitoring>`.
+
@@ -0,0 +1,36 @@
+.. include:: ../global.rst.inc
+
+.. _quickstart-backup-to-external-disk:
+
+Local Snapshots + Offline Backup to an External Disk
+====================================================
+
+This config example shows how we can use zrepl to make periodic snapshots of our local workstation and back it up to a zpool on an external disk which we occassionally connect.
+
+The local snapshots should be taken every 15 minutes for pain-free recovery from CLI disasters (``rm -rf /`` and the like).
+However, we do not want to keep the snapshots around for very long because our workstation is a little tight on disk space.
+Thus, we only keep one hour worth of high-resolution snapshots, then fade them out to one per hour for a day (24 hours), then one per day for 14 days.
+
+At the end of each work day, we connect our external disk that serves as our workstation's local offline backup.
+We want zrepl to inspect the filesystems and snapshots on the external pool, figure out which snapshots were created since the last time we connected the external disk, and use incremental replication to efficiently mirror our workstation to our backup disk.
+Afterwards, we want to clean up old snapshots on the backup pool: we want to keep all snapshots younger than one hour, 24 for each hor of the first day, then 360 daily backups.
+
+A few additional requirements:
+
+* Snapshot creation and pruning on our workstation should happen in the background, without interaction from our side.
+* However, we want to explicitly trigger replication via the command line.
+* We want to use OpenZFS native encryption to protect our data on the external disk.
+  It is absolutely critical that only encrypted data leaves our workstation.
+  **zrepl should provide an easy config knob for this and prevent replication of unencrypted datasets to the external disk.**
+* We want to be able to put off the backups for more than three weeks, i.e., longer than the lifetime of the automatically created snapshots on our workstation.
+  **zrepl should use bookmarks and holds to achieve this goal**.
+* When we yank out the drive during replication and go on a long vacation, we do *not* want the partially replicated snapshot to stick around as it would hold on to too much disk space over time.
+  Therefore, we want zrepl to deviate from its :ref:`default behavior <replication-option-protection>` and sacrifice resumability, but nonetheless retain the ability to do incremental replication once we return from our vacation.
+  **zrepl should provide an easy config knob to disable step holds for incremental replication**.
+
+The following config snippet implements the setup described above.
+You will likely want to customize some aspects mentioned in the top comment in the file.
+
+.. literalinclude:: ../../config/samples/quickstart_backup_to_external_disk.yml
+
+:ref:`Click here <quickstart-apply-config>` to go back to the quickstart guide.
@@ -0,0 +1,97 @@
+.. include:: ../global.rst.inc
+
+.. _quickstart-continuous-replication:
+
+Continuous Backup of a Server
+=============================
+
+This config example shows how we can backup our ZFS-based server to another machine using a zrepl push job.
+
+* Production server ``prod`` with filesystems to back up:
+
+  * ``zroot/var/db``
+  * ``zroot/usr/home`` and all its child filesystems
+  * **except** ``zroot/usr/home/paranoid`` belonging to a user doing backups themselves
+
+* Backup server ``backups`` with
+
+  * Filesystem ``storage/zrepl/sink/prod`` + children dedicated to backups of ``prod``
+
+Our backup solution should fulfill the following requirements:
+
+* Periodically snapshot the filesystems on ``prod`` *every 10 minutes*
+* Incrementally replicate these snapshots to ``storage/zrepl/sink/prod/*`` on ``backups``
+* Keep only very few snapshots on ``prod`` to save disk space
+* Keep a fading history (24 hourly, 30 daily, 6 monthly) of snapshots on ``backups``
+* The network is untrusted - zrepl should use TLS to protect its communication and our data.
+
+Analysis
+--------
+
+We can model this situation as two jobs:
+
+* A **push job** on ``prod``
+
+  * Creates the snapshots
+  * Keeps a short history of local snapshots to enable incremental replication to ``backups``
+  * Connects to the ``zrepl daemon`` process on ``backups``
+  * Pushes snapshots ``backups``
+  * Prunes snapshots on ``backups`` after replication is complete
+
+* A **sink job** on ``backups``
+
+  * Accepts connections & responds to requests from ``prod``
+  * Limits client ``prod`` access to filesystem sub-tree ``storage/zrepl/sink/prod``
+
+Generate TLS Certificates
+-------------------------
+
+We use the :ref:`TLS client authentication transport <transport-tcp+tlsclientauth>` to protect our data on the wire.
+To get things going quickly, we skip setting up a CA and generate two self-signed certificates as described :ref:`here <transport-tcp+tlsclientauth-2machineopenssl>`.
+For convenience, we generate the key pairs on our local machine and distribute them using ssh:
+
+.. code-block:: bash
+   :emphasize-lines: 6,13
+
+   openssl req -x509 -sha256 -nodes \
+      -newkey rsa:4096 \
+      -days 365 \
+      -keyout backups.key \
+      -out backups.crt
+   # ... and use "backups" as Common Name (CN)
+
+   openssl req -x509 -sha256 -nodes \
+      -newkey rsa:4096 \
+      -days 365 \
+      -keyout prod.key \
+      -out prod.crt
+   # ... and use "prod" as Common Name (CN)
+
+   ssh root@backups "mkdir /etc/zrepl"
+   scp  backups.key backups.crt prod.crt root@backups:/etc/zrepl
+
+   ssh root@prod "mkdir /etc/zrepl"
+   scp  prod.key prod.crt backups.crt root@prod:/etc/zrepl
+
+Note that alternative transports exist, e.g. via :ref:`TCP without TLS <transport-tcp>` or :ref:`ssh <transport-ssh+stdinserver>`.
+
+Configure server ``prod``
+-------------------------
+
+We define a **push job** named ``prod_to_backups`` in ``/etc/zrepl/zrepl.yml`` on host ``prod`` :
+
+.. literalinclude:: ../../config/samples/quickstart_continuous_server_backup_sender.yml
+
+.. _tutorial-configure-prod:
+
+Configure server ``backups``
+----------------------------
+
+We define a corresponding **sink job** named ``sink`` in ``/etc/zrepl/zrepl.yml`` on host ``backups`` :
+
+.. literalinclude:: ../../config/samples/quickstart_continuous_server_backup_receiver.yml
+
+Go Back To Quickstart Guide
+---------------------------
+
+:ref:`Click here <quickstart-apply-config>` to go back to the quickstart guide.
@@ -2,11 +2,11 @@

 .. _supporters:

-|Donate via Patreon| |Donate via Liberapay| |Donate via PayPal|
+|Donate via Patreon| |Donate via GitHub Sponsors| |Donate via Liberapay| |Donate via PayPal|

 zrepl is a spare-time project primarily developed by `Christian Schwarz <https://cschwarz.com>`_.
 You can support maintenance and feature development through one of the services listed above.
-For SEPA wire transfer and **commerical support**, please `contact Christian directly <https://cschwarz.com>`_.
+For SEPA wire transfer and **commercial support**, please `contact Christian directly <https://cschwarz.com>`_.

 **Thanks for your support!**

@@ -32,6 +32,10 @@ We would like to thank the following people and organizations for supporting zre

   <div class="fa fa-code" style="width: 1em;"></div>

+* |supporter-std| `Hans Schulz <https://github.com/schulzh>`_
+* |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>`_
@@ -1,177 +0,0 @@
-.. include:: global.rst.inc
-
-.. _tutorial:
-
-Tutorial
-========
-
-
-This tutorial shows how zrepl can be used to implement a ZFS-based push backup.
-We assume the following scenario:
-
-* Production server ``prod`` with filesystems to back up:
-
-  * ``zroot/var/db``
-  * ``zroot/usr/home`` and all its child filesystems
-  * **except** ``zroot/usr/home/paranoid`` belonging to a user doing backups themselves
-
-* Backup server ``backups`` with
-
-  * Filesystem ``storage/zrepl/sink/prod`` + children dedicated to backups of ``prod``
-
-Our backup solution should fulfill the following requirements:
-
-* Periodically snapshot the filesystems on ``prod`` *every 10 minutes*
-* Incrementally replicate these snapshots to ``storage/zrepl/sink/prod/*`` on ``backups``
-* Keep only very few snapshots on ``prod`` to save disk space
-* Keep a fading history (24 hourly, 30 daily, 6 monthly) of snapshots on ``backups``
-
-Analysis
--------
-
-We can model this situation as two jobs:
-
-* A **push job** on ``prod``
-
-  * Creates the snapshots
-  * Keeps a short history of local snapshots to enable incremental replication to ``backups``
-  * Connects to the ``zrepl daemon`` process on ``backups``
-  * Pushes snapshots ``backups``
-  * Prunes snapshots on ``backups`` after replication is complete
-
-* A **sink job** on ``backups``
-
-  * Accepts connections & responds to requests from ``prod``
-  * Limits client ``prod`` access to filesystem sub-tree ``storage/zrepl/sink/prod``
-
-Install zrepl
-------------
-
-Follow the :ref:`OS-specific installation instructions <installation>` and come back here.
-
-Generate TLS Certificates
-------------------------
-
-We use the :ref:`TLS client authentication transport <transport-tcp+tlsclientauth>` to protect our data on the wire.
-To get things going quickly, we skip setting up a CA and generate two self-signed certificates as described :ref:`here <transport-tcp+tlsclientauth-2machineopenssl>`.
-For convenience, we generate the key pairs on our local machine and distribute them using ssh:
-
-.. code-block:: bash
-   :emphasize-lines: 6,13
-
-   openssl req -x509 -sha256 -nodes \
-      -newkey rsa:4096 \
-      -days 365 \
-      -keyout backups.key \
-      -out backups.crt
-   # ... and use "backups" as Common Name (CN)
-
-   openssl req -x509 -sha256 -nodes \
-      -newkey rsa:4096 \
-      -days 365 \
-      -keyout prod.key \
-      -out prod.crt
-   # ... and use "prod" as Common Name (CN)
-
-   ssh root@backups "mkdir /etc/zrepl"
-   scp  backups.key backups.crt prod.crt root@backups:/etc/zrepl
-
-   ssh root@prod "mkdir /etc/zrepl"
-   scp  prod.key prod.crt backups.crt root@prod:/etc/zrepl
-
-
-Configure server ``prod``
-------------------------
-
-We define a **push job** named ``prod_to_backups`` in ``/etc/zrepl/zrepl.yml`` on host ``prod`` : ::
-
-    jobs:
-    - name: prod_to_backups
-      type: push 
-      connect:
-        type: tls
-        address: "backups.example.com:8888"
-        ca: /etc/zrepl/backups.crt
-        cert: /etc/zrepl/prod.crt
-        key:  /etc/zrepl/prod.key
-        server_cn: "backups"
-      filesystems: {
-        "zroot/var/db": true,
-        "zroot/usr/home<": true,
-        "zroot/usr/home/paranoid": false
-      }
-      snapshotting:
-        type: periodic
-        prefix: zrepl_
-        interval: 10m
-      pruning:
-        keep_sender:
-        - type: not_replicated
-        - type: last_n
-          count: 10
-        keep_receiver:
-        - type: grid
-          grid: 1x1h(keep=all) | 24x1h | 30x1d | 6x30d
-          regex: "^zrepl_"
-
-.. _tutorial-configure-prod:
-
-Configure server ``backups``
----------------------------
-
-We define a corresponding **sink job** named ``sink`` in ``/etc/zrepl/zrepl.yml`` on host ``backups`` : ::
-
-    jobs:
-    - name: sink
-      type: sink 
-      serve:
-         type: tls
-         listen: ":8888"
-         ca: "/etc/zrepl/prod.crt"
-         cert: "/etc/zrepl/backups.crt"
-         key: "/etc/zrepl/backups.key"
-         client_cns:
-           - "prod"
-      root_fs: "storage/zrepl/sink"
-
-
-Apply Configuration Changes
---------------------------
-
-We use ``zrepl configcheck`` to catch any configuration errors: no output indicates that everything is fine.
-If that is the case, restart the zrepl daemon on **both** ``prod`` and ``backups`` using ``service zrepl restart`` or ``systemctl restart zrepl``.
-
-
-Watch it Work
-------------
-
-Run ``zrepl status`` on ``prod`` to monitor the replication and pruning activity.
-To re-trigger replication (snapshots are separate!), use ``zrepl signal wakeup prod_to_backups`` on ``prod``.
-
-If you like tmux, here is a handy script that works on FreeBSD: ::
-
-    pkg install gnu-watch tmux
-    tmux new -s zrepl -d
-    tmux split-window -t zrepl "tail -f /var/log/messages"
-    tmux split-window -t zrepl "gnu-watch 'zfs list -t snapshot -o name,creation -s creation | grep zrepl_'"
-    tmux split-window -t zrepl "zrepl status"
-    tmux select-layout -t zrepl tiled
-    tmux attach -t zrepl
-
-The Linux equivalent might look like this: ::
-
-    # make sure tmux is installed & let's assume you use systemd + journald
-    tmux new -s zrepl -d
-    tmux split-window -t zrepl  "journalctl -f -u zrepl.service"
-    tmux split-window -t zrepl "watch 'zfs list -t snapshot -o name,creation -s creation | grep zrepl_'"
-    tmux split-window -t zrepl "zrepl status"
-    tmux select-layout -t zrepl tiled
-    tmux attach -t zrepl
-
-Summary
-------
-
-Congratulations, you have a working push backup. Where to go next?
-
-* Read more about :ref:`configuration format, options & job types <configuration_toc>`
-* Configure :ref:`logging <logging>` \& :ref:`monitoring <monitoring>`.
@@ -38,6 +38,8 @@ CLI Overview
    * - ``zrepl migrate``
      - | perform on-disk state / ZFS property migrations
        | (see :ref:`changelog <changelog>` for details)
+    * - ``zrepl zfs-abstraction``
+      - list and remove zrepl's abstractions on top of ZFS, e.g. holds and step bookmarks (see :ref:`overview <replication-cursor-and-last-received-hold>` )

 .. _usage-zrepl-daemon:

@@ -48,7 +50,7 @@ zrepl daemon
 All actual work zrepl does is performed by a daemon process.
 The daemon supports structured :ref:`logging <logging>` and provides :ref:`monitoring endpoints <monitoring>`.

-When installating from a package, the package maintainer should have provided an init script / systemd.service file.
+When installing from a package, the package maintainer should have provided an init script / systemd.service file.
 You should thus be able to start zrepl daemon using your init system.

 Alternatively, or for running zrepl in the foreground, simply execute ``zrepl daemon``.
@@ -73,6 +75,6 @@ The daemon exits as soon as all jobs have reported shut down.
 Systemd Unit File
 ~~~~~~~~~~~~~~~~~

-A systemd service defintion template is available in :repomasterlink:`dist/systemd`.
+A systemd service definition template is available in :repomasterlink:`dist/systemd`.
 Note that some of the options only work on recent versions of systemd.
 Any help & improvements are very welcome, see :issue:`145`.
@@ -3,25 +3,18 @@ package endpoint
 import (
 	"context"

+	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 )

 type contextKey int

 const (
-	contextKeyLogger contextKey = iota
-	ClientIdentityKey
+	ClientIdentityKey contextKey = iota
 )

 type Logger = logger.Logger

-func WithLogger(ctx context.Context, log Logger) context.Context {
-	return context.WithValue(ctx, contextKeyLogger, log)
-}
-
 func getLogger(ctx context.Context) Logger {
-	if l, ok := ctx.Value(contextKeyLogger).(Logger); ok {
-		return l
-	}
-	return logger.NewNullLogger()
+	return logging.GetLogger(ctx, logging.SubsysEndpoint)
 }
@@ -2,14 +2,18 @@
 package endpoint

 import (
+	"bytes"
 	"context"
 	"fmt"
+	"io"
 	"path"
-	"sync"

+	"github.com/kr/pretty"
 	"github.com/pkg/errors"
+	"github.com/zrepl/zrepl/daemon/logging/trace"

 	"github.com/zrepl/zrepl/replication/logic/pdu"
+	"github.com/zrepl/zrepl/util/chainedio"
 	"github.com/zrepl/zrepl/util/chainlock"
 	"github.com/zrepl/zrepl/util/envconst"
 	"github.com/zrepl/zrepl/util/semaphore"
@@ -70,6 +74,8 @@ func (s *Sender) filterCheckFS(fs string) (*zfs.DatasetPath, error) {
 }

 func (s *Sender) ListFilesystems(ctx context.Context, r *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error) {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+
 	fss, err := zfs.ZFSListMapping(ctx, s.FSFilter)
 	if err != nil {
 		return nil, err
@@ -92,11 +98,13 @@ func (s *Sender) ListFilesystems(ctx context.Context, r *pdu.ListFilesystemReq)
 }

 func (s *Sender) ListFilesystemVersions(ctx context.Context, r *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error) {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+
 	lp, err := s.filterCheckFS(r.GetFilesystem())
 	if err != nil {
 		return nil, err
 	}
-	fsvs, err := zfs.ZFSListFilesystemVersions(lp, nil)
+	fsvs, err := zfs.ZFSListFilesystemVersions(ctx, lp, zfs.ListFilesystemVersionsOptions{})
 	if err != nil {
 		return nil, err
 	}
@@ -109,36 +117,8 @@ func (s *Sender) ListFilesystemVersions(ctx context.Context, r *pdu.ListFilesyst

 }

-func (p *Sender) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.HintMostRecentCommonAncestorReq) (*pdu.HintMostRecentCommonAncestorRes, error) {
-	var err error
-
-	fs := r.GetFilesystem()
-	mostRecent, err := sendArgsFromPDUAndValidateExists(ctx, fs, r.GetSenderVersion())
-	if err != nil {
-		msg := "HintMostRecentCommonAncestor rpc with nonexistent most recent version"
-		getLogger(ctx).WithField("fs", fs).WithField("hinted_most_recent", fmt.Sprintf("%#v", mostRecent)).
-			Warn(msg)
-		return nil, errors.Wrap(err, msg)
-	}
-
-	// move replication cursor to this position
-	_, err = MoveReplicationCursor(ctx, fs, mostRecent, p.jobId)
-	if err == zfs.ErrBookmarkCloningNotSupported {
-		getLogger(ctx).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")
-	}
-
-	// cleanup previous steps
-	if err := ReleaseStepAll(ctx, fs, mostRecent, p.jobId); err != nil {
-		return nil, errors.Wrap(err, "cannot cleanup prior invocation's step holds and bookmarks")
-	}
-
-	return &pdu.HintMostRecentCommonAncestorRes{}, nil
-}
-
-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 {
@@ -147,18 +127,20 @@ func uncheckedSendArgsFromPDU(fsv *pdu.FilesystemVersion) *zfs.ZFSSendArgVersion
 	return &zfs.ZFSSendArgVersion{RelName: fsv.GetRelName(), GUID: fsv.Guid}
 }

-func sendArgsFromPDUAndValidateExists(ctx context.Context, fs string, fsv *pdu.FilesystemVersion) (*zfs.ZFSSendArgVersion, error) {
-	v := uncheckedSendArgsFromPDU(fsv)
-	if v == nil {
-		return nil, errors.New("must not be nil")
+func sendArgsFromPDUAndValidateExistsAndGetVersion(ctx context.Context, fs string, fsv *pdu.FilesystemVersion) (v zfs.FilesystemVersion, err error) {
+	sendArgs := uncheckedSendArgsFromPDU(fsv)
+	if sendArgs == nil {
+		return v, errors.New("must not be nil")
 	}
-	if err := v.ValidateExists(ctx, fs); err != nil {
-		return nil, err
+	version, err := sendArgs.ValidateExistsAndGetVersion(ctx, fs)
+	if err != nil {
+		return v, err
 	}
-	return v, nil
+	return version, nil
 }

-func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
+func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()

 	_, err := s.filterCheckFS(r.Filesystem)
 	if err != nil {
@@ -170,7 +152,7 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St
 		// ok, fallthrough outer
 	case pdu.Tri_False:
 		if s.encrypt.B {
-			return nil, nil, errors.New("only encrytped sends allowed (send -w + encryption!= off), but unencrytped send requested")
+			return nil, nil, errors.New("only encrypted sends allowed (send -w + encryption!= off), but unencrypted send requested")
 		}
 		// fallthrough outer
 	case pdu.Tri_True:
@@ -182,7 +164,7 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St
 		return nil, nil, fmt.Errorf("unknown pdu.Tri variant %q", r.Encrypted)
 	}

-	sendArgs := zfs.ZFSSendArgs{
+	sendArgsUnvalidated := zfs.ZFSSendArgsUnvalidated{
 		FS:          r.Filesystem,
 		From:        uncheckedSendArgsFromPDU(r.GetFrom()), // validated by zfs.ZFSSendDry / zfs.ZFSSend
 		To:          uncheckedSendArgsFromPDU(r.GetTo()),   // validated by zfs.ZFSSendDry / zfs.ZFSSend
@@ -190,6 +172,11 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St
 		ResumeToken: r.ResumeToken, // nil or not nil, depending on decoding success
 	}

+	sendArgs, err := sendArgsUnvalidated.Validate(ctx)
+	if err != nil {
+		return nil, nil, errors.Wrap(err, "validate send arguments")
+	}
+
 	getLogger(ctx).Debug("acquire concurrent send semaphore")
 	// TODO use try-acquire and fail with resource-exhaustion rpc status
 	// => would require handling on the client-side
@@ -206,7 +193,7 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St
 	}

 	// From now on, assume that sendArgs has been validated by ZFSSendDry
-	// (because validation invovles shelling out, it's actually a little expensive)
+	// (because validation involves shelling out, it's actually a little expensive)

 	var expSize int64 = 0      // protocol says 0 means no estimate
 	if si.SizeEstimate != -1 { // but si returns -1 for no size estimate
@@ -221,134 +208,157 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St
 		return res, nil, nil
 	}

-	// update replication cursor
-	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.From, s.jobId)
-		if err == zfs.ErrBookmarkCloningNotSupported {
-			getLogger(ctx).Debug("not creating replication cursor from bookmark because ZFS does not support it")
-			// fallthrough
-		} else if err != nil {
-			return nil, nil, errors.Wrap(err, "cannot set replication cursor to `from` version before starting send")
+	// create holds or bookmarks of `From` and `To` to guarantee one of the following:
+	// - that the replication step can always be resumed (`holds`),
+	// - that the replication step can be interrupted and a future replication
+	//   step with same or different `To` but same `From` is still possible (`bookmarks`)
+	// - nothing (`none`)
+	//
+	// ...
+	//
+	// ... actually create the abstractions
+	replicationGuaranteeOptions, err := replicationGuaranteeOptionsFromPDU(r.GetReplicationConfig().Protection)
+	if err != nil {
+		return nil, nil, err
+	}
+	replicationGuaranteeStrategy := replicationGuaranteeOptions.Strategy(sendArgs.From != nil)
+	liveAbs, err := replicationGuaranteeStrategy.SenderPreSend(ctx, s.jobId, &sendArgs)
+	if err != nil {
+		return nil, nil, err
+	}
+	for _, a := range liveAbs {
+		if a != nil {
+			abstractionsCacheSingleton.Put(a)
 		}
 	}

-	// make sure `From` doesn't go away in order to make this step resumable
-	if sendArgs.From != nil {
-		err := HoldStep(ctx, sendArgs.FS, sendArgs.From, s.jobId)
-		if err == zfs.ErrBookmarkCloningNotSupported {
-			getLogger(ctx).Debug("not creating step bookmark because ZFS does not support it")
-			// fallthrough
-		} else if err != nil {
-			return nil, nil, errors.Wrap(err, "cannot create step bookmark")
+	// 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()
+
+		keep := func(a Abstraction) (keep bool) {
+			keep = false
+			for _, k := range liveAbs {
+				keep = keep || AbstractionEquals(a, k)
+			}
+			return keep
 		}
-	}
-	// make sure `To` doesn't go away in order to make this step resumable
-	err = HoldStep(ctx, sendArgs.FS, sendArgs.To, s.jobId)
-	if err != nil {
-		return nil, nil, errors.Wrapf(err, "cannot hold `to` version %q before starting send", sendArgs.To.RelName)
-	}
+		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 {
+					isSendArg := zfs.FilesystemVersionEqualIdentity(mustLiveVersion, staleVersion.GetFilesystemVersion())
+					stepHoldBasedGuaranteeStrategy := false
+					k := replicationGuaranteeStrategy.Kind()
+					switch k {
+					case ReplicationGuaranteeKindResumability:
+						stepHoldBasedGuaranteeStrategy = true
+					case ReplicationGuaranteeKindIncremental:
+					case ReplicationGuaranteeKindNone:
+					default:
+						panic(fmt.Sprintf("this is supposed to be an exhaustive match, got %v", k))
+					}
+					isSnapshot := mustLiveVersion.IsSnapshot()
+					if isSendArg && (!isSnapshot || stepHoldBasedGuaranteeStrategy) {
+						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)))
+					}
+				}
+			}
+		}
+		destroyTypes := AbstractionTypeSet{
+			AbstractionStepHold:                           true,
+			AbstractionTentativeReplicationCursorBookmark: true,
+		}
+		abstractionsCacheSingleton.TryBatchDestroy(ctx, s.jobId, sendArgs.FS, destroyTypes, keep, check)
+	}()

-	// step holds & replication cursor released / moved forward in s.SendCompleted => s.moveCursorAndReleaseSendHolds
-
-	streamCopier, err := zfs.ZFSSend(ctx, sendArgs)
+	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
 	if err != nil {
+		// it's ok to not destroy the abstractions we just created here, a new send attempt will take care of it
 		return nil, nil, errors.Wrap(err, "zfs send failed")
 	}
-	return res, streamCopier, nil
+
+	return res, sendStream, nil
 }

 func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
-	orig := r.GetOriginalReq() // may be nil, always use proto getters
-	fs := orig.GetFilesystem()
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()

-	var err error
-	var from *zfs.ZFSSendArgVersion
+	orig := r.GetOriginalReq() // may be nil, always use proto getters
+	fsp, err := p.filterCheckFS(orig.GetFilesystem())
+	if err != nil {
+		return nil, err
+	}
+	fs := fsp.ToString()
+
+	var from *zfs.FilesystemVersion
 	if orig.GetFrom() != nil {
-		from, err = sendArgsFromPDUAndValidateExists(ctx, fs, orig.GetFrom()) // no shadow
+		f, err := sendArgsFromPDUAndValidateExistsAndGetVersion(ctx, fs, orig.GetFrom()) // no shadow
 		if err != nil {
 			return nil, errors.Wrap(err, "validate `from` exists")
 		}
+		from = &f
 	}
-	to, err := sendArgsFromPDUAndValidateExists(ctx, fs, orig.GetTo())
+	to, err := sendArgsFromPDUAndValidateExistsAndGetVersion(ctx, fs, orig.GetTo())
 	if err != nil {
 		return nil, errors.Wrap(err, "validate `to` exists")
 	}

-	log := getLogger(ctx).WithField("to_guid", to.GUID).
-		WithField("fs", fs).
-		WithField("to", to.RelName)
-	if from != nil {
-		log = log.WithField("from", from.RelName).WithField("from_guid", from.GUID)
-	}
-
-	log.Debug("move replication cursor to most recent common version")
-	destroyedCursors, err := MoveReplicationCursor(ctx, fs, to, p.jobId)
+	replicationGuaranteeOptions, err := replicationGuaranteeOptionsFromPDU(orig.GetReplicationConfig().Protection)
 	if err != nil {
-		if err == zfs.ErrBookmarkCloningNotSupported {
-			log.Debug("not setting replication cursor, bookmark cloning not supported")
-		} else {
-			msg := "cannot move replication cursor, keeping hold on `to` until successful"
-			log.WithError(err).Error(msg)
-			err = errors.Wrap(err, msg)
-			// it is correct to not release the hold if we can't move the cursor!
-			return &pdu.SendCompletedRes{}, err
-		}
-	} else {
-		log.Info("successfully moved replication cursor")
+		return nil, err
 	}
-
-	// 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()
-		log.Debug("release step-hold of or step-bookmark on `to`")
-		err = ReleaseStep(ctx, fs, to, p.jobId)
-		if err != nil {
-			log.WithError(err).Error("cannot release step-holds on or destroy step-bookmark of `to`")
-		} else {
-			log.Info("successfully released step-holds on or destroyed step-bookmark of `to`")
+	liveAbs, err := replicationGuaranteeOptions.Strategy(from != nil).SenderPostRecvConfirmed(ctx, p.jobId, fs, to)
+	if err != nil {
+		return nil, err
+	}
+	for _, a := range liveAbs {
+		if a != nil {
+			abstractionsCacheSingleton.Put(a)
 		}
-
-	}()
-	go func() {
-		defer wg.Done()
-		if from == nil {
-			return
+	}
+	keep := func(a Abstraction) (keep bool) {
+		keep = false
+		for _, k := range liveAbs {
+			keep = keep || AbstractionEquals(a, k)
 		}
-		log.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.
-				fsp, err := zfs.NewDatasetPath(fs)
-				if err != nil {
-					panic(err) // fs has been validated multiple times above
-				}
-				for _, fsv := range destroyedCursors {
-					if fsv.ToAbsPath(fsp) == dne.Path {
-						log.Info("`from` was a replication cursor and has already been destroyed")
-						return
-					}
-				}
-				// fallthrough
-			}
-			log.WithError(err).Error("cannot release step-holds on or destroy step-bookmark of `from`")
-		} else {
-			log.Info("successfully released step-holds on or destroyed step-bookmark of `from`")
-		}
-	}()
-	wg.Wait()
+		return keep
+	}
+	destroyTypes := AbstractionTypeSet{
+		AbstractionStepHold:                           true,
+		AbstractionTentativeReplicationCursorBookmark: true,
+		AbstractionReplicationCursorBookmarkV2:        true,
+	}
+	abstractionsCacheSingleton.TryBatchDestroy(ctx, p.jobId, fs, destroyTypes, keep, nil)

 	return &pdu.SendCompletedRes{}, nil
+
 }

 func (p *Sender) DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshotsReq) (*pdu.DestroySnapshotsRes, error) {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+
 	dp, err := p.filterCheckFS(req.Filesystem)
 	if err != nil {
 		return nil, err
@@ -357,6 +367,8 @@ func (p *Sender) DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshots
 }

 func (p *Sender) Ping(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, error) {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+
 	res := pdu.PingRes{
 		Echo: req.GetMessage(),
 	}
@@ -364,14 +376,20 @@ func (p *Sender) Ping(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, erro
 }

 func (p *Sender) PingDataconn(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, error) {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+
 	return p.Ping(ctx, req)
 }

 func (p *Sender) WaitForConnectivity(ctx context.Context) error {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+
 	return nil
 }

 func (p *Sender) ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error) {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+
 	dp, err := p.filterCheckFS(req.Filesystem)
 	if err != nil {
 		return nil, err
@@ -387,7 +405,7 @@ func (p *Sender) ReplicationCursor(ctx context.Context, req *pdu.ReplicationCurs
 	return &pdu.ReplicationCursorRes{Result: &pdu.ReplicationCursorRes_Guid{Guid: cursor.Guid}}, nil
 }

-func (p *Sender) Receive(ctx context.Context, r *pdu.ReceiveReq, receive zfs.StreamCopier) (*pdu.ReceiveRes, error) {
+func (p *Sender) Receive(ctx context.Context, r *pdu.ReceiveReq, _ io.ReadCloser) (*pdu.ReceiveRes, error) {
 	return nil, fmt.Errorf("sender does not implement Receive()")
 }

@@ -408,8 +426,6 @@ type ReceiverConfig struct {

 	RootWithoutClientComponent *zfs.DatasetPath // TODO use
 	AppendClientIdentity       bool
-
-	UpdateLastReceivedHold bool
 }

 func (c *ReceiverConfig) copyIn() {
@@ -502,6 +518,15 @@ func (f subroot) MapToLocal(fs string) (*zfs.DatasetPath, error) {
 }

 func (s *Receiver) ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error) {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+
+	// first make sure that root_fs is imported
+	if rphs, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, s.conf.RootWithoutClientComponent); err != nil {
+		return nil, errors.Wrap(err, "cannot determine whether root_fs exists")
+	} else if !rphs.FSExists {
+		return nil, errors.New("root_fs does not exist")
+	}
+
 	root := s.clientRootFromCtx(ctx)
 	filtered, err := zfs.ZFSListMapping(ctx, subroot{root})
 	if err != nil {
@@ -511,7 +536,7 @@ func (s *Receiver) ListFilesystems(ctx context.Context, req *pdu.ListFilesystemR
 	fss := make([]*pdu.Filesystem, 0, len(filtered))
 	for _, a := range filtered {
 		l := getLogger(ctx).WithField("fs", a)
-		ph, err := zfs.ZFSGetFilesystemPlaceholderState(a)
+		ph, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, a)
 		if err != nil {
 			l.WithError(err).Error("error getting placeholder state")
 			return nil, errors.Wrapf(err, "cannot get placeholder state for fs %q", a)
@@ -552,13 +577,16 @@ func (s *Receiver) ListFilesystems(ctx context.Context, req *pdu.ListFilesystemR
 }

 func (s *Receiver) ListFilesystemVersions(ctx context.Context, req *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error) {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+
 	root := s.clientRootFromCtx(ctx)
 	lp, err := subroot{root}.MapToLocal(req.GetFilesystem())
 	if err != nil {
 		return nil, err
 	}
+	// TODO share following code with sender

-	fsvs, err := zfs.ZFSListFilesystemVersions(lp, nil)
+	fsvs, err := zfs.ZFSListFilesystemVersions(ctx, lp, zfs.ListFilesystemVersionsOptions{})
 	if err != nil {
 		return nil, err
 	}
@@ -572,6 +600,8 @@ func (s *Receiver) ListFilesystemVersions(ctx context.Context, req *pdu.ListFile
 }

 func (s *Receiver) Ping(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, error) {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+
 	res := pdu.PingRes{
 		Echo: req.GetMessage(),
 	}
@@ -579,24 +609,30 @@ func (s *Receiver) Ping(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, er
 }

 func (s *Receiver) PingDataconn(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, error) {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 	return s.Ping(ctx, req)
 }

 func (s *Receiver) WaitForConnectivity(ctx context.Context) error {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 	return nil
 }

-func (s *Receiver) ReplicationCursor(context.Context, *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error) {
+func (s *Receiver) ReplicationCursor(ctx context.Context, _ *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error) {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 	return nil, fmt.Errorf("ReplicationCursor not implemented for Receiver")
 }

-func (s *Receiver) Send(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
+func (s *Receiver) Send(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 	return nil, nil, fmt.Errorf("receiver does not implement Send()")
 }

 var maxConcurrentZFSRecvSemaphore = semaphore.New(envconst.Int64("ZREPL_ENDPOINT_MAX_CONCURRENT_RECV", 10))

-func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs.StreamCopier) (*pdu.ReceiveRes, error) {
+func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.ReadCloser) (*pdu.ReceiveRes, error) {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+
 	getLogger(ctx).Debug("incoming Receive")
 	defer receive.Close()

@@ -610,9 +646,6 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 	if to == nil {
 		return nil, errors.New("`To` must not be nil")
 	}
-	if err := to.ValidateInMemory(lp.ToString()); err != nil {
-		return nil, errors.Wrap(err, "`To` invalid")
-	}
 	if !to.IsSnapshot() {
 		return nil, errors.New("`To` must be a snapshot")
 	}
@@ -625,19 +658,19 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 	//       ZFS dataset hierarchy subtrees.
 	var visitErr error
 	func() {
-		getLogger(ctx).Debug("begin aquire recvParentCreationMtx")
+		getLogger(ctx).Debug("begin acquire recvParentCreationMtx")
 		defer s.recvParentCreationMtx.Lock().Unlock()
-		getLogger(ctx).Debug("end aquire recvParentCreationMtx")
+		getLogger(ctx).Debug("end acquire recvParentCreationMtx")
 		defer getLogger(ctx).Debug("release recvParentCreationMtx")

 		f := zfs.NewDatasetPathForest()
 		f.Add(lp)
 		getLogger(ctx).Debug("begin tree-walk")
-		f.WalkTopDown(func(v zfs.DatasetPathVisit) (visitChildTree bool) {
+		f.WalkTopDown(func(v *zfs.DatasetPathVisit) (visitChildTree bool) {
 			if v.Path.Equal(lp) {
 				return false
 			}
-			ph, err := zfs.ZFSGetFilesystemPlaceholderState(v.Path)
+			ph, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, v.Path)
 			getLogger(ctx).
 				WithField("fs", v.Path.ToString()).
 				WithField("placeholder_state", fmt.Sprintf("%#v", ph)).
@@ -661,7 +694,7 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 				}
 				l := getLogger(ctx).WithField("placeholder_fs", v.Path)
 				l.Debug("create placeholder filesystem")
-				err := zfs.ZFSCreatePlaceholderFilesystem(v.Path)
+				err := zfs.ZFSCreatePlaceholderFilesystem(ctx, v.Path, v.Parent.Path)
 				if err != nil {
 					l.WithError(err).Error("cannot create placeholder filesystem")
 					visitErr = err
@@ -678,22 +711,31 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 		return nil, visitErr
 	}

+	log := getLogger(ctx).WithField("proto_fs", req.GetFilesystem()).WithField("local_fs", lp.ToString())
+
 	// determine whether we need to rollback the filesystem / change its placeholder state
 	var clearPlaceholderProperty bool
 	var recvOpts zfs.RecvOptions
-	ph, err := zfs.ZFSGetFilesystemPlaceholderState(lp)
-	if err == nil && ph.FSExists && ph.IsPlaceholder {
+	ph, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, lp)
+	if err != nil {
+		return nil, errors.Wrap(err, "cannot get placeholder state")
+	}
+	log.WithField("placeholder_state", fmt.Sprintf("%#v", ph)).Debug("placeholder state")
+	if ph.FSExists && ph.IsPlaceholder {
 		recvOpts.RollbackAndForceRecv = true
 		clearPlaceholderProperty = true
 	}
+
 	if clearPlaceholderProperty {
-		if err := zfs.ZFSSetPlaceholder(lp, false); err != nil {
+		log.Info("clearing placeholder property")
+		if err := zfs.ZFSSetPlaceholder(ctx, lp, false); err != nil {
 			return nil, fmt.Errorf("cannot clear placeholder property for forced receive: %s", err)
 		}
 	}

 	if req.ClearResumeToken && ph.FSExists {
-		if err := zfs.ZFSRecvClearResumeToken(lp.ToString()); err != nil {
+		log.Info("clearing resume token")
+		if err := zfs.ZFSRecvClearResumeToken(ctx, lp.ToString()); err != nil {
 			return nil, errors.Wrap(err, "cannot clear resume token")
 		}
 	}
@@ -703,7 +745,7 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 		return nil, errors.Wrap(err, "cannot determine whether we can use resumable send & recv")
 	}

-	getLogger(ctx).Debug("acquire concurrent recv semaphore")
+	log.Debug("acquire concurrent recv semaphore")
 	// TODO use try-acquire and fail with resource-exhaustion rpc status
 	// => would require handling on the client-side
 	// => this is a dataconn endpoint, doesn't have the status code semantics of gRPC
@@ -713,36 +755,140 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 	}
 	defer guard.Release()

-	getLogger(ctx).WithField("opts", fmt.Sprintf("%#v", recvOpts)).Debug("start receive command")
+	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")
+	}
+	var peekCopy bytes.Buffer
+	if n, err := peekCopy.Write(peek.Bytes()); err != nil || n != peek.Len() {
+		panic(peek.Len())
+	}
+
+	log.WithField("opts", fmt.Sprintf("%#v", recvOpts)).Debug("start receive command")

 	snapFullPath := to.FullPath(lp.ToString())
-	if err := zfs.ZFSRecv(ctx, lp.ToString(), to, receive, recvOpts); err != nil {
-		getLogger(ctx).
+	if err := zfs.ZFSRecv(ctx, lp.ToString(), to, chainedio.NewChainedReader(&peek, receive), recvOpts); err != nil {
+
+		// best-effort rollback of placeholder state if the recv didn't start
+		_, resumableStatePresent := err.(*zfs.RecvFailedWithResumeTokenErr)
+		disablePlaceholderRestoration := envconst.Bool("ZREPL_ENDPOINT_DISABLE_PLACEHOLDER_RESTORATION", false)
+		placeholderRestored := !ph.IsPlaceholder
+		if !disablePlaceholderRestoration && !resumableStatePresent && recvOpts.RollbackAndForceRecv && ph.FSExists && ph.IsPlaceholder && clearPlaceholderProperty {
+			log.Info("restoring placeholder property")
+			if phErr := zfs.ZFSSetPlaceholder(ctx, lp, true); phErr != nil {
+				log.WithError(phErr).Error("cannot restore placeholder property after failed receive, subsequent replications will likely fail with a different error")
+				// fallthrough
+			} else {
+				placeholderRestored = true
+			}
+			// fallthrough
+		}
+
+		// deal with failing initial encrypted send & recv
+		if _, ok := err.(*zfs.RecvDestroyOrOverwriteEncryptedErr); ok && ph.IsPlaceholder && placeholderRestored {
+			msg := `cannot automatically replace placeholder filesystem with incoming send stream - please see receive-side log for details`
+			err := errors.New(msg)
+			log.Error(msg)
+
+			log.Error(`zrepl creates placeholder filesystems on the receiving side of a replication to match the sending side's dataset hierarchy`)
+			log.Error(`zrepl uses zfs receive -F to replace those placeholders with incoming full sends`)
+			log.Error(`OpenZFS native encryption prohibits zfs receive -F for encrypted filesystems`)
+			log.Error(`the current zrepl placeholder filesystem concept is thus incompatible with OpenZFS native encryption`)
+
+			tempStartFullRecvFS := lp.Copy().ToString() + ".zrepl.initial-recv"
+			tempStartFullRecvFSDP, dpErr := zfs.NewDatasetPath(tempStartFullRecvFS)
+			if dpErr != nil {
+				log.WithError(dpErr).Error("cannot determine temporary filesystem name for initial encrypted recv workaround")
+				return nil, err // yes, err, not dpErr
+			}
+
+			log := log.WithField("temp_recv_fs", tempStartFullRecvFS)
+			log.Error(`as a workaround, zrepl will now attempt to re-receive the beginning of the stream into a temporary filesystem temp_recv_fs`)
+			log.Error(`if that step succeeds: shut down zrepl and use 'zfs rename' to swap temp_recv_fs with local_fs, then restart zrepl`)
+			log.Error(`replication will then resume using resumable send+recv`)
+
+			tempPH, phErr := zfs.ZFSGetFilesystemPlaceholderState(ctx, tempStartFullRecvFSDP)
+			if phErr != nil {
+				log.WithError(phErr).Error("cannot determine placeholder state of temp_recv_fs")
+				return nil, err // yes, err, not dpErr
+			}
+			if tempPH.FSExists {
+				log.Error("temp_recv_fs already exists, assuming a (partial) initial recv to that filesystem has already been done")
+				return nil, err
+			}
+
+			recvOpts.RollbackAndForceRecv = false
+			recvOpts.SavePartialRecvState = true
+			rerecvErr := zfs.ZFSRecv(ctx, tempStartFullRecvFS, to, chainedio.NewChainedReader(&peekCopy), recvOpts)
+			if _, isResumable := rerecvErr.(*zfs.RecvFailedWithResumeTokenErr); rerecvErr == nil || isResumable {
+				log.Error("completed re-receive into temporary filesystem temp_recv_fs, now shut down zrepl and use zfs rename to swap temp_recv_fs with local_fs")
+			} else {
+				log.WithError(rerecvErr).Error("failed to receive the beginning of the stream into temporary filesystem temp_recv_fs")
+				log.Error("we advise you to collect the error log and current configuration, open an issue on GitHub, and revert to your previous configuration in the meantime")
+			}
+
+			log.Error(`if you would like to see improvements to this situation, please open an issue on GitHub`)
+			return nil, err
+		}
+
+		log.
 			WithError(err).
-			WithField("opts", recvOpts).
+			WithField("opts", fmt.Sprintf("%#v", recvOpts)).
 			Error("zfs receive failed")
+
 		return nil, err
 	}

 	// validate that we actually received what the sender claimed
-	if err := to.ValidateExists(ctx, lp.ToString()); err != nil {
+	toRecvd, err := to.ValidateExistsAndGetVersion(ctx, lp.ToString())
+	if err != nil {
 		msg := "receive request's `To` version does not match what we received in the stream"
-		getLogger(ctx).WithError(err).WithField("snap", snapFullPath).Error(msg)
-		getLogger(ctx).Error("aborting recv request, but keeping received snapshot for inspection")
+		log.WithError(err).WithField("snap", snapFullPath).Error(msg)
+		log.Error("aborting recv request, but keeping received snapshot for inspection")
 		return nil, errors.Wrap(err, msg)
 	}

-	if s.conf.UpdateLastReceivedHold {
-		getLogger(ctx).Debug("move last-received-hold")
-		if err := MoveLastReceivedHold(ctx, lp.ToString(), *to, s.conf.JobID); err != nil {
-			return nil, errors.Wrap(err, "cannot move last-received-hold")
+	replicationGuaranteeOptions, err := replicationGuaranteeOptionsFromPDU(req.GetReplicationConfig().Protection)
+	if err != nil {
+		return nil, err
+	}
+	replicationGuaranteeStrategy := replicationGuaranteeOptions.Strategy(ph.FSExists)
+	liveAbs, err := replicationGuaranteeStrategy.ReceiverPostRecv(ctx, s.conf.JobID, lp.ToString(), toRecvd)
+	if err != nil {
+		return nil, err
+	}
+	for _, a := range liveAbs {
+		if a != nil {
+			abstractionsCacheSingleton.Put(a)
 		}
 	}
+	keep := func(a Abstraction) (keep bool) {
+		keep = false
+		for _, k := range liveAbs {
+			keep = keep || AbstractionEquals(a, k)
+		}
+		return keep
+	}
+	check := func(obsoleteAbs []Abstraction) {
+		for _, abs := range obsoleteAbs {
+			if zfs.FilesystemVersionEqualIdentity(abs.GetFilesystemVersion(), toRecvd) {
+				panic(fmt.Sprintf("would destroy endpoint abstraction around the filesystem version we just received %s", abs))
+			}
+		}
+	}
+	destroyTypes := AbstractionTypeSet{
+		AbstractionLastReceivedHold: true,
+	}
+	abstractionsCacheSingleton.TryBatchDestroy(ctx, s.conf.JobID, lp.ToString(), destroyTypes, keep, check)

 	return &pdu.ReceiveRes{}, nil
 }

 func (s *Receiver) DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshotsReq) (*pdu.DestroySnapshotsRes, error) {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+
 	root := s.clientRootFromCtx(ctx)
 	lp, err := subroot{root}.MapToLocal(req.Filesystem)
 	if err != nil {
@@ -751,16 +897,9 @@ 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) {
-	// 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)()

-func (p *Receiver) SendCompleted(context.Context, *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
 	return &pdu.SendCompletedRes{}, nil
 }

@@ -782,7 +921,7 @@ func doDestroySnapshots(ctx context.Context, lp *zfs.DatasetPath, snaps []*pdu.F
 			ErrOut:     &errs[i],
 		}
 	}
-	zfs.ZFSDestroyFilesystemVersions(reqs)
+	zfs.ZFSDestroyFilesystemVersions(ctx, reqs)
 	for i := range reqs {
 		if errs[i] != nil {
 			if de, ok := errs[i].(*zfs.DestroySnapshotsError); ok && len(de.Reason) == 1 {
--- a/Show More
+++ b/Show More