transport/ssh: bump go-netssh version to fix ssh client process leaks

fixes #322

config/samples/source.yml

@@ -5,7 +5,11 @@ jobs:
     type: tcp
     listen: "0.0.0.0:8888"
     clients: {
-      "192.168.122.123" : "client1"
+      "192.168.122.123" :               "mysql01",
+      "192.168.122.42" :                "mx01",
+      "2001:0db8:85a3::8a2e:0370:7334": "gateway",
+      "10.23.42.0/24":       "cluster-*",
+      "fde4:8dba:82e1::/64": "san-*",
     }
   filesystems: {
     "<": true,

docs/configuration/transports.rst

@@ -52,8 +52,15 @@ Serve
         listen: ":8888"
         listen_freebind: true # optional, default false
         clients: {
-          "192.168.122.123" : "mysql01"
+          "192.168.122.123" :               "mysql01",
-          "192.168.122.123" : "mx01"
+          "192.168.122.42" :                "mx01",
+          "2001:0db8:85a3::8a2e:0370:7334": "gateway",
+
+          # CIDR masks require a '*' in the client identity string
+          # that is expanded to the client's IP address
+
+          "10.23.42.0/24":       "cluster-*"
+          "fde4:8dba:82e1::/64": "san-*"
         }
       ...
 

docs/installation.rst

@@ -1,138 +1,19 @@
-.. _binary releases: https://github.com/zrepl/zrepl/releases
 
-.. _installation:
+.. _installation_toc:
 
+************
 Installation
-============
+************
 
 .. TIP::
 
     Note: check out the :ref:`tutorial` if you want a first impression of zrepl.
 
-User Privileges
+.. toctree::
----------------
 
-It is possible to run zrepl as an unprivileged user in combination with
+    installation/user-privileges
-`ZFS delegation <https://www.freebsd.org/doc/handbook/zfs-zfs-allow.html>`_.
+    installation/packages
-Also, there is the possibility to run it in a jail on FreeBSD by delegating a dataset to the jail.
+    installation/apt-repos
-However, until we get around documenting those setups, you will have to run zrepl as root or experiment yourself :)
+    installation/compile-from-source
-
+    installation/freebsd-jail-with-iocage
-Packages
+    installation/what-next
---------
-
-zrepl source releases are signed & tagged by the author in the git repository.
-Your OS vendor may provide binary packages of zrepl through the package manager.
-Additionally, `binary releases`_ are provided on GitHub.
-The following list may be incomplete, feel free to submit a PR with an update:
-
-.. list-table::
-    :header-rows: 1
-
-    * - OS / Distro
-      - Install Command
-      - Link
-    * - FreeBSD
-      - ``pkg install zrepl``
-      - `<https://www.freshports.org/sysutils/zrepl/>`_
-    * - MacOS
-      - ``brew install zrepl``
-      - Available on `homebrew <https://brew.sh>`_
-    * - Arch Linux
-      - ``yay install zrepl``
-      - Available on `AUR <https://aur.archlinux.org/packages/zrepl>`_
-    * - Fedora
-      - ``dnf install zrepl``
-      - Available on `COPR <https://copr.fedorainfracloud.org/coprs/poettlerric/zrepl/>`_
-    * - CentOS/RHEL
-      - ``yum install zrepl``
-      - Available on `COPR <https://copr.fedorainfracloud.org/coprs/poettlerric/zrepl/>`_
-    * - Debian + Ubuntu
-      - ``apt install zrepl``
-      - APT repository config :ref:`see below <installation-apt-repos>`
-    * - OmniOS
-      - ``pkg install zrepl``
-      - Available since `r151030 <https://pkg.omniosce.org/r151030/extra/en/search.shtml?token=zrepl&action=Search>`_
-    * - Void Linux
-      - ``xbps-install zrepl``
-      - Available since `a88a2a4 <https://github.com/void-linux/void-packages/commit/a88a2a4d7bf56072dadf61ab56b8424e39155890>`_
-    * - Others
-      -
-      - Use `binary releases`_ or build from source.
-
-.. _installation-apt-repos:
-
-Debian / Ubuntu APT repositories
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-
-We maintain APT repositories for Debian, Ubuntu and derivatives.
-The fingerprint of the signing key is ``E101 418F D3D6 FBCB 9D65  A62D 7086 99FC 5F2E BF16``.
-It is available at `<https://zrepl.cschwarz.com/apt/apt-key.asc>`_ .
-Please open an issue `in the packaging repository <https://github.com/zrepl/debian-binary-packaging>`_ if you encounter any issues with the repository.
-
-The following snippet configure the repository for your Debian or Ubuntu release:
-
-::
-
-    apt update && 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
-
-
-.. NOTE::
-
-   Until zrepl reaches 1.0, all APT repositories will be updated to the latest zrepl release immediately.
-   This includes breaking changes between zrepl versions.
-   Use ``apt-mark hold zrepl`` to prevent upgrades of zrepl.
-
-Compile From Source
-~~~~~~~~~~~~~~~~~~~
-
-Producing a release requires **Go 1.11** or newer and **Python 3** + **pip3** + ``docs/requirements.txt`` for the Sphinx documentation.
-A tutorial to install Go is available over at `golang.org <https://golang.org/doc/install>`_.
-Python and pip3 should probably be installed via your distro's package manager.
-
-Alternatively, you can use the Docker build process:
-it is used to produce the official zrepl `binary releases`_
-and serves as a reference for build dependencies and procedure:
-
-::
-
-    git clone https://github.com/zrepl/zrepl.git && \
-    cd zrepl && \
-    sudo docker build -t zrepl_build -f build.Dockerfile . && \
-    sudo docker run -it --rm \
-        -v "${PWD}:/src" \
-        --user "$(id -u):$(id -g)" \
-        zrepl_build make release
-
-Alternatively, you can install build dependencies on your local system and then build in your ``$GOPATH``:
-
-::
-
-    mkdir -p "${GOPATH}/src/github.com/zrepl/zrepl"
-    git clone https://github.com/zrepl/zrepl.git "${GOPATH}/src/github.com/zrepl/zrepl"
-    cd "${GOPATH}/src/github.com/zrepl/zrepl"
-    python3 -m venv3
-    source venv3/bin/activate
-    ./lazy.sh devsetup
-    make release
-
-The Python venv is used for the documentation build dependencies.
-If you just want to build the zrepl binary, leave it out and use `./lazy.sh godep` instead.
-Either way, all build results are located in the ``artifacts/`` directory.
-
-.. NOTE::
-
-    It is your job to install the appropriate binary in the zrepl users's ``$PATH``, e.g. ``/usr/local/bin/zrepl``.
-    Otherwise, the examples in the :ref:`tutorial` may need to be adjusted.
-
-What next?
-----------
-
-Read the :ref:`configuration chapter<configuration_toc>` and then continue with the :ref:`usage chapter<usage>`.
-
-**Reminder**: If you want a quick introduction, please read the :ref:`tutorial`.

docs/installation/apt-repos.rst

@@ -0,0 +1,29 @@
+
+.. _installation-apt-repos:
+
+Debian / Ubuntu APT repositories
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+We maintain APT repositories for Debian, Ubuntu and derivatives.
+The fingerprint of the signing key is ``E101 418F D3D6 FBCB 9D65  A62D 7086 99FC 5F2E BF16``.
+It is available at `<https://zrepl.cschwarz.com/apt/apt-key.asc>`_ .
+Please open an issue `in the packaging repository <https://github.com/zrepl/debian-binary-packaging>`_ if you encounter any issues with the repository.
+
+The following snippet configure the repository for your Debian or Ubuntu release:
+
+::
+
+    apt update && 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
+
+
+.. NOTE::
+
+   Until zrepl reaches 1.0, all APT repositories will be updated to the latest zrepl release immediately.
+   This includes breaking changes between zrepl versions.
+   Use ``apt-mark hold zrepl`` to prevent upgrades of zrepl.

docs/installation/compile-from-source.rst

@@ -0,0 +1,45 @@
+.. _binary releases: https://github.com/zrepl/zrepl/releases
+
+.. _installation-compile-from-source:
+
+Compile From Source
+~~~~~~~~~~~~~~~~~~~
+
+Producing a release requires **Go 1.11** or newer and **Python 3** + **pip3** + ``docs/requirements.txt`` for the Sphinx documentation.
+A tutorial to install Go is available over at `golang.org <https://golang.org/doc/install>`_.
+Python and pip3 should probably be installed via your distro's package manager.
+
+Alternatively, you can use the Docker build process:
+it is used to produce the official zrepl `binary releases`_
+and serves as a reference for build dependencies and procedure:
+
+::
+
+    git clone https://github.com/zrepl/zrepl.git && \
+    cd zrepl && \
+    sudo docker build -t zrepl_build -f build.Dockerfile . && \
+    sudo docker run -it --rm \
+        -v "${PWD}:/src" \
+        --user "$(id -u):$(id -g)" \
+        zrepl_build make release
+
+Alternatively, you can install build dependencies on your local system and then build in your ``$GOPATH``:
+
+::
+
+    mkdir -p "${GOPATH}/src/github.com/zrepl/zrepl"
+    git clone https://github.com/zrepl/zrepl.git "${GOPATH}/src/github.com/zrepl/zrepl"
+    cd "${GOPATH}/src/github.com/zrepl/zrepl"
+    python3 -m venv3
+    source venv3/bin/activate
+    ./lazy.sh devsetup
+    make release
+
+The Python venv is used for the documentation build dependencies.
+If you just want to build the zrepl binary, leave it out and use `./lazy.sh godep` instead.
+Either way, all build results are located in the ``artifacts/`` directory.
+
+.. NOTE::
+
+    It is your job to install the appropriate binary in the zrepl users's ``$PATH``, e.g. ``/usr/local/bin/zrepl``.
+    Otherwise, the examples in the :ref:`tutorial` may need to be adjusted.

docs/installation/freebsd-jail-with-iocage.rst

@@ -0,0 +1,140 @@
+.. include:: ../global.rst.inc
+
+.. _installation-freebsd-jail-with-iocage:
+
+FreeBSD Jail With iocage
+========================
+
+
+This tutorial shows how zrepl can be installed on FreeBSD, or FreeNAS in a jail using iocage.
+While this tutorial focuses on using iocage, much of the setup would be similar
+using a different jail manager.
+
+.. NOTE::
+
+   From a security perspective, just keep in mind that ``zfs send``/``recv`` was never designed with
+   jails in mind, an attacker could probably crash the receive-side kernel or worse induce stateful
+   damage to the receive-side pool if they were able to get access to the jail.
+
+   The jail doesn't provide security benefits, but only management ones.
+
+Requirements
+------------
+
+A dataset that will be delegated to the jail needs to be created if one does not already exist.
+For the tutorial ``tank/zrepl`` will be used.
+
+.. code-block:: bash
+
+   zfs create -o mountpoint=none tank/zrepl
+
+The only software requirements on the host system are ``iocage``, which can be installed
+from ports or packages.
+
+.. code-block:: bash
+
+   pkg install py37-iocage
+
+.. NOTE::
+
+   By default ``iocage`` will "activate" on first use which will set up some defaults such as
+   which pool will be used. To activate ``iocage`` manually the ``iocage activate`` command can be used.
+
+Jail Creation
+-------------
+
+There are two options for jail creation using FreeBSD.
+
+1. Manually set up the jail from scratch
+2. Create the jail using the ``zrepl`` plugin. On FreeNAS this is possible from the user interface using the community index.
+
+Manual Jail
+###########
+
+Create a jail, using the same release as the host, called ``zrepl`` that will be automatically started at boot.
+The jail will have ``tank/zrepl`` delegated into it.
+
+.. code-block:: bash
+
+   iocage create --release "$(freebsd-version -k | cut -d '-' -f '1,2')" --name zrepl \
+          boot=on nat=1 \
+          jail_zfs=on \
+          jail_zfs_dataset=zrepl \
+          jail_zfs_mountpoint='none'
+
+Enter the jail:
+
+.. code-block:: bash
+
+   iocage console zrepl
+
+Install ``zrepl``
+
+.. code-block:: bash
+
+   pkg update && pkg upgrade
+   pkg install zrepl
+
+Create the log file ``/var/log/zrepl.log``
+
+.. code-block:: bash
+
+   touch /var/log/zrepl.log && service newsyslog restart
+
+Tell syslogd to redirect facility local0 to the ``zrepl.log`` file:
+
+.. code-block:: bash
+
+   service syslogd reload
+
+Enable the zrepl daemon to start automatically at boot:
+
+.. code-block:: bash
+
+   sysrc zrepl_enable="YES"
+
+
+Plugin
+######
+
+When using the plugin, ``zrepl`` will be installed for you in a jail using the following ``iocage`` properties.
+
+* ``nat=1``
+* ``jail_zfs=on``
+* ``jail_zfs_mountpoint=none``
+
+Additionally the delegated dataset should be specified upon creation, and optionally start on boot can be set.
+This can also be done from the FreeNAS webui.
+
+.. code-block:: bash
+
+   fetch https://raw.githubusercontent.com/ix-plugin-hub/iocage-plugin-index/master/zrepl.json -o /tmp/zrepl.json
+   iocage fetch -P /tmp/zrepl.json --name zrepl jail_zfs_dataset=zrepl boot=on
+
+Configuration
+-------------
+
+Now ``zrepl`` can be configured.
+
+Enter the jail.
+
+.. code-block:: bash
+
+   iocage console zrepl
+
+Modify the ``/usr/local/etc/zrepl/zrepl.yml`` configuration file.
+
+.. TIP::
+
+    Note: check out the :ref:`tutorial` for examples of a ``sink`` job.
+
+Now ``zrepl`` can be started.
+
+.. code-block:: bash
+
+   service zrepl start
+
+Summary
+-------
+
+Congratulations, you have a working jail!

