deploy: add multi-writer for progress (#318)

There are still some minor visual glitches. See the examples for
reference.

## Image load (containerd):


[![asciicast](https://asciinema.org/a/W3br2Ttq2JrF5yQbCyUQJbNBe.svg)](https://asciinema.org/a/W3br2Ttq2JrF5yQbCyUQJbNBe)

## Image load (docker):


[![asciicast](https://asciinema.org/a/FpOJAaQzqa1oA43a67rDPDXpu.svg)](https://asciinema.org/a/FpOJAaQzqa1oA43a67rDPDXpu)


## Image push:


[![asciicast](https://asciinema.org/a/NjVjD8rI1QLaXC0Szx5w2Jo5d.svg)](https://asciinema.org/a/NjVjD8rI1QLaXC0Szx5w2Jo5d)

## Broken image push (minor visual glitch):


[![asciicast](https://asciinema.org/a/J0rA0oCswztkZsbSbDIFui3aw.svg)](https://asciinema.org/a/J0rA0oCswztkZsbSbDIFui3aw)

Author: malt3

img_tool/MODULE.bazel

@@ -27,4 +27,4 @@ go_sdk.from_file(
 
 go_deps = use_extension("@gazelle//:extensions.bzl", "go_deps")
 go_deps.from_file(go_mod = "//:go.mod")
-use_repo(go_deps, "com_github_aws_aws_sdk_go_v2", "com_github_aws_aws_sdk_go_v2_config", "com_github_aws_aws_sdk_go_v2_service_s3", "com_github_containerd_containerd_api", "com_github_containerd_platforms", "com_github_containerd_stargz_snapshotter_estargz", "com_github_google_uuid", "com_github_klauspost_compress", "com_github_klauspost_pgzip", "com_github_malt3_go_containerregistry", "com_github_opencontainers_go_digest", "com_github_opencontainers_image_spec", "com_github_schollz_progressbar_v3", "com_google_cloud_go_longrunning", "org_golang_google_genproto_googleapis_api", "org_golang_google_genproto_googleapis_bytestream", "org_golang_google_genproto_googleapis_rpc", "org_golang_google_grpc", "org_golang_google_protobuf", "org_golang_x_sync")
+use_repo(go_deps, "com_github_aws_aws_sdk_go_v2", "com_github_aws_aws_sdk_go_v2_config", "com_github_aws_aws_sdk_go_v2_service_s3", "com_github_containerd_containerd_api", "com_github_containerd_platforms", "com_github_containerd_stargz_snapshotter_estargz", "com_github_google_uuid", "com_github_jedib0t_go_pretty_v6", "com_github_klauspost_compress", "com_github_klauspost_pgzip", "com_github_malt3_go_containerregistry", "com_github_opencontainers_go_digest", "com_github_opencontainers_image_spec", "com_google_cloud_go_longrunning", "org_golang_google_genproto_googleapis_api", "org_golang_google_genproto_googleapis_bytestream", "org_golang_google_genproto_googleapis_rpc", "org_golang_google_grpc", "org_golang_google_protobuf", "org_golang_x_sync", "org_golang_x_term")

img_tool/go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/containerd/platforms v1.0.0-rc.2
 	github.com/containerd/stargz-snapshotter/estargz v0.18.1
 	github.com/google/uuid v1.6.0
+	github.com/jedib0t/go-pretty/v6 v6.7.5
 	github.com/klauspost/compress v1.18.2
 	github.com/klauspost/pgzip v1.2.6
 	github.com/malt3/go-containerregistry v0.0.0-20250724131542-7e98b20e9b45
@@ -48,17 +49,15 @@ require (
 	github.com/docker/cli v28.2.2+incompatible // indirect
 	github.com/docker/distribution v2.8.3+incompatible // indirect
 	github.com/docker/docker-credential-helpers v0.9.3 // indirect
-	github.com/mitchellh/colorstring v0.0.0-20190213212951-d06e56a500db // indirect
+	github.com/mattn/go-runewidth v0.0.16 // indirect
 	github.com/mitchellh/go-homedir v1.1.0 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/rivo/uniseg v0.4.7 // indirect
-	github.com/schollz/progressbar/v3 v3.18.0
 	github.com/sirupsen/logrus v1.9.3 // indirect
-	github.com/stretchr/testify v1.10.0 // indirect
 	github.com/vbatts/tar-split v0.12.2 // indirect
 	golang.org/x/net v0.46.1-0.20251013234738-63d1a5100f82 // indirect
 	golang.org/x/oauth2 v0.32.0 // indirect
 	golang.org/x/sys v0.37.0 // indirect
-	golang.org/x/term v0.36.0 // indirect
+	golang.org/x/term v0.36.0
 	golang.org/x/text v0.30.0 // indirect
 )

img_tool/go.sum

@@ -2,6 +2,7 @@ package docker
 
 import (
 	"archive/tar"
+	"context"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -58,7 +59,7 @@ func (t *TarWriter) WriteConfig(configData []byte) error {
 }
 
 // WriteLayer streams a layer to the tar
-func (t *TarWriter) WriteLayer(layerDigest registryv1.Hash, size int64, reader io.Reader) error {
+func (t *TarWriter) WriteLayer(ctx context.Context, layerDigest registryv1.Hash, size int64, reader io.Reader) error {
 	layerDir := layerDigest.Hex
 	layerPath := path.Join(layerDir, "layer.tar")
 
@@ -73,7 +74,7 @@ func (t *TarWriter) WriteLayer(layerDigest registryv1.Hash, size int64, reader i
 	// Create directory entry
 	if err := t.tw.WriteHeader(&tar.Header{
 		Name:     layerDir + "/",
-		Mode:     0755,
+		Mode:     0o755,
 		Typeflag: tar.TypeDir,
 	}); err != nil {
 		return fmt.Errorf("writing layer directory: %w", err)
@@ -87,15 +88,18 @@ func (t *TarWriter) WriteLayer(layerDigest registryv1.Hash, size int64, reader i
 	// Write layer content
 	hdr := &tar.Header{
 		Name: layerPath,
-		Mode: 0644,
+		Mode: 0o644,
 		Size: size,
 	}
 	if err := t.tw.WriteHeader(hdr); err != nil {
 		return fmt.Errorf("writing layer header: %w", err)
 	}
 
-	// Stream the layer content
-	pw := progress.Writer(size, layerDigest.Hex[:12])
+	// Stream the layer content with progress tracking
+	pw, err := progress.Writer(ctx, size, layerDigest.Hex[:12])
+	if err != nil {
+		return fmt.Errorf("creating progress writer: %w", err)
+	}
 	n, err := io.Copy(io.MultiWriter(t.tw, pw), reader)
 	if err != nil {
 		return fmt.Errorf("streaming layer content: %w", err)
@@ -134,7 +138,7 @@ func (t *TarWriter) Finalize() error {
 func (t *TarWriter) writeFile(name string, data []byte) error {
 	hdr := &tar.Header{
 		Name: name,
-		Mode: 0644,
+		Mode: 0o644,
 		Size: int64(len(data)),
 	}
 	if err := t.tw.WriteHeader(hdr); err != nil {

img_tool/pkg/docker/stream.go

@@ -154,7 +154,12 @@ func storeBlob(ctx context.Context, store containerd.Store, desc ocispec.Descrip
 	// Check if the blob already exists
 	info, err := store.Info(ctx, desc.Digest)
 	if err == nil && info.Digest == desc.Digest && info.Size == desc.Size {
-		// Blob already exists with correct size, nothing to do
+		// Blob already exists with correct size, nothing to do.
+		// We still print a progress writer for consistency.
+		if err := progress.CompletedWriter(ctx, desc.Size, desc.Digest.Hex()[:12]); err != nil {
+			return fmt.Errorf("creating completed progress writer: %w", err)
+		}
+
 		return nil
 	}
 
@@ -173,7 +178,10 @@ func storeBlob(ctx context.Context, store containerd.Store, desc ocispec.Descrip
 	}
 	defer reader.Close()
 
-	pw := progress.Writer(desc.Size, desc.Digest.Hex()[:12])
+	pw, err := progress.Writer(ctx, desc.Size, desc.Digest.Hex()[:12])
+	if err != nil {
+		return fmt.Errorf("creating progress bar: %w", err)
+	}
 	if _, err := io.Copy(io.MultiWriter(bufferedWriter, pw), reader); err != nil {
 		return fmt.Errorf("copying data to writer: %w", err)
 	}

img_tool/pkg/load/load.go

@@ -16,6 +16,7 @@ import (
 	"github.com/bazel-contrib/rules_img/img_tool/pkg/api"
 	"github.com/bazel-contrib/rules_img/img_tool/pkg/containerd"
 	"github.com/bazel-contrib/rules_img/img_tool/pkg/docker"
+	"github.com/bazel-contrib/rules_img/img_tool/pkg/progress"
 )
 
 type builder struct {
@@ -106,6 +107,23 @@ func (l *loader) LoadAll(ctx context.Context, ops []api.IndexedLoadDeployOperati
 
 			ctx = containerd.WithLease(ctx, lease)
 
+			// Setup progress tracking for concurrent blob uploads
+			ctx, stopProgress := progress.InitProgress(ctx, "loaded")
+			defer stopProgress()
+
+			// Pre-declare all trackers in deterministic order
+			blobNames := make([]string, len(blobs))
+			blobSizes := make([]int64, len(blobs))
+			for i, blob := range blobs {
+				digest, _ := blob.layer.Digest()
+				blobNames[i] = digest.Hex[:12]
+				blobSizes[i], err = blob.layer.Size()
+				if err != nil {
+					return nil, fmt.Errorf("getting size of blob %s: %w", digest.String(), err)
+				}
+			}
+			ctx = progress.DeclareTrackers(ctx, blobNames, blobSizes)
+
 			// Load all blobs in parallel...
 			contentStore := client.ContentStore()
 			if err := uploadBlobsParallel(ctx, contentStore, blobs, defaultWorkers); err != nil {
@@ -186,6 +204,10 @@ func (l *loader) loadContainerd(ctx context.Context, op api.IndexedLoadDeployOpe
 }
 
 func (l *loader) loadViaDocker(ctx context.Context, op api.IndexedLoadDeployOperation) ([]string, error) {
+	// Setup progress tracking for layer streaming
+	ctx, stopProgress := progress.InitProgress(ctx, "loaded")
+	defer stopProgress()
+
 	// Create a pipe to stream the tar to docker load
 	pr, pw := io.Pipe()
 
@@ -314,6 +336,16 @@ func (l *loader) streamManifestToTar(ctx context.Context, manifestInfo api.Manif
 		tw.SetTags(normalizedTags)
 	}
 
+	// Pre-declare trackers for all layers in order
+	layerNames := make([]string, len(manifestInfo.LayerBlobs))
+	layerSizes := make([]int64, len(manifestInfo.LayerBlobs))
+	for i, layerDesc := range manifestInfo.LayerBlobs {
+		digest, _ := registryv1.NewHash(layerDesc.Digest)
+		layerNames[i] = digest.Hex[:12]
+		layerSizes[i] = layerDesc.Size
+	}
+	ctx = progress.DeclareTrackers(ctx, layerNames, layerSizes)
+
 	// Stream layers
 	if err := l.streamLayers(ctx, manifestInfo, tw); err != nil {
 		return nil, fmt.Errorf("streaming layers: %w", err)
@@ -350,7 +382,7 @@ func (l *loader) streamLayers(ctx context.Context, manifestInfo api.ManifestDepl
 		}
 		defer rc.Close()
 
-		if err := tw.WriteLayer(digest, layerDesc.Size, rc); err != nil {
+		if err := tw.WriteLayer(ctx, digest, layerDesc.Size, rc); err != nil {
 			return err
 		}
 	}

img_tool/pkg/load/loader.go

@@ -5,5 +5,8 @@ go_library(
     srcs = ["progress.go"],
     importpath = "github.com/bazel-contrib/rules_img/img_tool/pkg/progress",
     visibility = ["//visibility:public"],
-    deps = ["@com_github_schollz_progressbar_v3//:progressbar"],
+    deps = [
+        "@com_github_jedib0t_go_pretty_v6//progress",
+        "@org_golang_x_term//:term",
+    ],
 )