WIP endpoint abstractions + pruning integration / pruner rewrite

daemon/pruner.v2/fsstate_enumer.go

@@ -0,0 +1,72 @@
+// Code generated by "enumer -type=FSState -json"; DO NOT EDIT.
+
+//
+package pruner
+
+import (
+	"encoding/json"
+	"fmt"
+)
+
+const _FSStateName = "FSStateInitializedFSStatePlanningFSStatePlanErrFSStateExecutingFSStateExecuteErrFSStateExecuteSuccess"
+
+var _FSStateIndex = [...]uint8{0, 18, 33, 47, 63, 80, 101}
+
+func (i FSState) String() string {
+	if i < 0 || i >= FSState(len(_FSStateIndex)-1) {
+		return fmt.Sprintf("FSState(%d)", i)
+	}
+	return _FSStateName[_FSStateIndex[i]:_FSStateIndex[i+1]]
+}
+
+var _FSStateValues = []FSState{0, 1, 2, 3, 4, 5}
+
+var _FSStateNameToValueMap = map[string]FSState{
+	_FSStateName[0:18]:   0,
+	_FSStateName[18:33]:  1,
+	_FSStateName[33:47]:  2,
+	_FSStateName[47:63]:  3,
+	_FSStateName[63:80]:  4,
+	_FSStateName[80:101]: 5,
+}
+
+// FSStateString retrieves an enum value from the enum constants string name.
+// Throws an error if the param is not part of the enum.
+func FSStateString(s string) (FSState, error) {
+	if val, ok := _FSStateNameToValueMap[s]; ok {
+		return val, nil
+	}
+	return 0, fmt.Errorf("%s does not belong to FSState values", s)
+}
+
+// FSStateValues returns all values of the enum
+func FSStateValues() []FSState {
+	return _FSStateValues
+}
+
+// IsAFSState returns "true" if the value is listed in the enum definition. "false" otherwise
+func (i FSState) IsAFSState() bool {
+	for _, v := range _FSStateValues {
+		if i == v {
+			return true
+		}
+	}
+	return false
+}
+
+// MarshalJSON implements the json.Marshaler interface for FSState
+func (i FSState) MarshalJSON() ([]byte, error) {
+	return json.Marshal(i.String())
+}
+
+// UnmarshalJSON implements the json.Unmarshaler interface for FSState
+func (i *FSState) UnmarshalJSON(data []byte) error {
+	var s string
+	if err := json.Unmarshal(data, &s); err != nil {
+		return fmt.Errorf("FSState should be a string, got %s", data)
+	}
+
+	var err error
+	*i, err = FSStateString(s)
+	return err
+}

daemon/pruner.v2/pruner.go

