build: detect if generate made things dirty and break release build in that case

An fsrep.Replication is either Ready, Retry or in a terminal state.
The queue prefers Ready over Retry:

Ready is sorted by nextStepDate to progress evenly..
Retry is sorted by error count, to de-prioritize filesystems that fail
often. This way we don't get stuck with individual filesystems
and lose other working filesystems to the watchdog.

fsrep.Replication no longer blocks in Retry state, we have
replication.WorkingWait for that.

Makefile

@@ -26,6 +26,9 @@ SUBPKGS += replication/internal/queue
 SUBPKGS += replication/internal/diff
 SUBPKGS += tlsconf
 SUBPKGS += util
+SUBPKGS += util/socketpair
+SUBPKGS += util/watchdog
+SUBPKGS += util/envconst
 SUBPKGS += version
 SUBPKGS += zfs
 
@@ -33,13 +36,16 @@ _TESTPKGS := $(ROOT) $(foreach p,$(SUBPKGS),$(ROOT)/$(p))
 
 ARTIFACTDIR := artifacts
 
-ifndef ZREPL_VERSION
-    ZREPL_VERSION := $(shell git describe --dirty 2>/dev/null || echo "ZREPL_BUILD_INVALID_VERSION" )
-    ifeq ($(ZREPL_VERSION),ZREPL_BUILD_INVALID_VERSION) # can't use .SHELLSTATUS because Debian Stretch is still on gmake 4.1
+ifdef ZREPL_VERSION
+    _ZREPL_VERSION := $(ZREPL_VERSION)
+endif
+ifndef _ZREPL_VERSION
+    _ZREPL_VERSION := $(shell git describe --dirty 2>/dev/null || echo "ZREPL_BUILD_INVALID_VERSION" )
+    ifeq ($(_ZREPL_VERSION),ZREPL_BUILD_INVALID_VERSION) # can't use .SHELLSTATUS because Debian Stretch is still on gmake 4.1
         $(error cannot infer variable ZREPL_VERSION using git and variable is not overriden by make invocation)
     endif
 endif
-GO_LDFLAGS := "-X github.com/zrepl/zrepl/version.zreplVersion=$(ZREPL_VERSION)"
+GO_LDFLAGS := "-X github.com/zrepl/zrepl/version.zreplVersion=$(_ZREPL_VERSION)"
 
 GO_BUILD := go build -ldflags $(GO_LDFLAGS)
 
@@ -115,7 +121,7 @@ $(RELEASE_BINS): $(ARTIFACTDIR)/zrepl-%-amd64: generate $(ARTIFACTDIR) vet test
 
 $(RELEASE_NOARCH): docs $(ARTIFACTDIR)/bash_completion $(ARTIFACTDIR)/go_version.txt
 	tar --mtime='1970-01-01' --sort=name \
-		--transform 's/$(ARTIFACTDIR)/zrepl-$(ZREPL_VERSION)-noarch/' \
+		--transform 's/$(ARTIFACTDIR)/zrepl-$(_ZREPL_VERSION)-noarch/' \
 		-acf $@ \
 		$(ARTIFACTDIR)/docs/html \
 		$(ARTIFACTDIR)/bash_completion \
@@ -126,10 +132,13 @@ release: $(RELEASE_BINS) $(RELEASE_NOARCH)
 	mkdir -p "$(ARTIFACTDIR)/release"
 	cp $^ "$(ARTIFACTDIR)/release"
 	cd "$(ARTIFACTDIR)/release" && sha512sum $$(ls | sort) > sha512sum.txt
-	@if echo "$(ZREPL_VERSION)" | grep dirty > /dev/null; then\
-		echo '[WARN] Do not publish the artifacts, make variable ZREPL_VERSION=$(ZREPL_VERSION) indicates they are dirty!'; \
-		exit 1; \
-	fi
+	@# note that we use ZREPL_VERSION and not _ZREPL_VERSION because we want to detect the override
+	@if git describe --dirty 2>/dev/null | grep dirty >/dev/null; then \
+		if [ "$(ZREPL_VERSION)" == "" ]; then \
+			echo "[WARN] git checkout is dirty and make variable ZREPL_VERSION was not used to override"; \
+			exit 1; \
+		fi; \
+	fi;
 
 clean: docs-clean
 	rm -rf "$(ARTIFACTDIR)"

client/configcheck.go

