fixups

cc @asomers

.circleci/config.yml

@@ -150,7 +150,7 @@ parameters:
 
   release_docker_baseimage_tag:
     type: string
-    default: "1.17"
+    default: "1.16"
 
 workflows:
   version: 2
@@ -160,15 +160,15 @@ workflows:
     jobs:
       - quickcheck-docs
       - quickcheck-go: &quickcheck-go-smoketest
-          name: quickcheck-go-amd64-linux-1.17
+          name: quickcheck-go-amd64-linux-1.16
-          goversion: &latest-go-release "1.17"
+          goversion: &latest-go-release "1.16"
           goos: linux
           goarch: amd64
       - test-go-on-latest-go-release:
           goversion: *latest-go-release
       - quickcheck-go:
           requires:
-            - quickcheck-go-amd64-linux-1.17 #quickcheck-go-smoketest.name
+            - quickcheck-go-amd64-linux-1.16 #quickcheck-go-smoketest.name
           matrix: &quickcheck-go-matrix
             alias: quickcheck-go-matrix
             parameters:
@@ -231,6 +231,8 @@ jobs:
       - install-docdep
       - run: make docs
 
+      - store_artifacts:
+          path: artifacts
       - download-and-install-minio-client
       - upload-minio:
           src: artifacts

Makefile

@@ -28,7 +28,7 @@ GO_BUILDFLAGS := $(GO_MOD_READONLY) $(GO_EXTRA_BUILDFLAGS)
 GO_BUILD := $(GO_ENV_VARS) $(GO) build $(GO_BUILDFLAGS) -ldflags $(GO_LDFLAGS)
 GOLANGCI_LINT := golangci-lint
 GOCOVMERGE := gocovmerge
-RELEASE_DOCKER_BASEIMAGE_TAG ?= 1.17
+RELEASE_DOCKER_BASEIMAGE_TAG ?= 1.16
 RELEASE_DOCKER_BASEIMAGE ?= golang:$(RELEASE_DOCKER_BASEIMAGE_TAG)
 
 ifneq ($(GOARM),)

build/tools.go

@@ -1,4 +1,3 @@
-//go:build tools
 // +build tools
 
 package main

client/status/viewmodel/bytecountbinary.go

@@ -2,22 +2,14 @@ package viewmodel
 
 import (
 	"fmt"
-	"math"
 )
 
