[#316] endpoint / replication protocol: more robust step-holds and replication cursor management

- drop HintMostRecentCommonAncestor rpc call
    - it is wrong to put faith into the active side of the replication to always make that call
      (we might not trust it, ref pull setup)
- clean up step holds + step bookmarks + replication cursor bookmarks on
  send RPC instead
    - this makes it symmetric with Receive RPC
- use a cache (endpoint.sendAbstractionsCache) to avoid the cost of
  listing the on-disk endpoint abstractions state on every step

The "create" methods for endpoint abstractions (CreateReplicationCursor, HoldStep) are now fully
idempotent and return an Abstraction.

Notes about endpoint.sendAbstractionsCache:
- fills lazily from disk state on first `Get` operation
- fill from disk is generally only attempted once
    - unless the `ListAbstractions` fails, in which case the fill from
      disk is retried on next `Get` (the current `Get` will observe a
      subset of the actual on-disk abstractions)
    - the `Invalidate` method is called
- it is a global (zrepl process-wide) cache

fixes #316

endpoint/endpoint.go

@@ -7,8 +7,8 @@ import (
 	"fmt"
 	"io"
 	"path"
-	"sync"
 
+	"github.com/kr/pretty"
 	"github.com/pkg/errors"
 	"github.com/zrepl/zrepl/daemon/logging/trace"
 
@@ -117,119 +117,8 @@ func (s *Sender) ListFilesystemVersions(ctx context.Context, r *pdu.ListFilesyst
 
 }
 
-func (p *Sender) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.HintMostRecentCommonAncestorReq) (*pdu.HintMostRecentCommonAncestorRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-
-	fsp, err := p.filterCheckFS(r.GetFilesystem())
-	if err != nil {
-		return nil, err
-	}
-	fs := fsp.ToString()
-
-	log := getLogger(ctx).WithField("fs", fs).WithField("hinted_most_recent", fmt.Sprintf("%#v", r.GetSenderVersion()))
-
-	log.WithField("full_hint", r).Debug("full hint")
-
-	if r.GetSenderVersion() == nil {
-		// no common ancestor found, likely due to failed prior replication attempt
-		// => release stale step holds to prevent them from accumulating
-		//    (they can accumulate on initial replication because each inital replication step might hold a different `to`)
-		// => replication cursors cannot accumulate because we always _move_ the replication cursor
-		log.Debug("releasing all step holds on the filesystem")
-		TryReleaseStepStaleFS(ctx, fs, p.jobId)
-		return &pdu.HintMostRecentCommonAncestorRes{}, nil
-	}
-
-	// we were hinted a specific common ancestor
-
-	mostRecentVersion, err := sendArgsFromPDUAndValidateExistsAndGetVersion(ctx, fs, r.GetSenderVersion())
-	if err != nil {
-		msg := "HintMostRecentCommonAncestor rpc with nonexistent most recent version"
-		log.Warn(msg)
-		return nil, errors.Wrap(err, msg)
-	}
-
-	// move replication cursor to this position
-	destroyedCursors, err := MoveReplicationCursor(ctx, fs, mostRecentVersion, p.jobId)
-	if err == zfs.ErrBookmarkCloningNotSupported {
-		log.Debug("not creating replication cursor from bookmark because ZFS does not support it")
-		// fallthrough
-	} else if err != nil {
-		return nil, errors.Wrap(err, "cannot set replication cursor to hinted version")
-	}
-
-	// take care of stale step holds
-	log.WithField("step-holds-cleanup-mode", senderHintMostRecentCommonAncestorStepCleanupMode).
-		Debug("taking care of possibly stale step holds")
-	doStepCleanup := false
-	var stepCleanupSince *CreateTXGRangeBound
-	switch senderHintMostRecentCommonAncestorStepCleanupMode {
-	case StepCleanupNoCleanup:
-		doStepCleanup = false
-	case StepCleanupRangeSinceUnbounded:
-		doStepCleanup = true
-		stepCleanupSince = nil
-	case StepCleanupRangeSinceReplicationCursor:
-		doStepCleanup = true
-		// Use the destroyed replication cursors as indicator how far the previous replication got.
-		// To be precise: We limit the amount of visisted snapshots to exactly those snapshots
-		// created since the last successful replication cursor movement (i.e. last successful replication step)
-		//
-		// If we crash now, we'll leak the step we are about to release, but the performance gain
-		// of limiting the amount of snapshots we visit makes up for that.
-		// Users have the `zrepl holds release-stale` command to cleanup leaked step holds.
-		for _, destroyed := range destroyedCursors {
-			if stepCleanupSince == nil {
-				stepCleanupSince = &CreateTXGRangeBound{
-					CreateTXG: destroyed.GetCreateTXG(),
-					Inclusive: &zfs.NilBool{B: true},
-				}
-			} else if destroyed.GetCreateTXG() < stepCleanupSince.CreateTXG {
-				stepCleanupSince.CreateTXG = destroyed.GetCreateTXG()
-			}
-		}
-	default:
-		panic(senderHintMostRecentCommonAncestorStepCleanupMode)
-	}
-	if !doStepCleanup {
-		log.Info("skipping cleanup of prior invocations' step holds due to environment variable setting")
-	} else {
-		if err := ReleaseStepCummulativeInclusive(ctx, fs, stepCleanupSince, mostRecentVersion, p.jobId); err != nil {
-			return nil, errors.Wrap(err, "cannot cleanup prior invocation's step holds and bookmarks")
-		} else {
-			log.Info("step hold cleanup done")
-		}
-	}
-
-	return &pdu.HintMostRecentCommonAncestorRes{}, nil
-}
-
-type HintMostRecentCommonAncestorStepCleanupMode struct{ string }
-
-var (
-	StepCleanupRangeSinceReplicationCursor = HintMostRecentCommonAncestorStepCleanupMode{"range-since-replication-cursor"}
-	StepCleanupRangeSinceUnbounded         = HintMostRecentCommonAncestorStepCleanupMode{"range-since-unbounded"}
-	StepCleanupNoCleanup                   = HintMostRecentCommonAncestorStepCleanupMode{"no-cleanup"}
-)
-
-func (m HintMostRecentCommonAncestorStepCleanupMode) String() string { return string(m.string) }
-func (m *HintMostRecentCommonAncestorStepCleanupMode) Set(s string) error {
-	switch s {
-	case StepCleanupRangeSinceReplicationCursor.String():
-		*m = StepCleanupRangeSinceReplicationCursor
-	case StepCleanupRangeSinceUnbounded.String():
-		*m = StepCleanupRangeSinceUnbounded
-	case StepCleanupNoCleanup.String():
-		*m = StepCleanupNoCleanup
-	default:
-		return fmt.Errorf("unknown step cleanup mode %q", s)
-	}
-	return nil
-}
-
-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))
+var maxConcurrentZFSSend = envconst.Int64("ZREPL_ENDPOINT_MAX_CONCURRENT_SEND", 10)
+var maxConcurrentZFSSendSemaphore = semaphore.New(maxConcurrentZFSSend)
 
 func uncheckedSendArgsFromPDU(fsv *pdu.FilesystemVersion) *zfs.ZFSSendArgVersion {
 	if fsv == nil {
@@ -319,10 +208,11 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
 		return res, nil, nil
 	}
 
-	// update replication cursor
+	// create a replication cursor for `From` (usually an idempotent no-op because SendCompleted already created it before)
+	var fromReplicationCursor Abstraction
 	if sendArgs.From != nil {
 		// For all but the first replication, this should always be a no-op because SendCompleted already moved the cursor
-		_, err = MoveReplicationCursor(ctx, sendArgs.FS, sendArgs.FromVersion, s.jobId)
+		fromReplicationCursor, err = CreateReplicationCursor(ctx, sendArgs.FS, *sendArgs.FromVersion, s.jobId) // no shadow
 		if err == zfs.ErrBookmarkCloningNotSupported {
 			getLogger(ctx).Debug("not creating replication cursor from bookmark because ZFS does not support it")
 			// fallthrough
@@ -331,9 +221,10 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
 		}
 	}
 
+	var fromHold, toHold Abstraction
 	// make sure `From` doesn't go away in order to make this step resumable
 	if sendArgs.From != nil {
-		_, err := HoldStep(ctx, sendArgs.FS, *sendArgs.FromVersion, s.jobId)
+		fromHold, err = HoldStep(ctx, sendArgs.FS, *sendArgs.FromVersion, s.jobId) // no shadow
 		if err == zfs.ErrBookmarkCloningNotSupported {
 			getLogger(ctx).Debug("not creating step bookmark because ZFS does not support it")
 			// fallthrough
@@ -342,17 +233,71 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
 		}
 	}
 	// make sure `To` doesn't go away in order to make this step resumable
-	_, err = HoldStep(ctx, sendArgs.FS, sendArgs.ToVersion, s.jobId)
+	toHold, err = HoldStep(ctx, sendArgs.FS, sendArgs.ToVersion, s.jobId)
 	if err != nil {
 		return nil, nil, errors.Wrapf(err, "cannot hold `to` version %q before starting send", sendArgs.ToVersion)
 	}
 
-	// step holds & replication cursor released / moved forward in s.SendCompleted => s.moveCursorAndReleaseSendHolds
+	// cleanup the mess that _this function_ might have created in prior failed attempts:
+	//
+	// In summary, we delete every endpoint ZFS abstraction created on this filesystem for this job id,
+	// except for the ones we just created above.
+	//
+	// This is the most robust approach to avoid leaking (= forgetting to clean up) endpoint ZFS abstractions,
+	// all under the assumption that there will only ever be one send for a (jobId,fs) combination at any given time.
+	//
+	// Note that the SendCompleted rpc can't be relied upon for this purpose:
+	// - it might be lost due to network errors,
+	// - or never be sent by a potentially malicious or buggy client,
+	// - or never be send because the replication step failed at some point
+	//   (potentially leaving a resumable state on the receiver, which is the case where we really do not want to blow away the step holds too soon.)
+	//
+	// Note further that a resuming send, due to the idempotent nature of func CreateReplicationCursor and HoldStep,
+	// will never lose its step holds because we just (idempotently re-)created them above, before attempting the cleanup.
+	func() {
+		ctx, endSpan := trace.WithSpan(ctx, "cleanup-stale-abstractions")
+		defer endSpan()
+
+		liveAbs := []Abstraction{fromHold, toHold, fromReplicationCursor}
+		keep := func(a Abstraction) (keep bool) {
+			keep = false
+			for _, k := range liveAbs {
+				keep = keep || AbstractionEquals(a, k)
+			}
+			return keep
+		}
+		check := func(obsoleteAbs []Abstraction) {
+			// last line of defense: check that we don't destroy the incremental `from` and `to`
+			// if we did that, we might be about to blow away the last common filesystem version between sender and receiver
+			mustLiveVersions := []zfs.FilesystemVersion{sendArgs.ToVersion}
+			if sendArgs.FromVersion != nil {
+				mustLiveVersions = append(mustLiveVersions, *sendArgs.FromVersion)
+			}
+			for _, staleVersion := range obsoleteAbs {
+				for _, mustLiveVersion := range mustLiveVersions {
+					if zfs.FilesystemVersionEqualIdentity(mustLiveVersion, staleVersion.GetFilesystemVersion()) {
+						panic(fmt.Sprintf("impl error: %q would be destroyed because it is considered stale but it is part of of sendArgs=%s", mustLiveVersion.String(), pretty.Sprint(sendArgs)))
+					}
+				}
+			}
+		}
+		sendAbstractionsCacheSingleton.TryBatchDestroy(ctx, s.jobId, sendArgs.FS, keep, check)
+	}()
+
+	if fromHold != nil {
+		sendAbstractionsCacheSingleton.Put(fromHold)
+	}
+	sendAbstractionsCacheSingleton.Put(toHold)
+	if fromReplicationCursor != nil {
+		sendAbstractionsCacheSingleton.Put(fromReplicationCursor)
+	}
 
 	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
 	if err != nil {
+		// it's ok to not destroy the abstractions we just created here, a new send attempt will take care of it
 		return nil, nil, errors.Wrap(err, "zfs send failed")
 	}
+
 	return res, sendStream, nil
 }
 
