[CELL-297] nix-cache push uploads all chunks in parallel — eliminates per-chunk manifest churn that serialized 29 blob uploads on ultimate images

- feat(nixstore): chunked push splits tar to temp files then uploads all layers via single remote.Write with 4 parallel blob streams — ultimate image push no longer serializes 29 manifest writes + 29 re-fetches, cutting wall-clock time significantly
- feat(nixstore): stall detector recognizes parallel upload finalization via TotalSizeHint byte comparison — no false stall abort while registry commits the single manifest after all blobs land
- feat(nixstore): UploadJobs package var (default 4) controls parallel blob concurrency — tunable without code changes
- docs(cli): nix-store push help text updated to describe parallel upload behavior — no functional change

Author: DimmKirr

cmd/nix_store.go

@@ -47,9 +47,10 @@ OCI tar+gzip layers atop --base, tagged --image, to the destination
 registry.
 
 By default, the tar is split on entry boundaries into ~512 MB chunks,
-each uploaded as a separate layer. This reduces per-layer upload size,
-making GHCR throttling less likely and enabling partial retry on
-failure (CELL-297). Set DEVCELL_NIX_CHUNK_SIZE=0 to disable chunking.
+buffered to temp files, then uploaded in parallel (4 concurrent blob
+streams, 1 manifest write). This eliminates per-chunk manifest churn
+and overlaps gzip+upload across layers (CELL-297). Set
+DEVCELL_NIX_CHUNK_SIZE=0 to disable chunking.
 
 Each layer is SINGLE-gzipped — this function is the deterministic
 alternative to ` + "`crane append --new_layer -`" + ` which re-gzipped

internal/nixstore/push.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/google/go-containerregistry/pkg/authn"
 	"github.com/google/go-containerregistry/pkg/name"
+	v1 "github.com/google/go-containerregistry/pkg/v1"
 	"github.com/google/go-containerregistry/pkg/v1/mutate"
 	"github.com/google/go-containerregistry/pkg/v1/remote"
 	"github.com/google/go-containerregistry/pkg/v1/stream"
@@ -34,6 +35,11 @@ var ProgressTick = 5 * time.Second
 // disable chunking (single layer, original behavior).
 var ChunkSize int64 = 512 << 20 // 512 MB
 
+// UploadJobs controls how many OCI layer blobs are uploaded concurrently
+// by remote.Write. Higher values improve throughput but risk GHCR
+// throttling; 16 caused session kills, 4 is a safe default.
+var UploadJobs = 4
+
 // StallTimeout is the maximum duration Push tolerates zero byte
 // progress before aborting the upload. Set to 0 to disable. GHCR
 // occasionally throttles multi-GB layer uploads to near-zero; without