@@ -0,0 +1,454 @@
+package pruner
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+
+	"github.com/pkg/errors"
+	"github.com/zrepl/zrepl/daemon/logging"
+	"github.com/zrepl/zrepl/daemon/logging/trace"
+	"github.com/zrepl/zrepl/endpoint"
+	"github.com/zrepl/zrepl/pruning"
+	"github.com/zrepl/zrepl/zfs"
+)
+
+type Pruner struct {
+	fsfilter  endpoint.FSFilter
+	jid       endpoint.JobID
+	side      Side
+	keepRules []pruning.KeepRule
+
+	// all channels consumed by the run loop
+	reportReqs chan reportRequest
+	stopReqs   chan stopRequest
+	done       chan struct{}
+	fsListRes  chan fsListRes
+
+	state State
+
+	listFilesystemsError error       // only in state StateListFilesystemsError
+	fsPruners            []*FSPruner // only in state StateFanOutFilesystems
+}
+
+//go:generate enumer -type=State -json
+type State int
+
+const (
+	StateInitialized State = iota
+	StateListFilesystems
+	StateListFilesystemsError
+	StateFanOutFilesystems
+	StateDone
+)
+
+type Report struct {
+	State                State
+	ListFilesystemsError error       // only valid in StateListFilesystemsError
+	Filesystems          []*FSReport // valid from StateFanOutFilesystems
+}
+
+type reportRequest struct {
+	ctx   context.Context
+	reply chan *Report
+}
+
+type runRequest struct {
+	complete chan struct{}
+}
+
+type stopRequest struct {
+	complete chan struct{}
+}
+
+type fsListRes struct {
+	filesystems []*zfs.DatasetPath
+	err         error
+}
+
+type Side interface {
+	// may return both nil, indicating there is no replication position
+	GetReplicationPosition(ctx context.Context, fs string) (*zfs.FilesystemVersion, error)
+	isSide() Side
+}
+
+func NewPruner(fsfilter endpoint.FSFilter, jid endpoint.JobID, side Side, keepRules []pruning.KeepRule) *Pruner {
+	return &Pruner{
+		fsfilter,
+		jid,
+		side,
+		keepRules,
+		make(chan reportRequest),
+		make(chan stopRequest),
+		make(chan struct{}),
+		make(chan fsListRes),
+		StateInitialized,
+		nil,
+		nil,
+	}
+}
+
+func (p *Pruner) Run(ctx context.Context) *Report {
+
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	if p.state != StateInitialized {
+		panic("Run can onl[y be called once")
+	}
+
+	go func() {
+		fss, err := zfs.ZFSListMapping(ctx, p.fsfilter)
+		p.fsListRes <- fsListRes{fss, err}
+	}()
+
+	for {
+		select {
+		case res := <-p.fsListRes:
+			if res.err != nil {
+				p.state = StateListFilesystemsError
+				p.listFilesystemsError = res.err
+				close(p.done)
+				continue
+			}
+
+			p.state = StateFanOutFilesystems
+
+			p.fsPruners = make([]*FSPruner, len(res.filesystems))
+			_, add, end := trace.WithTaskGroup(ctx, "pruner-fan-out-fs")
+			for i, fs := range res.filesystems {
+				p.fsPruners[i] = NewFSPruner(p.jid, p.side, p.keepRules, fs)
+				add(func(ctx context.Context) {
+					p.fsPruners[i].Run(ctx)
+				})
+			}
+			go func() {
+				end()
+				close(p.done)
+			}()
+
+		case req := <-p.stopReqs:
+			cancel()
+			go func() {
+				<-p.done
+				close(req.complete)
+			}()
+		case req := <-p.reportReqs:
+			req.reply <- p.report(req.ctx)
+		case <-p.done:
+			p.state = StateDone
+			return p.report(ctx)
+		}
+	}
+}
+
+func (p *Pruner) Report(ctx context.Context) *Report {
+	req := reportRequest{
+		ctx:   ctx,
+		reply: make(chan *Report, 1),
+	}
+	select {
+	case p.reportReqs <- req:
+		return <-req.reply
+	case <-ctx.Done():
+		return nil
+	case <-p.done:
+		return nil
+	}
+}
+
+func (p *Pruner) report(ctx context.Context) *Report {
+	fsreports := make([]*FSReport, len(p.fsPruners))
+	for i := range fsreports {
+		fsreports[i] = p.fsPruners[i].report()
+	}
+	return &Report{
+		State:                p.state,
+		ListFilesystemsError: p.listFilesystemsError,
+		Filesystems:          fsreports,
+	}
+}
+
+// implements pruning.Snapshot
+type snapshot struct {
+	replicated bool
+	stepHolds  []pruning.StepHold
+	zfs.FilesystemVersion
+
+	state     SnapState
+	destroyOp *zfs.DestroySnapOp
+}
+
+//go:generate enumer -type=SnapState -json
+type SnapState int
+
+const (
+	SnapStateInitialized SnapState = iota
+	SnapStateKeeping
+	SnapStateDeletePending
+	SnapStateDeleteAttempted
+)
+
+// implements pruning.StepHold
+type stepHold struct {
+	endpoint.Abstraction
+}
+
+func (s snapshot) Replicated() bool              { return s.replicated }
+func (s snapshot) StepHolds() []pruning.StepHold { return s.stepHolds }
+
+func (s stepHold) GetJobID() endpoint.JobID { return *s.Abstraction.GetJobID() }
+
+type FSPruner struct {
+	jid       endpoint.JobID
+	side      Side
+	keepRules []pruning.KeepRule
+	fsp       *zfs.DatasetPath
+
+	state FSState
+
+	// all channels consumed by the run loop
+	planned    chan fsPlanRes
+	executed   chan fsExecuteRes
+	done       chan struct{}
+	reportReqs chan fsReportReq
+
+	keepList    []*snapshot // valid in FSStateExecuting and forward
+	destroyList []*snapshot // valid in FSStateExecuting and forward, field .destroyOp is invalid until FSStateExecuting is left
+
+}
+
+type fsPlanRes struct {
+	keepList    []*snapshot
+	destroyList []*snapshot
+	err         error
+}
+
+type fsExecuteRes struct {
+	completedDestroyOps []*zfs.DestroySnapOp // same len() as FSPruner.destroyList
+}
+
+type fsReportReq struct {
+	res chan *FSReport
+}
+
+type FSReport struct {
+	State    FSState
+	KeepList []*SnapReport
+	Destroy  []*SnapReport
+}
+
+type SnapReport struct {
+	State         SnapState
+	Name          string
+	Replicated    bool
+	StepHoldCount int
+	DestroyError  error
+}
+
+//go:generate enumer -type=FSState -json
+type FSState int
+
+const (
+	FSStateInitialized FSState = iota
+	FSStatePlanning
+	FSStatePlanErr
+	FSStateExecuting
+	FSStateExecuteErr
+	FSStateExecuteSuccess
+)
+
+func (s FSState) IsTerminal() bool {
+	return s == FSStatePlanErr || s == FSStateExecuteErr || s == FSStateExecuteSuccess
+}
+
+func NewFSPruner(jid endpoint.JobID, side Side, keepRules []pruning.KeepRule, fsp *zfs.DatasetPath) *FSPruner {
+	return &FSPruner{
+		jid, side, keepRules, fsp,
+		FSStateInitialized,
+		make(chan fsPlanRes),
+		make(chan fsExecuteRes),
+		make(chan struct{}),
+		make(chan fsReportReq),
+		nil, nil,
+	}
+}
+
+func (p *FSPruner) Run(ctx context.Context) *FSReport {
+
+	defer func() {
+	}()
+
+	p.state = FSStatePlanning
+
+	go func() { p.planned <- p.plan(ctx) }()
+
+out:
+	for !p.state.IsTerminal() {
+		select {
+		case res := <-p.planned:
+
+			if res.err != nil {
+				p.state = FSStatePlanErr
+				continue
+			}
+			p.state = FSStateExecuting
+			p.keepList = res.keepList
+			p.destroyList = res.destroyList
+
+			go func() { p.executed <- p.execute(ctx, p.destroyList) }()
+
+		case res := <-p.executed:
+
+			if len(res.completedDestroyOps) != len(p.destroyList) {
+				panic("impl error: completedDestroyOps is a vector corresponding to entries in p.destroyList")
+			}
+
+			var erronous []*zfs.DestroySnapOp
+			for i, op := range res.completedDestroyOps {
+				if *op.ErrOut != nil {
+					erronous = append(erronous, op)
+				}
+				p.destroyList[i].destroyOp = op
+				p.destroyList[i].state = SnapStateDeleteAttempted
+			}
+			if len(erronous) > 0 {
+				p.state = FSStateExecuteErr
+			} else {
+				p.state = FSStateExecuteSuccess
+			}
+
+			close(p.done)
+
+		case <-p.reportReqs:
+			panic("unimp")
+		case <-p.done:
+			break out
+		}
+	}
+
+	// TODO render last FS report
+	return nil
+}
+
+func (p *FSPruner) plan(ctx context.Context) fsPlanRes {
+	fs := p.fsp.ToString()
+	vs, err := zfs.ZFSListFilesystemVersions(ctx, p.fsp, zfs.ListFilesystemVersionsOptions{})
+	if err != nil {
+		return fsPlanRes{err: errors.Wrap(err, "list filesystem versions")}
+	}
+
+	allJobsStepHolds, absErrs, err := endpoint.ListAbstractions(ctx, endpoint.ListZFSHoldsAndBookmarksQuery{
+		FS: endpoint.ListZFSHoldsAndBookmarksQueryFilesystemFilter{
+			FS: &fs,
+		},
+		What: endpoint.AbstractionTypeSet{
+			endpoint.AbstractionStepHold: true,
+		},
+		Concurrency: 1,
+	})
+	if err != nil {
+		return fsPlanRes{err: errors.Wrap(err, "list abstractions")}
+	}
+	if len(absErrs) > 0 {
+		logging.GetLogger(ctx, logging.SubsysPruning).WithError(endpoint.ListAbstractionsErrors(absErrs)).
+			Error("error listing some step holds, prune attempt might fail with 'dataset is busy' errors")
+	}
+
+	repPos, err := p.side.GetReplicationPosition(ctx, p.fsp.ToString())
+	if err != nil {
+		return fsPlanRes{err: errors.Wrap(err, "get replication position")}
+	}
+
+	vsAsSnaps := make([]pruning.Snapshot, len(vs))
+	for i := range vs {
+		var repPosCreateTxgOrZero uint64
+		if repPos != nil {
+			repPosCreateTxgOrZero = repPos.GetCreateTXG()
+		}
+		s := &snapshot{
+			state:             SnapStateInitialized,
+			FilesystemVersion: vs[i],
+			replicated:        vs[i].GetCreateTXG() <= repPosCreateTxgOrZero,
+		}
+		for _, h := range allJobsStepHolds {
+			if zfs.FilesystemVersionEqualIdentity(vs[i], h.GetFilesystemVersion()) {
+				s.stepHolds = append(s.stepHolds, stepHold{h})
+			}
+		}
+		vsAsSnaps[i] = s
+	}
+
+	downcastToSnapshots := func(l []pruning.Snapshot) (r []*snapshot) {
+		r = make([]*snapshot, len(l))
+		for i, e := range l {
+			r[i] = e.(*snapshot)
+		}
+		return r
+	}
+	pruningResult := pruning.PruneSnapshots(vsAsSnaps, p.keepRules)
+	remove, keep := downcastToSnapshots(pruningResult.Remove), downcastToSnapshots(pruningResult.Keep)
+	if len(remove)+len(keep) != len(vsAsSnaps) {
+		for _, s := range vsAsSnaps {
+			r, _ := json.MarshalIndent(s.(*snapshot).report(), "", "  ")
+			fmt.Fprintf(os.Stderr, "%s\n", string(r))
+		}
+		panic("indecisive")
+	}
+
+	for _, s := range remove {
+		s.state = SnapStateDeletePending
+	}
+	for _, s := range keep {
+		s.state = SnapStateKeeping
+	}
+
+	return fsPlanRes{keepList: keep, destroyList: remove, err: nil}
+}
+
+func (p *FSPruner) execute(ctx context.Context, destroyList []*snapshot) fsExecuteRes {
+	ops := make([]*zfs.DestroySnapOp, len(destroyList))
+	for i, fsv := range p.destroyList {
+		ops[i] = &zfs.DestroySnapOp{
+			Filesystem: p.fsp.ToString(),
+			Name:       fsv.GetName(),
+			ErrOut:     new(error),
+		}
+	}
+	zfs.ZFSDestroyFilesystemVersions(ctx, ops)
+
+	return fsExecuteRes{completedDestroyOps: ops}
+}
+
+func (p *FSPruner) report() *FSReport {
+	return &FSReport{
+		State:    p.state,
+		KeepList: p.reportRenderSnapReports(p.keepList),
+		Destroy:  p.reportRenderSnapReports(p.destroyList),
+	}
+}
+
+func (p *FSPruner) reportRenderSnapReports(l []*snapshot) (r []*SnapReport) {
+	r = make([]*SnapReport, len(l))
+	for i := range l {
+		r[i] = l[i].report()
+	}
+	return r
+}
+
+func (s *snapshot) report() *SnapReport {
+	var snapErr error
+	if s.state == SnapStateDeleteAttempted {
+		if *s.destroyOp.ErrOut != nil {
+			snapErr = (*s.destroyOp.ErrOut)
+		}
+	}
+	return &SnapReport{
+		State:         s.state,
+		Name:          s.Name,
+		Replicated:    s.Replicated(),
+		StepHoldCount: len(s.stepHolds),
+		DestroyError:  snapErr,
+	}
+}

