zfs: rewrite SendStream, fix bug in Close() on FreeBSD, add platformtests

This commit was motivated by https://github.com/zrepl/zrepl/issues/495 where, on FreeBSD with OpenZFS 2.0, a SendStream.Close() call might wait indefinitely for `zfs send` to exit. The reason is that, due to the refactoring done for redacted send & recv (https://github.com/openzfs/zfs/commit/30af21b02569ac192f52ce6e6511015f8a8d5729), the `dump_bytes` function, which writes to the pipe, executes in a separate thread (synctask taskq) iff not `HAVE_LARGE_STACKS`. The `zfs send` process/thread waits for that taskq thread using an uninterruptible primitive. So when we SIGKILL `zfs send`, that signal doesn't reach the right thread to interrupt the pipe write. Theoretically this affects both Linux and FreeBSD, but most Linux users `HAVE_LARGE_STACKS` and since https://github.com/penzfs/zfs/pull/12350/files OpenZFS on FreeBSD `HAVE_LARGE_STACKS` as well. However, at least until FreeBSD 13.1, possibly for the entire 13 lifecycle, we're going to have to live with that oddity. Measures taken in this commit: - Report the behavior as an upstream bug https://github.com/openzfs/zfs/issues/12500 - Change SendStream code so that it closes zrepl's read-end of the pipe (see comment in code) - Clean up and make explicit SendStream's state handling - Write extensive platformtests for SendStream - They pass on my Linux install and on FreeBSD 12 - FreeBSD 13 still needs testing. fixes https://github.com/zrepl/zrepl/issues/495
platformtest: fix 'active child tasks' panic for ReceiveForceRollbackWorksUnencrypted
2021-08-23 00:21:34 +02:00 · 2021-08-23 00:21:34 +02:00 · 2021-08-23 00:21:34 +02:00 · 2021-08-23 00:21:34 +02:00 · 2021-07-08 19:46:24 +02:00 · 2021-07-08 17:07:24 +02:00
36 changed files with 2206 additions and 945 deletions
@@ -358,7 +358,7 @@ jobs:
          subcommand: docdep
      - run:
          command: |
-            git config --global user.email "me@cschwarz.com"
+            git config --global user.email "zreplbot@cschwarz.com"
            git config --global user.name "zrepl-github-io-ci"

      # https://circleci.com/docs/2.0/add-ssh-key/#adding-multiple-keys-with-blank-hostnames
