Add throughput and ETA stats to fsck UI (#773)

Adds another line to the fsck TUI, showing the total and avg. throughput (bytes, resources, and errors) along with an ETA.

Author: AlCutter

cmd/fsck/tui/fsck_panel.go

@@ -25,7 +25,9 @@ import (
 
 // NewFsckPanel creates a new TUI panel showing information about an ongoing fsck operation.
 func NewFsckPanel() *FsckPanel {
-	r := &FsckPanel{}
+	r := &FsckPanel{
+		statsView: NewStatsView(),
+	}
 	return r
 }
 
@@ -37,6 +39,8 @@ type FsckPanel struct {
 	// The zeroth entry corresponds to the tiles on level zero.
 	tilesBars []*LayerProgressModel
 
+	statsView *StatsViewModel
+
 	// width is the width of the app window
 	width int
 }
@@ -90,6 +94,9 @@ func (m *FsckPanel) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
 			cmds = append(cmds, cmd)
 		}
 
+		_, cmd = m.statsView.Update(msg)
+		cmds = append(cmds, cmd)
+
 		return m, tea.Batch(cmds...)
 	default:
 		return m, nil
@@ -110,14 +117,17 @@ func (m *FsckPanel) View() string {
 	bars = append(bars, LayerProgressKey())
 	barsView := lipgloss.JoinVertical(lipgloss.Bottom, bars...)
 
-	content := lipgloss.NewStyle().
+	progress := lipgloss.NewStyle().
 		Width(m.width).
 		Height(lipgloss.Height(barsView)+1).
-		Align(lipgloss.Center, lipgloss.Top).
 		Border(lipgloss.NormalBorder(), true, false, false, false).
 		Render(barsView)
 
-	return lipgloss.JoinVertical(lipgloss.Top, content)
+	stats := lipgloss.NewStyle().Align(lipgloss.Left).Render(m.statsView.View())
+
+	return lipgloss.NewStyle().Align(lipgloss.Center).Render(
+		lipgloss.JoinVertical(lipgloss.Top, progress, stats),
+	)
 }
 
 // FsckPanelUpdateMsg is used to tell the FsckPanel about updated status from the fsck operation.

cmd/fsck/tui/stats_view.go

@@ -0,0 +1,138 @@
+// Copyright 2025 The Tessera authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package tui
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/dustin/go-humanize"
+	"github.com/transparency-dev/tessera/fsck"
+
+	tea "github.com/charmbracelet/bubbletea"
+	"github.com/charmbracelet/lipgloss"
+
+	movingaverage "github.com/RobinUS2/golang-moving-average"
+)
+
+// NewStatsView returns a widet which displays fsck statistics.
+func NewStatsView() *StatsViewModel {
+	m := &StatsViewModel{
+		bytesAvg:     movingaverage.New(10),
+		resourcesAvg: movingaverage.New(10),
+		errorsAvg:    movingaverage.New(10),
+	}
+	return m
+}
+
+// StatsViewModel is the UI model for a stats widget.
+type StatsViewModel struct {
+	// width available to this component.
+	width int
+
+	totalBytes     uint64
+	totalResources uint64
+	totalErrors    uint64
+
+	bytesAvg     *movingaverage.MovingAverage
+	resourcesAvg *movingaverage.MovingAverage
+	errorsAvg    *movingaverage.MovingAverage
+
+	startTime  time.Time
+	lastUpdate time.Time
+	eta        time.Time
+}
+
+func (m *StatsViewModel) Init() tea.Cmd {
+	m.startTime = time.Now()
+
+	return nil
+}
+
+// Update is used to update the widget.
+func (m *StatsViewModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
+	switch msg := msg.(type) {
+	case tea.WindowSizeMsg:
+		m.width = msg.Width
+		return m, nil
+	case FsckPanelUpdateMsg:
+		now := time.Now()
+		d := now.Sub(m.lastUpdate)
+		m.lastUpdate = now
+
+		b := msg.Status.BytesFetched
+		m.totalBytes += b
+		bytesPerSecond := float64(b) / d.Seconds()
+		m.bytesAvg.Add(bytesPerSecond)
+
+		r := msg.Status.ResourcesFetched
+		m.totalResources += r
+		resourcesPerSecond := float64(r) / d.Seconds()
+		m.resourcesAvg.Add(resourcesPerSecond)
+
+		e := msg.Status.ErrorsEncountered
+		m.totalErrors += e
+		errorsPerSecond := float64(e) / d.Seconds()
+		m.errorsAvg.Add(errorsPerSecond)
+
+		totalResources := totalSize(msg.Status.EntryRanges)
+		for _, rs := range msg.Status.TileRanges {
+			totalResources += totalSize(rs)
+		}
+		if totalResources > 0 {
+			elapsedTime := float64(now.Sub(m.startTime).Seconds())
+			complete := float64(m.totalResources) / float64(totalResources)
+			m.eta = now.Add(time.Duration(elapsedTime / complete))
+		}
+
+		return m, nil
+	default:
+		return m, nil
+	}
+}
+
+// totalSize returns the total number of elements covered by the provided ranges.
+func totalSize(rs []fsck.Range) uint64 {
+	tot := uint64(0)
+	for _, r := range rs {
+		tot += r.N
+	}
+	return tot
+}
+
+func (m *StatsViewModel) View() string {
+	bytesFetched := lipgloss.NewStyle().Width(27).Render(formatTotalAndAverage("Bytes", m.totalBytes, m.bytesAvg.Avg()))
+	resourcesFetched := lipgloss.NewStyle().Width(30).Render(formatTotalAndAverage("Resources", m.totalResources, m.resourcesAvg.Avg()))
+	errorsEncountered := lipgloss.NewStyle().Width(23).Render(formatTotalAndAverage("Errors", m.totalErrors, m.errorsAvg.Avg()))
+
+	eta := lipgloss.NewStyle().Width(55).Render(fmt.Sprintf("ETA: %s", humanize.Time(m.eta)))
+
+	return lipgloss.NewStyle().
+		Width(m.width).Render(
+		lipgloss.JoinHorizontal(
+			lipgloss.Right,
+			bytesFetched,
+			resourcesFetched,
+			errorsEncountered,
+			eta),
+	)
+}
+
+func formatTotalAndAverage(label string, total uint64, avg float64) string {
+	si, p := humanize.ComputeSI(float64(total))
+	avgSI, avgP := humanize.ComputeSI(avg)
+
+	return fmt.Sprintf("%s: %0.1f %s (%03.1f%s/s)", label, si, p, avgSI, avgP)
+}