@@ -389,8 +334,7 @@ func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*p
 		return log
 	}
 
-	log(ctx).Debug("move replication cursor to most recent common version")
-	destroyedCursors, err := MoveReplicationCursor(ctx, fs, to, p.jobId)
+	toReplicationCursor, err := CreateReplicationCursor(ctx, fs, to, p.jobId)
 	if err != nil {
 		if err == zfs.ErrBookmarkCloningNotSupported {
 			log(ctx).Debug("not setting replication cursor, bookmark cloning not supported")
@@ -402,59 +346,17 @@ func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*p
 			return &pdu.SendCompletedRes{}, err
 		}
 	} else {
-		log(ctx).Info("successfully moved replication cursor")
+		sendAbstractionsCacheSingleton.Put(toReplicationCursor)
+		log(ctx).WithField("to_cursor", toReplicationCursor.String()).Info("successfully created `to` replication cursor")
 	}
 
-	// kick off releasing of step holds / bookmarks
-	// if we fail to release them, don't bother the caller:
-	// they are merely an implementation detail on the sender for better resumability
-	var wg sync.WaitGroup
-	wg.Add(2)
-	go func() {
-		defer wg.Done()
-		ctx, endTask := trace.WithTask(ctx, "release-step-hold-to")
-		defer endTask()
-
-		log(ctx).Debug("release step-hold of or step-bookmark on `to`")
-		err = ReleaseStep(ctx, fs, to, p.jobId)
-		if err != nil {
-			log(ctx).WithError(err).Error("cannot release step-holds on or destroy step-bookmark of `to`")
-		} else {
-			log(ctx).Info("successfully released step-holds on or destroyed step-bookmark of `to`")
-		}
-
-	}()
-	go func() {
-		defer wg.Done()
-		ctx, endTask := trace.WithTask(ctx, "release-step-hold-from")
-		defer endTask()
-
-		if from == nil {
-			return
-		}
-		log(ctx).Debug("release step-hold of or step-bookmark on `from`")
-		err := ReleaseStep(ctx, fs, *from, p.jobId)
-		if err != nil {
-			if dne, ok := err.(*zfs.DatasetDoesNotExist); ok {
-				// If bookmark cloning is not supported, `from` might be the old replication cursor
-				// and thus have already been destroyed by MoveReplicationCursor above
-				// In that case, nonexistence of `from` is not an error, otherwise it is.
-				for _, c := range destroyedCursors {
-					if c.GetFullPath() == dne.Path {
-						log(ctx).Info("`from` was a replication cursor and has already been destroyed")
-						return
-					}
-				}
-				// fallthrough
-			}
-			log(ctx).WithError(err).Error("cannot release step-holds on or destroy step-bookmark of `from`")
-		} else {
-			log(ctx).Info("successfully released step-holds on or destroyed step-bookmark of `from`")
-		}
-	}()
-	wg.Wait()
+	keep := func(a Abstraction) bool {
+		return AbstractionEquals(a, toReplicationCursor)
+	}
+	sendAbstractionsCacheSingleton.TryBatchDestroy(ctx, p.jobId, fs, keep, nil)
 
 	return &pdu.SendCompletedRes{}, nil