@@ -1,816 +0,0 @@
-package client
-
-import (
-	"context"
-	"fmt"
-	"io"
-	"math"
-	"net/http"
-	"os"
-	"sort"
-	"strings"
-	"sync"
-	"time"
-
-	// tcell is the termbox-compatible library for abstracting away escape sequences, etc.
-	// as of tcell#252, the number of default distributed terminals is relatively limited
-	// additional terminal definitions can be included via side-effect import
-	// See https://github.com/gdamore/tcell/blob/master/terminfo/base/base.go
-	// See https://github.com/gdamore/tcell/issues/252#issuecomment-533836078
-	"github.com/gdamore/tcell/termbox"
-	_ "github.com/gdamore/tcell/terminfo/s/screen" // tmux on FreeBSD 11 & 12 without ncurses
-
-	"github.com/pkg/errors"
-	"github.com/spf13/pflag"
-	"github.com/zrepl/yaml-config"
-
-	"github.com/zrepl/zrepl/cli"
-	"github.com/zrepl/zrepl/daemon"
-	"github.com/zrepl/zrepl/daemon/job"
-	"github.com/zrepl/zrepl/daemon/pruner"
-	"github.com/zrepl/zrepl/daemon/snapper"
-	"github.com/zrepl/zrepl/replication/report"
-)
-
-type byteProgressMeasurement struct {
-	time time.Time
-	val  int64
-}
-
-type bytesProgressHistory struct {
-	last        *byteProgressMeasurement // pointer as poor man's optional
-	changeCount int
-	lastChange  time.Time
-	bpsAvg      float64
-}
-
-func (p *bytesProgressHistory) Update(currentVal int64) (bytesPerSecondAvg int64, changeCount int) {
-
-	if p.last == nil {
-		p.last = &byteProgressMeasurement{
-			time: time.Now(),
-			val:  currentVal,
-		}
-		return 0, 0
-	}
-
-	if p.last.val != currentVal {
-		p.changeCount++
-		p.lastChange = time.Now()
-	}
-
-	if time.Since(p.lastChange) > 3*time.Second {
-		p.last = nil
-		return 0, 0
-	}
-
-	deltaV := currentVal - p.last.val
-	deltaT := time.Since(p.last.time)
-	rate := float64(deltaV) / deltaT.Seconds()
-
-	factor := 0.3
-	p.bpsAvg = (1-factor)*p.bpsAvg + factor*rate
-
-	p.last.time = time.Now()
-	p.last.val = currentVal
-
-	return int64(p.bpsAvg), p.changeCount
-}
-
-type tui struct {
-	x, y   int
-	indent int
-
-	lock   sync.Mutex //For report and error
-	report map[string]*job.Status
-	err    error
-
-	jobFilter string
-
-	replicationProgress map[string]*bytesProgressHistory // by job name
-}
-
-func newTui() tui {
-	return tui{
-		replicationProgress: make(map[string]*bytesProgressHistory),
-	}
-}
-
-const INDENT_MULTIPLIER = 4
-
-func (t *tui) moveLine(dl int, col int) {
-	t.y += dl
-	t.x = t.indent*INDENT_MULTIPLIER + col
-}
-
-func (t *tui) write(text string) {
-	for _, c := range text {
-		if c == '\n' {
-			t.newline()
-			continue
-		}
-		termbox.SetCell(t.x, t.y, c, termbox.ColorDefault, termbox.ColorDefault)
-		t.x += 1
-	}
-}
-
-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.TrimRight(s[:rem], "\n\r")
-		s = s[rem:]
-		if len(untilNewline) == 0 {
-			continue
-		}
-		b.WriteString(untilNewline)
-		b.WriteString("\n")
-	}
-	return strings.TrimRight(b.String(), "\n\r")
-}
-
-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)
-}
-
-func (t *tui) setIndent(indent int) {
-	t.indent = indent
-	t.moveLine(0, 0)
-}
-
-func (t *tui) addIndent(indent int) {
-	t.indent += indent
-	t.moveLine(0, 0)
-}
-
-var statusFlags struct {
-	Raw bool
-	Job string
-}
-
-var StatusCmd = &cli.Subcommand{
-	Use:   "status",
-	Short: "show job activity or dump as JSON for monitoring",
-	SetupFlags: func(f *pflag.FlagSet) {
-		f.BoolVar(&statusFlags.Raw, "raw", false, "dump raw status description from zrepl daemon")
-		f.StringVar(&statusFlags.Job, "job", "", "only dump specified job")
-	},
-	Run: runStatus,
-}
-
-func runStatus(ctx context.Context, s *cli.Subcommand, args []string) error {
-	httpc, err := controlHttpClient(s.Config().Global.Control.SockPath)
-	if err != nil {
-		return err
-	}
-
-	if statusFlags.Raw {
-		resp, err := httpc.Get("http://unix" + daemon.ControlJobEndpointStatus)
-		if err != nil {
-			return err
-		}
-		defer resp.Body.Close()
-		if resp.StatusCode != http.StatusOK {
-			fmt.Fprintf(os.Stderr, "Received error response:\n")
-			_, err := io.CopyN(os.Stderr, resp.Body, 4096)
-			if err != nil {
-				return err
-			}
-			return errors.Errorf("exit")
-		}
-		if _, err := io.Copy(os.Stdout, resp.Body); err != nil {
-			return err
-		}
-		return nil
-	}
-
-	t := newTui()
-	t.lock.Lock()
-	t.err = errors.New("Got no report yet")
-	t.lock.Unlock()
-	t.jobFilter = statusFlags.Job
-
-	err = termbox.Init()
-	if err != nil {
-		return err
-	}
-	defer termbox.Close()
-
-	update := func() {
-		var m daemon.Status
-
-		err2 := jsonRequestResponse(httpc, daemon.ControlJobEndpointStatus,
-			struct{}{},
-			&m,
-		)
-
-		t.lock.Lock()
-		t.err = err2
-		t.report = m.Jobs
-		t.lock.Unlock()
-		t.draw()
-	}
-	update()
-
-	ticker := time.NewTicker(500 * time.Millisecond)
-	defer ticker.Stop()
-	go func() {
-		for range ticker.C {
-			update()
-		}
-	}()
-
-	termbox.HideCursor()
-	termbox.Clear(termbox.ColorDefault, termbox.ColorDefault)
-
-loop:
-	for {
-		switch ev := termbox.PollEvent(); ev.Type {
-		case termbox.EventKey:
-			switch ev.Key {
-			case termbox.KeyEsc:
-				break loop
-			case termbox.KeyCtrlC:
-				break loop
-			}
-		case termbox.EventResize:
-			t.draw()
-		}
-	}
-
-	return nil
-
-}
-
-func (t *tui) getReplicationProgressHistory(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()
-
-	termbox.Clear(termbox.ColorDefault, termbox.ColorDefault)
-	t.x = 0
-	t.y = 0
-	t.indent = 0
-
-	if t.err != nil {
-		t.write(t.err.Error())
-	} else {
-		//Iterate over map in alphabetical order
-		keys := make([]string, 0, len(t.report))
-		for k := range t.report {
-			if len(k) == 0 || daemon.IsInternalJobName(k) { //Internal job
-				continue
-			}
-			if t.jobFilter != "" && k != t.jobFilter {
-				continue
-			}
-			keys = append(keys, k)
-		}
-		sort.Strings(keys)
-
-		if len(keys) == 0 {
-			t.setIndent(0)
-			t.printf("no jobs to display")
-			t.newline()
-			termbox.Flush()
-			return
-		}
-
-		for _, k := range keys {
-			v := t.report[k]
-
-			t.setIndent(0)
-
-			t.printf("Job: %s", k)
-			t.setIndent(1)
-			t.newline()
-			t.printf("Type: %s", v.Type)
-			t.setIndent(1)
-			t.newline()
-
-			if v.Type == job.TypePush || v.Type == job.TypePull {
-				activeStatus, ok := v.JobSpecific.(*job.ActiveSideStatus)
-				if !ok || activeStatus == nil {
-					t.printf("ActiveSideStatus is null")
-					t.newline()
-					continue
-				}
-
-				t.printf("Replication:")
-				t.newline()
-				t.addIndent(1)
-				t.renderReplicationReport(activeStatus.Replication, t.getReplicationProgressHistory(k))
-				t.addIndent(-1)
-
-				t.printf("Pruning Sender:")
-				t.newline()
-				t.addIndent(1)
-				t.renderPrunerReport(activeStatus.PruningSender)
-				t.addIndent(-1)
-
-				t.printf("Pruning Receiver:")
-				t.newline()
-				t.addIndent(1)
-				t.renderPrunerReport(activeStatus.PruningReceiver)
-				t.addIndent(-1)
-
-				if v.Type == job.TypePush {
-					t.printf("Snapshotting:")
-					t.newline()
-					t.addIndent(1)
-					t.renderSnapperReport(activeStatus.Snapshotting)
-					t.addIndent(-1)
-				}
-
-			} else if v.Type == job.TypeSnap {
-				snapStatus, ok := v.JobSpecific.(*job.SnapJobStatus)
-				if !ok || snapStatus == nil {
-					t.printf("SnapJobStatus is null")
-					t.newline()
-					continue
-				}
-				t.printf("Pruning snapshots:")
-				t.newline()
-				t.addIndent(1)
-				t.renderPrunerReport(snapStatus.Pruning)
-				t.addIndent(-1)
-				t.printf("Snapshotting:")
-				t.newline()
-				t.addIndent(1)
-				t.renderSnapperReport(snapStatus.Snapshotting)
-				t.addIndent(-1)
-			} else if v.Type == job.TypeSource {
-
-				st := v.JobSpecific.(*job.PassiveStatus)
-				t.printf("Snapshotting:\n")
-				t.addIndent(1)
-				t.renderSnapperReport(st.Snapper)
-				t.addIndent(-1)
-
-			} else {
-				t.printf("No status representation for job type '%s', dumping as YAML", v.Type)
-				t.newline()
-				asYaml, err := yaml.Marshal(v.JobSpecific)
-				if err != nil {
-					t.printf("Error marshaling status to YAML: %s", err)
-					t.newline()
-					continue
-				}
-				t.write(string(asYaml))
-				t.newline()
-				continue
-			}
-		}
-	}
-	termbox.Flush()
-}
-
-func (t *tui) renderReplicationReport(rep *report.Report, history *bytesProgressHistory) {
-	if rep == nil {
-		t.printf("...\n")
-		return
-	}
-
-	if rep.WaitReconnectError != nil {
-		t.printfDrawIndentedAndWrappedIfMultiline("Connectivity: %s", rep.WaitReconnectError)
-		t.newline()
-	}
-	if !rep.WaitReconnectSince.IsZero() {
-		delta := time.Until(rep.WaitReconnectUntil).Round(time.Second)
-		if rep.WaitReconnectUntil.IsZero() || delta > 0 {
-			var until string
-			if rep.WaitReconnectUntil.IsZero() {
-				until = "waiting indefinitely"
-			} else {
-				until = fmt.Sprintf("hard fail in %s @ %s", delta, rep.WaitReconnectUntil)
-			}
-			t.printfDrawIndentedAndWrappedIfMultiline("Connectivity: reconnecting with exponential backoff (since %s) (%s)",
-				rep.WaitReconnectSince, until)
-		} else {
-			t.printfDrawIndentedAndWrappedIfMultiline("Connectivity: reconnects reached hard-fail timeout @ %s", rep.WaitReconnectUntil)
-		}
-		t.newline()
-	}
-
-	// TODO visualize more than the latest attempt by folding all attempts into one
-	if len(rep.Attempts) == 0 {
-		t.printf("no attempts made yet")
-		return
-	} else {
-		t.printf("Attempt #%d", len(rep.Attempts))
-		if len(rep.Attempts) > 1 {
-			t.printf(". Previous attempts failed with the following statuses:")
-			t.newline()
-			t.addIndent(1)
-			for i, a := range rep.Attempts[:len(rep.Attempts)-1] {
-				t.printfDrawIndentedAndWrappedIfMultiline("#%d: %s (failed at %s) (ran %s)", i+1, a.State, a.FinishAt, a.FinishAt.Sub(a.StartAt))
-				t.newline()
-			}
-			t.addIndent(-1)
-		} else {
-			t.newline()
-		}
-	}
-
-	latest := rep.Attempts[len(rep.Attempts)-1]
-	sort.Slice(latest.Filesystems, func(i, j int) bool {
-		return latest.Filesystems[i].Info.Name < latest.Filesystems[j].Info.Name
-	})
-
-	t.printf("Status: %s", latest.State)
-	t.newline()
-	if latest.State == report.AttemptPlanningError {
-		t.printf("Problem: ")
-		t.printfDrawIndentedAndWrappedIfMultiline("%s", latest.PlanError)
-		t.newline()
-	} else if latest.State == report.AttemptFanOutError {
-		t.printf("Problem: one or more of the filesystems encountered errors")
-		t.newline()
-	}
-
-	if latest.State != report.AttemptPlanning && latest.State != report.AttemptPlanningError {
-		// Draw global progress bar
-		// Progress: [---------------]
-		expected, replicated, containsInvalidSizeEstimates := latest.BytesSum()
-		rate, changeCount := history.Update(replicated)
-		eta := time.Duration(0)
-		if rate > 0 {
-			eta = time.Duration((expected-replicated)/rate) * time.Second
-		}
-		t.write("Progress: ")
-		t.drawBar(50, replicated, expected, changeCount)
-		t.write(fmt.Sprintf(" %s / %s @ %s/s", ByteCountBinary(replicated), ByteCountBinary(expected), ByteCountBinary(rate)))
-		if eta != 0 {
-			t.write(fmt.Sprintf(" (%s remaining)", humanizeDuration(eta)))
-		}
-		t.newline()
-		if containsInvalidSizeEstimates {
-			t.write("NOTE: not all steps could be size-estimated, total estimate is likely imprecise!")
-			t.newline()
-		}
-
-		if len(latest.Filesystems) == 0 {
-			t.write("NOTE: no filesystems were considered for replication!")
-			t.newline()
-		}
-
-		var maxFSLen int
-		for _, fs := range latest.Filesystems {
-			if len(fs.Info.Name) > maxFSLen {
-				maxFSLen = len(fs.Info.Name)
-			}
-		}
-		for _, fs := range latest.Filesystems {
-			t.printFilesystemStatus(fs, false, maxFSLen) // FIXME bring 'active' flag back
-		}
-
-	}
-
-}
-
-func (t *tui) renderPrunerReport(r *pruner.Report) {
-	if r == nil {
-		t.printf("...\n")
-		return
-	}
-
-	state, err := pruner.StateString(r.State)
-	if err != nil {
-		t.printf("Status: %q (parse error: %q)\n", r.State, err)
-		return
-	}
-
-	t.printf("Status: %s", state)
-	t.newline()
-
-	if r.Error != "" {
-		t.printf("Error: %s\n", r.Error)
-	}
-
-	type commonFS struct {
-		*pruner.FSReport
-		completed bool
-	}
-	all := make([]commonFS, 0, len(r.Pending)+len(r.Completed))
-	for i := range r.Pending {
-		all = append(all, commonFS{&r.Pending[i], false})
-	}
-	for i := range r.Completed {
-		all = append(all, commonFS{&r.Completed[i], true})
-	}
-
-	switch state {
-	case pruner.Plan:
-		fallthrough
-	case pruner.PlanErr:
-		return
-	}
-
-	if len(all) == 0 {
-		t.printf("nothing to do\n")
-		return
-	}
-
-	var totalDestroyCount, completedDestroyCount int
-	var maxFSname int
-	for _, fs := range all {
-		totalDestroyCount += len(fs.DestroyList)
-		if fs.completed {
-			completedDestroyCount += len(fs.DestroyList)
-		}
-		if maxFSname < len(fs.Filesystem) {
-			maxFSname = len(fs.Filesystem)
-		}
-	}
-
-	// global progress bar
-	progress := int(math.Round(80 * float64(completedDestroyCount) / float64(totalDestroyCount)))
-	t.write("Progress: ")
-	t.write("[")
-	t.write(times("=", progress))
-	t.write(">")
-	t.write(times("-", 80-progress))
-	t.write("]")
-	t.printf(" %d/%d snapshots", completedDestroyCount, totalDestroyCount)
-	t.newline()
-
-	sort.SliceStable(all, func(i, j int) bool {
-		return strings.Compare(all[i].Filesystem, all[j].Filesystem) == -1
-	})
-
-	// Draw a table-like representation of 'all'
-	for _, fs := range all {
-		t.write(rightPad(fs.Filesystem, maxFSname, " "))
-		t.write(" ")
-		if !fs.SkipReason.NotSkipped() {
-			t.printf("skipped: %s\n", fs.SkipReason)
-			continue
-		}
-		if fs.LastError != "" {
-			if strings.ContainsAny(fs.LastError, "\r\n") {
-				t.printf("ERROR:")
-				t.printfDrawIndentedAndWrappedIfMultiline("%s\n", fs.LastError)
-			} else {
-				t.printfDrawIndentedAndWrappedIfMultiline("ERROR: %s\n", fs.LastError)
-			}
-			t.newline()
-			continue
-		}
-
-		pruneRuleActionStr := fmt.Sprintf("(destroy %d of %d snapshots)",
-			len(fs.DestroyList), len(fs.SnapshotList))
-
-		if fs.completed {
-			t.printf("Completed  %s\n", pruneRuleActionStr)
-			continue
-		}
-
-		t.write("Pending    ") // whitespace is padding 10
-		if len(fs.DestroyList) == 1 {
-			t.write(fs.DestroyList[0].Name)
-		} else {
-			t.write(pruneRuleActionStr)
-		}
-		t.newline()
-	}
-
-}
-
-func (t *tui) renderSnapperReport(r *snapper.Report) {
-	if r == nil {
-		t.printf("<snapshot type does not have a report>\n")
-		return
-	}
-
-	t.printf("Status: %s", r.State)
-	t.newline()
-
-	if r.Error != "" {
-		t.printf("Error: %s\n", r.Error)
-	}
-	if !r.SleepUntil.IsZero() {
-		t.printf("Sleep until: %s\n", r.SleepUntil)
-	}
-
-	sort.Slice(r.Progress, func(i, j int) bool {
-		return strings.Compare(r.Progress[i].Path, r.Progress[j].Path) == -1
-	})
-
-	t.addIndent(1)
-	defer t.addIndent(-1)
-	dur := func(d time.Duration) string {
-		return d.Round(100 * time.Millisecond).String()
-	}
-
-	type row struct {
-		path, state, duration, remainder, hookReport string
-	}
-	var widths struct {
-		path, state, duration int
-	}
-	rows := make([]*row, len(r.Progress))
-	for i, fs := range r.Progress {
-		r := &row{
-			path:  fs.Path,
-			state: fs.State.String(),
-		}
-		if fs.HooksHadError {
-			r.hookReport = fs.Hooks // FIXME render here, not in daemon
-		}
-		switch fs.State {
-		case snapper.SnapPending:
-			r.duration = "..."
-			r.remainder = ""
-		case snapper.SnapStarted:
-			r.duration = dur(time.Since(fs.StartAt))
-			r.remainder = fmt.Sprintf("snap name: %q", fs.SnapName)
-		case snapper.SnapDone:
-			fallthrough
-		case snapper.SnapError:
-			r.duration = dur(fs.DoneAt.Sub(fs.StartAt))
-			r.remainder = fmt.Sprintf("snap name: %q", fs.SnapName)
-		}
-		rows[i] = r
-		if len(r.path) > widths.path {
-			widths.path = len(r.path)
-		}
-		if len(r.state) > widths.state {
-			widths.state = len(r.state)
-		}
-		if len(r.duration) > widths.duration {
-			widths.duration = len(r.duration)
-		}
-	}
-
-	for _, r := range rows {
-		path := rightPad(r.path, widths.path, " ")
-		state := rightPad(r.state, widths.state, " ")
-		duration := rightPad(r.duration, widths.duration, " ")
-		t.printf("%s %s %s", path, state, duration)
-		t.printfDrawIndentedAndWrappedIfMultiline(" %s", r.remainder)
-		if r.hookReport != "" {
-			t.printfDrawIndentedAndWrappedIfMultiline("%s", r.hookReport)
-		}
-		t.newline()
-	}
-
-}
-
-func times(str string, n int) (out string) {
-	for i := 0; i < n; i++ {
-		out += str
-	}
-	return
-}
-
-func rightPad(str string, length int, pad string) string {
-	if len(str) > length {
-		return str[:length]
-	}
-	return str + strings.Repeat(pad, length-len(str))
-}
-
-var arrowPositions = `>\|/`
-
-// changeCount = 0 indicates stall / no progress
-func (t *tui) drawBar(length int, bytes, totalBytes int64, changeCount int) {
-	var completedLength int
-	if totalBytes > 0 {
-		completedLength = int(int64(length) * bytes / totalBytes)
-		if completedLength > length {
-			completedLength = length
-		}
-	} else if totalBytes == bytes {
-		completedLength = length
-	}
-
-	t.write("[")
-	t.write(times("=", completedLength))
-	t.write(string(arrowPositions[changeCount%len(arrowPositions)]))
-	t.write(times("-", length-completedLength))
-	t.write("]")
-}
-
-func (t *tui) printFilesystemStatus(rep *report.FilesystemReport, active bool, maxFS int) {
-
-	expected, replicated, containsInvalidSizeEstimates := rep.BytesSum()
-	sizeEstimationImpreciseNotice := ""
-	if containsInvalidSizeEstimates {
-		sizeEstimationImpreciseNotice = " (some steps lack size estimation)"
-	}
-	if rep.CurrentStep < len(rep.Steps) && rep.Steps[rep.CurrentStep].Info.BytesExpected == 0 {
-		sizeEstimationImpreciseNotice = " (step lacks size estimation)"
-	}
-
-	status := fmt.Sprintf("%s (step %d/%d, %s/%s)%s",
-		strings.ToUpper(string(rep.State)),
-		rep.CurrentStep, len(rep.Steps),
-		ByteCountBinary(replicated), ByteCountBinary(expected),
-		sizeEstimationImpreciseNotice,
-	)
-
-	activeIndicator := " "
-	if active {
-		activeIndicator = "*"
-	}
-	t.printf("%s %s %s ",
-		activeIndicator,
-		rightPad(rep.Info.Name, maxFS, " "),
-		status)
-
-	next := ""
-	if err := rep.Error(); err != nil {
-		next = err.Err
-	} else if rep.State != report.FilesystemDone {
-		if nextStep := rep.NextStep(); nextStep != nil {
-			if nextStep.IsIncremental() {
-				next = fmt.Sprintf("next: %s => %s", nextStep.Info.From, nextStep.Info.To)
-			} else {
-				next = fmt.Sprintf("next: full send %s", nextStep.Info.To)
-			}
-			attribs := []string{}
-
-			if nextStep.Info.Resumed {
-				attribs = append(attribs, "resumed")
-			}
-
-			attribs = append(attribs, fmt.Sprintf("encrypted=%s", nextStep.Info.Encrypted))
-
-			next += fmt.Sprintf(" (%s)", strings.Join(attribs, ", "))
-		} else {
-			next = "" // individual FSes may still be in planning state
-		}
-
-	}
-	t.printfDrawIndentedAndWrappedIfMultiline("%s", next)
-
-	t.newline()
-}
-
-func ByteCountBinary(b int64) string {
-	const unit = 1024
-	if b < unit {
-		return fmt.Sprintf("%d B", b)
-	}
-	div, exp := int64(unit), 0
-	for n := b / unit; n >= unit; n /= unit {
-		div *= unit
-		exp++
-	}
-	return fmt.Sprintf("%.1f %ciB", float64(b)/float64(div), "KMGTPE"[exp])
-}
-
-func humanizeDuration(duration time.Duration) string {
-	days := int64(duration.Hours() / 24)
-	hours := int64(math.Mod(duration.Hours(), 24))
-	minutes := int64(math.Mod(duration.Minutes(), 60))
-	seconds := int64(math.Mod(duration.Seconds(), 60))
-
-	var parts []string
-
-	force := false
-	chunks := []int64{days, hours, minutes, seconds}
-	for i, chunk := range chunks {
-		if force || chunk > 0 {
-			padding := 0
-			if force {
-				padding = 2
-			}
-			parts = append(parts, fmt.Sprintf("%*d%c", padding, chunk, "dhms"[i]))
-			force = true
-		}
-	}
-
-	return strings.Join(parts, " ")
-}
@@ -0,0 +1,101 @@
+package client
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"io"
+	"net"
+	"net/http"
+
+	"github.com/pkg/errors"
+
+	"github.com/zrepl/zrepl/daemon"
+)
+
+type Client struct {
+	h http.Client
+}
+
+func New(network, addr string) (*Client, error) {
+	httpc, err := controlHttpClient(func(_ context.Context) (net.Conn, error) { return net.Dial(network, addr) })
+	if err != nil {
+		return nil, err
+	}
+	return &Client{httpc}, nil
+}
+
+func (c *Client) Status() (s daemon.Status, _ error) {
+	err := jsonRequestResponse(c.h, daemon.ControlJobEndpointStatus,
+		struct{}{},
+		&s,
+	)
+	return s, err
+}
+
+func (c *Client) StatusRaw() ([]byte, error) {
+	var r json.RawMessage
+	err := jsonRequestResponse(c.h, daemon.ControlJobEndpointStatus, struct{}{}, &r)
+	if err != nil {
+		return nil, err
+	}
+	return r, nil
+}
+
+func (c *Client) signal(job, sig string) error {
+	return jsonRequestResponse(c.h, daemon.ControlJobEndpointSignal,
+		struct {
+			Name string
+			Op   string
+		}{
+			Name: job,
+			Op:   sig,
+		},
+		struct{}{},
+	)
+}
+
+func (c *Client) SignalWakeup(job string) error {
+	return c.signal(job, "wakeup")
+}
+
+func (c *Client) SignalReset(job string) error {
+	return c.signal(job, "reset")
+}
+
+func controlHttpClient(dialfunc func(context.Context) (net.Conn, error)) (client http.Client, err error) {
+	return http.Client{
+		Transport: &http.Transport{
+			DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+				return dialfunc(ctx)
+			},
+		},
+	}, nil
+}
+
+func jsonRequestResponse(c http.Client, endpoint string, req interface{}, res interface{}) error {
+	var buf bytes.Buffer
+	encodeErr := json.NewEncoder(&buf).Encode(req)
+	if encodeErr != nil {
+		return encodeErr
+	}
+
+	resp, err := c.Post("http://unix"+endpoint, "application/json", &buf)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		var msg bytes.Buffer
+		_, _ = io.CopyN(&msg, resp.Body, 4096) // ignore error, just display what we got
+		return errors.Errorf("%s", msg.String())
+	}
+
+	decodeError := json.NewDecoder(resp.Body).Decode(&res)
+	if decodeError != nil {
+		return decodeError
+	}
+
+	return nil
+}
@@ -0,0 +1,97 @@
+package status
+
+import (
+	"context"
+	"os"
+	"time"
+
+	"github.com/mattn/go-isatty"
+	"github.com/pkg/errors"
+	"github.com/spf13/pflag"
+
+	"github.com/zrepl/zrepl/cli"
+	"github.com/zrepl/zrepl/client/status/client"
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon"
+	"github.com/zrepl/zrepl/util/choices"
+)
+
+type Client interface {
+	Status() (daemon.Status, error)
+	StatusRaw() ([]byte, error)
+	SignalWakeup(job string) error
+	SignalReset(job string) error
+}
+
+type statusFlags struct {
+	Mode  choices.Choices
+	Job   string
+	Delay time.Duration
+}
+
+var statusv2Flags statusFlags
+
+type statusv2Mode int
+
+const (
+	StatusV2ModeInteractive statusv2Mode = 1 + iota
+	StatusV2ModeDump
+	StatusV2ModeRaw
+	StatusV2ModeLegacy
+)
+
+var Subcommand = &cli.Subcommand{
+	Use:   "status",
+	Short: "retrieve & display daemon status information",
+	SetupFlags: func(f *pflag.FlagSet) {
+		statusv2Flags.Mode.Init(
+			"interactive", StatusV2ModeInteractive,
+			"dump", StatusV2ModeDump,
+			"raw", StatusV2ModeRaw,
+			"legacy", StatusV2ModeLegacy,
+		)
+		statusv2Flags.Mode.SetTypeString("mode")
+		statusv2Flags.Mode.SetDefaultValue(StatusV2ModeInteractive)
+		f.Var(&statusv2Flags.Mode, "mode", statusv2Flags.Mode.Usage())
+		f.StringVar(&statusv2Flags.Job, "job", "", "only show specified job (works in \"dump\" and \"interactive\" mode)")
+		f.DurationVarP(&statusv2Flags.Delay, "delay", "d", 1*time.Second, "use -d 3s for 3 seconds delay (minimum delay is 1s)")
+	},
+	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
+		return runStatusV2Command(ctx, subcommand.Config(), args)
+	},
+}
+
+func runStatusV2Command(ctx context.Context, config *config.Config, args []string) error {
+
+	c, err := client.New("unix", config.Global.Control.SockPath)
+	if err != nil {
+		return errors.Wrapf(err, "connect to daemon socket at %q", config.Global.Control.SockPath)
+	}
+
+	mode := statusv2Flags.Mode.Value().(statusv2Mode)
+
+	if !isatty.IsTerminal(os.Stdout.Fd()) && mode != StatusV2ModeDump && mode != StatusV2ModeRaw {
+		dumpmode, err := statusv2Flags.Mode.InputForChoice(StatusV2ModeDump)
+		if err != nil {
+			panic(err)
+		}
+		rawmode, err := statusv2Flags.Mode.InputForChoice(StatusV2ModeRaw)
+		if err != nil {
+			panic(err)
+		}
+		return errors.Errorf("error: stdout is not a tty, please use --mode %s or --mode %s", dumpmode, rawmode)
+	}
+
+	switch mode {
+	case StatusV2ModeInteractive:
+		return interactive(c, statusv2Flags)
+	case StatusV2ModeDump:
+		return dump(c, statusv2Flags.Job)
+	case StatusV2ModeRaw:
+		return raw(c)
+	case StatusV2ModeLegacy:
+		return legacy(c, statusv2Flags)
+	default:
+		panic("unreachable")
+	}
+}
@@ -0,0 +1,70 @@
+package status
+
+import (
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/gdamore/tcell"
+	"github.com/mattn/go-isatty"
+	"github.com/pkg/errors"
+
+	"github.com/zrepl/zrepl/client/status/viewmodel"
+)
+
+func dump(c Client, job string) error {
+	s, err := c.Status()
+	if err != nil {
+		return err
+	}
+
+	if job != "" {
+		if _, ok := s.Jobs[job]; !ok {
+			return errors.Errorf("job %q not found", job)
+		}
+	}
+
+	width := (1 << 31) - 1
+	wrap := false
+	hline := strings.Repeat("-", 80)
+	if isatty.IsTerminal(os.Stdout.Fd()) {
+		wrap = true
+		screen, err := tcell.NewScreen()
+		if err != nil {
+			return errors.Wrap(err, "get terminal dimensions")
+		}
+		if err := screen.Init(); err != nil {
+			return errors.Wrap(err, "init screen")
+		}
+		width, _ = screen.Size()
+		screen.Fini()
+		hline = strings.Repeat("-", width)
+	}
+
+	m := viewmodel.New()
+	params := viewmodel.Params{
+		Report:                  s.Jobs,
+		ReportFetchError:        nil,
+		SelectedJob:             nil,
+		FSFilter:                func(s string) bool { return true },
+		DetailViewWidth:         width,
+		DetailViewWrap:          wrap,
+		ShortKeybindingOverview: "",
+	}
+	m.Update(params)
+	for _, j := range m.Jobs() {
+		if job != "" && j.Name() != job {
+			continue
+		}
+		params.SelectedJob = j
+		m.Update(params)
+		fmt.Println(m.SelectedJob().FullDescription())
+		if job != "" {
+			return nil
+		} else {
+			fmt.Println(hline)
+		}
+	}
+
+	return nil
+}
@@ -0,0 +1,347 @@
+package status
+
+import (
+	"fmt"
+	"regexp"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/gdamore/tcell/v2"
+	tview "gitlab.com/tslocum/cview"
+
+	"github.com/zrepl/zrepl/client/status/viewmodel"
+)
+
+func interactive(c Client, flag statusFlags) error {
+
+	// Set this so we don't overwrite the default terminal colors
+	// See https://github.com/rivo/tview/blob/master/styles.go
+	tview.Styles.PrimitiveBackgroundColor = tcell.ColorDefault
+	tview.Styles.ContrastBackgroundColor = tcell.ColorDefault
+	tview.Styles.PrimaryTextColor = tcell.ColorDefault
+	tview.Styles.BorderColor = tcell.ColorDefault
+	app := tview.NewApplication()
+
+	jobDetailSplit := tview.NewFlex()
+	jobMenu := tview.NewTreeView()
+	jobMenuRoot := tview.NewTreeNode("jobs")
+	jobMenuRoot.SetSelectable(true)
+	jobMenu.SetRoot(jobMenuRoot)
+	jobMenu.SetCurrentNode(jobMenuRoot)
+	jobMenu.SetSelectedTextColor(tcell.ColorGreen)
+	jobTextDetail := tview.NewTextView()
+	jobTextDetail.SetWrap(false)
+
+	jobMenu.SetBorder(true)
+	jobTextDetail.SetBorder(true)
+
+	toolbarSplit := tview.NewFlex()
+	toolbarSplit.SetDirection(tview.FlexRow)
+	inputBarContainer := tview.NewFlex()
+	fsFilterInput := tview.NewInputField()
+	fsFilterInput.SetBorder(false)
+	fsFilterInput.SetFieldBackgroundColor(tcell.ColorDefault)
+	inputBarLabel := tview.NewTextView()
+	inputBarLabel.SetText("[::b]FILTER ")
+	inputBarLabel.SetDynamicColors(true)
+	inputBarContainer.AddItem(inputBarLabel, 7, 1, false)
+	inputBarContainer.AddItem(fsFilterInput, 0, 10, false)
+	toolbarSplit.AddItem(inputBarContainer, 1, 0, false)
+	toolbarSplit.AddItem(jobDetailSplit, 0, 10, false)
+
+	bottombar := tview.NewFlex()
+	bottombar.SetDirection(tview.FlexColumn)
+	bottombarDateView := tview.NewTextView()
+	bottombar.AddItem(bottombarDateView, len(time.Now().String()), 0, false)
+	bottomBarStatus := tview.NewTextView()
+	bottomBarStatus.SetDynamicColors(true)
+	bottomBarStatus.SetTextAlign(tview.AlignRight)
+	bottombar.AddItem(bottomBarStatus, 0, 10, false)
+	toolbarSplit.AddItem(bottombar, 1, 0, false)
+
+	tabbableWithJobMenu := []tview.Primitive{jobMenu, jobTextDetail, fsFilterInput}
+	tabbableWithoutJobMenu := []tview.Primitive{jobTextDetail, fsFilterInput}
+	var tabbable []tview.Primitive
+	tabbableActiveIndex := 0
+	tabbableRedraw := func() {
+		if len(tabbable) == 0 {
+			app.SetFocus(nil)
+			return
+		}
+		if tabbableActiveIndex >= len(tabbable) {
+			app.SetFocus(tabbable[0])
+			return
+		}
+		app.SetFocus(tabbable[tabbableActiveIndex])
+	}
+	tabbableCycle := func() {
+		if len(tabbable) == 0 {
+			return
+		}
+		tabbableActiveIndex = (tabbableActiveIndex + 1) % len(tabbable)
+		app.SetFocus(tabbable[tabbableActiveIndex])
+		tabbableRedraw()
+	}
+
+	jobMenuVisisble := false
+	reconfigureJobDetailSplit := func(setJobMenuVisible bool) {
+		if jobMenuVisisble == setJobMenuVisible {
+			return
+		}
+		jobMenuVisisble = setJobMenuVisible
+		if setJobMenuVisible {
+			jobDetailSplit.RemoveItem(jobTextDetail)
+			jobDetailSplit.AddItem(jobMenu, 0, 1, true)
+			jobDetailSplit.AddItem(jobTextDetail, 0, 5, false)
+			tabbable = tabbableWithJobMenu
+		} else {
+			jobDetailSplit.RemoveItem(jobMenu)
+			tabbable = tabbableWithoutJobMenu
+		}
+		tabbableRedraw()
+	}
+
+	showModal := func(m *tview.Modal, modalDoneFunc func(idx int, label string)) {
+		preModalFocus := app.GetFocus()
+		m.SetDoneFunc(func(idx int, label string) {
+			if modalDoneFunc != nil {
+				modalDoneFunc(idx, label)
+			}
+			app.SetRoot(toolbarSplit, true)
+			app.SetFocus(preModalFocus)
+			app.Draw()
+		})
+		app.SetRoot(m, true)
+		app.Draw()
+	}
+
+	app.SetRoot(toolbarSplit, true)
+	// initial focus
+	tabbableActiveIndex = len(tabbable)
+	tabbableCycle()
+	reconfigureJobDetailSplit(true)
+
+	m := viewmodel.New()
+	params := &viewmodel.Params{
+		Report:                  nil,
+		SelectedJob:             nil,
+		FSFilter:                func(_ string) bool { return true },
+		DetailViewWidth:         100,
+		DetailViewWrap:          false,
+		ShortKeybindingOverview: "[::b]Q[::-] quit  [::b]<TAB>[::-] switch panes  [::b]Shift+M[::-] toggle navbar  [::b]Shift+S[::-] signal job [::b]</>[::-] filter filesystems",
+	}
+	paramsMtx := &sync.Mutex{}
+	var redraw func()
+	viewmodelupdate := func(cb func(*viewmodel.Params)) {
+		paramsMtx.Lock()
+		defer paramsMtx.Unlock()
+		cb(params)
+		m.Update(*params)
+	}
+	redraw = func() {
+		jobs := m.Jobs()
+		if flag.Job != "" {
+			job_found := false
+			for _, job := range jobs {
+				if strings.Compare(flag.Job, job.Name()) == 0 {
+					jobs = []*viewmodel.Job{job}
+					job_found = true
+					break
+				}
+			}
+			if !job_found {
+				jobs = nil
+			}
+		}
+		redrawJobsList := false
+		var selectedJobN *tview.TreeNode
+		if len(jobMenuRoot.GetChildren()) == len(jobs) {
+			for i, jobN := range jobMenuRoot.GetChildren() {
+				if jobN.GetReference().(*viewmodel.Job) != jobs[i] {
+					redrawJobsList = true
+					break
+				}
+				if jobN.GetReference().(*viewmodel.Job) == m.SelectedJob() {
+					selectedJobN = jobN
+				}
+			}
+		} else {
+			redrawJobsList = true
+		}
+		if redrawJobsList {
+			selectedJobN = nil
+			children := make([]*tview.TreeNode, len(jobs))
+			for i := range jobs {
+				jobN := tview.NewTreeNode(jobs[i].JobTreeTitle())
+				jobN.SetReference(jobs[i])
+				jobN.SetSelectable(true)
+				children[i] = jobN
+				jobN.SetSelectedFunc(func() {
+					viewmodelupdate(func(p *viewmodel.Params) {
+						p.SelectedJob = jobN.GetReference().(*viewmodel.Job)
+					})
+				})
+				if jobs[i] == m.SelectedJob() {
+					selectedJobN = jobN
+				}
+			}
+			jobMenuRoot.SetChildren(children)
+		}
+
+		if selectedJobN != nil && jobMenu.GetCurrentNode() != selectedJobN {
+			jobMenu.SetCurrentNode(selectedJobN)
+		} else if selectedJobN == nil {
+			// select something, otherwise selection breaks (likely bug in tview)
+			jobMenu.SetCurrentNode(jobMenuRoot)
+		}
+
+		if selJ := m.SelectedJob(); selJ != nil {
+			jobTextDetail.SetText(selJ.FullDescription())
+		} else {
+			jobTextDetail.SetText("please select a job")
+		}
+
+		bottombardatestring := m.DateString()
+		bottombarDateView.SetText(bottombardatestring)
+		bottombar.ResizeItem(bottombarDateView, len(bottombardatestring), 0)
+
+		bottomBarStatus.SetText(m.BottomBarStatus())
+
+		app.Draw()
+
+	}
+
+	go func() {
+		defer func() {
+			if err := recover(); err != nil {
+				app.Suspend(func() {
+					panic(err)
+				})
+			}
+		}()
+		for {
+			st, err := c.Status()
+			viewmodelupdate(func(p *viewmodel.Params) {
+				p.Report = st.Jobs
+				p.ReportFetchError = err
+			})
+			app.QueueUpdateDraw(redraw)
+
+			time.Sleep(flag.Delay)
+		}
+	}()
+
+	jobMenu.SetChangedFunc(func(jobN *tview.TreeNode) {
+		viewmodelupdate(func(p *viewmodel.Params) {
+			p.SelectedJob, _ = jobN.GetReference().(*viewmodel.Job)
+		})
+		redraw()
+		jobTextDetail.ScrollToBeginning()
+	})
+	jobMenu.SetSelectedFunc(func(jobN *tview.TreeNode) {
+		app.SetFocus(jobTextDetail)
+	})
+
+	app.SetBeforeDrawFunc(func(screen tcell.Screen) bool {
+		viewmodelupdate(func(p *viewmodel.Params) {
+			_, _, p.DetailViewWidth, _ = jobTextDetail.GetInnerRect()
+		})
+		return false
+	})
+
+	app.SetInputCapture(func(e *tcell.EventKey) *tcell.EventKey {
+		if e.Key() == tcell.KeyTab {
+			tabbableCycle()
+			return nil
+		}
+
+		if e.Key() == tcell.KeyRune && app.GetFocus() == fsFilterInput {
+			return e
+		}
+
+		if e.Key() == tcell.KeyRune && e.Rune() == '/' {
+			if app.GetFocus() != fsFilterInput {
+				app.SetFocus(fsFilterInput)
+			}
+			return e
+		}
+
+		if e.Key() == tcell.KeyRune && e.Rune() == 'M' {
+			reconfigureJobDetailSplit(!jobMenuVisisble)
+			return nil
+		}
+
+		if e.Key() == tcell.KeyRune && e.Rune() == 'q' {
+			app.Stop()
+		}
+
+		if e.Key() == tcell.KeyRune && e.Rune() == 'S' {
+			job, ok := jobMenu.GetCurrentNode().GetReference().(*viewmodel.Job)
+			if !ok {
+				return nil
+			}
+			signals := []string{"wakeup", "reset"}
+			clientFuncs := []func(job string) error{c.SignalWakeup, c.SignalReset}
+			sigMod := tview.NewModal()
+			sigMod.SetBackgroundColor(tcell.ColorDefault)
+			sigMod.SetBorder(true)
+			sigMod.GetForm().SetButtonTextColorFocused(tcell.ColorGreen)
+			sigMod.AddButtons(signals)
+			sigMod.SetText(fmt.Sprintf("Send a signal to job %q", job.Name()))
+			showModal(sigMod, func(idx int, _ string) {
+				go func() {
+					if idx == -1 {
+						return
+					}
+					err := clientFuncs[idx](job.Name())
+					if err != nil {
+						app.QueueUpdate(func() {
+							me := tview.NewModal()
+							me.SetText(fmt.Sprintf("signal error: %s", err))
+							me.AddButtons([]string{"Close"})
+							showModal(me, nil)
+						})
+					}
+				}()
+			})
+		}
+
+		return e
+	})
+
+	fsFilterInput.SetChangedFunc(func(searchterm string) {
+		viewmodelupdate(func(p *viewmodel.Params) {
+			p.FSFilter = func(fs string) bool {
+				r, err := regexp.Compile(searchterm)
+				if err != nil {
+					return true
+				}
+				return r.MatchString(fs)
+			}
+		})
+		redraw()
+		jobTextDetail.ScrollToBeginning()
+	})
+	fsFilterInput.SetInputCapture(func(event *tcell.EventKey) *tcell.EventKey {
+		if event.Key() == tcell.KeyEnter {
+			app.SetFocus(jobTextDetail)
+			return nil
+		}
+		return event
+	})
+
+	jobTextDetail.SetInputCapture(func(event *tcell.EventKey) *tcell.EventKey {
+		if event.Key() == tcell.KeyRune && event.Rune() == 'w' {
+			// toggle wrapping
+			viewmodelupdate(func(p *viewmodel.Params) {
+				p.DetailViewWrap = !p.DetailViewWrap
+			})
+			redraw()
+			return nil
+		}
+		return event
+	})
+
+	return app.Run()
+}
@@ -0,0 +1,128 @@
+package status
+
+import (
+	"fmt"
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/gdamore/tcell/v2"
+	"github.com/mattn/go-isatty"
+	"github.com/pkg/errors"
+	tview "gitlab.com/tslocum/cview"
+
+	"github.com/zrepl/zrepl/client/status/viewmodel"
+)
+
+func legacy(c Client, flag statusFlags) error {
+
+	// Set this so we don't overwrite the default terminal colors
+	// See https://github.com/rivo/tview/blob/master/styles.go
+	tview.Styles.PrimitiveBackgroundColor = tcell.ColorDefault
+	tview.Styles.ContrastBackgroundColor = tcell.ColorDefault
+	tview.Styles.PrimaryTextColor = tcell.ColorDefault
+	tview.Styles.BorderColor = tcell.ColorDefault
+	app := tview.NewApplication()
+
+	textView := tview.NewTextView()
+	textView.SetWrap(true)
+	textView.SetScrollable(true) // so that it allows us to set scroll position
+	textView.SetScrollBarVisibility(tview.ScrollBarNever)
+
+	app.SetRoot(textView, true)
+
+	width := (1 << 31) - 1
+	wrap := false
+	if isatty.IsTerminal(os.Stdout.Fd()) {
+		wrap = true
+		screen, err := tcell.NewScreen()
+		if err != nil {
+			return errors.Wrap(err, "get terminal dimensions")
+		}
+		if err := screen.Init(); err != nil {
+			return errors.Wrap(err, "init screen")
+		}
+		width, _ = screen.Size()
+		screen.Fini()
+	}
+
+	paramsMtx := &sync.Mutex{}
+	params := viewmodel.Params{
+		Report:                  nil,
+		ReportFetchError:        nil,
+		SelectedJob:             nil,
+		FSFilter:                func(s string) bool { return true },
+		DetailViewWidth:         width,
+		DetailViewWrap:          wrap,
+		ShortKeybindingOverview: "",
+	}
+
+	redraw := func() {
+		textView.Clear()
+
+		paramsMtx.Lock()
+		defer paramsMtx.Unlock()
+
+		if params.ReportFetchError != nil {
+			fmt.Fprintln(textView, params.ReportFetchError.Error())
+		} else if params.Report != nil {
+			m := viewmodel.New()
+			m.Update(params)
+			for _, j := range m.Jobs() {
+				if flag.Job != "" && j.Name() != flag.Job {
+					continue
+				}
+				params.SelectedJob = j
+				m.Update(params)
+				fmt.Fprintln(textView, m.SelectedJob().FullDescription())
+				if flag.Job != "" {
+					break
+				} else {
+					hline := strings.Repeat("-", params.DetailViewWidth)
+					fmt.Fprintln(textView, hline)
+				}
+			}
+		} else {
+			fmt.Fprintln(textView, "waiting for request results")
+		}
+		textView.ScrollToBeginning()
+	}
+
+	app.SetBeforeDrawFunc(func(screen tcell.Screen) bool {
+		// sync resizes to `params`
+		paramsMtx.Lock()
+		_, _, newWidth, _ := textView.GetInnerRect()
+		if newWidth != params.DetailViewWidth {
+			params.DetailViewWidth = newWidth
+			app.QueueUpdateDraw(redraw)
+		}
+		paramsMtx.Unlock()
+
+		textView.ScrollToBeginning() // has the effect of inhibiting user scrolls
+		return false
+	})
+
+	go func() {
+		defer func() {
+			if err := recover(); err != nil {
+				app.Suspend(func() {
+					panic(err)
+				})
+			}
+		}()
+		for {
+			st, err := c.Status()
+			paramsMtx.Lock()
+			params.Report = st.Jobs
+			params.ReportFetchError = err
+			paramsMtx.Unlock()
+
+			app.QueueUpdateDraw(redraw)
+
+			time.Sleep(flag.Delay)
+		}
+	}()
+
+	return app.Run()
+}
@@ -0,0 +1,18 @@
+package status
+
+import (
+	"bytes"
+	"io"
+	"os"
+)
+
+func raw(c Client) error {
+	b, err := c.StatusRaw()
+	if err != nil {
+		return err
+	}
+	if _, err := io.Copy(os.Stdout, bytes.NewReader(b)); err != nil {
+		return err
+	}
+	return nil
+}
@@ -0,0 +1,18 @@
+package viewmodel
+
+import (
+	"fmt"
+)
+
+func ByteCountBinary(b int64) string {
+	const unit = 1024
+	if b < unit {
+		return fmt.Sprintf("%d B", b)
+	}
+	div, exp := int64(unit), 0
+	for n := b / unit; n >= unit; n /= unit {
+		div *= unit
+		exp++
+	}
+	return fmt.Sprintf("%.1f %ciB", float64(b)/float64(div), "KMGTPE"[exp])
+}
@@ -0,0 +1,48 @@
+package viewmodel
+
+import "time"
+
+type byteProgressMeasurement struct {
+	time time.Time
+	val  int64
+}
+
+type bytesProgressHistory struct {
+	last        *byteProgressMeasurement // pointer as poor man's optional
+	changeCount int
+	lastChange  time.Time
+	bpsAvg      float64
+}
+
+func (p *bytesProgressHistory) Update(currentVal int64) (bytesPerSecondAvg int64, changeCount int) {
+
+	if p.last == nil {
+		p.last = &byteProgressMeasurement{
+			time: time.Now(),
+			val:  currentVal,
+		}
+		return 0, 0
+	}
+
+	if p.last.val != currentVal {
+		p.changeCount++
+		p.lastChange = time.Now()
+	}
+
+	if time.Since(p.lastChange) > 3*time.Second {
+		p.last = nil
+		return 0, 0
+	}
+
+	deltaV := currentVal - p.last.val
+	deltaT := time.Since(p.last.time)
+	rate := float64(deltaV) / deltaT.Seconds()
+
+	factor := 0.3
+	p.bpsAvg = (1-factor)*p.bpsAvg + factor*rate
+
+	p.last.time = time.Now()
+	p.last.val = currentVal
+
+	return int64(p.bpsAvg), p.changeCount
+}
@@ -0,0 +1,614 @@
+package viewmodel
+
+import (
+	"fmt"
+	"math"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/go-playground/validator/v10"
+	yaml "github.com/zrepl/yaml-config"
+
+	"github.com/zrepl/zrepl/client/status/viewmodel/stringbuilder"
+	"github.com/zrepl/zrepl/daemon"
+	"github.com/zrepl/zrepl/daemon/job"
+	"github.com/zrepl/zrepl/daemon/pruner"
+	"github.com/zrepl/zrepl/daemon/snapper"
+	"github.com/zrepl/zrepl/replication/report"
+)
+
+type M struct {
+	jobs            map[string]*Job
+	jobsList        []*Job
+	selectedJob     *Job
+	dateString      string
+	bottomBarStatus string
+}
+
+type Job struct {
+	// long-lived
+	name         string
+	byteProgress *bytesProgressHistory
+
+	lastStatus      *job.Status
+	fulldescription string
+}
+
+func New() *M {
+	return &M{
+		jobs:        make(map[string]*Job),
+		jobsList:    make([]*Job, 0),
+		selectedJob: nil,
+	}
+}
+
+type FilterFunc func(string) bool
+
+type Params struct {
+	Report                  map[string]*job.Status
+	ReportFetchError        error
+	SelectedJob             *Job
+	FSFilter                FilterFunc `validate:"required"`
+	DetailViewWidth         int        `validate:"gte=1"`
+	DetailViewWrap          bool
+	ShortKeybindingOverview string
+}
+
+var validate = validator.New()
+
+func (m *M) Update(p Params) {
+
+	if err := validate.Struct(p); err != nil {
+		panic(err)
+	}
+
+	if p.ReportFetchError != nil {
+		m.bottomBarStatus = fmt.Sprintf("[red::]status fetch: %s", p.ReportFetchError)
+	} else {
+		m.bottomBarStatus = p.ShortKeybindingOverview
+		for jobname, st := range p.Report {
+			// TODO handle job renames & deletions
+			j, ok := m.jobs[jobname]
+			if !ok {
+				j = &Job{
+					name:         jobname,
+					byteProgress: &bytesProgressHistory{},
+				}
+				m.jobs[jobname] = j
+				m.jobsList = append(m.jobsList, j)
+			}
+			j.lastStatus = st
+		}
+	}
+
+	// filter out internal jobs
+	var jobsList []*Job
+	for _, j := range m.jobsList {
+		if daemon.IsInternalJobName(j.name) {
+			continue
+		}
+		jobsList = append(jobsList, j)
+	}
+	m.jobsList = jobsList
+
+	// determinism!
+	sort.Slice(m.jobsList, func(i, j int) bool {
+		return strings.Compare(m.jobsList[i].name, m.jobsList[j].name) < 0
+	})
+
+	// try to not lose the selected job
+	m.selectedJob = nil
+	for _, j := range m.jobsList {
+		j.updateFullDescription(p)
+		if j == p.SelectedJob {
+			m.selectedJob = j
+		}
+	}
+
+	m.dateString = time.Now().Format(time.RFC3339)
+
+}
+
+func (m *M) BottomBarStatus() string { return m.bottomBarStatus }
+
+func (m *M) Jobs() []*Job { return m.jobsList }
+
+// may be nil
+func (m *M) SelectedJob() *Job { return m.selectedJob }
+
+func (m *M) DateString() string { return m.dateString }
+
+func (j *Job) updateFullDescription(p Params) {
+	width := p.DetailViewWidth
+	if !p.DetailViewWrap {
+		width = 10000000 // FIXME
+	}
+	b := stringbuilder.New(stringbuilder.Config{
+		IndentMultiplier: 3,
+		Width:            width,
+	})
+	drawJob(b, j.name, j.lastStatus, j.byteProgress, p.FSFilter)
+	j.fulldescription = b.String()
+}
+
+func (j *Job) JobTreeTitle() string {
+	return j.name
+}
+
+func (j *Job) FullDescription() string {
+	return j.fulldescription
+}
+
+func (j *Job) Name() string {
+	return j.name
+}
+
+func drawJob(t *stringbuilder.B, name string, v *job.Status, history *bytesProgressHistory, fsfilter FilterFunc) {
+
+	t.Printf("Job: %s\n", name)
+	t.Printf("Type: %s\n\n", v.Type)
+
+	if v.Type == job.TypePush || v.Type == job.TypePull {
+		activeStatus, ok := v.JobSpecific.(*job.ActiveSideStatus)
+		if !ok || activeStatus == nil {
+			t.Printf("ActiveSideStatus is null")
+			t.Newline()
+			return
+		}
+
+		t.Printf("Replication:")
+		t.AddIndentAndNewline(1)
+		renderReplicationReport(t, activeStatus.Replication, history, fsfilter)
+		t.AddIndentAndNewline(-1)
+
+		t.Printf("Pruning Sender:")
+		t.AddIndentAndNewline(1)
+		renderPrunerReport(t, activeStatus.PruningSender, fsfilter)
+		t.AddIndentAndNewline(-1)
+
+		t.Printf("Pruning Receiver:")
+		t.AddIndentAndNewline(1)
+		renderPrunerReport(t, activeStatus.PruningReceiver, fsfilter)
+		t.AddIndentAndNewline(-1)
+
+		if v.Type == job.TypePush {
+			t.Printf("Snapshotting:")
+			t.AddIndentAndNewline(1)
+			renderSnapperReport(t, activeStatus.Snapshotting, fsfilter)
+			t.AddIndentAndNewline(-1)
+		}
+
+	} else if v.Type == job.TypeSnap {
+		snapStatus, ok := v.JobSpecific.(*job.SnapJobStatus)
+		if !ok || snapStatus == nil {
+			t.Printf("SnapJobStatus is null")
+			t.Newline()
+			return
+		}
+		t.Printf("Pruning snapshots:")
+		t.AddIndentAndNewline(1)
+		renderPrunerReport(t, snapStatus.Pruning, fsfilter)
+		t.AddIndentAndNewline(-1)
+		t.Printf("Snapshotting:")
+		t.AddIndentAndNewline(1)
+		renderSnapperReport(t, snapStatus.Snapshotting, fsfilter)
+		t.AddIndentAndNewline(-1)
+	} else if v.Type == job.TypeSource {
+
+		st := v.JobSpecific.(*job.PassiveStatus)
+		t.Printf("Snapshotting:\n")
+		t.AddIndent(1)
+		renderSnapperReport(t, st.Snapper, fsfilter)
+		t.AddIndentAndNewline(-1)
+
+	} else {
+		t.Printf("No status representation for job type '%s', dumping as YAML", v.Type)
+		t.Newline()
+		asYaml, err := yaml.Marshal(v.JobSpecific)
+		if err != nil {
+			t.Printf("Error marshaling status to YAML: %s", err)
+			t.Newline()
+			return
+		}
+		t.Write(string(asYaml))
+		t.Newline()
+	}
+}
+
+func printFilesystemStatus(t *stringbuilder.B, rep *report.FilesystemReport, active bool, maxFS int) {
+
+	expected, replicated, containsInvalidSizeEstimates := rep.BytesSum()
+	sizeEstimationImpreciseNotice := ""
+	if containsInvalidSizeEstimates {
+		sizeEstimationImpreciseNotice = " (some steps lack size estimation)"
+	}
+	if rep.CurrentStep < len(rep.Steps) && rep.Steps[rep.CurrentStep].Info.BytesExpected == 0 {
+		sizeEstimationImpreciseNotice = " (step lacks size estimation)"
+	}
+
+	status := fmt.Sprintf("%s (step %d/%d, %s/%s)%s",
+		strings.ToUpper(string(rep.State)),
+		rep.CurrentStep, len(rep.Steps),
+		ByteCountBinary(replicated), ByteCountBinary(expected),
+		sizeEstimationImpreciseNotice,
+	)
+
+	activeIndicator := " "
+	if active {
+		activeIndicator = "*"
+	}
+	t.AddIndent(1)
+	t.Printf("%s %s %s ",
+		activeIndicator,
+		stringbuilder.RightPad(rep.Info.Name, maxFS, " "),
+		status)
+
+	next := ""
+	if err := rep.Error(); err != nil {
+		next = err.Err
+	} else if rep.State != report.FilesystemDone {
+		if nextStep := rep.NextStep(); nextStep != nil {
+			if nextStep.IsIncremental() {
+				next = fmt.Sprintf("next: %s => %s", nextStep.Info.From, nextStep.Info.To)
+			} else {
+				next = fmt.Sprintf("next: full send %s", nextStep.Info.To)
+			}
+			attribs := []string{}
+
+			if nextStep.Info.Resumed {
+				attribs = append(attribs, "resumed")
+			}
+
+			attribs = append(attribs, fmt.Sprintf("encrypted=%s", nextStep.Info.Encrypted))
+
+			next += fmt.Sprintf(" (%s)", strings.Join(attribs, ", "))
+		} else {
+			next = "" // individual FSes may still be in planning state
+		}
+
+	}
+	t.Printf("%s", next)
+
+	t.AddIndent(-1)
+	t.Newline()
+}
+
+func renderReplicationReport(t *stringbuilder.B, rep *report.Report, history *bytesProgressHistory, fsfilter FilterFunc) {
+	if rep == nil {
+		t.Printf("...\n")
+		return
+	}
+
+	if rep.WaitReconnectError != nil {
+		t.PrintfDrawIndentedAndWrappedIfMultiline("Connectivity: %s", rep.WaitReconnectError)
+		t.Newline()
+	}
+	if !rep.WaitReconnectSince.IsZero() {
+		delta := time.Until(rep.WaitReconnectUntil).Round(time.Second)
+		if rep.WaitReconnectUntil.IsZero() || delta > 0 {
+			var until string
+			if rep.WaitReconnectUntil.IsZero() {
+				until = "waiting indefinitely"
+			} else {
+				until = fmt.Sprintf("hard fail in %s @ %s", delta, rep.WaitReconnectUntil)
+			}
+			t.PrintfDrawIndentedAndWrappedIfMultiline("Connectivity: reconnecting with exponential backoff (since %s) (%s)",
+				rep.WaitReconnectSince, until)
+		} else {
+			t.PrintfDrawIndentedAndWrappedIfMultiline("Connectivity: reconnects reached hard-fail timeout @ %s", rep.WaitReconnectUntil)
+		}
+		t.Newline()
+	}
+
+	// TODO visualize more than the latest attempt by folding all attempts into one
+	if len(rep.Attempts) == 0 {
+		t.Printf("no attempts made yet")
+		return
+	} else {
+		t.Printf("Attempt #%d", len(rep.Attempts))
+		if len(rep.Attempts) > 1 {
+			t.Printf(". Previous attempts failed with the following statuses:")
+			t.AddIndentAndNewline(1)
+			for i, a := range rep.Attempts[:len(rep.Attempts)-1] {
+				t.PrintfDrawIndentedAndWrappedIfMultiline("#%d: %s (failed at %s) (ran %s)\n", i+1, a.State, a.FinishAt, a.FinishAt.Sub(a.StartAt))
+			}
+			t.AddIndentAndNewline(-1)
+		} else {
+			t.Newline()
+		}
+	}
+
+	latest := rep.Attempts[len(rep.Attempts)-1]
+	sort.Slice(latest.Filesystems, func(i, j int) bool {
+		return latest.Filesystems[i].Info.Name < latest.Filesystems[j].Info.Name
+	})
+
+	// apply filter
+	filtered := make([]*report.FilesystemReport, 0, len(latest.Filesystems))
+	for _, fs := range latest.Filesystems {
+		if !fsfilter(fs.Info.Name) {
+			continue
+		}
+		filtered = append(filtered, fs)
+	}
+	latest.Filesystems = filtered
+
+	t.Printf("Status: %s", latest.State)
+	t.Newline()
+	if !latest.FinishAt.IsZero() {
+		t.Printf("Last Run: %s (lasted %s)\n", latest.FinishAt.Round(time.Second), latest.FinishAt.Sub(latest.StartAt).Round(time.Second))
+	} else {
+		t.Printf("Started: %s (lasting %s)\n", latest.StartAt.Round(time.Second), time.Since(latest.StartAt).Round(time.Second))
+	}
+
+	if latest.State == report.AttemptPlanningError {
+		t.Printf("Problem: ")
+		t.PrintfDrawIndentedAndWrappedIfMultiline("%s", latest.PlanError)
+		t.Newline()
+	} else if latest.State == report.AttemptFanOutError {
+		t.Printf("Problem: one or more of the filesystems encountered errors")
+		t.Newline()
+	}
+
+	if latest.State != report.AttemptPlanning && latest.State != report.AttemptPlanningError {
+		// Draw global progress bar
+		// Progress: [---------------]
+		expected, replicated, containsInvalidSizeEstimates := latest.BytesSum()
+		rate, changeCount := history.Update(replicated)
+		eta := time.Duration(0)
+		if rate > 0 {
+			eta = time.Duration((expected-replicated)/rate) * time.Second
+		}
+
+		t.Write("Progress: ")
+		t.DrawBar(50, replicated, expected, changeCount)
+		t.Write(fmt.Sprintf(" %s / %s @ %s/s", ByteCountBinary(replicated), ByteCountBinary(expected), ByteCountBinary(rate)))
+		if eta != 0 {
+			t.Write(fmt.Sprintf(" (%s remaining)", humanizeDuration(eta)))
+		}
+		t.Newline()
+		if containsInvalidSizeEstimates {
+			t.Write("NOTE: not all steps could be size-estimated, total estimate is likely imprecise!")
+			t.Newline()
+		}
+
+		if len(latest.Filesystems) == 0 {
+			t.Write("NOTE: no filesystems were considered for replication!")
+			t.Newline()
+		}
+
+		var maxFSLen int
+		for _, fs := range latest.Filesystems {
+			if len(fs.Info.Name) > maxFSLen {
+				maxFSLen = len(fs.Info.Name)
+			}
+		}
+		for _, fs := range latest.Filesystems {
+			printFilesystemStatus(t, fs, false, maxFSLen) // FIXME bring 'active' flag back
+		}
+
+	}
+}
+
+func humanizeDuration(duration time.Duration) string {
+	days := int64(duration.Hours() / 24)
+	hours := int64(math.Mod(duration.Hours(), 24))
+	minutes := int64(math.Mod(duration.Minutes(), 60))
+	seconds := int64(math.Mod(duration.Seconds(), 60))
+
+	var parts []string
+
+	force := false
+	chunks := []int64{days, hours, minutes, seconds}
+	for i, chunk := range chunks {
+		if force || chunk > 0 {
+			padding := 0
+			if force {
+				padding = 2
+			}
+			parts = append(parts, fmt.Sprintf("%*d%c", padding, chunk, "dhms"[i]))
+			force = true
+		}
+	}
+
+	return strings.Join(parts, " ")
+}
+
+func renderPrunerReport(t *stringbuilder.B, r *pruner.Report, fsfilter FilterFunc) {
+	if r == nil {
+		t.Printf("...\n")
+		return
+	}
+
+	state, err := pruner.StateString(r.State)
+	if err != nil {
+		t.Printf("Status: %q (parse error: %q)\n", r.State, err)
+		return
+	}
+
+	t.Printf("Status: %s", state)
+	t.Newline()
+
+	if r.Error != "" {
+		t.Printf("Error: %s\n", r.Error)
+	}
+
+	type commonFS struct {
+		*pruner.FSReport
+		completed bool
+	}
+	all := make([]commonFS, 0, len(r.Pending)+len(r.Completed))
+	for i := range r.Pending {
+		all = append(all, commonFS{&r.Pending[i], false})
+	}
+	for i := range r.Completed {
+		all = append(all, commonFS{&r.Completed[i], true})
+	}
+
+	// filter all
+	filtered := make([]commonFS, 0, len(all))
+	for _, fs := range all {
+		if fsfilter(fs.FSReport.Filesystem) {
+			filtered = append(filtered, fs)
+		}
+	}
+	all = filtered
+
+	switch state {
+	case pruner.Plan:
+		fallthrough
+	case pruner.PlanErr:
+		return
+	}
+
+	if len(all) == 0 {
+		t.Printf("nothing to do\n")
+		return
+	}
+
+	var totalDestroyCount, completedDestroyCount int
+	var maxFSname int
+	for _, fs := range all {
+		totalDestroyCount += len(fs.DestroyList)
+		if fs.completed {
+			completedDestroyCount += len(fs.DestroyList)
+		}
+		if maxFSname < len(fs.Filesystem) {
+			maxFSname = len(fs.Filesystem)
+		}
+	}
+
+	// global progress bar
+	progress := int(math.Round(80 * float64(completedDestroyCount) / float64(totalDestroyCount)))
+	t.Write("Progress: ")
+	t.Write("[")
+	t.Write(stringbuilder.Times("=", progress))
+	t.Write(">")
+	t.Write(stringbuilder.Times("-", 80-progress))
+	t.Write("]")
+	t.Printf(" %d/%d snapshots", completedDestroyCount, totalDestroyCount)
+	t.Newline()
+
+	sort.SliceStable(all, func(i, j int) bool {
+		return strings.Compare(all[i].Filesystem, all[j].Filesystem) == -1
+	})
+
+	// Draw a table-like representation of 'all'
+	for _, fs := range all {
+		t.Write(stringbuilder.RightPad(fs.Filesystem, maxFSname, " "))
+		t.Write(" ")
+		if !fs.SkipReason.NotSkipped() {
+			t.Printf("skipped: %s\n", fs.SkipReason)
+			continue
+		}
+		if fs.LastError != "" {
+			if strings.ContainsAny(fs.LastError, "\r\n") {
+				t.Printf("ERROR:")
+				t.PrintfDrawIndentedAndWrappedIfMultiline("%s\n", fs.LastError)
+			} else {
+				t.PrintfDrawIndentedAndWrappedIfMultiline("ERROR: %s\n", fs.LastError)
+			}
+			t.Newline()
+			continue
+		}
+
+		pruneRuleActionStr := fmt.Sprintf("(destroy %d of %d snapshots)",
+			len(fs.DestroyList), len(fs.SnapshotList))
+
+		if fs.completed {
+			t.Printf("Completed  %s\n", pruneRuleActionStr)
+			continue
+		}
+
+		t.Write("Pending    ") // whitespace is padding 10
+		if len(fs.DestroyList) == 1 {
+			t.Write(fs.DestroyList[0].Name)
+		} else {
+			t.Write(pruneRuleActionStr)
+		}
+		t.Newline()
+	}
+
+}
+
+func renderSnapperReport(t *stringbuilder.B, r *snapper.Report, fsfilter FilterFunc) {
+	if r == nil {
+		t.Printf("<snapshot type does not have a report>\n")
+		return
+	}
+
+	t.Printf("Status: %s", r.State)
+	t.Newline()
+
+	if r.Error != "" {
+		t.Printf("Error: %s\n", r.Error)
+	}
+	if !r.SleepUntil.IsZero() {
+		t.Printf("Sleep until: %s\n", r.SleepUntil)
+	}
+
+	sort.Slice(r.Progress, func(i, j int) bool {
+		return strings.Compare(r.Progress[i].Path, r.Progress[j].Path) == -1
+	})
+
+	dur := func(d time.Duration) string {
+		return d.Round(100 * time.Millisecond).String()
+	}
+
+	type row struct {
+		path, state, duration, remainder, hookReport string
+	}
+	var widths struct {
+		path, state, duration int
+	}
+	rows := make([]*row, 0, len(r.Progress))
+	for _, fs := range r.Progress {
+		if !fsfilter(fs.Path) {
+			continue
+		}
+		r := &row{
+			path:  fs.Path,
+			state: fs.State.String(),
+		}
+		if fs.HooksHadError {
+			r.hookReport = fs.Hooks // FIXME render here, not in daemon
+		}
+		switch fs.State {
+		case snapper.SnapPending:
+			r.duration = "..."
+			r.remainder = ""
+		case snapper.SnapStarted:
+			r.duration = dur(time.Since(fs.StartAt))
+			r.remainder = fmt.Sprintf("snap name: %q", fs.SnapName)
+		case snapper.SnapDone:
+			fallthrough
+		case snapper.SnapError:
+			r.duration = dur(fs.DoneAt.Sub(fs.StartAt))
+			r.remainder = fmt.Sprintf("snap name: %q", fs.SnapName)
+		}
+		rows = append(rows, r)
+		if len(r.path) > widths.path {
+			widths.path = len(r.path)
+		}
+		if len(r.state) > widths.state {
+			widths.state = len(r.state)
+		}
+		if len(r.duration) > widths.duration {
+			widths.duration = len(r.duration)
+		}
+	}
+
+	for _, r := range rows {
+		path := stringbuilder.RightPad(r.path, widths.path, " ")
+		state := stringbuilder.RightPad(r.state, widths.state, " ")
+		duration := stringbuilder.RightPad(r.duration, widths.duration, " ")
+		t.Printf("%s %s %s", path, state, duration)
+		t.PrintfDrawIndentedAndWrappedIfMultiline(" %s", r.remainder)
+		if r.hookReport != "" {
+			t.PrintfDrawIndentedAndWrappedIfMultiline("%s", r.hookReport)
+		}
+		t.Newline()
+	}
+
+}
@@ -0,0 +1,119 @@
+package stringbuilder
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/go-playground/validator/v10"
+)
+
+type B struct {
+	// const
+	indentMultiplier int
+
+	// mut
+	sb     *strings.Builder
+	indent int
+	width  int
+	x, y   int
+}
+
+type Config struct {
+	IndentMultiplier int `validate:"gte=1"`
+	Width            int `validate:"gte=1"`
+}
+
+var validate = validator.New()
+
+func New(config Config) *B {
+
+	if err := validate.Struct(config); err != nil {
+		panic(err)
+	}
+
+	return &B{sb: &strings.Builder{}, width: config.Width, indentMultiplier: config.IndentMultiplier}
+}
+
+func (b *B) String() string { return b.sb.String() }
+
+func (w *B) Newline() {
+	w.Write("\n")
+}
+
+func (w *B) PrintfDrawIndentedAndWrappedIfMultiline(format string, args ...interface{}) {
+	whole := fmt.Sprintf(format, args...)
+	if strings.ContainsAny(whole, "\n\r") {
+		w.AddIndent(1)
+		defer w.AddIndent(-1)
+	}
+	w.Write(whole)
+}
+
+func (w *B) Printf(format string, args ...interface{}) {
+	whole := fmt.Sprintf(format, args...)
+	w.Write(whole)
+}
+
+func (t *B) AddIndent(delta int) {
+	t.indent += delta * t.indentMultiplier
+}
+
+func (t *B) AddIndentAndNewline(delta int) {
+	t.indent += delta * t.indentMultiplier
+	t.Write("\n")
+}
+
+func (w *B) Write(s string) {
+	for _, c := range s {
+		if c == '\n' {
+			fmt.Fprint(w.sb, "\n")
+			w.x = 0
+			fmt.Fprint(w.sb, Times(" ", w.indent-w.x))
+			w.x = w.indent
+			w.y++
+			continue
+		}
+		if w.x >= w.width {
+			fmt.Fprint(w.sb, "\n")
+			w.x = 0
+			fmt.Fprint(w.sb, Times(" ", w.indent-w.x))
+			w.x = w.indent
+		}
+		fmt.Fprintf(w.sb, "%c", c)
+		w.x++
+	}
+}
+
+func Times(str string, n int) (out string) {
+	for i := 0; i < n; i++ {
+		out += str
+	}
+	return
+}
+
+func RightPad(str string, length int, pad string) string {
+	if len(str) > length {
+		return str[:length]
+	}
+	return str + strings.Repeat(pad, length-len(str))
+}
+
+// changeCount = 0 indicates stall / no progress
+func (w *B) DrawBar(length int, bytes, totalBytes int64, changeCount int) {
+	const arrowPositions = `>\|/`
+	var completedLength int
+	if totalBytes > 0 {
+		completedLength = int(int64(length) * bytes / totalBytes)
+		if completedLength > length {
+			completedLength = length
+		}
+	} else if totalBytes == bytes {
+		completedLength = length
+	}
+
+	w.Write("[")
+	w.Write(Times("=", completedLength))
+	w.Write(string(arrowPositions[changeCount%len(arrowPositions)]))
+	w.Write(Times("-", length-completedLength))
+	w.Write("]")
+}
@@ -70,9 +70,9 @@ func TestPrometheusMonitoring(t *testing.T) {
 global:
  monitoring:
    - type: prometheus
-      listen: ':9091'
+      listen: ':9811'
 `)
-	assert.Equal(t, ":9091", conf.Global.Monitoring[0].Ret.(*PrometheusMonitoring).Listen)
+	assert.Equal(t, ":9811", conf.Global.Monitoring[0].Ret.(*PrometheusMonitoring).Listen)
 }

 func TestSyslogLoggingOutletFacility(t *testing.T) {
@@ -5,7 +5,7 @@
 # quick start section which inlines this example.
 #
 # CUSTOMIZATIONS YOU WILL LIKELY WANT TO APPLY:
-# - adjust the name of the production pool `system` in the `filesystems` filter of jobs `snapjob` and `push_to_derive`
+# - adjust the name of the production pool `system` in the `filesystems` filter of jobs `snapjob` and `push_to_drive`
 # - adjust the name of the backup pool `backuppool` in the `backuppool_sink` job
 # - adjust the occurences of `myhostname` to the name of the system you are backing up (cannot be easily changed once you start replicating)
 # - make sure the `zrepl_` prefix is not being used by any other zfs tools you might have installed (it likely isn't)
@@ -85,4 +85,4 @@ jobs:
  root_fs: "backuppool/zrepl/sink"
  serve:
    type: local
-    listener_name: backuppool_sink
+    listener_name: backuppool_sink
@@ -6,6 +6,7 @@
 .. |docs| replace:: [DOCS]
 .. |feature| replace:: [FEATURE]
 .. |mig| replace:: **[MIGRATION]**
+.. |maint| replace:: [MAINT]

 .. _changelog:

@@ -26,6 +27,38 @@ We use the following annotations for classifying changes:
 * |feature| Change that introduces new functionality.
 * |bugfix| Change that fixes a bug, no regressions or incompatibilities expected.
 * |docs| Change to the documentation.
+* |maint| Maintenance changes.
+
+0.4.0
+-----
+
+* |feature| support setting zfs send / recv flags in the config (send: ``-wLcepbS`` , recv: ``-ox`` ).
+  Config docs :ref:`here <job-send-options>` and :ref:`here <job-recv-options>` .
+* |feature| parallel replication is now configurable (disabled by default, :ref:`config docs here <replication-option-concurrency>` ).
+* |feature| New ``zrepl status`` UI:
+
+  * Interactive job selection.
+  * Interactively ``zrepl signal`` jobs.
+  * Filter filesystems in the job view by name.
+  * An approximation of the old UI is still included as `--mode legacy` but will be removed in a future release of zrepl.
+
+* |bugfix| Actually use concurrency when listing zrepl abstractions & doing size estimation.
+  These operations were accidentally made sequential in zrepl 0.3.
+* |bugfix| Job hang-up during second replication attempt.
+* |bugfix| Data races conditions in the dataconn rpc stack.
+* |maint| Update to protobuf v1.25 and grpc 1.35.
+
+For users who skipped the 0.3.1 update: please make sure your pruning grid config is correct.
+The following bugfix in 0.3.1 :issue:`caused problems for some users <400>`:
+
+* |bugfix| pruning: ``grid``:  add all snapshots that do not match the regex to the rule's destroy list.
+
+.. NOTE::
+  |  zrepl is a spare-time project primarily developed by `Christian Schwarz <https://cschwarz.com>`_.
+  |  You can support maintenance and feature development through one of the following services:
+  |  |Donate via Patreon| |Donate via GitHub Sponsors| |Donate via Liberapay| |Donate via PayPal|
+  |  Note that PayPal processing fees are relatively high for small donations.
+  |  For SEPA wire transfer and **commercial support**, please `contact Christian directly <https://cschwarz.com>`_.

 0.3.1
 -----
@@ -45,13 +78,6 @@ Mostly a bugfix release for :ref:`zrepl 0.3 <release-0.3>`.
 * CI infrastructure rework
 * Continuous deployment of that new `stable` branch to zrepl.github.io.

-.. NOTE::
-   |  zrepl is a spare-time project primarily developed by `Christian Schwarz <https://cschwarz.com>`_.
-   |  You can support maintenance and feature development through one of the following services:
-   |  |Donate via Patreon| |Donate via GitHub Sponsors| |Donate via Liberapay| |Donate via PayPal|
-   |  Note that PayPal processing fees are relatively high for small donations.
-   |  For SEPA wire transfer and **commercial support**, please `contact Christian directly <https://cschwarz.com>`_.
-
 .. _release-0.3:

 0.3
@@ -13,7 +13,7 @@ Prometheus & Grafana
 --------------------

 zrepl can expose `Prometheus metrics <https://prometheus.io/docs/instrumenting/exposition_formats/>`_ via HTTP.
-The ``listen`` attribute is a `net.Listen <https://golang.org/pkg/net/#Listen>`_  string for tcp, e.g. ``:9091`` or ``127.0.0.1:9091``.
+The ``listen`` attribute is a `net.Listen <https://golang.org/pkg/net/#Listen>`_  string for tcp, e.g. ``:9811`` or ``127.0.0.1:9811`` (port 9811 was reserved to zrepl `on the official list <https://github.com/prometheus/prometheus/wiki/Default-port-allocations/_compare/43e495dd251ee328ac0d08b58084665b5c0f7a7e...459195059b55b414193ebeb80c5ba463d2606951>`_).
 The ``listen_freebind`` attribute is :ref:`explained here <listen-freebind-explanation>`.
 The Prometheus monitoring job appears in the ``zrepl control`` job list and may be specified **at most once**.

@@ -30,7 +30,7 @@ The dashboard also contains some advice on which metrics are important to monito
    global:
      monitoring:
        - type: prometheus
-          listen: ':9091'
+          listen: ':9811'
          listen_freebind: true # optional, default false


@@ -124,43 +124,51 @@ The syntax to describe the bucket list is as follows:

 ::

-          This grid spec produces the following list of adjacent buckets. For the sake of simplicity,
-          we subject all snapshots to the grid pruning policy by settings `regex: .*`.
+   Assume the following grid spec:

-          `
-           grid: 1x1h(keep=all) | 2x2h | 1x3h
-           regex: .*
-          `
+      grid: 1x1h(keep=all) | 2x2h | 1x3h

-          0h        1h        2h        3h        4h        5h        6h        7h        8h        9h
-          |         |         |         |         |         |         |         |         |         |
-          |-Bucket1-|-----Bucket 2------|------Bucket 3-----|-----------Bucket 4----------|
-          | keep=all|      keep=1       |       keep=1      |            keep=1           |
+   This grid spec produces the following constellation of buckets:
+
+   0h        1h        2h        3h        4h        5h        6h        7h        8h        9h
+   |         |         |         |         |         |         |         |         |         |
+   |-Bucket1-|-----Bucket2-------|------Bucket3------|-----------Bucket4-----------|
+   | keep=all|      keep=1       |       keep=1      |            keep=1           |



-          Let us consider the following set of snapshots @a-zA-C:
+   Now assume that we have a set of snapshots @a, @b, ..., @D.
+   Snapshot @a is the most recent snapshot.
+   Snapshot @D is the oldest snapshot, it is almost 9 hours older than snapshot @a.
+   We place the snapshots on the same timeline as the buckets:


-          |  a  b  c  d  e  f  g  h  i  j  k  l  m  n  o  p  q  r  s  t  u  v  w  x  y  z  A  B  C  D        |
+   0h        1h        2h        3h        4h        5h        6h        7h        8h        9h
+   |         |         |         |         |         |         |         |         |         |
+   |-Bucket1-|-----Bucket2-------|------Bucket3------|-----------Bucket4-----------|
+   | keep=all|      keep=1       |       keep=1      |            keep=1           |
+   |         |                   |                   |                             |
+   |  a  b  c| d  e  f  g  h  i  j  k  l  m  n  o  p |q  r  s  t  u  v  w  x  y  z |A  B  C  D

-          The `grid` algorithm maps them to their respective buckets:
+   The result is the following mapping of snapshots to buckets:

-          Bucket 1: a, b, c
-          Bucket 2: d,e,f,g,h,i,j
-          Bucket 3: k,l,m,n,o,p
-          Bucket 4: q,r, q,r,s,t,u,v,w,x,y,z
-          None:     A,B,C,D
+   Bucket1:   a, b, c
+   Bucket2:   d,e,f,g,h,i,j
+   Bucket3:   k,l,m,n,o,p
+   Bucket4:   q,r,s,t,u,v,w,x,y,z
+   No bucket: A,B,C,D

-          It then applies the per-bucket pruning logic described above which resulting in the
-          following list of remaining snapshots.
+   For each bucket, we now prune snapshots until it only contains `keep` snapshots.
+   Newer snapshots are destroyed first.
+   Snapshots that do not fall into a bucket are always destroyed.

-          |  a  b  c                    j                 p                             z                    |
-
-          Note that it only makes sense to grow (not shorten) the interval duration for buckets
-          further in the past since each bucket acts like a low-pass filter for incoming snapshots
-          and adding a less-low-pass-filter after a low-pass one has no effect.
+   Result after pruning:

+   0h        1h        2h        3h        4h        5h        6h        7h        8h        9h
+   |         |         |         |         |         |         |         |         |         |
+   |-Bucket1-|-----Bucket2-------|------Bucket3------|-----------Bucket4-----------|
+   |         |                   |                   |                             |
+   |  a  b  c|                   j                 p |                           z  |

 .. _prune-keep-last-n:

@@ -186,6 +186,12 @@ Mount behaviour
 * ``canmount``
 * ``overlay``

+Note: inheriting or overriding the ``mountpoint`` property on ZVOLs fails in ``zfs recv``.
+This is an `issue in OpenZFS <https://github.com/openzfs/zfs/issues/11416>`_ .
+As a workaround, consider creating separate zrepl jobs for your ZVOL and filesystem datasets.
+Please comment at zrepl :issue:`430` if you encounter this issue and/or would like zrepl to automatically work around it.
+
+
 Systemd
 -------

@@ -13,13 +13,13 @@ The following snippet configure the repository for your Debian or Ubuntu release

 ::

-    apt update && apt install curl gnupg lsb-release; \
+    sudo apt update && sudo apt install curl gnupg lsb-release; \
    ARCH="$(dpkg --print-architecture)"; \
    CODENAME="$(lsb_release -i -s | tr '[:upper:]' '[:lower:]') $(lsb_release -c -s | tr '[:upper:]' '[:lower:]')"; \
    echo "Using Distro and Codename: $CODENAME"; \
-    (curl https://zrepl.cschwarz.com/apt/apt-key.asc | apt-key add -) && \
-    (echo "deb [arch=$ARCH] https://zrepl.cschwarz.com/apt/$CODENAME main" > /etc/apt/sources.list.d/zrepl.list) && \
-    apt update
+    (curl https://zrepl.cschwarz.com/apt/apt-key.asc | sudo apt-key add -) && \
+    (echo "deb [arch=$ARCH] https://zrepl.cschwarz.com/apt/$CODENAME main" | sudo tee /etc/apt/sources.list.d/zrepl.list) && \
+    sudo apt update


 .. NOTE::
@@ -1,6 +1,5 @@
 #!/bin/bash
-set -eo pipefail
-
+set -euo pipefail

 NON_INTERACTIVE=false
 DO_CLONE=false
@@ -13,7 +12,7 @@ while getopts "ca" arg; do
            DO_CLONE=true
            ;;
        *)
-            echo invalid option
+            echo "invalid option '-$arg'"
            exit 1
            ;;
    esac
@@ -27,11 +26,6 @@ checkout_repo_msg() {
    echo "clone ${GHPAGESREPO} to ${PUBLICDIR}:"
 }

-exit_msg() {
-    echo "error, exiting..."
-}
-trap exit_msg EXIT
-
 if ! type sphinx-versioning >/dev/null; then
    echo "install sphinx-versioning and come back"
    exit 1
@@ -71,6 +65,9 @@ git reset --hard origin/master

 echo "cleaning GitHub pages repo"
 git rm -rf .
+cat > .gitignore <<EOF
+**/.doctrees
+EOF

 popd

@@ -96,7 +93,13 @@ COMMIT_MSG="sphinx-versioning render from publish.sh - $(date -u) - ${CURRENT_CO
 pushd "$PUBLICDIR"

 echo "adding and commiting all changes in GitHub pages repo"
+git add .gitignore
 git add -A
-git commit -m "$COMMIT_MSG"
+if [ "$(git status --porcelain)" != "" ]; then
+    git commit -m "$COMMIT_MSG"
+else
+    echo "nothing to commit"
+fi
+echo "pushing to GitHub pages repo"
 git push origin master

@@ -5,10 +5,12 @@ go 1.12
 require (
 	github.com/fatih/color v1.7.0
 	github.com/gdamore/tcell v1.2.0
+	github.com/gdamore/tcell/v2 v2.2.0
 	github.com/gitchander/permutation v0.0.0-20181107151852-9e56b92e9909
 	github.com/go-logfmt/logfmt v0.4.0
 	github.com/go-playground/universal-translator v0.17.0 // indirect
 	github.com/go-playground/validator v9.31.0+incompatible
+	github.com/go-playground/validator/v10 v10.4.1
 	github.com/go-sql-driver/mysql v1.4.1-0.20190907122137-b2c03bcae3d4
 	github.com/golang/protobuf v1.4.3
 	github.com/google/uuid v1.1.2
@@ -37,6 +39,7 @@ require (
 	github.com/yudai/gojsondiff v0.0.0-20170107030110-7b1b7adf999d
 	github.com/yudai/golcs v0.0.0-20170316035057-ecda9a501e82 // go1.12 thinks it needs this
 	github.com/zrepl/yaml-config v0.0.0-20191220194647-cbb6b0cf4bdd
+	gitlab.com/tslocum/cview v1.5.3
 	golang.org/x/net v0.0.0-20210119194325-5f4716e94777
 	golang.org/x/sync v0.0.0-20190423024810-112230192c58
 	golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c
@@ -46,6 +46,11 @@ github.com/gdamore/encoding v1.0.0 h1:+7OoQ1Bc6eTm5niUzBa0Ctsh6JbMW6Ra+YNuAtDBdk
 github.com/gdamore/encoding v1.0.0/go.mod h1:alR0ol34c49FCSBLjhosxzcPHQbf2trDkoo5dl+VrEg=
 github.com/gdamore/tcell v1.2.0 h1:ikixzsxc8K8o3V2/CEmyoEW8mJZaNYQQ3NP3VIQdUe4=
 github.com/gdamore/tcell v1.2.0/go.mod h1:Hjvr+Ofd+gLglo7RYKxxnzCBmev3BzsS67MebKS4zMM=
+github.com/gdamore/tcell v1.4.0 h1:vUnHwJRvcPQa3tzi+0QI4U9JINXYJlOz9yiaiPQ2wMU=
+github.com/gdamore/tcell/v2 v2.0.0-dev/go.mod h1:vSVL/GV5mCSlPC6thFP5kfOFdM9MGZcalipmpTxTgQA=
+github.com/gdamore/tcell/v2 v2.1.1-0.20201225194624-29bb185874fd/go.mod h1:vSVL/GV5mCSlPC6thFP5kfOFdM9MGZcalipmpTxTgQA=
+github.com/gdamore/tcell/v2 v2.2.0 h1:vSyEgKwraXPSOkvCk7IwOSyX+Pv3V2cV9CikJMXg4U4=
+github.com/gdamore/tcell/v2 v2.2.0/go.mod h1:cTTuF84Dlj/RqmaCIV5p4w8uG1zWdk0SF6oBpwHp4fU=
 github.com/gitchander/permutation v0.0.0-20181107151852-9e56b92e9909 h1:9NC8seTx6/zRmMTAdsHj/uOMi0EGHGQtjyLafBjk77Q=
 github.com/gitchander/permutation v0.0.0-20181107151852-9e56b92e9909/go.mod h1:lP+DW8LR6Rw3ru9Vo2/y/3iiLaLWmofYql/va+7zJOk=
 github.com/go-critic/go-critic v0.3.4/go.mod h1:AHR42Lk/E/aOznsrYdMYeIQS5RH10HZHSqP+rD6AJrc=
@@ -59,12 +64,15 @@ github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9
 github.com/go-logfmt/logfmt v0.4.0 h1:MP4Eh7ZCb31lleYCFuwm0oe4/YGak+5l1vA2NOE80nA=
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
 github.com/go-ole/go-ole v1.2.1/go.mod h1:7FAglXiTm7HKlQRDeOQ6ZNUHidzCWXuZWq/1dTyBNF8=
+github.com/go-playground/assert/v2 v2.0.1/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4=
 github.com/go-playground/locales v0.13.0 h1:HyWk6mgj5qFqCT5fjGBuRArbVDfE4hi8+e8ceBS/t7Q=
 github.com/go-playground/locales v0.13.0/go.mod h1:taPMhCMXrRLJO55olJkUXHZBHCxTMfnGwq/HNwmWNS8=
 github.com/go-playground/universal-translator v0.17.0 h1:icxd5fm+REJzpZx7ZfpaD876Lmtgy7VtROAbHHXk8no=
 github.com/go-playground/universal-translator v0.17.0/go.mod h1:UkSxE5sNxxRwHyU+Scu5vgOQjsIJAF8j9muTVoKLVtA=
 github.com/go-playground/validator v9.31.0+incompatible h1:UA72EPEogEnq76ehGdEDp4Mit+3FDh548oRqwVgNsHA=
 github.com/go-playground/validator v9.31.0+incompatible/go.mod h1:yrEkQXlcI+PugkyDjY2bRrL/UBU4f3rvrgkN3V8JEig=
+github.com/go-playground/validator/v10 v10.4.1 h1:pH2c5ADXtd66mxoE0Zm9SUhxE20r7aM3F26W0hOn+GE=
+github.com/go-playground/validator/v10 v10.4.1/go.mod h1:nlOn6nFhuKACm19sB/8EGNn9GlaMV7XkbRSipzJ0Ii4=
 github.com/go-sql-driver/mysql v1.4.1-0.20190907122137-b2c03bcae3d4 h1:0suja/iKSDbEIYLbrS/8C7iArJiWpgCNcR+zwAHu7Ig=
 github.com/go-sql-driver/mysql v1.4.1-0.20190907122137-b2c03bcae3d4/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
@@ -177,6 +185,7 @@ github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORN
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/leodido/go-urn v1.2.0/go.mod h1:+8+nEpDfqqsY+g338gtMEUOtuK+4dEMhiQEgxpxOKII=
 github.com/leodido/go-urn v1.2.1 h1:BqpAaACuzVSgi/VLzGZIobT2z4v53pjosyNd9Yv6n/w=
 github.com/leodido/go-urn v1.2.1/go.mod h1:zt4jvISO2HfUBqxjfIshjdMTYS56ZS/qv49ictyFfxY=
 github.com/lib/pq v1.2.0 h1:LXpIM/LZ5xGFhOpXAQUIMM1HdyqzVYM13zNdjCEEcA0=
@@ -184,6 +193,8 @@ github.com/lib/pq v1.2.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
 github.com/logrusorgru/aurora v0.0.0-20181002194514-a7b3b318ed4e/go.mod h1:7rIyQOR62GCctdiQpZ/zOJlFyk6y+94wXzv6RNZgaR4=
 github.com/lucasb-eyer/go-colorful v1.0.2 h1:mCMFu6PgSozg9tDNMMK3g18oJBX7oYGrC09mS6CXfO4=
 github.com/lucasb-eyer/go-colorful v1.0.2/go.mod h1:0MS4r+7BZKSJ5mw4/S5MPN+qHFF1fYclkSPilDOKW0s=
+github.com/lucasb-eyer/go-colorful v1.0.3 h1:QIbQXiugsb+q10B+MI+7DI1oQLdmnep86tWFlaaUAac=
+github.com/lucasb-eyer/go-colorful v1.0.3/go.mod h1:R4dSotOR9KMtayYi1e77YzuveK+i7ruzyGqttikkLy0=
 github.com/magiconair/properties v1.7.6/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
 github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
 github.com/mattn/go-colorable v0.0.9 h1:UVL0vNpWh04HeJXV0KLcaT7r06gOH2l4OW6ddYRUIY4=
@@ -196,6 +207,10 @@ github.com/mattn/go-isatty v0.0.8 h1:HLtExJ+uU2HOZ+wI0Tt5DtUDrx8yhUqDcp7fYERX4CE
 github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
 github.com/mattn/go-runewidth v0.0.4 h1:2BvfKmzob6Bmd4YsL0zygOqfdFnK7GR4QL06Do4/p7Y=
 github.com/mattn/go-runewidth v0.0.4/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzpuz5H//U1FU=
+github.com/mattn/go-runewidth v0.0.7/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
+github.com/mattn/go-runewidth v0.0.9/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
+github.com/mattn/go-runewidth v0.0.10 h1:CoZ3S2P7pvtP45xOtBw+/mDL2z0RKI576gSkzRRpdGg=
+github.com/mattn/go-runewidth v0.0.10/go.mod h1:RAqKPSqVFrSLVXbA8x7dzmKdmGzieGRCM46jaSJTDAk=
 github.com/mattn/goveralls v0.0.2/go.mod h1:8d1ZMHsd7fW6IRPKQh46F2WRpyib5/X4FOpevwGNQEw=
 github.com/matttproud/golang_protobuf_extensions v1.0.0/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=
@@ -264,6 +279,9 @@ github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsT
 github.com/prometheus/procfs v0.0.5 h1:3+auTFlqw+ZaQYJARz6ArODtkaIwtvBTx3N2NehQlL8=
 github.com/prometheus/procfs v0.0.5/go.mod h1:4A/X28fw3Fc593LaREMrKMqOKvUAntwMDaekg4FpcdQ=
 github.com/quasilyte/go-consistent v0.0.0-20190521200055-c6f3937de18c/go.mod h1:5STLWrekHfjyYwxBRVRXNOSewLJ3PWfDJd1VyTS21fI=
+github.com/rivo/uniseg v0.1.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
+github.com/rivo/uniseg v0.2.0 h1:S1pD9weZBuJdFmowNwbpi7BJ8TNftyUImj/0WQi72jY=
+github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
 github.com/rogpeppe/go-internal v1.1.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.2.1/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/ryanuber/go-glob v0.0.0-20170128012129-256dc444b735/go.mod h1:807d1WSdnB0XRJzKNil9Om6lcp/3a0v4qIHxIXzX/Yc=
@@ -332,11 +350,17 @@ github.com/zrepl/yaml-config v0.0.0-20190928121844-af7ca3f8448f/go.mod h1:JmNwis
 github.com/zrepl/yaml-config v0.0.0-20191220194647-cbb6b0cf4bdd h1:SSo67WLS+99QESvbW8Meibz7zCrxshP71U9dH5KOCXM=
 github.com/zrepl/yaml-config v0.0.0-20191220194647-cbb6b0cf4bdd/go.mod h1:JmNwisZzOvW4GfpfLvhZ+gtyKLsIiA+WC+wNKJGJaFg=
 github.com/zrepl/zrepl v0.2.0/go.mod h1:M3Zv2IGSO8iYpUjsZD6ayZ2LHy7zyMfzet9XatKOrZ8=
+gitlab.com/tslocum/cbind v0.1.4 h1:cbZXPPcieXspk8cShoT6efz7HAT8yMNQcofYWNizis4=
+gitlab.com/tslocum/cbind v0.1.4/go.mod h1:RvwYE3auSjBNlCmWeGspzn+jdLUVQ8C2QGC+0nP9ChI=
+gitlab.com/tslocum/cview v1.5.3 h1:6OTCtIUp1EkfGeLqQFRHtW8ynMJ66BhoBwuW8oZ84AQ=
+gitlab.com/tslocum/cview v1.5.3/go.mod h1:k/eLWRIF3B26VLDgtRRPkjLUXmcCsy+YCSPEAtNQgIY=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190313024323-a1f597ede03a/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190325154230-a5d413f7728c/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 h1:psW17arqaxU48Z5kZ0CQnkZWQJsqcURM6tKiBApRjXI=
+golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/exp v0.0.0-20180321215751-8460e604b9de/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20180807140117-3d87b88a115f/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
@@ -357,6 +381,7 @@ golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73r
 golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190313220215-9f648a60d977/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190613194153-d28f0bde5980 h1:dfGZHvZk057jK2MCeWus/TowKpJ8y4AmooUzdBSR9GU=
 golang.org/x/net v0.0.0-20190613194153-d28f0bde5980/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20210119194325-5f4716e94777 h1:003p0dJM77cxMSyCPFphvZf/Y5/NXf5fzg6ufd1/Oew=
@@ -376,6 +401,7 @@ golang.org/x/sys v0.0.0-20181205085412-a5c9d58dba9a/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190312061237-fead79001313/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190626150813-e07cf5db2756 h1:9nuHUbU8dRnRRfj9KjWUVrJeoexdbeMjttk6Oh1rD10=
 golang.org/x/sys v0.0.0-20190626150813-e07cf5db2756/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -383,10 +409,14 @@ golang.org/x/sys v0.0.0-20191010194322-b09406accb47 h1:/XfQ9z7ib8eEJX2hdgFTZJ/nt
 golang.org/x/sys v0.0.0-20191010194322-b09406accb47/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191026070338-33540a1f6037 h1:YyJpGZS1sBuBCzLAR1VEpK193GlqGZbnPFnPV/5Rsb4=
 golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20201013132646-2da7054afaeb/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210113181707-4bcb84eeeb78/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c h1:VwygUrnw9jn88c4u8GD3rZQbqrP/tgas88tPUbBxQrk=
 golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/term v0.0.0-20201210144234-2321bbc49cbf h1:MZ2shdL+ZM/XzY3ZGOnh4Nlpnxz5GSOhOmtHo3iPU6M=
+golang.org/x/term v0.0.0-20201210144234-2321bbc49cbf/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.0.0-20170915090833-1cbadb444a80/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -4,12 +4,13 @@ package main
 import (
 	"github.com/zrepl/zrepl/cli"
 	"github.com/zrepl/zrepl/client"
+	"github.com/zrepl/zrepl/client/status"
 	"github.com/zrepl/zrepl/daemon"
 )

 func init() {
 	cli.AddSubcommand(daemon.DaemonCmd)
-	cli.AddSubcommand(client.StatusCmd)
+	cli.AddSubcommand(status.Subcommand)
 	cli.AddSubcommand(client.SignalCmd)
 	cli.AddSubcommand(client.StdinserverCmd)
 	cli.AddSubcommand(client.ConfigcheckCmd)
@@ -34,5 +34,10 @@ var Cases = []Case{BatchDestroy,
 	SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden__EncryptionSupported_true,
 	SendArgsValidationResumeTokenDifferentFilesystemForbidden,
 	SendArgsValidationResumeTokenEncryptionMismatchForbidden,
+	SendStreamCloseAfterBlockedOnPipeWrite,
+	SendStreamCloseAfterEOFRead,
+	SendStreamMultipleCloseAfterEOF,
+	SendStreamMultipleCloseBeforeEOF,
+	SendStreamNonEOFReadErrorHandling,
 	UndestroyableSnapshotParsing,
 }
@@ -40,6 +40,7 @@ func ReceiveForceRollbackWorksUnencrypted(ctx *platformtest.Context) {

 	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
 	require.NoError(ctx, err)
+	defer sendStream.Close()

 	recvOpts := zfs.RecvOptions{
 		RollbackAndForceRecv: true,
@@ -7,6 +7,7 @@ import (
 	"os"
 	"path"
 	"sort"
+	"strings"
 	"time"

 	"github.com/kr/pretty"
@@ -90,6 +91,7 @@ func (i replicationInvocation) Do(ctx *platformtest.Context) *report.Report {
 		ReplicationConfig: &pdu.ReplicationConfig{
 			Protection: i.guarantee,
 		},
+		SizeEstimationConcurrency: 1,
 	}

 	report, wait := replication.Do(
@@ -1054,6 +1056,9 @@ func ReplicationPropertyReplicationWorks(ctx *platformtest.Context) {
 		rep := fsByName[fs]
 		require.Len(ctx, rep.Steps, 1)
 		require.Nil(ctx, rep.PlanError)
+		if rep.StepError != nil && strings.Contains(rep.StepError.Error(), "invalid option 'x'") {
+			ctx.SkipNow() // XXX feature detection
+		}
 		require.Nil(ctx, rep.StepError)
 		require.Len(ctx, rep.Steps, 1)
 		require.Equal(ctx, 1, rep.CurrentStep)
@@ -0,0 +1,186 @@
+package tests
+
+import (
+	"fmt"
+	"io"
+	"io/ioutil"
+	"os"
+	"os/exec"
+	"path"
+	"syscall"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sys/unix"
+
+	"github.com/zrepl/zrepl/platformtest"
+	"github.com/zrepl/zrepl/util/nodefault"
+	"github.com/zrepl/zrepl/zfs"
+)
+
+func sendStreamTest(ctx *platformtest.Context) *zfs.SendStream {
+
+	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+		DESTROYROOT
+		CREATEROOT
+		+  "sender"
+	`)
+
+	fs := fmt.Sprintf("%s/sender", ctx.RootDataset)
+
+	fsmpo, err := zfs.ZFSGetMountpoint(ctx, fs)
+	require.NoError(ctx, err)
+
+	writeDummyData(path.Join(fsmpo.Mountpoint, "dummy.data"), 1<<26)
+	mustSnapshot(ctx, fs+"@1")
+	snap := fsversion(ctx, fs, "@1")
+	snapSendArg := snap.ToSendArgVersion()
+
+	sendArgs, err := zfs.ZFSSendArgsUnvalidated{
+		FS:   fs,
+		From: nil,
+		To:   &snapSendArg,
+		ZFSSendFlags: zfs.ZFSSendFlags{
+			Encrypted: &nodefault.Bool{B: false},
+		},
+	}.Validate(ctx)
+	require.NoError(ctx, err)
+
+	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
+	require.NoError(ctx, err)
+
+	return sendStream
+
+}
+
+func SendStreamCloseAfterBlockedOnPipeWrite(ctx *platformtest.Context) {
+
+	sendStream := sendStreamTest(ctx)
+
+	// let the pipe buffer fill and the zfs process block uninterruptibly
+
+	ctx.Logf("waiting for pipe write to block")
+	time.Sleep(5 * time.Second) // XXX need a platform-neutral way to detect that the pipe is full and the writer is blocked
+
+	ctx.Logf("closing send stream")
+	err := sendStream.Close() // this is what this test case is about
+	ctx.Logf("close error: %T %s", err, err)
+	require.NoError(ctx, err)
+
+	exitErrZfsError := sendStream.TestOnly_ExitErr()
+	require.Contains(ctx, exitErrZfsError.Error(), "signal")
+	require.Error(ctx, exitErrZfsError)
+	exitErr, ok := exitErrZfsError.WaitErr.(*exec.ExitError)
+	require.True(ctx, ok)
+	if exitErr.Exited() {
+		// some ZFS impls (FreeBSD 12) behaves that way
+		return
+	}
+
+	// ProcessState is only available after exit
+	// => use as proxy that the process was wait()ed upon and is gone
+	ctx.Logf("%#v", exitErr.ProcessState)
+	require.NotNil(ctx, exitErr.ProcessState)
+	// and let's verify that the process got killed, so that we know it was the call to .Close() above
+	waitStatus := exitErr.ProcessState.Sys().(syscall.WaitStatus)
+	ctx.Logf("wait status: %#v", waitStatus)
+	ctx.Logf("exit status: %v", waitStatus.ExitStatus())
+	require.True(ctx, waitStatus.Signaled())
+	switch waitStatus.Signal() {
+	case unix.SIGKILL:
+		fallthrough
+	case unix.SIGPIPE:
+		// ok
+
+	default:
+		ctx.Errorf("%T %s\n%v", waitStatus.Signal(), waitStatus.Signal(), waitStatus.Signal())
+		ctx.FailNow()
+	}
+}
+
+func SendStreamCloseAfterEOFRead(ctx *platformtest.Context) {
+
+	sendStream := sendStreamTest(ctx)
+
+	_, err := io.Copy(ioutil.Discard, sendStream)
+	require.NoError(ctx, err)
+
+	var buf [128]byte
+	n, err := sendStream.Read(buf[:])
+	require.Zero(ctx, n)
+	require.Equal(ctx, io.EOF, err)
+
+	err = sendStream.Close()
+	require.NoError(ctx, err)
+
+	n, err = sendStream.Read(buf[:])
+	require.Zero(ctx, n)
+	require.Equal(ctx, os.ErrClosed, err, "same read error should be returned")
+}
+
+func SendStreamMultipleCloseAfterEOF(ctx *platformtest.Context) {
+
+	sendStream := sendStreamTest(ctx)
+
+	_, err := io.Copy(ioutil.Discard, sendStream)
+	require.NoError(ctx, err)
+
+	var buf [128]byte
+	n, err := sendStream.Read(buf[:])
+	require.Zero(ctx, n)
+	require.Equal(ctx, io.EOF, err)
+
+	err = sendStream.Close()
+	require.NoError(ctx, err)
+
+	err = sendStream.Close()
+	require.Equal(ctx, os.ErrClosed, err)
+}
+
+func SendStreamMultipleCloseBeforeEOF(ctx *platformtest.Context) {
+
+	sendStream := sendStreamTest(ctx)
+
+	err := sendStream.Close()
+	require.NoError(ctx, err)
+
+	err = sendStream.Close()
+	require.Equal(ctx, os.ErrClosed, err)
+}
+
+type failingReadCloser struct {
+	err error
+}
+
+var _ io.ReadCloser = &failingReadCloser{}
+
+func (c *failingReadCloser) Read(p []byte) (int, error) { return 0, c.err }
+func (c *failingReadCloser) Close() error               { return c.err }
+
+func SendStreamNonEOFReadErrorHandling(ctx *platformtest.Context) {
+
+	sendStream := sendStreamTest(ctx)
+
+	var buf [128]byte
+	n, err := sendStream.Read(buf[:])
+	require.Equal(ctx, len(buf), n)
+	require.NoError(ctx, err)
+
+	var mockError = fmt.Errorf("taeghaefow4piesahwahjocu7ul5tiachaiLipheijae8ooZ8Pies8shohGee9feeTeirai5aiFeiyaecai4kiaLoh4azeih0tea")
+	mock := &failingReadCloser{err: mockError}
+	orig := sendStream.TestOnly_ReplaceStdoutReader(mock)
+
+	n, err = sendStream.Read(buf[:])
+	require.Equal(ctx, 0, n)
+	require.Equal(ctx, mockError, err)
+
+	if sendStream.TestOnly_ReplaceStdoutReader(orig) != mock {
+		panic("incorrect test impl")
+	}
+
+	err = sendStream.Close()
+	require.NoError(ctx, err) // if we can't kill the child then this will be a flaky test, but let's assume we can kill the child
+
+	err = sendStream.Close()
+	require.Equal(ctx, os.ErrClosed, err)
+}
@@ -57,6 +57,11 @@ func (k KeepLastN) KeepRule(snaps []Snapshot) (destroyList []Snapshot) {
 		// then lexicographically descending (e.g. b, a)
 		return strings.Compare(matching[i].Name(), matching[j].Name()) == 1
 	})
-	destroyList = append(destroyList, matching[k.n:]...)
+
+	n := k.n
+	if n > len(matching) {
+		n = len(matching)
+	}
+	destroyList = append(destroyList, matching[n:]...)
 	return destroyList
 }
@@ -83,6 +83,19 @@ func TestKeepLastN(t *testing.T) {
 				"b1": true,
 			},
 		},
+		"keep_more_than_matching": {
+			inputs: []Snapshot{
+				stubSnap{"a1", false, o(10)},
+				stubSnap{"b1", false, o(11)},
+				stubSnap{"a2", false, o(12)},
+			},
+			rules: []KeepRule{
+				MustKeepLastN(3, "a"),
+			},
+			expDestroy: map[string]bool{
+				"b1": true,
+			},
+		},
 	}

 	testTable(tcs, t)
@@ -0,0 +1,98 @@
+// Package choice implements a flag.Value type that accepts a set of choices.
+//
+// See test cases or grep the code base for usage hints.
+package choices
+
+import (
+	"flag"
+	"fmt"
+	"strings"
+)
+
+type Choices struct {
+	choices    map[string]interface{}
+	typeString string
+	value      interface{}
+}
+
+var _ flag.Value = (*Choices)(nil)
+
+func new(pairs ...interface{}) Choices {
+	if (len(pairs) % 2) != 0 {
+		panic("must provide a sequence of key value pairs")
+	}
+	c := Choices{
+		choices: make(map[string]interface{}, len(pairs)/2),
+		value:   nil,
+	}
+	for i := 0; i < len(pairs); {
+		key, ok := pairs[i].(string)
+		if !ok {
+			panic(fmt.Sprintf("argument %d is %T but should be a string, value: %#v", i, pairs[i], pairs[i]))
+		}
+		c.choices[key] = pairs[i+1]
+		i += 2
+	}
+	c.typeString = strings.Join(c.choicesList(true), ",") // overrideable by setter
+	return c
+}
+
+func (c *Choices) Init(pairs ...interface{}) {
+	*c = new(pairs...)
+}
+
+func (c Choices) choicesList(escaped bool) []string {
+	keys := make([]string, len(c.choices))
+	i := 0
+	for k := range c.choices {
+		e := k
+		if escaped {
+			e = fmt.Sprintf("%q", k)
+		}
+		keys[i] = e
+		i += 1
+	}
+	return keys
+}
+
+func (c Choices) Usage() string {
+	return fmt.Sprintf("one of %s", strings.Join(c.choicesList(true), ","))
+}
+
+func (c Choices) InputForChoice(v interface{}) (string, error) {
+	for input, choice := range c.choices {
+		if choice == v {
+			return input, nil
+		}
+	}
+	return "", fmt.Errorf("choice not registered at .Init(): %v", v)
+}
+
+func (c *Choices) SetDefaultValue(v interface{}) {
+	c.value = v
+}
+
+func (c Choices) Value() interface{} {
+	return c.value
+}
+
+func (c *Choices) Set(input string) error {
+	v, ok := c.choices[input]
+	if !ok {
+		return fmt.Errorf("invalid value %q: must be one of %s", input, c.Usage())
+	}
+	c.value = v
+	return nil
+}
+
+func (c *Choices) String() string {
+	return "" // c.value.(fmt.Stringer).String()
+}
+
+func (c *Choices) SetTypeString(ts string) {
+	c.typeString = ts
+}
+
+func (c *Choices) Type() string {
+	return c.typeString
+}
@@ -0,0 +1,50 @@
+package choices_test
+
+import (
+	"bytes"
+	"flag"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/zrepl/zrepl/util/choices"
+)
+
+func TestChoices(t *testing.T) {
+
+	var c choices.Choices
+
+	fs := flag.NewFlagSet("testset", flag.ContinueOnError)
+	c.Init("append", os.O_APPEND, "overwrite", os.O_TRUNC|os.O_CREATE)
+	fs.Var(&c, "mode", c.Usage())
+	var o bytes.Buffer
+	fs.SetOutput(&o)
+
+	fs.Usage()
+	usage := o.String()
+	o.Reset()
+
+	t.Logf("usage:\n%s", usage)
+	require.Contains(t, usage, "\"append\"")
+	require.Contains(t, usage, "\"overwrite\"")
+
+	err := fs.Parse([]string{"-mode", "append"})
+	require.NoError(t, err)
+	o.Reset()
+	require.Equal(t, os.O_APPEND, c.Value())
+
+	c.SetDefaultValue(nil)
+	err = fs.Parse([]string{})
+	require.NoError(t, err)
+	o.Reset()
+	require.Nil(t, c.Value())
+
+	// a little whitebox testing: this is allowed ATM, we don't check that the default value was specified as a choice in init
+	c.SetDefaultValue(os.O_RDWR)
+	err = fs.Parse([]string{})
+	require.NoError(t, err)
+	o.Reset()
+	require.Equal(t, os.O_RDWR, c.Value())
+
+}
@@ -38,14 +38,14 @@ func (i PropertySource) String() string {
 		return _PropertySourceName_3
 	case i == 32:
 		return _PropertySourceName_4
-	case i == 18446744073709551615:
+	case i == 4294967295:
 		return _PropertySourceName_5
 	default:
 		return fmt.Sprintf("PropertySource(%d)", i)
 	}
 }

-var _PropertySourceValues = []PropertySource{1, 2, 4, 8, 16, 32, 18446744073709551615}
+var _PropertySourceValues = []PropertySource{1, 2, 4, 8, 16, 32, 4294967295}

 var _PropertySourceNameToValueMap = map[string]PropertySource{
 	_PropertySourceName_0[0:5]:  1,
@@ -54,7 +54,7 @@ var _PropertySourceNameToValueMap = map[string]PropertySource{
 	_PropertySourceName_2[0:4]:  8,
 	_PropertySourceName_3[0:9]:  16,
 	_PropertySourceName_4[0:8]:  32,
-	_PropertySourceName_5[0:3]:  18446744073709551615,
+	_PropertySourceName_5[0:3]:  4294967295,
 }

 // PropertySourceString retrieves an enum value from the enum constants string name.
@@ -14,7 +14,6 @@ import (
 	"strconv"
 	"strings"
 	"sync"
-	"time"

 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
@@ -26,11 +25,6 @@ import (
 	"github.com/zrepl/zrepl/zfs/zfscmd"
 )

-var (
-	ZFSSendPipeCapacityHint = int(envconst.Int64("ZFS_SEND_PIPE_CAPACITY_HINT", 1<<25))
-	ZFSRecvPipeCapacityHint = int(envconst.Int64("ZFS_RECV_PIPE_CAPACITY_HINT", 1<<25))
-)
-
 type DatasetPath struct {
 	comps []string
 }
@@ -328,8 +322,8 @@ func absVersion(fs string, v *ZFSSendArgVersion) (full string, err error) {
 }

 func pipeWithCapacityHint(capacity int) (r, w *os.File, err error) {
-	if capacity <= 0 {
-		panic(fmt.Sprintf("capacity must be positive %v", capacity))
+	if capacity < 0 {
+		panic(fmt.Sprintf("capacity must be non-negative, got %v", capacity))
 	}
 	stdoutReader, stdoutWriter, err := os.Pipe()
 	if err != nil {
@@ -339,60 +333,122 @@ func pipeWithCapacityHint(capacity int) (r, w *os.File, err error) {
 	return stdoutReader, stdoutWriter, nil
 }

-type SendStream struct {
-	cmd  *zfscmd.Cmd
-	kill context.CancelFunc
+type sendStreamState int

-	closeMtx     sync.Mutex
-	stdoutReader *os.File
+const (
+	sendStreamOpen sendStreamState = iota
+	sendStreamClosed
+)
+
+type SendStream struct {
+	cmd          *zfscmd.Cmd
+	kill         context.CancelFunc
+	stdoutReader io.ReadCloser // not *os.File for mocking during platformtest
 	stderrBuf    *circlog.CircularLog
-	opErr        error
+
+	mtx     sync.Mutex
+	state   sendStreamState
+	exitErr *ZFSError
 }

-func (s *SendStream) Read(p []byte) (n int, err error) {
-	s.closeMtx.Lock()
-	opErr := s.opErr
-	s.closeMtx.Unlock()
-	if opErr != nil {
-		return 0, opErr
-	}
+func (s *SendStream) Read(p []byte) (n int, _ error) {
+	s.mtx.Lock()
+	defer s.mtx.Unlock()

-	n, err = s.stdoutReader.Read(p)
-	if err != nil {
-		debug("sendStream: read err: %T %s", err, err)
-		// TODO we assume here that any read error is permanent
-		// which is most likely the case for a local zfs send
-		kwerr := s.killAndWait(err)
-		debug("sendStream: killAndWait n=%v err= %T %s", n, kwerr, kwerr)
-		// TODO we assume here that any read error is permanent
-		return n, kwerr
+	switch s.state {
+	case sendStreamClosed:
+		return 0, os.ErrClosed
+
+	case sendStreamOpen:
+		n, readErr := s.stdoutReader.Read(p)
+		if readErr != nil {
+			debug("sendStream: read: readErr=%T %s", readErr, readErr)
+			if readErr == io.EOF {
+				// io.EOF must be bubbled up as is so that consumers can handle it properly.
+				return n, readErr
+			}
+			// Assume that the error is not retryable.
+			// Try to kill now so that we can return a nice *ZFSError with captured stderr.
+			// If the kill doesn't work, it doesn't matter because the caller must by contract call Close() anyways.
+			killErr := s.killAndWait()
+			debug("sendStream: read: killErr=%T %s", killErr, killErr)
+			if killErr == nil {
+				s.state = sendStreamClosed
+				return n, s.exitErr // return the nice error
+			} else {
+				// we remain open so that we retry
+				return n, readErr // return the normal error
+			}
+		}
+		return n, readErr
+
+	default:
+		panic("unreachable")
 	}
-	return n, err
 }

 func (s *SendStream) Close() error {
 	debug("sendStream: close called")
-	return s.killAndWait(nil)
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+
+	switch s.state {
+	case sendStreamOpen:
+		err := s.killAndWait()
+		if err != nil {
+			return err
+		} else {
+			s.state = sendStreamClosed
+			return nil
+		}
+	case sendStreamClosed:
+		return os.ErrClosed
+	default:
+		panic("unreachable")
+	}
 }

-func (s *SendStream) killAndWait(precedingReadErr error) error {
+// returns nil iff the child process is gone (has been successfully waited upon)
+// in that case, s.exitErr is set
+func (s *SendStream) killAndWait() error {

 	debug("sendStream: killAndWait enter")
 	defer debug("sendStream: killAndWait leave")
-	if precedingReadErr == io.EOF {
-		// give the zfs process a little bit of time to terminate itself
-		// if it holds this deadline, exitErr will be nil
-		time.AfterFunc(200*time.Millisecond, s.kill)
-	} else {
-		s.kill()
-	}

-	// allow async kills from Close(), that's why we only take the mutex here
-	s.closeMtx.Lock()
-	defer s.closeMtx.Unlock()
+	// send SIGKILL
+	s.kill()

-	if s.opErr != nil {
-		return s.opErr
+	// Close our read-end of the pipe.
+	//
+	// We must do this before .Wait() because in some (not all) versions/build configs of ZFS,
+	// `zfs send` uses a separate kernel thread (taskq) to write the send stream (function `dump_bytes`).
+	// The `zfs send` thread then waits uinterruptably for the taskq thread to finish the write.
+	// And signalling the `zfs send` thread doesn't propagate to the taskq thread.
+	// So we end up in a state where we .Wait() forever.
+	// (See https://github.com/openzfs/zfs/issues/12500 and
+	//  https://github.com/zrepl/zrepl/issues/495#issuecomment-902530043)
+	//
+	// By closing our read end of the pipe before .Wait(), we unblock the taskq thread if there is any.
+	// If there is no separate taskq thread, the SIGKILL to `zfs end` would suffice and be most precise,
+	// but due to the circumstances above, there is no other portable & robust way.
+	//
+	// However, the fallout from closing the pipe is that (in non-taskq builds) `zfs sends` will get a SIGPIPE.
+	// And on Linux, that SIGPIPE appears to win over the previously issued SIGKILL.
+	// And thus, on Linux, the `zfs send` will be killed by the default SIGPIPE handler.
+	// We can observe this in the WaitStatus below.
+	// This behavior is slightly annoying because the *exec.ExitError's message ("signal: broken pipe")
+	// isn't as clear as ("signal: killed").
+	// However, it seems like we just have to live with that. (covered by platformtest)
+	var closePipeErr error
+	if s.stdoutReader != nil {
+		closePipeErr = s.stdoutReader.Close()
+		if closePipeErr == nil {
+			// avoid double-closes in case waiting below doesn't work
+			// and someone attempts Close again
+			s.stdoutReader = nil
+		} else {
+			return closePipeErr
+		}
 	}

 	waitErr := s.cmd.Wait()
@@ -407,39 +463,30 @@ func (s *SendStream) killAndWait(precedingReadErr error) error {
 		}
 	}

-	// now, after we know the program exited do we close the pipe
-	var closePipeErr error
-	if s.stdoutReader != nil {
-		closePipeErr = s.stdoutReader.Close()
-		if closePipeErr == nil {
-			// avoid double-closes in case anything below doesn't work
-			// and someone calls Close again
-			s.stdoutReader = nil
-		} else {
-			return closePipeErr
-		}
-	}
+	// invariant: at this point, the child is gone and we cleaned up everything related to the SendStream

-	// we managed to tear things down, no let's give the user some pretty *ZFSError
 	if exitErr != nil {
-		s.opErr = &ZFSError{
+		// zfs send exited with an error or was killed by a signal.
+		s.exitErr = &ZFSError{
 			Stderr:  []byte(s.stderrBuf.String()),
 			WaitErr: exitErr,
 		}
 	} else {
-		s.opErr = precedingReadErr
+		// zfs send exited successfully (we know that since waitErr was either nil or wasn't an *exec.ExitError)
+		s.exitErr = nil
 	}

-	// detect the edge where we're called from s.Read
-	// after the pipe EOFed and zfs send exited without errors
-	// this is actually the "hot" / nice path
-	if exitErr == nil && precedingReadErr == io.EOF {
-		return precedingReadErr
-	}
-
-	return s.opErr
+	return nil
 }

+func (s *SendStream) TestOnly_ReplaceStdoutReader(f io.ReadCloser) (prev io.ReadCloser) {
+	prev = s.stdoutReader
+	s.stdoutReader = f
+	return prev
+}
+
+func (s *SendStream) TestOnly_ExitErr() *ZFSError { return s.exitErr }
+
 // NOTE: When updating this struct, make sure to update funcs Validate ValidateCorrespondsToResumeToken
 type ZFSSendArgVersion struct {
 	RelName string
@@ -858,7 +905,7 @@ func ZFSSend(ctx context.Context, sendArgs ZFSSendArgsValidated) (*SendStream, e
 	ctx, cancel := context.WithCancel(ctx)

 	// setup stdout with an os.Pipe to control pipe buffer size
-	stdoutReader, stdoutWriter, err := pipeWithCapacityHint(ZFSSendPipeCapacityHint)
+	stdoutReader, stdoutWriter, err := pipeWithCapacityHint(getPipeCapacityHint("ZFS_SEND_PIPE_CAPACITY_HINT"))
 	if err != nil {
 		cancel()
 		return nil, err
@@ -1160,7 +1207,7 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, stream io.Rea

 	stderr := bytes.NewBuffer(make([]byte, 0, RecvStderrBufSiz))

-	stdin, stdinWriter, err := pipeWithCapacityHint(ZFSRecvPipeCapacityHint)
+	stdin, stdinWriter, err := pipeWithCapacityHint(getPipeCapacityHint("ZFS_RECV_PIPE_CAPACITY_HINT"))
 	if err != nil {
 		return err
 	}
@@ -1446,7 +1493,7 @@ func tryDatasetDoesNotExist(expectPath string, stderr []byte) *DatasetDoesNotExi
 }

 //go:generate enumer -type=PropertySource -trimprefix=Source
-type PropertySource uint
+type PropertySource uint32

 const (
 	SourceLocal PropertySource = 1 << iota
@@ -7,10 +7,14 @@ import (
 	"sync"
 )

+func getPipeCapacityHint(envvar string) int {
+	return 0 // not supported
+}
+
 var zfsPipeCapacityNotSupported sync.Once

 func trySetPipeCapacity(p *os.File, capacity int) {
-	if debugEnabled {
+	if debugEnabled && capacity != 0 {
 		zfsPipeCapacityNotSupported.Do(func() {
 			debug("trySetPipeCapacity error: OS does not support setting pipe capacity")
 		})
@@ -3,11 +3,33 @@ package zfs
 import (
 	"errors"
 	"fmt"
+	"io/ioutil"
 	"os"
+	"strconv"
+	"strings"

 	"golang.org/x/sys/unix"
+
+	"github.com/zrepl/zrepl/util/envconst"
 )

+func getPipeCapacityHint(envvar string) int {
+	var capacity int64 = 1 << 25
+
+	// Work around a race condition in Linux >= 5.8 related to pipe resizing.
+	// https://github.com/zrepl/zrepl/issues/424#issuecomment-800370928
+	// https://bugzilla.kernel.org/show_bug.cgi?id=212295
+	if _, err := os.Stat("/proc/sys/fs/pipe-max-size"); err == nil {
+		if dat, err := ioutil.ReadFile("/proc/sys/fs/pipe-max-size"); err == nil {
+			if capacity, err = strconv.ParseInt(strings.TrimSpace(string(dat)), 10, 64); err != nil {
+				capacity = 1 << 25
+			}
+		}
+	}
+
+	return int(envconst.Int64(envvar, capacity))
+}
+
 func trySetPipeCapacity(p *os.File, capacity int) {
 	res, err := unix.FcntlInt(p.Fd(), unix.F_SETPIPE_SZ, capacity)
 	if err != nil {
@@ -209,3 +209,7 @@ func (c *Cmd) Runtime() time.Duration {
 	}
 	return c.waitReturnedAt.Sub(c.startedAt)
 }
+
+func (c *Cmd) TestOnly_ExecCmd() *exec.Cmd {
+	return c.cmd
+}
Author	SHA1	Message	Date
Christian Schwarz	226962b4fb	zfs: rewrite SendStream, fix bug in Close() on FreeBSD, add platformtests This commit was motivated by https://github.com/zrepl/zrepl/issues/495 where, on FreeBSD with OpenZFS 2.0, a SendStream.Close() call might wait indefinitely for `zfs send` to exit. The reason is that, due to the refactoring done for redacted send & recv (https://github.com/openzfs/zfs/commit/30af21b02569ac192f52ce6e6511015f8a8d5729), the `dump_bytes` function, which writes to the pipe, executes in a separate thread (synctask taskq) iff not `HAVE_LARGE_STACKS`. The `zfs send` process/thread waits for that taskq thread using an uninterruptible primitive. So when we SIGKILL `zfs send`, that signal doesn't reach the right thread to interrupt the pipe write. Theoretically this affects both Linux and FreeBSD, but most Linux users `HAVE_LARGE_STACKS` and since https://github.com/penzfs/zfs/pull/12350/files OpenZFS on FreeBSD `HAVE_LARGE_STACKS` as well. However, at least until FreeBSD 13.1, possibly for the entire 13 lifecycle, we're going to have to live with that oddity. Measures taken in this commit: - Report the behavior as an upstream bug https://github.com/openzfs/zfs/issues/12500 - Change SendStream code so that it closes zrepl's read-end of the pipe (see comment in code) - Clean up and make explicit SendStream's state handling - Write extensive platformtests for SendStream - They pass on my Linux install and on FreeBSD 12 - FreeBSD 13 still needs testing. fixes https://github.com/zrepl/zrepl/issues/495	2021-08-23 00:21:34 +02:00
Christian Schwarz	ff6ebe87f3	platformtest: fix 'active child tasks' panic for ReceiveForceRollbackWorksUnencrypted Revealed by rework of SendStream in a prior commit.	2021-08-23 00:21:34 +02:00
Christian Schwarz	126874b423	platformtest: fix replication tests (SizeEstimationConcurrency field in PlannerPolicy was not set) fixup of `0ceea1b792` (parallel replication knobs)	2021-08-23 00:21:34 +02:00
Christian Schwarz	157753d781	platformtest: work around missing feature detection for test 'ReplicationPropertyReplicationWorks'	2021-08-23 00:21:34 +02:00
Christian Schwarz	009bd410af	docs: prune: improve `grid` example	2021-07-08 19:46:24 +02:00
Christian Schwarz	bcfcd7a134	docs / CI: stop creating churn with doc commits & commit as zreplbot@	2021-07-08 17:07:24 +02:00
Matthias Freund	bf1276f767	status: port status-v1 ETA calculation patch Must have forgotten to integrate it into the status-v2 branch at the time. refs https://github.com/zrepl/zrepl/issues/98#issuecomment-872154091 cc @dcdamien	2021-07-08 15:00:26 +02:00
James W. Brinkerhoff IV	9fa7a18351	docs: quickstart: external_disk: fix typo in example 'derive -> drive' closes #472	2021-04-19 23:36:03 +02:00
sre	50e8ee4549	docs: apt repo: use sudo in the snippet that sets up the repo I generally like when snippets are provided in a way which could be used without running as root, and uses sudo when applicable. This change allows for this. It will, however print out one extra line, which is possible to remove by adding '>/dev/null' after '/etc/apt/sources.list.d/zrepl.list'. closes #461	2021-04-17 21:50:36 +02:00
Lapo Luchini	3b5a1a8b9a	docs/monitoring: change suggested prometheus port to 9811 Change to 9811 as registered with the prometheus project now. Closes #444.	2021-03-28 18:18:02 +02:00
Christian Schwarz	f661d9429f	pruning/keep_last_n: correctly handle the case where `count` > matching snaps fixes #446	2021-03-25 22:42:01 +01:00
InsanePrawn	ac4b109872	status/interactive: Revert to simple wakeup/reset signalling Signed-off-by: InsanePrawn <insane.prawny@gmail.com> closes #452	2021-03-25 22:26:17 +01:00
InsanePrawn	b2c6e51a43	client/signal: Revert "add signal 'snapshot', rename existing signal 'wakeup' to 'replication'" This was merged to master prematurely as the job components are not decoupled well enough for these signals to be useful yet. This reverts commit `2c8c2cfa14`. closes #452	2021-03-25 22:26:17 +01:00
Lukas Schauer	ee2336a24b	zfs: pipe size: default to value of /proc/sys/fs/pipe-max-siz Addition by @problame: move getPipeCapacityHint() into platform-specific code. This has the added benefit of not recognizing the envvar as an envconst on platform that do not support resizing pipes. => won't show up in (zrepl status --raw).Global.Envconst fixes #424 cloes #449	2021-03-25 22:24:50 +01:00
Cole Helbling	1e85b1cb5f	client/status: allow raw mode without a tty A fairly common pattern when presented with JSON is to pipe it to `jq`. This also allows one to `zrepl status --mode raw > log` and operate on the JSON there. closes #442	2021-03-22 23:58:32 +01:00
Christian Schwarz	5c6d69a69c	zfs: PropertySource: set type to uint32 so that `enumer`-generated code is platform-independent make zrepl-bin test-platform-bin vet lint GOOS=freebsd GOARCH=386 make[2]: Entering directory '/src' GO111MODULE=on go build -mod=readonly -ldflags "-X github.com/zrepl/zrepl/version.zreplVersion=v0.3.1-20-g07f2bff" -o "artifacts/zrepl-freebsd-386" zfs/propertysource_enumer.go:41:9: constant 18446744073709551615 overflows PropertySource zfs/propertysource_enumer.go:48:66: constant 18446744073709551615 overflows PropertySource zfs/propertysource_enumer.go:57:23: constant 18446744073709551615 overflows PropertySource fixes #429	2021-03-14 22:32:45 +01:00
Christian Schwarz	299808aaaf	docs: 0.4 changelog: mention the 0.3.1 fix to prune:grid fixes #400 refs `3a4e841c73`	2021-03-14 21:13:26 +01:00
Christian Schwarz	7071adc774	docs: 0.4 changelog	2021-03-14 20:56:02 +01:00
InsanePrawn	8d678eed19	docs: add note about `zfs recv -x mountpoint` with ZVOLs refs #430	2021-03-14 20:26:39 +01:00
Calistoc	ab7abd9686	status: show the current replication attempt's runtime	2021-03-14 18:24:28 +01:00
Christian Schwarz	a58ce74ed0	implement new 'zrepl status' Primary goals: - Scrollable output ( fixes #245 ) - Sending job signals from status view - Filtering of output by filesystem Implementation: - original TUI framework: github.com/rivo/tview - but: tview is quasi-unmaintained, didn't support some features - => use fork https://gitlab.com/tslocum/cview - however, don't buy into either too much to avoid lock-in - instead: port over the existing status UI drawing code and adjust it to produce strings instead of directly drawing into the termbox buffer Co-authored-by: Calistoc <calistoc@protonmail.com> Co-authored-by: InsanePrawn <insane.prawny@gmail.com> fixes #245 fixes #220	2021-03-14 18:24:25 +01:00
Calistoc	2c8c2cfa14	add signal 'snapshot', rename existing signal 'wakeup' to 'replication'	2021-03-14 18:16:23 +01:00