daemon/pruner.v2/pruner_side.go

@@ -0,0 +1,27 @@
+package pruner
+
+import (
+	"context"
+
+	"github.com/zrepl/zrepl/endpoint"
+	"github.com/zrepl/zrepl/zfs"
+)
+
+type SideSender struct {
+	jobID endpoint.JobID
+}
+
+func NewSideSender(jid endpoint.JobID) *SideSender {
+	return &SideSender{jid}
+}
+
+func (s *SideSender) isSide() Side { return nil }
+
+var _ Side = (*SideSender)(nil)
+
+func (s *SideSender) GetReplicationPosition(ctx context.Context, fs string) (*zfs.FilesystemVersion, error) {
+	if fs == "" {
+		panic("must not pass zero value for fs")
+	}
+	return endpoint.GetMostRecentReplicationCursorOfJob(ctx, fs, s.jobID)
+}

daemon/pruner.v2/snapstate_enumer.go

@@ -0,0 +1,70 @@
+// Code generated by "enumer -type=SnapState -json"; DO NOT EDIT.
+
+//
+package pruner
+
+import (
+	"encoding/json"
+	"fmt"
+)
+
+const _SnapStateName = "SnapStateInitializedSnapStateKeepingSnapStateDeletePendingSnapStateDeleteAttempted"
+
+var _SnapStateIndex = [...]uint8{0, 20, 36, 58, 82}
+
+func (i SnapState) String() string {
+	if i < 0 || i >= SnapState(len(_SnapStateIndex)-1) {
+		return fmt.Sprintf("SnapState(%d)", i)
+	}
+	return _SnapStateName[_SnapStateIndex[i]:_SnapStateIndex[i+1]]
+}
+
+var _SnapStateValues = []SnapState{0, 1, 2, 3}
+
+var _SnapStateNameToValueMap = map[string]SnapState{
+	_SnapStateName[0:20]:  0,
+	_SnapStateName[20:36]: 1,
+	_SnapStateName[36:58]: 2,
+	_SnapStateName[58:82]: 3,
+}
+
+// SnapStateString retrieves an enum value from the enum constants string name.
+// Throws an error if the param is not part of the enum.
+func SnapStateString(s string) (SnapState, error) {
+	if val, ok := _SnapStateNameToValueMap[s]; ok {
+		return val, nil
+	}
+	return 0, fmt.Errorf("%s does not belong to SnapState values", s)
+}
+
+// SnapStateValues returns all values of the enum
+func SnapStateValues() []SnapState {
+	return _SnapStateValues
+}
+
+// IsASnapState returns "true" if the value is listed in the enum definition. "false" otherwise
+func (i SnapState) IsASnapState() bool {
+	for _, v := range _SnapStateValues {
+		if i == v {
+			return true
+		}
+	}
+	return false
+}
+
+// MarshalJSON implements the json.Marshaler interface for SnapState
+func (i SnapState) MarshalJSON() ([]byte, error) {
+	return json.Marshal(i.String())
+}
+
+// UnmarshalJSON implements the json.Unmarshaler interface for SnapState
+func (i *SnapState) UnmarshalJSON(data []byte) error {
+	var s string
+	if err := json.Unmarshal(data, &s); err != nil {
+		return fmt.Errorf("SnapState should be a string, got %s", data)
+	}
+
+	var err error
+	*i, err = SnapStateString(s)
+	return err
+}

