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

.circleci/config.yml

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

client/reset.go

@@ -1,88 +0,0 @@
-package client
-
-import (
-	"context"
-	"strconv"
-	"time"
-
-	"github.com/kr/pretty"
-	"github.com/pkg/errors"
-	"github.com/spf13/pflag"
-
-	"github.com/zrepl/zrepl/cli"
-	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon"
-	"github.com/zrepl/zrepl/daemon/job"
-)
-
-var resetCmdArgs struct {
-	verbose  bool
-	interval time.Duration
-	token    string
-}
-
-var ResetCmd = &cli.Subcommand{
-	Use:   "reset [-t TOKEN | JOB INVOCATION [replication|snapshotting|prune_sender|prune_receiver]]",
-	Short: "",
-	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
-		return runResetCmd(subcommand.Config(), args)
-	},
-	SetupFlags: func(f *pflag.FlagSet) {
-		f.BoolVarP(&resetCmdArgs.verbose, "verbose", "v", false, "verbose output")
-		f.DurationVarP(&resetCmdArgs.interval, "poll-interval", "i", 100*time.Millisecond, "poll interval")
-		f.StringVarP(&resetCmdArgs.token, "token", "t", "", "token produced by 'signal' subcommand")
-	},
-}
-
-func runResetCmd(config *config.Config, args []string) error {
-
-	httpc, err := controlHttpClient(config.Global.Control.SockPath)
-	if err != nil {
-		return err
-	}
-
-	var req daemon.ControlJobEndpointResetActiveRequest
-	if resetCmdArgs.token != "" {
-		var token TriggerToken
-		err := token.Decode(resetCmdArgs.token)
-		if err != nil {
-			return errors.Wrap(err, "cannot decode token")
-		}
-		req = token.ToReset()
-	} else {
-		jobName := args[0]
-
-		invocationId, err := strconv.ParseUint(args[1], 10, 64)
-		if err != nil {
-			return errors.Wrap(err, "parse invocation id")
-		}
-
-		// what := args[2]
-
-		// updated by subsequent requests
-		req = daemon.ControlJobEndpointResetActiveRequest{
-			Job: jobName,
-			ActiveSideResetRequest: job.ActiveSideResetRequest{
-				InvocationId: invocationId,
-			},
-		}
-	}
-
-	var res job.ActiveSideResetResponse
-	if resetCmdArgs.verbose {
-		pretty.Println("making request", req)
-	}
-	err = jsonRequestResponse(httpc, daemon.ControlJobEndpointResetActive,
-		req,
-		&res,
-	)
-	if err != nil {
-		return err
-	}
-
-	if resetCmdArgs.verbose {
-		pretty.Println("got response", res)
-	}
-
-	return nil
-}

client/signal.go

@@ -0,0 +1,42 @@
+package client
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+
+	"github.com/zrepl/zrepl/cli"
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon"
+)
+
+var SignalCmd = &cli.Subcommand{
+	Use:   "signal [wakeup|reset] JOB",
+	Short: "wake up a job from wait state or abort its current invocation",
+	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
+		return runSignalCmd(subcommand.Config(), args)
+	},
+}
+
+func runSignalCmd(config *config.Config, args []string) error {
+	if len(args) != 2 {
+		return errors.Errorf("Expected 2 arguments: [wakeup|reset] JOB")
+	}
+
+	httpc, err := controlHttpClient(config.Global.Control.SockPath)
+	if err != nil {
+		return err
+	}
+
+	err = jsonRequestResponse(httpc, daemon.ControlJobEndpointSignal,
+		struct {
+			Name string
+			Op   string
+		}{
+			Name: args[1],
+			Op:   args[0],
+		},
+		struct{}{},
+	)
+	return err
+}

client/status/client/client.go

