fix small typo

* Remove explicity state machine code for all but replication.Replication
* Introduce explicit error types that satisfy interfaces which provide
  sufficient information for replication.Replication to make intelligent
  retry + queuing decisions

  * Temporary()
  * LocalToFS()

* Remove the queue and replace it with a simple array that we sort each
  time (yay no generics :( )

Gopkg.lock

@@ -174,15 +174,15 @@
   revision = "391d2c78c8404a9683d79f75dd24ab53040f89f7"
 
 [[projects]]
-  digest = "1:c2ba1c9dc003c15856e4529dac028cacba08ee8924300f058b3467cde9acf7a9"
+  digest = "1:1bcbb0a7ad8d3392d446eb583ae5415ff987838a8f7331a36877789be20667e6"
   name = "github.com/problame/go-streamrpc"
   packages = [
     ".",
     "internal/pdu",
   ]
   pruneopts = ""
-  revision = "de6f6a4041c77f700f02d8fe749e54efa50811f7"
-  version = "v0.4"
+  revision = "d5d111e014342fe1c37f0b71cc37ec5f2afdfd13"
+  version = "v0.5"
 
 [[projects]]
   branch = "master"

Gopkg.toml

@@ -66,7 +66,7 @@ ignored = [ "github.com/inconshreveable/mousetrap" ]
 
 [[constraint]]
   name = "github.com/problame/go-streamrpc"
-  version = "0.4.0"
+  version = "0.5.0"
 
 
 

Makefile

@@ -22,10 +22,12 @@ SUBPKGS += pruning/retentiongrid
 SUBPKGS += replication
 SUBPKGS += replication/fsrep
 SUBPKGS += replication/pdu
-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 +35,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 +120,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 +131,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

@@ -22,6 +22,41 @@ import (
 	"time"
 )
 
+type bytesProgressHistory struct {
+	changeCounter     int
+	lastChangeAt      time.Time
+	last              int64
+	bpsIncreaseExpAvg float64
+}
+
+func (p *bytesProgressHistory) Update(currentVal int64) (bytesPerSecondAvg int64, changeCount int) {
+	if currentVal < p.last {
+		*p = bytesProgressHistory{
+			last: currentVal,
+			lastChangeAt: time.Now(),
+		}
+	}
+	defer func() {
+		p.last = currentVal
+	}()
+	if time.Now().Sub(p.lastChangeAt) > 3 *time.Second { // FIXME depends on refresh frequency
+		p.changeCounter = 0
+		p.bpsIncreaseExpAvg = 0
+	}
+	if currentVal != p.last {
+		p.changeCounter++
+		p.lastChangeAt = time.Now()
+	}
+
+	byteIncrease := float64(currentVal - p.last)
+	if byteIncrease < 0 {
+		byteIncrease = 0
+	}
+	const factor = 0.1
+	p.bpsIncreaseExpAvg = (1-factor) * p.bpsIncreaseExpAvg + factor *byteIncrease
+	return int64(p.bpsIncreaseExpAvg), p.changeCounter
+}
+
 type tui struct {
 	x, y   int
 	indent int
@@ -29,10 +64,14 @@ type tui struct {
 	lock   sync.Mutex //For report and error
 	report map[string]job.Status
 	err    error
+
+	replicationProgress map[string]*bytesProgressHistory // by job name
 }
 
 func newTui() tui {
-	return tui{}
+	return tui{
+		replicationProgress: make(map[string]*bytesProgressHistory, 0),
+	}
 }
 
 func (t *tui) moveCursor(x, y int) {
@@ -40,9 +79,11 @@ func (t *tui) moveCursor(x, y int) {
 	t.y += y
 }
 
+const INDENT_MULTIPLIER = 4
+
 func (t *tui) moveLine(dl int, col int) {
 	t.y += dl
-	t.x = t.indent*4 + col
+	t.x = t.indent*INDENT_MULTIPLIER + col
 }
 
 func (t *tui) write(text string) {
@@ -60,6 +101,40 @@ func (t *tui) printf(text string, a ...interface{}) {
 	t.write(fmt.Sprintf(text, a...))
 }
 
+func wrap(s string, width int) string {
+	var b strings.Builder
+	for len(s) > 0 {
+		rem := width
+		if rem > len(s) {
+			rem = len(s)
+		}
+		if idx := strings.IndexAny(s, "\n\r"); idx != -1 && idx < rem {
+			rem = idx+1
+		}
+		untilNewline := strings.TrimSpace(s[:rem])
+		s = s[rem:]
+		if len(untilNewline) == 0 {
+			continue
+		}
+		b.WriteString(untilNewline)
+		b.WriteString("\n")
+	}
+	return strings.TrimSpace(b.String())
+}
+
+func (t *tui) printfDrawIndentedAndWrappedIfMultiline(format string, a ...interface{}) {
+	whole := fmt.Sprintf(format, a...)
+	width, _ := termbox.Size()
+	if !strings.ContainsAny(whole, "\n\r") && t.x + len(whole) <= width {
+		t.printf(format, a...)
+	} else {
+		t.addIndent(1)
+		t.newline()
+		t.write(wrap(whole, width - INDENT_MULTIPLIER*t.indent))
+		t.addIndent(-1)
+	}
+}
+
 func (t *tui) newline() {
 	t.moveLine(1, 0)
 }
@@ -74,6 +149,7 @@ func (t *tui) addIndent(indent int) {
 	t.moveLine(0, 0)
 }
 
+
 var statusFlags struct {
 	Raw bool
 }
@@ -167,6 +243,15 @@ loop:
 
 }
 
+func (t *tui) getReplicationProgresHistory(jobName string) *bytesProgressHistory {
+	p, ok := t.replicationProgress[jobName]
+	if !ok {
+		p = &bytesProgressHistory{}
+		t.replicationProgress[jobName] = p
+	}
+	return p
+}
+
 func (t *tui) draw() {
 	t.lock.Lock()
 	defer t.lock.Unlock()
@@ -226,7 +311,7 @@ func (t *tui) draw() {
 			t.printf("Replication:")
 			t.newline()
 			t.addIndent(1)
-			t.renderReplicationReport(pushStatus.Replication)
+			t.renderReplicationReport(pushStatus.Replication, t.getReplicationProgresHistory(k))
 			t.addIndent(-1)
 
 			t.printf("Pruning Sender:")
@@ -246,7 +331,7 @@ func (t *tui) draw() {
 	termbox.Flush()
 }
 
-func (t *tui) renderReplicationReport(rep *replication.Report) {
+func (t *tui) renderReplicationReport(rep *replication.Report, history *bytesProgressHistory) {
 	if rep == nil {
 		t.printf("...\n")
 		return
@@ -271,11 +356,11 @@ func (t *tui) renderReplicationReport(rep *replication.Report) {
 	t.printf("Status: %s", state)
 	t.newline()
 	if rep.Problem != "" {
-		t.printf("Problem: %s", rep.Problem)
+		t.printf("Problem: ")
+		t.printfDrawIndentedAndWrappedIfMultiline("%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()))
 	}
 
@@ -298,9 +383,10 @@ func (t *tui) renderReplicationReport(rep *replication.Report) {
 			transferred += fstx
 			total += fstotal
 		}
+		rate, changeCount := history.Update(transferred)
 		t.write("Progress: ")
-		t.drawBar(80, transferred, total)
-		t.write(fmt.Sprintf(" %s / %s", ByteCountBinary(transferred), ByteCountBinary(total)))
+		t.drawBar(50, transferred, total, changeCount)
+		t.write(fmt.Sprintf(" %s / %s @ %s/s", ByteCountBinary(transferred), ByteCountBinary(total), ByteCountBinary(rate)))
 		t.newline()
 	}
 
@@ -311,7 +397,7 @@ func (t *tui) renderReplicationReport(rep *replication.Report) {
 		}
 	}
 	for _, fs := range all {
-		printFilesystemStatus(fs, t, fs == rep.Active, maxFSLen)
+		t.printFilesystemStatus(fs, fs == rep.Active, maxFSLen)
 	}
 }
 
@@ -392,8 +478,8 @@ func (t *tui) renderPrunerReport(r *pruner.Report) {
 	for _, fs := range all {
 		t.write(rightPad(fs.Filesystem, maxFSname, " "))
 		t.write(" ")
-		if fs.Error != "" {
-			t.printf("ERROR:    %s\n", fs.Error) // whitespace is padding
+		if fs.LastError != "" {
+			t.printf("ERROR (%d): %s\n", fs.ErrorCount, fs.LastError) // whitespace is padding
 			continue
 		}
 
@@ -401,11 +487,11 @@ func (t *tui) renderPrunerReport(r *pruner.Report) {
 			len(fs.DestroyList), len(fs.SnapshotList))
 
 		if fs.completed {
-			t.printf( "Completed %s\n", pruneRuleActionStr)
+			t.printf( "Completed  %s\n", pruneRuleActionStr)
 			continue
 		}
 
-		t.write("Pending   ") // whitespace is padding 10
+		t.write("Pending    ") // whitespace is padding 10
 		if len(fs.DestroyList) == 1 {
 			t.write(fs.DestroyList[0].Name)
 		} else {
@@ -457,7 +543,10 @@ func leftPad(str string, length int, pad string) string {
 	return times(pad, length-len(str)) + str
 }
 
-func (t *tui) drawBar(length int, bytes, totalBytes int64) {
+var arrowPositions = `>\|/`
+
+// changeCount = 0 indicates stall / no progresss
+func (t *tui) drawBar(length int, bytes, totalBytes int64, changeCount int) {
 	var completedLength int
 	if totalBytes > 0 {
 		completedLength = int(int64(length) * bytes / totalBytes)
@@ -470,7 +559,7 @@ func (t *tui) drawBar(length int, bytes, totalBytes int64) {
 
 	t.write("[")
 	t.write(times("=", completedLength))
-	t.write(">")
+	t.write( string(arrowPositions[changeCount%len(arrowPositions)]))
 	t.write(times("-", length-completedLength))
 	t.write("]")
 }
@@ -478,19 +567,17 @@ func (t *tui) drawBar(length int, bytes, totalBytes int64) {
 func StringStepState(s fsrep.StepState) string {
 	switch s {
 	case fsrep.StepReplicationReady: return "Ready"
-	case fsrep.StepReplicationRetry: return "Retry"
 	case fsrep.StepMarkReplicatedReady: return "MarkReady"
-	case fsrep.StepMarkReplicatedRetry: return "MarkRetry"
-	case fsrep.StepPermanentError: return "PermanentError"
 	case fsrep.StepCompleted: return "Completed"
 	default:
 		return fmt.Sprintf("UNKNOWN %d", s)
 	}
 }
 
-func filesystemStatusString(rep *fsrep.Report, active bool, fsWidth int) (line string, bytes, totalBytes int64) {
-	bytes = int64(0)
-	totalBytes = int64(0)
+func (t *tui) printFilesystemStatus(rep *fsrep.Report, active bool, maxFS int) {
+
+	bytes := int64(0)
+	totalBytes := int64(0)
 	for _, s := range rep.Pending {
 		bytes += s.Bytes
 		totalBytes += s.ExpectedBytes
@@ -500,36 +587,35 @@ func filesystemStatusString(rep *fsrep.Report, active bool, fsWidth int) (line s
 		totalBytes += s.ExpectedBytes
 	}
 
-	next := ""
-	if rep.Problem != "" {
-		next = " problem: " + rep.Problem
-	} else if len(rep.Pending) > 0 {
-		if rep.Pending[0].From != "" {
-			next = fmt.Sprintf(" next: %s => %s", rep.Pending[0].From, rep.Pending[0].To)
-		} else {
-			next = fmt.Sprintf(" next: %s (full)", rep.Pending[0].To)
-		}
-	}
-	status := fmt.Sprintf("%s (step %d/%d, %s/%s)%s",
+
+	status := fmt.Sprintf("%s (step %d/%d, %s/%s)",
 		rep.Status,
 		len(rep.Completed), len(rep.Pending) + len(rep.Completed),
 		ByteCountBinary(bytes), ByteCountBinary(totalBytes),
-		next,
+
 	)
+
 	activeIndicator := " "
 	if active {
 		activeIndicator = "*"
 	}
-	line = fmt.Sprintf("%s %s %s",
+	t.printf("%s %s %s ",
 		activeIndicator,
-		rightPad(rep.Filesystem, fsWidth, " "),
+		rightPad(rep.Filesystem, maxFS, " "),
 		status)
-	return line, bytes, totalBytes
-}
 
-func printFilesystemStatus(rep *fsrep.Report, t *tui, active bool, maxFS int) {
-	totalStatus, _, _ := filesystemStatusString(rep, active, maxFS)
-	t.write(totalStatus)
+	next := ""
+	if rep.Problem != "" {
+		next = rep.Problem
+	} else if len(rep.Pending) > 0 {
+		if rep.Pending[0].From != "" {
+			next = fmt.Sprintf("next: %s => %s", rep.Pending[0].From, rep.Pending[0].To)
+		} else {
+			next = fmt.Sprintf("next: %s (full)", rep.Pending[0].To)
+		}
+	}
+	t.printfDrawIndentedAndWrappedIfMultiline("%s", next)
+
 	t.newline()
 }
 

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

@@ -100,10 +100,11 @@ func (j *controlJob) Run(ctx context.Context) {
 		}}})
 
 	mux.Handle(ControlJobEndpointStatus,
-		requestLogger{log: log, handler: jsonResponder{func() (interface{}, error) {
+		// don't log requests to status endpoint, too spammy
+		jsonResponder{func() (interface{}, error) {
 			s := j.jobs.status()
 			return s, nil
-		}}})
+		}})
 
 	mux.Handle(ControlJobEndpointSignal,
 		requestLogger{log: log, handler: jsonRequestResponder{func(decoder jsonDecoder) (interface{}, error) {

daemon/job/active.go

@@ -6,17 +6,18 @@ import (
 	"github.com/problame/go-streamrpc"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon/filters"
 	"github.com/zrepl/zrepl/daemon/job/reset"
 	"github.com/zrepl/zrepl/daemon/job/wakeup"
-	"github.com/zrepl/zrepl/daemon/transport/connecter"
-	"github.com/zrepl/zrepl/daemon/filters"
+	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/daemon/pruner"
+	"github.com/zrepl/zrepl/daemon/snapper"
+	"github.com/zrepl/zrepl/daemon/transport/connecter"
 	"github.com/zrepl/zrepl/endpoint"
 	"github.com/zrepl/zrepl/replication"
+	"github.com/zrepl/zrepl/util/envconst"
 	"github.com/zrepl/zrepl/zfs"
 	"sync"
-	"github.com/zrepl/zrepl/daemon/logging"
-	"github.com/zrepl/zrepl/daemon/snapper"
 	"time"
 )
 
@@ -36,9 +37,30 @@ type ActiveSide struct {
 	tasks    activeSideTasks
 }
 
+
+//go:generate enumer -type=ActiveSideState
+type ActiveSideState int
+
+const (
+	ActiveSideReplicating ActiveSideState = 1 << iota
+	ActiveSidePruneSender
+	ActiveSidePruneReceiver
+	ActiveSideDone // also errors
+)
+
+
 type activeSideTasks struct {
+	state ActiveSideState
+
+	// valid for state ActiveSideReplicating, ActiveSidePruneSender, ActiveSidePruneReceiver, ActiveSideDone
 	replication *replication.Replication
+	replicationCancel context.CancelFunc
+
+	// valid for state ActiveSidePruneSender, ActiveSidePruneReceiver, ActiveSideDone
 	prunerSender, prunerReceiver *pruner.Pruner
+
+	// valid for state ActiveSidePruneReceiver, ActiveSideDone
+	prunerSenderCancel, prunerReceiverCancel context.CancelFunc
 }
 
 func (a *ActiveSide) updateTasks(u func(*activeSideTasks)) activeSideTasks {
@@ -238,11 +260,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)
@@ -250,11 +272,8 @@ func (j *ActiveSide) do(ctx context.Context, periodicWakeup <-chan struct{}) {
 	// allow cancellation of an invocation (this function)
 	ctx, cancelThisRun := context.WithCancel(ctx)
 	defer cancelThisRun()
-	runDone := make(chan struct{})
-	defer close(runDone)
 	go func() {
 		select {
-		case <-runDone:
 		case <-reset.Wait(ctx):
 			log.Info("reset received, cancelling current invocation")
 			cancelThisRun()
@@ -262,6 +281,78 @@ func (j *ActiveSide) do(ctx context.Context, periodicWakeup <-chan struct{}) {
 		}
 	}()
 
+	// The code after this watchdog goroutine is sequential and transitions the state from
+	//   ActiveSideReplicating -> ActiveSidePruneSender -> ActiveSidePruneReceiver -> ActiveSideDone
+	// If any of those sequential tasks 'gets stuck' (livelock, no progress), the watchdog will eventually
+	// cancel its context.
+	// If the task is written to support context cancellation, it will return immediately (in permanent error state),
+	// and the sequential code above transitions to the next state.
+	go func() {
+
+		wdto := envconst.Duration("ZREPL_JOB_WATCHDOG_TIMEOUT", 10*time.Minute)
+		jitter := envconst.Duration("ZREPL_JOB_WATCHDOG_JITTER", 1*time.Second)
+		// shadowing!
+		log := log.WithField("watchdog_timeout", wdto.String())
+
+		log.Debug("starting watchdog")
+		defer log.Debug("watchdog stopped")
+
+		t := time.NewTicker(wdto)
+		defer t.Stop()
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-t.C: // fall
+			}
+
+			j.updateTasks(func(tasks *activeSideTasks) {
+				// Since cancelling a task will cause the sequential code to transition to the next state immediately,
+				// we cannot check for its progress right then (no fallthrough).
+				// Instead, we return (not continue because we are in a closure) and give the new state another
+				// ZREPL_JOB_WATCHDOG_TIMEOUT interval to try make some progress.
+
+				log.WithField("state", tasks.state).Debug("watchdog firing")
+
+				const WATCHDOG_ENVCONST_NOTICE = " (adjust ZREPL_JOB_WATCHDOG_TIMEOUT env variable if inappropriate)"
+
+				switch tasks.state {
+				case ActiveSideReplicating:
+					log.WithField("replication_progress", tasks.replication.Progress.String()).
+						Debug("check replication progress")
+					if tasks.replication.Progress.CheckTimeout(wdto, jitter) {
+						log.Error("replication did not make progress, cancelling" + WATCHDOG_ENVCONST_NOTICE)
+						tasks.replicationCancel()
+						return
+					}
+				case ActiveSidePruneSender:
+					log.WithField("prune_sender_progress", tasks.replication.Progress.String()).
+						Debug("check pruner_sender progress")
+					if tasks.prunerSender.Progress.CheckTimeout(wdto, jitter) {
+						log.Error("pruner_sender did not make progress, cancelling" + WATCHDOG_ENVCONST_NOTICE)
+						tasks.prunerSenderCancel()
+						return
+					}
+				case ActiveSidePruneReceiver:
+					log.WithField("prune_receiver_progress", tasks.replication.Progress.String()).
+						Debug("check pruner_receiver progress")
+					if tasks.prunerReceiver.Progress.CheckTimeout(wdto, jitter) {
+						log.Error("pruner_receiver did not make progress, cancelling" + WATCHDOG_ENVCONST_NOTICE)
+						tasks.prunerReceiverCancel()
+						return
+					}
+				case ActiveSideDone:
+					// ignore, ctx will be Done() in a few milliseconds and the watchdog will exit
+				default:
+					log.WithField("state", tasks.state).
+						Error("watchdog implementation error: unknown active side state")
+				}
+			})
+
+		}
+	}()
+
 	client, err := j.clientFactory.NewClient()
 	if err != nil {
 		log.WithError(err).Error("factory cannot instantiate streamrpc client")
@@ -270,136 +361,62 @@ 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)
+			tasks.state = ActiveSideReplicating
+		})
+		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
+			tasks.state = ActiveSidePruneSender
 		})
+		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:
 		}
+		ctx, receiverCancel := context.WithCancel(ctx)
+		tasks := j.updateTasks(func(tasks *activeSideTasks) {
+			tasks.prunerReceiver = j.prunerFactory.BuildReceiverPruner(ctx, receiver, sender)
+			tasks.prunerReceiverCancel = receiverCancel
+			tasks.state = ActiveSidePruneReceiver
+		})
+		log.Info("start pruning receiver")
+		tasks.prunerReceiver.Prune()
+		log.Info("finished pruning receiver")
+		receiverCancel()
+	}
 
-	}()
-
-	pruningWg.Wait() // if pruners handle ctx cancellation correctly, we don't need to wait for it here
+	j.updateTasks(func(tasks *activeSideTasks) {
+		tasks.state = ActiveSideDone
+	})
 
 }

daemon/job/activesidestate_enumer.go

@@ -0,0 +1,66 @@
+// Code generated by "enumer -type=ActiveSideState"; DO NOT EDIT.
+
+package job
+
+import (
+	"fmt"
+)
+
+const (
+	_ActiveSideStateName_0 = "ActiveSideReplicatingActiveSidePruneSender"
+	_ActiveSideStateName_1 = "ActiveSidePruneReceiver"
+	_ActiveSideStateName_2 = "ActiveSideDone"
+)
+
+var (
+	_ActiveSideStateIndex_0 = [...]uint8{0, 21, 42}
+	_ActiveSideStateIndex_1 = [...]uint8{0, 23}
+	_ActiveSideStateIndex_2 = [...]uint8{0, 14}
+)
+
+func (i ActiveSideState) String() string {
+	switch {
+	case 1 <= i && i <= 2:
+		i -= 1
+		return _ActiveSideStateName_0[_ActiveSideStateIndex_0[i]:_ActiveSideStateIndex_0[i+1]]
+	case i == 4:
+		return _ActiveSideStateName_1
+	case i == 8:
+		return _ActiveSideStateName_2
+	default:
+		return fmt.Sprintf("ActiveSideState(%d)", i)
+	}
+}
+
+var _ActiveSideStateValues = []ActiveSideState{1, 2, 4, 8}
+
+var _ActiveSideStateNameToValueMap = map[string]ActiveSideState{
+	_ActiveSideStateName_0[0:21]:  1,
+	_ActiveSideStateName_0[21:42]: 2,
+	_ActiveSideStateName_1[0:23]:  4,
+	_ActiveSideStateName_2[0:14]:  8,
+}
+
+// ActiveSideStateString retrieves an enum value from the enum constants string name.
+// Throws an error if the param is not part of the enum.
+func ActiveSideStateString(s string) (ActiveSideState, error) {
+	if val, ok := _ActiveSideStateNameToValueMap[s]; ok {
+		return val, nil
+	}
+	return 0, fmt.Errorf("%s does not belong to ActiveSideState values", s)
+}
+
+// ActiveSideStateValues returns all values of the enum
+func ActiveSideStateValues() []ActiveSideState {
+	return _ActiveSideStateValues
+}
+
+// IsAActiveSideState returns "true" if the value is listed in the enum definition. "false" otherwise
+func (i ActiveSideState) IsAActiveSideState() bool {
+	for _, v := range _ActiveSideStateValues {
+		if i == v {
+			return true
+		}
+	}
+	return false
+}

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,8 +9,12 @@ 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"
+	"github.com/problame/go-streamrpc"
 	"net"
 	"sort"
+	"strings"
 	"sync"
 	"time"
 )
@@ -56,6 +60,8 @@ type args struct {
 type Pruner struct {
 	args args
 
+	Progress watchdog.KeepAlive
+
 	mtx sync.RWMutex
 
 	state State
@@ -65,8 +71,7 @@ type Pruner struct {
 	err        error
 
 	// State Exec
-	prunePending   []*fs
-	pruneCompleted []*fs
+	execQueue *execQueue
 }
 
 type PrunerFactory struct {
@@ -110,11 +115,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", 10 * time.Second),
+		considerSnapAtCursorReplicated: considerSnapAtCursorReplicated,
+		promPruneSecs: promPruneSecs,
 	}
 	return f, nil
 }
@@ -175,6 +180,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 +205,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")
+		}
 	}
 }
 
@@ -209,7 +224,8 @@ type Report struct {
 type FSReport struct {
 	Filesystem string
 	SnapshotList, DestroyList []SnapshotReport
-	Error string
+	ErrorCount int
+	LastError string
 }
 
 type SnapshotReport struct {
@@ -224,31 +240,37 @@ func (p *Pruner) Report() *Report {
 
 	r := Report{State: p.state.String()}
 
-	if p.state & PlanWait|ExecWait != 0 {
+	if p.state & (PlanWait|ExecWait) != 0 {
 		r.SleepUntil = p.sleepUntil
 	}
-	if p.state & PlanWait|ExecWait|ErrPerm != 0 {
+	if p.state & (PlanWait|ExecWait|ErrPerm) != 0 {
 		if p.err != nil {
 			r.Error = p.err.Error()
 		}
 	}
 
-	if p.state & Plan|PlanWait == 0 {
-		return &r
-	}
-
-	r.Pending = make([]FSReport, len(p.prunePending))
-	for i, fs := range p.prunePending{
-		r.Pending[i] = fs.Report()
-	}
-	r.Completed = make([]FSReport, len(p.pruneCompleted))
-	for i, fs := range p.pruneCompleted{
-		r.Completed[i] = fs.Report()
+	if p.execQueue != nil {
+		r.Pending, r.Completed = p.execQueue.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
 
@@ -260,14 +282,11 @@ type fs struct {
 	destroyList []pruning.Snapshot
 
 	mtx sync.RWMutex
-	// for Plan
-	err error
-}
 
-func (f *fs) Update(err error) {
-	f.mtx.Lock()
-	defer f.mtx.Unlock()
-	f.err = err
+	// only during Exec state, also used by execQueue
+	execErrLast error
+	execErrCount int
+
 }
 
 func (f *fs) Report() FSReport {
@@ -276,8 +295,9 @@ func (f *fs) Report() FSReport {
 
 	r := FSReport{}
 	r.Filesystem = f.path
-	if f.err != nil {
-		r.Error = f.err.Error()
+	r.ErrorCount = f.execErrCount
+	if f.execErrLast != nil {
+		r.LastError = f.execErrLast.Error()
 	}
 
 	r.SnapshotList = make([]SnapshotReport, len(f.snaps))
@@ -315,12 +335,17 @@ func (s snapshot) Replicated() bool { return s.replicated }
 
 func (s snapshot) Date() time.Time { return s.date }
 
+type Error interface {
+	error
+	Temporary() bool
+}
+
+var _ Error = net.Error(nil)
+var _ Error = streamrpc.Error(nil)
+
 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
 }
@@ -346,6 +371,10 @@ func onErr(u updater, e error) state {
 func statePlan(a *args, u updater) state {
 
 	ctx, target, receiver := a.ctx, a.target, a.receiver
+	var ka *watchdog.KeepAlive
+	u(func(pruner *Pruner) {
+		ka = &pruner.Progress
+	})
 
 	tfss, err := target.ListFilesystems(ctx)
 	if err != nil {
@@ -353,7 +382,6 @@ func statePlan(a *args, u updater) state {
 	}
 
 	pfss := make([]*fs, len(tfss))
-fsloop:
 	for i, tfs := range tfss {
 
 		l := GetLogger(ctx).WithField("fs", tfs.Path)
@@ -368,12 +396,10 @@ fsloop:
 		tfsvs, err := target.ListFilesystemVersions(ctx, tfs.Path)
 		if err != nil {
 			l.WithError(err).Error("cannot list filesystem versions")
-			if shouldRetry(err) {
-				return onErr(u, err)
-			}
-			pfs.err = err
-			continue fsloop
+			return onErr(u, err)
 		}
+		// no progress here since we could run in a live-lock (must have used target AND receiver before progress)
+
 		pfs.snaps = make([]pruning.Snapshot, 0, len(tfsvs))
 
 		rcReq := &pdu.ReplicationCursorReq{
@@ -385,17 +411,13 @@ fsloop:
 		rc, err := receiver.ReplicationCursor(ctx, rcReq)
 		if err != nil {
 			l.WithError(err).Error("cannot get replication cursor")
-			if shouldRetry(err) {
-				return onErr(u, err)
-			}
-			pfs.err = err
-			continue fsloop
+			return onErr(u, err)
 		}
 		if rc.GetError() != "" {
 			l.WithField("reqErr", rc.GetError()).Error("cannot get replication cursor")
-			pfs.err = fmt.Errorf("%s", rc.GetError())
-			continue fsloop
+			return onErr(u, err)
 		}
+		ka.MadeProgress()
 
 
 		// scan from older to newer, all snapshots older than cursor are interpreted as replicated
@@ -419,9 +441,11 @@ fsloop:
 			}
 			creation, err := tfsv.CreationAsTime()
 			if err != nil {
-				pfs.err = fmt.Errorf("%s%s has invalid creation date: %s", tfs, tfsv.RelName(), err)
-				l.WithError(pfs.err).Error("")
-				continue fsloop
+				err := fmt.Errorf("%s%s has invalid creation date: %s", tfs, tfsv.RelName(), err)
+				l.WithError(err).
+					WithField("tfsv", tfsv.RelName()).
+					Error("error with fileesystem version")
+				return onErr(u, err)
 			}
 			// note that we cannot use CreateTXG because target and receiver could be on different pools
 			atCursor := tfsv.Guid == rc.GetGuid()
@@ -433,23 +457,21 @@ fsloop:
 			})
 		}
 		if preCursor {
-			pfs.err = fmt.Errorf("replication cursor not found in prune target filesystem versions")
-			l.WithError(pfs.err).Error("")
-			continue fsloop
+			err := fmt.Errorf("replication cursor not found in prune target filesystem versions")
+			l.Error(err.Error())
+			return onErr(u, err)
 		}
 
 		// Apply prune rules
 		pfs.destroyList = pruning.PruneSnapshots(pfs.snaps, a.rules)
-
+		ka.MadeProgress()
 	}
 
 	return u(func(pruner *Pruner) {
+		pruner.Progress.MadeProgress()
+		pruner.execQueue = newExecQueue(len(pfss))
 		for _, pfs := range pfss {
-			if pfs.err != nil {
-				pruner.pruneCompleted = append(pruner.pruneCompleted, pfs)
-			} else {
-				pruner.prunePending = append(pruner.prunePending, pfs)
-			}
+			pruner.execQueue.Put(pfs, nil, false)
 		}
 		pruner.state = Exec
 	}).statefunc()
@@ -459,17 +481,15 @@ func stateExec(a *args, u updater) state {
 
 	var pfs *fs
 	state := u(func(pruner *Pruner) {
-		if len(pruner.prunePending) == 0 {
+		pfs = pruner.execQueue.Pop()
+		if pfs == nil {
 			nextState := Done
-			for _, pfs := range pruner.pruneCompleted {
-				if pfs.err != nil {
-					nextState = ErrPerm
-				}
+			if pruner.execQueue.HasCompletedFSWithErrors() {
+				nextState = ErrPerm
 			}
 			pruner.state = nextState
 			return
 		}
-		pfs = pruner.prunePending[0]
 	})
 	if state != Exec {
 		return state.statefunc()
@@ -483,21 +503,62 @@ func stateExec(a *args, u updater) state {
 			WithField("destroy_snap", destroyList[i].Name).
 			Debug("policy destroys snapshot")
 	}
-	pfs.Update(nil)
 	req := pdu.DestroySnapshotsReq{
 		Filesystem: pfs.path,
 		Snapshots:  destroyList,
 	}
-	_, err := a.target.DestroySnapshots(a.ctx, &req)
-	pfs.Update(err)
-	if err != nil && shouldRetry(err) {
+	GetLogger(a.ctx).WithField("fs", pfs.path).Debug("destroying snapshots")
+	res, err := a.target.DestroySnapshots(a.ctx, &req)
+	if err != nil {
+		u(func(pruner *Pruner) {
+			pruner.execQueue.Put(pfs, err, false)
+		})
+		return onErr(u, err)
+	}
+	// check if all snapshots were destroyed
+	destroyResults := make(map[string]*pdu.DestroySnapshotRes)
+	for _, fsres := range res.Results {
+		destroyResults[fsres.Snapshot.Name] = fsres
+	}
+	err = nil
+	destroyFails := make([]*pdu.DestroySnapshotRes, 0)
+	for _, reqDestroy := range destroyList {
+		 res, ok := destroyResults[reqDestroy.Name]
+		 if !ok {
+		 	err = fmt.Errorf("missing destroy-result for %s", reqDestroy.RelName())
+		 	break
+		 } else if res.Error != "" {
+		 	destroyFails = append(destroyFails, res)
+		 }
+	}
+	if err == nil && len(destroyFails) > 0 {
+		names := make([]string, len(destroyFails))
+		pairs := make([]string, len(destroyFails))
+		allSame := true
+		lastMsg := destroyFails[0].Error
+		for i := 0; i < len(destroyFails); i++{
+			allSame = allSame && destroyFails[i].Error == lastMsg
+			relname := destroyFails[i].Snapshot.RelName()
+			names[i] = relname
+			pairs[i] = fmt.Sprintf("(%s: %s)", relname, destroyFails[i].Error)
+		}
+		if allSame {
+			err = fmt.Errorf("destroys failed %s: %s",
+				strings.Join(names, ", "), lastMsg)
+		} else {
+			err = fmt.Errorf("destroys failed: %s", strings.Join(pairs, ", "))
+		}
+	}
+	u(func(pruner *Pruner) {
+		pruner.execQueue.Put(pfs, err, err == nil)
+	})
+	if err != nil {
+		GetLogger(a.ctx).WithError(err).Error("target could not destroy snapshots")
 		return onErr(u, err)
 	}
-	// if it's not retryable, treat is like as being done
 
 	return u(func(pruner *Pruner) {
-		pruner.pruneCompleted = append(pruner.pruneCompleted, pfs)
-		pruner.prunePending = pruner.prunePending[1:]
+		pruner.Progress.MadeProgress()
 	}).statefunc()
 }
 

daemon/pruner/pruner_queue.go

@@ -0,0 +1,89 @@
+package pruner
+
+import (
+	"sort"
+	"strings"
+	"sync"
+)
+
+type execQueue struct {
+	mtx sync.Mutex
+	pending, completed []*fs
+}
+
+func newExecQueue(cap int) *execQueue {
+	q := execQueue{
+		pending: make([]*fs, 0, cap),
+		completed: make([]*fs, 0, cap),
+	}
+	return &q
+}
+
+func (q *execQueue) Report() (pending, completed []FSReport) {
+	q.mtx.Lock()
+	defer q.mtx.Unlock()
+
+	pending = make([]FSReport, len(q.pending))
+	for i, fs := range q.pending {
+		pending[i] = fs.Report()
+	}
+	completed = make([]FSReport, len(q.completed))
+	for i, fs := range q.completed {
+		completed[i] = fs.Report()
+	}
+
+	return pending, completed
+}
+
+func (q *execQueue) HasCompletedFSWithErrors() bool {
+	q.mtx.Lock()
+	defer q.mtx.Unlock()
+	for _, fs := range q.completed {
+		if fs.execErrLast != nil {
+			return true
+		}
+	}
+	return false
+}
+
+func (q *execQueue) Pop() *fs {
+	if len(q.pending) == 0 {
+		return nil
+	}
+	fs := q.pending[0]
+	q.pending = q.pending[1:]
+	return fs
+}
+
+func(q *execQueue) Put(fs *fs, err error, done bool) {
+	fs.mtx.Lock()
+	fs.execErrLast = err
+	if err != nil {
+		fs.execErrCount++
+	}
+	if done || (err != nil && !shouldRetry(fs.execErrLast)) {
+		fs.mtx.Unlock()
+		q.mtx.Lock()
+		q.completed = append(q.completed, fs)
+		q.mtx.Unlock()
+		return
+	}
+	fs.mtx.Unlock()
+
+	q.mtx.Lock()
+	// inefficient priority q
+	q.pending = append(q.pending, fs)
+	sort.SliceStable(q.pending, func(i, j int) bool {
+		q.pending[i].mtx.Lock()
+		defer q.pending[i].mtx.Unlock()
+		q.pending[j].mtx.Lock()
+		defer q.pending[j].mtx.Unlock()
+		if q.pending[i].execErrCount != q.pending[j].execErrCount {
+			return q.pending[i].execErrCount < q.pending[j].execErrCount
+		}
+		return strings.Compare(q.pending[i].path, q.pending[j].path) == -1
+	})
+	q.mtx.Unlock()
+
+
+}

daemon/pruner/pruner_test.go

@@ -129,20 +129,18 @@ 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},
+				stubNetErr{msg: "fakeerror2", temporary: true,},
 			},
 		},
 		destroyErrs: map[string][]error{
-			"zroot/foo": {
-				fmt.Errorf("permanent error"),
-			},
-			"zroot/bar": {
-				stubNetErr{msg: "fakeerror3"},
+			"zroot/baz": {
+				stubNetErr{msg: "fakeerror3", temporary: true}, // first error puts it back in the queue
+				stubNetErr{msg:"permanent error"}, // so it will be last when pruner gives up due to permanent err
 			},
 		},
 		destroyed: make(map[string][]string),
@@ -176,10 +174,7 @@ func TestPruner_Prune(t *testing.T) {
 	history := &mockHistory{
 		errs: map[string][]error{
 			"zroot/foo": {
-				stubNetErr{msg: "fakeerror4"},
-			},
-			"zroot/baz": {
-				fmt.Errorf("permanent error2"),
+				stubNetErr{msg: "fakeerror4", temporary: true},
 			},
 		},
 	}
@@ -199,9 +194,8 @@ func TestPruner_Prune(t *testing.T) {
 	p.Prune()
 
 	exp := map[string][]string{
+		"zroot/foo": {"drop_c"},
 		"zroot/bar": {"drop_g"},
-		// drop_c is prohibited by failing destroy
-		// drop_i is prohibiteed by failing ReplicationCursor call
 	}
 
 	assert.Equal(t, exp, target.destroyed)

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,18 +29,28 @@ 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
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 * If the daemon crashes, the stack trace produced by the Go runtime and possibly diagnostic output of zrepl will be written to stderr.
   This behavior is independent from the ``stdout`` outlet type.
-  Please make sure the stderr output of the daemon is captured to a file.
-  Rotation should not be necessary because stderr is not written to under normal circumstances.
+  Please make sure the stderr output of the daemon is captured somewhere.
   To conserve precious stack traces, make sure that multiple service restarts do not directly discard previous stderr output.
 * Make it obvious for users how to set the ``GOTRACEBACK`` environment variable to ``GOTRACEBACK=crash``.
   This functionality will cause SIGABRT on panics and can be used to capture a coredump of the panicking process.
   To that extend, make sure that your package build system, your OS's coredump collection and the Go delve debugger work together.
   Use your build system to package the Go program in `this tutorial on Go coredumps and the delve debugger <https://rakyll.org/coredumps/>`_ , and make sure the symbol resolution etc. work on coredumps captured from the binary produced by your build system. (Special focus on symbol stripping, etc.)
+* Use of ``ssh+stdinserver`` :ref:`transport <transport-ssh+stdinserver>` is no longer encouraged.
+  Please encourage users to use the new ``tcp`` or ``tls`` transports.
+  You might as well mention some of the :ref:`tunneling options listed here <transport-tcp-tunneling>`.
 
 Changes
 ~~~~~~~
@@ -48,10 +58,10 @@ Changes
 * |feature| :issue:`55` : Push replication (see :ref:`push job <job-push>` and :ref:`sink job <job-sink>`)
 * |feature| :ref:`TCP Transport <transport-tcp>`
 * |feature| :ref:`TCP + TLS client authentication transport <transport-tcp+tlsclientauth>`
-* |feature| :issue:`78` TODO MERGE COMMIT Replication protocol rewrite
+* |feature| :issue:`78` :commit:`074f989` : Replication protocol rewrite
 
   * Uses ``github.com/problame/go-streamrpc`` for RPC layer
-  * |break| zrepl 0.1 and restart on both sides of a replication setup is required
+  * |break| Protocol breakage, update and restart of all zrepl daemons is required
   * |feature| :issue:`83`:  Improved error handling of network-level errors (zrepl retries instead of failing the entire job)
   * |bugfix| :issue:`75` :issue:`81`: use connection timeouts and protocol-level heartbeats
   * |break| |break_config|: mappings are no longer supported
@@ -63,7 +73,8 @@ Changes
 
   * |feature| :issue:`69`: include manually created snapshots in replication
   * |break_config| ``manual`` and ``periodic`` :ref:`snapshotting types <job-snapshotting-spec>`
-  * |feature| ``zrepl wakeup JOB`` subcommand to trigger *just* replication
+  * |feature| ``zrepl signal wakeup JOB`` subcommand to trigger replication + pruning
+  * |feature| ``zrepl signal reset JOB`` subcommand to abort current replication + pruning
 
 * |feature| |break| |break_config| New pruning system
 
@@ -77,7 +88,7 @@ Changes
   * |feature| |break| Bookmark pruning is no longer necessary
 
     * Per filesystem, zrepl creates a single bookmark (``#zrepl_replication_cursor``) and moves it forward with the most recent successfully replicated snapshot on the receiving side.
-    * Old bookmarks created prior to zrepl 0.1 (named like their corresponding snapshot) must be deleted manually.
+    * Old bookmarks created by prior versions of zrepl (named like their corresponding snapshot) must be deleted manually.
     * |break_config| ``keep_bookmarks`` parameter of the ``grid`` keep rule has been removed
 
 * |feature| ``zrepl status`` for live-updating replication progress (it's really cool!)

docs/configuration/jobs.rst

@@ -140,7 +140,7 @@ There is also a ``manual`` snapshotting type, which covers the following use cas
 * Run scripts before and after taking snapshots (like locking database tables).
   We are working on better integration for this use case: see :issue:`74`.
 
-Note that you will have to trigger replication manually using the ``zrepl wakeup JOB`` subcommand in that case.
+Note that you will have to trigger replication manually using the ``zrepl signal wakeup JOB`` subcommand in that case.
 
 ::
 

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::
 
@@ -97,7 +100,7 @@ the left edge of the leftmost (first) interval is the ``creation`` date of the y
 All intervals to its right describe time intervals further in the past.
 
 Each interval carries a maximum number of snapshots to keep.
-It is secified via ``(keep=N)``, where ``N`` is either ``all`` (all snapshots are kept) or a positive integer.
+It is specified via ``(keep=N)``, where ``N`` is either ``all`` (all snapshots are kept) or a positive integer.
 The default value is **keep=1**.
 
 The following procedure happens during pruning:

docs/configuration/transports.rst

@@ -10,6 +10,8 @@ On the passive (serving) side, the transport also provides the **client identity
 this string is used for access control and separation of filesystem sub-trees in :ref:`sink jobs <job-sink>`.
 Transports are specified in the ``connect`` or ``serve`` section of a job definition.
 
+.. contents::
+
 .. ATTENTION::
 
     The **client identities must be valid ZFS dataset path components**
@@ -27,6 +29,8 @@ This transport may also be used in conjunction with network-layer encryption and
 To make the IP-based client authentication effective, such solutions should provide authenticated IP addresses.
 Some options to consider:
 
+.. _transport-tcp-tunneling:
+
 * `WireGuard <https://www.wireguard.com/>`_: Linux-focussed, in-kernel TLS
 * `OpenVPN <https://openvpn.net/>`_: Cross-platform VPN, uses tun on \*nix
 * `IPSec <https://en.wikipedia.org/wiki/IPsec>`_: Properly standardized, in-kernel network-layer VPN
@@ -73,6 +77,7 @@ Connect
 The ``tls`` transport uses TCP + TLS with client authentication using client certificates.
 The client identity is the common name (CN) presented in the client certificate.
 It is recommended to set up a dedicated CA infrastructure for this transport, e.g. using OpenVPN's `EasyRSA <https://github.com/OpenVPN/easy-rsa>`_.
+For a simple 2-machine setup, see the :ref:`instructions below<transport-tcp+tlsclientauth-2machineopenssl>`.
 
 The implementation uses `Go's TLS library <https://golang.org/pkg/crypto/tls/>`_.
 Since Go binaries are statically linked, you or your distribution need to recompile zrepl when vulnerabilities in that library are disclosed.
@@ -122,6 +127,75 @@ The ``server_cn`` specifies the expected common name (CN) of the server's certif
 It overrides the hostname specified in ``address``.
 The connection fails if either do not match.
 
+.. _transport-tcp+tlsclientauth-2machineopenssl:
+
+Self-Signed Certificates
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+Tools like `EasyRSA <https://github.com/OpenVPN/easy-rsa>`_ make it easy to manage CA infrastructure for multiple clients, e.g. a central zrepl backup server (in sink mode).
+However, for a two-machine setup, self-signed certificates distributed using an out-of-band mechanism will also work just fine:
+
+Suppose you have a push-mode setup, with `backups.example.com` running the :ref:`sink job <job-sink>`, and `prod.example.com` running the :ref:`push job <job-push>`.
+Run the following OpenSSL commands on each host, substituting HOSTNAME in both filenames and the interactive input prompt by OpenSSL:
+
+.. code-block:: bash
+   :emphasize-lines: 1-5,24
+
+   openssl req -x509 -sha256 -nodes \
+      -newkey rsa:4096 \
+      -days 365 \
+      -keyout HOSTNAME.key \
+      -out HOSTNAME.crt
+
+   #Generating a 4096 bit RSA private key
+   #................++++
+   #.++++
+   #writing new private key to 'backups.key'
+   #-----
+   #You are about to be asked to enter information that will be incorporated
+   #into your certificate request.
+   #What you are about to enter is what is called a Distinguished Name or a DN.
+   #There are quite a few fields but you can leave some blank
+   #For some fields there will be a default value,
+   #If you enter '.', the field will be left blank.
+   #-----
+   #Country Name (2 letter code) [XX]:
+   #State or Province Name (full name) []:
+   #Locality Name (eg, city) [Default City]:
+   #Organization Name (eg, company) [Default Company Ltd]:
+   #Organizational Unit Name (eg, section) []:
+   #Common Name (eg, your name or your server's hostname) []:HOSTNAME
+   #Email Address []:
+
+Now copy each machine's ``HOSTNAME.crt`` to the other machine's ``/etc/zrepl/HOSTNAME.crt``, for example using `scp`.
+The serve & connect configuration will thus look like the following:
+
+::
+
+   # on backups.example.com
+   - type: sink
+     serve:
+       type: tls
+       listen: ":8888"
+       ca: "/etc/zrepl/prod.example.com.crt"
+       cert: "/etc/zrepl/backups.example.com.crt"
+       key: "/etc/zrepl/backups.example.com.key"
+       client_cns:
+         - "prod.example.com"
+     ...
+
+   # on prod.example.com
+   - type: push
+     connect:
+       type: tls
+       address:"backups.example.com:8888"
+       ca: /etc/zrepl/backups.example.com.crt
+       cert: /etc/zrepl/prod.example.com.crt
+       key:  /etc/zrepl/prod.example.com.key
+       server_cn: "backups.example.com"
+     ...
+
+
 .. _transport-ssh+stdinserver:
 
 ``ssh+stdinserver`` Transport
@@ -130,6 +204,11 @@ The connection fails if either do not match.
 ``ssh+stdinserver`` is inspired by `git shell <https://git-scm.com/docs/git-shell>`_ and `Borg Backup <https://borgbackup.readthedocs.io/en/stable/deployment.html>`_.
 It is provided by the Go package ``github.com/problame/go-netssh``.
 
+.. ATTENTION::
+
+   ``ssh+stdinserver`` has inferior error detection and handling compared to the ``tcp`` and ``tls`` transports.
+   If you require tested timeout & retry handling, use ``tcp`` or ``tls`` transports, or help improve package go-netssh.
+
 .. _transport-ssh+stdinserver-serve:
 
 Serve

docs/tutorial.rst

@@ -6,7 +6,7 @@ Tutorial
 ========
 
 
-This tutorial shows how zrepl can be used to implement a ZFS-based pull backup.
+This tutorial shows how zrepl can be used to implement a ZFS-based push backup.
 We assume the following scenario:
 
 * Production server ``prod`` with filesystems to back up:
@@ -17,12 +17,12 @@ We assume the following scenario:
 
 * Backup server ``backups`` with
 
-  * Filesystem ``storage/zrepl/pull/prod`` + children dedicated to backups of ``prod``
+  * Filesystem ``storage/zrepl/sink/prod`` + children dedicated to backups of ``prod``
 
 Our backup solution should fulfill the following requirements:
 
 * Periodically snapshot the filesystems on ``prod`` *every 10 minutes*
-* Incrementally replicate these snapshots to ``storage/zrepl/pull/prod/*`` on ``backups``
+* Incrementally replicate these snapshots to ``storage/zrepl/sink/prod/*`` on ``backups``
 * Keep only very few snapshots on ``prod`` to save disk space
 * Keep a fading history (24 hourly, 30 daily, 6 monthly) of snapshots on ``backups``
 
@@ -31,74 +31,70 @@ Analysis
 
 We can model this situation as two jobs:
 
-* A **source job** on ``prod``
+* A **push job** on ``prod``
 
   * Creates the snapshots
-  * Keeps a short history of snapshots to enable incremental replication to ``backups``
-  * Accepts connections from ``backups``
+  * Keeps a short history of local snapshots to enable incremental replication to ``backups``
+  * Connects to the ``zrepl daemon`` process on ``backups``
+  * Pushes snapshots ``backups``
+  * Prunes snapshots on ``backups`` after replication is complete
 
-* A **pull job** on ``backups``
+* A **sink job** on ``backups``
 
-  * Connects to the ``zrepl daemon`` process on ``prod``
-  * Pulls the snapshots to ``storage/zrepl/pull/prod/*``
-  * Fades out snapshots in ``storage/zrepl/pull/prod/*`` as they age
-
-
-Why doesn't the **pull job** create the snapshots before pulling?
-
-As is the case with all distributed systems, the link between ``prod`` and ``backups`` might be down for an hour or two.
-We do not want to sacrifice our required backup resolution of 10 minute intervals for a temporary connection outage.
-
-When the link comes up again, ``backups`` will catch up with the snapshots taken by ``prod`` in the meantime, without a gap in our backup history.
+  * Accepts connections & responds to requests from ``prod``
+  * Limits client ``prod`` access to filesystem sub-tree ``storage/zrepl/sink/prod``
 
 Install zrepl
 -------------
 
 Follow the :ref:`OS-specific installation instructions <installation>` and come back here.
 
-Configure server ``backups``
-----------------------------
-
-We define a **pull job** named ``pull_prod`` in ``/etc/zrepl/zrepl.yml`` or ``/usr/local/etc/zrepl/zrepl.yml`` on host ``backups`` : ::
-
-    jobs:
-    - name: pull_prod
-      type: pull
-      connect:
-        type: tcp
-        address: "192.168.2.20:2342"
-      root_fs: "storage/zrepl/pull/prod"
-      interval: 10m
-      pruning:
-        keep_sender:
-        - type: not_replicated
-        - type: last_n
-          count: 10
-        keep_receiver:
-        - type: grid
-          grid: 1x1h(keep=all) | 24x1h | 30x1d | 6x30d
-          regex: "^zrepl_"
-          interval: 10m
-
-The ``connect`` section instructs the zrepl daemon to use plain TCP transport.
-Check out the :ref:`transports <transport>` section for alternatives that support encryption.
-
-.. _tutorial-configure-prod:
-
-Configure server ``prod``
+Generate TLS Certificates
 -------------------------
 
-We define a corresponding **source job** named ``source_backups`` in ``/etc/zrepl/zrepl.yml`` or ``/usr/local/etc/zrepl/zrepl.yml`` on host ``prod`` : ::
+We use the `TLS client authentication transport <transport-tcp+tlsclientauth>` to protect our data on the wire.
+To get things going quickly, we skip setting up a CA and generate two self-signed certificates as described :ref:`here <transport-tcp+tlsclientauth-2machineopenssl>`.
+Again, for convenience, We generate the key pairs on our local machine and distribute them using ssh:
+
+.. code-block:: bash
+   :emphasize-lines: 6,13
+
+   openssl req -x509 -sha256 -nodes \
+      -newkey rsa:4096 \
+      -days 365 \
+      -keyout backups.key \
+      -out backups.crt
+   # ... and use "backups" as Common Name (CN)
+
+   openssl req -x509 -sha256 -nodes \
+      -newkey rsa:4096 \
+      -days 365 \
+      -keyout prod.key \
+      -out prod.crt
+   # ... and use "prod" as Common Name (CN)
+
+   ssh root@backups "mkdir /etc/zrepl"
+   scp  backups.key backups.crt prod.crt root@backups:/etc/zrepl
+
+   ssh root@prod "mkdir /etc/zrepl"
+   scp  prod.key prod.crt backups.crt root@prod:/etc/zrepl
+
+
+Configure server ``prod``
+----------------------------
+
+We define a **push job** named ``prod_to_backups`` in ``/etc/zrepl/zrepl.yml`` on host ``prod`` : ::
 
     jobs:
-    - name: source_backups
-      type: source
-      serve:
-        type: tcp
-        listen: ":2342"
-        clients: {
-          "192.168.2.10" : "backups"
-        }
+    - name: prod_to_backups
+      type: push 
+      connect:
+        type: tls
+        address: "backups.example.com:8888"
+        ca: /etc/zrepl/backups.crt
+        cert: /etc/zrepl/prod.crt
+        key:  /etc/zrepl/prod.key
+        server_cn: "backups"
       filesystems: {
         "zroot/var/db:": true,
         "zroot/usr/home<": true,
@@ -108,39 +104,69 @@ We define a corresponding **source job** named ``source_backups`` in ``/etc/zrep
         type: periodic
         prefix: zrepl_
         interval: 10m
+      pruning:
+        keep_sender:
+        - type: not_replicated
+        - type: last_n
+          count: 10
+        keep_receiver:
+        - type: grid
+          grid: 1x1h(keep=all) | 24x1h | 30x1d | 6x30d
+          regex: "^zrepl_"
 
+.. _tutorial-configure-prod:
+
+Configure server ``prod``
+-------------------------
+
+We define a corresponding **sink job** named ``sink`` in ``/etc/zrepl/zrepl.yml`` on host ``prod`` : ::
+
+    jobs:
+    - name: sink
+      type: sink 
+      serve:
+         type: tls
+         listen: ":8888"
+         ca: "/etc/zrepl/prod.crt"
+         cert: "/etc/zrepl/backups.crt"
+         key: "/etc/zrepl/backups.key"
+         client_cns:
+           - "prod"
+      root_fs: "storage/zrepl/sink"
 
-The ``serve`` section whitelists ``backups``'s IP address ``192.168.2.10`` and assigns it the client identity ``backups`` which will show up in the logs.
-Again, check the :ref:`docs for encrypted transports <transport>`.
 
 Apply Configuration Changes
 ---------------------------
 
-We need to restart the zrepl daemon on **both** ``prod`` and ``backups``.
-This is :ref:`OS-specific <usage-zrepl-daemon-restarting>`.
+We use ``zrepl configcheck`` before to catch any configuration errors: no output indicates that everything is fine.
+If that is the case, restart the zrepl daemon on **both** ``prod`` and ``backups`` using ``service zrepl restart`` or ``systemctl restart zrepl``.
+
 
 Watch it Work
 -------------
 
 Run ``zrepl status`` on ``prod`` to monitor the replication and pruning activity.
-
-Additionally, you can check the detailed structured logs of the `zrepl daemon` process and use GNU *watch* to view the snapshots present on both machines.
+To re-trigger replication (snapshots are separate!), use ``zrepl signal wakeup prod_to_backups`` on ``prod``.
 
 If you like tmux, here is a handy script that works on FreeBSD: ::
 
     pkg install gnu-watch tmux
-    tmux new-window
-    tmux split-window "tail -f /var/log/zrepl.log"
-    tmux split-window "gnu-watch 'zfs list -t snapshot -o name,creation -s creation | grep zrepl_'"
-    tmux select-layout tiled
+    tmux new -s zrepl -d
+    tmux split-window -t zrepl "tail -f /var/log/messages"
+    tmux split-window -t zrepl "gnu-watch 'zfs list -t snapshot -o name,creation -s creation | grep zrepl_'"
+    tmux split-window -t zrepl "zrepl status"
+    tmux select-layout -t zrepl tiled
+    tmux attach -t zrepl
 
 The Linux equivalent might look like this: ::
 
     # make sure tmux is installed & let's assume you use systemd + journald
-    tmux new-window
-    tmux split-window "journalctl -f -u zrepl.service"
-    tmux split-window "watch 'zfs list -t snapshot -o name,creation -s creation | grep zrepl_'"
-    tmux select-layout tiled
+    tmux new -s zrepl -d
+    tmux split-window -t zrepl  "journalctl -f -u zrepl.service"
+    tmux split-window -t zrepl "watch 'zfs list -t snapshot -o name,creation -s creation | grep zrepl_'"
+    tmux split-window -t zrepl "zrepl status"
+    tmux select-layout -t zrepl tiled
+    tmux attach -t zrepl
 
 Summary
 -------

docs/usage.rst

@@ -27,8 +27,10 @@ CLI Overview
       - show job activity, or with ``--raw`` for JSON output
     * - ``zrepl stdinserver``
       - see :ref:`transport-ssh+stdinserver`
-    * - ``zrepl wakeup JOB``
-      - manually trigger replication + pruning
+    * - ``zrepl signal wakeup JOB``
+      - manually trigger replication + pruning of JOB
+    * - ``zrepl signal reset JOB``
+      - manually abort current replication + pruning of JOB
     * - ``zrepl configcheck``
       - check if config can be parsed without errors
 

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"
@@ -70,35 +71,30 @@ type Report struct {
 	Completed, Pending []*StepReport
 }
 
-//go:generate stringer -type=State
+//go:generate enumer -type=State
 type State uint
 
 const (
 	Ready State = 1 << iota
-	RetryWait
-	PermanentError
 	Completed
 )
 
-func (s State) fsrsf() state {
-	m := map[State]state{
-		Ready:          stateReady,
-		RetryWait:      stateRetryWait,
-		PermanentError: nil,
-		Completed:      nil,
-	}
-	return m[s]
+type Error interface {
+	error
+	Temporary() bool
+	ContextErr() bool
+	LocalToFS() bool
 }
 
 type Replication struct {
 	promBytesReplicated prometheus.Counter
 
+	fs                 string
+
 	// lock protects all fields below it in this struct, but not the data behind pointers
 	lock               sync.Mutex
 	state              State
-	fs                 string
-	err                error
-	retryWaitUntil     time.Time
+	err                Error
 	completed, pending []*ReplicationStep
 }
 
@@ -108,6 +104,37 @@ func (f *Replication) State() State {
 	return f.state
 }
 
+func (f *Replication) FS() string { return f.fs }
+
+// returns zero value time.Time{} if no more pending steps
+func (f *Replication) NextStepDate() time.Time {
+	if len(f.pending) == 0 {
+		return time.Time{}
+	}
+	return f.pending[0].to.SnapshotTime()
+}
+
+func (f *Replication) Err() Error {
+	f.lock.Lock()
+	defer f.lock.Unlock()
+	return f.err
+}
+
+func (f *Replication) CanRetry() bool {
+	f.lock.Lock()
+	defer f.lock.Unlock()
+	if f.state == Completed {
+		return false
+	}
+	if f.state != Ready {
+		panic(fmt.Sprintf("implementation error: %v", f.state))
+	}
+	if f.err == nil {
+		return true
+	}
+	return f.err.Temporary()
+}
+
 func (f *Replication) UpdateSizeEsitmate(ctx context.Context, sender Sender) error {
 	f.lock.Lock()
 	defer f.lock.Unlock()
@@ -149,26 +176,39 @@ func (b *ReplicationBuilder) Done() (r *Replication) {
 	return r
 }
 
-func NewReplicationWithPermanentError(fs string, err error) *Replication {
+type ReplicationConflictError struct {
+	Err error
+}
+
+func (e *ReplicationConflictError) Timeout() bool { return false }
+
+func (e *ReplicationConflictError) Temporary() bool { return false }
+
+func (e *ReplicationConflictError) Error() string { return fmt.Sprintf("permanent error: %s", e.Err.Error()) }
+
+func (e *ReplicationConflictError) LocalToFS() bool { return true }
+
+func (e *ReplicationConflictError) ContextErr() bool { return false }
+
+func NewReplicationConflictError(fs string, err error) *Replication {
 	return &Replication{
-		state: PermanentError,
+		state: Completed,
 		fs:    fs,
-		err:   err,
+		err:   &ReplicationConflictError{Err: err},
 	}
 }
 
-//go:generate stringer -type=StepState
+//go:generate enumer -type=StepState
 type StepState uint
 
 const (
 	StepReplicationReady StepState = 1 << iota
-	StepReplicationRetry
 	StepMarkReplicatedReady
-	StepMarkReplicatedRetry
-	StepPermanentError
 	StepCompleted
 )
 
+func (s StepState) IsTerminal() bool { return s == StepCompleted }
+
 type FilesystemVersion interface {
 	SnapshotTime() time.Time
 	GetName() string // name without @ or #
@@ -191,7 +231,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) Retry(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver) Error {
 
 	var u updater = func(fu func(*Replication)) State {
 		f.lock.Lock()
@@ -201,93 +241,108 @@ func (f *Replication) TakeStep(ctx context.Context, sender Sender, receiver Rece
 		}
 		return f.state
 	}
-	var s state = u(nil).fsrsf()
 
-	pre := u(nil)
-	preTime := time.Now()
-	s = s(ctx, sender, receiver, u)
-	delta := time.Now().Sub(preTime)
-
-	post = u(func(f *Replication) {
-		if len(f.pending) == 0 {
-			return
-		}
-		nextStepDate = f.pending[0].to.SnapshotTime()
-		retryWaitUntil = f.retryWaitUntil
-	})
-
-	getLogger(ctx).
-		WithField("fs", f.fs).
-		WithField("transition", fmt.Sprintf("%s => %s", pre, post)).
-		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
-}
-
-type updater func(func(fsr *Replication)) State
-
-type state func(ctx context.Context, 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 {
 
 	var current *ReplicationStep
-	s := u(func(f *Replication) {
+	pre := u(nil)
+	getLogger(ctx).WithField("fsrep_state", pre).Debug("begin fsrep.Retry")
+	defer func() {
+		post := u(nil)
+		getLogger(ctx).WithField("fsrep_transition", post).Debug("end fsrep.Retry")
+	}()
+
+	st := u(func(f *Replication) {
 		if len(f.pending) == 0 {
 			f.state = Completed
 			return
 		}
 		current = f.pending[0]
 	})
-	if s != Ready {
-		return s.fsrsf()
+	if st == Completed {
+		return nil
+	}
+	if st != Ready {
+		panic(fmt.Sprintf("implementation error: %v", st))
 	}
 
-	stepState := current.doReplication(ctx, sender, receiver)
+	stepCtx := WithLogger(ctx, getLogger(ctx).WithField("step", current))
+	getLogger(stepCtx).Debug("take step")
+	err := current.Retry(stepCtx, ka, sender, receiver)
+	if err != nil {
+		getLogger(stepCtx).WithError(err).Error("step could not be completed")
+	}
 
-	return u(func(f *Replication) {
-		switch stepState {
-		case StepCompleted:
-			f.completed = append(f.completed, current)
-			f.pending = f.pending[1:]
-			if len(f.pending) > 0 {
-				f.state = Ready
-			} else {
-				f.state = Completed
-			}
-		case StepReplicationRetry:
-			fallthrough
-		case StepMarkReplicatedRetry:
-			f.retryWaitUntil = time.Now().Add(RetrySleepDuration)
-			f.state = RetryWait
-		case StepPermanentError:
-			f.state = PermanentError
-			f.err = errors.New("a replication step failed with a permanent error")
-		default:
-			panic(f)
+	u(func(fsr *Replication) {
+		if err != nil {
+			f.err = &StepError{stepStr: current.String(), err: err}
+			return
 		}
-	}).fsrsf()
+		if err == nil && current.state != StepCompleted {
+			panic(fmt.Sprintf("implementation error: %v", current.state))
+		}
+		f.err = nil
+		f.completed = append(f.completed, current)
+		f.pending = f.pending[1:]
+		if len(f.pending) > 0 {
+			f.state = Ready
+		} else {
+			f.state = Completed
+		}
+	})
+	var retErr Error = nil
+	u(func(fsr *Replication) {
+		retErr = fsr.err
+	})
+	return retErr
 }
 
-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()
+type updater func(func(fsr *Replication)) State
+
+type state func(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver, u updater) state
+
+type StepError struct {
+	stepStr string
+	err     error
+}
+
+var _ Error = &StepError{}
+
+func (e StepError) Error() string {
+	if e.LocalToFS() {
+		return fmt.Sprintf("step %s failed: %s", e.stepStr, e.err)
 	}
-	return u(func(f *Replication) {
-		f.state = Ready
-	}).fsrsf()
+	return e.err.Error()
+}
+
+func (e StepError) Timeout() bool {
+	if neterr, ok := e.err.(net.Error); ok {
+		return neterr.Timeout()
+	}
+	return false
+}
+
+func (e StepError) Temporary() bool {
+	if neterr, ok := e.err.(net.Error); ok {
+		return neterr.Temporary()
+	}
+	return false
+}
+
+func (e StepError) LocalToFS() bool {
+	if _, ok := e.err.(net.Error); ok {
+		return false
+	}
+	return true // conservative approximation: we'd like to check for specific errors returned over RPC here...
+}
+
+func (e StepError) ContextErr() bool {
+	switch e.err {
+	case context.Canceled:
+		return true
+	case context.DeadlineExceeded:
+		return true
+	}
+	return false
 }
 
 func (fsr *Replication) Report() *Report {
@@ -299,9 +354,8 @@ func (fsr *Replication) Report() *Report {
 		Status:     fsr.state.String(),
 	}
 
-	if fsr.state&PermanentError != 0 {
+	if fsr.err != nil && fsr.err.LocalToFS() {
 		rep.Problem = fsr.err.Error()
-		return &rep
 	}
 
 	rep.Completed = make([]*StepReport, len(fsr.completed))
@@ -313,73 +367,54 @@ 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?
-			rep.Problem = rep.Pending[0].Problem
-		}
-	}
-
 	return &rep
 }
 
-func shouldRetry(err error) bool {
-	switch err {
-	case io.EOF:
-		fallthrough
-	case io.ErrUnexpectedEOF:
-		fallthrough
-	case io.ErrClosedPipe:
-		return true
+func (s *ReplicationStep) Retry(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver) error {
+	switch s.state {
+	case StepReplicationReady:
+		return s.doReplication(ctx, ka, sender, receiver)
+	case StepMarkReplicatedReady:
+		return s.doMarkReplicated(ctx, ka, sender)
+	case StepCompleted:
+		return nil
 	}
-	if _, ok := err.(net.Error); ok {
-		return true
-	}
-	return false
+	panic(fmt.Sprintf("implementation error: %v", s.state))
 }
 
-func (s *ReplicationStep) doReplication(ctx context.Context, sender Sender, receiver Receiver) StepState {
+func (s *ReplicationStep) Error() error {
+	if s.state & (StepReplicationReady|StepMarkReplicatedReady) != 0 {
+		return s.err
+	}
+	return nil
+}
+
+func (s *ReplicationStep) doReplication(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver) error {
+
+	if s.state != StepReplicationReady {
+		panic(fmt.Sprintf("implementation error: %v", s.state))
+	}
 
 	fs := s.parent.fs
 
-	log := getLogger(ctx).
-		WithField("filesystem", fs).
-		WithField("step", s.String())
-
-	updateStateError := func(err error) StepState {
-		s.lock.Lock()
-		defer s.lock.Unlock()
-
-		s.err = err
-		if shouldRetry(s.err) {
-			s.state = StepReplicationRetry
-			return s.state
-		}
-		s.state = StepPermanentError
-		return s.state
-	}
-
-	updateStateCompleted := func() StepState {
-		s.lock.Lock()
-		defer s.lock.Unlock()
-		s.err = nil
-		s.state = StepMarkReplicatedReady
-		return s.state
-	}
-
+	log := getLogger(ctx)
 	sr := s.buildSendRequest(false)
 
 	log.Debug("initiate send request")
 	sres, sstream, err := sender.Send(ctx, sr)
 	if err != nil {
 		log.WithError(err).Error("send request failed")
-		return updateStateError(err)
+		return err
 	}
 	if sstream == nil {
 		err := errors.New("send request did not return a stream, broken endpoint implementation")
-		return updateStateError(err)
+		return err
 	}
 
 	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()))
 	}()
@@ -401,41 +436,23 @@ func (s *ReplicationStep) doReplication(ctx context.Context, sender Sender, rece
 		// 	- an unexpected exit of ZFS on the sending side
 		//  - an unexpected exit of ZFS on the receiving side
 		//  - a connectivity issue
-		return updateStateError(err)
+		return err
 	}
 	log.Debug("receive finished")
+	ka.MadeProgress()
 
-	updateStateCompleted()
-
-	return s.doMarkReplicated(ctx, sender)
+	s.state = StepMarkReplicatedReady
+	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) error {
 
-	log := getLogger(ctx).
-		WithField("filesystem", s.parent.fs).
-		WithField("step", s.String())
-
-	updateStateError := func(err error) StepState {
-		s.lock.Lock()
-		defer s.lock.Unlock()
-
-		s.err = err
-		if shouldRetry(s.err) {
-			s.state = StepMarkReplicatedRetry
-			return s.state
-		}
-		s.state = StepPermanentError
-		return s.state
+	if s.state != StepMarkReplicatedReady {
+		panic(fmt.Sprintf("implementation error: %v", s.state))
 	}
 
-	updateStateCompleted := func() StepState {
-		s.lock.Lock()
-		defer s.lock.Unlock()
-		s.state = StepCompleted
-		return s.state
-	}
+	log := getLogger(ctx)
 
 	log.Debug("advance replication cursor")
 	req := &pdu.ReplicationCursorReq{
@@ -449,24 +466,22 @@ func (s *ReplicationStep) doMarkReplicated(ctx context.Context, sender Sender) S
 	res, err := sender.ReplicationCursor(ctx, req)
 	if err != nil {
 		log.WithError(err).Error("error advancing replication cursor")
-		return updateStateError(err)
+		return err
 	}
 	if res.GetError() != "" {
 		err := fmt.Errorf("cannot advance replication cursor: %s", res.GetError())
 		log.Error(err.Error())
-		return updateStateError(err)
+		return err
 	}
+	ka.MadeProgress()
 
-	return updateStateCompleted()
+	s.state = StepCompleted
+	return err
 }
 
 func (s *ReplicationStep) updateSizeEstimate(ctx context.Context, sender Sender) error {
 
-	fs := s.parent.fs
-
-	log := getLogger(ctx).
-		WithField("filesystem", fs).
-		WithField("step", s.String())
+	log := getLogger(ctx)
 
 	sr := s.buildSendRequest(true)
 
@@ -485,7 +500,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_enumer.go

@@ -0,0 +1,50 @@
+// Code generated by "enumer -type=State"; DO NOT EDIT.
+
+package fsrep
+
+import (
+	"fmt"
+)
+
+const _StateName = "ReadyCompleted"
+
+var _StateIndex = [...]uint8{0, 5, 14}
+
+func (i State) String() string {
+	i -= 1
+	if i >= State(len(_StateIndex)-1) {
+		return fmt.Sprintf("State(%d)", i+1)
+	}
+	return _StateName[_StateIndex[i]:_StateIndex[i+1]]
+}
+
+var _StateValues = []State{1, 2}
+
+var _StateNameToValueMap = map[string]State{
+	_StateName[0:5]:  1,
+	_StateName[5:14]: 2,
+}
+
+// StateString retrieves an enum value from the enum constants string name.
+// Throws an error if the param is not part of the enum.
+func StateString(s string) (State, error) {
+	if val, ok := _StateNameToValueMap[s]; ok {
+		return val, nil
+	}
+	return 0, fmt.Errorf("%s does not belong to State values", s)
+}
+
+// StateValues returns all values of the enum
+func StateValues() []State {
+	return _StateValues
+}
+
+// IsAState returns "true" if the value is listed in the enum definition. "false" otherwise
+func (i State) IsAState() bool {
+	for _, v := range _StateValues {
+		if i == v {
+			return true
+		}
+	}
+	return false
+}

replication/fsrep/state_string.go

@@ -1,29 +0,0 @@
-// Code generated by "stringer -type=State"; DO NOT EDIT.
-
-package fsrep
-
-import "strconv"
-
-const (
-	_State_name_0 = "ReadyRetryWait"
-	_State_name_1 = "PermanentError"
-	_State_name_2 = "Completed"
-)
-
-var (
-	_State_index_0 = [...]uint8{0, 5, 14}
-)
-
-func (i State) String() string {
-	switch {
-	case 1 <= i && i <= 2:
-		i -= 1
-		return _State_name_0[_State_index_0[i]:_State_index_0[i+1]]
-	case i == 4:
-		return _State_name_1
-	case i == 8:
-		return _State_name_2
-	default:
-		return "State(" + strconv.FormatInt(int64(i), 10) + ")"
-	}
-}

replication/fsrep/stepstate_enumer.go

@@ -0,0 +1,61 @@
+// Code generated by "enumer -type=StepState"; DO NOT EDIT.
+
+package fsrep
+
+import (
+	"fmt"
+)
+
+const (
+	_StepStateName_0 = "StepReplicationReadyStepMarkReplicatedReady"
+	_StepStateName_1 = "StepCompleted"
+)
+
+var (
+	_StepStateIndex_0 = [...]uint8{0, 20, 43}
+	_StepStateIndex_1 = [...]uint8{0, 13}
+)
+
+func (i StepState) String() string {
+	switch {
+	case 1 <= i && i <= 2:
+		i -= 1
+		return _StepStateName_0[_StepStateIndex_0[i]:_StepStateIndex_0[i+1]]
+	case i == 4:
+		return _StepStateName_1
+	default:
+		return fmt.Sprintf("StepState(%d)", i)
+	}
+}
+
+var _StepStateValues = []StepState{1, 2, 4}
+
+var _StepStateNameToValueMap = map[string]StepState{
+	_StepStateName_0[0:20]:  1,
+	_StepStateName_0[20:43]: 2,
+	_StepStateName_1[0:13]:  4,
+}
+
+// StepStateString retrieves an enum value from the enum constants string name.
+// Throws an error if the param is not part of the enum.
+func StepStateString(s string) (StepState, error) {
+	if val, ok := _StepStateNameToValueMap[s]; ok {
+		return val, nil
+	}
+	return 0, fmt.Errorf("%s does not belong to StepState values", s)
+}
+
+// StepStateValues returns all values of the enum
+func StepStateValues() []StepState {
+	return _StepStateValues
+}
+
+// IsAStepState returns "true" if the value is listed in the enum definition. "false" otherwise
+func (i StepState) IsAStepState() bool {
+	for _, v := range _StepStateValues {
+		if i == v {
+			return true
+		}
+	}
+	return false
+}

replication/fsrep/stepstate_string.go

@@ -1,35 +0,0 @@
-// Code generated by "stringer -type=StepState"; DO NOT EDIT.
-
-package fsrep
-
-import "strconv"
-
-const (
-	_StepState_name_0 = "StepReplicationReadyStepReplicationRetry"
-	_StepState_name_1 = "StepMarkReplicatedReady"
-	_StepState_name_2 = "StepMarkReplicatedRetry"
-	_StepState_name_3 = "StepPermanentError"
-	_StepState_name_4 = "StepCompleted"
-)
-
-var (
-	_StepState_index_0 = [...]uint8{0, 20, 40}
-)
-
-func (i StepState) String() string {
-	switch {
-	case 1 <= i && i <= 2:
-		i -= 1
-		return _StepState_name_0[_StepState_index_0[i]:_StepState_index_0[i+1]]
-	case i == 4:
-		return _StepState_name_1
-	case i == 8:
-		return _StepState_name_2
-	case i == 16:
-		return _StepState_name_3
-	case i == 32:
-		return _StepState_name_4
-	default:
-		return "StepState(" + strconv.FormatInt(int64(i), 10) + ")"
-	}
-}

replication/internal/queue/queue.go

@@ -1,121 +0,0 @@
-package queue
-
-import (
-	"sort"
-	"time"
-
-	. "github.com/zrepl/zrepl/replication/fsrep"
-)
-
-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
-
-	fsr *Replication
-}
-
-type ReplicationQueue []*replicationQueueItem
-
-func NewReplicationQueue() *ReplicationQueue {
-	q := make(ReplicationQueue, 0)
-	return &q
-}
-
-func (q ReplicationQueue) Len() int      { return len(q) }
-func (q ReplicationQueue) Swap(i, j int) { q[i], q[j] = q[j], q[i] }
-
-type lessmapEntry struct {
-	prio int
-	less func(a, b *replicationQueueItem) bool
-}
-
-var lessmap = map[State]lessmapEntry{
-	Ready: {
-		prio: 0,
-		less: func(a, b *replicationQueueItem) bool {
-			return a.nextStepDate.Before(b.nextStepDate)
-		},
-	},
-	RetryWait: {
-		prio: 1,
-		less: func(a, b *replicationQueueItem) bool {
-			return a.retryWaitUntil.Before(b.retryWaitUntil)
-		},
-	},
-}
-
-func (q ReplicationQueue) Less(i, j int) bool {
-
-	a, b := q[i], q[j]
-	al, aok := lessmap[a.state]
-	if !aok {
-		panic(a)
-	}
-	bl, bok := lessmap[b.state]
-	if !bok {
-		panic(b)
-	}
-
-	if al.prio != bl.prio {
-		return al.prio < bl.prio
-	}
-
-	return al.less(a, b)
-}
-
-func (q *ReplicationQueue) sort() (done []*Replication) {
-	// pre-scan for everything that is not ready
-	newq := make(ReplicationQueue, 0, len(*q))
-	done = make([]*Replication, 0, len(*q))
-	for _, qitem := range *q {
-		if _, ok := lessmap[qitem.state]; !ok {
-			done = append(done, qitem.fsr)
-		} else {
-			newq = append(newq, qitem)
-		}
-	}
-	sort.Stable(newq) // stable to avoid flickering in reports
-	*q = newq
-	return done
-}
-
-// next remains valid until the next call to GetNext()
-func (q *ReplicationQueue) GetNext() (done []*Replication, next *ReplicationQueueItemHandle) {
-	done = q.sort()
-	if len(*q) == 0 {
-		return done, nil
-	}
-	next = &ReplicationQueueItemHandle{(*q)[0]}
-	return done, next
-}
-
-func (q *ReplicationQueue) Add(fsr *Replication) {
-	*q = append(*q, &replicationQueueItem{
-		fsr:   fsr,
-		state: fsr.State(),
-	})
-}
-
-func (q *ReplicationQueue) Foreach(fu func(*ReplicationQueueItemHandle)) {
-	for _, qitem := range *q {
-		fu(&ReplicationQueueItemHandle{qitem})
-	}
-}
-
-type ReplicationQueueItemHandle struct {
-	i *replicationQueueItem
-}
-
-func (h ReplicationQueueItemHandle) GetFSReplication() *Replication {
-	return h.i.fsr
-}
-
-func (h ReplicationQueueItemHandle) Update(newState State, nextStepDate, retryWaitUntil time.Time) {
-	h.i.state = newState
-	h.i.nextStepDate = nextStepDate
-	h.i.retryWaitUntil = retryWaitUntil
-}

replication/mainfsm.go

@@ -8,13 +8,17 @@ 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"
+	"github.com/problame/go-streamrpc"
 	"math/bits"
+	"net"
+	"sort"
 	"sync"
 	"time"
 
 	"github.com/zrepl/zrepl/replication/fsrep"
 	. "github.com/zrepl/zrepl/replication/internal/diff"
-	. "github.com/zrepl/zrepl/replication/internal/queue"
 	"github.com/zrepl/zrepl/replication/pdu"
 )
 
@@ -27,7 +31,7 @@ const (
 	Working
 	WorkingWait
 	Completed
-	ContextDone
+	PermanentError
 )
 
 func (s State) rsf() state {
@@ -46,6 +50,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,21 +63,20 @@ 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
 
 	state State
 
 	// Working, WorkingWait, Completed, ContextDone
-	queue     *ReplicationQueue
+	queue     []*fsrep.Replication
 	completed []*fsrep.Replication
-	active    *ReplicationQueueItemHandle
+	active    *fsrep.Replication // == queue[0] or nil, unlike in Report
 
-	// PlanningError
-	planningError error
-
-	// ContextDone
-	contextError error
+	// for PlanningError, WorkingWait and ContextError and Completed
+	err error
 
 	// PlanningError, WorkingWait
 	sleepUntil time.Time
@@ -81,7 +88,7 @@ type Report struct {
 	SleepUntil time.Time
 	Completed []*fsrep.Report
 	Pending   []*fsrep.Report
-	Active    *fsrep.Report
+	Active    *fsrep.Report // not contained in Pending, unlike in struct Replication
 }
 
 func NewReplication(secsPerState *prometheus.HistogramVec, bytesReplicated *prometheus.CounterVec) *Replication {
@@ -124,7 +131,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 +156,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 +194,41 @@ 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", 10 * time.Second)
 
-func statePlanning(ctx context.Context, sender Sender, receiver Receiver, u updater) state {
+type Error interface {
+	error
+	Temporary() bool
+}
+
+var _ Error = fsrep.Error(nil)
+var _ Error = net.Error(nil)
+var _ Error = streamrpc.Error(nil)
+
+func isPermanent(err error) bool {
+	if e, ok := err.(Error); ok {
+		return !e.Temporary()
+	}
+	return true
+}
+
+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
+			ge := GlobalError{Err: err, Temporary: !isPermanent(err)}
+			log.WithError(ge).Error("encountered global error while planning replication")
+			r.err = ge
+			if !ge.Temporary {
+				r.state = PermanentError
+			} else {
+				r.sleepUntil = time.Now().Add(RetryInterval)
+				r.state = PlanningError
+			}
 		}).rsf()
 	}
 
@@ -209,6 +237,7 @@ func statePlanning(ctx context.Context, sender Sender, receiver Receiver, u upda
 		log.WithError(err).Error("error listing sender filesystems")
 		return handlePlanningError(err)
 	}
+	// no progress here since we could run in a live-lock on connectivity issues
 
 	rfss, err := receiver.ListFilesystems(ctx)
 	if err != nil {
@@ -216,7 +245,9 @@ func statePlanning(ctx context.Context, sender Sender, receiver Receiver, u upda
 		return handlePlanningError(err)
 	}
 
-	q := NewReplicationQueue()
+	ka.MadeProgress() // for both sender and receiver
+
+	q := make([]*fsrep.Replication, 0, len(sfss))
 	mainlog := log
 	for _, fs := range sfss {
 
@@ -229,11 +260,12 @@ func statePlanning(ctx context.Context, sender Sender, receiver Receiver, u upda
 			log.WithError(err).Error("cannot get remote filesystem versions")
 			return handlePlanningError(err)
 		}
+		ka.MadeProgress()
 
 		if len(sfsvs) < 1 {
 			err := errors.New("sender does not have any versions")
 			log.Error(err.Error())
-			q.Add(fsrep.NewReplicationWithPermanentError(fs.Path, err))
+			q = append(q, fsrep.NewReplicationConflictError(fs.Path, err))
 			continue
 		}
 
@@ -258,6 +290,7 @@ func statePlanning(ctx context.Context, sender Sender, receiver Receiver, u upda
 		} else {
 			rfsvs = []*pdu.FilesystemVersion{}
 		}
+		ka.MadeProgress()
 
 		path, conflict := IncrementalPath(rfsvs, sfsvs)
 		if conflict != nil {
@@ -271,8 +304,9 @@ func statePlanning(ctx context.Context, sender Sender, receiver Receiver, u upda
 				log.WithField("problem", msg).Error("cannot resolve conflict")
 			}
 		}
+		ka.MadeProgress()
 		if path == nil {
-			q.Add(fsrep.NewReplicationWithPermanentError(fs.Path, conflict))
+			q = append(q, fsrep.NewReplicationConflictError(fs.Path, conflict))
 			continue
 		}
 
@@ -289,24 +323,29 @@ func statePlanning(ctx context.Context, sender Sender, receiver Receiver, u upda
 			}
 		}
 		qitem := fsrfsm.Done()
+		ka.MadeProgress()
 
 		log.Debug("compute send size estimate")
 		if err = qitem.UpdateSizeEsitmate(ctx, sender); err != nil {
 			log.WithError(err).Error("error computing size estimate")
 			return handlePlanningError(err)
 		}
-		q.Add(qitem)
+		ka.MadeProgress()
+
+		q = append(q, 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 +356,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,51 +367,143 @@ func statePlanningError(ctx context.Context, sender Sender, receiver Receiver, u
 	}).rsf()
 }
 
-func stateWorking(ctx context.Context, sender Sender, receiver Receiver, u updater) state {
+type GlobalError struct {
+	Err       error
+	Temporary bool
+}
 
-	var active *ReplicationQueueItemHandle
+func (e GlobalError) Error() string {
+	errClass := "temporary"
+	if !e.Temporary {
+		errClass = "permanent"
+	}
+	return fmt.Sprintf("%s global error: %s", errClass, e.Err)
+}
+
+type FilesystemsReplicationFailedError struct {
+	FilesystemsWithError []*fsrep.Replication
+}
+
+func (e FilesystemsReplicationFailedError) Error() string {
+	allSame := true
+	lastErr := e.FilesystemsWithError[0].Err().Error()
+	for _, fs := range e.FilesystemsWithError {
+		fsErr := fs.Err().Error()
+		allSame = allSame && lastErr == fsErr
+	}
+
+	fsstr := "multiple filesystems"
+	if len(e.FilesystemsWithError) == 1 {
+		fsstr = fmt.Sprintf("filesystem %s", e.FilesystemsWithError[0].FS())
+	}
+	errorStr := lastErr
+	if !allSame {
+		errorStr = "multiple different errors"
+	}
+	return fmt.Sprintf("%s could not be replicated: %s", fsstr, errorStr)
+}
+
+func stateWorking(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver, u updater) state {
+
+	var active *fsrep.Replication
 	rsfNext := u(func(r *Replication) {
-		done, next := r.queue.GetNext()
-		r.completed = append(r.completed, done...)
-		if next == nil {
-			r.state = Completed
+
+		r.err = nil
+
+		newq := make([]*fsrep.Replication, 0, len(r.queue))
+		for i := range r.queue {
+			if r.queue[i].CanRetry() {
+				newq = append(newq, r.queue[i])
+			} else {
+				r.completed = append(r.completed, r.queue[i])
+			}
 		}
-		r.active = next
-		active = next
+		sort.SliceStable(newq, func(i, j int) bool {
+			return newq[i].NextStepDate().Before(newq[j].NextStepDate())
+		})
+		r.queue = newq
+
+		if len(r.queue) == 0 {
+			r.state = Completed
+			fsWithErr := FilesystemsReplicationFailedError{ // prepare it
+				FilesystemsWithError: make([]*fsrep.Replication, 0, len(r.completed)),
+			}
+			for _, fs := range r.completed {
+				if fs.CanRetry() {
+					panic(fmt.Sprintf("implementation error: completed contains retryable FS %s %#v",
+						fs.FS(), fs.Err()))
+				}
+				if fs.Err() != nil {
+					fsWithErr.FilesystemsWithError = append(fsWithErr.FilesystemsWithError, fs)
+				}
+			}
+			if len(fsWithErr.FilesystemsWithError) > 0 {
+				r.err = fsWithErr
+				r.state = PermanentError
+			}
+			return
+		}
+
+		active =  r.queue[0] // do not dequeue: if it's done, it will be sorted the next time we check for more work
+		r.active = active
 	}).rsf()
 
 	if active == nil {
 		return rsfNext
 	}
 
-	retryWaitUntil := active.GetFSReplication().RetryWaitUntil()
-	if retryWaitUntil.After(time.Now()) {
-		return u(func(r *Replication) {
-			r.sleepUntil = retryWaitUntil
-			r.state = WorkingWait
-		}).rsf()
-	}
-
-	state, nextStepDate, retryWaitUntil := active.GetFSReplication().TakeStep(ctx, sender, receiver)
-	return u(func(r *Replication) {
-		active.Update(state, nextStepDate, retryWaitUntil)
+	activeCtx := fsrep.WithLogger(ctx, getLogger(ctx).WithField("fs", active.FS()))
+	err := active.Retry(activeCtx, ka, sender, receiver)
+	u(func(r *Replication) {
 		r.active = nil
 	}).rsf()
+
+	if err != nil {
+		if err.ContextErr() && ctx.Err() != nil {
+			getLogger(ctx).WithError(err).
+				Info("filesystem replication was cancelled")
+			u(func(r*Replication) {
+				r.err = GlobalError{Err: err, Temporary: false}
+				r.state = PermanentError
+			})
+		} else if err.LocalToFS() {
+			getLogger(ctx).WithError(err).
+				Error("filesystem replication encountered a filesystem-specific error")
+			// we stay in this state and let the queuing logic above de-prioritize this failing FS
+		} else if err.Temporary() {
+			getLogger(ctx).WithError(err).
+				Error("filesystem encountered a non-filesystem-specific temporary error, enter retry-wait")
+			u(func(r *Replication) {
+				r.err = GlobalError{Err: err, Temporary: true}
+				r.sleepUntil = time.Now().Add(RetryInterval)
+				r.state = WorkingWait
+			}).rsf()
+		} else {
+			getLogger(ctx).WithError(err).
+				Error("encountered a permanent non-filesystem-specific error")
+			u(func(r *Replication) {
+				r.err = GlobalError{Err: err, Temporary: false}
+				r.state = PermanentError
+			}).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 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,33 +526,35 @@ 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.err != nil {
+		rep.Problem = r.err.Error()
+	}
+
+	if r.state&(Planning|PlanningError) != 0 {
 		return &rep
 	}
 
-	rep.Pending = make([]*fsrep.Report, 0, r.queue.Len())
+	rep.Pending = make([]*fsrep.Report, 0, len(r.queue))
 	rep.Completed = make([]*fsrep.Report, 0, len(r.completed)) // room for active (potentially)
 
-	var active *fsrep.Replication
+	// since r.active == r.queue[0], do not contain it in pending output
+	pending := r.queue
 	if r.active != nil {
-		active = r.active.GetFSReplication()
-		rep.Active = active.Report()
+		rep.Active = r.active.Report()
+		pending = r.queue[1:]
+	}
+	for _, fsr := range pending {
+		rep.Pending= append(rep.Pending, fsr.Report())
 	}
-	r.queue.Foreach(func(h *ReplicationQueueItemHandle) {
-		fsr := h.GetFSReplication()
-		if active != fsr {
-			rep.Pending = append(rep.Pending, fsr.Report())
-		}
-	})
 	for _, fsr := range r.completed {
 		rep.Completed = append(rep.Completed, fsr.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,31 @@
+package watchdog
+
+import (
+	"fmt"
+	"sync"
+	"time"
+)
+
+type KeepAlive struct {
+	mtx sync.Mutex
+	lastUpd time.Time
+}
+
+func (p *KeepAlive) String() string {
+	if p.lastUpd.IsZero() {
+		return fmt.Sprintf("never updated")
+	}
+	return fmt.Sprintf("last update at %s", p.lastUpd)
+}
+
+func (k *KeepAlive) MadeProgress() {
+	k.mtx.Lock()
+	defer k.mtx.Unlock()
+	k.lastUpd = time.Now()
+}
+
+func (k *KeepAlive) CheckTimeout(timeout time.Duration, jitter time.Duration) (didTimeOut bool) {
+	k.mtx.Lock()
+	defer k.mtx.Unlock()
+	return k.lastUpd.Add(timeout - jitter).Before(time.Now())
+}

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