-func ByteCountBinaryUint(b uint64) string {
-	if b > math.MaxInt64 {
-		panic(b)
-	}
-	return ByteCountBinary(int64(b))
-}
-
 func ByteCountBinary(b int64) string {
 	const unit = 1024
 	if b < unit {
 		return fmt.Sprintf("%d B", b)
 	}
-	div, exp := unit, 0
+	div, exp := int64(unit), 0
 	for n := b / unit; n >= unit; n /= unit {
 		div *= unit
 		exp++

client/status/viewmodel/bytesprogresshistory.go

@@ -4,7 +4,7 @@ import "time"
 
 type byteProgressMeasurement struct {
 	time time.Time
-	val  uint64
+	val  int64
 }
 
 type bytesProgressHistory struct {
@@ -13,7 +13,7 @@ type bytesProgressHistory struct {
 	lastChange  time.Time
 }
 
-func (p *bytesProgressHistory) Update(currentVal uint64) (bytesPerSecondAvg int64, changeCount int) {
+func (p *bytesProgressHistory) Update(currentVal int64) (bytesPerSecondAvg int64, changeCount int) {
 
 	if p.last == nil {
 		p.last = &byteProgressMeasurement{
@@ -33,12 +33,7 @@ func (p *bytesProgressHistory) Update(currentVal uint64) (bytesPerSecondAvg int6
 		return 0, 0
 	}
 
-	var deltaV int64
+	deltaV := currentVal - p.last.val
-	if currentVal >= p.last.val {
-		deltaV = int64(currentVal - p.last.val)
-	} else {
-		deltaV = -int64(p.last.val - currentVal)
-	}
 	deltaT := time.Since(p.last.time)
 	rate := float64(deltaV) / deltaT.Seconds()
 

client/status/viewmodel/render.go

@@ -230,7 +230,7 @@ func printFilesystemStatus(t *stringbuilder.B, rep *report.FilesystemReport, act
 	status := fmt.Sprintf("%s (step %d/%d, %s/%s)%s",
 		strings.ToUpper(string(rep.State)),
 		rep.CurrentStep, len(rep.Steps),
-		ByteCountBinaryUint(replicated), ByteCountBinaryUint(expected),
+		ByteCountBinary(replicated), ByteCountBinary(expected),
 		sizeEstimationImpreciseNotice,
 	)
 
@@ -358,12 +358,12 @@ func renderReplicationReport(t *stringbuilder.B, rep *report.Report, history *by
 		rate, changeCount := history.Update(replicated)
 		eta := time.Duration(0)
 		if rate > 0 {
-			eta = time.Duration((float64(expected)-float64(replicated))/float64(rate)) * time.Second
+			eta = time.Duration((expected-replicated)/rate) * time.Second
 		}
 
 		t.Write("Progress: ")
 		t.DrawBar(50, replicated, expected, changeCount)
-		t.Write(fmt.Sprintf(" %s / %s @ %s/s", ByteCountBinaryUint(replicated), ByteCountBinaryUint(expected), ByteCountBinary(rate)))
+		t.Write(fmt.Sprintf(" %s / %s @ %s/s", ByteCountBinary(replicated), ByteCountBinary(expected), ByteCountBinary(rate)))
 		if eta != 0 {
 			t.Write(fmt.Sprintf(" (%s remaining)", humanizeDuration(eta)))
 		}

client/status/viewmodel/stringbuilder/stringbuilder.go

@@ -99,11 +99,11 @@ func RightPad(str string, length int, pad string) string {
 }
 
 // changeCount = 0 indicates stall / no progress
-func (w *B) DrawBar(length int, bytes, totalBytes uint64, changeCount int) {
+func (w *B) DrawBar(length int, bytes, totalBytes int64, changeCount int) {
 	const arrowPositions = `>\|/`
 	var completedLength int
 	if totalBytes > 0 {
-		completedLength = int(uint64(length) * bytes / totalBytes)
+		completedLength = int(int64(length) * bytes / totalBytes)
 		if completedLength > length {
 			completedLength = length
 		}

daemon/job/snapjob.go

@@ -10,7 +10,6 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/zrepl/zrepl/daemon/logging/trace"
-	"github.com/zrepl/zrepl/util/bandwidthlimit"
 	"github.com/zrepl/zrepl/util/nodefault"
 
 	"github.com/zrepl/zrepl/config"
@@ -147,7 +146,7 @@ type alwaysUpToDateReplicationCursorHistory struct {
 	target pruner.Target
 }
 
-var _ pruner.Sender = (*alwaysUpToDateReplicationCursorHistory)(nil)
+var _ pruner.History = (*alwaysUpToDateReplicationCursorHistory)(nil)
 
 func (h alwaysUpToDateReplicationCursorHistory) ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error) {
 	fsvReq := &pdu.ListFilesystemVersionsReq{
@@ -180,11 +179,8 @@ func (j *SnapJob) doPrune(ctx context.Context) {
 	sender := endpoint.NewSender(endpoint.SenderConfig{
 		JobID: j.name,
 		FSF:   j.fsfilter,
-		// FIXME the following config fields are irrelevant for SnapJob
+		// FIXME encryption setting is irrelevant for SnapJob because the endpoint is only used as pruner.Target
-		// because the endpoint is only used as pruner.Target.
-		// However, the implementation requires them to be set.
 		Encrypt: &nodefault.Bool{B: true},
-		BandwidthLimit: bandwidthlimit.NoLimitConfig(),
 	})
 	j.prunerMtx.Lock()
 	j.pruner = j.prunerFactory.BuildLocalPruner(ctx, sender, alwaysUpToDateReplicationCursorHistory{sender})

daemon/pruner/pruner.go

@@ -19,20 +19,12 @@ import (
 	"github.com/zrepl/zrepl/util/envconst"
 )
 
-// The sender in the replication setup.
-// The pruner uses the Sender to determine which of the Target's filesystems need to be pruned.
-// Also, it asks the Sender about the replication cursor of each filesystem
-// to enable the 'not_replicated' pruning rule.
-//
 // Try to keep it compatible with github.com/zrepl/zrepl/endpoint.Endpoint
-type Sender interface {
+type History interface {
 	ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error)
 	ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error)
 }
 
-// The pruning target, i.e., on which snapshots are destroyed.
-// This can be a replication sender or receiver.
-//
 // Try to keep it compatible with github.com/zrepl/zrepl/endpoint.Endpoint
 type Target interface {
 	ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error)
@@ -56,7 +48,7 @@ func GetLogger(ctx context.Context) Logger {
 type args struct {
 	ctx                            context.Context
 	target                         Target
-	sender                         Sender
+	receiver                       History
 	rules                          []pruning.KeepRule
 	retryWait                      time.Duration
 	considerSnapAtCursorReplicated bool
@@ -140,12 +132,12 @@ func NewPrunerFactory(in config.PruningSenderReceiver, promPruneSecs *prometheus
 	return f, nil
 }
 
-func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, target Target, sender Sender) *Pruner {
+func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, target Target, receiver History) *Pruner {
 	p := &Pruner{
 		args: args{
 			context.WithValue(ctx, contextKeyPruneSide, "sender"),
 			target,
-			sender,
+			receiver,
 			f.senderRules,
 			f.retryWait,
 			f.considerSnapAtCursorReplicated,
@@ -156,12 +148,12 @@ func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, target Target, se
 	return p
 }
 
-func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target, sender Sender) *Pruner {
+func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target, receiver History) *Pruner {
 	p := &Pruner{
 		args: args{
 			context.WithValue(ctx, contextKeyPruneSide, "receiver"),
 			target,
-			sender,
+			receiver,
 			f.receiverRules,
 			f.retryWait,
 			false, // senseless here anyways
@@ -172,12 +164,12 @@ func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target,
 	return p
 }
 
-func (f *LocalPrunerFactory) BuildLocalPruner(ctx context.Context, target Target, history Sender) *Pruner {
+func (f *LocalPrunerFactory) BuildLocalPruner(ctx context.Context, target Target, receiver History) *Pruner {
 	p := &Pruner{
 		args: args{
 			context.WithValue(ctx, contextKeyPruneSide, "local"),
 			target,
-			history,
+			receiver,
 			f.keepRules,
 			f.retryWait,
 			false, // considerSnapAtCursorReplicated is not relevant for local pruning
@@ -351,9 +343,9 @@ func (s snapshot) Date() time.Time { return s.date }
 
 func doOneAttempt(a *args, u updater) {
 
-	ctx, target, sender := a.ctx, a.target, a.sender
+	ctx, target, receiver := a.ctx, a.target, a.receiver
 
-	sfssres, err := sender.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
+	sfssres, err := receiver.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
 	if err != nil {
 		u(func(p *Pruner) {
 			p.state = PlanErr
@@ -418,7 +410,7 @@ tfss_loop:
 		rcReq := &pdu.ReplicationCursorReq{
 			Filesystem: tfs.Path,
 		}
-		rc, err := sender.ReplicationCursor(ctx, rcReq)
+		rc, err := receiver.ReplicationCursor(ctx, rcReq)
 		if err != nil {
 			pfsPlanErrAndLog(err, "cannot get replication cursor bookmark")
 			continue tfss_loop
@@ -464,7 +456,7 @@ tfss_loop:
 			})
 		}
 		if preCursor {
-			pfsPlanErrAndLog(fmt.Errorf("prune target has no snapshot that corresponds to sender replication cursor bookmark"), "")
+			pfsPlanErrAndLog(fmt.Errorf("replication cursor not found in prune target filesystem versions"), "")
 			continue tfss_loop
 		}
 

docs/configuration/prune.rst

@@ -67,9 +67,8 @@ Policy ``not_replicated``
      ...
 
 ``not_replicated`` keeps all snapshots that have not been replicated to the receiving side.
-It only makes sense to specify this rule for the ``keep_sender``.
+It only makes sense to specify this rule on a sender (source or push job).
-The reason is that, by definition, all snapshots on the receiver have already been replicated to there from the sender.
+The state required to evaluate this rule is stored in the :ref:`replication cursor bookmark <replication-cursor-and-last-received-hold>` on the sending side.
-To determine whether a sender-side snapshot has already been replicated, zrepl uses the :ref:`replication cursor bookmark <replication-cursor-and-last-received-hold>` which corresponds to the most recent successfully replicated snapshot.
 
 .. _prune-keep-retention-grid:
 

endpoint/endpoint.go

@@ -159,11 +159,12 @@ func sendArgsFromPDUAndValidateExistsAndGetVersion(ctx context.Context, fs strin
 	return version, nil
 }
 
-func (s *Sender) sendMakeArgs(ctx context.Context, r *pdu.SendReq) (sendArgs zfs.ZFSSendArgsValidated, _ error) {
+func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 
 	_, err := s.filterCheckFS(r.Filesystem)
 	if err != nil {
-		return sendArgs, err
+		return nil, nil, err
 	}
 	switch r.Encrypted {
 	case pdu.Tri_DontCare:
@@ -171,16 +172,16 @@ func (s *Sender) sendMakeArgs(ctx context.Context, r *pdu.SendReq) (sendArgs zfs
 		// ok, fallthrough outer
 	case pdu.Tri_False:
 		if s.config.Encrypt.B {
-			return sendArgs, errors.New("only encrypted sends allowed (send -w + encryption!= off), but unencrypted send requested")
+			return nil, nil, errors.New("only encrypted sends allowed (send -w + encryption!= off), but unencrypted send requested")
 		}
 		// fallthrough outer
 	case pdu.Tri_True:
 		if !s.config.Encrypt.B {
-			return sendArgs, errors.New("only unencrypted sends allowed, but encrypted send requested")
+			return nil, nil, errors.New("only unencrypted sends allowed, but encrypted send requested")
 		}
 		// fallthrough outer
 	default:
-		return sendArgs, fmt.Errorf("unknown pdu.Tri variant %q", r.Encrypted)
+		return nil, nil, fmt.Errorf("unknown pdu.Tri variant %q", r.Encrypted)
 	}
 
 	sendArgsUnvalidated := zfs.ZFSSendArgsUnvalidated{
@@ -200,19 +201,30 @@ func (s *Sender) sendMakeArgs(ctx context.Context, r *pdu.SendReq) (sendArgs zfs
 		},
 	}
 
-	sendArgs, err = sendArgsUnvalidated.Validate(ctx)
+	sendArgs, err := sendArgsUnvalidated.Validate(ctx)
 	if err != nil {
-		return sendArgs, errors.Wrap(err, "validate send arguments")
+		return nil, nil, errors.Wrap(err, "validate send arguments")
-	}
-	return sendArgs, nil
 	}
 
-func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
+	si, err := zfs.ZFSSendDry(ctx, sendArgs)
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-
-	sendArgs, err := s.sendMakeArgs(ctx, r)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, errors.Wrap(err, "zfs send dry failed")
+	}
+
+	// From now on, assume that sendArgs has been validated by ZFSSendDry
+	// (because validation involves shelling out, it's actually a little expensive)
+
+	var expSize int64 = 0      // protocol says 0 means no estimate
+	if si.SizeEstimate != -1 { // but si returns -1 for no size estimate
+		expSize = si.SizeEstimate
+	}
+	res := &pdu.SendRes{
+		ExpectedSize:    expSize,
+		UsedResumeToken: r.ResumeToken != "",
+	}
+
+	if r.DryRun {
+		return res, nil, nil
 	}
 
 	// create holds or bookmarks of `From` and `To` to guarantee one of the following:
@@ -310,37 +322,9 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
 	// apply rate limit
 	sendStream = s.bwLimit.WrapReadCloser(sendStream)
 
-	res := &pdu.SendRes{
-		ExpectedSize:    0,
-		UsedResumeToken: r.ResumeToken != "",
-	}
-
 	return res, sendStream, nil
 }
 
-func (s *Sender) SendDry(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-
-	sendArgs, err := s.sendMakeArgs(ctx, r)
-	if err != nil {
-		return nil, err
-	}
-
-	si, err := zfs.ZFSSendDry(ctx, sendArgs)
-	if err != nil {
-		return nil, errors.Wrap(err, "zfs send dry failed")
-	}
-
-	// From now on, assume that sendArgs has been validated by ZFSSendDry
-	// (because validation involves shelling out, it's actually a little expensive)
-
-	res := &pdu.SendRes{
-		ExpectedSize:    si.SizeEstimate,
-		UsedResumeToken: r.ResumeToken != "",
-	}
-	return res, nil
-}
-
 func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
 	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 
@@ -709,11 +693,6 @@ func (s *Receiver) Send(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, io
 	return nil, nil, fmt.Errorf("receiver does not implement Send()")
 }
 
-func (s *Receiver) SendDry(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-	return nil, fmt.Errorf("receiver does not implement SendDry()")
-}
-
 func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.ReadCloser) (*pdu.ReceiveRes, error) {
 	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 

go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/gdamore/tcell/v2 v2.2.0
 	github.com/gitchander/permutation v0.0.0-20181107151852-9e56b92e9909
 	github.com/go-logfmt/logfmt v0.4.0
+	github.com/go-playground/universal-translator v0.17.0 // indirect
 	github.com/go-playground/validator v9.31.0+incompatible
 	github.com/go-playground/validator/v10 v10.4.1
 	github.com/go-sql-driver/mysql v1.4.1-0.20190907122137-b2c03bcae3d4
@@ -43,6 +44,7 @@ require (
 	golang.org/x/net v0.0.0-20210119194325-5f4716e94777
 	golang.org/x/sync v0.0.0-20190423024810-112230192c58
 	golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c
+	golang.org/x/text v0.3.5 // indirect
 	golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135
 	gonum.org/v1/gonum v0.7.0 // indirect
 	google.golang.org/genproto v0.0.0-20210122163508-8081c04a3579 // indirect

platformtest/tests/generated_cases.go

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

platformtest/tests/replication.go

@@ -809,22 +809,13 @@ type NeverEndingSender struct {
 	*endpoint.Sender
 }
 
-func (s *NeverEndingSender) SendDry(ctx context.Context, req *pdu.SendReq) (r *pdu.SendRes, err error) {
-	r, _, err = s.sendImpl(ctx, req, true)
-	return r, err
-}
-
 func (s *NeverEndingSender) Send(ctx context.Context, req *pdu.SendReq) (r *pdu.SendRes, stream io.ReadCloser, _ error) {
-	return s.sendImpl(ctx, req, false)
-}
-
-func (s *NeverEndingSender) sendImpl(ctx context.Context, req *pdu.SendReq, dry bool) (r *pdu.SendRes, stream io.ReadCloser, _ error) {
 	stream = nil
 	r = &pdu.SendRes{
 		UsedResumeToken: false,
 		ExpectedSize:    1 << 30,
 	}
-	if dry {
+	if req.DryRun {
 		return r, stream, nil
 	}
 	dz, err := os.Open("/dev/zero")

platformtest/tests/sendStream.go

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

replication/logic/pdu/pdu.pb.go

@@ -518,7 +518,8 @@ type SendReq struct {
 	// encoded in the ResumeToken. Otherwise, the Sender MUST return an error.
 	ResumeToken       string             `protobuf:"bytes,4,opt,name=ResumeToken,proto3" json:"ResumeToken,omitempty"`
 	Encrypted         Tri                `protobuf:"varint,5,opt,name=Encrypted,proto3,enum=Tri" json:"Encrypted,omitempty"`
-	ReplicationConfig *ReplicationConfig `protobuf:"bytes,6,opt,name=ReplicationConfig,proto3" json:"ReplicationConfig,omitempty"`
+	DryRun            bool               `protobuf:"varint,6,opt,name=DryRun,proto3" json:"DryRun,omitempty"`
+	ReplicationConfig *ReplicationConfig `protobuf:"bytes,7,opt,name=ReplicationConfig,proto3" json:"ReplicationConfig,omitempty"`
 }
 
 func (x *SendReq) Reset() {
@@ -588,6 +589,13 @@ func (x *SendReq) GetEncrypted() Tri {
 	return Tri_DontCare
 }
 
+func (x *SendReq) GetDryRun() bool {
+	if x != nil {
+		return x.DryRun
+	}
+	return false
+}
+
 func (x *SendReq) GetReplicationConfig() *ReplicationConfig {
 	if x != nil {
 		return x.ReplicationConfig
@@ -758,11 +766,12 @@ type SendRes struct {
 	unknownFields protoimpl.UnknownFields
 
 	// Whether the resume token provided in the request has been used or not.
-	// If the SendReq.ResumeToken == "", this field MUST be false.
+	// If the SendReq.ResumeToken == "", this field has no meaning.
-	UsedResumeToken bool `protobuf:"varint,1,opt,name=UsedResumeToken,proto3" json:"UsedResumeToken,omitempty"`
+	UsedResumeToken bool `protobuf:"varint,2,opt,name=UsedResumeToken,proto3" json:"UsedResumeToken,omitempty"`
 	// Expected stream size determined by dry run, not exact.
 	// 0 indicates that for the given SendReq, no size estimate could be made.
-	ExpectedSize uint64 `protobuf:"varint,2,opt,name=ExpectedSize,proto3" json:"ExpectedSize,omitempty"`
+	ExpectedSize int64       `protobuf:"varint,3,opt,name=ExpectedSize,proto3" json:"ExpectedSize,omitempty"`
+	Properties   []*Property `protobuf:"bytes,4,rep,name=Properties,proto3" json:"Properties,omitempty"`
 }
 
 func (x *SendRes) Reset() {
@@ -804,13 +813,20 @@ func (x *SendRes) GetUsedResumeToken() bool {
 	return false
 }
 
-func (x *SendRes) GetExpectedSize() uint64 {
+func (x *SendRes) GetExpectedSize() int64 {
 	if x != nil {
 		return x.ExpectedSize
 	}
 	return 0
 }
 
+func (x *SendRes) GetProperties() []*Property {
+	if x != nil {
+		return x.Properties
+	}
+	return nil
+}
+
 type SendCompletedReq struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -1427,7 +1443,7 @@ var file_pdu_proto_rawDesc = []byte{
 	0x72, 0x65, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x22, 0x29, 0x0a, 0x0b, 0x56, 0x65, 0x72, 0x73, 0x69,
 	0x6f, 0x6e, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0c, 0x0a, 0x08, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68,
 	0x6f, 0x74, 0x10, 0x00, 0x12, 0x0c, 0x0a, 0x08, 0x42, 0x6f, 0x6f, 0x6b, 0x6d, 0x61, 0x72, 0x6b,
-	0x10, 0x01, 0x22, 0xfd, 0x01, 0x0a, 0x07, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65, 0x71, 0x12, 0x1e,
+	0x10, 0x01, 0x22, 0x95, 0x02, 0x0a, 0x07, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65, 0x71, 0x12, 0x1e,
 	0x0a, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x18, 0x01, 0x20, 0x01,
 	0x28, 0x09, 0x52, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x12, 0x26,
 	0x0a, 0x04, 0x46, 0x72, 0x6f, 0x6d, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x46,
@@ -1439,119 +1455,122 @@ var file_pdu_proto_rawDesc = []byte{
 	0x0b, 0x52, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x54, 0x6f, 0x6b, 0x65, 0x6e, 0x12, 0x22, 0x0a, 0x09,
 	0x45, 0x6e, 0x63, 0x72, 0x79, 0x70, 0x74, 0x65, 0x64, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0e, 0x32,
 	0x04, 0x2e, 0x54, 0x72, 0x69, 0x52, 0x09, 0x45, 0x6e, 0x63, 0x72, 0x79, 0x70, 0x74, 0x65, 0x64,
-	0x12, 0x40, 0x0a, 0x11, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43,
+	0x12, 0x16, 0x0a, 0x06, 0x44, 0x72, 0x79, 0x52, 0x75, 0x6e, 0x18, 0x06, 0x20, 0x01, 0x28, 0x08,
-	0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x52, 0x65,
+	0x52, 0x06, 0x44, 0x72, 0x79, 0x52, 0x75, 0x6e, 0x12, 0x40, 0x0a, 0x11, 0x52, 0x65, 0x70, 0x6c,
-	0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52,
+	0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x07, 0x20,
+	0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f,
+	0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52, 0x11, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61,
+	0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x22, 0x51, 0x0a, 0x11, 0x52, 0x65,
+	0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12,
+	0x3c, 0x0a, 0x0a, 0x70, 0x72, 0x6f, 0x74, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20,
+	0x01, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f,
+	0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x50, 0x72, 0x6f, 0x74, 0x65, 0x63, 0x74, 0x69, 0x6f,
+	0x6e, 0x52, 0x0a, 0x70, 0x72, 0x6f, 0x74, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x22, 0x8f, 0x01,
+	0x0a, 0x1b, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e,
+	0x66, 0x69, 0x67, 0x50, 0x72, 0x6f, 0x74, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x33, 0x0a,
+	0x07, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x6c, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x19,
+	0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x47, 0x75, 0x61, 0x72,
+	0x61, 0x6e, 0x74, 0x65, 0x65, 0x4b, 0x69, 0x6e, 0x64, 0x52, 0x07, 0x49, 0x6e, 0x69, 0x74, 0x69,
+	0x61, 0x6c, 0x12, 0x3b, 0x0a, 0x0b, 0x49, 0x6e, 0x63, 0x72, 0x65, 0x6d, 0x65, 0x6e, 0x74, 0x61,
+	0x6c, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x19, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63,
+	0x61, 0x74, 0x69, 0x6f, 0x6e, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x4b, 0x69,
+	0x6e, 0x64, 0x52, 0x0b, 0x49, 0x6e, 0x63, 0x72, 0x65, 0x6d, 0x65, 0x6e, 0x74, 0x61, 0x6c, 0x22,
+	0x34, 0x0a, 0x08, 0x50, 0x72, 0x6f, 0x70, 0x65, 0x72, 0x74, 0x79, 0x12, 0x12, 0x0a, 0x04, 0x4e,
+	0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x4e, 0x61, 0x6d, 0x65, 0x12,
+	0x14, 0x0a, 0x05, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05,
+	0x56, 0x61, 0x6c, 0x75, 0x65, 0x22, 0x82, 0x01, 0x0a, 0x07, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65,
+	0x73, 0x12, 0x28, 0x0a, 0x0f, 0x55, 0x73, 0x65, 0x64, 0x52, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x54,
+	0x6f, 0x6b, 0x65, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x08, 0x52, 0x0f, 0x55, 0x73, 0x65, 0x64,
+	0x52, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x54, 0x6f, 0x6b, 0x65, 0x6e, 0x12, 0x22, 0x0a, 0x0c, 0x45,
+	0x78, 0x70, 0x65, 0x63, 0x74, 0x65, 0x64, 0x53, 0x69, 0x7a, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28,
+	0x03, 0x52, 0x0c, 0x45, 0x78, 0x70, 0x65, 0x63, 0x74, 0x65, 0x64, 0x53, 0x69, 0x7a, 0x65, 0x12,
+	0x29, 0x0a, 0x0a, 0x50, 0x72, 0x6f, 0x70, 0x65, 0x72, 0x74, 0x69, 0x65, 0x73, 0x18, 0x04, 0x20,
+	0x03, 0x28, 0x0b, 0x32, 0x09, 0x2e, 0x50, 0x72, 0x6f, 0x70, 0x65, 0x72, 0x74, 0x79, 0x52, 0x0a,
+	0x50, 0x72, 0x6f, 0x70, 0x65, 0x72, 0x74, 0x69, 0x65, 0x73, 0x22, 0x3e, 0x0a, 0x10, 0x53, 0x65,
+	0x6e, 0x64, 0x43, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x71, 0x12, 0x2a,
+	0x0a, 0x0b, 0x4f, 0x72, 0x69, 0x67, 0x69, 0x6e, 0x61, 0x6c, 0x52, 0x65, 0x71, 0x18, 0x02, 0x20,
+	0x01, 0x28, 0x0b, 0x32, 0x08, 0x2e, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65, 0x71, 0x52, 0x0b, 0x4f,
+	0x72, 0x69, 0x67, 0x69, 0x6e, 0x61, 0x6c, 0x52, 0x65, 0x71, 0x22, 0x12, 0x0a, 0x10, 0x53, 0x65,
+	0x6e, 0x64, 0x43, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x73, 0x22, 0xbe,
+	0x01, 0x0a, 0x0a, 0x52, 0x65, 0x63, 0x65, 0x69, 0x76, 0x65, 0x52, 0x65, 0x71, 0x12, 0x1e, 0x0a,
+	0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28,
+	0x09, 0x52, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x12, 0x22, 0x0a,
+	0x02, 0x54, 0x6f, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x46, 0x69, 0x6c, 0x65,
+	0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x02, 0x54,
+	0x6f, 0x12, 0x2a, 0x0a, 0x10, 0x43, 0x6c, 0x65, 0x61, 0x72, 0x52, 0x65, 0x73, 0x75, 0x6d, 0x65,
+	0x54, 0x6f, 0x6b, 0x65, 0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x08, 0x52, 0x10, 0x43, 0x6c, 0x65,
+	0x61, 0x72, 0x52, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x54, 0x6f, 0x6b, 0x65, 0x6e, 0x12, 0x40, 0x0a,
 	0x11, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66,
-	0x69, 0x67, 0x22, 0x51, 0x0a, 0x11, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f,
+	0x69, 0x67, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69,
-	0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x3c, 0x0a, 0x0a, 0x70, 0x72, 0x6f, 0x74, 0x65,
+	0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52, 0x11, 0x52, 0x65,
-	0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x52, 0x65,
+	0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x22,
-	0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x50,
+	0x0c, 0x0a, 0x0a, 0x52, 0x65, 0x63, 0x65, 0x69, 0x76, 0x65, 0x52, 0x65, 0x73, 0x22, 0x67, 0x0a,
-	0x72, 0x6f, 0x74, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x0a, 0x70, 0x72, 0x6f, 0x74, 0x65,
+	0x13, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74,
-	0x63, 0x74, 0x69, 0x6f, 0x6e, 0x22, 0x8f, 0x01, 0x0a, 0x1b, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63,
+	0x73, 0x52, 0x65, 0x71, 0x12, 0x1e, 0x0a, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74,
-	0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x50, 0x72, 0x6f, 0x74, 0x65,
+	0x65, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79,
-	0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x33, 0x0a, 0x07, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x6c,
+	0x73, 0x74, 0x65, 0x6d, 0x12, 0x30, 0x0a, 0x09, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74,
-	0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x19, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61,
+	0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79,
-	0x74, 0x69, 0x6f, 0x6e, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x4b, 0x69, 0x6e,
+	0x73, 0x74, 0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x09, 0x53, 0x6e, 0x61,
-	0x64, 0x52, 0x07, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x6c, 0x12, 0x3b, 0x0a, 0x0b, 0x49, 0x6e,
+	0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x22, 0x5a, 0x0a, 0x12, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f,
-	0x63, 0x72, 0x65, 0x6d, 0x65, 0x6e, 0x74, 0x61, 0x6c, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0e, 0x32,
+	0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x52, 0x65, 0x73, 0x12, 0x2e, 0x0a, 0x08,
-	0x19, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x47, 0x75, 0x61,
+	0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12,
-	0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x4b, 0x69, 0x6e, 0x64, 0x52, 0x0b, 0x49, 0x6e, 0x63, 0x72,
+	0x2e, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69,
-	0x65, 0x6d, 0x65, 0x6e, 0x74, 0x61, 0x6c, 0x22, 0x34, 0x0a, 0x08, 0x50, 0x72, 0x6f, 0x70, 0x65,
+	0x6f, 0x6e, 0x52, 0x08, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x12, 0x14, 0x0a, 0x05,
-	0x72, 0x74, 0x79, 0x12, 0x12, 0x0a, 0x04, 0x4e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28,
+	0x45, 0x72, 0x72, 0x6f, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x45, 0x72, 0x72,
-	0x09, 0x52, 0x04, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x56, 0x61, 0x6c, 0x75, 0x65,
+	0x6f, 0x72, 0x22, 0x44, 0x0a, 0x13, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61,
-	0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x22, 0x57, 0x0a,
+	0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x52, 0x65, 0x73, 0x12, 0x2d, 0x0a, 0x07, 0x52, 0x65, 0x73,
-	0x07, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65, 0x73, 0x12, 0x28, 0x0a, 0x0f, 0x55, 0x73, 0x65, 0x64,
+	0x75, 0x6c, 0x74, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x44, 0x65, 0x73,
-	0x52, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x54, 0x6f, 0x6b, 0x65, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28,
+	0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x52, 0x65, 0x73, 0x52,
-	0x08, 0x52, 0x0f, 0x55, 0x73, 0x65, 0x64, 0x52, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x54, 0x6f, 0x6b,
+	0x07, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x73, 0x22, 0x36, 0x0a, 0x14, 0x52, 0x65, 0x70, 0x6c,
-	0x65, 0x6e, 0x12, 0x22, 0x0a, 0x0c, 0x45, 0x78, 0x70, 0x65, 0x63, 0x74, 0x65, 0x64, 0x53, 0x69,
+	0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x75, 0x72, 0x73, 0x6f, 0x72, 0x52, 0x65, 0x71,
-	0x7a, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0c, 0x45, 0x78, 0x70, 0x65, 0x63, 0x74,
-	0x65, 0x64, 0x53, 0x69, 0x7a, 0x65, 0x22, 0x3e, 0x0a, 0x10, 0x53, 0x65, 0x6e, 0x64, 0x43, 0x6f,
-	0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x71, 0x12, 0x2a, 0x0a, 0x0b, 0x4f, 0x72,
-	0x69, 0x67, 0x69, 0x6e, 0x61, 0x6c, 0x52, 0x65, 0x71, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32,
-	0x08, 0x2e, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65, 0x71, 0x52, 0x0b, 0x4f, 0x72, 0x69, 0x67, 0x69,
-	0x6e, 0x61, 0x6c, 0x52, 0x65, 0x71, 0x22, 0x12, 0x0a, 0x10, 0x53, 0x65, 0x6e, 0x64, 0x43, 0x6f,
-	0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x73, 0x22, 0xbe, 0x01, 0x0a, 0x0a, 0x52,
-	0x65, 0x63, 0x65, 0x69, 0x76, 0x65, 0x52, 0x65, 0x71, 0x12, 0x1e, 0x0a, 0x0a, 0x46, 0x69, 0x6c,
-	0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x46,
-	0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x12, 0x22, 0x0a, 0x02, 0x54, 0x6f, 0x18,
-	0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74,
-	0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x02, 0x54, 0x6f, 0x12, 0x2a, 0x0a,
-	0x10, 0x43, 0x6c, 0x65, 0x61, 0x72, 0x52, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x54, 0x6f, 0x6b, 0x65,
-	0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x08, 0x52, 0x10, 0x43, 0x6c, 0x65, 0x61, 0x72, 0x52, 0x65,
-	0x73, 0x75, 0x6d, 0x65, 0x54, 0x6f, 0x6b, 0x65, 0x6e, 0x12, 0x40, 0x0a, 0x11, 0x52, 0x65, 0x70,
-	0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x04,
-	0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69,
-	0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52, 0x11, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63,
-	0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x22, 0x0c, 0x0a, 0x0a, 0x52,
-	0x65, 0x63, 0x65, 0x69, 0x76, 0x65, 0x52, 0x65, 0x73, 0x22, 0x67, 0x0a, 0x13, 0x44, 0x65, 0x73,
-	0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x52, 0x65, 0x71,
 	0x12, 0x1e, 0x0a, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x18, 0x01,
 	0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d,
-	0x12, 0x30, 0x0a, 0x09, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x18, 0x02, 0x20,
+	0x22, 0x54, 0x0a, 0x14, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43,
-	0x03, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d,
+	0x75, 0x72, 0x73, 0x6f, 0x72, 0x52, 0x65, 0x73, 0x12, 0x14, 0x0a, 0x04, 0x47, 0x75, 0x69, 0x64,
-	0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x09, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f,
+	0x18, 0x01, 0x20, 0x01, 0x28, 0x04, 0x48, 0x00, 0x52, 0x04, 0x47, 0x75, 0x69, 0x64, 0x12, 0x1c,
-	0x74, 0x73, 0x22, 0x5a, 0x0a, 0x12, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61,
+	0x0a, 0x08, 0x4e, 0x6f, 0x74, 0x65, 0x78, 0x69, 0x73, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x08,
-	0x70, 0x73, 0x68, 0x6f, 0x74, 0x52, 0x65, 0x73, 0x12, 0x2e, 0x0a, 0x08, 0x53, 0x6e, 0x61, 0x70,
+	0x48, 0x00, 0x52, 0x08, 0x4e, 0x6f, 0x74, 0x65, 0x78, 0x69, 0x73, 0x74, 0x42, 0x08, 0x0a, 0x06,
-	0x73, 0x68, 0x6f, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x46, 0x69, 0x6c,
+	0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x22, 0x23, 0x0a, 0x07, 0x50, 0x69, 0x6e, 0x67, 0x52, 0x65,
-	0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x08,
+	0x71, 0x12, 0x18, 0x0a, 0x07, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x01, 0x20, 0x01,
-	0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x12, 0x14, 0x0a, 0x05, 0x45, 0x72, 0x72, 0x6f,
+	0x28, 0x09, 0x52, 0x07, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x1d, 0x0a, 0x07, 0x50,
-	0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x22, 0x44,
+	0x69, 0x6e, 0x67, 0x52, 0x65, 0x73, 0x12, 0x12, 0x0a, 0x04, 0x45, 0x63, 0x68, 0x6f, 0x18, 0x01,
-	0x0a, 0x13, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f,
+	0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x45, 0x63, 0x68, 0x6f, 0x2a, 0x28, 0x0a, 0x03, 0x54, 0x72,
-	0x74, 0x73, 0x52, 0x65, 0x73, 0x12, 0x2d, 0x0a, 0x07, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x73,
+	0x69, 0x12, 0x0c, 0x0a, 0x08, 0x44, 0x6f, 0x6e, 0x74, 0x43, 0x61, 0x72, 0x65, 0x10, 0x00, 0x12,
-	0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79,
+	0x09, 0x0a, 0x05, 0x46, 0x61, 0x6c, 0x73, 0x65, 0x10, 0x01, 0x12, 0x08, 0x0a, 0x04, 0x54, 0x72,
-	0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x52, 0x65, 0x73, 0x52, 0x07, 0x52, 0x65, 0x73,
+	0x75, 0x65, 0x10, 0x02, 0x2a, 0x86, 0x01, 0x0a, 0x18, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61,
-	0x75, 0x6c, 0x74, 0x73, 0x22, 0x36, 0x0a, 0x14, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74,
+	0x74, 0x69, 0x6f, 0x6e, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x4b, 0x69, 0x6e,
-	0x69, 0x6f, 0x6e, 0x43, 0x75, 0x72, 0x73, 0x6f, 0x72, 0x52, 0x65, 0x71, 0x12, 0x1e, 0x0a, 0x0a,
+	0x64, 0x12, 0x14, 0x0a, 0x10, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x49, 0x6e,
-	0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09,
+	0x76, 0x61, 0x6c, 0x69, 0x64, 0x10, 0x00, 0x12, 0x19, 0x0a, 0x15, 0x47, 0x75, 0x61, 0x72, 0x61,
-	0x52, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x22, 0x54, 0x0a, 0x14,
+	0x6e, 0x74, 0x65, 0x65, 0x52, 0x65, 0x73, 0x75, 0x6d, 0x61, 0x62, 0x69, 0x6c, 0x69, 0x74, 0x79,
-	0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x75, 0x72, 0x73, 0x6f,
+	0x10, 0x01, 0x12, 0x23, 0x0a, 0x1f, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x49,
-	0x72, 0x52, 0x65, 0x73, 0x12, 0x14, 0x0a, 0x04, 0x47, 0x75, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01,
+	0x6e, 0x63, 0x72, 0x65, 0x6d, 0x65, 0x6e, 0x74, 0x61, 0x6c, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63,
-	0x28, 0x04, 0x48, 0x00, 0x52, 0x04, 0x47, 0x75, 0x69, 0x64, 0x12, 0x1c, 0x0a, 0x08, 0x4e, 0x6f,
+	0x61, 0x74, 0x69, 0x6f, 0x6e, 0x10, 0x02, 0x12, 0x14, 0x0a, 0x10, 0x47, 0x75, 0x61, 0x72, 0x61,
-	0x74, 0x65, 0x78, 0x69, 0x73, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x08, 0x48, 0x00, 0x52, 0x08,
+	0x6e, 0x74, 0x65, 0x65, 0x4e, 0x6f, 0x74, 0x68, 0x69, 0x6e, 0x67, 0x10, 0x03, 0x32, 0xf0, 0x02,
-	0x4e, 0x6f, 0x74, 0x65, 0x78, 0x69, 0x73, 0x74, 0x42, 0x08, 0x0a, 0x06, 0x52, 0x65, 0x73, 0x75,
+	0x0a, 0x0b, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x1a, 0x0a,
-	0x6c, 0x74, 0x22, 0x23, 0x0a, 0x07, 0x50, 0x69, 0x6e, 0x67, 0x52, 0x65, 0x71, 0x12, 0x18, 0x0a,
+	0x04, 0x50, 0x69, 0x6e, 0x67, 0x12, 0x08, 0x2e, 0x50, 0x69, 0x6e, 0x67, 0x52, 0x65, 0x71, 0x1a,
-	0x07, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07,
+	0x08, 0x2e, 0x50, 0x69, 0x6e, 0x67, 0x52, 0x65, 0x73, 0x12, 0x39, 0x0a, 0x0f, 0x4c, 0x69, 0x73,
-	0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x1d, 0x0a, 0x07, 0x50, 0x69, 0x6e, 0x67, 0x52,
+	0x74, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x73, 0x12, 0x12, 0x2e, 0x4c,
-	0x65, 0x73, 0x12, 0x12, 0x0a, 0x04, 0x45, 0x63, 0x68, 0x6f, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09,
+	0x69, 0x73, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x52, 0x65, 0x71,
-	0x52, 0x04, 0x45, 0x63, 0x68, 0x6f, 0x2a, 0x28, 0x0a, 0x03, 0x54, 0x72, 0x69, 0x12, 0x0c, 0x0a,
+	0x1a, 0x12, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65,
-	0x08, 0x44, 0x6f, 0x6e, 0x74, 0x43, 0x61, 0x72, 0x65, 0x10, 0x00, 0x12, 0x09, 0x0a, 0x05, 0x46,
+	0x6d, 0x52, 0x65, 0x73, 0x12, 0x50, 0x0a, 0x16, 0x4c, 0x69, 0x73, 0x74, 0x46, 0x69, 0x6c, 0x65,
-	0x61, 0x6c, 0x73, 0x65, 0x10, 0x01, 0x12, 0x08, 0x0a, 0x04, 0x54, 0x72, 0x75, 0x65, 0x10, 0x02,
+	0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x1a,
-	0x2a, 0x86, 0x01, 0x0a, 0x18, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e,
+	0x2e, 0x4c, 0x69, 0x73, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x56,
-	0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x4b, 0x69, 0x6e, 0x64, 0x12, 0x14, 0x0a,
+	0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x52, 0x65, 0x71, 0x1a, 0x1a, 0x2e, 0x4c, 0x69, 0x73,
-	0x10, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x49, 0x6e, 0x76, 0x61, 0x6c, 0x69,
-	0x64, 0x10, 0x00, 0x12, 0x19, 0x0a, 0x15, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65,
-	0x52, 0x65, 0x73, 0x75, 0x6d, 0x61, 0x62, 0x69, 0x6c, 0x69, 0x74, 0x79, 0x10, 0x01, 0x12, 0x23,
-	0x0a, 0x1f, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x49, 0x6e, 0x63, 0x72, 0x65,
-	0x6d, 0x65, 0x6e, 0x74, 0x61, 0x6c, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f,
-	0x6e, 0x10, 0x02, 0x12, 0x14, 0x0a, 0x10, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65,
-	0x4e, 0x6f, 0x74, 0x68, 0x69, 0x6e, 0x67, 0x10, 0x03, 0x32, 0x8f, 0x03, 0x0a, 0x0b, 0x52, 0x65,
-	0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x1a, 0x0a, 0x04, 0x50, 0x69, 0x6e,
-	0x67, 0x12, 0x08, 0x2e, 0x50, 0x69, 0x6e, 0x67, 0x52, 0x65, 0x71, 0x1a, 0x08, 0x2e, 0x50, 0x69,
-	0x6e, 0x67, 0x52, 0x65, 0x73, 0x12, 0x39, 0x0a, 0x0f, 0x4c, 0x69, 0x73, 0x74, 0x46, 0x69, 0x6c,
-	0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x73, 0x12, 0x12, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x46,
-	0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x52, 0x65, 0x71, 0x1a, 0x12, 0x2e, 0x4c,
-	0x69, 0x73, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x52, 0x65, 0x73,
-	0x12, 0x50, 0x0a, 0x16, 0x4c, 0x69, 0x73, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74,
-	0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x1a, 0x2e, 0x4c, 0x69, 0x73,
 	0x74, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69,
-	0x6f, 0x6e, 0x73, 0x52, 0x65, 0x71, 0x1a, 0x1a, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x46, 0x69, 0x6c,
+	0x6f, 0x6e, 0x73, 0x52, 0x65, 0x73, 0x12, 0x3e, 0x0a, 0x10, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f,
-	0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x52,
+	0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x12, 0x14, 0x2e, 0x44, 0x65, 0x73,
-	0x65, 0x73, 0x12, 0x3e, 0x0a, 0x10, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61,
+	0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x52, 0x65, 0x71,
-	0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x12, 0x14, 0x2e, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79,
+	0x1a, 0x14, 0x2e, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68,
-	0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x52, 0x65, 0x71, 0x1a, 0x14, 0x2e, 0x44,
+	0x6f, 0x74, 0x73, 0x52, 0x65, 0x73, 0x12, 0x41, 0x0a, 0x11, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63,
-	0x65, 0x73, 0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x52,
+	0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x75, 0x72, 0x73, 0x6f, 0x72, 0x12, 0x15, 0x2e, 0x52, 0x65,
-	0x65, 0x73, 0x12, 0x41, 0x0a, 0x11, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f,
+	0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x75, 0x72, 0x73, 0x6f, 0x72, 0x52,
-	0x6e, 0x43, 0x75, 0x72, 0x73, 0x6f, 0x72, 0x12, 0x15, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63,
+	0x65, 0x71, 0x1a, 0x15, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e,
-	0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x75, 0x72, 0x73, 0x6f, 0x72, 0x52, 0x65, 0x71, 0x1a, 0x15,
+	0x43, 0x75, 0x72, 0x73, 0x6f, 0x72, 0x52, 0x65, 0x73, 0x12, 0x35, 0x0a, 0x0d, 0x53, 0x65, 0x6e,
-	0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x75, 0x72, 0x73,
+	0x64, 0x43, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x12, 0x11, 0x2e, 0x53, 0x65, 0x6e,
-	0x6f, 0x72, 0x52, 0x65, 0x73, 0x12, 0x1d, 0x0a, 0x07, 0x53, 0x65, 0x6e, 0x64, 0x44, 0x72, 0x79,
+	0x64, 0x43, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x71, 0x1a, 0x11, 0x2e,
-	0x12, 0x08, 0x2e, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65, 0x71, 0x1a, 0x08, 0x2e, 0x53, 0x65, 0x6e,
+	0x53, 0x65, 0x6e, 0x64, 0x43, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x73,
-	0x64, 0x52, 0x65, 0x73, 0x12, 0x35, 0x0a, 0x0d, 0x53, 0x65, 0x6e, 0x64, 0x43, 0x6f, 0x6d, 0x70,
+	0x42, 0x07, 0x5a, 0x05, 0x2e, 0x3b, 0x70, 0x64, 0x75, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f,
-	0x6c, 0x65, 0x74, 0x65, 0x64, 0x12, 0x11, 0x2e, 0x53, 0x65, 0x6e, 0x64, 0x43, 0x6f, 0x6d, 0x70,
+	0x33,
-	0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x71, 0x1a, 0x11, 0x2e, 0x53, 0x65, 0x6e, 0x64, 0x43,
-	0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x73, 0x42, 0x07, 0x5a, 0x05, 0x2e,
-	0x3b, 0x70, 0x64, 0x75, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -1606,31 +1625,30 @@ var file_pdu_proto_depIdxs = []int32{
 	11, // 7: ReplicationConfig.protection:type_name -> ReplicationConfigProtection
 	1,  // 8: ReplicationConfigProtection.Initial:type_name -> ReplicationGuaranteeKind
 	1,  // 9: ReplicationConfigProtection.Incremental:type_name -> ReplicationGuaranteeKind
-	9,  // 10: SendCompletedReq.OriginalReq:type_name -> SendReq
+	12, // 10: SendRes.Properties:type_name -> Property
-	8,  // 11: ReceiveReq.To:type_name -> FilesystemVersion
+	9,  // 11: SendCompletedReq.OriginalReq:type_name -> SendReq
-	10, // 12: ReceiveReq.ReplicationConfig:type_name -> ReplicationConfig
+	8,  // 12: ReceiveReq.To:type_name -> FilesystemVersion
-	8,  // 13: DestroySnapshotsReq.Snapshots:type_name -> FilesystemVersion
+	10, // 13: ReceiveReq.ReplicationConfig:type_name -> ReplicationConfig
-	8,  // 14: DestroySnapshotRes.Snapshot:type_name -> FilesystemVersion
+	8,  // 14: DestroySnapshotsReq.Snapshots:type_name -> FilesystemVersion
-	19, // 15: DestroySnapshotsRes.Results:type_name -> DestroySnapshotRes
+	8,  // 15: DestroySnapshotRes.Snapshot:type_name -> FilesystemVersion
-	23, // 16: Replication.Ping:input_type -> PingReq
+	19, // 16: DestroySnapshotsRes.Results:type_name -> DestroySnapshotRes
-	3,  // 17: Replication.ListFilesystems:input_type -> ListFilesystemReq
+	23, // 17: Replication.Ping:input_type -> PingReq
-	6,  // 18: Replication.ListFilesystemVersions:input_type -> ListFilesystemVersionsReq
+	3,  // 18: Replication.ListFilesystems:input_type -> ListFilesystemReq
-	18, // 19: Replication.DestroySnapshots:input_type -> DestroySnapshotsReq
+	6,  // 19: Replication.ListFilesystemVersions:input_type -> ListFilesystemVersionsReq
-	21, // 20: Replication.ReplicationCursor:input_type -> ReplicationCursorReq
+	18, // 20: Replication.DestroySnapshots:input_type -> DestroySnapshotsReq
-	9,  // 21: Replication.SendDry:input_type -> SendReq
+	21, // 21: Replication.ReplicationCursor:input_type -> ReplicationCursorReq
 	14, // 22: Replication.SendCompleted:input_type -> SendCompletedReq
 	24, // 23: Replication.Ping:output_type -> PingRes
 	4,  // 24: Replication.ListFilesystems:output_type -> ListFilesystemRes
 	7,  // 25: Replication.ListFilesystemVersions:output_type -> ListFilesystemVersionsRes
 	20, // 26: Replication.DestroySnapshots:output_type -> DestroySnapshotsRes
 	22, // 27: Replication.ReplicationCursor:output_type -> ReplicationCursorRes
-	13, // 28: Replication.SendDry:output_type -> SendRes
+	15, // 28: Replication.SendCompleted:output_type -> SendCompletedRes
-	15, // 29: Replication.SendCompleted:output_type -> SendCompletedRes
+	23, // [23:29] is the sub-list for method output_type
-	23, // [23:30] is the sub-list for method output_type
+	17, // [17:23] is the sub-list for method input_type
-	16, // [16:23] is the sub-list for method input_type
+	17, // [17:17] is the sub-list for extension type_name
-	16, // [16:16] is the sub-list for extension type_name
+	17, // [17:17] is the sub-list for extension extendee
-	16, // [16:16] is the sub-list for extension extendee
+	0,  // [0:17] is the sub-list for field type_name
-	0,  // [0:16] is the sub-list for field type_name
 }
 
 func init() { file_pdu_proto_init() }

replication/logic/pdu/pdu.proto

@@ -8,7 +8,6 @@ service Replication {
       returns (ListFilesystemVersionsRes);
   rpc DestroySnapshots(DestroySnapshotsReq) returns (DestroySnapshotsRes);
   rpc ReplicationCursor(ReplicationCursorReq) returns (ReplicationCursorRes);
-  rpc SendDry(SendReq) returns (SendRes);
   rpc SendCompleted(SendCompletedReq) returns (SendCompletedRes);
   // for Send and Recv, see package rpc
 }
@@ -61,7 +60,9 @@ message SendReq {
   string ResumeToken = 4;
   Tri Encrypted = 5;
 
-  ReplicationConfig ReplicationConfig = 6;
+  bool DryRun = 6;
+
+  ReplicationConfig ReplicationConfig = 7;
 }
 
 message ReplicationConfig {
@@ -88,12 +89,14 @@ message Property {
 
 message SendRes {
   // Whether the resume token provided in the request has been used or not.
-  // If the SendReq.ResumeToken == "", this field MUST be false.
+  // If the SendReq.ResumeToken == "", this field has no meaning.
-  bool UsedResumeToken = 1;
+  bool UsedResumeToken = 2;
 
   // Expected stream size determined by dry run, not exact.
   // 0 indicates that for the given SendReq, no size estimate could be made.
-  uint64 ExpectedSize = 2;
+  int64 ExpectedSize = 3;
+
+  repeated Property Properties = 4;
 }
 
 message SendCompletedReq {

replication/logic/pdu/pdu_grpc.pb.go

@@ -23,7 +23,6 @@ type ReplicationClient interface {
 	ListFilesystemVersions(ctx context.Context, in *ListFilesystemVersionsReq, opts ...grpc.CallOption) (*ListFilesystemVersionsRes, error)
 	DestroySnapshots(ctx context.Context, in *DestroySnapshotsReq, opts ...grpc.CallOption) (*DestroySnapshotsRes, error)
 	ReplicationCursor(ctx context.Context, in *ReplicationCursorReq, opts ...grpc.CallOption) (*ReplicationCursorRes, error)
-	SendDry(ctx context.Context, in *SendReq, opts ...grpc.CallOption) (*SendRes, error)
 	SendCompleted(ctx context.Context, in *SendCompletedReq, opts ...grpc.CallOption) (*SendCompletedRes, error)
 }
 
@@ -80,15 +79,6 @@ func (c *replicationClient) ReplicationCursor(ctx context.Context, in *Replicati
 	return out, nil
 }
 
-func (c *replicationClient) SendDry(ctx context.Context, in *SendReq, opts ...grpc.CallOption) (*SendRes, error) {
-	out := new(SendRes)
-	err := c.cc.Invoke(ctx, "/Replication/SendDry", in, out, opts...)
-	if err != nil {
-		return nil, err
-	}
-	return out, nil
-}
-
 func (c *replicationClient) SendCompleted(ctx context.Context, in *SendCompletedReq, opts ...grpc.CallOption) (*SendCompletedRes, error) {
 	out := new(SendCompletedRes)
 	err := c.cc.Invoke(ctx, "/Replication/SendCompleted", in, out, opts...)
@@ -107,7 +97,6 @@ type ReplicationServer interface {
 	ListFilesystemVersions(context.Context, *ListFilesystemVersionsReq) (*ListFilesystemVersionsRes, error)
 	DestroySnapshots(context.Context, *DestroySnapshotsReq) (*DestroySnapshotsRes, error)
 	ReplicationCursor(context.Context, *ReplicationCursorReq) (*ReplicationCursorRes, error)
-	SendDry(context.Context, *SendReq) (*SendRes, error)
 	SendCompleted(context.Context, *SendCompletedReq) (*SendCompletedRes, error)
 	mustEmbedUnimplementedReplicationServer()
 }
@@ -131,9 +120,6 @@ func (UnimplementedReplicationServer) DestroySnapshots(context.Context, *Destroy
 func (UnimplementedReplicationServer) ReplicationCursor(context.Context, *ReplicationCursorReq) (*ReplicationCursorRes, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method ReplicationCursor not implemented")
 }
-func (UnimplementedReplicationServer) SendDry(context.Context, *SendReq) (*SendRes, error) {
-	return nil, status.Errorf(codes.Unimplemented, "method SendDry not implemented")
-}
 func (UnimplementedReplicationServer) SendCompleted(context.Context, *SendCompletedReq) (*SendCompletedRes, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method SendCompleted not implemented")
 }
@@ -240,24 +226,6 @@ func _Replication_ReplicationCursor_Handler(srv interface{}, ctx context.Context
 	return interceptor(ctx, in, info, handler)
 }
 
-func _Replication_SendDry_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := new(SendReq)
-	if err := dec(in); err != nil {
-		return nil, err
-	}
-	if interceptor == nil {
-		return srv.(ReplicationServer).SendDry(ctx, in)
-	}
-	info := &grpc.UnaryServerInfo{
-		Server:     srv,
-		FullMethod: "/Replication/SendDry",
-	}
-	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(ReplicationServer).SendDry(ctx, req.(*SendReq))
-	}
-	return interceptor(ctx, in, info, handler)
-}
-
 func _Replication_SendCompleted_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(SendCompletedReq)
 	if err := dec(in); err != nil {
@@ -303,10 +271,6 @@ var Replication_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "ReplicationCursor",
 			Handler:    _Replication_ReplicationCursor_Handler,
 		},
-		{
-			MethodName: "SendDry",
-			Handler:    _Replication_SendDry_Handler,
-		},
 		{
 			MethodName: "SendCompleted",
 			Handler:    _Replication_SendCompleted_Handler,

replication/logic/replication_logic.go

@@ -41,7 +41,6 @@ type Sender interface {
 	// 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)
-	SendDry(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, error)
 	SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error)
 	ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error)
 }
@@ -159,7 +158,7 @@ type Step struct {
 	encrypt     tri
 	resumeToken string // empty means no resume token shall be used
 
-	expectedSize uint64 // 0 means no size estimate present / possible
+	expectedSize int64 // 0 means no size estimate present / possible
 
 	// byteCounter is nil initially, and set later in Step.doReplication
 	// => concurrent read of that pointer from Step.ReportInfo must be protected
@@ -190,7 +189,7 @@ func (s *Step) Step(ctx context.Context) error {
 func (s *Step) ReportInfo() *report.StepInfo {
 
 	// get current byteCounter value
-	var byteCounter uint64
+	var byteCounter int64
 	s.byteCounterMtx.Lock()
 	if s.byteCounter != nil {
 		byteCounter = s.byteCounter.Count()
@@ -547,10 +546,10 @@ func (s *Step) updateSizeEstimate(ctx context.Context) error {
 
 	log := getLogger(ctx)
 
-	sr := s.buildSendRequest()
+	sr := s.buildSendRequest(true)
 
 	log.Debug("initiate dry run send request")
-	sres, err := s.sender.SendDry(ctx, sr)
+	sres, _, err := s.sender.Send(ctx, sr)
 	if err != nil {
 		log.WithError(err).Error("dry run send request failed")
 		return err
@@ -564,7 +563,7 @@ func (s *Step) updateSizeEstimate(ctx context.Context) error {
 	return nil
 }
 
-func (s *Step) buildSendRequest() (sr *pdu.SendReq) {
+func (s *Step) buildSendRequest(dryRun bool) (sr *pdu.SendReq) {
 	fs := s.parent.Path
 	sr = &pdu.SendReq{
 		Filesystem:        fs,
@@ -572,6 +571,7 @@ func (s *Step) buildSendRequest() (sr *pdu.SendReq) {
 		To:                s.to,
 		Encrypted:         s.encrypt.ToPDU(),
 		ResumeToken:       s.resumeToken,
+		DryRun:            dryRun,
 		ReplicationConfig: s.parent.policy.ReplicationConfig,
 	}
 	return sr
@@ -582,7 +582,7 @@ func (s *Step) doReplication(ctx context.Context) error {
 	fs := s.parent.Path
 
 	log := getLogger(ctx).WithField("filesystem", fs)
-	sr := s.buildSendRequest()
+	sr := s.buildSendRequest(false)
 
 	log.Debug("initiate send request")
 	sres, stream, err := s.sender.Send(ctx, sr)

replication/report/replication_report.go

@@ -97,11 +97,11 @@ type StepInfo struct {
 	From, To        string
 	Resumed         bool
 	Encrypted       EncryptedEnum
-	BytesExpected   uint64
+	BytesExpected   int64
-	BytesReplicated uint64
+	BytesReplicated int64
 }
 
-func (a *AttemptReport) BytesSum() (expected, replicated uint64, containsInvalidSizeEstimates bool) {
+func (a *AttemptReport) BytesSum() (expected, replicated int64, containsInvalidSizeEstimates bool) {
 	for _, fs := range a.Filesystems {
 		e, r, fsContainsInvalidEstimate := fs.BytesSum()
 		containsInvalidSizeEstimates = containsInvalidSizeEstimates || fsContainsInvalidEstimate
@@ -111,7 +111,7 @@ func (a *AttemptReport) BytesSum() (expected, replicated uint64, containsInvalid
 	return expected, replicated, containsInvalidSizeEstimates
 }
 
-func (f *FilesystemReport) BytesSum() (expected, replicated uint64, containsInvalidSizeEstimates bool) {
+func (f *FilesystemReport) BytesSum() (expected, replicated int64, containsInvalidSizeEstimates bool) {
 	for _, step := range f.Steps {
 		expected += step.Info.BytesExpected
 		replicated += step.Info.BytesReplicated

rpc/dataconn/dataconn_client.go

@@ -114,6 +114,12 @@ func (c *Client) ReqSend(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, i
 	if err != nil {
 		return nil, nil, err
 	}
+	putWireOnReturn := true
+	defer func() {
+		if putWireOnReturn {
+			c.putWire(conn)
+		}
+	}()
 
 	if err := c.send(ctx, conn, EndpointSend, req, nil); err != nil {
 		return nil, nil, err
@@ -125,10 +131,14 @@ func (c *Client) ReqSend(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, i
 	}
 
 	var stream io.ReadCloser
+	if !req.DryRun {
+		putWireOnReturn = false
 		stream, err = conn.ReadStream(ZFSStream, true) // no shadow
 		if err != nil {
 			return nil, nil, err
 		}
+	}
+
 	return &res, stream, nil
 }
 

rpc/dataconn/dataconn_server.go

@@ -172,15 +172,6 @@ func (s *Server) serveConnRequest(ctx context.Context, endpoint string, c *strea
 			return
 		}
 		res, sendStream, handlerErr = s.h.Send(ctx, &req) // SHADOWING
-		// ensure that we always close the sendStream
-		if sendStream != nil {
-			defer func() {
-				err := sendStream.Close()
-				if err != nil {
-					s.log.WithError(err).Error("cannot close send stream")
-				}
-			}()
-		}
 	case EndpointRecv:
 		var req pdu.ReceiveReq
 		if err := proto.Unmarshal(reqStructured, &req); err != nil {
@@ -248,9 +239,12 @@ func (s *Server) serveConnRequest(ctx context.Context, endpoint string, c *strea
 
 	if sendStream != nil {
 		err := c.SendStream(ctx, sendStream, ZFSStream)
+		closeErr := sendStream.Close()
+		if closeErr != nil {
+			s.log.WithError(closeErr).Error("cannot close send stream")
+		}
 		if err != nil {
 			s.log.WithError(err).Error("cannot write send stream")
 		}
-		// sendStream.Close() done via defer above
 	}
 }

rpc/dataconn/timeoutconn/timeoutconn_readv_unsupported.go

@@ -1,4 +1,3 @@
-//go:build illumos || solaris
 // +build illumos solaris
 
 package timeoutconn

rpc/dataconn/timeoutconn/timoutconn_readv.go

@@ -1,5 +1,5 @@
-//go:build !illumos && !solaris
+// +build !illumos
-// +build !illumos,!solaris
+// +build !solaris
 
 package timeoutconn
 

rpc/rpc_client.go

@@ -108,13 +108,6 @@ func (c *Client) Receive(ctx context.Context, req *pdu.ReceiveReq, stream io.Rea
 	return c.dataClient.ReqRecv(ctx, req, stream)
 }
 
-func (c *Client) SendDry(ctx context.Context, in *pdu.SendReq) (*pdu.SendRes, error) {
-	ctx, endSpan := trace.WithSpan(ctx, "rpc.client.SendDry")
-	defer endSpan()
-
-	return c.controlClient.SendDry(ctx, in)
-}
-
 func (c *Client) ListFilesystems(ctx context.Context, in *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error) {
 	ctx, endSpan := trace.WithSpan(ctx, "rpc.client.ListFilesystems")
 	defer endSpan()

rpc/transportmux/transportmux.go

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

rpc/versionhandshake/versionhandshake.go

@@ -10,9 +10,12 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"os"
 	"strings"
 	"time"
 	"unicode/utf8"
+
+	"github.com/kr/pretty"
 )
 
 type HandshakeMessage struct {
@@ -93,6 +96,7 @@ func (m *HandshakeMessage) Encode() ([]byte, error) {
 func (m *HandshakeMessage) DecodeReader(r io.Reader, maxLen int) error {
 	var lenAndSpace [11]byte
 	if _, err := io.ReadFull(r, lenAndSpace[:]); err != nil {
+		fmt.Fprintf(os.Stderr, "HandshakeMessage.DecodeReader error: %T\n%s", err, pretty.Sprint(err))
 		return hsIOErr(err, "error reading protocol banner length: %s", err)
 	}
 	if !utf8.Valid(lenAndSpace[:]) {
@@ -152,7 +156,7 @@ func (m *HandshakeMessage) DecodeReader(r io.Reader, maxLen int) error {
 
 func DoHandshakeCurrentVersion(conn net.Conn, deadline time.Time) *HandshakeError {
 	// current protocol version is hardcoded here
-	return DoHandshakeVersion(conn, deadline, 6)
+	return DoHandshakeVersion(conn, deadline, 5)
 }
 
 const HandshakeMessageMaxLen = 16 * 4096

rpc/versionhandshake/versionhandshake_transport_wrappers.go

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

transport/tls/connect_tls.go

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

transport/tls/issue-517/main.go

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

transport/tls/serve_tls.go

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

util/bytecounter/bytecounter_readcloser.go

@@ -9,7 +9,7 @@ import (
 // its interface and counting the bytes written to during copying.
 type ReadCloser interface {
 	io.ReadCloser
-	Count() uint64
+	Count() int64
 }
 
 // NewReadCloser wraps rc.
@@ -19,11 +19,11 @@ func NewReadCloser(rc io.ReadCloser) ReadCloser {
 
 type readCloser struct {
 	rc    io.ReadCloser
-	count uint64
+	count int64
 }
 
-func (r *readCloser) Count() uint64 {
+func (r *readCloser) Count() int64 {
-	return atomic.LoadUint64(&r.count)
+	return atomic.LoadInt64(&r.count)
 }
 
 var _ io.ReadCloser = &readCloser{}
@@ -34,9 +34,6 @@ func (r *readCloser) Close() error {
 
 func (r *readCloser) Read(p []byte) (int, error) {
 	n, err := r.rc.Read(p)
-	if n < 0 {
+	atomic.AddInt64(&r.count, int64(n))
-		panic("expecting n >= 0")
-	}
-	atomic.AddUint64(&r.count, uint64(n))
 	return n, err
 }

util/envconst/envconst.go

@@ -73,24 +73,6 @@ func Int64(varname string, def int64) (d int64) {
 	return d
 }
 
-func Uint64(varname string, def uint64) (d uint64) {
-	var err error
-	if v, ok := cache.Load(varname); ok {
-		return v.(uint64)
-	}
-	e := os.Getenv(varname)
-	if e == "" {
-		d = def
-	} else {
-		d, err = strconv.ParseUint(e, 10, 64)
-		if err != nil {
-			panic(err)
-		}
-	}
-	cache.Store(varname, d)
-	return d
-}
-
 func Bool(varname string, def bool) (d bool) {
 	var err error
 	if v, ok := cache.Load(varname); ok {

util/tcpsock/tcpsock_freebind_freebsd.go

@@ -1,4 +1,3 @@
-//go:build freebsd
 // +build freebsd
 
 package tcpsock

util/tcpsock/tcpsock_freebind_linux.go

@@ -1,4 +1,3 @@
-//go:build linux
 // +build linux
 
 package tcpsock

util/tcpsock/tcpsock_freebind_unsupported.go

@@ -1,4 +1,3 @@
-//go:build !linux && !freebsd
 // +build !linux,!freebsd
 
 package tcpsock

zfs/zfs.go

@@ -7,7 +7,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
-	"math"
 	"os"
 	"os/exec"
 	"regexp"
@@ -15,6 +14,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
@@ -334,122 +334,60 @@ func pipeWithCapacityHint(capacity int) (r, w *os.File, err error) {
 	return stdoutReader, stdoutWriter, nil
 }
 
-type sendStreamState int
-
-const (
-	sendStreamOpen sendStreamState = iota
-	sendStreamClosed
-)
-
 type SendStream struct {
 	cmd  *zfscmd.Cmd
 	kill context.CancelFunc
-	stdoutReader io.ReadCloser // not *os.File for mocking during platformtest
+
+	closeMtx     sync.Mutex
+	stdoutReader *os.File
 	stderrBuf    *circlog.CircularLog
-
+	opErr        error
-	mtx     sync.Mutex
-	state   sendStreamState
-	exitErr *ZFSError
 }
 
-func (s *SendStream) Read(p []byte) (n int, _ error) {
+func (s *SendStream) Read(p []byte) (n int, err error) {
-	s.mtx.Lock()
+	s.closeMtx.Lock()
-	defer s.mtx.Unlock()
+	opErr := s.opErr
+	s.closeMtx.Unlock()
+	if opErr != nil {
+		return 0, opErr
+	}
 
-	switch s.state {
+	n, err = s.stdoutReader.Read(p)
-	case sendStreamClosed:
+	if err != nil {
-		return 0, os.ErrClosed
+		debug("sendStream: read err: %T %s", err, err)
-
+		// TODO we assume here that any read error is permanent
-	case sendStreamOpen:
+		// which is most likely the case for a local zfs send
-		n, readErr := s.stdoutReader.Read(p)
+		kwerr := s.killAndWait(err)
-		if readErr != nil {
+		debug("sendStream: killAndWait n=%v err= %T %s", n, kwerr, kwerr)
-			debug("sendStream: read: readErr=%T %s", readErr, readErr)
+		// TODO we assume here that any read error is permanent
-			if readErr == io.EOF {
+		return n, kwerr
-				// io.EOF must be bubbled up as is so that consumers can handle it properly.
-				return n, readErr
-			}
-			// Assume that the error is not retryable.
-			// Try to kill now so that we can return a nice *ZFSError with captured stderr.
-			// If the kill doesn't work, it doesn't matter because the caller must by contract call Close() anyways.
-			killErr := s.killAndWait()
-			debug("sendStream: read: killErr=%T %s", killErr, killErr)
-			if killErr == nil {
-				s.state = sendStreamClosed
-				return n, s.exitErr // return the nice error
-			} else {
-				// we remain open so that we retry
-				return n, readErr // return the normal error
-			}
-		}
-		return n, readErr
-
-	default:
-		panic("unreachable")
 	}
+	return n, err
 }
 
 func (s *SendStream) Close() error {
 	debug("sendStream: close called")
-	s.mtx.Lock()
+	return s.killAndWait(nil)
-	defer s.mtx.Unlock()
-
-	switch s.state {
-	case sendStreamOpen:
-		err := s.killAndWait()
-		if err != nil {
-			return err
-		} else {
-			s.state = sendStreamClosed
-			return nil
-		}
-	case sendStreamClosed:
-		return os.ErrClosed
-	default:
-		panic("unreachable")
-	}
 }
 
-// returns nil iff the child process is gone (has been successfully waited upon)
+func (s *SendStream) killAndWait(precedingReadErr error) error {
-// in that case, s.exitErr is set
-func (s *SendStream) killAndWait() error {
 
 	debug("sendStream: killAndWait enter")
 	defer debug("sendStream: killAndWait leave")
-
+	if precedingReadErr == io.EOF {
-	// send SIGKILL
+		// give the zfs process a little bit of time to terminate itself
-	s.kill()
+		// if it holds this deadline, exitErr will be nil
-
+		time.AfterFunc(200*time.Millisecond, s.kill)
-	// Close our read-end of the pipe.
-	//
-	// We must do this before .Wait() because in some (not all) versions/build configs of ZFS,
-	// `zfs send` uses a separate kernel thread (taskq) to write the send stream (function `dump_bytes`).
-	// The `zfs send` thread then waits uinterruptably for the taskq thread to finish the write.
-	// And signalling the `zfs send` thread doesn't propagate to the taskq thread.
-	// So we end up in a state where we .Wait() forever.
-	// (See https://github.com/openzfs/zfs/issues/12500 and
-	//  https://github.com/zrepl/zrepl/issues/495#issuecomment-902530043)
-	//
-	// By closing our read end of the pipe before .Wait(), we unblock the taskq thread if there is any.
-	// If there is no separate taskq thread, the SIGKILL to `zfs end` would suffice and be most precise,
-	// but due to the circumstances above, there is no other portable & robust way.
-	//
-	// However, the fallout from closing the pipe is that (in non-taskq builds) `zfs sends` will get a SIGPIPE.
-	// And on Linux, that SIGPIPE appears to win over the previously issued SIGKILL.
-	// And thus, on Linux, the `zfs send` will be killed by the default SIGPIPE handler.
-	// We can observe this in the WaitStatus below.
-	// This behavior is slightly annoying because the *exec.ExitError's message ("signal: broken pipe")
-	// isn't as clear as ("signal: killed").
-	// However, it seems like we just have to live with that. (covered by platformtest)
-	var closePipeErr error
-	if s.stdoutReader != nil {
-		closePipeErr = s.stdoutReader.Close()
-		if closePipeErr == nil {
-			// avoid double-closes in case waiting below doesn't work
-			// and someone attempts Close again
-			s.stdoutReader = nil
 	} else {
-			return closePipeErr
+		s.kill()
 	}
+
+	// allow async kills from Close(), that's why we only take the mutex here
+	s.closeMtx.Lock()
+	defer s.closeMtx.Unlock()
+
+	if s.opErr != nil {
+		return s.opErr
 	}
 
 	waitErr := s.cmd.Wait()
@@ -464,30 +402,39 @@ func (s *SendStream) killAndWait() error {
 		}
 	}
 
-	// invariant: at this point, the child is gone and we cleaned up everything related to the SendStream
+	// now, after we know the program exited do we close the pipe
+	var closePipeErr error
+	if s.stdoutReader != nil {
+		closePipeErr = s.stdoutReader.Close()
+		if closePipeErr == nil {
+			// avoid double-closes in case anything below doesn't work
+			// and someone calls Close again
+			s.stdoutReader = nil
+		} else {
+			return closePipeErr
+		}
+	}
 
+	// we managed to tear things down, no let's give the user some pretty *ZFSError
 	if exitErr != nil {
-		// zfs send exited with an error or was killed by a signal.
+		s.opErr = &ZFSError{
-		s.exitErr = &ZFSError{
 			Stderr:  []byte(s.stderrBuf.String()),
 			WaitErr: exitErr,
 		}
 	} else {
-		// zfs send exited successfully (we know that since waitErr was either nil or wasn't an *exec.ExitError)
+		s.opErr = precedingReadErr
-		s.exitErr = nil
 	}
 
-	return nil
+	// detect the edge where we're called from s.Read
+	// after the pipe EOFed and zfs send exited without errors
+	// this is actually the "hot" / nice path
+	if exitErr == nil && precedingReadErr == io.EOF {
+		return precedingReadErr
 	}
 
-func (s *SendStream) TestOnly_ReplaceStdoutReader(f io.ReadCloser) (prev io.ReadCloser) {
+	return s.opErr
-	prev = s.stdoutReader
-	s.stdoutReader = f
-	return prev
 }
 
-func (s *SendStream) TestOnly_ExitErr() *ZFSError { return s.exitErr }
-
 // NOTE: When updating this struct, make sure to update funcs Validate ValidateCorrespondsToResumeToken
 type ZFSSendArgVersion struct {
 	RelName string
@@ -961,7 +908,7 @@ type DrySendInfo struct {
 	Type         DrySendType
 	Filesystem   string // parsed from To field
 	From, To     string // direct copy from ZFS output
-	SizeEstimate uint64 // 0 if size estimate is not possible
+	SizeEstimate int64  // -1 if size estimate is not possible
 }
 
 var (
@@ -1034,10 +981,11 @@ func (s *DrySendInfo) unmarshalInfoLine(l string) (regexMatched bool, err error)
 		// see https://github.com/zrepl/zrepl/issues/289
 		fields["size"] = "0"
 	}
-	s.SizeEstimate, err = strconv.ParseUint(fields["size"], 10, 64)
+	s.SizeEstimate, err = strconv.ParseInt(fields["size"], 10, 64)
 	if err != nil {
 		return true, fmt.Errorf("cannot not parse size: %s", err)
 	}
+
 	return true, nil
 }
 
@@ -1047,7 +995,6 @@ func ZFSSendDry(ctx context.Context, sendArgs ZFSSendArgsValidated) (_ *DrySendI
 
 	if sendArgs.From != nil && strings.Contains(sendArgs.From.RelName, "#") {
 		/* TODO:
-		 * XXX feature check & support this as well
 		 * ZFS at the time of writing does not support dry-run send because size-estimation
 		 * uses fromSnap's deadlist. However, for a bookmark, that deadlist no longer exists.
 		 * Redacted send & recv will bring this functionality, see
@@ -1066,7 +1013,7 @@ func ZFSSendDry(ctx context.Context, sendArgs ZFSSendArgsValidated) (_ *DrySendI
 			Filesystem:   sendArgs.FS,
 			From:         fromAbs,
 			To:           toAbs,
-			SizeEstimate: 0}, nil
+			SizeEstimate: -1}, nil
 	}
 
 	args := make([]string, 0)
@@ -1086,19 +1033,6 @@ func ZFSSendDry(ctx context.Context, sendArgs ZFSSendArgsValidated) (_ *DrySendI
 	if err := si.unmarshalZFSOutput(output); err != nil {
 		return nil, fmt.Errorf("could not parse zfs send -n output: %s", err)
 	}
-
-	// There is a bug in OpenZFS where it estimates the size incorrectly.
-	// - zrepl: https://github.com/zrepl/zrepl/issues/463
-	// - resulting upstream bug: https://github.com/openzfs/zfs/issues/12265
-	//
-	// The wrong estimates are easy to detect because they are absurdly large.
-	// NB: we're doing the workaround for this late so that the test cases are not affected.
-	sizeEstimateThreshold := envconst.Uint64("ZREPL_ZFS_SEND_SIZE_ESTIMATE_INCORRECT_THRESHOLD", math.MaxInt64)
-	if sizeEstimateThreshold != 0 && si.SizeEstimate >= sizeEstimateThreshold {
-		debug("size estimate exceeds threshold %v, working around it: %#v %q", sizeEstimateThreshold, si, args)
-		si.SizeEstimate = 0
-	}
-
 	return &si, nil
 }
 

zfs/zfs_pipe.go

@@ -1,4 +1,3 @@
-//go:build !linux
 // +build !linux
 
 package zfs

zfs/zfscmd/zfscmd.go

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