replication/driver: fs.debug() helper that automatically prefixes with fs name

refs #140
refs #302

Makefile

@@ -23,7 +23,8 @@ GOHOSTARCH ?= $(shell bash -c 'source <($(GO) env) && echo "$$GOHOSTARCH"')
 GO_ENV_VARS := GO111MODULE=on
 GO_LDFLAGS := "-X github.com/zrepl/zrepl/version.zreplVersion=$(_ZREPL_VERSION)"
 GO_MOD_READONLY := -mod=readonly
-GO_BUILDFLAGS := $(GO_MOD_READONLY)
+GO_EXTRA_BUILDFLAGS :=
+GO_BUILDFLAGS := $(GO_MOD_READONLY) $(GO_EXTRA_BUILDFLAGS)
 GO_BUILD := $(GO_ENV_VARS) $(GO) build $(GO_BUILDFLAGS) -ldflags $(GO_LDFLAGS)
 GOLANGCI_LINT := golangci-lint
 ifneq ($(GOARM),)

endpoint/endpoint.go

@@ -150,11 +150,11 @@ func (p *Sender) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.HintMo
 	}
 
 	// take care of stale step holds
-	log.WithField("step-holds-cleanup-mode", hintMostRecentCommonAncestorStepCleanupMode).
+	log.WithField("step-holds-cleanup-mode", senderHintMostRecentCommonAncestorStepCleanupMode).
 		Debug("taking care of possibly stale step holds")
 	doStepCleanup := false
 	var stepCleanupSince *CreateTXGRangeBound
-	switch hintMostRecentCommonAncestorStepCleanupMode {
+	switch senderHintMostRecentCommonAncestorStepCleanupMode {
 	case StepCleanupNoCleanup:
 		doStepCleanup = false
 	case StepCleanupRangeSinceUnbounded:
@@ -180,7 +180,7 @@ func (p *Sender) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.HintMo
 			}
 		}
 	default:
