endpoing: better log message for v1 replication cursors

refs #196

Makefile

@@ -23,8 +23,7 @@ GOHOSTARCH ?= $(shell bash -c 'source <($(GO) env) && echo "$$GOHOSTARCH"')
 GO_ENV_VARS := GO111MODULE=on
 GO_LDFLAGS := "-X github.com/zrepl/zrepl/version.zreplVersion=$(_ZREPL_VERSION)"
 GO_MOD_READONLY := -mod=readonly
-GO_EXTRA_BUILDFLAGS :=
-GO_BUILDFLAGS := $(GO_MOD_READONLY) $(GO_EXTRA_BUILDFLAGS)
+GO_BUILDFLAGS := $(GO_MOD_READONLY)
 GO_BUILD := $(GO_ENV_VARS) $(GO) build $(GO_BUILDFLAGS) -ldflags $(GO_LDFLAGS)
 GOLANGCI_LINT := golangci-lint
 ifneq ($(GOARM),)
@@ -66,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
@@ -161,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):
@@ -169,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 > $@

cli/cli.go

@@ -19,38 +19,9 @@ 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",
-	},
-}
-
-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,
+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")
@@ -60,14 +31,17 @@ func init() {
 			}
 			os.Exit(1)
 		}
-				if err := info.genFunc(args[0]); err != nil {
-					fmt.Fprintf(os.Stderr, "error generating %s completion: %s", sh, err)
+		if err := rootCmd.GenBashCompletionFile(args[0]); err != nil {
+			fmt.Fprintf(os.Stderr, "error generating bash completion: %s", err)
 			os.Exit(1)
 		}
 	},
-		})
+	Hidden: true,
 }
