func listStaleFiltering(abs []Abstraction, sinceBound *CreateTXGRangeBound) *StalenessInfo {
	var noJobId []Abstraction
	by := make(map[fsAjobAtype][]Abstraction)
	for _, a := range abs {
		if a.GetJobID() == nil {
			noJobId = append(noJobId, a)
			continue
		}
		faj := fsAjobAtype{fsAndJobId{a.GetFS(), *a.GetJobID()}, a.GetType()}
		l := by[faj]
		l = append(l, a)
		by[faj] = l
	}
	type stepFirstNotStaleCandidate struct {
		cursor *Abstraction
		step   *Abstraction
	}
	stepFirstNotStaleCandidates := make(map[fsAndJobId]stepFirstNotStaleCandidate)  
	for _, a := range abs {
		if a.GetJobID() == nil {
			continue  
		}
		key := fsAndJobId{a.GetFS(), *a.GetJobID()}
		c := stepFirstNotStaleCandidates[key]
		switch a.GetType() {
		case AbstractionReplicationCursorBookmarkV2:
			if c.cursor == nil || (*c.cursor).GetCreateTXG() < a.GetCreateTXG() {
				a := a
				c.cursor = &a
			}
		case AbstractionStepHold:
			if c.step == nil || (*c.step).GetCreateTXG() < a.GetCreateTXG() {
				a := a
				c.step = &a
			}
		default:
			continue  
		}
		stepFirstNotStaleCandidates[key] = c
	}
	ret := &StalenessInfo{
		Live:  noJobId,
		Stale: []Abstraction{},
	}
	for k := range by {
		l := by[k]
		if k.Type == AbstractionStepHold {
			var untilBound *CreateTXGRangeBound
			{
				sfnsc := stepFirstNotStaleCandidates[k.fsAndJobId]
				if sfnsc.cursor != nil {
					untilBound = &CreateTXGRangeBound{
						CreateTXG: (*sfnsc.cursor).GetCreateTXG(),
						Inclusive: &nodefault.Bool{B: true},
					}
				} else if sfnsc.step != nil {
					untilBound = &CreateTXGRangeBound{
						CreateTXG: (*sfnsc.step).GetCreateTXG(),
						Inclusive: &nodefault.Bool{B: false},
					}
				} else {
					untilBound = nil  
				}
			}
			staleRange := CreateTXGRange{
				Since: sinceBound,
				Until: untilBound,
			}
			for _, a := range l {
				if staleRange.Contains(a.GetCreateTXG()) {
					ret.Stale = append(ret.Stale, a)
				} else {
					ret.Live = append(ret.Live, a)
				}
			}
		} else if k.Type == AbstractionReplicationCursorBookmarkV2 || k.Type == AbstractionLastReceivedHold {
			sort.Slice(l, func(i, j int) bool {
				return l[i].GetCreateTXG() > l[j].GetCreateTXG()
			})
			if len(l) > 0 {
				ret.Live = append(ret.Live, l[0])
				ret.Stale = append(ret.Stale, l[1:]...)
			}
		} else {
			ret.Live = append(ret.Live, l...)
		}
	}
	return ret
}
For step holds and bookmarks, only those older than the most recent replication cursor
of their (filesystem,job) are considered because younger ones cannot be stale by definition
(if we destroy them, we might actually lose the hold on the `To` for an ongoing incremental replication)

For replication cursors and last-received-holds, only the most recent one is kept.

