fix(distributed): restage swept cache hits

Treat files removed between cache probing and ownership claims as misses so HTTP and S3 workers can stage them again.

Assisted-by: Codex:gpt-6

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
 		}

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()

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
 	}

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")