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"
+  revision = "d5d111e014342fe1c37f0b71cc37ec5f2afdfd13"
-  version = "v0.4"
+  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,7 +22,6 @@ SUBPKGS += pruning/retentiongrid
 SUBPKGS += replication
 SUBPKGS += replication/fsrep
 SUBPKGS += replication/pdu
-SUBPKGS += replication/internal/queue
 SUBPKGS += replication/internal/diff
 SUBPKGS += tlsconf
 SUBPKGS += util

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,7 +356,8 @@ 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.IsTerminal() {
@@ -297,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.drawBar(50, transferred, total, changeCount)
-		t.write(fmt.Sprintf(" %s / %s", ByteCountBinary(transferred), ByteCountBinary(total)))
+		t.write(fmt.Sprintf(" %s / %s @ %s/s", ByteCountBinary(transferred), ByteCountBinary(total), ByteCountBinary(rate)))
 		t.newline()
 	}
 
@@ -310,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)
 	}
 }
 
@@ -391,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 != "" {
+		if fs.LastError != "" {
-			t.printf("ERROR:    %s\n", fs.Error) // whitespace is padding
+			t.printf("ERROR (%d): %s\n", fs.ErrorCount, fs.LastError) // whitespace is padding
 			continue
 		}
 
@@ -456,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)
@@ -469,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("]")
 }
@@ -477,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) {
+func (t *tui) printFilesystemStatus(rep *fsrep.Report, active bool, maxFS int) {
-	bytes = int64(0)
+
-	totalBytes = int64(0)
+	bytes := int64(0)
+	totalBytes := int64(0)
 	for _, s := range rep.Pending {
 		bytes += s.Bytes
 		totalBytes += s.ExpectedBytes
@@ -499,9 +587,26 @@ func filesystemStatusString(rep *fsrep.Report, active bool, fsWidth int) (line s
 		totalBytes += s.ExpectedBytes
 	}
 
+
+	status := fmt.Sprintf("%s (step %d/%d, %s/%s)",
+		rep.Status,
+		len(rep.Completed), len(rep.Pending) + len(rep.Completed),
+		ByteCountBinary(bytes), ByteCountBinary(totalBytes),
+
+	)
+
+	activeIndicator := " "
+	if active {
+		activeIndicator = "*"
+	}
+	t.printf("%s %s %s ",
+		activeIndicator,
+		rightPad(rep.Filesystem, maxFS, " "),
+		status)
+
 	next := ""
 	if rep.Problem != "" {
-		next = " problem: " + 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)
@@ -509,26 +614,8 @@ func filesystemStatusString(rep *fsrep.Report, active bool, fsWidth int) (line s
 			next = fmt.Sprintf("next: %s (full)", rep.Pending[0].To)
 		}
 	}
-	status := fmt.Sprintf("%s (step %d/%d, %s/%s)%s",
+	t.printfDrawIndentedAndWrappedIfMultiline("%s", next)
-		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",
-		activeIndicator,
-		rightPad(rep.Filesystem, fsWidth, " "),
-		status)
-	return line, bytes, totalBytes
-}
 
-func printFilesystemStatus(rep *fsrep.Report, t *tui, active bool, maxFS int) {
-	totalStatus, _, _ := filesystemStatusString(rep, active, maxFS)
-	t.write(totalStatus)
 	t.newline()
 }
 

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,19 +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/logging"
-	"github.com/zrepl/zrepl/daemon/filters"
 	"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/util/watchdog"
 	"github.com/zrepl/zrepl/zfs"
 	"sync"
-	"github.com/zrepl/zrepl/daemon/logging"
-	"github.com/zrepl/zrepl/daemon/snapper"
 	"time"
 )
 
@@ -38,10 +37,29 @@ 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
 }
 
@@ -254,11 +272,8 @@ func (j *ActiveSide) do(ctx context.Context) {
 	// 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()
@@ -266,53 +281,74 @@ func (j *ActiveSide) do(ctx context.Context) {
 		}
 	}()
 
