space: zfs diffs run side by side, give up after four minutes, and die properly

The timeout killed sudo and left zfs holding the pipe, so the refresh of tank sat on
immich's diff indefinitely. The process group is killed now, and a slow dataset is walked
by mtime instead.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

PLAN.md

@@ -356,7 +356,10 @@ something long in one, switch away and back, refresh — it's all still there.
   snapshots, the scan records the newest snapshot of each dataset at start (`bases`), and the
   next one runs `zfs diff -H -F <base>` per dataset: only the directories it names are
   read, their ancestors kept but looked into, everything else kept wholesale without a
-  stat. Elsewhere (ext4, or a dataset with no snapshot) a directory whose mtime — to the
+  stat. The diffs run side by side and one is given four minutes (its process group is
+  killed, sudo and zfs both), after which that dataset is walked by mtime instead — on
+  `tank`, `immich` changes constantly and its diff ran past twenty minutes the first time,
+  longer than walking it. Elsewhere (ext4, or a dataset with no snapshot) a directory whose mtime — to the
   nanosecond — is unchanged keeps its list and only its kept files are stat'ed again, since
   a file can grow without its directory noticing. A **cancelled** scan keeps what it walked,
   marks the folders it never got to (`partial`, "not yet read" in the tree) and **Resume**

internal/space/manager.go

@@ -299,21 +299,42 @@ func (m *Manager) plan(ctx context.Context, s, prev *Scan, datasets map[string]s
 	opts.Prev = prev.root
 	s.ByRoot = map[string]string{}
 	dirty, touched := map[string]bool{}, map[string]bool{}
+	// every dataset is asked at once what changed; the answers come back in root order
+	type diffed struct {
+		changed []change
+		err     error
+		took    time.Duration
+	}
+	answers := map[string]chan diffed{}
+	for _, r := range s.Roots {
+		if base := prev.Bases[r]; base != "" && datasets[r] != "" {
+			ch := make(chan diffed, 1)
+			answers[r] = ch
+			go func(r, base string) {
+				t0 := time.Now()
+				c, err := zfsDiff(ctx, base, r)
+				ch <- diffed{changed: c, err: err, took: time.Since(t0)}
+			}(r, base)
+		}
+	}
 	for _, r := range s.Roots {
 		mode := ModeMTime
 		how := "mtime"
-		if base := prev.Bases[r]; base != "" && datasets[r] != "" {
+		if ch, ok := answers[r]; ok {
-			changed, err := zfsDiff(ctx, base, r)
+			m.mu.Lock()
-			if err == nil {
+			s.Progress.Current = "asking zfs what changed in " + r
-				mode, how = ModeDiff, fmt.Sprintf("zfs diff since %s: %d changes", strings.TrimPrefix(base, datasets[r]), len(changed))
+			m.mu.Unlock()
-				for _, c := range changed {
+			d := <-ch
+			if d.err == nil {
+				mode, how = ModeDiff, fmt.Sprintf("zfs diff since %s: %d changes in %s", strings.TrimPrefix(prev.Bases[r], datasets[r]), len(d.changed), d.took.Round(time.Millisecond))
+				for _, c := range d.changed {
 					dirty[c.parent] = true
 					if c.dir {
 						dirty[c.path] = true
 					}
 				}
 			} else {
-				how = "mtime (" + err.Error() + ")"
+				how = "mtime (zfs diff: " + d.err.Error() + ")"
 			}
 		}
 		if Find(prev.root, r) == nil {
@@ -322,6 +343,9 @@ func (m *Manager) plan(ctx context.Context, s, prev *Scan, datasets map[string]s
 		h.Modes[r] = mode
 		s.ByRoot[r] = how
 	}
+	m.mu.Lock()
+	s.Progress.Current = ""
+	m.mu.Unlock()
 	// what the last walk never got to is read whatever the change list says
 	for _, p := range partialPaths(prev.root, "/") {
 		dirty[p] = true
@@ -388,6 +412,10 @@ func (m *Manager) run(ctx context.Context, s *Scan, prev *Scan, datasets map[str
 		}
 	}
 	h, _ := m.plan(ctx, s, prev, datasets)
+	if ctx.Err() != nil { // cancelled while zfs was being asked
+		finish(errors.New("cancelled"))
+		return
+	}
 	args := append([]string{m.self, "space-walk"}, s.Roots...)
 	if s.AsRoot {
 		args = priv.Wrap(args...)
@@ -543,10 +571,16 @@ func newestSnapshot(ctx context.Context, dataset string) string {
 // zfsDiff lists what changed in a dataset since a snapshot: every changed, added, removed
 // or renamed path, with the directory whose listing that touches.
 func zfsDiff(ctx context.Context, snapshot, mount string) ([]change, error) {
-	ctx, cancel := context.WithTimeout(ctx, 10*time.Minute)
+	// a diff slower than this is not worth waiting for: the walk by mtime is bounded
+	ctx, cancel := context.WithTimeout(ctx, 4*time.Minute)
 	defer cancel()
 	args := priv.Wrap("zfs", "diff", "-H", "-F", snapshot)
 	cmd := exec.CommandContext(ctx, args[0], args[1:]...)
+	// on timeout the whole process group goes, not just sudo: zfs itself would otherwise run
+	// on holding the pipe, and Wait would sit there with it
+	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+	cmd.Cancel = func() error { return syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL) }
+	cmd.WaitDelay = 5 * time.Second
 	var stderr strings.Builder
 	cmd.Stderr = &stderr
 	out, err := cmd.StdoutPipe()
@@ -575,6 +609,9 @@ func zfsDiff(ctx context.Context, snapshot, mount string) ([]change, error) {
 	}
 	if err := cmd.Wait(); err != nil {
 		msg := strings.TrimSpace(stderr.String())
+		if errors.Is(ctx.Err(), context.DeadlineExceeded) {
+			msg = "gave up after 4 min"
+		}
 		if msg == "" {
 			msg = err.Error()
 		}