DO NOT MERGE: timeoutconn: drop the only use of "unsafe" in the code base

Unless we're using the global source for math/rand, (*rand.Rand).Read
should not be called concurrently. We seed the rng in daemon.Run to
avoid ambiguity or hiding global side effects inside packages.

closes #414

daemon/daemon.go

@@ -3,6 +3,7 @@ package daemon
 import (
 	"context"
 	"fmt"
+	"math/rand"
 	"os"
 	"os/signal"
 	"strings"
@@ -38,6 +39,12 @@ func Run(ctx context.Context, conf *config.Config) error {
 		cancel()
 	}()
 
+	// The math/rand package is used presently for generating trace IDs, we
+	// seed it with the current time and pid so that the IDs are mostly
+	// unique.
+	rand.Seed(time.Now().UnixNano())
+	rand.Seed(int64(os.Getpid()))
+
 	outlets, err := logging.OutletsFromConfig(*conf.Global.Logging)
 	if err != nil {
 		return errors.Wrap(err, "cannot build logging from config")

daemon/job/snapjob.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"sort"
+	"sync"
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
@@ -29,7 +30,8 @@ type SnapJob struct {
 
 	promPruneSecs *prometheus.HistogramVec // labels: prune_side
 
-	pruner *pruner.Pruner
+	prunerMtx sync.Mutex
+	pruner    *pruner.Pruner
 }
 
 func (j *SnapJob) Name() string { return j.name.String() }
@@ -77,9 +79,11 @@ type SnapJobStatus struct {
 func (j *SnapJob) Status() *Status {
 	s := &SnapJobStatus{}
 	t := j.Type()
+	j.prunerMtx.Lock()
 	if j.pruner != nil {
 		s.Pruning = j.pruner.Report()
 	}
+	j.prunerMtx.Unlock()
 	s.Snapshotting = j.snapper.Report()
 	return &Status{Type: t, JobSpecific: s}
 }
@@ -177,7 +181,9 @@ func (j *SnapJob) doPrune(ctx context.Context) {
 		// FIXME encryption setting is irrelevant for SnapJob because the endpoint is only used as pruner.Target
 		Encrypt: &zfs.NilBool{B: true},
 	})
+	j.prunerMtx.Lock()
 	j.pruner = j.prunerFactory.BuildLocalPruner(ctx, sender, alwaysUpToDateReplicationCursorHistory{sender})
+	j.prunerMtx.Unlock()
 	log.Info("start pruning")
 	j.pruner.Prune()
 	log.Info("finished pruning")

daemon/logging/trace/trace_genID.go

@@ -3,20 +3,11 @@ package trace
 import (
 	"encoding/base64"
 	"math/rand"
-	"os"
 	"strings"
-	"time"
 
 	"github.com/zrepl/zrepl/util/envconst"
 )
 
-var genIdPRNG = rand.New(rand.NewSource(1))
-
-func init() {
-	genIdPRNG.Seed(time.Now().UnixNano())
-	genIdPRNG.Seed(int64(os.Getpid()))
-}
-
 var genIdNumBytes = envconst.Int("ZREPL_TRACE_ID_NUM_BYTES", 3)
 
 func init() {
@@ -30,7 +21,7 @@ func genID() string {
 	enc := base64.NewEncoder(base64.RawStdEncoding, &out)
 	buf := make([]byte, genIdNumBytes)
 	for i := 0; i < len(buf); {
-		n, err := genIdPRNG.Read(buf[i:])
+		n, err := rand.Read(buf[i:])
 		if err != nil {
 			panic(err)
 		}

go.mod

@@ -10,7 +10,6 @@ require (
 	github.com/go-sql-driver/mysql v1.4.1-0.20190907122137-b2c03bcae3d4
 	github.com/golang/protobuf v1.3.2
 	github.com/google/uuid v1.1.1
-	github.com/hashicorp/yamux v0.0.0-20200609203250-aecfd211c9ce
 	github.com/jinzhu/copier v0.0.0-20170922082739-db4671f3a9b8
 	github.com/kr/pretty v0.1.0
 	github.com/lib/pq v1.2.0
@@ -34,7 +33,6 @@ require (
 	github.com/yudai/gojsondiff v0.0.0-20170107030110-7b1b7adf999d
 	github.com/yudai/golcs v0.0.0-20170316035057-ecda9a501e82 // go1.12 thinks it needs this
 	github.com/zrepl/yaml-config v0.0.0-20191220194647-cbb6b0cf4bdd
-	golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2
 	golang.org/x/net v0.0.0-20190613194153-d28f0bde5980
 	golang.org/x/sync v0.0.0-20190423024810-112230192c58
 	golang.org/x/sys v0.0.0-20191026070338-33540a1f6037
@@ -42,5 +40,3 @@ require (
 	gonum.org/v1/gonum v0.7.0 // indirect
 	google.golang.org/grpc v1.17.0
 )
-
-replace github.com/problame/go-netssh => /home/cs/zrepl/go-netssh

go.sum

@@ -120,8 +120,6 @@ github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+
 github.com/gostaticanalysis/analysisutil v0.0.0-20190318220348-4088753ea4d3/go.mod h1:eEOZF4jCKGi+aprrirO9e7WKB3beBRtWgqGunKl6pKE=
 github.com/hashicorp/hcl v0.0.0-20180404174102-ef8a98b0bbce/go.mod h1:oZtUIOe8dh44I2q6ScRibXws4Ajl+d+nod3AaR9vL5w=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
-github.com/hashicorp/yamux v0.0.0-20200609203250-aecfd211c9ce h1:7UnVY3T/ZnHUrfviiAgIUjg2PXxsQfs5bphsG8F7Keo=
-github.com/hashicorp/yamux v0.0.0-20200609203250-aecfd211c9ce/go.mod h1:+NfK9FKeTrX5uv1uIXGdwYDTeHna2qgaIlx54MXqjAM=
 github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
 github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM=
@@ -302,7 +300,6 @@ github.com/zrepl/yaml-config v0.0.0-20191220194647-cbb6b0cf4bdd/go.mod h1:JmNwis
 github.com/zrepl/zrepl v0.2.0/go.mod h1:M3Zv2IGSO8iYpUjsZD6ayZ2LHy7zyMfzet9XatKOrZ8=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
-golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2 h1:VklqNMn3ovrHsnt90PveolxSbWFaJdECFbxSq0Mqo2M=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190313024323-a1f597ede03a/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190325154230-a5d413f7728c/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=

lazy.sh

@@ -39,12 +39,7 @@ godep() {
     export GOOS="$GOHOSTOS"
     export GOARCH="$GOHOSTARCH"
     # TODO GOARM=$GOHOSTARM?
-    go build -v -mod=readonly -o "$GOPATH/bin/stringer"      golang.org/x/tools/cmd/stringer
-    go build -v -mod=readonly -o "$GOPATH/bin/protoc-gen-go" github.com/golang/protobuf/protoc-gen-go
-    go build -v -mod=readonly -o "$GOPATH/bin/enumer"        github.com/alvaroloes/enumer
-    go build -v -mod=readonly -o "$GOPATH/bin/goimports"     golang.org/x/tools/cmd/goimports
-    go build -v -mod=readonly -o "$GOPATH/bin/golangci-lint" github.com/golangci/golangci-lint/cmd/golangci-lint
-    go build -v -mod=readonly -o "$GOPATH/bin/gocovmerge"    github.com/wadey/gocovmerge
+    cat tools.go | grep _ | awk -F'"' '{print $2}' | tee | xargs -tI '{}' go install '{}'
     set +x
     popd
     if ! type stringer || ! type protoc-gen-go || ! type enumer || ! type goimports || ! type golangci-lint || ! type gocovmerge; then

replication/driver/replication_driver.go

@@ -492,57 +492,37 @@ func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 	// => 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
-	// up to including the originally planned target snapshot
+	// for not-first attempts that succeeded in planning, only allow fs.planned.steps
+	// up to and including the originally planned target snapshot
 	if prev != nil && prev.planning.done && prev.planning.err == nil {
+		f.debug("attempting to correlate plan with previous attempt to find out what is left to do")
+		// find the highest of the previously uncompleted steps for which we can also find a step
+		// in our current plan
 		prevUncompleted := prev.planned.steps[prev.planned.step:]
-		if len(prevUncompleted) == 0 {
-			f.debug("prevUncompleted is empty")
-			return
-		}
-		if len(f.planned.steps) == 0 {
-			f.debug("fs.planned.steps is empty")
-			return
-		}
-		prevFailed := prevUncompleted[0]
-		curFirst := f.planned.steps[0]
-		// we assume that PlanFS retries prevFailed (using curFirst)
-		if !prevFailed.step.TargetEquals(curFirst.step) {
-			f.debug("Targets don't match")
-			// Two options:
-			// A: planning algorithm is broken
-			// B: manual user intervention inbetween
-			// Neither way will we make progress, so let's error out
-			stepFmt := func(step *step) string {
-				r := step.report()
-				s := r.Info
-				if r.IsIncremental() {
-					return fmt.Sprintf("%s=>%s", s.From, s.To)
-				} else {
-					return fmt.Sprintf("full=>%s", s.To)
+		var target struct{ prev, cur int }
+		target.prev = -1
+		target.cur = -1
+	out:
+		for p := len(prevUncompleted) - 1; p >= 0; p-- {
+			for q := len(f.planned.steps) - 1; q >= 0; q-- {
+				if prevUncompleted[p].step.TargetEquals(f.planned.steps[q].step) {
+					target.prev = p
+					target.cur = q
+					break out
 				}
 			}
-			msg := fmt.Sprintf("last attempt's uncompleted step %s does not correspond to this attempt's first planned step %s",
-				stepFmt(prevFailed), stepFmt(curFirst))
-			f.planned.stepErr = newTimedError(errors.New(msg), time.Now())
+		}
+		if target.prev == -1 || target.cur == -1 {
+			f.debug("no correlation possible between previous attempt and this attempt's plan")
+			f.planning.err = newTimedError(fmt.Errorf("cannot correlate previously failed attempt to current plan"), time.Now())
 			return
 		}
-		// only allow until step targets diverge
-		min := len(prevUncompleted)
-		if min > len(f.planned.steps) {
-			min = len(f.planned.steps)
-		}
-		diverge := 0
-		for ; diverge < min; diverge++ {
-			f.debug("diverge compare iteration %d", diverge)
-			if !f.planned.steps[diverge].step.TargetEquals(prevUncompleted[diverge].step) {
-				break
-			}
-		}
-		f.debug("diverge is %d", diverge)
-		f.planned.steps = f.planned.steps[0:diverge]
+
+		f.planned.steps = f.planned.steps[0:target.cur]
+		f.debug("found correlation, new steps are len(fs.planned.steps) = %d", len(f.planned.steps))
+	} else {
+		f.debug("previous attempt does not exist or did not finish planning, no correlation possible, taking this attempt's plan as is")
 	}
-	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

rpc/dataconn/frameconn/frameconn.go

@@ -47,7 +47,7 @@ func (f *FrameHeader) Unmarshal(buf []byte) {
 
 type Conn struct {
 	readMtx, writeMtx sync.Mutex
-	nc                timeoutconn.Conn
+	nc                *timeoutconn.Conn
 	readNextValid     bool
 	readNext          FrameHeader
 	nextReadErr       error
@@ -55,7 +55,7 @@ type Conn struct {
 	shutdown          shutdownFSM
 }
 
-func Wrap(nc timeoutconn.Conn) *Conn {
+func Wrap(nc *timeoutconn.Conn) *Conn {
 	return &Conn{
 		nc: nc,
 		//		ncBuf: bufio.NewReadWriter(bufio.NewReaderSize(nc, 1<<23), bufio.NewWriterSize(nc, 1<<23)),

rpc/dataconn/stream/stream_conn.go

@@ -144,6 +144,7 @@ func (c *Conn) ReadStream(frameType uint32, closeConnOnClose bool) (_ *StreamRea
 
 	c.readMtx.Lock()
 	if !c.readClean {
+		c.readMtx.Unlock()
 		return nil, errWriteStreamToErrorUnknownState
 	}
 

rpc/dataconn/timeoutconn/timeoutconn.go

@@ -11,7 +11,7 @@ import (
 	"errors"
 	"io"
 	"net"
-	"sync/atomic"
+	"sync"
 	"syscall"
 	"time"
 )
@@ -54,39 +54,60 @@ type Wire interface {
 }
 
 type Conn struct {
+	// immutable state
+
 	Wire
-	renewDeadlinesDisabled int32
-	idleTimeout            time.Duration
+	idleTimeout time.Duration
+
+	// mutable state (protected by mtx)
+
+	mtx                    sync.RWMutex
+	renewDeadlinesDisabled bool
 }
 
-func Wrap(conn Wire, idleTimeout time.Duration) Conn {
-	return Conn{Wire: conn, idleTimeout: idleTimeout}
+func Wrap(conn Wire, idleTimeout time.Duration) *Conn {
+	return &Conn{
+		Wire:        conn,
+		idleTimeout: idleTimeout,
+	}
 }
 
 // DisableTimeouts disables the idle timeout behavior provided by this package.
-// Existing deadlines are cleared iff the call is the first call to this method.
+// Existing deadlines are cleared iff the call is the first call to this method
+// or if the previous call produced an error.
 func (c *Conn) DisableTimeouts() error {
-	if atomic.CompareAndSwapInt32(&c.renewDeadlinesDisabled, 0, 1) {
-		return c.SetDeadline(time.Time{})
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+	if c.renewDeadlinesDisabled {
+		return nil
 	}
+	err := c.SetDeadline(time.Time{})
+	if err != nil {
+		return err
+	}
+	c.renewDeadlinesDisabled = true
 	return nil
 }
 
 func (c *Conn) renewReadDeadline() error {
-	if atomic.LoadInt32(&c.renewDeadlinesDisabled) != 0 {
+	c.mtx.RLock()
+	defer c.mtx.RUnlock()
+	if c.renewDeadlinesDisabled {
 		return nil
 	}
 	return c.SetReadDeadline(time.Now().Add(c.idleTimeout))
 }
 
 func (c *Conn) RenewWriteDeadline() error {
-	if atomic.LoadInt32(&c.renewDeadlinesDisabled) != 0 {
+	c.mtx.RLock()
+	defer c.mtx.RUnlock()
+	if c.renewDeadlinesDisabled {
 		return nil
 	}
 	return c.SetWriteDeadline(time.Now().Add(c.idleTimeout))
 }
 
-func (c Conn) Read(p []byte) (n int, err error) {
+func (c *Conn) Read(p []byte) (n int, err error) {
 	n = 0
 	err = nil
 restart:
@@ -103,7 +124,7 @@ restart:
 	return n, err
 }
 
-func (c Conn) Write(p []byte) (n int, err error) {
+func (c *Conn) Write(p []byte) (n int, err error) {
 	n = 0
 restart:
 	if err := c.RenewWriteDeadline(); err != nil {
@@ -123,7 +144,7 @@ restart:
 // but is guaranteed to use the writev system call if the wrapped Wire
 // support it.
 // Note the Conn does not support writev through io.Copy(aConn, aNetBuffers).
-func (c Conn) WritevFull(bufs net.Buffers) (n int64, err error) {
+func (c *Conn) WritevFull(bufs net.Buffers) (n int64, err error) {
 	n = 0
 restart:
 	if err := c.RenewWriteDeadline(); err != nil {
@@ -163,12 +184,12 @@ var _ SyscallConner = (*net.TCPConn)(nil)
 // If the connection returned io.EOF, the number of bytes written until
 // then + io.EOF is returned. This behavior is different to io.ReadFull
 // which returns io.ErrUnexpectedEOF.
-func (c Conn) ReadvFull(buffers net.Buffers) (n int64, err error) {
+func (c *Conn) ReadvFull(buffers net.Buffers) (n int64, err error) {
 	return c.readv(buffers)
 }
 
 // invoked by c.readv if readv system call cannot be used
-func (c Conn) readvFallback(nbuffers net.Buffers) (n int64, err error) {
+func (c *Conn) readvFallback(nbuffers net.Buffers) (n int64, err error) {
 	buffers := [][]byte(nbuffers)
 	for i := range buffers {
 		curBuf := buffers[i]

rpc/dataconn/timeoutconn/timeoutconn_readv_unsupported.go

@@ -1,10 +1,8 @@
-// +build illumos solaris
-
 package timeoutconn
 
 import "net"
 
-func (c Conn) readv(buffers net.Buffers) (n int64, err error) {
+func (c *Conn) readv(buffers net.Buffers) (n int64, err error) {
 	// Go does not expose the SYS_READV symbol for Solaris / Illumos - do they have it?
 	// Anyhow, use the fallback
 	return c.readvFallback(buffers)

rpc/dataconn/timeoutconn/timoutconn_readv.go

@@ -1,133 +0,0 @@
-// +build !illumos
-// +build !solaris
-
-package timeoutconn
-
-import (
-	"io"
-	"net"
-	"syscall"
-	"unsafe"
-)
-
-func buildIovecs(buffers net.Buffers) (totalLen int64, vecs []syscall.Iovec) {
-	vecs = make([]syscall.Iovec, 0, len(buffers))
-	for i := range buffers {
-		totalLen += int64(len(buffers[i]))
-		if len(buffers[i]) == 0 {
-			continue
-		}
-
-		v := syscall.Iovec{
-			Base: &buffers[i][0],
-		}
-		// syscall.Iovec.Len has platform-dependent size, thus use SetLen
-		v.SetLen(len(buffers[i]))
-
-		vecs = append(vecs, v)
-	}
-	return totalLen, vecs
-}
-
-func (c Conn) readv(buffers net.Buffers) (n int64, err error) {
-
-	scc, ok := c.Wire.(SyscallConner)
-	if !ok {
-		return c.readvFallback(buffers)
-	}
-	rawConn, err := scc.SyscallConn()
-	if err == SyscallConnNotSupported {
-		return c.readvFallback(buffers)
-	}
-	if err != nil {
-		return 0, err
-	}
-
-	_, iovecs := buildIovecs(buffers)
-
-	for len(iovecs) > 0 {
-		if err := c.renewReadDeadline(); err != nil {
-			return n, err
-		}
-		oneN, oneErr := c.doOneReadv(rawConn, &iovecs)
-		n += oneN
-		if netErr, ok := oneErr.(net.Error); ok && netErr.Timeout() && oneN > 0 { // TODO likely not working
-			continue
-		} else if oneErr == nil && oneN > 0 {
-			continue
-		} else {
-			return n, oneErr
-		}
-	}
-	return n, nil
-}
-
-func (c Conn) doOneReadv(rawConn syscall.RawConn, iovecs *[]syscall.Iovec) (n int64, err error) {
-	rawReadErr := rawConn.Read(func(fd uintptr) (done bool) {
-		// iovecs, n and err must not be shadowed!
-
-		// NOTE: unsafe.Pointer safety rules
-		// 		https://tip.golang.org/pkg/unsafe/#Pointer
-		//
-		//		(4) Conversion of a Pointer to a uintptr when calling syscall.Syscall.
-		// 		...
-		//		uintptr() conversions must appear within the syscall.Syscall argument list.
-		//      (even though we are not the escape analysis Likely not )
-		thisReadN, _, errno := syscall.Syscall(
-			syscall.SYS_READV,
-			fd,
-			uintptr(unsafe.Pointer(&(*iovecs)[0])),
-			uintptr(len(*iovecs)),
-		)
-		if thisReadN == ^uintptr(0) {
-			if errno == syscall.EAGAIN {
-				return false
-			}
-			err = syscall.Errno(errno)
-			return true
-		}
-		if int(thisReadN) < 0 {
-			panic("unexpected return value")
-		}
-		n += int64(thisReadN) // TODO check overflow
-
-		// shift iovecs forward
-		for left := int(thisReadN); left > 0; {
-			// conversion to uint does not change value, see TestIovecLenFieldIsMachineUint, and left > 0
-			thisIovecConsumedCompletely := uint((*iovecs)[0].Len) <= uint(left)
-			if thisIovecConsumedCompletely {
-				// Update left, cannot go below 0 due to
-				// a) definition of thisIovecConsumedCompletely
-				// b) left > 0 due to loop invariant
-				// Converting .Len to int64 is thus also safe now, because it is < left < INT_MAX
-				left -= int((*iovecs)[0].Len)
-				*iovecs = (*iovecs)[1:]
-			} else {
-				// trim this iovec to remaining length
-
-				// NOTE: unsafe.Pointer safety rules
-				// 		https://tip.golang.org/pkg/unsafe/#Pointer
-				// 		(3) Conversion of a Pointer to a uintptr and back, with arithmetic.
-				// 		...
-				//		Note that both conversions must appear in the same expression,
-				//		with only the intervening arithmetic between them:
-				(*iovecs)[0].Base = (*byte)(unsafe.Pointer(uintptr(unsafe.Pointer((*iovecs)[0].Base)) + uintptr(left)))
-				curVecNewLength := uint((*iovecs)[0].Len) - uint(left) // casts to uint do not change value
-				(*iovecs)[0].SetLen(int(curVecNewLength))              // int and uint have the same size, no change of value
-
-				break // inner
-			}
-		}
-		if thisReadN == 0 {
-			err = io.EOF
-			return true
-		}
-		return true
-	})
-
-	if rawReadErr != nil {
-		err = rawReadErr
-	}
-
-	return n, err
-}

zync/README.md

@@ -1,7 +0,0 @@
-A tool that re-used zrepl abstractions to be an rsync-like sync tool for ZFS.
-
-Test environment:
-
-```
-go build -o zync && rsync -a ./zync root@192.168.124.233:/usr/local/bin/ && sudo ./zync local:///rpool/zrepltlstest/src ssh://root:%2Fhome%2Fcs%2Fzrepl%2Fzrepl%2Fzync%2Ftestid@192.168.124.233/p1/zync_sink 
-```

zync/transport/sshdirect/dial.go

@@ -1,406 +0,0 @@
-package sshdirect
-
-import (
-	"bytes"
-	"context"
-	"fmt"
-	"io"
-	"net"
-	"os/exec"
-	"sync"
-	"syscall"
-	"time"
-
-	"github.com/hashicorp/yamux"
-	"github.com/zrepl/zrepl/transport"
-	"github.com/zrepl/zrepl/util/circlog"
-)
-
-type Endpoint struct {
-	Host         string
-	User         string
-	Port         uint16
-	IdentityFile string
-	SSHCommand   string
-	Options      []string
-	RunCommand   []string
-}
-
-func (e Endpoint) CmdArgs() (cmd string, args []string, env []string) {
-
-	if e.SSHCommand != "" {
-		cmd = e.SSHCommand
-	} else {
-		cmd = "ssh"
-	}
-
-	args = make([]string, 0, 2*len(e.Options)+4)
-	args = append(args,
-		"-p", fmt.Sprintf("%d", e.Port),
-		"-T",
-		"-i", e.IdentityFile,
-		"-o", "BatchMode=yes",
-	)
-	for _, option := range e.Options {
-		args = append(args, "-o", option)
-	}
-	args = append(args, fmt.Sprintf("%s@%s", e.User, e.Host))
-
-	args = append(args, e.RunCommand...)
-
-	env = []string{}
-
-	return
-}
-
-type SSHConn struct {
-	cmd    *exec.Cmd
-	stdin  io.WriteCloser
-	stdout io.ReadCloser
-
-	shutdownMtx    sync.Mutex
-	shutdownResult *shutdownResult // TODO not used anywhere
-	cmdCancel      context.CancelFunc
-}
-
-const go_network string = "netssh"
-
-type clientAddr struct {
-	pid int
-}
-
-func (a clientAddr) Network() string {
-	return go_network
-}
-
-func (a clientAddr) String() string {
-	return fmt.Sprintf("pid=%d", a.pid)
-}
-
-func (conn *SSHConn) LocalAddr() net.Addr {
-	proc := conn.cmd.Process
-	if proc == nil {
-		return clientAddr{-1}
-	}
-	return clientAddr{proc.Pid}
-}
-
-func (conn *SSHConn) RemoteAddr() net.Addr {
-	return conn.LocalAddr()
-}
-
-// Read implements io.Reader.
-// It returns *IOError for any non-nil error that is != io.EOF.
-func (conn *SSHConn) Read(p []byte) (int, error) {
-	n, err := conn.stdout.Read(p)
-	if err != nil && err != io.EOF {
-		return n, &IOError{err}
-	}
-	return n, err
-}
-
-// Write implements io.Writer.
-// It returns *IOError for any error != nil.
-func (conn *SSHConn) Write(p []byte) (int, error) {
-	n, err := conn.stdin.Write(p)
-	if err != nil {
-		return n, &IOError{err}
-	}
-	return n, err
-}
-
-func (conn *SSHConn) CloseWrite() error {
-	return conn.stdin.Close()
-}
-
-type deadliner interface {
-	SetReadDeadline(time.Time) error
-	SetWriteDeadline(time.Time) error
-}
-
-func (conn *SSHConn) SetReadDeadline(t time.Time) error {
-	// type assertion is covered by test TestExecCmdPipesDeadlineBehavior
-	return conn.stdout.(deadliner).SetReadDeadline(t)
-}
-
-func (conn *SSHConn) SetWriteDeadline(t time.Time) error {
-	// type assertion is covered by test TestExecCmdPipesDeadlineBehavior
-	return conn.stdin.(deadliner).SetWriteDeadline(t)
-}
-
-func (conn *SSHConn) SetDeadline(t time.Time) error {
-	// try both
-	rerr := conn.SetReadDeadline(t)
-	werr := conn.SetWriteDeadline(t)
-	if rerr != nil {
-		return rerr
-	}
-	if werr != nil {
-		return werr
-	}
-	return nil
-}
-
-func (conn *SSHConn) Close() error {
-	conn.shutdownProcess()
-	return nil // FIXME: waitError will be non-zero because we signaled it, shutdownProcess needs to distinguish that
-}
-
-type shutdownResult struct {
-	waitErr error
-}
-
-func (conn *SSHConn) shutdownProcess() *shutdownResult {
-	conn.shutdownMtx.Lock()
-	defer conn.shutdownMtx.Unlock()
-
-	if conn.shutdownResult != nil {
-		return conn.shutdownResult
-	}
-
-	wait := make(chan error, 1)
-	go func() {
-		if err := conn.cmd.Process.Signal(syscall.SIGTERM); err != nil {
-			// TODO log error
-			return
-		}
-		wait <- conn.cmd.Wait()
-	}()
-
-	timeout := time.NewTimer(1 * time.Second) // FIXME const
-	defer timeout.Stop()
-
-	select {
-	case waitErr := <-wait:
-		conn.shutdownResult = &shutdownResult{waitErr}
-	case <-timeout.C:
-		conn.cmdCancel()
-		waitErr := <-wait // reuse existing Wait invocation, must not call twice
-		conn.shutdownResult = &shutdownResult{waitErr}
-	}
-	return conn.shutdownResult
-}
-
-// Cmd returns the underlying *exec.Cmd (the ssh client process)
-// Use read-only, should not be necessary for regular users.
-func (conn *SSHConn) Cmd() *exec.Cmd {
-	return conn.cmd
-}
-
-// CmdCancel bypasses the normal shutdown mechanism of SSHConn
-// (that is, calling Close) and cancels the process's context,
-// which usually results in SIGKILL being sent to the process.
-// Intended for integration tests, regular users shouldn't use it.
-func (conn *SSHConn) CmdCancel() {
-	conn.cmdCancel()
-}
-
-const bannerMessageLen = 31
-
-var messages = make(map[string][]byte)
-
-func mustMessage(str string) []byte {
-	if len(str) > bannerMessageLen {
-		panic("message length must be smaller than bannerMessageLen")
-	}
-	if _, ok := messages[str]; ok {
-		panic("duplicate message")
-	}
-	var buf bytes.Buffer
-	n, _ := buf.WriteString(str)
-	if n != len(str) {
-		panic("message must only contain ascii / 8-bit chars")
-	}
-	buf.Write(bytes.Repeat([]byte{0}, bannerMessageLen-n))
-	return buf.Bytes()
-}
-
-var banner_msg = mustMessage("SSDIRECTHCON_HELO")
-var proxy_error_msg = mustMessage("SSDIRECTHCON_PROXY_ERROR") /* FIXME irrelevant, was copy-pasta */
-var begin_msg = mustMessage("SSDIRECTHCON_BEGIN")
-
-type SSHError struct {
-	RWCError      error
-	WhileActivity string
-}
-
-// Error() will try to present a one-line error message unless ssh stderr output is longer than one line
-func (e *SSHError) Error() string {
-
-	exitErr, ok := e.RWCError.(*exec.ExitError)
-	if !ok {
-		return fmt.Sprintf("ssh: %s", e.RWCError)
-	}
-
-	ws := exitErr.ProcessState.Sys().(syscall.WaitStatus)
-	var wsmsg string
-	if ws.Exited() {
-		wsmsg = fmt.Sprintf("(exit status %d)", ws.ExitStatus())
-	} else {
-		wsmsg = fmt.Sprintf("(%s)", ws.Signal())
-	}
-
-	haveSSHMessage := len(exitErr.Stderr) > 0
-	sshOnelineStderr := false
-	if i := bytes.Index(exitErr.Stderr, []byte("\n")); i == len(exitErr.Stderr)-1 {
-		sshOnelineStderr = true
-	}
-	stderr := bytes.TrimSpace(exitErr.Stderr)
-
-	if haveSSHMessage {
-		if sshOnelineStderr {
-			return fmt.Sprintf("ssh: '%s' %s", stderr, wsmsg) // FIXME proper single-quoting
-		} else {
-			return fmt.Sprintf("ssh %s\n%s", wsmsg, stderr)
-		}
-	}
-
-	return fmt.Sprintf("ssh terminated without stderr output %s", wsmsg)
-
-}
-
-type ProtocolError struct {
-	What string
-}
-
-func (e ProtocolError) Error() string {
-	return e.What
-}
-
-// Dial connects to the remote endpoint where it expects a command executing Proxy().
-// Dial performs a handshake consisting of the exchange of banner messages before returning the connection.
-// If the handshake cannot be completed before dialCtx is Done(), the underlying ssh command is killed
-// and the dialCtx.Err() returned.
-// If the handshake completes, dialCtx's deadline does not affect the returned connection.
-//
-// Errors returned are either dialCtx.Err(), or intances of ProtocolError or *SSHError
-func Dial(dialCtx context.Context, endpoint Endpoint) (*SSHConn, error) {
-
-	sshCmd, sshArgs, sshEnv := endpoint.CmdArgs()
-	commandCtx, commandCancel := context.WithCancel(context.Background())
-	cmd := exec.CommandContext(commandCtx, sshCmd, sshArgs...)
-	cmd.Env = sshEnv
-	stdin, err := cmd.StdinPipe()
-	if err != nil {
-		return nil, err
-	}
-	stdout, err := cmd.StdoutPipe()
-	if err != nil {
-		return nil, err
-	}
-
-	stderrBuf, err := circlog.NewCircularLog(1 << 15)
-	if err != nil {
-		panic(err) // wrong API usage
-	}
-	cmd.Stderr = stderrBuf
-
-	if err = cmd.Start(); err != nil {
-		return nil, err
-	}
-	cmdWaitErrOrIOErr := func(ioErr error, what string) *SSHError {
-		werr := cmd.Wait()
-		if werr, ok := werr.(*exec.ExitError); ok {
-			werr.Stderr = []byte(stderrBuf.String())
-			return &SSHError{werr, what}
-		}
-		return &SSHError{ioErr, what}
-	}
-
-	confErrChan := make(chan error, 1)
-	go func() {
-		defer close(confErrChan)
-		var buf bytes.Buffer
-		if _, err := io.CopyN(&buf, stdout, int64(len(banner_msg))); err != nil {
-			confErrChan <- cmdWaitErrOrIOErr(err, "read banner")
-			return
-		}
-		resp := buf.Bytes()
-		switch {
-		case bytes.Equal(resp, banner_msg):
-			break
-		case bytes.Equal(resp, proxy_error_msg):
-			_ = cmdWaitErrOrIOErr(nil, "")
-			confErrChan <- ProtocolError{"proxy error, check remote configuration"}
-			return
-		default:
-			_ = cmdWaitErrOrIOErr(nil, "")
-			confErrChan <- ProtocolError{fmt.Sprintf("unknown banner message: %v", resp)}
-			return
-		}
-		buf.Reset()
-		buf.Write(begin_msg)
-		if _, err := io.Copy(stdin, &buf); err != nil {
-			confErrChan <- cmdWaitErrOrIOErr(err, "send begin message")
-			return
-		}
-	}()
-
-	select {
-	case <-dialCtx.Done():
-
-		commandCancel()
-		// cancelling will make one of the calls in above goroutine fail,
-		// and the goroutine will send the error to confErrChan
-		//
-		// ignore the error and return the cancellation cause
-
-		// draining always terminates because we know the channel is always closed
-		for _ = range confErrChan {
-		}
-
-		// TODO collect stderr in this case
-		// can probably extend *SSHError for this but need to implement net.Error
-
-		return nil, dialCtx.Err()
-
-	case err := <-confErrChan:
-		if err != nil {
-			commandCancel()
-			return nil, err
-		}
-	}
-
-	return &SSHConn{
-		cmd:       cmd,
-		stdin:     stdin,
-		stdout:    stdout,
-		cmdCancel: commandCancel,
-	}, nil
-}
-
-type Connecter struct {
-	s        *yamux.Session
-	endpoint Endpoint
-}
-
-var _ transport.Connecter = (*Connecter)(nil)
-
-func NewConnecter(ctx context.Context, endpoint Endpoint) (*Connecter, error) {
-	conn, err := Dial(ctx, endpoint)
-	if err != nil {
-		return nil, err
-	}
-	s, err := yamux.Client(conn, nil)
-	if err != nil {
-		return nil, err
-	}
-	return &Connecter{
-		s:        s,
-		endpoint: endpoint,
-	}, nil
-}
-
-type fakeWire struct {
-	net.Conn
-}
-
-func (w *fakeWire) CloseWrite() error {
-	time.Sleep(1*time.Second) // HACKY
-	return fmt.Errorf("fakeWire does not support CloseWrite")
-}
-
-func (c *Connecter) Connect(ctx context.Context) (transport.Wire, error) {
-	conn, err := c.s.Open()
-	return &fakeWire{conn}, err
-}

zync/transport/sshdirect/error.go

@@ -1,48 +0,0 @@
-package sshdirect
-
-import (
-	"fmt"
-	"net"
-	"os"
-	"syscall"
-)
-
-type timeouter interface {
-	Timeout() bool
-}
-
-var _ timeouter = &os.PathError{}
-
-type IOError struct {
-	Cause error
-}
-
-var _ net.Error = &IOError{}
-
-func (e IOError) GoString() string {
-	return fmt.Sprintf("ServeConnIOError:%#v", e.Cause)
-}
-
-func (e IOError) Error() string {
-	// following case found by experiment
-	if pathErr, ok := e.Cause.(*os.PathError); ok {
-		if pathErr.Err == syscall.EPIPE {
-			return fmt.Sprintf("netssh %s: %s (likely: connection reset by peer)",
-				pathErr.Op, pathErr.Err,
-			)
-		}
-		return fmt.Sprintf("netssh: %s: %s", pathErr.Op, pathErr.Err)
-	}
-	return fmt.Sprintf("netssh: %s", e.Cause.Error())
-}
-
-func (e IOError) Timeout() bool {
-	if to, ok := e.Cause.(timeouter); ok {
-		return to.Timeout()
-	}
-	return false
-}
-
-func (e IOError) Temporary() bool {
-	return false
-}

zync/transport/sshdirect/serve.go

@@ -1,91 +0,0 @@
-package sshdirect
-
-import (
-	"bytes"
-	"io"
-	"log"
-	"net"
-	"os"
-	"time"
-
-	"github.com/hashicorp/yamux"
-)
-
-type ServeConn struct {
-	stdin, stdout *os.File
-}
-
-var _ net.Conn = (*ServeConn)(nil)
-
-func ServeStdin() (net.Listener, error) {
-
-	conn := &ServeConn{
-		stdin:  os.Stdin,
-		stdout: os.Stdout,
-	}
-
-	var buf bytes.Buffer
-	buf.Write(banner_msg)
-	if _, err := io.Copy(conn, &buf); err != nil {
-		log.Printf("error sending confirm message: %s", err)
-		conn.Close()
-		return nil, err
-	}
-	buf.Reset()
-	if _, err := io.CopyN(&buf, conn, int64(len(begin_msg))); err != nil {
-		log.Printf("error reading begin message: %s", err)
-		conn.Close()
-		return nil, err
-	}
-
-	return yamux.Server(conn, nil)
-}
-
-func (c *ServeConn) Read(p []byte) (int, error) {
-	return c.stdin.Read(p)
-}
-
-func (c *ServeConn) Write(p []byte) (int, error) {
-	return c.stdout.Write(p)
-}
-
-func (f *ServeConn) Close() (err error) {
-	e1 := f.stdin.Close()
-	e2 := f.stdout.Close()
-	// FIXME merge errors
-	if e1 != nil {
-		return e1
-	}
-	return e2
-}
-
-func (f *ServeConn) SetReadDeadline(t time.Time) error {
-	return f.stdin.SetReadDeadline(t)
-}
-
-func (f *ServeConn) SetWriteDeadline(t time.Time) error {
-	return f.stdout.SetReadDeadline(t)
-}
-
-func (f *ServeConn) SetDeadline(t time.Time) error {
-	// try both...
-	werr := f.SetWriteDeadline(t)
-	rerr := f.SetReadDeadline(t)
-	if werr != nil {
-		return werr
-	}
-	if rerr != nil {
-		return rerr
-	}
-	return nil
-}
-
-type serveAddr struct{}
-
-const GoNetwork string = "sshdirect"
-
-func (serveAddr) Network() string { return GoNetwork }
-func (serveAddr) String() string  { return "???" }
-
-func (f *ServeConn) LocalAddr() net.Addr  { return serveAddr{} }
-func (f *ServeConn) RemoteAddr() net.Addr { return serveAddr{} }

zync/transport/transportlistenerfromnetlistener/transportlistenerfromnetlistener.go

@@ -1,47 +0,0 @@
-package transportlistenerfromnetlistener
-
-import (
-	"context"
-	"fmt"
-	"net"
-	"time"
-
-	"github.com/zrepl/zrepl/transport"
-)
-
-type wrapFixed struct {
-	id string
-	l  net.Listener
-}
-
-var _ transport.AuthenticatedListener = (*wrapFixed)(nil)
-
-func WrapFixed(l net.Listener, identity string) transport.AuthenticatedListener {
-	return &wrapFixed{identity, l}
-}
-
-func (w *wrapFixed) Addr() net.Addr {
-	return w.l.Addr()
-}
-
-type fakeWire struct {
-	net.Conn
-}
-
-func (w *fakeWire) CloseWrite() error {
-	time.Sleep(1*time.Second) // HACKY
-	return fmt.Errorf("fakeWire does not support CloseWrite")
-}
-
-func (w *wrapFixed) Accept(ctx context.Context) (*transport.AuthConn, error) {
-	nc, err := w.l.Accept()
-	if err != nil {
-		return nil, err
-	}
-
-	return transport.NewAuthConn(&fakeWire{nc}, w.id), nil
-}
-
-func (w *wrapFixed) Close() error {
-	return w.l.Close()
-}

zync/zync.go

@@ -1,333 +0,0 @@
-package main
-
-import (
-	"context"
-	"flag"
-	"fmt"
-	"io/ioutil"
-	"net/url"
-	"os"
-	"os/signal"
-	"runtime"
-	"strconv"
-	"syscall"
-	"time"
-
-	"github.com/kr/pretty"
-	"github.com/pkg/errors"
-	"github.com/zrepl/zrepl/daemon/filters"
-	"github.com/zrepl/zrepl/daemon/logging"
-	"github.com/zrepl/zrepl/daemon/logging/trace"
-	"github.com/zrepl/zrepl/endpoint"
-	"github.com/zrepl/zrepl/logger"
-	"github.com/zrepl/zrepl/replication"
-	"github.com/zrepl/zrepl/replication/logic"
-	"github.com/zrepl/zrepl/replication/logic/pdu"
-	"github.com/zrepl/zrepl/rpc"
-	"github.com/zrepl/zrepl/transport"
-	"github.com/zrepl/zrepl/zfs"
-	"github.com/zrepl/zrepl/zync/transport/sshdirect"
-	"github.com/zrepl/zrepl/zync/transport/transportlistenerfromnetlistener"
-)
-
-var flagStdinserver = flag.String("stdinserver", "", "")
-var flagStderrToFile = flag.String("stderrtofile", "", "")
-
-const (
-	modeSender   = "sender"
-	modeReceiver = "receiver"
-)
-
-func connecter(ctx context.Context, u *url.URL, mode, fs string) (transport.Connecter, error) {
-	switch u.Scheme {
-	case "ssh":
-		return connecterSSH(ctx, u, mode, fs)
-	default:
-		panic(fmt.Sprintf("unknown scheme %q", u.Scheme))
-	}
-}
-
-func connecterSSH(ctx context.Context, u *url.URL, mode, fs string) (transport.Connecter, error) {
-	var port uint16
-	if u.Port() == "" {
-		port = 22
-	} else {
-		portU64, err := strconv.ParseUint(u.Port(), 10, 16)
-		if err != nil {
-			return nil, errors.Wrap(err, "invalid port")
-		}
-		port = uint16(portU64)
-	}
-
-	fmt.Println(u.User)
-	idFilePath, hasIdFile := u.User.Password()
-	if hasIdFile {
-		_, err := ioutil.ReadFile(idFilePath)
-		if err != nil {
-			fmt.Println(err)
-			hasIdFile = false
-		}
-	}
-	if !hasIdFile {
-		return nil, errors.New("must set password to identity file path")
-	}
-
-	ep := sshdirect.Endpoint{
-		Host:         u.Hostname(),
-		Port:         port,
-		User:         u.User.Username(),
-		IdentityFile: idFilePath,
-		RunCommand:   []string{"zync", "-stderrtofile", "/tmp/zync_server.log", "-stdinserver", mode, fs},
-	}
-	return sshdirect.NewConnecter(ctx, ep)
-}
-
-func onefsfilter(osname string) zfs.DatasetFilter {
-	f, err := filters.DatasetMapFilterFromConfig(map[string]bool{
-		osname: true,
-	})
-	if err != nil {
-		panic(err)
-	}
-	return f
-}
-
-func makeEndpoint(mode, fs string) (interface{}, error) {
-	switch mode {
-	case modeSender:
-		sc := endpoint.SenderConfig{
-			FSF:     onefsfilter(fs),
-			Encrypt: &zfs.NilBool{B: true},
-			JobID:   endpoint.MustMakeJobID("sender"),
-		}
-		return endpoint.NewSender(sc), nil
-	case modeReceiver:
-		rpath, err := zfs.NewDatasetPath(fs)
-		if err != nil {
-			panic(err)
-		}
-		if err != nil {
-			panic(err)
-		}
-		return endpoint.NewReceiver(endpoint.ReceiverConfig{
-			JobID:                      endpoint.MustMakeJobID("receiver"),
-			AppendClientIdentity:       false,
-			RootWithoutClientComponent: rpath,
-		}), nil
-	default:
-		return nil, fmt.Errorf("unknown mode %q", mode)
-	}
-}
-
-func serve(ctx context.Context, handler rpc.Handler) error {
-
-	// copy-pasta from passive.go
-	ctxInterceptor := func(handlerCtx context.Context, info rpc.HandlerContextInterceptorData, handler func(ctx context.Context)) {
-		// the handlerCtx is clean => need to inherit logging and tracing config from job context
-		handlerCtx = logging.WithInherit(handlerCtx, ctx)
-		handlerCtx = trace.WithInherit(handlerCtx, ctx)
-
-		handlerCtx, endTask := trace.WithTaskAndSpan(handlerCtx, "handler", fmt.Sprintf("method=%q", info.FullMethod()))
-		defer endTask()
-		handler(handlerCtx)
-	}
-
-	l, err := sshdirect.ServeStdin()
-	if err != nil {
-		panic(err)
-	}
-
-	srv := rpc.NewServer(handler, rpc.GetLoggersOrPanic(ctx), ctxInterceptor)
-	srv.Serve(ctx, transportlistenerfromnetlistener.WrapFixed(l, "fakeclientidentitymustnotbeempty"))
-	return nil
-}
-
-func parseFSArg(ctx context.Context, mode, arg string) (interface{}, error) {
-	u, err := url.Parse(arg)
-	if err != nil {
-		return nil, err
-	}
-	if !u.IsAbs() {
-		return nil, fmt.Errorf("URL must be absolute, got %q", arg)
-	}
-	if u.Path[0] != '/' {
-		panic("impl error: expecting leading /")
-	}
-	fs := u.Path[1:]
-	switch u.Scheme {
-	case "local":
-		if u.Host != "" {
-			panic("hostname must be empty for 'local' scheme")
-		}
-		return makeEndpoint(mode, fs)
-	default:
-		cn, err := connecter(ctx, u, mode, fs)
-		if err != nil {
-			return nil, err
-		}
-		return rpc.NewClient(cn, rpc.GetLoggersOrPanic(ctx)), nil
-	}
-}
-
-// func parseEndpoints(s, r string) (sender, receiver logic.Endpoint, _ error) {
-// 	sUrl, err := url.Parse(s)
-// 	if err != nil {
-// 		return nil, nil, err
-// 	}
-// 	rUrl, err := url.Parse((r))
-// 	if err != nil {
-// 		return nil, nil, err
-// 	}
-// 	if !sUrl.IsAbs() || !rUrl.IsAbs() {
-// 		return nil, nil, fmt.Errorf("must have a scheme")
-// 	}
-
-// 	if sUrl.Scheme == "local" {&& rUrl.Scheme == "local" {
-// 		var err error
-// 		sender, err = makeEndpoint(modeSender, sUrl.Path)
-// 		if err == nil {
-// 			receiver, err = makeEndpoint(modeReceiver, rUrl.Path)
-// 		}
-// 		return sender, receiver, err
-// 	} else {
-
-// 	}
-// }
-
-func main() {
-
-	cancelSigs := make(chan os.Signal)
-	signal.Notify(cancelSigs, os.Interrupt, syscall.SIGTERM)
-
-	ctx := context.Background()
-	trace.WithTaskFromStackUpdateCtx(&ctx)
-
-	ctx = logging.WithLoggers(ctx, logging.SubsystemLoggersWithUniversalLogger(logger.NewStderrDebugLogger()))
-
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
-	go func() {
-		for {
-			select {
-			case <-cancelSigs:
-				cancel()
-			}
-		}
-	}()
-
-	flag.Parse()
-
-	if *flagStderrToFile != "" {
-		f, err := os.Create(*flagStderrToFile)
-		if err != nil {
-			panic(err)
-		}
-		syscall.Dup2(int(f.Fd()), int(os.Stderr.Fd()))
-		runtime.KeepAlive(f)
-		// enough?
-	}
-
-	if *flagStdinserver != "" {
-		if flag.NArg() != 1 {
-			panic("usage: -stdinserver MODE FS")
-		}
-		h, err := makeEndpoint(*flagStdinserver, flag.Arg(0))
-		if err != nil {
-			panic(err)
-		}
-		err = serve(ctx, h.(rpc.Handler))
-		if err != nil {
-			panic(err)
-		}
-		return
-	}
-
-	if flag.NArg() != 2 {
-		panic("usage: zync SENDER RECEIVER")
-	}
-
-	sender, err := parseFSArg(ctx, modeSender, flag.Arg(0))
-	if err != nil {
-		panic(err)
-	}
-
-	receiver, err := parseFSArg(ctx, modeReceiver, flag.Arg(1))
-	if err != nil {
-		panic(err)
-	}
-
-	pp := logic.PlannerPolicy{
-		EncryptedSend: logic.TriFromBool(true), // FIXME add flag
-		ReplicationConfig: pdu.ReplicationConfig{
-			Protection: &pdu.ReplicationConfigProtection{
-				Initial:     pdu.ReplicationGuaranteeKind_GuaranteeNothing,
-				Incremental: pdu.ReplicationGuaranteeKind_GuaranteeNothing,
-			},
-		},
-	}
-	p := logic.NewPlanner(nil, nil, sender.(logic.Sender), receiver.(logic.Receiver), pp)
-	_, wait := replication.Do(ctx, p)
-	defer wait(true)
-
-}
-
-func local() {
-
-	cancelSigs := make(chan os.Signal)
-	signal.Notify(cancelSigs, os.Interrupt, syscall.SIGTERM)
-
-	ctx := context.Background()
-	trace.WithTaskFromStackUpdateCtx(&ctx)
-
-	ctx = logging.WithLoggers(ctx, logging.SubsystemLoggersWithUniversalLogger(logger.NewStderrDebugLogger()))
-
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
-	go func() {
-		for {
-			select {
-			case <-cancelSigs:
-				cancel()
-			}
-		}
-	}()
-
-	sc := endpoint.SenderConfig{
-		FSF:     onefsfilter(os.Args[1]),
-		Encrypt: &zfs.NilBool{B: true},
-		JobID:   endpoint.MustMakeJobID("sender"),
-	}
-	s := endpoint.NewSender(sc)
-	pretty.Println(s)
-	rpath, err := zfs.NewDatasetPath(os.Args[2])
-	if err != nil {
-		panic(err)
-	}
-	r := endpoint.NewReceiver(endpoint.ReceiverConfig{
-		JobID:                      endpoint.MustMakeJobID("receiver"),
-		AppendClientIdentity:       false,
-		RootWithoutClientComponent: rpath,
-	})
-	pretty.Println(r)
-	pp := logic.PlannerPolicy{
-		EncryptedSend: logic.TriFromBool(sc.Encrypt.B),
-		ReplicationConfig: pdu.ReplicationConfig{
-			Protection: &pdu.ReplicationConfigProtection{
-				Initial:     pdu.ReplicationGuaranteeKind_GuaranteeNothing,
-				Incremental: pdu.ReplicationGuaranteeKind_GuaranteeNothing,
-			},
-		},
-	}
-	p := logic.NewPlanner(nil, nil, s, r, pp)
-	report, wait := replication.Do(ctx, p)
-	defer wait(true)
-
-	ticker := time.NewTicker(2 * time.Second)
-	for !wait(false) {
-		select {
-		case <-ticker.C:
-			// pretty.Println(report())
-		}
-	}
-	pretty.Println(report())
-}