fix(nvsnap): do not treat an empty manifest as an existing capture

src/compute-plane-services/nvsnap/internal/rootfsonly/BUILD.bazel

@@ -34,6 +34,7 @@ go_test(
         "capture_method_test.go",
         "classify_test.go",
         "composer_test.go",
+        "empty_capture_test.go",
         "entry_argv_test.go",
         "enumerate_test.go",
         "orchestrator_test.go",

src/compute-plane-services/nvsnap/internal/rootfsonly/empty_capture_test.go

@@ -0,0 +1,91 @@
+/*
+SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+SPDX-License-Identifier: Apache-2.0
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package rootfsonly
+
+import (
+	"testing"
+
+	"github.com/NVIDIA/nvcf/src/compute-plane-services/nvsnap/internal/checkpointstore"
+)
+
+// The backend chain (Local -> ConfigMap -> PerCapturePVC) answers Stat from
+// the first tier that claims a hash, so the tiers can disagree. An L2 PVC left
+// behind by an earlier capture claims a hash whose manifest tier is gone, and
+// the manifest it returns describes no content. Skipping the capture on that
+// claim leaves the pod recorded as captured but unrestorable, and the skip is
+// logged as a successful commit -- so nothing downstream can tell the
+// difference until a restore needs the manifest that was never written.
+func TestUsableCapture(t *testing.T) {
+	cases := []struct {
+		name string
+		m    checkpointstore.Manifest
+		want bool
+	}{
+		{
+			name: "real capture",
+			m:    checkpointstore.Manifest{FileCount: 465, TotalSizeBytes: 141_733_920_768},
+			want: true,
+		},
+		{
+			name: "stale L2 claim: no files, no bytes",
+			m:    checkpointstore.Manifest{},
+			want: false,
+		},
+		{
+			name: "files but no bytes",
+			m:    checkpointstore.Manifest{FileCount: 12},
+			want: false,
+		},
+		{
+			name: "bytes but no files",
+			m:    checkpointstore.Manifest{TotalSizeBytes: 4096},
+			want: false,
+		},
+		{
+			name: "single small file is still a capture",
+			m:    checkpointstore.Manifest{FileCount: 1, TotalSizeBytes: 1},
+			want: true,
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := usableCapture(tc.m); got != tc.want {
+				t.Errorf("usableCapture(files=%d bytes=%d) = %v, want %v",
+					tc.m.FileCount, tc.m.TotalSizeBytes, got, tc.want)
+			}
+		})
+	}
+}
+
+// A manifest describing content must short-circuit the capture; one describing
+// nothing must not. Capture() itself needs a live pod to go further, so this
+// asserts the decision usableCapture drives rather than re-running the walk --
+// the decision is the whole behaviour change.
+func TestEmptyManifestIsNotTreatedAsExisting(t *testing.T) {
+	stale := checkpointstore.Manifest{CapturedOnNodes: []string{"node-a"}}
+	if usableCapture(stale) {
+		t.Fatal("a manifest with no files and no bytes must not count as an existing capture; " +
+			"skipping on it records the pod as captured while leaving it unrestorable")
+	}
+
+	real := checkpointstore.Manifest{FileCount: 1, TotalSizeBytes: 1, CapturedOnNodes: []string{"node-a"}}
+	if !usableCapture(real) {
+		t.Fatal("a manifest describing content must still short-circuit; " +
+			"re-capturing every pass would undo the idempotence the check exists for")
+	}
+}

src/compute-plane-services/nvsnap/internal/rootfsonly/orchestrator.go

@@ -213,9 +213,23 @@ func (c *Capturer) Capture(ctx context.Context, req CaptureRequest) (checkpoints
 	})
 
 	// Idempotent: if the backend already has this hash, we're done.
+	//
+	// Backend is a chain (Local -> ConfigMap -> PerCapturePVC) and Stat
+	// answers from the first tier that claims the hash, so the tiers can
+	// disagree: an L2 PVC left behind by an earlier capture will claim a hash
+	// whose manifest tier is gone. Skipping on that claim returns a manifest
+	// describing nothing, which restore cannot use -- and the caller logs it
+	// as a successful commit, so the pod looks captured while being
+	// unrestorable. Re-capture instead, loudly.
 	if existing, err := c.Backend.Stat(ctx, hash); err == nil {
-		log.Info("capture skipped: hash already exists")
-		return existing, nil
+		if usableCapture(existing) {
+			log.Info("capture skipped: hash already exists")
+			return existing, nil
+		}
+		log.WithFields(logrus.Fields{
+			"files": existing.FileCount,
+			"bytes": existing.TotalSizeBytes,
+		}).Warn("existing capture describes no content; re-capturing (backend tiers are inconsistent for this hash)")
 	} else if !errors.Is(err, checkpointstore.ErrNotFound) {
 		return checkpointstore.Manifest{}, fmt.Errorf("backend stat: %w", err)
 	}
@@ -629,6 +643,17 @@ func (c *Capturer) logger() logrus.FieldLogger {
 	return logrus.NewEntry(logrus.New()).WithField("subsys", "rootfsonly")
 }
 
+// usableCapture reports whether a manifest returned by Backend.Stat describes
+// content a restore could actually replay.
+//
+// FileCount and TotalSizeBytes are the only emptiness signal available here.
+// The consequence is that a workload which genuinely captured nothing is
+// re-captured on every pass rather than skipped: cheap, since there is nothing
+// to copy, and preferable to caching a capture that cannot serve a restore.
+func usableCapture(m checkpointstore.Manifest) bool {
+	return m.FileCount > 0 && m.TotalSizeBytes > 0
+}
+
 // readEntryArgv reads the source process's argv from
 // <procRoot>/<pid>/cmdline, which the kernel stores as NUL-separated,
 // NUL-terminated args. Best-effort: returns nil on any read error or