feat: Better parallelism -- Testing with Smelt

.gitignore

@@ -8,10 +8,6 @@ guppy
 # Environment Files
 .env*
 
-# Config Files
-*.yaml
-*.toml
-
 # IDE's
 *.idea/
 

Dockerfile

@@ -1,7 +1,7 @@
 # ============================================
-# Build stage (shared)
+# Build base stage
 # ============================================
-FROM golang:1.25.3-trixie AS build
+FROM golang:1.26.1-trixie AS build-base
 
 # Docker sets TARGETARCH automatically during multi-platform builds
 ARG TARGETARCH
@@ -10,25 +10,39 @@ WORKDIR /go/src/guppy
 
 COPY go.* .
 RUN go mod download
-COPY . .
+COPY --parents cmd internal pkg Makefile main.go version.json ./
+
+# ============================================
+# Production build stage
+# ============================================
+FROM build-base AS build-prod
+
+# Allow the Makefile to look up the git commit, at the expense of busting the
+# cache whenever `.git` changes.
+COPY --parents .git ./
 
 # Production build - with symbol stripping
-RUN CGO_ENABLED=0 GOOS=linux GOARCH=${TARGETARCH} make guppy-prod
+RUN --mount=type=cache,target=/root/.cache/go-build CGO_ENABLED=0 GOOS=linux GOARCH=${TARGETARCH} make guppy-prod
 
 # ============================================
 # Debug build stage
 # ============================================
-FROM build AS build-debug
+FROM build-base AS build-debug
+
+# Debug build - no optimizations, no inlining
+RUN --mount=type=cache,target=/root/.cache/go-build CGO_ENABLED=0 GOOS=linux GOARCH=${TARGETARCH} make guppy-debug
+
+# ============================================
+# Debug tools
+# ============================================
+FROM golang:1.26.1-trixie AS build-debug-tools
 
 ARG TARGETARCH
 
 # Install delve debugger and randdir tool for target architecture
 RUN GOARCH=${TARGETARCH} go install github.com/go-delve/delve/cmd/dlv@latest && \
     GOARCH=${TARGETARCH} go install github.com/storacha/randdir@latest
 
-# Debug build - no optimizations, no inlining
-RUN CGO_ENABLED=0 GOOS=linux GOARCH=${TARGETARCH} make guppy-debug
-
 # ============================================
 # Production image
 # ============================================
