func (f subroot) Filter(p *zfs.DatasetPath) (pass bool, err error) {
	return p.HasPrefix(f.localRoot) && !p.Equal(f.localRoot), nil
}
Filters local p
package endpoint
import (
	"bytes"
	"context"
	"fmt"
	"io"
	"path"
	"github.com/kr/pretty"
	"github.com/pkg/errors"
	"github.com/zrepl/zrepl/daemon/logging/trace"
	"github.com/zrepl/zrepl/replication/logic/pdu"
	"github.com/zrepl/zrepl/util/bandwidthlimit"
	"github.com/zrepl/zrepl/util/chainedio"
	"github.com/zrepl/zrepl/util/chainlock"
	"github.com/zrepl/zrepl/util/envconst"
	"github.com/zrepl/zrepl/util/nodefault"
	"github.com/zrepl/zrepl/zfs"
	zfsprop "github.com/zrepl/zrepl/zfs/property"
)
type SenderConfig struct {
	FSF   zfs.DatasetFilter
	JobID JobID
	Encrypt              *nodefault.Bool
	SendRaw              bool
	SendProperties       bool
	SendBackupProperties bool
	SendLargeBlocks      bool
	SendCompressed       bool
	SendEmbeddedData     bool
	SendSaved            bool
	BandwidthLimit bandwidthlimit.Config
}
func (c *SenderConfig) Validate() error {
	c.JobID.MustValidate()
	if err := c.Encrypt.ValidateNoDefault(); 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)
	}
	if err := bandwidthlimit.ValidateConfig(c.BandwidthLimit); err != nil {
		return errors.Wrap(err, "`BandwidthLimit` field invalid")
	}
	return nil
}
type Sender struct {
	pdu.UnsafeReplicationServer  
	FSFilter zfs.DatasetFilter
	jobId    JobID
	config   SenderConfig
	bwLimit  bandwidthlimit.Wrapper
}
func NewSender(conf SenderConfig) *Sender {
	if err := conf.Validate(); err != nil {
		panic("invalid config" + err.Error())
	}
	ratelimiter := bandwidthlimit.WrapperFromConfig(conf.BandwidthLimit)
	return &Sender{
		FSFilter: conf.FSF,
		jobId:    conf.JobID,
		config:   conf,
		bwLimit:  ratelimiter,
	}
}
func (s *Sender) filterCheckFS(fs string) (*zfs.DatasetPath, error) {
	dp, err := zfs.NewDatasetPath(fs)
	if err != nil {
		return nil, err
	}
	if dp.Length() == 0 {
		return nil, errors.New("empty filesystem not allowed")
	}
	pass, err := s.FSFilter.Filter(dp)
	if err != nil {
		return nil, err
	}
	if !pass {
		return nil, fmt.Errorf("endpoint does not allow access to filesystem %s", fs)
	}
	return dp, nil
}
func (s *Sender) ListFilesystems(ctx context.Context, r *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error) {
	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
	fss, err := zfs.ZFSListMapping(ctx, s.FSFilter)
	if err != nil {
		return nil, err
	}
	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(),
			IsPlaceholder: false,  
			IsEncrypted:   encEnabled,
		}
	}
	res := &pdu.ListFilesystemRes{Filesystems: rfss}
	return res, nil
}
func (s *Sender) ListFilesystemVersions(ctx context.Context, r *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error) {
	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
	lp, err := s.filterCheckFS(r.GetFilesystem())
	if err != nil {
		return nil, err
	}
	fsvs, err := zfs.ZFSListFilesystemVersions(ctx, lp, zfs.ListFilesystemVersionsOptions{})
	if err != nil {
		return nil, err
	}
	rfsvs := make([]*pdu.FilesystemVersion, len(fsvs))
	for i := range fsvs {
		rfsvs[i] = pdu.FilesystemVersionFromZFS(&fsvs[i])
	}
	res := &pdu.ListFilesystemVersionsRes{Versions: rfsvs}
	return res, nil
}
func uncheckedSendArgsFromPDU(fsv *pdu.FilesystemVersion) *zfs.ZFSSendArgVersion {
	if fsv == nil {
		return nil
	}
	return &zfs.ZFSSendArgVersion{RelName: fsv.GetRelName(), GUID: fsv.Guid}
}
func sendArgsFromPDUAndValidateExistsAndGetVersion(ctx context.Context, fs string, fsv *pdu.FilesystemVersion) (v zfs.FilesystemVersion, err error) {
	sendArgs := uncheckedSendArgsFromPDU(fsv)
	if sendArgs == nil {
		return v, errors.New("must not be nil")
	}
	version, err := sendArgs.ValidateExistsAndGetVersion(ctx, fs)
	if err != nil {
		return v, err
	}
	return version, nil
}
func (s *Sender) sendMakeArgs(ctx context.Context, r *pdu.SendReq) (sendArgs zfs.ZFSSendArgsValidated, _ error) {
	_, err := s.filterCheckFS(r.Filesystem)
	if err != nil {
		return sendArgs, err
	}
	switch r.Encrypted {
	case pdu.Tri_DontCare:
	case pdu.Tri_False:
		if s.config.Encrypt.B {
			return sendArgs, errors.New("only encrypted sends allowed (send -w + encryption!= off), but unencrypted send requested")
		}
	case pdu.Tri_True:
		if !s.config.Encrypt.B {
			return sendArgs, errors.New("only unencrypted sends allowed, but encrypted send requested")
		}
	default:
		return sendArgs, fmt.Errorf("unknown pdu.Tri variant %q", r.Encrypted)
	}
	sendArgsUnvalidated := zfs.ZFSSendArgsUnvalidated{
		FS:   r.Filesystem,
		From: uncheckedSendArgsFromPDU(r.GetFrom()),  
		To:   uncheckedSendArgsFromPDU(r.GetTo()),    
		ZFSSendFlags: zfs.ZFSSendFlags{
			ResumeToken:      r.ResumeToken,  
			Encrypted:        s.config.Encrypt,
			Properties:       s.config.SendProperties,
			BackupProperties: s.config.SendBackupProperties,
			Raw:              s.config.SendRaw,
			LargeBlocks:      s.config.SendLargeBlocks,
			Compressed:       s.config.SendCompressed,
			EmbeddedData:     s.config.SendEmbeddedData,
			Saved:            s.config.SendSaved,
		},
	}
	sendArgs, err = sendArgsUnvalidated.Validate(ctx)
	if err != nil {
		return sendArgs, errors.Wrap(err, "validate send arguments")
	}
	return sendArgs, nil
}
func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
	sendArgs, err := s.sendMakeArgs(ctx, r)
	if err != nil {
		return nil, nil, err
	}
	replicationGuaranteeOptions, err := replicationGuaranteeOptionsFromPDU(r.GetReplicationConfig().Protection)
	if err != nil {
		return nil, nil, err
	}
	replicationGuaranteeStrategy := replicationGuaranteeOptions.Strategy(sendArgs.From != nil)
	liveAbs, err := replicationGuaranteeStrategy.SenderPreSend(ctx, s.jobId, &sendArgs)
	if err != nil {
		return nil, nil, err
	}
	for _, a := range liveAbs {
		if a != nil {
			abstractionsCacheSingleton.Put(a)
		}
	}
	func() {
		ctx, endSpan := trace.WithSpan(ctx, "cleanup-stale-abstractions")
		defer endSpan()
		keep := func(a Abstraction) (keep bool) {
			keep = false
			for _, k := range liveAbs {
				keep = keep || AbstractionEquals(a, k)
			}
			return keep
		}
		check := func(obsoleteAbs []Abstraction) {
			mustLiveVersions := []zfs.FilesystemVersion{sendArgs.ToVersion}
			if sendArgs.FromVersion != nil {
				mustLiveVersions = append(mustLiveVersions, *sendArgs.FromVersion)
			}
			for _, staleVersion := range obsoleteAbs {
				for _, mustLiveVersion := range mustLiveVersions {
					isSendArg := zfs.FilesystemVersionEqualIdentity(mustLiveVersion, staleVersion.GetFilesystemVersion())
					stepHoldBasedGuaranteeStrategy := false
					k := replicationGuaranteeStrategy.Kind()
					switch k {
					case ReplicationGuaranteeKindResumability:
						stepHoldBasedGuaranteeStrategy = true
					case ReplicationGuaranteeKindIncremental:
					case ReplicationGuaranteeKindNone:
					default:
						panic(fmt.Sprintf("this is supposed to be an exhaustive match, got %v", k))
					}
					isSnapshot := mustLiveVersion.IsSnapshot()
					if isSendArg && (!isSnapshot || stepHoldBasedGuaranteeStrategy) {
						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)))
					}
				}
			}
		}
		destroyTypes := AbstractionTypeSet{
			AbstractionStepHold:                           true,
			AbstractionTentativeReplicationCursorBookmark: true,
		}
		abstractionsCacheSingleton.TryBatchDestroy(ctx, s.jobId, sendArgs.FS, destroyTypes, keep, check)
	}()
	var sendStream io.ReadCloser
	sendStream, err = zfs.ZFSSend(ctx, sendArgs)
	if err != nil {
		return nil, nil, errors.Wrap(err, "zfs send failed")
	}
	sendStream = s.bwLimit.WrapReadCloser(sendStream)
	res := &pdu.SendRes{
		ExpectedSize:    0,
		UsedResumeToken: r.ResumeToken != "",
	}
	return res, sendStream, nil
}
func (s *Sender) SendDry(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, error) {
	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
	sendArgs, err := s.sendMakeArgs(ctx, r)
	if err != nil {
		return nil, err
	}
	si, err := zfs.ZFSSendDry(ctx, sendArgs)
	if err != nil {
		return nil, errors.Wrap(err, "zfs send dry failed")
	}
	res := &pdu.SendRes{
		ExpectedSize:    si.SizeEstimate,
		UsedResumeToken: r.ResumeToken != "",
	}
	return res, nil
}
func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
	orig := r.GetOriginalReq()  
	fsp, err := p.filterCheckFS(orig.GetFilesystem())
	if err != nil {
		return nil, err
	}
	fs := fsp.ToString()
	var from *zfs.FilesystemVersion
	if orig.GetFrom() != nil {
		f, err := sendArgsFromPDUAndValidateExistsAndGetVersion(ctx, fs, orig.GetFrom())  
		if err != nil {
			return nil, errors.Wrap(err, "validate `from` exists")
		}
		from = &f
	}
	to, err := sendArgsFromPDUAndValidateExistsAndGetVersion(ctx, fs, orig.GetTo())
	if err != nil {
		return nil, errors.Wrap(err, "validate `to` exists")
	}
	replicationGuaranteeOptions, err := replicationGuaranteeOptionsFromPDU(orig.GetReplicationConfig().Protection)
	if err != nil {
		return nil, err
	}
	liveAbs, err := replicationGuaranteeOptions.Strategy(from != nil).SenderPostRecvConfirmed(ctx, p.jobId, fs, to)
	if err != nil {
		return nil, err
	}
	for _, a := range liveAbs {
		if a != nil {
			abstractionsCacheSingleton.Put(a)
		}
	}
	keep := func(a Abstraction) (keep bool) {
		keep = false
		for _, k := range liveAbs {
			keep = keep || AbstractionEquals(a, k)
		}
		return keep
	}
	destroyTypes := AbstractionTypeSet{
		AbstractionStepHold:                           true,
		AbstractionTentativeReplicationCursorBookmark: true,
		AbstractionReplicationCursorBookmarkV2:        true,
	}
	abstractionsCacheSingleton.TryBatchDestroy(ctx, p.jobId, fs, destroyTypes, keep, nil)
	return &pdu.SendCompletedRes{}, nil
}
func (p *Sender) DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshotsReq) (*pdu.DestroySnapshotsRes, error) {
	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
	dp, err := p.filterCheckFS(req.Filesystem)
	if err != nil {
		return nil, err
	}
	return doDestroySnapshots(ctx, dp, req.Snapshots)
}
func (p *Sender) Ping(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, error) {
	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
	res := pdu.PingRes{
		Echo: req.GetMessage(),
	}
	return &res, nil
}
func (p *Sender) PingDataconn(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, error) {
	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
	return p.Ping(ctx, req)
}
func (p *Sender) WaitForConnectivity(ctx context.Context) error {
	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
	return nil
}
func (p *Sender) ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error) {
	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
	dp, err := p.filterCheckFS(req.Filesystem)
	if err != nil {
		return nil, err
	}
	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, _ io.ReadCloser) (*pdu.ReceiveRes, error) {
	return nil, fmt.Errorf("sender does not implement Receive()")
}
type FSFilter interface {  
	Filter(path *zfs.DatasetPath) (pass bool, err error)
}
type FSMap interface {  
	FSFilter
	Map(path *zfs.DatasetPath) (*zfs.DatasetPath, error)
	Invert() (FSMap, error)
	AsFilter() FSFilter
}
type ReceiverConfig struct {
	JobID JobID
	RootWithoutClientComponent *zfs.DatasetPath
	AppendClientIdentity       bool
	InheritProperties  []zfsprop.Property
	OverrideProperties map[zfsprop.Property]string
	BandwidthLimit bandwidthlimit.Config
}
func (c *ReceiverConfig) copyIn() {
	c.RootWithoutClientComponent = c.RootWithoutClientComponent.Copy()
	pInherit := make([]zfsprop.Property, len(c.InheritProperties))
	copy(pInherit, c.InheritProperties)
	c.InheritProperties = pInherit
	pOverride := make(map[zfsprop.Property]string, len(c.OverrideProperties))
	for key, value := range c.OverrideProperties {
		pOverride[key] = value
	}
	c.OverrideProperties = pOverride
}
func (c *ReceiverConfig) Validate() error {
	c.JobID.MustValidate()
	for _, prop := range c.InheritProperties {
		err := prop.Validate()
		if err != nil {
			return errors.Wrapf(err, "inherit property %q", prop)
		}
	}
	for prop := range c.OverrideProperties {
		err := prop.Validate()
		if err != nil {
			return errors.Wrapf(err, "override property %q", prop)
		}
	}
	if c.RootWithoutClientComponent.Length() <= 0 {
		return errors.New("RootWithoutClientComponent must not be an empty dataset path")
	}
	if err := bandwidthlimit.ValidateConfig(c.BandwidthLimit); err != nil {
		return errors.Wrap(err, "`BandwidthLimit` field invalid")
	}
	return nil
}
type Receiver struct {
	pdu.UnsafeReplicationServer  
	conf ReceiverConfig  
	bwLimit bandwidthlimit.Wrapper
	recvParentCreationMtx *chainlock.L
}
func NewReceiver(config ReceiverConfig) *Receiver {
	config.copyIn()
	if err := config.Validate(); err != nil {
		panic(err)
	}
	return &Receiver{
		conf:                  config,
		recvParentCreationMtx: chainlock.New(),
		bwLimit:               bandwidthlimit.WrapperFromConfig(config.BandwidthLimit),
	}
}
func TestClientIdentity(rootFS *zfs.DatasetPath, clientIdentity string) error {
	_, err := clientRoot(rootFS, clientIdentity)
	return err
}
func clientRoot(rootFS *zfs.DatasetPath, clientIdentity string) (*zfs.DatasetPath, error) {
	rootFSLen := rootFS.Length()
	clientRootStr := path.Join(rootFS.ToString(), clientIdentity)
	clientRoot, err := zfs.NewDatasetPath(clientRootStr)
	if err != nil {
		return nil, err
	}
	if rootFSLen+1 != clientRoot.Length() {
		return nil, fmt.Errorf("client identity must be a single ZFS filesystem path component")
	}
	return clientRoot, nil
}
func (s *Receiver) clientRootFromCtx(ctx context.Context) *zfs.DatasetPath {
	if !s.conf.AppendClientIdentity {
		return s.conf.RootWithoutClientComponent.Copy()
	}
	clientIdentity, ok := ctx.Value(ClientIdentityKey).(string)
	if !ok {
		panic("ClientIdentityKey context value must be set")
	}
	clientRoot, err := clientRoot(s.conf.RootWithoutClientComponent, clientIdentity)
	if err != nil {
		panic(fmt.Sprintf("ClientIdentityContextKey must have been validated before invoking Receiver: %s", err))
	}
	return clientRoot
}
type subroot struct {
	localRoot *zfs.DatasetPath
}
var _ zfs.DatasetFilter = subroot{}