Optimize pipeline performance

- Parallelize R2 uploads with 20 concurrent workers and CopyObject for
  unchanged content-addressed files (incremental deploys)
- Tune HTTP transport with connection pooling matching concurrency
- Batch SVN discovery upserts in transactions of 500
- Batch update upserts via dedicated writer goroutine (batches of 100)
- Parallelize build file writes with 8-worker errgroup
- Pre-create p/wp-plugin, p/wp-theme, p2/ directories once upfront
- Add incremental builds: hard-link unchanged files from previous build
  based on content-addressed hash comparison
- Reuse serialized JSON bytes for p2/ files (same content as p/)
- Replace disk-based ValidateIntegrity with in-memory hash validation

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: 2 people

cmd/wpcomposer/cmd/build.go

@@ -5,6 +5,7 @@ import (
 	"path/filepath"
 	"time"
 
+	"github.com/roots/wp-composer/internal/deploy"
 	"github.com/roots/wp-composer/internal/repository"
 	"github.com/spf13/cobra"
 )
@@ -24,12 +25,23 @@ func runBuild(cmd *cobra.Command, args []string) error {
 		output = filepath.Join("storage", "repository", "builds")
 	}
 
+	// Resolve previous build dir for incremental builds
+	var previousBuildDir string
+	repoDir := filepath.Dir(output) // storage/repository
+	if latestID, err := deploy.LatestBuildID(repoDir); err == nil && latestID != "" {
+		candidate := deploy.BuildDirFromID(repoDir, latestID)
+		if err := deploy.ValidateBuild(candidate); err == nil {
+			previousBuildDir = candidate
+		}
+	}
+
 	result, err := repository.Build(cmd.Context(), application.DB, repository.BuildOpts{
-		OutputDir:   output,
-		AppURL:      application.Config.AppURL,
-		Force:       force,
-		PackageName: pkg,
-		Logger:      application.Logger,
+		OutputDir:        output,
+		AppURL:           application.Config.AppURL,
+		Force:            force,
+		PackageName:      pkg,
+		PreviousBuildDir: previousBuildDir,
+		Logger:           application.Logger,
 	})
 	if err != nil {
 		return err

cmd/wpcomposer/cmd/deploy.go

@@ -21,6 +21,7 @@ func runDeploy(cmd *cobra.Command, args []string) error {
 	repoDir := filepath.Join("storage", "repository")
 	cleanup, _ := cmd.Flags().GetBool("cleanup")
 	toR2, _ := cmd.Flags().GetBool("to-r2")
+	previousBuildID, _ := deploy.CurrentBuildID(repoDir)
 
 	r2Cleanup, _ := cmd.Flags().GetBool("r2-cleanup")
 	retainCount, _ := cmd.Flags().GetInt("retain")
@@ -77,7 +78,7 @@ func runDeploy(cmd *cobra.Command, args []string) error {
 
 		// Sync to R2 first, then promote locally
 		if toR2 || application.Config.R2.Enabled {
-			if err := deploy.SyncToR2(cmd.Context(), application.Config.R2, buildDir, target, application.Logger); err != nil {
+			if err := deploy.SyncToR2(cmd.Context(), application.Config.R2, buildDir, target, previousBuildID, application.Logger); err != nil {
 				return fmt.Errorf("R2 sync failed: %w", err)
 			}
 			recordR2Sync(cmd, target)
@@ -109,7 +110,7 @@ func runDeploy(cmd *cobra.Command, args []string) error {
 
 	// Sync to R2 first, then promote locally
 	if toR2 || application.Config.R2.Enabled {
-		if err := deploy.SyncToR2(cmd.Context(), application.Config.R2, buildDir, buildID, application.Logger); err != nil {
+		if err := deploy.SyncToR2(cmd.Context(), application.Config.R2, buildDir, buildID, previousBuildID, application.Logger); err != nil {
 			return fmt.Errorf("R2 sync failed: %w", err)
 		}
 		recordR2Sync(cmd, buildID)

cmd/wpcomposer/cmd/discover.go

@@ -109,6 +109,8 @@ func discoverFromConfig(ctx context.Context, pkgType string, limit, concurrency
 }
 
 func discoverFromSVN(ctx context.Context, pkgType string, limit int) error {
+	const svnBatchSize = 500
+
 	client := wporg.NewClient(application.Config.Discovery, application.Logger)
 
 	type svnSource struct {
@@ -129,20 +131,46 @@ func discoverFromSVN(ctx context.Context, pkgType string, limit int) error {
 		application.Logger.Info("discovering from SVN", "type", src.pkgType, "url", src.url)
 
 		var count, failed int
+		batch := make([]packages.ShellEntry, 0, svnBatchSize)
+
+		flush := func() {
+			if len(batch) == 0 {
+				return
+			}
+			if err := packages.BatchUpsertShellPackages(ctx, application.DB, batch); err != nil {
+				application.Logger.Warn("batch upsert failed, falling back to individual", "error", err)
+				for _, e := range batch {
+					if err := packages.UpsertShellPackage(ctx, application.DB, e.Type, e.Name, e.LastCommitted); err != nil {
+						application.Logger.Warn("failed to upsert shell package", "slug", e.Name, "error", err)
+						failed++
+						count--
+					}
+				}
+			}
+			batch = batch[:0]
+		}
+
 		err := client.ParseSVNListing(ctx, src.url, func(entry wporg.SVNEntry) error {
 			if limit > 0 && totalCount >= limit {
 				return errLimitReached
 			}
 
-			if err := packages.UpsertShellPackage(ctx, application.DB, src.pkgType, entry.Slug, entry.LastCommitted); err != nil {
-				application.Logger.Warn("failed to upsert shell package", "slug", entry.Slug, "error", err)
-				failed++
-				return nil
-			}
+			batch = append(batch, packages.ShellEntry{
+				Type:          src.pkgType,
+				Name:          entry.Slug,
+				LastCommitted: entry.LastCommitted,
+			})
 			count++
 			totalCount++
+
+			if len(batch) >= svnBatchSize {
+				flush()
+			}
 			return nil
 		})
+
+		flush()
+
 		if err != nil && err != errLimitReached {
 			return fmt.Errorf("SVN discovery for %s: %w", src.pkgType, err)
 		}

cmd/wpcomposer/cmd/update.go

@@ -62,10 +62,43 @@ func runUpdate(cmd *cobra.Command, args []string) error {
 
 	client := wporg.NewClient(application.Config.Discovery, application.Logger)
 
+	const writeBatchSize = 100
+
 	var succeeded, failed, deactivated atomic.Int64
 	g, gCtx := errgroup.WithContext(ctx)
 	g.SetLimit(concurrency)
 
+	// Writer goroutine batches DB writes
+	writeCh := make(chan *packages.Package, concurrency*2)
+	writeErrCh := make(chan error, 1)
+	go func() {
+		defer close(writeErrCh)
+		batch := make([]*packages.Package, 0, writeBatchSize)
+		flush := func() {
+			if len(batch) == 0 {
+				return
+			}
+			if err := packages.BatchUpsertPackages(ctx, application.DB, batch); err != nil {
+				application.Logger.Warn("batch upsert failed, falling back to individual", "error", err)
+				for _, pkg := range batch {
+					if err := packages.UpsertPackage(ctx, application.DB, pkg); err != nil {
+						application.Logger.Warn("failed to store", "type", pkg.Type, "name", pkg.Name, "error", err)
+						failed.Add(1)
+						succeeded.Add(-1)
+					}
+				}
+			}
+			batch = batch[:0]
+		}
+		for pkg := range writeCh {
+			batch = append(batch, pkg)
+			if len(batch) >= writeBatchSize {
+				flush()
+			}
+		}
+		flush()
+	}()
+
 	for _, p := range pkgs {
 		p := p
 		g.Go(func() error {
@@ -123,13 +156,9 @@ func runUpdate(cmd *cobra.Command, args []string) error {
 			pkg.LastSyncedAt = &now
 			pkg.LastSyncRunID = &syncRun.RunID
 
-			if err := packages.UpsertPackage(gCtx, application.DB, pkg); err != nil {
-				application.Logger.Warn("failed to store", "type", p.Type, "name", p.Name, "error", err)
-				failed.Add(1)
-				return nil
-			}
-
 			succeeded.Add(1)
+			writeCh <- pkg
+
 			total := succeeded.Load() + failed.Load() + deactivated.Load()
 			if total%500 == 0 {
 				application.Logger.Info("update progress",
@@ -149,6 +178,9 @@ func runUpdate(cmd *cobra.Command, args []string) error {
 		return err
 	}
 
+	close(writeCh)
+	<-writeErrCh // wait for writer to finish
+
 	stats := map[string]any{
 		"updated":     succeeded.Load(),
 		"failed":      failed.Load(),

internal/deploy/r2.go

@@ -10,8 +10,11 @@ import (
 	"math"
 	"sort"
 	"strings"
+	"sync/atomic"
 	"time"
 
+	"golang.org/x/sync/errgroup"
+
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/credentials"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
@@ -41,12 +44,18 @@ type buildFile struct {
 }
 
 // SyncToR2 uploads all files in a build directory to R2 under a versioned
-// release prefix (releases/<buildID>/). After all release files are uploaded,
-// it rewrites the root packages.json to point at the new prefix — the atomic
-// pointer swap that makes the new release live.
-func SyncToR2(ctx context.Context, cfg config.R2Config, buildDir, buildID string, logger *slog.Logger) error {
+// release prefix (releases/<buildID>/). When previousBuildID is non-empty,
+// content-addressed p/ files that exist in both builds are copied within R2
+// instead of re-uploaded. After all release files are uploaded, it rewrites
+// the root packages.json to point at the new prefix — the atomic pointer swap
+// that makes the new release live.
+func SyncToR2(ctx context.Context, cfg config.R2Config, buildDir, buildID, previousBuildID string, logger *slog.Logger) error {
 	client := newS3Client(cfg)
 	releasePrefix := "releases/" + buildID + "/"
+	previousPrefix := ""
+	if previousBuildID != "" {
+		previousPrefix = "releases/" + previousBuildID + "/"
+	}
 
 	// Collect all files, then sort: index files last.
 	var files []buildFile
@@ -60,23 +69,68 @@ func SyncToR2(ctx context.Context, cfg config.R2Config, buildDir, buildID string
 
 	sortBuildFiles(files)
 
-	// Upload all files under the release prefix.
-	var uploaded int
+	// Extract packages.json data before parallel upload.
 	var packagesData []byte
 	for _, f := range files {
-		key := releasePrefix + f.relPath
-		if err := putObjectWithRetry(ctx, client, cfg.Bucket, key, f.data, logger); err != nil {
-			return fmt.Errorf("R2 sync: %w", err)
-		}
 		if f.relPath == r2IndexFile {
 			packagesData = f.data
+			break
 		}
-		uploaded++
-		if uploaded%100 == 0 {
-			logger.Info("R2 upload progress", "uploaded", uploaded, "total", len(files))
+	}
+
+	// Build a set of content-addressed p/ files from the previous build for CopyObject.
+	previousKeys := make(map[string]bool)
+	if previousPrefix != "" {
+		for _, f := range files {
+			if strings.HasPrefix(f.relPath, "p/") && strings.Contains(f.relPath, "$") {
+				previousKeys[previousPrefix+f.relPath] = true
+			}
 		}
 	}
 
+	// Upload all files under the release prefix (parallel).
+	var uploaded, copied atomic.Int64
+	g, gCtx := errgroup.WithContext(ctx)
+	g.SetLimit(20)
+
+	for _, f := range files {
+		f := f
+		g.Go(func() error {
+			key := releasePrefix + f.relPath
+
+			// For content-addressed p/ files that existed in the previous build,
+			// use CopyObject within R2 instead of re-uploading.
+			if previousPrefix != "" && strings.HasPrefix(f.relPath, "p/") && strings.Contains(f.relPath, "$") {
+				srcKey := previousPrefix + f.relPath
+				if previousKeys[srcKey] {
+					if err := copyObjectWithRetry(gCtx, client, cfg.Bucket, srcKey, key, logger); err == nil {
+						copied.Add(1)
+						n := uploaded.Add(1) + copied.Load()
+						if n%500 == 0 {
+							logger.Info("R2 upload progress", "uploaded", uploaded.Load(), "copied", copied.Load(), "total", len(files))
+						}
+						return nil
+					}
+					// Fall through to upload on copy failure.
+					logger.Warn("CopyObject failed, falling back to upload", "key", key)
+				}
+			}
+
+			if err := putObjectWithRetry(gCtx, client, cfg.Bucket, key, f.data, logger); err != nil {
+				return fmt.Errorf("R2 sync: %w", err)
+			}
+			n := uploaded.Add(1)
+			if n%500 == 0 {
+				logger.Info("R2 upload progress", "uploaded", int(n), "total", len(files))
+			}
+			return nil
+		})
+	}
+
+	if err := g.Wait(); err != nil {
+		return err
+	}
+
 	// Rewrite and upload root packages.json — the atomic switch.
 	if packagesData == nil {
 		return fmt.Errorf("R2 sync: packages.json not found in build")
@@ -89,7 +143,7 @@ func SyncToR2(ctx context.Context, cfg config.R2Config, buildDir, buildID string
 		return fmt.Errorf("R2 sync (root packages.json): %w", err)
 	}
 
-	logger.Info("R2 sync complete", "files", uploaded, "release", releasePrefix)
+	logger.Info("R2 sync complete", "uploaded", uploaded.Load(), "copied", copied.Load(), "release", releasePrefix)
 	return nil
 }
 
@@ -175,6 +229,35 @@ func putObjectWithRetry(ctx context.Context, client *s3.Client, bucket, key stri
 	return fmt.Errorf("uploading %s after %d attempts: %w", key, r2MaxRetries, lastErr)
 }
 
+// copyObjectWithRetry copies a single object within R2 with exponential backoff retry.
+func copyObjectWithRetry(ctx context.Context, client *s3.Client, bucket, srcKey, dstKey string, logger *slog.Logger) error {
+	copySource := bucket + "/" + srcKey
+	cacheControl := CacheControlForPath(dstKey)
+
+	var lastErr error
+	for attempt := range r2MaxRetries {
+		if attempt > 0 {
+			delay := time.Duration(float64(r2RetryBaseMs)*math.Pow(2, float64(attempt-1))) * time.Millisecond
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-time.After(delay):
+			}
+		}
+
+		_, lastErr = client.CopyObject(ctx, &s3.CopyObjectInput{
+			Bucket:       aws.String(bucket),
+			CopySource:   aws.String(copySource),
+			Key:          aws.String(dstKey),
+			CacheControl: aws.String(cacheControl),
+		})
+		if lastErr == nil {
+			return nil
+		}
+	}
+	return fmt.Errorf("copying %s -> %s after %d attempts: %w", srcKey, dstKey, r2MaxRetries, lastErr)
+}
+
 // CleanupR2 removes old release prefixes from R2, keeping the live release,
 // releases within the grace period, and the top N most recent releases.
 // It no longer depends on the local filesystem — all state is read from R2.