@@ -39,7 +53,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
     curl \
     && rm -rf /var/lib/apt/lists/*
 
-COPY --from=build /go/src/guppy/guppy /usr/bin/guppy
+COPY --from=build-prod /go/src/guppy/guppy /usr/bin/guppy
 
 ENTRYPOINT ["/usr/bin/guppy"]
 
@@ -69,8 +83,8 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
     && rm -rf /var/lib/apt/lists/*
 
 # Delve debugger and randdir tool
-COPY --from=build-debug /go/bin/dlv /usr/bin/dlv
-COPY --from=build-debug /go/bin/randdir /usr/bin/randdir
+COPY --from=build-debug-tools /go/bin/dlv /usr/bin/dlv
+COPY --from=build-debug-tools /go/bin/randdir /usr/bin/randdir
 
 # Debug binary (with symbols, no optimizations)
 COPY --from=build-debug /go/src/guppy/guppy /usr/bin/guppy
@@ -87,3 +101,19 @@ RUN echo 'alias ll="ls -la"' >> /etc/bash.bashrc && \
 
 SHELL ["/bin/bash", "-c"]
 ENTRYPOINT ["/usr/bin/guppy"]
+
+# ============================================
+# Test image (without interactive tools)
+# ============================================
+FROM debian:bookworm-slim AS test
+
+# Debug binary (with symbols, no optimizations)
+COPY --from=build-debug /go/src/guppy/guppy /usr/bin/guppy
+
+# Create data directories
+RUN mkdir -p /root/.storacha/guppy /root/.config/guppy
+
+WORKDIR /root
+
+SHELL ["/bin/bash", "-c"]
+ENTRYPOINT ["/usr/bin/guppy"]

Makefile

@@ -6,7 +6,7 @@ DB_PATH ?= ~/.storacha/guppy/preparation.db
 GOOSE := go tool goose
 
 VERSION=$(shell awk -F'"' '/"version":/ {print $$4}' version.json)
-COMMIT=$(shell git rev-parse --short HEAD)
+COMMIT=$(shell git rev-parse --short HEAD 2>/dev/null || echo "unknown")
 DATE=$(shell date -u -Iseconds)
 GOFLAGS=-ldflags="-X github.com/storacha/guppy/pkg/build.version=$(VERSION) -X github.com/storacha/guppy/pkg/build.Commit=$(COMMIT) -X github.com/storacha/guppy/pkg/build.Date=$(DATE) -X github.com/storacha/guppy/pkg/build.BuiltBy=make"
 DOCKER?=$(shell which docker)
@@ -59,3 +59,7 @@ docker-prod: docker-setup
 
 docker-dev: docker-setup
 	$(DOCKER) buildx build --platform linux/amd64,linux/arm64 --target dev -t guppy:dev .
+
+test-upload:
+	@echo "Running upload test..."
+	./test/doupload

cmd/upload/root.go

@@ -107,6 +107,7 @@ var Cmd = &cobra.Command{
 			preparation.WithAssumeUnchangedSources(rootFlags.assumeUnchangedSources),
 			preparation.WithEventBus(eb),
 			preparation.WithReplicas(cfg.Upload.Replicas),
+			preparation.WithPutHTTPClient(cmdutil.TracedHTTPClient),
 		)
 		allUploads, err := api.FindOrCreateUploads(ctx, spaceDID)
 		if err != nil {

cmd/upload/source/add.go

@@ -73,7 +73,7 @@ var AddCmd = &cobra.Command{
 			return err
 		}
 
-		api := preparation.NewAPI(repo, client)
+		api := preparation.NewAPI(repo, client, preparation.WithPutHTTPClient(cmdutil.TracedHTTPClient))
 
 		// Parse shard size if provided
 		var spaceOptions []model.SpaceOption

internal/cmdutil/cmdutil.go

@@ -43,8 +43,10 @@ func envSigner() (principal.Signer, error) {
 	return signer.Parse(str)
 }
 
-var tracedHttpClient = &http.Client{
-	Transport: otelhttp.NewTransport(http.DefaultTransport),
+// TracedHTTPClient is an HTTP client with OpenTelemetry tracing and a guppy
+// User-Agent header on all outbound requests.
+var TracedHTTPClient = &http.Client{
+	Transport: newUserAgentTransport(otelhttp.NewTransport(http.DefaultTransport)),
 }
 
 // MustGetClient creates a new client suitable for the CLI, using stored data,
@@ -77,7 +79,7 @@ func MustGetClientForNetwork(storePath string, networkCfg config.NetworkConfig,
 
 	conn, err := uclient.NewConnection(
 		network.UploadID,
-		uhttp.NewChannel(&network.UploadURL, uhttp.WithClient(tracedHttpClient)),
+		uhttp.NewChannel(&network.UploadURL, uhttp.WithClient(TracedHTTPClient)),
 		uclient.WithOutboundCodec(car.NewOutboundCodec()),
 	)
 	if err != nil {
@@ -90,7 +92,7 @@ func MustGetClientForNetwork(storePath string, networkCfg config.NetworkConfig,
 			append(
 				options,
 				client.WithConnection(conn),
-				client.WithReceiptsClient(receiptclient.New(&network.ReceiptsURL, receiptclient.WithHTTPClient(tracedHttpClient))),
+				client.WithReceiptsClient(receiptclient.New(&network.ReceiptsURL, receiptclient.WithHTTPClient(TracedHTTPClient))),
 			)...,
 		)...,
 	)
@@ -132,7 +134,7 @@ func MustGetIndexClient(networkCfg config.NetworkConfig) (*indexclient.Client, u
 func MustGetIndexClientForNetwork(networkCfg config.NetworkConfig, flagName string) (*indexclient.Client, ucan.Principal) {
 	network := MustGetNetworkConfig(networkCfg, flagName)
 
-	client, err := indexclient.New(network.IndexerID, network.IndexerURL, indexclient.WithHTTPClient(tracedHttpClient))
+	client, err := indexclient.New(network.IndexerID, network.IndexerURL, indexclient.WithHTTPClient(TracedHTTPClient))
 	if err != nil {
 		log.Fatal(err)
 	}

internal/cmdutil/useragent.go

@@ -0,0 +1,26 @@
+package cmdutil
+
+import (
+	"fmt"
+	"net/http"
+
+	"github.com/storacha/guppy/pkg/build"
+)
+
+type userAgentTransport struct {
+	userAgent string
+	base      http.RoundTripper
+}
+
+func (t *userAgentTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	req = req.Clone(req.Context())
+	req.Header.Set("User-Agent", t.userAgent)
+	return t.base.RoundTrip(req)
+}
+
+func newUserAgentTransport(base http.RoundTripper) http.RoundTripper {
+	return &userAgentTransport{
+		userAgent: fmt.Sprintf("guppy/%s", build.Version),
+		base:      base,
+	}
+}

pkg/preparation/preparation.go

@@ -5,13 +5,15 @@ import (
 	"errors"
 	"fmt"
 	"io/fs"
+	"net/http"
 	"os"
 	"path/filepath"
 
 	"github.com/ipfs/go-cid"
 	"github.com/storacha/go-ucanto/did"
 
 	"github.com/storacha/guppy/pkg/bus"
+	clientpkg "github.com/storacha/guppy/pkg/client"
 	"github.com/storacha/guppy/pkg/preparation/blobs"
 	"github.com/storacha/guppy/pkg/preparation/dags"
 	"github.com/storacha/guppy/pkg/preparation/dags/nodereader"
@@ -61,6 +63,7 @@ type config struct {
 	assumeUnchangedSources bool
 	bus                    bus.Bus
 	replicas               uint
+	putHTTPClient          *http.Client
 }
 
 const (
@@ -156,14 +159,18 @@ func NewAPI(repo Repo, client StorachaClient, options ...Option) API {
 		ShardEncoder:     blobs.NewCAREncoder(),
 	}
 
+	var blobAddOptions []clientpkg.SpaceBlobAddOption
+	if cfg.putHTTPClient != nil {
+		blobAddOptions = append(blobAddOptions, clientpkg.WithPutClient(cfg.putHTTPClient))
+	}
 	storachaAPI := storacha.API{
-		Repo:                  repo,
-		Client:                client,
-		ReaderForShard:        blobsAPI.ReaderForShard,
-		ReaderForIndex:        blobsAPI.ReaderForIndex,
-		BlobUploadParallelism: cfg.blobUploadParallelism,
-		Bus:                   cfg.bus,
-		Replicas:              cfg.replicas,
+		Repo:           repo,
+		Client:         client,
+		ReaderForShard: blobsAPI.ReaderForShard,
+		ReaderForIndex: blobsAPI.ReaderForIndex,
+		Bus:            cfg.bus,
+		Replicas:       cfg.replicas,
+		BlobAddOptions: blobAddOptions,
 	}
 
 	uploadsAPI = uploads.API{
@@ -175,10 +182,11 @@ func NewAPI(repo Repo, client StorachaClient, options ...Option) API {
 		AddShardsToUploadIndexes:   blobsAPI.AddShardsToUploadIndexes,
 		CloseUploadShards:          blobsAPI.CloseUploadShards,
 		CloseUploadIndexes:         blobsAPI.CloseUploadIndexes,
-		AddShardsForUpload:         storachaAPI.AddShardsForUpload,
-		PostProcessUploadedShards:  storachaAPI.PostProcessUploadedShards,
-		PostProcessUploadedIndexes: storachaAPI.PostProcessUploadedIndexes,
-		AddIndexesForUpload:        storachaAPI.AddIndexesForUpload,
+		FindShardAddTasksForUpload: storachaAPI.FindShardAddTasksForUpload,
+		FindIndexAddTasksForUpload: storachaAPI.FindIndexAddTasksForUpload,
+		BlobUploadParallelism:      cfg.blobUploadParallelism,
+		FindShardPostProcessTasksForUpload: storachaAPI.FindShardPostProcessTasksForUpload,
+		FindIndexPostProcessTasksForUpload: storachaAPI.FindIndexPostProcessTasksForUpload,
 		AddStorachaUploadForUpload: storachaAPI.AddStorachaUploadForUpload,
 		RemoveBadFSEntry:           scansAPI.RemoveBadFSEntry,
 		RemoveBadNodes:             dagsAPI.RemoveBadNodes,
@@ -247,6 +255,14 @@ func WithReplicas(replicas uint) Option {
 	}
 }
 
+// WithPutHTTPClient sets the HTTP client used for blob PUT uploads.
+func WithPutHTTPClient(c *http.Client) Option {
+	return func(cfg *config) error {
+		cfg.putHTTPClient = c
+		return nil
+	}
+}
+
 func (a API) FindOrCreateSpace(ctx context.Context, spaceDID did.DID, name string, options ...spacesmodel.SpaceOption) (*spacesmodel.Space, error) {
 	return a.Spaces.FindOrCreateSpace(ctx, spaceDID, name, options...)
 }

pkg/preparation/preparation_test.go

@@ -405,7 +405,7 @@ func TestExecuteUpload(t *testing.T) {
 		// We don't know exactly how many successful PUTs there were, but we know it
 		// should be at least 2 and at most 6.
 		require.GreaterOrEqual(t, putBlobs.Size(), 2, "expected at least 2/5 shards to be added so far")
-		require.Less(t, putBlobs.Size(), 6, "expected at most 4/5 shards + 1 index to be added so far")
+		require.LessOrEqual(t, putBlobs.Size(), 6, "expected at most 5/5 shards + 1 index to be added so far")
 		require.Len(t, uploadAddCaps, 0, "expected `upload/add` not to have been called yet")
 
 		t.Log("Retrying the upload after error...")