-	// watchdog
+	// 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() {
-		// if no progress after 1 minute, kill the task
+
-		wdto := envconst.Duration("ZREPL_JOB_WATCHDOG_TIMEOUT", 1*time.Minute)
+		wdto := envconst.Duration("ZREPL_JOB_WATCHDOG_TIMEOUT", 10*time.Minute)
-		log.WithField("watchdog_timeout", wdto.String()).Debug("starting watchdog")
+		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()
 
-		var (
-			rep, prunerSender, prunerReceiver watchdog.Progress
-		)
 		for {
 			select {
-			case <-runDone:
-				return
 			case <-ctx.Done():
 				return
 			case <-t.C: // fall
 			}
 
-			log := log.WithField("watchdog_timeout", wdto.String()) // shadowing!
-
 			j.updateTasks(func(tasks *activeSideTasks) {
-				if tasks.replication != nil &&
+				// Since cancelling a task will cause the sequential code to transition to the next state immediately,
-					!tasks.replication.Progress.ExpectProgress(&rep) &&
+				// we cannot check for its progress right then (no fallthrough).
-					!tasks.replication.State().IsTerminal() {
+				// Instead, we return (not continue because we are in a closure) and give the new state another
-					log.Error("replication did not make progress, cancelling")
+				// 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
 					}
-				if tasks.prunerSender != nil &&
+				case ActiveSidePruneSender:
-					!tasks.prunerSender.Progress.ExpectProgress(&prunerSender) &&
+					log.WithField("prune_sender_progress", tasks.replication.Progress.String()).
-					!tasks.prunerSender.State().IsTerminal() {
+						Debug("check pruner_sender progress")
-					log.Error("pruner:sender did not make progress, cancelling")
+					if tasks.prunerSender.Progress.CheckTimeout(wdto, jitter) {
+						log.Error("pruner_sender did not make progress, cancelling" + WATCHDOG_ENVCONST_NOTICE)
 						tasks.prunerSenderCancel()
+						return
 					}
-				if tasks.prunerReceiver != nil &&
+				case ActiveSidePruneReceiver:
-					!tasks.prunerReceiver.Progress.ExpectProgress(&prunerReceiver) &&
+					log.WithField("prune_receiver_progress", tasks.replication.Progress.String()).
-					!tasks.prunerReceiver.State().IsTerminal() {
+						Debug("check pruner_receiver progress")
-					log.Error("pruner:receiver did not make progress, cancelling")
+					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")
 				}
 			})
-			log.WithField("replication_progress", rep.String()).
-				WithField("pruner_sender_progress", prunerSender.String()).
-				WithField("pruner_receiver_progress", prunerReceiver.String()).
-				Debug("watchdog did run")
 
 		}
 	}()
@@ -337,6 +373,7 @@ func (j *ActiveSide) do(ctx context.Context) {
 			*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)