docs/installation/packages.rst

@@ -0,0 +1,50 @@
+.. _installation-packages:
+
+.. _binary releases: https://github.com/zrepl/zrepl/releases
+
+Packages
+--------
+
+zrepl source releases are signed & tagged by the author in the git repository.
+Your OS vendor may provide binary packages of zrepl through the package manager.
+Additionally, `binary releases`_ are provided on GitHub.
+The following list may be incomplete, feel free to submit a PR with an update:
+
+.. list-table::
+    :header-rows: 1
+
+    * - OS / Distro
+      - Install Command
+      - Link
+    * - FreeBSD
+      - ``pkg install zrepl``
+      - `<https://www.freshports.org/sysutils/zrepl/>`_
+      
+        :ref:`installation-freebsd-jail-with-iocage`
+    * - FreeNAS
+      -
+      - :ref:`installation-freebsd-jail-with-iocage`
+    * - MacOS
+      - ``brew install zrepl``
+      - Available on `homebrew <https://brew.sh>`_
+    * - Arch Linux
+      - ``yay install zrepl``
+      - Available on `AUR <https://aur.archlinux.org/packages/zrepl>`_
+    * - Fedora
+      - ``dnf install zrepl``
+      - Available on `COPR <https://copr.fedorainfracloud.org/coprs/poettlerric/zrepl/>`_
+    * - CentOS/RHEL
+      - ``yum install zrepl``
+      - Available on `COPR <https://copr.fedorainfracloud.org/coprs/poettlerric/zrepl/>`_
+    * - Debian + Ubuntu
+      - ``apt install zrepl``
+      - APT repository config :ref:`see below <installation-apt-repos>`
+    * - OmniOS
+      - ``pkg install zrepl``
+      - Available since `r151030 <https://pkg.omniosce.org/r151030/extra/en/search.shtml?token=zrepl&action=Search>`_
+    * - Void Linux
+      - ``xbps-install zrepl``
+      - Available since `a88a2a4 <https://github.com/void-linux/void-packages/commit/a88a2a4d7bf56072dadf61ab56b8424e39155890>`_
+    * - Others
+      -
+      - Use `binary releases`_ or build from source.

docs/installation/user-privileges.rst

@@ -0,0 +1,12 @@
+.. _installation-user-privileges:
+
+User Privileges
+---------------
+
+It is possible to run zrepl as an unprivileged user in combination with
+`ZFS delegation <https://www.freebsd.org/doc/handbook/zfs-zfs-allow.html>`_.
+Also, there is the possibility to run it in a jail on FreeBSD by delegating a dataset to the jail.
+
+.. TIP::
+
+    Note: check out the :ref:`installation-freebsd-jail-with-iocage` for FreeBSD jail setup instructions.

docs/installation/what-next.rst

@@ -0,0 +1,8 @@
+.. _installation-what-next:
+
+What next?
+----------
+
+Read the :ref:`configuration chapter<configuration_toc>` and then continue with the :ref:`usage chapter<usage>`.
+
+**Reminder**: If you want a quick introduction, please read the :ref:`tutorial`.

docs/tutorial.rst

@@ -47,7 +47,7 @@ We can model this situation as two jobs:
 Install zrepl
 -------------
 
-Follow the :ref:`OS-specific installation instructions <installation>` and come back here.
+Follow the :ref:`OS-specific installation instructions <installation_toc>` and come back here.
 
 Generate TLS Certificates
 -------------------------

endpoint/endpoint.go

@@ -2,17 +2,14 @@
 package endpoint
 
 import (
-	"bytes"
 	"context"
 	"fmt"
-	"io"
 	"path"
 	"sync"
 
 	"github.com/pkg/errors"
 
 	"github.com/zrepl/zrepl/replication/logic/pdu"
-	"github.com/zrepl/zrepl/util/chainedio"
 	"github.com/zrepl/zrepl/util/chainlock"
 	"github.com/zrepl/zrepl/util/envconst"
 	"github.com/zrepl/zrepl/util/semaphore"
@@ -244,7 +241,7 @@ func sendArgsFromPDUAndValidateExistsAndGetVersion(ctx context.Context, fs strin
 	return version, nil
 }
 
-func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
+func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
 
 	_, err := s.filterCheckFS(r.Filesystem)
 	if err != nil {
@@ -342,11 +339,11 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
 
 	// step holds & replication cursor released / moved forward in s.SendCompleted => s.moveCursorAndReleaseSendHolds
 
-	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
+	streamCopier, err := zfs.ZFSSend(ctx, sendArgs)
 	if err != nil {
 		return nil, nil, errors.Wrap(err, "zfs send failed")
 	}
-	return res, sendStream, nil
+	return res, streamCopier, nil
 }
 
 func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
@@ -479,7 +476,7 @@ func (p *Sender) ReplicationCursor(ctx context.Context, req *pdu.ReplicationCurs
 	return &pdu.ReplicationCursorRes{Result: &pdu.ReplicationCursorRes_Guid{Guid: cursor.Guid}}, nil
 }
 
-func (p *Sender) Receive(ctx context.Context, r *pdu.ReceiveReq, _ io.ReadCloser) (*pdu.ReceiveRes, error) {
+func (p *Sender) Receive(ctx context.Context, r *pdu.ReceiveReq, receive zfs.StreamCopier) (*pdu.ReceiveRes, error) {
 	return nil, fmt.Errorf("sender does not implement Receive()")
 }
 
@@ -683,13 +680,13 @@ func (s *Receiver) ReplicationCursor(context.Context, *pdu.ReplicationCursorReq)
 	return nil, fmt.Errorf("ReplicationCursor not implemented for Receiver")
 }
 
-func (s *Receiver) Send(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
+func (s *Receiver) Send(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
 	return nil, nil, fmt.Errorf("receiver does not implement Send()")
 }
 
 var maxConcurrentZFSRecvSemaphore = semaphore.New(envconst.Int64("ZREPL_ENDPOINT_MAX_CONCURRENT_RECV", 10))
 
-func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.ReadCloser) (*pdu.ReceiveRes, error) {
+func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs.StreamCopier) (*pdu.ReceiveRes, error) {
 	getLogger(ctx).Debug("incoming Receive")
 	defer receive.Close()
 
@@ -768,30 +765,21 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.
 		return nil, visitErr
 	}
 
-	log := getLogger(ctx).WithField("proto_fs", req.GetFilesystem()).WithField("local_fs", lp.ToString())
-
 	// determine whether we need to rollback the filesystem / change its placeholder state
 	var clearPlaceholderProperty bool
 	var recvOpts zfs.RecvOptions
 	ph, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, lp)
-	if err != nil {
+	if err == nil && ph.FSExists && ph.IsPlaceholder {
-		return nil, errors.Wrap(err, "cannot get placeholder state")
-	}
-	log.WithField("placeholder_state", fmt.Sprintf("%#v", ph)).Debug("placeholder state")
-	if ph.FSExists && ph.IsPlaceholder {
 		recvOpts.RollbackAndForceRecv = true
 		clearPlaceholderProperty = true
 	}
-
 	if clearPlaceholderProperty {
-		log.Info("clearing placeholder property")
 		if err := zfs.ZFSSetPlaceholder(ctx, lp, false); err != nil {
 			return nil, fmt.Errorf("cannot clear placeholder property for forced receive: %s", err)
 		}
 	}
 
 	if req.ClearResumeToken && ph.FSExists {
-		log.Info("clearing resume token")
 		if err := zfs.ZFSRecvClearResumeToken(ctx, lp.ToString()); err != nil {
 			return nil, errors.Wrap(err, "cannot clear resume token")
 		}
@@ -802,7 +790,7 @@ 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")
 	}
 
-	log.Debug("acquire concurrent recv semaphore")
+	getLogger(ctx).Debug("acquire concurrent recv semaphore")
 	// TODO use try-acquire and fail with resource-exhaustion rpc status
 	// => would require handling on the client-side
 	// => this is a dataconn endpoint, doesn't have the status code semantics of gRPC
@@ -812,89 +800,14 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.
 	}
 	defer guard.Release()
 
-	log.Info("peeking 1M ahead")
+	getLogger(ctx).WithField("opts", fmt.Sprintf("%#v", recvOpts)).Debug("start receive command")
-	var peek bytes.Buffer
-	var MaxPeek = envconst.Int64("ZREPL_ENDPOINT_RECV_PEEK_SIZE", 1<<20)
-	if _, err := io.Copy(&peek, io.LimitReader(receive, MaxPeek)); err != nil {
-		log.WithError(err).Error("cannot read peek-buffer from send stream")
-	}
-	var peekCopy bytes.Buffer
-	if n, err := peekCopy.Write(peek.Bytes()); err != nil || n != peek.Len() {
-		panic(peek.Len())
-	}
-
-	log.WithField("opts", fmt.Sprintf("%#v", recvOpts)).Debug("start receive command")
 
 	snapFullPath := to.FullPath(lp.ToString())
