fix(nvsnap): capture watcher and cachedir restore shim, so a restored pod actually serves

src/compute-plane-services/nvsnap/docs/BENCHMARK.md

@@ -69,6 +69,31 @@ bring-up floor (scheduling, framework import, CUDA init, tensor-parallel
 worker spawn) that both cold and warm starts pay; cachedir compresses the
 disk work, not live process initialization.
 
+### Page-cache prewarm A/B, September 2026
+
+The cachedir restore adds an `nvsnap-prewarm` init container that reads the
+whole rox tree (`find | xargs -P 6 cat`) before the engine starts, so the
+engine's weight load hits page cache instead of storage. Opt out per pod with
+`NVSNAP_PREWARM=0`. Measured on Llama-3.1-70B-Instruct, vLLM v0.20.0 TP=4,
+131.6 GB rox on a network block volume (NVMesh, ~1 GB/s per volume), same node,
+page cache dropped before every run:
+
+| Restore | Ready | Prewarm init |
+|---|---:|---:|
+| prewarm | 245 s, 249 s | 138 s, 137 s |
+| no prewarm | 263 s, 251 s | -- |
+
+Cold start on the same node: 632 s. The prewarm is neutral here: a single
+reader already saturates the volume, so the prewarm only moves the same
+storage-bound read ahead of the engine instead of overlapping it with process
+bring-up. The prewarm pays off on volumes whose single-stream reads are
+latency-bound but whose aggregate throughput is high (Hyperdisk ML), where the
+engine's per-fault mmap reads leave the volume idle and the parallel sweep does
+not. Treat the prewarm value as a property of the storage class, not the model:
+the default and the reader count live in the storage profile (`prewarm`,
+`prewarmParallelism` in the `nvsnap-storage-profiles` ConfigMap), see
+`docs/design/STORAGE-AGNOSTIC-L2-PROMOTION.md`.
+
 ## Environment
 
 - **Cluster**: example-gpu-cluster (GKE)

src/compute-plane-services/nvsnap/docs/design/STORAGE-AGNOSTIC-L2-PROMOTION.md

@@ -215,6 +215,7 @@ data:
       strategy: shared-volume
       volumeHandleTransform: nvmesh
       mountOptions: [ro, norecovery, nouuid]  # xfs RO multi-mount needs nouuid
+      prewarm: false                       # one reader saturates the volume
     efs.csi.aws.com:                       # EFS — shared RWX filesystem
       strategy: shared-volume
       volumeHandleTransform: none
@@ -230,6 +231,20 @@ disables L2.
 The built-in table carries these same defaults so a stock install needs no
 ConfigMap; the ConfigMap exists for new/3rd-party backends and overrides.
 
+Two optional keys tune the cachedir restore's page-cache prewarm per storage
+class, because whether the parallel read-ahead helps depends on the volume,
+not the model (see the 70B A/B in `docs/BENCHMARK.md`):
+
+- `prewarm` (default `true`): add the `nvsnap-prewarm` init container that
+  reads the whole rox tree before the engine starts. Set `false` where a
+  single reader already saturates the volume.
+- `prewarmParallelism` (default 6): concurrent readers in the sweep. Raise it
+  on volumes whose throughput scales with parallel readers (Hyperdisk ML).
+
+A pod's own `NVSNAP_PREWARM=0` or `NVSNAP_PREWARM=1` overrides the profile.
+The agent resolves the profile once at startup and hands it to the webhook,
+so a ConfigMap edit takes effect on the next agent restart.
+
 ## What stays put (no change)
 
 - Hash-keyed Lease serialization, single-copy mount-holder write phase, the

src/compute-plane-services/nvsnap/internal/agent/agent.go

@@ -269,6 +269,11 @@ type Agent struct {
 	// → CRIU dumps stay on L1 hostpath + L4 blobstore; multi-node
 	// restore falls back to L3 peer cascade.
 	l2Backend checkpointstore.Backend
+	// l2Profile is the StorageProfile resolved for the L2 StorageClass at
+	// startup (nil when L2 is off or nothing matched). The webhook reads
+	// its prewarm policy; the promoter strategy is already baked into
+	// l2Backend.
+	l2Profile *checkpointstore.StorageProfile
 
 	// kubeClient is the shared K8s API client used by the rootfs-only
 	// capture watcher AND the admission-webhook cascade-fetch path

src/compute-plane-services/nvsnap/internal/agent/checkpoint_v2.go

@@ -503,4 +503,3 @@ func tailOfFile(path string, n int) string {
 	}
 	return strings.Join(lines, " | ")
 }