the returned StalenessInfo.ConstructedWithQuery is not set
package endpoint
import (
	"context"
	"encoding/json"
	"fmt"
	"math"
	"sort"
	"strings"
	"sync"
	"github.com/pkg/errors"
	"github.com/zrepl/zrepl/daemon/logging/trace"
	"github.com/zrepl/zrepl/util/envconst"
	"github.com/zrepl/zrepl/util/nodefault"
	"github.com/zrepl/zrepl/util/semaphore"
	"github.com/zrepl/zrepl/zfs"
)
type AbstractionType string
const (
	AbstractionStepHold                           AbstractionType = "step-hold"
	AbstractionLastReceivedHold                   AbstractionType = "last-received-hold"
	AbstractionTentativeReplicationCursorBookmark AbstractionType = "tentative-replication-cursor-bookmark-v2"
	AbstractionReplicationCursorBookmarkV1        AbstractionType = "replication-cursor-bookmark-v1"
	AbstractionReplicationCursorBookmarkV2        AbstractionType = "replication-cursor-bookmark-v2"
)
var AbstractionTypesAll = map[AbstractionType]bool{
	AbstractionStepHold:                           true,
	AbstractionLastReceivedHold:                   true,
	AbstractionTentativeReplicationCursorBookmark: true,
	AbstractionReplicationCursorBookmarkV1:        true,
	AbstractionReplicationCursorBookmarkV2:        true,
}
type Abstraction interface {
	GetType() AbstractionType
	GetFS() string
	GetName() string
	GetFullPath() string
	GetJobID() *JobID  
	GetCreateTXG() uint64
	GetFilesystemVersion() zfs.FilesystemVersion
	String() string
	Destroy(context.Context) error
	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 AbstractionStepHold:
		return nil
	case AbstractionLastReceivedHold:
		return nil
	case AbstractionTentativeReplicationCursorBookmark:
		return nil
	case AbstractionReplicationCursorBookmarkV1:
		return nil
	case AbstractionReplicationCursorBookmarkV2:
		return nil
	default:
		return errors.Errorf("unknown abstraction type %q", t)
	}
}
func (t AbstractionType) MustValidate() error {
	if err := t.Validate(); err != nil {
		panic(err)
	}
	return nil
}
type AbstractionJSON struct{ Abstraction }
var _ json.Marshaler = (*AbstractionJSON)(nil)
func (a AbstractionJSON) MarshalJSON() ([]byte, error) {
	type S struct {
		Type              AbstractionType
		FS                string
		Name              string
		FullPath          string
		JobID             *JobID  
		CreateTXG         uint64
		FilesystemVersion zfs.FilesystemVersion
		String            string
	}
	v := S{
		Type:              a.Abstraction.GetType(),
		FS:                a.Abstraction.GetFS(),
		Name:              a.Abstraction.GetName(),
		FullPath:          a.Abstraction.GetFullPath(),
		JobID:             a.Abstraction.GetJobID(),
		CreateTXG:         a.Abstraction.GetCreateTXG(),
		FilesystemVersion: a.Abstraction.GetFilesystemVersion(),
		String:            a.Abstraction.String(),
	}
	return json.Marshal(v)
}
type AbstractionTypeSet map[AbstractionType]bool
func AbstractionTypeSetFromStrings(sts []string) (AbstractionTypeSet, error) {
	ats := make(map[AbstractionType]bool, len(sts))
	for i, t := range sts {
		at := AbstractionType(t)
		if err := at.Validate(); err != nil {
			return nil, errors.Wrapf(err, "invalid abstraction type #%d %q", i+1, t)
		}
		ats[at] = true
	}
	return ats, nil
}
func (s AbstractionTypeSet) ContainsAll(q AbstractionTypeSet) bool {
	for k := range q {
		if _, ok := s[k]; !ok {
			return false
		}
	}
	return true
}
func (s AbstractionTypeSet) ContainsAnyOf(q AbstractionTypeSet) bool {
	for k := range q {
		if _, ok := s[k]; ok {
			return true
		}
	}
	return false
}
func (s AbstractionTypeSet) String() string {
	sts := make([]string, 0, len(s))
	for i := range s {
		sts = append(sts, string(i))
	}
	sts = sort.StringSlice(sts)
	return strings.Join(sts, ",")
}
func (s AbstractionTypeSet) Validate() error {
	for k := range s {
		if err := k.Validate(); err != nil {
			return err
		}
	}
	return nil
}
type BookmarkExtractor func(fs *zfs.DatasetPath, v zfs.FilesystemVersion) Abstraction
func (t AbstractionType) BookmarkExtractor() BookmarkExtractor {
	switch t {
	case AbstractionTentativeReplicationCursorBookmark:
		return TentativeReplicationCursorExtractor
	case AbstractionReplicationCursorBookmarkV1:
		return ReplicationCursorV1Extractor
	case AbstractionReplicationCursorBookmarkV2:
		return ReplicationCursorV2Extractor
	case AbstractionStepHold:
		return nil
	case AbstractionLastReceivedHold:
		return nil
	default:
		panic(fmt.Sprintf("unimpl: %q", t))
	}
}
type HoldExtractor = func(fs *zfs.DatasetPath, v zfs.FilesystemVersion, tag string) Abstraction
func (t AbstractionType) HoldExtractor() HoldExtractor {
	switch t {
	case AbstractionTentativeReplicationCursorBookmark:
		return nil
	case AbstractionReplicationCursorBookmarkV1:
		return nil
	case AbstractionReplicationCursorBookmarkV2:
		return nil
	case AbstractionStepHold:
		return StepHoldExtractor
	case AbstractionLastReceivedHold:
		return LastReceivedHoldExtractor
	default:
		panic(fmt.Sprintf("unimpl: %q", t))
	}
}
func (t AbstractionType) BookmarkNamer() func(fs string, guid uint64, jobId JobID) (string, error) {
	switch t {
	case AbstractionTentativeReplicationCursorBookmark:
		return TentativeReplicationCursorBookmarkName
	case AbstractionReplicationCursorBookmarkV1:
		panic("shouldn't be creating new ones")
	case AbstractionReplicationCursorBookmarkV2:
		return ReplicationCursorBookmarkName
	case AbstractionStepHold:
		return nil
	case AbstractionLastReceivedHold:
		return nil
	default:
		panic(fmt.Sprintf("unimpl: %q", t))
	}
}
type ListZFSHoldsAndBookmarksQuery struct {
	FS ListZFSHoldsAndBookmarksQueryFilesystemFilter
	What AbstractionTypeSet
	JobID *JobID
	CreateTXG CreateTXGRange
	Concurrency int64
}
type CreateTXGRangeBound struct {
	CreateTXG uint64
	Inclusive *nodefault.Bool  
}
type CreateTXGRange struct {
	Since *CreateTXGRangeBound
	Until *CreateTXGRangeBound
}
type ListZFSHoldsAndBookmarksQueryFilesystemFilter struct {
	FS     *string
	Filter zfs.DatasetFilter
}
func (q *ListZFSHoldsAndBookmarksQuery) Validate() error {
	if err := q.FS.Validate(); err != nil {
		return errors.Wrap(err, "FS")
	}
	if q.JobID != nil {
		q.JobID.MustValidate()  
	}
	if err := q.CreateTXG.Validate(); err != nil {
		return errors.Wrap(err, "CreateTXGRange")
	}
	if err := q.What.Validate(); err != nil {
		return err
	}
	if q.Concurrency < 1 {
		return errors.New("Concurrency must be >= 1")
	}
	return nil
}
var createTXGRangeBoundAllowCreateTXG0 = envconst.Bool("ZREPL_ENDPOINT_LIST_ABSTRACTIONS_QUERY_CREATETXG_RANGE_BOUND_ALLOW_0", false)
func (i *CreateTXGRangeBound) Validate() error {
	if err := i.Inclusive.ValidateNoDefault(); err != nil {
		return errors.Wrap(err, "Inclusive")
	}
	if i.CreateTXG == 0 && !createTXGRangeBoundAllowCreateTXG0 {
		return errors.New("CreateTXG must be non-zero")
	}
	return nil
}
func (f *ListZFSHoldsAndBookmarksQueryFilesystemFilter) Validate() error {
	if f == nil {
		return nil
	}
	fsSet := f.FS != nil
	filterSet := f.Filter != nil
	if fsSet && filterSet || !fsSet && !filterSet {
		return fmt.Errorf("must set FS or Filter field, but fsIsSet=%v and filterIsSet=%v", fsSet, filterSet)
	}
	if fsSet {
		if err := zfs.EntityNamecheck(*f.FS, zfs.EntityTypeFilesystem); err != nil {
			return errors.Wrap(err, "FS invalid")
		}
	}
	return nil
}
func (f *ListZFSHoldsAndBookmarksQueryFilesystemFilter) Filesystems(ctx context.Context) ([]string, error) {
	if err := f.Validate(); err != nil {
		panic(err)
	}
	if f.FS != nil {
		return []string{*f.FS}, nil
	}
	if f.Filter != nil {
		dps, err := zfs.ZFSListMapping(ctx, f.Filter)
		if err != nil {
			return nil, err
		}
		fss := make([]string, len(dps))
		for i, dp := range dps {
			fss[i] = dp.ToString()
		}
		return fss, nil
	}
	panic("unreachable")
}
func (r *CreateTXGRange) Validate() error {
	if r.Since != nil {
		if err := r.Since.Validate(); err != nil {
			return errors.Wrap(err, "Since")
		}
	}
	if r.Until != nil {
		if err := r.Until.Validate(); err != nil {
			return errors.Wrap(err, "Until")
		}
	}
	if _, err := r.effectiveBounds(); err != nil {
		return errors.Wrapf(err, "specified range %s is semantically invalid", r)
	}
	return nil
}
type effectiveBounds struct {
	sinceInclusive uint64
	sinceUnbounded bool
	untilInclusive uint64
	untilUnbounded bool
}
func (r *CreateTXGRange) effectiveBounds() (bounds effectiveBounds, err error) {
	bounds.sinceUnbounded = r.Since == nil
	bounds.untilUnbounded = r.Until == nil
	if r.Since == nil && r.Until == nil {
		return bounds, nil
	}
	if r.Since != nil {
		bounds.sinceInclusive = r.Since.CreateTXG
		if !r.Since.Inclusive.B {
			if r.Since.CreateTXG == math.MaxUint64 {
				return bounds, errors.Errorf("Since-exclusive (%v) must be less than math.MaxUint64 (%v)",
					r.Since.CreateTXG, uint64(math.MaxUint64))
			}
			bounds.sinceInclusive++
		}
	}
	if r.Until != nil {
		bounds.untilInclusive = r.Until.CreateTXG
		if !r.Until.Inclusive.B {
			if r.Until.CreateTXG == 0 {
				return bounds, errors.Errorf("Until-exclusive (%v) must be greater than 0", r.Until.CreateTXG)
			}
			bounds.untilInclusive--
		}
	}
	if !bounds.sinceUnbounded && !bounds.untilUnbounded {
		if bounds.sinceInclusive >= bounds.untilInclusive {
			return bounds, errors.Errorf("effective range bounds are [%v,%v] which is empty or invalid", bounds.sinceInclusive, bounds.untilInclusive)
		}
	}
	return bounds, nil
}
func (r *CreateTXGRange) String() string {
	var buf strings.Builder
	if r.Since == nil {
		fmt.Fprintf(&buf, "~")
	} else {
		if err := r.Since.Inclusive.ValidateNoDefault(); err != nil {
			fmt.Fprintf(&buf, "?")
		} else if r.Since.Inclusive.B {
			fmt.Fprintf(&buf, "[")
		} else {
			fmt.Fprintf(&buf, "(")
		}
		fmt.Fprintf(&buf, "%d", r.Since.CreateTXG)
	}
	fmt.Fprintf(&buf, ",")
	if r.Until == nil {
		fmt.Fprintf(&buf, "~")
	} else {
		fmt.Fprintf(&buf, "%d", r.Until.CreateTXG)
		if err := r.Until.Inclusive.ValidateNoDefault(); err != nil {
			fmt.Fprintf(&buf, "?")
		} else if r.Until.Inclusive.B {
			fmt.Fprintf(&buf, "]")
		} else {
			fmt.Fprintf(&buf, ")")
		}
	}
	return buf.String()
}
func (r *CreateTXGRange) IsUnbounded() bool {
	if err := r.Validate(); err != nil {
		panic(err)
	}
	bounds, err := r.effectiveBounds()
	if err != nil {
		panic(err)
	}
	return bounds.sinceUnbounded && bounds.untilUnbounded
}
func (r *CreateTXGRange) Contains(qCreateTxg uint64) bool {
	if err := r.Validate(); err != nil {
		panic(err)
	}
	bounds, err := r.effectiveBounds()
	if err != nil {
		panic(err)
	}
	sinceMatches := bounds.sinceUnbounded || bounds.sinceInclusive <= qCreateTxg
	untilMatches := bounds.untilUnbounded || qCreateTxg <= bounds.untilInclusive
	return sinceMatches && untilMatches
}
type ListAbstractionsError struct {
	FS   string
	Snap string
	What string
	Err  error
}
func (e ListAbstractionsError) Error() string {
	if e.FS == "" {
		return fmt.Sprintf("list endpoint abstractions: %s: %s", e.What, e.Err)
	} else {
		v := e.FS
		if e.Snap != "" {
			v = fmt.Sprintf("%s@%s", e.FS, e.Snap)
		}
		return fmt.Sprintf("list endpoint abstractions on %q: %s: %s", v, e.What, e.Err)
	}
}
type putListAbstractionErr func(err error, fs string, what string)
type putListAbstraction func(a Abstraction)
type ListAbstractionsErrors []ListAbstractionsError
func (e ListAbstractionsErrors) Error() string {
	if len(e) == 0 {
		panic(e)
	}
	if len(e) == 1 {
		return fmt.Sprintf("list endpoint abstractions: %s", e[0])
	}
	msgs := make([]string, len(e))
	for i := range e {
		msgs[i] = e[i].Error()
	}
	return fmt.Sprintf("list endpoint abstractions: multiple errors:\n%s", strings.Join(msgs, "\n"))
}
func ListAbstractions(ctx context.Context, query ListZFSHoldsAndBookmarksQuery) (out []Abstraction, outErrs []ListAbstractionsError, err error) {
	outChan, outErrsChan, err := ListAbstractionsStreamed(ctx, query)
	if err != nil {
		return nil, nil, err
	}
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		for a := range outChan {
			out = append(out, a)
		}
	}()
	wg.Add(1)
	go func() {
		defer wg.Done()
		for err := range outErrsChan {
			outErrs = append(outErrs, err)
		}
	}()
	wg.Wait()
	return out, outErrs, nil
}
func ListAbstractionsStreamed(ctx context.Context, query ListZFSHoldsAndBookmarksQuery) (<-chan Abstraction, <-chan ListAbstractionsError, error) {
	if err := query.Validate(); err != nil {
		return nil, nil, errors.Wrap(err, "validate query")
	}
	fss, err := query.FS.Filesystems(ctx)
	if err != nil {
		return nil, nil, errors.Wrap(err, "list filesystems")
	}
	outErrs := make(chan ListAbstractionsError)
	out := make(chan Abstraction)
	errCb := func(err error, fs string, what string) {
		outErrs <- ListAbstractionsError{Err: err, FS: fs, What: what}
	}
	emitAbstraction := func(a Abstraction) {
		jobIdMatches := query.JobID == nil || a.GetJobID() == nil || *a.GetJobID() == *query.JobID
		createTXGMatches := query.CreateTXG.Contains(a.GetCreateTXG())
		if jobIdMatches && createTXGMatches {
			out <- a
		}
	}
	sem := semaphore.New(int64(query.Concurrency))
	ctx, endTask := trace.WithTask(ctx, "list-abstractions-streamed-producer")
	go func() {
		defer endTask()
		defer close(out)
		defer close(outErrs)
		_, add, wait := trace.WithTaskGroup(ctx, "list-abstractions-impl-fs")
		defer wait()
		for _, fs := range fss {
			fs := fs  
			add(func(ctx context.Context) {
				g, err := sem.Acquire(ctx)
				if err != nil {
					errCb(err, fs, err.Error())
					return
				}
				func() {
					defer g.Release()
					listAbstractionsImplFS(ctx, fs, &query, emitAbstraction, errCb)
				}()
			})
		}
	}()
	return out, outErrs, nil
}
func listAbstractionsImplFS(ctx context.Context, fs string, query *ListZFSHoldsAndBookmarksQuery, emitCandidate putListAbstraction, errCb putListAbstractionErr) {
	fsp, err := zfs.NewDatasetPath(fs)
	if err != nil {
		panic(err)
	}
	if len(query.What) == 0 {
		return
	}
	whatTypes := zfs.VersionTypeSet{}
	for what := range query.What {
		if e := what.BookmarkExtractor(); e != nil {
			whatTypes[zfs.Bookmark] = true
		}
		if e := what.HoldExtractor(); e != nil {
			whatTypes[zfs.Snapshot] = true
		}
	}
	fsvs, err := zfs.ZFSListFilesystemVersions(ctx, fsp, zfs.ListFilesystemVersionsOptions{
		Types: whatTypes,
	})
	if err != nil {
		errCb(err, fs, "list filesystem versions")
		return
	}
	for at := range query.What {
		bmE := at.BookmarkExtractor()
		holdE := at.HoldExtractor()
		if bmE == nil && holdE == nil || bmE != nil && holdE != nil {
			panic("implementation error: extractors misconfigured for " + at)
		}
		for _, v := range fsvs {
			if v.Type == zfs.Bookmark && bmE != nil {
				if a := bmE(fsp, v); a != nil {
					emitCandidate(a)
				}
			}
			if v.Type == zfs.Snapshot && holdE != nil && query.CreateTXG.Contains(v.GetCreateTXG()) && (!v.UserRefs.Valid || v.UserRefs.Value > 0) {  
				holds, err := zfs.ZFSHolds(ctx, fsp.ToString(), v.Name)
				if err != nil {
					errCb(err, v.ToAbsPath(fsp), "get hold on snap")
					continue
				}
				for _, tag := range holds {
					if a := holdE(fsp, v, tag); a != nil {
						emitCandidate(a)
					}
				}
			}
		}
	}
}
type BatchDestroyResult struct {
	Abstraction
	DestroyErr error
}
var _ json.Marshaler = (*BatchDestroyResult)(nil)
func (r BatchDestroyResult) MarshalJSON() ([]byte, error) {
	err := ""
	if r.DestroyErr != nil {
		err = r.DestroyErr.Error()
	}
	s := struct {
		Abstraction AbstractionJSON
		DestroyErr  string
	}{
		AbstractionJSON{r.Abstraction},
		err,
	}
	return json.Marshal(s)
}
func BatchDestroy(ctx context.Context, abs []Abstraction) <-chan BatchDestroyResult {
	res := make(chan BatchDestroyResult, len(abs))
	go func() {
		for _, a := range abs {
			res <- BatchDestroyResult{
				a,
				a.Destroy(ctx),
			}
		}
		close(res)
	}()
	return res
}
type StalenessInfo struct {
	ConstructedWithQuery ListZFSHoldsAndBookmarksQuery
	Live                 []Abstraction
	Stale                []Abstraction
}
type fsAndJobId struct {
	fs    string
	jobId JobID
}
type ListStaleQueryError struct {
	error
}
func ListStale(ctx context.Context, q ListZFSHoldsAndBookmarksQuery) (*StalenessInfo, error) {
	if !q.CreateTXG.IsUnbounded() {
		return nil, &ListStaleQueryError{errors.New("ListStale cannot have Until != nil set on query")}
	}
	ifAnyThenAll := AbstractionTypeSet{
		AbstractionStepHold:                    true,
		AbstractionReplicationCursorBookmarkV2: true,
	}
	if q.What.ContainsAnyOf(ifAnyThenAll) && !q.What.ContainsAll(ifAnyThenAll) {
		return nil, &ListStaleQueryError{errors.Errorf("ListStale requires query to ask for all of %s", ifAnyThenAll.String())}
	}
	qAbs, absErr, err := ListAbstractions(ctx, q)
	if err != nil {
		return nil, err
	}
	if len(absErr) > 0 {
		return nil, ListAbstractionsErrors(absErr)
	}
	si := listStaleFiltering(qAbs, q.CreateTXG.Since)
	si.ConstructedWithQuery = q
	return si, nil
}
type fsAjobAtype struct {
	fsAndJobId
	Type AbstractionType
}