feat(cache): add immediate upload mode for external interruption resilience

Add LEEWAY_CACHE_UPLOAD_IMMEDIATE environment variable to enable uploading
packages to remote cache immediately after each successful build, rather than
waiting until the end of the entire build process.

This addresses scenarios where builds are interrupted externally (CI timeouts,
manual cancellation, SIGTERM/SIGKILL) before the final upload phase, causing
loss of all successfully built packages.

Changes:
- Add LEEWAY_CACHE_UPLOAD_IMMEDIATE environment variable (default: false)
- Implement immediate per-package upload in RegisterNewlyBuilt()
- Track uploaded packages to prevent duplicates
- Skip already-uploaded packages in final batch upload
- Add uploadPackageImmediately() method with error handling
- Document new environment variable in README

Trade-offs:
- Immediate mode: More network calls but survives interruptions
- Default mode: Single batch upload, more efficient but vulnerable to interrupts

Usage:
  export LEEWAY_CACHE_UPLOAD_IMMEDIATE=true
  leeway build your-package

Co-authored-by: Ona <no-reply@ona.com>

Author: geropl

README.md

@@ -491,6 +491,7 @@ variables have an effect on leeway:
           For details on configuring AWS credentials see https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-files.html
 - `LEEWAY_CACHE_DIR`: Location of the local build cache. The directory does not have to exist yet.
 - `LEEWAY_BUILD_DIR`: Working location of leeway (i.e. where the actual builds happen). This location will see heavy I/O which makes it advisable to place this on a fast SSD or in RAM.
+- `LEEWAY_CACHE_UPLOAD_IMMEDIATE`: When set to "true", packages are uploaded to remote cache immediately after successful build, rather than waiting until the end of the entire build. This is useful when builds may be interrupted externally (e.g., CI timeouts, manual cancellation) to ensure partial results are cached. Defaults to "false".
 - `LEEWAY_YARN_MUTEX`: Configures the mutex flag leeway will pass to yarn. Defaults to "network". See https://yarnpkg.com/lang/en/docs/cli/#toc-concurrency-and-mutex for possible values.
 - `LEEWAY_EXPERIMENTAL`: Enables exprimental features
 

pkg/leeway/build.go

@@ -68,9 +68,11 @@ type buildContext struct {
 	mu                 sync.Mutex
 	newlyBuiltPackages map[string]*Package
 
-	pkgLockCond *sync.Cond
-	pkgLocks    map[string]struct{}
-	buildLimit  *semaphore.Weighted
+	pkgLockCond           *sync.Cond
+	pkgLocks              map[string]struct{}
+	buildLimit            *semaphore.Weighted
+	immediateCacheUpload  bool
+	uploadedPackages      map[string]struct{} // Track packages already uploaded to avoid duplicates
 }
 
 const (
@@ -85,6 +87,12 @@ const (
 	// Defaults to "network".
 	EnvvarYarnMutex = "LEEWAY_YARN_MUTEX"
 
+	// EnvvarImmediateCacheUpload controls whether packages are uploaded to remote cache immediately
+	// after successful build, rather than waiting until the end of the entire build.
+	// Set to "true" to enable immediate uploads. This is useful when builds may be interrupted
+	// externally (e.g., CI timeouts, manual cancellation) to ensure partial results are cached.
+	EnvvarImmediateCacheUpload = "LEEWAY_CACHE_UPLOAD_IMMEDIATE"
+
 	// dockerImageNamesFiles is the name of the file store in poushed Docker build artifacts
 	// which contains the names of the Docker images we just pushed
 	dockerImageNamesFiles = "imgnames.txt"
@@ -145,15 +153,20 @@ func newBuildContext(options buildOptions) (ctx *buildContext, err error) {
 		return nil, xerrors.Errorf("cannot compute hash of myself: %w", err)
 	}
 
+	// Check if immediate cache upload is enabled
+	immediateCacheUpload := os.Getenv(EnvvarImmediateCacheUpload) == "true"
+
 	ctx = &buildContext{
-		buildOptions:       options,
-		buildDir:           buildDir,
-		buildID:            buildID,
-		newlyBuiltPackages: make(map[string]*Package),
-		pkgLockCond:        sync.NewCond(&sync.Mutex{}),
-		pkgLocks:           make(map[string]struct{}),
-		buildLimit:         buildLimit,
-		leewayHash:         hex.EncodeToString(leewayHash.Sum(nil)),
+		buildOptions:         options,
+		buildDir:             buildDir,
+		buildID:              buildID,
+		newlyBuiltPackages:   make(map[string]*Package),
+		pkgLockCond:          sync.NewCond(&sync.Mutex{}),
+		pkgLocks:             make(map[string]struct{}),
+		buildLimit:           buildLimit,
+		leewayHash:           hex.EncodeToString(leewayHash.Sum(nil)),
+		immediateCacheUpload: immediateCacheUpload,
+		uploadedPackages:     make(map[string]struct{}),
 	}
 
 	err = os.MkdirAll(buildDir, 0755)
@@ -235,16 +248,61 @@ func (c *buildContext) RegisterNewlyBuilt(p *Package) error {
 	c.mu.Lock()
 	c.newlyBuiltPackages[ver] = p
 	c.mu.Unlock()
+
+	// If immediate cache upload is enabled, upload this package right away
+	if c.immediateCacheUpload && !p.Ephemeral {
+		if err := c.uploadPackageImmediately(p); err != nil {
+			// Log the error but don't fail the build
+			log.WithError(err).WithField("package", p.FullName()).Warn("immediate cache upload failed")
+			c.Reporter.CacheUploadFailed([]cache.Package{p}, err)
+		}
+	}
+
+	return nil
+}
+
+// uploadPackageImmediately uploads a single package to remote cache immediately after build
+func (c *buildContext) uploadPackageImmediately(p *Package) error {
+	ver, err := p.Version()
+	if err != nil {
+		return err
+	}
+
+	c.mu.Lock()
+	// Check if already uploaded to avoid duplicate uploads
+	if _, uploaded := c.uploadedPackages[ver]; uploaded {
+		c.mu.Unlock()
+		return nil
+	}
+	c.uploadedPackages[ver] = struct{}{}
+	c.mu.Unlock()
+
+	log.WithField("package", p.FullName()).Debug("uploading package to remote cache immediately")
+
+	pkgsToUpload := []cache.Package{p}
+	err = c.RemoteCache.Upload(context.Background(), c.LocalCache, pkgsToUpload)
+	if err != nil {
+		// Remove from uploaded set on failure so it can be retried at the end
+		c.mu.Lock()
+		delete(c.uploadedPackages, ver)
+		c.mu.Unlock()
+		return err
+	}
+
 	return nil
 }
 
 func (c *buildContext) GetNewPackagesForCache() []*Package {
 	res := make([]*Package, 0, len(c.newlyBuiltPackages))
 	c.mu.Lock()
-	for _, pkg := range c.newlyBuiltPackages {
+	for ver, pkg := range c.newlyBuiltPackages {
 		if pkg.Ephemeral {
 			continue
 		}
+		// Skip packages that were already uploaded immediately
+		if _, uploaded := c.uploadedPackages[ver]; uploaded {
+			continue
+		}
 		res = append(res, pkg)
 	}
 	c.mu.Unlock()