From d89fb1717d3631cf19159a76c238f3826270a0c6 Mon Sep 17 00:00:00 2001 From: Mike Sul Date: Tue, 9 Jun 2026 12:26:03 +0200 Subject: [PATCH 1/5] refactor(compose): extract single-blob fetch into helper FetchBlobs inlined the whole per-blob download path inside an anonymous function, leaving no seam to reuse it. The upcoming concurrent worker pool needs to invoke that exact path from each goroutine, so it is lifted into fetchSingleBlob. This is a pure refactor: the sequential loop and observable behavior are unchanged. Signed-off-by: Mike Sul Assisted-by: Claude Code:claude-opus-4-8 --- pkg/compose/fetch.go | 47 ++++++++++++++++++++++---------------------- 1 file changed, 24 insertions(+), 23 deletions(-) diff --git a/pkg/compose/fetch.go b/pkg/compose/fetch.go index e24369e..01d4c11 100644 --- a/pkg/compose/fetch.go +++ b/pkg/compose/fetch.go @@ -165,29 +165,7 @@ func FetchBlobs(ctx context.Context, cfg *Config, blobs BlobsInfo, options ...Fe } for _, bi := range getOrderedBlobsToFetch(blobsToFetch) { - err = func() error { - // Get the reader without digest calculation and verification because the writer/ingester of - // the local store (`ls`) will do that. - //r, err := blobProvider.GetReadCloser(ctx, WithRef(bi.Ref()), WithDescriptor(*bi.Descriptor)) - r, err := blobProvider.GetReadCloser(ctx, WithRef(bi.Ref()), WithDescriptor(*bi.Descriptor), WithSecureReadOff()) - if err != nil { - return fmt.Errorf("failed to initiate request to fetch blob %s: %w", bi.Descriptor.Digest, err) - } - defer r.Close() - blobReader, ok := r.(io.ReadSeekCloser) - if !ok { - return fmt.Errorf("blob fetch reader for %s does not implement io.ReadSeekCloser", bi.Ref()) - } - bi.FetchStartTime = time.Now() - rm := NewReadMonitor(ctx, blobReader, bi) - rm.Start() - defer rm.Stop() - if err := CopyBlob(ctx, rm, bi.Ref(), *bi.Descriptor, ls, true); err != nil { - return fmt.Errorf("failed to fetch blob %s: %w", bi.Descriptor.Digest, err) - } - return nil - }() - if err != nil { + if err = fetchSingleBlob(ctx, blobProvider, ls, bi); err != nil { break } } @@ -205,6 +183,29 @@ func FetchBlobs(ctx context.Context, cfg *Config, blobs BlobsInfo, options ...Fe return ctx.Err() } +// fetchSingleBlob downloads one blob into the local store. The reader is created +// without digest calculation/verification because the local store ingester (`ls`) +// performs that during the copy. +func fetchSingleBlob(ctx context.Context, blobProvider BlobProvider, ls content.Store, bi *BlobFetchProgress) error { + r, err := blobProvider.GetReadCloser(ctx, WithRef(bi.Ref()), WithDescriptor(*bi.Descriptor), WithSecureReadOff()) + if err != nil { + return fmt.Errorf("failed to initiate request to fetch blob %s: %w", bi.Descriptor.Digest, err) + } + defer r.Close() + blobReader, ok := r.(io.ReadSeekCloser) + if !ok { + return fmt.Errorf("blob fetch reader for %s does not implement io.ReadSeekCloser", bi.Ref()) + } + bi.FetchStartTime = time.Now() + rm := NewReadMonitor(ctx, blobReader, bi) + rm.Start() + defer rm.Stop() + if err := CopyBlob(ctx, rm, bi.Ref(), *bi.Descriptor, ls, true); err != nil { + return fmt.Errorf("failed to fetch blob %s: %w", bi.Descriptor.Digest, err) + } + return nil +} + func checkAndUpdateBlobStatus(ctx context.Context, fetchProgress *FetchProgress, ls content.Store, sr progress.Reporter[FetchProgress]) { for _, b := range fetchProgress.Blobs { if b.State == BlobOk { From 97fc916613e2d1c273c1664d60836728a0fa6cb1 Mon Sep 17 00:00:00 2001 From: Mike Sul Date: Tue, 9 Jun 2026 12:32:39 +0200 Subject: [PATCH 2/5] feat(compose): pull blobs via bounded concurrent worker pool Blobs were fetched strictly one at a time, so on high-latency links or apps with many small blobs the transfer was bottlenecked by round-trip latency rather than bandwidth. FetchBlobs now drives the downloads through a bounded errgroup pool so several blobs transfer at once while capping how many run together. The pool is fail-fast: the first error cancels the derived context, aborting in-flight and queued downloads, and that error is returned. Each blob is owned by exactly one worker and the shared progress counters stay single-writer, so no new shared mutable state is introduced. Worker count is configurable via WithFetchWorkers and defaults to DefaultFetchWorkers (3) when unset. Signed-off-by: Mike Sul Assisted-by: Claude Code:claude-opus-4-8 --- go.mod | 2 +- pkg/compose/fetch.go | 31 +++++- pkg/compose/fetch_test.go | 215 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 243 insertions(+), 5 deletions(-) create mode 100644 pkg/compose/fetch_test.go diff --git a/go.mod b/go.mod index 5293d28..967fcc8 100644 --- a/go.mod +++ b/go.mod @@ -19,6 +19,7 @@ require ( github.com/spf13/cobra v1.8.0 go.etcd.io/bbolt v1.3.7 golang.org/x/net v0.26.0 + golang.org/x/sync v0.7.0 gopkg.in/yaml.v3 v3.0.1 ) @@ -71,7 +72,6 @@ require ( go.opentelemetry.io/otel/trace v1.21.0 // indirect golang.org/x/exp v0.0.0-20230713183714-613f0c0eb8a1 // indirect golang.org/x/mod v0.17.0 // indirect - golang.org/x/sync v0.7.0 // indirect golang.org/x/sys v0.29.0 // indirect golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d // indirect diff --git a/pkg/compose/fetch.go b/pkg/compose/fetch.go index 01d4c11..59cdc2c 100644 --- a/pkg/compose/fetch.go +++ b/pkg/compose/fetch.go @@ -9,6 +9,7 @@ import ( "github.com/containerd/containerd/errdefs" "github.com/foundriesio/composeapp/internal/progress" "github.com/opencontainers/go-digest" + "golang.org/x/sync/errgroup" "io" "sync" "sync/atomic" @@ -17,6 +18,7 @@ import ( const ( DefaultPollInterval = 300 // Default interval between polling/checking blob download status in milliseconds + DefaultFetchWorkers = 3 // Default number of concurrent blob-fetch workers ) type ( @@ -41,6 +43,7 @@ type ( ProgressHandler FetchProgressFunc ProgressPollInterval int // interval between polling/checking blob download status in milliseconds SourcePath string // path to the source directory containing blobs to fetch, if specified, the blobs will be fetched from this directory instead of remote registry + Workers int // number of concurrent blob-fetch workers; defaults to DefaultFetchWorkers when not positive } FetchOption func(*FetchOptions) @@ -79,6 +82,12 @@ func WithSourcePath(sourcePath string) FetchOption { } } +func WithFetchWorkers(workers int) FetchOption { + return func(opts *FetchOptions) { + opts.Workers = workers + } +} + func FetchBlobs(ctx context.Context, cfg *Config, blobs BlobsInfo, options ...FetchOption) error { opts := FetchOptions{} for _, o := range options { @@ -164,11 +173,11 @@ func FetchBlobs(ctx context.Context, cfg *Config, blobs BlobsInfo, options ...Fe }(stopChan) } - for _, bi := range getOrderedBlobsToFetch(blobsToFetch) { - if err = fetchSingleBlob(ctx, blobProvider, ls, bi); err != nil { - break - } + workers := opts.Workers + if workers <= 0 { + workers = DefaultFetchWorkers } + err = runBlobFetchPool(ctx, blobProvider, ls, getOrderedBlobsToFetch(blobsToFetch), workers) if progressReporter != nil { if ctx.Err() == nil { @@ -183,6 +192,20 @@ func FetchBlobs(ctx context.Context, cfg *Config, blobs BlobsInfo, options ...Fe return ctx.Err() } +// runBlobFetchPool downloads the given blobs concurrently using a bounded pool of +// workers. It is fail-fast: the first error cancels the derived context so that +// any in-flight or queued downloads stop, and that error is returned. +func runBlobFetchPool(ctx context.Context, blobProvider BlobProvider, ls content.Store, blobs []*BlobFetchProgress, workers int) error { + g, gctx := errgroup.WithContext(ctx) + g.SetLimit(workers) + for _, bi := range blobs { + g.Go(func() error { + return fetchSingleBlob(gctx, blobProvider, ls, bi) + }) + } + return g.Wait() +} + // fetchSingleBlob downloads one blob into the local store. The reader is created // without digest calculation/verification because the local store ingester (`ls`) // performs that during the copy. diff --git a/pkg/compose/fetch_test.go b/pkg/compose/fetch_test.go new file mode 100644 index 0000000..3be6c7e --- /dev/null +++ b/pkg/compose/fetch_test.go @@ -0,0 +1,215 @@ +package compose + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "sync" + "testing" + "time" + + "github.com/containerd/containerd/content" + "github.com/containerd/containerd/content/local" + "github.com/opencontainers/go-digest" + ocispec "github.com/opencontainers/image-spec/specs-go/v1" +) + +// errInjected is the sentinel a fakeBlobProvider wraps when asked to fail a blob, +// so tests can assert error propagation with errors.Is. +var errInjected = errors.New("injected fetch failure") + +// fakeReadSeekCloser serves blob bytes from memory and, on the first read, sleeps +// for `delay` to widen the window during which a fetch is "in flight" so the test +// can observe real overlap between concurrent workers. +type fakeReadSeekCloser struct { + *bytes.Reader + delay time.Duration + slept bool + onClose func() +} + +func (f *fakeReadSeekCloser) Read(p []byte) (int, error) { + if !f.slept { + f.slept = true + if f.delay > 0 { + time.Sleep(f.delay) + } + } + return f.Reader.Read(p) +} + +func (f *fakeReadSeekCloser) Close() error { + if f.onClose != nil { + f.onClose() + } + return nil +} + +// fakeBlobProvider is an in-memory BlobProvider that tracks how many fetches are +// concurrently in flight and can inject a failure for a specific digest. +type fakeBlobProvider struct { + blobs map[digest.Digest][]byte + failDigest digest.Digest + readDelay time.Duration + + mu sync.Mutex + active int + maxActive int +} + +func (p *fakeBlobProvider) Type() BlobProviderType { return BlobProviderTypeMemory } + +func (p *fakeBlobProvider) Info(_ context.Context, _ digest.Digest) (content.Info, error) { + return content.Info{}, fmt.Errorf("not implemented") +} + +func (p *fakeBlobProvider) GetReadCloser(_ context.Context, opts ...SecureReadOptions) (io.ReadCloser, error) { + dgst := GetSecureReadParams(opts...).Descriptor.Digest + if p.failDigest != "" && dgst == p.failDigest { + return nil, fmt.Errorf("blob %s: %w", dgst, errInjected) + } + data, ok := p.blobs[dgst] + if !ok { + return nil, fmt.Errorf("blob %s not found", dgst) + } + p.enter() + return &fakeReadSeekCloser{ + Reader: bytes.NewReader(data), + delay: p.readDelay, + onClose: p.leave, + }, nil +} + +func (p *fakeBlobProvider) enter() { + p.mu.Lock() + defer p.mu.Unlock() + p.active++ + if p.active > p.maxActive { + p.maxActive = p.active + } +} + +func (p *fakeBlobProvider) leave() { + p.mu.Lock() + defer p.mu.Unlock() + p.active-- +} + +func (p *fakeBlobProvider) maxConcurrency() int { + p.mu.Lock() + defer p.mu.Unlock() + return p.maxActive +} + +func newTestStore(t *testing.T) content.Store { + t.Helper() + ls, err := local.NewStore(t.TempDir()) + if err != nil { + t.Fatalf("failed to create local store: %s", err) + } + return ls +} + +// newTestBlob builds a fetchable blob descriptor whose digest matches its bytes so +// the local store's digest verification during CopyBlob succeeds. +func newTestBlob(data []byte) *BlobFetchProgress { + dgst := digest.FromBytes(data) + desc := &ocispec.Descriptor{ + MediaType: "application/octet-stream", + Digest: dgst, + Size: int64(len(data)), + URLs: []string{"fake://" + dgst.String()}, + } + return &BlobFetchProgress{ + BlobInfo: BlobInfo{ + Descriptor: desc, + State: BlobMissing, + Type: BlobTypeImageLayer, + }, + } +} + +func TestRunBlobFetchPoolRespectsWorkerLimit(t *testing.T) { + const ( + workers = 3 + numBlobs = 9 + ) + provider := &fakeBlobProvider{blobs: map[digest.Digest][]byte{}, readDelay: 100 * time.Millisecond} + var blobs []*BlobFetchProgress + for i := 0; i < numBlobs; i++ { + data := []byte(fmt.Sprintf("limit-blob-%d-payload", i)) + provider.blobs[digest.FromBytes(data)] = data + blobs = append(blobs, newTestBlob(data)) + } + + ls := newTestStore(t) + if err := runBlobFetchPool(context.Background(), provider, ls, blobs, workers); err != nil { + t.Fatalf("runBlobFetchPool returned an unexpected error: %s", err) + } + + // The pool must never run more than `workers` fetches at once, yet with more + // blobs than workers it must saturate the pool to exactly `workers`. + if got := provider.maxConcurrency(); got != workers { + t.Errorf("expected peak concurrency to equal the worker limit %d, got %d", workers, got) + } + + for _, b := range blobs { + if _, err := ls.Info(context.Background(), b.Descriptor.Digest); err != nil { + t.Errorf("blob %s missing from store after fetch: %s", b.Descriptor.Digest, err) + } + } +} + +func TestRunBlobFetchPoolFailsFast(t *testing.T) { + provider := &fakeBlobProvider{blobs: map[digest.Digest][]byte{}} + var blobs []*BlobFetchProgress + for i := 0; i < 5; i++ { + data := []byte(fmt.Sprintf("failfast-blob-%d", i)) + provider.blobs[digest.FromBytes(data)] = data + blobs = append(blobs, newTestBlob(data)) + } + // Inject a failure for one of the blobs. + provider.failDigest = blobs[2].Descriptor.Digest + + ls := newTestStore(t) + err := runBlobFetchPool(context.Background(), provider, ls, blobs, 2) + if err == nil { + t.Fatal("expected runBlobFetchPool to return an error, got nil") + } + if !errors.Is(err, errInjected) { + t.Errorf("expected returned error to wrap the injected failure, got: %s", err) + } + // The failing blob must never be committed to the store. + if _, infoErr := ls.Info(context.Background(), blobs[2].Descriptor.Digest); infoErr == nil { + t.Errorf("failing blob %s should not be present in the store", blobs[2].Descriptor.Digest) + } +} + +func TestRunBlobFetchPoolFetchesAllBlobs(t *testing.T) { + const workers = DefaultFetchWorkers + provider := &fakeBlobProvider{blobs: map[digest.Digest][]byte{}} + var blobs []*BlobFetchProgress + for i := 0; i < 5; i++ { + data := []byte(fmt.Sprintf("all-blob-%d-payload", i)) + provider.blobs[digest.FromBytes(data)] = data + blobs = append(blobs, newTestBlob(data)) + } + + ls := newTestStore(t) + if err := runBlobFetchPool(context.Background(), provider, ls, blobs, workers); err != nil { + t.Fatalf("runBlobFetchPool returned an unexpected error: %s", err) + } + + for _, b := range blobs { + info, err := ls.Info(context.Background(), b.Descriptor.Digest) + if err != nil { + t.Errorf("blob %s missing from store: %s", b.Descriptor.Digest, err) + continue + } + if info.Size != b.Descriptor.Size { + t.Errorf("blob %s stored size %d, want %d", b.Descriptor.Digest, info.Size, b.Descriptor.Size) + } + } +} From 19a0a8f8be687d434c5ddbfd43c025e7984efcc7 Mon Sep 17 00:00:00 2001 From: Mike Sul Date: Tue, 9 Jun 2026 12:43:42 +0200 Subject: [PATCH 3/5] feat(cli): add --workers flag to pull for concurrent fetch The concurrent fetch pool added to FetchBlobs had no user-facing control, so the parallelism was fixed and could not be tuned to a link or a registry's tolerance. This exposes it as a --workers/-w flag on the pull command, defaulting to 3. The value is validated up front against a strict 1-10 range, mirroring the existing watermark check, so an out-of-bounds argument fails with a clear message and a non-zero exit before any pull work starts. An e2e test covers both the range validation and a successful multi-worker pull. Signed-off-by: Mike Sul Assisted-by: Claude Code:claude-opus-4-8 --- cmd/composectl/cmd/pull.go | 20 ++++++++++- test/fixtures/composectl_cmds.go | 22 ++++++++++++ test/integration/concurrent_pull_test.go | 46 ++++++++++++++++++++++++ 3 files changed, 87 insertions(+), 1 deletion(-) create mode 100644 test/integration/concurrent_pull_test.go diff --git a/cmd/composectl/cmd/pull.go b/cmd/composectl/cmd/pull.go index 5df6b6b..70b833d 100644 --- a/cmd/composectl/cmd/pull.go +++ b/cmd/composectl/cmd/pull.go @@ -20,9 +20,16 @@ type ( SrcStorePath string PrintUsageStat bool Quick bool + Workers uint } ) +const ( + MinWorkers = 1 + MaxWorkers = 10 + DefaultWorkers = 3 +) + var ( exitCodeInsufficientSpace int = 100 ) @@ -41,14 +48,24 @@ func init() { pullCmd.Flags().StringVarP(&opts.SrcStorePath, "source-store-path", "l", "", "A path to the source store root directory") pullCmd.Flags().BoolVarP(&opts.PrintUsageStat, "print-usage-stat", "p", false, "A flag to enable/disable usage statistic output to stderr") pullCmd.Flags().BoolVar(&opts.Quick, "quick", false, "Skip checking hash of app blobs; verify only their presence and size") + pullCmd.Flags().UintVarP(&opts.Workers, "workers", "w", DefaultWorkers, + fmt.Sprintf("Number of concurrent blob download workers in range %d-%d", MinWorkers, MaxWorkers)) pullCmd.Run = func(cmd *cobra.Command, args []string) { checkWatermark(opts.UsageWatermark) + checkWorkers(opts.Workers) pullApps(cmd, args, &opts) } rootCmd.AddCommand(pullCmd) } +func checkWorkers(workers uint) { + if workers < MinWorkers || workers > MaxWorkers { + DieNotNilWithCode(fmt.Errorf("invalid `--workers` value: %d; should be between %d and %d", + workers, MinWorkers, MaxWorkers), 1, "invalid argument") + } +} + func pullApps(cmd *cobra.Command, args []string, opts *pullOptions) { if len(args) > 1 { fmt.Printf("Pulling %d apps to %s\n", len(args), config.StoreRoot) @@ -78,7 +95,8 @@ func pullApps(cmd *cobra.Command, args []string, opts *pullOptions) { err := compose.FetchBlobs(cmd.Context(), config, cr.MissingBlobs, compose.WithProgressPollInterval(1000), compose.WithFetchProgress(getFetchProgressHandler()), - compose.WithSourcePath(opts.SrcStorePath)) + compose.WithSourcePath(opts.SrcStorePath), + compose.WithFetchWorkers(int(opts.Workers))) DieNotNil(err, "failed to fetch blobs") fmt.Println("\n\nApp blobs pull completed at " + time.Now().UTC().Format("15:04:05 02 Jan 2006")) } diff --git a/test/fixtures/composectl_cmds.go b/test/fixtures/composectl_cmds.go index c07472d..3a66b4b 100644 --- a/test/fixtures/composectl_cmds.go +++ b/test/fixtures/composectl_cmds.go @@ -17,6 +17,7 @@ import ( "os/exec" "path" "path/filepath" + "strconv" "testing" "time" ) @@ -192,6 +193,11 @@ func (a *App) Pull(t *testing.T) { a.runCmd(t, "pull app", "pull", a.PublishedUri, "-u", "90") } +func (a *App) PullWithWorkers(t *testing.T, workers int) { + t.Helper() + a.runCmd(t, "pull app", "pull", a.PublishedUri, "-u", "90", "-w", strconv.Itoa(workers)) +} + func (a *App) Remove(t *testing.T) { t.Helper() a.runCmd(t, "remove app", "rm", a.PublishedUri) @@ -444,6 +450,22 @@ func runCmd(t *testing.T, appDir string, args ...string) []byte { return output } +// RunCmdExpectFail runs composectl with the given args and fails the test unless +// the command exits with a non-zero status. It returns the combined output so the +// caller can assert on the user-facing error message. +func RunCmdExpectFail(t *testing.T, appDir string, args ...string) []byte { + t.Helper() + c := exec.Command(composeExec, args...) + if len(appDir) > 0 { + c.Dir = appDir + } + output, err := c.CombinedOutput() + if err == nil { + t.Fatalf("expected `%s` command to fail, but it succeeded; output: %s\n", args[0], output) + } + return output +} + func randomString(length int) string { const charset = "abcdefghijklmnopqrstuvwxyz0123456789" seededRand := rand2.New(rand2.NewSource(time.Now().UnixNano())) diff --git a/test/integration/concurrent_pull_test.go b/test/integration/concurrent_pull_test.go new file mode 100644 index 0000000..c522e6e --- /dev/null +++ b/test/integration/concurrent_pull_test.go @@ -0,0 +1,46 @@ +package e2e_tests + +import ( + "strings" + "testing" + + f "github.com/foundriesio/composeapp/test/fixtures" +) + +// TestPullWorkersFlagValidation asserts that out-of-range --workers values are +// rejected with a clear, user-facing error before any pull work begins. +func TestPullWorkersFlagValidation(t *testing.T) { + for _, tc := range []struct { + name string + value string + }{ + {"below minimum", "0"}, + {"above maximum", "11"}, + } { + t.Run(tc.name, func(t *testing.T) { + out := f.RunCmdExpectFail(t, "", + "pull", "registry:5000/factory/does-not-exist:latest", "-w", tc.value) + msg := string(out) + if !strings.Contains(msg, "invalid `--workers` value") || !strings.Contains(msg, "between 1 and 10") { + t.Fatalf("expected an --workers range validation error, got: %s", msg) + } + }) + } +} + +// TestPullWithConcurrentWorkers verifies the --workers flag plumbs through to the +// concurrent fetch and that an app pulled with multiple workers is fully fetched. +func TestPullWithConcurrentWorkers(t *testing.T) { + appComposeDef := ` +services: + busybox: + image: ghcr.io/foundriesio/busybox:1.36 + command: sh -c "while true; do sleep 60; done" +` + app := f.NewApp(t, appComposeDef) + app.Publish(t) + defer app.Remove(t) + + app.PullWithWorkers(t, 5) + app.CheckFetched(t) +} From 9345c38bc31fd55a85fed6dc980123d06565d72e Mon Sep 17 00:00:00 2001 From: Mike Sul Date: Tue, 9 Jun 2026 13:09:04 +0200 Subject: [PATCH 4/5] build(dev): add cgo toolchain and run unit tests with -race The dev container image is Alpine-based and shipped no C compiler, so CGO resolved to disabled and the Go race detector could not run. That left the new concurrent blob-fetch worker pool without automated data-race verification. Install gcc and musl-dev in the dev image and enable CGO_ENABLED=1 with -race in the test-unit target so every unit run is race-instrumented. The cgo setting is scoped to the test-unit recipe line, leaving the static production build (which is built with cgo off elsewhere) unchanged. Signed-off-by: Mike Sul Assisted-by: Claude Code:claude-opus-4-8 --- Makefile | 2 +- test/compose/Dockerfile | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Makefile b/Makefile index 4852220..f6f5baf 100644 --- a/Makefile +++ b/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 $@ diff --git a/test/compose/Dockerfile b/test/compose/Dockerfile index e5bd10b..1996fae 100644 --- a/test/compose/Dockerfile +++ b/test/compose/Dockerfile @@ -3,7 +3,7 @@ FROM golang:1.24-alpine ARG REG_CERT=test/compose/registry/certs/registry.crt ARG SRC_DIR=$PWD -RUN apk add make curl docker docker-compose git python3 py3-requests py3-expandvars py3-yaml skopeo golangci-lint +RUN apk add make curl docker docker-compose git python3 py3-requests py3-expandvars py3-yaml skopeo golangci-lint gcc musl-dev # fetch ci-scripts, needed for calculating sizes of extacted app layers RUN git clone https://github.com/foundriesio/ci-scripts.git /ci-scripts From ecf1265f8b04c5cb6b7b5540366f77fee52fbf41 Mon Sep 17 00:00:00 2001 From: Mike Sul Date: Wed, 10 Jun 2026 19:27:51 +0200 Subject: [PATCH 5/5] fix(cli): render pull blob progress one line per blob The pull progress handler tracked every blob in a single redraw region and moved the cursor up by its full length each frame. Because completed blobs were never dropped, that region grew to the total blob count; once it exceeded the terminal height the cursor-up clamped at the top of the screen and bottom-row newlines scrolled it, desynchronizing the redraw. In-flight blobs were then reprinted on fresh lines, so a single blob appeared on several lines. Bound the live region to only the in-flight blobs and graduate each blob to one permanent line when it completes, so the region can never outgrow the screen and history scrolling is harmless. Each in-flight blob thus owns exactly one in-place updated line. Without a TTY cursor control is meaningless, so output is append-only instead: each blob gets a line when it first appears and whenever it crosses another 10% of its size, plus one final completion line. Previously only the completion line was printed, so a pull dominated by one large layer looked hung for minutes in log collectors such as journalctl. The shared line formatter now reads the cross-goroutine byte counters via atomic loads, removing a latent race with the concurrent fetch workers. Signed-off-by: Mike Sul Assisted-by: Claude Code:claude-fable-5 --- cmd/composectl/cmd/fetch_progress.go | 159 ++++++++++++++++++ cmd/composectl/cmd/fetch_progress_test.go | 187 ++++++++++++++++++++++ cmd/composectl/cmd/pull.go | 59 +------ 3 files changed, 347 insertions(+), 58 deletions(-) create mode 100644 cmd/composectl/cmd/fetch_progress.go create mode 100644 cmd/composectl/cmd/fetch_progress_test.go diff --git a/cmd/composectl/cmd/fetch_progress.go b/cmd/composectl/cmd/fetch_progress.go new file mode 100644 index 0000000..6ed16fb --- /dev/null +++ b/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 +} diff --git a/cmd/composectl/cmd/fetch_progress_test.go b/cmd/composectl/cmd/fetch_progress_test.go new file mode 100644 index 0000000..2d3f86c --- /dev/null +++ b/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()) + } +} diff --git a/cmd/composectl/cmd/pull.go b/cmd/composectl/cmd/pull.go index 70b833d..5218f73 100644 --- a/cmd/composectl/cmd/pull.go +++ b/cmd/composectl/cmd/pull.go @@ -4,7 +4,6 @@ import ( "encoding/json" "fmt" "os" - "sync/atomic" "time" "github.com/containerd/containerd/platforms" @@ -108,62 +107,6 @@ func pullApps(cmd *cobra.Command, args []string, opts *pullOptions) { } func getFetchProgressHandler() func(progress *compose.FetchProgress) { - var blobsBeingFetched []struct { - blob *compose.BlobFetchProgress - done bool - } isTty := term.IsTerminal(os.Stdout.Fd()) || os.Getenv("PARENT_HAS_TTY") == "1" - return func(p *compose.FetchProgress) { - currentBlobsBeingFetchedNumb := len(blobsBeingFetched) - for _, bi := range p.Blobs { - found := false - for _, bf := range blobsBeingFetched { - if bf.blob.Descriptor.Digest == bi.Descriptor.Digest { - found = true - break - } - } - if !found && (bi.State == compose.BlobFetching || bi.State == compose.BlobOk) { - blobsBeingFetched = append(blobsBeingFetched, struct { - blob *compose.BlobFetchProgress - done bool - }{blob: bi, done: false}) - } - } - - if isTty && currentBlobsBeingFetchedNumb > 0 { - fmt.Printf("\033[%dA\r", currentBlobsBeingFetchedNumb) - } - - for i := range blobsBeingFetched { - if blobsBeingFetched[i].done { - if isTty { - // Move cursor down one line since this blob fetch is completed - fmt.Print("\033[1B") - } - continue - } - - b := blobsBeingFetched[i].blob - - fmt.Printf("\n [%-12s] %.12s start: %s from: %10s ", - b.Type, - b.Descriptor.Digest.Encoded(), - b.FetchStartTime.UTC().Format(time.TimeOnly), - compose.FormatBytesInt64(b.BlobInfo.BytesFetched)) - - fmt.Printf("progress: %10s / %10s (%3.0f%%) avg: %10s/s cur: %10s/s", - compose.FormatBytesInt64(atomic.LoadInt64(&b.BytesFetched)), - compose.FormatBytesInt64(b.Descriptor.Size), - 100*float64(b.BytesFetched)/float64(b.Descriptor.Size), - compose.FormatBytesInt64(atomic.LoadInt64(&b.ReadSpeedAvg)), - compose.FormatBytesInt64(atomic.LoadInt64(&b.ReadSpeedCur))) - - if b.BytesFetched == b.Descriptor.Size { - fmt.Printf("; done at %s", - time.Now().UTC().Format(time.TimeOnly)) - blobsBeingFetched[i].done = true - } - } - } + return NewFetchProgressRenderer(os.Stdout, isTty).Render }