fix(operator): fence workloads until wipe workers drain

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/GameplanePanel/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), &current); 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, &current); 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)
+			}
+		})
+	}
+}

operator/internal/controller/gameserver_controller.go

@@ -239,6 +239,15 @@ 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.
+	var wipeState wipeWorkerState
+	if changed, err := r.ensureWipeGuard(ctx, &gs, &wipeState); 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,12 +381,12 @@ 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
 	}
-	if err := r.reconcileStatefulSet(ctx, &gs, &tmpl, ver, mc, replicas); err != nil {
+	if err := r.reconcileStatefulSet(ctx, &gs, &tmpl, ver, mc, replicas, &wipeState); err != nil {
 		if apierrors.IsConflict(err) {
 			return ctrl.Result{Requeue: true}, nil
 		}
@@ -397,7 +406,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 {
+	// Share the version-bound preflight/workload inventory with wipe handling.
+	// A live GameServer change can refresh it; an ordinary pass does not rescan.
+	var wipeErr error
+	if gs.Annotations[wipeGuardAnnotation] != "" || pendingWipe(&gs) {
+		wipeErr = r.reconcileWipe(ctx, &gs, &tmpl, &wipeState)
+	} else if wipeState.matches(&gs) && len(wipeState.jobs) != 0 {
+		wipeErr = r.deleteWipeJob(ctx, &gs, gs.Name+"-wipe")
+	}
+	if err := wipeErr; err != nil {
 		if apierrors.IsConflict(err) {
 			return ctrl.Result{Requeue: true}, nil
 		}
@@ -1175,7 +1192,20 @@ 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
+		}
+		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] != "" {
@@ -1387,8 +1417,10 @@ func (r *GameServerReconciler) clearStopAnnotation(ctx context.Context, gs *game
 func (r *GameServerReconciler) reconcileStatefulSet(
 	ctx context.Context, gs *gameplanev1alpha1.GameServer, tmpl *gameplanev1alpha1.GameTemplate,
 	ver *gameplanev1alpha1.GameVersion, mc *materializedConfig, replicas int32,
+	inventory ...*wipeWorkerState,
 ) error {
 	image := resolveImage(gs, tmpl, ver)
+	workers := wipeInventory(inventory)
 
 	// Validate gs.Spec.Env secret and configmap references: any SecretKeyRef
 	// or ConfigMapKeyRef must target a server-owned resource. Referencing
@@ -1408,24 +1440,55 @@ 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")
+		}
+		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
 				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) {
+			if err := r.loadWipeWorkers(ctx, &live, workers); 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
@@ -1895,7 +1958,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

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)})