daemon/pruner.v2/state_enumer.go

@@ -0,0 +1,71 @@
+// Code generated by "enumer -type=State -json"; DO NOT EDIT.
+
+//
+package pruner
+
+import (
+	"encoding/json"
+	"fmt"
+)
+
+const _StateName = "StateInitializedStateListFilesystemsStateListFilesystemsErrorStateFanOutFilesystemsStateDone"
+
+var _StateIndex = [...]uint8{0, 16, 36, 61, 83, 92}
+
+func (i State) String() string {
+	if i < 0 || i >= State(len(_StateIndex)-1) {
+		return fmt.Sprintf("State(%d)", i)
+	}
+	return _StateName[_StateIndex[i]:_StateIndex[i+1]]
+}
+
+var _StateValues = []State{0, 1, 2, 3, 4}
+
+var _StateNameToValueMap = map[string]State{
+	_StateName[0:16]:  0,
+	_StateName[16:36]: 1,
+	_StateName[36:61]: 2,
+	_StateName[61:83]: 3,
+	_StateName[83:92]: 4,
+}
+
+// StateString retrieves an enum value from the enum constants string name.
+// Throws an error if the param is not part of the enum.
+func StateString(s string) (State, error) {
+	if val, ok := _StateNameToValueMap[s]; ok {
+		return val, nil
+	}
+	return 0, fmt.Errorf("%s does not belong to State values", s)
+}
+
+// StateValues returns all values of the enum
+func StateValues() []State {
+	return _StateValues
+}
+
+// IsAState returns "true" if the value is listed in the enum definition. "false" otherwise
+func (i State) IsAState() bool {
+	for _, v := range _StateValues {
+		if i == v {
+			return true
+		}
+	}
+	return false
+}
+
+// MarshalJSON implements the json.Marshaler interface for State
+func (i State) MarshalJSON() ([]byte, error) {
+	return json.Marshal(i.String())
+}
+
+// UnmarshalJSON implements the json.Unmarshaler interface for State
+func (i *State) UnmarshalJSON(data []byte) error {
+	var s string
+	if err := json.Unmarshal(data, &s); err != nil {
+		return fmt.Errorf("State should be a string, got %s", data)
+	}
+
+	var err error
+	*i, err = StateString(s)
+	return err
+}