fsck/fsck.go

@@ -44,10 +44,12 @@ type Fetcher interface {
 type Fsck struct {
 	origin       string
 	verifier     note.Verifier
-	fetcher      Fetcher
+	fetcher      *countingFetcher
 	bundleHasher func([]byte) ([][]byte, error)
 	rangeTracker *rangeTracker
 	opts         Opts
+
+	fsckTree *fsckTree
 }
 
 type Opts struct {
@@ -72,13 +74,15 @@ func New(origin string, verifier note.Verifier, f Fetcher, bundleHasher func([]b
 	return &Fsck{
 		origin:       origin,
 		verifier:     verifier,
-		fetcher:      f,
+		fetcher:      newCountingFetcher(f),
 		opts:         opts,
 		bundleHasher: bundleHasher,
 	}
 }
 
 // Check performs an integrity check against the log.
+//
+// Check may only be called once per instance of Fsck.
 func (f *Fsck) Check(ctx context.Context) error {
 	cp, cpRaw, _, err := client.FetchCheckpoint(ctx, f.fetcher.ReadCheckpoint, f.verifier, f.origin)
 	if err != nil {
@@ -90,7 +94,7 @@ func (f *Fsck) Check(ctx context.Context) error {
 
 	f.rangeTracker = newRangeTracker(cp.Size)
 
-	fTree := fsckTree{
+	f.fsckTree = &fsckTree{
 		fetcher:           f.fetcher,
 		bundleHasher:      f.bundleHasher,
 		tree:              (&compact.RangeFactory{Hash: rfc6962.DefaultHasher.HashChildren}).NewEmptyRange(0),
@@ -106,17 +110,17 @@ func (f *Fsck) Check(ctx context.Context) error {
 
 	// Kick off resource comparing workers
 	for range f.opts.N {
-		eg.Go(fTree.resourceCheckWorker(ctx))
+		eg.Go(f.fsckTree.resourceCheckWorker(ctx))
 	}
 
 	trackBundle := func(ctx context.Context, idx uint64, p uint8) ([]byte, error) {
-		fTree.rangeTracker.Update(-1, idx, Fetching)
+		f.fsckTree.rangeTracker.Update(-1, idx, Fetching)
 		r, err := f.fetcher.ReadEntryBundle(ctx, idx, p)
 		if err != nil {
-			fTree.rangeTracker.Update(-1, idx, FetchError)
+			f.fsckTree.rangeTracker.Update(-1, idx, FetchError)
 			return nil, err
 		}
-		fTree.rangeTracker.Update(-1, idx, Fetched)
+		f.fsckTree.rangeTracker.Update(-1, idx, Fetched)
 		return r, nil
 	}
 
@@ -127,27 +131,27 @@ func (f *Fsck) Check(ctx context.Context) error {
 		if err != nil {
 			return fmt.Errorf("error while streaming bundles: %v", err)
 		}
-		fTree.rangeTracker.Update(-1, b.RangeInfo.Index, OK)
-		if err := fTree.AppendBundle(b.RangeInfo, b.Data); err != nil {
+		f.fsckTree.rangeTracker.Update(-1, b.RangeInfo.Index, OK)
+		if err := f.fsckTree.AppendBundle(b.RangeInfo, b.Data); err != nil {
 			return fmt.Errorf("failure calling AppendBundle(%v): %v", b.RangeInfo, err)
 		}
-		if fTree.tree.End() >= cp.Size {
+		if f.fsckTree.tree.End() >= cp.Size {
 			break
 		}
 	}
 
 	// Ensure we see process any partial tiles too.
-	fTree.flushPartialTiles()
+	f.fsckTree.flushPartialTiles()
 	// Signal that there will be no more resource checking jobs coming so workers can exit when the channel is drained.
-	close(fTree.expectedResources)
+	close(f.fsckTree.expectedResources)
 
 	// Wait for all the work to be done.
 	if err := eg.Wait(); err != nil {
 		return fmt.Errorf("failed: %v", err)
 	}
 
 	// Finally, check that the claimed root hash matches what we calculated.
-	gotRoot, err := fTree.GetRootHash()
+	gotRoot, err := f.fsckTree.GetRootHash()
 	switch {
 	case err != nil:
 		return fmt.Errorf("failed to calculate root: %v", err)
@@ -160,19 +164,34 @@ func (f *Fsck) Check(ctx context.Context) error {
 	return nil
 }
 
+// Status represents the current status of an ongoing fsck check.
 type Status struct {
+	// EntryRanges describes the status of the entrybundles in the target log.
 	EntryRanges []Range
-	TileRanges  [][]Range
+	// TileRanges describes the status of the tiles in the target log.
+	// The zeroth entry in the slice represents the lower-most tile level, just above the entry bundles.
+	TileRanges [][]Range
+
+	// BytesFetched is the total number of bytes fetched from the target log since the last time Status() was called.
+	BytesFetched uint64
+	// ResourcesFetched is the total number of resources fetched from the target log since the last time Status() was called.
+	ResourcesFetched uint64
+	// ErrorsEncountered is the total number of errors encountered since the last time Status() was called.
+	ErrorsEncountered uint64
 }
 
+// Status returns a struct representing the current status of the fsck operation.
 func (f *Fsck) Status() Status {
 	if f.rangeTracker == nil {
 		return Status{}
 	}
 	e, t := f.rangeTracker.Ranges()
 	r := Status{
-		EntryRanges: e,
-		TileRanges:  t,
+		EntryRanges:       e,
+		TileRanges:        t,
+		BytesFetched:      f.fsckTree.fetcher.bytesFetched.Swap(0),
+		ResourcesFetched:  f.fsckTree.fetcher.resourcesFetched.Swap(0),
+		ErrorsEncountered: f.fsckTree.fetcher.errorsEncountered.Swap(0),
 	}
 	return r
 }
@@ -206,7 +225,7 @@ type resource struct {
 // fsckTree represents the tree we're currently checking.
 type fsckTree struct {
 	// fetcher knows how to retrieve static tlog-tile resources.
-	fetcher Fetcher
+	fetcher *countingFetcher
 	// bundleHasher knows how to convert entry bundles into leaf hashes.
 	bundleHasher func([]byte) ([][]byte, error)
 	// tree contains the running state of the leaves we've appended so far.
@@ -332,3 +351,39 @@ func (f *fsckTree) resourceCheckWorker(ctx context.Context) func() error {
 		return nil
 	}
 }
+
+// countingFetcher is a Fetcher which keeps track of the total number of bytes and resources fetched.
+type countingFetcher struct {
+	f                 Fetcher
+	bytesFetched      atomic.Uint64
+	resourcesFetched  atomic.Uint64
+	errorsEncountered atomic.Uint64
+}
+
+func newCountingFetcher(f Fetcher) *countingFetcher {
+	return &countingFetcher{
+		f: f,
+	}
+}
+
+func (c *countingFetcher) count(r []byte, err error) ([]byte, error) {
+	if err != nil {
+		c.errorsEncountered.Add(1)
+		return nil, err
+	}
+	c.bytesFetched.Add(uint64(len(r)))
+	c.resourcesFetched.Add(1)
+	return r, nil
+}
+
+func (c *countingFetcher) ReadCheckpoint(ctx context.Context) ([]byte, error) {
+	return c.count(c.f.ReadCheckpoint(ctx))
+}
+
+func (c *countingFetcher) ReadEntryBundle(ctx context.Context, i uint64, p uint8) ([]byte, error) {
+	return c.count(c.f.ReadEntryBundle(ctx, i, p))
+}
+
+func (c *countingFetcher) ReadTile(ctx context.Context, l, i uint64, p uint8) ([]byte, error) {
+	return c.count(c.f.ReadTile(ctx, l, i, p))
+}

fsck/fsck_test.go

@@ -65,9 +65,9 @@ func TestTrimFullToPartial(t *testing.T) {
 
 			f := &fsckTree{
 				expectedResources: make(chan resource, 1),
-				fetcher: &fakeFetcher{
+				fetcher: newCountingFetcher(&fakeFetcher{
 					tile: test.storedTile,
-				},
+				}),
 				rangeTracker: newRangeTracker(1),
 			}