new features: {resumable,encrypted,hold-protected} send-recv, last-received-hold

- **Resumable Send & Recv Support**
  No knobs required, automatically used where supported.
- **Hold-Protected Send & Recv**
  Automatic ZFS holds to ensure that we can always resume a replication step.
- **Encrypted Send & Recv Support** for OpenZFS native encryption.
  Configurable at the job level, i.e., for all filesystems a job is responsible for.
- **Receive-side hold on last received dataset**
  The counterpart to the replication cursor bookmark on the send-side.
  Ensures that incremental replication will always be possible between a sender and receiver.

Design Doc
----------

`replication/design.md` doc describes how we use ZFS holds and bookmarks to ensure that a single replication step is always resumable.

The replication algorithm described in the design doc introduces the notion of job IDs (please read the details on this design doc).
We reuse the job names for job IDs and use `JobID` type to ensure that a job name can be embedded into hold tags, bookmark names, etc.
This might BREAK CONFIG on upgrade.

Protocol Version Bump
---------------------

This commit makes backwards-incompatible changes to the replication/pdu protobufs.
Thus, bump the version number used in the protocol handshake.

Replication Cursor Format Change
--------------------------------

The new replication cursor bookmark format is: `#zrepl_CURSOR_G_${this.GUID}_J_${jobid}`
Including the GUID enables transaction-safe moving-forward of the cursor.
Including the job id enables that multiple sending jobs can send the same filesystem without interfering.
The `zrepl migrate replication-cursor:v1-v2` subcommand can be used to safely destroy old-format cursors once zrepl has created new-format cursors.

Changes in This Commit
----------------------

- package zfs
  - infrastructure for holds
  - infrastructure for resume token decoding
  - implement a variant of OpenZFS's `entity_namecheck` and use it for validation in new code
  - ZFSSendArgs to specify a ZFS send operation
    - validation code protects against malicious resume tokens by checking that the token encodes the same send parameters that the send-side would use if no resume token were available (i.e. same filesystem, `fromguid`, `toguid`)
  - RecvOptions support for `recv -s` flag
  - convert a bunch of ZFS operations to be idempotent
    - achieved through more differentiated error message scraping / additional pre-/post-checks

- package replication/pdu
  - add field for encryption to send request messages
  - add fields for resume handling to send & recv request messages
  - receive requests now contain `FilesystemVersion To` in addition to the filesystem into which the stream should be `recv`d into
    - can use `zfs recv $root_fs/$client_id/path/to/dataset@${To.Name}`, which enables additional validation after recv (i.e. whether `To.Guid` matched what we received in the stream)
    - used to set `last-received-hold`
- package replication/logic
  - introduce `PlannerPolicy` struct, currently only used to configure whether encrypted sends should be requested from the sender
  - integrate encryption and resume token support into `Step` struct

- package endpoint
  - move the concepts that endpoint builds on top of ZFS to a single file `endpoint/endpoint_zfs.go`
    - step-holds + step-bookmarks
    - last-received-hold
    - new replication cursor + old replication cursor compat code
  - adjust `endpoint/endpoint.go` handlers for
    - encryption
    - resumability
    - new replication cursor
    - last-received-hold

- client subcommand `zrepl holds list`: list all holds and hold-like bookmarks that zrepl thinks belong to it
- client subcommand `zrepl migrate replication-cursor:v1-v2`

endpoint/endpoint.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"fmt"
 	"path"
+	"sync"
 
 	"github.com/pkg/errors"
 
@@ -15,13 +16,39 @@ import (
 	"github.com/zrepl/zrepl/zfs"
 )
 
+type SenderConfig struct {
+	FSF     zfs.DatasetFilter
+	Encrypt *zfs.NilBool
+	JobID   JobID
+}
+
+func (c *SenderConfig) Validate() error {
+	c.JobID.MustValidate()
+	if err := c.Encrypt.Validate(); err != nil {
+		return errors.Wrap(err, "`Encrypt` field invalid")
+	}
+	if _, err := StepHoldTag(c.JobID); err != nil {
+		return fmt.Errorf("JobID cannot be used for hold tag: %s", err)
+	}
+	return nil
+}
+
 // Sender implements replication.ReplicationEndpoint for a sending side
 type Sender struct {
 	FSFilter zfs.DatasetFilter
+	encrypt  *zfs.NilBool
+	jobId    JobID
 }
 
-func NewSender(fsf zfs.DatasetFilter) *Sender {
-	return &Sender{FSFilter: fsf}
+func NewSender(conf SenderConfig) *Sender {
+	if err := conf.Validate(); err != nil {
+		panic("invalid config" + err.Error())
+	}
+	return &Sender{
+		FSFilter: conf.FSF,
+		encrypt:  conf.Encrypt,
+		jobId:    conf.JobID,
+	}
 }
 
 func (s *Sender) filterCheckFS(fs string) (*zfs.DatasetPath, error) {
@@ -49,10 +76,15 @@ func (s *Sender) ListFilesystems(ctx context.Context, r *pdu.ListFilesystemReq)
 	}
 	rfss := make([]*pdu.Filesystem, len(fss))
 	for i := range fss {
+		encEnabled, err := zfs.ZFSGetEncryptionEnabled(ctx, fss[i].ToString())
+		if err != nil {
+			return nil, errors.Wrap(err, "cannot get filesystem encryption status")
+		}
 		rfss[i] = &pdu.Filesystem{
 			Path: fss[i].ToString(),
-			// FIXME: not supporting ResumeToken yet
+			// ResumeToken does not make sense from Sender
 			IsPlaceholder: false, // sender FSs are never placeholders
+			IsEncrypted:   encEnabled,
 		}
 	}
 	res := &pdu.ListFilesystemRes{Filesystems: rfss}
@@ -77,13 +109,86 @@ func (s *Sender) ListFilesystemVersions(ctx context.Context, r *pdu.ListFilesyst
 
 }
 
+func (p *Sender) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.HintMostRecentCommonAncestorReq) (*pdu.HintMostRecentCommonAncestorRes, error) {
+	var err error
+
+	fs := r.GetFilesystem()
+	mostRecent, err := sendArgsFromPDUAndValidateExists(ctx, fs, r.GetSenderVersion())
+	if err != nil {
+		msg := "HintMostRecentCommonAncestor rpc with nonexistent most recent version"
+		getLogger(ctx).WithField("fs", fs).WithField("hinted_most_recent", fmt.Sprintf("%#v", mostRecent)).
+			Warn(msg)
+		return nil, errors.Wrap(err, msg)
+	}
+
+	// move replication cursor to this position
+	_, err = MoveReplicationCursor(ctx, fs, mostRecent, p.jobId)
+	if err == zfs.ErrBookmarkCloningNotSupported {
+		getLogger(ctx).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")
+	}
+
+	// cleanup previous steps
+	if err := ReleaseStepAll(ctx, fs, mostRecent, p.jobId); err != nil {
+		return nil, errors.Wrap(err, "cannot cleanup prior invocation's step holds and bookmarks")
+	}
+
+	return &pdu.HintMostRecentCommonAncestorRes{}, nil
+}
+
 var maxConcurrentZFSSendSemaphore = semaphore.New(envconst.Int64("ZREPL_ENDPOINT_MAX_CONCURRENT_SEND", 10))
 