daemon/pruner/pruner.go

@@ -20,15 +20,14 @@ import (
 )
 
 // Try to keep it compatible with github.com/zrepl/zrepl/endpoint.Endpoint
-type History interface {
-	ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error)
+type Endpoint interface {
 	ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error)
+	ListFilesystemVersions(ctx context.Context, req *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error)
 }
 
 // Try to keep it compatible with github.com/zrepl/zrepl/endpoint.Endpoint
 type Target interface {
-	ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error)
-	ListFilesystemVersions(ctx context.Context, req *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error)
+	Endpoint
 	DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshotsReq) (*pdu.DestroySnapshotsRes, error)
 }
 
@@ -46,13 +45,14 @@ func GetLogger(ctx context.Context) Logger {
 }
 
 type args struct {
-	ctx                            context.Context
-	target                         Target
-	receiver                       History
-	rules                          []pruning.KeepRule
-	retryWait                      time.Duration
-	considerSnapAtCursorReplicated bool
-	promPruneSecs                  prometheus.Observer
+	ctx                              context.Context
+	target                           Target
+	sender, receiver                 Endpoint
+	rules                            []pruning.KeepRule
+	retryWait                        time.Duration
+	considerSnapAtCursorReplicated   bool
+	convertAnyStepHoldToStepBookmark bool
+	promPruneSecs                    prometheus.Observer
 }
 
 type Pruner struct {
@@ -70,11 +70,12 @@ type Pruner struct {
 }
 
 type PrunerFactory struct {
-	senderRules                    []pruning.KeepRule
-	receiverRules                  []pruning.KeepRule
-	retryWait                      time.Duration
-	considerSnapAtCursorReplicated bool
-	promPruneSecs                  *prometheus.HistogramVec
+	senderRules                      []pruning.KeepRule
+	receiverRules                    []pruning.KeepRule
+	retryWait                        time.Duration
+	considerSnapAtCursorReplicated   bool
+	convertAnyStepHoldToStepBookmark bool
+	promPruneSecs                    *prometheus.HistogramVec
 }
 
 type LocalPrunerFactory struct {
@@ -122,25 +123,35 @@ func NewPrunerFactory(in config.PruningSenderReceiver, promPruneSecs *prometheus
 		}
 		considerSnapAtCursorReplicated = considerSnapAtCursorReplicated || !knr.KeepSnapshotAtCursor
 	}
+
+	convertAnyStepHoldToStepBookmark := false
+	for _, r := range in.KeepSender {
+		_, ok := r.Ret.(*config.PruneKeepStepHolds)
+		convertAnyStepHoldToStepBookmark = convertAnyStepHoldToStepBookmark || ok
+	}
+
 	f := &PrunerFactory{
-		senderRules:                    keepRulesSender,
-		receiverRules:                  keepRulesReceiver,
-		retryWait:                      envconst.Duration("ZREPL_PRUNER_RETRY_INTERVAL", 10*time.Second),
-		considerSnapAtCursorReplicated: considerSnapAtCursorReplicated,
-		promPruneSecs:                  promPruneSecs,
+		senderRules:                      keepRulesSender,
+		receiverRules:                    keepRulesReceiver,
+		retryWait:                        envconst.Duration("ZREPL_PRUNER_RETRY_INTERVAL", 10*time.Second),
+		considerSnapAtCursorReplicated:   considerSnapAtCursorReplicated,
+		convertAnyStepHoldToStepBookmark: convertAnyStepHoldToStepBookmark,
+		promPruneSecs:                    promPruneSecs,
 	}
 	return f, nil
 }
 
