add tracing package (forgot to commit)

.circleci/config.yml

@@ -6,7 +6,6 @@ workflows:
       - build-1.11
       - build-1.12
       - build-1.13
-      - build-1.14
       - build-latest
       - test-build-in-docker
 jobs:
@@ -112,11 +111,6 @@ 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
   #

Makefile

@@ -65,7 +65,6 @@ wrapup-and-checksum:
 		-acf $(NOARCH_TARBALL) \
 		$(ARTIFACTDIR)/docs/html \
 		$(ARTIFACTDIR)/bash_completion \
-		$(ARTIFACTDIR)/_zrepl.zsh_completion \
 		$(ARTIFACTDIR)/go_env.txt \
 		dist \
 		config/samples
@@ -160,7 +159,7 @@ platformtest: # do not track dependency on platformtest-bin to allow build of pl
 		$(ZREPL_PLATFORMTEST_ARGS)
 
 ##################### NOARCH #####################
-.PHONY: noarch $(ARTIFACTDIR)/bash_completion $(ARTIFACTDIR)/_zrepl.zsh_completion $(ARTIFACTDIR)/go_env.txt docs docs-clean
+.PHONY: noarch $(ARTIFACTDIR)/bash_completion $(ARTIFACTDIR)/go_env.txt docs docs-clean
 
 
 $(ARTIFACTDIR):
@@ -168,16 +167,12 @@ $(ARTIFACTDIR):
 $(ARTIFACTDIR)/docs: $(ARTIFACTDIR)
 	mkdir -p "$@"
 
-noarch: $(ARTIFACTDIR)/bash_completion $(ARTIFACTDIR)/_zrepl.zsh_completion $(ARTIFACTDIR)/go_env.txt docs
+noarch: $(ARTIFACTDIR)/bash_completion $(ARTIFACTDIR)/go_env.txt docs
 	# pass
 
 $(ARTIFACTDIR)/bash_completion:
 	$(MAKE) zrepl-bin GOOS=$(GOHOSTOS) GOARCH=$(GOHOSTARCH)
-	artifacts/zrepl-$(GOHOSTOS)-$(GOHOSTARCH) gencompletion bash "$@"
-
-$(ARTIFACTDIR)/_zrepl.zsh_completion:
-	$(MAKE) zrepl-bin GOOS=$(GOHOSTOS) GOARCH=$(GOHOSTARCH)
-	artifacts/zrepl-$(GOHOSTOS)-$(GOHOSTARCH) gencompletion zsh "$@"
+	artifacts/zrepl-$(GOHOSTOS)-$(GOHOSTARCH) bashcomp "$@"
 
 $(ARTIFACTDIR)/go_env.txt:
 	$(GO_ENV_VARS) $(GO) env > $@

README.md

