feat: validate remote archive/opcode caches against the origin (#195)

## Summary

Before this change, `pkg/executor` cached every HTTP(S) download under a
path derived from `sha256(URL)`. Once downloaded, the cached file was
reused forever for the same URL with zero revalidation against the
origin. If you republished a `.tar.gz` at the same URL (common for
GitHub release assets, re-built CI artifacts), every runner kept using
the stale copy indefinitely — you had to manually delete the cache dir
to force a re-download.

This PR validates cached remote files against the origin on every
resolve using standard HTTP validators (`ETag` / `Last-Modified` /
`Content-Length`):

- On each resolve, send a HEAD to the origin and compare the response to
a sidecar `<cached-file>.meta` JSON stored next to the cached blob.
- **Validators match** → reuse cache (just cost: one HEAD roundtrip).
- **Validators differ** → re-download and refresh the sidecar.
- **No validators returned** or **HEAD fails** → reuse the cache with a
warning so transient network errors or dumb servers don't force
re-downloads.
- **Legacy cache entry without a sidecar** (populated by an older
benchmarkoor) → re-download once to refresh the validators.

Applies to all three HTTP cache sites in `pkg/executor`:

- `archive.file` — single file
- `archive.parts[]` — each part validated independently
- `archive.parts` combined concatenated file — rebuilt when any
underlying part was re-downloaded
- `opcode_source.file`

### Cache layout

```
<cacheDir>/archive-<hash>         # the cached file (unchanged)
<cacheDir>/archive-<hash>.meta    # new: {url, etag, last_modified, content_length}
```

## Test plan

- [x] Unit tests (`pkg/executor/cache_test.go`):
  - `TestFetchCached_HitWhenValidatorsMatch`
  - `TestFetchCached_MissWhenETagChanges`
  - `TestFetchCached_LegacySidecarMissingReDownloads`
  - `TestFetchCached_NoValidatorsFallsBackToCache`
  - `TestFetchCached_HeadFailureFallsBackToCache`
  - `TestFetchCached_CacheMissCoexistsWithOtherPrefixes`
- [x] `TestArchiveSource_PartsReconcatenateOnOriginChange` verifies that
flipping a part's ETag triggers re-concatenation of the combined archive
- [x] All existing archive_source tests still pass
- [x] `go vet -tags exclude_graphdriver_btrfs ./...` clean
- [x] Manual: republish a `.tar.gz` at a previously-used URL and confirm
the next run re-downloads; confirm an unchanged URL continues to use the
cache (only adds one HEAD roundtrip)

Author: skylenet

pkg/executor/archive_source.go

@@ -107,10 +107,12 @@ func (s *ArchiveSource) GetSourceInfo() (*SuiteSource, error) {
 	return &SuiteSource{Archive: info}, nil
 }
 
-// resolveFile returns the local path to the archive file. For URLs, it checks
-// the cache directory first and only downloads if the file is not already cached.
-// When the config uses `parts`, the parts are downloaded (with caching) and
-// concatenated into a single cached file.
+// resolveFile returns the local path to the archive file. For URLs, it
+// checks the cache directory first and validates any existing cached copy
+// against the origin (via HEAD + ETag/Last-Modified) before reuse. If the
+// remote has changed or has no cache entry yet, a fresh copy is downloaded.
+// When the config uses `parts`, the parts are resolved (each with its own
+// validation) and concatenated into a single cached file.
 func (s *ArchiveSource) resolveFile(ctx context.Context) (string, error) {
 	if len(s.cfg.Parts) > 0 {
 		return s.resolvePartsFile(ctx)
@@ -119,43 +121,14 @@ func (s *ArchiveSource) resolveFile(ctx context.Context) (string, error) {
 	file := s.cfg.File
 
 	if strings.HasPrefix(file, "http://") || strings.HasPrefix(file, "https://") {
-		cachedPath := s.cachedArchivePath()
-
-		if _, err := os.Stat(cachedPath); err == nil {
-			s.log.WithFields(logrus.Fields{
-				"url":  file,
-				"path": cachedPath,
-			}).Info("Using cached archive")
-
-			return cachedPath, nil
-		}
-
-		s.log.WithField("url", file).Info("Downloading archive")
-
 		downloadURL, token := s.resolveDownloadURL(file)
 
-		// Download to a temp file first, then rename for atomic cache writes.
-		tmpPath := cachedPath + ".tmp"
-
-		if err := os.MkdirAll(filepath.Dir(cachedPath), 0755); err != nil {
-			return "", fmt.Errorf("creating cache directory: %w", err)
-		}
-
-		if err := downloadToFile(ctx, downloadURL, tmpPath, token, s.log); err != nil {
-			_ = os.Remove(tmpPath)
-
+		res, err := fetchCached(ctx, s.log, file, downloadURL, token, s.cacheDir, "archive")
+		if err != nil {
 			return "", err
 		}
 
-		if err := os.Rename(tmpPath, cachedPath); err != nil {
-			_ = os.Remove(tmpPath)
-
-			return "", fmt.Errorf("caching archive: %w", err)
-		}
-
-		s.log.WithField("path", cachedPath).Info("Archive cached")
-
-		return cachedPath, nil
+		return res.Path, nil
 	}
 
 	// Local file path — resolve relative paths.
@@ -175,62 +148,60 @@ func (s *ArchiveSource) resolveFile(ctx context.Context) (string, error) {
 	return file, nil
 }
 
-// cachedArchivePath returns a stable file path in the cache directory derived
-// from the configured URL, so repeated runs reuse the same downloaded file.
-func (s *ArchiveSource) cachedArchivePath() string {
-	hash := sha256.Sum256([]byte(s.cfg.File))
-	name := "archive-" + hex.EncodeToString(hash[:8])
-
+// resolvePartsFile resolves each configured part (validating HTTP caches
+// against the origin) and concatenates them into a single combined cache
+// file. The combined file is rebuilt whenever any part was (re-)downloaded
+// during resolution, so updating a part on the origin causes the combined
+// archive to be regenerated on the next run.
+func (s *ArchiveSource) resolvePartsFile(ctx context.Context) (string, error) {
 	cacheDir := s.cacheDir
 	if cacheDir == "" {
 		cacheDir = os.TempDir()
 	}
 
-	return filepath.Join(cacheDir, name)
-}
-
-// resolvePartsFile downloads (with caching) all configured parts and
-// concatenates them in order into a single cached file, returning its path.
-// Local paths and URLs can be mixed freely in the parts list.
-func (s *ArchiveSource) resolvePartsFile(ctx context.Context) (string, error) {
-	cacheDir := s.cacheDir
-	if cacheDir == "" {
-		cacheDir = os.TempDir()
+	if err := os.MkdirAll(cacheDir, 0755); err != nil {
+		return "", fmt.Errorf("creating cache directory: %w", err)
 	}
 
 	// Combined cache key is derived from the full ordered parts list so any
 	// change to the list produces a fresh combined file.
 	combined := sha256.Sum256([]byte(strings.Join(s.cfg.Parts, "\n")))
 	combinedPath := filepath.Join(cacheDir, "archive-parts-"+hex.EncodeToString(combined[:8]))
 
-	if _, err := os.Stat(combinedPath); err == nil {
-		s.log.WithField("path", combinedPath).Info("Using cached combined archive")
-
-		return combinedPath, nil
-	}
-
-	if err := os.MkdirAll(cacheDir, 0755); err != nil {
-		return "", fmt.Errorf("creating cache directory: %w", err)
-	}
-
-	// Resolve each part to a local file path (downloading URLs as needed).
+	// Resolve each part to a local file path (validating URL caches).
 	partPaths := make([]string, 0, len(s.cfg.Parts))
 
+	anyPartChanged := false
+
 	for i, part := range s.cfg.Parts {
 		s.log.WithFields(logrus.Fields{
 			"part":  i + 1,
 			"total": len(s.cfg.Parts),
 			"ref":   part,
 		}).Info("Resolving archive part")
 
-		partPath, err := s.resolvePart(ctx, part, cacheDir)
+		partPath, changed, err := s.resolvePart(ctx, part, cacheDir)
 		if err != nil {
 			return "", fmt.Errorf("resolving part %d (%s): %w", i+1, part, err)
 		}
 
+		if changed {
+			anyPartChanged = true
+		}
+
 		partPaths = append(partPaths, partPath)
 	}
 
+	// Reuse the combined file only when it exists AND no underlying part
+	// was refreshed during this call.
+	if !anyPartChanged {
+		if _, err := os.Stat(combinedPath); err == nil {
+			s.log.WithField("path", combinedPath).Info("Using cached combined archive")
+
+			return combinedPath, nil
+		}
+	}
+
 	// Concatenate all parts into a temporary file, then atomically rename.
 	tmpPath := combinedPath + ".tmp"
 
@@ -251,56 +222,38 @@ func (s *ArchiveSource) resolvePartsFile(ctx context.Context) (string, error) {
 	return combinedPath, nil
 }
 
-// resolvePart resolves a single part reference (URL or local path) to a local
-// file path, downloading and caching remote parts in cacheDir.
-func (s *ArchiveSource) resolvePart(ctx context.Context, part, cacheDir string) (string, error) {
+// resolvePart resolves a single part reference (URL or local path) to a
+// local file path. Remote parts are cache-validated via fetchCached.
+// Returns (path, changed, error), where `changed` is true when the call
+// produced a fresh download; used by resolvePartsFile to know when the
+// concatenated combined file needs to be rebuilt.
+func (s *ArchiveSource) resolvePart(ctx context.Context, part, cacheDir string) (string, bool, error) {
 	if strings.HasPrefix(part, "http://") || strings.HasPrefix(part, "https://") {
-		hash := sha256.Sum256([]byte(part))
-		cachedPath := filepath.Join(cacheDir, "archive-part-"+hex.EncodeToString(hash[:8]))
-
-		if _, err := os.Stat(cachedPath); err == nil {
-			s.log.WithFields(logrus.Fields{
-				"url":  part,
-				"path": cachedPath,
-			}).Info("Using cached archive part")
-
-			return cachedPath, nil
-		}
-
 		downloadURL, token := s.resolveDownloadURL(part)
 
-		tmpPath := cachedPath + ".tmp"
-
-		if err := downloadToFile(ctx, downloadURL, tmpPath, token, s.log); err != nil {
-			_ = os.Remove(tmpPath)
-
-			return "", err
-		}
-
-		if err := os.Rename(tmpPath, cachedPath); err != nil {
-			_ = os.Remove(tmpPath)
-
-			return "", fmt.Errorf("caching archive part: %w", err)
+		res, err := fetchCached(ctx, s.log, part, downloadURL, token, cacheDir, "archive-part")
+		if err != nil {
+			return "", false, err
 		}
 
-		return cachedPath, nil
+		return res.Path, res.Changed, nil
 	}
 
 	// Local file path — resolve relative paths.
 	if !filepath.IsAbs(part) {
 		absPath, err := filepath.Abs(part)
 		if err != nil {
-			return "", fmt.Errorf("resolving path %q: %w", part, err)
+			return "", false, fmt.Errorf("resolving path %q: %w", part, err)
 		}
 
 		part = absPath
 	}
 
 	if _, err := os.Stat(part); os.IsNotExist(err) {
-		return "", fmt.Errorf("archive part %q does not exist", part)
+		return "", false, fmt.Errorf("archive part %q does not exist", part)
 	}
 
-	return part, nil
+	return part, false, nil
 }
 
 // concatFiles concatenates src files (in order) into dst. dst is created (or

pkg/executor/archive_source_test.go

@@ -7,12 +7,16 @@ import (
 	"compress/gzip"
 	"context"
 	"crypto/rand"
+	"crypto/sha256"
+	"encoding/hex"
 	"io"
 	"net/http"
 	"net/http/httptest"
 	"os"
 	"path/filepath"
 	"strconv"
+	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -356,6 +360,131 @@ func splitBytes(b []byte, n int) [][]byte {
 	return chunks
 }
 
+func TestArchiveSource_PartsReconcatenateOnOriginChange(t *testing.T) {
+	tmpDir := t.TempDir()
+
+	// Build two tar.gz archives, A (one test) and B (two tests). Split
+	// each into two parts. Serve A's parts initially, then flip the ETag
+	// on the second part's server to serve B's second half and verify the
+	// combined file is regenerated on the next Prepare().
+	pathA := filepath.Join(tmpDir, "a.tar.gz")
+	createTestTarGz(t, pathA, map[string]string{
+		"mytest/test/a.txt": "content-a",
+	})
+	pathB := filepath.Join(tmpDir, "b.tar.gz")
+	createTestTarGz(t, pathB, map[string]string{
+		"mytest/test/b1.txt": "content-b1",
+		"mytest/test/b2.txt": "content-b2",
+	})
+
+	bytesA, err := os.ReadFile(pathA)
+	require.NoError(t, err)
+	bytesB, err := os.ReadFile(pathB)
+	require.NoError(t, err)
+
+	chunksA := splitBytes(bytesA, 2)
+	chunksB := splitBytes(bytesB, 2)
+
+	var currentPart2 atomic.Value
+	currentPart2.Store(chunksA[1])
+
+	var etagPart2 atomic.Value
+	etagPart2.Store(`"a"`)
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var data []byte
+		var etag string
+
+		switch r.URL.Path {
+		case "/part1":
+			data = chunksA[0]
+			etag = `"a"`
+		case "/part2":
+			data = currentPart2.Load().([]byte)
+			etag = etagPart2.Load().(string)
+		default:
+			http.NotFound(w, r)
+			return
+		}
+
+		w.Header().Set("ETag", etag)
+		w.Header().Set("Content-Length", strconv.Itoa(len(data)))
+		w.WriteHeader(http.StatusOK)
+
+		if r.Method == http.MethodGet {
+			_, _ = w.Write(data)
+		}
+	}))
+	defer srv.Close()
+
+	log := logrus.New()
+	log.SetLevel(logrus.DebugLevel)
+
+	makeSrc := func() *ArchiveSource {
+		return &ArchiveSource{
+			log:      log.WithField("source", "archive"),
+			cacheDir: tmpDir,
+			cfg: &config.ArchiveSourceConfig{
+				Parts: []string{srv.URL + "/part1", srv.URL + "/part2"},
+				Steps: &config.StepsConfig{
+					Test: []string{"mytest/test/*"},
+				},
+			},
+		}
+	}
+
+	// First run: downloads both parts and writes the combined archive.
+	s1 := makeSrc()
+	result, err := s1.Prepare(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, 1, len(result.Tests), "first run sees only A's single test")
+	require.NoError(t, s1.Cleanup())
+
+	// Origin publishes a new second half with a different ETag. The
+	// rebuilt combined archive must reflect it.
+	currentPart2.Store(chunksB[1])
+	etagPart2.Store(`"b"`)
+
+	// We also need to replace part1 on the origin since A and B differ
+	// from byte 0 — but here we care about the control flow, so instead
+	// we make A -> A+B2 (first half of A, second half of B). That's
+	// gibberish as a tar.gz; switch both halves to B.
+	// Simpler: also update part1 so the combined is B.
+	// Since we only have one endpoint for /part1, stub it out:
+	//  -> we already serve chunksA[0] unconditionally. Update via a flag.
+	//
+	// For this test the primary assertion is that the re-download of
+	// part2 (ETag change) triggers a rebuild of the combined file, even
+	// if its content is garbled.
+
+	s2 := makeSrc()
+	_, err = s2.Prepare(context.Background())
+	// Extraction will likely fail (chunksA[0] + chunksB[1] is not a valid
+	// tar.gz). We only care that the combined cached file got rebuilt.
+	_ = err
+	require.NoError(t, s2.Cleanup())
+
+	// Verify: the combined cached file now has byte content reflecting
+	// chunksA[0] + chunksB[1], not the original chunksA[0] + chunksA[1].
+	combinedKey := filepath.Join(
+		tmpDir,
+		"archive-parts-"+shortHashCombined(srv.URL+"/part1", srv.URL+"/part2"),
+	)
+	got, err := os.ReadFile(combinedKey) //nolint:gosec // test-only
+	require.NoError(t, err)
+
+	want := append([]byte{}, chunksA[0]...)
+	want = append(want, chunksB[1]...)
+	assert.Equal(t, want, got, "combined file must be rebuilt when any part ETag changes")
+}
+
+// shortHashCombined mirrors how resolvePartsFile computes its combined
+// cache key from the ordered parts list.
+func shortHashCombined(parts ...string) string {
+	h := sha256.Sum256([]byte(strings.Join(parts, "\n")))
+	return hex.EncodeToString(h[:8])
+}
+
 func TestArchiveSource_PrepareWithParts(t *testing.T) {
 	tmpDir := t.TempDir()