From ddd35f23a20830838c87e2cc75eb1f23e895d85d Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Tue, 8 Sep 2026 10:43:49 +0000 Subject: [PATCH] fix(worker): claim cached ephemeral inputs Keep startup-scanned cache hits owned while inference uses them and reconcile their actual size against capacity. Assisted-by: Codex:gpt-6 --- core/services/worker/ephemeral_capacity.go | 60 +++++++++++++++ core/services/worker/file_staging.go | 2 +- .../worker/file_staging_release_test.go | 73 +++++++++++++++++++ 3 files changed, 134 insertions(+), 1 deletion(-) diff --git a/core/services/worker/ephemeral_capacity.go b/core/services/worker/ephemeral_capacity.go index 0fb5c0a0b..38a2aeff6 100644 --- a/core/services/worker/ephemeral_capacity.go +++ b/core/services/worker/ephemeral_capacity.go @@ -257,6 +257,66 @@ func (g *EphemeralCapacityGuard) Commit(path string) error { return nil } +// Claim makes an existing regular file request-owned until Release. It +// serializes with recovery deletion, preserves bytes already discovered by a +// startup scan, and only admits growth that fits the configured capacity. +// Repeated claims of the same committed file are idempotent. +func (g *EphemeralCapacityGuard) Claim(path string) error { + cleanPath, root, err := g.registeredPath(path) + if err != nil { + return err + } + + g.mu.Lock() + defer g.mu.Unlock() + + info, err := os.Lstat(cleanPath) + if err != nil { + return fmt.Errorf("stating claimed ephemeral file %q: %w", cleanPath, err) + } + if !info.Mode().IsRegular() { + return fmt.Errorf("claimed ephemeral path %q is not a regular file", cleanPath) + } + + entry, found := g.entries[cleanPath] + if found && entry.isOwned() && entry.state != ephemeralCapacityCommitted { + return &EphemeralReservationConflictError{ + Path: cleanPath, ActiveBytes: entry.reserved, RequestedBytes: info.Size(), + } + } + + accounted := int64(0) + if found { + accounted = entry.baseline + entry.reserved + } + delta := info.Size() - accounted + if delta > 0 { + if err := g.checkCapacityLocked(root, delta); err != nil { + // The file already occupies the filesystem, so keep accounting + // truthful even though a new request cannot claim it. Preserve + // ownership if an earlier claim is still awaiting Release. + state := ephemeralCapacityExisting + owned := false + if found && entry.state == ephemeralCapacityCommitted && entry.isOwned() { + state = ephemeralCapacityCommitted + owned = true + } + entry = ephemeralCapacityEntry{ + state: state, owned: owned, baseline: info.Size(), + } + g.entries[cleanPath] = entry + g.usage += delta + return err + } + } + g.usage += delta + entry = ephemeralCapacityEntry{ + state: ephemeralCapacityCommitted, owned: true, baseline: info.Size(), + } + g.entries[cleanPath] = entry + return nil +} + // Release forgets all accounting for path. It is safe to call repeatedly. func (g *EphemeralCapacityGuard) Release(path string) error { cleanPath, _, err := g.registeredPath(path) diff --git a/core/services/worker/file_staging.go b/core/services/worker/file_staging.go index 6378120dd..7afcae19e 100644 --- a/core/services/worker/file_staging.go +++ b/core/services/worker/file_staging.go @@ -287,7 +287,7 @@ func ensureWorkerFile(ctx context.Context, fm *storage.FileManager, capacity *Ep if !info.Mode().IsRegular() { return "", fmt.Errorf("ephemeral cache path %q is not a regular file", cachePath) } - if err := capacity.Account(cachePath, info.Size()); err != nil { + if err := capacity.Claim(cachePath); err != nil { return "", err } return cachePath, nil diff --git a/core/services/worker/file_staging_release_test.go b/core/services/worker/file_staging_release_test.go index 07e91b375..d94b0ade7 100644 --- a/core/services/worker/file_staging_release_test.go +++ b/core/services/worker/file_staging_release_test.go @@ -70,6 +70,79 @@ func (m *releaseMessagingClient) IsConnected() bool { return true } func (m *releaseMessagingClient) Close() {} var _ = Describe("Worker exact-key staging release", func() { + It("claims a startup-scanned cache hit against stale recovery until release", func() { + cacheDir := GinkgoT().TempDir() + root := filepath.Join(cacheDir, "ephemeral") + key := "ephemeral/audio/request-id/input.wav" + cachePath := filepath.Join(cacheDir, filepath.FromSlash(key)) + Expect(os.MkdirAll(filepath.Dir(cachePath), 0o750)).To(Succeed()) + Expect(os.WriteFile(cachePath, []byte("data"), 0o600)).To(Succeed()) + old := time.Now().Add(-2 * time.Hour) + Expect(os.Chtimes(cachePath, old, old)).To(Succeed()) + Expect(os.Chtimes(filepath.Dir(cachePath), old, old)).To(Succeed()) + + guard, err := NewEphemeralCapacityGuard([]string{root}, 4, 0) + Expect(err).NotTo(HaveOccurred()) + store := &stagingObjectStore{payload: []byte("unused")} + fm, err := storage.NewFileManager(store, cacheDir) + Expect(err).NotTo(HaveOccurred()) + + localPath, err := ensureWorkerFile(context.Background(), fm, guard, key) + Expect(err).NotTo(HaveOccurred()) + Expect(localPath).To(Equal(cachePath)) + Expect(store.getCalls).To(BeZero()) + CleanEphemeralRoots([]string{root}, time.Hour, guard) + Expect(cachePath).To(BeAnExistingFile()) + + Expect(guard.Release(cachePath)).To(Succeed()) + CleanEphemeralRoots([]string{root}, time.Hour, guard) + Expect(cachePath).NotTo(BeAnExistingFile()) + }) + + It("makes repeated cache-hit claims idempotent", func() { + cacheDir := GinkgoT().TempDir() + root := filepath.Join(cacheDir, "ephemeral") + key := "ephemeral/audio/request-id/input.wav" + cachePath := filepath.Join(cacheDir, filepath.FromSlash(key)) + Expect(os.MkdirAll(filepath.Dir(cachePath), 0o750)).To(Succeed()) + Expect(os.WriteFile(cachePath, []byte("data"), 0o600)).To(Succeed()) + guard, err := NewEphemeralCapacityGuard([]string{root}, 4, 0) + Expect(err).NotTo(HaveOccurred()) + fm, err := storage.NewFileManager(&stagingObjectStore{}, cacheDir) + Expect(err).NotTo(HaveOccurred()) + + for range 2 { + localPath, ensureErr := ensureWorkerFile(context.Background(), fm, guard, key) + Expect(ensureErr).NotTo(HaveOccurred()) + Expect(localPath).To(Equal(cachePath)) + } + err = guard.Reserve(filepath.Join(root, "other", "request-id", "input.wav"), 1) + var capacityErr *EphemeralCapacityError + Expect(errors.As(err, &capacityErr)).To(BeTrue()) + Expect(capacityErr.UsageBytes).To(Equal(int64(4))) + }) + + It("capacity-checks growth of a startup-scanned cache file", func() { + cacheDir := GinkgoT().TempDir() + root := filepath.Join(cacheDir, "ephemeral") + key := "ephemeral/audio/request-id/input.wav" + cachePath := filepath.Join(cacheDir, filepath.FromSlash(key)) + Expect(os.MkdirAll(filepath.Dir(cachePath), 0o750)).To(Succeed()) + Expect(os.WriteFile(cachePath, []byte("12"), 0o600)).To(Succeed()) + guard, err := NewEphemeralCapacityGuard([]string{root}, 4, 0) + Expect(err).NotTo(HaveOccurred()) + Expect(os.WriteFile(cachePath, []byte("12345"), 0o600)).To(Succeed()) + fm, err := storage.NewFileManager(&stagingObjectStore{}, cacheDir) + Expect(err).NotTo(HaveOccurred()) + + _, err = ensureWorkerFile(context.Background(), fm, guard, key) + var capacityErr *EphemeralCapacityError + Expect(errors.As(err, &capacityErr)).To(BeTrue()) + Expect(capacityErr.RequestedBytes).To(Equal(int64(3))) + Expect(capacityErr.UsageBytes).To(Equal(int64(2))) + Expect(guard.HasActiveReservation(cachePath)).To(BeFalse()) + }) + It("reserves S3 object size before download and releases it with the exact key", func() { cacheDir := GinkgoT().TempDir() root := filepath.Join(cacheDir, "ephemeral")