From fc25ddb1e55f01621190909659f2aebb36133668 Mon Sep 17 00:00:00 2001 From: Amp Date: Fri, 28 Aug 2026 06:37:07 +0000 Subject: [PATCH] Echo entry_ref on cache expire for scope-aware invalidation The backend now returns an opaque entry_ref in cache retrieve responses, identifying exactly the entry that was served. When a restore degrades to a miss (missing blob, digest mismatch, unreadable archive) the agent now echoes that ref on expire, so the server deletes only the entry that was actually retrieved and never a concurrent replacement. The expire response is now typed ({message, existed}): the agent reports the entry as invalidated only when the server confirms it existed and was deleted, and logs a distinct no-op message when it was already gone. The legacy target_paths/cache_key address is still sent for servers that don't understand entry_ref yet. Part of A-1774. Amp-Thread-ID: https://ampcode.com/threads/T-01a046fc-3919-7509-af75-03f3f25535d8 Co-authored-by: Kate Sy --- api/cache.go | 29 ++++++-- api/cache_test.go | 38 +++++++++- internal/cache/client.go | 2 +- internal/cache/integration_test.go | 108 +++++++++++++++++++++++++++-- internal/cache/restore.go | 23 ++++-- 5 files changed, 179 insertions(+), 21 deletions(-) diff --git a/api/cache.go b/api/cache.go index 9f364ff40b..178ad5eab4 100644 --- a/api/cache.go +++ b/api/cache.go @@ -88,13 +88,26 @@ type CacheEntryRetrieveResp struct { Multipart bool `json:"multipart"` DownloadInstructions []string `json:"download_instructions"` Message string `json:"message"` + EntryRef string `json:"entry_ref"` } // CacheEntryExpireReq is the request body for invalidating a cache entry. -// The address should be the resolved entry, as echoed by the retrieve response. +// EntryRef is the opaque reference echoed from the retrieve response; when +// set, the server invalidates exactly the entry that was retrieved. The +// target paths and cache key are the legacy address, kept for compatibility +// with servers that don't understand entry_ref. type CacheEntryExpireReq struct { TargetPaths []string `json:"target_paths"` CacheKey []CacheKeyPart `json:"cache_key"` + EntryRef string `json:"entry_ref,omitempty"` +} + +// CacheEntryExpireResp is the response body for invalidating a cache entry. +// Existed reports whether the referenced entry existed and was deleted; +// false means it was already gone (an idempotent no-op). +type CacheEntryExpireResp struct { + Message string `json:"message"` + Existed bool `json:"existed"` } // CacheEntryPeekReq is the request body for checking whether an entry exists. @@ -243,23 +256,25 @@ func (c *Client) CacheEntryRetrieve(ctx context.Context, registry string, retrie } // CacheEntryExpire invalidates a cache entry so a subsequent save re-uploads it. -func (c *Client) CacheEntryExpire(ctx context.Context, registry string, expire CacheEntryExpireReq) (*Response, error) { +func (c *Client) CacheEntryExpire(ctx context.Context, registry string, expire CacheEntryExpireReq) (CacheEntryExpireResp, *Response, error) { ctx, span := cacheTracer.Start(ctx, "Client.CacheEntryExpire") defer span.End() + var expireResp CacheEntryExpireResp + req, err := c.newRequest(ctx, http.MethodPost, cachePath("/cache_registries/%s/expire", registry), &expire) if err != nil { - return nil, cacheSpanErr(span, "failed to create request: %w", err) + return expireResp, nil, cacheSpanErr(span, "failed to create request: %w", err) } - apiResp, err := c.cacheDo(req, &struct{}{}) + apiResp, err := c.cacheDo(req, &expireResp) if err != nil { - return apiResp, cacheSpanErr(span, "%w", err) + return expireResp, apiResp, cacheSpanErr(span, "%w", err) } if apiResp.StatusCode < 200 || apiResp.StatusCode >= 300 { - return apiResp, cacheSpanErr(span, "failed to expire cache entry: %s", apiResp.Status) + return expireResp, apiResp, cacheSpanErr(span, "failed to expire cache entry: %s", apiResp.Status) } - return apiResp, nil + return expireResp, apiResp, nil } // cachePath formats a cache API path with URL-safe escaping for path components. diff --git a/api/cache_test.go b/api/cache_test.go index fe0d342c4b..ff31a15c64 100644 --- a/api/cache_test.go +++ b/api/cache_test.go @@ -218,6 +218,7 @@ func TestCacheEntryRetrieve_Success(t *testing.T) { Multipart: false, DownloadInstructions: []string{"curl -X GET..."}, Message: "Retrieved successfully", + EntryRef: "opaque-entry-ref", }) })) defer server.Close() @@ -240,6 +241,9 @@ func TestCacheEntryRetrieve_Success(t *testing.T) { if resp.Fallback { t.Error("resp.Fallback = true, want false") } + if got, want := resp.EntryRef, "opaque-entry-ref"; got != want { + t.Errorf("resp.EntryRef = %q, want %q", got, want) + } } func TestCacheEntryRetrieve_NotFound(t *testing.T) { @@ -332,6 +336,9 @@ func TestCacheEntryExpire_Success(t *testing.T) { if len(req.TargetPaths) != 1 || req.TargetPaths[0] != "node_modules" { t.Errorf("req.TargetPaths = %v, want [node_modules]", req.TargetPaths) } + if got, want := req.EntryRef, "opaque-entry-ref"; got != want { + t.Errorf("req.EntryRef = %q, want %q", got, want) + } w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusOK) @@ -341,13 +348,40 @@ func TestCacheEntryExpire_Success(t *testing.T) { client := newTestCacheClient(t, server.URL) - _, err := client.CacheEntryExpire(t.Context(), "test-slug", api.CacheEntryExpireReq{ + resp, _, err := client.CacheEntryExpire(t.Context(), "test-slug", api.CacheEntryExpireReq{ TargetPaths: []string{"node_modules"}, CacheKey: []api.CacheKeyPart{{Value: "v1", Mandatory: true}}, + EntryRef: "opaque-entry-ref", + }) + if err != nil { + t.Fatalf("CacheEntryExpire error = %v, want nil", err) + } + if !resp.Existed { + t.Error("resp.Existed = false, want true") + } +} + +// An expire of an entry that is already gone is an idempotent no-op: the +// server responds 2xx with existed=false, and no error is returned. +func TestCacheEntryExpire_AlreadyGone(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]any{"message": "Cache entry not found", "existed": false}) + })) + defer server.Close() + + client := newTestCacheClient(t, server.URL) + + resp, _, err := client.CacheEntryExpire(t.Context(), "test-slug", api.CacheEntryExpireReq{ + EntryRef: "opaque-entry-ref", }) if err != nil { t.Fatalf("CacheEntryExpire error = %v, want nil", err) } + if resp.Existed { + t.Error("resp.Existed = true, want false") + } } // A non-2xx status (e.g. a missing /expire route during backend rollout, or a @@ -363,7 +397,7 @@ func TestCacheEntryExpire_NonSuccessIsError(t *testing.T) { client := newTestCacheClient(t, server.URL) - _, err := client.CacheEntryExpire(t.Context(), "test-slug", api.CacheEntryExpireReq{ + _, _, err := client.CacheEntryExpire(t.Context(), "test-slug", api.CacheEntryExpireReq{ TargetPaths: []string{"node_modules"}, CacheKey: []api.CacheKeyPart{{Value: "v1", Mandatory: true}}, }) diff --git a/internal/cache/client.go b/internal/cache/client.go index eede5562c4..31dbe2fd20 100644 --- a/internal/cache/client.go +++ b/internal/cache/client.go @@ -31,7 +31,7 @@ type cacheAPI interface { CacheEntryCreate(ctx context.Context, registry string, req api.CacheEntryCreateReq) (api.CacheEntryCreateResp, *api.Response, error) 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.Response, error) + CacheEntryExpire(ctx context.Context, registry string, req api.CacheEntryExpireReq) (api.CacheEntryExpireResp, *api.Response, error) } // Sentinel errors for common scenarios. diff --git a/internal/cache/integration_test.go b/internal/cache/integration_test.go index fc431ed8c3..f17aeac15c 100644 --- a/internal/cache/integration_test.go +++ b/internal/cache/integration_test.go @@ -24,6 +24,10 @@ type mockAPIClient struct { registries map[string]*mockRegistry // expireCalls records the addresses passed to CacheEntryExpire expireCalls []api.CacheEntryExpireReq + // expireForceNotExisted makes CacheEntryExpire report existed=false + // without deleting anything, simulating an entry that was concurrently + // replaced (entry ref no longer matches) or already gone. + expireForceNotExisted bool } type mockRegistry struct { @@ -41,6 +45,9 @@ type mockCacheEntry struct { committed bool expiresAt time.Time platform string + // entryRef is the opaque expiration reference the backend mints for each + // stored entry and echoes in retrieve responses. + entryRef string } // cacheAddr builds the v2 entry address from the order-insensitive target_paths @@ -126,6 +133,7 @@ func (m *mockAPIClient) CacheEntryCreate(ctx context.Context, registry string, r committed: false, expiresAt: time.Now().Add(7 * 24 * time.Hour), platform: req.Platform, + entryRef: fmt.Sprintf("entry-ref-%d", time.Now().UnixNano()), } reg.cache[cacheAddr(req.TargetPaths, req.CacheKey)] = entry @@ -166,24 +174,43 @@ func (m *mockAPIClient) CacheEntryRetrieve(ctx context.Context, registry string, Blobs: entry.blobs, Fallback: false, ExpiresAt: entry.expiresAt, + EntryRef: entry.entryRef, }, true, nil, nil } return api.CacheEntryRetrieveResp{Message: api.CacheEntryNotFound}, false, nil, nil } -func (m *mockAPIClient) CacheEntryExpire(ctx context.Context, registry string, req api.CacheEntryExpireReq) (*api.Response, error) { +func (m *mockAPIClient) CacheEntryExpire(ctx context.Context, registry string, req api.CacheEntryExpireReq) (api.CacheEntryExpireResp, *api.Response, error) { m.expireCalls = append(m.expireCalls, req) reg, ok := m.registries[registry] if !ok { - return nil, fmt.Errorf("registry not found: %s", registry) + return api.CacheEntryExpireResp{}, nil, fmt.Errorf("registry not found: %s", registry) } - // Mirror the backend's delete_item so a subsequent save - // re-uploads the invalidated entry. - delete(reg.cache, cacheAddr(req.TargetPaths, req.CacheKey)) - return nil, nil + if m.expireForceNotExisted { + return api.CacheEntryExpireResp{Message: "Cache entry not found", Existed: false}, nil, nil + } + + // Mirror the backend: an entry ref deletes exactly the entry it was minted + // for; the legacy address falls back to the composed cache key. + if req.EntryRef != "" { + for addr, entry := range reg.cache { + if entry.entryRef == req.EntryRef { + delete(reg.cache, addr) + return api.CacheEntryExpireResp{Message: "Cache entry expired", Existed: true}, nil, nil + } + } + return api.CacheEntryExpireResp{Message: "Cache entry not found", Existed: false}, nil, nil + } + + addr := cacheAddr(req.TargetPaths, req.CacheKey) + if _, exists := reg.cache[addr]; !exists { + return api.CacheEntryExpireResp{Message: "Cache entry not found", Existed: false}, nil, nil + } + delete(reg.cache, addr) + return api.CacheEntryExpireResp{Message: "Cache entry expired", Existed: true}, nil, nil } // createRandomFile creates a file filled with random data @@ -524,6 +551,16 @@ func TestCacheIntegration_RestoreMissingBlobInvalidates(t *testing.T) { t.Fatal("expected initial save to create an entry") } + // Capture the entry ref the mock minted at save so we can assert the expire + // echoes exactly the retrieved entry's ref. + var wantRef string + for _, e := range mockClient.registries["~"].cache { + wantRef = e.entryRef + } + if wantRef == "" { + t.Fatal("expected the saved mock entry to have an entry ref") + } + // Simulate the blob being lifecycle/TTL-deleted while the entry survives. blobEntries, err := os.ReadDir(storageDir) if err != nil { @@ -555,6 +592,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 got.EntryRef != wantRef { + t.Errorf("expire echoed EntryRef = %q, want the retrieved entry's ref %q", got.EntryRef, wantRef) + } // A subsequent save must re-upload, proving the entry was invalidated. resaveResult, err := cacheClient.Save(ctx, "test-cache") @@ -898,6 +938,15 @@ func TestCacheIntegration_RestorePoisonedBlobIsMiss(t *testing.T) { t.Fatal("expected save to create an entry") } + // Capture the legitimate entry's ref so we can assert the expire echoes it. + var wantRef string + for _, e := range mockClient.registries["~"].cache { + wantRef = e.entryRef + } + if wantRef == "" { + t.Fatal("expected the saved mock entry to have an entry ref") + } + // Poison the store: keep the legitimate blob's content-addressed name but // swap in the poisoned bytes, so the fingerprint no longer matches the name. blobPath := filepath.Join(storageDir, saveResult.Archive.Sha256Sum) @@ -942,6 +991,9 @@ func TestCacheIntegration_RestorePoisonedBlobIsMiss(t *testing.T) { if len(expired.CacheKey) != 1 || expired.CacheKey[0].Value != "v1-test-key" { t.Errorf("expire targeted cache_key %+v, want single part v1-test-key", expired.CacheKey) } + if expired.EntryRef != wantRef { + t.Errorf("expire echoed EntryRef = %q, want the retrieved entry's ref %q", expired.EntryRef, wantRef) + } // A subsequent save must re-upload, proving the entry was invalidated. resaveResult, err := cacheClient.Save(ctx, "test-cache") @@ -953,6 +1005,50 @@ func TestCacheIntegration_RestorePoisonedBlobIsMiss(t *testing.T) { } } +// TestCacheIntegration_RestoreMissingBlobExpireNoOp covers the scope-aware +// no-op: the entry was concurrently replaced or removed between retrieve and +// expire, so the server reports existed=false and nothing is deleted. The +// restore must still be a clean miss rather than an error. +func TestCacheIntegration_RestoreMissingBlobExpireNoOp(t *testing.T) { + ctx := t.Context() + + cacheClient, _, storageDir := setupTestCache(t, "local_file") + mockClient := cacheClient.api.(*mockAPIClient) + + if _, err := cacheClient.Save(ctx, "test-cache"); err != nil { + t.Fatalf("Save: %v", err) + } + + // Simulate the blob disappearing while the entry survives. + blobEntries, err := os.ReadDir(storageDir) + if err != nil { + t.Fatalf("ReadDir: %v", err) + } + for _, e := range blobEntries { + if err := os.RemoveAll(filepath.Join(storageDir, e.Name())); err != nil { + t.Fatalf("RemoveAll: %v", err) + } + } + + // The expire lands after the entry was concurrently replaced: the server + // reports existed=false and deletes nothing. + mockClient.expireForceNotExisted = true + + restoreResult, err := cacheClient.Restore(ctx, "test-cache") + if err != nil { + t.Fatalf("Restore with a no-op expire should not error, got: %v", err) + } + if restoreResult.CacheRestored { + t.Error("missing blob should degrade to CacheRestored=false") + } + if restoreResult.CacheHit { + t.Error("missing blob should not be a cache hit") + } + if len(mockClient.expireCalls) != 1 { + t.Fatalf("expire calls = %d, want 1", len(mockClient.expireCalls)) + } +} + // TODO: restore fallback-matching coverage (was TestCacheIntegration_RestoreWithFallback, // removed in the v2 migration) once agent-side fallback_limit parsing lands and the agent // sends mandatory:false parts. Until then the agent only addresses exact matches. diff --git a/internal/cache/restore.go b/internal/cache/restore.go index 7919d55c51..e408870cb3 100644 --- a/internal/cache/restore.go +++ b/internal/cache/restore.go @@ -192,7 +192,7 @@ func (c *client) Restore(ctx context.Context, cacheID string) (RestoreResult, er attribute.Bool("cache.invalidated", invalidated), ) span.SetStatus(codes.Ok, "cache miss (missing blob)") - c.callProgress(cacheID, "complete", "Cache miss (missing blob, invalidated stale entry)", 0, 0) + c.callProgress(cacheID, "complete", "Cache miss (missing blob)", 0, 0) return result, nil } if errors.Is(err, ErrDigestMismatch) { @@ -214,7 +214,7 @@ func (c *client) Restore(ctx context.Context, cacheID string) (RestoreResult, er attribute.Bool("cache.invalidated", invalidated), ) span.SetStatus(codes.Ok, "cache miss (digest mismatch)") - c.callProgress(cacheID, "complete", "Cache miss (blob digest mismatch, invalidated entry)", 0, 0) + c.callProgress(cacheID, "complete", "Cache miss (blob digest mismatch)", 0, 0) return result, nil } span.RecordError(err) @@ -353,22 +353,30 @@ func (c *client) Restore(ctx context.Context, cacheID string) (RestoreResult, er // invalidateStaleEntry uses the retrieve response to expire a cache entry whose // blob is missing or fails digest verification, so a subsequent save re-uploads it. +// It returns true only when the server confirms the entry existed and was deleted; +// false means the entry was already gone (or the request failed), so a concurrent +// replacement is left intact. func (c *client) invalidateStaleEntry(ctx context.Context, retrieveResp api.CacheEntryRetrieveResp) bool { - if len(retrieveResp.TargetPaths) == 0 || len(retrieveResp.CacheKey) == 0 { - slog.Warn("cannot invalidate stale cache entry: retrieve response missing resolved address") + if retrieveResp.EntryRef == "" && (len(retrieveResp.TargetPaths) == 0 || len(retrieveResp.CacheKey) == 0) { + slog.Warn("cannot invalidate stale cache entry: retrieve response missing entry ref and resolved address") return false } + // EntryRef scopes the expire to exactly the entry that was retrieved; the + // legacy resolved address is sent alongside for servers that don't + // understand entry_ref yet. req := api.CacheEntryExpireReq{ TargetPaths: retrieveResp.TargetPaths, CacheKey: retrieveResp.CacheKey, + EntryRef: retrieveResp.EntryRef, } + var expireResp api.CacheEntryExpireResp 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.CacheEntryExpire(ctx, c.registry, req) + resp, apiResp, err := c.api.CacheEntryExpire(ctx, c.registry, req) if api.BreakOnNonRetryable(r, apiResp, err) { return err } @@ -376,12 +384,17 @@ func (c *client) invalidateStaleEntry(ctx context.Context, retrieveResp api.Cach slog.Warn("cache entry invalidation failed, retrying", "err", err, "retrier", r.String()) return err } + expireResp = resp return nil }) if err != nil { slog.Warn("cache entry invalidation failed", "registry", c.registry, "err", err) return false } + if !expireResp.Existed { + slog.Info("stale cache entry was already gone, nothing invalidated", "registry", c.registry) + return false + } return true }