-		panic(hintMostRecentCommonAncestorStepCleanupMode)
+		panic(senderHintMostRecentCommonAncestorStepCleanupMode)
 	}
 	if !doStepCleanup {
 		log.Info("skipping cleanup of prior invocations' step holds due to environment variable setting")
@@ -218,7 +218,7 @@ func (m *HintMostRecentCommonAncestorStepCleanupMode) Set(s string) error {
 	return nil
 }
 
-var hintMostRecentCommonAncestorStepCleanupMode = *envconst.Var("ZREPL_ENDPOINT_HINT_MOST_RECENT_STEP_HOLD_CLEANUP_MODE", &StepCleanupRangeSinceReplicationCursor).(*HintMostRecentCommonAncestorStepCleanupMode)
+var senderHintMostRecentCommonAncestorStepCleanupMode = *envconst.Var("ZREPL_ENDPOINT_SENDER_HINT_MOST_RECENT_STEP_HOLD_CLEANUP_MODE", &StepCleanupRangeSinceReplicationCursor).(*HintMostRecentCommonAncestorStepCleanupMode)
 
 var maxConcurrentZFSSendSemaphore = semaphore.New(envconst.Int64("ZREPL_ENDPOINT_MAX_CONCURRENT_SEND", 10))
 
@@ -806,7 +806,7 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 	if err := zfs.ZFSRecv(ctx, lp.ToString(), to, receive, recvOpts); err != nil {
 		getLogger(ctx).
 			WithError(err).
-			WithField("opts", recvOpts).
+			WithField("opts", fmt.Sprintf("%#v", recvOpts)).
 			Error("zfs receive failed")
 		return nil, err
 	}

replication/design.md

@@ -18,7 +18,32 @@ Hence, when trying to map algorithm to implementation, use the code in package `
       * the recv-side fs doesn't get newer snapshots than `to` in the meantime
         * guaranteed because the zrepl model of the receiver assumes ownership of the filesystems it receives into
           * if that assumption is broken, future replication attempts will fail with a conflict
-* The [Algorithm for Planning and Executing Replication of an Filesystems](#zrepl-algo-filesystem) is a design draft and not used
+* The [Algorithm for Planning and Executing Replication of an Filesystems](#zrepl-algo-filesystem) is a design draft and not used.
+  However, there were some noteworthy lessons learned when implementing the algorithm for a single step:
+  * In order to avoid leaking `step-hold`s and `step-bookmarks`, if the replication planner is invoked a second time after a replication step (either initial or incremental) has been attempted but failed to completed, the replication planner must
+    * A) either guarantee that it will resume that replication step, and continue as if nothing happened or
+    * B) release the step holds and bookmarks and clear the partially received state on the sending side.
+  * Option A is what we want to do: we use the step algorithm to achieve resumability in the first place!
+  * Option B is not done by zrepl except if the sending side doesn't support resuming.
+    In that case however, we need not release any holds since the behavior is to re-start the send
+    from the beginning.
+  * However, there is one **edge-case to Option A for initial replication**:
+    * If initial replication "`full a`" fails without leaving resumable state, the step holds on the sending side are still present, which makes sense because
+      * a) we want resumability and
+      * b) **the sending side cannot immediately be informed post-failure whether the initial replication left any state that would mandate keeping the step hold**, because the network connection might have failed.
+    * Thus, the sending side must keep the step hold for "`full a`" until it knows more.
+    * In the current implementation, it knows more when the next replication attempt is made, the planner is invoked, the diffing algorithm run, and the `HintMostRecentCommonAncestor` RPC is sent by the active side, communicating the most recent common version shared betwen sender and receiver.
+    * At this point, the sender can safely throw away any step holds with CreateTXG's older than that version.
+  * **The `step-hold`, `step-bookmark`, `last-received-hold` and `replication-cursor` abstractions are currently local concepts of package `endpoint` and not part of the replication protocol**
+    * This is not necessarilty the best design decision and should be revisited some point:
+      * The (quite expensive) `HintMostRecentCommonAncestor` RPC impl on the sender would not be necessary if step holds were part of the replication protocol:
+        * We only need the `HintMostRecentCommonAncestor` info for the aforementioned edge-case during initial replication, where the receive is aborted without any partial received state being stored on the receiver (due to network failure, wrong zfs invocation, bad permissions, etc):
+        * **The replication planner does not know about the step holds, thus it cannot deterministically pick up where it left of (right at the start of the last failing initial replication).**
+        * Instead, it will seem like no prior invocation happened at all, and it will apply its policy for initial replication to pick a new `full b != full a`, **thereby leaking the step holds of `full a`**.
+        * In contrast, if the replication planner created the step holds and knew about them, it could use the step holds as an indicator where it left off and re-start from there (of course asserting that the thereby inferred step is compatible with the state of the receiving side).
+      * (What we do in zrepl right now is to hard-code the initial replication policy, and hard-code that assumption in `endpoint.ListStale` as well.)
+      * The cummulative cleanup done in `HintMostRecentCommonAncestor` provides a nice self-healing aspect, though.
+
 * We also have [Notes on Planning and Executing Replication of Multiple Filesystems](#zrepl-algo-multiple-filesystems-notes)
 
 ---

replication/driver/replication_driver.go

@@ -355,7 +355,7 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 	// invariant: prevs contains an entry for each unambiguous correspondence
 
 	stepQueue := newStepQueue()
-	defer stepQueue.Start(1)() // TODO parallel replication
+	defer stepQueue.Start(envconst.Int("ZREPL_REPLICATION_EXPERIMENTAL_REPLICATION_CONCURRENCY", 1))() // TODO parallel replication
 	var fssesDone sync.WaitGroup
 	for _, f := range a.fss {
 		fssesDone.Add(1)
@@ -370,54 +370,57 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 	a.finishedAt = time.Now()
 }
 
-func (fs *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
+func (f *fs) debug(format string, args ...interface{}) {
+	debugPrefix("fs=%s", f.fs.ReportInfo().Name)(format, args...)
+}
 
-	defer fs.l.Lock().Unlock()
+func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
+
+	defer f.l.Lock().Unlock()
 
 	// get planned steps from replication logic
 	var psteps []Step
 	var errTime time.Time
 	var err error
-	fs.l.DropWhile(func() {
+	f.l.DropWhile(func() {
 		// TODO hacky
 		// choose target time that is earlier than any snapshot, so fs planning is always prioritized
 		targetDate := time.Unix(0, 0)
-		defer pq.WaitReady(fs, targetDate)()
-		psteps, err = fs.fs.PlanFS(ctx) // no shadow
-		errTime = time.Now()            // no shadow
+		defer pq.WaitReady(f, targetDate)()
+		psteps, err = f.fs.PlanFS(ctx) // no shadow
+		errTime = time.Now()           // no shadow
 	})
-	debug := debugPrefix("fs=%s", fs.fs.ReportInfo().Name)
-	fs.planning.done = true
+	f.planning.done = true
 	if err != nil {
-		fs.planning.err = newTimedError(err, errTime)
+		f.planning.err = newTimedError(err, errTime)
 		return
 	}
 	for _, pstep := range psteps {
 		step := &step{
-			l:    fs.l,
+			l:    f.l,
 			step: pstep,
 		}
-		fs.planned.steps = append(fs.planned.steps, step)
+		f.planned.steps = append(f.planned.steps, step)
 	}
-	debug("initial len(fs.planned.steps) = %d", len(fs.planned.steps))
+	f.debug("initial len(fs.planned.steps) = %d", len(f.planned.steps))
 
 	// for not-first attempts, only allow fs.planned.steps
 	// up to including the originally planned target snapshot
 	if prev != nil && prev.planning.done && prev.planning.err == nil {
 		prevUncompleted := prev.planned.steps[prev.planned.step:]
 		if len(prevUncompleted) == 0 {
-			debug("prevUncompleted is empty")
+			f.debug("prevUncompleted is empty")
 			return
 		}
-		if len(fs.planned.steps) == 0 {
-			debug("fs.planned.steps is empty")
+		if len(f.planned.steps) == 0 {
+			f.debug("fs.planned.steps is empty")
 			return
 		}
 		prevFailed := prevUncompleted[0]
-		curFirst := fs.planned.steps[0]
+		curFirst := f.planned.steps[0]
 		// we assume that PlanFS retries prevFailed (using curFirst)
 		if !prevFailed.step.TargetEquals(curFirst.step) {
-			debug("Targets don't match")
+			f.debug("Targets don't match")
 			// Two options:
 			// A: planning algorithm is broken
 			// B: manual user intervention inbetween
@@ -433,43 +436,43 @@ func (fs *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 			}
 			msg := fmt.Sprintf("last attempt's uncompleted step %s does not correspond to this attempt's first planned step %s",
 				stepFmt(prevFailed), stepFmt(curFirst))
-			fs.planned.stepErr = newTimedError(errors.New(msg), time.Now())
+			f.planned.stepErr = newTimedError(errors.New(msg), time.Now())
 			return
 		}
 		// only allow until step targets diverge
 		min := len(prevUncompleted)
-		if min > len(fs.planned.steps) {
-			min = len(fs.planned.steps)
+		if min > len(f.planned.steps) {
+			min = len(f.planned.steps)
 		}
 		diverge := 0
 		for ; diverge < min; diverge++ {
-			debug("diverge compare iteration %d", diverge)
-			if !fs.planned.steps[diverge].step.TargetEquals(prevUncompleted[diverge].step) {
+			f.debug("diverge compare iteration %d", diverge)
+			if !f.planned.steps[diverge].step.TargetEquals(prevUncompleted[diverge].step) {
 				break
 			}
 		}
-		debug("diverge is %d", diverge)
-		fs.planned.steps = fs.planned.steps[0:diverge]
+		f.debug("diverge is %d", diverge)
+		f.planned.steps = f.planned.steps[0:diverge]
 	}
-	debug("post-prev-merge len(fs.planned.steps) = %d", len(fs.planned.steps))
+	f.debug("post-prev-merge len(fs.planned.steps) = %d", len(f.planned.steps))
 
-	for i, s := range fs.planned.steps {
+	for i, s := range f.planned.steps {
 		var (
 			err     error
 			errTime time.Time
 		)
 		// lock must not be held while executing step in order for reporting to work
-		fs.l.DropWhile(func() {
+		f.l.DropWhile(func() {
 			targetDate := s.step.TargetDate()
-			defer pq.WaitReady(fs, targetDate)()
+			defer pq.WaitReady(f, targetDate)()
 			err = s.step.Step(ctx) // no shadow
 			errTime = time.Now()   // no shadow
 		})
 		if err != nil {
-			fs.planned.stepErr = newTimedError(err, errTime)
+			f.planned.stepErr = newTimedError(err, errTime)
 			break
 		}
-		fs.planned.step = i + 1 // fs.planned.step must be == len(fs.planned.steps) if all went OK
+		f.planned.step = i + 1 // fs.planned.step must be == len(fs.planned.steps) if all went OK
 	}
 }
 

replication/logic/pdu/pdu.pb.go

@@ -46,7 +46,7 @@ func (x Tri) String() string {
 	return proto.EnumName(Tri_name, int32(x))
 }
 func (Tri) EnumDescriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{0}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{0}
 }
 
 type FilesystemVersion_VersionType int32
@@ -69,7 +69,7 @@ func (x FilesystemVersion_VersionType) String() string {
 	return proto.EnumName(FilesystemVersion_VersionType_name, int32(x))
 }
 func (FilesystemVersion_VersionType) EnumDescriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{5, 0}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{5, 0}
 }
 
 type ListFilesystemReq struct {
@@ -82,7 +82,7 @@ func (m *ListFilesystemReq) Reset()         { *m = ListFilesystemReq{} }
 func (m *ListFilesystemReq) String() string { return proto.CompactTextString(m) }
 func (*ListFilesystemReq) ProtoMessage()    {}
 func (*ListFilesystemReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{0}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{0}
 }
 func (m *ListFilesystemReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ListFilesystemReq.Unmarshal(m, b)
@@ -113,7 +113,7 @@ func (m *ListFilesystemRes) Reset()         { *m = ListFilesystemRes{} }
 func (m *ListFilesystemRes) String() string { return proto.CompactTextString(m) }
 func (*ListFilesystemRes) ProtoMessage()    {}
 func (*ListFilesystemRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{1}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{1}
 }
 func (m *ListFilesystemRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ListFilesystemRes.Unmarshal(m, b)
@@ -154,7 +154,7 @@ func (m *Filesystem) Reset()         { *m = Filesystem{} }
 func (m *Filesystem) String() string { return proto.CompactTextString(m) }
 func (*Filesystem) ProtoMessage()    {}
 func (*Filesystem) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{2}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{2}
 }
 func (m *Filesystem) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_Filesystem.Unmarshal(m, b)
@@ -213,7 +213,7 @@ func (m *ListFilesystemVersionsReq) Reset()         { *m = ListFilesystemVersion
 func (m *ListFilesystemVersionsReq) String() string { return proto.CompactTextString(m) }
 func (*ListFilesystemVersionsReq) ProtoMessage()    {}
 func (*ListFilesystemVersionsReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{3}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{3}
 }
 func (m *ListFilesystemVersionsReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ListFilesystemVersionsReq.Unmarshal(m, b)
@@ -251,7 +251,7 @@ func (m *ListFilesystemVersionsRes) Reset()         { *m = ListFilesystemVersion
 func (m *ListFilesystemVersionsRes) String() string { return proto.CompactTextString(m) }
 func (*ListFilesystemVersionsRes) ProtoMessage()    {}
 func (*ListFilesystemVersionsRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{4}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{4}
 }
 func (m *ListFilesystemVersionsRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ListFilesystemVersionsRes.Unmarshal(m, b)
@@ -293,7 +293,7 @@ func (m *FilesystemVersion) Reset()         { *m = FilesystemVersion{} }
 func (m *FilesystemVersion) String() string { return proto.CompactTextString(m) }
 func (*FilesystemVersion) ProtoMessage()    {}
 func (*FilesystemVersion) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{5}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{5}
 }
 func (m *FilesystemVersion) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_FilesystemVersion.Unmarshal(m, b)
@@ -371,7 +371,7 @@ func (m *SendReq) Reset()         { *m = SendReq{} }
 func (m *SendReq) String() string { return proto.CompactTextString(m) }
 func (*SendReq) ProtoMessage()    {}
 func (*SendReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{6}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{6}
 }
 func (m *SendReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_SendReq.Unmarshal(m, b)
@@ -445,7 +445,7 @@ func (m *Property) Reset()         { *m = Property{} }
 func (m *Property) String() string { return proto.CompactTextString(m) }
 func (*Property) ProtoMessage()    {}
 func (*Property) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{7}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{7}
 }
 func (m *Property) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_Property.Unmarshal(m, b)
@@ -496,7 +496,7 @@ func (m *SendRes) Reset()         { *m = SendRes{} }
 func (m *SendRes) String() string { return proto.CompactTextString(m) }
 func (*SendRes) ProtoMessage()    {}
 func (*SendRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{8}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{8}
 }
 func (m *SendRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_SendRes.Unmarshal(m, b)
@@ -548,7 +548,7 @@ func (m *SendCompletedReq) Reset()         { *m = SendCompletedReq{} }
 func (m *SendCompletedReq) String() string { return proto.CompactTextString(m) }
 func (*SendCompletedReq) ProtoMessage()    {}
 func (*SendCompletedReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{9}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{9}
 }
 func (m *SendCompletedReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_SendCompletedReq.Unmarshal(m, b)
@@ -585,7 +585,7 @@ func (m *SendCompletedRes) Reset()         { *m = SendCompletedRes{} }
 func (m *SendCompletedRes) String() string { return proto.CompactTextString(m) }
 func (*SendCompletedRes) ProtoMessage()    {}
 func (*SendCompletedRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{10}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{10}
 }
 func (m *SendCompletedRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_SendCompletedRes.Unmarshal(m, b)
@@ -608,7 +608,7 @@ var xxx_messageInfo_SendCompletedRes proto.InternalMessageInfo
 type ReceiveReq struct {
 	Filesystem string             `protobuf:"bytes,1,opt,name=Filesystem,proto3" json:"Filesystem,omitempty"`
 	To         *FilesystemVersion `protobuf:"bytes,2,opt,name=To,proto3" json:"To,omitempty"`
-	// If true, the receiver should clear the resume token before perfoming the
+	// If true, the receiver should clear the resume token before performing the
 	// zfs recv of the stream in the request
 	ClearResumeToken     bool     `protobuf:"varint,3,opt,name=ClearResumeToken,proto3" json:"ClearResumeToken,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
@@ -620,7 +620,7 @@ func (m *ReceiveReq) Reset()         { *m = ReceiveReq{} }
 func (m *ReceiveReq) String() string { return proto.CompactTextString(m) }
 func (*ReceiveReq) ProtoMessage()    {}
 func (*ReceiveReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{11}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{11}
 }
 func (m *ReceiveReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ReceiveReq.Unmarshal(m, b)
@@ -671,7 +671,7 @@ func (m *ReceiveRes) Reset()         { *m = ReceiveRes{} }
 func (m *ReceiveRes) String() string { return proto.CompactTextString(m) }
 func (*ReceiveRes) ProtoMessage()    {}
 func (*ReceiveRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{12}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{12}
 }
 func (m *ReceiveRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ReceiveRes.Unmarshal(m, b)
@@ -704,7 +704,7 @@ func (m *DestroySnapshotsReq) Reset()         { *m = DestroySnapshotsReq{} }
 func (m *DestroySnapshotsReq) String() string { return proto.CompactTextString(m) }
 func (*DestroySnapshotsReq) ProtoMessage()    {}
 func (*DestroySnapshotsReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{13}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{13}
 }
 func (m *DestroySnapshotsReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_DestroySnapshotsReq.Unmarshal(m, b)
@@ -750,7 +750,7 @@ func (m *DestroySnapshotRes) Reset()         { *m = DestroySnapshotRes{} }
 func (m *DestroySnapshotRes) String() string { return proto.CompactTextString(m) }
 func (*DestroySnapshotRes) ProtoMessage()    {}
 func (*DestroySnapshotRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{14}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{14}
 }
 func (m *DestroySnapshotRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_DestroySnapshotRes.Unmarshal(m, b)
@@ -795,7 +795,7 @@ func (m *DestroySnapshotsRes) Reset()         { *m = DestroySnapshotsRes{} }
 func (m *DestroySnapshotsRes) String() string { return proto.CompactTextString(m) }
 func (*DestroySnapshotsRes) ProtoMessage()    {}
 func (*DestroySnapshotsRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{15}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{15}
 }
 func (m *DestroySnapshotsRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_DestroySnapshotsRes.Unmarshal(m, b)
@@ -833,7 +833,7 @@ func (m *ReplicationCursorReq) Reset()         { *m = ReplicationCursorReq{} }
 func (m *ReplicationCursorReq) String() string { return proto.CompactTextString(m) }
 func (*ReplicationCursorReq) ProtoMessage()    {}
 func (*ReplicationCursorReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{16}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{16}
 }
 func (m *ReplicationCursorReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ReplicationCursorReq.Unmarshal(m, b)
@@ -874,7 +874,7 @@ func (m *ReplicationCursorRes) Reset()         { *m = ReplicationCursorRes{} }
 func (m *ReplicationCursorRes) String() string { return proto.CompactTextString(m) }
 func (*ReplicationCursorRes) ProtoMessage()    {}
 func (*ReplicationCursorRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{17}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{17}
 }
 func (m *ReplicationCursorRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ReplicationCursorRes.Unmarshal(m, b)
@@ -1010,7 +1010,7 @@ func (m *PingReq) Reset()         { *m = PingReq{} }
 func (m *PingReq) String() string { return proto.CompactTextString(m) }
 func (*PingReq) ProtoMessage()    {}
 func (*PingReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{18}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{18}
 }
 func (m *PingReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_PingReq.Unmarshal(m, b)
@@ -1049,7 +1049,7 @@ func (m *PingRes) Reset()         { *m = PingRes{} }
 func (m *PingRes) String() string { return proto.CompactTextString(m) }
 func (*PingRes) ProtoMessage()    {}
 func (*PingRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{19}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{19}
 }
 func (m *PingRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_PingRes.Unmarshal(m, b)
@@ -1077,7 +1077,16 @@ func (m *PingRes) GetEcho() string {
 }
 
 type HintMostRecentCommonAncestorReq struct {
-	Filesystem           string             `protobuf:"bytes,1,opt,name=Filesystem,proto3" json:"Filesystem,omitempty"`
+	Filesystem string `protobuf:"bytes,1,opt,name=Filesystem,proto3" json:"Filesystem,omitempty"`
+	// A copy of the FilesystemVersion on the sending side that the replication
+	// algorithm identified as a shared most recent common version between sending
+	// and receiving side.
+	//
+	// If nil, this is an indication that the replication algorithm could not
+	// find a common ancestor between the two sides.
+	// NOTE: nilness does not mean that replication never happened - there could
+	// as well be a replication conflict. thus, dont' jump to conclusions too
+	// rapidly here.
 	SenderVersion        *FilesystemVersion `protobuf:"bytes,2,opt,name=SenderVersion,proto3" json:"SenderVersion,omitempty"`
 	XXX_NoUnkeyedLiteral struct{}           `json:"-"`
 	XXX_unrecognized     []byte             `json:"-"`
@@ -1088,7 +1097,7 @@ func (m *HintMostRecentCommonAncestorReq) Reset()         { *m = HintMostRecentC
 func (m *HintMostRecentCommonAncestorReq) String() string { return proto.CompactTextString(m) }
 func (*HintMostRecentCommonAncestorReq) ProtoMessage()    {}
 func (*HintMostRecentCommonAncestorReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{20}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{20}
 }
 func (m *HintMostRecentCommonAncestorReq) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_HintMostRecentCommonAncestorReq.Unmarshal(m, b)
@@ -1132,7 +1141,7 @@ func (m *HintMostRecentCommonAncestorRes) Reset()         { *m = HintMostRecentC
 func (m *HintMostRecentCommonAncestorRes) String() string { return proto.CompactTextString(m) }
 func (*HintMostRecentCommonAncestorRes) ProtoMessage()    {}
 func (*HintMostRecentCommonAncestorRes) Descriptor() ([]byte, []int) {
-	return fileDescriptor_pdu_0f43b713cd3bf056, []int{21}
+	return fileDescriptor_pdu_e59763dc61674a79, []int{21}
 }
 func (m *HintMostRecentCommonAncestorRes) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_HintMostRecentCommonAncestorRes.Unmarshal(m, b)
@@ -1449,9 +1458,9 @@ var _Replication_serviceDesc = grpc.ServiceDesc{
 	Metadata: "pdu.proto",
 }
 
-func init() { proto.RegisterFile("pdu.proto", fileDescriptor_pdu_0f43b713cd3bf056) }
+func init() { proto.RegisterFile("pdu.proto", fileDescriptor_pdu_e59763dc61674a79) }
 
-var fileDescriptor_pdu_0f43b713cd3bf056 = []byte{
+var fileDescriptor_pdu_e59763dc61674a79 = []byte{
 	// 892 bytes of a gzipped FileDescriptorProto
 	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x56, 0xdf, 0x6f, 0xdb, 0x36,
 	0x10, 0x8e, 0x6c, 0x39, 0x91, 0xcf, 0xe9, 0xea, 0x5c, 0xb2, 0x42, 0x13, 0xba, 0xce, 0xe3, 0x86,

replication/logic/pdu/pdu.proto

@@ -128,7 +128,17 @@ message PingRes {
 }
 
 message HintMostRecentCommonAncestorReq {
-    string Filesystem = 1;
-    FilesystemVersion SenderVersion = 2;
+  string Filesystem = 1;
+
+  // A copy of the FilesystemVersion on the sending side that the replication
+  // algorithm identified as a shared most recent common version between sending
+  // and receiving side.
+  //
+  // If nil, this is an indication that the replication algorithm could not
+  // find a common ancestor between the two sides.
+  // NOTE: nilness does not mean that replication never happened - there could
+  // as well be a replication conflict. thus, dont' jump to conclusions too
+  // rapidly here.
+  FilesystemVersion SenderVersion = 2;
 }
 message HintMostRecentCommonAncestorRes {}

rpc/versionhandshake/versionhandshake.go

@@ -152,7 +152,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, 2)
+	return DoHandshakeVersion(conn, deadline, 3)
 }
 
 const HandshakeMessageMaxLen = 16 * 4096