@@ -126,11 +132,11 @@ func fmtBytes(n uint64) string {
 // more OCI tar+gzip layers atop baseRef.
 //
 // When ChunkSize > 0 (default 512 MB), the input tar is split on entry
-// boundaries into ~ChunkSize chunks. Each chunk is pushed as a separate
-// OCI layer via its own remote.Write call — one upload at a time. After
-// each chunk the manifest is re-fetched so previous layers are referenced
-// by digest and skipped on subsequent pushes. No temp files; bytes flow
-// stdin → pipe → gzip → registry.
+// boundaries into ~ChunkSize temp files. All chunks are then uploaded
+// in parallel (UploadJobs concurrent blob streams, default 4) via a
+// single remote.Write call — one manifest write at the end. This
+// eliminates per-chunk manifest churn and overlaps gzip+upload across
+// layers. Temp files are removed after the push completes.
 //
 // When ChunkSize <= 0, the original single-layer streaming behavior
 // is preserved: bytes flow r → gzip → registry with no disk staging.
@@ -192,8 +198,9 @@ func Push(ctx context.Context, baseRef, dstRef string, r io.ReadCloser) error {
 				if StallTimeout > 0 && cur > 0 && cur == lastBytes {
 					ar := activeReader.Load()
 					draining := ar != nil && ar.drained.Load()
-					if draining {
-						progressLog("[nix-store push] sent %s in %s — pipe drained, waiting for registry to finalize\n",
+					allRead := TotalSizeHint > 0 && cur >= uint64(TotalSizeHint)
+					if draining || allRead {
+						progressLog("[nix-store push] sent %s in %s — all data read, waiting for registry to finalize\n",
 							fmtBytes(cur), now.Sub(start).Round(time.Second))
 						stallSince = time.Time{}
 					} else if stallSince.IsZero() {
@@ -256,56 +263,91 @@ func Push(ctx context.Context, baseRef, dstRef string, r io.ReadCloser) error {
 	if ChunkSize > 0 {
 		defer r.Close()
 		tr := tar.NewReader(r)
-		currentImg := baseImg
 
-		uploadOpts := make([]remote.Option, len(remoteOpts))
-		copy(uploadOpts, remoteOpts)
-		uploadOpts = append(uploadOpts, remote.WithContext(uploadCtx))
+		// Phase 1: split tar into temp files on disk.
+		splitStart := time.Now()
+		var chunkFiles []string
+		defer func() {
+			for _, f := range chunkFiles {
+				os.Remove(f)
+			}
+		}()
 
 		for chunkIdx := 0; ; chunkIdx++ {
+			tmpFile, err := os.CreateTemp("", fmt.Sprintf("nix-chunk-%03d-", chunkIdx))
+			if err != nil {
+				return fmt.Errorf("create temp chunk %d: %w", chunkIdx+1, err)
+			}
+
 			pr, pw := io.Pipe()
 			more := make(chan bool, 1)
 			go writeOneChunk(tr, pw, ChunkSize, more)
 
-			progressLog("[nix-store push] chunk %d streaming\n", chunkIdx+1)
-
-			chunkPR := &progressReader{rc: pr, count: &byteCount}
-			activeReader.Store(chunkPR)
+			n, copyErr := io.Copy(tmpFile, pr)
+			tmpFile.Close()
 
-			layer := stream.NewLayer(
-				chunkPR,
-				stream.WithMediaType(types.OCILayer),
-			)
-
-			img, err := mutate.AppendLayers(currentImg, layer)
-			if err != nil {
-				pr.CloseWithError(err)
+			if copyErr != nil {
+				os.Remove(tmpFile.Name())
 				<-more
-				return fmt.Errorf("append chunk %d: %w", chunkIdx+1, err)
+				return fmt.Errorf("buffer chunk %d: %w", chunkIdx+1, copyErr)
 			}
 
-			writeStart := time.Now()
-			if err := remote.Write(dst, img, uploadOpts...); err != nil {
-				pr.CloseWithError(err)
-				<-more
-				if stallDetected.Load() {
-					return fmt.Errorf("push %q: upload stalled — no progress for %s", dstRef, StallTimeout.Round(time.Second))
-				}
-				return fmt.Errorf("push chunk %d of %q: %w", chunkIdx+1, dstRef, err)
-			}
-			progressLog("[nix-store push] chunk %d committed (%s)\n", chunkIdx+1, time.Since(writeStart).Round(time.Millisecond))
+			chunkFiles = append(chunkFiles, tmpFile.Name())
+			progressLog("[nix-store push] chunk %d split (%s)\n", chunkIdx+1, fmtBytes(uint64(n)))
 
 			if !(<-more) {
 				break
 			}
+		}
 
-			refetchStart := time.Now()
-			currentImg, err = remote.Image(dst, uploadOpts...)
+		var totalFileBytes int64
+		for _, cf := range chunkFiles {
+			if fi, err := os.Stat(cf); err == nil {
+				totalFileBytes += fi.Size()
+			}
+		}
+		TotalSizeHint = totalFileBytes
+		progressLog("[nix-store push] %d chunks (%s) split in %s, uploading with %d parallel streams\n",
+			len(chunkFiles), fmtBytes(uint64(totalFileBytes)), time.Since(splitStart).Round(time.Millisecond), UploadJobs)
+
+		// Phase 2: open all chunk files, wrap in layers, one remote.Write.
+		var layers []v1.Layer
+		var openFiles []*os.File
+		defer func() {
+			for _, f := range openFiles {
+				f.Close()
+			}
+		}()
+
+		for _, chunkFile := range chunkFiles {
+			f, err := os.Open(chunkFile)
 			if err != nil {
-				return fmt.Errorf("re-fetch after chunk %d: %w", chunkIdx+1, err)
+				return fmt.Errorf("open chunk: %w", err)
+			}
+			openFiles = append(openFiles, f)
+
+			chunkPR := &progressReader{rc: f, count: &byteCount}
+			layer := stream.NewLayer(chunkPR, stream.WithMediaType(types.OCILayer))
+			layers = append(layers, layer)
+		}
+
+		img, err := mutate.AppendLayers(baseImg, layers...)
+		if err != nil {
+			return fmt.Errorf("append %d layers: %w", len(layers), err)
+		}
+
+		uploadOpts := make([]remote.Option, len(remoteOpts))
+		copy(uploadOpts, remoteOpts)
+		uploadOpts = append(uploadOpts, remote.WithContext(uploadCtx), remote.WithJobs(UploadJobs))
+
+		writeStart := time.Now()
+		if err := remote.Write(dst, img, uploadOpts...); err != nil {
+			if stallDetected.Load() {
+				return fmt.Errorf("push %q: upload stalled — no progress for %s", dstRef, StallTimeout.Round(time.Second))
 			}
-			progressLog("[nix-store push] manifest re-fetched (%s)\n", time.Since(refetchStart).Round(time.Millisecond))
+			return fmt.Errorf("push %q: %w", dstRef, err)
 		}
+		progressLog("[nix-store push] %d chunks committed (%s)\n", len(layers), time.Since(writeStart).Round(time.Millisecond))
 	} else {
 		pr := &progressReader{rc: r, count: &byteCount}
 		stallCloser = pr