[#345] fix broken identification of parent-fs for initial replication ordering

fixup of 02807279

fixes #345

platformtest/tests/generated_cases.go

@@ -16,6 +16,7 @@ var Cases = []Case{BatchDestroy,
 	ListFilesystemsNoFilter,
 	ReceiveForceIntoEncryptedErr,
 	ReceiveForceRollbackWorksUnencrypted,
+	ReplicationFailingInitialParentProhibitsChildReplication,
 	ReplicationIncrementalCleansUpStaleAbstractionsWithCacheOnSecondReplication,
 	ReplicationIncrementalCleansUpStaleAbstractionsWithoutCacheOnSecondReplication,
 	ReplicationIncrementalDestroysStepHoldsIffIncrementalStepHoldsAreDisabledButStepHoldsExist,

platformtest/tests/replication.go

@@ -30,6 +30,7 @@ import (
 type replicationInvocation struct {
 	sjid, rjid        endpoint.JobID
 	sfs               string
+	sfilter           *filters.DatasetMapFilter
 	rfsRoot           string
 	interceptSender   func(e *endpoint.Sender) logic.Sender
 	interceptReceiver func(e *endpoint.Receiver) logic.Receiver
@@ -45,11 +46,16 @@ func (i replicationInvocation) Do(ctx *platformtest.Context) *report.Report {
 		i.interceptReceiver = func(e *endpoint.Receiver) logic.Receiver { return e }
 	}
 
-	sfilter := filters.NewDatasetMapFilter(1, true)
-	err := sfilter.Add(i.sfs, "ok")
+	if i.sfs != "" && i.sfilter != nil || i.sfs == "" && i.sfilter == nil {
+		panic("either sfs or sfilter must be set")
+	}
+	if i.sfilter == nil {
+		i.sfilter = filters.NewDatasetMapFilter(1, true)
+		err := i.sfilter.Add(i.sfs, "ok")
 		require.NoError(ctx, err)
+	}
 	sender := i.interceptSender(endpoint.NewSender(endpoint.SenderConfig{
-		FSF:     sfilter.AsFilter(),
+		FSF:     i.sfilter.AsFilter(),
 		Encrypt: &zfs.NilBool{B: false},
 		JobID:   i.sjid,
 	}))
@@ -835,3 +841,76 @@ func ReplicationReceiverErrorWhileStillSending(ctx *platformtest.Context) {
 	require.NotNil(ctx, afs.StepError)
 	require.Contains(ctx, afs.StepError.Err, mockRecvErr.Error())
 }
+
+func ReplicationFailingInitialParentProhibitsChildReplication(ctx *platformtest.Context) {
+
+	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+		CREATEROOT
+		+  "sender"
+		+  "sender/a"
+		+  "sender/a/child"
+		+  "sender/aa"
+		+  "receiver"
+		R  zfs create -p "${ROOTDS}/receiver/${ROOTDS}"
+		R  zfs snapshot -r ${ROOTDS}/sender@initial
+	`)
+
+	sjid := endpoint.MustMakeJobID("sender-job")
+	rjid := endpoint.MustMakeJobID("receiver-job")
+
+	fsA := ctx.RootDataset + "/sender/a"
+	fsAChild := ctx.RootDataset + "/sender/a/child"
+	fsAA := ctx.RootDataset + "/sender/aa"
+
+	sfilter := filters.NewDatasetMapFilter(3, true)
+	mustAddToSFilter := func(fs string) {
+		err := sfilter.Add(fs, "ok")
+		require.NoError(ctx, err)
+	}
+	mustAddToSFilter(fsA)
+	mustAddToSFilter(fsAChild)
+	mustAddToSFilter(fsAA)
+	rfsRoot := ctx.RootDataset + "/receiver"
+
+	mockRecvErr := fmt.Errorf("yifae4ohPhaquaes0hohghiep9oufie4roo7quoWooluaj2ee8")
+
+	rep := replicationInvocation{
+		sjid:      sjid,
+		rjid:      rjid,
+		sfilter:   sfilter,
+		rfsRoot:   rfsRoot,
+		guarantee: *pdu.ReplicationConfigProtectionWithKind(pdu.ReplicationGuaranteeKind_GuaranteeNothing),
+		interceptReceiver: func(r *endpoint.Receiver) logic.Receiver {
+			return &ErroringReceiver{recvErr: mockRecvErr, Receiver: r}
+		},
+	}
+
+	r := rep.Do(ctx)
+	ctx.Logf("\n%s", pretty.Sprint(r))
+
+	require.Len(ctx, r.Attempts, 1)
+	attempt := r.Attempts[0]
+	require.Nil(ctx, attempt.PlanError)
+	require.Len(ctx, attempt.Filesystems, 3)
+
+	fsByName := make(map[string]*report.FilesystemReport, len(attempt.Filesystems))
+	for _, fs := range attempt.Filesystems {
+		fsByName[fs.Info.Name] = fs
+	}
+
+	require.Contains(ctx, fsByName, fsA)
+	require.Contains(ctx, fsByName, fsAA)
+	require.Contains(ctx, fsByName, fsAA)
+
+	checkFS := func(fs string, expectErrMsg string) {
+		rep := fsByName[fs]
+		require.Len(ctx, rep.Steps, 1)
+		require.Nil(ctx, rep.PlanError)
+		require.NotNil(ctx, rep.StepError)
+		require.Contains(ctx, rep.StepError.Err, expectErrMsg)
+	}
+
+	checkFS(fsA, mockRecvErr.Error())
+	checkFS(fsAChild, "parent(s) failed during initial replication")
+	checkFS(fsAA, mockRecvErr.Error()) // fsAA is not treated as a child of fsA
+}

replication/driver/replication_driver.go

@@ -2,7 +2,6 @@ package driver
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"net"
 	"sort"
@@ -10,7 +9,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/pkg/errors"
 	"github.com/zrepl/zrepl/daemon/logging/trace"
+	"github.com/zrepl/zrepl/zfs"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
@@ -374,11 +375,28 @@ func (a *attempt) doGlobalPlanning(ctx context.Context, prev *attempt) map[*fs]*
 	// invariant: prevs contains an entry for each unambiguous correspondence
 
 	// build up parent-child relationship (FIXME (O(n^2), but who's going to have that many filesystems...))
+	mustDatasetPathOrPlanFail := func(fs string) *zfs.DatasetPath {
+		dp, err := zfs.NewDatasetPath(fs)
+		if err != nil {
+			now := time.Now()
+			a.planErr = newTimedError(errors.Wrapf(err, "%q", fs), now)
+			a.fss = nil
+			a.finishedAt = now
+			return nil
+		}
+		return dp
+	}
 	for _, f1 := range a.fss {
-		fs1 := f1.fs.ReportInfo().Name
+		fs1 := mustDatasetPathOrPlanFail(f1.fs.ReportInfo().Name)
+		if fs1 == nil {
+			return nil
+		}
 		for _, f2 := range a.fss {
-			fs2 := f2.fs.ReportInfo().Name
-			if strings.HasPrefix(fs1, fs2) && fs1 != fs2 {
+			fs2 := mustDatasetPathOrPlanFail(f2.fs.ReportInfo().Name)
+			if fs2 == nil {
+				return nil
+			}
+			if fs1.HasPrefix(fs2) && !fs1.Equal(fs2) {
 				f1.initialRepOrd.parents = append(f1.initialRepOrd.parents, f2)
 				f2.initialRepOrd.children = append(f2.initialRepOrd.children, f1)
 			}