-	if err := zfs.ZFSRecv(ctx, lp.ToString(), to, chainedio.NewChainedReader(&peek, receive), recvOpts); err != nil {
+	if err := zfs.ZFSRecv(ctx, lp.ToString(), to, receive, recvOpts); err != nil {
-
+		getLogger(ctx).
-		// best-effort rollback of placeholder state if the recv didn't start
-		_, resumableStatePresent := err.(*zfs.RecvFailedWithResumeTokenErr)
-		disablePlaceholderRestoration := envconst.Bool("ZREPL_ENDPOINT_DISABLE_PLACEHOLDER_RESTORATION", false)
-		placeholderRestored := !ph.IsPlaceholder
-		if !disablePlaceholderRestoration && !resumableStatePresent && recvOpts.RollbackAndForceRecv && ph.FSExists && ph.IsPlaceholder && clearPlaceholderProperty {
-			log.Info("restoring placeholder property")
-			if phErr := zfs.ZFSSetPlaceholder(ctx, lp, true); phErr != nil {
-				log.WithError(phErr).Error("cannot restore placeholder property after failed receive, subsequent replications will likely fail with a different error")
-				// fallthrough
-			} else {
-				placeholderRestored = true
-			}
-			// fallthrough
-		}
-
-		// deal with failing initial encrypted send & recv
-		if _, ok := err.(*zfs.RecvDestroyOrOverwriteEncryptedErr); ok && ph.IsPlaceholder && placeholderRestored {
-			msg := `cannot automatically replace placeholder filesystem with incoming send stream - please see receive-side log for details`
-			err := errors.New(msg)
-			log.Error(msg)
-
-			log.Error(`zrepl creates placeholder filesystems on the receiving side of a replication to match the sending side's dataset hierarchy`)
-			log.Error(`zrepl uses zfs receive -F to replace those placeholders with incoming full sends`)
-			log.Error(`OpenZFS native encryption prohibits zfs receive -F for encrypted filesystems`)
-			log.Error(`the current zrepl placeholder filesystem concept is thus incompatible with OpenZFS native encryption`)
-
-			tempStartFullRecvFS := lp.Copy().ToString() + ".zrepl.initial-recv"
-			tempStartFullRecvFSDP, dpErr := zfs.NewDatasetPath(tempStartFullRecvFS)
-			if dpErr != nil {
-				log.WithError(dpErr).Error("cannot determine temporary filesystem name for initial encrypted recv workaround")
-				return nil, err // yes, err, not dpErr
-			}
-
-			log := log.WithField("temp_recv_fs", tempStartFullRecvFS)
-			log.Error(`as a workaround, zrepl will now attempt to re-receive the beginning of the stream into a temporary filesystem temp_recv_fs`)
-			log.Error(`if that step succeeds: shut down zrepl and use 'zfs rename' to swap temp_recv_fs with local_fs, then restart zrepl`)
-			log.Error(`replication will then resume using resumable send+recv`)
-
-			tempPH, phErr := zfs.ZFSGetFilesystemPlaceholderState(ctx, tempStartFullRecvFSDP)
-			if phErr != nil {
-				log.WithError(phErr).Error("cannot determine placeholder state of temp_recv_fs")
-				return nil, err // yes, err, not dpErr
-			}
-			if tempPH.FSExists {
-				log.Error("temp_recv_fs already exists, assuming a (partial) initial recv to that filesystem has already been done")
-				return nil, err
-			}
-
-			recvOpts.RollbackAndForceRecv = false
-			recvOpts.SavePartialRecvState = true
-			rerecvErr := zfs.ZFSRecv(ctx, tempStartFullRecvFS, to, chainedio.NewChainedReader(&peekCopy), recvOpts)
-			if _, isResumable := rerecvErr.(*zfs.RecvFailedWithResumeTokenErr); rerecvErr == nil || isResumable {
-				log.Error("completed re-receive into temporary filesystem temp_recv_fs, now shut down zrepl and use zfs rename to swap temp_recv_fs with local_fs")
-			} else {
-				log.WithError(rerecvErr).Error("failed to receive the beginning of the stream into temporary filesystem temp_recv_fs")
-				log.Error("we advise you to collect the error log and current configuration, open an issue on GitHub, and revert to your previous configuration in the meantime")
-			}
-
-			log.Error(`if you would like to see improvements to this situation, please open an issue on GitHub`)
-			return nil, err
-		}
-
-		log.
 			WithError(err).
 			WithField("opts", fmt.Sprintf("%#v", recvOpts)).
 			Error("zfs receive failed")
-
 		return nil, err
 	}
 
@@ -902,13 +815,13 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.
 	toRecvd, err := to.ValidateExistsAndGetVersion(ctx, lp.ToString())
 	if err != nil {
 		msg := "receive request's `To` version does not match what we received in the stream"
-		log.WithError(err).WithField("snap", snapFullPath).Error(msg)
+		getLogger(ctx).WithError(err).WithField("snap", snapFullPath).Error(msg)
-		log.Error("aborting recv request, but keeping received snapshot for inspection")
+		getLogger(ctx).Error("aborting recv request, but keeping received snapshot for inspection")
 		return nil, errors.Wrap(err, msg)
 	}
 
 	if s.conf.UpdateLastReceivedHold {
-		log.Debug("move last-received-hold")
+		getLogger(ctx).Debug("move last-received-hold")
 		if err := MoveLastReceivedHold(ctx, lp.ToString(), toRecvd, s.conf.JobID); err != nil {
 			return nil, errors.Wrap(err, "cannot move last-received-hold")
 		}

endpoint/endpoint_zfs_abstraction.go

@@ -467,7 +467,7 @@ func (e ListAbstractionsErrors) Error() string {
 	}
 	msgs := make([]string, len(e))
 	for i := range e {
-		msgs[i] = e.Error()
+		msgs[i] = e[i].Error()
 	}
 	return fmt.Sprintf("list endpoint abstractions: multiple errors:\n%s", strings.Join(msgs, "\n"))
 }
@@ -731,6 +731,9 @@ func listStaleFiltering(abs []Abstraction, sinceBound *CreateTXGRangeBound) *Sta
 	}
 	stepFirstNotStaleCandidates := make(map[fsAndJobId]stepFirstNotStaleCandidate) // empty map => will always return nil
 	for _, a := range abs {
+		if a.GetJobID() == nil {
+			continue // already put those into always-live list noJobId in above loop
+		}
 		key := fsAndJobId{a.GetFS(), *a.GetJobID()}
 		c := stepFirstNotStaleCandidates[key]
 

go.mod

@@ -21,7 +21,7 @@ require (
 	github.com/onsi/gomega v1.7.0 // indirect
 	github.com/pkg/errors v0.8.1
 	github.com/pkg/profile v1.2.1
-	github.com/problame/go-netssh v0.0.0-20191209123953-18d8aa6923c7
+	github.com/problame/go-netssh v0.0.0-20200601114649-26439f9f0dc5
 	github.com/prometheus/client_golang v1.2.1
 	github.com/sergi/go-diff v1.0.1-0.20180205163309-da645544ed44 // go1.12 thinks it needs this
 	github.com/spf13/cobra v0.0.2

go.sum

@@ -208,6 +208,8 @@ github.com/problame/go-netssh v0.0.0-20191026123024-f34099f4f6b1 h1:HH8yzlaZq/A8
 github.com/problame/go-netssh v0.0.0-20191026123024-f34099f4f6b1/go.mod h1:JRs3nyBjvOv/9YHC+Vlw9z1Jfzl5s5t0BNnTzmclj1c=
 github.com/problame/go-netssh v0.0.0-20191209123953-18d8aa6923c7 h1:Oik8tLrLhoMq4fduDRuNNOAQ+q0M0dXkjAYLUf4+mAc=
 github.com/problame/go-netssh v0.0.0-20191209123953-18d8aa6923c7/go.mod h1:Ba6RaFpK1+IHWs1wLZ6sCBuXkt4iAVLeOG5GinXHusM=
+github.com/problame/go-netssh v0.0.0-20200601114649-26439f9f0dc5 h1:1eSrO2WAeSXjMol8WRKW9LekL1252VIMaKQwWkmEdvs=
+github.com/problame/go-netssh v0.0.0-20200601114649-26439f9f0dc5/go.mod h1:Ba6RaFpK1+IHWs1wLZ6sCBuXkt4iAVLeOG5GinXHusM=
 github.com/prometheus/client_golang v0.0.0-20180410130117-e11c6ff8170b/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
 github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
 github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=
@@ -263,6 +265,7 @@ github.com/spf13/viper v1.0.2/go.mod h1:A8kyI5cUJhb8N+3pkfONlcEcZbueH6nhAm0Fq7Sr
 github.com/spf13/viper v1.3.2/go.mod h1:ZiWeW+zYFKm7srdB9IoDzzZXaJaI5eL9QjNiN/DMA2s=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.2.1/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=

platformtest/platformtest.go

@@ -27,6 +27,7 @@ func (c *Context) Logf(format string, args ...interface{}) {
 
 func (c *Context) Errorf(format string, args ...interface{}) {
 	GetLog(c).Printf(format, args...)
+	c.FailNow()
 }
 
 func (c *Context) FailNow() {

platformtest/tests/helpers.go

@@ -136,7 +136,7 @@ func makeResumeSituation(ctx *platformtest.Context, src dummySnapshotSituation,
 		return situation
 	}
 
-	limitedCopier := limitio.ReadCloser(copier, src.dummyDataLen/2)
+	limitedCopier := zfs.NewReadCloserCopier(limitio.ReadCloser(copier, src.dummyDataLen/2))
 	defer limitedCopier.Close()
 
 	require.NotNil(ctx, sendArgs.To)

platformtest/tests/sendArgsValidation.go

@@ -35,7 +35,7 @@ func SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden(ctx *platformt
 		ResumeToken: "",
 	}.Validate(ctx)
 
-	var stream *zfs.SendStream
+	var stream *zfs.ReadCloserCopier
 	if err == nil {
 		stream, err = zfs.ZFSSend(ctx, sendArgs) // no shadow
 		if err == nil {

replication/driver/replication_driver.go

@@ -146,12 +146,6 @@ type fs struct {
 
 	l *chainlock.L
 
-	// ordering relationship that must be maintained for initial replication
-	initialRepOrd struct {
-		parents, children []*fs
-		parentDidUpdate   chan struct{}
-	}
-
 	planning struct {
 		done bool
 		err  *timedError
@@ -302,7 +296,6 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 			fs: pfs,
 			l:  a.l,
 		}
-		fs.initialRepOrd.parentDidUpdate = make(chan struct{}, 1)
 		a.fss = append(a.fss, fs)
 	}
 
@@ -361,18 +354,6 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 	}
 	// invariant: prevs contains an entry for each unambiguous correspondence
 
-	// build up parent-child relationship (FIXME (O(n^2), but who's going to have that many filesystems...))
-	for _, f1 := range a.fss {
-		fs1 := f1.fs.ReportInfo().Name
-		for _, f2 := range a.fss {
-			fs2 := f2.fs.ReportInfo().Name
-			if strings.HasPrefix(fs1, fs2) && fs1 != fs2 {
-				f1.initialRepOrd.parents = append(f1.initialRepOrd.parents, f2)
-				f2.initialRepOrd.children = append(f2.initialRepOrd.children, f1)
-			}
-		}
-	}
-
 	stepQueue := newStepQueue()
 	defer stepQueue.Start(envconst.Int("ZREPL_REPLICATION_EXPERIMENTAL_REPLICATION_CONCURRENCY", 1))() // TODO parallel replication
 	var fssesDone sync.WaitGroup
@@ -393,27 +374,9 @@ func (f *fs) debug(format string, args ...interface{}) {
 	debugPrefix("fs=%s", f.fs.ReportInfo().Name)(format, args...)
 }
 
-// wake up children that watch for f.{planning.{err,done},planned.{step,stepErr}}
-func (f *fs) initialRepOrdWakeupChildren() {
-	var children []string
-	for _, c := range f.initialRepOrd.children {
-		// no locking required, c.fs does not change
-		children = append(children, c.fs.ReportInfo().Name)
-	}
-	f.debug("wakeup children %s", children)
-	for _, child := range f.initialRepOrd.children {
-		select {
-		// no locking required, child.initialRepOrd does not change
-		case child.initialRepOrd.parentDidUpdate <- struct{}{}:
-		default:
-		}
-	}
-}
-
 func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 
 	defer f.l.Lock().Unlock()
-	defer f.initialRepOrdWakeupChildren()
 
 	// get planned steps from replication logic
 	var psteps []Step
@@ -427,6 +390,7 @@ func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 		psteps, err = f.fs.PlanFS(ctx) // no shadow
 		errTime = time.Now()           // no shadow
 	})
+	f.planning.done = true
 	if err != nil {
 		f.planning.err = newTimedError(err, errTime)
 		return
@@ -438,8 +402,6 @@ func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 		}
 		f.planned.steps = append(f.planned.steps, step)
 	}
-	// we're not done planning yet, f.planned.steps might still be changed by next block
-	// => don't set f.planning.done just yet
 	f.debug("initial len(fs.planned.steps) = %d", len(f.planned.steps))
 
 	// for not-first attempts, only allow fs.planned.steps
@@ -494,110 +456,24 @@ func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 	}
 	f.debug("post-prev-merge len(fs.planned.steps) = %d", len(f.planned.steps))
 
-	// now we are done planning (f.planned.steps won't change from now on)
-	f.planning.done = true
-
-	// wait for parents' initial replication
-	var parents []string
-	for _, p := range f.initialRepOrd.parents {
-		parents = append(parents, p.fs.ReportInfo().Name)
-	}
-	f.debug("wait for parents %s", parents)
-	for {
-		var initialReplicatingParentsWithErrors []string
-		allParentsPresentOnReceiver := true
-		f.l.DropWhile(func() {
-			for _, p := range f.initialRepOrd.parents {
-				p.l.HoldWhile(func() {
-					// (get the preconditions that allow us to inspect p.planned)
-					parentHasPlanningDone := p.planning.done && p.planning.err == nil
-					if !parentHasPlanningDone {
-						// if the parent couldn't be planned, we cannot know whether it needs initial replication
-						// or incremental replication => be conservative and assume it was initial replication
-						allParentsPresentOnReceiver = false
-						if p.planning.err != nil {
-							initialReplicatingParentsWithErrors = append(initialReplicatingParentsWithErrors, p.fs.ReportInfo().Name)
-						}
-						return
-					}
-					// now allowed to inspect p.planned
-
-					// if there are no steps to be done, the filesystem must exist on the receiving side
-					// (otherwise we'd replicate it, and there would be a step for that)
-					// (FIXME hardcoded initial replication policy, assuming the policy will always do _some_ initial replication)
-					parentHasNoSteps := len(p.planned.steps) == 0
-
-					// OR if it has completed at least one step
-					// (remember that .step points to the next step to be done)
-					// (TODO technically, we could make this step ready in the moment the recv-side
-					//  dataset exists, i.e. after the first few megabytes of transferred data, but we'd have to ask the receiver for that -> poll ListFilesystems RPC)
-					parentHasTakenAtLeastOneSuccessfulStep := !parentHasNoSteps && p.planned.step >= 1
-
-					parentFirstStepIsIncremental := // no need to lock for .report() because step.l == it's fs.l
-						len(p.planned.steps) > 0 && p.planned.steps[0].report().IsIncremental()
-
-					f.debug("parentHasNoSteps=%v parentFirstStepIsIncremental=%v parentHasTakenAtLeastOneSuccessfulStep=%v",
-						parentHasNoSteps, parentFirstStepIsIncremental, parentHasTakenAtLeastOneSuccessfulStep)
-
-					parentPresentOnReceiver := parentHasNoSteps || parentFirstStepIsIncremental || parentHasTakenAtLeastOneSuccessfulStep
-
-					allParentsPresentOnReceiver = allParentsPresentOnReceiver && parentPresentOnReceiver // no shadow
-
-					if !parentPresentOnReceiver && p.planned.stepErr != nil {
-						initialReplicatingParentsWithErrors = append(initialReplicatingParentsWithErrors, p.fs.ReportInfo().Name)
-					}
-
-				})
-			}
-		})
-
-		if len(initialReplicatingParentsWithErrors) > 0 {
-			f.planned.stepErr = newTimedError(fmt.Errorf("parent(s) failed during initial replication: %s", initialReplicatingParentsWithErrors), time.Now())
-			return
-		}
-
-		if allParentsPresentOnReceiver {
-			break // good to go
-		}
-
-		// wait for wakeups from parents, then check again
-		// lock must not be held while waiting in order for reporting to work
-		f.l.DropWhile(func() {
-			select {
-			case <-ctx.Done():
-				f.planned.stepErr = newTimedError(ctx.Err(), time.Now())
-				return
-			case <-f.initialRepOrd.parentDidUpdate:
-				// loop
-			}
-		})
-		if f.planned.stepErr != nil {
-			return
-		}
-	}
-
-	f.debug("all parents ready, start replication %s", parents)
-
-	// do our steps
 	for i, s := range f.planned.steps {
+		var (
+			err     error
+			errTime time.Time
+		)
 		// lock must not be held while executing step in order for reporting to work
 		f.l.DropWhile(func() {
-			// wait for parallel replication
 			targetDate := s.step.TargetDate()
 			defer pq.WaitReady(f, targetDate)()
-			// do the step
+			err = s.step.Step(ctx) // no shadow
-			err, errTime = s.step.Step(ctx), time.Now() // no shadow
+			errTime = time.Now()   // no shadow
 		})
-
 		if err != nil {
 			f.planned.stepErr = newTimedError(err, errTime)
 			break
 		}
 		f.planned.step = i + 1 // fs.planned.step must be == len(fs.planned.steps) if all went OK
-
-		f.initialRepOrdWakeupChildren()
 	}
-
 }
 
 // caller must hold lock l

