fix: Wait for the cache to show a StatefulSet shrink before the next pass

CLAUDE.md

@@ -77,6 +77,7 @@ Test philosophy: assert external behavior, never internal call ordering or priva
 - **Logs live on the lib claim** (decided 2026-10-07; the separate `log-storage` claim and its four `log*` knobs are gone): `storage.<role>.fileLogging` (bool, default `true`, a default-on toggle like `configureCorePattern`) pins `--log-file=/var/lib/memgraph/logs/memgraph.log`, off pins `--log-file=""`. Never a path knob and `log_file` stays denied in `spec.flags`: Memgraph's `CleanLogsDir` deletes files older than `log_retention_days` from the log file's **parent directory**, so the file must sit in a directory of its own, never the claim root, and any path off a writable mount crash-loops on the read-only root. The empty value is load-bearing: the image's `memgraph.conf` sets `log_file` to `/var/log/memgraph/memgraph.log`. Toggling is an ordinary roll (no CEL pin, no claim template involved); turning it off leaves old log files on the claim, never cleaned by the operator. Logs count toward `libPVCSize`, which is why the defaults moved to 10Gi and `log_level=INFO`.
 - Both StatefulSets use `updateStrategy: OnDelete`, so **nothing but the operator ever restarts a workload pod**. A pod-template change no reconcile acts on takes effect never, which is what the `Updated` condition exists to report. A run-time flag change is deliberately *not* a pod-template change (see Memgraph flags above). `RollingUpdate` cannot express the required order (it sweeps highest ordinal to lowest, and `partition` is a descending cutoff, not a set), so a MAIN on any ordinal but 0 would be restarted mid-sweep and each such restart buys another failover.
 - The operator never promotes a MAIN outside bootstrap and the scale-down handover: a roll deletes the MAIN's pod and lets the Raft coordinators promote. That is only safe on a Memgraph reporting an unreachable MAIN as `role=main, health=down` — a release that vacates the `main` row makes `planner.Plan` believe there is no MAIN and race the failover.