@@ -2,15 +2,22 @@ package client
 
 import (
 	"encoding/json"
+	"fmt"
 	"github.com/kr/pretty"
+	"github.com/pkg/errors"
 	"github.com/spf13/pflag"
 	"github.com/zrepl/yaml-config"
 	"github.com/zrepl/zrepl/cli"
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon/job"
+	"github.com/zrepl/zrepl/daemon/logging"
+	"github.com/zrepl/zrepl/logger"
 	"os"
 )
 
 var configcheckArgs struct {
 	format string
+	what string
 }
 
 var ConfigcheckCmd = &cli.Subcommand{
@@ -18,19 +25,88 @@ var ConfigcheckCmd = &cli.Subcommand{
 	Short: "check if config can be parsed without errors",
 	SetupFlags: func(f *pflag.FlagSet) {
 		f.StringVar(&configcheckArgs.format, "format", "", "dump parsed config object [pretty|yaml|json]")
+		f.StringVar(&configcheckArgs.what, "what", "all", "what to print [all|config|jobs|logging]")
 	},
 	Run: func(subcommand *cli.Subcommand, args []string) error {
-		switch configcheckArgs.format {
-		case "pretty":
-			_, err := pretty.Println(subcommand.Config())
-			return err
-		case "json":
-			return json.NewEncoder(os.Stdout).Encode(subcommand.Config())
-		case "yaml":
-			return yaml.NewEncoder(os.Stdout).Encode(subcommand.Config())
-		default: // no output
+		formatMap := map[string]func(interface{}) {
+			"": func(i interface{}) {},
+			"pretty": func(i interface{}) { pretty.Println(i) },
+			"json": func(i interface{}) {
+				json.NewEncoder(os.Stdout).Encode(subcommand.Config())
+			},
+			"yaml": func(i interface{}) {
+				yaml.NewEncoder(os.Stdout).Encode(subcommand.Config())
+			},
+		}
+
+		formatter, ok := formatMap[configcheckArgs.format]
+		if !ok {
+			return fmt.Errorf("unsupported --format %q", configcheckArgs.format)
+		}
+
+		var hadErr bool
+		// further: try to build jobs
+		confJobs, err := job.JobsFromConfig(subcommand.Config())
+		if err != nil {
+			err := errors.Wrap(err, "cannot build jobs from config")
+			if configcheckArgs.what == "jobs" {
+				return err
+			} else {
+				fmt.Fprintf(os.Stderr, "%s\n", err)
+				confJobs = nil
+				hadErr = true
+			}
+		}
+
+		// further: try to build logging outlets
+		outlets, err := logging.OutletsFromConfig(*subcommand.Config().Global.Logging)
+		if err != nil {
+			err := errors.Wrap(err, "cannot build logging from config")
+			if configcheckArgs.what == "logging" {
+				return err
+			} else {
+				fmt.Fprintf(os.Stderr, "%s\n", err)
+				outlets = nil
+				hadErr = true
+			}
+		}
+
+
+		whatMap := map[string]func() {
+			"all": func() {
+				o := struct {
+					config *config.Config
+					jobs []job.Job
+					logging *logger.Outlets
+				}{
+					subcommand.Config(),
+					confJobs,
+					outlets,
+				}
+				formatter(o)
+			},
+			"config": func() {
+				formatter(subcommand.Config())
+			},
+			"jobs": func() {
+				formatter(confJobs)
+			},
+			"logging": func() {
+				formatter(outlets)
+			},
+		}
+
+		wf, ok := whatMap[configcheckArgs.what]
+		if !ok {
+			return fmt.Errorf("unsupported --format %q", configcheckArgs.what)
+		}
+		wf()
+
+		if hadErr {
+			return fmt.Errorf("config parsing failed")
+		} else {
+			return nil
 		}
-		return nil
 	},
 }
 

client/status.go

@@ -274,8 +274,7 @@ func (t *tui) renderReplicationReport(rep *replication.Report) {
 		t.printf("Problem: %s", rep.Problem)
 		t.newline()
 	}
-	if rep.SleepUntil.After(time.Now()) &&
-		state & ^(replication.ContextDone|replication.Completed) != 0 {
+	if rep.SleepUntil.After(time.Now()) && !state.IsTerminal() {
 		t.printf("Sleeping until %s (%s left)\n", rep.SleepUntil, rep.SleepUntil.Sub(time.Now()))
 	}
 

config/config.go

@@ -115,7 +115,7 @@ time: true
 level: "warn"
 format: "human"
 `
-	s := StdoutLoggingOutlet{}
+	s := &StdoutLoggingOutlet{}
 	err := yaml.UnmarshalStrict([]byte(def), &s)
 	if err != nil {
 		panic(err)

config/config_global_test.go

@@ -57,7 +57,7 @@ global:
 func TestDefaultLoggingOutlet(t *testing.T) {
 	conf := testValidGlobalSection(t, "")
 	assert.Equal(t, 1, len(*conf.Global.Logging))
-	o := (*conf.Global.Logging)[0].Ret.(StdoutLoggingOutlet)
+	o := (*conf.Global.Logging)[0].Ret.(*StdoutLoggingOutlet)
 	assert.Equal(t, "warn", o.Level)
 	assert.Equal(t, "human", o.Format)
 }
@@ -77,6 +77,6 @@ func TestLoggingOutletEnumList_SetDefaults(t *testing.T) {
 	var i yaml.Defaulter = e
 	require.NotPanics(t, func() {
 		i.SetDefault()
-		assert.Equal(t, "warn", (*e)[0].Ret.(StdoutLoggingOutlet).Level)
+		assert.Equal(t, "warn", (*e)[0].Ret.(*StdoutLoggingOutlet).Level)
 	})
 }

daemon/job/active.go

@@ -13,6 +13,8 @@ import (
 	"github.com/zrepl/zrepl/daemon/pruner"
 	"github.com/zrepl/zrepl/endpoint"
 	"github.com/zrepl/zrepl/replication"
+	"github.com/zrepl/zrepl/util/envconst"
+	"github.com/zrepl/zrepl/util/watchdog"
 	"github.com/zrepl/zrepl/zfs"
 	"sync"
 	"github.com/zrepl/zrepl/daemon/logging"
@@ -38,7 +40,9 @@ type ActiveSide struct {
 
 type activeSideTasks struct {
 	replication *replication.Replication
+	replicationCancel context.CancelFunc
 	prunerSender, prunerReceiver *pruner.Pruner
+	prunerSenderCancel, prunerReceiverCancel context.CancelFunc
 }
 
 func (a *ActiveSide) updateTasks(u func(*activeSideTasks)) activeSideTasks {
@@ -238,11 +242,11 @@ outer:
 		}
 		invocationCount++
 		invLog := log.WithField("invocation", invocationCount)
-		j.do(WithLogger(ctx, invLog), periodicDone)
+		j.do(WithLogger(ctx, invLog))
 	}
 }
 
-func (j *ActiveSide) do(ctx context.Context, periodicWakeup <-chan struct{}) {
+func (j *ActiveSide) do(ctx context.Context) {
 
 	log := GetLogger(ctx)
 	ctx = logging.WithSubsystemLoggers(ctx, log)
@@ -262,6 +266,57 @@ func (j *ActiveSide) do(ctx context.Context, periodicWakeup <-chan struct{}) {
 		}
 	}()
 
+	// watchdog
+	go func() {
+		// if no progress after 1 minute, kill the task
+		wdto := envconst.Duration("ZREPL_JOB_WATCHDOG_TIMEOUT", 1*time.Minute)
+		log.WithField("watchdog_timeout", wdto.String()).Debug("starting watchdog")
+
+		t := time.NewTicker(wdto)
+		defer t.Stop()
+
+		var (
+			rep, prunerSender, prunerReceiver watchdog.Progress
+		)
+		for {
+			select {
+			case <-runDone:
+				return
+			case <-ctx.Done():
+				return
+			case <-t.C: // fall
+			}
+
+			log := log.WithField("watchdog_timeout", wdto.String()) // shadowing!
+
+			j.updateTasks(func(tasks *activeSideTasks) {
+				if tasks.replication != nil &&
+					!tasks.replication.Progress.ExpectProgress(&rep) &&
+					!tasks.replication.State().IsTerminal() {
+					log.Error("replication did not make progress, cancelling")
+					tasks.replicationCancel()
+				}
+				if tasks.prunerSender != nil &&
+					!tasks.prunerSender.Progress.ExpectProgress(&prunerSender) &&
+					!tasks.prunerSender.State().IsTerminal() {
+					log.Error("pruner:sender did not make progress, cancelling")
+					tasks.prunerSenderCancel()
+				}
+				if tasks.prunerReceiver != nil &&
+					!tasks.prunerReceiver.Progress.ExpectProgress(&prunerReceiver) &&
+					!tasks.prunerReceiver.State().IsTerminal() {
+					log.Error("pruner:receiver did not make progress, cancelling")
+					tasks.prunerReceiverCancel()
+				}
+			})
+			log.WithField("replication_progress", rep.String()).
+				WithField("pruner_sender_progress", prunerSender.String()).
+				WithField("pruner_receiver_progress", prunerReceiver.String()).
+				Debug("watchdog did run")
+
+		}
+	}()
+
 	client, err := j.clientFactory.NewClient()
 	if err != nil {
 		log.WithError(err).Error("factory cannot instantiate streamrpc client")
@@ -270,136 +325,54 @@ func (j *ActiveSide) do(ctx context.Context, periodicWakeup <-chan struct{}) {
 
 	sender, receiver, err := j.mode.SenderReceiver(client)
 
-	tasks := j.updateTasks(func(tasks *activeSideTasks) {
-		// reset it
-		*tasks = activeSideTasks{}
-		tasks.replication = replication.NewReplication(j.promRepStateSecs, j.promBytesReplicated)
-	})
-
-	log.Info("start replication")
-	replicationDone := make(chan struct{})
-	replicationCtx, replicationCancel := context.WithCancel(ctx)
-	defer replicationCancel()
-	go func() {
-		tasks.replication.Drive(replicationCtx, sender, receiver)
-		close(replicationDone)
-	}()
-	outer:
-	for {
+	{
 		select {
-		case <-replicationDone:
-			// fine!
-			break outer
-		case <-periodicWakeup:
-			// Replication took longer than the periodic interval.
-			//
-			// For pull jobs, this isn't so bad because nothing changes on the active side
-			// if replication doesn't go forward.
-			//
-			// For push jobs, this means snapshots were taken.
-			// We need to invoke the pruner now, because otherwise an infinitely stuck replication
-			// will cause this side to fill up with snapshots.
-			//
-			// However, there are cases where replication progresses and just takes longer,
-			// and we don't want these situations be interrupted by a prune, which will require
-			// re-planning and starting over (think of initial replication as an example).
-			//
-			// Therefore, we prohibit pruning of snapshots that are part of the current replication plan.
-			// If there is no such plan, we kill the replication.
-
-			if j.mode.Type() == TypePush {
-
-				rep := tasks.replication.Report()
-				state, err := replication.StateString(rep.Status)
-				if err != nil {
-					panic(err)
-				}
-
-				switch state {
-				case replication.Planning:
-					fallthrough
-				case replication.PlanningError:
-					fallthrough
-				case replication.WorkingWait:
-					log.WithField("repl_state", state.String()).
-						Info("cancelling replication after new snapshots invalidated its current state")
-					replicationCancel()
-					log.Info("waiting for replication to stop")
-					<-replicationDone // no need to wait for ctx.Done, replication is already bound to global cancel
-					break outer
-				default:
-					log.WithField("repl_state", state.String()).
-						Warn("new snapshots while replication is running and making progress")
-				}
-
-			}
-
+		case <-ctx.Done():
+			return
+		default:
 		}
+		ctx, repCancel := context.WithCancel(ctx)
+		tasks := j.updateTasks(func(tasks *activeSideTasks) {
+			// reset it
+			*tasks = activeSideTasks{}
+			tasks.replicationCancel = repCancel
+			tasks.replication = replication.NewReplication(j.promRepStateSecs, j.promBytesReplicated)
+		})
+		log.Info("start replication")
+		tasks.replication.Drive(ctx, sender, receiver)
+		repCancel() // always cancel to free up context resources
 	}
 
-
-	var pruningWg sync.WaitGroup
-	log.Info("start pruning sender")
-	pruningWg.Add(1)
-	go func() {
-		defer pruningWg.Done()
+	{
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+		ctx, senderCancel := context.WithCancel(ctx)
 		tasks := j.updateTasks(func(tasks *activeSideTasks) {
 			tasks.prunerSender = j.prunerFactory.BuildSenderPruner(ctx, sender, sender)
+			tasks.prunerSenderCancel = senderCancel
 		})
+		log.Info("start pruning sender")
 		tasks.prunerSender.Prune()
-		// FIXME no need to do the cancellation dance with sender, we know it's local for push
-		// FIXME and we don't worry about pull ATM
-	}()
-	log.Info("start pruning receiver")
-	pruningWg.Add(1)
-	go func() {
-		defer pruningWg.Done()
-
-		receiverPrunerCtx, receiverPrunerCancel := context.WithCancel(ctx)
-		defer receiverPrunerCancel()
-		tasks := j.updateTasks(func(tasks *activeSideTasks) {
-			tasks.prunerReceiver = j.prunerFactory.BuildReceiverPruner(receiverPrunerCtx, receiver, sender)
-		})
-		receiverPrunerDone := make(chan struct{})
-		go func() {
-			defer close(receiverPrunerDone)
-			tasks.prunerReceiver.Prune()
-		}()
-
-		outer:
-		for {
-			select {
-			case <-receiverPrunerDone:
-				// fine!
-				break outer
-			case <-periodicWakeup:
-				// see comments for similar apporach with replication above
-				if j.mode.Type() == TypePush {
-					rep := tasks.prunerReceiver.Report()
-					state, err := pruner.StateString(rep.State)
-					if err != nil {
-						panic(err)
-					}
-					switch state {
-					case pruner.PlanWait:
-						fallthrough
-					case pruner.ExecWait:
-						log.WithField("pruner_state", state.String()).
-							Info("cancelling failing prune on receiver because new snapshots were taken on sender")
-						receiverPrunerCancel()
-						log.Info("waiting for receiver pruner to stop")
-						<-receiverPrunerDone
-						break outer
-					default:
-						log.WithField("pruner_state", state.String()).
-							Warn("new snapshots while prune on receiver is still running")
-					}
-				}
-			}
+		log.Info("finished pruning sender")
+		senderCancel()
+	}
+	{
+		select {
+		case <-ctx.Done():
+			return
+		default:
 		}
-
-	}()
-
-	pruningWg.Wait() // if pruners handle ctx cancellation correctly, we don't need to wait for it here
-
+		ctx, receiverCancel := context.WithCancel(ctx)
+		tasks := j.updateTasks(func(tasks *activeSideTasks) {
+			tasks.prunerReceiver = j.prunerFactory.BuildReceiverPruner(ctx, receiver, sender)
+			tasks.prunerReceiverCancel = receiverCancel
+		})
+		log.Info("start pruning receiver")
+		tasks.prunerReceiver.Prune()
+		log.Info("finished pruning receiver")
+		receiverCancel()
+	}
 }

daemon/job/build_jobs.go

@@ -13,6 +13,9 @@ func JobsFromConfig(c *config.Config) ([]Job, error) {
 		if err != nil {
 			return nil, err
 		}
+		if j == nil || j.Name() == "" {
+			panic(fmt.Sprintf("implementation error: job builder returned nil job type %T", c.Jobs[i].Ret))
+		}
 		js[i] = j
 	}
 	return js, nil
@@ -20,7 +23,7 @@ func JobsFromConfig(c *config.Config) ([]Job, error) {
 
 func buildJob(c *config.Global, in config.JobEnum) (j Job, err error) {
 	cannotBuildJob := func(e error, name string) (Job, error) {
-		return nil, errors.Wrapf(err, "cannot build job %q", name)
+		return nil, errors.Wrapf(e, "cannot build job %q", name)
 	}
 	// FIXME prettify this
 	switch v := in.Ret.(type) {

daemon/pruner/pruner.go

@@ -9,6 +9,8 @@ import (
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/pruning"
 	"github.com/zrepl/zrepl/replication/pdu"
+	"github.com/zrepl/zrepl/util/envconst"
+	"github.com/zrepl/zrepl/util/watchdog"
 	"net"
 	"sort"
 	"sync"
@@ -56,6 +58,8 @@ type args struct {
 type Pruner struct {
 	args args
 
+	Progress watchdog.KeepAlive
+
 	mtx sync.RWMutex
 
 	state State
@@ -110,11 +114,11 @@ func NewPrunerFactory(in config.PruningSenderReceiver, promPruneSecs *prometheus
 		considerSnapAtCursorReplicated = considerSnapAtCursorReplicated || !knr.KeepSnapshotAtCursor
 	}
 	f := &PrunerFactory{
-		keepRulesSender,
-		keepRulesReceiver,
-		10 * time.Second, //FIXME constant
-		considerSnapAtCursorReplicated,
-		promPruneSecs,
+		senderRules: keepRulesSender,
+		receiverRules: keepRulesReceiver,
+		retryWait: envconst.Duration("ZREPL_PRUNER_RETRY_INTERVAL", 4 * time.Second),
+		considerSnapAtCursorReplicated: considerSnapAtCursorReplicated,
+		promPruneSecs: promPruneSecs,
 	}
 	return f, nil
 }
@@ -175,6 +179,10 @@ func (s State) statefunc() state {
 	return statemap[s]
 }
 
+func (s State) IsTerminal() bool {
+	return s.statefunc() == nil
+}
+
 type updater func(func(*Pruner)) State
 type state func(args *args, u updater) state
 
@@ -196,6 +204,12 @@ func (p *Pruner) prune(args args) {
 		GetLogger(args.ctx).
 			WithField("transition", fmt.Sprintf("%s=>%s", pre, post)).
 			Debug("state transition")
+		if err := p.Error(); err != nil {
+			GetLogger(args.ctx).
+				WithError(p.err).
+				WithField("state", post.String()).
+				Error("entering error state after error")
+		}
 	}
 }
 
@@ -249,6 +263,21 @@ func (p *Pruner) Report() *Report {
 	return &r
 }
 
+func (p *Pruner) State() State {
+	p.mtx.Lock()
+	defer p.mtx.Unlock()
+	return p.state
+}
+
+func (p *Pruner) Error() error {
+	p.mtx.Lock()
+	defer p.mtx.Unlock()
+	if p.state & (PlanWait|ExecWait|ErrPerm) != 0 {
+		return p.err
+	}
+	return nil
+}
+
 type fs struct {
 	path  string
 
@@ -316,11 +345,8 @@ func (s snapshot) Replicated() bool { return s.replicated }
 func (s snapshot) Date() time.Time { return s.date }
 
 func shouldRetry(e error) bool {
-	switch e.(type) {
-	case nil:
-		return true
-	case net.Error:
-		return true
+	if neterr, ok := e.(net.Error); ok {
+		return neterr.Temporary()
 	}
 	return false
 }

daemon/pruner/pruner_test.go

@@ -129,12 +129,12 @@ func TestPruner_Prune(t *testing.T) {
 	var _ net.Error = &net.OpError{} // we use it below
 	target := &mockTarget{
 		listFilesystemsErr: []error{
-			stubNetErr{msg: "fakerror0"},
+			stubNetErr{msg: "fakerror0", temporary: true},
 		},
 		listVersionsErrs: map[string][]error{
 			"zroot/foo": {
-				stubNetErr{msg: "fakeerror1"}, // should be classified as temporaty
-				stubNetErr{msg: "fakeerror2"},
+				stubNetErr{msg: "fakeerror1", temporary: true}, // should be classified as temporaty
+				stubNetErr{msg: "fakeerror2", temporary: true,},
 			},
 		},
 		destroyErrs: map[string][]error{
@@ -142,7 +142,7 @@ func TestPruner_Prune(t *testing.T) {
 				fmt.Errorf("permanent error"),
 			},
 			"zroot/bar": {
-				stubNetErr{msg: "fakeerror3"},
+				stubNetErr{msg: "fakeerror3", temporary: true},
 			},
 		},
 		destroyed: make(map[string][]string),
@@ -176,7 +176,7 @@ func TestPruner_Prune(t *testing.T) {
 	history := &mockHistory{
 		errs: map[string][]error{
 			"zroot/foo": {
-				stubNetErr{msg: "fakeerror4"},
+				stubNetErr{msg: "fakeerror4", temporary: true},
 			},
 			"zroot/baz": {
 				fmt.Errorf("permanent error2"),

daemon/transport/connecter/connect_tls.go

@@ -39,5 +39,9 @@ func TLSConnecterFromConfig(in *config.TLSConnect) (*TLSConnecter, error) {
 }
 
 func (c *TLSConnecter) Connect(dialCtx context.Context) (conn net.Conn, err error) {
-	return tls.DialWithDialer(&c.dialer, "tcp", c.Address, c.tlsConfig)
+	conn, err = c.dialer.DialContext(dialCtx, "tcp", c.Address)
+	if err != nil {
+		return nil, err
+	}
+	return tls.Client(conn, c.tlsConfig), nil
 }

docs/changelog.rst

@@ -29,6 +29,14 @@ We use the following annotations for classifying changes:
 This release is a milestone for zrepl and required significant refactoring if not rewrites of substantial parts of the application.
 It breaks both configuration and transport format, and thus requires manual intervention and updates on both sides of a replication setup.
 
+.. DANGER::
+   The changes in the pruning system for this release require you to explicitly define **keep rules**:
+   for any snapshot that you want to keep, at least one rule must match.
+   This is different from previous releases where pruning only affected snapshots with the configured snapshotting prefix.
+   Make sure that snapshots to be kept or ignored by zrepl are covered, e.g. by using the ``regex`` keep rule.
+   :ref:`Learn more in the config docs... <prune>`
+
+
 Notes to Package Maintainers
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

docs/configuration/prune.rst

@@ -45,6 +45,9 @@ Example Configuration:
              regex: "^zrepl_.*"
            # manually created snapshots will be kept forever on receiver
 
+.. DANGER::
+    You might have **existing snapshots** of filesystems affected by pruning which you want to keep, i.e. not be destroyed by zrepl.
+    Make sure to actually add the necessary ``regex`` keep rules on both sides, like with ``manual`` in the example above.
 
 .. ATTENTION::
 

endpoint/endpoint.go

@@ -79,16 +79,17 @@ func (p *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
 	}
 
 	if r.DryRun {
-		size, err := zfs.ZFSSendDry(r.Filesystem, r.From, r.To)
-		if err == zfs.BookmarkSizeEstimationNotSupported {
-			return &pdu.SendRes{ExpectedSize: 0}, nil, nil
-		}
+		si, err := zfs.ZFSSendDry(r.Filesystem, r.From, r.To, "")
 		if err != nil {
 			return nil, nil, err
 		}
-		return &pdu.SendRes{ExpectedSize: size}, nil, nil
+		var expSize int64 = 0 // protocol says 0 means no estimate
+		if si.SizeEstimate != -1 { // but si returns -1 for no size estimate
+			expSize = si.SizeEstimate
+		}
+		return &pdu.SendRes{ExpectedSize: expSize}, nil, nil
 	} else {
-		stream, err := zfs.ZFSSend(r.Filesystem, r.From, r.To)
+		stream, err := zfs.ZFSSend(ctx, r.Filesystem, r.From, r.To, "")
 		if err != nil {
 			return nil, nil, err
 		}
@@ -278,7 +279,7 @@ func (e *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, sendStream
 
 	getLogger(ctx).Debug("start receive command")
 
-	if err := zfs.ZFSRecv(lp.ToString(), sendStream, args...); err != nil {
+	if err := zfs.ZFSRecv(ctx, lp.ToString(), sendStream, args...); err != nil {
 		getLogger(ctx).
 			WithError(err).
 			WithField("args", args).

replication/fsrep/fsfsm.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/zrepl/zrepl/util/watchdog"
 	"io"
 	"net"
 	"sync"
@@ -75,7 +76,7 @@ type State uint
 
 const (
 	Ready State = 1 << iota
-	RetryWait
+	Retry
 	PermanentError
 	Completed
 )
@@ -83,13 +84,17 @@ const (
 func (s State) fsrsf() state {
 	m := map[State]state{
 		Ready:          stateReady,
-		RetryWait:      stateRetryWait,
+		Retry:          stateRetry,
 		PermanentError: nil,
 		Completed:      nil,
 	}
 	return m[s]
 }
 
+func (s State) IsErrorState() bool {
+	return s & (Retry|PermanentError) != 0
+}
+
 type Replication struct {
 	promBytesReplicated prometheus.Counter
 
@@ -98,7 +103,6 @@ type Replication struct {
 	state              State
 	fs                 string
 	err                error
-	retryWaitUntil     time.Time
 	completed, pending []*ReplicationStep
 }
 
@@ -108,6 +112,15 @@ func (f *Replication) State() State {
 	return f.state
 }
 
+func (f *Replication) Err() error {
+	f.lock.Lock()
+	defer f.lock.Unlock()
+	if f.state & (Retry|PermanentError) != 0 {
+		return f.err
+	}
+	return nil
+}
+
 func (f *Replication) UpdateSizeEsitmate(ctx context.Context, sender Sender) error {
 	f.lock.Lock()
 	defer f.lock.Unlock()
@@ -191,7 +204,7 @@ type ReplicationStep struct {
 	expectedSize int64 // 0 means no size estimate present / possible
 }
 
-func (f *Replication) TakeStep(ctx context.Context, sender Sender, receiver Receiver) (post State, nextStepDate, retryWaitUntil time.Time) {
+func (f *Replication) TakeStep(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver) (post State, nextStepDate time.Time) {
 
 	var u updater = func(fu func(*Replication)) State {
 		f.lock.Lock()
@@ -205,7 +218,7 @@ func (f *Replication) TakeStep(ctx context.Context, sender Sender, receiver Rece
 
 	pre := u(nil)
 	preTime := time.Now()
-	s = s(ctx, sender, receiver, u)
+	s = s(ctx, ka, sender, receiver, u)
 	delta := time.Now().Sub(preTime)
 
 	post = u(func(f *Replication) {
@@ -213,7 +226,6 @@ func (f *Replication) TakeStep(ctx context.Context, sender Sender, receiver Rece
 			return
 		}
 		nextStepDate = f.pending[0].to.SnapshotTime()
-		retryWaitUntil = f.retryWaitUntil
 	})
 
 	getLogger(ctx).
@@ -222,22 +234,14 @@ func (f *Replication) TakeStep(ctx context.Context, sender Sender, receiver Rece
 		WithField("duration", delta).
 		Debug("fsr step taken")
 
-	return post, nextStepDate, retryWaitUntil
-}
-
-func (f *Replication) RetryWaitUntil() time.Time {
-	f.lock.Lock()
-	defer f.lock.Unlock()
-	return f.retryWaitUntil
+	return post, nextStepDate
 }
 
 type updater func(func(fsr *Replication)) State
 
-type state func(ctx context.Context, sender Sender, receiver Receiver, u updater) state
+type state func(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver, u updater) state
 
-var RetrySleepDuration = 10 * time.Second // FIXME make configurable
-
-func stateReady(ctx context.Context, sender Sender, receiver Receiver, u updater) state {
+func stateReady(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver, u updater) state {
 
 	var current *ReplicationStep
 	s := u(func(f *Replication) {
@@ -251,7 +255,7 @@ func stateReady(ctx context.Context, sender Sender, receiver Receiver, u updater
 		return s.fsrsf()
 	}
 
-	stepState := current.doReplication(ctx, sender, receiver)
+	stepState := current.doReplication(ctx, ka, sender, receiver)
 
 	return u(func(f *Replication) {
 		switch stepState {
@@ -266,8 +270,7 @@ func stateReady(ctx context.Context, sender Sender, receiver Receiver, u updater
 		case StepReplicationRetry:
 			fallthrough
 		case StepMarkReplicatedRetry:
-			f.retryWaitUntil = time.Now().Add(RetrySleepDuration)
-			f.state = RetryWait
+			f.state = Retry
 		case StepPermanentError:
 			f.state = PermanentError
 			f.err = errors.New("a replication step failed with a permanent error")
@@ -277,16 +280,9 @@ func stateReady(ctx context.Context, sender Sender, receiver Receiver, u updater
 	}).fsrsf()
 }
 
-func stateRetryWait(ctx context.Context, sender Sender, receiver Receiver, u updater) state {
-	var sleepUntil time.Time
-	u(func(f *Replication) {
-		sleepUntil = f.retryWaitUntil
-	})
-	if time.Now().Before(sleepUntil) {
-		return u(nil).fsrsf()
-	}
-	return u(func(f *Replication) {
-		f.state = Ready
+func stateRetry(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver, u updater) state {
+	return u(func(fsr *Replication) {
+		fsr.state = Ready
 	}).fsrsf()
 }
 
@@ -313,8 +309,8 @@ func (fsr *Replication) Report() *Report {
 		rep.Pending[i] = fsr.pending[i].Report()
 	}
 
-	if fsr.state&RetryWait != 0 {
-		if len(rep.Pending) != 0 { // should always be true for RetryWait == true?
+	if fsr.state&Retry != 0 {
+		if len(rep.Pending) != 0 { // should always be true for Retry == true?
 			rep.Problem = rep.Pending[0].Problem
 		}
 	}
@@ -337,7 +333,7 @@ func shouldRetry(err error) bool {
 	return false
 }
 
-func (s *ReplicationStep) doReplication(ctx context.Context, sender Sender, receiver Receiver) StepState {
+func (s *ReplicationStep) doReplication(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver) StepState {
 
 	fs := s.parent.fs
 
@@ -380,6 +376,9 @@ func (s *ReplicationStep) doReplication(ctx context.Context, sender Sender, rece
 	}
 
 	s.byteCounter = util.NewByteCounterReader(sstream)
+	s.byteCounter.SetCallback(1*time.Second, func(i int64) {
+		ka.MadeProgress()
+	})
 	defer func() {
 		s.parent.promBytesReplicated.Add(float64(s.byteCounter.Bytes()))
 	}()
@@ -404,14 +403,15 @@ func (s *ReplicationStep) doReplication(ctx context.Context, sender Sender, rece
 		return updateStateError(err)
 	}
 	log.Debug("receive finished")
+	ka.MadeProgress()
 
 	updateStateCompleted()
 
-	return s.doMarkReplicated(ctx, sender)
+	return s.doMarkReplicated(ctx, ka, sender)
 
 }
 
-func (s *ReplicationStep) doMarkReplicated(ctx context.Context, sender Sender) StepState {
+func (s *ReplicationStep) doMarkReplicated(ctx context.Context, ka *watchdog.KeepAlive, sender Sender) StepState {
 
 	log := getLogger(ctx).
 		WithField("filesystem", s.parent.fs).
@@ -456,6 +456,7 @@ func (s *ReplicationStep) doMarkReplicated(ctx context.Context, sender Sender) S
 		log.Error(err.Error())
 		return updateStateError(err)
 	}
+	ka.MadeProgress()
 
 	return updateStateCompleted()
 }
@@ -485,7 +486,7 @@ func (s *ReplicationStep) buildSendRequest(dryRun bool) (sr *pdu.SendReq) {
 	if s.from == nil {
 		sr = &pdu.SendReq{
 			Filesystem: fs,
-			From:       s.to.RelName(), // FIXME fix protocol to use To, like zfs does internally
+			To:         s.to.RelName(),
 			DryRun:     dryRun,
 		}
 	} else {

replication/fsrep/state_string.go

@@ -5,13 +5,13 @@ package fsrep
 import "strconv"
 
 const (
-	_State_name_0 = "ReadyRetryWait"
+	_State_name_0 = "ReadyRetry"
 	_State_name_1 = "PermanentError"
 	_State_name_2 = "Completed"
 )
 
 var (
-	_State_index_0 = [...]uint8{0, 5, 14}
+	_State_index_0 = [...]uint8{0, 5, 10}
 )
 
 func (i State) String() string {

replication/internal/queue/queue.go

@@ -11,9 +11,8 @@ type replicationQueueItem struct {
 	// duplicates fsr.state to avoid accessing and locking fsr
 	state State
 	// duplicates fsr.current.nextStepDate to avoid accessing & locking fsr
-	nextStepDate time.Time
-	// duplicates fsr.retryWaitUntil to avoid accessing & locking fsr
-	retryWaitUntil time.Time
+	nextStepDate         time.Time
+	errorStateEnterCount int
 
 	fsr *Replication
 }
@@ -40,10 +39,10 @@ var lessmap = map[State]lessmapEntry{
 			return a.nextStepDate.Before(b.nextStepDate)
 		},
 	},
-	RetryWait: {
+	Retry: {
 		prio: 1,
 		less: func(a, b *replicationQueueItem) bool {
-			return a.retryWaitUntil.Before(b.retryWaitUntil)
+			return a.errorStateEnterCount < b.errorStateEnterCount
 		},
 	},
 }
@@ -114,8 +113,10 @@ func (h ReplicationQueueItemHandle) GetFSReplication() *Replication {
 	return h.i.fsr
 }
 
-func (h ReplicationQueueItemHandle) Update(newState State, nextStepDate, retryWaitUntil time.Time) {
+func (h ReplicationQueueItemHandle) Update(newState State, nextStepDate time.Time) {
 	h.i.state = newState
 	h.i.nextStepDate = nextStepDate
-	h.i.retryWaitUntil = retryWaitUntil
+	if h.i.state.IsErrorState() {
+		h.i.errorStateEnterCount++
+	}
 }

replication/mainfsm.go

@@ -8,7 +8,10 @@ import (
 	"fmt"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/zrepl/zrepl/daemon/job/wakeup"
+	"github.com/zrepl/zrepl/util/envconst"
+	"github.com/zrepl/zrepl/util/watchdog"
 	"math/bits"
+	"net"
 	"sync"
 	"time"
 
@@ -27,7 +30,7 @@ const (
 	Working
 	WorkingWait
 	Completed
-	ContextDone
+	PermanentError
 )
 
 func (s State) rsf() state {
@@ -46,6 +49,10 @@ func (s State) rsf() state {
 	return m[idx]
 }
 
+func (s State) IsTerminal() bool {
+	return s.rsf() == nil
+}
+
 // Replication implements the replication of multiple file systems from a Sender to a Receiver.
 //
 // It is a state machine that is driven by the Drive method
@@ -55,6 +62,8 @@ type Replication struct {
 	promSecsPerState *prometheus.HistogramVec // labels: state
 	promBytesReplicated *prometheus.CounterVec // labels: filesystem
 
+	Progress watchdog.KeepAlive
+
 	// lock protects all fields of this struct (but not the fields behind pointers!)
 	lock sync.Mutex
 
@@ -65,11 +74,8 @@ type Replication struct {
 	completed []*fsrep.Replication
 	active    *ReplicationQueueItemHandle
 
-	// PlanningError
-	planningError error
-
-	// ContextDone
-	contextError error
+	// for PlanningError, WorkingWait and ContextError and Completed
+	err error
 
 	// PlanningError, WorkingWait
 	sleepUntil time.Time
@@ -124,7 +130,7 @@ func NewFilteredError(fs string) *FilteredError {
 func (f FilteredError) Error() string { return "endpoint does not allow access to filesystem " + f.fs }
 
 type updater func(func(*Replication)) (newState State)
-type state func(ctx context.Context, sender Sender, receiver Receiver, u updater) state
+type state func(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver, u updater) state
 
 // Drive starts the state machine and returns only after replication has finished (with or without errors).
 // The Logger in ctx is used for both debug and error logging, but is not guaranteed to be stable
@@ -149,7 +155,7 @@ func (r *Replication) Drive(ctx context.Context, sender Sender, receiver Receive
 	for s != nil {
 		preTime := time.Now()
 		pre = u(nil)
-		s = s(ctx, sender, receiver, u)
+		s = s(ctx, &r.Progress, sender, receiver, u)
 		delta := time.Now().Sub(preTime)
 		r.promSecsPerState.WithLabelValues(pre.String()).Observe(delta.Seconds())
 		post = u(nil)
@@ -187,20 +193,34 @@ func resolveConflict(conflict error) (path []*pdu.FilesystemVersion, msg string)
 	return nil, "no automated way to handle conflict type"
 }
 
-var PlanningRetryInterval = 10 * time.Second // FIXME make constant onfigurable
+var RetryInterval = envconst.Duration("ZREPL_REPLICATION_RETRY_INTERVAL", 4 * time.Second)
 
-func statePlanning(ctx context.Context, sender Sender, receiver Receiver, u updater) state {
+func isPermanent(err error) bool {
+	switch err {
+	case context.Canceled: return true
+	case context.DeadlineExceeded: return true
+	}
+	if operr, ok := err.(net.Error); ok {
+		return !operr.Temporary()
+	}
+	return false
+}
+
+func statePlanning(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver, u updater) state {
 
 	log := getLogger(ctx)
 
 	log.Info("start planning")
 
 	handlePlanningError := func(err error) state {
-		// FIXME classify error as temporary or permanent / max retry counter
 		return u(func(r *Replication) {
-			r.sleepUntil = time.Now().Add(PlanningRetryInterval)
-			r.planningError = err
-			r.state = PlanningError
+			r.err = err
+			if isPermanent(err) {
+				r.state = PermanentError
+			} else {
+				r.sleepUntil = time.Now().Add(RetryInterval)
+				r.state = PlanningError
+			}
 		}).rsf()
 	}
 
@@ -295,18 +315,21 @@ func statePlanning(ctx context.Context, sender Sender, receiver Receiver, u upda
 			log.WithError(err).Error("error computing size estimate")
 			return handlePlanningError(err)
 		}
+
 		q.Add(qitem)
 	}
 
+	ka.MadeProgress()
+
 	return u(func(r *Replication) {
 		r.completed = nil
 		r.queue = q
-		r.planningError = nil
+		r.err = nil
 		r.state = Working
 	}).rsf()
 }
 
-func statePlanningError(ctx context.Context, sender Sender, receiver Receiver, u updater) state {
+func statePlanningError(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver, u updater) state {
 	var sleepUntil time.Time
 	u(func(r *Replication) {
 		sleepUntil = r.sleepUntil
@@ -317,8 +340,8 @@ func statePlanningError(ctx context.Context, sender Sender, receiver Receiver, u
 	select {
 	case <-ctx.Done():
 		return u(func(r *Replication) {
-			r.state = ContextDone
-			r.contextError = ctx.Err()
+			r.state = PermanentError
+			r.err = ctx.Err()
 		}).rsf()
 	case <-t.C:
 	case <-wakeup.Wait(ctx):
@@ -328,7 +351,7 @@ func statePlanningError(ctx context.Context, sender Sender, receiver Receiver, u
 	}).rsf()
 }
 
-func stateWorking(ctx context.Context, sender Sender, receiver Receiver, u updater) state {
+func stateWorking(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver, u updater) state {
 
 	var active *ReplicationQueueItemHandle
 	rsfNext := u(func(r *Replication) {
@@ -345,34 +368,49 @@ func stateWorking(ctx context.Context, sender Sender, receiver Receiver, u updat
 		return rsfNext
 	}
 
-	retryWaitUntil := active.GetFSReplication().RetryWaitUntil()
-	if retryWaitUntil.After(time.Now()) {
+	state, nextStepDate := active.GetFSReplication().TakeStep(ctx, ka, sender, receiver)
+	u(func(r *Replication) {
+		active.Update(state, nextStepDate)
+		r.active = nil
+	}).rsf()
+
+	select {
+	case <-ctx.Done():
 		return u(func(r *Replication) {
-			r.sleepUntil = retryWaitUntil
-			r.state = WorkingWait
+			r.err = ctx.Err()
+			r.state = PermanentError
+		}).rsf()
+	default:
+	}
+
+	if err := active.GetFSReplication().Err(); err != nil {
+		return u(func(r *Replication) {
+			r.err = err
+			if isPermanent(err) {
+				r.state = PermanentError
+			} else {
+				r.sleepUntil = time.Now().Add(RetryInterval)
+				r.state = WorkingWait
+			}
 		}).rsf()
 	}
 
-	state, nextStepDate, retryWaitUntil := active.GetFSReplication().TakeStep(ctx, sender, receiver)
-	return u(func(r *Replication) {
-		active.Update(state, nextStepDate, retryWaitUntil)
-		r.active = nil
-	}).rsf()
+	return u(nil).rsf()
 }
 
-func stateWorkingWait(ctx context.Context, sender Sender, receiver Receiver, u updater) state {
+func stateWorkingWait(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver, u updater) state {
 	var sleepUntil time.Time
 	u(func(r *Replication) {
 		sleepUntil = r.sleepUntil
 	})
-	t := time.NewTimer(PlanningRetryInterval)
-	getLogger(ctx).WithField("until", sleepUntil).Info("retry wait because no filesystems are ready")
+	t := time.NewTimer(RetryInterval)
+	getLogger(ctx).WithField("until", sleepUntil).Info("retry wait after replication step error")
 	defer t.Stop()
 	select {
 	case <-ctx.Done():
 		return u(func(r *Replication) {
-			r.state = ContextDone
-			r.contextError = ctx.Err()
+			r.state = PermanentError
+			r.err = ctx.Err()
 		}).rsf()
 
 	case <-t.C:
@@ -395,12 +433,9 @@ func (r *Replication) Report() *Report {
 		SleepUntil: r.sleepUntil,
 	}
 
-	if r.state&(Planning|PlanningError|ContextDone) != 0 {
-		switch r.state {
-		case PlanningError:
-			rep.Problem = r.planningError.Error()
-		case ContextDone:
-			rep.Problem = r.contextError.Error()
+	if r.state&(Planning|PlanningError|PermanentError) != 0 {
+		if r.err != nil {
+			rep.Problem = r.err.Error()
 		}
 		return &rep
 	}
@@ -425,3 +460,9 @@ func (r *Replication) Report() *Report {
 
 	return &rep
 }
+
+func (r *Replication) State() State {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+	return r.state
+}

replication/state_enumer.go

@@ -11,7 +11,7 @@ const (
 	_StateName_1 = "Working"
 	_StateName_2 = "WorkingWait"
 	_StateName_3 = "Completed"
-	_StateName_4 = "ContextDone"
+	_StateName_4 = "PermanentError"
 )
 
 var (
@@ -19,7 +19,7 @@ var (
 	_StateIndex_1 = [...]uint8{0, 7}
 	_StateIndex_2 = [...]uint8{0, 11}
 	_StateIndex_3 = [...]uint8{0, 9}
-	_StateIndex_4 = [...]uint8{0, 11}
+	_StateIndex_4 = [...]uint8{0, 14}
 )
 
 func (i State) String() string {
@@ -48,7 +48,7 @@ var _StateNameToValueMap = map[string]State{
 	_StateName_1[0:7]:  4,
 	_StateName_2[0:11]: 8,
 	_StateName_3[0:9]:  16,
-	_StateName_4[0:11]: 32,
+	_StateName_4[0:14]: 32,
 }
 
 // StateString retrieves an enum value from the enum constants string name.

util/envconst/envconst.go

@@ -0,0 +1,25 @@
+package envconst
+
+import (
+	"os"
+	"sync"
+	"time"
+)
+
+var cache sync.Map
+
+func Duration(varname string, def time.Duration) time.Duration {
+	if v, ok := cache.Load(varname); ok {
+		return v.(time.Duration)
+	}
+	e := os.Getenv(varname)
+	if e == "" {
+		return def
+	}
+	d, err := time.ParseDuration(e)
+	if err != nil {
+		panic(err)
+	}
+	cache.Store(varname, d)
+	return d
+}

util/io.go

@@ -5,6 +5,7 @@ import (
 	"net"
 	"os"
 	"sync/atomic"
+	"time"
 )
 
 type NetConnLogger struct {
@@ -101,6 +102,14 @@ func (c *ChainedReader) Read(buf []byte) (n int, err error) {
 
 type ByteCounterReader struct {
 	reader    io.ReadCloser
+
+	// called & accessed synchronously during Read, no external access
+	cb        func(full int64)
+	cbEvery   time.Duration
+	lastCbAt  time.Time
+	bytesSinceLastCb int64
+
+	// set atomically because it may be read by multiple threads
 	bytes     int64
 }
 
@@ -110,13 +119,23 @@ func NewByteCounterReader(reader io.ReadCloser) *ByteCounterReader {
 	}
 }
 
+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)
-	atomic.AddInt64(&b.bytes, int64(n))
+	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
 }
 

util/iocommand.go

@@ -2,6 +2,7 @@ package util
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"io"
 	"os"
@@ -34,8 +35,8 @@ func (e IOCommandError) Error() string {
 	return fmt.Sprintf("underlying process exited with error: %s\nstderr: %s\n", e.WaitErr, e.Stderr)
 }
 
-func RunIOCommand(command string, args ...string) (c *IOCommand, err error) {
-	c, err = NewIOCommand(command, args, IOCommandStderrBufSize)
+func RunIOCommand(ctx context.Context, command string, args ...string) (c *IOCommand, err error) {
+	c, err = NewIOCommand(ctx, command, args, IOCommandStderrBufSize)
 	if err != nil {
 		return
 	}
@@ -43,7 +44,7 @@ func RunIOCommand(command string, args ...string) (c *IOCommand, err error) {
 	return
 }
 
-func NewIOCommand(command string, args []string, stderrBufSize int) (c *IOCommand, err error) {
+func NewIOCommand(ctx context.Context, command string, args []string, stderrBufSize int) (c *IOCommand, err error) {
 
 	if stderrBufSize == 0 {
 		stderrBufSize = IOCommandStderrBufSize
@@ -51,7 +52,7 @@ func NewIOCommand(command string, args []string, stderrBufSize int) (c *IOComman
 
 	c = &IOCommand{}
 
-	c.Cmd = exec.Command(command, args...)
+	c.Cmd = exec.CommandContext(ctx, command, args...)
 
 	if c.Stdout, err = c.Cmd.StdoutPipe(); err != nil {
 		return

util/watchdog/watchdog.go

@@ -0,0 +1,41 @@
+package watchdog
+
+import (
+	"fmt"
+	"sync"
+	"time"
+)
+
+type Progress struct {
+	lastUpd time.Time
+}
+
+func (p *Progress) String() string {
+	return fmt.Sprintf("last update at %s", p.lastUpd)
+}
+
+func (p *Progress) madeProgressSince(p2 *Progress) bool {
+	if p.lastUpd.IsZero() && p2.lastUpd.IsZero() {
+		return false
+	}
+	return p.lastUpd.After(p2.lastUpd)
+}
+
+type KeepAlive struct {
+	mtx sync.Mutex
+	p Progress
+}
+
+func (k *KeepAlive) MadeProgress() {
+	k.mtx.Lock()
+	defer k.mtx.Unlock()
+	k.p.lastUpd = time.Now()
+}
+
+func (k *KeepAlive) ExpectProgress(last *Progress) (madeProgress bool) {
+	k.mtx.Lock()
+	defer k.mtx.Unlock()
+	madeProgress = k.p.madeProgressSince(last)
+	*last = k.p
+	return madeProgress
+}

zfs/zfs.go

@@ -286,92 +286,176 @@ func absVersion(fs, v string) (full string, err error) {
 	return fmt.Sprintf("%s%s", fs, v), nil
 }
 
-func ZFSSend(fs string, from, to string) (stream io.ReadCloser, err error) {
+func buildCommonSendArgs(fs string, from, to string, token string) ([]string, error) {
+	args := make([]string, 0, 3)
+	if token != "" {
+		args = append(args, "-t", token)
+		return args, nil
+	}
 
-	fromV, err := absVersion(fs, from)
+	toV, err := absVersion(fs, to)
 	if err != nil {
 		return nil, err
 	}
 
-	toV := ""
-	if to != "" {
-		toV, err = absVersion(fs, to)
+	fromV := ""
+	if from != "" {
+		fromV, err = absVersion(fs, from)
 		if err != nil {
 			return nil, err
 		}
 	}
 
-	args := make([]string, 0)
-	args = append(args, "send")
-
-	if toV == "" { // Initial
-		args = append(args, fromV)
+	if fromV == "" { // Initial
+		args = append(args, toV)
 	} else {
 		args = append(args, "-i", fromV, toV)
 	}
+	return args, nil
+}
 
-	stream, err = util.RunIOCommand(ZFS_BINARY, args...)
+// if token != "", then send -t token is used
+// otherwise send [-i from] to is used
+// (if from is "" a full ZFS send is done)
+func ZFSSend(ctx context.Context, fs string, from, to string, token string) (stream io.ReadCloser, err error) {
+
+	args := make([]string, 0)
+	args = append(args, "send")
+
+	sargs, err := buildCommonSendArgs(fs, from, to, token)
+	if err != nil {
+		return nil, err
+	}
+	args = append(args, sargs...)
+
+	stream, err = util.RunIOCommand(ctx, ZFS_BINARY, args...)
 
 	return
 }
 
-var BookmarkSizeEstimationNotSupported error = fmt.Errorf("size estimation is not supported for bookmarks")
 
-// May return BookmarkSizeEstimationNotSupported as err if from is a bookmark.
-func ZFSSendDry(fs string, from, to string) (size int64, err error) {
+type DrySendType string
 
-	fromV, err := absVersion(fs, from)
-	if err != nil {
-		return 0, err
+const (
+	DrySendTypeFull        DrySendType = "full"
+	DrySendTypeIncremental DrySendType = "incremental"
+)
+
+func DrySendTypeFromString(s string) (DrySendType, error) {
+	switch s {
+	case string(DrySendTypeFull): return DrySendTypeFull, nil
+	case string(DrySendTypeIncremental): return DrySendTypeIncremental, nil
+	default:
+		return "", fmt.Errorf("unknown dry send type %q", s)
 	}
+}
 
-	toV := ""
-	if to != "" {
-		toV, err = absVersion(fs, to)
+type DrySendInfo struct {
+	Type DrySendType
+	Filesystem string // parsed from To field
+	From, To string // direct copy from ZFS output
+	SizeEstimate int64 // -1 if size estimate is not possible
+}
+
+var sendDryRunInfoLineRegex = regexp.MustCompile(`^(full|incremental)(\t(\S+))?\t(\S+)\t([0-9]+)$`)
+
+// see test cases for example output
+func (s *DrySendInfo) unmarshalZFSOutput(output []byte) (err error) {
+	lines := strings.Split(string(output), "\n")
+	for _, l := range lines {
+		regexMatched, err := s.unmarshalInfoLine(l)
 		if err != nil {
-			return 0, err
+			return fmt.Errorf("line %q: %s", l, err)
 		}
+		if !regexMatched {
+			continue
+		}
+		return nil
+	}
+	return fmt.Errorf("no match for info line (regex %s)", sendDryRunInfoLineRegex)
+}
+
+
+// unmarshal info line, looks like this:
+//   full	zroot/test/a@1	5389768
+//   incremental	zroot/test/a@1	zroot/test/a@2	5383936
+// => see test cases
+func (s *DrySendInfo) unmarshalInfoLine(l string) (regexMatched bool, err error) {
+
+	m := sendDryRunInfoLineRegex.FindStringSubmatch(l)
+	if m == nil {
+		return false, nil
+	}
+	s.Type, err = DrySendTypeFromString(m[1])
+	if err != nil {
+		return true, err
 	}
 
-	if strings.Contains(fromV, "#") {
+	s.From = m[3]
+	s.To = m[4]
+	toFS, _, _ , err := DecomposeVersionString(s.To)
+	if err != nil {
+		return true, fmt.Errorf("'to' is not a valid filesystem version: %s", err)
+	}
+	s.Filesystem = toFS
+
+	s.SizeEstimate, err = strconv.ParseInt(m[5], 10, 64)
+	if err != nil {
+		return true, fmt.Errorf("cannot not parse size: %s", err)
+	}
+
+	return true, nil
+}
+
+// from may be "", in which case a full ZFS send is done
+// May return BookmarkSizeEstimationNotSupported as err if from is a bookmark.
+func ZFSSendDry(fs string, from, to string, token string) (_ *DrySendInfo, err error) {
+
+	if strings.Contains(from, "#") {
 		/* TODO:
 		 * ZFS at the time of writing does not support dry-run send because size-estimation
 		 * uses fromSnap's deadlist. However, for a bookmark, that deadlist no longer exists.
 		 * Redacted send & recv will bring this functionality, see
 		 * 	https://github.com/openzfs/openzfs/pull/484
 		 */
-		 return 0, BookmarkSizeEstimationNotSupported
+		 fromAbs, err := absVersion(fs, from)
+		 if err != nil {
+		 	return nil, fmt.Errorf("error building abs version for 'from': %s", err)
+		 }
+		 toAbs, err := absVersion(fs, to)
+		 if err != nil {
+		 	return nil, fmt.Errorf("error building abs version for 'to': %s", err)
+		 }
+		 return &DrySendInfo{
+			Type: DrySendTypeIncremental,
+			Filesystem: fs,
+			From: fromAbs,
+			To: toAbs,
+			SizeEstimate: -1}, nil
 	}
 
 	args := make([]string, 0)
 	args = append(args, "send", "-n", "-v", "-P")
-
-	if toV == "" { // Initial
-		args = append(args, fromV)
-	} else {
-		args = append(args, "-i", fromV, toV)
+	sargs, err := buildCommonSendArgs(fs, from, to, token)
+	if err != nil {
+		return nil, err
 	}
+	args = append(args, sargs...)
 
 	cmd := exec.Command(ZFS_BINARY, args...)
 	output, err := cmd.CombinedOutput()
 	if err != nil {
-		return 0, err
+		return nil, err
 	}
-	o := string(output)
-	lines := strings.Split(o, "\n")
-	if len(lines) < 2 {
-		return 0, errors.New("zfs send -n did not return the expected number of lines")
+	var si DrySendInfo
+	if err := si.unmarshalZFSOutput(output); err != nil {
+		return nil, fmt.Errorf("could not parse zfs send -n output: %s", err)
 	}
-	fields := strings.Fields(lines[1])
-	if len(fields) != 2 {
-		return 0, errors.New("zfs send -n returned unexpexted output")
-	}
-
-	size, err = strconv.ParseInt(fields[1], 10, 64)
-	return size, err
+	return &si, nil
 }
 
-func ZFSRecv(fs string, stream io.Reader, additionalArgs ...string) (err error) {
+
+func ZFSRecv(ctx context.Context, fs string, stream io.Reader, additionalArgs ...string) (err error) {
 
 	if err := validateZFSFilesystem(fs); err != nil {
 		return err
@@ -384,7 +468,7 @@ func ZFSRecv(fs string, stream io.Reader, additionalArgs ...string) (err error)
 	}
 	args = append(args, fs)
 
-	cmd := exec.Command(ZFS_BINARY, args...)
+	cmd := exec.CommandContext(ctx, ZFS_BINARY, args...)
 
 	stderr := bytes.NewBuffer(make([]byte, 0, 1024))
 	cmd.Stderr = stderr
@@ -413,25 +497,30 @@ func ZFSRecv(fs string, stream io.Reader, additionalArgs ...string) (err error)
 	return nil
 }
 
-func ZFSRecvWriter(fs *DatasetPath, additionalArgs ...string) (io.WriteCloser, error) {
+type ClearResumeTokenError struct {
+	ZFSOutput []byte
+	CmdError error
+}
 
-	args := make([]string, 0)
-	args = append(args, "recv")
-	if len(args) > 0 {
-		args = append(args, additionalArgs...)
+func (e ClearResumeTokenError) Error() string {
+	return fmt.Sprintf("could not clear resume token: %q", string(e.ZFSOutput))
+}
+
+// always returns *ClearResumeTokenError
+func ZFSRecvClearResumeToken(fs string) (err error) {
+	if err := validateZFSFilesystem(fs); err != nil {
+		return err
 	}
-	args = append(args, fs.ToString())
 
-	cmd, err := util.NewIOCommand(ZFS_BINARY, args, 1024)
+	cmd := exec.Command(ZFS_BINARY, "recv", "-A", fs)
+	o, err := cmd.CombinedOutput()
 	if err != nil {
-		return nil, err
+		if bytes.Contains(o, []byte("does not have any resumable receive state to abort")) {
+			return nil
+		}
+		return &ClearResumeTokenError{o, err}
 	}
-
-	if err = cmd.Start(); err != nil {
-		return nil, err
-	}
-
-	return cmd.Stdin, nil
+	return nil
 }
 
 type ZFSProperties struct {

zfs/zfs_test.go

@@ -68,3 +68,140 @@ func TestZFSPropertySource(t *testing.T) {
 	}
 
 }
+
+func TestDrySendInfo(t *testing.T) {
+
+	// # full send
+	// $ zfs send -Pnv -t 1-9baebea70-b8-789c636064000310a500c4ec50360710e72765a52697303030419460caa7a515a79680647ce0f26c48f2499525a9c5405ac3c90fabfe92fcf4d2cc140686b30972c7850efd0cd24092e704cbe725e6a632305415e5e797e803cd2ad14f743084b805001b201795
+	fullSend := `
+resume token contents:
+nvlist version: 0
+	object = 0x2
+	offset = 0x4c0000
+	bytes = 0x4e4228
+	toguid = 0x52f9c212c71e60cd
+	toname = zroot/test/a@1
+full	zroot/test/a@1	5389768
+`
+
+	// # incremental send with token
+	// $ zfs send -nvP -t 1-ef01e717e-e0-789c636064000310a501c49c50360710a715e5e7a69766a63040c1d904b9e342877e062900d9ec48eaf293b252934b181898a0ea30e4d3d28a534b40323e70793624f9a4ca92d46220fdc1ce0fabfe927c882bc46c8a0a9f71ad3baf8124cf0996cf4bcc4d6560a82acacf2fd1079a55a29fe86004710b00d8ae1f93
+	incSend := `
+resume token contents:
+nvlist version: 0
+	fromguid = 0x52f9c212c71e60cd
+	object = 0x2
+	offset = 0x4c0000
+	bytes = 0x4e3ef0
+	toguid = 0xcfae0ae671723c16
+	toname = zroot/test/a@2
+incremental	zroot/test/a@1	zroot/test/a@2	5383936
+`
+
+	// # incremental send with token + bookmarmk
+	// $ zfs send -nvP -t 1-ef01e717e-e0-789c636064000310a501c49c50360710a715e5e7a69766a63040c1d904b9e342877e062900d9ec48eaf293b252934b181898a0ea30e4d3d28a534b40323e70793624f9a4ca92d46220fdc1ce0fabfe927c882bc46c8a0a9f71ad3baf8124cf0996cf4bcc4d6560a82acacf2fd1079a55a29fe86004710b00d8ae1f93
+	incSendBookmark := `
+resume token contents:
+nvlist version: 0
+	fromguid = 0x52f9c212c71e60cd
+	object = 0x2
+	offset = 0x4c0000
+	bytes = 0x4e3ef0
+	toguid = 0xcfae0ae671723c16
+	toname = zroot/test/a@2
+incremental	zroot/test/a#1	zroot/test/a@2	5383312
+
+`
+
+	// incremental send without token
+	// $ sudo zfs send -nvP -i @1 zroot/test/a@2
+	incNoToken := `
+incremental	1	zroot/test/a@2	10511856
+size	10511856
+`
+	// full send without token
+	// $ sudo zfs send -nvP  zroot/test/a@3
+	fullNoToken := `
+full	zroot/test/a@3	10518512
+size	10518512
+`
+
+	type tc struct {
+		name string
+		in string
+		exp *DrySendInfo
+		expErr bool
+	}
+
+	tcs := []tc{
+		{
+			name: "fullSend", in: fullSend,
+			exp: &DrySendInfo{
+				Type: DrySendTypeFull,
+				Filesystem: "zroot/test/a",
+				From: "",
+				To: "zroot/test/a@1",
+				SizeEstimate: 5389768,
+			},
+		},
+		{
+			name: "incSend", in: incSend,
+			exp: &DrySendInfo{
+				Type:         DrySendTypeIncremental,
+				Filesystem: "zroot/test/a",
+				From:         "zroot/test/a@1",
+				To:           "zroot/test/a@2",
+				SizeEstimate: 5383936,
+			},
+		},
+		{
+			name: "incSendBookmark", in: incSendBookmark,
+			exp: &DrySendInfo{
+				Type:         DrySendTypeIncremental,
+				Filesystem: "zroot/test/a",
+				From:         "zroot/test/a#1",
+				To:           "zroot/test/a@2",
+				SizeEstimate: 5383312,
+			},
+		},
+		{
+			name: "incNoToken", in: incNoToken,
+			exp: &DrySendInfo{
+				Type:         DrySendTypeIncremental,
+				Filesystem: "zroot/test/a",
+				// as can be seen in the string incNoToken,
+				// we cannot infer whether the incremental source is a snapshot or bookmark
+				From:         "1", // yes, this is actually correct on ZoL 0.7.11
+				To:           "zroot/test/a@2",
+				SizeEstimate: 10511856,
+			},
+		},
+		{
+			name: "fullNoToken", in: fullNoToken,
+			exp: &DrySendInfo{
+				Type: DrySendTypeFull,
+				Filesystem: "zroot/test/a",
+				From: "",
+				To: "zroot/test/a@3",
+				SizeEstimate: 10518512,
+			},
+		},
+	}
+
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+
+			in := tc.in[1:] // strip first newline
+			var si DrySendInfo
+			err := si.unmarshalZFSOutput([]byte(in))
+			if tc.expErr {
+				assert.Error(t, err)
+			}
+			t.Logf("%#v", &si)
+			if tc.exp != nil {
+				assert.Equal(t, tc.exp, &si)
+			}
+
+		})
+	}
+}