diff --git a/pkg/atproto/sweep.go b/pkg/atproto/sweep.go index 5a5c4b36..f6a756b9 100644 --- a/pkg/atproto/sweep.go +++ b/pkg/atproto/sweep.go @@ -536,6 +536,7 @@ func (atsync *ATProtoSynchronizer) sweepDeepen(ctx context.Context, progress *sw if err := ctx.Err(); err != nil { return err } + progress.setRound(round + 1) var mu sync.Mutex var next []sweepItem err := runLanes(ctx, atsync.sweepConcurrency(), hostLanes(pending), func(ctx context.Context, item sweepItem) { @@ -609,6 +610,7 @@ type sweepProgress struct { phase string done int windows int + round int total int horizon time.Time started bool @@ -622,6 +624,7 @@ func (p *sweepProgress) begin(phase string, total int, horizon time.Time) { p.total = total p.done = 0 p.windows = 0 + p.round = 0 p.horizon = horizon p.started = true } @@ -642,6 +645,16 @@ func (p *sweepProgress) window() { p.windows++ } +// setRound records which deepening round is running. Rounds are barriers -- +// every repo gets its first window before any repo gets a second -- so repos +// cannot finish before the last round, and users=0 is the expected reading for +// most of the phase. The round is what makes that legible on the status line. +func (p *sweepProgress) setRound(round int) { + p.mu.Lock() + defer p.mu.Unlock() + p.round = round +} + // setHorizon updates how far back the sweep has taken every repo it is working // on. func (p *sweepProgress) setHorizon(horizon time.Time) { @@ -662,7 +675,7 @@ func (p *sweepProgress) status() []any { } kv := []any{"phase", p.phase, "users", p.done, "total", p.total, "horizon", horizon} if p.phase == sweepPhaseDeepen { - kv = append(kv, "windows", p.windows) + kv = append(kv, "round", p.round, "windows", p.windows) } return kv } diff --git a/pkg/atproto/sweep_test.go b/pkg/atproto/sweep_test.go index 9c5f3e51..93529acc 100644 --- a/pkg/atproto/sweep_test.go +++ b/pkg/atproto/sweep_test.go @@ -508,8 +508,9 @@ func TestSweepProgressStatusLine(t *testing.T) { deeper := time.Now().Add(-30 * 24 * time.Hour) progress.begin(sweepPhaseDeepen, 2, deeper) require.Equal(t, - []any{"phase", "deepen", "users", 0, "total", 2, "horizon", deeper.Unix(), "windows", 0}, + []any{"phase", "deepen", "users", 0, "total", 2, "horizon", deeper.Unix(), "round", 0, "windows", 0}, progress.status()) + progress.setRound(1) progress.window() progress.window() progress.window() @@ -518,7 +519,7 @@ func TestSweepProgressStatusLine(t *testing.T) { deepest := time.Now().Add(-180 * 24 * time.Hour) progress.setHorizon(deepest) require.Equal(t, - []any{"phase", "deepen", "users", 2, "total", 2, "horizon", deepest.Unix(), "windows", 3}, + []any{"phase", "deepen", "users", 2, "total", 2, "horizon", deepest.Unix(), "round", 1, "windows", 3}, progress.status()) // The ticker stops when told to, without leaking a goroutine.