fix(operator): do not record a config Job that ran on replaced ConfigMap data

docs/designs/2026-07-10-package-execution-as-jobs.md

@@ -120,7 +120,7 @@ Interrupt Jobs are the exception and keep an unbounded limit. Under `OnFailure`
 
 Because completed pods now linger, pod-deletion can no longer be the processed-once marker. A persisted Job annotation, `nodewright.nvidia.com/state-recorded: "true"`, replaces it. For each unprocessed Job on the node:
 
-- **`Complete`, not yet marked** → read the `Succeeded` child pod for the container name (selected by the Job's controller UID, not by name — a same-named prior Job's pod can still be terminating), record completion through the existing node-state path (upgrade/uninstall/interrupt special cases unchanged), then one Job Update sets `state-recorded` and the success TTL. If the child pod was already GC'd, fall back to the Job's own stage/node/interrupt labels. Already-marked Jobs are skipped, so duplicate events are harmless. The recorded attempt count now comes from the Job's failed-pod count — a user-visible improvement over today's usually-zero restart count.
+- **`Complete`, not yet marked** → read the `Succeeded` child pod for the container name (selected by the Job's controller UID, not by name — a same-named prior Job's pod can still be terminating), record completion through the existing node-state path (upgrade/uninstall/interrupt special cases unchanged), then one Job Update sets `state-recorded` and the success TTL. If the child pod was already GC'd, fall back to the Job's own stage/node/interrupt labels. Already-marked Jobs are skipped, so duplicate events are harmless. A config-stage Job also carries `nodewright.nvidia.com/config-hash`, a hash of the ConfigMap data it was created against; if the live ConfigMap no longer matches, the completion is not recorded and the Job is deleted so the stage re-runs on the current data (a config-update reset puts the entry back at `config`, which otherwise looks identical to the run the Job did). The recorded attempt count now comes from the Job's failed-pod count — a user-visible improvement over today's usually-zero restart count.
 - **`Failed`, on genuine evidence** → the stage is out of attempts (`BackoffLimitExceeded` with at least one retained archive that really failed) or hit the Job-level ceiling (`DeadlineExceeded`, which needs no archive — see Stage deadline, below). Record `erroring`, mark processed with the failure TTL, and **leave the Job in place as the timeout marker**.
 - **`Failed`, without it** → every retained attempt was a pod the package never ran in, the kubelet-admission case below. No node-state write (matching today's silent disruption behavior); one Update carries the marker and failure TTL, and the sweep then clears the Job so the stage re-runs.
 - **Node gone** → if the node `Get` returns NotFound, there is no state to record: mark the Job processed and stop, rather than error-looping on a node that will never return.
@@ -223,14 +223,14 @@ Scale: succeeded pods are retained for the success window — roughly `nodes ×
 Job names are deterministic (same inputs as today's pod names), which makes creation idempotent — the name *is* the claim. But a finished Job keeps its name until TTL, so every "run this stage again" path must first clear the name. Five rules replace today's delete-on-success; "processed" means `state-recorded` is set, for both outcomes, and Job deletions are always **foreground** so a deterministic name frees only after its child pods are gone.
 
 1. **Validation is scoped to unfinished Jobs.** The spec-mismatch / not-in-node-state / stage-mismatch checks apply only to unfinished Jobs; applying them to finished Jobs would delete every retained Job the moment the node progresses, gutting Goal 2.
-2. **The rerun predicate.** A processed finished Job for stage `S` is deleted once node state no longer records `S` as done (entry absent, or progress at/before `S` without `S` complete — compared via the state machine's stage ordering, not string equality). **Timeout exception:** a `Failed` Job is *not* deleted while the entry sits at (`S`, `erroring`) — that pair is the timed-out state; every other value clears it, which is what sweeps away a failure the operator judged non-genuine and never wrote state for. **Unprocessed** finished Jobs are never deleted except by the two sweeps — for `Complete` that protects an unrecorded completion, and for `Failed` it keeps this sweep from racing the erroring write, a window a finite `backoffLimit` shrinks from an hour to seconds.
+2. **The rerun predicate.** A processed finished Job for stage `S` is deleted once node state no longer records `S` as done (entry absent, or progress at/before `S` without `S` complete — compared via the state machine's stage ordering, not string equality). **Timeout exception:** a `Failed` Job is *not* deleted while the entry sits at (`S`, `erroring`) — that pair is the timed-out state; every other value clears it, which is what sweeps away a failure the operator judged non-genuine and never wrote state for. **Unprocessed** finished Jobs are never deleted except by the two sweeps and two exceptions (the config-hash check on a config-stage completion, above, and the config-update reset's deletion of unfinished, failed, or unrecorded config-stage `Complete` Jobs, which #788 added) — for `Complete` that protects an unrecorded completion, and for `Failed` it keeps this sweep from racing the erroring write, a window a finite `backoffLimit` shrinks from an hour to seconds.
 3. **`AlreadyExists` on create is not blindly benign.** GET the Job: unfinished and matching → another pass won the race, return. Timed out → return; the timeout is doing its job (this branch is what absorbs level-triggered recreate attempts on an erroring stage). Not yet marked → don't delete, whatever the outcome: a `Complete` one holds an unrecorded completion, and a `Failed` one has not yet had its chance to write `erroring`, so deleting it would take its retained attempts with it and restart the stage on a fresh budget. This path reaches that window first, because a finished Job does not satisfy the existence gate. Otherwise processed, or unfinished-but-mismatched → foreground-delete and requeue, without recording an `in_progress` that isn't happening.
 4. **Orphaned-node sweep.** Jobs whose node no longer exists are deleted regardless of status — an unfinished one would otherwise churn PodGC↔replacement forever against the missing node, and a finished one has no node state left to claim it.
 5. **Reboot-reset sweep.** When `REAPPLY_ON_REBOOT` resets a node's state, that node's Jobs for the Skyhook are deleted regardless of status, including unprocessed `Complete` ones — a completion from a previous boot must not land on freshly-reset state. Rule 2's protection of unprocessed completions yields to this sweep.
 
 Two retention carve-outs follow and are accepted: a successful no-interrupt **uninstall** Job's completion *is* entry removal, so it satisfies the absent-entry branch immediately and gets no retention window; and because uninstall precedes apply in the ordering, starting an uninstall also clears the package's retained apply/config Jobs. Uninstall is the "remove all trace" operation, so prompt cleanup is coherent — and crash-loop *failures* of uninstall still retain logs (Active Job). The cost is that a *successful* uninstall's output is unrecoverable, since the Job and its pod are gone before the success TTL applies: a known limitation, tracked in [#443](https://github.com/NVIDIA/nodewright/issues/443) and deferred to a later release.
 
-The existence gates move with these rules: the "is this stage in flight" check and the "wait until nothing is running" interrupt gate both count **unfinished Jobs only** (a retained Succeeded pod must not hold an interrupt hostage — node cordoned and drained — until its TTL expires), and config-update handling deletes the package's **unfinished or terminally failed** Job rather than its child pod (which would just trigger a same-config replacement) — a config change invalidates that failure history anyway, and the deterministic name has to be freed for the rerun.
+The existence gates move with these rules: the "is this stage in flight" check and the "wait until nothing is running" interrupt gate both count **unfinished Jobs only** (a retained Succeeded pod must not hold an interrupt hostage — node cordoned and drained — until its TTL expires), and config-update handling deletes the package's **unfinished, terminally failed or unrecorded config-stage `Complete`** Job on the reset nodes, rather than just its child pod, which would only trigger a same-config replacement. A config change invalidates that failure history anyway, and the deterministic name has to be freed for the rerun. The `Complete` one is also dropped by the config-hash check when the ConfigMap has since changed; and once the ConfigMap is replaced, each reset node's entry is rewritten from a fresh read (`restartConfigStage`), so a completion recorded between the pass's node read and the ConfigMap update does not survive on the old data.
 
 ### Labels
 

operator/internal/controller/job_builder.go

@@ -34,6 +34,7 @@ import (
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 // doneContainerName replaces the forever-running pause main container. A Job records
@@ -298,7 +299,7 @@ func createPodFromPackage(opts SkyhookOperatorOptions, _package *v1alpha1.Packag
 			VolumeSource: corev1.VolumeSource{
 				ConfigMap: &corev1.ConfigMapVolumeSource{
 					LocalObjectReference: corev1.LocalObjectReference{
-						Name: strings.ToLower(fmt.Sprintf("%s-%s-%s", skyhook.Name, _package.Name, _package.Version)),
+						Name: packageConfigMapName(skyhook.Name, _package.Name, _package.Version),
 					},
 				},
 			},
@@ -655,3 +656,19 @@ func podMatchesPackage(opts SkyhookOperatorOptions, _package *v1alpha1.Package,
 
 	return true
 }
+
+// packageConfigMapName is the name of a package's ConfigMap, and the one the package pod mounts it
+// from. The config-hash check looks the ConfigMap up by this name, so a copy that drifted would turn
+// the check off without an error.
+func packageConfigMapName(skyhook, pkg, version string) string {
+	return strings.ToLower(fmt.Sprintf("%s-%s-%s", skyhook, pkg, version))
+}
+
+// uncachedReader is the apiserver reader when one is configured, else the (cached) client, which is
+// what the fake-client tests use.
+func uncachedReader(uncached client.Reader, c client.Client) client.Reader {
+	if uncached != nil {
+		return uncached
+	}
+	return c
+}

operator/internal/controller/job_controller.go

@@ -20,6 +20,8 @@ package controller
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"fmt"
 	"slices"
 	"sort"
@@ -74,6 +76,11 @@ const (
 	// interrupt Job created before the bound existed, is never ended by it.
 	annotationRestartAllowance = v1alpha1.METADATA_PREFIX + "/restart-allowance"
 
+	// annotationConfigHash is stamped at creation on a config-stage Job: a hash of the package
+	// ConfigMap data the pod mounts. The pod reads each key once at start, so a Job that ran
+	// against data since replaced must not have its completion recorded as the new data's.
+	annotationConfigHash = v1alpha1.METADATA_PREFIX + "/config-hash"
+
 	// interruptRestartLimitAction is the event action for an interrupt Job ended that way.
 	interruptRestartLimitAction = "RestartLimitExceeded"
 
@@ -126,7 +133,9 @@ const (
 type JobReconciler struct {
 	client.Client
 	// uncached reads straight from the apiserver: a Node re-read after a patch conflict (see
-	// patchNodeState), and a terminal Job's child pods (see packageJobRestarts).
+	// patchNodeState), a terminal Job's child pods (see packageJobRestarts), and the package
+	// ConfigMap behind the config-hash check (see configSuperseded), where a stale cache would
+	// match the old hash.
 	uncached client.Reader
 	recorder events.EventRecorder
 	opts     JobOperatorOptions
@@ -285,13 +294,57 @@ func (r *JobReconciler) handleCompleteJob(ctx context.Context, job *batchv1.Job)
 		return ctrl.Result{}, r.markJobProcessed(ctx, job, r.opts.JobTTLSucceeded)
 	}
 
+	// The entry is back at config after a config-update reset, which looks the same as the run this
+	// Job did, so recording would complete the node on data it never saw. Drop the Job instead;
+	// the stage is re-created from the entry against the current ConfigMap.
+	if superseded, err := r.configSuperseded(ctx, job, pkg); err != nil {
+		return ctrl.Result{}, err
+	} else if superseded {
+		return ctrl.Result{}, deleteJobForeground(ctx, r.Client, job, "config changed while the config stage ran")
+	}
+
 	if err := r.recordJobCompletion(ctx, job, pkg); err != nil {
 		return ctrl.Result{}, err
 	}
 
 	return ctrl.Result{}, r.markJobProcessed(ctx, job, r.opts.JobTTLSucceeded)
 }
 
+// configSuperseded reports a config-stage Job whose stamped ConfigMap hash no longer matches the
+// live ConfigMap. A Job without the stamp (created by an older operator) or a ConfigMap that is
+// gone is never superseded, so those still record as before.
+func (r *JobReconciler) configSuperseded(ctx context.Context, job *batchv1.Job, pkg *PackageSkyhook) (bool, error) {
+	stamped, ok := job.Annotations[annotationConfigHash]
+	if !ok || pkg.Stage != v1alpha1.StageConfig {
+		return false, nil
+	}
+	cm := &corev1.ConfigMap{}
+	key := types.NamespacedName{Namespace: job.Namespace, Name: packageConfigMapName(pkg.Skyhook, pkg.Name, pkg.Version)}
+	// uncached: a stale cache could match the old hash and let a superseded completion record
+	if err := uncachedReader(r.uncached, r.Client).Get(ctx, key, cm); err != nil {
+		if apierrors.IsNotFound(err) {
+			return false, nil
+		}
+		return false, fmt.Errorf("getting config map %s for job %s: %w", key.Name, job.Name, err)
+	}
+	return configDataHash(cm.Data) != stamped, nil
+}
+
+// configDataHash is a stable digest of ConfigMap data: keys sorted, each key and value
+// length-prefixed so neither can bleed into the next.
+func configDataHash(data map[string]string) string {
+	keys := make([]string, 0, len(data))
+	for k := range data {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	h := sha256.New()
+	for _, k := range keys {
+		_, _ = fmt.Fprintf(h, "%d:%s%d:%s", len(k), k, len(data[k]), data[k])
+	}
+	return hex.EncodeToString(h.Sum(nil))
+}
+
 // shouldRecordCompletion is the re-serve guard. It records a completion only when the
 // package's node-state entry is still at this Job's stage and not yet complete: a valid
 // forward transition. An absent entry (uninstalled or node reset), an already-complete

operator/internal/controller/job_controller_test.go

@@ -237,6 +237,95 @@ var _ = Describe("JobReconcile", func() {
 		Expect(getNodeState(r)[pkgRef.GetUniqueName()].Stage).To(Equal(v1alpha1.StageConfig))
 	})
 
+	// A config-update reset puts the entry back at (config, in_progress), the same shape the old
+	// run had, so the stamped ConfigMap hash is the only thing that says which data a Job saw.
+	Describe("config Job completion after the ConfigMap changed", func() {
+		configMap := func(data map[string]string) *corev1.ConfigMap {
+			return &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{
+				Namespace: namespace,
+				Name:      packageConfigMapName(skyhookName, pkgRef.Name, pkgRef.Version),
+			}, Data: data}
+		}
+		configJob := func(stamped map[string]string) *batchv1.Job {
+			job := packageJob(v1alpha1.StageConfig, false, trueCondition(batchv1.JobComplete, ""))
+			if stamped != nil {
+				job.Annotations[annotationConfigHash] = configDataHash(stamped)
+			}
+			return job
+		}
+
+		It("does not record a Job that ran on data since replaced, and drops the Job", func() {
+			node := nodeWithState(v1alpha1.StateInProgress, v1alpha1.StageConfig)
+			job := configJob(map[string]string{"a.properties": "old"})
+			r := newReconciler(node, job, configMap(map[string]string{"a.properties": "new"}))
+
+			_, err := r.JobReconcile(ctx, job)
+			Expect(err).ToNot(HaveOccurred())
+
+			Expect(getNodeState(r)[pkgRef.GetUniqueName()].State).To(Equal(v1alpha1.StateInProgress))
+			err = r.Get(ctx, types.NamespacedName{Namespace: namespace, Name: job.Name}, &batchv1.Job{})
+			Expect(apierrors.IsNotFound(err)).To(BeTrue(), "the stale Job is deleted so the stage re-runs")
+		})
+
+		It("judges the Job by the apiserver's ConfigMap, not a cache that still holds the old one", func() {
+			node := nodeWithState(v1alpha1.StateInProgress, v1alpha1.StageConfig)
+			job := configJob(map[string]string{"a.properties": "old"})
+
+			scheme := runtime.NewScheme()
+			Expect(corev1.AddToScheme(scheme)).To(Succeed())
+			Expect(batchv1.AddToScheme(scheme)).To(Succeed())
+			Expect(v1alpha1.AddToScheme(scheme)).To(Succeed())
+			// the cache still serves the data the Job ran on; only the live reader has the new one
+			cached := fake.NewClientBuilder().WithScheme(scheme).WithObjects(node, job, configMap(map[string]string{"a.properties": "old"})).Build()
+			live := fake.NewClientBuilder().WithScheme(scheme).WithObjects(configMap(map[string]string{"a.properties": "new"})).Build()
+			r := NewJobReconciler(cached, live, k8sfake.NewClientset(), events.NewFakeRecorder(50), validOpts().JobOperatorOptions)
+
+			_, err := r.JobReconcile(ctx, job)
+			Expect(err).ToNot(HaveOccurred())
+
+			Expect(getNodeState(r)[pkgRef.GetUniqueName()].State).To(Equal(v1alpha1.StateInProgress))
+		})
+
+		It("records a Job whose data still matches", func() {
+			node := nodeWithState(v1alpha1.StateInProgress, v1alpha1.StageConfig)
+			data := map[string]string{"a.properties": "new"}
+			job := configJob(data)
+			r := newReconciler(node, job, configMap(data))
+
+			_, err := r.JobReconcile(ctx, job)
+			Expect(err).ToNot(HaveOccurred())
+
+			Expect(getNodeState(r)[pkgRef.GetUniqueName()].State).To(Equal(v1alpha1.StateComplete))
+		})
+
+		It("records a Job created without the stamp", func() {
+			node := nodeWithState(v1alpha1.StateInProgress, v1alpha1.StageConfig)
+			job := configJob(nil)
+			r := newReconciler(node, job, configMap(map[string]string{"a.properties": "new"}))
+
+			_, err := r.JobReconcile(ctx, job)
+			Expect(err).ToNot(HaveOccurred())
+
+			Expect(getNodeState(r)[pkgRef.GetUniqueName()].State).To(Equal(v1alpha1.StateComplete))
+		})
+
+		It("records when the ConfigMap is gone", func() {
+			node := nodeWithState(v1alpha1.StateInProgress, v1alpha1.StageConfig)
+			job := configJob(map[string]string{"a.properties": "old"})
+			r := newReconciler(node, job)
+
+			_, err := r.JobReconcile(ctx, job)
+			Expect(err).ToNot(HaveOccurred())
+
+			Expect(getNodeState(r)[pkgRef.GetUniqueName()].State).To(Equal(v1alpha1.StateComplete))
+		})
+
+		It("hashes the data independent of key order and without key/value bleed", func() {
+			Expect(configDataHash(map[string]string{"a": "1", "b": "2"})).To(Equal(configDataHash(map[string]string{"b": "2", "a": "1"})))
+			Expect(configDataHash(map[string]string{"a": "bc"})).ToNot(Equal(configDataHash(map[string]string{"ab": "c"})))
+		})
+	})
+
 	It("does not regress a stage when a completion is re-served after the node advanced", func() {
 		// The apply Job re-fires, but node state has already moved on to config.
 		node := nodeWithState(v1alpha1.StateInProgress, v1alpha1.StageConfig)