-func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, target Target, receiver History) *Pruner {
+func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, sender Target, receiver Endpoint) *Pruner {
 	p := &Pruner{
 		args: args{
 			context.WithValue(ctx, contextKeyPruneSide, "sender"),
-			target,
+			sender,
+			sender,
 			receiver,
 			f.senderRules,
 			f.retryWait,
 			f.considerSnapAtCursorReplicated,
+			f.convertAnyStepHoldToStepBookmark,
 			f.promPruneSecs.WithLabelValues("sender"),
 		},
 		state: Plan,
@@ -148,15 +159,17 @@ func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, target Target, re
 	return p
 }
 
-func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target, receiver History) *Pruner {
+func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, receiver Target, sender Endpoint) *Pruner {
 	p := &Pruner{
 		args: args{
 			context.WithValue(ctx, contextKeyPruneSide, "receiver"),
-			target,
+			receiver,
+			sender,
 			receiver,
 			f.receiverRules,
 			f.retryWait,
 			false, // senseless here anyways
+			false, // senseless here anyways
 			f.promPruneSecs.WithLabelValues("receiver"),
 		},
 		state: Plan,
@@ -164,15 +177,17 @@ func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target,
 	return p
 }
 
-func (f *LocalPrunerFactory) BuildLocalPruner(ctx context.Context, target Target, receiver History) *Pruner {
+func (f *LocalPrunerFactory) BuildLocalPruner(ctx context.Context, target Target) *Pruner {
 	p := &Pruner{
 		args: args{
 			context.WithValue(ctx, contextKeyPruneSide, "local"),
 			target,
-			receiver,
+			target,
+			target,
 			f.keepRules,
 			f.retryWait,
 			false, // considerSnapAtCursorReplicated is not relevant for local pruning
+			false, // convertAnyStepHoldToStepBookmark is not relevant for local pruning
 			f.promPruneSecs.WithLabelValues("local"),
 		},
 		state: Plan,
@@ -341,11 +356,13 @@ func (s snapshot) Replicated() bool { return s.replicated }
 
 func (s snapshot) Date() time.Time { return s.date }
 
+func (s snapshot) CreateTXG() uint64 { return s.fsv.GetCreateTXG() }
+
 func doOneAttempt(a *args, u updater) {
 
-	ctx, target, receiver := a.ctx, a.target, a.receiver
+	ctx, sender, receiver, target := a.ctx, a.sender, a.receiver, a.target
 
-	sfssres, err := receiver.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
+	sfssres, err := sender.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
 	if err != nil {
 		u(func(p *Pruner) {
 			p.state = PlanErr
@@ -407,6 +424,10 @@ tfss_loop:
 
 		pfs.snaps = make([]pruning.Snapshot, 0, len(tfsvs))
 
+		receiver.ListFilesystemVersions(ctx, &pdu.ListFilesystemVersionsReq{
+			Filesystem: tfs.Path,
+		})
+
 		rcReq := &pdu.ReplicationCursorReq{
 			Filesystem: tfs.Path,
 		}
@@ -415,6 +436,7 @@ tfss_loop:
 			pfsPlanErrAndLog(err, "cannot get replication cursor bookmark")
 			continue tfss_loop
 		}
+
 		if rc.GetNotexist() {
 			err := errors.New("replication cursor bookmark does not exist (one successful replication is required before pruning works)")
 			pfsPlanErrAndLog(err, "")