diff --git a/api/cache.go b/api/cache.go index 5500ff5991..1288de74e1 100644 --- a/api/cache.go +++ b/api/cache.go @@ -115,6 +115,23 @@ type CacheEntryExpireResp struct { Existed bool `json:"existed"` } +// CacheEntryConfirmReq is the request body for confirming a successful +// restore, so the registry can refresh the entry's retention now that the +// backing blob is actually known to be good. +type CacheEntryConfirmReq struct { + TargetPaths []string `json:"target_paths"` + CacheKey []CacheKeyPart `json:"cache_key"` + // Scopes should be copied verbatim from the CacheEntryRetrieveResp that + // resolved this entry, not recomputed — omit only when the entry was + // retrieved unscoped. + Scopes map[string]string `json:"scopes,omitempty"` +} + +// CacheEntryConfirmResp acknowledges a confirm request. +type CacheEntryConfirmResp struct { + Message string `json:"message"` +} + // CacheEntryPeekReq is the request body for checking whether an entry exists. type CacheEntryPeekReq struct { TargetPaths []string `json:"target_paths"` @@ -284,6 +301,29 @@ func (c *Client) CacheEntryExpire(ctx context.Context, registry string, expire C return cacheResp, apiResp, nil } +// CacheEntryConfirm confirms a successful, verified restore so the registry +// can refresh the entry's retention. +func (c *Client) CacheEntryConfirm(ctx context.Context, registry string, confirm CacheEntryConfirmReq) (CacheEntryConfirmResp, *Response, error) { + ctx, span := cacheTracer.Start(ctx, "Client.CacheEntryConfirm") + defer span.End() + + var cacheResp CacheEntryConfirmResp + + req, err := c.newRequest(ctx, http.MethodPost, cachePath("/cache_registries/%s/confirm", registry), &confirm) + if err != nil { + return cacheResp, nil, cacheSpanErr(span, "failed to create request: %w", err) + } + + apiResp, err := c.cacheDo(req, &cacheResp) + if err != nil { + return cacheResp, apiResp, cacheSpanErr(span, "%w", err) + } + if apiResp.StatusCode < 200 || apiResp.StatusCode >= 300 { + return cacheResp, apiResp, cacheSpanErr(span, "failed to confirm cache restore: %s", apiResp.Status) + } + return cacheResp, apiResp, nil +} + // cachePath formats a cache API path with URL-safe escaping for path components. func cachePath(format string, args ...any) string { escaped := make([]any, len(args)) diff --git a/api/cache_test.go b/api/cache_test.go index 1209ac6925..2815cfc41c 100644 --- a/api/cache_test.go +++ b/api/cache_test.go @@ -465,3 +465,57 @@ func TestCacheEntryRetrieveAndExpire_ScopesRoundTripThroughWire(t *testing.T) { t.Error("expireResp.Existed = false, want true") } } + +func TestCacheEntryConfirm_Success(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + t.Errorf("method = %q, want %q", r.Method, http.MethodPost) + } + if !strings.HasSuffix(r.URL.Path, "/cache_registries/test-slug/confirm") { + t.Errorf("path = %q, want suffix %q", r.URL.Path, "/cache_registries/test-slug/confirm") + } + var req api.CacheEntryConfirmReq + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Fatalf("decode request body: %v", err) + } + if len(req.TargetPaths) != 1 || req.TargetPaths[0] != "node_modules" { + t.Errorf("req.TargetPaths = %v, want [node_modules]", req.TargetPaths) + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]any{"message": "Restore confirmed"}) + })) + defer server.Close() + + client := newTestCacheClient(t, server.URL) + + _, _, err := client.CacheEntryConfirm(t.Context(), "test-slug", api.CacheEntryConfirmReq{ + TargetPaths: []string{"node_modules"}, + CacheKey: []api.CacheKeyPart{{Value: "v1", Mandatory: true}}, + }) + if err != nil { + t.Fatalf("CacheEntryConfirm error = %v, want nil", err) + } +} + +// A non-2xx status must surface as an error rather than be swallowed as a +// successful confirmation. +func TestCacheEntryConfirm_NonSuccessIsError(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusNotFound) + _ = json.NewEncoder(w).Encode(map[string]string{"message": "Not Found"}) + })) + defer server.Close() + + client := newTestCacheClient(t, server.URL) + + _, _, err := client.CacheEntryConfirm(t.Context(), "test-slug", api.CacheEntryConfirmReq{ + TargetPaths: []string{"node_modules"}, + CacheKey: []api.CacheKeyPart{{Value: "v1", Mandatory: true}}, + }) + if err == nil { + t.Error("CacheEntryConfirm error = nil, want non-nil for non-2xx status") + } +} diff --git a/internal/cache/client.go b/internal/cache/client.go index 31dbe2fd20..1e4937b807 100644 --- a/internal/cache/client.go +++ b/internal/cache/client.go @@ -32,6 +32,7 @@ type cacheAPI interface { CacheEntryCommit(ctx context.Context, registry string, req api.CacheEntryCommitReq) (api.CacheEntryCommitResp, *api.Response, error) CacheEntryRetrieve(ctx context.Context, registry string, req api.CacheEntryRetrieveReq) (api.CacheEntryRetrieveResp, bool, *api.Response, error) CacheEntryExpire(ctx context.Context, registry string, req api.CacheEntryExpireReq) (api.CacheEntryExpireResp, *api.Response, error) + CacheEntryConfirm(ctx context.Context, registry string, req api.CacheEntryConfirmReq) (api.CacheEntryConfirmResp, *api.Response, error) } // Sentinel errors for common scenarios. diff --git a/internal/cache/integration_test.go b/internal/cache/integration_test.go index 996cc0561a..d10ff95731 100644 --- a/internal/cache/integration_test.go +++ b/internal/cache/integration_test.go @@ -4,6 +4,7 @@ import ( "archive/zip" "context" "crypto/rand" + "errors" "fmt" "math" "net/http" @@ -25,6 +26,15 @@ type mockAPIClient struct { registries map[string]*mockRegistry // expireCalls records the addresses passed to CacheEntryExpire expireCalls []api.CacheEntryExpireReq + // confirmCalls records the addresses passed to CacheEntryConfirm + confirmCalls []api.CacheEntryConfirmReq + // confirmErr and confirmResp, if confirmErr is set, are returned by every + // CacheEntryConfirm call instead of succeeding. + confirmErr error + confirmResp *api.Response + // confirmBlock, if true, makes CacheEntryConfirm hang until ctx is done + // instead of returning, simulating a blocked/slow confirmation request. + confirmBlock bool } type mockRegistry struct { @@ -195,6 +205,30 @@ func (m *mockAPIClient) CacheEntryExpire(ctx context.Context, registry string, r return api.CacheEntryExpireResp{Existed: existed}, nil, nil } +func (m *mockAPIClient) CacheEntryConfirm(ctx context.Context, registry string, req api.CacheEntryConfirmReq) (api.CacheEntryConfirmResp, *api.Response, error) { + m.confirmCalls = append(m.confirmCalls, req) + + if m.confirmBlock { + <-ctx.Done() + return api.CacheEntryConfirmResp{}, nil, ctx.Err() + } + + if m.confirmErr != nil { + return api.CacheEntryConfirmResp{}, m.confirmResp, m.confirmErr + } + + reg, ok := m.registries[registry] + if !ok { + return api.CacheEntryConfirmResp{}, nil, fmt.Errorf("registry not found: %s", registry) + } + + // Mirror the backend refreshing the entry's retention. + if entry, exists := reg.cache[cacheAddr(req.TargetPaths, req.CacheKey)]; exists { + entry.expiresAt = time.Now().Add(7 * 24 * time.Hour) + } + return api.CacheEntryConfirmResp{Message: "Restore confirmed"}, nil, nil +} + // createRandomFile creates a file filled with random data func createRandomFile(t *testing.T, path string, sizeBytes int64) { t.Helper() @@ -518,6 +552,89 @@ func TestCacheIntegration_RestoreCacheMiss(t *testing.T) { // recovery path: the entry still exists in the registry but its backing blob is // gone. Restore must degrade to a cache miss, invalidate the stale entry, // and let a subsequent save re-upload it. +// TestCacheIntegration_RestoreConfirmsSuccessfulExactMatch exercises the +// happy path through the real public Restore path: once the blob is +// downloaded and verified, Restore must confirm the restore with the server +// so retention is refreshed off a verified restore, not optimistically at +// retrieve. +func TestCacheIntegration_RestoreConfirmsSuccessfulExactMatch(t *testing.T) { + ctx := t.Context() + + cacheClient, cacheDir, _ := setupTestCache(t, "local_file") + mockClient := cacheClient.api.(*mockAPIClient) + + saveResult, err := cacheClient.Save(ctx, "test-cache") + if err != nil { + t.Fatalf("Save: %v", err) + } + if !saveResult.CacheEntryCreated { + t.Fatal("expected save to create an entry") + } + + // Simulate a fresh checkout so restore actually has something to do. + if err := os.RemoveAll(cacheDir); err != nil { + t.Fatalf("RemoveAll: %v", err) + } + + restoreResult, err := cacheClient.Restore(ctx, "test-cache") + if err != nil { + t.Fatalf("Restore: %v", err) + } + if !restoreResult.CacheRestored { + t.Fatal("expected a successful restore") + } + + if len(mockClient.confirmCalls) != 1 { + t.Fatalf("confirm calls = %d, want 1", len(mockClient.confirmCalls)) + } + got := mockClient.confirmCalls[0] + if len(got.CacheKey) != 1 || got.CacheKey[0].Value != "v1-test-key" { + t.Errorf("confirm targeted cache_key %+v, want single part v1-test-key", got.CacheKey) + } +} + +// TestCacheIntegration_RestoreSucceedsWhenConfirmFails covers confirmation's +// best-effort contract through the real public Restore path: by the time +// confirmRestoreSucceeded runs, the blob has already been downloaded, +// digest-verified, and extracted, so a failing confirm call is cosmetic (a +// missed retention refresh) and must not turn an otherwise-successful restore +// into a failure. +func TestCacheIntegration_RestoreSucceedsWhenConfirmFails(t *testing.T) { + ctx := t.Context() + + cacheClient, cacheDir, _ := setupTestCache(t, "local_file") + mockClient := cacheClient.api.(*mockAPIClient) + + saveResult, err := cacheClient.Save(ctx, "test-cache") + if err != nil { + t.Fatalf("Save: %v", err) + } + if !saveResult.CacheEntryCreated { + t.Fatal("expected save to create an entry") + } + + // Simulate a fresh checkout so restore actually has something to do. + if err := os.RemoveAll(cacheDir); err != nil { + t.Fatalf("RemoveAll: %v", err) + } + + // Non-retryable status so the confirm attempt fails on its first try, + // keeping the test fast while still exercising the failure path. + mockClient.confirmErr = errors.New("confirm unavailable") + mockClient.confirmResp = &api.Response{Response: &http.Response{StatusCode: http.StatusForbidden}} + + restoreResult, err := cacheClient.Restore(ctx, "test-cache") + if err != nil { + t.Fatalf("Restore should not fail when confirmation fails, got: %v", err) + } + if !restoreResult.CacheRestored { + t.Error("CacheRestored = false, want true even though confirmation failed") + } + if len(mockClient.confirmCalls) != 1 { + t.Fatalf("confirm calls = %d, want 1", len(mockClient.confirmCalls)) + } +} + func TestCacheIntegration_RestoreMissingBlobInvalidates(t *testing.T) { ctx := t.Context() @@ -564,6 +681,9 @@ func TestCacheIntegration_RestoreMissingBlobInvalidates(t *testing.T) { if len(got.CacheKey) != 1 || got.CacheKey[0].Value != "v1-test-key" { t.Errorf("expire targeted cache_key %+v, want single part v1-test-key", got.CacheKey) } + if len(mockClient.confirmCalls) != 0 { + t.Errorf("confirm calls = %d, want 0 for a restore that ended in invalidation, not success", len(mockClient.confirmCalls)) + } // A subsequent save must re-upload, proving the entry was invalidated. resaveResult, err := cacheClient.Save(ctx, "test-cache") diff --git a/internal/cache/restore.go b/internal/cache/restore.go index cee218cb9e..9fc10e6f52 100644 --- a/internal/cache/restore.go +++ b/internal/cache/restore.go @@ -330,12 +330,18 @@ func (c *client) Restore(ctx context.Context, cacheID string) (RestoreResult, er } result.CacheRestored = true + + // The restore is now fully verified (downloaded, digest-checked, and + // extracted) — confirm it with the server so retention refreshes off a + // restore that's actually known to be good. + confirmed := c.confirmRestoreSucceeded(ctx, retrieveResp) result.TotalDuration = time.Since(startTime) // Add result attributes to span span.SetAttributes( attribute.Bool("cache.hit", result.CacheHit), attribute.Bool("cache.restored", result.CacheRestored), + attribute.Bool("cache.confirmed", confirmed), attribute.Int64("cache.archive_size_bytes", result.Archive.Size), attribute.Int64("cache.written_bytes", result.Archive.WrittenBytes), attribute.Int64("cache.written_entries", result.Archive.WrittenEntries), @@ -390,6 +396,69 @@ func (c *client) invalidateStaleEntry(ctx context.Context, retrieveResp api.Cach return existed } +// confirmRestoreTimeout bounds the entire confirmRestoreSucceeded retry loop. +// Confirmation is best-effort and non-essential to a successful restore, so it +// must not be allowed to hold up job startup for anywhere near the API +// client's per-attempt timeout multiplied by roko's max attempts. +// +// A var, not a const, so tests can shrink it to exercise deadline behaviour +// without waiting out the real timeout. +var confirmRestoreTimeout = 5 * time.Second + +// confirmRestoreSucceeded tells the server this restore's blob was verified, +// so the registry can safely refresh the entry's retention now rather than +// when retrieve first served it, before the blob was known to be good. +// +// Skipped for a fallback match — mirrors the backend's own (now-removed) +// `unless entry_result.fallback_used?` bump guard: a fallback hit means the +// entry was saved under a shorter key sequence, and refreshing it would +// reset the clock on an entry the caller didn't explicitly target. +// +// Best-effort: never fails the build. The whole retry loop is bounded by +// confirmRestoreTimeout, not just each individual attempt — retrieve's exact-match +// success path (unlike the failure path invalidateStaleEntry handles) runs on +// every cache hit, so a network partition here must not be able to stall an +// otherwise-successful, latency-sensitive restore by using up 5 full client +// timeouts. +func (c *client) confirmRestoreSucceeded(ctx context.Context, retrieveResp api.CacheEntryRetrieveResp) bool { + if retrieveResp.Fallback { + return false + } + if len(retrieveResp.TargetPaths) == 0 || len(retrieveResp.CacheKey) == 0 { + slog.Warn("cannot confirm cache restore: retrieve response missing resolved address") + return false + } + + ctx, cancel := context.WithTimeout(ctx, confirmRestoreTimeout) + defer cancel() + + req := api.CacheEntryConfirmReq{ + TargetPaths: retrieveResp.TargetPaths, + CacheKey: retrieveResp.CacheKey, + Scopes: retrieveResp.Scopes, + } + err := roko.NewRetrier( + roko.WithMaxAttempts(5), + roko.WithStrategy(roko.ExponentialSubsecond(500*time.Millisecond)), + roko.WithJitter(), + ).DoWithContext(ctx, func(r *roko.Retrier) error { + _, apiResp, err := c.api.CacheEntryConfirm(ctx, c.registry, req) + if api.BreakOnNonRetryable(r, apiResp, err) { + return err + } + if err != nil { + slog.Warn("cache restore confirmation failed, retrying", "err", err, "retrier", r.String()) + return err + } + return nil + }) + if err != nil { + slog.Warn("cache restore confirmation failed", "registry", c.registry, "err", err) + return false + } + return true +} + // missCompleteMessage builds the progress text for a stale-entry miss, // distinguishing an actual deletion from a no-op so callers aren't told an // entry was invalidated when it wasn't. diff --git a/internal/cache/restore_test.go b/internal/cache/restore_test.go index 00fdc731de..8090e5cfba 100644 --- a/internal/cache/restore_test.go +++ b/internal/cache/restore_test.go @@ -9,6 +9,7 @@ import ( "runtime" "strings" "testing" + "time" "github.com/buildkite/agent/v4/api" "github.com/buildkite/agent/v4/internal/cache/store" @@ -68,6 +69,106 @@ func TestInvalidateStaleEntry_ExistedFalseIsNotReportedAsInvalidated(t *testing. } } +func TestConfirmRestoreSucceeded_EchoesScopesFromRetrieve(t *testing.T) { + mockClient := newMockAPIClient("s3") + c := &client{api: mockClient, registry: "~"} + + scopes := map[string]string{"branch": "main"} + retrieveResp := api.CacheEntryRetrieveResp{ + TargetPaths: []string{"node_modules"}, + CacheKey: []api.CacheKeyPart{{Value: "v1-test-key", Mandatory: true}}, + Scopes: scopes, + Fallback: false, + } + + confirmed := c.confirmRestoreSucceeded(t.Context(), retrieveResp) + if !confirmed { + t.Fatalf("confirmRestoreSucceeded() = false, want true") + } + + if len(mockClient.confirmCalls) != 1 { + t.Fatalf("confirm calls = %d, want 1", len(mockClient.confirmCalls)) + } + if got := mockClient.confirmCalls[0].Scopes; !reflect.DeepEqual(got, scopes) { + t.Errorf("confirm request scopes = %+v, want %+v", got, scopes) + } +} + +// TestConfirmRestoreSucceeded_SkipsFallbackMatch mirrors the backend's +// (now-removed) `unless entry_result.fallback_used?` TTL-bump guard: a +// fallback hit means the entry was saved under a shorter key sequence, and +// confirming it would refresh a blob the caller didn't explicitly target. +func TestConfirmRestoreSucceeded_SkipsFallbackMatch(t *testing.T) { + mockClient := newMockAPIClient("s3") + c := &client{api: mockClient, registry: "~"} + + retrieveResp := api.CacheEntryRetrieveResp{ + TargetPaths: []string{"node_modules"}, + CacheKey: []api.CacheKeyPart{{Value: "v1-test-key", Mandatory: true}}, + Fallback: true, + } + + confirmed := c.confirmRestoreSucceeded(t.Context(), retrieveResp) + if confirmed { + t.Error("confirmRestoreSucceeded() = true, want false for a fallback match") + } + if len(mockClient.confirmCalls) != 0 { + t.Errorf("confirm calls = %d, want 0 for a fallback match", len(mockClient.confirmCalls)) + } +} + +// TestConfirmRestoreSucceeded_DeadlineBoundsEntireOperation guards against a +// regression where the timeout resets on every retry attempt instead of +// bounding the whole confirmRestoreSucceeded call. A confirmation request that +// hangs (network partition, blocked connection) must be cut off by +// confirmRestoreTimeout once, not retried up to roko's max attempts with a +// fresh timeout budget handed to each one. +func TestConfirmRestoreSucceeded_DeadlineBoundsEntireOperation(t *testing.T) { + origTimeout := confirmRestoreTimeout + confirmRestoreTimeout = 100 * time.Millisecond + t.Cleanup(func() { confirmRestoreTimeout = origTimeout }) + + mockClient := newMockAPIClient("s3") + mockClient.confirmBlock = true + c := &client{api: mockClient, registry: "~"} + + retrieveResp := api.CacheEntryRetrieveResp{ + TargetPaths: []string{"node_modules"}, + CacheKey: []api.CacheKeyPart{{Value: "v1-test-key", Mandatory: true}}, + } + + start := time.Now() + confirmed := c.confirmRestoreSucceeded(t.Context(), retrieveResp) + elapsed := time.Since(start) + + if confirmed { + t.Error("confirmRestoreSucceeded() = true, want false for a request that never returns") + } + + // A single blocked request should be cut off at roughly the deadline. If the + // deadline instead reset per attempt, up to 5 attempts would each hang for + // the full timeout, taking several times as long. + if elapsed > confirmRestoreTimeout*2 { + t.Errorf("confirmRestoreSucceeded() took %v, want close to the %v deadline (retries must not each get their own timeout)", elapsed, confirmRestoreTimeout) + } + if len(mockClient.confirmCalls) != 1 { + t.Errorf("confirm calls = %d, want exactly 1 -- the deadline should expire before any retry is attempted", len(mockClient.confirmCalls)) + } +} + +func TestConfirmRestoreSucceeded_MissingResolvedAddress(t *testing.T) { + mockClient := newMockAPIClient("s3") + c := &client{api: mockClient, registry: "~"} + + confirmed := c.confirmRestoreSucceeded(t.Context(), api.CacheEntryRetrieveResp{}) + if confirmed { + t.Error("confirmRestoreSucceeded() = true, want false when the retrieve response has no resolved address") + } + if len(mockClient.confirmCalls) != 0 { + t.Errorf("confirm calls = %d, want 0", len(mockClient.confirmCalls)) + } +} + // TestMissCompleteMessage guards the progress text agents/users see against // claiming an invalidation happened when invalidateStaleEntry reported an // idempotent no-op (existed: false).