diff --git a/internal/operator/packagesource_reconciler.go b/internal/operator/packagesource_reconciler.go index e79370bdd..8a6baacdb 100644 --- a/internal/operator/packagesource_reconciler.go +++ b/internal/operator/packagesource_reconciler.go @@ -19,7 +19,9 @@ package operator import ( "context" "fmt" + "strconv" "strings" + "time" cozyv1alpha1 "github.com/cozystack/cozystack/api/v1alpha1" sourcewatcherv1beta1 "github.com/fluxcd/source-watcher/api/v2/v1beta1" @@ -34,6 +36,62 @@ import ( "sigs.k8s.io/controller-runtime/pkg/log" ) +// Constants tuning the workaround for fluxcd/pkg#934 (patch.Helper split-write +// race in source-watcher v2.1.0). See the block comment on +// maybeRecoverArtifactGenerator for the mechanism. +// +// Version-skew note. Comments in this file reference upstream source-watcher +// internals by exact file:line (e.g. `drift.go:52`, `controller.go:164/254`). +// Those anchors are for the DEPLOYED source-watcher binary, currently pinned +// to v2.1.0 at `internal/fluxinstall/manifests/fluxcd.yaml`. The Go module +// dependency in `go.mod` (`github.com/fluxcd/source-watcher/api/v2 v2.0.3`) +// only pulls the CRD types package — behavioural code lives in the deployed +// binary. When the flux distribution bumps source-watcher, re-verify the +// line anchors AND the watch predicate (`ReconcileRequestedPredicate`) and +// drift-detection heuristics against the new version — those are the +// load-bearing upstream assumptions this workaround stands on. +// +// The schedule is deliberately conservative: +// +// - `stuckGracePeriod = 30s` — source-watcher normally settles an +// ArtifactGenerator in under 10s once it takes the drifted branch, so a +// 30s Unknown window comfortably distinguishes a real stall from a +// legitimate in-flight rebuild. +// - `initialBackoff = 30s`, `maxBackoff = 4m`, `maxRecoveryAttempts = 5` — +// exponential 30s + 60s + 2m + 4m + 4m gives ~11.5m of total budget +// before we give up, safely inside the 15m upstream HelmRelease install +// timeout so a real stall surfaces as SourceWatcherStalled rather than +// silently timing out an install. +// +// HA note: the retry driver stores its state (attempt counter, last-force +// timestamp) as annotations on the ArtifactGenerator itself, so a single +// replica converges deterministically. Running cozystack-operator with +// multiple replicas WITHOUT leader election will let both replicas race on +// annotation writes, corrupting the counter and the elapsed-time math. Deploy +// this operator with `--leader-elect=true` (or replicas=1). +const ( + stuckGracePeriod = 30 * time.Second + initialBackoff = 30 * time.Second + maxBackoff = 4 * time.Minute + maxRecoveryAttempts = 5 + + annotationFluxRequestedAt = "reconcile.fluxcd.io/requestedAt" + annotationRecoveryAttempts = "cozystack.io/source-watcher-recovery-attempts" + annotationLastRecoveryAt = "cozystack.io/source-watcher-last-recovery-at" + + reasonAwaitingRecovery = "AwaitingSourceWatcherRecovery" + reasonSourceWatcherBad = "SourceWatcherStalled" + // reasonRecoveryForced is the Reason we stamp on the AG's Ready condition + // when forcing drift. Kept as a constant because updateStatus needs to + // recognise its own writes reflecting back through the Owns() watch and + // skip the not-stuck / clearRecoveryTracking path for them (see B2 in + // lexfrei's review on PR #3182). + reasonRecoveryForced = "SourceWatcherRecoveryForced" +) + +// nowFunc is overridable in tests; production code always uses time.Now. +var nowFunc = time.Now + // PackageSourceReconciler reconciles PackageSource resources type PackageSourceReconciler struct { client.Client @@ -57,19 +115,32 @@ func (r *PackageSourceReconciler) Reconcile(ctx context.Context, req ctrl.Reques return ctrl.Result{}, err } + // Skip work on objects that are already being torn down. Without this + // guard the retry driver could keep writing annotations to an + // ArtifactGenerator whose owning PackageSource is about to be garbage + // collected, and Status().Update below would leave a misleading + // "AwaitingSourceWatcherRecovery" condition on a doomed object. + // ownerReference cascade takes care of the actual AG cleanup. + if !packageSource.DeletionTimestamp.IsZero() { + return ctrl.Result{}, nil + } + // Generate ArtifactGenerator for package source if err := r.reconcileArtifactGenerators(ctx, packageSource); err != nil { logger.Error(err, "failed to reconcile ArtifactGenerator") return ctrl.Result{}, err } - // Update PackageSource status (variants and conditions from ArtifactGenerator) - if err := r.updateStatus(ctx, packageSource); err != nil { - logger.Error(err, "failed to update status") - // Don't return error, status update is not critical - } - - return ctrl.Result{}, nil + // Update PackageSource status (variants and conditions from ArtifactGenerator). + // The status update may schedule a follow-up reconcile via RequeueAfter when + // it detects a source-watcher status-patch stall and needs to wait for the + // next backoff window; that Result is honoured on the way out. `now` is + // threaded down as a single sample so the whole reconcile agrees on one + // wall-clock reading. Errors from updateStatus are propagated so + // controller-runtime's exponential backoff can retry transient failures — + // swallowing them would drop force-drift or backoff-schedule writes on the + // floor and leave the retry driver in a stale state. + return r.updateStatus(ctx, packageSource, nowFunc()) } // reconcileArtifactGenerators generates a single ArtifactGenerator for the package source @@ -327,8 +398,14 @@ func (r *PackageSourceReconciler) createOrUpdate(ctx context.Context, obj client return r.Patch(ctx, obj, client.Apply, client.FieldOwner("cozystack-packagesource-controller")) } -// updateStatus updates PackageSource status (variants and conditions from ArtifactGenerator) -func (r *PackageSourceReconciler) updateStatus(ctx context.Context, packageSource *cozyv1alpha1.PackageSource) error { +// updateStatus updates PackageSource status (variants and conditions from +// ArtifactGenerator). It may return a Result with RequeueAfter set when the +// ArtifactGenerator's upstream Ready condition is stuck and the reconciler is +// driving source-watcher through a bounded requeue schedule; see +// maybeRecoverArtifactGenerator for the strategy. `now` is passed by the +// caller so every time-sensitive check inside this reconcile agrees on the +// same wall-clock reading. +func (r *PackageSourceReconciler) updateStatus(ctx context.Context, packageSource *cozyv1alpha1.PackageSource, now time.Time) (ctrl.Result, error) { logger := log.FromContext(ctx) // Update variants in status from spec @@ -342,12 +419,13 @@ func (r *PackageSourceReconciler) updateStatus(ctx context.Context, packageSourc if packageSource.Spec.SourceRef == nil { // Set status to unknown if SourceRef is not set meta.SetStatusCondition(&packageSource.Status.Conditions, metav1.Condition{ - Type: "Ready", - Status: metav1.ConditionUnknown, - Reason: "SourceRefNotSet", - Message: "SourceRef is not configured", + Type: "Ready", + Status: metav1.ConditionUnknown, + Reason: "SourceRefNotSet", + Message: "SourceRef is not configured", + ObservedGeneration: packageSource.Generation, }) - return r.Status().Update(ctx, packageSource) + return ctrl.Result{}, r.Status().Update(ctx, packageSource) } // Get ArtifactGenerator @@ -361,27 +439,102 @@ func (r *PackageSourceReconciler) updateStatus(ctx context.Context, packageSourc if apierrors.IsNotFound(err) { // ArtifactGenerator not found, set status to unknown meta.SetStatusCondition(&packageSource.Status.Conditions, metav1.Condition{ - Type: "Ready", - Status: metav1.ConditionUnknown, - Reason: "ArtifactGeneratorNotFound", - Message: "ArtifactGenerator not found", + Type: "Ready", + Status: metav1.ConditionUnknown, + Reason: "ArtifactGeneratorNotFound", + Message: "ArtifactGenerator not found", + ObservedGeneration: packageSource.Generation, }) - return r.Status().Update(ctx, packageSource) + return ctrl.Result{}, r.Status().Update(ctx, packageSource) } - return fmt.Errorf("failed to get ArtifactGenerator: %w", err) + return ctrl.Result{}, fmt.Errorf("failed to get ArtifactGenerator: %w", err) } // Find Ready condition in ArtifactGenerator readyCondition := meta.FindStatusCondition(ag.Status.Conditions, "Ready") + + // Route to the recovery state machine if any of the following holds: + // + // (a) the AG's Ready condition carries our own reasonRecoveryForced + // marker (Owns(&ArtifactGenerator{}) re-firing on the writes + // from forceArtifactGeneratorDrift, or a still-un-reacted force + // from a previous reconcile); + // (b) recovery is already in progress (attempts > 0) and source- + // watcher has moved off our marker into Progressing + // (Ready=Unknown on the current generation) — the rebuild write + // cycle after taking the drifted branch; + // (c) the AG has been sitting in the fluxcd/pkg#934 stall signature + // (Inventory populated + digest set + Ready=Unknown / absent past + // grace) — the classic first-time stuck-detection path. + // + // All three funnel into maybeRecoverArtifactGenerator so its + // decideRecovery state machine is the single arbiter of "wait vs + // re-force vs give up". Splitting (a) and (b) into a separate + // wait-forever helper would trap us in a tight requeue loop if + // source-watcher never responds — the state machine's Wait branch is + // bounded by the same backoff schedule that drives Force, so + // exhaustion always surfaces SourceWatcherStalled honestly. + // + // Signals for (a) and (b) are exact: reasonRecoveryForced is only ever + // written by forceArtifactGeneratorDrift, so any Ready condition + // carrying it is necessarily our own reflection; the attempts counter + // is only ever set by that same function. Neither is observable from + // outside this reconciler. + // + // Asymmetric ObservedGeneration handling — deliberate, not oversight. + // `isProgressingInRecovery` requires `ObservedGeneration == Generation` + // so a Progressing condition left over from a previous spec cycle does + // NOT match the pattern (the AG is being rebuilt against fresh spec — + // no need to hold tracking around a stale Progressing). `isOwnMarker` + // does NOT check ObservedGeneration on purpose: if we wrote our + // marker at gen=N and the user then edits PS spec so AG.generation=N+1, + // our marker sits with `condition.ObservedGeneration=N`. That marker + // is stale, but we must still route through the state machine on it + // rather than falling into the not-stuck / clearRecoveryTracking + // branch — otherwise the copy-through path would leak our synthetic + // Ready=False/SourceWatcherRecoveryForced onto the PackageSource + // (exactly the B2 bug this whole gate exists to prevent). + // source-watcher's GenerationChangedPredicate fires on the AG's own + // generation bump, so the rebuild is triggered regardless of our + // marker's staleness — one wasted attempt in the worst case, and the + // state machine's Wait branch usually absorbs it without incrementing. + attempts, _ := readRecoveryTracking(ag) + isOwnMarker := readyCondition != nil && readyCondition.Reason == reasonRecoveryForced + isProgressingInRecovery := attempts > 0 && readyCondition != nil && + readyCondition.Status == metav1.ConditionUnknown && + readyCondition.ObservedGeneration == ag.Generation + if isOwnMarker || isProgressingInRecovery || artifactGeneratorStuck(ag, readyCondition, now) { + logger.V(1).Info("routing to recovery state machine", + "packageSource", packageSource.Name, + "artifactGenerator", ag.Name, + "attempts", attempts, + "ownMarker", isOwnMarker, + "progressingInRecovery", isProgressingInRecovery) + return r.maybeRecoverArtifactGenerator(ctx, packageSource, ag, readyCondition, now) + } + + // AG is not stuck and not mid-recovery — clear any recovery-tracking + // annotations left behind by a previous stuck cycle, then either + // surface the missing-Ready case or copy the real condition through. + if err := r.clearRecoveryTracking(ctx, ag); err != nil { + logger.Error(err, "failed to clear recovery tracking annotations", "artifactGenerator", ag.Name) + // Non-fatal: annotations are best-effort bookkeeping. + } + if readyCondition == nil { // No Ready condition in ArtifactGenerator, set status to unknown meta.SetStatusCondition(&packageSource.Status.Conditions, metav1.Condition{ - Type: "Ready", - Status: metav1.ConditionUnknown, - Reason: "ArtifactGeneratorNotReady", - Message: "ArtifactGenerator Ready condition not found", + Type: "Ready", + Status: metav1.ConditionUnknown, + Reason: "ArtifactGeneratorNotReady", + Message: "ArtifactGenerator Ready condition not found", + ObservedGeneration: packageSource.Generation, }) - return r.Status().Update(ctx, packageSource) + // Fresh AG with no Ready yet — if source-watcher doesn't act within + // the grace period, artifactGeneratorStuck will fire on the next + // reconcile. Schedule that reconcile explicitly so we don't sit + // waiting for an AG-status change that may never come. + return ctrl.Result{RequeueAfter: agFollowUpDelay(ag.CreationTimestamp.Time, now)}, r.Status().Update(ctx, packageSource) } // Copy Ready condition from ArtifactGenerator to PackageSource @@ -399,7 +552,463 @@ func (r *PackageSourceReconciler) updateStatus(ctx context.Context, packageSourc "status", readyCondition.Status, "reason", readyCondition.Reason) - return r.Status().Update(ctx, packageSource) + // If we just copied Ready=Unknown across, the artifactGeneratorStuck + // predicate will fire once the grace period elapses. Without an + // explicit RequeueAfter here, this reconciler would wait for a fresh + // AG-status change to re-fire via the Owns() watch — and if + // source-watcher's next write is the one lost to the fluxcd/pkg#934 + // split-patch race (i.e., exactly the scenario this workaround exists + // for), no such change will land and we would sit dormant until the + // 10h informer resync. Schedule a follow-up reconcile at grace-period + // expiry so the recovery driver can take over. On Ready=True/False, + // the AG has resolved and no explicit follow-up is needed. + result := ctrl.Result{} + if readyCondition.Status == metav1.ConditionUnknown { + result.RequeueAfter = agFollowUpDelay(readyCondition.LastTransitionTime.Time, now) + } + return result, r.Status().Update(ctx, packageSource) +} + +// agFollowUpDelay returns how long to wait before re-reconciling an AG whose +// Ready=Unknown state is still within the grace period. Anchor is the last +// state transition (or AG creation, when no Ready condition exists yet). The +// result is at least one second so we never schedule a zero-delay requeue +// tight loop when the grace period has just elapsed. +func agFollowUpDelay(transitionOrCreation time.Time, now time.Time) time.Duration { + remaining := transitionOrCreation.Add(stuckGracePeriod).Sub(now) + if remaining < time.Second { + return time.Second + } + return remaining +} + +// maybeRecoverArtifactGenerator advances the bounded-recovery schedule for an +// AG whose Ready condition is stuck in the fluxcd/pkg#934 window (Inventory +// and ObservedSourcesDigest persisted on the current spec generation, Ready +// condition write lost to the split patch). It forces source-watcher onto its +// drifted-branch reconcile by patching Ready=False on the AG's status +// subresource with exponential backoff and — after maxRecoveryAttempts +// fruitless attempts — surfaces the failure as PackageSource.Ready=False with +// reason SourceWatcherStalled so an operator can intervene. +// +// Why forcing Ready=False AND bumping requestedAt together works: the two +// signals are complementary — neither alone is enough. +// +// - The status write (Ready=False, reason=SourceWatcherRecoveryForced) is +// what source-watcher v2.1.0's detectDrift +// (internal/controller/artifactgenerator_drift.go:52) treats as drift = +// "NotReady", promoting the reconcile past the no-drift early-return at +// controller.go:164 that traps a lost-Ready-condition state. +// - The annotation write (reconcile.fluxcd.io/requestedAt = now) is what +// ReconcileRequestedPredicate on source-watcher's ArtifactGenerator +// watch keys on to enqueue that reconcile in the first place. A +// status-only patch changes neither the AG's generation nor the +// requestedAt annotation, so it never triggers a watch event — the AG +// would sit until the periodic self-requeue (hardcoded to 1h), +// well outside the 15m HR install timeout. +// +// On successful rebuild source-watcher writes Ready=True at +// controller.go:254; on a genuine rebuild failure it writes Ready=False +// with a real reason. Either way an honest condition from upstream +// replaces our synthetic False, and our Owns(&ArtifactGenerator{}) watch +// fires this reconciler to copy the resolved condition onto the +// PackageSource. +// +// The recovery state (attempt count + last-force timestamp) lives on the +// AG itself as annotations so it survives operator restarts. The counter +// is NOT reset on a fresh Ready.LastTransitionTime — source-watcher's +// Progressing write during rebuild always advances that timestamp, and a +// reset there would make the SourceWatcherStalled give-up unreachable +// under the very race this driver targets. The canonical reset is the +// natural clearRecoveryTracking on updateStatus's not-stuck path, once +// source-watcher lands a definitive Ready=True or Ready=False with a +// real upstream reason. +// +// TODO(remove once fluxcd/pkg#934 lands and is rolled out): once source-watcher +// consumes a patch.Helper that either serialises or transactionally combines +// the .status / .status.conditions writes, this whole recovery driver can be +// deleted and updateStatus can copy the AG's Ready condition through +// unconditionally. +func (r *PackageSourceReconciler) maybeRecoverArtifactGenerator(ctx context.Context, packageSource *cozyv1alpha1.PackageSource, ag *sourcewatcherv1beta1.ArtifactGenerator, ready *metav1.Condition, now time.Time) (ctrl.Result, error) { + logger := log.FromContext(ctx) + + // The attempt counter is intentionally NOT reset here on a fresh + // Ready.LastTransitionTime. source-watcher's response to a forced drift + // includes a Ready=Unknown (Progressing) write before the rebuild + // completes; that transition's timestamp is always newer than + // lastRecoveryAt, and resetting on it would make the give-up budget + // unreachable under the very race this driver targets — force → Progressing + // → reset → force → Progressing → reset → forever. Instead, the unified + // Owns()-re-entrancy gate in updateStatus routes those Progressing + // transitions back through this same maybeRecoverArtifactGenerator so + // decideRecovery holds tracking through Wait until either source-watcher + // resolves the AG or the attempt budget exhausts. The natural + // clearRecoveryTracking on the not-stuck path (Ready=True from + // source-watcher, or Ready=False with an upstream reason) is the + // canonical reset. The trade-off — an operator restart between a prior + // give-up and a genuine fresh stall would preserve the stale counter — + // is contained by the leader-election requirement documented on the + // constants block. + attempts, lastRecoveryAt := readRecoveryTracking(ag) + decision := decideRecovery(attempts, lastRecoveryAt, now) + + switch decision.action { + case recoveryActionGiveUp: + message := fmt.Sprintf( + "ArtifactGenerator %s/%s has been stuck with a lost Ready condition through %d force-drift attempts; "+ + "source-watcher is not recovering. See https://github.com/fluxcd/pkg/issues/934. "+ + "An operator must restart source-watcher or manually inspect the ArtifactGenerator.", + ag.Namespace, ag.Name, maxRecoveryAttempts, + ) + meta.SetStatusCondition(&packageSource.Status.Conditions, metav1.Condition{ + Type: "Ready", + Status: metav1.ConditionFalse, + Reason: reasonSourceWatcherBad, + Message: message, + ObservedGeneration: packageSource.Generation, + }) + logger.Info("source-watcher stalled after bounded force-drift attempts; surfacing PackageSource Ready=False", + "packageSource", packageSource.Name, "artifactGenerator", ag.Name, "attempts", attempts) + return ctrl.Result{}, r.Status().Update(ctx, packageSource) + + case recoveryActionWait: + meta.SetStatusCondition(&packageSource.Status.Conditions, metav1.Condition{ + Type: "Ready", + Status: metav1.ConditionUnknown, + Reason: reasonAwaitingRecovery, + Message: fmt.Sprintf( + "ArtifactGenerator Ready condition lost to fluxcd/pkg#934 patch.Helper race; "+ + "force-drift %d/%d issued, waiting for source-watcher to rebuild.", + attempts, maxRecoveryAttempts, + ), + ObservedGeneration: packageSource.Generation, + }) + if err := r.Status().Update(ctx, packageSource); err != nil { + return ctrl.Result{}, err + } + return ctrl.Result{RequeueAfter: decision.wait}, nil + + case recoveryActionForce: + nextAttempt := attempts + 1 + if err := r.forceArtifactGeneratorDrift(ctx, ag, now, nextAttempt); err != nil { + return ctrl.Result{}, fmt.Errorf("failed to force ArtifactGenerator drift: %w", err) + } + meta.SetStatusCondition(&packageSource.Status.Conditions, metav1.Condition{ + Type: "Ready", + Status: metav1.ConditionUnknown, + Reason: reasonAwaitingRecovery, + Message: fmt.Sprintf( + "ArtifactGenerator Ready condition lost to fluxcd/pkg#934 patch.Helper race; "+ + "forced drift on AG.status.conditions[Ready]=False (attempt %d/%d) so source-watcher rebuilds.", + nextAttempt, maxRecoveryAttempts, + ), + ObservedGeneration: packageSource.Generation, + }) + if err := r.Status().Update(ctx, packageSource); err != nil { + return ctrl.Result{}, err + } + logger.Info("forced source-watcher drift via AG status Ready=False patch", + "packageSource", packageSource.Name, "artifactGenerator", ag.Name, "attempt", nextAttempt) + return ctrl.Result{RequeueAfter: backoffFor(nextAttempt)}, nil + + default: + // Force a loud failure if a new recoveryAction is ever added without + // a matching case here — a silent return would leave the PackageSource + // Ready condition frozen at whatever value it had before this call. + // controller-runtime v0.15+ recovers panics inside Reconcile by + // default (Options.RecoverPanic), so this manifests as a requeue with + // a logged stack trace rather than a crash-loop; if that default is + // ever flipped off in SetupWithManager, this needs to become an + // error return instead. + panic(fmt.Sprintf("unhandled recoveryAction %d", decision.action)) + } +} + +// artifactGeneratorStuck detects the fluxcd/pkg#934 stall signature: artifacts +// are demonstrably produced (Inventory populated and ObservedSourcesDigest set) +// on the current spec generation, yet the Ready condition is either missing or +// has been sitting in Unknown longer than stuckGracePeriod. The grace period +// keeps this predicate off the fast path during a normal in-flight rebuild; +// only genuinely quiescent AGs land here. +// +// Ready=True and Ready=False are BOTH pass-through (return false) — the retry +// driver only runs on the specific stuck-Unknown case. A real regeneration +// failure surfaces as Ready=False and is copied through unchanged. +func artifactGeneratorStuck(ag *sourcewatcherv1beta1.ArtifactGenerator, ready *metav1.Condition, now time.Time) bool { + if len(ag.Status.Inventory) == 0 { + return false + } + if ag.Status.ObservedSourcesDigest == "" { + return false + } + if ready == nil { + // Ready condition entirely absent: this is the half-persisted case the + // PR was written for. Wait out the grace period from AG creation before + // intervening so we don't fight a fresh-install AG that just hasn't + // been touched yet. + return ag.CreationTimestamp.Time.Add(stuckGracePeriod).Before(now) + } + if ready.Status != metav1.ConditionUnknown { + return false + } + if ready.ObservedGeneration != ag.Generation { + return false + } + // Only intervene if Unknown has held for the grace period; otherwise source + // -watcher is legitimately mid-rebuild and will settle on its own. + return ready.LastTransitionTime.Time.Add(stuckGracePeriod).Before(now) +} + +// recoveryAction enumerates what maybeRecoverArtifactGenerator should do given +// the current retry state. +type recoveryAction int + +const ( + recoveryActionForce recoveryAction = iota // enough time elapsed — issue a fresh force-drift status patch + recoveryActionWait // in backoff window — schedule a follow-up reconcile at wait + recoveryActionGiveUp // exceeded maxRecoveryAttempts — surface as Ready=False +) + +type recoveryDecision struct { + action recoveryAction + wait time.Duration +} + +// decideRecovery is the pure decision function driving maybeRecoverArtifactGenerator. +// Split out so it can be unit-tested without a cluster. +// +// The exhaustion branch (`attempts >= maxRecoveryAttempts`) also respects the +// final backoff window before switching to GiveUp. Without that gate, the +// Owns()-watch re-fire from the N-th force-drift's own writes triggers a +// reconcile milliseconds later, `decideRecovery` sees `attempts == max` and +// returns GiveUp immediately — preempting the response window the final +// force was supposed to grant source-watcher and collapsing the documented +// budget from ~11.5m (30s+60s+2m+4m+4m) down to ~7.5m (30s+60s+2m+4m). The +// wait branch here holds until the full final backoff has elapsed, so the +// last nudge actually gets its chance to succeed (Owns → Ready=True copy- +// through) before the driver surfaces SourceWatcherStalled. +func decideRecovery(attempts int, lastRecoveryAt time.Time, now time.Time) recoveryDecision { + if attempts >= maxRecoveryAttempts { + elapsed := now.Sub(lastRecoveryAt) + needed := backoffFor(maxRecoveryAttempts) + if elapsed < needed { + return recoveryDecision{action: recoveryActionWait, wait: needed - elapsed} + } + return recoveryDecision{action: recoveryActionGiveUp} + } + if attempts == 0 { + return recoveryDecision{action: recoveryActionForce} + } + elapsed := now.Sub(lastRecoveryAt) + needed := backoffFor(attempts) + if elapsed >= needed { + return recoveryDecision{action: recoveryActionForce} + } + return recoveryDecision{action: recoveryActionWait, wait: needed - elapsed} +} + +// backoffFor returns the backoff duration to wait AFTER the Nth force before +// the (N+1)th. Attempts are 1-indexed. Exponential up to maxBackoff. +func backoffFor(attempt int) time.Duration { + if attempt < 1 { + attempt = 1 + } + d := initialBackoff + for i := 1; i < attempt; i++ { + d *= 2 + if d >= maxBackoff { + return maxBackoff + } + } + return d +} + +// forceArtifactGeneratorDrift writes Ready=False on the AG's status subresource +// to trigger source-watcher's drifted-branch reconcile (see the block comment +// on maybeRecoverArtifactGenerator for why this is what actually recovers a +// stuck AG). Then updates our tracking annotations so the retry loop can back +// off. +// +// Two subresources are patched — status for the Ready condition, metadata for +// the annotations — because status is a separate endpoint and cannot be +// combined with metadata in a single PATCH. Order is: status first, then +// metadata. If the status patch fails, we return the error and leave both the +// caller's `ag` and the apiserver untouched. If the status patch succeeds but +// the annotations patch fails, we still return an error — the retry loop will +// pick up on the next reconcile and, because source-watcher already got the +// Ready=False signal from the successful status write, the extra force-drift +// on that retry is benign (source-watcher is already mid-rebuild). +// +// JSON merge patch semantics — honest trade-off. `client.MergeFrom` produces +// a JSON merge patch, which replaces arrays in full per RFC 7396 (custom +// resources like ArtifactGenerator do not carry the `patchStrategy: merge` +// / `patchMergeKey` markers that would enable strategic-merge-patch +// per-element merging on Kubernetes-native APIs). Consequences: +// +// - If source-watcher writes a fresh Reconciling condition between our +// Get (in updateStatus, several stack frames up) and this status Patch, +// our patch REPLACES the entire status.conditions array with the array +// we hold locally — Reconciling gets transiently wiped. +// source-watcher restores it on its next reconcile (which our +// requestedAt bump in step 2 is about to trigger), so the wipe is +// short-lived. Downstream tooling watching for Reconciling (Grafana +// panels, alertmanager rules) may still observe the momentary absence. +// - Similarly, any other condition source-watcher added concurrently +// (e.g. a custom Stalled marker in future versions) would be wiped. +// Currently source-watcher v2.1.0 only writes Ready and Reconciling, +// so the blast radius is Reconciling only. +// +// A full fix would require server-side apply with per-condition field +// ownership, at the cost of introducing SSA into the operator's write +// path. Not doing it here because the transient wipe is acceptable in +// this workaround's scope; if a future refactor moves the operator to +// SSA globally, this call should migrate too. +// +// On success, `ag.Status.Conditions` and `ag.Annotations` reflect the +// persisted state so the caller can re-read the same pointer. On any Patch +// failure the corresponding field is rolled back to its pre-call value. +func (r *PackageSourceReconciler) forceArtifactGeneratorDrift(ctx context.Context, ag *sourcewatcherv1beta1.ArtifactGenerator, now time.Time, nextAttempt int) error { + logger := log.FromContext(ctx) + // Step 1: patch AG.status.conditions[Ready]=False. This is the signal + // source-watcher's detectDrift picks up as `NotReady` drift. + statusBase := ag.DeepCopy() + priorConditions := cloneConditions(ag.Status.Conditions) + meta.SetStatusCondition(&ag.Status.Conditions, metav1.Condition{ + Type: "Ready", + Status: metav1.ConditionFalse, + Reason: reasonRecoveryForced, + Message: "cozystack-operator forced drift after fluxcd/pkg#934 stall; source-watcher will rebuild", + ObservedGeneration: ag.Generation, + }) + if err := r.Status().Patch(ctx, ag, client.MergeFrom(statusBase)); err != nil { + ag.Status.Conditions = priorConditions + return fmt.Errorf("status patch to force drift: %w", err) + } + + // Step 2: bump `reconcile.fluxcd.io/requestedAt` alongside the tracking + // annotations. Both parts are load-bearing: + // + // - source-watcher v2.1.0 registers the ArtifactGenerator watch with + // `predicate.Or(GenerationChangedPredicate, ReconcileRequestedPredicate)` + // (source-watcher `artifactgenerator_manager.go` in the v2.1.0 pinned + // distribution). A status-only patch changes neither the generation + // nor the requestedAt annotation, so it does NOT enqueue a reconcile. + // The only fallback is the AG's periodic self-requeue, hardcoded to + // `time.Hour` — well outside the 15m HR install timeout this driver + // exists to beat. + // - The Ready=False write from step 1 is what source-watcher's + // `detectDrift` keys on once it does reconcile (drift.go:52, + // `IsFalse(Ready)` → `NotReady` drift → drifted branch → MarkTrue). + // + // Together: the annotation triggers the reconcile, the status condition + // makes that reconcile take the drifted branch and rebuild. Either alone + // is a no-op. + // + // If this metadata patch fails, the status write from step 1 has already + // taken effect, but source-watcher will not have been enqueued yet. The + // retry loop notices on the next reconcile and re-issues both parts — + // benign, source-watcher is not doing anything with the stale Ready=False + // alone. + metadataBase := ag.DeepCopy() + priorAnnotations := cloneAnnotations(ag.Annotations) + if ag.Annotations == nil { + ag.Annotations = map[string]string{} + } + nowStr := now.UTC().Format(time.RFC3339Nano) + ag.Annotations[annotationFluxRequestedAt] = nowStr + ag.Annotations[annotationRecoveryAttempts] = strconv.Itoa(nextAttempt) + ag.Annotations[annotationLastRecoveryAt] = nowStr + if err := r.Patch(ctx, ag, client.MergeFrom(metadataBase)); err != nil { + ag.Annotations = priorAnnotations + // The status patch already landed, so source-watcher sees our + // Ready=False. If this metadata patch keeps failing on retry, + // the recovery tracking annotations never persist to the + // apiserver — every subsequent reconcile reads attempts=0 and + // re-forces from scratch instead of accumulating toward + // SourceWatcherStalled. Log loudly so operators can spot a + // metadata-write-only apiserver malfunction. + logger.Error(err, "recovery tracking metadata patch failed after successful status patch — "+ + "source-watcher has been nudged but the attempt counter did not advance; "+ + "if this repeats persistently, force-drift will loop without ever giving up", + "artifactGenerator", ag.Name, + "intendedAttempts", nextAttempt) + return fmt.Errorf("metadata patch to update recovery tracking: %w", err) + } + return nil +} + +// clearRecoveryTracking removes our bookkeeping annotations once the AG is +// healthy again. +// +// Same success/failure contract as forceArtifactGeneratorDrift: on success +// `ag.Annotations` matches the persisted state; on Patch failure the caller's +// annotations are rolled back to their pre-call values. +func (r *PackageSourceReconciler) clearRecoveryTracking(ctx context.Context, ag *sourcewatcherv1beta1.ArtifactGenerator) error { + if ag.Annotations == nil { + return nil + } + _, hasAttempts := ag.Annotations[annotationRecoveryAttempts] + _, hasLast := ag.Annotations[annotationLastRecoveryAt] + if !hasAttempts && !hasLast { + return nil + } + patchBase := ag.DeepCopy() + priorAnnotations := cloneAnnotations(ag.Annotations) + delete(ag.Annotations, annotationRecoveryAttempts) + delete(ag.Annotations, annotationLastRecoveryAt) + if err := r.Patch(ctx, ag, client.MergeFrom(patchBase)); err != nil { + ag.Annotations = priorAnnotations + return err + } + return nil +} + +// cloneConditions returns a shallow copy of a conditions slice suitable for +// rollback on Patch failure. Each metav1.Condition holds only value-typed +// fields (strings, ints, Time), so a shallow slice copy is sufficient. +func cloneConditions(src []metav1.Condition) []metav1.Condition { + if src == nil { + return nil + } + dst := make([]metav1.Condition, len(src)) + copy(dst, src) + return dst +} + +// cloneAnnotations returns a shallow copy suitable for rollback on Patch +// failure. A shallow copy is enough because annotation values are plain +// strings, which are immutable in Go; there is no shared mutable state to +// deep-copy. Nil in → nil out (preserved as a distinct sentinel from an empty +// map so the caller sees exactly the same state it started with). +func cloneAnnotations(src map[string]string) map[string]string { + if src == nil { + return nil + } + dst := make(map[string]string, len(src)) + for k, v := range src { + dst[k] = v + } + return dst +} + +// readRecoveryTracking pulls the retry-attempt counter and last-force timestamp +// off the AG. Missing/malformed annotations are treated as "no prior attempts" +// so a corrupted counter can't wedge the retry loop. +func readRecoveryTracking(ag *sourcewatcherv1beta1.ArtifactGenerator) (attempts int, lastRecoveryAt time.Time) { + if ag.Annotations == nil { + return 0, time.Time{} + } + if raw, ok := ag.Annotations[annotationRecoveryAttempts]; ok { + if parsed, err := strconv.Atoi(raw); err == nil && parsed >= 0 { + attempts = parsed + } + } + if raw, ok := ag.Annotations[annotationLastRecoveryAt]; ok { + if parsed, err := time.Parse(time.RFC3339Nano, raw); err == nil { + lastRecoveryAt = parsed + } + } + return attempts, lastRecoveryAt } // SetupWithManager sets up the controller with the Manager. diff --git a/internal/operator/packagesource_reconciler_test.go b/internal/operator/packagesource_reconciler_test.go new file mode 100644 index 000000000..f21cb6978 --- /dev/null +++ b/internal/operator/packagesource_reconciler_test.go @@ -0,0 +1,1434 @@ +/* +Copyright 2025 The Cozystack Authors. + +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 operator + +import ( + "context" + "errors" + "strconv" + "testing" + "time" + + cozyv1alpha1 "github.com/cozystack/cozystack/api/v1alpha1" + sourcewatcherv1beta1 "github.com/fluxcd/source-watcher/api/v2/v1beta1" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" +) + +// referenceTime is a fixed anchor for time-sensitive tests; every duration in +// the assertions is expressed as an offset from it so the tests read as +// "given the AG was last touched at reference-8s, and now is reference, ...". +var referenceTime = time.Date(2026, 7, 3, 12, 0, 0, 0, time.UTC) + +// agFixture builds an ArtifactGenerator with a fixed .metadata.generation and +// creation timestamp so tests can exercise the Generation and grace-period +// branches of artifactGeneratorStuck without a full apiserver round-trip. +func agFixture(gen int64, createdOffset time.Duration, status sourcewatcherv1beta1.ArtifactGeneratorStatus) *sourcewatcherv1beta1.ArtifactGenerator { + return &sourcewatcherv1beta1.ArtifactGenerator{ + ObjectMeta: metav1.ObjectMeta{ + Generation: gen, + CreationTimestamp: metav1.NewTime(referenceTime.Add(createdOffset)), + }, + Status: status, + } +} + +// TestArtifactGeneratorStuck locks in detection of the fluxcd/pkg#934 split +// -write stall. The predicate must fire ONLY when artifacts are demonstrably +// produced on the current generation AND the Ready condition has been sitting +// in Unknown longer than the grace period (or is missing on an old-enough AG). +// Ready=True and Ready=False must both pass through untouched — the retry +// driver only runs for the stuck-Unknown case, and a real regeneration failure +// (Ready=False) must never be papered over. +func TestArtifactGeneratorStuck(t *testing.T) { + nonEmptyInventory := []sourcewatcherv1beta1.ExternalArtifactReference{ + {Name: "one", Namespace: "cozy-system", Digest: "sha256:aaa"}, + } + const observedDigest = "sha256:0e7" + + // beyondGrace: how far into Unknown the AG has been before we'll intervene. + beyondGrace := stuckGracePeriod + 5*time.Second + + tests := []struct { + name string + ag *sourcewatcherv1beta1.ArtifactGenerator + want bool + }{ + { + name: "fresh ArtifactGenerator with nothing observed yet", + ag: agFixture(1, -time.Hour, sourcewatcherv1beta1.ArtifactGeneratorStatus{}), + want: false, + }, + { + name: "inventory present but sources not yet observed", + ag: agFixture(1, -time.Hour, sourcewatcherv1beta1.ArtifactGeneratorStatus{ + Inventory: nonEmptyInventory, + }), + want: false, + }, + { + name: "sources observed but inventory empty", + ag: agFixture(1, -time.Hour, sourcewatcherv1beta1.ArtifactGeneratorStatus{ + ObservedSourcesDigest: observedDigest, + }), + want: false, + }, + { + name: "artifacts produced, Ready condition absent, AG within grace period — do NOT intervene yet", + ag: agFixture(1, -5*time.Second, sourcewatcherv1beta1.ArtifactGeneratorStatus{ + Inventory: nonEmptyInventory, + ObservedSourcesDigest: observedDigest, + }), + want: false, + }, + { + name: "artifacts produced, Ready condition absent, AG older than grace period — intervene", + ag: agFixture(1, -beyondGrace, sourcewatcherv1beta1.ArtifactGeneratorStatus{ + Inventory: nonEmptyInventory, + ObservedSourcesDigest: observedDigest, + }), + want: true, + }, + { + name: "artifacts produced, Ready=Unknown within grace period — legitimate in-flight rebuild, do NOT intervene", + ag: agFixture(1, -time.Hour, sourcewatcherv1beta1.ArtifactGeneratorStatus{ + Inventory: nonEmptyInventory, + ObservedSourcesDigest: observedDigest, + Conditions: []metav1.Condition{{ + Type: "Ready", + Status: metav1.ConditionUnknown, + Reason: "Progressing", + Message: "Reconciliation in progress", + ObservedGeneration: 1, + LastTransitionTime: metav1.NewTime(referenceTime.Add(-5 * time.Second)), + }}, + }), + want: false, + }, + { + name: "artifacts produced, Ready=Unknown beyond grace period on current Generation — the stuck-status case", + ag: agFixture(1, -time.Hour, sourcewatcherv1beta1.ArtifactGeneratorStatus{ + Inventory: nonEmptyInventory, + ObservedSourcesDigest: observedDigest, + Conditions: []metav1.Condition{{ + Type: "Ready", + Status: metav1.ConditionUnknown, + Reason: "Progressing", + Message: "Reconciliation in progress", + ObservedGeneration: 1, + LastTransitionTime: metav1.NewTime(referenceTime.Add(-beyondGrace)), + }}, + }), + want: true, + }, + { + name: "artifacts observed on a PRIOR Generation (spec was updated) — do NOT intervene, condition is legitimately stale", + ag: agFixture(2, -time.Hour, sourcewatcherv1beta1.ArtifactGeneratorStatus{ + Inventory: nonEmptyInventory, + ObservedSourcesDigest: observedDigest, + Conditions: []metav1.Condition{{ + Type: "Ready", + Status: metav1.ConditionUnknown, + Reason: "Progressing", + Message: "Reconciliation in progress", + ObservedGeneration: 1, + LastTransitionTime: metav1.NewTime(referenceTime.Add(-beyondGrace)), + }}, + }), + want: false, + }, + { + name: "upstream Ready=True — pass through the real condition, do NOT intervene", + ag: agFixture(1, -time.Hour, sourcewatcherv1beta1.ArtifactGeneratorStatus{ + Inventory: nonEmptyInventory, + ObservedSourcesDigest: observedDigest, + Conditions: []metav1.Condition{{ + Type: "Ready", + Status: metav1.ConditionTrue, + Reason: "Succeeded", + ObservedGeneration: 1, + LastTransitionTime: metav1.NewTime(referenceTime.Add(-beyondGrace)), + }}, + }), + want: false, + }, + { + name: "upstream Ready=False — real failure, must NOT be masked", + ag: agFixture(1, -time.Hour, sourcewatcherv1beta1.ArtifactGeneratorStatus{ + Inventory: nonEmptyInventory, + ObservedSourcesDigest: observedDigest, + Conditions: []metav1.Condition{{ + Type: "Ready", + Status: metav1.ConditionFalse, + Reason: "SourceRefFailed", + Message: "cannot fetch source", + ObservedGeneration: 1, + LastTransitionTime: metav1.NewTime(referenceTime.Add(-beyondGrace)), + }}, + }), + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ready := meta.FindStatusCondition(tt.ag.Status.Conditions, "Ready") + got := artifactGeneratorStuck(tt.ag, ready, referenceTime) + if got != tt.want { + t.Errorf("artifactGeneratorStuck() = %v, want %v", got, tt.want) + } + }) + } +} + +// TestDecideRecovery locks in the bounded-retry state machine that drives +// source-watcher out of the stuck window. First attempt forces immediately; +// the N-th subsequent attempt waits backoffFor(N) since the last force before +// firing again; after maxRecoveryAttempts fruitless attempts we surface +// Ready=False. Wait durations are asserted for exact equality because +// decideRecovery is deterministic (`needed - elapsed`) — a loose bound would +// silently miss a backoff-schedule regression. +func TestDecideRecovery(t *testing.T) { + tests := []struct { + name string + attempts int + lastRecoveryAt time.Time + now time.Time + wantAction recoveryAction + wantWait time.Duration // 0 means don't check (Force / GiveUp branches ignore wait) + }{ + { + name: "first-ever detection — force immediately", + attempts: 0, + now: referenceTime, + wantAction: recoveryActionForce, + }, + { + name: "backoff not yet elapsed after attempt 1 — wait", + attempts: 1, + lastRecoveryAt: referenceTime.Add(-10 * time.Second), + now: referenceTime, + wantAction: recoveryActionWait, + wantWait: 20 * time.Second, // initialBackoff (30s) minus elapsed 10s + }, + { + name: "backoff elapsed after attempt 1 — force", + attempts: 1, + lastRecoveryAt: referenceTime.Add(-45 * time.Second), + now: referenceTime, + wantAction: recoveryActionForce, + }, + { + name: "backoff after attempt 3 (2m) — wait 90s", + attempts: 3, + lastRecoveryAt: referenceTime.Add(-30 * time.Second), + now: referenceTime, + wantAction: recoveryActionWait, + wantWait: 90 * time.Second, // backoffFor(3)=2m minus elapsed 30s + }, + { + name: "attempts exhausted with elapsed >> final backoff — give up", + attempts: maxRecoveryAttempts, + now: referenceTime, + wantAction: recoveryActionGiveUp, + }, + { + name: "attempts exceeded — still give up", + attempts: maxRecoveryAttempts + 3, + now: referenceTime, + wantAction: recoveryActionGiveUp, + }, + { + // lexfrei's follow-up: without holding the final backoff window, + // the Owns()-watch re-fire from the N-th force-drift's own + // writes preempts the response window and give-up fires within + // milliseconds of the last force. Budget collapses to ~7.5m + // instead of the documented ~11.5m. This case pins the fix: + // attempts=max but no time elapsed since the final force, + // expect Wait for the full backoffFor(maxRecoveryAttempts). + name: "attempts exhausted but final backoff not elapsed — wait, do not give up", + attempts: maxRecoveryAttempts, + lastRecoveryAt: referenceTime.Add(-time.Second), + now: referenceTime, + wantAction: recoveryActionWait, + wantWait: backoffFor(maxRecoveryAttempts) - time.Second, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := decideRecovery(tt.attempts, tt.lastRecoveryAt, tt.now) + if got.action != tt.wantAction { + t.Fatalf("action = %v, want %v", got.action, tt.wantAction) + } + if tt.wantWait > 0 && got.wait != tt.wantWait { + t.Errorf("wait = %v, want %v", got.wait, tt.wantWait) + } + }) + } +} + +// TestBackoffFor pins the exponential-with-cap schedule so a future edit that +// silently doubles the ceiling or drops the exponent gets caught here. +func TestBackoffFor(t *testing.T) { + // initialBackoff=30s, maxBackoff=4m. Expected: 30s, 60s, 120s, 240s, 240s, 240s, ... + tests := []struct { + attempt int + want time.Duration + }{ + {0, 30 * time.Second}, // clamped to attempt=1 + {1, 30 * time.Second}, + {2, 60 * time.Second}, + {3, 120 * time.Second}, + {4, 240 * time.Second}, + {5, 240 * time.Second}, + {10, 240 * time.Second}, + } + for _, tt := range tests { + t.Run(strconv.Itoa(tt.attempt), func(t *testing.T) { + got := backoffFor(tt.attempt) + if got != tt.want { + t.Errorf("backoffFor(%d) = %v, want %v", tt.attempt, got, tt.want) + } + }) + } +} + +// TestReadRecoveryTracking verifies that a corrupted or malformed retry-tracking +// annotation cannot wedge the retry loop by producing a nonsensical counter — +// missing, empty, non-numeric, or negative values must all read back as +// "no prior attempts". +func TestReadRecoveryTracking(t *testing.T) { + fixed := referenceTime.UTC().Format(time.RFC3339Nano) + + tests := []struct { + name string + annotations map[string]string + wantAttempt int + wantTimeSet bool + }{ + { + name: "no annotations", + annotations: nil, + wantAttempt: 0, + wantTimeSet: false, + }, + { + name: "valid attempt + valid timestamp", + annotations: map[string]string{ + annotationRecoveryAttempts: "3", + annotationLastRecoveryAt: fixed, + }, + wantAttempt: 3, + wantTimeSet: true, + }, + { + name: "corrupt attempt counter — read as zero", + annotations: map[string]string{ + annotationRecoveryAttempts: "not-a-number", + annotationLastRecoveryAt: fixed, + }, + wantAttempt: 0, + wantTimeSet: true, + }, + { + name: "negative attempt — clamped to zero", + annotations: map[string]string{ + annotationRecoveryAttempts: "-5", + }, + wantAttempt: 0, + wantTimeSet: false, + }, + { + name: "corrupt timestamp — read as zero time", + annotations: map[string]string{ + annotationRecoveryAttempts: "2", + annotationLastRecoveryAt: "not-a-date", + }, + wantAttempt: 2, + wantTimeSet: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ag := &sourcewatcherv1beta1.ArtifactGenerator{ + ObjectMeta: metav1.ObjectMeta{Annotations: tt.annotations}, + } + attempt, ts := readRecoveryTracking(ag) + if attempt != tt.wantAttempt { + t.Errorf("attempt = %d, want %d", attempt, tt.wantAttempt) + } + if tt.wantTimeSet && ts.IsZero() { + t.Error("timestamp is zero, want set") + } + if !tt.wantTimeSet && !ts.IsZero() { + t.Errorf("timestamp = %v, want zero", ts) + } + }) + } +} + +// TestCloneAnnotations pins the "shallow copy, nil in / nil out" contract that +// the rollback path in forceArtifactGeneratorDrift / clearRecoveryTracking +// depends on. If a future refactor "normalises" nil to an empty map, the +// rollback would leave the caller with an empty map instead of the original +// nil, subtly changing observable state in downstream code that treats +// annotations == nil as a distinct signal. +func TestCloneAnnotations(t *testing.T) { + t.Run("nil in, nil out", func(t *testing.T) { + if got := cloneAnnotations(nil); got != nil { + t.Errorf("cloneAnnotations(nil) = %v, want nil", got) + } + }) + t.Run("empty in, non-nil empty out", func(t *testing.T) { + got := cloneAnnotations(map[string]string{}) + if got == nil { + t.Error("cloneAnnotations(empty) returned nil, want empty map") + } + if len(got) != 0 { + t.Errorf("cloneAnnotations(empty) len = %d, want 0", len(got)) + } + }) + t.Run("populated in, independent copy out", func(t *testing.T) { + src := map[string]string{"a": "1", "b": "2"} + got := cloneAnnotations(src) + if len(got) != 2 || got["a"] != "1" || got["b"] != "2" { + t.Errorf("cloneAnnotations(populated) = %v, want copy of %v", got, src) + } + // Mutation-independence: modifying the source must not touch the copy. + src["a"] = "changed" + if got["a"] != "1" { + t.Errorf("clone was aliased to source; got[a]=%q after src[a]=changed", got["a"]) + } + }) +} + +// testScheme is a shared scheme with the CRDs both reconciler test suites need. +func testScheme(t *testing.T) *runtime.Scheme { + t.Helper() + s := runtime.NewScheme() + if err := cozyv1alpha1.AddToScheme(s); err != nil { + t.Fatalf("cozyv1alpha1.AddToScheme: %v", err) + } + if err := sourcewatcherv1beta1.AddToScheme(s); err != nil { + t.Fatalf("sourcewatcherv1beta1.AddToScheme: %v", err) + } + return s +} + +// newAG builds a minimal ArtifactGenerator suitable for fake-client tests. +func newAG(annotations map[string]string) *sourcewatcherv1beta1.ArtifactGenerator { + return &sourcewatcherv1beta1.ArtifactGenerator{ + ObjectMeta: metav1.ObjectMeta{ + Name: "example", + Namespace: "cozy-system", + Annotations: annotations, + }, + } +} + +// TestForceArtifactGeneratorDrift_Success asserts that on a successful +// force-drift call: +// - the AG's status.conditions[Ready] is set to False in the apiserver (this +// is the signal source-watcher's detectDrift keys on), +// - the tracking annotations land on the caller's `ag` AND in the apiserver, +// - the same in-memory `ag` reflects both mutations (mutation contract). +// +// This locks in the two-subresource shape (status patch + metadata patch) and +// the mutation contract documented on the function. +func TestForceArtifactGeneratorDrift_Success(t *testing.T) { + ag := newAG(nil) + c := fake.NewClientBuilder().WithScheme(testScheme(t)).WithStatusSubresource(ag).WithObjects(ag).Build() + r := &PackageSourceReconciler{Client: c, Scheme: testScheme(t)} + + if err := r.forceArtifactGeneratorDrift(context.Background(), ag, referenceTime, 2); err != nil { + t.Fatalf("forceArtifactGeneratorDrift: %v", err) + } + + // In-memory ag: Ready=False + tracking annotations reflect the writes. + ready := meta.FindStatusCondition(ag.Status.Conditions, "Ready") + if ready == nil || ready.Status != metav1.ConditionFalse || ready.Reason != reasonRecoveryForced { + t.Fatalf("ag.Status Ready condition = %+v, want False/%s", ready, reasonRecoveryForced) + } + if got := ag.Annotations[annotationRecoveryAttempts]; got != "2" { + t.Errorf("ag.Annotations[%s] = %q, want 2", annotationRecoveryAttempts, got) + } + if got := ag.Annotations[annotationLastRecoveryAt]; got == "" { + t.Errorf("ag.Annotations[%s] not set", annotationLastRecoveryAt) + } + // B1 (lexfrei): the requestedAt annotation MUST also be set — it is the + // signal source-watcher's ReconcileRequestedPredicate keys on to enqueue + // a reconcile. Without it the status patch alone is a no-op for up to + // the AG's 1h self-requeue interval. + if got := ag.Annotations[annotationFluxRequestedAt]; got == "" { + t.Errorf("ag.Annotations[%s] not set — source-watcher will not enqueue a reconcile", annotationFluxRequestedAt) + } + + // apiserver reflects both writes. + persisted := &sourcewatcherv1beta1.ArtifactGenerator{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(ag), persisted); err != nil { + t.Fatalf("Get after force-drift: %v", err) + } + persistedReady := meta.FindStatusCondition(persisted.Status.Conditions, "Ready") + if persistedReady == nil || persistedReady.Status != metav1.ConditionFalse { + t.Errorf("persisted AG Ready condition = %+v, want False", persistedReady) + } + for _, k := range []string{annotationFluxRequestedAt, annotationRecoveryAttempts, annotationLastRecoveryAt} { + if _, ok := persisted.Annotations[k]; !ok { + t.Errorf("persisted AG missing annotation %s", k) + } + } +} + +// TestForceArtifactGeneratorDrift_StatusPatchFailure_Rollback drives the +// status-patch failure branch: an interceptor errors on SubResourcePatch. +// The caller's `ag` must be untouched (both conditions and annotations) and +// no metadata patch should have been issued. +func TestForceArtifactGeneratorDrift_StatusPatchFailure_Rollback(t *testing.T) { + ag := newAG(map[string]string{"unrelated": "value"}) + baseClient := fake.NewClientBuilder().WithScheme(testScheme(t)).WithStatusSubresource(ag).WithObjects(ag).Build() + metadataPatches := 0 + failing := interceptor.NewClient(baseClient, interceptor.Funcs{ + Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + metadataPatches++ + return c.Patch(ctx, obj, patch, opts...) + }, + SubResourcePatch: func(_ context.Context, _ client.Client, _ string, _ client.Object, _ client.Patch, _ ...client.SubResourcePatchOption) error { + return errors.New("simulated status patch failure") + }, + }) + r := &PackageSourceReconciler{Client: failing, Scheme: testScheme(t)} + + if err := r.forceArtifactGeneratorDrift(context.Background(), ag, referenceTime, 3); err == nil { + t.Fatal("forceArtifactGeneratorDrift succeeded, want error") + } + if ready := meta.FindStatusCondition(ag.Status.Conditions, "Ready"); ready != nil { + t.Errorf("ag.Status Ready condition = %+v after failed status patch, want unset (rolled back)", ready) + } + if _, ok := ag.Annotations[annotationRecoveryAttempts]; ok { + t.Errorf("tracking annotation leaked despite status-patch failure: %v", ag.Annotations) + } + if metadataPatches != 0 { + t.Errorf("metadata patch issued %d times despite status-patch failure, want 0", metadataPatches) + } + if ag.Annotations["unrelated"] != "value" { + t.Errorf("pre-existing annotation lost: %v", ag.Annotations) + } +} + +// TestForceArtifactGeneratorDrift_MetadataPatchFailure_RollbackAnnotations +// drives the second-step failure: status patch succeeds, metadata patch fails. +// Ready=False must remain in the caller's `ag` (successfully persisted) but +// the tracking annotations must roll back to pre-call state so the counter +// doesn't advance without landing in the apiserver. +func TestForceArtifactGeneratorDrift_MetadataPatchFailure_RollbackAnnotations(t *testing.T) { + ag := newAG(map[string]string{"unrelated": "value"}) + baseClient := fake.NewClientBuilder().WithScheme(testScheme(t)).WithStatusSubresource(ag).WithObjects(ag).Build() + failing := interceptor.NewClient(baseClient, interceptor.Funcs{ + Patch: func(_ context.Context, _ client.WithWatch, _ client.Object, _ client.Patch, _ ...client.PatchOption) error { + return errors.New("simulated metadata patch failure") + }, + }) + r := &PackageSourceReconciler{Client: failing, Scheme: testScheme(t)} + + if err := r.forceArtifactGeneratorDrift(context.Background(), ag, referenceTime, 4); err == nil { + t.Fatal("forceArtifactGeneratorDrift succeeded, want error from metadata patch") + } + // Status patch already landed, so ag.Status.Conditions carries Ready=False + // even though the operation as a whole errored. + ready := meta.FindStatusCondition(ag.Status.Conditions, "Ready") + if ready == nil || ready.Status != metav1.ConditionFalse { + t.Errorf("ag.Status Ready condition = %+v, want False (status patch succeeded)", ready) + } + // Annotations must roll back — the counter never advanced in apiserver so + // leaving it locally advanced would desync the caller's view. Both the + // tracking annotations AND the requestedAt bump (added for B1) must be + // symmetrically rolled back — otherwise source-watcher would see a bumped + // requestedAt with no matching backoff-state on our side. + if _, ok := ag.Annotations[annotationRecoveryAttempts]; ok { + t.Errorf("tracking annotation not rolled back after metadata patch failure: %v", ag.Annotations) + } + if _, ok := ag.Annotations[annotationFluxRequestedAt]; ok { + t.Errorf("requestedAt annotation not rolled back after metadata patch failure: %v", ag.Annotations) + } + if ag.Annotations["unrelated"] != "value" { + t.Errorf("pre-existing annotation lost: %v", ag.Annotations) + } +} + +// TestForceArtifactGeneratorDrift_MetadataPatchFailure_RollbackRestoresPriorBump +// exercises the subtle case where prior tracking annotations were set by an +// EARLIER force-drift and the current metadata patch fails. Rollback must +// restore prior values, not delete them. +func TestForceArtifactGeneratorDrift_MetadataPatchFailure_RollbackRestoresPriorBump(t *testing.T) { + priorTimestamp := referenceTime.Add(-time.Minute).UTC().Format(time.RFC3339Nano) + ag := newAG(map[string]string{ + annotationRecoveryAttempts: "1", + annotationLastRecoveryAt: priorTimestamp, + }) + baseClient := fake.NewClientBuilder().WithScheme(testScheme(t)).WithStatusSubresource(ag).WithObjects(ag).Build() + failing := interceptor.NewClient(baseClient, interceptor.Funcs{ + Patch: func(_ context.Context, _ client.WithWatch, _ client.Object, _ client.Patch, _ ...client.PatchOption) error { + return errors.New("simulated metadata patch failure") + }, + }) + r := &PackageSourceReconciler{Client: failing, Scheme: testScheme(t)} + + if err := r.forceArtifactGeneratorDrift(context.Background(), ag, referenceTime, 2); err == nil { + t.Fatal("force-drift succeeded, want error") + } + if got := ag.Annotations[annotationRecoveryAttempts]; got != "1" { + t.Errorf("annotationRecoveryAttempts = %q after rollback, want prior value 1", got) + } + if got := ag.Annotations[annotationLastRecoveryAt]; got != priorTimestamp { + t.Errorf("annotationLastRecoveryAt = %q after rollback, want prior %q", got, priorTimestamp) + } +} + +// TestClearRecoveryTracking_Noop asserts that if the AG has no tracking +// annotations, we don't issue a Patch at all — otherwise every healthy +// reconcile would generate a no-op write. +func TestClearRecoveryTracking_Noop(t *testing.T) { + ag := newAG(map[string]string{"unrelated": "value"}) + baseClient := fake.NewClientBuilder().WithScheme(testScheme(t)).WithObjects(ag).Build() + patchCount := 0 + watched := interceptor.NewClient(baseClient, interceptor.Funcs{ + Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + patchCount++ + return c.Patch(ctx, obj, patch, opts...) + }, + }) + r := &PackageSourceReconciler{Client: watched, Scheme: testScheme(t)} + + if err := r.clearRecoveryTracking(context.Background(), ag); err != nil { + t.Fatalf("clearRecoveryTracking: %v", err) + } + if patchCount != 0 { + t.Errorf("clearRecoveryTracking issued %d Patch calls, want 0", patchCount) + } +} + +// TestClearRecoveryTracking_PatchFailure_Rollback drives the failure branch and +// asserts the tracking annotations are restored on the caller's `ag`. +func TestClearRecoveryTracking_PatchFailure_Rollback(t *testing.T) { + original := map[string]string{ + annotationRecoveryAttempts: "3", + annotationLastRecoveryAt: referenceTime.UTC().Format(time.RFC3339Nano), + "unrelated": "value", + } + ag := newAG(cloneAnnotations(original)) + baseClient := fake.NewClientBuilder().WithScheme(testScheme(t)).WithObjects(ag).Build() + failing := interceptor.NewClient(baseClient, interceptor.Funcs{ + Patch: func(_ context.Context, _ client.WithWatch, _ client.Object, _ client.Patch, _ ...client.PatchOption) error { + return errors.New("simulated Patch failure") + }, + }) + r := &PackageSourceReconciler{Client: failing, Scheme: testScheme(t)} + + if err := r.clearRecoveryTracking(context.Background(), ag); err == nil { + t.Fatal("clearRecoveryTracking succeeded, want error") + } + if len(ag.Annotations) != len(original) { + t.Fatalf("ag.Annotations size = %d, want %d after rollback", len(ag.Annotations), len(original)) + } + for k, v := range original { + if ag.Annotations[k] != v { + t.Errorf("ag.Annotations[%s] = %q, want %q after rollback", k, ag.Annotations[k], v) + } + } +} + +// TestReconcile_DeletionTimestamp_EarlyReturn asserts that a PackageSource +// carrying a deletionTimestamp does not have its status touched and does not +// trigger any writes to its ArtifactGenerator's annotations — the retry driver +// must not keep chasing an object that's already on its way out. +func TestReconcile_DeletionTimestamp_EarlyReturn(t *testing.T) { + now := metav1.NewTime(referenceTime) + ps := &cozyv1alpha1.PackageSource{ + ObjectMeta: metav1.ObjectMeta{ + Name: "doomed", + Namespace: "cozy-system", + DeletionTimestamp: &now, + Finalizers: []string{"cozystack.io/pretend"}, + }, + Spec: cozyv1alpha1.PackageSourceSpec{}, + } + ag := newAG(map[string]string{annotationRecoveryAttempts: "2"}) + + writes := 0 + base := fake.NewClientBuilder().WithScheme(testScheme(t)).WithObjects(ps, ag).Build() + watched := interceptor.NewClient(base, interceptor.Funcs{ + Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + writes++ + return c.Patch(ctx, obj, patch, opts...) + }, + Update: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.UpdateOption) error { + writes++ + return c.Update(ctx, obj, opts...) + }, + SubResourcePatch: func(ctx context.Context, c client.Client, subResourceName string, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error { + writes++ + return c.SubResource(subResourceName).Patch(ctx, obj, patch, opts...) + }, + SubResourceUpdate: func(ctx context.Context, c client.Client, subResourceName string, obj client.Object, opts ...client.SubResourceUpdateOption) error { + writes++ + return c.SubResource(subResourceName).Update(ctx, obj, opts...) + }, + }) + r := &PackageSourceReconciler{Client: watched, Scheme: testScheme(t)} + + res, err := r.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: ps.Name, Namespace: ps.Namespace}, + }) + if err != nil { + t.Fatalf("Reconcile: %v", err) + } + if res.RequeueAfter != 0 { + t.Errorf("Reconcile scheduled RequeueAfter=%v on a deleted object, want 0", res.RequeueAfter) + } + if writes != 0 { + t.Errorf("Reconcile issued %d writes on a deleted object, want 0", writes) + } +} + +// TestUpdateStatus_UnknownWithinGrace_SchedulesFollowUp locks in the fix for +// the dormancy bug caught on dev3: when Ready=Unknown but grace hasn't elapsed +// yet, artifactGeneratorStuck returns false, we fall through to the copy path +// and — without an explicit RequeueAfter — the reconciler would sleep until +// the AG's next status change (which may never come if it's exactly the +// split-patch race we exist to fix) or the 10h informer resync. This test +// asserts we schedule a follow-up reconcile at grace-period expiry so the +// recovery driver can take over. +func TestUpdateStatus_UnknownWithinGrace_SchedulesFollowUp(t *testing.T) { + // Ready=Unknown, transitioned 10s ago — grace is 30s, so 20s remaining. + transitionAt := referenceTime.Add(-10 * time.Second) + ps := &cozyv1alpha1.PackageSource{ + ObjectMeta: metav1.ObjectMeta{Name: "example", Namespace: "cozy-system", Generation: 1}, + Spec: cozyv1alpha1.PackageSourceSpec{ + SourceRef: &cozyv1alpha1.PackageSourceRef{ + Name: "src", + Kind: "OCIRepository", + Namespace: "cozy-system", + }, + }, + } + ag := &sourcewatcherv1beta1.ArtifactGenerator{ + ObjectMeta: metav1.ObjectMeta{ + Name: "example", Namespace: "cozy-system", Generation: 1, + CreationTimestamp: metav1.NewTime(referenceTime.Add(-time.Hour)), + }, + Status: sourcewatcherv1beta1.ArtifactGeneratorStatus{ + Inventory: []sourcewatcherv1beta1.ExternalArtifactReference{ + {Name: "one", Namespace: "cozy-system", Digest: "sha256:aaa"}, + }, + ObservedSourcesDigest: "sha256:0e7", + Conditions: []metav1.Condition{{ + Type: "Ready", + Status: metav1.ConditionUnknown, + Reason: "Progressing", + Message: "Reconciliation in progress", + ObservedGeneration: 1, + LastTransitionTime: metav1.NewTime(transitionAt), + }}, + }, + } + c := fake.NewClientBuilder().WithScheme(testScheme(t)).WithStatusSubresource(ps, ag).WithObjects(ps, ag).Build() + r := &PackageSourceReconciler{Client: c, Scheme: testScheme(t)} + + res, err := r.updateStatus(context.Background(), ps, referenceTime) + if err != nil { + t.Fatalf("updateStatus: %v", err) + } + // remaining = transitionAt + stuckGracePeriod - now = -10s + 30s = 20s + if res.RequeueAfter != 20*time.Second { + t.Errorf("RequeueAfter = %v, want 20s (remaining grace after 10s elapsed)", res.RequeueAfter) + } +} + +// TestUpdateStatus_MissingReadyWithinGrace_SchedulesFollowUp — same dormancy +// fix but for the ready==nil branch: a fresh AG with no Ready condition yet +// must also schedule a follow-up so the operator notices when grace elapses. +func TestUpdateStatus_MissingReadyWithinGrace_SchedulesFollowUp(t *testing.T) { + createdAt := referenceTime.Add(-5 * time.Second) // 25s remaining on grace + ps := &cozyv1alpha1.PackageSource{ + ObjectMeta: metav1.ObjectMeta{Name: "example", Namespace: "cozy-system", Generation: 1}, + Spec: cozyv1alpha1.PackageSourceSpec{ + SourceRef: &cozyv1alpha1.PackageSourceRef{ + Name: "src", + Kind: "OCIRepository", + Namespace: "cozy-system", + }, + }, + } + ag := &sourcewatcherv1beta1.ArtifactGenerator{ + ObjectMeta: metav1.ObjectMeta{ + Name: "example", Namespace: "cozy-system", Generation: 1, + CreationTimestamp: metav1.NewTime(createdAt), + }, + Status: sourcewatcherv1beta1.ArtifactGeneratorStatus{}, // no Ready + } + c := fake.NewClientBuilder().WithScheme(testScheme(t)).WithStatusSubresource(ps, ag).WithObjects(ps, ag).Build() + r := &PackageSourceReconciler{Client: c, Scheme: testScheme(t)} + + res, err := r.updateStatus(context.Background(), ps, referenceTime) + if err != nil { + t.Fatalf("updateStatus: %v", err) + } + if res.RequeueAfter != 25*time.Second { + t.Errorf("RequeueAfter = %v, want 25s", res.RequeueAfter) + } +} + +// TestAgFollowUpDelay pins the follow-up-delay computation, including the +// one-second floor that prevents a zero-delay tight loop when the anchor is +// right at or beyond grace expiry. +func TestAgFollowUpDelay(t *testing.T) { + tests := []struct { + name string + anchorOffset time.Duration + want time.Duration + }{ + {"anchor 10s ago — grace has 20s left", -10 * time.Second, 20 * time.Second}, + {"anchor 5s ago — grace has 25s left", -5 * time.Second, 25 * time.Second}, + {"anchor right at grace expiry — floor to 1s", -stuckGracePeriod, time.Second}, + {"anchor past grace expiry — floor to 1s", -2 * stuckGracePeriod, time.Second}, + {"anchor now — full grace remaining", 0, stuckGracePeriod}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := agFollowUpDelay(referenceTime.Add(tt.anchorOffset), referenceTime) + if got != tt.want { + t.Errorf("agFollowUpDelay(offset=%v) = %v, want %v", tt.anchorOffset, got, tt.want) + } + }) + } +} + +// stuckAG builds an ArtifactGenerator in the fluxcd/pkg#934 stall signature: +// Inventory populated, ObservedSourcesDigest set, Ready=Unknown on the current +// generation, LastTransitionTime old enough to clear the grace period. Used by +// the maybeRecoverArtifactGenerator tests below. +func stuckAG(t *testing.T, annotations map[string]string, readyTransition time.Time) *sourcewatcherv1beta1.ArtifactGenerator { + t.Helper() + return &sourcewatcherv1beta1.ArtifactGenerator{ + ObjectMeta: metav1.ObjectMeta{ + Name: "example", + Namespace: "cozy-system", + Generation: 1, + Annotations: annotations, + }, + Status: sourcewatcherv1beta1.ArtifactGeneratorStatus{ + Inventory: []sourcewatcherv1beta1.ExternalArtifactReference{ + {Name: "one", Namespace: "cozy-system", Digest: "sha256:aaa"}, + }, + ObservedSourcesDigest: "sha256:0e7", + Conditions: []metav1.Condition{{ + Type: "Ready", + Status: metav1.ConditionUnknown, + Reason: "Progressing", + Message: "Reconciliation in progress", + ObservedGeneration: 1, + LastTransitionTime: metav1.NewTime(readyTransition), + }}, + }, + } +} + +// TestMaybeRecoverArtifactGenerator_ForceBranch pins the first-time-stuck path: +// no prior tracking annotations, so decideRecovery returns Force, we patch +// Ready=False on the AG's status, update tracking annotations, and set the +// PackageSource to Unknown/AwaitingSourceWatcherRecovery with a RequeueAfter +// equal to backoffFor(1). +func TestMaybeRecoverArtifactGenerator_ForceBranch(t *testing.T) { + ps := &cozyv1alpha1.PackageSource{ + ObjectMeta: metav1.ObjectMeta{Name: "example", Namespace: "cozy-system", Generation: 1}, + } + ag := stuckAG(t, nil, referenceTime.Add(-2*time.Minute)) + c := fake.NewClientBuilder().WithScheme(testScheme(t)).WithStatusSubresource(ps, ag).WithObjects(ps, ag).Build() + r := &PackageSourceReconciler{Client: c, Scheme: testScheme(t)} + + ready := meta.FindStatusCondition(ag.Status.Conditions, "Ready") + res, err := r.maybeRecoverArtifactGenerator(context.Background(), ps, ag, ready, referenceTime) + if err != nil { + t.Fatalf("maybeRecoverArtifactGenerator: %v", err) + } + if res.RequeueAfter != backoffFor(1) { + t.Errorf("RequeueAfter = %v, want %v", res.RequeueAfter, backoffFor(1)) + } + + // AG status.Ready must be False in apiserver. + persistedAG := &sourcewatcherv1beta1.ArtifactGenerator{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(ag), persistedAG); err != nil { + t.Fatalf("Get AG: %v", err) + } + agReady := meta.FindStatusCondition(persistedAG.Status.Conditions, "Ready") + if agReady == nil || agReady.Status != metav1.ConditionFalse { + t.Errorf("AG Ready in apiserver = %+v, want False", agReady) + } + if persistedAG.Annotations[annotationRecoveryAttempts] != "1" { + t.Errorf("AG %s annotation = %q, want 1", annotationRecoveryAttempts, persistedAG.Annotations[annotationRecoveryAttempts]) + } + + // PackageSource must be Unknown/AwaitingSourceWatcherRecovery in apiserver. + persistedPS := &cozyv1alpha1.PackageSource{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(ps), persistedPS); err != nil { + t.Fatalf("Get PS: %v", err) + } + psReady := meta.FindStatusCondition(persistedPS.Status.Conditions, "Ready") + if psReady == nil || psReady.Status != metav1.ConditionUnknown || psReady.Reason != reasonAwaitingRecovery { + t.Errorf("PS Ready = %+v, want Unknown/%s", psReady, reasonAwaitingRecovery) + } +} + +// TestMaybeRecoverArtifactGenerator_WaitBranch pins the in-backoff-window path: +// prior attempt exists, backoff not yet elapsed, so decideRecovery returns +// Wait and we do NOT touch the AG's status/annotations — only the PackageSource +// gets Unknown/AwaitingSourceWatcherRecovery and RequeueAfter set to the +// remaining backoff. +func TestMaybeRecoverArtifactGenerator_WaitBranch(t *testing.T) { + priorForceAt := referenceTime.Add(-10 * time.Second) // 20s remaining on initialBackoff=30s + ps := &cozyv1alpha1.PackageSource{ + ObjectMeta: metav1.ObjectMeta{Name: "example", Namespace: "cozy-system", Generation: 1}, + } + ag := stuckAG(t, map[string]string{ + annotationRecoveryAttempts: "1", + annotationLastRecoveryAt: priorForceAt.UTC().Format(time.RFC3339Nano), + }, referenceTime.Add(-2*time.Minute)) + c := fake.NewClientBuilder().WithScheme(testScheme(t)).WithStatusSubresource(ps, ag).WithObjects(ps, ag).Build() + agWriteCount := 0 + watched := interceptor.NewClient(c, interceptor.Funcs{ + Patch: func(ctx context.Context, cli client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + if _, ok := obj.(*sourcewatcherv1beta1.ArtifactGenerator); ok { + agWriteCount++ + } + return cli.Patch(ctx, obj, patch, opts...) + }, + SubResourcePatch: func(ctx context.Context, cli client.Client, sub string, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error { + if _, ok := obj.(*sourcewatcherv1beta1.ArtifactGenerator); ok { + agWriteCount++ + } + return cli.SubResource(sub).Patch(ctx, obj, patch, opts...) + }, + }) + r := &PackageSourceReconciler{Client: watched, Scheme: testScheme(t)} + + ready := meta.FindStatusCondition(ag.Status.Conditions, "Ready") + res, err := r.maybeRecoverArtifactGenerator(context.Background(), ps, ag, ready, referenceTime) + if err != nil { + t.Fatalf("maybeRecoverArtifactGenerator: %v", err) + } + if res.RequeueAfter != 20*time.Second { + t.Errorf("RequeueAfter = %v, want 20s", res.RequeueAfter) + } + if agWriteCount != 0 { + t.Errorf("wait branch wrote %d times to AG, want 0", agWriteCount) + } + persistedPS := &cozyv1alpha1.PackageSource{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(ps), persistedPS); err != nil { + t.Fatalf("Get PS: %v", err) + } + psReady := meta.FindStatusCondition(persistedPS.Status.Conditions, "Ready") + if psReady == nil || psReady.Reason != reasonAwaitingRecovery { + t.Errorf("PS Ready = %+v, want reason %s", psReady, reasonAwaitingRecovery) + } +} + +// TestMaybeRecoverArtifactGenerator_GiveUpBranch pins the exhaustion path: +// attempts equals maxRecoveryAttempts, so we surface the failure as +// PackageSource Ready=False/SourceWatcherStalled and schedule no follow-up +// reconcile (an operator must intervene). The AG is left untouched. +func TestMaybeRecoverArtifactGenerator_GiveUpBranch(t *testing.T) { + ps := &cozyv1alpha1.PackageSource{ + ObjectMeta: metav1.ObjectMeta{Name: "example", Namespace: "cozy-system", Generation: 1}, + } + priorForceAt := referenceTime.Add(-time.Hour).UTC().Format(time.RFC3339Nano) + ag := stuckAG(t, map[string]string{ + annotationRecoveryAttempts: strconv.Itoa(maxRecoveryAttempts), + annotationLastRecoveryAt: priorForceAt, + }, referenceTime.Add(-time.Hour)) // last transition BEFORE lastRecoveryAt so reset-counter doesn't fire + c := fake.NewClientBuilder().WithScheme(testScheme(t)).WithStatusSubresource(ps, ag).WithObjects(ps, ag).Build() + r := &PackageSourceReconciler{Client: c, Scheme: testScheme(t)} + + ready := meta.FindStatusCondition(ag.Status.Conditions, "Ready") + res, err := r.maybeRecoverArtifactGenerator(context.Background(), ps, ag, ready, referenceTime) + if err != nil { + t.Fatalf("maybeRecoverArtifactGenerator: %v", err) + } + if res.RequeueAfter != 0 { + t.Errorf("RequeueAfter = %v, want 0 (give-up path stops rescheduling)", res.RequeueAfter) + } + persistedPS := &cozyv1alpha1.PackageSource{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(ps), persistedPS); err != nil { + t.Fatalf("Get PS: %v", err) + } + psReady := meta.FindStatusCondition(persistedPS.Status.Conditions, "Ready") + if psReady == nil || psReady.Status != metav1.ConditionFalse || psReady.Reason != reasonSourceWatcherBad { + t.Errorf("PS Ready = %+v, want False/%s", psReady, reasonSourceWatcherBad) + } +} + +// TestMaybeRecoverArtifactGenerator_DoesNotResetCounterOnFreshTransition pins +// the fix for lexfrei's B3: source-watcher writes Ready=Unknown (Progressing) +// as it takes the drifted branch after our force, and that transition's +// LastTransitionTime is always newer than lastRecoveryAt. A reset-counter +// heuristic that keys on that comparison would restart the budget on every +// Progressing write, making the SourceWatcherStalled give-up unreachable +// under the very race this driver targets. Assertion: with attempts already +// at maxRecoveryAttempts, we go straight to GiveUp even when Ready has just +// been touched — the natural clearRecoveryTracking on the not-stuck path is +// the only counter reset, and the unified own-marker / progressing-in-recovery +// gate in updateStatus holds tracking through Progressing by routing back +// through this same maybeRecoverArtifactGenerator's Wait branch. +func TestMaybeRecoverArtifactGenerator_DoesNotResetCounterOnFreshTransition(t *testing.T) { + priorForceAt := referenceTime.Add(-time.Hour) + freshTransitionAt := referenceTime.Add(-2 * time.Minute) // AFTER priorForceAt — must NOT reset + ps := &cozyv1alpha1.PackageSource{ + ObjectMeta: metav1.ObjectMeta{Name: "example", Namespace: "cozy-system", Generation: 1}, + } + ag := stuckAG(t, map[string]string{ + annotationRecoveryAttempts: strconv.Itoa(maxRecoveryAttempts), + annotationLastRecoveryAt: priorForceAt.UTC().Format(time.RFC3339Nano), + }, freshTransitionAt) + c := fake.NewClientBuilder().WithScheme(testScheme(t)).WithStatusSubresource(ps, ag).WithObjects(ps, ag).Build() + r := &PackageSourceReconciler{Client: c, Scheme: testScheme(t)} + + ready := meta.FindStatusCondition(ag.Status.Conditions, "Ready") + res, err := r.maybeRecoverArtifactGenerator(context.Background(), ps, ag, ready, referenceTime) + if err != nil { + t.Fatalf("maybeRecoverArtifactGenerator: %v", err) + } + // Give-up path returns no RequeueAfter (operator must intervene). + if res.RequeueAfter != 0 { + t.Errorf("RequeueAfter = %v, want 0 — give-up should not schedule a follow-up", res.RequeueAfter) + } + persistedPS := &cozyv1alpha1.PackageSource{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(ps), persistedPS); err != nil { + t.Fatalf("Get PS: %v", err) + } + psReady := meta.FindStatusCondition(persistedPS.Status.Conditions, "Ready") + if psReady == nil || psReady.Status != metav1.ConditionFalse || psReady.Reason != reasonSourceWatcherBad { + t.Errorf("PS Ready = %+v, want False/%s — reset would have taken the Force branch instead", psReady, reasonSourceWatcherBad) + } +} + +// TestUpdateStatus_OwnMarkerHoldsTrackingViaStateMachine pins lexfrei's B2 +// fix: when the AG's Ready condition carries our own reasonRecoveryForced +// marker (our previous forceArtifactGeneratorDrift write reflecting back +// through the Owns() watch), updateStatus must (a) NOT clear the +// recovery-tracking annotations we just wrote, and (b) NOT copy the +// synthetic Ready=False through to the PackageSource. It routes through the +// unified own-marker / progressing-in-recovery / stuck gate into +// maybeRecoverArtifactGenerator, and — because we are inside the backoff +// window — the state machine's Wait branch keeps the PS in +// Unknown/AwaitingSourceWatcherRecovery with a follow-up requeue. +func TestUpdateStatus_OwnMarkerHoldsTrackingViaStateMachine(t *testing.T) { + ps := &cozyv1alpha1.PackageSource{ + ObjectMeta: metav1.ObjectMeta{Name: "example", Namespace: "cozy-system", Generation: 1}, + Spec: cozyv1alpha1.PackageSourceSpec{ + SourceRef: &cozyv1alpha1.PackageSourceRef{Name: "src", Kind: "OCIRepository", Namespace: "cozy-system"}, + }, + } + forceAt := referenceTime.Add(-5 * time.Second) + ag := &sourcewatcherv1beta1.ArtifactGenerator{ + ObjectMeta: metav1.ObjectMeta{ + Name: "example", Namespace: "cozy-system", Generation: 1, + CreationTimestamp: metav1.NewTime(referenceTime.Add(-time.Hour)), + Annotations: map[string]string{ + annotationRecoveryAttempts: "1", + annotationLastRecoveryAt: forceAt.UTC().Format(time.RFC3339Nano), + }, + }, + Status: sourcewatcherv1beta1.ArtifactGeneratorStatus{ + Inventory: []sourcewatcherv1beta1.ExternalArtifactReference{ + {Name: "one", Namespace: "cozy-system", Digest: "sha256:aaa"}, + }, + ObservedSourcesDigest: "sha256:0e7", + Conditions: []metav1.Condition{{ + Type: "Ready", + Status: metav1.ConditionFalse, + Reason: reasonRecoveryForced, + Message: "cozystack-operator forced drift", + ObservedGeneration: 1, + LastTransitionTime: metav1.NewTime(forceAt), + }}, + }, + } + c := fake.NewClientBuilder().WithScheme(testScheme(t)).WithStatusSubresource(ps, ag).WithObjects(ps, ag).Build() + agWrites := 0 + watched := interceptor.NewClient(c, interceptor.Funcs{ + Patch: func(ctx context.Context, cli client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + if _, ok := obj.(*sourcewatcherv1beta1.ArtifactGenerator); ok { + agWrites++ + } + return cli.Patch(ctx, obj, patch, opts...) + }, + }) + r := &PackageSourceReconciler{Client: watched, Scheme: testScheme(t)} + + res, err := r.updateStatus(context.Background(), ps, referenceTime) + if err != nil { + t.Fatalf("updateStatus: %v", err) + } + // The routing goes through maybeRecoverArtifactGenerator's Wait branch: + // decideRecovery(attempts=1, lastRecoveryAt=referenceTime-5s, now=referenceTime) + // → elapsed 5s, backoffFor(1)=30s → wait 25s. Exact equality catches + // silent backoff-schedule regressions. + if res.RequeueAfter != 25*time.Second { + t.Errorf("RequeueAfter = %v, want 25s (backoffFor(1)=30s minus 5s elapsed)", res.RequeueAfter) + } + if agWrites != 0 { + t.Errorf("clearRecoveryTracking wrote to AG %d times — tracking must be held while our own marker is visible", agWrites) + } + // AG's recovery-tracking annotations must still be intact. + persistedAG := &sourcewatcherv1beta1.ArtifactGenerator{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(ag), persistedAG); err != nil { + t.Fatalf("Get AG: %v", err) + } + if got := persistedAG.Annotations[annotationRecoveryAttempts]; got != "1" { + t.Errorf("AG %s = %q after own-marker reconcile, want 1 (held)", annotationRecoveryAttempts, got) + } + // PS must be Unknown/AwaitingSourceWatcherRecovery — NOT False/SourceWatcherRecoveryForced. + persistedPS := &cozyv1alpha1.PackageSource{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(ps), persistedPS); err != nil { + t.Fatalf("Get PS: %v", err) + } + psReady := meta.FindStatusCondition(persistedPS.Status.Conditions, "Ready") + if psReady == nil || psReady.Status != metav1.ConditionUnknown || psReady.Reason != reasonAwaitingRecovery { + t.Errorf("PS Ready = %+v, want Unknown/%s — synthetic marker must not leak to PS status", psReady, reasonAwaitingRecovery) + } +} + +// TestUpdateStatus_ProgressingDuringRecoveryHoldsTracking pins lexfrei's B2 +// fix for the second Owns()-re-entrancy path: source-watcher writes +// Ready=Unknown/Progressing after taking the drifted branch and before +// finalising the rebuild. That transition arrives via the Owns() watch, and +// if we cleared tracking on it the attempts counter would never accumulate. +// Assertion: with tracking annotations present (attempts>0) and Ready=Unknown +// visible on the current generation, updateStatus routes through the unified +// own-marker / progressing-in-recovery gate into maybeRecoverArtifactGenerator +// so the state machine holds tracking through Wait instead of clearing it. +func TestUpdateStatus_ProgressingDuringRecoveryHoldsTracking(t *testing.T) { + ps := &cozyv1alpha1.PackageSource{ + ObjectMeta: metav1.ObjectMeta{Name: "example", Namespace: "cozy-system", Generation: 1}, + Spec: cozyv1alpha1.PackageSourceSpec{ + SourceRef: &cozyv1alpha1.PackageSourceRef{Name: "src", Kind: "OCIRepository", Namespace: "cozy-system"}, + }, + } + forceAt := referenceTime.Add(-10 * time.Second) + progressingAt := referenceTime.Add(-3 * time.Second) // Newer than forceAt + ag := &sourcewatcherv1beta1.ArtifactGenerator{ + ObjectMeta: metav1.ObjectMeta{ + Name: "example", Namespace: "cozy-system", Generation: 1, + CreationTimestamp: metav1.NewTime(referenceTime.Add(-time.Hour)), + Annotations: map[string]string{ + annotationRecoveryAttempts: "2", + annotationLastRecoveryAt: forceAt.UTC().Format(time.RFC3339Nano), + }, + }, + Status: sourcewatcherv1beta1.ArtifactGeneratorStatus{ + Inventory: []sourcewatcherv1beta1.ExternalArtifactReference{ + {Name: "one", Namespace: "cozy-system", Digest: "sha256:aaa"}, + }, + ObservedSourcesDigest: "sha256:0e7", + Conditions: []metav1.Condition{{ + Type: "Ready", + Status: metav1.ConditionUnknown, + Reason: "Progressing", + Message: "Reconciliation in progress", + ObservedGeneration: 1, + LastTransitionTime: metav1.NewTime(progressingAt), + }}, + }, + } + c := fake.NewClientBuilder().WithScheme(testScheme(t)).WithStatusSubresource(ps, ag).WithObjects(ps, ag).Build() + agWrites := 0 + watched := interceptor.NewClient(c, interceptor.Funcs{ + Patch: func(ctx context.Context, cli client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + if _, ok := obj.(*sourcewatcherv1beta1.ArtifactGenerator); ok { + agWrites++ + } + return cli.Patch(ctx, obj, patch, opts...) + }, + }) + r := &PackageSourceReconciler{Client: watched, Scheme: testScheme(t)} + + if _, err := r.updateStatus(context.Background(), ps, referenceTime); err != nil { + t.Fatalf("updateStatus: %v", err) + } + if agWrites != 0 { + t.Errorf("tracking cleared during recovery Progressing — attempts=2 wrote to AG %d times, want 0", agWrites) + } + persistedAG := &sourcewatcherv1beta1.ArtifactGenerator{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(ag), persistedAG); err != nil { + t.Fatalf("Get AG: %v", err) + } + if got := persistedAG.Annotations[annotationRecoveryAttempts]; got != "2" { + t.Errorf("AG %s = %q, want 2 (held through Progressing)", annotationRecoveryAttempts, got) + } +} + +// TestUpdateStatus_ReadyTrueClearsTracking pins the canonical reset: once +// source-watcher writes Ready=True (recovery succeeded), the not-stuck path +// clears the recovery-tracking annotations and copies the real condition +// through. This is the only counter reset that remains after B3. +func TestUpdateStatus_ReadyTrueClearsTracking(t *testing.T) { + ps := &cozyv1alpha1.PackageSource{ + ObjectMeta: metav1.ObjectMeta{Name: "example", Namespace: "cozy-system", Generation: 1}, + Spec: cozyv1alpha1.PackageSourceSpec{ + SourceRef: &cozyv1alpha1.PackageSourceRef{Name: "src", Kind: "OCIRepository", Namespace: "cozy-system"}, + }, + } + ag := &sourcewatcherv1beta1.ArtifactGenerator{ + ObjectMeta: metav1.ObjectMeta{ + Name: "example", Namespace: "cozy-system", Generation: 1, + CreationTimestamp: metav1.NewTime(referenceTime.Add(-time.Hour)), + Annotations: map[string]string{ + annotationRecoveryAttempts: "3", + annotationLastRecoveryAt: referenceTime.Add(-time.Minute).UTC().Format(time.RFC3339Nano), + }, + }, + Status: sourcewatcherv1beta1.ArtifactGeneratorStatus{ + Inventory: []sourcewatcherv1beta1.ExternalArtifactReference{ + {Name: "one", Namespace: "cozy-system", Digest: "sha256:aaa"}, + }, + ObservedSourcesDigest: "sha256:0e7", + Conditions: []metav1.Condition{{ + Type: "Ready", + Status: metav1.ConditionTrue, + Reason: "Succeeded", + ObservedGeneration: 1, + LastTransitionTime: metav1.NewTime(referenceTime.Add(-time.Second)), + }}, + }, + } + c := fake.NewClientBuilder().WithScheme(testScheme(t)).WithStatusSubresource(ps, ag).WithObjects(ps, ag).Build() + r := &PackageSourceReconciler{Client: c, Scheme: testScheme(t)} + + if _, err := r.updateStatus(context.Background(), ps, referenceTime); err != nil { + t.Fatalf("updateStatus: %v", err) + } + persistedAG := &sourcewatcherv1beta1.ArtifactGenerator{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(ag), persistedAG); err != nil { + t.Fatalf("Get AG: %v", err) + } + if _, ok := persistedAG.Annotations[annotationRecoveryAttempts]; ok { + t.Errorf("recovery-attempts annotation still present after Ready=True: %v", persistedAG.Annotations) + } + persistedPS := &cozyv1alpha1.PackageSource{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(ps), persistedPS); err != nil { + t.Fatalf("Get PS: %v", err) + } + psReady := meta.FindStatusCondition(persistedPS.Status.Conditions, "Ready") + if psReady == nil || psReady.Status != metav1.ConditionTrue { + t.Errorf("PS Ready = %+v, want True (copied through from AG)", psReady) + } +} + +// TestUpdateStatus_OwnMarkerStale_TriggersRetry regression-tests the +// source-watcher-unresponsive path that an earlier iteration of this driver +// tripped on: after we force-drift, source-watcher never enqueues (network +// partition, missing predicate, pod dead), and our previous force's marker +// keeps reflecting back via Owns(). If updateStatus routed that into a +// perpetual "await" helper the recovery would tight-loop forever and +// SourceWatcherStalled would be unreachable. Asserted here: past +// backoffFor(attempts), the state machine issues another force (bumps +// attempts, re-stamps requestedAt) rather than idling. +func TestUpdateStatus_OwnMarkerStale_TriggersRetry(t *testing.T) { + // attempts=1, lastRecoveryAt=referenceTime-60s. backoffFor(1)=30s, so + // 60s > 30s → decideRecovery must Force again. + forceAt := referenceTime.Add(-60 * time.Second) + ps := &cozyv1alpha1.PackageSource{ + ObjectMeta: metav1.ObjectMeta{Name: "example", Namespace: "cozy-system", Generation: 1}, + Spec: cozyv1alpha1.PackageSourceSpec{ + SourceRef: &cozyv1alpha1.PackageSourceRef{Name: "src", Kind: "OCIRepository", Namespace: "cozy-system"}, + }, + } + ag := &sourcewatcherv1beta1.ArtifactGenerator{ + ObjectMeta: metav1.ObjectMeta{ + Name: "example", Namespace: "cozy-system", Generation: 1, + CreationTimestamp: metav1.NewTime(referenceTime.Add(-time.Hour)), + Annotations: map[string]string{ + annotationRecoveryAttempts: "1", + annotationLastRecoveryAt: forceAt.UTC().Format(time.RFC3339Nano), + annotationFluxRequestedAt: forceAt.UTC().Format(time.RFC3339Nano), + }, + }, + Status: sourcewatcherv1beta1.ArtifactGeneratorStatus{ + Inventory: []sourcewatcherv1beta1.ExternalArtifactReference{ + {Name: "one", Namespace: "cozy-system", Digest: "sha256:aaa"}, + }, + ObservedSourcesDigest: "sha256:0e7", + Conditions: []metav1.Condition{{ + Type: "Ready", + Status: metav1.ConditionFalse, + Reason: reasonRecoveryForced, + Message: "cozystack-operator forced drift (source-watcher never responded)", + ObservedGeneration: 1, + LastTransitionTime: metav1.NewTime(forceAt), + }}, + }, + } + c := fake.NewClientBuilder().WithScheme(testScheme(t)).WithStatusSubresource(ps, ag).WithObjects(ps, ag).Build() + r := &PackageSourceReconciler{Client: c, Scheme: testScheme(t)} + + if _, err := r.updateStatus(context.Background(), ps, referenceTime); err != nil { + t.Fatalf("updateStatus: %v", err) + } + // attempts must have incremented from 1 → 2. + persistedAG := &sourcewatcherv1beta1.ArtifactGenerator{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(ag), persistedAG); err != nil { + t.Fatalf("Get AG: %v", err) + } + if got := persistedAG.Annotations[annotationRecoveryAttempts]; got != "2" { + t.Errorf("AG %s = %q, want 2 — stale own-marker must escalate through Force, not idle", annotationRecoveryAttempts, got) + } + // requestedAt must have been re-stamped to a fresh timestamp so + // source-watcher's next reconcile-request predicate fires. + if got := persistedAG.Annotations[annotationFluxRequestedAt]; got == forceAt.UTC().Format(time.RFC3339Nano) { + t.Errorf("AG %s = %q — must be re-stamped on retry", annotationFluxRequestedAt, got) + } +} + +// TestUpdateStatus_OwnMarkerAttemptsExhausted_GivesUp regression-tests the +// endpoint of the retry loop: with attempts=maxRecoveryAttempts and our own +// marker still reflecting back (source-watcher never reacted through the +// whole budget), updateStatus must surface Ready=False / +// SourceWatcherStalled on the PackageSource instead of looping forever. +func TestUpdateStatus_OwnMarkerAttemptsExhausted_GivesUp(t *testing.T) { + forceAt := referenceTime.Add(-10 * time.Minute) // well past any backoff + ps := &cozyv1alpha1.PackageSource{ + ObjectMeta: metav1.ObjectMeta{Name: "example", Namespace: "cozy-system", Generation: 1}, + Spec: cozyv1alpha1.PackageSourceSpec{ + SourceRef: &cozyv1alpha1.PackageSourceRef{Name: "src", Kind: "OCIRepository", Namespace: "cozy-system"}, + }, + } + ag := &sourcewatcherv1beta1.ArtifactGenerator{ + ObjectMeta: metav1.ObjectMeta{ + Name: "example", Namespace: "cozy-system", Generation: 1, + CreationTimestamp: metav1.NewTime(referenceTime.Add(-time.Hour)), + Annotations: map[string]string{ + annotationRecoveryAttempts: strconv.Itoa(maxRecoveryAttempts), + annotationLastRecoveryAt: forceAt.UTC().Format(time.RFC3339Nano), + }, + }, + Status: sourcewatcherv1beta1.ArtifactGeneratorStatus{ + Inventory: []sourcewatcherv1beta1.ExternalArtifactReference{ + {Name: "one", Namespace: "cozy-system", Digest: "sha256:aaa"}, + }, + ObservedSourcesDigest: "sha256:0e7", + Conditions: []metav1.Condition{{ + Type: "Ready", + Status: metav1.ConditionFalse, + Reason: reasonRecoveryForced, + ObservedGeneration: 1, + LastTransitionTime: metav1.NewTime(forceAt), + }}, + }, + } + c := fake.NewClientBuilder().WithScheme(testScheme(t)).WithStatusSubresource(ps, ag).WithObjects(ps, ag).Build() + r := &PackageSourceReconciler{Client: c, Scheme: testScheme(t)} + + res, err := r.updateStatus(context.Background(), ps, referenceTime) + if err != nil { + t.Fatalf("updateStatus: %v", err) + } + if res.RequeueAfter != 0 { + t.Errorf("RequeueAfter = %v on give-up, want 0 (operator must intervene)", res.RequeueAfter) + } + persistedPS := &cozyv1alpha1.PackageSource{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(ps), persistedPS); err != nil { + t.Fatalf("Get PS: %v", err) + } + psReady := meta.FindStatusCondition(persistedPS.Status.Conditions, "Ready") + if psReady == nil || psReady.Status != metav1.ConditionFalse || psReady.Reason != reasonSourceWatcherBad { + t.Errorf("PS Ready = %+v, want False/%s — unresponsive source-watcher must eventually surface as stalled", psReady, reasonSourceWatcherBad) + } +} + +// TestUpdateStatus_OwnMarkerAttemptsExhaustedButFinalBackoffPending pins +// lexfrei's follow-up on the give-up path: the Owns()-watch re-fire from the +// 5th force-drift's own writes triggers a reconcile milliseconds after the +// force. Without the "hold final backoff window" gate in decideRecovery, +// that re-fire would see `attempts >= maxRecoveryAttempts` and immediately +// return GiveUp, preempting the 4-minute response window the last nudge +// was supposed to grant source-watcher — effective time-to-give-up +// collapses from the documented ~11.5m to ~7.5m. Assertion: with +// attempts=maxRecoveryAttempts AND lastRecoveryAt=now-eps (the exact shape +// of the re-fire on the 5th force's own writes), updateStatus must return +// Wait with RequeueAfter equal to the remaining backoffFor(maxRecoveryAttempts), +// NOT GiveUp. +func TestUpdateStatus_OwnMarkerAttemptsExhaustedButFinalBackoffPending(t *testing.T) { + // Simulate the moment right after the 5th force landed: attempts=max, + // lastRecoveryAt=now-eps, marker still on the AG. + forceAt := referenceTime.Add(-time.Second) + ps := &cozyv1alpha1.PackageSource{ + ObjectMeta: metav1.ObjectMeta{Name: "example", Namespace: "cozy-system", Generation: 1}, + Spec: cozyv1alpha1.PackageSourceSpec{ + SourceRef: &cozyv1alpha1.PackageSourceRef{Name: "src", Kind: "OCIRepository", Namespace: "cozy-system"}, + }, + } + ag := &sourcewatcherv1beta1.ArtifactGenerator{ + ObjectMeta: metav1.ObjectMeta{ + Name: "example", Namespace: "cozy-system", Generation: 1, + CreationTimestamp: metav1.NewTime(referenceTime.Add(-time.Hour)), + Annotations: map[string]string{ + annotationRecoveryAttempts: strconv.Itoa(maxRecoveryAttempts), + annotationLastRecoveryAt: forceAt.UTC().Format(time.RFC3339Nano), + }, + }, + Status: sourcewatcherv1beta1.ArtifactGeneratorStatus{ + Inventory: []sourcewatcherv1beta1.ExternalArtifactReference{ + {Name: "one", Namespace: "cozy-system", Digest: "sha256:aaa"}, + }, + ObservedSourcesDigest: "sha256:0e7", + Conditions: []metav1.Condition{{ + Type: "Ready", + Status: metav1.ConditionFalse, + Reason: reasonRecoveryForced, + ObservedGeneration: 1, + LastTransitionTime: metav1.NewTime(forceAt), + }}, + }, + } + c := fake.NewClientBuilder().WithScheme(testScheme(t)).WithStatusSubresource(ps, ag).WithObjects(ps, ag).Build() + r := &PackageSourceReconciler{Client: c, Scheme: testScheme(t)} + + res, err := r.updateStatus(context.Background(), ps, referenceTime) + if err != nil { + t.Fatalf("updateStatus: %v", err) + } + wantWait := backoffFor(maxRecoveryAttempts) - time.Second + if res.RequeueAfter != wantWait { + t.Errorf("RequeueAfter = %v, want %v — final backoff window must not be preempted", res.RequeueAfter, wantWait) + } + // PS must still be Unknown/AwaitingSourceWatcherRecovery, NOT False/SourceWatcherStalled. + persistedPS := &cozyv1alpha1.PackageSource{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(ps), persistedPS); err != nil { + t.Fatalf("Get PS: %v", err) + } + psReady := meta.FindStatusCondition(persistedPS.Status.Conditions, "Ready") + if psReady == nil || psReady.Status != metav1.ConditionUnknown || psReady.Reason != reasonAwaitingRecovery { + t.Errorf("PS Ready = %+v, want Unknown/%s — give-up must wait for final backoff to elapse", psReady, reasonAwaitingRecovery) + } +}