config: support zrepl's day and week units for snapshotting.interval

Originally, I had a patch that would replace all usages of
time.Duration in package config with the new config.Duration
types, but:
1. these are all timeouts/retry intervals that have default values.
   Most users don't touch them, and if they do, they don't need
   day or week units.
2. go-yaml's error reporting for yaml.Unmarshaler is inferior to
   built-in types (line numbers are missing, so the error would not have
   sufficient context)

fixes https://github.com/zrepl/zrepl/issues/486

.circleci/config.yml

@@ -24,12 +24,7 @@ commands:
 
   apt-update-and-install-common-deps:
     steps:
-      - run: sudo apt-get update
+      - run: sudo apt update && sudo apt install gawk make
-      - run: sudo apt-get install -y gawk make
-      # CircleCI doesn't update its cimg/go images.
-      # So, need to update manually to get up-to-date trust chains.
-      # The need for this was required for cimg/go:1.12, but let's future proof this here and now.
-      - run: sudo apt-get install -y git ca-certificates
 
   restore-cache-gomod:
     steps:
@@ -155,7 +150,7 @@ parameters:
 
   release_docker_baseimage_tag:
     type: string
-    default: "1.19"
+    default: "1.17"
 
 workflows:
   version: 2
@@ -165,19 +160,19 @@ workflows:
     jobs:
       - quickcheck-docs
       - quickcheck-go: &quickcheck-go-smoketest
-          name: quickcheck-go-amd64-linux-1.19
+          name: quickcheck-go-amd64-linux-1.17
-          goversion: &latest-go-release "1.19"
+          goversion: &latest-go-release "1.17"
           goos: linux
           goarch: amd64
       - test-go-on-latest-go-release:
           goversion: *latest-go-release
       - quickcheck-go:
           requires:
-            - quickcheck-go-amd64-linux-1.19 #quickcheck-go-smoketest.name
+            - quickcheck-go-amd64-linux-1.17 #quickcheck-go-smoketest.name
           matrix: &quickcheck-go-matrix
             alias: quickcheck-go-matrix
             parameters:
-              goversion: [*latest-go-release, "1.18"]
+              goversion: [*latest-go-release, "1.12"]
               goos: ["linux", "freebsd"]
               goarch: ["amd64", "arm64"]
             exclude:
@@ -185,6 +180,10 @@ workflows:
               - goversion: *latest-go-release
                 goos: linux
                 goarch: amd64
+              # not supported by Go 1.12
+              - goversion: "1.12"
+                goos: freebsd
+                goarch: arm64
 
   release:
     when: << pipeline.parameters.do_release >>
@@ -250,7 +249,7 @@ jobs:
       goarch:
         type: string
     docker:
-      - image: cimg/go:<<parameters.goversion>>
+      - image: circleci/golang:<<parameters.goversion>>
     environment:
       GOOS: <<parameters.goos>>
       GOARCH: <<parameters.goarch>>
@@ -258,13 +257,12 @@ jobs:
     steps:
       - checkout
 
-      - install-godep
-
       - restore-cache-gomod
       - run: go mod download
       - run: cd build && go mod download
       - save-cache-gomod
 
+      - install-godep
       - run: make formatcheck
       - run: make generate-platform-test-list
       - run: make zrepl-bin test-platform-bin
@@ -288,7 +286,7 @@ jobs:
       goversion:
         type: string
     docker:
-      - image: cimg/go:<<parameters.goversion>>
+      - image: circleci/golang:<<parameters.goversion>>
     steps:
       - checkout
       - restore-cache-gomod

Makefile

@@ -28,7 +28,7 @@ GO_BUILDFLAGS := $(GO_MOD_READONLY) $(GO_EXTRA_BUILDFLAGS)
 GO_BUILD := $(GO_ENV_VARS) $(GO) build $(GO_BUILDFLAGS) -ldflags $(GO_LDFLAGS)
 GOLANGCI_LINT := golangci-lint
 GOCOVMERGE := gocovmerge
-RELEASE_DOCKER_BASEIMAGE_TAG ?= 1.19
+RELEASE_DOCKER_BASEIMAGE_TAG ?= 1.17
 RELEASE_DOCKER_BASEIMAGE ?= golang:$(RELEASE_DOCKER_BASEIMAGE_TAG)
 
 ifneq ($(GOARM),)
@@ -134,13 +134,7 @@ endif
 
 deb-docker:
 	docker build -t zrepl_debian_pkg --pull -f packaging/deb/Dockerfile .
-	# Use a small open file limit to make fakeroot work. If we don't
-	# specify it, docker daemon will use its file limit. I don't know
-	# what changed (Docker, its systemd service, its Go version). But I
-	# observed fakeroot iterating close(i) up to i > 1000000, which costs
-	# a good amount of CPU time and makes the build slow.
 	docker run --rm -i -v $(CURDIR):/build/src -u $$(id -u):$$(id -g) \
-		--ulimit nofile=1024:1024 \
 		zrepl_debian_pkg \
 		make deb GOOS=$(GOOS) GOARCH=$(GOARCH) GOARM=$(GOARM)
 

build/go.mod

@@ -4,10 +4,13 @@ go 1.12
 
 require (
 	github.com/alvaroloes/enumer v1.1.1
-	github.com/golangci/golangci-lint v1.50.1
+	github.com/golangci/golangci-lint v1.35.2
+	github.com/golangci/misspell v0.3.4 // indirect
+	github.com/golangci/revgrep v0.0.0-20180812185044-276a5c0a1039 // indirect
+	github.com/spf13/afero v1.2.2 // indirect
 	github.com/spf13/jwalterweatherman v1.1.0 // indirect
 	github.com/wadey/gocovmerge v0.0.0-20160331181800-b5bfa59ec0ad
-	golang.org/x/tools v0.2.0
+	golang.org/x/tools v0.0.0-20210105210202-9ed45478a130
 	google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.1.0 // indirect
-	google.golang.org/protobuf v1.28.0
+	google.golang.org/protobuf v1.25.0
 )

client/zfsabstractions.go

@@ -57,7 +57,7 @@ func (f *zabsFilterFlags) registerZabsFilterFlags(s *pflag.FlagSet, verb string)
 	for v := range endpoint.AbstractionTypesAll {
 		variants = append(variants, string(v))
 	}
-	sort.Strings(variants)
+	variants = sort.StringSlice(variants)
 	variantsJoined := strings.Join(variants, "|")
 	s.Var(&f.Types, "type", fmt.Sprintf("only %s holds of the specified type [default: all] [comma-separated list of %s]", verb, variantsJoined))
 

config/config.go

@@ -223,11 +223,10 @@ type SnapshottingEnum struct {
 }
 
 type SnapshottingPeriodic struct {
-	Type            string            `yaml:"type"`
+	Type     string            `yaml:"type"`
-	Prefix          string            `yaml:"prefix"`
+	Prefix   string            `yaml:"prefix"`
-	Interval        *PositiveDuration `yaml:"interval"`
+	Interval *PositiveDuration `yaml:"interval"`
-	Hooks           HookList          `yaml:"hooks,optional"`
+	Hooks    HookList          `yaml:"hooks,optional"`
-	TimestampFormat string            `yaml:"timestamp_format,optional,default=dense"`
 }
 
 type CronSpec struct {
@@ -256,11 +255,10 @@ func (s *CronSpec) UnmarshalYAML(unmarshal func(v interface{}, not_strict bool)
 }
 
 type SnapshottingCron struct {
-	Type            string   `yaml:"type"`
+	Type   string   `yaml:"type"`
-	Prefix          string   `yaml:"prefix"`
+	Prefix string   `yaml:"prefix"`
-	Cron            CronSpec `yaml:"cron"`
+	Cron   CronSpec `yaml:"cron"`
-	Hooks           HookList `yaml:"hooks,optional"`
+	Hooks  HookList `yaml:"hooks,optional"`
-	TimestampFormat string   `yaml:"timestamp_format,optional,default=dense"`
 }
 
 type SnapshottingManual struct {

config/config_snapshotting_test.go

@@ -35,16 +35,8 @@ jobs:
   snapshotting:
     type: periodic
     prefix: zrepl_
-    timestamp_format: dense
     interval: 10m
 `
-	cron := `
-  snapshotting:
-    type: cron
-    prefix: zrepl_
-    timestamp_format: human
-    cron: "10 * * * *"
-`
 
 	periodicDaily := `
   snapshotting:
@@ -99,15 +91,6 @@ jobs:
 		assert.Equal(t, "periodic", snp.Type)
 		assert.Equal(t, 24*time.Hour, snp.Interval.Duration())
 		assert.Equal(t, "zrepl_", snp.Prefix)
-		assert.Equal(t, "dense", snp.TimestampFormat)
-	})
-
-	t.Run("cron", func(t *testing.T) {
-		c = testValidConfig(t, fillSnapshotting(cron))
-		snp := c.Jobs[0].Ret.(*PushJob).Snapshotting.Ret.(*SnapshottingCron)
-		assert.Equal(t, "cron", snp.Type)
-		assert.Equal(t, "zrepl_", snp.Prefix)
-		assert.Equal(t, "human", snp.TimestampFormat)
 	})
 
 	t.Run("hooks", func(t *testing.T) {
@@ -120,57 +103,3 @@ jobs:
 	})
 
 }