@@ -1,9 +1,8 @@
 [![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 Liberapay](https://img.shields.io/liberapay/receives/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
@@ -24,7 +23,6 @@ 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.
@@ -48,8 +46,11 @@ zrepl is written in [Go](https://golang.org) and uses [Go modules](https://githu
 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`.
-`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 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.
 
 ### Project Structure
 
@@ -61,15 +62,14 @@ If you just want to install the Go dependencies, run `./lazy.sh godep`.
 │   └── 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
-├── dist                    # supplemental material for users & package maintainers
 ├── docs                    # sphinx-based documentation
+├── dist                    # supplemental material for users & package maintainers
 │   ├── **/*.rst            # documentation in reStructuredText
 │   ├── sphinxconf
 │   │   └── conf.py         # sphinx config (see commit 445a280 why its not in docs/)
@@ -78,7 +78,6 @@ If you just want to install the Go dependencies, run `./lazy.sh godep`.
 │   └── 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
@@ -93,13 +92,14 @@ If you just want to install the Go dependencies, run `./lazy.sh godep`.
 │   ├── 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               # transport implementations
+├── transport               # transports implementation
 │   ├── fromconfig
 │   ├── local
 │   ├── ssh
 │   ├── tcp
 │   └── tls
 ├── util
+├── vendor                  # managed by dep
 ├── version                 # abstraction for versions (filled during build by Makefile)
 └── zfs                     # zfs(8) wrappers
 ```
@@ -134,15 +134,3 @@ 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
-

cli/cli.go

@@ -19,55 +19,29 @@ var rootCmd = &cobra.Command{
 	Short: "One-stop ZFS replication solution",
 }
 
-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",
+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)
+		}
 	},
+	Hidden: true,
 }
 
 func init() {
-	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)
+	rootCmd.PersistentFlags().StringVar(&rootArgs.configPath, "config", "", "config file path")
+	rootCmd.AddCommand(bashcompCmd)
 }
 
 type Subcommand struct {

client/holds.go

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

client/migrate.go

@@ -99,7 +99,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(ctx, fs, migratePlaceholder0_1Args.dryRun)
+			r, err := zfs.ZFSMigrateHashBasedPlaceholderToCurrent(fs, migratePlaceholder0_1Args.dryRun)
 			if err != nil {
 				fmt.Printf("error: %s\n", err)
 			} else if !r.NeedsModification {
@@ -165,7 +165,7 @@ func doMigrateReplicationCursor(sc *cli.Subcommand, args []string) error {
 	var hadError bool
 	for _, fs := range fss {
 
-		bold.Printf("INSPECT FILESYSTEM %q\n", fs.ToString())
+		bold.Printf("INSPECT FILESYTEM %q\n", fs.ToString())
 
 		err := doMigrateReplicationCursorFS(ctx, v1cursorJobs, fs)
 		if err == migrateReplicationCursorSkipSentinel {
@@ -211,15 +211,17 @@ func doMigrateReplicationCursorFS(ctx context.Context, v1CursorJobs []job.Job, f
 	}
 	fmt.Printf("identified owning job %q\n", owningJob.Name())
 
-	bookmarks, err := zfs.ZFSListFilesystemVersions(fs, zfs.ListFilesystemVersionsOptions{
-		Types: zfs.Bookmarks,
-	})
+	versions, err := zfs.ZFSListFilesystemVersions(fs, nil)
 	if err != nil {
 		return errors.Wrapf(err, "list filesystem versions of %q", fs.ToString())
 	}
 
 	var oldCursor *zfs.FilesystemVersion
-	for i, fsv := range bookmarks {
+	for i, fsv := range versions {
+		if fsv.Type != zfs.Bookmark {
+			continue
+		}
+
 		_, _, err := endpoint.ParseReplicationCursorBookmarkName(fsv.ToAbsPath(fs))
 		if err != endpoint.ErrV1ReplicationCursor {
 			continue
@@ -230,7 +232,7 @@ func doMigrateReplicationCursorFS(ctx context.Context, v1CursorJobs []job.Job, f
 			return errors.Wrap(err, "multiple filesystem versions identified as v1 replication cursors")
 		}
 
-		oldCursor = &bookmarks[i]
+		oldCursor = &versions[i]
 
 	}
 
@@ -262,7 +264,7 @@ func doMigrateReplicationCursorFS(ctx context.Context, v1CursorJobs []job.Job, f
 	if migrateReplicationCursorArgs.dryRun {
 		succ.Printf("DRY RUN\n")
 	} else {
-		if err := zfs.ZFSDestroyFilesystemVersion(ctx, fs, oldCursor); err != nil {
+		if err := zfs.ZFSDestroyFilesystemVersion(fs, oldCursor); err != nil {
 			return err
 		}
 	}

client/status.go

@@ -11,7 +11,7 @@ import (
 	"sync"
 	"time"
 
-	// tcell is the termbox-compatible library for abstracting away escape sequences, etc.
+	// tcell is the termbox-compatbile 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 +81,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
@@ -219,7 +219,7 @@ func runStatus(s *cli.Subcommand, args []string) error {
 	defer termbox.Close()
 
 	update := func() {
-		var m daemon.Status
+		m := make(map[string]job.Status)
 
 		err2 := jsonRequestResponse(httpc, daemon.ControlJobEndpointStatus,
 			struct{}{},
@@ -228,7 +228,7 @@ func runStatus(s *cli.Subcommand, args []string) error {
 
 		t.lock.Lock()
 		t.err = err2
-		t.report = m.Jobs
+		t.report = m
 		t.lock.Unlock()
 		t.draw()
 	}
@@ -264,7 +264,7 @@ loop:
 
 }
 
-func (t *tui) getReplicationProgressHistory(jobName string) *bytesProgressHistory {
+func (t *tui) getReplicationProgresHistory(jobName string) *bytesProgressHistory {
 	p, ok := t.replicationProgress[jobName]
 	if !ok {
 		p = &bytesProgressHistory{}
@@ -329,7 +329,7 @@ func (t *tui) draw() {
 				t.printf("Replication:")
 				t.newline()
 				t.addIndent(1)
-				t.renderReplicationReport(activeStatus.Replication, t.getReplicationProgressHistory(k))
+				t.renderReplicationReport(activeStatus.Replication, t.getReplicationProgresHistory(k))
 				t.addIndent(-1)
 
 				t.printf("Pruning Sender:")
@@ -697,7 +697,7 @@ func rightPad(str string, length int, pad string) string {
 
 var arrowPositions = `>\|/`
 
-// changeCount = 0 indicates stall / no progress
+// changeCount = 0 indicates stall / no progresss
 func (t *tui) drawBar(length int, bytes, totalBytes int64, changeCount int) {
 	var completedLength int
 	if totalBytes > 0 {

client/testcmd.go

@@ -49,7 +49,6 @@ func runTestFilterCmd(subcommand *cli.Subcommand, args []string) error {
 	}
 
 	conf := subcommand.Config()
-	ctx := context.Background()
 
 	var confFilter config.FilesystemsFilter
 	job, err := conf.Job(testFilterArgs.job)
@@ -61,8 +60,6 @@ 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)
 	}
@@ -76,7 +73,7 @@ func runTestFilterCmd(subcommand *cli.Subcommand, args []string) error {
 	if testFilterArgs.input != "" {
 		fsnames = []string{testFilterArgs.input}
 	} else {
-		out, err := zfs.ZFSList(ctx, []string{"name"})
+		out, err := zfs.ZFSList([]string{"name"})
 		if err != nil {
 			return fmt.Errorf("could not list ZFS filesystems: %s", err)
 		}
@@ -140,11 +137,10 @@ var testPlaceholder = &cli.Subcommand{
 func runTestPlaceholder(subcommand *cli.Subcommand, args []string) error {
 
 	var checkDPs []*zfs.DatasetPath
-	ctx := context.Background()
 
 	// all actions first
 	if testPlaceholderArgs.all {
-		out, err := zfs.ZFSList(ctx, []string{"name"})
+		out, err := zfs.ZFSList([]string{"name"})
 		if err != nil {
 			return errors.Wrap(err, "could not list ZFS filesystems")
 		}
@@ -168,7 +164,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(ctx, dp)
+		ph, err := zfs.ZFSGetFilesystemPlaceholderState(dp)
 		if err != nil {
 			return errors.Wrap(err, "cannot get placeholder state")
 		}

client/zfsabstractions.go

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

client/zfsabstractions_create.go

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

client/zfsabstractions_create_step_hold.go

@@ -1,60 +0,0 @@
-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(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")
-	}
-
-	ctx := context.Background()
-
-	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
-}

client/zfsabstractions_list.go

@@ -1,100 +0,0 @@
-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(sc *cli.Subcommand, args []string) error {
-	var err error
-	ctx := context.Background()
-
-	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(abstractions); 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
-	}
-
-}

client/zfsabstractions_release.go

@@ -1,145 +0,0 @@
-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(sc *cli.Subcommand, args []string) error {
-	var err error
-	ctx := context.Background()
-
-	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(sc *cli.Subcommand, args []string) error {
-
-	var err error
-	ctx := context.Background()
-
-	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
-	}
-}

config/config.go

@@ -620,7 +620,7 @@ func ParseConfigBytes(bytes []byte) (*Config, error) {
 
 var durationStringRegex *regexp.Regexp = regexp.MustCompile(`^\s*(\d+)\s*(s|m|h|d|w)\s*$`)
 
-func parsePositiveDuration(e string) (d time.Duration, err error) {
+func parsePostitiveDuration(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)

config/config_global_test.go

@@ -21,7 +21,7 @@ jobs:
     clients: {
       "10.0.0.1":"foo"
     }
-  root_fs: zroot/foo
+  root_fs: zoot/foo
 `
 	_, err := ParseConfigBytes([]byte(jobdef))
 	require.NoError(t, err)

config/config_test.go

@@ -42,7 +42,7 @@ func TestSampleConfigsAreParsedWithoutErrors(t *testing.T) {
 
 }
 
-// template must be a template/text template with a single '{{ . }}' as placeholder for val
+// template must be a template/text template with a single '{{ . }}' as placehodler for val
 //nolint[:deadcode,unused]
 func testValidConfigTemplate(t *testing.T, tmpl string, val string) *Config {
 	tmp, err := template.New("master").Parse(tmpl)

config/retentiongrid.go

@@ -64,7 +64,7 @@ func parseRetentionGridIntervalString(e string) (intervals []RetentionInterval,
 		return nil, fmt.Errorf("contains factor <= 0")
 	}
 
-	duration, err := parsePositiveDuration(comps[2])
+	duration, err := parsePostitiveDuration(comps[2])
 	if err != nil {
 		return nil, err
 	}

daemon/control.go

@@ -20,7 +20,6 @@ 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 {
@@ -65,7 +64,7 @@ func (j *controlJob) RegisterMetrics(registerer prometheus.Registerer) {
 		Namespace: "zrepl",
 		Subsystem: "control",
 		Name:      "request_finished",
-		Help:      "time it took a request to finish",
+		Help:      "time it took a request to finih",
 		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)
@@ -118,9 +117,7 @@ func (j *controlJob) Run(ctx context.Context) {
 	mux.Handle(ControlJobEndpointStatus,
 		// don't log requests to status endpoint, too spammy
 		jsonResponder{log, func() (interface{}, error) {
-			jobs := j.jobs.status()
-			globalZFS := zfscmd.GetReport()
-			s := Status{Jobs: jobs, Global: GlobalStatus{ZFSCmds: globalZFS}}
+			s := j.jobs.status()
 			return s, nil
 		}})
 
@@ -253,7 +250,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 marshal error")
+		j.log.WithError(producerErr).Error("control handler json marhsal error")
 		w.WriteHeader(http.StatusInternalServerError)
 		_, err := io.WriteString(w, encodeErr.Error())
 		logIoErr(err)

daemon/daemon.go

@@ -20,7 +20,6 @@ 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 {
@@ -82,9 +81,6 @@ func Run(conf *config.Config) error {
 		jobs.start(ctx, job, true)
 	}
 
-	// register global (=non job-local) metrics
-	zfscmd.RegisterMetrics(prometheus.DefaultRegisterer)
-
 	log.Info("starting daemon")
 
 	// start regular jobs
@@ -132,15 +128,6 @@ func (s *jobs) wait() <-chan struct{} {
 	return ch
 }
 
-type Status struct {
-	Jobs   map[string]*job.Status
-	Global GlobalStatus
-}
-
-type GlobalStatus struct {
-	ZFSCmds *zfscmd.Report
-}
-
 func (s *jobs) status() map[string]*job.Status {
 	s.m.RLock()
 	defer s.m.RUnlock()
@@ -220,7 +207,6 @@ func (s *jobs) start(ctx context.Context, j job.Job, internal bool) {
 
 	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

daemon/filters/fsmapfilter.go

@@ -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 exactly those paths that were not forbidden by the mapping.
+// The new filter allows excactly those paths that were not forbidden by the mapping.
 func (m DatasetMapFilter) InvertedFilter() (inv *DatasetMapFilter, err error) {
 
 	if m.filterMode {

daemon/filters/fsvfilter.go

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

daemon/job/active.go

@@ -147,7 +147,7 @@ func modePushFromConfig(g *config.Global, in *config.PushJob, jobID endpoint.Job
 
 	fsf, err := filters.DatasetMapFilterFromConfig(in.Filesystems)
 	if err != nil {
-		return nil, errors.Wrap(err, "cannot build filesystem filter")
+		return nil, errors.Wrap(err, "cannnot build filesystem filter")
 	}
 
 	m.senderConfig = &endpoint.SenderConfig{

daemon/job/passive.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/golang/protobuf/proto"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 
@@ -75,7 +76,7 @@ func modeSourceFromConfig(g *config.Global, in *config.SourceJob, jobID endpoint
 	m = &modeSource{}
 	fsf, err := filters.DatasetMapFilterFromConfig(in.Filesystems)
 	if err != nil {
-		return nil, errors.Wrap(err, "cannot build filesystem filter")
+		return nil, errors.Wrap(err, "cannnot build filesystem filter")
 	}
 	m.senderConfig = &endpoint.SenderConfig{
 		FSF:     fsf,
@@ -180,11 +181,39 @@ func (j *PassiveSide) Run(ctx context.Context) {
 	}
 
 	ctxInterceptor := func(handlerCtx context.Context) context.Context {
-		return logging.WithSubsystemLoggers(handlerCtx, log)
+		reqLog := log.WithField(logging.ReqIDField, rpc.MustGetRequestID(handlerCtx))
+		return logging.WithSubsystemLoggers(handlerCtx, reqLog)
+	}
+
+	// ctxInterceptor already injected request loggers (with request id)
+	pre := func(ctx context.Context, method string, req interface{}) {
+		l := endpoint.GetLogger(ctx).WithField("method", method).WithField("reqT", fmt.Sprintf("%T", req))
+		if p, ok := req.(proto.Message); ok {
+			l = l.WithField("req", proto.CompactTextString(p))
+		}
+		ci, ok := ctx.Value(endpoint.ClientIdentityKey).(string)
+		if !ok {
+			// like endpoint, we must assume that rpc.Server sets it
+			panic(method)
+		}
+		l = l.WithField("clientIdentity", ci)
+		l.Debug("incoming")
+	}
+	post := func(ctx context.Context, res interface{}, err error) {
+		l := endpoint.GetLogger(ctx).
+			WithField("responseT", fmt.Sprintf("%T", res)).
+			WithField("errT", fmt.Sprintf("%T", err))
+		if s, ok := res.(fmt.Stringer); ok {
+			l = l.WithField("response", s.String())
+		}
+		if err != nil {
+			l = l.WithError(err)
+		}
+		l.Debug("handler done")
 	}
 
 	rpcLoggers := rpc.GetLoggersOrPanic(ctx) // WithSubsystemLoggers above
-	server := rpc.NewServer(handler, rpcLoggers, ctxInterceptor)
+	server := rpc.NewServer(handler, rpcLoggers, ctxInterceptor, pre, post)
 
 	listener, err := j.listen()
 	if err != nil {

daemon/logging/build_logging.go

@@ -22,7 +22,6 @@ import (
 	"github.com/zrepl/zrepl/rpc/transportmux"
 	"github.com/zrepl/zrepl/tlsconf"
 	"github.com/zrepl/zrepl/transport"
-	"github.com/zrepl/zrepl/zfs/zfscmd"
 )
 
 func OutletsFromConfig(in config.LoggingOutletEnumList) (*logger.Outlets, error) {
@@ -71,7 +70,7 @@ type Subsystem string
 
 const (
 	SubsysReplication  Subsystem = "repl"
-	SubsysEndpoint     Subsystem = "endpoint"
+	SubsyEndpoint      Subsystem = "endpoint"
 	SubsysPruning      Subsystem = "pruning"
 	SubsysSnapshot     Subsystem = "snapshot"
 	SubsysHooks        Subsystem = "hook"
@@ -80,25 +79,28 @@ const (
 	SubsysRPC          Subsystem = "rpc"
 	SubsysRPCControl   Subsystem = "rpc.ctrl"
 	SubsysRPCData      Subsystem = "rpc.data"
-	SubsysZFSCmd       Subsystem = "zfs.cmd"
 )
 
 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, SubsysEndpoint))
+	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 = zfscmd.WithLogger(ctx, log.WithField(SubsysField, SubsysZFSCmd))
 	ctx = rpc.WithLoggers(ctx,
 		rpc.Loggers{
 			General: log.WithField(SubsysField, SubsysRPC),
 			Control: log.WithField(SubsysField, SubsysRPCControl),
 			Data:    log.WithField(SubsysField, SubsysRPCData),
 		},
+		// rpc.Loggers{
+		// 	General: logger.NewNullLogger(),
+		// 	Control: logger.NewNullLogger(),
+		// 	Data:    logger.NewNullLogger(),
+		// },
 	)
 	return ctx
 }

daemon/logging/logging_formatters.go

@@ -22,6 +22,7 @@ const (
 const (
 	JobField    string = "job"
 	SubsysField string = "subsystem"
+	ReqIDField  string = "reqid"
 )
 
 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, ReqIDField}
 	prefixed := make(map[string]bool, len(prefixFields)+2)
 	for _, field := range prefixFields {
 		val, ok := e.Fields[field]
@@ -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 unsupported value type Go type")
+			return errors.Wrap(err, "cannot encode unsuuported value type Go type")
 		}
 		return nil
 	}

daemon/logging/logging_outlets.go

@@ -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 previous is in io.Copy()
+	entryChan := make(chan *bytes.Buffer, 1) // allow one message in flight while previos is in io.Copy()
 
 	o := &TCPOutlet{
 		formatter: formatter,

daemon/pruner/pruner.go

@@ -18,13 +18,13 @@ import (
 	"github.com/zrepl/zrepl/util/envconst"
 )
 
-// Try to keep it compatible with github.com/zrepl/zrepl/endpoint.Endpoint
+// Try to keep it compatible with gitub.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 github.com/zrepl/zrepl/endpoint.Endpoint
+// Try to keep it compatible with gitub.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)

daemon/snapper/snapper.go

@@ -277,7 +277,7 @@ func snapshot(a args, u updater) state {
 
 		jobCallback := hooks.NewCallbackHookForFilesystem("snapshot", fs, func(_ context.Context) (err error) {
 			l.Debug("create snapshot")
-			err = zfs.ZFSSnapshot(a.ctx, fs, snapname, false) // TODO propagate context to ZFSSnapshot
+			err = zfs.ZFSSnapshot(fs, snapname, false) // TODO propagagte context to ZFSSnapshot
 			if err != nil {
 				l.WithError(err).Error("cannot create snapshot")
 			}
@@ -485,10 +485,7 @@ 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) {
 
-	fsvs, err := zfs.ZFSListFilesystemVersions(d, zfs.ListFilesystemVersionsOptions{
-		Types:           zfs.Snapshots,
-		ShortnamePrefix: prefix,
-	})
+	fsvs, err := zfs.ZFSListFilesystemVersions(d, filters.NewTypedPrefixFilter(prefix, zfs.Snapshot))
 	if err != nil {
 		return time.Time{}, errors.Wrap(err, "list filesystem versions")
 	}

docs/changelog.rst

@@ -62,9 +62,7 @@ Actual changelog:
 * |bugfix| |docs| snapshotting: clarify sync-up behavior and warn about filesystems
   that will not be snapshotted until the sync-up phase is over
 * |docs| Document new replication features in the :ref:`config overview <overview-how-replication-works>` and :repomasterlink:`replication/design.md`.
-* |feature| documented subcommand to generate ``bash`` and ``zsh`` completions
 * **[MAINTAINER NOTICE]** New platform tests in this version, please make sure you run them for your distro!
-* **[MAINTAINER NOTICE]** Please add the shell completions to the zrepl packages.
 
 0.2.1
 -----
@@ -107,7 +105,7 @@ Actual changelog:
    |  You can support maintenance and feature development through one of the following services:
    |  |Donate via Patreon| |Donate via Liberapay| |Donate via PayPal|
    |  Note that PayPal processing fees are relatively high for small donations.
-   |  For SEPA wire transfer and **commercial support**, please `contact Christian directly <https://cschwarz.com>`_.
+   |  For SEPA wire transfer and **commerical support**, please `contact Christian directly <https://cschwarz.com>`_.
 
 
 0.1.1

docs/configuration/logging.rst

@@ -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 before messages are dropped.
+    Note that TCP buffering in the kernel must first run full becfore 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>` ).

docs/configuration/monitoring.rst

@@ -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 Prometheus monitoring job appears in the ``zrepl control`` job list and may be specified **at most once**.
+The Prometheues 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`.

docs/configuration/overview.rst

@@ -26,9 +26,9 @@ Config File Structure
      type: push
    - ...
 
-zrepl is configured using a single YAML configuration file with two main sections: ``global`` and ``jobs``.
+zrepl is confgured 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 going to explain now.
+The ``jobs`` section is a list of jobs which we are goind to explain now.
 
 .. _job-overview:
 
@@ -41,7 +41,7 @@ Jobs are identified by their ``name``, both in log files and the ``zrepl status`
 
 Replication always happens between a pair of jobs: one is the **active side**, and one the **passive side**.
 The active side connects to the passive side using a :ref:`transport <transport>` and starts executing the replication logic.
-The passive side responds to requests from the active side after checking its permissions.
+The passive side responds to requests from the active side after checking its persmissions.
 
 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.
@@ -120,7 +120,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}``.
-  * Acquire send-side *step-holds* on the step's `from` and `to` snapshots.
+  * Aquire 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).
@@ -141,7 +141,7 @@ The ``zrepl holds list`` provides a listing of all bookmarks and holds managed b
 .. _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 hierarchy without replicating every filesystem at every intermediary dataset path component.
+Placeholders allow the receiving side to mirror the sender's ZFS dataset hierachy 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.
@@ -181,7 +181,7 @@ 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 assumed 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 asssumed 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
@@ -198,5 +198,5 @@ Multiple pull jobs pulling from the same source have potential for race conditio
 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 possible 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 possibe to just manually assert that the prune rules of all pull jobs are disjoint to avoid replication-prune and prune-prune races.
 

docs/configuration/prune.rst

@@ -154,7 +154,7 @@ Policy ``regex``
            negate: true
            regex: "^zrepl_.*"
 
-``regex`` keeps all snapshots whose names are matched by the regular expression in ``regex``.
+``regex`` keeps all snapshots whose names are matched by the regular expressionin ``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.
 

docs/configuration/sendrecvoptions.rst

@@ -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 specifically, if ``encryption=true``, zrepl
+More specificially, 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

docs/installation.rst

@@ -101,9 +101,9 @@ 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 . && \
+    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)" \
@@ -127,7 +127,7 @@ 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``.
+    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?

docs/supporters.rst

@@ -6,7 +6,7 @@
 
 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 **commercial support**, please `contact Christian directly <https://cschwarz.com>`_.
+For SEPA wire transfer and **commerical support**, please `contact Christian directly <https://cschwarz.com>`_.
 
 **Thanks for your support!**
 

docs/usage.rst

@@ -38,8 +38,6 @@ CLI Overview
     * - ``zrepl migrate``
       - | perform on-disk state / ZFS property migrations
         | (see :ref:`changelog <changelog>` for details)
-    * - ``zrepl zfs-abstractions``
-      - 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:
 
@@ -50,7 +48,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 installing from a package, the package maintainer should have provided an init script / systemd.service file.
+When installating 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``.
@@ -75,6 +73,6 @@ The daemon exits as soon as all jobs have reported shut down.
 Systemd Unit File
 ~~~~~~~~~~~~~~~~~
 
-A systemd service definition template is available in :repomasterlink:`dist/systemd`.
+A systemd service defintion 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`.

endpoint/context.go

@@ -19,6 +19,10 @@ func WithLogger(ctx context.Context, log Logger) context.Context {
 	return context.WithValue(ctx, contextKeyLogger, log)
 }
 
+func GetLogger(ctx context.Context) Logger {
+	return getLogger(ctx)
+}
+
 func getLogger(ctx context.Context) Logger {
 	if l, ok := ctx.Value(contextKeyLogger).(Logger); ok {
 		return l

endpoint/endpoint.go

@@ -96,7 +96,7 @@ func (s *Sender) ListFilesystemVersions(ctx context.Context, r *pdu.ListFilesyst
 	if err != nil {
 		return nil, err
 	}
-	fsvs, err := zfs.ZFSListFilesystemVersions(lp, zfs.ListFilesystemVersionsOptions{})
+	fsvs, err := zfs.ZFSListFilesystemVersions(lp, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -110,116 +110,34 @@ 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
 
-	fsp, err := p.filterCheckFS(r.GetFilesystem())
-	if err != nil {
-		return nil, err
-	}
-	fs := fsp.ToString()
-
-	log := getLogger(ctx).WithField("fs", fs).WithField("hinted_most_recent", fmt.Sprintf("%#v", r.GetSenderVersion()))
-
-	log.WithField("full_hint", r).Debug("full hint")
-
-	if r.GetSenderVersion() == nil {
-		// no common ancestor found, likely due to failed prior replication attempt
-		// => release stale step holds to prevent them from accumulating
-		//    (they can accumulate on initial replication because each inital replication step might hold a different `to`)
-		// => replication cursors cannot accumulate because we always _move_ the replication cursor
-		log.Debug("releasing all step holds on the filesystem")
-		TryReleaseStepStaleFS(ctx, fs, p.jobId)
-		return &pdu.HintMostRecentCommonAncestorRes{}, nil
-	}
-
-	// we were hinted a specific common ancestor
-
-	mostRecentVersion, err := sendArgsFromPDUAndValidateExistsAndGetVersion(ctx, fs, r.GetSenderVersion())
+	fs := r.GetFilesystem()
+	mostRecent, err := sendArgsFromPDUAndValidateExists(ctx, fs, r.GetSenderVersion())
 	if err != nil {
 		msg := "HintMostRecentCommonAncestor rpc with nonexistent most recent version"
-		log.Warn(msg)
+		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
-	destroyedCursors, err := MoveReplicationCursor(ctx, fs, mostRecentVersion, p.jobId)
+	_, err = MoveReplicationCursor(ctx, fs, mostRecent, p.jobId)
 	if err == zfs.ErrBookmarkCloningNotSupported {
-		log.Debug("not creating replication cursor from bookmark because ZFS does not support it")
+		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")
 	}
 
-	// take care of stale step holds
-	log.WithField("step-holds-cleanup-mode", senderHintMostRecentCommonAncestorStepCleanupMode).
-		Debug("taking care of possibly stale step holds")
-	doStepCleanup := false
-	var stepCleanupSince *CreateTXGRangeBound
-	switch senderHintMostRecentCommonAncestorStepCleanupMode {
-	case StepCleanupNoCleanup:
-		doStepCleanup = false
-	case StepCleanupRangeSinceUnbounded:
-		doStepCleanup = true
-		stepCleanupSince = nil
-	case StepCleanupRangeSinceReplicationCursor:
-		doStepCleanup = true
-		// Use the destroyed replication cursors as indicator how far the previous replication got.
-		// To be precise: We limit the amount of visisted snapshots to exactly those snapshots
-		// created since the last successful replication cursor movement (i.e. last successful replication step)
-		//
-		// If we crash now, we'll leak the step we are about to release, but the performance gain
-		// of limiting the amount of snapshots we visit makes up for that.
-		// Users have the `zrepl holds release-stale` command to cleanup leaked step holds.
-		for _, destroyed := range destroyedCursors {
-			if stepCleanupSince == nil {
-				stepCleanupSince = &CreateTXGRangeBound{
-					CreateTXG: destroyed.GetCreateTXG(),
-					Inclusive: &zfs.NilBool{B: true},
-				}
-			} else if destroyed.GetCreateTXG() < stepCleanupSince.CreateTXG {
-				stepCleanupSince.CreateTXG = destroyed.GetCreateTXG()
-			}
-		}
-	default:
-		panic(senderHintMostRecentCommonAncestorStepCleanupMode)
-	}
-	if !doStepCleanup {
-		log.Info("skipping cleanup of prior invocations' step holds due to environment variable setting")
-	} else {
-		if err := ReleaseStepCummulativeInclusive(ctx, fs, stepCleanupSince, mostRecentVersion, p.jobId); err != nil {
-			return nil, errors.Wrap(err, "cannot cleanup prior invocation's step holds and bookmarks")
-		} else {
-			log.Info("step hold cleanup done")
-		}
+	// 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
 }
 
-type HintMostRecentCommonAncestorStepCleanupMode struct{ string }
-
-var (
-	StepCleanupRangeSinceReplicationCursor = HintMostRecentCommonAncestorStepCleanupMode{"range-since-replication-cursor"}
-	StepCleanupRangeSinceUnbounded         = HintMostRecentCommonAncestorStepCleanupMode{"range-since-unbounded"}
-	StepCleanupNoCleanup                   = HintMostRecentCommonAncestorStepCleanupMode{"no-cleanup"}
-)
-
-func (m HintMostRecentCommonAncestorStepCleanupMode) String() string { return string(m.string) }
-func (m *HintMostRecentCommonAncestorStepCleanupMode) Set(s string) error {
-	switch s {
-	case StepCleanupRangeSinceReplicationCursor.String():
-		*m = StepCleanupRangeSinceReplicationCursor
-	case StepCleanupRangeSinceUnbounded.String():
-		*m = StepCleanupRangeSinceUnbounded
-	case StepCleanupNoCleanup.String():
-		*m = StepCleanupNoCleanup
-	default:
-		return fmt.Errorf("unknown step cleanup mode %q", s)
-	}
-	return nil
-}
-
-var senderHintMostRecentCommonAncestorStepCleanupMode = *envconst.Var("ZREPL_ENDPOINT_SENDER_HINT_MOST_RECENT_STEP_HOLD_CLEANUP_MODE", &StepCleanupRangeSinceReplicationCursor).(*HintMostRecentCommonAncestorStepCleanupMode)
-
 var maxConcurrentZFSSendSemaphore = semaphore.New(envconst.Int64("ZREPL_ENDPOINT_MAX_CONCURRENT_SEND", 10))
 
 func uncheckedSendArgsFromPDU(fsv *pdu.FilesystemVersion) *zfs.ZFSSendArgVersion {
@@ -229,19 +147,19 @@ func uncheckedSendArgsFromPDU(fsv *pdu.FilesystemVersion) *zfs.ZFSSendArgVersion
 	return &zfs.ZFSSendArgVersion{RelName: fsv.GetRelName(), GUID: fsv.Guid}
 }
 
-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")
+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")
 	}
-	version, err := sendArgs.ValidateExistsAndGetVersion(ctx, fs)
-	if err != nil {
-		return v, err
+	if err := v.ValidateExists(ctx, fs); err != nil {
+		return nil, err
 	}
-	return version, nil
+	return v, nil
 }
 
 func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
+	getLogger(ctx).WithField("req", r.String()).Debug("incoming Send request")
 
 	_, err := s.filterCheckFS(r.Filesystem)
 	if err != nil {
@@ -253,7 +171,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 encrypted sends allowed (send -w + encryption!= off), but unencrypted send requested")
+			return nil, nil, errors.New("only encrytped sends allowed (send -w + encryption!= off), but unencrytped send requested")
 		}
 		// fallthrough outer
 	case pdu.Tri_True:
@@ -265,7 +183,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)
 	}
 
-	sendArgsUnvalidated := zfs.ZFSSendArgsUnvalidated{
+	sendArgs := zfs.ZFSSendArgs{
 		FS:          r.Filesystem,
 		From:        uncheckedSendArgsFromPDU(r.GetFrom()), // validated by zfs.ZFSSendDry / zfs.ZFSSend
 		To:          uncheckedSendArgsFromPDU(r.GetTo()),   // validated by zfs.ZFSSendDry / zfs.ZFSSend
@@ -273,11 +191,6 @@ 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
@@ -294,7 +207,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 involves shelling out, it's actually a little expensive)
+	// (because validation invovles 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
@@ -312,7 +225,7 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St
 	// 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.FromVersion, s.jobId)
+		_, 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
@@ -323,18 +236,18 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St
 
 	// make sure `From` doesn't go away in order to make this step resumable
 	if sendArgs.From != nil {
-		_, err := HoldStep(ctx, sendArgs.FS, *sendArgs.FromVersion, s.jobId)
+		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.Wrapf(err, "cannot hold `from` version %q before starting send", *sendArgs.FromVersion)
+			return nil, nil, errors.Wrap(err, "cannot create step bookmark")
 		}
 	}
 	// make sure `To` doesn't go away in order to make this step resumable
-	_, err = HoldStep(ctx, sendArgs.FS, sendArgs.ToVersion, s.jobId)
+	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.ToVersion)
+		return nil, nil, errors.Wrapf(err, "cannot hold `to` version %q before starting send", sendArgs.To.RelName)
 	}
 
 	// step holds & replication cursor released / moved forward in s.SendCompleted => s.moveCursorAndReleaseSendHolds
@@ -347,32 +260,28 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St
 }
 
 func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
-
+	getLogger(ctx).WithField("req", r.String()).Debug("incoming SendCompleted request")
 	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()
+	fs := orig.GetFilesystem()
 
-	var from *zfs.FilesystemVersion
+	var err error
+	var from *zfs.ZFSSendArgVersion
 	if orig.GetFrom() != nil {
-		f, err := sendArgsFromPDUAndValidateExistsAndGetVersion(ctx, fs, orig.GetFrom()) // no shadow
+		from, err = sendArgsFromPDUAndValidateExists(ctx, fs, orig.GetFrom()) // no shadow
 		if err != nil {
 			return nil, errors.Wrap(err, "validate `from` exists")
 		}
-		from = &f
 	}
-	to, err := sendArgsFromPDUAndValidateExistsAndGetVersion(ctx, fs, orig.GetTo())
+	to, err := sendArgsFromPDUAndValidateExists(ctx, fs, orig.GetTo())
 	if err != nil {
 		return nil, errors.Wrap(err, "validate `to` exists")
 	}
 
-	log := getLogger(ctx).WithField("to_guid", to.Guid).
+	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 = log.WithField("from", from.RelName).WithField("from_guid", from.GUID)
 	}
 
 	log.Debug("move replication cursor to most recent common version")
@@ -413,14 +322,18 @@ func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*p
 			return
 		}
 		log.Debug("release step-hold of or step-bookmark on `from`")
-		err := ReleaseStep(ctx, fs, *from, p.jobId)
+		err := ReleaseStep(ctx, fs, from, p.jobId)
 		if err != nil {
 			if dne, ok := err.(*zfs.DatasetDoesNotExist); ok {
 				// If bookmark cloning is not supported, `from` might be the old replication cursor
 				// and thus have already been destroyed by MoveReplicationCursor above
 				// In that case, nonexistence of `from` is not an error, otherwise it is.
-				for _, c := range destroyedCursors {
-					if c.GetFullPath() == dne.Path {
+				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
 					}
@@ -599,8 +512,8 @@ func (s *Receiver) ListFilesystems(ctx context.Context, req *pdu.ListFilesystemR
 	// present filesystem without the root_fs prefix
 	fss := make([]*pdu.Filesystem, 0, len(filtered))
 	for _, a := range filtered {
-		l := getLogger(ctx).WithField("fs", a)
-		ph, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, a)
+		l := getLogger(ctx).WithField("fs", a.ToString())
+		ph, err := zfs.ZFSGetFilesystemPlaceholderState(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)
@@ -646,9 +559,8 @@ func (s *Receiver) ListFilesystemVersions(ctx context.Context, req *pdu.ListFile
 	if err != nil {
 		return nil, err
 	}
-	// TODO share following code with sender
 
-	fsvs, err := zfs.ZFSListFilesystemVersions(lp, zfs.ListFilesystemVersionsOptions{})
+	fsvs, err := zfs.ZFSListFilesystemVersions(lp, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -687,7 +599,7 @@ func (s *Receiver) Send(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, zf
 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) {
-	getLogger(ctx).Debug("incoming Receive")
+	getLogger(ctx).WithField("rr", req.String()).Debug("incoming Receive")
 	defer receive.Close()
 
 	root := s.clientRootFromCtx(ctx)
@@ -700,6 +612,9 @@ 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")
 	}
@@ -712,9 +627,9 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 	//       ZFS dataset hierarchy subtrees.
 	var visitErr error
 	func() {
-		getLogger(ctx).Debug("begin acquire recvParentCreationMtx")
+		getLogger(ctx).Debug("begin aquire recvParentCreationMtx")
 		defer s.recvParentCreationMtx.Lock().Unlock()
-		getLogger(ctx).Debug("end acquire recvParentCreationMtx")
+		getLogger(ctx).Debug("end aquire recvParentCreationMtx")
 		defer getLogger(ctx).Debug("release recvParentCreationMtx")
 
 		f := zfs.NewDatasetPathForest()
@@ -724,7 +639,7 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 			if v.Path.Equal(lp) {
 				return false
 			}
-			ph, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, v.Path)
+			ph, err := zfs.ZFSGetFilesystemPlaceholderState(v.Path)
 			getLogger(ctx).
 				WithField("fs", v.Path.ToString()).
 				WithField("placeholder_state", fmt.Sprintf("%#v", ph)).
@@ -748,7 +663,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(ctx, v.Path)
+				err := zfs.ZFSCreatePlaceholderFilesystem(v.Path)
 				if err != nil {
 					l.WithError(err).Error("cannot create placeholder filesystem")
 					visitErr = err
@@ -768,19 +683,19 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 	// determine whether we need to rollback the filesystem / change its placeholder state
 	var clearPlaceholderProperty bool
 	var recvOpts zfs.RecvOptions
-	ph, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, lp)
+	ph, err := zfs.ZFSGetFilesystemPlaceholderState(lp)
 	if err == nil && ph.FSExists && ph.IsPlaceholder {
 		recvOpts.RollbackAndForceRecv = true
 		clearPlaceholderProperty = true
 	}
 	if clearPlaceholderProperty {
-		if err := zfs.ZFSSetPlaceholder(ctx, lp, false); err != nil {
+		if err := zfs.ZFSSetPlaceholder(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(ctx, lp.ToString()); err != nil {
+		if err := zfs.ZFSRecvClearResumeToken(lp.ToString()); err != nil {
 			return nil, errors.Wrap(err, "cannot clear resume token")
 		}
 	}
@@ -812,8 +727,7 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 	}
 
 	// validate that we actually received what the sender claimed
-	toRecvd, err := to.ValidateExistsAndGetVersion(ctx, lp.ToString())
-	if err != nil {
+	if err := to.ValidateExists(ctx, lp.ToString()); 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")
@@ -822,7 +736,7 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 
 	if s.conf.UpdateLastReceivedHold {
 		getLogger(ctx).Debug("move last-received-hold")
-		if err := MoveLastReceivedHold(ctx, lp.ToString(), toRecvd, s.conf.JobID); err != nil {
+		if err := MoveLastReceivedHold(ctx, lp.ToString(), *to, s.conf.JobID); err != nil {
 			return nil, errors.Wrap(err, "cannot move last-received-hold")
 		}
 	}

endpoint/endpoint_zfs.go

@@ -0,0 +1,503 @@
+package endpoint
+
+import (
+	"context"
+	"fmt"
+	"regexp"
+	"sort"
+
+	"github.com/kr/pretty"
+	"github.com/pkg/errors"
+
+	"github.com/zrepl/zrepl/zfs"
+)
+
+var stepHoldTagRE = regexp.MustCompile("^zrepl_STEP_J_(.+)")
+
+func StepHoldTag(jobid JobID) (string, error) {
+	return stepHoldTagImpl(jobid.String())
+}
+
+func stepHoldTagImpl(jobid string) (string, error) {
+	t := fmt.Sprintf("zrepl_STEP_J_%s", jobid)
+	if err := zfs.ValidHoldTag(t); err != nil {
+		return "", err
+	}
+	return t, nil
+}
+
+// err != nil always means that the bookmark is not a step bookmark
+func ParseStepHoldTag(tag string) (JobID, error) {
+	match := stepHoldTagRE.FindStringSubmatch(tag)
+	if match == nil {
+		return JobID{}, fmt.Errorf("parse hold tag: match regex %q", stepHoldTagRE)
+	}
+	jobID, err := MakeJobID(match[1])
+	if err != nil {
+		return JobID{}, errors.Wrap(err, "parse hold tag: invalid job id field")
+	}
+	return jobID, nil
+}
+
+const stepBookmarkNamePrefix = "zrepl_STEP"
+
+// v must be validated by caller
+func StepBookmarkName(fs string, guid uint64, id JobID) (string, error) {
+	return stepBookmarkNameImpl(fs, guid, id.String())
+}
+
+func stepBookmarkNameImpl(fs string, guid uint64, jobid string) (string, error) {
+	return makeJobAndGuidBookmarkName(stepBookmarkNamePrefix, fs, guid, jobid)
+}
+
+// name is the full bookmark name, including dataset path
+//
+// err != nil always means that the bookmark is not a step bookmark
+func ParseStepBookmarkName(fullname string) (guid uint64, jobID JobID, err error) {
+	guid, jobID, err = parseJobAndGuidBookmarkName(fullname, stepBookmarkNamePrefix)
+	if err != nil {
+		err = errors.Wrap(err, "parse step bookmark name") // no shadow!
+	}
+	return guid, jobID, err
+}
+
+const replicationCursorBookmarkNamePrefix = "zrepl_CURSOR"
+
+func ReplicationCursorBookmarkName(fs string, guid uint64, id JobID) (string, error) {
+	return replicationCursorBookmarkNameImpl(fs, guid, id.String())
+}
+
+func replicationCursorBookmarkNameImpl(fs string, guid uint64, jobid string) (string, error) {
+	return makeJobAndGuidBookmarkName(replicationCursorBookmarkNamePrefix, fs, guid, jobid)
+}
+
+var ErrV1ReplicationCursor = fmt.Errorf("bookmark name is a v1-replication cursor")
+
+//err != nil always means that the bookmark is not a valid replication bookmark
+//
+// Returns ErrV1ReplicationCursor as error if the bookmark is a v1 replication cursor
+func ParseReplicationCursorBookmarkName(fullname string) (uint64, JobID, error) {
+
+	// check for legacy cursors
+	{
+		if err := zfs.EntityNamecheck(fullname, zfs.EntityTypeBookmark); err != nil {
+			return 0, JobID{}, errors.Wrap(err, "parse replication cursor bookmark name")
+		}
+		_, _, name, err := zfs.DecomposeVersionString(fullname)
+		if err != nil {
+			return 0, JobID{}, errors.Wrap(err, "parse replication cursor bookmark name: decompose version string")
+		}
+		const V1ReplicationCursorBookmarkName = "zrepl_replication_cursor"
+		if name == V1ReplicationCursorBookmarkName {
+			return 0, JobID{}, ErrV1ReplicationCursor
+		}
+	}
+
+	guid, jobID, err := parseJobAndGuidBookmarkName(fullname, replicationCursorBookmarkNamePrefix)
+	if err != nil {
+		err = errors.Wrap(err, "parse replication cursor bookmark name") // no shadow
+	}
+	return guid, jobID, err
+}
+
+// may return nil for both values, indicating there is no cursor
+func GetMostRecentReplicationCursorOfJob(ctx context.Context, fs string, jobID JobID) (*zfs.FilesystemVersion, error) {
+	fsp, err := zfs.NewDatasetPath(fs)
+	if err != nil {
+		return nil, err
+	}
+	candidates, err := GetReplicationCursors(ctx, fsp, jobID)
+	if err != nil || len(candidates) == 0 {
+		return nil, err
+	}
+
+	sort.Slice(candidates, func(i, j int) bool {
+		return candidates[i].CreateTXG < candidates[j].CreateTXG
+	})
+
+	mostRecent := candidates[len(candidates)-1]
+	return &mostRecent, nil
+}
+
+func GetReplicationCursors(ctx context.Context, fs *zfs.DatasetPath, jobID JobID) ([]zfs.FilesystemVersion, error) {
+
+	listOut := &ListHoldsAndBookmarksOutput{}
+	if err := listZFSHoldsAndBookmarksImplFS(ctx, listOut, fs); err != nil {
+		return nil, errors.Wrap(err, "get replication cursor: list bookmarks and holds")
+	}
+
+	if len(listOut.V1ReplicationCursors) > 0 {
+		getLogger(ctx).WithField("bookmark", pretty.Sprint(listOut.V1ReplicationCursors)).
+			Warn("found v1-replication cursor bookmarks, consider running migration 'replication-cursor:v1-v2' after successful replication with this zrepl version")
+	}
+
+	candidates := make([]zfs.FilesystemVersion, 0)
+	for _, v := range listOut.ReplicationCursorBookmarks {
+		zv := zfs.ZFSSendArgVersion{
+			RelName: "#" + v.Name,
+			GUID:    v.Guid,
+		}
+		if err := zv.ValidateExists(ctx, v.FS); err != nil {
+			getLogger(ctx).WithError(err).WithField("bookmark", zv.FullPath(v.FS)).
+				Error("found invalid replication cursor bookmark")
+			continue
+		}
+		candidates = append(candidates, v.v)
+	}
+
+	return candidates, nil
+}
+
+// `target` is validated before replication cursor is set. if validation fails, the cursor is not moved.
+//
+// returns ErrBookmarkCloningNotSupported if version is a bookmark and bookmarking bookmarks is not supported by ZFS
+func MoveReplicationCursor(ctx context.Context, fs string, target *zfs.ZFSSendArgVersion, jobID JobID) (destroyedCursors []zfs.FilesystemVersion, err error) {
+
+	if !target.IsSnapshot() {
+		return nil, zfs.ErrBookmarkCloningNotSupported
+	}
+
+	snapProps, err := target.ValidateExistsAndGetCheckedProps(ctx, fs)
+	if err != nil {
+		return nil, errors.Wrapf(err, "invalid replication cursor target %q (guid=%v)", target.RelName, target.GUID)
+	}
+
+	bookmarkname, err := ReplicationCursorBookmarkName(fs, snapProps.Guid, jobID)
+	if err != nil {
+		return nil, errors.Wrap(err, "determine replication cursor name")
+	}
+
+	// idempotently create bookmark (guid is encoded in it, hence we'll most likely add a new one
+	// cleanup the old one afterwards
+
+	err = zfs.ZFSBookmark(fs, *target, bookmarkname)
+	if err != nil {
+		if err == zfs.ErrBookmarkCloningNotSupported {
+			return nil, err // TODO go1.13 use wrapping
+		}
+		return nil, errors.Wrapf(err, "cannot create bookmark")
+	}
+
+	destroyedCursors, err = DestroyObsoleteReplicationCursors(ctx, fs, target, jobID)
+	if err != nil {
+		return nil, errors.Wrap(err, "destroy obsolete replication cursors")
+	}
+
+	return destroyedCursors, nil
+}
+
+func DestroyObsoleteReplicationCursors(ctx context.Context, fs string, target *zfs.ZFSSendArgVersion, jobID JobID) (destroyed []zfs.FilesystemVersion, err error) {
+	return destroyBookmarksOlderThan(ctx, fs, target, jobID, func(shortname string) (accept bool) {
+		_, parsedID, err := ParseReplicationCursorBookmarkName(fs + "#" + shortname)
+		return err == nil && parsedID == jobID
+	})
+}
+
+// idempotently hold / step-bookmark `version`
+//
+// returns ErrBookmarkCloningNotSupported if version is a bookmark and bookmarking bookmarks is not supported by ZFS
+func HoldStep(ctx context.Context, fs string, v *zfs.ZFSSendArgVersion, jobID JobID) error {
+	if err := v.ValidateExists(ctx, fs); err != nil {
+		return err
+	}
+	if v.IsSnapshot() {
+
+		tag, err := StepHoldTag(jobID)
+		if err != nil {
+			return errors.Wrap(err, "step hold tag")
+		}
+
+		if err := zfs.ZFSHold(ctx, fs, *v, tag); err != nil {
+			return errors.Wrap(err, "step hold: zfs")
+		}
+
+		return nil
+	}
+
+	v.MustBeBookmark()
+
+	bmname, err := StepBookmarkName(fs, v.GUID, jobID)
+	if err != nil {
+		return errors.Wrap(err, "create step bookmark: determine bookmark name")
+	}
+	// idempotently create bookmark
+	err = zfs.ZFSBookmark(fs, *v, bmname)
+	if err != nil {
+		if err == zfs.ErrBookmarkCloningNotSupported {
+			// TODO we could actually try to find a local snapshot that has the requested GUID
+			// 		however, the replication algorithm prefers snapshots anyways, so this quest
+			// 		is most likely not going to be successful. Also, there's the possibility that
+			//      the caller might want to filter what snapshots are eligibile, and this would
+			//      complicate things even further.
+			return err // TODO go1.13 use wrapping
+		}
+		return errors.Wrap(err, "create step bookmark: zfs")
+	}
+	return nil
+}
+
+// idempotently release the step-hold on v if v is a snapshot
+// or idempotently destroy the step-bookmark  of v if v is a bookmark
+//
+// note that this operation leaves v itself untouched, unless v is the step-bookmark itself, in which case v is destroyed
+//
+// returns an instance of *zfs.DatasetDoesNotExist if `v` does not exist
+func ReleaseStep(ctx context.Context, fs string, v *zfs.ZFSSendArgVersion, jobID JobID) error {
+
+	if err := v.ValidateExists(ctx, fs); err != nil {
+		return err
+	}
+
+	if v.IsSnapshot() {
+		tag, err := StepHoldTag(jobID)
+		if err != nil {
+			return errors.Wrap(err, "step release tag")
+		}
+
+		if err := zfs.ZFSRelease(ctx, tag, v.FullPath(fs)); err != nil {
+			return errors.Wrap(err, "step release: zfs")
+		}
+
+		return nil
+	}
+
+	v.MustBeBookmark()
+
+	bmname, err := StepBookmarkName(fs, v.GUID, jobID)
+	if err != nil {
+		return errors.Wrap(err, "step release: determine bookmark name")
+	}
+	// idempotently destroy bookmark
+
+	if err := zfs.ZFSDestroyIdempotent(bmname); err != nil {
+		return errors.Wrap(err, "step release: bookmark destroy: zfs")
+	}
+
+	return nil
+}
+
+// release {step holds, step bookmarks} earlier and including `mostRecent`
+func ReleaseStepAll(ctx context.Context, fs string, mostRecent *zfs.ZFSSendArgVersion, jobID JobID) error {
+
+	if err := mostRecent.ValidateInMemory(fs); err != nil {
+		return err
+	}
+
+	tag, err := StepHoldTag(jobID)
+	if err != nil {
+		return errors.Wrap(err, "step release all: tag")
+	}
+
+	err = zfs.ZFSReleaseAllOlderAndIncludingGUID(ctx, fs, mostRecent.GUID, tag)
+	if err != nil {
+		return errors.Wrapf(err, "step release all: release holds older and including %q", mostRecent.FullPath(fs))
+	}
+
+	_, err = destroyBookmarksOlderThan(ctx, fs, mostRecent, jobID, func(shortname string) bool {
+		_, parsedId, parseErr := ParseStepBookmarkName(fs + "#" + shortname)
+		return parseErr == nil && parsedId == jobID
+	})
+	if err != nil {
+		return errors.Wrapf(err, "step release all: destroy bookmarks older than %q", mostRecent.FullPath(fs))
+	}
+
+	return nil
+}
+
+var lastReceivedHoldTagRE = regexp.MustCompile("^zrepl_last_received_J_(.+)$")
+
+// err != nil always means that the bookmark is not a step bookmark
+func ParseLastReceivedHoldTag(tag string) (JobID, error) {
+	match := lastReceivedHoldTagRE.FindStringSubmatch(tag)
+	if match == nil {
+		return JobID{}, errors.Errorf("parse last-received-hold tag: does not match regex %s", lastReceivedHoldTagRE.String())
+	}
+	jobId, err := MakeJobID(match[1])
+	if err != nil {
+		return JobID{}, errors.Wrap(err, "parse last-received-hold tag: invalid job id field")
+	}
+	return jobId, nil
+}
+
+func LastReceivedHoldTag(jobID JobID) (string, error) {
+	return lastReceivedHoldImpl(jobID.String())
+}
+
+func lastReceivedHoldImpl(jobid string) (string, error) {
+	tag := fmt.Sprintf("zrepl_last_received_J_%s", jobid)
+	if err := zfs.ValidHoldTag(tag); err != nil {
+		return "", err
+	}
+	return tag, nil
+}
+
+func MoveLastReceivedHold(ctx context.Context, fs string, to zfs.ZFSSendArgVersion, jobID JobID) error {
+	if err := to.ValidateExists(ctx, fs); err != nil {
+		return err
+	}
+	if err := zfs.EntityNamecheck(to.FullPath(fs), zfs.EntityTypeSnapshot); err != nil {
+		return err
+	}
+
+	tag, err := LastReceivedHoldTag(jobID)
+	if err != nil {
+		return errors.Wrap(err, "last-received-hold: hold tag")
+	}
+
+	// we never want to be without a hold
+	// => hold new one before releasing old hold
+
+	err = zfs.ZFSHold(ctx, fs, to, tag)
+	if err != nil {
+		return errors.Wrap(err, "last-received-hold: hold newly received")
+	}
+
+	err = zfs.ZFSReleaseAllOlderThanGUID(ctx, fs, to.GUID, tag)
+	if err != nil {
+		return errors.Wrap(err, "last-received-hold: release older holds")
+	}
+
+	return nil
+}
+
+type ListHoldsAndBookmarksOutputBookmarkV1ReplicationCursor struct {
+	FS   string
+	Name string
+}
+
+type ListHoldsAndBookmarksOutput struct {
+	StepBookmarks []*ListHoldsAndBookmarksOutputBookmark
+	StepHolds     []*ListHoldsAndBookmarksOutputHold
+
+	ReplicationCursorBookmarks []*ListHoldsAndBookmarksOutputBookmark
+	V1ReplicationCursors       []*ListHoldsAndBookmarksOutputBookmarkV1ReplicationCursor
+	LastReceivedHolds          []*ListHoldsAndBookmarksOutputHold
+}
+
+type ListHoldsAndBookmarksOutputBookmark struct {
+	FS, Name string
+	Guid     uint64
+	JobID    JobID
+	v        zfs.FilesystemVersion
+}
+
+type ListHoldsAndBookmarksOutputHold struct {
+	FS            string
+	Snap          string
+	SnapGuid      uint64
+	SnapCreateTXG uint64
+	Tag           string
+	JobID         JobID
+}
+
+// List all holds and bookmarks managed by endpoint
+func ListZFSHoldsAndBookmarks(ctx context.Context, fsfilter zfs.DatasetFilter) (*ListHoldsAndBookmarksOutput, error) {
+
+	// initialize all fields so that JSON serializion of output looks pretty (see client/holds.go)
+	// however, listZFSHoldsAndBookmarksImplFS shouldn't rely on it
+	out := &ListHoldsAndBookmarksOutput{
+		StepBookmarks:              make([]*ListHoldsAndBookmarksOutputBookmark, 0),
+		StepHolds:                  make([]*ListHoldsAndBookmarksOutputHold, 0),
+		ReplicationCursorBookmarks: make([]*ListHoldsAndBookmarksOutputBookmark, 0),
+		V1ReplicationCursors:       make([]*ListHoldsAndBookmarksOutputBookmarkV1ReplicationCursor, 0),
+		LastReceivedHolds:          make([]*ListHoldsAndBookmarksOutputHold, 0),
+	}
+
+	fss, err := zfs.ZFSListMapping(ctx, fsfilter)
+	if err != nil {
+		return nil, errors.Wrap(err, "list filesystems")
+	}
+
+	for _, fs := range fss {
+		err := listZFSHoldsAndBookmarksImplFS(ctx, out, fs)
+		if err != nil {
+			return nil, errors.Wrapf(err, "list holds and bookmarks on %q", fs.ToString())
+		}
+	}
+	return out, nil
+}
+
+func listZFSHoldsAndBookmarksImplFS(ctx context.Context, out *ListHoldsAndBookmarksOutput, fs *zfs.DatasetPath) error {
+	fsvs, err := zfs.ZFSListFilesystemVersions(fs, nil)
+	if err != nil {
+		return errors.Wrapf(err, "list filesystem versions of %q", fs)
+	}
+	for _, v := range fsvs {
+		switch v.Type {
+		case zfs.Bookmark:
+			listZFSHoldsAndBookmarksImplTryParseBookmark(ctx, out, fs, v)
+		case zfs.Snapshot:
+			holds, err := zfs.ZFSHolds(ctx, fs.ToString(), v.Name)
+			if err != nil {
+				return errors.Wrapf(err, "get holds of %q", v.ToAbsPath(fs))
+			}
+			for _, tag := range holds {
+				listZFSHoldsAndBookmarksImplSnapshotTryParseHold(ctx, out, fs, v, tag)
+			}
+		default:
+			continue
+		}
+	}
+	return nil
+}
+
+// pure function, err != nil always indicates parsing error
+func listZFSHoldsAndBookmarksImplTryParseBookmark(ctx context.Context, out *ListHoldsAndBookmarksOutput, fs *zfs.DatasetPath, v zfs.FilesystemVersion) {
+	var err error
+
+	if v.Type != zfs.Bookmark {
+		panic("impl error")
+	}
+
+	fullname := v.ToAbsPath(fs)
+
+	bm := &ListHoldsAndBookmarksOutputBookmark{
+		FS: fs.ToString(), Name: v.Name, v: v,
+	}
+	bm.Guid, bm.JobID, err = ParseStepBookmarkName(fullname)
+	if err == nil {
+		out.StepBookmarks = append(out.StepBookmarks, bm)
+		return
+	}
+
+	bm.Guid, bm.JobID, err = ParseReplicationCursorBookmarkName(fullname)
+	if err == nil {
+		out.ReplicationCursorBookmarks = append(out.ReplicationCursorBookmarks, bm)
+		return
+	} else if err == ErrV1ReplicationCursor {
+		v1rc := &ListHoldsAndBookmarksOutputBookmarkV1ReplicationCursor{
+			FS: fs.ToString(), Name: v.Name,
+		}
+		out.V1ReplicationCursors = append(out.V1ReplicationCursors, v1rc)
+		return
+	}
+}
+
+// pure function, err != nil always indicates parsing error
+func listZFSHoldsAndBookmarksImplSnapshotTryParseHold(ctx context.Context, out *ListHoldsAndBookmarksOutput, fs *zfs.DatasetPath, v zfs.FilesystemVersion, holdTag string) {
+	var err error
+
+	if v.Type != zfs.Snapshot {
+		panic("impl error")
+	}
+
+	hold := &ListHoldsAndBookmarksOutputHold{
+		FS:            fs.ToString(),
+		Snap:          v.Name,
+		SnapGuid:      v.Guid,
+		SnapCreateTXG: v.CreateTXG,
+		Tag:           holdTag,
+	}
+	hold.JobID, err = ParseStepHoldTag(holdTag)
+	if err == nil {
+		out.StepHolds = append(out.StepHolds, hold)
+		return
+	}
+
+	hold.JobID, err = ParseLastReceivedHoldTag(holdTag)
+	if err == nil {
+		out.LastReceivedHolds = append(out.LastReceivedHolds, hold)
+		return
+	}
+
+}

endpoint/endpoint_zfs_abstraction.go

@@ -1,835 +0,0 @@
-package endpoint
-
-import (
-	"context"
-	"encoding/json"
-	"fmt"
-	"math"
-	"sort"
-	"strings"
-	"sync"
-
-	"github.com/pkg/errors"
-
-	"github.com/zrepl/zrepl/util/envconst"
-	"github.com/zrepl/zrepl/util/semaphore"
-	"github.com/zrepl/zrepl/zfs"
-)
-
-type AbstractionType string
-
-// Implementation note:
-// There are a lot of exhaustive switches on AbstractionType in the code base.
-// When adding a new abstraction type, make sure to search and update them!
-const (
-	AbstractionStepBookmark                AbstractionType = "step-bookmark"
-	AbstractionStepHold                    AbstractionType = "step-hold"
-	AbstractionLastReceivedHold            AbstractionType = "last-received-hold"
-	AbstractionReplicationCursorBookmarkV1 AbstractionType = "replication-cursor-bookmark-v1"
-	AbstractionReplicationCursorBookmarkV2 AbstractionType = "replication-cursor-bookmark-v2"
-)
-
-var AbstractionTypesAll = map[AbstractionType]bool{
-	AbstractionStepBookmark:                true,
-	AbstractionStepHold:                    true,
-	AbstractionLastReceivedHold:            true,
-	AbstractionReplicationCursorBookmarkV1: true,
-	AbstractionReplicationCursorBookmarkV2: true,
-}
-
-// Implementation Note:
-// Whenever you add a new accessor, adjust AbstractionJSON.MarshalJSON accordingly
-type Abstraction interface {
-	GetType() AbstractionType
-	GetFS() string
-	GetName() string
-	GetFullPath() string
-	GetJobID() *JobID // may return nil if the abstraction does not have a JobID
-	GetCreateTXG() uint64
-	GetFilesystemVersion() zfs.FilesystemVersion
-	String() string
-	// destroy the abstraction: either releases the hold or destroys the bookmark
-	Destroy(context.Context) error
-	json.Marshaler
-}
-
-func (t AbstractionType) Validate() error {
-	switch t {
-	case AbstractionStepBookmark:
-		return nil
-	case AbstractionStepHold:
-		return nil
-	case AbstractionLastReceivedHold:
-		return nil
-	case AbstractionReplicationCursorBookmarkV1:
-		return nil
-	case AbstractionReplicationCursorBookmarkV2:
-		return nil
-	default:
-		return errors.Errorf("unknown abstraction type %q", t)
-	}
-}
-
-func (t AbstractionType) MustValidate() error {
-	if err := t.Validate(); err != nil {
-		panic(err)
-	}
-	return nil
-}
-
-type AbstractionJSON struct{ Abstraction }
-
-var _ json.Marshaler = (*AbstractionJSON)(nil)
-
-func (a AbstractionJSON) MarshalJSON() ([]byte, error) {
-	type S struct {
-		Type              AbstractionType
-		FS                string
-		Name              string
-		FullPath          string
-		JobID             *JobID // may return nil if the abstraction does not have a JobID
-		CreateTXG         uint64
-		FilesystemVersion zfs.FilesystemVersion
-		String            string
-	}
-	v := S{
-		Type:              a.Abstraction.GetType(),
-		FS:                a.Abstraction.GetFS(),
-		Name:              a.Abstraction.GetName(),
-		FullPath:          a.Abstraction.GetFullPath(),
-		JobID:             a.Abstraction.GetJobID(),
-		CreateTXG:         a.Abstraction.GetCreateTXG(),
-		FilesystemVersion: a.Abstraction.GetFilesystemVersion(),
-		String:            a.Abstraction.String(),
-	}
-	return json.Marshal(v)
-}
-
-type AbstractionTypeSet map[AbstractionType]bool
-
-func AbstractionTypeSetFromStrings(sts []string) (AbstractionTypeSet, error) {
-	ats := make(map[AbstractionType]bool, len(sts))
-	for i, t := range sts {
-		at := AbstractionType(t)
-		if err := at.Validate(); err != nil {
-			return nil, errors.Wrapf(err, "invalid abstraction type #%d %q", i+1, t)
-		}
-		ats[at] = true
-	}
-	return ats, nil
-}
-
-func (s AbstractionTypeSet) ContainsAll(q AbstractionTypeSet) bool {
-	for k := range q {
-		if _, ok := s[k]; !ok {
-			return false
-		}
-	}
-	return true
-}
-
-func (s AbstractionTypeSet) ContainsAnyOf(q AbstractionTypeSet) bool {
-	for k := range q {
-		if _, ok := s[k]; ok {
-			return true
-		}
-	}
-	return false
-}
-
-func (s AbstractionTypeSet) String() string {
-	sts := make([]string, 0, len(s))
-	for i := range s {
-		sts = append(sts, string(i))
-	}
-	sts = sort.StringSlice(sts)
-	return strings.Join(sts, ",")
-}
-
-func (s AbstractionTypeSet) Validate() error {
-	for k := range s {
-		if err := k.Validate(); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-type BookmarkExtractor func(fs *zfs.DatasetPath, v zfs.FilesystemVersion) Abstraction
-
-// returns nil if the abstraction type is not bookmark-based
-func (t AbstractionType) BookmarkExtractor() BookmarkExtractor {
-	switch t {
-	case AbstractionStepBookmark:
-		return StepBookmarkExtractor
-	case AbstractionReplicationCursorBookmarkV1:
-		return ReplicationCursorV1Extractor
-	case AbstractionReplicationCursorBookmarkV2:
-		return ReplicationCursorV2Extractor
-	case AbstractionStepHold:
-		return nil
-	case AbstractionLastReceivedHold:
-		return nil
-	default:
-		panic(fmt.Sprintf("unimpl: %q", t))
-	}
-}
-
-type HoldExtractor = func(fs *zfs.DatasetPath, v zfs.FilesystemVersion, tag string) Abstraction
-
-// returns nil if the abstraction type is not hold-based
-func (t AbstractionType) HoldExtractor() HoldExtractor {
-	switch t {
-	case AbstractionStepBookmark:
-		return nil
-	case AbstractionReplicationCursorBookmarkV1:
-		return nil
-	case AbstractionReplicationCursorBookmarkV2:
-		return nil
-	case AbstractionStepHold:
-		return StepHoldExtractor
-	case AbstractionLastReceivedHold:
-		return LastReceivedHoldExtractor
-	default:
-		panic(fmt.Sprintf("unimpl: %q", t))
-	}
-}
-
-type ListZFSHoldsAndBookmarksQuery struct {
-	FS ListZFSHoldsAndBookmarksQueryFilesystemFilter
-	// What abstraction types should match (any contained in the set)
-	What AbstractionTypeSet
-
-	// The output for the query must satisfy _all_ (AND) requirements of all fields in this query struct.
-
-	// if not nil: JobID of the hold or bookmark in question must be equal
-	// else: JobID of the hold or bookmark can be any value
-	JobID *JobID
-
-	// zero-value means any CreateTXG is acceptable
-	CreateTXG CreateTXGRange
-
-	// Number of concurrently queried filesystems. Must be >= 1
-	Concurrency int64
-}
-
-type CreateTXGRangeBound struct {
-	CreateTXG uint64
-	Inclusive *zfs.NilBool // must not be nil
-}
-
-// A non-empty range of CreateTXGs
-//
-// If both Since and Until are nil, any CreateTXG is acceptable
-type CreateTXGRange struct {
-	// if not nil: The hold's snapshot or the bookmark's createtxg must be greater than (or equal) Since
-	// else: CreateTXG of the hold or bookmark can be any value accepted by Until
-	Since *CreateTXGRangeBound
-	// if not nil: The hold's snapshot or the bookmark's createtxg must be less than (or equal) Until
-	// else: CreateTXG of the hold or bookmark can be any value accepted by Since
-	Until *CreateTXGRangeBound
-}
-
-// FS == nil XOR Filter == nil
-type ListZFSHoldsAndBookmarksQueryFilesystemFilter struct {
-	FS     *string
-	Filter zfs.DatasetFilter
-}
-
-func (q *ListZFSHoldsAndBookmarksQuery) Validate() error {
-	if err := q.FS.Validate(); err != nil {
-		return errors.Wrap(err, "FS")
-	}
-	if q.JobID != nil {
-		q.JobID.MustValidate() // FIXME
-	}
-	if err := q.CreateTXG.Validate(); err != nil {
-		return errors.Wrap(err, "CreateTXGRange")
-	}
-	if err := q.What.Validate(); err != nil {
-		return err
-	}
-	if q.Concurrency < 1 {
-		return errors.New("Concurrency must be >= 1")
-	}
-	return nil
-}
-
-var createTXGRangeBoundAllowCreateTXG0 = envconst.Bool("ZREPL_ENDPOINT_LIST_ABSTRACTIONS_QUERY_CREATETXG_RANGE_BOUND_ALLOW_0", false)
-
-func (i *CreateTXGRangeBound) Validate() error {
-	if err := i.Inclusive.Validate(); err != nil {
-		return errors.Wrap(err, "Inclusive")
-	}
-	if i.CreateTXG == 0 && !createTXGRangeBoundAllowCreateTXG0 {
-		return errors.New("CreateTXG must be non-zero")
-	}
-	return nil
-
-}
-
-func (f *ListZFSHoldsAndBookmarksQueryFilesystemFilter) Validate() error {
-	if f == nil {
-		return nil
-	}
-	fsSet := f.FS != nil
-	filterSet := f.Filter != nil
-	if fsSet && filterSet || !fsSet && !filterSet {
-		return fmt.Errorf("must set FS or Filter field, but fsIsSet=%v and filterIsSet=%v", fsSet, filterSet)
-	}
-	if fsSet {
-		if err := zfs.EntityNamecheck(*f.FS, zfs.EntityTypeFilesystem); err != nil {
-			return errors.Wrap(err, "FS invalid")
-		}
-	}
-	return nil
-}
-
-func (f *ListZFSHoldsAndBookmarksQueryFilesystemFilter) Filesystems(ctx context.Context) ([]string, error) {
-	if err := f.Validate(); err != nil {
-		panic(err)
-	}
-	if f.FS != nil {
-		return []string{*f.FS}, nil
-	}
-	if f.Filter != nil {
-		dps, err := zfs.ZFSListMapping(ctx, f.Filter)
-		if err != nil {
-			return nil, err
-		}
-		fss := make([]string, len(dps))
-		for i, dp := range dps {
-			fss[i] = dp.ToString()
-		}
-		return fss, nil
-	}
-	panic("unreachable")
-}
-
-func (r *CreateTXGRange) Validate() error {
-	if r.Since != nil {
-		if err := r.Since.Validate(); err != nil {
-			return errors.Wrap(err, "Since")
-		}
-	}
-	if r.Until != nil {
-		if err := r.Until.Validate(); err != nil {
-			return errors.Wrap(err, "Until")
-		}
-	}
-	if _, err := r.effectiveBounds(); err != nil {
-		return errors.Wrapf(err, "specified range %s is semantically invalid", r)
-	}
-	return nil
-}
-
-// inclusive-inclusive bounds
-type effectiveBounds struct {
-	sinceInclusive uint64
-	sinceUnbounded bool
-	untilInclusive uint64
-	untilUnbounded bool
-}
-
-// callers must have validated r.Since and r.Until before calling this method
-func (r *CreateTXGRange) effectiveBounds() (bounds effectiveBounds, err error) {
-
-	bounds.sinceUnbounded = r.Since == nil
-	bounds.untilUnbounded = r.Until == nil
-
-	if r.Since == nil && r.Until == nil {
-		return bounds, nil
-	}
-
-	if r.Since != nil {
-		bounds.sinceInclusive = r.Since.CreateTXG
-		if !r.Since.Inclusive.B {
-			if r.Since.CreateTXG == math.MaxUint64 {
-				return bounds, errors.Errorf("Since-exclusive (%v) must be less than math.MaxUint64 (%v)",
-					r.Since.CreateTXG, uint64(math.MaxUint64))
-			}
-			bounds.sinceInclusive++
-		}
-	}
-
-	if r.Until != nil {
-		bounds.untilInclusive = r.Until.CreateTXG
-		if !r.Until.Inclusive.B {
-			if r.Until.CreateTXG == 0 {
-				return bounds, errors.Errorf("Until-exclusive (%v) must be greater than 0", r.Until.CreateTXG)
-			}
-			bounds.untilInclusive--
-		}
-	}
-
-	if !bounds.sinceUnbounded && !bounds.untilUnbounded {
-		if bounds.sinceInclusive >= bounds.untilInclusive {
-			return bounds, errors.Errorf("effective range bounds are [%v,%v] which is empty or invalid", bounds.sinceInclusive, bounds.untilInclusive)
-		}
-		// fallthrough
-	}
-
-	return bounds, nil
-}
-
-func (r *CreateTXGRange) String() string {
-	var buf strings.Builder
-	if r.Since == nil {
-		fmt.Fprintf(&buf, "~")
-	} else {
-		if err := r.Since.Inclusive.Validate(); err != nil {
-			fmt.Fprintf(&buf, "?")
-		} else if r.Since.Inclusive.B {
-			fmt.Fprintf(&buf, "[")
-		} else {
-			fmt.Fprintf(&buf, "(")
-		}
-		fmt.Fprintf(&buf, "%d", r.Since.CreateTXG)
-	}
-
-	fmt.Fprintf(&buf, ",")
-
-	if r.Until == nil {
-		fmt.Fprintf(&buf, "~")
-	} else {
-		fmt.Fprintf(&buf, "%d", r.Until.CreateTXG)
-		if err := r.Until.Inclusive.Validate(); err != nil {
-			fmt.Fprintf(&buf, "?")
-		} else if r.Until.Inclusive.B {
-			fmt.Fprintf(&buf, "]")
-		} else {
-			fmt.Fprintf(&buf, ")")
-		}
-	}
-
-	return buf.String()
-}
-
-// panics if not .Validate()
-func (r *CreateTXGRange) IsUnbounded() bool {
-	if err := r.Validate(); err != nil {
-		panic(err)
-	}
-	bounds, err := r.effectiveBounds()
-	if err != nil {
-		panic(err)
-	}
-	return bounds.sinceUnbounded && bounds.untilUnbounded
-}
-
-// panics if not .Validate()
-func (r *CreateTXGRange) Contains(qCreateTxg uint64) bool {
-	if err := r.Validate(); err != nil {
-		panic(err)
-	}
-
-	bounds, err := r.effectiveBounds()
-	if err != nil {
-		panic(err)
-	}
-
-	sinceMatches := bounds.sinceUnbounded || bounds.sinceInclusive <= qCreateTxg
-	untilMatches := bounds.untilUnbounded || qCreateTxg <= bounds.untilInclusive
-
-	return sinceMatches && untilMatches
-}
-
-type ListAbstractionsError struct {
-	FS   string
-	Snap string
-	What string
-	Err  error
-}
-
-func (e ListAbstractionsError) Error() string {
-	if e.FS == "" {
-		return fmt.Sprintf("list endpoint abstractions: %s: %s", e.What, e.Err)
-	} else {
-		v := e.FS
-		if e.Snap != "" {
-			v = fmt.Sprintf("%s@%s", e.FS, e.Snap)
-		}
-		return fmt.Sprintf("list endpoint abstractions on %q: %s: %s", v, e.What, e.Err)
-	}
-}
-
-type putListAbstractionErr func(err error, fs string, what string)
-type putListAbstraction func(a Abstraction)
-
-type ListAbstractionsErrors []ListAbstractionsError
-
-func (e ListAbstractionsErrors) Error() string {
-	if len(e) == 0 {
-		panic(e)
-	}
-	if len(e) == 1 {
-		return fmt.Sprintf("list endpoint abstractions: %s", e[0])
-	}
-	msgs := make([]string, len(e))
-	for i := range e {
-		msgs[i] = e.Error()
-	}
-	return fmt.Sprintf("list endpoint abstractions: multiple errors:\n%s", strings.Join(msgs, "\n"))
-}
-
-func ListAbstractions(ctx context.Context, query ListZFSHoldsAndBookmarksQuery) (out []Abstraction, outErrs []ListAbstractionsError, err error) {
-	outChan, outErrsChan, err := ListAbstractionsStreamed(ctx, query)
-	if err != nil {
-		return nil, nil, err
-	}
-	var wg sync.WaitGroup
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		for a := range outChan {
-			out = append(out, a)
-		}
-	}()
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		for err := range outErrsChan {
-			outErrs = append(outErrs, err)
-		}
-	}()
-	wg.Wait()
-	return out, outErrs, nil
-}
-
-// if err != nil, the returned channels are both nil
-// if err == nil, both channels must be fully drained by the caller to avoid leaking goroutines
-func ListAbstractionsStreamed(ctx context.Context, query ListZFSHoldsAndBookmarksQuery) (<-chan Abstraction, <-chan ListAbstractionsError, error) {
-
-	// impl note: structure the query processing in such a way that
-	// a minimum amount of zfs shell-outs needs to be done
-
-	if err := query.Validate(); err != nil {
-		return nil, nil, errors.Wrap(err, "validate query")
-	}
-
-	fss, err := query.FS.Filesystems(ctx)
-	if err != nil {
-		return nil, nil, errors.Wrap(err, "list filesystems")
-	}
-
-	outErrs := make(chan ListAbstractionsError)
-	out := make(chan Abstraction)
-
-	errCb := func(err error, fs string, what string) {
-		outErrs <- ListAbstractionsError{Err: err, FS: fs, What: what}
-	}
-	emitAbstraction := func(a Abstraction) {
-		jobIdMatches := query.JobID == nil || a.GetJobID() == nil || *a.GetJobID() == *query.JobID
-
-		createTXGMatches := query.CreateTXG.Contains(a.GetCreateTXG())
-
-		if jobIdMatches && createTXGMatches {
-			out <- a
-		}
-	}
-
-	sem := semaphore.New(int64(query.Concurrency))
-	go func() {
-		defer close(out)
-		defer close(outErrs)
-		var wg sync.WaitGroup
-		defer wg.Wait()
-		for i := range fss {
-			wg.Add(1)
-			go func(i int) {
-				defer wg.Done()
-				g, err := sem.Acquire(ctx)
-				if err != nil {
-					errCb(err, fss[i], err.Error())
-					return
-				}
-				func() {
-					defer g.Release()
-					listAbstractionsImplFS(ctx, fss[i], &query, emitAbstraction, errCb)
-				}()
-			}(i)
-		}
-	}()
-
-	return out, outErrs, nil
-}
-
-func listAbstractionsImplFS(ctx context.Context, fs string, query *ListZFSHoldsAndBookmarksQuery, emitCandidate putListAbstraction, errCb putListAbstractionErr) {
-	fsp, err := zfs.NewDatasetPath(fs)
-	if err != nil {
-		panic(err)
-	}
-
-	if len(query.What) == 0 {
-		return
-	}
-
-	whatTypes := zfs.VersionTypeSet{}
-	for what := range query.What {
-		if e := what.BookmarkExtractor(); e != nil {
-			whatTypes[zfs.Bookmark] = true
-		}
-		if e := what.HoldExtractor(); e != nil {
-			whatTypes[zfs.Snapshot] = true
-		}
-	}
-	fsvs, err := zfs.ZFSListFilesystemVersions(fsp, zfs.ListFilesystemVersionsOptions{
-		Types: whatTypes,
-	})
-	if err != nil {
-		errCb(err, fs, "list filesystem versions")
-		return
-	}
-
-	for at := range query.What {
-		bmE := at.BookmarkExtractor()
-		holdE := at.HoldExtractor()
-		if bmE == nil && holdE == nil || bmE != nil && holdE != nil {
-			panic("implementation error: extractors misconfigured for " + at)
-		}
-		for _, v := range fsvs {
-			var a Abstraction
-			if v.Type == zfs.Bookmark && bmE != nil {
-				a = bmE(fsp, v)
-			}
-			if v.Type == zfs.Snapshot && holdE != nil && query.CreateTXG.Contains(v.GetCreateTXG()) && (!v.UserRefs.Valid || v.UserRefs.Value > 0) {
-				holds, err := zfs.ZFSHolds(ctx, fsp.ToString(), v.Name)
-				if err != nil {
-					errCb(err, v.ToAbsPath(fsp), "get hold on snap")
-					continue
-				}
-				for _, tag := range holds {
-					a = holdE(fsp, v, tag)
-				}
-			}
-			if a != nil {
-				emitCandidate(a)
-			}
-		}
-	}
-}
-
-type BatchDestroyResult struct {
-	Abstraction
-	DestroyErr error
-}
-
-var _ json.Marshaler = (*BatchDestroyResult)(nil)
-
-func (r BatchDestroyResult) MarshalJSON() ([]byte, error) {
-	err := ""
-	if r.DestroyErr != nil {
-		err = r.DestroyErr.Error()
-	}
-	s := struct {
-		Abstraction AbstractionJSON
-		DestroyErr  string
-	}{
-		AbstractionJSON{r.Abstraction},
-		err,
-	}
-	return json.Marshal(s)
-}
-
-func BatchDestroy(ctx context.Context, abs []Abstraction) <-chan BatchDestroyResult {
-	// hold-based batching: per snapshot
-	// bookmark-based batching: none possible via CLI
-	// => not worth the trouble for now, will be worth it once we start using channel programs
-	// => TODO: actual batching using channel programs
-	res := make(chan BatchDestroyResult, len(abs))
-	go func() {
-		for _, a := range abs {
-			res <- BatchDestroyResult{
-				a,
-				a.Destroy(ctx),
-			}
-		}
-		close(res)
-	}()
-	return res
-}
-
-type StalenessInfo struct {
-	ConstructedWithQuery ListZFSHoldsAndBookmarksQuery
-	Live                 []Abstraction
-	Stale                []Abstraction
-}
-
-type fsAndJobId struct {
-	fs    string
-	jobId JobID
-}
-
-type ListStaleQueryError struct {
-	error
-}
-
-// returns *ListStaleQueryError if the given query cannot be used for determining staleness info
-func ListStale(ctx context.Context, q ListZFSHoldsAndBookmarksQuery) (*StalenessInfo, error) {
-	if !q.CreateTXG.IsUnbounded() {
-		// we must determine the most recent step per FS, can't allow that
-		return nil, &ListStaleQueryError{errors.New("ListStale cannot have Until != nil set on query")}
-	}
-
-	// if asking for step holds, must also as for step bookmarks (same kind of abstraction)
-	// as well as replication cursor bookmarks (for firstNotStale)
-	ifAnyThenAll := AbstractionTypeSet{
-		AbstractionStepHold:                    true,
-		AbstractionStepBookmark:                true,
-		AbstractionReplicationCursorBookmarkV2: true,
-	}
-	if q.What.ContainsAnyOf(ifAnyThenAll) && !q.What.ContainsAll(ifAnyThenAll) {
-		return nil, &ListStaleQueryError{errors.Errorf("ListStale requires query to ask for all of %s", ifAnyThenAll.String())}
-	}
-
-	// ----------------- done validating query for listStaleFiltering -----------------------
-
-	qAbs, absErr, err := ListAbstractions(ctx, q)
-	if err != nil {
-		return nil, err
-	}
-	if len(absErr) > 0 {
-		// can't go on here because we can't determine the most recent step
-		return nil, ListAbstractionsErrors(absErr)
-	}
-
-	si := listStaleFiltering(qAbs, q.CreateTXG.Since)
-	si.ConstructedWithQuery = q
-	return si, nil
-}
-
-type fsAjobAtype struct {
-	fsAndJobId
-	Type AbstractionType
-}
-
-// For step holds and bookmarks, only those older than the most recent replication cursor
-// of their (filesystem,job) is considered because younger ones cannot be stale by definition
-// (if we destroy them, we might actually lose the hold on the `To` for an ongoing incremental replication)
-//
-// For replication cursors and last-received-holds, only the most recent one is kept.
-//
-// the returned StalenessInfo.ConstructedWithQuery is not set
-func listStaleFiltering(abs []Abstraction, sinceBound *CreateTXGRangeBound) *StalenessInfo {
-
-	var noJobId []Abstraction
-	by := make(map[fsAjobAtype][]Abstraction)
-	for _, a := range abs {
-		if a.GetJobID() == nil {
-			noJobId = append(noJobId, a)
-			continue
-		}
-		faj := fsAjobAtype{fsAndJobId{a.GetFS(), *a.GetJobID()}, a.GetType()}
-		l := by[faj]
-		l = append(l, a)
-		by[faj] = l
-	}
-
-	type stepFirstNotStaleCandidate struct {
-		cursor *Abstraction
-		step   *Abstraction
-	}
-	stepFirstNotStaleCandidates := make(map[fsAndJobId]stepFirstNotStaleCandidate) // empty map => will always return nil
-	for _, a := range abs {
-		key := fsAndJobId{a.GetFS(), *a.GetJobID()}
-		c := stepFirstNotStaleCandidates[key]
-
-		switch a.GetType() {
-		// stepFirstNotStaleCandidate.cursor
-		case AbstractionReplicationCursorBookmarkV2:
-			if c.cursor == nil || (*c.cursor).GetCreateTXG() < a.GetCreateTXG() {
-				a := a
-				c.cursor = &a
-			}
-
-		// stepFirstNotStaleCandidate.step
-		case AbstractionStepBookmark:
-			fallthrough
-		case AbstractionStepHold:
-			if c.step == nil || (*c.step).GetCreateTXG() < a.GetCreateTXG() {
-				a := a
-				c.step = &a
-			}
-
-		// not interested in the others
-		default:
-			continue // not relevant
-
-		}
-
-		stepFirstNotStaleCandidates[key] = c
-	}
-
-	ret := &StalenessInfo{
-		Live:  noJobId,
-		Stale: []Abstraction{},
-	}
-
-	for k := range by {
-		l := by[k]
-
-		if k.Type == AbstractionStepHold || k.Type == AbstractionStepBookmark {
-			// all older than the most recent cursor are stale, others are always live
-
-			// if we don't have a replication cursor yet, use untilBound = nil
-			// to consider all steps stale (...at first)
-			var untilBound *CreateTXGRangeBound
-			{
-				sfnsc := stepFirstNotStaleCandidates[k.fsAndJobId]
-
-				// if there's a replication cursor, use it as a cutoff between live and stale
-				// if there's none, we are in initial replication and only need to keep
-				// the most recent step hold live, since that's what our initial replication strategy
-				// uses (both initially and on resume)
-				// (FIXME hardcoded replication strategy)
-				if sfnsc.cursor != nil {
-					untilBound = &CreateTXGRangeBound{
-						CreateTXG: (*sfnsc.cursor).GetCreateTXG(),
-						// if we have a cursor, can throw away step hold on both From and To
-						Inclusive: &zfs.NilBool{B: true},
-					}
-				} else if sfnsc.step != nil {
-					untilBound = &CreateTXGRangeBound{
-						CreateTXG: (*sfnsc.step).GetCreateTXG(),
-						// if we don't have a cursor, the step most recent step hold is our
-						// initial replication cursor and it's possibly still live (interrupted initial replication)
-						Inclusive: &zfs.NilBool{B: false},
-					}
-				} else {
-					untilBound = nil // consider everything stale
-				}
-			}
-			staleRange := CreateTXGRange{
-				Since: sinceBound,
-				Until: untilBound,
-			}
-
-			// partition by staleRange
-			for _, a := range l {
-				if staleRange.Contains(a.GetCreateTXG()) {
-					ret.Stale = append(ret.Stale, a)
-				} else {
-					ret.Live = append(ret.Live, a)
-				}
-			}
-
-		} else if k.Type == AbstractionReplicationCursorBookmarkV2 || k.Type == AbstractionLastReceivedHold {
-			// all but the most recent are stale by definition (we always _move_ them)
-			// NOTE: must not use firstNotStale in this branch, not computed for these types
-
-			// sort descending (highest createtxg first), then cut off
-			sort.Slice(l, func(i, j int) bool {
-				return l[i].GetCreateTXG() > l[j].GetCreateTXG()
-			})
-			if len(l) > 0 {
-				ret.Live = append(ret.Live, l[0])
-				ret.Stale = append(ret.Stale, l[1:]...)
-			}
-		} else {
-			ret.Live = append(ret.Live, l...)
-		}
-	}
-
-	return ret
-
-}

endpoint/endpoint_zfs_abstraction_cursor_and_last_received_hold.go

@@ -1,386 +0,0 @@
-package endpoint
-
-import (
-	"context"
-	"encoding/json"
-	"fmt"
-	"regexp"
-	"sort"
-
-	"github.com/pkg/errors"
-
-	"github.com/zrepl/zrepl/util/errorarray"
-	"github.com/zrepl/zrepl/zfs"
-)
-
-const replicationCursorBookmarkNamePrefix = "zrepl_CURSOR"
-
-func ReplicationCursorBookmarkName(fs string, guid uint64, id JobID) (string, error) {
-	return replicationCursorBookmarkNameImpl(fs, guid, id.String())
-}
-
-func replicationCursorBookmarkNameImpl(fs string, guid uint64, jobid string) (string, error) {
-	return makeJobAndGuidBookmarkName(replicationCursorBookmarkNamePrefix, fs, guid, jobid)
-}
-
-var ErrV1ReplicationCursor = fmt.Errorf("bookmark name is a v1-replication cursor")
-
-//err != nil always means that the bookmark is not a valid replication bookmark
-//
-// Returns ErrV1ReplicationCursor as error if the bookmark is a v1 replication cursor
-func ParseReplicationCursorBookmarkName(fullname string) (uint64, JobID, error) {
-
-	// check for legacy cursors
-	{
-		if err := zfs.EntityNamecheck(fullname, zfs.EntityTypeBookmark); err != nil {
-			return 0, JobID{}, errors.Wrap(err, "parse replication cursor bookmark name")
-		}
-		_, _, name, err := zfs.DecomposeVersionString(fullname)
-		if err != nil {
-			return 0, JobID{}, errors.Wrap(err, "parse replication cursor bookmark name: decompose version string")
-		}
-		const V1ReplicationCursorBookmarkName = "zrepl_replication_cursor"
-		if name == V1ReplicationCursorBookmarkName {
-			return 0, JobID{}, ErrV1ReplicationCursor
-		}
-		// fallthrough to main parser
-	}
-
-	guid, jobID, err := parseJobAndGuidBookmarkName(fullname, replicationCursorBookmarkNamePrefix)
-	if err != nil {
-		err = errors.Wrap(err, "parse replication cursor bookmark name") // no shadow
-	}
-	return guid, jobID, err
-}
-
-// may return nil for both values, indicating there is no cursor
-func GetMostRecentReplicationCursorOfJob(ctx context.Context, fs string, jobID JobID) (*zfs.FilesystemVersion, error) {
-	fsp, err := zfs.NewDatasetPath(fs)
-	if err != nil {
-		return nil, err
-	}
-	candidates, err := GetReplicationCursors(ctx, fsp, jobID)
-	if err != nil || len(candidates) == 0 {
-		return nil, err
-	}
-
-	sort.Slice(candidates, func(i, j int) bool {
-		return candidates[i].CreateTXG < candidates[j].CreateTXG
-	})
-
-	mostRecent := candidates[len(candidates)-1]
-	return &mostRecent, nil
-}
-
-func GetReplicationCursors(ctx context.Context, dp *zfs.DatasetPath, jobID JobID) ([]zfs.FilesystemVersion, error) {
-
-	fs := dp.ToString()
-	q := ListZFSHoldsAndBookmarksQuery{
-		FS: ListZFSHoldsAndBookmarksQueryFilesystemFilter{FS: &fs},
-		What: map[AbstractionType]bool{
-			AbstractionReplicationCursorBookmarkV1: true,
-			AbstractionReplicationCursorBookmarkV2: true,
-		},
-		JobID:       &jobID,
-		CreateTXG:   CreateTXGRange{},
-		Concurrency: 1,
-	}
-	abs, absErr, err := ListAbstractions(ctx, q)
-	if err != nil {
-		return nil, errors.Wrap(err, "get replication cursor: list bookmarks and holds")
-	}
-	if len(absErr) > 0 {
-		return nil, ListAbstractionsErrors(absErr)
-	}
-
-	var v1, v2 []Abstraction
-	for _, a := range abs {
-		switch a.GetType() {
-		case AbstractionReplicationCursorBookmarkV1:
-			v1 = append(v1, a)
-		case AbstractionReplicationCursorBookmarkV2:
-			v2 = append(v2, a)
-		default:
-			panic("unexpected abstraction: " + a.GetType())
-		}
-	}
-
-	if len(v1) > 0 {
-		getLogger(ctx).WithField("bookmark", v1).
-			Warn("found v1-replication cursor bookmarks, consider running migration 'replication-cursor:v1-v2' after successful replication with this zrepl version")
-	}
-
-	candidates := make([]zfs.FilesystemVersion, 0)
-	for _, v := range v2 {
-		candidates = append(candidates, v.GetFilesystemVersion())
-	}
-
-	return candidates, nil
-}
-
-type ReplicationCursorTarget interface {
-	IsSnapshot() bool
-	GetGuid() uint64
-	GetCreateTXG() uint64
-	ToSendArgVersion() zfs.ZFSSendArgVersion
-}
-
-// `target` is validated before replication cursor is set. if validation fails, the cursor is not moved.
-//
-// returns ErrBookmarkCloningNotSupported if version is a bookmark and bookmarking bookmarks is not supported by ZFS
-func MoveReplicationCursor(ctx context.Context, fs string, target ReplicationCursorTarget, jobID JobID) (destroyedCursors []Abstraction, err error) {
-
-	if !target.IsSnapshot() {
-		return nil, zfs.ErrBookmarkCloningNotSupported
-	}
-
-	bookmarkname, err := ReplicationCursorBookmarkName(fs, target.GetGuid(), jobID)
-	if err != nil {
-		return nil, errors.Wrap(err, "determine replication cursor name")
-	}
-
-	// idempotently create bookmark (guid is encoded in it, hence we'll most likely add a new one
-	// cleanup the old one afterwards
-
-	err = zfs.ZFSBookmark(ctx, fs, target.ToSendArgVersion(), bookmarkname)
-	if err != nil {
-		if err == zfs.ErrBookmarkCloningNotSupported {
-			return nil, err // TODO go1.13 use wrapping
-		}
-		return nil, errors.Wrapf(err, "cannot create bookmark")
-	}
-
-	destroyedCursors, err = DestroyObsoleteReplicationCursors(ctx, fs, target, jobID)
-	if err != nil {
-		return nil, errors.Wrap(err, "destroy obsolete replication cursors")
-	}
-
-	return destroyedCursors, nil
-}
-
-type ReplicationCursor interface {
-	GetCreateTXG() uint64
-}
-
-func DestroyObsoleteReplicationCursors(ctx context.Context, fs string, current ReplicationCursor, jobID JobID) (_ []Abstraction, err error) {
-
-	q := ListZFSHoldsAndBookmarksQuery{
-		FS: ListZFSHoldsAndBookmarksQueryFilesystemFilter{
-			FS: &fs,
-		},
-		What: AbstractionTypeSet{
-			AbstractionReplicationCursorBookmarkV2: true,
-		},
-		JobID: &jobID,
-		CreateTXG: CreateTXGRange{
-			Since: nil,
-			Until: &CreateTXGRangeBound{
-				CreateTXG: current.GetCreateTXG(),
-				Inclusive: &zfs.NilBool{B: false},
-			},
-		},
-		Concurrency: 1,
-	}
-	abs, absErr, err := ListAbstractions(ctx, q)
-	if err != nil {
-		return nil, errors.Wrap(err, "list abstractions")
-	}
-	if len(absErr) > 0 {
-		return nil, errors.Wrap(ListAbstractionsErrors(absErr), "list abstractions")
-	}
-
-	var destroyed []Abstraction
-	var errs []error
-	for res := range BatchDestroy(ctx, abs) {
-		log := getLogger(ctx).
-			WithField("replication_cursor_bookmark", res.Abstraction)
-		if res.DestroyErr != nil {
-			errs = append(errs, res.DestroyErr)
-			log.WithError(err).
-				Error("cannot destroy obsolete replication cursor bookmark")
-		} else {
-			destroyed = append(destroyed, res.Abstraction)
-			log.Info("destroyed obsolete replication cursor bookmark")
-		}
-	}
-	if len(errs) == 0 {
-		return destroyed, nil
-	} else {
-		return destroyed, errorarray.Wrap(errs, "destroy obsolete replication cursor")
-	}
-}
-
-var lastReceivedHoldTagRE = regexp.MustCompile("^zrepl_last_received_J_(.+)$")
-
-// err != nil always means that the bookmark is not a step bookmark
-func ParseLastReceivedHoldTag(tag string) (JobID, error) {
-	match := lastReceivedHoldTagRE.FindStringSubmatch(tag)
-	if match == nil {
-		return JobID{}, errors.Errorf("parse last-received-hold tag: does not match regex %s", lastReceivedHoldTagRE.String())
-	}
-	jobId, err := MakeJobID(match[1])
-	if err != nil {
-		return JobID{}, errors.Wrap(err, "parse last-received-hold tag: invalid job id field")
-	}
-	return jobId, nil
-}
-
-func LastReceivedHoldTag(jobID JobID) (string, error) {
-	return lastReceivedHoldImpl(jobID.String())
-}
-
-func lastReceivedHoldImpl(jobid string) (string, error) {
-	tag := fmt.Sprintf("zrepl_last_received_J_%s", jobid)
-	if err := zfs.ValidHoldTag(tag); err != nil {
-		return "", err
-	}
-	return tag, nil
-}
-
-func MoveLastReceivedHold(ctx context.Context, fs string, to zfs.FilesystemVersion, jobID JobID) error {
-
-	if !to.IsSnapshot() {
-		return errors.Errorf("last-received-hold: target must be a snapshot: %s", to.FullPath(fs))
-	}
-
-	tag, err := LastReceivedHoldTag(jobID)
-	if err != nil {
-		return errors.Wrap(err, "last-received-hold: hold tag")
-	}
-
-	// we never want to be without a hold
-	// => hold new one before releasing old hold
-
-	err = zfs.ZFSHold(ctx, fs, to, tag)
-	if err != nil {
-		return errors.Wrap(err, "last-received-hold: hold newly received")
-	}
-
-	q := ListZFSHoldsAndBookmarksQuery{
-		What: AbstractionTypeSet{
-			AbstractionLastReceivedHold: true,
-		},
-		FS: ListZFSHoldsAndBookmarksQueryFilesystemFilter{
-			FS: &fs,
-		},
-		JobID: &jobID,
-		CreateTXG: CreateTXGRange{
-			Since: nil,
-			Until: &CreateTXGRangeBound{
-				CreateTXG: to.GetCreateTXG(),
-				Inclusive: &zfs.NilBool{B: false},
-			},
-		},
-		Concurrency: 1,
-	}
-	abs, absErrs, err := ListAbstractions(ctx, q)
-	if err != nil {
-		return errors.Wrap(err, "last-received-hold: list")
-	}
-	if len(absErrs) > 0 {
-		return errors.Wrap(ListAbstractionsErrors(absErrs), "last-received-hold: list")
-	}
-
-	getLogger(ctx).WithField("last-received-holds", fmt.Sprintf("%s", abs)).Debug("releasing last-received-holds")
-
-	var errs []error
-	for res := range BatchDestroy(ctx, abs) {
-		log := getLogger(ctx).
-			WithField("last-received-hold", res.Abstraction)
-		if res.DestroyErr != nil {
-			errs = append(errs, res.DestroyErr)
-			log.WithError(err).
-				Error("cannot release last-received-hold")
-		} else {
-			log.Info("released last-received-hold")
-		}
-	}
-	if len(errs) == 0 {
-		return nil
-	} else {
-		return errorarray.Wrap(errs, "last-received-hold: release")
-	}
-}
-
-func ReplicationCursorV2Extractor(fs *zfs.DatasetPath, v zfs.FilesystemVersion) (_ Abstraction) {
-	if v.Type != zfs.Bookmark {
-		panic("impl error")
-	}
-	fullname := v.ToAbsPath(fs)
-	guid, jobid, err := ParseReplicationCursorBookmarkName(fullname)
-	if err == nil {
-		if guid != v.Guid {
-			// TODO log this possibly tinkered-with bookmark
-			return nil
-		}
-		return &bookmarkBasedAbstraction{
-			Type:              AbstractionReplicationCursorBookmarkV2,
-			FS:                fs.ToString(),
-			FilesystemVersion: v,
-			JobID:             jobid,
-		}
-	}
-	return nil
-}
-
-func ReplicationCursorV1Extractor(fs *zfs.DatasetPath, v zfs.FilesystemVersion) (_ Abstraction) {
-	if v.Type != zfs.Bookmark {
-		panic("impl error")
-	}
-	fullname := v.ToAbsPath(fs)
-	_, _, err := ParseReplicationCursorBookmarkName(fullname)
-	if err == ErrV1ReplicationCursor {
-		return &ReplicationCursorV1{
-			Type:              AbstractionReplicationCursorBookmarkV1,
-			FS:                fs.ToString(),
-			FilesystemVersion: v,
-		}
-	}
-	return nil
-}
-
-var _ HoldExtractor = LastReceivedHoldExtractor
-
-func LastReceivedHoldExtractor(fs *zfs.DatasetPath, v zfs.FilesystemVersion, holdTag string) Abstraction {
-	var err error
-
-	if v.Type != zfs.Snapshot {
-		panic("impl error")
-	}
-
-	jobID, err := ParseLastReceivedHoldTag(holdTag)
-	if err == nil {
-		return &holdBasedAbstraction{
-			Type:              AbstractionLastReceivedHold,
-			FS:                fs.ToString(),
-			FilesystemVersion: v,
-			Tag:               holdTag,
-			JobID:             jobID,
-		}
-	}
-	return nil
-}
-
-type ReplicationCursorV1 struct {
-	Type AbstractionType
-	FS   string
-	zfs.FilesystemVersion
-}
-
-func (c ReplicationCursorV1) GetType() AbstractionType                    { return c.Type }
-func (c ReplicationCursorV1) GetFS() string                               { return c.FS }
-func (c ReplicationCursorV1) GetFullPath() string                         { return fmt.Sprintf("%s#%s", c.FS, c.GetName()) }
-func (c ReplicationCursorV1) GetJobID() *JobID                            { return nil }
-func (c ReplicationCursorV1) GetFilesystemVersion() zfs.FilesystemVersion { return c.FilesystemVersion }
-func (c ReplicationCursorV1) MarshalJSON() ([]byte, error) {
-	return json.Marshal(AbstractionJSON{c})
-}
-func (c ReplicationCursorV1) String() string {
-	return fmt.Sprintf("%s %s", c.Type, c.GetFullPath())
-}
-func (c ReplicationCursorV1) Destroy(ctx context.Context) error {
-	if err := zfs.ZFSDestroyIdempotent(ctx, c.GetFullPath()); err != nil {
-		return errors.Wrapf(err, "destroy %s %s: zfs", c.Type, c.GetFullPath())
-	}
-	return nil
-}

endpoint/endpoint_zfs_abstraction_step.go

@@ -1,286 +0,0 @@
-package endpoint
-
-import (
-	"context"
-	"fmt"
-	"regexp"
-
-	"github.com/pkg/errors"
-
-	"github.com/zrepl/zrepl/util/errorarray"
-	"github.com/zrepl/zrepl/zfs"
-)
-
-var stepHoldTagRE = regexp.MustCompile("^zrepl_STEP_J_(.+)")
-
-func StepHoldTag(jobid JobID) (string, error) {
-	return stepHoldTagImpl(jobid.String())
-}
-
-func stepHoldTagImpl(jobid string) (string, error) {
-	t := fmt.Sprintf("zrepl_STEP_J_%s", jobid)
-	if err := zfs.ValidHoldTag(t); err != nil {
-		return "", err
-	}
-	return t, nil
-}
-
-// err != nil always means that the bookmark is not a step bookmark
-func ParseStepHoldTag(tag string) (JobID, error) {
-	match := stepHoldTagRE.FindStringSubmatch(tag)
-	if match == nil {
-		return JobID{}, fmt.Errorf("parse hold tag: match regex %q", stepHoldTagRE)
-	}
-	jobID, err := MakeJobID(match[1])
-	if err != nil {
-		return JobID{}, errors.Wrap(err, "parse hold tag: invalid job id field")
-	}
-	return jobID, nil
-}
-
-const stepBookmarkNamePrefix = "zrepl_STEP"
-
-// v must be validated by caller
-func StepBookmarkName(fs string, guid uint64, id JobID) (string, error) {
-	return stepBookmarkNameImpl(fs, guid, id.String())
-}
-
-func stepBookmarkNameImpl(fs string, guid uint64, jobid string) (string, error) {
-	return makeJobAndGuidBookmarkName(stepBookmarkNamePrefix, fs, guid, jobid)
-}
-
-// name is the full bookmark name, including dataset path
-//
-// err != nil always means that the bookmark is not a step bookmark
-func ParseStepBookmarkName(fullname string) (guid uint64, jobID JobID, err error) {
-	guid, jobID, err = parseJobAndGuidBookmarkName(fullname, stepBookmarkNamePrefix)
-	if err != nil {
-		err = errors.Wrap(err, "parse step bookmark name") // no shadow!
-	}
-	return guid, jobID, err
-}
-
-// idempotently hold / step-bookmark `version`
-//
-// returns ErrBookmarkCloningNotSupported if version is a bookmark and bookmarking bookmarks is not supported by ZFS
-func HoldStep(ctx context.Context, fs string, v zfs.FilesystemVersion, jobID JobID) (Abstraction, error) {
-	if v.IsSnapshot() {
-
-		tag, err := StepHoldTag(jobID)
-		if err != nil {
-			return nil, errors.Wrap(err, "step hold tag")
-		}
-
-		if err := zfs.ZFSHold(ctx, fs, v, tag); err != nil {
-			return nil, errors.Wrap(err, "step hold: zfs")
-		}
-
-		return &holdBasedAbstraction{
-			Type:              AbstractionStepHold,
-			FS:                fs,
-			Tag:               tag,
-			JobID:             jobID,
-			FilesystemVersion: v,
-		}, nil
-	}
-
-	if !v.IsBookmark() {
-		panic(fmt.Sprintf("version must bei either snapshot or bookmark, got %#v", v))
-	}
-
-	bmname, err := StepBookmarkName(fs, v.Guid, jobID)
-	if err != nil {
-		return nil, errors.Wrap(err, "create step bookmark: determine bookmark name")
-	}
-	// idempotently create bookmark
-	err = zfs.ZFSBookmark(ctx, fs, v.ToSendArgVersion(), bmname)
-	if err != nil {
-		if err == zfs.ErrBookmarkCloningNotSupported {
-			// TODO we could actually try to find a local snapshot that has the requested GUID
-			// 		however, the replication algorithm prefers snapshots anyways, so this quest
-			// 		is most likely not going to be successful. Also, there's the possibility that
-			//      the caller might want to filter what snapshots are eligibile, and this would
-			//      complicate things even further.
-			return nil, err // TODO go1.13 use wrapping
-		}
-		return nil, errors.Wrap(err, "create step bookmark: zfs")
-	}
-	return &bookmarkBasedAbstraction{
-		Type:              AbstractionStepBookmark,
-		FS:                fs,
-		FilesystemVersion: v,
-		JobID:             jobID,
-	}, nil
-}
-
-// idempotently release the step-hold on v if v is a snapshot
-// or idempotently destroy the step-bookmark  of v if v is a bookmark
-//
-// note that this operation leaves v itself untouched, unless v is the step-bookmark itself, in which case v is destroyed
-//
-// returns an instance of *zfs.DatasetDoesNotExist if `v` does not exist
-func ReleaseStep(ctx context.Context, fs string, v zfs.FilesystemVersion, jobID JobID) error {
-
-	if v.IsSnapshot() {
-		tag, err := StepHoldTag(jobID)
-		if err != nil {
-			return errors.Wrap(err, "step release tag")
-		}
-
-		if err := zfs.ZFSRelease(ctx, tag, v.FullPath(fs)); err != nil {
-			return errors.Wrap(err, "step release: zfs")
-		}
-
-		return nil
-	}
-	if !v.IsBookmark() {
-		panic(fmt.Sprintf("impl error: expecting version to be a bookmark, got %#v", v))
-	}
-
-	bmname, err := StepBookmarkName(fs, v.Guid, jobID)
-	if err != nil {
-		return errors.Wrap(err, "step release: determine bookmark name")
-	}
-	// idempotently destroy bookmark
-
-	if err := zfs.ZFSDestroyIdempotent(ctx, bmname); err != nil {
-		return errors.Wrap(err, "step release: bookmark destroy: zfs")
-	}
-
-	return nil
-}
-
-// release {step holds, step bookmarks} earlier and including `mostRecent`
-func ReleaseStepCummulativeInclusive(ctx context.Context, fs string, since *CreateTXGRangeBound, mostRecent zfs.FilesystemVersion, jobID JobID) error {
-	q := ListZFSHoldsAndBookmarksQuery{
-		What: AbstractionTypeSet{
-			AbstractionStepHold:     true,
-			AbstractionStepBookmark: true,
-		},
-		FS: ListZFSHoldsAndBookmarksQueryFilesystemFilter{
-			FS: &fs,
-		},
-		JobID: &jobID,
-		CreateTXG: CreateTXGRange{
-			Since: since,
-			Until: &CreateTXGRangeBound{
-				CreateTXG: mostRecent.CreateTXG,
-				Inclusive: &zfs.NilBool{B: true},
-			},
-		},
-		Concurrency: 1,
-	}
-	abs, absErrs, err := ListAbstractions(ctx, q)
-	if err != nil {
-		return errors.Wrap(err, "step release cummulative: list")
-	}
-	if len(absErrs) > 0 {
-		return errors.Wrap(ListAbstractionsErrors(absErrs), "step release cummulative: list")
-	}
-
-	getLogger(ctx).WithField("step_holds_and_bookmarks", fmt.Sprintf("%s", abs)).Debug("releasing step holds and bookmarks")
-
-	var errs []error
-	for res := range BatchDestroy(ctx, abs) {
-		log := getLogger(ctx).
-			WithField("step_hold_or_bookmark", res.Abstraction)
-		if res.DestroyErr != nil {
-			errs = append(errs, res.DestroyErr)
-			log.WithError(err).
-				Error("cannot release step hold or bookmark")
-		} else {
-			log.Info("released step hold or bookmark")
-		}
-	}
-	if len(errs) == 0 {
-		return nil
-	} else {
-		return errorarray.Wrap(errs, "step release cummulative: release")
-	}
-}
-
-func TryReleaseStepStaleFS(ctx context.Context, fs string, jobID JobID) {
-
-	q := ListZFSHoldsAndBookmarksQuery{
-		FS: ListZFSHoldsAndBookmarksQueryFilesystemFilter{
-			FS: &fs,
-		},
-		JobID: &jobID,
-		What: AbstractionTypeSet{
-			AbstractionStepHold:                    true,
-			AbstractionStepBookmark:                true,
-			AbstractionReplicationCursorBookmarkV2: true,
-		},
-		Concurrency: 1,
-	}
-	staleness, err := ListStale(ctx, q)
-	if _, ok := err.(*ListStaleQueryError); ok {
-		panic(err)
-	} else if err != nil {
-		getLogger(ctx).WithError(err).Error("cannot list stale step holds and bookmarks")
-		return
-	}
-	for _, s := range staleness.Stale {
-		getLogger(ctx).WithField("stale_step_hold_or_bookmark", s).Info("batch-destroying stale step hold or bookmark")
-	}
-	for res := range BatchDestroy(ctx, staleness.Stale) {
-		if res.DestroyErr != nil {
-			getLogger(ctx).
-				WithField("stale_step_hold_or_bookmark", res.Abstraction).
-				WithError(res.DestroyErr).
-				Error("cannot destroy stale step-hold or bookmark")
-		} else {
-			getLogger(ctx).
-				WithField("stale_step_hold_or_bookmark", res.Abstraction).
-				WithError(res.DestroyErr).
-				Info("destroyed stale step-hold or bookmark")
-		}
-	}
-
-}
-
-var _ BookmarkExtractor = StepBookmarkExtractor
-
-func StepBookmarkExtractor(fs *zfs.DatasetPath, v zfs.FilesystemVersion) (_ Abstraction) {
-	if v.Type != zfs.Bookmark {
-		panic("impl error")
-	}
-
-	fullname := v.ToAbsPath(fs)
-
-	guid, jobid, err := ParseStepBookmarkName(fullname)
-	if guid != v.Guid {
-		// TODO log this possibly tinkered-with bookmark
-		return nil
-	}
-	if err == nil {
-		bm := &bookmarkBasedAbstraction{
-			Type:              AbstractionStepBookmark,
-			FS:                fs.ToString(),
-			FilesystemVersion: v,
-			JobID:             jobid,
-		}
-		return bm
-	}
-	return nil
-}
-
-var _ HoldExtractor = StepHoldExtractor
-
-func StepHoldExtractor(fs *zfs.DatasetPath, v zfs.FilesystemVersion, holdTag string) Abstraction {
-	if v.Type != zfs.Snapshot {
-		panic("impl error")
-	}
-
-	jobID, err := ParseStepHoldTag(holdTag)
-	if err == nil {
-		return &holdBasedAbstraction{
-			Type:              AbstractionStepHold,
-			FS:                fs.ToString(),
-			Tag:               holdTag,
-			FilesystemVersion: v,
-			JobID:             jobID,
-		}
-	}
-	return nil
-}

endpoint/endpoint_zfs_abstraction_test.go

@@ -1,234 +0,0 @@
-package endpoint
-
-import (
-	"fmt"
-	"math"
-	"runtime/debug"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-
-	"github.com/zrepl/zrepl/zfs"
-)
-
-func TestCreateTXGRange(t *testing.T) {
-
-	type testCaseExpectation struct {
-		input  uint64
-		expect bool
-	}
-	type testCase struct {
-		name                     string
-		config                   *CreateTXGRange
-		configAllowZeroCreateTXG bool
-		expectInvalid            bool
-		expectString             string
-		expect                   []testCaseExpectation
-	}
-
-	tcs := []testCase{
-		{
-			name:          "unbounded",
-			expectInvalid: false,
-			config: &CreateTXGRange{
-				Since: nil,
-				Until: nil,
-			},
-			expectString: "~,~",
-			expect: []testCaseExpectation{
-				{0, true},
-				{math.MaxUint64, true},
-				{1, true},
-				{math.MaxUint64 - 1, true},
-			},
-		},
-		{
-			name:          "wrong order obvious",
-			expectInvalid: true,
-			config: &CreateTXGRange{
-				Since: &CreateTXGRangeBound{23, &zfs.NilBool{B: true}},
-				Until: &CreateTXGRangeBound{20, &zfs.NilBool{B: true}},
-			},
-			expectString: "[23,20]",
-		},
-		{
-			name:          "wrong order edge-case could also be empty",
-			expectInvalid: true,
-			config: &CreateTXGRange{
-				Since: &CreateTXGRangeBound{23, &zfs.NilBool{B: false}},
-				Until: &CreateTXGRangeBound{22, &zfs.NilBool{B: true}},
-			},
-			expectString: "(23,22]",
-		},
-		{
-			name:          "empty",
-			expectInvalid: true,
-			config: &CreateTXGRange{
-				Since: &CreateTXGRangeBound{2, &zfs.NilBool{B: false}},
-				Until: &CreateTXGRangeBound{2, &zfs.NilBool{B: false}},
-			},
-			expectString: "(2,2)",
-		},
-		{
-			name:          "inclusive-since-exclusive-until",
-			expectInvalid: false,
-			config: &CreateTXGRange{
-				Since: &CreateTXGRangeBound{2, &zfs.NilBool{B: true}},
-				Until: &CreateTXGRangeBound{5, &zfs.NilBool{B: false}},
-			},
-			expectString: "[2,5)",
-			expect: []testCaseExpectation{
-				{0, false},
-				{1, false},
-				{2, true},
-				{3, true},
-				{4, true},
-				{5, false},
-				{6, false},
-			},
-		},
-		{
-			name:          "exclusive-since-inclusive-until",
-			expectInvalid: false,
-			config: &CreateTXGRange{
-				Since: &CreateTXGRangeBound{2, &zfs.NilBool{B: false}},
-				Until: &CreateTXGRangeBound{5, &zfs.NilBool{B: true}},
-			},
-			expectString: "(2,5]",
-			expect: []testCaseExpectation{
-				{0, false},
-				{1, false},
-				{2, false},
-				{3, true},
-				{4, true},
-				{5, true},
-				{6, false},
-			},
-		},
-		{
-			name:          "zero-createtxg-not-allowed-because-likely-programmer-error",
-			expectInvalid: true,
-			config: &CreateTXGRange{
-				Since: nil,
-				Until: &CreateTXGRangeBound{0, &zfs.NilBool{B: true}},
-			},
-			expectString: "~,0]",
-		},
-		{
-			name:          "half-open-no-until",
-			expectInvalid: false,
-			config: &CreateTXGRange{
-				Since: &CreateTXGRangeBound{2, &zfs.NilBool{B: false}},
-				Until: nil,
-			},
-			expectString: "(2,~",
-			expect: []testCaseExpectation{
-				{0, false},
-				{1, false},
-				{2, false},
-				{3, true},
-				{4, true},
-				{5, true},
-				{6, true},
-			},
-		},
-		{
-			name:          "half-open-no-since",
-			expectInvalid: false,
-			config: &CreateTXGRange{
-				Since: nil,
-				Until: &CreateTXGRangeBound{4, &zfs.NilBool{B: true}},
-			},
-			expectString: "~,4]",
-			expect: []testCaseExpectation{
-				{0, true},
-				{1, true},
-				{2, true},
-				{3, true},
-				{4, true},
-				{5, false},
-			},
-		},
-		{
-			name:          "edgeSince",
-			expectInvalid: false,
-			config: &CreateTXGRange{
-				Since: &CreateTXGRangeBound{math.MaxUint64, &zfs.NilBool{B: true}},
-				Until: nil,
-			},
-			expectString: "[18446744073709551615,~",
-			expect: []testCaseExpectation{
-				{math.MaxUint64, true},
-				{math.MaxUint64 - 1, false},
-				{0, false},
-				{1, false},
-			},
-		},
-		{
-			name:          "edgeSinceNegative",
-			expectInvalid: true,
-			config: &CreateTXGRange{
-				Since: &CreateTXGRangeBound{math.MaxUint64, &zfs.NilBool{B: false}},
-				Until: nil,
-			},
-			expectString: "(18446744073709551615,~",
-		},
-		{
-			name:          "edgeUntil",
-			expectInvalid: false,
-			config: &CreateTXGRange{
-				Until: &CreateTXGRangeBound{0, &zfs.NilBool{B: true}},
-			},
-			configAllowZeroCreateTXG: true,
-			expectString:             "~,0]",
-			expect: []testCaseExpectation{
-				{0, true},
-				{math.MaxUint64, false},
-				{1, false},
-			},
-		},
-		{
-			name:                     "edgeUntilNegative",
-			expectInvalid:            true,
-			configAllowZeroCreateTXG: true,
-			config: &CreateTXGRange{
-				Until: &CreateTXGRangeBound{0, &zfs.NilBool{B: false}},
-			},
-			expectString: "~,0)",
-		},
-	}
-
-	for _, tc := range tcs {
-		t.Run(tc.name, func(t *testing.T) {
-			require.True(t, tc.expectInvalid != (len(tc.expect) > 0), "invalid test config: must either expect invalid or have expectations: %s", tc.name)
-			require.NotEmpty(t, tc.expectString)
-			assert.Equal(t, tc.expectString, tc.config.String())
-
-			save := createTXGRangeBoundAllowCreateTXG0
-			createTXGRangeBoundAllowCreateTXG0 = tc.configAllowZeroCreateTXG
-			defer func() {
-				createTXGRangeBoundAllowCreateTXG0 = save
-			}()
-
-			if tc.expectInvalid {
-				t.Run(tc.name, func(t *testing.T) {
-					assert.Error(t, tc.config.Validate())
-				})
-			} else {
-				for i, e := range tc.expect {
-					t.Run(fmt.Sprint(i), func(t *testing.T) {
-						defer func() {
-							v := recover()
-							if v != nil {
-								t.Fatalf("should not panic: %T %v\n%s", v, v, debug.Stack())
-							}
-						}()
-						assert.Equal(t, e.expect, tc.config.Contains(e.input))
-					})
-				}
-			}
-		})
-	}
-
-}

endpoint/endpoint_zfs_helpers.go

@@ -1,6 +1,7 @@
 package endpoint
 
 import (
+	"context"
 	"fmt"
 	"regexp"
 	"strconv"
@@ -56,3 +57,53 @@ func parseJobAndGuidBookmarkName(fullname string, prefix string) (guid uint64, j
 
 	return guid, jobID, nil
 }
+
+func destroyBookmarksOlderThan(ctx context.Context, fs string, mostRecent *zfs.ZFSSendArgVersion, jobID JobID, filter func(shortname string) (accept bool)) (destroyed []zfs.FilesystemVersion, err error) {
+	if filter == nil {
+		panic(filter)
+	}
+
+	fsp, err := zfs.NewDatasetPath(fs)
+	if err != nil {
+		return nil, errors.Wrap(err, "invalid filesystem path")
+	}
+
+	mostRecentProps, err := mostRecent.ValidateExistsAndGetCheckedProps(ctx, fs)
+	if err != nil {
+		return nil, errors.Wrap(err, "validate most recent version argument")
+	}
+
+	stepBookmarks, err := zfs.ZFSListFilesystemVersions(fsp, zfs.FilterFromClosure(
+		func(t zfs.VersionType, name string) (accept bool, err error) {
+			if t != zfs.Bookmark {
+				return false, nil
+			}
+			return filter(name), nil
+		}))
+	if err != nil {
+		return nil, errors.Wrap(err, "list bookmarks")
+	}
+
+	// cut off all bookmarks prior to mostRecent's CreateTXG
+	var destroy []zfs.FilesystemVersion
+	for _, v := range stepBookmarks {
+		if v.Type != zfs.Bookmark {
+			panic("implementation error")
+		}
+		if !filter(v.Name) {
+			panic("inconsistent filter result")
+		}
+		if v.CreateTXG < mostRecentProps.CreateTXG {
+			destroy = append(destroy, v)
+		}
+	}
+
+	// FIXME use batch destroy, must adopt code to handle bookmarks
+	for _, v := range destroy {
+		if err := zfs.ZFSDestroyIdempotent(v.ToAbsPath(fsp)); err != nil {
+			return nil, errors.Wrap(err, "destroy bookmark")
+		}
+	}
+
+	return destroy, nil
+}

endpoint/endpoint_zfs_helpers_types.go

@@ -1,74 +0,0 @@
-package endpoint
-
-import (
-	"context"
-	"encoding/json"
-	"fmt"
-
-	"github.com/pkg/errors"
-
-	"github.com/zrepl/zrepl/zfs"
-)
-
-type bookmarkBasedAbstraction struct {
-	Type AbstractionType
-	FS   string
-	zfs.FilesystemVersion
-	JobID JobID
-}
-
-func (b bookmarkBasedAbstraction) GetType() AbstractionType { return b.Type }
-func (b bookmarkBasedAbstraction) GetFS() string            { return b.FS }
-func (b bookmarkBasedAbstraction) GetJobID() *JobID         { return &b.JobID }
-func (b bookmarkBasedAbstraction) GetFullPath() string {
-	return fmt.Sprintf("%s#%s", b.FS, b.Name) // TODO use zfs.FilesystemVersion.ToAbsPath
-}
-func (b bookmarkBasedAbstraction) MarshalJSON() ([]byte, error) {
-	return json.Marshal(AbstractionJSON{b})
-}
-func (b bookmarkBasedAbstraction) String() string {
-	return fmt.Sprintf("%s %s", b.Type, b.GetFullPath())
-}
-
-func (b bookmarkBasedAbstraction) GetFilesystemVersion() zfs.FilesystemVersion {
-	return b.FilesystemVersion
-}
-
-func (b bookmarkBasedAbstraction) Destroy(ctx context.Context) error {
-	if err := zfs.ZFSDestroyIdempotent(ctx, b.GetFullPath()); err != nil {
-		return errors.Wrapf(err, "destroy %s: zfs", b)
-	}
-	return nil
-}
-
-type holdBasedAbstraction struct {
-	Type AbstractionType
-	FS   string
-	zfs.FilesystemVersion
-	Tag   string
-	JobID JobID
-}
-
-func (h holdBasedAbstraction) GetType() AbstractionType { return h.Type }
-func (h holdBasedAbstraction) GetFS() string            { return h.FS }
-func (h holdBasedAbstraction) GetJobID() *JobID         { return &h.JobID }
-func (h holdBasedAbstraction) GetFullPath() string {
-	return fmt.Sprintf("%s@%s", h.FS, h.GetName()) // TODO use zfs.FilesystemVersion.ToAbsPath
-}
-func (h holdBasedAbstraction) MarshalJSON() ([]byte, error) {
-	return json.Marshal(AbstractionJSON{h})
-}
-func (h holdBasedAbstraction) String() string {
-	return fmt.Sprintf("%s %q on %s", h.Type, h.Tag, h.GetFullPath())
-}
-
-func (h holdBasedAbstraction) GetFilesystemVersion() zfs.FilesystemVersion {
-	return h.FilesystemVersion
-}
-
-func (h holdBasedAbstraction) Destroy(ctx context.Context) error {
-	if err := zfs.ZFSRelease(ctx, h.Tag, h.GetFullPath()); err != nil {
-		return errors.Wrapf(err, "release %s: zfs", h)
-	}
-	return nil
-}

endpoint/jobid.go

@@ -9,8 +9,8 @@ import (
 	"github.com/zrepl/zrepl/zfs"
 )
 
-// JobID instances returned by MakeJobID() guarantee their JobID.String()
-// can be used in ZFS dataset names and hold tags.
+// An instance of this type returned by MakeJobID guarantees
+// that that instance's JobID.String() can be used in a ZFS dataset name and hold tag.
 type JobID struct {
 	jid string
 }
@@ -21,7 +21,7 @@ func MakeJobID(s string) (JobID, error) {
 	}
 
 	if err := zfs.ComponentNamecheck(s); err != nil {
-		return JobID{}, errors.Wrap(err, "must be usable as a dataset path component")
+		return JobID{}, errors.Wrap(err, "muse be usable as a dataset path component")
 	}
 
 	if _, err := stepBookmarkNameImpl("pool/ds", 0xface601d, s); err != nil {
@@ -34,7 +34,7 @@ func MakeJobID(s string) (JobID, error) {
 	}
 
 	if _, err := lastReceivedHoldImpl(s); err != nil {
-		return JobID{}, errors.Wrap(err, "must be usable as a last-received-hold tag")
+		return JobID{}, errors.Wrap(err, "must be usabel as a last-recieved-hold tag")
 	}
 
 	// FIXME replication cursor bookmark name
@@ -57,7 +57,7 @@ func MustMakeJobID(s string) JobID {
 
 func (j JobID) expectInitialized() {
 	if j.jid == "" {
-		panic("use of uninitialized JobID")
+		panic("use of unitialized JobID")
 	}
 }
 

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/gdamore/tcell v1.2.0
 	github.com/go-logfmt/logfmt v0.4.0
 	github.com/go-sql-driver/mysql v1.4.1-0.20190907122137-b2c03bcae3d4
+	github.com/gogo/protobuf v1.1.1
 	github.com/golang/protobuf v1.3.2
 	github.com/google/uuid v1.1.1
 	github.com/jinzhu/copier v0.0.0-20170922082739-db4671f3a9b8
@@ -33,6 +34,5 @@ require (
 	golang.org/x/net v0.0.0-20190613194153-d28f0bde5980
 	golang.org/x/sync v0.0.0-20190423024810-112230192c58
 	golang.org/x/sys v0.0.0-20191026070338-33540a1f6037
-	gonum.org/v1/gonum v0.7.0 // indirect
 	google.golang.org/grpc v1.17.0
 )

go.sum

@@ -5,7 +5,6 @@ github.com/DATA-DOG/go-sqlmock v1.3.3/go.mod h1:f/Ixk793poVmq4qj/V1dPUg2JEAKC73Q
 github.com/OpenPeeDeeP/depguard v0.0.0-20180806142446-a69c782687b2/go.mod h1:7/4sitnI9YlQgTLLk734QlzXT8DuHVnAyztLplQjk+o=
 github.com/OpenPeeDeeP/depguard v0.0.0-20181229194401-1f388ab2d810/go.mod h1:7/4sitnI9YlQgTLLk734QlzXT8DuHVnAyztLplQjk+o=
 github.com/StackExchange/wmi v0.0.0-20180116203802-5d049714c4a6/go.mod h1:3eOhrUMpNV+6aFIbp5/iudMxNCF27Vw2OZgy4xEx0Fg=
-github.com/ajstarks/svgo v0.0.0-20180226025133-644b8db467af/go.mod h1:K08gAheRH3/J6wwsYMMT4xOr94bZjxIelGM0+d/wbFw=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
@@ -29,7 +28,6 @@ github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/fatih/color v1.6.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
 github.com/fatih/color v1.7.0 h1:DkWD4oS2D8LGGgTQ6IvwJJXSL5Vp2ffcQg58nFV38Ys=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
-github.com/fogleman/gg v1.2.1-0.20190220221249-0403632d5b90/go.mod h1:R/bRT+9gY/C5z7JzPU0zXsXHKM4/ayA+zqcVNZzPa1k=
 github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV9I=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/ftrvxmtrx/fd v0.0.0-20150925145434-c6d800382fff h1:zk1wwii7uXmI0znwU+lqg+wFL9G5+vm5I+9rv2let60=
@@ -70,9 +68,9 @@ github.com/go-toolsmith/strparse v1.0.0/go.mod h1:YI2nUKP9YGZnL/L1/DLFBfixrcjslW
 github.com/go-toolsmith/typep v0.0.0-20181030061450-d63dc7650676/go.mod h1:JSQCQMUPdRlMZFswiq3TGpNp1GMktqkR2Ns5AIQkATU=
 github.com/go-toolsmith/typep v1.0.0/go.mod h1:JSQCQMUPdRlMZFswiq3TGpNp1GMktqkR2Ns5AIQkATU=
 github.com/gobwas/glob v0.2.3/go.mod h1:d3Ez4x06l9bZtSvzIay5+Yzi0fmZzPgnTbPcKjJAkT8=
+github.com/gogo/protobuf v1.1.1 h1:72R+M5VuhED/KujmZVcIquuo8mBgX4oVda//DQb3PXo=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
 github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
-github.com/golang/freetype v0.0.0-20170609003504-e2365dfdc4a0/go.mod h1:E/TSTwGwJL78qG/PmXZO1EjYhfJinVAhrmmHX6Z8B9k=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b h1:VKtxabqXZkF25pY9ekfRL6a582T4P37/31XEstQ5p58=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
 github.com/golang/lint v0.0.0-20180702182130-06c8688daad7/go.mod h1:tluoj9z5200jBnyusfRPU2LqT6J+DAorxEvtC7LHB+E=
@@ -125,7 +123,6 @@ github.com/jinzhu/copier v0.0.0-20170922082739-db4671f3a9b8/go.mod h1:yL958EeXv8
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/json-iterator/go v1.1.7/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
 github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
-github.com/jung-kurt/gofpdf v1.0.3-0.20190309125859-24315acbbda5/go.mod h1:7Id9E/uU8ce6rXgefFLlgrJj/GYY22cpxn+r32jIOes=
 github.com/k0kubun/colorstring v0.0.0-20150214042306-9440f1994b88 h1:uC1QfSlInpQF+M0ao65imhwqKnz3Q2z/d8PWZRMQvDM=
 github.com/k0kubun/colorstring v0.0.0-20150214042306-9440f1994b88/go.mod h1:3w7q1U84EfirKl04SVQ/s7nPm1ZPhiXd34z40TNz36k=
 github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
@@ -293,11 +290,6 @@ golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnf
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190313024323-a1f597ede03a/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190325154230-a5d413f7728c/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
-golang.org/x/exp v0.0.0-20180321215751-8460e604b9de/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
-golang.org/x/exp v0.0.0-20180807140117-3d87b88a115f/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
-golang.org/x/exp v0.0.0-20190125153040-c74c464bbbf2 h1:y102fOLFqhV41b+4GPiJoa0k/x+pJcEi2/HB1Y5T6fU=
-golang.org/x/exp v0.0.0-20190125153040-c74c464bbbf2/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
-golang.org/x/image v0.0.0-20180708004352-c73c2afc3b81/go.mod h1:ux5Hcp/YLpHSI86hEcLt0YII63i6oz57MZXIpbrjZUs=
 golang.org/x/lint v0.0.0-20180702182130-06c8688daad7/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/net v0.0.0-20170915142106-8351a756f30f/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -337,24 +329,17 @@ golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/tools v0.0.0-20170915040203-e531a2a1c15f/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20180221164845-07fd8470d635/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
-golang.org/x/tools v0.0.0-20180525024113-a5b4c53f6e8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20180828015842-6cd1fcedba52/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20181117154741-2ddaf7f79a09/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20181205014116-22934f0fdb62/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190110163146-51295c7ec13a/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190121143147-24cd39ecf745/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
-golang.org/x/tools v0.0.0-20190206041539-40960b6deb8e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190213192042-740235f6c0d8/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190311215038-5c2858a9cfe5/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190322203728-c1a832b0ad89/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190521203540-521d6ed310dd/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
 golang.org/x/tools v0.0.0-20190524210228-3d17549cdc6b/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
-gonum.org/v1/gonum v0.0.0-20180816165407-929014505bf4/go.mod h1:Y+Yx5eoAFn32cQvJDxZx5Dpnq+c3wtXuadVZAcxbbBo=
-gonum.org/v1/gonum v0.7.0 h1:Hdks0L0hgznZLG9nzXb8vZ0rRvqNvAcgAp84y7Mwkgw=
-gonum.org/v1/gonum v0.7.0/go.mod h1:L02bwd0sqlsvRv41G7wGWFCsVNZFv/k1xzGIxeANHGM=
-gonum.org/v1/netlib v0.0.0-20190313105609-8cb42192e0e0/go.mod h1:wa6Ws7BG/ESfp6dHfk7C6KdzKA7wR7u/rKwOGE66zvw=
-gonum.org/v1/plot v0.0.0-20190515093506-e2840ee46a6b/go.mod h1:Wt8AAjI+ypCyYX3nZBvf6cAIx93T+c/OS2HFAYskSZc=
 google.golang.org/appengine v1.1.0 h1:igQkv0AAhEIvTEpD5LIpAfav2eeVO9HBTjvKHVJPRSs=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
 google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8 h1:Nw54tB0rB7hY/N0NQvRW8DG4Yk3Q6T9cu9RcFQDu1tc=
@@ -366,7 +351,6 @@ google.golang.org/grpc v1.17.0 h1:TRJYBgMclJvGYn2rIMjj+h9KtMt5r1Ij7ODVRIZkwhk=
 google.golang.org/grpc v1.17.0/go.mod h1:6QZJwpn2B+Zp71q/5VxRsJ6NXXVCE5NRUHRo+f3cWCs=
 gopkg.in/airbrake/gobrake.v2 v2.0.9/go.mod h1:/h5ZAUhDkGaJfjzjKLSjv6zCL6O0LLBxU4K+aSYdM/U=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
@@ -385,6 +369,5 @@ mvdan.cc/interfacer v0.0.0-20180901003855-c20040233aed/go.mod h1:Xkxe497xwlCKkIa
 mvdan.cc/lint v0.0.0-20170908181259-adc824a0674b/go.mod h1:2odslEg/xrtNQqCYg2/jCoyKnw3vv5biOc3JnIcYfL4=
 mvdan.cc/unparam v0.0.0-20190209190245-fbb59629db34/go.mod h1:H6SUd1XjIs+qQCyskXg5OFSrilMRUkD8ePJpHKDPaeY=
 mvdan.cc/unparam v0.0.0-20190310220240-1b9ccfa71afe/go.mod h1:BnhuWBAqxH3+J5bDybdxgw5ZfS+DsVd4iylsKQePN8o=
-rsc.io/pdf v0.1.1/go.mod h1:n8OzWcQ6Sp37PL01nO98y4iUCRdTGarVfzxY20ICaU4=
 sourcegraph.com/sqs/pbtypes v0.0.0-20180604144634-d3ebe8f20ae4/go.mod h1:ketZ/q3QxT9HOBeFhu6RdvsftgpsbFHBF5Cas6cDKZ0=
 sourcegraph.com/sqs/pbtypes v1.0.0/go.mod h1:3AciMUv4qUuRHRHhOG4TZOB+72GdPVz5k+c648qsFS4=

main.go

@@ -17,7 +17,7 @@ func init() {
 	cli.AddSubcommand(client.PprofCmd)
 	cli.AddSubcommand(client.TestCmd)
 	cli.AddSubcommand(client.MigrateCmd)
-	cli.AddSubcommand(client.ZFSAbstractionsCmd)
+	cli.AddSubcommand(client.HoldsCmd)
 }
 
 func main() {

platformtest/harness/harness.go

@@ -172,21 +172,21 @@ type testCaseResult struct {
 func runTestCase(ctx *platformtest.Context, ex platformtest.Execer, c tests.Case) *testCaseResult {
 
 	// run case
-	var panicked = false
+	var paniced = false
 	var panicValue interface{} = nil
 	var panicStack error
 	func() {
 		defer func() {
 			if item := recover(); item != nil {
 				panicValue = item
-				panicked = true
+				paniced = true
 				panicStack = errors.Errorf("panic while running test: %v", panicValue)
 			}
 		}()
 		c(ctx)
 	}()
 
-	if panicked {
+	if paniced {
 		switch panicValue {
 		case platformtest.SkipNowSentinel:
 			return &testCaseResult{skipped: true}

platformtest/platformtest_ops.go

@@ -24,7 +24,6 @@ type Stmt interface {
 type Op string
 
 const (
-	Comment         Op = "#"
 	AssertExists    Op = "!E"
 	AssertNotExists Op = "!N"
 	Add             Op = "+"
@@ -103,26 +102,6 @@ func (o *SnapOp) Run(ctx context.Context, e Execer) error {
 	}
 }
 
-type BookmarkOp struct {
-	Op       Op
-	Existing string
-	Bookmark string
-}
-
-func (o *BookmarkOp) Run(ctx context.Context, e Execer) error {
-	switch o.Op {
-	case Add:
-		return e.RunExpectSuccessNoOutput(ctx, "zfs", "bookmark", o.Existing, o.Bookmark)
-	case Del:
-		if o.Existing != "" {
-			panic("existing must be empty for destroy, got " + o.Existing)
-		}
-		return e.RunExpectSuccessNoOutput(ctx, "zfs", "destroy", o.Bookmark)
-	default:
-		panic(o.Op)
-	}
-}
-
 type RunOp struct {
 	RootDS string
 	Script string
@@ -201,8 +180,8 @@ func splitQuotedWords(data []byte, atEOF bool) (advance int, token []byte, err e
 				// unescaped quote, end of this string
 				// remove backslash-escapes
 				withBackslash := data[begin+1 : end]
-				withoutBackslash := bytes.Replace(withBackslash, []byte("\\\""), []byte("\""), -1)
-				return end + 1, withoutBackslash, nil
+				withoutBaskslash := bytes.Replace(withBackslash, []byte("\\\""), []byte("\""), -1)
+				return end + 1, withoutBaskslash, nil
 			} else {
 				// continue to next quote
 				end += 1
@@ -276,26 +255,16 @@ nextLine:
 			op = AssertExists
 		case string(AssertNotExists):
 			op = AssertNotExists
-		case string(Comment):
-			op = Comment
-			continue
 		default:
 			return nil, &LineError{scan.Text(), fmt.Sprintf("invalid op %q", comps.Text())}
 		}
 
-		// FS / SNAP / BOOKMARK
+		// FS / SNAP
 		if err := expectMoreTokens(); err != nil {
 			return nil, err
 		}
 		if strings.ContainsAny(comps.Text(), "@") {
 			stmts = append(stmts, &SnapOp{Op: op, Path: fmt.Sprintf("%s/%s", rootds, comps.Text())})
-		} else if strings.ContainsAny(comps.Text(), "#") {
-			bookmark := fmt.Sprintf("%s/%s", rootds, comps.Text())
-			if err := expectMoreTokens(); err != nil {
-				return nil, err
-			}
-			existing := fmt.Sprintf("%s/%s", rootds, comps.Text())
-			stmts = append(stmts, &BookmarkOp{Op: op, Existing: existing, Bookmark: bookmark})
 		} else {
 			// FS
 			fs := comps.Text()

platformtest/platformtest_zpool.go

@@ -34,7 +34,7 @@ func (a ZpoolCreateArgs) Validate() error {
 		return errors.Errorf("Mountpoint must be an absolute path to a directory")
 	}
 	if a.PoolName == "" {
-		return errors.Errorf("PoolName must not be empty")
+		return errors.Errorf("PoolName must not be emtpy")
 	}
 	return nil
 }
@@ -45,7 +45,7 @@ func CreateOrReplaceZpool(ctx context.Context, e Execer, args ZpoolCreateArgs) (
 	}
 
 	// export pool if it already exists (idempotence)
-	if _, err := zfs.ZFSGetRawAnySource(ctx, args.PoolName, []string{"name"}); err != nil {
+	if _, err := zfs.ZFSGetRawAnySource(args.PoolName, []string{"name"}); err != nil {
 		if _, ok := err.(*zfs.DatasetDoesNotExist); ok {
 			// we'll create it shortly
 		} else {

platformtest/tests/batchRelease.go

@@ -0,0 +1,154 @@
+package tests
+
+import (
+	"fmt"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/zrepl/zrepl/platformtest"
+	"github.com/zrepl/zrepl/zfs"
+)
+
+type rollupReleaseExpectTags struct {
+	Snap  string
+	Holds map[string]bool
+}
+
+func rollupReleaseTest(ctx *platformtest.Context, cb func(fs string) []rollupReleaseExpectTags) {
+
+	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+	DESTROYROOT
+	CREATEROOT
+	+  "foo bar"
+	+  "foo bar@1"
+	+  "foo bar@2"
+	+  "foo bar@3"
+	+  "foo bar@4"
+	+  "foo bar@5"
+	+  "foo bar@6"
+	R  zfs hold zrepl_platformtest   "${ROOTDS}/foo bar@1"
+	R  zfs hold zrepl_platformtest_2 "${ROOTDS}/foo bar@2"
+	R  zfs hold zrepl_platformtest   "${ROOTDS}/foo bar@3"
+	R  zfs hold zrepl_platformtest   "${ROOTDS}/foo bar@5"
+	R  zfs hold zrepl_platformtest   "${ROOTDS}/foo bar@6"
+	R  zfs bookmark "${ROOTDS}/foo bar@5" "${ROOTDS}/foo bar#5"
+`)
+
+	fs := fmt.Sprintf("%s/foo bar", ctx.RootDataset)
+
+	expTags := cb(fs)
+
+	for _, exp := range expTags {
+		holds, err := zfs.ZFSHolds(ctx, fs, exp.Snap)
+		if err != nil {
+			panic(err)
+		}
+		for _, h := range holds {
+			if e, ok := exp.Holds[h]; !ok || !e {
+				panic(fmt.Sprintf("tag %q on snap %q not expected", h, exp.Snap))
+			}
+		}
+	}
+
+}
+
+func RollupReleaseIncluding(ctx *platformtest.Context) {
+	rollupReleaseTest(ctx, func(fs string) []rollupReleaseExpectTags {
+		guid5, err := zfs.ZFSGetGUID(fs, "@5")
+		require.NoError(ctx, err)
+
+		err = zfs.ZFSReleaseAllOlderAndIncludingGUID(ctx, fs, guid5, "zrepl_platformtest")
+		require.NoError(ctx, err)
+
+		return []rollupReleaseExpectTags{
+			{"1", map[string]bool{}},
+			{"2", map[string]bool{"zrepl_platformtest_2": true}},
+			{"3", map[string]bool{}},
+			{"4", map[string]bool{}},
+			{"5", map[string]bool{}},
+			{"6", map[string]bool{"zrepl_platformtest": true}},
+		}
+	})
+}
+
+func RollupReleaseExcluding(ctx *platformtest.Context) {
+	rollupReleaseTest(ctx, func(fs string) []rollupReleaseExpectTags {
+		guid5, err := zfs.ZFSGetGUID(fs, "@5")
+		require.NoError(ctx, err)
+
+		err = zfs.ZFSReleaseAllOlderThanGUID(ctx, fs, guid5, "zrepl_platformtest")
+		require.NoError(ctx, err)
+
+		return []rollupReleaseExpectTags{
+			{"1", map[string]bool{}},
+			{"2", map[string]bool{"zrepl_platformtest_2": true}},
+			{"3", map[string]bool{}},
+			{"4", map[string]bool{}},
+			{"5", map[string]bool{"zrepl_platformtest": true}},
+			{"6", map[string]bool{"zrepl_platformtest": true}},
+		}
+	})
+}
+
+func RollupReleaseMostRecentIsBookmarkWithoutSnapshot(ctx *platformtest.Context) {
+	rollupReleaseTest(ctx, func(fs string) []rollupReleaseExpectTags {
+		guid5, err := zfs.ZFSGetGUID(fs, "#5")
+		require.NoError(ctx, err)
+
+		err = zfs.ZFSRelease(ctx, "zrepl_platformtest", fs+"@5")
+		require.NoError(ctx, err)
+
+		err = zfs.ZFSDestroy(fs + "@5")
+		require.NoError(ctx, err)
+
+		err = zfs.ZFSReleaseAllOlderAndIncludingGUID(ctx, fs, guid5, "zrepl_platformtest")
+		require.NoError(ctx, err)
+
+		return []rollupReleaseExpectTags{
+			{"1", map[string]bool{}},
+			{"2", map[string]bool{"zrepl_platformtest_2": true}},
+			{"3", map[string]bool{}},
+			{"4", map[string]bool{}},
+			// {"5", map[string]bool{}}, doesn't exist
+			{"6", map[string]bool{"zrepl_platformtest": true}},
+		}
+	})
+}
+
+func RollupReleaseMostRecentIsBookmarkAndSnapshotStillExists(ctx *platformtest.Context) {
+	rollupReleaseTest(ctx, func(fs string) []rollupReleaseExpectTags {
+		guid5, err := zfs.ZFSGetGUID(fs, "#5")
+		require.NoError(ctx, err)
+
+		err = zfs.ZFSReleaseAllOlderAndIncludingGUID(ctx, fs, guid5, "zrepl_platformtest")
+		require.NoError(ctx, err)
+
+		return []rollupReleaseExpectTags{
+			{"1", map[string]bool{}},
+			{"2", map[string]bool{"zrepl_platformtest_2": true}},
+			{"3", map[string]bool{}},
+			{"4", map[string]bool{}},
+			{"5", map[string]bool{}},
+			{"6", map[string]bool{"zrepl_platformtest": true}},
+		}
+	})
+}
+
+func RollupReleaseMostRecentDoesntExist(ctx *platformtest.Context) {
+	rollupReleaseTest(ctx, func(fs string) []rollupReleaseExpectTags {
+
+		const nonexistentGuid = 0 // let's take our chances...
+		err := zfs.ZFSReleaseAllOlderAndIncludingGUID(ctx, fs, nonexistentGuid, "zrepl_platformtest")
+		require.Error(ctx, err)
+		require.Contains(ctx, err.Error(), "cannot find snapshot or bookmark with guid 0")
+
+		return []rollupReleaseExpectTags{
+			{"1", map[string]bool{"zrepl_platformtest": true}},
+			{"2", map[string]bool{"zrepl_platformtest_2": true}},
+			{"3", map[string]bool{"zrepl_platformtest": true}},
+			{"4", map[string]bool{"zrepl_platformtest": true}},
+			{"5", map[string]bool{"zrepl_platformtest": true}},
+			{"6", map[string]bool{"zrepl_platformtest": true}},
+		}
+	})
+}

platformtest/tests/getNonexistent.go

@@ -17,14 +17,14 @@ func GetNonexistent(ctx *platformtest.Context) {
 	`)
 
 	// test raw
-	_, err := zfs.ZFSGetRawAnySource(ctx, fmt.Sprintf("%s/foo bar", ctx.RootDataset), []string{"name"})
+	_, err := zfs.ZFSGetRawAnySource(fmt.Sprintf("%s/foo bar", ctx.RootDataset), []string{"name"})
 	if err != nil {
 		panic(err)
 	}
 
 	// test nonexistent filesystem
 	nonexistent := fmt.Sprintf("%s/nonexistent filesystem", ctx.RootDataset)
-	props, err := zfs.ZFSGetRawAnySource(ctx, nonexistent, []string{"name"})
+	props, err := zfs.ZFSGetRawAnySource(nonexistent, []string{"name"})
 	if err == nil {
 		panic(props)
 	}
@@ -37,7 +37,7 @@ func GetNonexistent(ctx *platformtest.Context) {
 
 	// test nonexistent snapshot
 	nonexistent = fmt.Sprintf("%s/foo bar@non existent", ctx.RootDataset)
-	props, err = zfs.ZFSGetRawAnySource(ctx, nonexistent, []string{"name"})
+	props, err = zfs.ZFSGetRawAnySource(nonexistent, []string{"name"})
 	if err == nil {
 		panic(props)
 	}
@@ -50,7 +50,7 @@ func GetNonexistent(ctx *platformtest.Context) {
 
 	// test nonexistent bookmark
 	nonexistent = fmt.Sprintf("%s/foo bar#non existent", ctx.RootDataset)
-	props, err = zfs.ZFSGetRawAnySource(ctx, nonexistent, []string{"name"})
+	props, err = zfs.ZFSGetRawAnySource(nonexistent, []string{"name"})
 	if err == nil {
 		panic(props)
 	}

platformtest/tests/helpers.go

@@ -5,7 +5,6 @@ import (
 	"math/rand"
 	"os"
 	"path"
-	"sort"
 	"strings"
 
 	"github.com/stretchr/testify/require"
@@ -15,8 +14,8 @@ import (
 	"github.com/zrepl/zrepl/zfs"
 )
 
-func sendArgVersion(ctx *platformtest.Context, fs, relName string) zfs.ZFSSendArgVersion {
-	guid, err := zfs.ZFSGetGUID(ctx, fs, relName)
+func sendArgVersion(fs, relName string) zfs.ZFSSendArgVersion {
+	guid, err := zfs.ZFSGetGUID(fs, relName)
 	if err != nil {
 		panic(err)
 	}
@@ -26,14 +25,6 @@ func sendArgVersion(ctx *platformtest.Context, fs, relName string) zfs.ZFSSendAr
 	}
 }
 
-func fsversion(ctx *platformtest.Context, fs, relname string) zfs.FilesystemVersion {
-	v, err := zfs.ZFSGetFilesystemVersion(ctx, fs+relname)
-	if err != nil {
-		panic(err)
-	}
-	return v
-}
-
 func mustDatasetPath(fs string) *zfs.DatasetPath {
 	p, err := zfs.NewDatasetPath(fs)
 	if err != nil {
@@ -42,7 +33,7 @@ func mustDatasetPath(fs string) *zfs.DatasetPath {
 	return p
 }
 
-func mustSnapshot(ctx *platformtest.Context, snap string) {
+func mustSnapshot(snap string) {
 	if err := zfs.EntityNamecheck(snap, zfs.EntityTypeSnapshot); err != nil {
 		panic(err)
 	}
@@ -50,16 +41,16 @@ func mustSnapshot(ctx *platformtest.Context, snap string) {
 	if len(comps) != 2 {
 		panic(comps)
 	}
-	err := zfs.ZFSSnapshot(ctx, mustDatasetPath(comps[0]), comps[1], false)
+	err := zfs.ZFSSnapshot(mustDatasetPath(comps[0]), comps[1], false)
 	if err != nil {
 		panic(err)
 	}
 }
 
-func mustGetFilesystemVersion(ctx *platformtest.Context, snapOrBookmark string) zfs.FilesystemVersion {
-	v, err := zfs.ZFSGetFilesystemVersion(ctx, snapOrBookmark)
+func mustGetProps(entity string) zfs.ZFSPropCreateTxgAndGuidProps {
+	props, err := zfs.ZFSGetCreateTXGAndGuid(entity)
 	check(err)
-	return v
+	return props
 }
 
 func check(err error) {
@@ -87,7 +78,7 @@ type dummySnapshotSituation struct {
 }
 
 type resumeSituation struct {
-	sendArgs         zfs.ZFSSendArgsUnvalidated
+	sendArgs         zfs.ZFSSendArgs
 	recvOpts         zfs.RecvOptions
 	sendErr, recvErr error
 	recvErrDecoded   *zfs.RecvFailedWithResumeTokenErr
@@ -96,7 +87,7 @@ type resumeSituation struct {
 func makeDummyDataSnapshots(ctx *platformtest.Context, sendFS string) (situation dummySnapshotSituation) {
 
 	situation.sendFS = sendFS
-	sendFSMount, err := zfs.ZFSGetMountpoint(ctx, sendFS)
+	sendFSMount, err := zfs.ZFSGetMountpoint(sendFS)
 	require.NoError(ctx, err)
 	require.True(ctx, sendFSMount.Mounted)
 
@@ -104,33 +95,28 @@ func makeDummyDataSnapshots(ctx *platformtest.Context, sendFS string) (situation
 	situation.dummyDataLen = dummyLen
 
 	writeDummyData(path.Join(sendFSMount.Mountpoint, "dummy_data"), dummyLen)
-	mustSnapshot(ctx, sendFS+"@a snapshot")
-	snapA := sendArgVersion(ctx, sendFS, "@a snapshot")
+	mustSnapshot(sendFS + "@a snapshot")
+	snapA := sendArgVersion(sendFS, "@a snapshot")
 	situation.snapA = &snapA
 
 	writeDummyData(path.Join(sendFSMount.Mountpoint, "dummy_data"), dummyLen)
-	mustSnapshot(ctx, sendFS+"@b snapshot")
-	snapB := sendArgVersion(ctx, sendFS, "@b snapshot")
+	mustSnapshot(sendFS + "@b snapshot")
+	snapB := sendArgVersion(sendFS, "@b snapshot")
 	situation.snapB = &snapB
 
 	return situation
 }
 
-func makeResumeSituation(ctx *platformtest.Context, src dummySnapshotSituation, recvFS string, sendArgs zfs.ZFSSendArgsUnvalidated, recvOptions zfs.RecvOptions) *resumeSituation {
+func makeResumeSituation(ctx *platformtest.Context, src dummySnapshotSituation, recvFS string, sendArgs zfs.ZFSSendArgs, recvOptions zfs.RecvOptions) *resumeSituation {
 
 	situation := &resumeSituation{}
 
 	situation.sendArgs = sendArgs
 	situation.recvOpts = recvOptions
-	require.True(ctx, recvOptions.SavePartialRecvState, "this method would be pointless otherwise")
+	require.True(ctx, recvOptions.SavePartialRecvState, "this method would be pointeless otherwise")
 	require.Equal(ctx, sendArgs.FS, src.sendFS)
-	sendArgsValidated, err := sendArgs.Validate(ctx)
-	situation.sendErr = err
-	if err != nil {
-		return situation
-	}
 
-	copier, err := zfs.ZFSSend(ctx, sendArgsValidated)
+	copier, err := zfs.ZFSSend(ctx, sendArgs)
 	situation.sendErr = err
 	if err != nil {
 		return situation
@@ -151,26 +137,3 @@ func makeResumeSituation(ctx *platformtest.Context, src dummySnapshotSituation,
 
 	return situation
 }
-
-func versionRelnamesSorted(versions []zfs.FilesystemVersion) []string {
-	var vstrs []string
-	for _, v := range versions {
-		vstrs = append(vstrs, v.RelName())
-	}
-	sort.Strings(vstrs)
-	return vstrs
-}
-
-func datasetToStringSortedTrimPrefix(prefix *zfs.DatasetPath, paths []*zfs.DatasetPath) []string {
-	var pstrs []string
-	for _, p := range paths {
-		trimmed := p.Copy()
-		trimmed.TrimPrefix(prefix)
-		if trimmed.Length() == 0 {
-			continue
-		}
-		pstrs = append(pstrs, trimmed.ToString())
-	}
-	sort.Strings(pstrs)
-	return pstrs
-}

platformtest/tests/idempotentBookmark.go

@@ -19,22 +19,22 @@ func IdempotentBookmark(ctx *platformtest.Context) {
 
 	fs := fmt.Sprintf("%s/foo bar", ctx.RootDataset)
 
-	asnap := sendArgVersion(ctx, fs, "@a snap")
-	anotherSnap := sendArgVersion(ctx, fs, "@another snap")
+	asnap := sendArgVersion(fs, "@a snap")
+	anotherSnap := sendArgVersion(fs, "@another snap")
 
-	err := zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
+	err := zfs.ZFSBookmark(fs, asnap, "a bookmark")
 	if err != nil {
 		panic(err)
 	}
 
 	// do it again, should be idempotent
-	err = zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
+	err = zfs.ZFSBookmark(fs, asnap, "a bookmark")
 	if err != nil {
 		panic(err)
 	}
 
 	// should fail for another snapshot
-	err = zfs.ZFSBookmark(ctx, fs, anotherSnap, "a bookmark")
+	err = zfs.ZFSBookmark(fs, anotherSnap, "a bookmark")
 	if err == nil {
 		panic(err)
 	}
@@ -43,12 +43,12 @@ func IdempotentBookmark(ctx *platformtest.Context) {
 	}
 
 	// destroy the snapshot
-	if err := zfs.ZFSDestroy(ctx, fmt.Sprintf("%s@a snap", fs)); err != nil {
+	if err := zfs.ZFSDestroy(fmt.Sprintf("%s@a snap", fs)); err != nil {
 		panic(err)
 	}
 
 	// do it again, should fail with special error type
-	err = zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
+	err = zfs.ZFSBookmark(fs, asnap, "a bookmark")
 	if err == nil {
 		panic(err)
 	}

platformtest/tests/idempotentDestroy.go

@@ -18,8 +18,8 @@ func IdempotentDestroy(ctx *platformtest.Context) {
 	`)
 
 	fs := fmt.Sprintf("%s/foo bar", ctx.RootDataset)
-	asnap := sendArgVersion(ctx, fs, "@a snap")
-	err := zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
+	asnap := sendArgVersion(fs, "@a snap")
+	err := zfs.ZFSBookmark(fs, asnap, "a bookmark")
 	if err != nil {
 		panic(err)
 	}
@@ -41,17 +41,17 @@ func IdempotentDestroy(ctx *platformtest.Context) {
 			log.Printf("SUBBEGIN testing idempotent destroy %q for path %q", c.description, c.path)
 
 			log.Println("destroy existing")
-			err = zfs.ZFSDestroy(ctx, c.path)
+			err = zfs.ZFSDestroy(c.path)
 			if err != nil {
 				panic(err)
 			}
 			log.Println("destroy again, non-idempotently, must error")
-			err = zfs.ZFSDestroy(ctx, c.path)
+			err = zfs.ZFSDestroy(c.path)
 			if _, ok := err.(*zfs.DatasetDoesNotExist); !ok {
 				panic(fmt.Sprintf("%T: %s", err, err))
 			}
 			log.Println("destroy again, idempotently, must not error")
-			err = zfs.ZFSDestroyIdempotent(ctx, c.path)
+			err = zfs.ZFSDestroyIdempotent(c.path)
 			if err != nil {
 				panic(err)
 			}
@@ -62,12 +62,12 @@ func IdempotentDestroy(ctx *platformtest.Context) {
 	}
 
 	// also test idempotent destroy for cases where the parent dataset does not exist
-	err = zfs.ZFSDestroyIdempotent(ctx, fmt.Sprintf("%s/not foo bar@nonexistent snapshot", ctx.RootDataset))
+	err = zfs.ZFSDestroyIdempotent(fmt.Sprintf("%s/not foo bar@nonexistent snapshot", ctx.RootDataset))
 	if err != nil {
 		panic(err)
 	}
 
-	err = zfs.ZFSDestroyIdempotent(ctx, fmt.Sprintf("%s/not foo bar#nonexistent bookmark", ctx.RootDataset))
+	err = zfs.ZFSDestroyIdempotent(fmt.Sprintf("%s/not foo bar#nonexistent bookmark", ctx.RootDataset))
 	if err != nil {
 		panic(err)
 	}

platformtest/tests/idempotentHold.go

@@ -22,7 +22,7 @@ func IdempotentHold(ctx *platformtest.Context) {
 	`)
 
 	fs := fmt.Sprintf("%s/foo bar", ctx.RootDataset)
-	v1 := fsversion(ctx, fs, "@1")
+	v1 := sendArgVersion(fs, "@1")
 
 	tag := "zrepl_platformtest"
 	err := zfs.ZFSHold(ctx, fs, v1, tag)
@@ -34,4 +34,14 @@ func IdempotentHold(ctx *platformtest.Context) {
 	if err != nil {
 		panic(err)
 	}
+
+	vnonexistent := zfs.ZFSSendArgVersion{
+		RelName: "@nonexistent",
+		GUID:    0xbadf00d,
+	}
+	err = zfs.ZFSHold(ctx, fs, vnonexistent, tag)
+	if err == nil {
+		panic("still expecting error for nonexistent snapshot")
+	}
+
 }

platformtest/tests/listFilesystemVersions.go

@@ -1,179 +0,0 @@
-package tests
-
-import (
-	"fmt"
-	"sort"
-	"strings"
-
-	"github.com/stretchr/testify/require"
-
-	"github.com/zrepl/zrepl/platformtest"
-	"github.com/zrepl/zrepl/zfs"
-)
-
-func ListFilesystemVersionsTypeFilteringAndPrefix(t *platformtest.Context) {
-	platformtest.Run(t, platformtest.PanicErr, t.RootDataset, `
-		DESTROYROOT
-		CREATEROOT
-		+  "foo bar"
-		+  "foo bar@foo 1"
-		+  "foo bar#foo 1" "foo bar@foo 1"
-		+  "foo bar#bookfoo 1" "foo bar@foo 1"
-		+  "foo bar@foo 2"
-		+  "foo bar#foo 2" "foo bar@foo 2"
-		+  "foo bar#bookfoo 2" "foo bar@foo 2"
-		+  "foo bar@blup 1"
-		+  "foo bar#blup 1" "foo bar@blup 1"
-		+  "foo bar@ foo with leading whitespace"
-
-		# repeat the whole thing for a child dataset to make sure we disable recursion
-
-		+  "foo bar/child dataset"
-		+  "foo bar/child dataset@foo 1"
-		+  "foo bar/child dataset#foo 1" "foo bar/child dataset@foo 1"
-		+  "foo bar/child dataset#bookfoo 1" "foo bar/child dataset@foo 1"
-		+  "foo bar/child dataset@foo 2"
-		+  "foo bar/child dataset#foo 2" "foo bar/child dataset@foo 2"
-		+  "foo bar/child dataset#bookfoo 2" "foo bar/child dataset@foo 2"
-		+  "foo bar/child dataset@blup 1"
-		+  "foo bar/child dataset#blup 1" "foo bar/child dataset@blup 1"
-		+  "foo bar/child dataset@ foo with leading whitespace"
-	`)
-
-	fs := fmt.Sprintf("%s/foo bar", t.RootDataset)
-
-	// no options := all types
-	vs, err := zfs.ZFSListFilesystemVersions(mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
-	require.NoError(t, err)
-	require.Equal(t, []string{
-		"#blup 1", "#bookfoo 1", "#bookfoo 2", "#foo 1", "#foo 2",
-		"@ foo with leading whitespace", "@blup 1", "@foo 1", "@foo 2",
-	}, versionRelnamesSorted(vs))
-
-	// just snapshots
-	vs, err = zfs.ZFSListFilesystemVersions(mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{
-		Types: zfs.Snapshots,
-	})
-	require.NoError(t, err)
-	require.Equal(t, []string{"@ foo with leading whitespace", "@blup 1", "@foo 1", "@foo 2"}, versionRelnamesSorted(vs))
-
-	// just bookmarks
-	vs, err = zfs.ZFSListFilesystemVersions(mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{
-		Types: zfs.Bookmarks,
-	})
-	require.NoError(t, err)
-	require.Equal(t, []string{"#blup 1", "#bookfoo 1", "#bookfoo 2", "#foo 1", "#foo 2"}, versionRelnamesSorted(vs))
-
-	// just with prefix foo
-	vs, err = zfs.ZFSListFilesystemVersions(mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{
-		ShortnamePrefix: "foo",
-	})
-	require.NoError(t, err)
-	require.Equal(t, []string{"#foo 1", "#foo 2", "@foo 1", "@foo 2"}, versionRelnamesSorted(vs))
-
-}
-
-func ListFilesystemVersionsZeroExistIsNotAnError(t *platformtest.Context) {
-	platformtest.Run(t, platformtest.PanicErr, t.RootDataset, `
-	DESTROYROOT
-	CREATEROOT
-	+ "foo bar"
-	`)
-
-	fs := fmt.Sprintf("%s/foo bar", t.RootDataset)
-
-	vs, err := zfs.ZFSListFilesystemVersions(mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
-	require.Empty(t, vs)
-	require.NoError(t, err)
-	dsne, ok := err.(*zfs.DatasetDoesNotExist)
-	require.True(t, ok)
-	require.Equal(t, fs, dsne.Path)
-}
-
-func ListFilesystemVersionsFilesystemNotExist(t *platformtest.Context) {
-	platformtest.Run(t, platformtest.PanicErr, t.RootDataset, `
-	DESTROYROOT
-	CREATEROOT
-	`)
-
-	nonexistentFS := fmt.Sprintf("%s/not existent", t.RootDataset)
-
-	vs, err := zfs.ZFSListFilesystemVersions(mustDatasetPath(nonexistentFS), zfs.ListFilesystemVersionsOptions{})
-	require.Empty(t, vs)
-	require.Error(t, err)
-	t.Logf("err = %T\n%s", err, err)
-	dsne, ok := err.(*zfs.DatasetDoesNotExist)
-	require.True(t, ok)
-	require.Equal(t, nonexistentFS, dsne.Path)
-}
-
-func ListFilesystemVersionsUserrefs(t *platformtest.Context) {
-	platformtest.Run(t, platformtest.PanicErr, t.RootDataset, `
-	DESTROYROOT
-	CREATEROOT
-	+ "foo bar"
-	+ "foo bar@snap 1"
-	+ "foo bar#snap 1" "foo bar@snap 1"
-	+ "foo bar@snap 2"
-	+ "foo bar#snap 2" "foo bar@snap 2"
-	R zfs hold zrepl_platformtest "${ROOTDS}/foo bar@snap 2"
-	+ "foo bar@snap 3"
-	+ "foo bar#snap 3" "foo bar@snap 3"
-	R zfs hold zrepl_platformtest "${ROOTDS}/foo bar@snap 3"
-	R zfs hold zrepl_platformtest_second_hold "${ROOTDS}/foo bar@snap 3"
-	+ "foo bar@snap 4"
-	+ "foo bar#snap 4" "foo bar@snap 4"
-
-
-	+ "foo bar/child datset"
-	+ "foo bar/child datset@snap 1"
-	+ "foo bar/child datset#snap 1" "foo bar/child datset@snap 1"
-	+ "foo bar/child datset@snap 2"
-	+ "foo bar/child datset#snap 2" "foo bar/child datset@snap 2"
-	R zfs hold zrepl_platformtest "${ROOTDS}/foo bar/child datset@snap 2"
-	+ "foo bar/child datset@snap 3"
-	+ "foo bar/child datset#snap 3" "foo bar/child datset@snap 3"
-	R zfs hold zrepl_platformtest "${ROOTDS}/foo bar/child datset@snap 3"
-	R zfs hold zrepl_platformtest_second_hold "${ROOTDS}/foo bar/child datset@snap 3"
-	+ "foo bar/child datset@snap 4"
-	+ "foo bar/child datset#snap 4" "foo bar/child datset@snap 4"
-	`)
-
-	fs := fmt.Sprintf("%s/foo bar", t.RootDataset)
-
-	vs, err := zfs.ZFSListFilesystemVersions(mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
-	require.NoError(t, err)
-
-	type expectation struct {
-		relName  string
-		userrefs zfs.OptionUint64
-	}
-
-	expect := []expectation{
-		{"#snap 1", zfs.OptionUint64{Valid: false}},
-		{"#snap 2", zfs.OptionUint64{Valid: false}},
-		{"#snap 3", zfs.OptionUint64{Valid: false}},
-		{"#snap 4", zfs.OptionUint64{Valid: false}},
-		{"@snap 1", zfs.OptionUint64{Value: 0, Valid: true}},
-		{"@snap 2", zfs.OptionUint64{Value: 1, Valid: true}},
-		{"@snap 3", zfs.OptionUint64{Value: 2, Valid: true}},
-		{"@snap 4", zfs.OptionUint64{Value: 0, Valid: true}},
-	}
-
-	sort.Slice(vs, func(i, j int) bool {
-		return strings.Compare(vs[i].RelName(), vs[j].RelName()) < 0
-	})
-
-	var expectRelNames []string
-	for _, e := range expect {
-		expectRelNames = append(expectRelNames, e.relName)
-	}
-
-	require.Equal(t, expectRelNames, versionRelnamesSorted(vs))
-
-	for i, e := range expect {
-		require.Equal(t, e.relName, vs[i].RelName())
-		require.Equal(t, e.userrefs, vs[i].UserRefs)
-	}
-
-}

platformtest/tests/listFilesystems.go

@@ -1,34 +0,0 @@
-package tests
-
-import (
-	"strings"
-
-	"github.com/stretchr/testify/require"
-
-	"github.com/zrepl/zrepl/platformtest"
-	"github.com/zrepl/zrepl/zfs"
-)
-
-func ListFilesystemsNoFilter(t *platformtest.Context) {
-	platformtest.Run(t, platformtest.PanicErr, t.RootDataset, `
-		DESTROYROOT
-		CREATEROOT
-		R  zfs create -V 10M "${ROOTDS}/bar baz"
-		+  "foo bar"
-		+  "foo bar/bar blup"
-		+  "foo bar/blah"
-		R  zfs create -V 10M "${ROOTDS}/foo bar/blah/a volume"
-	`)
-
-	fss, err := zfs.ZFSListMapping(t, zfs.NoFilter())
-	require.NoError(t, err)
-	var onlyTestPool []*zfs.DatasetPath
-	for _, fs := range fss {
-		if strings.HasPrefix(fs.ToString(), t.RootDataset) {
-			onlyTestPool = append(onlyTestPool, fs)
-		}
-	}
-	onlyTestPoolStr := datasetToStringSortedTrimPrefix(mustDatasetPath(t.RootDataset), onlyTestPool)
-	require.Equal(t, []string{"bar baz", "foo bar", "foo bar/bar blup", "foo bar/blah", "foo bar/blah/a volume"}, onlyTestPoolStr)
-
-}

platformtest/tests/replicationCursor.go

@@ -5,7 +5,6 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-
 	"github.com/zrepl/zrepl/endpoint"
 	"github.com/zrepl/zrepl/platformtest"
 	"github.com/zrepl/zrepl/zfs"
@@ -32,7 +31,7 @@ func ReplicationCursor(ctx *platformtest.Context) {
 	}
 
 	fs := ds.ToString()
-	snap := fsversion(ctx, fs, "@1 with space")
+	snap := sendArgVersion(fs, "@1 with space")
 
 	destroyed, err := endpoint.MoveReplicationCursor(ctx, fs, &snap, jobid)
 	if err != nil {
@@ -40,7 +39,7 @@ func ReplicationCursor(ctx *platformtest.Context) {
 	}
 	assert.Empty(ctx, destroyed)
 
-	snapProps, err := zfs.ZFSGetFilesystemVersion(ctx, snap.FullPath(fs))
+	snapProps, err := zfs.ZFSGetCreateTXGAndGuid(snap.FullPath(fs))
 	if err != nil {
 		panic(err)
 	}
@@ -57,13 +56,13 @@ func ReplicationCursor(ctx *platformtest.Context) {
 	}
 
 	// try moving
-	cursor1BookmarkName, err := endpoint.ReplicationCursorBookmarkName(fs, snap.Guid, jobid)
+	cursor1BookmarkName, err := endpoint.ReplicationCursorBookmarkName(fs, snap.GUID, jobid)
 	require.NoError(ctx, err)
 
-	snap2 := fsversion(ctx, fs, "@2 with space")
+	snap2 := sendArgVersion(fs, "@2 with space")
 	destroyed, err = endpoint.MoveReplicationCursor(ctx, fs, &snap2, jobid)
 	require.NoError(ctx, err)
 	require.Equal(ctx, 1, len(destroyed))
-	require.Equal(ctx, endpoint.AbstractionReplicationCursorBookmarkV2, destroyed[0].GetType())
-	require.Equal(ctx, cursor1BookmarkName, destroyed[0].GetName())
+	require.Equal(ctx, zfs.Bookmark, destroyed[0].Type)
+	require.Equal(ctx, cursor1BookmarkName, destroyed[0].Name)
 }

platformtest/tests/resumableRecvAndTokenHandling.go

@@ -25,7 +25,7 @@ func ResumableRecvAndTokenHandling(ctx *platformtest.Context) {
 
 	src := makeDummyDataSnapshots(ctx, sendFS)
 
-	s := makeResumeSituation(ctx, src, recvFS, zfs.ZFSSendArgsUnvalidated{
+	s := makeResumeSituation(ctx, src, recvFS, zfs.ZFSSendArgs{
 		FS:          sendFS,
 		To:          src.snapA,
 		Encrypted:   &zfs.NilBool{B: false},
@@ -40,7 +40,7 @@ func ResumableRecvAndTokenHandling(ctx *platformtest.Context) {
 		require.True(ctx, ok)
 
 		// we know that support on sendFS implies support on recvFS
-		// => assert that if we don't support resumed recv, the method returns ""
+		// => asser that if we don't support resumed recv, the method returns ""
 		tok, err := zfs.ZFSGetReceiveResumeTokenOrEmptyStringIfNotSupported(ctx, mustDatasetPath(recvFS))
 		check(err)
 		require.Equal(ctx, "", tok)

platformtest/tests/resumeTokenParsing.go

@@ -16,7 +16,7 @@ type resumeTokenTest struct {
 
 func (rtt *resumeTokenTest) Test(t *platformtest.Context) {
 
-	resumeSendSupported, err := zfs.ResumeSendSupported(t)
+	resumeSendSupported, err := zfs.ResumeSendSupported()
 	if err != nil {
 		t.Errorf("cannot determine whether resume supported: %T %s", err, err)
 		t.FailNow()

platformtest/tests/sendArgsValidation.go

@@ -23,9 +23,9 @@ func SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden(ctx *platformt
 	`)
 
 	fs := fmt.Sprintf("%s/send er", ctx.RootDataset)
-	props := mustGetFilesystemVersion(ctx, fs+"@a snap")
+	props := mustGetProps(fs + "@a snap")
 
-	sendArgs, err := zfs.ZFSSendArgsUnvalidated{
+	sendArgs := zfs.ZFSSendArgs{
 		FS: fs,
 		To: &zfs.ZFSSendArgVersion{
 			RelName: "@a snap",
@@ -33,15 +33,10 @@ func SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden(ctx *platformt
 		},
 		Encrypted:   &zfs.NilBool{B: true},
 		ResumeToken: "",
-	}.Validate(ctx)
-
-	var stream *zfs.ReadCloserCopier
+	}
+	stream, err := zfs.ZFSSend(ctx, sendArgs)
 	if err == nil {
-		stream, err = zfs.ZFSSend(ctx, sendArgs) // no shadow
-		if err == nil {
-			defer stream.Close()
-		}
-		// fallthrough
+		defer stream.Close()
 	}
 
 	if expectNotSupportedErr {
@@ -63,7 +58,7 @@ func SendArgsValidationResumeTokenEncryptionMismatchForbidden(ctx *platformtest.
 	if !supported {
 		ctx.SkipNow()
 	}
-	supported, err = zfs.ResumeSendSupported(ctx)
+	supported, err = zfs.ResumeSendSupported()
 	check(err)
 	if !supported {
 		ctx.SkipNow()
@@ -81,7 +76,7 @@ func SendArgsValidationResumeTokenEncryptionMismatchForbidden(ctx *platformtest.
 
 	src := makeDummyDataSnapshots(ctx, sendFS)
 
-	unencS := makeResumeSituation(ctx, src, unencRecvFS, zfs.ZFSSendArgsUnvalidated{
+	unencS := makeResumeSituation(ctx, src, unencRecvFS, zfs.ZFSSendArgs{
 		FS:        sendFS,
 		To:        src.snapA,
 		Encrypted: &zfs.NilBool{B: false}, // !
@@ -90,7 +85,7 @@ func SendArgsValidationResumeTokenEncryptionMismatchForbidden(ctx *platformtest.
 		SavePartialRecvState: true,
 	})
 
-	encS := makeResumeSituation(ctx, src, encRecvFS, zfs.ZFSSendArgsUnvalidated{
+	encS := makeResumeSituation(ctx, src, encRecvFS, zfs.ZFSSendArgs{
 		FS:        sendFS,
 		To:        src.snapA,
 		Encrypted: &zfs.NilBool{B: true}, // !
@@ -102,10 +97,16 @@ func SendArgsValidationResumeTokenEncryptionMismatchForbidden(ctx *platformtest.
 	// threat model: use of a crafted resume token that requests an unencrypted send
 	//               but send args require encrypted send
 	{
-		var maliciousSend zfs.ZFSSendArgsUnvalidated = encS.sendArgs
+		var maliciousSend zfs.ZFSSendArgs = encS.sendArgs
 		maliciousSend.ResumeToken = unencS.recvErrDecoded.ResumeTokenRaw
 
-		_, err := maliciousSend.Validate(ctx)
+		stream, err := zfs.ZFSSend(ctx, maliciousSend)
+		if err == nil {
+			defer stream.Close()
+		}
+		require.Nil(ctx, stream)
+		require.Error(ctx, err)
+		ctx.Logf("send err: %T %s", err, err)
 		validationErr, ok := err.(*zfs.ZFSSendArgsValidationError)
 		require.True(ctx, ok)
 		require.Equal(ctx, validationErr.What, zfs.ZFSSendArgsResumeTokenMismatch)
@@ -116,13 +117,17 @@ func SendArgsValidationResumeTokenEncryptionMismatchForbidden(ctx *platformtest.
 		require.Equal(ctx, mismatchError.What, zfs.ZFSSendArgsResumeTokenMismatchEncryptionNotSet)
 	}
 
-	// threat model: use of a crafted resume token that requests an encrypted send
+	// threat model: use of a crafted resume token that requests an encryped send
 	//               but send args require unencrypted send
 	{
-		var maliciousSend zfs.ZFSSendArgsUnvalidated = unencS.sendArgs
+		var maliciousSend zfs.ZFSSendArgs = unencS.sendArgs
 		maliciousSend.ResumeToken = encS.recvErrDecoded.ResumeTokenRaw
 
-		_, err := maliciousSend.Validate(ctx)
+		stream, err := zfs.ZFSSend(ctx, maliciousSend)
+		if err == nil {
+			defer stream.Close()
+		}
+		require.Nil(ctx, stream)
 		require.Error(ctx, err)
 		ctx.Logf("send err: %T %s", err, err)
 		validationErr, ok := err.(*zfs.ZFSSendArgsValidationError)
@@ -144,7 +149,7 @@ func SendArgsValidationResumeTokenDifferentFilesystemForbidden(ctx *platformtest
 	if !supported {
 		ctx.SkipNow()
 	}
-	supported, err = zfs.ResumeSendSupported(ctx)
+	supported, err = zfs.ResumeSendSupported()
 	check(err)
 	if !supported {
 		ctx.SkipNow()
@@ -164,7 +169,7 @@ func SendArgsValidationResumeTokenDifferentFilesystemForbidden(ctx *platformtest
 	src1 := makeDummyDataSnapshots(ctx, sendFS1)
 	src2 := makeDummyDataSnapshots(ctx, sendFS2)
 
-	rs := makeResumeSituation(ctx, src1, recvFS, zfs.ZFSSendArgsUnvalidated{
+	rs := makeResumeSituation(ctx, src1, recvFS, zfs.ZFSSendArgs{
 		FS:        sendFS1,
 		To:        src1.snapA,
 		Encrypted: &zfs.NilBool{B: false},
@@ -175,7 +180,7 @@ func SendArgsValidationResumeTokenDifferentFilesystemForbidden(ctx *platformtest
 
 	// threat model: forged resume token tries to steal a full send of snapA on fs2 by
 	//               presenting a resume token for full send of snapA on fs1
-	var maliciousSend zfs.ZFSSendArgsUnvalidated = zfs.ZFSSendArgsUnvalidated{
+	var maliciousSend zfs.ZFSSendArgs = zfs.ZFSSendArgs{
 		FS: sendFS2,
 		To: &zfs.ZFSSendArgVersion{
 			RelName: src2.snapA.RelName,
@@ -184,7 +189,12 @@ func SendArgsValidationResumeTokenDifferentFilesystemForbidden(ctx *platformtest
 		Encrypted:   &zfs.NilBool{B: false},
 		ResumeToken: rs.recvErrDecoded.ResumeTokenRaw,
 	}
-	_, err = maliciousSend.Validate(ctx)
+
+	stream, err := zfs.ZFSSend(ctx, maliciousSend)
+	if err == nil {
+		defer stream.Close()
+	}
+	require.Nil(ctx, stream)
 	require.Error(ctx, err)
 	ctx.Logf("send err: %T %s", err, err)
 	validationErr, ok := err.(*zfs.ZFSSendArgsValidationError)

platformtest/tests/tests.go

@@ -18,6 +18,11 @@ var Cases = []Case{
 	UndestroyableSnapshotParsing,
 	GetNonexistent,
 	ReplicationCursor,
+	RollupReleaseIncluding,
+	RollupReleaseExcluding,
+	RollupReleaseMostRecentIsBookmarkWithoutSnapshot,
+	RollupReleaseMostRecentIsBookmarkAndSnapshotStillExists,
+	RollupReleaseMostRecentDoesntExist,
 	IdempotentHold,
 	IdempotentBookmark,
 	IdempotentDestroy,
@@ -26,9 +31,4 @@ var Cases = []Case{
 	SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden,
 	SendArgsValidationResumeTokenEncryptionMismatchForbidden,
 	SendArgsValidationResumeTokenDifferentFilesystemForbidden,
-	ListFilesystemVersionsTypeFilteringAndPrefix,
-	ListFilesystemVersionsFilesystemNotExist,
-	ListFilesystemVersionsFilesystemNotExist,
-	ListFilesystemVersionsUserrefs,
-	ListFilesystemsNoFilter,
 }

platformtest/tests/undestroyableSnapshotParsing.go

@@ -20,7 +20,7 @@ func UndestroyableSnapshotParsing(t *platformtest.Context) {
 		R  zfs hold zrepl_platformtest "${ROOTDS}/foo bar@4 5 6"
 	`)
 
-	err := zfs.ZFSDestroy(t, fmt.Sprintf("%s/foo bar@1 2 3,4 5 6,7 8 9", t.RootDataset))
+	err := zfs.ZFSDestroy(fmt.Sprintf("%s/foo bar@1 2 3,4 5 6,7 8 9", t.RootDataset))
 	if err == nil {
 		panic("expecting destroy error due to hold")
 	}

replication/design.md

@@ -18,32 +18,7 @@ Hence, when trying to map algorithm to implementation, use the code in package `
       * the recv-side fs doesn't get newer snapshots than `to` in the meantime
         * guaranteed because the zrepl model of the receiver assumes ownership of the filesystems it receives into
           * if that assumption is broken, future replication attempts will fail with a conflict
-* The [Algorithm for Planning and Executing Replication of an Filesystems](#zrepl-algo-filesystem) is a design draft and not used.
-  However, there were some noteworthy lessons learned when implementing the algorithm for a single step:
-  * In order to avoid leaking `step-hold`s and `step-bookmarks`, if the replication planner is invoked a second time after a replication step (either initial or incremental) has been attempted but failed to completed, the replication planner must
-    * A) either guarantee that it will resume that replication step, and continue as if nothing happened or
-    * B) release the step holds and bookmarks and clear the partially received state on the sending side.
-  * Option A is what we want to do: we use the step algorithm to achieve resumability in the first place!
-  * Option B is not done by zrepl except if the sending side doesn't support resuming.
-    In that case however, we need not release any holds since the behavior is to re-start the send
-    from the beginning.
-  * However, there is one **edge-case to Option A for initial replication**:
-    * If initial replication "`full a`" fails without leaving resumable state, the step holds on the sending side are still present, which makes sense because
-      * a) we want resumability and
-      * b) **the sending side cannot immediately be informed post-failure whether the initial replication left any state that would mandate keeping the step hold**, because the network connection might have failed.
-    * Thus, the sending side must keep the step hold for "`full a`" until it knows more.
-    * In the current implementation, it knows more when the next replication attempt is made, the planner is invoked, the diffing algorithm run, and the `HintMostRecentCommonAncestor` RPC is sent by the active side, communicating the most recent common version shared betwen sender and receiver.
-    * At this point, the sender can safely throw away any step holds with CreateTXG's older than that version.
-  * **The `step-hold`, `step-bookmark`, `last-received-hold` and `replication-cursor` abstractions are currently local concepts of package `endpoint` and not part of the replication protocol**
-    * This is not necessarilty the best design decision and should be revisited some point:
-      * The (quite expensive) `HintMostRecentCommonAncestor` RPC impl on the sender would not be necessary if step holds were part of the replication protocol:
-        * We only need the `HintMostRecentCommonAncestor` info for the aforementioned edge-case during initial replication, where the receive is aborted without any partial received state being stored on the receiver (due to network failure, wrong zfs invocation, bad permissions, etc):
-        * **The replication planner does not know about the step holds, thus it cannot deterministically pick up where it left of (right at the start of the last failing initial replication).**
-        * Instead, it will seem like no prior invocation happened at all, and it will apply its policy for initial replication to pick a new `full b != full a`, **thereby leaking the step holds of `full a`**.
-        * In contrast, if the replication planner created the step holds and knew about them, it could use the step holds as an indicator where it left off and re-start from there (of course asserting that the thereby inferred step is compatible with the state of the receiving side).
-      * (What we do in zrepl right now is to hard-code the initial replication policy, and hard-code that assumption in `endpoint.ListStale` as well.)
-      * The cummulative cleanup done in `HintMostRecentCommonAncestor` provides a nice self-healing aspect, though.
-
+* The [Algorithm for Planning and Executing Replication of an Filesystems](#zrepl-algo-filesystem) is a design draft and not used
 * We also have [Notes on Planning and Executing Replication of Multiple Filesystems](#zrepl-algo-multiple-filesystems-notes)
 
 ---
@@ -60,7 +35,7 @@ The algorithm **ensures resumability** of the replication step in presence of
 * network failures at any time
 * other instances of this algorithm executing the same step in parallel (e.g. concurrent replication to different destinations)
 
-To accomplish this goal, the algorithm **assumes ownership of parts of the ZFS hold tag namespace and the bookmark namespace**:
+To accomplish this goal, the algorithm **assumes ownersip of parts of the ZFS hold tag namespace and the bookmark namespace**:
 * holds with prefix `zrepl_STEP` on any snapshot are reserved for zrepl
 * bookmarks with prefix `zrepl_STEP` are reserved for zrepl
 
@@ -80,7 +55,7 @@ The replication step (full `to` send or `from => to` send) is *complete* iff the
 Specifically, the algorithm may be invoked with the same `from` and `to` arguments, and potentially a `resume_token`, after a temporary (like network-related) failure:
 **Unless permanent errors occur, repeated invocations of the algorithm with updated resume token will converge monotonically (but not strictly monotonically) toward completion.**
 
-Note that the mere existence of `to` on the receiving side does not constitute completion, since there may still be post-recv actions to be performed on sender and receiver.
+Note that the mere existence of `to` on the receiving side does not constitue completion, since there may still be post-recv actions to be performed on sender and receiver.
 
 #### Job and Job ID
 This algorithm supports that *multiple* instance of it run in parallel on the *same* step (full `to` / `from => to` pair).
@@ -142,7 +117,7 @@ Recv-side: no-op
     # => doesn't work, because zfs recv is implemented as a `clone` internally, that's exactly what we want
     ```
 
-- if recv-side `to` exists, goto cleanup-phase (no replication to do)
+- if recv-side `to` exists, goto cleaup-phase (no replication to do)
   
 - `to` cannot be destroyed while being received, because it isn't visible as a snapshot yet (it isn't yet one after all)
 
@@ -167,7 +142,7 @@ Network failures during replication can be recovered from using resumable send &
 - Network failure during the replication
   - send-side `from` and `to` are still present due to zfs holds
   - recv-side `from` is still present because the partial receive state prevents its destruction (see prepare-phase)
-  - if recv-side has a resume token, the resume token will continue to work on the sender because `from`s and `to` are still present
+  - if recv-side hasa resume token, the resume token will continue to work on the sender because `from`s and `to` are still present
 - Network failure at the end of the replication step stream transmission
   - Variant A: failure from the sender's perspective, success from the receiver's perspective
     - receive-side `to` doesn't have a hold and could be destroyed anytime
@@ -246,7 +221,7 @@ It builds a diff between the sender and receiver filesystem bookmarks+snapshots
 In case of conflict, the algorithm errors out with a conflict description that can be used to manually or automatically resolve the conflict.
 Otherwise, the algorithm builds a list of replication steps that are then worked on sequentially by the "Algorithm for a Single Replication Step".
 
-The algorithm ensures that a plan can be executed exactly as planned by acquiring appropriate zfs holds.
+The algorithm ensures that a plan can be executed exactly as planned by aquiring appropriate zfs holds.
 The algorithm can be configured to retry a plan when encountering non-permanent errors (e.g. network errors).
 However, permanent errors result in the plan being cancelled.
 
@@ -287,22 +262,22 @@ If fast-forward is not possible, produce a conflict description and ERROR OUT.<b
 TODOs:
 - make it configurable what snapshots are included in the list (i.e. every one we see, only most recent, at least one every X hours, ...)
 
-**Ensure that we will be able to carry out all steps** by acquiring holds or fsstep bookmarks on the sending side
+**Ensure that we will be able to carry out all steps** by aquiring holds or fsstep bookmarks on the sending side
 - `idempotent_hold([s.to for s in STEPS], zrepl_FS_J_${jobid})`
 - `if STEPS[0].from != nil: idempotent_FSSTEP_bookmark(STEPS[0].from, zrepl_FSSTEP_bm_G_${STEPS[0].from.guid}_J_${jobid})` 
   
 **Determine which steps have not been completed (`uncompleted_steps`)** (we might be in an second invocation of this algorithm after a network failure and some steps might already be done):
 - `res_tok := receiver.ResumeToken(fs)`
 - `rmrfsv := receiver.MostRecentFilesystemVersion(fs)`
-- if `res_tok !=  nil`: ensure that `res_tok` has a corresponding step in `STEPS`, otherwise ERROR OUT
-- if `rmrfsv != nil`: ensure that `res_tok` has a corresponding step in `STEPS`, otherwise ERROR OUT
+- if `res_tok !=  nil`: ensure that `res_tok` has a correspondinng step in `STEPS`, otherwise ERROR OUT
+- if `rmrfsv != nil`: ensure that `res_tok` has a correspondinng step in `STEPS`, otherwise ERROR OUT
 - if `(res_token != nil && rmrfsv != nil)`: ensure that `res_tok` is the subsequent step to the one we found for `rmrfsv`
 - if both are nil, we are at the beginning, `uncompleted_steps = STEPS` and goto next block
 - `rstep := if res_tok != nil { res_tok } else { rmrfsv }`
 - `uncompleted_steps := STEPS[find_step_idx(STEPS, rstep).expect("must exist, checked above"):]`
   - Note that we do not explicitly check for the completion of prior replication steps.
     All we care about is what needs to be done from `rstep`.
-  - This is intentional and necessary because we cumulatively release all holds and step bookmarks made for steps that precede a just-completed step (see next paragraph)
+  - This is intentional and necessary because we cummutatively release all holds and step bookmarks made for steps that preceed a just-completed step (see next paragraph)
 
 **Execute uncompleted steps**<br/>
 Invoke the "Algorithm for a Single Replication Step" for each step in `uncompleted_steps`.

replication/driver/replication_driver.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"os"
 	"sort"
 	"strings"
 	"sync"
@@ -14,6 +15,7 @@ import (
 	"google.golang.org/grpc/status"
 
 	"github.com/zrepl/zrepl/replication/report"
+	"github.com/zrepl/zrepl/tracing"
 	"github.com/zrepl/zrepl/util/chainlock"
 	"github.com/zrepl/zrepl/util/envconst"
 )
@@ -206,6 +208,7 @@ func Do(ctx context.Context, planner Planner) (ReportFunc, WaitFunc) {
 			}
 			run.attempts = append(run.attempts, cur)
 			run.l.DropWhile(func() {
+				ctx := tracing.Child(ctx, fmt.Sprintf("attempt#%d", ano)) // shadow
 				cur.do(ctx, prev)
 			})
 			prev = cur
@@ -281,7 +284,7 @@ func Do(ctx context.Context, planner Planner) (ReportFunc, WaitFunc) {
 }
 
 func (a *attempt) do(ctx context.Context, prev *attempt) {
-	pfss, err := a.planner.Plan(ctx)
+	pfss, err := a.planner.Plan(tracing.Child(ctx, "plan"))
 	errTime := time.Now()
 	defer a.l.Lock().Unlock()
 	if err != nil {
@@ -352,7 +355,7 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 			}
 		}
 	}
-	// invariant: prevs contains an entry for each unambiguous correspondence
+	// invariant: prevs contains an entry for each unambigious correspondence
 
 	stepQueue := newStepQueue()
 	defer stepQueue.Start(1)() // TODO parallel replication
@@ -361,7 +364,7 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 		fssesDone.Add(1)
 		go func(f *fs) {
 			defer fssesDone.Done()
-			f.do(ctx, stepQueue, prevs[f])
+			f.do(tracing.Child(ctx, f.fs.ReportInfo().Name), stepQueue, prevs[f])
 		}(f)
 	}
 	a.l.DropWhile(func() {
@@ -372,6 +375,8 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 
 func (fs *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 
+	fmt.Fprintf(os.Stderr, "CHILD STACK: %v\n", tracing.GetStack(ctx))
+
 	defer fs.l.Lock().Unlock()
 
 	// get planned steps from replication logic
@@ -399,7 +404,7 @@ func (fs *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 		}
 		fs.planned.steps = append(fs.planned.steps, step)
 	}
-	debug("initial len(fs.planned.steps) = %d", len(fs.planned.steps))
+	debug("iniital len(fs.planned.steps) = %d", len(fs.planned.steps))
 
 	// for not-first attempts, only allow fs.planned.steps
 	// up to including the originally planned target snapshot

replication/driver/replication_driver_logging.go

@@ -8,12 +8,12 @@ import (
 
 type Logger = logger.Logger
 
-type contextKey int
+type contexKey int
 
-const contextKeyLogger contextKey = iota + 1
+const contexKeyLogger contexKey = iota + 1
 
 func getLog(ctx context.Context) Logger {
-	l, ok := ctx.Value(contextKeyLogger).(Logger)
+	l, ok := ctx.Value(contexKeyLogger).(Logger)
 	if !ok {
 		l = logger.NewNullLogger()
 	}
@@ -21,5 +21,5 @@ func getLog(ctx context.Context) Logger {
 }
 
 func WithLogger(ctx context.Context, log Logger) context.Context {
-	return context.WithValue(ctx, contextKeyLogger, log)
+	return context.WithValue(ctx, contexKeyLogger, log)
 }

replication/driver/replication_driver_test.go

@@ -174,7 +174,7 @@ func TestReplication(t *testing.T) {
 	waitBegin := time.Now()
 	wait(true)
 	waitDuration := time.Since(waitBegin)
-	assert.True(t, waitDuration < 10*time.Millisecond, "%v", waitDuration) // and that's gracious
+	assert.True(t, waitDuration < 10*time.Millisecond, "%v", waitDuration) // and that's gratious
 
 	prev, err := json.Marshal(reports[0])
 	require.NoError(t, err)

replication/driver/replication_stepqueue_test.go

@@ -14,7 +14,7 @@ import (
 )
 
 // FIXME: this test relies on timing and is thus rather flaky
-// (relies on scheduler responsiveness of < 500ms)
+// (relies on scheduler responsivity of < 500ms)
 func TestPqNotconcurrent(t *testing.T) {
 	var ctr uint32
 	q := newStepQueue()

replication/logic/diff/diff_test.go

@@ -17,7 +17,7 @@ func fsvlist(fsv ...string) (r []*FilesystemVersion) {
 	r = make([]*FilesystemVersion, len(fsv))
 	for i, f := range fsv {
 
-		// parse the id from fsvlist. it is used to derive Guid,CreateTXG and Creation attrs
+		// parse the id from fsvlist. it is used to derivce Guid,CreateTXG and Creation attrs
 		split := strings.Split(f, ",")
 		if len(split) != 2 {
 			panic("invalid fsv spec")
@@ -114,7 +114,7 @@ func TestIncrementalPath_BookmarkSupport(t *testing.T) {
 		assert.Equal(t, l("#a,1", "@b,2"), path)
 	})
 
-	// bookmarks are stripped from IncrementalPath (cannot send incrementally)
+	// boomarks are stripped from IncrementalPath (cannot send incrementally)
 	doTest(l("@a,1"), l("#a,1", "#b,2", "@c,3"), func(path []*FilesystemVersion, conflict error) {
 		assert.Equal(t, l("#a,1", "@c,3"), path)
 	})

replication/logic/pdu/pdu.pb.go

@@ -46,7 +46,7 @@ func (x Tri) String() string {
 	return proto.EnumName(Tri_name, int32(x))
 }
 func (Tri) EnumDescriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{0}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{0}
 }
 
 type FilesystemVersion_VersionType int32
@@ -69,7 +69,7 @@ func (x FilesystemVersion_VersionType) String() string {
 	return proto.EnumName(FilesystemVersion_VersionType_name, int32(x))
 }
 func (FilesystemVersion_VersionType) EnumDescriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{5, 0}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{5, 0}
 }
 
 type ListFilesystemReq struct {
@@ -82,7 +82,7 @@ func (m *ListFilesystemReq) Reset()         { *m = ListFilesystemReq{} }
 func (m *ListFilesystemReq) String() string { return proto.CompactTextString(m) }
 func (*ListFilesystemReq) ProtoMessage()    {}
 func (*ListFilesystemReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{0}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{0}
 }
 func (m *ListFilesystemReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ListFilesystemReq.Unmarshal(m, b)
@@ -113,7 +113,7 @@ func (m *ListFilesystemRes) Reset()         { *m = ListFilesystemRes{} }
 func (m *ListFilesystemRes) String() string { return proto.CompactTextString(m) }
 func (*ListFilesystemRes) ProtoMessage()    {}
 func (*ListFilesystemRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{1}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{1}
 }
 func (m *ListFilesystemRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ListFilesystemRes.Unmarshal(m, b)
@@ -154,7 +154,7 @@ func (m *Filesystem) Reset()         { *m = Filesystem{} }
 func (m *Filesystem) String() string { return proto.CompactTextString(m) }
 func (*Filesystem) ProtoMessage()    {}
 func (*Filesystem) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{2}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{2}
 }
 func (m *Filesystem) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_Filesystem.Unmarshal(m, b)
@@ -213,7 +213,7 @@ func (m *ListFilesystemVersionsReq) Reset()         { *m = ListFilesystemVersion
 func (m *ListFilesystemVersionsReq) String() string { return proto.CompactTextString(m) }
 func (*ListFilesystemVersionsReq) ProtoMessage()    {}
 func (*ListFilesystemVersionsReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{3}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{3}
 }
 func (m *ListFilesystemVersionsReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ListFilesystemVersionsReq.Unmarshal(m, b)
@@ -251,7 +251,7 @@ func (m *ListFilesystemVersionsRes) Reset()         { *m = ListFilesystemVersion
 func (m *ListFilesystemVersionsRes) String() string { return proto.CompactTextString(m) }
 func (*ListFilesystemVersionsRes) ProtoMessage()    {}
 func (*ListFilesystemVersionsRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{4}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{4}
 }
 func (m *ListFilesystemVersionsRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ListFilesystemVersionsRes.Unmarshal(m, b)
@@ -293,7 +293,7 @@ func (m *FilesystemVersion) Reset()         { *m = FilesystemVersion{} }
 func (m *FilesystemVersion) String() string { return proto.CompactTextString(m) }
 func (*FilesystemVersion) ProtoMessage()    {}
 func (*FilesystemVersion) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{5}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{5}
 }
 func (m *FilesystemVersion) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_FilesystemVersion.Unmarshal(m, b)
@@ -371,7 +371,7 @@ func (m *SendReq) Reset()         { *m = SendReq{} }
 func (m *SendReq) String() string { return proto.CompactTextString(m) }
 func (*SendReq) ProtoMessage()    {}
 func (*SendReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{6}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{6}
 }
 func (m *SendReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_SendReq.Unmarshal(m, b)
@@ -445,7 +445,7 @@ func (m *Property) Reset()         { *m = Property{} }
 func (m *Property) String() string { return proto.CompactTextString(m) }
 func (*Property) ProtoMessage()    {}
 func (*Property) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{7}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{7}
 }
 func (m *Property) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_Property.Unmarshal(m, b)
@@ -496,7 +496,7 @@ func (m *SendRes) Reset()         { *m = SendRes{} }
 func (m *SendRes) String() string { return proto.CompactTextString(m) }
 func (*SendRes) ProtoMessage()    {}
 func (*SendRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{8}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{8}
 }
 func (m *SendRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_SendRes.Unmarshal(m, b)
@@ -548,7 +548,7 @@ func (m *SendCompletedReq) Reset()         { *m = SendCompletedReq{} }
 func (m *SendCompletedReq) String() string { return proto.CompactTextString(m) }
 func (*SendCompletedReq) ProtoMessage()    {}
 func (*SendCompletedReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{9}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{9}
 }
 func (m *SendCompletedReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_SendCompletedReq.Unmarshal(m, b)
@@ -585,7 +585,7 @@ func (m *SendCompletedRes) Reset()         { *m = SendCompletedRes{} }
 func (m *SendCompletedRes) String() string { return proto.CompactTextString(m) }
 func (*SendCompletedRes) ProtoMessage()    {}
 func (*SendCompletedRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{10}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{10}
 }
 func (m *SendCompletedRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_SendCompletedRes.Unmarshal(m, b)
@@ -608,7 +608,7 @@ var xxx_messageInfo_SendCompletedRes proto.InternalMessageInfo
 type ReceiveReq struct {
 	Filesystem string             `protobuf:"bytes,1,opt,name=Filesystem,proto3" json:"Filesystem,omitempty"`
 	To         *FilesystemVersion `protobuf:"bytes,2,opt,name=To,proto3" json:"To,omitempty"`
-	// If true, the receiver should clear the resume token before performing the
+	// If true, the receiver should clear the resume token before perfoming the
 	// zfs recv of the stream in the request
 	ClearResumeToken     bool     `protobuf:"varint,3,opt,name=ClearResumeToken,proto3" json:"ClearResumeToken,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
@@ -620,7 +620,7 @@ func (m *ReceiveReq) Reset()         { *m = ReceiveReq{} }
 func (m *ReceiveReq) String() string { return proto.CompactTextString(m) }
 func (*ReceiveReq) ProtoMessage()    {}
 func (*ReceiveReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{11}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{11}
 }
 func (m *ReceiveReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ReceiveReq.Unmarshal(m, b)
@@ -671,7 +671,7 @@ func (m *ReceiveRes) Reset()         { *m = ReceiveRes{} }
 func (m *ReceiveRes) String() string { return proto.CompactTextString(m) }
 func (*ReceiveRes) ProtoMessage()    {}
 func (*ReceiveRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{12}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{12}
 }
 func (m *ReceiveRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ReceiveRes.Unmarshal(m, b)
@@ -704,7 +704,7 @@ func (m *DestroySnapshotsReq) Reset()         { *m = DestroySnapshotsReq{} }
 func (m *DestroySnapshotsReq) String() string { return proto.CompactTextString(m) }
 func (*DestroySnapshotsReq) ProtoMessage()    {}
 func (*DestroySnapshotsReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{13}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{13}
 }
 func (m *DestroySnapshotsReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_DestroySnapshotsReq.Unmarshal(m, b)
@@ -750,7 +750,7 @@ func (m *DestroySnapshotRes) Reset()         { *m = DestroySnapshotRes{} }
 func (m *DestroySnapshotRes) String() string { return proto.CompactTextString(m) }
 func (*DestroySnapshotRes) ProtoMessage()    {}
 func (*DestroySnapshotRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{14}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{14}
 }
 func (m *DestroySnapshotRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_DestroySnapshotRes.Unmarshal(m, b)
@@ -795,7 +795,7 @@ func (m *DestroySnapshotsRes) Reset()         { *m = DestroySnapshotsRes{} }
 func (m *DestroySnapshotsRes) String() string { return proto.CompactTextString(m) }
 func (*DestroySnapshotsRes) ProtoMessage()    {}
 func (*DestroySnapshotsRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{15}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{15}
 }
 func (m *DestroySnapshotsRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_DestroySnapshotsRes.Unmarshal(m, b)
@@ -833,7 +833,7 @@ func (m *ReplicationCursorReq) Reset()         { *m = ReplicationCursorReq{} }
 func (m *ReplicationCursorReq) String() string { return proto.CompactTextString(m) }
 func (*ReplicationCursorReq) ProtoMessage()    {}
 func (*ReplicationCursorReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{16}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{16}
 }
 func (m *ReplicationCursorReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ReplicationCursorReq.Unmarshal(m, b)
@@ -874,7 +874,7 @@ func (m *ReplicationCursorRes) Reset()         { *m = ReplicationCursorRes{} }
 func (m *ReplicationCursorRes) String() string { return proto.CompactTextString(m) }
 func (*ReplicationCursorRes) ProtoMessage()    {}
 func (*ReplicationCursorRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{17}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{17}
 }
 func (m *ReplicationCursorRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ReplicationCursorRes.Unmarshal(m, b)
@@ -1010,7 +1010,7 @@ func (m *PingReq) Reset()         { *m = PingReq{} }
 func (m *PingReq) String() string { return proto.CompactTextString(m) }
 func (*PingReq) ProtoMessage()    {}
 func (*PingReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{18}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{18}
 }
 func (m *PingReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_PingReq.Unmarshal(m, b)
@@ -1049,7 +1049,7 @@ func (m *PingRes) Reset()         { *m = PingRes{} }
 func (m *PingRes) String() string { return proto.CompactTextString(m) }
 func (*PingRes) ProtoMessage()    {}
 func (*PingRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{19}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{19}
 }
 func (m *PingRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_PingRes.Unmarshal(m, b)
@@ -1077,16 +1077,7 @@ func (m *PingRes) GetEcho() string {
 }
 
 type HintMostRecentCommonAncestorReq struct {
-	Filesystem string `protobuf:"bytes,1,opt,name=Filesystem,proto3" json:"Filesystem,omitempty"`
-	// A copy of the FilesystemVersion on the sending side that the replication
-	// algorithm identified as a shared most recent common version between sending
-	// and receiving side.
-	//
-	// If nil, this is an indication that the replication algorithm could not
-	// find a common ancestor between the two sides.
-	// NOTE: nilness does not mean that replication never happened - there could
-	// as well be a replication conflict. thus, dont' jump to conclusions too
-	// rapidly here.
+	Filesystem           string             `protobuf:"bytes,1,opt,name=Filesystem,proto3" json:"Filesystem,omitempty"`
 	SenderVersion        *FilesystemVersion `protobuf:"bytes,2,opt,name=SenderVersion,proto3" json:"SenderVersion,omitempty"`
 	XXX_NoUnkeyedLiteral struct{}           `json:"-"`
 	XXX_unrecognized     []byte             `json:"-"`
@@ -1097,7 +1088,7 @@ func (m *HintMostRecentCommonAncestorReq) Reset()         { *m = HintMostRecentC
 func (m *HintMostRecentCommonAncestorReq) String() string { return proto.CompactTextString(m) }
 func (*HintMostRecentCommonAncestorReq) ProtoMessage()    {}
 func (*HintMostRecentCommonAncestorReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{20}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{20}
 }
 func (m *HintMostRecentCommonAncestorReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_HintMostRecentCommonAncestorReq.Unmarshal(m, b)
@@ -1141,7 +1132,7 @@ func (m *HintMostRecentCommonAncestorRes) Reset()         { *m = HintMostRecentC
 func (m *HintMostRecentCommonAncestorRes) String() string { return proto.CompactTextString(m) }
 func (*HintMostRecentCommonAncestorRes) ProtoMessage()    {}
 func (*HintMostRecentCommonAncestorRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_e59763dc61674a79, []int{21}
+	return fileDescriptor_pdu_0f43b713cd3bf056, []int{21}
 }
 func (m *HintMostRecentCommonAncestorRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_HintMostRecentCommonAncestorRes.Unmarshal(m, b)
@@ -1458,9 +1449,9 @@ var _Replication_serviceDesc = grpc.ServiceDesc{
 	Metadata: "pdu.proto",
 }
 
-func init() { proto.RegisterFile("pdu.proto", fileDescriptor_pdu_e59763dc61674a79) }
+func init() { proto.RegisterFile("pdu.proto", fileDescriptor_pdu_0f43b713cd3bf056) }
 
-var fileDescriptor_pdu_e59763dc61674a79 = []byte{
+var fileDescriptor_pdu_0f43b713cd3bf056 = []byte{
 	// 892 bytes of a gzipped FileDescriptorProto
 	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x56, 0xdf, 0x6f, 0xdb, 0x36,
 	0x10, 0x8e, 0x6c, 0x39, 0x91, 0xcf, 0xe9, 0xea, 0x5c, 0xb2, 0x42, 0x13, 0xba, 0xce, 0xe3, 0x86,

replication/logic/pdu/pdu.proto

@@ -91,7 +91,7 @@ message ReceiveReq {
   string Filesystem = 1;
   FilesystemVersion To = 2;
 
-  // If true, the receiver should clear the resume token before performing the
+  // If true, the receiver should clear the resume token before perfoming the
   // zfs recv of the stream in the request
   bool ClearResumeToken = 3;
 }
@@ -128,17 +128,7 @@ message PingRes {
 }
 
 message HintMostRecentCommonAncestorReq {
-  string Filesystem = 1;
-
-  // A copy of the FilesystemVersion on the sending side that the replication
-  // algorithm identified as a shared most recent common version between sending
-  // and receiving side.
-  //
-  // If nil, this is an indication that the replication algorithm could not
-  // find a common ancestor between the two sides.
-  // NOTE: nilness does not mean that replication never happened - there could
-  // as well be a replication conflict. thus, dont' jump to conclusions too
-  // rapidly here.
-  FilesystemVersion SenderVersion = 2;
+    string Filesystem = 1;
+    FilesystemVersion SenderVersion = 2;
 }
 message HintMostRecentCommonAncestorRes {}

replication/logic/replication_logic.go

@@ -234,7 +234,6 @@ func resolveConflict(conflict error) (path []*pdu.FilesystemVersion, msg string)
 	if noCommonAncestor, ok := conflict.(*ConflictNoCommonAncestor); ok {
 		if len(noCommonAncestor.SortedReceiverVersions) == 0 {
 			// TODO this is hard-coded replication policy: most recent snapshot as source
-			// NOTE: Keep in sync with listStaleFiltering, it depends on this hard-coded assumption
 			var mostRecentSnap *pdu.FilesystemVersion
 			for n := len(noCommonAncestor.SortedSenderVersions) - 1; n >= 0; n-- {
 				if noCommonAncestor.SortedSenderVersions[n].Type == pdu.FilesystemVersion_Snapshot {
@@ -263,6 +262,7 @@ func (p *Planner) doPlanning(ctx context.Context) ([]*Filesystem, error) {
 		return nil, err
 	}
 	sfss := slfssres.GetFilesystems()
+	// no progress here since we could run in a live-lock on connectivity issues
 
 	rlfssres, err := p.receiver.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
 	if err != nil {
@@ -352,19 +352,18 @@ func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
 		log.WithField("token", resumeToken).Debug("decode resume token")
 	}
 
-	// give both sides a hint about how far prior replication attempts got
+	// give both sides a hint about how far the replication got
 	// This serves as a cummulative variant of SendCompleted and can be useful
 	// for example to release stale holds from an earlier (interrupted) replication.
 	// TODO FIXME: enqueue this as a replication step instead of doing it here during planning
 	//             then again, the step should run regardless of planning success
 	//             so maybe a separate phase before PLANNING, then?
 	path, conflict := IncrementalPath(rfsvs, sfsvs)
-	var sender_mrca *pdu.FilesystemVersion
+	var sender_mrca *pdu.FilesystemVersion // from sfsvs
 	if conflict == nil && len(path) > 0 {
 		sender_mrca = path[0] // shadow
 	}
-	// yes, sender_mrca may be nil, indicating that we do not have an mrca
-	{
+	if sender_mrca != nil {
 		var wg sync.WaitGroup
 		doHint := func(ep Endpoint, name string) {
 			defer wg.Done()
@@ -384,6 +383,8 @@ func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
 		go doHint(fs.sender, "sender")
 		go doHint(fs.receiver, "receiver")
 		wg.Wait()
+	} else {
+		log.Debug("cannot identify most recent common ancestor, skipping hint")
 	}
 
 	var steps []*Step
@@ -605,7 +606,7 @@ func (s *Step) doReplication(ctx context.Context) error {
 	log := getLogger(ctx)
 	sr := s.buildSendRequest(false)
 
-	log.Debug("initiate send request")
+	log.WithField("sr", sr.String()).Debug("initiate send request")
 	sres, sstreamCopier, err := s.sender.Send(ctx, sr)
 	if err != nil {
 		log.WithError(err).Error("send request failed")
@@ -632,7 +633,7 @@ func (s *Step) doReplication(ctx context.Context) error {
 		To:               sr.GetTo(),
 		ClearResumeToken: !sres.UsedResumeToken,
 	}
-	log.Debug("initiate receive request")
+	log.WithField("rr", rr.String()).Debug("initiate receive request")
 	_, err = s.receiver.Receive(ctx, rr, byteCountingStream)
 	if err != nil {
 		log.
@@ -648,10 +649,11 @@ func (s *Step) doReplication(ctx context.Context) error {
 	}
 	log.Debug("receive finished")
 
-	log.Debug("tell sender replication completed")
-	_, err = s.sender.SendCompleted(ctx, &pdu.SendCompletedReq{
+	scr := &pdu.SendCompletedReq{
 		OriginalReq: sr,
-	})
+	}
+	log.WithField("scr", scr.String()).Debug("tell sender replication completed")
+	_, err = s.sender.SendCompleted(ctx, scr)
 	if err != nil {
 		log.WithError(err).Error("error telling sender that replication completed successfully")
 		return err

rpc/dataconn/dataconn_client.go

@@ -75,7 +75,7 @@ func (c *Client) recv(ctx context.Context, conn *stream.Conn, res proto.Message)
 	if err != nil {
 		return err
 	}
-	header := string(headerBuf)
+	header := string(headerBuf) // FIXME
 	if strings.HasPrefix(header, responseHeaderHandlerErrorPrefix) {
 		// FIXME distinguishable error type
 		return &RemoteHandlerError{strings.TrimPrefix(header, responseHeaderHandlerErrorPrefix)}

rpc/dataconn/dataconn_server.go

@@ -14,8 +14,8 @@ import (
 	"github.com/zrepl/zrepl/zfs"
 )
 
-// WireInterceptor has a chance to exchange the context and connection on each client connection.
-type WireInterceptor func(ctx context.Context, rawConn *transport.AuthConn) (context.Context, *transport.AuthConn)
+// ReqInterceptor has a chance to exchange the context and connection on each request.
+type ReqInterceptor func(ctx context.Context, rawConn *transport.AuthConn) (context.Context, *transport.AuthConn)
 
 // Handler implements the functionality that is exposed by Server to the Client.
 type Handler interface {
@@ -30,19 +30,27 @@ type Handler interface {
 	PingDataconn(ctx context.Context, r *pdu.PingReq) (*pdu.PingRes, error)
 }
 
+type PreHandlerInspector = func(ctx context.Context, endpoint string, req interface{})
+
+type PostHandlerInspector = func(ctx context.Context, response interface{}, err error)
+
 type Logger = logger.Logger
 
 type Server struct {
-	h   Handler
-	wi  WireInterceptor
-	log Logger
+	h    Handler
+	ri   ReqInterceptor
+	log  Logger
+	pre  PreHandlerInspector
+	post PostHandlerInspector
 }
 
-func NewServer(wi WireInterceptor, logger Logger, handler Handler) *Server {
+func NewServer(ri ReqInterceptor, logger Logger, pre PreHandlerInspector, handler Handler, post PostHandlerInspector) *Server {
 	return &Server{
 		h:   handler,
-		wi:  wi,
+		ri:  ri,
 		log: logger,
+		pre: pre,
+		post: post,
 	}
 }
 
@@ -83,8 +91,8 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
 	defer s.log.Debug("serveConn done")
 
 	ctx := context.Background()
-	if s.wi != nil {
-		ctx, nc = s.wi(ctx, nc)
+	if s.ri != nil {
+		ctx, nc = s.ri(ctx, nc)
 	}
 
 	c := stream.Wrap(nc, HeartbeatInterval, HeartbeatPeerTimeout)
@@ -100,7 +108,7 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
 		s.log.WithError(err).Error("error reading structured part")
 		return
 	}
-	endpoint := string(header)
+	endpoint := string(header) // FIXME
 
 	reqStructured, err := c.ReadStreamedMessage(ctx, RequestStructuredMaxSize, ReqStructured)
 	if err != nil {
@@ -120,6 +128,7 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
 			s.log.WithError(err).Error("cannot unmarshal send request")
 			return
 		}
+		s.pre(ctx, endpoint, &req)
 		res, sendStream, handlerErr = s.h.Send(ctx, &req) // SHADOWING
 	case EndpointRecv:
 		var req pdu.ReceiveReq
@@ -127,6 +136,7 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
 			s.log.WithError(err).Error("cannot unmarshal receive request")
 			return
 		}
+		s.pre(ctx, endpoint, &req)
 		res, handlerErr = s.h.Receive(ctx, &req, &streamCopier{streamConn: c, closeStreamOnClose: false}) // SHADOWING
 	case EndpointPing:
 		var req pdu.PingReq
@@ -134,6 +144,7 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
 			s.log.WithError(err).Error("cannot unmarshal ping request")
 			return
 		}
+		s.pre(ctx, endpoint, &req)
 		res, handlerErr = s.h.PingDataconn(ctx, &req) // SHADOWING
 	default:
 		s.log.WithField("endpoint", endpoint).Error("unknown endpoint")
@@ -142,6 +153,8 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
 
 	s.log.WithField("endpoint", endpoint).WithField("errType", fmt.Sprintf("%T", handlerErr)).Debug("handler returned")
 
+	s.post(ctx, res, handlerErr)
+
 	// prepare protobuf now to return the protobuf error in the header
 	// if marshaling fails. We consider failed marshaling a handler error
 	var protobuf *bytes.Buffer

rpc/dataconn/dataconn_shared.go

@@ -52,7 +52,7 @@ func (s *streamCopier) WriteStreamTo(w io.Writer) zfs.StreamCopierError {
 	s.mtx.Lock()
 	defer s.mtx.Unlock()
 	if s.used {
-		panic("streamCopier used multiple times")
+		panic("streamCopier used mulitple times")
 	}
 	s.used = true
 	return s.streamConn.ReadStreamInto(w, ZFSStream)

rpc/dataconn/frameconn/frameconn.go

@@ -100,7 +100,7 @@ func (c *Conn) ReadFrame() (Frame, error) {
 		return Frame{}, ErrShutdown
 	}
 
-	// only acquire readMtx now to prioritize the draining in Shutdown()
+	// only aquire readMtx now to prioritize the draining in Shutdown()
 	// over external callers (= drain public callers)
 
 	c.readMtx.Lock()
@@ -148,7 +148,7 @@ func (c *Conn) readFrame() (Frame, error) {
 		//                | |         |           |
 		//                | |         |           F3
 		//                | |         |
-		//                | F2        |significant time between frames because
+		//                | F2        |signficant time between frames because
 		//                F1           the peer has nothing to say to us
 		//
 		// Assume we're at the point were F2's header is in c.readNext.
@@ -246,7 +246,7 @@ func (c *Conn) Shutdown(deadline time.Time) error {
 	//
 	// 1. Naive Option: We just call Close() right after CloseWrite.
 	// This yields the same race condition as explained above (DIF, first
-	// paragraph): The situation just became a little more unlikely because
+	// paragraph): The situation just becomae a little more unlikely because
 	// our rstFrameType + CloseWrite dance gave the client a full RTT worth of
 	// time to read the data from its TCP recv buffer.
 	//
@@ -295,7 +295,7 @@ func (c *Conn) Shutdown(deadline time.Time) error {
 	defer prometheus.NewTimer(prom.ShutdownSeconds).ObserveDuration()
 
 	closeWire := func(step string) error {
-		// TODO SetLinger(0) or similar (we want RST frames here, not FINS)
+		// TODO SetLinger(0) or similiar (we want RST frames here, not FINS)
 		closeErr := c.nc.Close()
 		if closeErr == nil {
 			return nil
@@ -321,10 +321,10 @@ func (c *Conn) Shutdown(deadline time.Time) error {
 
 	c.shutdown.Begin()
 	// new calls to c.ReadFrame and c.WriteFrame will now return ErrShutdown
-	// Acquiring writeMtx and readMtx afterwards ensures that already-running calls exit successfully
+	// Aquiring writeMtx and readMtx afterwards ensures that already-running calls exit successfully
 
 	// disable renewing timeouts now, enforce the requested deadline instead
-	// we need to do this before acquiring locks to enforce the timeout on slow
+	// we need to do this before aquiring locks to enforce the timeout on slow
 	// clients / if something hangs (DoS mitigation)
 	if err := c.nc.DisableTimeouts(); err != nil {
 		return hardclose(err, "disable_timeouts")

rpc/dataconn/microbenchmark/microbenchmark.go

@@ -1,4 +1,4 @@
-// microbenchmark to manually test rpc/dataconn performance
+// microbenchmark to manually test rpc/dataconn perforamnce
 //
 // With stdin / stdout on client and server, simulating zfs send|recv piping
 //
@@ -6,7 +6,7 @@
 //   ./microbenchmark -appmode client -direction recv < /dev/zero
 //
 //
-// Without the overhead of pipes (just protocol performance, mostly useful with perf bc no bw measurement)
+// Without the overhead of pipes (just protocol perforamnce, mostly useful with perf bc no bw measurement)
 //
 //   ./microbenchmark -appmode client -direction recv -devnoopWriter -devnoopReader
 //   ./microbenchmark -appmode server -devnoopReader -devnoopWriter

rpc/dataconn/timeoutconn/timeoutconn.go

@@ -49,7 +49,7 @@ type Wire interface {
 	// No data that could otherwise be Read is lost as a consequence of this call.
 	// The use case for this API is abortive connection shutdown.
 	// To provide any value over draining Wire using io.Read, an implementation
-	// will likely use out-of-band messaging mechanisms.
+	// will likely use out-of-bounds messaging mechanisms.
 	// TODO WaitForPeerClose() (supported bool, err error)
 }
 
@@ -119,7 +119,7 @@ restart:
 	return n, err
 }
 
-// Writes the given buffers to Conn, following the semantics of io.Copy,
+// Writes the given buffers to Conn, following the sematincs of io.Copy,
 // but is guaranteed to use the writev system call if the wrapped Wire
 // support it.
 // Note the Conn does not support writev through io.Copy(aConn, aNetBuffers).
@@ -158,9 +158,9 @@ var _ SyscallConner = (*net.TCPConn)(nil)
 // Think of io.ReadvFull, but for net.Buffers + using the readv syscall.
 //
 // If the underlying Wire is not a SyscallConner, a fallback
-// implementation based on repeated Conn.Read invocations is used.
+// ipmlementation based on repeated Conn.Read invocations is used.
 //
-// If the connection returned io.EOF, the number of bytes written until
+// If the connection returned io.EOF, the number of bytes up ritten until
 // then + io.EOF is returned. This behavior is different to io.ReadFull
 // which returns io.ErrUnexpectedEOF.
 func (c Conn) ReadvFull(buffers net.Buffers) (n int64, err error) {

rpc/dataconn/timeoutconn/timoutconn_readv.go

@@ -99,7 +99,7 @@ func (c Conn) doOneReadv(rawConn syscall.RawConn, iovecs *[]syscall.Iovec) (n in
 				// Update left, cannot go below 0 due to
 				// a) definition of thisIovecConsumedCompletely
 				// b) left > 0 due to loop invariant
-				// Converting .Len to int64 is thus also safe now, because it is < left < INT_MAX
+				// Convertion .Len to int64 is thus also safe now, because it is < left < INT_MAX
 				left -= int((*iovecs)[0].Len)
 				*iovecs = (*iovecs)[1:]
 			} else {

rpc/grpcclientidentity/authlistener_grpc_adaptor.go

@@ -101,7 +101,11 @@ func (*transportCredentials) OverrideServerName(string) error {
 
 type ContextInterceptor = func(ctx context.Context) context.Context
 
-func NewInterceptors(logger Logger, clientIdentityKey interface{}, ctxInterceptor ContextInterceptor) (unary grpc.UnaryServerInterceptor, stream grpc.StreamServerInterceptor) {
+type PreHandlerInspector func(ctx context.Context, endpoint string, req interface{})
+
+type PostHandlerInspector func(ctx context.Context, response interface{}, err error)
+
+func NewInterceptors(logger Logger, clientIdentityKey interface{}, ctxInterceptor ContextInterceptor, pre PreHandlerInspector, post PostHandlerInspector) (unary grpc.UnaryServerInterceptor, stream grpc.StreamServerInterceptor) {
 	unary = func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
 		logger.WithField("fullMethod", info.FullMethod).Debug("request")
 		p, ok := peer.FromContext(ctx)
@@ -118,7 +122,10 @@ func NewInterceptors(logger Logger, clientIdentityKey interface{}, ctxIntercepto
 		if ctxInterceptor != nil {
 			ctx = ctxInterceptor(ctx)
 		}
-		return handler(ctx, req)
+		pre(ctx, info.FullMethod, req)
+		res, err := handler(ctx, req)
+		post(ctx, res, err)
+		return res, err
 	}
 	stream = func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
 		panic("unimplemented")

rpc/grpcclientidentity/grpchelper/authlistener_grpc_adaptor_wrapper.go

@@ -4,14 +4,18 @@ package grpchelper
 
 import (
 	"context"
+	"fmt"
+	"strings"
 	"time"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/keepalive"
+	"google.golang.org/grpc/metadata"
 
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/rpc/grpcclientidentity"
 	"github.com/zrepl/zrepl/rpc/netadaptor"
+	"github.com/zrepl/zrepl/tracing"
 	"github.com/zrepl/zrepl/transport"
 )
 
@@ -28,15 +32,40 @@ type Logger = logger.Logger
 
 // ClientConn is an easy-to-use wrapper around the Dialer and TransportCredentials interface
 // to produce a grpc.ClientConn
-func ClientConn(cn transport.Connecter, log Logger) *grpc.ClientConn {
+func ClientConn(cn transport.Connecter, log Logger, genReqID func() string) *grpc.ClientConn {
 	ka := grpc.WithKeepaliveParams(keepalive.ClientParameters{
 		Time:                StartKeepalivesAfterInactivityDuration,
 		Timeout:             KeepalivePeerTimeout,
 		PermitWithoutStream: true,
 	})
+	unaryIntcpt := grpc.WithUnaryInterceptor(func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
+
+		traceStack := tracing.GetStack(ctx)
+		if len(traceStack) == 0 {
+			panic("implementation error: expecting trace stack")
+		}
+		reqId := genReqID() + " " + strings.Join(traceStack, " <> ")
+
+		l := log
+		l = l.WithField("reqID", reqId)
+		l = l.WithField("reqT", fmt.Sprintf("%T", req))
+		l = l.WithField("repT", fmt.Sprintf("%T", reply))
+		l = l.WithField("ctxT", fmt.Sprintf("%T", ctx))
+		l.Debug("unary intercepted")
+
+		ctx = metadata.AppendToOutgoingContext(ctx, "zrepl-req-id", reqId)
+
+		err := invoker(ctx, method, req, reply, cc, opts...)
+
+		l = log.WithField("repT", fmt.Sprintf("%T", reply))
+		l = log.WithField("errT", fmt.Sprintf("%T", err))
+		l.Debug("reply intercepted")
+
+		return err
+	})
 	dialerOption := grpc.WithDialer(grpcclientidentity.NewDialer(log, cn))
 	cred := grpc.WithTransportCredentials(grpcclientidentity.NewTransportCredentials(log))
-	cc, err := grpc.DialContext(context.Background(), "doesn't matter done by dialer", dialerOption, cred, ka)
+	cc, err := grpc.DialContext(context.Background(), "doesn't matter done by dialer", dialerOption, cred, ka, unaryIntcpt)
 	if err != nil {
 		log.WithError(err).Error("cannot create gRPC client conn (non-blocking)")
 		// It's ok to panic here: the we call grpc.DialContext without the
@@ -50,7 +79,7 @@ func ClientConn(cn transport.Connecter, log Logger) *grpc.ClientConn {
 }
 
 // NewServer is a convenience interface around the TransportCredentials and Interceptors interface.
-func NewServer(authListener transport.AuthenticatedListener, clientIdentityKey interface{}, logger grpcclientidentity.Logger, ctxInterceptor grpcclientidentity.ContextInterceptor) (srv *grpc.Server, serve func() error) {
+func NewServer(authListener transport.AuthenticatedListener, clientIdentityKey interface{}, logger grpcclientidentity.Logger, interceptor grpcclientidentity.ContextInterceptor, pre grpcclientidentity.PreHandlerInspector, post grpcclientidentity.PostHandlerInspector) (srv *grpc.Server, serve func() error) {
 	ka := grpc.KeepaliveParams(keepalive.ServerParameters{
 		Time:    StartKeepalivesAfterInactivityDuration,
 		Timeout: KeepalivePeerTimeout,
@@ -60,8 +89,17 @@ func NewServer(authListener transport.AuthenticatedListener, clientIdentityKey i
 		PermitWithoutStream: true,
 	})
 	tcs := grpcclientidentity.NewTransportCredentials(logger)
-	unary, stream := grpcclientidentity.NewInterceptors(logger, clientIdentityKey, ctxInterceptor)
-	srv = grpc.NewServer(grpc.Creds(tcs), grpc.UnaryInterceptor(unary), grpc.StreamInterceptor(stream), ka, ep)
+	unary, stream := grpcclientidentity.NewInterceptors(logger, clientIdentityKey, interceptor, pre, post)
+	unary2 := func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
+		md, ok := metadata.FromIncomingContext(ctx)
+		if !ok {
+			logger.Error("no request id in incoming request")
+		} else {
+			logger.Info(fmt.Sprintf("req-id = %v", md.Get("zrepl-req-id")))
+		}
+		return unary(ctx, req, info, handler)
+	}
+	srv = grpc.NewServer(grpc.Creds(tcs), grpc.UnaryInterceptor(unary2), grpc.StreamInterceptor(stream), ka, ep)
 
 	serve = func() error {
 		if err := srv.Serve(netadaptor.New(authListener, logger)); err != nil {

rpc/rpc_client.go

@@ -26,6 +26,7 @@ import (
 // Client implements the active side of a replication setup.
 // It satisfies the Endpoint, Sender and Receiver interface defined by package replication.
 type Client struct {
+	reqIDGen      *requestIDGenerator
 	dataClient    *dataconn.Client
 	controlClient pdu.ReplicationClient // this the grpc client instance, see constructor
 	controlConn   *grpc.ClientConn
@@ -47,10 +48,13 @@ func NewClient(cn transport.Connecter, loggers Loggers) *Client {
 	muxedConnecter := mux(cn)
 
 	c := &Client{
-		loggers: loggers,
-		closed:  make(chan struct{}),
+		loggers:  loggers,
+		closed:   make(chan struct{}),
+		reqIDGen: newRequestIDGenerator(),
 	}
-	grpcConn := grpchelper.ClientConn(muxedConnecter.control, loggers.Control)
+	grpcConn := grpchelper.ClientConn(muxedConnecter.control, loggers.Control, func() string {
+		return c.reqIDGen.newID().String()
+	})
 
 	go func() {
 		ctx, cancel := context.WithCancel(context.Background())

rpc/rpc_requestid.go

@@ -0,0 +1,58 @@
+package rpc
+
+import (
+	"context"
+	"encoding/base64"
+	"strings"
+
+	"github.com/google/uuid"
+)
+
+type requestIDGenerator struct{}
+
+func newRequestIDGenerator() *requestIDGenerator {
+	return &requestIDGenerator{}
+}
+
+type RequestID struct{ s string }
+
+func (r RequestID) String() string { return r.s }
+
+type requestIDContextKey int
+
+const (
+	requestIDContextKeyRequestID requestIDContextKey = 1 + iota
+)
+
+func (g *requestIDGenerator) newID() RequestID {
+	id := uuid.New()
+	var buf strings.Builder
+	enc := base64.NewEncoder(base64.RawStdEncoding, &buf)
+	n, err := enc.Write(id[:])
+	if err != nil {
+		panic(err)
+	} else if n != len(id) {
+		panic(n)
+	}
+	if err := enc.Close(); err != nil {
+		panic(err)
+	}
+	return RequestID{buf.String()}
+}
+
+func (g *requestIDGenerator) inject(ctx context.Context) context.Context {
+	return context.WithValue(ctx, requestIDContextKeyRequestID, g.newID())
+}
+
+func GetRequestID(ctx context.Context) (id RequestID, ok bool) {
+	id, ok = ctx.Value(requestIDContextKeyRequestID).(RequestID)
+	return id, ok
+}
+
+func MustGetRequestID(ctx context.Context) (id RequestID) {
+	id, ok := GetRequestID(ctx)
+	if !ok {
+		panic("calling context expectes request id to bet set")
+	}
+	return id
+}

rpc/rpc_server.go

@@ -7,6 +7,7 @@ import (
 	"github.com/zrepl/zrepl/endpoint"
 	"github.com/zrepl/zrepl/replication/logic/pdu"
 	"github.com/zrepl/zrepl/rpc/dataconn"
+	"github.com/zrepl/zrepl/rpc/grpcclientidentity"
 	"github.com/zrepl/zrepl/rpc/grpcclientidentity/grpchelper"
 	"github.com/zrepl/zrepl/rpc/versionhandshake"
 	"github.com/zrepl/zrepl/transport"
@@ -30,15 +31,35 @@ type Server struct {
 	dataServerServe    serveFunc
 }
 
-type HandlerContextInterceptor func(ctx context.Context) context.Context
+type CtxInterceptor = grpcclientidentity.ContextInterceptor
+
+func chainInterceptors(interceptors []CtxInterceptor) CtxInterceptor {
+	return func(ctx context.Context) (_ context.Context) {
+		for _, i := range interceptors {
+			ctx = i(ctx)
+		}
+		return ctx
+	}
+}
+
+type PreHandlerInspector = func(ctx context.Context, endpoint string, req interface{})
+
+type PostHandlerInspector = func(ctx context.Context, response interface{}, err error)
 
 // config must be valid (use its Validate function).
-func NewServer(handler Handler, loggers Loggers, ctxInterceptor HandlerContextInterceptor) *Server {
+func NewServer(handler Handler, loggers Loggers, ctxInterceptor CtxInterceptor, pre PreHandlerInspector, post PostHandlerInspector) *Server {
+
+	reqIdGen := newRequestIDGenerator()
+
+	ctxInterceptor = chainInterceptors([]CtxInterceptor{
+		reqIdGen.inject,
+		ctxInterceptor,
+	})
 
 	// setup control server
 	controlServerServe := func(ctx context.Context, controlListener transport.AuthenticatedListener, errOut chan<- error) {
 
-		controlServer, serve := grpchelper.NewServer(controlListener, endpoint.ClientIdentityKey, loggers.Control, ctxInterceptor)
+		controlServer, serve := grpchelper.NewServer(controlListener, endpoint.ClientIdentityKey, loggers.Control, ctxInterceptor, pre, post)
 		pdu.RegisterReplicationServer(controlServer, handler)
 
 		// give time for graceful stop until deadline expires, then hard stop
@@ -63,7 +84,7 @@ func NewServer(handler Handler, loggers Loggers, ctxInterceptor HandlerContextIn
 		}
 		return ctx, wire
 	}
-	dataServer := dataconn.NewServer(dataServerClientIdentitySetter, loggers.Data, handler)
+	dataServer := dataconn.NewServer(dataServerClientIdentitySetter, loggers.Data, pre, handler, post)
 	dataServerServe := func(ctx context.Context, dataListener transport.AuthenticatedListener, errOut chan<- error) {
 		dataServer.Serve(ctx, dataListener)
 		errOut <- nil // TODO bad design of dataServer?

rpc/versionhandshake/versionhandshake.go

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

tlsconf/tlsconf.go

@@ -56,7 +56,7 @@ func NewClientAuthListener(
 }
 
 // Accept() accepts a connection from the *net.TCPListener passed to the constructor
-// and sets up the TLS connection, including handshake and peer CommonName validation
+// and sets up the TLS connection, including handshake and peer CommmonName validation
 // within the specified handshakeTimeout.
 //
 // It returns both the raw TCP connection (tcpConn) and the TLS connection (tlsConn) on top of it.
@@ -121,6 +121,7 @@ func ClientAuthClient(serverName string, rootCA *x509.CertPool, clientCert tls.C
 		ServerName:   serverName,
 		KeyLogWriter: keylogFromEnv(),
 	}
+	tlsConfig.BuildNameToCertificate()
 	return tlsConfig, nil
 }
 

tracing/tracing.go

@@ -0,0 +1,53 @@
+package tracing
+
+import "context"
+
+type tracingContextKey int
+
+const (
+	CallerContext tracingContextKey = 1 + iota
+)
+
+type jobSubtree struct {
+	jobid string
+}
+
+type ctx struct {
+	parent *ctx
+	job    *jobSubtree
+	ident  string
+}
+
+var root = &ctx{nil, nil, ""}
+
+func getParentOrRoot(c context.Context) *ctx {
+	parent, ok := c.Value(CallerContext).(*ctx)
+	if !ok {
+		parent = root
+	}
+	return parent
+}
+
+func makeChild(c context.Context, child *ctx) context.Context {
+	if child.parent == nil {
+		panic(child)
+	}
+	return context.WithValue(c, CallerContext, child)
+}
+
+func Child(c context.Context, ident string) context.Context {
+	parent := getParentOrRoot(c)
+	return makeChild(c, &ctx{parent: parent, ident: ident})
+}
+
+func GetStack(c context.Context) (idents []string) {
+	ct, ok := c.Value(CallerContext).(*ctx)
+	if !ok {
+		return idents
+	}
+	for ct.parent != nil {
+		idents = append(idents, ct.ident)
+		ct = ct.parent
+	}
+	return idents
+}

tracing/tracing_test.go

@@ -0,0 +1,21 @@
+package tracing
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestIt(t *testing.T) {
+	ctx := context.Background()
+
+	ctx = Child(ctx, "a")
+	ctx = Child(ctx, "b")
+	ctx = Child(ctx, "c")
+	ctx = Child(ctx, "d")
+
+	assert.Equal(t, "dcba", strings.Join(GetStack(ctx), ""))
+
+}

transport/local/serve_local.go

@@ -100,7 +100,7 @@ func (l *LocalListener) Accept(ctx context.Context) (*transport.AuthConn, error)
 			WithField("res.conn", res.conn).WithField("res.err", res.err).
 			Debug("responding to client request")
 
-		// contract between Connect and Accept is that Connect sends a req.callback
+		// contract bewteen Connect and Accept is that Connect sends a req.callback
 		// into which we can send one result non-blockingly.
 		// We want to panic if that contract is violated (impl error)
 		//
@@ -112,7 +112,7 @@ func (l *LocalListener) Accept(ctx context.Context) (*transport.AuthConn, error)
 		defer func() {
 			errv := recover()
 			if errv == clientCallbackBlocked {
-				// this would be a violation of contract between Connect and Accept, see above
+				// this would be a violation of contract betwee Connect and Accept, see above
 				panic(clientCallbackBlocked)
 			} else {
 				transport.GetLogger(ctx).WithField("recover_err", errv).

transport/ssh/serve_stdinserver.go

@@ -116,7 +116,7 @@ func (m *MultiStdinserverListener) Close() error {
 	return oneErr
 }
 
-// a single stdinserverListener (part of multiStdinserverListener)
+// a single stdinserverListener (part of multiStinserverListener)
 type stdinserverListener struct {
 	l              *netssh.Listener
 	clientIdentity string

transport/tls/serve_tls.go

@@ -32,7 +32,7 @@ func TLSListenerFactoryFromConfig(c *config.Global, in *config.TLSServe) (transp
 
 	serverCert, err := tls.LoadX509KeyPair(in.Cert, in.Key)
 	if err != nil {
-		return nil, errors.Wrap(err, "cannot parse cert/key pair")
+		return nil, errors.Wrap(err, "cannot parse cer/key pair")
 	}
 
 	clientCNs := make(map[string]struct{}, len(in.ClientCNs))

transport/transport.go

@@ -61,7 +61,7 @@ func ValidateClientIdentity(in string) (err error) {
 		return err
 	}
 	if path.Length() != 1 {
-		return errors.New("client identity must be a single path component (not empty, no '/')")
+		return errors.New("client identity must be a single path comonent (not empty, no '/')")
 	}
 	return nil
 }

util/bytecounter/bytecounter_streamcopier.go

@@ -7,7 +7,7 @@ import (
 	"github.com/zrepl/zrepl/zfs"
 )
 
-// StreamCopier wraps a zfs.StreamCopier, reimplementing
+// StreamCopier wraps a zfs.StreamCopier, reimplemening
 // its interface and counting the bytes written to during copying.
 type StreamCopier interface {
 	zfs.StreamCopier

util/envconst/envconst.go

@@ -1,10 +1,7 @@
 package envconst
 
 import (
-	"flag"
-	"fmt"
 	"os"
-	"reflect"
 	"strconv"
 	"sync"
 	"time"
@@ -88,31 +85,3 @@ func String(varname string, def string) string {
 	cache.Store(varname, e)
 	return e
 }
-
-func Var(varname string, def flag.Value) interface{} {
-
-	// use def's type to instantiate a new object of that same type
-	// and call flag.Value.Set() on it
-	defType := reflect.TypeOf(def)
-	if defType.Kind() != reflect.Ptr {
-		panic(fmt.Sprintf("envconst var must be a pointer, got %T", def))
-	}
-	defElemType := defType.Elem()
-
-	if v, ok := cache.Load(varname); ok {
-		return v.(string)
-	}
-	e := os.Getenv(varname)
-	if e == "" {
-		return def
-	}
-
-	newInstance := reflect.New(defElemType)
-	if err := newInstance.Interface().(flag.Value).Set(e); err != nil {
-		panic(err)
-	}
-
-	res := newInstance.Interface()
-	cache.Store(varname, res)
-	return res
-}

util/envconst/envconst_test.go

@@ -1,50 +0,0 @@
-package envconst_test
-
-import (
-	"fmt"
-	"os"
-	"testing"
-
-	"github.com/stretchr/testify/require"
-
-	"github.com/zrepl/zrepl/util/envconst"
-)
-
-type ExampleVarType struct{ string }
-
-var (
-	Var1 = ExampleVarType{"var1"}
-	Var2 = ExampleVarType{"var2"}
-)
-
-func (m ExampleVarType) String() string { return string(m.string) }
-func (m *ExampleVarType) Set(s string) error {
-	switch s {
-	case Var1.String():
-		*m = Var1
-	case Var2.String():
-		*m = Var2
-	default:
-		return fmt.Errorf("unknown var %q", s)
-	}
-	return nil
-}
-
-const EnvVarName = "ZREPL_ENVCONST_UNIT_TEST_VAR"
-
-func TestVar(t *testing.T) {
-	_, set := os.LookupEnv(EnvVarName)
-	require.False(t, set)
-	defer os.Unsetenv(EnvVarName)
-
-	val := envconst.Var(EnvVarName, &Var1)
-	if &Var1 != val {
-		t.Errorf("default value shut be same address")
-	}
-
-	err := os.Setenv(EnvVarName, "var2")
-	require.NoError(t, err)
-
-	val = envconst.Var(EnvVarName, &Var1)
-	require.Equal(t, &Var2, val, "only structural identity is required for non-default vars")
-}