Feat/concurrent blob pull

Makefile

@@ -57,7 +57,7 @@ format:
 	@$(GO) fmt ./...
 
 test-unit:
-	@$(GO) test -v ./pkg/compose/... ./internal/...
+	@CGO_ENABLED=1 $(GO) test -race -v ./pkg/compose/... ./internal/...
 
 $(bd):
 	@mkdir -p $@

cmd/composectl/cmd/fetch_progress.go

@@ -0,0 +1,159 @@
+package composectl
+
+import (
+	"fmt"
+	"io"
+	"sort"
+	"sync/atomic"
+	"time"
+
+	"github.com/foundriesio/composeapp/pkg/compose"
+	"github.com/opencontainers/go-digest"
+)
+
+// fetchProgressRenderer renders per-blob download progress. In a TTY it keeps
+// only the in-flight blobs in an in-place redrawn region (bounded by the worker
+// count) and "graduates" a blob to a single permanent line above that region
+// once it completes. Because the live region never exceeds the worker count it
+// cannot outgrow the terminal height, so the cursor-up redraw never clamps at
+// the top of the screen (the defect that caused the same blob to be reprinted
+// on several lines). Without a TTY each blob gets a line when it crosses each
+// 10% of its size and a final completion line, so log collectors such as
+// journalctl still depict download progress.
+//
+// Render is invoked serially from the progress reporter goroutine, so the
+// renderer state needs no locking.
+type fetchProgressRenderer struct {
+	out       io.Writer
+	isTty     bool
+	live      []*compose.BlobFetchProgress // in-flight blobs currently redrawn, in stable order
+	seen      map[digest.Digest]bool       // blobs ever added to the live set
+	liveCount int                          // number of live lines drawn in the previous frame
+	reported  map[digest.Digest]int64      // last reported progress decile per in-flight blob (non-TTY)
+}
+
+func NewFetchProgressRenderer(out io.Writer, isTty bool) *fetchProgressRenderer {
+	return &fetchProgressRenderer{
+		out:      out,
+		isTty:    isTty,
+		seen:     map[digest.Digest]bool{},
+		reported: map[digest.Digest]int64{},
+	}
+}
+
+// Render consumes one progress snapshot and updates the terminal accordingly.
+func (r *fetchProgressRenderer) Render(p *compose.FetchProgress) {
+	done, live := r.track(p)
+	if r.isTty {
+		r.renderTty(done, live)
+		return
+	}
+	r.renderPlain(done, live)
+}
+
+// track folds the latest snapshot into the live set: newly observed in-flight
+// blobs are appended (deterministically ordered), and the live set is split into
+// blobs that have just completed and those still in flight. r.live is left
+// holding only the still-in-flight blobs.
+func (r *fetchProgressRenderer) track(p *compose.FetchProgress) (done, live []*compose.BlobFetchProgress) {
+	var newcomers []*compose.BlobFetchProgress
+	for _, b := range p.Blobs {
+		if r.seen[b.Descriptor.Digest] {
+			continue
+		}
+		if b.State == compose.BlobFetching || b.State == compose.BlobOk {
+			r.seen[b.Descriptor.Digest] = true
+			newcomers = append(newcomers, b)
+		}
+	}
+	sort.Slice(newcomers, func(i, j int) bool {
+		if !newcomers[i].FetchStartTime.Equal(newcomers[j].FetchStartTime) {
+			return newcomers[i].FetchStartTime.Before(newcomers[j].FetchStartTime)
+		}
+		return newcomers[i].Descriptor.Digest < newcomers[j].Descriptor.Digest
+	})
+	r.live = append(r.live, newcomers...)
+
+	for _, b := range r.live {
+		if atomic.LoadInt64(&b.BytesFetched) >= b.Descriptor.Size {
+			done = append(done, b)
+		} else {
+			live = append(live, b)
+		}
+	}
+	r.live = live
+	return done, live
+}
+
+// renderTty graduates completed blobs to permanent lines and redraws the live
+// region in place. The leading cursor-up only covers the live region, so a
+// terminal scroll of the permanent history above it is harmless.
+func (r *fetchProgressRenderer) renderTty(done, live []*compose.BlobFetchProgress) {
+	if r.liveCount > 0 {
+		fmt.Fprintf(r.out, "\033[%dA", r.liveCount)
+	}
+	// Return to column 0 and clear from here to the end of the screen so stale
+	// live lines do not linger.
+	fmt.Fprint(r.out, "\r\033[J")
+	for _, b := range done {
+		fmt.Fprintf(r.out, " %s\n", formatBlobLine(b))
+	}
+	for _, b := range live {
+		fmt.Fprintf(r.out, " %s\n", formatBlobLine(b))
+	}
+	r.liveCount = len(live)
+}
+
+// renderPlain handles non-TTY output (pipes, logs): cursor control is
+// meaningless, so each in-flight blob gets a line when it first appears and
+// then whenever it crosses another 10% of its size, plus one final line when
+// it completes.
+func (r *fetchProgressRenderer) renderPlain(done, live []*compose.BlobFetchProgress) {
+	for _, b := range done {
+		delete(r.reported, b.Descriptor.Digest)
+		fmt.Fprintf(r.out, " %s\n", formatBlobLine(b))
+	}
+	for _, b := range live {
+		if r.crossedDecile(b) {
+			fmt.Fprintf(r.out, " %s\n", formatBlobLine(b))
+		}
+	}
+}
+
+// crossedDecile reports whether the blob has entered a 10%-of-size bucket that
+// has not been reported yet, recording the new bucket if so. The first call for
+// a blob always reports, so its download start shows up in the log.
+func (r *fetchProgressRenderer) crossedDecile(b *compose.BlobFetchProgress) bool {
+	if b.Descriptor.Size <= 0 {
+		return false
+	}
+	decile := 100 * atomic.LoadInt64(&b.BytesFetched) / b.Descriptor.Size / 10
+	last, ok := r.reported[b.Descriptor.Digest]
+	if ok && decile <= last {
+		return false
+	}
+	r.reported[b.Descriptor.Digest] = decile
+	return true
+}
+
+// formatBlobLine renders a single blob's status line. All cross-goroutine
+// counters are read via atomic loads since the fetch workers update them while
+// this runs on the reporter goroutine.
+func formatBlobLine(b *compose.BlobFetchProgress) string {
+	fetched := atomic.LoadInt64(&b.BytesFetched)
+	line := fmt.Sprintf("[%-12s] %.12s start: %s from: %10s ",
+		b.Type,
+		b.Descriptor.Digest.Encoded(),
+		b.FetchStartTime.UTC().Format(time.TimeOnly),
+		compose.FormatBytesInt64(b.BlobInfo.BytesFetched))
+	line += fmt.Sprintf("progress: %10s / %10s (%3.0f%%) avg: %10s/s cur: %10s/s",
+		compose.FormatBytesInt64(fetched),
+		compose.FormatBytesInt64(b.Descriptor.Size),
+		100*float64(fetched)/float64(b.Descriptor.Size),
+		compose.FormatBytesInt64(atomic.LoadInt64(&b.ReadSpeedAvg)),
+		compose.FormatBytesInt64(atomic.LoadInt64(&b.ReadSpeedCur)))
+	if fetched >= b.Descriptor.Size {
+		line += "; done at " + time.Now().UTC().Format(time.TimeOnly)
+	}
+	return line
+}

