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

This commit was motivated by https://github.com/zrepl/zrepl/issues/495
where, on FreeBSD with OpenZFS 2.0, a SendStream.Close() call might wait indefinitely for `zfs send` to exit.
The reason is that, due to the refactoring done for redacted send & recv
(https://github.com/openzfs/zfs/commit/30af21b02569ac192f52ce6e6511015f8a8d5729),
the `dump_bytes` function, which writes to the pipe, executes in a separate thread (synctask taskq) iff not `HAVE_LARGE_STACKS`.
The `zfs send` process/thread waits for that taskq thread using an uninterruptible primitive.
So when we SIGKILL `zfs send`, that signal doesn't reach the right thread to interrupt the pipe write.

Theoretically this affects both Linux and FreeBSD, but most Linux users `HAVE_LARGE_STACKS` and since https://github.com/penzfs/zfs/pull/12350/files OpenZFS on FreeBSD `HAVE_LARGE_STACKS` as well.
However, at least until FreeBSD 13.1, possibly for the entire 13 lifecycle, we're going to have to live with that oddity.

Measures taken in this commit:
- Report the behavior as an upstream bug https://github.com/openzfs/zfs/issues/12500
- Change SendStream code so that it closes zrepl's read-end of the pipe (see comment in code)
- Clean up and make explicit SendStream's state handling
- Write extensive platformtests for SendStream
    - They pass on my Linux install and on FreeBSD 12
    - FreeBSD 13 still needs testing.

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

.circleci/config.yml

@@ -358,7 +358,7 @@ jobs:
           subcommand: docdep
       - run:
           command: |
-            git config --global user.email "me@cschwarz.com"
+            git config --global user.email "zreplbot@cschwarz.com"
             git config --global user.name "zrepl-github-io-ci"
 
       # https://circleci.com/docs/2.0/add-ssh-key/#adding-multiple-keys-with-blank-hostnames

client/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_global_test.go

@@ -70,9 +70,9 @@ func TestPrometheusMonitoring(t *testing.T) {
 global:
   monitoring:
     - type: prometheus
-      listen: ':9091'
+      listen: ':9811'
 `)
-	assert.Equal(t, ":9091", conf.Global.Monitoring[0].Ret.(*PrometheusMonitoring).Listen)
+	assert.Equal(t, ":9811", conf.Global.Monitoring[0].Ret.(*PrometheusMonitoring).Listen)
 }
 
 func TestSyslogLoggingOutletFacility(t *testing.T) {

config/samples/quickstart_backup_to_external_disk.yml

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

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

platformtest/tests/generated_cases.go

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

platformtest/tests/recvRollback.go

@@ -40,6 +40,7 @@ func ReceiveForceRollbackWorksUnencrypted(ctx *platformtest.Context) {
 
 	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
 	require.NoError(ctx, err)
+	defer sendStream.Close()
 
 	recvOpts := zfs.RecvOptions{
 		RollbackAndForceRecv: true,

platformtest/tests/replication.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"path"
 	"sort"
+	"strings"
 	"time"
 
 	"github.com/kr/pretty"
@@ -90,6 +91,7 @@ func (i replicationInvocation) Do(ctx *platformtest.Context) *report.Report {
 		ReplicationConfig: &pdu.ReplicationConfig{
 			Protection: i.guarantee,
 		},
+		SizeEstimationConcurrency: 1,
 	}
 
 	report, wait := replication.Do(
@@ -1054,6 +1056,9 @@ func ReplicationPropertyReplicationWorks(ctx *platformtest.Context) {
 		rep := fsByName[fs]
 		require.Len(ctx, rep.Steps, 1)
 		require.Nil(ctx, rep.PlanError)
+		if rep.StepError != nil && strings.Contains(rep.StepError.Error(), "invalid option 'x'") {
+			ctx.SkipNow() // XXX feature detection
+		}
 		require.Nil(ctx, rep.StepError)
 		require.Len(ctx, rep.Steps, 1)
 		require.Equal(ctx, 1, rep.CurrentStep)

platformtest/tests/sendStream.go

@@ -0,0 +1,186 @@
+package tests
+
+import (
+	"fmt"
+	"io"
+	"io/ioutil"
+	"os"
+	"os/exec"
+	"path"
+	"syscall"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sys/unix"
+
+	"github.com/zrepl/zrepl/platformtest"
+	"github.com/zrepl/zrepl/util/nodefault"
+	"github.com/zrepl/zrepl/zfs"
+)
+
+func sendStreamTest(ctx *platformtest.Context) *zfs.SendStream {
+
+	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+		DESTROYROOT
+		CREATEROOT
+		+  "sender"
+	`)
+
+	fs := fmt.Sprintf("%s/sender", ctx.RootDataset)
+
+	fsmpo, err := zfs.ZFSGetMountpoint(ctx, fs)
+	require.NoError(ctx, err)
+
+	writeDummyData(path.Join(fsmpo.Mountpoint, "dummy.data"), 1<<26)
+	mustSnapshot(ctx, fs+"@1")
+	snap := fsversion(ctx, fs, "@1")
+	snapSendArg := snap.ToSendArgVersion()
+
+	sendArgs, err := zfs.ZFSSendArgsUnvalidated{
+		FS:   fs,
+		From: nil,
+		To:   &snapSendArg,
+		ZFSSendFlags: zfs.ZFSSendFlags{
+			Encrypted: &nodefault.Bool{B: false},
+		},
+	}.Validate(ctx)
+	require.NoError(ctx, err)
+
+	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
+	require.NoError(ctx, err)
+
+	return sendStream
+
+}
+
+func SendStreamCloseAfterBlockedOnPipeWrite(ctx *platformtest.Context) {
+
+	sendStream := sendStreamTest(ctx)
+
+	// let the pipe buffer fill and the zfs process block uninterruptibly
+
+	ctx.Logf("waiting for pipe write to block")
+	time.Sleep(5 * time.Second) // XXX need a platform-neutral way to detect that the pipe is full and the writer is blocked
+
+	ctx.Logf("closing send stream")
+	err := sendStream.Close() // this is what this test case is about
+	ctx.Logf("close error: %T %s", err, err)
+	require.NoError(ctx, err)
+
+	exitErrZfsError := sendStream.TestOnly_ExitErr()
+	require.Contains(ctx, exitErrZfsError.Error(), "signal")
+	require.Error(ctx, exitErrZfsError)
+	exitErr, ok := exitErrZfsError.WaitErr.(*exec.ExitError)
+	require.True(ctx, ok)
+	if exitErr.Exited() {
+		// some ZFS impls (FreeBSD 12) behaves that way
+		return
+	}
+
+	// ProcessState is only available after exit
+	// => use as proxy that the process was wait()ed upon and is gone
+	ctx.Logf("%#v", exitErr.ProcessState)
+	require.NotNil(ctx, exitErr.ProcessState)
+	// and let's verify that the process got killed, so that we know it was the call to .Close() above
+	waitStatus := exitErr.ProcessState.Sys().(syscall.WaitStatus)
+	ctx.Logf("wait status: %#v", waitStatus)
+	ctx.Logf("exit status: %v", waitStatus.ExitStatus())
+	require.True(ctx, waitStatus.Signaled())
+	switch waitStatus.Signal() {
+	case unix.SIGKILL:
+		fallthrough
+	case unix.SIGPIPE:
+		// ok
+
+	default:
+		ctx.Errorf("%T %s\n%v", waitStatus.Signal(), waitStatus.Signal(), waitStatus.Signal())
+		ctx.FailNow()
+	}
+}
+
+func SendStreamCloseAfterEOFRead(ctx *platformtest.Context) {
+
+	sendStream := sendStreamTest(ctx)
+
+	_, err := io.Copy(ioutil.Discard, sendStream)
+	require.NoError(ctx, err)
+
+	var buf [128]byte
+	n, err := sendStream.Read(buf[:])
+	require.Zero(ctx, n)
+	require.Equal(ctx, io.EOF, err)
+
+	err = sendStream.Close()
+	require.NoError(ctx, err)
+
+	n, err = sendStream.Read(buf[:])
+	require.Zero(ctx, n)
+	require.Equal(ctx, os.ErrClosed, err, "same read error should be returned")
+}
+
+func SendStreamMultipleCloseAfterEOF(ctx *platformtest.Context) {
+
+	sendStream := sendStreamTest(ctx)
+
+	_, err := io.Copy(ioutil.Discard, sendStream)
+	require.NoError(ctx, err)
+
+	var buf [128]byte
+	n, err := sendStream.Read(buf[:])
+	require.Zero(ctx, n)
+	require.Equal(ctx, io.EOF, err)
+
+	err = sendStream.Close()
+	require.NoError(ctx, err)
+
+	err = sendStream.Close()
+	require.Equal(ctx, os.ErrClosed, err)
+}
+
+func SendStreamMultipleCloseBeforeEOF(ctx *platformtest.Context) {
+
+	sendStream := sendStreamTest(ctx)
+
+	err := sendStream.Close()
+	require.NoError(ctx, err)
+
+	err = sendStream.Close()
+	require.Equal(ctx, os.ErrClosed, err)
+}
+
+type failingReadCloser struct {
+	err error
+}
+
+var _ io.ReadCloser = &failingReadCloser{}
+
+func (c *failingReadCloser) Read(p []byte) (int, error) { return 0, c.err }
+func (c *failingReadCloser) Close() error               { return c.err }
+
+func SendStreamNonEOFReadErrorHandling(ctx *platformtest.Context) {
+
+	sendStream := sendStreamTest(ctx)
+
+	var buf [128]byte
+	n, err := sendStream.Read(buf[:])
+	require.Equal(ctx, len(buf), n)
+	require.NoError(ctx, err)
+
+	var mockError = fmt.Errorf("taeghaefow4piesahwahjocu7ul5tiachaiLipheijae8ooZ8Pies8shohGee9feeTeirai5aiFeiyaecai4kiaLoh4azeih0tea")
+	mock := &failingReadCloser{err: mockError}
+	orig := sendStream.TestOnly_ReplaceStdoutReader(mock)
+
+	n, err = sendStream.Read(buf[:])
+	require.Equal(ctx, 0, n)
+	require.Equal(ctx, mockError, err)
+
+	if sendStream.TestOnly_ReplaceStdoutReader(orig) != mock {
+		panic("incorrect test impl")
+	}
+
+	err = sendStream.Close()
+	require.NoError(ctx, err) // if we can't kill the child then this will be a flaky test, but let's assume we can kill the child
+
+	err = sendStream.Close()
+	require.Equal(ctx, os.ErrClosed, err)
+}

