From 0af4ba88d0fd7e8d3fc8748ffbaf2932b39b2298 Mon Sep 17 00:00:00 2001 From: Alex Yuskauskas Date: Tue, 11 Aug 2026 14:22:02 -0700 Subject: [PATCH 1/4] fix(operator): never resurrect a package entry on interrupt completion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An interrupt Job is package-agnostic — one per node, stage and interrupt type, deduped across packages — so its completion also has to promote siblings left at (interrupt, skipped). shouldRecordCompletion's interrupt branch authorizes the write on the strength of such a sibling, before looking at the Job's own package at all. recordJobCompletion then took that as license for its own package too: HandleCompletePod's interrupt branch only promotes and reports no update, so the fallback Upsert always ran, and Upsert creates. A rerun, reset or finalizer-driven uninstall that removed the entry while the interrupt Job was completing therefore got it back at (interrupt, complete) — and with the entry present and complete, the rerun predicate keeps the Job, so the stage never runs again. The rerun the user asked for silently does nothing until the failure TTL. The self-write is now gated on entryAwaitsCompletion (present, at this stage, not complete), extracted so shouldRecordCompletion's non-interrupt tail and this guard cannot drift. Promotion is untouched. State is re-read after HandleCompletePod, since promotion can move this package's own entry. Closes #426. Signed-off-by: Alex Yuskauskas --- .../internal/controller/job_controller.go | 28 +++++++++++++++-- .../controller/job_controller_test.go | 31 +++++++++++++++++++ 2 files changed, 57 insertions(+), 2 deletions(-) diff --git a/operator/internal/controller/job_controller.go b/operator/internal/controller/job_controller.go index bfb4dadf7..477f1edbc 100644 --- a/operator/internal/controller/job_controller.go +++ b/operator/internal/controller/job_controller.go @@ -223,6 +223,9 @@ func (r *JobReconciler) shouldRecordCompletion(job *batchv1.Job, pkg *PackageSky if isInterruptJob(job) { // ProgressSkipped promotes only StageInterrupt-skipped packages, so re-run the record // only while such a package remains, matching what the promotion can actually advance. + // + // This authorizes the write on a SIBLING's state, so it says nothing about this Job's own + // package — recordJobCompletion gates its self-write on entryAwaitsCompletion separately. for _, s := range state { if s.Stage == v1alpha1.StageInterrupt && s.State == v1alpha1.StateSkipped { return true, nil @@ -230,8 +233,16 @@ func (r *JobReconciler) shouldRecordCompletion(job *batchv1.Job, pkg *PackageSky } } + return entryAwaitsCompletion(state, pkg), nil +} + +// entryAwaitsCompletion reports whether the package's entry is in the only shape a completion may +// be written onto: present, still at this Job's stage, and not already complete. Absent means +// removed (rerun, reset, uninstall) and writing would resurrect it; a later stage or an existing +// complete would regress or duplicate. +func entryAwaitsCompletion(state v1alpha1.NodeState, pkg *PackageSkyhook) bool { status, present := state[pkg.GetUniqueName()] - return present && status.Stage == pkg.Stage && status.State != v1alpha1.StateComplete, nil + return present && status.Stage == pkg.Stage && status.State != v1alpha1.StateComplete } // patchNodeState re-reads the node, applies mutate, and patches under an optimistic-lock @@ -318,7 +329,20 @@ func (r *JobReconciler) recordJobCompletion(ctx context.Context, job *batchv1.Jo if err != nil { return false, fmt.Errorf("recording completion for job %s: %w", job.Name, err) } - if !updated { + + // Read after HandleCompletePod: its interrupt branch promotes skipped packages, which can + // move this package's own entry. + state, err := skyhookNode.State() + if err != nil { + return false, fmt.Errorf("reading node state for job %s: %w", job.Name, err) + } + + // Upsert creates, so it must never run on an entry that is not there. An interrupt Job + // reaches this on a sibling's behalf (see shouldRecordCompletion), so its own package may + // have been removed by a rerun, reset or uninstall since the Job started; re-creating it + // at (interrupt, complete) would then make the rerun predicate keep the Job, and the stage + // would never run again. The promotion above still lands either way. + if !updated && entryAwaitsCompletion(state, pkg) { if err := skyhookNode.Upsert(pkg.PackageRef, pkg.Image, v1alpha1.StateComplete, pkg.Stage, job.Status.Failed, pkg.ContainerSHA); err != nil { return false, fmt.Errorf("upserting complete state for job %s: %w", job.Name, err) } diff --git a/operator/internal/controller/job_controller_test.go b/operator/internal/controller/job_controller_test.go index ccd5b6fda..ffa8540d6 100644 --- a/operator/internal/controller/job_controller_test.go +++ b/operator/internal/controller/job_controller_test.go @@ -412,6 +412,37 @@ var _ = Describe("JobReconcile", func() { Expect(state[sibling.GetUniqueName()].State).To(Equal(v1alpha1.StateComplete)) }) + It("promotes the sibling without resurrecting a package whose entry was removed", func() { + // A rerun/reset/uninstall clears an entry while that package's interrupt Job is + // completing. The Job is package-agnostic and still owes the sibling its promotion, but + // re-creating the cleared entry would put it back at (interrupt, complete) — the rerun + // predicate would then keep the Job and the stage would never run again, so the rerun + // the user asked for would silently do nothing. + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: nodeName}} + sn, err := wrapper.NewSkyhookNodeOnly(node, skyhookName) + Expect(err).ToNot(HaveOccurred()) + sibling := v1alpha1.PackageRef{Name: "other", Version: "2.0.0"} + Expect(sn.Upsert(sibling, image, v1alpha1.StateSkipped, v1alpha1.StageInterrupt, 0, "")).To(Succeed()) + // deliberately no entry for pkgRef: that is the removal + + scr := &v1alpha1.NodeWright{ + ObjectMeta: metav1.ObjectMeta{Name: skyhookName}, + Spec: v1alpha1.NodeWrightSpec{Packages: v1alpha1.Packages{ + "tuning": {PackageRef: pkgRef, Image: image}, + "other": {PackageRef: sibling, Image: image}, + }}, + } + job := packageJob(v1alpha1.StageInterrupt, true, trueCondition(batchv1.JobComplete, "")) + r := newReconciler(node, scr, job) + + _, err = r.JobReconcile(ctx, job) + Expect(err).ToNot(HaveOccurred()) + + state := getNodeState(r) + Expect(state).ToNot(HaveKey(pkgRef.GetUniqueName()), "the removal must stand") + Expect(state[sibling.GetUniqueName()].State).To(Equal(v1alpha1.StateComplete), "the sibling is still promoted") + }) + It("records erroring (state only, no marker) for a stale FailureTarget on an unreachable node", func() { node := nodeWithState(v1alpha1.StateInProgress, v1alpha1.StageConfig) job := packageJob(v1alpha1.StageConfig, false) From 2bac8006c2713cf459ebf35927456e883129873f Mon Sep 17 00:00:00 2001 From: Alex Yuskauskas Date: Tue, 11 Aug 2026 15:09:36 -0700 Subject: [PATCH 2/4] fix(operator): guard the interrupt branch's CR read and the completion event MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-ups from review of the previous commit. A deleted NodeWright reads as (nil, nil) from GetSkyhook, and wrapper.Convert dereferences it, so HandleCompletePod's interrupt branch panicked when the CR was removed while a completed interrupt Job was still unprocessed — the same window the entry guard is about. The uninstall branch two blocks down already nil-checks; this one now does too. Pre-existing, but it is the same failure family, and the new spec panics without the guard. The success event could also lie once the self-write is gated: an interrupt Job that reaches the recorder purely on a sibling's promotion left skyhookNode.Changed() true, so the operator announced "Package [x:1.0.0] state complete" for a package whose entry a reset had just cleared. The event now follows what was actually written. Also folds shouldRecordPodErroring onto entryAwaitsCompletion — it was a third verbatim copy of the same predicate in the same package — and notes on recordJobErroring why its erroring-exclusion is deliberately not the same helper. Specs added for the CR-gone panic, the upgrade branch's RemoveState path (previously untested), and no-regression from a later stage on the interrupt path, which is the direction only the new guard covers. The design doc now records that the create-nothing rule binds the Job path too, not just the Pod watch. Signed-off-by: Alex Yuskauskas --- .../2026-07-10-package-execution-as-jobs.md | 2 + .../internal/controller/job_controller.go | 43 ++++++++++--- .../controller/job_controller_test.go | 63 +++++++++++++++++++ .../internal/controller/pod_controller.go | 3 +- 4 files changed, 100 insertions(+), 11 deletions(-) diff --git a/docs/designs/2026-07-10-package-execution-as-jobs.md b/docs/designs/2026-07-10-package-execution-as-jobs.md index 7a1b9436c..5b17cb597 100644 --- a/docs/designs/2026-07-10-package-execution-as-jobs.md +++ b/docs/designs/2026-07-10-package-execution-as-jobs.md @@ -128,6 +128,8 @@ Because completed pods now linger, pod-deletion can no longer be the processed-o The Pod watch is **evidence, not authority**: it may only update an entry that already exists at the pod's stage and is not yet complete. It must never create one. Under `restartPolicy: Never` a failing Job mints a fresh pod per attempt, indefinitely, so an unguarded write becomes a repeating one. If it could create an entry, a node-state reset would be undone by the very Job the reset is meant to clear: the resurrected entry makes the reset invisible to the not-in-node-state check in rule 1 below, so the stale Job is never invalidated, and the existence gate then blocks the package's new stage forever. The same guard also stops a retained failed-attempt archive pod (kept on purpose, below) from regressing a completion the Job path already recorded. +**Neither may the Job path create an entry**, for the same reason, even though it *is* the completion authority. Authority means it decides when a stage is done, not that it may write a package state says is gone. This bites on interrupt Jobs specifically: they are package-agnostic, so a completion is recorded on the strength of a *sibling* still sitting at (interrupt, `skipped`), which says nothing about whether this Job's own package still has an entry. A rerun, reset or uninstall landing in that window would otherwise see the entry return at (interrupt, `complete`), where the rerun predicate keeps the Job and the stage never runs again. The promotion of skipped siblings happens either way; only the self-write is gated. + Recording a completion is two writes to two objects (node state, then the Job marker) and cannot be atomic; a crash between them re-serves the event. The re-processing path is guarded by per-transition postcondition checks so a re-served completion is only marked, not re-applied — detail in [Edge cases](#edge-cases-and-correctness-arguments). This is strictly better than today, which has the same two-write window with no guard. ### Retry and the failed-attempt archive diff --git a/operator/internal/controller/job_controller.go b/operator/internal/controller/job_controller.go index 477f1edbc..0e6f54575 100644 --- a/operator/internal/controller/job_controller.go +++ b/operator/internal/controller/job_controller.go @@ -342,17 +342,33 @@ func (r *JobReconciler) recordJobCompletion(ctx context.Context, job *batchv1.Jo // have been removed by a rerun, reset or uninstall since the Job started; re-creating it // at (interrupt, complete) would then make the rerun predicate keep the Job, and the stage // would never run again. The promotion above still lands either way. + // + // Both guards are kept deliberately: updated says HandleCompletePod already wrote this + // entry, entryAwaitsCompletion says there is an entry to write onto. They agree today — + // every branch that sets updated leaves the entry absent or at another stage — and neither + // is safe to drop on the strength of the other. + recorded := false if !updated && entryAwaitsCompletion(state, pkg) { if err := skyhookNode.Upsert(pkg.PackageRef, pkg.Image, v1alpha1.StateComplete, pkg.Stage, job.Status.Failed, pkg.ContainerSHA); err != nil { return false, fmt.Errorf("upserting complete state for job %s: %w", job.Name, err) } + recorded = true } if !skyhookNode.Changed() { return false, nil } - r.recorder.Eventf(node, nil, EventTypeNormal, EventsReasonSkyhookStateChange, "JobComplete", - "Package [%s:%s] state %s on [skyhook:%s]", pkg.Name, pkg.Version, v1alpha1.StateComplete, pkg.Skyhook) + + // The event follows what was actually written. An interrupt Job can reach here purely on a + // sibling's promotion with its own entry gone, and announcing that package complete would + // name the one a reset just cleared. + if recorded || updated { + r.recorder.Eventf(node, nil, EventTypeNormal, EventsReasonSkyhookStateChange, "JobComplete", + "Package [%s:%s] state %s on [skyhook:%s]", pkg.Name, pkg.Version, v1alpha1.StateComplete, pkg.Skyhook) + } else { + r.recorder.Eventf(node, nil, EventTypeNormal, EventsReasonSkyhookStateChange, "JobComplete", + "Interrupt complete on [skyhook:%s]: promoted packages skipped during interrupt sequencing", pkg.Skyhook) + } return true, nil }) } @@ -373,14 +389,20 @@ func (r *JobReconciler) HandleCompletePod(ctx context.Context, skyhookNode wrapp return false, err } - upgraded, err := wrapper.Convert(skyhookNode, skyhook) - if err != nil { - return false, fmt.Errorf("error converting node wrapper: %w", err) - } + // A deleted CR reads as (nil, nil), and Convert dereferences it. That is reachable in the + // same window this stage's guards are about: the CR removed while a completed interrupt + // Job is still unprocessed. Nothing is left to promote then, so skip the branch rather + // than panic — matching the nil check the uninstall branch below already does. + if skyhook != nil { + upgraded, err := wrapper.Convert(skyhookNode, skyhook) + if err != nil { + return false, fmt.Errorf("error converting node wrapper: %w", err) + } - // progress forward any skipped packages that this interrupt completed - if err := upgraded.ProgressSkipped(); err != nil { - return false, fmt.Errorf("error progressing skipped packages: %w", err) + // progress forward any skipped packages that this interrupt completed + if err := upgraded.ProgressSkipped(); err != nil { + return false, fmt.Errorf("error progressing skipped packages: %w", err) + } } } else if packagePtr.Stage == v1alpha1.StageUpgrade { nodeState, err := skyhookNode.State() @@ -546,6 +568,9 @@ func (r *JobReconciler) recordJobErroring(ctx context.Context, job *batchv1.Job, return false, fmt.Errorf("reading node state for job %s: %w", job.Name, err) } + // Deliberately NOT entryAwaitsCompletion: the completion guard excludes an entry that is + // already complete, this one excludes an entry that is already erroring, for idempotence + // on a re-served terminal event. Same shape, different exclusion — do not unify them. status, present := state[pkg.GetUniqueName()] if !present || status.Stage != pkg.Stage || status.State == v1alpha1.StateErroring { return false, nil diff --git a/operator/internal/controller/job_controller_test.go b/operator/internal/controller/job_controller_test.go index ffa8540d6..2ba84e212 100644 --- a/operator/internal/controller/job_controller_test.go +++ b/operator/internal/controller/job_controller_test.go @@ -443,6 +443,69 @@ var _ = Describe("JobReconcile", func() { Expect(state[sibling.GetUniqueName()].State).To(Equal(v1alpha1.StateComplete), "the sibling is still promoted") }) + It("does not regress an entry that already advanced past the interrupt", func() { + // The non-interrupt path never reaches the write here — shouldRecordCompletion returns + // false on a mismatched stage. On the interrupt path a skipped sibling makes it return + // true, so this direction is guarded only by the completion check. + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: nodeName}} + sn, err := wrapper.NewSkyhookNodeOnly(node, skyhookName) + Expect(err).ToNot(HaveOccurred()) + Expect(sn.Upsert(pkgRef, image, v1alpha1.StateComplete, v1alpha1.StagePostInterrupt, 0, "")).To(Succeed()) + sibling := v1alpha1.PackageRef{Name: "other", Version: "2.0.0"} + Expect(sn.Upsert(sibling, image, v1alpha1.StateSkipped, v1alpha1.StageInterrupt, 0, "")).To(Succeed()) + + scr := &v1alpha1.NodeWright{ + ObjectMeta: metav1.ObjectMeta{Name: skyhookName}, + Spec: v1alpha1.NodeWrightSpec{Packages: v1alpha1.Packages{ + "tuning": {PackageRef: pkgRef, Image: image}, + "other": {PackageRef: sibling, Image: image}, + }}, + } + job := packageJob(v1alpha1.StageInterrupt, true, trueCondition(batchv1.JobComplete, "")) + r := newReconciler(node, scr, job) + + _, err = r.JobReconcile(ctx, job) + Expect(err).ToNot(HaveOccurred()) + + Expect(getNodeState(r)[pkgRef.GetUniqueName()].Stage).To(Equal(v1alpha1.StagePostInterrupt)) + }) + + It("marks an interrupt completion without panicking when the CR is already gone", func() { + // Same window as the resurrection case: the NodeWright deleted while a completed interrupt + // Job is still unprocessed. GetSkyhook reads a missing CR as (nil, nil). + node := nodeWithState(v1alpha1.StateInProgress, v1alpha1.StageInterrupt) + job := packageJob(v1alpha1.StageInterrupt, true, trueCondition(batchv1.JobComplete, "")) + r := newReconciler(node, job) // no NodeWright seeded + + _, err := r.JobReconcile(ctx, job) + Expect(err).ToNot(HaveOccurred()) + + Expect(getNodeState(r)[pkgRef.GetUniqueName()].State).To(Equal(v1alpha1.StateComplete)) + Expect(getJob(r, job.Name).Annotations).To(HaveKeyWithValue(annotationStateRecorded, annotationValueTrue)) + }) + + It("removes the superseded version's entry on upgrade completion", func() { + // The upgrade branch is the other place that calls RemoveState and then leans on the + // guarded fallback; NodeState is keyed name|version, so only the old key goes. + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: nodeName}} + sn, err := wrapper.NewSkyhookNodeOnly(node, skyhookName) + Expect(err).ToNot(HaveOccurred()) + old := v1alpha1.PackageRef{Name: "tuning", Version: "0.9.0"} + Expect(sn.Upsert(old, image, v1alpha1.StateComplete, v1alpha1.StageConfig, 0, "")).To(Succeed()) + Expect(sn.Upsert(pkgRef, image, v1alpha1.StateInProgress, v1alpha1.StageUpgrade, 0, "")).To(Succeed()) + + job := packageJob(v1alpha1.StageUpgrade, false, trueCondition(batchv1.JobComplete, "")) + r := newReconciler(node, job) + + _, err = r.JobReconcile(ctx, job) + Expect(err).ToNot(HaveOccurred()) + + state := getNodeState(r) + Expect(state).ToNot(HaveKey(old.GetUniqueName())) + Expect(state[pkgRef.GetUniqueName()].State).To(Equal(v1alpha1.StateComplete)) + Expect(state[pkgRef.GetUniqueName()].Stage).To(Equal(v1alpha1.StageUpgrade)) + }) + It("records erroring (state only, no marker) for a stale FailureTarget on an unreachable node", func() { node := nodeWithState(v1alpha1.StateInProgress, v1alpha1.StageConfig) job := packageJob(v1alpha1.StageConfig, false) diff --git a/operator/internal/controller/pod_controller.go b/operator/internal/controller/pod_controller.go index 37095edb2..a40013a12 100644 --- a/operator/internal/controller/pod_controller.go +++ b/operator/internal/controller/pod_controller.go @@ -242,8 +242,7 @@ func shouldRecordPodErroring(skyhookNode wrapper.SkyhookNodeOnly, packagePtr *Pa return false, fmt.Errorf("error reading node state for package %s: %w", packagePtr.GetUniqueName(), err) } - status, present := state[packagePtr.GetUniqueName()] - return present && status.Stage == packagePtr.Stage && status.State != v1alpha1.StateComplete, nil + return entryAwaitsCompletion(state, packagePtr), nil } const ( From 0db7d2894ab863dd15f9f6dc39ced0fad49f103b Mon Sep 17 00:00:00 2001 From: Alex Yuskauskas Date: Tue, 11 Aug 2026 15:17:14 -0700 Subject: [PATCH 3/4] refactor(operator): name the shared guard for its shape, not one caller MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review pointed out that reusing a predicate called entryAwaitsCompletion for the Pod watch's erroring write reads as the wrong thing, which is the same complaint that renamed isParkedJob. It is now entryOpenAtStage — present, at this stage, not complete — which is what both callers actually require, with the doc comment saying so and recording why an already-erroring entry is deliberately still open (a rising restart count must land; an identical write is dropped by the Changed() check). Also tightens the no-regression spec to assert State as well as Stage, and fixes a garbled sentence in the design doc. Signed-off-by: Alex Yuskauskas --- .../2026-07-10-package-execution-as-jobs.md | 2 +- .../internal/controller/job_controller.go | 26 ++++++++++++------- .../controller/job_controller_test.go | 4 ++- .../internal/controller/pod_controller.go | 2 +- 4 files changed, 21 insertions(+), 13 deletions(-) diff --git a/docs/designs/2026-07-10-package-execution-as-jobs.md b/docs/designs/2026-07-10-package-execution-as-jobs.md index 5b17cb597..d52a76f3f 100644 --- a/docs/designs/2026-07-10-package-execution-as-jobs.md +++ b/docs/designs/2026-07-10-package-execution-as-jobs.md @@ -128,7 +128,7 @@ Because completed pods now linger, pod-deletion can no longer be the processed-o The Pod watch is **evidence, not authority**: it may only update an entry that already exists at the pod's stage and is not yet complete. It must never create one. Under `restartPolicy: Never` a failing Job mints a fresh pod per attempt, indefinitely, so an unguarded write becomes a repeating one. If it could create an entry, a node-state reset would be undone by the very Job the reset is meant to clear: the resurrected entry makes the reset invisible to the not-in-node-state check in rule 1 below, so the stale Job is never invalidated, and the existence gate then blocks the package's new stage forever. The same guard also stops a retained failed-attempt archive pod (kept on purpose, below) from regressing a completion the Job path already recorded. -**Neither may the Job path create an entry**, for the same reason, even though it *is* the completion authority. Authority means it decides when a stage is done, not that it may write a package state says is gone. This bites on interrupt Jobs specifically: they are package-agnostic, so a completion is recorded on the strength of a *sibling* still sitting at (interrupt, `skipped`), which says nothing about whether this Job's own package still has an entry. A rerun, reset or uninstall landing in that window would otherwise see the entry return at (interrupt, `complete`), where the rerun predicate keeps the Job and the stage never runs again. The promotion of skipped siblings happens either way; only the self-write is gated. +**Neither may the Job path create an entry**, for the same reason, even though it *is* the completion authority. Authority means it decides when a stage is done, not that it may write an entry node state says is gone. This bites on interrupt Jobs specifically: they are package-agnostic, so a completion is recorded on the strength of a *sibling* still sitting at (interrupt, `skipped`) — which says nothing about whether this Job's own package still has an entry. A rerun, reset or uninstall landing in that window would otherwise see the entry return at (interrupt, `complete`), where the rerun predicate keeps the Job and the stage never runs again. The promotion of skipped siblings happens either way; only the self-write is gated. Recording a completion is two writes to two objects (node state, then the Job marker) and cannot be atomic; a crash between them re-serves the event. The re-processing path is guarded by per-transition postcondition checks so a re-served completion is only marked, not re-applied — detail in [Edge cases](#edge-cases-and-correctness-arguments). This is strictly better than today, which has the same two-write window with no guard. diff --git a/operator/internal/controller/job_controller.go b/operator/internal/controller/job_controller.go index 0e6f54575..5ec28162f 100644 --- a/operator/internal/controller/job_controller.go +++ b/operator/internal/controller/job_controller.go @@ -225,7 +225,7 @@ func (r *JobReconciler) shouldRecordCompletion(job *batchv1.Job, pkg *PackageSky // only while such a package remains, matching what the promotion can actually advance. // // This authorizes the write on a SIBLING's state, so it says nothing about this Job's own - // package — recordJobCompletion gates its self-write on entryAwaitsCompletion separately. + // package — recordJobCompletion gates its self-write on entryOpenAtStage separately. for _, s := range state { if s.Stage == v1alpha1.StageInterrupt && s.State == v1alpha1.StateSkipped { return true, nil @@ -233,14 +233,20 @@ func (r *JobReconciler) shouldRecordCompletion(job *batchv1.Job, pkg *PackageSky } } - return entryAwaitsCompletion(state, pkg), nil + return entryOpenAtStage(state, pkg), nil } -// entryAwaitsCompletion reports whether the package's entry is in the only shape a completion may -// be written onto: present, still at this Job's stage, and not already complete. Absent means -// removed (rerun, reset, uninstall) and writing would resurrect it; a later stage or an existing -// complete would regress or duplicate. -func entryAwaitsCompletion(state v1alpha1.NodeState, pkg *PackageSkyhook) bool { +// entryOpenAtStage reports whether the package's entry is present, still at the stage the caller +// is reporting on, and not already complete — the only shape an executor may write onto at all. +// Absent means removed (rerun, reset, uninstall) and writing would resurrect it; a later stage or +// an existing complete would regress or duplicate. +// +// Named for the shape it tests, not for one caller's use: both the completion write and the Pod +// watch's erroring write require exactly this, and the two must not drift. It deliberately does +// not exclude an already-erroring entry — the Pod watch re-reports erroring so a rising restart +// count still lands, and an otherwise identical write is a no-op the Changed() check drops. +// recordJobErroring adds that exclusion itself; see the note there. +func entryOpenAtStage(state v1alpha1.NodeState, pkg *PackageSkyhook) bool { status, present := state[pkg.GetUniqueName()] return present && status.Stage == pkg.Stage && status.State != v1alpha1.StateComplete } @@ -344,11 +350,11 @@ func (r *JobReconciler) recordJobCompletion(ctx context.Context, job *batchv1.Jo // would never run again. The promotion above still lands either way. // // Both guards are kept deliberately: updated says HandleCompletePod already wrote this - // entry, entryAwaitsCompletion says there is an entry to write onto. They agree today — + // entry, entryOpenAtStage says there is an entry to write onto. They agree today — // every branch that sets updated leaves the entry absent or at another stage — and neither // is safe to drop on the strength of the other. recorded := false - if !updated && entryAwaitsCompletion(state, pkg) { + if !updated && entryOpenAtStage(state, pkg) { if err := skyhookNode.Upsert(pkg.PackageRef, pkg.Image, v1alpha1.StateComplete, pkg.Stage, job.Status.Failed, pkg.ContainerSHA); err != nil { return false, fmt.Errorf("upserting complete state for job %s: %w", job.Name, err) } @@ -568,7 +574,7 @@ func (r *JobReconciler) recordJobErroring(ctx context.Context, job *batchv1.Job, return false, fmt.Errorf("reading node state for job %s: %w", job.Name, err) } - // Deliberately NOT entryAwaitsCompletion: the completion guard excludes an entry that is + // Deliberately NOT entryOpenAtStage: the completion guard excludes an entry that is // already complete, this one excludes an entry that is already erroring, for idempotence // on a re-served terminal event. Same shape, different exclusion — do not unify them. status, present := state[pkg.GetUniqueName()] diff --git a/operator/internal/controller/job_controller_test.go b/operator/internal/controller/job_controller_test.go index 2ba84e212..cb79e0d6f 100644 --- a/operator/internal/controller/job_controller_test.go +++ b/operator/internal/controller/job_controller_test.go @@ -467,7 +467,9 @@ var _ = Describe("JobReconcile", func() { _, err = r.JobReconcile(ctx, job) Expect(err).ToNot(HaveOccurred()) - Expect(getNodeState(r)[pkgRef.GetUniqueName()].Stage).To(Equal(v1alpha1.StagePostInterrupt)) + entry := getNodeState(r)[pkgRef.GetUniqueName()] + Expect(entry.Stage).To(Equal(v1alpha1.StagePostInterrupt)) + Expect(entry.State).To(Equal(v1alpha1.StateComplete)) }) It("marks an interrupt completion without panicking when the CR is already gone", func() { diff --git a/operator/internal/controller/pod_controller.go b/operator/internal/controller/pod_controller.go index a40013a12..ff809b201 100644 --- a/operator/internal/controller/pod_controller.go +++ b/operator/internal/controller/pod_controller.go @@ -242,7 +242,7 @@ func shouldRecordPodErroring(skyhookNode wrapper.SkyhookNodeOnly, packagePtr *Pa return false, fmt.Errorf("error reading node state for package %s: %w", packagePtr.GetUniqueName(), err) } - return entryAwaitsCompletion(state, packagePtr), nil + return entryOpenAtStage(state, packagePtr), nil } const ( From 93caf2ac1f4b86d0b8d9eb38d7d284714f08cb75 Mon Sep 17 00:00:00 2001 From: Alex Yuskauskas Date: Tue, 11 Aug 2026 15:28:04 -0700 Subject: [PATCH 4/4] test(operator): prove the no-regression spec exercised the completion path The spec seeded the entry at (post-interrupt, complete) and asserted it was still there, which a reconcile that did nothing at all would also satisfy. It now asserts the sibling was promoted and the Job carries the state-recorded marker, so the interrupt completion path demonstrably ran while the entry was left alone. Also inserts the relative pronoun the design doc sentence was missing. Signed-off-by: Alex Yuskauskas --- docs/designs/2026-07-10-package-execution-as-jobs.md | 2 +- operator/internal/controller/job_controller_test.go | 8 +++++++- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/docs/designs/2026-07-10-package-execution-as-jobs.md b/docs/designs/2026-07-10-package-execution-as-jobs.md index d52a76f3f..bba2ca770 100644 --- a/docs/designs/2026-07-10-package-execution-as-jobs.md +++ b/docs/designs/2026-07-10-package-execution-as-jobs.md @@ -128,7 +128,7 @@ Because completed pods now linger, pod-deletion can no longer be the processed-o The Pod watch is **evidence, not authority**: it may only update an entry that already exists at the pod's stage and is not yet complete. It must never create one. Under `restartPolicy: Never` a failing Job mints a fresh pod per attempt, indefinitely, so an unguarded write becomes a repeating one. If it could create an entry, a node-state reset would be undone by the very Job the reset is meant to clear: the resurrected entry makes the reset invisible to the not-in-node-state check in rule 1 below, so the stale Job is never invalidated, and the existence gate then blocks the package's new stage forever. The same guard also stops a retained failed-attempt archive pod (kept on purpose, below) from regressing a completion the Job path already recorded. -**Neither may the Job path create an entry**, for the same reason, even though it *is* the completion authority. Authority means it decides when a stage is done, not that it may write an entry node state says is gone. This bites on interrupt Jobs specifically: they are package-agnostic, so a completion is recorded on the strength of a *sibling* still sitting at (interrupt, `skipped`) — which says nothing about whether this Job's own package still has an entry. A rerun, reset or uninstall landing in that window would otherwise see the entry return at (interrupt, `complete`), where the rerun predicate keeps the Job and the stage never runs again. The promotion of skipped siblings happens either way; only the self-write is gated. +**Neither may the Job path create an entry**, for the same reason, even though it *is* the completion authority. Authority means it decides when a stage is done, not that it may write a package entry that node state says is gone. This bites on interrupt Jobs specifically: they are package-agnostic, so a completion is recorded on the strength of a *sibling* still sitting at (interrupt, `skipped`) — which says nothing about whether this Job's own package still has an entry. A rerun, reset or uninstall landing in that window would otherwise see the entry return at (interrupt, `complete`), where the rerun predicate keeps the Job and the stage never runs again. The promotion of skipped siblings happens either way; only the self-write is gated. Recording a completion is two writes to two objects (node state, then the Job marker) and cannot be atomic; a crash between them re-serves the event. The re-processing path is guarded by per-transition postcondition checks so a re-served completion is only marked, not re-applied — detail in [Edge cases](#edge-cases-and-correctness-arguments). This is strictly better than today, which has the same two-write window with no guard. diff --git a/operator/internal/controller/job_controller_test.go b/operator/internal/controller/job_controller_test.go index cb79e0d6f..f634c0a68 100644 --- a/operator/internal/controller/job_controller_test.go +++ b/operator/internal/controller/job_controller_test.go @@ -467,7 +467,13 @@ var _ = Describe("JobReconcile", func() { _, err = r.JobReconcile(ctx, job) Expect(err).ToNot(HaveOccurred()) - entry := getNodeState(r)[pkgRef.GetUniqueName()] + state := getNodeState(r) + // The sibling's promotion is what proves the interrupt completion path actually ran: + // without it this spec would pass on a reconcile that did nothing at all. + Expect(state[sibling.GetUniqueName()].State).To(Equal(v1alpha1.StateComplete)) + Expect(getJob(r, job.Name).Annotations).To(HaveKeyWithValue(annotationStateRecorded, annotationValueTrue)) + + entry := state[pkgRef.GetUniqueName()] Expect(entry.Stage).To(Equal(v1alpha1.StagePostInterrupt)) Expect(entry.State).To(Equal(v1alpha1.StateComplete)) })