cmd/composectl/cmd/fetch_progress_test.go

@@ -0,0 +1,187 @@
+package composectl
+
+import (
+	"bytes"
+	"regexp"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/foundriesio/composeapp/pkg/compose"
+	"github.com/opencontainers/go-digest"
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
+)
+
+var (
+	ansiRe     = regexp.MustCompile("\x1b\\[[0-9;]*[A-Za-z]")
+	cursorUpRe = regexp.MustCompile("\x1b\\[([0-9]+)A")
+)
+
+// newProgressBlob builds a blob whose digest is unique per id, sized so progress
+// can be advanced across several frames.
+func newProgressBlob(t *testing.T, id string, size int64, start time.Time) *compose.BlobFetchProgress {
+	t.Helper()
+	dgst := digest.FromString(id)
+	return &compose.BlobFetchProgress{
+		BlobInfo: compose.BlobInfo{
+			Descriptor: &ocispec.Descriptor{Digest: dgst, Size: size},
+			Type:       compose.BlobTypeImageLayer,
+			State:      compose.BlobFetching,
+		},
+		FetchStartTime: start,
+	}
+}
+
+func shortDigest(b *compose.BlobFetchProgress) string {
+	enc := b.Descriptor.Digest.Encoded()
+	return enc[:12]
+}
+
+func snapshot(blobs ...*compose.BlobFetchProgress) *compose.FetchProgress {
+	m := compose.BlobsFetchProgress{}
+	for _, b := range blobs {
+		m[b.Descriptor.Digest] = b
+	}
+	return &compose.FetchProgress{Blobs: m}
+}
+
+// advance sets a blob's fetched bytes and flips its state to done once complete.
+func advance(b *compose.BlobFetchProgress, fetched int64) {
+	b.BytesFetched = fetched
+	if fetched >= b.Descriptor.Size {
+		b.State = compose.BlobOk
+	}
+}
+
+func doneLineCount(output string, b *compose.BlobFetchProgress) int {
+	count := 0
+	for _, line := range strings.Split(ansiRe.ReplaceAllString(output, ""), "\n") {
+		if strings.Contains(line, shortDigest(b)) && strings.Contains(line, "; done at") {
+			count++
+		}
+	}
+	return count
+}
+
+func anyLineFor(output string, b *compose.BlobFetchProgress) int {
+	count := 0
+	for _, line := range strings.Split(ansiRe.ReplaceAllString(output, ""), "\n") {
+		if strings.Contains(line, shortDigest(b)) {
+			count++
+		}
+	}
+	return count
+}
+
+// runFrames feeds a fixed multi-frame scenario into the renderer: blob A finishes
+// in frame 1, blob B spans all three frames, blob C appears in frame 2 and
+// finishes in frame 3.
+func runFrames(r *fetchProgressRenderer) (a, b, c *compose.BlobFetchProgress) {
+	t0 := time.Unix(0, 0).UTC()
+	a = &compose.BlobFetchProgress{
+		BlobInfo: compose.BlobInfo{
+			Descriptor: &ocispec.Descriptor{Digest: digest.FromString("blob-a"), Size: 10},
+			Type:       compose.BlobTypeImageManifest,
+			State:      compose.BlobOk,
+		},
+		FetchStartTime: t0,
+		BytesFetched:   10,
+	}
+	b = &compose.BlobFetchProgress{
+		BlobInfo: compose.BlobInfo{
+			Descriptor: &ocispec.Descriptor{Digest: digest.FromString("blob-b"), Size: 100},
+			Type:       compose.BlobTypeImageLayer,
+			State:      compose.BlobFetching,
+		},
+		FetchStartTime: t0.Add(time.Second),
+		BytesFetched:   30,
+	}
+	c = &compose.BlobFetchProgress{
+		BlobInfo: compose.BlobInfo{
+			Descriptor: &ocispec.Descriptor{Digest: digest.FromString("blob-c"), Size: 50},
+			Type:       compose.BlobTypeImageLayer,
+			State:      compose.BlobFetching,
+		},
+		FetchStartTime: t0.Add(2 * time.Second),
+	}
+
+	r.Render(snapshot(a, b))    // frame 1: A done, B at 30%
+	advance(b, 70)              // frame 2: B at 70%, C appears at 0%
+	r.Render(snapshot(a, b, c)) //
+	advance(b, 100)             // frame 3: B and C both complete
+	advance(c, 50)              //
+	r.Render(snapshot(a, b, c)) //
+	return a, b, c
+}
+
+func TestFetchProgressRendererPlainReportsProgressAndCompletion(t *testing.T) {
+	var out bytes.Buffer
+	r := NewFetchProgressRenderer(&out, false)
+	a, b, c := runFrames(r)
+
+	for _, blob := range []*compose.BlobFetchProgress{a, b, c} {
+		if got := doneLineCount(out.String(), blob); got != 1 {
+			t.Errorf("blob %s: expected exactly one completion line in non-TTY mode, got %d\n%s",
+				shortDigest(blob), got, out.String())
+		}
+	}
+	// A completes within its first frame: just the completion line.
+	// B is seen at 30% and 70% before completing: two progress lines + done.
+	// C is seen at 0% before completing: one progress line + done.
+	for blob, want := range map[*compose.BlobFetchProgress]int{a: 1, b: 3, c: 2} {
+		if got := anyLineFor(out.String(), blob); got != want {
+			t.Errorf("blob %s: expected %d printed lines in non-TTY mode, got %d\n%s",
+				shortDigest(blob), want, got, out.String())
+		}
+	}
+}
+
+func TestFetchProgressRendererTtyGraduatesEachBlobOnce(t *testing.T) {
+	var out bytes.Buffer
+	r := NewFetchProgressRenderer(&out, true)
+	a, b, c := runFrames(r)
+
+	for _, blob := range []*compose.BlobFetchProgress{a, b, c} {
+		if got := doneLineCount(out.String(), blob); got != 1 {
+			t.Errorf("blob %s: expected exactly one completion line in TTY mode, got %d\n%s",
+				shortDigest(blob), got, out.String())
+		}
+	}
+
+	// The live region (and thus every cursor-up move) must never exceed the peak
+	// number of concurrently in-flight blobs (here B and C in frame 2 => 2).
+	const maxConcurrent = 2
+	for _, m := range cursorUpRe.FindAllStringSubmatch(out.String(), -1) {
+		n, err := strconv.Atoi(m[1])
+		if err != nil {
+			t.Fatalf("failed to parse cursor-up count %q: %s", m[1], err)
+		}
+		if n > maxConcurrent {
+			t.Errorf("cursor moved up %d lines, exceeding the in-flight bound %d; "+
+				"the redraw region is growing unbounded again", n, maxConcurrent)
+		}
+	}
+}
+
+// Ensures the non-TTY decile reporting prints a line only when a blob enters a
+// new 10% bucket, and never marks an in-flight blob as completed.
+func TestFetchProgressRendererPlainReportsOncePerDecile(t *testing.T) {
+	var out bytes.Buffer
+	r := NewFetchProgressRenderer(&out, false)
+	b := newProgressBlob(t, "incomplete", 100, time.Unix(0, 0).UTC())
+
+	advance(b, 40)
+	r.Render(snapshot(b)) // first observation at 40% reports
+	advance(b, 49)
+	r.Render(snapshot(b)) // still in the 40% bucket: no new line
+	advance(b, 50)
+	r.Render(snapshot(b)) // crossed into the 50% bucket: reports
+
+	if got := anyLineFor(out.String(), b); got != 2 {
+		t.Errorf("expected one line per crossed decile (2), got %d:\n%s", got, out.String())
+	}
+	if got := doneLineCount(out.String(), b); got != 0 {
+		t.Errorf("expected no completion line for an in-flight blob, got %d:\n%s", got, out.String())
+	}
+}