diff --git a/core/services/nodes/file_transfer_server.go b/core/services/nodes/file_transfer_server.go index 9e26c9b17..5ccc43cde 100644 --- a/core/services/nodes/file_transfer_server.go +++ b/core/services/nodes/file_transfer_server.go @@ -225,6 +225,10 @@ func handleClaimWithCapacity(w http.ResponseWriter, _ *http.Request, stagingDir, } if capacity != nil { if err := capacity.Claim(filePath); err != nil { + if errors.Is(err, os.ErrNotExist) { + http.Error(w, "not found", http.StatusNotFound) + return + } http.Error(w, err.Error(), http.StatusInsufficientStorage) return } diff --git a/core/services/nodes/file_transfer_server_test.go b/core/services/nodes/file_transfer_server_test.go index 164019b7a..3c156ff82 100644 --- a/core/services/nodes/file_transfer_server_test.go +++ b/core/services/nodes/file_transfer_server_test.go @@ -567,6 +567,21 @@ var _ = Describe("FileTransferServer", func() { // --- EnsureRemote skip tests --- Describe("EnsureRemote skip-if-exists", func() { + It("reports a claim-time disappearance as a cache miss", func() { + stagingDir := GinkgoT().TempDir() + key := "ephemeral/audio/request/input.wav" + remotePath := filepath.Join(stagingDir, filepath.FromSlash(key)) + Expect(os.MkdirAll(filepath.Dir(remotePath), 0o750)).To(Succeed()) + Expect(os.WriteFile(remotePath, []byte("stale"), 0o600)).To(Succeed()) + guard := &recordingEphemeralCapacity{claimErr: fmt.Errorf("claim raced recovery: %w", os.ErrNotExist)} + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/v1/files/"+key, nil) + + handleClaimWithCapacity(recorder, request, stagingDir, key, guard) + + Expect(recorder.Code).To(Equal(http.StatusNotFound)) + }) + It("claims a matching ephemeral file before returning the worker path", func() { stagingDir := GinkgoT().TempDir() modelsDir := GinkgoT().TempDir() diff --git a/core/services/worker/file_staging.go b/core/services/worker/file_staging.go index 7afcae19e..f7d3aadbf 100644 --- a/core/services/worker/file_staging.go +++ b/core/services/worker/file_staging.go @@ -3,6 +3,7 @@ package worker import ( "context" "encoding/json" + "errors" "fmt" "os" "path" @@ -272,7 +273,21 @@ func releaseEphemeralCachePathWithCapacity(cacheDir, key, filePath string, capac return nil } +type ephemeralStagingCapacity interface { + Reserve(path string, size int64) error + Commit(path string) error + Claim(path string) error + Release(path string) error +} + func ensureWorkerFile(ctx context.Context, fm *storage.FileManager, capacity *EphemeralCapacityGuard, key string) (string, error) { + if capacity == nil { + return fm.Download(ctx, key) + } + return ensureWorkerFileWithCapacity(ctx, fm, capacity, key) +} + +func ensureWorkerFileWithCapacity(ctx context.Context, fm *storage.FileManager, capacity ephemeralStagingCapacity, key string) (string, error) { if capacity == nil || !strings.HasPrefix(key, "ephemeral/") { return fm.Download(ctx, key) } @@ -288,9 +303,12 @@ func ensureWorkerFile(ctx context.Context, fm *storage.FileManager, capacity *Ep return "", fmt.Errorf("ephemeral cache path %q is not a regular file", cachePath) } if err := capacity.Claim(cachePath); err != nil { - return "", err + if !errors.Is(err, os.ErrNotExist) { + return "", err + } + } else { + return cachePath, nil } - return cachePath, nil } else if !os.IsNotExist(statErr) { return "", statErr } diff --git a/core/services/worker/file_staging_release_test.go b/core/services/worker/file_staging_release_test.go index 36755368a..e2a9c6af7 100644 --- a/core/services/worker/file_staging_release_test.go +++ b/core/services/worker/file_staging_release_test.go @@ -26,6 +26,18 @@ type stagingObjectStore struct { getErr error } +type disappearingStagingCapacity struct{} + +func (*disappearingStagingCapacity) Reserve(string, int64) error { return nil } +func (*disappearingStagingCapacity) Commit(string) error { return nil } +func (*disappearingStagingCapacity) Release(string) error { return nil } +func (*disappearingStagingCapacity) Claim(path string) error { + if err := os.Remove(path); err != nil { + return err + } + return fmt.Errorf("claim raced recovery: %w", os.ErrNotExist) +} + func (*stagingObjectStore) Put(context.Context, string, io.Reader) error { return nil } func (s *stagingObjectStore) Get(context.Context, string) (io.ReadCloser, error) { s.getCalls++ @@ -145,6 +157,24 @@ var _ = Describe("Worker exact-key staging release", func() { Expect(cachePath).NotTo(BeAnExistingFile()) }) + It("downloads again when a cache file disappears while being claimed", func() { + cacheDir := GinkgoT().TempDir() + 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("stale"), 0o600)).To(Succeed()) + store := &stagingObjectStore{payload: []byte("fresh")} + fm, err := storage.NewFileManager(store, cacheDir) + Expect(err).NotTo(HaveOccurred()) + + localPath, err := ensureWorkerFileWithCapacity(context.Background(), fm, &disappearingStagingCapacity{}, key) + + Expect(err).NotTo(HaveOccurred()) + Expect(localPath).To(Equal(cachePath)) + Expect(os.ReadFile(localPath)).To(Equal([]byte("fresh"))) + Expect(store.getCalls).To(Equal(1)) + }) + It("makes repeated cache-hit claims idempotent", func() { cacheDir := GinkgoT().TempDir() root := filepath.Join(cacheDir, "ephemeral")