fixups

cc @asomers

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

@@ -11,8 +11,8 @@ import (
 )
 
 var SignalCmd = &cli.Subcommand{
-	Use:   "signal [replication|reset|snapshot] JOB",
-	Short: "run a job replication, abort its current invocation, run a snapshot job",
+	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)
 	},
@@ -20,7 +20,7 @@ var SignalCmd = &cli.Subcommand{
 
 func runSignalCmd(config *config.Config, args []string) error {
 	if len(args) != 2 {
-		return errors.Errorf("Expected 2 arguments: [replication|reset|snapshot] JOB")
+		return errors.Errorf("Expected 2 arguments: [wakeup|reset] JOB")
 	}
 
 	httpc, err := controlHttpClient(config.Global.Control.SockPath)

client/status/client/client.go

@@ -55,12 +55,8 @@ func (c *Client) signal(job, sig string) error {
 	)
 }
 
-func (c *Client) SignalReplication(job string) error {
-	return c.signal(job, "replication")
-}
-
-func (c *Client) SignalSnapshot(job string) error {
-	return c.signal(job, "snapshot")
+func (c *Client) SignalWakeup(job string) error {
+	return c.signal(job, "wakeup")
 }
 
 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
-	SignalSnapshot(job string) error
+	SignalWakeup(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 {
-		usemode, err := statusv2Flags.Mode.InputForChoice(StatusV2ModeDump)
+	if !isatty.IsTerminal(os.Stdout.Fd()) && mode != StatusV2ModeDump && mode != StatusV2ModeRaw {
+		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"}
-			clientFuncs := []func(job string) error{c.SignalReplication, c.SignalSnapshot, c.SignalReset}
+			signals := []string{"wakeup", "reset"}
+			clientFuncs := []func(job string) error{c.SignalWakeup, c.SignalReset}
 			sigMod := tview.NewModal()
 			sigMod.SetBackgroundColor(tcell.ColorDefault)
 			sigMod.SetBorder(true)

client/status/viewmodel/bytesprogresshistory.go

@@ -11,7 +11,6 @@ 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) {
@@ -38,11 +37,8 @@ func (p *bytesProgressHistory) Update(currentVal int64) (bytesPerSecondAvg int64
 	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
+	return int64(rate), p.changeCount
 }

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

config/config.go

@@ -13,6 +13,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/zrepl/yaml-config"
 
+	"github.com/zrepl/zrepl/util/datasizeunit"
 	zfsprop "github.com/zrepl/zrepl/zfs/property"
 )
 
@@ -87,6 +88,8 @@ type SendOptions struct {
 	Compressed       bool `yaml:"compressed,optional,default=false"`
 	EmbeddedData     bool `yaml:"embbeded_data,optional,default=false"`
 	Saved            bool `yaml:"saved,optional,default=false"`
+
+	BandwidthLimit *BandwidthLimit `yaml:"bandwidth_limit,optional,fromdefaults"`
 }
 
 type RecvOptions struct {
@@ -96,6 +99,15 @@ type RecvOptions struct {
 	// Reencrypt bool `yaml:"reencrypt"`
 
 	Properties *PropertyRecvOptions `yaml:"properties,fromdefaults"`
+
+	BandwidthLimit *BandwidthLimit `yaml:"bandwidth_limit,optional,fromdefaults"`
+}
+
+var _ yaml.Unmarshaler = &datasizeunit.Bits{}
+
+type BandwidthLimit struct {
+	Max            datasizeunit.Bits `yaml:"max,default=-1 B"`
+	BucketCapacity datasizeunit.Bits `yaml:"bucket_capacity,default=128 KiB"`
 }
 
 type Replication struct {
@@ -113,10 +125,6 @@ type ReplicationOptionsConcurrency struct {
 	SizeEstimates int `yaml:"size_estimates,optional,default=4"`
 }
 
-func (l *RecvOptions) SetDefault() {
-	*l = RecvOptions{Properties: &PropertyRecvOptions{}}
-}
-
 type PropertyRecvOptions struct {
 	Inherit  []zfsprop.Property          `yaml:"inherit,optional"`
 	Override map[zfsprop.Property]string `yaml:"override,optional"`

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/bandwidth_limit.yml

@@ -0,0 +1,41 @@
+
+jobs:
+  - type: sink
+    name: "limited_sink"
+    root_fs: "fs0"
+    recv:
+      bandwidth_limit:
+        max: 12345 B
+    serve:
+      type: local
+      listener_name: localsink
+
+  - type: push
+    name: "limited_push"
+    connect:
+      type: local
+      listener_name: localsink
+      client_identity: local_backup
+    filesystems: {
+      "root<": true,
+    }
+    send:
+      bandwidth_limit:
+        max: 54321 B
+        bucket_capacity: 1024 B
+    snapshotting:
+      type: manual
+    pruning:
+      keep_sender:
+        - type: last_n
+          count: 1
+      keep_receiver:
+        - type: last_n
+          count: 1
+
+  - type: sink
+    name: "nolimit_sink"
+    root_fs: "fs1"
+    serve:
+      type: local
+      listener_name: localsink

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

@@ -143,12 +143,10 @@ func (j *controlJob) Run(ctx context.Context) {
 
 			var err error
 			switch req.Op {
-			case "replication":
-				err = j.jobs.doreplication(req.Name)
+			case "wakeup":
+				err = j.jobs.wakeup(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)
 			}

daemon/daemon.go

@@ -20,9 +20,8 @@ import (
 
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/job"
-	"github.com/zrepl/zrepl/daemon/job/doreplication"
-	"github.com/zrepl/zrepl/daemon/job/dosnapshot"
 	"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"
@@ -132,19 +131,17 @@ type jobs struct {
 	wg sync.WaitGroup
 
 	// m protects all fields below it
-	m              sync.RWMutex
-	doreplications map[string]doreplication.Func // by Job.Name
-	resets         map[string]reset.Func         // by Job.Name
-	dosnapshots    map[string]dosnapshot.Func    // by Job.Name
-	jobs           map[string]job.Job
+	m       sync.RWMutex
+	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{
-		doreplications: make(map[string]doreplication.Func),
-		resets:         make(map[string]reset.Func),
-		dosnapshots:    make(map[string]dosnapshot.Func),
-		jobs:           make(map[string]job.Job),
+		wakeups: make(map[string]wakeup.Func),
+		resets:  make(map[string]reset.Func),
+		jobs:    make(map[string]job.Job),
 	}
 }
 
@@ -193,11 +190,11 @@ func (s *jobs) status() map[string]*job.Status {
 	return ret
 }
 
-func (s *jobs) doreplication(job string) error {
+func (s *jobs) wakeup(job string) error {
 	s.m.RLock()
 	defer s.m.RUnlock()
 
-	wu, ok := s.doreplications[job]
+	wu, ok := s.wakeups[job]
 	if !ok {
 		return errors.Errorf("Job %s does not exist", job)
 	}
@@ -215,17 +212,6 @@ func (s *jobs) reset(job string) error {
 	return wu()
 }
 
-func (s *jobs) dosnapshot(job string) error {
-	s.m.RLock()
-	defer s.m.RUnlock()
-
-	wu, ok := s.dosnapshots[job]
-	if !ok {
-		return errors.Errorf("Job %s does not exist", job)
-	}
-	return wu()
-}
-
 const (
 	jobNamePrometheus = "_prometheus"
 	jobNameControl    = "_control"
@@ -256,12 +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, doreplication := doreplication.Context(ctx)
+	ctx, wakeup := wakeup.Context(ctx)
 	ctx, resetFunc := reset.Context(ctx)
-	ctx, dosnapshotFunc := dosnapshot.Context(ctx)
-	s.doreplications[jobName] = doreplication
+	s.wakeups[jobName] = wakeup
 	s.resets[jobName] = resetFunc
-	s.dosnapshots[jobName] = dosnapshotFunc
 
 	s.wg.Add(1)
 	go func() {

daemon/job/active.go

@@ -14,8 +14,8 @@ import (
 	"github.com/zrepl/zrepl/util/envconst"
 
 	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon/job/doreplication"
 	"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"
@@ -89,7 +89,7 @@ type activeMode interface {
 	SenderReceiver() (logic.Sender, logic.Receiver)
 	Type() Type
 	PlannerPolicy() logic.PlannerPolicy
-	RunPeriodic(ctx context.Context, 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, replicationCommon chan<- struct{}) {
-	m.snapper.Run(ctx, replicationCommon)
+func (m *modePush) RunPeriodic(ctx context.Context, wakeUpCommon chan<- struct{}) {
+	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, 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, replicationCommon chan<- str
 		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
@@ -435,13 +435,13 @@ func (j *ActiveSide) 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):
 			j.mode.ResetConnectBackoff()
 		case <-periodicDone:
 		}

daemon/job/build_jobs.go

@@ -8,6 +8,7 @@ import (
 	"github.com/pkg/errors"
 
 	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/util/bandwidthlimit"
 )
 
 func JobsFromConfig(c *config.Config) ([]Job, error) {
@@ -107,3 +108,13 @@ func validateReceivingSidesDoNotOverlap(receivingRootFSs []string) error {
 	}
 	return nil
 }
+
+func buildBandwidthLimitConfig(in *config.BandwidthLimit) (c bandwidthlimit.Config, _ error) {
+	if in.Max.ToBytes() > 0 && int64(in.Max.ToBytes()) == 0 {
+		return c, fmt.Errorf("bandwidth limit `max` is too small, must at least specify one byte")
+	}
+	return bandwidthlimit.Config{
+		Max:            int64(in.Max.ToBytes()),
+		BucketCapacity: int64(in.BucketCapacity.ToBytes()),
+	}, nil
+}

daemon/job/build_jobs_sendrecvoptions.go

@@ -22,7 +22,12 @@ func buildSenderConfig(in SendingJobConfig, jobID endpoint.JobID) (*endpoint.Sen
 		return nil, errors.Wrap(err, "cannot build filesystem filter")
 	}
 	sendOpts := in.GetSendOptions()
-	return &endpoint.SenderConfig{
+	bwlim, err := buildBandwidthLimitConfig(sendOpts.BandwidthLimit)
+	if err != nil {
+		return nil, errors.Wrap(err, "cannot build bandwith limit config")
+	}
+
+	sc := &endpoint.SenderConfig{
 		FSF:   fsf,
 		JobID: jobID,
 
@@ -34,7 +39,15 @@ func buildSenderConfig(in SendingJobConfig, jobID endpoint.JobID) (*endpoint.Sen
 		SendCompressed:       sendOpts.Compressed,
 		SendEmbeddedData:     sendOpts.EmbeddedData,
 		SendSaved:            sendOpts.Saved,
-	}, nil
+
+		BandwidthLimit: bwlim,
+	}
+
+	if err := sc.Validate(); err != nil {
+		return nil, errors.Wrap(err, "cannot build sender config")
+	}
+
+	return sc, nil
 }
 
 type ReceivingJobConfig interface {
@@ -53,6 +66,12 @@ func buildReceiverConfig(in ReceivingJobConfig, jobID endpoint.JobID) (rc endpoi
 	}
 
 	recvOpts := in.GetRecvOptions()
+
+	bwlim, err := buildBandwidthLimitConfig(recvOpts.BandwidthLimit)
+	if err != nil {
+		return rc, errors.Wrap(err, "cannot build bandwith limit config")
+	}
+
 	rc = endpoint.ReceiverConfig{
 		JobID:                      jobID,
 		RootWithoutClientComponent: rootFs,
@@ -60,6 +79,8 @@ func buildReceiverConfig(in ReceivingJobConfig, jobID endpoint.JobID) (rc endpoi
 
 		InheritProperties:  recvOpts.Properties.Inherit,
 		OverrideProperties: recvOpts.Properties.Override,
+
+		BandwidthLimit: bwlim,
 	}
 	if err := rc.Validate(); err != nil {
 		return rc, errors.Wrap(err, "cannot build receiver config")

daemon/job/build_jobs_test.go

@@ -119,6 +119,14 @@ func TestSampleConfigsAreBuiltWithoutErrors(t *testing.T) {
 		t.Errorf("glob failed: %+v", err)
 	}
 
+	type additionalCheck struct {
+		state int
+		test  func(t *testing.T, jobs []Job)
+	}
+	additionalChecks := map[string]*additionalCheck{
+		"bandwidth_limit.yml": {test: testSampleConfig_BandwidthLimit},
+	}
+
 	for _, p := range paths {
 
 		if path.Ext(p) != ".yml" {
@@ -126,10 +134,20 @@ func TestSampleConfigsAreBuiltWithoutErrors(t *testing.T) {
 			continue
 		}
 
+		filename := path.Base(p)
+		t.Logf("checking for presence additonal checks for file %q", filename)
+		additionalCheck := additionalChecks[filename]
+		if additionalCheck == nil {
+			t.Logf("no additional checks")
+		} else {
+			t.Logf("additional check present")
+			additionalCheck.state = 1
+		}
+
 		t.Run(p, func(t *testing.T) {
 			c, err := config.ParseConfig(p)
 			if err != nil {
-				t.Errorf("error parsing %s:\n%+v", p, err)
+				t.Fatalf("error parsing %s:\n%+v", p, err)
 			}
 
 			t.Logf("file: %s", p)
@@ -138,11 +156,57 @@ func TestSampleConfigsAreBuiltWithoutErrors(t *testing.T) {
 			tls.FakeCertificateLoading(t)
 			jobs, err := JobsFromConfig(c)
 			t.Logf("jobs: %#v", jobs)
-			assert.NoError(t, err)
+			require.NoError(t, err)
+
+			if additionalCheck != nil {
+				additionalCheck.test(t, jobs)
+				additionalCheck.state = 2
+			}
 		})
 
 	}
 
+	for basename, c := range additionalChecks {
+		if c.state == 0 {
+			panic("univisited additional check " + basename)
+		}
+	}
+
+}
+
+func testSampleConfig_BandwidthLimit(t *testing.T, jobs []Job) {
+	require.Len(t, jobs, 3)
+
+	{
+		limitedSink, ok := jobs[0].(*PassiveSide)
+		require.True(t, ok, "%T", jobs[0])
+		limitedSinkMode, ok := limitedSink.mode.(*modeSink)
+		require.True(t, ok, "%T", limitedSink)
+
+		assert.Equal(t, int64(12345), limitedSinkMode.receiverConfig.BandwidthLimit.Max)
+		assert.Equal(t, int64(1<<17), limitedSinkMode.receiverConfig.BandwidthLimit.BucketCapacity)
+	}
+
+	{
+		limitedPush, ok := jobs[1].(*ActiveSide)
+		require.True(t, ok, "%T", jobs[1])
+		limitedPushMode, ok := limitedPush.mode.(*modePush)
+		require.True(t, ok, "%T", limitedPush)
+
+		assert.Equal(t, int64(54321), limitedPushMode.senderConfig.BandwidthLimit.Max)
+		assert.Equal(t, int64(1024), limitedPushMode.senderConfig.BandwidthLimit.BucketCapacity)
+	}
+
+	{
+		unlimitedSink, ok := jobs[2].(*PassiveSide)
+		require.True(t, ok, "%T", jobs[2])
+		unlimitedSinkMode, ok := unlimitedSink.mode.(*modeSink)
+		require.True(t, ok, "%T", unlimitedSink)
+
+		max := unlimitedSinkMode.receiverConfig.BandwidthLimit.Max
+		assert.Less(t, max, int64(0), max, "unlimited mode <=> negative value for .Max, see bandwidthlimit.Config")
+	}
+
 }
 
 func TestReplicationOptions(t *testing.T) {

daemon/job/doreplication/doreplication.go

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

daemon/job/snapjob.go

@@ -9,12 +9,12 @@ import (
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 
-	"github.com/zrepl/zrepl/daemon/job/doreplication"
 	"github.com/zrepl/zrepl/daemon/logging/trace"
 	"github.com/zrepl/zrepl/util/nodefault"
 
 	"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"
@@ -112,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/wakeup/wakeup.go

@@ -1,4 +1,4 @@
-package dosnapshot
+package wakeup
 
 import (
 	"context"
@@ -7,10 +7,10 @@ import (
 
 type contextKey int
 
-const contextKeyDosnapshot contextKey = iota
+const contextKeyWakeup contextKey = iota
 
 func Wait(ctx context.Context) <-chan struct{} {
-	wc, ok := ctx.Value(contextKeyDosnapshot).(chan struct{})
+	wc, ok := ctx.Value(contextKeyWakeup).(chan struct{})
 	if !ok {
 		wc = make(chan struct{})
 	}
@@ -19,7 +19,7 @@ func Wait(ctx context.Context) <-chan struct{} {
 
 type Func func() error
 
-var AlreadyDosnapshot = errors.New("already snapshotting")
+var AlreadyWokenUp = errors.New("already woken up")
 
 func Context(ctx context.Context) (context.Context, Func) {
 	wc := make(chan struct{})
@@ -28,8 +28,8 @@ func Context(ctx context.Context) (context.Context, Func) {
 		case wc <- struct{}{}:
 			return nil
 		default:
-			return AlreadyDosnapshot
+			return AlreadyWokenUp
 		}
 	}
-	return context.WithValue(ctx, contextKeyDosnapshot, wc), wuf
+	return context.WithValue(ctx, contextKeyWakeup, wc), wuf
 }

daemon/snapper/snapper.go

@@ -9,7 +9,6 @@ import (
 
 	"github.com/pkg/errors"
 
-	"github.com/zrepl/zrepl/daemon/job/dosnapshot"
 	"github.com/zrepl/zrepl/daemon/logging/trace"
 
 	"github.com/zrepl/zrepl/config"
@@ -211,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)
 	}
@@ -383,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,
-          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:
 

docs/configuration/sendrecvoptions.rst

@@ -36,6 +36,9 @@ See the `upstream man page <https://openzfs.github.io/openzfs-docs/man/8/zfs-sen
     * - ``encrypted``
       -
       - Specific to zrepl, :ref:`see below <job-send-options-encrypted>`.
+    * - ``bandwidth_limit``
+      -
+      - Specific to zrepl, :ref:`see below <job-send-recv-options-bandwidth-limit>`.
     * - ``raw``
       - ``-w``
       - Use ``encrypted`` to only allow encrypted sends.
@@ -138,6 +141,7 @@ Recv Options
          override: {
            "org.openzfs.systemd:ignore": "on"
          }
+       bandwidth_limit: ... # see below
      ...
 
 .. _job-recv-options--inherit-and-override:
@@ -212,3 +216,25 @@ and property replication is enabled, the receiver must :ref:`inherit the followi
 * ``keylocation``
 * ``keyformat``
 * ``encryption``
+
+Common Options
+~~~~~~~~~~~~~~
+
+.. _job-send-recv-options-bandwidth-limit:
+
+Bandwidth Limit (send & recv)
+-----------------------------
+
+::
+
+   bandwidth_limit:
+     max: 23.5 MiB # -1 is the default and disabled rate limiting
+     bucket_capacity: # token bucket capacity in bytes; defaults to 128KiB
+
+Both ``send`` and ``recv`` can be limited to a maximum bandwidth through ``bandwidth_limit``.
+For most users, it should be sufficient to just set ``bandwidth_limit.max``.
+The ``bandwidth_limit.bucket_capacity`` refers to the `token bucket size <https://github.com/juju/ratelimit>`_.
+
+The bandwidth limit only applies to the payload data, i.e., the ZFS send stream.
+It does not account for transport protocol overheads.
+The scope is the job level, i.e., all :ref:`concurrent <replication-option-concurrency>` sends or incoming receives of a job share the bandwidth limit.

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 -) && \
-    (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::

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``.
-(refer to the example use case document if you are uncertain which job you want to start replication).
+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 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/supporters.rst

@@ -32,6 +32,8 @@ We would like to thank the following people and organizations for supporting zre
 
    <div class="fa fa-code" style="width: 1em;"></div>
 
+* |supporter-gold| Prominic.NET, Inc.
+* |supporter-std| Torsten Blum
 * |supporter-gold| Cyberiada GmbH
 * |supporter-std| `Gordon Schulz <https://github.com/azmodude>`_
 * |supporter-std| `@jwittlincohen <https://github.com/jwittlincohen>`_

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

endpoint/endpoint.go

@@ -14,6 +14,7 @@ import (
 	"github.com/zrepl/zrepl/daemon/logging/trace"
 
 	"github.com/zrepl/zrepl/replication/logic/pdu"
+	"github.com/zrepl/zrepl/util/bandwidthlimit"
 	"github.com/zrepl/zrepl/util/chainedio"
 	"github.com/zrepl/zrepl/util/chainlock"
 	"github.com/zrepl/zrepl/util/envconst"
@@ -34,6 +35,8 @@ type SenderConfig struct {
 	SendCompressed       bool
 	SendEmbeddedData     bool
 	SendSaved            bool
+
+	BandwidthLimit bandwidthlimit.Config
 }
 
 func (c *SenderConfig) Validate() error {
@@ -44,6 +47,9 @@ func (c *SenderConfig) Validate() error {
 	if _, err := StepHoldTag(c.JobID); err != nil {
 		return fmt.Errorf("JobID cannot be used for hold tag: %s", err)
 	}
+	if err := bandwidthlimit.ValidateConfig(c.BandwidthLimit); err != nil {
+		return errors.Wrap(err, "`BandwidthLimit` field invalid")
+	}
 	return nil
 }
 
@@ -54,16 +60,21 @@ type Sender struct {
 	FSFilter zfs.DatasetFilter
 	jobId    JobID
 	config   SenderConfig
+	bwLimit  bandwidthlimit.Wrapper
 }
 
 func NewSender(conf SenderConfig) *Sender {
 	if err := conf.Validate(); err != nil {
 		panic("invalid config" + err.Error())
 	}
+
+	ratelimiter := bandwidthlimit.WrapperFromConfig(conf.BandwidthLimit)
+
 	return &Sender{
 		FSFilter: conf.FSF,
 		jobId:    conf.JobID,
 		config:   conf,
+		bwLimit:  ratelimiter,
 	}
 }
 
@@ -301,12 +312,16 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
 		abstractionsCacheSingleton.TryBatchDestroy(ctx, s.jobId, sendArgs.FS, destroyTypes, keep, check)
 	}()
 
-	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
+	var sendStream io.ReadCloser
+	sendStream, err = zfs.ZFSSend(ctx, sendArgs)
 	if err != nil {
 		// it's ok to not destroy the abstractions we just created here, a new send attempt will take care of it
 		return nil, nil, errors.Wrap(err, "zfs send failed")
 	}
 
+	// apply rate limit
+	sendStream = s.bwLimit.WrapReadCloser(sendStream)
+
 	return res, sendStream, nil
 }
 
@@ -439,6 +454,8 @@ type ReceiverConfig struct {
 
 	InheritProperties  []zfsprop.Property
 	OverrideProperties map[zfsprop.Property]string
+
+	BandwidthLimit bandwidthlimit.Config
 }
 
 func (c *ReceiverConfig) copyIn() {
@@ -475,6 +492,11 @@ func (c *ReceiverConfig) Validate() error {
 	if c.RootWithoutClientComponent.Length() <= 0 {
 		return errors.New("RootWithoutClientComponent must not be an empty dataset path")
 	}
+
+	if err := bandwidthlimit.ValidateConfig(c.BandwidthLimit); err != nil {
+		return errors.Wrap(err, "`BandwidthLimit` field invalid")
+	}
+
 	return nil
 }
 
@@ -484,6 +506,8 @@ type Receiver struct {
 
 	conf ReceiverConfig // validated
 
+	bwLimit bandwidthlimit.Wrapper
+
 	recvParentCreationMtx *chainlock.L
 }
 
@@ -495,6 +519,7 @@ func NewReceiver(config ReceiverConfig) *Receiver {
 	return &Receiver{
 		conf:                  config,
 		recvParentCreationMtx: chainlock.New(),
+		bwLimit:               bandwidthlimit.WrapperFromConfig(config.BandwidthLimit),
 	}
 }
 
@@ -787,6 +812,9 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.
 		return nil, errors.Wrap(err, "cannot determine whether we can use resumable send & recv")
 	}
 
+	// apply rate limit
+	receive = s.bwLimit.WrapReadCloser(receive)
+
 	var peek bytes.Buffer
 	var MaxPeek = envconst.Int64("ZREPL_ENDPOINT_RECV_PEEK_SIZE", 1<<20)
 	log.WithField("max_peek_bytes", MaxPeek).Info("peeking incoming stream")

go.mod

@@ -15,6 +15,7 @@ require (
 	github.com/golang/protobuf v1.4.3
 	github.com/google/uuid v1.1.2
 	github.com/jinzhu/copier v0.0.0-20170922082739-db4671f3a9b8
+	github.com/juju/ratelimit v1.0.1
 	github.com/kisielk/gotool v1.0.0 // indirect
 	github.com/kr/pretty v0.1.0
 	github.com/leodido/go-urn v1.2.1 // indirect

go.sum

@@ -164,6 +164,8 @@ github.com/jinzhu/copier v0.0.0-20170922082739-db4671f3a9b8 h1:+dKzeuiDYbD/Cfi/s
 github.com/jinzhu/copier v0.0.0-20170922082739-db4671f3a9b8/go.mod h1:yL958EeXv8Ylng6IfnvG4oflryUi3vgA3xPs9hmII1s=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/json-iterator/go v1.1.7/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
+github.com/juju/ratelimit v1.0.1 h1:+7AIFJVQ0EQgq/K9+0Krm7m530Du7tIz0METWzN0RgY=
+github.com/juju/ratelimit v1.0.1/go.mod h1:qapgC/Gy+xNh9UxzV13HGGl/6UXNN+ct+vwSgWNm/qk=
 github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
 github.com/jung-kurt/gofpdf v1.0.3-0.20190309125859-24315acbbda5/go.mod h1:7Id9E/uU8ce6rXgefFLlgrJj/GYY22cpxn+r32jIOes=
 github.com/k0kubun/colorstring v0.0.0-20150214042306-9440f1994b88 h1:uC1QfSlInpQF+M0ao65imhwqKnz3Q2z/d8PWZRMQvDM=

packaging/deb/Dockerfile

@@ -1,9 +1,13 @@
 FROM debian:latest
 
+# binutils are for cross-compilation to work in bullseye
 RUN apt-get update && apt-get install -y \
 	build-essential \
     devscripts \
-    dh-exec
+    dh-exec \
+	binutils-aarch64-linux-gnu \
+	binutils-arm-linux-gnueabihf  \
+	binutils-i686-linux-gnu
 
 RUN mkdir -p /build/src && chmod -R 0777 /build
 

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"
@@ -20,6 +21,7 @@ import (
 	"github.com/zrepl/zrepl/replication/logic"
 	"github.com/zrepl/zrepl/replication/logic/pdu"
 	"github.com/zrepl/zrepl/replication/report"
+	"github.com/zrepl/zrepl/util/bandwidthlimit"
 	"github.com/zrepl/zrepl/util/limitio"
 	"github.com/zrepl/zrepl/util/nodefault"
 	"github.com/zrepl/zrepl/zfs"
@@ -63,9 +65,10 @@ func (i replicationInvocation) Do(ctx *platformtest.Context) *report.Report {
 	}
 
 	senderConfig := endpoint.SenderConfig{
-		FSF:     i.sfilter.AsFilter(),
-		Encrypt: &nodefault.Bool{B: false},
-		JobID:   i.sjid,
+		FSF:            i.sfilter.AsFilter(),
+		Encrypt:        &nodefault.Bool{B: false},
+		JobID:          i.sjid,
+		BandwidthLimit: bandwidthlimit.NoLimitConfig(),
 	}
 	if i.senderConfigHook != nil {
 		i.senderConfigHook(&senderConfig)
@@ -75,6 +78,7 @@ func (i replicationInvocation) Do(ctx *platformtest.Context) *report.Report {
 		JobID:                      i.rjid,
 		AppendClientIdentity:       false,
 		RootWithoutClientComponent: mustDatasetPath(i.rfsRoot),
+		BandwidthLimit:             bandwidthlimit.NoLimitConfig(),
 	}
 	if i.receiverConfigHook != nil {
 		i.receiverConfigHook(&receiverConfig)
@@ -90,6 +94,7 @@ func (i replicationInvocation) Do(ctx *platformtest.Context) *report.Report {
 		ReplicationConfig: &pdu.ReplicationConfig{
 			Protection: i.guarantee,
 		},
+		SizeEstimationConcurrency: 1,
 	}
 
 	report, wait := replication.Do(
@@ -1054,6 +1059,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)

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)

rpc/dataconn/dataconn_server.go

@@ -241,7 +241,7 @@ func (s *Server) serveConnRequest(ctx context.Context, endpoint string, c *strea
 		err := c.SendStream(ctx, sendStream, ZFSStream)
 		closeErr := sendStream.Close()
 		if closeErr != nil {
-			s.log.WithError(err).Error("cannot close send stream")
+			s.log.WithError(closeErr).Error("cannot close send stream")
 		}
 		if err != nil {
 			s.log.WithError(err).Error("cannot write send stream")

rpc/transportmux/transportmux.go

@@ -7,6 +7,7 @@ package transportmux
 
 import (
 	"context"
+	"os"
 	"sync/atomic"
 	"syscall"
 
@@ -15,6 +16,7 @@ import (
 	"net"
 	"time"
 
+	"github.com/kr/pretty"
 	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/transport"
@@ -151,6 +153,7 @@ func Demux(ctx context.Context, rawListener transport.AuthenticatedListener, lab
 
 			rawConn, err := rawListener.Accept(ctx)
 			if err != nil {
+				fmt.Fprintf(os.Stderr, "transportmux.Demux: rawListener.Accept() returned error: %T %s\n%s\n", err, err, pretty.Sprint(err))
 				if ctx.Err() != nil {
 					return
 				}

rpc/versionhandshake/versionhandshake.go

@@ -10,9 +10,12 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"os"
 	"strings"
 	"time"
 	"unicode/utf8"
+
+	"github.com/kr/pretty"
 )
 
 type HandshakeMessage struct {
@@ -93,6 +96,7 @@ func (m *HandshakeMessage) Encode() ([]byte, error) {
 func (m *HandshakeMessage) DecodeReader(r io.Reader, maxLen int) error {
 	var lenAndSpace [11]byte
 	if _, err := io.ReadFull(r, lenAndSpace[:]); err != nil {
+		fmt.Fprintf(os.Stderr, "HandshakeMessage.DecodeReader error: %T\n%s", err, pretty.Sprint(err))
 		return hsIOErr(err, "error reading protocol banner length: %s", err)
 	}
 	if !utf8.Valid(lenAndSpace[:]) {

rpc/versionhandshake/versionhandshake_transport_wrappers.go

@@ -2,7 +2,9 @@ package versionhandshake
 
 import (
 	"context"
+	"fmt"
 	"net"
+	"os"
 	"time"
 
 	"github.com/zrepl/zrepl/transport"
@@ -23,6 +25,7 @@ func (c HandshakeConnecter) Connect(ctx context.Context) (transport.Wire, error)
 		dl = time.Now().Add(c.timeout)
 	}
 	if err := DoHandshakeCurrentVersion(conn, dl); err != nil {
+		fmt.Fprintf(os.Stderr, "HandshakeConnecter error: %T\n\t%s\n\t%v\n\t%#v\n\n", err, err, err, err)
 		conn.Close()
 		return nil, err
 	}

transport/tcp/serve_tcp_ipmap_test.go

@@ -91,15 +91,12 @@ func TestIPMap(t *testing.T) {
 				"fde4:8dba:82e1::/64": "sub64-*",
 			},
 			expect: map[string]testCaseExpect{
-				"10.1.2.3":        {expectNoMapping: true},
-				"192.168.23.1":    {expectIdent: "db-192.168.23.1"},
-				"192.168.23.23":   {expectIdent: "db-twentythree"},
-				"192.168.023.001": {expectIdent: "db-192.168.23.1"},
-				"10.1.4.5":        {expectIdent: "my-10.1.4.5-server"},
+				"10.1.2.3":      {expectNoMapping: true},
+				"192.168.23.1":  {expectIdent: "db-192.168.23.1"},
+				"192.168.42.1":  {expectIdent: "web-192.168.42.1"},
+				"192.168.23.23": {expectIdent: "db-twentythree"},
+				"10.1.4.5":      {expectIdent: "my-10.1.4.5-server"},
 
-				// normalization
-				"192.168.42.1":    {expectIdent: "web-192.168.42.1"},
-				"192.168.042.001": {expectIdent: "web-192.168.42.1"},
 				// v6 matching
 				"fe80::23:42%eth1": {expectIdent: "san-fe80::23:42-eth1"},
 				"fe80::23:42%eth2": {expectNoMapping: true},
@@ -179,7 +176,8 @@ func TestIPMap(t *testing.T) {
 			for input, expect := range c.expect {
 				// reuse newIPMapEntry to parse test case input
 				// "test" is not used during testing but must not be empty.
-				ipMapEntry, _ := newIPMapEntry(input, "test")
+				ipMapEntry, err := newIPMapEntry(input, "test")
+				require.NoError(t, err)
 				ones, bits := ipMapEntry.subnet.Mask.Size()
 				require.Equal(t, bits, net.IPv6len*8, "and we know ipMapEntry always expands its IPs to 16bytes")
 				require.Equal(t, ones, net.IPv6len*8, "test case addresses must be fully specified")

transport/tls/connect_tls.go

@@ -3,7 +3,9 @@ package tls
 import (
 	"context"
 	"crypto/tls"
+	"fmt"
 	"net"
+	"os"
 
 	"github.com/pkg/errors"
 
@@ -48,6 +50,7 @@ func TLSConnecterFromConfig(in *config.TLSConnect) (*TLSConnecter, error) {
 func (c *TLSConnecter) Connect(dialCtx context.Context) (transport.Wire, error) {
 	conn, err := c.dialer.DialContext(dialCtx, "tcp", c.Address)
 	if err != nil {
+		fmt.Fprintf(os.Stderr, "tls connecter error %T\n\t%s\n\t%v\n\t%#v\n\n", err, err, err, err)
 		return nil, err
 	}
 	tcpConn := conn.(*net.TCPConn)

transport/tls/issue-517/main.go

@@ -0,0 +1,140 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"io"
+	"log"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/transport/tls"
+)
+
+var servConf = config.TLSServe{
+	ServeCommon: config.ServeCommon{
+		Type: "tls",
+	},
+	HandshakeTimeout: 10 * time.Second,
+}
+
+var clientConf = config.TLSConnect{
+	ConnectCommon: config.ConnectCommon{
+		Type: "",
+	},
+	Address:     "",
+	Ca:          "",
+	Cert:        "",
+	Key:         "",
+	ServerCN:    "",
+	DialTimeout: 10 * time.Second,
+}
+
+var ca string
+var mode string
+
+func main() {
+
+	flag.StringVar(&mode, "mode", "", "server|client")
+
+	flag.StringVar(&ca, "ca", "", "path")
+
+	flag.StringVar(&servConf.Listen, "serve.listen", "", "")
+	flag.StringVar(&servConf.Cert, "serve.cert", "", "path")
+	flag.StringVar(&servConf.Key, "serve.key", "", "path")
+	var clientCN string
+	flag.StringVar(&clientCN, "serve.client_cn", "", "")
+
+	flag.StringVar(&clientConf.Address, "client.address", "", "")
+	flag.StringVar(&clientConf.Cert, "client.cert", "", "path")
+	flag.StringVar(&clientConf.Key, "client.key", "", "path")
+	flag.StringVar(&clientConf.ServerCN, "client.server_cn", "", "")
+
+	flag.Parse()
+
+	servConf.ClientCNs = append(servConf.ClientCNs, clientCN)
+
+	servConf.Ca = ca
+	clientConf.Ca = ca
+
+	switch mode {
+	case "server":
+		server()
+	case "client":
+		client()
+	default:
+		panic(mode)
+	}
+
+}
+
+func server() {
+
+	servFactory, err := tls.TLSListenerFactoryFromConfig(nil, &servConf)
+	if err != nil {
+		panic(err)
+	}
+
+	listener, err := servFactory()
+	if err != nil {
+		panic(err)
+	}
+
+	ctx := context.Background()
+
+	for {
+		conn, err := listener.Accept(ctx)
+		if err != nil {
+			log.Printf("accept error: %s", err)
+			continue
+		}
+		go func() {
+			defer conn.Close()
+
+			log.Printf("handling connection %s", conn.RemoteAddr())
+			_, err = io.Copy(conn, strings.NewReader("here is the server\n"))
+			if err != nil {
+				log.Printf("%s: respond to client error: %s", conn.RemoteAddr(), err)
+				return
+			}
+
+			err = conn.CloseWrite()
+			if err != nil {
+				log.Printf("%s: failed to close write connection: err", conn.RemoteAddr(), err)
+			}
+
+			log.Printf("%s: waiting for client to close connection", conn.RemoteAddr())
+
+			_, err = io.Copy(io.Discard, conn)
+			if err != nil {
+				log.Printf("%s: error draining client connection: %s", conn.RemoteAddr(), err)
+				return
+			}
+
+			log.Printf("%s: done", conn.RemoteAddr())
+			return
+		}()
+	}
+
+}
+
+func client() {
+	connecter, err := tls.TLSConnecterFromConfig(&clientConf)
+	if err != nil {
+		panic(err)
+	}
+
+	ctx := context.Background()
+	conn, err := connecter.Connect(ctx)
+	if err != nil {
+		panic(err)
+	}
+	defer conn.Close()
+
+	_, err = io.Copy(os.Stdout, conn)
+	if err != nil {
+		panic(err)
+	}
+}

transport/tls/serve_tls.go

@@ -4,8 +4,10 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+	"os"
 	"time"
 
+	"github.com/kr/pretty"
 	"github.com/pkg/errors"
 
 	"github.com/zrepl/zrepl/config"
@@ -68,6 +70,7 @@ type tlsAuthListener struct {
 func (l tlsAuthListener) Accept(ctx context.Context) (*transport.AuthConn, error) {
 	tcpConn, tlsConn, cn, err := l.ClientAuthListener.Accept()
 	if err != nil {
+		fmt.Fprintf(os.Stderr, "tlsAuthListener.Accept: l.ClientAuthListener.Accept returned error %T %s\n%s\n", err, err, pretty.Sprint(err))
 		return nil, err
 	}
 	if _, ok := l.clientCNs[cn]; !ok {

util/bandwidthlimit/bandwidthlimit.go

@@ -0,0 +1,79 @@
+package bandwidthlimit
+
+import (
+	"errors"
+	"io"
+
+	"github.com/juju/ratelimit"
+)
+
+type Wrapper interface {
+	WrapReadCloser(io.ReadCloser) io.ReadCloser
+}
+
+type Config struct {
+	// Units in this struct are in _bytes_.
+
+	Max            int64 // < 0 means no limit, BucketCapacity is irrelevant then
+	BucketCapacity int64
+}
+
+func NoLimitConfig() Config {
+	return Config{
+		Max:            -1,
+		BucketCapacity: -1,
+	}
+}
+
+func ValidateConfig(conf Config) error {
+	if conf.BucketCapacity == 0 {
+		return errors.New("BucketCapacity must not be zero")
+	}
+	return nil
+}
+
+func WrapperFromConfig(conf Config) Wrapper {
+	if err := ValidateConfig(conf); err != nil {
+		panic(err)
+	}
+
+	if conf.Max < 0 {
+		return noLimit{}
+	}
+
+	return &withLimit{
+		bucket: ratelimit.NewBucketWithRate(float64(conf.Max), conf.BucketCapacity),
+	}
+}
+
+type noLimit struct{}
+
+func (_ noLimit) WrapReadCloser(rc io.ReadCloser) io.ReadCloser { return rc }
+
+type withLimit struct {
+	bucket *ratelimit.Bucket
+}
+
+func (l *withLimit) WrapReadCloser(rc io.ReadCloser) io.ReadCloser {
+	return WrapReadCloser(rc, l.bucket)
+}
+
+type withLimitReadCloser struct {
+	orig    io.Closer
+	limited io.Reader
+}
+
+func (r *withLimitReadCloser) Read(buf []byte) (int, error) {
+	return r.limited.Read(buf)
+}
+
+func (r *withLimitReadCloser) Close() error {
+	return r.orig.Close()
+}
+
+func WrapReadCloser(rc io.ReadCloser, bucket *ratelimit.Bucket) io.ReadCloser {
+	return &withLimitReadCloser{
+		limited: ratelimit.Reader(rc, bucket),
+		orig:    rc,
+	}
+}

util/bandwidthlimit/bandwidthlimit_test.go

@@ -0,0 +1,19 @@
+package bandwidthlimit
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestNoLimitConfig(t *testing.T) {
+
+	conf := NoLimitConfig()
+
+	err := ValidateConfig(conf)
+	require.NoError(t, err)
+
+	require.NotPanics(t, func() {
+		_ = WrapperFromConfig(conf)
+	})
+}

util/datasizeunit/datasizeunit.go

@@ -0,0 +1,86 @@
+package datasizeunit
+
+import (
+	"errors"
+	"fmt"
+	"math"
+	"regexp"
+	"strconv"
+	"strings"
+)
+
+type Bits struct {
+	bits float64
+}
+
+func (b Bits) ToBits() float64    { return b.bits }
+func (b Bits) ToBytes() float64   { return b.bits / 8 }
+func FromBytesInt64(i int64) Bits { return Bits{float64(i) * 8} }
+
+var datarateRegex = regexp.MustCompile(`^([-0-9\.]*)\s*(bit|(|K|Ki|M|Mi|G|Gi|T|Ti)([bB]))$`)
+
+func (r *Bits) UnmarshalYAML(u func(interface{}, bool) error) (_ error) {
+
+	var s string
+	err := u(&s, false)
+	if err != nil {
+		return err
+	}
+
+	genericErr := func(err error) error {
+		var buf strings.Builder
+		fmt.Fprintf(&buf, "cannot parse %q using regex %s", s, datarateRegex)
+		if err != nil {
+			fmt.Fprintf(&buf, ": %s", err)
+		}
+		return errors.New(buf.String())
+	}
+
+	match := datarateRegex.FindStringSubmatch(s)
+	if match == nil {
+		return genericErr(nil)
+	}
+
+	bps, err := strconv.ParseFloat(match[1], 64)
+	if err != nil {
+		return genericErr(err)
+	}
+
+	if match[2] == "bit" {
+		if math.Round(bps) != bps {
+			return genericErr(fmt.Errorf("unit bit must be an integer value"))
+		}
+		r.bits = bps
+		return nil
+	}
+
+	factorMap := map[string]uint64{
+		"": 1,
+
+		"K": 1e3,
+		"M": 1e6,
+		"G": 1e9,
+		"T": 1e12,
+
+		"Ki": 1 << 10,
+		"Mi": 1 << 20,
+		"Gi": 1 << 30,
+		"Ti": 1 << 40,
+	}
+	factor, ok := factorMap[match[3]]
+	if !ok {
+		panic(match)
+	}
+
+	baseUnitFactorMap := map[string]uint64{
+		"b": 1,
+		"B": 8,
+	}
+	baseUnitFactor, ok := baseUnitFactorMap[match[4]]
+	if !ok {
+		panic(match)
+	}
+
+	r.bits = bps * float64(factor) * float64(baseUnitFactor)
+	return nil
+}

util/datasizeunit/datasizeunit_test.go

@@ -0,0 +1,57 @@
+package datasizeunit
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/zrepl/yaml-config"
+)
+
+func TestBits(t *testing.T) {
+
+	tcs := []struct {
+		input      string
+		expectRate float64
+		expectErr  string
+	}{
+		{`23 bit`, 23, ""}, // bit special case works
+		{`23bit`, 23, ""},  // also without space
+
+		{`10MiB`, 10 * (1 << 20) * 8, ""},  // integer unit without space
+		{`10 MiB`, 8 * 10 * (1 << 20), ""}, // integer unit with space
+
+		{`10.5 Kib`, 10.5 * (1 << 10), ""}, // floating point with bit unit works with space
+		{`10.5Kib`, 10.5 * (1 << 10), ""},  // floating point with bit unit works without space
+
+		// unit checks
+		{`1 bit`, 1, ""},
+		{`1 B`, 1 * 8, ""},
+		{`1 Kb`, 1e3, ""},
+		{`1 Kib`, 1 << 10, ""},
+		{`1 Mb`, 1e6, ""},
+		{`1 Mib`, 1 << 20, ""},
+		{`1 Gb`, 1e9, ""},
+		{`1 Gib`, 1 << 30, ""},
+		{`1 Tb`, 1e12, ""},
+		{`1 Tib`, 1 << 40, ""},
+	}
+
+	for _, tc := range tcs {
+		t.Run(tc.input, func(t *testing.T) {
+
+			var bits Bits
+			err := yaml.Unmarshal([]byte(tc.input), &bits)
+			if tc.expectErr != "" {
+				assert.Error(t, err)
+				assert.Regexp(t, tc.expectErr, err.Error())
+				assert.Zero(t, bits.bits)
+			} else {
+				require.NoError(t, err)
+				assert.Equal(t, tc.expectRate, bits.bits)
+			}
+		})
+
+	}
+
+}

zfs/zfs.go

@@ -26,11 +26,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 +323,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 {
@@ -858,7 +853,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 +1155,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 {