Skip to content

Echo entry_ref on cache expire for scope-aware invalidation #4281

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
buildkate wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Echo entry_ref on cache expire for scope-aware invalidation #4281

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 22 additions & 7 deletions api/cache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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.
Expand Down
38 changes: 36 additions & 2 deletions api/cache_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand All @@ -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) {
Expand Down Expand Up @@ -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)
Expand All @@ -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
Expand All @@ -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}},
})
Expand Down
2 changes: 1 addition & 1 deletion internal/cache/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
108 changes: 102 additions & 6 deletions internal/cache/integration_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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")
Expand All @@ -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.
Expand Down
Loading
Loading