-

src/compute-plane-services/nvsnap/internal/agent/l2_integration.go

@@ -51,12 +51,16 @@ const storageProfilesConfigMap = "nvsnap-storage-profiles"
 // bad ConfigMap, construct error) — the backend then falls back to its
 // default snapshot-clone ROX promoter. Always logs the resolved strategy
 // (or the reason for fallback) so an operator can see what L2 will do.
-func resolveL2Promoter(ctx context.Context, kc kubernetes.Interface, dyn dynamic.Interface, scName, namespace string, log logrus.FieldLogger) checkpointstore.Promoter {
+//
+// The resolved profile is returned alongside the promoter because the
+// webhook consumes it too (cachedir prewarm policy); nil when nothing
+// matched, so callers fall back to the profile's zero-value defaults.
+func resolveL2Promoter(ctx context.Context, kc kubernetes.Interface, dyn dynamic.Interface, scName, namespace string, log logrus.FieldLogger) (checkpointstore.Promoter, *checkpointstore.StorageProfile) {
 	sc, err := kc.StorageV1().StorageClasses().Get(ctx, scName, metav1.GetOptions{})
 	if err != nil {
 		log.WithError(err).WithField("sc", scName).
 			Warn("L2 storage profile: cannot read StorageClass; falling back to default snapshot-clone ROX promoter")
-		return nil
+		return nil, nil
 	}
 	provisioner := sc.Provisioner
 	volType := sc.Parameters["type"] // disambiguates pd.csi (hyperdisk-ml vs pd-ssd)
@@ -76,20 +80,21 @@ func resolveL2Promoter(ctx context.Context, kc kubernetes.Interface, dyn dynamic
 	if !ok {
 		log.WithFields(logrus.Fields{"sc": scName, "provisioner": provisioner, "type": volType}).
 			Warn("L2 storage profile: no profile for provisioner[/type]; falling back to default snapshot-clone ROX promoter (add an entry to the nvsnap-storage-profiles ConfigMap to support this backend)")
-		return nil
+		return nil, nil
 	}
 	promoter, err := checkpointstore.NewPromoterFromProfile(profile, scName, kc, dyn, log)
 	if err != nil {
 		log.WithError(err).WithField("strategy", profile.Strategy).
 			Warn("L2 storage profile: cannot construct promoter; falling back to default")
-		return nil
+		return nil, &profile
 	}
 	log.WithFields(logrus.Fields{
 		"sc": scName, "provisioner": provisioner, "type": volType,
 		"profile_key": key, "strategy": profile.Strategy,
 		"read_only_many": profile.ReadOnlyMany, "vh_transform": profile.VolumeHandleTransform,
+		"prewarm": profile.PrewarmEnabled(), "prewarm_workers": profile.PrewarmWorkers(),
 	}).Info("L2 storage profile resolved")
-	return promoter
+	return promoter, &profile
 }
 
 // startL2Backend constructs the PerCapturePVCBackend if L2 is enabled
@@ -169,7 +174,8 @@ func (a *Agent) startL2Backend(_ context.Context, cfg L2BackendConfig) (checkpoi
 	// parameters.type (nvsnap#171). nil ⇒ the backend's applyDefaults
 	// falls back to the snapshot-clone ROX promoter (Hyperdisk-ML
 	// behavior) — back-compat for clusters with no profile match.
-	promoter := resolveL2Promoter(context.Background(), kc, dyn, cfg.StorageClass, cfg.Namespace, log)
+	promoter, profile := resolveL2Promoter(context.Background(), kc, dyn, cfg.StorageClass, cfg.Namespace, log)
+	a.l2Profile = profile
 
 	// SnapshotClass is only meaningful for the snapshot-clone strategy.
 	// Shared-volume backends (NVMesh/EFS/Filestore) never snapshot — they

src/compute-plane-services/nvsnap/internal/agent/l2_profile_prewarm_test.go

@@ -0,0 +1,62 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package agent
+
+import (
+	"context"
+	"testing"
+
+	"github.com/sirupsen/logrus"
+	corev1 "k8s.io/api/core/v1"
+	storagev1 "k8s.io/api/storage/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+// The resolver returns the profile next to the promoter so the webhook can
+// read the prewarm policy, and a nvsnap-storage-profiles ConfigMap entry
+// that turns the prewarm off for a provisioner reaches it unchanged.
+func TestResolveL2Promoter_ReturnsProfileWithPrewarmPolicy(t *testing.T) {
+	log := logrus.New()
+	log.SetLevel(logrus.PanicLevel)
+	sc := &storagev1.StorageClass{
+		ObjectMeta:  metav1.ObjectMeta{Name: "nvcf-sc"},
+		Provisioner: "nvmesh-csi.excelero.com",
+	}
+
+	kc := fake.NewSimpleClientset(sc)
+	promoter, profile := resolveL2Promoter(context.Background(), kc, nil, "nvcf-sc", "nvsnap-system", log)
+	if promoter == nil || profile == nil {
+		t.Fatalf("built-in NVMesh profile: promoter=%v profile=%v, want both", promoter, profile)
+	}
+	if !profile.PrewarmEnabled() || profile.PrewarmWorkers() != 6 {
+		t.Errorf("built-in profile ships prewarm on with 6 readers, got enabled=%v workers=%d", profile.PrewarmEnabled(), profile.PrewarmWorkers())
+	}
+
+	cm := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{Name: storageProfilesConfigMap, Namespace: "nvsnap-system"},
+		Data: map[string]string{"profiles.yaml": `
+nvmesh-csi.excelero.com:
+  strategy: shared-volume
+  volumeHandleTransform: nvmesh
+  mountOptions: [ro, norecovery, nouuid]
+  prewarm: false
+  prewarmParallelism: 2
+`},
+	}
+	kc = fake.NewSimpleClientset(sc, cm)
+	promoter, profile = resolveL2Promoter(context.Background(), kc, nil, "nvcf-sc", "nvsnap-system", log)
+	if promoter == nil || profile == nil {
+		t.Fatalf("ConfigMap overlay: promoter=%v profile=%v, want both", promoter, profile)
+	}
+	if profile.PrewarmEnabled() || profile.PrewarmWorkers() != 2 {
+		t.Errorf("ConfigMap prewarm policy lost on the way to the webhook: enabled=%v workers=%d", profile.PrewarmEnabled(), profile.PrewarmWorkers())
+	}
+
+	// No match: nothing to hand the webhook, it falls back to defaults.
+	unknown := &storagev1.StorageClass{ObjectMeta: metav1.ObjectMeta{Name: "weka"}, Provisioner: "csi.weka.io"}
+	if _, profile := resolveL2Promoter(context.Background(), fake.NewSimpleClientset(unknown), nil, "weka", "nvsnap-system", log); profile != nil {
+		t.Errorf("unmatched provisioner must yield a nil profile, got %+v", profile)
+	}
+}

src/compute-plane-services/nvsnap/internal/agent/webhook_integration.go

@@ -149,6 +149,9 @@ func (a *Agent) startWebhook(ctx context.Context, cfg WebhookConfig, backend che
 		// restore-side RO mount agree. Empty = standard paths.
 		CacheDir:     a.config.RootfsCapture.PodCacheDir,
 		CacheEnvFile: a.config.RootfsCapture.PodCacheEnvFile,
+		// Storage profile of the L2 StorageClass: the cachedir restore
+		// reads its prewarm policy (on/off, reader count). nil = defaults.
+		StorageProfile: a.l2Profile,
 		Composer: &rootfsonly.HashInputComposer{
 			CUDADriverMajor: a.config.RootfsCapture.CUDADriverMajor,
 		},

src/compute-plane-services/nvsnap/internal/checkpointstore/storage_profile.go

@@ -69,6 +69,35 @@ type StorageProfile struct {
 	// (RO mount of a possibly-dirty log). Empty ⇒ inherit the primary PV's
 	// options unchanged (fine for NFS-like backends: EFS, Filestore).
 	MountOptions []string `json:"mountOptions,omitempty"`
+	// Prewarm controls the nvsnap-prewarm init container the webhook adds
+	// to a cachedir restore: a parallel sequential read of the whole rox
+	// tree so the engine's weight load hits page cache. nil (absent) means
+	// on. Whether it helps is a property of the volume, not the model: on
+	// Hyperdisk ML the engine's per-fault mmap reads leave a high-throughput
+	// volume idle and the sweep wins; on a volume a single reader already
+	// saturates (NVMesh, measured 70B TP=4: 247 s vs 257 s) it is neutral.
+	// A pod's own NVSNAP_PREWARM=0/1 overrides this either way.
+	Prewarm *bool `json:"prewarm,omitempty"`
+	// PrewarmParallelism is the number of concurrent readers in the sweep.
+	// 0 means DefaultPrewarmParallelism.
+	PrewarmParallelism int `json:"prewarmParallelism,omitempty"`
+}
+
+// DefaultPrewarmParallelism is the reader count when a profile does not set
+// one; matches the retired Go prewarmer.
+const DefaultPrewarmParallelism = 6
+
+// PrewarmEnabled reports the profile's prewarm default (on when unset).
+func (p StorageProfile) PrewarmEnabled() bool {
+	return p.Prewarm == nil || *p.Prewarm
+}
+
+// PrewarmWorkers returns the reader count, defaulted and clamped to >= 1.
+func (p StorageProfile) PrewarmWorkers() int {
+	if p.PrewarmParallelism < 1 {
+		return DefaultPrewarmParallelism
+	}
+	return p.PrewarmParallelism
 }
 
 // builtinProfiles is the compiled-in provider table. Keyed by

src/compute-plane-services/nvsnap/internal/checkpointstore/storage_profile_test.go

@@ -122,3 +122,55 @@ func TestNewPromoterFromProfile(t *testing.T) {
 		t.Error("unknown strategy should error")
 	}
 }
+
+// The prewarm policy defaults to on with six readers when a profile says
+// nothing, and a ConfigMap entry can turn it off or resize it.
+func TestStorageProfile_PrewarmPolicy(t *testing.T) {
+	var unset StorageProfile
+	if !unset.PrewarmEnabled() || unset.PrewarmWorkers() != DefaultPrewarmParallelism {
+		t.Errorf("zero profile: enabled=%v workers=%d, want on/%d", unset.PrewarmEnabled(), unset.PrewarmWorkers(), DefaultPrewarmParallelism)
+	}
+	off := false
+	if (StorageProfile{Prewarm: &off}).PrewarmEnabled() {
+		t.Error("prewarm: false must disable")
+	}
+	on := true
+	if !(StorageProfile{Prewarm: &on}).PrewarmEnabled() {
+		t.Error("prewarm: true must enable")
+	}
+	if got := (StorageProfile{PrewarmParallelism: 12}).PrewarmWorkers(); got != 12 {
+		t.Errorf("parallelism 12 -> %d", got)
+	}
+	if got := (StorageProfile{PrewarmParallelism: -3}).PrewarmWorkers(); got != DefaultPrewarmParallelism {
+		t.Errorf("negative parallelism must fall back to the default, got %d", got)
+	}
+
+	m, err := ParseConfigMapProfiles(`
+nvmesh-csi.excelero.com:
+  strategy: shared-volume
+  volumeHandleTransform: nvmesh
+  prewarm: false
+pd.csi.storage.gke.io/hyperdisk-ml:
+  strategy: snapshot-clone
+  snapshotClass: hdml-images-snapshot-class
+  readOnlyMany: true
+  prewarmParallelism: 16
+`)
+	if err != nil {
+		t.Fatalf("parse: %v", err)
+	}
+	if m["nvmesh-csi.excelero.com"].PrewarmEnabled() {
+		t.Error("ConfigMap prewarm: false not honoured")
+	}
+	hd := m["pd.csi.storage.gke.io/hyperdisk-ml"]
+	if !hd.PrewarmEnabled() || hd.PrewarmWorkers() != 16 {
+		t.Errorf("hyperdisk entry: enabled=%v workers=%d, want on/16", hd.PrewarmEnabled(), hd.PrewarmWorkers())
+	}
+	// Every built-in ships with the prewarm on: nothing measured so far
+	// justifies turning it off by default anywhere.
+	for k, p := range builtinProfiles {
+		if !p.PrewarmEnabled() {
+			t.Errorf("built-in %s ships with prewarm off", k)
+		}
+	}
+}

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

@@ -145,10 +145,13 @@ func (w *Watcher) handlePodEvent(ctx context.Context, obj any) {
 	if !ok || pod == nil {
 		return
 	}
+	plog := w.logger().WithField("pod", pod.Namespace+"/"+pod.Name)
 	if !w.isLabeledForCapture(pod) {
+		plog.Debug("watcher: skipping, not labeled for capture")
 		return
 	}
 	if !IsPodReady(pod) {
+		plog.Debug("watcher: skipping, pod not Ready yet")
 		return
 	}
 	// Rootfs-only capture is the multi-GPU fallback (cuda-checkpoint can't
@@ -165,15 +168,18 @@ func (w *Watcher) handlePodEvent(ctx context.Context, obj any) {
 	// signal (the watch already filters by it at line 130), so we honor it
 	// even for single-GPU pods and let rootfs replace CRIU.
 	gpus := podGPURequest(pod)
-	if gpus < 2 {
-		w.logger().WithFields(logrus.Fields{
-			"pod":  pod.Namespace + "/" + pod.Name,
-			"gpus": gpus,
-		}).Info("rootfs-only watcher capturing single-GPU pod (explicit nvsnap.io/capture=true opt-in)")
-	}
 	if _, alreadyScheduled := w.captured.LoadOrStore(pod.UID, struct{}{}); alreadyScheduled {
+		plog.Debug("watcher: skipping, a capture is already scheduled or committed for this pod UID")
 		return
 	}
+	// Logged for every pod, not only single-GPU ones. This used to fire only
+	// when gpus < 2, so a multi-GPU pod -- the case this path exists to serve
+	// -- produced no output on any branch, and a capture that silently never
+	// happened was indistinguishable from one never triggered.
+	plog.WithFields(logrus.Fields{
+		"gpus":   gpus,
+		"warmup": w.WarmupDelay,
+	}).Info("watcher: scheduling rootfs capture (nvsnap.io/capture=true)")
 	go w.runCapture(ctx, pod.DeepCopy())
 }
 
@@ -228,6 +234,19 @@ func (w *Watcher) refreshPodForCapture(ctx context.Context, pod *corev1.Pod, log
 // retried by re-firing on subsequent Pod Update events (we clear captured
 // on persistent error so the next event re-tries).
 func (w *Watcher) runCapture(ctx context.Context, pod *corev1.Pod) {
+	// captured marks a pod UID as scheduled, and handlePodEvent treats a
+	// marked UID as nothing-to-do. Only a committed capture may keep the
+	// mark: every other exit here (warmup cancelled, semaphore wait
+	// cancelled, pod refresh failed, capture failed) has to release it or
+	// the pod is never retried and every later event returns silently.
+	// Previously only the capture-error path released it, so an abort left
+	// the UID poisoned for the life of the agent.
+	committed := false
+	defer func() {
+		if !committed {
+			w.captured.Delete(pod.UID)
+		}
+	}()
 	log := w.logger().WithFields(logrus.Fields{
 		"pod":     pod.Namespace + "/" + pod.Name,
 		"pod_uid": string(pod.UID),
@@ -236,13 +255,15 @@ func (w *Watcher) runCapture(ctx context.Context, pod *corev1.Pod) {
 		select {
 		case <-time.After(w.WarmupDelay):
 		case <-ctx.Done():
+			log.Debug("watcher: warmup cancelled before capture; releasing the pod for retry")
 			return
 		}
 	}
 	select {
 	case w.sem <- struct{}{}:
 		defer func() { <-w.sem }()
 	case <-ctx.Done():
+		log.Debug("watcher: cancelled waiting for a capture slot; releasing the pod for retry")
 		return
 	}
 
@@ -282,15 +303,15 @@ func (w *Watcher) runCapture(ctx context.Context, pod *corev1.Pod) {
 	m, err := w.Capturer.Capture(captureCtx, req)
 	if err != nil {
 		log.WithError(err).Warn("capture failed; will retry on next Update event")
-		// Allow retry on subsequent events.
-		w.captured.Delete(pod.UID)
+		// The deferred release above re-arms the pod; no explicit Delete.
 		return
 	}
 	log.WithFields(logrus.Fields{
 		"hash":     m.Hash[:12],
 		"size_mib": m.TotalSizeBytes / 1024 / 1024,
 		"files":    m.FileCount,
 	}).Info("capture committed for pod")
+	committed = true
 }
 
 func (w *Watcher) isLabeledForCapture(pod *corev1.Pod) bool {