@@ -353,6 +390,7 @@ func (j *ActiveSide) do(ctx context.Context) {
 		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()
@@ -369,10 +407,16 @@ func (j *ActiveSide) do(ctx context.Context) {
 		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()
 	}
+
+	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/pruner/pruner.go

@@ -11,8 +11,10 @@ import (
 	"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"
 )
@@ -69,8 +71,7 @@ type Pruner struct {
 	err        error
 
 	// State Exec
-	prunePending   []*fs
+	execQueue *execQueue
-	pruneCompleted []*fs
 }
 
 type PrunerFactory struct {
@@ -116,7 +117,7 @@ func NewPrunerFactory(in config.PruningSenderReceiver, promPruneSecs *prometheus
 	f := &PrunerFactory{
 		senderRules: keepRulesSender,
 		receiverRules: keepRulesReceiver,
-		retryWait: envconst.Duration("ZREPL_PRUNER_RETRY_INTERVAL", 4 * time.Second),
+		retryWait: envconst.Duration("ZREPL_PRUNER_RETRY_INTERVAL", 10 * time.Second),
 		considerSnapAtCursorReplicated: considerSnapAtCursorReplicated,
 		promPruneSecs: promPruneSecs,
 	}
@@ -223,7 +224,8 @@ type Report struct {
 type FSReport struct {
 	Filesystem string
 	SnapshotList, DestroyList []SnapshotReport
-	Error string
+	ErrorCount int
+	LastError string
 }
 
 type SnapshotReport struct {
@@ -238,26 +240,17 @@ 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 {
+	if p.execQueue != nil {
-		return &r
+		r.Pending, r.Completed = p.execQueue.Report()
-	}
-
-	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()
 	}
 
 	return &r
@@ -289,14 +282,11 @@ type fs struct {
 	destroyList []pruning.Snapshot
 
 	mtx sync.RWMutex
-	// for Plan
-	err error
-}
 
-func (f *fs) Update(err error) {
+	// only during Exec state, also used by execQueue
-	f.mtx.Lock()
+	execErrLast error
-	defer f.mtx.Unlock()
+	execErrCount int
-	f.err = err
+
 }
 
 func (f *fs) Report() FSReport {
@@ -305,8 +295,9 @@ func (f *fs) Report() FSReport {
 
 	r := FSReport{}
 	r.Filesystem = f.path
-	if f.err != nil {
+	r.ErrorCount = f.execErrCount
-		r.Error = f.err.Error()
+	if f.execErrLast != nil {
+		r.LastError = f.execErrLast.Error()
 	}
 
 	r.SnapshotList = make([]SnapshotReport, len(f.snaps))
@@ -344,6 +335,14 @@ 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 {
 	if neterr, ok := e.(net.Error); ok {
 		return neterr.Temporary()
@@ -372,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 {
@@ -379,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)
@@ -394,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
+		// no progress here since we could run in a live-lock (must have used target AND receiver before progress)
-			continue fsloop
+
-		}
 		pfs.snaps = make([]pruning.Snapshot, 0, len(tfsvs))
 
 		rcReq := &pdu.ReplicationCursorReq{
@@ -411,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
-		}
 		if rc.GetError() != "" {
 			l.WithField("reqErr", rc.GetError()).Error("cannot get replication cursor")
-			pfs.err = fmt.Errorf("%s", rc.GetError())
+			return onErr(u, err)
-			continue fsloop
 		}
+		ka.MadeProgress()
 
 
 		// scan from older to newer, all snapshots older than cursor are interpreted as replicated
@@ -445,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)
+				err := fmt.Errorf("%s%s has invalid creation date: %s", tfs, tfsv.RelName(), err)
-				l.WithError(pfs.err).Error("")
+				l.WithError(err).
-				continue fsloop
+					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()
@@ -459,23 +457,21 @@ fsloop:
 			})
 		}
 		if preCursor {
-			pfs.err = fmt.Errorf("replication cursor not found in prune target filesystem versions")
+			err := fmt.Errorf("replication cursor not found in prune target filesystem versions")
-			l.WithError(pfs.err).Error("")
+			l.Error(err.Error())
-			continue fsloop
+			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.execQueue.Put(pfs, nil, false)
-				pruner.pruneCompleted = append(pruner.pruneCompleted, pfs)
-			} else {
-				pruner.prunePending = append(pruner.prunePending, pfs)
-			}
 		}
 		pruner.state = Exec
 	}).statefunc()
@@ -485,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 pruner.execQueue.HasCompletedFSWithErrors() {
-				if pfs.err != nil {
 				nextState = ErrPerm
 			}
-			}
 			pruner.state = nextState
 			return
 		}
-		pfs = pruner.prunePending[0]
 	})
 	if state != Exec {
 		return state.statefunc()
@@ -509,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)
+	GetLogger(a.ctx).WithField("fs", pfs.path).Debug("destroying snapshots")
-	pfs.Update(err)
+	res, err := a.target.DestroySnapshots(a.ctx, &req)
-	if err != nil && shouldRetry(err) {
+	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.Progress.MadeProgress()
-		pruner.prunePending = pruner.prunePending[1:]
 	}).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