+- **The shrink is the one write with an expectation** (`internal/expectations`, decided 2026-10-09). The operator reads its StatefulSets from the informer cache, and `replicaCounts` applies `max(declared, cached spec.replicas)`, so a pass reading the StatefulSet from before `shedRetiredPods` lowered it would grow it back (and shrink it again once it found the member gone, the pod recreated in between). `shedRetiredPods` records the UID and generation the apply returned; `replicaCounts` marks a role whose cached StatefulSet is the same UID at a lower generation, and the pass ends before applying anything, waiting on the StatefulSet event (`Owns`). It is in-memory on purpose: a restarted operator lists into a fresh cache that already holds every write it made, so only the writing process can be ahead of its cache. Every other cached read is safe without one (the apply restates desired state, pod deletes carry a UID precondition, `rollout`'s `Stale` covers status, membership is read live over Bolt); a new write whose stale read would undo it, or a decision pushed into Memgraph from a cached count, needs one too.
 - Log messages follow Kubernetes style: capital first letter, no trailing period, past tense, object type named (see AGENTS.md for examples).
 - **External access** (`spec.externalAccess`, `type: LoadBalancer | Gateway`) is operator-owned on both sides: the external objects *and* the announced `bolt_server` of every exposed member. Both modes build the same Services, one shared by the coordinators plus one per data instance (selected by the StatefulSet pod-name label); with `LoadBalancer` they are the way in, with `Gateway` they are ClusterIPs behind one operator-owned Gateway (`gatewayClassName` required, listeners `coordinators-bolt` on 7687 and `data-<n>-bolt` on `dataPortBase + n`, default 9000) fed by one TCPRoute each. The address is discovered, never declared: external-dns hostname annotation (on the Service, or in Gateway mode on the TCPRoute, then the Gateway), then the reported hostname, then the IP, pod DNS until any exists (`Converged=False/ExternalAddressPending`). `{ordinal}` in per-instance annotation values is substituted with the pod ordinal and CEL requires it on a data hostname annotation. Only the bolt address ever leaves the cluster network; the operator itself always dials coordinators on pod DNS (`resources.CoordinatorEndpoints`), never through the LoadBalancer or Gateway. External objects follow the data pods the operator *runs* (applied count), are pruned by label once the spec stops describing them, and removing the block reverts every `bolt_server` to pod DNS. The Gateway API CRDs are never bundled: `main.go` discovers once at start (`controller.GatewayAPIServed`) whether `Gateway` and `TCPRoute` are served at `v1`, watches the kinds only if so, and a cluster asking for `Gateway` without that gets `Converged=False/ApplyFailed` naming what is missing. TCPRoute is `v1` only from Gateway API v1.6 (Envoy Gateway ≥ v1.9); older releases serve it as `v1alpha2` alone and v1.6 stops serving `v1alpha2`, so no single version works on both sides — the operator requires ≥ v1.6. The e2e suite gets LoadBalancer addresses from MetalLB and a Gateway from Envoy Gateway (`hack/kind-metallb.sh`, `hack/kind-envoy-gateway.sh`, run by `make setup-test-e2e`); envtest loads the Gateway API CRDs from the `sigs.k8s.io/gateway-api` module.
 - **TLS** (`spec.tls`) is presence-based per mode and takes one shared `kubernetes.io/tls` Secret per mode for both roles, fixed keys, no key-name knobs: a StatefulSet has one pod template, and Memgraph verifies no member hostname, so per-pod identities buy nothing. `bolt` (`secretName`; `tls.crt`/`tls.key` mounted read-only at `/etc/memgraph/ssl`, `--bolt-cert-file`/`--bolt-key-file`) also moves 9091 to https because Memgraph serves metrics from the Bolt server context. Everything downstream is derived, never a knob: the operator dials `bolt+ssc` (TLS, unverified — verification would force pod-DNS SANs and a CA onto a user-facing cert and buys nothing while Bolt is unauthenticated; Bolt auth is the trigger to revisit) and falls back to the other scheme once, because a roll turning TLS on or off leaves coordinators in the old mode while the spec says the new one (`memgraph.dialSchemes`); the ServiceMonitor switches to https with `insecureSkipVerify`. Probes are TCP and untouched. Rotation is not the operator's: the Secret is mounted without `subPath` so the kubelet refreshes it, and Memgraph picks it up on `RELOAD BOLT SERVER TLS` until the core reloads by mtime (planned upstream); the operator watches no Secret and never restarts for an in-place update, and a `secretName` change is an ordinary roll. `intraCluster` (`secretName`; `tls.crt`/`tls.key`/`ca.crt` mounted read-only at `/etc/memgraph/intra_cluster_tls`, the three `--cluster-*-file` flags always together) is mTLS proving membership, not identity: Memgraph verifies the chain and demands a client cert but checks no hostname, so one shared cert is the shape (nine of thirteen surveyed operators do the same). Its presence is pinned by a CEL transition rule on `MemgraphClusterSpec`, not on `TLSSpec`, because a rule on the block does not fire when the block is added or removed whole: a member with it cannot talk to one without, so a one-pod-at-a-time roll deadlocks on lag. `secretName` changes roll normally; a CA cut-over needs a bundled `ca.crt` in both Secrets, documented, not detected. Renewal is `RELOAD INTRA_CLUSTER TLS` per instance until the core reloads by mtime (Raft already does).

internal/controller/memgraphcluster_controller.go

@@ -52,6 +52,7 @@ import (
 	monitoringv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
 
 	memgraphcomv1alpha1 "github.com/memgraph/kubernetes-operator/api/v1alpha1"
+	"github.com/memgraph/kubernetes-operator/internal/expectations"
 	"github.com/memgraph/kubernetes-operator/internal/memgraph"
 	"github.com/memgraph/kubernetes-operator/internal/planner"
 	"github.com/memgraph/kubernetes-operator/internal/resources"
@@ -130,6 +131,11 @@ type MemgraphClusterReconciler struct {
 	// ServiceMonitorAPIMissing says what the discovery found lacking when
 	// ServiceMonitorAPI is false, so the resource is told what to install.
 	ServiceMonitorAPIMissing string
+
+	// shrinks are the StatefulSet shrinks this process applied and its cache
+	// may not show yet; a pass waits for them before resolving replica
+	// counts (see replicaCounts). The zero value is ready to use.
+	shrinks expectations.StatefulSets
 }
 
 // The install chart's ClusterRole is generated from these markers, so they are
@@ -249,6 +255,13 @@ func (r *MemgraphClusterReconciler) reconcileCluster(
 	if err != nil {
 		return ctrl.Result{}, err
 	}
+	if len(replicas.unseenShrinks) > 0 {
+		// The StatefulSet event that clears it requeues the cluster, so the
+		// delay is only a backstop.
+		log.Info("Deferred the pass until the cache shows the StatefulSet shrink it applied",
+			"statefulsets", replicas.unseenShrinks)
+		return ctrl.Result{RequeueAfter: requeueWhilePending}, nil
+	}
 
 	// A grown storage size goes to the claims before anything is applied: the
 	// apply restates the live claim templates, which Kubernetes refuses to
@@ -685,13 +698,13 @@ func (r *MemgraphClusterReconciler) applyDesired(
 	ctx context.Context,
 	cluster *memgraphcomv1alpha1.MemgraphCluster,
 	desired ...client.Object,
-) (appliedGenerations, error) {
-	applied := appliedGenerations{}
+) (appliedObjects, error) {
+	applied := appliedObjects{}
 	for _, obj := range desired {
 		if err := controllerutil.SetControllerReference(cluster, obj, r.Scheme); err != nil {
 			return nil, fmt.Errorf("setting owner reference on %T %s: %w", obj, obj.GetName(), err)
 		}
-		generation, err := r.apply(ctx, obj)
+		written, err := r.apply(ctx, obj)
 		if err != nil {
 			applyErr := fmt.Errorf("applying %T %s: %w", obj, obj.GetName(), err)
 			msg := truncateMessage(applyErr.Error())
@@ -700,24 +713,31 @@ func (r *MemgraphClusterReconciler) applyDesired(
 				notConvergedCondition(memgraphcomv1alpha1.ReasonApplyFailed, msg))
 			return nil, applyErr
 		}
-		applied[appliedKey(obj)] = generation
+		applied[appliedKey(obj)] = appliedObject{uid: written.GetUID(), generation: written.GetGeneration()}
 	}
 	return applied, nil
 }
 
-// appliedGenerations is the metadata.generation the API server assigned each
-// object a pass applied, keyed by kind and name — two kinds share a name here,
-// the headless Service and the StatefulSet of a role.
-type appliedGenerations map[string]int64
+// appliedObjects is the UID and metadata.generation the API server
+// returned for each object a pass applied, keyed by kind and name — two kinds
+// share a name here, the headless Service and the StatefulSet of a role.
+type appliedObjects map[string]appliedObject
+
+// appliedObject is one applied object as the API server returned it.
+type appliedObject struct {
+	uid        types.UID
+	generation int64
+}
 
 func appliedKey(obj client.Object) string {
 	return obj.GetObjectKind().GroupVersionKind().Kind + "/" + obj.GetName()
 }
 
-// statefulSet is the generation this pass applied to the named StatefulSet, or
-// zero when the pass applied none — which is what a StatefulSet read from the
-// cache is then measured against, and zero is what any generation clears.
-func (a appliedGenerations) statefulSet(name string) int64 {
+// statefulSet is the named StatefulSet as this pass applied it, or the zero
+// value when the pass applied none — whose generation is what a StatefulSet
+// read from the cache is then measured against, and zero is what any
+// generation clears.
+func (a appliedObjects) statefulSet(name string) appliedObject {
 	return a["StatefulSet/"+name]
 }
 
@@ -741,6 +761,10 @@ type roleReplicas struct {
 type replicaCounts struct {
 	coordinators roleReplicas
 	data         roleReplicas
+	// unseenShrinks names the StatefulSets whose shrink this process applied
+	// and its cache does not show yet. While it is non-empty the counts above
+	// are not to be acted on.
+	unseenShrinks []string
 }
 
 // retirementMessage names the members a lowered count is shedding, and is empty
@@ -828,6 +852,12 @@ func (r *MemgraphClusterReconciler) replicaCounts(
 		if err != nil {
 			return replicaCounts{}, err
 		}
+		// A cached StatefulSet older than the shrink a previous pass applied
+		// still carries the count before it, and the max below would apply
+		// that count again: the pods just shed would come back.
+		if !r.shrinks.Satisfied(types.NamespacedName{Namespace: cluster.Namespace, Name: role.name}, sts) {
+			counts.unseenShrinks = append(counts.unseenShrinks, role.name)
+		}
 		// The count the operator's own previous apply left, zero before the
 		// first one.
 		var current int32
@@ -1096,18 +1126,18 @@ func (r *MemgraphClusterReconciler) observeRollout(
 	ctx context.Context,
 	cluster *memgraphcomv1alpha1.MemgraphCluster,
 	replicas replicaCounts,
-	applied appliedGenerations,
+	applied appliedObjects,
 ) (rolloutRoles, error) {
 	var roles rolloutRoles
 	coordinators, err := r.observeRolloutRole(ctx, cluster, replicas.coordinators,
 		resources.CoordinatorPodSelector(cluster), resources.CoordinatorInstanceName,
-		applied.statefulSet(replicas.coordinators.name))
+		applied.statefulSet(replicas.coordinators.name).generation)
 	if err != nil {
 		return rolloutRoles{}, err
 	}
 	data, err := r.observeRolloutRole(ctx, cluster, replicas.data,
 		resources.DataPodSelector(cluster), resources.DataInstanceName,
-		applied.statefulSet(replicas.data.name))
+		applied.statefulSet(replicas.data.name).generation)
 	if err != nil {
 		return rolloutRoles{}, err
 	}
@@ -2298,9 +2328,13 @@ func (r *MemgraphClusterReconciler) shedRetiredPods(
 		}
 		shrunk := role.build(cluster, role.replicas.declared)
 		volumes.KeepLiveSizes(shrunk, role.replicas.statefulSet)
-		if _, err := r.applyDesired(ctx, cluster, shrunk); err != nil {
+		applied, err := r.applyDesired(ctx, cluster, shrunk)
+		if err != nil {
 			return err
 		}
+		written := applied.statefulSet(role.replicas.name)
+		r.shrinks.Expect(types.NamespacedName{Namespace: cluster.Namespace, Name: role.replicas.name},
+			written.uid, written.generation)
 		log.Info("Shrank a StatefulSet to the declared replica count",
 			"statefulset", role.replicas.name, "replicas", role.replicas.declared)
 	}
@@ -2613,12 +2647,13 @@ func (r *MemgraphClusterReconciler) showInstances(
 // apply server-side-applies a desired object built by the resource builders.
 // Builders set only the fields the operator owns, so the converted apply
 // configuration claims exactly those fields for this controller. It returns the
-// generation the API server assigned the object: the response is decoded back
-// into the applied configuration, so no second read is needed for it.
-func (r *MemgraphClusterReconciler) apply(ctx context.Context, obj client.Object) (int64, error) {
+// object as the API server answered, its UID and generation included: the
+// response is decoded back into the applied configuration, so no second read
+// is needed for them.
+func (r *MemgraphClusterReconciler) apply(ctx context.Context, obj client.Object) (metav1.Object, error) {
 	content, err := runtime.DefaultUnstructuredConverter.ToUnstructured(obj)
 	if err != nil {
-		return 0, fmt.Errorf("converting to unstructured: %w", err)
+		return nil, fmt.Errorf("converting to unstructured: %w", err)
 	}
 	u := &unstructured.Unstructured{Object: content}
 	// Zero-valued struct fields survive the conversion; drop them so the
@@ -2629,9 +2664,9 @@ func (r *MemgraphClusterReconciler) apply(ctx context.Context, obj client.Object
 
 	if err := r.Apply(ctx, client.ApplyConfigurationFromUnstructured(u),
 		client.FieldOwner(fieldOwner), client.ForceOwnership); err != nil {
-		return 0, err
+		return nil, err
 	}
-	return u.GetGeneration(), nil
+	return u, nil
 }
 
 // SetupWithManager sets up the controller with the Manager. The Gateway API

internal/controller/memgraphcluster_controller_test.go

@@ -115,6 +115,24 @@ func libClaim(sts *appsv1.StatefulSet) corev1.PersistentVolumeClaimSpec {
 	return corev1.PersistentVolumeClaimSpec{}
 }
 
+// laggingCache is a client whose reads of the given StatefulSets return the
+// copies it holds, the way an informer cache serves an object until the watch
+// event of a write reaches it. Everything else, writes included, goes through.
+type laggingCache struct {
+	client.Client
+	statefulSets map[types.NamespacedName]*appsv1.StatefulSet
+}
+
+func (c laggingCache) Get(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
+	if sts, ok := obj.(*appsv1.StatefulSet); ok {
+		if held, ok := c.statefulSets[key]; ok {
+			held.DeepCopyInto(sts)
+			return nil
+		}
+	}
+	return c.Client.Get(ctx, key, obj, opts...)
+}
+
 var _ = Describe("MemgraphCluster Controller", func() {
 	const resourceNamespace = "default"
 
@@ -1000,6 +1018,47 @@ var _ = Describe("MemgraphCluster Controller", func() {
 				memgraphcomv1alpha1.ConditionConverged)).To(BeTrue())
 		})
 
+		// The shrink is the one write a lagging cache would undo: the count a pass
+		// applies is the larger of the declared one and the one the cached
+		// StatefulSet carries, so a pass reading the pre-shrink StatefulSet would
+		// apply the old count again and bring the shed pod back.
+		It("should not grow a StatefulSet it shrank back while the cache shows the old count", func() {
+			baseline := bootstrapped()
+			setCounts(3, 1)
+			reconcileCluster(resourceName)
+			Expect(sinceBootstrap(baseline)).To(HaveLen(1))
+
+			beforeShrink := &appsv1.StatefulSet{}
+			get(resourceName+dataSuffix, beforeShrink)
+			reconcileCluster(resourceName)
+			shrunk := &appsv1.StatefulSet{}
+			get(resourceName+dataSuffix, shrunk)
+			Expect(*shrunk.Spec.Replicas).To(Equal(int32(1)))
+
+			By("reconciling while the cache still serves the StatefulSet from before the shrink")
+			reconciler.Client = laggingCache{Client: k8sClient, statefulSets: map[types.NamespacedName]*appsv1.StatefulSet{
+				client.ObjectKeyFromObject(beforeShrink): beforeShrink,
+			}}
+			result := reconcileCluster(resourceName)
+			// The generation, not just the count: a pass that grew the StatefulSet
+			// back would shrink it again further on, once it found the member
+			// already gone, and the end state alone would hide the pod the
+			// StatefulSet controller recreated in between.
+			afterLag := &appsv1.StatefulSet{}
+			get(resourceName+dataSuffix, afterLag)
+			Expect(afterLag.Generation).To(Equal(shrunk.Generation), "the StatefulSet must not be written at all")
+			Expect(result.RequeueAfter).To(BeNumerically(">", 0), "the pass waits for the cache, it does not give up")
+			Expect(sinceBootstrap(baseline)).To(HaveLen(1), "nothing is issued while the cache lags")
+
+			By("converging once the cache shows the shrink")
+			reconciler.Client = k8sClient
+			reconcileCluster(resourceName)
+			Expect(replicas(dataSuffix)).To(Equal(int32(1)))
+			Expect(status().DataInstances).To(Equal(int32(1)))
+			Expect(apimeta.IsStatusConditionTrue(status().Conditions,
+				memgraphcomv1alpha1.ConditionConverged)).To(BeTrue())
+		})
+
 		// Memgraph refuses to unregister the MAIN, so a retiring instance holding it
 		// is demoted and a survivor promoted in its place — within the same pass, on
 		// the same leader connection, so the cluster is MAIN-less for the time