fixup e0b5bd7: crash on endpoint.ListStale if replication-cursor-v1 bookmark present

```
cs@cstp:[~/zrepl/zrepl]: artifacts/zrepl-linux-amd64 zfs-abstraction release-stale --dry-run
panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x0 pc=0x9de971]

goroutine 1 [running]:
github.com/zrepl/zrepl/endpoint.listStaleFiltering(0xc00012b700, 0x5, 0x8, 0x0, 0x13ccb58)
        /endpoint/endpoint_zfs_abstraction.go:736 +0x281
github.com/zrepl/zrepl/endpoint.ListStale(0xe3ae20, 0xc00026b740, 0x0, 0xe27c60, 0x13ccb58, 0xc0001d8690, 0x0, 0x0, 0x0, 0x1, ...)
        /endpoint/endpoint_zfs_abstraction.go:698 +0x3fd
github.com/zrepl/zrepl/client.doZabsReleaseStale(0xe3ae20, 0xc00026b740, 0x13a28a0, 0xc000151be0, 0x0, 0x1, 0x5705b4, 0xc0002686e0)
        /client/zfsabstractions_release.go:83 +0x1a0
github.com/zrepl/zrepl/cli.(*Subcommand).run(0x13a28a0, 0xc000264c80, 0xc000151be0, 0x0, 0x1)
        /cli/cli.go:104 +0xf5
github.com/spf13/cobra.(*Command).execute(0xc000264c80, 0xc000151bd0, 0x1, 0x1, 0xc000264c80, 0xc000151bd0)
        GOROOT/pkg/mod/github.com/spf13/cobra@v0.0.2/command.go:760 +0x2aa
github.com/spf13/cobra.(*Command).ExecuteC(0x13a43c0, 0x0, 0x0, 0x0)
        GOROOT/pkg/mod/github.com/spf13/cobra@v0.0.2/command.go:846 +0x2ea
github.com/spf13/cobra.(*Command).Execute(...)
        GOROOT/pkg/mod/github.com/spf13/cobra@v0.0.2/command.go:794
github.com/zrepl/zrepl/cli.Run()
        /cli/cli.go:151 +0x2d
main.main()
        /main.go:24 +0x20
```

client/migrate.go

@@ -211,7 +211,7 @@ func doMigrateReplicationCursorFS(ctx context.Context, v1CursorJobs []job.Job, f
 	}
 	fmt.Printf("identified owning job %q\n", owningJob.Name())
 