@@ -133,16 +133,14 @@ func TestPruner_Prune(t *testing.T) {
 		},
 		listVersionsErrs: map[string][]error{
 			"zroot/foo": {
-				stubNetErr{msg: "fakeerror1", temporary: true}, // should be classified as temporaty
+				stubNetErr{msg: "fakeerror1", temporary: true},
 				stubNetErr{msg: "fakeerror2", temporary: true,},
 			},
 		},
 		destroyErrs: map[string][]error{
-			"zroot/foo": {
+			"zroot/baz": {
-				fmt.Errorf("permanent error"),
+				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
-			"zroot/bar": {
-				stubNetErr{msg: "fakeerror3", temporary: true},
 			},
 		},
 		destroyed: make(map[string][]string),
@@ -178,9 +176,6 @@ func TestPruner_Prune(t *testing.T) {
 			"zroot/foo": {
 				stubNetErr{msg: "fakeerror4", temporary: true},
 			},
-			"zroot/baz": {
-				fmt.Errorf("permanent error2"),
-			},
 		},
 	}
 
@@ -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)

docs/changelog.rst

@@ -42,13 +42,15 @@ 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.
+  Please make sure the stderr output of the daemon is captured somewhere.
-  Rotation should not be necessary because stderr is not written to under normal circumstances.
   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
 ~~~~~~~
@@ -56,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
@@ -71,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
 
@@ -85,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

