From d91f0112b5c21b3fabfec1365bb32854d935ba10 Mon Sep 17 00:00:00 2001 From: Baylem Date: Wed, 7 Oct 2026 14:49:11 -0400 Subject: [PATCH 1/9] fix(operator): fence workloads until wipe workers drain Signed-off-by: Baylem --- .../controller/gameserver_controller.go | 72 ++- .../internal/controller/gameserver_wipe.go | 189 +++---- .../controller/gameserver_wipe_guard.go | 276 +++++++++ .../controller/gameserver_wipe_guard_test.go | 534 ++++++++++++++++++ .../gameserver_wipe_ownership_test.go | 1 + .../controller/gameserver_wipe_test.go | 86 +-- .../internal/controller/restore_lifecycle.go | 6 +- operator/specs.md | 2 +- 8 files changed, 987 insertions(+), 179 deletions(-) create mode 100644 operator/internal/controller/gameserver_wipe_guard.go create mode 100644 operator/internal/controller/gameserver_wipe_guard_test.go diff --git a/operator/internal/controller/gameserver_controller.go b/operator/internal/controller/gameserver_controller.go index 0ccb3cf70..0a156cf46 100644 --- a/operator/internal/controller/gameserver_controller.go +++ b/operator/internal/controller/gameserver_controller.go @@ -372,6 +372,13 @@ func (r *GameServerReconciler) Reconcile(ctx context.Context, req ctrl.Request) return ctrl.Result{}, err } + // Persist wipe ownership before any workload mutation, including recovery + // from legacy workers after restart. Acquisition gets its own reconcile. + if changed, err := r.ensureWipeGuard(ctx, &gs); err != nil { + return requeueOnConflict(ctrl.Result{}, err) + } else if changed { + return ctrl.Result{Requeue: true}, nil + } replicas, stopRequeue, err := r.desiredReplicas(ctx, &gs, &tmpl, idle) if err != nil { logger.Error(err, "compute desired replicas") @@ -397,7 +404,15 @@ func (r *GameServerReconciler) Reconcile(ctx context.Context, req ctrl.Request) logger.Error(err, "reconcile BackupSchedule") return ctrl.Result{}, err } - if err := r.reconcileWipe(ctx, &gs, &tmpl); err != nil { + // Preflight already recovered legacy workers. An ordinary pass needs only + // finished-Job cleanup; avoid repeating namespace inventories here. + var wipeErr error + if gs.Annotations[wipeGuardAnnotation] != "" || pendingWipe(&gs) { + wipeErr = r.reconcileWipe(ctx, &gs, &tmpl) + } else { + wipeErr = r.deleteWipeJob(ctx, &gs, gs.Name+"-wipe") + } + if err := wipeErr; err != nil { if apierrors.IsConflict(err) { return ctrl.Result{Requeue: true}, nil } @@ -1176,6 +1191,13 @@ func (r *GameServerReconciler) desiredReplicas( ctx context.Context, gs *gameplanev1alpha1.GameServer, tmpl *gameplanev1alpha1.GameTemplate, idle idleState, ) (int32, time.Duration, error) { + if gs.Annotations[wipeGuardAnnotation] != "" || pendingWipe(gs) { + replicas, poll, err := r.softStop(ctx, gs, tmpl) + if poll == 0 || poll > wipeDrainPoll { + poll = wipeDrainPoll + } + return replicas, poll, err + } // Restore owns the data PVC until its workers have drained. Keep the // workload stopped even if a concurrent power request clears suspend. if gs.Annotations[restoreGuardAnnotation] != "" { @@ -1408,24 +1430,46 @@ func (r *GameServerReconciler) reconcileStatefulSet( "gameplane.local/template": tmpl.Name, } actualReplicas := replicas - if ss.Annotations[restoreGuardAnnotation] != "" { + if ss.Annotations[restoreGuardAnnotation] != "" || ss.Annotations[wipeGuardAnnotation] != "" { actualReplicas = 0 - } else if ss.ResourceVersion == "" && r.APIReader != nil { - // A new StatefulSet can be created from a stale GameServer cache - // snapshot. Consult the live object before starting its first pod. - var live gameplanev1alpha1.GameServer - if err := r.APIReader.Get(ctx, client.ObjectKeyFromObject(gs), &live); err != nil { - return err - } - if live.UID != gs.UID { - return fmt.Errorf("GameServer changed identity before StatefulSet creation") - } - if guard := live.Annotations[restoreGuardAnnotation]; guard != "" { + } + // Both creates and updates can originate from stale cache snapshots. + // Read live ownership even after a marker was cleared for release: a + // conflicting GameServer release must still keep this workload fenced. + var live gameplanev1alpha1.GameServer + if err := r.wipeReader().Get(ctx, client.ObjectKeyFromObject(gs), &live); err != nil { + return err + } + if live.UID != gs.UID { + return fmt.Errorf("GameServer changed identity before StatefulSet mutation") + } + for _, annotation := range []string{restoreGuardAnnotation, wipeGuardAnnotation} { + if guard := live.Annotations[annotation]; guard != "" { actualReplicas = 0 if ss.Annotations == nil { ss.Annotations = make(map[string]string) } - ss.Annotations[restoreGuardAnnotation] = guard + ss.Annotations[annotation] = guard + } + } + if pendingWipe(&live) { + actualReplicas = 0 + } + if live.Annotations[wipeGuardAnnotation] == "" && !pendingWipe(&live) { + workers, err := wipeWorkers(ctx, r.wipeReader(), &live) + if err != nil { + return err + } + if workers.busy { + actualReplicas = 0 + } else { + // A callback may have restamped the marker between workload + // cleanup and the checked GameServer release. Recover that + // interrupted release once the live guard and workers are gone. + delete(ss.Annotations, wipeGuardAnnotation) + if ss.Annotations[restoreGuardAnnotation] == "" { + actualReplicas = replicas + } } } ss.Spec.Replicas = &actualReplicas diff --git a/operator/internal/controller/gameserver_wipe.go b/operator/internal/controller/gameserver_wipe.go index 83f5dcfd1..59c3b0932 100644 --- a/operator/internal/controller/gameserver_wipe.go +++ b/operator/internal/controller/gameserver_wipe.go @@ -4,7 +4,6 @@ import ( "context" "fmt" - appsv1 "k8s.io/api/apps/v1" batchv1 "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" @@ -13,7 +12,6 @@ import ( "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" - "sigs.k8s.io/controller-runtime/pkg/log" gameplanev1alpha1 "github.com/ValgulNecron/gameplane/operator/api/v1alpha1" ) @@ -38,113 +36,109 @@ if [ -n "$left" ]; then fi` ) -// reconcileWipe runs a one-shot Job that empties the GameServer's data PVC -// when a wipe has been requested and the server is suspended (so the game -// pod isn't holding the ReadWriteOnce volume). It's idempotent: the request -// is acked once the Job succeeds and the same token never re-runs. +// reconcileWipe holds exclusive access from guard acquisition through worker +// drain. Reconciles for a GameServer key are serialized; restore/API writers +// arbitrate through the checked GameServer guard update. func (r *GameServerReconciler) reconcileWipe( ctx context.Context, gs *gameplanev1alpha1.GameServer, tmpl *gameplanev1alpha1.GameTemplate, ) error { - // A previous acknowledgement may be newer than this cache snapshot. - // Refresh the request and guard before any wipe side effect, so a stale - // pending token cannot recreate a writer after a restore acquires it. - reader := r.APIReader - if reader == nil { - reader = r.Client - } - var live gameplanev1alpha1.GameServer - if err := reader.Get(ctx, client.ObjectKeyFromObject(gs), &live); err != nil { - return client.IgnoreNotFound(err) - } - if live.UID != gs.UID { + changed, err := r.ensureWipeGuard(ctx, gs) + if apierrors.IsNotFound(err) { return nil } - gs = &live - // A restic restore has exclusive access to this data volume. Do not - // create a wipe Job or acknowledge a wipe by resuming the server yet. + if err != nil { + return err + } if gs.Annotations[restoreGuardAnnotation] != "" { return nil } req := gs.Annotations[WipeRequestedAnnotation] - done := gs.Annotations[WipeCompletedAnnotation] + guard := gs.Annotations[wipeGuardAnnotation] jobName := gs.Name + "-wipe" - - if req == "" || req == done { - // Nothing pending — clean up any finished wipe Job left behind. + state, err := wipeWorkers(ctx, r.wipeReader(), gs) + if err != nil { + return err + } + // Foreign Jobs can never prove completion or authorize cleanup. + var job batchv1.Job + err = r.wipeReader().Get(ctx, types.NamespacedName{Name: jobName, Namespace: gs.Namespace}, &job) + if err != nil && !apierrors.IsNotFound(err) { + return err + } + exists := err == nil + if exists && !metav1.IsControlledBy(&job, gs) { + if pendingWipe(gs) { + return fmt.Errorf("wipe Job %s/%s is not controlled by this GameServer", gs.Namespace, jobName) + } + exists = false + } + if changed { + return nil + } + if guard == "" { return r.deleteWipeJob(ctx, gs, jobName) } - - // Only wipe while suspended; otherwise the game pod still mounts the - // volume. The API sets suspend=true when requesting a wipe. - if !gs.Spec.Suspend { - log.FromContext(ctx).Info("data wipe requested but server not suspended; waiting", "server", gs.Name) + if !pendingWipe(gs) || req != guard || (exists && job.Labels[wipeTokenLabel] != guard) { + // Foreground deletion cannot hide independently inventoried pods. + for _, old := range state.jobs { + if old.DeletionTimestamp.IsZero() { + if err := r.deleteWipeJob(ctx, gs, old.Name); err != nil { + return err + } + } + } + if len(state.jobs) != 0 || state.livePods { + return nil + } + return r.releaseWipeGuard(ctx, gs, guard, false) + } + if gs.Annotations[wipeSuccessAnnotation] == guard { + // Success is durable before Job deletion, and acknowledgment waits + // for foreground cleanup, including terminal pods now terminating. + if exists && job.DeletionTimestamp.IsZero() { + return r.deleteWipeJob(ctx, gs, jobName) + } + if len(state.jobs) != 0 || state.livePods { + return nil + } + return r.ackWipe(ctx, gs, guard) + } + if exists && !job.DeletionTimestamp.IsZero() { return nil } - - // Wait for the pod to actually be gone (Status.Replicas == 0) before mounting - // the PVC. A slowly-stopping game would still hold the ReadWriteOnce volume - // if the Job ran during the graceful stop, causing the Job to fail permanently. - var ss appsv1.StatefulSet - switch err := r.Get(ctx, types.NamespacedName{Namespace: gs.Namespace, Name: gs.Name}, &ss); { - case apierrors.IsNotFound(err): - // StatefulSet is gone — pod is definitely gone. - case err != nil: - return err - case ss.Status.Replicas > 0: - // Pod still present — requeue and wait for scale-down to complete. - log.FromContext(ctx).Info("data wipe requested but pod still draining; waiting", "server", gs.Name) + if exists { + if job.Status.Succeeded > 0 { + return r.recordWipeSuccess(ctx, gs, guard) + } + if jobPermanentlyFailed(&job) { + // Retain guard, failed Job and logs even after a direct start request. + if !gs.Spec.Suspend { + gs.Spec.Suspend = true + if err := r.Update(ctx, gs); err != nil { + return err + } + } + return r.setWipeFailed(ctx, gs) + } return nil } - - // Also verify the game pod object itself is gone, not just scaled down — - // a pod in Terminating state still holds the ReadWriteOnce PVC. - var pod corev1.Pod - podName := types.NamespacedName{Namespace: gs.Namespace, Name: gs.Name + "-0"} - switch err := r.Get(ctx, podName, &pod); { - case apierrors.IsNotFound(err): - // Pod is definitely gone. - case err != nil: - return err - default: - // Pod still exists (including Terminating state) — requeue. - log.FromContext(ctx).Info("data wipe requested but pod still terminating; waiting", "server", gs.Name) + if state.busy { return nil } - - var job batchv1.Job - err := r.Get(ctx, types.NamespacedName{Name: jobName, Namespace: gs.Namespace}, &job) - switch { - case apierrors.IsNotFound(err): - return r.createWipeJob(ctx, gs, tmpl, jobName, req) - case err != nil: + stopped, err := r.wipeTargetStopped(ctx, gs) + if err != nil || !stopped { return err } - - if !metav1.IsControlledBy(&job, gs) { - return fmt.Errorf("wipe Job %s/%s is not controlled by this GameServer", gs.Namespace, jobName) - } - - // A leftover Job from a previous request — replace it. - if job.Labels[wipeTokenLabel] != req { - return r.deleteWipeJob(ctx, gs, jobName) - } - // The current request finished successfully — ack and clean up. - if job.Status.Succeeded > 0 { - if err := r.ackWipe(ctx, gs, req); err != nil { - return err - } - return r.deleteWipeJob(ctx, gs, jobName) + // Revalidate immediately before launch. Another GameServer reconcile for + // this key cannot concurrently release the guard during this launch. + var live gameplanev1alpha1.GameServer + if err := r.wipeReader().Get(ctx, client.ObjectKeyFromObject(gs), &live); err != nil { + return client.IgnoreNotFound(err) } - // The Job exhausted its retries without succeeding (e.g. a permission - // error the wipe container's uid can't get past) — report the failure - // on the GameServer instead of leaving the request silently pending. - // Do not ack and do not delete the Job, so its pod logs stay available - // for the operator to inspect; a new request (a different token) will - // still replace it via the "leftover Job" branch above. - if jobPermanentlyFailed(&job) { - return r.setWipeFailed(ctx, gs) + if live.UID != gs.UID || live.Annotations[wipeGuardAnnotation] != guard || live.Annotations[WipeRequestedAnnotation] != guard || !pendingWipe(&live) || live.Annotations[restoreGuardAnnotation] != "" { + return nil } - return nil + return r.createWipeJob(ctx, &live, tmpl, jobName, guard) } func (r *GameServerReconciler) createWipeJob( @@ -163,6 +157,7 @@ func (r *GameServerReconciler) createWipeJob( Namespace: gs.Namespace, Labels: map[string]string{ wipeTokenLabel: token, + wipeUIDLabel: string(gs.UID), "app.kubernetes.io/managed-by": "gameplane", "app.kubernetes.io/name": gs.Name, }, @@ -170,6 +165,7 @@ func (r *GameServerReconciler) createWipeJob( Spec: batchv1.JobSpec{ BackoffLimit: &backoff, Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{wipeTokenLabel: token, wipeUIDLabel: string(gs.UID)}}, Spec: corev1.PodSpec{ RestartPolicy: corev1.RestartPolicyNever, SecurityContext: &corev1.PodSecurityContext{ @@ -230,19 +226,12 @@ func (r *GameServerReconciler) createWipeJob( } func (r *GameServerReconciler) ackWipe(ctx context.Context, gs *gameplanev1alpha1.GameServer, token string) error { - patch := client.MergeFrom(gs.DeepCopy()) - if gs.Annotations == nil { - gs.Annotations = map[string]string{} - } - gs.Annotations[WipeCompletedAnnotation] = token - // A successful wipe restarts the server on a fresh world. Set suspend=false - // to resume the server after the data has been cleared. - gs.Spec.Suspend = false - if err := r.Patch(ctx, gs, patch); err != nil { + if err := r.releaseWipeGuard(ctx, gs, token, true); err != nil { return err } - // A successful wipe clears a DataWipe=False left by an earlier failed - // request, so the condition only ever describes the latest wipe. + if gs.Annotations[WipeCompletedAnnotation] != token { + return nil + } if meta.FindStatusCondition(gs.Status.Conditions, gameplanev1alpha1.GameServerConditionDataWipe) == nil { return nil } @@ -256,13 +245,13 @@ func (r *GameServerReconciler) ackWipe(ctx context.Context, gs *gameplanev1alpha func (r *GameServerReconciler) deleteWipeJob(ctx context.Context, gs *gameplanev1alpha1.GameServer, name string) error { var job batchv1.Job - if err := r.Get(ctx, types.NamespacedName{Name: name, Namespace: gs.Namespace}, &job); err != nil { + if err := r.wipeReader().Get(ctx, types.NamespacedName{Name: name, Namespace: gs.Namespace}, &job); err != nil { return client.IgnoreNotFound(err) } if !metav1.IsControlledBy(&job, gs) { return nil } - policy := metav1.DeletePropagationBackground + policy := metav1.DeletePropagationForeground uid := job.UID return client.IgnoreNotFound(r.Delete(ctx, &job, &client.DeleteOptions{PropagationPolicy: &policy, Preconditions: &metav1.Preconditions{UID: &uid}})) } diff --git a/operator/internal/controller/gameserver_wipe_guard.go b/operator/internal/controller/gameserver_wipe_guard.go new file mode 100644 index 000000000..a3ab4fa92 --- /dev/null +++ b/operator/internal/controller/gameserver_wipe_guard.go @@ -0,0 +1,276 @@ +package controller + +import ( + "context" + "fmt" + "time" + + appsv1 "k8s.io/api/apps/v1" + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + + gameplanev1alpha1 "github.com/ValgulNecron/gameplane/operator/api/v1alpha1" +) + +const ( + wipeGuardAnnotation = "gameplane.local/wipe-owner" + wipeSuccessAnnotation = "gameplane.local/wipe-succeeded" + wipeUIDLabel = "gameplane.local/wipe-server-uid" + wipeDrainPoll = 5 * time.Second + wipeOrphanGuard = "orphaned-worker" +) + +func pendingWipe(gs *gameplanev1alpha1.GameServer) bool { + return gs.Annotations[WipeRequestedAnnotation] != "" && gs.Annotations[WipeRequestedAnnotation] != gs.Annotations[WipeCompletedAnnotation] +} + +type wipeWorkerState struct { + jobs []*batchv1.Job + livePods bool + busy bool + token string +} + +// Inventory bypasses the cache and keeps terminating/orphan pods visible after +// foreground Job deletion. Labels identify new workers; legacy pods use the +// owned Job UID, or conservatively the fixed Job owner name and target PVC. +// A replacement Job at the same name must not hide old legacy worker pods. +// Labels never authorize deleting a Job. +func wipeWorkers(ctx context.Context, reader client.Reader, gs *gameplanev1alpha1.GameServer) (wipeWorkerState, error) { + state := wipeWorkerState{} + var job batchv1.Job + err := reader.Get(ctx, types.NamespacedName{Namespace: gs.Namespace, Name: gs.Name + "-wipe"}, &job) + if err != nil && !apierrors.IsNotFound(err) { + return state, err + } + ownedUIDs := map[types.UID]bool{} + jobExists := err == nil + if jobExists && metav1.IsControlledBy(&job, gs) { + state.jobs = append(state.jobs, &job) + if job.UID != "" { + ownedUIDs[job.UID] = true + } + if state.token == "" { + state.token = job.Labels[wipeTokenLabel] + } + if !job.DeletionTimestamp.IsZero() || job.Status.Active > 0 || (job.Status.Succeeded == 0 && !jobPermanentlyFailed(&job)) { + state.busy = true + } + } + var pods corev1.PodList + if err := reader.List(ctx, &pods, client.InNamespace(gs.Namespace)); err != nil { + return state, err + } + for i := range pods.Items { + pod := &pods.Items[i] + if !liveRestorePod(pod) { + continue + } + owner := metav1.GetControllerOf(pod) + owned := gs.UID != "" && pod.Labels[wipeUIDLabel] == string(gs.UID) + if owner != nil && owner.Kind == "Job" { + owned = owned || ownedUIDs[owner.UID] || (owner.Name == gs.Name+"-wipe" && pod.Labels[wipeUIDLabel] == "" && podMountsRestoreTarget(pod, gs.Name)) + } + if owned { + state.livePods, state.busy = true, true + if state.token == "" { + state.token = pod.Labels[wipeTokenLabel] + } + } + } + return state, nil +} + +func (r *GameServerReconciler) wipeReader() client.Reader { + if r.APIReader != nil { + return r.APIReader + } + return r.Client +} + +// Called before workload mutation as well as before wipe side effects. Recovery +// from a legacy Job-only/orphan state must fence a start after operator restart. +// The resourceVersion Update arbitrates with restore claims and API writes. +func (r *GameServerReconciler) ensureWipeGuard(ctx context.Context, gs *gameplanev1alpha1.GameServer) (changed bool, err error) { + var live gameplanev1alpha1.GameServer + if err := r.wipeReader().Get(ctx, client.ObjectKeyFromObject(gs), &live); err != nil { + return false, err + } + if live.UID != gs.UID { + return false, fmt.Errorf("GameServer changed identity before wipe acquisition") + } + *gs = live + if gs.Annotations[wipeGuardAnnotation] != "" || gs.Annotations[restoreGuardAnnotation] != "" { + return false, nil + } + state, err := wipeWorkers(ctx, r.wipeReader(), gs) + if err != nil { + return false, err + } + needsGuard := pendingWipe(gs) || state.busy + for _, job := range state.jobs { + // An acknowledged, drained success can be removed without reacquiring. + if job.Status.Succeeded == 0 || job.Labels[wipeTokenLabel] != gs.Annotations[WipeCompletedAnnotation] { + needsGuard = true + } + } + if !needsGuard { + return false, nil + } + token := state.token + if token == "" && pendingWipe(gs) && !state.busy && len(state.jobs) == 0 { + token = gs.Annotations[WipeRequestedAnnotation] + } + if token == "" { + token = wipeOrphanGuard + } + if gs.Annotations == nil { + gs.Annotations = make(map[string]string) + } + gs.Annotations[wipeGuardAnnotation] = token + // Recovery without a pending request is cleanup, including legacy orphan + // pods with no token. The guard fences live workers without changing the + // current resume/stop intent. Pending and failed matching wipes suspend. + if pendingWipe(gs) { + gs.Spec.Suspend = true + } + if err := r.Update(ctx, gs); err != nil { + return false, err + } + return true, nil +} + +// Stamp the owned workload with a checked write before workers launch. A stale +// workload update then conflicts, while callbacks reading the fence stay at 0. +func (r *GameServerReconciler) wipeTargetStopped(ctx context.Context, gs *gameplanev1alpha1.GameServer) (bool, error) { + var ss appsv1.StatefulSet + err := r.wipeReader().Get(ctx, client.ObjectKeyFromObject(gs), &ss) + if apierrors.IsNotFound(err) { + // An absent workload cannot fence an in-flight Create. + return false, nil + } + if err != nil { + return false, err + } + if !metav1.IsControlledBy(&ss, gs) { + return false, fmt.Errorf("wipe target StatefulSet is not controlled by this GameServer") + } + if ss.Annotations[wipeGuardAnnotation] != gs.Annotations[wipeGuardAnnotation] { + if ss.Annotations == nil { + ss.Annotations = make(map[string]string) + } + ss.Annotations[wipeGuardAnnotation] = gs.Annotations[wipeGuardAnnotation] + zero := int32(0) + ss.Spec.Replicas = &zero + return false, r.Update(ctx, &ss) + } + if ss.Spec.Replicas == nil || *ss.Spec.Replicas != 0 || ss.Status.Replicas != 0 || ss.Status.ObservedGeneration < ss.Generation { + return false, nil + } + var pods corev1.PodList + if err := r.wipeReader().List(ctx, &pods, client.InNamespace(gs.Namespace)); err != nil { + return false, err + } + for i := range pods.Items { + pod := &pods.Items[i] + if !liveRestorePod(pod) { + continue + } + owner := metav1.GetControllerOf(pod) + if pod.Name == gs.Name+"-0" || podMountsRestoreTarget(pod, gs.Name) || (owner != nil && ((owner.Kind == "StatefulSet" && owner.Name == gs.Name) || (owner.Kind == "GameServer" && owner.UID == gs.UID))) { + return false, nil + } + } + var jobs batchv1.JobList + if err := r.wipeReader().List(ctx, &jobs, client.InNamespace(gs.Namespace)); err != nil { + return false, err + } + for i := range jobs.Items { + job := &jobs.Items[i] + if !job.DeletionTimestamp.IsZero() || job.Status.Active > 0 || (job.Status.Succeeded == 0 && !jobPermanentlyFailed(job)) { + if podMountsRestoreTarget(&corev1.Pod{Spec: job.Spec.Template.Spec}, gs.Name) { + return false, nil + } + } + } + return true, nil +} + +// Clear the matching workload marker while the GameServer still holds its +// guard. Callbacks always read the live guard, closing a release-conflict gap. +func (r *GameServerReconciler) clearWipeWorkloadGuard(ctx context.Context, gs *gameplanev1alpha1.GameServer, token string) error { + var ss appsv1.StatefulSet + err := r.wipeReader().Get(ctx, client.ObjectKeyFromObject(gs), &ss) + if apierrors.IsNotFound(err) { + return nil + } + if err != nil { + return err + } + if metav1.IsControlledBy(&ss, gs) && ss.Annotations[wipeGuardAnnotation] == token { + delete(ss.Annotations, wipeGuardAnnotation) + return r.Update(ctx, &ss) + } + return nil +} + +func (r *GameServerReconciler) releaseWipeGuard(ctx context.Context, gs *gameplanev1alpha1.GameServer, token string, success bool) error { + var live gameplanev1alpha1.GameServer + if err := r.wipeReader().Get(ctx, client.ObjectKeyFromObject(gs), &live); err != nil { + return client.IgnoreNotFound(err) + } + if live.UID != gs.UID || live.Annotations[wipeGuardAnnotation] != token || live.Annotations[restoreGuardAnnotation] != "" { + return nil + } + if success && (live.Annotations[WipeRequestedAnnotation] != token || !pendingWipe(&live)) { + return nil + } + if success && live.Annotations[wipeSuccessAnnotation] != token { + return nil + } + if err := r.clearWipeWorkloadGuard(ctx, &live, token); err != nil { + return err + } + if success { + live.Annotations[WipeCompletedAnnotation] = token + live.Spec.Suspend = false + delete(live.Annotations, wipeGuardAnnotation) + } else if pendingWipe(&live) { + // Transfer only after old Jobs and live pods have drained; next pass + // fences the workload for this new token before launching a worker. + live.Annotations[wipeGuardAnnotation] = live.Annotations[WipeRequestedAnnotation] + live.Spec.Suspend = true + } else { + delete(live.Annotations, wipeGuardAnnotation) + } + if live.Annotations[wipeSuccessAnnotation] == token { + delete(live.Annotations, wipeSuccessAnnotation) + } + if err := r.Update(ctx, &live); err != nil { + return err + } + *gs = live + return nil +} + +// Persist success before deleting its Job, so restart during foreground drain +// cannot lose success proof and recreate a destructive writer for this token. +func (r *GameServerReconciler) recordWipeSuccess(ctx context.Context, gs *gameplanev1alpha1.GameServer, token string) error { + var live gameplanev1alpha1.GameServer + if err := r.wipeReader().Get(ctx, client.ObjectKeyFromObject(gs), &live); err != nil { + return client.IgnoreNotFound(err) + } + if live.UID != gs.UID || live.Annotations[wipeGuardAnnotation] != token || live.Annotations[WipeRequestedAnnotation] != token || !pendingWipe(&live) || live.Annotations[restoreGuardAnnotation] != "" { + return nil + } + live.Annotations[wipeSuccessAnnotation] = token + if err := r.Update(ctx, &live); err != nil { + return err + } + *gs = live + return nil +} diff --git a/operator/internal/controller/gameserver_wipe_guard_test.go b/operator/internal/controller/gameserver_wipe_guard_test.go new file mode 100644 index 000000000..ae20f24a6 --- /dev/null +++ b/operator/internal/controller/gameserver_wipe_guard_test.go @@ -0,0 +1,534 @@ +package controller + +import ( + "context" + "fmt" + "testing" + + appsv1 "k8s.io/api/apps/v1" + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + + gameplanev1alpha1 "github.com/ValgulNecron/gameplane/operator/api/v1alpha1" +) + +func wipeStoppedSet(gs *gameplanev1alpha1.GameServer) *appsv1.StatefulSet { + zero := int32(0) + return &appsv1.StatefulSet{ + ObjectMeta: metav1.ObjectMeta{Name: gs.Name, Namespace: gs.Namespace, UID: "workload", Generation: 2, + Annotations: map[string]string{wipeGuardAnnotation: gs.Annotations[wipeGuardAnnotation]}, + OwnerReferences: []metav1.OwnerReference{{APIVersion: gameplanev1alpha1.GroupVersion.String(), Kind: "GameServer", Name: gs.Name, UID: gs.UID, Controller: ownerBoolPtr(true)}}}, + Spec: appsv1.StatefulSetSpec{Replicas: &zero}, + Status: appsv1.StatefulSetStatus{ObservedGeneration: 2}, + } +} + +func wipePasses(t *testing.T, r *GameServerReconciler, gs *gameplanev1alpha1.GameServer, count int) { + t.Helper() + for range count { + if err := r.reconcileWipe(t.Context(), gs, &gameplanev1alpha1.GameTemplate{}); err != nil { + t.Fatal(err) + } + } +} + +func wipeCurrent(t *testing.T, c client.Client, gs *gameplanev1alpha1.GameServer) *gameplanev1alpha1.GameServer { + t.Helper() + var current gameplanev1alpha1.GameServer + if err := c.Get(t.Context(), client.ObjectKeyFromObject(gs), ¤t); err != nil { + t.Fatal(err) + } + return ¤t +} + +func TestWipePendingStartAcquiresGuardBeforeLaunching(t *testing.T) { + s := wipeScheme(t) + gs := wipeGameServer(false, "current") + c := fake.NewClientBuilder().WithScheme(s).WithObjects(gs, wipeStoppedSet(gs)).Build() + r := &GameServerReconciler{Client: c, APIReader: c, Scheme: s} + wipePasses(t, r, gs, 1) + current := wipeCurrent(t, c, gs) + if current.Annotations[wipeGuardAnnotation] != "current" || !current.Spec.Suspend { + t.Fatal("direct start bypassed durable wipe acquisition") + } + var job batchv1.Job + if err := c.Get(t.Context(), client.ObjectKey{Namespace: gs.Namespace, Name: gs.Name + "-wipe"}, &job); !apierrors.IsNotFound(err) { + t.Fatal("worker launched in guard acquisition pass") + } + wipePasses(t, r, gs, 2) + if err := c.Get(t.Context(), client.ObjectKey{Namespace: gs.Namespace, Name: gs.Name + "-wipe"}, &job); err != nil { + t.Fatal(err) + } + if job.Spec.Template.Labels[wipeUIDLabel] != string(gs.UID) || job.Spec.Template.Labels[wipeTokenLabel] != "current" { + t.Fatal("worker pod lacks durable identity labels") + } +} + +func TestWipeCancellationAndReplacementWaitForOrphanWorkers(t *testing.T) { + for _, request := range []string{"", "next"} { + for _, legacy := range []bool{false, true} { + t.Run(fmt.Sprintf("request=%q/legacy=%t", request, legacy), func(t *testing.T) { + s := wipeScheme(t) + gs := wipeGameServer(false, request) + gs.Annotations[wipeGuardAnnotation] = "old" + job := &batchv1.Job{ObjectMeta: metav1.ObjectMeta{Name: "alpha-wipe", Namespace: "ns", UID: "worker", Labels: map[string]string{wipeTokenLabel: "old"}}} + if err := controllerutil.SetControllerReference(gs, job, s); err != nil { + t.Fatal(err) + } + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "worker-pod", Namespace: "ns", Finalizers: []string{"test/drain"}, + Labels: map[string]string{wipeUIDLabel: string(gs.UID), wipeTokenLabel: "old"}, + OwnerReferences: []metav1.OwnerReference{{APIVersion: "batch/v1", Kind: "Job", Name: job.Name, UID: job.UID, Controller: ownerBoolPtr(true)}}}, + Spec: corev1.PodSpec{Volumes: []corev1.Volume{{Name: "data", VolumeSource: corev1.VolumeSource{PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ClaimName: "alpha-data"}}}}}, + Status: corev1.PodStatus{Phase: corev1.PodSucceeded}} + if legacy { + pod.Labels = nil + } + foreground := false + c := fake.NewClientBuilder().WithScheme(s).WithObjects(gs, wipeStoppedSet(gs), job, pod).WithInterceptorFuncs(interceptor.Funcs{ + Delete: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.DeleteOption) error { + if _, ok := obj.(*batchv1.Job); ok { + o := &client.DeleteOptions{} + o.ApplyOptions(opts) + foreground = o.PropagationPolicy != nil && *o.PropagationPolicy == metav1.DeletePropagationForeground && o.Preconditions != nil && o.Preconditions.UID != nil && *o.Preconditions.UID == job.UID + } + return c.Delete(ctx, obj, opts...) + }, + }).Build() + r := &GameServerReconciler{Client: c, APIReader: c, Scheme: s} + if err := c.Delete(t.Context(), pod); err != nil { + t.Fatal(err) + } + wipePasses(t, r, gs, 3) + current := wipeCurrent(t, c, gs) + if !foreground || current.Annotations[wipeGuardAnnotation] != "old" || current.Annotations[WipeCompletedAnnotation] != "" { + t.Fatal("old worker was released or acknowledged before termination") + } + replicas, poll, err := r.desiredReplicas(t.Context(), current, &gameplanev1alpha1.GameTemplate{}, idleAwake) + if err != nil || replicas != 0 || poll == 0 { + t.Fatalf("start during drain: replicas=%d poll=%s err=%v", replicas, poll, err) + } + var remaining corev1.Pod + if err := c.Get(t.Context(), client.ObjectKeyFromObject(pod), &remaining); err != nil { + t.Fatal(err) + } + remaining.Finalizers = nil + if err := c.Update(t.Context(), &remaining); err != nil { + t.Fatal(err) + } + wipePasses(t, r, gs, 1) + current = wipeCurrent(t, c, gs) + if current.Annotations[wipeGuardAnnotation] != request || current.Annotations[WipeCompletedAnnotation] != "" { + t.Fatal("drained worker did not release/hand off safely") + } + if request == "" && current.Spec.Suspend { + t.Fatal("cancellation overwrote current start intent") + } + }) + } + } +} + +func TestWipeSuccessWaitsForWorkerAndChecksCurrentRequest(t *testing.T) { + s := wipeScheme(t) + gs := wipeGameServer(false, "old") + gs.Annotations[wipeGuardAnnotation] = "old" + job := &batchv1.Job{ObjectMeta: metav1.ObjectMeta{Name: "alpha-wipe", Namespace: "ns", UID: "worker", Labels: map[string]string{wipeTokenLabel: "old"}}, Status: batchv1.JobStatus{Succeeded: 1}} + if err := controllerutil.SetControllerReference(gs, job, s); err != nil { + t.Fatal(err) + } + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "worker-pod", Namespace: "ns", Labels: map[string]string{wipeUIDLabel: string(gs.UID)}}, Status: corev1.PodStatus{Phase: corev1.PodRunning}} + c := fake.NewClientBuilder().WithScheme(s).WithObjects(gs, wipeStoppedSet(gs), job, pod).Build() + r := &GameServerReconciler{Client: c, APIReader: c, Scheme: s} + wipePasses(t, r, gs, 1) + if wipeCurrent(t, c, gs).Annotations[WipeCompletedAnnotation] != "" { + t.Fatal("success acked while worker live") + } + if err := c.Delete(t.Context(), pod); err != nil { + t.Fatal(err) + } + wipePasses(t, r, gs, 1) // Delete the successful Job after durable proof. + r.Client = interceptor.NewClient(c, interceptor.Funcs{ + Update: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.UpdateOption) error { + if update, ok := obj.(*gameplanev1alpha1.GameServer); ok && update.Annotations[WipeCompletedAnnotation] == "old" { + current := wipeCurrent(t, c, gs) + current.Annotations[WipeRequestedAnnotation] = "next" + if err := c.Update(ctx, current); err != nil { + return err + } + } + return c.Update(ctx, obj, opts...) + }, + }) + if err := r.reconcileWipe(t.Context(), gs, &gameplanev1alpha1.GameTemplate{}); !apierrors.IsConflict(err) { + t.Fatalf("ack race error=%v, want conflict", err) + } + current := wipeCurrent(t, c, gs) + if current.Annotations[WipeRequestedAnnotation] != "next" || current.Annotations[WipeCompletedAnnotation] != "" || current.Annotations[wipeGuardAnnotation] != "old" { + t.Fatal("old success overwrote replacement request") + } +} + +func TestWipeRecoversGuardBeforeLegacyWorkerCanStartGame(t *testing.T) { + s := wipeScheme(t) + gs := wipeGameServer(false, "") + job := &batchv1.Job{ObjectMeta: metav1.ObjectMeta{Name: "alpha-wipe", Namespace: "ns", UID: "worker", Labels: map[string]string{wipeTokenLabel: "old"}}} + if err := controllerutil.SetControllerReference(gs, job, s); err != nil { + t.Fatal(err) + } + c := fake.NewClientBuilder().WithScheme(s).WithObjects(gs, job).Build() + r := &GameServerReconciler{Client: c, APIReader: c, Scheme: s} + if changed, err := r.ensureWipeGuard(t.Context(), gs); err != nil || !changed { + t.Fatalf("legacy writer exclusion recovery: changed=%t err=%v", changed, err) + } + replicas, poll, err := r.desiredReplicas(t.Context(), gs, &gameplanev1alpha1.GameTemplate{}, idleAwake) + if err != nil || replicas != 0 || poll == 0 { + t.Fatalf("legacy writer allowed start: replicas=%d poll=%s err=%v", replicas, poll, err) + } + if wipeCurrent(t, c, gs).Annotations[wipeGuardAnnotation] != "old" { + t.Fatal("legacy writer exclusion was not durable") + } +} + +func TestWipeStatefulSetHonorsLiveGuardAgainstStaleStart(t *testing.T) { + for _, existing := range []bool{false, true} { + t.Run(fmt.Sprintf("existing=%t", existing), func(t *testing.T) { + s := wipeScheme(t) + stale := wipeGameServer(false, "") + live := stale.DeepCopy() + live.Annotations[wipeGuardAnnotation] = "wipe" + objects := []client.Object{live} + if existing { + objects = append(objects, wipeStoppedSet(stale)) + } + c := fake.NewClientBuilder().WithScheme(s).WithObjects(objects...).Build() + r := &GameServerReconciler{Client: c, APIReader: c, Scheme: s} + if err := r.reconcileStatefulSet(t.Context(), stale, &gameplanev1alpha1.GameTemplate{}, nil, &materializedConfig{}, 1); err != nil { + t.Fatal(err) + } + var ss appsv1.StatefulSet + if err := c.Get(t.Context(), client.ObjectKeyFromObject(stale), &ss); err != nil { + t.Fatal(err) + } + if ss.Spec.Replicas == nil || *ss.Spec.Replicas != 0 || ss.Annotations[wipeGuardAnnotation] != "wipe" { + t.Fatal("stale workload callback restarted guarded server") + } + }) + } +} + +func TestRestoreCannotClaimCanceledWipeWithGuardOrOrphan(t *testing.T) { + for _, guarded := range []bool{false, true} { + t.Run(fmt.Sprintf("guarded=%t", guarded), func(t *testing.T) { + r, _, rs := restoreLifecycleFixture(t) + gs := lifecycleServer(t, r) + if gs.Annotations == nil { + gs.Annotations = map[string]string{} + } + if guarded { + gs.Annotations[wipeGuardAnnotation] = "canceled" + } + if err := r.Update(t.Context(), gs); err != nil { + t.Fatal(err) + } + if !guarded { + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "orphan-wipe", Namespace: gs.Namespace, Labels: map[string]string{wipeUIDLabel: string(gs.UID)}}, Status: corev1.PodStatus{Phase: corev1.PodRunning}} + if err := r.Create(t.Context(), pod); err != nil { + t.Fatal(err) + } + } + lifecyclePasses(t, r, "first", 4) + if lifecycleServer(t, r).Annotations[restoreGuardAnnotation] != "" { + t.Fatal("restore claimed canceled wipe before drain") + } + if err := r.Get(t.Context(), client.ObjectKeyFromObject(rs), rs); err != nil { + t.Fatal(err) + } + if rs.Status.Message == "" { + t.Fatal("wipe exclusion wait was not reported") + } + }) + } +} + +func TestWipeLegacyPodSurvivesSameNameJobReplacement(t *testing.T) { + for _, foreign := range []bool{false, true} { + t.Run(fmt.Sprintf("foreign=%t", foreign), func(t *testing.T) { + s := wipeScheme(t) + gs := wipeGameServer(false, "") + gs.Annotations[wipeGuardAnnotation] = "old" + job := &batchv1.Job{ObjectMeta: metav1.ObjectMeta{Name: "alpha-wipe", Namespace: "ns", UID: "replacement", Labels: map[string]string{wipeTokenLabel: "replacement"}}} + if !foreign { + if err := controllerutil.SetControllerReference(gs, job, s); err != nil { + t.Fatal(err) + } + } + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "legacy-worker", Namespace: "ns", + OwnerReferences: []metav1.OwnerReference{{APIVersion: "batch/v1", Kind: "Job", Name: job.Name, UID: "old-job", Controller: ownerBoolPtr(true)}}}, + Spec: corev1.PodSpec{Volumes: []corev1.Volume{{Name: "data", VolumeSource: corev1.VolumeSource{PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ClaimName: "alpha-data"}}}}}, + Status: corev1.PodStatus{Phase: corev1.PodRunning}} + c := fake.NewClientBuilder().WithScheme(s).WithObjects(gs, job, pod).Build() + r := &GameServerReconciler{Client: c, APIReader: c, Scheme: s} + wipePasses(t, r, gs, 3) + if wipeCurrent(t, c, gs).Annotations[wipeGuardAnnotation] != "old" { + t.Fatal("replacement hid surviving legacy worker") + } + var remaining batchv1.Job + if err := c.Get(t.Context(), client.ObjectKeyFromObject(job), &remaining); foreign && (err != nil || remaining.UID != job.UID) { + t.Fatal("foreign replacement Job was removed") + } + }) + } +} + +func TestWipeFailureRetainsGuardSuspensionAndLogs(t *testing.T) { + s := wipeScheme(t) + gs := wipeGameServer(false, "failed") + gs.Annotations[wipeGuardAnnotation] = "failed" + job := &batchv1.Job{ObjectMeta: metav1.ObjectMeta{Name: "alpha-wipe", Namespace: "ns", UID: "failed-worker", Labels: map[string]string{wipeTokenLabel: "failed"}}, + Status: batchv1.JobStatus{Conditions: []batchv1.JobCondition{{Type: batchv1.JobFailed, Status: corev1.ConditionTrue}}}} + if err := controllerutil.SetControllerReference(gs, job, s); err != nil { + t.Fatal(err) + } + c := fake.NewClientBuilder().WithScheme(s).WithObjects(gs, job).WithStatusSubresource(gs).Build() + r := &GameServerReconciler{Client: c, APIReader: c, Scheme: s} + wipePasses(t, r, gs, 2) + current := wipeCurrent(t, c, gs) + if !current.Spec.Suspend || current.Annotations[wipeGuardAnnotation] != "failed" || current.Annotations[WipeCompletedAnnotation] != "" { + t.Fatal("failed wipe was acknowledged or resumed") + } + conditionFound := false + for _, condition := range current.Status.Conditions { + if condition.Type == gameplanev1alpha1.GameServerConditionDataWipe && condition.Status == metav1.ConditionFalse && condition.Reason == "JobFailed" { + conditionFound = true + } + } + if !conditionFound { + t.Fatal("failed wipe did not report DataWipe failure") + } + var retained batchv1.Job + if err := c.Get(t.Context(), client.ObjectKeyFromObject(job), &retained); err != nil || retained.UID != job.UID { + t.Fatal("failed Job/logs were removed") + } +} + +func TestWipeStatefulSetFenceConflictsWithInFlightStaleUpdate(t *testing.T) { + s := wipeScheme(t) + gs := wipeGameServer(false, "") + ss := wipeStoppedSet(gs) + intercepted := false + c := fake.NewClientBuilder().WithScheme(s).WithObjects(gs, ss).WithInterceptorFuncs(interceptor.Funcs{ + Update: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.UpdateOption) error { + if workload, ok := obj.(*appsv1.StatefulSet); ok && workload.Spec.Replicas != nil && *workload.Spec.Replicas == 1 && !intercepted { + intercepted = true + current := wipeCurrent(t, c, gs) + current.Annotations[WipeRequestedAnnotation] = "wipe" + current.Annotations[wipeGuardAnnotation] = "wipe" + current.Spec.Suspend = true + if err := c.Update(ctx, current); err != nil { + return err + } + var fenced appsv1.StatefulSet + if err := c.Get(ctx, client.ObjectKeyFromObject(ss), &fenced); err != nil { + return err + } + fenced.Annotations[wipeGuardAnnotation] = "wipe" + if err := c.Update(ctx, &fenced); err != nil { + return err + } + } + return c.Update(ctx, obj, opts...) + }, + }).Build() + r := &GameServerReconciler{Client: c, APIReader: c, Scheme: s} + if err := r.reconcileStatefulSet(t.Context(), gs, &gameplanev1alpha1.GameTemplate{}, nil, &materializedConfig{}, 1); !apierrors.IsConflict(err) { + t.Fatalf("stale StatefulSet write err=%v, want conflict", err) + } + var remaining appsv1.StatefulSet + if err := c.Get(t.Context(), client.ObjectKeyFromObject(ss), &remaining); err != nil { + t.Fatal(err) + } + if remaining.Annotations[wipeGuardAnnotation] != "wipe" || *remaining.Spec.Replicas != 0 { + t.Fatal("stale update defeated wipe fence") + } +} + +func TestWipeInterruptedReleaseDoesNotLeaveWorkloadStoppedForever(t *testing.T) { + s := wipeScheme(t) + gs := wipeGameServer(false, "finished") + gs.Annotations[WipeCompletedAnnotation] = "finished" + ss := wipeStoppedSet(gs) + ss.Annotations[wipeGuardAnnotation] = "finished" + c := fake.NewClientBuilder().WithScheme(s).WithObjects(gs, ss).Build() + r := &GameServerReconciler{Client: c, APIReader: c, Scheme: s} + if err := r.reconcileStatefulSet(t.Context(), gs, &gameplanev1alpha1.GameTemplate{}, nil, &materializedConfig{}, 1); err != nil { + t.Fatal(err) + } + var current appsv1.StatefulSet + if err := c.Get(t.Context(), client.ObjectKeyFromObject(ss), ¤t); err != nil { + t.Fatal(err) + } + if current.Annotations[wipeGuardAnnotation] != "" || *current.Spec.Replicas != 1 { + t.Fatal("interrupted release marker prevented successful resume") + } +} + +func TestWipeSuccessForegroundDrainSurvivesRestartBeforeAck(t *testing.T) { + s := wipeScheme(t) + gs := wipeGameServer(true, "success") + gs.Annotations[wipeGuardAnnotation] = "success" + job := &batchv1.Job{ObjectMeta: metav1.ObjectMeta{Name: "alpha-wipe", Namespace: "ns", UID: "worker", Finalizers: []string{"test/drain"}, Labels: map[string]string{wipeTokenLabel: "success"}}, Status: batchv1.JobStatus{Succeeded: 1}} + if err := controllerutil.SetControllerReference(gs, job, s); err != nil { + t.Fatal(err) + } + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "worker-pod", Namespace: "ns", Finalizers: []string{"test/drain"}, Labels: map[string]string{wipeUIDLabel: string(gs.UID), wipeTokenLabel: "success"}}, Status: corev1.PodStatus{Phase: corev1.PodSucceeded}} + c := fake.NewClientBuilder().WithScheme(s).WithObjects(gs, wipeStoppedSet(gs), job, pod).Build() + r := &GameServerReconciler{Client: c, APIReader: c, Scheme: s} + wipePasses(t, r, gs, 2) // Persist success, then foreground-delete its Job. + current := wipeCurrent(t, c, gs) + if current.Annotations[wipeSuccessAnnotation] != "success" || current.Annotations[WipeCompletedAnnotation] != "" || !current.Spec.Suspend { + t.Fatal("success was acknowledged before foreground drain") + } + if err := c.Delete(t.Context(), pod); err != nil { + t.Fatal(err) + } + var deletingJob batchv1.Job + if err := c.Get(t.Context(), client.ObjectKeyFromObject(job), &deletingJob); err != nil { + t.Fatal(err) + } + if deletingJob.DeletionTimestamp.IsZero() { + t.Fatal("foreground Job was not retained during drain") + } + deletingJob.Finalizers = nil + if err := c.Update(t.Context(), &deletingJob); err != nil { + t.Fatal(err) + } + // Restart the reconciler with the Job now absent and a terminating terminal + // pod present. Durable success must prevent a replacement destructive Job. + r = &GameServerReconciler{Client: c, APIReader: c, Scheme: s} + wipePasses(t, r, gs, 2) + if wipeCurrent(t, c, gs).Annotations[WipeCompletedAnnotation] != "" { + t.Fatal("terminating success worker was acknowledged") + } + if err := c.Get(t.Context(), client.ObjectKeyFromObject(job), &deletingJob); !apierrors.IsNotFound(err) { + t.Fatal("success proof was lost and worker was recreated") + } + var deletingPod corev1.Pod + if err := c.Get(t.Context(), client.ObjectKeyFromObject(pod), &deletingPod); err != nil { + t.Fatal(err) + } + deletingPod.Finalizers = nil + if err := c.Update(t.Context(), &deletingPod); err != nil { + t.Fatal(err) + } + wipePasses(t, r, gs, 1) + current = wipeCurrent(t, c, gs) + if current.Spec.Suspend || current.Annotations[WipeCompletedAnnotation] != "success" || current.Annotations[wipeGuardAnnotation] != "" || current.Annotations[wipeSuccessAnnotation] != "" { + t.Fatal("drained successful wipe did not acknowledge and resume atomically") + } +} + +func TestWipeAcknowledgedForegroundCleanupPreservesResumeIntent(t *testing.T) { + s := wipeScheme(t) + gs := wipeGameServer(false, "done") + gs.Annotations[WipeCompletedAnnotation] = "done" + deleting := metav1.Now() + job := &batchv1.Job{ObjectMeta: metav1.ObjectMeta{Name: "alpha-wipe", Namespace: "ns", UID: "worker", DeletionTimestamp: &deleting, Finalizers: []string{"test/drain"}, Labels: map[string]string{wipeTokenLabel: "done"}}, Status: batchv1.JobStatus{Succeeded: 1}} + if err := controllerutil.SetControllerReference(gs, job, s); err != nil { + t.Fatal(err) + } + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "worker-pod", Namespace: "ns", DeletionTimestamp: &deleting, Finalizers: []string{"test/drain"}, Labels: map[string]string{wipeUIDLabel: string(gs.UID), wipeTokenLabel: "done"}}, Status: corev1.PodStatus{Phase: corev1.PodSucceeded}} + c := fake.NewClientBuilder().WithScheme(s).WithObjects(gs, job, pod).Build() + r := &GameServerReconciler{Client: c, APIReader: c, Scheme: s} + wipePasses(t, r, gs, 2) + current := wipeCurrent(t, c, gs) + replicas, _, err := r.desiredReplicas(t.Context(), current, &gameplanev1alpha1.GameTemplate{}, idleAwake) + if err != nil || replicas != 0 || current.Annotations[wipeGuardAnnotation] != "done" || current.Spec.Suspend { + t.Fatal("acknowledged cleanup lost resume intent or allowed early start") + } + var drainingJob batchv1.Job + if err := c.Get(t.Context(), client.ObjectKeyFromObject(job), &drainingJob); err != nil { + t.Fatal(err) + } + drainingJob.Finalizers = nil + if err := c.Update(t.Context(), &drainingJob); err != nil { + t.Fatal(err) + } + var drainingPod corev1.Pod + if err := c.Get(t.Context(), client.ObjectKeyFromObject(pod), &drainingPod); err != nil { + t.Fatal(err) + } + drainingPod.Finalizers = nil + if err := c.Update(t.Context(), &drainingPod); err != nil { + t.Fatal(err) + } + wipePasses(t, r, gs, 1) + current = wipeCurrent(t, c, gs) + if current.Spec.Suspend || current.Annotations[wipeGuardAnnotation] != "" || current.Annotations[WipeCompletedAnnotation] != "done" { + t.Fatal("foreground cleanup stranded acknowledged success suspended") + } +} + +func TestWipeAcknowledgedLegacyOrphanPreservesPowerIntent(t *testing.T) { + for _, suspended := range []bool{false, true} { + t.Run(fmt.Sprintf("suspended=%t", suspended), func(t *testing.T) { + s := wipeScheme(t) + gs := wipeGameServer(suspended, "done") + gs.Annotations[WipeCompletedAnnotation] = "done" + deleting := metav1.Now() + // The old implementation did not label PodTemplate workers. Its + // Job is already gone, so neither server UID nor token is available. + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "legacy-worker", Namespace: gs.Namespace, DeletionTimestamp: &deleting, Finalizers: []string{"test/drain"}, + OwnerReferences: []metav1.OwnerReference{{APIVersion: "batch/v1", Kind: "Job", Name: gs.Name + "-wipe", UID: "gone-job", Controller: ownerBoolPtr(true)}}}, + Spec: corev1.PodSpec{Volumes: []corev1.Volume{{Name: "data", VolumeSource: corev1.VolumeSource{PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ClaimName: gs.Name + "-data"}}}}}, + Status: corev1.PodStatus{Phase: corev1.PodSucceeded}} + c := fake.NewClientBuilder().WithScheme(s).WithObjects(gs, wipeStoppedSet(gs), pod).Build() + r := &GameServerReconciler{Client: c, APIReader: c, Scheme: s} + wipePasses(t, r, gs, 2) + current := wipeCurrent(t, c, gs) + replicas, poll, err := r.desiredReplicas(t.Context(), current, &gameplanev1alpha1.GameTemplate{}, idleAwake) + if err != nil || replicas != 0 || poll == 0 || current.Annotations[wipeGuardAnnotation] != wipeOrphanGuard || current.Spec.Suspend != suspended { + t.Fatalf("legacy orphan drain lost power intent or exclusion: replicas=%d poll=%s suspended=%t guard=%q err=%v", replicas, poll, current.Spec.Suspend, current.Annotations[wipeGuardAnnotation], err) + } + // Even a stale start callback must keep the live orphan fence. + if err := r.reconcileStatefulSet(t.Context(), gs, &gameplanev1alpha1.GameTemplate{}, nil, &materializedConfig{}, 1); err != nil { + t.Fatal(err) + } + var ss appsv1.StatefulSet + if err := c.Get(t.Context(), client.ObjectKeyFromObject(gs), &ss); err != nil { + t.Fatal(err) + } + if ss.Spec.Replicas == nil || *ss.Spec.Replicas != 0 || ss.Annotations[wipeGuardAnnotation] != wipeOrphanGuard { + t.Fatal("legacy orphan allowed premature workload resume") + } + var remaining corev1.Pod + if err := c.Get(t.Context(), client.ObjectKeyFromObject(pod), &remaining); err != nil { + t.Fatal(err) + } + remaining.Finalizers = nil + if err := c.Update(t.Context(), &remaining); err != nil { + t.Fatal(err) + } + wipePasses(t, r, gs, 1) + current = wipeCurrent(t, c, gs) + if current.Spec.Suspend != suspended || current.Annotations[wipeGuardAnnotation] != "" || current.Annotations[WipeCompletedAnnotation] != "done" { + t.Fatal("drained legacy orphan changed power intent or retained its guard") + } + if err := c.Get(t.Context(), client.ObjectKeyFromObject(gs), &ss); err != nil { + t.Fatal(err) + } + if ss.Annotations[wipeGuardAnnotation] != "" { + t.Fatal("drained legacy orphan retained workload guard") + } + var job batchv1.Job + if err := c.Get(t.Context(), client.ObjectKey{Name: gs.Name + "-wipe", Namespace: gs.Namespace}, &job); !apierrors.IsNotFound(err) { + t.Fatal("acknowledged legacy cleanup recreated wipe worker") + } + }) + } +} diff --git a/operator/internal/controller/gameserver_wipe_ownership_test.go b/operator/internal/controller/gameserver_wipe_ownership_test.go index cbbd9af6d..4c225a51c 100644 --- a/operator/internal/controller/gameserver_wipe_ownership_test.go +++ b/operator/internal/controller/gameserver_wipe_ownership_test.go @@ -72,6 +72,7 @@ func TestWipePreservesForeignJobs(t *testing.T) { func TestWipeCleanupPreservesJobReplacedDuringDelete(t *testing.T) { scheme := wipeScheme(t) gs := wipeGameServer(false, "") + gs.Annotations[wipeGuardAnnotation] = wipeOrphanGuard job := &batchv1.Job{ObjectMeta: metav1.ObjectMeta{Name: gs.Name + "-wipe", Namespace: gs.Namespace, UID: "owned-job"}} if err := controllerutil.SetControllerReference(gs, job, scheme); err != nil { t.Fatal(err) diff --git a/operator/internal/controller/gameserver_wipe_test.go b/operator/internal/controller/gameserver_wipe_test.go index 2f16a4870..492cabd9b 100644 --- a/operator/internal/controller/gameserver_wipe_test.go +++ b/operator/internal/controller/gameserver_wipe_test.go @@ -49,21 +49,14 @@ func wipeGameServer(suspend bool, req string) *gameplanev1alpha1.GameServer { func TestReconcileWipe_CreatesJobWhenSuspended(t *testing.T) { s := wipeScheme(t) gs := wipeGameServer(true, "tok1") - ss := &appsv1.StatefulSet{ - ObjectMeta: metav1.ObjectMeta{ - Name: "alpha", - Namespace: "ns", - }, - Status: appsv1.StatefulSetStatus{Replicas: 0}, - } + ss := wipeStoppedSet(gs) + ss.Status.Replicas = 0 tmpl := &gameplanev1alpha1.GameTemplate{} tmpl.Name = "mc" cl := fake.NewClientBuilder().WithScheme(s).WithObjects(gs, ss).Build() r := &GameServerReconciler{Client: cl, APIReader: cl, Scheme: s} - if err := r.reconcileWipe(context.Background(), gs, tmpl); err != nil { - t.Fatalf("reconcileWipe: %v", err) - } + wipePasses(t, r, gs, 4) var job batchv1.Job if err := cl.Get(context.Background(), types.NamespacedName{Name: "alpha-wipe", Namespace: "ns"}, &job); err != nil { @@ -77,7 +70,7 @@ func TestReconcileWipe_CreatesJobWhenSuspended(t *testing.T) { } } -func TestReconcileWipe_SkipsWhenNotSuspended(t *testing.T) { +func TestReconcileWipe_AcquiresGuardBeforeWorkloadExists(t *testing.T) { s := wipeScheme(t) gs := wipeGameServer(false, "tok1") tmpl := &gameplanev1alpha1.GameTemplate{} @@ -85,8 +78,10 @@ func TestReconcileWipe_SkipsWhenNotSuspended(t *testing.T) { cl := fake.NewClientBuilder().WithScheme(s).WithObjects(gs).Build() r := &GameServerReconciler{Client: cl, APIReader: cl, Scheme: s} - if err := r.reconcileWipe(context.Background(), gs, tmpl); err != nil { - t.Fatalf("reconcileWipe: %v", err) + wipePasses(t, r, gs, 4) + current := wipeCurrent(t, cl, gs) + if current.Annotations[wipeGuardAnnotation] != "tok1" || !current.Spec.Suspend { + t.Fatal("pending request did not fence direct start") } var job batchv1.Job err := cl.Get(context.Background(), types.NamespacedName{Name: "alpha-wipe", Namespace: "ns"}, &job) @@ -98,13 +93,8 @@ func TestReconcileWipe_SkipsWhenNotSuspended(t *testing.T) { func TestReconcileWipe_AcksWhenJobSucceeded(t *testing.T) { s := wipeScheme(t) gs := wipeGameServer(true, "tok1") - ss := &appsv1.StatefulSet{ - ObjectMeta: metav1.ObjectMeta{ - Name: "alpha", - Namespace: "ns", - }, - Status: appsv1.StatefulSetStatus{Replicas: 0}, - } + ss := wipeStoppedSet(gs) + ss.Status.Replicas = 0 tmpl := &gameplanev1alpha1.GameTemplate{} tmpl.Name = "mc" @@ -121,9 +111,7 @@ func TestReconcileWipe_AcksWhenJobSucceeded(t *testing.T) { } cl := fake.NewClientBuilder().WithScheme(s).WithObjects(gs, ss, job).Build() r := &GameServerReconciler{Client: cl, APIReader: cl, Scheme: s} - if err := r.reconcileWipe(context.Background(), gs, tmpl); err != nil { - t.Fatalf("reconcileWipe: %v", err) - } + wipePasses(t, r, gs, 4) // The request is acked on the GameServer. var got gameplanev1alpha1.GameServer @@ -147,13 +135,8 @@ func TestReconcileWipe_AcksWhenJobSucceeded(t *testing.T) { func TestReconcileWipe_RestartsServerAfterWipe(t *testing.T) { s := wipeScheme(t) gs := wipeGameServer(true, "tok1") - ss := &appsv1.StatefulSet{ - ObjectMeta: metav1.ObjectMeta{ - Name: "alpha", - Namespace: "ns", - }, - Status: appsv1.StatefulSetStatus{Replicas: 0}, - } + ss := wipeStoppedSet(gs) + ss.Status.Replicas = 0 tmpl := &gameplanev1alpha1.GameTemplate{} tmpl.Name = "mc" @@ -170,9 +153,7 @@ func TestReconcileWipe_RestartsServerAfterWipe(t *testing.T) { } cl := fake.NewClientBuilder().WithScheme(s).WithObjects(gs, ss, job).Build() r := &GameServerReconciler{Client: cl, APIReader: cl, Scheme: s} - if err := r.reconcileWipe(context.Background(), gs, tmpl); err != nil { - t.Fatalf("reconcileWipe: %v", err) - } + wipePasses(t, r, gs, 4) // After a successful wipe, suspend must be false to restart the server. var got gameplanev1alpha1.GameServer @@ -194,18 +175,11 @@ func TestReconcileWipe_WaitsForPodToBeGone(t *testing.T) { tmpl.Name = "mc" // StatefulSet exists with replicas > 0 (pod still draining). - ss := &appsv1.StatefulSet{ - ObjectMeta: metav1.ObjectMeta{ - Name: "alpha", - Namespace: "ns", - }, - Status: appsv1.StatefulSetStatus{Replicas: 1}, - } + ss := wipeStoppedSet(gs) + ss.Status.Replicas = 1 cl := fake.NewClientBuilder().WithScheme(s).WithObjects(gs, ss).Build() r := &GameServerReconciler{Client: cl, APIReader: cl, Scheme: s} - if err := r.reconcileWipe(context.Background(), gs, tmpl); err != nil { - t.Fatalf("reconcileWipe: %v", err) - } + wipePasses(t, r, gs, 4) // No wipe job should be created while the pod is still running. var job batchv1.Job @@ -222,13 +196,8 @@ func TestReconcileWipe_WaitsForPodObjectToBeGone(t *testing.T) { tmpl.Name = "mc" // StatefulSet exists with replicas == 0 but pod still terminating. - ss := &appsv1.StatefulSet{ - ObjectMeta: metav1.ObjectMeta{ - Name: "alpha", - Namespace: "ns", - }, - Status: appsv1.StatefulSetStatus{Replicas: 0}, - } + ss := wipeStoppedSet(gs) + ss.Status.Replicas = 0 pod := &corev1.Pod{ ObjectMeta: metav1.ObjectMeta{ Name: "alpha-0", @@ -240,9 +209,7 @@ func TestReconcileWipe_WaitsForPodObjectToBeGone(t *testing.T) { } cl := fake.NewClientBuilder().WithScheme(s).WithObjects(gs, ss, pod).Build() r := &GameServerReconciler{Client: cl, APIReader: cl, Scheme: s} - if err := r.reconcileWipe(context.Background(), gs, tmpl); err != nil { - t.Fatalf("reconcileWipe: %v", err) - } + wipePasses(t, r, gs, 4) // No wipe job should be created while the pod object exists, even with replicas == 0. var job batchv1.Job @@ -259,18 +226,11 @@ func TestReconcileWipe_CreatesJobWhenPodGone(t *testing.T) { tmpl.Name = "mc" // StatefulSet exists with replicas == 0 (pod is gone). - ss := &appsv1.StatefulSet{ - ObjectMeta: metav1.ObjectMeta{ - Name: "alpha", - Namespace: "ns", - }, - Status: appsv1.StatefulSetStatus{Replicas: 0}, - } + ss := wipeStoppedSet(gs) + ss.Status.Replicas = 0 cl := fake.NewClientBuilder().WithScheme(s).WithObjects(gs, ss).Build() r := &GameServerReconciler{Client: cl, APIReader: cl, Scheme: s} - if err := r.reconcileWipe(context.Background(), gs, tmpl); err != nil { - t.Fatalf("reconcileWipe: %v", err) - } + wipePasses(t, r, gs, 4) // Wipe job should be created once the pod is gone. var job batchv1.Job diff --git a/operator/internal/controller/restore_lifecycle.go b/operator/internal/controller/restore_lifecycle.go index c117937e4..150262aea 100644 --- a/operator/internal/controller/restore_lifecycle.go +++ b/operator/internal/controller/restore_lifecycle.go @@ -71,7 +71,11 @@ func (r *RestoreReconciler) bindRestoreTarget(ctx context.Context, rs *gameplane // is written in that same update; a crash cannot leave a lock without suspension. func (r *RestoreReconciler) claimRestoreTarget(ctx context.Context, rs *gameplanev1alpha1.Restore, gs *gameplanev1alpha1.GameServer) (bool, error) { owner := restoreIdentityJSON(rs.Name, rs.UID) - if req := gs.Annotations[WipeRequestedAnnotation]; gs.Annotations[restoreGuardAnnotation] == "" && req != "" && req != gs.Annotations[WipeCompletedAnnotation] { + workers, err := wipeWorkers(ctx, r.apiReader(), gs) + if err != nil { + return false, err + } + if req := gs.Annotations[WipeRequestedAnnotation]; gs.Annotations[wipeGuardAnnotation] != "" || workers.busy || (gs.Annotations[restoreGuardAnnotation] == "" && pendingWipe(gs)) { message := "Waiting for pending data wipe to finish before restoring" var job batchv1.Job err := r.apiReader().Get(ctx, types.NamespacedName{Namespace: gs.Namespace, Name: gs.Name + "-wipe"}, &job) diff --git a/operator/specs.md b/operator/specs.md index 8ca6058d4..3d2066af8 100644 --- a/operator/specs.md +++ b/operator/specs.md @@ -156,7 +156,7 @@ Primary reconcilers register with the manager in `cmd/main.go` and handle CRD li - `gameserver_modcreds.go`: mount mod credentials Secrets. - `gameserver_rcon.go`: RCON port exposure, lifecycle sequences. - `gameserver_restart.go`: restart action, pod deletion. - - `gameserver_wipe.go`: data wipe sequence. Request idempotency: token-based (request token acked once Job succeeds, same token never re-runs). Wipe gate: requires spec.suspend=true (API sets this); waits for StatefulSet scaled to 0 replicas AND game pod object deleted (pod by name `-0`) before starting the destructive Job. Restart: ackWipe sets spec.suspend=false after successful Job completion to resume the server on the fresh world. Job: runs rm against the existing PVC (never deleted/recreated); carries BackoffLimit 2 and no ActiveDeadlineSeconds. + - `gameserver_wipe.go` / `gameserver_wipe_guard.go`: data wipe sequence. Request idempotency is token-based: the current request is acknowledged only after its Job succeeds and all live/terminating workers drain. A durable `gameplane.local/wipe-owner` guard is acquired with a resourceVersion-checked GameServer update that also sets suspend=true for pending wipes; workload creates and updates honor the live guard even after a direct start request. Before launching, the operator fences the owned StatefulSet, requires requested and observed replicas=0 with observedGeneration current, and waits for game/target-mounting pods and writer Jobs to drain. Cancellation or token replacement foreground-deletes owned Jobs with UID preconditions and retains exclusion until Jobs and independently inventoried orphan/terminating worker pods disappear; a replacement then acquires the guard for its own token. Success is recorded under the guard in `gameplane.local/wipe-succeeded` before foreground cleanup, so restart during drain cannot recreate the worker. Matching success clears the guard/proof, acknowledges the token, and resumes automatically in one checked update after Job and pod cleanup. Recovery of an already-acknowledged success keeps the current power intent while fencing remaining cleanup. Failure retains suspension, the guard, the DataWipe failure condition, and the Job/logs until canceled or replaced. Restore cannot acquire the volume while a wipe guard, pending request, or attributable live worker remains, even if the request annotation was removed. Wipe waits poll at most every five seconds. The Job empties and verifies the existing PVC (never deleted/recreated), carries BackoffLimit 2 and no ActiveDeadlineSeconds, and labels both Job and worker pods with the server UID and request token. - `gameserver_version.go`: track/propagate game version. - `gameserver_node.go`: node affinity, pod anti-affinity. - `gameserver_status.go`: phase computation from StatefulSet/Pod state + agent heartbeat, AddressAssignment condition. From 4fa43cbfa37e9ad69f071e4143a234d4ce72ff17 Mon Sep 17 00:00:00 2001 From: Baylem Date: Wed, 7 Oct 2026 15:03:28 -0400 Subject: [PATCH 2/9] fix: preserve a coherent snapshot during wipe preflight Signed-off-by: Baylem --- .../internal/controller/gameserver_controller.go | 15 ++++++++------- .../controller/gameserver_wipe_guard_test.go | 14 ++++++++++---- 2 files changed, 18 insertions(+), 11 deletions(-) diff --git a/operator/internal/controller/gameserver_controller.go b/operator/internal/controller/gameserver_controller.go index 0a156cf46..df43b6107 100644 --- a/operator/internal/controller/gameserver_controller.go +++ b/operator/internal/controller/gameserver_controller.go @@ -239,6 +239,14 @@ func (r *GameServerReconciler) Reconcile(ctx context.Context, req ctrl.Request) if err := r.Get(ctx, req.NamespacedName, &gs); err != nil { return ctrl.Result{}, client.IgnoreNotFound(err) } + // Refresh and recover wipe ownership before deriving config or idle state. + // A later refresh would mix an older idle calculation with a newer status + // patch base and could overwrite a concurrent idle-clock update. + if changed, err := r.ensureWipeGuard(ctx, &gs); err != nil { + return requeueOnConflict(ctrl.Result{}, err) + } else if changed { + return ctrl.Result{Requeue: true}, nil + } // Resolve the template this GameServer points at. Templates are // cluster-scoped so no namespace is needed. @@ -372,13 +380,6 @@ func (r *GameServerReconciler) Reconcile(ctx context.Context, req ctrl.Request) return ctrl.Result{}, err } - // Persist wipe ownership before any workload mutation, including recovery - // from legacy workers after restart. Acquisition gets its own reconcile. - if changed, err := r.ensureWipeGuard(ctx, &gs); err != nil { - return requeueOnConflict(ctrl.Result{}, err) - } else if changed { - return ctrl.Result{Requeue: true}, nil - } replicas, stopRequeue, err := r.desiredReplicas(ctx, &gs, &tmpl, idle) if err != nil { logger.Error(err, "compute desired replicas") diff --git a/operator/internal/controller/gameserver_wipe_guard_test.go b/operator/internal/controller/gameserver_wipe_guard_test.go index ae20f24a6..741b6e2aa 100644 --- a/operator/internal/controller/gameserver_wipe_guard_test.go +++ b/operator/internal/controller/gameserver_wipe_guard_test.go @@ -156,9 +156,12 @@ func TestWipeSuccessWaitsForWorkerAndChecksCurrentRequest(t *testing.T) { r.Client = interceptor.NewClient(c, interceptor.Funcs{ Update: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.UpdateOption) error { if update, ok := obj.(*gameplanev1alpha1.GameServer); ok && update.Annotations[WipeCompletedAnnotation] == "old" { - current := wipeCurrent(t, c, gs) + var current gameplanev1alpha1.GameServer + if err := c.Get(ctx, client.ObjectKeyFromObject(gs), ¤t); err != nil { + return err + } current.Annotations[WipeRequestedAnnotation] = "next" - if err := c.Update(ctx, current); err != nil { + if err := c.Update(ctx, ¤t); err != nil { return err } } @@ -326,11 +329,14 @@ func TestWipeStatefulSetFenceConflictsWithInFlightStaleUpdate(t *testing.T) { Update: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.UpdateOption) error { if workload, ok := obj.(*appsv1.StatefulSet); ok && workload.Spec.Replicas != nil && *workload.Spec.Replicas == 1 && !intercepted { intercepted = true - current := wipeCurrent(t, c, gs) + var current gameplanev1alpha1.GameServer + if err := c.Get(ctx, client.ObjectKeyFromObject(gs), ¤t); err != nil { + return err + } current.Annotations[WipeRequestedAnnotation] = "wipe" current.Annotations[wipeGuardAnnotation] = "wipe" current.Spec.Suspend = true - if err := c.Update(ctx, current); err != nil { + if err := c.Update(ctx, ¤t); err != nil { return err } var fenced appsv1.StatefulSet From 43f3693e831544c36576cce326dd24e88087932e Mon Sep 17 00:00:00 2001 From: Baylem Date: Wed, 7 Oct 2026 15:10:24 -0400 Subject: [PATCH 3/9] test: register wipe jobs in missing-template fixture Signed-off-by: Baylem --- .../internal/controller/gameserver_template_retry_test.go | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/operator/internal/controller/gameserver_template_retry_test.go b/operator/internal/controller/gameserver_template_retry_test.go index e10070c26..9e36f6fd1 100644 --- a/operator/internal/controller/gameserver_template_retry_test.go +++ b/operator/internal/controller/gameserver_template_retry_test.go @@ -4,6 +4,7 @@ import ( "testing" "time" + batchv1 "k8s.io/api/batch/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" @@ -14,7 +15,11 @@ import ( func TestGameServer_MissingTemplateKeepsBoundedRetry(t *testing.T) { gs := &gameplanev1alpha1.GameServer{ObjectMeta: metav1.ObjectMeta{Name: "waiting", Namespace: "games"}, Spec: gameplanev1alpha1.GameServerSpec{TemplateRef: gameplanev1alpha1.GameTemplateRef{Name: "missing"}}} - c := fake.NewClientBuilder().WithScheme(testScheme(t)).WithObjects(gs).WithStatusSubresource(gs).Build() + scheme := testScheme(t) + if err := batchv1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(gs).WithStatusSubresource(gs).Build() r := &GameServerReconciler{Client: c} for range 2 { result, err := r.Reconcile(t.Context(), ctrl.Request{NamespacedName: client.ObjectKeyFromObject(gs)}) From 54cf527511efb0abb6397cde57140314ef8351a4 Mon Sep 17 00:00:00 2001 From: Baylem Date: Wed, 7 Oct 2026 15:26:50 -0400 Subject: [PATCH 4/9] fix: preserve idle snapshot across capture status writes Signed-off-by: Baylem --- .../gameserver_capture_status_test.go | 85 +++++++++++++++++++ .../controller/gameserver_controller.go | 5 +- 2 files changed, 89 insertions(+), 1 deletion(-) create mode 100644 operator/internal/controller/gameserver_capture_status_test.go diff --git a/operator/internal/controller/gameserver_capture_status_test.go b/operator/internal/controller/gameserver_capture_status_test.go new file mode 100644 index 000000000..f59cfba9a --- /dev/null +++ b/operator/internal/controller/gameserver_capture_status_test.go @@ -0,0 +1,85 @@ +package controller + +import ( + "context" + "fmt" + "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" + + gameplanev1alpha1 "github.com/ValgulNecron/gameplane/operator/api/v1alpha1" +) + +func TestCaptureStatusPatchPreservesConcurrentIdleClock(t *testing.T) { + for _, captureWasReady := range []bool{false, true} { + t.Run(fmt.Sprintf("captureWasReady=%t", captureWasReady), func(t *testing.T) { + gs := wipeGameServer(false, "") + gs.Spec.Idle = enabledIdle(5) + now := metav1.NewTime(time.Now().UTC().Truncate(time.Second)) + oldHeartbeat := metav1.NewTime(now.Add(-time.Second)) + recent := metav1.NewTime(now.Add(-time.Minute)) + backdated := metav1.NewTime(now.Add(-10 * time.Minute)) + players := int32(0) + gs.Status.Phase = gameplanev1alpha1.GameServerPhaseRunning + gs.Status.Agent = &gameplanev1alpha1.AgentStatus{LastHeartbeat: &oldHeartbeat, PlayersOnline: &players} + gs.Status.Idle = &gameplanev1alpha1.IdleStatus{EmptySince: &recent} + gs.Status.Capture = &gameplanev1alpha1.CaptureStatus{Ready: captureWasReady} + ss := wipeStoppedSet(gs) + one := int32(1) + ss.Spec.Replicas = &one + ss.Status.Replicas, ss.Status.ReadyReplicas = 1, 1 + injected := false + c := fake.NewClientBuilder().WithScheme(testScheme(t)).WithObjects(gs, ss).WithStatusSubresource(gs, ss). + WithInterceptorFuncs(interceptor.Funcs{ + SubResourcePatch: func(ctx context.Context, c client.Client, subresource string, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error { + if subresource == "status" && !injected { + injected = true + var current gameplanev1alpha1.GameServer + if err := c.Get(ctx, client.ObjectKeyFromObject(gs), ¤t); err != nil { + return err + } + // Model a clock backdate and fresh agent heartbeat arriving + // after idle derivation but during the capture status write. + current.Status.Idle.EmptySince = &backdated + current.Status.Agent.LastHeartbeat = &now + if err := c.Status().Update(ctx, ¤t); err != nil { + return err + } + } + return c.SubResource(subresource).Patch(ctx, obj, patch, opts...) + }, + }).Build() + r := &GameServerReconciler{Client: c, Scheme: c.Scheme()} + idle, idleStatus, _, scheduleErr, err := r.reconcileIdle(t.Context(), gs) + if err != nil || idle != idleAwake || idleStatus == nil || idleStatus.EmptySince == nil || !idleStatus.EmptySince.Equal(&recent) { + t.Fatalf("initial idle derivation: state=%v err=%v", idle, err) + } + if err := r.reconcileCapture(t.Context(), gs); err != nil { + t.Fatal(err) + } + if _, err := r.reconcileStatus(t.Context(), gs, idle, idleStatus, scheduleErr, tunnelPlan{}, &gameplanev1alpha1.GameTemplate{}, nil, ""); err != nil { + t.Fatal(err) + } + current := wipeCurrent(t, c, gs) + if !injected || current.Status.Idle == nil || current.Status.Idle.EmptySince == nil || !current.Status.Idle.EmptySince.Equal(&backdated) { + t.Fatal("final status merge overwrote the concurrently backdated idle clock") + } + if current.Status.Agent == nil || current.Status.Agent.LastHeartbeat == nil || !current.Status.Agent.LastHeartbeat.Equal(&now) { + t.Fatal("status merge overwrote the fresh agent heartbeat") + } + if current.Status.Capture == nil || current.Status.Capture.Ready { + t.Fatal("disabled capture status was not persisted") + } + // The preserved backdate must drive the next sleep decision, + // rather than silently resetting the server's idle interval. + nextIdle, _, _, _, err := r.reconcileIdle(t.Context(), current) + if err != nil || nextIdle != idleAsleep || current.Annotations[IdleAsleepSinceAnnotation] == "" { + t.Fatalf("backdated clock did not produce sleep: state=%v err=%v", nextIdle, err) + } + }) + } +} diff --git a/operator/internal/controller/gameserver_controller.go b/operator/internal/controller/gameserver_controller.go index df43b6107..5dfb2f4f9 100644 --- a/operator/internal/controller/gameserver_controller.go +++ b/operator/internal/controller/gameserver_controller.go @@ -1940,7 +1940,10 @@ func (r *GameServerReconciler) stopActiveCaptures(ctx context.Context, gs *gamep // function never clobbers fields other reconcile steps concurrently own // (e.g. the agent sidecar's status.agent heartbeat). func (r *GameServerReconciler) patchCaptureStatus(ctx context.Context, gs, base *gameplanev1alpha1.GameServer) error { - if err := r.Status().Patch(ctx, gs, client.MergeFrom(base)); err != nil { + // Patch responses contain the full GameServer, even for a no-op patch. + // Keep the caller's snapshot coherent with its already-derived idle state + // so a fresh response cannot turn an older idle clock into a stale write. + if err := r.Status().Patch(ctx, gs.DeepCopy(), client.MergeFrom(base)); err != nil { return fmt.Errorf("patch capture status for %s: %w", gs.Name, err) } return nil From ffa86c91b45cbbf422f3aa241c9f7c834ce47ec0 Mon Sep 17 00:00:00 2001 From: Baylem Date: Wed, 7 Oct 2026 15:52:11 -0400 Subject: [PATCH 5/9] fix: recover workload guards after restore completion Signed-off-by: Baylem --- .../controller/gameserver_controller.go | 10 + .../controller/restore_release_test.go | 175 ++++++++++++++++++ 2 files changed, 185 insertions(+) create mode 100644 operator/internal/controller/restore_release_test.go diff --git a/operator/internal/controller/gameserver_controller.go b/operator/internal/controller/gameserver_controller.go index 5dfb2f4f9..653429fbe 100644 --- a/operator/internal/controller/gameserver_controller.go +++ b/operator/internal/controller/gameserver_controller.go @@ -1444,6 +1444,16 @@ func (r *GameServerReconciler) reconcileStatefulSet( if live.UID != gs.UID { return fmt.Errorf("GameServer changed identity before StatefulSet mutation") } + if owner := metav1.GetControllerOf(ss); owner != nil && owner.UID != live.UID { + return fmt.Errorf("StatefulSet belongs to a different GameServer identity") + } + if live.Annotations[restoreGuardAnnotation] == "" { + // Restore cleanup can race a callback that restamps the workload + // before the checked GameServer release. That release guarantees + // worker drain, so the same live owner's absent guard permits + // removing the stale marker without changing requested replicas. + delete(ss.Annotations, restoreGuardAnnotation) + } for _, annotation := range []string{restoreGuardAnnotation, wipeGuardAnnotation} { if guard := live.Annotations[annotation]; guard != "" { actualReplicas = 0 diff --git a/operator/internal/controller/restore_release_test.go b/operator/internal/controller/restore_release_test.go new file mode 100644 index 000000000..0e8bd45b3 --- /dev/null +++ b/operator/internal/controller/restore_release_test.go @@ -0,0 +1,175 @@ +package controller + +import ( + "context" + "fmt" + "testing" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" + + gameplanev1alpha1 "github.com/ValgulNecron/gameplane/operator/api/v1alpha1" +) + +func TestRestoreReleaseRecoversRestampedStatefulSetGuard(t *testing.T) { + for _, requested := range []int32{0, 1} { + t.Run(fmt.Sprintf("requested-%d", requested), func(t *testing.T) { + r, _, _ := restoreLifecycleFixture(t) + lifecyclePasses(t, r, "first", 8) + job := lifecycleJobs(t, r)[0] + job.Status.Succeeded = 1 + if err := r.Status().Update(t.Context(), &job); err != nil { + t.Fatal(err) + } + + // Run the workload callback after cleanup removes its marker but + // before the restore controller releases the live GameServer guard. + restamped := false + r.Client = interceptor.NewClient(r.Client.(client.WithWatch), interceptor.Funcs{Update: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.UpdateOption) error { + ss, ok := obj.(*appsv1.StatefulSet) + if !ok || restamped || ss.Annotations[restoreGuardAnnotation] != "" { + return c.Update(ctx, obj, opts...) + } + if err := c.Update(ctx, obj, opts...); err != nil { + return err + } + var gs gameplanev1alpha1.GameServer + if err := c.Get(ctx, client.ObjectKeyFromObject(ss), &gs); err != nil { + return err + } + gr := &GameServerReconciler{Client: c, APIReader: c, Scheme: r.Scheme} + if err := gr.reconcileStatefulSet(ctx, &gs, &gameplanev1alpha1.GameTemplate{}, nil, &materializedConfig{}, 1); err != nil { + return err + } + var fenced appsv1.StatefulSet + if err := c.Get(ctx, client.ObjectKeyFromObject(ss), &fenced); err != nil { + return err + } + if *fenced.Spec.Replicas != 0 || fenced.Annotations[restoreGuardAnnotation] != restoreIdentityJSON("first", "restore-first") { + return fmt.Errorf("workload resumed before live restore guard release") + } + restamped = true + return nil + }}) + lifecyclePasses(t, r, "first", 4) + gs := lifecycleServer(t, r) + if !restamped || gs.Spec.Suspend || gs.Annotations[restoreGuardAnnotation] != "" { + t.Fatal("successful restore did not release its restamped workload fence") + } + var ss appsv1.StatefulSet + if err := r.Get(t.Context(), client.ObjectKeyFromObject(gs), &ss); err != nil { + t.Fatal(err) + } + if ss.Annotations[restoreGuardAnnotation] == "" || *ss.Spec.Replicas != 0 { + t.Fatal("interleaving did not leave the stale marker to recover") + } + gs.Spec.Suspend = requested == 0 + if err := r.Update(t.Context(), gs); err != nil { + t.Fatal(err) + } + gr := &GameServerReconciler{Client: r.Client, APIReader: r.APIReader, Scheme: r.Scheme} + if err := gr.reconcileStatefulSet(t.Context(), gs, &gameplanev1alpha1.GameTemplate{}, nil, &materializedConfig{}, requested); err != nil { + t.Fatal(err) + } + if err := r.Get(t.Context(), client.ObjectKeyFromObject(gs), &ss); err != nil { + t.Fatal(err) + } + if ss.Annotations[restoreGuardAnnotation] != "" || *ss.Spec.Replicas != requested { + t.Fatalf("stale restore fence remains: marker=%q replicas=%d, want %d", ss.Annotations[restoreGuardAnnotation], *ss.Spec.Replicas, requested) + } + }) + } +} + +func TestRestoreStaleWorkloadMarkerPreservesLiveBlockers(t *testing.T) { + for _, blocker := range []string{"replacement-restore", "wipe-guard", "pending-wipe", "wipe-worker"} { + t.Run(blocker, func(t *testing.T) { + r, _, _ := restoreLifecycleFixture(t) + gs := lifecycleServer(t, r) + gs.Spec.Suspend = false + stale := gs.DeepCopy() + gs.Annotations = make(map[string]string) + switch blocker { + case "replacement-restore": + gs.Annotations[restoreGuardAnnotation] = restoreIdentityJSON("second", "restore-second") + case "wipe-guard": + gs.Annotations[wipeGuardAnnotation] = "wipe" + case "pending-wipe": + gs.Annotations[WipeRequestedAnnotation] = "wipe" + case "wipe-worker": + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "wipe-worker", Namespace: gs.Namespace, Labels: map[string]string{wipeUIDLabel: string(gs.UID)}}, Status: corev1.PodStatus{Phase: corev1.PodRunning}} + if err := r.Create(t.Context(), pod); err != nil { + t.Fatal(err) + } + } + if err := r.Update(t.Context(), gs); err != nil { + t.Fatal(err) + } + var ss appsv1.StatefulSet + if err := r.Get(t.Context(), client.ObjectKeyFromObject(gs), &ss); err != nil { + t.Fatal(err) + } + ss.Annotations = map[string]string{restoreGuardAnnotation: restoreIdentityJSON("first", "restore-first")} + if err := r.Update(t.Context(), &ss); err != nil { + t.Fatal(err) + } + gr := &GameServerReconciler{Client: r.Client, APIReader: r.APIReader, Scheme: r.Scheme} + if err := gr.reconcileStatefulSet(t.Context(), stale, &gameplanev1alpha1.GameTemplate{}, nil, &materializedConfig{}, 1); err != nil { + t.Fatal(err) + } + if err := r.Get(t.Context(), client.ObjectKeyFromObject(gs), &ss); err != nil { + t.Fatal(err) + } + if *ss.Spec.Replicas != 0 { + t.Fatal("stale restore-marker recovery bypassed the live blocker") + } + if ss.Annotations[restoreGuardAnnotation] != gs.Annotations[restoreGuardAnnotation] || ss.Annotations[wipeGuardAnnotation] != gs.Annotations[wipeGuardAnnotation] { + t.Fatalf("workload did not adopt live guard ownership: %v", ss.Annotations) + } + }) + } +} + +func TestRestoreStaleWorkloadMarkerRefusesReplacedIdentity(t *testing.T) { + for _, replaced := range []string{"gameserver", "statefulset-owner"} { + t.Run(replaced, func(t *testing.T) { + r, _, _ := restoreLifecycleFixture(t) + gs := lifecycleServer(t, r) + var ss appsv1.StatefulSet + if err := r.Get(t.Context(), client.ObjectKeyFromObject(gs), &ss); err != nil { + t.Fatal(err) + } + ss.Annotations = map[string]string{restoreGuardAnnotation: restoreIdentityJSON("first", "restore-first")} + if replaced == "statefulset-owner" { + ss.OwnerReferences[0].UID = "replacement-uid" + } else { + if err := r.Delete(t.Context(), gs); err != nil { + t.Fatal(err) + } + replacement := gs.DeepCopy() + replacement.ResourceVersion = "" + replacement.UID = "replacement-uid" + if err := r.Create(t.Context(), replacement); err != nil { + t.Fatal(err) + } + } + if err := r.Update(t.Context(), &ss); err != nil { + t.Fatal(err) + } + revision := ss.ResourceVersion + gr := &GameServerReconciler{Client: r.Client, APIReader: r.APIReader, Scheme: r.Scheme} + if err := gr.reconcileStatefulSet(t.Context(), gs, &gameplanev1alpha1.GameTemplate{}, nil, &materializedConfig{}, 1); err == nil { + t.Fatal("stale workload reconcile accepted a replaced identity") + } + if err := r.Get(t.Context(), client.ObjectKeyFromObject(gs), &ss); err != nil { + t.Fatal(err) + } + if ss.ResourceVersion != revision || *ss.Spec.Replicas != 0 || ss.Annotations[restoreGuardAnnotation] != restoreIdentityJSON("first", "restore-first") { + t.Fatal("failed identity check mutated the workload") + } + }) + } +} From 5154eb16f38735e77d082482a9729ae1b0004b75 Mon Sep 17 00:00:00 2001 From: Baylem Date: Wed, 7 Oct 2026 17:48:06 -0400 Subject: [PATCH 6/9] ci: retry checks after runner package installation stalled Signed-off-by: Baylem From 3f972228432db3420bf2001abe5d4535ec7c7d34 Mon Sep 17 00:00:00 2001 From: Baylem Date: Wed, 14 Oct 2026 19:53:34 -0400 Subject: [PATCH 7/9] docs(operator): clarify wipe guard release preconditions Signed-off-by: Baylem --- operator/internal/controller/gameserver_wipe_guard.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/operator/internal/controller/gameserver_wipe_guard.go b/operator/internal/controller/gameserver_wipe_guard.go index a2a30b73a..90b85d18e 100644 --- a/operator/internal/controller/gameserver_wipe_guard.go +++ b/operator/internal/controller/gameserver_wipe_guard.go @@ -258,6 +258,9 @@ func (r *GameServerReconciler) clearWipeWorkloadGuard(ctx context.Context, gs *g return nil } +// releaseWipeGuard requires the caller to have observed all workers drained. +// Success acknowledges the current token using durable success proof; cancellation +// preserves power intent or transfers ownership to a newer pending request. func (r *GameServerReconciler) releaseWipeGuard(ctx context.Context, gs *gameplanev1alpha1.GameServer, token string, success bool) error { var live gameplanev1alpha1.GameServer if err := r.wipeReader().Get(ctx, client.ObjectKeyFromObject(gs), &live); err != nil { From 2cad5d528348e589c6fb0a703a6f06004036fd13 Mon Sep 17 00:00:00 2001 From: Baylem Date: Fri, 16 Oct 2026 01:03:15 -0400 Subject: [PATCH 8/9] test(operator): avoid envtest scheme name collision Signed-off-by: Baylem --- .../internal/controller/gameserver_wipe_inventory_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/operator/internal/controller/gameserver_wipe_inventory_test.go b/operator/internal/controller/gameserver_wipe_inventory_test.go index 6a301bfcd..e4075147d 100644 --- a/operator/internal/controller/gameserver_wipe_inventory_test.go +++ b/operator/internal/controller/gameserver_wipe_inventory_test.go @@ -13,7 +13,7 @@ import ( corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/client-go/kubernetes/scheme" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" "k8s.io/client-go/util/cert" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" @@ -25,7 +25,7 @@ import ( func TestGameServerReconcileInventoriesWipeWorkersOnce(t *testing.T) { s := wipeScheme(t) - if err := scheme.AddToScheme(s); err != nil { + if err := clientgoscheme.AddToScheme(s); err != nil { t.Fatal(err) } gs := wipeGameServer(false, "") From fa23541fa7cb053ae5b30c9322cc2f2b234fd84b Mon Sep 17 00:00:00 2001 From: Baylem Date: Fri, 16 Oct 2026 01:16:20 -0400 Subject: [PATCH 9/9] fix(operator): stop fast polling settled wipe failures Signed-off-by: Baylem --- .../controller/gameserver_controller.go | 8 +- .../controller/gameserver_wipe_guard.go | 21 +++ .../controller/gameserver_wipe_poll_test.go | 147 ++++++++++++++++++ operator/specs.md | 2 +- 4 files changed, 176 insertions(+), 2 deletions(-) create mode 100644 operator/internal/controller/gameserver_wipe_poll_test.go diff --git a/operator/internal/controller/gameserver_controller.go b/operator/internal/controller/gameserver_controller.go index 6b4f39d6d..7b95d3daf 100644 --- a/operator/internal/controller/gameserver_controller.go +++ b/operator/internal/controller/gameserver_controller.go @@ -381,7 +381,7 @@ func (r *GameServerReconciler) Reconcile(ctx context.Context, req ctrl.Request) return ctrl.Result{}, err } - replicas, stopRequeue, err := r.desiredReplicas(ctx, &gs, &tmpl, idle) + replicas, stopRequeue, err := r.desiredReplicas(ctx, &gs, &tmpl, idle, &wipeState) if err != nil { logger.Error(err, "compute desired replicas") return ctrl.Result{}, err @@ -1192,9 +1192,15 @@ func svcPortsFromTemplate( func (r *GameServerReconciler) desiredReplicas( ctx context.Context, gs *gameplanev1alpha1.GameServer, tmpl *gameplanev1alpha1.GameTemplate, idle idleState, + inventory ...*wipeWorkerState, ) (int32, time.Duration, error) { if gs.Annotations[wipeGuardAnnotation] != "" || pendingWipe(gs) { replicas, poll, err := r.softStop(ctx, gs, tmpl) + if err == nil && r.settledWipeFailure(ctx, gs, wipeInventory(inventory)) { + // Retain softStop's own deadline. The owned failed Job and current + // request changes are watched; no worker drain needs a 5s backstop. + return replicas, poll, nil + } if poll == 0 || poll > wipeDrainPoll { poll = wipeDrainPoll } diff --git a/operator/internal/controller/gameserver_wipe_guard.go b/operator/internal/controller/gameserver_wipe_guard.go index 90b85d18e..f86a1969d 100644 --- a/operator/internal/controller/gameserver_wipe_guard.go +++ b/operator/internal/controller/gameserver_wipe_guard.go @@ -9,6 +9,7 @@ import ( batchv1 "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/client" @@ -67,6 +68,26 @@ func wipeInventory(supplied []*wipeWorkerState) *wipeWorkerState { return &wipeWorkerState{} } +// A failure condition is not token-bound: annotation-only replacement requests +// can retain it. Drop the drain timer only with the current owned failed Job +// and a drained inventory; cancellation, success cleanup and unknown reads +// must keep their retry. The authoritative wipe step reports inventory errors. +func (r *GameServerReconciler) settledWipeFailure(ctx context.Context, gs *gameplanev1alpha1.GameServer, state *wipeWorkerState) bool { + condition := meta.FindStatusCondition(gs.Status.Conditions, gameplanev1alpha1.GameServerConditionDataWipe) + guard := gs.Annotations[wipeGuardAnnotation] + if condition == nil || condition.Status != metav1.ConditionFalse || condition.Reason != "JobFailed" || + guard == "" || guard != gs.Annotations[WipeRequestedAnnotation] || !pendingWipe(gs) || + gs.Annotations[wipeSuccessAnnotation] != "" || gs.Annotations[restoreGuardAnnotation] != "" { + return false + } + if err := r.loadWipeWorkers(ctx, gs, state); err != nil { + return false + } + job := state.job + return !state.busy && job != nil && metav1.IsControlledBy(job, gs) && + job.Labels[wipeTokenLabel] == guard && job.Status.Succeeded == 0 && jobPermanentlyFailed(job) +} + // Inventory bypasses the cache and keeps terminating/orphan pods visible after // foreground Job deletion. Labels identify new workers; legacy pods use the // owned Job UID, or conservatively the fixed Job owner name and target PVC. diff --git a/operator/internal/controller/gameserver_wipe_poll_test.go b/operator/internal/controller/gameserver_wipe_poll_test.go new file mode 100644 index 000000000..3f13a0abb --- /dev/null +++ b/operator/internal/controller/gameserver_wipe_poll_test.go @@ -0,0 +1,147 @@ +package controller + +import ( + "context" + "errors" + "testing" + "time" + + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + + gameplanev1alpha1 "github.com/GameplanePanel/gameplane/operator/api/v1alpha1" +) + +func failedWipePollFixture(t *testing.T) (*gameplanev1alpha1.GameServer, *batchv1.Job) { + t.Helper() + gs := wipeGameServer(true, "current") + gs.Annotations[wipeGuardAnnotation] = "current" + gs.Status.Conditions = []metav1.Condition{{Type: gameplanev1alpha1.GameServerConditionDataWipe, Status: metav1.ConditionFalse, Reason: "JobFailed"}} + job := &batchv1.Job{ObjectMeta: metav1.ObjectMeta{Name: gs.Name + "-wipe", Namespace: gs.Namespace, UID: "failed-worker", Labels: map[string]string{wipeTokenLabel: "current"}}, + Status: batchv1.JobStatus{Conditions: []batchv1.JobCondition{{Type: batchv1.JobFailed, Status: corev1.ConditionTrue}}}} + if err := controllerutil.SetControllerReference(gs, job, wipeScheme(t)); err != nil { + t.Fatal(err) + } + return gs, job +} + +func TestWipeFailurePollRequiresCurrentRetainedAndDrainedJob(t *testing.T) { + for _, mode := range []string{"retained failure", "new active request with stale condition", "missing Job", "foreign Job", "old Job token", "cancelled", "replaced", "acknowledged", "deleting Job", "active Job", "live worker", "terminating terminal worker", "success cleanup", "condition not yet reported"} { + t.Run(mode, func(t *testing.T) { + gs, job := failedWipePollFixture(t) + objects := []client.Object{gs} + switch mode { + case "new active request with stale condition": + gs.Annotations[WipeRequestedAnnotation], gs.Annotations[wipeGuardAnnotation] = "replacement", "replacement" + job.Labels[wipeTokenLabel] = "replacement" + job.Status = batchv1.JobStatus{Active: 1} + case "foreign Job": + job.OwnerReferences = nil + case "old Job token": + job.Labels[wipeTokenLabel] = "old" + case "cancelled": + delete(gs.Annotations, WipeRequestedAnnotation) + case "replaced": + gs.Annotations[WipeRequestedAnnotation] = "replacement" + case "acknowledged": + gs.Annotations[WipeCompletedAnnotation] = "current" + case "deleting Job": + deleting := metav1.Now() + job.DeletionTimestamp, job.Finalizers = &deleting, []string{"test/drain"} + case "active Job": + job.Status.Active = 1 + case "live worker", "terminating terminal worker": + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "remaining-worker", Namespace: gs.Namespace, Labels: map[string]string{wipeUIDLabel: string(gs.UID)}}, Status: corev1.PodStatus{Phase: corev1.PodRunning}} + if mode == "terminating terminal worker" { + deleting := metav1.Now() + pod.DeletionTimestamp, pod.Finalizers, pod.Status.Phase = &deleting, []string{"test/drain"}, corev1.PodFailed + } + objects = append(objects, pod) + case "success cleanup": + gs.Annotations[wipeSuccessAnnotation] = "current" + case "condition not yet reported": + gs.Status.Conditions = nil + } + if mode != "missing Job" { + objects = append(objects, job) + } + s := wipeScheme(t) + c := fake.NewClientBuilder().WithScheme(s).WithObjects(objects...).Build() + r := &GameServerReconciler{Client: c, APIReader: c, Scheme: s} + replicas, poll, err := r.desiredReplicas(t.Context(), gs, &gameplanev1alpha1.GameTemplate{}, idleAwake) + if err != nil || replicas != 0 { + t.Fatalf("wipe exclusion changed: replicas=%d err=%v", replicas, err) + } + wantPoll := wipeDrainPoll + if mode == "retained failure" { + wantPoll = 0 + } + if poll != wantPoll { + t.Fatalf("poll=%s want=%s", poll, wantPoll) + } + }) + } +} + +type unavailableWipePollReader struct{ client.Reader } + +func TestWipeFailurePollSharesInventoryWithLifecycle(t *testing.T) { + gs, job := failedWipePollFixture(t) + s := wipeScheme(t) + c := fake.NewClientBuilder().WithScheme(s).WithObjects(gs, job).WithStatusSubresource(gs).Build() + reader := &wipeInventoryReader{Reader: c} + r := &GameServerReconciler{Client: c, APIReader: reader, Scheme: s} + var workers wipeWorkerState + if changed, err := r.ensureWipeGuard(t.Context(), gs, &workers); err != nil || changed { + t.Fatalf("preflight: changed=%t err=%v", changed, err) + } + if _, poll, err := r.desiredReplicas(t.Context(), gs, &gameplanev1alpha1.GameTemplate{}, idleAwake, &workers); err != nil || poll != 0 { + t.Fatalf("settled failure: poll=%s err=%v", poll, err) + } + if err := r.reconcileWipe(t.Context(), gs, &gameplanev1alpha1.GameTemplate{}, &workers); err != nil { + t.Fatal(err) + } + if reader.jobGets != 1 || reader.podLists != 1 { + t.Fatalf("failure polling repeated worker inventory: Job GETs=%d Pod LISTs=%d", reader.jobGets, reader.podLists) + } +} + +func (unavailableWipePollReader) List(context.Context, client.ObjectList, ...client.ListOption) error { + return errors.New("worker inventory unavailable") +} + +func TestWipeFailureInventoryErrorRetainsStopAndPoll(t *testing.T) { + gs, job := failedWipePollFixture(t) + s := wipeScheme(t) + c := fake.NewClientBuilder().WithScheme(s).WithObjects(gs, job).Build() + r := &GameServerReconciler{Client: c, APIReader: unavailableWipePollReader{Reader: c}, Scheme: s} + replicas, poll, err := r.desiredReplicas(t.Context(), gs, &gameplanev1alpha1.GameTemplate{}, idleAwake) + if err != nil || replicas != 0 || poll != wipeDrainPoll { + t.Fatalf("observation error changed stop/retry: replicas=%d poll=%s err=%v", replicas, poll, err) + } +} + +type wipePollAttacher struct{} + +func (wipePollAttacher) Stop(context.Context, string, string, string, []string) error { return nil } + +func TestWipeRetainedFailurePreservesSoftStopDeadline(t *testing.T) { + gs, job := failedWipePollFixture(t) + gs.Annotations[stopRequestedAtAnnotation] = time.Now().UTC().Format(time.RFC3339) + grace := int32(60) + gs.Spec.StopGracePeriodSeconds = &grace + workload := wipeStoppedSet(gs) + workload.Status.ReadyReplicas = 1 + s := wipeScheme(t) + c := fake.NewClientBuilder().WithScheme(s).WithObjects(gs, job, workload).Build() + r := &GameServerReconciler{Client: c, APIReader: c, Scheme: s, PodAttacher: wipePollAttacher{}} + tmpl := &gameplanev1alpha1.GameTemplate{Spec: gameplanev1alpha1.GameTemplateSpec{ConsoleMode: "pty", Capabilities: &gameplanev1alpha1.CapabilitiesSpec{Lifecycle: &gameplanev1alpha1.LifecycleSpec{Stop: []string{"exit"}}}}} + replicas, poll, err := r.desiredReplicas(t.Context(), gs, tmpl, idleAwake) + if err != nil || replicas != 1 || poll <= wipeDrainPoll || poll > time.Duration(grace)*time.Second { + t.Fatalf("soft-stop deadline was dropped or replaced: replicas=%d poll=%s err=%v", replicas, poll, err) + } +} diff --git a/operator/specs.md b/operator/specs.md index a6fa3664a..5346946ea 100644 --- a/operator/specs.md +++ b/operator/specs.md @@ -156,7 +156,7 @@ Primary reconcilers register with the manager in `cmd/main.go` and handle CRD li - `gameserver_modcreds.go`: mount mod credentials Secrets. - `gameserver_rcon.go`: RCON port exposure, lifecycle sequences. - `gameserver_restart.go`: restart action, pod deletion. - - `gameserver_wipe.go` / `gameserver_wipe_guard.go`: data wipe sequence. Request idempotency is token-based: the current request is acknowledged only after its Job succeeds and all live/terminating workers drain. A durable `gameplane.local/wipe-owner` guard is acquired with a resourceVersion-checked GameServer update that also sets suspend=true for pending wipes; workload creates and updates honor the live guard even after a direct start request. Before launching, the operator fences the owned StatefulSet, requires both desired and observed replica counts to be zero with observedGeneration current, and waits for game/target-mounting pods and writer Jobs to drain. Cancellation or token replacement foreground-deletes owned Jobs with UID preconditions and retains exclusion until Jobs and independently inventoried orphan/terminating worker pods disappear; a replacement then acquires the guard for its own token. Success is recorded under the guard in `gameplane.local/wipe-succeeded` before foreground cleanup, so restart during drain cannot recreate the worker. Matching success clears the guard/proof, acknowledges the token, and resumes automatically in one checked update after Job and pod cleanup. Recovery of an already-acknowledged success keeps the current power intent while fencing remaining cleanup. Failure retains suspension, the guard, the DataWipe failure condition, and the Job/logs until canceled or replaced. Restore cannot acquire the volume while a wipe guard, pending request, or attributable live worker remains, even if the request annotation was removed. Wipe waits poll at most every five seconds. The Job empties and verifies the existing PVC (never deleted/recreated), carries BackoffLimit 2 and no ActiveDeadlineSeconds, and labels both Job and worker pods with the server UID and request token. + - `gameserver_wipe.go` / `gameserver_wipe_guard.go`: data wipe sequence. Request idempotency is token-based: the current request is acknowledged only after its Job succeeds and all live/terminating workers drain. A durable `gameplane.local/wipe-owner` guard is acquired with a resourceVersion-checked GameServer update that also sets suspend=true for pending wipes; workload creates and updates honor the live guard even after a direct start request. Before launching, the operator fences the owned StatefulSet, requires both desired and observed replica counts to be zero with observedGeneration current, and waits for game/target-mounting pods and writer Jobs to drain. Cancellation or token replacement foreground-deletes owned Jobs with UID preconditions and retains exclusion until Jobs and independently inventoried orphan/terminating worker pods disappear; a replacement then acquires the guard for its own token. Success is recorded under the guard in `gameplane.local/wipe-succeeded` before foreground cleanup, so restart during drain cannot recreate the worker. Matching success clears the guard/proof, acknowledges the token, and resumes automatically in one checked update after Job and pod cleanup. Recovery of an already-acknowledged success keeps the current power intent while fencing remaining cleanup. Failure retains suspension, the guard, the DataWipe failure condition, and the Job/logs until canceled or replaced. Restore cannot acquire the volume while a wipe guard, pending request, or attributable live worker remains, even if the request annotation was removed. Active wipe and worker-drain waits poll at most every five seconds. A retained failed Job for the current guarded request drops this extra poll only after attributable workers drain; its guard, failure status and logs remain. Owned Job and GameServer watches, the ordinary status refresh and any graceful-stop deadline still apply. The Job empties and verifies the existing PVC (never deleted/recreated), carries BackoffLimit 2 and no ActiveDeadlineSeconds, and labels both Job and worker pods with the server UID and request token. - `gameserver_version.go`: track/propagate game version. - `gameserver_node.go`: node affinity, pod anti-affinity. - `gameserver_status.go`: phase computation from StatefulSet/Pod state + agent heartbeat, AddressAssignment condition.