+func uncheckedSendArgsFromPDU(fsv *pdu.FilesystemVersion) *zfs.ZFSSendArgVersion {
+	if fsv == nil {
+		return nil
+	}
+	return &zfs.ZFSSendArgVersion{RelName: fsv.GetRelName(), GUID: fsv.Guid}
+}
+
+func sendArgsFromPDUAndValidateExists(ctx context.Context, fs string, fsv *pdu.FilesystemVersion) (*zfs.ZFSSendArgVersion, error) {
+	v := uncheckedSendArgsFromPDU(fsv)
+	if v == nil {
+		return nil, errors.New("must not be nil")
+	}
+	if err := v.ValidateExists(ctx, fs); err != nil {
+		return nil, err
+	}
+	return v, nil
+}
+
 func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
+
 	_, err := s.filterCheckFS(r.Filesystem)
 	if err != nil {
 		return nil, nil, err
 	}
+	switch r.Encrypted {
+	case pdu.Tri_DontCare:
+		// use s.encrypt setting
+		// ok, fallthrough outer
+	case pdu.Tri_False:
+		if s.encrypt.B {
+			return nil, nil, errors.New("only encrytped sends allowed (send -w + encryption!= off), but unencrytped send requested")
+		}
+		// fallthrough outer
+	case pdu.Tri_True:
+		if !s.encrypt.B {
+			return nil, nil, errors.New("only unencrypted sends allowed, but encrypted send requested")
+		}
+		// fallthrough outer
+	default:
+		return nil, nil, fmt.Errorf("unknown pdu.Tri variant %q", r.Encrypted)
+	}
+
+	sendArgs := zfs.ZFSSendArgs{
+		FS:          r.Filesystem,
+		From:        uncheckedSendArgsFromPDU(r.GetFrom()), // validated by zfs.ZFSSendDry / zfs.ZFSSend
+		To:          uncheckedSendArgsFromPDU(r.GetTo()),   // validated by zfs.ZFSSendDry / zfs.ZFSSend
+		Encrypted:   s.encrypt,
+		ResumeToken: r.ResumeToken, // nil or not nil, depending on decoding success
+	}
 
 	getLogger(ctx).Debug("acquire concurrent send semaphore")
 	// TODO use try-acquire and fail with resource-exhaustion rpc status
@@ -95,28 +200,154 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St
 	}
 	defer guard.Release()
 
-	si, err := zfs.ZFSSendDry(r.Filesystem, r.From, r.To, "")
+	si, err := zfs.ZFSSendDry(ctx, sendArgs)
 	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 invovles 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}
+	res := &pdu.SendRes{
+		ExpectedSize:    expSize,
+		UsedResumeToken: r.ResumeToken != "",
+	}
 
 	if r.DryRun {
 		return res, nil, nil
 	}
 
-	streamCopier, err := zfs.ZFSSend(ctx, r.Filesystem, r.From, r.To, "")
+	// update replication cursor
+	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.From, s.jobId)
+		if err == zfs.ErrBookmarkCloningNotSupported {
+			getLogger(ctx).Debug("not creating replication cursor from bookmark because ZFS does not support it")
+			// fallthrough
+		} else if err != nil {
+			return nil, nil, errors.Wrap(err, "cannot set replication cursor to `from` version before starting send")
+		}
+	}
+
+	// make sure `From` doesn't go away in order to make this step resumable
+	if sendArgs.From != nil {
+		err := HoldStep(ctx, sendArgs.FS, sendArgs.From, s.jobId)
+		if err == zfs.ErrBookmarkCloningNotSupported {
+			getLogger(ctx).Debug("not creating step bookmark because ZFS does not support it")
+			// fallthrough
+		} else if err != nil {
+			return nil, nil, errors.Wrap(err, "cannot create step bookmark")
+		}
+	}
+	// make sure `To` doesn't go away in order to make this step resumable
+	err = HoldStep(ctx, sendArgs.FS, sendArgs.To, s.jobId)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, errors.Wrapf(err, "cannot hold `to` version %q before starting send", sendArgs.To.RelName)
+	}
+
+	// step holds & replication cursor released / moved forward in s.SendCompleted => s.moveCursorAndReleaseSendHolds
+
+	streamCopier, err := zfs.ZFSSend(ctx, sendArgs)
+	if err != nil {
+		return nil, nil, errors.Wrap(err, "zfs send failed")
 	}
 	return res, streamCopier, nil
 }
 