@@ -100,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,43 +31,78 @@ 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``
+  * Keeps a short history of local snapshots to enable incremental replication to ``backups``
-  * Accepts connections from ``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``
+  * Accepts connections & responds to requests from ``prod``
-  * Pulls the snapshots to ``storage/zrepl/pull/prod/*``
+  * Limits client ``prod`` access to filesystem sub-tree ``storage/zrepl/sink/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.
 
 Install zrepl
 -------------
 
 Follow the :ref:`OS-specific installation instructions <installation>` and come back here.
 
-Configure server ``backups``
+Generate TLS Certificates
+-------------------------
+
+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 **pull job** named ``pull_prod`` in ``/etc/zrepl/zrepl.yml`` or ``/usr/local/etc/zrepl/zrepl.yml`` on host ``backups`` : ::
+We define a **push job** named ``prod_to_backups`` in ``/etc/zrepl/zrepl.yml`` on host ``prod`` : ::
 
     jobs:
-    - name: pull_prod
+    - name: prod_to_backups
-      type: pull
+      type: push 
       connect:
-        type: tcp
+        type: tls
-        address: "192.168.2.20:2342"
+        address: "backups.example.com:8888"
-      root_fs: "storage/zrepl/pull/prod"
+        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,
+        "zroot/usr/home/paranoid": false
+      }
+      snapshotting:
+        type: periodic
+        prefix: zrepl_
         interval: 10m
       pruning:
         keep_sender:
@@ -78,69 +113,60 @@ We define a **pull job** named ``pull_prod`` in ``/etc/zrepl/zrepl.yml`` or ``/u
         - 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``
 -------------------------
 
-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 define a corresponding **sink job** named ``sink`` in ``/etc/zrepl/zrepl.yml`` on host ``prod`` : ::
 
     jobs:
-    - name: source_backups
+    - name: sink
-      type: source
+      type: sink 
       serve:
-        type: tcp
+         type: tls
-        listen: ":2342"
+         listen: ":8888"
-        clients: {
+         ca: "/etc/zrepl/prod.crt"
-          "192.168.2.10" : "backups"
+         cert: "/etc/zrepl/backups.crt"
-        }
+         key: "/etc/zrepl/backups.key"
-      filesystems: {
+         client_cns:
-        "zroot/var/db:": true,
+           - "prod"
-        "zroot/usr/home<": true,
+      root_fs: "storage/zrepl/sink"
-        "zroot/usr/home/paranoid": false
-      }
-      snapshotting:
-        type: periodic
-        prefix: zrepl_
-        interval: 10m
 
 
-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``.
+We use ``zrepl configcheck`` before to catch any configuration errors: no output indicates that everything is fine.
-This is :ref:`OS-specific <usage-zrepl-daemon-restarting>`.
+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.
-
+To re-trigger replication (snapshots are separate!), use ``zrepl signal wakeup prod_to_backups`` on ``prod``.
-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.
 
 If you like tmux, here is a handy script that works on FreeBSD: ::
 
     pkg install gnu-watch tmux
-    tmux new-window
+    tmux new -s zrepl -d
-    tmux split-window "tail -f /var/log/zrepl.log"
+    tmux split-window -t zrepl "tail -f /var/log/messages"
-    tmux split-window "gnu-watch 'zfs list -t snapshot -o name,creation -s creation | grep zrepl_'"
+    tmux split-window -t zrepl "gnu-watch 'zfs list -t snapshot -o name,creation -s creation | grep zrepl_'"
-    tmux select-layout tiled
+    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 new -s zrepl -d
-    tmux split-window "journalctl -f -u zrepl.service"
+    tmux split-window -t zrepl  "journalctl -f -u zrepl.service"
-    tmux split-window "watch 'zfs list -t snapshot -o name,creation -s creation | grep zrepl_'"
+    tmux split-window -t zrepl "watch 'zfs list -t snapshot -o name,creation -s creation | grep zrepl_'"
-    tmux select-layout tiled
+    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``
+    * - ``zrepl signal wakeup JOB``
-      - manually trigger replication + pruning
+      - 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
 

replication/fsrep/fsfsm.go

@@ -71,38 +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
-	Retry
-	PermanentError
 	Completed
 )
 
-func (s State) fsrsf() state {
+type Error interface {
-	m := map[State]state{
+	error
-		Ready:          stateReady,
+	Temporary() bool
-		Retry:          stateRetry,
+	ContextErr() bool
-		PermanentError: nil,
+	LocalToFS() bool
-		Completed:      nil,
-	}
-	return m[s]
-}
-
-func (s State) IsErrorState() bool {
-	return s & (Retry|PermanentError) != 0
 }
 
 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
-	err                error
 	completed, pending []*ReplicationStep
 }
 
@@ -112,13 +104,35 @@ func (f *Replication) State() State {
 	return f.state
 }
 
-func (f *Replication) Err() error {
+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()
-	if f.state & (Retry|PermanentError) != 0 {
 	return f.err
 }
-	return nil
+
+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 {
@@ -162,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 #
@@ -204,7 +231,7 @@ type ReplicationStep struct {
 	expectedSize int64 // 0 means no size estimate present / possible
 }
 
-func (f *Replication) TakeStep(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver) (post State, nextStepDate 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()
@@ -214,52 +241,46 @@ func (f *Replication) TakeStep(ctx context.Context, ka *watchdog.KeepAlive, send
 		}
 		return f.state
 	}
-	var s state = u(nil).fsrsf()
 
-	pre := u(nil)
-	preTime := time.Now()
-	s = s(ctx, ka, 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()
-	})
-
-	getLogger(ctx).
-		WithField("fs", f.fs).
-		WithField("transition", fmt.Sprintf("%s => %s", pre, post)).
-		WithField("duration", delta).
-		Debug("fsr step taken")
-
-	return post, nextStepDate
-}
-
-type updater func(func(fsr *Replication)) State
-
-type state func(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver, u updater) state
-
-func stateReady(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver, u updater) state {
 
 	var current *ReplicationStep
-	s := u(func(f *Replication) {
+	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 {
+	if st == Completed {
-		return s.fsrsf()
+		return nil
+	}
+	if st != Ready {
+		panic(fmt.Sprintf("implementation error: %v", st))
 	}
 
-	stepState := current.doReplication(ctx, ka, 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) {
+	u(func(fsr *Replication) {
-		switch stepState {
+		if err != nil {
-		case StepCompleted:
+			f.err = &StepError{stepStr: current.String(), err: err}
+			return
+		}
+		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 {
@@ -267,23 +288,61 @@ func stateReady(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, rece
 		} else {
 			f.state = Completed
 		}
-		case StepReplicationRetry:
+	})
-			fallthrough
+	var retErr Error = nil
-		case StepMarkReplicatedRetry:
+	u(func(fsr *Replication) {
-			f.state = Retry
+		retErr = fsr.err
-		case StepPermanentError:
+	})
-			f.state = PermanentError
+	return retErr
-			f.err = errors.New("a replication step failed with a permanent error")
-		default:
-			panic(f)
-		}
-	}).fsrsf()
 }
 
-func stateRetry(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver, u updater) state {
+type updater func(func(fsr *Replication)) State
-	return u(func(fsr *Replication) {
+
-		fsr.state = Ready
+type state func(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver, u updater) state
-	}).fsrsf()
+
+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 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 {
@@ -295,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))
@@ -309,70 +367,48 @@ func (fsr *Replication) Report() *Report {
 		rep.Pending[i] = fsr.pending[i].Report()
 	}
 
-	if fsr.state&Retry != 0 {
-		if len(rep.Pending) != 0 { // should always be true for Retry == true?
-			rep.Problem = rep.Pending[0].Problem
-		}
-	}
-
 	return &rep
 }
 
-func shouldRetry(err error) bool {
+func (s *ReplicationStep) Retry(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver) error {
-	switch err {
+	switch s.state {
-	case io.EOF:
+	case StepReplicationReady:
-		fallthrough
+		return s.doReplication(ctx, ka, sender, receiver)
-	case io.ErrUnexpectedEOF:
+	case StepMarkReplicatedReady:
-		fallthrough
+		return s.doMarkReplicated(ctx, ka, sender)
-	case io.ErrClosedPipe:
+	case StepCompleted:
-		return true
+		return nil
 	}
-	if _, ok := err.(net.Error); ok {
+	panic(fmt.Sprintf("implementation error: %v", s.state))
-		return true
-	}
-	return false
 }
 
-func (s *ReplicationStep) doReplication(ctx context.Context, ka *watchdog.KeepAlive, 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).
+	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
-	}
-
 	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)
@@ -400,42 +436,23 @@ func (s *ReplicationStep) doReplication(ctx context.Context, ka *watchdog.KeepAl
 		// 	- 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()
+	s.state = StepMarkReplicatedReady
-
 	return s.doMarkReplicated(ctx, ka, sender)
 
 }
 
-func (s *ReplicationStep) doMarkReplicated(ctx context.Context, ka *watchdog.KeepAlive, sender Sender) StepState {
+func (s *ReplicationStep) doMarkReplicated(ctx context.Context, ka *watchdog.KeepAlive, sender Sender) error {
 
-	log := getLogger(ctx).
+	if s.state != StepMarkReplicatedReady {
-		WithField("filesystem", s.parent.fs).
+		panic(fmt.Sprintf("implementation error: %v", s.state))
-		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
 	}
 
-	updateStateCompleted := func() StepState {
+	log := getLogger(ctx)
-		s.lock.Lock()
-		defer s.lock.Unlock()
-		s.state = StepCompleted
-		return s.state
-	}
 
 	log.Debug("advance replication cursor")
 	req := &pdu.ReplicationCursorReq{
@@ -449,25 +466,22 @@ func (s *ReplicationStep) doMarkReplicated(ctx context.Context, ka *watchdog.Kee
 	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)
-
-	log := getLogger(ctx).
-		WithField("filesystem", fs).
-		WithField("step", s.String())
 
 	sr := s.buildSendRequest(true)
 

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 = "ReadyRetry"
-	_State_name_1 = "PermanentError"
-	_State_name_2 = "Completed"
-)
-
-var (
-	_State_index_0 = [...]uint8{0, 5, 10}
-)
-
-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,122 +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
-	errorStateEnterCount int
-
-	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)
-		},
-	},
-	Retry: {
-		prio: 1,
-		less: func(a, b *replicationQueueItem) bool {
-			return a.errorStateEnterCount < b.errorStateEnterCount
-		},
-	},
-}
-
-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 time.Time) {
-	h.i.state = newState
-	h.i.nextStepDate = nextStepDate
-	if h.i.state.IsErrorState() {
-		h.i.errorStateEnterCount++
-	}
-}

replication/mainfsm.go

@@ -10,14 +10,15 @@ import (
 	"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"
 )
 
@@ -70,9 +71,9 @@ type Replication struct {
 	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
 
 	// for PlanningError, WorkingWait and ContextError and Completed
 	err error
@@ -87,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 {
@@ -193,17 +194,22 @@ func resolveConflict(conflict error) (path []*pdu.FilesystemVersion, msg string)
 	return nil, "no automated way to handle conflict type"
 }
 
-var RetryInterval = envconst.Duration("ZREPL_REPLICATION_RETRY_INTERVAL", 4 * time.Second)
+var RetryInterval = envconst.Duration("ZREPL_REPLICATION_RETRY_INTERVAL", 10 * time.Second)
+
+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 {
-	switch err {
+	if e, ok := err.(Error); ok {
-	case context.Canceled: return true
+		return !e.Temporary()
-	case context.DeadlineExceeded: return true
 	}
-	if operr, ok := err.(net.Error); ok {
+	return true
-		return !operr.Temporary()
-	}
-	return false
 }
 
 func statePlanning(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, receiver Receiver, u updater) state {
@@ -214,8 +220,10 @@ func statePlanning(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, r
 
 	handlePlanningError := func(err error) state {
 		return u(func(r *Replication) {
-			r.err = err
+			ge := GlobalError{Err: err, Temporary: !isPermanent(err)}
-			if 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)
@@ -229,6 +237,7 @@ func statePlanning(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, r
 		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 {
@@ -236,7 +245,9 @@ func statePlanning(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, r
 		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 {
 
@@ -249,11 +260,12 @@ func statePlanning(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, r
 			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
 		}
 
@@ -278,6 +290,7 @@ func statePlanning(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, r
 		} else {
 			rfsvs = []*pdu.FilesystemVersion{}
 		}
+		ka.MadeProgress()
 
 		path, conflict := IncrementalPath(rfsvs, sfsvs)
 		if conflict != nil {
@@ -291,8 +304,9 @@ func statePlanning(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, r
 				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
 		}
 
@@ -309,14 +323,16 @@ func statePlanning(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, r
 			}
 		}
 		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)
 		}
+		ka.MadeProgress()
 
-		q.Add(qitem)
+		q = append(q, qitem)
 	}
 
 	ka.MadeProgress()
@@ -351,48 +367,125 @@ func statePlanningError(ctx context.Context, ka *watchdog.KeepAlive, sender Send
 	}).rsf()
 }
 
+type GlobalError struct {
+	Err       error
+	Temporary bool
+}
+
+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 *ReplicationQueueItemHandle
+	var active *fsrep.Replication
 	rsfNext := u(func(r *Replication) {
-		done, next := r.queue.GetNext()
+
-		r.completed = append(r.completed, done...)
+		r.err = nil
-		if next == nil {
+
-			r.state = Completed
+		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
 	}
 
-	state, nextStepDate := active.GetFSReplication().TakeStep(ctx, ka, sender, receiver)
+	activeCtx := fsrep.WithLogger(ctx, getLogger(ctx).WithField("fs", active.FS()))
+	err := active.Retry(activeCtx, ka, sender, receiver)
 	u(func(r *Replication) {
-		active.Update(state, nextStepDate)
 		r.active = nil
 	}).rsf()
 
-	select {
+	if err != nil {
-	case <-ctx.Done():
+		if err.ContextErr() && ctx.Err() != nil {
-		return u(func(r *Replication) {
+			getLogger(ctx).WithError(err).
-			r.err = ctx.Err()
+				Info("filesystem replication was cancelled")
+			u(func(r*Replication) {
+				r.err = GlobalError{Err: err, Temporary: false}
 				r.state = PermanentError
-		}).rsf()
+			})
-	default:
+		} else if err.LocalToFS() {
-	}
+			getLogger(ctx).WithError(err).
-
+				Error("filesystem replication encountered a filesystem-specific error")
-	if err := active.GetFSReplication().Err(); err != nil {
+			// we stay in this state and let the queuing logic above de-prioritize this failing FS
-		return u(func(r *Replication) {
+		} else if err.Temporary() {
-			r.err = err
+			getLogger(ctx).WithError(err).
-			if isPermanent(err) {
+				Error("filesystem encountered a non-filesystem-specific temporary error, enter retry-wait")
-				r.state = PermanentError
+			u(func(r *Replication) {
-			} else {
+				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()
@@ -404,7 +497,7 @@ func stateWorkingWait(ctx context.Context, ka *watchdog.KeepAlive, sender Sender
 		sleepUntil = r.sleepUntil
 	})
 	t := time.NewTimer(RetryInterval)
-	getLogger(ctx).WithField("until", sleepUntil).Info("retry wait after replication step error")
+	getLogger(ctx).WithField("until", sleepUntil).Info("retry wait after error")
 	defer t.Stop()
 	select {
 	case <-ctx.Done():
@@ -433,27 +526,26 @@ func (r *Replication) Report() *Report {
 		SleepUntil: r.sleepUntil,
 	}
 
-	if r.state&(Planning|PlanningError|PermanentError) != 0 {
 	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 = r.active.Report()
-		rep.Active = active.Report()
+		pending = r.queue[1:]
 	}
-	r.queue.Foreach(func(h *ReplicationQueueItemHandle) {
+	for _, fsr := range pending {
-		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())
 	}

util/watchdog/watchdog.go

@@ -6,36 +6,26 @@ import (
 	"time"
 )
 
-type Progress struct {
+type KeepAlive struct {
+	mtx sync.Mutex
 	lastUpd time.Time
 }
 
-func (p *Progress) String() string {
+func (p *KeepAlive) String() string {
+	if p.lastUpd.IsZero() {
+		return fmt.Sprintf("never updated")
+	}
 	return fmt.Sprintf("last update at %s", p.lastUpd)
 }
 
-func (p *Progress) madeProgressSince(p2 *Progress) bool {
-	if p.lastUpd.IsZero() && p2.lastUpd.IsZero() {
-		return false
-	}
-	return p.lastUpd.After(p2.lastUpd)
-}
-
-type KeepAlive struct {
-	mtx sync.Mutex
-	p Progress
-}
-
 func (k *KeepAlive) MadeProgress() {
 	k.mtx.Lock()
 	defer k.mtx.Unlock()
-	k.p.lastUpd = time.Now()
+	k.lastUpd = time.Now()
 }
 
-func (k *KeepAlive) ExpectProgress(last *Progress) (madeProgress bool) {
+func (k *KeepAlive) CheckTimeout(timeout time.Duration, jitter time.Duration) (didTimeOut bool) {
 	k.mtx.Lock()
 	defer k.mtx.Unlock()
-	madeProgress = k.p.madeProgressSince(last)
+	return k.lastUpd.Add(timeout - jitter).Before(time.Now())
-	*last = k.p
-	return madeProgress
 }