zfs/zfs.go

@@ -14,7 +14,6 @@ import (
 	"strconv"
 	"strings"
 	"sync"
-	"time"
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
@@ -334,60 +333,122 @@ func pipeWithCapacityHint(capacity int) (r, w *os.File, err error) {
 	return stdoutReader, stdoutWriter, nil
 }
 
-type SendStream struct {
-	cmd  *zfscmd.Cmd
-	kill context.CancelFunc
+type sendStreamState int
 
-	closeMtx     sync.Mutex
-	stdoutReader *os.File
+const (
+	sendStreamOpen sendStreamState = iota
+	sendStreamClosed
+)
+
+type SendStream struct {
+	cmd          *zfscmd.Cmd
+	kill         context.CancelFunc
+	stdoutReader io.ReadCloser // not *os.File for mocking during platformtest
 	stderrBuf    *circlog.CircularLog
-	opErr        error
+
+	mtx     sync.Mutex
+	state   sendStreamState
+	exitErr *ZFSError
 }
 
-func (s *SendStream) Read(p []byte) (n int, err error) {
-	s.closeMtx.Lock()
-	opErr := s.opErr
-	s.closeMtx.Unlock()
-	if opErr != nil {
-		return 0, opErr
-	}
+func (s *SendStream) Read(p []byte) (n int, _ error) {
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
 
-	n, err = s.stdoutReader.Read(p)
-	if err != nil {
-		debug("sendStream: read err: %T %s", err, err)
-		// TODO we assume here that any read error is permanent
-		// which is most likely the case for a local zfs send
-		kwerr := s.killAndWait(err)
-		debug("sendStream: killAndWait n=%v err= %T %s", n, kwerr, kwerr)
-		// TODO we assume here that any read error is permanent
-		return n, kwerr
+	switch s.state {
+	case sendStreamClosed:
+		return 0, os.ErrClosed
+
+	case sendStreamOpen:
+		n, readErr := s.stdoutReader.Read(p)
+		if readErr != nil {
+			debug("sendStream: read: readErr=%T %s", readErr, readErr)
+			if readErr == io.EOF {
+				// io.EOF must be bubbled up as is so that consumers can handle it properly.
+				return n, readErr
+			}
+			// Assume that the error is not retryable.
+			// Try to kill now so that we can return a nice *ZFSError with captured stderr.
+			// If the kill doesn't work, it doesn't matter because the caller must by contract call Close() anyways.
+			killErr := s.killAndWait()
+			debug("sendStream: read: killErr=%T %s", killErr, killErr)
+			if killErr == nil {
+				s.state = sendStreamClosed
+				return n, s.exitErr // return the nice error
+			} else {
+				// we remain open so that we retry
+				return n, readErr // return the normal error
+			}
+		}
+		return n, readErr
+
+	default:
+		panic("unreachable")
 	}
-	return n, err
 }
 
 func (s *SendStream) Close() error {
 	debug("sendStream: close called")
-	return s.killAndWait(nil)
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+
+	switch s.state {
+	case sendStreamOpen:
+		err := s.killAndWait()
+		if err != nil {
+			return err
+		} else {
+			s.state = sendStreamClosed
+			return nil
+		}
+	case sendStreamClosed:
+		return os.ErrClosed
+	default:
+		panic("unreachable")
+	}
 }
 
-func (s *SendStream) killAndWait(precedingReadErr error) error {
+// returns nil iff the child process is gone (has been successfully waited upon)
+// in that case, s.exitErr is set
+func (s *SendStream) killAndWait() error {
 
 	debug("sendStream: killAndWait enter")
 	defer debug("sendStream: killAndWait leave")
-	if precedingReadErr == io.EOF {
-		// give the zfs process a little bit of time to terminate itself
-		// if it holds this deadline, exitErr will be nil
-		time.AfterFunc(200*time.Millisecond, s.kill)
-	} else {
-		s.kill()
-	}
 
-	// allow async kills from Close(), that's why we only take the mutex here
-	s.closeMtx.Lock()
-	defer s.closeMtx.Unlock()
+	// send SIGKILL
+	s.kill()
 
-	if s.opErr != nil {
-		return s.opErr
+	// Close our read-end of the pipe.
+	//
+	// We must do this before .Wait() because in some (not all) versions/build configs of ZFS,
+	// `zfs send` uses a separate kernel thread (taskq) to write the send stream (function `dump_bytes`).
+	// The `zfs send` thread then waits uinterruptably for the taskq thread to finish the write.
+	// And signalling the `zfs send` thread doesn't propagate to the taskq thread.
+	// So we end up in a state where we .Wait() forever.
+	// (See https://github.com/openzfs/zfs/issues/12500 and
+	//  https://github.com/zrepl/zrepl/issues/495#issuecomment-902530043)
+	//
+	// By closing our read end of the pipe before .Wait(), we unblock the taskq thread if there is any.
+	// If there is no separate taskq thread, the SIGKILL to `zfs end` would suffice and be most precise,
+	// but due to the circumstances above, there is no other portable & robust way.
+	//
+	// However, the fallout from closing the pipe is that (in non-taskq builds) `zfs sends` will get a SIGPIPE.
+	// And on Linux, that SIGPIPE appears to win over the previously issued SIGKILL.
+	// And thus, on Linux, the `zfs send` will be killed by the default SIGPIPE handler.
+	// We can observe this in the WaitStatus below.
+	// This behavior is slightly annoying because the *exec.ExitError's message ("signal: broken pipe")
+	// isn't as clear as ("signal: killed").
+	// However, it seems like we just have to live with that. (covered by platformtest)
+	var closePipeErr error
+	if s.stdoutReader != nil {
+		closePipeErr = s.stdoutReader.Close()
+		if closePipeErr == nil {
+			// avoid double-closes in case waiting below doesn't work
+			// and someone attempts Close again
+			s.stdoutReader = nil
+		} else {
+			return closePipeErr
+		}
 	}
 
 	waitErr := s.cmd.Wait()
@@ -402,39 +463,30 @@ func (s *SendStream) killAndWait(precedingReadErr error) error {
 		}
 	}
 
-	// now, after we know the program exited do we close the pipe
-	var closePipeErr error
-	if s.stdoutReader != nil {
-		closePipeErr = s.stdoutReader.Close()
-		if closePipeErr == nil {
-			// avoid double-closes in case anything below doesn't work
-			// and someone calls Close again
-			s.stdoutReader = nil
-		} else {
-			return closePipeErr
-		}
-	}
+	// invariant: at this point, the child is gone and we cleaned up everything related to the SendStream
 
-	// we managed to tear things down, no let's give the user some pretty *ZFSError
 	if exitErr != nil {
-		s.opErr = &ZFSError{
+		// zfs send exited with an error or was killed by a signal.
+		s.exitErr = &ZFSError{
 			Stderr:  []byte(s.stderrBuf.String()),
 			WaitErr: exitErr,
 		}
 	} else {
-		s.opErr = precedingReadErr
+		// zfs send exited successfully (we know that since waitErr was either nil or wasn't an *exec.ExitError)
+		s.exitErr = nil
 	}
 
-	// detect the edge where we're called from s.Read
-	// after the pipe EOFed and zfs send exited without errors
-	// this is actually the "hot" / nice path
-	if exitErr == nil && precedingReadErr == io.EOF {
-		return precedingReadErr
-	}
-
-	return s.opErr
+	return nil
 }
 
+func (s *SendStream) TestOnly_ReplaceStdoutReader(f io.ReadCloser) (prev io.ReadCloser) {
+	prev = s.stdoutReader
+	s.stdoutReader = f
+	return prev
+}
+
+func (s *SendStream) TestOnly_ExitErr() *ZFSError { return s.exitErr }
+
 // NOTE: When updating this struct, make sure to update funcs Validate ValidateCorrespondsToResumeToken
 type ZFSSendArgVersion struct {
 	RelName string

zfs/zfscmd/zfscmd.go

@@ -209,3 +209,7 @@ func (c *Cmd) Runtime() time.Duration {
 	}
 	return c.waitReturnedAt.Sub(c.startedAt)
 }
+
+func (c *Cmd) TestOnly_ExecCmd() *exec.Cmd {
+	return c.cmd
+}