From d2ee5d390a435a658b32905fd6277357d239c245 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Tue, 8 Sep 2026 10:27:48 +0000 Subject: [PATCH] fix(worker): retain staged input ownership Keep committed request inputs protected from age recovery until exact release ends their ownership. Startup-scanned files remain reclaimable and can acquire ownership through reservation. Assisted-by: Codex:gpt-6 --- core/services/worker/ephemeral_capacity.go | 29 +++++++++++++------ .../worker/ephemeral_capacity_test.go | 4 ++- .../services/worker/ephemeral_cleanup_test.go | 23 +++++++++++++++ 3 files changed, 46 insertions(+), 10 deletions(-) diff --git a/core/services/worker/ephemeral_capacity.go b/core/services/worker/ephemeral_capacity.go index 5d9b8158e..0fb5c0a0b 100644 --- a/core/services/worker/ephemeral_capacity.go +++ b/core/services/worker/ephemeral_capacity.go @@ -98,6 +98,7 @@ const ( type ephemeralCapacityEntry struct { state ephemeralCapacityState + owned bool baseline int64 reserved int64 pending int64 @@ -189,6 +190,7 @@ func (g *EphemeralCapacityGuard) Reserve(path string, size int64) error { return err } entry.state = ephemeralCapacityActive + entry.owned = true entry.reserved = size entry.pending = size entry.inflight = 0 @@ -198,7 +200,8 @@ func (g *EphemeralCapacityGuard) Reserve(path string, size int64) error { } // Commit replaces the path's baseline and reservation with the regular file's -// actual size. It waits for every bounded writer for the path to close. +// actual size. It waits for every bounded writer for the path to close and +// preserves request ownership until Release. func (g *EphemeralCapacityGuard) Commit(path string) error { cleanPath, root, err := g.registeredPath(path) if err != nil { @@ -245,6 +248,7 @@ func (g *EphemeralCapacityGuard) Commit(path string) error { } g.usage -= charged - info.Size() entry.state = ephemeralCapacityCommitted + entry.owned = true entry.baseline = info.Size() entry.reserved = 0 entry.pending = 0 @@ -275,8 +279,9 @@ func (g *EphemeralCapacityGuard) Release(path string) error { } } -// Account records bytes found by recovery after startup. Existing active -// reservations are left unchanged so recovery cannot erase in-flight charges. +// Account records unowned bytes found by recovery after startup. Existing +// request-owned entries are left unchanged so recovery cannot erase live +// charges or ownership. func (g *EphemeralCapacityGuard) Account(path string, size int64) error { if size < 0 { return fmt.Errorf("accounted size must not be negative") @@ -289,7 +294,7 @@ func (g *EphemeralCapacityGuard) Account(path string, size int64) error { g.mu.Lock() defer g.mu.Unlock() entry, found := g.entries[cleanPath] - if found && entry.isActive() { + if found && entry.isOwned() { return &EphemeralReservationConflictError{ Path: cleanPath, ActiveBytes: entry.reserved, RequestedBytes: size, } @@ -329,7 +334,7 @@ func (g *EphemeralCapacityGuard) ReleaseTree(path string) error { } // RemoveTreeIfInactive serializes recovery deletion with new reservations so -// cleanup cannot remove a request tree between an activity check and Reserve. +// cleanup cannot remove a request tree between an ownership check and Reserve. func (g *EphemeralCapacityGuard) RemoveTreeIfInactive(path string, remove func() error) (bool, error) { if remove == nil { return false, fmt.Errorf("ephemeral tree remover is nil") @@ -342,7 +347,7 @@ func (g *EphemeralCapacityGuard) RemoveTreeIfInactive(path string, remove func() g.mu.Lock() defer g.mu.Unlock() for entryPath, entry := range g.entries { - if entry.isActive() && ephemeralPathAtOrBelow(entryPath, cleanPath) { + if entry.isOwned() && ephemeralPathAtOrBelow(entryPath, cleanPath) { return false, nil } } @@ -357,8 +362,9 @@ func (g *EphemeralCapacityGuard) RemoveTreeIfInactive(path string, remove func() return true, nil } -// HasActiveReservation reports whether path itself or a descendant has an -// active reservation. Recovery cleanup uses it to avoid active request trees. +// HasActiveReservation reports whether path itself or a descendant is owned +// by a request. Recovery cleanup uses it to avoid live request trees, including +// inputs whose upload has committed while inference is still running. func (g *EphemeralCapacityGuard) HasActiveReservation(path string) bool { cleanPath, _, err := g.registeredPath(path) if err != nil { @@ -368,7 +374,7 @@ func (g *EphemeralCapacityGuard) HasActiveReservation(path string) bool { g.mu.Lock() defer g.mu.Unlock() for entryPath, entry := range g.entries { - if entry.isActive() && ephemeralPathAtOrBelow(entryPath, cleanPath) { + if entry.isOwned() && ephemeralPathAtOrBelow(entryPath, cleanPath) { return true } } @@ -409,6 +415,7 @@ func (g *EphemeralCapacityGuard) NewWriter(path string, destination io.Writer) ( entry.pending = 0 entry.inflight = 0 } + entry.owned = true entry.state = ephemeralCapacityWriting entry.openWriters++ g.entries[cleanPath] = entry @@ -480,6 +487,10 @@ func (e ephemeralCapacityEntry) isActive() bool { return e.state == ephemeralCapacityActive || e.state == ephemeralCapacityWriting } +func (e ephemeralCapacityEntry) isOwned() bool { + return e.owned +} + func (g *EphemeralCapacityGuard) accountExistingFiles(root string) error { err := filepath.WalkDir(root, func(path string, dirEntry os.DirEntry, walkErr error) error { if walkErr != nil { diff --git a/core/services/worker/ephemeral_capacity_test.go b/core/services/worker/ephemeral_capacity_test.go index ce03231bd..6fe883f50 100644 --- a/core/services/worker/ephemeral_capacity_test.go +++ b/core/services/worker/ephemeral_capacity_test.go @@ -133,7 +133,7 @@ var _ = Describe("EphemeralCapacityGuard", func() { Expect(guard.Reserve(path, 4)).To(Succeed()) Expect(os.WriteFile(path, make([]byte, 4), 0o600)).To(Succeed()) Expect(guard.Commit(path)).To(Succeed()) - Expect(guard.HasActiveReservation(path)).To(BeFalse()) + Expect(guard.HasActiveReservation(path)).To(BeTrue()) Expect(guard.Reserve(path, 6)).To(Succeed()) Expect(guard.HasActiveReservation(path)).To(BeTrue()) @@ -149,8 +149,10 @@ var _ = Describe("EphemeralCapacityGuard", func() { Expect(os.WriteFile(path, make([]byte, 4), 0o600)).To(Succeed()) guard, err := NewEphemeralCapacityGuard([]string{root}, 10, 0) Expect(err).NotTo(HaveOccurred()) + Expect(guard.HasActiveReservation(path)).To(BeFalse()) Expect(guard.Reserve(path, 6)).To(Succeed()) + Expect(guard.HasActiveReservation(path)).To(BeTrue()) err = guard.Reserve(filepath.Join(root, "overflow.bin"), 1) var capacityErr *EphemeralCapacityError Expect(errors.As(err, &capacityErr)).To(BeTrue()) diff --git a/core/services/worker/ephemeral_cleanup_test.go b/core/services/worker/ephemeral_cleanup_test.go index 0c54fb048..c8cb9e150 100644 --- a/core/services/worker/ephemeral_cleanup_test.go +++ b/core/services/worker/ephemeral_cleanup_test.go @@ -88,4 +88,27 @@ var _ = Describe("Worker ephemeral staging cleanup", func() { Expect(freshChildRequest).To(BeADirectory()) Expect(guard.Reserve(filepath.Join(httpRoot, "audio", "replacement", "input.bin"), 4)).To(Succeed()) }) + + It("keeps committed request inputs until exact release ends ownership", func() { + root := filepath.Join(stagingDir, "ephemeral") + requestDir := filepath.Join(root, "audio", "owned") + path := filepath.Join(requestDir, "input.bin") + Expect(os.MkdirAll(requestDir, 0o750)).To(Succeed()) + guard, err := NewEphemeralCapacityGuard([]string{root}, 8, 0) + Expect(err).NotTo(HaveOccurred()) + + Expect(guard.Reserve(path, 4)).To(Succeed()) + Expect(os.WriteFile(path, []byte("data"), 0o600)).To(Succeed()) + Expect(guard.Commit(path)).To(Succeed()) + old := time.Now().Add(-2 * time.Hour) + Expect(os.Chtimes(path, old, old)).To(Succeed()) + Expect(os.Chtimes(requestDir, old, old)).To(Succeed()) + + CleanEphemeralRoots([]string{root}, time.Hour, guard) + Expect(requestDir).To(BeADirectory()) + + Expect(guard.Release(path)).To(Succeed()) + CleanEphemeralRoots([]string{root}, time.Hour, guard) + Expect(requestDir).NotTo(BeADirectory()) + }) })