WIP fix racy holds

fixes #280

README.md

@@ -1,8 +1,9 @@
 [![GitHub license](https://img.shields.io/github/license/zrepl/zrepl.svg)](https://github.com/zrepl/zrepl/blob/master/LICENSE)
 [![Language: Go](https://img.shields.io/badge/language-Go-6ad7e5.svg)](https://golang.org/)
 [![User Docs](https://img.shields.io/badge/docs-web-blue.svg)](https://zrepl.github.io)
-[![Donate via PayPal](https://img.shields.io/badge/donate-paypal-yellow.svg)](https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=R5QSXJVYHGX96)
+[![Donate via Patreon](https://img.shields.io/endpoint.svg?url=https%3A%2F%2Fshieldsio-patreon.herokuapp.com%2Fzrepl%2Fpledges&style=flat&color=yellow)](https://www.patreon.com/zrepl)
 [![Donate via Liberapay](https://img.shields.io/liberapay/receives/zrepl.svg?logo=liberapay)](https://liberapay.com/zrepl/donate)
+[![Donate via PayPal](https://img.shields.io/badge/donate-paypal-yellow.svg)](https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=R5QSXJVYHGX96)
 [![Twitter](https://img.shields.io/twitter/url/https/github.com/zrepl/zrepl.svg?style=social)](https://twitter.com/intent/tweet?text=Wow:&url=https%3A%2F%2Fgithub.com%2Fzrepl%2Fzrepl)
 
 # zrepl
@@ -23,6 +24,7 @@ zrepl is a one-stop ZFS backup & replication solution.
    If so, think of an expressive configuration example.
 2. Think of at least one use case that generalizes from your concrete application.
 3. Open an issue on GitHub with example conf & use case attached.
+4. **Optional**: [Post a bounty](https://www.bountysource.com/teams/zrepl) on the issue, or [contact Christian Schwarz](https://cschwarz.com) for contract work.
 
 The above does not apply if you already implemented everything.
 Check out the *Coding Workflow* section below for details.
@@ -46,11 +48,8 @@ zrepl is written in [Go](https://golang.org) and uses [Go modules](https://githu
 The documentation is written in [ReStructured Text](http://docutils.sourceforge.net/rst.html) using the [Sphinx](https://www.sphinx-doc.org) framework.
 
 To get started, run `./lazy.sh devsetup` to easily install build dependencies and read `docs/installation.rst -> Compiling from Source`.
-
+`lazy.sh` uses `python3-pip` to fetch the build dependencies for the docs - you might want to use a [venv](https://docs.python.org/3/library/venv.html).
-### Overall Architecture
+If you just want to install the Go dependencies, run `./lazy.sh godep`.
-
-The application architecture is documented as part of the user docs in the *Implementation* section (`docs/content/impl`).
-Make sure to develop an understanding how zrepl is typically used by studying the user docs first.
 
 ### Project Structure
 
@@ -62,14 +61,15 @@ Make sure to develop an understanding how zrepl is typically used by studying th
 │   └── samples
 ├── daemon                  # the implementation of `zrepl daemon` subcommand
 │   ├── filters
+│   ├── hooks               # snapshot hooks
 │   ├── job                 # job implementations
 │   ├── logging             # logging outlets + formatters
 │   ├── nethelpers
 │   ├── prometheus
 │   ├── pruner              # pruner implementation
 │   ├── snapper             # snapshotter implementation
-├── docs                    # sphinx-based documentation
 ├── dist                    # supplemental material for users & package maintainers
+├── docs                    # sphinx-based documentation
 │   ├── **/*.rst            # documentation in reStructuredText
 │   ├── sphinxconf
 │   │   └── conf.py         # sphinx config (see commit 445a280 why its not in docs/)
@@ -78,6 +78,7 @@ Make sure to develop an understanding how zrepl is typically used by studying th
 │   └── public_git          # checkout of zrepl.github.io managed by above shell script
 ├── endpoint                # implementation of replication endpoints (=> package replication)
 ├── logger                  # our own logger package
+├── platformtest            # test suite for our zfs abstractions (error classification, etc)
 ├── pruning                 # pruning rules (the logic, not the actual execution)
 │   └── retentiongrid
 ├── replication
@@ -92,14 +93,13 @@ Make sure to develop an understanding how zrepl is typically used by studying th
 │   ├── transportmux        # TCP connecter and listener used to split control & data traffic
 │   └── versionhandshake    # replication protocol version handshake perfomed on newly established connections
 ├── tlsconf                 # abstraction for Go TLS server + client config
-├── transport               # transports implementation
+├── transport               # transport implementations
 │   ├── fromconfig
 │   ├── local
 │   ├── ssh
 │   ├── tcp
 │   └── tls
 ├── util
-├── vendor                  # managed by dep
 ├── version                 # abstraction for versions (filled during build by Makefile)
 └── zfs                     # zfs(8) wrappers
 ```
@@ -134,3 +134,15 @@ There will not be a big refactoring (an attempt was made, but it's destroying to
 
 However, new contributions & patches should fix naming without further notice in the commit message.
 
+### RPC debugging
+
+Optionally, there are various RPC-related environment varibles, that if set to something != `""` will produce additional debug output on stderr:
+
+https://github.com/zrepl/zrepl/blob/master/rpc/rpc_debug.go#L11
+
+https://github.com/zrepl/zrepl/blob/master/rpc/dataconn/dataconn_debug.go#L11
+
+https://github.com/zrepl/zrepl/blob/master/rpc/dataconn/stream/stream_debug.go#L11
+
+https://github.com/zrepl/zrepl/blob/master/rpc/dataconn/heartbeatconn/heartbeatconn_debug.go#L11
+

client/testcmd.go

@@ -60,6 +60,8 @@ func runTestFilterCmd(subcommand *cli.Subcommand, args []string) error {
 		confFilter = j.Filesystems
 	case *config.PushJob:
 		confFilter = j.Filesystems
+	case *config.SnapJob:
+		confFilter = j.Filesystems
 	default:
 		return fmt.Errorf("job type %T does not have filesystems filter", j)
 	}

docs/installation.rst

@@ -101,9 +101,9 @@ and serves as a reference for build dependencies and procedure:
 
 ::
 
-    git clone https://github.com/zrepl/zrepl.git
+    git clone https://github.com/zrepl/zrepl.git && \
-    cd zrepl
+    cd zrepl && \
-    sudo docker build -t zrepl_build -f build.Dockerfile .
+    sudo docker build -t zrepl_build -f build.Dockerfile . && \
     sudo docker run -it --rm \
         -v "${PWD}:/src" \
         --user "$(id -u):$(id -g)" \

endpoint/endpoint_zfs.go

@@ -411,6 +411,7 @@ func ListZFSHoldsAndBookmarks(ctx context.Context, fsfilter zfs.DatasetFilter) (
 	for _, fs := range fss {
 		err := listZFSHoldsAndBookmarksImplFS(ctx, out, fs)
 		if err != nil {
+			// FIXME if _, ok := err.(*zfs.DatasetDoesNotExist); ok { noop }
 			return nil, errors.Wrapf(err, "list holds and bookmarks on %q", fs.ToString())
 		}
 	}
@@ -420,7 +421,8 @@ func ListZFSHoldsAndBookmarks(ctx context.Context, fsfilter zfs.DatasetFilter) (
 func listZFSHoldsAndBookmarksImplFS(ctx context.Context, out *ListHoldsAndBookmarksOutput, fs *zfs.DatasetPath) error {
 	fsvs, err := zfs.ZFSListFilesystemVersions(fs, nil)
 	if err != nil {
-		return errors.Wrapf(err, "list filesystem versions of %q", fs)
+		// FIXME if _, ok := err.(*zfs.DatasetDoesNotExist); ok { noop }
+		return errors.Wrapf(err, "list filesystem versions of %q", fs.ToString())
 	}
 	for _, v := range fsvs {
 		switch v.Type {
@@ -429,8 +431,13 @@ func listZFSHoldsAndBookmarksImplFS(ctx context.Context, out *ListHoldsAndBookma
 		case zfs.Snapshot:
 			holds, err := zfs.ZFSHolds(ctx, fs.ToString(), v.Name)
 			if err != nil {
+				if _, ok := err.(*zfs.DatasetDoesNotExist); ok {
+					holds  = []string{}
+					// fallthrough
+				} else {
 					return errors.Wrapf(err, "get holds of %q", v.ToAbsPath(fs))
 				}
+			}
 			for _, tag := range holds {
 				listZFSHoldsAndBookmarksImplSnapshotTryParseHold(ctx, out, fs, v, tag)
 			}

platformtest/tests/idempotentHold.go

@@ -3,6 +3,7 @@ package tests
 import (
 	"fmt"
 
+	"github.com/stretchr/testify/require"
 	"github.com/zrepl/zrepl/platformtest"
 	"github.com/zrepl/zrepl/zfs"
 )
@@ -13,35 +14,28 @@ func IdempotentHold(ctx *platformtest.Context) {
 		DESTROYROOT
 		CREATEROOT
 		+  "foo bar"
-		+  "foo bar@1"
+		+  "foo bar@1 2"
-	`)
-	defer platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
-		R zfs release zrepl_platformtest "${ROOTDS}/foo bar@1"
-		-  "foo bar@1"
-		-  "foo bar"
 	`)
 
 	fs := fmt.Sprintf("%s/foo bar", ctx.RootDataset)
-	v1 := sendArgVersion(fs, "@1")
+	snap := sendArgVersion(fs, "@1 2")
 
 	tag := "zrepl_platformtest"
-	err := zfs.ZFSHold(ctx, fs, v1, tag)
+	err := zfs.ZFSHold(ctx, fs, snap, tag)
 	if err != nil {
 		panic(err)
 	}
 
-	err = zfs.ZFSHold(ctx, fs, v1, tag)
+	// existing holds
-	if err != nil {
+	holds, err := zfs.ZFSHolds(ctx, fs, "1 2")
-		panic(err)
+	require.NoError(ctx, err)
-	}
+	require.Equal(ctx, []string{tag}, holds)
-
-	vnonexistent := zfs.ZFSSendArgVersion{
-		RelName: "@nonexistent",
-		GUID:    0xbadf00d,
-	}
-	err = zfs.ZFSHold(ctx, fs, vnonexistent, tag)
-	if err == nil {
-		panic("still expecting error for nonexistent snapshot")
-	}
 
+	holds, err = zfs.ZFSHolds(ctx, fs, "non existent")
+	ctx.Logf("holds=%v", holds)
+	ctx.Logf("errT=%T", err)
+	ctx.Logf("err=%s", err)
+	notExist, ok := err.(*zfs.DatasetDoesNotExist)
+	require.True(ctx, ok)
+	require.Equal(ctx, fs+"@non existent", notExist.Path)
 }

platformtest/tests/listHolds.go

@@ -0,0 +1,47 @@
+package tests
+
+import (
+	"fmt"
+
+	"github.com/zrepl/zrepl/platformtest"
+	"github.com/zrepl/zrepl/zfs"
+)
+
+func ListHolds(ctx *platformtest.Context) {
+
+	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+		DESTROYROOT
+		CREATEROOT
+		+  "foo bar"
+		+  "foo bar@1"
+	`)
+	defer platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+		R zfs release zrepl_platformtest "${ROOTDS}/foo bar@1"
+		-  "foo bar@1"
+		-  "foo bar"
+	`)
+
+	fs := fmt.Sprintf("%s/foo bar", ctx.RootDataset)
+	v1 := sendArgVersion(fs, "@1")
+
+	tag := "zrepl_platformtest"
+	err := zfs.ZFSHold(ctx, fs, v1, tag)
+	if err != nil {
+		panic(err)
+	}
+
+	err = zfs.ZFSHold(ctx, fs, v1, tag)
+	if err != nil {
+		panic(err)
+	}
+
+	vnonexistent := zfs.ZFSSendArgVersion{
+		RelName: "@nonexistent",
+		GUID:    0xbadf00d,
+	}
+	err = zfs.ZFSHold(ctx, fs, vnonexistent, tag)
+	if err == nil {
+		panic("still expecting error for nonexistent snapshot")
+	}
+
+}

platformtest/tests/tests.go

@@ -31,4 +31,5 @@ var Cases = []Case{
 	SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden,
 	SendArgsValidationResumeTokenEncryptionMismatchForbidden,
 	SendArgsValidationResumeTokenDifferentFilesystemForbidden,
+	ListHolds,
 }

replication/logic/replication_logic.go

@@ -262,7 +262,6 @@ func (p *Planner) doPlanning(ctx context.Context) ([]*Filesystem, error) {
 		return nil, err
 	}
 	sfss := slfssres.GetFilesystems()
-	// no progress here since we could run in a live-lock on connectivity issues
 
 	rlfssres, err := p.receiver.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
 	if err != nil {

rpc/dataconn/timeoutconn/timeoutconn.go

@@ -49,7 +49,7 @@ type Wire interface {
 	// No data that could otherwise be Read is lost as a consequence of this call.
 	// The use case for this API is abortive connection shutdown.
 	// To provide any value over draining Wire using io.Read, an implementation
-	// will likely use out-of-bounds messaging mechanisms.
+	// will likely use out-of-band messaging mechanisms.
 	// TODO WaitForPeerClose() (supported bool, err error)
 }
 

transport/tls/serve_tls.go

@@ -32,7 +32,7 @@ func TLSListenerFactoryFromConfig(c *config.Global, in *config.TLSServe) (transp
 
 	serverCert, err := tls.LoadX509KeyPair(in.Cert, in.Key)
 	if err != nil {
-		return nil, errors.Wrap(err, "cannot parse cer/key pair")
+		return nil, errors.Wrap(err, "cannot parse cert/key pair")
 	}
 
 	clientCNs := make(map[string]struct{}, len(in.ClientCNs))

util/envconst/envconst.go

@@ -9,16 +9,6 @@ import (
 
 var cache sync.Map
 
-// capture the actual envconst values used at runtime
-func DebugDump() map[string]interface{} {
-	m := make(map[string]interface{})
-	cache.Range(func(k, v interface{}) bool {
-		m[k.(string)] = v
-		return true
-	})
-	return m
-}
-
 func Duration(varname string, def time.Duration) time.Duration {
 	if v, ok := cache.Load(varname); ok {
 		return v.(time.Duration)

zfs/holds.go

@@ -59,6 +59,7 @@ success:
 	return nil
 }
 
+// If the snapshot does not exist, the returned error is of type *DatasetDoesNotExist
 func ZFSHolds(ctx context.Context, fs, snap string) ([]string, error) {
 	if err := validateZFSFilesystem(fs); err != nil {
 		return nil, errors.Wrap(err, "`fs` is not a valid filesystem path")