replication/driver/replication_driver_debug.go

@@ -26,6 +26,6 @@ type debugFunc func(format string, args ...interface{})
 func debugPrefix(prefixFormat string, prefixFormatArgs ...interface{}) debugFunc {
 	prefix := fmt.Sprintf(prefixFormat, prefixFormatArgs...)
 	return func(format string, args ...interface{}) {
-		debug("%s: %s", prefix, fmt.Sprintf(format, args...))
+		debug("%s: %s", prefix, fmt.Sprintf(format, args))
 	}
 }

replication/logic/replication_logic.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"io"
 	"sync"
 	"time"
 
@@ -39,7 +38,7 @@ type Sender interface {
 	// If a non-nil io.ReadCloser is returned, it is guaranteed to be closed before
 	// any next call to the parent github.com/zrepl/zrepl/replication.Endpoint.
 	// If the send request is for dry run the io.ReadCloser will be nil
-	Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error)
+	Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error)
 	SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error)
 	ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error)
 }
@@ -48,7 +47,7 @@ type Receiver interface {
 	Endpoint
 	// Receive sends r and sendStream (the latter containing a ZFS send stream)
 	// to the parent github.com/zrepl/zrepl/replication.Endpoint.
-	Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.ReadCloser) (*pdu.ReceiveRes, error)
+	Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs.StreamCopier) (*pdu.ReceiveRes, error)
 }
 
 type PlannerPolicy struct {
@@ -163,7 +162,7 @@ type Step struct {
 
 	// byteCounter is nil initially, and set later in Step.doReplication
 	// => concurrent read of that pointer from Step.ReportInfo must be protected
-	byteCounter    bytecounter.ReadCloser
+	byteCounter    bytecounter.StreamCopier
 	byteCounterMtx chainlock.L
 }
 
@@ -603,23 +602,23 @@ func (s *Step) doReplication(ctx context.Context) error {
 
 	fs := s.parent.Path
 
-	log := getLogger(ctx)
+	log := getLogger(ctx).WithField("filesystem", fs)
 	sr := s.buildSendRequest(false)
 
 	log.Debug("initiate send request")
-	sres, stream, err := s.sender.Send(ctx, sr)
+	sres, sstreamCopier, err := s.sender.Send(ctx, sr)
 	if err != nil {
 		log.WithError(err).Error("send request failed")
 		return err
 	}
-	if stream == nil {
+	if sstreamCopier == nil {
 		err := errors.New("send request did not return a stream, broken endpoint implementation")
 		return err
 	}
-	defer stream.Close()
+	defer sstreamCopier.Close()
 
 	// Install a byte counter to track progress + for status report
-	byteCountingStream := bytecounter.NewReadCloser(stream)
+	byteCountingStream := bytecounter.NewStreamCopier(sstreamCopier)
 	s.byteCounterMtx.Lock()
 	s.byteCounter = byteCountingStream
 	s.byteCounterMtx.Unlock()

rpc/dataconn/dataconn_client.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"io"
 	"strings"
 
 	"github.com/golang/protobuf/proto"
@@ -12,6 +11,7 @@ import (
 	"github.com/zrepl/zrepl/replication/logic/pdu"
 	"github.com/zrepl/zrepl/rpc/dataconn/stream"
 	"github.com/zrepl/zrepl/transport"
+	"github.com/zrepl/zrepl/zfs"
 )
 
 type Client struct {
@@ -26,7 +26,7 @@ func NewClient(connecter transport.Connecter, log Logger) *Client {
 	}
 }
 
-func (c *Client) send(ctx context.Context, conn *stream.Conn, endpoint string, req proto.Message, stream io.ReadCloser) error {
+func (c *Client) send(ctx context.Context, conn *stream.Conn, endpoint string, req proto.Message, streamCopier zfs.StreamCopier) error {
 
 	var buf bytes.Buffer
 	_, memErr := buf.WriteString(endpoint)
@@ -46,8 +46,8 @@ func (c *Client) send(ctx context.Context, conn *stream.Conn, endpoint string, r
 		return err
 	}
 
-	if stream != nil {
+	if streamCopier != nil {
-		return conn.SendStream(ctx, stream, ZFSStream)
+		return conn.SendStream(ctx, streamCopier, ZFSStream)
 	} else {
 		return nil
 	}
@@ -109,7 +109,7 @@ func (c *Client) putWire(conn *stream.Conn) {
 	}
 }
 
-func (c *Client) ReqSend(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
+func (c *Client) ReqSend(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
 	conn, err := c.getWire(ctx)
 	if err != nil {
 		return nil, nil, err
@@ -130,19 +130,16 @@ func (c *Client) ReqSend(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, i
 		return nil, nil, err
 	}
 
-	var stream io.ReadCloser
+	var copier zfs.StreamCopier = nil
 	if !req.DryRun {
 		putWireOnReturn = false
-		stream, err = conn.ReadStream(ZFSStream, true) // no shadow
+		copier = &streamCopier{streamConn: conn, closeStreamOnClose: true}
-		if err != nil {
-			return nil, nil, err
-		}
 	}
 
-	return &res, stream, nil
+	return &res, copier, nil
 }
 
-func (c *Client) ReqRecv(ctx context.Context, req *pdu.ReceiveReq, stream io.ReadCloser) (*pdu.ReceiveRes, error) {
+func (c *Client) ReqRecv(ctx context.Context, req *pdu.ReceiveReq, streamCopier zfs.StreamCopier) (*pdu.ReceiveRes, error) {
 
 	defer c.log.Debug("ReqRecv returns")
 	conn, err := c.getWire(ctx)
@@ -169,7 +166,7 @@ func (c *Client) ReqRecv(ctx context.Context, req *pdu.ReceiveReq, stream io.Rea
 
 	sendErrChan := make(chan error)
 	go func() {
-		if err := c.send(ctx, conn, EndpointRecv, req, stream); err != nil {
+		if err := c.send(ctx, conn, EndpointRecv, req, streamCopier); err != nil {
 			sendErrChan <- err
 		} else {
 			sendErrChan <- nil

rpc/dataconn/dataconn_server.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"io"
 
 	"github.com/golang/protobuf/proto"
 
@@ -12,6 +11,7 @@ import (
 	"github.com/zrepl/zrepl/replication/logic/pdu"
 	"github.com/zrepl/zrepl/rpc/dataconn/stream"
 	"github.com/zrepl/zrepl/transport"
+	"github.com/zrepl/zrepl/zfs"
 )
 
 // WireInterceptor has a chance to exchange the context and connection on each client connection.
@@ -21,11 +21,11 @@ type WireInterceptor func(ctx context.Context, rawConn *transport.AuthConn) (con
 type Handler interface {
 	// Send handles a SendRequest.
 	// The returned io.ReadCloser is allowed to be nil, for example if the requested Send is a dry-run.
-	Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error)
+	Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error)
 	// Receive handles a ReceiveRequest.
 	// It is guaranteed that Server calls Receive with a stream that holds the IdleConnTimeout
 	// configured in ServerConfig.Shared.IdleConnTimeout.
-	Receive(ctx context.Context, r *pdu.ReceiveReq, receive io.ReadCloser) (*pdu.ReceiveRes, error)
+	Receive(ctx context.Context, r *pdu.ReceiveReq, receive zfs.StreamCopier) (*pdu.ReceiveRes, error)
 	// PingDataconn handles a PingReq
 	PingDataconn(ctx context.Context, r *pdu.PingReq) (*pdu.PingRes, error)
 }
@@ -111,7 +111,7 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
 	s.log.WithField("endpoint", endpoint).Debug("calling handler")
 
 	var res proto.Message
-	var sendStream io.ReadCloser
+	var sendStream zfs.StreamCopier
 	var handlerErr error
 	switch endpoint {
 	case EndpointSend:
@@ -127,12 +127,7 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
 			s.log.WithError(err).Error("cannot unmarshal receive request")
 			return
 		}
-		stream, err := c.ReadStream(ZFSStream, false)
+		res, handlerErr = s.h.Receive(ctx, &req, &streamCopier{streamConn: c, closeStreamOnClose: false}) // SHADOWING
-		if err != nil {
-			s.log.WithError(err).Error("cannot open stream in receive request")
-			return
-		}
-		res, handlerErr = s.h.Receive(ctx, &req, stream) // SHADOWING
 	case EndpointPing:
 		var req pdu.PingReq
 		if err := proto.Unmarshal(reqStructured, &req); err != nil {

rpc/dataconn/dataconn_shared.go

@@ -1,7 +1,12 @@
 package dataconn
 
 import (
+	"io"
+	"sync"
 	"time"
+
+	"github.com/zrepl/zrepl/rpc/dataconn/stream"
+	"github.com/zrepl/zrepl/zfs"
 )
 
 const (
@@ -34,3 +39,33 @@ const (
 	responseHeaderHandlerOk          = "HANDLER OK\n"
 	responseHeaderHandlerErrorPrefix = "HANDLER ERROR:\n"
 )
+
+type streamCopier struct {
+	mtx                sync.Mutex
+	used               bool
+	streamConn         *stream.Conn
+	closeStreamOnClose bool
+}
+
+// WriteStreamTo implements zfs.StreamCopier
+func (s *streamCopier) WriteStreamTo(w io.Writer) zfs.StreamCopierError {
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+	if s.used {
+		panic("streamCopier used multiple times")
+	}
+	s.used = true
+	return s.streamConn.ReadStreamInto(w, ZFSStream)
+}
+
+// Close implements zfs.StreamCopier
+func (s *streamCopier) Close() error {
+	// only record the close here, what we do actually depends on whether
+	// the streamCopier is instantiated server-side or client-side
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+	if s.closeStreamOnClose {
+		return s.streamConn.Close()
+	}
+	return nil
+}

rpc/dataconn/microbenchmark/microbenchmark.go

@@ -29,6 +29,7 @@ import (
 	"github.com/zrepl/zrepl/rpc/dataconn/timeoutconn"
 	"github.com/zrepl/zrepl/transport"
 	"github.com/zrepl/zrepl/util/devnoop"
+	"github.com/zrepl/zrepl/zfs"
 )
 
 func orDie(err error) {
@@ -41,9 +42,23 @@ type readerStreamCopier struct{ io.Reader }
 
 func (readerStreamCopier) Close() error { return nil }
 
+type readerStreamCopierErr struct {
+	error
+}
+
+func (readerStreamCopierErr) IsReadError() bool  { return false }
+func (readerStreamCopierErr) IsWriteError() bool { return true }
+
+func (c readerStreamCopier) WriteStreamTo(w io.Writer) zfs.StreamCopierError {
+	var buf [1 << 21]byte
+	_, err := io.CopyBuffer(w, c.Reader, buf[:])
+	// always assume write error
+	return readerStreamCopierErr{err}
+}
+
 type devNullHandler struct{}
 
-func (devNullHandler) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
+func (devNullHandler) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
 	var res pdu.SendRes
 	if args.devnoopReader {
 		return &res, readerStreamCopier{devnoop.Get()}, nil
@@ -52,12 +67,12 @@ func (devNullHandler) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, i
 	}
 }
 
-func (devNullHandler) Receive(ctx context.Context, r *pdu.ReceiveReq, stream io.ReadCloser) (*pdu.ReceiveRes, error) {
+func (devNullHandler) Receive(ctx context.Context, r *pdu.ReceiveReq, stream zfs.StreamCopier) (*pdu.ReceiveRes, error) {
 	var out io.Writer = os.Stdout
 	if args.devnoopWriter {
 		out = devnoop.Get()
 	}
-	_, err := io.Copy(out, stream)
+	err := stream.WriteStreamTo(out)
 	var res pdu.ReceiveRes
 	return &res, err
 }
@@ -157,7 +172,7 @@ func client() {
 		req := pdu.SendReq{}
 		_, stream, err := client.ReqSend(ctx, &req)
 		orDie(err)
-		_, err = io.Copy(os.Stdout, stream)
+		err = stream.WriteStreamTo(os.Stdout)
 		orDie(err)
 	case "recv":
 		var r io.Reader = os.Stdin

rpc/dataconn/stream/stream.go

@@ -14,6 +14,7 @@ import (
 	"github.com/zrepl/zrepl/rpc/dataconn/base2bufpool"
 	"github.com/zrepl/zrepl/rpc/dataconn/frameconn"
 	"github.com/zrepl/zrepl/rpc/dataconn/heartbeatconn"
+	"github.com/zrepl/zrepl/zfs"
 )
 
 type Logger = logger.Logger
@@ -197,7 +198,7 @@ func (e ReadStreamError) Temporary() bool {
 	return false
 }
 
-var _ net.Error = &ReadStreamError{}
+var _ zfs.StreamCopierError = &ReadStreamError{}
 
 func (e ReadStreamError) IsReadError() bool {
 	return e.Kind != ReadStreamErrorKindWrite

rpc/dataconn/stream/stream_conn.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/zrepl/zrepl/rpc/dataconn/heartbeatconn"
 	"github.com/zrepl/zrepl/rpc/dataconn/timeoutconn"
+	"github.com/zrepl/zrepl/zfs"
 )
 
 type Conn struct {
@@ -39,7 +40,15 @@ type Conn struct {
 
 var readMessageSentinel = fmt.Errorf("read stream complete")
 
-var errWriteStreamToErrorUnknownState = fmt.Errorf("dataconn read stream: connection is in unknown state")
+type writeStreamToErrorUnknownState struct{}
+
+func (e writeStreamToErrorUnknownState) Error() string {
+	return "dataconn read stream: connection is in unknown state"
+}
+
+func (e writeStreamToErrorUnknownState) IsReadError() bool { return true }
+
+func (e writeStreamToErrorUnknownState) IsWriteError() bool { return false }
 
 func Wrap(nc timeoutconn.Wire, sendHeartbeatInterval, peerTimeout time.Duration) *Conn {
 	hc := heartbeatconn.Wrap(nc, sendHeartbeatInterval, peerTimeout)
@@ -114,28 +123,14 @@ func (c *Conn) ReadStreamedMessage(ctx context.Context, maxSize uint32, frameTyp
 	}
 }
 
-type StreamReader struct {
-	*io.PipeReader
-	conn             *Conn
-	closeConnOnClose bool
-}
-
-func (r *StreamReader) Close() error {
-	err := r.PipeReader.Close()
-	if r.closeConnOnClose {
-		r.conn.Close() // TODO error logging
-	}
-	return err
-}
-
 // WriteStreamTo reads a stream from Conn and writes it to w.
-func (c *Conn) ReadStream(frameType uint32, closeConnOnClose bool) (_ *StreamReader, err error) {
+func (c *Conn) ReadStreamInto(w io.Writer, frameType uint32) (err zfs.StreamCopierError) {
 
 	// if we are closed while writing, return that as an error
 	if closeGuard, cse := c.closeState.RWEntry(); cse != nil {
-		return nil, cse
+		return cse
 	} else {
-		defer func(err *error) {
+		defer func(err *zfs.StreamCopierError) {
 			if closed := closeGuard.RWExit(); closed != nil {
 				*err = closed
 			}
@@ -143,23 +138,18 @@ func (c *Conn) ReadStream(frameType uint32, closeConnOnClose bool) (_ *StreamRea
 	}
 
 	c.readMtx.Lock()
+	defer c.readMtx.Unlock()
 	if !c.readClean {
-		return nil, errWriteStreamToErrorUnknownState
+		return writeStreamToErrorUnknownState{}
 	}
+	var rse *ReadStreamError = readStream(c.frameReads, c.hc, w, frameType)
+	c.readClean = isConnCleanAfterRead(rse)
 
-	r, w := io.Pipe()
+	// https://golang.org/doc/faq#nil_error
-	go func() {
+	if rse == nil {
-		defer c.readMtx.Unlock()
+		return nil
-		var err *ReadStreamError = readStream(c.frameReads, c.hc, w, frameType)
+	}
-		if err != nil {
+	return rse
-			_ = w.CloseWithError(err) // doc guarantees that error will always be nil
-		} else {
-			w.Close()
-		}
-		c.readClean = isConnCleanAfterRead(err)
-	}()
-
-	return &StreamReader{PipeReader: r, conn: c, closeConnOnClose: closeConnOnClose}, nil
 }
 
 func (c *Conn) WriteStreamedMessage(ctx context.Context, buf io.Reader, frameType uint32) (err error) {
@@ -188,7 +178,7 @@ func (c *Conn) WriteStreamedMessage(ctx context.Context, buf io.Reader, frameTyp
 	return errConn
 }
 
-func (c *Conn) SendStream(ctx context.Context, stream io.ReadCloser, frameType uint32) (err error) {
+func (c *Conn) SendStream(ctx context.Context, src zfs.StreamCopier, frameType uint32) (err error) {
 
 	// if we are closed while reading, return that as an error
 	if closeGuard, cse := c.closeState.RWEntry(); cse != nil {
@@ -207,17 +197,49 @@ func (c *Conn) SendStream(ctx context.Context, stream io.ReadCloser, frameType u
 		return fmt.Errorf("dataconn send stream: connection is in unknown state")
 	}
 
-	errStream, errConn := writeStream(ctx, c.hc, stream, frameType)
+	// avoid io.Pipe if zfs.StreamCopier is an io.Reader
-
+	var r io.Reader
-	c.writeClean = isConnCleanAfterWrite(errConn) // TODO correct?
+	var w *io.PipeWriter
-
+	streamCopierErrChan := make(chan zfs.StreamCopierError, 1)
-	if errStream != nil {
+	if reader, ok := src.(io.Reader); ok {
-		return errStream
+		r = reader
-	} else if errConn != nil {
+		streamCopierErrChan <- nil
-		return errConn
+		close(streamCopierErrChan)
+	} else {
+		r, w = io.Pipe()
+		go func() {
+			streamCopierErrChan <- src.WriteStreamTo(w)
+			w.Close()
+		}()
+	}
+
+	type writeStreamRes struct {
+		errStream, errConn error
+	}
+	writeStreamErrChan := make(chan writeStreamRes, 1)
+	go func() {
+		var res writeStreamRes
+		res.errStream, res.errConn = writeStream(ctx, c.hc, r, frameType)
+		if w != nil {
+			_ = w.CloseWithError(res.errStream) // always returns nil
+		}
+		writeStreamErrChan <- res
+	}()
+
+	writeRes := <-writeStreamErrChan
+	streamCopierErr := <-streamCopierErrChan
+	c.writeClean = isConnCleanAfterWrite(writeRes.errConn) // TODO correct?
+	if streamCopierErr != nil && streamCopierErr.IsReadError() {
+		return streamCopierErr // something on our side is bad
+	} else {
+		if writeRes.errStream != nil {
+			return writeRes.errStream
+		} else if writeRes.errConn != nil {
+			return writeRes.errConn
+		}
+		// TODO combined error?
+		return streamCopierErr
 	}
-	// TODO combined error?
-	return nil
 }
 
 type closeState struct {
@@ -226,13 +248,17 @@ type closeState struct {
 
 type closeStateErrConnectionClosed struct{}
 
+var _ zfs.StreamCopierError = (*closeStateErrConnectionClosed)(nil)
+var _ error = (*closeStateErrConnectionClosed)(nil)
 var _ net.Error = (*closeStateErrConnectionClosed)(nil)
 
 func (e *closeStateErrConnectionClosed) Error() string {
 	return "connection closed"
 }
-func (e *closeStateErrConnectionClosed) Timeout() bool   { return false }
+func (e *closeStateErrConnectionClosed) IsReadError() bool  { return true }
-func (e *closeStateErrConnectionClosed) Temporary() bool { return false }
+func (e *closeStateErrConnectionClosed) IsWriteError() bool { return true }
+func (e *closeStateErrConnectionClosed) Timeout() bool      { return false }
+func (e *closeStateErrConnectionClosed) Temporary() bool    { return false }
 
 func (s *closeState) CloseEntry() error {
 	firstCloser := atomic.AddUint32(&s.closeCount, 1) == 1
@@ -247,7 +273,7 @@ type closeStateEntry struct {
 	entryCount uint32
 }
 
-func (s *closeState) RWEntry() (e *closeStateEntry, err net.Error) {
+func (s *closeState) RWEntry() (e *closeStateEntry, err zfs.StreamCopierError) {
 	entry := &closeStateEntry{s, atomic.LoadUint32(&s.closeCount)}
 	if entry.entryCount > 0 {
 		return nil, &closeStateErrConnectionClosed{}
@@ -255,7 +281,7 @@ func (s *closeState) RWEntry() (e *closeStateEntry, err net.Error) {
 	return entry, nil
 }
 
-func (e *closeStateEntry) RWExit() net.Error {
+func (e *closeStateEntry) RWExit() zfs.StreamCopierError {
 	if atomic.LoadUint32(&e.entryCount) == e.entryCount {
 		// no calls to Close() while running rw operation
 		return nil

rpc/rpc_client.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"io"
 	"net"
 	"sync"
 	"sync/atomic"
@@ -21,6 +20,7 @@ import (
 	"github.com/zrepl/zrepl/rpc/versionhandshake"
 	"github.com/zrepl/zrepl/transport"
 	"github.com/zrepl/zrepl/util/envconst"
+	"github.com/zrepl/zrepl/zfs"
 )
 
 // Client implements the active side of a replication setup.
@@ -82,22 +82,22 @@ func (c *Client) Close() {
 
 // callers must ensure that the returned io.ReadCloser is closed
 // TODO expose dataClient interface to the outside world
-func (c *Client) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
+func (c *Client) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
 	// TODO the returned sendStream may return a read error created by the remote side
-	res, stream, err := c.dataClient.ReqSend(ctx, r)
+	res, streamCopier, err := c.dataClient.ReqSend(ctx, r)
 	if err != nil {
 		return nil, nil, err
 	}
-	if stream == nil {
+	if streamCopier == nil {
 		return res, nil, nil
 	}
 
-	return res, stream, nil
+	return res, streamCopier, nil
 
 }
 
-func (c *Client) Receive(ctx context.Context, req *pdu.ReceiveReq, stream io.ReadCloser) (*pdu.ReceiveRes, error) {
+func (c *Client) Receive(ctx context.Context, req *pdu.ReceiveReq, streamCopier zfs.StreamCopier) (*pdu.ReceiveRes, error) {
-	return c.dataClient.ReqRecv(ctx, req, stream)
+	return c.dataClient.ReqRecv(ctx, req, streamCopier)
 }
 
 func (c *Client) ListFilesystems(ctx context.Context, in *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error) {

transport/tcp/serve_tcp.go

@@ -11,39 +11,6 @@ import (
 	"github.com/zrepl/zrepl/util/tcpsock"
 )
 
-type ipMapEntry struct {
-	ip    net.IP
-	ident string
-}
-
-type ipMap struct {
-	entries []ipMapEntry
-}
-
-func ipMapFromConfig(clients map[string]string) (*ipMap, error) {
-	entries := make([]ipMapEntry, 0, len(clients))
-	for clientIPString, clientIdent := range clients {
-		clientIP := net.ParseIP(clientIPString)
-		if clientIP == nil {
-			return nil, errors.Errorf("cannot parse client IP %q", clientIPString)
-		}
-		if err := transport.ValidateClientIdentity(clientIdent); err != nil {
-			return nil, errors.Wrapf(err, "invalid client identity for IP %q", clientIPString)
-		}
-		entries = append(entries, ipMapEntry{clientIP, clientIdent})
-	}
-	return &ipMap{entries: entries}, nil
-}
-
-func (m *ipMap) Get(ip net.IP) (string, error) {
-	for _, e := range m.entries {
-		if e.ip.Equal(ip) {
-			return e.ident, nil
-		}
-	}
-	return "", errors.Errorf("no identity mapping for client IP %s", ip)
-}
-
 func TCPListenerFactoryFromConfig(c *config.Global, in *config.TCPServe) (transport.AuthenticatedListenerFactory, error) {
 	clientMap, err := ipMapFromConfig(in.Clients)
 	if err != nil {
@@ -69,10 +36,13 @@ func (f *TCPAuthListener) Accept(ctx context.Context) (*transport.AuthConn, erro
 	if err != nil {
 		return nil, err
 	}
-	clientIP := nc.RemoteAddr().(*net.TCPAddr).IP
+	clientAddr := &net.IPAddr{
-	clientIdent, err := f.clientMap.Get(clientIP)
+		IP:   nc.RemoteAddr().(*net.TCPAddr).IP,
+		Zone: nc.RemoteAddr().(*net.TCPAddr).Zone,
+	}
+	clientIdent, err := f.clientMap.Get(clientAddr)
 	if err != nil {
-		transport.GetLogger(ctx).WithField("ip", clientIP).Error("client IP not in client map")
+		transport.GetLogger(ctx).WithField("ipaddr", clientAddr).Error("client IP not in client map")
 		nc.Close()
 		return nil, err
 	}

transport/tcp/serve_tcp_ipmap.go

@@ -0,0 +1,166 @@
+package tcp
+
+import (
+	"bytes"
+	"fmt"
+	"net"
+	"sort"
+	"strings"
+
+	"github.com/pkg/errors"
+	"github.com/zrepl/zrepl/transport"
+	"golang.org/x/sys/unix"
+)
+
+type ipMapEntry struct {
+	subnet *net.IPNet
+	// ident is always not empty
+	ident string
+	// zone may be empty (e.g. for IPv4)
+	zone string
+}
+
+type ipMap struct {
+	entries []*ipMapEntry
+}
+
+func ipMapFromConfig(clients map[string]string) (*ipMap, error) {
+
+	entries := make([]*ipMapEntry, 0, len(clients))
+	for clientInput, clientIdent := range clients {
+		userIPMapEntry, err := newIPMapEntry(clientInput, clientIdent)
+		if err != nil {
+			return nil, errors.Wrapf(err, "cannot not parse %q:%q", clientInput, clientIdent)
+		}
+
+		entries = append(entries, userIPMapEntry)
+	}
+
+	sort.Sort(byPrefixlen(entries))
+
+	return &ipMap{entries: entries}, nil
+}
+
+func (m *ipMap) Get(ipAddr *net.IPAddr) (string, error) {
+	for _, e := range m.entries {
+		if e.zone != ipAddr.Zone {
+			continue
+		}
+		if e.subnet.Contains(ipAddr.IP) {
+			return zfsDatasetPathComponentCompatibleRepresentation(e.ident, ipAddr), nil
+		}
+	}
+	return "", errors.Errorf("no identity mapping for client IP: %s%%%s", ipAddr.IP, ipAddr.Zone)
+}
+
+var ipv6FullySpecifiedMask = bytes.Repeat([]byte{0xff}, net.IPv6len)
+
+func newIPMapEntry(input string, ident string) (*ipMapEntry, error) {
+	ip := input
+	var zone string
+
+	ipZoneSplit := strings.SplitN(input, "%", 2)
+	if len(ipZoneSplit) > 1 {
+		ip = ipZoneSplit[0]
+		zone = ipZoneSplit[1]
+	}
+
+	_, subnet, err := net.ParseCIDR(ip)
+	if err != nil {
+		// expect full IP, no '*' placeholder expansion
+
+		if strings.Count(ident, "*") != 0 {
+			return nil, fmt.Errorf("non-CIDR matches must not contain '*' placeholder")
+		}
+
+		if err := transport.ValidateClientIdentity(ident); err != nil {
+			return nil, errors.Wrapf(err, "invalid client identity %q for IP %q", ident, ip)
+		}
+
+		parsedIP := net.ParseIP(ip)
+		if parsedIP == nil {
+			return nil, errors.Wrapf(err, "invalid client address %q", ip)
+		}
+		parsedIP = parsedIP.To16()
+
+		return &ipMapEntry{
+			subnet: &net.IPNet{
+				IP:   parsedIP,
+				Mask: ipv6FullySpecifiedMask,
+			},
+			zone:  zone,
+			ident: ident,
+		}, nil
+	}
+
+	// expect CIDR and '*' placeholder expansion
+
+	if strings.Count(ident, "*") != 1 {
+		err = fmt.Errorf("CIDRs require 1 IP placeholder")
+		return nil, errors.Wrapf(err, "invalid client identity %q for IP %q", ident, ip)
+	}
+
+	longestIPAddr := net.IPAddr{
+		IP:   net.IP(bytes.Repeat([]byte{0xff}, net.IPv6len)),
+		Zone: strings.Repeat("i", unix.IFNAMSIZ),
+	}
+	longestIdent := zfsDatasetPathComponentCompatibleRepresentation(ident, &longestIPAddr)
+
+	if err := transport.ValidateClientIdentity(longestIdent); err != nil {
+		return nil, errors.Wrapf(err, "invalid client identity for IP %q", ip)
+	}
+
+	ones, _ := subnet.Mask.Size()
+	preExpansionAddrlen := len(subnet.IP) * 8
+	expanded := subnet.IP.To16()
+	postExpansionAddrlen := len(expanded) * 8
+	return &ipMapEntry{
+		subnet: &net.IPNet{
+			IP:   expanded,
+			Mask: net.CIDRMask(postExpansionAddrlen-preExpansionAddrlen+ones, postExpansionAddrlen),
+		},
+		zone:  zone,
+		ident: ident,
+	}, nil
+}
+
+func zfsDatasetPathComponentCompatibleRepresentation(identityWithWildcard string, addr *net.IPAddr) string {
+	// If a Zone exists we append it after the IP using a "-" because "%"
+	// is a zfs dataset forbidden char.
+	if addr.Zone != "" {
+		return strings.Replace(identityWithWildcard, "*", addr.IP.String()+"-"+addr.Zone, 1)
+	}
+	// newIPMapEntry validates that the line contains exactly one "*"
+	return strings.Replace(identityWithWildcard, "*", addr.IP.String(), 1)
+}
+
+func (e *ipMapEntry) String() string {
+	return fmt.Sprintf("&ipMapEntry{subnet=%q, ident=%q, zone=%q}", e.subnet.String(), e.ident, e.zone)
+}
+
+func (e *ipMapEntry) PrefixLen() int {
+	ones, bits := e.subnet.Mask.Size()
+	if bits != net.IPv6len*8 {
+		panic(fmt.Sprintf("impl error: we represent all addresses as 16byte internally ones=%v bits=%v: %s", ones, bits, e))
+	}
+	return ones
+}
+
+// newtype to support sorting by prefixlength
+type byPrefixlen []*ipMapEntry
+
+func (m byPrefixlen) Len() int { return len(m) }
+func (m byPrefixlen) Less(i, j int) bool {
+
+	if m[i].PrefixLen() != m[j].PrefixLen() {
+		return m[i].PrefixLen() > m[j].PrefixLen()
+	}
+
+	addrCmp := bytes.Compare(m[i].subnet.IP.To16(), m[j].subnet.IP.To16())
+	if addrCmp != 0 {
+		return addrCmp < 0
+	}
+
+	return strings.Compare(m[i].zone, m[j].zone) < 0
+}
+func (m byPrefixlen) Swap(i, j int) { m[i], m[j] = m[j], m[i] }

transport/tcp/serve_tcp_ipmap_test.go

@@ -0,0 +1,204 @@
+package tcp
+
+import (
+	"net"
+	"os"
+	"testing"
+
+	"github.com/kr/pretty"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestIPMap(t *testing.T) {
+	type testCaseExpect struct {
+		expectNoMapping bool
+		expectIdent     string
+	}
+	type testCase struct {
+		name          string
+		config        map[string]string
+		expectInitErr bool
+		expect        map[string]testCaseExpect
+	}
+
+	cases := []testCase{
+		{
+			name:          "regular ips",
+			expectInitErr: false,
+			config: map[string]string{
+				"192.168.123.234": "ident1",
+				"192.168.20.23":   "ident2",
+			},
+			expect: map[string]testCaseExpect{
+				"192.168.123.234": {expectIdent: "ident1"},
+				"192.168.20.23":   {expectIdent: "ident2"},
+				"192.168.20.10":   {expectNoMapping: true},
+			},
+		},
+		{
+			name:          "full wildcard",
+			expectInitErr: false,
+			config: map[string]string{
+				"0.0.0.0/0": "*",
+				"0::/0":     "*",
+			},
+			expect: map[string]testCaseExpect{
+				"10.123.234.24": {expectIdent: "10.123.234.24"},
+				// '[' and ']' are forbiddenin dataset names
+				"fe80::23:42": {expectIdent: "fe80::23:42"},
+			},
+		},
+		{
+			name:          "longest prefix matching",
+			expectInitErr: false,
+			config: map[string]string{
+				"10.1.2.3":              "specific-host",
+				"10.1.2.0/24":           "subnet-one-two-*",
+				"10.1.1.0/24":           "subnet-one-one-*",
+				"10.1.0.0/24":           "subnet-one-zero-*",
+				"10.1.0.0/16":           "subnet-one-*",
+				"fde4:8dba:82e1:1::1":   "v6-48-1-specialhost",
+				"fde4:8dba:82e1::/48":   "v6-48-*",
+				"fde4:8dba:82e1:1::/64": "v6-64-1-*",
+				"fde4:8dba:82e1:2::/64": "v6-64-2-*",
+			},
+			expect: map[string]testCaseExpect{
+				"10.1.2.3":             {expectIdent: "specific-host"},
+				"10.1.2.1":             {expectIdent: "subnet-one-two-10.1.2.1"},
+				"10.1.1.1":             {expectIdent: "subnet-one-one-10.1.1.1"},
+				"10.1.0.23":            {expectIdent: "subnet-one-zero-10.1.0.23"},
+				"10.1.3.1":             {expectIdent: "subnet-one-10.1.3.1"},
+				"10.2.1.1":             {expectNoMapping: true},
+				"fde4:8dba:82e1:1::1":  {expectIdent: "v6-48-1-specialhost"},
+				"fde4:8dba:82e1:23::1": {expectIdent: "v6-48-fde4:8dba:82e1:23::1"},
+				"fde4:8dba:82e1:1::2":  {expectIdent: "v6-64-1-fde4:8dba:82e1:1::2"},
+				"fde4:8dba:82e1:2::1":  {expectIdent: "v6-64-2-fde4:8dba:82e1:2::1"},
+				"fde4:8dba:82e2::1":    {expectNoMapping: true},
+			},
+		},
+		{
+			name:          "different prefixes, mixed ipv4 ipv6, with interface ids",
+			expectInitErr: false,
+			config: map[string]string{
+				"192.168.23.0/24": "db-*",
+				"192.168.23.23":   "db-twentythree",
+				"192.168.42.0/24": "web-*",
+				"10.1.4.0/24":     "my-*-server",
+				"2001:0db8:85a3:0000:0000:8a2e:0370:7334":      "aspecifichost",
+				"2001:0db8:85a3:0000:0000:8a2e:0370:7334%eth1": "aspecifichost",
+				"fe80::/16%eth1":      "san-*",
+				"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"},
+
+				// 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},
+
+				// v6 subnet matching
+				"fde4:8dba:82e1::1": {expectIdent: "sub64-fde4:8dba:82e1::1"},
+				// v6 subnet matching with suffix that matches another allowed IPv4
+				"fde4:8dba:82e1::c0a8:1717": {expectIdent: "sub64-fde4:8dba:82e1::c0a8:1717"},
+
+				"2001:0db8:85a3:0000:0000:8a2e:0370:7334":      {expectIdent: "aspecifichost"},
+				"2001:0db8:85a3:0000:0000:8a2e:0370:7334%eth1": {expectIdent: "aspecifichost"},
+				"2001:0db8:85a3:0000:0000:8a2e:0370:7334%eth2": {expectNoMapping: true},
+			},
+		},
+		{
+			name:          "invalid user input: non ip or cidr",
+			expectInitErr: true,
+			config: map[string]string{
+				"jimmy": "db",
+			},
+		},
+		{
+			name:          "invalid user input: v4 ip with an identity containing *",
+			expectInitErr: true,
+			config: map[string]string{
+				"192.168.1.2": "db-*",
+			},
+		},
+		{
+			name:          "invalid user input: v4 subnet without an identity containing *",
+			expectInitErr: true,
+			config: map[string]string{
+				"192.168.1.0/24": "db-",
+			},
+		},
+		{
+			name:          "invalid user input: v6 ip with an identity containing *",
+			expectInitErr: true,
+			config: map[string]string{
+				"2001:0db8:85a3:0000:0000:8a2e:0370:7334": "aspecifichost*",
+			},
+		},
+		{
+			name:          "invalid user input: v6 subnet without identity containing *",
+			expectInitErr: true,
+			config: map[string]string{
+				"fe80::/16%eth1": "db-",
+			},
+		},
+		{
+			name:          "invalid user input with subnet match: client identity with forbidden zfs dataset name char @",
+			expectInitErr: true,
+			config: map[string]string{
+				"fe80::/16": "db@-*",
+			},
+		},
+		{
+			name:          "invalid user input with IP match: client identity with forbidden zfs dataset name char @",
+			expectInitErr: true,
+			config: map[string]string{
+				"fe80::1": "db@foo",
+			},
+		},
+	}
+
+	for i := range cases {
+		c := cases[i]
+		t.Run(c.name, func(t *testing.T) {
+			pretty.Fprintf(os.Stderr, "running %#v\n", c)
+			m, err := ipMapFromConfig(c.config)
+			if c.expectInitErr {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.NotNil(t, m)
+			}
+			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")
+				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")
+				require.NotNil(t, ipMapEntry)
+				ident, err := m.Get(&net.IPAddr{
+					IP:   ipMapEntry.subnet.IP,
+					Zone: ipMapEntry.zone,
+				})
+				if expect.expectNoMapping {
+					assert.Empty(t, ident)
+				} else {
+					assert.NoError(t, err)
+					assert.Equal(t, expect.expectIdent, ident)
+				}
+			}
+		})
+	}
+}
+
+func TestPackageNetAssumptions(t *testing.T) {
+
+}

transport/transport.go

@@ -4,10 +4,10 @@ package transport
 
 import (
 	"context"
-	"errors"
 	"net"
 	"syscall"
 
+	"github.com/pkg/errors"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/rpc/dataconn/timeoutconn"
 	"github.com/zrepl/zrepl/zfs"
@@ -55,13 +55,10 @@ type Connecter interface {
 }
 
 // A client identity must be a single component in a ZFS filesystem path
-func ValidateClientIdentity(in string) (err error) {
+func ValidateClientIdentity(in string) error {
-	path, err := zfs.NewDatasetPath(in)
+	err := zfs.ComponentNamecheck(in)
 	if err != nil {
-		return err
+		return errors.Wrap(err, "client identity must be usable as a single dataset path component")
-	}
-	if path.Length() != 1 {
-		return errors.New("client identity must be a single path component (not empty, no '/')")
 	}
 	return nil
 }

util/bytecounter/bytecounter_readcloser.go

@@ -1,39 +0,0 @@
-package bytecounter
-
-import (
-	"io"
-	"sync/atomic"
-)
-
-// ReadCloser wraps an io.ReadCloser, reimplementing
-// its interface and counting the bytes written to during copying.
-type ReadCloser interface {
-	io.ReadCloser
-	Count() int64
-}
-
-// NewReadCloser wraps rc.
-func NewReadCloser(rc io.ReadCloser) ReadCloser {
-	return &readCloser{rc, 0}
-}
-
-type readCloser struct {
-	rc    io.ReadCloser
-	count int64
-}
-
-func (r *readCloser) Count() int64 {
-	return atomic.LoadInt64(&r.count)
-}
-
-var _ io.ReadCloser = &readCloser{}
-
-func (r *readCloser) Close() error {
-	return r.rc.Close()
-}
-
-func (r *readCloser) Read(p []byte) (int, error) {
-	n, err := r.rc.Read(p)
-	atomic.AddInt64(&r.count, int64(n))
-	return n, err
-}

util/bytecounter/bytecounter_reader.go

@@ -0,0 +1,49 @@
+package bytecounter
+
+import (
+	"io"
+	"sync/atomic"
+	"time"
+)
+
+type ByteCounterReader struct {
+	reader io.ReadCloser
+
+	// called & accessed synchronously during Read, no external access
+	cb       func(full int64)
+	cbEvery  time.Duration
+	lastCbAt time.Time
+
+	// set atomically because it may be read by multiple threads
+	bytes int64
+}
+
+func NewByteCounterReader(reader io.ReadCloser) *ByteCounterReader {
+	return &ByteCounterReader{
+		reader: reader,
+	}
+}
+
+func (b *ByteCounterReader) SetCallback(every time.Duration, cb func(full int64)) {
+	b.cbEvery = every
+	b.cb = cb
+}
+
+func (b *ByteCounterReader) Close() error {
+	return b.reader.Close()
+}
+
+func (b *ByteCounterReader) Read(p []byte) (n int, err error) {
+	n, err = b.reader.Read(p)
+	full := atomic.AddInt64(&b.bytes, int64(n))
+	now := time.Now()
+	if b.cb != nil && now.Sub(b.lastCbAt) > b.cbEvery {
+		b.cb(full)
+		b.lastCbAt = now
+	}
+	return n, err
+}
+
+func (b *ByteCounterReader) Bytes() int64 {
+	return atomic.LoadInt64(&b.bytes)
+}

util/bytecounter/bytecounter_streamcopier.go

@@ -0,0 +1,71 @@
+package bytecounter
+
+import (
+	"io"
+	"sync/atomic"
+
+	"github.com/zrepl/zrepl/zfs"
+)
+
+// StreamCopier wraps a zfs.StreamCopier, reimplementing
+// its interface and counting the bytes written to during copying.
+type StreamCopier interface {
+	zfs.StreamCopier
+	Count() int64
+}
+
+// NewStreamCopier wraps sc into a StreamCopier.
+// If sc is io.Reader, it is guaranteed that the returned StreamCopier
+// implements that interface, too.
+func NewStreamCopier(sc zfs.StreamCopier) StreamCopier {
+	bsc := &streamCopier{sc, 0}
+	if scr, ok := sc.(io.Reader); ok {
+		return streamCopierAndReader{bsc, scr}
+	} else {
+		return bsc
+	}
+}
+
+type streamCopier struct {
+	sc    zfs.StreamCopier
+	count int64
+}
+
+// proxy writer used by streamCopier
+type streamCopierWriter struct {
+	parent *streamCopier
+	w      io.Writer
+}
+
+func (w streamCopierWriter) Write(p []byte) (n int, err error) {
+	n, err = w.w.Write(p)
+	atomic.AddInt64(&w.parent.count, int64(n))
+	return
+}
+
+func (s *streamCopier) Count() int64 {
+	return atomic.LoadInt64(&s.count)
+}
+
+var _ zfs.StreamCopier = &streamCopier{}
+
+func (s streamCopier) Close() error {
+	return s.sc.Close()
+}
+
+func (s *streamCopier) WriteStreamTo(w io.Writer) zfs.StreamCopierError {
+	ww := streamCopierWriter{s, w}
+	return s.sc.WriteStreamTo(ww)
+}
+
+// a streamCopier whose underlying sc is an io.Reader
+type streamCopierAndReader struct {
+	*streamCopier
+	asReader io.Reader
+}
+
+func (scr streamCopierAndReader) Read(p []byte) (int, error) {
+	n, err := scr.asReader.Read(p)
+	atomic.AddInt64(&scr.streamCopier.count, int64(n))
+	return n, err
+}

util/bytecounter/bytecounter_streamcopier_test.go

@@ -0,0 +1,39 @@
+package bytecounter
+
+import (
+	"io"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/zrepl/zrepl/zfs"
+)
+
+type mockStreamCopierAndReader struct {
+	zfs.StreamCopier // to satisfy interface
+	reads            int
+}
+
+func (r *mockStreamCopierAndReader) Read(p []byte) (int, error) {
+	r.reads++
+	return len(p), nil
+}
+
+var _ io.Reader = &mockStreamCopierAndReader{}
+
+func TestNewStreamCopierReexportsReader(t *testing.T) {
+	mock := &mockStreamCopierAndReader{}
+	x := NewStreamCopier(mock)
+
+	r, ok := x.(io.Reader)
+	if !ok {
+		t.Fatalf("%T does not implement io.Reader, hence reader cannout have been wrapped", x)
+	}
+
+	var buf [23]byte
+	n, err := r.Read(buf[:])
+	assert.True(t, mock.reads == 1)
+	assert.True(t, n == len(buf))
+	assert.NoError(t, err)
+	assert.True(t, x.Count() == 23)
+}

util/chainedio/chainedio_readcloser.go

@@ -1,43 +0,0 @@
-package chainedio
-
-import "io"
-
-type ChainedReadCloser struct {
-	readers   []io.Reader
-	curReader int
-}
-
-func NewChainedReader(reader ...io.Reader) *ChainedReadCloser {
-	return &ChainedReadCloser{
-		readers:   reader,
-		curReader: 0,
-	}
-}
-
-func (c *ChainedReadCloser) Read(buf []byte) (n int, err error) {
-
-	n = 0
-
-	for c.curReader < len(c.readers) {
-		n, err = c.readers[c.curReader].Read(buf)
-		if err == io.EOF {
-			c.curReader++
-			continue
-		}
-		break
-	}
-	if c.curReader == len(c.readers) {
-		err = io.EOF // actually, there was no gap
-	}
-
-	return
-}
-
-func (c *ChainedReadCloser) Close() error {
-	for _, r := range c.readers {
-		if c, ok := r.(io.Closer); ok {
-			c.Close() // TODO debug log error?
-		}
-	}
-	return nil
-}

util/chainedio/chainedio_reader.go

@@ -0,0 +1,34 @@
+package chainedio
+
+import "io"
+
+type ChainedReader struct {
+	Readers   []io.Reader
+	curReader int
+}
+
+func NewChainedReader(reader ...io.Reader) *ChainedReader {
+	return &ChainedReader{
+		Readers:   reader,
+		curReader: 0,
+	}
+}
+
+func (c *ChainedReader) Read(buf []byte) (n int, err error) {
+
+	n = 0
+
+	for c.curReader < len(c.Readers) {
+		n, err = c.Readers[c.curReader].Read(buf)
+		if err == io.EOF {
+			c.curReader++
+			continue
+		}
+		break
+	}
+	if c.curReader == len(c.Readers) {
+		err = io.EOF // actually, there was no gap
+	}
+
+	return
+}

util/chainlock/chainlock.go

@@ -40,8 +40,3 @@ func (l *L) DropWhile(f func()) {
 	defer l.Unlock().Lock()
 	f()
 }
-
-func (l *L) HoldWhile(f func()) {
-	defer l.Lock().Unlock()
-	f()
-}

zfs/zfs.go

@@ -354,6 +354,63 @@ func (a ZFSSendArgsUnvalidated) buildCommonSendArgs() ([]string, error) {
 	return args, nil
 }
 
+type ReadCloserCopier struct {
+	recorder readErrRecorder
+}
+
+type readErrRecorder struct {
+	io.ReadCloser
+	readErr error
+}
+
+type sendStreamCopierError struct {
+	isReadErr bool // if false, it's a write error
+	err       error
+}
+
+func (e sendStreamCopierError) Error() string {
+	if e.isReadErr {
+		return fmt.Sprintf("stream: read error: %s", e.err)
+	} else {
+		return fmt.Sprintf("stream: writer error: %s", e.err)
+	}
+}
+
+func (e sendStreamCopierError) IsReadError() bool  { return e.isReadErr }
+func (e sendStreamCopierError) IsWriteError() bool { return !e.isReadErr }
+
+func (r *readErrRecorder) Read(p []byte) (n int, err error) {
+	n, err = r.ReadCloser.Read(p)
+	r.readErr = err
+	return n, err
+}
+
+func NewReadCloserCopier(stream io.ReadCloser) *ReadCloserCopier {
+	return &ReadCloserCopier{recorder: readErrRecorder{stream, nil}}
+}
+
+func (c *ReadCloserCopier) WriteStreamTo(w io.Writer) StreamCopierError {
+	debug("sendStreamCopier.WriteStreamTo: begin")
+	_, err := io.Copy(w, &c.recorder)
+	debug("sendStreamCopier.WriteStreamTo: copy done")
+	if err != nil {
+		if c.recorder.readErr != nil {
+			return sendStreamCopierError{isReadErr: true, err: c.recorder.readErr}
+		} else {
+			return sendStreamCopierError{isReadErr: false, err: err}
+		}
+	}
+	return nil
+}
+
+func (c *ReadCloserCopier) Read(p []byte) (n int, err error) {
+	return c.recorder.Read(p)
+}
+
+func (c *ReadCloserCopier) Close() error {
+	return c.recorder.ReadCloser.Close()
+}
+
 func pipeWithCapacityHint(capacity int) (r, w *os.File, err error) {
 	if capacity <= 0 {
 		panic(fmt.Sprintf("capacity must be positive %v", capacity))
@@ -366,7 +423,7 @@ func pipeWithCapacityHint(capacity int) (r, w *os.File, err error) {
 	return stdoutReader, stdoutWriter, nil
 }
 
-type SendStream struct {
+type sendStream struct {
 	cmd  *zfscmd.Cmd
 	kill context.CancelFunc
 
@@ -376,7 +433,7 @@ type SendStream struct {
 	opErr        error
 }
 
-func (s *SendStream) Read(p []byte) (n int, err error) {
+func (s *sendStream) Read(p []byte) (n int, err error) {
 	s.closeMtx.Lock()
 	opErr := s.opErr
 	s.closeMtx.Unlock()
@@ -397,12 +454,12 @@ func (s *SendStream) Read(p []byte) (n int, err error) {
 	return n, err
 }
 
-func (s *SendStream) Close() error {
+func (s *sendStream) Close() error {
 	debug("sendStream: close called")
 	return s.killAndWait(nil)
 }
 
-func (s *SendStream) killAndWait(precedingReadErr error) error {
+func (s *sendStream) killAndWait(precedingReadErr error) error {
 
 	debug("sendStream: killAndWait enter")
 	defer debug("sendStream: killAndWait leave")
@@ -773,7 +830,7 @@ var ErrEncryptedSendNotSupported = fmt.Errorf("raw sends which are required for
 // (if from is "" a full ZFS send is done)
 //
 // Returns ErrEncryptedSendNotSupported if encrypted send is requested but not supported by CLI
-func ZFSSend(ctx context.Context, sendArgs ZFSSendArgsValidated) (*SendStream, error) {
+func ZFSSend(ctx context.Context, sendArgs ZFSSendArgsValidated) (*ReadCloserCopier, error) {
 
 	args := make([]string, 0)
 	args = append(args, "send")
@@ -822,14 +879,14 @@ func ZFSSend(ctx context.Context, sendArgs ZFSSendArgsValidated) (*SendStream, e
 	// close our writing-end of the pipe so that we don't wait for ourselves when reading from the reading  end
 	stdoutWriter.Close()
 
-	stream := &SendStream{
+	stream := &sendStream{
 		cmd:          cmd,
 		kill:         cancel,
 		stdoutReader: stdoutReader,
 		stderrBuf:    stderrBuf,
 	}
 
-	return stream, nil
+	return NewReadCloserCopier(stream), nil
 }
 
 type DrySendType string
@@ -968,6 +1025,24 @@ func ZFSSendDry(ctx context.Context, sendArgs ZFSSendArgsValidated) (_ *DrySendI
 	return &si, nil
 }
 
+type StreamCopierError interface {
+	error
+	IsReadError() bool
+	IsWriteError() bool
+}
+
+type StreamCopier interface {
+	// WriteStreamTo writes the stream represented by this StreamCopier
+	// to the given io.Writer.
+	WriteStreamTo(w io.Writer) StreamCopierError
+	// Close must be called as soon as it is clear that no more data will
+	// be read from the StreamCopier.
+	// If StreamCopier gets its data from a connection, it might hold
+	// a lock on the connection until Close is called. Only closing ensures
+	// that the connection can be used afterwards.
+	Close() error
+}
+
 type RecvOptions struct {
 	// Rollback to the oldest snapshot, destroy it, then perform `recv -F`.
 	// Note that this doesn't change property values, i.e. an existing local property value will be kept.
@@ -992,9 +1067,7 @@ func (e *ErrRecvResumeNotSupported) Error() string {
 	return buf.String()
 }
 
-const RecvStderrBufSiz = 1 << 15
+func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, streamCopier StreamCopier, opts RecvOptions) (err error) {
-
-func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, stream io.ReadCloser, opts RecvOptions) (err error) {
 
 	if err := v.ValidateInMemory(fs); err != nil {
 		return errors.Wrap(err, "invalid version")
@@ -1061,7 +1134,7 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, stream io.Rea
 	//   cannot receive new filesystem stream: invalid backup stream
 	stdout := bytes.NewBuffer(make([]byte, 0, 1024))
 
-	stderr := bytes.NewBuffer(make([]byte, 0, RecvStderrBufSiz))
+	stderr := bytes.NewBuffer(make([]byte, 0, 1024))
 
 	stdin, stdinWriter, err := pipeWithCapacityHint(ZFSRecvPipeCapacityHint)
 	if err != nil {
@@ -1089,10 +1162,9 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, stream io.Rea
 
 	debug("started")
 
-	copierErrChan := make(chan error)
+	copierErrChan := make(chan StreamCopierError)
 	go func() {
-		_, err := io.Copy(stdinWriter, stream)
+		copierErrChan <- streamCopier.WriteStreamTo(stdinWriter)
-		copierErrChan <- err
 		stdinWriter.Close()
 	}()
 	waitErrChan := make(chan error)
@@ -1101,10 +1173,6 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, stream io.Rea
 		if err = cmd.Wait(); err != nil {
 			if rtErr := tryRecvErrorWithResumeToken(ctx, stderr.String()); rtErr != nil {
 				waitErrChan <- rtErr
-			} else if owErr := tryRecvDestroyOrOverwriteEncryptedErr(stderr.Bytes()); owErr != nil {
-				waitErrChan <- owErr
-			} else if readErr := tryRecvCannotReadFromStreamErr(stderr.Bytes()); readErr != nil {
-				waitErrChan <- readErr
 			} else {
 				waitErrChan <- &ZFSError{
 					Stderr:  stderr.Bytes(),
@@ -1115,23 +1183,22 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, stream io.Rea
 		}
 	}()
 
+	// streamCopier always fails before or simultaneously with Wait
+	// thus receive from it first
 	copierErr := <-copierErrChan
 	debug("copierErr: %T %s", copierErr, copierErr)
 	if copierErr != nil {
-		debug("killing zfs recv command after copierErr")
 		cancelCmd()
 	}
 
 	waitErr := <-waitErrChan
 	debug("waitErr: %T %s", waitErr, waitErr)
-
 	if copierErr == nil && waitErr == nil {
 		return nil
-	} else if _, isReadErr := waitErr.(*RecvCannotReadFromStreamErr); isReadErr {
+	} else if waitErr != nil && (copierErr == nil || copierErr.IsWriteError()) {
-		return copierErr // likely network error reading from stream
+		return waitErr // has more interesting info in that case
-	} else {
-		return waitErr // almost always more interesting info. NOTE: do not wrap!
 	}
+	return copierErr // if it's not a write error, the copier error is more interesting
 }
 
 type RecvFailedWithResumeTokenErr struct {
@@ -1161,43 +1228,6 @@ func (e *RecvFailedWithResumeTokenErr) Error() string {
 	return fmt.Sprintf("receive failed, resume token available: %s\n%#v", e.ResumeTokenRaw, e.ResumeTokenParsed)
 }
 
-type RecvDestroyOrOverwriteEncryptedErr struct {
-	Msg string
-}
-
-func (e *RecvDestroyOrOverwriteEncryptedErr) Error() string {
-	return e.Msg
-}
-
-var recvDestroyOrOverwriteEncryptedErrRe = regexp.MustCompile(`^(cannot receive new filesystem stream: zfs receive -F cannot be used to destroy an encrypted filesystem or overwrite an unencrypted one with an encrypted one)`)
-
-func tryRecvDestroyOrOverwriteEncryptedErr(stderr []byte) *RecvDestroyOrOverwriteEncryptedErr {
-	debug("tryRecvDestroyOrOverwriteEncryptedErr: %v", stderr)
-	m := recvDestroyOrOverwriteEncryptedErrRe.FindSubmatch(stderr)
-	if m == nil {
-		return nil
-	}
-	return &RecvDestroyOrOverwriteEncryptedErr{Msg: string(m[1])}
-}
-
-type RecvCannotReadFromStreamErr struct {
-	Msg string
-}
-
-func (e *RecvCannotReadFromStreamErr) Error() string {
-	return e.Msg
-}
-
-var reRecvCannotReadFromStreamErr = regexp.MustCompile(`^(cannot receive: failed to read from stream)$`)
-
-func tryRecvCannotReadFromStreamErr(stderr []byte) *RecvCannotReadFromStreamErr {
-	m := reRecvCannotReadFromStreamErr.FindSubmatch(stderr)
-	if m == nil {
-		return nil
-	}
-	return &RecvCannotReadFromStreamErr{Msg: string(m[1])}
-}
-
 type ClearResumeTokenError struct {
 	ZFSOutput []byte
 	CmdError  error

zfs/zfs_test.go

@@ -2,11 +2,9 @@ package zfs
 
 import (
 	"context"
-	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
 )
 
 func TestZFSListHandlesProducesZFSErrorOnNonZeroExit(t *testing.T) {
@@ -261,12 +259,3 @@ size	10518512
 		})
 	}
 }
-
-func TestTryRecvDestroyOrOverwriteEncryptedErr(t *testing.T) {
-	msg := "cannot receive new filesystem stream: zfs receive -F cannot be used to destroy an encrypted filesystem or overwrite an unencrypted one with an encrypted one\n"
-	assert.GreaterOrEqual(t, RecvStderrBufSiz, len(msg))
-
-	err := tryRecvDestroyOrOverwriteEncryptedErr([]byte(msg))
-	require.NotNil(t, err)
-	assert.EqualError(t, err, strings.TrimSpace(msg))
-}

zfs/zfscmd/zfscmd_platform_test.go

@@ -5,14 +5,11 @@ import (
 	"bytes"
 	"context"
 	"io"
-	"os"
 	"os/exec"
-	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"github.com/zrepl/zrepl/util/circlog"
 )
 
 const testBin = "./zfscmd_platform_test.bash"
@@ -88,37 +85,5 @@ func TestCmdProcessState(t *testing.T) {
 	require.True(t, ok)
 	require.NotNil(t, ee.ProcessState)
 	require.Contains(t, ee.Error(), "killed")
-}
-
-func TestSigpipe(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	cmd := CommandContext(ctx, "bash", "-c", "sleep 5; echo invalid input; exit 23")
-	r, w, err := os.Pipe()
-	require.NoError(t, err)
-	output := circlog.MustNewCircularLog(1 << 20)
-	cmd.SetStdio(Stdio{
-		Stdin:  r,
-		Stdout: output,
-		Stderr: output,
-	})
-	err = cmd.Start()
-	require.NoError(t, err)
-	err = r.Close()
-	require.NoError(t, err)
-
-	// the script doesn't read stdin, but this input is almost certainly smaller than the pipe buffer
-	const LargerThanPipeBuffer = 1 << 21
-	_, err = io.Copy(w, bytes.NewBuffer(bytes.Repeat([]byte("i"), LargerThanPipeBuffer)))
-	// => io.Copy is going to block because the pipe buffer is full and the
-	//    script is not reading from it
-	// => the script is going to exit after 5s
-	// => we should expect a broken pipe error from the copier's perspective
-	t.Logf("copy err = %T: %s", err, err)
-	require.NotNil(t, err)
-	require.True(t, strings.Contains(err.Error(), "broken pipe"))
-
-	err = cmd.Wait()
-	require.EqualError(t, err, "exit status 23")
 
 }