-
-func TestSnapshottingTimestampDefaults(t *testing.T) {
-	tmpl := `
-jobs:
-- name: foo
-  type: push
-  connect:
-    type: local
-    listener_name: foo
-    client_identity: bar
-  filesystems: {"<": true}
-  %s
-  pruning:
-    keep_sender:
-    - type: last_n
-      count: 10
-    keep_receiver:
-    - type: last_n
-      count: 10
-`
-
-	periodic := `
-  snapshotting:
-    type: periodic
-    prefix: zrepl_
-    interval: 10m
-`
-	cron := `
-  snapshotting:
-    type: cron
-    prefix: zrepl_
-    cron: "10 * * * *"
-`
-
-	fillSnapshotting := func(s string) string { return fmt.Sprintf(tmpl, s) }
-	var c *Config
-
-	t.Run("periodic", func(t *testing.T) {
-		c = testValidConfig(t, fillSnapshotting(periodic))
-		snp := c.Jobs[0].Ret.(*PushJob).Snapshotting.Ret.(*SnapshottingPeriodic)
-		assert.Equal(t, "periodic", snp.Type)
-		assert.Equal(t, 10*time.Minute, snp.Interval.Duration())
-		assert.Equal(t, "zrepl_", snp.Prefix)
-		assert.Equal(t, "dense", snp.TimestampFormat) // default was set correctly
-	})
-
-	t.Run("cron", func(t *testing.T) {
-		c = testValidConfig(t, fillSnapshotting(cron))
-		snp := c.Jobs[0].Ret.(*PushJob).Snapshotting.Ret.(*SnapshottingCron)
-		assert.Equal(t, "cron", snp.Type)
-		assert.Equal(t, "zrepl_", snp.Prefix)
-		assert.Equal(t, "dense", snp.TimestampFormat) // default was set correctly
-	})
-}

config/config_test.go

@@ -44,8 +44,7 @@ func TestSampleConfigsAreParsedWithoutErrors(t *testing.T) {
 }
 
 // template must be a template/text template with a single '{{ . }}' as placeholder for val