+
 }
 
 func (p *Sender) DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshotsReq) (*pdu.DestroySnapshotsRes, error) {
@@ -974,16 +876,6 @@ func (s *Receiver) DestroySnapshots(ctx context.Context, req *pdu.DestroySnapsho
 	return doDestroySnapshots(ctx, lp, req.Snapshots)
 }
 
-func (p *Receiver) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.HintMostRecentCommonAncestorReq) (*pdu.HintMostRecentCommonAncestorRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-	// we don't move last-received-hold as part of this hint
-	// because that wouldn't give us any benefit wrt resumability.
-	//
-	// Other reason: the replication logic that issues this RPC would require refactoring
-	// to include the receiver's FilesystemVersion in the request)
-	return &pdu.HintMostRecentCommonAncestorRes{}, nil
-}
-
 func (p *Receiver) SendCompleted(ctx context.Context, _ *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
 	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 

endpoint/endpoint_metrics.go

@@ -0,0 +1,7 @@
+package endpoint
+
+import "github.com/prometheus/client_golang/prometheus"
+
+func RegisterMetrics(r prometheus.Registerer) {
+	r.MustRegister(sendAbstractionsCacheMetrics.count)
+}

endpoint/endpoint_send_abstractions_cache.go

@@ -0,0 +1,203 @@
+package endpoint
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/zrepl/zrepl/daemon/logging/trace"
+	"github.com/zrepl/zrepl/util/chainlock"
+)
+
+var sendAbstractionsCacheMetrics struct {
+	count prometheus.Gauge
+}
+
+func init() {
+	sendAbstractionsCacheMetrics.count = prometheus.NewGauge(prometheus.GaugeOpts{
+		Namespace: "zrepl",
+		Subsystem: "endpoint",
+		Name:      "send_abstractions_cache_entry_count",
+		Help:      "number of send abstractions tracked in the sendAbstractionsCache data structure",
+	})
+}
+
+var sendAbstractionsCacheSingleton = newSendAbstractionsCache()
+
+type sendAbstractionsCacheDidLoadFSState int
+
+const (
+	sendAbstractionsCacheDidLoadFSStateNo sendAbstractionsCacheDidLoadFSState = iota // 0-value has meaning
+	sendAbstractionsCacheDidLoadFSStateInProgress
+	sendAbstractionsCacheDidLoadFSStateDone
+)
+
+type sendAbstractionsCache struct {
+	mtx              chainlock.L
+	abstractions     []Abstraction
+	didLoadFS        map[string]sendAbstractionsCacheDidLoadFSState
+	didLoadFSChanged *sync.Cond
+}
+
+func newSendAbstractionsCache() *sendAbstractionsCache {
+	c := &sendAbstractionsCache{
+		didLoadFS: make(map[string]sendAbstractionsCacheDidLoadFSState),
+	}
+	c.didLoadFSChanged = c.mtx.NewCond()
+	return c
+}
+
+func (s *sendAbstractionsCache) Put(a Abstraction) {
+	defer s.mtx.Lock().Unlock()
+
+	var zeroJobId JobID
+	if a.GetJobID() == nil {
+		panic("abstraction must not have nil job id")
+	} else if *a.GetJobID() == zeroJobId {
+		panic(fmt.Sprintf("abstraction must not have zero-value job id: %s", a))
+	}
+
+	s.abstractions = append(s.abstractions, a)
+	sendAbstractionsCacheMetrics.count.Set(float64(len(s.abstractions)))
+}
+
+func (s *sendAbstractionsCache) InvalidateFSCache(fs string) {
+	// FIXME: O(n)
+	newAbs := make([]Abstraction, 0, len(s.abstractions))
+	for _, a := range s.abstractions {
+		if a.GetFS() != fs {
+			newAbs = append(newAbs, a)
+		}
+	}
+	s.abstractions = newAbs
+	sendAbstractionsCacheMetrics.count.Set(float64(len(s.abstractions)))
+
+	s.didLoadFS[fs] = sendAbstractionsCacheDidLoadFSStateNo
+	s.didLoadFSChanged.Broadcast()
+
+}
+
+// - logs errors in getting on-disk abstractions
+// - only fetches on-disk abstractions once, but every time from the in-memory store
+//
+// That means that for precise results, all abstractions created by the endpoint must be .Put into this cache.
+func (s *sendAbstractionsCache) GetAndDeleteByJobIDAndFS(ctx context.Context, jobID JobID, fs string, keep func(a Abstraction) bool) (ret []Abstraction) {
+	defer s.mtx.Lock().Unlock()
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+	var zeroJobId JobID
+	if jobID == zeroJobId {
+		panic("must not pass zero-value job id")
+	}
+	if fs == "" {
+		panic("must not pass zero-value fs")
+	}
+
+	s.tryLoadOnDiskSendAbstractions(ctx, fs)
+
+	// FIXME O(n)
+	var remaining []Abstraction
+	for _, a := range s.abstractions {
+		aJobId := *a.GetJobID()
+		aFS := a.GetFS()
+		if aJobId == jobID && aFS == fs && !keep(a) {
+			ret = append(ret, a)
+		} else {
+			remaining = append(remaining, a)
+		}
+	}
+	s.abstractions = remaining
+	sendAbstractionsCacheMetrics.count.Set(float64(len(s.abstractions)))
+
+	return ret
+}
+
+// caller must hold s.mtx
+func (s *sendAbstractionsCache) tryLoadOnDiskSendAbstractions(ctx context.Context, fs string) {
+	for s.didLoadFS[fs] != sendAbstractionsCacheDidLoadFSStateDone {
+		if s.didLoadFS[fs] == sendAbstractionsCacheDidLoadFSStateInProgress {
+			s.didLoadFSChanged.Wait()
+			continue
+		}
+		if s.didLoadFS[fs] != sendAbstractionsCacheDidLoadFSStateNo {
+			panic(fmt.Sprintf("unreachable: %v", s.didLoadFS[fs]))
+		}
+
+		s.didLoadFS[fs] = sendAbstractionsCacheDidLoadFSStateInProgress
+		defer s.didLoadFSChanged.Broadcast()
+
+		var onDiskAbs []Abstraction
+		var err error
+		s.mtx.DropWhile(func() {
+			onDiskAbs, err = s.tryLoadOnDiskSendAbstractionsImpl(ctx, fs) // no shadow
+		})
+
+		if err != nil {
+			s.didLoadFS[fs] = sendAbstractionsCacheDidLoadFSStateNo
+			getLogger(ctx).WithField("fs", fs).WithError(err).Error("cannot list send step abstractions for filesystem")
+		} else {
+			s.didLoadFS[fs] = sendAbstractionsCacheDidLoadFSStateDone
+			s.abstractions = append(s.abstractions, onDiskAbs...)
+			getLogger(ctx).WithField("fs", fs).WithField("abstractions", onDiskAbs).Debug("loaded step abstractions for filesystem")
+		}
+		return
+	}
+}
+
+// caller should _not hold s.mtx
+func (s *sendAbstractionsCache) tryLoadOnDiskSendAbstractionsImpl(ctx context.Context, fs string) ([]Abstraction, error) {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+
+	q := ListZFSHoldsAndBookmarksQuery{
+		FS: ListZFSHoldsAndBookmarksQueryFilesystemFilter{
+			FS: &fs,
+		},
+		JobID: nil,
+		What: AbstractionTypeSet{
+			AbstractionStepHold:                    true,
+			AbstractionStepBookmark:                true,
+			AbstractionReplicationCursorBookmarkV2: true,
+		},
+		Concurrency: 1,
+	}
+	abs, absErrs, err := ListAbstractions(ctx, q)
+	if err != nil {
+		return nil, err
+	}
+	// safe to ignore absErrs here, this is best-effort cleanup
+	if len(absErrs) > 0 {
+		return nil, ListAbstractionsErrors(absErrs)
+	}
+	return abs, nil
+}
+
+func (s *sendAbstractionsCache) TryBatchDestroy(ctx context.Context, jobId JobID, fs string, keep func(a Abstraction) bool, check func(willDestroy []Abstraction)) {
+	// no s.mtx, we only use the public interface in this function
+
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+
+	obsoleteAbs := s.GetAndDeleteByJobIDAndFS(ctx, jobId, fs, keep)
+
+	if check != nil {
+		check(obsoleteAbs)
+	}
+
+	hadErr := false
+	for res := range BatchDestroy(ctx, obsoleteAbs) {
+		if res.DestroyErr != nil {
+			hadErr = true
+			getLogger(ctx).
+				WithField("abstraction", res.Abstraction).
+				WithError(res.DestroyErr).
+				Error("cannot destroy stale send step abstraction")
+		} else {
+			getLogger(ctx).
+				WithField("abstraction", res.Abstraction).
+				Info("destroyed stale send step abstraction")
+		}
+	}
+	if hadErr {
+		s.InvalidateFSCache(fs)
+	}
+
+}