-	rootCmd.AddCommand(genCompletionCmd)
+
+func init() {
+	rootCmd.PersistentFlags().StringVar(&rootArgs.configPath, "config", "", "config file path")
+	rootCmd.AddCommand(bashcompCmd)
 }
 
 type Subcommand struct {

client/hold_create_step_hold.go

@@ -12,28 +12,28 @@ import (
 	"github.com/zrepl/zrepl/zfs"
 )
 
-var zabsCreateStepHoldFlags struct {
+var holdsCreateStepHoldFlags struct {
 	target string
 	jobid  JobIDFlag
 }
 
-var zabsCmdCreateStepHold = &cli.Subcommand{
+var holdsCmdCreateStepHold = &cli.Subcommand{
 	Use:             "step",
-	Run:             doZabsCreateStep,
+	Run:             doHoldsCreateStep,
 	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")
+		f.StringVarP(&holdsCreateStepHoldFlags.target, "target", "t", "", "snapshot to be held / bookmark to be held")
+		f.VarP(&holdsCreateStepHoldFlags.jobid, "jobid", "j", "jobid for which the hold is installed")
 	},
 }
 
-func doZabsCreateStep(sc *cli.Subcommand, args []string) error {
+func doHoldsCreateStep(sc *cli.Subcommand, args []string) error {
 	if len(args) > 0 {
 		return errors.New("subcommand takes no arguments")
 	}
 
-	f := &zabsCreateStepHoldFlags
+	f := &holdsCreateStepHoldFlags
 
 	fs, _, _, err := zfs.DecomposeVersionString(f.target)
 	if err != nil {

client/holds.go

@@ -14,15 +14,15 @@ import (
 )
 
 var (
-	ZFSAbstractionsCmd = &cli.Subcommand{
-		Use:   "zfs-abstraction",
-		Short: "manage abstractions that zrepl builds on top of ZFS",
+	HoldsCmd = &cli.Subcommand{
+		Use:   "holds",
+		Short: "manage holds & step bookmarks",
 		SetupSubcommands: func() []*cli.Subcommand {
 			return []*cli.Subcommand{
-				zabsCmdList,
-				zabsCmdReleaseAll,
-				zabsCmdReleaseStale,
-				zabsCmdCreate,
+				holdsCmdList,
+				holdsCmdReleaseAll,
+				holdsCmdReleaseStale,
+				holdsCmdCreate,
 			}
 		},
 	}
@@ -30,7 +30,7 @@ var (
 
 // a common set of CLI flags that map to the fields of an
 // endpoint.ListZFSHoldsAndBookmarksQuery
-type zabsFilterFlags struct {
+type holdsFilterFlags struct {
 	Filesystems FilesystemsFilterFlag
 	Job         JobIDFlag
 	Types       AbstractionTypesFlag
@@ -38,7 +38,7 @@ type zabsFilterFlags struct {
 }
 
 // produce a query from the CLI flags
-func (f zabsFilterFlags) Query() (endpoint.ListZFSHoldsAndBookmarksQuery, error) {
+func (f holdsFilterFlags) Query() (endpoint.ListZFSHoldsAndBookmarksQuery, error) {
 	q := endpoint.ListZFSHoldsAndBookmarksQuery{
 		FS:          f.Filesystems.FlagValue(),
 		What:        f.Types.FlagValue(),
@@ -48,7 +48,7 @@ func (f zabsFilterFlags) Query() (endpoint.ListZFSHoldsAndBookmarksQuery, error)
 	return q, q.Validate()
 }
 
-func (f *zabsFilterFlags) registerZabsFilterFlags(s *pflag.FlagSet, verb string) {
+func (f *holdsFilterFlags) registerHoldsFilterFlags(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))

client/holds_create.go

@@ -0,0 +1,14 @@
+package client
+
+import "github.com/zrepl/zrepl/cli"
+
+var holdsCmdCreate = &cli.Subcommand{
+	Use:             "create",
+	NoRequireConfig: true,
+	Short:           `create zrepl-managed holds and boomkmarks (for debugging & development only!)`,
+	SetupSubcommands: func() []*cli.Subcommand {
+		return []*cli.Subcommand{
+			holdsCmdCreateStepHold,
+		}
+	},
+}

client/holds_list.go

@@ -16,23 +16,23 @@ import (
 	"github.com/zrepl/zrepl/util/chainlock"
 )
 
-var zabsListFlags struct {
-	Filter zabsFilterFlags
+var holdsListFlags struct {
+	Filter holdsFilterFlags
 	Json   bool
 }
 
-var zabsCmdList = &cli.Subcommand{
+var holdsCmdList = &cli.Subcommand{
 	Use:             "list",
-	Short:           `list zrepl ZFS abstractions`,
-	Run:             doZabsList,
+	Run:             doHoldsList,
 	NoRequireConfig: true,
+	Short:           "list holds and bookmarks",
 	SetupFlags: func(f *pflag.FlagSet) {
-		zabsListFlags.Filter.registerZabsFilterFlags(f, "list")
-		f.BoolVar(&zabsListFlags.Json, "json", false, "emit JSON")
+		holdsListFlags.Filter.registerHoldsFilterFlags(f, "list")
+		f.BoolVar(&holdsListFlags.Json, "json", false, "emit JSON")
 	},
 }
 
-func doZabsList(sc *cli.Subcommand, args []string) error {
+func doHoldsList(sc *cli.Subcommand, args []string) error {
 	var err error
 	ctx := context.Background()
 
@@ -40,7 +40,7 @@ func doZabsList(sc *cli.Subcommand, args []string) error {
 		return errors.New("this subcommand takes no positional arguments")
 	}
 
-	q, err := zabsListFlags.Filter.Query()
+	q, err := holdsListFlags.Filter.Query()
 	if err != nil {
 		return errors.Wrap(err, "invalid filter specification on command line")
 	}
@@ -62,7 +62,7 @@ func doZabsList(sc *cli.Subcommand, args []string) error {
 		for a := range abstractions {
 			func() {
 				defer line.Lock().Unlock()
-				if zabsListFlags.Json {
+				if holdsListFlags.Json {
 					enc.SetIndent("", "  ")
 					if err := enc.Encode(abstractions); err != nil {
 						panic(err)

client/holds_release.go

@@ -15,35 +15,35 @@ import (
 )
 
 // shared between release-all and release-step
-var zabsReleaseFlags struct {
-	Filter zabsFilterFlags
+var holdsReleaseFlags struct {
+	Filter holdsFilterFlags
 	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")
+func registerHoldsReleaseFlags(s *pflag.FlagSet) {
+	holdsReleaseFlags.Filter.registerHoldsFilterFlags(s, "release")
+	s.BoolVar(&holdsReleaseFlags.Json, "json", false, "emit json instead of pretty-printed")
+	s.BoolVar(&holdsReleaseFlags.DryRun, "dry-run", false, "do a dry-run")
 }
 
-var zabsCmdReleaseAll = &cli.Subcommand{
+var holdsCmdReleaseAll = &cli.Subcommand{
 	Use:             "release-all",
-	Run:             doZabsReleaseAll,
+	Run:             doHoldsReleaseAll,
 	NoRequireConfig: true,
-	Short:           `(DANGEROUS) release ALL zrepl ZFS abstractions (mostly useful for uninstalling zrepl completely or for "de-zrepl-ing" a filesystem)`,
-	SetupFlags:      registerZabsReleaseFlags,
+	Short:           `(DANGEROUS) release all zrepl-managed holds and bookmarks, mostly useful for uninstalling zrepl`,
+	SetupFlags:      registerHoldsReleaseFlags,
 }
 
-var zabsCmdReleaseStale = &cli.Subcommand{
+var holdsCmdReleaseStale = &cli.Subcommand{
 	Use:             "release-stale",
-	Run:             doZabsReleaseStale,
+	Run:             doHoldsReleaseStale,
 	NoRequireConfig: true,
-	Short:           `release stale zrepl ZFS abstractions (useful if zrepl has a bug and does not do it by itself)`,
-	SetupFlags:      registerZabsReleaseFlags,
+	Short:           `release stale zrepl-managed holds and boomkarks (useful if zrepl has a bug and doesn't do it by itself)`,
+	SetupFlags:      registerHoldsReleaseFlags,
 }
 
-func doZabsReleaseAll(sc *cli.Subcommand, args []string) error {
+func doHoldsReleaseAll(sc *cli.Subcommand, args []string) error {
 	var err error
 	ctx := context.Background()
 
@@ -51,7 +51,7 @@ func doZabsReleaseAll(sc *cli.Subcommand, args []string) error {
 		return errors.New("this subcommand takes no positional arguments")
 	}
 
-	q, err := zabsReleaseFlags.Filter.Query()
+	q, err := holdsReleaseFlags.Filter.Query()
 	if err != nil {
 		return errors.Wrap(err, "invalid filter specification on command line")
 	}
@@ -65,10 +65,10 @@ func doZabsReleaseAll(sc *cli.Subcommand, args []string) error {
 		// proceed anyways with rest of abstractions
 	}
 
-	return doZabsRelease_Common(ctx, abstractions)
+	return doHoldsRelease_Common(ctx, abstractions)
 }
 
-func doZabsReleaseStale(sc *cli.Subcommand, args []string) error {
+func doHoldsReleaseStale(sc *cli.Subcommand, args []string) error {
 
 	var err error
 	ctx := context.Background()
@@ -77,7 +77,7 @@ func doZabsReleaseStale(sc *cli.Subcommand, args []string) error {
 		return errors.New("this subcommand takes no positional arguments")
 	}
 
-	q, err := zabsReleaseFlags.Filter.Query()
+	q, err := holdsReleaseFlags.Filter.Query()
 	if err != nil {
 		return errors.Wrap(err, "invalid filter specification on command line")
 	}
@@ -87,13 +87,13 @@ func doZabsReleaseStale(sc *cli.Subcommand, args []string) error {
 		return err // context clear by invocation of command
 	}
 
-	return doZabsRelease_Common(ctx, stalenessInfo.Stale)
+	return doHoldsRelease_Common(ctx, stalenessInfo.Stale)
 }
 
-func doZabsRelease_Common(ctx context.Context, destroy []endpoint.Abstraction) error {
+func doHoldsRelease_Common(ctx context.Context, destroy []endpoint.Abstraction) error {
 
-	if zabsReleaseFlags.DryRun {
-		if zabsReleaseFlags.Json {
+	if holdsReleaseFlags.DryRun {
+		if holdsReleaseFlags.Json {
 			m, err := json.MarshalIndent(destroy, "", "  ")
 			if err != nil {
 				panic(err)
@@ -121,7 +121,7 @@ func doZabsRelease_Common(ctx context.Context, destroy []endpoint.Abstraction) e
 
 	for res := range outcome {
 		hadErr = hadErr || res.DestroyErr != nil
-		if zabsReleaseFlags.Json {
+		if holdsReleaseFlags.Json {
 			err := enc.Encode(res)
 			if err != nil {
 				colorErr.Fprintf(os.Stderr, "cannot marshal there were errors in destroying the abstractions")

client/migrate.go

@@ -211,7 +211,7 @@ func doMigrateReplicationCursorFS(ctx context.Context, v1CursorJobs []job.Job, f
 	}
 	fmt.Printf("identified owning job %q\n", owningJob.Name())
 
-	bookmarks, err := zfs.ZFSListFilesystemVersions(ctx, fs, zfs.ListFilesystemVersionsOptions{
+	bookmarks, err := zfs.ZFSListFilesystemVersions(fs, zfs.ListFilesystemVersionsOptions{
 		Types: zfs.Bookmarks,
 	})
 	if err != nil {

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,
-		}
-	},
-}

config/samples/source.yml

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

daemon/snapper/snapper.go

@@ -485,7 +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(context.TODO(), d, zfs.ListFilesystemVersionsOptions{
+	fsvs, err := zfs.ZFSListFilesystemVersions(d, zfs.ListFilesystemVersionsOptions{
 		Types:           zfs.Snapshots,
 		ShortnamePrefix: prefix,
 	})

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

docs/configuration/transports.rst

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

docs/installation.rst

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

docs/installation/apt-repos.rst

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

docs/installation/compile-from-source.rst

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

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

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

docs/installation/packages.rst

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

docs/installation/user-privileges.rst

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

docs/installation/what-next.rst

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

docs/tutorial.rst

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

docs/usage.rst

@@ -38,8 +38,8 @@ 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>` )
+    * - ``zrepl holds``
+      - list and remove holds and step bookmarks created by zrepl (see :ref:`overview <replication-cursor-and-last-received-hold>` )
 
 .. _usage-zrepl-daemon:
 

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(ctx, lp, zfs.ListFilesystemVersionsOptions{})
+	fsvs, err := zfs.ZFSListFilesystemVersions(lp, zfs.ListFilesystemVersionsOptions{})
 	if err != nil {
 		return nil, err
 	}
@@ -150,11 +150,11 @@ func (p *Sender) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.HintMo
 	}
 
 	// take care of stale step holds
-	log.WithField("step-holds-cleanup-mode", senderHintMostRecentCommonAncestorStepCleanupMode).
+	log.WithField("step-holds-cleanup-mode", hintMostRecentCommonAncestorStepCleanupMode).
 		Debug("taking care of possibly stale step holds")
 	doStepCleanup := false
 	var stepCleanupSince *CreateTXGRangeBound
-	switch senderHintMostRecentCommonAncestorStepCleanupMode {
+	switch hintMostRecentCommonAncestorStepCleanupMode {
 	case StepCleanupNoCleanup:
 		doStepCleanup = false
 	case StepCleanupRangeSinceUnbounded:
@@ -180,7 +180,7 @@ func (p *Sender) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.HintMo
 			}
 		}
 	default:
-		panic(senderHintMostRecentCommonAncestorStepCleanupMode)
+		panic(hintMostRecentCommonAncestorStepCleanupMode)
 	}
 	if !doStepCleanup {
 		log.Info("skipping cleanup of prior invocations' step holds due to environment variable setting")
@@ -218,7 +218,7 @@ func (m *HintMostRecentCommonAncestorStepCleanupMode) Set(s string) error {
 	return nil
 }
 
-var senderHintMostRecentCommonAncestorStepCleanupMode = *envconst.Var("ZREPL_ENDPOINT_SENDER_HINT_MOST_RECENT_STEP_HOLD_CLEANUP_MODE", &StepCleanupRangeSinceReplicationCursor).(*HintMostRecentCommonAncestorStepCleanupMode)
+var hintMostRecentCommonAncestorStepCleanupMode = *envconst.Var("ZREPL_ENDPOINT_HINT_MOST_RECENT_STEP_HOLD_CLEANUP_MODE", &StepCleanupRangeSinceReplicationCursor).(*HintMostRecentCommonAncestorStepCleanupMode)
 
 var maxConcurrentZFSSendSemaphore = semaphore.New(envconst.Int64("ZREPL_ENDPOINT_MAX_CONCURRENT_SEND", 10))
 
@@ -648,7 +648,7 @@ func (s *Receiver) ListFilesystemVersions(ctx context.Context, req *pdu.ListFile
 	}
 	// TODO share following code with sender
 
-	fsvs, err := zfs.ZFSListFilesystemVersions(ctx, lp, zfs.ListFilesystemVersionsOptions{})
+	fsvs, err := zfs.ZFSListFilesystemVersions(lp, zfs.ListFilesystemVersionsOptions{})
 	if err != nil {
 		return nil, err
 	}
@@ -806,7 +806,7 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 	if err := zfs.ZFSRecv(ctx, lp.ToString(), to, receive, recvOpts); err != nil {
 		getLogger(ctx).
 			WithError(err).
-			WithField("opts", fmt.Sprintf("%#v", recvOpts)).
+			WithField("opts", recvOpts).
 			Error("zfs receive failed")
 		return nil, err
 	}

endpoint/endpoint_zfs_abstraction.go

@@ -467,7 +467,7 @@ func (e ListAbstractionsErrors) Error() string {
 	}
 	msgs := make([]string, len(e))
 	for i := range e {
-		msgs[i] = e[i].Error()
+		msgs[i] = e.Error()
 	}
 	return fmt.Sprintf("list endpoint abstractions: multiple errors:\n%s", strings.Join(msgs, "\n"))
 }
@@ -573,7 +573,7 @@ func listAbstractionsImplFS(ctx context.Context, fs string, query *ListZFSHoldsA
 			whatTypes[zfs.Snapshot] = true
 		}
 	}
-	fsvs, err := zfs.ZFSListFilesystemVersions(ctx, fsp, zfs.ListFilesystemVersionsOptions{
+	fsvs, err := zfs.ZFSListFilesystemVersions(fsp, zfs.ListFilesystemVersionsOptions{
 		Types: whatTypes,
 	})
 	if err != nil {
@@ -731,9 +731,6 @@ func listStaleFiltering(abs []Abstraction, sinceBound *CreateTXGRangeBound) *Sta
 	}
 	stepFirstNotStaleCandidates := make(map[fsAndJobId]stepFirstNotStaleCandidate) // empty map => will always return nil
 	for _, a := range abs {
-		if a.GetJobID() == nil {
-			continue // already put those into always-live list noJobId in above loop
-		}
 		key := fsAndJobId{a.GetFS(), *a.GetJobID()}
 		c := stepFirstNotStaleCandidates[key]
 

go.mod

@@ -21,7 +21,7 @@ require (
 	github.com/onsi/gomega v1.7.0 // indirect
 	github.com/pkg/errors v0.8.1
 	github.com/pkg/profile v1.2.1
-	github.com/problame/go-netssh v0.0.0-20200601114649-26439f9f0dc5
+	github.com/problame/go-netssh v0.0.0-20191209123953-18d8aa6923c7
 	github.com/prometheus/client_golang v1.2.1
 	github.com/sergi/go-diff v1.0.1-0.20180205163309-da645544ed44 // go1.12 thinks it needs this
 	github.com/spf13/cobra v0.0.2

go.sum

@@ -208,8 +208,6 @@ github.com/problame/go-netssh v0.0.0-20191026123024-f34099f4f6b1 h1:HH8yzlaZq/A8
 github.com/problame/go-netssh v0.0.0-20191026123024-f34099f4f6b1/go.mod h1:JRs3nyBjvOv/9YHC+Vlw9z1Jfzl5s5t0BNnTzmclj1c=
 github.com/problame/go-netssh v0.0.0-20191209123953-18d8aa6923c7 h1:Oik8tLrLhoMq4fduDRuNNOAQ+q0M0dXkjAYLUf4+mAc=
 github.com/problame/go-netssh v0.0.0-20191209123953-18d8aa6923c7/go.mod h1:Ba6RaFpK1+IHWs1wLZ6sCBuXkt4iAVLeOG5GinXHusM=
-github.com/problame/go-netssh v0.0.0-20200601114649-26439f9f0dc5 h1:1eSrO2WAeSXjMol8WRKW9LekL1252VIMaKQwWkmEdvs=
-github.com/problame/go-netssh v0.0.0-20200601114649-26439f9f0dc5/go.mod h1:Ba6RaFpK1+IHWs1wLZ6sCBuXkt4iAVLeOG5GinXHusM=
 github.com/prometheus/client_golang v0.0.0-20180410130117-e11c6ff8170b/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
 github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
 github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=
@@ -265,7 +263,6 @@ github.com/spf13/viper v1.0.2/go.mod h1:A8kyI5cUJhb8N+3pkfONlcEcZbueH6nhAm0Fq7Sr
 github.com/spf13/viper v1.3.2/go.mod h1:ZiWeW+zYFKm7srdB9IoDzzZXaJaI5eL9QjNiN/DMA2s=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/testify v1.2.1/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=

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

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

platformtest/tests/listFilesystemVersions.go

@@ -43,7 +43,7 @@ func ListFilesystemVersionsTypeFilteringAndPrefix(t *platformtest.Context) {
 	fs := fmt.Sprintf("%s/foo bar", t.RootDataset)
 
 	// no options := all types
-	vs, err := zfs.ZFSListFilesystemVersions(t, mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
+	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",
@@ -51,21 +51,21 @@ func ListFilesystemVersionsTypeFilteringAndPrefix(t *platformtest.Context) {
 	}, versionRelnamesSorted(vs))
 
 	// just snapshots
-	vs, err = zfs.ZFSListFilesystemVersions(t, mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{
+	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(t, mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{
+	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(t, mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{
+	vs, err = zfs.ZFSListFilesystemVersions(mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{
 		ShortnamePrefix: "foo",
 	})
 	require.NoError(t, err)
@@ -82,7 +82,7 @@ func ListFilesystemVersionsZeroExistIsNotAnError(t *platformtest.Context) {
 
 	fs := fmt.Sprintf("%s/foo bar", t.RootDataset)
 
-	vs, err := zfs.ZFSListFilesystemVersions(t, mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
+	vs, err := zfs.ZFSListFilesystemVersions(mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
 	require.Empty(t, vs)
 	require.NoError(t, err)
 	dsne, ok := err.(*zfs.DatasetDoesNotExist)
@@ -98,7 +98,7 @@ func ListFilesystemVersionsFilesystemNotExist(t *platformtest.Context) {
 
 	nonexistentFS := fmt.Sprintf("%s/not existent", t.RootDataset)
 
-	vs, err := zfs.ZFSListFilesystemVersions(t, mustDatasetPath(nonexistentFS), zfs.ListFilesystemVersionsOptions{})
+	vs, err := zfs.ZFSListFilesystemVersions(mustDatasetPath(nonexistentFS), zfs.ListFilesystemVersionsOptions{})
 	require.Empty(t, vs)
 	require.Error(t, err)
 	t.Logf("err = %T\n%s", err, err)
@@ -141,7 +141,7 @@ func ListFilesystemVersionsUserrefs(t *platformtest.Context) {
 
 	fs := fmt.Sprintf("%s/foo bar", t.RootDataset)
 
-	vs, err := zfs.ZFSListFilesystemVersions(t, mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
+	vs, err := zfs.ZFSListFilesystemVersions(mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
 	require.NoError(t, err)
 
 	type expectation struct {

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

replication/driver/replication_driver.go

@@ -355,7 +355,7 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 	// invariant: prevs contains an entry for each unambiguous correspondence
 
 	stepQueue := newStepQueue()
-	defer stepQueue.Start(envconst.Int("ZREPL_REPLICATION_EXPERIMENTAL_REPLICATION_CONCURRENCY", 1))() // TODO parallel replication
+	defer stepQueue.Start(1)() // TODO parallel replication
 	var fssesDone sync.WaitGroup
 	for _, f := range a.fss {
 		fssesDone.Add(1)
@@ -370,57 +370,54 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 	a.finishedAt = time.Now()
 }
 
-func (f *fs) debug(format string, args ...interface{}) {
-	debugPrefix("fs=%s", f.fs.ReportInfo().Name)(format, args...)
-}
+func (fs *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 
-func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
-
-	defer f.l.Lock().Unlock()
+	defer fs.l.Lock().Unlock()
 
 	// get planned steps from replication logic
 	var psteps []Step
 	var errTime time.Time
 	var err error
-	f.l.DropWhile(func() {
+	fs.l.DropWhile(func() {
 		// TODO hacky
 		// choose target time that is earlier than any snapshot, so fs planning is always prioritized
 		targetDate := time.Unix(0, 0)
-		defer pq.WaitReady(f, targetDate)()
-		psteps, err = f.fs.PlanFS(ctx) // no shadow
+		defer pq.WaitReady(fs, targetDate)()
+		psteps, err = fs.fs.PlanFS(ctx) // no shadow
 		errTime = time.Now()            // no shadow
 	})
-	f.planning.done = true
+	debug := debugPrefix("fs=%s", fs.fs.ReportInfo().Name)
+	fs.planning.done = true
 	if err != nil {
-		f.planning.err = newTimedError(err, errTime)
+		fs.planning.err = newTimedError(err, errTime)
 		return
 	}
 	for _, pstep := range psteps {
 		step := &step{
-			l:    f.l,
+			l:    fs.l,
 			step: pstep,
 		}
-		f.planned.steps = append(f.planned.steps, step)
+		fs.planned.steps = append(fs.planned.steps, step)
 	}
-	f.debug("initial len(fs.planned.steps) = %d", len(f.planned.steps))
+	debug("initial 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
 	if prev != nil && prev.planning.done && prev.planning.err == nil {
 		prevUncompleted := prev.planned.steps[prev.planned.step:]
 		if len(prevUncompleted) == 0 {
-			f.debug("prevUncompleted is empty")
+			debug("prevUncompleted is empty")
 			return
 		}
-		if len(f.planned.steps) == 0 {
-			f.debug("fs.planned.steps is empty")
+		if len(fs.planned.steps) == 0 {
+			debug("fs.planned.steps is empty")
 			return
 		}
 		prevFailed := prevUncompleted[0]
-		curFirst := f.planned.steps[0]
+		curFirst := fs.planned.steps[0]
 		// we assume that PlanFS retries prevFailed (using curFirst)
 		if !prevFailed.step.TargetEquals(curFirst.step) {
-			f.debug("Targets don't match")
+			debug("Targets don't match")
 			// Two options:
 			// A: planning algorithm is broken
 			// B: manual user intervention inbetween
@@ -436,43 +433,43 @@ func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 			}
 			msg := fmt.Sprintf("last attempt's uncompleted step %s does not correspond to this attempt's first planned step %s",
 				stepFmt(prevFailed), stepFmt(curFirst))
-			f.planned.stepErr = newTimedError(errors.New(msg), time.Now())
+			fs.planned.stepErr = newTimedError(errors.New(msg), time.Now())
 			return
 		}
 		// only allow until step targets diverge
 		min := len(prevUncompleted)
-		if min > len(f.planned.steps) {
-			min = len(f.planned.steps)
+		if min > len(fs.planned.steps) {
+			min = len(fs.planned.steps)
 		}
 		diverge := 0
 		for ; diverge < min; diverge++ {
-			f.debug("diverge compare iteration %d", diverge)
-			if !f.planned.steps[diverge].step.TargetEquals(prevUncompleted[diverge].step) {
+			debug("diverge compare iteration %d", diverge)
+			if !fs.planned.steps[diverge].step.TargetEquals(prevUncompleted[diverge].step) {
 				break
 			}
 		}
-		f.debug("diverge is %d", diverge)
-		f.planned.steps = f.planned.steps[0:diverge]
+		debug("diverge is %d", diverge)
+		fs.planned.steps = fs.planned.steps[0:diverge]
 	}
-	f.debug("post-prev-merge len(fs.planned.steps) = %d", len(f.planned.steps))
+	debug("post-prev-merge len(fs.planned.steps) = %d", len(fs.planned.steps))
 
-	for i, s := range f.planned.steps {
+	for i, s := range fs.planned.steps {
 		var (
 			err     error
 			errTime time.Time
 		)
 		// lock must not be held while executing step in order for reporting to work
-		f.l.DropWhile(func() {
+		fs.l.DropWhile(func() {
 			targetDate := s.step.TargetDate()
-			defer pq.WaitReady(f, targetDate)()
+			defer pq.WaitReady(fs, targetDate)()
 			err = s.step.Step(ctx) // no shadow
 			errTime = time.Now()   // no shadow
 		})
 		if err != nil {
-			f.planned.stepErr = newTimedError(err, errTime)
+			fs.planned.stepErr = newTimedError(err, errTime)
 			break
 		}
-		f.planned.step = i + 1 // fs.planned.step must be == len(fs.planned.steps) if all went OK
+		fs.planned.step = i + 1 // fs.planned.step must be == len(fs.planned.steps) if all went OK
 	}
 }
 

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)
@@ -1078,15 +1078,6 @@ 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.
 	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

@@ -129,16 +129,6 @@ 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;
 }
 message HintMostRecentCommonAncestorRes {}

replication/logic/replication_logic.go

@@ -602,7 +602,7 @@ func (s *Step) doReplication(ctx context.Context) error {
 
 	fs := s.parent.Path
 
-	log := getLogger(ctx).WithField("filesystem", fs)
+	log := getLogger(ctx)
 	sr := s.buildSendRequest(false)
 
 	log.Debug("initiate send request")

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

transport/tcp/serve_tcp.go

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

transport/tcp/serve_tcp_ipmap.go

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

transport/tcp/serve_tcp_ipmap_test.go

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

transport/transport.go

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

zfs/versions.go

@@ -206,13 +206,13 @@ func (o *ListFilesystemVersionsOptions) matches(v FilesystemVersion) bool {
 }
 
 // returned versions are sorted by createtxg FIXME drop sort by createtxg requirement
-func ZFSListFilesystemVersions(ctx context.Context, fs *DatasetPath, options ListFilesystemVersionsOptions) (res []FilesystemVersion, err error) {
+func ZFSListFilesystemVersions(fs *DatasetPath, options ListFilesystemVersionsOptions) (res []FilesystemVersion, err error) {
 	listResults := make(chan ZFSListResult)
 
 	promTimer := prometheus.NewTimer(prom.ZFSListFilesystemVersionDuration.WithLabelValues(fs.ToString()))
 	defer promTimer.ObserveDuration()
 
-	ctx, cancel := context.WithCancel(ctx)
+	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	go ZFSListChan(ctx, listResults,
 		[]string{"name", "guid", "createtxg", "creation", "userrefs"},

zfs/zfs.go

@@ -1084,7 +1084,7 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, streamCopier
 	if opts.RollbackAndForceRecv {
 		// destroy all snapshots before `recv -F` because `recv -F`
 		// does not perform a rollback unless `send -R` was used (which we assume hasn't been the case)
-		snaps, err := ZFSListFilesystemVersions(ctx, fsdp, ListFilesystemVersionsOptions{
+		snaps, err := ZFSListFilesystemVersions(fsdp, ListFilesystemVersionsOptions{
 			Types: Snapshots,
 		})
 		if err != nil {

zfs/zfscmd/zfscmd.go

@@ -21,7 +21,7 @@ type Cmd struct {
 	cmd                       *exec.Cmd
 	ctx                       context.Context
 	mtx                       sync.RWMutex
-	startedAt, waitStartedAt, waitReturnedAt time.Time
+	startedAt, waitReturnedAt time.Time
 }
 
 func CommandContext(ctx context.Context, name string, arg ...string) *Cmd {
@@ -119,65 +119,19 @@ func (c *Cmd) startPost(err error) {
 }
 
 func (c *Cmd) waitPre() {
-	now := time.Now()
-
-	// ignore duplicate waits
-	c.mtx.Lock()
-	// ignore duplicate waits
-	if !c.waitStartedAt.IsZero() {
-		c.mtx.Unlock()
-		return
-	}
-	c.waitStartedAt = now
-	c.mtx.Unlock()
-
-	waitPreLogging(c, now)
-}
-
-type usage struct {
-	total_secs, system_secs, user_secs float64
+	waitPreLogging(c, time.Now())
 }
 
 func (c *Cmd) waitPost(err error) {
 	now := time.Now()
 
 	c.mtx.Lock()
-	// ignore duplicate waits
-	if !c.waitReturnedAt.IsZero() {
-		c.mtx.Unlock()
-		return
-	}
 	c.waitReturnedAt = now
 	c.mtx.Unlock()
 
-	// build usage
-	var u usage
-	{
-		var s *os.ProcessState
-		if err == nil {
-			s = c.cmd.ProcessState
-		} else if ee, ok := err.(*exec.ExitError); ok {
-			s = ee.ProcessState
-		}
-
-		if s == nil {
-			u = usage{
-				total_secs:  c.Runtime().Seconds(),
-				system_secs: -1,
-				user_secs:   -1,
-			}
-		} else {
-			u = usage{
-				total_secs:  c.Runtime().Seconds(),
-				system_secs: s.SystemTime().Seconds(),
-				user_secs:   s.UserTime().Seconds(),
-			}
-		}
-	}
-
-	waitPostReport(c, u, now)
-	waitPostLogging(c, u, err, now)
-	waitPostPrometheus(c, u, err, now)
+	waitPostReport(c, now)
+	waitPostLogging(c, err, now)
+	waitPostPrometheus(c, err, now)
 }
 
 // returns 0 if the command did not yet finish

zfs/zfscmd/zfscmd_logging.go

@@ -27,12 +27,11 @@ func waitPreLogging(c *Cmd, now time.Time) {
 	c.log().Debug("start waiting")
 }
 
-func waitPostLogging(c *Cmd, u usage, err error, now time.Time) {
-
+func waitPostLogging(c *Cmd, err error, now time.Time) {
 	log := c.log().
-		WithField("total_time_s", u.total_secs).
-		WithField("systemtime_s", u.system_secs).
-		WithField("usertime_s", u.user_secs)
+		WithField("total_time_s", c.Runtime().Seconds()).
+		WithField("systemtime_s", c.cmd.ProcessState.SystemTime().Seconds()).
+		WithField("usertime_s", c.cmd.ProcessState.UserTime().Seconds())
 
 	if err == nil {
 		log.Info("command exited without error")

zfs/zfscmd/zfscmd_platform_test.go

@@ -1,9 +1,7 @@
 package zfscmd
 
 import (
-	"bufio"
 	"bytes"
-	"context"
 	"io"
 	"os/exec"
 	"testing"
@@ -60,30 +58,3 @@ func TestCmdStderrBehaviorStdoutPipe(t *testing.T) {
 	require.True(t, ok)
 	require.Empty(t, ee.Stderr) // !!!!! probably not what one would expect if we only redirect stdout
 }
-
-func TestCmdProcessState(t *testing.T) {
-
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	cmd := exec.CommandContext(ctx, "bash", "-c", "echo running; sleep 3600")
-	stdout, err := cmd.StdoutPipe()
-	require.NoError(t, err)
-	err = cmd.Start()
-	require.NoError(t, err)
-
-	r := bufio.NewReader(stdout)
-	line, err := r.ReadString('\n')
-	require.NoError(t, err)
-	require.Equal(t, "running\n", line)
-
-	// we know it's running and sleeping
-	cancel()
-	err = cmd.Wait()
-	t.Logf("wait err %T\n%s", err, err)
-	require.Error(t, err)
-	ee, ok := err.(*exec.ExitError)
-	require.True(t, ok)
-	require.NotNil(t, ee.ProcessState)
-	require.Contains(t, ee.Error(), "killed")
-
-}

zfs/zfscmd/zfscmd_prometheus.go

@@ -46,7 +46,7 @@ func RegisterMetrics(r prometheus.Registerer) {
 	r.MustRegister(metrics.usertime)
 }
 
-func waitPostPrometheus(c *Cmd, u usage, err error, now time.Time) {
+func waitPostPrometheus(c *Cmd, err error, now time.Time) {
 
 	if len(c.cmd.Args) < 2 {
 		getLogger(c.ctx).WithField("args", c.cmd.Args).
@@ -64,10 +64,10 @@ func waitPostPrometheus(c *Cmd, u usage, err error, now time.Time) {
 
 	metrics.totaltime.
 		WithLabelValues(labelValues...).
-		Observe(u.total_secs)
+		Observe(c.Runtime().Seconds())
 	metrics.systemtime.WithLabelValues(labelValues...).
-		Observe(u.system_secs)
+		Observe(c.cmd.ProcessState.SystemTime().Seconds())
 	metrics.usertime.WithLabelValues(labelValues...).
-		Observe(u.user_secs)
+		Observe(c.cmd.ProcessState.UserTime().Seconds())
 
 }

zfs/zfscmd/zfscmd_report.go

@@ -57,7 +57,7 @@ func startPostReport(c *Cmd, err error, now time.Time) {
 	active.mtx.Unlock()
 }
 
-func waitPostReport(c *Cmd, _ usage, now time.Time) {
+func waitPostReport(c *Cmd, now time.Time) {
 	active.mtx.Lock()
 	defer active.mtx.Unlock()
 	prev := active.cmds[c]