From 881141ab43a5bbf7f860d8a271f6fd0cb7da9355 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Tue, 8 Sep 2026 13:22:30 +0000 Subject: [PATCH] fix(distributed): enforce staging admission Propagate multimodal staging failures before inference and claim matching ephemeral HTTP cache entries. Fall back to PUT when an older worker does not support claims. Assisted-by: Codex:gpt-6 --- core/services/nodes/file_stager_http.go | 64 ++++++-- core/services/nodes/file_staging_client.go | 40 +++-- .../nodes/file_staging_lifecycle_test.go | 45 +++++- core/services/nodes/file_transfer_server.go | 37 +++++ .../nodes/file_transfer_server_test.go | 149 ++++++++++++++++++ .../worker/file_staging_release_test.go | 46 ++++++ 6 files changed, 356 insertions(+), 25 deletions(-) diff --git a/core/services/nodes/file_stager_http.go b/core/services/nodes/file_stager_http.go index 681eaa3d4..4d7c46a30 100644 --- a/core/services/nodes/file_stager_http.go +++ b/core/services/nodes/file_stager_http.go @@ -120,7 +120,9 @@ func (h *HTTPFileStager) EnsureRemote(ctx context.Context, nodeID, localPath, ke } // Probe: check if the remote already has the file with matching content hash. - if remotePath, ok := h.probeExisting(ctx, addr, localPath, key); ok { + if remotePath, ok, probeErr := h.probeExisting(ctx, addr, localPath, key); probeErr != nil { + return "", fmt.Errorf("claiming existing file on node %s: %w", nodeID, probeErr) + } else if ok { xlog.Info("Upload skipped (file already exists with matching hash)", "node", nodeID, "key", key, "remotePath", remotePath) return remotePath, nil } @@ -469,14 +471,15 @@ func isTransientError(err error) bool { // probeExisting sends a HEAD request to check if the remote already has the // file with a matching SHA-256 hash. Returns the remote path and true if the -// upload can be skipped. Any errors (including 405 from older servers) silently -// fall through so the caller proceeds with a normal PUT. -func (h *HTTPFileStager) probeExisting(ctx context.Context, addr, localPath, key string) (string, bool) { +// upload can be skipped. HEAD and hash errors fall through to a normal PUT. +// Matching ephemeral files are claimed first; a 404 or 405 claim response +// identifies an older worker and also falls through to PUT. +func (h *HTTPFileStager) probeExisting(ctx context.Context, addr, localPath, key string) (string, bool, error) { url := fmt.Sprintf("http://%s/v1/files/%s", addr, key) req, err := http.NewRequestWithContext(ctx, http.MethodHead, url, nil) if err != nil { - return "", false + return "", false, nil } if h.token != "" { req.Header.Set("Authorization", "Bearer "+h.token) @@ -484,18 +487,18 @@ func (h *HTTPFileStager) probeExisting(ctx context.Context, addr, localPath, key resp, err := h.client.Do(req) if err != nil { - return "", false + return "", false, nil } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - return "", false + return "", false, nil } remotePath := resp.Header.Get(HeaderLocalPath) remoteHash := resp.Header.Get(HeaderContentSHA256) if remotePath == "" || remoteHash == "" { - return "", false + return "", false, nil } // A 200 with a content hash is proof the worker is alive and serving right @@ -505,14 +508,53 @@ func (h *HTTPFileStager) probeExisting(ctx context.Context, addr, localPath, key localHash, err := hashLocalCached(ctx, localPath) if err != nil { - return "", false + return "", false, nil } if localHash != remoteHash { - return "", false + return "", false, nil } - return remotePath, true + if strings.HasPrefix(key, "ephemeral/") { + claimed, err := h.claimExisting(ctx, addr, key) + if err != nil { + return "", false, err + } + if !claimed { + return "", false, nil + } + } + + return remotePath, true, nil +} + +func (h *HTTPFileStager) claimExisting(ctx context.Context, addr, key string) (bool, error) { + claimURL := (&url.URL{ + Scheme: "http", + Host: addr, + Path: "/v1/files/" + key, + RawQuery: "claim=1", + }).String() + req, err := http.NewRequestWithContext(ctx, http.MethodPost, claimURL, nil) + if err != nil { + return false, fmt.Errorf("creating claim request for %q: %w", key, err) + } + if h.token != "" { + req.Header.Set("Authorization", "Bearer "+h.token) + } + resp, err := h.client.Do(req) + if err != nil { + return false, fmt.Errorf("claiming %q: %w", key, err) + } + defer resp.Body.Close() + if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusMethodNotAllowed { + return false, nil + } + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return false, fmt.Errorf("claiming %q: status %d: %s", key, resp.StatusCode, strings.TrimSpace(string(body))) + } + return true, nil } // hashChunkSize is how much of a file is hashed between activity ticks and diff --git a/core/services/nodes/file_staging_client.go b/core/services/nodes/file_staging_client.go index 57c5bad9b..31def99bc 100644 --- a/core/services/nodes/file_staging_client.go +++ b/core/services/nodes/file_staging_client.go @@ -149,7 +149,11 @@ func (f *FileStagingClient) Predict(ctx context.Context, in *pb.PredictOptions, lifecycle := f.newStagedInputLifecycle() defer lifecycle.release() in = proto.Clone(in).(*pb.PredictOptions) - in = f.stageMultimodalInputs(ctx, lifecycle, in) + var err error + in, err = f.stageMultimodalInputs(ctx, lifecycle, in) + if err != nil { + return nil, err + } return f.Backend.Predict(ctx, in, opts...) } @@ -157,7 +161,11 @@ func (f *FileStagingClient) PredictStream(ctx context.Context, in *pb.PredictOpt lifecycle := f.newStagedInputLifecycle() defer lifecycle.release() in = proto.Clone(in).(*pb.PredictOptions) - in = f.stageMultimodalInputs(ctx, lifecycle, in) + var err error + in, err = f.stageMultimodalInputs(ctx, lifecycle, in) + if err != nil { + return err + } return f.Backend.PredictStream(ctx, in, fn, opts...) } @@ -528,11 +536,21 @@ func (f *FileStagingClient) stageMultimodalInputs( ctx context.Context, lifecycle *stagedInputLifecycle, in *pb.PredictOptions, -) *pb.PredictOptions { - in.Images = f.stagePathSlice(ctx, lifecycle, in.Images, "inputs") - in.Videos = f.stagePathSlice(ctx, lifecycle, in.Videos, "inputs") - in.Audios = f.stagePathSlice(ctx, lifecycle, in.Audios, "inputs") - return in +) (*pb.PredictOptions, error) { + var err error + in.Images, err = f.stagePathSlice(ctx, lifecycle, in.Images, "inputs") + if err != nil { + return nil, fmt.Errorf("staging predict images: %w", err) + } + in.Videos, err = f.stagePathSlice(ctx, lifecycle, in.Videos, "inputs") + if err != nil { + return nil, fmt.Errorf("staging predict videos: %w", err) + } + in.Audios, err = f.stagePathSlice(ctx, lifecycle, in.Audios, "inputs") + if err != nil { + return nil, fmt.Errorf("staging predict audios: %w", err) + } + return in, nil } func (f *FileStagingClient) stagePathSlice( @@ -540,22 +558,20 @@ func (f *FileStagingClient) stagePathSlice( lifecycle *stagedInputLifecycle, paths []string, category string, -) []string { +) ([]string, error) { result := make([]string, len(paths)) for i, p := range paths { if isFilePath(p) { backendPath, err := f.stageInputFile(ctx, lifecycle, p, category) if err != nil { - xlog.Warn("Failed to stage multimodal file, passing through", "path", p, "error", err) - result[i] = p - continue + return nil, fmt.Errorf("staging %q: %w", p, err) } result[i] = backendPath } else { result[i] = p } } - return result + return result, nil } // isFilePath checks if a string looks like a local file path (not base64 or URL). diff --git a/core/services/nodes/file_staging_lifecycle_test.go b/core/services/nodes/file_staging_lifecycle_test.go index 25892e998..3f33b5064 100644 --- a/core/services/nodes/file_staging_lifecycle_test.go +++ b/core/services/nodes/file_staging_lifecycle_test.go @@ -19,6 +19,7 @@ const fullUUIDPattern = `[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}- type lifecycleStager struct { fakeFileStager ensureErr error + ensureErrAt int releaseErr error releasedKeys []string releaseCtxErr []error @@ -28,7 +29,7 @@ type lifecycleStager struct { func (s *lifecycleStager) EnsureRemote(ctx context.Context, nodeID, localPath, key string) (string, error) { s.fakeFileStager.EnsureRemote(ctx, nodeID, localPath, key) - if s.ensureErr != nil { + if s.ensureErr != nil && (s.ensureErrAt == 0 || len(s.ensureCalls) == s.ensureErrAt) { return "", s.ensureErr } return "/remote/" + key, nil @@ -47,11 +48,16 @@ type lifecycleBackend struct { grpc.Backend predictResult *pb.Reply predictErr error + predictCalls int + predictInput *pb.PredictOptions + streamCalls int streamBlock <-chan struct{} streamStarted chan<- struct{} } -func (b *lifecycleBackend) Predict(_ context.Context, _ *pb.PredictOptions, _ ...ggrpc.CallOption) (*pb.Reply, error) { +func (b *lifecycleBackend) Predict(_ context.Context, in *pb.PredictOptions, _ ...ggrpc.CallOption) (*pb.Reply, error) { + b.predictCalls++ + b.predictInput = proto.Clone(in).(*pb.PredictOptions) if b.predictResult == nil { b.predictResult = &pb.Reply{} } @@ -59,6 +65,7 @@ func (b *lifecycleBackend) Predict(_ context.Context, _ *pb.PredictOptions, _ .. } func (b *lifecycleBackend) PredictStream(_ context.Context, _ *pb.PredictOptions, _ func(*pb.Reply), _ ...ggrpc.CallOption) error { + b.streamCalls++ if b.streamStarted != nil { b.streamStarted <- struct{}{} } @@ -234,6 +241,40 @@ var _ = Describe("FileStagingClient request lifecycle", func() { Expect(stager.releasedKeys).To(Equal(keysFromEnsureCalls(stager.ensureCalls))) }) + It("does not invoke predict when multimodal staging fails", func(ctx SpecContext) { + uploadErr := errors.New("ephemeral capacity exceeded") + stager := &lifecycleStager{ensureErr: uploadErr, ensureErrAt: 2} + backend := &lifecycleBackend{} + client := NewFileStagingClient(backend, stager, "worker-1") + request := &pb.PredictOptions{Images: []string{"/tmp/first.png", "/tmp/second.png"}} + original := proto.Clone(request) + + result, err := client.Predict(ctx, request) + + Expect(result).To(BeNil()) + Expect(err).To(MatchError(ContainSubstring("ephemeral capacity exceeded"))) + Expect(backend.predictCalls).To(BeZero()) + Expect(proto.Equal(request, original)).To(BeTrue()) + Expect(stager.ensureCalls).To(HaveLen(2)) + Expect(stager.releasedKeys).To(Equal(keysFromEnsureCalls(stager.ensureCalls))) + }) + + It("does not invoke streaming predict when multimodal staging fails", func(ctx SpecContext) { + uploadErr := errors.New("ephemeral capacity exceeded") + stager := &lifecycleStager{ensureErr: uploadErr} + backend := &lifecycleBackend{} + client := NewFileStagingClient(backend, stager, "worker-1") + request := &pb.PredictOptions{Audios: []string{"/tmp/audio.wav"}} + original := proto.Clone(request) + + err := client.PredictStream(ctx, request, func(*pb.Reply) {}) + + Expect(err).To(MatchError(ContainSubstring("ephemeral capacity exceeded"))) + Expect(backend.streamCalls).To(BeZero()) + Expect(proto.Equal(request, original)).To(BeTrue()) + Expect(stager.releasedKeys).To(Equal(keysFromEnsureCalls(stager.ensureCalls))) + }) + It("uses an active bounded cleanup context after caller cancellation", func() { ctx, cancel := context.WithCancel(context.Background()) cancel() diff --git a/core/services/nodes/file_transfer_server.go b/core/services/nodes/file_transfer_server.go index 93c175db9..42fcc3a02 100644 --- a/core/services/nodes/file_transfer_server.go +++ b/core/services/nodes/file_transfer_server.go @@ -136,6 +136,8 @@ func startFileTransferServer(lis net.Listener, stagingDir, modelsDir, dataDir, t case http.MethodPost: if key == "temp" { handleAllocTemp(w, r, stagingDir) + } else if r.URL.Query().Get("claim") == "1" { + handleClaimWithCapacity(w, r, stagingDir, key, capacity) } else { http.Error(w, "not found", http.StatusNotFound) } @@ -196,6 +198,40 @@ func startFileTransferServer(lis net.Listener, stagingDir, modelsDir, dataDir, t return server, nil } +// handleClaimWithCapacity marks an existing ephemeral file as owned by the +// request that just verified its content. +func handleClaimWithCapacity(w http.ResponseWriter, _ *http.Request, stagingDir, key string, capacity EphemeralCapacity) { + if err := validateEphemeralReleaseKey(key); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + filePath := filepath.Join(stagingDir, filepath.FromSlash(key)) + if err := validatePathInDir(filePath, stagingDir); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + info, err := os.Lstat(filePath) + if err != nil { + if os.IsNotExist(err) { + http.Error(w, "not found", http.StatusNotFound) + } else { + http.Error(w, err.Error(), http.StatusInternalServerError) + } + return + } + if !info.Mode().IsRegular() { + http.Error(w, "ephemeral path is not a regular file", http.StatusBadRequest) + return + } + if capacity != nil { + if err := capacity.Claim(filePath); err != nil { + http.Error(w, err.Error(), http.StatusInsufficientStorage) + return + } + } + w.WriteHeader(http.StatusNoContent) +} + func handleRelease(w http.ResponseWriter, _ *http.Request, stagingDir, key string) { handleReleaseWithCapacity(w, nil, stagingDir, key, nil) } @@ -300,6 +336,7 @@ type contentRange struct { type EphemeralCapacity interface { Reserve(path string, size int64) error Commit(path string) error + Claim(path string) error Release(path string) error CapacityWriter(path string, destination io.Writer) (io.WriteCloser, error) } diff --git a/core/services/nodes/file_transfer_server_test.go b/core/services/nodes/file_transfer_server_test.go index 2bbe80387..03912a965 100644 --- a/core/services/nodes/file_transfer_server_test.go +++ b/core/services/nodes/file_transfer_server_test.go @@ -17,6 +17,7 @@ import ( "sync" "time" + pb "github.com/mudler/LocalAI/pkg/grpc/proto" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" ) @@ -25,6 +26,8 @@ type recordingEphemeralCapacity struct { reserved int64 reserveErr error writerErr error + claimCalls []string + claimErr error } type nopWriteCloser struct{ io.Writer } @@ -43,6 +46,10 @@ func (g *recordingEphemeralCapacity) Reserve(_ string, size int64) error { func (*recordingEphemeralCapacity) Commit(string) error { return nil } func (*recordingEphemeralCapacity) Release(string) error { return nil } +func (g *recordingEphemeralCapacity) Claim(path string) error { + g.claimCalls = append(g.claimCalls, path) + return g.claimErr +} func (g *recordingEphemeralCapacity) CapacityWriter(_ string, destination io.Writer) (io.WriteCloser, error) { if g.writerErr != nil { return failingWriteCloser{err: g.writerErr}, nil @@ -494,6 +501,148 @@ var _ = Describe("FileTransferServer", func() { // --- EnsureRemote skip tests --- Describe("EnsureRemote skip-if-exists", func() { + It("claims a matching ephemeral file before returning the worker path", func() { + stagingDir := GinkgoT().TempDir() + modelsDir := GinkgoT().TempDir() + dataDir := GinkgoT().TempDir() + guard := &recordingEphemeralCapacity{} + key := "ephemeral/audio/request/input.wav" + remotePath := filepath.Join(stagingDir, filepath.FromSlash(key)) + content := []byte("already on worker") + Expect(os.MkdirAll(filepath.Dir(remotePath), 0o750)).To(Succeed()) + Expect(os.WriteFile(remotePath, content, 0o600)).To(Succeed()) + Expect(os.WriteFile(remotePath+hashSidecarSuffix, []byte(sha256Hex(content)), 0o600)).To(Succeed()) + localPath := filepath.Join(GinkgoT().TempDir(), "input.wav") + Expect(os.WriteFile(localPath, content, 0o600)).To(Succeed()) + + mux := http.NewServeMux() + mux.HandleFunc("/v1/files/", func(w http.ResponseWriter, r *http.Request) { + requestKey := strings.TrimPrefix(r.URL.Path, "/v1/files/") + switch r.Method { + case http.MethodHead: + handleHead(w, r, stagingDir, modelsDir, dataDir, requestKey) + case http.MethodPost: + handleClaimWithCapacity(w, r, stagingDir, requestKey, guard) + default: + http.Error(w, "unexpected upload", http.StatusInternalServerError) + } + }) + ts := httptest.NewServer(mux) + DeferCleanup(ts.Close) + stager := NewHTTPFileStager(func(string) (string, error) { + return strings.TrimPrefix(ts.URL, "http://"), nil + }, "") + + for range 2 { + path, err := stager.EnsureRemote(context.Background(), "node-1", localPath, key) + Expect(err).NotTo(HaveOccurred()) + Expect(path).To(Equal(remotePath)) + } + Expect(guard.claimCalls).To(Equal([]string{remotePath, remotePath})) + }) + + It("propagates an ephemeral cache-hit claim failure", func() { + content := []byte("already on worker") + localPath := filepath.Join(GinkgoT().TempDir(), "input.wav") + Expect(os.WriteFile(localPath, content, 0o600)).To(Succeed()) + mux := http.NewServeMux() + mux.HandleFunc("/v1/files/", func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodHead { + w.Header().Set(HeaderLocalPath, "/remote/ephemeral/audio/request/input.wav") + w.Header().Set(HeaderContentSHA256, sha256Hex(content)) + w.WriteHeader(http.StatusOK) + return + } + http.Error(w, "ephemeral capacity exceeded", http.StatusInsufficientStorage) + }) + ts := httptest.NewServer(mux) + DeferCleanup(ts.Close) + stager := NewHTTPFileStager(func(string) (string, error) { + return strings.TrimPrefix(ts.URL, "http://"), nil + }, "") + + path, err := stager.EnsureRemote(context.Background(), "node-1", localPath, "ephemeral/audio/request/input.wav") + + Expect(path).To(BeEmpty()) + Expect(err).To(MatchError(ContainSubstring("ephemeral capacity exceeded"))) + }) + + DescribeTable("uploads a matching ephemeral file when an old worker cannot claim it", + func(claimStatus int) { + stagingDir := GinkgoT().TempDir() + content := []byte("compatible upload") + localPath := filepath.Join(GinkgoT().TempDir(), "input.wav") + Expect(os.WriteFile(localPath, content, 0o600)).To(Succeed()) + cacheHitPath := filepath.Join(stagingDir, "ephemeral", "audio", "stale", "input.wav") + putCalls := 0 + putRemotePath := "" + mux := http.NewServeMux() + mux.HandleFunc("/v1/files/", func(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodHead: + w.Header().Set(HeaderLocalPath, cacheHitPath) + w.Header().Set(HeaderContentSHA256, sha256Hex(content)) + w.WriteHeader(http.StatusOK) + case http.MethodPost: + http.Error(w, "claim unsupported", claimStatus) + case http.MethodPut: + putCalls++ + key := strings.TrimPrefix(r.URL.Path, "/v1/files/") + putRemotePath = filepath.Join(stagingDir, filepath.FromSlash(key)) + handleUpload(w, r, stagingDir, "", "", key, 0) + case http.MethodDelete: + w.WriteHeader(http.StatusNoContent) + } + }) + ts := httptest.NewServer(mux) + DeferCleanup(ts.Close) + stager := NewHTTPFileStager(func(string) (string, error) { + return strings.TrimPrefix(ts.URL, "http://"), nil + }, "") + + backend := &lifecycleBackend{} + client := NewFileStagingClient(backend, stager, "node-1") + request := &pb.PredictOptions{Audios: []string{localPath}} + _, err := client.Predict(context.Background(), request) + + Expect(err).NotTo(HaveOccurred()) + Expect(putCalls).To(Equal(1)) + Expect(backend.predictInput).NotTo(BeNil()) + Expect(backend.predictInput.Audios).To(Equal([]string{putRemotePath})) + }, + Entry("404", http.StatusNotFound), + Entry("405", http.StatusMethodNotAllowed), + ) + + It("keeps matching model probes read-only", func() { + content := []byte("model") + localPath := filepath.Join(GinkgoT().TempDir(), "model.bin") + Expect(os.WriteFile(localPath, content, 0o600)).To(Succeed()) + unexpectedWrites := 0 + mux := http.NewServeMux() + mux.HandleFunc("/v1/files/", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodHead { + unexpectedWrites++ + http.Error(w, "unexpected write", http.StatusInternalServerError) + return + } + w.Header().Set(HeaderLocalPath, "/models/tracking/model.bin") + w.Header().Set(HeaderContentSHA256, sha256Hex(content)) + w.WriteHeader(http.StatusOK) + }) + ts := httptest.NewServer(mux) + DeferCleanup(ts.Close) + stager := NewHTTPFileStager(func(string) (string, error) { + return strings.TrimPrefix(ts.URL, "http://"), nil + }, "") + + path, err := stager.EnsureRemote(context.Background(), "node-1", localPath, "models/tracking/model.bin") + + Expect(err).NotTo(HaveOccurred()) + Expect(path).To(Equal("/models/tracking/model.bin")) + Expect(unexpectedWrites).To(BeZero()) + }) + It("skips upload when file exists with matching hash", func() { ts, stagingDir, _, _ := setupTestServer("tok", 0) diff --git a/core/services/worker/file_staging_release_test.go b/core/services/worker/file_staging_release_test.go index d94b0ade7..36755368a 100644 --- a/core/services/worker/file_staging_release_test.go +++ b/core/services/worker/file_staging_release_test.go @@ -2,15 +2,19 @@ package worker import ( "context" + "crypto/sha256" "encoding/json" "errors" + "fmt" "io" + "net" "os" "path/filepath" "strings" "time" "github.com/mudler/LocalAI/core/services/messaging" + "github.com/mudler/LocalAI/core/services/nodes" "github.com/mudler/LocalAI/core/services/storage" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -70,6 +74,48 @@ func (m *releaseMessagingClient) IsConnected() bool { return true } func (m *releaseMessagingClient) Close() {} var _ = Describe("Worker exact-key staging release", func() { + It("protects a startup-accounted HTTP cache hit through authenticated repeated probes", func() { + stagingDir := GinkgoT().TempDir() + root := filepath.Join(stagingDir, "ephemeral") + key := "ephemeral/audio/request-id/input.wav" + remotePath := filepath.Join(stagingDir, filepath.FromSlash(key)) + content := []byte("data") + Expect(os.MkdirAll(filepath.Dir(remotePath), 0o750)).To(Succeed()) + Expect(os.WriteFile(remotePath, content, 0o600)).To(Succeed()) + hash := sha256.Sum256(content) + Expect(os.WriteFile(remotePath+".sha256", []byte(fmt.Sprintf("%x", hash)), 0o600)).To(Succeed()) + old := time.Now().Add(-2 * time.Hour) + for _, path := range []string{remotePath, remotePath + ".sha256", filepath.Dir(remotePath)} { + Expect(os.Chtimes(path, old, old)).To(Succeed()) + } + guard, err := NewEphemeralCapacityGuard([]string{root}, int64(len(content)+sha256.Size*2), 0) + Expect(err).NotTo(HaveOccurred()) + + listener, err := net.Listen("tcp", "127.0.0.1:0") + Expect(err).NotTo(HaveOccurred()) + addr := listener.Addr().String() + Expect(listener.Close()).To(Succeed()) + server, err := nodes.StartFileTransferServerWithCapacity(addr, stagingDir, GinkgoT().TempDir(), GinkgoT().TempDir(), "secret", 0, nil, guard) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(nodes.ShutdownFileTransferServer, server) + + localPath := filepath.Join(GinkgoT().TempDir(), "input.wav") + Expect(os.WriteFile(localPath, content, 0o600)).To(Succeed()) + stager := nodes.NewHTTPFileStager(func(string) (string, error) { return addr, nil }, "secret") + for range 2 { + path, ensureErr := stager.EnsureRemote(context.Background(), "worker", localPath, key) + Expect(ensureErr).NotTo(HaveOccurred()) + Expect(path).To(Equal(remotePath)) + } + + CleanEphemeralRoots([]string{root}, time.Hour, guard) + Expect(remotePath).To(BeAnExistingFile()) + + Expect(stager.ReleaseRemote(context.Background(), "worker", key)).To(Succeed()) + CleanEphemeralRoots([]string{root}, time.Hour, guard) + Expect(remotePath).NotTo(BeAnExistingFile()) + }) + It("claims a startup-scanned cache hit against stale recovery until release", func() { cacheDir := GinkgoT().TempDir() root := filepath.Join(cacheDir, "ephemeral")