Show the progress rate over the last 5 seconds, and where a phase waits after 10 seconds with no item done (closes #125)

The rate was the phase's average, so a burst after a long wait read as
over 1,000 files a second. It now covers the last 5 seconds, or the
time since the previous line off a terminal; the ETA still comes from
the average. Once no item has completed for 10 seconds, the rate gives
way to "stalled" and where the phase waits: disk, db, out (stdout) or
cpu, or two of them, as "stalled disk/db". A line too wide for the
terminal loses its bar first, then its rate or those words, so the
phase, progress, elapsed time and ETA stay whole. The content phase's
last partial batch is now written under a display of its own.

Deviation: the stall words hold no time and name where the phase waits, not what it does.
Judgement call: trees' group phase, which works in memory, says "cpu".

Model: opus-5-5

progress_test.go

@@ -6,7 +6,9 @@ import (
 	"math"
 	"os"
 	"path/filepath"
+	"slices"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 	"unicode/utf8"
@@ -20,6 +22,33 @@ const testRedraw = time.Millisecond
 // it expects.
 const testDeadline = 5 * time.Second
 
+// hashStalledOn is where the hash phase says it waits once it has
+// stalled, as readRuns gives it.
+const hashStalledOn = "disk/db"
+
+// testClock is a clock a test moves by hand, for a display that reads
+// it from its own goroutine too.
+type testClock struct {
+	mu  sync.Mutex
+	now time.Time
+}
+
+// Now returns the clock's time.
+func (c *testClock) Now() time.Time {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	return c.now
+}
+
+// Add moves the clock on by d.
+func (c *testClock) Add(d time.Duration) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.now = c.now.Add(d)
+}
+
 //nolint:paralleltest // replaces the process-wide os.Stderr
 func TestStderrIsTTYFalseForNonTerminals(t *testing.T) {
 	r, pipe, err := os.Pipe()
@@ -72,11 +101,13 @@ func TestPhaseLabel(t *testing.T) {
 }
 
 // TestProgressLine checks the line a display shows in each state, ten
-// seconds into the phase.
+// seconds into the phase, with no draw since the first and the last
+// item just completed.
 func TestProgressLine(t *testing.T) {
 	t.Parallel()
 
-	ago := time.Now().Add(-10 * time.Second)
+	now := time.Now()
+	ago := now.Add(-10 * time.Second)
 
 	cases := []struct {
 		p    *progress
@@ -129,90 +160,120 @@ func TestProgressLine(t *testing.T) {
 	}}
 
 	for _, tc := range cases {
-		tc.p.start, tc.p.rateStart = ago, ago
+		tc.p.start, tc.p.rateStart, tc.p.lastItem = ago, ago, now
+		tc.p.drawn = []drawnCount{{at: ago}}
 
-		if got, _ := tc.p.line(math.MaxInt, true); got != tc.want {
+		if got := tc.p.line(now, math.MaxInt, true); got != tc.want {
 			t.Errorf("line = %q, want %q", got, tc.want)
 		}
 	}
 }
 
 // TestLineFitsRoom checks how a line is fitted to the room it has, ten
-// seconds into the phase: it leaves out the rate, or what the phase
-// waits for, and is cut only when what is left is still too wide.
+// seconds into the phase, and fifty into a phase with no file read for
+// the last 42: it leaves out the rate, what the phase waits for, or the
+// words of a stalled phase, and is cut only when what is left is still
+// too wide.
 func TestLineFitsRoom(t *testing.T) {
 	t.Parallel()
 
-	ago := time.Now().Add(-10 * time.Second)
+	now := time.Now()
+	ago := now.Add(-10 * time.Second)
 	hash := &progress{
 		label: scanPhase("hash"), unit: scanUnit, total: 100, count: 20,
-		start: ago, rateStart: ago,
+		start: ago, rateStart: ago, lastItem: now,
+		drawn: []drawnCount{{at: ago}},
 	}
 	content := &progress{
 		label: scanPhase("content"), total: -1,
 		waitingFor: "searching the database", start: ago, rateStart: ago,
 	}
 
+	started := now.Add(-50 * time.Second)
+	stalled := &progress{
+		label: scanPhase("hash"), unit: scanUnit, total: 100, count: 20,
+		stalledOn: hashStalledOn, start: started, rateStart: started,
+		lastItem: now.Add(-42 * time.Second),
+		drawn:    []drawnCount{{at: now.Add(-rateWindow), count: 20}},
+	}
+
 	cases := []struct {
 		p       *progress
 		room    int
 		percent bool
 		want    string
-		fits    bool
 	}{{
 		p: hash, room: 57, percent: true,
 		want: "[3/5] hash: [20/100] 20%, 2 files/s, elapsed 10s, eta 40s",
-		fits: true,
 	}, {
 		p: hash, room: 56, percent: true,
 		want: "[3/5] hash: [20/100] 20%, elapsed 10s, eta 40s",
-		fits: true,
 	}, {
 		p: hash, room: 45, percent: true,
 		want: "[3/5] hash: [20/100] 20%, elapsed 10s, eta 40",
-		fits: false,
 	}, {
 		// Before the library's bar, which shows the percent.
 		p: hash, room: 57, percent: false,
 		want: "[3/5] hash: [20/100], 2 files/s, elapsed 10s, eta 40s",
-		fits: true,
 	}, {
 		p: content, room: 40, percent: true,
 		want: "[5/5] content: elapsed 10s, eta unknown",
-		fits: true,
+	}, {
+		p: stalled, room: 65, percent: true,
+		want: "[3/5] hash: [20/100] 20%, stalled disk/db, elapsed 50s, " +
+			"eta 3m20s",
+	}, {
+		p: stalled, room: 64, percent: true,
+		want: "[3/5] hash: [20/100] 20%, elapsed 50s, eta 3m20s",
 	}}
 
 	for _, tc := range cases {
-		got, fits := tc.p.line(tc.room, tc.percent)
-		if got != tc.want || fits != tc.fits {
-			t.Errorf("line(%d, %t) = %q, %t, want %q, %t",
-				tc.room, tc.percent, got, fits, tc.want, tc.fits)
+		if got := tc.p.line(now, tc.room, tc.percent); got != tc.want {
+			t.Errorf("line(%d, %t) = %q, want %q",
+				tc.room, tc.percent, got, tc.want)
 		}
 	}
 }
 
 // TestSetTotalStartsRate checks that setTotal starts a phase's rate,
-// and so its ETA, from when the total becomes known, not from when the
-// phase started waiting for it.
+// its ETA and its time with no record read from when the total becomes
+// known, not from when the phase started waiting for it: after 20
+// seconds of waiting for the total, the phase has not stalled 5 seconds
+// after it, and has 10 seconds after it with no record read.
 func TestSetTotalStartsRate(t *testing.T) {
 	t.Parallel()
 
-	waited := time.Now().Add(-20 * time.Second)
+	clock := &testClock{now: time.Now()}
+	waited := clock.Now().Add(-20 * time.Second)
 	p := &progress{
-		label: scanPhase("load"), unit: scanUnit, total: -1, start: waited,
-		rateStart: waited,
+		now: clock.Now, label: scanPhase("load"), unit: scanUnit, total: -1,
+		stalledOn: "db", start: waited, rateStart: waited,
+		lastItem: waited, drawn: []drawnCount{{at: waited}},
 	}
 
 	p.setTotal(100)
 
-	// Ten seconds after the total came, without waiting ten seconds,
-	// 20 records are read.
-	p.start = p.start.Add(-10 * time.Second)
-	p.rateStart = p.rateStart.Add(-10 * time.Second)
-	p.count = 20
+	clock.Add(5 * time.Second)
 
-	want := "[1/5] load: [20/100] 20%, 2 files/s, elapsed 30s, eta 40s"
-	if got, _ := p.line(math.MaxInt, true); got != want {
+	want := "[1/5] load: [0/100] 0%, 0 files/s, elapsed 25s, eta unknown"
+	if got := p.line(clock.Now(), math.MaxInt, true); got != want {
+		t.Errorf("line = %q, want %q", got, want)
+	}
+
+	clock.Add(5 * time.Second)
+
+	want = "[1/5] load: [0/100] 0%, stalled db, elapsed 30s, eta unknown"
+	if got := p.line(clock.Now(), math.MaxInt, true); got != want {
+		t.Errorf("line = %q, want %q", got, want)
+	}
+
+	// Then 20 records are read.
+	for range 20 {
+		p.increment()
+	}
+
+	want = "[1/5] load: [20/100] 20%, 2 files/s, elapsed 30s, eta 40s"
+	if got := p.line(clock.Now(), math.MaxInt, true); got != want {
 		t.Errorf("line = %q, want %q", got, want)
 	}
 }
@@ -272,13 +333,137 @@ func TestPlainProgressPrintsWhileWaiting(t *testing.T) {
 	p := startProgress(progressParams{
 		label: scanPhase("load"), total: -1,
 		waitingFor: countingRecords,
-	}, nil, testRedraw)
+	}, nil, testRedraw, time.Now)
 
 	awaitOutput(t, stderr,
 		"[1/5] load: counting the records, elapsed 0s, eta unknown\n", 3)
 	p.finish()
 }
 
+// TestPlainProgressStalledProducer checks that a phase whose files stop
+// coming, as when one large file is being read or a disk is slow, goes
+// on printing its line every interval: the rate falls to 0 once no file
+// has been read for rateWindow, and once none has for stallTime the
+// line says the phase has stalled, and where it waits, in place of the
+// rate. The test moves the display's clock by hand.
+//
+//nolint:paralleltest // captureStderr replaces the process-wide os.Stderr
+func TestPlainProgressStalledProducer(t *testing.T) {
+	stderr := captureStderr(t)
+	clock := &testClock{now: time.Now()}
+	p := startProgress(progressParams{
+		label: scanPhase("hash"), unit: scanUnit, total: 100,
+		stalledOn: hashStalledOn,
+	}, nil, testRedraw, clock.Now)
+
+	// Pass or fail, the display ends before stderr is put back.
+	t.Cleanup(p.finish)
+
+	// 20 files read as the phase starts, then none.
+	for range 20 {
+		p.increment()
+	}
+
+	clock.Add(5 * time.Second)
+	awaitOutput(t, stderr,
+		"[3/5] hash: [20/100] 20%, 4 files/s, elapsed 5s, eta 20s\n", 3)
+
+	clock.Add(time.Second)
+	awaitOutput(t, stderr,
+		"[3/5] hash: [20/100] 20%, 0 files/s, elapsed 6s, eta 24s\n", 3)
+
+	clock.Add(4 * time.Second)
+	awaitOutput(t, stderr,
+		"[3/5] hash: [20/100] 20%, stalled disk/db, elapsed 10s, eta 40s\n",
+		3)
+}
+
+// TestPlainProgressRateAfterStall draws a phase's plain line by hand, a
+// little over plainInterval after the line before, as a late tick
+// does, through a stall and a burst of files after it: the rate covers
+// the files read since the line before, while the ETA comes from the
+// phase's average.
+//
+//nolint:paralleltest // captureStderr replaces the process-wide os.Stderr
+func TestPlainProgressRateAfterStall(t *testing.T) {
+	stderr := captureStderr(t)
+	clock := &testClock{now: time.Now()}
+
+	// No timed draws: the test draws.
+	p := startProgress(progressParams{
+		label: scanPhase("hash"), unit: scanUnit, total: 1000,
+		stalledOn: hashStalledOn,
+	}, nil, time.Hour, clock.Now)
+
+	late := plainInterval + time.Millisecond
+
+	clock.Add(late)
+	p.draw()
+	clock.Add(late)
+	p.draw()
+
+	for range 100 {
+		p.increment()
+	}
+
+	clock.Add(late)
+	p.finish()
+
+	want := []string{
+		"[3/5] hash: [0/1000] 0%, 0 files/s, elapsed 0s, eta unknown",
+		"[3/5] hash: [0/1000] 0%, 0 files/s, elapsed 5s, eta unknown",
+		"[3/5] hash: [0/1000] 0%, stalled disk/db, elapsed 10s, eta unknown",
+		"[3/5] hash: [100/1000] 10%, 20 files/s, elapsed 15s, eta 2m15s",
+	}
+
+	lines := strings.Split(strings.TrimSuffix(stderr(), "\n"), "\n")
+	if !slices.Equal(lines, want) {
+		t.Errorf("stderr lines = %q, want %q", lines, want)
+	}
+}
+
+// TestProgressRateAfterStallRedrawn draws a phase by hand every
+// redrawInterval, as a terminal does, through a stall of stallTime and
+// a burst of files after it: the rate covers the files read over the
+// last rateWindow, and falls to 0 once the burst is older than that,
+// while the ETA comes from the phase's average. Plain lines show each
+// draw.
+//
+//nolint:paralleltest // captureStderr replaces the process-wide os.Stderr
+func TestProgressRateAfterStallRedrawn(t *testing.T) {
+	stderr := captureStderr(t)
+	clock := &testClock{now: time.Now()}
+
+	// No timed draws: the test draws.
+	p := startProgress(progressParams{
+		label: scanPhase("hash"), unit: scanUnit, total: 1000,
+		stalledOn: hashStalledOn,
+	}, nil, time.Hour, clock.Now)
+
+	// redraw draws the display every redrawInterval for d.
+	redraw := func(d time.Duration) {
+		for range d / redrawInterval {
+			clock.Add(redrawInterval)
+			p.draw()
+		}
+	}
+
+	redraw(stallTime)
+
+	for range 50 {
+		p.increment()
+	}
+
+	redraw(rateWindow + redrawInterval)
+	p.finish()
+
+	want := "[3/5] hash: [50/1000] 5%, 10 files/s, elapsed 15s, eta 4m45s\n" +
+		"[3/5] hash: [50/1000] 5%, 0 files/s, elapsed 15s, eta 4m47s\n"
+	if !strings.Contains(stderr(), want) {
+		t.Errorf("stderr = %q, want the lines %q", stderr(), want)
+	}
+}
+
 // lastFrame returns what a terminal shows of the frames a display drew:
 // the last one. The library starts each frame with a carriage return
 // and erases the previous one with spaces first.
@@ -321,7 +506,7 @@ func TestTerminalRedrawsWhileWaiting(t *testing.T) {
 	for _, tc := range cases {
 		before := len(stderr())
 		since := func() string { return stderr()[before:] }
-		p := startProgress(tc.params, terminalWidth, testRedraw)
+		p := startProgress(tc.params, terminalWidth, testRedraw, time.Now)
 
 		for range 50 {
 			p.increment()
@@ -343,21 +528,23 @@ func TestTerminalRedrawsWhileWaiting(t *testing.T) {
 
 // drawLate draws the display params describes on a terminal width
 // columns wide as it starts and once more as it ends, with count items
-// done two and a half hours into the phase, and returns what it wrote
-// to stderr.
+// done two and a half hours into the phase, the last just now, and
+// returns what it wrote to stderr.
 func drawLate(stderr func() string, params progressParams, width int,
 	count int64,
 ) string {
 	before := len(stderr())
+	clock := &testClock{now: time.Now()}
 
 	// No timed draws come between the two.
-	p := startProgress(params, func() int { return width }, time.Hour)
+	p := startProgress(params, func() int { return width }, time.Hour,
+		clock.Now)
 
 	// Without counting the items one by one or waiting the hours.
+	clock.Add(150 * time.Minute)
+
 	p.mu.Lock()
-	p.count = count
-	p.start = p.start.Add(-150 * time.Minute)
-	p.rateStart = p.start
+	p.count, p.lastItem = count, clock.Now()
 	p.mu.Unlock()
 
 	p.finish()
@@ -413,6 +600,85 @@ func TestTerminalLinesKeepETA(t *testing.T) {
 	}
 }
 
+// TestTerminalStalledLines draws every phase's line on an 80-column
+// terminal once the phase has stalled, its last item done 42 seconds
+// before. At an ordinary scan's counts and times, 12345 of 45678 items
+// done four and a half minutes in, the line shows where the phase waits
+// whole, beside the phase, progress, elapsed time and ETA. At a large
+// scan's, 1234567 of 9876543 two and a half hours in, those four still
+// show whole, as the words give way first where they do not fit.
+//
+//nolint:paralleltest // captureStderr replaces the process-wide os.Stderr
+func TestTerminalStalledLines(t *testing.T) {
+	stderr := captureStderr(t)
+
+	// drawStalled draws the display params describes as it starts and
+	// once more as it ends, with count of total items done elapsed into
+	// the phase, and returns what the terminal shows.
+	drawStalled := func(params progressParams, total, count int64,
+		elapsed time.Duration,
+	) string {
+		before := len(stderr())
+		clock := &testClock{now: time.Now()}
+		params.total = total
+
+		// No timed draws come between the two.
+		p := startProgress(params, func() int { return 80 }, time.Hour,
+			clock.Now)
+
+		clock.Add(elapsed - 42*time.Second)
+
+		p.mu.Lock()
+		p.count, p.lastItem = count, clock.Now()
+		p.mu.Unlock()
+
+		clock.Add(42 * time.Second)
+		p.finish()
+
+		return lastFrame(stderr()[before:])
+	}
+
+	phases := []progressParams{
+		{label: scanPhase("load"), stalledOn: "db"},
+		{label: scanPhase("walk"), estimate: true, stalledOn: "disk"},
+		{label: scanPhase("hash"), stalledOn: hashStalledOn},
+		{label: scanPhase("update"), stalledOn: "db"},
+		// The content phase's checks and reads, then its last write.
+		{label: scanPhase("content"), stalledOn: "disk/db"},
+		{label: scanPhase("content"), stalledOn: "db"},
+		{label: treesPhase("read"), stalledOn: "db"},
+		{label: treesPhase("group"), stalledOn: "cpu"},
+		{label: treesPhase("write"), stalledOn: "out"},
+		{label: reportPhase("read"), stalledOn: "db/out"},
+	}
+
+	for _, params := range phases {
+		approx := ""
+		if params.estimate {
+			approx = "~"
+		}
+
+		shown := drawStalled(params, 45678, 12345,
+			4*time.Minute+30*time.Second)
+		want := params.label + ": [12345/" + approx + "45678] 27%, stalled " +
+			params.stalledOn + ", elapsed 4m30s, eta 12m9s"
+
+		if !strings.Contains(shown, want) {
+			t.Errorf("terminal shows %q, want %q", shown, want)
+		}
+
+		shown = drawStalled(params, 9876543, 1234567, 150*time.Minute)
+		for _, want := range []string{
+			params.label + ": [1234567/" + approx + "9876543] 12%, ",
+			", elapsed 2h30m0s, eta 17h30m0s",
+		} {
+			if !strings.Contains(shown, want) {
+				t.Errorf("terminal shows %q, want %q", shown, want)
+			}
+		}
+	}
+}
+
 // TestTerminalLinesFitWidth draws a large scan's bar display and a small
 // scan's, two and a half hours into the phase, on terminals from 120
 // columns down to 40, and checks that every line stays clear of the
@@ -455,7 +721,7 @@ func TestTerminalLinesFitWidth(t *testing.T) {
 // times. Nothing written after that, the erase before each redraw
 // included, may reach the narrower terminal's last column: it would
 // wrap, and every redraw would leave a line behind. At these widths the
-// spinner stays a spinner and the bar stays a bar.
+// spinner stays a spinner and the bar gives way to the spinner.
 //
 //nolint:paralleltest // captureStderr replaces the process-wide os.Stderr
 func TestTerminalNarrowedRedrawsInPlace(t *testing.T) {
@@ -479,14 +745,16 @@ func TestTerminalNarrowedRedrawsInPlace(t *testing.T) {
 
 	for _, tc := range cases {
 		width := 120
+		clock := &testClock{now: time.Now()}
 
 		// No timed draws: the test draws.
-		p := startProgress(tc.params, func() int { return width }, time.Hour)
+		p := startProgress(tc.params, func() int { return width }, time.Hour,
+			clock.Now)
+
+		clock.Add(150 * time.Minute)
 
 		p.mu.Lock()
-		p.count = 1234567
-		p.start = p.start.Add(-150 * time.Minute)
-		p.rateStart = p.start
+		p.count, p.lastItem = 1234567, clock.Now()
 		p.mu.Unlock()
 
 		p.draw()
@@ -528,7 +796,7 @@ func TestProgressWarningsOnOwnLines(t *testing.T) {
 	issued := 0
 
 	for _, params := range cases {
-		p := startProgress(params, terminalWidth, testRedraw)
+		p := startProgress(params, terminalWidth, testRedraw, time.Now)
 
 		// No pause between warnings: one written straight to stderr is
 		// garbled only if a redraw lands while it is being written.
@@ -581,7 +849,7 @@ func TestProgressWriterRowsOnOwnLines(t *testing.T) {
 	var want []string
 
 	for _, params := range cases {
-		p := startProgress(params, terminalWidth, testRedraw)
+		p := startProgress(params, terminalWidth, testRedraw, time.Now)
 		out := p.writer(os.Stderr)
 
 		// No pause between rows: one is garbled only if a redraw lands
@@ -644,7 +912,7 @@ func TestProgressWriterStalledStdoutKeepsLines(t *testing.T) {
 	})
 
 	p := startProgress(progressParams{label: treesPhase("write"), total: 1},
-		nil, testRedraw)
+		nil, testRedraw, time.Now)
 	out := p.writer(w)
 	written := make(chan error, 1)
 
@@ -687,7 +955,7 @@ func TestProgressWriterStalledStdoutKeepsLines(t *testing.T) {
 func TestTerminalStoppedShortKeepsCount(t *testing.T) {
 	stderr := captureStderr(t)
 	p := startProgress(progressParams{label: scanPhase("hash"), total: 10},
-		terminalWidth, redrawInterval)
+		terminalWidth, redrawInterval, time.Now)
 
 	p.increment()
 	p.increment()
@@ -706,7 +974,7 @@ func TestTerminalStoppedShortKeepsCount(t *testing.T) {
 func TestWarningAfterLastItemKeepsBar(t *testing.T) {
 	stderr := captureStderr(t)
 	p := startProgress(progressParams{label: scanPhase("hash"), total: 2},
-		terminalWidth, testRedraw)
+		terminalWidth, testRedraw, time.Now)
 
 	p.increment()
 	p.increment()