@@ -11,7 +11,6 @@ import (
 	"github.com/pkg/errors"
 
 	"github.com/zrepl/zrepl/daemon"
-	"github.com/zrepl/zrepl/daemon/job"
 )
 
 type Client struct {
@@ -43,27 +42,21 @@ func (c *Client) StatusRaw() ([]byte, error) {
 	return r, nil
 }
 
-func (c *Client) signal(jobName, sig string) error {
+func (c *Client) signal(job, sig string) error {
-	return jsonRequestResponse(c.h, daemon.ControlJobEndpointTriggerActive,
+	return jsonRequestResponse(c.h, daemon.ControlJobEndpointSignal,
 		struct {
-			Job string
+			Name string
-			job.ActiveSideTriggerRequest
+			Op   string
 		}{
-			Job: jobName,
+			Name: job,
-			ActiveSideTriggerRequest: job.ActiveSideTriggerRequest{
+			Op:   sig,
-				What: sig,
-			},
 		},
 		struct{}{},
 	)
 }
 
-func (c *Client) SignalReplication(job string) error {
+func (c *Client) SignalWakeup(job string) error {
-	return c.signal(job, "replication")
+	return c.signal(job, "wakeup")
-}
-
-func (c *Client) SignalSnapshot(job string) error {
-	return c.signal(job, "snapshot")
 }
 
 func (c *Client) SignalReset(job string) error {

client/status/status.go

@@ -19,8 +19,7 @@ import (
 type Client interface {
 	Status() (daemon.Status, error)
 	StatusRaw() ([]byte, error)
-	SignalReplication(job string) error
+	SignalWakeup(job string) error
-	SignalSnapshot(job string) error
 	SignalReset(job string) error
 }
 
@@ -71,12 +70,16 @@ func runStatusV2Command(ctx context.Context, config *config.Config, args []strin
 
 	mode := statusv2Flags.Mode.Value().(statusv2Mode)
 
-	if !isatty.IsTerminal(os.Stdout.Fd()) && mode != StatusV2ModeDump {
+	if !isatty.IsTerminal(os.Stdout.Fd()) && mode != StatusV2ModeDump && mode != StatusV2ModeRaw {
-		usemode, err := statusv2Flags.Mode.InputForChoice(StatusV2ModeDump)
+		dumpmode, err := statusv2Flags.Mode.InputForChoice(StatusV2ModeDump)
 		if err != nil {
 			panic(err)
 		}
-		return errors.Errorf("error: stdout is not a tty, please use --mode %s", usemode)
+		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 {

client/status/status_interactive.go

@@ -281,8 +281,8 @@ func interactive(c Client, flag statusFlags) error {
 			if !ok {
 				return nil
 			}
-			signals := []string{"replication", "snapshot", "reset"}
+			signals := []string{"wakeup", "reset"}
-			clientFuncs := []func(job string) error{c.SignalReplication, c.SignalSnapshot, c.SignalReset}
+			clientFuncs := []func(job string) error{c.SignalWakeup, c.SignalReset}
 			sigMod := tview.NewModal()
 			sigMod.SetBackgroundColor(tcell.ColorDefault)
 			sigMod.SetBorder(true)

client/status/viewmodel/render.go

@@ -356,9 +356,17 @@ func renderReplicationReport(t *stringbuilder.B, rep *report.Report, history *by
 		// 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!")
@@ -383,6 +391,30 @@ func renderReplicationReport(t *stringbuilder.B, rep *report.Report, history *by
 	}
 }
 
+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")

client/trigger.go

@@ -1,94 +0,0 @@
-package client
-
-import (
-	"context"
-	"encoding/json"
-	"fmt"
-
-	"github.com/pkg/errors"
-
-	"github.com/zrepl/zrepl/cli"
-	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon"
-	"github.com/zrepl/zrepl/daemon/job"
-)
-
-var TriggerCmd = &cli.Subcommand{
-	Use:   "trigger JOB [replication|snapshot]",
-	Short: "",
-	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
-		return runTriggerCmd(subcommand.Config(), args)
-	},
-}
-
-type TriggerToken struct {
-	// TODO version, daemon invocation id, etc.
-	Job          string
-	InvocationId uint64
-}
-
-func (t TriggerToken) Encode() string {
-	j, err := json.Marshal(t)
-	if err != nil {
-		panic(err)
-	}
-	return string(j)
-}
-
-func (t *TriggerToken) Decode(s string) error {
-	return json.Unmarshal([]byte(s), t)
-}
-
-func (t TriggerToken) ToReset() daemon.ControlJobEndpointResetActiveRequest {
-	return daemon.ControlJobEndpointResetActiveRequest{
-		Job: t.Job,
-		ActiveSideResetRequest: job.ActiveSideResetRequest{
-			InvocationId: t.InvocationId,
-		},
-	}
-}
-
-func (t TriggerToken) ToWait() daemon.ControlJobEndpointWaitActiveRequest {
-	return daemon.ControlJobEndpointWaitActiveRequest{
-		Job: t.Job,
-		ActiveSidePollRequest: job.ActiveSidePollRequest{
-			InvocationId: t.InvocationId,
-		},
-	}
-}
-
-func runTriggerCmd(config *config.Config, args []string) error {
-	if len(args) != 2 {
-		return errors.Errorf("Expected 2 arguments: [replication|reset|snapshot] JOB")
-	}
-
-	httpc, err := controlHttpClient(config.Global.Control.SockPath)
-	if err != nil {
-		return err
-	}
-
-	jobName := args[0]
-	what := args[1]
-
-	var res job.ActiveSideSignalResponse
-	err = jsonRequestResponse(httpc, daemon.ControlJobEndpointTriggerActive,
-		struct {
-			Job string
-			job.ActiveSideTriggerRequest
-		}{
-			Job: jobName,
-			ActiveSideTriggerRequest: job.ActiveSideTriggerRequest{
-				What: what,
-			},
-		},
-		&res,
-	)
-
-	token := TriggerToken{
-		Job:          jobName,
-		InvocationId: res.InvocationId,
-	}
-
-	fmt.Println(token.Encode())
-	return err
-}

client/wait.go

@@ -1,110 +0,0 @@
-package client
-
-import (
-	"context"
-	"fmt"
-	"strconv"
-	"time"
-
-	"github.com/kr/pretty"
-	"github.com/pkg/errors"
-	"github.com/spf13/pflag"
-
-	"github.com/zrepl/zrepl/cli"
-	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon"
-	"github.com/zrepl/zrepl/daemon/job"
-)
-
-var waitCmdArgs struct {
-	verbose  bool
-	interval time.Duration
-	token    string
-}
-
-var WaitCmd = &cli.Subcommand{
-	Use:   "wait [-t TOKEN | JOB INVOCATION [replication|snapshotting|prune_sender|prune_receiver]]",
-	Short: "",
-	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
-		return runWaitCmd(subcommand.Config(), args)
-	},
-	SetupFlags: func(f *pflag.FlagSet) {
-		f.BoolVarP(&waitCmdArgs.verbose, "verbose", "v", false, "verbose output")
-		f.DurationVarP(&waitCmdArgs.interval, "poll-interval", "i", 100*time.Millisecond, "poll interval")
-		f.StringVarP(&waitCmdArgs.token, "token", "t", "", "token produced by 'signal' subcommand")
-	},
-}
-
-func runWaitCmd(config *config.Config, args []string) error {
-
-	httpc, err := controlHttpClient(config.Global.Control.SockPath)
-	if err != nil {
-		return err
-	}
-
-	var req daemon.ControlJobEndpointWaitActiveRequest
-	if waitCmdArgs.token != "" {
-		var token TriggerToken
-		err := token.Decode(resetCmdArgs.token)
-		if err != nil {
-			return errors.Wrap(err, "cannot decode token")
-		}
-		req = token.ToWait()
-	} else {
-
-		jobName := args[0]
-
-		invocationId, err := strconv.ParseUint(args[1], 10, 64)
-		if err != nil {
-			return errors.Wrap(err, "parse invocation id")
-		}
-
-		// updated by subsequent requests
-		req = daemon.ControlJobEndpointWaitActiveRequest{
-			Job: jobName,
-			ActiveSidePollRequest: job.ActiveSidePollRequest{
-				InvocationId: invocationId,
-			},
-		}
-	}
-
-	doneErr := fmt.Errorf("done")
-
-	pollOnce := func() error {
-		var res job.ActiveSidePollResponse
-		if waitCmdArgs.verbose {
-			pretty.Println("making poll request", req)
-		}
-		err = jsonRequestResponse(httpc, daemon.ControlJobEndpointPollActive,
-			req,
-			&res,
-		)
-		if err != nil {
-			return err
-		}
-
-		if waitCmdArgs.verbose {
-			pretty.Println("got poll response", res)
-		}
-
-		if res.Done {
-			return doneErr
-		}
-
-		req.InvocationId = res.InvocationId
-
-		return nil
-	}
-
-	t := time.NewTicker(waitCmdArgs.interval)
-	for range t.C {
-		err := pollOnce()
-		if err == doneErr {
-			return nil
-		} else if err != nil {
-			return err
-		}
-	}
-
-	return err
-}

config/config_global_test.go

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

config/samples/quickstart_backup_to_external_disk.yml

@@ -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)
@@ -40,7 +40,7 @@ jobs:
 
 # This job pushes to the local sink defined in job `backuppool_sink`.
 # We trigger replication manually from the command line / udev rules using
-#  `zrepl signal replication push_to_drive`
+#  `zrepl signal wakeup push_to_drive`
 - type: push
   name: push_to_drive
   connect:
@@ -85,4 +85,4 @@ jobs:
   root_fs: "backuppool/zrepl/sink"
   serve:
     type: local
-    listener_name: backuppool_sink
+    listener_name: backuppool_sink

daemon/control.go

@@ -73,29 +73,12 @@ func (j *controlJob) RegisterMetrics(registerer prometheus.Registerer) {
 }
 
 const (
-	ControlJobEndpointPProf         string = "/debug/pprof"
+	ControlJobEndpointPProf   string = "/debug/pprof"
-	ControlJobEndpointVersion       string = "/version"
+	ControlJobEndpointVersion string = "/version"
-	ControlJobEndpointStatus        string = "/status"
+	ControlJobEndpointStatus  string = "/status"
-	ControlJobEndpointTriggerActive string = "/signal/active"
+	ControlJobEndpointSignal  string = "/signal"
-	ControlJobEndpointPollActive    string = "/poll/active"
-	ControlJobEndpointResetActive   string = "/reset/active"
 )
 
-type ControlJobEndpointTriggerActiveRequest struct {
-	Job string
-	job.ActiveSideTriggerRequest
-}
-
-type ControlJobEndpointResetActiveRequest struct {
-	Job string
-	job.ActiveSideResetRequest
-}
-
-type ControlJobEndpointWaitActiveRequest struct {
-	Job string
-	job.ActiveSidePollRequest
-}
-
 func (j *controlJob) Run(ctx context.Context) {
 
 	log := job.GetLogger(ctx)
@@ -147,106 +130,29 @@ func (j *controlJob) Run(ctx context.Context) {
 			return s, nil
 		}})
 
-	mux.Handle(ControlJobEndpointPollActive, requestLogger{log: log, handler: jsonRequestResponder{log, func(decoder jsonDecoder) (v interface{}, err error) {
+	mux.Handle(ControlJobEndpointSignal,
-		var req ControlJobEndpointWaitActiveRequest
+		requestLogger{log: log, handler: jsonRequestResponder{log, func(decoder jsonDecoder) (interface{}, error) {
-		if decoder(&req) != nil {
-			return nil, errors.Errorf("decode failed")
-		}
-
-		j.jobs.m.RLock()
-
-		jo, ok := j.jobs.jobs[req.Job]
-		if !ok {
-			j.jobs.m.RUnlock()
-			return struct{}{}, fmt.Errorf("unknown job name %q", req.Job)
-		}
-
-		ajo, ok := jo.(*job.ActiveSide)
-		if !ok {
-			v, err = struct{}{}, fmt.Errorf("job %q is not an active side (it's a %T)", jo.Name(), jo)
-			j.jobs.m.RUnlock()
-			return v, err
-		}
-
-		res, err := ajo.Poll(req.ActiveSidePollRequest)
-
-		j.jobs.m.RUnlock()
-
-		return res, err
-	}}})
-
-	mux.Handle(ControlJobEndpointTriggerActive,
-		requestLogger{log: log, handler: jsonRequestResponder{log, func(decoder jsonDecoder) (v interface{}, err error) {
 			type reqT struct {
-				Job string
+				Name string
-				job.ActiveSideTriggerRequest
+				Op   string
 			}
 			var req reqT
 			if decoder(&req) != nil {
 				return nil, errors.Errorf("decode failed")
 			}
 
-			// FIXME dedup the following code with ControlJobEndpointPollActive
+			var err error
-
+			switch req.Op {
-			j.jobs.m.RLock()
+			case "wakeup":
-
+				err = j.jobs.wakeup(req.Name)
-			jo, ok := j.jobs.jobs[req.Job]
+			case "reset":
-			if !ok {
+				err = j.jobs.reset(req.Name)
-				j.jobs.m.RUnlock()
+			default:
-				return struct{}{}, fmt.Errorf("unknown job name %q", req.Job)
+				err = fmt.Errorf("operation %q is invalid", req.Op)
 			}
 
-			ajo, ok := jo.(*job.ActiveSide)
+			return struct{}{}, err
-			if !ok {
-				v, err = struct{}{}, fmt.Errorf("job %q is not an active side (it's a %T)", jo.Name(), jo)
-				j.jobs.m.RUnlock()
-				return v, err
-			}
-
-			res, err := ajo.Trigger(req.ActiveSideTriggerRequest)
-
-			j.jobs.m.RUnlock()
-
-			return res, err
-
 		}}})
-
-	mux.Handle(ControlJobEndpointResetActive,
-		requestLogger{log: log, handler: jsonRequestResponder{log, func(decoder jsonDecoder) (v interface{}, err error) {
-			type reqT struct {
-				Job string
-				job.ActiveSideResetRequest
-			}
-			var req reqT
-			if decoder(&req) != nil {
-				return nil, errors.Errorf("decode failed")
-			}
-
-			// FIXME dedup the following code with ControlJobEndpointPollActive
-
-			j.jobs.m.RLock()
-
-			jo, ok := j.jobs.jobs[req.Job]
-			if !ok {
-				j.jobs.m.RUnlock()
-				return struct{}{}, fmt.Errorf("unknown job name %q", req.Job)
-			}
-
-			ajo, ok := jo.(*job.ActiveSide)
-			if !ok {
-				v, err = struct{}{}, fmt.Errorf("job %q is not an active side (it's a %T)", jo.Name(), jo)
-				j.jobs.m.RUnlock()
-				return v, err
-			}
-
-			res, err := ajo.Reset(req.ActiveSideResetRequest)
-
-			j.jobs.m.RUnlock()
-
-			return res, err
-
-		}}})
-
 	server := http.Server{
 		Handler: mux,
 		// control socket is local, 1s timeout should be more than sufficient, even on a loaded system

daemon/daemon.go

@@ -20,6 +20,8 @@ import (
 
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/job"
+	"github.com/zrepl/zrepl/daemon/job/reset"
+	"github.com/zrepl/zrepl/daemon/job/wakeup"
 	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/version"
@@ -129,13 +131,17 @@ type jobs struct {
 	wg sync.WaitGroup
 
 	// m protects all fields below it
-	m    sync.RWMutex
+	m       sync.RWMutex
-	jobs map[string]job.Job
+	wakeups map[string]wakeup.Func // by Job.Name
+	resets  map[string]reset.Func  // by Job.Name
+	jobs    map[string]job.Job
 }
 
 func newJobs() *jobs {
 	return &jobs{
-		jobs: make(map[string]job.Job),
+		wakeups: make(map[string]wakeup.Func),
+		resets:  make(map[string]reset.Func),
+		jobs:    make(map[string]job.Job),
 	}
 }
 
@@ -184,6 +190,28 @@ func (s *jobs) status() map[string]*job.Status {
 	return ret
 }
 
+func (s *jobs) wakeup(job string) error {
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	wu, ok := s.wakeups[job]
+	if !ok {
+		return errors.Errorf("Job %s does not exist", job)
+	}
+	return wu()
+}
+
+func (s *jobs) reset(job string) error {
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	wu, ok := s.resets[job]
+	if !ok {
+		return errors.Errorf("Job %s does not exist", job)
+	}
+	return wu()
+}
+
 const (
 	jobNamePrometheus = "_prometheus"
 	jobNameControl    = "_control"
@@ -214,6 +242,10 @@ func (s *jobs) start(ctx context.Context, j job.Job, internal bool) {
 
 	s.jobs[jobName] = j
 	ctx = zfscmd.WithJobID(ctx, j.Name())
+	ctx, wakeup := wakeup.Context(ctx)
+	ctx, resetFunc := reset.Context(ctx)
+	s.wakeups[jobName] = wakeup
+	s.resets[jobName] = resetFunc
 
 	s.wg.Add(1)
 	go func() {

daemon/job/active.go

@@ -8,11 +8,14 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/common/log"
 
 	"github.com/zrepl/zrepl/daemon/logging/trace"
 	"github.com/zrepl/zrepl/util/envconst"
 
 	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon/job/reset"
+	"github.com/zrepl/zrepl/daemon/job/wakeup"
 	"github.com/zrepl/zrepl/daemon/pruner"
 	"github.com/zrepl/zrepl/daemon/snapper"
 	"github.com/zrepl/zrepl/endpoint"
@@ -40,12 +43,8 @@ type ActiveSide struct {
 	promBytesReplicated   *prometheus.CounterVec   // labels: filesystem
 	promReplicationErrors prometheus.Gauge
 
-	tasksMtx           sync.Mutex
+	tasksMtx sync.Mutex
-	tasks              activeSideTasks
+	tasks    activeSideTasks
-	nextInvocationId   uint64
-	activeInvocationId uint64 // 0 <=> inactive
-	trigger            chan struct{}
-	reset              chan uint64
 }
 
 //go:generate enumer -type=ActiveSideState
@@ -58,17 +57,18 @@ const (
 	ActiveSideDone // also errors
 )
 
-type activeSideReplicationAndTriggerRemotePruneSequence struct {
+type activeSideTasks struct {
 	state ActiveSideState
 
 	// valid for state ActiveSideReplicating, ActiveSidePruneSender, ActiveSidePruneReceiver, ActiveSideDone
 	replicationReport driver.ReportFunc
 	replicationCancel context.CancelFunc
-	replicationDone   *report.Report
 
 	// valid for state ActiveSidePruneSender, ActiveSidePruneReceiver, ActiveSideDone
-	pruneRemote *pruner.Pruner
+	prunerSender, prunerReceiver *pruner.Pruner
-	pruneRemoteCancel context.CancelFunc
+
+	// valid for state ActiveSidePruneReceiver, ActiveSideDone
+	prunerSenderCancel, prunerReceiverCancel context.CancelFunc
 }
 
 func (a *ActiveSide) updateTasks(u func(*activeSideTasks)) activeSideTasks {
@@ -89,7 +89,7 @@ type activeMode interface {
 	SenderReceiver() (logic.Sender, logic.Receiver)
 	Type() Type
 	PlannerPolicy() logic.PlannerPolicy
-	RunPeriodic(ctx context.Context, wakePeriodic <-chan struct{}, replicationCommon chan<- struct{})
+	RunPeriodic(ctx context.Context, wakeUpCommon chan<- struct{})
 	SnapperReport() *snapper.Report
 	ResetConnectBackoff()
 }
@@ -131,8 +131,8 @@ func (m *modePush) Type() Type { return TypePush }
 
 func (m *modePush) PlannerPolicy() logic.PlannerPolicy { return *m.plannerPolicy }
 
-func (m *modePush) RunPeriodic(ctx context.Context, wakePeriodic <-chan struct{}, replicationCommon chan<- struct{}) {
+func (m *modePush) RunPeriodic(ctx context.Context, wakeUpCommon chan<- struct{}) {
-	m.snapper.Run(ctx, replicationCommon)
+	m.snapper.Run(ctx, wakeUpCommon)
 }
 
 func (m *modePush) SnapperReport() *snapper.Report {
@@ -214,10 +214,10 @@ func (*modePull) Type() Type { return TypePull }
 
 func (m *modePull) PlannerPolicy() logic.PlannerPolicy { return *m.plannerPolicy }
 
-func (m *modePull) RunPeriodic(ctx context.Context, wakePeriodic <-chan struct{}, replicationCommon chan<- struct{}) {
+func (m *modePull) RunPeriodic(ctx context.Context, wakeUpCommon chan<- struct{}) {
 	if m.interval.Manual {
 		GetLogger(ctx).Info("manual pull configured, periodic pull disabled")
-		// "waiting for wakeup replications" is printed in common ActiveSide.do
+		// "waiting for wakeups" is printed in common ActiveSide.do
 		return
 	}
 	t := time.NewTicker(m.interval.Interval)
@@ -226,12 +226,12 @@ func (m *modePull) RunPeriodic(ctx context.Context, wakePeriodic <-chan struct{}
 		select {
 		case <-t.C:
 			select {
-			case replicationCommon <- struct{}{}:
+			case wakeUpCommon <- struct{}{}:
 			default:
 				GetLogger(ctx).
 					WithField("pull_interval", m.interval).
 					Warn("pull job took longer than pull interval")
-				replicationCommon <- struct{}{} // block anyways, to queue up the wakeup replication
+				wakeUpCommon <- struct{}{} // block anyways, to queue up the wakeup
 			}
 		case <-ctx.Done():
 			return
@@ -425,205 +425,50 @@ func (j *ActiveSide) Run(ctx context.Context) {
 
 	defer log.Info("job exiting")
 
-	type Activity interface {
-		Trigger() (interface{}, error)
-	}
-
-	var periodicActivity Activity
-	var replicationActivity Activity
-	
 	periodicDone := make(chan struct{})
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	periodicCtx, endTask := trace.WithTask(ctx, "periodic")
 	defer endTask()
+	go j.mode.RunPeriodic(periodicCtx, periodicDone)
 
-	wakePeriodic := make(chan struct{})
+	invocationCount := 0
-	go j.mode.RunPeriodic(periodicCtx, wakePeriodic, periodicDone)
+outer:
-
-	j.trigger = make(chan struct{})
-	j.reset = make(chan uint64)
-	j.nextInvocationId = 1
-
-	type WaitTriggerResult interface {
-		
-	}
-	var t interface{
-		WaitForTrigger(context.Context) (context.Context, <-chan WaitTriggerResult)
-	}
-
 	for {
-		log.Info("wait for triggers")
+		log.Info("wait for wakeups")
+		select {
+		case <-ctx.Done():
+			log.WithError(ctx.Err()).Info("context")
+			break outer
 
-			// j.tasksMtx.Lock()
+		case <-wakeup.Wait(ctx):
-		// j.activeInvocationId = j.nextInvocationId
+			j.mode.ResetConnectBackoff()
-		// j.nextInvocationId++
+		case <-periodicDone:
-		// thisInvocation := j.activeInvocationId // stack-local, for use in reset-handler goroutine below
-		// j.tasksMtx.Unlock()
-		
-		// // setup the goroutine that waits for task resets
-		// // Task resets are converted into cancellations of the invocation context.
-		
-		invocationCtx, cancelInvocation := context.WithCancel(invocationCtx)
-		// waitForResetCtx, stopWaitForReset := context.WithCancel(ctx)
-		// var wg sync.WaitGroup
-		// wg.Add(1)
-		// go func() {
-		// 	defer wg.Done()
-		// 	select {
-		// 	case <-waitForResetCtx.Done():
-		// 		return
-		// 	case reqResetInvocation := <-j.reset:
-		// 		l := log.WithField("requested_invocation_id", reqResetInvocation).
-		// 			WithField("this_invocation_id", thisInvocation)
-		// 		if reqResetInvocation == thisInvocation {
-		// 			l.Info("reset received, cancelling current invocation")
-		// 			cancelInvocation()
-		// 		} else {
-		// 			l.Debug("received reset for invocation id that is not us, discarding request")
-		// 		}
-		// 	}
-		// }()
-
-		// j.tasksMtx.Lock()
-		// j.activeInvocationId = 0
-		// j.tasksMtx.Unlock()
-
-		invocationCtx, err := t.WaitForTrigger(ctx)
-		if err != nil {
-			log.WithError(ctx.Err()).Info("error waiting for trigger")
-			break
 		}
-
+		invocationCount++
-		j.mode.ResetConnectBackoff()
+		invocationCtx, endSpan := trace.WithSpan(ctx, fmt.Sprintf("invocation-%d", invocationCount))
-
-		// setup the invocation context
-		invocationCtx, endSpan := trace.WithSpan(invocationCtx, fmt.Sprintf("invocation-%d", j.nextInvocationId))
-		
-
 		j.do(invocationCtx)
-		stopWaitForReset()
-		wg.Wait()
-
-	
-
 		endSpan()
 	}
 }
 
-type ActiveSidePollRequest struct {
-	InvocationId uint64
-}
-
-type ActiveSidePollResponse struct {
-	Done         bool
-	InvocationId uint64
-}
-
-func (j *ActiveSide) Poll(req ActiveSidePollRequest) (*ActiveSidePollResponse, error) {
-	j.tasksMtx.Lock()
-	defer j.tasksMtx.Unlock()
-
-	waitForId := req.InvocationId
-	if req.InvocationId == 0 {
-		// handle the case where the client doesn't know what the current invocation id is
-		if j.activeInvocationId != 0 {
-			waitForId = j.activeInvocationId
-		} else {
-			waitForId = j.nextInvocationId
-		}
-	}
-
-	var done bool
-	if j.activeInvocationId == 0 {
-		done = waitForId < j.nextInvocationId
-	} else {
-		done = waitForId < j.activeInvocationId
-	}
-	res := &ActiveSidePollResponse{Done: done, InvocationId: waitForId}
-	return res, nil
-}
-
-type ActiveSideResetRequest struct {
-	InvocationId uint64
-}
-
-type ActiveSideResetResponse struct {
-	InvocationId uint64
-}
-
-func (j *ActiveSide) Reset(req ActiveSideResetRequest) (*ActiveSideResetResponse, error) {
-	j.tasksMtx.Lock()
-	defer j.tasksMtx.Unlock()
-
-	resetId := req.InvocationId
-	if req.InvocationId == 0 {
-		// handle the case where the client doesn't know what the current invocation id is
-		resetId = j.activeInvocationId
-	}
-
-	if resetId == 0 {
-		return nil, fmt.Errorf("no active invocation")
-	}
-
-	if resetId != j.activeInvocationId {
-		return nil, fmt.Errorf("active invocation (%d) is not the invocation requested for reset (%d); (active invocation '0' indicates no active invocation)", j.activeInvocationId, resetId)
-	}
-
-	// non-blocking send (.Run() must not hold mutex while waiting for resets)
-	select {
-	case j.reset <- resetId:
-	default:
-	}
-
-	return &ActiveSideResetResponse{InvocationId: resetId}, nil
-}
-
-type ActiveSideTriggerRequest struct {
-}
-
-type ActiveSideSignalResponse struct {
-	InvocationId uint64
-	
-}
-
-func (j *ActiveSide) Trigger(req ActiveSideTriggerRequest) (*ActiveSideSignalResponse, error) {
-	// switch req.What {
-	// case "replication":
-	// 	invocationId, err = j.jobs.doreplication(req.Name)
-	// case "reset":
-	// 	err = j.jobs.reset(req.Name)
-	// case "snapshot":
-	// 	err = j.jobs.dosnapshot(req.Name)
-	// default:
-	// 	err = fmt.Errorf("operation %q is invalid", req.Op)
-	// }
-
-	j.tasksMtx.Lock()
-	var invocationId uint64
-	if j.activeInvocationId != 0 {
-		invocationId = j.activeInvocationId
-	} else {
-		invocationId = j.nextInvocationId
-	}
-	// non-blocking send (.Run() must not hold mutex while waiting for signals)
-	select {
-	case j.trigger <- struct{}{}:
-	default:
-	}
-	j.tasksMtx.Unlock()
-	return &ActiveSideSignalResponse{InvocationId: invocationId}, nil
-}
-
-type ReplicationPlusRemotePruneSequence struct {
-
-}
-
 func (j *ActiveSide) do(ctx context.Context) {
 
 	j.mode.ConnectEndpoints(ctx, j.connecter)
 	defer j.mode.DisconnectEndpoints()
 
+	// allow cancellation of an invocation (this function)
+	ctx, cancelThisRun := context.WithCancel(ctx)
+	defer cancelThisRun()
+	go func() {
+		select {
+		case <-reset.Wait(ctx):
+			log.Info("reset received, cancelling current invocation")
+			cancelThisRun()
+		case <-ctx.Done():
+		}
+	}()
+
 	sender, receiver := j.mode.SenderReceiver()
 
 	{
@@ -647,11 +492,10 @@ func (j *ActiveSide) do(ctx context.Context) {
 		GetLogger(ctx).Info("start replication")
 		repWait(true) // wait blocking
 		repCancel()   // always cancel to free up context resources
+
 		replicationReport := j.tasks.replicationReport()
 		j.promReplicationErrors.Set(float64(replicationReport.GetFailedFilesystemsCountInLatestAttempt()))
-		j.updateTasks(func(tasks *activeSideTasks) {
+
-			tasks.replicationDone = replicationReport
-		})
 		endSpan()
 	}
 

daemon/job/job.go

@@ -19,7 +19,6 @@ func GetLogger(ctx context.Context) Logger {
 	return logging.GetLogger(ctx, logging.SubsysJob)
 }
 
-
 type Job interface {
 	Name() string
 	Run(ctx context.Context)

daemon/job/reset/reset.go

@@ -0,0 +1,35 @@
+package reset
+
+import (
+	"context"
+	"errors"
+)
+
+type contextKey int
+
+const contextKeyReset contextKey = iota
+
+func Wait(ctx context.Context) <-chan struct{} {
+	wc, ok := ctx.Value(contextKeyReset).(chan struct{})
+	if !ok {
+		wc = make(chan struct{})
+	}
+	return wc
+}
+
+type Func func() error
+
+var AlreadyReset = errors.New("already reset")
+
+func Context(ctx context.Context) (context.Context, Func) {
+	wc := make(chan struct{})
+	wuf := func() error {
+		select {
+		case wc <- struct{}{}:
+			return nil
+		default:
+			return AlreadyReset
+		}
+	}
+	return context.WithValue(ctx, contextKeyReset, wc), wuf
+}

daemon/job/snapjob.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/filters"
+	"github.com/zrepl/zrepl/daemon/job/wakeup"
 	"github.com/zrepl/zrepl/daemon/pruner"
 	"github.com/zrepl/zrepl/daemon/snapper"
 	"github.com/zrepl/zrepl/endpoint"
@@ -111,13 +112,13 @@ func (j *SnapJob) Run(ctx context.Context) {
 	invocationCount := 0
 outer:
 	for {
-		log.Info("wait for replications")
+		log.Info("wait for wakeups")
 		select {
 		case <-ctx.Done():
 			log.WithError(ctx.Err()).Info("context")
 			break outer
 
-		// case <-doreplication.Wait(ctx):
+		case <-wakeup.Wait(ctx):
 		case <-periodicDone:
 		}
 		invocationCount++

daemon/job/trigger/trigger.go

@@ -1,184 +0,0 @@
-//
-//
-// Alternative Design (in "RustGo")
-//
-//	enum InternalMsg {
-//	    Trigger((), chan (TriggerResponse, error)),
-//	    Poll(PollRequest, chan PollResponse),
-//	    Reset(ResetRequest, chan (ResetResponse, error)),
-//	}
-//
-//	enum State {
-//	    Running{
-//	        invocationId: u32,
-//	        cancelCurrentInvocation: context.CancelFunc
-//	    }
-//	    Waiting{
-//	        nextInvocationId: u32,
-//	    }
-//	}
-//
-//	for msg := <- t.internalMsgs {
-//	    match (msg, state) {
-//	        ...
-//	    }
-//	}
-package trigger
-
-import (
-	"context"
-	"fmt"
-	"math"
-	"sync"
-)
-
-type T struct {
-	mtx sync.Mutex
-	cv  sync.Cond
-
-	nextInvocationId        uint64
-	activeInvocationId      uint64 // 0 <=> inactive
-	triggerPending          bool
-	contextDone             bool
-	reset                   chan uint64
-	stopWaitForReset        chan struct{}
-	cancelCurrentInvocation context.CancelFunc
-}
-
-func New() *T {
-	t := &T{
-		activeInvocationId: math.MaxUint64,
-		nextInvocationId:   1,
-	}
-	t.cv.L = &t.mtx
-	return t
-}
-
-func (t *T) WaitForTrigger(ctx context.Context) (rctx context.Context, err error) {
-	t.mtx.Lock()
-	defer t.mtx.Unlock()
-
-	if t.activeInvocationId == 0 {
-		return nil, fmt.Errorf("must be running when calling this function")
-	}
-	t.activeInvocationId = 0
-	t.cancelCurrentInvocation = nil
-
-	if t.contextDone == true {
-		panic("implementation error: this variable is only true while in WaitForTrigger, and that's a mutually exclusive function")
-	}
-	stopWaitingForDone := make(chan struct{})
-	go func() {
-		select {
-		case <-stopWaitingForDone:
-		case <-ctx.Done():
-			t.mtx.Lock()
-			t.contextDone = true
-			t.cv.Broadcast()
-			t.mtx.Unlock()
-		}
-	}()
-
-	defer func() {
-		t.triggerPending = false
-		t.contextDone = false
-	}()
-	for !t.triggerPending && !t.contextDone {
-		t.cv.Wait()
-	}
-	close(stopWaitingForDone)
-	if t.contextDone {
-		if ctx.Err() == nil {
-			panic("implementation error: contextDone <=> ctx.Err() != nil")
-		}
-		return nil, ctx.Err()
-	}
-
-	t.activeInvocationId = t.nextInvocationId
-	t.nextInvocationId++
-	rctx, t.cancelCurrentInvocation = context.WithCancel(ctx)
-
-	return rctx, nil
-}
-
-type TriggerResponse struct {
-	InvocationId uint64
-}
-
-func (t *T) Trigger() (TriggerResponse, error) {
-	t.mtx.Lock()
-	defer t.mtx.Unlock()
-	var invocationId uint64
-	if t.activeInvocationId != 0 {
-		invocationId = t.activeInvocationId
-	} else {
-		invocationId = t.nextInvocationId
-	}
-	// non-blocking send (.Run() must not hold mutex while waiting for signals)
-	t.triggerPending = true
-	t.cv.Broadcast()
-	return TriggerResponse{InvocationId: invocationId}, nil
-}
-
-type PollRequest struct {
-	InvocationId uint64
-}
-
-type PollResponse struct {
-	Done         bool
-	InvocationId uint64
-}
-
-func (t *T) Poll(req PollRequest) (res PollResponse) {
-	t.mtx.Lock()
-	defer t.mtx.Unlock()
-
-	waitForId := req.InvocationId
-	if req.InvocationId == 0 {
-		// handle the case where the client doesn't know what the current invocation id is
-		if t.activeInvocationId != 0 {
-			waitForId = t.activeInvocationId
-		} else {
-			waitForId = t.nextInvocationId
-		}
-	}
-
-	var done bool
-	if t.activeInvocationId == 0 {
-		done = waitForId < t.nextInvocationId
-	} else {
-		done = waitForId < t.activeInvocationId
-	}
-	return PollResponse{Done: done, InvocationId: waitForId}
-}
-
-type ResetRequest struct {
-	InvocationId uint64
-}
-
-type ResetResponse struct {
-	InvocationId uint64
-}
-
-func (t *T) Reset(req ResetRequest) (*ResetResponse, error) {
-	t.mtx.Lock()
-	defer t.mtx.Unlock()
-
-	resetId := req.InvocationId
-	if req.InvocationId == 0 {
-		// handle the case where the client doesn't know what the current invocation id is
-		resetId = t.activeInvocationId
-	}
-
-	if resetId == 0 {
-		return nil, fmt.Errorf("no active invocation")
-	}
-
-	if resetId != t.activeInvocationId {
-		return nil, fmt.Errorf("active invocation (%d) is not the invocation requested for reset (%d); (active invocation '0' indicates no active invocation)", t.activeInvocationId, resetId)
-	}
-
-	t.cancelCurrentInvocation()
-
-	return &ResetResponse{InvocationId: resetId}, nil
-}

daemon/job/trigger/trigger_test.go

@@ -1,205 +0,0 @@
-package trigger_test
-
-import (
-	"context"
-	"net/http"
-	"sync"
-	"testing"
-
-	"github.com/stretchr/testify/require"
-	"github.com/zrepl/zrepl/daemon/job/trigger"
-	"github.com/zrepl/zrepl/daemon/logging/trace"
-	"github.com/zrepl/zrepl/replication"
-	"github.com/zrepl/zrepl/replication/driver"
-	"github.com/zrepl/zrepl/replication/logic"
-)
-
-func TestBasics(t *testing.T) {
-
-	var wg sync.WaitGroup
-	defer wg.Wait()
-
-	tr := trigger.New()
-
-	triggered := make(chan int)
-	waitForTriggerError := make(chan error)
-	waitForResetCallToBeMadeByMainGoroutine := make(chan struct{})
-	postResetAssertionsDone := make(chan struct{})
-
-	taskCtx := context.Background()
-	taskCtx, cancelTaskCtx := context.WithCancel(taskCtx)
-
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-
-		taskCtx := context.WithValue(taskCtx, "mykey", "myvalue")
-
-		triggers := 0
-
-	outer:
-		for {
-			invocationCtx, err := tr.WaitForTrigger(taskCtx)
-			if err != nil {
-				waitForTriggerError <- err
-				return
-			}
-			require.Equal(t, invocationCtx.Value("mykey"), "myvalue")
-
-			triggers++
-			triggered <- triggers
-
-			switch triggers {
-			case 1:
-				continue outer
-			case 2:
-				<-waitForResetCallToBeMadeByMainGoroutine
-				require.Equal(t, context.Canceled, invocationCtx.Err(), "Reset() cancels invocation context")
-				require.Nil(t, taskCtx.Err(), "Reset() does not cancel task context")
-				close(postResetAssertionsDone)
-			}
-
-		}
-
-	}()
-
-	t.Logf("trigger 1")
-	_, err := tr.Trigger()
-	require.NoError(t, err)
-	v := <-triggered
-	require.Equal(t, 1, v)
-
-	t.Logf("trigger 2")
-	triggerResponse, err := tr.Trigger()
-	require.NoError(t, err)
-	v = <-triggered
-	require.Equal(t, 2, v)
-
-	t.Logf("reset")
-	resetResponse, err := tr.Reset(trigger.ResetRequest{InvocationId: triggerResponse.InvocationId})
-	require.NoError(t, err)
-	t.Logf("reset response: %#v", resetResponse)
-	close(waitForResetCallToBeMadeByMainGoroutine)
-	<-postResetAssertionsDone
-
-	t.Logf("cancel the context")
-	cancelTaskCtx()
-	wfte := <-waitForTriggerError
-	require.Equal(t, taskCtx.Err(), wfte)
-
-}
-
-type PushJob struct {
-	snap *Snapshotter
-	repl *ReplicationAndTriggerRemotePruningSequence
-}
-
-func (j *PushJob) Handle(w http.ResponseWriter, r *http.Request) {
-	panic("unimplemented")
-}
-
-func (j *PushJob) HandleTrigger(w http.ResponseWriter, r *http.Request) {
-	panic("unimplemented")
-}
-
-func (j *PushJob) Run(ctx context.Context) {
-
-}
-
-type SnapshotSequence struct {
-
-}
-
-type ReplicationAndTriggerRemotePruningSequence struct {
-
-}
-
-func (s ReplicationAndTriggerRemotePruningSequence) Run(ctx context.Context) {
-
-	j.mode.ConnectEndpoints(ctx, j.connecter)
-	defer j.mode.DisconnectEndpoints()
-
-	sender, receiver := j.mode.SenderReceiver()
-
-	{
-		select {
-		case <-ctx.Done():
-			return
-		default:
-		}
-		ctx, endSpan := trace.WithSpan(ctx, "replication")
-		ctx, repCancel := context.WithCancel(ctx)
-		var repWait driver.WaitFunc
-		j.updateTasks(func(tasks *activeSideTasks) {
-			// reset it
-			*tasks = activeSideTasks{}
-			tasks.replicationCancel = func() { repCancel(); endSpan() }
-			tasks.replicationReport, repWait = replication.Do(
-				ctx, j.replicationDriverConfig, logic.NewPlanner(j.promRepStateSecs, j.promBytesReplicated, sender, receiver, j.mode.PlannerPolicy()),
-			)
-			tasks.state = ActiveSideReplicating
-		})
-		GetLogger(ctx).Info("start replication")
-		repWait(true) // wait blocking
-		repCancel()   // always cancel to free up context resources
-		replicationReport := j.tasks.replicationReport()
-		j.promReplicationErrors.Set(float64(replicationReport.GetFailedFilesystemsCountInLatestAttempt()))
-		j.updateTasks(func(tasks *activeSideTasks) {
-			tasks.replicationDone = replicationReport
-		})
-		endSpan()
-	}
-
-	{
-		select {
-		case <-ctx.Done():
-			return
-		default:
-		}
-		ctx, endSpan := trace.WithSpan(ctx, "prune_sender")
-		ctx, senderCancel := context.WithCancel(ctx)
-		tasks := j.updateTasks(func(tasks *activeSideTasks) {
-			tasks.prunerSender = j.prunerFactory.BuildSenderPruner(ctx, sender, sender)
-			tasks.prunerSenderCancel = func() { senderCancel(); endSpan() }
-			tasks.state = ActiveSidePruneSender
-		})
-		GetLogger(ctx).Info("start pruning sender")
-		tasks.prunerSender.Prune()
-		GetLogger(ctx).Info("finished pruning sender")
-		senderCancel()
-		endSpan()
-	}
-	{
-		select {
-		case <-ctx.Done():
-			return
-		default:
-		}
-		ctx, endSpan := trace.WithSpan(ctx, "prune_recever")
-		ctx, receiverCancel := context.WithCancel(ctx)
-		tasks := j.updateTasks(func(tasks *activeSideTasks) {
-			tasks.prunerReceiver = j.prunerFactory.BuildReceiverPruner(ctx, receiver, sender)
-			tasks.prunerReceiverCancel = func() { receiverCancel(); endSpan() }
-			tasks.state = ActiveSidePruneReceiver
-		})
-		GetLogger(ctx).Info("start pruning receiver")
-		tasks.prunerReceiver.Prune()
-		GetLogger(ctx).Info("finished pruning receiver")
-		receiverCancel()
-		endSpan()
-	}
-
-	j.updateTasks(func(tasks *activeSideTasks) {
-		tasks.state = ActiveSideDone
-	})
-
-}
-
-
-func TestUseCase(t *testing.T) {
-
-	var as ActiveSide
-
-	as.
-	
-}

daemon/job/wakeup/wakeup.go

@@ -0,0 +1,35 @@
+package wakeup
+
+import (
+	"context"
+	"errors"
+)
+
+type contextKey int
+
+const contextKeyWakeup contextKey = iota
+
+func Wait(ctx context.Context) <-chan struct{} {
+	wc, ok := ctx.Value(contextKeyWakeup).(chan struct{})
+	if !ok {
+		wc = make(chan struct{})
+	}
+	return wc
+}
+
+type Func func() error
+
+var AlreadyWokenUp = errors.New("already woken up")
+
+func Context(ctx context.Context) (context.Context, Func) {
+	wc := make(chan struct{})
+	wuf := func() error {
+		select {
+		case wc <- struct{}{}:
+			return nil
+		default:
+			return AlreadyWokenUp
+		}
+	}
+	return context.WithValue(ctx, contextKeyWakeup, wc), wuf
+}

daemon/snapper/snapper.go

@@ -210,10 +210,6 @@ func syncUp(a args, u updater) state {
 		return u(func(s *Snapper) {
 			s.state = Planning
 		}).sf()
-	// case <-dosnapshot.Wait(a.ctx):
-	// 	return u(func(s *Snapper) {
-	// 		s.state = Planning
-	// 	}).sf()
 	case <-a.ctx.Done():
 		return onMainCtxDone(a.ctx, u)
 	}
@@ -382,10 +378,6 @@ func wait(a args, u updater) state {
 		return u(func(snapper *Snapper) {
 			snapper.state = Planning
 		}).sf()
-	// case <-dosnapshot.Wait(a.ctx):
-	// 	return u(func(snapper *Snapper) {
-	// 		snapper.state = Planning
-	// 	}).sf()
 	case <-a.ctx.Done():
 		return onMainCtxDone(a.ctx, u)
 	}

daemon/snapper/snapper_all.go

@@ -18,9 +18,9 @@ type PeriodicOrManual struct {
 	s *Snapper
 }
 
-func (s *PeriodicOrManual) Run(ctx context.Context, replicationCommon chan<- struct{}) {
+func (s *PeriodicOrManual) Run(ctx context.Context, wakeUpCommon chan<- struct{}) {
 	if s.s != nil {
-		s.s.Run(ctx, replicationCommon)
+		s.s.Run(ctx, wakeUpCommon)
 	}
 }
 

docs/changelog.rst

@@ -32,7 +32,6 @@ We use the following annotations for classifying changes:
 0.4.0
 -----
 
-* |break| Change syntax to trigger a job replication, rename ``zrepl signal wakeup JOB`` to ``zrepl signal replication JOB``
 * |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>` ).

docs/configuration/jobs.rst

@@ -78,7 +78,7 @@ Job Type ``pull``
         ``$root_fs/$source_path``
     * - ``interval``
       - | Interval at which to pull from the source job (e.g. ``10m``).
-        | ``manual`` disables periodic pulling, replication then only happens on :ref:`replication <cli-signal-replication>`.
+        | ``manual`` disables periodic pulling, replication then only happens on :ref:`wakeup <cli-signal-wakeup>`.
     * - ``pruning``
       - |pruning-spec|
 

docs/configuration/monitoring.rst

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

docs/configuration/prune.rst

@@ -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,
+   Assume the following grid spec:
-          we subject all snapshots to the grid pruning policy by settings `regex: .*`.
 
-          `
+      grid: 1x1h(keep=all) | 2x2h | 1x3h
-           grid: 1x1h(keep=all) | 2x2h | 1x3h
-           regex: .*
-          `
 
-          0h        1h        2h        3h        4h        5h        6h        7h        8h        9h
+   This grid spec produces the following constellation of buckets:
-          |         |         |         |         |         |         |         |         |         |
+
-          |-Bucket1-|-----Bucket 2------|------Bucket 3-----|-----------Bucket 4----------|
+   0h        1h        2h        3h        4h        5h        6h        7h        8h        9h
-          | keep=all|      keep=1       |       keep=1      |            keep=1           |
+   |         |         |         |         |         |         |         |         |         |
+   |-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
+   Bucket1:   a, b, c
-          Bucket 2: d,e,f,g,h,i,j
+   Bucket2:   d,e,f,g,h,i,j
-          Bucket 3: k,l,m,n,o,p
+   Bucket3:   k,l,m,n,o,p
-          Bucket 4: q,r, q,r,s,t,u,v,w,x,y,z
+   Bucket4:   q,r,s,t,u,v,w,x,y,z
-          None:     A,B,C,D
+   No bucket: A,B,C,D
 
-          It then applies the per-bucket pruning logic described above which resulting in the
+   For each bucket, we now prune snapshots until it only contains `keep` snapshots.
-          following list of remaining snapshots.
+   Newer snapshots are destroyed first.
+   Snapshots that do not fall into a bucket are always destroyed.
 
-          |  a  b  c                    j                 p                             z                    |
+   Result after pruning:
-
-          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.
 
+   0h        1h        2h        3h        4h        5h        6h        7h        8h        9h
+   |         |         |         |         |         |         |         |         |         |
+   |-Bucket1-|-----Bucket2-------|------Bucket3------|-----------Bucket4-----------|
+   |         |                   |                   |                             |
+   |  a  b  c|                   j                 p |                           z  |
 
 .. _prune-keep-last-n:
 

docs/configuration/snapshotting.rst

@@ -15,7 +15,7 @@ A filesystem that does not have snapshots by the snapshotter has lower priority
 
 For ``push`` jobs, replication is automatically triggered after all filesystems have been snapshotted.
 
-Note that the ``zrepl signal replication JOB`` subcommand does not trigger snapshotting.
+Note that the ``zrepl signal wakeup JOB`` subcommand does not trigger snapshotting.
 
 
 ::
@@ -38,7 +38,7 @@ There is also a ``manual`` snapshotting type, which covers the following use cas
 * Existing infrastructure for automatic snapshots: you only want to use this zrepl job for replication.
 * Handling snapshotting through a separate ``snap`` job.
 
-Note that you will have to trigger replication manually using the ``zrepl signal replication JOB`` subcommand in that case.
+Note that you will have to trigger replication manually using the ``zrepl signal wakeup JOB`` subcommand in that case.
 
 ::
 

docs/installation/apt-repos.rst

@@ -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 -) && \
+    (curl https://zrepl.cschwarz.com/apt/apt-key.asc | sudo apt-key add -) && \
-    (echo "deb [arch=$ARCH] https://zrepl.cschwarz.com/apt/$CODENAME main" > /etc/apt/sources.list.d/zrepl.list) && \
+    (echo "deb [arch=$ARCH] https://zrepl.cschwarz.com/apt/$CODENAME main" | sudo tee /etc/apt/sources.list.d/zrepl.list) && \
-    apt update
+    sudo apt update
 
 
 .. NOTE::

docs/publish.sh

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

docs/quickstart.rst

@@ -55,8 +55,8 @@ Watch it Work
 =============
 
 Run ``zrepl status`` on the active side of the replication setup to monitor snaphotting, replication and pruning activity.
-To re-trigger replication (snapshots are separate!), use ``zrepl signal replication JOBNAME``.
+To re-trigger replication (snapshots are separate!), use ``zrepl signal wakeup JOBNAME``.
-(refer to the example use case document if you are uncertain which job you want to start replication).
+(refer to the example use case document if you are uncertain which job you want to wake up).
 
 You can also use basic UNIX tools to inspect see what's going on.
 If you like tmux, here is a handy script that works on FreeBSD: ::

docs/usage.rst

@@ -13,7 +13,7 @@ CLI Overview
     The zrepl binary is self-documenting:
     run ``zrepl help`` for an overview of the available subcommands or ``zrepl SUBCOMMAND --help`` for information on available flags, etc.
 
-.. _cli-signal-replication:
+.. _cli-signal-wakeup:
 
 .. list-table::
     :widths: 30 70
@@ -29,7 +29,7 @@ CLI Overview
       - show job activity, or with ``--raw`` for JSON output
     * - ``zrepl stdinserver``
       - see :ref:`transport-ssh+stdinserver`
-    * - ``zrepl signal replication JOB``
+    * - ``zrepl signal wakeup JOB``
       - manually trigger replication + pruning of JOB
     * - ``zrepl signal reset JOB``
       - manually abort current replication + pruning of JOB

main.go

@@ -11,9 +11,7 @@ import (
 func init() {
 	cli.AddSubcommand(daemon.DaemonCmd)
 	cli.AddSubcommand(status.Subcommand)
-	cli.AddSubcommand(client.TriggerCmd)
+	cli.AddSubcommand(client.SignalCmd)
-	cli.AddSubcommand(client.ResetCmd)
-	cli.AddSubcommand(client.WaitCmd)
 	cli.AddSubcommand(client.StdinserverCmd)
 	cli.AddSubcommand(client.ConfigcheckCmd)
 	cli.AddSubcommand(client.VersionCmd)

platformtest/tests/generated_cases.go

@@ -34,5 +34,10 @@ var Cases = []Case{BatchDestroy,
 	SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden__EncryptionSupported_true,
 	SendArgsValidationResumeTokenDifferentFilesystemForbidden,
 	SendArgsValidationResumeTokenEncryptionMismatchForbidden,
+	SendStreamCloseAfterBlockedOnPipeWrite,
+	SendStreamCloseAfterEOFRead,
+	SendStreamMultipleCloseAfterEOF,
+	SendStreamMultipleCloseBeforeEOF,
+	SendStreamNonEOFReadErrorHandling,
 	UndestroyableSnapshotParsing,
 }

platformtest/tests/recvRollback.go

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

platformtest/tests/replication.go

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

platformtest/tests/sendStream.go

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

pruning/keep_last_n.go

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

pruning/keep_last_n_test.go

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

zfs/zfs.go

@@ -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 {
+	if capacity < 0 {
-		panic(fmt.Sprintf("capacity must be positive %v", capacity))
+		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 {
+type sendStreamState int
-	cmd  *zfscmd.Cmd
-	kill context.CancelFunc
 
-	closeMtx     sync.Mutex
+const (
-	stdoutReader *os.File
+	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) {
+func (s *SendStream) Read(p []byte) (n int, _ error) {
-	s.closeMtx.Lock()
+	s.mtx.Lock()
-	opErr := s.opErr
+	defer s.mtx.Unlock()
-	s.closeMtx.Unlock()
-	if opErr != nil {
-		return 0, opErr
-	}
 
-	n, err = s.stdoutReader.Read(p)
+	switch s.state {
-	if err != nil {
+	case sendStreamClosed:
-		debug("sendStream: read err: %T %s", err, err)
+		return 0, os.ErrClosed
-		// TODO we assume here that any read error is permanent
+
-		// which is most likely the case for a local zfs send
+	case sendStreamOpen:
-		kwerr := s.killAndWait(err)
+		n, readErr := s.stdoutReader.Read(p)
-		debug("sendStream: killAndWait n=%v err= %T %s", n, kwerr, kwerr)
+		if readErr != nil {
-		// TODO we assume here that any read error is permanent
+			debug("sendStream: read: readErr=%T %s", readErr, readErr)
-		return n, kwerr
+			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
+	// send SIGKILL
-	s.closeMtx.Lock()
+	s.kill()
-	defer s.closeMtx.Unlock()
 
-	if s.opErr != nil {
+	// Close our read-end of the pipe.
-		return s.opErr
+	//
+	// 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
+	// invariant: at this point, the child is gone and we cleaned up everything related to the SendStream
-	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
-		}
-	}
 
-	// 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
+	return nil
-	// 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
 }
 
+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
 	}

zfs/zfs_pipe.go

@@ -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")
 		})

zfs/zfs_pipe_linux.go

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

zfs/zfscmd/zfscmd.go

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