+func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
+	orig := r.GetOriginalReq() // may be nil, always use proto getters
+	fs := orig.GetFilesystem()
+
+	var err error
+	var from *zfs.ZFSSendArgVersion
+	if orig.GetFrom() != nil {
+		from, err = sendArgsFromPDUAndValidateExists(ctx, fs, orig.GetFrom()) // no shadow
+		if err != nil {
+			return nil, errors.Wrap(err, "validate `from` exists")
+		}
+	}
+	to, err := sendArgsFromPDUAndValidateExists(ctx, fs, orig.GetTo())
+	if err != nil {
+		return nil, errors.Wrap(err, "validate `to` exists")
+	}
+
+	log := getLogger(ctx).WithField("to_guid", to.GUID).
+		WithField("fs", fs).
+		WithField("to", to.RelName)
+	if from != nil {
+		log = log.WithField("from", from.RelName).WithField("from_guid", from.GUID)
+	}
+
+	log.Debug("move replication cursor to most recent common version")
+	destroyedCursors, err := MoveReplicationCursor(ctx, fs, to, p.jobId)
+	if err != nil {
+		if err == zfs.ErrBookmarkCloningNotSupported {
+			log.Debug("not setting replication cursor, bookmark cloning not supported")
+		} else {
+			msg := "cannot move replication cursor, keeping hold on `to` until successful"
+			log.WithError(err).Error(msg)
+			err = errors.Wrap(err, msg)
+			// it is correct to not release the hold if we can't move the cursor!
+			return &pdu.SendCompletedRes{}, err
+		}
+	} else {
+		log.Info("successfully moved 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()
+		log.Debug("release step-hold of or step-bookmark on `to`")
+		err = ReleaseStep(ctx, fs, to, p.jobId)
+		if err != nil {
+			log.WithError(err).Error("cannot release step-holds on or destroy step-bookmark of `to`")
+		} else {
+			log.Info("successfully released step-holds on or destroyed step-bookmark of `to`")
+		}
+
+	}()
+	go func() {
+		defer wg.Done()
+		if from == nil {
+			return
+		}
+		log.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.
+				fsp, err := zfs.NewDatasetPath(fs)
+				if err != nil {
+					panic(err) // fs has been validated multiple times above
+				}
+				for _, fsv := range destroyedCursors {
+					if fsv.ToAbsPath(fsp) == dne.Path {
+						log.Info("`from` was a replication cursor and has already been destroyed")
+						return
+					}
+				}
+				// fallthrough
+			}
+			log.WithError(err).Error("cannot release step-holds on or destroy step-bookmark of `from`")
+		} else {
+			log.Info("successfully released step-holds on or destroyed step-bookmark of `from`")
+		}
+	}()
+	wg.Wait()
+
+	return &pdu.SendCompletedRes{}, nil
+}
+
 func (p *Sender) DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshotsReq) (*pdu.DestroySnapshotsRes, error) {
 	dp, err := p.filterCheckFS(req.Filesystem)
 	if err != nil {
@@ -146,25 +377,14 @@ func (p *Sender) ReplicationCursor(ctx context.Context, req *pdu.ReplicationCurs
 		return nil, err
 	}
 
-	switch op := req.Op.(type) {
-	case *pdu.ReplicationCursorReq_Get:
-		cursor, err := zfs.ZFSGetReplicationCursor(dp)
-		if err != nil {
-			return nil, err
-		}
-		if cursor == nil {
-			return &pdu.ReplicationCursorRes{Result: &pdu.ReplicationCursorRes_Notexist{Notexist: true}}, nil
-		}
-		return &pdu.ReplicationCursorRes{Result: &pdu.ReplicationCursorRes_Guid{Guid: cursor.Guid}}, nil
-	case *pdu.ReplicationCursorReq_Set:
-		guid, err := zfs.ZFSSetReplicationCursor(dp, op.Set.Snapshot)
-		if err != nil {
-			return nil, err
-		}
-		return &pdu.ReplicationCursorRes{Result: &pdu.ReplicationCursorRes_Guid{Guid: guid}}, nil
-	default:
-		return nil, errors.Errorf("unknown op %T", op)
+	cursor, err := GetMostRecentReplicationCursorOfJob(ctx, dp.ToString(), p.jobId)
+	if err != nil {
+		return nil, err
 	}
+	if cursor == nil {
+		return &pdu.ReplicationCursorRes{Result: &pdu.ReplicationCursorRes_Notexist{Notexist: true}}, nil
+	}
+	return &pdu.ReplicationCursorRes{Result: &pdu.ReplicationCursorRes_Guid{Guid: cursor.Guid}}, nil
 }
 
 func (p *Sender) Receive(ctx context.Context, r *pdu.ReceiveReq, receive zfs.StreamCopier) (*pdu.ReceiveRes, error) {
@@ -183,22 +403,42 @@ type FSMap interface { // FIXME unused
 	AsFilter() FSFilter
 }
 
+type ReceiverConfig struct {
+	JobID JobID
+
+	RootWithoutClientComponent *zfs.DatasetPath // TODO use
+	AppendClientIdentity       bool
+
+	UpdateLastReceivedHold bool
+}
+
+func (c *ReceiverConfig) copyIn() {
+	c.RootWithoutClientComponent = c.RootWithoutClientComponent.Copy()
+}
+
+func (c *ReceiverConfig) Validate() error {
+	c.JobID.MustValidate()
+	if c.RootWithoutClientComponent.Length() <= 0 {
+		return errors.New("RootWithoutClientComponent must not be an empty dataset path")
+	}
+	return nil
+}
+
 // Receiver implements replication.ReplicationEndpoint for a receiving side
 type Receiver struct {
-	rootWithoutClientComponent *zfs.DatasetPath
-	appendClientIdentity       bool
+	conf ReceiverConfig // validated
 
 	recvParentCreationMtx *chainlock.L
 }
 
-func NewReceiver(rootDataset *zfs.DatasetPath, appendClientIdentity bool) *Receiver {
-	if rootDataset.Length() <= 0 {
-		panic(fmt.Sprintf("root dataset must not be an empty path: %v", rootDataset))
+func NewReceiver(config ReceiverConfig) *Receiver {
+	config.copyIn()
+	if err := config.Validate(); err != nil {
+		panic(err)
 	}
 	return &Receiver{
-		rootWithoutClientComponent: rootDataset.Copy(),
-		appendClientIdentity:       appendClientIdentity,
-		recvParentCreationMtx:      chainlock.New(),
+		conf:                  config,
+		recvParentCreationMtx: chainlock.New(),
 	}
 }
 
@@ -221,8 +461,8 @@ func clientRoot(rootFS *zfs.DatasetPath, clientIdentity string) (*zfs.DatasetPat
 }
 
 func (s *Receiver) clientRootFromCtx(ctx context.Context) *zfs.DatasetPath {
-	if !s.appendClientIdentity {
-		return s.rootWithoutClientComponent.Copy()
+	if !s.conf.AppendClientIdentity {
+		return s.conf.RootWithoutClientComponent.Copy()
 	}
 
 	clientIdentity, ok := ctx.Value(ClientIdentityKey).(string)
@@ -230,7 +470,7 @@ func (s *Receiver) clientRootFromCtx(ctx context.Context) *zfs.DatasetPath {
 		panic(fmt.Sprintf("ClientIdentityKey context value must be set"))
 	}
 
-	clientRoot, err := clientRoot(s.rootWithoutClientComponent, clientIdentity)
+	clientRoot, err := clientRoot(s.conf.RootWithoutClientComponent, clientIdentity)
 	if err != nil {
 		panic(fmt.Sprintf("ClientIdentityContextKey must have been validated before invoking Receiver: %s", err))
 	}
@@ -282,8 +522,27 @@ func (s *Receiver) ListFilesystems(ctx context.Context, req *pdu.ListFilesystemR
 			err := errors.Errorf("inconsistent placeholder state: filesystem %q must exist in this context", a.ToString())
 			return nil, err
 		}
+		token, err := zfs.ZFSGetReceiveResumeTokenOrEmptyStringIfNotSupported(ctx, a)
+		if err != nil {
+			l.WithError(err).Error("cannot get receive resume token")
+			return nil, err
+		}
+		encEnabled, err := zfs.ZFSGetEncryptionEnabled(ctx, a.ToString())
+		if err != nil {
+			l.WithError(err).Error("cannot get encryption enabled status")
+			return nil, err
+		}
+		l.WithField("receive_resume_token", token).Debug("receive resume token")
+
 		a.TrimPrefix(root)
-		fss = append(fss, &pdu.Filesystem{Path: a.ToString(), IsPlaceholder: ph.IsPlaceholder})
+
+		fs := &pdu.Filesystem{
+			Path:          a.ToString(),
+			IsPlaceholder: ph.IsPlaceholder,
+			ResumeToken:   token,
+			IsEncrypted:   encEnabled,
+		}
+		fss = append(fss, fs)
 	}
 	if len(fss) == 0 {
 		getLogger(ctx).Debug("no filesystems found")
@@ -344,7 +603,18 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 	root := s.clientRootFromCtx(ctx)
 	lp, err := subroot{root}.MapToLocal(req.Filesystem)
 	if err != nil {
-		return nil, err
+		return nil, errors.Wrap(err, "`Filesystem` invalid")
+	}
+
+	to := uncheckedSendArgsFromPDU(req.GetTo())
+	if to == nil {
+		return nil, errors.New("`To` must not be nil")
+	}
+	if err := to.ValidateInMemory(lp.ToString()); err != nil {
+		return nil, errors.Wrap(err, "`To` invalid")
+	}
+	if !to.IsSnapshot() {
+		return nil, errors.New("`To` must be a snapshot")
 	}
 
 	// create placeholder parent filesystems as appropriate
@@ -380,11 +650,11 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 			}
 
 			if !ph.FSExists {
-				if s.rootWithoutClientComponent.HasPrefix(v.Path) {
+				if s.conf.RootWithoutClientComponent.HasPrefix(v.Path) {
 					if v.Path.Length() == 1 {
 						visitErr = fmt.Errorf("pool %q not imported", v.Path.ToString())
 					} else {
-						visitErr = fmt.Errorf("root_fs %q does not exist", s.rootWithoutClientComponent.ToString())
+						visitErr = fmt.Errorf("root_fs %q does not exist", s.conf.RootWithoutClientComponent.ToString())
 					}
 					getLogger(ctx).WithError(visitErr).Error("placeholders are only created automatically below root_fs")
 					return false
@@ -422,6 +692,17 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 		}
 	}
 
+	if req.ClearResumeToken && ph.FSExists {
+		if err := zfs.ZFSRecvClearResumeToken(lp.ToString()); err != nil {
+			return nil, errors.Wrap(err, "cannot clear resume token")
+		}
+	}
+
+	recvOpts.SavePartialRecvState, err = zfs.ResumeRecvSupported(ctx, lp)
+	if err != nil {
+		return nil, errors.Wrap(err, "cannot determine whether we can use resumable send & recv")
+	}
+
 	getLogger(ctx).Debug("acquire concurrent recv semaphore")
 	// TODO use try-acquire and fail with resource-exhaustion rpc status
 	// => would require handling on the client-side
@@ -434,13 +715,30 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 
 	getLogger(ctx).WithField("opts", fmt.Sprintf("%#v", recvOpts)).Debug("start receive command")
 
-	if err := zfs.ZFSRecv(ctx, lp.ToString(), receive, recvOpts); err != nil {
+	snapFullPath := to.FullPath(lp.ToString())
+	if err := zfs.ZFSRecv(ctx, lp.ToString(), to, receive, recvOpts); err != nil {
 		getLogger(ctx).
 			WithError(err).
 			WithField("opts", recvOpts).
 			Error("zfs receive failed")
 		return nil, err
 	}
+
+	// validate that we actually received what the sender claimed
+	if err := to.ValidateExists(ctx, lp.ToString()); err != nil {
+		msg := "receive request's `To` version does not match what we received in the stream"
+		getLogger(ctx).WithError(err).WithField("snap", snapFullPath).Error(msg)
+		getLogger(ctx).Error("aborting recv request, but keeping received snapshot for inspection")
+		return nil, errors.Wrap(err, msg)
+	}
+
+	if s.conf.UpdateLastReceivedHold {
+		getLogger(ctx).Debug("move last-received-hold")
+		if err := MoveLastReceivedHold(ctx, lp.ToString(), *to, s.conf.JobID); err != nil {
+			return nil, errors.Wrap(err, "cannot move last-received-hold")
+		}
+	}
+
 	return &pdu.ReceiveRes{}, nil
 }
 
@@ -453,6 +751,19 @@ 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) {
+	// 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(context.Context, *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
+	return &pdu.SendCompletedRes{}, nil
+}
+
 func doDestroySnapshots(ctx context.Context, lp *zfs.DatasetPath, snaps []*pdu.FilesystemVersion) (*pdu.DestroySnapshotsRes, error) {
 	reqs := make([]*zfs.DestroySnapOp, len(snaps))
 	ress := make([]*pdu.DestroySnapshotRes, len(snaps))

endpoint/endpoint_zfs.go

@@ -0,0 +1,503 @@
+package endpoint
+
+import (
+	"context"
+	"fmt"
+	"regexp"
+	"sort"
+
+	"github.com/kr/pretty"
+	"github.com/pkg/errors"
+
+	"github.com/zrepl/zrepl/zfs"
+)
+
+var stepHoldTagRE = regexp.MustCompile("^zrepl_STEP_J_(.+)")
+
+func StepHoldTag(jobid JobID) (string, error) {
+	return stepHoldTagImpl(jobid.String())
+}
+
+func stepHoldTagImpl(jobid string) (string, error) {
+	t := fmt.Sprintf("zrepl_STEP_J_%s", jobid)
+	if err := zfs.ValidHoldTag(t); err != nil {
+		return "", err
+	}
+	return t, nil
+}
+
+// err != nil always means that the bookmark is not a step bookmark
+func ParseStepHoldTag(tag string) (JobID, error) {
+	match := stepHoldTagRE.FindStringSubmatch(tag)
+	if match == nil {
+		return JobID{}, fmt.Errorf("parse hold tag: match regex %q", stepHoldTagRE)
+	}
+	jobID, err := MakeJobID(match[1])
+	if err != nil {
+		return JobID{}, errors.Wrap(err, "parse hold tag: invalid job id field")
+	}
+	return jobID, nil
+}
+
+const stepBookmarkNamePrefix = "zrepl_STEP"
+
+// v must be validated by caller
+func StepBookmarkName(fs string, guid uint64, id JobID) (string, error) {
+	return stepBookmarkNameImpl(fs, guid, id.String())
+}
+
+func stepBookmarkNameImpl(fs string, guid uint64, jobid string) (string, error) {
+	return makeJobAndGuidBookmarkName(stepBookmarkNamePrefix, fs, guid, jobid)
+}
+
+// name is the full bookmark name, including dataset path
+//
+// err != nil always means that the bookmark is not a step bookmark
+func ParseStepBookmarkName(fullname string) (guid uint64, jobID JobID, err error) {
+	guid, jobID, err = parseJobAndGuidBookmarkName(fullname, stepBookmarkNamePrefix)
+	if err != nil {
+		err = errors.Wrap(err, "parse step bookmark name") // no shadow!
+	}
+	return guid, jobID, err
+}
+
+const replicationCursorBookmarkNamePrefix = "zrepl_CURSOR"
+
+func ReplicationCursorBookmarkName(fs string, guid uint64, id JobID) (string, error) {
+	return replicationCursorBookmarkNameImpl(fs, guid, id.String())
+}
+
+func replicationCursorBookmarkNameImpl(fs string, guid uint64, jobid string) (string, error) {
+	return makeJobAndGuidBookmarkName(replicationCursorBookmarkNamePrefix, fs, guid, jobid)
+}
+
+var ErrV1ReplicationCursor = fmt.Errorf("bookmark name is a v1-replication cursor")
+
+//err != nil always means that the bookmark is not a valid replication bookmark
+//
+// Returns ErrV1ReplicationCursor as error if the bookmark is a v1 replication cursor
+func ParseReplicationCursorBookmarkName(fullname string) (uint64, JobID, error) {
+
+	// check for legacy cursors
+	{
+		if err := zfs.EntityNamecheck(fullname, zfs.EntityTypeBookmark); err != nil {
+			return 0, JobID{}, errors.Wrap(err, "parse replication cursor bookmark name")
+		}
+		_, _, name, err := zfs.DecomposeVersionString(fullname)
+		if err != nil {
+			return 0, JobID{}, errors.Wrap(err, "parse replication cursor bookmark name: decompose version string")
+		}
+		const V1ReplicationCursorBookmarkName = "zrepl_replication_cursor"
+		if name == V1ReplicationCursorBookmarkName {
+			return 0, JobID{}, ErrV1ReplicationCursor
+		}
+	}
+
+	guid, jobID, err := parseJobAndGuidBookmarkName(fullname, replicationCursorBookmarkNamePrefix)
+	if err != nil {
+		err = errors.Wrap(err, "parse replication cursor bookmark name") // no shadow
+	}
+	return guid, jobID, err
+}
+
+// may return nil for both values, indicating there is no cursor
+func GetMostRecentReplicationCursorOfJob(ctx context.Context, fs string, jobID JobID) (*zfs.FilesystemVersion, error) {
+	fsp, err := zfs.NewDatasetPath(fs)
+	if err != nil {
+		return nil, err
+	}
+	candidates, err := GetReplicationCursors(ctx, fsp, jobID)
+	if err != nil || len(candidates) == 0 {
+		return nil, err
+	}
+
+	sort.Slice(candidates, func(i, j int) bool {
+		return candidates[i].CreateTXG < candidates[j].CreateTXG
+	})
+
+	mostRecent := candidates[len(candidates)-1]
+	return &mostRecent, nil
+}
+
+func GetReplicationCursors(ctx context.Context, fs *zfs.DatasetPath, jobID JobID) ([]zfs.FilesystemVersion, error) {
+
+	listOut := &ListHoldsAndBookmarksOutput{}
+	if err := listZFSHoldsAndBookmarksImplFS(ctx, listOut, fs); err != nil {
+		return nil, errors.Wrap(err, "get replication cursor: list bookmarks and holds")
+	}
+
+	if len(listOut.V1ReplicationCursors) > 0 {
+		getLogger(ctx).WithField("bookmark", pretty.Sprint(listOut.V1ReplicationCursors)).
+			Warn("found v1-replication cursor bookmarks, consider running migration 'replication-cursor:v1-v2' after successful replication with this zrepl version")
+	}
+
+	candidates := make([]zfs.FilesystemVersion, 0)
+	for _, v := range listOut.ReplicationCursorBookmarks {
+		zv := zfs.ZFSSendArgVersion{
+			RelName: "#" + v.Name,
+			GUID:    v.Guid,
+		}
+		if err := zv.ValidateExists(ctx, v.FS); err != nil {
+			getLogger(ctx).WithError(err).WithField("bookmark", zv.FullPath(v.FS)).
+				Error("found invalid replication cursor bookmark")
+			continue
+		}
+		candidates = append(candidates, v.v)
+	}
+
+	return candidates, nil
+}
+
+// `target` is validated before replication cursor is set. if validation fails, the cursor is not moved.
+//
+// returns ErrBookmarkCloningNotSupported if version is a bookmark and bookmarking bookmarks is not supported by ZFS
+func MoveReplicationCursor(ctx context.Context, fs string, target *zfs.ZFSSendArgVersion, jobID JobID) (destroyedCursors []zfs.FilesystemVersion, err error) {
+
+	if !target.IsSnapshot() {
+		return nil, zfs.ErrBookmarkCloningNotSupported
+	}
+
+	snapProps, err := target.ValidateExistsAndGetCheckedProps(ctx, fs)
+	if err != nil {
+		return nil, errors.Wrapf(err, "invalid replication cursor target %q (guid=%v)", target.RelName, target.GUID)
+	}
+
+	bookmarkname, err := ReplicationCursorBookmarkName(fs, snapProps.Guid, jobID)
+	if err != nil {
+		return nil, errors.Wrap(err, "determine replication cursor name")
+	}
+
+	// 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(fs, *target, bookmarkname)
+	if err != nil {
+		if err == zfs.ErrBookmarkCloningNotSupported {
+			return nil, err // TODO go1.13 use wrapping
+		}
+		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
+}
+
+func DestroyObsoleteReplicationCursors(ctx context.Context, fs string, target *zfs.ZFSSendArgVersion, jobID JobID) (destroyed []zfs.FilesystemVersion, err error) {
+	return destroyBookmarksOlderThan(ctx, fs, target, jobID, func(shortname string) (accept bool) {
+		_, parsedID, err := ParseReplicationCursorBookmarkName(fs + "#" + shortname)
+		return err == nil && parsedID == jobID
+	})
+}
+
+// idempotently hold / step-bookmark `version`
+//
+// returns ErrBookmarkCloningNotSupported if version is a bookmark and bookmarking bookmarks is not supported by ZFS
+func HoldStep(ctx context.Context, fs string, v *zfs.ZFSSendArgVersion, jobID JobID) error {
+	if err := v.ValidateExists(ctx, fs); err != nil {
+		return err
+	}
+	if v.IsSnapshot() {
+
+		tag, err := StepHoldTag(jobID)
+		if err != nil {
+			return errors.Wrap(err, "step hold tag")
+		}
+
+		if err := zfs.ZFSHold(ctx, fs, *v, tag); err != nil {
+			return errors.Wrap(err, "step hold: zfs")
+		}
+
+		return nil
+	}
+
+	v.MustBeBookmark()
+
+	bmname, err := StepBookmarkName(fs, v.GUID, jobID)
+	if err != nil {
+		return errors.Wrap(err, "create step bookmark: determine bookmark name")
+	}
+	// idempotently create bookmark
+	err = zfs.ZFSBookmark(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
+			// 		however, the replication algorithm prefers snapshots anyways, so this quest
+			// 		is most likely not going to be successful. Also, there's the possibility that
+			//      the caller might want to filter what snapshots are eligibile, and this would
+			//      complicate things even further.
+			return err // TODO go1.13 use wrapping
+		}
+		return errors.Wrap(err, "create step bookmark: zfs")
+	}
+	return nil
+}
+
+// idempotently release the step-hold on v if v is a snapshot
+// or idempotently destroy the step-bookmark  of v if v is a bookmark
+//
+// note that this operation leaves v itself untouched, unless v is the step-bookmark itself, in which case v is destroyed
+//
+// returns an instance of *zfs.DatasetDoesNotExist if `v` does not exist
+func ReleaseStep(ctx context.Context, fs string, v *zfs.ZFSSendArgVersion, jobID JobID) error {
+
+	if err := v.ValidateExists(ctx, fs); err != nil {
+		return err
+	}
+
+	if v.IsSnapshot() {
+		tag, err := StepHoldTag(jobID)
+		if err != nil {
+			return errors.Wrap(err, "step release tag")
+		}
+
+		if err := zfs.ZFSRelease(ctx, tag, v.FullPath(fs)); err != nil {
+			return errors.Wrap(err, "step release: zfs")
+		}
+
+		return nil
+	}
+
+	v.MustBeBookmark()
+
+	bmname, err := StepBookmarkName(fs, v.GUID, jobID)
+	if err != nil {
+		return errors.Wrap(err, "step release: determine bookmark name")
+	}
+	// idempotently destroy bookmark
+
+	if err := zfs.ZFSDestroyIdempotent(bmname); err != nil {
+		return errors.Wrap(err, "step release: bookmark destroy: zfs")
+	}
+
+	return nil
+}
+
+// release {step holds, step bookmarks} earlier and including `mostRecent`
+func ReleaseStepAll(ctx context.Context, fs string, mostRecent *zfs.ZFSSendArgVersion, jobID JobID) error {
+
+	if err := mostRecent.ValidateInMemory(fs); err != nil {
+		return err
+	}
+
+	tag, err := StepHoldTag(jobID)
+	if err != nil {
+		return errors.Wrap(err, "step release all: tag")
+	}
+
+	err = zfs.ZFSReleaseAllOlderAndIncludingGUID(ctx, fs, mostRecent.GUID, tag)
+	if err != nil {
+		return errors.Wrapf(err, "step release all: release holds older and including %q", mostRecent.FullPath(fs))
+	}
+
+	_, err = destroyBookmarksOlderThan(ctx, fs, mostRecent, jobID, func(shortname string) bool {
+		_, parsedId, parseErr := ParseStepBookmarkName(fs + "#" + shortname)
+		return parseErr == nil && parsedId == jobID
+	})
+	if err != nil {
+		return errors.Wrapf(err, "step release all: destroy bookmarks older than %q", mostRecent.FullPath(fs))
+	}
+
+	return nil
+}
+
+var lastReceivedHoldTagRE = regexp.MustCompile("^zrepl_last_received_J_(.+)$")
+
+// err != nil always means that the bookmark is not a step bookmark
+func ParseLastReceivedHoldTag(tag string) (JobID, error) {
+	match := lastReceivedHoldTagRE.FindStringSubmatch(tag)
+	if match == nil {
+		return JobID{}, errors.Errorf("parse last-received-hold tag: does not match regex %s", lastReceivedHoldTagRE.String())
+	}
+	jobId, err := MakeJobID(match[1])
+	if err != nil {
+		return JobID{}, errors.Wrap(err, "parse last-received-hold tag: invalid job id field")
+	}
+	return jobId, nil
+}
+
+func LastReceivedHoldTag(jobID JobID) (string, error) {
+	return lastReceivedHoldImpl(jobID.String())
+}
+
+func lastReceivedHoldImpl(jobid string) (string, error) {
+	tag := fmt.Sprintf("zrepl_last_received_J_%s", jobid)
+	if err := zfs.ValidHoldTag(tag); err != nil {
+		return "", err
+	}
+	return tag, nil
+}
+
+func MoveLastReceivedHold(ctx context.Context, fs string, to zfs.ZFSSendArgVersion, jobID JobID) error {
+	if err := to.ValidateExists(ctx, fs); err != nil {
+		return err
+	}
+	if err := zfs.EntityNamecheck(to.FullPath(fs), zfs.EntityTypeSnapshot); err != nil {
+		return err
+	}
+
+	tag, err := LastReceivedHoldTag(jobID)
+	if err != nil {
+		return errors.Wrap(err, "last-received-hold: hold tag")
+	}
+
+	// we never want to be without a hold
+	// => hold new one before releasing old hold
+
+	err = zfs.ZFSHold(ctx, fs, to, tag)
+	if err != nil {
+		return errors.Wrap(err, "last-received-hold: hold newly received")
+	}
+
+	err = zfs.ZFSReleaseAllOlderThanGUID(ctx, fs, to.GUID, tag)
+	if err != nil {
+		return errors.Wrap(err, "last-received-hold: release older holds")
+	}
+
+	return nil
+}
+
+type ListHoldsAndBookmarksOutputBookmarkV1ReplicationCursor struct {
+	FS   string
+	Name string
+}
+
+type ListHoldsAndBookmarksOutput struct {
+	StepBookmarks []*ListHoldsAndBookmarksOutputBookmark
+	StepHolds     []*ListHoldsAndBookmarksOutputHold
+
+	ReplicationCursorBookmarks []*ListHoldsAndBookmarksOutputBookmark
+	V1ReplicationCursors       []*ListHoldsAndBookmarksOutputBookmarkV1ReplicationCursor
+	LastReceivedHolds          []*ListHoldsAndBookmarksOutputHold
+}
+
+type ListHoldsAndBookmarksOutputBookmark struct {
+	FS, Name string
+	Guid     uint64
+	JobID    JobID
+	v        zfs.FilesystemVersion
+}
+
+type ListHoldsAndBookmarksOutputHold struct {
+	FS            string
+	Snap          string
+	SnapGuid      uint64
+	SnapCreateTXG uint64
+	Tag           string
+	JobID         JobID
+}
+
+// List all holds and bookmarks managed by endpoint
+func ListZFSHoldsAndBookmarks(ctx context.Context, fsfilter zfs.DatasetFilter) (*ListHoldsAndBookmarksOutput, error) {
+
+	// initialize all fields so that JSON serializion of output looks pretty (see client/holds.go)
+	// however, listZFSHoldsAndBookmarksImplFS shouldn't rely on it
+	out := &ListHoldsAndBookmarksOutput{
+		StepBookmarks:              make([]*ListHoldsAndBookmarksOutputBookmark, 0),
+		StepHolds:                  make([]*ListHoldsAndBookmarksOutputHold, 0),
+		ReplicationCursorBookmarks: make([]*ListHoldsAndBookmarksOutputBookmark, 0),
+		V1ReplicationCursors:       make([]*ListHoldsAndBookmarksOutputBookmarkV1ReplicationCursor, 0),
+		LastReceivedHolds:          make([]*ListHoldsAndBookmarksOutputHold, 0),
+	}
+
+	fss, err := zfs.ZFSListMapping(ctx, fsfilter)
+	if err != nil {
+		return nil, errors.Wrap(err, "list filesystems")
+	}
+
+	for _, fs := range fss {
+		err := listZFSHoldsAndBookmarksImplFS(ctx, out, fs)
+		if err != nil {
+			return nil, errors.Wrapf(err, "list holds and bookmarks on %q", fs.ToString())
+		}
+	}
+	return out, nil
+}
+
+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)
+	}
+	for _, v := range fsvs {
+		switch v.Type {
+		case zfs.Bookmark:
+			listZFSHoldsAndBookmarksImplTryParseBookmark(ctx, out, fs, v)
+		case zfs.Snapshot:
+			holds, err := zfs.ZFSHolds(ctx, fs.ToString(), v.Name)
+			if err != nil {
+				return errors.Wrapf(err, "get holds of %q", v.ToAbsPath(fs))
+			}
+			for _, tag := range holds {
+				listZFSHoldsAndBookmarksImplSnapshotTryParseHold(ctx, out, fs, v, tag)
+			}
+		default:
+			continue
+		}
+	}
+	return nil
+}
+
+// pure function, err != nil always indicates parsing error
+func listZFSHoldsAndBookmarksImplTryParseBookmark(ctx context.Context, out *ListHoldsAndBookmarksOutput, fs *zfs.DatasetPath, v zfs.FilesystemVersion) {
+	var err error
+
+	if v.Type != zfs.Bookmark {
+		panic("impl error")
+	}
+
+	fullname := v.ToAbsPath(fs)
+
+	bm := &ListHoldsAndBookmarksOutputBookmark{
+		FS: fs.ToString(), Name: v.Name, v: v,
+	}
+	bm.Guid, bm.JobID, err = ParseStepBookmarkName(fullname)
+	if err == nil {
+		out.StepBookmarks = append(out.StepBookmarks, bm)
+		return
+	}
+
+	bm.Guid, bm.JobID, err = ParseReplicationCursorBookmarkName(fullname)
+	if err == nil {
+		out.ReplicationCursorBookmarks = append(out.ReplicationCursorBookmarks, bm)
+		return
+	} else if err == ErrV1ReplicationCursor {
+		v1rc := &ListHoldsAndBookmarksOutputBookmarkV1ReplicationCursor{
+			FS: fs.ToString(), Name: v.Name,
+		}
+		out.V1ReplicationCursors = append(out.V1ReplicationCursors, v1rc)
+		return
+	}
+}
+
+// pure function, err != nil always indicates parsing error
+func listZFSHoldsAndBookmarksImplSnapshotTryParseHold(ctx context.Context, out *ListHoldsAndBookmarksOutput, fs *zfs.DatasetPath, v zfs.FilesystemVersion, holdTag string) {
+	var err error
+
+	if v.Type != zfs.Snapshot {
+		panic("impl error")
+	}
+
+	hold := &ListHoldsAndBookmarksOutputHold{
+		FS:            fs.ToString(),
+		Snap:          v.Name,
+		SnapGuid:      v.Guid,
+		SnapCreateTXG: v.CreateTXG,
+		Tag:           holdTag,
+	}
+	hold.JobID, err = ParseStepHoldTag(holdTag)
+	if err == nil {
+		out.StepHolds = append(out.StepHolds, hold)
+		return
+	}
+
+	hold.JobID, err = ParseLastReceivedHoldTag(holdTag)
+	if err == nil {
+		out.LastReceivedHolds = append(out.LastReceivedHolds, hold)
+		return
+	}
+
+}

endpoint/endpoint_zfs_helpers.go

@@ -0,0 +1,109 @@
+package endpoint
+
+import (
+	"context"
+	"fmt"
+	"regexp"
+	"strconv"
+
+	"github.com/pkg/errors"
+
+	"github.com/zrepl/zrepl/zfs"
+)
+
+// returns the short name (no fs# prefix)
+func makeJobAndGuidBookmarkName(prefix string, fs string, guid uint64, jobid string) (string, error) {
+	bmname := fmt.Sprintf(prefix+"_G_%016x_J_%s", guid, jobid)
+	if err := zfs.EntityNamecheck(fmt.Sprintf("%s#%s", fs, bmname), zfs.EntityTypeBookmark); err != nil {
+		return "", err
+	}
+	return bmname, nil
+}
+
+var jobAndGuidBookmarkRE = regexp.MustCompile(`(.+)_G_([0-9a-f]{16})_J_(.+)$`)
+
+func parseJobAndGuidBookmarkName(fullname string, prefix string) (guid uint64, jobID JobID, _ error) {
+
+	if len(prefix) == 0 {
+		panic("prefix must not be empty")
+	}
+
+	if err := zfs.EntityNamecheck(fullname, zfs.EntityTypeBookmark); err != nil {
+		return 0, JobID{}, err
+	}
+
+	_, _, name, err := zfs.DecomposeVersionString(fullname)
+	if err != nil {
+		return 0, JobID{}, errors.Wrap(err, "decompose bookmark name")
+	}
+
+	match := jobAndGuidBookmarkRE.FindStringSubmatch(name)
+	if match == nil {
+		return 0, JobID{}, errors.Errorf("bookmark name does not match regex %q", jobAndGuidBookmarkRE.String())
+	}
+	if match[1] != prefix {
+		return 0, JobID{}, errors.Errorf("prefix component does not match: expected %q, got %q", prefix, match[1])
+	}
+
+	guid, err = strconv.ParseUint(match[2], 16, 64)
+	if err != nil {
+		return 0, JobID{}, errors.Wrapf(err, "parse guid component: %q", match[2])
+	}
+
+	jobID, err = MakeJobID(match[3])
+	if err != nil {
+		return 0, JobID{}, errors.Wrapf(err, "parse jobid component: %q", match[3])
+	}
+
+	return guid, jobID, nil
+}
+
+func destroyBookmarksOlderThan(ctx context.Context, fs string, mostRecent *zfs.ZFSSendArgVersion, jobID JobID, filter func(shortname string) (accept bool)) (destroyed []zfs.FilesystemVersion, err error) {
+	if filter == nil {
+		panic(filter)
+	}
+
+	fsp, err := zfs.NewDatasetPath(fs)
+	if err != nil {
+		return nil, errors.Wrap(err, "invalid filesystem path")
+	}
+
+	mostRecentProps, err := mostRecent.ValidateExistsAndGetCheckedProps(ctx, fs)
+	if err != nil {
+		return nil, errors.Wrap(err, "validate most recent version argument")
+	}
+
+	stepBookmarks, err := zfs.ZFSListFilesystemVersions(fsp, zfs.FilterFromClosure(
+		func(t zfs.VersionType, name string) (accept bool, err error) {
+			if t != zfs.Bookmark {
+				return false, nil
+			}
+			return filter(name), nil
+		}))
+	if err != nil {
+		return nil, errors.Wrap(err, "list bookmarks")
+	}
+
+	// cut off all bookmarks prior to mostRecent's CreateTXG
+	var destroy []zfs.FilesystemVersion
+	for _, v := range stepBookmarks {
+		if v.Type != zfs.Bookmark {
+			panic("implementation error")
+		}
+		if !filter(v.Name) {
+			panic("inconsistent filter result")
+		}
+		if v.CreateTXG < mostRecentProps.CreateTXG {
+			destroy = append(destroy, v)
+		}
+	}
+
+	// FIXME use batch destroy, must adopt code to handle bookmarks
+	for _, v := range destroy {
+		if err := zfs.ZFSDestroyIdempotent(v.ToAbsPath(fsp)); err != nil {
+			return nil, errors.Wrap(err, "destroy bookmark")
+		}
+	}
+
+	return destroy, nil
+}

endpoint/endpoint_zfs_helpers_test.go

@@ -0,0 +1,51 @@
+package endpoint
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestParseJobAndGuidBookmarkName(t *testing.T) {
+
+	type Case struct {
+		input     string
+		expectErr bool
+
+		guid  uint64
+		jobid string
+	}
+
+	cases := []Case{
+		{
+			`p1/sync#zrepl_CURSOR_G_932f3a7089080ce2_J_push with legitimate name`,
+			false, 0x932f3a7089080ce2, "push with legitimate name",
+		},
+		{
+			input:     `p1/sync#zrepl_CURSOR_G_932f3a7089_J_push with legitimate name`,
+			expectErr: true,
+		},
+		{
+			input:     `p1/sync#zrepl_CURSOR_G_932f3a7089080ce2_J_push with il\tlegitimate name`,
+			expectErr: true,
+		},
+		{
+			input:     `p1/sync#otherprefix_G_932f3a7089080ce2_J_push with legitimate name`,
+			expectErr: true,
+		},
+	}
+
+	for i := range cases {
+		t.Run(cases[i].input, func(t *testing.T) {
+			guid, jobid, err := parseJobAndGuidBookmarkName(cases[i].input, replicationCursorBookmarkNamePrefix)
+			if cases[i].expectErr {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, cases[i].guid, guid)
+				assert.Equal(t, MustMakeJobID(cases[i].jobid), jobid)
+			}
+		})
+	}
+
+}

endpoint/jobid.go

@@ -0,0 +1,78 @@
+package endpoint
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/pkg/errors"
+
+	"github.com/zrepl/zrepl/zfs"
+)
+
+// An instance of this type returned by MakeJobID guarantees
+// that that instance's JobID.String() can be used in a ZFS dataset name and hold tag.
+type JobID struct {
+	jid string
+}
+
+func MakeJobID(s string) (JobID, error) {
+	if len(s) == 0 {
+		return JobID{}, fmt.Errorf("must not be empty string")
+	}
+
+	if err := zfs.ComponentNamecheck(s); err != nil {
+		return JobID{}, errors.Wrap(err, "muse be usable as a dataset path component")
+	}
+
+	if _, err := stepBookmarkNameImpl("pool/ds", 0xface601d, s); err != nil {
+		// note that this might still fail due to total maximum name length, but we can't enforce that
+		return JobID{}, errors.Wrap(err, "must be usable for a step bookmark")
+	}
+
+	if _, err := stepHoldTagImpl(s); err != nil {
+		return JobID{}, errors.Wrap(err, "must be usable for a step hold tag")
+	}
+
+	if _, err := lastReceivedHoldImpl(s); err != nil {
+		return JobID{}, errors.Wrap(err, "must be usabel as a last-recieved-hold tag")
+	}
+
+	// FIXME replication cursor bookmark name
+
+	_, err := zfs.NewDatasetPath(s)
+	if err != nil {
+		return JobID{}, fmt.Errorf("must be usable in a ZFS dataset path: %s", err)
+	}
+
+	return JobID{s}, nil
+}
+
+func MustMakeJobID(s string) JobID {
+	jid, err := MakeJobID(s)
+	if err != nil {
+		panic(err)
+	}
+	return jid
+}
+
+func (j JobID) expectInitialized() {
+	if j.jid == "" {
+		panic("use of unitialized JobID")
+	}
+}
+
+func (j JobID) String() string {
+	j.expectInitialized()
+	return j.jid
+}
+
+var _ json.Marshaler = JobID{}
+var _ json.Unmarshaler = (*JobID)(nil)
+
+func (j JobID) MarshalJSON() ([]byte, error) { return json.Marshal(j.jid) }
+
+func (j *JobID) UnmarshalJSON(b []byte) error {
+	return json.Unmarshal(b, &j.jid)
+}
+
+func (j JobID) MustValidate() { j.expectInitialized() }