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_BUILDFLAGS := $(GO_MOD_READONLY) $(GO_EXTRA_BUILDFLAGS)
 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 "$@"
+	artifacts/zrepl-$(GOHOSTOS)-$(GOHOSTARCH) bashcomp "$@"
-
-$(ARTIFACTDIR)/_zrepl.zsh_completion:
-	$(MAKE) zrepl-bin GOOS=$(GOHOSTOS) GOARCH=$(GOHOSTARCH)
-	artifacts/zrepl-$(GOHOSTOS)-$(GOHOSTARCH) gencompletion zsh "$@"
 
 $(ARTIFACTDIR)/go_env.txt:
 	$(GO_ENV_VARS) $(GO) env > $@

cli/cli.go

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

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.StringVarP(&holdsCreateStepHoldFlags.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.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{
+	HoldsCmd = &cli.Subcommand{
-		Use:   "zfs-abstraction",
+		Use:   "holds",
-		Short: "manage abstractions that zrepl builds on top of ZFS",
+		Short: "manage holds & step bookmarks",
 		SetupSubcommands: func() []*cli.Subcommand {
 			return []*cli.Subcommand{
-				zabsCmdList,
+				holdsCmdList,
-				zabsCmdReleaseAll,
+				holdsCmdReleaseAll,
-				zabsCmdReleaseStale,
+				holdsCmdReleaseStale,
-				zabsCmdCreate,
+				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 {
+var holdsListFlags struct {
-	Filter zabsFilterFlags
+	Filter holdsFilterFlags
 	Json   bool
 }
 
-var zabsCmdList = &cli.Subcommand{
+var holdsCmdList = &cli.Subcommand{
 	Use:             "list",
-	Short:           `list zrepl ZFS abstractions`,
+	Run:             doHoldsList,
-	Run:             doZabsList,
 	NoRequireConfig: true,
+	Short:           "list holds and bookmarks",
 	SetupFlags: func(f *pflag.FlagSet) {
-		zabsListFlags.Filter.registerZabsFilterFlags(f, "list")
+		holdsListFlags.Filter.registerHoldsFilterFlags(f, "list")
-		f.BoolVar(&zabsListFlags.Json, "json", false, "emit JSON")
+		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 {
+var holdsReleaseFlags struct {
-	Filter zabsFilterFlags
+	Filter holdsFilterFlags
 	Json   bool
 	DryRun bool
 }
 
-func registerZabsReleaseFlags(s *pflag.FlagSet) {
+func registerHoldsReleaseFlags(s *pflag.FlagSet) {
-	zabsReleaseFlags.Filter.registerZabsFilterFlags(s, "release")
+	holdsReleaseFlags.Filter.registerHoldsFilterFlags(s, "release")
-	s.BoolVar(&zabsReleaseFlags.Json, "json", false, "emit json instead of pretty-printed")
+	s.BoolVar(&holdsReleaseFlags.Json, "json", false, "emit json instead of pretty-printed")
-	s.BoolVar(&zabsReleaseFlags.DryRun, "dry-run", false, "do a dry-run")
+	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)`,
+	Short:           `(DANGEROUS) release all zrepl-managed holds and bookmarks, mostly useful for uninstalling zrepl`,
-	SetupFlags:      registerZabsReleaseFlags,
+	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)`,
+	Short:           `release stale zrepl-managed holds and boomkarks (useful if zrepl has a bug and doesn't do it by itself)`,
-	SetupFlags:      registerZabsReleaseFlags,
+	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 holdsReleaseFlags.DryRun {
-		if zabsReleaseFlags.Json {
+		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,
-		}
-	},
-}

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/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``
+    * - ``zrepl holds``
-      - 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>` )
+      - 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

@@ -2,17 +2,14 @@
 package endpoint
 
 import (
-	"bytes"
 	"context"
 	"fmt"
-	"io"
 	"path"
 	"sync"
 
 	"github.com/pkg/errors"
 
 	"github.com/zrepl/zrepl/replication/logic/pdu"
-	"github.com/zrepl/zrepl/util/chainedio"
 	"github.com/zrepl/zrepl/util/chainlock"
 	"github.com/zrepl/zrepl/util/envconst"
 	"github.com/zrepl/zrepl/util/semaphore"
@@ -99,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
 	}
@@ -153,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:
@@ -183,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")
@@ -221,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))
 
@@ -244,7 +241,7 @@ func sendArgsFromPDUAndValidateExistsAndGetVersion(ctx context.Context, fs strin
 	return version, nil
 }
 
-func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
+func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
 
 	_, err := s.filterCheckFS(r.Filesystem)
 	if err != nil {
@@ -342,11 +339,11 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
 
 	// step holds & replication cursor released / moved forward in s.SendCompleted => s.moveCursorAndReleaseSendHolds
 
-	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
+	streamCopier, err := zfs.ZFSSend(ctx, sendArgs)
 	if err != nil {
 		return nil, nil, errors.Wrap(err, "zfs send failed")
 	}
-	return res, sendStream, nil
+	return res, streamCopier, nil
 }
 
 func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
@@ -479,7 +476,7 @@ func (p *Sender) ReplicationCursor(ctx context.Context, req *pdu.ReplicationCurs
 	return &pdu.ReplicationCursorRes{Result: &pdu.ReplicationCursorRes_Guid{Guid: cursor.Guid}}, nil
 }
 
-func (p *Sender) Receive(ctx context.Context, r *pdu.ReceiveReq, _ io.ReadCloser) (*pdu.ReceiveRes, error) {
+func (p *Sender) Receive(ctx context.Context, r *pdu.ReceiveReq, receive zfs.StreamCopier) (*pdu.ReceiveRes, error) {
 	return nil, fmt.Errorf("sender does not implement Receive()")
 }
 
@@ -651,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
 	}
@@ -683,13 +680,13 @@ func (s *Receiver) ReplicationCursor(context.Context, *pdu.ReplicationCursorReq)
 	return nil, fmt.Errorf("ReplicationCursor not implemented for Receiver")
 }
 
-func (s *Receiver) Send(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
+func (s *Receiver) Send(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
 	return nil, nil, fmt.Errorf("receiver does not implement Send()")
 }
 
 var maxConcurrentZFSRecvSemaphore = semaphore.New(envconst.Int64("ZREPL_ENDPOINT_MAX_CONCURRENT_RECV", 10))
 
-func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.ReadCloser) (*pdu.ReceiveRes, error) {
+func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs.StreamCopier) (*pdu.ReceiveRes, error) {
 	getLogger(ctx).Debug("incoming Receive")
 	defer receive.Close()
 
@@ -768,30 +765,21 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.
 		return nil, visitErr
 	}
 
-	log := getLogger(ctx).WithField("proto_fs", req.GetFilesystem()).WithField("local_fs", lp.ToString())
-
 	// determine whether we need to rollback the filesystem / change its placeholder state
 	var clearPlaceholderProperty bool
 	var recvOpts zfs.RecvOptions
 	ph, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, lp)
-	if err != nil {
+	if err == nil && ph.FSExists && ph.IsPlaceholder {
-		return nil, errors.Wrap(err, "cannot get placeholder state")
-	}
-	log.WithField("placeholder_state", fmt.Sprintf("%#v", ph)).Debug("placeholder state")
-	if ph.FSExists && ph.IsPlaceholder {
 		recvOpts.RollbackAndForceRecv = true
 		clearPlaceholderProperty = true
 	}
-
 	if clearPlaceholderProperty {
-		log.Info("clearing placeholder property")
 		if err := zfs.ZFSSetPlaceholder(ctx, lp, false); err != nil {
 			return nil, fmt.Errorf("cannot clear placeholder property for forced receive: %s", err)
 		}
 	}
 
 	if req.ClearResumeToken && ph.FSExists {
-		log.Info("clearing resume token")
 		if err := zfs.ZFSRecvClearResumeToken(ctx, lp.ToString()); err != nil {
 			return nil, errors.Wrap(err, "cannot clear resume token")
 		}
@@ -802,7 +790,7 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.
 		return nil, errors.Wrap(err, "cannot determine whether we can use resumable send & recv")
 	}
 
-	log.Debug("acquire concurrent recv semaphore")
+	getLogger(ctx).Debug("acquire concurrent recv semaphore")
 	// TODO use try-acquire and fail with resource-exhaustion rpc status
 	// => would require handling on the client-side
 	// => this is a dataconn endpoint, doesn't have the status code semantics of gRPC
@@ -812,89 +800,14 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.
 	}
 	defer guard.Release()
 
-	log.Info("peeking 1M ahead")
+	getLogger(ctx).WithField("opts", fmt.Sprintf("%#v", recvOpts)).Debug("start receive command")
-	var peek bytes.Buffer
-	var MaxPeek = envconst.Int64("ZREPL_ENDPOINT_RECV_PEEK_SIZE", 1<<20)
-	if _, err := io.Copy(&peek, io.LimitReader(receive, MaxPeek)); err != nil {
-		log.WithError(err).Error("cannot read peek-buffer from send stream")
-	}
-	var peekCopy bytes.Buffer
-	if n, err := peekCopy.Write(peek.Bytes()); err != nil || n != peek.Len() {
-		panic(peek.Len())
-	}
-
-	log.WithField("opts", fmt.Sprintf("%#v", recvOpts)).Debug("start receive command")
 
 	snapFullPath := to.FullPath(lp.ToString())
-	if err := zfs.ZFSRecv(ctx, lp.ToString(), to, chainedio.NewChainedReader(&peek, receive), recvOpts); err != nil {
+	if err := zfs.ZFSRecv(ctx, lp.ToString(), to, receive, recvOpts); err != nil {
-
+		getLogger(ctx).
-		// best-effort rollback of placeholder state if the recv didn't start
-		_, resumableStatePresent := err.(*zfs.RecvFailedWithResumeTokenErr)
-		disablePlaceholderRestoration := envconst.Bool("ZREPL_ENDPOINT_DISABLE_PLACEHOLDER_RESTORATION", false)
-		placeholderRestored := !ph.IsPlaceholder
-		if !disablePlaceholderRestoration && !resumableStatePresent && recvOpts.RollbackAndForceRecv && ph.FSExists && ph.IsPlaceholder && clearPlaceholderProperty {
-			log.Info("restoring placeholder property")
-			if phErr := zfs.ZFSSetPlaceholder(ctx, lp, true); phErr != nil {
-				log.WithError(phErr).Error("cannot restore placeholder property after failed receive, subsequent replications will likely fail with a different error")
-				// fallthrough
-			} else {
-				placeholderRestored = true
-			}
-			// fallthrough
-		}
-
-		// deal with failing initial encrypted send & recv
-		if _, ok := err.(*zfs.RecvDestroyOrOverwriteEncryptedErr); ok && ph.IsPlaceholder && placeholderRestored {
-			msg := `cannot automatically replace placeholder filesystem with incoming send stream - please see receive-side log for details`
-			err := errors.New(msg)
-			log.Error(msg)
-
-			log.Error(`zrepl creates placeholder filesystems on the receiving side of a replication to match the sending side's dataset hierarchy`)
-			log.Error(`zrepl uses zfs receive -F to replace those placeholders with incoming full sends`)
-			log.Error(`OpenZFS native encryption prohibits zfs receive -F for encrypted filesystems`)
-			log.Error(`the current zrepl placeholder filesystem concept is thus incompatible with OpenZFS native encryption`)
-
-			tempStartFullRecvFS := lp.Copy().ToString() + ".zrepl.initial-recv"
-			tempStartFullRecvFSDP, dpErr := zfs.NewDatasetPath(tempStartFullRecvFS)
-			if dpErr != nil {
-				log.WithError(dpErr).Error("cannot determine temporary filesystem name for initial encrypted recv workaround")
-				return nil, err // yes, err, not dpErr
-			}
-
-			log := log.WithField("temp_recv_fs", tempStartFullRecvFS)
-			log.Error(`as a workaround, zrepl will now attempt to re-receive the beginning of the stream into a temporary filesystem temp_recv_fs`)
-			log.Error(`if that step succeeds: shut down zrepl and use 'zfs rename' to swap temp_recv_fs with local_fs, then restart zrepl`)
-			log.Error(`replication will then resume using resumable send+recv`)
-
-			tempPH, phErr := zfs.ZFSGetFilesystemPlaceholderState(ctx, tempStartFullRecvFSDP)
-			if phErr != nil {
-				log.WithError(phErr).Error("cannot determine placeholder state of temp_recv_fs")
-				return nil, err // yes, err, not dpErr
-			}
-			if tempPH.FSExists {
-				log.Error("temp_recv_fs already exists, assuming a (partial) initial recv to that filesystem has already been done")
-				return nil, err
-			}
-
-			recvOpts.RollbackAndForceRecv = false
-			recvOpts.SavePartialRecvState = true
-			rerecvErr := zfs.ZFSRecv(ctx, tempStartFullRecvFS, to, chainedio.NewChainedReader(&peekCopy), recvOpts)
-			if _, isResumable := rerecvErr.(*zfs.RecvFailedWithResumeTokenErr); rerecvErr == nil || isResumable {
-				log.Error("completed re-receive into temporary filesystem temp_recv_fs, now shut down zrepl and use zfs rename to swap temp_recv_fs with local_fs")
-			} else {
-				log.WithError(rerecvErr).Error("failed to receive the beginning of the stream into temporary filesystem temp_recv_fs")
-				log.Error("we advise you to collect the error log and current configuration, open an issue on GitHub, and revert to your previous configuration in the meantime")
-			}
-
-			log.Error(`if you would like to see improvements to this situation, please open an issue on GitHub`)
-			return nil, err
-		}
-
-		log.
 			WithError(err).
-			WithField("opts", fmt.Sprintf("%#v", recvOpts)).
+			WithField("opts", recvOpts).
 			Error("zfs receive failed")
-
 		return nil, err
 	}
 
@@ -902,13 +815,13 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.
 	toRecvd, err := to.ValidateExistsAndGetVersion(ctx, lp.ToString())
 	if err != nil {
 		msg := "receive request's `To` version does not match what we received in the stream"
-		log.WithError(err).WithField("snap", snapFullPath).Error(msg)
+		getLogger(ctx).WithError(err).WithField("snap", snapFullPath).Error(msg)
-		log.Error("aborting recv request, but keeping received snapshot for inspection")
+		getLogger(ctx).Error("aborting recv request, but keeping received snapshot for inspection")
 		return nil, errors.Wrap(err, msg)
 	}
 
 	if s.conf.UpdateLastReceivedHold {
-		log.Debug("move last-received-hold")
+		getLogger(ctx).Debug("move last-received-hold")
 		if err := MoveLastReceivedHold(ctx, lp.ToString(), toRecvd, s.conf.JobID); err != nil {
 			return nil, errors.Wrap(err, "cannot move last-received-hold")
 		}

endpoint/endpoint_zfs_abstraction.go

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

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/tests/helpers.go

@@ -136,7 +136,7 @@ func makeResumeSituation(ctx *platformtest.Context, src dummySnapshotSituation,
 		return situation
 	}
 
-	limitedCopier := limitio.ReadCloser(copier, src.dummyDataLen/2)
+	limitedCopier := zfs.NewReadCloserCopier(limitio.ReadCloser(copier, src.dummyDataLen/2))
 	defer limitedCopier.Close()
 
 	require.NotNil(ctx, sendArgs.To)

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 {

platformtest/tests/sendArgsValidation.go

@@ -35,7 +35,7 @@ func SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden(ctx *platformt
 		ResumeToken: "",
 	}.Validate(ctx)
 
-	var stream *zfs.SendStream
+	var stream *zfs.ReadCloserCopier
 	if err == nil {
 		stream, err = zfs.ZFSSend(ctx, sendArgs) // no shadow
 		if err == nil {

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.
+* 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.
-
 * We also have [Notes on Planning and Executing Replication of Multiple Filesystems](#zrepl-algo-multiple-filesystems-notes)
 
 ---

replication/driver/replication_driver.go

@@ -146,12 +146,6 @@ type fs struct {
 
 	l *chainlock.L
 
-	// ordering relationship that must be maintained for initial replication
-	initialRepOrd struct {
-		parents, children []*fs
-		parentDidUpdate   chan struct{}
-	}
-
 	planning struct {
 		done bool
 		err  *timedError
@@ -302,7 +296,6 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 			fs: pfs,
 			l:  a.l,
 		}
-		fs.initialRepOrd.parentDidUpdate = make(chan struct{}, 1)
 		a.fss = append(a.fss, fs)
 	}
 
@@ -361,20 +354,8 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 	}
 	// invariant: prevs contains an entry for each unambiguous correspondence
 
-	// build up parent-child relationship (FIXME (O(n^2), but who's going to have that many filesystems...))
-	for _, f1 := range a.fss {
-		fs1 := f1.fs.ReportInfo().Name
-		for _, f2 := range a.fss {
-			fs2 := f2.fs.ReportInfo().Name
-			if strings.HasPrefix(fs1, fs2) && fs1 != fs2 {
-				f1.initialRepOrd.parents = append(f1.initialRepOrd.parents, f2)
-				f2.initialRepOrd.children = append(f2.initialRepOrd.children, f1)
-			}
-		}
-	}
-
 	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)
@@ -389,76 +370,54 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 	a.finishedAt = time.Now()
 }
 
-func (f *fs) debug(format string, args ...interface{}) {
+func (fs *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
-	debugPrefix("fs=%s", f.fs.ReportInfo().Name)(format, args...)
-}
 
-// wake up children that watch for f.{planning.{err,done},planned.{step,stepErr}}
+	defer fs.l.Lock().Unlock()
-func (f *fs) initialRepOrdWakeupChildren() {
-	var children []string
-	for _, c := range f.initialRepOrd.children {
-		// no locking required, c.fs does not change
-		children = append(children, c.fs.ReportInfo().Name)
-	}
-	f.debug("wakeup children %s", children)
-	for _, child := range f.initialRepOrd.children {
-		select {
-		// no locking required, child.initialRepOrd does not change
-		case child.initialRepOrd.parentDidUpdate <- struct{}{}:
-		default:
-		}
-	}
-}
-
-func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
-
-	defer f.l.Lock().Unlock()
-	defer f.initialRepOrdWakeupChildren()
 
 	// 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)()
+		defer pq.WaitReady(fs, targetDate)()
-		psteps, err = f.fs.PlanFS(ctx) // no shadow
+		psteps, err = fs.fs.PlanFS(ctx) // no shadow
-		errTime = time.Now()           // no shadow
+		errTime = time.Now()            // no shadow
 	})
+	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)
 	}
-	// we're not done planning yet, f.planned.steps might still be changed by next block
+	debug("initial len(fs.planned.steps) = %d", len(fs.planned.steps))
-	// => don't set f.planning.done just yet
-	f.debug("initial len(fs.planned.steps) = %d", len(f.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 {
+		if len(fs.planned.steps) == 0 {
-			f.debug("fs.planned.steps is empty")
+			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
@@ -474,130 +433,44 @@ 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) {
+		if min > len(fs.planned.steps) {
-			min = len(f.planned.steps)
+			min = len(fs.planned.steps)
 		}
 		diverge := 0
 		for ; diverge < min; diverge++ {
-			f.debug("diverge compare iteration %d", diverge)
+			debug("diverge compare iteration %d", diverge)
-			if !f.planned.steps[diverge].step.TargetEquals(prevUncompleted[diverge].step) {
+			if !fs.planned.steps[diverge].step.TargetEquals(prevUncompleted[diverge].step) {
 				break
 			}
 		}
-		f.debug("diverge is %d", diverge)
+		debug("diverge is %d", diverge)
-		f.planned.steps = f.planned.steps[0: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))
 
-	// now we are done planning (f.planned.steps won't change from now on)
+	for i, s := range fs.planned.steps {
-	f.planning.done = true
+		var (
-
+			err     error
-	// wait for parents' initial replication
+			errTime time.Time
-	var parents []string
+		)
-	for _, p := range f.initialRepOrd.parents {
-		parents = append(parents, p.fs.ReportInfo().Name)
-	}
-	f.debug("wait for parents %s", parents)
-	for {
-		var initialReplicatingParentsWithErrors []string
-		allParentsPresentOnReceiver := true
-		f.l.DropWhile(func() {
-			for _, p := range f.initialRepOrd.parents {
-				p.l.HoldWhile(func() {
-					// (get the preconditions that allow us to inspect p.planned)
-					parentHasPlanningDone := p.planning.done && p.planning.err == nil
-					if !parentHasPlanningDone {
-						// if the parent couldn't be planned, we cannot know whether it needs initial replication
-						// or incremental replication => be conservative and assume it was initial replication
-						allParentsPresentOnReceiver = false
-						if p.planning.err != nil {
-							initialReplicatingParentsWithErrors = append(initialReplicatingParentsWithErrors, p.fs.ReportInfo().Name)
-						}
-						return
-					}
-					// now allowed to inspect p.planned
-
-					// if there are no steps to be done, the filesystem must exist on the receiving side
-					// (otherwise we'd replicate it, and there would be a step for that)
-					// (FIXME hardcoded initial replication policy, assuming the policy will always do _some_ initial replication)
-					parentHasNoSteps := len(p.planned.steps) == 0
-
-					// OR if it has completed at least one step
-					// (remember that .step points to the next step to be done)
-					// (TODO technically, we could make this step ready in the moment the recv-side
-					//  dataset exists, i.e. after the first few megabytes of transferred data, but we'd have to ask the receiver for that -> poll ListFilesystems RPC)
-					parentHasTakenAtLeastOneSuccessfulStep := !parentHasNoSteps && p.planned.step >= 1
-
-					parentFirstStepIsIncremental := // no need to lock for .report() because step.l == it's fs.l
-						len(p.planned.steps) > 0 && p.planned.steps[0].report().IsIncremental()
-
-					f.debug("parentHasNoSteps=%v parentFirstStepIsIncremental=%v parentHasTakenAtLeastOneSuccessfulStep=%v",
-						parentHasNoSteps, parentFirstStepIsIncremental, parentHasTakenAtLeastOneSuccessfulStep)
-
-					parentPresentOnReceiver := parentHasNoSteps || parentFirstStepIsIncremental || parentHasTakenAtLeastOneSuccessfulStep
-
-					allParentsPresentOnReceiver = allParentsPresentOnReceiver && parentPresentOnReceiver // no shadow
-
-					if !parentPresentOnReceiver && p.planned.stepErr != nil {
-						initialReplicatingParentsWithErrors = append(initialReplicatingParentsWithErrors, p.fs.ReportInfo().Name)
-					}
-
-				})
-			}
-		})
-
-		if len(initialReplicatingParentsWithErrors) > 0 {
-			f.planned.stepErr = newTimedError(fmt.Errorf("parent(s) failed during initial replication: %s", initialReplicatingParentsWithErrors), time.Now())
-			return
-		}
-
-		if allParentsPresentOnReceiver {
-			break // good to go
-		}
-
-		// wait for wakeups from parents, then check again
-		// lock must not be held while waiting in order for reporting to work
-		f.l.DropWhile(func() {
-			select {
-			case <-ctx.Done():
-				f.planned.stepErr = newTimedError(ctx.Err(), time.Now())
-				return
-			case <-f.initialRepOrd.parentDidUpdate:
-				// loop
-			}
-		})
-		if f.planned.stepErr != nil {
-			return
-		}
-	}
-
-	f.debug("all parents ready, start replication %s", parents)
-
-	// do our steps
-	for i, s := range f.planned.steps {
 		// lock must not be held while executing step in order for reporting to work
-		f.l.DropWhile(func() {
+		fs.l.DropWhile(func() {
-			// wait for parallel replication
 			targetDate := s.step.TargetDate()
-			defer pq.WaitReady(f, targetDate)()
+			defer pq.WaitReady(fs, targetDate)()
-			// do the step
+			err = s.step.Step(ctx) // no shadow
-			err, errTime = s.step.Step(ctx), time.Now() // 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
-
-		f.initialRepOrdWakeupChildren()
 	}
-
 }
 
 // caller must hold lock l

replication/driver/replication_driver_debug.go

@@ -26,6 +26,6 @@ type debugFunc func(format string, args ...interface{})
 func debugPrefix(prefixFormat string, prefixFormatArgs ...interface{}) debugFunc {
 	prefix := fmt.Sprintf(prefixFormat, prefixFormatArgs...)
 	return func(format string, args ...interface{}) {
-		debug("%s: %s", prefix, fmt.Sprintf(format, args...))
+		debug("%s: %s", prefix, fmt.Sprintf(format, args))
 	}
 }

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"`
+	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

@@ -128,17 +128,7 @@ message PingRes {
 }
 
 message HintMostRecentCommonAncestorReq {
-  string Filesystem = 1;
+    string Filesystem = 1;
-
+    FilesystemVersion SenderVersion = 2;
-  // 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

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"io"
 	"sync"
 	"time"
 
@@ -39,7 +38,7 @@ type Sender interface {
 	// If a non-nil io.ReadCloser is returned, it is guaranteed to be closed before
 	// any next call to the parent github.com/zrepl/zrepl/replication.Endpoint.
 	// If the send request is for dry run the io.ReadCloser will be nil
-	Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error)
+	Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error)
 	SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error)
 	ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error)
 }
@@ -48,7 +47,7 @@ type Receiver interface {
 	Endpoint
 	// Receive sends r and sendStream (the latter containing a ZFS send stream)
 	// to the parent github.com/zrepl/zrepl/replication.Endpoint.
-	Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.ReadCloser) (*pdu.ReceiveRes, error)
+	Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs.StreamCopier) (*pdu.ReceiveRes, error)
 }
 
 type PlannerPolicy struct {
@@ -163,7 +162,7 @@ type Step struct {
 
 	// byteCounter is nil initially, and set later in Step.doReplication
 	// => concurrent read of that pointer from Step.ReportInfo must be protected
-	byteCounter    bytecounter.ReadCloser
+	byteCounter    bytecounter.StreamCopier
 	byteCounterMtx chainlock.L
 }
 
@@ -607,19 +606,19 @@ func (s *Step) doReplication(ctx context.Context) error {
 	sr := s.buildSendRequest(false)
 
 	log.Debug("initiate send request")
-	sres, stream, err := s.sender.Send(ctx, sr)
+	sres, sstreamCopier, err := s.sender.Send(ctx, sr)
 	if err != nil {
 		log.WithError(err).Error("send request failed")
 		return err
 	}
-	if stream == nil {
+	if sstreamCopier == nil {
 		err := errors.New("send request did not return a stream, broken endpoint implementation")
 		return err
 	}
-	defer stream.Close()
+	defer sstreamCopier.Close()
 
 	// Install a byte counter to track progress + for status report
-	byteCountingStream := bytecounter.NewReadCloser(stream)
+	byteCountingStream := bytecounter.NewStreamCopier(sstreamCopier)
 	s.byteCounterMtx.Lock()
 	s.byteCounter = byteCountingStream
 	s.byteCounterMtx.Unlock()

rpc/dataconn/dataconn_client.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"io"
 	"strings"
 
 	"github.com/golang/protobuf/proto"
@@ -12,6 +11,7 @@ import (
 	"github.com/zrepl/zrepl/replication/logic/pdu"
 	"github.com/zrepl/zrepl/rpc/dataconn/stream"
 	"github.com/zrepl/zrepl/transport"
+	"github.com/zrepl/zrepl/zfs"
 )
 
 type Client struct {
@@ -26,7 +26,7 @@ func NewClient(connecter transport.Connecter, log Logger) *Client {
 	}
 }
 
-func (c *Client) send(ctx context.Context, conn *stream.Conn, endpoint string, req proto.Message, stream io.ReadCloser) error {
+func (c *Client) send(ctx context.Context, conn *stream.Conn, endpoint string, req proto.Message, streamCopier zfs.StreamCopier) error {
 
 	var buf bytes.Buffer
 	_, memErr := buf.WriteString(endpoint)
@@ -46,8 +46,8 @@ func (c *Client) send(ctx context.Context, conn *stream.Conn, endpoint string, r
 		return err
 	}
 
-	if stream != nil {
+	if streamCopier != nil {
-		return conn.SendStream(ctx, stream, ZFSStream)
+		return conn.SendStream(ctx, streamCopier, ZFSStream)
 	} else {
 		return nil
 	}
@@ -109,7 +109,7 @@ func (c *Client) putWire(conn *stream.Conn) {
 	}
 }
 
-func (c *Client) ReqSend(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
+func (c *Client) ReqSend(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
 	conn, err := c.getWire(ctx)
 	if err != nil {
 		return nil, nil, err
@@ -130,19 +130,16 @@ func (c *Client) ReqSend(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, i
 		return nil, nil, err
 	}
 
-	var stream io.ReadCloser
+	var copier zfs.StreamCopier = nil
 	if !req.DryRun {
 		putWireOnReturn = false
-		stream, err = conn.ReadStream(ZFSStream, true) // no shadow
+		copier = &streamCopier{streamConn: conn, closeStreamOnClose: true}
-		if err != nil {
-			return nil, nil, err
-		}
 	}
 
-	return &res, stream, nil
+	return &res, copier, nil
 }
 
-func (c *Client) ReqRecv(ctx context.Context, req *pdu.ReceiveReq, stream io.ReadCloser) (*pdu.ReceiveRes, error) {
+func (c *Client) ReqRecv(ctx context.Context, req *pdu.ReceiveReq, streamCopier zfs.StreamCopier) (*pdu.ReceiveRes, error) {
 
 	defer c.log.Debug("ReqRecv returns")
 	conn, err := c.getWire(ctx)
@@ -169,7 +166,7 @@ func (c *Client) ReqRecv(ctx context.Context, req *pdu.ReceiveReq, stream io.Rea
 
 	sendErrChan := make(chan error)
 	go func() {
-		if err := c.send(ctx, conn, EndpointRecv, req, stream); err != nil {
+		if err := c.send(ctx, conn, EndpointRecv, req, streamCopier); err != nil {
 			sendErrChan <- err
 		} else {
 			sendErrChan <- nil

rpc/dataconn/dataconn_server.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"io"
 
 	"github.com/golang/protobuf/proto"
 
@@ -12,6 +11,7 @@ import (
 	"github.com/zrepl/zrepl/replication/logic/pdu"
 	"github.com/zrepl/zrepl/rpc/dataconn/stream"
 	"github.com/zrepl/zrepl/transport"
+	"github.com/zrepl/zrepl/zfs"
 )
 
 // WireInterceptor has a chance to exchange the context and connection on each client connection.
@@ -21,11 +21,11 @@ type WireInterceptor func(ctx context.Context, rawConn *transport.AuthConn) (con
 type Handler interface {
 	// Send handles a SendRequest.
 	// The returned io.ReadCloser is allowed to be nil, for example if the requested Send is a dry-run.
-	Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error)
+	Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error)
 	// Receive handles a ReceiveRequest.
 	// It is guaranteed that Server calls Receive with a stream that holds the IdleConnTimeout
 	// configured in ServerConfig.Shared.IdleConnTimeout.
-	Receive(ctx context.Context, r *pdu.ReceiveReq, receive io.ReadCloser) (*pdu.ReceiveRes, error)
+	Receive(ctx context.Context, r *pdu.ReceiveReq, receive zfs.StreamCopier) (*pdu.ReceiveRes, error)
 	// PingDataconn handles a PingReq
 	PingDataconn(ctx context.Context, r *pdu.PingReq) (*pdu.PingRes, error)
 }
@@ -111,7 +111,7 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
 	s.log.WithField("endpoint", endpoint).Debug("calling handler")
 
 	var res proto.Message
-	var sendStream io.ReadCloser
+	var sendStream zfs.StreamCopier
 	var handlerErr error
 	switch endpoint {
 	case EndpointSend:
@@ -127,12 +127,7 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
 			s.log.WithError(err).Error("cannot unmarshal receive request")
 			return
 		}
-		stream, err := c.ReadStream(ZFSStream, false)
+		res, handlerErr = s.h.Receive(ctx, &req, &streamCopier{streamConn: c, closeStreamOnClose: false}) // SHADOWING
-		if err != nil {
-			s.log.WithError(err).Error("cannot open stream in receive request")
-			return
-		}
-		res, handlerErr = s.h.Receive(ctx, &req, stream) // SHADOWING
 	case EndpointPing:
 		var req pdu.PingReq
 		if err := proto.Unmarshal(reqStructured, &req); err != nil {

rpc/dataconn/dataconn_shared.go

@@ -1,7 +1,12 @@
 package dataconn
 
 import (
+	"io"
+	"sync"
 	"time"
+
+	"github.com/zrepl/zrepl/rpc/dataconn/stream"
+	"github.com/zrepl/zrepl/zfs"
 )
 
 const (
@@ -34,3 +39,33 @@ const (
 	responseHeaderHandlerOk          = "HANDLER OK\n"
 	responseHeaderHandlerErrorPrefix = "HANDLER ERROR:\n"
 )
+
+type streamCopier struct {
+	mtx                sync.Mutex
+	used               bool
+	streamConn         *stream.Conn
+	closeStreamOnClose bool
+}
+
+// WriteStreamTo implements zfs.StreamCopier
+func (s *streamCopier) WriteStreamTo(w io.Writer) zfs.StreamCopierError {
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+	if s.used {
+		panic("streamCopier used multiple times")
+	}
+	s.used = true
+	return s.streamConn.ReadStreamInto(w, ZFSStream)
+}
+
+// Close implements zfs.StreamCopier
+func (s *streamCopier) Close() error {
+	// only record the close here, what we do actually depends on whether
+	// the streamCopier is instantiated server-side or client-side
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+	if s.closeStreamOnClose {
+		return s.streamConn.Close()
+	}
+	return nil
+}

rpc/dataconn/microbenchmark/microbenchmark.go

@@ -29,6 +29,7 @@ import (
 	"github.com/zrepl/zrepl/rpc/dataconn/timeoutconn"
 	"github.com/zrepl/zrepl/transport"
 	"github.com/zrepl/zrepl/util/devnoop"
+	"github.com/zrepl/zrepl/zfs"
 )
 
 func orDie(err error) {
@@ -41,9 +42,23 @@ type readerStreamCopier struct{ io.Reader }
 
 func (readerStreamCopier) Close() error { return nil }
 
+type readerStreamCopierErr struct {
+	error
+}
+
+func (readerStreamCopierErr) IsReadError() bool  { return false }
+func (readerStreamCopierErr) IsWriteError() bool { return true }
+
+func (c readerStreamCopier) WriteStreamTo(w io.Writer) zfs.StreamCopierError {
+	var buf [1 << 21]byte
+	_, err := io.CopyBuffer(w, c.Reader, buf[:])
+	// always assume write error
+	return readerStreamCopierErr{err}
+}
+
 type devNullHandler struct{}
 
-func (devNullHandler) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
+func (devNullHandler) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
 	var res pdu.SendRes
 	if args.devnoopReader {
 		return &res, readerStreamCopier{devnoop.Get()}, nil
@@ -52,12 +67,12 @@ func (devNullHandler) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, i
 	}
 }
 
-func (devNullHandler) Receive(ctx context.Context, r *pdu.ReceiveReq, stream io.ReadCloser) (*pdu.ReceiveRes, error) {
+func (devNullHandler) Receive(ctx context.Context, r *pdu.ReceiveReq, stream zfs.StreamCopier) (*pdu.ReceiveRes, error) {
 	var out io.Writer = os.Stdout
 	if args.devnoopWriter {
 		out = devnoop.Get()
 	}
-	_, err := io.Copy(out, stream)
+	err := stream.WriteStreamTo(out)
 	var res pdu.ReceiveRes
 	return &res, err
 }
@@ -157,7 +172,7 @@ func client() {
 		req := pdu.SendReq{}
 		_, stream, err := client.ReqSend(ctx, &req)
 		orDie(err)
-		_, err = io.Copy(os.Stdout, stream)
+		err = stream.WriteStreamTo(os.Stdout)
 		orDie(err)
 	case "recv":
 		var r io.Reader = os.Stdin

rpc/dataconn/stream/stream.go

@@ -14,6 +14,7 @@ import (
 	"github.com/zrepl/zrepl/rpc/dataconn/base2bufpool"
 	"github.com/zrepl/zrepl/rpc/dataconn/frameconn"
 	"github.com/zrepl/zrepl/rpc/dataconn/heartbeatconn"
+	"github.com/zrepl/zrepl/zfs"
 )
 
 type Logger = logger.Logger
@@ -197,7 +198,7 @@ func (e ReadStreamError) Temporary() bool {
 	return false
 }
 
-var _ net.Error = &ReadStreamError{}
+var _ zfs.StreamCopierError = &ReadStreamError{}
 
 func (e ReadStreamError) IsReadError() bool {
 	return e.Kind != ReadStreamErrorKindWrite

rpc/dataconn/stream/stream_conn.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/zrepl/zrepl/rpc/dataconn/heartbeatconn"
 	"github.com/zrepl/zrepl/rpc/dataconn/timeoutconn"
+	"github.com/zrepl/zrepl/zfs"
 )
 
 type Conn struct {
@@ -39,7 +40,15 @@ type Conn struct {
 
 var readMessageSentinel = fmt.Errorf("read stream complete")
 
-var errWriteStreamToErrorUnknownState = fmt.Errorf("dataconn read stream: connection is in unknown state")
+type writeStreamToErrorUnknownState struct{}
+
+func (e writeStreamToErrorUnknownState) Error() string {
+	return "dataconn read stream: connection is in unknown state"
+}
+
+func (e writeStreamToErrorUnknownState) IsReadError() bool { return true }
+
+func (e writeStreamToErrorUnknownState) IsWriteError() bool { return false }
 
 func Wrap(nc timeoutconn.Wire, sendHeartbeatInterval, peerTimeout time.Duration) *Conn {
 	hc := heartbeatconn.Wrap(nc, sendHeartbeatInterval, peerTimeout)
@@ -114,28 +123,14 @@ func (c *Conn) ReadStreamedMessage(ctx context.Context, maxSize uint32, frameTyp
 	}
 }
 
-type StreamReader struct {
-	*io.PipeReader
-	conn             *Conn
-	closeConnOnClose bool
-}
-
-func (r *StreamReader) Close() error {
-	err := r.PipeReader.Close()
-	if r.closeConnOnClose {
-		r.conn.Close() // TODO error logging
-	}
-	return err
-}
-
 // WriteStreamTo reads a stream from Conn and writes it to w.
-func (c *Conn) ReadStream(frameType uint32, closeConnOnClose bool) (_ *StreamReader, err error) {
+func (c *Conn) ReadStreamInto(w io.Writer, frameType uint32) (err zfs.StreamCopierError) {
 
 	// if we are closed while writing, return that as an error
 	if closeGuard, cse := c.closeState.RWEntry(); cse != nil {
-		return nil, cse
+		return cse
 	} else {
-		defer func(err *error) {
+		defer func(err *zfs.StreamCopierError) {
 			if closed := closeGuard.RWExit(); closed != nil {
 				*err = closed
 			}
@@ -143,23 +138,18 @@ func (c *Conn) ReadStream(frameType uint32, closeConnOnClose bool) (_ *StreamRea
 	}
 
 	c.readMtx.Lock()
+	defer c.readMtx.Unlock()
 	if !c.readClean {
-		return nil, errWriteStreamToErrorUnknownState
+		return writeStreamToErrorUnknownState{}
 	}
+	var rse *ReadStreamError = readStream(c.frameReads, c.hc, w, frameType)
+	c.readClean = isConnCleanAfterRead(rse)
 
-	r, w := io.Pipe()
+	// https://golang.org/doc/faq#nil_error
-	go func() {
+	if rse == nil {
-		defer c.readMtx.Unlock()
+		return nil
-		var err *ReadStreamError = readStream(c.frameReads, c.hc, w, frameType)
+	}
-		if err != nil {
+	return rse
-			_ = w.CloseWithError(err) // doc guarantees that error will always be nil
-		} else {
-			w.Close()
-		}
-		c.readClean = isConnCleanAfterRead(err)
-	}()
-
-	return &StreamReader{PipeReader: r, conn: c, closeConnOnClose: closeConnOnClose}, nil
 }
 
 func (c *Conn) WriteStreamedMessage(ctx context.Context, buf io.Reader, frameType uint32) (err error) {
@@ -188,7 +178,7 @@ func (c *Conn) WriteStreamedMessage(ctx context.Context, buf io.Reader, frameTyp
 	return errConn
 }
 
-func (c *Conn) SendStream(ctx context.Context, stream io.ReadCloser, frameType uint32) (err error) {
+func (c *Conn) SendStream(ctx context.Context, src zfs.StreamCopier, frameType uint32) (err error) {
 
 	// if we are closed while reading, return that as an error
 	if closeGuard, cse := c.closeState.RWEntry(); cse != nil {
@@ -207,17 +197,49 @@ func (c *Conn) SendStream(ctx context.Context, stream io.ReadCloser, frameType u
 		return fmt.Errorf("dataconn send stream: connection is in unknown state")
 	}
 
-	errStream, errConn := writeStream(ctx, c.hc, stream, frameType)
+	// avoid io.Pipe if zfs.StreamCopier is an io.Reader
-
+	var r io.Reader
-	c.writeClean = isConnCleanAfterWrite(errConn) // TODO correct?
+	var w *io.PipeWriter
-
+	streamCopierErrChan := make(chan zfs.StreamCopierError, 1)
-	if errStream != nil {
+	if reader, ok := src.(io.Reader); ok {
-		return errStream
+		r = reader
-	} else if errConn != nil {
+		streamCopierErrChan <- nil
-		return errConn
+		close(streamCopierErrChan)
+	} else {
+		r, w = io.Pipe()
+		go func() {
+			streamCopierErrChan <- src.WriteStreamTo(w)
+			w.Close()
+		}()
+	}
+
+	type writeStreamRes struct {
+		errStream, errConn error
+	}
+	writeStreamErrChan := make(chan writeStreamRes, 1)
+	go func() {
+		var res writeStreamRes
+		res.errStream, res.errConn = writeStream(ctx, c.hc, r, frameType)
+		if w != nil {
+			_ = w.CloseWithError(res.errStream) // always returns nil
+		}
+		writeStreamErrChan <- res
+	}()
+
+	writeRes := <-writeStreamErrChan
+	streamCopierErr := <-streamCopierErrChan
+	c.writeClean = isConnCleanAfterWrite(writeRes.errConn) // TODO correct?
+	if streamCopierErr != nil && streamCopierErr.IsReadError() {
+		return streamCopierErr // something on our side is bad
+	} else {
+		if writeRes.errStream != nil {
+			return writeRes.errStream
+		} else if writeRes.errConn != nil {
+			return writeRes.errConn
+		}
+		// TODO combined error?
+		return streamCopierErr
 	}
-	// TODO combined error?
-	return nil
 }
 
 type closeState struct {
@@ -226,13 +248,17 @@ type closeState struct {
 
 type closeStateErrConnectionClosed struct{}
 
+var _ zfs.StreamCopierError = (*closeStateErrConnectionClosed)(nil)
+var _ error = (*closeStateErrConnectionClosed)(nil)
 var _ net.Error = (*closeStateErrConnectionClosed)(nil)
 
 func (e *closeStateErrConnectionClosed) Error() string {
 	return "connection closed"
 }
-func (e *closeStateErrConnectionClosed) Timeout() bool   { return false }
+func (e *closeStateErrConnectionClosed) IsReadError() bool  { return true }
-func (e *closeStateErrConnectionClosed) Temporary() bool { return false }
+func (e *closeStateErrConnectionClosed) IsWriteError() bool { return true }
+func (e *closeStateErrConnectionClosed) Timeout() bool      { return false }
+func (e *closeStateErrConnectionClosed) Temporary() bool    { return false }
 
 func (s *closeState) CloseEntry() error {
 	firstCloser := atomic.AddUint32(&s.closeCount, 1) == 1
@@ -247,7 +273,7 @@ type closeStateEntry struct {
 	entryCount uint32
 }
 
-func (s *closeState) RWEntry() (e *closeStateEntry, err net.Error) {
+func (s *closeState) RWEntry() (e *closeStateEntry, err zfs.StreamCopierError) {
 	entry := &closeStateEntry{s, atomic.LoadUint32(&s.closeCount)}
 	if entry.entryCount > 0 {
 		return nil, &closeStateErrConnectionClosed{}
@@ -255,7 +281,7 @@ func (s *closeState) RWEntry() (e *closeStateEntry, err net.Error) {
 	return entry, nil
 }
 
-func (e *closeStateEntry) RWExit() net.Error {
+func (e *closeStateEntry) RWExit() zfs.StreamCopierError {
 	if atomic.LoadUint32(&e.entryCount) == e.entryCount {
 		// no calls to Close() while running rw operation
 		return nil

rpc/rpc_client.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"io"
 	"net"
 	"sync"
 	"sync/atomic"
@@ -21,6 +20,7 @@ import (
 	"github.com/zrepl/zrepl/rpc/versionhandshake"
 	"github.com/zrepl/zrepl/transport"
 	"github.com/zrepl/zrepl/util/envconst"
+	"github.com/zrepl/zrepl/zfs"
 )
 
 // Client implements the active side of a replication setup.
@@ -82,22 +82,22 @@ func (c *Client) Close() {
 
 // callers must ensure that the returned io.ReadCloser is closed
 // TODO expose dataClient interface to the outside world
-func (c *Client) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
+func (c *Client) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
 	// TODO the returned sendStream may return a read error created by the remote side
-	res, stream, err := c.dataClient.ReqSend(ctx, r)
+	res, streamCopier, err := c.dataClient.ReqSend(ctx, r)
 	if err != nil {
 		return nil, nil, err
 	}
-	if stream == nil {
+	if streamCopier == nil {
 		return res, nil, nil
 	}
 
-	return res, stream, nil
+	return res, streamCopier, nil
 
 }
 
-func (c *Client) Receive(ctx context.Context, req *pdu.ReceiveReq, stream io.ReadCloser) (*pdu.ReceiveRes, error) {
+func (c *Client) Receive(ctx context.Context, req *pdu.ReceiveReq, streamCopier zfs.StreamCopier) (*pdu.ReceiveRes, error) {
-	return c.dataClient.ReqRecv(ctx, req, stream)
+	return c.dataClient.ReqRecv(ctx, req, streamCopier)
 }
 
 func (c *Client) ListFilesystems(ctx context.Context, in *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error) {

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

util/bytecounter/bytecounter_readcloser.go

@@ -1,39 +0,0 @@
-package bytecounter
-
-import (
-	"io"
-	"sync/atomic"
-)
-
-// ReadCloser wraps an io.ReadCloser, reimplementing
-// its interface and counting the bytes written to during copying.
-type ReadCloser interface {
-	io.ReadCloser
-	Count() int64
-}
-
-// NewReadCloser wraps rc.
-func NewReadCloser(rc io.ReadCloser) ReadCloser {
-	return &readCloser{rc, 0}
-}
-
-type readCloser struct {
-	rc    io.ReadCloser
-	count int64
-}
-
-func (r *readCloser) Count() int64 {
-	return atomic.LoadInt64(&r.count)
-}
-
-var _ io.ReadCloser = &readCloser{}
-
-func (r *readCloser) Close() error {
-	return r.rc.Close()
-}
-
-func (r *readCloser) Read(p []byte) (int, error) {
-	n, err := r.rc.Read(p)
-	atomic.AddInt64(&r.count, int64(n))
-	return n, err
-}

util/bytecounter/bytecounter_reader.go

@@ -0,0 +1,49 @@
+package bytecounter
+
+import (
+	"io"
+	"sync/atomic"
+	"time"
+)
+
+type ByteCounterReader struct {
+	reader io.ReadCloser
+
+	// called & accessed synchronously during Read, no external access
+	cb       func(full int64)
+	cbEvery  time.Duration
+	lastCbAt time.Time
+
+	// set atomically because it may be read by multiple threads
+	bytes int64
+}
+
+func NewByteCounterReader(reader io.ReadCloser) *ByteCounterReader {
+	return &ByteCounterReader{
+		reader: reader,
+	}
+}
+
+func (b *ByteCounterReader) SetCallback(every time.Duration, cb func(full int64)) {
+	b.cbEvery = every
+	b.cb = cb
+}
+
+func (b *ByteCounterReader) Close() error {
+	return b.reader.Close()
+}
+
+func (b *ByteCounterReader) Read(p []byte) (n int, err error) {
+	n, err = b.reader.Read(p)
+	full := atomic.AddInt64(&b.bytes, int64(n))
+	now := time.Now()
+	if b.cb != nil && now.Sub(b.lastCbAt) > b.cbEvery {
+		b.cb(full)
+		b.lastCbAt = now
+	}
+	return n, err
+}
+
+func (b *ByteCounterReader) Bytes() int64 {
+	return atomic.LoadInt64(&b.bytes)
+}

util/bytecounter/bytecounter_streamcopier.go

@@ -0,0 +1,71 @@
+package bytecounter
+
+import (
+	"io"
+	"sync/atomic"
+
+	"github.com/zrepl/zrepl/zfs"
+)
+
+// StreamCopier wraps a zfs.StreamCopier, reimplementing
+// its interface and counting the bytes written to during copying.
+type StreamCopier interface {
+	zfs.StreamCopier
+	Count() int64
+}
+
+// NewStreamCopier wraps sc into a StreamCopier.
+// If sc is io.Reader, it is guaranteed that the returned StreamCopier
+// implements that interface, too.
+func NewStreamCopier(sc zfs.StreamCopier) StreamCopier {
+	bsc := &streamCopier{sc, 0}
+	if scr, ok := sc.(io.Reader); ok {
+		return streamCopierAndReader{bsc, scr}
+	} else {
+		return bsc
+	}
+}
+
+type streamCopier struct {
+	sc    zfs.StreamCopier
+	count int64
+}
+
+// proxy writer used by streamCopier
+type streamCopierWriter struct {
+	parent *streamCopier
+	w      io.Writer
+}
+
+func (w streamCopierWriter) Write(p []byte) (n int, err error) {
+	n, err = w.w.Write(p)
+	atomic.AddInt64(&w.parent.count, int64(n))
+	return
+}
+
+func (s *streamCopier) Count() int64 {
+	return atomic.LoadInt64(&s.count)
+}
+
+var _ zfs.StreamCopier = &streamCopier{}
+
+func (s streamCopier) Close() error {
+	return s.sc.Close()
+}
+
+func (s *streamCopier) WriteStreamTo(w io.Writer) zfs.StreamCopierError {
+	ww := streamCopierWriter{s, w}
+	return s.sc.WriteStreamTo(ww)
+}
+
+// a streamCopier whose underlying sc is an io.Reader
+type streamCopierAndReader struct {
+	*streamCopier
+	asReader io.Reader
+}
+
+func (scr streamCopierAndReader) Read(p []byte) (int, error) {
+	n, err := scr.asReader.Read(p)
+	atomic.AddInt64(&scr.streamCopier.count, int64(n))
+	return n, err
+}

util/bytecounter/bytecounter_streamcopier_test.go

@@ -0,0 +1,39 @@
+package bytecounter
+
+import (
+	"io"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/zrepl/zrepl/zfs"
+)
+
+type mockStreamCopierAndReader struct {
+	zfs.StreamCopier // to satisfy interface
+	reads            int
+}
+
+func (r *mockStreamCopierAndReader) Read(p []byte) (int, error) {
+	r.reads++
+	return len(p), nil
+}
+
+var _ io.Reader = &mockStreamCopierAndReader{}
+
+func TestNewStreamCopierReexportsReader(t *testing.T) {
+	mock := &mockStreamCopierAndReader{}
+	x := NewStreamCopier(mock)
+
+	r, ok := x.(io.Reader)
+	if !ok {
+		t.Fatalf("%T does not implement io.Reader, hence reader cannout have been wrapped", x)
+	}
+
+	var buf [23]byte
+	n, err := r.Read(buf[:])
+	assert.True(t, mock.reads == 1)
+	assert.True(t, n == len(buf))
+	assert.NoError(t, err)
+	assert.True(t, x.Count() == 23)
+}

util/chainedio/chainedio_readcloser.go

@@ -1,43 +0,0 @@
-package chainedio
-
-import "io"
-
-type ChainedReadCloser struct {
-	readers   []io.Reader
-	curReader int
-}
-
-func NewChainedReader(reader ...io.Reader) *ChainedReadCloser {
-	return &ChainedReadCloser{
-		readers:   reader,
-		curReader: 0,
-	}
-}
-
-func (c *ChainedReadCloser) Read(buf []byte) (n int, err error) {
-
-	n = 0
-
-	for c.curReader < len(c.readers) {
-		n, err = c.readers[c.curReader].Read(buf)
-		if err == io.EOF {
-			c.curReader++
-			continue
-		}
-		break
-	}
-	if c.curReader == len(c.readers) {
-		err = io.EOF // actually, there was no gap
-	}
-
-	return
-}
-
-func (c *ChainedReadCloser) Close() error {
-	for _, r := range c.readers {
-		if c, ok := r.(io.Closer); ok {
-			c.Close() // TODO debug log error?
-		}
-	}
-	return nil
-}

util/chainedio/chainedio_reader.go

@@ -0,0 +1,34 @@
+package chainedio
+
+import "io"
+
+type ChainedReader struct {
+	Readers   []io.Reader
+	curReader int
+}
+
+func NewChainedReader(reader ...io.Reader) *ChainedReader {
+	return &ChainedReader{
+		Readers:   reader,
+		curReader: 0,
+	}
+}
+
+func (c *ChainedReader) Read(buf []byte) (n int, err error) {
+
+	n = 0
+
+	for c.curReader < len(c.Readers) {
+		n, err = c.Readers[c.curReader].Read(buf)
+		if err == io.EOF {
+			c.curReader++
+			continue
+		}
+		break
+	}
+	if c.curReader == len(c.Readers) {
+		err = io.EOF // actually, there was no gap
+	}
+
+	return
+}

util/chainlock/chainlock.go

@@ -40,8 +40,3 @@ func (l *L) DropWhile(f func()) {
 	defer l.Unlock().Lock()
 	f()
 }
-
-func (l *L) HoldWhile(f func()) {
-	defer l.Lock().Unlock()
-	f()
-}

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

@@ -354,6 +354,63 @@ func (a ZFSSendArgsUnvalidated) buildCommonSendArgs() ([]string, error) {
 	return args, nil
 }
 
+type ReadCloserCopier struct {
+	recorder readErrRecorder
+}
+
+type readErrRecorder struct {
+	io.ReadCloser
+	readErr error
+}
+
+type sendStreamCopierError struct {
+	isReadErr bool // if false, it's a write error
+	err       error
+}
+
+func (e sendStreamCopierError) Error() string {
+	if e.isReadErr {
+		return fmt.Sprintf("stream: read error: %s", e.err)
+	} else {
+		return fmt.Sprintf("stream: writer error: %s", e.err)
+	}
+}
+
+func (e sendStreamCopierError) IsReadError() bool  { return e.isReadErr }
+func (e sendStreamCopierError) IsWriteError() bool { return !e.isReadErr }
+
+func (r *readErrRecorder) Read(p []byte) (n int, err error) {
+	n, err = r.ReadCloser.Read(p)
+	r.readErr = err
+	return n, err
+}
+
+func NewReadCloserCopier(stream io.ReadCloser) *ReadCloserCopier {
+	return &ReadCloserCopier{recorder: readErrRecorder{stream, nil}}
+}
+
+func (c *ReadCloserCopier) WriteStreamTo(w io.Writer) StreamCopierError {
+	debug("sendStreamCopier.WriteStreamTo: begin")
+	_, err := io.Copy(w, &c.recorder)
+	debug("sendStreamCopier.WriteStreamTo: copy done")
+	if err != nil {
+		if c.recorder.readErr != nil {
+			return sendStreamCopierError{isReadErr: true, err: c.recorder.readErr}
+		} else {
+			return sendStreamCopierError{isReadErr: false, err: err}
+		}
+	}
+	return nil
+}
+
+func (c *ReadCloserCopier) Read(p []byte) (n int, err error) {
+	return c.recorder.Read(p)
+}
+
+func (c *ReadCloserCopier) Close() error {
+	return c.recorder.ReadCloser.Close()
+}
+
 func pipeWithCapacityHint(capacity int) (r, w *os.File, err error) {
 	if capacity <= 0 {
 		panic(fmt.Sprintf("capacity must be positive %v", capacity))
@@ -366,7 +423,7 @@ func pipeWithCapacityHint(capacity int) (r, w *os.File, err error) {
 	return stdoutReader, stdoutWriter, nil
 }
 
-type SendStream struct {
+type sendStream struct {
 	cmd  *zfscmd.Cmd
 	kill context.CancelFunc
 
@@ -376,7 +433,7 @@ type SendStream struct {
 	opErr        error
 }
 
-func (s *SendStream) Read(p []byte) (n int, err error) {
+func (s *sendStream) Read(p []byte) (n int, err error) {
 	s.closeMtx.Lock()
 	opErr := s.opErr
 	s.closeMtx.Unlock()
@@ -397,12 +454,12 @@ func (s *SendStream) Read(p []byte) (n int, err error) {
 	return n, err
 }
 
-func (s *SendStream) Close() error {
+func (s *sendStream) Close() error {
 	debug("sendStream: close called")
 	return s.killAndWait(nil)
 }
 
-func (s *SendStream) killAndWait(precedingReadErr error) error {
+func (s *sendStream) killAndWait(precedingReadErr error) error {
 
 	debug("sendStream: killAndWait enter")
 	defer debug("sendStream: killAndWait leave")
@@ -773,7 +830,7 @@ var ErrEncryptedSendNotSupported = fmt.Errorf("raw sends which are required for
 // (if from is "" a full ZFS send is done)
 //
 // Returns ErrEncryptedSendNotSupported if encrypted send is requested but not supported by CLI
-func ZFSSend(ctx context.Context, sendArgs ZFSSendArgsValidated) (*SendStream, error) {
+func ZFSSend(ctx context.Context, sendArgs ZFSSendArgsValidated) (*ReadCloserCopier, error) {
 
 	args := make([]string, 0)
 	args = append(args, "send")
@@ -822,14 +879,14 @@ func ZFSSend(ctx context.Context, sendArgs ZFSSendArgsValidated) (*SendStream, e
 	// close our writing-end of the pipe so that we don't wait for ourselves when reading from the reading  end
 	stdoutWriter.Close()
 
-	stream := &SendStream{
+	stream := &sendStream{
 		cmd:          cmd,
 		kill:         cancel,
 		stdoutReader: stdoutReader,
 		stderrBuf:    stderrBuf,
 	}
 
-	return stream, nil
+	return NewReadCloserCopier(stream), nil
 }
 
 type DrySendType string
@@ -968,6 +1025,24 @@ func ZFSSendDry(ctx context.Context, sendArgs ZFSSendArgsValidated) (_ *DrySendI
 	return &si, nil
 }
 
+type StreamCopierError interface {
+	error
+	IsReadError() bool
+	IsWriteError() bool
+}
+
+type StreamCopier interface {
+	// WriteStreamTo writes the stream represented by this StreamCopier
+	// to the given io.Writer.
+	WriteStreamTo(w io.Writer) StreamCopierError
+	// Close must be called as soon as it is clear that no more data will
+	// be read from the StreamCopier.
+	// If StreamCopier gets its data from a connection, it might hold
+	// a lock on the connection until Close is called. Only closing ensures
+	// that the connection can be used afterwards.
+	Close() error
+}
+
 type RecvOptions struct {
 	// Rollback to the oldest snapshot, destroy it, then perform `recv -F`.
 	// Note that this doesn't change property values, i.e. an existing local property value will be kept.
@@ -992,9 +1067,7 @@ func (e *ErrRecvResumeNotSupported) Error() string {
 	return buf.String()
 }
 
-const RecvStderrBufSiz = 1 << 15
+func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, streamCopier StreamCopier, opts RecvOptions) (err error) {
-
-func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, stream io.ReadCloser, opts RecvOptions) (err error) {
 
 	if err := v.ValidateInMemory(fs); err != nil {
 		return errors.Wrap(err, "invalid version")
@@ -1011,7 +1084,7 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, stream io.Rea
 	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 {
@@ -1061,7 +1134,7 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, stream io.Rea
 	//   cannot receive new filesystem stream: invalid backup stream
 	stdout := bytes.NewBuffer(make([]byte, 0, 1024))
 
-	stderr := bytes.NewBuffer(make([]byte, 0, RecvStderrBufSiz))
+	stderr := bytes.NewBuffer(make([]byte, 0, 1024))
 
 	stdin, stdinWriter, err := pipeWithCapacityHint(ZFSRecvPipeCapacityHint)
 	if err != nil {
@@ -1089,10 +1162,9 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, stream io.Rea
 
 	debug("started")
 
-	copierErrChan := make(chan error)
+	copierErrChan := make(chan StreamCopierError)
 	go func() {
-		_, err := io.Copy(stdinWriter, stream)
+		copierErrChan <- streamCopier.WriteStreamTo(stdinWriter)
-		copierErrChan <- err
 		stdinWriter.Close()
 	}()
 	waitErrChan := make(chan error)
@@ -1101,10 +1173,6 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, stream io.Rea
 		if err = cmd.Wait(); err != nil {
 			if rtErr := tryRecvErrorWithResumeToken(ctx, stderr.String()); rtErr != nil {
 				waitErrChan <- rtErr
-			} else if owErr := tryRecvDestroyOrOverwriteEncryptedErr(stderr.Bytes()); owErr != nil {
-				waitErrChan <- owErr
-			} else if readErr := tryRecvCannotReadFromStreamErr(stderr.Bytes()); readErr != nil {
-				waitErrChan <- readErr
 			} else {
 				waitErrChan <- &ZFSError{
 					Stderr:  stderr.Bytes(),
@@ -1115,23 +1183,22 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, stream io.Rea
 		}
 	}()
 
+	// streamCopier always fails before or simultaneously with Wait
+	// thus receive from it first
 	copierErr := <-copierErrChan
 	debug("copierErr: %T %s", copierErr, copierErr)
 	if copierErr != nil {
-		debug("killing zfs recv command after copierErr")
 		cancelCmd()
 	}
 
 	waitErr := <-waitErrChan
 	debug("waitErr: %T %s", waitErr, waitErr)
-
 	if copierErr == nil && waitErr == nil {
 		return nil
-	} else if _, isReadErr := waitErr.(*RecvCannotReadFromStreamErr); isReadErr {
+	} else if waitErr != nil && (copierErr == nil || copierErr.IsWriteError()) {
-		return copierErr // likely network error reading from stream
+		return waitErr // has more interesting info in that case
-	} else {
-		return waitErr // almost always more interesting info. NOTE: do not wrap!
 	}
+	return copierErr // if it's not a write error, the copier error is more interesting
 }
 
 type RecvFailedWithResumeTokenErr struct {
@@ -1161,43 +1228,6 @@ func (e *RecvFailedWithResumeTokenErr) Error() string {
 	return fmt.Sprintf("receive failed, resume token available: %s\n%#v", e.ResumeTokenRaw, e.ResumeTokenParsed)
 }
 
-type RecvDestroyOrOverwriteEncryptedErr struct {
-	Msg string
-}
-
-func (e *RecvDestroyOrOverwriteEncryptedErr) Error() string {
-	return e.Msg
-}
-
-var recvDestroyOrOverwriteEncryptedErrRe = regexp.MustCompile(`^(cannot receive new filesystem stream: zfs receive -F cannot be used to destroy an encrypted filesystem or overwrite an unencrypted one with an encrypted one)`)
-
-func tryRecvDestroyOrOverwriteEncryptedErr(stderr []byte) *RecvDestroyOrOverwriteEncryptedErr {
-	debug("tryRecvDestroyOrOverwriteEncryptedErr: %v", stderr)
-	m := recvDestroyOrOverwriteEncryptedErrRe.FindSubmatch(stderr)
-	if m == nil {
-		return nil
-	}
-	return &RecvDestroyOrOverwriteEncryptedErr{Msg: string(m[1])}
-}
-
-type RecvCannotReadFromStreamErr struct {
-	Msg string
-}
-
-func (e *RecvCannotReadFromStreamErr) Error() string {
-	return e.Msg
-}
-
-var reRecvCannotReadFromStreamErr = regexp.MustCompile(`^(cannot receive: failed to read from stream)$`)
-
-func tryRecvCannotReadFromStreamErr(stderr []byte) *RecvCannotReadFromStreamErr {
-	m := reRecvCannotReadFromStreamErr.FindSubmatch(stderr)
-	if m == nil {
-		return nil
-	}
-	return &RecvCannotReadFromStreamErr{Msg: string(m[1])}
-}
-
 type ClearResumeTokenError struct {
 	ZFSOutput []byte
 	CmdError  error

zfs/zfs_test.go

@@ -2,11 +2,9 @@ package zfs
 
 import (
 	"context"
-	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
 )
 
 func TestZFSListHandlesProducesZFSErrorOnNonZeroExit(t *testing.T) {
@@ -261,12 +259,3 @@ size	10518512
 		})
 	}
 }
-
-func TestTryRecvDestroyOrOverwriteEncryptedErr(t *testing.T) {
-	msg := "cannot receive new filesystem stream: zfs receive -F cannot be used to destroy an encrypted filesystem or overwrite an unencrypted one with an encrypted one\n"
-	assert.GreaterOrEqual(t, RecvStderrBufSiz, len(msg))
-
-	err := tryRecvDestroyOrOverwriteEncryptedErr([]byte(msg))
-	require.NotNil(t, err)
-	assert.EqualError(t, err, strings.TrimSpace(msg))
-}

zfs/zfscmd/zfscmd.go

@@ -18,10 +18,10 @@ import (
 )
 
 type Cmd struct {
-	cmd                                      *exec.Cmd
+	cmd                       *exec.Cmd
-	ctx                                      context.Context
+	ctx                       context.Context
-	mtx                                      sync.RWMutex
+	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()
+	waitPreLogging(c, 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
 }
 
 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
+	waitPostReport(c, now)
-	var u usage
+	waitPostLogging(c, err, now)
-	{
+	waitPostPrometheus(c, err, now)
-		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)
 }
 
 // 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("total_time_s", c.Runtime().Seconds()).
-		WithField("systemtime_s", u.system_secs).
+		WithField("systemtime_s", c.cmd.ProcessState.SystemTime().Seconds()).
-		WithField("usertime_s", u.user_secs)
+		WithField("usertime_s", c.cmd.ProcessState.UserTime().Seconds())
 
 	if err == nil {
 		log.Info("command exited without error")

zfs/zfscmd/zfscmd_platform_test.go

@@ -1,18 +1,13 @@
 package zfscmd
 
 import (
-	"bufio"
 	"bytes"
-	"context"
 	"io"
-	"os"
 	"os/exec"
-	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"github.com/zrepl/zrepl/util/circlog"
 )
 
 const testBin = "./zfscmd_platform_test.bash"
@@ -63,62 +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")
-}
-
-func TestSigpipe(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	cmd := CommandContext(ctx, "bash", "-c", "sleep 5; echo invalid input; exit 23")
-	r, w, err := os.Pipe()
-	require.NoError(t, err)
-	output := circlog.MustNewCircularLog(1 << 20)
-	cmd.SetStdio(Stdio{
-		Stdin:  r,
-		Stdout: output,
-		Stderr: output,
-	})
-	err = cmd.Start()
-	require.NoError(t, err)
-	err = r.Close()
-	require.NoError(t, err)
-
-	// the script doesn't read stdin, but this input is almost certainly smaller than the pipe buffer
-	const LargerThanPipeBuffer = 1 << 21
-	_, err = io.Copy(w, bytes.NewBuffer(bytes.Repeat([]byte("i"), LargerThanPipeBuffer)))
-	// => io.Copy is going to block because the pipe buffer is full and the
-	//    script is not reading from it
-	// => the script is going to exit after 5s
-	// => we should expect a broken pipe error from the copier's perspective
-	t.Logf("copy err = %T: %s", err, err)
-	require.NotNil(t, err)
-	require.True(t, strings.Contains(err.Error(), "broken pipe"))
-
-	err = cmd.Wait()
-	require.EqualError(t, err, "exit status 23")
-
-}

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]