-//
+//nolint[:deadcode,unused]
-//nolint:deadcode,unused
 func testValidConfigTemplate(t *testing.T, tmpl string, val string) *Config {
 	tmp, err := template.New("master").Parse(tmpl)
 	if err != nil {

daemon/hooks/hook_docs.go

@@ -5,7 +5,7 @@
 //
 // This package also provides all supported hook type implementations and abstractions around them.
 //
-// # Use For Other Kinds Of ExpectStepReports
+// Use For Other Kinds Of ExpectStepReports
 //
 // This package REQUIRES REFACTORING before it can be used for other activities than snapshots, e.g. pre- and post-replication:
 //
@@ -15,7 +15,7 @@
 // The hook implementations should move out of this package.
 // However, there is a lot of tight coupling which to untangle isn't worth it ATM.
 //
-// # How This Package Is Used By Package Snapper
+// How This Package Is Used By Package Snapper
 //
 // Deserialize a config.List using ListFromConfig().
 // Then it MUST filter the list to only contain hooks for a particular filesystem using
@@ -30,4 +30,5 @@
 // Command hooks make it available in the environment variable ZREPL_DRYRUN.
 //
 // Plan.Report() can be called while Plan.Run() is executing to give an overview of plan execution progress (future use in "zrepl status").
+//
 package hooks

daemon/hooks/hook_type_mysql_lock_tables.go

@@ -17,22 +17,19 @@ import (
 	"github.com/zrepl/zrepl/zfs"
 )
 
-// Hook to implement the following recommmendation from MySQL docs
 // https://dev.mysql.com/doc/mysql-backup-excerpt/5.7/en/backup-methods.html
 //
-//	Making Backups Using a File System Snapshot:
+// Making Backups Using a File System Snapshot:
 //
-//	If you are using a Veritas file system, you can make a backup like this:
+//  If you are using a Veritas file system, you can make a backup like this:
 //
-//	   	From a client program, execute FLUSH TABLES WITH READ LOCK.
+//    	From a client program, execute FLUSH TABLES WITH READ LOCK.
-//	   	From another shell, execute mount vxfs snapshot.
+//    	From another shell, execute mount vxfs snapshot.
-//	   	From the first client, execute UNLOCK TABLES.
+//    	From the first client, execute UNLOCK TABLES.
-//	   	Copy files from the snapshot.
+//    	Copy files from the snapshot.
-//	   	Unmount the snapshot.
+//    	Unmount the snapshot.
 //
-//		Similar snapshot capabilities may be available in other file systems, such as LVM or ZFS.
+//	Similar snapshot capabilities may be available in other file systems, such as LVM or ZFS.
-//
-
 type MySQLLockTables struct {
 	errIsFatal  bool
 	connector   sqldriver.Connector

daemon/job/snapjob.go

@@ -138,7 +138,7 @@ outer:
 // TODO:
 // This is a work-around for the current package daemon/pruner
 // and package pruning.Snapshot limitation: they require the
-// `Replicated` getter method be present, but obviously,
+//  `Replicated` getter method be present, but obviously,
 // a local job like SnapJob can't deliver on that.
 // But the pruner.Pruner gives up on an FS if no replication
 // cursor is present, which is why this pruner returns the

daemon/logging/trace/trace.go

@@ -1,6 +1,6 @@
 // package trace provides activity tracing via ctx through Tasks and Spans
 //
-// # Basic Concepts
+// Basic Concepts
 //
 // Tracing can be used to identify where a piece of code spends its time.
 //
@@ -10,50 +10,51 @@
 // to tech-savvy users (albeit not developers).
 //
 // This package provides the concept of Tasks and Spans to express what activity is happening within an application:
-// - Neither task nor span is really tangible but instead contained within the context.Context tree
+//
-// - Tasks represent concurrent activity (i.e. goroutines).
+//  - Neither task nor span is really tangible but instead contained within the context.Context tree
-// - Spans represent a semantic stack trace within a task.
+//  - Tasks represent concurrent activity (i.e. goroutines).
+//  - Spans represent a semantic stack trace within a task.
+//
 // As a consequence, whenever a context is propagated across goroutine boundary, you need to create a child task:
 //
-//	go func(ctx context.Context) {
+//   go func(ctx context.Context) {
-//	  ctx, endTask = WithTask(ctx, "what-happens-inside-the-child-task")
+//     ctx, endTask = WithTask(ctx, "what-happens-inside-the-child-task")
-//	  defer endTask()
+//     defer endTask()
-//	  // ...
+//     // ...
-//	}(ctx)
+//   }(ctx)
 //
 // Within the task, you can open up a hierarchy of spans.
 // In contrast to tasks, which have can multiple concurrently running child tasks,
 // spans must nest and not cross the goroutine boundary.
 //
-//	ctx, endSpan = WithSpan(ctx, "copy-dir")
+//  ctx, endSpan = WithSpan(ctx, "copy-dir")
-//	defer endSpan()
+//  defer endSpan()
-//	for _, f := range dir.Files() {
+//  for _, f := range dir.Files() {
-//	  func() {
+//    func() {
-//	    ctx, endSpan := WithSpan(ctx, fmt.Sprintf("copy-file %q", f))
+//      ctx, endSpan := WithSpan(ctx, fmt.Sprintf("copy-file %q", f))
-//	    defer endspan()
+//      defer endspan()
-//	    b, _ := ioutil.ReadFile(f)
+//      b, _ := ioutil.ReadFile(f)
-//	    _ = ioutil.WriteFile(f + ".copy", b, 0600)
+//      _ = ioutil.WriteFile(f + ".copy", b, 0600)
-//	  }()
+//    }()
-//	}
+//  }
 //
 // In combination:
-//
+//  ctx, endTask = WithTask(ctx, "copy-dirs")
-//	ctx, endTask = WithTask(ctx, "copy-dirs")
+//  defer endTask()
-//	defer endTask()
+//  for i := range dirs {
-//	for i := range dirs {
+//    go func(dir string) {
-//	  go func(dir string) {
+//      ctx, endTask := WithTask(ctx, "copy-dir")
-//	    ctx, endTask := WithTask(ctx, "copy-dir")
+//      defer endTask()
-//	    defer endTask()
+//      for _, f := range filesIn(dir) {
-//	    for _, f := range filesIn(dir) {
+//        func() {
-//	      func() {
+//          ctx, endSpan := WithSpan(ctx, fmt.Sprintf("copy-file %q", f))
-//	        ctx, endSpan := WithSpan(ctx, fmt.Sprintf("copy-file %q", f))
+//          defer endspan()
-//	        defer endspan()
+//          b, _ := ioutil.ReadFile(f)
-//	        b, _ := ioutil.ReadFile(f)
+//          _ = ioutil.WriteFile(f + ".copy", b, 0600)
-//	        _ = ioutil.WriteFile(f + ".copy", b, 0600)
+//        }()
-//	      }()
+//      }
-//	    }
+//    }()
-//	  }()
+//  }
-//	}
 //
 // Note that a span ends at the time you call endSpan - not before and not after that.
 // If you violate the stack-like nesting of spans by forgetting an endSpan() invocation,
@@ -64,7 +65,8 @@
 //
 // Recovering from endSpan() or endTask() panics will corrupt the trace stack and lead to corrupt tracefile output.
 //
-// # Best Practices For Naming Tasks And Spans
+//
+// Best Practices For Naming Tasks And Spans
 //
 // Tasks should always have string constants as names, and must not contain the `#` character. WHy?
 // First, the visualization by chrome://tracing draws a horizontal bar for each task in the trace.
@@ -72,7 +74,8 @@
 // Note that the `#NUM` suffix will be reused if a task has ended, in order to avoid an
 // infinite number of horizontal bars in the visualization.
 //
-// # Chrome-compatible Tracefile Support
+//
+// Chrome-compatible Tracefile Support
 //
 // The activity trace generated by usage of WithTask and WithSpan can be rendered to a JSON output file
 // that can be loaded into chrome://tracing .

daemon/logging/trace/trace_convenience.go

@@ -11,6 +11,8 @@ import (
 // use like this:
 //
 //	defer WithSpanFromStackUpdateCtx(&existingCtx)()
+//
+//
 func WithSpanFromStackUpdateCtx(ctx *context.Context) DoneFunc {
 	childSpanCtx, end := WithSpan(*ctx, getMyCallerOrPanic())
 	*ctx = childSpanCtx

daemon/snapper/cron.go

@@ -10,7 +10,6 @@ import (
 
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/hooks"
-	"github.com/zrepl/zrepl/util/suspendresumesafetimer"
 	"github.com/zrepl/zrepl/zfs"
 )
 
@@ -21,9 +20,8 @@ func cronFromConfig(fsf zfs.DatasetFilter, in config.SnapshottingCron) (*Cron, e
 		return nil, errors.Wrap(err, "hook config error")
 	}
 	planArgs := planArgs{
-		prefix:          in.Prefix,
+		prefix: in.Prefix,
-		timestampFormat: in.TimestampFormat,
+		hooks:  hooksList,
-		hooks:           hooksList,
 	}
 	return &Cron{config: in, fsf: fsf, planArgs: planArgs}, nil
 }
@@ -44,45 +42,67 @@ type Cron struct {
 
 func (s *Cron) Run(ctx context.Context, snapshotsTaken chan<- struct{}) {
 
+	t := time.NewTimer(0)
+	defer func() {
+		if !t.Stop() {
+			select {
+			case <-t.C:
+			default:
+			}
+		}
+	}()
 	for {
 		now := time.Now()
 		s.mtx.Lock()
 		s.wakeupTime = s.config.Cron.Schedule.Next(now)
 		s.mtx.Unlock()
 
-		ctxDone := suspendresumesafetimer.SleepUntil(ctx, s.wakeupTime)
+		// Re-arm the timer.
-		if ctxDone != nil {
+		// Need to Stop before Reset, see docs.
-			return
+		if !t.Stop() {
-		}
+			// Use non-blocking read from timer channel
-
+			// because, except for the first loop iteration,
-		getLogger(ctx).Debug("cron timer fired")
+			// the channel is already drained
-		s.mtx.Lock()
-		if s.running {
-			getLogger(ctx).Warn("snapshotting triggered according to cron rules but previous snapshotting is not done; not taking a snapshot this time")
-			s.wakeupWhileRunningCount++
-			s.mtx.Unlock()
-			continue
-		}
-		s.lastError = nil
-		s.lastPlan = nil
-		s.wakeupWhileRunningCount = 0
-		s.running = true
-		s.mtx.Unlock()
-		go func() {
-			err := s.do(ctx)
-			s.mtx.Lock()
-			s.lastError = err
-			s.running = false
-			s.mtx.Unlock()
-
 			select {
-			case snapshotsTaken <- struct{}{}:
+			case <-t.C:
 			default:
-				if snapshotsTaken != nil {
-					getLogger(ctx).Warn("callback channel is full, discarding snapshot update event")
-				}
 			}
-		}()
+		}
+		t.Reset(s.wakeupTime.Sub(now))
+
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			getLogger(ctx).Debug("cron timer fired")
+			s.mtx.Lock()
+			if s.running {
+				getLogger(ctx).Warn("snapshotting triggered according to cron rules but previous snapshotting is not done; not taking a snapshot this time")
+				s.wakeupWhileRunningCount++
+				s.mtx.Unlock()
+				continue
+			}
+			s.lastError = nil
+			s.lastPlan = nil
+			s.wakeupWhileRunningCount = 0
+			s.running = true
+			s.mtx.Unlock()
+			go func() {
+				err := s.do(ctx)
+				s.mtx.Lock()
+				s.lastError = err
+				s.running = false
+				s.mtx.Unlock()
+
+				select {
+				case snapshotsTaken <- struct{}{}:
+				default:
+					if snapshotsTaken != nil {
+						getLogger(ctx).Warn("callback channel is full, discarding snapshot update event")
+					}
+				}
+			}()
+		}
 	}
 
 }

daemon/snapper/impl.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"sort"
-	"strconv"
 	"strings"
 	"time"
 
@@ -15,9 +14,8 @@ import (
 )
 
 type planArgs struct {
-	prefix          string
+	prefix string
-	timestampFormat string
+	hooks  *hooks.List
-	hooks           *hooks.List
 }
 
 type plan struct {
@@ -60,21 +58,6 @@ type snapProgress struct {
 	runResults hooks.PlanReport
 }
 
-func (plan *plan) formatNow(format string) string {
-	now := time.Now().UTC()
-	switch strings.ToLower(format) {
-	case "dense":
-		format = "20060102_150405_000"
-	case "human":
-		format = "2006-01-02_15:04:05"
-	case "iso-8601":
-		format = "2006-01-02T15:04:05.000Z"
-	case "unix-seconds":
-		return strconv.FormatInt(now.Unix(), 10)
-	}
-	return now.Format(format)
-}
-
 func (plan *plan) execute(ctx context.Context, dryRun bool) (ok bool) {
 
 	hookMatchCount := make(map[hooks.Hook]int, len(*plan.args.hooks))
@@ -85,7 +68,7 @@ func (plan *plan) execute(ctx context.Context, dryRun bool) (ok bool) {
 	anyFsHadErr := false
 	// TODO channel programs -> allow a little jitter?
 	for fs, progress := range plan.snaps {
-		suffix := plan.formatNow(plan.args.timestampFormat)
+		suffix := time.Now().In(time.UTC).Format("20060102_150405_000")
 		snapname := fmt.Sprintf("%s%s", plan.args.prefix, suffix)
 
 		ctx := logging.WithInjectedField(ctx, "fs", fs.ToString())

daemon/snapper/periodic.go

@@ -15,7 +15,6 @@ import (
 	"github.com/zrepl/zrepl/daemon/hooks"
 	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/util/envconst"
-	"github.com/zrepl/zrepl/util/suspendresumesafetimer"
 	"github.com/zrepl/zrepl/zfs"
 )
 
@@ -36,9 +35,8 @@ func periodicFromConfig(g *config.Global, fsf zfs.DatasetFilter, in *config.Snap
 		interval: in.Interval.Duration(),
 		fsf:      fsf,
 		planArgs: planArgs{
-			prefix:          in.Prefix,
+			prefix: in.Prefix,
-			timestampFormat: in.TimestampFormat,
+			hooks:  hookList,
-			hooks:           hookList,
 		},
 		// ctx and log is set in Run()
 	}
@@ -173,13 +171,16 @@ func periodicStateSyncUp(a periodicArgs, u updater) state {
 	u(func(s *Periodic) {
 		s.sleepUntil = syncPoint
 	})
-	ctxDone := suspendresumesafetimer.SleepUntil(a.ctx, syncPoint)
+	t := time.NewTimer(time.Until(syncPoint))
-	if ctxDone != nil {
+	defer t.Stop()
+	select {
+	case <-t.C:
+		return u(func(s *Periodic) {
+			s.state = Planning
+		}).sf()
+	case <-a.ctx.Done():
 		return onMainCtxDone(a.ctx, u)
 	}
-	return u(func(s *Periodic) {
-		s.state = Planning
-	}).sf()
 }
 
 func periodicStatePlan(a periodicArgs, u updater) state {
@@ -240,13 +241,17 @@ func periodicStateWait(a periodicArgs, u updater) state {
 		logFunc("enter wait-state after error")
 	})
 
-	ctxDone := suspendresumesafetimer.SleepUntil(a.ctx, sleepUntil)
+	t := time.NewTimer(time.Until(sleepUntil))
-	if ctxDone != nil {
+	defer t.Stop()
+
+	select {
+	case <-t.C:
+		return u(func(snapper *Periodic) {
+			snapper.state = Planning
+		}).sf()
+	case <-a.ctx.Done():
 		return onMainCtxDone(a.ctx, u)
 	}
-	return u(func(snapper *Periodic) {
-		snapper.state = Planning
-	}).sf()
 }
 
 func listFSes(ctx context.Context, mf zfs.DatasetFilter) (fss []*zfs.DatasetPath, err error) {

dist/grafana/grafana-prometheus-zrepl.json

@@ -1235,7 +1235,7 @@
         ]
     },
     "timezone": "",
-    "title": "zrepl",
+    "title": "zrepl 0.3",
     "uid": "etQuvBnGz",
-    "version": 8
+    "version": 7
 }

dist/systemd/zrepl.service

@@ -9,9 +9,6 @@ ExecStart=/usr/local/bin/zrepl --config /etc/zrepl/zrepl.yml daemon
 RuntimeDirectory=zrepl zrepl/stdinserver
 RuntimeDirectoryMode=0700
 
-# Make Go produce coredumps
-Environment=GOTRACEBACK='crash'
-
 ProtectSystem=strict
 #PrivateDevices=yes # TODO ZFS needs access to /dev/zfs, could we limit this?
 ProtectKernelTunables=yes
@@ -30,8 +27,7 @@ ProtectHome=read-only
 # SystemCallFilter
 #   ~@privileged doesn't work with Ubuntu 18.04 ssh
 SystemCallFilter=~ @mount @cpu-emulation @keyring @module @obsolete @raw-io @debug @clock @resources
-# Go1.19 added automatic RLIMIT_NOFILE changes, so, we need to allow that
+
-SystemCallFilter= setrlimit
 
 [Install]
 WantedBy=multi-user.target

docs/changelog.rst

@@ -16,42 +16,13 @@ Changelog
 The changelog summarizes bugfixes that are deemed relevant for users and package maintainers.
 Developers should consult the git commit log or GitHub issue tracker.
 
-Next Release
+0.6 (Unreleased)
-------------
+----------------
 
-The plan for the next release is to revisit how zrepl does snapshot management.
+* `Feature Wishlist on GitHub <https://github.com/zrepl/zrepl/discussions/547>`_
-High-level goals:
-
-- Make it easy to decouple snapshot management (snapshotting, pruning) from replication.
-- Ability to include/exclude snapshots from replication.
-  This is useful for aforementioned decoupling, e.g., separate snapshot prefixes for local & remote replication.
-  Also, it makes explicit that by default, zrepl replicates all snapshots, and that
-  replication has no concept of "zrepl-created snapshots", which is a common misconception.
-- Use of ``zfs snapshot`` comma syntax or channel programs to take snapshots of multiple
-  datasets atomically.
-- Provide an alternative to the ``grid`` pruning policy.
-  Most likely something based on hourly/daily/weekly/monthly "trains" plus a count.
-- Ability to prune at the granularity of the **group** of snapshots created at a given
-  time, as opposed to the individual snapshots within a dataset.
-  Maybe this will be addressed by the alternative to the ``grid`` pruning policy,
-  as it will likely be more predictable.
-
-Those changes will likely come with some breakage in the config.
-However, I want to avoid breaking **use cases** that are satisfied by the current design.
-There will be beta/RC releases to give users a chance to evaluate.
-
-0.6
----
 
 * |feature| :ref:`Schedule-based snapshotting<job-snapshotting--cron>` using ``cron`` syntax instead of an interval.
-* |feature| Configurable initial replication policy.
+* |feature| Add ``ZREPL_DESTROY_MAX_BATCH_SIZE`` env var (default 0=unlimited).
-  When a filesystem is first replicated to a receiver, this control whether just the newest
-  snapshot will be replicated vs. all existing snapshots. Learn more :ref:`in the docs <conflict_resolution-initial_replication>`.
-* |feature| Configurable timestamp format for snapshot names via :ref:`timestamp_format<job-snapshotting-timestamp_format>`
-  (Thanks, `@ydylla <https://github.com/ydylla>`_).
-* |feature| Add ``ZREPL_DESTROY_MAX_BATCH_SIZE`` env var (default 0=unlimited)
-  (Thanks, `@3nprob <https://github.com/3nprob>`_).
-* |feature| Add ``zrepl configcheck --skip-cert-check`` flag (Thanks, `@cole-h <https://github.com/cole-h>`_).
 * |bugfix| Fix resuming from interrupted replications that use ``send.raw`` on unencrypted datasets.
 
   * The send options introduced in zrepl 0.4 allow users to specify additional zfs send flags for zrepl to use.
@@ -63,7 +34,7 @@ There will be beta/RC releases to give users a chance to evaluate.
   * However, this means that the ``zrepl status`` UI no longer indicates whether a replication step uses encrypted sends or not.
     The setting is still effective though.
 
-* |break| convert Prometheus metric ``zrepl_version_daemon`` to ``zrepl_start_time`` metric
+* |break| |feature| convert Prometheus metric ``zrepl_version_daemon`` to ``zrepl_start_time`` metric
 
   * The metric still reports the zrepl version in a label.
     But the metric *value* is now the Unix timestamp at the time the daemon was started.
@@ -72,13 +43,6 @@ There will be beta/RC releases to give users a chance to evaluate.
 * |bugfix| transient zrepl status error: ``Post "http://unix/status": EOF``
 * |bugfix| don't treat receive-side bookmarks as a replication conflict.
   This facilitates chaining of replication jobs. See :issue:`490`.
-* |bugfix| workaround for Go/gRPC problem on Illumos where zrepl would
-  crash when using the ``local`` transport type (:issue:`598`).
-* |bugfix| fix active child tasks panic that cold occur during replication plannig (:issue:`193abbe`)
-* |bugfix| ``zrepl status`` off-by-one error in display of completed step count (:commit:`ce6701f`)
-* |bugfix| Allow using day & week units for ``snapshotting.interval`` (:commit:`ffb1d89`)
-* |docs| ``docs/overview`` improvements (Thanks, `@jtagcat <https://github.com/jtagcat>`_).
-* |maint| Update to Go 1.19.
 
 0.5
 ---

docs/configuration/conflict_resolution.rst

@@ -15,7 +15,7 @@ Conflict Resolution Options
 
      ...
 
-.. _conflict_resolution-initial_replication:
+.. _conflict_resolution-initial_replication-option-send_all_snapshots:
 
 
 ``initial_replication`` option

docs/configuration/snapshotting.rst

@@ -62,9 +62,6 @@ The ``periodic`` and ``cron`` snapshotting types share some common options and b
        type: periodic
        prefix: zrepl_
        interval: 10m
-       # Timestamp format that is used as snapshot suffix.
-       # Can be any of "dense" (default), "human", "iso-8601", "unix-seconds" or a custom Go time format (see https://go.dev/src/time/format.go)
-       timestamp_format: dense
        hooks: ...
     pruning: ...
 
@@ -94,9 +91,6 @@ The snapshotter uses the ``prefix`` to identify which snapshots it created.
        # (second, optional) minute hour day-of-month month day-of-week
        # This example takes snapshots daily at 3:00.
        cron: "0 3 * * *"
-       # Timestamp format that is used as snapshot suffix.
-       # Can be any of "dense" (default), "human", "iso-8601", "unix-seconds" or a custom Go time format (see https://go.dev/src/time/format.go)
-       timestamp_format: dense
      pruning: ...
 
 In ``cron`` mode, the snapshotter takes snaphots at fixed points in time.
@@ -104,21 +98,6 @@ See https://en.wikipedia.org/wiki/Cron for details on the syntax.
 zrepl uses the ``the github.com/robfig/cron/v3`` Go package for parsing.
 An optional field for "seconds" is supported to take snapshots at sub-minute frequencies.
 
-.. _job-snapshotting-timestamp_format:
-
-Timestamp Format
-~~~~~~~~~~~~~~~~
-
-The ``cron`` and ``periodic`` snapshotter support configuring a custom timestamp format that is used as suffix for the snapshot name.
-It can be used by setting ``timestamp_format`` to any of the following values:
-
-* ``dense`` (default) looks like ``20060102_150405_000``
-* ``human`` looks like ``2006-01-02_15:04:05``
-* ``iso-8601`` looks like ``2006-01-02T15:04:05.000Z``
-* ``unix-seconds`` looks like ``1136214245``
-* Any custom Go time format accepted by `time.Time#Format <https://go.dev/src/time/format.go>`_.
-
-
 ``manual`` Snapshotting
 -----------------------
 

endpoint/endpoint.go

@@ -439,7 +439,7 @@ type FSMap interface { // FIXME unused
 }
 
 // NOTE: when adding members to this struct, remember
-// to add them to `ReceiverConfig.copyIn()`
+//       to add them to `ReceiverConfig.copyIn()`
 type ReceiverConfig struct {
 	JobID JobID
 

endpoint/endpoint_zfs_abstraction.go

@@ -168,7 +168,7 @@ func (s AbstractionTypeSet) String() string {
 	for i := range s {
 		sts = append(sts, string(i))
 	}
-	sort.Strings(sts)
+	sts = sort.StringSlice(sts)
 	return strings.Join(sts, ",")
 }
 

endpoint/endpoint_zfs_abstraction_replication_cursor.go

@@ -23,7 +23,7 @@ func replicationCursorBookmarkNameImpl(fs string, guid uint64, jobid string) (st
 
 var ErrV1ReplicationCursor = fmt.Errorf("bookmark name is a v1-replication cursor")
 
-// err != nil always means that the bookmark is not a valid replication bookmark
+//err != nil always means that the bookmark is not a valid replication bookmark
 //
 // Returns ErrV1ReplicationCursor as error if the bookmark is a v1 replication cursor
 func ParseReplicationCursorBookmarkName(fullname string) (uint64, JobID, error) {

replication/driver/replication_driver.go

@@ -797,7 +797,7 @@ func (a *attempt) errorReport() *errorReport {
 			r.byClass[class] = errs
 		}
 		for _, err := range r.flattened {
-			if neterr, ok := err.Err.(net.Error); ok && neterr.Timeout() {
+			if neterr, ok := err.Err.(net.Error); ok && neterr.Temporary() {
 				putClass(err, errorClassTemporaryConnectivityRelated)
 				continue
 			}

replication/driver/replication_driver_debug.go

@@ -13,7 +13,7 @@ func init() {
 	}
 }
 
-//nolint:deadcode,unused
+//nolint[:deadcode,unused]
 func debug(format string, args ...interface{}) {
 	if debugEnabled {
 		fmt.Fprintf(os.Stderr, "repl: driver: %s\n", fmt.Sprintf(format, args...))
@@ -22,7 +22,7 @@ func debug(format string, args ...interface{}) {
 
 type debugFunc func(format string, args ...interface{})
 
-//nolint:deadcode,unused
+//nolint[:deadcode,unused]
 func debugPrefix(prefixFormat string, prefixFormatArgs ...interface{}) debugFunc {
 	prefix := fmt.Sprintf(prefixFormat, prefixFormatArgs...)
 	return func(format string, args ...interface{}) {

rpc/dataconn/dataconn_debug.go

@@ -14,7 +14,7 @@ func init() {
 	}
 }
 
-//nolint:deadcode,unused
+//nolint[:deadcode,unused]
 func debug(format string, args ...interface{}) {
 	if debugEnabled {
 		fmt.Fprintf(os.Stderr, "rpc/dataconn: %s\n", fmt.Sprintf(format, args...))

rpc/dataconn/heartbeatconn/heartbeatconn.go

@@ -24,9 +24,6 @@ func (e HeartbeatTimeout) Error() string {
 	return "heartbeat timeout"
 }
 
-// This function is deprecated in net.Error and since this
-// function is not involved in .Accept() code path, nothing
-// really needs this method to be here.
 func (e HeartbeatTimeout) Temporary() bool { return true }
 
 func (e HeartbeatTimeout) Timeout() bool { return true }

rpc/dataconn/heartbeatconn/heartbeatconn_debug.go

@@ -13,7 +13,7 @@ func init() {
 	}
 }
 
-//nolint:deadcode,unused
+//nolint[:deadcode,unused]
 func debug(format string, args ...interface{}) {
 	if debugEnabled {
 		fmt.Fprintf(os.Stderr, "rpc/dataconn/heartbeatconn: %s\n", fmt.Sprintf(format, args...))

rpc/dataconn/heartbeatconn/integration_test_variablereceiverate/heartbeatconn_integration_variablereceiverate.go

@@ -6,12 +6,12 @@
 // In commit 082335df5d85e1b0b9faa35ff182c71886142d3e and earlier, heartbeatconn would fail
 // this benchmark with a writev I/O timeout (here the ss(8) output at the time of failure)
 //
-//	 ESTAB        33369    0                                        127.0.0.1:12345                                                127.0.0.1:57282                    users:(("heartbeatconn_i",pid=25953,fd=5))
+//  ESTAB        33369    0                                        127.0.0.1:12345                                                127.0.0.1:57282                    users:(("heartbeatconn_i",pid=25953,fd=5))
-//		 cubic wscale:7,7 rto:203 rtt:2.992/5.849 ato:162 mss:32768 pmtu:65535 rcvmss:32741 advmss:65483 cwnd:10 bytes_sent:48 bytes_acked:48 bytes_received:195401 segs_out:44 segs_in:57 data_segs_out:6 data_segs_in:34 send 876.1Mbps lastsnd:125 lastrcv:9390 lastack:125 pacing_rate 1752.0Mbps delivery_rate 6393.8Mbps delivered:7 app_limited busy:42ms rcv_rtt:1 rcv_space:65483 rcv_ssthresh:65483 minrtt:0.029
+//	 cubic wscale:7,7 rto:203 rtt:2.992/5.849 ato:162 mss:32768 pmtu:65535 rcvmss:32741 advmss:65483 cwnd:10 bytes_sent:48 bytes_acked:48 bytes_received:195401 segs_out:44 segs_in:57 data_segs_out:6 data_segs_in:34 send 876.1Mbps lastsnd:125 lastrcv:9390 lastack:125 pacing_rate 1752.0Mbps delivery_rate 6393.8Mbps delivered:7 app_limited busy:42ms rcv_rtt:1 rcv_space:65483 rcv_ssthresh:65483 minrtt:0.029
-//		 --
+//	 --
-//		 ESTAB        0        3956805                                  127.0.0.1:57282                                                127.0.0.1:12345                    users:(("heartbeatconn_i",pid=26100,fd=3))
+//	 ESTAB        0        3956805                                  127.0.0.1:57282                                                127.0.0.1:12345                    users:(("heartbeatconn_i",pid=26100,fd=3))
-//			  cubic wscale:7,7 rto:211 backoff:5 rtt:10.38/16.937 ato:40 mss:32768 pmtu:65535 rcvmss:536 advmss:65483 cwnd:10 bytes_sent:195401 bytes_acked:195402 bytes_received:48 segs_out:57 segs_in:45 data_segs_out:34 data_segs_in:6 send 252.5Mbps lastsnd:9390 lastrcv:125 lastack:125 pacing_rate 505.1Mbps delivery_rate 1971.0Mbps delivered:35 busy:30127ms rwnd_limited:30086ms(99.9%) rcv_space:65495 rcv_ssthresh:65495 notsent:3956805 minrtt:0.007
+//		  cubic wscale:7,7 rto:211 backoff:5 rtt:10.38/16.937 ato:40 mss:32768 pmtu:65535 rcvmss:536 advmss:65483 cwnd:10 bytes_sent:195401 bytes_acked:195402 bytes_received:48 segs_out:57 segs_in:45 data_segs_out:34 data_segs_in:6 send 252.5Mbps lastsnd:9390 lastrcv:125 lastack:125 pacing_rate 505.1Mbps delivery_rate 1971.0Mbps delivered:35 busy:30127ms rwnd_limited:30086ms(99.9%) rcv_space:65495 rcv_ssthresh:65495 notsent:3956805 minrtt:0.007
-//		 panic: writev tcp 127.0.0.1:57282->127.0.0.1:12345: i/o timeout
+//	 panic: writev tcp 127.0.0.1:57282->127.0.0.1:12345: i/o timeout
 //
 // The assumed reason for those writev timeouts is the following:
 // - Sporadic server stalls (sever data handling, usually I/O) cause TCP exponential backoff on the client for client->server
@@ -22,23 +22,27 @@
 // The fix contained in the commit this message was committed with resets the deadline whenever
 // a heartbeat is received from the server.
 //
+//
 // How to run this integration test:
 //
-//	Terminal 1:
-//	   $ ZREPL_RPC_DATACONN_HEARTBEATCONN_DEBUG=1 go run heartbeatconn_integration_variablereceiverate.go  -mode server -addr 127.0.0.1:12345
-//	   rpc/dataconn/heartbeatconn: send heartbeat
-//	   rpc/dataconn/heartbeatconn: send heartbeat
-//	   ...
 //
-//	Terminal 2:
+//  Terminal 1:
-//	   $ ZREPL_RPC_DATACONN_HEARTBEATCONN_DEBUG=1 go run heartbeatconn_integration_variablereceiverate.go  -mode  client -addr 127.0.0.1:12345
+//     $ ZREPL_RPC_DATACONN_HEARTBEATCONN_DEBUG=1 go run heartbeatconn_integration_variablereceiverate.go  -mode server -addr 127.0.0.1:12345
-//	   rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout
+//     rpc/dataconn/heartbeatconn: send heartbeat
-//	   rpc/dataconn/heartbeatconn: renew frameconn write timeout returned errT=<nil> err=%!s(<nil>)
+//     rpc/dataconn/heartbeatconn: send heartbeat
-//	   rpc/dataconn/heartbeatconn: send heartbeat
+//     ...
-//	   rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout
+//
-//	   rpc/dataconn/heartbeatconn: renew frameconn write timeout returned errT=<nil> err=%!s(<nil>)
+//  Terminal 2:
-//	   rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout
+//     $ ZREPL_RPC_DATACONN_HEARTBEATCONN_DEBUG=1 go run heartbeatconn_integration_variablereceiverate.go  -mode  client -addr 127.0.0.1:12345
-//	   ...
+//     rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout
+//     rpc/dataconn/heartbeatconn: renew frameconn write timeout returned errT=<nil> err=%!s(<nil>)
+//     rpc/dataconn/heartbeatconn: send heartbeat
+//     rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout
+//     rpc/dataconn/heartbeatconn: renew frameconn write timeout returned errT=<nil> err=%!s(<nil>)
+//     rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout
+//     ...
+//
+//  You should observe
 package main
 
 import (

rpc/dataconn/microbenchmark/microbenchmark.go

@@ -2,13 +2,15 @@
 //
 // With stdin / stdout on client and server, simulating zfs send|recv piping
 //
-//	./microbenchmark -appmode server  | pv -r > /dev/null
+//   ./microbenchmark -appmode server  | pv -r > /dev/null
-//	./microbenchmark -appmode client -direction recv < /dev/zero
+//   ./microbenchmark -appmode client -direction recv < /dev/zero
+//
 //
 // Without the overhead of pipes (just protocol performance, mostly useful with perf bc no bw measurement)
 //
-//	./microbenchmark -appmode client -direction recv -devnoopWriter -devnoopReader
+//   ./microbenchmark -appmode client -direction recv -devnoopWriter -devnoopReader
-//	./microbenchmark -appmode server -devnoopReader -devnoopWriter
+//   ./microbenchmark -appmode server -devnoopReader -devnoopWriter
+//
 package main
 
 import (

rpc/dataconn/stream/stream.go

@@ -29,7 +29,7 @@ func WithLogger(ctx context.Context, log Logger) context.Context {
 	return context.WithValue(ctx, contextKeyLogger, log)
 }
 
-//nolint:deadcode,unused
+//nolint[:deadcode,unused]
 func getLog(ctx context.Context) Logger {
 	log, ok := ctx.Value(contextKeyLogger).(Logger)
 	if !ok {
@@ -181,19 +181,23 @@ func (e *ReadStreamError) Error() string {
 
 var _ net.Error = &ReadStreamError{}
 
-func (e ReadStreamError) Timeout() bool {
+func (e ReadStreamError) netErr() net.Error {
 	if netErr, ok := e.Err.(net.Error); ok {
+		return netErr
+	}
+	return nil
+}
+
+func (e ReadStreamError) Timeout() bool {
+	if netErr := e.netErr(); netErr != nil {
 		return netErr.Timeout()
 	}
 	return false
 }
 
-// This function is deprecated in net.Error and since this
-// function is not involved in .Accept() code path, nothing
-// really needs this method to be here.
 func (e ReadStreamError) Temporary() bool {
-	if te, ok := e.Err.(interface{ Temporary() bool }); ok {
+	if netErr := e.netErr(); netErr != nil {
-		return te.Temporary()
+		return netErr.Temporary()
 	}
 	return false
 }

rpc/dataconn/stream/stream_conn.go

@@ -232,12 +232,7 @@ var _ net.Error = (*closeStateErrConnectionClosed)(nil)
 func (e *closeStateErrConnectionClosed) Error() string {
 	return "connection closed"
 }
-
+func (e *closeStateErrConnectionClosed) Timeout() bool   { return false }
-func (e *closeStateErrConnectionClosed) Timeout() bool { return false }
-
-// This function is deprecated in net.Error and since this
-// function is not involved in .Accept() code path, nothing
-// really needs this method to be here.
 func (e *closeStateErrConnectionClosed) Temporary() bool { return false }
 
 func (s *closeState) CloseEntry() error {

rpc/dataconn/stream/stream_debug.go

@@ -13,7 +13,7 @@ func init() {
 	}
 }
 
-//nolint:deadcode,unused
+//nolint[:deadcode,unused]
 func debug(format string, args ...interface{}) {
 	if debugEnabled {
 		fmt.Fprintf(os.Stderr, "rpc/dataconn/stream: %s\n", fmt.Sprintf(format, args...))

rpc/netadaptor/authlistener_netlistener_adaptor.go

@@ -1,5 +1,25 @@
 // Package netadaptor implements an adaptor from
 // transport.AuthenticatedListener to net.Listener.
+//
+// In contrast to transport.AuthenticatedListener,
+// net.Listener is commonly expected (e.g. by net/http.Server.Serve),
+// to return errors that fulfill the Temporary interface:
+//   interface Temporary { Temporary() bool }
+// Common behavior of packages consuming net.Listener is to return
+// from the serve-loop if an error returned by Accept is not Temporary,
+// i.e., does not implement the interface or is !net.Error.Temporary().
+//
+// The zrepl transport infrastructure was written with the
+// idea that Accept() may return any kind of error, and the consumer
+// would just log the error and continue calling Accept().
+// We have to adapt these listeners' behavior to the expectations
+// of net/http.Server.
+//
+// Hence, Listener does not return an error at all but blocks the
+// caller of Accept() until we get a (successfully authenticated)
+// connection without errors from the transport.
+// Accept errors returned from the transport are logged as errors
+// to the logger passed on initialization.
 package netadaptor
 
 import (

rpc/rpc_client.go

@@ -188,8 +188,8 @@ func (c *Client) WaitForConnectivity(ctx context.Context) error {
 			data, dataErr := c.dataClient.ReqPing(ctx, &req)
 			// dataClient uses transport.Connecter, which doesn't expose WaitForReady(true)
 			// => we need to mask dial timeouts
-			if err, ok := dataErr.(interface{ Timeout() bool }); ok && err.Timeout() {
+			if err, ok := dataErr.(interface{ Temporary() bool }); ok && err.Temporary() {
-				// Rate-limit pings here in case Timeout() == true is a mis-classification
+				// Rate-limit pings here in case Temporary() is a mis-classification
 				// or returns immediately (this is a tight loop in that case)
 				// TODO keep this in lockstep with controlClient
 				// 		=> don't use FailFast for control, but check that both control and data worked

rpc/rpc_debug.go

@@ -14,7 +14,7 @@ func init() {
 	}
 }
 
-//nolint:deadcode,unused
+//nolint[:deadcode,unused]
 func debug(format string, args ...interface{}) {
 	if debugEnabled {
 		fmt.Fprintf(os.Stderr, "rpc: %s\n", fmt.Sprintf(format, args...))

rpc/rpc_doc.go

@@ -3,7 +3,7 @@
 // The zrepl documentation refers to the client as the
 // `active side` and to the server as the `passive side`.
 //
-// # Design Considerations
+// Design Considerations
 //
 // zrepl has non-standard requirements to remote procedure calls (RPC):
 // whereas the coordination of replication (the planning phase) mostly
@@ -35,7 +35,7 @@
 //
 // Hence, this package attempts to combine the best of both worlds:
 //
-// # GRPC for Coordination and Dataconn for Bulk Data Transfer
+// GRPC for Coordination and Dataconn for Bulk Data Transfer
 //
 // This package's Client uses its transport.Connecter to maintain
 // separate control and data connections to the Server.
@@ -47,66 +47,68 @@
 // The following ASCII diagram gives an overview of how the individual
 // building blocks are glued together:
 //
-//	                 +------------+
+//                    +------------+
-//	                 | rpc.Client |
+//                    | rpc.Client |
-//	                 +------------+
+//                    +------------+
-//	                   |         |
+//                      |         |
-//	          +--------+         +------------+
+//             +--------+         +------------+
-//	          |                               |
+//             |                               |
-//	+---------v-----------+          +--------v------+
+//   +---------v-----------+          +--------v------+
-//	|pdu.ReplicationClient|          |dataconn.Client|
+//   |pdu.ReplicationClient|          |dataconn.Client|
-//	+---------------------+          +--------v------+
+//   +---------------------+          +--------v------+
-//	          | label:            label:      |
+//             | label:            label:      |
-//	          | zrepl_control      zrepl_data |
+//             | zrepl_control      zrepl_data |
-//	          +--------+         +------------+
+//             +--------+         +------------+
-//	                   |         |
+//                      |         |
-//	                +--v---------v---+
+//                   +--v---------v---+
-//	                |  transportmux  |
+//                   |  transportmux  |
-//	                +-------+--------+
+//                   +-------+--------+
-//	                        | uses
+//                           | uses
-//	                +-------v--------+
+//                   +-------v--------+
-//	                |versionhandshake|
+//                   |versionhandshake|
-//	                +-------+--------+
+//                   +-------+--------+
-//	                        | uses
+//                           | uses
-//	                 +------v------+
+//                    +------v------+
-//	                 |  transport  |
+//                    |  transport  |
-//	                 +------+------+
+//                    +------+------+
-//	                        |
+//                           |
-//	                     NETWORK
+//                        NETWORK
-//	                        |
+//                           |
-//	                 +------+------+
+//                    +------+------+
-//	                 |  transport  |
+//                    |  transport  |
-//	                 +------^------+
+//                    +------^------+
-//	                        | uses
+//                           | uses
-//	                +-------+--------+
+//                   +-------+--------+
-//	                |versionhandshake|
+//                   |versionhandshake|
-//	                +-------^--------+
+//                   +-------^--------+
-//	                        | uses
+//                           | uses
-//	                +-------+--------+
+//                   +-------+--------+
-//	                |  transportmux  |
+//                   |  transportmux  |
-//	                +--^--------^----+
+//                   +--^--------^----+
-//	                   |        |
+//                      |        |
-//	          +--------+        --------------+       ---
+//             +--------+        --------------+       ---
-//	          |                               |        |
+//             |                               |        |
-//	          | label:            label:      |        |
+//             | label:            label:      |        |
-//	          | zrepl_control      zrepl_data |        |
+//             | zrepl_control      zrepl_data |        |
-//	    +-----+----+              +-----------+---+    |
+//       +-----+----+              +-----------+---+    |
-//	    |netadaptor|              |dataconn.Server|    | rpc.Server
+//       |netadaptor|              |dataconn.Server|    | rpc.Server
-//	    |     +    |              +------+--------+    |
+//       |     +    |              +------+--------+    |
-//	    |grpcclient|                     |             |
+//       |grpcclient|                     |             |
-//	    |identity  |                     |             |
+//       |identity  |                     |             |
-//	    +-----+----+                     |             |
+//       +-----+----+                     |             |
-//	          |                          |             |
+//             |                          |             |
-//	+---------v-----------+              |             |
+//   +---------v-----------+              |             |
-//	|pdu.ReplicationServer|              |             |
+//   |pdu.ReplicationServer|              |             |
-//	+---------+-----------+              |             |
+//   +---------+-----------+              |             |
-//	          |                          |            ---
+//             |                          |            ---
-//	          +----------+  +------------+
+//             +----------+  +------------+
-//	                     |  |
+//                        |  |
-//	                 +---v--v-----+
+//                    +---v--v-----+
-//	                 |  Handler   |
+//                    |  Handler   |
-//	                 +------------+
+//                    +------------+
-//	        (usually endpoint.{Sender,Receiver})
+//           (usually endpoint.{Sender,Receiver})
+//
+//
 package rpc
 
 // edit trick for the ASCII art above:

rpc/rpc_logging.go

@@ -9,7 +9,7 @@ import (
 
 type Logger = logger.Logger
 
-// All fields must be non-nil
+/// All fields must be non-nil
 type Loggers struct {
 	General Logger
 	Control Logger

rpc/versionhandshake/versionhandshake.go

@@ -33,47 +33,8 @@ var _ net.Error = &HandshakeError{}
 
 func (e HandshakeError) Error() string { return e.msg }
 
-// When a net.Listener.Accept() returns an error, the server must
+// Like with net.OpErr (Go issue 6163), a client failing to handshake
-// decide whether to retry calling Accept() or not.
+// should be a temporary Accept error toward the Listener .
-// On some platforms (e.g., Linux), Accept() can return errors
-// related to the specific protocol connection that was supposed
-// to be returned as asocket FD. Obviously, we want to ignore,
-// maybe log, those errors and retry Accept() immediately to
-// serve other connections.
-// But there are also conditions where we get Accept() errors because
-// the process has run out of file descriptors. In that case, retrying
-// won't help. We need to close some file descriptor to make progress.
-// Note that there could be lots of open file descriptors because we
-// have accepted, and not yet closed, lots of connections in the past.
-// And then, of course there can be errors where we just want
-// to return, e.g., if there's a programming error and we're getting
-// an EBADFD or whatever.
-//
-// So, the serve loops in net/http.Server.Serve() or gRPC's server.Serve()
-// must inspect the error and decide what to do.
-// The vehicle for this is the
-//
-//	interface { Temporary() bool }
-//
-// Behavior in both of the aforementioned Serve() loops:
-//
-//   - if the error doesn't implement the interface, stop serving and return
-//   - `Temporary() == true`: retry with back-off
-//   - `Temporary() == false`: stop serving and return
-//
-// So, to make this package's HandshakeListener work with these
-// Serve() loops, we return Temporary() == true if the handshake fails.
-// In the aforementioned categories, that's the case of a per-connection
-// protocol error.
-//
-// Note: the net.Error interface has deprecated the Temporary() method
-// in go.dev/issue/45729, but there is no replacement for users of .Accept().
-// Existing users of .Accept() continue to check for the interface.
-// So, we need to continue supporting Temporary() until there's a different
-// mechanism for serve loops to decide whether to retry or not.
-// The following mailing list post proposes to eliminate the retries
-// completely, but it seems like the effort has stalled.
-// https://groups.google.com/g/golang-nuts/c/-JcZzOkyqYI/m/xwaZzjCgAwAJ
 func (e HandshakeError) Temporary() bool {
 	if e.isAcceptError {
 		return true

util/chainlock/chainlock.go

@@ -3,12 +3,13 @@
 //
 // Intended Usage
 //
-//	  defer s.lock().unlock()
+//   defer s.lock().unlock()
-//	  // drop lock while waiting for wait group
+//   // drop lock while waiting for wait group
-//	  func() {
+//   func() {
-//		     defer a.l.Unlock().Lock()
+//	     defer a.l.Unlock().Lock()
-//		     fssesDone.Wait()
+//	     fssesDone.Wait()
-//		 }()
+//	 }()
+//
 package chainlock
 
 import "sync"

util/nodefault/nodefault.go

@@ -7,25 +7,26 @@
 //
 // Example:
 //
-//	type Config struct {
+//  type Config struct {
-//	    // This field must be set to a non-nil value,
+//      // This field must be set to a non-nil value,
-//	    // forcing the caller to make their mind up
+//      // forcing the caller to make their mind up
-//	    // about this field.
+//      // about this field.
-//	    CriticalSetting   *nodefault.Bool
+//      CriticalSetting   *nodefault.Bool
-//	}
+//  }
 //
 // An function that takes such a Config should _not_ check for nil-ness:
 // and instead unconditionally dereference:
 //
-//	func f(c Config) {
+//  func f(c Config) {
-//	    if (c.CriticalSetting) { }
+//      if (c.CriticalSetting) { }
-//	}
+//  }
 //
 // If the caller of f forgot to specify the .CriticalSetting
 // field, the Go runtime will issue a nil-pointer deref panic
 // and it'll be clear that the caller did not read the docs of Config.
 //
-//	f(Config{}) // crashes
+//  f(Config{}) // crashes
+//
+//  f Config{ CriticalSetting: &nodefault.Bool{B: false}} // doesn't crash
 //
-//	f Config{ CriticalSetting: &nodefault.Bool{B: false}} // doesn't crash
 package nodefault

util/suspendresumesafetimer/suspendresumesafetimer.go

@@ -1,96 +0,0 @@
-package suspendresumesafetimer
-
-import (
-	"context"
-	"time"
-
-	"github.com/zrepl/zrepl/util/envconst"
-)
-
-// The returned error is guaranteed to be the ctx.Err()
-func SleepUntil(ctx context.Context, sleepUntil time.Time) error {
-
-	// We use .Round(0) to strip the monotonic clock reading from the time.Time
-	// returned by time.Now(). That will make the before/after check in the ticker
-	// for-loop compare wall-clock times instead of monotonic time.
-	// Comparing wall clock time is necessary because monotonic time does not progress
-	// while the system is suspended.
-	//
-	// Background
-	//
-	// A time.Time carries a wallclock timestamp and optionally a monotonic clock timestamp.
-	// time.Now() returns a time.Time that carries both.
-	// time.Time.Add() applies the same delta to both timestamps in the time.Time.
-	// x.Sub(y) will return the *monotonic* delta if both x and y carry a monotonic timestamp.
-	// time.Until(x) == x.Sub(now) where `now` will have a monotonic timestamp.
-	// So, time.Until(x) with an `x` that has monotonic timestamp will return monotonic delta.
-	//
-	// Why Do We Care?
-	//
-	// On systems that suspend/resume, wall clock time progresses during suspend but
-	// monotonic time does not.
-	//
-	// So, suppose the following sequence of events:
-	//   x <== time.Now()
-	//   System suspends for 1 hour
-	//   delta <== time.Now().Sub(x)
-	// `delta` will be near 0 because time.Until() subtracts the monotonic
-	// timestamps, and monotonic time didn't progress during suspend.
-	//
-	// Now strip the timestamp using .Round(0)
-	//   x <== time.Now().Round(0)
-	//   System suspends for 1 hour
-	//   delta <== time.Now().Sub(x)
-	// `delta` will be 1 hour because time.Sub() subtracted wallclock timestamps
-	// because x didn't have a monotonic timestamp because we stripped it using .Round(0).
-	//
-	//
-	sleepUntil = sleepUntil.Round(0)
-
-	// Set up a timer so that, if the system doesn't suspend/resume,
-	// we get a precise wake-up time from the native Go timer.
-	monotonicClockTimer := time.NewTimer(time.Until(sleepUntil))
-	defer func() {
-		if !monotonicClockTimer.Stop() {
-			// non-blocking read since we can come here when
-			// we've already drained the channel through
-			//		case <-monotonicClockTimer.C
-			// in the `for` loop below.
-			select {
-			case <-monotonicClockTimer.C:
-			default:
-			}
-		}
-	}()
-
-	// Set up a ticker so that we're guaranteed to wake up periodically.
-	// We'll then get the current wall-clock time and check ourselves
-	// whether we're past the requested expiration time.
-	// Pick a 10 second check interval by default since it's rare that
-	// suspend/resume is done more frequently.
-	ticker := time.NewTicker(envconst.Duration("ZREPL_WALLCLOCKTIMER_MAX_DELAY", 10*time.Second))
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-monotonicClockTimer.C:
-			return nil
-		case <-ticker.C:
-			now := time.Now()
-			if now.Before(sleepUntil) {
-				// Continue waiting.
-
-				// Reset the monotonic timer to reset drift.
-				if !monotonicClockTimer.Stop() {
-					<-monotonicClockTimer.C
-				}
-				monotonicClockTimer.Reset(time.Until(sleepUntil))
-
-				continue
-			}
-			return nil
-		case <-ctx.Done():
-			return ctx.Err()
-		}
-	}
-}

zfs/namecheck.go

@@ -53,6 +53,7 @@ var componentValidChar = regexp.MustCompile(`^[0-9a-zA-Z-_\.: ]+$`)
 // characters:
 //
 //	[-_.: ]
+//
 func ComponentNamecheck(datasetPathComponent string) error {
 	if len(datasetPathComponent) == 0 {
 		return fmt.Errorf("path component must not be empty")
@@ -90,8 +91,8 @@ func (e *PathValidationError) Error() string {
 
 // combines
 //
-//	lib/libzfs/libzfs_dataset.c: zfs_validate_name
+//    lib/libzfs/libzfs_dataset.c: zfs_validate_name
-//	module/zcommon/zfs_namecheck.c: entity_namecheck
+//    module/zcommon/zfs_namecheck.c: entity_namecheck
 //
 // The '%' character is not allowed because it's reserved for zfs-internal use
 func EntityNamecheck(path string, t EntityType) (err *PathValidationError) {

zfs/zfs.go

@@ -343,6 +343,7 @@ const (
 
 type SendStream struct {
 	cmd          *zfscmd.Cmd
+	kill         context.CancelFunc
 	stdoutReader io.ReadCloser // not *os.File for mocking during platformtest
 	stderrBuf    *circlog.CircularLog
 
@@ -415,29 +416,8 @@ func (s *SendStream) killAndWait() error {
 	debug("sendStream: killAndWait enter")
 	defer debug("sendStream: killAndWait leave")
 
-	// ensure this function is only called once
-	if s.state != sendStreamOpen {
-		panic(s.state)
-	}
-
 	// send SIGKILL
-	// In an earlier version, we used the starting context.Context's cancel function
+	s.kill()
-	// for this, but in Go > 1.19, doing so will cause .Wait() to return the
-	// context cancel error instead of the *exec.ExitError.
-	err := s.cmd.Process().Kill()
-	if err != nil {
-		if err == os.ErrProcessDone {
-			// This can happen if
-			// (1) the process has already been .Wait()ed, or
-			// (2) some other goroutine cancels the context, likely further up
-			//     the context tree.
-			// Case (1) can't happen to us because we only call
-			// this function in sendStreamOpen state.
-			// In Case (2), it's still our job to .Wait(), so, fallthrough.
-		} else {
-			return err
-		}
-	}
 
 	// Close our read-end of the pipe.
 	//
@@ -983,10 +963,10 @@ func ZFSSend(ctx context.Context, sendArgs ZFSSendArgsValidated) (*SendStream, e
 
 	stream := &SendStream{
 		cmd:          cmd,
+		kill:         cancel,
 		stdoutReader: stdoutReader,
 		stderrBuf:    stderrBuf,
 	}
-	_ = cancel // the SendStream.killAndWait() will kill the process
 
 	return stream, nil
 }
@@ -1042,10 +1022,8 @@ func (s *DrySendInfo) unmarshalZFSOutput(output []byte) (err error) {
 }
 
 // unmarshal info line, looks like this:
-//
+//   full	zroot/test/a@1	5389768
-//	full	zroot/test/a@1	5389768
+//   incremental	zroot/test/a@1	zroot/test/a@2	5383936
-//	incremental	zroot/test/a@1	zroot/test/a@2	5383936
-//
 // => see test cases
 func (s *DrySendInfo) unmarshalInfoLine(l string) (regexMatched bool, err error) {
 
@@ -1877,6 +1855,7 @@ var ErrBookmarkCloningNotSupported = fmt.Errorf("bookmark cloning feature is not
 // unless a bookmark with the name `bookmark` exists and has the same idenitty (zfs.FilesystemVersionEqualIdentity)
 //
 // v must be validated by the caller
+//
 func ZFSBookmark(ctx context.Context, fs string, v FilesystemVersion, bookmark string) (bm FilesystemVersion, err error) {
 
 	bm = FilesystemVersion{

zfs/zfs_debug.go

@@ -13,7 +13,7 @@ func init() {
 	}
 }
 
-//nolint:deadcode,unused
+//nolint[:deadcode,unused]
 func debug(format string, args ...interface{}) {
 	if debugEnabled {
 		fmt.Fprintf(os.Stderr, "zfs: %s\n", fmt.Sprintf(format, args...))