-	bookmarks, err := zfs.ZFSListFilesystemVersions(fs, zfs.ListFilesystemVersionsOptions{
+	bookmarks, err := zfs.ZFSListFilesystemVersions(ctx, fs, zfs.ListFilesystemVersionsOptions{
 		Types: zfs.Bookmarks,
 	})
 	if err != nil {

daemon/snapper/snapper.go

@@ -485,7 +485,7 @@ var findSyncPointFSNoFilesystemVersionsErr = fmt.Errorf("no filesystem versions"
 
 func findSyncPointFSNextOptimalSnapshotTime(l Logger, now time.Time, interval time.Duration, prefix string, d *zfs.DatasetPath) (time.Time, error) {
 
-	fsvs, err := zfs.ZFSListFilesystemVersions(d, zfs.ListFilesystemVersionsOptions{
+	fsvs, err := zfs.ZFSListFilesystemVersions(context.TODO(), d, zfs.ListFilesystemVersionsOptions{
 		Types:           zfs.Snapshots,
 		ShortnamePrefix: prefix,
 	})

endpoint/endpoint.go

@@ -96,7 +96,7 @@ func (s *Sender) ListFilesystemVersions(ctx context.Context, r *pdu.ListFilesyst
 	if err != nil {
 		return nil, err
 	}
-	fsvs, err := zfs.ZFSListFilesystemVersions(lp, zfs.ListFilesystemVersionsOptions{})
+	fsvs, err := zfs.ZFSListFilesystemVersions(ctx, lp, zfs.ListFilesystemVersionsOptions{})
 	if err != nil {
 		return nil, err
 	}
@@ -648,7 +648,7 @@ func (s *Receiver) ListFilesystemVersions(ctx context.Context, req *pdu.ListFile
 	}
 	// TODO share following code with sender
 
-	fsvs, err := zfs.ZFSListFilesystemVersions(lp, zfs.ListFilesystemVersionsOptions{})
+	fsvs, err := zfs.ZFSListFilesystemVersions(ctx, lp, zfs.ListFilesystemVersionsOptions{})
 	if err != nil {
 		return nil, err
 	}

endpoint/endpoint_zfs_abstraction.go

@@ -573,7 +573,7 @@ func listAbstractionsImplFS(ctx context.Context, fs string, query *ListZFSHoldsA
 			whatTypes[zfs.Snapshot] = true
 		}
 	}
-	fsvs, err := zfs.ZFSListFilesystemVersions(fsp, zfs.ListFilesystemVersionsOptions{
+	fsvs, err := zfs.ZFSListFilesystemVersions(ctx, fsp, zfs.ListFilesystemVersionsOptions{
 		Types: whatTypes,
 	})
 	if err != nil {
@@ -731,6 +731,9 @@ func listStaleFiltering(abs []Abstraction, sinceBound *CreateTXGRangeBound) *Sta
 	}
 	stepFirstNotStaleCandidates := make(map[fsAndJobId]stepFirstNotStaleCandidate) // empty map => will always return nil
 	for _, a := range abs {
+		if a.GetJobID() == nil {
+			continue // already put those into always-live list noJobId in above loop
+		}
 		key := fsAndJobId{a.GetFS(), *a.GetJobID()}
 		c := stepFirstNotStaleCandidates[key]
 

platformtest/tests/listFilesystemVersions.go

@@ -43,7 +43,7 @@ func ListFilesystemVersionsTypeFilteringAndPrefix(t *platformtest.Context) {
 	fs := fmt.Sprintf("%s/foo bar", t.RootDataset)
 
 	// no options := all types
-	vs, err := zfs.ZFSListFilesystemVersions(mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
+	vs, err := zfs.ZFSListFilesystemVersions(t, mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
 	require.NoError(t, err)
 	require.Equal(t, []string{
 		"#blup 1", "#bookfoo 1", "#bookfoo 2", "#foo 1", "#foo 2",
@@ -51,21 +51,21 @@ func ListFilesystemVersionsTypeFilteringAndPrefix(t *platformtest.Context) {
 	}, versionRelnamesSorted(vs))
 
 	// just snapshots
-	vs, err = zfs.ZFSListFilesystemVersions(mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{
+	vs, err = zfs.ZFSListFilesystemVersions(t, mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{
 		Types: zfs.Snapshots,
 	})
 	require.NoError(t, err)
 	require.Equal(t, []string{"@ foo with leading whitespace", "@blup 1", "@foo 1", "@foo 2"}, versionRelnamesSorted(vs))
 
 	// just bookmarks
-	vs, err = zfs.ZFSListFilesystemVersions(mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{
+	vs, err = zfs.ZFSListFilesystemVersions(t, mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{
 		Types: zfs.Bookmarks,
 	})
 	require.NoError(t, err)
 	require.Equal(t, []string{"#blup 1", "#bookfoo 1", "#bookfoo 2", "#foo 1", "#foo 2"}, versionRelnamesSorted(vs))
 
 	// just with prefix foo
-	vs, err = zfs.ZFSListFilesystemVersions(mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{
+	vs, err = zfs.ZFSListFilesystemVersions(t, mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{
 		ShortnamePrefix: "foo",
 	})
 	require.NoError(t, err)
@@ -82,7 +82,7 @@ func ListFilesystemVersionsZeroExistIsNotAnError(t *platformtest.Context) {
 
 	fs := fmt.Sprintf("%s/foo bar", t.RootDataset)
 
-	vs, err := zfs.ZFSListFilesystemVersions(mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
+	vs, err := zfs.ZFSListFilesystemVersions(t, mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
 	require.Empty(t, vs)
 	require.NoError(t, err)
 	dsne, ok := err.(*zfs.DatasetDoesNotExist)
@@ -98,7 +98,7 @@ func ListFilesystemVersionsFilesystemNotExist(t *platformtest.Context) {
 
 	nonexistentFS := fmt.Sprintf("%s/not existent", t.RootDataset)
 
-	vs, err := zfs.ZFSListFilesystemVersions(mustDatasetPath(nonexistentFS), zfs.ListFilesystemVersionsOptions{})
+	vs, err := zfs.ZFSListFilesystemVersions(t, mustDatasetPath(nonexistentFS), zfs.ListFilesystemVersionsOptions{})
 	require.Empty(t, vs)
 	require.Error(t, err)
 	t.Logf("err = %T\n%s", err, err)
@@ -141,7 +141,7 @@ func ListFilesystemVersionsUserrefs(t *platformtest.Context) {
 
 	fs := fmt.Sprintf("%s/foo bar", t.RootDataset)
 
-	vs, err := zfs.ZFSListFilesystemVersions(mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
+	vs, err := zfs.ZFSListFilesystemVersions(t, mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
 	require.NoError(t, err)
 
 	type expectation struct {

zfs/versions.go

@@ -206,13 +206,13 @@ func (o *ListFilesystemVersionsOptions) matches(v FilesystemVersion) bool {
 }
 
 // returned versions are sorted by createtxg FIXME drop sort by createtxg requirement
-func ZFSListFilesystemVersions(fs *DatasetPath, options ListFilesystemVersionsOptions) (res []FilesystemVersion, err error) {
+func ZFSListFilesystemVersions(ctx context.Context, fs *DatasetPath, options ListFilesystemVersionsOptions) (res []FilesystemVersion, err error) {
 	listResults := make(chan ZFSListResult)
 
 	promTimer := prometheus.NewTimer(prom.ZFSListFilesystemVersionDuration.WithLabelValues(fs.ToString()))
 	defer promTimer.ObserveDuration()
 
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	go ZFSListChan(ctx, listResults,
 		[]string{"name", "guid", "createtxg", "creation", "userrefs"},

zfs/zfs.go

@@ -1084,7 +1084,7 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, streamCopier
 	if opts.RollbackAndForceRecv {
 		// destroy all snapshots before `recv -F` because `recv -F`
 		// does not perform a rollback unless `send -R` was used (which we assume hasn't been the case)
-		snaps, err := ZFSListFilesystemVersions(fsdp, ListFilesystemVersionsOptions{
+		snaps, err := ZFSListFilesystemVersions(ctx, fsdp, ListFilesystemVersionsOptions{
 			Types: Snapshots,
 		})
 		if err != nil {

zfs/zfscmd/zfscmd.go

@@ -18,10 +18,10 @@ import (
 )
 
 type Cmd struct {
-	cmd                       *exec.Cmd
-	ctx                       context.Context
-	mtx                       sync.RWMutex
-	startedAt, waitReturnedAt time.Time
+	cmd                                      *exec.Cmd
+	ctx                                      context.Context
+	mtx                                      sync.RWMutex
+	startedAt, waitStartedAt, waitReturnedAt time.Time
 }
 
 func CommandContext(ctx context.Context, name string, arg ...string) *Cmd {
@@ -119,19 +119,65 @@ func (c *Cmd) startPost(err error) {
 }
 
 func (c *Cmd) waitPre() {
-	waitPreLogging(c, time.Now())
+	now := time.Now()
+
+	// ignore duplicate waits
+	c.mtx.Lock()
+	// ignore duplicate waits
+	if !c.waitStartedAt.IsZero() {
+		c.mtx.Unlock()
+		return
+	}
+	c.waitStartedAt = now
+	c.mtx.Unlock()
+
+	waitPreLogging(c, now)
+}
+
+type usage struct {
+	total_secs, system_secs, user_secs float64
 }
 
 func (c *Cmd) waitPost(err error) {
 	now := time.Now()
 
 	c.mtx.Lock()
+	// ignore duplicate waits
+	if !c.waitReturnedAt.IsZero() {
+		c.mtx.Unlock()
+		return
+	}
 	c.waitReturnedAt = now
 	c.mtx.Unlock()
 
-	waitPostReport(c, now)
-	waitPostLogging(c, err, now)
-	waitPostPrometheus(c, err, now)
+	// build usage
+	var u usage
+	{
+		var s *os.ProcessState
+		if err == nil {
+			s = c.cmd.ProcessState
+		} else if ee, ok := err.(*exec.ExitError); ok {
+			s = ee.ProcessState
+		}
+
+		if s == nil {
+			u = usage{
+				total_secs:  c.Runtime().Seconds(),
+				system_secs: -1,
+				user_secs:   -1,
+			}
+		} else {
+			u = usage{
+				total_secs:  c.Runtime().Seconds(),
+				system_secs: s.SystemTime().Seconds(),
+				user_secs:   s.UserTime().Seconds(),
+			}
+		}
+	}
+
+	waitPostReport(c, u, now)
+	waitPostLogging(c, u, err, now)
+	waitPostPrometheus(c, u, err, now)
 }
 
 // returns 0 if the command did not yet finish

zfs/zfscmd/zfscmd_logging.go

@@ -1,7 +1,6 @@
 package zfscmd
 
 import (
-	"os/exec"
 	"time"
 )
 
@@ -28,23 +27,12 @@ func waitPreLogging(c *Cmd, now time.Time) {
 	c.log().Debug("start waiting")
 }
 
-func waitPostLogging(c *Cmd, err error, now time.Time) {
-
-	var total, system, user float64
-
-	total = c.Runtime().Seconds()
-	if ee, ok := err.(*exec.ExitError); ok {
-		system = ee.ProcessState.SystemTime().Seconds()
-		user = ee.ProcessState.UserTime().Seconds()
-	} else {
-		system = -1
-		user = -1
-	}
+func waitPostLogging(c *Cmd, u usage, err error, now time.Time) {
 
 	log := c.log().
-		WithField("total_time_s", total).
-		WithField("systemtime_s", system).
-		WithField("usertime_s", user)
+		WithField("total_time_s", u.total_secs).
+		WithField("systemtime_s", u.system_secs).
+		WithField("usertime_s", u.user_secs)
 
 	if err == nil {
 		log.Info("command exited without error")

zfs/zfscmd/zfscmd_prometheus.go

@@ -46,7 +46,7 @@ func RegisterMetrics(r prometheus.Registerer) {
 	r.MustRegister(metrics.usertime)
 }
 
-func waitPostPrometheus(c *Cmd, err error, now time.Time) {
+func waitPostPrometheus(c *Cmd, u usage, err error, now time.Time) {
 
 	if len(c.cmd.Args) < 2 {
 		getLogger(c.ctx).WithField("args", c.cmd.Args).
@@ -64,10 +64,10 @@ func waitPostPrometheus(c *Cmd, err error, now time.Time) {
 
 	metrics.totaltime.
 		WithLabelValues(labelValues...).
-		Observe(c.Runtime().Seconds())
+		Observe(u.total_secs)
 	metrics.systemtime.WithLabelValues(labelValues...).
-		Observe(c.cmd.ProcessState.SystemTime().Seconds())
+		Observe(u.system_secs)
 	metrics.usertime.WithLabelValues(labelValues...).
-		Observe(c.cmd.ProcessState.UserTime().Seconds())
+		Observe(u.user_secs)
 
 }

zfs/zfscmd/zfscmd_report.go

@@ -57,7 +57,7 @@ func startPostReport(c *Cmd, err error, now time.Time) {
 	active.mtx.Unlock()
 }
 
-func waitPostReport(c *Cmd, now time.Time) {
+func waitPostReport(c *Cmd, _ usage, now time.Time) {
 	active.mtx.Lock()
 	defer active.mtx.Unlock()
 	prev := active.cmds[c]