endpoint/endpoint_zfs_abstraction.go

@@ -54,6 +54,30 @@ type Abstraction interface {
 	json.Marshaler
 }
 
+func AbstractionEquals(a, b Abstraction) bool {
+	if (a != nil) != (b != nil) {
+		return false
+	}
+	if a == nil && b == nil {
+		return true
+	}
+	var aJobId, bJobId JobID
+	if aJid := a.GetJobID(); aJid != nil {
+		aJobId = *aJid
+	}
+	if bJid := b.GetJobID(); bJid != nil {
+		bJobId = *bJid
+	}
+	return a.GetType() == b.GetType() &&
+		a.GetFS() == b.GetFS() &&
+		a.GetName() == b.GetName() &&
+		a.GetFullPath() == b.GetFullPath() &&
+		aJobId == bJobId &&
+		a.GetCreateTXG() == b.GetCreateTXG() &&
+		zfs.FilesystemVersionEqualIdentity(a.GetFilesystemVersion(), b.GetFilesystemVersion()) &&
+		a.String() == b.String()
+}
+
 func (t AbstractionType) Validate() error {
 	switch t {
 	case AbstractionStepBookmark:

endpoint/endpoint_zfs_abstraction_cursor_and_last_received_hold.go

@@ -118,31 +118,33 @@ func GetReplicationCursors(ctx context.Context, dp *zfs.DatasetPath, jobID JobID
 	return candidates, nil
 }
 
-type ReplicationCursorTarget interface {
-	IsSnapshot() bool
-	GetGuid() uint64
-	GetCreateTXG() uint64
-	ToSendArgVersion() zfs.ZFSSendArgVersion
-}
-
-// `target` is validated before replication cursor is set. if validation fails, the cursor is not moved.
+// idempotently create a replication cursor targeting `target`
 //
 // returns ErrBookmarkCloningNotSupported if version is a bookmark and bookmarking bookmarks is not supported by ZFS
-func MoveReplicationCursor(ctx context.Context, fs string, target ReplicationCursorTarget, jobID JobID) (destroyedCursors []Abstraction, err error) {
-
-	if !target.IsSnapshot() {
-		return nil, zfs.ErrBookmarkCloningNotSupported
-	}
+func CreateReplicationCursor(ctx context.Context, fs string, target zfs.FilesystemVersion, jobID JobID) (a Abstraction, err error) {
 
 	bookmarkname, err := ReplicationCursorBookmarkName(fs, target.GetGuid(), jobID)
 	if err != nil {
 		return nil, errors.Wrap(err, "determine replication cursor name")
 	}
 
+	if target.IsBookmark() && target.GetName() == bookmarkname {
+		return &bookmarkBasedAbstraction{
+			Type:              AbstractionReplicationCursorBookmarkV2,
+			FS:                fs,
+			FilesystemVersion: target,
+			JobID:             jobID,
+		}, nil
+	}
+
+	if !target.IsSnapshot() {
+		return nil, zfs.ErrBookmarkCloningNotSupported
+	}
+
 	// idempotently create bookmark (guid is encoded in it, hence we'll most likely add a new one
 	// cleanup the old one afterwards
 
-	err = zfs.ZFSBookmark(ctx, fs, target.ToSendArgVersion(), bookmarkname)
+	cursorBookmark, err := zfs.ZFSBookmark(ctx, fs, target, bookmarkname)
 	if err != nil {
 		if err == zfs.ErrBookmarkCloningNotSupported {
 			return nil, err // TODO go1.13 use wrapping
@@ -150,12 +152,12 @@ func MoveReplicationCursor(ctx context.Context, fs string, target ReplicationCur
 		return nil, errors.Wrapf(err, "cannot create bookmark")
 	}
 
-	destroyedCursors, err = DestroyObsoleteReplicationCursors(ctx, fs, target, jobID)
-	if err != nil {
-		return nil, errors.Wrap(err, "destroy obsolete replication cursors")
-	}
-
-	return destroyedCursors, nil
+	return &bookmarkBasedAbstraction{
+		Type:              AbstractionReplicationCursorBookmarkV2,
+		FS:                fs,
+		FilesystemVersion: cursorBookmark,
+		JobID:             jobID,
+	}, nil
 }
 
 type ReplicationCursor interface {

endpoint/endpoint_zfs_abstraction_step.go

@@ -93,7 +93,7 @@ func HoldStep(ctx context.Context, fs string, v zfs.FilesystemVersion, jobID Job
 		return nil, errors.Wrap(err, "create step bookmark: determine bookmark name")
 	}
 	// idempotently create bookmark
-	err = zfs.ZFSBookmark(ctx, fs, v.ToSendArgVersion(), bmname)
+	stepBookmark, err := zfs.ZFSBookmark(ctx, fs, v, bmname)
 	if err != nil {
 		if err == zfs.ErrBookmarkCloningNotSupported {
 			// TODO we could actually try to find a local snapshot that has the requested GUID
@@ -108,7 +108,7 @@ func HoldStep(ctx context.Context, fs string, v zfs.FilesystemVersion, jobID Job
 	return &bookmarkBasedAbstraction{
 		Type:              AbstractionStepBookmark,
 		FS:                fs,
-		FilesystemVersion: v,
+		FilesystemVersion: stepBookmark,
 		JobID:             jobID,
 	}, nil
 }
@@ -236,7 +236,6 @@ func TryReleaseStepStaleFS(ctx context.Context, fs string, jobID JobID) {
 				Info("destroyed stale step-hold or bookmark")
 		}
 	}
-
 }
 
 var _ BookmarkExtractor = StepBookmarkExtractor